From c8ee83253f8fe9516e1c6d87697e596f612e584e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Emirhan=20Durmu=C5=9F?= Date: Mon, 27 Jul 2026 14:31:18 +0300 Subject: [PATCH] Fix SQLite write-queue freeze with self-recovery and readiness fast path. Add transaction timeouts, pool recycle, and background queue shedding so stuck sqlite writers cannot wedge the Controller indefinitely. Readiness SELECT 1 bypasses the write queue; checkFogToken runs handlers inside the auth transaction to prevent nested-enqueue deadlocks on agent hot paths. --- CHANGELOG.md | 12 +- docs/operations/database-transactions.md | 21 +- src/config/config.yaml | 14 +- src/config/env-mapping.js | 4 + src/decorators/authorization-decorator.js | 19 +- src/helpers/agent-auth-error-utils.js | 7 +- src/helpers/db-metrics.js | 24 ++ src/helpers/errors.js | 22 ++ src/helpers/transaction-runner.js | 271 ++++++++++++++++-- src/services/controller-readiness-service.js | 29 +- .../helpers/transaction-grep-gates.test.js | 10 + test/src/helpers/transaction-runner.test.js | 107 ++++++- .../src/services/controller-readiness.test.js | 43 ++- 13 files changed, 530 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8df97ea2..9f7cd9d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,15 +1,17 @@ # Changelog -## [v3.8.2] - 2026-07-25 +## [v3.8.2] - 2026-07-30 -Plan 21: liveness/readiness probe split and structured agent auth errors (coordinate with Edgelet v3.8.2). Also embedded OAuth logout/re-login hardening, EdgeOps Console **v1.0.12**, and operator sizing docs. +Plan 21: liveness/readiness probe split and structured agent auth errors (coordinate with Edgelet v3.8.2). Also embedded OAuth logout/re-login hardening, EdgeOps Console **v1.0.12**, operator sizing docs, and **SQLite write-queue self-recovery** for long-running single-node deployments. ### Added - **`GET /api/v3/live`** — public **liveness** probe (process up; always 200 while HTTP server listens). - Structured agent fog JWT errors on **`/api/v3/agent/*`**: **`code`** + **`retryable`** on 401 (credential failure) and 503 (Controller dependency failure). - **`docs/operations/sizing.md`** — hardware sizing guide by fog count (Kubernetes and Remote ControlPlane). +- **SQLite transaction recovery settings** — `settings.dbWriteQueueBackpressureDepth` (default **32**), `settings.dbTransactionTimeoutReadinessMs` (**5000**), `settings.dbTransactionTimeoutInteractiveMs` (**15000**), `settings.dbTransactionTimeoutBackgroundMs` (**120000**); env overrides **`DB_WRITE_QUEUE_BACKPRESSURE_DEPTH`**, **`DB_TRANSACTION_TIMEOUT_*_MS`** (see `docs/operations/database-transactions.md`). +- **OTEL DB metrics** — `db.transaction.timeouts` and `db.write_queue.background_shed` for queue surgery and stuck-transaction visibility. ### Changed @@ -20,9 +22,15 @@ Plan 21: liveness/readiness probe split and structured agent auth errors (coordi - Embedded OAuth BFF authorize sends **`prompt=login`** so each browser sign-in starts a fresh issuer interaction. - **`POST /api/v3/user/logout`** (embedded) clears issuer Session/Grant/Interaction state and destroys the BFF express-session when present (refresh-token revocation unchanged). - Dependency bumps: OpenTelemetry **0.221.x**, **`body-parser` 1.20.6**, **`js-yaml` 4.3.0**, **`undici` ^7.28.0**; Dockerfile base image digest pins refreshed. +- **SQLite write queue backpressure** — when total queued depth exceeds **`dbWriteQueueBackpressureDepth`**, new **background** enqueue is rejected (`QueueBackpressureError` → **503** when surfaced through agent auth); **`dbWriteQueueMaxDepth`** (**256**) remains alert-only. +- **SQLite transaction timeouts** — per-lane timeouts abort stuck work, recycle the sqlite pool, and allow the queue worker to continue (interactive **15s**, background **120s**). +- **`checkFogToken` transaction scope** — agent handler runs inside the auth transaction via **`runWithTransactionContext`**, so nested **`runInTransaction()`** reuses the parent writer instead of enqueueing a second sqlite transaction. +- **Readiness database probe (SQLite)** — **`SELECT 1`** for **`GET /api/v3/status`** runs outside the global write queue with a **5s** timeout, so health checks stay responsive under write-queue pressure. ### Fixed +- **SQLite ControlPlane freeze after multi-day uptime** — a hung sqlite transaction could block the global write queue indefinitely (queue depth > **256**, console/potctl and **`/api/v3/status`** unresponsive until restart). Self-recovery: transaction timeouts, pool recycle, and background queue shedding under sustained backpressure. +- **Agent auth nested-transaction deadlock (SQLite)** — **`checkFogToken`** authenticated in one transaction then invoked the handler outside ALS, allowing nested writes to deadlock the single-connection pool on hot agent routes (e.g. status/config polling). - **Secret PATCH** — omitted **`type`** in the update body now defaults to the existing secret type instead of **400** `Secret type mismatch` (fixes JSON PATCH and YAML secret updates that send only **`data`**). - **Embedded OAuth re-login after logout** — stale issuer session could yield **`access_denied`** on callback; logout now tears down issuer/BFF OAuth state and authorize forces fresh login. - **OAuth callback errors** — issuer errors (e.g. **`access_denied`**) redirect to **`{consoleUrl}/login?oauthError=...`** instead of **401 JSON** on the API port. diff --git a/docs/operations/database-transactions.md b/docs/operations/database-transactions.md index 38236e51..cf03c860 100644 --- a/docs/operations/database-transactions.md +++ b/docs/operations/database-transactions.md @@ -85,6 +85,10 @@ Phased reconcile and grep gates complement ALS: short txs for NATS/platform phas |---------|---------|--------------| | `settings.sqliteEnterpriseFogWarningThreshold` | 50 | `SQLITE_ENTERPRISE_FOG_WARNING_THRESHOLD` | | `settings.dbWriteQueueMaxDepth` | 256 | `DB_WRITE_QUEUE_MAX_DEPTH` | +| `settings.dbWriteQueueBackpressureDepth` | 32 | `DB_WRITE_QUEUE_BACKPRESSURE_DEPTH` | +| `settings.dbTransactionTimeoutReadinessMs` | 5000 | `DB_TRANSACTION_TIMEOUT_READINESS_MS` | +| `settings.dbTransactionTimeoutInteractiveMs` | 15000 | `DB_TRANSACTION_TIMEOUT_INTERACTIVE_MS` | +| `settings.dbTransactionTimeoutBackgroundMs` | 120000 | `DB_TRANSACTION_TIMEOUT_BACKGROUND_MS` | | `settings.dbBusyRetryMaxAttempts` | 8 | `DB_BUSY_RETRY_MAX_ATTEMPTS` | | `settings.dbBusyRetryBaseMs` | 25 | `DB_BUSY_RETRY_BASE_MS` | | `settings.reconcileOutboxDrainerIntervalSeconds` | 1 | `RECONCILE_OUTBOX_DRAINER_INTERVAL_SECONDS` | @@ -97,7 +101,19 @@ See `src/config/config.yaml` and [architecture.md](../architecture.md) for pool ## SQLite write queue backpressure -When total queued work (`interactive` + `background` lanes) exceeds `settings.dbWriteQueueMaxDepth` (default **256**), Controller logs an **error** once per overflow episode. **Interactive requests are not rejected** — the queue continues to drain in priority order. Operators should investigate background job pressure or migrate to mysql/postgres. +When total queued work (`interactive` + `background` lanes) exceeds `settings.dbWriteQueueBackpressureDepth` (default **32**), **new background enqueue attempts are rejected** with `QueueBackpressureError` (503 to callers when surfaced through agent auth). Interactive work continues to enqueue and drain in priority order. + +When depth exceeds `settings.dbWriteQueueMaxDepth` (default **256**), Controller logs an **error** once per overflow episode. Interactive requests are still not rejected at the alert threshold — operators should investigate background job pressure or migrate to mysql/postgres. + +### Self-recovery (SQLite) + +| Layer | Trigger | Action | +|-------|---------|--------| +| **L1 timeout** | Transaction exceeds lane timeout | Reject with `TransactionTimeoutError`; worker continues | +| **L2 pool recycle** | After interactive/background timeout | Close and re-init sqlite connection pool | +| **L3 queue surgery** | 3+ interactive timeouts in 60s **or** backpressure sustained 30s | Shed all queued background tasks with `QueueBackpressureError` | + +Readiness probes (`SELECT 1` for `/api/v3/status`) run **outside** the global write queue on SQLite so health checks stay responsive when the queue is wedged. Timeout defaults to `dbTransactionTimeoutReadinessMs` (5s); failures return **503** with `Retry-After`. --- @@ -110,6 +126,8 @@ Instruments are registered at startup in `src/helpers/db-metrics.js` (requires ` | `db.transaction.duration` | histogram | `label`, `priority`, `provider` | p99 spike correlated with load | | `db.write_queue.depth` | gauge | `priority` | **> 100 for 5 min** → investigate background pressure | | `db.write_queue.wait_ms` | histogram | `priority` | Sustained high wait → scale DB or reduce background load | +| `db.transaction.timeouts` | counter | `label`, `priority` | **Any sustained rate** → stuck tx or load spike | +| `db.write_queue.background_shed` | counter | `reason` | **Any increment** → queue surgery fired; check stuck writers | | `db.busy_retries` | counter | `label` | **> 10/min** → lock contention | | `db.connection.invalidated` | counter | `provider` | **Any increment** → investigate pool / connection errors | | `db.sqlite.fog_count_warning` | counter | — | Fleet exceeded sqlite recommended size | @@ -202,6 +220,7 @@ npm test -- --grep "grep gates" | **Fog platform phased reconcile** | Separate `fogPlatform.*` labels; no monolithic `fogPlatform.natsEnsure` | | **OIDC adapter** | All adapter reads/writes through `runInTransaction` with `oidc.adapter.*` labels | | **JTI cleanup** | `fog-token-cleanup-job.js` routes through `runInTransaction`, not bare manager call | +| **Agent auth ALS** | `checkFogToken` runs the handler inside `runWithTransactionContext` so nested `runInTransaction` reuses the auth tx | When a gate fails, fix the **minimal** violation (pass parent `transaction`, move I/O outside the tx body, or split phases) — do not disable the gate. diff --git a/src/config/config.yaml b/src/config/config.yaml index 53851641..80d796ae 100644 --- a/src/config/config.yaml +++ b/src/config/config.yaml @@ -98,11 +98,15 @@ settings: jobStartupDelaySeconds: 3 # Delay before reconcile-heavy background jobs start (default: 3) reconcileOutboxDrainerIntervalSeconds: 1 # Outbox drainer poll interval in seconds (default: 1) reconcileOutboxDrainerBatchSize: 32 # Max outbox rows per drainer batch (default: 32) - wsSessionReconcileIntervalSeconds: 60 # WS exec/log stale DB row reconcile interval (R89) - sqliteEnterpriseFogWarningThreshold: 50 # Log + metric when sqlite fog count exceeds (R124) - dbWriteQueueMaxDepth: 256 # SQLite write queue depth before error log (R123) - dbBusyRetryMaxAttempts: 8 # SQLITE_BUSY retry attempts per transaction (Plan 19) - dbBusyRetryBaseMs: 25 # Exponential backoff base for busy retry (Plan 19) + wsSessionReconcileIntervalSeconds: 60 # WS exec/log stale DB row reconcile interval + sqliteEnterpriseFogWarningThreshold: 50 # Log + metric when sqlite fog count exceeds + dbWriteQueueMaxDepth: 256 # SQLite write queue depth before error log + dbWriteQueueBackpressureDepth: 32 # Shed background enqueue when total queue depth exceeds (sqlite) + dbTransactionTimeoutReadinessMs: 5000 # Readiness SELECT 1 timeout (sqlite, outside write queue) + dbTransactionTimeoutInteractiveMs: 15000 # Interactive sqlite transaction timeout + dbTransactionTimeoutBackgroundMs: 120000 # Background sqlite transaction timeout + dbBusyRetryMaxAttempts: 8 # SQLITE_BUSY retry attempts per transaction + dbBusyRetryBaseMs: 25 # Exponential backoff base for busy retry # Database Configuration database: diff --git a/src/config/env-mapping.js b/src/config/env-mapping.js index 999a79d2..b977b33c 100644 --- a/src/config/env-mapping.js +++ b/src/config/env-mapping.js @@ -80,6 +80,10 @@ module.exports = { RECONCILE_OUTBOX_DRAINER_BATCH_SIZE: 'settings.reconcileOutboxDrainerBatchSize', SQLITE_ENTERPRISE_FOG_WARNING_THRESHOLD: 'settings.sqliteEnterpriseFogWarningThreshold', DB_WRITE_QUEUE_MAX_DEPTH: 'settings.dbWriteQueueMaxDepth', + DB_WRITE_QUEUE_BACKPRESSURE_DEPTH: 'settings.dbWriteQueueBackpressureDepth', + DB_TRANSACTION_TIMEOUT_READINESS_MS: 'settings.dbTransactionTimeoutReadinessMs', + DB_TRANSACTION_TIMEOUT_INTERACTIVE_MS: 'settings.dbTransactionTimeoutInteractiveMs', + DB_TRANSACTION_TIMEOUT_BACKGROUND_MS: 'settings.dbTransactionTimeoutBackgroundMs', DB_BUSY_RETRY_MAX_ATTEMPTS: 'settings.dbBusyRetryMaxAttempts', DB_BUSY_RETRY_BASE_MS: 'settings.dbBusyRetryBaseMs', diff --git a/src/decorators/authorization-decorator.js b/src/decorators/authorization-decorator.js index 810526f9..65626f43 100644 --- a/src/decorators/authorization-decorator.js +++ b/src/decorators/authorization-decorator.js @@ -5,7 +5,7 @@ const Errors = require('../helpers/errors') const CODES = require('../helpers/agent-auth-error-codes') const { classifyCheckFogTokenFailure } = require('../helpers/agent-auth-error-utils') const { isTest } = require('../helpers/app-helper') -const { runInTransaction } = require('../helpers/transaction-runner') +const { runInTransaction, runWithTransactionContext, PRIORITY_INTERACTIVE } = require('../helpers/transaction-runner') function checkFogToken (f) { return async function (...fArgs) { @@ -53,10 +53,10 @@ function checkFogToken (f) { throw new Errors.AgentAuthenticationError(CODES.AGENT_JWT_MISSING_SUBJECT, 'Agent JWT missing subject claim') } - const fog = await runInTransaction(async (transaction) => { + return runInTransaction(async (transaction) => { const foundFog = await FogManager.findOne({ uuid: fogUuid }, transaction) if (!foundFog) { - return null + throw new Errors.AgentAuthenticationError(CODES.AGENT_FOG_NOT_FOUND, 'Agent fog not found') } await FogKeyService.verifyJWT(token, fogUuid, transaction) @@ -64,17 +64,10 @@ function checkFogToken (f) { const timestamp = Date.now() await FogManager.updateLastActive(foundFog.uuid, timestamp, transaction) - return foundFog - }, { label: 'checkFogToken' }) - - if (!fog) { - logger.error(`Fog with UUID ${fogUuid} not found`) - throw new Errors.AgentAuthenticationError(CODES.AGENT_FOG_NOT_FOUND, 'Agent fog not found') - } + fArgs.push(foundFog) - fArgs.push(fog) - - return f.apply(this, fArgs) + return runWithTransactionContext(transaction, PRIORITY_INTERACTIVE, () => f.apply(this, fArgs)) + }, { label: 'checkFogToken' }) } catch (error) { const classified = classifyCheckFogTokenFailure(error) logger.error(`Agent authentication failed: ${classified.message}`, { code: classified.code }) diff --git a/src/helpers/agent-auth-error-utils.js b/src/helpers/agent-auth-error-utils.js index 185214ce..e7b7d36d 100644 --- a/src/helpers/agent-auth-error-utils.js +++ b/src/helpers/agent-auth-error-utils.js @@ -2,7 +2,9 @@ const { isSqliteBusyError } = require('./db-busy-retry') const CODES = require('./agent-auth-error-codes') const { AgentAuthenticationError, - ServiceUnavailableError + ServiceUnavailableError, + TransactionTimeoutError, + QueueBackpressureError } = require('./errors') function isVaultInfrastructureError (error) { @@ -19,6 +21,9 @@ function isDatabaseInfrastructureError (error) { if (!error) { return false } + if (error instanceof TransactionTimeoutError || error instanceof QueueBackpressureError) { + return true + } if (isSqliteBusyError(error)) { return true } diff --git a/src/helpers/db-metrics.js b/src/helpers/db-metrics.js index 847e2ad6..1a973274 100644 --- a/src/helpers/db-metrics.js +++ b/src/helpers/db-metrics.js @@ -10,6 +10,8 @@ let busyRetries = null let connectionInvalidated = null let sqliteFogCountWarning = null let writeQueueDepthInteractive = null +let transactionTimeouts = null +let backgroundQueueShed = null function getMeter () { if (!meter) { @@ -74,6 +76,13 @@ function initDbMetrics (_sequelize, _provider, queueReader) { result.observe(depth.background, { priority: 'background' }) }) } + + transactionTimeouts = m.createCounter('db.transaction.timeouts', { + description: 'SQLite transaction timeouts by label and priority lane' + }) + backgroundQueueShed = m.createCounter('db.write_queue.background_shed', { + description: 'Background SQLite write queue tasks shed during recovery surgery' + }) } function recordTransactionDuration (attributes, durationMs) { @@ -100,6 +109,19 @@ function recordSqliteFogCountWarning () { sqliteFogCountWarning?.add(1) } +function recordTransactionTimeout (label, priority) { + transactionTimeouts?.add(1, { + label: label || 'unknown', + priority: priority || 'unknown' + }) +} + +function recordBackgroundQueueShed (count, reason) { + if (count > 0) { + backgroundQueueShed?.add(count, { reason: reason || 'unknown' }) + } +} + function maybeRecordConnectionInvalidated (error, provider) { if (isConnectionInvalidatedError(error)) { recordConnectionInvalidated(provider) @@ -114,5 +136,7 @@ module.exports = { recordConnectionInvalidated, recordSqliteFogCountWarning, recordTransactionDuration, + recordTransactionTimeout, + recordBackgroundQueueShed, recordWriteQueueWaitMs } diff --git a/src/helpers/errors.js b/src/helpers/errors.js index 352febc4..c8fb4c90 100644 --- a/src/helpers/errors.js +++ b/src/helpers/errors.js @@ -159,6 +159,26 @@ class ServiceUnavailableError extends Error { } } +class TransactionTimeoutError extends Error { + constructor (label, priority, timeoutMs) { + const message = `Database transaction timed out after ${timeoutMs}ms (${label || 'unknown'}, ${priority || 'unknown'})` + super(message) + this.name = 'TransactionTimeoutError' + this.label = label || 'unknown' + this.priority = priority || 'unknown' + this.timeoutMs = timeoutMs + this.retryable = true + } +} + +class QueueBackpressureError extends Error { + constructor (message = 'SQLite write queue backpressure active') { + super(message) + this.name = 'QueueBackpressureError' + this.retryable = true + } +} + class ReadinessNotReadyError extends Error { constructor (code, message, basePayload) { super(message) @@ -184,6 +204,8 @@ module.exports = { AuthenticationError, AgentAuthenticationError, ServiceUnavailableError, + TransactionTimeoutError, + QueueBackpressureError, ReadinessNotReadyError, TransactionError, ValidationError, diff --git a/src/helpers/transaction-runner.js b/src/helpers/transaction-runner.js index a9717edf..175d6e30 100644 --- a/src/helpers/transaction-runner.js +++ b/src/helpers/transaction-runner.js @@ -3,26 +3,38 @@ const databaseProvider = require('../data/providers/database-factory') const config = require('../config') const logger = require('../logger') const { withDbBusyRetry } = require('./db-busy-retry') +const { TransactionTimeoutError, QueueBackpressureError } = require('./errors') const { maybeRecordConnectionInvalidated, recordTransactionDuration, - recordWriteQueueWaitMs + recordWriteQueueWaitMs, + recordTransactionTimeout, + recordBackgroundQueueShed } = require('./db-metrics') const PRIORITY_INTERACTIVE = 'interactive' const PRIORITY_BACKGROUND = 'background' +const READINESS_LABEL = 'readiness.database' + const interactiveLane = [] const backgroundLane = [] let workerPromise = null let queueDepthExceededLogged = false +let backpressureActiveSince = null const activeTransactionStore = new AsyncLocalStorage() +const interactiveTimeoutTimestamps = [] + const queueDepth = { interactive: 0, background: 0 } +const SURGERY_INTERACTIVE_TIMEOUT_THRESHOLD = 3 +const SURGERY_INTERACTIVE_TIMEOUT_WINDOW_MS = 60000 +const SURGERY_BACKPRESSURE_SUSTAINED_MS = 30000 + function getProviderName () { return process.env.DB_PROVIDER || config.get('database.provider', 'sqlite') || 'sqlite' } @@ -35,6 +47,27 @@ function getWriteQueueMaxDepth () { return config.get('settings.dbWriteQueueMaxDepth', 256) } +function getWriteQueueBackpressureDepth () { + return config.get('settings.dbWriteQueueBackpressureDepth', 32) +} + +function getTransactionTimeoutMs (options = {}) { + if (options.timeoutMs != null) { + return options.timeoutMs + } + + if (options.label === READINESS_LABEL) { + return config.get('settings.dbTransactionTimeoutReadinessMs', 5000) + } + + const priority = options.priority || PRIORITY_INTERACTIVE + if (priority === PRIORITY_BACKGROUND) { + return config.get('settings.dbTransactionTimeoutBackgroundMs', 120000) + } + + return config.get('settings.dbTransactionTimeoutInteractiveMs', 15000) +} + function updateQueueDepth () { queueDepth.interactive = interactiveLane.length queueDepth.background = backgroundLane.length @@ -44,15 +77,33 @@ function getWriteQueueDepth () { return { ...queueDepth } } +function getTotalQueueDepth () { + return queueDepth.interactive + queueDepth.background +} + +function isBackpressureActive () { + return getTotalQueueDepth() > getWriteQueueBackpressureDepth() +} + +function trackBackpressureState () { + if (isBackpressureActive()) { + if (!backpressureActiveSince) { + backpressureActiveSince = Date.now() + } + return + } + backpressureActiveSince = null +} + function checkQueueBackpressure () { const depth = getWriteQueueDepth() const totalDepth = depth.interactive + depth.background const maxDepth = getWriteQueueMaxDepth() + trackBackpressureState() + if (totalDepth <= maxDepth) { queueDepthExceededLogged = false - return - } - if (!queueDepthExceededLogged) { + } else if (!queueDepthExceededLogged) { queueDepthExceededLogged = true logger.error( `SQLite write queue depth ${totalDepth} exceeds configured maximum ${maxDepth} ` + @@ -61,6 +112,139 @@ function checkQueueBackpressure () { 'Interactive requests are not rejected; see docs/operations/database-transactions.md.' ) } + + maybePerformQueueSurgery('sustained-backpressure') +} + +function withTransactionTimeout (promise, options = {}) { + const timeoutMs = getTransactionTimeoutMs(options) + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { + return promise + } + + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new TransactionTimeoutError(options.label, options.priority, timeoutMs)) + }, timeoutMs) + + Promise.resolve(promise) + .then((value) => { + clearTimeout(timer) + resolve(value) + }) + .catch((error) => { + clearTimeout(timer) + reject(error) + }) + }) +} + +async function recoverSqliteConnection () { + if (!isSqliteProvider()) { + return + } + + const sequelize = databaseProvider.sequelize + const connectionManager = sequelize && sequelize.connectionManager + if (!connectionManager) { + return + } + + try { + if (connectionManager.pool) { + await connectionManager.pool.drain() + await connectionManager.pool.destroyAllNow() + } + } catch (error) { + logger.warn({ err: error.message }, 'SQLite pool drain during recovery failed') + } + + try { + connectionManager.initPools() + await sequelize.authenticate() + } catch (error) { + logger.error({ err: error.message }, 'SQLite connection pool re-init failed after recovery') + throw error + } +} + +function recordInteractiveTimeoutForSurgery () { + const now = Date.now() + interactiveTimeoutTimestamps.push(now) + while ( + interactiveTimeoutTimestamps.length > 0 && + now - interactiveTimeoutTimestamps[0] > SURGERY_INTERACTIVE_TIMEOUT_WINDOW_MS + ) { + interactiveTimeoutTimestamps.shift() + } +} + +function maybePerformQueueSurgery (reason) { + if (!isSqliteProvider()) { + return 0 + } + + let shouldShed = false + + if (reason === 'interactive-timeout') { + shouldShed = interactiveTimeoutTimestamps.length >= SURGERY_INTERACTIVE_TIMEOUT_THRESHOLD + } else if (reason === 'sustained-backpressure') { + shouldShed = backpressureActiveSince != null && + Date.now() - backpressureActiveSince >= SURGERY_BACKPRESSURE_SUSTAINED_MS + } + + if (!shouldShed || backgroundLane.length === 0) { + return 0 + } + + return shedBackgroundLane(reason) +} + +function shedBackgroundLane (reason) { + const dropped = backgroundLane.splice(0, backgroundLane.length) + updateQueueDepth() + trackBackpressureState() + + for (const item of dropped) { + item.reject(new QueueBackpressureError(`SQLite background queue shed (${reason})`)) + } + + if (dropped.length > 0) { + recordBackgroundQueueShed(dropped.length, reason) + logger.error( + `SQLite write queue surgery: shed ${dropped.length} background task(s) (${reason}); ` + + `interactiveQueued=${queueDepth.interactive}` + ) + } + + return dropped.length +} + +async function handleTransactionFailure (error, options) { + if (error instanceof TransactionTimeoutError) { + recordTransactionTimeout(options.label, options.priority) + logger.error({ + msg: 'SQLite transaction timed out', + label: options.label, + priority: options.priority, + timeoutMs: error.timeoutMs + }) + + if (isSqliteProvider()) { + try { + await recoverSqliteConnection() + } catch (recoveryError) { + logger.error({ err: recoveryError.message }, 'SQLite connection recovery failed after transaction timeout') + } + + if (options.priority === PRIORITY_INTERACTIVE) { + recordInteractiveTimeoutForSurgery() + maybePerformQueueSurgery('interactive-timeout') + } + } + } + + throw error } function dequeueNext () { @@ -87,18 +271,22 @@ async function executeTransaction (fn, options) { const provider = getProviderName() const startedAt = Date.now() const priority = options.priority || PRIORITY_INTERACTIVE + const label = options.label || 'unknown' try { - const result = await withDbBusyRetry( - () => sequelize.transaction((transaction) => { - return activeTransactionStore.run({ transaction, priority }, async () => fn(transaction)) - }), - options + const result = await withTransactionTimeout( + withDbBusyRetry( + () => sequelize.transaction((transaction) => { + return activeTransactionStore.run({ transaction, priority }, async () => fn(transaction)) + }), + options + ), + { priority, label, timeoutMs: options.timeoutMs } ) recordTransactionDuration( { - label: options.label || 'unknown', - priority: options.priority || PRIORITY_INTERACTIVE, + label, + priority, provider }, Date.now() - startedAt @@ -106,7 +294,7 @@ async function executeTransaction (fn, options) { return result } catch (error) { maybeRecordConnectionInvalidated(error, provider) - throw error + return handleTransactionFailure(error, { priority, label, timeoutMs: options.timeoutMs }) } } @@ -139,6 +327,10 @@ function ensureWorker () { } function enqueueSqlite (fn, options) { + if (options.priority === PRIORITY_BACKGROUND && isBackpressureActive()) { + return Promise.reject(new QueueBackpressureError('SQLite background enqueue rejected due to queue backpressure')) + } + return new Promise((resolve, reject) => { const item = { fn, @@ -148,7 +340,8 @@ function enqueueSqlite (fn, options) { enqueuedAt: Date.now(), retryOptions: { label: options.label, - priority: options.priority + priority: options.priority, + timeoutMs: options.timeoutMs } } @@ -167,10 +360,35 @@ function enqueueSqlite (fn, options) { async function runInTransactionPool (fn, options) { return executeTransaction(fn, { label: options.label, - priority: options.priority || PRIORITY_INTERACTIVE + priority: options.priority || PRIORITY_INTERACTIVE, + timeoutMs: options.timeoutMs }) } +/** + * Run a lightweight SQLite read outside the global write queue (WAL-safe). + * Used for readiness probes so health checks do not sit behind stuck writers. + */ +async function runSqliteReadOutsideQueue (fn, options = {}) { + if (!isSqliteProvider()) { + return fn() + } + + const label = options.label || READINESS_LABEL + const priority = options.priority || PRIORITY_INTERACTIVE + const timeoutMs = getTransactionTimeoutMs({ ...options, label }) + + try { + return await withTransactionTimeout(Promise.resolve().then(fn), { + label, + priority, + timeoutMs + }) + } catch (error) { + return handleTransactionFailure(error, { label, priority, timeoutMs }) + } +} + /** * Run a callback inside a real Sequelize transaction. * SQLite: serialized through a global priority write queue (interactive before background). @@ -181,7 +399,7 @@ async function runInTransactionPool (fn, options) { * events scheduled via setImmediate after a handler commit) cannot reuse a stale tx. * * @param {Function} fn - async (transaction) => result - * @param {{ priority?: string, label?: string }} [options] + * @param {{ priority?: string, label?: string, timeoutMs?: number }} [options] */ async function runInTransaction (fn, options = {}) { const priority = options.priority || PRIORITY_INTERACTIVE @@ -192,10 +410,10 @@ async function runInTransaction (fn, options = {}) { if (parentCtx && parentCtx.transaction && priority !== PRIORITY_BACKGROUND) { return fn(parentCtx.transaction) } - return enqueueSqlite(fn, { priority, label }) + return enqueueSqlite(fn, { priority, label, timeoutMs: options.timeoutMs }) } - return runInTransactionPool(fn, { priority, label }) + return runInTransactionPool(fn, { priority, label, timeoutMs: options.timeoutMs }) } /** @@ -225,7 +443,15 @@ async function runWithTransactionContext (transaction, priority, fn) { */ function schedulePostCommitBackground (label, fn) { setImmediate(async () => { - await runInTransaction(fn, { priority: PRIORITY_BACKGROUND, label }) + try { + await runInTransaction(fn, { priority: PRIORITY_BACKGROUND, label }) + } catch (error) { + if (error instanceof QueueBackpressureError) { + logger.warn({ msg: 'Deferred background transaction rejected by queue backpressure', label }) + return + } + logger.error({ err: error, msg: 'Deferred background transaction failed', label }) + } }) } @@ -236,19 +462,26 @@ function _resetQueueForTests () { queueDepth.interactive = 0 queueDepth.background = 0 queueDepthExceededLogged = false + backpressureActiveSince = null + interactiveTimeoutTimestamps.length = 0 } module.exports = { PRIORITY_BACKGROUND, PRIORITY_INTERACTIVE, + READINESS_LABEL, _resetQueueForTests, getActiveTransaction, getActiveTransactionContext, getProviderName, + getWriteQueueBackpressureDepth, getWriteQueueDepth, getWriteQueueMaxDepth, + getTransactionTimeoutMs, isSqliteProvider, runInTransaction, + runSqliteReadOutsideQueue, runWithTransactionContext, - schedulePostCommitBackground + schedulePostCommitBackground, + shedBackgroundLane } diff --git a/src/services/controller-readiness-service.js b/src/services/controller-readiness-service.js index ffb7fc4d..04a2d457 100644 --- a/src/services/controller-readiness-service.js +++ b/src/services/controller-readiness-service.js @@ -5,12 +5,19 @@ const authJwks = require('../config/auth-jwks') const transactionRunner = require('../helpers/transaction-runner') const { isSqliteBusyError } = require('../helpers/db-busy-retry') const CODES = require('../helpers/agent-auth-error-codes') -const { ReadinessNotReadyError } = require('../helpers/errors') +const { ReadinessNotReadyError, TransactionTimeoutError } = require('../helpers/errors') async function checkDatabaseReady () { + if (transactionRunner.isSqliteProvider()) { + await transactionRunner.runSqliteReadOutsideQueue(async () => { + await databaseProvider.sequelize.query('SELECT 1') + }, { label: transactionRunner.READINESS_LABEL }) + return + } + await transactionRunner.runInTransaction(async (transaction) => { await databaseProvider.sequelize.query('SELECT 1', { transaction }) - }, { label: 'readiness.database' }) + }, { label: transactionRunner.READINESS_LABEL }) } async function checkVaultReady () { @@ -41,14 +48,24 @@ function buildReadinessFailure (code, message) { return new ReadinessNotReadyError(code, message, null) } +function classifyDatabaseReadinessFailure (error) { + if (isSqliteBusyError(error)) { + return CODES.CONTROLLER_DB_BUSY + } + if (error instanceof TransactionTimeoutError) { + return CODES.CONTROLLER_DB_UNAVAILABLE + } + return CODES.CONTROLLER_DB_UNAVAILABLE +} + async function assertReadiness () { try { await checkDatabaseReady() } catch (error) { - const code = isSqliteBusyError(error) - ? CODES.CONTROLLER_DB_BUSY - : CODES.CONTROLLER_DB_UNAVAILABLE - throw buildReadinessFailure(code, 'Database is not ready') + throw buildReadinessFailure( + classifyDatabaseReadinessFailure(error), + 'Database is not ready' + ) } try { diff --git a/test/src/helpers/transaction-grep-gates.test.js b/test/src/helpers/transaction-grep-gates.test.js index c2786301..6147e2c0 100644 --- a/test/src/helpers/transaction-grep-gates.test.js +++ b/test/src/helpers/transaction-grep-gates.test.js @@ -245,4 +245,14 @@ describe('grep gates', () => { expect(source).to.not.match(/findOne\(\{[\s\S]*?\}, \{ transaction \}\)/) expect(source).to.not.match(/findAll\(\{[\s\S]*?\}, \{ transaction \}\)/) }) + + it('runs checkFogToken handler inside runWithTransactionContext', () => { + const authSource = fs.readFileSync( + path.join(REPO_ROOT, 'src/decorators/authorization-decorator.js'), + 'utf8' + ) + expect(authSource).to.include('runWithTransactionContext') + expect(authSource).to.match(/runWithTransactionContext\(transaction, PRIORITY_INTERACTIVE, \(\) => f\.apply\(this, fArgs\)\)/) + expect(authSource).to.not.match(/return f\.apply\(this, fArgs\)\s*\n\s*\} catch/) + }) }) diff --git a/test/src/helpers/transaction-runner.test.js b/test/src/helpers/transaction-runner.test.js index 7006f9c5..9b99d986 100644 --- a/test/src/helpers/transaction-runner.test.js +++ b/test/src/helpers/transaction-runner.test.js @@ -13,9 +13,13 @@ const { _resetQueueForTests, getActiveTransactionContext, getWriteQueueDepth, + getWriteQueueBackpressureDepth, runInTransaction, - runWithTransactionContext + runSqliteReadOutsideQueue, + runWithTransactionContext, + shedBackgroundLane } = require('../../../src/helpers/transaction-runner') +const { TransactionTimeoutError, QueueBackpressureError } = require('../../../src/helpers/errors') const { registerSqlitePragmas, applySqlitePragmas } = require('../../../src/helpers/sqlite-pragmas') describe('transaction-runner', () => { @@ -285,4 +289,105 @@ describe('transaction-runner', () => { expect(nestedCtx.priority).to.equal(PRIORITY_BACKGROUND) }) + + it('rejects background enqueue when sqlite queue exceeds backpressure depth', async () => { + let release + const gate = new Promise((resolve) => { + release = resolve + }) + + const blocker = runInTransaction(async () => { + await gate + }, { priority: PRIORITY_INTERACTIVE, label: 'blocker', timeoutMs: 60000 }) + + await new Promise((resolve) => setTimeout(resolve, 20)) + + const backpressureDepth = getWriteQueueBackpressureDepth() + for (let i = 0; i <= backpressureDepth; i++) { + runInTransaction(async () => {}, { priority: PRIORITY_BACKGROUND, label: `fill-${i}` }) + } + + await new Promise((resolve) => setTimeout(resolve, 20)) + + await expect( + runInTransaction(async () => {}, { priority: PRIORITY_BACKGROUND, label: 'rejected' }) + ).to.be.rejectedWith(QueueBackpressureError) + + release() + await blocker + }) + + it('times out hung sqlite transactions and allows subsequent work', async () => { + let release + const gate = new Promise((resolve) => { + release = resolve + }) + + const hung = runInTransaction(async () => { + await gate + }, { priority: PRIORITY_INTERACTIVE, label: 'hung', timeoutMs: 50 }) + + await expect(hung).to.be.rejectedWith(TransactionTimeoutError) + release() + + await runInTransaction(async (transaction) => { + await sequelize.query('INSERT INTO tx_runner_test (label) VALUES (\'after-timeout\')', { transaction }) + }, { label: 'after-timeout' }) + + const [rows] = await sequelize.query('SELECT label FROM tx_runner_test') + expect(rows.map((row) => row.label)).to.include('after-timeout') + }) + + it('shedBackgroundLane rejects all queued background tasks', async () => { + let release + const gate = new Promise((resolve) => { + release = resolve + }) + + const blocker = runInTransaction(async () => { + await gate + }, { priority: PRIORITY_INTERACTIVE, label: 'blocker', timeoutMs: 60000 }) + + await new Promise((resolve) => setTimeout(resolve, 20)) + + const results = [] + for (let i = 0; i < 3; i++) { + runInTransaction(async () => { + results.push(`bg-${i}`) + }, { priority: PRIORITY_BACKGROUND, label: `bg-${i}` }).catch((error) => { + results.push(error.name) + }) + } + + await new Promise((resolve) => setTimeout(resolve, 20)) + const shed = shedBackgroundLane('test') + expect(shed).to.equal(3) + + release() + await blocker + + expect(results.filter((entry) => entry === 'QueueBackpressureError')).to.have.length(3) + }) + + it('runSqliteReadOutsideQueue executes SELECT without waiting behind queued writers', async () => { + let release + const gate = new Promise((resolve) => { + release = resolve + }) + + const writerBlock = runInTransaction(async () => { + await gate + }, { priority: PRIORITY_INTERACTIVE, label: 'writer-block', timeoutMs: 60000 }) + + await new Promise((resolve) => setTimeout(resolve, 20)) + + const startedAt = Date.now() + await runSqliteReadOutsideQueue(async () => { + await sequelize.query('SELECT 1') + }, { label: 'readiness.database', timeoutMs: 5000 }) + expect(Date.now() - startedAt).to.be.lessThan(500) + + release() + await writerBlock + }) }) diff --git a/test/src/services/controller-readiness.test.js b/test/src/services/controller-readiness.test.js index 9e3e0842..81a78b06 100644 --- a/test/src/services/controller-readiness.test.js +++ b/test/src/services/controller-readiness.test.js @@ -9,7 +9,7 @@ const vaultManager = require('../../../src/vault/vault-manager') const oidcConfig = require('../../../src/config/oidc') const authJwks = require('../../../src/config/auth-jwks') const CODES = require('../../../src/helpers/agent-auth-error-codes') -const { ReadinessNotReadyError } = require('../../../src/helpers/errors') +const { ReadinessNotReadyError, TransactionTimeoutError } = require('../../../src/helpers/errors') describe('controller readiness', () => { def('sandbox', () => sinon.createSandbox()) @@ -18,7 +18,8 @@ describe('controller readiness', () => { describe('assertReadiness()', () => { it('passes when database, vault, and auth checks succeed', async () => { - $sandbox.stub(transactionRunner, 'runInTransaction').callsFake(async (fn) => fn({})) + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').callsFake(async (fn) => fn()) $sandbox.stub(databaseProvider.sequelize, 'query').resolves([[1]]) $sandbox.stub(vaultManager, 'isEnabled').returns(false) $sandbox.stub(oidcConfig, 'isAuthConfigured').returns(false) @@ -26,8 +27,22 @@ describe('controller readiness', () => { await readinessService.assertReadiness() }) + it('uses sqlite read path outside the write queue for database readiness', async () => { + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + const readOutside = $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').callsFake(async (fn) => fn()) + $sandbox.stub(databaseProvider.sequelize, 'query').resolves([[1]]) + $sandbox.stub(vaultManager, 'isEnabled').returns(false) + $sandbox.stub(oidcConfig, 'isAuthConfigured').returns(false) + + await readinessService.assertReadiness() + + expect(readOutside).to.have.been.calledOnce + expect(readOutside.firstCall.args[1]).to.deep.include({ label: transactionRunner.READINESS_LABEL }) + }) + it('throws ReadinessNotReadyError when database check fails', async () => { - $sandbox.stub(transactionRunner, 'runInTransaction').rejects(new Error('SQLITE_BUSY: database is locked')) + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').rejects(new Error('SQLITE_BUSY: database is locked')) try { await readinessService.assertReadiness() @@ -39,8 +54,25 @@ describe('controller readiness', () => { } }) + it('throws ReadinessNotReadyError when database readiness times out', async () => { + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').rejects( + new TransactionTimeoutError('readiness.database', 'interactive', 5000) + ) + + try { + await readinessService.assertReadiness() + throw new Error('expected failure') + } catch (error) { + expect(error).to.be.instanceOf(ReadinessNotReadyError) + expect(error.code).to.equal(CODES.CONTROLLER_DB_UNAVAILABLE) + expect(error.retryable).to.equal(true) + } + }) + it('throws ReadinessNotReadyError when vault check fails', async () => { - $sandbox.stub(transactionRunner, 'runInTransaction').callsFake(async (fn) => fn({})) + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').callsFake(async (fn) => fn()) $sandbox.stub(databaseProvider.sequelize, 'query').resolves([[1]]) $sandbox.stub(vaultManager, 'isEnabled').returns(true) $sandbox.stub(vaultManager, 'getProvider').returns({ @@ -57,7 +89,8 @@ describe('controller readiness', () => { }) it('throws ReadinessNotReadyError when embedded auth is not ready', async () => { - $sandbox.stub(transactionRunner, 'runInTransaction').callsFake(async (fn) => fn({})) + $sandbox.stub(transactionRunner, 'isSqliteProvider').returns(true) + $sandbox.stub(transactionRunner, 'runSqliteReadOutsideQueue').callsFake(async (fn) => fn()) $sandbox.stub(databaseProvider.sequelize, 'query').resolves([[1]]) $sandbox.stub(vaultManager, 'isEnabled').returns(false) $sandbox.stub(oidcConfig, 'isAuthConfigured').returns(true)