From 2bb0a6461df9254078877bba048708a669f9e560 Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:51:47 +0100 Subject: [PATCH] Publish a module's ledger history and snapshots only on its own ledger connection, read from the runner's own secret ledgers and followed without a restart (issue 1966 T6) --- node-tools/cmd/node-tools/main.go | 11 +- node-tools/internal/bus/bus.go | 16 +- .../internal/runtime/ledger_credentials.go | 143 ++++++++++++++++++ .../runtime/ledger_credentials_test.go | 137 +++++++++++++++++ node-tools/internal/runtime/runtime.go | 20 ++- node-tools/module.json | 3 +- 6 files changed, 323 insertions(+), 7 deletions(-) create mode 100644 node-tools/internal/runtime/ledger_credentials.go create mode 100644 node-tools/internal/runtime/ledger_credentials_test.go diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index f053f93..bf55106 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -5,6 +5,7 @@ // wherever MESH_CONSOLE_LISTEN says — it is also the console, MCP over HTTP on loopback. // // MESH_BROKER_FILE the credential the mesh sealed to this machine for the runtime +// MESH_LEDGERS_FILE its own secret ledgers, each module's ledger credential; beside the broker file when unset // MESH_TOOL_MODULES =,… the bundles to launch // MESH_TOOL_ENV {"": {"": ""}} what each is given (ADR 0192) // MESH_OPERATOR_ACCOUNT whose machine this is, and MESH_OPERATOR_HOME their home @@ -56,7 +57,15 @@ func main() { if err != nil { log.Fatalf("node-tools: %v", err) } - stop, err := runtime.Run(conn, served, envs, log.Printf) + // Its own secret ledgers: each module's ledger connection, the only one its history is published on (novox/hq ADR + // 0597 §9). Followed, so a credential minted or retired is taken with no restart. + var opts []runtime.Option + if file := runtime.LedgersFileOf(os.Getenv("MESH_BROKER_FILE")); file != "" { + ledgers := runtime.FollowLedgerCredentials(file, bus.Dial, log.Printf) + defer ledgers.Stop() + opts = append(opts, runtime.WithLedgerWriters(ledgers.Writer)) + } + stop, err := runtime.Run(conn, served, envs, log.Printf, opts...) if err != nil { log.Fatalf("node-tools: %v", err) } diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 8027e53..d5e5e3d 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -288,9 +288,20 @@ func (c *Conn) refused(err error) { }) } -// Connect dials the bus as the credential's module. A module's subjects come from its credential, -// never from its calls (ADR 0074). +// Connect dials the bus as the credential's module and follows its membership. A module's subjects come from its +// credential, never from its calls (ADR 0074). func Connect(cred Credential) (*Conn, error) { + c, err := Dial(cred) + if err != nil { + return nil, err + } + c.Follow(cred.Module) + return c, nil +} + +// Dial dials the bus as the credential's module and follows nothing: a connection that only publishes for its module, +// as a module's ledger connection does (novox/hq ADR 0597 §9). +func Dial(cred Credential) (*Conn, error) { if cred.Module == "" { return nil, errors.New("a broker credential with no module: the runtime derives its subjects " + "from the account the mesh issued, and cannot guess which module it is") @@ -329,7 +340,6 @@ func Connect(cred Credential) (*Conn, error) { } c = &Conn{nc: nc, js: js, self: cred.Module, node: cred.Node, cred: cred, issued: map[string]*Membership{}, Logf: log.Printf, answering: map[string]*answered{}} - c.Follow(cred.Module) return c, nil } diff --git a/node-tools/internal/runtime/ledger_credentials.go b/node-tools/internal/runtime/ledger_credentials.go new file mode 100644 index 0000000..3abca81 --- /dev/null +++ b/node-tools/internal/runtime/ledger_credentials.go @@ -0,0 +1,143 @@ +package runtime + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "sync" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// The ledger credentials (novox/hq ADR 0597 §9, design 25 § 12.4; issue 1966 T6): one per module assignment, held by +// the machine's tool runner in its own secret `ledgers`, a map of module to credential (`{}` when none), which the vault +// rewrites on each mint and retirement. **A module's ledger history is appended, and its snapshots written, only on +// that module's own ledger connection**; everything else stays on the runner's own. + +// LedgersFile is where the runner reads its own secret `ledgers`: MESH_LEDGERS_FILE when set, else beside its broker +// credential, both in its directory mesh-state (node-tools' manifest: `${dir:mesh-state}/ledgers`). +const LedgersFile = "MESH_LEDGERS_FILE" + +// LedgersFileOf is the path of the secret `ledgers` for a runtime whose broker credential is at broker. +func LedgersFileOf(broker string) string { + if file := os.Getenv(LedgersFile); file != "" { + return file + } + if broker == "" { + return "" + } + return filepath.Join(filepath.Dir(broker), "ledgers") +} + +// LedgerCredentials are the ledger connections of the modules on this machine, read from the secret `ledgers` and +// followed as it changes, with no restart. +type LedgerCredentials struct { + file string + dial func(bus.Credential) (*bus.Conn, error) + logf func(string, ...any) + + mu sync.Mutex + creds map[string]bus.Credential + conns map[string]*bus.Conn + why string // why the file could not be read, when it could not + stop func() +} + +// FollowLedgerCredentials reads the secret `ledgers` at file and follows it (WatchFile), dialling a module's connection +// with dial the first time it is asked for. +func FollowLedgerCredentials(file string, dial func(bus.Credential) (*bus.Conn, error), logf func(string, ...any)) *LedgerCredentials { + l := &LedgerCredentials{file: file, dial: dial, logf: logf, creds: map[string]bus.Credential{}, conns: map[string]*bus.Conn{}} + l.read() + stop, err := WatchFile(file, l.read, logf, "a change of the ledger credentials is taken only after a restart") + if err != nil { + logf("[mesh-tools] cannot follow the ledger credentials at %s: %v; a change is taken only after a restart", file, err) + } else { + l.stop = stop + } + return l +} + +// read takes the file as it is now: a module whose credential changed or went has its connection closed, and is +// dialled again from the new one when next asked for. +func (l *LedgerCredentials) read() { + creds, why := map[string]bus.Credential{}, "" + raw, err := os.ReadFile(l.file) + switch { + case errors.Is(err, os.ErrNotExist): + why = "the file " + l.file + " is not there yet" + case err != nil: + why = err.Error() + case len(bytes.TrimSpace(raw)) == 0: + why = "the file " + l.file + " is empty" + default: + if err := json.Unmarshal(raw, &creds); err != nil { + why = "the file " + l.file + " is not a map of module to credential: " + err.Error() + creds = map[string]bus.Credential{} + } + } + l.mu.Lock() + var closing []*bus.Conn + for module, conn := range l.conns { + if now, has := creds[module]; !has || !reflect.DeepEqual(now, l.creds[module]) { + closing = append(closing, conn) + delete(l.conns, module) + } + } + saidBefore := l.why + l.creds, l.why = creds, why + l.mu.Unlock() + for _, c := range closing { + c.Close() + } + if why != "" && why != saidBefore { + l.logf("[mesh-tools] the ledger credentials cannot be read (%s); no ledger is written until they can", why) + } +} + +// Writer is a module's ledger connection, dialled from its credential the first time it is asked for; refused, naming +// what is missing, when the secret holds none for it. +func (l *LedgerCredentials) Writer(module string) (*bus.Conn, error) { + l.mu.Lock() + defer l.mu.Unlock() + if c := l.conns[module]; c != nil { + return c, nil + } + cred, has := l.creds[module] + if !has { + why := "" + if l.why != "" { + why = " (" + l.why + ")" + } + return nil, fmt.Errorf("%s's ledger credential is missing from the tool runner's own secret ledgers%s: its history "+ + "is published only on its own ledger connection, which the grants part issues with its membership and the "+ + "vault writes there (novox/hq ADR 0597 §9)", module, why) + } + if cred.Module == "" { + cred.Module = module + } + c, err := l.dial(cred) + if err != nil { + return nil, fmt.Errorf("%s's ledger connection could not be opened: %w", module, err) + } + c.Logf = l.logf + l.conns[module] = c + return c, nil +} + +// Stop ends the following and closes every ledger connection. +func (l *LedgerCredentials) Stop() { + if l.stop != nil { + l.stop() + } + l.mu.Lock() + conns := l.conns + l.conns = map[string]*bus.Conn{} + l.mu.Unlock() + for _, c := range conns { + c.Close() + } +} diff --git a/node-tools/internal/runtime/ledger_credentials_test.go b/node-tools/internal/runtime/ledger_credentials_test.go new file mode 100644 index 0000000..a517f7c --- /dev/null +++ b/node-tools/internal/runtime/ledger_credentials_test.go @@ -0,0 +1,137 @@ +package runtime + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// 1966/AC22: given the secret ledgers holding m's credential, when m applies, then the append goes on m's connection +// (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.>"} + users := []*server.User{ + {Username: "node-tools", Password: "pw", Permissions: &server.Permissions{ + Publish: &server.SubjectPermission{Allow: []string{">"}, Deny: ledgerSubjects}}}, + {Username: "ledger.anchor.m", Password: "pw", Permissions: &server.Permissions{ + Publish: &server.SubjectPermission{Allow: append([]string{"$JS.API.>", "_INBOX.>"}, ledgerSubjects...)}}}, + {Username: "admin", Password: "pw"}, + } + s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), + NoLog: true, NoSigs: true, Users: users}) + if err != nil { + t.Fatal(err) + } + go s.Start() + if !s.ReadyForConnections(30 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(s.Shutdown) + url := s.ClientURL() + + admin, err := nats.Connect(url, nats.UserInfo("admin", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(admin.Close) + 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}, + } { + if _, err := js.CreateStream(ctx, cfg); err != nil { + t.Fatal(err) + } + } + for _, b := range []string{"m_records", buscompose.GrantsMembershipsBucket} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { + t.Fatal(err) + } + } + 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"}} + raw, _ := json.Marshal(mem) + var sdk membership.Membership + _ = json.Unmarshal(raw, &sdk) + grants, _ := js.KeyValue(ctx, buscompose.GrantsMembershipsBucket) + if _, err := grants.Put(ctx, buscompose.MembershipKey("anchor", "m"), membership.EncodeGrantsMembership(sdk)); err != nil { + t.Fatal(err) + } + + conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor", User: "node-tools", Password: "pw"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + conn.Follow("m") + mt.Until(t, func() error { + if conn.Membership("m") == nil { + return os.ErrNotExist + } + return nil + }) + + file := filepath.Join(t.TempDir(), "ledgers") + if err := os.WriteFile(file, []byte(`{}`), 0o600); err != nil { + t.Fatal(err) + } + creds := FollowLedgerCredentials(file, bus.Dial, t.Logf) + t.Cleanup(creds.Stop) + types := func(string, string) ([]catalogue.LedgerType, error) { + return []catalogue.LedgerType{{Name: "decision", Operations: []catalogue.LedgerOperation{ + {Name: "title", Effect: catalogue.EffectSetFields, Fields: []string{"title"}}}}}, nil + } + all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}} + WithLedgerTypes(types)(all) + WithLedgerWriters(creds.Writer)(all) + m := all.forModule("m").(*moduleBus) + apply := func(instance string) error { + _, err := m.BusState(buscalls.LedgerApply, json.RawMessage(`{"ledger":"records","instance":"`+instance+`","operation":"title","body":{"title":"T"}}`)) + return err + } + + // Missing: refused, naming it. + if err := apply("decision/1"); err == nil || !strings.Contains(err.Error(), "m's ledger credential is missing from the tool runner's own secret ledgers") { + t.Fatalf("an apply with no credential: %v", err) + } + // Written: taken without a restart, on m's connection (the runner's own account is denied m's ledger subjects, so an + // append that lands went on m's), whose account stamps the operation. + cred, _ := json.Marshal(map[string]bus.Credential{"m": {URL: url, Node: "anchor", Module: "m", User: "ledger.anchor.m", Password: "pw"}}) + if err := os.WriteFile(file, cred, 0o600); err != nil { + t.Fatal(err) + } + mt.Until(t, func() error { return apply("decision/1") }) + page, err := m.BusState(buscalls.LedgerHistory, json.RawMessage(`{"ledger":"records","instance":"decision/1"}`)) + if err != nil || !strings.Contains(string(page), `"author":{"account":"ledger.anchor.m"}`) { + t.Fatalf("the history holds %s, %v", page, err) + } + // Gone again: refused again, the change taken without a restart. + if err := os.WriteFile(file, []byte(`{}`), 0o600); err != nil { + t.Fatal(err) + } + mt.Until(t, func() error { + if err := apply("decision/2"); err == nil || !strings.Contains(err.Error(), "credential is missing") { + return os.ErrExist + } + return nil + }) +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index b7d6e41..7e4756e 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -61,7 +61,7 @@ const NodeRuntimeModule = "node-tools" // runtimeOnly are the words of the runtime's own environment no bundle is given. var runtimeOnly = map[string]bool{ToolEnv: true, ToolModules: true, "MESH_BROKER_FILE": true, - "MESH_BROKER_URL": true, "MESH_CONSOLE_LISTEN": true} + "MESH_BROKER_URL": true, "MESH_CONSOLE_LISTEN": true, LedgersFile: true} // ServedModulesFrom reads MESH_TOOL_MODULES: `=` entries, comma-separated, several // per module. The one-module form — a bare path, or the runtime's own module — is the per-module @@ -234,9 +234,22 @@ type runner struct { stops []func() // the watch, the announcement and the seats, in the order they began } +// Option sets how Run serves, beside what it is given. +type Option func(*consumers) + +// WithLedgerWriters gives the runtime the connection each module's ledger history is published on (ledger_credentials.go). +func WithLedgerWriters(writer func(module string) (*bus.Conn, error)) Option { + return func(c *consumers) { c.ledgers.writer = writer } +} + +// WithLedgerTypes gives the runtime where it reads a ledger's types (ledgers.go). +func WithLedgerTypes(types LedgerTypes) Option { + return func(c *consumers) { c.ledgers.types = types } +} + // Run launches, binds and serves, and serves each module's bundle again when it appears, changes or // goes, with no restart. It answers a stop function. -func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, logf func(string, ...any)) (func(), error) { +func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, logf func(string, ...any), opts ...Option) (func(), error) { if account := os.Getenv(OperatorAccount); account != "" { home := "" if h := os.Getenv(OperatorHome); h != "" { @@ -259,6 +272,9 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log // The module's events, for every child of it that subscribes (ADR 0198): one consumer per module, // bound the first time any of its children subscribes, each event handed to every child that did. r.events = &consumers{conn: conn, logf: logf, of: map[string]*moduleEvents{}} + for _, opt := range opts { + opt(r.events) + } for _, s := range served { r.of[s.Module] = r.load(s) diff --git a/node-tools/module.json b/node-tools/module.json index e6cb6db..e456563 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -28,7 +28,8 @@ } }, "own-secrets": { - "broker": "${dir:mesh-state}/broker" + "broker": "${dir:mesh-state}/broker", + "ledgers": "${dir:mesh-state}/ledgers" }, "listens": [ {