| author | |
| committer | |
| log | 4cd0076ec65d51974dbe3efdd03be6ed5bf91c80 |
| tree | 6f4c60c167041bd17e8ca3231c732391bb3bbe0a |
| parent | 173a97097659be5f77ab2c05cf5dfe84be3eb094 |
| signature | Signed by PGP key B5690EEEBB952194 |
Takeover of #17785 by @tris203. Executions binding more than 64 KiB of text or blob data run on a throwaway statement, in the writer and the reader worker, so cached statements no longer pin large payloads in native memory.
Co-authored-by: tris203 <admin@snappeh.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>2 files changed, 188 insertions(+), 3 deletions(-)
packages/shared/src/nodeSqliteClient.test.ts+128| ... | @@ -131,6 +131,102 @@ layer("NodeSqliteClient", (it) => { | ... | @@ -131,6 +131,102 @@ layer("NodeSqliteClient", (it) => { |
| 131 | ); | 131 | ); |
| 132 | }); | 132 | }); |
| 133 | 133 | ||
| 134 | const largeText = "x".repeat(64 * 1024 + 1); | ||
| 135 | |||
| 136 | it.effect("keeps large text and blob parameters off cached statements", () => | ||
| 137 | Effect.gen(function* () { | ||
| 138 | const sql = yield* SqlClient.SqlClient; | ||
| 139 | const statements: Array<NodeSqlite.StatementSync> = []; | ||
| 140 | const all = NodeSqlite.StatementSync.prototype.all; | ||
| 141 | yield* Effect.acquireRelease( | ||
| 142 | Effect.sync(() => | ||
| 143 | vi.spyOn(NodeSqlite.StatementSync.prototype, "all").mockImplementation(function ( | ||
| 144 | this: NodeSqlite.StatementSync, | ||
| 145 | ...params | ||
| 146 | ) { | ||
| 147 | statements.push(this); | ||
| 148 | return all.apply(this, params); | ||
| 149 | }), | ||
| 150 | ), | ||
| 151 | (spy) => Effect.sync(() => spy.mockRestore()), | ||
| 152 | ); | ||
| 153 | |||
| 154 | const small = "small"; | ||
| 155 | const queries = [ | ||
| 156 | (value: string | Uint8Array | ArrayBuffer) => sql`SELECT length(${value}) AS size`, | ||
| 157 | (value: string | Uint8Array | ArrayBuffer) => | ||
| 158 | sql.unsafe("SELECT length($value) AS size", [{ $value: value }]), | ||
| 159 | (value: string | Uint8Array | ArrayBuffer) => | ||
| 160 | sql.unsafe("SELECT length($value) AS callable", [ | ||
| 161 | Object.assign(() => {}, { $value: value }), | ||
| 162 | ]), | ||
| 163 | ]; | ||
| 164 | const cached = new Set<NodeSqlite.StatementSync>(); | ||
| 165 | const large = new Set<NodeSqlite.StatementSync>(); | ||
| 166 | const largeBlobs = [new Uint8Array(64 * 1024 + 1), new ArrayBuffer(64 * 1024 + 1)]; | ||
| 167 | for (const value of [small, largeText, ...largeBlobs, small]) { | ||
| 168 | const expected = typeof value === "string" ? value.length : value.byteLength; | ||
| 169 | for (const query of queries) { | ||
| 170 | assert.deepEqual((yield* query(value)).map(Object.values), [[expected]]); | ||
| 171 | assert.deepEqual(yield* query(value).values, [[expected]]); | ||
| 172 | for (const statement of statements.splice(0)) { | ||
| 173 | (value === small ? cached : large).add(statement); | ||
| 174 | } | ||
| 175 | } | ||
| 176 | } | ||
| 177 | |||
| 178 | // Small parameters keep reusing the one cached statement for each query. | ||
| 179 | assert.equal(cached.size, queries.length); | ||
| 180 | // Each large execution gets a statement that nothing holds on to afterwards. | ||
| 181 | assert.equal(large.size, queries.length * 6); | ||
| 182 | assert.isTrue(cached.isDisjointFrom(large)); | ||
| 183 | }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))), | ||
| 184 | ); | ||
| 185 | |||
| 186 | it.effect("reads integers on the values path as the caller asked, cached or not", () => | ||
| 187 | Effect.gen(function* () { | ||
| 188 | const sql = yield* SqlClient.SqlClient; | ||
| 189 | for (const value of ["small", largeText]) { | ||
| 190 | const values = sql`SELECT 1 AS id, length(${value}) AS size`.values; | ||
| 191 | assert.deepEqual(yield* values.pipe(Effect.provideService(SqlClient.SafeIntegers, true)), [ | ||
| 192 | [1n, BigInt(value.length)], | ||
| 193 | ]); | ||
| 194 | assert.deepEqual(yield* values, [[1, value.length]]); | ||
| 195 | } | ||
| 196 | }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))), | ||
| 197 | ); | ||
| 198 | |||
| 199 | it.effect("leaves a parameter it cannot measure for the statement to handle", () => | ||
| 200 | Effect.gen(function* () { | ||
| 201 | const sql = yield* SqlClient.SqlClient; | ||
| 202 | const detached = new DataView(new ArrayBuffer(8)); | ||
| 203 | structuredClone(detached.buffer, { transfer: [detached.buffer] }); | ||
| 204 | // What a detached view binds as varies by Node version; match the statement itself. | ||
| 205 | const native = new NodeSqlite.DatabaseSync(":memory:"); | ||
| 206 | const expected = native.prepare("SELECT length(?) AS size").all(detached); | ||
| 207 | native.close(); | ||
| 208 | assert.deepEqual(yield* sql`SELECT length(${detached}) AS size`, expected); | ||
| 209 | |||
| 210 | const throwing = { | ||
| 211 | get $value(): string { | ||
| 212 | throw new Error("unreadable"); | ||
| 213 | }, | ||
| 214 | }; | ||
| 215 | const error = yield* sql.unsafe("SELECT length($value) AS size", [throwing]).pipe(Effect.flip); | ||
| 216 | assert.equal(error.reason.operation, "execute"); | ||
| 217 | |||
| 218 | // Only the statement reads a getter, so it runs once as it would natively. | ||
| 219 | let reads = 0; | ||
| 220 | const counting = { | ||
| 221 | get $value(): number { | ||
| 222 | return ++reads; | ||
| 223 | }, | ||
| 224 | }; | ||
| 225 | assert.deepEqual(yield* sql.unsafe("SELECT $value AS value", [counting]), [{ value: 1 }]); | ||
| 226 | assert.equal(reads, 1); | ||
| 227 | }).pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:" }))), | ||
| 228 | ); | ||
| 229 | |||
| 134 | const makeTempDatabase = Effect.gen(function* () { | 230 | const makeTempDatabase = Effect.gen(function* () { |
| 135 | const fs = yield* FileSystem.FileSystem; | 231 | const fs = yield* FileSystem.FileSystem; |
| 136 | const path = yield* Path.Path; | 232 | const path = yield* Path.Path; |
| ... | @@ -311,6 +407,38 @@ it.effect("keeps read-only work inside a write transaction on the writer", () => | ... | @@ -311,6 +407,38 @@ it.effect("keeps read-only work inside a write transaction on the writer", () => |
| 311 | ), | 407 | ), |
| 312 | ); | 408 | ); |
| 313 | 409 | ||
| 410 | it.effect("keeps large parameters off the reader worker's cached statements", () => | ||
| 411 | withReaderClient((sql) => | ||
| 412 | Effect.gen(function* () { | ||
| 413 | const read = (value: string) => | ||
| 414 | sql<{ size: number }>`SELECT length(${value}) AS size`.pipe( | ||
| 415 | Effect.map((rows) => rows[0]?.size), | ||
| 416 | ); | ||
| 417 | // Start the worker so its requests can be recorded. | ||
| 418 | yield* countEntries(sql).pipe(SqliteClient.readOnly); | ||
| 419 | const worker = startedWorkers.at(-1)!; | ||
| 420 | const cached: Array<boolean> = []; | ||
| 421 | const postMessage = worker.postMessage; | ||
| 422 | yield* Effect.acquireRelease( | ||
| 423 | Effect.sync(() => | ||
| 424 | vi.spyOn(worker, "postMessage").mockImplementation(function ( | ||
| 425 | this: NodeWorkerThreads.Worker, | ||
| 426 | message: { readonly cached: boolean }, | ||
| 427 | ) { | ||
| 428 | cached.push(message.cached); | ||
| 429 | return postMessage.call(this, message); | ||
| 430 | }), | ||
| 431 | ), | ||
| 432 | (spy) => Effect.sync(() => spy.mockRestore()), | ||
| 433 | ); | ||
| 434 | |||
| 435 | assert.equal(yield* read("small").pipe(SqliteClient.readOnly), 5); | ||
| 436 | assert.equal(yield* read(largeText).pipe(SqliteClient.readOnly), largeText.length); | ||
| 437 | assert.deepEqual(cached, [true, false]); | ||
| 438 | }).pipe(Effect.scoped), | ||
| 439 | ), | ||
| 440 | ); | ||
| 441 | |||
| 314 | it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () => | 442 | it.effect("fails a read-only transaction whose reader stops, then starts a new reader", () => |
| 315 | withReaderClient((sql) => | 443 | withReaderClient((sql) => |
| 316 | Effect.gen(function* () { | 444 | Effect.gen(function* () { |
packages/shared/src/nodeSqliteClient.ts+60-3| ... | @@ -5,6 +5,7 @@ | ... | @@ -5,6 +5,7 @@ |
| 5 | * @module SqliteClient | 5 | * @module SqliteClient |
| 6 | */ | 6 | */ |
| 7 | import * as NodeSqlite from "node:sqlite"; | 7 | import * as NodeSqlite from "node:sqlite"; |
| 8 | import * as NodeUtil from "node:util"; | ||
| 8 | import * as NodeWorkerThreads from "node:worker_threads"; | 9 | import * as NodeWorkerThreads from "node:worker_threads"; |
| 9 | 10 | ||
| 10 | import * as Cache from "effect/Cache"; | 11 | import * as Cache from "effect/Cache"; |
| ... | @@ -126,6 +127,54 @@ const classifyError = (cause: unknown, message: string, operation: string) => { | ... | @@ -126,6 +127,54 @@ const classifyError = (cause: unknown, message: string, operation: string) => { |
| 126 | return classifySqliteError(cause, { message, operation }); | 127 | return classifySqliteError(cause, { message, operation }); |
| 127 | }; | 128 | }; |
| 128 | 129 | ||
| 130 | /** | ||
| 131 | * SQLite keeps its copy of a statement's text and blob parameters until the | ||
| 132 | * statement is bound again or finalized, and `node:sqlite` cannot clear them. | ||
| 133 | * A cached statement would hold its last payload for as long as it stays | ||
| 134 | * cached, so executions binding more than this many string code units and | ||
| 135 | * blob bytes, or a named parameter behind a getter, get a throwaway statement | ||
| 136 | * instead. | ||
| 137 | */ | ||
| 138 | const LARGE_BINDING_SIZE = 64 * 1024; | ||
| 139 | |||
| 140 | // `node:sqlite` binds both buffer views and raw buffers as blobs. | ||
| 141 | const isBlob = (value: unknown): value is ArrayBufferView | ArrayBufferLike => | ||
| 142 | ArrayBuffer.isView(value) || NodeUtil.types.isAnyArrayBuffer(value); | ||
| 143 | |||
| 144 | const bindingSize = (value: unknown): number => { | ||
| 145 | if (typeof value === "string") { | ||
| 146 | return value.length; | ||
| 147 | } | ||
| 148 | return isBlob(value) ? value.byteLength : 0; | ||
| 149 | }; | ||
| 150 | |||
| 151 | const hasLargeBinding = (params: ReadonlyArray<unknown>): boolean => { | ||
| 152 | try { | ||
| 153 | let size = 0; | ||
| 154 | for (const param of params) { | ||
| 155 | if ( | ||
| 156 | (typeof param === "object" || typeof param === "function") && | ||
| 157 | param !== null && | ||
| 158 | !isBlob(param) | ||
| 159 | ) { | ||
| 160 | // `node:sqlite` takes named parameters as one object, callable or not. | ||
| 161 | // Read data properties only: calling a getter here would run it twice. | ||
| 162 | for (const property of Object.values(Object.getOwnPropertyDescriptors(param))) { | ||
| 163 | if (!property.enumerable) continue; | ||
| 164 | if (!("value" in property)) return true; | ||
| 165 | size += bindingSize(property.value); | ||
| 166 | } | ||
| 167 | } else { | ||
| 168 | size += bindingSize(param); | ||
| 169 | } | ||
| 170 | } | ||
| 171 | return size > LARGE_BINDING_SIZE; | ||
| 172 | } catch { | ||
| 173 | // A parameter that cannot be measured is left for the statement to report. | ||
| 174 | return true; | ||
| 175 | } | ||
| 176 | }; | ||
| 177 | |||
| 129 | // Evaluated source instead of a worker file, so the reader also starts from | 178 | // Evaluated source instead of a worker file, so the reader also starts from |
| 130 | // bundles and single-executable builds that have no module file on disk. | 179 | // bundles and single-executable builds that have no module file on disk. |
| 131 | const readerWorkerSource = ` | 180 | const readerWorkerSource = ` |
| ... | @@ -327,8 +376,10 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { | ... | @@ -327,8 +376,10 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) { |
| 327 | }, | 376 | }, |
| 328 | }); | 377 | }); |
| 329 | if (pending.size === 1) worker.ref(); | 378 | if (pending.size === 1) worker.ref(); |
| 379 | // The worker's statement cache would pin large parameters the same way. | ||
| 380 | const cached = input.cached && !hasLargeBinding(input.params); | ||
| 330 | // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. | 381 | // oxlint-disable-next-line unicorn/require-post-message-target-origin -- Node workers do not accept a target origin. |
| 331 | worker.postMessage({ id, safeIntegers, ...input }); | 382 | worker.postMessage({ id, safeIntegers, ...input, cached }); |
| 332 | }); | 383 | }); |
| 333 | }); | 384 | }); |
| 334 | 385 | ||
| ... | @@ -487,6 +538,9 @@ const make = Effect.fn("makeWithDatabase")(function* ( | ... | @@ -487,6 +538,9 @@ const make = Effect.fn("makeWithDatabase")(function* ( |
| 487 | Exit.isSuccess(exit) ? (options.prepareCacheTTL ?? Duration.minutes(10)) : Duration.zero, | 538 | Exit.isSuccess(exit) ? (options.prepareCacheTTL ?? Duration.minutes(10)) : Duration.zero, |
| 488 | }); | 539 | }); |
| 489 | 540 | ||
| 541 | const statementFor = (sql: string, params: ReadonlyArray<unknown>) => | ||
| 542 | hasLargeBinding(params) ? prepare(sql) : Cache.get(prepareCache, sql); | ||
| 543 | |||
| 490 | const runStatement = ( | 544 | const runStatement = ( |
| 491 | statement: NodeSqlite.StatementSync, | 545 | statement: NodeSqlite.StatementSync, |
| 492 | params: ReadonlyArray<unknown>, | 546 | params: ReadonlyArray<unknown>, |
| ... | @@ -510,7 +564,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( | ... | @@ -510,7 +564,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( |
| 510 | }); | 564 | }); |
| 511 | 565 | ||
| 512 | const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) => | 566 | const run = (sql: string, params: ReadonlyArray<unknown>, raw = false) => |
| 513 | Effect.flatMap(Cache.get(prepareCache, sql), (s) => runStatement(s, params, raw)); | 567 | Effect.flatMap(statementFor(sql, params), (s) => runStatement(s, params, raw)); |
| 514 | 568 | ||
| 515 | const runStatementValues = ( | 569 | const runStatementValues = ( |
| 516 | statement: NodeSqlite.StatementSync, | 570 | statement: NodeSqlite.StatementSync, |
| ... | @@ -521,6 +575,9 @@ const make = Effect.fn("makeWithDatabase")(function* ( | ... | @@ -521,6 +575,9 @@ const make = Effect.fn("makeWithDatabase")(function* ( |
| 521 | (statement) => | 575 | (statement) => |
| 522 | Effect.try({ | 576 | Effect.try({ |
| 523 | try: () => { | 577 | try: () => { |
| 578 | statement.setReadBigInts( | ||
| 579 | Boolean(Context.get(Fiber.getCurrent()!.context, Client.SafeIntegers)), | ||
| 580 | ); | ||
| 524 | if (hasRows(statement)) { | 581 | if (hasRows(statement)) { |
| 525 | statement.setReturnArrays(true); | 582 | statement.setReturnArrays(true); |
| 526 | // Safe to cast to array after we've setReturnArrays(true) | 583 | // Safe to cast to array after we've setReturnArrays(true) |
| ... | @@ -555,7 +612,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( | ... | @@ -555,7 +612,7 @@ const make = Effect.fn("makeWithDatabase")(function* ( |
| 555 | ); | 612 | ); |
| 556 | 613 | ||
| 557 | const runValues = (sql: string, params: ReadonlyArray<unknown>) => | 614 | const runValues = (sql: string, params: ReadonlyArray<unknown>) => |
| 558 | Effect.flatMap(Cache.get(prepareCache, sql), (statement) => | 615 | Effect.flatMap(statementFor(sql, params), (statement) => |
| 559 | runStatementValues(statement, params), | 616 | runStatementValues(statement, params), |
| 560 | ); | 617 | ); |
| 561 | 618 |