Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Dai Ha
|
ea41bbf6b9 | ||
|
Dai Ha
|
823976c1b5 | ||
|
Dai Ha
|
c8388a7f92 | ||
|
Dai Ha
|
02e6aef98c | ||
|
Dai Ha
|
6d493bc7bb | ||
|
Dai Ha
|
e545c08082 | ||
|
Dai Ha
|
efa0deb9b2 | ||
|
Dai Ha
|
ca47e90c01 |
@@ -39,12 +39,32 @@ import java.util.function.Supplier;
|
||||
* keeps the old value until a restart: {@code lifecycle:}, {@code leadHeartbeat:},
|
||||
* {@code spawnReadyTimeoutMs} / {@code spawnReadyPollMs}, {@code quarantineCooldownSeconds}
|
||||
* (CB-578 stage B — baked once into the {@code BackendQuarantine} built at startup),
|
||||
* {@code guard:}, {@code worktreeRoot:}, adding or removing a profile (a new backend needs its own launcher,
|
||||
* {@code guard:}, {@code worktreeRoot:} and {@code worktreeGroup:} (both baked once into the
|
||||
* {@code GitWorktrees} built at {@code Fleetd.java:251} and never rebuilt — fleetd #323
|
||||
* instance 2 found {@code worktreeGroup} missing from this list and from
|
||||
* {@link #changedDeferredKeys}), {@code primary:} (fleetd #326 — {@code Fleetd.java:506, 519,
|
||||
* 520} read {@code cfg.primary()} only off the startup snapshot to build {@code
|
||||
* PrimaryRegistry} and size {@code ReplyPushLoop}'s reminder cap/backoff, and neither is
|
||||
* rebuilt on reload. Say the consequence exactly: {@code primary.terminal} is DEPRECATED
|
||||
* (CB-532, and {@code Fleetd.java:511} warns about it at startup) — a lead's identity comes
|
||||
* from {@code leaders:}/{@code leadScan:}, so changing this pin does not demote or promote a
|
||||
* lead that uses those. What a changed pin still does not take effect on until a restart is
|
||||
* the fallback nudge destination the pin remains, the deprecated identity path for an operator
|
||||
* who still relies on it, and {@code pushReminders}/{@code pushBackoffMs}), {@code configReload:} (fleetd #326 — {@code
|
||||
* Fleetd.java:679-680} read it only at startup to decide whether to build a {@code
|
||||
* ConfigWatcher} at all and with what interval; the watcher that would apply a later change is
|
||||
* itself built once, so a running watcher keeps polling on its original enabled flag and
|
||||
* interval regardless of what a reload changes it to, the same shape as {@code lifecycle} —
|
||||
* not cold, because no already-open resource goes inconsistent with the new value, the watcher
|
||||
* (if any) simply keeps its old settings), adding or removing a profile (a new backend needs its own launcher,
|
||||
* which is constructed once), <em>and an existing profile's launch settings</em> —
|
||||
* {@code model}, {@code baseUrl}, {@code argv}, {@code env}, {@code mcpUrl},
|
||||
* {@code exhaustedPattern} (CB-578 stage A — compiled once into {@code Fleetd.main}'s
|
||||
* pattern map at startup), {@code errorPattern} (fleetd #201 Unit 5 — compiled once into
|
||||
* {@code Fleetd.main}'s backend-error pattern map at startup, the same way), and the rest.
|
||||
* {@code Fleetd.main}'s backend-error pattern map at startup, the same way),
|
||||
* {@code ideProjectDir} / {@code ideOpenCommand} / {@code autoCompactWindow} (fleetd #323
|
||||
* instance 1 — all three are read at spawn off the same frozen profile map and were missing
|
||||
* from {@link #sameLaunchSettings}), and the rest of {@link #sameLaunchSettings}.
|
||||
* {@code credentialId} (CB-578 stage B) is NOT on
|
||||
* this list — it is read live off the config supplier at every quarantine check and
|
||||
* exhaustion event, exactly like {@code weight} / {@code maxLoad}, so it is hot instead.
|
||||
@@ -57,6 +77,24 @@ import java.util.function.Supplier;
|
||||
* listening.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p><strong>The denominator, measured on 2026-09-04 (fleetd #326).</strong> {@code FleetConfig} has
|
||||
* 22 top-level record components. Four of them are named nowhere in this file, and the reason
|
||||
* differs per key, so do not read "absent" as "forgotten":
|
||||
* <ul>
|
||||
* <li>{@code memberCredentials} and {@code memberLoginShell} are <strong>hot</strong> and
|
||||
* correctly absent — both are read live off {@code config.get()} at spawn time
|
||||
* ({@code Fleetd.java:198, 205, 729} and {@code HerdrPeerLauncher#configuredMemberLoginShell}),
|
||||
* so a reload takes effect on the next spawn with no entry needed here.</li>
|
||||
* <li>{@code health} and {@code coordinator} are <strong>undecided</strong>, not hot. Each is read
|
||||
* both ways at different sites, so neither fits the three classes above as a whole key. Until
|
||||
* that is settled a reload touching them reports a bare "config reloaded", which under-claims.
|
||||
* Deciding it is what a top-level coverage checker (the {@link ConfigRefProfileCoverageTest}
|
||||
* shape, one level up) is waiting on — without it, such a checker cannot express the answer.</li>
|
||||
* </ul>
|
||||
* The point of writing the count down: "not mentioned in this file" looks identical for a key that
|
||||
* is correctly hot and for a key nobody triaged. Twice now — {@code worktreeGroup} (#323) and
|
||||
* {@code primary}/{@code configReload} (#326) — the second kind hid among the first.
|
||||
*
|
||||
* <p><strong>A cold change refuses the whole reload.</strong> Not the hot half applied and the cold
|
||||
* half warned about: that would leave the running daemon in a state matching no file on disk, which
|
||||
* is the worst thing a reload can do to an operator debugging one. Refusing keeps the invariant that
|
||||
@@ -221,6 +259,29 @@ public final class ConfigRef implements Supplier<FleetConfig> {
|
||||
if (!Objects.equals(old.worktreeRoot(), fresh.worktreeRoot())) {
|
||||
changed.add("worktreeRoot");
|
||||
}
|
||||
// Baked into the same GitWorktrees as worktreeRoot (Fleetd.java:251) and never rebuilt
|
||||
// either — see the class doc. Missing this check was fleetd #323 instance 2: a reload
|
||||
// that changed only worktreeGroup reported "config reloaded" with nothing deferred, and
|
||||
// newly provisioned worktrees kept the old sharing behaviour.
|
||||
if (!Objects.equals(old.worktreeGroup(), fresh.worktreeGroup())) {
|
||||
changed.add("worktreeGroup");
|
||||
}
|
||||
// fleetd #326: Fleetd.java:506, 519, 520 read cfg.primary() only off the startup snapshot
|
||||
// (PrimaryRegistry's pinned terminal, ReplyPushLoop's reminder cap and backoff) — neither is
|
||||
// rebuilt on reload, so a changed value needs a restart. Note what it does NOT mean:
|
||||
// primary.terminal is deprecated (CB-532), identity comes from leaders:/leadScan:, so a lead
|
||||
// using those is unaffected by this pin either way. See the class doc for the exact scope.
|
||||
if (!Objects.equals(old.primary(), fresh.primary())) {
|
||||
changed.add("primary");
|
||||
}
|
||||
// fleetd #326: Fleetd.java:679-680 read cfg.configReload() only at startup to decide whether
|
||||
// to build a ConfigWatcher at all and with what interval — the watcher that would apply a
|
||||
// later change is itself built once, so a running watcher keeps its original enabled flag and
|
||||
// interval regardless of what a reload changes it to. Not cold: no already-open resource goes
|
||||
// inconsistent with the new value, a watcher (if any) simply keeps polling on the old settings.
|
||||
if (!Objects.equals(old.configReload(), fresh.configReload())) {
|
||||
changed.add("configReload");
|
||||
}
|
||||
if (!Objects.equals(old.spawnReadyTimeoutMs(), fresh.spawnReadyTimeoutMs())
|
||||
|| !Objects.equals(old.spawnReadyPollMs(), fresh.spawnReadyPollMs())) {
|
||||
changed.add("spawnReady*");
|
||||
@@ -265,14 +326,34 @@ public final class ConfigRef implements Supplier<FleetConfig> {
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether two versions of a profile would launch a peer identically. Compares every component
|
||||
* the launcher reads at spawn; {@code weight}, {@code maxLoad} and {@code credentialId} are
|
||||
* excluded because those are read live (by the placement policy and, for credentialId, by
|
||||
* {@code CompositePeerLauncher}/the CB-578 stage B exhaustion sink) and really do take effect on
|
||||
* the next spawn.
|
||||
* {@link FleetConfig.Profile} record components deliberately left out of
|
||||
* {@link #sameLaunchSettings} because they are read <em>live</em>, not baked in at spawn — see
|
||||
* the class doc's <em>Hot</em> bullet. {@code weight} and {@code maxLoad} are read live by the
|
||||
* placement policy on every spawn; {@code credentialId} is read live by
|
||||
* {@code CompositePeerLauncher} and the CB-578 stage B exhaustion sink. Nothing else is
|
||||
* excluded — see {@code sameLaunchSettingsComparesEveryProfileComponentOrExcludesIt} in
|
||||
* {@code ConfigRefProfileCoverageTest}, which enumerates every {@code Profile} record component
|
||||
* by reflection and fails the build if one is neither compared below nor named here.
|
||||
*/
|
||||
private static boolean sameLaunchSettings(FleetConfig.Profile a, FleetConfig.Profile b) {
|
||||
return Objects.equals(a.baseUrl(), b.baseUrl())
|
||||
static final Set<String> LAUNCH_SETTINGS_EXCLUDED = Set.of("weight", "maxLoad", "credentialId");
|
||||
|
||||
/**
|
||||
* Whether two versions of a profile would launch a peer identically.
|
||||
*
|
||||
* <p>This must compare every {@link FleetConfig.Profile} record component except the three in
|
||||
* {@link #LAUNCH_SETTINGS_EXCLUDED}. That is not a claim this javadoc can make good on by
|
||||
* itself — a javadoc saying "compares every component" is exactly what fleetd #323 found to be
|
||||
* false for three fields (and a sibling method's field list, for a fourth). The actual
|
||||
* guarantee comes from {@code ConfigRefProfileCoverageTest}: it enumerates every record
|
||||
* component of {@code FleetConfig.Profile} by reflection, mutates each one not in
|
||||
* {@code LAUNCH_SETTINGS_EXCLUDED} on a base profile, and asserts this method reports a
|
||||
* difference — so a new component that is neither compared here nor added to
|
||||
* {@code LAUNCH_SETTINGS_EXCLUDED} (with a reason) fails that test by name, rather than
|
||||
* silently reporting "config reloaded" for a value the daemon never picked up.
|
||||
*/
|
||||
static boolean sameLaunchSettings(FleetConfig.Profile a, FleetConfig.Profile b) {
|
||||
return Objects.equals(a.profile(), b.profile())
|
||||
&& Objects.equals(a.baseUrl(), b.baseUrl())
|
||||
&& Objects.equals(a.model(), b.model())
|
||||
&& Objects.equals(a.configDir(), b.configDir())
|
||||
&& Objects.equals(a.tokenEnv(), b.tokenEnv())
|
||||
@@ -298,6 +379,16 @@ public final class ConfigRef implements Supplier<FleetConfig> {
|
||||
// fleetd #201 Unit 5: errorPattern is compiled once into Fleetd.main's backend-error
|
||||
// pattern map at startup (see BackendErrorPatternLookup wiring), the same way
|
||||
// exhaustedPattern is — a reload never re-reads it either.
|
||||
&& Objects.equals(a.errorPattern(), b.errorPattern());
|
||||
&& Objects.equals(a.errorPattern(), b.errorPattern())
|
||||
// fleetd #323 instance 1: ideProjectDir and ideOpenCommand are read at spawn off the
|
||||
// same frozen profile map as ideMcpUrl above (ClaudeCodeLauncher.java:267/269,
|
||||
// OpenCodeLauncher.java:474/480/486) and were missing from this comparison.
|
||||
&& Objects.equals(a.ideProjectDir(), b.ideProjectDir())
|
||||
&& Objects.equals(a.ideOpenCommand(), b.ideOpenCommand())
|
||||
// fleetd #323 instance 1: autoCompactWindow is read at spawn the same way
|
||||
// (ClaudeCodeLauncher.java:926, OpenCodeLauncher.java:650). Comparing it here only
|
||||
// makes the reload REPORT that a restart is needed — it deliberately does not make
|
||||
// autoCompactWindow take effect live, which is a separate, larger change.
|
||||
&& Objects.equals(a.autoCompactWindow(), b.autoCompactWindow());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -466,8 +466,18 @@ public final class MessageService {
|
||||
if (candidates.size() == 1) {
|
||||
Task orphan = candidates.get(0);
|
||||
if (orphan.future.complete(new Reply(Outcome.REPLIED, content))) {
|
||||
if (orphan.turnId != null) {
|
||||
asyncTasksByTurn.remove(orphan.turnId, orphan);
|
||||
// fleetd #329 (F3): read orphan.turnId once. It used to be read twice, under no
|
||||
// lock — once for this null check, once as the removal key — the same double-read
|
||||
// shape fleetd #324 fixed in finishAsyncTask. clearAsyncQuestion's unlocked
|
||||
// forgetTurn=true path (ask()'s timeout cleanup) can null the field between the two
|
||||
// reads; capturing it once removes the torn read here too.
|
||||
String turnId = orphan.turnId;
|
||||
if (turnId != null) {
|
||||
if (replyOrphanTurnIdRaceHookForTest != null) {
|
||||
// Test-only (fleetd #329, F3): see the field's own javadoc.
|
||||
replyOrphanTurnIdRaceHookForTest.run();
|
||||
}
|
||||
asyncTasksByTurn.remove(turnId, orphan);
|
||||
}
|
||||
count(FleetMetrics.REPLIES, "path", "async-recovered");
|
||||
return true; // the ticket itself took it — no inbox stranding at all
|
||||
@@ -1002,13 +1012,36 @@ public final class MessageService {
|
||||
// Measured when #282 was merged: this guard is DEFENCE IN DEPTH, not the thing
|
||||
// that makes the chained ask work. ask() calls markAsyncQuestion (:860) before
|
||||
// resolveQuestion (:861), so by the time this thread wakes, the task has already
|
||||
// moved to the new turnId and finishAsyncTask(oldTurnId, ...) finds nothing. Removing
|
||||
// this guard alone leaves the test green. Keep it anyway: it mirrors sendAsync's
|
||||
// sibling guard, and that sibling's own comment (:1017) warns the two orderings are
|
||||
// not something to rely on. Do NOT delete it as dead code without re-checking that
|
||||
// ordering, and do not treat it as the sole protection either.
|
||||
if (result.outcome() != Outcome.QUESTION) {
|
||||
finishAsyncTask(turnId, result);
|
||||
// moved to the new turnId — this outcome check, not a Task lookup, is what tells the
|
||||
// two cases apart (see fleetd #329 below). Removing this guard alone leaves the test
|
||||
// green. Keep it anyway: it mirrors sendAsync's sibling guard, and that sibling's own
|
||||
// comment (:1017) warns the two orderings are not something to rely on. Do NOT delete
|
||||
// it as dead code without re-checking that ordering, and do not treat it as the sole
|
||||
// protection either.
|
||||
//
|
||||
// fleetd #329 (F1): complete the SAME Task object this method already looked up at
|
||||
// :991, rather than re-resolving it from turnId a second time. The old
|
||||
// finishAsyncTask(turnId, result) did its own asyncTasksByTurn.get(turnId) here, and
|
||||
// that second lookup races ask()'s own timeout path: ask()'s ticket.answer().get(...)
|
||||
// can time out (or lose that exact race) at essentially the same instant this method's
|
||||
// rendezvous.answerAsk(turnId, ...) above already succeeded, running
|
||||
// markAskTimedOut + clearAsyncQuestion(turnId, true) with no lock at all — which
|
||||
// forgets turnId (removes it from asyncTasksByTurn, nulls Task.turnId) before this
|
||||
// thread ever gets here. The worker's real reply then arrived, this method's own wait
|
||||
// woke up with it, and the by-turnId lookup found nothing: the async ticket's future
|
||||
// was never completed, so fleet_poll{ticket} stayed PENDING forever even though
|
||||
// answer() itself correctly returned REPLIED. Reusing the reference captured at :991
|
||||
// — before that race window opens — sidesteps the second lookup entirely: it is the
|
||||
// identical Task whatever asyncTasksByTurn or Task.turnId say by the time we reach
|
||||
// this point, so finishAsyncTask(Task, Reply) can still complete its future and detach
|
||||
// it using whatever turnId it now reads. The #282 chained-ask case is unaffected
|
||||
// because it is still gated purely by result.outcome() == QUESTION above, which does
|
||||
// not depend on this lookup — widening what "task" means here cannot complete a ticket
|
||||
// the chained ask deliberately left open. When task is null (this was never an async
|
||||
// ticket — a blocking fleet_ask's answer() call has no Task at all), there is nothing
|
||||
// to complete, matching the old lookup-miss behaviour.
|
||||
if (result.outcome() != Outcome.QUESTION && task != null) {
|
||||
finishAsyncTask(task, result);
|
||||
}
|
||||
return result;
|
||||
} catch (TimeoutException e) {
|
||||
@@ -1059,7 +1092,7 @@ public final class MessageService {
|
||||
// this fires exactly once, from whichever path completes it: finishAsyncTask(task, result)
|
||||
// below on any non-QUESTION outcome of send() — a worker's fleet_reply, the CB-106
|
||||
// completion fallback, a CB-109 wedge, TIMED_OUT, BUSY, or BACKEND_EXHAUSTED — the same
|
||||
// finishAsyncTask reached via answer()'s finishAsyncTask(turnId, result) once a QUESTION
|
||||
// finishAsyncTask reached via answer()'s own finishAsyncTask(task, result) once a QUESTION
|
||||
// is resolved, completeExceptionally(t) just below when send() itself throws, or a CB-516
|
||||
// abandon() on teardown. Without this, MessageService.reply's rendezvous fast path (the
|
||||
// one an async ticket always takes) never told the push loop anything happened — see the
|
||||
@@ -1079,7 +1112,22 @@ public final class MessageService {
|
||||
finishAsyncTask(task, result);
|
||||
}
|
||||
} catch (Throwable t) {
|
||||
task.future.completeExceptionally(t);
|
||||
// fleetd #329 (F2): finishAsyncTask above already completes task.future — on its very
|
||||
// first line — before doing anything else, so anything that throws afterward (inside
|
||||
// finishAsyncTask's own cleanup, or from a future addition to this try block) lands
|
||||
// here with the future already resolved. completeExceptionally on an already-completed
|
||||
// future is a silent no-op: it returns false and does nothing, so without the check
|
||||
// below the exception simply vanished — no log, no metric, nothing. Measured (see the
|
||||
// ticket): temporarily reintroducing the fleetd #324 NPE reproduced 19 real exceptions
|
||||
// on the ordinary path, with 74/74 tests staying green and not one log line produced.
|
||||
// Do not stop completing the future first — finishAsyncTask completing before it
|
||||
// cleans up is what makes a late failure harmless to the ticket's own result — only
|
||||
// add the missing visibility for the case where that step, or whatever ran after it,
|
||||
// has already lost the race to report through the future.
|
||||
if (!task.future.completeExceptionally(t)) {
|
||||
log.error("async send {} -> {} threw after its ticket was already resolved",
|
||||
ticket, target, t);
|
||||
}
|
||||
}
|
||||
});
|
||||
pruneTerminalTickets();
|
||||
@@ -1247,6 +1295,10 @@ public final class MessageService {
|
||||
*/
|
||||
private void finishAsyncTask(Task task, Reply result) {
|
||||
task.future.complete(result);
|
||||
if (afterFinishAsyncTaskCompleteHookForTest != null) {
|
||||
// Test-only (fleetd #329, F2): see the field's own javadoc.
|
||||
afterFinishAsyncTaskCompleteHookForTest.run();
|
||||
}
|
||||
String turnId = task.turnId;
|
||||
if (turnId != null) {
|
||||
if (finishAsyncTaskRaceHook != null) {
|
||||
@@ -1276,6 +1328,49 @@ public final class MessageService {
|
||||
this.finishAsyncTaskRaceHook = hook;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null in production; test seam for fleetd #329 (F2) — invoked from {@link #finishAsyncTask(Task,
|
||||
* Reply)} unconditionally, immediately after {@code task.future.complete(result)} runs (before
|
||||
* {@link Task#turnId} is even read, so it fires regardless of whether this task was ever asked).
|
||||
* A test installs this to force an exception into exactly the shape fleetd #329 identified:
|
||||
* something throws inside {@link #sendAsync}'s executor task after the async ticket's future is
|
||||
* already resolved, so the surrounding {@code catch (Throwable t)} can only report failure through
|
||||
* {@code completeExceptionally} — a silent no-op on an already-completed future. Engineering a
|
||||
* real exception to land in that exact post-completion window is what the ticket itself had to do
|
||||
* by temporarily deleting a production guard (fleetd #324's {@code turnId != null} check); this
|
||||
* hook drives the identical shape deterministically instead.
|
||||
*/
|
||||
private volatile Runnable afterFinishAsyncTaskCompleteHookForTest;
|
||||
|
||||
/**
|
||||
* Test-only (fleetd #329, F2): install {@link #afterFinishAsyncTaskCompleteHookForTest}.
|
||||
* Package-private so the test, in the same package, can reach it without widening any production
|
||||
* API.
|
||||
*/
|
||||
void setAfterFinishAsyncTaskCompleteHookForTest(Runnable hook) {
|
||||
this.afterFinishAsyncTaskCompleteHookForTest = hook;
|
||||
}
|
||||
|
||||
/**
|
||||
* Null in production; test seam for fleetd #329 (F3) — invoked from {@link #reply} right after
|
||||
* the single local read of {@code orphan.turnId} passes its null-check and before that (now-local)
|
||||
* value is used as the {@code asyncTasksByTurn} removal key. Mirrors {@link
|
||||
* #finishAsyncTaskRaceHook} exactly, for the structurally identical double-read fleetd #324 fixed
|
||||
* in {@link #finishAsyncTask}: a test installs this to force, deterministically, {@code
|
||||
* clearAsyncQuestion}'s unlocked {@code forgetTurn=true} path nulling {@link Task#turnId} in that
|
||||
* exact window, and to confirm the single-read fix tolerates it (the captured local is used
|
||||
* unconditionally, so a hook that nulls the field afterward cannot affect this call).
|
||||
*/
|
||||
private volatile Runnable replyOrphanTurnIdRaceHookForTest;
|
||||
|
||||
/**
|
||||
* Test-only (fleetd #329, F3): install {@link #replyOrphanTurnIdRaceHookForTest}. Package-private
|
||||
* so the test, in the same package, can reach it without widening any production API.
|
||||
*/
|
||||
void setReplyOrphanTurnIdRaceHookForTest(Runnable hook) {
|
||||
this.replyOrphanTurnIdRaceHookForTest = hook;
|
||||
}
|
||||
|
||||
/**
|
||||
* Test-only (fleetd #324): run the exact production cleanup {@link #ask}'s own timeout path runs
|
||||
* unlocked — {@link #clearAsyncQuestion(String, boolean)} with {@code forgetTurn=true} — so a test
|
||||
@@ -1285,14 +1380,6 @@ public final class MessageService {
|
||||
clearAsyncQuestion(turnId, true);
|
||||
}
|
||||
|
||||
/** Complete the async ticket correlated to a specific answered turn. */
|
||||
private void finishAsyncTask(String turnId, Reply result) {
|
||||
Task task = asyncTasksByTurn.get(turnId);
|
||||
if (task != null) {
|
||||
finishAsyncTask(task, result);
|
||||
}
|
||||
}
|
||||
|
||||
/** A new send must not open a waiter while an async ticket owns this worker's paused turn. */
|
||||
private boolean hasAsyncQuestion(String target) {
|
||||
return asyncTasksByTurn.values().stream().anyMatch(task -> target.equals(task.target));
|
||||
|
||||
@@ -0,0 +1,202 @@
|
||||
package dev.ltms.fleet.config;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.lang.reflect.Constructor;
|
||||
import java.lang.reflect.RecordComponent;
|
||||
import java.util.Arrays;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
/**
|
||||
* fleetd #323: {@code ConfigRef.sameLaunchSettings} javadoc used to claim it "compares every
|
||||
* component the launcher reads at spawn". It did not — {@code ideProjectDir}, {@code
|
||||
* ideOpenCommand} and {@code autoCompactWindow} were all baked in at daemon startup (see
|
||||
* {@code ClaudeCodeLauncher}/{@code OpenCodeLauncher}) and missing from the comparison, so a reload
|
||||
* that changed only one of them reported "config reloaded" and the running daemon kept the old
|
||||
* value.
|
||||
*
|
||||
* <p>This class is the mechanism the issue asked for: it enumerates every record component of
|
||||
* {@link FleetConfig.Profile} by reflection and proves — by actually mutating a base profile one
|
||||
* field at a time and calling the real method — that each component is either compared by
|
||||
* {@link ConfigRef#sameLaunchSettings} or named in {@link ConfigRef#LAUNCH_SETTINGS_EXCLUDED} with
|
||||
* a reason. A new profile field that is neither fails this test by name, not a hand-maintained list
|
||||
* going stale.
|
||||
*/
|
||||
class ConfigRefProfileCoverageTest {
|
||||
|
||||
private static final RecordComponent[] COMPONENTS = FleetConfig.Profile.class.getRecordComponents();
|
||||
|
||||
/**
|
||||
* One valid, non-blank value per record component — "the a value". None of these trip any
|
||||
* defaulting/normalization in {@code Profile}'s compact constructor (see {@code
|
||||
* FleetConfig.java}), so what goes in is what {@code sameLaunchSettings} sees back out.
|
||||
*/
|
||||
private static final Map<String, Object> BASE = baseValues();
|
||||
|
||||
/** The same shape, each value distinct from {@link #BASE} — "the b value". */
|
||||
private static final Map<String, Object> ALT = altValues();
|
||||
|
||||
private static Map<String, Object> baseValues() {
|
||||
Map<String, Object> v = new LinkedHashMap<>();
|
||||
v.put("profile", "sonnet");
|
||||
v.put("baseUrl", "http://gx00.gw:8000");
|
||||
v.put("model", "sonnet");
|
||||
v.put("configDir", "/config/a");
|
||||
v.put("tokenEnv", "TOKEN_A");
|
||||
v.put("argv", List.of("claude", "--flag-a"));
|
||||
v.put("placement", "tab");
|
||||
v.put("workspace", "workspace-a");
|
||||
v.put("tabLabel", "label-a");
|
||||
v.put("mcpUrl", "http://mcp-a");
|
||||
v.put("cwd", "/cwd/a");
|
||||
v.put("parityOverlay", List.of(".env", ".env.a"));
|
||||
v.put("gitTokenEnv", "GIT_TOKEN_A");
|
||||
v.put("gitHostEnv", "GITEA_HOST_A");
|
||||
v.put("kind", "claude-code");
|
||||
v.put("env", Map.of("K", "A"));
|
||||
v.put("weight", 1.0f);
|
||||
v.put("maxLoad", 5);
|
||||
v.put("subscription", Boolean.TRUE);
|
||||
v.put("exhaustedPattern", "usage limit a");
|
||||
v.put("credentialId", "cred-a");
|
||||
v.put("ideMcpUrl", "http://ide-mcp-a");
|
||||
v.put("ideProjectDir", "modules/a");
|
||||
v.put("ideOpenCommand", "open-cmd-a {dir}");
|
||||
v.put("autoCompactWindow", 150000);
|
||||
v.put("errorPattern", "error a");
|
||||
assertNamesMatchComponents(v);
|
||||
return v;
|
||||
}
|
||||
|
||||
private static Map<String, Object> altValues() {
|
||||
Map<String, Object> v = new LinkedHashMap<>();
|
||||
v.put("profile", "sonnet-b");
|
||||
v.put("baseUrl", "http://gx01.gw:8000");
|
||||
v.put("model", "haiku");
|
||||
v.put("configDir", "/config/b");
|
||||
v.put("tokenEnv", "TOKEN_B");
|
||||
v.put("argv", List.of("claude", "--flag-b"));
|
||||
v.put("placement", "weighted");
|
||||
v.put("workspace", "workspace-b");
|
||||
v.put("tabLabel", "label-b");
|
||||
v.put("mcpUrl", "http://mcp-b");
|
||||
v.put("cwd", "/cwd/b");
|
||||
v.put("parityOverlay", List.of(".env", ".env.b"));
|
||||
v.put("gitTokenEnv", "GIT_TOKEN_B");
|
||||
v.put("gitHostEnv", "GITEA_HOST_B");
|
||||
v.put("kind", "opencode");
|
||||
v.put("env", Map.of("K", "B"));
|
||||
v.put("weight", 2.0f);
|
||||
v.put("maxLoad", 9);
|
||||
v.put("subscription", Boolean.FALSE);
|
||||
v.put("exhaustedPattern", "usage limit b");
|
||||
v.put("credentialId", "cred-b");
|
||||
v.put("ideMcpUrl", "http://ide-mcp-b");
|
||||
v.put("ideProjectDir", "modules/b");
|
||||
v.put("ideOpenCommand", "open-cmd-b {dir}");
|
||||
v.put("autoCompactWindow", 250000);
|
||||
v.put("errorPattern", "error b");
|
||||
assertNamesMatchComponents(v);
|
||||
return v;
|
||||
}
|
||||
|
||||
private static void assertNamesMatchComponents(Map<String, Object> values) {
|
||||
Set<String> componentNames = new TreeSet<>();
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
componentNames.add(rc.getName());
|
||||
}
|
||||
assertEquals(componentNames, new TreeSet<>(values.keySet()),
|
||||
"this test's value map has drifted from FleetConfig.Profile's actual components — "
|
||||
+ "update BASE/ALT alongside the record");
|
||||
}
|
||||
|
||||
private static FleetConfig.Profile profileOf(Map<String, Object> values) throws ReflectiveOperationException {
|
||||
Class<?>[] types = Arrays.stream(COMPONENTS).map(RecordComponent::getType).toArray(Class<?>[]::new);
|
||||
Object[] args = Arrays.stream(COMPONENTS)
|
||||
.map(rc -> values.get(rc.getName()))
|
||||
.toArray();
|
||||
Constructor<FleetConfig.Profile> ctor = FleetConfig.Profile.class.getDeclaredConstructor(types);
|
||||
return ctor.newInstance(args);
|
||||
}
|
||||
|
||||
/** {@code BASE} with exactly one named component swapped for its {@code ALT} value. */
|
||||
private static FleetConfig.Profile mutate(String componentName) throws ReflectiveOperationException {
|
||||
Map<String, Object> values = new LinkedHashMap<>(BASE);
|
||||
values.put(componentName, ALT.get(componentName));
|
||||
return profileOf(values);
|
||||
}
|
||||
|
||||
/**
|
||||
* The mechanism fleetd #323 asked for: enumerate {@link FleetConfig.Profile}'s record
|
||||
* components, mutate each non-excluded one, and prove {@code sameLaunchSettings} actually
|
||||
* notices — not just that some hand-maintained list claims it does. Prints the denominator
|
||||
* (total / compared / excluded) the issue required: a checker that cannot state its own
|
||||
* denominator is the failure this repo keeps hitting.
|
||||
*/
|
||||
@Test
|
||||
void sameLaunchSettingsComparesEveryProfileComponentOrExcludesIt() throws ReflectiveOperationException {
|
||||
int total = COMPONENTS.length;
|
||||
Set<String> excluded = ConfigRef.LAUNCH_SETTINGS_EXCLUDED;
|
||||
|
||||
Set<String> allNames = new TreeSet<>();
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
allNames.add(rc.getName());
|
||||
}
|
||||
assertTrue(allNames.containsAll(excluded),
|
||||
"ConfigRef.LAUNCH_SETTINGS_EXCLUDED names a component that does not exist on "
|
||||
+ "FleetConfig.Profile — check for a typo: " + excluded);
|
||||
|
||||
// The exclusion set is this mechanism's own escape hatch, so it has to be pinned too.
|
||||
// Found by mutation while verifying fleetd #323: moving autoCompactWindow and
|
||||
// ideOpenCommand OUT of the comparison and INTO the exclusion set left the whole suite
|
||||
// green — the loop below simply skips them, and the denominator assertion still balances.
|
||||
// That is exactly the lazy move a failing coverage test invites, and it silently restores
|
||||
// the #323 bug. Only ideProjectDir and worktreeGroup were saved by a behavioural test in
|
||||
// ConfigRefTest; the other two had none. So: growing this set now requires editing this
|
||||
// line as well, which is a visible, deliberate diff rather than a quiet one.
|
||||
assertEquals(Set.of("weight", "maxLoad", "credentialId"), excluded,
|
||||
"ConfigRef.LAUNCH_SETTINGS_EXCLUDED changed. A component belongs in it ONLY if it "
|
||||
+ "is read live off the config supplier, not baked into a launcher at "
|
||||
+ "startup. If you are adding one to silence this test, that is fleetd #323 "
|
||||
+ "happening again: compare it in sameLaunchSettings instead. If it really "
|
||||
+ "is read live, name where it is read and update this assertion.");
|
||||
|
||||
FleetConfig.Profile base = profileOf(BASE);
|
||||
List<String> uncovered = new java.util.ArrayList<>();
|
||||
int compared = 0;
|
||||
for (RecordComponent rc : COMPONENTS) {
|
||||
String name = rc.getName();
|
||||
if (excluded.contains(name)) {
|
||||
continue;
|
||||
}
|
||||
FleetConfig.Profile mutated = mutate(name);
|
||||
if (ConfigRef.sameLaunchSettings(base, mutated)) {
|
||||
uncovered.add(name);
|
||||
} else {
|
||||
compared++;
|
||||
}
|
||||
}
|
||||
|
||||
System.out.printf(
|
||||
"ConfigRef.sameLaunchSettings coverage — %d Profile components total, %d compared, "
|
||||
+ "%d excluded (%s)%n",
|
||||
total, compared, excluded.size(), excluded);
|
||||
|
||||
assertEquals(List.of(), uncovered,
|
||||
"these FleetConfig.Profile components changed but ConfigRef.sameLaunchSettings "
|
||||
+ "reported no difference — add each one to the comparison (it is read at "
|
||||
+ "spawn and baked in until a restart) or to ConfigRef.LAUNCH_SETTINGS_EXCLUDED "
|
||||
+ "with a reason it is genuinely read live: " + uncovered);
|
||||
assertEquals(total, compared + excluded.size(),
|
||||
"every FleetConfig.Profile record component must be either compared or excluded — "
|
||||
+ total + " components, " + compared + " compared, " + excluded.size()
|
||||
+ " excluded");
|
||||
}
|
||||
}
|
||||
@@ -434,6 +434,142 @@ class ConfigRefTest {
|
||||
assertEquals("provider 5xx", ref.get().profiles().get("sonnet").errorPattern());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #323 instance 1: {@code ideProjectDir} is read at spawn off the frozen profile map
|
||||
* (see {@code ClaudeCodeLauncher}/{@code OpenCodeLauncher}) exactly like {@code model}, but was
|
||||
* missing from {@code sameLaunchSettings} — a reload changing only this field used to report a
|
||||
* bare "config reloaded" and the running daemon kept launching with the old value.
|
||||
*/
|
||||
@Test
|
||||
void changingAProfilesIdeProjectDirIsReportedAsDeferred(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("fleetd.yaml");
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
host: 127.0.0.1
|
||||
port: 8765
|
||||
herdrSocket: ~/.config/herdr/herdr.sock
|
||||
profiles:
|
||||
sonnet:
|
||||
baseUrl: http://gx00.gw:8000
|
||||
model: sonnet
|
||||
ideProjectDir: fleetd
|
||||
guard:
|
||||
offSubscriptionHosts:
|
||||
- gx00.gw
|
||||
""");
|
||||
ConfigRef ref = refFor(f);
|
||||
|
||||
Files.writeString(f, """
|
||||
bind:
|
||||
host: 127.0.0.1
|
||||
port: 8765
|
||||
herdrSocket: ~/.config/herdr/herdr.sock
|
||||
profiles:
|
||||
sonnet:
|
||||
baseUrl: http://gx00.gw:8000
|
||||
model: sonnet
|
||||
ideProjectDir: fleetd-renamed
|
||||
guard:
|
||||
offSubscriptionHosts:
|
||||
- gx00.gw
|
||||
""");
|
||||
ConfigRef.Outcome out = ref.reload();
|
||||
|
||||
assertTrue(out.applied());
|
||||
assertEquals(1, out.deferred().size(), out.deferred().toString());
|
||||
assertTrue(out.deferred().getFirst().contains("sonnet"), out.deferred().toString());
|
||||
assertTrue(out.deferred().getFirst().contains("launch settings"), out.deferred().toString());
|
||||
// The snapshot still carries the new value — a restart is what makes it take effect.
|
||||
assertEquals("fleetd-renamed", ref.get().profiles().get("sonnet").ideProjectDir());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #323 instance 2: {@code worktreeGroup} is baked into the same {@code GitWorktrees}
|
||||
* as {@code worktreeRoot} (Fleetd.java:251) and never rebuilt, but only {@code worktreeRoot}
|
||||
* was on {@code changedDeferredKeys} — a reload changing only the group reported a bare
|
||||
* "config reloaded" and newly provisioned worktrees kept the old sharing behaviour.
|
||||
*/
|
||||
@Test
|
||||
void changingWorktreeGroupIsReportedAsDeferred(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("fleetd.yaml");
|
||||
Files.writeString(f, yaml("worktreeGroup: devgroup\n"));
|
||||
ConfigRef ref = refFor(f);
|
||||
|
||||
Files.writeString(f, yaml("worktreeGroup: devgroup2\n"));
|
||||
ConfigRef.Outcome out = ref.reload();
|
||||
|
||||
assertTrue(out.applied());
|
||||
assertEquals(java.util.List.of("worktreeGroup"), out.deferred());
|
||||
assertTrue(out.summary().contains("needs a restart") || out.summary().contains("need a restart"),
|
||||
out.summary());
|
||||
// The snapshot still carries the new value — a restart is what makes it take effect.
|
||||
assertEquals("devgroup2", ref.get().worktreeGroup());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #326: {@code primary} is read only off the startup snapshot — {@code Fleetd.java:506,
|
||||
* 519, 520} feed {@code PrimaryRegistry} and {@code ReplyPushLoop} at construction and neither is
|
||||
* rebuilt on reload — but it was missing from {@link ConfigRef#changedDeferredKeys}, so a reload
|
||||
* that only changed the pinned primary terminal reported a bare "config reloaded" while a lead
|
||||
* whose tab no longer matched stayed demoted to worker.
|
||||
*/
|
||||
@Test
|
||||
void changingPrimaryIsReportedAsDeferred(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("fleetd.yaml");
|
||||
Files.writeString(f, yaml("""
|
||||
primary:
|
||||
terminal: term-a
|
||||
"""));
|
||||
ConfigRef ref = refFor(f);
|
||||
|
||||
Files.writeString(f, yaml("""
|
||||
primary:
|
||||
terminal: term-b
|
||||
"""));
|
||||
ConfigRef.Outcome out = ref.reload();
|
||||
|
||||
assertTrue(out.applied());
|
||||
assertEquals(java.util.List.of("primary"), out.deferred());
|
||||
assertTrue(out.summary().contains("need") && out.summary().contains("restart"), out.summary());
|
||||
// The snapshot still carries the new value — a restart is what makes it take effect.
|
||||
assertEquals("term-b", ref.get().primary().terminal());
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #326: {@code configReload} itself is read only at startup ({@code Fleetd.java:679-680})
|
||||
* to decide whether to build a {@code ConfigWatcher} at all, and with what interval — the watcher
|
||||
* that would apply a later change is itself built once, so it is deferred rather than cold (see
|
||||
* {@link ConfigRef}'s class doc: cold means an already-open resource would go inconsistent with
|
||||
* the new value, and there is no such resource here — a running watcher just keeps polling on its
|
||||
* original enabled/interval until a restart, exactly like {@code lifecycle} or {@code guard}).
|
||||
* Before this fix, turning reload off (or changing its interval) through a reload reported a bare
|
||||
* "config reloaded" — the obvious joke the issue names.
|
||||
*/
|
||||
@Test
|
||||
void changingConfigReloadIsReportedAsDeferred(@TempDir Path dir) throws Exception {
|
||||
Path f = dir.resolve("fleetd.yaml");
|
||||
Files.writeString(f, yaml("""
|
||||
configReload:
|
||||
enabled: true
|
||||
intervalSeconds: 10
|
||||
"""));
|
||||
ConfigRef ref = refFor(f);
|
||||
|
||||
Files.writeString(f, yaml("""
|
||||
configReload:
|
||||
enabled: false
|
||||
intervalSeconds: 30
|
||||
"""));
|
||||
ConfigRef.Outcome out = ref.reload();
|
||||
|
||||
assertTrue(out.applied());
|
||||
assertEquals(java.util.List.of("configReload"), out.deferred());
|
||||
assertTrue(out.summary().contains("need") && out.summary().contains("restart"), out.summary());
|
||||
// The snapshot still carries the new value — a restart is what makes it take effect.
|
||||
assertFalse(ref.get().configReload().isEnabled());
|
||||
assertEquals(30, ref.get().configReload().intervalSeconds());
|
||||
}
|
||||
|
||||
@Test
|
||||
void aFixedRefHasNoFileAndRefusesToReload() {
|
||||
FleetConfig cfg = new FleetConfig(null, null, null, null, null, null,
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
package dev.ltms.fleet.msg;
|
||||
|
||||
import ch.qos.logback.classic.Level;
|
||||
import ch.qos.logback.classic.Logger;
|
||||
import ch.qos.logback.classic.spi.ILoggingEvent;
|
||||
import ch.qos.logback.core.read.ListAppender;
|
||||
import dev.ltms.fleet.herdr.AgentControl;
|
||||
import dev.ltms.fleet.herdr.AgentStatus;
|
||||
import dev.ltms.fleet.herdr.FakeHerdr;
|
||||
@@ -11,6 +15,7 @@ import dev.ltms.fleet.mcp.PrimaryRegistry;
|
||||
import dev.ltms.fleet.inject.Injector;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@@ -1041,6 +1046,105 @@ class MessageServiceTest {
|
||||
"the reply completed its own ticket directly and never touched the inbox");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #329 (F1). {@code answer()} completes the async ticket by looking {@code turnId} up in
|
||||
* {@code asyncTasksByTurn} a SECOND time (the first is at :991, purely to re-register the
|
||||
* {@code asyncTasksByWaiter} entry for #282's chained-ask case). That second lookup races
|
||||
* {@code ask()}'s own unlocked timeout cleanup ({@code clearAsyncQuestion(turnId, true)}, run from
|
||||
* {@code markAskTimedOut} + forgetting): {@code ask()}'s {@code ticket.answer().get(timeoutMillis)}
|
||||
* can time out at essentially the same instant {@code answer()}'s {@code rendezvous.answerAsk}
|
||||
* call above already succeeded and unblocked the worker. fleetd #324's single-read fix does not
|
||||
* help here — that fixed a torn read of one already-held {@link MessageService} internal
|
||||
* {@code Task}; this is a second, independent map lookup by a caller that no longer holds the
|
||||
* {@code Task} it already found once.
|
||||
*
|
||||
* <p>This test does not wait for that real race to land on its own schedule — it drives the exact
|
||||
* sequence the ticket describes (worker asks, primary answers, worker's real reply arrives) and
|
||||
* fires the identical production cleanup {@code ask()}'s timeout path runs
|
||||
* ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd #324
|
||||
* already merged) at the point between "the worker's own {@code ask()} call has unblocked" and
|
||||
* "the worker's real {@code fleet_reply} arrives" — the exact window fleetd #329 names.
|
||||
*
|
||||
* <p>What this proves: given that exact interleaving, the async ticket must still resolve to the
|
||||
* worker's real reply, not stay stuck {@link MessageService.Phase#PENDING} forever. What it does
|
||||
* not prove: that the interleaving itself is reachable in production on its own timing — that is
|
||||
* established by reading the code (see the ticket's "path in"), not by this test, for the same
|
||||
* reason fleetd #324's own race test says so.
|
||||
*/
|
||||
@Test
|
||||
void aReplyRacingAsksTimeoutCleanupStillCompletesTheAsyncTicket() throws Exception {
|
||||
String ticket = messages.sendAsync(T, "task that asks");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
|
||||
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
|
||||
String turnId = asking.turnId();
|
||||
|
||||
CompletableFuture<MessageService.Reply> answer =
|
||||
CompletableFuture.supplyAsync(() -> messages.answer(turnId, "config.yaml", 5000));
|
||||
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer(),
|
||||
"the worker's own ask() call must have already unblocked with the primary's answer "
|
||||
+ "before we force the race below");
|
||||
|
||||
// ask()'s own unlocked timeout cleanup can forget this exact turnId at essentially the same
|
||||
// instant answer() has already unblocked the worker and is now waiting on the resumed turn's
|
||||
// real reply — reproduce that interleaving directly instead of trying to win a real race.
|
||||
messages.forgetTurnForTest(turnId);
|
||||
|
||||
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
|
||||
|
||||
assertEquals(MessageService.Outcome.REPLIED, answer.get(5, TimeUnit.SECONDS).outcome(),
|
||||
"the primary's own answer() call must still see the worker's real reply");
|
||||
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
|
||||
assertEquals("PR opened: https://example/pulls/42", done.reply(),
|
||||
"fleet_poll{ticket} must return the worker's real reply, not stay PENDING forever "
|
||||
+ "just because ask()'s timeout cleanup forgot this turnId first");
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #329 (F3). {@link MessageService#reply} reads the same orphan {@code Task}'s {@code
|
||||
* turnId} twice on this path (once to check it is non-null, once as the {@code
|
||||
* asyncTasksByTurn.remove} key) — the exact double-read shape fleetd #324 fixed in {@code
|
||||
* finishAsyncTask}. This test drives the same real sequence as {@code
|
||||
* aReplyAfterAnswerTimesOutStillCompletesTheAsyncTicket} (worker asks, primary answers, the
|
||||
* primary's own bounded wait for the resumed turn expires) to reach a task with {@code turnId}
|
||||
* genuinely stamped and no live rendezvous waiter open — the state {@code askAnsweredAsyncTasks}
|
||||
* matches here — then fires the identical production cleanup {@code ask()}'s own timeout path
|
||||
* runs ({@code clearAsyncQuestion(turnId, true)}, via the {@code forgetTurnForTest} seam fleetd
|
||||
* #324 merged) at the point between the check and the removal use, via a dedicated test hook
|
||||
* mirroring {@code finishAsyncTaskRaceHook}.
|
||||
*/
|
||||
@Test
|
||||
void replyToAnOrphanedTaskSurvivesTurnIdGoingNullBetweenItsTwoReads() throws Exception {
|
||||
String ticket = messages.sendAsync(T, "task that asks");
|
||||
awaitWaiting();
|
||||
injectDelivery();
|
||||
|
||||
CompletableFuture<MessageService.AskResult> ask =
|
||||
CompletableFuture.supplyAsync(() -> messages.ask(T, "which config?", 5000));
|
||||
MessageService.TaskView asking = awaitTicketPhase(ticket, MessageService.Phase.ASKING);
|
||||
String turnId = asking.turnId();
|
||||
|
||||
MessageService.Reply answerReply = messages.answer(turnId, "config.yaml", 150);
|
||||
assertEquals("config.yaml", ask.get(5, TimeUnit.SECONDS).answer());
|
||||
assertEquals(MessageService.Outcome.TIMED_OUT_WORKING, answerReply.outcome(),
|
||||
"the primary's own bounded wait must give up first, leaving turnId stamped with no "
|
||||
+ "live waiter — the state reply()'s F3 code path matches");
|
||||
|
||||
messages.setReplyOrphanTurnIdRaceHookForTest(() -> messages.forgetTurnForTest(turnId));
|
||||
try {
|
||||
assertTrue(messages.reply(T, "PR opened: https://example/pulls/42"));
|
||||
|
||||
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
|
||||
assertEquals("PR opened: https://example/pulls/42", done.reply(),
|
||||
"the ticket must still resolve to the worker's real reply despite the forced race");
|
||||
} finally {
|
||||
messages.setReplyOrphanTurnIdRaceHookForTest(null);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* fleetd #307's ambiguity guard: an ask timeout frees its target ({@code hasAsyncQuestion}
|
||||
* becomes false the instant it lapses — proven above), so a second, independent delegation can
|
||||
@@ -1151,6 +1255,63 @@ class MessageServiceTest {
|
||||
assertEquals("done", done.reply());
|
||||
}
|
||||
|
||||
// --- fleetd #329 (F2): an exception after the ticket's future completes must reach a log -----
|
||||
//
|
||||
// sendAsync's own executor task ends with catch (Throwable t) { task.future.completeExceptionally(t); }
|
||||
// — but finishAsyncTask completes that same future on its first line, so anything that throws
|
||||
// afterward hits an already-completed future: completeExceptionally returns false and does
|
||||
// nothing, and (before this fix) nothing logged it either. Measured on the ticket: temporarily
|
||||
// reintroducing the fleetd #324 NPE reproduced 19 real exceptions on the ordinary path with
|
||||
// 74/74 tests staying green and zero log lines. There is no reachable production call site where
|
||||
// finishAsyncTask throws after completing the future (fleetd #324 already closed the one that
|
||||
// used to), so this test drives the shape directly via a dedicated test-only hook
|
||||
// (afterFinishAsyncTaskCompleteHookForTest) rather than trying to engineer a real exception into
|
||||
// that narrow window — the same technique fleetd #324's own race test and this ticket's F1/F3
|
||||
// tests use for their own races.
|
||||
|
||||
private static ListAppender<ILoggingEvent> attachMessageServiceLog() {
|
||||
Logger logger = (Logger) LoggerFactory.getLogger(MessageService.class);
|
||||
ListAppender<ILoggingEvent> appender = new ListAppender<>();
|
||||
appender.start();
|
||||
logger.addAppender(appender);
|
||||
return appender;
|
||||
}
|
||||
|
||||
private static void detachMessageServiceLog(ListAppender<ILoggingEvent> appender) {
|
||||
((Logger) LoggerFactory.getLogger(MessageService.class)).detachAppender(appender);
|
||||
}
|
||||
|
||||
@Test
|
||||
void anExceptionAfterTheTicketFutureCompletesStillReachesTheLog() throws Exception {
|
||||
ListAppender<ILoggingEvent> appender = attachMessageServiceLog();
|
||||
try {
|
||||
messages.setAfterFinishAsyncTaskCompleteHookForTest(() -> {
|
||||
throw new RuntimeException("PROBE-329-F2");
|
||||
});
|
||||
|
||||
String ticket = messages.sendAsync(T, "do the task");
|
||||
awaitWaiting();
|
||||
assertTrue(rendezvous.resolve(T, "done"));
|
||||
|
||||
// finishAsyncTask's own first line already completed the future with the real result
|
||||
// BEFORE the hook threw — F2 is a visibility gap, not a correctness gap for the ticket
|
||||
// itself, so the ticket's own outcome must be unaffected by the swallowed exception.
|
||||
MessageService.TaskView done = awaitTicketPhase(ticket, MessageService.Phase.DONE);
|
||||
assertEquals("done", done.reply());
|
||||
|
||||
assertTrue(appender.list.stream().anyMatch(e ->
|
||||
e.getLevel() == Level.ERROR
|
||||
&& e.getFormattedMessage().contains(ticket)
|
||||
&& e.getThrowableProxy() != null
|
||||
&& "PROBE-329-F2".equals(e.getThrowableProxy().getMessage())),
|
||||
"an exception thrown after the ticket's future already completed must still "
|
||||
+ "reach the log, not vanish silently");
|
||||
} finally {
|
||||
messages.setAfterFinishAsyncTaskCompleteHookForTest(null);
|
||||
detachMessageServiceLog(appender);
|
||||
}
|
||||
}
|
||||
|
||||
// --- CB-582: fleet_status pendingAsk() ------------------------------------------------------
|
||||
|
||||
@Test
|
||||
|
||||
Reference in New Issue
Block a user