From d7cf8aaa8d4fbcbdd523b4f4bc86fda5c47b4a70 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 4 Sep 2026 11:56:05 -0700 Subject: [PATCH] perf(client): stop thread streams when unused (#9740) Close a thread's live detail stream when its last consumer leaves. Retain a registry-local completed state and replay cursor for five idle minutes. Keep warm lookup independent of collectible raw atom definitions. Preserve paging and deletion state, reject stale owners, and cache only completed data/cursor updates. Skip repeated saves of unchanged data while retaining dirty-data flushes and failed-write retries. Verify real GC/remount behavior and focused lifecycle cases. Created with GPT-6 Astra (preview) in Codex. --- docs/internals/connection-runtime.md | 15 + .../client-runtime/src/state/threadDetail.ts | 45 +- .../src/state/threadRetention.ts | 6 +- .../src/state/threads-atoms.test.ts | 430 +++++++++++++++++- .../src/state/threads-sync.test.ts | 282 +++++++++++- packages/client-runtime/src/state/threads.ts | 262 ++++++++--- 6 files changed, 911 insertions(+), 129 deletions(-) diff --git a/docs/internals/connection-runtime.md b/docs/internals/connection-runtime.md index 1b686c365770a493ce7cde87fac45450273378b6..e84d9ed4d703cbc6da3b8c65f8045adb78f66f46 100644 --- a/docs/internals/connection-runtime.md +++ b/docs/internals/connection-runtime.md @@ -170,6 +170,20 @@ Application code must not construct RPC clients, retry loops, or raw orchestration commands. Persistence paths belong to the platform registration and cache stores, with explicit migration or invalidation policy. +### Thread detail ownership + +Mounted detail consumers share one live thread subscription. The subscription +ends when the last consumer leaves. Hidden routes that remain mounted still +count as consumers. + +A separate registry-local atom retains the last thread state and replay cursor +for five idle minutes. A warm mount renders that state and resumes the stream +without another snapshot download. The cache holds completed state and cursor +updates together. It retains pagination data but clears canceled loading state. +An ownership token prevents an old scope from replacing its successor's cache. +Closing a thread skips snapshots that were already saved. Changed snapshots and +failed background writes still get a final save attempt. + ## Verification Core state-machine tests use `@effect/vitest` and deterministic service layers. @@ -184,6 +198,7 @@ Required coverage includes: - relay token reuse and refresh; - progressive relay discovery; - shell and thread cache hydration; +- thread stream shutdown, warm resume, and snapshot expiry; - durable subscriptions switching sessions; - command metadata and idempotent queued-command metadata. diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index 5a2ffa442e04e986e354c6ba3a59adb8c6a02b36..0233cee0e22e10719fafe6be82c77f74c70380db 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -15,7 +15,6 @@ import type { EnvironmentThread, EnvironmentThreadShell } from "./models.ts"; import { scopeThread } from "./models.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; -import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; const EMPTY_MESSAGES: ReadonlyArray = Object.freeze([]); const EMPTY_ACTIVITIES: ReadonlyArray = Object.freeze([]); @@ -80,10 +79,7 @@ export function createEnvironmentThreadDetailAtoms( AsyncResult.value(get(threadStateAtom(ref.environmentId, ref.threadId))), () => EMPTY_ENVIRONMENT_THREAD_STATE, ), - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-state-value:${key}`), - ); + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state-value:${key}`)); }); const threadDetailAtomFamily = Atom.family((key: string) => { @@ -98,22 +94,19 @@ export function createEnvironmentThreadDetailAtoms( previousSource = source; previousValue = source === null ? null : scopeThread(ref.environmentId, source); return previousValue; - }).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-detail:${key}`), - ); + }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-detail:${key}`)); }); const threadStatusAtomFamily = Atom.family((key: string) => Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), + Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-status:${key}`), ), ); const threadErrorAtomFamily = Atom.family((key: string) => Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), + Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-error:${key}`), ), ); @@ -122,58 +115,40 @@ export function createEnvironmentThreadDetailAtoms( Atom.make( (get): ReadonlyArray => get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-messages:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-messages:${key}`)), ); const threadActivitiesAtomFamily = Atom.family((key: string) => Atom.make( (get): ReadonlyArray => get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-activities:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-activities:${key}`)), ); const threadProposedPlansAtomFamily = Atom.family((key: string) => Atom.make( (get): ReadonlyArray => get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-proposed-plans:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-proposed-plans:${key}`)), ); const threadCheckpointsAtomFamily = Atom.family((key: string) => Atom.make( (get): ReadonlyArray => get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-checkpoints:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-checkpoints:${key}`)), ); const threadSessionAtomFamily = Atom.family((key: string) => Atom.make( (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-session:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-session:${key}`)), ); const threadLatestTurnAtomFamily = Atom.family((key: string) => Atom.make( (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null, - ).pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-latest-turn:${key}`), - ), + ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-latest-turn:${key}`)), ); return { diff --git a/packages/client-runtime/src/state/threadRetention.ts b/packages/client-runtime/src/state/threadRetention.ts index 119b963167cd4ce3dd7b29082ca89bb3730874a4..e921a10d31ad424fac8733b59109938e294d888a 100644 --- a/packages/client-runtime/src/state/threadRetention.ts +++ b/packages/client-runtime/src/state/threadRetention.ts @@ -1,3 +1,3 @@ -// Mobile thread routes unmount during back navigation. Retain the stream-backed -// state across short subscriber gaps without keeping every opened thread alive. -export const THREAD_STATE_IDLE_TTL_MS = 5 * 60_000; +// Keep recent thread snapshots for back navigation. Live subscriptions end +// when the last detail consumer leaves. +export const THREAD_SNAPSHOT_IDLE_TTL_MS = 5 * 60_000; diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 0a9b2cd21b18e9ca41b39e30d5cc01c023c5a1c9..d7a415b2999ddcb604a2c089f215d9d0c23fdf23 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -1,26 +1,416 @@ -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; -import { describe, expect, it } from "@effect/vitest"; +import { + EnvironmentId, + EventId, + MessageId, + ORCHESTRATION_WS_METHODS, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationThread, + type OrchestrationThreadDetailSnapshot, + type OrchestrationThreadStreamItem, +} from "@t3tools/contracts"; +import { afterEach, describe, expect, it, vi } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; -import { Atom } from "effect/unstable/reactivity"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; +import { Atom, AtomRegistry } from "effect/unstable/reactivity"; -import type { EnvironmentRegistry } from "../connection/registry.ts"; -import type { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; -import { createEnvironmentThreadStateAtoms, type ThreadSnapshotLoader } from "./threads.ts"; +import type { ConnectionCatalogEntry } from "../connection/catalog.ts"; +import { EnvironmentRegistry } from "../connection/registry.ts"; +import { + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, + type NetworkStatus, + type PreparedConnection, +} from "../connection/model.ts"; +import { EnvironmentSupervisor } from "../connection/supervisor.ts"; +import { EnvironmentCacheStore } from "../platform/persistence.ts"; +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import type { RpcSession } from "../rpc/session.ts"; +import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts"; +import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; +import type { ThreadSnapshotWindow } from "./threadSnapshotHttp.ts"; +import { + createEnvironmentThreadStateAtoms, + requestOlderThreadTurns, + ThreadSnapshotLoader, + type EnvironmentThreadState, +} from "./threads.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); +const THREAD_ID = ThreadId.make("thread-1"); +const THREAD: OrchestrationThread = { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + title: "Cached thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + latestTurn: null, + createdAt: "2026-04-01T00:00:00.000Z", + updatedAt: "2026-04-01T00:00:00.000Z", + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, +}; +const SNAPSHOT: OrchestrationThreadDetailSnapshot = { snapshotSequence: 7, thread: THREAD }; + +const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: { + readonly snapshot?: OrchestrationThreadDetailSnapshot; +}) { + const subscriptions = yield* Queue.unbounded<{ + readonly afterSequence: number | undefined; + readonly events: Queue.Queue; + readonly closed: Deferred.Deferred; + }>(); + const olderLoads = yield* Queue.unbounded<{ + readonly window: ThreadSnapshotWindow; + readonly response: Deferred.Deferred>; + readonly closed: Deferred.Deferred; + }>(); + const snapshot = options?.snapshot ?? SNAPSHOT; + let httpLoads = 0; + let diskLoads = 0; + let opened = 0; + let active = 0; + const client = { + [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) => + Stream.unwrap( + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const closed = yield* Deferred.make(); + yield* Effect.acquireRelease( + Effect.sync(() => { + opened += 1; + active += 1; + }), + () => + Effect.sync(() => { + active -= 1; + }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))), + ); + yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed }); + return Stream.fromQueue(events); + }), + ), + } as unknown as WsRpcProtocolClient; + const session: RpcSession = { + client, + initialConfig: Effect.succeed({ + threadResumeCompletionMarker: true, + threadSnapshotPagination: true, + } as never), + subscribeServerConfig: (input) => client.subscribeServerConfig(input), + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + }; + const supervisor = EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: yield* SubscriptionRef.make(Option.some(session)), + prepared: yield* SubscriptionRef.make>( + Option.some({ + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, + }), + ), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const environmentRegistry = EnvironmentRegistry.of({ + entries: yield* SubscriptionRef.make>( + new Map(), + ), + networkStatus: yield* SubscriptionRef.make("online"), + start: Effect.void, + register: () => Effect.die("Unexpected environment registration"), + registerPlatform: () => Effect.die("Unexpected environment registration"), + reconcilePlatform: () => Effect.die("Unexpected environment reconciliation"), + remove: () => Effect.die("Unexpected environment removal"), + removeRelayEnvironments: () => Effect.die("Unexpected environment removal"), + retryNow: () => Effect.void, + state: () => SubscriptionRef.get(supervisor.state), + stateChanges: () => SubscriptionRef.changes(supervisor.state), + run: (_environmentId, effect) => + Effect.provideService(effect, EnvironmentSupervisor, supervisor), + runStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor, supervisor), + followStream: (_environmentId, stream) => + Stream.provideService(stream, EnvironmentSupervisor, supervisor), + }); + const runtime = Atom.runtime( + Layer.mergeAll( + Layer.succeed(EnvironmentRegistry, environmentRegistry), + Layer.succeed( + EnvironmentCacheStore, + EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => + Effect.sync(() => { + diskLoads += 1; + return Option.none(); + }), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }), + ), + Layer.succeed( + ThreadSnapshotLoader, + ThreadSnapshotLoader.of({ + load: (_prepared, _threadId, window) => { + if (window?.beforeCursor === undefined) { + return Effect.sync(() => { + httpLoads += 1; + return Option.some(snapshot); + }); + } + return Effect.gen(function* () { + const response = + yield* Deferred.make>(); + const closed = yield* Deferred.make(); + yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined)); + yield* Queue.offer(olderLoads, { window, response, closed }); + return yield* Deferred.await(response); + }).pipe(Effect.scoped); + }, + }), + ), + ), + ); + const raw = createEnvironmentThreadStateAtoms(runtime); + const details = createEnvironmentThreadDetailAtoms(raw.stateAtom); + const ref = { environmentId: TARGET.environmentId, threadId: THREAD_ID }; + const stateAtom = details.stateAtom(ref); + const makeRegistry = Effect.acquireRelease( + Effect.sync(() => AtomRegistry.make({ defaultIdleTTL: 60_000, timeoutResolution: 1 })), + (registry) => Effect.sync(() => registry.dispose()), + ); + const registry = yield* makeRegistry; + + return { + registry, + makeRegistry, + rawAtoms: raw, + stateAtom, + details, + ref, + subscriptions, + olderLoads, + counts: () => ({ httpLoads, diskLoads, opened, active }), + }; +}); + +function observeState( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom, + predicate: (state: EnvironmentThreadState) => boolean, +) { + return AtomRegistry.toStream(registry, atom).pipe( + Stream.filter(predicate), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); +} + +function currentThread( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom, +) { + return Option.getOrThrow(registry.get(atom).data); +} describe("createEnvironmentThreadStateAtoms", () => { - it("retains thread state across short subscriber gaps", () => { - const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime< - EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader, - never - >; - const threads = createEnvironmentThreadStateAtoms(runtime); - const environmentId = EnvironmentId.make("environment-1"); - const threadId = ThreadId.make("thread-1"); - const atom = threads.stateAtom(environmentId, threadId); - - expect(atom.idleTTL).toBe(THREAD_STATE_IDLE_TTL_MS); - expect(threads.stateAtom(environmentId, threadId)).toBe(atom); - expect(threads.stateAtom(environmentId, ThreadId.make("thread-2"))).not.toBe(atom); + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); }); + + it.effect("shares one live stream and closes it after the last detail consumer leaves", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const unmountMessages = h.registry.mount(h.details.messagesAtom(h.ref)); + const first = yield* Queue.take(h.subscriptions); + const unmountStatus = h.registry.mount(h.details.statusAtom(h.ref)); + expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 1, active: 1 }); + unmountMessages(); + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); + expect(h.counts().active).toBe(1); + unmountStatus(); + yield* Deferred.await(first.closed); + expect(h.counts().active).toBe(0); + }), + ); + + it.effect("keeps warm data and resumes a completed cursor without loading another snapshot", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const unmount = h.registry.mount(h.stateAtom); + const first = yield* Queue.take(h.subscriptions); + yield* Queue.offer(first.events, { + kind: "event", + event: { + type: "thread.message-sent", + sequence: 8, + eventId: EventId.make("message-1"), + aggregateKind: "thread", + aggregateId: THREAD_ID, + occurredAt: THREAD.createdAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { + threadId: THREAD_ID, + messageId: MessageId.make("message-1"), + role: "assistant", + text: "Retained text", + turnId: null, + streaming: true, + createdAt: THREAD.createdAt, + updatedAt: THREAD.createdAt, + }, + }, + }); + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live"); + const before = currentThread(h.registry, h.stateAtom); + unmount(); + yield* Deferred.await(first.closed); + const remount = h.registry.mount(h.stateAtom); + expect(currentThread(h.registry, h.stateAtom)).toBe(before); + const next = yield* Queue.take(h.subscriptions); + expect(next.afterSequence).toBe(8); + expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 }); + remount(); + yield* Deferred.await(next.closed); + }), + ); + + it.effect("keeps warm data when the raw atom family's weak entry is collected", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const oldRaw = h.rawAtoms.stateAtom(TARGET.environmentId, THREAD_ID); + const unmount = h.registry.mount(h.stateAtom); + const first = yield* Queue.take(h.subscriptions); + const latest = { ...THREAD, title: "Newer cached thread" }; + yield* Queue.offer(first.events, { + kind: "snapshot", + snapshot: { snapshotSequence: 8, thread: latest }, + }); + yield* Queue.offer(first.events, { kind: "synchronized" }); + yield* observeState(h.registry, h.stateAtom, (value) => value.status === "live"); + unmount(); + yield* Deferred.await(first.closed); + + // Force the weak-family miss without depending on host GC timing. + const deref = WeakRef.prototype.deref; + vi.spyOn(WeakRef.prototype, "deref").mockImplementation(function (this: WeakRef) { + const value = deref.call(this); + return value === oldRaw ? undefined : value; + }); + const remount = h.registry.mount(h.stateAtom); + expect(currentThread(h.registry, h.stateAtom)).toBe(latest); + const next = yield* Queue.take(h.subscriptions); + expect(next.afterSequence).toBe(8); + expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 }); + remount(); + yield* Deferred.await(next.closed); + }), + ); + + it.effect("keeps cached snapshots local to each registry", () => + Effect.gen(function* () { + const h = yield* makeHarness(); + const unmount = h.registry.mount(h.stateAtom); + const first = yield* Queue.take(h.subscriptions); + unmount(); + yield* Deferred.await(first.closed); + const otherRegistry = yield* h.makeRegistry; + const unmountOther = otherRegistry.mount(h.stateAtom); + const other = yield* Queue.take(h.subscriptions); + expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 }); + unmountOther(); + yield* Deferred.await(other.closed); + }), + ); + + it.effect("expires the plain snapshot after five idle minutes", () => + Effect.gen(function* () { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + const h = yield* makeHarness(); + const unmount = h.registry.mount(h.stateAtom); + const first = yield* Queue.take(h.subscriptions); + unmount(); + yield* Deferred.await(first.closed); + yield* Effect.yieldNow; + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(THREAD_SNAPSHOT_IDLE_TTL_MS + 1)); + const remount = h.registry.mount(h.stateAtom); + const next = yield* Queue.take(h.subscriptions); + expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 }); + remount(); + yield* Deferred.await(next.closed); + }), + ); + + it.effect("cancels older-page work on unmount and permits it again on a warm return", () => + Effect.gen(function* () { + const h = yield* makeHarness({ + snapshot: { + ...SNAPSHOT, + page: { beforeCursor: "older-1", hasMore: true, snapshotSequence: 7 }, + }, + }); + const unmount = h.registry.mount(h.stateAtom); + const first = yield* Queue.take(h.subscriptions); + expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true); + const older = yield* Queue.take(h.olderLoads); + expect(Option.getOrThrow(h.registry.get(h.stateAtom).page).loadingOlder).toBe(true); + unmount(); + yield* Deferred.await(first.closed); + yield* Deferred.await(older.closed); + const remount = h.registry.mount(h.stateAtom); + const next = yield* Queue.take(h.subscriptions); + expect(Option.getOrThrow(h.registry.get(h.stateAtom).page).loadingOlder).toBe(false); + expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true); + const retried = yield* Queue.take(h.olderLoads); + expect(retried.window.beforeCursor).toBe("older-1"); + expect(h.counts().httpLoads).toBe(1); + remount(); + yield* Deferred.await(next.closed); + yield* Deferred.await(retried.closed); + }), + ); }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index d94ed3a3fd7432cf381beb7acd252507d9021670..619b5f43425d7e408ba2f7ad0a75e2ce10fc887a 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -11,10 +11,14 @@ import { type OrchestrationThreadStreamItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; @@ -134,6 +138,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly cached?: OrchestrationThread; readonly httpSnapshot?: Option.Option; readonly completionMarker?: boolean; + readonly resumeCache?: NonNullable[1]>; + readonly loadCached?: Effect.Effect>; + readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -202,6 +209,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o loadShell: () => Effect.succeed(Option.none()), saveShell: () => Effect.void, loadThread: (_environmentId, threadId) => + options?.loadCached ?? Effect.succeed( threadId === THREAD_ID && options?.cached !== undefined ? Option.some({ @@ -210,8 +218,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o }) : Option.none(), ), - saveThread: (_environmentId, thread) => - Ref.update(savedThreads, (current) => [...current, thread]), + saveThread: (environmentId, thread) => + Ref.update(savedThreads, (current) => [...current, thread]).pipe( + Effect.andThen(options?.saveThread?.(environmentId, thread) ?? Effect.void), + ), removeThread: (_environmentId, threadId) => Ref.update(removedThreads, (current) => [...current, threadId]), loadServerConfig: () => Effect.succeed(Option.none()), @@ -222,7 +232,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o clearVcsRefs: () => Effect.void, clear: () => Effect.void, }); - const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( + const threadState = yield* makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ThreadSnapshotLoader, snapshotLoader), @@ -239,6 +249,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o ); return { + threadState, inputs, observed, latest, @@ -316,6 +327,271 @@ const deleted = (): OrchestrationThreadStreamItem => ({ }); describe("EnvironmentThreads", () => { + for (const source of ["disk", "HTTP"] as const) { + it.effect(`does not rewrite an unchanged ${source} snapshot on navigation or warm return`, () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + const firstSaved = yield* Effect.scoped( + Effect.gen(function* () { + const h = yield* makeHarness({ + resumeCache, + ...(source === "disk" + ? { cached: BASE_THREAD } + : { httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }) }), + }); + yield* awaitThreadState(h.observed, (value) => value.status === "live"); + if (source === "HTTP") yield* TestClock.adjust("500 millis"); + return h.savedThreads; + }), + ); + expect(yield* Ref.get(firstSaved)).toHaveLength(source === "disk" ? 0 : 1); + const nextSaved = yield* Effect.scoped( + Effect.gen(function* () { + const h = yield* makeHarness({ resumeCache }); + yield* awaitThreadState(h.observed, (value) => value.status === "live"); + return h.savedThreads; + }), + ); + expect(yield* Ref.get(nextSaved)).toEqual([]); + }), + ); + } + + it.effect("retries a failed background cache write when the scope closes", () => + Effect.gen(function* () { + let attempts = 0; + yield* Effect.scoped( + Effect.gen(function* () { + const h = yield* makeHarness({ + httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }), + saveThread: () => + Effect.suspend(() => { + attempts += 1; + return attempts === 1 + ? Effect.fail( + new Persistence.ConnectionPersistenceError({ + operation: "save-thread", + message: "Test storage failure", + }), + ) + : Effect.void; + }), + }); + yield* awaitThreadState(h.observed, (value) => value.status === "live"); + yield* TestClock.adjust("500 millis"); + expect(attempts).toBe(1); + }), + ); + expect(attempts).toBe(2); + }), + ); + + it.effect("flushes newer data after an older background write completes", () => + Effect.gen(function* () { + const writing = yield* Deferred.make(); + const written = yield* Deferred.make(); + const release = yield* Deferred.make(); + const saved = yield* Effect.scoped( + Effect.gen(function* () { + const h = yield* makeHarness({ + cached: BASE_THREAD, + saveThread: (_environmentId, snapshot) => + snapshot.snapshotSequence === 8 + ? Deferred.succeed(writing, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen(Deferred.succeed(written, undefined)), + ) + : Effect.void, + }); + yield* Queue.offer(h.inputs, titleUpdated("First update", 8)); + yield* awaitThreadState( + h.observed, + (value) => Option.getOrNull(value.data)?.title === "First update", + ); + yield* TestClock.adjust("500 millis"); + yield* Deferred.await(writing); + yield* Queue.offer(h.inputs, titleUpdated("Newer update", 9)); + yield* awaitThreadState( + h.observed, + (value) => Option.getOrNull(value.data)?.title === "Newer update", + ); + yield* Deferred.succeed(release, undefined); + yield* Deferred.await(written); + return h.savedThreads; + }), + ); + expect( + (yield* Ref.get(saved)).map((snapshot) => [ + snapshot.snapshotSequence, + snapshot.thread.title, + ]), + ).toEqual([ + [8, "First update"], + [9, "Newer update"], + ]); + }), + ); + + it.effect("does not resume past a canceled event whose data was not applied", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + const scope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + const first = yield* makeHarness({ + httpSnapshot: Option.some({ + thread: BASE_THREAD, + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + }), + resumeCache, + }).pipe(Effect.provideService(Scope.Scope, scope)); + yield* awaitThreadState(first.observed, (value) => value.status === "live"); + const applying = yield* Deferred.make(); + const update = titleUpdated("Not applied", 8); + if (update.kind !== "event") return yield* Effect.die("Expected an event"); + Object.defineProperty(update.event.payload, "title", { + get: () => { + Deferred.doneUnsafe(applying, Exit.void); + return "Not applied"; + }, + }); + // The reducer reads the title after it advances the cursor. Hold only + // the data write so closing the scope interrupts that exact interval. + yield* first.threadState.semaphore.take(1); + yield* Queue.offer(first.inputs, update); + yield* Deferred.await(applying); + yield* Scope.close(scope, Exit.void); + yield* first.threadState.semaphore.release(1); + + const resumed = yield* makeHarness({ resumeCache }); + const state = yield* awaitThreadState(resumed.observed, (value) => value.status === "live"); + expect(Option.getOrThrow(state.data).title).toBe(BASE_THREAD.title); + expect(yield* Ref.get(resumed.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE); + expect(yield* Ref.get(first.savedThreads)).toEqual([ + { + thread: BASE_THREAD, + snapshotSequence: CACHED_SNAPSHOT_SEQUENCE, + }, + ]); + }), + ); + + it.effect("prevents an old scope from replacing its successor's cached data", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + const oldScope = yield* Effect.acquireRelease(Scope.make(), (scope) => + Scope.close(scope, Exit.void), + ); + const old = yield* makeHarness({ cached: BASE_THREAD, resumeCache }).pipe( + Effect.provideService(Scope.Scope, oldScope), + ); + yield* awaitThreadState(old.observed, (value) => value.status === "live"); + const successor = yield* makeHarness({ resumeCache }); + yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9)); + yield* awaitThreadState( + successor.observed, + (value) => Option.getOrNull(value.data)?.title === "Successor title", + ); + yield* Queue.offer(old.inputs, titleUpdated("Old scope title", 8)); + yield* awaitThreadState( + old.observed, + (value) => Option.getOrNull(value.data)?.title === "Old scope title", + ); + yield* Scope.close(oldScope, Exit.void); + + expect(resumeCache.snapshot?.sequence).toBe(9); + expect(Option.getOrThrow(resumeCache.snapshot!.state.data).title).toBe("Successor title"); + expect(yield* Ref.get(old.savedThreads)).toEqual([]); + }), + ); + + it.effect("does not let a delayed cache read replace an initialized successor", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + const loading = yield* Deferred.make(); + const response = yield* Deferred.make>(); + const oldFiber = yield* makeHarness({ + resumeCache, + loadCached: Deferred.succeed(loading, undefined).pipe( + Effect.andThen(Deferred.await(response)), + ), + }).pipe(Effect.forkScoped); + yield* Deferred.await(loading); + const successor = yield* makeHarness({ cached: BASE_THREAD, resumeCache }); + yield* awaitThreadState(successor.observed, (value) => value.status === "live"); + yield* Deferred.succeed( + response, + Option.some({ snapshotSequence: 3, thread: { ...BASE_THREAD, title: "Old disk data" } }), + ); + const old = yield* Fiber.join(oldFiber); + yield* awaitThreadState(old.observed, (value) => value.status === "live"); + + expect(resumeCache.snapshot?.sequence).toBe(CACHED_SNAPSHOT_SEQUENCE); + expect(Option.getOrThrow(resumeCache.snapshot!.state.data).title).toBe(BASE_THREAD.title); + }), + ); + + it.effect("does not let an old deletion remove its successor's persisted cache", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + const old = yield* makeHarness({ cached: BASE_THREAD, resumeCache }); + yield* awaitThreadState(old.observed, (value) => value.status === "live"); + const successor = yield* makeHarness({ resumeCache }); + yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9)); + yield* awaitThreadState( + successor.observed, + (value) => Option.getOrNull(value.data)?.title === "Successor title", + ); + const update = deleted(); + if (update.kind !== "event") return yield* Effect.die("Expected an event"); + yield* Queue.offer(old.inputs, { ...update, event: { ...update.event, sequence: 8 } }); + yield* awaitThreadState(old.observed, (value) => value.status === "deleted"); + + expect(resumeCache.snapshot?.sequence).toBe(9); + expect(yield* Ref.get(old.removedThreads)).toEqual([]); + }), + ); + + it.effect("retains a deletion instead of restoring the old disk snapshot", () => + Effect.gen(function* () { + const resumeCache: NonNullable[1]> = { + snapshot: undefined, + owner: undefined, + }; + yield* Effect.scoped( + Effect.gen(function* () { + const first = yield* makeHarness({ cached: BASE_THREAD, resumeCache }); + const update = deleted(); + if (update.kind !== "event") return yield* Effect.die("Expected an event"); + yield* Queue.offer(first.inputs, { ...update, event: { ...update.event, sequence: 8 } }); + yield* awaitThreadState(first.observed, (value) => value.status === "deleted"); + }), + ); + const resumed = yield* makeHarness({ cached: BASE_THREAD, resumeCache }); + const state = yield* awaitThreadState( + resumed.observed, + (value) => value.status === "deleted", + ); + expect(Option.isNone(state.data)).toBe(true); + expect(yield* Ref.get(resumed.loaderCalls)).toBe(0); + }), + ); + it.effect("publishes cached data immediately from a warm cache", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 39561afdb416bde46475af085c127112862c7e61..a0055b6cab3cb3bc712689b251ba16d3898c127d 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -27,7 +27,7 @@ import { subscribeDynamic } from "../rpc/client.ts"; import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; -import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; +import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, @@ -131,40 +131,93 @@ function shouldPersistThread(thread: OrchestrationThread): boolean { return status !== "starting" && status !== "running"; } +interface ThreadResumeSnapshot { + readonly state: EnvironmentThreadState; + readonly sequence: number; + readonly persisted: boolean; +} + +interface ThreadResumeCache { + snapshot: ThreadResumeSnapshot | undefined; + owner: object | undefined; +} + +function matchesThreadSnapshot( + current: ThreadResumeSnapshot, + thread: OrchestrationThread | null, + sequence: number, + page: Pick | undefined, +): boolean { + if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== thread) + return false; + const currentPage = Option.getOrUndefined(current.state.page); + return currentPage === undefined + ? page === undefined + : page !== undefined && + currentPage.beforeCursor === page.beforeCursor && + currentPage.hasMore === page.hasMore; +} + +function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState { + return { + ...value, + status: value.status === "deleted" ? "deleted" : statusWithoutLiveData(value.data), + error: Option.none(), + page: Option.map(value.page, (page) => ({ ...page, loadingOlder: false })), + }; +} + export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* ( threadId: ThreadIdType, + resumeCache?: ThreadResumeCache, ) { const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; - const cached = yield* cache.loadThread(environmentId, threadId).pipe( - Effect.catch((error) => - Effect.logWarning("Could not load cached thread.").pipe( - Effect.annotateLogs({ - environmentId, - threadId, - error: error.message, - }), - Effect.as(Option.none()), - ), - ), - ); + const retained = resumeCache?.snapshot; + const owner = {}; + if (resumeCache) resumeCache.owner = owner; + const cached = + retained === undefined + ? yield* cache.loadThread(environmentId, threadId).pipe( + Effect.catch((error) => + Effect.logWarning("Could not load cached thread.").pipe( + Effect.annotateLogs({ + environmentId, + threadId, + error: error.message, + }), + Effect.as(Option.none()), + ), + ), + ) + : Option.none(); const cachedThread = Option.map(cached, (snapshot) => snapshot.thread); - const state = yield* SubscriptionRef.make({ - data: cachedThread, - status: statusWithoutLiveData(cachedThread), - error: Option.none(), - // A cached windowed snapshot restores its page cursor so "load earlier" - // works while rendering from cache; a cached full snapshot has no page. - page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)), - }); + const initialState: EnvironmentThreadState = retained + ? cachedThreadState(retained.state) + : { + data: cachedThread, + status: statusWithoutLiveData(cachedThread), + error: Option.none(), + // A cached windowed snapshot restores its page cursor so "load earlier" + // works while rendering from cache; a cached full snapshot has no page. + page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)), + }; + const state = yield* SubscriptionRef.make(initialState); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. - const lastSequence = yield* SubscriptionRef.make( - Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }), - ); + const initialSequence = + retained?.sequence ?? + Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }); + const lastSequence = yield* SubscriptionRef.make(initialSequence); + let committed: ThreadResumeSnapshot = { + state: initialState, + sequence: initialSequence, + persisted: retained?.persisted ?? Option.isSome(cached), + }; + if (resumeCache?.owner === owner) resumeCache.snapshot = committed; const awaitingCompletion = yield* Ref.make(false); // Bumped whenever loaded history may have been rewritten out from under an // in-flight older-page fetch (snapshot replacement, revert, deletion). A @@ -174,6 +227,25 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // merges. Without it, a revert or snapshot processed between loadOlderTurns' // epoch check and its merge could still slip resurrected history in. const applyLock = yield* Semaphore.make(1); + // Save only completed data/cursor updates. A canceled scope must not cache + // a cursor whose event has not reached the data yet. + const remember = Effect.gen(function* () { + const current = yield* SubscriptionRef.get(state); + const sequence = yield* SubscriptionRef.get(lastSequence); + committed = { + state: current, + sequence, + persisted: + committed.persisted && + matchesThreadSnapshot( + committed, + Option.getOrNull(current.data), + sequence, + Option.getOrUndefined(current.page), + ), + }; + if (resumeCache?.owner === owner) resumeCache.snapshot = committed; + }); // Whether the connected server accepts windowed reads; set per subscription // from the session config. Gates loadOlderTurns so a reconnect to a // pre-pagination server never sends unsupported window parameters. @@ -190,7 +262,28 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationThreadDetailSnapshot, ) { + if (resumeCache !== undefined && resumeCache.owner !== owner) return; + if ( + committed.persisted && + matchesThreadSnapshot(committed, snapshot.thread, snapshot.snapshotSequence, snapshot.page) + ) + return; yield* cache.saveThread(environmentId, snapshot).pipe( + Effect.tap(() => + Effect.sync(() => { + if ( + !matchesThreadSnapshot( + committed, + snapshot.thread, + snapshot.snapshotSequence, + snapshot.page, + ) + ) + return; + committed = { ...committed, persisted: true }; + if (resumeCache?.owner === owner) resumeCache.snapshot = committed; + }), + ), Effect.catch((error) => Effect.logWarning("Could not persist the thread cache.").pipe( Effect.annotateLogs({ @@ -300,6 +393,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make error: Option.none(), page: Option.none(), }); + yield* remember; + if (resumeCache !== undefined && resumeCache.owner !== owner) return; yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => Effect.logWarning("Could not remove the cached thread.").pipe( @@ -402,7 +497,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( item: OrchestrationThreadStreamItem, ) { - yield* applyLock.withPermits(1)(applyItemLocked(item)); + yield* applyLock.withPermits(1)(applyItemLocked(item).pipe(Effect.andThen(remember))); }); // Merges an older disjoint page below the currently loaded window. All four @@ -459,6 +554,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ...(snapshot.page === undefined ? {} : { page: { ...snapshot.page, snapshotSequence } }), }); } + yield* remember; }); const loadOlderTurns = Effect.fn("EnvironmentThreadState.loadOlderTurns")(function* () { @@ -579,15 +675,21 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // trap: afterSequence resume keeps only the window, and the missing // older turns can never be loaded (the server has no cursor reads). // Drop the window marker and treat the data as needing a full reload. - if (!supportsPagination && Option.isSome(current.page)) { - yield* Ref.update(historyEpoch, (epoch) => epoch + 1); - yield* SubscriptionRef.update(state, (value) => ({ - ...value, - data: Option.none(), - status: value.status === "deleted" ? value.status : ("empty" as const), - page: Option.none(), - })); - yield* SubscriptionRef.set(lastSequence, 0); + if (!supportsPagination) { + yield* applyLock.withPermits(1)( + Effect.gen(function* () { + if (Option.isNone((yield* SubscriptionRef.get(state)).page)) return; + yield* Ref.update(historyEpoch, (epoch) => epoch + 1); + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + data: Option.none(), + status: value.status === "deleted" ? value.status : ("empty" as const), + page: Option.none(), + })); + yield* SubscriptionRef.set(lastSequence, 0); + yield* remember; + }), + ); current = yield* SubscriptionRef.get(state); } if (Option.isNone(current.data) && current.status !== "deleted") { @@ -663,40 +765,45 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* Effect.addFinalizer(() => Effect.sync(deregister)); yield* Effect.addFinalizer(() => - Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe( - Effect.flatMap(([current, snapshotSequence]) => - Option.match(current.data, { - onNone: () => Effect.void, - onSome: (thread) => - shouldPersistThread(thread) - ? persist({ - snapshotSequence, - thread, - ...Option.match(current.page, { - onNone: () => ({}), - onSome: (page) => - ({ - page: { - beforeCursor: page.beforeCursor, - hasMore: page.hasMore, - snapshotSequence, - }, - }) as const, - }), - }) - : Effect.void, - }), - ), - ), + Effect.suspend(() => { + const { state: current, sequence: snapshotSequence } = committed; + return Option.match(current.data, { + onNone: () => Effect.void, + onSome: (thread) => + shouldPersistThread(thread) + ? persist({ + snapshotSequence, + thread, + ...Option.match(current.page, { + onNone: () => ({}), + onSome: (page) => + ({ + page: { + beforeCursor: page.beforeCursor, + hasMore: page.hasMore, + snapshotSequence, + }, + }) as const, + }), + }) + : Effect.void, + }); + }), ); return state; }); -export function threadStateChanges(environmentId: EnvironmentIdType, threadId: ThreadIdType) { +export function threadStateChanges( + environmentId: EnvironmentIdType, + threadId: ThreadIdType, + resumeCache?: ThreadResumeCache, +) { return followStreamInEnvironment( environmentId, - Stream.unwrap(makeEnvironmentThreadState(threadId).pipe(Effect.map(SubscriptionRef.changes))), + Stream.unwrap( + makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)), + ), ); } @@ -706,16 +813,35 @@ export function createEnvironmentThreadStateAtoms( E >, ) { + // Cache definitions must outlive collectible live-atom definitions. The + // registry retains these nodes without retaining environment or RPC scopes. + const resumeFamily = Atom.family((key: string) => + Atom.make((): ThreadResumeCache => ({ + snapshot: undefined, + owner: undefined, + })).pipe( + Atom.setIdleTTL(THREAD_SNAPSHOT_IDLE_TTL_MS), + Atom.withLabel(`environment-thread-resume:${key}`), + ), + ); const family = Atom.family((key: string) => { const { environmentId, threadId } = parseThreadKey(key); + const resumeAtom = resumeFamily(key); return runtime - .atom(threadStateChanges(environmentId, threadId), { - initialValue: EMPTY_ENVIRONMENT_THREAD_STATE, - }) - .pipe( - Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS), - Atom.withLabel(`environment-thread-state:${key}`), - ); + .atom( + (get) => { + get.mount(resumeAtom); + const resume = get.once(resumeAtom); + const live = threadStateChanges(environmentId, threadId, resume); + return resume.snapshot === undefined + ? live + : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live); + }, + { + initialValue: EMPTY_ENVIRONMENT_THREAD_STATE, + }, + ) + .pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state:${key}`)); }); return { -- 2.54.0