| author |  Theo Browne <me@t3.gg> 2026-09-25 20:24:35-07:00 |
| committer |  GitHub <noreply@github.com> 2026-09-25 20:24:35-07:00 |
| log | 574b18090281225de3449816c8366f0ee9ab886c |
| tree | ca1f1126b5722492f5337b0b328bc4c3086f1d4c |
| parent | 1dc8cbe6d14a7b4e131a2806f43bde8b60d6aaa7 |
| signature | Signed by PGP key B5690EEEBB952194 |
perf(server): shutdown no longer rewrites every stopped session row (#13688)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>2 files changed, 89 insertions(+), 3 deletions(-)
apps
server
src
provider
apps/server/src/provider/Layers/ProviderService.test.ts+72
| ... | ... | @@ -648,6 +648,78 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => |
| 648 | 648 | }), |
| 649 | 649 | ); |
| 650 | 650 | |
| 651 | it.effect("ProviderServiceLive shutdown leaves settled session rows untouched", () => | |
| 652 | Effect.gen(function* () { | |
| 653 | const recordedAnalytics = makeRecordingAnalytics(); | |
| 654 | const codex = makeFakeCodexAdapter(); | |
| 655 | const persistence = yield* Layer.build( | |
| 656 | ProviderSessionDirectoryLive.pipe( | |
| 657 | Layer.provide(ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory))), | |
| 658 | ), | |
| 659 | ); | |
| 660 | const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory.pipe( | |
| 661 | Effect.provide(persistence), | |
| 662 | ); | |
| 663 | const seed = (threadId: ThreadId, status: "running" | "stopped", activeTurnId: TurnId | null) => | |
| 664 | directory.upsert({ | |
| 665 | threadId, | |
| 666 | provider: CODEX_DRIVER, | |
| 667 | providerInstanceId: codexInstanceId, | |
| 668 | status, | |
| 669 | runtimePayload: { cwd: "/repo", activeTurnId }, | |
| 670 | }); | |
| 671 | const readBindings = directory | |
| 672 | .listBindings() | |
| 673 | .pipe( | |
| 674 | Effect.map((bindings) => new Map(bindings.map((binding) => [binding.threadId, binding]))), | |
| 675 | ); | |
| 676 | const settledId = asThreadId("shutdown-settled"); | |
| 677 | const runningId = asThreadId("shutdown-running"); | |
| 678 | const stoppedWithTurnId = asThreadId("shutdown-stopped-with-turn"); | |
| 679 | yield* seed(settledId, "stopped", null); | |
| 680 | yield* seed(runningId, "running", asTurnId("running-turn")); | |
| 681 | yield* seed(stoppedWithTurnId, "stopped", asTurnId("stale-turn")); | |
| 682 | const settledBefore = (yield* readBindings).get(settledId); | |
| 683 | assert(settledBefore !== undefined); | |
| 684 | ||
| 685 | const scope = yield* Scope.make(); | |
| 686 | yield* Layer.build( | |
| 687 | makeProviderServiceLive().pipe( | |
| 688 | Layer.provide(NodeServices.layer), | |
| 689 | Layer.provide(Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, directory)), | |
| 690 | Layer.provide( | |
| 691 | Layer.succeed( | |
| 692 | ProviderAdapterRegistry.ProviderAdapterRegistry, | |
| 693 | makeStaticInstanceRegistry([[codexInstanceId, codex.adapter]]), | |
| 694 | ), | |
| 695 | ), | |
| 696 | Layer.provide(defaultServerSettingsLayer), | |
| 697 | Layer.provide(serverConfigTestLayer), | |
| 698 | Layer.provide(recordedAnalytics.layer), | |
| 699 | Layer.provide( | |
| 700 | Layer.succeed( | |
| 701 | ProviderEventLoggers.ProviderEventLoggers, | |
| 702 | ProviderEventLoggers.NoOpProviderEventLoggers, | |
| 703 | ), | |
| 704 | ), | |
| 705 | ), | |
| 706 | ).pipe(Scope.provide(scope)); | |
| 707 | yield* TestClock.adjust("1 minute"); | |
| 708 | yield* Scope.close(scope, Exit.void); | |
| 709 | ||
| 710 | const byThread = yield* readBindings; | |
| 711 | assert.deepStrictEqual(byThread.get(settledId), settledBefore); | |
| 712 | for (const threadId of [runningId, stoppedWithTurnId]) { | |
| 713 | const binding = byThread.get(threadId); | |
| 714 | assert.equal(binding?.status, "stopped"); | |
| 715 | assert.propertyVal(binding?.runtimePayload, "activeTurnId", null); | |
| 716 | assert.propertyVal(binding?.runtimePayload, "lastRuntimeEvent", "provider.stopAll"); | |
| 717 | } | |
| 718 | const [stoppedAll] = recordedAnalytics.eventsByName("provider.sessions.stopped_all"); | |
| 719 | assert.equal(stoppedAll?.properties?.stoppedSessionCount, 2); | |
| 720 | }).pipe(Effect.provide(NodeServices.layer)), | |
| 721 | ); | |
| 722 | ||
| 651 | 723 | it.effect("ProviderServiceLive flushes deferred completions during shutdown", () => |
| 652 | 724 | Effect.gen(function* () { |
| 653 | 725 | const recordedAnalytics = makeRecordingAnalytics(); |
apps/server/src/provider/Layers/ProviderService.ts+17-3
| ... | ... | @@ -427,6 +427,14 @@ function readPersistedCwd( |
| 427 | 427 | return trimmed.length > 0 ? trimmed : undefined; |
| 428 | 428 | } |
| 429 | 429 | |
| 430 | /** Stopped rows with no active turn are settled; shutdown leaves them untouched. */ | |
| 431 | function isSettledBinding(binding: ProviderSessionDirectory.ProviderRuntimeBinding): boolean { | |
| 432 | if (binding.status !== "stopped") return false; | |
| 433 | const payload = binding.runtimePayload; | |
| 434 | if (!payload || typeof payload !== "object" || Array.isArray(payload)) return true; | |
| 435 | return !("activeTurnId" in payload) || payload.activeTurnId == null; | |
| 436 | } | |
| 437 | ||
| 430 | 438 | const dieOnMissingBindingInstanceId = ( |
| 431 | 439 | operation: string, |
| 432 | 440 | payload: { |
| ... | ... | @@ -2331,7 +2339,6 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( |
| 2331 | 2339 | return [completed, state] as const; |
| 2332 | 2340 | }); |
| 2333 | 2341 | yield* recordCompletedTurnProperties(properties); |
| 2334 | const threadIds = yield* directory.listThreadIds(); | |
| 2335 | 2342 | const currentAdapters = yield* getAdapterEntries; |
| 2336 | 2343 | const activeSessions = yield* Effect.forEach(currentAdapters, ([instanceId, adapter]) => |
| 2337 | 2344 | adapter.listSessions().pipe( |
| ... | ... | @@ -2362,7 +2369,12 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( |
| 2362 | 2369 | yield* Effect.forEach(currentAdapters, ([, adapter]) => adapter.stopAll()).pipe(Effect.asVoid); |
| 2363 | 2370 | yield* McpSessionRegistry.revokeAllActiveMcpCredentials(); |
| 2364 | 2371 | McpProviderSession.clearAllMcpProviderSessions(); |
| 2365 | const bindings = yield* directory.listBindings().pipe(Effect.orElseSucceed(() => [])); | |
| 2372 | // Stopped rows stay for their resume cursors, so long-lived installs hold | |
| 2373 | // thousands. Only rewrite the ones this shutdown actually stops. | |
| 2374 | const bindings = yield* directory.listBindings().pipe( | |
| 2375 | Effect.map((all) => all.filter((binding) => !isSettledBinding(binding))), | |
| 2376 | Effect.orElseSucceed(() => []), | |
| 2377 | ); | |
| 2366 | 2378 | yield* Effect.forEach(bindings, (binding) => |
| 2367 | 2379 | Effect.gen(function* () { |
| 2368 | 2380 | const providerInstanceId = dieOnMissingBindingInstanceId( |
| ... | ... | @@ -2382,8 +2394,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( |
| 2382 | 2394 | }); |
| 2383 | 2395 | }), |
| 2384 | 2396 | ).pipe(Effect.asVoid); |
| 2397 | // Not `sessionCount`: that older property counted every row, so a new name | |
| 2398 | // keeps the two meanings in separate series. | |
| 2385 | 2399 | yield* analytics.record("provider.sessions.stopped_all", { |
| 2386 | sessionCount: threadIds.length, | |
| 2400 | stoppedSessionCount: bindings.length, | |
| 2387 | 2401 | }); |
| 2388 | 2402 | yield* analytics.flush; |
| 2389 | 2403 | }); |