authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-05 00:15:46-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-05 00:15:46-07:00
logf87ecf0cc307f974745da8eea856fab5711023f3
tree8922d65ce23c37f4cb6a63c7d84eff5ff72293ae
parentfc1f543d6c67da9cf1edb25b647646faa0105b41
signature Signed by PGP key B5690EEEBB952194

fix(client): keep warm thread resumes live instead of flashing sync (#9913)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

2 files changed, 110 insertions(+), 5 deletions(-)

packages/client-runtime/src/state/threads-atoms.test.ts+89-2
...@@ -27,6 +27,7 @@ import {...@@ -27,6 +27,7 @@ import {
27 PrimaryConnectionTarget,27 PrimaryConnectionTarget,
28 type NetworkStatus,28 type NetworkStatus,
29 type PreparedConnection,29 type PreparedConnection,
30 type SupervisorConnectionState,
30} from "../connection/model.ts";31} from "../connection/model.ts";
31import { EnvironmentSupervisor } from "../connection/supervisor.ts";32import { EnvironmentSupervisor } from "../connection/supervisor.ts";
32import { EnvironmentCacheStore } from "../platform/persistence.ts";33import { EnvironmentCacheStore } from "../platform/persistence.ts";
...@@ -73,8 +74,18 @@ const THREAD: OrchestrationThread = {...@@ -73,8 +74,18 @@ const THREAD: OrchestrationThread = {
73};74};
74const SNAPSHOT: OrchestrationThreadDetailSnapshot = { snapshotSequence: 7, thread: THREAD };75const SNAPSHOT: OrchestrationThreadDetailSnapshot = { snapshotSequence: 7, thread: THREAD };
7576
77const CONNECTED_STATE: SupervisorConnectionState = {
78 ...AVAILABLE_CONNECTION_STATE,
79 desired: true,
80 network: "online",
81 phase: "connected",
82 attempt: 1,
83 generation: 1,
84};
85
76const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {86const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
77 readonly snapshot?: OrchestrationThreadDetailSnapshot;87 readonly snapshot?: OrchestrationThreadDetailSnapshot;
88 readonly connected?: boolean;
78}) {89}) {
79 const subscriptions = yield* Queue.unbounded<{90 const subscriptions = yield* Queue.unbounded<{
80 readonly afterSequence: number | undefined;91 readonly afterSequence: number | undefined;
...@@ -123,10 +134,14 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?...@@ -123,10 +134,14 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
123 probe: Effect.void,134 probe: Effect.void,
124 closed: Effect.never,135 closed: Effect.never,
125 };136 };
137 const connectionState = yield* SubscriptionRef.make(
138 options?.connected ? CONNECTED_STATE : AVAILABLE_CONNECTION_STATE,
139 );
140 const sessionRef = yield* SubscriptionRef.make(Option.some(session));
126 const supervisor = EnvironmentSupervisor.of({141 const supervisor = EnvironmentSupervisor.of({
127 target: TARGET,142 target: TARGET,
128 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),143 state: connectionState,
129 session: yield* SubscriptionRef.make(Option.some(session)),144 session: sessionRef,
130 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(145 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
131 Option.some({146 Option.some({
132 environmentId: TARGET.environmentId,147 environmentId: TARGET.environmentId,
...@@ -228,6 +243,9 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?...@@ -228,6 +243,9 @@ const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?
228 ref,243 ref,
229 subscriptions,244 subscriptions,
230 olderLoads,245 olderLoads,
246 connectionState,
247 session,
248 sessionRef,
231 counts: () => ({ httpLoads, diskLoads, opened, active }),249 counts: () => ({ httpLoads, diskLoads, opened, active }),
232 };250 };
233});251});
...@@ -319,6 +337,75 @@ describe("createEnvironmentThreadStateAtoms", () => {...@@ -319,6 +337,75 @@ describe("createEnvironmentThreadStateAtoms", () => {
319 }),337 }),
320 );338 );
321339
340 it.effect.each([
341 { replayed: false, statuses: ["live"] },
342 { replayed: true, statuses: ["live", "synchronizing", "live"] },
343 ])(
344 "keeps a warm resume live until it replays events (replayed: $replayed)",
345 ({ replayed, statuses }) =>
346 Effect.gen(function* () {
347 const h = yield* makeHarness({ connected: true });
348 const unmount = h.registry.mount(h.stateAtom);
349 const first = yield* Queue.take(h.subscriptions);
350 yield* Queue.offer(first.events, { kind: "synchronized" });
351 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
352 unmount();
353 yield* Deferred.await(first.closed);
354
355 const observed: Array<EnvironmentThreadState["status"]> = [];
356 const stop = h.registry.subscribe(h.stateAtom, (state) => observed.push(state.status), {
357 immediate: true,
358 });
359 const remount = h.registry.mount(h.stateAtom);
360 const next = yield* Queue.take(h.subscriptions);
361 expect(next.afterSequence).toBe(7);
362 if (replayed) {
363 yield* Queue.offer(next.events, {
364 kind: "snapshot",
365 snapshot: { snapshotSequence: 9, thread: { ...THREAD, title: "Replayed" } },
366 });
367 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "synchronizing");
368 }
369 yield* Queue.offer(next.events, { kind: "synchronized" });
370 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
371 expect(observed.filter((status, index) => observed[index - 1] !== status)).toEqual(
372 statuses,
373 );
374 stop();
375 remount();
376 yield* Deferred.await(next.closed);
377 }),
378 );
379
380 it.effect("downgrades a warm resume when the connection dropped while away", () =>
381 Effect.gen(function* () {
382 const h = yield* makeHarness({ connected: true });
383 const unmount = h.registry.mount(h.stateAtom);
384 const first = yield* Queue.take(h.subscriptions);
385 yield* Queue.offer(first.events, { kind: "synchronized" });
386 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
387 unmount();
388 yield* Deferred.await(first.closed);
389
390 yield* SubscriptionRef.set(h.sessionRef, Option.none());
391 yield* SubscriptionRef.set(h.connectionState, AVAILABLE_CONNECTION_STATE);
392 const remount = h.registry.mount(h.stateAtom);
393 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "cached");
394 expect(currentThread(h.registry, h.stateAtom)).toBe(THREAD);
395 expect(h.counts().opened).toBe(1);
396
397 yield* SubscriptionRef.set(h.connectionState, CONNECTED_STATE);
398 yield* SubscriptionRef.set(h.sessionRef, Option.some(h.session));
399 const next = yield* Queue.take(h.subscriptions);
400 expect(next.afterSequence).toBe(7);
401 expect(h.registry.get(h.stateAtom).status).toBe("synchronizing");
402 yield* Queue.offer(next.events, { kind: "synchronized" });
403 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
404 remount();
405 yield* Deferred.await(next.closed);
406 }),
407 );
408
322 it.effect("keeps warm data when the raw atom family's weak entry is collected", () =>409 it.effect("keeps warm data when the raw atom family's weak entry is collected", () =>
323 Effect.gen(function* () {410 Effect.gen(function* () {
324 const h = yield* makeHarness();411 const h = yield* makeHarness();
packages/client-runtime/src/state/threads.ts+21-3
...@@ -158,10 +158,18 @@ function matchesThreadSnapshot(...@@ -158,10 +158,18 @@ function matchesThreadSnapshot(
158 currentPage.hasMore === page.hasMore;158 currentPage.hasMore === page.hasMore;
159}159}
160160
161// A retained "live" state stays live: the cursor resume that follows only
162// replays what the thread missed, and on servers that send the completion
163// marker the first replayed event moves the status to "synchronizing" on its
164// own. Downgrading here would flash a sync label on every return to a
165// recently viewed thread.
161function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState {166function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState {
162 return {167 return {
163 ...value,168 ...value,
164 status: value.status === "deleted" ? "deleted" : statusWithoutLiveData(value.data),169 status:
170 value.status === "deleted" || (value.status === "live" && Option.isSome(value.data))
171 ? value.status
172 : statusWithoutLiveData(value.data),
165 error: Option.none(),173 error: Option.none(),
166 page: Option.map(value.page, (page) => ({ ...page, loadingOlder: false })),174 page: Option.map(value.page, (page) => ({ ...page, loadingOlder: false })),
167 };175 };
...@@ -647,7 +655,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -647,7 +655,16 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
647 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),655 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
648 });656 });
649657
650 yield* setSynchronizing;658 // Only the first subscription after a warm live resume keeps the retained
659 // status. A replacement session or foreground resubscribe on the same scope
660 // may have missed events, so those show sync progress until confirmed.
661 const resumingLive = yield* Ref.make(initialState.status === "live");
662 const markSynchronizing = Effect.gen(function* () {
663 if (yield* Ref.get(resumingLive)) return;
664 yield* setSynchronizing;
665 });
666
667 yield* markSynchronizing;
651 yield* Effect.forkScoped(668 yield* Effect.forkScoped(
652 subscribeDynamic(669 subscribeDynamic(
653 ORCHESTRATION_WS_METHODS.subscribeThread,670 ORCHESTRATION_WS_METHODS.subscribeThread,
...@@ -668,7 +685,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -668,7 +685,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
668 const supportsPagination = config.threadSnapshotPagination === true;685 const supportsPagination = config.threadSnapshotPagination === true;
669 yield* Ref.set(paginationSupported, supportsPagination);686 yield* Ref.set(paginationSupported, supportsPagination);
670 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);687 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
671 yield* setSynchronizing;688 yield* markSynchronizing;
689 yield* Ref.set(resumingLive, false);
672690
673 let current = yield* SubscriptionRef.get(state);691 let current = yield* SubscriptionRef.get(state);
674 // A windowed cache resuming against a server without pagination is a692 // A windowed cache resuming against a server without pagination is a