Skip to content

Commit 59c3e32

Browse files
JacobNWolfJacobNWolf
committed
fix(core): split sqlite read/write lanes and bound WAL checkpoints
Run plain SELECT/EXPLAIN reads on a second sqlite connection in both drivers so they never queue behind the write semaphore; keep writes gated and route in-transaction statements through the transaction connection. Reader connections enforce readonly access and PRAGMA query_only, and stay aliased to the writer for :memory: and disableWAL databases. Bound WAL growth with wal_autocheckpoint 4000, journal_size_limit 32MB, and a 5-minute PASSIVE checkpoint loop; TRUNCATE stalled the writer for about five seconds per run while holding the write permit. Throttle shell per-chunk metadata writes to 200ms with a guaranteed dirty flush, annotate GlobalBus listeners, and drop the EventEmitter generic that broke under local bun 1.4 typings. Add file-backed split-driver coverage: cross-connection read-after-write, in-transaction reads, and rollback isolation.
1 parent cdd1f38 commit 59c3e32

7 files changed

Lines changed: 197 additions & 26 deletions

File tree

‎packages/core/src/database/database.ts‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ export * as Database from "./database"
22

33
import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
44
import { layer as sqliteLayer } from "#sqlite"
5-
import { Context, Effect, Layer } from "effect"
5+
import { Context, Effect, Layer, Schedule } from "effect"
66
import { Global } from "../global"
77
import { Flag } from "../flag/flag"
88
import { isAbsolute, join } from "path"
@@ -30,8 +30,14 @@ const layer = Layer.effect(
3030
yield* db.run("PRAGMA cache_size = -64000")
3131
yield* db.run("PRAGMA foreign_keys = ON")
3232
yield* db.run("PRAGMA wal_checkpoint(PASSIVE)")
33+
yield* db.run("PRAGMA wal_autocheckpoint = 4000")
34+
yield* db.run("PRAGMA journal_size_limit = 33554432")
3335
yield* DatabaseMigration.apply(db)
3436

37+
yield* db
38+
.run("PRAGMA wal_checkpoint(PASSIVE)")
39+
.pipe(Effect.ignore, Effect.repeat(Schedule.spaced("5 minutes")), Effect.forkScoped)
40+
3541
return { db }
3642
}).pipe(Effect.orDie),
3743
)

‎packages/core/src/database/sqlite.bun.ts‎

Lines changed: 49 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@ import { Sqlite } from "./sqlite"
1717

1818
const ATTR_DB_SYSTEM_NAME = "db.system.name"
1919

20+
// Plain reads run on a second connection so they never queue behind the write semaphore.
21+
const READS = /^\s*(?:select|explain)\b/i
22+
2023
const TypeId = "~@opencode-ai/core/database/SqliteBun" as const
2124
type TypeId = typeof TypeId
2225

@@ -53,9 +56,21 @@ const make = (options: Config) =>
5356
? Statement.defaultTransforms(options.transformResultNames).array
5457
: undefined
5558

56-
const run = (query: string, params: ReadonlyArray<unknown> = []) =>
59+
const inMemory = options.filename.includes(":memory:") || options.filename.includes("mode=memory")
60+
const readNative =
61+
inMemory || options.disableWAL
62+
? native
63+
: new Database(options.filename, { readonly: options.readonly, readwrite: !options.readonly, create: false })
64+
if (readNative !== native) {
65+
yield* Effect.addFinalizer(() => Effect.sync(() => readNative.close()))
66+
readNative.run("PRAGMA busy_timeout = 5000;")
67+
readNative.run("PRAGMA cache_size = -64000;")
68+
readNative.run("PRAGMA query_only = ON;")
69+
}
70+
71+
const run = (target: Database, query: string, params: ReadonlyArray<unknown> = []) =>
5772
Effect.withFiber<Array<Record<string, unknown>>, SqlError>((fiber) => {
58-
const statement = native.query(query)
73+
const statement = target.query(query)
5974
// @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627
6075
statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers))
6176
try {
@@ -69,9 +84,9 @@ const make = (options: Config) =>
6984
}
7085
})
7186

72-
const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
87+
const runValues = (target: Database, query: string, params: ReadonlyArray<unknown> = []) =>
7388
Effect.withFiber<Array<unknown[]>, SqlError>((fiber) => {
74-
const statement = native.query(query)
89+
const statement = target.query(query)
7590
// @ts-ignore bun-types missing safeIntegers method, fixed in https://github.com/oven-sh/bun/pull/26627
7691
statement.safeIntegers(Context.get(fiber.context, Client.SafeIntegers))
7792
try {
@@ -85,15 +100,18 @@ const make = (options: Config) =>
85100
}
86101
})
87102

