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)

2026-11-03 13:51:47 +01:00
parent 38a634723b
commit 2bb0a6461d
6 changed files with 323 additions and 7 deletions
@@ -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()
+	}
+}
@@ -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
+	})
+}
@@ -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: `<module>=<entrypoint>` 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)