API index and search · Build metadata
Source snapshot
packages/sql-client/src/client.ts
1import { RindleSqlError, isRindleSqlError, protocolError } from "./errors.ts";2import { newIdempotencyKey, newRequestId } from "./id.ts";3import type {4 BatchOptions,5 BatchResult,6 BeginMutationInput,7 BeginMutationResult,8 ClientOptions,9 ExecuteMutationInput,10 ExecuteOptions,11 ExecuteResult,12 Fetch,13 IntMode,14 MigrationInput,15 MigrationResult,16 MutationIdentity,17 MutationReceipt,18 MutationRows,19 OperationOptions,20 ReadConsistency,21 RejectMutationInput,22 RetryOptions,23 RetryScope,24 RoutingMetadata,25 SqlClient,26 SqlMutationTransaction,27 SqlSession,28 SqlTransaction,29 Statement,30 StatementResult,31 TransactionOptions,32 TransactionState,33 WireRoutingMetadata,34 WireStatementResult,35} from "./types.ts";36import { decodeSqlValue, encodeStatement } from "./value.ts";3738type JsonObject = Record<string, unknown>;3940interface CursorState {41 cursor: string | null;42}4344interface RequestOptions {45 method?: "GET" | "POST";46 body?: unknown;47 signal?: AbortSignal;48 authenticated?: boolean;49 retrySafe?: boolean;50 expectJson?: boolean;51 /** Scope advertised after this invocation exhausts its bounded request retries. */52 exhaustedRequestRetryScope?: RetryScope;53}5455const REQUEST_ATTEMPTS = 3;5657/** How many times `withTransaction` re-drives a commit whose outcome is still unknown. */58const COMMIT_RESOLUTION_ATTEMPTS = 3;59const COMMIT_RESOLUTION_BASE_DELAY_MS = 10;6061function isRecord(value: unknown): value is JsonObject {62 return value !== null && typeof value === "object" && !Array.isArray(value);63}6465function requiredString(object: JsonObject, key: string): string {66 const value = object[key];67 if (typeof value !== "string") throw protocolError(`server response field ${key} must be a string`);68 return value;69}7071function nullableString(object: JsonObject, key: string): string | null {72 const value = object[key];73 if (value !== null && typeof value !== "string") {74 throw protocolError(`server response field ${key} must be a string or null`);75 }76 return value;77}7879function nullableNumber(object: JsonObject, key: string): number | null {80 const value = object[key];81 if (value !== null && (typeof value !== "number" || !Number.isFinite(value))) {82 throw protocolError(`server response field ${key} must be a finite number or null`);83 }84 return value;85}8687function requiredNumber(object: JsonObject, key: string): number {88 const value = object[key];89 if (typeof value !== "number" || !Number.isFinite(value)) {90 throw protocolError(`server response field ${key} must be a finite number`);91 }92 return value;93}9495function optionalString(object: JsonObject, key: string): string | null {96 const value = object[key];97 if (value === undefined || value === null) return null;98 if (typeof value !== "string") throw protocolError(`server response field ${key} must be a string when present`);99 return value;100}101102function assertMutationIdentity(input: MutationIdentity): void {103 if (!isRecord(input)) throw new TypeError("mutation input must be an object");104 if (typeof input.clientId !== "string" || input.clientId.length === 0) {105 throw new TypeError("mutation clientId must be a non-empty string");106 }107 if (!Number.isSafeInteger(input.mid) || input.mid < 1) {108 throw new TypeError("mutation mid must be a positive safe integer");109 }110}111112function normalizeRetryScope(value: unknown): RetryScope {113 return value === "request" || value === "transaction" || value === "closure" || value === "never"114 ? value115 : "never";116}117118function normalizeTransactionState(value: unknown): TransactionState | undefined {119 return value === "open" || value === "closed" || value === "unknown" ? value : undefined;120}121122function httpError(response: Response, payload: unknown, intMode: IntMode, requestId: string): RindleSqlError {123 const outer = isRecord(payload) ? payload : undefined;124 const body = outer && isRecord(outer.error) ? outer.error : outer;125 const fallbackMessage = `Rindle SQL request failed with HTTP ${response.status}`;126 const message = body && typeof body.message === "string"127 ? body.message128 : outer && typeof outer.error === "string"129 ? outer.error130 : fallbackMessage;131 const code = body && typeof body.code === "string" ? body.code : `HTTP_${response.status}`;132 const sqliteCodeRaw = body?.sqlite_code ?? body?.sqliteCode;133 const sqliteCode = typeof sqliteCodeRaw === "number" ? sqliteCodeRaw : undefined;134 const retryScopeRaw = body?.retry_scope ?? body?.retryScope;135 const inferredRetryScope = response.status >= 500 ? "request" : "never";136 const retryScope = retryScopeRaw === undefined ? inferredRetryScope : normalizeRetryScope(retryScopeRaw);137 const transactionStateRaw = body?.transaction_state ?? body?.transactionState;138 const statementIndexRaw = body?.statement_index ?? body?.statementIndex;139 const hasStatementIndex = statementIndexRaw !== undefined;140 let partialResults: StatementResult[] | undefined;141 const partialResultsRaw = body?.partial_results ?? body?.partialResults;142 const hasPartialResults = partialResultsRaw !== undefined;143 if (hasStatementIndex !== hasPartialResults) {144 throw protocolError("server error statement_index and partial_results must occur together");145 }146 let statementIndex: number | undefined;147 if (hasStatementIndex) {148 if (typeof statementIndexRaw !== "number" || !Number.isSafeInteger(statementIndexRaw) || statementIndexRaw < 0) {149 throw protocolError("server error statement_index must be a non-negative safe integer");150 }151 if (!Array.isArray(partialResultsRaw) || partialResultsRaw.length !== statementIndexRaw) {152 throw protocolError("server error partial_results length must equal statement_index");153 }154 statementIndex = statementIndexRaw;155 partialResults = partialResultsRaw.map((result) => decodeStatementResult(result, intMode));156 }157 return new RindleSqlError({158 code,159 message,160 sqliteCode,161 retryScope,162 transactionState: normalizeTransactionState(transactionStateRaw),163 status: response.status,164 // This is the logical request identity minted by this transport. A proxy's X-Request-Id is a165 // different, per-hop concept; a missing/malformed echo must not replace our stable identity.166 requestId,167 statementIndex,168 partialResults,169 });170}171172function replaceRetryScope(error: RindleSqlError, retryScope: RetryScope): RindleSqlError {173 return new RindleSqlError({174 code: error.code,175 message: error.message,176 sqliteCode: error.sqliteCode,177 retryScope,178 transactionState: error.transactionState,179 status: error.status,180 requestId: error.requestId,181 statementIndex: error.statementIndex,182 partialResults: error.partialResults,183 cause: error,184 });185}186187function isAbort(error: unknown, signal?: AbortSignal): boolean {188 return signal?.aborted === true || (error instanceof Error && error.name === "AbortError");189}190191function throwAbort(signal?: AbortSignal): never {192 if (signal?.reason !== undefined) throw signal.reason;193 if (typeof DOMException !== "undefined") throw new DOMException("The operation was aborted", "AbortError");194 const error = new Error("The operation was aborted");195 error.name = "AbortError";196 throw error;197}198199function delay(ms: number, signal?: AbortSignal): Promise<void> {200 if (signal?.aborted) throwAbort(signal);201 return new Promise((resolve, reject) => {202 const timer = setTimeout(() => {203 signal?.removeEventListener("abort", abort);204 resolve();205 }, ms);206 const abort = (): void => {207 clearTimeout(timer);208 signal?.removeEventListener("abort", abort);209 try {210 throwAbort(signal);211 } catch (error) {212 reject(error);213 }214 };215 signal?.addEventListener("abort", abort, { once: true });216 });217}218219function linkedSignal(one: AbortSignal | undefined, two: AbortSignal): { signal: AbortSignal; dispose: () => void } {220 if (one === undefined) return { signal: two, dispose: () => {} };221 const controller = new AbortController();222 const abortOne = (): void => controller.abort(one.reason);223 const abortTwo = (): void => controller.abort(two.reason);224 if (one.aborted) abortOne();225 else one.addEventListener("abort", abortOne, { once: true });226 if (two.aborted) abortTwo();227 else two.addEventListener("abort", abortTwo, { once: true });228 return {229 signal: controller.signal,230 dispose: () => {231 one.removeEventListener("abort", abortOne);232 two.removeEventListener("abort", abortTwo);233 },234 };235}236237function retryAfterMs(response: Response, attempt: number): number {238 const value = response.headers.get("retry-after");239 if (value !== null) {240 const seconds = Number(value);241 if (Number.isFinite(seconds) && seconds >= 0) return Math.min(seconds * 1_000, 1_000);242 const date = Date.parse(value);243 if (Number.isFinite(date)) return Math.max(0, Math.min(date - Date.now(), 1_000));244 }245 return Math.min(20 * 2 ** attempt, 160);246}247248class Transport {249 readonly baseUrl: string;250 readonly authToken: string;251 readonly consistency: ReadConsistency;252 readonly intMode: IntMode;253 private readonly fetchImpl: Fetch;254 private readonly closeController = new AbortController();255 private closed = false;256257 constructor(options: ClientOptions) {258 if (typeof options.url !== "string" || options.url.trim() === "") throw new TypeError("url must be a non-empty string");259 if (typeof options.authToken !== "string" || options.authToken === "") {260 throw new TypeError("authToken must be a non-empty string");261 }262 if (options.consistency !== undefined && !["session", "strong", "eventual"].includes(options.consistency)) {263 throw new TypeError(`unsupported consistency: ${String(options.consistency)}`);264 }265 if (options.intMode !== undefined && !["bigint", "number", "string"].includes(options.intMode)) {266 throw new TypeError(`unsupported intMode: ${String(options.intMode)}`);267 }268 const runtimeFetch = globalThis.fetch?.bind(globalThis) as Fetch | undefined;269 this.fetchImpl = options.fetch ?? runtimeFetch ?? (() => Promise.reject(new TypeError("global fetch is unavailable")));270 this.baseUrl = options.url.replace(/\/+$/, "");271 this.authToken = options.authToken;272 this.consistency = options.consistency ?? "session";273 this.intMode = options.intMode ?? "bigint";274 }275276 isClosed(): boolean {277 return this.closed;278 }279280 close(): void {281 if (this.closed) return;282 this.closed = true;283 this.closeController.abort(new RindleSqlError({ code: "CLIENT_CLOSED", message: "SQL client is closed" }));284 }285286 async request(path: string, options: RequestOptions = {}): Promise<unknown> {287 if (this.closed) throw new RindleSqlError({ code: "CLIENT_CLOSED", message: "SQL client is closed" });288 const method = options.method ?? "POST";289 const authenticated = options.authenticated ?? true;290 const expectJson = options.expectJson ?? true;291 const encodedBody = options.body === undefined ? undefined : JSON.stringify(options.body);292 const requestId = newRequestId();293 const attempts = options.retrySafe === true ? REQUEST_ATTEMPTS : 1;294 let lastError: unknown;295296 for (let attempt = 0; attempt < attempts; attempt += 1) {297 const linked = linkedSignal(options.signal, this.closeController.signal);298 try {299 if (linked.signal.aborted) throwAbort(linked.signal);300 const headers: Record<string, string> = {301 Accept: "application/json",302 "Rindle-Request-Id": requestId,303 };304 if (encodedBody !== undefined) headers["Content-Type"] = "application/json";305 if (authenticated) headers.Authorization = `Bearer ${this.authToken}`;306 const response = await this.fetchImpl(`${this.baseUrl}${path}`, {307 method,308 headers,309 body: encodedBody,310 signal: linked.signal,311 });312 let payload: unknown;313 const text = response.status === 204 ? "" : await response.text();314 if (text !== "") {315 try {316 payload = JSON.parse(text) as unknown;317 } catch (cause) {318 if (response.ok) throw protocolError("Rindle SQL returned invalid JSON", cause);319 }320 }321 if (response.ok) {322 if (expectJson && payload === undefined) throw protocolError("Rindle SQL returned an empty JSON response");323 return payload;324 }325326 const error = httpError(response, payload, this.intMode, requestId);327 lastError = error;328 if (attempt + 1 >= attempts || error.retryScope !== "request") throw error;329 await delay(retryAfterMs(response, attempt), options.signal);330 } catch (error) {331 if (isAbort(error, linked.signal)) {332 if (this.closed) throw new RindleSqlError({ code: "CLIENT_CLOSED", message: "SQL client is closed", cause: error });333 throwAbort(options.signal ?? linked.signal);334 }335 if (isRindleSqlError(error)) {336 lastError = error;337 if (attempt + 1 >= attempts) {338 const exhaustedScope = options.exhaustedRequestRetryScope ?? "never";339 throw error.retryScope === "request" && exhaustedScope !== "request"340 ? replaceRetryScope(error, exhaustedScope)341 : error;342 }343 if (error.retryScope !== "request") throw error;344 } else {345 lastError = error;346 if (attempt + 1 >= attempts) {347 throw new RindleSqlError({348 code: "TRANSPORT_ERROR",349 message: error instanceof Error ? error.message : "Rindle SQL transport failed",350 retryScope: options.exhaustedRequestRetryScope ?? "never",351 cause: error,352 });353 }354 }355 await delay(Math.min(20 * 2 ** attempt, 160), options.signal);356 } finally {357 linked.dispose();358 }359 }360 throw lastError;361 }362}363364function decodeStatementResult(value: unknown, intMode: IntMode): StatementResult {365 if (!isRecord(value)) throw protocolError("server statement result must be an object");366 if (!Array.isArray(value.columns)) throw protocolError("server statement result columns must be an array");367 const columns = value.columns.map((column) => {368 if (!isRecord(column) || typeof column.name !== "string" || (column.decltype !== null && typeof column.decltype !== "string")) {369 throw protocolError("server returned a malformed result column");370 }371 return { name: column.name, decltype: column.decltype };372 });373 if (!Array.isArray(value.rows)) throw protocolError("server statement result rows must be an array");374 const rows = value.rows.map((row) => {375 if (!Array.isArray(row)) throw protocolError("server returned a malformed result row");376 if (row.length !== columns.length) throw protocolError("server returned a result row with the wrong column count");377 return row.map((cell) => decodeSqlValue(cell, intMode));378 });379 return {380 columns,381 rows,382 rowsAffected: requiredNumber(value, "rows_affected"),383 lastInsertRowid: nullableString(value, "last_insert_rowid"),384 rowsRead: nullableNumber(value, "rows_read"),385 rowsWritten: nullableNumber(value, "rows_written"),386 };387}388389function decodeRouting(value: unknown): RoutingMetadata {390 if (!isRecord(value)) throw protocolError("server routing metadata must be an object");391 if (value.served_by !== "master" && value.served_by !== "follower" && value.served_by !== "standalone") {392 throw protocolError("server routing served_by must be master, follower, or standalone");393 }394 if (typeof value.fence_fallback !== "boolean") throw protocolError("server routing fence_fallback must be boolean");395 return {396 servedBy: value.served_by,397 appliedLagMs: nullableNumber(value, "applied_lag_ms"),398 fenceFallback: value.fence_fallback,399 };400}401402function decodeExecute(value: unknown, intMode: IntMode): ExecuteResult {403 if (!isRecord(value)) throw protocolError("server execute response must be an object");404 return {405 result: decodeStatementResult(value.result as WireStatementResult, intMode),406 commitCursor: nullableString(value, "commit_cursor"),407 routing: decodeRouting(value.routing as WireRoutingMetadata),408 };409}410411function decodeBatch(value: unknown, intMode: IntMode): BatchResult {412 if (!isRecord(value) || !Array.isArray(value.results)) throw protocolError("server batch response must contain results");413 return {414 results: value.results.map((result) => decodeStatementResult(result, intMode)),415 commitCursor: nullableString(value, "commit_cursor"),416 routing: decodeRouting(value.routing),417 };418}419420function decodeMutationReceipt(value: unknown): MutationReceipt {421 if (!isRecord(value) || typeof value.applied !== "boolean") {422 throw protocolError("server mutation response must contain applied and lmid");423 }424 const lmid = requiredNumber(value, "lmid");425 if (!Number.isSafeInteger(lmid) || lmid < 0) {426 throw protocolError("server mutation lmid must be a non-negative safe integer");427 }428 return {429 applied: value.applied,430 lmid,431 commitCursor: optionalString(value, "cursor"),432 };433}434435function decodeMutationRows(value: unknown, intMode: IntMode): MutationRows {436 if (!isRecord(value) || !Array.isArray(value.cols) || !value.cols.every((column) => typeof column === "string")) {437 throw protocolError("server mutation read response must contain string cols");438 }439 if (!Array.isArray(value.rows)) throw protocolError("server mutation read response must contain rows");440 const columns = value.cols as string[];441 const rows = value.rows.map((row) => {442 if (!Array.isArray(row) || row.length !== columns.length) {443 throw protocolError("server mutation read row has the wrong column count");444 }445 // The mutation-session wire predates the public SQL tagged-value response and can still carry446 // a raw boolean. All other cells use the public decoder (including future tagged values).447 return row.map((cell) => (typeof cell === "boolean" ? cell : decodeSqlValue(cell, intMode)));448 });449 return { columns, rows };450}451452function addCursor(body: JsonObject, options: { sessionCursor?: string | null } | undefined, state: CursorState): void {453 const hasOverride = options?.sessionCursor !== undefined;454 const cursor = hasOverride ? options.sessionCursor! : state.cursor;455 if (cursor !== null || hasOverride) body.session_cursor = cursor;456}457458function acceptCursor(state: CursorState, cursor: string | null): void {459 if (cursor === null) return;460461 // The current server emits `w:` plus sixteen lowercase hex digits. Responses from concurrent462 // requests can arrive out of order, so do not let an older acknowledgment move that463 // same-format fence backwards. Any other shape remains opaque and authoritative: a future464 // timeline-aware encoding must be allowed to re-anchor the client rather than being compared465 // with this process-local compatibility rule.466 const current = state.cursor;467 const parseCurrentCursor = (value: string): bigint | null => {468 if (!/^w:[0-9a-f]{16}$/.test(value)) return null;469 return BigInt(`0x${value.slice(2)}`);470 };471 if (current !== null) {472 const currentSequence = parseCurrentCursor(current);473 const nextSequence = parseCurrentCursor(cursor);474 if (currentSequence !== null && nextSequence !== null && nextSequence < currentSequence) return;475 }476 state.cursor = cursor;477}478479class Transaction implements SqlTransaction {480 private open = true;481 private busy = false;482 private operationSequence = 0n;483 private pendingExecute: { operationId: string; requestIdentity: string | null } | null = null;484 private commitOperationId: string | null = null;485 private readonly transport: Transport;486 private readonly id: string;487 private readonly cursorState: CursorState;488489 constructor(transport: Transport, id: string, cursorState: CursorState) {490 this.transport = transport;491 this.id = id;492 this.cursorState = cursorState;493 }494495 private ensureUsable(): void {496 if (!this.open) {497 throw new RindleSqlError({498 code: "TRANSACTION_CLOSED",499 message: "transaction is closed",500 retryScope: "never",501 transactionState: "closed",502 });503 }504 if (this.busy) {505 throw new RindleSqlError({506 code: "TRANSACTION_BUSY",507 message: "transaction operations must be awaited serially",508 retryScope: "never",509 transactionState: "open",510 });511 }512 }513514 private path(suffix: string): string {515 return `/v1/sql/transactions/${encodeURIComponent(this.id)}/${suffix}`;516 }517518 /** Transaction-local canonical operation IDs: 1, 2, 3, ... */519 private nextOperationId(): string {520 this.operationSequence += 1n;521 return this.operationSequence.toString();522 }523524 private async cancelAndRollback(operationId: string): Promise<void> {525 if (!this.open || this.transport.isClosed()) return;526 try {527 await this.transport.request(this.path("cancel"), {528 body: { operation_id: operationId },529 retrySafe: true,530 expectJson: false,531 });532 } catch {533 // The original abort remains the caller-visible error.534 }535 try {536 await this.transport.request(this.path("rollback"), {537 body: { operation_id: this.nextOperationId() },538 retrySafe: true,539 expectJson: false,540 });541 } catch {542 // Best effort cleanup after cancellation.543 }544 this.open = false;545 }546547 private async executeStatements(statements: Statement[], options?: OperationOptions): Promise<StatementResult[]> {548 this.ensureUsable();549 if (statements.length === 0) throw new TypeError("transaction batch requires at least one statement");550 if (this.commitOperationId !== null) {551 throw new RindleSqlError({552 code: "TRANSACTION_COMMIT_PENDING",553 message: "the commit outcome is pending; retry commit() or roll back this transaction",554 retryScope: "never",555 transactionState: "unknown",556 });557 }558 const encodedStatements = statements.map((statement) => encodeStatement(statement));559 const requestIdentity = JSON.stringify(encodedStatements);560 const pendingExecute = this.pendingExecute;561 if (pendingExecute !== null && pendingExecute.requestIdentity !== null && pendingExecute.requestIdentity !== requestIdentity) {562 throw new RindleSqlError({563 code: "TRANSACTION_OPERATION_PENDING",564 message: "retry the same transaction statement request before starting a different operation",565 retryScope: "never",566 transactionState: "open",567 });568 }569 const operationId = this.pendingExecute?.operationId ?? this.nextOperationId();570 this.pendingExecute = { operationId, requestIdentity };571 this.busy = true;572 try {573 const payload = await this.transport.request(this.path("execute"), {574 body: { statements: encodedStatements, operation_id: operationId },575 signal: options?.signal,576 retrySafe: true,577 // Unlike an autocommit method, this transaction object can preserve the exact operation578 // identity across a later explicit retry after all internal transport attempts fail.579 exhaustedRequestRetryScope: "request",580 });581 if (!isRecord(payload) || !Array.isArray(payload.results)) {582 throw protocolError("server transaction execute response must contain results");583 }584 const results = payload.results.map((result) => decodeStatementResult(result, this.transport.intMode));585 if (results.length !== encodedStatements.length) {586 throw protocolError("server transaction execute result count must match the statement count");587 }588 this.pendingExecute = null;589 return results;590 } catch (error) {591 const requestRetryable =592 isRindleSqlError(error) && error.retryScope === "request" && error.transactionState !== "closed";593 const rejectedBeforeAdmission =594 isRindleSqlError(error) &&595 error.retryScope !== "request" &&596 (error.transactionState === "unknown" ||597 (error.transactionState === undefined &&598 (error.status === 400 || error.status === 413) &&599 error.code === `HTTP_${error.status}`));600 if (rejectedBeforeAdmission) this.pendingExecute = { operationId, requestIdentity: null };601 else if (!requestRetryable) this.pendingExecute = null;602 if (options?.signal?.aborted) {603 this.pendingExecute = null;604 await this.cancelAndRollback(operationId);605 }606 if (isRindleSqlError(error) && error.transactionState === "closed") this.open = false;607 throw error;608 } finally {609 this.busy = false;610 }611 }612613 async execute(statement: Statement | string, options?: OperationOptions): Promise<StatementResult> {614 const results = await this.executeStatements([typeof statement === "string" ? { sql: statement } : statement], options);615 return results[0]!;616 }617618 batch(statements: Statement[], options?: OperationOptions): Promise<StatementResult[]> {619 return this.executeStatements(statements, options);620 }621622 async commit(options?: OperationOptions): Promise<{ commitCursor: string | null }> {623 this.ensureUsable();624 const pendingExecute = this.pendingExecute;625 if (pendingExecute !== null && pendingExecute.requestIdentity !== null) {626 throw new RindleSqlError({627 code: "TRANSACTION_OPERATION_PENDING",628 message: "retry the pending transaction statement request before committing, or roll back",629 retryScope: "never",630 transactionState: "open",631 });632 }633 // A commit may have reached the server even when its response did not reach the caller. Reuse634 // the same terminal operation identity across explicit retries so an open session still sees635 // its next expected sequence and a closed session can replay the retained terminal outcome.636 const reusableOperationId = pendingExecute?.operationId;637 this.pendingExecute = null;638 const operationId =639 this.commitOperationId ??640 (this.commitOperationId = reusableOperationId ?? this.nextOperationId());641 this.busy = true;642 try {643 const payload = await this.transport.request(this.path("commit"), {644 body: { operation_id: operationId },645 signal: options?.signal,646 retrySafe: true,647 // commitOperationId is retained across an explicit later commit(), so an exhausted648 // request error remains safely request-retryable by callers.649 exhaustedRequestRetryScope: "request",650 });651 if (!isRecord(payload)) throw protocolError("server transaction commit response must be an object");652 const commitCursor = nullableString(payload, "commit_cursor");653 acceptCursor(this.cursorState, commitCursor);654 this.open = false;655 return { commitCursor };656 } catch (error) {657 if (options?.signal?.aborted) await this.cancelAndRollback(operationId);658 if (659 isRindleSqlError(error) &&660 (error.transactionState === "closed" ||661 error.code === "TRANSACTION_CONFLICT" ||662 error.code === "TRANSACTION_EXPIRED" ||663 error.code === "TRANSACTION_CLOSED")664 ) {665 this.open = false;666 }667 throw error;668 } finally {669 this.busy = false;670 }671 }672673 async rollback(options?: OperationOptions): Promise<void> {674 if (!this.open) return;675 this.ensureUsable();676 // If a commit transport failed with an unknown outcome, its terminal sequence is still the677 // server's next expected operation when the request never arrived. Reusing it lets rollback678 // close that live transaction; if the commit did arrive, rollback of the now-missing handle is679 // already an idempotent success.680 const operationId = this.commitOperationId ?? this.nextOperationId();681 this.busy = true;682 try {683 await this.transport.request(this.path("rollback"), {684 body: { operation_id: operationId },685 signal: options?.signal,686 retrySafe: true,687 expectJson: false,688 exhaustedRequestRetryScope: "request",689 });690 this.open = false;691 } catch (error) {692 if (options?.signal?.aborted) await this.cancelAndRollback(operationId);693 if (isRindleSqlError(error) && error.transactionState === "closed") this.open = false;694 throw error;695 } finally {696 this.busy = false;697 }698 }699}700701class MutationTransaction implements SqlMutationTransaction {702 /** `closed` means the SERVER declared the transaction gone (410, or an explicit closed state), so703 * no rollback is owed. `unknown` means a request failed without an answer (5xx / transport): the704 * writer may still be held, so rollback MUST still go out even though no further work may. Only705 * `closed` suppresses the rollback — collapsing these two is what leaked the server-side writer706 * until its deadline. */707 private state: "open" | "unknown" | "closed" = "open";708 private busy = false;709 private readonly transport: Transport;710 private readonly id: string;711 private readonly cursorState: CursorState;712713 constructor(transport: Transport, id: string, cursorState: CursorState) {714 this.transport = transport;715 this.id = id;716 this.cursorState = cursorState;717 }718719 private ensureUsable(): void {720 if (this.state !== "open") {721 throw new RindleSqlError({722 code: "TRANSACTION_CLOSED",723 message:724 this.state === "closed"725 ? "mutation transaction is closed"726 : "mutation transaction is in an unknown state after a failed request",727 retryScope: "never",728 transactionState: this.state === "closed" ? "closed" : "unknown",729 });730 }731 if (this.busy) {732 throw new RindleSqlError({733 code: "TRANSACTION_BUSY",734 message: "mutation transaction operations must be awaited serially",735 retryScope: "never",736 transactionState: "open",737 });738 }739 }740741 private path(suffix: string): string {742 return `/v1/sql/mutations/transactions/${encodeURIComponent(this.id)}/${suffix}`;743 }744745 /** Classify a failed request. Only a server-DECLARED closure retires the rollback obligation; an746 * unanswered request leaves the writer possibly held, which is `unknown`, not `closed`. */747 private noteTerminalError(error: unknown): void {748 if (!isRindleSqlError(error)) return;749 if (error.status === 410 || error.transactionState === "closed") {750 this.state = "closed";751 return;752 }753 if (error.status !== undefined && error.status >= 500) this.state = "unknown";754 }755756 async execute(statement: Statement | string, options?: OperationOptions): Promise<void> {757 await this.batch([typeof statement === "string" ? { sql: statement } : statement], options);758 }759760 async batch(statements: Statement[], options?: OperationOptions): Promise<void> {761 this.ensureUsable();762 if (statements.length === 0) throw new TypeError("mutation transaction batch requires at least one statement");763 this.busy = true;764 try {765 await this.transport.request(this.path("execute"), {766 body: { statements: statements.map((statement) => encodeStatement(statement)) },767 signal: options?.signal,768 retrySafe: false,769 });770 } catch (error) {771 this.noteTerminalError(error);772 throw error;773 } finally {774 this.busy = false;775 }776 }777778 async query(statement: Statement | string, options?: OperationOptions): Promise<MutationRows> {779 this.ensureUsable();780 this.busy = true;781 try {782 return decodeMutationRows(783 await this.transport.request(this.path("query"), {784 body: { query: encodeStatement(statement) },785 signal: options?.signal,786 retrySafe: false,787 }),788 this.transport.intMode,789 );790 } catch (error) {791 this.noteTerminalError(error);792 throw error;793 } finally {794 this.busy = false;795 }796 }797798 async commit(options?: OperationOptions): Promise<MutationReceipt> {799 this.ensureUsable();800 this.busy = true;801 try {802 const receipt = decodeMutationReceipt(803 await this.transport.request(this.path("commit"), {804 body: {},805 signal: options?.signal,806 retrySafe: false,807 }),808 );809 acceptCursor(this.cursorState, receipt.commitCursor);810 this.state = "closed";811 return receipt;812 } catch (error) {813 this.noteTerminalError(error);814 throw error;815 } finally {816 this.busy = false;817 }818 }819820 async rollback(options?: OperationOptions): Promise<void> {821 // Only a server-declared closure retires the obligation. An `unknown` transaction still owes a822 // rollback — that is the whole point of the state split — so it deliberately does NOT go through823 // `ensureUsable`, which refuses everything except `open`.824 if (this.state === "closed") return;825 if (this.busy) {826 throw new RindleSqlError({827 code: "TRANSACTION_BUSY",828 message: "mutation transaction operations must be awaited serially",829 retryScope: "never",830 transactionState: "open",831 });832 }833 this.busy = true;834 try {835 await this.transport.request(this.path("rollback"), {836 body: {},837 signal: options?.signal,838 retrySafe: true,839 expectJson: false,840 exhaustedRequestRetryScope: "request",841 });842 this.state = "closed";843 } catch (error) {844 // A rollback of an already-gone transaction is the outcome the caller wanted: the server845 // answering 410 (or declaring it closed) IS the success case, not a failure to report.846 if (isRindleSqlError(error) && (error.status === 410 || error.transactionState === "closed")) {847 this.state = "closed";848 return;849 }850 throw error;851 } finally {852 this.busy = false;853 }854 }855}856857class Session implements SqlSession {858 protected readonly transport: Transport;859 protected readonly cursorState: CursorState;860861 constructor(transport: Transport, cursorState: CursorState) {862 this.transport = transport;863 this.cursorState = cursorState;864 }865866 async execute(statement: Statement | string, options?: ExecuteOptions): Promise<ExecuteResult> {867 const body: JsonObject = {868 statement: encodeStatement(statement),869 default_consistency: this.transport.consistency,870 idempotency_key: newIdempotencyKey(),871 };872 if (options?.consistency !== undefined) body.consistency = options.consistency;873 addCursor(body, options, this.cursorState);874 const result = decodeExecute(875 await this.transport.request("/v1/sql/execute", { body, signal: options?.signal, retrySafe: true }),876 this.transport.intMode,877 );878 acceptCursor(this.cursorState, result.commitCursor);879 return result;880 }881882 async batch(statements: Statement[], options?: BatchOptions): Promise<BatchResult> {883 if (statements.length === 0) throw new TypeError("batch requires at least one statement");884 const body: JsonObject = {885 statements: statements.map((statement) => encodeStatement(statement)),886 default_consistency: this.transport.consistency,887 idempotency_key: newIdempotencyKey(),888 };889 if (options?.consistency !== undefined) body.consistency = options.consistency;890 addCursor(body, options, this.cursorState);891 const result = decodeBatch(892 await this.transport.request("/v1/sql/batch", { body, signal: options?.signal, retrySafe: true }),893 this.transport.intMode,894 );895 if (result.results.length !== statements.length) {896 throw protocolError("server batch result count must match the statement count");897 }898 acceptCursor(this.cursorState, result.commitCursor);899 return result;900 }901902 async begin(options?: TransactionOptions): Promise<SqlTransaction> {903 const readOnly = options?.readOnly ?? false;904 const body: JsonObject = {905 read_only: readOnly,906 isolation: options?.isolation ?? "serializable",907 };908 if (readOnly) body.default_consistency = this.transport.consistency;909 if (options?.consistency !== undefined) body.consistency = options.consistency;910 const hasCursorOverride = options?.sessionCursor !== undefined;911 if (readOnly || hasCursorOverride) addCursor(body, options, this.cursorState);912 const payload = await this.transport.request("/v1/sql/transactions", {913 body,914 signal: options?.signal,915 retrySafe: false,916 });917 if (!isRecord(payload)) throw protocolError("server transaction begin response must be an object");918 return new Transaction(this.transport, requiredString(payload, "transaction_id"), this.cursorState);919 }920921 async withTransaction<T>(fn: (tx: SqlTransaction) => Promise<T>, options?: TransactionOptions): Promise<T> {922 if (typeof fn !== "function") throw new TypeError("withTransaction requires a callback");923 const tx = await this.begin(options);924 let value: T;925 try {926 value = await fn(tx);927 } catch (error) {928 // The callback failed, so nothing was ever submitted for commit: rolling back is both929 // truthful and the fastest way to release the server's writer connection.930 try {931 await tx.rollback();932 } catch {933 // Preserve the callback error.934 }935 throw error;936 }937 await this.commitToKnownOutcome(tx, options);938 return value;939 }940941 /**942 * Drive a transaction's commit to a KNOWN outcome.943 *944 * A commit whose response is lost is outcome-*unknown*, not failed — the server may already hold945 * a durable terminal record for it. `commit()` retains its operation id precisely so repeating946 * the same call reads that record back, and reports `retryScope: "request"` to say the call is947 * safe to repeat. Re-driving it here is the only way an application using this ergonomic surface948 * can reach that machinery.949 *950 * If the outcome is still unknown once the attempts are spent, the error is rethrown WITHOUT a951 * rollback. Rolling back would report a possibly-durable commit as aborted and leave the session952 * fence unadvanced, so the caller's natural response — re-running the mutation — would953 * double-apply it. A genuinely-open transaction is instead reclaimed by the server's session954 * lease, and the retained operation id keeps `tx.commit()` resolvable until then.955 */956 private async commitToKnownOutcome(tx: SqlTransaction, options?: TransactionOptions): Promise<void> {957 for (let attempt = 1; ; attempt += 1) {958 try {959 await tx.commit({ signal: options?.signal });960 return;961 } catch (error) {962 const outcomeUnknown = isRindleSqlError(error) && error.retryScope === "request";963 if (!outcomeUnknown) {964 // A definite negative (conflict, closed, rejected before admission): nothing committed,965 // so release the transaction before surfacing it.966 try {967 await tx.rollback();968 } catch {969 // Preserve the commit error.970 }971 throw error;972 }973 if (attempt >= COMMIT_RESOLUTION_ATTEMPTS || options?.signal?.aborted) throw error;974 await delay(COMMIT_RESOLUTION_BASE_DELAY_MS * 2 ** (attempt - 1), options?.signal);975 }976 }977 }978979 async withTransactionRetry<T>(fn: (tx: SqlTransaction) => Promise<T>, options: RetryOptions = {}): Promise<T> {980 const maxAttempts = options.maxAttempts ?? 5;981 const baseDelayMs = options.baseDelayMs ?? 10;982 const maxDelayMs = options.maxDelayMs ?? 250;983 if (!Number.isInteger(maxAttempts) || maxAttempts < 1) throw new TypeError("maxAttempts must be a positive integer");984 if (!Number.isFinite(baseDelayMs) || baseDelayMs < 0) throw new TypeError("baseDelayMs must be non-negative");985 if (!Number.isFinite(maxDelayMs) || maxDelayMs < baseDelayMs) {986 throw new TypeError("maxDelayMs must be at least baseDelayMs");987 }988 const transactionOptions: TransactionOptions = {};989 if (options.readOnly !== undefined) transactionOptions.readOnly = options.readOnly;990 if (options.isolation !== undefined) transactionOptions.isolation = options.isolation;991 if (options.consistency !== undefined) transactionOptions.consistency = options.consistency;992 if (options.sessionCursor !== undefined) transactionOptions.sessionCursor = options.sessionCursor;993 if (options.signal !== undefined) transactionOptions.signal = options.signal;994995 for (let attempt = 1; ; attempt += 1) {996 try {997 return await this.withTransaction(fn, transactionOptions);998 } catch (error) {999 const conflict = isRindleSqlError(error) && error.code === "TRANSACTION_CONFLICT";1000 if (!conflict || attempt >= maxAttempts) throw error;1001 const cap = Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1));1002 await delay(Math.random() * cap, options.signal);1003 }1004 }1005 }10061007 async executeMutation(input: ExecuteMutationInput, options?: OperationOptions): Promise<MutationReceipt> {1008 assertMutationIdentity(input);1009 if (!Array.isArray(input.statements)) throw new TypeError("mutation statements must be an array");1010 const body: JsonObject = {1011 client_id: input.clientId,1012 mid: input.mid,1013 statements: input.statements.map((statement) => encodeStatement(statement)),1014 };1015 const receipt = decodeMutationReceipt(1016 await this.transport.request("/v1/sql/mutations/execute", {1017 body,1018 signal: options?.signal,1019 // The server's mid dedup makes a response-loss retry safe and returns the stored lmid.1020 retrySafe: true,1021 exhaustedRequestRetryScope: "request",1022 }),1023 );1024 acceptCursor(this.cursorState, receipt.commitCursor);1025 return receipt;1026 }10271028 async beginMutation(input: BeginMutationInput, options?: OperationOptions): Promise<BeginMutationResult> {1029 assertMutationIdentity(input);1030 if (input.statements !== undefined && !Array.isArray(input.statements)) {1031 throw new TypeError("mutation statements must be an array when present");1032 }1033 const body: JsonObject = {1034 client_id: input.clientId,1035 mid: input.mid,1036 statements: (input.statements ?? []).map((statement) => encodeStatement(statement)),1037 };1038 if (input.query !== undefined) body.query = encodeStatement(input.query);1039 const payload = await this.transport.request("/v1/sql/mutations/transactions", {1040 body,1041 signal: options?.signal,1042 // A lost begin response may have left a live writer transaction, so begin is never retried.1043 retrySafe: false,1044 });1045 if (!isRecord(payload)) throw protocolError("server mutation begin response must be an object");1046 if (payload.absorbed === true) {1047 return { absorbed: true, receipt: decodeMutationReceipt(payload) };1048 }1049 const transactionId = requiredString(payload, "sessionId");1050 const result: BeginMutationResult = {1051 absorbed: false,1052 transaction: new MutationTransaction(this.transport, transactionId, this.cursorState),1053 };1054 if (payload.read !== undefined) result.read = decodeMutationRows(payload.read, this.transport.intMode);1055 return result;1056 }10571058 async rejectMutation(input: RejectMutationInput, options?: OperationOptions): Promise<MutationReceipt> {1059 assertMutationIdentity(input);1060 if (input.reason !== undefined && typeof input.reason !== "string") {1061 throw new TypeError("mutation rejection reason must be a string when present");1062 }1063 const body: JsonObject = { client_id: input.clientId, mid: input.mid };1064 if (input.reason !== undefined) body.reason = input.reason;1065 const receipt = decodeMutationReceipt(1066 await this.transport.request("/v1/sql/mutations/reject", {1067 body,1068 signal: options?.signal,1069 retrySafe: true,1070 exhaustedRequestRetryScope: "request",1071 }),1072 );1073 acceptCursor(this.cursorState, receipt.commitCursor);1074 return receipt;1075 }10761077 async executeDdl(sql: string, options?: OperationOptions): Promise<ExecuteResult> {1078 if (typeof sql !== "string" || sql.length === 0) throw new TypeError("sql must be a non-empty string");1079 const body: JsonObject = {1080 statement: encodeStatement(sql),1081 idempotency_key: newIdempotencyKey(),1082 };1083 const result = decodeExecute(1084 await this.transport.request("/v1/sql/execute", { body, signal: options?.signal, retrySafe: true }),1085 this.transport.intMode,1086 );1087 acceptCursor(this.cursorState, result.commitCursor);1088 return result;1089 }10901091 async migrate(input: MigrationInput, options?: OperationOptions): Promise<MigrationResult> {1092 if (!isRecord(input) || typeof input.id !== "string" || input.id.length === 0) {1093 throw new TypeError("migration id must be a non-empty string");1094 }1095 if (typeof input.checksum !== "string" || input.checksum.length === 0) {1096 throw new TypeError("migration checksum must be a non-empty string");1097 }1098 if (!Array.isArray(input.statements) || input.statements.length === 0 || !input.statements.every((sql) => typeof sql === "string")) {1099 throw new TypeError("migration statements must be a non-empty string array");1100 }1101 const payload = await this.transport.request("/v1/sql/migrate", {1102 body: { id: input.id, checksum: input.checksum, statements: input.statements },1103 signal: options?.signal,1104 retrySafe: true,1105 exhaustedRequestRetryScope: "request",1106 });1107 if (!isRecord(payload) || typeof payload.applied !== "boolean") {1108 throw protocolError("server migration response must contain applied and commit_cursor");1109 }1110 const commitCursor = requiredString(payload, "commit_cursor");1111 acceptCursor(this.cursorState, commitCursor);1112 return { applied: payload.applied, commitCursor };1113 }11141115 async executeMultiple(sql: string, options?: OperationOptions): Promise<void> {1116 if (typeof sql !== "string" || sql.length === 0) throw new TypeError("sql must be a non-empty string");1117 const payload = await this.transport.request("/v1/sql/execute-multiple", {1118 body: { sql, idempotency_key: newIdempotencyKey() },1119 signal: options?.signal,1120 retrySafe: true,1121 });1122 if (!isRecord(payload)) throw protocolError("server execute-multiple response must be an object");1123 acceptCursor(this.cursorState, nullableString(payload, "commit_cursor"));1124 }11251126 session(cursor: string | null = null): SqlSession {1127 if (cursor !== null && typeof cursor !== "string") throw new TypeError("session cursor must be a string or null");1128 return new Session(this.transport, { cursor });1129 }11301131 getSessionCursor(): string | null {1132 return this.cursorState.cursor;1133 }11341135 resetSessionCursor(): void {1136 this.cursorState.cursor = null;1137 }11381139 async ping(options?: OperationOptions): Promise<void> {1140 await this.transport.request("/version", {1141 method: "GET",1142 signal: options?.signal,1143 authenticated: false,1144 retrySafe: true,1145 expectJson: false,1146 exhaustedRequestRetryScope: "request",1147 });1148 }1149}11501151class Client extends Session implements SqlClient {1152 close(): void {1153 this.transport.close();1154 }1155}11561157export function createSqlClient(options: ClientOptions): SqlClient {1158 const transport = new Transport(options);1159 return new Client(transport, { cursor: options.sessionCursor ?? null });1160}1161