diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index fb5eb276103..2581bd5ec59 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -2,7 +2,7 @@ export * as Database from "./database.js" import { EffectDrizzleSqlite } from "./drizzle.js" import { sqliteLayer, supportsForeignKeyToggle, supportsTuningPragmas } from "#sqlite" -import { Context, Effect, Layer, Schema } from "effect" +import { Context, Effect, Layer, Schema, Semaphore } from "effect" import type { SqlClient } from "effect/unstable/sql" import { Global } from "@opencode-ai/util/global" import { isAbsolute, join } from "path" @@ -23,30 +23,52 @@ export type Options = typeof Options.Type export class Service extends Context.Service()("@opencode/storage/Database") {} -const databaseLayer = Layer.effect( - Service, - Effect.gen(function* () { - const db = yield* makeDatabase +// The bootstrap lock is scoped to the database being built, never to this +// module: on workerd every Durable Object in an isolate shares module state, and +// releasing a shared semaphore resumes the waiting object's fiber inside the +// releasing object's I/O context, where its first storage call is rejected as +// cross-object I/O. +const databaseLayer = (lock: Effect.Effect) => + Layer.effect( + Service, + Effect.gen(function* () { + const db = yield* makeDatabase - if (supportsTuningPragmas) { - yield* db.run("PRAGMA journal_mode = WAL") - yield* db.run("PRAGMA synchronous = NORMAL") - yield* db.run("PRAGMA busy_timeout = 5000") - yield* db.run("PRAGMA cache_size = -64000") - yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") - } - // Durable Object SQLite always enforces foreign keys and rejects the pragma. - if (supportsForeignKeyToggle) yield* db.run("PRAGMA foreign_keys = ON") - yield* DatabaseMigration.apply(db) + if (supportsTuningPragmas) { + yield* db.run("PRAGMA journal_mode = WAL") + yield* db.run("PRAGMA synchronous = NORMAL") + yield* db.run("PRAGMA busy_timeout = 5000") + yield* db.run("PRAGMA cache_size = -64000") + yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + } + // Durable Object SQLite always enforces foreign keys and rejects the pragma. + if (supportsForeignKeyToggle) yield* db.run("PRAGMA foreign_keys = ON") + const semaphore = yield* lock + yield* semaphore.withPermit(DatabaseMigration.apply(db)) - return { db } - }).pipe(Effect.orDie), -) + return { db } + }).pipe(Effect.orDie), + ) + +// Two instances over one file bootstrap the same schema, so file databases +// share a lock per path. Each in-memory database is its own connection. +const locks = new Map() + +function lockFor(filename: string) { + const existing = locks.get(filename) + if (existing) return existing + const lock = Semaphore.makeUnsafe(1) + locks.set(filename, lock) + return lock +} export function layer(options: Options = { path: ":memory:" }) { return Layer.unwrap( Effect.gen(function* () { - const provide = (filename: string) => layerFromClient.pipe(Layer.provide(sqliteLayer({ filename }))) + const provide = (filename: string) => + databaseLayer(filename === ":memory:" ? Semaphore.make(1) : Effect.succeed(lockFor(filename))).pipe( + Layer.provide(sqliteLayer({ filename })), + ) const filename = options.path ?? ":memory:" if (filename === ":memory:" || isAbsolute(filename)) return provide(filename) const global = yield* Global.Service @@ -58,8 +80,11 @@ export function layer(options: Options = { path: ":memory:" }) { // The database service over an injected SqlClient, for runtimes that receive // database storage instead of opening a filesystem path. Any client provided // here still goes through the pragma guards and migrations; Global is required -// because migrations may read it (the v1 import). -export const layerFromClient: Layer.Layer = databaseLayer +// because migrations may read it (the v1 import). The lock is created per build +// because every Durable Object builds this layer over its own storage. +export const layerFromClient: Layer.Layer = databaseLayer( + Semaphore.make(1), +) export function configured(options?: Options) { return makeGlobalNode({ service: Service, layer: layer(options), deps: [Global.node] }) diff --git a/packages/core/src/database/migration.ts b/packages/core/src/database/migration.ts index a9104a0f2a4..2d0c05a2a89 100644 --- a/packages/core/src/database/migration.ts +++ b/packages/core/src/database/migration.ts @@ -1,7 +1,7 @@ export * as DatabaseMigration from "./migration.js" import { sql } from "drizzle-orm" -import { Effect, Semaphore } from "effect" +import { Effect } from "effect" import { supportsForeignKeyToggle } from "#sqlite" import type { EffectDrizzleSqlite } from "./drizzle.js" import { migrations } from "./migration.gen.js" @@ -10,7 +10,6 @@ import { Global } from "@opencode-ai/util/global" type Database = EffectDrizzleSqlite.EffectSQLiteDatabase type Transaction = Parameters[0]>[0] -const lock = Semaphore.makeUnsafe(1) export type Migration = { id: string @@ -18,38 +17,38 @@ export type Migration = { up: (tx: Transaction) => Effect.Effect } +// Not serialized here: the Database layer holds a lock scoped to the database +// it is bootstrapping, since two instances over one file must not race. export function apply(db: Database) { - return lock.withPermit( - Effect.gen(function* () { - // OpenCode owns the unprefixed table namespace. Embedders sharing this - // database may own underscore-prefixed tables, which bootstrap ignores. - const tables = yield* db.all<{ name: string }>( - sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND substr(name, 1, 1) <> '_'`, - ) - if (tables.some((table) => table.name === "session" || table.name === "session_v2")) - return yield* applyOnly(db, migrations) - if (tables.length > 0) return yield* Effect.die(new Error("Database is not empty and has no session table")) - const started = Date.now() - yield* Effect.logInfo("database schema bootstrap started", { migrations: migrations.length }) - yield* db.transaction((tx) => - Effect.gen(function* () { - yield* schema.up(tx) - yield* tx.run( - sql`CREATE TABLE ${sql.identifier("migration")} (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`, - ) - yield* Effect.forEach(migrations, (migration) => - tx.run( - sql`INSERT INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`, - ), - ) - }), - ) - yield* Effect.logInfo("database schema bootstrap completed", { - migrations: migrations.length, - durationMs: Date.now() - started, - }) - }), - ) + return Effect.gen(function* () { + // OpenCode owns the unprefixed table namespace. Embedders sharing this + // database may own underscore-prefixed tables, which bootstrap ignores. + const tables = yield* db.all<{ name: string }>( + sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND substr(name, 1, 1) <> '_'`, + ) + if (tables.some((table) => table.name === "session" || table.name === "session_v2")) + return yield* applyOnly(db, migrations) + if (tables.length > 0) return yield* Effect.die(new Error("Database is not empty and has no session table")) + const started = Date.now() + yield* Effect.logInfo("database schema bootstrap started", { migrations: migrations.length }) + yield* db.transaction((tx) => + Effect.gen(function* () { + yield* schema.up(tx) + yield* tx.run( + sql`CREATE TABLE ${sql.identifier("migration")} (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`, + ) + yield* Effect.forEach(migrations, (migration) => + tx.run( + sql`INSERT INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`, + ), + ) + }), + ) + yield* Effect.logInfo("database schema bootstrap completed", { + migrations: migrations.length, + durationMs: Date.now() - started, + }) + }) } export function applyOnly(db: Database, input: Migration[]) { diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index 03814cda009..3708a5dc6ca 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -4,14 +4,15 @@ import { fileURLToPath } from "url" import path from "path" import { SqliteClient } from "@effect/sql-sqlite-bun" import { EffectDrizzleSqlite } from "@opencode-ai/core/database/drizzle" -import { Effect, Layer } from "effect" +import { Deferred, Effect, Fiber, Layer } from "effect" +import { Reactivity } from "effect/unstable/reactivity" +import { SqlClient, Statement } from "effect/unstable/sql" import { sql } from "drizzle-orm" import { DatabaseMigration } from "@opencode-ai/core/database/migration" import { migrations } from "@opencode-ai/core/database/migration.gen" import workspaceNameMigration from "@opencode-ai/core/database/migration/20260410174513_workspace-name" import { Database } from "@opencode-ai/core/database/database" import { tmpdir } from "./fixture/tmpdir" -import type { SqlClient } from "effect/unstable/sql/SqlClient" import legacyCredentialsMigration from "@opencode-ai/core/database/migration/20260805200742_import_legacy_credentials" import worktreeMigration from "@opencode-ai/core/database/migration/20260812213948_worktree" import previousV2Migration from "@opencode-ai/core/database/migration/20260804233008_loose_psylocke" @@ -22,7 +23,7 @@ import sessionViewedStateMigration from "@opencode-ai/core/database/migration/20 import { Global } from "@opencode-ai/util/global" const run = ( - effect: Effect.Effect, + effect: Effect.Effect, global = Global.make({ data: path.join(process.cwd(), ".test-data") }), ) => Effect.runPromise( @@ -35,6 +36,31 @@ const run = ( const makeDb = EffectDrizzleSqlite.makeWithDefaults() +// A real in-memory SqlClient whose schema inspection signals `arrived` and then +// waits on `gate`. Bootstrap inspects the schema as its first locked statement, +// so a database built over this client parks while holding its migration lock. +const parkedClient = (arrived: Deferred.Deferred, gate: Deferred.Deferred) => + Layer.effect( + SqlClient.SqlClient, + Effect.gen(function* () { + const client = yield* SqlClient.SqlClient + const connection = yield* client.reserve + const park = (query: string, effect: Effect.Effect) => + query.includes("sqlite_master") + ? Deferred.succeed(arrived, undefined).pipe(Effect.andThen(Deferred.await(gate)), Effect.andThen(effect)) + : effect + return yield* SqlClient.make({ + acquirer: Effect.succeed({ + ...connection, + execute: (query, params, transform) => park(query, connection.execute(query, params, transform)), + executeRaw: (query, params) => park(query, connection.executeRaw(query, params)), + }), + compiler: Statement.makeCompilerSqlite(), + spanAttributes: [], + }) + }), + ).pipe(Layer.provide(SqliteClient.layer({ filename: ":memory:", disableWAL: true })), Layer.provide(Reactivity.layer)) + describe("DatabaseMigration", () => { test("defaults missing workspace names while preserving legacy workspace data", async () => { await run( @@ -112,6 +138,31 @@ describe("DatabaseMigration", () => { ) }) + test("bootstraps distinct databases without waiting on each other's lock", async () => { + await Effect.runPromise( + Effect.gen(function* () { + const arrived = yield* Deferred.make() + const gate = yield* Deferred.make() + // Park the first database inside its bootstrap, after it holds its lock. + const parked = yield* Effect.forkScoped( + Layer.build(Database.layerFromClient.pipe(Layer.provide(parkedClient(arrived, gate)))), + ) + yield* Deferred.await(arrived) + + yield* Layer.build( + Database.layerFromClient.pipe(Layer.provide(SqliteClient.layer({ filename: ":memory:", disableWAL: true }))), + ).pipe(Effect.timeout("2 seconds")) + + expect(parked.pollUnsafe()).toBeUndefined() + yield* Deferred.succeed(gate, undefined) + yield* Fiber.join(parked) + }).pipe( + Effect.provideService(Global.Service, Global.make({ data: path.join(process.cwd(), ".test-data") })), + Effect.scoped, + ), + ) + }) + if (process.platform === "linux") { test("declared schema has no ungenerated migrations", async () => { const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check`