diff --git a/src/backend/clients/database/MySQLDatabaseClient.test.ts b/src/backend/clients/database/MySQLDatabaseClient.test.ts
index 1d74adde6..bbade448b 100644
--- a/src/backend/clients/database/MySQLDatabaseClient.test.ts
+++ b/src/backend/clients/database/MySQLDatabaseClient.test.ts
@@ -17,8 +17,12 @@
* along with this program. If not, see .
*/
-import { describe, expect, it } from 'vitest';
-import { compareMigrationFilenames } from './MySQLDatabaseClient.js';
+import { describe, expect, it, vi } from 'vitest';
+import type { IConfig } from '../../types';
+import {
+ MySQLDatabaseClient,
+ compareMigrationFilenames,
+} from './MySQLDatabaseClient.js';
describe('compareMigrationFilenames', () => {
it('orders numbered migrations numerically, not lexically', () => {
@@ -93,3 +97,76 @@ describe('compareMigrationFilenames', () => {
expect([...sorted].sort(compareMigrationFilenames)).toEqual(sorted);
});
});
+
+// ── Replica read failover ───────────────────────────────────────────
+//
+// `read()` normally goes to the replica batcher; when the replica side is
+// degraded (batcher load-shed or a transient connection error) and a real
+// replica is configured, the read retries once on the primary batcher.
+
+type Batcher = { execute: ReturnType };
+
+const makeClient = (opts: {
+ replica: Batcher;
+ primary: Batcher;
+ multiNode?: boolean;
+}) => {
+ const client = new MySQLDatabaseClient({
+ database: { engine: 'mysql' },
+ } as IConfig);
+ // Bypass onServerStart (which would connect to a real database) and
+ // inject the batchers directly. Configuration enum: SINGLE=0, REPLICA=1.
+ Object.assign(client as unknown as Record, {
+ dbReplica: opts.replica,
+ db: opts.primary,
+ configuration: opts.multiNode === false ? 0 : 1,
+ });
+ return client;
+};
+
+const codedError = (code: string) => {
+ const err = new Error(code) as Error & { code: string };
+ err.code = code;
+ return err;
+};
+
+describe('MySQLDatabaseClient.read — replica failover', () => {
+ it('fails over to the primary on batcher load-shed errors', async () => {
+ const replica = { execute: vi.fn().mockRejectedValue(codedError('dbBatchFailed')) };
+ const primary = { execute: vi.fn().mockResolvedValue([[{ ok: 1 }]]) };
+ const client = makeClient({ replica, primary });
+
+ await expect(client.read('SELECT 1')).resolves.toEqual([{ ok: 1 }]);
+ expect(primary.execute).toHaveBeenCalledTimes(1);
+ });
+
+ it('fails over on transient connection errors', async () => {
+ const replica = { execute: vi.fn().mockRejectedValue(codedError('ECONNRESET')) };
+ const primary = { execute: vi.fn().mockResolvedValue([[{ ok: 1 }]]) };
+ const client = makeClient({ replica, primary });
+
+ await expect(client.read('SELECT 1')).resolves.toEqual([{ ok: 1 }]);
+ });
+
+ it('rethrows deterministic SQL errors without touching the primary', async () => {
+ const replica = { execute: vi.fn().mockRejectedValue(codedError('ER_PARSE_ERROR')) };
+ const primary = { execute: vi.fn() };
+ const client = makeClient({ replica, primary });
+
+ await expect(client.read('SELEC oops')).rejects.toMatchObject({
+ code: 'ER_PARSE_ERROR',
+ });
+ expect(primary.execute).not.toHaveBeenCalled();
+ });
+
+ it('does not fail over in single-node configuration', async () => {
+ const replica = { execute: vi.fn().mockRejectedValue(codedError('dbBatchFailed')) };
+ const primary = { execute: vi.fn() };
+ const client = makeClient({ replica, primary, multiNode: false });
+
+ await expect(client.read('SELECT 1')).rejects.toMatchObject({
+ code: 'dbBatchFailed',
+ });
+ expect(primary.execute).not.toHaveBeenCalled();
+ });
+});
diff --git a/src/backend/clients/database/MySQLDatabaseClient.ts b/src/backend/clients/database/MySQLDatabaseClient.ts
index 964d8986f..e2bd21a80 100644
--- a/src/backend/clients/database/MySQLDatabaseClient.ts
+++ b/src/backend/clients/database/MySQLDatabaseClient.ts
@@ -19,32 +19,24 @@
import { readdirSync, readFileSync } from 'fs';
import { isAbsolute, resolve as resolvePath } from 'path';
+import { metrics } from '@opentelemetry/api';
import { createPool, type Pool } from 'mysql2';
import { Span } from '../../util/span.js';
import { AbstractDatabaseClient, type WriteResult } from './DatabaseClient';
import { SQLBatcher } from './SQLBatcher.js';
+import { isRetriableError } from './retriableErrors.js';
import { splitMysqlStatements } from './splitMysqlStatements.js';
import { compareMigrationFilenames } from './migrationFilenames.js';
import type { IConfig } from '../../types';
-const RETRIABLE_ERROR_CODES = new Set([
- 'PROTOCOL_CONNECTION_LOST',
- 'PROTOCOL_SEQUENCE_TIMEOUT',
- 'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR',
- 'ECONNRESET',
- 'ETIMEDOUT',
- 'EPIPE',
- 'ECONNREFUSED',
- 'EHOSTUNREACH',
- 'ENETUNREACH',
- 'EAI_AGAIN',
-]);
+const DEFAULT_SELECT_TIMEOUT_MS = 30_000;
-const RETRIABLE_ERROR_MESSAGES = [
- 'Connection lost',
- 'read ECONNRESET',
- 'ETIMEDOUT',
-];
+const replicaFailoverCounter = metrics
+ .getMeter('puter-backend')
+ .createCounter('db.read.replica_failover', {
+ description:
+ 'Reads that failed on the replica batcher and were retried on the primary',
+ });
export { compareMigrationFilenames };
@@ -86,7 +78,7 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
});
console.log('[mysql] connected to primary');
- this.db = new SQLBatcher(this.primaryPool, 30, 5);
+ this.db = this.createPrimaryBatcher(this.primaryPool);
if (dbConf.replica) {
this.replicaPool = this.createPool(dbConf.replica);
@@ -97,7 +89,7 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
this.configuration = Configuration.SINGLE;
}
- this.dbReplica = new SQLBatcher(this.replicaPool, 10, 5);
+ this.dbReplica = this.createReplicaBatcher(this.replicaPool);
await this.runMigrations();
}
@@ -144,7 +136,23 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
query: string,
params: unknown[] = [],
): Promise[]> {
- const result = await this.dbReplica.execute(query, params);
+ let result;
+ try {
+ result = await this.dbReplica.execute(query, params);
+ } catch (error) {
+ // Replica-side degradation (batcher load-shed or a transient
+ // connection failure) shouldn't fail reads while the primary is
+ // healthy. Deterministic errors (bad SQL) are rethrown — they
+ // would fail identically on the primary.
+ if (
+ this.configuration !== Configuration.REPLICA ||
+ !MySQLDatabaseClient.isFailoverWorthy(error)
+ ) {
+ throw error;
+ }
+ replicaFailoverCounter.add(1);
+ result = await this.db.execute(query, params);
+ }
if (!result) return [];
return (result[0] as Record[]) ?? [];
}
@@ -307,12 +315,53 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
// ------------------------------------------------------------------
private createPool(poolConfig: PoolConfig): Pool {
- return createPool({
+ const pool = createPool({
maxPreparedStatements: 900,
connectionLimit: 30,
+ enableKeepAlive: true,
...poolConfig,
multipleStatements: true,
} as PoolConfig);
+
+ // Server-side kill switch for runaway reads: MySQL applies
+ // max_execution_time to SELECT statements only, so this is
+ // write-safe. Without it, a stalled database turns reads into
+ // indefinite hangs that no client-side timeout ever converts
+ // into a failure. 0 disables.
+ const selectTimeoutMs = Math.floor(
+ Number(
+ this.config.database?.selectTimeoutMs ??
+ DEFAULT_SELECT_TIMEOUT_MS,
+ ),
+ );
+ if (selectTimeoutMs > 0) {
+ pool.on('connection', (conn) => {
+ conn.query(
+ `SET SESSION max_execution_time = ${selectTimeoutMs}`,
+ );
+ });
+ }
+
+ return pool;
+ }
+
+ private createPrimaryBatcher(pool: Pool): SQLBatcher {
+ return new SQLBatcher(pool, {
+ maxTimeInQueue: 30,
+ maxBatchSize: 5,
+ poolLabel: 'primary',
+ acquireTimeoutMs: this.config.database?.acquireTimeoutMs,
+ });
+ }
+
+ private createReplicaBatcher(pool: Pool): SQLBatcher {
+ return new SQLBatcher(pool, {
+ maxTimeInQueue: 10,
+ maxBatchSize: 5,
+ poolLabel: 'replica',
+ readOnly: true,
+ acquireTimeoutMs: this.config.database?.acquireTimeoutMs,
+ });
}
/** Reinitialize the primary pool (e.g. after a health-check failure). */
@@ -328,11 +377,11 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
password: dbConf.password ?? '',
database: dbConf.database ?? 'puter',
});
- this.db = new SQLBatcher(this.primaryPool, 30, 5);
+ this.db = this.createPrimaryBatcher(this.primaryPool);
if (this.configuration === Configuration.SINGLE) {
this.replicaPool = this.primaryPool;
- this.dbReplica = new SQLBatcher(this.primaryPool, 10, 5);
+ this.dbReplica = this.createReplicaBatcher(this.primaryPool);
}
if (previous && previous !== this.primaryPool) {
@@ -346,7 +395,7 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
const previous = this.replicaPool;
this.replicaPool = this.createPool(this.config.database.replica);
- this.dbReplica = new SQLBatcher(this.replicaPool, 10, 5);
+ this.dbReplica = this.createReplicaBatcher(this.replicaPool);
if (
previous &&
@@ -362,11 +411,14 @@ export class MySQLDatabaseClient extends AbstractDatabaseClient {
// ------------------------------------------------------------------
static isRetriableError(error: unknown): boolean {
- const code = (error as { code?: string })?.code;
- if (code && RETRIABLE_ERROR_CODES.has(code)) return true;
+ return isRetriableError(error);
+ }
- const msg = String((error as Error)?.message ?? '');
- return RETRIABLE_ERROR_MESSAGES.some((m) => msg.includes(m));
+ /** Replica failures worth retrying on the primary: batcher load-shed
+ * or transient connection errors — never deterministic SQL errors. */
+ private static isFailoverWorthy(error: unknown): boolean {
+ const code = (error as { code?: string })?.code;
+ return code === 'dbBatchFailed' || isRetriableError(error);
}
async readWithRetry(
diff --git a/src/backend/clients/database/SQLBatcher.js b/src/backend/clients/database/SQLBatcher.js
index a358f7ea7..e4fe66456 100644
--- a/src/backend/clients/database/SQLBatcher.js
+++ b/src/backend/clients/database/SQLBatcher.js
@@ -18,10 +18,19 @@
*/
import { metrics } from '@opentelemetry/api';
+import {
+ POOL_ACQUIRE_TIMEOUT,
+ isNeverSentError,
+ isRetriableError,
+} from './retriableErrors.js';
const DEFAULT_MAX_QUEUE_SIZE = 1000;
const DEFAULT_FAILURE_THRESHOLD = 5;
const DEFAULT_COOLDOWN_MS = 5_000;
+const DEFAULT_ACQUIRE_TIMEOUT_MS = 5_000;
+const ACQUIRE_ATTEMPTS = 3;
+const ITEM_RETRY_ATTEMPTS = 2;
+const RETRY_BASE_BACKOFF_MS = 100;
const FALLBACK_RETRY_CONCURRENCY = 8;
const meter = metrics.getMeter('puter-backend');
@@ -53,6 +62,15 @@ const fallbackItemFailuresCounter = meter.createCounter(
'Per-item failures observed during SQLBatcher per-item retry',
},
);
+const fallbackItemRetriesCounter = meter.createCounter(
+ 'sql_batcher.fallback.item_retries',
+ {
+ description:
+ 'Transient per-item failures retried during SQLBatcher fallback',
+ },
+);
+
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
export class SQLBatcher {
dbPool;
@@ -61,18 +79,43 @@ export class SQLBatcher {
maxQueueSize;
failureThreshold;
cooldownMs;
+ poolLabel;
+ readOnly;
+ acquireTimeoutMs;
queue = [];
timeouts = [];
#consecutiveFailures = 0;
#lastFailureAt = 0;
+ #metricAttrs;
+ /**
+ * @param {object} dbPool mysql2 pool
+ * @param {object} [opts]
+ * @param {number} [opts.maxTimeInQueue] ms an item may wait before flush
+ * @param {number} [opts.maxBatchSize] items coalesced per flush
+ * @param {number} [opts.maxQueueSize] drop-oldest high-water mark
+ * @param {number} [opts.failureThreshold] consecutive failures to open the breaker
+ * @param {number} [opts.cooldownMs] breaker open duration after last failure
+ * @param {'primary'|'replica'} [opts.poolLabel] role label on metrics; in
+ * single-node setups the 'replica' batcher shares the primary pool, so
+ * this reflects the read/write role rather than a physical instance
+ * @param {boolean} [opts.readOnly] this batcher only ever carries SELECTs,
+ * so any transient failure is safe to retry
+ * @param {number} [opts.acquireTimeoutMs] max wait for a pooled
+ * connection; 0 disables the bound
+ */
constructor(
dbPool,
- maxTimeInQueue = 20,
- maxBatchSize = 50,
- maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
- failureThreshold = DEFAULT_FAILURE_THRESHOLD,
- cooldownMs = DEFAULT_COOLDOWN_MS,
+ {
+ maxTimeInQueue = 20,
+ maxBatchSize = 50,
+ maxQueueSize = DEFAULT_MAX_QUEUE_SIZE,
+ failureThreshold = DEFAULT_FAILURE_THRESHOLD,
+ cooldownMs = DEFAULT_COOLDOWN_MS,
+ poolLabel = 'primary',
+ readOnly = false,
+ acquireTimeoutMs = DEFAULT_ACQUIRE_TIMEOUT_MS,
+ } = {},
) {
this.dbPool = dbPool;
this.maxTimeInQueue = maxTimeInQueue;
@@ -80,6 +123,10 @@ export class SQLBatcher {
this.maxQueueSize = maxQueueSize;
this.failureThreshold = failureThreshold;
this.cooldownMs = cooldownMs;
+ this.poolLabel = poolLabel;
+ this.readOnly = readOnly;
+ this.acquireTimeoutMs = acquireTimeoutMs;
+ this.#metricAttrs = { pool: poolLabel };
}
async execute(sql, values) {
@@ -90,9 +137,13 @@ export class SQLBatcher {
return this;
}
- #createPublicBatchError() {
+ // The public error is deliberately opaque (no SQL, no internals), but
+ // `reason` distinguishes the load-shed path for logs and callers:
+ // breakerOpen | queueOverflow | connAcquire.
+ #createPublicBatchError(reason) {
const error = new Error('Database operation failed');
error.code = 'dbBatchFailed';
+ error.reason = reason;
return error;
}
@@ -106,8 +157,8 @@ export class SQLBatcher {
async query(sql, values) {
if (this.#isBreakerOpen()) {
- enqueueRejectedCounter.add(1);
- throw this.#createPublicBatchError();
+ enqueueRejectedCounter.add(1, this.#metricAttrs);
+ throw this.#createPublicBatchError('breakerOpen');
}
const { promise, resolve, reject } = Promise.withResolvers();
@@ -117,8 +168,8 @@ export class SQLBatcher {
// to have already exceeded any caller-side timeout anyway.
while (this.queue.length >= this.maxQueueSize) {
const dropped = this.queue.shift();
- dropped.reject(this.#createPublicBatchError());
- enqueueDroppedCounter.add(1);
+ dropped.reject(this.#createPublicBatchError('queueOverflow'));
+ enqueueDroppedCounter.add(1, this.#metricAttrs);
}
this.queue.push({
@@ -142,6 +193,61 @@ export class SQLBatcher {
return promise;
}
+ // Bounded wait for a pooled connection. Without a bound, a stalled
+ // database turns every flush into an indefinite hang — nothing fails,
+ // so neither the breaker nor callers' own timeouts ever engage.
+ #getConnectionWithTimeout() {
+ const acquire = this.dbPool.promise().getConnection();
+ if (!this.acquireTimeoutMs) return acquire;
+
+ return new Promise((resolve, reject) => {
+ let timedOut = false;
+ const timer = setTimeout(() => {
+ timedOut = true;
+ // A connection that arrives late must go back to the pool.
+ acquire.then(
+ (conn) => conn.release(),
+ () => {},
+ );
+ const error = new Error(
+ 'Timed out acquiring database connection',
+ );
+ error.code = POOL_ACQUIRE_TIMEOUT;
+ reject(error);
+ }, this.acquireTimeoutMs);
+
+ acquire.then(
+ (conn) => {
+ if (timedOut) return;
+ clearTimeout(timer);
+ resolve(conn);
+ },
+ (err) => {
+ if (timedOut) return;
+ clearTimeout(timer);
+ reject(err);
+ },
+ );
+ });
+ }
+
+ // Acquisition failures never sent a statement, so retrying is always
+ // safe regardless of what the batch contains.
+ async #acquireConnection() {
+ let lastError;
+ for (let attempt = 1; attempt <= ACQUIRE_ATTEMPTS; attempt++) {
+ try {
+ return await this.#getConnectionWithTimeout();
+ } catch (error) {
+ lastError = error;
+ if (attempt < ACQUIRE_ATTEMPTS) {
+ await sleep(RETRY_BASE_BACKOFF_MS * attempt);
+ }
+ }
+ }
+ throw lastError;
+ }
+
async flush(batch) {
const timeout = this.timeouts.shift();
if (timeout && !timeout._destroyed) {
@@ -154,17 +260,17 @@ export class SQLBatcher {
let connection;
try {
- connection = await this.dbPool.promise().getConnection();
+ connection = await this.#acquireConnection();
} catch (error) {
this.#consecutiveFailures++;
this.#lastFailureAt = Date.now();
- flushFailureCounter.add(1);
+ flushFailureCounter.add(1, this.#metricAttrs);
console.warn(
'SQLBatcher could not acquire connection for flush:',
error,
);
for (const b of batch) {
- b.reject(this.#createPublicBatchError());
+ b.reject(this.#createPublicBatchError('connAcquire'));
}
return;
}
@@ -206,8 +312,8 @@ export class SQLBatcher {
// committed; re-running each item independently produces clean
// success/failure outcomes for each caller. Concurrency is capped to
// avoid briefly saturating the pool when a large batch fails.
- flushFailureCounter.add(1);
- fallbackInvocationsCounter.add(1);
+ flushFailureCounter.add(1, this.#metricAttrs);
+ fallbackInvocationsCounter.add(1, this.#metricAttrs);
const settled = new Array(batch.length);
let cursor = 0;
@@ -216,17 +322,7 @@ export class SQLBatcher {
async () => {
while (cursor < batch.length) {
const i = cursor++;
- const b = batch[i];
- try {
- settled[i] = {
- ok: true,
- value: await this.dbPool
- .promise()
- .query(b.sql, b.values ?? []),
- };
- } catch (error) {
- settled[i] = { ok: false, error };
- }
+ settled[i] = await this.#runFallbackItem(batch[i]);
}
},
);
@@ -246,7 +342,7 @@ export class SQLBatcher {
}
}
if (failureCount > 0) {
- fallbackItemFailuresCounter.add(failureCount);
+ fallbackItemFailuresCounter.add(failureCount, this.#metricAttrs);
}
// Only escalate the breaker when the database itself looks unhealthy
@@ -259,4 +355,39 @@ export class SQLBatcher {
this.#consecutiveFailures++;
}
}
+
+ // Run one fallback item, retrying transient failures with backoff.
+ // A read-only batcher may retry anything transient; a batcher that
+ // carries writes only retries failures where the statement provably
+ // never reached the server — a write that died mid-flight may have
+ // committed, and re-running it would double-apply.
+ async #runFallbackItem(b) {
+ let attempt = 0;
+ while (true) {
+ let connection;
+ try {
+ connection = await this.#acquireConnection();
+ } catch (error) {
+ return { ok: false, error };
+ }
+ try {
+ return {
+ ok: true,
+ value: await connection.query(b.sql, b.values ?? []),
+ };
+ } catch (error) {
+ const canRetry = this.readOnly
+ ? isRetriableError(error)
+ : isNeverSentError(error);
+ if (!canRetry || attempt >= ITEM_RETRY_ATTEMPTS) {
+ return { ok: false, error };
+ }
+ attempt++;
+ fallbackItemRetriesCounter.add(1, this.#metricAttrs);
+ await sleep(RETRY_BASE_BACKOFF_MS * attempt);
+ } finally {
+ connection.release();
+ }
+ }
+ }
}
diff --git a/src/backend/clients/database/SQLBatcher.test.ts b/src/backend/clients/database/SQLBatcher.test.ts
new file mode 100644
index 000000000..f85bf07d0
--- /dev/null
+++ b/src/backend/clients/database/SQLBatcher.test.ts
@@ -0,0 +1,239 @@
+/**
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ */
+
+import { describe, expect, it, vi } from 'vitest';
+import { SQLBatcher } from './SQLBatcher.js';
+
+const makeError = (code: string, message = code): Error & { code: string } => {
+ const error = new Error(message) as Error & { code: string };
+ error.code = code;
+ return error;
+};
+
+interface FakeConnection {
+ beginTransaction: ReturnType;
+ query: ReturnType;
+ commit: ReturnType;
+ rollback: ReturnType;
+ release: ReturnType;
+}
+
+// The batch flush sends one coalesced multi-statement (always suffixed
+// `; SELECT 1`); fallback items send their original single statement.
+const isBatchQuery = (sql: string) => sql.endsWith('; SELECT 1');
+
+const makeConnection = (
+ onQuery: (sql: string, values: unknown[]) => unknown,
+): FakeConnection => ({
+ beginTransaction: vi.fn(async () => {}),
+ query: vi.fn(async (sql: string, values: unknown[]) => onQuery(sql, values)),
+ commit: vi.fn(async () => {}),
+ rollback: vi.fn(async () => {}),
+ release: vi.fn(),
+});
+
+const makePool = (connection: FakeConnection | (() => Promise)) => {
+ const getConnection = vi.fn(async () =>
+ typeof connection === 'function' ? connection() : connection,
+ );
+ return {
+ pool: { promise: () => ({ getConnection }) },
+ getConnection,
+ };
+};
+
+// Happy-path onQuery: batch returns one result row-set per statement plus
+// the trailing SELECT 1 row-set.
+const happyBatch = (sql: string) => {
+ if (!isBatchQuery(sql)) throw new Error('unexpected fallback query');
+ const statements = sql.split(';').length - 1;
+ return [
+ Array.from({ length: statements + 1 }, (_, i) => [{ n: i }]),
+ undefined,
+ ];
+};
+
+describe('SQLBatcher', () => {
+ it('resolves each batched item with its own result', async () => {
+ const conn = makeConnection(happyBatch);
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, { maxTimeInQueue: 5 });
+
+ const [a, b] = await Promise.all([
+ batcher.query('SELECT a', []),
+ batcher.query('SELECT b', []),
+ ]);
+ expect(a[0]).toEqual([{ n: 0 }]);
+ expect(b[0]).toEqual([{ n: 1 }]);
+ expect(conn.beginTransaction).toHaveBeenCalledTimes(1);
+ expect(conn.commit).toHaveBeenCalledTimes(1);
+ expect(conn.release).toHaveBeenCalledTimes(1);
+ });
+
+ it('drops the oldest item with reason queueOverflow at the high-water mark', async () => {
+ const conn = makeConnection(happyBatch);
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, {
+ maxTimeInQueue: 5,
+ maxQueueSize: 1,
+ });
+
+ const first = batcher.query('SELECT a', []);
+ const second = batcher.query('SELECT b', []);
+
+ await expect(first).rejects.toMatchObject({
+ code: 'dbBatchFailed',
+ reason: 'queueOverflow',
+ });
+ await expect(second).resolves.toBeTruthy();
+ });
+
+ it('rejects with reason connAcquire after exhausting acquisition retries', async () => {
+ const getConnection = vi.fn(async () => {
+ throw makeError('ECONNREFUSED');
+ });
+ const pool = { promise: () => ({ getConnection }) };
+ const batcher = new SQLBatcher(pool, { maxTimeInQueue: 5 });
+
+ await expect(batcher.query('SELECT a', [])).rejects.toMatchObject({
+ code: 'dbBatchFailed',
+ reason: 'connAcquire',
+ });
+ expect(getConnection).toHaveBeenCalledTimes(3);
+ });
+
+ it('bounds connection acquisition and rejects when the pool never answers', async () => {
+ const getConnection = vi.fn(
+ () => new Promise(() => {}), // pool never yields a connection
+ );
+ const pool = { promise: () => ({ getConnection }) };
+ const batcher = new SQLBatcher(pool, {
+ maxTimeInQueue: 5,
+ acquireTimeoutMs: 30,
+ });
+
+ await expect(batcher.query('SELECT a', [])).rejects.toMatchObject({
+ code: 'dbBatchFailed',
+ reason: 'connAcquire',
+ });
+ expect(getConnection).toHaveBeenCalledTimes(3);
+ });
+
+ it('opens the breaker after consecutive failures and rejects with reason breakerOpen', async () => {
+ // Batch and fallback both fail with an ambiguous (non-retriable for
+ // writes) connection error, so no item gets through.
+ const conn = makeConnection(() => {
+ throw makeError('ECONNRESET');
+ });
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, {
+ maxTimeInQueue: 5,
+ failureThreshold: 1,
+ cooldownMs: 60_000,
+ });
+
+ await expect(batcher.query('INSERT x', [])).rejects.toMatchObject({
+ code: 'ECONNRESET',
+ });
+ await expect(batcher.query('INSERT y', [])).rejects.toMatchObject({
+ code: 'dbBatchFailed',
+ reason: 'breakerOpen',
+ });
+ });
+
+ it('retries transient fallback failures when readOnly', async () => {
+ let fallbackAttempts = 0;
+ const conn = makeConnection((sql) => {
+ if (isBatchQuery(sql)) throw makeError('ECONNRESET');
+ fallbackAttempts++;
+ if (fallbackAttempts === 1) throw makeError('ECONNRESET');
+ return [[{ ok: 1 }], undefined];
+ });
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, {
+ maxTimeInQueue: 5,
+ readOnly: true,
+ });
+
+ const result = await batcher.query('SELECT a', []);
+ expect(result[0]).toEqual([{ ok: 1 }]);
+ expect(fallbackAttempts).toBe(2);
+ });
+
+ it('does not retry ambiguous failures on a batcher that carries writes', async () => {
+ let fallbackAttempts = 0;
+ const conn = makeConnection((sql) => {
+ if (isBatchQuery(sql)) throw makeError('ECONNRESET');
+ fallbackAttempts++;
+ throw makeError('ECONNRESET');
+ });
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, { maxTimeInQueue: 5 });
+
+ await expect(batcher.query('INSERT x', [])).rejects.toMatchObject({
+ code: 'ECONNRESET',
+ });
+ expect(fallbackAttempts).toBe(1);
+ });
+
+ it('retries never-sent failures even on a batcher that carries writes', async () => {
+ let fallbackAttempts = 0;
+ const conn = makeConnection((sql) => {
+ if (isBatchQuery(sql)) throw makeError('ECONNRESET');
+ fallbackAttempts++;
+ if (fallbackAttempts === 1) throw makeError('ECONNREFUSED');
+ return [[{ ok: 1 }], undefined];
+ });
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, { maxTimeInQueue: 5 });
+
+ const result = await batcher.query('INSERT x', []);
+ expect(result[0]).toEqual([{ ok: 1 }]);
+ expect(fallbackAttempts).toBe(2);
+ });
+
+ it('never retries deterministic row-level errors and does not escalate the breaker', async () => {
+ let fallbackAttempts = 0;
+ const conn = makeConnection((sql) => {
+ if (isBatchQuery(sql)) throw makeError('ER_DUP_ENTRY');
+ fallbackAttempts++;
+ if (sql === 'INSERT dup') throw makeError('ER_DUP_ENTRY');
+ return [[{ ok: 1 }], undefined];
+ });
+ const { pool } = makePool(conn);
+ const batcher = new SQLBatcher(pool, {
+ maxTimeInQueue: 5,
+ failureThreshold: 1,
+ cooldownMs: 60_000,
+ readOnly: true,
+ });
+
+ const dup = batcher.query('INSERT dup', []);
+ const fine = batcher.query('INSERT fine', []);
+ await expect(dup).rejects.toMatchObject({ code: 'ER_DUP_ENTRY' });
+ await expect(fine).resolves.toBeTruthy();
+ expect(fallbackAttempts).toBe(2);
+
+ // One fallback item succeeded, so the breaker must stay closed.
+ const conn2 = makeConnection(happyBatch);
+ // reuse same batcher/pool: next query must not be rejected upfront
+ conn.query.mockImplementation(conn2.query.getMockImplementation()!);
+ await expect(batcher.query('SELECT a', [])).resolves.toBeTruthy();
+ });
+});
diff --git a/src/backend/clients/database/retriableErrors.ts b/src/backend/clients/database/retriableErrors.ts
new file mode 100644
index 000000000..7bd044603
--- /dev/null
+++ b/src/backend/clients/database/retriableErrors.ts
@@ -0,0 +1,78 @@
+/**
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see .
+ */
+
+/** Error code set by the pool-acquisition timeout in SQLBatcher. */
+export const POOL_ACQUIRE_TIMEOUT = 'POOL_ACQUIRE_TIMEOUT';
+
+const RETRIABLE_ERROR_CODES = new Set([
+ 'PROTOCOL_CONNECTION_LOST',
+ 'PROTOCOL_SEQUENCE_TIMEOUT',
+ 'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR',
+ 'ECONNRESET',
+ 'ETIMEDOUT',
+ 'EPIPE',
+ 'ECONNREFUSED',
+ 'EHOSTUNREACH',
+ 'ENETUNREACH',
+ 'EAI_AGAIN',
+ POOL_ACQUIRE_TIMEOUT,
+]);
+
+const RETRIABLE_ERROR_MESSAGES = [
+ 'Connection lost',
+ 'read ECONNRESET',
+ 'ETIMEDOUT',
+];
+
+/**
+ * Failures where the statement provably never reached the server, so a
+ * retry can never double-apply it — safe even for writes. Anything that
+ * can occur after the statement was sent (resets, protocol drops) is
+ * deliberately excluded: the server may have committed before the
+ * connection died.
+ */
+const NEVER_SENT_ERROR_CODES = new Set([
+ 'ECONNREFUSED',
+ 'EHOSTUNREACH',
+ 'ENETUNREACH',
+ 'EAI_AGAIN',
+ POOL_ACQUIRE_TIMEOUT,
+]);
+
+const errorCode = (error: unknown): string | undefined =>
+ (error as { code?: string } | null)?.code;
+
+/**
+ * Transient connection-level failures worth retrying — but only for
+ * statements that are safe to run twice (reads). Row-level errors
+ * (duplicate key, constraint violations) are deterministic and never
+ * match.
+ */
+export const isRetriableError = (error: unknown): boolean => {
+ const code = errorCode(error);
+ if (code && RETRIABLE_ERROR_CODES.has(code)) return true;
+
+ const msg = String((error as Error)?.message ?? '');
+ return RETRIABLE_ERROR_MESSAGES.some((m) => msg.includes(m));
+};
+
+export const isNeverSentError = (error: unknown): boolean => {
+ const code = errorCode(error);
+ return Boolean(code && NEVER_SENT_ERROR_CODES.has(code));
+};
diff --git a/src/backend/controllers/fs/FSController.ts b/src/backend/controllers/fs/FSController.ts
index 0047cd376..0f84c6ef0 100644
--- a/src/backend/controllers/fs/FSController.ts
+++ b/src/backend/controllers/fs/FSController.ts
@@ -901,7 +901,19 @@ export class FSController extends PuterController {
child.suggestedApps = rootSuggestions[index] ?? [];
}
}
- res.json(rootChildren.map((child) => this.#toClientEntry(child)));
+ const rootItems = rootChildren.map((child) =>
+ this.#toClientEntry(child),
+ );
+ if (paginated) {
+ res.json({
+ items: rootItems,
+ ...(body.includeTotal === true
+ ? { total: rootItems.length }
+ : {}),
+ });
+ return;
+ }
+ res.json(rootItems);
return;
}
@@ -944,7 +956,7 @@ export class FSController extends PuterController {
? await this.services.fs.countDirectory(parent.uuid)
: undefined;
res.json({
- items: page.entries,
+ items: page.entries.map((child) => this.#toClientEntry(child)),
...(page.cursor ? { cursor: page.cursor } : {}),
...(total !== undefined ? { total } : {}),
});
@@ -958,7 +970,7 @@ export class FSController extends PuterController {
sortOrder,
});
await this.#attachSuggestedApps(children);
- res.json(children);
+ res.json(children.map((child) => this.#toClientEntry(child)));
}
async #attachSuggestedApps(entries: FSEntry[]): Promise {
@@ -972,8 +984,6 @@ export class FSController extends PuterController {
child.suggestedApps = suggestions[index] ?? [];
}
}
-
- res.json(children.map((child) => this.#toClientEntry(child)));
}
@Post('/search', { subdomain: 'api', requireVerified: true })
@@ -1863,7 +1873,8 @@ export class FSController extends PuterController {
// the ActorUser type. Access via the escape hatch until a proper
// storage-quota mechanism is in place.
const actorUser = req.actor?.user as
- Record | undefined;
+ | Record
+ | undefined;
const candidates = [
this.#toStorageCapacityCandidate(actorUser?.free_storage),
diff --git a/src/backend/core/http/middleware/errorHandler.test.ts b/src/backend/core/http/middleware/errorHandler.test.ts
index 7ea4216f2..804683e7d 100644
--- a/src/backend/core/http/middleware/errorHandler.test.ts
+++ b/src/backend/core/http/middleware/errorHandler.test.ts
@@ -232,3 +232,50 @@ describe('createErrorHandler — when the response has already started streaming
expect(onError).toHaveBeenCalledTimes(1);
});
});
+
+// ── Database load-shed translation ──────────────────────────────────
+
+describe('createErrorHandler — dbBatchFailed load-shed errors', () => {
+ const makeDbBatchError = (reason: string) => {
+ const err = new Error('Database operation failed') as Error & {
+ code: string;
+ reason: string;
+ };
+ err.code = 'dbBatchFailed';
+ err.reason = reason;
+ return err;
+ };
+
+ it('maps to 503 + Retry-After instead of a generic 500', () => {
+ const onUnhandled = vi.fn();
+ const handler = createErrorHandler({ onUnhandled });
+ const { out } = runHandler(handler, makeDbBatchError('breakerOpen'));
+
+ expect(out.statusCode).toBe(503);
+ expect(out.headers['Retry-After']).toBe(5);
+ expect(out.body).toMatchObject({
+ code: 'db_unavailable',
+ error: 'Service temporarily unavailable',
+ });
+ // Translated errors are expected degradation, not unhandled bugs.
+ expect(onUnhandled).not.toHaveBeenCalled();
+ });
+
+ it('translates every load-shed reason the batcher emits', () => {
+ const handler = createErrorHandler({ onUnhandled: () => {} });
+ for (const reason of ['breakerOpen', 'queueOverflow', 'connAcquire']) {
+ const { out } = runHandler(handler, makeDbBatchError(reason));
+ expect(out.statusCode).toBe(503);
+ }
+ });
+
+ it('leaves unrelated coded errors on the generic 500 path', () => {
+ const onUnhandled = vi.fn();
+ const handler = createErrorHandler({ onUnhandled });
+ const err = new Error('boom') as Error & { code: string };
+ err.code = 'somethingElse';
+ const { out } = runHandler(handler, err);
+ expect(out.statusCode).toBe(500);
+ expect(onUnhandled).toHaveBeenCalledTimes(1);
+ });
+});
diff --git a/src/backend/core/http/middleware/errorHandler.ts b/src/backend/core/http/middleware/errorHandler.ts
index 824ea93db..f59188db7 100644
--- a/src/backend/core/http/middleware/errorHandler.ts
+++ b/src/backend/core/http/middleware/errorHandler.ts
@@ -85,6 +85,17 @@ export const createErrorHandler = (
err = translated;
}
+ // Database-batcher load-shed (circuit open, queue overflow, or no
+ // connection available): the persistence layer is temporarily
+ // degraded, not a programming bug. Surface as 503 so clients back
+ // off and retry instead of treating it as a hard failure.
+ if ((err as { code?: string } | null)?.code === 'dbBatchFailed') {
+ res.setHeader('Retry-After', 5);
+ err = new HttpError(503, 'Service temporarily unavailable', {
+ code: 'db_unavailable',
+ });
+ }
+
if (isHttpError(err)) {
opts.onError?.(err, req);
if (err.statusCode === 402 || err.statusCode === 413) {
diff --git a/src/backend/core/http/middleware/notFoundHandler.test.ts b/src/backend/core/http/middleware/notFoundHandler.test.ts
index 5c035d555..89d3bef45 100644
--- a/src/backend/core/http/middleware/notFoundHandler.test.ts
+++ b/src/backend/core/http/middleware/notFoundHandler.test.ts
@@ -22,25 +22,102 @@ import { describe, expect, it, vi } from 'vitest';
import { isHttpError } from '../HttpError';
import { createNotFoundHandler } from './notFoundHandler';
+const makeReq = (over: Partial = {}): Request =>
+ ({
+ method: 'GET',
+ hostname: 'puter.example',
+ path: '/no-such-page',
+ ...over,
+ }) as Request;
+
+const makeRes = () => ({
+ status: vi.fn(),
+ json: vi.fn(),
+ redirect: vi.fn(),
+});
+
+const expect404 = (next: ReturnType) => {
+ expect(next).toHaveBeenCalledTimes(1);
+ const err = next.mock.calls[0][0];
+ expect(isHttpError(err)).toBe(true);
+ expect(err.statusCode).toBe(404);
+ expect(err.legacyCode).toBe('not_found');
+};
+
describe('createNotFoundHandler', () => {
it("forwards an HttpError(404, 'not_found') to next() — does not write the response itself", () => {
// The handler must NOT call res.json/status — that's the error
// handler's job, so every failure goes through the same serializer.
const handler = createNotFoundHandler();
const next = vi.fn();
- const res = {
- status: vi.fn(),
- json: vi.fn(),
- } as unknown as Response;
- handler({} as Request, res, next);
+ const res = makeRes();
+ handler(makeReq(), res as unknown as Response, next);
- expect(next).toHaveBeenCalledTimes(1);
- const err = next.mock.calls[0][0];
- expect(isHttpError(err)).toBe(true);
- expect(err.statusCode).toBe(404);
- expect(err.legacyCode).toBe('not_found');
+ expect404(next);
// Never wrote a response directly.
- expect((res.status as ReturnType)).not.toHaveBeenCalled();
- expect((res.json as ReturnType)).not.toHaveBeenCalled();
+ expect(res.status).not.toHaveBeenCalled();
+ expect(res.json).not.toHaveBeenCalled();
+ expect(res.redirect).not.toHaveBeenCalled();
+ });
+
+ describe('with guiDomain set', () => {
+ const handler = createNotFoundHandler({ guiDomain: 'puter.example' });
+
+ it('redirects an unmatched GET on the GUI domain to /', () => {
+ const next = vi.fn();
+ const res = makeRes();
+ handler(makeReq(), res as unknown as Response, next);
+
+ expect(res.redirect).toHaveBeenCalledWith('/');
+ expect(next).not.toHaveBeenCalled();
+ });
+
+ it('redirects HEAD like GET', () => {
+ const next = vi.fn();
+ const res = makeRes();
+ handler(
+ makeReq({ method: 'HEAD' }),
+ res as unknown as Response,
+ next,
+ );
+
+ expect(res.redirect).toHaveBeenCalledWith('/');
+ expect(next).not.toHaveBeenCalled();
+ });
+
+ it('still 404s non-GET methods on the GUI domain', () => {
+ const next = vi.fn();
+ const res = makeRes();
+ handler(
+ makeReq({ method: 'POST' }),
+ res as unknown as Response,
+ next,
+ );
+
+ expect404(next);
+ expect(res.redirect).not.toHaveBeenCalled();
+ });
+
+ it('still 404s on subdomains (api., etc.)', () => {
+ const next = vi.fn();
+ const res = makeRes();
+ handler(
+ makeReq({ hostname: 'api.puter.example' }),
+ res as unknown as Response,
+ next,
+ );
+
+ expect404(next);
+ expect(res.redirect).not.toHaveBeenCalled();
+ });
+
+ it('never redirects / to itself', () => {
+ const next = vi.fn();
+ const res = makeRes();
+ handler(makeReq({ path: '/' }), res as unknown as Response, next);
+
+ expect404(next);
+ expect(res.redirect).not.toHaveBeenCalled();
+ });
});
});
diff --git a/src/backend/core/http/middleware/notFoundHandler.ts b/src/backend/core/http/middleware/notFoundHandler.ts
index 230d80ad4..564848e42 100644
--- a/src/backend/core/http/middleware/notFoundHandler.ts
+++ b/src/backend/core/http/middleware/notFoundHandler.ts
@@ -20,6 +20,16 @@
import type { RequestHandler } from 'express';
import { HttpError } from '../HttpError';
+export interface NotFoundHandlerOptions {
+ /**
+ * The bare GUI domain (`config.domain`). When set, unmatched GET/HEAD
+ * requests whose host is exactly this domain redirect to `/` instead of
+ * 404ing, so a typo'd or stale URL lands back on the desktop. Subdomains
+ * (api., etc.) and custom domains are unaffected and still 404.
+ */
+ guiDomain?: string;
+}
+
/**
* Catch-all 404 middleware. Install last (just before the error handler);
* any request that didn't match a route lands here.
@@ -28,8 +38,22 @@ import { HttpError } from '../HttpError';
* the same error-handler pipeline serializes the body — keeps the wire shape
* consistent with every other failure (`{ error: '...', code: 'not_found' }`).
*/
-export const createNotFoundHandler = (): RequestHandler => {
- return (_req, _res, next): void => {
+export const createNotFoundHandler = (
+ opts: NotFoundHandlerOptions = {},
+): RequestHandler => {
+ const guiDomain = opts.guiDomain?.trim().toLowerCase() || null;
+ return (req, res, next): void => {
+ if (
+ guiDomain &&
+ (req.method === 'GET' || req.method === 'HEAD') &&
+ req.hostname?.toLowerCase() === guiDomain &&
+ // '/' always matches the shell route; the guard just makes a
+ // misconfigured deployment 404 instead of redirect-looping.
+ req.path !== '/'
+ ) {
+ res.redirect('/');
+ return;
+ }
next(new HttpError(404, 'Not Found', { legacyCode: 'not_found' }));
};
};
diff --git a/src/backend/server.ts b/src/backend/server.ts
index 9b72d3168..c342fde49 100644
--- a/src/backend/server.ts
+++ b/src/backend/server.ts
@@ -711,7 +711,9 @@ export class PuterServer {
* without `next(err)` ceremony.
*/
#installTerminalMiddleware() {
- this.#app.use(createNotFoundHandler());
+ this.#app.use(
+ createNotFoundHandler({ guiDomain: this.#config.domain }),
+ );
this.#app.use(
createErrorHandler({
onError: (err, req) => {
diff --git a/src/backend/types.ts b/src/backend/types.ts
index 865641c09..5d5a70195 100644
--- a/src/backend/types.ts
+++ b/src/backend/types.ts
@@ -378,6 +378,18 @@ export interface IDatabaseConfig {
connectionString?: string;
url?: string;
};
+ /**
+ * Server-side execution cap for SELECT statements in ms (mysql engine;
+ * applies to both pools — MySQL only enforces it on SELECTs, so writes
+ * are unaffected). 0 disables. Default 30000.
+ */
+ selectTimeoutMs?: number;
+ /**
+ * Max time to wait for a pooled connection before the query batcher
+ * treats acquisition as failed, in ms (mysql engine). 0 disables the
+ * bound. Default 5000.
+ */
+ acquireTimeoutMs?: number;
/**
* Ordered list of directories whose `.sql` files are run sequentially at
* server start (mysql/postgres engines). Numbered migration filenames sort