From 961ec0ef96b1c830af3b71db88996c440d77854f Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 15:17:12 +0100 Subject: [PATCH] Append and snapshot a ledger on mesh-sdk's ledger subjects, the ones its account is granted, refusing a binding naming others (issue 1966 T6) --- node-tools/internal/bus/ledger.go | 23 +++++++++++++------ .../runtime/ledger_credentials_test.go | 8 +++---- .../runtime/ledger_integration_test.go | 10 ++++---- node-tools/internal/runtime/ledgers.go | 5 +++- node-tools/internal/runtime/ledgers_test.go | 15 +++++++----- 5 files changed, 38 insertions(+), 23 deletions(-) diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go index 417c64d..c9080eb 100644 --- a/node-tools/internal/bus/ledger.go +++ b/node-tools/internal/bus/ledger.go @@ -10,6 +10,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/membership" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -18,17 +19,25 @@ import ( ) // A ledger on the bus (novox/hq ADR 0597, design 25 § 12.4; issue 1966): its history a stream with a subject per -// instance under the binding's subject, its state a key-value bucket, its snapshots a subject per instance under the -// binding's snapshot subject, the latest only (the stream that holds them keeps one per subject, as the bus module +// instance under mesh-sdk's buscompose.LedgerHistorySubject, its state a key-value bucket, its snapshots a subject per +// instance under buscompose.LedgerSnapshotSubject, the latest only (the stream that holds them keeps one per subject, as the bus module // creates it). **Only the module's ledger connection publishes on its history and its snapshots** (ADR 0597 §9): what // is read, and the state, go on the runner's own. -// LedgerStores are the stores of one ledger binding: reads and the state on c, the history's appends and the snapshots' -// writes on writer, the module's ledger connection. -func (c *Conn) LedgerStores(b membership.BusBinding, writer *Conn) (ledger.History, ledger.State, ledger.Snapshots) { - return &busHistory{read: c, write: writer, stream: b.History, prefix: prefixOf(b.Subject)}, +// LedgerStores are the stores of a module's ledger binding: reads and the state on c, the history's appends and the +// snapshots' writes on writer, the module's ledger connection. The subjects are mesh-sdk's, the ones the grants part +// grants the module's ledger account byte for byte; a binding naming others is refused, naming both. +func (c *Conn) LedgerStores(module string, b membership.BusBinding, writer *Conn) (ledger.History, ledger.State, ledger.Snapshots, error) { + history, snapshot := buscompose.LedgerHistorySubject(module, b.Local), buscompose.LedgerSnapshotSubject(module, b.Local) + for _, named := range []struct{ what, bound, want string }{{"history", b.Subject, history}, {"snapshot", b.Snapshot, snapshot}} { + if named.bound != "" && prefixOf(named.bound) != named.want { + return nil, nil, nil, fmt.Errorf("%s's ledger %q is bound with the %s subject %s, and its ledger account is granted "+ + "%s alone (mesh-sdk buscompose); the binding is refused", module, b.Local, named.what, named.bound, named.want) + } + } + return &busHistory{read: c, write: writer, stream: b.History, prefix: history}, &busState{c: c, bucket: b.State}, - &busSnapshots{read: c, write: writer, prefix: prefixOf(b.Snapshot)} + &busSnapshots{read: c, write: writer, prefix: snapshot}, nil } // prefixOf is a binding's subject as the prefix an instance's key is added to: written with or without its `.>`. diff --git a/node-tools/internal/runtime/ledger_credentials_test.go b/node-tools/internal/runtime/ledger_credentials_test.go index 0a801d9..783cddb 100644 --- a/node-tools/internal/runtime/ledger_credentials_test.go +++ b/node-tools/internal/runtime/ledger_credentials_test.go @@ -25,7 +25,7 @@ import ( // (the runner's own account may not publish on m's ledger, as the grants part grants it); given it missing, apply is // refused naming it; a change of the file is taken without a restart. func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) { - ledgerSubjects := []string{"mesh.ledger.m.records.>", "mesh.ledger-snapshot.m.records.>"} + ledgerSubjects := []string{buscompose.LedgerHistorySubject("m", "records") + ".>", buscompose.LedgerSnapshotSubject("m", "records") + ".>"} users := []*server.User{ {Username: "node-tools", Password: "pw", Permissions: &server.Permissions{ Publish: &server.SubjectPermission{Allow: []string{">"}, Deny: ledgerSubjects}}}, @@ -53,8 +53,8 @@ func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) { js, _ := jetstream.New(admin) 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}, + {Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.history.>"}}, + {Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger.m.records.snapshot.>"}, MaxMsgsPerSubject: 1}, } { if _, err := js.CreateStream(ctx, cfg); err != nil { t.Fatal(err) @@ -67,7 +67,7 @@ func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) { } mem := mt.MembershipOf("m", "anchor", false, nil) mem.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, - History: "LEDGER_m_records", Subject: "mesh.ledger.m.records", State: "m_records", Snapshot: "mesh.ledger-snapshot.m.records", + History: "LEDGER_m_records", Subject: "mesh.ledger.m.records.history", State: "m_records", Snapshot: "mesh.ledger.m.records.snapshot", Types: []catalogue.LedgerType{{Name: "decision", Operations: []catalogue.LedgerOperation{ {Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}}}}}}} raw, _ := json.Marshal(mem) diff --git a/node-tools/internal/runtime/ledger_integration_test.go b/node-tools/internal/runtime/ledger_integration_test.go index c58e304..802fb03 100644 --- a/node-tools/internal/runtime/ledger_integration_test.go +++ b/node-tools/internal/runtime/ledger_integration_test.go @@ -66,8 +66,8 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) { 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}, + {Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.history.>"}}, + {Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger.m.records.snapshot.>"}, MaxMsgsPerSubject: 1}, } { if _, err := js.CreateStream(ctx, cfg); err != nil { t.Fatal(err) @@ -85,8 +85,8 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) { }, Indexes: []catalogue.LedgerIndex{{Name: "by-title", On: []string{"title"}}}}} 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"} + History: "LEDGER_m_records", Subject: "mesh.ledger.m.records.history", State: "m_records", + Indexes: map[string]string{"by-title": "m_records_by_title"}, Snapshot: "mesh.ledger.m.records.snapshot"} mem := mt.MembershipOf("m", "anchor", false, nil) binding.Types = types mem.Bindings = []membership.BusBinding{binding} @@ -207,7 +207,7 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(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 { + if err := s.Purge(ctx, jetstream.WithPurgeSubject("mesh.ledger.m.records.history.decision.expired")); err != nil { t.Fatal(err) } kv, _ := js.KeyValue(ctx, "m_records") diff --git a/node-tools/internal/runtime/ledgers.go b/node-tools/internal/runtime/ledgers.go index b3d62dc..ab2be1e 100644 --- a/node-tools/internal/runtime/ledgers.go +++ b/node-tools/internal/runtime/ledgers.go @@ -98,7 +98,10 @@ func (c *consumers) ledger(module, local string) (*ledger.Ledger, error) { return nil, err } } - history, state, snapshots := c.conn.LedgerStores(b, writer) + history, state, snapshots, err := c.conn.LedgerStores(module, b, writer) + if err != nil { + return nil, err + } // The owning module's authenticated account: the user of the connection its history is published on. account := module if writer != nil && writer.Credential().User != "" { diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index 8b17dbf..b688711 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -24,8 +24,8 @@ func TestTheLedgerRepairsAreToolsOfEveryLedgerOwner(t *testing.T) { mesh := mt.New(t) owner := mt.MembershipOf("owner", "anchor", true, nil) owner.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, - History: "LEDGER_owner_records", Subject: "mesh.ledger.owner.records", State: "owner_records", - Snapshot: "mesh.ledger-snapshot.owner.records"}} + History: "LEDGER_owner_records", Subject: "mesh.ledger.owner.records.history", State: "owner_records", + Snapshot: "mesh.ledger.owner.records.snapshot"}} plain := mt.MembershipOf("plain", "anchor", true, nil) mesh.Issue(t, owner) mesh.Issue(t, plain) @@ -91,8 +91,8 @@ func aLedger(t *testing.T) (m, other *moduleBus, mesh *mt.Mesh) { 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}, + {Name: "LEDGER_m_records", Subjects: []string{"mesh.ledger.m.records.history.>"}}, + {Name: "SNAPSHOT_m_records", Subjects: []string{"mesh.ledger.m.records.snapshot.>"}, MaxMsgsPerSubject: 1}, } { if _, err := js.CreateStream(ctx, cfg); err != nil { t.Fatal(err) @@ -103,11 +103,13 @@ func aLedger(t *testing.T) (m, other *moduleBus, mesh *mt.Mesh) { } mem := mt.MembershipOf("m", "anchor", false, nil) mem.Bindings = []membership.BusBinding{{Local: "records", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, - History: "LEDGER_m_records", Subject: "mesh.ledger.m.records", State: "m_records", Snapshot: "mesh.ledger-snapshot.m.records", + History: "LEDGER_m_records", Subject: "mesh.ledger.m.records.history", State: "m_records", Snapshot: "mesh.ledger.m.records.snapshot", 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}}}}}, {Local: "untyped", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, History: "LEDGER_m_records", - Subject: "mesh.ledger.m.records", State: "m_records", Snapshot: "mesh.ledger-snapshot.m.records"}} + State: "m_records"}, + {Local: "elsewhere", Storage: catalogue.StorageLedger, Template: catalogue.TemplateLedger, History: "LEDGER_m_records", + Subject: "mesh.ledger.m.records.history", State: "m_records", Types: []catalogue.LedgerType{{Name: "decision"}}}} mesh.Issue(t, mem) mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil)) conn := connect(t, "node-tools", "anchor") @@ -139,6 +141,7 @@ func TestAnApplyIsStampedWithTheCallerTheRuntimeProved(t *testing.T) { {"naming no call stamps the account alone", `{"ledger":"records","instance":"decision/2","operation":"title",` + forged + `}`, "", ""}, {"1966/AC18 another module's call is refused, naming it", `{"ledger":"records","instance":"decision/3","operation":"title",` + forged + `,"call":"` + theirs + `"}`, "", "the call " + theirs + ", which is another module's"}, {"a ledger bound with no types is refused, naming it", `{"ledger":"untyped","instance":"decision/3","operation":"title",` + forged + `}`, "", `m's ledger "untyped" is bound with no types`}, + {"a binding naming subjects its account is not granted is refused, naming both", `{"ledger":"elsewhere","instance":"decision/3","operation":"title",` + forged + `}`, "", "bound with the history subject mesh.ledger.m.records.history, and its ledger account is granted mesh.ledger.m.elsewhere.history alone"}, {"a call not in flight is refused, naming it", `{"ledger":"records","instance":"decision/3","operation":"title",` + forged + `,"call":"c0ffee"}`, "", "the call c0ffee, which is not a tool call in flight"}, } { t.Run(c.name, func(t *testing.T) {