diff --git a/cmd/mesh-controller/build.go b/cmd/mesh-controller/build.go index 9331dec0..2997292d 100644 --- a/cmd/mesh-controller/build.go +++ b/cmd/mesh-controller/build.go @@ -820,6 +820,9 @@ type answers struct { sources map[string]inventory.Source // waiting is every machine not running what the mesh would send it. waiting []inventory.Machine + // recordedWaits is, per machine, every move of a recorded module that waits there for a person (novox/hq ADR + // 0377 §6): the machine is behind for these, and `push ` sends them. + recordedWaits map[string][]string // reported is every machine's last word beside when it was last sent a declaration — the // pair that answers "has it caught up", which waiting alone cannot (the sent digest is // recorded at send, not at apply). diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index 0700d79b..f56bdc47 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/cmd/mesh-controller/busobjects.go @@ -49,10 +49,12 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra // composed from, so a module the mesh grants a consumer's subjects has that consumer waiting. // Done on every raise, not only when a credential is issued: every module moved onto this bus // by the rollout was issued on the old one, and came up with nothing to bind to (2026-09-28). - consumers, err := moduleConsumers(ctx, inv) + consumers, records, err := moduleConsumersAndRecords(ctx, inv) if err != nil { return nil, err } + // And a grant of a restricted stream that composes nothing, said on every raise and send (novox/hq ADR 0376 §2). + sayWithheld(records, " ") // And the work queues of seats that name their caller or their kind, with each holder's worker // (novox/hq ADR 0259 §3): an ask queues until the router takes it, a channel's work until that kind // takes it. @@ -165,15 +167,21 @@ func unassertedObservation(err error) conditions.Observation { // moduleConsumers is every module's durable consumer, from the records the user list is composed from. func moduleConsumers(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, error) { + consumers, _, err := moduleConsumersAndRecords(ctx, inv) + return consumers, err +} + +// moduleConsumersAndRecords is moduleConsumers with the records they were composed from. +func moduleConsumersAndRecords(ctx context.Context, inv *inventory.Inventory) ([]broker.ModuleConsumer, broker.Records, error) { records, err := inv.BusRecords(ctx) if err != nil { - return nil, err + return nil, broker.Records{}, err } users, err := broker.Users(records) if err != nil { - return nil, err + return nil, broker.Records{}, err } - return broker.ConsumersOf(users), nil + return broker.ConsumersOf(users), records, nil } // seatTrafficObjects is the work queues and workers of seats that name their caller or their kind, from diff --git a/cmd/mesh-controller/changeplan.go b/cmd/mesh-controller/changeplan.go index b895614d..2b3ae219 100644 --- a/cmd/mesh-controller/changeplan.go +++ b/cmd/mesh-controller/changeplan.go @@ -73,6 +73,11 @@ func changePlanOf(repository, base, head string, r mergeReach, entries []invento "(issue 280)", busUpgradeNeeded, name, busUpgradeVerb, strings.Join(e.On, ", "))) case name == "nats" || catalogue.ProvidesBus(e.Manifest): p.Steps = append(p.Steps, "a planned bus step: the bus is upgraded by `bus upgrade`, never by an ordinary send") + case waits && len(e.On) > 0 && (u.From == catalogue.FromModule || u.From == catalogue.FromData): + // Judged per machine by the walk, against the build each was last sent (novox/hq ADR 0377). + p.Steps = append(p.Steps, fmt.Sprintf("%s waits for a person: its policy records (%s) — unless on a machine "+ + "its move recreates none of its containers and moves none of its data, which the walk judges and then sends", + name, orNone(u.From))) case waits && len(e.On) > 0: p.Steps = append(p.Steps, fmt.Sprintf("%s waits for a person: its policy records (%s)", name, orNone(u.From))) diff --git a/cmd/mesh-controller/deadletters_aging.go b/cmd/mesh-controller/deadletters_aging.go new file mode 100644 index 00000000..e7b2b466 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging.go @@ -0,0 +1,117 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// A dead letter from a restricted stream ages out with its stream (novox/hq ADR 0376 §3, §6). +// +// DEAD_LETTERS lets such a message go when its stream's maximum age runs out, so a consumer's `max-deliveries` +// condition can clear with nobody having delivered it again or dropped it. That is said: the watchdogs remember, +// from one look to the next, when each restricted dead letter ages out, count those gone once their time passed, +// say each in the controller's log as it goes, and clear the condition with words saying how many aged out +// undelivered and from which stream. + +// agingTally is what the watchdogs remember of restricted dead letters between looks. +type agingTally struct { + // seen is each consumer's restricted dead letters at the last look, by id, with when each ages out. + seen map[string]map[uint64]time.Time + // agedOut is how many of each consumer's aged out undelivered while its condition was open. + agedOut map[string]int +} + +// look compares this look's restricted dead letters with the last: newly is how many of each consumer's aged out +// since, and cleared how many aged out in all for each consumer DEAD_LETTERS no longer holds anything of, when any +// did. held is every consumer's count this look; aging its restricted dead letters. +func (t *agingTally) look(now time.Time, held map[string]int, aging map[string][]link.Aging) (newly, cleared map[string]int) { + if t.seen == nil { + t.seen, t.agedOut = map[string]map[uint64]time.Time{}, map[string]int{} + } + newly, cleared = map[string]int{}, map[string]int{} + for key, before := range t.seen { + still := map[uint64]bool{} + for _, a := range aging[key] { + still[a.ID] = true + } + for id, agesOut := range before { + if !still[id] && !agesOut.After(now) { + newly[key]++ + } + } + if newly[key] == 0 { + delete(newly, key) + } + t.agedOut[key] += newly[key] + } + for key, n := range t.agedOut { + if held[key] > 0 { + continue + } + if n > 0 { + cleared[key] = n + } + delete(t.agedOut, key) + } + t.seen = map[string]map[uint64]time.Time{} + for key, letters := range aging { + t.seen[key] = map[uint64]time.Time{} + for _, a := range letters { + t.seen[key][a.ID] = a.AgesOut + } + } + return newly, cleared +} + +// agedOutResolved is the line a consumer's condition clears with when messages it gave up on aged out undelivered. +func agedOutResolved(key string, n int) string { + stream, _, _ := strings.Cut(key, ".") + messages := "a message" + if n > 1 { + messages = fmt.Sprintf("%d messages", n) + } + return fmt.Sprintf("Resolved: %s of the stream %s aged out undelivered", messages, stream) +} + +// sayAgedOut says, in the log, what aged out since the last look, and clears the condition of every consumer whose +// dead letters are all gone with the words saying how many aged out. Before the rows are reconciled, so the +// condition is cleared with these words and not its own. +func (w *watchdogs) sayAgedOut(ctx context.Context, f *signalFacts) []string { + if f.advisoriesErr != nil || f.deadLettersAging == nil { + return nil + } + newly, cleared := w.aging.look(f.now, f.deadLetters, f.deadLettersAging) + var problems []string + for _, key := range countedKeys(newly) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%d message(s) %s gave up on aged out of %s undelivered: %s keeps them only for its maximum age "+ + "(novox/hq ADR 0376)\n", newly[key], link.ConsumerInWords(stream, consumer), "DEAD_LETTERS", stream) + } + for _, key := range countedKeys(cleared) { + stream, consumer, _ := strings.Cut(key, ".") + fmt.Printf("%s no longer has a message kept: %d aged out of %s undelivered, so its condition clears "+ + "(novox/hq ADR 0376)\n", link.ConsumerInWords(stream, consumer), cleared[key], stream) + if _, err := w.keeper.ClearSaying(ctx, conditions.Key(conditions.ScopeBus, key, link.AdvisoryMaxDeliveries), + fmt.Sprintf("its %d dead letter(s) of %s aged out undelivered", cleared[key], stream), + agedOutResolved(key, cleared[key])); err != nil { + problems = append(problems, err.Error()) + } + } + return problems +} + +// countedKeys is a count's keys in order. +func countedKeys(m map[string]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/cmd/mesh-controller/deadletters_aging_test.go b/cmd/mesh-controller/deadletters_aging_test.go new file mode 100644 index 00000000..867f97b0 --- /dev/null +++ b/cmd/mesh-controller/deadletters_aging_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/link" +) + +// **A dead letter that ages out clears its condition and says so** (novox/hq ADR 0376 §6), on a fake clock: while +// DEAD_LETTERS holds a restricted stream's messages for a consumer its max-deliveries condition is open; once they +// have aged out it clears, with words saying how many aged out undelivered and from which stream. One dropped before +// its age is not counted as aged out, and a consumer of EVENTS clears as before. +func TestADeadLetterThatAgesOutClearsItsConditionAndSaysSo(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + store := conditions.NewInMemory() + told := &conditions.Told{} + k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: told, + Now: func() time.Time { return now }}) + w := &watchdogs{keeper: k, started: now.Add(-time.Hour)} + + look := func(at time.Time, held map[string]int, aging map[string][]link.Aging) { + f := calm(at) + f.deadLetters, f.deadLettersAging = held, aging + w.see(t.Context(), f) + } + mail := "MAIL.ace_mailer" + look(now, map[string]int{mail: 3, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 1, AgesOut: now.Add(time.Hour)}, {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + if open, _ := k.Open(t.Context()); len(open) != 2 { + t.Fatalf("open: %+v", open) + } + // One dropped by a person before its age: not aged out. + look(now.Add(30*time.Minute), map[string]int{mail: 2, "EVENTS.media_sonarr": 1}, map[string][]link.Aging{mail: { + {ID: 2, AgesOut: now.Add(2 * time.Hour)}, {ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + look(now.Add(150*time.Minute), map[string]int{mail: 1}, map[string][]link.Aging{mail: {{ID: 3, AgesOut: now.Add(3 * time.Hour)}}}) + open, _ := k.Open(t.Context()) + if len(open) != 1 || open[0].Key != conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries) { + t.Fatalf("after one aged out: %+v", open) + } + look(now.Add(4*time.Hour), map[string]int{}, map[string][]link.Aging{}) + if open, _ := k.Open(t.Context()); len(open) != 0 { + t.Fatalf("after all aged out: %+v", open) + } + k.Close(context.Background()) // everything said is said once it closes + var mailCleared, sonarrCleared string + for _, e := range told.Said() { + if e.Change != conditions.ChangeCleared { + continue + } + switch e.Condition.Key { + case conditions.Key(conditions.ScopeBus, mail, link.AdvisoryMaxDeliveries): + mailCleared = e.Condition.Resolved + case conditions.Key(conditions.ScopeBus, "EVENTS.media_sonarr", link.AdvisoryMaxDeliveries): + sonarrCleared = e.Condition.Resolved + } + } + if mailCleared != "Resolved: 2 messages of the stream MAIL aged out undelivered" { + t.Fatalf("MAIL's condition cleared with %q", mailCleared) + } + if sonarrCleared != "Resolved: the messages it gave up on were delivered again or dropped" { + t.Fatalf("an EVENTS consumer's condition cleared with %q", sonarrCleared) + } + if why, ok := conditions.Plain(mailCleared); !ok { + t.Fatalf("not plain: %s", why) + } +} diff --git a/cmd/mesh-controller/desk_secret.go b/cmd/mesh-controller/desk_secret.go index a0da69d4..67f242cb 100644 --- a/cmd/mesh-controller/desk_secret.go +++ b/cmd/mesh-controller/desk_secret.go @@ -54,6 +54,10 @@ type deskGive struct { trusted func(module string) (bool, error) // ask asks one machine's node-launcher.secret and answers its result, or the holder's refusal. ask func(machine string, args map[string]any) (json.RawMessage, error) + // promptFor is the one line the secret's declaration asks for it as (novox/hq issue 789): read from the + // controller's own catalogue, never from the call or the module's runtime. Nil, or an empty answer, asks by + // name alone, as the desk does today. + promptFor func(module, name string) (string, error) // accept seals the value as `secret accept` does, and says whether it lives until the module's start. accept func(value string) (untilStart bool, err error) // record writes the act in the hand-act log. @@ -153,15 +157,28 @@ func (d deskGive) give(node, module, name, desk string) (string, error) { } // By name, never by words: the holder writes the prompt from these, and says the controller asks, which // the bus alone makes true (broker.ControllerOnly). Who asked is the bus's word on the caller, cut to a - // name's characters — never an argument of the call. - raw, err := d.ask(desk, map[string]any{ + // name's characters — never an argument of the call. `what` is the catalogue's own words for the secret + // (novox/hq issue 789), never a word the caller chose; omitted when the module declares none, so an old + // desk and an old module behave as today. + args := map[string]any{ "module": module, "secret": name, "node": node, "asked_by": askedByName(d.askedBy), "seal_to": public, "timeout_seconds": deskPromptWithin, - }) + } + if d.promptFor != nil { + what, err := d.promptFor(module, name) + if err != nil { + outcome = "refused" + return "", fmt.Errorf("nobody was asked to type anything: %w", err) + } + if what != "" { + args["what"] = what + } + } + raw, err := d.ask(desk, args) if err != nil { outcome = "refused" return "", fmt.Errorf("the desk on %s could not be asked: %w", desk, err) @@ -247,7 +264,8 @@ func askAtDesk(ctx context.Context, node, module, name, desk string) error { _, err := open.inventory.NodeByName(ctx, machine) return err }, - trusted: func(module string) (bool, error) { return open.inventory.RunsAsItsOwnAccount(ctx, module) }, + trusted: func(module string) (bool, error) { return open.inventory.RunsAsItsOwnAccount(ctx, module) }, + promptFor: func(module, name string) (string, error) { return open.inventory.OwnSecretPrompt(ctx, module, name) }, ask: func(machine string, args map[string]any) (json.RawMessage, error) { var result json.RawMessage err := onTheBus(func(conn *nats.Conn) error { diff --git a/cmd/mesh-controller/desk_secret_test.go b/cmd/mesh-controller/desk_secret_test.go index 170df795..d2cc26a9 100644 --- a/cmd/mesh-controller/desk_secret_test.go +++ b/cmd/mesh-controller/desk_secret_test.go @@ -456,6 +456,52 @@ func TestThePromptNamesWhoAskedFromTheBussWordAlone(t *testing.T) { } } +// The desk's prompt names which account of a family is asked for, from the catalogue's own words (novox/hq +// issue 789): a family declaring `prompt` with `{member}` sends `what` with the member's suffix filled in. +func TestThePromptNamesWhichAccountOfAFamilyIsAsked(t *testing.T) { + d, _, _, asked := aDesk(t, sealedTo(t, typed)) + d.declares = func(module, name string) error { + if module != "mesh-mail" || name != "mail-credential-gmail" { + return errors.New(module + " does not declare " + name + " as an own secret") + } + return nil + } + d.promptFor = func(module, name string) (string, error) { + if module == "mesh-mail" && name == "mail-credential-gmail" { + return "the password of the mail account gmail", nil + } + return "", nil + } + if _, err := d.give("anchor", "mesh-mail", "mail-credential-gmail", "laptop"); err != nil { + t.Fatal(err) + } + if got := (*asked)[0]["what"]; got != "the password of the mail account gmail" { + t.Errorf(`"what" = %v`, got) + } +} + +// Without a declared prompt, no `what` is sent: an old module, or one that declares none, is asked by name alone, +// as the desk does today (novox/hq issue 789). +func TestWithNoDeclaredPromptNoWhatIsSent(t *testing.T) { + d, _, _, asked := aDesk(t, sealedTo(t, typed)) + d.promptFor = func(module, name string) (string, error) { return "", nil } + if _, err := d.give("anchor", "telegram", "telegram-token", "laptop"); err != nil { + t.Fatal(err) + } + if _, there := (*asked)[0]["what"]; there { + t.Errorf(`"what" was sent with no declared prompt: %v`, (*asked)[0]) + } + // And with no promptFor wired at all (an older caller of deskGive), same thing: `give` still works, asking + // nothing more than it does today. + d2, _, _, asked2 := aDesk(t, sealedTo(t, typed)) + if _, err := d2.give("anchor", "telegram", "telegram-token", "laptop"); err != nil { + t.Fatal(err) + } + if _, there := (*asked2)[0]["what"]; there { + t.Errorf(`"what" was sent with promptFor nil: %v`, (*asked2)[0]) + } +} + // An ask for a secret is bounded before anybody is asked to type (ADR 0277): the record refuses it, nothing is // asked; and how every ask ends is recorded. func TestASecretAskIsBoundedAndItsEndRecorded(t *testing.T) { diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index a79e5675..93d9799e 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -146,6 +146,11 @@ var probeRegistry = []probe{ {ID: probeFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + "said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", Kind: kindBucketOrLogFilling, Raises: []string{kindBucketOrLogUnread}, Phase: 1, run: probeFill}, + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2): the module is carried by the + // machine's tool runner, which never hears a restricted stream, so it hears nothing it was granted. + {ID: probeWithheldID, Asserts: "no grant of a restricted stream names a module that composes nothing for it: every " + + "module granted one runs as an account of its own", From: "ADR 0376 §2", Kind: kindGrantComposesNothing, + Phase: 1, run: probeWithheldHearing}, {ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs}, // The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is diff --git a/cmd/mesh-controller/healers.go b/cmd/mesh-controller/healers.go index 401fe0fc..7bc1c999 100644 --- a/cmd/mesh-controller/healers.go +++ b/cmd/mesh-controller/healers.go @@ -642,7 +642,8 @@ func planStale(ctx context.Context, inv *inventory.Inventory, p inventory.Plan) if err != nil { return "", "", err } - if u.RollOut { + // A recorded module whose move goes out on a machine is still the walk's to send (novox/hq ADR 0377). + if u.RollOut || len(wentOut(s.Recorded)) > 0 { return "", "", nil } } diff --git a/cmd/mesh-controller/heard_withheld.go b/cmd/mesh-controller/heard_withheld.go new file mode 100644 index 00000000..31c86821 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld.go @@ -0,0 +1,66 @@ +package main + +import ( + "context" + "fmt" + "strings" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). +// +// A restricted stream is heard only by a module that runs as an account of its own (`runs-as`, ADR 0259): the +// tool runner's bus account is the one every agent session on its machine reaches the bus as, so the controller +// composes no consumer, grant or bind word for a module it carries (broker.ComposedHears). Where a grant still +// names such a module, the module hears nothing it was granted, and that is said — in `doctor` and in the +// controller's log on every raise and send — naming the module, the stream and the remedy. Never silent. + +const ( + // probeWithheldID is the self-check's probe of it, and kindGrantComposesNothing the condition it raises. + probeWithheldID = "D-heard" + kindGrantComposesNothing = "grant-composes-nothing" +) + +// probeWithheldHearing says every granted hearing of a restricted stream composed for nothing. +func probeWithheldHearing(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + records, err := d.open.inventory.BusRecords(ctx) + if err != nil { + return nil, err + } + return withheldObservations(broker.WithheldOf(records)), nil +} + +// withheldObservations is each withheld hearing as a condition of its module on its machine. +func withheldObservations(withheld []broker.Withheld) []conditions.Observation { + var out []conditions.Observation + for _, w := range withheld { + out = append(out, conditions.Observation{Scope: conditions.ScopeModule, + ID: w.Module + "." + strings.ToLower(w.Stream) + "." + w.Node, + Kind: kindGrantComposesNothing, Machine: w.Node, Severity: conditions.Warning, + Summary: w.Said(), Said: fmt.Sprintf("%s on %s granted %s; composed nothing", w.Module, w.Node, w.Stream)}) + } + return sortedFound(out) +} + +// sayWithheld puts every withheld hearing in the controller's log, once per raise or send that composes it. +func sayWithheld(records broker.Records, indent string) { + for _, w := range broker.WithheldOf(records) { + fmt.Printf("%sA GRANT COMPOSES NOTHING: %s\n", indent, w.Said()) + } +} + +// withheldWords are the plain words of a withheld hearing: the module, the machine and what to do. +func withheldWords(o conditions.Observation) words { + module, m := "a module", machineOr(o, "a machine") + if o.Scope == conditions.ScopeModule && idPart(o, 0) != "" { + module = idPart(o, 0) + } + return words{Headline: clip(conditions.Capital(fmt.Sprintf("%s on %s hears nothing it was granted", module, m)), 60), + Needs: fmt.Sprintf("give %s an account of its own in its manifest; the details say which stream.", module), + Explanation: fmt.Sprintf("You granted %s part of a stream that only modules running under an account of "+ + "their own may hear. It runs inside the shared tool runner on %s, which every assistant session there "+ + "also uses, so the mesh gives it none of that stream.", module, m), + Resolved: fmt.Sprintf("%s on %s hears what it was granted, or is no longer granted it", module, m)} +} diff --git a/cmd/mesh-controller/heard_withheld_test.go b/cmd/mesh-controller/heard_withheld_test.go new file mode 100644 index 00000000..76fbf102 --- /dev/null +++ b/cmd/mesh-controller/heard_withheld_test.go @@ -0,0 +1,43 @@ +package main + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// A grant naming a module the tool runner carries composes nothing, and doctor says so in plain words, naming the +// module, the stream and the remedy (novox/hq ADR 0376 §2). +func TestAGrantThatComposesNothingIsSaid(t *testing.T) { + records := broker.Records{Nodes: []string{"ace"}, Assigned: map[string][]broker.Declared{"ace": { + {Module: broker.RuntimeModule}, + {Module: "reader", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + {Module: "mailer", RunsAs: "mailer", Hears: []broker.Heard{{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, + Filters: []string{"mesh.stream.mesh-mail.event.mail.mesh.>"}}}}, + }}} + found := withheldObservations(broker.WithheldOf(records)) + if len(found) != 1 { + t.Fatalf("said %+v", found) + } + o := found[0] + if o.Kind != kindGrantComposesNothing || o.Machine != "ace" || !strings.Contains(o.Summary, "reader on ace") || + !strings.Contains(o.Summary, "MAIL") || !strings.Contains(o.Summary, "runs-as") { + t.Fatalf("%+v", o) + } + w := plainWordings[kindGrantComposesNothing](o) + if why, ok := conditions.PlainWords(w, "ace"); !ok { + t.Fatalf("not plain: %s (%+v)", why, w) + } + registered := false + for _, p := range probeRegistry { + if p.ID == probeWithheldID && p.Kind == kindGrantComposesNothing && p.run != nil { + registered = true + } + } + if !registered { + t.Fatal("doctor does not run the probe of a grant that composes nothing") + } +} diff --git a/cmd/mesh-controller/held_back.go b/cmd/mesh-controller/held_back.go index a0b4c6ba..66644ec2 100644 --- a/cmd/mesh-controller/held_back.go +++ b/cmd/mesh-controller/held_back.go @@ -146,7 +146,23 @@ func heldMachines(ctx context.Context, open *stores, names []string) (map[string same[m] = current[m].Commit } } - why := heldBack(node, modules, same, known, current, plans) + // A move under a module's own `record` that recreates nothing and moves no data goes as a module that rolls + // out goes (novox/hq ADR 0377): a walk's to send, not a person's. + here, copied := current, false + for m, was := range same { + if now, held := current[m]; held && now.OwnRecord() && !f.identical(m, was, now.Commit) && f.rolls(m, was) { + if !copied { + copied = true + here = make(map[string]inventory.CurrentBuild, len(current)) + for k, v := range current { + here[k] = v + } + } + now.RollOut = true + here[m] = now + } + } + why := heldBack(node, modules, same, known, here, plans) // And a build no gate has seen is not carried by a send that does not judge it (ADR 0236). for _, mv := range f.moves(node, modules, same, known, false) { if planStillToSend(plans, mv.Module, node) == "" { diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index da0511e5..a1049696 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -117,6 +117,8 @@ var plainWordings = map[string]func(conditions.Observation) words{ "reaches it. It keeps running what it has.", m), Resolved: m + " can get new instructions again"} }), + // A grant of a restricted stream that composes nothing (novox/hq ADR 0376 §2). + kindGrantComposesNothing: worded(withheldWords), kindAgentCanBecomeRoot: worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") return words{Headline: "Sessions on " + m + " could become root", diff --git a/cmd/mesh-controller/probes.go b/cmd/mesh-controller/probes.go index 481571f7..deafc4e9 100644 --- a/cmd/mesh-controller/probes.go +++ b/cmd/mesh-controller/probes.go @@ -929,6 +929,12 @@ func streamDiffers(want broker.Stream, have nats.StreamConfig) string { differs = append(differs, fmt.Sprintf("keeps one of a message id for %s, defined %s", have.Duplicates, time.Duration(want.DuplicatesSeconds)*time.Second)) } + if want.AllowMsgTTL && !have.AllowMsgTTL { + differs = append(differs, "takes no time to live of a message's own, defined to (novox/hq ADR 0376)") + } + if want.Restricted && have.Metadata[broker.RestrictedMetadata] != "true" { + differs = append(differs, "does not say it is restricted, so a message kept from it would not age out with it") + } if want.DiscardNew && have.Discard != nats.DiscardNew { differs = append(differs, "drops what it holds when full, defined to refuse what comes next") } diff --git a/cmd/mesh-controller/recorded_kept.go b/cmd/mesh-controller/recorded_kept.go index bc9560ff..397bbd2b 100644 --- a/cmd/mesh-controller/recorded_kept.go +++ b/cmd/mesh-controller/recorded_kept.go @@ -158,6 +158,12 @@ func recordedKept(ctx context.Context, open *stores, node string) (map[string]st if f.identical(m, was, now.Commit) { continue } + // **Under its own `record`, a move that recreates nothing and moves no data is not kept** (novox/hq ADR 0377 + // §1): it goes as a module that rolls out goes. A person's `record`, the bus, and a move that cannot be + // judged are kept as before (§4, §5). + if !every && f.rolls(m, was) { + continue + } out[m] = was } // **And the controller and the node-engine at the build the machine was last sent, over an older record** diff --git a/cmd/mesh-controller/recorded_moves.go b/cmd/mesh-controller/recorded_moves.go new file mode 100644 index 00000000..e72bb0da --- /dev/null +++ b/cmd/mesh-controller/recorded_moves.go @@ -0,0 +1,213 @@ +package main + +import ( + "context" + "fmt" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// Under a module's own `record`, a move that recreates no container and moves no data rolls out (novox/hq ADR +// 0377, changing ADR 0242 §1 and ADR 0360 §1). +// +// A module records because its manifest says so, because it keeps irreplaceable data, because a person set it, or +// because it provides the bus. The first two hold for a reason the manifest shows — a change that recreates mail's +// containers interrupts everyone's mail, a new database image may upgrade its databases in place — and that hold +// also held the moves that interrupt nothing: on 2026-10-15 the database server moved with only its tools bundle +// changed, its walk said it "recreates none of its containers", and ace stayed behind until a person pushed. +// +// So **a move under a module's own `record` is judged per machine, against the build that machine was last sent** +// (§2): when it recreates none of the module's containers and moves none of its data (catalogue.Holds, §3) it +// goes as a module that rolls out goes — a walk sends it, and every other send treats it as a move waiting for its +// walk — and otherwise it is kept at the build the machine runs, as before. The module's policy stays `record`. A +// move that cannot be judged waits (§4); a person's `record` and the bus hold every move (§5); and each walk says +// per machine which way a recorded module went (§6). + +// recordedRollsWhy is what a walk says of a recorded module's move that went out (ADR 0377 §6). +const recordedRollsWhy = "went out: it recreates none of its containers and moves none of its data (novox/hq ADR 0377)" + +// recordVerdict is one machine's move of a recorded module: whether it rolls out, and what the walk says of it. +type recordVerdict struct { + rolls bool + // said is what the walk says: recordedRollsWhy, or "waits: " and why — the recreate words, the data change + // named, the person's or the bus's hold, or that it could not be judged. + said string +} + +// recordedMove judges a module's move on a machine from the build it was last sent (`was`) to the one it would be +// sent now. A module that rolls out rolls; one whose policy is a person's or the bus's waits, said; one under its +// own `record` rolls when the move recreates nothing and moves no data, and waits when it does or cannot be judged. +func (f moveFacts) recordedMove(module, was string) recordVerdict { + now := f.current[module] + switch { + case now.RollOut: + return recordVerdict{rolls: true} + case now.From == catalogue.FromPerson: + return recordVerdict{said: "waits: a person set it to record, and holds every move of it until they push (novox/hq ADR 0377 §5)"} + case now.From == catalogue.FromBus: + return recordVerdict{said: "waits: it provides the bus, which moves only by the bus step a person starts (novox/hq ADR 0377 §5)"} + case !now.OwnRecord(): + return recordVerdict{said: "waits: its policy records"} + case f.judge == nil: + return recordVerdict{said: "waits: it could not be judged — the build records were not read (novox/hq ADR 0377 §4)"} + } + return f.judge(module, was, now.Commit) +} + +// rolls is whether a module's move on a machine from `was` goes as a move that rolls out (ADR 0377 §1). +func (f moveFacts) rolls(module, was string) bool { + return f.recordedMove(module, was).rolls +} + +// recordJudge is the judging of moves under a module's own `record`, reading the build records once per move and +// the catalogue once (ADR 0377 §2–§4). +func recordJudge(ctx context.Context, inv *inventory.Inventory) func(module, was, now string) recordVerdict { + judged := map[string]recordVerdict{} + var shelf map[string]catalogue.Manifest + var shelfErr error + shelfRead := false + return func(module, was, now string) recordVerdict { + key := module + "\x00" + was + "\x00" + now + if v, seen := judged[key]; seen { + return v + } + v := func() recordVerdict { + cannot := func(why string) recordVerdict { + return recordVerdict{said: "waits: it could not be judged — " + why + " (novox/hq ADR 0377 §4)"} + } + if was == "" { + return cannot("what the machine was last sent of it is not known") + } + from, found, err := inv.ManifestAt(ctx, module, was) + if err != nil { + return cannot(fmt.Sprintf("its build %s, which the machine was last sent, cannot be read: %v", short(was), err)) + } + if !found { + return cannot(fmt.Sprintf("its build %s, which the machine was last sent, is not in the build records", short(was))) + } + if !shelfRead { + shelf, shelfErr = inv.Catalogue(ctx) + shelfRead = true + } + if shelfErr != nil { + return cannot(fmt.Sprintf("the build it would be sent cannot be read: %v", shelfErr)) + } + to, held := shelf[module] + if !held { + return cannot("the build it would be sent is not in the catalogue") + } + if why := catalogue.Holds(from, to); why != "" { + return recordVerdict{said: "waits: " + why} + } + return recordVerdict{rolls: true, said: recordedRollsWhy} + }() + judged[key] = v + return v + } +} + +// judgeRecordedInWalk judges, for a walk carrying a module whose policy records, each machine running it whose +// build would move (ADR 0377 §2), keeps on the plan which way each went (§6), says it on the controller's log, and +// answers the machines its move goes out to. Judged once per walk: a walk read again answers what it kept. +func judgeRecordedInWalk(ctx context.Context, inv *inventory.Inventory, p *inventory.Plan, module string, + state *inventory.PlanModule) ([]string, error) { + if state.Recorded != nil { + return wentOut(state.Recorded), nil + } + running, err := inv.Running(ctx, module) + if err != nil { + return nil, err + } + f, err := readMoveFacts(ctx, inv) + if err != nil { + return nil, err + } + now := f.current[module] + said := map[string]string{} + for _, n := range running { + sent, known, err := inv.SentBuilds(ctx, n) + if err != nil { + return nil, err + } + was, carried := sent[module] + if known && carried && f.identical(module, was, now.Commit) { + continue // no move there + } + v := f.recordedMove(module, was) + if !known { + v = recordVerdict{said: "waits: it could not be judged — what " + n + " was last sent is not known (novox/hq ADR 0377 §4)"} + } + said[n] = v.said + fmt.Printf("%s: %s on %s %s\n", p.ID, module, n, v.said) + } + if len(said) > 0 { + state.Recorded = said + } + return wentOut(said), nil +} + +// wentOut is the machines a recorded module's move went out to, sorted. +func wentOut(recorded map[string]string) []string { + var out []string + for _, n := range sortedKeysOf(recorded) { + if recorded[n] == recordedRollsWhy { + out = append(out, n) + } + } + return out +} + +// recordedSaid is, for a walk's note at its end, which way each recorded module it carried went on each machine +// (ADR 0377 §6). +func recordedSaid(p inventory.Plan) []string { + var out []string + for _, m := range sortedKeysOf(planModuleNames(p)) { + s := p.Modules[m] + for _, n := range sortedKeysOf(s.Recorded) { + out = append(out, fmt.Sprintf("%s on %s %s", m, n, s.Recorded[n])) + } + } + return out +} + +func planModuleNames(p inventory.Plan) map[string]string { + out := map[string]string{} + for m, s := range p.Modules { + if s != nil && len(s.Recorded) > 0 { + out[m] = m + } + } + return out +} + +// recordedWaiting is, on one machine, every move of a recorded module that waits for a person, each said with why +// (ADR 0377 §6): what `status` names the machine behind for, with `push ` as the remedy. A move under a +// module's own `record` that recreates nothing and moves no data is not among them: it goes with its walk. +func recordedWaiting(ctx context.Context, inv *inventory.Inventory, f moveFacts, node string) ([]string, error) { + sent, known, err := inv.SentBuilds(ctx, node) + if err != nil || !known { + return nil, err + } + var out []string + for _, m := range sortedKeysOf(sent) { + was := sent[m] + now, held := f.current[m] + if !held || now.RollOut || now.Commit == "" || was == "" || f.identical(m, was, now.Commit) { + continue + } + if v := f.recordedMove(m, was); !v.rolls { + out = append(out, fmt.Sprintf("%s %s → %s %s", m, short(was), short(now.Commit), v.said)) + } + } + return out, nil +} + +// stringsKeys is a map of lists as a map of its keys, for sortedKeysOf. +func stringsKeys(m map[string][]string) map[string]string { + out := map[string]string{} + for k := range m { + out[k] = k + } + return out +} diff --git a/cmd/mesh-controller/recorded_moves_test.go b/cmd/mesh-controller/recorded_moves_test.go new file mode 100644 index 00000000..cc11869a --- /dev/null +++ b/cmd/mesh-controller/recorded_moves_test.go @@ -0,0 +1,361 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "strings" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/catalogue" + "github.com/novox/mesh-controller/internal/inventory" +) + +// Under a module's own `record`, a move that recreates no container and moves no data rolls out (novox/hq ADR +// 0377). Seen 2026-10-15: the database server moved with only its tools bundle changed, its walk said it +// "recreates none of its containers", and ace stayed behind until a person pushed. + +// aMailu is a mail module as the catalogue declares it, small: it records because its manifest says so, keeps its +// mailboxes as data, mounts its environment file, and serves its tools from a bundle. +func aMailu(policy bool) catalogue.Manifest { + raw := `{"module":"mailu","version":"1","tools":["mailu_status"], + "data":{"own":[{"id":"mail","path":"${dir:data-mail}","class":"valuable","why":"every mailbox"}]}, + "bundles":[{"name":"tools","source":"store/mailu-tools@sha256:aaa","digest":"sha256:aaa"}], + "resources":[ + {"id":"state","type":"directory"}, + {"id":"mesh-state","type":"directory"}, + {"id":"data-mail","type":"directory"}, + {"id":"config-env","type":"file","path":"${dir:state}/mailu.env","content":"A=1"}, + {"id":"runtime-config","type":"file","path":"${dir:mesh-state}/config.json","content":"{}"}, + {"id":"imap","type":"container","name":"mailu-imap","image":"registry.invalid/imap@sha256:1", + "volumes":["${dir:data-mail}:/mail"],"env-file":["${dir:state}/mailu.env"]}]}` + var m catalogue.Manifest + if err := json.Unmarshal([]byte(raw), &m); err != nil { + panic(err) + } + if policy { + m.Upgrade = &catalogue.UpgradePolicy{Policy: catalogue.PolicyRecord, Why: "people's mail: a person chooses the moment"} + } + return m +} + +func resourceOf(m *catalogue.Manifest, id string) map[string]any { + for _, r := range m.Resources { + if r["id"] == id { + return r + } + } + panic("no resource " + id) +} + +// recordedMesh is a mesh running mailu at build c1 on anchor and laptop, as each was last sent. +type recordedMesh struct { + open *stores + sent [][]string +} + +func aRecordedMesh(t *testing.T, c1 catalogue.Manifest) *recordedMesh { + t.Helper() + open := aMesh(t) + ctx := t.Context() + inv := open.inventory + rm := &recordedMesh{open: open} + rm.register(t, c1, "c1", time.Now().Add(-2*time.Hour)) + for _, n := range []string{"anchor", "laptop"} { + if _, err := inv.Assign(ctx, n, c1.Module); err != nil { + t.Fatal(err) + } + if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{c1.Module: "c1"}); err != nil { + t.Fatal(err) + } + } + return rm +} + +// register keeps a build of a module and makes it the mesh's. +func (rm *recordedMesh) register(t *testing.T, m catalogue.Manifest, commit string, at time.Time) { + t.Helper() + ctx := t.Context() + raw, _ := json.Marshal(m) + if err := rm.open.inventory.RecordBuild(ctx, inventory.Build{ID: "build-" + m.Module + "-" + commit, Module: m.Module, + Commit: commit, Repository: "novox/mesh-catalog", Path: "modules/" + m.Module, Manifest: raw, Asked: at, At: at}); err != nil { + t.Fatal(err) + } + if err := rm.open.inventory.RegisterModule(ctx, m, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", + Path: "modules/" + m.Module, BuiltFrom: commit, Head: commit, Asked: at}); err != nil { + t.Fatal(err) + } +} + +// keptOn is what an ordinary send — a walk's, a healer's — keeps on a machine: module → the build it keeps. +func (rm *recordedMesh) keptOn(t *testing.T, node string) map[string]string { + t.Helper() + ctx := t.Context() + under, err := sendKeeps(ctx, rm.open.inventory) + if err != nil { + t.Fatal(err) + } + kept, err := recordedKept(under, rm.open, node) + if err != nil { + t.Fatal(err) + } + return kept +} + +// walk opens a merge's walk of mailu's c2, built, and walks it with every send composed as an ordinary send +// composes it — a recorded module kept where its move waits — and applied at once. +func (rm *recordedMesh) walk(t *testing.T) inventory.Plan { + t.Helper() + ctx := t.Context() + inv := rm.open.inventory + wasSend := sendRollout + n := 0 + sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) { + rm.sent = append(rm.sent, append([]string(nil), names...)) + current, err := open.inventory.CurrentBuilds(ctx) + if err != nil { + return nil, err + } + for _, node := range names { + kept := rm.keptOn(t, node) + builds := map[string]string{"mailu": current["mailu"].Commit} + if was, held := kept["mailu"]; held { + builds["mailu"] = was + } + n++ + digest := fmt.Sprintf("d-%s-%d", node, n) + if err := inv.RecordSent(ctx, nodeID(t, open, node), digest, builds); err != nil { + return nil, err + } + if _, err := inv.RecordDoing(ctx, nodeID(t, open, node), inventory.Doing{Node: node, + Outcome: inventory.OutcomeApplied, Declared: digest, Applied: 1, At: time.Now()}); err != nil { + return nil, err + } + } + return names, nil + } + t.Cleanup(func() { sendRollout = wasSend }) + built := time.Now().UTC() + plan := inventory.Plan{ID: "plan-mailu", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", + Created: built, State: inventory.PlanBuilding, Tiers: [][]string{{"mailu"}}, + Modules: map[string]*inventory.PlanModule{"mailu": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-mailu-c2"}}} + if err := inv.SavePlan(ctx, &plan); err != nil { + t.Fatal(err) + } + for i := 0; i < 8; i++ { + advancePlans(ctx, rm.open) + } + p, err := inv.PlanByID(ctx, "plan-mailu") + if err != nil { + t.Fatal(err) + } + return p +} + +func (rm *recordedMesh) sentBuild(t *testing.T, node string) string { + t.Helper() + sent, _, err := rm.open.inventory.SentBuilds(t.Context(), node) + if err != nil { + t.Fatal(err) + } + return sent["mailu"] +} + +// recordedWaitsOn is what `status` names a machine behind for, waiting for a person (ADR 0377 §6). +func (rm *recordedMesh) recordedWaitsOn(t *testing.T, node string) []string { + t.Helper() + f, err := readMoveFacts(t.Context(), rm.open.inventory) + if err != nil { + t.Fatal(err) + } + held, err := recordedWaiting(t.Context(), rm.open.inventory, f, node) + if err != nil { + t.Fatal(err) + } + return held +} + +// §1, §6: a module that records because its manifest says so, whose move changes only its tools bundle, goes out +// with its walk to every machine, without a push; the walk says per machine that it went out and why; the policy +// still reads record; and status names no machine behind for it. +func TestARecordedModulesBundleOnlyMoveRollsOutInItsWalk(t *testing.T) { + c1 := aMailu(true) + rm := aRecordedMesh(t, c1) + c2 := aMailu(true) + c2.Bundles[0].Digest, c2.Bundles[0].Source = "sha256:bbb", "store/mailu-tools@sha256:bbb" + c2.Tools = append(c2.Tools, "mailu_queue") + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + + for _, n := range []string{"anchor", "laptop"} { + if _, kept := rm.keptOn(t, n)["mailu"]; kept { + t.Fatalf("an ordinary send to %s keeps mailu at c1 for a move that recreates nothing", n) + } + } + p := rm.walk(t) + if p.State != inventory.PlanDone { + t.Fatalf("the walk is %s: %s", p.State, p.Note) + } + for _, n := range []string{"anchor", "laptop"} { + if got := rm.sentBuild(t, n); got != "c2" { + t.Fatalf("%s was sent mailu %s; want c2, with its walk", n, got) + } + if said := p.Modules["mailu"].Recorded[n]; said != recordedRollsWhy { + t.Fatalf("the walk says of mailu on %s %q", n, said) + } + if held := rm.recordedWaitsOn(t, n); len(held) != 0 { + t.Fatalf("status names %s behind for %v", n, held) + } + } + if !strings.Contains(p.Note, "mailu on anchor went out") || !strings.Contains(p.Note, "mailu on laptop went out") { + t.Fatalf("the walk's note does not say which way mailu went: %q", p.Note) + } + if u, err := rm.open.inventory.UpgradeOf(t.Context(), "mailu"); err != nil || u.Policy() != catalogue.PolicyRecord { + t.Fatalf("mailu's policy is %q (%v); want record still", u.Policy(), err) + } +} + +// §1, §2, §6: a move that recreates a container — a new image, or a changed declaration — still waits for a push; +// the walk says per machine that it waits, with the recreate words; and status names the machine behind, with +// `push ` as the remedy. A bundle-only change on top of it stays held: it is judged against what the machine +// was last sent. +func TestARecordedModulesMoveThatRecreatesAContainerWaits(t *testing.T) { + for name, change := range map[string]func(m *catalogue.Manifest){ + "image": func(m *catalogue.Manifest) { resourceOf(m, "imap")["image"] = "registry.invalid/imap@sha256:2" }, + "spec": func(m *catalogue.Manifest) { resourceOf(m, "imap")["health"] = map[string]any{"kind": "runtime"} }, + } { + t.Run(name, func(t *testing.T) { + rm := aRecordedMesh(t, aMailu(true)) + c2 := aMailu(true) + change(&c2) + c2.Bundles[0].Digest = "sha256:bbb" + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + p := rm.walk(t) + for _, n := range []string{"anchor", "laptop"} { + if got := rm.sentBuild(t, n); got != "c1" { + t.Fatalf("%s was sent mailu %s; want c1 kept, waiting for a push", n, got) + } + if said := p.Modules["mailu"].Recorded[n]; !strings.HasPrefix(said, "waits: recreates 1 of mailu's 1") { + t.Fatalf("the walk says of mailu on %s %q", n, said) + } + if held := rm.recordedWaitsOn(t, n); len(held) != 1 || !strings.Contains(held[0], "recreates") { + t.Fatalf("status names %s behind for %v", n, held) + } + } + // §2: the next build changes only the bundle, and the machines were last sent c1: still held. + c3 := c2 + c3.Bundles = []catalogue.Bundle{{Name: "tools", Source: "store/mailu-tools@sha256:ccc", Digest: "sha256:ccc"}} + rm.register(t, c3, "c3", time.Now()) + if kept := rm.keptOn(t, "anchor"); kept["mailu"] != "c1" { + t.Fatalf("a bundle-only build on top of a held container change is not held: %v", kept) + } + }) + } +} + +// §3: a move touching a file a container mounts, a directory holding declared data, the data declaration or a +// provider waits. +func TestARecordedModulesMoveThatTouchesDataOrAProviderWaits(t *testing.T) { + for name, c := range map[string]struct { + change func(m *catalogue.Manifest) + says string + }{ + "a mounted file": {func(m *catalogue.Manifest) { resourceOf(m, "config-env")["content"] = "A=2" }, "which its container imap mounts"}, + "a data directory": {func(m *catalogue.Manifest) { resourceOf(m, "data-mail")["mode"] = "0700" }, "data-mail"}, + "the data declaration": {func(m *catalogue.Manifest) { + m.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Path: "${dir:data-mail}", Class: "irreplaceable", Why: "x"}}} + }, "data declaration"}, + "a provider": {func(m *catalogue.Manifest) { m.Requires = append(m.Requires, "postgres-database") }, "what it requires"}, + } { + t.Run(name, func(t *testing.T) { + rm := aRecordedMesh(t, aMailu(true)) + c2 := aMailu(true) + c.change(&c2) + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + kept := rm.keptOn(t, "anchor") + if kept["mailu"] != "c1" { + t.Fatalf("an ordinary send does not keep mailu at c1: %v", kept) + } + held := rm.recordedWaitsOn(t, "anchor") + if len(held) != 1 || !strings.Contains(held[0], c.says) { + t.Fatalf("status says %v; want it to name %q", held, c.says) + } + }) + } +} + +// §5: a module a person set to record, and the bus, wait on a bundle-only move. +func TestAPersonsRecordAndTheBusHoldEveryMove(t *testing.T) { + t.Run("a person sets record", func(t *testing.T) { + rm := aRecordedMesh(t, aMailu(false)) + c2 := aMailu(false) + c2.Bundles[0].Digest = "sha256:bbb" + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + if err := rm.open.inventory.SetUpgradeOf(t.Context(), "mailu", inventory.Upgrade{Why: "the migration week", By: "jochen"}); err != nil { + t.Fatal(err) + } + p := rm.walk(t) + if got := rm.sentBuild(t, "anchor"); got != "c1" { + t.Fatalf("a person's record was walked past: anchor was sent %s", got) + } + if said := p.Modules["mailu"].Recorded["anchor"]; !strings.Contains(said, "a person set it to record") { + t.Fatalf("the walk says %q", said) + } + if held := rm.recordedWaitsOn(t, "anchor"); len(held) != 1 { + t.Fatalf("status names anchor behind for %v", held) + } + }) + t.Run("the bus", func(t *testing.T) { + bus := aMailu(false) + bus.Module = "nats" + bus.Provides = []catalogue.Offer{{Name: "mesh-bus"}} + bus.Data = nil + rm := aRecordedMesh(t, bus) + c2 := bus + c2.Bundles = []catalogue.Bundle{{Name: "tools", Source: "store/nats-tools@sha256:bbb", Digest: "sha256:bbb"}} + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + if kept := rm.keptOn(t, "anchor"); kept["nats"] != "c1" { + t.Fatalf("the bus moved by an ordinary send: %v", kept) + } + f, err := readMoveFacts(t.Context(), rm.open.inventory) + if err != nil { + t.Fatal(err) + } + if v := f.recordedMove("nats", "c1"); v.rolls || !strings.Contains(v.said, "bus") { + t.Fatalf("the bus's move: %+v", v) + } + }) +} + +// §4: a move whose earlier build is not in the build records cannot be judged, waits, and is said so. +func TestARecordedMoveThatCannotBeJudgedWaits(t *testing.T) { + rm := aRecordedMesh(t, aMailu(true)) + c2 := aMailu(true) + c2.Bundles[0].Digest = "sha256:bbb" + rm.register(t, c2, "c2", time.Now().Add(-time.Minute)) + // anchor was last sent a build the records do not hold. + if err := rm.open.inventory.RecordSent(t.Context(), nodeID(t, rm.open, "anchor"), "d-anchor-c0", + map[string]string{"mailu": "c0"}); err != nil { + t.Fatal(err) + } + f, err := readMoveFacts(t.Context(), rm.open.inventory) + if err != nil { + t.Fatal(err) + } + if v := f.recordedMove("mailu", "c0"); v.rolls || !strings.Contains(v.said, "could not be judged") || + !strings.Contains(v.said, "not in the build records") { + t.Fatalf("an unjudgeable move: %+v", v) + } + if _, _, err := planFor(keepingRecorded(t.Context()), rm.open, "anchor"); err == nil || !strings.Contains(err.Error(), "push anchor") { + t.Fatalf("a send that cannot keep the unjudged build did not refuse with its remedy: %v", err) + } + // laptop, judged from c1, goes; anchor waits. + p := rm.walk(t) + if said := p.Modules["mailu"].Recorded["anchor"]; !strings.Contains(said, "could not be judged") { + t.Fatalf("the walk says of anchor %q", said) + } + if !slices.Contains(wentOut(p.Modules["mailu"].Recorded), "laptop") || rm.sentBuild(t, "laptop") != "c2" { + t.Fatalf("laptop, judged from c1, did not go: %+v", p.Modules["mailu"].Recorded) + } +} diff --git a/cmd/mesh-controller/release.go b/cmd/mesh-controller/release.go index 4937b5fe..4e2b3634 100644 --- a/cmd/mesh-controller/release.go +++ b/cmd/mesh-controller/release.go @@ -83,6 +83,9 @@ type moveFacts struct { plans []inventory.Plan // self is the release walking these moves, whose own gates are no other walk's (novox/hq ADR 0353). self string + // judge judges a move under a module's own `record` (novox/hq ADR 0377): nil when the facts were not read, + // and every such move then waits. + judge func(module, was, now string) recordVerdict } type walkingKey struct{} @@ -109,6 +112,7 @@ func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, er } f.plans, err = inv.OpenPlans(ctx) f.self, _ = ctx.Value(walkingKey{}).(string) + f.judge = recordJudge(ctx, inv) return f, err } @@ -172,7 +176,8 @@ func inRun(run map[string]bool, commit string) bool { } // moves is what a machine's next send would move that no gate has seen: modules whose policy rolls out -// (a recorded one is a person's push), that the machine was sent before, moving to a build not identical +// (a recorded one is a person's push, unless its move under its own record recreates nothing and moves no data — +// novox/hq ADR 0377), that the machine was sent before, moving to a build not identical // to the one it runs and that has passed no gate. A machine whose last send's builds are not known names // none: the cascade holds it whole (ADR 0221). // @@ -185,7 +190,7 @@ func (f moveFacts) moves(node string, modules []string, sent map[string]string, for _, m := range modules { now := f.current[m] was, carried := sent[m] - if !now.RollOut || !carried || f.identical(m, was, now.Commit) || (!all && f.gated(m, now.Commit)) { + if !carried || f.identical(m, was, now.Commit) || (!all && f.gated(m, now.Commit)) || !f.rolls(m, was) { continue } out = append(out, inventory.CarriedMove{Module: m, Node: node, From: was, To: now.Commit}) diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index bb49ccad..fa523935 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -865,13 +865,32 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, together := map[string]bool{} for _, m := range tier { state := p.Modules[m] - if state == nil || state.SentAt != nil || state.State == planDeleted || !rollsOut(m) { + if state == nil || state.SentAt != nil || state.State == planDeleted { continue } + // **A recorded module goes where its move recreates nothing and moves no data** (novox/hq ADR 0377): judged + // per machine against the build it was last sent, said per machine, and sent only there. + var goes map[string]bool + if !rollsOut(m) { + out, err := judgeRecordedInWalk(ctx, inv, p, m, state) + if err != nil { + return false, err + } + if len(out) == 0 { + continue + } + goes = map[string]bool{} + for _, n := range out { + goes[n] = true + } + } running, err := inv.Running(ctx, m) if err != nil { return false, err } + if goes != nil { + running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] }) + } runningOf[m] = running // **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source // of the build every machine running it was sent, it is registered with that build's artifacts — @@ -1150,6 +1169,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan, said = append(said, fmt.Sprintf("said nothing on its send within %s: %s", inventory.ApplySilentAfter, strings.Join(silent, ", "))) } + said = append(said, recordedSaid(*p)...) end = strings.Join(said, "; ") // **The new controller's user list follows its walk** (ADR 0236 §34, read without the gate since ADR 0360): // once a walk that moved the controller ends, the bus's machine is sent the list it composes, if behind. @@ -1955,6 +1975,12 @@ func plansCommand(ctx context.Context, args []string) error { case s != nil && s.Gate != nil: fmt.Printf(" %-22s %s\n", "", gateLine(s.Gate)) } + // Which way a recorded module went on each machine (novox/hq ADR 0377 §6). + if s != nil { + for _, n := range sortedKeysOf(s.Recorded) { + fmt.Printf(" %-22s on %s %s\n", "", n, s.Recorded[n]) + } + } // What the send to its first machine did to its containers (ADR 0245). if s != nil && s.Gate != nil { for _, c := range s.Gate.Carried { diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index af68b7f5..01cf7f98 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -205,6 +205,10 @@ func printStatus(asked answers) error { fmt.Printf("%d machine(s) have never been sent anything:\n %s\n", len(never), strings.Join(never, ", ")) } + for _, n := range sortedKeysOf(stringsKeys(asked.recordedWaits)) { + fmt.Printf(" %s is behind, waiting for a person: %s — `push %s` sends it\n", n, + strings.Join(asked.recordedWaits[n], "; "), n) + } fmt.Printf("\n `push --behind` sends them\n\n") } @@ -540,6 +544,29 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { if err != nil { return answers{}, err } + // And, of those, what waits for a person's push (novox/hq ADR 0377 §6): a recorded module's move that went out + // with its walk raises nothing here. + if len(out.waiting) > 0 { + f, err := readMoveFacts(ctx, inv) + if err != nil { + return answers{}, err + } + for _, m := range out.waiting { + if m.Never { + continue + } + held, err := recordedWaiting(ctx, inv, f, m.Node) + if err != nil { + return answers{}, err + } + if len(held) > 0 { + if out.recordedWaits == nil { + out.recordedWaits = map[string][]string{} + } + out.recordedWaits[m.Node] = held + } + } + } out.reported, err = inv.LastReports(ctx) if err != nil { return answers{}, err diff --git a/cmd/mesh-controller/watchdogs.go b/cmd/mesh-controller/watchdogs.go index a8d2f793..4b5fde61 100644 --- a/cmd/mesh-controller/watchdogs.go +++ b/cmd/mesh-controller/watchdogs.go @@ -91,7 +91,10 @@ type signalFacts struct { // deadLettersNewest is when DEAD_LETTERS kept the newest message it holds for each of those // consumers: the condition counts the messages given up on, not the looks (novox/hq issue 440). deadLettersNewest map[string]time.Time - advisoriesErr error + // deadLettersAging is, for each consumer whose dead letters came from a restricted stream, when each ages out + // (novox/hq ADR 0376 §6); nil when it was not read. + deadLettersAging map[string][]link.Aging + advisoriesErr error // busFaultsErr is why the bus module's events could not be read this tick (novox/hq ADR 0357). busFaultsErr error @@ -211,6 +214,9 @@ type watchdogs struct { // confirm holds back a row blind for one tick: a read that timed out once on a loaded store. confirm confirming + // aging is what is remembered of restricted dead letters between looks (novox/hq ADR 0376 §6). + aging agingTally + mu sync.Mutex ticked time.Time last *signalFacts @@ -264,6 +270,8 @@ func (w *watchdogs) tick(ctx context.Context) { func (w *watchdogs) see(running context.Context, f *signalFacts) { var problems []string var blind []conditions.Observation + // What aged out of DEAD_LETTERS, said and cleared in its own words before the rows clear it in theirs. + problems = append(problems, w.sayAgedOut(running, f)...) for _, row := range watchedRows() { if err := row.needs(f); err != nil { blind = append(blind, blindRow(row, err)) @@ -382,6 +390,9 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts { } } } + if f.advisoriesErr == nil && w.js != nil { + f.deadLettersAging, f.advisoriesErr = link.AgingDeadLetters(w.js.Context(), f.deadLetters) + } f.handActs, f.handActsErr = w.gatherHandActs(ctx, now) f.refusedSends, f.refusedSendsErr = inv.RefusedSendsSince(ctx, now.Add(-generationRefusalsSaid)) f.doing, f.doingErr = inv.NotDoingWhatTheyWereTold(ctx) diff --git a/internal/broker/derived.go b/internal/broker/derived.go index 1a770ecf..cd370df1 100644 --- a/internal/broker/derived.go +++ b/internal/broker/derived.go @@ -210,7 +210,7 @@ func ConsumersOf(users []Principal) []ModuleConsumer { for _, d := range p.Carries { add(Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears}) + Streams: d.Streams, Hears: ComposedHears(d)}) } } return out diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 9f9dca93..495c5290 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -173,6 +173,10 @@ func (j *JetStream) EnsureStream(s Stream) error { Description: s.Why, } want.AllowDirect = s.Direct + want.AllowMsgTTL = s.AllowMsgTTL + if s.Restricted { + want.Metadata = map[string]string{RestrictedMetadata: "true"} + } if s.MaxBytes > 0 { want.MaxBytes = s.MaxBytes } diff --git a/internal/broker/modstreams.go b/internal/broker/modstreams.go index e6e3a315..fc584b67 100644 --- a/internal/broker/modstreams.go +++ b/internal/broker/modstreams.go @@ -77,6 +77,8 @@ func (s ModuleStream) Stream() Stream { MaxAge: s.MaxAgeDays * 24 * 60 * 60, MaxBytes: int64(s.MaxMiB) * 1024 * 1024, Why: why, + // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3). + Restricted: s.Restricted, } } @@ -98,6 +100,62 @@ type Heard struct { Stream string Emitter string Filters []string + // Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it + // is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears). + Restricted bool +} + +// ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it +// hears, but a restricted one only for a module that runs in a runtime of its own on a bus account of its own +// (`runs-as`, ADR 0259). **The tool runner never hears a restricted stream**: a module it carries reaches the bus +// through the tool runner's account, which every agent session on that machine reaches the bus as too, so a +// consumer, its grants or its bind word composed there would hand the stream to any session. A module that +// declares no `runs-as` and keeps its own container credential gets none either: only `runs-as` is an account +// the mesh keeps from agents. +func ComposedHears(d Declared) []Heard { + if d.RunsAs != "" { + return d.Hears + } + var out []Heard + for _, h := range d.Hears { + if !h.Restricted { + out = append(out, h) + } + } + return out +} + +// Withheld is a restricted stream a grant gives a module that the controller composes nothing for, because the +// module has no account of its own (novox/hq ADR 0376 §2): said in `doctor` and in the controller's log, never +// silent. +type Withheld struct { + Node, Module, Stream, Emitter string +} + +// Said is a withheld hearing in the operator's words: the module, the stream and the remedy. +func (w Withheld) Said() string { + return fmt.Sprintf("%s on %s is granted part of %s's restricted stream %s, and hears none of it: it is carried "+ + "by the machine's tool runner, whose bus account every agent session there shares, so nothing is composed "+ + "for it. Declare runs-as in %s's manifest, so it runs as an account of its own (novox/hq ADR 0376)", + w.Module, w.Node, w.Emitter, w.Stream, w.Module) +} + +// WithheldOf is every granted hearing of a restricted stream composed for nothing, by machine and module. +func WithheldOf(r Records) []Withheld { + var out []Withheld + for _, node := range sortedCopy(r.Nodes) { + for _, d := range r.Assigned[node] { + if d.RunsAs != "" { + continue + } + for _, h := range d.Hears { + if h.Restricted && len(h.Filters) > 0 { + out = append(out, Withheld{Node: node, Module: d.Module, Stream: h.Stream, Emitter: h.Emitter}) + } + } + } + } + return out } // heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer @@ -177,7 +235,8 @@ func streamsIssuedFor(d Declared) []StreamIssued { // HeardIssuedFor is every module stream an assignment hears, as its membership names it. func HeardIssuedFor(node string, d Declared) []HeardIssued { var out []HeardIssued - for _, h := range d.Hears { + // A restricted stream's bind word only for a module of its own account (novox/hq ADR 0376 §1). + for _, h := range ComposedHears(d) { if len(h.Filters) == 0 { continue } diff --git a/internal/broker/modstreams_test.go b/internal/broker/modstreams_test.go index ffa04bd2..9445a64d 100644 --- a/internal/broker/modstreams_test.go +++ b/internal/broker/modstreams_test.go @@ -14,9 +14,14 @@ var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL", MaxAgeDays: 7, MaxMiB: 1024, Restricted: true} // mailHeard is what a reader granted the account `mesh` hears of it. -var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{ +var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{ "mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}} +// newsStream is a module's stream without a grant: heard by whoever asks (ADR 0363 §5), the tool runner included. +var newsStream = ModuleStream{Module: "news", Name: "NEWS", Events: []string{"story.**"}, MaxAgeDays: 7, MaxMiB: 16} + +var newsHeard = Heard{Stream: "NEWS", Emitter: "news", Filters: []string{"mesh.stream.news.event.story.>"}} + // A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2). func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) { s := mailStream.Stream() @@ -91,12 +96,14 @@ func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing. } } -// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS -// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both. +// A granted consumer of its own account has a durable consumer on the stream, with its granted subjects as +// filters, and an EVENTS consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); a module of its +// own account is no module the runtime carries, so the runtime reads neither (ADR 0376 §1). func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}}, - "ace": {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}}, + "ace": {{Module: RuntimeModule}, {Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, + {Module: "stranger", HearsAStream: true}}, }} users, err := Users(records) if err != nil { @@ -118,22 +125,112 @@ func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) { t.Fatalf("on EVENTS: %+v", onEvents) } for _, u := range users { - if u.Kind != KindNodeTools || u.Node != "ace" { + if u.Node != "ace" || (u.Kind != KindNodeTools && u.Module != "reader") { continue } perms, err := PermissionsFor(u) if err != nil { t.Fatal(err) } - if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") { - t.Fatal("the runtime carrying the reader may not read its consumer on MAIL") + reads := MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") && MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") + if u.Kind == KindNodeTools && reads { + t.Fatal("the runtime may read the consumer on MAIL of a module of its own account") + } + if u.Kind == KindModule && !reads { + t.Fatal("the reader's own account may not read its consumer on MAIL") } if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") { - t.Fatal("the runtime may read a consumer no module it carries is granted") + t.Fatal("a consumer no module is granted may be read") } } } +// **The tool runner never hears a restricted stream** (ADR 0376 §1, §2): a module it carries, granted part of +// MAIL, gets no consumer on MAIL, the tool runner's account no grant to ask about, pull or acknowledge one there, +// and its membership no word to bind one; a module of its own account granted the same part gets all three, on +// its own account. The grant that composes nothing is said. A stream without a grant is heard through the tool +// runner as before. +func TestTheToolRunnerNeverHearsARestrictedStream(t *testing.T) { + records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{ + "novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, {Module: "news", Streams: []ModuleStream{newsStream}}}, + "ace": {{Module: RuntimeModule}, + {Module: "carried", Hears: []Heard{mailHeard, newsHeard}, HearsAStream: true}, + {Module: "mailer", RunsAs: "mailer", Hears: []Heard{mailHeard}, HearsAStream: true}}, + }} + users, err := Users(records) + if err != nil { + t.Fatal(err) + } + var onMail, onNews []string + for _, c := range ConsumersOf(users) { + switch c.Consumer.Stream { + case "MAIL": + onMail = append(onMail, c.Consumer.Name) + case "NEWS": + onNews = append(onNews, c.Consumer.Name) + } + } + if !slices.Equal(onMail, []string{"ace_mailer"}) { + t.Fatalf("consumers on MAIL: %v, want only the mailer's", onMail) + } + if !slices.Equal(onNews, []string{"ace_carried"}) { + t.Fatalf("consumers on NEWS: %v, want the carried module's as before", onNews) + } + for _, u := range users { + if u.Node != "ace" { + continue + } + perms, err := PermissionsFor(u) + if err != nil { + t.Fatal(err) + } + switch { + case u.Kind == KindNodeTools: + for _, s := range perms.Publish { + if strings.Contains(s, ".MAIL.") { + t.Errorf("the tool runner's account may publish %s", s) + } + } + for _, s := range perms.Subscribe { + if strings.HasPrefix(s, "mesh.stream.mesh-mail.") { + t.Errorf("the tool runner's account may subscribe %s", s) + } + } + if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.NEWS.ace_carried") { + t.Error("the tool runner may no longer read the carried module's consumer on a stream without a grant") + } + case u.Kind == KindModule && u.Module == "carried": + if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_carried") { + t.Error("a module without an account of its own may pull a consumer on MAIL") + } + case u.Kind == KindModule && u.Module == "mailer": + for _, s := range []string{"$JS.API.CONSUMER.INFO.MAIL.ace_mailer", "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_mailer", + "$JS.ACK.MAIL.ace_mailer.1"} { + if !MayPublish(perms, s) { + t.Errorf("the mailer's own account may not publish %s", s) + } + } + } + } + where := Placements{Nodes: map[string][]string{}} + carried := MembershipFor("ace", records.Assigned["ace"][1], where) + if !reflect.DeepEqual(carried.Hears, []HeardIssued{{Stream: "NEWS", Consumer: "ace_carried"}}) { + t.Fatalf("the carried module is told to bind %+v", carried.Hears) + } + mailer := MembershipFor("ace", records.Assigned["ace"][2], where) + if !reflect.DeepEqual(mailer.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_mailer"}}) { + t.Fatalf("the mailer is told to bind %+v", mailer.Hears) + } + withheld := WithheldOf(records) + if len(withheld) != 1 || withheld[0] != (Withheld{Node: "ace", Module: "carried", Stream: "MAIL", Emitter: "mesh-mail"}) { + t.Fatalf("withheld %+v", withheld) + } + if said := withheld[0].Said(); !strings.Contains(said, "carried on ace") || !strings.Contains(said, "MAIL") || + !strings.Contains(said, "runs-as") { + t.Fatalf("said %q", said) + } +} + // The membership says where a module's stream events go and which consumers on module streams to bind // (ADR 0363 §2, §8). func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { @@ -144,7 +241,7 @@ func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) { !slices.Equal(provider.Streams[0].Events, mailStream.Events) { t.Fatalf("the provider's membership: %+v", provider.Streams) } - reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where) + reader := MembershipFor("ace", Declared{Module: "reader", RunsAs: "reader", Hears: []Heard{mailHeard}}, where) if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) { t.Fatalf("the reader's membership: %+v", reader.Hears) } diff --git a/internal/broker/msgttl_live_test.go b/internal/broker/msgttl_live_test.go new file mode 100644 index 00000000..02589420 --- /dev/null +++ b/internal/broker/msgttl_live_test.go @@ -0,0 +1,86 @@ +package broker + +import ( + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/testbus" +) + +// **DEAD_LETTERS and EVENTS take a message's own time to live once updated in place** (novox/hq ADR 0376 §3, §5), +// on the pinned server: each made as it was before, without one, is brought to its definition by the controller's +// assertion, keeps what it held, and removes a message once its time to live passes. A restricted module stream says +// so in its metadata; one without a grant does not. +// +// go test ./internal/broker/ -run TestTheStreamsTakeAMessagesTimeToLive +func TestTheStreamsTakeAMessagesTimeToLive(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + for _, s := range MeshStreams() { + if s.Name != EventsStream && s.Name != DeadLettersStream { + if s.AllowMsgTTL { + t.Errorf("%s takes a time to live of a message's own, which nothing asks of it", s.Name) + } + continue + } + if !s.AllowMsgTTL { + t.Fatalf("%s is not defined to take a message's time to live", s.Name) + } + _ = js.js.DeleteStream(s.Name) + before := s + before.AllowMsgTTL = false + if err := js.EnsureStream(before); err != nil { + t.Fatalf("%s as it was: %v", s.Name, err) + } + subject := DeadLetterSubject("EVENTS", "ace_reader") + if s.Name == EventsStream { + subject = "mesh.mod.gitea.event.pull.merged" + } + if _, err := js.js.Publish(subject, []byte("held")); err != nil { + t.Fatal(err) + } + if err := js.EnsureStream(s); err != nil { + t.Fatalf("%s updated in place: %v", s.Name, err) + } + info, err := js.js.StreamInfo(s.Name) + if err != nil || !info.Config.AllowMsgTTL || info.State.Msgs != 1 { + t.Fatalf("%s after the update: %+v (%v)", s.Name, info, err) + } + msg := &nats.Msg{Subject: subject, Data: []byte("brief"), Header: nats.Header{}} + msg.Header.Set(nats.MsgTTLHdr, "1s") + ack, err := js.js.PublishMsg(msg) + if err != nil { + t.Fatalf("%s refused a message's time to live: %v", s.Name, err) + } + deadline := time.Now().Add(8 * time.Second) + for { + if _, err := js.js.GetMsg(s.Name, ack.Sequence); err != nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("%s still holds a message past its time to live", s.Name) + } + time.Sleep(200 * time.Millisecond) + } + if info, _ := js.js.StreamInfo(s.Name); info.State.Msgs != 1 { + t.Fatalf("%s holds %d, want only the message without an age", s.Name, info.State.Msgs) + } + _ = js.js.PurgeStream(s.Name) + } + if !mailStream.Stream().Restricted || newsStream.Stream().Restricted { + t.Fatal("a module stream does not say whether it is restricted") + } + _ = js.js.DeleteStream("MAIL") + if err := js.EnsureStream(mailStream.Stream()); err != nil { + t.Fatal(err) + } + defer func() { _ = js.js.DeleteStream("MAIL") }() + if info, _ := js.js.StreamInfo("MAIL"); info.Config.Metadata[RestrictedMetadata] != "true" { + t.Fatalf("MAIL's metadata: %v", info.Config.Metadata) + } +} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 6554a0b0..b1f0b831 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -857,9 +857,10 @@ func PermissionsFor(p Principal) (Permissions, error) { for _, d := range p.Carries { own := Principal{Kind: KindModule, Node: p.Node, Module: d.Module, Emits: d.Emits, Consumes: d.Consumes, Serves: d.Serves, Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, - Streams: d.Streams, Hears: d.Hears} - // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8). - pub = append(pub, heardGrants(own, d.Hears)...) + Streams: d.Streams, Hears: ComposedHears(d)} + // And the module streams it hears, each through its own consumer there (novox/hq ADR 0363 §8) — never + // a restricted one, which only a module on an account of its own hears (ADR 0376 §1). + pub = append(pub, heardGrants(own, own.Hears)...) if _, consumes := ConsumerFor(own); !consumes { continue } diff --git a/internal/broker/streams.go b/internal/broker/streams.go index 124dfa86..54d8c4c1 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -59,8 +59,17 @@ type Stream struct { // DuplicatesSeconds is the window in which a message id published twice is kept once; zero for the // server's default (two minutes). DuplicatesSeconds int + // AllowMsgTTL lets a message carry a time to live of its own (`Nats-TTL`), for what must not outlive the + // stream it came from (novox/hq ADR 0376 §3, §5). + AllowMsgTTL bool + // Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own + // metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3). + Restricted bool } +// RestrictedMetadata is the key a restricted stream's metadata says it under. +const RestrictedMetadata = "mesh-restricted" + // AssignmentsStream holds every assignment's membership, the newest per subject. const AssignmentsStream = "ASSIGNMENTS" @@ -174,6 +183,8 @@ func MeshStreams() []Stream { Retention: RetentionLimits, MaxAge: 7 * 24 * 60 * 60, MaxMsgsPerSubject: 10000, + // A restricted dead letter delivered again keeps what is left of its stream's age (ADR 0376 §5). + AllowMsgTTL: true, Why: "a subscriber that was down catches up; tool traffic under the same prefix is " + "excluded by the event token; per-subject caps keep a noisy emitter from " + "evicting a quiet one without splitting the stream", @@ -194,9 +205,11 @@ func MeshStreams() []Stream { MaxBytes: DeadLettersBytes, DiscardNew: true, DuplicatesSeconds: 24 * 60 * 60, + AllowMsgTTL: true, Why: "every message a consumer gave up on, with its consumer, subject, deliveries and when, kept " + - "until a person delivers it again or drops it with why (novox/hq issue 330); no age, and full " + - "it refuses the next copy rather than drop one it holds", + "until a person delivers it again or drops it with why (novox/hq issue 330); no age but a restricted " + + "stream's message, which ages out with its stream (ADR 0376), and full it refuses the next copy " + + "rather than drop one it holds", }, } } diff --git a/internal/broker/users.go b/internal/broker/users.go index 7ec4beb3..9ff3f208 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -139,13 +139,15 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, - Logs: logNames(d.Logs), Streams: d.Streams, Hears: d.Hears, + Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), }) } if runtimeHere { var carried []Declared for _, d := range r.Assigned[node] { if d.RunsAs == "" { + // 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/internal/catalogue/holds_test.go b/internal/catalogue/holds_test.go new file mode 100644 index 00000000..2a39fb71 --- /dev/null +++ b/internal/catalogue/holds_test.go @@ -0,0 +1,100 @@ +package catalogue + +import ( + "encoding/json" + "strings" + "testing" +) + +// novox/hq ADR 0377 §3: a move under a module's own `record` rolls out when it recreates none of its containers +// and moves none of its data, judged from the two builds' manifests. +func TestWhatAMoveUnderRecordHoldsFor(t *testing.T) { + base := func() Manifest { + raw := `{"module":"mailu","tools":["mailu_status"], + "requires":["postgres-database"], + "data":{"own":[{"id":"mail","path":"${dir:data-mail}","class":"valuable"}]}, + "bundles":[{"name":"tools","source":"store/mailu-tools@sha256:aaa","digest":"sha256:aaa"}], + "resources":[ + {"id":"state","type":"directory"}, + {"id":"mesh-state","type":"directory"}, + {"id":"data-mail","type":"directory"}, + {"id":"config-env","type":"file","path":"${dir:state}/mailu.env","content":"A=1"}, + {"id":"runtime-config","type":"file","path":"${dir:mesh-state}/config.json","content":"{}"}, + {"id":"inbox-note","type":"file","path":"${dir:data-mail}/README","content":"x"}, + {"id":"imap","type":"container","image":"registry/imap@sha256:1","volumes":["${dir:data-mail}:/mail"], + "env-file":["${dir:state}/mailu.env"]}, + {"id":"watcher","type":"process","command":"watch"}, + {"id":"installation","type":"archive","path":"/opt/mailu-tools","artifact":"archive"}, + {"id":"guard","type":"package","package":"nftables"}]}` + var m Manifest + if err := json.Unmarshal([]byte(raw), &m); err != nil { + t.Fatal(err) + } + return m + } + resource := func(m *Manifest, id string) map[string]any { + for _, r := range m.Resources { + if r["id"] == id { + return r + } + } + t.Fatalf("no resource %s", id) + return nil + } + cases := []struct { + name string + change func(m *Manifest) + holds string // empty: it rolls out + }{ + {"nothing", func(*Manifest) {}, ""}, + {"a tools bundle", func(m *Manifest) { + m.Bundles[0].Digest, m.Bundles[0].Source = "sha256:bbb", "store/mailu-tools@sha256:bbb" + }, ""}, + {"the tool list", func(m *Manifest) { m.Tools = append(m.Tools, "mailu_queue") }, ""}, + {"a file no container mounts", func(m *Manifest) { resource(m, "runtime-config")["content"] = `{"a":1}` }, ""}, + {"a container's image", func(m *Manifest) { resource(m, "imap")["image"] = "registry/imap@sha256:2" }, "recreates 1 of mailu's 1"}, + {"a container's declaration", func(m *Manifest) { resource(m, "imap")["health"] = map[string]any{"kind": "runtime"} }, "recreates 1"}, + {"a container added", func(m *Manifest) { + m.Resources = append(m.Resources, map[string]any{"id": "smtp", "type": "container", "image": "x"}) + }, "recreates"}, + {"a mounted file", func(m *Manifest) { resource(m, "config-env")["content"] = "A=2" }, "which its container imap mounts"}, + {"a file within declared data", func(m *Manifest) { resource(m, "inbox-note")["content"] = "y" }, "which its container imap mounts"}, + {"a directory holding declared data", func(m *Manifest) { resource(m, "data-mail")["mode"] = "0700" }, "data-mail"}, + {"the data declaration", func(m *Manifest) { m.Data.Own[0].Class = ClassIrreplaceable }, "data declaration"}, + {"a provider", func(m *Manifest) { m.Requires = append(m.Requires, "s3") }, "what it requires"}, + {"what it provides", func(m *Manifest) { m.Provides = append(m.Provides, Offer{Name: "smtp"}) }, "what it provides"}, + {"a process of a module with data", func(m *Manifest) { resource(m, "watcher")["command"] = "watch2" }, "long-running process watcher"}, + {"an archive resource", func(m *Manifest) { resource(m, "installation")["artifact"] = "archive2" }, "the archive installation"}, + {"a package resource", func(m *Manifest) { resource(m, "guard")["package"] = "nftables2" }, "the package guard"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + from, to := base(), base() + c.change(&to) + got := Holds(from, to) + switch { + case c.holds == "" && got != "": + t.Fatalf("held: %s", got) + case c.holds != "" && !strings.Contains(got, c.holds): + t.Fatalf("holds %q; want it to say %q", got, c.holds) + } + }) + } + // A data directory nothing mounts still holds its data. + from, to := base(), base() + from.Data.Own = append(from.Data.Own, DataItem{ID: "archive", Path: "${dir:mesh-state}/archive", Class: "valuable"}) + to.Data = &Data{Own: append([]DataItem(nil), from.Data.Own...)} + resource(&to, "runtime-config")["path"] = "${dir:mesh-state}/archive/config.json" + resource(&from, "runtime-config")["path"] = "${dir:mesh-state}/archive/config.json" + resource(&to, "runtime-config")["content"] = "new" + if got := Holds(from, to); !strings.Contains(got, "lies within its data archive") { + t.Fatalf("a file within data nothing mounts: %q", got) + } + // A process of a module with no data of its own may change. + from, to = base(), base() + from.Data, to.Data = nil, nil + resource(&to, "watcher")["command"] = "watch2" + if got := Holds(from, to); got != "" { + t.Fatalf("a process of a module with no data held: %s", got) + } +} diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index 845b9122..2d82cbff 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -14,6 +14,7 @@ import ( "sort" "strconv" "strings" + "unicode/utf8" ) // Scopes a claim can have. @@ -2190,6 +2191,7 @@ func ParseManifest(raw []byte) (Manifest, error) { "may make it (novox/hq ADR 0228)", m.Module, name, own.IssuedBy, IssuedOutside)) } + problems = append(problems, ownSecretPromptProblems(m, name, own.Prompt, IsFamily(name))...) } localOf := map[string]string{} for _, to := range m.SecretRequirements() { @@ -2759,6 +2761,11 @@ type OwnSecret struct { Path string Taken string IssuedBy string + // Prompt is the one line the desk asks for this secret as, in place of its bare name (novox/hq + // issue 789): at most PromptMax characters, one line, holding only the {member} placeholder — a + // family's member suffix, filled where the secret is asked for. Optional; reviewed in the + // module's pull request like the rest of the manifest. + Prompt string } // MeshMayMake says the mesh may put a value it makes in place of the one the secret holds: the @@ -2769,6 +2776,50 @@ func (s OwnSecret) MeshMayMake() bool { return s.Taken == TakenAtStart && s.IssuedBy != IssuedOutside } +// PromptMax is the longest an own secret's prompt may be (novox/hq issue 789): one line, short enough that the +// desk's prompt reads at a glance. +const PromptMax = 120 + +// promptPlaceholder is a `{word}` token in a prompt's line; the only one a prompt may hold is `{member}`, a +// family's member suffix (novox/hq issue 789). +var promptPlaceholder = regexp.MustCompile(`\{[^{}]*\}`) + +// ownSecretPromptProblems is what is wrong with an own secret's declared prompt, in the module check's words: +// empty is fine (no prompt is declared, and the desk asks by name alone as it does today); otherwise one line, +// at most PromptMax characters — counting what a person reads, not its bytes — holding only the `{member}` +// placeholder, and only where the secret is a family: a single secret's prompt has no member to fill, so +// `{member}` there would stay in the prompt literally (novox/hq issue 789). +func ownSecretPromptProblems(m Manifest, name, prompt string, family bool) []string { + if prompt == "" { + return nil + } + var problems []string + if strings.ContainsAny(prompt, "\n\r") { + problems = append(problems, fmt.Sprintf( + "%s names its secret %q with a prompt that holds more than one line; a prompt is one line (novox/hq issue 789)", + m.Module, name)) + } + if n := utf8.RuneCountInString(prompt); n > PromptMax { + problems = append(problems, fmt.Sprintf( + "%s names its secret %q with a prompt %d characters long; a prompt is at most %d characters", + m.Module, name, n, PromptMax)) + } + for _, ph := range promptPlaceholder.FindAllString(prompt, -1) { + if ph != "{member}" { + problems = append(problems, fmt.Sprintf( + "%s names its secret %q with a prompt that holds %s; a prompt holds only the {member} placeholder, "+ + "filled with a family's member suffix (novox/hq issue 789)", + m.Module, name, ph)) + } else if !family { + problems = append(problems, fmt.Sprintf( + "%s names its secret %q with a prompt that holds {member}; {member} is replaced only for a "+ + "family's member (novox/hq issue 789)", + m.Module, name)) + } + } + return problems +} + // OwnSecrets is a module's own secrets by name. On the wire each is a path, or an object naming // the path and how it is taken; written back the way it was read, so a manifest the mesh holds // keeps its bytes. @@ -2790,13 +2841,14 @@ func (o *OwnSecrets) UnmarshalJSON(raw []byte) error { Path string `json:"path"` Taken string `json:"taken,omitempty"` IssuedBy string `json:"issued-by,omitempty"` + Prompt string `json:"prompt,omitempty"` } dec := json.NewDecoder(bytes.NewReader(body)) dec.DisallowUnknownFields() if err := dec.Decode(&long); err != nil { - return fmt.Errorf("own-secrets.%s: a path, or {\"path\", \"taken\", \"issued-by\"}: %w", name, typedUnknown(err)) + return fmt.Errorf("own-secrets.%s: a path, or {\"path\", \"taken\", \"issued-by\", \"prompt\"}: %w", name, typedUnknown(err)) } - out[name] = OwnSecret{Path: long.Path, Taken: long.Taken, IssuedBy: long.IssuedBy} + out[name] = OwnSecret{Path: long.Path, Taken: long.Taken, IssuedBy: long.IssuedBy, Prompt: long.Prompt} } *o = out return nil @@ -2805,7 +2857,7 @@ func (o *OwnSecrets) UnmarshalJSON(raw []byte) error { func (o OwnSecrets) MarshalJSON() ([]byte, error) { entries := make(map[string]any, len(o)) for name, s := range o { - if s.Taken == "" && s.IssuedBy == "" { + if s.Taken == "" && s.IssuedBy == "" && s.Prompt == "" { entries[name] = s.Path continue } @@ -2816,6 +2868,9 @@ func (o OwnSecrets) MarshalJSON() ([]byte, error) { if s.IssuedBy != "" { long["issued-by"] = s.IssuedBy } + if s.Prompt != "" { + long["prompt"] = s.Prompt + } entries[name] = long } return json.Marshal(entries) @@ -2854,7 +2909,8 @@ func (o OwnSecrets) Lookup(name string) (s OwnSecret, family string, ok bool) { continue } if family == "" || len(declared) > len(family) { - s, family, ok = OwnSecret{Path: strings.Replace(f.Path, "*", rest, 1), Taken: f.Taken, IssuedBy: f.IssuedBy}, declared, true + s, family, ok = OwnSecret{Path: strings.Replace(f.Path, "*", rest, 1), Taken: f.Taken, IssuedBy: f.IssuedBy, + Prompt: f.Prompt}, declared, true } } return s, family, ok diff --git a/internal/catalogue/own_secret_prompt_test.go b/internal/catalogue/own_secret_prompt_test.go new file mode 100644 index 00000000..f0d65750 --- /dev/null +++ b/internal/catalogue/own_secret_prompt_test.go @@ -0,0 +1,111 @@ +package catalogue + +import ( + "encoding/json" + "strings" + "testing" +) + +// An own secret (or a family) may declare `prompt`, one line naming what it is for a desk's prompt in place of +// its bare name (novox/hq issue 789): reviewed in the module's pull request, and the controller validates it in +// the module check. +func TestAnOwnSecretsPromptParsesAndIsWrittenBack(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","own-secrets":{ + "mail-credential-*":{"path":"/var/lib/mesh-mail/mail-credential-*.secret","issued-by":"outside", + "prompt":"the password of the mail account {member}"}, + "smtp-relay-key":{"path":"/var/lib/mesh-mail/smtp-relay-key","taken":"at-start", + "prompt":"the key for the outbound relay"}}}`)) + if err != nil { + t.Fatalf("a manifest with a well-formed prompt was refused: %v", err) + } + if got := m.OwnSecrets["mail-credential-*"].Prompt; got != "the password of the mail account {member}" { + t.Fatalf("the family's prompt: %q", got) + } + if got := m.OwnSecrets["smtp-relay-key"].Prompt; got != "the key for the outbound relay" { + t.Fatalf("the plain secret's prompt: %q", got) + } + // Written back the way it was read, so a registered manifest keeps its bytes. + out, err := json.Marshal(m.OwnSecrets) + if err != nil { + t.Fatal(err) + } + var again OwnSecrets + if err := json.Unmarshal(out, &again); err != nil { + t.Fatal(err) + } + if again["mail-credential-*"].Prompt != "the password of the mail account {member}" || + again["smtp-relay-key"].Prompt != "the key for the outbound relay" { + t.Fatalf("the round trip lost the prompt: %s", out) + } +} + +// No prompt declared is fine: the desk asks by name alone, as it does today. +func TestAnOwnSecretWithNoPromptIsFine(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"telegram","version":"1","own-secrets":{ + "telegram-token":{"path":"/var/lib/telegram/telegram-token","issued-by":"outside"}}}`)) + if err != nil { + t.Fatal(err) + } + if m.OwnSecrets["telegram-token"].Prompt != "" { + t.Fatalf("an undeclared prompt: %q", m.OwnSecrets["telegram-token"].Prompt) + } + out, err := json.Marshal(m.OwnSecrets) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(out), "prompt") { + t.Fatalf("an empty prompt was written: %s", out) + } +} + +// PromptMax counts characters, not bytes: a prompt of 100 multi-byte characters is well within it even +// though it holds well over 120 bytes (novox/hq issue 789). +func TestAPromptOfMultiByteCharactersIsCountedByCharacter(t *testing.T) { + prompt := strings.Repeat("é", 100) + if n := len(prompt); n <= 120 { + t.Fatalf("the test prompt is %d bytes, not over 120 as intended", n) + } + _, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","own-secrets":{ + "smtp-relay-key":{"path":"/var/lib/mesh-mail/smtp-relay-key","taken":"at-start", + "prompt":` + jsonString(prompt) + `}}}`)) + if err != nil { + t.Fatalf("a 100-character prompt was refused: %v", err) + } +} + +// {member} is replaced only for a family's member: a single secret's prompt may not hold it (novox/hq issue 789). +func TestAPromptHoldingMemberOnASingleSecretIsRefused(t *testing.T) { + _, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","own-secrets":{ + "smtp-relay-key":{"path":"/var/lib/mesh-mail/smtp-relay-key","taken":"at-start", + "prompt":"For {member}"}}}`)) + if err == nil || !strings.Contains(err.Error(), "{member} is replaced only for a family's member") { + t.Fatalf("a single secret's prompt holding {member}: %v; want a refusal saying so", err) + } +} + +// A prompt that is not one short line naming only {member} is refused by the module check (novox/hq issue 789). +func TestAMalformedPromptIsRefused(t *testing.T) { + for name, c := range map[string]struct{ prompt, says string }{ + "a newline": {"the password of\nthe mail account {member}", "one line"}, + "over 120 characters": {"the password of the mail account {member}, which this mesh needs to log in to " + + "send and receive mail for the operator's account on this machine today", "120 characters"}, + "a placeholder other than {member}": {"the password of the mail account {name}", "{member}"}, + "two placeholders, one unknown": {"the {kind} secret of the mail account {member}", "{member}"}, + } { + _, err := ParseManifest([]byte(`{"module":"mesh-mail","version":"1","own-secrets":{ + "mail-credential-*":{"path":"/var/lib/mesh-mail/mail-credential-*.secret","issued-by":"outside", + "prompt":` + jsonString(c.prompt) + `}}}`)) + if err == nil || !strings.Contains(err.Error(), c.says) { + t.Errorf("%s: %v; want a refusal saying %q", name, err, c.says) + } + } +} + +// jsonString quotes a Go string the way encoding/json would, for building a literal into a test's manifest JSON. +func jsonString(s string) string { + out, err := json.Marshal(s) + if err != nil { + panic(err) + } + return string(out) +} diff --git a/internal/catalogue/recreates.go b/internal/catalogue/recreates.go index db8aab2c..431a3f81 100644 --- a/internal/catalogue/recreates.go +++ b/internal/catalogue/recreates.go @@ -109,3 +109,224 @@ func sameDeclaration(a, b any) bool { _ = json.Unmarshal(rb, &nb) return reflect.DeepEqual(na, nb) } + +// What a move under a module's own `record` may do without a person (novox/hq ADR 0377). +// +// A module records because its manifest says so or because it keeps irreplaceable data: mail, whose containers +// recreated together interrupt everyone's mail, and the database server, whose restart drops every consumer. That +// hold also held the moves that interrupt nothing — on 2026-10-15 the database server moved with only its tools +// bundle changed, its walk said it "recreates none of its containers", and the machine stayed behind until a +// person pushed. So such a move rolls out like any other, and only a move that recreates a container or touches +// data still waits. + +// Holds is why a move from one build's manifest to another's still waits for a person under a module's own +// `record` (ADR 0377 §3), empty when it recreates none of its containers and moves none of its data: no container +// changed, added or gone; no file or directory changed, added or gone that a container mounts, or that holds or +// lies within a path of the module's declared data; no change to its `data` declaration, nor to what it provides, +// requires or contributes; and, for a module that declares data of its own, no change to a long-running process. +// Its tools bundles, a file nothing mounts that holds none of its data, and manifest fields that recreate +// nothing — its tool list — may differ. +func Holds(from, to Manifest) string { + module := to.Module + if module == "" { + module = from.Module + } + if said := Recreates(from, to).Say(module); said != "" { + return said + } + if !sameDeclaration(from.Data, to.Data) { + return "it changes " + module + "'s data declaration" + } + for _, f := range []struct { + what string + was, s any + }{ + {"what it provides", from.Provides, to.Provides}, + {"what it requires", from.Requires, to.Requires}, + {"which of its requirements are optional", from.OptionalRequires, to.OptionalRequires}, + {"what it contributes to a provider", from.Contributes, to.Contributes}, + {"what it contributes to a provider", from.ContributesMany, to.ContributesMany}, + // A setting is filled into the files the module writes: a changed default changes what a container may + // mount without the file's own declaration changing. + {"its settings, which the files it writes are filled from", from.Settings, to.Settings}, + } { + if !sameDeclaration(emptyAsNil(f.was), emptyAsNil(f.s)) { + return "it changes " + f.what + } + } + mounted := append(mountSources(from), mountSources(to)...) + data := append(dataPaths(from), dataPaths(to)...) + keepsData := (from.Data != nil && len(from.Data.Own) > 0) || (to.Data != nil && len(to.Data.Own) > 0) + before, after := resourcesByID(from), resourcesByID(to) + var changed []string + for id, r := range after { + if was, held := before[id]; !held || !sameDeclaration(was, r) { + changed = append(changed, id) + } + } + for id := range before { + if _, kept := after[id]; !kept { + changed = append(changed, id) + } + } + sort.Strings(changed) + dirs := absoluteDirectories(from, to) + for _, key := range changed { + for _, r := range []map[string]any{before[key], after[key]} { + if r == nil { + continue + } + kind, id := fmt.Sprint(r["type"]), fmt.Sprint(r["id"]) + switch kind { + case "file", "directory": + for _, p := range pathsOf(r, dirs) { + for _, m := range mounted { + if pathsMeet(p, expandDir(m.source, dirs)) { + return fmt.Sprintf("it changes the %s %s, which its container %s mounts", kind, id, m.container) + } + } + for _, d := range data { + if pathsMeet(p, expandDir(d.path, dirs)) { + return fmt.Sprintf("it changes the %s %s, which holds or lies within its data %s", kind, id, d.id) + } + } + } + case "process", "service": + if keepsData { + return fmt.Sprintf("it changes its long-running %s %s, and it keeps data of its own", kind, id) + } + default: + // Anything that is not a file or directory (judged above by whether a container mounts it + // or it holds data) always holds: an archive, a package, a user, a network, or any type + // added later (novox/hq ADR 0377 §3's allowlist names only tools bundles and manifest + // fields that recreate nothing). + return fmt.Sprintf("it changes the %s %s", kind, id) + } + } + } + return "" +} + +type mountSource struct{ container, source string } + +// mountSources is every path a container of the module reads from its machine: its volumes' sources and its +// environment files. +func mountSources(m Manifest) []mountSource { + var out []mountSource + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) != "container" { + continue + } + id := fmt.Sprint(r["id"]) + vols, _ := r["volumes"].([]any) + for _, v := range vols { + s, _ := v.(string) + if mount := volumeMount.FindStringSubmatch(s); mount != nil { + out = append(out, mountSource{id, mount[1]}) + } + } + switch files := r["env-file"].(type) { + case string: + out = append(out, mountSource{id, files}) + case []any: + for _, f := range files { + if s, ok := f.(string); ok { + out = append(out, mountSource{id, s}) + } + } + } + } + return out +} + +type dataPath struct{ id, path string } + +func dataPaths(m Manifest) []dataPath { + if m.Data == nil { + return nil + } + var out []dataPath + for _, it := range m.Data.Own { + out = append(out, dataPath{it.ID, it.Path}) + } + return out +} + +func resourcesByID(m Manifest) map[string]map[string]any { + out := map[string]map[string]any{} + for _, r := range m.Resources { + out[fmt.Sprint(r["type"])+":"+fmt.Sprint(r["id"])] = r + } + return out +} + +// absoluteDirectories is, per directory id, the absolute path a directory of either build states. +func absoluteDirectories(ms ...Manifest) map[string]string { + out := map[string]string{} + for _, m := range ms { + for _, r := range m.Resources { + if fmt.Sprint(r["type"]) != "directory" { + continue + } + if p, ok := r["path"].(string); ok && strings.HasPrefix(p, "/") { + out[fmt.Sprint(r["id"])] = strings.TrimRight(p, "/") + } + } + } + return out +} + +// pathsOf is every way a file or directory resource is named: its path, and a directory's `${dir:}`. +func pathsOf(r map[string]any, dirs map[string]string) []string { + var out []string + if p, ok := r["path"].(string); ok && p != "" { + out = append(out, expandDir(p, dirs)) + } + if fmt.Sprint(r["type"]) == "directory" { + out = append(out, expandDir("${dir:"+fmt.Sprint(r["id"])+"}", dirs)) + } + return out +} + +// expandDir writes a leading `${dir:}` as the directory's absolute path when one is stated, so a path named +// either way is one path. +func expandDir(p string, dirs map[string]string) string { + p = strings.TrimRight(p, "/") + if ref := dirPlain.FindStringSubmatch(p); ref != nil { + if abs, ok := dirs[ref[1]]; ok { + return abs + } + return p + } + if strings.HasPrefix(p, "${dir:") { + end := strings.Index(p, "}") + if abs, ok := dirs[p[len("${dir:"):end]]; ok { + return abs + p[end+1:] + } + } + return p +} + +// pathsMeet is whether one path is the other or lies within it. +func pathsMeet(a, b string) bool { + if a == "" || b == "" { + return false + } + return a == b || strings.HasPrefix(a, b+"/") || strings.HasPrefix(b, a+"/") +} + +// emptyAsNil reads an empty list or map as nothing, so a field written empty is no change from one left out. +func emptyAsNil(v any) any { + rv := reflect.ValueOf(v) + switch rv.Kind() { + case reflect.Slice, reflect.Map: + if rv.Len() == 0 { + return nil + } + case reflect.Ptr: + if rv.IsNil() { + return nil + } + } + return v +} diff --git a/internal/catalogue/seats_declared.go b/internal/catalogue/seats_declared.go index e6e51cb0..77c64810 100644 --- a/internal/catalogue/seats_declared.go +++ b/internal/catalogue/seats_declared.go @@ -247,6 +247,8 @@ func CatalogueProblems(shelf Shelf) []string { var problems []string // A stream name has one owner (novox/hq ADR 0363). problems = append(problems, StreamNameProblems(shelf)...) + // A module that hears a restricted stream runs as an account of its own (novox/hq ADR 0376 §2). + problems = append(problems, RestrictedHearingProblems(shelf)...) // Who declares what, and who declared it first. declaredBy := map[string]string{} diff --git a/internal/catalogue/streams.go b/internal/catalogue/streams.go index 0fd19126..105d3d50 100644 --- a/internal/catalogue/streams.go +++ b/internal/catalogue/streams.go @@ -311,6 +311,41 @@ func StreamNameProblems(shelf Shelf) []string { return problems } +// RestrictedHearingProblems refuses a module that would hear a restricted stream without an account of its own +// (novox/hq ADR 0376 §2): a `consumes` naming a module literally and asking for events a restricted stream of +// that module carries, by a manifest that declares no `runs-as`. A module the machine's tool runner carries +// reaches the bus as the tool runner's account, which every agent session there shares, so the controller would +// compose nothing for it. Judged only where the provider is on the shelf, as a consumer may be installed before +// its emitter; a wildcard emitter never reaches a module's stream (ADR 0363 §5) and is not judged. +func RestrictedHearingProblems(shelf Shelf) []string { + var problems []string + for _, module := range shelfOrder(shelf) { + m := shelf[module] + if m.RunsAs != "" { + continue + } + for _, c := range m.Consumes { + emitter, rest, named := strings.Cut(c, ".") + provider, known := shelf[emitter] + if !named || emitter == OneName || emitter == TheRest || !known { + continue + } + hearing, _ := provider.StreamsHearing(rest) + for _, s := range hearing { + if s.Grant == nil { + continue + } + problems = append(problems, fmt.Sprintf( + "%s consumes %s, which %s's restricted stream %s carries, so %s must run as an account of its "+ + "own (runs-as): the machine's runtime runs as the operator's account, which every agent "+ + "runs as, and never hears a restricted stream (novox/hq ADR 0376)", + module, c, emitter, s.BusName(), module)) + } + } + } + return problems +} + // GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the // module wherever it runs, or the module on this machine as `.`. A value of another shape // grants nothing; a key that is not one usable name grants nothing. diff --git a/internal/catalogue/streams_test.go b/internal/catalogue/streams_test.go index 5ea8c4e7..a5971e3c 100644 --- a/internal/catalogue/streams_test.go +++ b/internal/catalogue/streams_test.go @@ -204,3 +204,41 @@ func TestNarrowingTwoPatterns(t *testing.T) { t.Fatalf("uncovered: %v", got) } } + +// **A module that hears a restricted stream runs as an account of its own** (ADR 0376 §2): registration refuses a +// `consumes` naming a module and asking for events its restricted stream carries, unless the manifest declares +// `runs-as`; a stream without a grant, an event the provider says on EVENTS, and a wildcard emitter stay as they +// were, and a consumer whose provider is not on the shelf is judged once it is. +func TestHearingARestrictedStreamNeedsAnAccountOfItsOwn(t *testing.T) { + provider, err := ParseManifest([]byte(mailManifest(t, mailStream, `"emits":["account.added"],`))) + if err != nil { + t.Fatal(err) + } + feed := Manifest{Module: "feed", Streams: []StreamDeclaration{{Name: "items", Events: []string{"item.**"}}}} + carried := Manifest{Module: "reader", Consumes: []string{"mesh-mail.mail.*.message.added"}} + got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": carried}) + if len(got) != 1 || !strings.Contains(got[0], "reader consumes mesh-mail.mail.*.message.added") || + !strings.Contains(got[0], "MAIL") || !strings.Contains(got[0], "runs-as") || !strings.Contains(got[0], "ADR 0376") { + t.Fatalf("a carried module hearing MAIL was not refused as it should be: %v", got) + } + if all := strings.Join(CatalogueProblems(Shelf{"mesh-mail": provider, "reader": carried}), "; "); !strings.Contains(all, got[0]) { + t.Fatalf("the catalogue's check does not refuse it: %s", all) + } + own := carried + own.RunsAs = "reader" + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "reader": own}); len(got) != 0 { + t.Fatalf("a module of its own account was refused: %v", got) + } + for _, m := range []Manifest{ + {Module: "events", Consumes: []string{"mesh-mail.account.added"}}, + {Module: "items", Consumes: []string{"feed.item.**"}}, + {Module: "audit", Consumes: []string{"**", "*.mail.**"}}, + } { + if got := RestrictedHearingProblems(Shelf{"mesh-mail": provider, "feed": feed, m.Module: m}); len(got) != 0 { + t.Errorf("%s was refused: %v", m.Module, got) + } + } + if got := RestrictedHearingProblems(Shelf{"reader": carried}); len(got) != 0 { + t.Fatalf("judged without its provider: %v", got) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index e8cf32f5..bb6f9491 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -496,7 +496,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node }) var out []broker.Heard for _, p := range parts { - h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter} + h := broker.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil} for _, pattern := range p.Patterns { h.Filters = append(h.Filters, broker.ModuleStreamSubject(p.Emitter, broker.WirePattern(pattern))) } diff --git a/internal/inventory/catalogue.go b/internal/inventory/catalogue.go index 6ce3740c..723df3cc 100644 --- a/internal/inventory/catalogue.go +++ b/internal/inventory/catalogue.go @@ -1476,6 +1476,16 @@ type CurrentBuild struct { Commit string // RollOut is the module's upgrade policy, as Upgrade.RollOut. RollOut bool + // From is where the policy came from, as Upgrade.From: a move under a module's own `record` — its manifest's or + // its irreplaceable data's — rolls out when it recreates nothing and moves no data (novox/hq ADR 0377), one under + // a person's or the bus's never does. + From string +} + +// OwnRecord is whether a module records because its manifest says so or because it keeps irreplaceable data +// (novox/hq ADR 0377 §1): the `record` whose moves that recreate nothing roll out. +func (b CurrentBuild) OwnRecord() bool { + return !b.RollOut && (b.From == catalogue.FromModule || b.From == catalogue.FromData) } // CurrentBuilds is every module's current build and upgrade policy, in one read: what a send records @@ -1497,7 +1507,8 @@ func (i *Inventory) CurrentBuilds(ctx context.Context) (map[string]CurrentBuild, if err := rows.Scan(&name, &b.Commit, &chosen, &together, &why, &by, &manifest); err != nil { return nil, err } - b.RollOut = upgradeFrom(chosen, together, why, by, manifest).RollOut + u := upgradeFrom(chosen, together, why, by, manifest) + b.RollOut, b.From = u.RollOut, u.From out[name] = b } return out, rows.Err() diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index c6f6d323..4b158c7e 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -313,6 +313,11 @@ type PlanModule struct { // Rest is, per machine of the rest, the declaration the send after the first-node gate carried there (novox/hq ADR // 0282 decision 6): the machine's first report of it, applied, is when the build runs there. Rest map[string]SentDeclaration `json:"rest,omitempty"` + // Recorded is, for a module whose policy records, per machine running it with a move, which way the walk took it + // (novox/hq ADR 0377 §6): "went out: …" when the move recreates none of its containers and moves none of its + // data, or "waits: …" with the recreate words, the data change named, the person's or the bus's hold, or why it + // could not be judged. + Recorded map[string]string `json:"recorded,omitempty"` } // PlanGate is one module's rollout record at its gate (to-be 45 §8): the component, the first machine, diff --git a/internal/inventory/secrets.go b/internal/inventory/secrets.go index 0d283381..a4bc7078 100644 --- a/internal/inventory/secrets.go +++ b/internal/inventory/secrets.go @@ -595,6 +595,26 @@ func (i *Inventory) DeclaresOwnSecret(ctx context.Context, module, name string) return GivableAtDesk(m, name) } +// OwnSecretPrompt is the one line the desk asks for this secret as — the module's declared `prompt`, read from +// the controller's own catalogue, never from the call or the module's runtime (novox/hq issue 789). For a family +// member it is the family's prompt with `{member}` replaced by the member's suffix. Empty when the module or the +// secret is unknown, or declares no prompt — the desk then asks by name alone, as it does today. +func (i *Inventory) OwnSecretPrompt(ctx context.Context, module, name string) (string, error) { + m, err := i.declared(ctx, module) + if err != nil { + return "", err + } + own, family, ok := m.OwnSecrets.Lookup(name) + if !ok || own.Prompt == "" { + return "", nil + } + if family == "" { + return own.Prompt, nil + } + member := strings.TrimPrefix(name, catalogue.FamilyPrefix(family)) + return strings.ReplaceAll(own.Prompt, "{member}", member), nil +} + // RunsAsItsOwnAccount says a module runs as an account of its own (novox/hq ADR 0259 §8): a trusted party — the // router, a channel that proves its sender or shows a link's code — whose own secret is what the operator's // answers are believed by. Its value is given at the controller's terminal alone. diff --git a/internal/inventory/sent_builds_test.go b/internal/inventory/sent_builds_test.go index f3888e1f..44015d8e 100644 --- a/internal/inventory/sent_builds_test.go +++ b/internal/inventory/sent_builds_test.go @@ -73,10 +73,10 @@ func TestTheCurrentBuildsAreTheCatalogues(t *testing.T) { t.Fatal(err) } // Rolled out by default, one machine first and gated (novox/hq ADR 0236). - if got := current["resolver"]; got != (CurrentBuild{Commit: "c1", RollOut: true}) { + if got := current["resolver"]; got != (CurrentBuild{Commit: "c1", RollOut: true, From: catalogue.FromDefault}) { t.Errorf("resolver is at %+v", got) } - if got := current["by-hand"]; got != (CurrentBuild{RollOut: true}) { + if got := current["by-hand"]; got != (CurrentBuild{RollOut: true, From: catalogue.FromPerson}) { t.Errorf("a module with no source is at %+v", got) } } diff --git a/internal/inventory/streams_test.go b/internal/inventory/streams_test.go index 9efef62f..dbddf979 100644 --- a/internal/inventory/streams_test.go +++ b/internal/inventory/streams_test.go @@ -32,7 +32,7 @@ func TestWhatAModuleHearsIsComposedOntoTheStreamsSubjects(t *testing.T) { reader := catalogue.Manifest{Module: "reader"} hears, touches, onEvents := hearsOf(reader, map[string]catalogue.Manifest{"mesh-mail": provider}, "ace", grants, []string{"mesh-mail.mail.**", "billing.paid"}) - if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || + if !touches || len(hears) != 1 || hears[0].Stream != "MAIL" || !hears[0].Restricted || !slices.Equal(hears[0].Filters, []string{"mesh.stream.mesh-mail.event.mail.mesh.message.added"}) { t.Fatalf("heard %+v", hears) } diff --git a/internal/link/calls.go b/internal/link/calls.go index 262defee..6332c705 100644 --- a/internal/link/calls.go +++ b/internal/link/calls.go @@ -106,6 +106,13 @@ type CallKeeper interface { Kept(ctx context.Context, id string) (Call, bool, error) // Recent is the kept calls, newest first, without their answers. Recent(ctx context.Context) ([]Call, error) + // MarkAbandoned marks the call id abandoned, unless its record already says it finished (novox/hq + // issue 711): a finish recorded before, or racing, the mark must stand — a call whose answer was + // kept never reads abandoned. It is the record's own compare-and-set that decides the race, not + // the order these two writers happen to run in, so it retries against a finish landing in + // between. It returns the record as it stands once this returns: marked abandoned, the finish + // that won instead, or a zero Call when the record is gone. + MarkAbandoned(ctx context.Context, id string) (Call, error) } // CallLog keeps the latest calls a holder served. @@ -171,14 +178,15 @@ func (l *CallLog) Durably(ctx context.Context, keeper CallKeeper, holder string, if c.State != CallRunning || c.Holder == holder { continue } - whole, found, err := keeper.Kept(ctx, c.ID) - if err != nil || !found { - whole = c - } - whole.State = CallAbandoned - if err := keeper.Keep(ctx, whole); err != nil { + final, err := keeper.MarkAbandoned(ctx, c.ID) + if err != nil { return fmt.Errorf("marking %s abandoned: %w", c.ID, err) } + if final.State != CallAbandoned { + // Its finish was recorded before this sweep reached it, or while it raced this mark + // (novox/hq issue 711): that word stands, with its kept answer, and is never overwritten. + continue + } if logger != nil { logger.Printf("%s (%s.%s, asked %s by %s) was running under %s (epoch %d), which stopped: marked "+ "abandoned — it may have done part of what it was asked, and nothing will finish it", c.ID, c.Seat, diff --git a/internal/link/calls_kept.go b/internal/link/calls_kept.go index 06e4fff2..5adac863 100644 --- a/internal/link/calls_kept.go +++ b/internal/link/calls_kept.go @@ -87,6 +87,53 @@ func (b *BusCalls) Kept(ctx context.Context, id string) (Call, bool, error) { return c, true, nil } +// markAbandonedTries bounds the compare-and-set loop below (novox/hq issue 711 review round 1): a +// record that keeps changing on every read is not this sweep's to keep retrying forever over. +const markAbandonedTries = 10 + +// MarkAbandoned marks id abandoned unless its record already says it finished (novox/hq issue 711): +// a controller starting finds a call running under a holder that stopped, but that holder may have +// written the finish — with its kept answer — around the same moment. The record's own +// compare-and-set decides the race rather than the order these two writers happen to run in: this +// reads the record, and if it still says running, writes abandoned at that revision; a finish landing +// in between fails the write with a revision mismatch, and this rereads and finds the finish, which +// then stands. A record already resolved, or gone, is returned as found, so the caller never writes +// abandoned over it. Past markAbandonedTries attempts, it gives up and returns an error naming the +// call instead of looping forever (novox/hq issue 711 review round 1); Durably reports that as it +// reports any other keeper error. +func (b *BusCalls) MarkAbandoned(ctx context.Context, id string) (Call, error) { + for try := 0; try < markAbandonedTries; try++ { + entry, err := b.kv.Get(ctx, id) + if errors.Is(err, jetstream.ErrKeyNotFound) || errors.Is(err, jetstream.ErrInvalidKey) { + return Call{}, nil + } + if err != nil { + return Call{}, err + } + var c Call + if err := json.Unmarshal(entry.Value(), &c); err != nil { + return Call{}, fmt.Errorf("the record of %s on the bus is not a call: %w", id, err) + } + if c.State != CallRunning { + // A finish (or an earlier mark) is already the record's word: it stands, untouched. + return c, nil + } + c.State = CallAbandoned + record, err := json.Marshal(c) + if err != nil { + return Call{}, err + } + if _, err := b.kv.Update(ctx, id, record, entry.Revision()); err != nil { + if errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + continue // something — a finish, most likely — wrote since: reread and judge it again + } + return Call{}, err + } + return c, nil + } + return Call{}, fmt.Errorf("%s: its record kept changing; it is left as it stands", id) +} + // Recent is every call the bucket holds, newest first, without answers: read once through a watch // of the records, which hands over the current value of each and then says it has. func (b *BusCalls) Recent(ctx context.Context) ([]Call, error) { diff --git a/internal/link/calls_kept_test.go b/internal/link/calls_kept_test.go index 84a23be7..175d14a2 100644 --- a/internal/link/calls_kept_test.go +++ b/internal/link/calls_kept_test.go @@ -105,6 +105,165 @@ func TestNatsACallsOutcomeOutlivesItsController(t *testing.T) { } } +// racingRecent wraps a CallKeeper so a finish lands right after its Recent() snapshot is taken, as +// if that finish raced the moment the snapshot was read (novox/hq issue 711): the snapshot still +// says a call is running, but by the time anything acts on that entry the record has already moved +// on. +type racingRecent struct { + CallKeeper + thenLand func() +} + +func (r racingRecent) Recent(ctx context.Context) ([]Call, error) { + out, err := r.CallKeeper.Recent(ctx) + if err == nil && r.thenLand != nil { + r.thenLand() + } + return out, err +} + +// **A finish recorded before the sweep reaches a call stands, with its kept answer** (novox/hq issue +// 711): a stale "running" snapshot is no excuse to overwrite a record that has since said how the +// call ended. A caller reading `abandoned` for a call that succeeded would redo work already done. +func TestNatsAFinishRacingTheSweepsSnapshotReadsFinished(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-1", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + finished := running + finished.State = CallAnswered + finished.Answer = json.RawMessage(`"anchor told"`) + wrapped := racingRecent{CallKeeper: keeper, thenLand: func() { + if err := keeper.Keep(context.Background(), finished); err != nil { + t.Error(err) + } + waitKept(t, keeper, running.ID, CallAnswered) + }} + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), wrapped, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallAnswered || !strings.Contains(string(got.Answer), "anchor told") { + t.Fatalf("its finish did not stand: %+v %v %v", got, found, err) + } + if strings.Contains(said.String(), running.ID) { + t.Errorf("a finished call was said abandoned: %q", said.String()) + } +} + +// A call left running with no finish recorded is still marked abandoned — unchanged (novox/hq issue +// 711): only a finish actually recorded for it may stand instead. +func TestNatsARunningCallWithNoFinishIsStillAbandoned(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-2", Seat: "mesh-controller", Verb: "plans", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), keeper, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallAbandoned || !strings.Contains(said.String(), running.ID) { + t.Fatalf("got %+v %v %v; said %q", got, found, err, said.String()) + } +} + +// delayedKV wraps a jetstream.KeyValue so a hook runs just before Update, to land a write between a +// reader's Get and its own compare-and-set write. +type delayedKV struct { + jetstream.KeyValue + before func() +} + +func (d delayedKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + d.before() + return d.KeyValue.Update(ctx, key, value, revision) +} + +// **The race where a finish lands after the sweep's own read but before its write: the finish +// wins** (novox/hq issue 711). The sweep's write is a compare-and-set on the revision it read, so a +// finish landing in between invalidates it; the sweep rereads, finds the finish, and leaves it alone. +func TestNatsAFinishRacingTheSweepsWriteWins(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-3", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + finished := running + finished.State = CallFinishedAfter + finished.Answer = json.RawMessage(`"anchor told"`) + landed := false + racey := &BusCalls{kv: delayedKV{KeyValue: keeper.kv, before: func() { + if landed { + return + } + landed = true + if err := keeper.Keep(context.Background(), finished); err != nil { + t.Error(err) + } + waitKept(t, keeper, running.ID, CallFinishedAfter) + }}} + + var said bytes.Buffer + second := NewCallLog() + if err := second.Durably(t.Context(), racey, "controller@two", log.New(&said, "", 0)); err != nil { + t.Fatal(err) + } + got, found, err := second.Get(running.ID) + if err != nil || !found || got.State != CallFinishedAfter || !strings.Contains(string(got.Answer), "anchor told") { + t.Fatalf("the finish did not win the race: %+v %v %v", got, found, err) + } + if strings.Contains(said.String(), running.ID) { + t.Errorf("a finished call was said abandoned: %q", said.String()) + } +} + +// alwaysMismatchKV wraps a jetstream.KeyValue whose Update always fails with a revision mismatch, as +// if some other writer kept winning the race forever (novox/hq issue 711 review round 1). +type alwaysMismatchKV struct { + jetstream.KeyValue + tries *int +} + +func (a alwaysMismatchKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + *a.tries++ + return 0, jetstream.ErrKeyRevisionMismatch +} + +// **MarkAbandoned gives up after markAbandonedTries rather than loop forever** (novox/hq issue 711 +// review round 1): a record whose compare-and-set keeps losing is left as it stands, said in an +// error naming the call, not retried without end. +func TestNatsAMarkAbandonedGivesUpAfterTenTries(t *testing.T) { + keeper := callsBucket(t) + running := Call{ID: "call-711-4", Seat: "mesh-controller", Verb: "push", State: CallRunning, Holder: "controller@one"} + if err := keeper.Keep(t.Context(), running); err != nil { + t.Fatal(err) + } + waitKept(t, keeper, running.ID, CallRunning) + + tries := 0 + stuck := &BusCalls{kv: alwaysMismatchKV{KeyValue: keeper.kv, tries: &tries}} + _, err := stuck.MarkAbandoned(t.Context(), running.ID) + if err == nil || !strings.Contains(err.Error(), running.ID) { + t.Fatalf("expected an error naming %s, got %v", running.ID, err) + } + if tries != markAbandonedTries { + t.Fatalf("tried %d times, want %d", tries, markAbandonedTries) + } +} + // The bucket holds the last thousand calls or fourteen days: a call is two keys, so the stream under // it holds twice as many messages, and asserting it again keeps the count. func TestNatsTheCallsBucketIsBounded(t *testing.T) { diff --git a/internal/link/deadletters.go b/internal/link/deadletters.go index 2b6002dd..94b442cb 100644 --- a/internal/link/deadletters.go +++ b/internal/link/deadletters.go @@ -41,6 +41,10 @@ const ( DeadHeaderLost = "Mesh-Dead-Lost" // DeadHeaderMsgID is the message's own de-duplication id, kept, since the copy carries its own. DeadHeaderMsgID = "Mesh-Dead-Msg-Id" + // DeadHeaderRestricted says the message came from a restricted stream (novox/hq ADR 0376), and + // DeadHeaderAgesOut when its stream's maximum age runs out, counted from when it was first published. + DeadHeaderRestricted = "Mesh-Dead-Restricted" + DeadHeaderAgesOut = "Mesh-Dead-Ages-Out" // AgainHeader marks a message delivered again, with the kept copy's id it came from. AgainHeader = "Mesh-Delivered-Again" ) @@ -65,12 +69,46 @@ type DeadLetter struct { // Original says what became of the ask it was kept from, in its seat's work queue, when it was // delivered again (novox/hq issue 334). Original string `json:"original,omitempty"` - // Body and Headers are the message itself, given only for one dead letter asked by its id; a header - // with several values keeps them all. + // Restricted says it came from a restricted stream (novox/hq ADR 0376): its body and its own headers are + // never shown, said in Withheld, and it ages out at AgesOut, when its stream's maximum age runs out. + Restricted bool `json:"restricted,omitempty"` + AgesOut time.Time `json:"ages_out,omitzero"` + Withheld string `json:"withheld,omitempty"` + // Body and Headers are the message itself, given only for one dead letter asked by its id, and never for + // a restricted one; a header with several values keeps them all. Body string `json:"body,omitempty"` Headers map[string][]string `json:"headers,omitempty"` } +// deadLetterNow is the clock dead letters age by: a variable, so a test can move it. +var deadLetterNow = time.Now + +// leastLeftToDeliver is the least of its stream's age a restricted dead letter must have left to be delivered +// again (novox/hq ADR 0376 §5). +const leastLeftToDeliver = time.Minute + +// withheldWords is what the dead-letters verb says in place of a restricted dead letter's body. +const withheldWords = "its body and its own headers are not shown: it came from a restricted stream, which only " + + "the modules its grant names may read (novox/hq ADR 0376)" + +// sourceAge is whether a stream is restricted (novox/hq ADR 0363 §4), and its maximum age: read from the stream +// itself, where the controller says both. A stream that no longer exists is neither. +func sourceAge(js nats.JetStreamContext, stream string) (bool, time.Duration, error) { + info, err := js.StreamInfo(stream) + if errors.Is(err, nats.ErrStreamNotFound) { + return false, 0, nil + } + if err != nil { + return false, 0, fmt.Errorf("whether %s is restricted cannot be read: %w", stream, err) + } + return info.Config.Metadata[broker.RestrictedMetadata] == "true", info.Config.MaxAge, nil +} + +// ttlOf is a time left as a message's time to live, in whole seconds and never more than what is left. +func ttlOf(left time.Duration) string { + return strconv.FormatInt(int64(left/time.Second), 10) + "s" +} + // maxDeliveries is the part of the server's notice a kept message is made from. type maxDeliveries struct { Stream string `json:"stream"` @@ -94,8 +132,28 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) gaveUp = time.Now() } kept := &nats.Msg{Subject: broker.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}} + // **A restricted stream's message ages out with its stream** (novox/hq ADR 0376 §3): kept with a time to live + // of what is left of the stream's maximum age, counted from when it was first published; one already past it + // is kept as a record without its body. Every other dead letter keeps no age, as before (ADR 0264). + restricted, maxAge, err := sourceAge(js, a.Stream) + if err != nil { + return DeadLetter{}, err + } + if restricted { + kept.Header.Set(DeadHeaderRestricted, "true") + } raw, err := js.GetMsg(a.Stream, a.StreamSeq) + agesOut := time.Time{} + if err == nil && restricted && maxAge > 0 { + agesOut = raw.Time.Add(maxAge) + kept.Header.Set(DeadHeaderAgesOut, agesOut.UTC().Format(time.RFC3339Nano)) + } switch { + case err == nil && !agesOut.IsZero() && agesOut.Sub(deadLetterNow()) < time.Second: + kept.Header.Set(DeadHeaderSubject, raw.Subject) + kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) + kept.Header.Set(DeadHeaderLost, fmt.Sprintf("message %d of %s had passed its stream's maximum age of %s when "+ + "the notice was taken, so it is kept without its body (novox/hq ADR 0376)", a.StreamSeq, a.Stream, maxAge)) case err == nil: for k, v := range raw.Header { if strings.HasPrefix(k, "Nats-") { @@ -109,6 +167,9 @@ func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) kept.Header.Set(DeadHeaderSubject, raw.Subject) kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano)) kept.Data = raw.Data + if !agesOut.IsZero() { + kept.Header.Set(nats.MsgTTLHdr, ttlOf(agesOut.Sub(deadLetterNow()))) + } case errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound): kept.Header.Set(DeadHeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v", a.Stream, a.StreamSeq, err)) @@ -143,6 +204,8 @@ func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole b d.Deliveries, _ = strconv.ParseUint(h.Get(DeadHeaderDeliveries), 10, 64) d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderGaveUp)) d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderPublished)) + d.Restricted = h.Get(DeadHeaderRestricted) == "true" + d.AgesOut, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderAgesOut)) if whole { d.Body = string(data) d.Headers = map[string][]string{} @@ -230,6 +293,9 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer { continue } + if d.Restricted { + d.Withheld = withheldWords + } out = append(out, d) } return out, total, nil @@ -238,8 +304,21 @@ func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLet // ErrNoDeadLetter is a dead letter's id DEAD_LETTERS does not hold. var ErrNoDeadLetter = errors.New("no such dead letter") -// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with. +// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with — **but for one from +// a restricted stream, whose body and own headers are never shown** (novox/hq ADR 0376 §4). A grant names modules, +// and this is asked by people and by tool runners' bus accounts, which a grant never names: withholding the body +// from every caller is that grant's check, said plainly. func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) { + d, err := deadLetterWhole(js, id) + if err != nil || !d.Restricted { + return d, err + } + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + return d, nil +} + +// deadLetterWhole is one kept message whole, a restricted one's body too: for delivering it again, never to show. +func deadLetterWhole(js nats.JetStreamContext, id uint64) (DeadLetter, error) { raw, err := js.GetMsg(broker.DeadLettersStream, id) if errors.Is(err, nats.ErrMsgNotFound) { return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, "+ @@ -286,6 +365,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // 0363 §7): its consumer on that stream must still be there and still filter the subject, or delivering // it again would hand a module what its grant no longer names. Then it is delivered as an event is, on // EVENTS under the consumer's own again-subject, which the module's consumer there filters. + // + // **A restricted one only while at least a minute of its stream's age is left** (novox/hq ADR 0376 §5): + // past that it can only be dropped, and the copy keeps what is left (DeliverAgain). + if d.Restricted { + if left := d.AgesOut.Sub(deadLetterNow()); d.AgesOut.IsZero() || left < leastLeftToDeliver { + return "", fmt.Errorf("dead letter %d has aged out: %s keeps a message only for its maximum age, and "+ + "less than a minute of it is left (it ages out at %s), so it is not delivered again. Drop it "+ + "(novox/hq ADR 0376)", d.ID, d.Stream, d.AgesOut.UTC().Format(time.RFC3339)) + } + } info, err := js.ConsumerInfo(d.Stream, d.Consumer) if errors.Is(err, nats.ErrConsumerNotFound) { return "", fmt.Errorf("dead letter %d was given up by %s, which no longer hears %s, so it would reach a "+ @@ -332,7 +421,16 @@ func AgainTo(js nats.JetStreamContext, d DeadLetter) (string, error) { // from DEAD_LETTERS; an ask's original is removed from its work queue before. The message carries its own headers and AgainHeader; its de-duplication id is the // kept copy's, so asking twice delivers it once. func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { - d, err := DeadLetterNamed(js, id) + d, to, err := deliverAgain(js, id) + if d.Restricted { + // What it said stays unsaid to whoever asked it delivered (novox/hq ADR 0376 §4). + d.Body, d.Headers, d.Withheld = "", nil, withheldWords + } + return d, to, err +} + +func deliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) { + d, err := deadLetterWhole(js, id) if err != nil { return d, "", err } @@ -349,6 +447,10 @@ func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, erro } again.Header.Set(AgainHeader, strconv.FormatUint(id, 10)) again.Header.Set(nats.MsgIdHdr, "again."+broker.DeadLettersStream+"."+strconv.FormatUint(id, 10)) + if d.Restricted { + // The copy keeps what is left of its stream's age (novox/hq ADR 0376 §5). + again.Header.Set(nats.MsgTTLHdr, ttlOf(d.AgesOut.Sub(deadLetterNow()))) + } if _, err := js.PublishMsg(again); err != nil { if d.Original != "" { return d, to, fmt.Errorf("dead letter %d could not be delivered again on %s: %w; it is still kept, so "+ @@ -473,3 +575,65 @@ func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscripti _ = m.Ack() }, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck()) } + +// Aging is one restricted dead letter as it ages (novox/hq ADR 0376 §6): its id, and when its stream's maximum age +// runs out and DEAD_LETTERS lets it go. +type Aging struct { + ID uint64 + AgesOut time.Time +} + +// AgingDeadLetters is, for each consumer of held whose dead letters came from a restricted stream, each one that +// ages out and when: read by their headers alone, never their bodies. A consumer of any other stream is left out, +// and so is a record that keeps no age. +func AgingDeadLetters(js nats.JetStreamContext, held map[string]int) (map[string][]Aging, error) { + out := map[string][]Aging{} + restricted := map[string]bool{} + for key, n := range held { + stream, consumer, _ := strings.Cut(key, ".") + is, known := restricted[stream] + if !known { + r, _, err := sourceAge(js, stream) + if err != nil { + return nil, err + } + restricted[stream], is = r, r + } + if !is || n <= 0 { + continue + } + sub, err := js.SubscribeSync(broker.DeadLetterSubject(stream, consumer), nats.BindStream(broker.DeadLettersStream), + nats.OrderedConsumer(), nats.HeadersOnly(), nats.DeliverAll()) + if err != nil { + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + // held can count more than DEAD_LETTERS still holds for this consumer — one aged out, + // delivered again or dropped since it was counted (novox/hq issue 728 review round 1): read + // the ordered consumer's pending count once, from right when it was subscribed, and look for + // at most that many messages, so a look never waits out NextMsg's bound for one that is not + // there. That bound stays only as a guard against a pending count that is itself stale. + limit := n + if pending, err := sub.InitialConsumerPending(); err == nil && int(pending) < limit { + limit = int(pending) + } + for i := 0; i < limit; i++ { + m, err := sub.NextMsg(2 * time.Second) + if errors.Is(err, nats.ErrTimeout) { + break // delivered again, dropped or aged out since they were counted + } + if err != nil { + _ = sub.Unsubscribe() + return nil, fmt.Errorf("the dead letters of %s cannot be read: %w", key, err) + } + meta, err := m.Metadata() + if err != nil { + continue + } + if at, err := time.Parse(time.RFC3339Nano, m.Header.Get(DeadHeaderAgesOut)); err == nil { + out[key] = append(out[key], Aging{ID: meta.Sequence.Stream, AgesOut: at}) + } + } + _ = sub.Unsubscribe() + } + return out, nil +} diff --git a/internal/link/deadletters_restricted_test.go b/internal/link/deadletters_restricted_test.go new file mode 100644 index 00000000..c621bbc2 --- /dev/null +++ b/internal/link/deadletters_restricted_test.go @@ -0,0 +1,251 @@ +package link + +import ( + "fmt" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// A restricted stream of a module, MAIL, kept a week, on the bus (novox/hq ADR 0363), and the subject an event of +// the account `mesh` is published on. +func aRestrictedStream(t *testing.T, js *broker.JetStream) string { + t.Helper() + mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"}, + MaxAgeDays: 7, MaxMiB: 16, Restricted: true} + _ = js.Context().DeleteStream("MAIL") + if err := js.EnsureStream(mail.Stream()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") }) + return broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added") +} + +// withClock sets the clock dead letters are kept and delivered again by, until the test ends. +func withClock(t *testing.T, now func() time.Time) { + t.Helper() + before := deadLetterNow + deadLetterNow = now + t.Cleanup(func() { deadLetterNow = before }) +} + +// publishMail puts one mail event on MAIL with a header of its own, and answers its sequence and when it was stored. +func publishMail(t *testing.T, js *broker.JetStream, subject string) (uint64, time.Time) { + t.Helper() + msg := &nats.Msg{Subject: subject, Data: []byte(`{"from":"someone@example.org"}`), Header: nats.Header{}} + msg.Header.Set("x-account", "mesh") + ack, err := js.Context().PublishMsg(msg) + if err != nil { + t.Fatal(err) + } + raw, err := js.Context().GetMsg("MAIL", ack.Sequence) + if err != nil { + t.Fatal(err) + } + return ack.Sequence, raw.Time +} + +// notice is the server's maximum-deliveries notice for one message of MAIL. +func notice(seq uint64) []byte { + return []byte(fmt.Sprintf(`{"stream":"MAIL","consumer":"ace_mailer","stream_seq":%d,"deliveries":5}`, seq)) +} + +// **A dead letter from a restricted stream is listed without its body, and ages out with its stream** (ADR 0376 +// §3, §4): given up on by its consumer and kept, the list and the show by id give what it was and when it ages out, +// never its body or its own headers, and the kept copy carries a time to live of what is left of MAIL's week. +func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) { + js := aBus(t) + keeping(t, js) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + _, published := publishMail(t, js, subject) + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + mailer, err := api.Consumer(t.Context(), "MAIL", "ace_mailer") + if err != nil { + t.Fatal(err) + } + d := givenUp(t, js, mailer) + weekLeft := published.Add(7 * 24 * time.Hour) + if !d.Restricted || d.Body != "" || d.Headers != nil || d.Subject != subject || d.Sequence != 1 || d.Deliveries != 5 || + d.Published.IsZero() || d.GaveUp.IsZero() || d.Size == 0 || d.AgesOut.Sub(weekLeft).Abs() > time.Second { + t.Fatalf("listed as %+v", d) + } + shown, err := DeadLetterNamed(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if shown.Body != "" || shown.Headers != nil || !shown.Restricted || shown.Withheld == "" || shown.AgesOut.IsZero() { + t.Fatalf("shown by its id as %+v", shown) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 7*24*time.Hour || ttl < 7*24*time.Hour-time.Minute { + t.Fatalf("kept with a time to live of %q (%v), want what is left of the week", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` { + t.Fatal("the kept copy does not hold the message, so it could not be delivered again") + } +} + +// **Past its stream's age, a restricted dead letter is gone** (ADR 0376 §3), on a fake clock: kept with two seconds +// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left, it is kept as a record +// without its body, which cannot be delivered again. +func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour - 2*time.Second) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if !d.Restricted || d.Lost != "" { + t.Fatalf("kept as %+v", d) + } + eventually(t, "the dead letter aged out", func() bool { + _, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID) + return err != nil + }) + if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 { + t.Fatalf("still held after its age: %v", held) + } + + seq, published = publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) }) + past, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID) + if err != nil { + t.Fatal(err) + } + if len(raw.Data) != 0 || raw.Header.Get("x-account") != "" || !strings.Contains(past.Lost, "maximum age") { + t.Fatalf("kept past its age as %+v with %q", past, raw.Data) + } + if _, _, err := DeliverAgain(js.Context(), past.ID); err == nil || !strings.Contains(err.Error(), "Drop it") { + t.Fatalf("a record past its age was delivered again: %v", err) + } +} + +// **A restricted dead letter is delivered again only within its age, and the copy keeps that age** (ADR 0376 §5): +// with more than a minute left the copy goes on EVENTS under the consumer's again subject with a time to live of +// what is left; with less, it is refused as aged out. +func TestARestrictedDeadLetterGoesBackOnlyWithinItsAge(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}} + for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer", + Hears: []broker.Heard{heard}, PasswordHash: "x"}}) { + if err := js.EnsureConsumer(c.Consumer); err != nil { + t.Fatal(err) + } + } + seq, published := publishMail(t, js, subject) + at := published.Add(7*24*time.Hour - 10*time.Minute) + withClock(t, func() time.Time { return at }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + at = published.Add(7*24*time.Hour - 30*time.Second) + if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "aged out") { + t.Fatalf("delivered again with under a minute left: %v", err) + } + at = published.Add(7*24*time.Hour - 5*time.Minute) + _, to, err := DeliverAgain(js.Context(), d.ID) + if err != nil { + t.Fatal(err) + } + if to != broker.AgainSubject("ace_mailer", subject) { + t.Fatalf("delivered again on %s", to) + } + raw, err := js.Context().GetLastMsg(broker.EventsStream, to) + if err != nil { + t.Fatal(err) + } + ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr)) + if err != nil || ttl > 5*time.Minute || ttl < 4*time.Minute { + t.Fatalf("the copy carries a time to live of %q (%v), want what is left", raw.Header.Get(nats.MsgTTLHdr), err) + } + if string(raw.Data) != `{"from":"someone@example.org"}` || raw.Header.Get("x-account") != "mesh" { + t.Fatalf("the copy is %q %v", raw.Data, raw.Header) + } +} + +// What of a consumer's dead letters ages out, and when, is read without their bodies, and only for a restricted +// stream's (ADR 0376 §6). +func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(time.Hour) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+ + `"deliveries":5}`)); err != nil { + t.Fatal(err) + } + held, err := HeldDeadLetters(js.Context()) + if err != nil { + t.Fatal(err) + } + aging, err := AgingDeadLetters(js.Context(), held) + if err != nil { + t.Fatal(err) + } + if len(aging) != 1 || len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID || + !aging["MAIL.ace_mailer"][0].AgesOut.Equal(d.AgesOut) { + t.Fatalf("aging %+v, kept %+v", aging, d) + } +} + +// **A look never waits out NextMsg's bound for a dead letter that is not there** (novox/hq issue 728 review round +// 1): held can count more than DEAD_LETTERS still holds for a consumer, one aged out, delivered again or dropped +// since it was counted. AgingDeadLetters reads the ordered consumer's pending count once and looks for at most +// that many, so it returns in well under the 2-second NextMsg bound rather than waiting it out for messages that +// are gone. +func TestTheAgingOfRestrictedDeadLettersDoesNotWaitForWhatHeldOvercounts(t *testing.T) { + js := aBus(t) + subject := aRestrictedStream(t, js) + seq, published := publishMail(t, js, subject) + withClock(t, func() time.Time { return published.Add(time.Hour) }) + d, err := KeepDeadLetter(js.Context(), notice(seq)) + if err != nil { + t.Fatal(err) + } + // held says 3 for this consumer; DEAD_LETTERS holds only the one just kept. + held := map[string]int{"MAIL.ace_mailer": 3} + + started := time.Now() + aging, err := AgingDeadLetters(js.Context(), held) + elapsed := time.Since(started) + if err != nil { + t.Fatal(err) + } + if elapsed >= time.Second { + t.Fatalf("took %s, well over the 2-second bound it should never wait out", elapsed) + } + if len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID { + t.Fatalf("aging %+v, kept %+v", aging, d) + } +} diff --git a/testdata/store-tests b/testdata/store-tests index 8d09e44a..8f060418 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -74,3 +74,8 @@ ./cmd/mesh-controller TestOneProviderDownAndThreeConsumersFailingAreOneCondition ./cmd/mesh-controller TestAConsumerFailingBesideAHealthyProviderIsItsOwn ./internal/inventory TestWhatRunsIsTheRegisteredBuildNotANewerOnePutBack +./cmd/mesh-controller TestARecordedModulesBundleOnlyMoveRollsOutInItsWalk +./cmd/mesh-controller TestARecordedModulesMoveThatRecreatesAContainerWaits +./cmd/mesh-controller TestARecordedModulesMoveThatTouchesDataOrAProviderWaits +./cmd/mesh-controller TestAPersonsRecordAndTheBusHoldEveryMove +./cmd/mesh-controller TestARecordedMoveThatCannotBeJudgedWaits