chore: merge rename-event-to-bus into main

Author: edy

MERGE.md

@@ -2,7 +2,7 @@ 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 { Context, Effect, Layer, Schedule } from "effect"
 import { Global } from "../global"
 import { Flag } from "../flag/flag"
 import { isAbsolute, join } from "path"
@@ -24,9 +24,22 @@ const layer = Layer.effect(
   Effect.gen(function* () {
     const db = yield* makeDatabase
 
-    yield* db.run("PRAGMA journal_mode = WAL")
-    yield* db.run("PRAGMA synchronous = NORMAL")
+    // busy_timeout must precede journal_mode: the WAL switch takes a write lock
+    // and would fail with SQLITE_BUSY under concurrent startup at zero timeout.
     yield* db.run("PRAGMA busy_timeout = 5000")
+    // The WAL mode switch needs an exclusive lock; busy_timeout does not always
+    // cover it, so retry briefly when two processes cold-start the same file.
+    yield* db
+      .run("PRAGMA journal_mode = WAL")
+      .pipe(
+        Effect.retry(
+          Schedule.exponential(25).pipe(
+            Schedule.jittered,
+            Schedule.while((meta) => meta.elapsed < 5_000),
+          ),
+        ),
+      )
+    yield* db.run("PRAGMA synchronous = NORMAL")
     yield* db.run("PRAGMA cache_size = -64000")
     yield* db.run("PRAGMA foreign_keys = ON")
     yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")

packages/core/src/bus.ts

@@ -23,19 +23,26 @@ export function apply(db: Database) {
       )
       if (tables.some((table) => table.name === "session")) return yield* applyOnly(db, migrations)
       if (tables.length > 0) return yield* Effect.die("Database is not empty and has no session table")
-      yield* db.transaction((tx) =>
+      // Bootstrap DDL cannot replay, so the emptiness decision must be re-verified
+      // inside this immediate write transaction: another process may have
+      // bootstrapped the same file after the read above.
+      const bootstrapped = 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)`,
+            sql`CREATE TABLE IF NOT EXISTS ${sql.identifier("migration")} (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`,
           )
+          if (yield* tx.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = ${"session"}`))
+            return false
+          yield* schema.up(tx)
           yield* Effect.forEach(migrations, (migration) =>
             tx.run(
               sql`INSERT OR IGNORE INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`,
             ),
           )
+          return true
         }),
       )
+      if (!bootstrapped) return yield* applyOnly(db, migrations)
     }),
   )
 }
@@ -68,8 +75,12 @@ export function applyOnly(db: Database, input: Migration[]) {
 
     for (const migration of input) {
       if (completed.has(migration.id)) continue
+      // Migration DDL cannot replay, so completion must be re-verified inside
+      // this immediate write transaction: another process may have claimed the
+      // migration after the journal read above.
       yield* db.transaction((tx) =>
         Effect.gen(function* () {
+          if (yield* tx.get(sql`SELECT id FROM ${sql.identifier("migration")} WHERE id = ${migration.id}`)) return
           yield* migration.up(tx)
           yield* tx.run(
             sql`INSERT OR IGNORE INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`,