fleet/fleetd
3
0
Fork 0
Code Issues 73 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fleetd #556: make turn registration structural, independent of any TurnListener #566

Merged
ltms merged 2 commits from worker/556-injector-owns-registration-e027a5-1 into main 2026-09-12 12:18:16 +02:00
Conversation 1 Commits 2 Files Changed 6
+430 -37
6 changed files with 430 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fleetd/src/main/java/dev/ltms/fleet/Fleetd.java
+4 -1
View File
@@ -528,8 +528,11 @@ public final class Fleetd {
}
};
Predicate<String> deliverable = deliverableTo(presence, leads);
// fleetd #556: registration is wired directly to `completion`, not folded into the
// `turnListener` fan-out above — so it survives `sessions.onDelivered` (or any future
// listener) throwing, regardless of call order. See TurnRegistrar's javadoc.
Injector injector = new Injector(router, turnListener, deliverable,
presence::forget);
presence::forget, completion::register);
StatusPoller poller = new StatusPoller(router, injector, Injector.POLL_INTERVAL_MILLIS);
poller.start();
fleetd/src/main/java/dev/ltms/fleet/inject/CompletionResolver.java
+29 -2
View File
@@ -50,7 +50,7 @@ import java.util.regex.Pattern;
* so the scrape's herdr round-trip never stalls the status poller. The captured waiter is read on the
* poller thread (before any next-turn delivery can overwrite it) and passed into the virtual thread.
*/
public final class CompletionResolver implements TurnListener {
public final class CompletionResolver implements TurnListener, TurnRegistrar {
private static final Logger log = LoggerFactory.getLogger(CompletionResolver.class);
@@ -235,6 +235,26 @@ public final class CompletionResolver implements TurnListener {
this.worktreeBranches = Objects.requireNonNull(worktreeBranches, "worktreeBranches");
}
/**
* fleetd #556: {@link TurnRegistrar}'s structural half of what {@link #captureBaseline} used to
* do alone — record the waiter, no I/O. Called directly and unconditionally by the {@link
* Injector} as part of delivering a turn, before {@link #onDelivered} ever runs, so this
* registration cannot be skipped by a {@link TurnListener} throwing (from {@code onDelivered} or
* any other callback). {@link #captureBaseline} still performs this exact check-and-put itself
* as well — harmless and idempotent when it runs right after this — so a caller that only wires
* the {@link TurnListener} path (e.g. an existing test that never mentions {@link TurnRegistrar})
* keeps working unchanged.
*/
@Override
public void register(String target, TurnToken token) {
CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
if (waiter == null) {
inFlight.remove(target); // no send is waiting on this delivery — nothing to resolve later
return;
}
inFlight.put(target, new InFlight(waiter, null, nowNanos.getAsLong(), token.injectedText()));
}
@Override
public void onDelivered(String target, TurnToken token) {
// Capture the exact waiter this turn belongs to (CB-116) and snapshot the pane's pre-turn
@@ -244,7 +264,14 @@ public final class CompletionResolver implements TurnListener {
captureBaseline(target, token);
}
/** Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link #onDelivered}). */
/**
* Capture the in-flight turn: its waiter and pre-turn baseline (the testable core of {@link
* #onDelivered}). fleetd #556: this still does its own full waiter check and {@code inFlight.put}
* — the same registration {@link #register} performs — so it keeps working standalone (as every
* test calling it directly already does) even though the {@link Injector} now also calls {@link
* #register} on its own, earlier and unconditionally. The two writes are idempotent with each
* other; only the second (this one) carries a real baseline, since only this one pays for a scrape.
*/
void captureBaseline(String target, TurnToken token) {
CompletableFuture<Rendezvous.Resolution> waiter = token.waiter();
if (waiter == null) {
fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java
+100 -34
View File
@@ -11,6 +11,7 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
@@ -93,6 +94,11 @@ public final class Injector {
private final AgentControl agents;
private final HerdrRouter router;
private final TurnListener turnListener;
/**
* fleetd #556: the Injector's own registration invariant, called directly and unconditionally —
* never folded into {@link #turnListener}'s fan-out. See {@link TurnRegistrar}.
*/
private final TurnRegistrar registrar;
private final Predicate<String> ready; // CB-113: a target is deliverable only when available
private final Consumer<String> forget; // CB-114: clear a gone worker's readiness/presence
/**
@@ -134,15 +140,43 @@ public final class Injector {
*/
public Injector(AgentControl agents, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget) {
this(agents, turnListener, ready, forget, System::currentTimeMillis);
// fleetd #556: no explicit registrar named here, so fall back to turnListener itself when it
// happens to also implement TurnRegistrar (true for CompletionResolver, the only production
// TurnListener that needs CB-106 registration) — every existing caller of this overload keeps
// working unchanged. A turnListener that does NOT implement TurnRegistrar (a fan-out object,
// or a bare test lambda) gets TurnRegistrar.NOOP here, same as before this ticket.
this(agents, turnListener, ready, forget,
turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP);
}
/** Full constructor — for tests: an injectable wall-clock supplier (fleetd #501). */
/**
* fleetd #556: explicit-registrar overload. Use this whenever {@code turnListener} does not
* itself implement {@link TurnRegistrar} — e.g. a fan-out object that also notifies unrelated
* observers — so registration is wired directly to the one component that owns it, rather than
* relying on {@code turnListener} happening to implement both interfaces.
*/
public Injector(AgentControl agents, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget, TurnRegistrar registrar) {
this(agents, turnListener, ready, forget, registrar, System::currentTimeMillis);
}
/**
* Package-private constructor — for tests: an injectable wall-clock supplier (fleetd #501),
* with no explicit registrar (fleetd #556 fallback, same as the public 4-arg overload above).
*/
Injector(AgentControl agents, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget, LongSupplier nowMillis) {
this(agents, turnListener, ready, forget,
turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP, nowMillis);
}
/** Full constructor — for tests: an explicit registrar and an injectable wall-clock supplier. */
Injector(AgentControl agents, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget, TurnRegistrar registrar, LongSupplier nowMillis) {
this.agents = agents;
this.router = null;
this.turnListener = turnListener;
this.registrar = Objects.requireNonNull(registrar, "registrar");
this.ready = ready;
this.forget = forget;
this.nowMillis = nowMillis;
@@ -150,15 +184,28 @@ public final class Injector {
public Injector(HerdrRouter router, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget) {
this(router, turnListener, ready, forget, System::currentTimeMillis);
// fleetd #556: see the AgentControl overload above for the fallback rationale.
this(router, turnListener, ready, forget,
turnListener instanceof TurnRegistrar r ? r : TurnRegistrar.NOOP);
}
/**
* fleetd #556: explicit-registrar overload — production wiring ({@code Fleetd}) uses this, since
* its {@code turnListener} is a fan-out object that also notifies {@code SessionManager} and does
* not itself implement {@link TurnRegistrar}.
*/
public Injector(HerdrRouter router, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget, TurnRegistrar registrar) {
this(router, turnListener, ready, forget, registrar, System::currentTimeMillis);
}
/** Full constructor — for tests: an injectable wall-clock supplier (fleetd #501). */
Injector(HerdrRouter router, TurnListener turnListener, Predicate<String> ready,
Consumer<String> forget, LongSupplier nowMillis) {
Consumer<String> forget, TurnRegistrar registrar, LongSupplier nowMillis) {
this.agents = null;
this.router = router;
this.turnListener = turnListener;
this.registrar = Objects.requireNonNull(registrar, "registrar");
this.ready = ready;
this.forget = forget;
this.nowMillis = nowMillis;
@@ -506,20 +553,25 @@ public final class Injector {
//
// There are TWO futures at stake here, not one: `sent.delivered()` (the delivery future)
// and `sent.token().waiter()` (the rendezvous waiter a blocking fleet_send actually waits
// on for the worker's ANSWER). The waiter is registered only by turnListener.onDelivered(),
// called from inside the `if (sent != null)` block below. Completing `sent.delivered()`
// without also calling onDelivered leaves the waiter unregistered forever — a hang with a
// success receipt, which is worse than the plain hang this ticket is about. So the
// `finally` below does not just complete the future: on the path where the `if (sent !=
// null)` block never ran, it does that block's whole job — onDelivered, then complete.
// on for the worker's ANSWER). fleetd #556: the waiter is registered by `registrar.register`
// now — called directly and unconditionally, structural rather than delegated to a listener
// callback — never by turnListener.onDelivered() alone (before this ticket, that WAS the
// only path, and any TurnListener wired ahead of the one that mattered could throw and skip
// it; #553 could only make the reachable listener behave, not remove that dependency).
// Completing `sent.delivered()` without also calling `registrar.register` leaves the waiter
// unregistered forever — a hang with a success receipt, which is worse than the plain hang
// #553 was about. So the `finally` below does not just complete the future: on the path
// where the `if (sent != null)` block never ran, it does that block's whole job —
// register, then onDelivered (notification only, may throw), then complete.
//
// `sentHandled` is NOT allowed to move earlier than this: an earlier comment on #553
// proposed running `if (sent != null)` first, before turnCompleted/turnFailed, and that
// was withdrawn — onDelivered WRITES CompletionResolver's inFlight record for this turn,
// onTurnComplete READS it for the PREVIOUS turn, and running onDelivered first makes
// was withdrawn — `registrar.register` WRITES CompletionResolver's inFlight record for this
// turn, onTurnComplete READS it for the PREVIOUS turn, and running register first makes
// onTurnComplete resolve the NEW turn's waiter with the PREVIOUS turn's stale output
// (CB-116). The `if (sent != null)` block stays last; the `finally` is a backstop for it,
// not a replacement.
// not a replacement. fleetd #556 keeps this ordering unchanged — it only splits what used
// to be one listener call (register + notify) into two calls in the same position.
boolean sentHandled = false;
try {
if (resubmit) {
@@ -585,13 +637,13 @@ public final class Injector {
}
if (sent != null) {
// fleetd #553: record the INTENT to handle before any side effect, not the fact of
// having completed it. If onDelivered() throws part way through — e.g. after
// CompletionResolver's captureBaseline has already done its inFlight.put — a flag
// set only after this block would still read false, and the `finally` below would
// call onDelivered() a SECOND time. A second captureBaseline runs later, once
// onTurnComplete has already thrown and possibly scraped the pane, and can snapshot
// a pane that already absorbed this turn's output — which means the pane's tail
// never differs from that baseline again and CompletionResolver.resolve's own
// having completed it. If register()/onDelivered() throws part way through — e.g.
// after CompletionResolver's captureBaseline has already done its inFlight.put — a
// flag set only after this block would still read false, and the `finally` below
// would re-run this block's job a SECOND time. A second captureBaseline runs later,
// once onTurnComplete has already thrown and possibly scraped the pane, and can
// snapshot a pane that already absorbed this turn's output — which means the pane's
// tail never differs from that baseline again and CompletionResolver.resolve's own
// suppression at :364-369 (`return` keeping the in-flight record) drops every future
// completion for this turn, permanently. So this flag must go true FIRST.
sentHandled = true;
@@ -599,32 +651,41 @@ public final class Injector {
log.warn("inject to {} failed, dropped message: {}", target, sendError.getMessage());
sent.delivered().completeExceptionally(sendError);
} else {
// Baseline the pane's pre-turn content so a misattributed completion (no new output)
// can't resolve this send with the previous turn's stale answer (CB-115).
// fleetd #556: register the waiter FIRST and unconditionally — this is the
// Injector's own invariant, kept structural rather than delegated. Even if
// turnListener.onDelivered below throws (a bug in an unrelated observer, e.g.
// SessionManager's bookkeeping, or a fan-out wired in a different order), the
// waiter is already registered and resolvable — a throwing TurnListener can no
// longer take this invariant down with it.
registrar.register(target, sent.token());
// Notification only, from here down: baseline the pane's pre-turn content so a
// misattributed completion (no new output) can't resolve this send with the
// previous turn's stale answer (CB-115). Allowed to throw and to fail (the herdr
// scrape inside already fails open with baseline == null on its own).
turnListener.onDelivered(target, sent.token());
sent.delivered().complete(null);
}
}
} finally {
// Backstop (fleetd #553): if an earlier block in this try threw before the `if (sent !=
// null)` block above ran, `sentHandled` is still false here, and this does that block's
// WHOLE job — not just the future completion. Skipping onDelivered() here would leave
// sent.token().waiter() never registered with CompletionResolver, so a blocking
// fleet_send would be told its message was delivered and then wait out its full timeout
// for an answer that can never resolve — worse than the plain hang, because it now looks
// like success. onDelivered() runs only when sendError == null: nothing was delivered on
// the error path, so there is nothing to register.
// Backstop (fleetd #553, extended #556): if an earlier block in this try threw before the
// `if (sent != null)` block above ran, `sentHandled` is still false here, and this does
// that block's WHOLE job — not just the future completion. Skipping registrar.register()
// here would leave sent.token().waiter() never registered with CompletionResolver, so a
// blocking fleet_send would be told its message was delivered and then wait out its full
// timeout for an answer that can never resolve — worse than the plain hang, because it now
// looks like success. This block runs only when sendError == null: nothing was delivered
// on the error path, so there is nothing to register.
//
// fleetd #553 (lead review, comment 16916): `sentHandled` guards ONLY the onDelivered
// fleetd #553 (lead review, comment 16916): `sentHandled` guards ONLY the register/notify
// re-call below, never the future completion outside this inner try. One boolean cannot
// carry both meanings — "onDelivered was called" and "the future has been dealt with" —
// carry both meanings — "the turn was handled" and "the future has been dealt with" —
// because they come apart exactly when onDelivered throws PART WAY THROUGH: sentHandled
// is already true (set before the call, correctly — see the `if (sent != null)` block
// above), so a guard on the outer `if` here would skip this whole recovery, including the
// completion, and leave sent.delivered() pending forever even though the message really
// was typed into the pane and taken off the queue. So `sent != null` alone gates whether
// this target has anything to finish; `!sentHandled` gates only the onDelivered re-call
// inside. CompletableFuture.complete/completeExceptionally are idempotent — on the
// this target has anything to finish; `!sentHandled` gates only the register/notify
// re-call inside. CompletableFuture.complete/completeExceptionally are idempotent — on the
// ordinary path (sentHandled == true, no throw) the `if (sent != null)` block above has
// already completed this future, so the calls below are a no-op returning false.
//
@@ -635,6 +696,11 @@ public final class Injector {
if (sent != null) {
try {
if (!sentHandled && sendError == null) {
// fleetd #556: register first here too, same as the ordinary path above — this
// recovery path is reached only when an EARLIER block (resubmit/turnCompleted/
// turnFailed) threw before the ordinary path ever ran, so nothing has
// registered this turn yet.
registrar.register(target, sent.token());
turnListener.onDelivered(target, sent.token());
}
} catch (Throwable recoveryError) {
fleetd/src/main/java/dev/ltms/fleet/inject/TurnRegistrar.java
+29
View File
@@ -0,0 +1,29 @@
package dev.ltms.fleet.inject;
import dev.ltms.fleet.msg.TurnToken;
/**
* fleetd #556: the {@link Injector}'s own invariant — every delivered turn has a registered
* waiter — kept structural rather than delegated to a {@link TurnListener} callback that can
* throw. The {@link Injector} calls {@link #register} directly and unconditionally as part of
* delivering a turn, before it ever calls {@code turnListener.onDelivered}. A {@link TurnListener}
* that throws from every callback (a bug in an unrelated observer, e.g. session bookkeeping, or a
* fan-out object wired in the wrong order) can therefore never leave a delivered turn unregistered
* — the shape #553 could not close, since that fix could only make the reachable listener behave,
* never remove the Injector's dependency on a listener behaving at all.
*
* <p>Deliberately narrow: registration only, no I/O, nothing that scrapes a pane or can reasonably
* fail. The CB-115 staleness baseline (a herdr round-trip, allowed to fail) stays a
* {@link TurnListener#onDelivered} concern — fired by the {@link Injector} only after this call has
* already run, so its own failure cannot un-register anything.
*/
@FunctionalInterface
public interface TurnRegistrar {
/** Record {@code token}'s waiter as the turn currently in flight for {@code target}. */
void register(String target, TurnToken token);
/** No-op registrar for callers that don't need CB-106 rendezvous tracking. */
TurnRegistrar NOOP = (_, _) -> {
};
}
fleetd/src/test/java/dev/ltms/fleet/inject/CompletionResolverTest.java
+106
View File
@@ -19,6 +19,7 @@ import java.util.regex.Pattern;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Unit behaviour of the CB-106 completion resolver in isolation from the injector. */
@@ -519,6 +520,111 @@ class CompletionResolverTest {
assertTrue(rendezvous.isWaiting("term_a"), "turn N+1 is still awaiting its own resolution");
}
// --- fleetd #556 (comment 16908): name the invariant the two-arg remove(target, turn) idiom
// exists to maintain — a superseded turn's terminal handling must not evict its successor's
// registration — and test THAT, not the idiom's literal spelling. Three entry points reach a
// conditional remove: resolve()'s plain completion path, resolve()'s echoed-brief/noReportMessage
// sub-path, and fail(). This is a non-goal-to-break for fleetd #556's redesign: flattening any of
// these removes to the one-arg form must turn one of these three red. -------------------------
@Test
void aSupersededTurnsPlainCompletionMustNotEvictItsSuccessorsRegistration() {
// Turn A is overtaken by turn B on the same target (a rapid back-to-back send): B's own
// delivery overwrites the map entry A's delivery installed. A's late completion fallback
// then finally fires and reaches resolve()'s ordinary (non-echoed) completion branch — the
// two-arg remove(target, turnA) there must be a no-op once the map holds B, not a blind
// evict.
FakeHerdr herdr = new FakeHerdr().readText("⏺ A's pre-turn pane\n❯ ");
Rendezvous rendezvous = new Rendezvous();
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none());
var waiterA = rendezvous.open("term_a");
var turnA = new CompletionResolver.InFlight(waiterA, null); // baseline null: fails open, never suppressed
rendezvous.close("term_a", waiterA); // A's own send's finally, freeing the session for B
var waiterB = rendezvous.open("term_a");
resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB)); // the injector registering B
herdr.readText("⏺ A's late, genuine answer\n❯ "); // what the pane shows when A's fallback fires
resolver.resolve("term_a", turnA);
assertTrue(waiterA.isDone(), "A's own resolve must still complete A's waiter normally");
assertEquals(Rendezvous.Kind.COMPLETION, waiterA.getNow(null).kind());
CompletionResolver.InFlight afterA = resolver.inFlight("term_a");
assertNotNull(afterA, "B's registration must still be present — a one-arg remove(target) "
+ "here would evict it even though the map no longer holds turnA");
assertEquals(waiterB, afterA.waiter(), "the surviving entry must still be B's, untouched by A's "
+ "terminal handling");
assertTrue(rendezvous.resolve("term_a", "B replied"),
"B must still resolve normally through the ordinary reply path");
assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind());
}
@Test
void aSupersededTurnsEchoedNoReportSubPathMustNotEvictItsSuccessorsRegistration() {
// Same invariant, driven down resolve()'s OTHER branch: a pane that merely echoes the
// injected brief back (no real report) resolves through noReportMessage(), not the plain
// tail — comment 16908 calls this "a sub-path of the first, not a separate method", reaching
// the same terminal remove() call site from different logic above it.
String echoedBrief = "y".repeat(450); // >= CompletionResolver.ECHO_MIN_CHARS normalised chars
FakeHerdr herdr = new FakeHerdr().readText("⏺ " + echoedBrief + "\n❯ ");
Rendezvous rendezvous = new Rendezvous();
CompletionResolver resolver = new CompletionResolver(new AgentControl(herdr), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none());
var waiterA = rendezvous.open("term_a");
// Back-dated deliveredAtNanos (like the 2-arg InFlight convenience ctor) so MIN_TURN_NANOS
// never fires; injectedText == the pane's own content so echoesInjectedBrief() is true.
var turnA = new CompletionResolver.InFlight(waiterA, null, Long.MIN_VALUE / 2, echoedBrief);
rendezvous.close("term_a", waiterA);
var waiterB = rendezvous.open("term_a");
resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
resolver.resolve("term_a", turnA);
assertTrue(waiterA.isDone(), "A's own resolve must still complete A's waiter via the echoed "
+ "no-report branch");
assertTrue(waiterA.getNow(null).text().contains(CompletionResolver.NO_REPORT_PREFIX),
"sanity: A must actually have gone down the noReportMessage sub-path, not the plain one");
CompletionResolver.InFlight afterA = resolver.inFlight("term_a");
assertNotNull(afterA, "B's registration must still be present after A's echoed-brief resolve");
assertEquals(waiterB, afterA.waiter());
assertTrue(rendezvous.resolve("term_a", "B replied"),
"B must still resolve normally through the ordinary reply path");
assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind());
}
@Test
void aSupersededTurnsFailMustNotEvictItsSuccessorsRegistration() {
// The fail() entry point (CB-109 wedge / CB-110 drop): turn A is failed while turn B has
// already superseded it in the map. fail()'s own two-arg remove(target, turnA) must be a
// no-op once the map holds B.
Rendezvous rendezvous = new Rendezvous();
CompletionResolver resolver = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none());
var waiterA = rendezvous.open("term_a");
var turnA = new CompletionResolver.InFlight(waiterA, null);
rendezvous.close("term_a", waiterA);
var waiterB = rendezvous.open("term_a");
resolver.captureBaseline("term_a", new TurnToken("term_a", waiterB));
resolver.fail("term_a", turnA, "A wedged in an unknown state");
assertTrue(waiterA.isDone(), "A's own fail must still complete A's waiter");
assertEquals(Rendezvous.Kind.FAILED, waiterA.getNow(null).kind());
CompletionResolver.InFlight afterA = resolver.inFlight("term_a");
assertNotNull(afterA, "B's registration must still be present — a one-arg remove(target) in "
+ "fail() would evict it even though the map no longer holds turnA");
assertEquals(waiterB, afterA.waiter());
assertTrue(rendezvous.resolve("term_a", "B replied"),
"B must still resolve normally through the ordinary reply path");
assertEquals(Rendezvous.Kind.REPLY, waiterB.getNow(null).kind());
}
// --- CB-578 stage A: backend-exhausted classification ---------------------------------
@Test
fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java
+162
View File
@@ -1197,4 +1197,166 @@ class InjectorTest {
+ "every path out of onStatus, including the one where onDelivered itself is "
+ "what threw. Leaving it pending here is a hang, not a fix");
}
// --- fleetd #556: registration is the Injector's own invariant now, structural rather than
// delegated to a TurnListener callback that can throw. #553 could only make the ONE reachable
// listener behave; this closes the shape itself. ---
@Test
void aTurnListenerThatThrowsFromEveryCallbackStillLeavesTheDeliveredTurnRegisteredAndResolvable() {
// fleetd #556 acceptance criterion 1 (the ticket's own, restated by comment 16966 as the
// replacement for #561's order-dependent test): a TurnListener that throws from EVERY
// callback it implements — nothing about it is safe to lean on — must still leave a
// delivered turn registered, with its waiter still resolvable through the ordinary path.
// Before this ticket, registration lived INSIDE onDelivered, one of the callbacks that
// throws here — this test is red against that shape, because the only listener callback
// this specific delivery scenario exercises (a first delivery, no previous turn to
// complete) is exactly the one that throws.
RuntimeException boom = new RuntimeException("fleetd #556: throws from every callback");
TurnListener throwsEverywhere = new TurnListener() {
@Override
public void onTurnComplete(String target) {
throw boom;
}
@Override
public boolean hasPostTurnAction(String target) {
throw boom;
}
@Override
public boolean onTurnCompleteWithPostAction(String target) {
throw boom;
}
@Override
public void onTurnFailed(String target) {
throw boom;
}
@Override
public void onTurnFailed(String target, String reason) {
throw boom;
}
@Override
public void onDelivered(String target, TurnToken token) {
throw boom;
}
};
Rendezvous rendezvous = new Rendezvous();
CompletionResolver completion = new CompletionResolver(new AgentControl(new FakeHerdr()), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none());
Injector inj = new Injector(new AgentControl(herdr), throwsEverywhere, _ -> true, _ -> {
}, completion::register);
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.open(T);
inj.enqueue(T, "brief", new TurnToken(T, waiter));
RuntimeException thrown = assertThrows(RuntimeException.class,
() -> inj.onStatus(T, AgentStatus.IDLE), // delivers "brief"; onDelivered throws
"the throwing listener's own bug must still escape onStatus — a listener bug stays "
+ "loud (fleetd #553's own guarantee, unchanged by this ticket)");
assertSame(boom, thrown);
CompletionResolver.InFlight inFlight = completion.inFlight(T);
assertNotNull(inFlight,
"fleetd #556: the delivered turn must be registered even though the ONLY "
+ "TurnListener wired throws from every single callback, including "
+ "onDelivered — registration must not depend on that call succeeding");
assertSame(waiter, inFlight.waiter(),
"the registered entry must carry the exact waiter this turn's send opened");
assertFalse(waiter.isDone(), "sanity: nothing has resolved this waiter yet");
assertTrue(rendezvous.resolve(T, "hello"),
"the waiter registrar.register captured must be the real, live rendezvous waiter — "
+ "still resolvable through the ordinary fleet_reply path, unaffected by the "
+ "throwing listener");
assertEquals(Rendezvous.Kind.REPLY, waiter.getNow(null).kind());
}
@Test
void theNewTurnsRegistrationRunsAfterThePreviousTurnsCompletionIsRead() {
// fleetd #556 acceptance criterion 2 (CB-116 guard, ticket body + comment 16966): the
// ordering constraint from fleetd #553 must survive this redesign — onTurnComplete reads
// CompletionResolver's inFlight entry for the PREVIOUS turn, so the new turn's
// registrar.register() call must not run before that read, or a redesign trades this
// ticket's defect for the CB-116 cross-turn stale reply. Pinned here as a call-ORDER
// assertion so a future redesign that hoists registrar.register() earlier (e.g. "for
// simplicity", ahead of the turnCompleted block) goes red immediately, before it can ever
// reach a real cross-turn scrape race.
List<String> events = new ArrayList<>();
TurnListener listener = new TurnListener() {
@Override
public void onTurnComplete(String target) {
events.add("onTurnComplete:" + target);
}
};
TurnRegistrar registrar = (target, token) -> events.add("register:" + target);
Injector inj = new Injector(new AgentControl(herdr), listener, _ -> true, _ -> {
}, registrar);
inj.enqueue(T, "first", TestTurnTokens.inert(T));
inj.enqueue(T, "second", TestTurnTokens.inert(T));
inj.onStatus(T, AgentStatus.IDLE); // delivers "first" -> register:term_a (not asserted below)
inj.onStatus(T, AgentStatus.WORKING); // picked up
events.clear();
inj.onStatus(T, AgentStatus.IDLE); // "first" completes AND "second" delivers, same cycle
assertEquals(List.of("onTurnComplete:" + T, "register:" + T), events,
"onTurnComplete for the previous turn (\"first\") must run before register for the "
+ "new turn (\"second\") within the same onStatus cycle — reversing this order "
+ "is the CB-116 cross-turn stale reply fleetd #553 fixed, and this redesign "
+ "must not reopen it");
}
@Test
void aRuntimeExceptionFromOnTurnCompleteStillLeavesTheNextDeliveryRegisteredOnTheRecoveryPath() {
// fleetd #556 rework (PR #566, comment 17009): there are TWO registrar.register call sites
// in the delivery method — the ordinary path inside `if (sent != null)`, and the fleetd
// #553 `finally` backstop, reached only when an earlier block (here, onTurnComplete for the
// PREVIOUS turn) throws before the ordinary path ever runs. The existing #553 regression
// test for this exact scenario
// (aRuntimeExceptionFromOnTurnCompleteStillCompletesTheNextDelivery, above) asserts only
// that "second"'s DELIVERED FUTURE completes — never that "second" is REGISTERED with
// CompletionResolver — so a redesign that dropped registrar.register() from the backstop
// passed every existing test while reopening this ticket's own defect on the one path
// fleetd #553 exists for. This test closes that gap: on the recovery path, the backstop is
// the ONLY thing that registers "second", so its waiter must still be resolvable afterward.
RuntimeException boom = new RuntimeException("fleetd #556 rework: boom from onTurnComplete");
TurnListener throwing = new TurnListener() {
@Override
public void onTurnComplete(String target) {
throw boom;
}
};
Rendezvous rendezvous = new Rendezvous();
CompletionResolver completion = new CompletionResolver(new AgentControl(herdr), rendezvous,
ExhaustedPatternLookup.none(), ExhaustionSink.none());
Injector inj = new Injector(new AgentControl(herdr), throwing, _ -> true, _ -> {
}, completion::register);
inj.enqueue(T, "first", TestTurnTokens.inert(T));
CompletableFuture<Rendezvous.Resolution> secondWaiter = rendezvous.open(T);
inj.enqueue(T, "second", new TurnToken(T, secondWaiter));
inj.onStatus(T, AgentStatus.IDLE); // delivers "first" (inert token — nothing to register)
inj.onStatus(T, AgentStatus.WORKING); // picked up
RuntimeException thrown = assertThrows(RuntimeException.class,
// "first" completes (onTurnComplete throws) before "second"'s ordinary delivery
// path ever runs; only the finally backstop is left to register "second".
() -> inj.onStatus(T, AgentStatus.IDLE),
"the throwable from onTurnComplete must still escape onStatus");
assertSame(boom, thrown);
CompletionResolver.InFlight inFlight = completion.inFlight(T);
assertNotNull(inFlight,
"fleetd #556: \"second\" must be registered by the fleetd #553 finally backstop "
+ "even though onTurnComplete threw for \"first\"'s completion before the "
+ "ordinary registration path ever ran for \"second\"");
assertSame(secondWaiter, inFlight.waiter(),
"the registered entry must carry \"second\"'s own waiter, not some other value");
}
}