Fan the agent command drain out per destination so a pass places work on every node (GH-3698)

[jeremydmiller]

Fixes the first half of #3698. Single commit, deliberately narrow — see "Scope" at the bottom for what was split out and why.

The reporter's own description of what they need:

1. let a pass place work on all eligible nodes instead of filling one and stopping
2. do not gate the next pass on the previous batch fully starting, or bound that wait

This PR is (1). (2) turned out to need a design change and is being developed separately in draft #3719.

Reproduces on 6.24.0 (the issue's original "6.23.0" was a misread tag out of a persisted pause_reason).

The defect

executeAgentCommandsAsync executed one command at a time across the whole cluster. Because
NodeAgentController.batchCommands groups a wave's AssignAgent commands by destination and appends each
destination's chunks contiguously, the drain was destination-at-a-time: every chunk for node A was awaited
to completion before node B was sent its first.

Cluster-wide agent start concurrency was therefore capped at a single node's MaxAgentStartParallelism
no matter how many nodes were in the cluster — adding nodes did nothing for convergence time. This is
also why the reporter's AgentStartBatchSize 50 → 1500 change helped without changing the shape: bigger
chunks mean fewer control-queue round-trips per node, but the starts inside them were funnelled through one
node either way.

WO-4 of the GH-3604 work chunked the batches and parallelised starts within a chunk; it never parallelised
across destinations.

Reproduction

src/Testing/SlowTests/Agents/agent_assignment_at_scale.cs — three Postgres-backed nodes, 900 agents whose
StartAsync takes real wall-clock time (standing in for a Marten daemon shard spin-up under a
GateSideEffectsBehindPriorVersion replay), shipped defaults for AgentStartBatchSize (50) and
MaxAgentStartParallelism (10). It reproduced every reported symptom on the first run:

elapsed | assigned per node        | total | AssignmentChanged
     22s | 1:   0  2:   0  3:   0  |     0 |   901      <- one burst of telemetry
     57s | 1:   0  2: 300  3:   0  |   300 |   901      <- nodes 1 and 3 still at ZERO
     94s | 1:   0  2: 300  3: 300  |   600 |   901      <- node 1 still 0, 72s in
    124s | 1: 300  2: 300  3: 300  |   900 |   901

Converged at:                  125.5s
Peak concurrent agent starts:  10  (cluster-wide; 3 nodes x 10 parallelism = 30 available)

The change

IAgentCommand gains a Guid? DestinationNodeId default interface member (null = "no single destination"),
so existing and third-party commands stay source-compatible. The new AgentCommandDrain partitions each
round into one lane per destination: serial within a lane (one chunk in flight per node, which is what
the WO-5 pending-assignment ledger assumes), concurrent across lanes. Commands naming no single
destination share one strictly-serial lane, exactly as before.

Cascades are deferred to the next round rather than appended to the producing lane, so "a cascade runs after
its producer" still holds while being re-keyed onto the right lane.

A failing command no longer discards the rest of the wave — previously any exception, most realistically a
TimeoutException from an AssignAgents chunk that outran its reply window, unwound out of the drain and
dropped every command still queued for every other node. Failures are returned rather than thrown so
ApplyRestrictionsAsync, which answers to an operator, still surfaces them as an AggregateException once
the whole wave has run instead of silently reporting success.

On ordering across lanes: a stop and a start for the same agent on the same node land in the same lane and
so stay ordered, and the only cross-node pairing the leader emits for one agent — a split-brain
StopRemoteAgent against the older copy alongside a reassignment away from the newer one — targets two
different nodes with two idempotent stops, and defers its start to the next round.

Result

	before	after
Peak cluster-wide concurrent starts	10 of 30	30 of 30
Converged	125.5s	57.3 / 58.2 / 58.2s
Node ramp	one node at a time, others at 0	all three in lockstep

Three runs, each against a freshly recreated Postgres container.

Tests

• SlowTests/Agents/agent_assignment_at_scale.cs — end-to-end repro, red-baselined against origin/main at
  125.5s. SlowTests is not wired into any Nuke target, so this is a local investigation tool rather than a CI
  gate.
• CoreTests/Runtime/Agents/per_destination_command_drain.cs — the drain's contract, for CI. The two
  behavioural tests were red-checked by temporarily forcing the drain back to
  MaxDegreeOfParallelism = 1 and removing the per-command catch.

CoreTests 2107/2107. Full wolverine.slnx Release build clean.

RavenDbTests.LeaderElection 28 of 28 on this commit, against 1 of 28 failing on clean origin/main
(the known spin_up_several_nodes_take_away_original_node flake) — so this commit is if anything cleaner
than main there.

PersistenceTests.Agents.durability_modes.start_in_balanced_mode fails, and reproduces on clean
origin/main with these changes stashed, so it is pre-existing.

Scope — what was split out

This started as a larger branch that also decoupled the assignment evaluation from the drain (the reporter's
ask 2). Bisecting the RavenDb leadership-election suite showed that half regresses 7 tests:

clean origin/main       1 of 28 failures (known flake)
this commit             28 of 28 PASS
+ decoupling commit     7 failures

The failure dump shows the database holding a clean 3/3/3/3 while the runtimes disagree — two agents each
running on two nodes at once. Once the evaluation stops waiting for the drain, a dispatched-but-unconfirmed
agent looks entirely unassigned (Agent.OriginalNode comes from persisted ActiveAgents, and the assignment
row only appears once the agent is running), so a later pass emits a plain AssignAgent to a different node
with no stop for the copy already starting elsewhere.

Fixing that needs pending assignment to become a first-class grid state rather than a TTL'd suppression list,
which is a bigger change than belongs in this PR. It continues in draft
#3719, which should be rebased once this lands.

Also found, not in this PR

• NodeStarted records carry a random node_number. DurabilitySettings.AssignedNodeNumber defaults to
  Guid.NewGuid().ToString().GetDeterministicHashCode(), and StartLocalAgentProcessingAsync calls
  _observer.NodeStarted() one line before assigning the real number. This is the reporter's 1827159984
  / -1841074440 values.
• DormantNodeEjected fires for nodes that were never ejected. ejectStaleNodes continues past the
  delete for nodes the Exclusive listener agent does not reassign reliably to a survivor after its node leaves the cluster #3604 hysteresis deliberately spares, then calls _observer.StaleNodes() on the
  whole stale list.
• AssignmentsChanged is observed before batchCommands, so telemetry writes one row per agent while
  the wire carries one message per chunk — 901 rows for 19 chunks in the repro, ~5,366 rows per pass in the
  reported cluster.
• A chunk that outruns its 30s + chunkSize reply window no longer discards the wave, but still burns the
  full timeout in its lane. Arming the pending-assignment ledger on dispatch rather than emission would help.

🤖 Generated with Claude Code