From 7ecf7ba3569bfd81dc6df5a4053f2575867e3a24 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:06:52 +0100 Subject: [PATCH 1/4] Ask a refused subscription again on a new link and announce the node seat as discovery reads it, so node-engine.root-free is answered again (issue 774) Every engine's node-seat subscriptions were refused at 12:10 by a user list composed without them. nats asks a refused subscription again only on a new connection, and a user-list reload makes none, so the seat answered nobody all afternoon and the messenger held every Telegram ask. The endpoints also lacked the kind and scope discovery reads, so the console said nothing in the mesh is called node-engine. Quick fix: restart the engines. Thorough fix (chosen, the second time a stale user list left an engine deaf): a refused subscription of any kind is said and the link is opened again after a pause (1 min, doubling to 30 min), so every subscription the engine makes is asked again; the seat's endpoints carry kind, seat, scope, module, description and schema like every tool runner's. --- internal/link/engineseat.go | 22 +++++++++- internal/link/handover_test.go | 29 +++++++++++++ internal/link/hearing_nats.go | 46 +++++++++++++++++++- internal/link/run.go | 41 ++++++++++++++++++ internal/link/session.go | 36 ++++++++++++++++ internal/link/session_test.go | 77 ++++++++++++++++++++++++++++++++++ 6 files changed, 247 insertions(+), 4 deletions(-) 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/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") + } +} From 8bff438e5eda3ceda9753626f9134878203b2f49 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:09:16 +0100 Subject: [PATCH 2/4] Ask a data module's backup again while nothing on the machine answers it, within three minutes, so a tool-runner restart no longer holds the module silently (issue 774) On 2026-10-30 ace's redis, postgres and restic (14:03), step-ca (13:55) and mesh-vault on novox (15:22) were held on a single backup ask that met no responder while the tool runner restarted. Quick fix: ask node-backup.now twice. Thorough fix (chosen): every node-backup ask (backed-up and now) is asked again with the engine's backoff while it is known never to have been delivered (ErrNotAsked), bounded at three minutes; an ask that may have been delivered (a timeout, a refusal) is never asked twice, so no backup starts twice. Past the bound the backup fails saying how often and how long it was asked, the module is held and its have carries those words for the assigner's ask and condition. This was the engine's only one-shot seat ask. --- cmd/mesh-host/backup_event_test.go | 34 ++++++++++++++++++++++++++ cmd/mesh-host/nodeagent.go | 39 ++++++++++++++++++++++++++++-- 2 files changed, 71 insertions(+), 2 deletions(-) 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/nodeagent.go b/cmd/mesh-host/nodeagent.go index a67f6a0..7d3ba5f 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. From 744bfec4a018936d0c00e6ad25d85a7ac2550a9e Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:12:56 +0100 Subject: [PATCH 3/4] Hand the health loop's judges what runs once a module settles, so a resource its new composition drops is no longer watched (issue 774) restic's composition dropped its process restic.measure on g14 and shanks at 16:14; the engine removed it and said it down on every look after. The judges were set only when facts were said, and the facts said inside an apply read the module's composition from before it (the agent records the new one only after the apply returns), so the removed resource was put back. Judge.Set itself prunes what is not declared. Quick fix: set the judges after the restic apply. Thorough fix (chosen): one function hands every judge (liveness, resolver, units, accounts, libraries) what is declared; it runs from the facts as before and now also whenever a module settles (applied or removed) or is taken as running, so every removed health entry of every module is pruned at recompose. --- cmd/mesh-host/main.go | 46 ++++++++++++++++++++++++++++++---- cmd/mesh-host/nodeagent.go | 12 ++++++--- cmd/mesh-host/whatruns_test.go | 43 +++++++++++++++++++++++++++++++ 3 files changed, 93 insertions(+), 8 deletions(-) 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/nodeagent.go b/cmd/mesh-host/nodeagent.go index 7d3ba5f..9e55e69 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -555,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") From b9c24a7b42cb2db5fe9fe32e8176b303c1cf2239 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:36:38 +0100 Subject: [PATCH 4/4] Read the vault's structured move-in refusal and take a value it holds another of as superseded, so the move is not asked again for ever (issue 774) mesh-vault (mesh-catalog f3b13de2) answers a move it refuses with `refused: {reason, held}` and `why`, and no read-back. The engine read only the read-back, so every such row failed "answered with no read-back" and the controller asked it again on every pass. Quick fix: treat any answer without a read-back as done. Thorough fix (chosen): read `refused`. For holds-another-value the vault's value is the live one: the move answers done, superseded, at the held generation with the held record in its words, so the controller records the row and stops asking; the journal says it once per secret. Any other reason fails in the vault's own words. --- cmd/mesh-host/movesecret.go | 55 +++++++++++++++++++++++++++++++- cmd/mesh-host/movesecret_test.go | 41 ++++++++++++++++++++---- internal/link/movesecret.go | 12 ++++++- internal/link/movesecret_test.go | 5 +-- 4 files changed, 102 insertions(+), 11 deletions(-) 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/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) } }