Fix workload broadcast ordering and record dedup keys only after successful emit

README.md

@@ -239,6 +239,9 @@ Each entry assumes the ones before it.
 8. **[Developer guide](docs/developing.mdx)** — read this before contributing:
    where the code lives, how a change travels through the layers, and the
    conventions the project enforces.
+9. **[Workload broadcast reliability](docs/workload-broadcast-reliability.mdx)**
+   — ordered inter-node broadcasts and dedup-after-emit in the workload
+   manager.
 
 Component references, for when you already know what you are looking for:
 

docs/workload-broadcast-reliability.mdx

@@ -0,0 +1,126 @@
+{/*
+SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+SPDX-License-Identifier: Apache-2.0
+*/}
+
+# Workload broadcast reliability
+
+`services/nvpair-workload-manager` replicates workload lifecycle events to
+cluster peers over mutual TLS so every node's scheduler sees the same pending
+load. Two ordering bugs in that replication could corrupt a peer's view of the
+cluster: ghost workloads that never existed, and real events silently dropped.
+Both are fixed here. No JSON-RPC surface changed.
+
+## 1. Broadcasts go out in origin order
+
+### The problem
+
+`broadcastFrame` fanned every outbound frame out in its own goroutine so a
+slow peer could never block the read loop. Goroutines are not ordered, so a
+`workloads:remove` could overtake the lifecycle upsert it followed. The late
+upsert then landed on a peer that had already processed the remove and
+**resurrected a ghost workload** — phantom pending load in that peer's
+scheduler view, with nothing on the origin to ever correct it.
+
+```mermaid
+sequenceDiagram
+    participant Origin as origin read loop
+    participant Peer as peer
+    Note over Origin,Peer: before: one goroutine per frame
+    Origin->>Peer: workloads:remove (id 7)
+    Origin->>Peer: workload:started (id 7)
+    Note over Peer: remove applied, then stale<br/>upsert resurrects id 7
+```
+
+### The fix
+
+A single ordered worker now drains a bounded queue:
+
+- `broadcastCh` (capacity 1024) is created in `NewManager` and consumed by
+  `broadcastLoop`, started in `Run`.
+- `broadcastFrame` only enqueues — it still never blocks on network I/O, so a
+  slow peer still cannot wedge the read loop.
+- A full queue drops the frame with a warning instead of blocking. Frames are
+  small JSON notifications, so 1024 is far beyond steady-state volume; during
+  a peer outage the heartbeat and discovery backfill re-sync state, and a
+  dropped frame degrades to delayed convergence rather than a wedged node.
+- At shutdown the loop exits on context cancellation; queued frames are
+  dropped with the process.
+
+```mermaid
+sequenceDiagram
+    participant Origin as origin read loop
+    participant Queue as broadcastCh
+    participant Worker as broadcastLoop
+    participant Peer as peer
+    Note over Origin,Peer: after: single ordered consumer
+    Origin->>Queue: enqueue workload:started (id 7)
+    Origin->>Queue: enqueue workloads:remove (id 7)
+    Worker->>Queue: dequeue in order
+    Worker->>Peer: workload:started (id 7)
+    Worker->>Peer: workloads:remove (id 7)
+    Note over Peer: transitions apply in order;<br/>no resurrection
+```
+
+## 2. Dedup keys are recorded only after a successful emit
+
+### The problem
+
+The inter-node server deduplicates retried peer events with a key index. The
+old code recorded the key **before** emitting to the broker
+(`dedup.seenOrAdd`). If the emit failed, the server answered `500` — but the
+peer's retry arrived to find the key already recorded and was swallowed as a
+duplicate. The event was lost permanently: no anti-entropy repair exists for
+it, and the origin had already moved on.
+
+The removal path had the identical flaw: a failed `workloads:remove` emit
+meant the retry was deduplicated away and the ghost workload stayed in the
+broker catalog.
+
+### The fix
+
+`dedupIndex` split the check from the record: `seen()` reports without
+recording, `add()` records. Both `handleLifecycle` and `handleRemove` now
+follow the same sequence:
+
+1. If `seen(key)` → answer `200`, skip (genuine duplicate).
+2. Emit to the broker. On failure → answer `500` **without** recording the
+   key, so the peer's retry is treated as new.
+3. On success → `add(key)`.
+
+Resync/backfill frames still bypass dedup, as before — they intentionally
+re-assert state so the broker store can reconcile.
+
+```mermaid
+sequenceDiagram
+    participant Peer
+    participant Server as inter-node server
+    participant Broker
+
+    Peer->>Server: workload:started (id 7)
+    Server->>Server: seen(key)? no
+    Server->>Broker: emit upsert → fails
+    Server->>Peer: 500 (key NOT recorded)
+    Peer->>Server: retry workload:started (id 7)
+    Server->>Server: seen(key)? no
+    Server->>Broker: emit upsert → ok
+    Server->>Server: add(key)
+    Server->>Peer: 200
+    Peer->>Server: duplicate (id 7)
+    Server->>Server: seen(key)? yes
+    Server->>Peer: 200 (deduplicated)
+```
+
+A concurrent duplicate that passes `seen()` before either emit runs simply
+emits twice; the broker store is idempotent, so the double emit is absorbed.
+
+## Validation
+
+- `reliability_test.go` drives both fixes through the real cluster-mTLS
+  broadcast path: 25 frames arrive in exact origin order, and failed
+  lifecycle/remove emits return `500` without recording the key — the retry
+  emits exactly once, and a genuine duplicate afterwards is still
+  deduplicated. Run with `go test -race ./...` from
+  `services/nvpair-workload-manager`.
+- Component version bumped per `services/VERSIONING.md`:
+  `nvpair-workload-manager` 0.13.3 → 0.13.4.

