Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Dai Ha
|
a4dbc8f8b7 |
@@ -2,7 +2,6 @@ package dev.ltms.fleet.inject;
|
||||
|
||||
import dev.ltms.fleet.herdr.AgentControl;
|
||||
import dev.ltms.fleet.herdr.AgentStatus;
|
||||
import dev.ltms.fleet.herdr.HerdrException;
|
||||
import dev.ltms.fleet.herdr.HerdrRouter;
|
||||
import dev.ltms.fleet.msg.TurnToken;
|
||||
import org.slf4j.Logger;
|
||||
@@ -47,17 +46,6 @@ import java.util.stream.Collectors;
|
||||
* turn — the pickup-grace path (a turn too fast to sample) unwedges the queue but does not fire
|
||||
* completion, since without a sampled {@code working} there is no trustworthy "the worker just
|
||||
* finished the task" signal to act on.
|
||||
*
|
||||
* <p><strong>Delivery honesty (fleetd #551).</strong> The one external send call
|
||||
* ({@link AgentControl#send}) is irreversible, and its own response can still fail after the text
|
||||
* has already reached the worker's pane — herdr replied (or may have), so the request was already
|
||||
* processed, but the reply itself then failed to parse or carried an error. A queued entry is
|
||||
* polled off the queue and marked {@code ATTEMPTED} <em>before</em> that call is made, not after,
|
||||
* so a failure from the call itself can never be recorded as a confident {@code NOT_DELIVERED} for
|
||||
* text that may already be sitting in the pane. {@code NOT_DELIVERED} stays reserved for the cases
|
||||
* where nothing was ever attempted — the readiness grace expiring, {@link #drop}, or a herdr
|
||||
* {@code *_not_found} error, which the rest of this codebase already treats as a confirmed absence
|
||||
* rather than a merely inconclusive failure (see {@code StatusPoller}, {@code AgentControl}).
|
||||
*/
|
||||
public final class Injector {
|
||||
|
||||
@@ -229,23 +217,8 @@ public final class Injector {
|
||||
|
||||
/** The result of trying to remove an undelivered message from the injector. */
|
||||
public enum Cancellation {
|
||||
/** The message was still queued and this call removed it; the target saw nothing. */
|
||||
CANCELLED,
|
||||
/** The message reached the worker's pane and is confirmed delivered. */
|
||||
DELIVERED,
|
||||
/**
|
||||
* fleetd #551: the injector called {@link AgentControl#send} for this message and that call
|
||||
* threw before its outcome was known — the text may or may not have reached the pane. A
|
||||
* caller reporting this to an operator must say "uncertain", not "definitely not
|
||||
* delivered": reading it as a confident negative invites a resend of text that may already
|
||||
* be sitting in the pane (a double delivery), which is worse than the ambiguity itself.
|
||||
*/
|
||||
ATTEMPTED,
|
||||
/**
|
||||
* The message never reached the worker's pane — nothing was ever attempted for it (the
|
||||
* readiness grace expired, the target was dropped, or send failed with a herdr
|
||||
* {@code *_not_found} error, which is a confirmed absence, not merely inconclusive).
|
||||
*/
|
||||
NOT_DELIVERED
|
||||
}
|
||||
|
||||
@@ -268,29 +241,7 @@ public final class Injector {
|
||||
|
||||
/** A pending message and the future that completes when it has been delivered. */
|
||||
private static final class Pending {
|
||||
enum State {
|
||||
/** Still in the target's queue, not yet attempted. */
|
||||
QUEUED,
|
||||
/**
|
||||
* fleetd #551: {@link AgentControl#send} has been called for this entry and its outcome
|
||||
* is not yet known — recorded BEFORE the call (see {@link #onStatus}), so a Throwable
|
||||
* from send() can never leave the entry either still QUEUED or falsely marked
|
||||
* NOT_DELIVERED. Upgraded to DELIVERED on success; left as ATTEMPTED on an ordinary
|
||||
* failure, since reaching the catch does not prove the text never reached the pane.
|
||||
*/
|
||||
ATTEMPTED,
|
||||
/** {@code send()} returned normally: the text is confirmed to have reached the pane. */
|
||||
DELIVERED,
|
||||
/**
|
||||
* Confirmed — not merely inconclusive — that nothing was ever sent for this entry: the
|
||||
* worker never became ready ({@link #onStatus}'s readiness-grace expiry), the target was
|
||||
* dropped ({@link #drop}), or send failed with a herdr {@code *_not_found} error. Never
|
||||
* written for an entry whose send() outcome is unknown; see ATTEMPTED.
|
||||
*/
|
||||
NOT_DELIVERED,
|
||||
/** Removed from the queue by {@link #cancel} before it was ever attempted. */
|
||||
CANCELLED
|
||||
}
|
||||
enum State { QUEUED, DELIVERED, NOT_DELIVERED, CANCELLED }
|
||||
|
||||
final String target;
|
||||
final String text;
|
||||
@@ -382,14 +333,7 @@ public final class Injector {
|
||||
}
|
||||
|
||||
private static Cancellation cancellationOf(Pending p) {
|
||||
// fleetd #551 (comment 17037): a two-way split on a now-three-way question folded ATTEMPTED
|
||||
// into NOT_DELIVERED with no compiler error and no failing test — the exact defect this
|
||||
// ticket exists to fix, one layer up. ATTEMPTED gets its own answer instead.
|
||||
return switch (p.state) {
|
||||
case DELIVERED -> Cancellation.DELIVERED;
|
||||
case ATTEMPTED -> Cancellation.ATTEMPTED;
|
||||
case QUEUED, NOT_DELIVERED, CANCELLED -> Cancellation.NOT_DELIVERED;
|
||||
};
|
||||
return p.state == Pending.State.DELIVERED ? Cancellation.DELIVERED : Cancellation.NOT_DELIVERED;
|
||||
}
|
||||
|
||||
private static boolean isQuiescent(Target t) {
|
||||
@@ -482,18 +426,9 @@ public final class Injector {
|
||||
if (p != null && ready.test(target)) {
|
||||
t.notReadySincePoll = 0;
|
||||
t.notReadySinceMillis = 0;
|
||||
// fleetd #551: poll and record BEFORE the irreversible send, not after.
|
||||
// The entry comes off the queue and its state is set to ATTEMPTED here,
|
||||
// unconditionally — so a Throwable escaping the send call below (caught or
|
||||
// not) can never leave the entry QUEUED at the head of t.queue (the fleetd
|
||||
// #546 hazard, since peek() alone would let the next onStatus round re-enter
|
||||
// this block and send the same text again), and no path can write a
|
||||
// confident DELIVERED or NOT_DELIVERED before we actually know which one
|
||||
// happened.
|
||||
t.queue.poll();
|
||||
p.state = Pending.State.ATTEMPTED;
|
||||
try {
|
||||
agentsFor(target).send(target, p.text());
|
||||
t.queue.poll();
|
||||
p.state = Pending.State.DELIVERED;
|
||||
t.awaitingPickup = true;
|
||||
t.awaitingCompletion = true;
|
||||
@@ -501,24 +436,15 @@ public final class Injector {
|
||||
t.injectableSincePickup = 0;
|
||||
sent = p;
|
||||
} catch (Throwable e) {
|
||||
// fleetd #551: leave p.state == ATTEMPTED (recorded above, before the
|
||||
// call) rather than downgrading it to NOT_DELIVERED here — reaching this
|
||||
// catch does not prove the text never reached the pane. Three of the
|
||||
// four HerdrException throw sites in HerdrCodec fire only after herdr
|
||||
// has already replied (so it processed the request), and the fourth (a
|
||||
// transport IOException) leaves it genuinely unknown whether herdr even
|
||||
// received the bytes — see #551 comment 16867. The one exception is a
|
||||
// herdr `*_not_found` error: that family is already read as "definitely
|
||||
// absent, not merely inconclusive" everywhere else in this codebase
|
||||
// (StatusPoller, AgentControl's own retry, WorkspaceControl,
|
||||
// HerdrPeerLauncher, FleetApp, ReplyPushLoop) because it means the
|
||||
// target pane/agent does not exist at all, so nothing could have been
|
||||
// pasted anywhere — #551 keeps the new state consistent with that
|
||||
// existing vocabulary rather than inventing a second one.
|
||||
if (e instanceof HerdrException he && he.code() != null
|
||||
&& he.code().endsWith("_not_found")) {
|
||||
p.state = Pending.State.NOT_DELIVERED;
|
||||
}
|
||||
// Delivery failed at herdr; drop the poisoned message and surface it
|
||||
// rather than blocking the queue behind it. Catches Throwable, not just
|
||||
// RuntimeException: fleetd #546 — an Error escaping this send (e.g. a
|
||||
// NoClassDefFoundError, see #413) would otherwise leave the entry QUEUED
|
||||
// at the head of t.queue. Line :378 peeks rather than polls, so the next
|
||||
// onStatus round would re-enter this try and send the same text again,
|
||||
// typing the same brief into the member's pane a second time.
|
||||
t.queue.poll();
|
||||
p.state = Pending.State.NOT_DELIVERED;
|
||||
sent = p;
|
||||
sendError = e;
|
||||
}
|
||||
|
||||
@@ -94,22 +94,16 @@ public final class MessageService {
|
||||
TIMED_OUT_WORKING,
|
||||
/**
|
||||
* Timed out with no confirmed delivery. Despite the name, this does not mean the message
|
||||
* is sitting in a queue. {@link #send} reaches this outcome through {@link Injector#cancel},
|
||||
* whose result tells three routes apart:
|
||||
* is sitting in a queue — on every route it will not arrive later. {@link #send} reaches
|
||||
* this outcome through {@link Injector#cancel}, whose result tells the routes apart:
|
||||
* {@link Injector.Cancellation#CANCELLED} means the message was still queued and this call
|
||||
* removed it, so the target saw nothing and it will not arrive later;
|
||||
* {@link Injector.Cancellation#NOT_DELIVERED} means nothing was ever sent — the queue was
|
||||
* cleared because the target never became ready or was abandoned, or the injector's call to
|
||||
* the target's terminal ({@link dev.ltms.fleet.herdr.AgentControl#send}) failed with a herdr
|
||||
* error that this codebase already treats as a confirmed absence — so this route too
|
||||
* establishes that the target saw nothing and it will not arrive later; but
|
||||
* {@link Injector.Cancellation#ATTEMPTED} (fleetd #551) means that call was made and its
|
||||
* outcome is unknown. {@code agent.prompt} pastes <em>and submits</em> in one call, so on
|
||||
* this route the target may hold a complete, already-submitted turn and be working on it
|
||||
* right now — {@link Outcome#TIMED_OUT_WORKING}'s meaning, reported here as
|
||||
* {@code TIMED_OUT_QUEUED} only because this caller never observed the pickup. Only
|
||||
* {@code CANCELLED} and {@code NOT_DELIVERED} establish that the target saw nothing;
|
||||
* {@code ATTEMPTED} does not.
|
||||
* removed it, so the target saw nothing; {@link Injector.Cancellation#NOT_DELIVERED} means
|
||||
* an earlier attempt already decided the message's fate — the injector's call to the
|
||||
* target's terminal ({@link dev.ltms.fleet.herdr.AgentControl#send}) threw, or the queue
|
||||
* was cleared because the target never became ready or was abandoned — and {@code cancel}
|
||||
* is only reporting that pre-existing state. On the failed-attempt route the terminal may
|
||||
* already hold a partial paste from before the call threw, so only the {@code CANCELLED}
|
||||
* case establishes that the target saw nothing.
|
||||
*/
|
||||
TIMED_OUT_QUEUED,
|
||||
/** Another send to this session was in flight for the whole window. */
|
||||
@@ -319,18 +313,16 @@ public final class MessageService {
|
||||
/**
|
||||
* Targets whose last send timed out with {@link Outcome#TIMED_OUT_QUEUED} (CB-640) — {@link
|
||||
* #send} called {@link Injector#cancel} and got back something other than {@code DELIVERED}.
|
||||
* That covers three histories, not one: {@link Injector.Cancellation#CANCELLED} — the message
|
||||
* was still queued and {@code cancel} removed it right there; {@link
|
||||
* Injector.Cancellation#NOT_DELIVERED} — nothing was ever sent, because the target never became
|
||||
* ready, was torn down, or the call to its terminal failed with a herdr error this codebase
|
||||
* already treats as a confirmed absence; or {@link Injector.Cancellation#ATTEMPTED} (fleetd
|
||||
* #551) — the call to the target's terminal was made and its outcome is unknown, so the target
|
||||
* may already hold a complete, submitted turn. Only the first two mean the message will not
|
||||
* arrive later and the target saw nothing; on the third it may already have arrived in full.
|
||||
* Set where {@link #send} already computes {@code wasDelivered} for that outcome; no queue is
|
||||
* kept here, only the fact that the send ended with no confirmed delivery. Cleared the same way
|
||||
* as {@link #strandedReplies}: the next accepted delivery for the target ({@link #send} opening
|
||||
* a fresh waiter) or a teardown ({@link #abandon}).
|
||||
* That covers more than one history: the message may have still been queued and {@code cancel}
|
||||
* removed it right there, or an earlier attempt may already have failed (the call to the
|
||||
* target's terminal threw) or been abandoned (the target never became ready, or was torn
|
||||
* down). What holds on every route: the message will not arrive later, and it is not sitting
|
||||
* in any queue. What does NOT hold on every route: that the target saw nothing — a failed
|
||||
* delivery attempt can leave a partial paste behind. Set where {@link #send} already computes
|
||||
* {@code wasDelivered} for that outcome; no queue is kept here, only the fact that the send
|
||||
* ended with no confirmed delivery. Cleared the same way as {@link #strandedReplies}: the next
|
||||
* accepted delivery for the target ({@link #send} opening a fresh waiter) or a teardown
|
||||
* ({@link #abandon}).
|
||||
*/
|
||||
private final ConcurrentHashMap<String, Boolean> queuedDeliveries = new ConcurrentHashMap<>();
|
||||
private final AtomicLong ticketSeq = new AtomicLong();
|
||||
@@ -420,18 +412,14 @@ public final class MessageService {
|
||||
* with no confirmed delivery — the caller saw {@link Outcome#TIMED_OUT_QUEUED} (see the
|
||||
* {@code TimeoutException} branch of {@link #send}). Despite the method's name, this is not
|
||||
* proof that a message is sitting in a queue: {@link Injector#cancel} reports this outcome
|
||||
* through three routes. {@link Injector.Cancellation#CANCELLED} means the message was still
|
||||
* queued and got removed right there. {@link Injector.Cancellation#NOT_DELIVERED} means
|
||||
* nothing was ever sent — the target never became ready, was torn down, or the call to its
|
||||
* terminal failed with a herdr error this codebase already treats as a confirmed absence.
|
||||
* Only these two routes mean the message will not arrive later. {@link
|
||||
* Injector.Cancellation#ATTEMPTED} (fleetd #551) means the call to the target's terminal was
|
||||
* made and its outcome is unknown: {@code agent.prompt} pastes <em>and submits</em> in one
|
||||
* call, so on this route the target may already hold a complete, submitted turn and be
|
||||
* working on it right now — it does NOT follow that the target saw nothing. Distinct from
|
||||
* {@link Outcome#TIMED_OUT_WORKING}, where delivery already happened and only the reply is
|
||||
* outstanding. Cleared the next time this target's delivery is accepted or the target is
|
||||
* abandoned — see {@link #queuedDeliveries}.
|
||||
* either because the message was still queued and got removed right there, or because an
|
||||
* earlier attempt already failed (the call to the target's terminal threw) or was abandoned
|
||||
* (the target never became ready, or was torn down). Either way the message will not arrive
|
||||
* later. It does NOT follow that the target saw nothing — on the failed-attempt route the
|
||||
* terminal may already hold a partial paste. Distinct from {@link Outcome#TIMED_OUT_WORKING},
|
||||
* where delivery already happened and only the reply is outstanding. Cleared the next time
|
||||
* this target's delivery is accepted or the target is abandoned — see
|
||||
* {@link #queuedDeliveries}.
|
||||
*/
|
||||
public boolean hasQueuedDelivery(String target) {
|
||||
return target != null && queuedDeliveries.containsKey(target);
|
||||
@@ -986,12 +974,11 @@ public final class MessageService {
|
||||
}
|
||||
log.debug("send to {} timed out (delivered={})", target, wasDelivered);
|
||||
if (!wasDelivered) {
|
||||
// CB-640: record that delivery is not confirmed, for fleet health (see
|
||||
// CB-640: record that delivery did not happen for fleet health (see
|
||||
// queuedDeliveries). Whatever injector.cancel() reported above — this call
|
||||
// removed a still-queued Pending (CANCELLED), an earlier attempt already
|
||||
// failed with a confirmed absence (NOT_DELIVERED), or an earlier attempt was
|
||||
// made and its outcome is unknown (ATTEMPTED, fleetd #551 — the message may
|
||||
// already have arrived in full) — the send ends with no confirmed delivery.
|
||||
// removed a still-queued Pending, or an earlier attempt already failed or
|
||||
// was abandoned — the send ends with no confirmed delivery and will not
|
||||
// arrive later.
|
||||
queuedDeliveries.put(target, Boolean.TRUE);
|
||||
}
|
||||
return recorded(new Reply(
|
||||
|
||||
@@ -720,13 +720,10 @@ class InjectorTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void anErrorFromSendRemovesTheMessageAndMarksItAttempted() {
|
||||
void anErrorFromSendRemovesTheMessageAndMarksItNotDelivered() {
|
||||
// fleetd #546, acceptance test 1: an Error (not a RuntimeException) escaping the send seam
|
||||
// must still be caught and the message dropped from the queue, never left QUEUED at the
|
||||
// head. fleetd #551 updates what it is marked: an AssertionError from the send call itself
|
||||
// proves nothing about whether the text reached the pane, so it is recorded as ATTEMPTED
|
||||
// (uncertain), not a confident NOT_DELIVERED — see anErrorFromSendMarksItNotDeliveredOnlyForANotFoundCode
|
||||
// below for the one case that still gets NOT_DELIVERED.
|
||||
// at Injector.java:383 must still be caught, the message dropped from the queue, and its
|
||||
// state set to NOT_DELIVERED — never left QUEUED at the head.
|
||||
ErrorOnPrompt throwing = new ErrorOnPrompt(new FakeHerdr());
|
||||
Injector inj = new Injector(new AgentControl(throwing));
|
||||
Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
|
||||
@@ -737,9 +734,9 @@ class InjectorTest {
|
||||
|
||||
assertTrue(delivery.completion().isCompletedExceptionally(),
|
||||
"the delivery's future must surface the send failure");
|
||||
assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery),
|
||||
"fleetd #551: the message must be dropped and marked ATTEMPTED (not a confident "
|
||||
+ "NOT_DELIVERED), and never left QUEUED at the head of the queue");
|
||||
assertEquals(Injector.Cancellation.NOT_DELIVERED, inj.cancel(delivery),
|
||||
"the message must be dropped and marked NOT_DELIVERED, not left QUEUED at the "
|
||||
+ "head of the queue");
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -761,14 +758,10 @@ class InjectorTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
void aHerdrExceptionFromSendStillSurfacesButNowReportsAttempted() {
|
||||
void aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged() {
|
||||
// fleetd #546, acceptance test 3 (control): HerdrException extends RuntimeException, so it
|
||||
// was already caught before #546's widening. The send-failure/surface behaviour is
|
||||
// unchanged by #551 — still dropped, still surfaced to the caller — but fleetd #551
|
||||
// deliberately changes WHAT it is marked: "send_failed" is not a herdr `*_not_found` code,
|
||||
// so this is exactly the response-half case #551 exists to fix. See
|
||||
// aHerdrExceptionAfterThePasteIsNeverRecordedAsConfidentlyNotDelivered for the acceptance
|
||||
// test this ticket was filed for.
|
||||
// was already caught before this ticket's widening. This pins that the ordinary path is
|
||||
// unchanged — still dropped, still NOT_DELIVERED, still surfaced to the caller.
|
||||
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
|
||||
Injector inj = new Injector(new AgentControl(failing));
|
||||
Injector.Delivery delivery = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
|
||||
@@ -778,90 +771,9 @@ class InjectorTest {
|
||||
assertTrue(delivery.completion().isCompletedExceptionally(),
|
||||
"a HerdrException at the send seam must still surface to the caller, unchanged by "
|
||||
+ "fleetd #546's widening");
|
||||
assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery),
|
||||
"fleetd #551: a non-*_not_found HerdrException must now report ATTEMPTED, not a "
|
||||
+ "confident NOT_DELIVERED");
|
||||
}
|
||||
|
||||
// --- fleetd #551: the injector records the delivery attempt BEFORE the irreversible send, not
|
||||
// after, so a failure in the send's response window can never be written down as a confident
|
||||
// NOT_DELIVERED for text that may already be sitting in the worker's pane. ---
|
||||
|
||||
@Test
|
||||
void aHerdrExceptionAfterThePasteIsNeverRecordedAsConfidentlyNotDelivered() {
|
||||
// fleetd #551, the ticket's acceptance test 1, and comment 17037's added test ("a test
|
||||
// pinning that a caller cannot receive a confident NOT_DELIVERED for a message that reached
|
||||
// the paste"): a HerdrException thrown from the response half of the send seam — herdr
|
||||
// replied, with an error, so it definitely processed the request — must not leave a record
|
||||
// claiming the text was never delivered. Red before the fix: on main at ba2f4d1 this
|
||||
// asserted (and got) Injector.Cancellation.NOT_DELIVERED.
|
||||
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
|
||||
Injector inj = new Injector(new AgentControl(failing));
|
||||
Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
assertTrue(delivery.completion().isCompletedExceptionally(),
|
||||
"the caller must still see the send failure");
|
||||
assertEquals(Injector.Cancellation.ATTEMPTED, inj.cancel(delivery),
|
||||
"fleetd #551: a HerdrException from the response half of the send seam must not be "
|
||||
+ "recorded as a confident NOT_DELIVERED — the text may already be sitting "
|
||||
+ "in the pane");
|
||||
}
|
||||
|
||||
@Test
|
||||
void ordinarySuccessStillReportsDeliveredExactlyOnce() {
|
||||
// fleetd #551, the ticket's acceptance test 2: the record-before-send reorder must not
|
||||
// change the ordinary success path — exactly one send, and the caller-facing Cancellation
|
||||
// for it is still DELIVERED, never left at the new ATTEMPTED value.
|
||||
Injector.Delivery delivery = injector.enqueue(T, "hello", TestTurnTokens.inert(T));
|
||||
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
assertEquals(List.of("hello"), sent(), "the text must be sent exactly once");
|
||||
assertTrue(delivery.completion().isDone() && !delivery.completion().isCompletedExceptionally(),
|
||||
"the ordinary success path must still complete normally");
|
||||
assertEquals(Injector.Cancellation.DELIVERED, injector.cancel(delivery),
|
||||
"fleetd #551: the ordinary success path must still report DELIVERED, unaffected by "
|
||||
+ "the record-before-send reorder");
|
||||
}
|
||||
|
||||
@Test
|
||||
void transportDownStillSurfacesTheErrorAndDropsTheEntry() {
|
||||
// fleetd #551, the ticket's acceptance test 3: the ordinary transport-down path (herdr
|
||||
// unreachable — a plain HerdrException with no error code) must be unaffected by the #551
|
||||
// reorder — the failure still surfaces to the caller, and the entry is not left sitting in
|
||||
// the queue for a second onStatus round to resend.
|
||||
FakeHerdr unreachable = new FakeHerdr().healthy(false);
|
||||
Injector inj = new Injector(new AgentControl(unreachable));
|
||||
Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
assertTrue(delivery.completion().isCompletedExceptionally(),
|
||||
"a transport-down failure must still surface to the caller");
|
||||
assertTrue(inj.activeTargets().isEmpty(),
|
||||
"the entry must not be left queued for a second round to resend");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aNotFoundCodeFromSendStillReportsAConfidentNotDelivered() {
|
||||
// fleetd #551 (comment 17037, point 3): a herdr `*_not_found` error means the target
|
||||
// pane/agent does not exist at all, so nothing could have been pasted anywhere — the
|
||||
// codebase already treats this family as a confirmed absence everywhere else (StatusPoller,
|
||||
// AgentControl's own retry). #551's new ATTEMPTED state must not swallow this case: it stays
|
||||
// a confident NOT_DELIVERED.
|
||||
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("agent_not_found");
|
||||
Injector inj = new Injector(new AgentControl(failing));
|
||||
Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
|
||||
|
||||
inj.onStatus(T, AgentStatus.IDLE);
|
||||
|
||||
assertTrue(delivery.completion().isCompletedExceptionally(),
|
||||
"an agent_not_found failure must still surface to the caller");
|
||||
assertEquals(Injector.Cancellation.NOT_DELIVERED, inj.cancel(delivery),
|
||||
"fleetd #551: a herdr *_not_found error is a confirmed absence, not merely "
|
||||
+ "inconclusive — it must stay NOT_DELIVERED, not the new ATTEMPTED");
|
||||
"a HerdrException must still be dropped and marked NOT_DELIVERED, unchanged by the "
|
||||
+ "wider Throwable catch");
|
||||
}
|
||||
|
||||
// --- fleetd #553: a throwable from any listener callback in onStatus must not skip the
|
||||
|
||||
@@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit;
|
||||
|
||||
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.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
@@ -579,6 +580,183 @@ class MessageServiceTest {
|
||||
assertEquals("config.yaml", a.answer());
|
||||
}
|
||||
|
||||
// --- fleetd #572: answer() must release the session lock on EVERY exit, not just return it -----
|
||||
//
|
||||
// answer()'s session lock is released in an outer `finally` (MessageService.java:1217-1219) that
|
||||
// wraps its whole body. Removing that one line survives the entire suite: coverage is not the
|
||||
// gap, assertion is — every existing answer() test checks the RETURN VALUE, never that the lock
|
||||
// it took is reacquirable afterward. If it is not, the session is wedged forever with no
|
||||
// exception and no log line. Each test below drives answer() through one of its four exits
|
||||
// (normal reply, TIMED_OUT_WORKING, ExecutionException, InterruptedException) and then proves
|
||||
// reacquisition the only way that actually proves it: a bounded follow-up `send` on the SAME
|
||||
// session must not come back BUSY. A `send` only reports BUSY when `tryLock` itself timed out
|
||||
// (MessageService.java:926-928) — every other early return in `send` still passes through its own
|
||||
// lock-acquired `try`, so a non-BUSY probe result is specifically evidence the lock was free.
|
||||
|
||||
/** The REPLIED exit (the happy path) — the resumed worker's real {@code fleet_reply} arrives. */
|
||||
@Test
|
||||
void answerReleasesTheSessionLockAfterANormalReply() throws Exception {
|
||||
CompletableFuture<MessageService.Reply> send = sendAsync();
|
||||
awaitWaiting();
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
injector.onStatus(T, AgentStatus.WORKING);
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
|
||||
MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.QUESTION, q.outcome());
|
||||
|
||||
CompletableFuture<MessageService.Reply> answer =
|
||||
CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 5000));
|
||||
ask.get(5, TimeUnit.SECONDS); // worker resumed with the answer
|
||||
|
||||
awaitWaiting(); // the answering call has (re)opened its own forward waiter
|
||||
assertTrue(rendezvous.resolve(T, "done"), "the worker's final reply resolves the answering send");
|
||||
MessageService.Reply done = answer.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.REPLIED, done.outcome());
|
||||
|
||||
MessageService.Reply probe = messages.send(T, "probe after normal reply", 300);
|
||||
assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
|
||||
"the session lock must be released after a normal REPLIED answer(), or this bounded "
|
||||
+ "follow-up send would come back BUSY instead of timing out on its own work");
|
||||
}
|
||||
|
||||
/**
|
||||
* The {@code TIMED_OUT_WORKING} exit. Same setup as {@link #answerTimesOutWhenTheResumedWorkerNeverReplies}
|
||||
* (which only checks {@code answer()}'s return value — exactly the assertion the fleetd #572
|
||||
* mutation survives), plus the reacquisition proof that test does not make.
|
||||
*
|
||||
* <p>{@code answer()} MUST run on its own thread here, not the test's main thread: {@code lock}
|
||||
* is a {@link java.util.concurrent.locks.ReentrantLock}, so a probe {@code send} issued by the
|
||||
* SAME thread that (under the mutation) still "holds" it would reenter for free and report a
|
||||
* false pass — reentrancy, not release. Measured while writing this test: with {@code answer()}
|
||||
* called inline, this test stayed green under the mutation while its three siblings correctly
|
||||
* went red.
|
||||
*/
|
||||
@Test
|
||||
void answerReleasesTheSessionLockAfterATimedOutWorkingReturn() throws Exception {
|
||||
CompletableFuture<MessageService.Reply> send = sendAsync();
|
||||
awaitWaiting();
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
injector.onStatus(T, AgentStatus.WORKING);
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
|
||||
MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.QUESTION, q.outcome());
|
||||
|
||||
// The primary answers, unblocking the worker; the worker never sends its follow-up
|
||||
// fleet_reply, so the answering call rides out its short window as still-working.
|
||||
CompletableFuture<MessageService.Reply> answer =
|
||||
CompletableFuture.supplyAsync(() -> messages.answer(q.turnId(), "config.yaml", 200));
|
||||
MessageService.Reply answered = answer.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answered.outcome(),
|
||||
"an answered worker that never replies times out as still working");
|
||||
ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer
|
||||
|
||||
MessageService.Reply probe = messages.send(T, "probe after timed-out-working", 300);
|
||||
assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
|
||||
"the session lock must be released after a TIMED_OUT_WORKING answer(), or this bounded "
|
||||
+ "follow-up send would come back BUSY instead of timing out on its own work");
|
||||
}
|
||||
|
||||
/**
|
||||
* The {@code ExecutionException} exit. Forces it directly on answer()'s own reopened forward
|
||||
* waiter — {@code rendezvous.currentWaiter(T)} is the exact {@code CompletableFuture} its
|
||||
* {@code reply.get(...)} is blocked on — rather than trying to make a real worker fail, since the
|
||||
* failure mode under test is in answer()'s own wait, not in how it got triggered.
|
||||
*/
|
||||
@Test
|
||||
void answerReleasesTheSessionLockWhenTheReplyFutureFailsExceptionally() throws Exception {
|
||||
CompletableFuture<MessageService.Reply> send = sendAsync();
|
||||
awaitWaiting();
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
injector.onStatus(T, AgentStatus.WORKING);
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
|
||||
MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.QUESTION, q.outcome());
|
||||
String turnId = q.turnId();
|
||||
|
||||
java.util.concurrent.atomic.AtomicReference<Throwable> caught =
|
||||
new java.util.concurrent.atomic.AtomicReference<>();
|
||||
Thread answerer = new Thread(() -> {
|
||||
try {
|
||||
messages.answer(turnId, "config.yaml", 5000);
|
||||
caught.set(new AssertionError("expected answer() to throw"));
|
||||
} catch (Throwable t) {
|
||||
caught.set(t);
|
||||
}
|
||||
});
|
||||
answerer.start();
|
||||
awaitWaiting(); // answer() re-opened its forward waiter and is about to block on it
|
||||
|
||||
CompletableFuture<Rendezvous.Resolution> waiter = rendezvous.currentWaiter(T);
|
||||
assertNotNull(waiter, "answer() should have registered a forward waiter for the worker session");
|
||||
waiter.completeExceptionally(new RuntimeException("boom"));
|
||||
|
||||
answerer.join(5000);
|
||||
assertFalse(answerer.isAlive(), "answer() should have left by throwing once its reply future failed");
|
||||
Throwable thrown = caught.get();
|
||||
assertNotNull(thrown, "answer() should have thrown");
|
||||
assertTrue(thrown instanceof RuntimeException, "a RuntimeException cause is rethrown as-is: " + thrown);
|
||||
assertEquals("boom", thrown.getMessage());
|
||||
|
||||
ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer
|
||||
|
||||
MessageService.Reply probe = messages.send(T, "probe after exceptional failure", 300);
|
||||
assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
|
||||
"the session lock must be released when answer()'s reply future fails exceptionally, "
|
||||
+ "or this bounded follow-up send would come back BUSY instead of timing out on its own work");
|
||||
}
|
||||
|
||||
/**
|
||||
* The {@code InterruptedException} exit. Same shape as the {@code ExecutionException} case above,
|
||||
* but the thread blocked in {@code reply.get(...)} is interrupted instead of the future failing.
|
||||
*/
|
||||
@Test
|
||||
void answerReleasesTheSessionLockWhenInterrupted() throws Exception {
|
||||
CompletableFuture<MessageService.Reply> send = sendAsync();
|
||||
awaitWaiting();
|
||||
injector.onStatus(T, AgentStatus.IDLE);
|
||||
injector.onStatus(T, AgentStatus.WORKING);
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config file?", 5000));
|
||||
MessageService.Reply q = send.get(5, TimeUnit.SECONDS);
|
||||
assertEquals(MessageService.Outcome.QUESTION, q.outcome());
|
||||
String turnId = q.turnId();
|
||||
|
||||
java.util.concurrent.atomic.AtomicReference<Throwable> caught =
|
||||
new java.util.concurrent.atomic.AtomicReference<>();
|
||||
Thread answerer = new Thread(() -> {
|
||||
try {
|
||||
messages.answer(turnId, "config.yaml", 5000);
|
||||
caught.set(new AssertionError("expected answer() to throw"));
|
||||
} catch (Throwable t) {
|
||||
caught.set(t);
|
||||
}
|
||||
});
|
||||
answerer.start();
|
||||
awaitWaiting(); // answer() re-opened its forward waiter and is about to block on it
|
||||
|
||||
answerer.interrupt();
|
||||
answerer.join(5000);
|
||||
assertFalse(answerer.isAlive(), "answer() should have left by throwing once its thread was interrupted");
|
||||
Throwable thrown = caught.get();
|
||||
assertNotNull(thrown, "answer() should have thrown");
|
||||
assertTrue(thrown instanceof IllegalStateException,
|
||||
"an interrupted wait is wrapped in IllegalStateException: " + thrown);
|
||||
|
||||
ask.get(5, TimeUnit.SECONDS); // drain: the worker resumed with the answer
|
||||
|
||||
MessageService.Reply probe = messages.send(T, "probe after interruption", 300);
|
||||
assertNotEquals(MessageService.Outcome.BUSY, probe.outcome(),
|
||||
"the session lock must be released when answer()'s wait is interrupted, or this bounded "
|
||||
+ "follow-up send would come back BUSY instead of timing out on its own work");
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollReturnsNullForAnUnknownTicket() {
|
||||
assertNull(messages.poll("task-999999"), "a ticket that was never minted is unknown");
|
||||
|
||||
Reference in New Issue
Block a user