diff --git a/README.md b/README.md
index 0f0a7242..09127950 100644
--- a/README.md
+++ b/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:
diff --git a/docs/workload-broadcast-reliability.mdx b/docs/workload-broadcast-reliability.mdx
new file mode 100644
index 00000000..3826012e
--- /dev/null
+++ b/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
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;
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.
diff --git a/services/nvpair-workload-manager/broadcast_order_test.go b/services/nvpair-workload-manager/broadcast_order_test.go
new file mode 100644
index 00000000..d4f02413
--- /dev/null
+++ b/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)
+ }
+ }
+}
diff --git a/services/nvpair-workload-manager/dedup.go b/services/nvpair-workload-manager/dedup.go
index c265a622..eb1b0f2c 100644
--- a/services/nvpair-workload-manager/dedup.go
+++ b/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
diff --git a/services/nvpair-workload-manager/dedup_after_emit_test.go b/services/nvpair-workload-manager/dedup_after_emit_test.go
new file mode 100644
index 00000000..db3371fe
--- /dev/null
+++ b/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)
+ }
+}
diff --git a/services/nvpair-workload-manager/manager.go b/services/nvpair-workload-manager/manager.go
index d9895e56..61793b3b 100644
--- a/services/nvpair-workload-manager/manager.go
+++ b/services/nvpair-workload-manager/manager.go
@@ -92,6 +92,14 @@ type Manager struct {
activeMu sync.Mutex
activeLocal map[workloadKey]workloadEvent
+ // broadcastCh serializes outbound inter-node frames in the order the
+ // read loop produced them. broadcastFrame only enqueues (never blocks on
+ // network I/O), and a single worker drains the queue in order — so a
+ // remove can never overtake the lifecycle upsert it follows. Without
+ // this, each frame fanned out in its own goroutine and a late upsert
+ // could resurrect a workload on peers that had already removed it.
+ broadcastCh chan []byte
+
ctx context.Context
cancel context.CancelFunc
}
@@ -124,6 +132,7 @@ func NewManager(codec *Codec, port int, selfUUID, clusterDir string) *Manager {
peerSource: relaySource,
relaySource: relaySource,
activeLocal: make(map[workloadKey]workloadEvent),
+ broadcastCh: make(chan []byte, broadcastQueueDepth),
}
m.server = NewServer(port, dedup, mesh, m.emitUpsert, m.emitRemove)
return m
@@ -157,6 +166,10 @@ func (m *Manager) Run(ctx context.Context) error {
go m.discoveryLoop(ctx)
go m.resyncLoop(ctx)
+ // The single ordered broadcast consumer: frames go out in the order the
+ // read loop produced them, so a remove can never overtake the lifecycle
+ // event it follows.
+ go m.broadcastLoop(ctx)
// Follow this node into and out of a cluster. Every gate already reads live
// membership, so the watch exists to notice a change with no traffic flowing
// and to re-assert our workloads immediately: peers that could not receive
@@ -351,19 +364,54 @@ func (m *Manager) handleLocalRemove(msg *Message) {
m.broadcastFrame(msg.Method, msg.Params)
}
-// broadcastFrame re-marshals a single notification and fans it out to every peer
-// asynchronously, so a slow peer never blocks the read loop. Delivery is
-// immediate and per-event (no batching/conflation): the origin's own view
-// already updated synchronously in the broker, and peers must see each
-// transition promptly and individually — a batching window would add latency and
-// drop intermediate states, skewing each node's independent scheduling view.
+// broadcastQueueDepth bounds how many outbound frames can wait for the
+// ordered broadcast worker. Frames are small JSON notifications; 1024 is far
+// beyond steady-state volume and only binds memory during a peer outage, when
+// overflow frames are dropped (with a warning) rather than wedging the read
+// loop — the heartbeat and peer backfill re-sync state, so a dropped frame
+// degrades to delayed convergence.
+const broadcastQueueDepth = 1024
+
+// broadcastFrame re-marshals a single notification and enqueues it for the
+// ordered broadcast worker. Delivery is still asynchronous — a slow peer never
+// blocks the read loop — but frames now go out in the order they were produced
+// (no batching/conflation): the origin's own view already updated
+// synchronously in the broker, and peers must see each transition promptly and
+// individually — a batching window would add latency and drop intermediate
+// states, skewing each node's independent scheduling view.
func (m *Manager) broadcastFrame(method string, params json.RawMessage) {
frame, err := json.Marshal(&Message{JSONRPC: "2.0", Method: method, Params: params})
if err != nil {
slog.Error("failed to marshal broadcast frame", "method", method, "err", err)
return
}
- go m.broadcaster.Broadcast(m.ctx, frame)
+ if m.broadcastCh == nil {
+ // Only reachable by hand-built Managers (tests); NewManager always
+ // installs the queue.
+ slog.Warn("broadcast queue not initialized, dropping frame", "method", method)
+ return
+ }
+ select {
+ case m.broadcastCh <- frame:
+ default:
+ slog.Warn("broadcast queue full, dropping frame", "method", method)
+ }
+}
+
+// broadcastLoop is the single ordered consumer of broadcastCh. One worker (not
+// a goroutine per frame) is what guarantees a remove never overtakes the
+// lifecycle event it follows. Broadcast aborts in-flight attempts on ctx
+// cancellation, so at shutdown the loop just exits; frames still queued are
+// dropped with the process.
+func (m *Manager) broadcastLoop(ctx context.Context) {
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case frame := <-m.broadcastCh:
+ m.broadcaster.Broadcast(ctx, frame)
+ }
+ }
}
// trackActive records the latest event for a local-origin workload. A
diff --git a/services/nvpair-workload-manager/server.go b/services/nvpair-workload-manager/server.go
index 742b5607..ef1c0b9e 100644
--- a/services/nvpair-workload-manager/server.go
+++ b/services/nvpair-workload-manager/server.go
@@ -170,10 +170,22 @@ func (s *Server) handleLifecycle(w http.ResponseWriter, msg *Message) {
// discovery backfill), which intentionally re-asserts the same key and must
// reach the broker so its store can reconcile (e.g. un-stick a wrongly
// inferred failed). The store is idempotent, so bypassing dedup here is safe.
- if !isResyncFrame(msg.Params) && s.dedup.seenOrAdd(keyLifecycle(wl)) {
- slog.Debug("inter-node lifecycle deduplicated", "method", msg.Method, "id", wl.ID, "state", wl.State)
- s.ok(w)
- return
+ //
+ // The key is recorded only after the broker emit succeeds: a failed emit
+ // answers 500 so the peer's retry budget kicks in, and that retry must not
+ // be mistaken for a duplicate (the old code recorded the key first, so a
+ // failed emit permanently dropped the event with no anti-entropy repair).
+ // A concurrent duplicate that passes seen() before either emit runs just
+ // emits twice, which the idempotent store absorbs.
+ resync := isResyncFrame(msg.Params)
+ lifecycleKey := ""
+ if !resync {
+ lifecycleKey = keyLifecycle(wl)
+ if s.dedup.seen(lifecycleKey) {
+ slog.Debug("inter-node lifecycle deduplicated", "method", msg.Method, "id", wl.ID, "state", wl.State)
+ s.ok(w)
+ return
+ }
}
if err := s.emitUpsert(wl); err != nil {
@@ -184,6 +196,9 @@ func (s *Server) handleLifecycle(w http.ResponseWriter, msg *Message) {
http.Error(w, "broker unavailable", http.StatusInternalServerError)
return
}
+ if !resync {
+ s.dedup.add(lifecycleKey)
+ }
slog.Info("relayed remote lifecycle as upsert", "method", msg.Method, "id", wl.ID, "state", wl.State, "node", wl.OriginatedFrom)
s.ok(w)
}
@@ -204,7 +219,11 @@ func (s *Server) handleRemove(w http.ResponseWriter, msg *Message) {
return
}
- if s.dedup.seenOrAdd(keyRemove(nodeID, workloadID)) {
+ // The removal key, like the lifecycle key, is recorded only after the
+ // broker emit succeeds, so a failed emit's retry isn't swallowed as a
+ // duplicate.
+ removeKey := keyRemove(nodeID, workloadID)
+ if s.dedup.seen(removeKey) {
slog.Debug("inter-node removal deduplicated", "workloadId", workloadID, "node", nodeID)
s.ok(w)
return
@@ -215,6 +234,7 @@ func (s *Server) handleRemove(w http.ResponseWriter, msg *Message) {
http.Error(w, "broker unavailable", http.StatusInternalServerError)
return
}
+ s.dedup.add(removeKey)
slog.Info("relayed remote removal", "workloadId", workloadID, "node", nodeID)
s.ok(w)
}
diff --git a/services/versions.json b/services/versions.json
index 29d8c230..f64443ed 100644
--- a/services/versions.json
+++ b/services/versions.json
@@ -8,7 +8,7 @@
"nvpair-node-info": "0.13.3",
"nvpair-node-scanner": "0.20.3",
"nvpair-manual-nodes": "0.11.1",
- "nvpair-workload-manager": "0.13.3",
+ "nvpair-workload-manager": "0.13.4",
"nvpair-errors": "0.7.4",
"nvpair-node-settings": "1.0.4",
"nvpair-ui-broker": "0.40.2",