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") + } +}