diff --git a/cmd/mesh-host/backup_event_test.go b/cmd/mesh-host/backup_event_test.go index 0a773cc..5871dc1 100644 --- a/cmd/mesh-host/backup_event_test.go +++ b/cmd/mesh-host/backup_event_test.go @@ -3,9 +3,13 @@ package main import ( "context" "encoding/json" + "errors" + "fmt" "strings" "testing" "time" + + "github.com/novox/mesh-host/internal/link" ) func backupEvent(node, module string, ok bool, at time.Time, why string) []byte { @@ -39,4 +43,34 @@ func TestABackupEndsWhenItsEventSaysSo(t *testing.T) { if done, words := run(50 * time.Millisecond); done || !strings.Contains(words, "did not say it was done") { t.Errorf("silence within the bound: done=%v words=%q", done, words) } + + // The ask that starts it (sweep 2026-10-30): one that never reached the holder (nothing on this machine answers it + // while the tool runner restarts) is asked again within its bound, and past it fails saying so; one that may have + // reached it is never asked twice. + notAsked := fmt.Errorf("node-backup.now could not be asked: %w: nothing on this machine answers "+ + "mesh.seat.node-backup.tool.now.ace", link.ErrNotAsked) + for _, row := range []struct { + name string + answers []error + within time.Duration + asked int + words string + }{ + {"heard after the tool runner is back", []error{notAsked, notAsked, nil}, time.Second, 3, ""}, + // Asked again until the bound, however slow the machine: at least twice, and fewer than it would answer. + {"never heard within the bound", []error{notAsked, notAsked, notAsked, notAsked, notAsked, nil}, 25 * time.Millisecond, + 0, "times over"}, + {"a timeout may have started it", []error{errors.New("no answer from x: context deadline exceeded"), nil}, + time.Second, 1, "deadline exceeded"}, + {"a refusal in words", []error{errors.New("restic is not on this machine"), nil}, time.Second, 1, "not on this"}, + } { + asked := 0 + _, err := askUntilHeard(context.Background(), func(context.Context) (json.RawMessage, error) { + asked++ + return json.RawMessage(`{}`), row.answers[asked-1] + }, row.within, func(int) time.Duration { return 10 * time.Millisecond }) + if row.asked == 0 && (asked < 2 || asked >= len(row.answers)) || row.asked != 0 && asked != row.asked || (row.words == "") != (err == nil) || err != nil && !strings.Contains(err.Error(), row.words) { + t.Errorf("%s: asked %d times, %v", row.name, asked, err) + } + } } diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 505bc93..c689a92 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1840,9 +1840,8 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio if j := judging; j != nil { // A container that waits for a library whose disk or mount is not there was not created: the library says // why, and liveness judges it neither alive nor dead (novox/hq ADR 0328 §9). - j.Set(liveness.LongRunning(declared, withWaiting(held, apply.WaitingOnLibrary(declared)))) + judgeWhatIsDeclared(declared, held) st, _ := j.Look(ctx) - netJudge.declare(declaredResolvConf(declared)) var netSt *network.Statement if n := netJudge.get(); n != nil { ns := n.Last() @@ -1850,7 +1849,6 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio } var unitSt *units.Statement if u := unitJudge; u != nil { - u.Set(units.OwnedBy(declared, held)) us := u.Last() unitSt = &us } @@ -1858,11 +1856,9 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio // said with it, relogin needed included, in the report that says the apply. var accountSt *accounts.Statement if a := accountJudge; a != nil { - a.Set(accounts.Of(declared, held)) as, _ := a.Look(ctx) accountSt = &as } - accesses.set(declared, held) // A take act stopped at its bound goes on, beside (ADR 0328 §7). keeping.poke(ctx) accessSt, _ := accesses.look(time.Now()) @@ -1957,6 +1953,46 @@ func factsOf(ctx context.Context, opts options, declared *declaration.Declaratio return report } +// judgeWhatIsDeclared hands every judge of the health loop what is declared here now: the long-running resources, the +// resolver file, the units, the accounts and the libraries. A resource, unit, account or library no longer declared is +// forgotten by its judge, so a module whose composition drops one is no longer watched for it. +func judgeWhatIsDeclared(declared *declaration.Declaration, held map[string]bool) { + if j := judging; j != nil { + // A container that waits for a library whose disk or mount is not there was not created: the library says + // why, and liveness judges it neither alive nor dead (novox/hq ADR 0328 §9). + j.Set(liveness.LongRunning(declared, withWaiting(held, apply.WaitingOnLibrary(declared)))) + } + netJudge.declare(declaredResolvConf(declared)) + if u := unitJudge; u != nil { + u.Set(units.OwnedBy(declared, held)) + } + if a := accountJudge; a != nil { + a.Set(accounts.Of(declared, held)) + } + accesses.set(declared, held) +} + +// judgeWhatRunsNow hands the judges what runs now, as the node agent holds it once a module settled (applied or +// removed) or was taken as running: the facts said inside an apply read the module's composition from before it, so a +// resource its new composition dropped stayed watched until the next facts (g14 and shanks, 2026-10-30 16:14: restic's +// process restic.measure, removed, then said down every look). +func judgeWhatRunsNow(statePath string, signer ed25519.PublicKey) { + if judging == nil { + return + } + declared, err := declaredNow(statePath, signer) + if err != nil { + return + } + held := map[string]bool{} + if known, err := store.Load(statePath); err == nil { + for _, h := range known.Held { + held[h.ID] = true + } + } + judgeWhatIsDeclared(declared, held) +} + // sayFacts says this node's facts on the link open now (novox/hq ADR 0539 §9): what the node is made of, how what runs // here is, and its filters and ports. What runs is the compositions the node agent holds running (ag), or, before it // has run, the declaration the node kept before the switch. It applies nothing; the haves are the agent's. diff --git a/cmd/mesh-host/movesecret.go b/cmd/mesh-host/movesecret.go index da6c23c..0748460 100644 --- a/cmd/mesh-host/movesecret.go +++ b/cmd/mesh-host/movesecret.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "os" + "sync" "time" "golang.org/x/crypto/nacl/box" @@ -40,6 +41,11 @@ func moveSecretFrom(statePath string, queue *link.Session) link.MoveSecretFunc { if err != nil { fmt.Fprintf(os.Stderr, "mesh-host: a value was not moved into the vault: %v\n", err) } + if moved.Superseded != "" { + if _, said := supersededSaid.LoadOrStore(moved.Secret, true); !said { + fmt.Fprintf(os.Stderr, "mesh-host: %s\n", moved.Superseded) + } + } return moved, err } } @@ -104,8 +110,30 @@ func moveIntoTheVault(ask vaultAsk, key identity.SealingKey, m link.MoveSecret) Generation int `json:"generation"` Unchanged bool `json:"unchanged"` ReadBack string `json:"read_back"` + // Refused is the vault's refusal as fields (mesh-vault's MoveRefused, mesh-catalog f3b13de2): nothing was taken. + Refused *struct { + Reason string `json:"reason"` + Held *heldValue `json:"held"` + } `json:"refused"` + Why string `json:"why"` } - if err := json.Unmarshal(raw, &kept); err != nil || kept.Name == "" || kept.ReadBack == "" { + if err := json.Unmarshal(raw, &kept); err != nil { + return link.MovedSecret{}, fmt.Errorf("the vault's answer for %s does not read: %v", opened.Secret, err) + } + if r := kept.Refused; r != nil { + // The vault holds another value for the name, and the move never replaces one: the vault's value is the live + // one, and this machine's older value is superseded, never asked again (the sweep of 2026-10-30: it failed + // "answered with no read-back" on every ask). + if r.Reason == vaultHoldsAnother && r.Held != nil && r.Held.Name != "" { + return link.MovedSecret{Secret: r.Held.Name, Generation: r.Held.Generation, Superseded: r.Held.words()}, nil + } + why := kept.Why + if why == "" { + why = r.Reason + } + return link.MovedSecret{}, fmt.Errorf("the vault refused %s: %s", opened.Secret, why) + } + if kept.Name == "" || kept.ReadBack == "" { return link.MovedSecret{}, fmt.Errorf("the vault answered %s with no read-back", opened.Secret) } blob, err := base64.StdEncoding.DecodeString(kept.ReadBack) @@ -124,6 +152,31 @@ func moveIntoTheVault(ask vaultAsk, key identity.SealingKey, m link.MoveSecret) return link.MovedSecret{Secret: kept.Name, Generation: kept.Generation, Unchanged: kept.Unchanged}, nil } +// vaultHoldsAnother is the vault's reason when it holds another value for the name (mesh-vault RefusedHeldAnother). +const vaultHoldsAnother = "holds-another-value" + +// heldValue is the record of the value the vault holds instead, never the value (mesh-vault HeldAnother). +type heldValue struct { + Name string `json:"name"` + Generation int `json:"generation"` + ChangedBy string `json:"changed_by"` + ChangedAt time.Time `json:"changed_at"` + Origin string `json:"origin"` +} + +// words say the held value's record, for the journal and the controller. +func (h heldValue) words() string { + w := fmt.Sprintf("the vault holds another value for %s, generation %d, changed by %s at %s", h.Name, h.Generation, + h.ChangedBy, h.ChangedAt.UTC().Format(time.RFC3339)) + if h.Origin != "" { + w += ", origin " + h.Origin + } + return w + ": that value is the live one, and this machine's older value is superseded and not moved" +} + +// supersededSaid holds the names a superseded move was said of, so the journal says each once. +var supersededSaid sync.Map + func wipe(b []byte) { for i := range b { b[i] = 0 diff --git a/cmd/mesh-host/movesecret_test.go b/cmd/mesh-host/movesecret_test.go index e362c77..2da4d47 100644 --- a/cmd/mesh-host/movesecret_test.go +++ b/cmd/mesh-host/movesecret_test.go @@ -17,13 +17,15 @@ import ( // standInVault answers move-in's two steps as the vault does: a key made for the call, the value opened with it, // and the value kept read back sealed to the key the second step sent. lie makes its read-back another value. type standInVault struct { - t *testing.T - pub *[32]byte - priv *[32]byte - opened map[string]any - kept string - lie string - refuse string + t *testing.T + pub *[32]byte + priv *[32]byte + opened map[string]any + kept string + lie string + refuse string + // answer replaces the second step's answer: the vault's refusal as fields (mesh-catalog f3b13de2). + answer map[string]any carried []string } @@ -51,6 +53,9 @@ func (v *standInVault) ask(verb string, args any) (json.RawMessage, error) { v.t.Fatal("what the engine sealed does not open with the call's key") } v.kept = string(value) + if v.answer != nil { + return json.Marshal(v.answer) + } back := v.kept if v.lie != "" { back = v.lie @@ -119,6 +124,28 @@ func TestTheEngineSaysAMoveThatDidNotHold(t *testing.T) { !strings.Contains(err.Error(), "never replaces") { t.Fatalf("the vault's refusal: %v", err) } + // The vault's refusal as fields (mesh-catalog f3b13de2, no read-back): for another value held, the vault's is the + // live one and the move is done, superseded, at the held generation; any other reason fails in the vault's words. + held := map[string]any{"name": "own/anchor/postgres/superuser-password", "generation": 3, + "changed_by": "mesh-vault@novox", "changed_at": "2026-10-30T15:50:00Z", "origin": "made"} + for _, row := range []struct { + refused map[string]any + gen int + fails string + }{ + {map[string]any{"reason": "holds-another-value", "held": held}, 3, ""}, + {map[string]any{"reason": "locked"}, 0, "the vault refused"}, + } { + v := &standInVault{t: t, answer: map[string]any{"refused": row.refused, "why": "in words"}} + moved, err := moveIntoTheVault(v.ask, key, aMoveOf(t, key)) + switch { + case row.fails != "" && (err == nil || !strings.Contains(err.Error(), row.fails)): + t.Errorf("refused %v: %v", row.refused["reason"], err) + case row.fails == "" && (err != nil || moved.Generation != row.gen || moved.Secret != held["name"] || + !strings.Contains(moved.Superseded, "changed by mesh-vault@novox")): + t.Errorf("refused %v: moved %+v, %v", row.refused["reason"], moved, err) + } + } // A pair credential names both its ends. pair := aMoveOf(t, key) pair.Module, pair.Name = "", "" diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index a67f6a0..9e55e69 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -165,19 +165,54 @@ func backupNode(session *link.Session) func(ctx context.Context, module string) defer stop() // A module this machine backs nothing of up moves without a backup: never held for one that cannot run (issue // 1945). An answer that cannot be had or read asks for the backup as before. - if said, err := session.AskSeat(ctx, "node-backup", "backed-up", map[string]any{"module": module}, 30*time.Second); err == nil { + ask := func(verb string) func(context.Context) (json.RawMessage, error) { + return func(ctx context.Context) (json.RawMessage, error) { + return session.AskSeat(ctx, "node-backup", verb, map[string]any{"module": module}, 30*time.Second) + } + } + if said, err := askUntilHeard(ctx, ask("backed-up"), backupAskFor, link.Backoff); err == nil { if nothing, read := declaresNothingToBackUp(said, module); read && nothing { return agent.Backup{Done: true, Words: module + " declares nothing to back up on this machine"} } } asked := time.Now() - if _, err := session.AskSeat(ctx, "node-backup", "now", map[string]any{"module": module}, 30*time.Second); err != nil { + if _, err := askUntilHeard(ctx, ask("now"), backupAskFor, link.Backoff); err != nil { return agent.Backup{Words: err.Error()} } return awaitBackup(ctx, events, session.Membership.Node, module, asked, backupWithin) } } +// backupAskFor is how long the backup's ask is asked again while it is known never to have reached the holder of +// node-backup: nothing on this machine answers it (the tool runner restarting, its subscriptions not made yet) or no +// link is open. The sweep of 2026-10-30 met it on ace's redis, postgres and restic (14:03), step-ca (13:55) and +// mesh-vault on novox (15:22): each was held on one unanswered ask. Past the bound the backup failed, said with how +// often and how long it was asked: the module is held, its have says so, and the assigner asks the operator. +var backupAskFor = 3 * time.Minute + +// askUntilHeard asks, and asks again after wait(try) while the answer says the ask was never delivered (link.ErrNotAsked), +// for at most within: an ask that may have reached its holder (a timeout, a refusal in words) is never asked twice, so a +// backup is never started twice. +func askUntilHeard(ctx context.Context, ask func(context.Context) (json.RawMessage, error), within time.Duration, + wait func(int) time.Duration) (json.RawMessage, error) { + began := time.Now() + for try := 1; ; try++ { + said, err := ask(ctx) + if err == nil || !errors.Is(err, link.ErrNotAsked) { + return said, err + } + pause := wait(try) + if time.Since(began)+pause > within { + return nil, fmt.Errorf("asked %d times over %s and never heard: %w", try, time.Since(began).Round(time.Second), err) + } + select { + case <-ctx.Done(): + return nil, fmt.Errorf("the node agent was stopped while it asked again: %w", err) + case <-time.After(pause): + } + } +} + // awaitBackup waits for the event that says the backup of a module on this node ended, within the bound: done when it // says ok, held on its words when it does not, and held when it says nothing within the bound. An event of another node // or module, or from before the ask, is not the answer. @@ -520,11 +555,17 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus Backup: backupNode(session), // What the filling refused before this apply may be named now: the provisioner's client is told to fill // again (ace, 2026-10-25 21:17). - Settled: func(module string) { knockFilling(module+" settled", say) }, + Settled: func(module string) { + knockFilling(module+" settled", say) + judgeWhatRunsNow(opts.state, signer) + }, // And once the haves are read at the start, and whenever a module is taken as running without an apply, so // a refusal from the engine's start does not wait for some module to settle (issue 924). - RunsChanged: func(why string) { knockFilling(why, say) }, - Say: say, + RunsChanged: func(why string) { + knockFilling(why, say) + judgeWhatRunsNow(opts.state, signer) + }, + Say: say, // A failure of the bus not answering for a moment is tried again: 2 s, 5 s, 15 s, 60 s, then every 5 min. Backoff: link.Backoff, Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result { diff --git a/cmd/mesh-host/whatruns_test.go b/cmd/mesh-host/whatruns_test.go index 1d4601e..bdace6b 100644 --- a/cmd/mesh-host/whatruns_test.go +++ b/cmd/mesh-host/whatruns_test.go @@ -1,10 +1,12 @@ package main import ( + "context" "crypto/ed25519" "encoding/json" "errors" "path/filepath" + "sort" "strings" "testing" @@ -12,6 +14,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/liveness" "github.com/novox/mesh-host/internal/store" ) @@ -62,6 +65,46 @@ func TestDeclaredNowReadsWhatRunsWhenTheAgentIsRunning(t *testing.T) { t.Fatalf("declaredNow read %+v (%v), not what runs", d, err) } + // And the health loop's judges are handed what runs once a module settles (g14 and shanks, 2026-10-30 16:14): + // restic's composition dropped its process restic.measure, which stayed watched and said down on every look. + j, err := liveness.Open(filepath.Join(t.TempDir(), "liveness.json"), fakeRuntime{}) + if err != nil { + t.Fatal(err) + } + judging = j + defer func() { judging = nil }() + process := func(name string) json.RawMessage { + raw, _ := json.Marshal(map[string]any{"id": "restic." + name, "type": "process", "name": "restic-" + name, + "source": "https://store.invalid/restic", "digest": "sha256:" + strings.Repeat("a", 64), + "run": []string{"./restic"}, "module": "restic"}) + return raw + } + for _, row := range []struct { + runs []string + watch string + }{{[]string{"night", "measure"}, "restic.measure restic.night"}, {[]string{"night"}, "restic.night"}} { + var raws []json.RawMessage + for _, name := range row.runs { + raws = append(raws, process(name)) + } + runs, err := whatRuns([]delivery.Composed{{Node: "n1", Module: "restic", Resources: raws}}, false, + func() (*declaration.Declaration, error) { return nil, errors.New("none kept") }) + if err != nil { + t.Fatal(err) + } + setRunsHere(func(string, ed25519.PublicKey) (*declaration.Declaration, error) { return runs, nil }) + judgeWhatRunsNow(filepath.Join(t.TempDir(), "state.json"), nil) + st, _ := j.Look(context.Background()) + var watched []string + for _, r := range st.Resources { + watched = append(watched, r.ID) + } + sort.Strings(watched) + if got := strings.Join(watched, " "); got != row.watch { + t.Errorf("running %v, the judge watches %q, want %q", row.runs, got, row.watch) + } + } + // L3 s9 (ADR 0550 §2): a node whose placed modules all have a have reads the kept declaration no more, though the // file is there and readable and the controller's key is dropped; until then it reads it. state := filepath.Join(t.TempDir(), "state", "state.json") diff --git a/internal/link/engineseat.go b/internal/link/engineseat.go index 258e3f5..7261559 100644 --- a/internal/link/engineseat.go +++ b/internal/link/engineseat.go @@ -111,7 +111,8 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { return } svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0", - Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true}) + Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true, + Metadata: map[string]string{"seat": EngineSeat, "scope": "node", "node": l.node}}) if err != nil { return } @@ -120,7 +121,7 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { answer := micro.HandlerFunc(func(r micro.Request) { _ = r.Respond(AnswerEngineCall(handle, verb, l.node, r.Subject(), r.Data())) }) - meta := map[string]string{"seat": EngineSeat, "tool": verb, "node": l.node} + meta := EngineEndpointMetadata(v, l.node) _ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)), micro.WithEndpointMetadata(meta)) _ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)), @@ -129,6 +130,23 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) { l.engine = svc } +// EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's. +const EngineModule = "mesh-host" + +// EngineEndpointMetadata is what discovery reads of one verb's endpoint, in the shape every tool runner and the +// controller announce a seat's verb in (mesh-tools node-tools' announce package): its kind, the seat, its scope, the +// module whose code answers, the verb, the node, its description and its schema. Without the kind and the scope the +// console listed none of it, and `/node-engine.` answered that nothing in the mesh is called node-engine +// (the sweep of 2026-10-30). +func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string { + schema, err := json.Marshal(v.Input) + if err != nil || v.Input == nil { + schema = []byte("{}") + } + return map[string]string{"kind": "seat", "seat": EngineSeat, "scope": "node", "module": EngineModule, + "tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)} +} + // SayEngine says one of the seat's events about this node, on the events stream. func (l *natsLink) SayEngine(event string, body any) error { raw, err := json.Marshal(body) diff --git a/internal/link/handover_test.go b/internal/link/handover_test.go index a19d009..54bceb6 100644 --- a/internal/link/handover_test.go +++ b/internal/link/handover_test.go @@ -342,6 +342,35 @@ func TestNatsAHandOverIsAskedOfTheNode(t *testing.T) { if _, err := conn.Request(EngineToolSubject("acts", "another"), []byte(`{}`), 300*time.Millisecond); err == nil { t.Fatal("another node's verb was answered here") } + // Announced in the shape the console reads a seat's verb in (sweep 2026-10-30): without kind and scope it listed + // none of it, and /node-engine.root-free answered that nothing in the mesh is called node-engine. + info, err := conn.Request("$SRV.INFO."+EngineSeat, nil, 5*time.Second) + if err != nil { + t.Fatal(err) + } + var said struct { + Endpoints []struct { + Subject string `json:"subject"` + Metadata map[string]string `json:"metadata"` + } `json:"endpoints"` + } + if err := json.Unmarshal(info.Data, &said); err != nil { + t.Fatal(err) + } + announced := 0 + for _, e := range said.Endpoints { + md := e.Metadata + if md["kind"] != "seat" || md["scope"] != "node" || md["seat"] != EngineSeat || md["module"] != "mesh-host" || + md["node"] != m.Node || md["description"] == "" || !strings.HasPrefix(md["schema"], "{") { + t.Errorf("the endpoint on %s announces %v", e.Subject, md) + } + if e.Subject == EngineToolSubject("root-free", m.Node) { + announced++ + } + } + if announced != 1 { + t.Errorf("root-free is announced on its tool subject %d times", announced) + } for seat, want := range map[string]bool{EngineSeat: true, "node-login-shell": false} { verb := "acts" if !want { diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index 9bd0cde..afc81b4 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -5,6 +5,8 @@ import ( "crypto/ed25519" "errors" "fmt" + "os" + "regexp" "strings" "sync" "time" @@ -57,6 +59,9 @@ type natsLink struct { engine micro.Service warrants *nats.Subscription node string + // refused carries each subject the bus refused this link a subscription to (RefusedSubscription), which serve drops + // the link for after a pause, so every subscription is asked again on the next. + refused chan string // signer is the mesh's key, which a hand-over ask is verified with as a declaration is (issue 356). signer ed25519.PublicKey lost chan error @@ -108,9 +113,13 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e l := &natsLink{ conn: conn, js: js, node: m.Node, signer: m.Signer, - lost: make(chan error, 1), - asked: make(chan struct{}, 1), + lost: make(chan error, 1), + asked: make(chan struct{}, 1), + refused: make(chan string, 64), } + // Set before anything subscribes, so the node seat's service (micro), which wraps the handler it finds, hands every + // refusal of its own subjects on to it too. + conn.SetErrorHandler(l.asyncError) l.hearAsks() @@ -259,6 +268,23 @@ func (l *natsLink) HearMoveSecrets(handle MoveSecretFunc) { } } +// asyncError is said by the client as before, and a subscription the bus refused is handed to serve (Refused). +func (l *natsLink) asyncError(_ *nats.Conn, _ *nats.Subscription, err error) { + if err == nil { + return + } + fmt.Fprintf(os.Stderr, "%v\n", err) + if subject, ok := RefusedSubscription(err); ok && l.refused != nil { + select { + case l.refused <- subject: + default: + } + } +} + +// Refused says each subject the bus refused this link a subscription to. +func (l *natsLink) Refused() <-chan string { return l.refused } + func (l *natsLink) AskedToReport() <-chan struct{} { return l.asked } func (l *natsLink) Lost() <-chan error { return l.lost } @@ -339,3 +365,19 @@ var ( _ EngineSeats = (*natsLink)(nil) _ OperatorAsks = (*natsLink)(nil) ) + +// refusedSubscription is how the server words a subscription its user list does not grant. +var refusedSubscription = regexp.MustCompile(`Permissions Violation for Subscription to "([^"]+)"`) + +// RefusedSubscription is the subject a permissions violation names when it refused a subscription; false for any other +// error, a refused publish among them (a publish is refused each time it is tried, and is said where it is tried). +func RefusedSubscription(err error) (string, bool) { + if err == nil || !errors.Is(err, nats.ErrPermissionViolation) && !strings.Contains(err.Error(), "permissions violation") { + return "", false + } + m := refusedSubscription.FindStringSubmatch(err.Error()) + if m == nil { + return "", false + } + return m[1], true +} diff --git a/internal/link/movesecret.go b/internal/link/movesecret.go index 95463bb..4342796 100644 --- a/internal/link/movesecret.go +++ b/internal/link/movesecret.go @@ -54,6 +54,9 @@ type MoveSecretAnswer struct { Secret string `json:"secret,omitempty"` Generation int `json:"generation,omitempty"` Unchanged bool `json:"unchanged,omitempty"` + // Superseded is set when the vault holds another value for the name, which is the live one: the row is done and + // is not asked again; Generation is the held value's, and Said its record in words. + Superseded bool `json:"superseded,omitempty"` } // MoveSecret is one move asked and believed, as the handler is given it: the row, this machine, and who asked. @@ -70,6 +73,9 @@ type MovedSecret struct { Secret string Generation int Unchanged bool + // Superseded says, in words, that the vault holds another value for the name, which is the live one: nothing was + // moved, and the move is done (the vault's record of the held value, its generation in Generation). + Superseded string } // MoveSecretFunc moves one value into the vault: what it kept, or why not. @@ -133,7 +139,11 @@ func AnswerMoveSecret(check HandOverCheck, handle MoveSecretFunc, body []byte) M if moved.Unchanged { said = fmt.Sprintf("the vault held %s already, the same value (generation %d)", moved.Secret, moved.Generation) } - return MoveSecretAnswer{Said: said, Secret: moved.Secret, Generation: moved.Generation, Unchanged: moved.Unchanged} + if moved.Superseded != "" { + said = moved.Superseded + } + return MoveSecretAnswer{Said: said, Secret: moved.Secret, Generation: moved.Generation, Unchanged: moved.Unchanged, + Superseded: moved.Superseded != ""} } // MoveSecretReply is what a move request is answered with, and whether to answer, as HandOverReply. diff --git a/internal/link/movesecret_test.go b/internal/link/movesecret_test.go index 41ae611..fc01756 100644 --- a/internal/link/movesecret_test.go +++ b/internal/link/movesecret_test.go @@ -28,8 +28,9 @@ func TestTheMovesAskKeepsItsFieldNamesAndContext(t *testing.T) { if MoveSecretContext != "novox-mesh move-secret v1\n" { t.Fatalf("the signing context is %q", MoveSecretContext) } - answer, _ := json.Marshal(MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true}) - if got := keysOf(t, answer); got != "generation refused said secret unchanged" { + answer, _ := json.Marshal(MoveSecretAnswer{Said: "s", Refused: "r", Secret: "own/a/b/c", Generation: 1, Unchanged: true, + Superseded: true}) + if got := keysOf(t, answer); got != "generation refused said secret superseded unchanged" { t.Fatalf("the answer's fields are %q", got) } } diff --git a/internal/link/run.go b/internal/link/run.go index eac39ef..70c441c 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -150,6 +150,15 @@ func holdWith(ctx context.Context, run attempt, say Announce, roused Roused) err } } +// Refusals is a link that says each subject the bus refused it a subscription to. Optional. +type Refusals interface { + Refused() <-chan string +} + +// refusalsSettle is how long a link runs with no subscription refused before it counts as holding them: refusals arrive +// within moments of subscribing. +const refusalsSettle = 30 * time.Second + // Run holds the link open once, hearing what the controller asks and saying the node's facts. // // Outbound only, and nothing listens on this machine. Returns when the link ends, for any reason; @@ -226,6 +235,21 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann s.HearMoveSecrets(queue.MoveSecret) } + // And every subscription the bus refused, because the user list it loaded does not grant it (yet): nats asks a + // refused subscription again only on a new connection, and a reload of the user list does not make one, so the node + // seat, the hand-over and every other subject heard here stayed deaf until the engine restarted (the sweep of + // 2026-10-30: every engine was refused its node seat at 12:10 and answered nobody all afternoon). The link is kept + // for a pause, said, and then dropped, so Hold opens it again and every subscription is asked anew. + var refused <-chan string + if r, ok := link.(Refusals); ok { + refused = r.Refused() + } + var again <-chan time.Time + var first string + refusedCount := 0 + held := time.NewTimer(refusalsSettle) + defer held.Stop() + for { if ctx.Err() != nil { return nil @@ -237,6 +261,23 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann return reason case <-asks: queue.FactsAsked(ctx) + case subject := <-refused: + refusedCount++ + if again == nil { + first = subject + pause := queue.refusedPause() + say(fmt.Sprintf("the bus refused this node a subscription to %s: the user list it loaded does not grant "+ + "it, so nothing here answers there; the link is opened again in %s so every subscription is asked "+ + "again", subject, pause)) + again = time.After(pause) + } + case <-held.C: + if refusedCount == 0 { + queue.refusalsHeld() + } + case <-again: + return fmt.Errorf("the bus refused %d of this node's subscriptions, %s first; opened again so each is "+ + "asked again", refusedCount, first) } } } diff --git a/internal/link/session.go b/internal/link/session.go index 5b2fb89..5f63d1f 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -60,6 +60,12 @@ type Session struct { Keys func() RecordKeys Say Announce Timeout time.Duration + // RefusedPause is how long a link keeps running after the bus refused it a subscription before it is dropped and + // opened again (zero is DefaultRefusedPause); doubled for each link in a row that is refused again, to RefusedPauseMost. + RefusedPause time.Duration + + // refusedInARow is how many links in a row the bus refused a subscription to. + refusedInARow int once sync.Once @@ -91,6 +97,36 @@ type Session struct { healthWritten *NodeHealth } +// DefaultRefusedPause and RefusedPauseMost bound how soon a link refused a subscription is opened again: the grants +// part composes the user list again within a minute of what it is composed from changing, so the first pause waits for +// that; one that is still refused waits longer each time, never more than half an hour, and always says so. +const ( + DefaultRefusedPause = time.Minute + RefusedPauseMost = 30 * time.Minute +) + +// refusedPause is the pause before a link the bus refused a subscription to is opened again, counting it. +func (q *Session) refusedPause() time.Duration { + q.mu.Lock() + defer q.mu.Unlock() + pause := q.RefusedPause + if pause <= 0 { + pause = DefaultRefusedPause + } + for i := 0; i < q.refusedInARow && pause < RefusedPauseMost; i++ { + pause *= 2 + } + q.refusedInARow++ + return min(pause, RefusedPauseMost) +} + +// refusalsHeld says a link has held its subscriptions: the next refusal starts from the first pause again. +func (q *Session) refusalsHeld() { + q.mu.Lock() + q.refusedInARow = 0 + q.mu.Unlock() +} + func (q *Session) init() { q.once.Do(func() { if q.Say == nil { diff --git a/internal/link/session_test.go b/internal/link/session_test.go index 909ed82..331efb0 100644 --- a/internal/link/session_test.go +++ b/internal/link/session_test.go @@ -5,7 +5,9 @@ import ( "crypto/ed25519" "encoding/json" "errors" + "fmt" "os" + "strings" "sync" "testing" "time" @@ -162,3 +164,78 @@ func eventuallyHeld(t *testing.T, what string, ok func() bool) { time.Sleep(5 * time.Millisecond) } } + +// refusingLink is a quiet link the bus refused subscriptions to. +type refusingLink struct { + *quietLink + refused chan string +} + +func (l *refusingLink) Refused() <-chan string { return l.refused } + +// The sweep of 2026-10-30: every engine was refused its node seat's subscriptions at 12:10, by a user list composed +// without them, and nats asks a refused subscription again only on a new connection; the seat answered nobody all +// afternoon. A refused subscription is read from the server's words, said, and the link opened again after a pause +// that grows while it is refused again; a refused publish is not one. +func TestASubscriptionTheBusRefusedIsAskedAgainOnANewLink(t *testing.T) { + // The server's words, as the client hands them on (journal of nox-mesh-host on novox, 2026-10-30 12:10:44). + for _, row := range []struct { + err error + want string + }{ + {fmt.Errorf("%w: %s", nats.ErrPermissionViolation, + `Permissions Violation for Subscription to "mesh.seat.node-engine.tool.root-free.novox"`), + "mesh.seat.node-engine.tool.root-free.novox"}, + {errors.Join(fmt.Errorf("%w: %s", nats.ErrPermissionViolation, + `Permissions Violation for Subscription to "$SRV.PING.node-engine"`), errors.New("stopping service: x")), + "$SRV.PING.node-engine"}, + {fmt.Errorf("%w: %s", nats.ErrPermissionViolation, + `Permissions Violation for Publish to "$JS.API.STREAM.INFO.KV_mesh-host_health"`), ""}, + {errors.New("nats: timeout"), ""}, + } { + got, ok := RefusedSubscription(row.err) + if got != row.want || ok != (row.want != "") { + t.Errorf("%v reads as %q, %v", row.err, got, ok) + } + l := &natsLink{refused: make(chan string, 1)} + l.asyncError(nil, nil, row.err) + if heard := len(l.refused); heard != map[bool]int{true: 1}[row.want != ""] { + t.Errorf("%v handed %d refusals on", row.err, heard) + } + } + + m, _ := aMember(t) + q := &Session{Membership: m, Timeout: time.Second, RefusedPause: 50 * time.Millisecond} + var said []string + var mu sync.Mutex + say := func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() } + for _, want := range []time.Duration{50 * time.Millisecond, 100 * time.Millisecond} { + l := &refusingLink{quietLink: newQuietLink(), refused: make(chan string, 2)} + l.refused <- "mesh.seat.node-engine.tool.root-free.n1" + l.refused <- "$SRV.PING" + began := time.Now() + done := make(chan error, 1) + go func() { done <- serve(context.Background(), l, m, q, say, time.Second) }() + select { + case err := <-done: + if err == nil || !strings.Contains(err.Error(), "refused 2 of this node's subscriptions, "+ + "mesh.seat.node-engine.tool.root-free.n1 first") { + t.Fatalf("the link ended with %v", err) + } + if took := time.Since(began); took < want { + t.Errorf("the link was dropped after %s, before its pause of %s", took, want) + } + case <-time.After(5 * time.Second): + t.Fatal("a link the bus refused a subscription was kept, and the subscription never asked again") + } + } + mu.Lock() + defer mu.Unlock() + if n := strings.Count(strings.Join(said, "\n"), "the bus refused this node a subscription to "+ + "mesh.seat.node-engine.tool.root-free.n1"); n != 2 { + t.Errorf("the refusal was said %d times in two links: %q", n, said) + } + if q.refusalsHeld(); q.refusedPause() != 50*time.Millisecond { + t.Error("a link that held its subscriptions did not start from the first pause again") + } +}