88-
const connection = identity<SqliteConnection>({
103+
const semaphore = yield* Semaphore.make(1)
104+
const gated = <A, E>(effect: Effect.Effect<A, E>) => semaphore.withPermits(1)(effect)
105+
106+
const writer = identity<SqliteConnection>({
89107
execute(query, params, transformRows) {
90-
return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
108+
return transformRows ? Effect.map(run(native, query, params), transformRows) : run(native, query, params)
91109
},
92110
executeRaw(query, params) {
93-
return run(query, params)
111+
return run(native, query, params)
94112
},
95113
executeValues(query, params) {
96-
return runValues(query, params)
114+
return runValues(native, query, params)
97115
},
98116
executeUnprepared(query, params, transformRows) {
99117
return this.execute(query, params, transformRows)
@@ -118,14 +136,34 @@ const make = (options: Config) =>
118136
}),
119137
})
120138

121-
const semaphore = yield* Semaphore.make(1)
122-
const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
139+
const connection = identity<SqliteConnection>({
140+
execute(query, params, transformRows) {
141+
const effect = READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
142+
return transformRows ? Effect.map(effect, transformRows) : effect
143+
},
144+
executeRaw(query, params) {
145+
return READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
146+
},
147+
executeValues(query, params) {
148+
return READS.test(query) ? runValues(readNative, query, params) : gated(runValues(native, query, params))
149+
},
150+
executeUnprepared(query, params, transformRows) {
151+
return this.execute(query, params, transformRows)
152+
},
153+
executeStream() {
154+
return Stream.die("executeStream not implemented")
155+
},
156+
export: writer.export,
157+
loadExtension: writer.loadExtension,
158+
})
159+
160+
const acquirer = Effect.succeed(connection)
123161
const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
124162
const fiber = Fiber.getCurrent()!
125163
const scope = Context.getUnsafe(fiber.context, Scope.Scope)
126164
return Effect.as(
127165
Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
128-
connection,
166+
writer,
129167
)
130168
})
131169

‎packages/core/src/database/sqlite.node.ts‎

Lines changed: 51 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,9 @@ import { Sqlite } from "./sqlite"
1717

1818
const ATTR_DB_SYSTEM_NAME = "db.system.name"
1919

20+
// Plain reads run on a second connection so they never queue behind the write semaphore.
21+
const READS = /^\s*(?:select|explain)\b/i
22+
2023
const TypeId = "~@opencode-ai/core/database/SqliteNode" as const
2124
type TypeId = typeof TypeId
2225

@@ -53,9 +56,25 @@ const make = (options: Config) =>
5356
? Statement.defaultTransforms(options.transformResultNames).array
5457
: undefined
5558

56-
const run = (query: string, params: ReadonlyArray<unknown> = []) =>
59+
const inMemory = options.filename.includes(":memory:") || options.filename.includes("mode=memory")
60+
const readNative =
61+
inMemory || options.disableWAL
62+
? native
63+
: new DatabaseSync(options.filename, {
64+
readOnly: options.readonly,
65+
timeout: options.timeout,
66+
open: true,
67+
})
68+
if (readNative !== native) {
69+
yield* Effect.addFinalizer(() => Effect.sync(() => readNative.close()))
70+
readNative.exec("PRAGMA busy_timeout = 5000;")
71+
readNative.exec("PRAGMA cache_size = -64000;")
72+
readNative.exec("PRAGMA query_only = ON;")
73+
}
74+
75+
const run = (target: DatabaseSync, query: string, params: ReadonlyArray<unknown> = []) =>
5776
Effect.withFiber<Array<Record<string, unknown>>, SqlError>((fiber) => {
58-
const statement = native.prepare(query)
77+
const statement = target.prepare(query)
5978
statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
6079
try {
6180
return Effect.succeed(statement.all(...(params as SQLInputValue[])) as Array<Record<string, unknown>>)
@@ -68,9 +87,9 @@ const make = (options: Config) =>
6887
}
6988
})
7089

