262 lines
10 KiB
Go
262 lines
10 KiB
Go
//go:build integration
|
|
|
|
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"testing"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
|
"github.com/novox/mesh-tools/node-tools/internal/ledger"
|
|
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
|
|
)
|
|
|
|
// racing is a history whose append lets another writer apply first, once: before its own append, or after it.
|
|
type racing struct {
|
|
ledger.History
|
|
before, after func()
|
|
}
|
|
|
|
func (r *racing) Append(key string, op []byte, expect uint64) (uint64, error) {
|
|
if f := r.before; f != nil {
|
|
r.before = nil
|
|
f()
|
|
}
|
|
seq, err := r.History.Append(key, op, expect)
|
|
if f := r.after; f != nil && err == nil {
|
|
r.after = nil
|
|
f()
|
|
}
|
|
return seq, err
|
|
}
|
|
|
|
// cut is a state whose next write is lost, as a writer stopped between its history and its state would lose it.
|
|
type cut struct {
|
|
ledger.State
|
|
once bool
|
|
}
|
|
|
|
func (c *cut) Write(key string, value []byte, revision uint64) (uint64, error) {
|
|
if c.once {
|
|
c.once = false
|
|
return 0, errors.New("the writer stopped")
|
|
}
|
|
return c.State.Write(key, value, revision)
|
|
}
|
|
|
|
// 1966/AC19: the ledger keeps its promises against the real bus (ADR 0597's How it is checked; design 25 § 12.7), one
|
|
// case each, and every apply leaves the snapshot equal to the state.
|
|
func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) {
|
|
mesh := mt.New(t)
|
|
nc, err := nats.Connect(mt.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, _ := jetstream.New(nc)
|
|
ctx := context.Background()
|
|
for _, cfg := range []jetstream.StreamConfig{
|
|
{Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.>"}},
|
|
{Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger-snapshot.m.records.>"}, MaxMsgsPerSubject: 1},
|
|
} {
|
|
if _, err := js.CreateStream(ctx, cfg); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
for _, b := range []string{"m_records", "m_records_by_title"} {
|
|
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
binding := membership.BusBinding{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger,
|
|
History: "LEDGER_m_records", Subject: "mesh.ledger.m.records", State: "m_records",
|
|
Indexes: map[string]string{"by-title": "m_records_by_title"}, Snapshot: "mesh.ledger-snapshot.m.records"}
|
|
mem := mt.MembershipOf("m", "anchor", false, nil)
|
|
mem.Bindings = []membership.BusBinding{binding}
|
|
mesh.Issue(t, mem)
|
|
conn := connect(t, "node-tools", "anchor")
|
|
conn.Follow("m")
|
|
mt.Until(t, func() error {
|
|
if conn.Membership("m") == nil {
|
|
return errors.New("not issued yet")
|
|
}
|
|
return nil
|
|
})
|
|
types := []catalogue.LedgerType{{Name: "decision", Schema: map[string]catalogue.LedgerField{"title": {Type: "string"}},
|
|
Operations: []catalogue.LedgerOperation{
|
|
{Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}, Author: true},
|
|
{Name: "note", Effect: catalogue.EffectAppend, Into: "notes", Fields: []string{"says"}},
|
|
},
|
|
Indexes: []catalogue.LedgerIndex{{Name: "by-title", On: []string{"title"}}}}}
|
|
all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
|
|
WithLedgerTypes(func(string, string) ([]catalogue.LedgerType, error) { return types, nil })(all)
|
|
WithLedgerWriters(func(string) (*bus.Conn, error) { return conn, nil })(all)
|
|
m := all.forModule("m").(*moduleBus)
|
|
|
|
open := func(t *testing.T) *ledger.Ledger {
|
|
t.Helper()
|
|
l, err := all.ledger("m", "records")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return l
|
|
}
|
|
apply := func(t *testing.T, l *ledger.Ledger, instance, operation, body string) (buscalls.Applied, error) {
|
|
t.Helper()
|
|
return l.Apply(buscalls.ApplyRequest{Ledger: "records", Instance: instance, Operation: operation, Body: json.RawMessage(body)},
|
|
buscalls.Author{Account: "m"})
|
|
}
|
|
must := func(t *testing.T, l *ledger.Ledger, instance, operation, body string) buscalls.Applied {
|
|
t.Helper()
|
|
got, err := apply(t, l, instance, operation, body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return got
|
|
}
|
|
held := func(t *testing.T, key string) ledger.Record {
|
|
t.Helper()
|
|
raw, _, err := open(t).State.Get(key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var r ledger.Record
|
|
if raw != nil {
|
|
_ = json.Unmarshal(raw, &r)
|
|
}
|
|
return r
|
|
}
|
|
snapshotIsState := func(t *testing.T, key string) {
|
|
t.Helper()
|
|
l := open(t)
|
|
state, _, _ := l.State.Get(key)
|
|
snap, _, _ := l.Snapshots.Get(key)
|
|
if string(snap) != string(state) {
|
|
t.Fatalf("the snapshot of %s is %s, its state %s", key, snap, state)
|
|
}
|
|
}
|
|
|
|
cases := []struct {
|
|
name, key string
|
|
run func(t *testing.T)
|
|
}{
|
|
{"two writers race and one is refused", "decision.race", func(t *testing.T) {
|
|
l := open(t)
|
|
l.History = &racing{History: l.History, before: func() { must(t, open(t), "decision/race", "note", `{"says":"first"}`) }}
|
|
_, err := apply(t, l, "decision/race", "note", `{"says":"second"}`)
|
|
if !errors.Is(err, ledger.ErrRaced) || !strings.Contains(err.Error(), "not taken") {
|
|
t.Fatalf("the racing writer was answered %v", err)
|
|
}
|
|
if got := held(t, "decision.race"); string(got.State) != `{"notes":[{"says":"first"}]}` {
|
|
t.Fatalf("the state is %s", got.State)
|
|
}
|
|
}},
|
|
{"two interleaved applies leave the state at the higher sequence", "decision.interleave", func(t *testing.T) {
|
|
var inner buscalls.Applied
|
|
l := open(t)
|
|
l.History = &racing{History: l.History, after: func() { inner = must(t, open(t), "decision/interleave", "note", `{"says":"later"}`) }}
|
|
outer := must(t, l, "decision/interleave", "note", `{"says":"earlier"}`)
|
|
got := held(t, "decision.interleave")
|
|
if inner.Sequence <= outer.Sequence || got.Sequence != inner.Sequence ||
|
|
string(got.State) != `{"notes":[{"says":"earlier"},{"says":"later"}]}` {
|
|
t.Fatalf("applied %d then %d, the state at %d: %s", outer.Sequence, inner.Sequence, got.Sequence, got.State)
|
|
}
|
|
}},
|
|
{"a write cut between history and state is repaired by the next apply", "decision.cut", func(t *testing.T) {
|
|
must(t, open(t), "decision/cut", "note", `{"says":"one"}`)
|
|
l := open(t)
|
|
l.State = &cut{State: l.State, once: true}
|
|
if got := must(t, l, "decision/cut", "note", `{"says":"cut"}`); !got.Taken || !got.StateBehind {
|
|
t.Fatalf("the cut apply was answered %+v", got)
|
|
}
|
|
must(t, open(t), "decision/cut", "note", `{"says":"three"}`)
|
|
if got := held(t, "decision.cut"); string(got.State) != `{"notes":[{"says":"one"},{"says":"cut"},{"says":"three"}]}` {
|
|
t.Fatalf("the state is %s", got.State)
|
|
}
|
|
}},
|
|
{"a full state bucket answers taken with its sequence and catches up", "decision.full", func(t *testing.T) {
|
|
must(t, open(t), "decision/full", "note", `{"says":"small"}`)
|
|
kv, _ := js.KeyValue(ctx, "m_records")
|
|
status, _ := kv.Status(ctx)
|
|
if _, err := js.UpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "m_records", MaxBytes: int64(status.Bytes()) + 64}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got := must(t, open(t), "decision/full", "note", fmt.Sprintf(`{"says":%q}`, strings.Repeat("x", 512)))
|
|
if !got.Taken || got.Sequence == 0 || !got.StateBehind {
|
|
t.Fatalf("the full bucket's apply was answered %+v", got)
|
|
}
|
|
if _, err := js.UpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "m_records", MaxBytes: -1}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
after := must(t, open(t), "decision/full", "note", `{"says":"room"}`)
|
|
if st := held(t, "decision.full"); after.StateBehind || st.Sequence != after.Sequence || !strings.Contains(string(st.State), strings.Repeat("x", 512)) {
|
|
t.Fatalf("after room: %+v, the state at %d", after, st.Sequence)
|
|
}
|
|
}},
|
|
{"an instance whose operations expired rebuilds from its snapshot", "decision.expired", func(t *testing.T) {
|
|
must(t, open(t), "decision/expired", "title", `{"title":"Kept"}`)
|
|
want := held(t, "decision.expired")
|
|
s, _ := js.Stream(ctx, "LEDGER_m_records")
|
|
if err := s.Purge(ctx, jetstream.WithPurgeSubject("mesh.ledger.m.records.decision.expired")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kv, _ := js.KeyValue(ctx, "m_records")
|
|
_ = kv.Purge(ctx, "decision.expired")
|
|
got, err := open(t).Rebuild(buscalls.RebuildRequest{Instance: "decision/expired", From: buscalls.RebuildFromSnapshot})
|
|
if st := held(t, "decision.expired"); err != nil || got.Written != 1 || string(st.State) != string(want.State) {
|
|
t.Fatalf("rebuilt %+v, %v: the state %s, want %s", got, err, st.State, want.State)
|
|
}
|
|
}},
|
|
{"rebuild, reindex and lag repair by hand", "decision.repair", func(t *testing.T) {
|
|
must(t, open(t), "decision/repair", "title", `{"title":"Repaired"}`)
|
|
kv, _ := js.KeyValue(ctx, "m_records")
|
|
_ = kv.Purge(ctx, "decision.repair")
|
|
l := open(t)
|
|
lag, err := l.Lag(buscalls.LedgerRequest{})
|
|
if err != nil || !strings.Contains(fmt.Sprint(lag), "decision/repair") || len(lag.Indexes) == 0 {
|
|
t.Fatalf("lag says %+v, %v", lag, err)
|
|
}
|
|
if got, err := l.Rebuild(buscalls.RebuildRequest{Instance: "decision/repair"}); err != nil || got.Written != 1 {
|
|
t.Fatalf("rebuilt %+v, %v", got, err)
|
|
}
|
|
if _, err := l.Reindex(buscalls.ReindexRequest{Index: "by-title"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if lag, err := l.Lag(buscalls.LedgerRequest{}); err != nil || len(lag.Instances) != 0 || len(lag.Indexes) != 0 {
|
|
t.Fatalf("lag after the repairs says %+v, %v", lag, err)
|
|
}
|
|
q, err := l.Query(buscalls.QueryRequest{Index: "by-title", Key: "Repaired.>"})
|
|
if err != nil || len(q.Entries) != 1 {
|
|
t.Fatalf("queried %+v, %v", q, err)
|
|
}
|
|
}},
|
|
{"a forged author is refused", "decision.forged", func(t *testing.T) {
|
|
_, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"decision/forged","operation":"title","body":{"title":"T","author":"op@shanks"},"call":"forged"}`))
|
|
if err == nil || !strings.Contains(err.Error(), "the call forged, which is not a tool call in flight") {
|
|
t.Fatalf("a forged call was answered %v", err)
|
|
}
|
|
if last, _ := open(t).History.Last("decision.forged"); last != 0 {
|
|
t.Fatalf("the forged apply was appended at %d", last)
|
|
}
|
|
}},
|
|
}
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
c.run(t)
|
|
if c.key != "decision.forged" {
|
|
snapshotIsState(t, c.key)
|
|
}
|
|
})
|
|
}
|
|
}
|