diff --git a/packages/core/src/database/database.ts b/packages/core/src/database/database.ts index d61adf047eac..88a79375b92d 100644 --- a/packages/core/src/database/database.ts +++ b/packages/core/src/database/database.ts @@ -2,10 +2,11 @@ export * as Database from "./database" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { layer as sqliteLayer } from "#sqlite" -import { Context, Effect, Layer } from "effect" +import { Cause, Context, Effect, Layer } from "effect" import { Global } from "../global" import { Flag } from "../flag/flag" import { isAbsolute, join } from "path" +import { rename, stat } from "fs/promises" import { DatabaseMigration } from "./migration" import { InstallationChannel } from "../installation/version" import { makeGlobalNode } from "../effect/app-node" @@ -19,9 +20,109 @@ export interface Interface { export class Service extends Context.Service()("@opencode/v2/storage/Database") {} -const layer = Layer.effect( - Service, +function isCorruptedDatabase(cause: Cause.Cause) { + const error = Cause.squash(cause) + const message = error instanceof Error ? error.message : String(error) + return message.includes("file is not a database") || message.includes("database disk image is malformed") +} + +const backupCorruptedFiles = (filename: string) => + Effect.gen(function* () { + const timestamp = Date.now() + const backedUp = yield* Effect.forEach(["", "-wal", "-shm"] as const, (ext) => + Effect.tryPromise({ + try: async () => { + const src = filename + ext + await rename(src, `${src}.corrupt-${timestamp}`) + return true + }, + catch: () => false, + }).pipe(Effect.orElseSucceed(() => false)), + ) + + if (backedUp.some(Boolean)) { + yield* Effect.logWarning(`Database corrupted. Backed up to: ${filename}.corrupt-${timestamp}`) + return `${filename}.corrupt-${timestamp}` + } + + yield* Effect.logWarning(`Database corrupted, but no files could be moved aside: ${filename}`) + return undefined + }) + +const salvageFromBackup = (backupPath: string, targetFilename: string) => Effect.gen(function* () { + const exists = yield* Effect.tryPromise({ + try: () => stat(backupPath), + catch: () => null, + }).pipe(Effect.orElseSucceed(() => null)) + if (!exists) return + + yield* Effect.try({ + try: () => { + const { Database: BunDatabase } = require("bun:sqlite") + const source = new BunDatabase(backupPath, { readwrite: true, create: false }) + const target = new BunDatabase(targetFilename, { readwrite: true, create: true }) + + try { + target.run("PRAGMA journal_mode = WAL") + target.run("PRAGMA foreign_keys = OFF") + + const tables = target + .query("SELECT name FROM sqlite_master WHERE type = 'table' AND name != 'migration'") + .all() as Array<{ name: string }> + + let salvaged = 0 + for (const { name } of tables) { + try { + const sourceHasTable = source + .query(`SELECT name FROM sqlite_master WHERE type = 'table' AND name = '${name}'`) + .all() + if (sourceHasTable.length === 0) continue + + const targetColumns = (target.query(`PRAGMA table_info('${name}')`).all() as Array<{ name: string }>).map((c) => c.name) + const sourceColumns = (source.query(`PRAGMA table_info('${name}')`).all() as Array<{ name: string }>).map((c) => c.name) + const shared = targetColumns.filter((col) => sourceColumns.includes(col)) + if (shared.length === 0) continue + + const columnNames = shared.join(", ") + const placeholders = shared.map(() => "?").join(", ") + + const rows = source.query(`SELECT ${columnNames} FROM ${name}`).all() as Array> + if (rows.length === 0) continue + + const insert = target.prepare(`INSERT OR IGNORE INTO ${name} (${columnNames}) VALUES (${placeholders})`) + const tx = target.transaction((batch: Array>) => { + for (const row of batch) { + insert.run(...(shared.map((col) => row[col]) as Array)) + } + }) + tx(rows) + salvaged += rows.length + } catch { + continue + } + } + + if (salvaged > 0) { + // eslint-disable-next-line no-console + console.warn(`[opencode] Salvaged ${salvaged} rows from corrupted database`) + } + } finally { + target.run("PRAGMA wal_checkpoint(TRUNCATE)") + source.close() + target.close() + } + }, + catch: (e) => { + // eslint-disable-next-line no-console + console.warn(`[opencode] Failed to salvage from corrupted database:`, e) + return undefined + }, + }) + }) + +function initializeDb() { + return Effect.gen(function* () { const db = yield* makeDatabase yield* db.run("PRAGMA journal_mode = WAL") @@ -30,14 +131,43 @@ const layer = Layer.effect( yield* db.run("PRAGMA cache_size = -64000") yield* db.run("PRAGMA foreign_keys = ON") yield* db.run("PRAGMA wal_checkpoint(PASSIVE)") + + const rows = yield* db.all<{ quick_check: string }>("PRAGMA quick_check") + if (rows.length !== 1 || rows[0]!.quick_check !== "ok") { + const details = rows.map((r) => r.quick_check).join("; ") + yield* Effect.die(new Error(`database disk image is malformed (quick_check: ${details})`)) + } + yield* DatabaseMigration.apply(db) - return { db } - }).pipe(Effect.orDie), -) + return Service.of({ db }) + }) +} + +function baseLayer(filename: string) { + return Layer.effect( + Service, + initializeDb().pipe(Effect.orDie), + ).pipe( + Layer.provide(sqliteLayer({ filename, disableWAL: true })), + ) +} export function layerFromPath(filename: string) { - return layer.pipe(Layer.provide(sqliteLayer({ filename }))) + return Layer.catchCause(baseLayer(filename), (cause) => + isCorruptedDatabase(cause) + ? Layer.unwrap( + Effect.gen(function* () { + const backupPath = yield* backupCorruptedFiles(filename) + const recovered = baseLayer(filename) + if (backupPath) { + return Layer.tap(recovered, () => salvageFromBackup(backupPath, filename)) + } + return recovered + }), + ) + : Layer.effectContext(Effect.failCause(cause)), + ) } export function path() { diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index b381cc7418a3..dc95137343ec 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -1,10 +1,11 @@ import { describe, expect, test } from "bun:test" import { $ } from "bun" +import fs from "fs/promises" import { fileURLToPath } from "url" import path from "path" import { SqliteClient } from "@effect/sql-sqlite-bun" import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" -import { Effect, Layer } from "effect" +import { Context, Effect, Exit, Layer, Scope } from "effect" import { eq, inArray, sql } from "drizzle-orm" import { DatabaseMigration } from "@opencode-ai/core/database/migration" import { migrations } from "@opencode-ai/core/database/migration.gen" @@ -38,6 +39,115 @@ const run = (effect: Effect.Effect) => const makeDb = EffectDrizzleSqlite.makeWithDefaults() describe("DatabaseMigration", () => { + test("backs up a corrupted database file and recreates it", async () => { + await using tmp = await tmpdir() + const filename = path.join(tmp.path, "recovery.sqlite") + await Bun.write(filename, new TextEncoder().encode(`SQLite format 3\0${"x".repeat(256)}`)) + + await Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.make() + const context = yield* Layer.buildWithScope(Database.layerFromPath(filename), scope) + const db = Context.get(context, Database.Service).db + + expect((yield* Effect.promise(() => fs.readdir(tmp.path))).some((file) => file.startsWith("recovery.sqlite.corrupt-"))).toBe(true) + expect(yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'migration'`)).toEqual({ + name: "migration", + }) + yield* Scope.close(scope, Exit.void) + }), + ) + }) + + test("backs up a partially corrupted database detected by quick_check", async () => { + await using tmp = await tmpdir() + const filename = path.join(tmp.path, "partial-corrupt.sqlite") + + const { Database: BunDatabase } = await import("bun:sqlite") + const native = new BunDatabase(filename, { create: true }) + native.run("PRAGMA page_size = 4096") + native.run("PRAGMA journal_mode = DELETE") + native.run("CREATE TABLE test_data (id INTEGER PRIMARY KEY, payload TEXT)") + for (let i = 0; i < 200; i++) { + native.run(`INSERT INTO test_data VALUES (${i}, '${"a".repeat(200)}')`) + } + native.close() + + const data = await Bun.file(filename).arrayBuffer() + const bytes = new Uint8Array(data) + const pageSize = 4096 + const totalPages = Math.floor(bytes.length / pageSize) + if (totalPages > 3) { + const targetPage = totalPages - 1 + const offset = targetPage * pageSize + for (let i = offset + 8; i < offset + 64; i++) { + bytes[i] = bytes[i]! ^ 0xaa + } + } + await Bun.write(filename, bytes) + + await Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.make() + const context = yield* Layer.buildWithScope(Database.layerFromPath(filename), scope) + const db = Context.get(context, Database.Service).db + + const files = yield* Effect.promise(() => fs.readdir(tmp.path)) + expect(files.some((file) => file.startsWith("partial-corrupt.sqlite.corrupt-"))).toBe(true) + expect(yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'migration'`)).toEqual({ + name: "migration", + }) + yield* Scope.close(scope, Exit.void) + }), + ) + }) + + test("salvages readable rows from a partially corrupted database", async () => { + await using tmp = await tmpdir() + const filename = path.join(tmp.path, "salvage.sqlite") + + const { Database: BunDatabase } = await import("bun:sqlite") + const native = new BunDatabase(filename, { create: true }) + native.run("PRAGMA page_size = 4096") + native.run("PRAGMA journal_mode = DELETE") + native.run("CREATE TABLE project (id TEXT PRIMARY KEY, worktree TEXT NOT NULL, sandboxes TEXT NOT NULL, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL)") + native.run("CREATE TABLE session (id TEXT PRIMARY KEY, project_id TEXT, slug TEXT, directory TEXT, title TEXT, version TEXT, time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL)") + native.run(`INSERT INTO project VALUES ('proj_1', '/code', '[]', 1, 1)`) + native.run(`INSERT INTO session VALUES ('ses_1', 'proj_1', 'test', '/code', 'My Session', 'v1', 1, 1)`) + native.run(`INSERT INTO session VALUES ('ses_2', 'proj_1', 'test2', '/code', 'Another', 'v1', 2, 2)`) + native.close() + + const data = await Bun.file(filename).arrayBuffer() + const bytes = new Uint8Array(data) + const pageSize = 4096 + const totalPages = Math.floor(bytes.length / pageSize) + if (totalPages > 3) { + const targetPage = totalPages - 1 + const offset = targetPage * pageSize + for (let i = offset + 8; i < offset + 64; i++) { + bytes[i] = bytes[i]! ^ 0xaa + } + } + await Bun.write(filename, bytes) + + await Effect.runPromise( + Effect.gen(function* () { + const scope = yield* Scope.make() + const context = yield* Layer.buildWithScope(Database.layerFromPath(filename), scope) + const db = Context.get(context, Database.Service).db + + const files = yield* Effect.promise(() => fs.readdir(tmp.path)) + expect(files.some((file) => file.startsWith("salvage.sqlite.corrupt-"))).toBe(true) + + const sessions = yield* db.all<{ id: string; title: string }>(sql`SELECT id, title FROM session ORDER BY id`) + const projects = yield* db.all<{ id: string }>(sql`SELECT id FROM project`) + expect(sessions.length + projects.length).toBeGreaterThan(0) + + yield* Scope.close(scope, Exit.void) + }), + ) + }) + test("serializes concurrent embedded initialization for one database path", async () => { await using tmp = await tmpdir() const filename = path.join(tmp.path, "embedded.sqlite")