71-
const runValues = (query: string, params: ReadonlyArray<unknown> = []) =>
90+
const runValues = (target: DatabaseSync, query: string, params: ReadonlyArray<unknown> = []) =>
7291
Effect.withFiber<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>((fiber) => {
73-
const statement = native.prepare(query)
92+
const statement = target.prepare(query)
7493
statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
7594
statement.setReturnArrays(true)
7695
try {
@@ -86,15 +105,15 @@ const make = (options: Config) =>
86105
}
87106
})
88107

89-
const connection = identity<SqliteConnection>({
108+
const writer = identity<SqliteConnection>({
90109
execute(query, params, transformRows) {
91-
return transformRows ? Effect.map(run(query, params), transformRows) : run(query, params)
110+
return transformRows ? Effect.map(run(native, query, params), transformRows) : run(native, query, params)
92111
},
93112
executeRaw(query, params) {
94-
return run(query, params)
113+
return run(native, query, params)
95114
},
96115
executeValues(query, params) {
97-
return runValues(query, params)
116+
return runValues(native, query, params)
98117
},
99118
executeUnprepared(query, params, transformRows) {
100119
return this.execute(query, params, transformRows)
@@ -113,13 +132,35 @@ const make = (options: Config) =>
113132
})
114133

115134
const semaphore = yield* Semaphore.make(1)
116-
const acquirer = semaphore.withPermits(1)(Effect.succeed(connection))
135+
const gated = <A, E>(effect: Effect.Effect<A, E>) => semaphore.withPermits(1)(effect)
136+
137+
const connection = identity<SqliteConnection>({
138+
execute(query, params, transformRows) {
139+
const result = READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
140+
return transformRows ? Effect.map(result, transformRows) : result
141+
},
142+
executeRaw(query, params) {
143+
return READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
144+
},
145+
executeValues(query, params) {
146+
return READS.test(query) ? runValues(readNative, query, params) : gated(runValues(native, query, params))
147+
},
148+
executeUnprepared(query, params, transformRows) {
149+
return this.execute(query, params, transformRows)
150+
},
151+
executeStream() {
152+
return Stream.die("executeStream not implemented")
153+
},
154+
loadExtension: (path) => writer.loadExtension(path),
155+
})
156+
157+
const acquirer = Effect.succeed(connection)
117158
const transactionAcquirer = Effect.uninterruptibleMask((restore) => {
118159
const fiber = Fiber.getCurrent()!
119160
const scope = Context.getUnsafe(fiber.context, Scope.Scope)
120161
return Effect.as(
121162
Effect.tap(restore(semaphore.take(1)), () => Scope.addFinalizer(scope, semaphore.release(1))),
122-
connection,
163+
writer,
123164
)
124165
})
125166

‎packages/core/test/sqlite-split.test.ts‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
import { describe, expect, test } from "bun:test"
2+
import path from "path"
3+
import { sql } from "drizzle-orm"
4+
import { Effect } from "effect"
5+
import { Database } from "@opencode-ai/core/database/database"
6+
import { tmpdir } from "./fixture/tmpdir"
7+
8+
// The split driver aliases its reader connection to the writer for :memory: databases,
9+
// so these tests must use a real file to exercise the second connection.
10+
const run = <A, E>(filename: string, effect: Effect.Effect<A, E, Database.Service>) =>
11+
Effect.runPromise(effect.pipe(Effect.provide(Database.layerFromPath(filename))))
12+
13+
describe("SqliteReadWriteSplit", () => {
14+
test("plain reads see committed writes through the reader connection", async () => {
15+
await using tmp = await tmpdir()
16+
const filename = path.join(tmp.path, "split.sqlite")
17+
await run(
18+
filename,
19+
Effect.gen(function* () {
20+
const { db } = yield* Database.Service
21+
yield* db.run(sql`CREATE TABLE probe (id INTEGER PRIMARY KEY, value TEXT NOT NULL)`)
22+
yield* db.run(sql`INSERT INTO probe (id, value) VALUES (1, 'committed')`)
23+
expect(yield* db.get(sql`SELECT value FROM probe WHERE id = 1`)).toEqual({ value: "committed" })
24+
}),
25+
)
26+
})
27+
28+
test("reads inside a transaction see uncommitted rows from the same connection", async () => {
29+
await using tmp = await tmpdir()
30+
const filename = path.join(tmp.path, "split.sqlite")
31+
await run(
32+
filename,
33+
Effect.gen(function* () {
34+
const { db } = yield* Database.Service
35+
yield* db.run(sql`CREATE TABLE probe (id INTEGER PRIMARY KEY, value TEXT NOT NULL)`)
36+
yield* db.transaction((tx) =>
37+
Effect.gen(function* () {
38+
yield* tx.run(sql`INSERT INTO probe (id, value) VALUES (1, 'uncommitted')`)
39+
expect(yield* tx.get(sql`SELECT value FROM probe WHERE id = 1`)).toEqual({ value: "uncommitted" })
40+
}),
41+
)
42+
expect(yield* db.get(sql`SELECT value FROM probe WHERE id = 1`)).toEqual({ value: "uncommitted" })
43+
}),
44+
)
45+
})
46+
47+
test("a rollback hides uncommitted rows from the reader connection", async () => {
48+
await using tmp = await tmpdir()
49+
const filename = path.join(tmp.path, "split.sqlite")
50+
await run(
51+
filename,
52+
Effect.gen(function* () {
53+
const { db } = yield* Database.Service
54+
yield* db.run(sql`CREATE TABLE probe (id INTEGER PRIMARY KEY, value TEXT NOT NULL)`)
55+
yield* db
56+
.transaction((tx) =>
57+
Effect.gen(function* () {
58+
yield* tx.run(sql`INSERT INTO probe (id, value) VALUES (1, 'rollback')`)
59+
return yield* Effect.die(new Error("rollback"))
60+
}),
61+
)
62+
.pipe(Effect.catchCause(() => Effect.void))
63+
expect(yield* db.get(sql`SELECT COUNT(*) AS count FROM probe`)).toEqual({ count: 0 })
64+
}),
65+
)
66+
})
67+
})

‎packages/opencode/src/bus/global.ts‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,7 @@ export type GlobalEvent = {
88
payload: any
99
}
1010

11-
class GlobalBusEmitter extends EventEmitter<{
12-
event: [GlobalEvent]
13-
}> {
11+
class GlobalBusEmitter extends EventEmitter {
1412
override emit(eventName: "event", event: GlobalEvent): boolean {
1513
if (event.payload && typeof event.payload === "object" && !("id" in event.payload)) {
1614
event.payload.id = event.payload.syncEvent?.id ?? Identifier.create("evt", "ascending")

‎packages/opencode/src/cli/tui/worker.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ function flushEvents() {
3333
Rpc.emit("global.event.batch", collapseEventBatch(pendingEvents.splice(0)))
3434
}
3535

36-
GlobalBus.on("event", (event) => {
36+
GlobalBus.on("event", (event: GlobalEvent) => {
3737
pendingEvents.push(event)
3838
if (pendingEvents.length >= EVENT_BATCH_LIMIT) {
3939
if (flushTimer !== undefined) clearTimeout(flushTimer)

‎packages/opencode/src/tool/shell.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,9 @@ import { BashArity } from "@/permission/arity"
2525
export { Parameters } from "./shell/prompt"
2626

2727
const MAX_METADATA_LENGTH = 30_000
28+
29+
// Bound per-chunk preview writes: every metadata call durably rewrites the full part.
30+
const METADATA_THROTTLE_MS = 200
2831
const CWD = new Set(["cd", "chdir", "popd", "pushd", "push-location", "set-location"])
2932
const FILES = new Set([
3033
...CWD,
@@ -446,6 +449,8 @@ export const ShellTool = Tool.define(
446449
let cut = false
447450
let expired = false
448451
let aborted = false
452+
let lastMetadataFlush = 0
453+
let metadataDirty = false
449454

450455
const closeSink = Effect.fnUntraced(function* () {
451456
const stream = sink
@@ -522,6 +527,13 @@ export const ShellTool = Tool.define(
522527
}
523528
}
524529

530+
const now = Date.now()
531+
if (now - lastMetadataFlush < METADATA_THROTTLE_MS) {
532+
metadataDirty = true
533+
return Effect.void
534+
}
535+
lastMetadataFlush = now
536+
metadataDirty = false
525537
return ctx.metadata({
526538
metadata: {
527539
output: last,
@@ -558,6 +570,15 @@ export const ShellTool = Tool.define(
558570
}),
559571
).pipe(Effect.orDie)
560572

573+
if (metadataDirty) {
574+
metadataDirty = false
575+
yield* ctx.metadata({
576+
metadata: {
577+
output: last,
578+
},
579+
})
580+
}
581+
561582
const meta: string[] = []
562583
if (expired) {
563584
meta.push(

0 commit comments

Comments
 (0)