From bb993dd606412166ddec503a35e61fee327cd9a4 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:28:21 +0100 Subject: [PATCH] Keep the builder's history as its own log that readers declare and follow, the memberships as the grants part's state, and refuse a module's own stream, so every module's data is in the bus's predefined forms and readable by its tools (issue 1955, ADR 0571 clauses 1, 4, 5, 7) --- go/buscompose/buscompose_test.go | 33 +++++- go/buscompose/log.go | 59 +++++++++- go/buscompose/membership.go | 7 +- go/buscompose/nats.go | 20 ++++ go/buscompose/users.go | 10 +- go/catalogue/seats_declared.go | 2 + go/catalogue/streams.go | 28 +++++ go/catalogue/streams_test.go | 8 ++ go/delivery/control_records_test.go | 42 ++++++- go/delivery/history_records.go | 149 +++++++++++++++++++----- go/log.go | 106 ++++++++++++++++- go/log_test.go | 56 ++++++++- go/process/fixture_integration_test.go | 8 ++ go/process/log.go | 152 +++++++++++++++++++++++++ go/process/state_integration_test.go | 38 +++++++ go/stdio.go | 28 ++++- 16 files changed, 695 insertions(+), 51 deletions(-) create mode 100644 go/process/log.go diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index 5e356b4..a83b771 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -59,6 +59,19 @@ func beyondTheControllers(user string) map[string]bool { // mayBeyondTheControllers is what the copy may grant beyond the controller's where the records allow it, and need not // where they do not: a node's ask of the vault's `own` while the vault serves it (issue 1126, ADR 0550 §5). func mayBeyondTheControllers(user string) map[string]bool { + // A runtime's read of its membership from the grants part's state (novox/hq ADR 0571 clause 7): its own key, or the + // tool runner's node's. + if parts := strings.Split(user, "."); len(parts) == 2 && parts[0] != "node" { + key := user + if parts[1] == RuntimeModule { + key = parts[0] + ".*" + } + out := map[string]bool{} + for _, s := range readGrants(GrantsMembershipsBucket, key) { + out[s] = true + } + return out + } if _, ok := strings.CutPrefix(user, "node."); !ok { return nil } @@ -618,6 +631,11 @@ func healthChecks(t *testing.T) { // ADR 0571 clause 3: disk-load keeps the readings; the backup holder reads them. {Module: "disk-load", State: []Bucket{bucket("disk-load", "readings")}}, {Module: "restic", Reads: []string{"disk-load.readings"}}, + // ADR 0571 clauses 1, 9: the indexer reads the builder's log through its own consumer; the catalogue module asks + // the builder's tool. + {Module: delivery.BuilderModule, Logs: []Log{{Module: delivery.BuilderModule, Name: delivery.BuilderLog, MaxAgeDays: 30}}}, + {Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead}}, + {Module: "mesh-catalog", Invokes: []string{delivery.BuilderModule + "." + delivery.BuildsSinceTool}}, {Module: "telegram", Holds: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}, Processes: []ProcessPrincipal{{Process: "w", Queues: []QueueOwner{{Process: "w", Stream: "M_WORK"}, {Process: "w", Stream: "KV_x"}}}}}, {Module: "nats", SnapshotsTheBus: true, Processes: []ProcessPrincipal{{Process: "nats-tools", @@ -634,7 +652,14 @@ func healthChecks(t *testing.T) { "$JS.API.DIRECT.GET.KV_mesh-provisioner_plans.$KV.mesh-provisioner_plans.a.keycloak"}, "a.mesh-hand-acts": {"$KV.mesh-hand-acts_warrants.mesh-config.ask-1", "mesh.log.mesh-hand-acts.acts.act-1"}, "a.disk-load": {"$KV.disk-load_readings.a.pg.data"}, - "a.restic": {"$JS.API.DIRECT.GET.KV_disk-load_readings.$KV.disk-load_readings.a.pg.data"}, + "a.restic": {"$JS.API.DIRECT.GET.KV_disk-load_readings.$KV.disk-load_readings.a.pg.data", + // ADR 0571 clause 7: its own membership from the grants part's state. + "$JS.API.DIRECT.GET.KV_mesh-grants_memberships.$KV.mesh-grants_memberships.a.restic"}, + "a.mesh-delivery-indexer": {"$JS.API.CONSUMER.CREATE.LOG_mesh-delivery-builder_build-log.r1.mesh.log.mesh-delivery-builder.build-log.letta", + "$JS.API.CONSUMER.MSG.NEXT.LOG_mesh-delivery-builder_build-log.r1", "$JS.API.CONSUMER.DELETE.LOG_mesh-delivery-builder_build-log.r1", + "$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log.mesh.log.mesh-delivery-builder.build-log.letta"}, + "a.mesh-delivery-builder": {"mesh.log.mesh-delivery-builder.build-log.letta"}, + "a.mesh-catalog": {"mesh.mod.mesh-delivery-builder.tool.builder_builds_since"}, WatchmanModule: {"$JS.API.CONSUMER.CREATE.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTER_NOTICES." + WatchmanModule, "$JS.API.CONSUMER.INFO.DEAD_LETTER_NOTICES.controller", "$JS.API.STREAM.MSG.GET.DEAD_LETTERS"}, WatchmanModule + " sources": {"$JS.API.STREAM.MSG.GET.EVENTS", "$JS.API.STREAM.INFO.SEAT_TELEGRAM_SENDER", @@ -678,6 +703,12 @@ func healthChecks(t *testing.T) { "a.nats": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-host_acts.g14.a1"}, "a.shop.nats-tools": {"$JS.API.CONSUMER.DELETE.DEAD_LETTER_NOTICES.controller", "$KV.mesh-conditions_conditions.x"}, } + // ADR 0571 clause 1: a read naming another module's log is issued to the reader as a log it reads, never writes. + if m := MembershipFor("a", Declared{Module: delivery.IndexerModule, Reads: []string{delivery.BuildLogRead, "x.y"}}, + Placements{Logs: map[string]bool{delivery.BuildLogRead: true}}); len(m.Logs) != 1 || m.Logs[0].Writes || + m.Logs[0].Stream != "LOG_mesh-delivery-builder_build-log" || m.Logs[0].Name != delivery.BuildLogRead { + t.Errorf("the indexer's membership issues the logs %+v", m.Logs) + } // ADR 0565 clauses 6, 7 (issue 1953 L5-S4): a node's engine writes its own packages key and no other node's; only the // assigner writes its policies. for _, u := range users { diff --git a/go/buscompose/log.go b/go/buscompose/log.go index 9f1370f..d179b2b 100644 --- a/go/buscompose/log.go +++ b/go/buscompose/log.go @@ -15,8 +15,10 @@ import ( // A log follows a bucket in every respect (ADR 0201): the module names it locally, the mesh derives // its stream and subjects, the controller creates it from the catalogue on every raise and from // registration, never a module, and **no path of the mesh removes one** — a log whose declaration -// is gone is reported, as a bucket is. Unlike a bucket, a log is its owner's alone: no other module -// reads it, so there is no read of a log to grant or issue. +// is gone is reported, as a bucket is. Only its owner appends to it. Another module reads it when its manifest's +// `reads` names it, `.` (novox/hq ADR 0571 clause 1; the lead, 2026-10-29): by direct gets, and through an +// ordered consumer of its own it creates and lets go of to read from a time or the newest per key and then follow new +// entries; never a durable consumer, never a write. // // Pure, but for the stream's configuration, which is the server's own type so that what is tested // is exactly what is sent; jetstream.go is the part that asks a server. @@ -138,6 +140,59 @@ func logGrants(module string, names []string) []string { return out } +// DeclaredLogs is every log a module placed anywhere declares, as a read names it: `.`. +func DeclaredLogs(r Records) map[string]bool { + out := map[string]bool{} + for _, ds := range r.Assigned { + for _, d := range ds { + for _, l := range d.Logs { + out[d.Module+"."+l.Name] = true + } + } + } + return out +} + +// withReadLogs is d with ReadLogs set: of its reads, those naming a log of logs. +func withReadLogs(d Declared, logs map[string]bool) Declared { + d.ReadLogs = nil + for _, r := range d.Reads { + if logs[strings.TrimSpace(r)] { + d.ReadLogs = append(d.ReadLogs, strings.TrimSpace(r)) + } + } + return d +} + +// readLogGrants is what a reader of other modules' logs publishes, for each `.` read: the stream's info, +// direct gets (an entry by sequence, the last of a key), and an ordered consumer of its own (created with any name and +// filter, pulled, read and deleted). No append, no purge, no delete of an entry: the log's stream refuses those anyway. +func readLogGrants(reads []string) []string { + var out []string + for _, r := range reads { + module, name, ok := strings.Cut(r, ".") + if !ok || !safeSubject.MatchString(module) || !safeSubject.MatchString(name) { + continue + } + stream := LogStreamName(module, name) + out = append(out, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream, "$JS.API.DIRECT.GET."+stream+".>", + "$JS.API.CONSUMER.CREATE."+stream+".>", "$JS.API.CONSUMER.MSG.NEXT."+stream+".*", + "$JS.API.CONSUMER.INFO."+stream+".*", "$JS.API.CONSUMER.DELETE."+stream+".*") + } + return out +} + +// readLogsIssuedFor is every other module's log a module's code reads, as its membership lists them: by the read's +// name, never written. +func readLogsIssuedFor(d Declared) []LogIssued { + var out []LogIssued + for _, r := range d.ReadLogs { + module, name, _ := strings.Cut(r, ".") + out = append(out, LogIssued{Name: r, Stream: LogStreamName(module, name), Subject: LogSubject(module, name)}) + } + return out +} + // logNames is the local names of a module's logs. func logNames(logs []Log) []string { out := make([]string, 0, len(logs)) diff --git a/go/buscompose/membership.go b/go/buscompose/membership.go index d28ff68..4b4b716 100644 --- a/go/buscompose/membership.go +++ b/go/buscompose/membership.go @@ -24,6 +24,9 @@ type Placements struct { Interchangeable map[string]bool // Kinds is every kind held of a kinded bench, by whom and with what capabilities (ADR 0259 §5). Kinds []KindHeld + // Logs is every log a module placed anywhere declares, `.` (DeclaredLogs): a read naming one is issued + // as a log read (novox/hq ADR 0571 clause 1). + Logs map[string]bool } // AnswersForTheModule says whether an instance of a module on one machine is issued the module's @@ -58,7 +61,7 @@ func MembershipFor(node string, d Declared, where Placements) Membership { } } m.State = stateIssuedFor(d, node) - m.Logs = logsIssuedFor(d) + m.Logs = append(logsIssuedFor(d), readLogsIssuedFor(withReadLogs(d, where.Logs))...) m.Streams = streamsIssuedFor(d) m.Hears = HeardIssuedFor(node, d) m.Waits = waitsIssuedFor(d) @@ -103,7 +106,7 @@ func MembershipFor(node string, d Declared, where Placements) Membership { // PlacementsOf reads where everything runs from the records the bus's accounts are composed from. func PlacementsOf(r Records, interchangeable map[string]bool) Placements { - p := Placements{Nodes: map[string][]string{}, Interchangeable: interchangeable} + p := Placements{Nodes: map[string][]string{}, Interchangeable: interchangeable, Logs: DeclaredLogs(r)} for node, declared := range r.Assigned { for _, d := range declared { p.Nodes[d.Module] = append(p.Nodes[d.Module], node) diff --git a/go/buscompose/nats.go b/go/buscompose/nats.go index 83378f1..61bdfa9 100644 --- a/go/buscompose/nats.go +++ b/go/buscompose/nats.go @@ -181,6 +181,8 @@ type Principal struct { KeyedReads []KeyedRead // Logs is the local names of the logs this principal's module keeps (novox/hq ADR 0297). Logs []string + // ReadLogs are, of Reads, those naming another module's log (novox/hq ADR 0571 clause 1). + ReadLogs []string // Streams are the streams this principal's module owns, and Hears the module streams it hears with the // subjects it was granted (novox/hq ADR 0363). Streams []ModuleStream @@ -1063,6 +1065,9 @@ func PermissionsFor(p Principal) (Permissions, error) { // directly from the stream and followed live. Nothing else's. sub = append(sub, MembershipSubject(p.Node, p.Module)) pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+"."+MembershipSubject(p.Node, p.Module)) + // And from the grants part's state memberships, its own key (novox/hq ADR 0571 clause 7): the read that replaces + // the stream's at the lease write, when the grants part stops publishing there. + pub = append(pub, readGrants(GrantsMembershipsBucket, MembershipKey(p.Node, p.Module))...) // 1b. The tools it calls, if its manifest says it calls any (novox/hq ADR 0152). The same // grant a person gets and derived the same way, so "what may this module ask" is @@ -1180,6 +1185,7 @@ func PermissionsFor(p Principal) (Permissions, error) { sub = append(sub, stateCreatedHeard(kept)...) // And its logs (novox/hq ADR 0297): appended to and read directly, its own alone. pub = append(pub, logGrants(p.Module, p.Logs)...) + pub = append(pub, readLogGrants(p.ReadLogs)...) // 6. Its traffic on seats that name their caller or their kind, ask proofs or keep records // (novox/hq ADR 0259 §3). @@ -1227,6 +1233,7 @@ func PermissionsFor(p Principal) (Permissions, error) { // that varies is the module, so the pattern is the machine's own assignments. sub = append(sub, "mesh.assignment."+p.Node+".*") pub = append(pub, "$JS.API.DIRECT.GET."+AssignmentsStream+".mesh.assignment."+p.Node+".*") + pub = append(pub, readGrants(GrantsMembershipsBucket, p.Node+".*")...) // And that it is there (novox/hq to-be 45 §3, S11): its own heartbeat, under its machine's // name and no other's, on core NATS like the host's. pub = append(pub, "mesh.control."+p.Node+".tools-alive") @@ -1283,6 +1290,7 @@ func PermissionsFor(p Principal) (Permissions, error) { // runtime's to keep, from the logs each membership lists. for _, d := range p.Carries { pub = append(pub, logGrants(d.Module, logNames(d.Logs))...) + pub = append(pub, readLogGrants(d.ReadLogs)...) } // **And the event families a session may wait on** (novox/hq ADR 0367 §2): the tool runner's module's own, // named one by one in its manifest, as a plain subscription — no consumer is made for them, and no @@ -1728,6 +1736,18 @@ func CreateConsumerGrants(consumers []ModuleConsumer) []string { // part, which composes them from the manifests and placement. It defines no stream and is granted no other subject here. const GrantsModule = "mesh-grants" +// The grants part's state memberships (novox/hq ADR 0571 clause 7): one key per assignment, `.`, its +// membership (Membership) as the assignments stream carries it, written only when it differs. A module's runtime reads +// its own key, the tool runner its node's; the grants part alone writes it. +const ( + GrantsMembershipsState = "memberships" + GrantsMembershipsBucket = GrantsModule + "_" + GrantsMembershipsState + GrantsMembershipsRead = GrantsModule + "." + GrantsMembershipsState +) + +// MembershipKey is an assignment's key in the grants part's memberships: `.`. +func MembershipKey(node, module string) string { return node + "." + module } + // The bus's own events, imported into MESH from the default system account `$SYS` under subjects of MESH's own // (novox/hq ADR 0431 §1, issue 1147): the server says an account's connections opening and closing only to the system // account, and the list of an account's connections only to a request made there. `$SYS` holds no user, so nobody diff --git a/go/buscompose/users.go b/go/buscompose/users.go index ca23cf5..c370f5d 100644 --- a/go/buscompose/users.go +++ b/go/buscompose/users.go @@ -64,6 +64,9 @@ type Declared struct { // ReadsRefused are the reads its manifest names that the owner's announcement does not make readable (novox/hq ADR // 0495 clause 3, issue 1790): granted nothing, kept to be said. ReadsRefused []string + // ReadLogs are, of its reads, those naming another module's log (`.`, novox/hq ADR 0571 clause 1): filled + // by Users and PlacementsOf from the logs the modules placed anywhere declare, never by a caller. + ReadLogs []string // KeyedReads are keys of other modules' state it reads, each one key alone: what a seat's holder is // granted for the pieces the modules beside it offer (novox/hq ADR 0260). KeyedReads []KeyedRead @@ -157,6 +160,7 @@ func Users(r Records) ([]Principal, error) { AsksTheConfig: servedOf(r, VerbsTheControllerAsksTheConfig), FilesFor: sortedCopy(r.Nodes)}} served := servedReadsOf(r) + logs := DeclaredLogs(r) grants := -1 for _, node := range sortedCopy(r.Nodes) { witness := false @@ -198,7 +202,7 @@ func Users(r Records) ([]Principal, error) { for _, d := range r.Assigned[node] { // A state with named readers is read by them alone (ADR 0547 clause 9): a read named for another module is // granted nothing, here and in the runtime carrying the module. - d = withNamedReaders(d) + d = withReadLogs(withNamedReaders(d), logs) // **A process account is its own, delivered in its own secret** (broker-, novox/hq // ADR 0436 §1, §2), never the module's `broker` — so it is composed whether or not the module // itself has an account, and is never carried by the machine's runtime, which runs as the @@ -226,7 +230,7 @@ func Users(r Records) ([]Principal, error) { Kind: KindModule, Node: node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, - State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, + State: stateNames(d.State), Reads: d.Reads, ReadLogs: d.ReadLogs, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues, }) @@ -250,7 +254,7 @@ func Users(r Records) ([]Principal, error) { var carried []Declared for _, d := range r.Assigned[node] { if d.RunsAs == "" && runnerServes(d) { - d = withNamedReaders(d) + d = withReadLogs(withNamedReaders(d), logs) // Never a restricted stream for what the tool runner carries (novox/hq ADR 0376 §1). d.Hears = ComposedHears(d) carried = append(carried, d) diff --git a/go/catalogue/seats_declared.go b/go/catalogue/seats_declared.go index 90087e2..b8342e7 100644 --- a/go/catalogue/seats_declared.go +++ b/go/catalogue/seats_declared.go @@ -285,6 +285,8 @@ func CatalogueProblems(shelf Shelf) []string { // Every offer says its credential's form, never left to a default (the lead, 2026-10-29, for novox/hq ADR 0570). for _, module := range shelfOrder(shelf) { problems = append(problems, CredentialFormProblems(shelf[module])...) + // And its data in the bus's predefined forms, a stream of its own refused (the lead, 2026-10-29). + problems = append(problems, PredefinedFormProblems(shelf[module])...) } // Who declares what, and who declared it first. diff --git a/go/catalogue/streams.go b/go/catalogue/streams.go index 15c6c2b..fa166e2 100644 --- a/go/catalogue/streams.go +++ b/go/catalogue/streams.go @@ -630,3 +630,31 @@ func uncovered(patterns []string) []string { slices.Sort(out) return out } + +// **A module keeps its data in the bus's predefined forms** (the lead and the operator, 2026-10-29; the operator's rule of +// 2026-10-22, "data storage is done using our predefined nats bus options"): a state, a log or an object store, which the +// runtime reads and writes for its tools and other modules read by a declared `reads`. A stream of its own is outside +// them and refused, naming it. StreamsKeptUntilMoved are the streams declared before this check, each allowed by name +// until its module moves it; the list only shrinks. +var StreamsKeptUntilMoved = map[string][]string{ + "brother": {"device-asks"}, + "mesh-build-go": {"toolchains-go"}, + "mesh-build-node": {"toolchains-node"}, + "mesh-mail": {"mail", "mail-sends"}, + "only-office": {"convert-asks"}, + "mesh-delivery-assigner": {"placement"}, + "mesh-delivery-checker": {"checks"}, + "mesh-delivery-forge-bridge": {"forge"}, +} + +// PredefinedFormProblems refuses each stream a module declares that StreamsKeptUntilMoved does not name for it. +func PredefinedFormProblems(m Manifest) []string { + var out []string + for _, s := range m.Streams { + if !slices.Contains(StreamsKeptUntilMoved[m.Module], s.Name) { + out = append(out, fmt.Sprintf("%s keeps the stream %q, outside the bus's predefined forms: keep it as a state, "+ + "a log or an object store, which the runtime reads for the module's tools and others read by `reads`", m.Module, s.Name)) + } + } + return out +} diff --git a/go/catalogue/streams_test.go b/go/catalogue/streams_test.go index a5971e3..cb02362 100644 --- a/go/catalogue/streams_test.go +++ b/go/catalogue/streams_test.go @@ -96,6 +96,14 @@ func TestTwoModulesCannotDeclareOneStream(t *testing.T) { if got := StreamNameProblems(Shelf{"a": a}); len(got) != 0 { t.Fatalf("one owner refused: %v", got) } + // And a stream of its own is outside the bus's predefined forms (the lead, 2026-10-29): refused, naming it, unless it + // is one kept by name until its module moves it. + if !strings.Contains(got, `a keeps the stream "mail", outside the bus's predefined forms`) { + t.Fatalf("a stream of its own was not refused: %s", got) + } + if got := PredefinedFormProblems(Manifest{Module: "mesh-mail", Streams: a.Streams}); len(got) != 0 { + t.Fatalf("a stream kept until it moves was refused: %v", got) + } } // The setting's value names who hears each key: a module wherever it runs, or on one machine (ADR 0363 §4). diff --git a/go/delivery/control_records_test.go b/go/delivery/control_records_test.go index 94683f4..2c64037 100644 --- a/go/delivery/control_records_test.go +++ b/go/delivery/control_records_test.go @@ -58,9 +58,13 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) { } // ADR 0571 clause 4: who holds which seat, from placement; a node seat by its node. held := SeatHoldersOf(map[string]Placement{ - "ace": {Node: "ace", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "node-files", Scope: "node", Module: "mesh-files"}}}, - "novox": {Node: "novox", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "mesh-broker", Scope: "mesh", Module: "nats"}}}, + "ace": {Node: "ace", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "node-files", Scope: "node", Module: "mesh-files"}}}, + "novox": {Node: "novox", Holds: []Holding{{Claim: "mesh-dns-resolver", Scope: "mesh", Module: "dnsmasq"}, {Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, + {Claim: "photo-gallery", Scope: "mesh", Module: "photos"}}}, }) + if o := held["photo-gallery"]; !o.Outside || len(o.Holders) != 1 || held["mesh-broker"].Outside { + t.Errorf("a claim outside the set is %+v", o) + } keys := map[string][]byte{} for k, v := range held { keys[k], _ = json.Marshal(v) @@ -68,9 +72,19 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) { if h, err := HoldersOf(keys, "mesh-dns-resolver", ""); err != nil || len(h) != 2 || h[0].Node != "ace" { t.Errorf("a replicated seat's holders are %v, %v", h, err) } - if h, _ := HoldersOf(keys, "node-files", "ace"); len(h) != 1 || h[0].Module != "mesh-files" || len(held) != 3 { + if h, _ := HoldersOf(keys, "node-files", "ace"); len(h) != 1 || h[0].Module != "mesh-files" || len(held) != 4 { t.Errorf("a node seat's holder is %v, keys %v", h, held) } + // ADR 0571 clauses 4, 5: the modules and the machines, from the indexer's and the placement seat's keys. + if m, err := ModuleNames(map[string][]byte{"letta": nil, "gitea": nil}); err != nil || strings.Join(m, ",") != "gitea,letta" { + t.Errorf("the modules are %v, %v", m, err) + } + if _, err := ModuleNames(map[string][]byte{"a.b": nil}); err == nil { + t.Error("a key that is no module's was read as one") + } + if n := MachineNames(map[string][]byte{"placement.novox": nil, "placement.ace": nil, "proposal.p1": nil}); strings.Join(n, ",") != "ace,novox" { + t.Errorf("the machines are %v", n) + } if h, _ := HoldersOf(keys, "node-files", "novox"); h != nil { t.Errorf("a node seat nobody holds there is held by %v", h) } @@ -79,7 +93,8 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) { start := BuildLogEntry{Event: BuildStartedEvent, Module: "letta", FP: "src2:a", Source: "abc", Repository: "novox/mesh-catalog", Directory: "modules/letta", Commit: "abc123", Node: "ace", At: at} end := start - end.Event, end.At, end.Outcome, end.Artifacts = BuildEndedEvent, at.Add(3*time.Minute), "done", []BuiltDigest{{Name: "image", Digest: "sha256:1"}} + end.Event, end.At, end.Outcome, end.Manifest, end.Against = BuildEndedEvent, at.Add(3*time.Minute), "done", "sha256:m", []string{"artifact-store://base/image@sha256:b"} + end.Made = []BuiltArtifact{{Name: "image", Kind: "image", Digest: "sha256:1", Reference: "artifact-store://letta/image@sha256:1"}} for _, e := range []BuildLogEntry{start, end} { raw, err := EncodeBuildLogEntry(e) back, derr := DecodeBuildLogEntry(raw) @@ -91,7 +106,24 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) { if d, ok := BuildDuration(start, end); !ok || d != 3*time.Minute { t.Errorf("the build took %s, %v", d, ok) } - for _, edit := range []func(*BuildLogEntry){func(e *BuildLogEntry) { e.Outcome = "" }, func(e *BuildLogEntry) { e.Commit = "" }} { + older := end + older.At, older.FP = at.Add(-time.Hour), "src2:b" + other := end + other.Module = "mesh-catalog" + if n := NewestPerModule([]BuildLogEntry{older, end, other}); n["letta"].FP != "src2:a" || len(n) != 2 { + t.Errorf("each module's newest is %v", n) + } + if e := EndedSince([]BuildLogEntry{end, start, older}, at.Add(-2*time.Hour)); len(e) != 2 || e[0].FP != "src2:b" { + t.Errorf("the ended builds since are %v", e) + } + if k, err := BuildLogKey("letta"); k != "letta" || err != nil || BuildLogRead != "mesh-delivery-builder.build-log" { + t.Errorf("the key is %q, %v", k, err) + } + if _, err := BuildLogKey("a.b"); err == nil { + t.Error("a module with a dot was taken as a key") + } + for _, edit := range []func(*BuildLogEntry){func(e *BuildLogEntry) { e.Outcome = "" }, func(e *BuildLogEntry) { e.Commit = "" }, + func(e *BuildLogEntry) { e.Event, e.Outcome, e.Against = BuildStartedEvent, "", nil }} { bad := end edit(&bad) if _, err := EncodeBuildLogEntry(bad); err == nil { diff --git a/go/delivery/history_records.go b/go/delivery/history_records.go index 6284549..714ff85 100644 --- a/go/delivery/history_records.go +++ b/go/delivery/history_records.go @@ -5,6 +5,8 @@ import ( "fmt" "sort" "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" ) // **Who holds which seat, and the builder's log** (novox/hq ADR 0571 clauses 1 and 4, issue 1955). @@ -24,12 +26,15 @@ type SeatHolder struct { } // SeatHolders is one key of seat-holders: the seat, its scope, the node for a node seat's key, and its holders -// (several only for a replicated mesh seat), sorted. +// (several only for a replicated mesh seat), sorted. Outside says the claim names no seat of the set (catalogue.Seats): +// a module registered before the set closed, or built outside the indexed repositories, keeps its claim listed, as the +// controller's `seats` listed it under outside. type SeatHolders struct { V int `json:"v"` Seat string `json:"seat"` Scope string `json:"scope"` Node string `json:"node,omitempty"` + Outside bool `json:"outside,omitempty"` Holders []SeatHolder `json:"holders"` } @@ -49,7 +54,8 @@ func SeatHoldersKey(seat, node string) (string, error) { } // SeatHoldersOf is the state composed from every node's placement (ADR 0567 §1): each holding by the seat it claims, -// keyed as SeatHoldersKey says, with the module holding it and its node. Deterministic, so a writer writes only a key +// keyed as SeatHoldersKey says, with the module holding it and its node. A claim naming a seat's former name is keyed by +// the seat's name (ADR 0122); one naming no seat is kept, marked Outside. Deterministic, so a writer writes only a key // that differs. func SeatHoldersOf(placements map[string]Placement) map[string]SeatHolders { out := map[string]SeatHolders{} @@ -64,12 +70,17 @@ func SeatHoldersOf(placements map[string]Placement) map[string]SeatHolders { if h.Scope == "node" { node = n } - key, err := SeatHoldersKey(h.Claim, node) + seat, known := catalogue.SeatNamed(h.Claim) + name := h.Claim + if known { + name = seat.Name + } + key, err := SeatHoldersKey(name, node) if err != nil { continue } sh := out[key] - sh.V, sh.Seat, sh.Scope, sh.Node = Version, h.Claim, h.Scope, node + sh.V, sh.Seat, sh.Scope, sh.Node, sh.Outside = Version, name, h.Scope, node, !known sh.Holders = append(sh.Holders, SeatHolder{Module: h.Module, Node: n}) out[key] = sh } @@ -110,39 +121,89 @@ func HoldersOf(keys map[string][]byte, seat, node string) ([]SeatHolder, error) return nil, nil } -// The builder's log (clause 1): a stream of its own, `log`, one entry when a build of a fingerprint starts and one when -// it ends, kept 30 days. +// The builder's log (clause 1; the lead and the operator, 2026-10-29): the builder's own module log (ADR 0297) `build-log`, +// kept 30 days (max-age-days), one entry under the key `` when a build of a fingerprint starts and one when it +// ends, so the newest entry of a key is the module's newest start or end. The builder's tools read it through the +// runtime's log verbs; another module reads it by naming BuildLogRead in its manifest's `reads`: a bundle through the +// runtime (stdio.Log(BuildLogRead)), a process on its own credential (process.Conn.ReadLog), each by newest per key, +// from a time, or following new entries. The catalogue module asks the builder's tool BuildsSinceTool instead. const ( - BuilderLogStream = "log" + BuilderLog = "build-log" + BuildLogRead = BuilderModule + "." + BuilderLog BuildStartedEvent = "build.started" BuildEndedEvent = "build.ended" // BuilderLogDays is how long the log keeps its entries. BuilderLogDays = 30 + // BuildsSinceTool is the builder's tool answering every ended build since a time, in order (clause 9): its + // argument `since` (RFC 3339), its answer BuildsSince. + BuildsSinceTool = "builder_builds_since" ) -// BuiltDigest is one artifact a build ended with, by its name and digest. -type BuiltDigest struct { - Name string `json:"name"` - Digest string `json:"digest"` +// BuildLogKey is the key of a module's entries in the builder's log: the module's name, refused when a log key could not +// hold it. +func BuildLogKey(module string) (string, error) { + if err := checkName("module", module); err != nil { + return "", err + } + return module, nil } // BuildLogEntry is one entry of the builder's log: the module, the fingerprint, the source revision, where the module is -// built from (its repository, its directory there and the commit), the build node and the time; at the end, the outcome -// and each artifact's digest. The repository, directory and commit are what the indexer's PathOf, the outside-built -// registrations and the assigner's running commits read (the lead, 2026-10-29). +// built from (its repository, its directory there, the commit and the ref), the build node and the time; at the end, +// the outcome, the manifest built (its digest, read from the indexer's state `manifests` under ManifestsKey), what it was +// built against (the bases' references, as the controller's built-before said them) and each artifact made +// (BuiltArtifact, the done key's: its name, kind, digest and reference). The repository, directory and commit are what +// the indexer's PathOf, the outside-built registrations and the assigner's running commits read; the manifest, against +// and made what the catalogue module registers a build from (the lead, 2026-10-29). type BuildLogEntry struct { - V int `json:"v"` - Event string `json:"event"` - Module string `json:"module"` - FP string `json:"fp"` - Source string `json:"source"` - Repository string `json:"repository"` - Directory string `json:"directory"` - Commit string `json:"commit"` - Node string `json:"node"` - At time.Time `json:"at"` - Outcome string `json:"outcome,omitempty"` - Artifacts []BuiltDigest `json:"artifacts,omitempty"` + V int `json:"v"` + Event string `json:"event"` + Module string `json:"module"` + FP string `json:"fp"` + Source string `json:"source"` + Repository string `json:"repository"` + Directory string `json:"directory"` + Commit string `json:"commit"` + Ref string `json:"ref,omitempty"` + Node string `json:"node"` + At time.Time `json:"at"` + Outcome string `json:"outcome,omitempty"` + Manifest string `json:"manifest,omitempty"` + Against []string `json:"against,omitempty"` + Made []BuiltArtifact `json:"made,omitempty"` +} + +// BuildsSince is the answer of BuildsSinceTool: every ended build since a time, oldest first, and whether the log held +// more than one answer carries (then ask again from the last one's time). +type BuildsSince struct { + Since time.Time `json:"since"` + Builds []BuildLogEntry `json:"builds"` + More bool `json:"more,omitempty"` +} + +// NewestPerModule is each module's newest entry of entries (one kind, read from the last per subject or whole), by +// module: the reader's helper for the five newest and the running commits. +func NewestPerModule(entries []BuildLogEntry) map[string]BuildLogEntry { + out := map[string]BuildLogEntry{} + for _, e := range entries { + if have, ok := out[e.Module]; !ok || e.At.After(have.At) { + out[e.Module] = e + } + } + return out +} + +// EndedSince is, of entries, the ended builds at or after since, oldest first: the reader's helper for history and for +// BuildsSinceTool's answer. +func EndedSince(entries []BuildLogEntry, since time.Time) []BuildLogEntry { + var out []BuildLogEntry + for _, e := range entries { + if e.Event == BuildEndedEvent && !e.At.Before(since) { + out = append(out, e) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) }) + return out } // EncodeBuildLogEntry is e as JSON at Version; it refuses an entry a reader could not place. @@ -155,8 +216,8 @@ func EncodeBuildLogEntry(e BuildLogEntry) ([]byte, error) { return nil, fmt.Errorf("delivery: a build log entry names its module, fingerprint, repository, commit, node and time") case e.Event == BuildEndedEvent && e.Outcome == "": return nil, fmt.Errorf("delivery: a build's end names its outcome") - case e.Event == BuildStartedEvent && (e.Outcome != "" || len(e.Artifacts) > 0): - return nil, fmt.Errorf("delivery: a build's start names no outcome and no artifact") + case e.Event == BuildStartedEvent && (e.Outcome != "" || len(e.Made) > 0 || e.Manifest != "" || len(e.Against) > 0): + return nil, fmt.Errorf("delivery: a build's start names no outcome, manifest, base or artifact") } return json.Marshal(e) } @@ -182,3 +243,35 @@ func BuildDuration(start, end BuildLogEntry) (time.Duration, bool) { } return end.At.Sub(start.At), true } + +// **Which modules exist and which machines the mesh has, from the bus** (novox/hq ADR 0571 clauses 4, 5; the lead for a6, +// 2026-10-29): the tool runner reads the modules from the indexer's state `modules` and the machines from the placement +// seat's keys, and no longer asks the controller's `modules` or `nodes`. + +// ModuleNames is every module the indexer holds, sorted, from its state `modules` as a reader holds its keys (key to +// value; a deleted key absent). A key that is not a module's is refused, naming it. +func ModuleNames(keys map[string][]byte) ([]string, error) { + out := make([]string, 0, len(keys)) + for k := range keys { + module, err := ParseModulesKey(k) + if err != nil { + return nil, err + } + out = append(out, module) + } + sort.Strings(out) + return out, nil +} + +// MachineNames is every machine the placement seat places modules on, sorted, from its state `placement` as a reader +// holds its keys: each `placement.`. Keys of the state's other kinds (a proposal's) are passed over. +func MachineNames(keys map[string][]byte) []string { + out := []string{} + for k := range keys { + if node, err := ParsePlacementKey(k); err == nil { + out = append(out, node) + } + } + sort.Strings(out) + return out +} diff --git a/go/log.go b/go/log.go index 7b1030d..aad2169 100644 --- a/go/log.go +++ b/go/log.go @@ -9,8 +9,9 @@ import ( // Log — a module's record of operations on the bus, kept as long as the log is (novox/hq ADR 0297). Not the // current value: that is state, and a log's state is its fold. Each entry is one operation on one key, its // sequence in the log is its identity, and the order of the log is the order of the operations. A module -// declares each log it owns in its manifest (`logs`) and names it here as it named it there; a log is its -// owner's alone, and the runtime refuses, with the reason, one it did not issue. +// declares each log it owns in its manifest (`logs`) and names it here as it named it there; only its owner appends. +// Another module's log is read by naming it `.` in the manifest's `reads` and here (novox/hq ADR 0571 +// clause 1): read, newest per key, from a time, and followed. The runtime refuses, with the reason, one it did not issue. // LogEntry is one operation on one key, as the log holds it. type LogEntry struct { @@ -40,7 +41,7 @@ type LogInfo struct { // KeptLog is one log this module owns. type KeptLog struct{ name string } -// Log is a log this module owns, by its local name. +// Log is a log this module owns, by its local name, or reads, `.`. func Log(name string) KeptLog { return KeptLog{name: name} } const logTimeout = 30 * time.Second @@ -121,3 +122,102 @@ func (l KeptLog) Info() (*LogInfo, error) { var i LogInfo return &i, json.Unmarshal(raw, &i) } + +// Newest is the newest entry of every key keys names (a key, `*` for one name, `**` for the rest; empty is every key), +// in the log's order: runtime verb `mesh/log.newest` {log, keys}, answered as a list of entries. +func (l KeptLog) Newest(keys string) ([]LogEntry, error) { + params := map[string]any{} + if keys != "" { + params["keys"] = keys + } + raw, err := l.ask("newest", params) + if err != nil { + return nil, err + } + var out []LogEntry + return out, json.Unmarshal(raw, &out) +} + +// Since is one page of the entries from a time on, as Read pages them: runtime verb `mesh/log.read` with `since` (RFC +// 3339) in place of `from`. +func (l KeptLog) Since(at time.Time, keys string, limit int) (*LogPage, error) { + if limit < 0 { + return nil, fmt.Errorf("a read of %s asks for at least no entries: limit %d", l.name, limit) + } + params := map[string]any{"since": at.UTC().Format(time.RFC3339Nano)} + if keys != "" { + params["keys"] = keys + } + if limit > 0 { + params["limit"] = limit + } + raw, err := l.ask("read", params) + if err != nil { + return nil, err + } + var p LogPage + return &p, json.Unmarshal(raw, &p) +} + +// LogFrom is where a log's watch starts: LogFromNewest (the newest entry of each key, then every new one), LogFromAll +// (every entry the log keeps), LogFromSince (every entry from a time) or LogFromNow (only new entries). +type LogFrom struct { + Kind string + Since time.Time +} + +// The kinds of LogFrom. +const ( + LogFromNewestKind = "newest" + LogFromAllKind = "all" + LogFromSinceKind = "since" + LogFromNowKind = "now" +) + +var ( + LogFromNewest = LogFrom{Kind: LogFromNewestKind} + LogFromAll = LogFrom{Kind: LogFromAllKind} + LogFromNow = LogFrom{Kind: LogFromNowKind} +) + +// LogFromSince starts a watch at a time. +func LogFromSince(at time.Time) LogFrom { return LogFrom{Kind: LogFromSinceKind, Since: at} } + +func (f LogFrom) params() (map[string]any, error) { + p := map[string]any{"from": f.Kind} + switch f.Kind { + case LogFromNewestKind, LogFromAllKind, LogFromNowKind: + case LogFromSinceKind: + p["since"] = f.Since.UTC().Format(time.RFC3339Nano) + default: + return nil, fmt.Errorf("a log's watch starts from newest, all, since or now, not %q", f.Kind) + } + return p, nil +} + +// Watch hands handle each entry of keys from where from says, in the log's order, and then every new one as it is +// appended, and returns once the entries there at its start have been handled: runtime verb `mesh/log.watch` {log, +// keys, from, since}, each entry then sent as `mesh/log` {watch, entry}. Event-driven: nothing reads the log again on +// a timer. The watch lasts as long as the process. +func (l KeptLog) Watch(keys string, from LogFrom, handle func(LogEntry) error) error { + params, err := from.params() + if err != nil { + return err + } + c, err := live() + if err != nil { + return err + } + id := c.newID() + c.mu.Lock() + c.logWatches[id] = handle + c.mu.Unlock() + params["log"], params["keys"] = l.name, keys + if _, err := c.askAs(id, "mesh/log.watch", params, logTimeout); err != nil { + c.mu.Lock() + delete(c.logWatches, id) + c.mu.Unlock() + return err + } + return nil +} diff --git a/go/log_test.go b/go/log_test.go index b650925..98bba28 100644 --- a/go/log_test.go +++ b/go/log_test.go @@ -58,12 +58,13 @@ func (r *logRuntime) answer(m message) map[string]any { From uint64 `json:"from"` Keys string `json:"keys"` Limit int `json:"limit"` + Since time.Time `json:"since"` } _ = json.Unmarshal(m.Params, &p) r.mu.Lock() defer r.mu.Unlock() r.asked = append(r.asked, m) - if p.Log != "changes" { + if p.Log != "changes" && p.Log != "builder.build-log" { return map[string]any{"error": map[string]any{"code": -32000, "message": "issues was issued no log " + p.Log}} } switch strings.TrimPrefix(m.Method, "mesh/log.") { @@ -83,7 +84,7 @@ func (r *logRuntime) answer(m message) map[string]any { page := map[string]any{"entries": []LogEntry{}, "next": 0, "last": len(r.entries)} var got []LogEntry for _, e := range r.entries { - if e.Sequence < p.From || !matches(p.Keys, e.Key) { + if e.Sequence < p.From || !matches(p.Keys, e.Key) || e.At.Before(p.Since) { continue } if len(got) == limit { @@ -96,6 +97,16 @@ func (r *logRuntime) answer(m message) map[string]any { page["entries"] = got } return map[string]any{"result": page} + case "newest": + var out []LogEntry + for _, e := range r.entries { + if matches(p.Keys, e.Key) && r.last(e.Key) == e.Sequence { + out = append(out, e) + } + } + return map[string]any{"result": out} + case "watch": + return map[string]any{"result": map[string]any{}} case "last": seq := r.last(p.Key) if seq == 0 { @@ -148,6 +159,13 @@ func logging(t *testing.T, r *logRuntime, run func() (any, error)) map[string]an a := r.answer(m) a["id"] = m.ID send(a) + // A watch's entries follow its answer, each named by the watch's id (ADR 0571 clause 1). + if m.Method == "mesh/log.watch" { + r.mu.Lock() + e := r.entries[len(r.entries)-1] + r.mu.Unlock() + send(map[string]any{"id": "entry-1", "method": "mesh/log", "params": map[string]any{"watch": m.ID, "entry": e}}) + } }(m) continue } @@ -193,7 +211,9 @@ func TestABundleKeepsALogThroughTheRuntime(t *testing.T) { pages []*LogPage last, none *LogEntry info *LogInfo - narrowed *LogPage + narrowed, since *LogPage + newest []LogEntry + watched *LogEntry ) got := logging(t, r, func() (any, error) { changes := Log("changes") @@ -232,8 +252,27 @@ func TestABundleKeepsALogThroughTheRuntime(t *testing.T) { if none, err = changes.Last("999"); err != nil { return nil, err } - info, err = changes.Info() - return "done", err + if info, err = changes.Info(); err != nil { + return nil, err + } + // ADR 0571 clause 1: another module's log, newest per key, from a time, and followed. + if newest, err = changes.Newest(""); err != nil { + return nil, err + } + if since, err = changes.Since(time.Date(2026, 10, 11, 18, 0, 5, 0, time.UTC), "", 0); err != nil { + return nil, err + } + heard := make(chan LogEntry, 1) + if err = Log("builder.build-log").Watch("letta", LogFromSince(time.Date(2026, 10, 11, 18, 0, 0, 0, time.UTC)), + func(e LogEntry) error { heard <- e; return nil }); err != nil { + return nil, err + } + select { + case e := <-heard: + watched = &e + case <-time.After(5 * time.Second): + } + return "done", nil }) if out, isError := text(t, got); isError || out != `"done"` { t.Fatalf("the tool answered %s", out) @@ -296,6 +335,13 @@ func TestABundleKeepsALogThroughTheRuntime(t *testing.T) { if none != nil { t.Fatalf("a key with no entry has a last: %+v", none) } + if len(newest) != 5 || newest[0].Key != "402" || newest[1].Sequence != 3 || len(since.Entries) != 2 || since.Entries[0].Sequence != 5 { + t.Fatalf("the newest per key are %+v, since a time %+v", newest, since) + } + if method, p = r.params(t, 15); method != "mesh/log.watch" || p["log"] != "builder.build-log" || p["keys"] != "letta" || + p["from"] != "since" || p["since"] != "2026-10-11T18:00:00Z" || watched == nil || watched.Sequence != 6 { + t.Fatalf("a watch asked %s %v and heard %+v", method, p, watched) + } if info == nil || info.Entries != 6 || info.MaxBytes != 1<<30 || info.First != 1 || info.Last != 6 || info.Bytes == 0 { t.Fatalf("the log's info is %+v", info) } diff --git a/go/process/fixture_integration_test.go b/go/process/fixture_integration_test.go index ce92cfa..91ed0a3 100644 --- a/go/process/fixture_integration_test.go +++ b/go/process/fixture_integration_test.go @@ -43,6 +43,14 @@ var sendGrants = []string{ "mesh.seat.operator-channel.call.tell.novox~mesh-mail~send", "mesh.seat.operator-channel.tool.tell", "mesh.seat.operator-channel.tool.tell.*", + // And the builder's log, as buscompose's readLogGrants grants a declared read of it (novox/hq ADR 0571 clause 1). + "$JS.API.STREAM.INFO.LOG_mesh-delivery-builder_build-log", + "$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log", + "$JS.API.DIRECT.GET.LOG_mesh-delivery-builder_build-log.>", + "$JS.API.CONSUMER.CREATE.LOG_mesh-delivery-builder_build-log.>", + "$JS.API.CONSUMER.MSG.NEXT.LOG_mesh-delivery-builder_build-log.*", + "$JS.API.CONSUMER.INFO.LOG_mesh-delivery-builder_build-log.*", + "$JS.API.CONSUMER.DELETE.LOG_mesh-delivery-builder_build-log.*", } // grantedSendProcess is a test bus with JetStream, with two users: novox.mesh-mail.send, permissioned to diff --git a/go/process/log.go b/go/process/log.go new file mode 100644 index 0000000..176c408 --- /dev/null +++ b/go/process/log.go @@ -0,0 +1,152 @@ +package process + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// **Another module's log, read on this process's own credential** (novox/hq ADR 0571 clause 1; the lead, 2026-10-29): a +// log the module's manifest names in `reads`, `.`, read through an ordered consumer this connection creates +// and lets go of, never a durable one. Newest is the newest entry of each key, Since every entry from a time, and Watch +// those or every entry, then each new one as it is appended: event-driven, nothing reads the log again on a timer. The +// grants are buscompose's readLogGrants, given for a declared read of a log alone. + +// LogReader is one module's log this connection reads. +type LogReader struct { + Module, Name string + js func() (jetstream.JetStream, error) + closed func() bool +} + +// ReadLog is module's log name, read on this connection. +func (c *Conn) ReadLog(module, name string) LogReader { + return LogReader{Module: module, Name: name, closed: func() bool { return c.nc.IsClosed() }, + js: func() (jetstream.JetStream, error) { return jetstream.New(c.nc) }} +} + +func (l LogReader) stream() string { return "LOG_" + l.Module + "_" + l.Name } +func (l LogReader) subject() string { return "mesh.log." + l.Module + "." + l.Name } + +// filter is keys as a subject filter: a key, `*` for one name, `**` for the rest; empty is every key. +func (l LogReader) filter(keys string) string { + if keys == "" { + return l.subject() + ".>" + } + return l.subject() + "." + watchPattern(keys) +} + +func (l LogReader) config(keys string, from stdio.LogFrom) (jetstream.OrderedConsumerConfig, error) { + cfg := jetstream.OrderedConsumerConfig{FilterSubjects: []string{l.filter(keys)}} + switch from.Kind { + case stdio.LogFromNewestKind: + cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy + case stdio.LogFromAllKind: + cfg.DeliverPolicy = jetstream.DeliverAllPolicy + case stdio.LogFromSinceKind: + at := from.Since + cfg.DeliverPolicy, cfg.OptStartTime = jetstream.DeliverByStartTimePolicy, &at + case stdio.LogFromNowKind: + cfg.DeliverPolicy = jetstream.DeliverNewPolicy + default: + return cfg, fmt.Errorf("%s's log %s: a read starts from newest, all, since or now, not %q", l.Module, l.Name, from.Kind) + } + return cfg, nil +} + +func (l LogReader) entry(m jetstream.Msg) (stdio.LogEntry, error) { + meta, err := m.Metadata() + if err != nil { + return stdio.LogEntry{}, err + } + return stdio.LogEntry{Sequence: meta.Sequence.Stream, Key: strings.TrimPrefix(m.Subject(), l.subject()+"."), + Value: append([]byte(nil), m.Data()...), At: meta.Timestamp}, nil +} + +// Newest is the newest entry of each key keys names, in the log's order. +func (l LogReader) Newest(keys string) ([]stdio.LogEntry, error) { + return l.collect(keys, stdio.LogFromNewest) +} + +// Since is every entry of keys from at on, in the log's order. +func (l LogReader) Since(at time.Time, keys string) ([]stdio.LogEntry, error) { + return l.collect(keys, stdio.LogFromSince(at)) +} + +func (l LogReader) collect(keys string, from stdio.LogFrom) ([]stdio.LogEntry, error) { + var out []stdio.LogEntry + stop, err := l.watch(keys, from, func(e stdio.LogEntry) error { out = append(out, e); return nil }, false) + if stop != nil { + stop() + } + return out, err +} + +// Watch hands handle each entry from where from says and then every new one, in the log's order, and returns once the +// entries there at its start have been handled. The watch lasts as long as the connection; an error from handle ends +// it, as the connection closing does. Before the entries at its start are handled, either is returned as the error. +func (l LogReader) Watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error) error { + _, err := l.watch(keys, from, handle, true) + return err +} + +func (l LogReader) watch(keys string, from stdio.LogFrom, handle func(stdio.LogEntry) error, follow bool) (func(), error) { + cfg, err := l.config(keys, from) + if err != nil { + return nil, err + } + js, err := l.js() + if err != nil { + return nil, err + } + bind, stopBind := context.WithTimeout(context.Background(), stateTimeout) + defer stopBind() + cons, err := js.OrderedConsumer(bind, l.stream(), cfg) + if err != nil { + return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err) + } + info, err := cons.Info(bind) + if err != nil { + return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err) + } + pending := info.NumPending + it, err := cons.Messages() + if err != nil { + return nil, fmt.Errorf("%s's log %s (stream %s): %w", l.Module, l.Name, l.stream(), err) + } + for ; pending > 0; pending-- { + m, err := it.Next() + if err == nil { + var e stdio.LogEntry + if e, err = l.entry(m); err == nil { + err = handle(e) + } + } + if err != nil { + it.Stop() + return nil, fmt.Errorf("%s's log %s: %w", l.Module, l.Name, err) + } + } + if !follow { + return it.Stop, nil + } + go func() { + defer it.Stop() + for { + m, err := it.Next() + if err != nil || (l.closed != nil && l.closed()) { + return + } + e, err := l.entry(m) + if err != nil || handle(e) != nil { + return + } + } + }() + return it.Stop, nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index 5c5f07f..7d86e70 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -274,6 +274,44 @@ func TestStateWatchOnAGrantedBucket(t *testing.T) { t.Fatalf("narrowed watch got %+v for a key it did not name", c) case <-time.After(500 * time.Millisecond): } + + // ADR 0571 clause 1: another module's log, read on the process's own credential with exactly the read grants: the + // newest per key, every entry from a time, and a watch from the newest that then hears an appended entry. + if _, err := adminJetStream.CreateStream(ctx, jetstream.StreamConfig{Name: "LOG_mesh-delivery-builder_build-log", + Subjects: []string{"mesh.log.mesh-delivery-builder.build-log.>"}, AllowDirect: true}); err != nil { + t.Fatal(err) + } + for _, k := range []string{"letta", "gitea", "letta"} { + if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log."+k, []byte(`{"module":"`+k+`"}`)); err != nil { + t.Fatal(err) + } + time.Sleep(5 * time.Millisecond) + } + builds := conn.ReadLog("mesh-delivery-builder", "build-log") + newest, err := builds.Newest("") + if err != nil || len(newest) != 2 || newest[0].Key != "gitea" || newest[1].Sequence != 3 { + t.Fatalf("the newest per key are %+v: %v", newest, err) + } + since, err := builds.Since(newest[0].At, "letta") + if err != nil || len(since) != 1 || since[0].Sequence != 3 { + t.Fatalf("letta since gitea's build is %+v: %v", since, err) + } + heard := make(chan stdio.LogEntry, 4) + if err := builds.Watch("letta", stdio.LogFromNewest, func(e stdio.LogEntry) error { heard <- e; return nil }); err != nil || len(heard) != 1 { + t.Fatalf("a watch from the newest handed %d at its start: %v", len(heard), err) + } + <-heard + if _, err := adminJetStream.Publish(ctx, "mesh.log.mesh-delivery-builder.build-log.letta", []byte(`{}`)); err != nil { + t.Fatal(err) + } + select { + case e := <-heard: + if e.Sequence != 4 || e.Key != "letta" { + t.Fatalf("the watch heard %+v", e) + } + case <-time.After(10 * time.Second): + t.Fatal("the watch did not hear the appended entry") + } } // 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket. diff --git a/go/stdio.go b/go/stdio.go index c3de051..82568f6 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -72,6 +72,8 @@ type channel struct { watches map[string]func(StateChange) error // provisioned are this bundle's watches of its values, by the id each was asked with (provision.go). provisioned map[string]func(ProvisionChange) error + // logWatches are this bundle's watches of a log, by the id each was asked with (log.go). + logWatches map[string]func(LogEntry) error // takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259). takers map[string]func(Work) error provers map[string]func(subject string, body json.RawMessage) (any, error) @@ -182,8 +184,8 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { byName[t.Name] = t } c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}, - provisioned: map[string]func(ProvisionChange) error{}, - takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} + provisioned: map[string]func(ProvisionChange) error{}, logWatches: map[string]func(LogEntry) error{}, + takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} currentMu.Lock() current = c currentMu.Unlock() @@ -315,6 +317,28 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { } c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Change) + case "mesh/log": + // One entry of a log this bundle watches (log.go), named by the watch's id, answered once its handler has + // taken it, in order: the runtime sends the next only after this answer. + var p struct { + Watch string `json:"watch"` + Entry *LogEntry `json:"entry"` + } + _ = json.Unmarshal(m.Params, &p) + c.mu.Lock() + handle := c.logWatches[p.Watch] + c.mu.Unlock() + if handle == nil || p.Entry == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/log names no watch of this bundle"}}) + continue + } + go func(id any, e LogEntry) { + if err := handle(e); err != nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) + return + } + c.say(map[string]any{"id": id, "result": map[string]any{}}) + }(id, *p.Entry) case "mesh/provision": // One change to a value of this module's that this bundle watches (provision.go), named by the // watch's id, answered once its handler has taken it. Beside the loop, like a tool call.