Merge pull request 'The node-engine asks again a subscription the bus refused, so its seat answers; backups and health follow what runs (issue 774)' (#179) from fix/sweep-2026-10-30 into main

Reviewed-on: #179
This commit was merged in pull request #179.
This commit is contained in:
2026-10-30 18:13:06 +00:00
14 changed files with 513 additions and 25 deletions
+34
View File
@@ -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)
}
}
}
+41 -5
View File
@@ -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.
+54 -1
View File
@@ -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
+34 -7
View File
@@ -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 = "", ""
+46 -5
View File
@@ -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 {
+43
View File
@@ -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")
+20 -2
View File
@@ -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>/node-engine.<verb>` 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)
+29
View File
@@ -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>/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 {
+44 -2
View File
@@ -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
}
+11 -1
View File
@@ -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.
+3 -2
View File
@@ -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)
}
}
+41
View File
@@ -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)
}
}
}
+36
View File
@@ -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 {
+77
View File
@@ -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")
}
}