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

fleetd #551: record the delivery attempt before the irreversible send #569

Merged
ltms merged 3 commits from worker/551-record-before-send-7cbf56-1 into main 2026-09-12 13:19:20 +02:00
Conversation 0 Commits 3 Files Changed 3
+229 -54
3 changed files with 229 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fleetd/src/main/java/dev/ltms/fleet/inject/Injector.java
+86 -12
View File
@@ -2,6 +2,7 @@ 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;
@@ -46,6 +47,17 @@ 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 {
@@ -217,8 +229,23 @@ 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
}
@@ -241,7 +268,29 @@ public final class Injector {
/** A pending message and the future that completes when it has been delivered. */
private static final class Pending {
enum State { QUEUED, DELIVERED, NOT_DELIVERED, CANCELLED }
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
}
final String target;
final String text;
@@ -333,7 +382,14 @@ public final class Injector {
}
private static Cancellation cancellationOf(Pending p) {
return p.state == Pending.State.DELIVERED ? Cancellation.DELIVERED : Cancellation.NOT_DELIVERED;
// 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;
};
}
private static boolean isQuiescent(Target t) {
@@ -426,9 +482,18 @@ 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;
@@ -436,15 +501,24 @@ public final class Injector {
t.injectableSincePickup = 0;
sent = p;
} catch (Throwable e) {
// 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;
// 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;
}
sent = p;
sendError = e;
}
fleetd/src/main/java/dev/ltms/fleet/msg/MessageService.java
+44 -31
View File
@@ -94,16 +94,22 @@ 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 — on every route it will not arrive later. {@link #send} reaches
* this outcome through {@link Injector#cancel}, whose result tells the routes apart:
* is sitting in a queue. {@link #send} reaches this outcome through {@link Injector#cancel},
* whose result tells three routes apart:
* {@link Injector.Cancellation#CANCELLED} means the message was still queued and this call
* 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.
* 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.
*/
TIMED_OUT_QUEUED,
/** Another send to this session was in flight for the whole window. */
@@ -313,16 +319,18 @@ 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 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}).
* 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}).
*/
private final ConcurrentHashMap<String, Boolean> queuedDeliveries = new ConcurrentHashMap<>();
private final AtomicLong ticketSeq = new AtomicLong();
@@ -412,14 +420,18 @@ 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
* 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}.
* 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}.
*/
public boolean hasQueuedDelivery(String target) {
return target != null && queuedDeliveries.containsKey(target);
@@ -974,11 +986,12 @@ public final class MessageService {
}
log.debug("send to {} timed out (delivered={})", target, wasDelivered);
if (!wasDelivered) {
// CB-640: record that delivery did not happen for fleet health (see
// CB-640: record that delivery is not confirmed, for fleet health (see
// queuedDeliveries). Whatever injector.cancel() reported above — this call
// 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.
// 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.
queuedDeliveries.put(target, Boolean.TRUE);
}
return recorded(new Reply(
fleetd/src/test/java/dev/ltms/fleet/inject/InjectorTest.java
+99 -11
View File
@@ -720,10 +720,13 @@ class InjectorTest {
}
@Test
void anErrorFromSendRemovesTheMessageAndMarksItNotDelivered() {
void anErrorFromSendRemovesTheMessageAndMarksItAttempted() {
// fleetd #546, acceptance test 1: an Error (not a RuntimeException) escaping the send seam
// 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.
// 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.
ErrorOnPrompt throwing = new ErrorOnPrompt(new FakeHerdr());
Injector inj = new Injector(new AgentControl(throwing));
Injector.Delivery delivery = inj.enqueue(T, "brief", TestTurnTokens.inert(T));
@@ -734,9 +737,9 @@ class InjectorTest {
assertTrue(delivery.completion().isCompletedExceptionally(),
"the delivery's future must surface the send failure");
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");
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");
}
@Test
@@ -758,10 +761,14 @@ class InjectorTest {
}
@Test
void aHerdrExceptionFromSendStillProducesNotDeliveredUnchanged() {
void aHerdrExceptionFromSendStillSurfacesButNowReportsAttempted() {
// fleetd #546, acceptance test 3 (control): HerdrException extends RuntimeException, so it
// 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.
// 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.
FakeHerdr failing = new FakeHerdr().agentSendFailsWith("send_failed");
Injector inj = new Injector(new AgentControl(failing));
Injector.Delivery delivery = inj.enqueue(T, "boom", TestTurnTokens.inert(T));
@@ -771,9 +778,90 @@ 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),
"a HerdrException must still be dropped and marked NOT_DELIVERED, unchanged by the "
+ "wider Throwable catch");
"fleetd #551: a herdr *_not_found error is a confirmed absence, not merely "
+ "inconclusive — it must stay NOT_DELIVERED, not the new ATTEMPTED");
}
// --- fleetd #553: a throwable from any listener callback in onStatus must not skip the