From b45e4a75dc1b14596e492c9b43ab128b8e088b72 Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Fri, 11 Sep 2026 00:53:11 -0500 Subject: [PATCH 1/4] fix(cluster-manager): announce only persisted endorsement changes Emit truststore change announcements only after an endorsement merge adds and persists a new endorsement. Keep idempotent re-pins, empty or duplicate merges, and failed writes silent, with component-level regression and race coverage. Signed-off-by: Josh Stevenson --- services/nvpair-cluster-manager/truststore.go | 19 +++-- .../truststore_announce_test.go | 85 +++++++++++++++++-- 2 files changed, 91 insertions(+), 13 deletions(-) diff --git a/services/nvpair-cluster-manager/truststore.go b/services/nvpair-cluster-manager/truststore.go index 592897b9..aa425a11 100644 --- a/services/nvpair-cluster-manager/truststore.go +++ b/services/nvpair-cluster-manager/truststore.go @@ -207,8 +207,8 @@ func (ts *TrustStore) Pin(pin *TrustedPin) error { } // Identical re-pin: fold in any newly-seen endorsements so the // gossiped trust web thickens (idempotent on the cert itself). - err := ts.mergeEndorsementsLocked(pin.NodeUUID, pin.Endorsements) - changed = err == nil + merged, err := ts.mergeEndorsementsLocked(pin.NodeUUID, pin.Endorsements) + changed = merged return err } return fmt.Errorf("uuid %s already pinned to a different certificate; remove it first to re-pin", pin.NodeUUID) @@ -239,10 +239,10 @@ func (ts *TrustStore) writePinLocked(pin *TrustedPin) error { // mergeEndorsementsLocked unions incoming endorsements into an existing pin // (dedup by signer+signature) and persists if anything new was added. Caller // holds ts.mu. A missing pin is a silent no-op. -func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsement) error { +func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsement) (bool, error) { pin, ok := ts.pins[uuid] if !ok || len(incoming) == 0 { - return nil + return false, nil } seen := make(map[string]struct{}, len(pin.Endorsements)) for _, e := range pin.Endorsements { @@ -260,9 +260,12 @@ func (ts *TrustStore) mergeEndorsementsLocked(uuid string, incoming []Endorsemen added = true } if !added { - return nil + return false, nil } - return ts.writePinLocked(updated) + if err := ts.writePinLocked(updated); err != nil { + return false, err + } + return true, nil } func endorsementKey(e Endorsement) string { @@ -279,8 +282,8 @@ func (ts *TrustStore) AddEndorsements(uuid string, endorsements []Endorsement) e defer ts.announce(&changed) ts.mu.Lock() defer ts.mu.Unlock() - err := ts.mergeEndorsementsLocked(uuid, endorsements) - changed = err == nil + merged, err := ts.mergeEndorsementsLocked(uuid, endorsements) + changed = merged return err } diff --git a/services/nvpair-cluster-manager/truststore_announce_test.go b/services/nvpair-cluster-manager/truststore_announce_test.go index 1fb7c387..cf61b7ae 100644 --- a/services/nvpair-cluster-manager/truststore_announce_test.go +++ b/services/nvpair-cluster-manager/truststore_announce_test.go @@ -4,6 +4,8 @@ package main import ( + "os" + "path/filepath" "testing" "time" ) @@ -80,6 +82,76 @@ func TestTrustStoreAnnouncesEveryMutation(t *testing.T) { } } +func TestTrustStoreAnnouncesNewEndorsementOnce(t *testing.T) { + ts, count := newAnnouncingStore(t) + const uuid = "principal-peer" + + if err := ts.Pin(testPin(t, uuid)); err != nil { + t.Fatalf("pin: %v", err) + } + endorsement := Endorsement{ + By: "trusted-peer", + Fingerprint: "sha256:target", + ClusterID: "cluster-1", + IssuedAt: 1, + Sig: "signature-1", + } + + if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err != nil { + t.Fatalf("add endorsement: %v", err) + } + if count() != 2 { + t.Fatalf("announcements after new endorsement = %d, want 2", count()) + } + stored, ok := ts.Get(uuid) + if !ok || len(stored.Endorsements) != 1 || stored.Endorsements[0] != endorsement { + t.Fatalf("endorsement was not persisted: %+v", stored) + } + + if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err != nil { + t.Fatalf("repeat endorsement: %v", err) + } + if count() != 2 { + t.Fatalf("announcements after duplicate endorsement = %d, want 2", count()) + } + stored, ok = ts.Get(uuid) + if !ok || len(stored.Endorsements) != 1 { + t.Fatalf("duplicate endorsement changed persisted state: %+v", stored) + } +} + +func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { + ts, count := newAnnouncingStore(t) + const uuid = "principal-peer" + if err := ts.Pin(testPin(t, uuid)); err != nil { + t.Fatalf("pin: %v", err) + } + beforeCount := count() + + blocker := filepath.Join(t.TempDir(), "not-a-directory") + if err := os.WriteFile(blocker, []byte("x"), 0o600); err != nil { + t.Fatalf("create persistence blocker: %v", err) + } + ts.dir = blocker + endorsement := Endorsement{ + By: "trusted-peer", + Fingerprint: "sha256:target", + ClusterID: "cluster-1", + IssuedAt: 1, + Sig: "signature-1", + } + if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err == nil { + t.Fatal("failed endorsement write unexpectedly succeeded") + } + if count() != beforeCount { + t.Fatalf("announcements after failed endorsement write = %d, want %d", count(), beforeCount) + } + stored, ok := ts.Get(uuid) + if !ok || len(stored.Endorsements) != 0 { + t.Fatalf("failed endorsement write mutated live state: %+v", stored) + } +} + // TestTrustStoreStaysSilentWhenNothingChanged keeps the announcement meaningful. // The scanner answers it by walking its whole directory, and the broker relays // it, so a store that announced on every call — including the idempotent re-pin @@ -96,11 +168,14 @@ func TestTrustStoreStaysSilentWhenNothingChanged(t *testing.T) { before := count() // An identical re-pin folds in no new endorsements and rewrites nothing. - if err := ts.Pin(testPin(t, uuid)); err == nil { - // A fresh leaf for the same uuid is a DIFFERENT certificate, which the - // store refuses rather than silently re-pinning; either way it must not - // announce a change it did not make. - t.Log("re-pin with a new certificate was accepted") + // Reuse the exact pin: generating another fixture would mint a different + // certificate and exercise the key-rotation rejection path instead. + if err := ts.Pin(pin); err != nil { + t.Fatalf("identical re-pin: %v", err) + } + // An empty endorsement merge is also a no-op and must stay silent. + if err := ts.AddEndorsements(uuid, nil); err != nil { + t.Fatalf("empty endorsement merge: %v", err) } // A rename to the values already stored changes nothing. if ok, err := ts.UpdateIdentity(uuid, uuid, uuid); err != nil || ok { From a2f0380e75eb525db0a78095280b0d240374b3fa Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Fri, 11 Sep 2026 01:35:48 -0500 Subject: [PATCH 2/4] test(cluster-manager): cover endorsement on identical pins Signed-off-by: Josh Stevenson --- .../truststore_announce_test.go | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/services/nvpair-cluster-manager/truststore_announce_test.go b/services/nvpair-cluster-manager/truststore_announce_test.go index cf61b7ae..d6f50fe8 100644 --- a/services/nvpair-cluster-manager/truststore_announce_test.go +++ b/services/nvpair-cluster-manager/truststore_announce_test.go @@ -120,6 +120,43 @@ func TestTrustStoreAnnouncesNewEndorsementOnce(t *testing.T) { } } +func TestTrustStoreAnnouncesNewEndorsementOnIdenticalPinOnce(t *testing.T) { + ts, count := newAnnouncingStore(t) + const uuid = "principal-peer" + pin := testPin(t, uuid) + + if err := ts.Pin(pin); err != nil { + t.Fatalf("pin: %v", err) + } + endorsement := Endorsement{ + By: "trusted-peer", + Fingerprint: "sha256:target", + ClusterID: "cluster-1", + IssuedAt: 1, + Sig: "signature-1", + } + withEndorsement := *pin + withEndorsement.Endorsements = []Endorsement{endorsement} + + if err := ts.Pin(&withEndorsement); err != nil { + t.Fatalf("re-pin with new endorsement: %v", err) + } + if count() != 2 { + t.Fatalf("announcements after new endorsement on identical pin = %d, want 2", count()) + } + stored, ok := ts.Get(uuid) + if !ok || len(stored.Endorsements) != 1 || stored.Endorsements[0] != endorsement { + t.Fatalf("endorsement was not persisted through identical pin: %+v", stored) + } + + if err := ts.Pin(&withEndorsement); err != nil { + t.Fatalf("repeat re-pin with endorsement: %v", err) + } + if count() != 2 { + t.Fatalf("announcements after duplicate endorsement on identical pin = %d, want 2", count()) + } +} + func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { ts, count := newAnnouncingStore(t) const uuid = "principal-peer" From eff875a45e269cc2a1b7d2bd3796bd53e14c2192 Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Sat, 12 Sep 2026 23:16:26 -0500 Subject: [PATCH 3/4] test(cluster-manager): verify persisted endorsement notifications Verify callback read-back and disk reload for direct and identical-pin merges, including mixed and duplicate batches. Exercise replace failures through both entry points, preserve old disk and live state, and verify recovery without temporary residue. Add concurrent identical submissions and missing-target regression coverage. Signed-off-by: Josh Stevenson --- .../truststore_announce_test.go | 284 ++++++++++++------ 1 file changed, 193 insertions(+), 91 deletions(-) diff --git a/services/nvpair-cluster-manager/truststore_announce_test.go b/services/nvpair-cluster-manager/truststore_announce_test.go index d6f50fe8..1dbe8b15 100644 --- a/services/nvpair-cluster-manager/truststore_announce_test.go +++ b/services/nvpair-cluster-manager/truststore_announce_test.go @@ -4,8 +4,13 @@ package main import ( + "bytes" + "errors" "os" "path/filepath" + "reflect" + "sync" + "sync/atomic" "testing" "time" ) @@ -46,8 +51,8 @@ func testPin(t *testing.T, uuid string) *TrustedPin { // announcement exists to prevent — so the hook lives on the store rather than at // the ~19 call sites that pin and unpin peers. // -// Pinning, removing, forgetting, and a display-name update all mutate what is on -// disk and must each announce exactly once. +// Pinning, removing, and a display-name update mutate disk state; forgetting +// intentionally changes only live authorization. Each must announce once. func TestTrustStoreAnnouncesEveryMutation(t *testing.T) { ts, count := newAnnouncingStore(t) const uuid = "principal-peer" @@ -82,110 +87,207 @@ func TestTrustStoreAnnouncesEveryMutation(t *testing.T) { } } -func TestTrustStoreAnnouncesNewEndorsementOnce(t *testing.T) { - ts, count := newAnnouncingStore(t) - const uuid = "principal-peer" - - if err := ts.Pin(testPin(t, uuid)); err != nil { - t.Fatalf("pin: %v", err) - } - endorsement := Endorsement{ - By: "trusted-peer", - Fingerprint: "sha256:target", - ClusterID: "cluster-1", - IssuedAt: 1, - Sig: "signature-1", - } - - if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err != nil { - t.Fatalf("add endorsement: %v", err) - } - if count() != 2 { - t.Fatalf("announcements after new endorsement = %d, want 2", count()) - } - stored, ok := ts.Get(uuid) - if !ok || len(stored.Endorsements) != 1 || stored.Endorsements[0] != endorsement { - t.Fatalf("endorsement was not persisted: %+v", stored) - } - - if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err != nil { - t.Fatalf("repeat endorsement: %v", err) +// assertStoredEndorsements checks the live snapshot and a separately loaded +// store. It is also safe to call from an onChange callback in a joined worker. +func assertStoredEndorsements(t *testing.T, ts *TrustStore, uuid string, want []Endorsement) { + t.Helper() + pin, ok := ts.Get(uuid) + if !ok || !reflect.DeepEqual(pin.Endorsements, want) { + t.Errorf("live endorsements = %+v, want %+v", pin, want) } - if count() != 2 { - t.Fatalf("announcements after duplicate endorsement = %d, want 2", count()) + reloaded, err := newTrustStore(filepath.Dir(ts.dir)) + if err != nil { + t.Errorf("reload trust store: %v", err) + return } - stored, ok = ts.Get(uuid) - if !ok || len(stored.Endorsements) != 1 { - t.Fatalf("duplicate endorsement changed persisted state: %+v", stored) + pin, ok = reloaded.Get(uuid) + if !ok || !reflect.DeepEqual(pin.Endorsements, want) { + t.Errorf("reloaded endorsements = %+v, want %+v", pin, want) } } -func TestTrustStoreAnnouncesNewEndorsementOnIdenticalPinOnce(t *testing.T) { - ts, count := newAnnouncingStore(t) - const uuid = "principal-peer" - pin := testPin(t, uuid) - - if err := ts.Pin(pin); err != nil { - t.Fatalf("pin: %v", err) - } - endorsement := Endorsement{ - By: "trusted-peer", - Fingerprint: "sha256:target", - ClusterID: "cluster-1", - IssuedAt: 1, - Sig: "signature-1", +func TestTrustStoreAnnouncesNewEndorsementsAfterPersistence(t *testing.T) { + for _, identicalPin := range []bool{false, true} { + name := "AddEndorsements" + if identicalPin { + name = "IdenticalPin" + } + t.Run(name, func(t *testing.T) { + ts, _ := newAnnouncingStore(t) + pin := testPin(t, "principal-peer") + first := Endorsement{By: "trusted-peer", Sig: "signature-1"} + second := Endorsement{By: "trusted-peer", SigV2: "signature-2"} + pin.Endorsements = []Endorsement{first} + if err := ts.Pin(pin); err != nil { + t.Fatalf("pin: %v", err) + } + want := []Endorsement{first, second} + calls := 0 + ts.SetOnChange(func() { + calls++ + // Acquiring Get's read lock here also witnesses that the + // mutation lock was released before announcing the change. + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) + merge := func(batch []Endorsement) error { + if identicalPin { + updated := *pin + updated.Endorsements = batch + return ts.Pin(&updated) + } + return ts.AddEndorsements(pin.NodeUUID, batch) + } + // Mix an existing endorsement, a new endorsement, and an + // in-batch duplicate. One operation causes one announcement. + batch := []Endorsement{first, second, second} + if err := merge(batch); err != nil { + t.Fatalf("merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after merge = %d, want 1", calls) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + before, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatal(err) + } + for _, noOp := range [][]Endorsement{batch, nil} { + if err := merge(noOp); err != nil { + t.Fatalf("no-op merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after no-op = %d, want 1", calls) + } + } + after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil || !bytes.Equal(before, after) { + t.Fatalf("no-op changed disk contents: %v", err) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) } - withEndorsement := *pin - withEndorsement.Endorsements = []Endorsement{endorsement} +} - if err := ts.Pin(&withEndorsement); err != nil { - t.Fatalf("re-pin with new endorsement: %v", err) - } - if count() != 2 { - t.Fatalf("announcements after new endorsement on identical pin = %d, want 2", count()) - } - stored, ok := ts.Get(uuid) - if !ok || len(stored.Endorsements) != 1 || stored.Endorsements[0] != endorsement { - t.Fatalf("endorsement was not persisted through identical pin: %+v", stored) +func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { + for _, identicalPin := range []bool{false, true} { + name := "AddEndorsements" + if identicalPin { + name = "IdenticalPin" + } + t.Run(name, func(t *testing.T) { + ts, count := newAnnouncingStore(t) + pin := testPin(t, "principal-peer") + first := Endorsement{By: "trusted-peer", Sig: "signature-1"} + second := Endorsement{By: "trusted-peer", SigV2: "signature-2"} + pin.Endorsements = []Endorsement{first} + if err := ts.Pin(pin); err != nil { + t.Fatalf("pin: %v", err) + } + before, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil { + t.Fatal(err) + } + beforeCount := count() + // Fail the final replace, after the temporary file was written. + // The existing pin must remain intact on disk and in memory. + originalRename := renameFile + writeErr := errors.New("injected endorsement replace failure") + renameFile = func(_, _ string) error { return writeErr } + t.Cleanup(func() { renameFile = originalRename }) + merge := func() error { + if identicalPin { + updated := *pin + updated.Endorsements = []Endorsement{second} + return ts.Pin(&updated) + } + return ts.AddEndorsements(pin.NodeUUID, []Endorsement{second}) + } + if err := merge(); !errors.Is(err, writeErr) { + t.Fatalf("merge error = %v, want injected replace failure", err) + } + if count() != beforeCount { + t.Fatalf("announcements after failed write = %d, want %d", count(), beforeCount) + } + after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) + if err != nil || !bytes.Equal(before, after) { + t.Fatalf("failed write changed disk contents: %v", err) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first}) + entries, err := os.ReadDir(ts.dir) + if err != nil || len(entries) != 1 || entries[0].Name() != pin.NodeUUID+".json" { + t.Fatalf("failed write left temporary residue: entries=%v err=%v", entries, err) + } + renameFile = originalRename + if err := merge(); err != nil { + t.Fatalf("retry after storage recovery: %v", err) + } + if count() != beforeCount+1 { + t.Fatalf("announcements after retry = %d, want %d", count(), beforeCount+1) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first, second}) + }) } +} - if err := ts.Pin(&withEndorsement); err != nil { - t.Fatalf("repeat re-pin with endorsement: %v", err) - } - if count() != 2 { - t.Fatalf("announcements after duplicate endorsement on identical pin = %d, want 2", count()) +func TestTrustStoreConcurrentDuplicateEndorsementsAnnounceOnce(t *testing.T) { + ts, err := newTrustStore(t.TempDir()) + if err != nil { + t.Fatal(err) } + pin := testPin(t, "principal-peer") + if err := ts.Pin(pin); err != nil { + t.Fatal(err) + } + endorsement := Endorsement{By: "trusted-peer", SigV2: "signature-1"} + want := []Endorsement{endorsement} + var calls atomic.Int32 + ts.SetOnChange(func() { + calls.Add(1) + assertStoredEndorsements(t, ts, pin.NodeUUID, want) + }) + const workers = 16 + start := make(chan struct{}) + errs := make(chan error, workers) + var wg sync.WaitGroup + for i := range workers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if i%2 == 0 { + updated := *pin + updated.Endorsements = want + errs <- ts.Pin(&updated) + } else { + errs <- ts.AddEndorsements(pin.NodeUUID, want) + } + }() + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Errorf("concurrent merge: %v", err) + } + } + if got := calls.Load(); got != 1 { + t.Errorf("announcements for concurrent identical submissions = %d, want 1", got) + } + assertStoredEndorsements(t, ts, pin.NodeUUID, want) } -func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { +func TestTrustStoreMissingEndorsementTargetStaysSilent(t *testing.T) { ts, count := newAnnouncingStore(t) - const uuid = "principal-peer" - if err := ts.Pin(testPin(t, uuid)); err != nil { - t.Fatalf("pin: %v", err) - } - beforeCount := count() - - blocker := filepath.Join(t.TempDir(), "not-a-directory") - if err := os.WriteFile(blocker, []byte("x"), 0o600); err != nil { - t.Fatalf("create persistence blocker: %v", err) - } - ts.dir = blocker - endorsement := Endorsement{ - By: "trusted-peer", - Fingerprint: "sha256:target", - ClusterID: "cluster-1", - IssuedAt: 1, - Sig: "signature-1", - } - if err := ts.AddEndorsements(uuid, []Endorsement{endorsement}); err == nil { - t.Fatal("failed endorsement write unexpectedly succeeded") + if err := ts.AddEndorsements("principal-stranger", []Endorsement{{By: "trusted-peer", SigV2: "signature-1"}}); err != nil { + t.Fatal(err) } - if count() != beforeCount { - t.Fatalf("announcements after failed endorsement write = %d, want %d", count(), beforeCount) + if count() != 0 || len(ts.List()) != 0 { + t.Fatalf("missing-target merge changed live state: announcements=%d pins=%v", count(), ts.List()) } - stored, ok := ts.Get(uuid) - if !ok || len(stored.Endorsements) != 0 { - t.Fatalf("failed endorsement write mutated live state: %+v", stored) + entries, err := os.ReadDir(ts.dir) + if err != nil || len(entries) != 0 { + t.Fatalf("missing-target merge changed disk state: entries=%v err=%v", entries, err) } } From 6cdc6c3572540e2cd7a645fdc497c7f93161f65a Mon Sep 17 00:00:00 2001 From: Kaylee Lubick Date: Wed, 23 Sep 2026 14:13:58 -0400 Subject: [PATCH 4/4] slight refactor of test Signed-off-by: Kaylee Lubick --- services/nvpair-cluster-manager/spec.md | 5 +- .../truststore_announce_test.go | 89 ++++++++++--------- 2 files changed, 51 insertions(+), 43 deletions(-) diff --git a/services/nvpair-cluster-manager/spec.md b/services/nvpair-cluster-manager/spec.md index ea65943e..fc633ffe 100644 --- a/services/nvpair-cluster-manager/spec.md +++ b/services/nvpair-cluster-manager/spec.md @@ -164,6 +164,7 @@ All traffic is JSON-RPC 2.0. The local interface uses `stdin`/`stdout` (or a nam | `cluster:invite-declined` | CM → Broker | notification | Inviter-side: the joiner declined; carries the `Invite` with `state:"declined"` so the UI can clear the PIN / abandon a throwaway solo cluster. | | `cluster:invite-failed` | CM → Broker | notification | Inviter-side: the Completion Exchange failed (e.g. a wrong PIN); carries the `Invite` with `state:"failed"` (and `reason:"incorrect-pin"` on a wrong PIN) so the UI can show an "Incorrect PIN" error / abandon a throwaway solo cluster. | | `cluster:identity-changed` | CM → Broker | notification | The local `clusterId` changed (e.g. the joiner adopted the inviter's on pairing); payload `{clusterId, clusterFriendlyName}` so the Broker can persist it to `nvpair-node-settings`. | +| `cluster:trust-changed` | CM → Broker | notification | The trusted-peer store changed, or live authorization was revoked; empty payload. The Broker refreshes consumers that cache trust-derived state. | | `nodes:changed` _(proposed, §4)_ | CM → Broker | notification | Full membership snapshot pushed on every change. | - **`cluster:get-node-id` (Broker → CM)**: pure read of *this* node's own identity. Params: none. Result `{nodeUuid, nodeId, name, certFingerprint, clusterId}`. The `nodeUuid` is the self-generated, persisted identity (§7.4); `nodeId`/`name` default to the OS hostname; `clusterId` is `""` until set, created, or adopted — a `""` here is the unclustered signal a UI can surface (e.g. render "not in a cluster" or offer an explicit `cluster:create`). The caller no longer needs to create before inviting: `cluster:invite-node` auto-founds when unclustered (§7.0/§7.2), so this read is informational rather than a required gate on the invite path. Always available (the identity is minted at startup), so the UI can render "this is me" immediately. @@ -183,6 +184,7 @@ All traffic is JSON-RPC 2.0. The local interface uses `stdin`/`stdout` (or a nam - **`cluster:invite-received` (CM → Broker)**: emitted once the inbound Initial Exchange completes. Params: the `Invite` (with `state: "pending"`, no `pin`). The Broker prompts the user to enter the PIN shown on the inviting node, then drives `cluster:respond-to-invite`. - **`cluster:invite-canceled` (CM → Broker)**: emitted on the **joiner** when the inviter cancels a still-pending inbound invite (a `phase:"cancel"` arrives on the pairing channel, §7.2), or when a newer invite from the same authenticated sender supersedes it. Params: the `Invite` (now `state: "canceled"`, no `pin`). The joiner drops the old pending-inbound invite/session; on supersession it retains the sender's single pending member row for the replacement invite. The Broker uses this event to dismiss the old PIN prompt before handling the replacement `cluster:invite-received`. Idempotent — a duplicate or late cancel for an already-resolved invite is a no-op and emits nothing. - **`cluster:identity-changed` (CM → Broker)**: emitted when the local `clusterId` originates **here** rather than from a `cluster:set-identity` call — i.e. in the two cases where this service creates or adopts a cluster identity: (a) `cluster:create` founding a new cluster, and (b) the joiner adopting the inviter's cluster on a successful pair (§4 working model). Params `{clusterId, clusterFriendlyName}`; the Broker persists it to `nvpair-node-settings`. Not emitted for changes the Broker itself drove via `cluster:set-identity` (those already came from node-settings). +- **`cluster:trust-changed` (CM → Broker)**: emitted after a trusted peer's pin is added, updated, or removed, or its live authorization is forgotten. An endorsement merge triggers it only after a new endorsement is persisted. Duplicate endorsements, same-cert and same-admission re-pins with no new endorsement, empty merges, missing endorsement targets, and failed endorsement writes remain silent. The payload is `{}`: the Broker re-derives trust-dependent state rather than applying an event diff. The store releases its mutation lock before emitting the notification, so a recipient can read the updated state. - **`nodes:changed` (CM → Broker, proposed)**: full `{nodes: ClusterNode[]}` snapshot pushed whenever membership changes (accept, decline, removal, peer-initiated removal). Lets the Broker stay live without re-polling `nodes:get-initial`. Marked proposed in §4. Example `cluster:invite-node` request and result (`stdin` → `stdout`): @@ -421,7 +423,8 @@ Operations on the store map directly to single-file filesystem actions, so an ed - **Pin** (on pairing success): write `trusted/.json.tmp` then atomically rename to `trusted/.json` (`0600`). - **Remove** (`nodes:remove`): `os.Remove("trusted/.json")` — a single atomic delete; a missing file is treated as already-removed (idempotent success). - **Load** (startup): read `trusted/*.json` into the in-memory `uuid → pin` map, **skipping** `*.tmp` and any file that doesn't parse. On load, verify the file's inner `nodeUuid` (and the `certPem` subject `CN`) matches the filename ``; reject/skip-and-log a mismatch so a renamed or tampered file cannot masquerade as another UUID. -- **Re-pin guard** (§7.2): re-pinning the *same* cert for an existing `.json` is a no-op; a *different* cert for an already-present `.json` is rejected, not overwritten (explicit re-pin/re-invite required — §12 key rotation). +- **Re-pin guard** (§7.2): re-pinning the *same* cert and admission epoch for an existing `.json` merges newly received endorsements. An older admission epoch is ignored; a newer one for the same cert is persisted as a new incarnation. A *different* cert for an already-present `.json` is rejected, not overwritten (explicit re-pin/re-invite required — §12 key rotation). +- **Endorsement merge**: both an identical re-pin and a direct addition to an existing pin deduplicate endorsements by signer and signature. A merge with no new endorsement does not rewrite the file or emit `cluster:trust-changed`; a failed endorsement write leaves the live and stored pin unchanged and emits nothing. Adding endorsements for a missing peer is a no-op and does not create a pin. Writes are serialized (membership changes are rare and human-driven), so readers tolerate seeing a just-added or just-removed file without a global lock; there is no single-file consistent snapshot, which is acceptable for this low-churn store. diff --git a/services/nvpair-cluster-manager/truststore_announce_test.go b/services/nvpair-cluster-manager/truststore_announce_test.go index 1dbe8b15..054e9521 100644 --- a/services/nvpair-cluster-manager/truststore_announce_test.go +++ b/services/nvpair-cluster-manager/truststore_announce_test.go @@ -106,12 +106,20 @@ func assertStoredEndorsements(t *testing.T, ts *TrustStore, uuid string, want [] } } +type endorsementMerge func(*TrustStore, *TrustedPin, []Endorsement) error + +func addEndorsements(ts *TrustStore, pin *TrustedPin, batch []Endorsement) error { + return ts.AddEndorsements(pin.NodeUUID, batch) +} + +func pinWithEndorsements(ts *TrustStore, pin *TrustedPin, batch []Endorsement) error { + updated := *pin + updated.Endorsements = batch + return ts.Pin(&updated) +} + func TestTrustStoreAnnouncesNewEndorsementsAfterPersistence(t *testing.T) { - for _, identicalPin := range []bool{false, true} { - name := "AddEndorsements" - if identicalPin { - name = "IdenticalPin" - } + test := func(name string, merge endorsementMerge) { t.Run(name, func(t *testing.T) { ts, _ := newAnnouncingStore(t) pin := testPin(t, "principal-peer") @@ -129,18 +137,10 @@ func TestTrustStoreAnnouncesNewEndorsementsAfterPersistence(t *testing.T) { // mutation lock was released before announcing the change. assertStoredEndorsements(t, ts, pin.NodeUUID, want) }) - merge := func(batch []Endorsement) error { - if identicalPin { - updated := *pin - updated.Endorsements = batch - return ts.Pin(&updated) - } - return ts.AddEndorsements(pin.NodeUUID, batch) - } // Mix an existing endorsement, a new endorsement, and an // in-batch duplicate. One operation causes one announcement. batch := []Endorsement{first, second, second} - if err := merge(batch); err != nil { + if err := merge(ts, pin, batch); err != nil { t.Fatalf("merge: %v", err) } if calls != 1 { @@ -151,29 +151,34 @@ func TestTrustStoreAnnouncesNewEndorsementsAfterPersistence(t *testing.T) { if err != nil { t.Fatal(err) } - for _, noOp := range [][]Endorsement{batch, nil} { - if err := merge(noOp); err != nil { - t.Fatalf("no-op merge: %v", err) - } - if calls != 1 { - t.Fatalf("announcements after no-op = %d, want 1", calls) - } + if err := merge(ts, pin, batch); err != nil { + t.Fatalf("duplicate merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after duplicate merge = %d, want 1", calls) + } + if err := merge(ts, pin, nil); err != nil { + t.Fatalf("empty merge: %v", err) + } + if calls != 1 { + t.Fatalf("announcements after empty merge = %d, want 1", calls) } after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) - if err != nil || !bytes.Equal(before, after) { - t.Fatalf("no-op changed disk contents: %v", err) + if err != nil { + t.Fatalf("read pin after no-op merges: %v", err) + } + if !bytes.Equal(before, after) { + t.Fatal("no-op merges changed disk contents") } assertStoredEndorsements(t, ts, pin.NodeUUID, want) }) } + test("AddEndorsements", addEndorsements) + test("IdenticalPin", pinWithEndorsements) } func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { - for _, identicalPin := range []bool{false, true} { - name := "AddEndorsements" - if identicalPin { - name = "IdenticalPin" - } + test := func(name string, merge endorsementMerge) { t.Run(name, func(t *testing.T) { ts, count := newAnnouncingStore(t) pin := testPin(t, "principal-peer") @@ -194,31 +199,29 @@ func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { writeErr := errors.New("injected endorsement replace failure") renameFile = func(_, _ string) error { return writeErr } t.Cleanup(func() { renameFile = originalRename }) - merge := func() error { - if identicalPin { - updated := *pin - updated.Endorsements = []Endorsement{second} - return ts.Pin(&updated) - } - return ts.AddEndorsements(pin.NodeUUID, []Endorsement{second}) - } - if err := merge(); !errors.Is(err, writeErr) { + if err := merge(ts, pin, []Endorsement{second}); !errors.Is(err, writeErr) { t.Fatalf("merge error = %v, want injected replace failure", err) } if count() != beforeCount { t.Fatalf("announcements after failed write = %d, want %d", count(), beforeCount) } after, err := os.ReadFile(ts.pinPath(pin.NodeUUID)) - if err != nil || !bytes.Equal(before, after) { - t.Fatalf("failed write changed disk contents: %v", err) + if err != nil { + t.Fatalf("read pin after failed write: %v", err) + } + if !bytes.Equal(before, after) { + t.Fatal("failed write changed disk contents") } assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first}) entries, err := os.ReadDir(ts.dir) - if err != nil || len(entries) != 1 || entries[0].Name() != pin.NodeUUID+".json" { - t.Fatalf("failed write left temporary residue: entries=%v err=%v", entries, err) + if err != nil { + t.Fatalf("list trusted directory after failed write: %v", err) + } + if len(entries) != 1 || entries[0].Name() != pin.NodeUUID+".json" { + t.Fatalf("failed write left temporary residue: entries=%v", entries) } renameFile = originalRename - if err := merge(); err != nil { + if err := merge(ts, pin, []Endorsement{second}); err != nil { t.Fatalf("retry after storage recovery: %v", err) } if count() != beforeCount+1 { @@ -227,6 +230,8 @@ func TestTrustStoreStaysSilentWhenEndorsementWriteFails(t *testing.T) { assertStoredEndorsements(t, ts, pin.NodeUUID, []Endorsement{first, second}) }) } + test("AddEndorsements", addEndorsements) + test("IdenticalPin", pinWithEndorsements) } func TestTrustStoreConcurrentDuplicateEndorsementsAnnounceOnce(t *testing.T) {