services/nvpair-workload-manager/broadcast_order_test.go

@@ -0,0 +1,107 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"nvpair-shared/clustertrust"
+)
+
+// codecNop is a throwaway io.ReadWriter for a Manager whose broker side is
+// never exercised.
+type codecNop struct{}
+
+func (codecNop) Read([]byte) (int, error)    { return 0, io.EOF }
+func (codecNop) Write(p []byte) (int, error) { return len(p), nil }
+
+// TestBroadcastPreservesOriginOrder: frames must reach the peer in the order
+// the read loop produced them. The old code fanned each frame out in its own
+// goroutine, so a remove could overtake the lifecycle upsert it followed and
+// the late upsert resurrected a ghost workload on the peer (phantom pending
+// load in the scheduler's view). The ordered broadcast queue fixes this; the
+// test drives 25 frames through the real cluster-mTLS broadcast path and
+// asserts arrival order.
+func TestBroadcastPreservesOriginOrder(t *testing.T) {
+	selfCert, selfKey := genLeaf(t, "uuid-self")
+	peerCert, peerKey := genLeaf(t, "uuid-peer")
+	selfDir := setupNode(t, selfCert, selfKey, map[string][]byte{"uuid-peer": peerCert})
+	peerMesh := clustertrust.Open(setupNode(t, peerCert, peerKey, map[string][]byte{"uuid-self": selfCert}))
+
+	var mu sync.Mutex
+	var got []string
+	mux := http.NewServeMux()
+	mux.HandleFunc(eventsPath, func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(io.LimitReader(r.Body, 1<<20))
+		_ = r.Body.Close()
+		mu.Lock()
+		got = append(got, string(body))
+		mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	})
+	ts := httptest.NewUnstartedServer(mux)
+	ts.TLS = peerMesh.ServerTLSConfig()
+	ts.StartTLS()
+	t.Cleanup(ts.Close)
+
+	m := NewManager(NewCodec(codecNop{}), 0, "uuid-self", selfDir)
+	u, err := url.Parse(ts.URL)
+	if err != nil {
+		t.Fatalf("parse test server URL: %v", err)
+	}
+	host, portStr, err := net.SplitHostPort(u.Host)
+	if err != nil {
+		t.Fatalf("split test server hostport: %v", err)
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		t.Fatalf("parse test server port: %v", err)
+	}
+	m.peers.Replace([]PeerNode{{
+		ID: "peer-1", Addresses: []string{host}, Port: port,
+		TXT: []string{"cluster-uuid=uuid-peer"},
+	}})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go m.broadcastLoop(ctx)
+
+	const n = 25
+	for i := 0; i < n; i++ {
+		m.broadcastFrame("workload:started", []byte(fmt.Sprintf(`{"seq":%d}`, i)))
+	}
+
+	deadline := time.Now().Add(15 * time.Second)
+	for {
+		mu.Lock()
+		l := len(got)
+		mu.Unlock()
+		if l >= n || time.Now().After(deadline) {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(got) != n {
+		t.Fatalf("peer received %d of %d frames", len(got), n)
+	}
+	for i, f := range got {
+		if want := fmt.Sprintf(`"seq":%d`, i); !strings.Contains(f, want) {
+			t.Fatalf("frame %d arrived out of order: %s", i, f)
+		}
+	}
+}

services/nvpair-workload-manager/dedup.go

@@ -47,14 +47,38 @@ func newDedupIndex(capacity int) *dedupIndex {
 // first sighting it records the key and returns false. Either way the key is
 // promoted to most-recently-seen.
 func (d *dedupIndex) seenOrAdd(key string) bool {
+	if d.seen(key) {
+		d.mu.Lock()
+		d.ll.MoveToFront(d.items[key])
+		d.mu.Unlock()
+		return true
+	}
+	d.add(key)
+	return false
+}
+
+// seen reports whether the key is already recorded, without recording it.
+// Split from add so a caller can record the key only after the work the key
+// guards has actually succeeded (e.g. the inter-node server records a peer
+// event's dedup key only once the broker emit succeeded, so a failed emit's
+// retry isn't mistaken for a duplicate).
+func (d *dedupIndex) seen(key string) bool {
 	d.mu.Lock()
 	defer d.mu.Unlock()
+	_, ok := d.items[key]
+	return ok
+}
 
+// add records the key, promoting it to most-recently-seen and evicting the
+// least-recently-seen key past capacity. Recording an already-present key is a
+// no-op recency promotion.
+func (d *dedupIndex) add(key string) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
 	if el, ok := d.items[key]; ok {
 		d.ll.MoveToFront(el)
-		return true
+		return
 	}
-
 	el := d.ll.PushFront(key)
 	d.items[key] = el
 	if d.ll.Len() > d.capacity {
@@ -64,7 +88,6 @@ func (d *dedupIndex) seenOrAdd(key string) bool {
 			delete(d.items, oldest.Value.(string))
 		}
 	}
-	return false
 }
 
 // keyLifecycle builds the dedup key for a lifecycle event. Workload.id is only

services/nvpair-workload-manager/dedup_after_emit_test.go

@@ -0,0 +1,104 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+)
+
+// TestInterNodeDedupRecordedOnlyAfterSuccessfulEmit: the old code recorded the
+// dedup key before the broker emit, so a failed emit (500) was followed by the
+// peer's retry being swallowed as a duplicate — the event was lost with no
+// anti-entropy repair. Now the first (failed) attempt answers 500 without
+// recording the key, and the retry is emitted to the broker exactly once.
+func TestInterNodeDedupRecordedOnlyAfterSuccessfulEmit(t *testing.T) {
+	self, peer := newPinnedPeerMeshes(t)
+	dedup := newDedupIndex(100)
+
+	var emitted int
+	failEmit := true
+	srv := NewServer(0, dedup, self,
+		func(w *Workload) error {
+			if failEmit {
+				return fmt.Errorf("broker gone")
+			}
+			emitted++
+			return nil
+		},
+		func(workloadID, nodeID string) error { return nil },
+	)
+	post := serveEventsOverMTLS(t, srv, self, peer)
+
+	frame := []byte(`{"jsonrpc":"2.0","method":"workload:started","params":` +
+		`{"workloadInfo":{"id":"7","model":"llama3","engine":"ollama",` +
+		`"runId":"r1","state":"running","originatedFrom":"uuid-peer"}}}`)
+
+	if code := post(frame); code != http.StatusInternalServerError {
+		t.Fatalf("failed emit status = %d, want 500", code)
+	}
+	if emitted != 0 {
+		t.Fatalf("emitted = %d after failed emit, want 0", emitted)
+	}
+
+	// The retry must NOT be treated as a duplicate: it reaches the broker.
+	failEmit = false
+	if code := post(frame); code != http.StatusOK {
+		t.Fatalf("retry status = %d, want 200", code)
+	}
+	if emitted != 1 {
+		t.Fatalf("emitted = %d after retry, want 1", emitted)
+	}
+
+	// And a genuine duplicate afterwards is still deduplicated.
+	if code := post(frame); code != http.StatusOK {
+		t.Fatalf("duplicate status = %d, want 200", code)
+	}
+	if emitted != 1 {
+		t.Fatalf("emitted = %d after duplicate, want still 1", emitted)
+	}
+}
+
+// TestInterNodeRemoveDedupRecordedOnlyAfterSuccessfulEmit: the removal path
+// had the same record-before-emit flaw — a failed workloads:remove emit would
+// swallow the retry and leave the ghost workload in the broker catalog.
+func TestInterNodeRemoveDedupRecordedOnlyAfterSuccessfulEmit(t *testing.T) {
+	self, peer := newPinnedPeerMeshes(t)
+	dedup := newDedupIndex(100)
+
+	var emitted int
+	failEmit := true
+	srv := NewServer(0, dedup, self,
+		func(w *Workload) error { return nil },
+		func(workloadID, nodeID string) error {
+			if failEmit {
+				return fmt.Errorf("broker gone")
+			}
+			emitted++
+			return nil
+		},
+	)
+	post := serveEventsOverMTLS(t, srv, self, peer)
+
+	frame := []byte(`{"jsonrpc":"2.0","method":"workloads:remove",` +
+		`"params":{"workloadId":"7","originatedFrom":"uuid-peer"}}`)
+
+	if code := post(frame); code != http.StatusInternalServerError {
+		t.Fatalf("failed remove status = %d, want 500", code)
+	}
+	failEmit = false
+	if code := post(frame); code != http.StatusOK {
+		t.Fatalf("remove retry status = %d, want 200", code)
+	}
+	if emitted != 1 {
+		t.Fatalf("remove emitted = %d after retry, want 1", emitted)
+	}
+	if code := post(frame); code != http.StatusOK {
+		t.Fatalf("remove duplicate status = %d, want 200", code)
+	}
+	if emitted != 1 {
+		t.Fatalf("remove emitted = %d after duplicate, want still 1", emitted)
+	}
+}