The loop is triggered by {@link #onReplyQueued(String)} (called from - * {@link MessageService#reply} after the durable inbox publish). It checks four conditions - * at each tick via {@link #decide(String, int)}, then either injects a drain nudge, - * waits for the primary to become injectable, or stops reminding. + *
CB-590: one schedule per lead. Both kinds of work are triggered through + * their own entry point — {@link #onReplyQueued(String)} and + * {@link #onTicketTerminal(String, String, boolean)} — but both resolve the lead that should be + * nudged and coalesce onto a single per-lead reminder schedule, tracked in {@link #activeLeads}. + * Earlier this was two independent schedules (one keyed by worker target for replies, one keyed + * by lead for tickets) that could both decide to inject into the same pane in the same window — + * a race, not routine behaviour, but the expensive kind: it interrupts the lead's live turn + * twice. Collapsing to one schedule per lead makes that structurally impossible: at most one + * scheduled tick chain is ever live for a given lead (guarded by {@link #activeLeads}' + * compare-and-set), so at most one {@code agents.send} to that lead's pane is ever in flight. * - *
Bounded: at most {@link #maxReminders} nudges per target, with a configurable backoff - * between them. The reply is never lost — the durable inbox is the backstop. - * - *
CB-588 ticket nudges. {@link #onTicketTerminal(String, String, boolean)} is a - * second, independent entry point for an async delegation ticket ({@code bridge_send(wait:false)}) - * reaching a terminal phase. That path takes the rendezvous fast path in {@link MessageService#reply} - * and never reaches {@link #onReplyQueued}, so without this the lead's own charter — prefer - * {@code wait:false} for anything non-trivial — was exactly the mode this loop failed to cover. It - * reuses the same status gating, bounded/backoff reminders, and metrics, keyed by the nudge-receiving - * lead terminal rather than the worker target so several tickets finishing together coalesce into one - * nudge. The two entry points do not interact: {@link #onReplyQueued} / {@link #decide} / their nudge - * text and bound are unchanged. + *
Each tick examines everything pending for that lead — reply targets whose inbox
+ * still holds an unacked message ({@link #pendingReplies}) and tickets not yet collected
+ * ({@link #pendingTickets}) — and sends at most one combined nudge per tick
+ * ({@link #injectNudge(String, int)}). Work that arrives while the lead is busy is never lost:
+ * it is re-read fresh on every tick until the lead is injectable or the shared reminder cap
+ * ({@link #maxReminders}) is reached, whichever the durable inbox / pending-ticket set doesn't
+ * already answer via {@code STOP}.
*/
public final class ReplyPushLoop {
private static final Logger log = LoggerFactory.getLogger(ReplyPushLoop.class);
static final String NUDGE_FORMAT = "Worker %s returned a reply — run bridge_poll(target=%s) to collect it";
- /** CB-588: singular form, one uncollected ticket. */
+ /** Coalesced form, several uncollected replies for the same lead. */
+ static final String REPLIES_NUDGE_FORMAT =
+ "%d workers returned replies — run bridge_poll(target=...) for each to collect them: %s";
+ /** Singular form, one uncollected ticket. */
static final String TICKET_NUDGE_FORMAT =
"Ticket %s finished%s — run bridge_poll(ticket=%s) to collect it";
- /** CB-588: coalesced form, several uncollected tickets for the same lead. */
+ /** Coalesced form, several uncollected tickets for the same lead. */
static final String TICKETS_NUDGE_FORMAT =
"%d tickets finished%s — run bridge_poll(ticket=...) for each to collect them: %s";
@@ -56,11 +64,11 @@ public final class ReplyPushLoop {
private final long backoffMs;
private final Metrics metrics; // CB-512: nullable — no registry in unit tests
- /** Track targets that have an active schedule. */
- private final ConcurrentHashMap Idempotent per lead: several tickets going terminal for the same lead while its schedule is
- * already active coalesce onto that schedule's next tick rather than firing a nudge each.
- *
- * @param ticket the ticket to nudge about
- * @param target the worker session the ticket was sent to — resolves which lead delegated it
- * @param failed whether the ticket ended in a failure phase rather than {@code DONE}
- */
- public void onTicketTerminal(String ticket, String target, boolean failed) {
- var lead = primaryRegistry.nudgeTargetFor(target);
- if (lead.isEmpty()) {
- log.debug("push: no lead is known to be waiting on ticket {} (target {}), skipping nudge",
- ticket, target);
- return;
- }
- pendingTickets.put(ticket, new PendingTicket(ticket, lead.get(), failed));
- if (activeLeads.putIfAbsent(lead.get(), Boolean.TRUE) != null) {
- log.debug("push: ticket reminder loop already active for lead {}, {} coalesced in",
- lead.get(), ticket);
- return;
- }
- log.debug("push: starting ticket reminder loop for lead {}", lead.get());
- scheduleTicketTick(lead.get(), 0);
- }
-
- /**
- * Called when a ticket's terminal state has been collected via {@code bridge_poll}. Removes it
- * from the pending set so a scheduled tick — and any nudge it sends — never names a ticket the
- * lead already has (CB-588 acceptance #5). A ticket that was never pending (unknown ticket, or
- * one nudged with no push loop configured) is a no-op.
- */
- public void ticketCollected(String ticket) {
- pendingTickets.remove(ticket);
- }
-
/** Tickets still pending for {@code lead}, snapshotted fresh for one tick. */
- private List Restarting on ANY non-empty {@code pendingFor(lead)} would be wrong: when STOP is reached
- * because the reminder cap was hit rather than the backlog draining, the same never-collected
- * ticket is expected to still be sitting there — that is the cap doing its job — and restarting
- * would nudge about it forever, defeating the bound (the original CB-307 bounded-reminder
- * guarantee, carried into CB-588 by acceptance criterion #7 — this exact regression showed up as
- * two existing tests failing once a naive "any pending ticket restarts" version of this fix went
- * in: {@code successfulTicketNudgeIncrementsDelivered} and {@code ticketNudgesSendUpToCapThenStop}).
- * Diffing the current pending set against {@code pendingBefore} tells the two cases apart: a
- * ticket present before this tick's decision is stale backlog, not a race; only a ticket absent
- * from {@code pendingBefore} can only have arrived during the decision-to-release window, which is
- * exactly the race this method closes.
+ * Resolves the delegating lead the same way {@link #onReplyQueued} does and coalesces onto
+ * the same per-lead schedule (CB-590) — several tickets, or a ticket and a reply, finishing
+ * for the same lead while its schedule is already active all ride the existing schedule's next
+ * tick rather than firing a nudge each.
+ *
+ * @param ticket the ticket to nudge about
+ * @param target the worker session the ticket was sent to — resolves which lead delegated it
+ * @param failed whether the ticket ended in a failure phase rather than {@code DONE}
+ */
+ public void onTicketTerminal(String ticket, String target, boolean failed) {
+ var lead = primaryRegistry.nudgeTargetFor(target);
+ if (lead.isEmpty()) {
+ log.debug("push: no lead is known to be waiting on ticket {} (target {}), skipping nudge",
+ ticket, target);
+ return;
+ }
+ pendingTickets.put(ticket, new PendingTicket(ticket, lead.get(), failed));
+ startOrCoalesce(lead.get());
+ }
+
+ /**
+ * Called when a ticket's terminal state has been collected via {@code bridge_poll}. Removes it
+ * from the pending set so a scheduled tick — and any nudge it sends — never names a ticket the
+ * lead already has. A ticket that was never pending (unknown ticket, or one nudged with no push
+ * loop configured) is a no-op.
+ */
+ public void ticketCollected(String ticket) {
+ pendingTickets.remove(ticket);
+ }
+
+ // --- the schedule ----------------------------------------------------------------------------
+
+ /** Start a reminder schedule for {@code lead}, or join the one already running. */
+ private void startOrCoalesce(String lead) {
+ if (activeLeads.putIfAbsent(lead, Boolean.TRUE) != null) {
+ log.debug("push: reminder loop already active for lead {}, work coalesced in", lead);
+ return;
+ }
+ log.debug("push: starting reminder loop for lead {}", lead);
+ scheduleNext(lead, 0);
+ }
+
+ /** Execute one loop tick — called on the scheduler thread. */
+ private void tick(String lead, int reminderCount) {
+ Set Restarting on ANY non-empty pending set would be wrong: when STOP is reached because the
+ * reminder cap was hit rather than the backlog draining, the same never-collected work is
+ * expected to still be sitting there — that is the cap doing its job — and restarting would
+ * nudge about it forever, defeating the bound. Diffing the current pending sets against the
+ * "before" snapshots tells the two cases apart: an item present before this tick's decision is
+ * stale backlog, not a race; only an item absent from the "before" snapshot can only have
+ * arrived during the decision-to-release window, which is exactly the race this method closes.
*
* Package-private so a test can drive the interleaving directly rather than trying to force a
- * genuine thread race: pass the exact {@code pendingBefore} snapshot a race requires (or does
- * not) and call this to prove the recheck responds correctly either way.
+ * genuine thread race.
*
- * Terminates rather than spinning: this method restarts the schedule at most once per call, and
- * a fresh {@link #onTicketTerminal} racing the recheck below still terminates in one of two ways —
- * either it observes the slot already vacated (by the {@code activeLeads.remove} above, which
- * happens-before this recheck in program order) and claims it itself, or it lands first and this
- * recheck then observes its ticket in {@code pendingTickets} and reclaims the slot instead. Exactly
- * one side always wins; neither can miss the other, so this never loops on its own account.
+ * Terminates rather than spinning: this method restarts the schedule at most once per call,
+ * and a fresh {@link #onReplyQueued} / {@link #onTicketTerminal} racing the recheck below still
+ * terminates in one of two ways — either it observes the slot already vacated (by the
+ * {@code activeLeads.remove} above, which happens-before this recheck in program order) and
+ * claims it itself, or it lands first and this recheck then observes its work in
+ * {@link #pendingReplies} / {@link #pendingTickets} and reclaims the slot instead. Exactly one
+ * side always wins; neither can miss the other, so this never loops on its own account.
*/
- void stopOrRestartTicketLoop(String lead, Set CB-590 collapsed the CB-307 reply-nudge schedule and the CB-588 ticket-nudge schedule into
+ * one schedule per lead ({@link ReplyPushLoop#decide}), so most tests below register pending work
+ * through the public entry points ({@code onReplyQueued} / {@code onTicketTerminal}) before
+ * exercising {@code decide} directly, mirroring how the two entry points now share one decision
+ * function keyed by the lead terminal rather than by worker target.
+ *
* Uses a {@link RecordingHerdrClient} that synchronizes access to its call list so the
* scheduler thread and test thread never have memory ordering issues. The {@code decide()}
* tests use a simple client with no concurrency concern.
@@ -58,38 +65,50 @@ class ReplyPushLoopTest {
// --- decide() logic ------------------------------------------------------------------------
@Test
- void decideWithoutPrimaryIsStop() {
- agents = agentWithStatus("idle");
- var loop = new ReplyPushLoop(
- new PrimaryRegistry(null), agents, inbox, scheduler, 5, 100);
- assertEquals(ReplyPushLoop.Action.STOP, loop.decide(WORKER, 0));
+ void onReplyQueuedWithNoKnownLeadNeverStartsASchedule() throws Exception {
+ var rec = recordingClient();
+ agents = new AgentControl(rec);
+ inbox.publish(WORKER, "m1", "hello");
+ var loop = new ReplyPushLoop(new PrimaryRegistry(null), agents, inbox, scheduler, 5, 50);
+
+ loop.onReplyQueued(WORKER); // no lead known -> never registered, never scheduled
+
+ assertFalse(loop.isActive(), "no lead known means nothing to nudge yet");
+ Thread.sleep(150);
+ assertEquals(0, rec.sendCount(), "must not nudge when no lead is known to be waiting");
}
@Test
- void decideWithEmptyInboxIsStop() {
+ void decideWithNothingPendingIsStop() {
agents = agentWithStatus("idle");
- assertEquals(ReplyPushLoop.Action.STOP, loop().decide(WORKER, 0));
+ assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0));
}
@Test
void decideAtCapIsStop() {
agents = agentWithStatus("idle");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.STOP, loop(2, 100).decide(WORKER, 2));
+ var loop = loop(2, 100);
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
}
@Test
void decideUnderCapWithInjectablePrimaryIsInject() {
agents = agentWithStatus("idle");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.INJECT, loop().decide(WORKER, 0));
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
}
@Test
void decideUnderCapWithBlockedPrimaryIsInject() {
agents = agentWithStatus("blocked");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.INJECT, loop().decide(WORKER, 0),
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0),
"BLOCKED is injectable");
}
@@ -97,7 +116,9 @@ class ReplyPushLoopTest {
void decideUnderCapWithDonePrimaryIsInject() {
agents = agentWithStatus("done");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.INJECT, loop().decide(WORKER, 0),
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0),
"DONE is injectable");
}
@@ -105,23 +126,29 @@ class ReplyPushLoopTest {
void decideUnderCapWithBusyPrimaryIsWaitBusy() {
agents = agentWithStatus("working");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop().decide(WORKER, 0));
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
}
@Test
void decideUnderCapWithUnknownPrimaryIsWaitBusy() {
agents = agentWithStatus("unknown");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop().decide(WORKER, 0));
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
}
@Test
void decideStopsAfterInboxIsEmptied() {
agents = agentWithStatus("idle");
inbox.publish(WORKER, "m1", "hello");
- assertEquals(ReplyPushLoop.Action.INJECT, loop().decide(WORKER, 0));
+ var loop = loop();
+ loop.onReplyQueued(WORKER);
+ assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
inbox.ack(WORKER, "m1");
- assertEquals(ReplyPushLoop.Action.STOP, loop().decide(WORKER, 0));
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0));
}
// --- onReplyQueued integration -------------------------------------------------------------
@@ -201,12 +228,19 @@ class ReplyPushLoopTest {
assertTrue(nudge.contains("bridge_poll(target=term_worker)"));
}
- // --- CB-588: async ticket terminal nudges — decideTickets() logic --------------------------
+ @Test
+ void repliesNudgeFormatIsCorrect() {
+ String multi = ReplyPushLoop.REPLIES_NUDGE_FORMAT.formatted(2, "term_worker1, term_worker2");
+ assertTrue(multi.contains("2 workers"));
+ assertTrue(multi.contains("bridge_poll(target=...)"));
+ }
+
+ // --- CB-588: async ticket terminal nudges — decide() logic on tickets -----------------------
@Test
void decideTicketsWithNothingPendingIsStop() {
agents = agentWithStatus("idle");
- assertEquals(ReplyPushLoop.Action.STOP, loop().decideTickets(PRIMARY, 0));
+ assertEquals(ReplyPushLoop.Action.STOP, loop().decide(PRIMARY, 0));
}
@Test
@@ -214,7 +248,7 @@ class ReplyPushLoopTest {
agents = agentWithStatus("idle");
var loop = loop(2, 100_000);
loop.onTicketTerminal("task-1", WORKER, false);
- assertEquals(ReplyPushLoop.Action.STOP, loop.decideTickets(PRIMARY, 2));
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
}
@Test
@@ -222,7 +256,7 @@ class ReplyPushLoopTest {
agents = agentWithStatus("idle");
var loop = loop(5, 100_000);
loop.onTicketTerminal("task-1", WORKER, false);
- assertEquals(ReplyPushLoop.Action.INJECT, loop.decideTickets(PRIMARY, 0));
+ assertEquals(ReplyPushLoop.Action.INJECT, loop.decide(PRIMARY, 0));
}
@Test
@@ -230,7 +264,7 @@ class ReplyPushLoopTest {
agents = agentWithStatus("working");
var loop = loop(5, 100_000);
loop.onTicketTerminal("task-1", WORKER, false);
- assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decideTickets(PRIMARY, 0));
+ assertEquals(ReplyPushLoop.Action.WAIT_BUSY, loop.decide(PRIMARY, 0));
}
@Test
@@ -238,7 +272,7 @@ class ReplyPushLoopTest {
agents = agentWithStatus("idle");
var loop = new ReplyPushLoop(new PrimaryRegistry(null), agents, inbox, scheduler, 5, 100_000);
loop.onTicketTerminal("task-1", WORKER, false); // no lead known -> never registered as pending
- assertEquals(ReplyPushLoop.Action.STOP, loop.decideTickets(PRIMARY, 0));
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 0));
}
// --- CB-588: onTicketTerminal integration ---------------------------------------------------
@@ -255,7 +289,7 @@ class ReplyPushLoopTest {
String nudge = rec.sentParams().getFirst().getValue().toString();
assertTrue(nudge.contains("task-1"), "nudge should name the ticket");
assertTrue(nudge.contains("bridge_poll(ticket="), "nudge should name the exact ticket-poll call");
- assertFalse(nudge.contains("bridge_poll(target="), "a ticket nudge must not tell the lead to run the target-poll call");
+ assertFalse(nudge.contains("bridge_poll(target="), "a ticket-only nudge must not tell the lead to run the target-poll call");
}
@Test
@@ -346,11 +380,11 @@ class ReplyPushLoopTest {
void aTicketStillPendingWhenTheLoopStopsIsNotStranded() {
// Regression for the race a reviewer found in gitea PR #73: onTicketTerminal's
// activeLeads.putIfAbsent can see the lead's slot as still occupied a moment before
- // decideTickets' STOP releases it, so the ticket coalesces onto a schedule that is about to
+ // decide's STOP releases it, so the ticket coalesces onto a schedule that is about to
// die and nothing ever nudges about it. Forcing that exact thread interleaving is not
- // reliable, so this drives stopOrRestartTicketLoop — the STOP path's own release-and-recheck —
+ // reliable, so this drives stopOrRestart — the STOP path's own release-and-recheck —
// directly, arranging the state it must not lose a ticket in: a ticket pending for the lead
- // that was NOT part of the pre-decision snapshot (pendingBefore=empty), standing in for one
+ // that was NOT part of the pre-decision snapshot (ticketsBefore=empty), standing in for one
// that races in during the decision-to-release window.
var rec = recordingClient();
agents = new AgentControl(rec);
@@ -358,9 +392,9 @@ class ReplyPushLoopTest {
loop.onTicketTerminal("task-1", WORKER, false); // pendingTickets={task-1}; activeLeads={PRIMARY}
- // Stand in for the scheduler thread reaching decideTickets==STOP for this lead — with nothing
+ // Stand in for the scheduler thread reaching decide==STOP for this lead — with nothing
// pending at decide time — while task-1 races in before the release below runs.
- loop.stopOrRestartTicketLoop(PRIMARY, Set.of());
+ loop.stopOrRestart(PRIMARY, Set.of(), Set.of());
assertTrue(loop.isActive(), "a ticket that raced the loop's stop must reclaim the schedule "
+ "slot, not be stranded with no schedule left to ever nudge about it");
@@ -368,18 +402,19 @@ class ReplyPushLoopTest {
@Test
void aStaleUncollectedTicketAtCapDoesNotRestartTheLoop() {
- // The other direction of the same fix: restarting on ANY non-empty pendingFor(lead) would be
+ // The other direction of the same fix: restarting on ANY non-empty pending set would be
// wrong. When STOP is reached because the reminder cap was hit, the same never-collected
// ticket is expected to still be there — that is the cap doing its job (acceptance criterion
- // #7: nudges stay bounded). task-1 here was already accounted for at decide time (it is in
- // pendingBefore), so it must not restart the loop just because it is still sitting there.
+ // #5: no spin / nudges stay bounded). task-1 here was already accounted for at decide time
+ // (it is in ticketsBefore), so it must not restart the loop just because it is still sitting
+ // there.
var rec = recordingClient();
agents = new AgentControl(rec);
ReplyPushLoop loop = loop(1, 100_000);
loop.onTicketTerminal("task-1", WORKER, false); // pendingTickets={task-1}; activeLeads={PRIMARY}
- loop.stopOrRestartTicketLoop(PRIMARY, Set.of("task-1"));
+ loop.stopOrRestart(PRIMARY, Set.of(), Set.of("task-1"));
assertFalse(loop.isActive(), "a stale ticket already accounted for at decide time must not "
+ "restart the loop — that would defeat the reminder cap");
@@ -402,7 +437,52 @@ class ReplyPushLoopTest {
void inboxNudgeStillUsesTheOriginalTargetPollCall() {
String nudge = ReplyPushLoop.NUDGE_FORMAT.formatted(WORKER, WORKER);
assertTrue(nudge.contains("bridge_poll(target=" + WORKER + ")"),
- "CB-588 must not change the CB-307 inbox nudge's call shape");
+ "CB-588/CB-590 must not change the CB-307 inbox nudge's call shape");
+ }
+
+ // --- CB-590: one schedule per lead — no overlap, no lost nudges -----------------------------
+
+ @Test
+ void replyAndTicketForTheSameLeadCoalesceIntoOneSendNeverOverlapping() throws Exception {
+ var rec = recordingClient();
+ agents = new AgentControl(rec);
+ inbox.publish(WORKER, "m1", "hello");
+ var loop = loop(1, 300); // backoff wide enough that both entry points land before the first tick
+
+ loop.onReplyQueued(WORKER);
+ loop.onTicketTerminal("task-1", WORKER, false);
+
+ assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS), "one combined nudge should have been sent");
+ Thread.sleep(300);
+ assertEquals(1, rec.sendCount(),
+ "a reply and a ticket for the same lead must coalesce onto ONE schedule — "
+ + "two nudge injections into the same lead pane must never overlap");
+ String nudge = rec.sentParams().getFirst().getValue().toString();
+ assertTrue(nudge.contains("bridge_poll(target=" + WORKER + ")"),
+ "the combined nudge must still mention the reply: " + nudge);
+ assertTrue(nudge.contains("task-1"), "the combined nudge must still mention the ticket: " + nudge);
+ }
+
+ @Test
+ void aReplyQueuedWhileTheLeadIsBusyIsNotLostWhenATicketArrivesToo() throws Exception {
+ // The lead is busy for its first two status checks, then becomes injectable. A reply is
+ // queued while busy; a ticket for the same lead arrives before the lead frees up. Neither
+ // may be dropped — deferred is fine, lost is not (acceptance criterion #2).
+ var rec = new BusyThenIdleHerdrClient(2);
+ agents = new AgentControl(rec);
+ inbox.publish(WORKER, "m1", "hello");
+ var loop = loop(1, 50); // cap=1: WAIT_BUSY doesn't count against it, so exactly one send once injectable
+
+ loop.onReplyQueued(WORKER); // schedule starts, first tick(s) WAIT_BUSY
+ loop.onTicketTerminal("task-1", WORKER, false); // coalesces onto the same waiting schedule
+
+ assertTrue(rec.sendLatch.await(3, TimeUnit.SECONDS),
+ "once the lead becomes injectable, the deferred work must still be nudged");
+ Thread.sleep(200);
+ assertEquals(1, rec.sendCount(), "exactly one nudge once injectable — reply and ticket coalesced");
+ String nudge = rec.sentParams().getFirst().getValue().toString();
+ assertTrue(nudge.contains("bridge_poll(target=" + WORKER + ")"), "the reply must not be dropped: " + nudge);
+ assertTrue(nudge.contains("task-1"), "the ticket must not be dropped: " + nudge);
}
// --- metrics (CB-512) ----------------------------------------------------------------------
@@ -430,8 +510,10 @@ class ReplyPushLoopTest {
agents = agentWithStatus("idle");
inbox.publish(WORKER, "m1", "hello");
Metrics metrics = new Metrics();
+ var loop = loop(2, 100, metrics);
+ loop.onReplyQueued(WORKER);
- assertEquals(ReplyPushLoop.Action.STOP, loop(2, 100, metrics).decide(WORKER, 2));
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
assertEquals(1, metrics.count(BridgedMetrics.PUSH_NUDGES, "outcome", "exhausted"),
"hitting the reminder cap must count as exhausted");
@@ -459,7 +541,7 @@ class ReplyPushLoopTest {
var loop = loop(2, 100_000, metrics);
loop.onTicketTerminal("task-1", WORKER, false);
- assertEquals(ReplyPushLoop.Action.STOP, loop.decideTickets(PRIMARY, 2));
+ assertEquals(ReplyPushLoop.Action.STOP, loop.decide(PRIMARY, 2));
assertEquals(1, metrics.count(BridgedMetrics.PUSH_NUDGES, "outcome", "exhausted"),
"hitting the ticket reminder cap must count as exhausted");
@@ -547,4 +629,49 @@ class ReplyPushLoopTest {
private static RecordingHerdrClient recordingClient() {
return new RecordingHerdrClient();
}
+
+ /**
+ * Thread-safe fake that reports {@code working} (not injectable) for its first
+ * {@code busyChecks} status calls, then {@code idle} forever after — used to prove work queued
+ * while the lead is busy is deferred, not dropped, once it becomes injectable.
+ */
+ private static final class BusyThenIdleHerdrClient implements HerdrClient {
+ private final List