diff --git a/node-tools/internal/runtime/ledger_integration_test.go b/node-tools/internal/runtime/ledger_integration_test.go new file mode 100644 index 0000000..11d8153 --- /dev/null +++ b/node-tools/internal/runtime/ledger_integration_test.go @@ -0,0 +1,261 @@ +//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) + } + }) + } +}