authorgravatar for maxwellbrown1990@gmail.comMaxwell Brown <maxwellbrown1990@gmail.com> 2026-03-27 21:05:14-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-27 18:05:14-07:00
loge08cea3a233263660af3f5f83ab1e58d8b13ec5b
tree67a93303916159b9878ed6f5f69ffe4bb0c8f61b
parent83eb396c407f167e70e29b29826d94f9ba1eb3aa
signature Signed by PGP key B5690EEEBB952194

Improve drain semantics via STM in `DrainableWorker` (#1474)

Co-authored-by: Julius Marminge <julius0216@outlook.com>

9 files changed, 93 insertions(+), 100 deletions(-)

apps/server/src/git/Layers/GitCore.test.ts+4-1
...@@ -415,6 +415,9 @@ it.layer(TestLayer)("git integration", (it) => {...@@ -415,6 +415,9 @@ it.layer(TestLayer)("git integration", (it) => {
415415
416 it.effect("refreshes upstream behind count after checkout when remote branch advanced", () =>416 it.effect("refreshes upstream behind count after checkout when remote branch advanced", () =>
417 Effect.gen(function* () {417 Effect.gen(function* () {
418 const services = yield* Effect.services();
419 const runPromise = Effect.runPromiseWith(services);
420
418 const remote = yield* makeTmpDir();421 const remote = yield* makeTmpDir();
419 const source = yield* makeTmpDir();422 const source = yield* makeTmpDir();
420 const clone = yield* makeTmpDir();423 const clone = yield* makeTmpDir();
...@@ -449,7 +452,7 @@ it.layer(TestLayer)("git integration", (it) => {...@@ -449,7 +452,7 @@ it.layer(TestLayer)("git integration", (it) => {
449 const core = yield* GitCore;452 const core = yield* GitCore;
450 yield* Effect.promise(() =>453 yield* Effect.promise(() =>
451 vi.waitFor(async () => {454 vi.waitFor(async () => {
452 const details = await Effect.runPromise(core.statusDetails(source));455 const details = await runPromise(core.statusDetails(source));
453 expect(details.branch).toBe(featureBranch);456 expect(details.branch).toBe(featureBranch);
454 expect(details.aheadCount).toBe(0);457 expect(details.aheadCount).toBe(0);
455 expect(details.behindCount).toBe(1);458 expect(details.behindCount).toBe(1);
apps/server/src/main.test.ts+2-2
...@@ -368,7 +368,7 @@ it.layer(testLayer)("server CLI command", (it) => {...@@ -368,7 +368,7 @@ it.layer(testLayer)("server CLI command", (it) => {
368368
369 it.effect("does not start server for invalid --mode values", () =>369 it.effect("does not start server for invalid --mode values", () =>
370 Effect.gen(function* () {370 Effect.gen(function* () {
371 yield* runCli(["--mode", "invalid"]);371 yield* runCli(["--mode", "invalid"]).pipe(Effect.catch(() => Effect.void));
372372
373 assert.equal(start.mock.calls.length, 0);373 assert.equal(start.mock.calls.length, 0);
374 assert.equal(stop.mock.calls.length, 0);374 assert.equal(stop.mock.calls.length, 0);
...@@ -386,7 +386,7 @@ it.layer(testLayer)("server CLI command", (it) => {...@@ -386,7 +386,7 @@ it.layer(testLayer)("server CLI command", (it) => {
386386
387 it.effect("does not start server for out-of-range --port values", () =>387 it.effect("does not start server for out-of-range --port values", () =>
388 Effect.gen(function* () {388 Effect.gen(function* () {
389 yield* runCli(["--port", "70000"]);389 yield* runCli(["--port", "70000"]).pipe(Effect.catch(() => Effect.void));
390390
391 // effect/unstable/cli renders help/errors for parse failures and returns success.391 // effect/unstable/cli renders help/errors for parse failures and returns success.
392 assert.equal(start.mock.calls.length, 0);392 assert.equal(start.mock.calls.length, 0);
apps/server/src/persistence/NodeSqliteClient.ts+17-4
...@@ -20,7 +20,7 @@ import * as Stream from "effect/Stream";...@@ -20,7 +20,7 @@ import * as Stream from "effect/Stream";
20import * as Reactivity from "effect/unstable/reactivity/Reactivity";20import * as Reactivity from "effect/unstable/reactivity/Reactivity";
21import * as Client from "effect/unstable/sql/SqlClient";21import * as Client from "effect/unstable/sql/SqlClient";
22import type { Connection } from "effect/unstable/sql/SqlConnection";22import type { Connection } from "effect/unstable/sql/SqlConnection";
23import { SqlError } from "effect/unstable/sql/SqlError";23import { SqlError, classifySqliteError } from "effect/unstable/sql/SqlError";
24import * as Statement from "effect/unstable/sql/Statement";24import * as Statement from "effect/unstable/sql/Statement";
2525
26const ATTR_DB_SYSTEM_NAME = "db.system.name";26const ATTR_DB_SYSTEM_NAME = "db.system.name";
...@@ -29,6 +29,9 @@ export const TypeId: TypeId = "~local/sqlite-node/SqliteClient";...@@ -29,6 +29,9 @@ export const TypeId: TypeId = "~local/sqlite-node/SqliteClient";
2929
30export type TypeId = "~local/sqlite-node/SqliteClient";30export type TypeId = "~local/sqlite-node/SqliteClient";
3131
32const classifyError = (cause: unknown, message: string, operation: string) =>
33 classifySqliteError(cause, { message, operation });
34
32/**35/**
33 * SqliteClient - Effect service tag for the sqlite SQL client.36 * SqliteClient - Effect service tag for the sqlite SQL client.
34 */37 */
...@@ -109,7 +112,10 @@ const makeWithDatabase = (...@@ -109,7 +112,10 @@ const makeWithDatabase = (
109 lookup: (sql: string) =>112 lookup: (sql: string) =>
110 Effect.try({113 Effect.try({
111 try: () => db.prepare(sql),114 try: () => db.prepare(sql),
112 catch: (cause) => new SqlError({ cause, message: "Failed to prepare statement" }),115 catch: (cause) =>
116 new SqlError({
117 reason: classifyError(cause, "Failed to prepare statement", "prepare"),
118 }),
113 }),119 }),
114 });120 });
115121
...@@ -127,7 +133,11 @@ const makeWithDatabase = (...@@ -127,7 +133,11 @@ const makeWithDatabase = (
127 const result = statement.run(...(params as any));133 const result = statement.run(...(params as any));
128 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);134 return Effect.succeed(raw ? (result as unknown as ReadonlyArray<any>) : []);
129 } catch (cause) {135 } catch (cause) {
130 return Effect.fail(new SqlError({ cause, message: "Failed to execute statement" }));136 return Effect.fail(
137 new SqlError({
138 reason: classifyError(cause, "Failed to execute statement", "execute"),
139 }),
140 );
131 }141 }
132 });142 });
133143
...@@ -150,7 +160,10 @@ const makeWithDatabase = (...@@ -150,7 +160,10 @@ const makeWithDatabase = (
150 statement.run(...(params as any));160 statement.run(...(params as any));
151 return [];161 return [];
152 },162 },
153 catch: (cause) => new SqlError({ cause, message: "Failed to execute statement" }),163 catch: (cause) =>
164 new SqlError({
165 reason: classifyError(cause, "Failed to execute statement", "execute"),
166 }),
154 }),167 }),
155 (statement) =>168 (statement) =>
156 Effect.sync(() => {169 Effect.sync(() => {
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+8-2
...@@ -1100,10 +1100,13 @@ describe("ClaudeAdapterLive", () => {...@@ -1100,10 +1100,13 @@ describe("ClaudeAdapterLive", () => {
1100 it.effect("closes the session when the Claude stream aborts after a turn starts", () => {1100 it.effect("closes the session when the Claude stream aborts after a turn starts", () => {
1101 const harness = makeHarness();1101 const harness = makeHarness();
1102 return Effect.gen(function* () {1102 return Effect.gen(function* () {
1103 const services = yield* Effect.services();
1104 const runFork = Effect.runForkWith(services);
1105
1103 const adapter = yield* ClaudeAdapter;1106 const adapter = yield* ClaudeAdapter;
1104 const runtimeEvents: Array<ProviderRuntimeEvent> = [];1107 const runtimeEvents: Array<ProviderRuntimeEvent> = [];
11051108
1106 const runtimeEventsFiber = Effect.runFork(1109 const runtimeEventsFiber = runFork(
1107 Stream.runForEach(adapter.streamEvents, (event) =>1110 Stream.runForEach(adapter.streamEvents, (event) =>
1108 Effect.sync(() => {1111 Effect.sync(() => {
1109 runtimeEvents.push(event);1112 runtimeEvents.push(event);
...@@ -1197,9 +1200,12 @@ describe("ClaudeAdapterLive", () => {...@@ -1197,9 +1200,12 @@ describe("ClaudeAdapterLive", () => {
1197 );1200 );
11981201
1199 return Effect.gen(function* () {1202 return Effect.gen(function* () {
1203 const services = yield* Effect.services();
1204 const runFork = Effect.runForkWith(services);
1205
1200 const adapter = yield* ClaudeAdapter;1206 const adapter = yield* ClaudeAdapter;
12011207
1202 const runtimeEventsFiber = Effect.runFork(1208 const runtimeEventsFiber = runFork(
1203 Stream.runForEach(adapter.streamEvents, () => Effect.void),1209 Stream.runForEach(adapter.streamEvents, () => Effect.void),
1204 );1210 );
12051211
apps/server/src/provider/Layers/ClaudeAdapter.ts+9-5
...@@ -2380,6 +2380,10 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2380,6 +2380,10 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2380 existingResumeSessionId === undefined ? yield* Random.nextUUIDv4 : undefined;2380 existingResumeSessionId === undefined ? yield* Random.nextUUIDv4 : undefined;
2381 const sessionId = existingResumeSessionId ?? newSessionId;2381 const sessionId = existingResumeSessionId ?? newSessionId;
23822382
2383 const services = yield* Effect.services();
2384 const runFork = Effect.runForkWith(services);
2385 const runPromise = Effect.runPromiseWith(services);
2386
2383 const promptQueue = yield* Queue.unbounded<PromptQueueItem>();2387 const promptQueue = yield* Queue.unbounded<PromptQueueItem>();
2384 const prompt = Stream.fromQueue(promptQueue).pipe(2388 const prompt = Stream.fromQueue(promptQueue).pipe(
2385 Stream.filter((item) => item.type === "message"),2389 Stream.filter((item) => item.type === "message"),
...@@ -2461,7 +2465,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2461,7 +2465,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2461 }2465 }
2462 aborted = true;2466 aborted = true;
2463 pendingUserInputs.delete(requestId);2467 pendingUserInputs.delete(requestId);
2464 Effect.runFork(Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers));2468 runFork(Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers));
2465 };2469 };
2466 callbackOptions.signal.addEventListener("abort", onAbort, { once: true });2470 callbackOptions.signal.addEventListener("abort", onAbort, { once: true });
24672471
...@@ -2607,7 +2611,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2607,7 +2611,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2607 return;2611 return;
2608 }2612 }
2609 pendingApprovals.delete(requestId);2613 pendingApprovals.delete(requestId);
2610 Effect.runFork(Deferred.succeed(decisionDeferred, "cancel"));2614 runFork(Deferred.succeed(decisionDeferred, "cancel"));
2611 };2615 };
26122616
2613 callbackOptions.signal.addEventListener("abort", onAbort, {2617 callbackOptions.signal.addEventListener("abort", onAbort, {
...@@ -2662,7 +2666,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2662,7 +2666,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2662 });2666 });
26632667
2664 const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) =>2668 const canUseTool: CanUseTool = (toolName, toolInput, callbackOptions) =>
2665 Effect.runPromise(canUseToolEffect(toolName, toolInput, callbackOptions));2669 runPromise(canUseToolEffect(toolName, toolInput, callbackOptions));
26662670
2667 const claudeSettings = yield* serverSettingsService.getSettings.pipe(2671 const claudeSettings = yield* serverSettingsService.getSettings.pipe(
2668 Effect.map((settings) => settings.providers.claudeAgent),2672 Effect.map((settings) => settings.providers.claudeAgent),
...@@ -2813,7 +2817,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2813,7 +2817,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2813 providerRefs: {},2817 providerRefs: {},
2814 });2818 });
28152819
2816 const streamFiber = Effect.runFork(runSdkStream(context));2820 const streamFiber = runFork(runSdkStream(context));
2817 context.streamFiber = streamFiber;2821 context.streamFiber = streamFiber;
2818 streamFiber.addObserver((exit) => {2822 streamFiber.addObserver((exit) => {
2819 if (context.stopped) {2823 if (context.stopped) {
...@@ -2822,7 +2826,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2822,7 +2826,7 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2822 if (context.streamFiber === streamFiber) {2826 if (context.streamFiber === streamFiber) {
2823 context.streamFiber = undefined;2827 context.streamFiber = undefined;
2824 }2828 }
2825 Effect.runFork(handleStreamExit(context, exit));2829 runFork(handleStreamExit(context, exit));
2826 });2830 });
28272831
2828 return {2832 return {
apps/server/src/wsServer.ts+8-8
...@@ -252,6 +252,11 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -252,6 +252,11 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
252 } = serverConfig;252 } = serverConfig;
253 const availableEditors = resolveAvailableEditors();253 const availableEditors = resolveAvailableEditors();
254254
255 const runtimeServices = yield* Effect.services<
256 ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path
257 >();
258 const runPromise = Effect.runPromiseWith(runtimeServices);
259
255 const gitManager = yield* GitManager;260 const gitManager = yield* GitManager;
256 const terminalManager = yield* TerminalManager;261 const terminalManager = yield* TerminalManager;
257 const keybindingsManager = yield* Keybindings;262 const keybindingsManager = yield* Keybindings;
...@@ -429,7 +434,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -429,7 +434,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
429 res.end(body);434 res.end(body);
430 };435 };
431436
432 void Effect.runPromise(437 void runPromise(
433 Effect.gen(function* () {438 Effect.gen(function* () {
434 const url = new URL(req.url ?? "/", `http://localhost:${port}`);439 const url = new URL(req.url ?? "/", `http://localhost:${port}`);
435 if (tryHandleProjectFaviconRequest(url, res)) {440 if (tryHandleProjectFaviconRequest(url, res)) {
...@@ -713,13 +718,8 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<...@@ -713,13 +718,8 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return<
713 );718 );
714 }719 }
715720
716 const runtimeServices = yield* Effect.services<721 const unsubscribeTerminalEvents = yield* terminalManager.subscribe((event) =>
717 ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path722 runPromise(pushBus.publishAll(WS_CHANNELS.terminalEvent, event)),
718 >();
719 const runPromise = Effect.runPromiseWith(runtimeServices);
720
721 const unsubscribeTerminalEvents = yield* terminalManager.subscribe(
722 (event) => void Effect.runPromise(pushBus.publishAll(WS_CHANNELS.terminalEvent, event)),
723 );723 );
724 yield* Effect.addFinalizer(() => Effect.sync(() => unsubscribeTerminalEvents()));724 yield* Effect.addFinalizer(() => Effect.sync(() => unsubscribeTerminalEvents()));
725 yield* readiness.markTerminalSubscriptionsReady;725 yield* readiness.markTerminalSubscriptionsReady;
bun.lock+11-19
...@@ -173,13 +173,13 @@...@@ -173,13 +173,13 @@
173 "vite": "^8.0.0",173 "vite": "^8.0.0",
174 },174 },
175 "catalog": {175 "catalog": {
176 "@effect/language-service": "0.75.1",176 "@effect/language-service": "0.84.1",
177 "@effect/platform-node": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/platform-node@8881a9b",177 "@effect/platform-node": "4.0.0-beta.42",
178 "@effect/sql-sqlite-bun": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/sql-sqlite-bun@8881a9b",178 "@effect/sql-sqlite-bun": "4.0.0-beta.42",
179 "@effect/vitest": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/vitest@8881a9b",179 "@effect/vitest": "4.0.0-beta.42",
180 "@types/bun": "^1.3.9",180 "@types/bun": "^1.3.9",
181 "@types/node": "^24.10.13",181 "@types/node": "^24.10.13",
182 "effect": "https://pkg.pr.new/Effect-TS/effect-smol/effect@8881a9b",182 "effect": "4.0.0-beta.42",
183 "tsdown": "^0.20.3",183 "tsdown": "^0.20.3",
184 "typescript": "^5.7.3",184 "typescript": "^5.7.3",
185 "vitest": "^4.0.0",185 "vitest": "^4.0.0",
...@@ -265,15 +265,15 @@...@@ -265,15 +265,15 @@
265265
266 "@dnd-kit/utilities": ["@dnd-kit/utilities@3.2.2", "", { "dependencies": { "tslib": "^2.0.0" }, "peerDependencies": { "react": ">=16.8.0" } }, "sha512-+MKAJEOfaBe5SmV6t34p80MMKhjvUz0vRrvVJbPT0WElzaOJ/1xs+D+KDv+tD/NE5ujfrChEcshd4fLn0wpiqg=="],266 "@dnd-kit/utilities": ["@dnd-kit/utilities@3.2.2", "", { "dependencies": { "tslib": "^2.0.0" }, "peerDependencies": { "react": ">=16.8.0" } }, "sha512-+MKAJEOfaBe5SmV6t34p80MMKhjvUz0vRrvVJbPT0WElzaOJ/1xs+D+KDv+tD/NE5ujfrChEcshd4fLn0wpiqg=="],
267267
268 "@effect/language-service": ["@effect/language-service@0.75.1", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-g9xD2tAQgRFpYC2YgpZq02VeSL5fBbFJ0B/g1o+14NuNmwtaYJc7SjiLWAA9eyhJHosNrn6h1Ye+Kx6j5mN0AA=="],268 "@effect/language-service": ["@effect/language-service@0.84.1", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-YUqjJU24HeYgPV453cR2fDqkZ+zZKMuxGnmxWAPscWJ6gt6FB7JZohMCOczRTIOGPrQMcloJX7BjCaPu+RNhpw=="],
269269
270 "@effect/platform-node": ["@effect/platform-node@https://pkg.pr.new/Effect-TS/effect-smol/@effect/platform-node@8881a9b", { "dependencies": { "@effect/platform-node-shared": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/platform-node-shared@8881a9b606d84a6f5eb6615279138322984f5368", "mime": "^4.1.0", "undici": "^7.20.0" }, "peerDependencies": { "effect": "^4.0.0-beta.25", "ioredis": "^5.7.0" } }],270 "@effect/platform-node": ["@effect/platform-node@4.0.0-beta.42", "", { "dependencies": { "@effect/platform-node-shared": "^4.0.0-beta.42", "mime": "^4.1.0", "undici": "^7.24.0" }, "peerDependencies": { "effect": "^4.0.0-beta.42", "ioredis": "^5.7.0" } }, "sha512-kbdRML2FBa4q8U8rZQcnmLKZ5zN/z1bAA7t5D1/UsBHZqJgnfRgu1CP6kaEfb1Nie6YyaWshxTktZQryjvW/Yg=="],
271271
272 "@effect/platform-node-shared": ["@effect/platform-node-shared@https://pkg.pr.new/Effect-TS/effect-smol/@effect/platform-node-shared@8881a9b606d84a6f5eb6615279138322984f5368", { "dependencies": { "@types/ws": "^8.18.1", "ws": "^8.19.0" }, "peerDependencies": { "effect": "^4.0.0-beta.25" } }],272 "@effect/platform-node-shared": ["@effect/platform-node-shared@4.0.0-beta.42", "", { "dependencies": { "@types/ws": "^8.18.1", "ws": "^8.19.0" }, "peerDependencies": { "effect": "^4.0.0-beta.42" } }, "sha512-PC+lxLsrwob3+nBChAPrQq32olCeyApgXBvs1NrRsoArLViNT76T/68CttuCAksCZj5e1bZ1ZibLPel3vUmx2g=="],
273273
274 "@effect/sql-sqlite-bun": ["@effect/sql-sqlite-bun@https://pkg.pr.new/Effect-TS/effect-smol/@effect/sql-sqlite-bun@8881a9b", { "peerDependencies": { "effect": "^4.0.0-beta.25" } }],274 "@effect/sql-sqlite-bun": ["@effect/sql-sqlite-bun@4.0.0-beta.42", "", { "peerDependencies": { "effect": "^4.0.0-beta.42" } }, "sha512-Ah2QfkeV+I9r5OBVJijSDnFXCv51giBXngSwhju5gefc0uWiM3G1tsYAqrNX24HlvFFEnOAZqNf/Sq1h4NqOAA=="],
275275
276 "@effect/vitest": ["@effect/vitest@https://pkg.pr.new/Effect-TS/effect-smol/@effect/vitest@8881a9b", { "peerDependencies": { "effect": "^4.0.0-beta.25", "vitest": "^3.0.0 || ^4.0.0" } }],276 "@effect/vitest": ["@effect/vitest@4.0.0-beta.42", "", { "peerDependencies": { "effect": "^4.0.0-beta.42", "vitest": "^3.0.0 || ^4.0.0" } }, "sha512-/11arjUnCRhIrBRvOn/nrbg5p/FadjAPvStddZlpl1VrCxtB2s0n39cbG9uTyDdf1ZrRBG73Upo1ZDF1CTWy8w=="],
277277
278 "@electron/get": ["@electron/get@2.0.3", "", { "dependencies": { "debug": "^4.1.1", "env-paths": "^2.2.0", "fs-extra": "^8.1.0", "got": "^11.8.5", "progress": "^2.0.3", "semver": "^6.2.0", "sumchecker": "^3.0.1" }, "optionalDependencies": { "global-agent": "^3.0.0" } }, "sha512-Qkzpg2s9GnVV2I2BjRksUi43U5e6+zaQMcjoJy0C+C5oxaKl+fmckGDQFtRpZpZV0NQekuZZ+tGz7EA9TVnQtQ=="],278 "@electron/get": ["@electron/get@2.0.3", "", { "dependencies": { "debug": "^4.1.1", "env-paths": "^2.2.0", "fs-extra": "^8.1.0", "got": "^11.8.5", "progress": "^2.0.3", "semver": "^6.2.0", "sumchecker": "^3.0.1" }, "optionalDependencies": { "global-agent": "^3.0.0" } }, "sha512-Qkzpg2s9GnVV2I2BjRksUi43U5e6+zaQMcjoJy0C+C5oxaKl+fmckGDQFtRpZpZV0NQekuZZ+tGz7EA9TVnQtQ=="],
279279
...@@ -1017,7 +1017,7 @@...@@ -1017,7 +1017,7 @@
10171017
1018 "dunder-proto": ["dunder-proto@1.0.1", "", { "dependencies": { "call-bind-apply-helpers": "^1.0.1", "es-errors": "^1.3.0", "gopd": "^1.2.0" } }, "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A=="],1018 "dunder-proto": ["dunder-proto@1.0.1", "", { "dependencies": { "call-bind-apply-helpers": "^1.0.1", "es-errors": "^1.3.0", "gopd": "^1.2.0" } }, "sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A=="],
10191019
1020 "effect": ["effect@https://pkg.pr.new/Effect-TS/effect-smol/effect@8881a9b", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }],1020 "effect": ["effect@4.0.0-beta.42", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }, "sha512-c1UrRP+tLzyHb4Fepl8XBDJlLQLkrcMXrRBba441GQRxMbeQ/aIOSFcBwSda1iMJ5l9F0lYc3Bhe33/whrmavQ=="],
10211021
1022 "electron": ["electron@40.6.0", "", { "dependencies": { "@electron/get": "^2.0.0", "@types/node": "^24.9.0", "extract-zip": "^2.0.1" }, "bin": { "electron": "cli.js" } }, "sha512-ett8W+yOFGDuM0vhJMamYSkrbV3LoaffzJd9GfjI96zRAxyrNqUSKqBpf/WGbQCweDxX2pkUCUfrv4wwKpsFZA=="],1022 "electron": ["electron@40.6.0", "", { "dependencies": { "@electron/get": "^2.0.0", "@types/node": "^24.9.0", "extract-zip": "^2.0.1" }, "bin": { "electron": "cli.js" } }, "sha512-ett8W+yOFGDuM0vhJMamYSkrbV3LoaffzJd9GfjI96zRAxyrNqUSKqBpf/WGbQCweDxX2pkUCUfrv4wwKpsFZA=="],
10231023
...@@ -1905,14 +1905,6 @@...@@ -1905,14 +1905,6 @@
19051905
1906 "@babel/types/@babel/helper-validator-identifier": ["@babel/helper-validator-identifier@7.28.5", "", {}, "sha512-qSs4ifwzKJSV39ucNjsvc6WVHs6b7S03sOh2OcHF9UHfVPqWWALUsNUVzhSBiItjRZoLHx7nIarVjqKVusUZ1Q=="],1906 "@babel/types/@babel/helper-validator-identifier": ["@babel/helper-validator-identifier@7.28.5", "", {}, "sha512-qSs4ifwzKJSV39ucNjsvc6WVHs6b7S03sOh2OcHF9UHfVPqWWALUsNUVzhSBiItjRZoLHx7nIarVjqKVusUZ1Q=="],
19071907
1908 "@effect/platform-node/effect": ["effect@4.0.0-beta.33", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }, "sha512-ln9emWPd1SemokSdOV43r2CbH1j8GTe9qbPvttmh9/j2OR0WNmj7UpjbN34llQgF9QV4IdcN6QdV2w8G7B7RyQ=="],
1909
1910 "@effect/platform-node-shared/effect": ["effect@4.0.0-beta.33", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }, "sha512-ln9emWPd1SemokSdOV43r2CbH1j8GTe9qbPvttmh9/j2OR0WNmj7UpjbN34llQgF9QV4IdcN6QdV2w8G7B7RyQ=="],
1911
1912 "@effect/sql-sqlite-bun/effect": ["effect@4.0.0-beta.33", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }, "sha512-ln9emWPd1SemokSdOV43r2CbH1j8GTe9qbPvttmh9/j2OR0WNmj7UpjbN34llQgF9QV4IdcN6QdV2w8G7B7RyQ=="],
1913
1914 "@effect/vitest/effect": ["effect@4.0.0-beta.33", "", { "dependencies": { "@standard-schema/spec": "^1.1.0", "fast-check": "^4.5.3", "find-my-way-ts": "^0.1.6", "ini": "^6.0.0", "kubernetes-types": "^1.30.0", "msgpackr": "^1.11.8", "multipasta": "^0.2.7", "toml": "^3.0.0", "uuid": "^13.0.0", "yaml": "^2.8.2" } }, "sha512-ln9emWPd1SemokSdOV43r2CbH1j8GTe9qbPvttmh9/j2OR0WNmj7UpjbN34llQgF9QV4IdcN6QdV2w8G7B7RyQ=="],
1915
1916 "@electron/get/fs-extra": ["fs-extra@8.1.0", "", { "dependencies": { "graceful-fs": "^4.2.0", "jsonfile": "^4.0.0", "universalify": "^0.1.0" } }, "sha512-yhlQgA6mnOJUKOsRUFsgJdQCvkKhcz8tlZG5HBQfReYZy46OwLcY+Zia0mtdHsOo9y/hP+CxMN0TU9QxoOtG4g=="],1908 "@electron/get/fs-extra": ["fs-extra@8.1.0", "", { "dependencies": { "graceful-fs": "^4.2.0", "jsonfile": "^4.0.0", "universalify": "^0.1.0" } }, "sha512-yhlQgA6mnOJUKOsRUFsgJdQCvkKhcz8tlZG5HBQfReYZy46OwLcY+Zia0mtdHsOo9y/hP+CxMN0TU9QxoOtG4g=="],
19171909
1918 "@electron/get/semver": ["semver@6.3.1", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA=="],1910 "@electron/get/semver": ["semver@6.3.1", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA=="],
package.json+5-5
...@@ -8,11 +8,11 @@...@@ -8,11 +8,11 @@
8 "scripts"8 "scripts"
9 ],9 ],
10 "catalog": {10 "catalog": {
11 "effect": "https://pkg.pr.new/Effect-TS/effect-smol/effect@8881a9b",11 "effect": "4.0.0-beta.42",
12 "@effect/platform-node": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/platform-node@8881a9b",12 "@effect/platform-node": "4.0.0-beta.42",
13 "@effect/sql-sqlite-bun": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/sql-sqlite-bun@8881a9b",13 "@effect/sql-sqlite-bun": "4.0.0-beta.42",
14 "@effect/vitest": "https://pkg.pr.new/Effect-TS/effect-smol/@effect/vitest@8881a9b",14 "@effect/vitest": "4.0.0-beta.42",
15 "@effect/language-service": "0.75.1",15 "@effect/language-service": "0.84.1",
16 "@types/bun": "^1.3.9",16 "@types/bun": "^1.3.9",
17 "@types/node": "^24.10.13",17 "@types/node": "^24.10.13",
18 "tsdown": "^0.20.3",18 "tsdown": "^0.20.3",
packages/shared/src/DrainableWorker.ts+29-54
...@@ -1,14 +1,12 @@...@@ -1,14 +1,12 @@
1/**1/**
2 * DrainableWorker - A queue-based worker that exposes a `drain()` effect.2 * DrainableWorker - A queue-based worker with deterministic `drain()`.
3 *3 *
4 * Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds4 * Tracks outstanding work in STM so `drain()` resolves only when no items
5 * a signal that resolves when the queue is empty **and** the current item5 * are queued or in flight. Useful in tests instead of timing-based waits.
6 * has finished processing. This lets tests replace timing-sensitive
7 * `Effect.sleep` calls with deterministic `drain()`.
8 *6 *
9 * @module DrainableWorker7 * @module DrainableWorker
10 */8 */
11import { Deferred, Effect, Queue, Ref } from "effect";9import { Effect, TxQueue, TxRef } from "effect";
12import type { Scope } from "effect";10import type { Scope } from "effect";
1311
14export interface DrainableWorker<A> {12export interface DrainableWorker<A> {
...@@ -39,63 +37,40 @@ export const makeDrainableWorker = <A, E, R>(...@@ -39,63 +37,40 @@ export const makeDrainableWorker = <A, E, R>(
39 process: (item: A) => Effect.Effect<void, E, R>,37 process: (item: A) => Effect.Effect<void, E, R>,
40): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>38): Effect.Effect<DrainableWorker<A>, never, Scope.Scope | R> =>
41 Effect.gen(function* () {39 Effect.gen(function* () {
42 const queue = yield* Queue.unbounded<A>();40 const ref = yield* TxRef.make(0);
43 const initialIdle = yield* Deferred.make<void>();
44 yield* Deferred.succeed(initialIdle, undefined).pipe(Effect.orDie);
45 const state = yield* Ref.make({
46 outstanding: 0,
47 idle: initialIdle,
48 });
4941
50 yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid));42 const queue = yield* Effect.acquireRelease(TxQueue.unbounded<A>(), (queue) =>
43 TxQueue.shutdown(queue),
44 );
5145
52 const finishOne = Ref.modify(state, (current) => {46 const takeItem = Effect.tx(
53 const remaining = Math.max(0, current.outstanding - 1);47 Effect.gen(function* () {
54 return [48 const item = yield* TxQueue.take(queue);
55 remaining === 0 ? current.idle : null,49 yield* TxRef.update(ref, (n) => n + 1);
56 {50 return item;
57 outstanding: remaining,51 }),
58 idle: current.idle,
59 },
60 ] as const;
61 }).pipe(
62 Effect.flatMap((idle) =>
63 idle === null ? Effect.void : Deferred.succeed(idle, undefined).pipe(Effect.orDie),
64 ),
65 );52 );
6653
67 yield* Effect.forkScoped(54 yield* takeItem.pipe(
68 Effect.forever(55 Effect.flatMap((item) =>
69 Queue.take(queue).pipe(56 process(item).pipe(Effect.ensuring(TxRef.update(ref, (n) => n - 1))),
70 Effect.flatMap((item) => process(item).pipe(Effect.ensuring(finishOne))),
71 ),
72 ),57 ),
58 Effect.forever,
59 Effect.forkScoped,
73 );60 );
7461
75 const enqueue: DrainableWorker<A>["enqueue"] = (item) =>62 const drain: DrainableWorker<A>["drain"] = Effect.tx(
76 Effect.gen(function* () {63 Effect.gen(function* () {
77 const nextIdle = yield* Deferred.make<void>();64 const inFlight = yield* TxRef.get(ref);
78 yield* Ref.update(state, (current) =>65 const isEmpty = yield* TxQueue.isEmpty(queue);
79 current.outstanding === 066 if (inFlight > 0 || !isEmpty) {
80 ? {67 return yield* Effect.txRetry;
81 outstanding: 1,
82 idle: nextIdle,
83 }
84 : {
85 outstanding: current.outstanding + 1,
86 idle: current.idle,
87 },
88 );
89
90 const accepted = yield* Queue.offer(queue, item);
91 if (!accepted) {
92 yield* finishOne;
93 }68 }
94 });69 }),
95
96 const drain: DrainableWorker<A>["drain"] = Ref.get(state).pipe(
97 Effect.flatMap(({ idle }) => Deferred.await(idle)),
98 );70 );
9971
100 return { enqueue, drain } satisfies DrainableWorker<A>;72 return {
73 enqueue: (item) => TxQueue.offer(queue, item),
74 drain,
75 } satisfies DrainableWorker<A>;
101 });76 });