diff --git a/cmd/mesh-host/logins.go b/cmd/mesh-host/logins.go index 574a3fd..f7e923c 100644 --- a/cmd/mesh-host/logins.go +++ b/cmd/mesh-host/logins.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "fmt" "os" "path/filepath" "time" @@ -84,3 +85,69 @@ func followLogins(ctx context.Context, session *link.Session, node string, hear cancel() } } + +// loginSecretModule is the module the vault names this engine's login values under: own//mesh-host/ (ADR 0583 §3, +// insight). +const loginSecretModule = "mesh-host" + +// fetchLogin asks the vault, through its existing hand-out of a machine's own secrets, for the value of login, sealed to this +// node's key, and opens it with the sealing key; no new verb is asked for it. +func fetchLogin(queue *link.Session, statePath string) func(context.Context, string) (string, error) { + unseal := sealOpener(statePath) + return func(ctx context.Context, login string) (string, error) { + raw, err := queue.AskVault(ctx, "own", map[string]any{}) + if err != nil { + return "", err + } + var a struct { + Own []struct { + Module string `json:"module"` + Name string `json:"name"` + Sealed json.RawMessage `json:"sealed"` + } `json:"own"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return "", fmt.Errorf("the vault's own answer is not one: %w", err) + } + for _, o := range a.Own { + if o.Module != loginSecretModule || o.Name != login { + continue + } + sealed := string(o.Sealed) + var text string + if json.Unmarshal(o.Sealed, &text) == nil { + sealed = text + } + value, err := unseal(sealed) + if err != nil { + return "", fmt.Errorf("the value of %s cannot be opened with this node's sealing key: %w", login, err) + } + return string(value), nil + } + return "", fmt.Errorf("the vault hands this node no value for %s (own/%s/%s/%s)", login, queueNode(queue), loginSecretModule, login) + } +} + +func queueNode(q *link.Session) string { return q.Membership.Node } + +// newRotator is the engine's side of a rotation over this session (ADR 0583 §3-§6): the value fetched through the vault's own +// hand-out, the test-dial on a second connection, the confirmation over the new link, and the finding said in the journal. +func newRotator(queue *link.Session, statePath string, timeout time.Duration, say link.Announce) *link.Rotator { + r := &link.Rotator{Logins: queue.Membership.Logins, Say: say, + Fetch: fetchLogin(queue, statePath), + TestDial: func(ctx context.Context, login, password string) error { + return link.TestDial(ctx, queue.Membership, login, password, timeout) + }, + Confirm: func(ctx context.Context) error { + login, _ := queue.Membership.Logins.Current() + if err := queue.WhenLinkedAs(ctx, login); err != nil { + return err + } + _, err := queue.AskVault(ctx, "node-account-confirm", map[string]any{}) + return err + }, + // The finding the watchman raises (ADR 0583 §6): said in the journal, the route an engine says a finding by. + Finding: func(name, text string) { say("finding " + name + ": " + text) }} + r.Watch() + return r +} diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 8b656eb..592b1ab 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1194,10 +1194,12 @@ func runLink(ctx context.Context, opts options) error { trust.onChange(func() { factsOfNow(aside, current.get()) }) go followTrust(aside, queue, trust, say) // And the login it dials as, which the vault's node-logins names for this node (novox/hq ADR 0583 §1). + rotator := newRotator(queue, opts.state, opts.timeout, say) go followLogins(aside, queue, mine.Node, func(nl delivery.NodeLogin) { if line := membership.Logins.Heard(nl); line != "" { say(line) } + rotator.Heard(aside, nl) }, say) go nodeAgent(aside, opts, mine.Membership.Signer, trust, queue, sched, say, afterApply, current, factsOfNow) go busyPasses(aside, func() { diff --git a/internal/identity/identity_test.go b/internal/identity/identity_test.go index 9dd7293..d775e46 100644 --- a/internal/identity/identity_test.go +++ b/internal/identity/identity_test.go @@ -350,7 +350,8 @@ func TestASealingKeyOpensOnlyWhatWasSealedToIt(t *testing.T) { } } -func TestSealingTheSameValueTwiceLooksDifferent(t *testing.T) { +// Sealing is randomised and a node with no sealing key is told what to do (two tests merged). +func TestSealingIsRandomisedAndANodeWithNoSealingKeySaysWhatToDo(t *testing.T) { // Sealed boxes are randomised, so an observer cannot tell that two nodes were given the same // password, nor that a rotation changed nothing. Worth asserting because the alternative is // a subtle leak nobody would look for. @@ -360,9 +361,6 @@ func TestSealingTheSameValueTwiceLooksDifferent(t *testing.T) { if first == second { t.Fatal("sealing is deterministic, so equal secrets are visible as equal blobs") } -} - -func TestANodeWithNoSealingKeySaysWhatToDo(t *testing.T) { // Rather than making one. A key the mesh was never told about is a key nothing can be sealed // to, so a node that quietly created one would look fine and receive nothing for ever. _, err := LoadSealingKey(t.TempDir() + "/absent.key") diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index 0ba79dc..06cedf2 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -59,6 +59,8 @@ type natsLink struct { engine micro.Service warrants *nats.Subscription node string + // user is the bus user this link authenticated as (node. or node.--b, ADR 0583 §1). + user 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 @@ -101,6 +103,7 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e } conn, err := nats.Connect(natsURL(m.Broker), opts...) + m.Logins.dialed(user, err) if err != nil { if errors.Is(err, ErrWrongCertificate) { return nil, err @@ -114,7 +117,7 @@ func dialNats(ctx context.Context, m Membership, timeout time.Duration) (Link, e } l := &natsLink{ - conn: conn, js: js, node: m.Node, signer: m.Signer, + conn: conn, js: js, node: m.Node, user: user, signer: m.Signer, lost: make(chan error, 1), asked: make(chan struct{}, 1), refused: make(chan string, 64), @@ -383,3 +386,6 @@ func RefusedSubscription(err error) (string, bool) { } return m[1], true } + +// User is the bus user this link authenticated as. +func (l *natsLink) User() string { return l.user } diff --git a/internal/link/logins.go b/internal/link/logins.go index 8cb08be..1ee7fa7 100644 --- a/internal/link/logins.go +++ b/internal/link/logins.go @@ -1,8 +1,12 @@ package link import ( + "errors" + "strings" "sync" + "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -19,6 +23,8 @@ type Logins struct { values map[string]string save func(LoginsState) error changed chan struct{} + // OnDial is told how each dial as a login went: authenticated, or refused for its credentials (the Rotator's fallback). + OnDial func(user string, authenticated bool) } // LoginsState is what the engine keeps of its logins between runs. @@ -179,3 +185,26 @@ func (l *Logins) Heard(nl delivery.NodeLogin) (said string) { func rotatingToUs(nl delivery.NodeLogin, current string) bool { return nl.Rotating != "" && nl.Rotating == current } + +// dialed tells OnDial how a dial as user went; err is the dial's, nil for a connection made. +func (l *Logins) dialed(user string, err error) { + if l == nil || l.OnDial == nil { + return + } + switch { + case err == nil: + l.OnDial(user, true) + case refusedForCredentials(err): + l.OnDial(user, false) + } +} + +// refusedForCredentials says a failed dial was the bus refusing the user or its password, not the network failing. +func refusedForCredentials(err error) bool { + if errors.Is(err, nats.ErrAuthorization) || errors.Is(err, nats.ErrAuthExpired) { + return true + } + text := strings.ToLower(err.Error()) + return strings.Contains(text, "authorization violation") || strings.Contains(text, "authentication error") || + strings.Contains(text, "user authentication") +} diff --git a/internal/link/rotate.go b/internal/link/rotate.go new file mode 100644 index 0000000..d5c8f02 --- /dev/null +++ b/internal/link/rotate.go @@ -0,0 +1,178 @@ +package link + +import ( + "context" + "fmt" + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Rotator is the engine's side of a node's account rotation (novox/hq ADR 0583 §3-§6, issue 774). The vault begins a rotation +// and writes the node's key of node-logins: the login in use, the login rotating to and the stage. On that key's ready the +// engine fetches the new value, dials a second connection as the new login, and switches only if it authenticates; over the new +// connection it confirms. It keeps the old value until the vault says the rotation is done, switches back before the new value +// is dropped when the rotation is abandoned, and falls back to the old value after repeated authentication failures with the +// new one, saying so. Nothing here polls: every step answers a change of the key, or a failed dial. +type Rotator struct { + Logins *Logins + // Fetch opens the value of a login the vault hands this node as own//mesh-host/ (ADR 0583 §3, insight). + Fetch func(ctx context.Context, login string) (string, error) + // TestDial dials a second connection as login with password and answers whether it authenticated, closing it. + TestDial func(ctx context.Context, login, password string) error + // Confirm calls mesh-vault.node-account-confirm over the link as the login now in use, once it is open. + Confirm func(ctx context.Context) error + // Say is the engine's journal; Finding says a finding the watchman raises (node-account-fallback). + Say func(string) + Finding func(name, text string) + // FailBound is how many authentication failures with the new login, before rotation-done, it falls back after; 3 when 0. + FailBound int + + mu sync.Mutex + old string // the login switched away from, while the rotation is under way + failures int + // fellBack is the key's `since` a fallback was made for: the same rotation is not tried again. + fellBackSince string + switchedSince string +} + +func (r *Rotator) say(f string, a ...any) { + if r.Say != nil { + r.Say(fmt.Sprintf(f, a...)) + } +} + +// Heard takes this node's key of node-logins as read. +func (r *Rotator) Heard(ctx context.Context, nl delivery.NodeLogin) { + r.mu.Lock() + defer r.mu.Unlock() + cur, _ := r.Logins.Current() + switch { + case nl.Rotating == "" && nl.Login == cur: + // Done (the switch happened and the old login is retired), or no rotation at all: the other login's value goes (§4), + // only now that the vault says so. + r.old, r.failures = "", 0 + if other := delivery.OtherLogin(nl.Node, cur); other != "" { + if _, held := r.Logins.Value(other); held { + if err := r.Logins.Drop(other); err != nil { + r.say("the old login %s's value is not dropped: %v", other, err) + } else { + r.say("rotation-done: the value of %s is dropped, %s is the login in use", other, cur) + } + } + } + case nl.Rotating == "": + // Abandoned after the switch (before or after the confirmation), or the vault moved on: back to the login the vault keeps, + // before the value of the one it drops. + if _, held := r.Logins.Value(nl.Login); !held { + r.say("the vault keeps %s in use and this engine holds no value for it: it stays on %s", nl.Login, cur) + return + } + if err := r.Logins.Use(nl.Login); err != nil { + r.say("cannot switch back to %s: %v", nl.Login, err) + return + } + r.say("the rotation is abandoned: this engine dials as %s again", nl.Login) + r.old, r.failures = "", 0 + if err := r.Logins.Drop(cur); err != nil { + r.say("the abandoned login %s's value is not dropped: %v", cur, err) + } + case nl.Login == cur && nl.Stage != delivery.RotationBegun: + if r.fellBackSince != "" && r.fellBackSince == nl.Since.String() { + return // fallen back for this rotation: not tried again until the vault begins another + } + r.switchTo(ctx, nl) + case nl.Rotating == cur && nl.Stage != delivery.RotationConfirmed: + r.confirm(ctx, nl.Rotating) + } +} + +// switchTo fetches the new value, test-dials it and, only if it authenticates, switches and confirms (§3). +func (r *Rotator) switchTo(ctx context.Context, nl delivery.NodeLogin) { + next := nl.Rotating + password, held := r.Logins.Value(next) + if !held { + var err error + if password, err = r.Fetch(ctx, next); err != nil { + r.say("the value of %s is not fetched, so this node stays on %s: %v", next, nl.Login, err) + return + } + if err := r.Logins.Hold(next, password); err != nil { + r.say("the value of %s is not kept, so this node stays on %s: %v", next, nl.Login, err) + return + } + } + if err := r.TestDial(ctx, next, password); err != nil { + r.say("the test-dial as %s did not authenticate, so this node stays on %s: %v", next, nl.Login, err) + return + } + if err := r.Logins.Use(next); err != nil { + r.say("this node cannot switch to %s: %v", next, err) + return + } + r.old, r.failures, r.switchedSince = nl.Login, 0, nl.Since.String() + r.say("the test-dial as %s authenticated: this node dials as %s from now, and keeps the value of %s until the vault says the rotation is done", + next, next, nl.Login) + r.confirm(ctx, next) +} + +func (r *Rotator) confirm(ctx context.Context, login string) { + if r.Confirm == nil { + return + } + if err := r.Confirm(ctx); err != nil { + r.say("node-account-confirm as %s is not answered yet (asked again when the key changes or the link opens): %v", login, err) + return + } + r.say("node-account-confirm as %s is answered", login) +} + +// AuthFailed is told of each failed authentication as login: before rotation-done, repeated failures with the new login (the +// one switched to) bring the engine back to the old value, saying node-account-fallback (§6). +func (r *Rotator) AuthFailed(login string) { + r.mu.Lock() + defer r.mu.Unlock() + cur, _ := r.Logins.Current() + if r.old == "" || login != cur { + return + } + r.failures++ + bound := r.FailBound + if bound == 0 { + bound = 3 + } + if r.failures < bound { + return + } + if err := r.Logins.Use(r.old); err != nil { + r.say("cannot fall back to %s: %v", r.old, err) + return + } + text := fmt.Sprintf("%s failed to authenticate %d times before the vault said rotation-done: this node dials as %s again", login, r.failures, r.old) + r.say("%s", text) + if r.Finding != nil { + r.Finding(delivery.NodeAccountFallbackFinding, text) + } + r.old, r.failures = "", 0 + r.fellBackSince = r.switchedSince +} + +// AuthOK is told of each successful connection as login: the failures with it are forgotten. +func (r *Rotator) AuthOK(login string) { + r.mu.Lock() + defer r.mu.Unlock() + if cur, _ := r.Logins.Current(); login == cur { + r.failures = 0 + } +} + +// Watch makes the engine's dials tell r how they went: every refusal of the login in use counts towards the fallback. +func (r *Rotator) Watch() { + r.Logins.OnDial = func(user string, authenticated bool) { + if authenticated { + r.AuthOK(user) + } else { + r.AuthFailed(user) + } + } +} diff --git a/internal/link/rotate_nats.go b/internal/link/rotate_nats.go new file mode 100644 index 0000000..b693404 --- /dev/null +++ b/internal/link/rotate_nats.go @@ -0,0 +1,35 @@ +package link + +import ( + "context" + "fmt" + "time" + + "github.com/nats-io/nats.go" +) + +// TestDial dials a second connection to the bus as login with password, on the pinned certificate and the node's one inbox, +// and answers whether it authenticated (novox/hq ADR 0583 §3): the connection is closed at once, nothing is subscribed or said. +func TestDial(ctx context.Context, m Membership, login, password string, timeout time.Duration) error { + config, err := PinnedConfig(m.Fingerprint) + if err != nil { + return err + } + conn, err := nats.Connect(natsURL(m.Broker), + nats.Secure(config), + nats.UserInfo(login, password), + nats.CustomInboxPrefix("_INBOX.node."+m.Node), + nats.Name("mesh-host/"+m.Node+"/test-dial"), + nats.Timeout(timeout), + nats.NoReconnect()) + if err != nil { + return fmt.Errorf("cannot dial as %s: %w", login, err) + } + defer conn.Close() + flush, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + if err := conn.FlushWithContext(flush); err != nil { + return fmt.Errorf("the bus did not answer as %s: %w", login, err) + } + return nil +} diff --git a/internal/link/rotate_test.go b/internal/link/rotate_test.go new file mode 100644 index 0000000..89388e9 --- /dev/null +++ b/internal/link/rotate_test.go @@ -0,0 +1,148 @@ +package link + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// rotRig is a Rotator over stand-ins: what the vault hands, what a test-dial answers, and what it was asked, in order. +type rotRig struct { + r *Rotator + logins *Logins + states []LoginsState + steps []string + dialErr error + said []string + finding string +} + +func newRotRig() *rotRig { + g := &rotRig{} + g.logins = NewLogins("ace", "first", LoginsState{}, func(s LoginsState) error { + g.states = append(g.states, LoginsState{Current: s.Current, Values: map[string]string{}}) + for k, v := range s.Values { + g.states[len(g.states)-1].Values[k] = v + } + return nil + }) + g.r = &Rotator{Logins: g.logins, + Fetch: func(_ context.Context, login string) (string, error) { + g.steps = append(g.steps, "fetch "+login) + return "secret-of-" + login, nil + }, + TestDial: func(_ context.Context, login, _ string) error { + g.steps = append(g.steps, "test-dial "+login) + return g.dialErr + }, + Confirm: func(context.Context) error { + cur, _ := g.logins.Current() + g.steps = append(g.steps, "confirm as "+cur) + return nil + }, + Say: func(s string) { g.said = append(g.said, s) }, + Finding: func(name, text string) { g.finding = name }, + FailBound: 2} + g.r.Watch() + return g +} + +func key(login, rotating, stage string) delivery.NodeLogin { + return delivery.NodeLogin{Node: "ace", Login: login, Rotating: rotating, Stage: stage, Since: time.Date(2026, 11, 2, 12, 0, 0, 0, time.UTC)} +} + +// 774/S62/AC1 and AC2 (ADR 0583 §3): at ready the engine fetches the new value and test-dials it; one that does not +// authenticate leaves it on the old login, saying why; one that does is switched to, confirmed over the new connection, and +// the old value is kept. +func TestTheEngineSwitchesOnlyAfterATestDialAndConfirmsAsTheNewLogin(t *testing.T) { + g := newRotRig() + ctx := context.Background() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationBegun)) + if len(g.steps) != 0 { + t.Fatalf("before ready: %v", g.steps) + } + g.dialErr = errors.New("authorization violation") + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if cur, _ := g.logins.Current(); cur != "node.ace" || !strings.Contains(strings.Join(g.said, "|"), "stays on node.ace") || + slices.Contains(g.steps, "confirm as node.ace--b") { + t.Fatalf("a test-dial that failed: %s, said %v, steps %v", cur, g.said, g.steps) + } + g.dialErr = nil + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + cur, pw := g.logins.Current() + if cur != "node.ace--b" || pw != "secret-of-node.ace--b" { + t.Fatalf("after a good test-dial: %s %s", cur, pw) + } + if want := []string{"fetch node.ace--b", "test-dial node.ace--b", "test-dial node.ace--b", "confirm as node.ace--b"}; !slices.Equal(g.steps, want) { + t.Errorf("steps %v, want %v", g.steps, want) + } + if v, ok := g.logins.Value("node.ace"); !ok || v != "first" { + t.Error("the old value was not kept at the switch") + } + select { + case <-g.logins.Changed(): + default: + t.Error("the link was not told to open again as the new login") + } +} + +// 774/S63/AC1-AC3 (ADR 0583 §4-§6): the old value is dropped at rotation-done and not before; an abandon after the switch, +// before or after the confirmation, switches back before the new value is dropped; repeated failures with the new value before +// rotation-done bring the old one back, said as node-account-fallback, and that rotation is not tried again. +func TestTheOldValueGoesAtRotationDoneAbandonSwitchesBackAndFailuresFallBack(t *testing.T) { + ctx := context.Background() + for _, stage := range []string{"before confirming", "after confirming"} { + g := newRotRig() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if stage == "after confirming" { + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationConfirmed)) + } + if _, held := g.logins.Value("node.ace"); !held { + t.Fatalf("%s: the old value was dropped before rotation-done", stage) + } + g.states = nil + g.r.Heard(ctx, key("node.ace", "", "")) // abandon: the vault keeps the old login in use, no rotation + if cur, _ := g.logins.Current(); cur != "node.ace" { + t.Fatalf("%s: after abandon the engine dials as %s", stage, cur) + } + for _, s := range g.states { + if _, held := s.Values["node.ace--b"]; !held && s.Current != "node.ace" { + t.Errorf("%s: the new value was dropped while %s was in use: %+v", stage, s.Current, s) + } + } + if _, held := g.logins.Value("node.ace--b"); held { + t.Errorf("%s: the abandoned value is kept", stage) + } + } + g := newRotRig() + g.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + g.r.Heard(ctx, key("node.ace--b", "", "")) // rotation-done: the vault retired the old login + if _, held := g.logins.Value("node.ace"); held { + t.Error("the old value is kept after rotation-done") + } + if cur, _ := g.logins.Current(); cur != "node.ace--b" { + t.Errorf("in use after done: %s", cur) + } + // Fallback: two failures with the new value (the bound) before rotation-done. + f := newRotRig() + f.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + f.logins.dialed("node.ace--b", errors.New("nats: authorization violation")) + if cur, _ := f.logins.Current(); cur != "node.ace--b" { + t.Fatalf("fell back after one failure: %s", cur) + } + f.logins.dialed("node.ace--b", errors.New("nats: authorization violation")) + if cur, _ := f.logins.Current(); cur != "node.ace" || f.finding != delivery.NodeAccountFallbackFinding { + t.Fatalf("after the bound: %s, finding %q", cur, f.finding) + } + steps := len(f.steps) + f.r.Heard(ctx, key("node.ace", "node.ace--b", delivery.RotationReady)) + if len(f.steps) != steps { + t.Errorf("the rotation fallen back from was tried again: %v", f.steps[steps:]) + } + f.logins.dialed("node.ace", errors.New("dial tcp: connection refused")) // not a refusal of credentials: counts for nothing +} diff --git a/internal/link/session.go b/internal/link/session.go index 5f63d1f..dfe1d3b 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -24,6 +24,8 @@ const Writer = "node-engine" // event of the node, ordered by the node-engine's own epoch and sequence. Nothing here receives, orders or reconciles // a whole declaration. type Session struct { + // linkedUser is the bus user the link open now authenticated as, guarded by mu (WhenLinkedAs). + linkedUser string // Membership is the node and the mesh's key a controller's ask is verified with. Membership Membership // OnLink is run, off the link's own loop, each time a link opens and when the controller asks the node to say its @@ -197,6 +199,10 @@ func (q *Session) Attach(ctx context.Context, bus Bus) { q.init() q.mu.Lock() q.bus = bus + q.linkedUser = "" + if u, ok := bus.(interface{ User() string }); ok { + q.linkedUser = u.User() + } q.gone = make(chan struct{}) close(q.changed) q.changed = make(chan struct{}) @@ -224,6 +230,7 @@ func (q *Session) Detach(bus Bus) { q.mu.Lock() if q.bus == bus { q.bus = nil + q.linkedUser = "" close(q.gone) close(q.changed) q.changed = make(chan struct{}) diff --git a/internal/link/watch.go b/internal/link/watch.go index 145d5aa..9f03840 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -196,6 +196,26 @@ func (q *Session) WhenLinked(ctx context.Context) (context.Context, context.Canc } } +// WhenLinkedAs waits for a link open that authenticated as user: after a switch of login, the vault is asked over the new +// connection and no other (ADR 0583 §3). +func (q *Session) WhenLinkedAs(ctx context.Context, user string) error { + q.init() + for { + q.mu.Lock() + if q.bus != nil && q.linkedUser == user { + q.mu.Unlock() + return nil + } + changed := q.changed + q.mu.Unlock() + select { + case <-ctx.Done(): + return ctx.Err() + case <-changed: + } + } +} + // EventBus is a link that can hear an event on a subject of core NATS: the node agent waits for a backup's end with it, // and never polls. type EventBus interface {