diff --git a/apps/web/src/connection/storage.test.ts b/apps/web/src/connection/storage.test.ts index 206220768e9a75e68bb47f1daadd8f398d66a28b..dbb22baa0bd199557eeefbbb73275c0d4b776129 100644 --- a/apps/web/src/connection/storage.test.ts +++ b/apps/web/src/connection/storage.test.ts @@ -2,8 +2,8 @@ import { ConnectionTransientError, PrimaryConnectionTarget, } from "@t3tools/client-runtime/connection"; -import { EnvironmentId } from "@t3tools/contracts"; -import { ConnectionCatalogDocument } from "@t3tools/client-runtime/platform"; +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { ConnectionCatalogDocument, Persistence } from "@t3tools/client-runtime/platform"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; @@ -16,6 +16,7 @@ import { makeBrowserGitHubRoutingPermissions, makeCatalogBackend, makeCatalogStore, + layer as connectionStorageLayer, } from "./storage"; const emptyCatalog = { @@ -68,7 +69,31 @@ describe("makeCatalogStore", () => { ); }); +const fixedHandle = (database: IDBDatabase) => ({ + get: Effect.succeed(database), + invalidate: () => Effect.void, +}); + describe("makeCatalogBackend", () => { + it.effect("reports a closed IndexedDB connection as a typed read and write failure", () => + Effect.gen(function* () { + vi.stubGlobal("window", {}); + const database = { + transaction: () => { + throw new DOMException("The database connection is closing.", "InvalidStateError"); + }, + } as unknown as IDBDatabase; + const backend = makeCatalogBackend(fixedHandle(database)); + + const readError = yield* Effect.flip(backend.read); + const writeError = yield* Effect.flip(backend.write("{}")); + + expect(readError).toBeInstanceOf(ConnectionTransientError); + expect(readError.message).toContain("The database connection is closing."); + expect(writeError).toBeInstanceOf(ConnectionTransientError); + }), + ); + it.effect("fails writes when desktop secure storage declines the catalog", () => Effect.gen(function* () { const setConnectionCatalog = vi.fn().mockResolvedValue(false); @@ -78,7 +103,7 @@ describe("makeCatalogBackend", () => { setConnectionCatalog, }, }); - const backend = makeCatalogBackend({} as IDBDatabase); + const backend = makeCatalogBackend(fixedHandle({} as IDBDatabase)); const error = yield* backend.write("{}").pipe(Effect.flip); @@ -103,7 +128,9 @@ describe("makeCatalogBackend", () => { }, }), }); - const backend = makeCatalogBackend({ transaction: () => transaction } as never); + const backend = makeCatalogBackend( + fixedHandle({ transaction: () => transaction } as unknown as IDBDatabase), + ); const error = yield* backend.write("{}").pipe(Effect.flip); @@ -112,6 +139,174 @@ describe("makeCatalogBackend", () => { ); }); +describe("environment cache removal", () => { + it.effect("fails both removal operations when IndexedDB aborts their commits", () => + Effect.gen(function* () { + vi.stubGlobal("window", {}); + vi.stubGlobal("IDBKeyRange", { bound: () => ({}) }); + const database = Object.assign(new EventTarget(), { + transaction: () => { + const transaction = Object.assign(new EventTarget(), { + error: new DOMException("Commit aborted", "AbortError"), + objectStore: () => ({ + delete: () => queueMicrotask(() => transaction.dispatchEvent(new Event("abort"))), + openCursor: () => { + queueMicrotask(() => transaction.dispatchEvent(new Event("abort"))); + return new EventTarget(); + }, + }), + }); + return transaction; + }, + close: vi.fn(), + }) as unknown as IDBDatabase; + const openRequest = Object.assign(new EventTarget(), { result: database, error: null }); + vi.stubGlobal("indexedDB", { + open: () => { + queueMicrotask(() => openRequest.dispatchEvent(new Event("success"))); + return openRequest; + }, + }); + + const [threadError, refsError] = yield* Effect.gen(function* () { + const cache = yield* Persistence.EnvironmentCacheStore; + return [ + yield* Effect.flip( + cache.removeThread(EnvironmentId.make("env"), ThreadId.make("thread")), + ), + yield* Effect.flip(cache.clearVcsRefs(EnvironmentId.make("env"))), + ] as const; + }).pipe(Effect.provide(connectionStorageLayer)); + + expect(threadError.message).toContain("Commit aborted"); + expect(refsError.message).toContain("Commit aborted"); + expect(database.close).toHaveBeenCalledOnce(); + }), + ); +}); + +describe("IndexedDB connection recovery", () => { + it.effect("reports an initial open failure from the cache operation", () => + Effect.gen(function* () { + vi.stubGlobal("window", {}); + const open = vi.fn(() => { + throw new DOMException("Storage is unavailable", "InvalidStateError"); + }); + vi.stubGlobal("indexedDB", { open }); + + yield* Effect.gen(function* () { + const cache = yield* Persistence.EnvironmentCacheStore; + expect(open).not.toHaveBeenCalled(); + const error = yield* Effect.flip( + cache.loadThread(EnvironmentId.make("env"), ThreadId.make("thread")), + ); + expect(error.message).toContain("Storage is unavailable"); + }).pipe(Effect.provide(connectionStorageLayer)); + + expect(open).toHaveBeenCalledOnce(); + }), + ); + + it.effect("reopens after a forced close and finalizes the current connection", () => + Effect.gen(function* () { + vi.stubGlobal("window", {}); + const makeDatabase = () => + Object.assign(new EventTarget(), { + close: vi.fn(), + transaction: () => ({ + objectStore: () => ({ + get: () => { + const request = Object.assign(new EventTarget(), { + result: undefined, + error: null, + }); + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }, + }), + }), + }) as unknown as IDBDatabase; + const first = makeDatabase(); + const second = makeDatabase(); + const databases = [first, second]; + let openCount = 0; + const open = vi.fn(() => { + const request = Object.assign(new EventTarget(), { + result: databases[openCount++], + error: null, + }); + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }); + vi.stubGlobal("indexedDB", { open }); + + yield* Effect.gen(function* () { + const cache = yield* Persistence.EnvironmentCacheStore; + const environmentId = EnvironmentId.make("env"); + const threadId = ThreadId.make("thread"); + expect(Option.isNone(yield* cache.loadThread(environmentId, threadId))).toBe(true); + expect(open).toHaveBeenCalledTimes(1); + + first.dispatchEvent(new Event("close")); + const recovered = yield* Effect.all( + [cache.loadThread(environmentId, threadId), cache.loadThread(environmentId, threadId)], + { concurrency: 2 }, + ); + expect(recovered.every(Option.isNone)).toBe(true); + expect(open).toHaveBeenCalledTimes(2); + }).pipe(Effect.provide(connectionStorageLayer)); + + expect(first.close).not.toHaveBeenCalled(); + expect(second.close).toHaveBeenCalledOnce(); + }), + ); +}); + +describe("IndexedDB connection closed without a close event", () => { + it.effect("reopens and retries the failing operation once", () => + Effect.gen(function* () { + vi.stubGlobal("window", {}); + const closing = Object.assign(new EventTarget(), { + close: vi.fn(), + transaction: () => { + // Chromium force-closed this connection; this tab never saw "close". + throw new DOMException("The database connection is closing.", "InvalidStateError"); + }, + }) as unknown as IDBDatabase; + const fresh = Object.assign(new EventTarget(), { + close: vi.fn(), + transaction: () => ({ + objectStore: () => ({ + get: () => { + const request = Object.assign(new EventTarget(), { result: undefined, error: null }); + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }, + }), + }), + }) as unknown as IDBDatabase; + const databases = [closing, fresh]; + let openCount = 0; + const open = vi.fn(() => { + const request = Object.assign(new EventTarget(), { + result: databases[openCount++], + error: null, + }); + queueMicrotask(() => request.dispatchEvent(new Event("success"))); + return request; + }); + vi.stubGlobal("indexedDB", { open }); + + yield* Effect.gen(function* () { + const cache = yield* Persistence.EnvironmentCacheStore; + const loaded = yield* cache.loadThread(EnvironmentId.make("env"), ThreadId.make("thread")); + expect(Option.isNone(loaded)).toBe(true); + expect(open).toHaveBeenCalledTimes(2); + }).pipe(Effect.provide(connectionStorageLayer)); + }), + ); +}); + describe("browser GitHub routing permissions", () => { it.effect("revokes across runtimes before storage events and resists stale catalog writes", () => Effect.gen(function* () { diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 03ecbdb3f6a9e50ff64ea8f966ea642342cb349e..15fca39e174e7dcd55e61777e94874f6d8480ace 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -118,110 +118,253 @@ const openDatabase = Effect.fn("web.connectionStorage.openDatabase")(function* ( ); return; } - const request = indexedDB.open(DATABASE_NAME, DATABASE_VERSION); - request.addEventListener("upgradeneeded", () => { - if (!request.result.objectStoreNames.contains(CATALOG_STORE_NAME)) { - request.result.createObjectStore(CATALOG_STORE_NAME); - } - if (!request.result.objectStoreNames.contains(SHELL_STORE_NAME)) { - request.result.createObjectStore(SHELL_STORE_NAME); - } - if (!request.result.objectStoreNames.contains(THREAD_STORE_NAME)) { - request.result.createObjectStore(THREAD_STORE_NAME); - } - if (!request.result.objectStoreNames.contains(SERVER_CONFIG_STORE_NAME)) { - request.result.createObjectStore(SERVER_CONFIG_STORE_NAME); - } - if (!request.result.objectStoreNames.contains(VCS_REFS_STORE_NAME)) { - request.result.createObjectStore(VCS_REFS_STORE_NAME); - } - }); - request.addEventListener("error", () => { - resume(Effect.fail(catalogError("open", request.error ?? "Unknown IndexedDB error"))); - }); - request.addEventListener("success", () => { - resume(Effect.succeed(request.result)); - }); + try { + const request = indexedDB.open(DATABASE_NAME, DATABASE_VERSION); + request.addEventListener("upgradeneeded", () => { + if (!request.result.objectStoreNames.contains(CATALOG_STORE_NAME)) { + request.result.createObjectStore(CATALOG_STORE_NAME); + } + if (!request.result.objectStoreNames.contains(SHELL_STORE_NAME)) { + request.result.createObjectStore(SHELL_STORE_NAME); + } + if (!request.result.objectStoreNames.contains(THREAD_STORE_NAME)) { + request.result.createObjectStore(THREAD_STORE_NAME); + } + if (!request.result.objectStoreNames.contains(SERVER_CONFIG_STORE_NAME)) { + request.result.createObjectStore(SERVER_CONFIG_STORE_NAME); + } + if (!request.result.objectStoreNames.contains(VCS_REFS_STORE_NAME)) { + request.result.createObjectStore(VCS_REFS_STORE_NAME); + } + }); + request.addEventListener("error", () => { + resume(Effect.fail(catalogError("open", request.error ?? "Unknown IndexedDB error"))); + }); + request.addEventListener("success", () => { + resume(Effect.succeed(request.result)); + }); + } catch (cause) { + resume(Effect.fail(catalogError("open", cause))); + } }); }); -function readDatabaseValue(database: IDBDatabase, storeName: string, key: IDBValidKey) { +interface DatabaseHandle { + readonly get: Effect.Effect; + /** Forget `database` if it is still the shared connection, so the next access reopens. */ + readonly invalidate: (database: IDBDatabase) => Effect.Effect; +} + +/** Share a connection until the browser closes it; the next access reopens it. */ +const makeDatabaseHandle = Effect.fn("web.connectionStorage.makeDatabaseHandle")(function* () { + const lock = yield* Semaphore.make(1); + let current: IDBDatabase | null = null; + const forget = (database: IDBDatabase) => { + if (current === database) current = null; + }; + const get = Effect.suspend(() => + current !== null + ? Effect.succeed(current) + : lock.withPermits(1)( + Effect.gen(function* () { + if (current !== null) return current; + const opened = yield* openDatabase(); + current = opened; + opened.addEventListener("close", () => forget(opened)); + // Another tab upgrading the schema waits on this connection. + opened.addEventListener("versionchange", () => { + forget(opened); + opened.close(); + }); + return opened; + }), + ), + ); + const close = lock.withPermits(1)( + Effect.sync(() => { + current?.close(); + current = null; + }), + ); + const handle: DatabaseHandle = { + get, + invalidate: (database) => Effect.sync(() => forget(database)), + }; + return { handle, close }; +}); + +/** + * Runs `use` on the shared connection. A connection the browser already closed + * throws InvalidStateError even when no close event reached this tab, so drop + * it and retry once on a fresh one instead of failing every later operation. + */ +function withDatabase( + database: DatabaseHandle, + use: (opened: IDBDatabase) => Effect.Effect, +) { + // Only a connection that opened and then failed is stale; a failing open is + // storage being unavailable, which a retry would not fix. + const closedConnection = Symbol("closedConnection"); + const attempt = Effect.flatMap(database.get, (opened) => + use(opened).pipe( + Effect.catchIf( + (error) => error.detail.includes("InvalidStateError"), + (error) => + database + .invalidate(opened) + .pipe(Effect.andThen(Effect.fail({ [closedConnection]: error } as const))), + ), + ), + ); + return attempt.pipe( + Effect.catchIf( + (error): error is { readonly [closedConnection]: ConnectionTransientError } => + closedConnection in error, + () => Effect.flatMap(database.get, use), + ), + ); +} + +function readDatabaseValueOnConnection(database: IDBDatabase, storeName: string, key: IDBValidKey) { return Effect.callback((resume) => { - const request = database.transaction(storeName, "readonly").objectStore(storeName).get(key); - request.addEventListener("error", () => { - resume(Effect.fail(catalogError("read", request.error ?? "Unknown IndexedDB read error"))); - }); - request.addEventListener("success", () => { - resume(Effect.succeed(request.result)); - }); + try { + const request = database.transaction(storeName, "readonly").objectStore(storeName).get(key); + request.addEventListener("error", () => { + resume(Effect.fail(catalogError("read", request.error ?? "Unknown IndexedDB read error"))); + }); + request.addEventListener("success", () => { + resume(Effect.succeed(request.result)); + }); + } catch (cause) { + resume(Effect.fail(catalogError("read", cause))); + } }).pipe(Effect.withSpan("web.connectionStorage.readDatabaseValue")); } -function writeDatabaseValue( +function writeDatabaseValueOnConnection( database: IDBDatabase, storeName: string, key: IDBValidKey, value: unknown, ) { return Effect.callback((resume) => { - const transaction = database.transaction(storeName, "readwrite"); - // Every failed write fires "abort". A failed commit, such as - // QuotaExceededError, fires only "abort" and no "error". - transaction.addEventListener("abort", () => { - resume( - Effect.fail(catalogError("write", transaction.error ?? "Unknown IndexedDB write error")), - ); - }); - transaction.addEventListener("complete", () => { - resume(Effect.void); - }); - transaction.objectStore(storeName).put(value, key); + try { + const transaction = database.transaction(storeName, "readwrite"); + // Every failed write fires "abort". A failed commit, such as + // QuotaExceededError, fires only "abort" and no "error". + transaction.addEventListener("abort", () => { + resume( + Effect.fail(catalogError("write", transaction.error ?? "Unknown IndexedDB write error")), + ); + }); + transaction.addEventListener("complete", () => { + resume(Effect.void); + }); + transaction.objectStore(storeName).put(value, key); + } catch (cause) { + resume(Effect.fail(catalogError("write", cause))); + } }).pipe(Effect.withSpan("web.connectionStorage.writeDatabaseValue")); } -function removeDatabaseValue(database: IDBDatabase, storeName: string, key: IDBValidKey) { +function removeDatabaseValueOnConnection( + database: IDBDatabase, + storeName: string, + key: IDBValidKey, +) { return Effect.callback((resume) => { - const transaction = database.transaction(storeName, "readwrite"); - transaction.addEventListener("error", () => { - resume( - Effect.fail(catalogError("remove", transaction.error ?? "Unknown IndexedDB remove error")), - ); - }); - transaction.addEventListener("complete", () => { - resume(Effect.void); - }); - transaction.objectStore(storeName).delete(key); + try { + const transaction = database.transaction(storeName, "readwrite"); + transaction.addEventListener("abort", () => { + resume( + Effect.fail( + catalogError("remove", transaction.error ?? "Unknown IndexedDB remove error"), + ), + ); + }); + transaction.addEventListener("complete", () => { + resume(Effect.void); + }); + transaction.objectStore(storeName).delete(key); + } catch (cause) { + resume(Effect.fail(catalogError("remove", cause))); + } }).pipe(Effect.withSpan("web.connectionStorage.removeDatabaseValue")); } -function removeDatabaseValuesInRange(database: IDBDatabase, storeName: string, range: IDBKeyRange) { +function removeDatabaseValuesInRangeOnConnection( + database: IDBDatabase, + storeName: string, + range: IDBKeyRange, +) { return Effect.callback((resume) => { - const transaction = database.transaction(storeName, "readwrite"); - transaction.addEventListener("error", () => { - resume( - Effect.fail(catalogError("remove", transaction.error ?? "Unknown IndexedDB cursor error")), - ); - }); - transaction.addEventListener("complete", () => { - resume(Effect.void); - }); - const request = transaction.objectStore(storeName).openCursor(range); - request.addEventListener("error", () => { - resume( - Effect.fail(catalogError("remove", request.error ?? "Unknown IndexedDB cursor error")), - ); - }); - request.addEventListener("success", () => { - const cursor = request.result; - if (cursor === null) { - return; - } - cursor.delete(); - cursor.continue(); - }); + try { + const transaction = database.transaction(storeName, "readwrite"); + transaction.addEventListener("abort", () => { + resume( + Effect.fail( + catalogError("remove", transaction.error ?? "Unknown IndexedDB cursor error"), + ), + ); + }); + transaction.addEventListener("complete", () => { + resume(Effect.void); + }); + const request = transaction.objectStore(storeName).openCursor(range); + request.addEventListener("error", () => { + resume( + Effect.fail(catalogError("remove", request.error ?? "Unknown IndexedDB cursor error")), + ); + }); + request.addEventListener("success", () => { + const cursor = request.result; + if (cursor === null) { + return; + } + try { + cursor.delete(); + cursor.continue(); + } catch (cause) { + resume(Effect.fail(catalogError("remove", cause))); + } + }); + } catch (cause) { + resume(Effect.fail(catalogError("remove", cause))); + } }).pipe(Effect.withSpan("web.connectionStorage.removeDatabaseValuesInRange")); } +function readDatabaseValue(database: DatabaseHandle, storeName: string, key: IDBValidKey) { + return withDatabase(database, (opened) => readDatabaseValueOnConnection(opened, storeName, key)); +} + +function writeDatabaseValue( + database: DatabaseHandle, + storeName: string, + key: IDBValidKey, + value: unknown, +) { + return withDatabase(database, (opened) => + writeDatabaseValueOnConnection(opened, storeName, key, value), + ); +} + +function removeDatabaseValue(database: DatabaseHandle, storeName: string, key: IDBValidKey) { + return withDatabase(database, (opened) => + removeDatabaseValueOnConnection(opened, storeName, key), + ); +} + +function removeDatabaseValuesInRange( + database: DatabaseHandle, + storeName: string, + range: IDBKeyRange, +) { + return withDatabase(database, (opened) => + removeDatabaseValuesInRangeOnConnection(opened, storeName, range), + ); +} + function threadCacheKey(environmentId: EnvironmentId, threadId: ThreadId) { return `${environmentId}:${threadId}`; } @@ -250,7 +393,7 @@ export interface CatalogBackend { readonly quarantine?: (raw: string) => Effect.Effect; } -export function makeCatalogBackend(database: IDBDatabase): CatalogBackend { +export function makeCatalogBackend(database: DatabaseHandle): CatalogBackend { const bridge = window.desktopBridge; if (bridge?.getConnectionCatalog !== undefined && bridge.setConnectionCatalog !== undefined) { return { @@ -459,8 +602,9 @@ export function makeBrowserGitHubRoutingPermissions( export const layer = Layer.effectContext( Effect.gen(function* () { - const database = yield* Effect.acquireRelease(openDatabase(), (database) => - Effect.sync(() => database.close()), + const { handle: database } = yield* Effect.acquireRelease( + makeDatabaseHandle(), + (owned) => owned.close, ); const catalog = yield* makeCatalogStore(makeCatalogBackend(database)); const githubRoutingPermissions = makeBrowserGitHubRoutingPermissions(); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 7bc377c0b4e40216b550d0d47148bcf5c66ae06c..a832d46b673257b17cea14344408976b81520f3c 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -108,7 +108,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly httpSnapshot?: ThreadSnapshotLoadResult; readonly completionMarker?: boolean; readonly resumeCache?: NonNullable[1]>; - readonly loadCached?: Effect.Effect>; + readonly loadCached?: ReturnType; readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"]; readonly historyPaging?: "enabled" | "no-http" | "no-controller"; readonly historyHttpClient?: HttpClient.HttpClient; @@ -315,6 +315,28 @@ const deleted = (sequence = 3): OrchestrationV2ThreadStreamItem => { }; describe("EnvironmentThreads", () => { + it.effect("loads the server thread when its local cache read fails", () => + Effect.gen(function* () { + const h = yield* makeHarness({ + loadCached: Effect.fail( + new Persistence.ConnectionPersistenceError({ + operation: "load-thread", + message: "The database connection is closing.", + }), + ), + httpSnapshot: { + _tag: "present", + snapshot: { snapshotSequence: 7, projection: BASE_PROJECTION }, + }, + }); + + const state = yield* awaitThreadState(h.observed, (value) => value.status === "live"); + expect(Option.getOrThrow(state.data)).toEqual(BASE_PROJECTION); + expect(yield* Ref.get(h.loaderCalls)).toBe(1); + expect(yield* Ref.get(h.subscriptionCount)).toBe(1); + }), + ); + it.effect.each(["disk", "HTTP"] as const)( "does not rewrite an unchanged %s snapshot on navigation or warm return", (source) =>