From df39f2e9b609d4c890b86db3a938a65b6755154f Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Fri, 11 Sep 2026 00:53:11 -0500 Subject: [PATCH 1/3] 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 +++++++++++++++++-- services/versions.json | 2 +- 3 files changed, 92 insertions(+), 14 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 { diff --git a/services/versions.json b/services/versions.json index 29d8c230..19241b5a 100644 --- a/services/versions.json +++ b/services/versions.json @@ -13,7 +13,7 @@ "nvpair-node-settings": "1.0.4", "nvpair-ui-broker": "0.40.2", "nvpair-engine-manager": "0.17.4", - "nvpair-cluster-manager": "1.1.4", + "nvpair-cluster-manager": "1.1.5", "nvpair-job-scheduler": "0.4.1", "nvpair-tui": "0.7.2" } From be34de20a8d0cccf9a0c5ac6f2994def4979d44a Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Fri, 11 Sep 2026 01:35:48 -0500 Subject: [PATCH 2/3] 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 da7df0a48060d7e729519fb5d6884778fb604c29 Mon Sep 17 00:00:00 2001 From: Josh Stevenson Date: Sat, 12 Sep 2026 23:16:26 -0500 Subject: [PATCH 3/3] 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) } }