diff --git a/packages/core/schema.json b/packages/core/schema.json index 8ec46b52df..d419909444 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,10 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "68b6e591-ccd7-415e-aa5f-94e0ec9da5aa", - "prevIds": [ - "b0355fd9-bf41-42e3-9dca-76107de27ecd" - ], + "id": "b0355fd9-bf41-42e3-9dca-76107de27ecd", + "prevIds": ["95328a41-789d-44de-9643-6ac6ecd6b4ec"], "ddl": [ { "name": "workspace", @@ -66,10 +64,6 @@ "name": "part", "entityType": "tables" }, - { - "name": "session_execution_recovery", - "entityType": "tables" - }, { "name": "session_input", "entityType": "tables" @@ -990,26 +984,6 @@ "entityType": "columns", "table": "part" }, - { - "type": "text", - "notNull": false, - "autoincrement": false, - "default": null, - "generated": null, - "name": "session_id", - "entityType": "columns", - "table": "session_execution_recovery" - }, - { - "type": "integer", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "interrupted_seq", - "entityType": "columns", - "table": "session_execution_recovery" - }, { "type": "text", "notNull": false, @@ -1601,13 +1575,9 @@ "table": "session_share" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1616,13 +1586,9 @@ "table": "workspace" }, { - "columns": [ - "active_account_id" - ], + "columns": ["active_account_id"], "tableTo": "account", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "SET NULL", "nameExplicit": false, @@ -1631,13 +1597,9 @@ "table": "account_state" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "tableTo": "event_sequence", - "columnsTo": [ - "aggregate_id" - ], + "columnsTo": ["aggregate_id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1646,13 +1608,9 @@ "table": "event" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1661,13 +1619,9 @@ "table": "permission" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1676,13 +1630,9 @@ "table": "project_directory" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1691,13 +1641,9 @@ "table": "instruction_checkpoint" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1706,13 +1652,9 @@ "table": "instruction_entry" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1721,13 +1663,9 @@ "table": "message" }, { - "columns": [ - "message_id" - ], + "columns": ["message_id"], "tableTo": "message", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1736,28 +1674,9 @@ "table": "part" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], - "onUpdate": "NO ACTION", - "onDelete": "CASCADE", - "nameExplicit": false, - "name": "fk_session_execution_recovery_session_id_session_id_fk", - "entityType": "fks", - "table": "session_execution_recovery" - }, - { - "columns": [ - "session_id" - ], - "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1766,13 +1685,9 @@ "table": "session_input" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1781,13 +1696,9 @@ "table": "session_message" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1796,13 +1707,9 @@ "table": "session" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1811,13 +1718,9 @@ "table": "todo" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1826,193 +1729,140 @@ "table": "session_share" }, { - "columns": [ - "email", - "url" - ], + "columns": ["email", "url"], "nameExplicit": false, "name": "control_account_pk", "entityType": "pks", "table": "control_account" }, { - "columns": [ - "project_id", - "directory" - ], + "columns": ["project_id", "directory"], "nameExplicit": false, "name": "project_directory_pk", "entityType": "pks", "table": "project_directory" }, { - "columns": [ - "session_id", - "key" - ], + "columns": ["session_id", "key"], "nameExplicit": false, "name": "instruction_entry_pk", "entityType": "pks", "table": "instruction_entry" }, { - "columns": [ - "session_id", - "position" - ], + "columns": ["session_id", "position"], "nameExplicit": false, "name": "todo_pk", "entityType": "pks", "table": "todo" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "workspace_pk", "table": "workspace", "entityType": "pks" }, { - "columns": [ - "name" - ], + "columns": ["name"], "nameExplicit": false, "name": "data_migration_pk", "table": "data_migration", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_state_pk", "table": "account_state", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_pk", "table": "account", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "credential_pk", "table": "credential", "entityType": "pks" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "nameExplicit": false, "name": "event_sequence_pk", "table": "event_sequence", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "event_pk", "table": "event", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "permission_pk", "table": "permission", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "project_pk", "table": "project", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "instruction_checkpoint_pk", "table": "instruction_checkpoint", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "message_pk", "table": "message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "part_pk", "table": "part", "entityType": "pks" }, { - "columns": [ - "session_id" - ], - "nameExplicit": false, - "name": "session_execution_recovery_pk", - "table": "session_execution_recovery", - "entityType": "pks" - }, - { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_input_pk", "table": "session_input", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_message_pk", "table": "session_message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_pk", "table": "session", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "session_share_pk", "table": "session_share", @@ -2348,4 +2198,4 @@ } ], "renames": [] -} \ No newline at end of file +} diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index e84ed55fb5..99ca206c1f 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -48,6 +48,5 @@ export const migrations = ( import("./migration/20260705180000_rename_instructions"), import("./migration/20260706223930_add-session-fork"), import("./migration/20260707010146_durable_session_inbox"), - import("./migration/20260708004136_session-execution-recovery"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260708004136_session-execution-recovery.ts b/packages/core/src/database/migration/20260708004136_session-execution-recovery.ts deleted file mode 100644 index dce74446c0..0000000000 --- a/packages/core/src/database/migration/20260708004136_session-execution-recovery.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { Effect } from "effect" -import type { DatabaseMigration } from "../migration" - -export default { - id: "20260708004136_session-execution-recovery", - up(tx) { - return Effect.gen(function* () { - yield* tx.run(` - CREATE TABLE \`session_execution_recovery\` ( - \`session_id\` text PRIMARY KEY, - \`interrupted_seq\` integer NOT NULL, - CONSTRAINT \`fk_session_execution_recovery_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE - ); - `) - }) - }, -} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 015c51e93d..0bb6050bf6 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -166,13 +166,6 @@ export default { CONSTRAINT \`fk_part_message_id_message_id_fk\` FOREIGN KEY (\`message_id\`) REFERENCES \`message\`(\`id\`) ON DELETE CASCADE ); `) - yield* tx.run(` - CREATE TABLE \`session_execution_recovery\` ( - \`session_id\` text PRIMARY KEY, - \`interrupted_seq\` integer NOT NULL, - CONSTRAINT \`fk_session_execution_recovery_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE - ); - `) yield* tx.run(` CREATE TABLE \`session_input\` ( \`id\` text PRIMARY KEY, diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index 41b73622f7..2ea2991a7f 100644 --- a/packages/core/src/session/execution/local.ts +++ b/packages/core/src/session/execution/local.ts @@ -9,11 +9,11 @@ import { SessionEvent } from "../event" import { SessionRunCoordinator } from "../run-coordinator" import { SessionRunner } from "../runner" import { SessionSchema } from "../schema" -import { SessionExecutionRecoveryTable, SessionTable } from "../sql" import { SessionStore } from "../store" import { SessionExecution } from "../execution" import { toSessionError } from "../to-session-error" import { UserInterruptedError } from "../error" +import { EffectFlock } from "../../util/effect-flock" export function terminal(exit: Exit.Exit, reason?: "user" | "shutdown" | "superseded") { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } @@ -23,14 +23,12 @@ export function terminal(exit: Exit.Exit, reason?: return { type: "failed" as const, error: toSessionError(failure) } } -export const claimSessionsInterruptedByShutdown = Effect.fn("SessionExecutionLocal.claimSessionsInterruptedByShutdown")( +export const sessionsInterruptedByShutdown = Effect.fn("SessionExecutionLocal.sessionsInterruptedByShutdown")( function* (db: Database.Interface["db"]) { - const claimed = yield* db.all<{ sessionID: string }>(sql` - INSERT INTO ${SessionExecutionRecoveryTable} (session_id, interrupted_seq) - SELECT lifecycle.aggregate_id, lifecycle.seq + const latest = yield* db.all<{ sessionID: string; type: string; reason: string | null }>(sql` + SELECT aggregate_id AS sessionID, type, json_extract(data, '$.reason') AS reason FROM ( SELECT aggregate_id, type, data, - seq, row_number() OVER (PARTITION BY aggregate_id ORDER BY seq DESC) AS rank FROM ${EventTable} WHERE type IN ( @@ -39,16 +37,16 @@ export const claimSessionsInterruptedByShutdown = Effect.fn("SessionExecutionLoc ${EventV2.versionedType(SessionEvent.Execution.Failed.type, 1)}, ${EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1)} ) - ) AS lifecycle - INNER JOIN ${SessionTable} ON ${SessionTable.id} = lifecycle.aggregate_id - WHERE lifecycle.rank = 1 - AND lifecycle.type = ${EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1)} - AND json_extract(lifecycle.data, '$.reason') = 'shutdown' - ON CONFLICT (session_id) DO UPDATE SET interrupted_seq = excluded.interrupted_seq - WHERE interrupted_seq < excluded.interrupted_seq - RETURNING session_id AS sessionID + ) + WHERE rank = 1 `) - return claimed.map((event) => SessionSchema.ID.make(event.sessionID)) + return latest + .filter( + (event) => + event.type === EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1) && + event.reason === "shutdown", + ) + .map((event) => SessionSchema.ID.make(event.sessionID)) }, ) @@ -59,6 +57,7 @@ const layer = Layer.effect( const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service const events = yield* EventV2.Service + const flock = yield* EffectFlock.Service const { db } = yield* Database.Service const reportLifecycle = (sessionID: SessionSchema.ID, effect: Effect.Effect) => effect.pipe( @@ -111,19 +110,31 @@ const layer = Layer.effect( ), }) - const interrupted = yield* claimSessionsInterruptedByShutdown(db) - yield* Effect.forEach( - interrupted, - (sessionID) => - coordinator.run(sessionID).pipe( - Effect.tapCause((cause) => - Effect.logError("Failed to recover Session after shutdown", cause).pipe(Effect.annotateLogs({ sessionID })), - ), - Effect.ignore, - Effect.forkScoped, - ), - { discard: true }, - ) + yield* flock + .withLock( + Effect.gen(function* () { + const interrupted = yield* sessionsInterruptedByShutdown(db) + yield* Effect.forEach( + interrupted, + (sessionID) => + coordinator.run(sessionID).pipe( + Effect.tapCause((cause) => + Effect.logError("Failed to recover Session after shutdown", cause).pipe( + Effect.annotateLogs({ sessionID }), + ), + ), + Effect.ignore, + ), + { concurrency: "unbounded", discard: true }, + ) + }), + "session-shutdown-recovery", + ) + .pipe( + Effect.tapCause((cause) => Effect.logError("Failed to acquire Session recovery ownership", cause)), + Effect.ignore, + Effect.forkScoped, + ) return SessionExecution.Service.of({ active: coordinator.active, @@ -138,7 +149,7 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: SessionExecution.Service, layer, - deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, Database.node], + deps: [SessionStore.node, LocationServiceMap.node, EventV2.node, Database.node, EffectFlock.node], }) export * as SessionExecutionLocal from "./local" diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 2973ab97f2..e7d3bdca58 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -173,14 +173,6 @@ export const SessionInputTable = sqliteTable( ], ) -export const SessionExecutionRecoveryTable = sqliteTable("session_execution_recovery", { - session_id: text() - .$type() - .primaryKey() - .references(() => SessionTable.id, { onDelete: "cascade" }), - interrupted_seq: integer().notNull(), -}) - export const InstructionEntryTable = sqliteTable( "instruction_entry", { diff --git a/packages/core/test/session-execution-local.test.ts b/packages/core/test/session-execution-local.test.ts index 589a78ebed..3643e8a172 100644 --- a/packages/core/test/session-execution-local.test.ts +++ b/packages/core/test/session-execution-local.test.ts @@ -5,14 +5,11 @@ import { EventV2 } from "@opencode-ai/core/event" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { SessionEvent } from "@opencode-ai/core/session/event" -import { claimSessionsInterruptedByShutdown, terminal } from "@opencode-ai/core/session/execution/local" +import { sessionsInterruptedByShutdown, terminal } from "@opencode-ai/core/session/execution/local" import { SessionV2 } from "@opencode-ai/core/session" import { UserInterruptedError } from "@opencode-ai/core/session/error" import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" -import { sql } from "drizzle-orm" -import { Context, Effect, Exit, Layer } from "effect" -import path from "node:path" -import { tmpdir } from "./fixture/tmpdir" +import { Effect, Exit } from "effect" import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node]))) @@ -46,40 +43,7 @@ describe("SessionExecutionLocal lifecycle", () => { expect(terminal(Exit.fail(new UserInterruptedError()))).toEqual({ type: "interrupted", reason: "user" }) }) - test("claims shutdown recovery once across database connections", async () => { - await using tmp = await tmpdir() - const filename = path.join(tmp.path, "recovery.sqlite") - await Effect.runPromise( - Effect.scoped( - Effect.gen(function* () { - const first = Context.get(yield* Layer.build(Database.layerFromPath(filename)), Database.Service).db - const second = Context.get(yield* Layer.build(Database.layerFromPath(filename)), Database.Service).db - const sessionID = SessionV2.ID.make("ses_recover_concurrent") - yield* first.run(sql` - INSERT INTO project (id, worktree, sandboxes, time_created, time_updated) - VALUES ('prj_recovery_concurrent', '/tmp/recovery', '[]', 0, 0) - `) - yield* first.run(sql` - INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated) - VALUES (${sessionID}, 'prj_recovery_concurrent', ${sessionID}, '/tmp/recovery', ${sessionID}, 'test', 0, 0) - `) - yield* first.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES (${sessionID}, 1)`) - yield* first.run(sql` - INSERT INTO event (id, aggregate_id, seq, created, type, data) - VALUES ('evt_recovery_concurrent', ${sessionID}, 1, 0, 'session.execution.interrupted.1', '{"sessionID":"ses_recover_concurrent","reason":"shutdown"}') - `) - - const claims = yield* Effect.all( - [claimSessionsInterruptedByShutdown(first), claimSessionsInterruptedByShutdown(second)], - { concurrency: "unbounded" }, - ) - expect(claims.flat()).toEqual([sessionID]) - }), - ), - ) - }) - - it.effect("claims each latest shutdown interruption once", () => + it.effect("selects only sessions whose latest execution ended for shutdown", () => Effect.gen(function* () { const events = yield* EventV2.Service const { db } = yield* Database.Service @@ -88,17 +52,6 @@ describe("SessionExecutionLocal lifecycle", () => { const user = SessionV2.ID.make("ses_user") const crashed = SessionV2.ID.make("ses_crashed") - yield* db.run(sql` - INSERT INTO project (id, worktree, sandboxes, time_created, time_updated) - VALUES ('prj_recovery', '/tmp/recovery', '[]', 0, 0) - `) - yield* Effect.forEach([recover, completed, user, crashed], (sessionID) => - db.run(sql` - INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated) - VALUES (${sessionID}, 'prj_recovery', ${sessionID}, '/tmp/recovery', ${sessionID}, 'test', 0, 0) - `), - ) - yield* events.publish(SessionEvent.Execution.Started, { sessionID: recover }) yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: recover, reason: "shutdown" }) yield* events.publish(SessionEvent.InstructionsUpdated, { sessionID: recover, text: "later non-lifecycle event" }) @@ -108,16 +61,7 @@ describe("SessionExecutionLocal lifecycle", () => { yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: user, reason: "user" }) yield* events.publish(SessionEvent.Execution.Started, { sessionID: crashed }) - const claims = yield* Effect.all( - [claimSessionsInterruptedByShutdown(db), claimSessionsInterruptedByShutdown(db)], - { concurrency: "unbounded" }, - ) - expect(claims.flat()).toEqual([recover]) - - yield* events.publish(SessionEvent.Execution.Started, { sessionID: recover }) - yield* events.publish(SessionEvent.Execution.Interrupted, { sessionID: recover, reason: "shutdown" }) - expect(yield* claimSessionsInterruptedByShutdown(db)).toEqual([recover]) - expect(yield* claimSessionsInterruptedByShutdown(db)).toEqual([]) + expect(yield* sessionsInterruptedByShutdown(db)).toEqual([recover]) }), ) })