fix(cluster-manager): announce only persisted endorsement changes

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/<uuid>.json.tmp` then atomically rename to `trusted/<uuid>.json` (`0600`).
 - **Remove** (`nodes:remove`): `os.Remove("trusted/<uuid>.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 `<uuid>`; 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 `<uuid>.json` is a no-op; a *different* cert for an already-present `<uuid>.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 `<uuid>.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 `<uuid>.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.
 

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
 }
 

services/nvpair-cluster-manager/truststore_announce_test.go

@@ -4,6 +4,13 @@
 package main
 
 import (
+	"bytes"
+	"errors"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -44,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"
@@ -80,6 +87,215 @@ func TestTrustStoreAnnouncesEveryMutation(t *testing.T) {
 	}
 }
 
+// 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)
+	}
+	reloaded, err := newTrustStore(filepath.Dir(ts.dir))
+	if err != nil {
+		t.Errorf("reload trust store: %v", err)
+		return
+	}
+	pin, ok = reloaded.Get(uuid)
+	if !ok || !reflect.DeepEqual(pin.Endorsements, want) {
+		t.Errorf("reloaded endorsements = %+v, want %+v", pin, 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) {
+	test := func(name string, merge endorsementMerge) {
+		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)
+			})
+			// 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(ts, pin, 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)
+			}
+			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 {
+				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) {
+	test := func(name string, merge endorsementMerge) {
+		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 })
+			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 {
+				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 {
+				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(ts, pin, []Endorsement{second}); 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})
+		})
+	}
+	test("AddEndorsements", addEndorsements)
+	test("IdenticalPin", pinWithEndorsements)
+}
+
+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 TestTrustStoreMissingEndorsementTargetStaysSilent(t *testing.T) {
+	ts, count := newAnnouncingStore(t)
+	if err := ts.AddEndorsements("principal-stranger", []Endorsement{{By: "trusted-peer", SigV2: "signature-1"}}); err != nil {
+		t.Fatal(err)
+	}
+	if count() != 0 || len(ts.List()) != 0 {
+		t.Fatalf("missing-target merge changed live state: announcements=%d pins=%v", count(), ts.List())
+	}
+	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)
+	}
+}
+
 // 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 +312,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 {