From 52c7270d43bc29444e87c4cdf8987f4a65d20eae Mon Sep 17 00:00:00 2001 From: Chris Barber Date: Thu, 16 Jul 2026 01:49:33 -0500 Subject: [PATCH 1/3] Add RocksDB managed backup and restore operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New operations, also runnable from the CLI under their operation name (and offline against a stopped server): create_backup, list_backups, verify_backup, delete_backup, purge_backups, restore_backup — plus RocksDB support for get_backup, which streams a gzipped-by-default tar (gzip=false for a plain tar) and works against a remote target. Managed backups are an incremental, checksum-verified, server-side repository under storage.backupPath (default /backup), one subdirectory per database, and capture the transaction log alongside the data. Restore closes the database across all worker threads, verifies process-wide closure (registryStatus), then purges and rewrites the directory — guarded by a per-database flock plus an fsynced marker so an interrupted restore is detected and not loaded on restart. A database a loaded component holds open, and always the system database, cannot be restored online (409 pointing at running it offline with the server stopped); offline restore additionally probes RocksDB's own lock before purging. Job workers release their RocksDB handles on exit so the closure check can pass. Co-Authored-By: Claude Opus 4.8 (1M context) --- DESIGN.md | 57 ++ bin/backup.ts | 128 ++++ bin/cliOperations.ts | 112 ++-- bin/harper.ts | 11 +- components/mcp/tools/operations.ts | 3 + .../tools/schemas/operationDescriptions.ts | 3 + components/mcp/tools/schemas/operations.ts | 9 + config-root.schema.json | 4 + dataLayer/harperBridge/ResourceBridge.ts | 33 + dataLayer/restoreMarker.ts | 127 ++++ dataLayer/rocksdbBackup.ts | 567 ++++++++++++++++++ resources/databases.ts | 189 +++++- server/itc/serverHandlers.js | 8 +- server/jobs/jobProcess.ts | 19 +- server/jobs/jobRunner.ts | 10 + server/jobs/jobs.ts | 11 + server/serverHelpers/serverHandlers.js | 6 +- server/serverHelpers/serverUtilities.ts | 19 + static/defaultConfig.yaml | 1 + unitTests/dataLayer/restoreMarker.test.js | 122 ++++ unitTests/dataLayer/rocksdbBackup.test.js | 285 +++++++++ .../resources/closeLoadedDatabases.test.js | 72 +++ utility/hdbTerms.ts | 12 +- utility/operation_authorization.ts | 25 + 24 files changed, 1765 insertions(+), 68 deletions(-) create mode 100644 bin/backup.ts create mode 100644 dataLayer/restoreMarker.ts create mode 100644 dataLayer/rocksdbBackup.ts create mode 100644 unitTests/dataLayer/restoreMarker.test.js create mode 100644 unitTests/dataLayer/rocksdbBackup.test.js create mode 100644 unitTests/resources/closeLoadedDatabases.test.js diff --git a/DESIGN.md b/DESIGN.md index e26b359ff9..f1031ae134 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -305,3 +305,60 @@ guarded against (in the scan or in tar-fs's own pack walk); that's a pre-existin this fix doesn't attempt to solve. `deploy_component`/`package_component` still never validate that declared entry points (`jsResource`/`graphqlSchema`) survived extraction — a truncation from some other future cause would still report success silently; that's a deferred, separate fix. + +## RocksDB backup/restore: the restore lock + marker protocol (`dataLayer/restoreMarker.ts`, `dataLayer/rocksdbBackup.ts`) + +The `restore_backup` operation restores a user database on a live server by closing it across all +worker threads, purging its directory (`backups.restore` with `purgeAllFiles`), and reloading it. +Three non-obvious mechanics keep that safe: + +- **Two files next to (never inside) the database directory**: `.restore.lock`, an OS-level + exclusive flock (rocksdb-js `tryFileLock`, auto-released on process death), serializes restores; + `.restoring`, a marker written+fsynced (file _and_ parent directory) after the lock and + before any destructive step, means "a restore started and has not finished". Startup/rescan + detection (`databasesBlockedByRestore` in `resources/databases.ts`) checks the **marker first** + and only probes the lock when the marker exists — probes take the flock and are mutually + exclusive across threads, so probing the (persistent) lock file of every long-ago-restored + database on every rescan would make concurrent rescans misclassify healthy databases as + in-progress. Marker-present + lock-held = restore in progress (don't load); marker-present + + lock-free = crashed mid-restore (don't load; rerun the restore to recover). +- **The ITC close broadcast is best-effort, so closure is verified before the purge.** The SCHEMA + broadcast (`signalSchemaChange`) resolves after remote handlers complete but times out at 30s + "best-effort", swallows errors, and never reaches job-worker threads at all (their ports are + excluded from broadcasts to avoid re-entrant deadlocks). A destructive purge cannot trust it: + `restoreBackup` polls rocksdb-js `registryStatus()` (process-global across worker threads) until + the database path has no open instance, and aborts with a 409 — _cleaning up the marker, since + nothing was destroyed_ — if handles remain. +- **Online restore is impossible for a database a component holds open — and that failure is + correct.** rocksdb-js's registry is process-global but records only a per-path refCount, with no + attribution to a thread or component; Harper keeps no component→database ownership map. So when a + loaded component (or the `system` database, which Harper itself never stops while running) holds + its own handle on the target database, `registryStatus()` stays non-zero, Harper can neither + identify nor force-close that handle, and an in-place purge would corrupt a live instance. + `verifyDatabaseClosed` therefore waits only a short grace period (`DATABASE_CLOSE_WAIT_MS`, for a + just-finished job worker's own close to drain) and then fails fast with a 409 that points at + running the operation offline (`harper restore_backup` with the server stopped, where no + components are loaded and nothing holds the database open). Offline restore is the supported path + for component-held and `system` databases; online restore serves databases not actively held by a + component. The CLI exposes each backup operation under its operation name only (`create_backup`, + `restore_backup`, …) — no hyphenated alias — and `bin/backup.ts` routes it to a reachable server + or, when the local server is stopped, to the equivalent offline function. +- **Job workers must release their RocksDB handles on exit, or the closure check can never pass.** + rocksdb-js's registry is process-global across worker threads, and a thread that exits WITHOUT + closing leaks its handles (the refCount never drops); the only alternative, `shutdown()`, tears + down rocksdb for the _entire_ process. A job worker (`server/jobs/jobProcess.ts`) opens the whole + database graph via `getDatabases()` and exits when the job finishes — and `create_backup` is + itself a job, so before any `restore_backup` there is always at least one exited job worker that + touched the database. Without cleanup those leaked handles keep `registryStatus()` non-zero and + would fail the closure check even when no component holds the database. `jobProcess` therefore + calls `closeLoadedDatabases()` (`resources/databases.ts`) in its `finally`, closing every loaded + user database on that thread (the non-enumerable `system` DB is intentionally skipped), so an + exited job worker leaves no residual handle to be mistaken for a live holder. +- **Every path that can (re)open or delete a RocksDB database checks the restore state**, not just + the `getDatabases()` scan: `database()`'s on-demand open (`create_table`/`create_schema` would + otherwise resurrect a half-purged directory as a fresh empty DB, defeating the recovery + protocol) and `dropDatabase` (whose `destroy()` racing a restore's copy would gut a completed + restore with the marker already gone) both throw 409 via `throwIfBlockedByRestore`. + +Known limitation: the flock is process-owned; if the restore job's worker _thread_ dies without +the process exiting, the lock stays held (restores 409) until Harper restarts. diff --git a/bin/backup.ts b/bin/backup.ts new file mode 100644 index 0000000000..05faf93b0b --- /dev/null +++ b/bin/backup.ts @@ -0,0 +1,128 @@ +'use strict'; + +import { createWriteStream } from 'node:fs'; +import { basename } from 'node:path'; +import { pipeline } from 'node:stream/promises'; +import * as YAML from 'yaml'; +import * as envMgr from '../utility/environment/environmentManager.ts'; +envMgr.initSync(); +import * as terms from '../utility/hdbTerms.ts'; +import { buildRequest, cliOperations, resolveRequestOptions } from './cliOperations.ts'; +import { loadCredentials } from './cliCredentials.ts'; +import { httpRequest } from '../utility/common_utils.ts'; +import { initConfig } from '../config/configUtils.ts'; +import { getHdbPid } from '../utility/processManagement/processManagement.js'; +import { + backupDirForDatabase, + createBackupOffline, + deleteBackupOffline, + listBackupsInDir, + purgeBackupsOffline, + restoreBackupOffline, + verifyBackupOffline, +} from '../dataLayer/rocksdbBackup.ts'; + +const { OPERATIONS_ENUM } = terms; + +/** + * Runs a backup operation from the CLI. `command` is the operation name (e.g. `create_backup`); + * there is no hyphenated alias. Each works whether or not Harper is running: RocksDB is + * single-writer, so when a server is reachable the running process must own the database handle + * and the operation is forwarded to it (also handling remote `target=` and auth); only when the + * *local* server is stopped do we open the database/backup files directly. `get_backup` is the + * exception — it streams a live snapshot from a running server and has no offline form. + */ +export async function runBackupCommand(command: string): Promise { + const request = buildRequest(); + const databaseName = request.database || 'data'; + + // We print results here (not by returning a string) so the CLI's top-level handler doesn't + // re-emit them through logger.notify — this keeps output identical to forwarding the operation + // to a running server, which the online branch below simply delegates to. + if (command === OPERATIONS_ENUM.GET_BACKUP) { + console.log(await downloadBackup(request, databaseName)); + return; + } + + // Fall back to direct file access only when there is no server to talk to (no target and the + // local instance is stopped). We never fall back on a *reachable* server's error response — + // e.g. restore_backup rejecting the system/component-held database (whose offline path is to + // stop the server and rerun this same command, which then takes the branch below). + if (useOperationApi(request)) { + request.operation = command; + await cliOperations(request); // prints its own result, exactly like any forwarded operation + return; + } + + let result: any; + switch (command) { + case OPERATIONS_ENUM.CREATE_BACKUP: + result = await createBackupOffline(databaseName); + break; + case OPERATIONS_ENUM.LIST_BACKUPS: + result = await listBackupsInDir(backupDirForDatabase(databaseName)); + break; + case OPERATIONS_ENUM.VERIFY_BACKUP: + result = await verifyBackupOffline(databaseName, request.backup_id, request.verify_checksum); + break; + case OPERATIONS_ENUM.DELETE_BACKUP: + result = await deleteBackupOffline(databaseName, request.backup_id); + break; + case OPERATIONS_ENUM.PURGE_BACKUPS: + result = await purgeBackupsOffline(databaseName, request.keep_count); + break; + case OPERATIONS_ENUM.RESTORE_BACKUP: + result = await restoreBackupOffline(databaseName, request.backup_id, request.target_database); + break; + default: + throw new Error(`Unknown backup command '${command}'`); + } + console.log(YAML.stringify(result).trim()); +} + +/** + * Whether to route a backup command through the operation API rather than direct file access. + * A configured target (explicit `target=`, env, or saved `last_target`) always implies a remote + * server; otherwise we use the operation API only while the local instance is actually running + * (`getHdbPid` verifies process liveness, so a stale pid file reads as stopped). + */ +function useOperationApi(request: any): boolean { + if (request.target || process.env.HARPER_CLI_TARGET || process.env.CLI_TARGET || loadCredentials()?.last_target) { + return true; + } + initConfig(); + return Boolean(getHdbPid()); +} + +/** + * `get_backup` streams a full-snapshot tar of the database's current state from a running server + * into a local file. It works against a remote server (`target=…`) or the local instance, resolving + * the target and auth the same way as any other CLI operation; a local connection requires Harper to + * be running (there is no offline form). For RocksDB the stream is gzipped by default; pass + * `gzip=false` for a plain tar. While stopped, use `create_backup` — a local incremental directory + * backup — instead. + */ +async function downloadBackup(request: any, databaseName: string): Promise { + // only forward gzip when the user passed it — it is a RocksDB-only option (sending it + // unconditionally would fail LMDB downloads), and when unset the server applies its own + // default (RocksDB gzips) + const body: any = { operation: terms.OPERATIONS_ENUM.GET_BACKUP, database: databaseName }; + if (request.gzip !== undefined) body.gzip = request.gzip; + // resolveRequestOptions connects to a remote target (target=/env/saved) or the local domain + // socket with auth; for a local connection it enforces that Harper is running. + const { options } = await resolveRequestOptions(request); + options.streamResponse = true; + const response = await httpRequest(options, body); + if (response.statusCode !== 200) { + const chunks: Buffer[] = []; + for await (const chunk of response) chunks.push(Buffer.from(chunk)); + throw new Error(`get_backup failed (${response.statusCode}): ${Buffer.concat(chunks).toString('utf8')}`); + } + // name the file after what the server is sending (RocksDB: .tar[.gz]; LMDB: the database file). + // basename() strips any path from the server-supplied filename — a malicious/compromised target + // could otherwise return `../../…` and steer the write outside the cwd. + const serverFilename = response.headers['content-disposition']?.match(/filename="([^"]+)"/)?.[1]; + const outputPath = request.out || (serverFilename && basename(serverFilename)) || `${databaseName}.backup`; + await pipeline(response, createWriteStream(outputPath)); + return `Backup of database '${databaseName}' written to ${outputPath}`; +} diff --git a/bin/cliOperations.ts b/bin/cliOperations.ts index 9ff41eef52..76850df6a1 100644 --- a/bin/cliOperations.ts +++ b/bin/cliOperations.ts @@ -180,9 +180,16 @@ function resolveTarget(req, allCredentials) { * @param skipResponseLog By default, the response is logged to the console. Set this to true to skip logging it, which can be useful for sensitive responses like login calls! * @returns {Promise} */ -async function cliOperations(req: any, skipResponseLog = false) { - require('dotenv').config(); - +/** + * Resolve the transport options for a CLI operation request: a remote target URL (with auth) when + * one is configured (`target=`, env, or a saved `last_target`), otherwise the local domain socket. + * Returns the `options` object ready for `httpRequest` (method + Content-Type, and an Authorization + * header for remote targets, refreshing an expired operation token when possible) plus the resolved + * `target` (undefined for a local connection). Exits the process if a local connection is required + * but Harper is not running or has no domain socket. Shared by `cliOperations` and the CLI's + * streaming `get_backup` download so both reach local and remote servers the same way. + */ +export async function resolveRequestOptions(req: any): Promise<{ options: any; target: any }> { const allCredentials = loadCredentials(); req.target = normalizeTarget(resolveTarget(req, allCredentials)); let target; @@ -222,58 +229,65 @@ async function cliOperations(req: any, skipResponseLog = false) { process.exit(1); } } - await PREPARE_OPERATION[req.operation]?.(req); - try { - let options = target ?? { - protocol: 'http:', - socketPath: getConfigPath(terms.CONFIG_PARAMS.OPERATIONSAPI_NETWORK_DOMAINSOCKET), - }; - options.method = 'POST'; - options.headers = { 'Content-Type': 'application/json' }; - if (target?.username) { - options.headers.Authorization = `Basic ${Buffer.from(`${target.username}:${target.password}`).toString('base64')}`; - } else if (allCredentials) { - let tokens = null; - let lookupKey = null; - if (target && allCredentials.targets) { - lookupKey = target.resolvedTarget; - tokens = allCredentials.targets[lookupKey] ?? null; - } + let options = target ?? { + protocol: 'http:', + socketPath: getConfigPath(terms.CONFIG_PARAMS.OPERATIONSAPI_NETWORK_DOMAINSOCKET), + }; + options.method = 'POST'; + options.headers = { 'Content-Type': 'application/json' }; + if (target?.username) { + options.headers.Authorization = `Basic ${Buffer.from(`${target.username}:${target.password}`).toString('base64')}`; + } else if (allCredentials) { + let tokens = null; + let lookupKey = null; + if (target && allCredentials.targets) { + lookupKey = target.resolvedTarget; + tokens = allCredentials.targets[lookupKey] ?? null; + } - if (tokens?.operation_token) { - if (tokens.refresh_token && isJWTExpired(tokens.operation_token)) { - console.error('Operation token expired, attempting to refresh...'); - try { - const refreshOptions = { ...options }; - refreshOptions.headers = { ...options.headers, Authorization: `Bearer ${tokens.refresh_token}` }; - const refreshResponse = await httpRequest(refreshOptions, { - operation: 'refresh_operation_token', - }); - if (refreshResponse.statusCode === 200) { - const refreshData = JSON.parse(refreshResponse.body); - if (refreshData.operation_token) { - tokens.operation_token = refreshData.operation_token; - saveCredentials(lookupKey || target?.resolvedTarget, { - operation_token: tokens.operation_token, - refresh_token: tokens.refresh_token, - }); - console.error('Operation token refreshed successfully.'); - // Update the original request's authorization header with the new token - options.headers.Authorization = `Bearer ${tokens.operation_token}`; - } - } else if (refreshResponse.statusCode === 401) { - console.error('Refresh token expired or invalid. Please run harper login again.'); - process.exit(1); - } else { - console.error(`Failed to refresh operation token: ${refreshResponse.statusCode}`); + if (tokens?.operation_token) { + if (tokens.refresh_token && isJWTExpired(tokens.operation_token)) { + console.error('Operation token expired, attempting to refresh...'); + try { + const refreshOptions = { ...options }; + refreshOptions.headers = { ...options.headers, Authorization: `Bearer ${tokens.refresh_token}` }; + const refreshResponse = await httpRequest(refreshOptions, { + operation: 'refresh_operation_token', + }); + if (refreshResponse.statusCode === 200) { + const refreshData = JSON.parse(refreshResponse.body); + if (refreshData.operation_token) { + tokens.operation_token = refreshData.operation_token; + saveCredentials(lookupKey || target?.resolvedTarget, { + operation_token: tokens.operation_token, + refresh_token: tokens.refresh_token, + }); + console.error('Operation token refreshed successfully.'); + // Update the original request's authorization header with the new token + options.headers.Authorization = `Bearer ${tokens.operation_token}`; } - } catch (refreshErr) { - console.error(`Error refreshing operation token: ${refreshErr.message}`); + } else if (refreshResponse.statusCode === 401) { + console.error('Refresh token expired or invalid. Please run harper login again.'); + process.exit(1); + } else { + console.error(`Failed to refresh operation token: ${refreshResponse.statusCode}`); } + } catch (refreshErr) { + console.error(`Error refreshing operation token: ${refreshErr.message}`); } - options.headers.Authorization = `Bearer ${tokens.operation_token}`; } + options.headers.Authorization = `Bearer ${tokens.operation_token}`; } + } + return { options, target }; +} + +async function cliOperations(req: any, skipResponseLog = false) { + require('dotenv').config(); + + const { options, target } = await resolveRequestOptions(req); + await PREPARE_OPERATION[req.operation]?.(req); + try { // Streaming deploy (multipart upload + SSE progress) only works against >= 5.1 servers. // When deploying to a remote target, probe its version first and downgrade to the // legacy JSON deploy if it predates 5.1. Local (domain-socket) deploys always diff --git a/bin/harper.ts b/bin/harper.ts index 7f8bc8bafe..83586a1cb9 100644 --- a/bin/harper.ts +++ b/bin/harper.ts @@ -8,7 +8,7 @@ import * as cliOperations from './cliOperations.ts'; import { packageJson } from '../utility/packageUtils.js'; import checkNode from '../launchServiceScripts/utility/checkNodeVersion.js'; import * as hdbTerms from '../utility/hdbTerms.ts'; -const { SERVICE_ACTIONS_ENUM } = hdbTerms as any; +const { SERVICE_ACTIONS_ENUM, OPERATIONS_ENUM } = hdbTerms as any; if (typeof process.setSourceMapsEnabled === 'function') { process.setSourceMapsEnabled(true); // this is necessary for source maps to work, at least on the main thread. } @@ -25,6 +25,7 @@ By default, the CLI also supports certain Operation APIs. Specify the operation Commands: copy-db - Copies a database from source path to target path dev - Run the application in dev mode with debugging, foreground logging, no auth +get_backup database= [gzip=false] [out=] - Download a full backup from a running server (RocksDB: gzipped tar stream, gzip=false for plain tar) install - Install harperdb = - Run an API operation and return result to the CLI, not all operations are supported login [target] [username] - Login to a remote or local Harper instance @@ -112,6 +113,14 @@ async function harper() { let targetDbPath = process.argv[4]; return require('./copyDb').copyDb(sourceDb, targetDbPath); } + case OPERATIONS_ENUM.CREATE_BACKUP: + case OPERATIONS_ENUM.LIST_BACKUPS: + case OPERATIONS_ENUM.VERIFY_BACKUP: + case OPERATIONS_ENUM.DELETE_BACKUP: + case OPERATIONS_ENUM.PURGE_BACKUPS: + case OPERATIONS_ENUM.GET_BACKUP: + case OPERATIONS_ENUM.RESTORE_BACKUP: + return require('./backup').runBackupCommand(service); case SERVICE_ACTIONS_ENUM.DEV: process.env.DEV_MODE = 'true'; // fall through diff --git a/components/mcp/tools/operations.ts b/components/mcp/tools/operations.ts index c5e1bc954c..7ad6355ddb 100644 --- a/components/mcp/tools/operations.ts +++ b/components/mcp/tools/operations.ts @@ -175,6 +175,9 @@ const DESTRUCTIVE_OPERATIONS: ReadonlySet = new Set([ 'restart_service', 'set_configuration', 'remove_node', + 'restore_backup', + 'delete_backup', + 'purge_backups', ]); /** diff --git a/components/mcp/tools/schemas/operationDescriptions.ts b/components/mcp/tools/schemas/operationDescriptions.ts index 86e7d3b3e9..8eb26cec1a 100644 --- a/components/mcp/tools/schemas/operationDescriptions.ts +++ b/components/mcp/tools/schemas/operationDescriptions.ts @@ -60,6 +60,9 @@ export const OPERATION_DESCRIPTIONS: Record = { // list_deployments: components/deploymentOperations.ts:50 — Deployment history with filtering and pagination. list_deployments: 'Lists deployments from system.hdb_deployment with optional filtering by project, status, and date range, plus pagination.', + // list_backups: dataLayer/rocksdbBackup.ts (listBackups) — Managed RocksDB backup metadata for a database. + list_backups: + 'Lists the managed RocksDB directory backups for a database (id, timestamp, size, file count). Requires super_user; RocksDB databases only.', // ─── DEFAULT_ALLOW: search_* ────────────────────────────────────────── // search_by_conditions: dataLayer/search.ts:6 — Multi-condition search with comparators. diff --git a/components/mcp/tools/schemas/operations.ts b/components/mcp/tools/schemas/operations.ts index efec4e4c3d..d59bb29235 100644 --- a/components/mcp/tools/schemas/operations.ts +++ b/components/mcp/tools/schemas/operations.ts @@ -101,6 +101,15 @@ export const OPERATION_INPUT_SCHEMAS: Record = { properties: {}, description: 'Lists deployed component versions in this Harper instance.', }, + list_backups: { + type: 'object', + properties: { + database: { + type: 'string', + description: 'Database name whose managed RocksDB backups to list. Defaults to `data`.', + }, + }, + }, // ─── search_* ───────────────────────────────────────────────────────── search_by_hash: { diff --git a/config-root.schema.json b/config-root.schema.json index 2d65919cb9..bd07685ad6 100644 --- a/config-root.schema.json +++ b/config-root.schema.json @@ -486,6 +486,10 @@ }, "prefetchWrites": { "type": "boolean", "description": "Load data prior to write transactions. Default: true" }, "path": { "type": "string", "description": "Directory for all database files. Default: /database" }, + "backupPath": { + "type": "string", + "description": "Directory for managed database backups (RocksDB directory backups created by create_backup), one subdirectory per database. Default: /backup" + }, "blobPaths": { "oneOf": [{ "type": "string" }, { "type": "array", "items": { "type": "string" } }], "description": "Path or array of paths for blob storage. Default: /blobs" diff --git a/dataLayer/harperBridge/ResourceBridge.ts b/dataLayer/harperBridge/ResourceBridge.ts index f00d58028a..3b66662407 100644 --- a/dataLayer/harperBridge/ResourceBridge.ts +++ b/dataLayer/harperBridge/ResourceBridge.ts @@ -27,6 +27,7 @@ import { errorToString } from '../../utility/logging/harper_logger.ts'; import { RocksDatabase } from '@harperfast/rocksdb-js'; import { BridgeMethods } from './BridgeMethods.ts'; import lmdbGetBackup from './lmdbBridge/lmdbMethods/lmdbGetBackup.js'; +import { createBackupStream, resolveSingleRootStore } from '../rocksdbBackup.ts'; import { DeleteTransactionLogsBeforeResults } from './DeleteTransactionLogsBeforeResults.ts'; import type { Readable } from 'node:stream'; @@ -556,7 +557,39 @@ export class ResourceBridge extends BridgeMethods { schema?: string; table?: string; tables?: string[]; + include_audit?: boolean; + gzip?: boolean; }): Promise { + const databaseName = getBackupObj.database || getBackupObj.schema || 'data'; + const database = getDatabases()[databaseName]; + if (!database) { + throw new ClientError(`Database '${databaseName}' does not exist`, 404); + } + const firstTable = database[Object.keys(database)[0]]; + if (!firstTable) { + throw new ClientError(`Database '${databaseName}' has no tables to back up`); + } + if (firstTable.primaryStore.rootStore instanceof RocksDatabase) { + // RocksDB: stream a fresh full-snapshot tar of the database's current state — no + // scratch disk, nothing to clean up. Per-engine params are validated descriptively. + if (getBackupObj.tables || getBackupObj.table) { + throw new ClientError(`'tables'/'table' are LMDB-only options; RocksDB backups are always whole-database`); + } + if (getBackupObj.include_audit !== undefined) { + throw new ClientError( + `'include_audit' is an LMDB-only option; RocksDB backups always include the transaction log` + ); + } + if (getBackupObj.gzip !== undefined && typeof getBackupObj.gzip !== 'boolean') { + throw new ClientError(`'gzip' must be a boolean`); + } + const rootStore = resolveSingleRootStore(databaseName); + // gzip defaults on (it compresses the snapshot substantially); gzip=false opts out. + return createBackupStream(rootStore, databaseName, getBackupObj.gzip !== false); + } + if (getBackupObj.gzip !== undefined) { + throw new ClientError(`'gzip' is a RocksDB-only option; LMDB backups are gzipped per the accept-encoding header`); + } return lmdbGetBackup(getBackupObj); } } diff --git a/dataLayer/restoreMarker.ts b/dataLayer/restoreMarker.ts new file mode 100644 index 0000000000..a51e09e80d --- /dev/null +++ b/dataLayer/restoreMarker.ts @@ -0,0 +1,127 @@ +'use strict'; + +import { closeSync, existsSync, fsyncSync, openSync, unlinkSync, writeSync } from 'node:fs'; +import { dirname } from 'node:path'; +import { tryFileLock, fileLockRelease } from '@harperfast/rocksdb-js'; + +/** + * Restore lock + marker protocol for RocksDB database restores (online operation and offline CLI). + * + * Two files live *next to* the database directory (never inside it, since a restore purges the + * destination): + * + * - `.restore.lock` — an OS-level exclusive file lock (via rocksdb-js `tryFileLock`), + * effective across processes, containers, and worker threads, auto-released on process exit. + * Only *held-ness* is meaningful; the file itself persists after release (harmless). + * Known limitation: the lock is owned by the process, so if the restore job's worker *thread* + * dies without the process exiting, the lock stays held (restores 409) until Harper restarts. + * - `.restoring` — the completion marker. Written (and fsynced) after the lock is + * acquired and before the destructive restore begins; deleted only after the restore completes + * successfully, while still holding the lock. Its *existence* means "a restore started and has + * not finished successfully". + */ + +export const RESTORE_LOCK_SUFFIX = '.restore.lock'; +export const RESTORING_MARKER_SUFFIX = '.restoring'; + +export function restoreLockPath(dbPath: string): string { + return dbPath + RESTORE_LOCK_SUFFIX; +} + +export function restoringMarkerPath(dbPath: string): string { + return dbPath + RESTORING_MARKER_SUFFIX; +} + +export type RestoreState = 'in-progress' | 'incomplete' | 'clear'; + +/** + * Determine the restore state of a database directory. Used by startup database detection and + * the open-database guards: + * - 'in-progress': marker present and the restore lock is held (a restore is running in some + * process) — do not load. + * - 'incomplete': marker present but the lock is free (crashed mid-restore; the directory may + * be partial garbage) — do not load; rerun the restore. + * - 'clear': no marker — load normally (a stale, unheld lock file alone is fine). + * + * The marker is checked FIRST and the lock is only probed when the marker exists. Probing takes + * and releases the flock, and probes are mutually exclusive across threads — if every rescan on + * every thread probed the (persistent) lock file of a long-ago-restored database, concurrent + * rescans would collide and misclassify healthy databases as 'in-progress'. Marker-first is + * safe: `beginRestore` writes (and fsyncs) the marker immediately after taking the lock and + * before any destructive step, so a database without a marker has nothing to protect yet. + */ +export function checkRestoreState(dbPath: string): RestoreState { + if (!existsSync(restoringMarkerPath(dbPath))) return 'clear'; + const lockPath = restoreLockPath(dbPath); + if (existsSync(lockPath)) { + const token = tryFileLock(lockPath); + if (token === 0) return 'in-progress'; + fileLockRelease(token); + } + return 'incomplete'; +} + +/** + * Acquire the per-database restore lock and write the restoring marker. Call before any + * destructive step. Returns the lock token for `completeRestore`/`abandonRestore`. + * Throws (statusCode 409) if another restore already holds the lock. + */ +export function beginRestore(dbPath: string): number { + const token = tryFileLock(restoreLockPath(dbPath)); + if (token === 0) { + const error: any = new Error(`Restore already in progress for database at ${dbPath}`); + error.statusCode = 409; + throw error; + } + try { + const fd = openSync(restoringMarkerPath(dbPath), 'w'); + try { + writeSync(fd, `restore started ${new Date().toISOString()}\n`); + fsyncSync(fd); + } finally { + closeSync(fd); + } + // fsync the parent directory so the marker's directory entry is durable — without this a + // power loss can lose the entry, and a half-purged database would load as healthy + const dirFd = openSync(dirname(dbPath), 'r'); + try { + fsyncSync(dirFd); + } finally { + closeSync(dirFd); + } + } catch (error) { + fileLockRelease(token); + throw error; + } + return token; +} + +/** + * Mark the restore successful: delete the marker (while still holding the lock), then release + * the lock. + */ +export function completeRestore(dbPath: string, token: number): void { + try { + unlinkSync(restoringMarkerPath(dbPath)); + // fsync the parent directory so the marker's *removal* is durable — symmetric with the + // creation fsync in beginRestore. Without it, a power loss could resurrect the marker's + // directory entry and misclassify a fully-restored database as incomplete (blocking its + // load until a needless rerun). + const dirFd = openSync(dirname(dbPath), 'r'); + try { + fsyncSync(dirFd); + } finally { + closeSync(dirFd); + } + } finally { + fileLockRelease(token); + } +} + +/** + * Release the lock after a failed restore, leaving the marker in place so the database is + * detected as an incomplete restore (and not loaded) until a rerun succeeds. + */ +export function abandonRestore(token: number): void { + fileLockRelease(token); +} diff --git a/dataLayer/rocksdbBackup.ts b/dataLayer/rocksdbBackup.ts new file mode 100644 index 0000000000..8c8f2a698f --- /dev/null +++ b/dataLayer/rocksdbBackup.ts @@ -0,0 +1,567 @@ +'use strict'; + +import { existsSync, readdirSync } from 'node:fs'; +import { join, resolve } from 'node:path'; +import { PassThrough, Writable } from 'node:stream'; +import { setTimeout as delay } from 'node:timers/promises'; +import { RocksDatabase, backups, registryStatus, type BackupInfo } from '@harperfast/rocksdb-js'; +import { getDatabases, resolveDatabasePath } from '../resources/databases.ts'; +import { getHdbBasePath } from '../utility/environment/environmentManager.ts'; +import { getConfigPath } from '../config/configUtils.ts'; +import { getBackupDirPath } from '../config/configHelpers.ts'; +import { CONFIG_PARAMS, OPERATIONS_ENUM } from '../utility/hdbTerms.ts'; +import { ClientError } from '../utility/errors/hdbError.ts'; +import * as signalling from '../utility/signalling.ts'; +import { SchemaEventMsg } from '../server/threads/itc.js'; +import { beginRestore, completeRestore, abandonRestore, checkRestoreState } from './restoreMarker.ts'; +import logger from '../utility/logging/harper_logger.ts'; + +/** + * Shared core for the RocksDB managed-backup operations (`create_backup`, `list_backups`, + * `verify_backup`, `delete_backup`, `purge_backups`, `restore_backup`) and the RocksDB path of + * `get_backup`. Used by both the operation API (running server) and the CLI (stopped server) so + * the two behave identically. + * + * Directory backups are confined to `//` where the backups root comes + * from the `storage.backupPath` config (default `/backups`); operations never accept + * arbitrary filesystem paths. + */ + +export class BackupNotFoundError extends ClientError { + constructor(message: string) { + super(message, 404); + this.name = 'BackupNotFoundError'; + } +} + +export class BackupInProgressError extends ClientError { + constructor(message: string) { + super(message, 409); + this.name = 'BackupInProgressError'; + } +} + +export function getBackupsRoot(): string { + const configured = getConfigPath(CONFIG_PARAMS.STORAGE_BACKUPPATH); + if (configured && typeof configured === 'string') return configured; + // same /backup directory as config-file backups; databases get subdirectories + return getBackupDirPath(getHdbBasePath()); +} + +function getDatabaseName(request: any): string { + const databaseName = request.database || request.schema || 'data'; + validateDatabaseName(databaseName); + return databaseName; +} + +/** + * The database name becomes a path segment under the backups root and the databases root — + * reject anything that could traverse outside them. + */ +export function validateDatabaseName(databaseName: any): void { + if (typeof databaseName !== 'string' || databaseName.length === 0) { + throw new ClientError(`'database' must be a non-empty string`); + } + if ( + databaseName.includes('/') || + databaseName.includes('\\') || + databaseName.includes('\0') || + databaseName === '.' || + databaseName === '..' + ) { + throw new ClientError(`Invalid database name '${databaseName}'`); + } +} + +export function backupDirForDatabase(databaseName: string): string { + validateDatabaseName(databaseName); + return join(getBackupsRoot(), databaseName); +} + +/** + * Resolve the single root store for a database. A database can span multiple root stores when a + * table has a per-table `path` config; backing up such a database is not supported and errors + * descriptively. Engine gating (RocksDB vs LMDB) is done inline at each call site. + */ +export function resolveSingleRootStore(databaseName: string): any { + const database = getDatabases()[databaseName]; + if (!database) { + throw new BackupNotFoundError(`Database '${databaseName}' does not exist`); + } + const rootStores = new Set(); + for (const tableName in database) { + const rootStore = database[tableName]?.primaryStore?.rootStore; + if (rootStore) rootStores.add(rootStore); + } + if (rootStores.size > 1) { + throw new ClientError( + `Database '${databaseName}' spans multiple root stores (tables with a per-table 'path' config); backup operations only support single-root databases` + ); + } + if (rootStores.size === 0) { + throw new ClientError(`Database '${databaseName}' has no tables to back up`); + } + return rootStores.values().next().value; +} + +function requireRocksRootStore(databaseName: string, operation: string): RocksDatabase { + const rootStore = resolveSingleRootStore(databaseName); + if (!(rootStore instanceof RocksDatabase)) { + throw new ClientError( + `Operation '${operation}' requires a RocksDB database; '${databaseName}' uses the LMDB storage engine (use 'get_backup' to download an LMDB backup)` + ); + } + return rootStore as RocksDatabase; +} + +function requireBackupId(backupId: any): number { + if (!Number.isSafeInteger(backupId) || backupId <= 0) { + throw new ClientError(`'backup_id' must be a positive integer`); + } + return backupId; +} + +function requireBooleanOption(value: any, name: string): boolean { + if (value !== undefined && typeof value !== 'boolean') { + throw new ClientError(`'${name}' must be a boolean`); + } + return value === true; +} + +/** + * The binding serializes backup-directory writers with an on-disk `.backup.lock`; a concurrent + * writer rejects with a "locked" error. Map it to a descriptive 409 — fail fast, no queueing. + */ +function mapLockedError(error: any, databaseName: string): any { + if (typeof error?.message === 'string' && error.message.includes('is locked')) { + return new BackupInProgressError(`Backup operation already in progress for database '${databaseName}'`); + } + return error; +} + +// --- directory helpers (operate on a backup directory only; no open database, usable offline) --- + +export async function listBackupsInDir(backupDir: string): Promise { + // the backup dir doesn't exist until the first create_backup + if (!existsSync(backupDir)) return []; + return backups.list(backupDir); +} + +async function findBackup(backupDir: string, backupId: number, databaseName: string): Promise { + requireBackupId(backupId); // every id-taking path flows through here, including the offline CLI + const list = await listBackupsInDir(backupDir); + const info = list.find((backup) => backup.backupId === backupId); + if (!info) { + throw new BackupNotFoundError(`Backup ${backupId} not found for database '${databaseName}'`); + } + return info; +} + +// --- synchronous operations --- + +export async function listBackups(request: any) { + const databaseName = getDatabaseName(request); + logger.info(`Listing backups for database '${databaseName}'`); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.LIST_BACKUPS); + return listBackupsInDir(backupDirForDatabase(databaseName)); +} + +export async function deleteBackup(request: any) { + const databaseName = getDatabaseName(request); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.DELETE_BACKUP); + const backupId = requireBackupId(request.backup_id); + const backupDir = backupDirForDatabase(databaseName); + await findBackup(backupDir, backupId, databaseName); + try { + await backups.delete(backupDir, backupId); + } catch (error) { + throw mapLockedError(error, databaseName); + } + return { ok: true }; +} + +export async function purgeBackups(request: any) { + const databaseName = getDatabaseName(request); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.PURGE_BACKUPS); + const keepCount = request.keep_count; + if (!Number.isSafeInteger(keepCount) || keepCount < 0) { + throw new ClientError(`'keep_count' must be a non-negative integer`); + } + const backupDir = backupDirForDatabase(databaseName); + const before = await listBackupsInDir(backupDir); + if (before.length === 0) { + throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); + } + try { + await backups.purge(backupDir, keepCount); + } catch (error) { + throw mapLockedError(error, databaseName); + } + const remaining = (await listBackupsInDir(backupDir)).length; + // clamp: a concurrent create between the two lists can otherwise make this negative + return { deleted: Math.max(0, before.length - remaining), remaining }; +} + +// --- job operations: create_backup / verify_backup / restore_backup --- +// Each has a synchronous-validation function (run by jobs.addJob before the job record is +// created) and the job function itself (run in the job worker thread). + +export async function validateCreateBackup(request: any) { + const databaseName = getDatabaseName(request); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.CREATE_BACKUP); +} + +export async function createBackup(request: any) { + const databaseName = getDatabaseName(request); + const rootStore = requireRocksRootStore(databaseName, OPERATIONS_ENUM.CREATE_BACKUP); + const backupDir = backupDirForDatabase(databaseName); + let backupId; + try { + backupId = await rootStore.backup(backupDir, { transactionLogs: true }); + } catch (error) { + throw mapLockedError(error, databaseName); + } + return { database: databaseName, backupId, ...(await describeBackup(backupDir, backupId)) }; +} + +/** + * db.backup() returns only the id; size/timestamp come from a list() match. A missing match + * (e.g. a concurrent delete/purge between the two calls) is logged rather than silently + * reported as undefined fields. + */ +async function describeBackup(backupDir: string, backupId: number): Promise<{ size?: number; timestamp?: any }> { + const info = (await listBackupsInDir(backupDir)).find((backup) => backup.backupId === backupId); + if (!info) { + logger.warn(`Backup ${backupId} was created but is no longer listed in ${backupDir} (deleted concurrently?)`); + return {}; + } + return { size: info.size, timestamp: info.timestamp }; +} + +export async function validateVerifyBackup(request: any) { + const databaseName = getDatabaseName(request); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.VERIFY_BACKUP); + requireBooleanOption(request.verify_checksum, 'verify_checksum'); + await findBackup(backupDirForDatabase(databaseName), requireBackupId(request.backup_id), databaseName); +} + +export async function verifyBackup(request: any) { + const databaseName = getDatabaseName(request); + requireRocksRootStore(databaseName, OPERATIONS_ENUM.VERIFY_BACKUP); + const backupId = requireBackupId(request.backup_id); + const verifyWithChecksum = requireBooleanOption(request.verify_checksum, 'verify_checksum'); + const backupDir = backupDirForDatabase(databaseName); + await findBackup(backupDir, backupId, databaseName); + await backups.verify(backupDir, backupId, { verifyWithChecksum }); + return { database: databaseName, backupId, ok: true }; +} + +export async function validateRestoreBackup(request: any) { + const databaseName = getDatabaseName(request); + if (databaseName === 'system') { + throw new ClientError( + `The 'system' database cannot be restored while Harper is running; stop the server and run: harper restore_backup database=system` + ); + } + if (request.target_database !== undefined) { + // silently ignoring this would destructively restore over the source database instead of + // the copy the caller asked for + throw new ClientError( + `'target_database' is not supported while Harper is running (restore_backup always restores in place); stop the server and run: harper restore_backup database=${databaseName} target_database=` + ); + } + const backupDir = backupDirForDatabase(databaseName); + if (request.backup_id !== undefined) { + await findBackup(backupDir, requireBackupId(request.backup_id), databaseName); + } else if ((await listBackupsInDir(backupDir)).length === 0) { + throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); + } + // The database is normally loaded and RocksDB; a database left unloaded by an interrupted + // restore is also restorable (rerunning the restore is how it is recovered). + const databaseDir = resolveDatabasePath(databaseName); + if (checkRestoreState(databaseDir) === 'clear' || getDatabases()[databaseName]) { + requireRocksRootStore(databaseName, OPERATIONS_ENUM.RESTORE_BACKUP); + } +} + +/** + * Online restore of a user database (see the design's restore lock + marker protocol): + * take the per-database restore lock, write the restoring marker, close the database across all + * worker threads, restore, delete the marker, release the lock, and reload everywhere. + */ +export async function restoreBackup(request: any) { + const databaseName = getDatabaseName(request); + if (databaseName === 'system') { + throw new ClientError( + `The 'system' database cannot be restored while Harper is running; stop the server and run: harper restore_backup database=system` + ); + } + if (request.target_database !== undefined) { + throw new ClientError( + `'target_database' is not supported while Harper is running (restore_backup always restores in place); stop the server and run: harper restore_backup database=${databaseName} target_database=` + ); + } + const backupDir = backupDirForDatabase(databaseName); + const available = await listBackupsInDir(backupDir); + if (available.length === 0) { + throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); + } + let backupId; + if (request.backup_id !== undefined) { + backupId = requireBackupId(request.backup_id); + await findBackup(backupDir, backupId, databaseName); + } else { + backupId = available.reduce((latest, backup) => Math.max(latest, backup.backupId), 0); + } + // a loaded database's root store knows its real directory (which can differ from the + // computed default, e.g. legacy layouts); fall back to the computed path only when the + // database is unloaded (recovering an interrupted restore) + const databaseDir = getDatabases()[databaseName] + ? requireRocksRootStore(databaseName, OPERATIONS_ENUM.RESTORE_BACKUP).path + : resolveDatabasePath(databaseName); + const lockToken = beginRestoreForDatabase(databaseDir, databaseName); + let destructionStarted = false; + try { + // close the database across all worker threads (each thread also rescans, and the + // restoring marker keeps the scan from reloading it mid-restore) + await signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName)); + // A live component (or the system database) can hold its own handle on the database that + // Harper does not track and cannot close, so verify actual process-wide closure before + // purging — restoring under an open instance would corrupt it. If handles remain, fail + // with a clear pointer to the offline CLI path rather than purging. + await verifyDatabaseClosed(databaseDir, databaseName); + destructionStarted = true; + await backups.restore(backupDir, databaseDir, { backupId, mode: 'purgeAllFiles' }); + } catch (error) { + if (destructionStarted) { + // leave the marker: startup/rescan detection reports the incomplete restore until a rerun succeeds + abandonRestore(lockToken); + error.message = `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`; + } else { + // nothing destructive happened — clear the marker and let every thread reload the intact database + completeRestore(databaseDir, lockToken); + await signalling.signalSchemaChange( + new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName) + ); + } + throw error; + } + completeRestore(databaseDir, lockToken); + // signal again: with the marker gone, every thread's rescan reloads the restored database + await signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName)); + return { database: databaseName, backupId }; +} + +// After the close broadcast is acknowledged, every worker thread has released its Harper-managed +// handles; a short grace period covers a just-finished job worker still draining its own close. +// Anything still open past that is a handle Harper neither tracks nor controls (a loaded component +// holding its own instance, or the system database), which will never close on its own — so fail +// fast rather than waiting out a long timeout. +const DATABASE_CLOSE_WAIT_MS = 3000; +const DATABASE_CLOSE_POLL_INTERVAL_MS = 250; + +/** + * Verify no thread in this process still has the database open (rocksdb-js's registry is + * process-global across worker threads), polling briefly to let a just-finished job worker's own + * close drain. Throws 409 with an actionable message if handles remain — which means a loaded + * component is holding the database open (Harper can neither detect which component nor force its + * handle closed), so an online in-place restore is not possible and the offline CLI is the path. + */ +async function verifyDatabaseClosed(databaseDir: string, databaseName: string): Promise { + const targetPath = resolve(databaseDir); + const deadline = Date.now() + DATABASE_CLOSE_WAIT_MS; + for (;;) { + const stillOpen = registryStatus().some( + (instance) => resolve(instance.path) === targetPath && instance.refCount > 0 + ); + if (!stillOpen) return; + if (Date.now() >= deadline) { + throw new BackupInProgressError( + `Cannot restore database '${databaseName}' while Harper is running: it is held open by a loaded component (or is the system database). ` + + `Restore it offline instead — stop the server and run: harper restore_backup database=${databaseName}` + + (databaseName === 'system' ? '' : ` backup_id=`) + ); + } + await delay(DATABASE_CLOSE_POLL_INTERVAL_MS); + } +} + +/** + * beginRestore's own error message carries the filesystem path (useful in CLI/server logs); + * client-facing operations report by database name instead. + */ +function beginRestoreForDatabase(databaseDir: string, databaseName: string): number { + try { + return beginRestore(databaseDir); + } catch (error) { + if (error.statusCode === 409) { + throw new BackupInProgressError(`Restore already in progress for database '${databaseName}'`); + } + throw error; + } +} + +// --- get_backup (RocksDB path): stream a fresh full-snapshot tar in the HTTP response --- + +/** + * Returns a Readable (with `.headers`) streaming a full-snapshot tar (optionally gzipped) of the + * database's current state. No scratch disk; a consumer error aborts the native backup cleanly. + * `noCompression` opts out of serverHandlers' accept-encoding auto-gzip — the binding handles + * gzip when requested, and this response must never be compressed by the server. + */ +export function createBackupStream(rootStore: RocksDatabase, databaseName: string, gzip: boolean): PassThrough { + const stream: any = new PassThrough(); + // database names may legally contain `"` and `\` (schemaRegex) — sanitize so the quoted + // content-disposition filename stays parseable + const filename = `${databaseName.replace(/["\\]/g, '_')}.tar${gzip ? '.gz' : ''}`; + stream.headers = new Map([ + ['content-type', gzip ? 'application/gzip' : 'application/x-tar'], + ['content-disposition', `attachment; filename="${filename}"`], + ]); + stream.noCompression = true; + rootStore + .backup(Writable.toWeb(stream) as any, { gzip, transactionLogs: true }) + .catch((error) => stream.destroy(error)); + return stream; +} + +// --- offline CLI paths (server stopped) --- + +/** + * Offline create: open the RocksDatabase directly, run an ordinary incremental directory backup + * into the configured backup root, and close. RocksDB is single-writer, so this collides on the + * database lock if the server is running — callers guard on the server being stopped. + */ +export async function createBackupOffline(databaseName: string) { + validateDatabaseName(databaseName); + const databaseDir = resolveDatabasePath(databaseName); + if (!existsSync(join(databaseDir, 'CURRENT'))) { + throw new BackupNotFoundError(`No RocksDB database found at ${databaseDir}`); + } + const restoreState = checkRestoreState(databaseDir); + if (restoreState !== 'clear') { + throw new BackupInProgressError( + `Database '${databaseName}' has an ${restoreState === 'in-progress' ? 'active' : 'incomplete'} restore; rerun restore_backup before backing up` + ); + } + const database = RocksDatabase.open(databaseDir); + try { + const backupDir = backupDirForDatabase(databaseName); + let backupId; + try { + backupId = await database.backup(backupDir, { transactionLogs: true }); + } catch (error) { + throw mapLockedError(error, databaseName); + } + return { database: databaseName, backupId, ...(await describeBackup(backupDir, backupId)) }; + } finally { + database.close(); + } +} + +/** + * Offline restore (required for the `system` database; works for any database). Runs the same + * lock + marker protocol as the online operation so a crashed CLI restore is detected at next + * server start. `targetDatabase` restores into a different database directory (non-destructive + * for the source database); the server picks it up on next start via normal engine detection. + */ +export async function restoreBackupOffline(databaseName: string, backupId?: number, targetDatabase?: string) { + validateDatabaseName(databaseName); + const backupDir = backupDirForDatabase(databaseName); + const available = await listBackupsInDir(backupDir); + if (available.length === 0) { + throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); + } + if (backupId !== undefined) { + await findBackup(backupDir, backupId, databaseName); + } else { + backupId = available.reduce((latest, backup) => Math.max(latest, backup.backupId), 0); + } + if (targetDatabase !== undefined) validateDatabaseName(targetDatabase); + const databaseDir = resolveDatabasePath(targetDatabase ?? databaseName); + if (targetDatabase !== undefined && targetDatabase !== databaseName && !isMissingOrEmptyDir(databaseDir)) { + // target_database is documented as non-destructive: never purge an existing database of + // that name out from under the operator + throw new ClientError( + `target_database '${targetDatabase}' already exists at ${databaseDir}; restoring into it would destroy it — choose a new name, or restore in place by omitting target_database` + ); + } + // The offline path is entered only when the CLI sees no running server (getHdbPid), but that is + // a heuristic: the PID file is briefly absent mid-`harper restart`, and backups.restore's + // purgeAllFiles never takes RocksDB's own lock. Probe that lock by opening the database — a live + // holder makes open throw "is locked" — so we refuse rather than purge a database another + // process still has open. A directory that fails to open for any *other* reason (corrupt or + // half-restored) is exactly what restore recovers, so only a lock conflict aborts. + if (existsSync(join(databaseDir, 'CURRENT'))) { + try { + RocksDatabase.open(databaseDir).close(); + } catch (error: any) { + if (typeof error?.message === 'string' && error.message.includes('is locked')) { + throw new BackupInProgressError( + `Cannot restore database '${databaseName}': it is open by a running Harper process — stop Harper before restoring offline` + ); + } + } + } + const lockToken = beginRestoreForDatabase(databaseDir, targetDatabase ?? databaseName); + try { + await backups.restore(backupDir, databaseDir, { backupId, mode: 'purgeAllFiles' }); + } catch (error) { + abandonRestore(lockToken); + error.message = `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`; + throw error; + } + completeRestore(databaseDir, lockToken); + return { database: databaseName, backupId, restoredTo: databaseDir }; +} + +function isMissingOrEmptyDir(path: string): boolean { + try { + return readdirSync(path).length === 0; + } catch (error) { + if (error.code === 'ENOENT') return true; + throw error; + } +} + +// --- offline management wrappers (no engine validation: they operate on the directory only) --- + +export async function verifyBackupOffline(databaseName: string, backupId: number, verifyChecksum?: boolean) { + validateDatabaseName(databaseName); + const verifyWithChecksum = requireBooleanOption(verifyChecksum, 'verify_checksum'); + const backupDir = backupDirForDatabase(databaseName); + await findBackup(backupDir, backupId, databaseName); + await backups.verify(backupDir, backupId, { verifyWithChecksum }); + return { database: databaseName, backupId, ok: true }; +} + +export async function deleteBackupOffline(databaseName: string, backupId: number) { + validateDatabaseName(databaseName); + const backupDir = backupDirForDatabase(databaseName); + await findBackup(backupDir, backupId, databaseName); + try { + await backups.delete(backupDir, backupId); + } catch (error) { + throw mapLockedError(error, databaseName); + } + return { ok: true }; +} + +export async function purgeBackupsOffline(databaseName: string, keepCount: number) { + validateDatabaseName(databaseName); + if (!Number.isSafeInteger(keepCount) || keepCount < 0) { + throw new ClientError(`'keep_count' must be a non-negative integer`); + } + const backupDir = backupDirForDatabase(databaseName); + const before = await listBackupsInDir(backupDir); + if (before.length === 0) { + throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); + } + try { + await backups.purge(backupDir, keepCount); + } catch (error) { + throw mapLockedError(error, databaseName); + } + const remaining = (await listBackupsInDir(backupDir)).length; + return { deleted: Math.max(0, before.length - remaining), remaining }; +} diff --git a/resources/databases.ts b/resources/databases.ts index 8824498f74..b7e23e6a5c 100644 --- a/resources/databases.ts +++ b/resources/databases.ts @@ -34,6 +34,7 @@ import { RocksIndexStore } from './RocksIndexStore.ts'; import { when } from '../utility/when.ts'; import { resolveRocksMemoryConfig } from '../utility/rocksMemoryConfig.ts'; import { isProcessRunning } from '../utility/processManagement/processManagement.js'; +import { checkRestoreState, RESTORE_LOCK_SUFFIX, RESTORING_MARKER_SUFFIX } from '../dataLayer/restoreMarker.ts'; /** * Check if Harper is running in read-only mode. @@ -259,9 +260,12 @@ export function getDatabases(): Databases { if (databasePath && existsSync(databasePath)) { // First load all the databases from our main database folder // TODO: Load any databases defined with explicit storage paths from the config - for (const databaseEntry of readdirSync(databasePath, { withFileTypes: true })) { + const entries = readdirSync(databasePath, { withFileTypes: true }); + const blockedByRestore = databasesBlockedByRestore(databasePath, entries); + for (const databaseEntry of entries) { const dbName = basename(databaseEntry.name, '.mdb'); const dbPath = join(databasePath, databaseEntry.name); + if (blockedByRestore.has(dbName)) continue; if ( databaseEntry.isFile() && @@ -318,7 +322,10 @@ export function getDatabases(): Databases { const schemaConfig = schemaConfigs[dbName]; const databasePath = schemaConfig.path; if (existsSync(databasePath)) { - for (const databaseEntry of readdirSync(databasePath, { withFileTypes: true })) { + const entries = readdirSync(databasePath, { withFileTypes: true }); + const blockedByRestore = databasesBlockedByRestore(databasePath, entries); + for (const databaseEntry of entries) { + if (blockedByRestore.has(basename(databaseEntry.name, '.mdb'))) continue; if (databaseEntry.isFile() && extname(databaseEntry.name).toLowerCase() === '.mdb') { readMetaDb(join(databasePath, databaseEntry.name), basename(databaseEntry.name, '.mdb'), dbName); } else { @@ -393,6 +400,35 @@ export function getDatabases(): Databases { return databases; } +/** + * Scan a databases directory's entries for restore lock/marker files and return the names of + * databases that must not be loaded: a held restore lock means a restore is in progress in some + * process; an unheld lock with a surviving `.restoring` marker means a restore was interrupted + * mid-purge (the directory may be partial garbage) and must be rerun. The files live *next to* + * the database directory, so this also covers a database whose directory is missing or empty. + */ +function databasesBlockedByRestore(databasePath: string, entries: { name: string }[]): Set { + const blocked = new Set(); + const candidates = new Set(); + for (const { name } of entries) { + if (name.endsWith(RESTORING_MARKER_SUFFIX)) candidates.add(name.slice(0, -RESTORING_MARKER_SUFFIX.length)); + else if (name.endsWith(RESTORE_LOCK_SUFFIX)) candidates.add(name.slice(0, -RESTORE_LOCK_SUFFIX.length)); + } + for (const dbName of candidates) { + const state = checkRestoreState(join(databasePath, dbName)); + if (state === 'in-progress') { + logger.warn(`A restore of database '${dbName}' is in progress; not loading it`); + blocked.add(dbName); + } else if (state === 'incomplete') { + logger.error( + `Incomplete restore of database '${dbName}' detected (a restore started but did not finish); not loading it — rerun the restore to recover` + ); + blocked.add(dbName); + } + } + return blocked; +} + /** * This is responsible for reading the internal dbi of a single database file to get a list of all the tables and * their indexed or registered attributes @@ -820,18 +856,12 @@ function setTable(tables, tableName, Table) { return Table; } /** - * Get root store for a database - * @param options - * @returns + * Resolve the directory that holds (or would hold) a database's storage, from the databases + * config, storage path config/env, or the hdb root — without opening anything. This is the + * parent directory selection used by `database()`; a RocksDB database lives at + * `join(resolveDatabaseStorageRoot(...), databaseName)`. */ -export function database({ database: databaseName, table: tableName }) { - if (!databaseName) databaseName = DEFAULT_DATABASE_NAME; - getDatabases(); - ensureDB(databaseName); - const definedDatabase = definedDatabases.get(databaseName); - if ((definedDatabase as any)?.rootStore) { - return (definedDatabase as any).rootStore; - } +export function resolveDatabaseStorageRoot(databaseName: string, tableName?: string): string { const databaseConfig = envGet(CONFIG_PARAMS.DATABASES) || {}; if (process.env.SCHEMAS_DATA_PATH) { databaseConfig.data = { path: process.env.SCHEMAS_DATA_PATH }; @@ -856,6 +886,35 @@ export function database({ database: databaseName, table: tableName }) { `Unable to determine database storage path. Ensure STORAGE_PATH, HDB_ROOT, or a valid config path is set.` ); } + return databasePath; +} + +/** + * Resolve the directory path of a RocksDB database (whether or not it exists or is loaded). + */ +export function resolveDatabasePath(databaseName: string): string { + return join(resolveDatabaseStorageRoot(databaseName), databaseName); +} + +/** + * Get root store for a database + * @param options + * @returns + */ +export function database({ database: databaseName, table: tableName }) { + if (!databaseName) databaseName = DEFAULT_DATABASE_NAME; + getDatabases(); + ensureDB(databaseName); + const definedDatabase = definedDatabases.get(databaseName); + if ((definedDatabase as any)?.rootStore) { + return (definedDatabase as any).rootStore; + } + const databaseConfig = envGet(CONFIG_PARAMS.DATABASES) || {}; + if (process.env.SCHEMAS_DATA_PATH) { + databaseConfig.data = { path: process.env.SCHEMAS_DATA_PATH }; + } + const tablePath = tableName && databaseConfig[databaseName]?.tables?.[tableName]?.path; + const databasePath = resolveDatabaseStorageRoot(databaseName, tableName); let rootStore: RootDatabaseKind; const useRocksdb = (process.env.HARPER_STORAGE_ENGINE || envGet(CONFIG_PARAMS.STORAGE_ENGINE)) !== 'lmdb'; @@ -863,6 +922,10 @@ export function database({ database: databaseName, table: tableName }) { const path = join(databasePath, tablePath ? tableName : databaseName); rootStore = rocksdbDatabaseEnvs.get(path); if (!rootStore || rootStore.status === 'closed') { + // this on-demand open (create_table/create_database and friends) must not resurrect a + // database that a restore is rewriting (or left half-purged) — the scan-time restore + // checks don't cover this path + throwIfBlockedByRestore(path, databaseName); rootStore = openRocksDatabase(path, { disableWAL: false, enableStats: true, @@ -885,6 +948,19 @@ export function database({ database: databaseName, table: tableName }) { if (definedDatabase) (definedDatabase as any).rootStore = rootStore; return rootStore; } +function throwIfBlockedByRestore(dbPath: string, databaseName: string): void { + const restoreState = checkRestoreState(dbPath); + if (restoreState !== 'clear') { + const error: any = new Error( + restoreState === 'in-progress' + ? `Database '${databaseName}' is being restored; retry when the restore completes` + : `Database '${databaseName}' has an incomplete restore; rerun restore_backup to recover it` + ); + error.statusCode = 409; + throw error; + } +} + /** * Delete the database * @param databaseName @@ -897,6 +973,9 @@ export async function dropDatabase(databaseName) { for (const tableName in dbTables) { const table = dbTables[tableName]; rootStore = table.primaryStore.rootStore; + // a drop's file deletion must not interleave with a restore's purge-and-copy on the same + // directory (destroy landing after the restore's copy would gut a "successful" restore) + if (rootStore instanceof RocksDatabase) throwIfBlockedByRestore(rootStore.path, databaseName); lmdbDatabaseEnvs.delete(rootStore.path); rocksdbDatabaseEnvs.delete(rootStore.path); } @@ -938,6 +1017,90 @@ export async function dropDatabase(databaseName) { await deleteRootBlobPathsForDB(rootStore); } + +/** + * Close a RocksDB database's store handles on the current thread and unregister it, without + * touching its files. Used by the restore_backup flow: every thread must release its handles so + * `backups.restore()` can purge and rewrite the (fully closed) database directory. A subsequent + * `resetDatabases()`/`getDatabases()` rescan reloads it (or skips it while a restore is in + * progress, per the restore marker checks in the scan). + */ +export function closeDatabase(databaseName: string): boolean { + const dbTables = databases[databaseName]; + if (!dbTables) return false; + const rootStores = new Set(); + const closeStore = (store: any, description: string) => { + try { + store?.close?.(); + } catch (error) { + logger.warn(`Error closing ${description} while closing database ${databaseName}:`, error); + } + }; + for (const tableName in dbTables) { + const table: any = dbTables[tableName]; + if (!table?.primaryStore) continue; + if (table.primaryStore.rootStore) rootStores.add(table.primaryStore.rootStore); + for (const indexName in table.indices || {}) { + closeStore(table.indices[indexName], `index ${tableName}.${indexName}`); + } + closeStore(table.primaryStore, `table ${tableName}`); + } + // a database with no tables (an empty schema, or one whose tables were all dropped) still holds + // an open root store, tracked only on the defined-database entry rather than any table — include + // it so its handles are released too (the Set dedupes it against the per-table root stores above) + const definedRoot = (definedDatabases?.get(databaseName) as any)?.rootStore; + if (definedRoot) rootStores.add(definedRoot); + for (const rootStore of rootStores) { + closeStore(rootStore.dbisDb, 'attributes store'); + closeStore(rootStore, 'root store'); + lmdbDatabaseEnvs.delete(rootStore.path); + rocksdbDatabaseEnvs.delete(rootStore.path); + } + const definedDatabase = definedDatabases?.get(databaseName); + if (definedDatabase) (definedDatabase as any).rootStore = undefined; + if (databaseName === 'data') { + for (const tableName in tables) { + delete tables[tableName]; + } + delete tables[DEFINED_TABLES]; + } + delete databases[databaseName]; + return true; +} + +/** + * Close every RocksDB (user) database this thread has open, releasing its native handles. + * + * rocksdb-js's registry is process-global across worker threads, and a thread that exits WITHOUT + * closing leaks its handles (the process-global refCount never drops), while the only alternative, + * `shutdown()`, tears down rocksdb for the entire process. So a worker thread that opens databases + * and then exits — notably a job worker (jobProcess), which opens the whole database graph via + * `getDatabases()` and exits when the job finishes — must close its handles explicitly, or those + * handles linger process-wide (and, e.g., block an online `restore_backup` from confirming the + * database is closed). The `system` database is intentionally left open: it is non-enumerable here + * (skipped by the loop), is never restored online, and the exiting worker may still touch the job + * table during teardown. Best-effort: closing failures are swallowed inside `closeDatabase`. + */ +export function closeLoadedDatabases(): void { + // snapshot the names first: closeDatabase() deletes from `databases` as it goes + for (const databaseName of Object.keys(databases)) { + const dbTables = databases[databaseName]; + if (!dbTables) continue; + let isRocks = false; + for (const tableName in dbTables) { + if (dbTables[tableName]?.primaryStore?.rootStore instanceof RocksDatabase) { + isRocks = true; + break; + } + } + // a tableless database exposes no table root store, so also check the defined-database + // entry — otherwise its open root store would leak on worker exit + if (!isRocks && (definedDatabases?.get(databaseName) as any)?.rootStore instanceof RocksDatabase) { + isRocks = true; + } + if (isRocks) closeDatabase(databaseName); + } +} // HNSW_NO_AUTOVERSION kill-switch: when set, a NEW index initializes as legacy rather than // versioned. process.env values are strings, so a bare truthiness check would treat "0"/"false" // as enabling the switch — the opposite of intent. Treat "" / "0" / "false" (and unset) as NOT set. diff --git a/server/itc/serverHandlers.js b/server/itc/serverHandlers.js index 1c97d127ea..9b48ef3266 100644 --- a/server/itc/serverHandlers.js +++ b/server/itc/serverHandlers.js @@ -12,7 +12,7 @@ const harperBridge = require('../../dataLayer/harperBridge/harperBridge.ts'); const process = require('process'); const { isMainThread, workerData } = require('worker_threads'); -const { resetDatabases } = require('../../resources/databases.ts'); +const { resetDatabases, closeDatabase } = require('../../resources/databases.ts'); /** * This object/functions are passed to the ITC client instance and dynamically added as event handlers. @@ -46,6 +46,12 @@ async function schemaHandler(event) { } hdbLogger.trace(`ITC schemaHandler received schema event:`, event); + // restore_backup: this thread must release its store handles so the restore can purge and + // rewrite the database directory. The rescan below (resetDatabases) skips reloading it while + // the restoring marker is present, and reloads it on the completion signal (marker gone). + if (event.message?.operation === hdbTerms.OPERATIONS_ENUM.RESTORE_BACKUP && event.message.schema) { + closeDatabase(event.message.schema); + } await cleanLmdbMap(event.message); await syncSchemaMetadata(event.message); for (let listener of schemaListeners) { diff --git a/server/jobs/jobProcess.ts b/server/jobs/jobProcess.ts index 2526ee1dd9..9688a3c32b 100644 --- a/server/jobs/jobProcess.ts +++ b/server/jobs/jobProcess.ts @@ -77,7 +77,24 @@ const JOB_ID = JOB_NAME.substring(4); jobObj.message = err.message ? err.message : err; jobObj.end_datetime = moment().valueOf(); } finally { - await jobs.updateJob(jobObj); + // A rejected updateJob must not skip handle cleanup and exit scheduling below (that would + // leak this worker's process-global RocksDB handles and leave the worker hanging). + try { + await jobs.updateJob(jobObj); + } catch (updateErr) { + harperLogger.error('Error updating job record on job worker exit:', updateErr); + } + // Release this worker's RocksDB handles before it exits. A job worker opens the whole + // database graph via getDatabases(); rocksdb-js's registry is process-global and a thread + // that exits without closing leaks its handles process-wide, which (among other costs) + // blocks an online restore_backup from confirming the target database is closed. Best + // effort — never let cleanup mask the job result. + try { + const { closeLoadedDatabases } = await import('../../resources/databases.ts'); + closeLoadedDatabases(); + } catch (closeErr) { + harperLogger.warn('Error releasing database handles on job worker exit:', closeErr); + } // On Bun 1.3.13, calling process.exit() in a worker thread with lmdb-js loaded // while sibling workers are running causes a NAPI fatal error crash. Unref // parentPort (which broadcastWithAcknowledgement may have ref'd during schema diff --git a/server/jobs/jobRunner.ts b/server/jobs/jobRunner.ts index 654c632a11..4c1d7c9549 100644 --- a/server/jobs/jobRunner.ts +++ b/server/jobs/jobRunner.ts @@ -10,6 +10,7 @@ import log from '../../utility/logging/harper_logger.ts'; import * as jobs from './jobs.ts'; import * as hdbExport from '../../dataLayer/export.ts'; import * as hdbDelete from '../../dataLayer/delete.ts'; +import * as rocksdbBackup from '../../dataLayer/rocksdbBackup.ts'; import * as threadsStart from '../threads/manageThreads.js'; import * as transactionLog from '../../utility/logging/transactionLog.ts'; import * as restart from '../../bin/restart.ts'; @@ -79,6 +80,15 @@ async function parseMessage(runnerMessage: any) { case hdbTerms.JOB_TYPE_ENUM.delete_transaction_logs_before: await runJob(runnerMessage, transactionLog.deleteTransactionLogsBefore); break; + case hdbTerms.JOB_TYPE_ENUM.create_backup: + await runJob(runnerMessage, rocksdbBackup.createBackup); + break; + case hdbTerms.JOB_TYPE_ENUM.verify_backup: + await runJob(runnerMessage, rocksdbBackup.verifyBackup); + break; + case hdbTerms.JOB_TYPE_ENUM.restore_backup: + await runJob(runnerMessage, rocksdbBackup.restoreBackup); + break; case hdbTerms.JOB_TYPE_ENUM.restart_service: await runJob(runnerMessage, restart.restartService); return `Restarting ${runnerMessage.json.service}`; diff --git a/server/jobs/jobs.ts b/server/jobs/jobs.ts index 4bdfaebd7c..69ae1063a0 100644 --- a/server/jobs/jobs.ts +++ b/server/jobs/jobs.ts @@ -20,6 +20,7 @@ import * as hdbUtil from '../../utility/common_utils.ts'; import { promisify } from 'util'; import moment from 'moment'; import * as fileLoadValidator from '../../validation/fileLoadValidator.ts'; +import * as rocksdbBackup from '../../dataLayer/rocksdbBackup.ts'; import bulkDeleteValidator from '../../validation/bulkDeleteValidator.ts'; import { deleteTransactionLogsBeforeValidator } from '../../validation/transactionLogValidator.ts'; import { handleHDBError, ClientError } from '../../utility/errors/hdbError.ts'; @@ -122,6 +123,16 @@ export async function addJob(jsonBody: any) { throw handleHDBError(new Error(), 'Invalid service', HTTP_STATUS_CODES.BAD_REQUEST, undefined, undefined, true); } break; + // backup job validators throw ClientError (with statusCode) directly on failure + case hdbTerms.OPERATIONS_ENUM.CREATE_BACKUP: + await rocksdbBackup.validateCreateBackup(jsonBody); + break; + case hdbTerms.OPERATIONS_ENUM.VERIFY_BACKUP: + await rocksdbBackup.validateVerifyBackup(jsonBody); + break; + case hdbTerms.OPERATIONS_ENUM.RESTORE_BACKUP: + await rocksdbBackup.validateRestoreBackup(jsonBody); + break; default: break; } diff --git a/server/serverHelpers/serverHandlers.js b/server/serverHelpers/serverHandlers.js index c0f132ba40..ea2f69ab19 100644 --- a/server/serverHelpers/serverHandlers.js +++ b/server/serverHelpers/serverHandlers.js @@ -174,8 +174,10 @@ async function handlePostRequest(req, res, _bypassAuth = false) { res.header(name, value); } // fastify-compress has one job. I don't know why it can't do it. So we compress here to - // handle the case of returning a stream - if (req.headers['accept-encoding']?.includes('gzip')) { + // handle the case of returning a stream. Streams marked `noCompression` opt out (e.g. + // RocksDB get_backup tars: the binding gzips when requested; compressing here would + // mislabel a gzip:false tar or double-compress a gzip:true one). + if (req.headers['accept-encoding']?.includes('gzip') && !result.noCompression) { res.header('content-encoding', 'gzip'); result = result.pipe(createGzip({ level: constants.Z_BEST_SPEED })); // go fast } diff --git a/server/serverHelpers/serverUtilities.ts b/server/serverHelpers/serverUtilities.ts index 02599dbcd6..ec8ede7c00 100644 --- a/server/serverHelpers/serverUtilities.ts +++ b/server/serverHelpers/serverUtilities.ts @@ -10,6 +10,7 @@ import customFunctionOperations from '../../components/operations.js'; import harperLogger from '../../utility/logging/harper_logger.ts'; import readLog from '../../utility/logging/readLog.ts'; import * as export_ from '../../dataLayer/export.ts'; +import * as rocksdbBackup from '../../dataLayer/rocksdbBackup.ts'; import * as opAuth from '../../utility/operation_authorization.ts'; import * as jobs from '../jobs/jobs.ts'; import * as terms from '../../utility/hdbTerms.ts'; @@ -363,6 +364,9 @@ export async function executeJob(json: OperationRequestBody): Promise }; } } catch (err) { + // errors that already carry a statusCode (e.g. ClientError from job validation) are + // client-facing as-is; wrapping them here would turn a 400/404/409 into a 500 + if (err instanceof Error && typeof (err as any).statusCode === 'number') throw err; const error = err instanceof Error ? err : null; const message = `There was an error executing job: ${error && 'http_resp_msg' in error ? error.http_resp_msg : err}`; operationLog.error(message); @@ -548,6 +552,21 @@ function initializeOperationFunctionMap(): Map beginRestore(dbPath), + (error) => error.statusCode === 409 && /already in progress/.test(error.message) + ); + } finally { + completeRestore(dbPath, token); + } + }); + + it('a rerun after an abandoned restore succeeds and clears the marker', function () { + abandonRestore(beginRestore(dbPath)); + assert.strictEqual(checkRestoreState(dbPath), 'incomplete'); + const token = beginRestore(dbPath); + completeRestore(dbPath, token); + assert.strictEqual(checkRestoreState(dbPath), 'clear'); + }); + }); +}); diff --git a/unitTests/dataLayer/rocksdbBackup.test.js b/unitTests/dataLayer/rocksdbBackup.test.js new file mode 100644 index 0000000000..d6ddcddc64 --- /dev/null +++ b/unitTests/dataLayer/rocksdbBackup.test.js @@ -0,0 +1,285 @@ +'use strict'; + +const assert = require('node:assert'); +const { existsSync, mkdtempSync, rmSync } = require('node:fs'); +const { join } = require('node:path'); +const { tmpdir } = require('node:os'); +const { RocksDatabase } = require('@harperfast/rocksdb-js'); +const { + backupDirForDatabase, + createBackupOffline, + deleteBackupOffline, + getBackupsRoot, + listBackupsInDir, + purgeBackupsOffline, + restoreBackup, + restoreBackupOffline, + validateDatabaseName, + validateRestoreBackup, + verifyBackupOffline, + createBackupStream, +} = require('#src/dataLayer/rocksdbBackup'); +const { beginRestore, completeRestore, checkRestoreState } = require('#src/dataLayer/restoreMarker'); + +const DB_NAME = 'rocksdb-backup-unit-test'; + +describe('rocksdbBackup', function () { + let storageDir; + let databaseDir; + let savedStoragePath; + + before(function () { + storageDir = mkdtempSync(join(tmpdir(), 'harper.unit-test.rocksdb-backup-')); + // resolveDatabasePath honors STORAGE_PATH, so databases created here resolve to this dir + savedStoragePath = process.env.STORAGE_PATH; + process.env.STORAGE_PATH = storageDir; + databaseDir = join(storageDir, DB_NAME); + }); + + after(function () { + if (savedStoragePath === undefined) delete process.env.STORAGE_PATH; + else process.env.STORAGE_PATH = savedStoragePath; + rmSync(storageDir, { recursive: true, force: true }); + rmSync(backupDirForDatabase(DB_NAME), { recursive: true, force: true }); + }); + + function writeRecords(records) { + const database = RocksDatabase.open(databaseDir); + try { + for (const [key, value] of records) { + database.putSync(key, value); + } + } finally { + database.close(); + } + } + + describe('validateDatabaseName', function () { + it('accepts ordinary names', function () { + validateDatabaseName('data'); + validateDatabaseName('my_db-2'); + validateDatabaseName('my.db'); + }); + + it('rejects traversal and separator names', function () { + for (const name of ['..', '.', 'a/b', 'a\\b', '', 'a\0b', 42, undefined]) { + assert.throws( + () => validateDatabaseName(name), + (error) => error.statusCode === 400 + ); + } + }); + }); + + describe('backupDirForDatabase', function () { + it('confines the backup directory to the backups root', function () { + assert.strictEqual(backupDirForDatabase(DB_NAME), join(getBackupsRoot(), DB_NAME)); + }); + }); + + describe('listBackupsInDir', function () { + it('returns [] for a directory that does not exist yet', async function () { + assert.deepStrictEqual(await listBackupsInDir(join(storageDir, 'no-such-dir')), []); + }); + }); + + describe('offline backup lifecycle', function () { + it('createBackupOffline errors descriptively when there is no database', async function () { + await assert.rejects(createBackupOffline('no-such-database'), (error) => error.statusCode === 404); + }); + + it('creates, lists, verifies, restores, deletes, and purges backups', async function () { + this.timeout(30000); + writeRecords([ + ['alpha', { n: 1 }], + ['beta', { n: 2 }], + ]); + + const first = await createBackupOffline(DB_NAME); + assert.strictEqual(first.database, DB_NAME); + assert.strictEqual(first.backupId, 1); + assert.ok(first.size > 0, 'size should come from the backups.list match'); + assert.ok(first.timestamp !== undefined); + + writeRecords([['gamma', { n: 3 }]]); + const second = await createBackupOffline(DB_NAME); + assert.strictEqual(second.backupId, 2); + + const backupDir = backupDirForDatabase(DB_NAME); + const listed = await listBackupsInDir(backupDir); + assert.deepStrictEqual( + listed.map((backup) => backup.backupId), + [1, 2] + ); + + const verified = await verifyBackupOffline(DB_NAME, 1, true); + assert.deepStrictEqual(verified, { database: DB_NAME, backupId: 1, ok: true }); + + // restore the first backup into a separate target database directory + const restored = await restoreBackupOffline(DB_NAME, 1, `${DB_NAME}-restored`); + assert.strictEqual(restored.backupId, 1); + const restoredDir = join(storageDir, `${DB_NAME}-restored`); + assert.strictEqual(restored.restoredTo, restoredDir); + assert.strictEqual(checkRestoreState(restoredDir), 'clear'); + const restoredDb = RocksDatabase.open(restoredDir); + try { + assert.deepStrictEqual(restoredDb.getSync('alpha'), { n: 1 }); + assert.strictEqual(restoredDb.getSync('gamma'), undefined, 'backup 1 predates gamma'); + } finally { + restoredDb.close(); + } + + // restore latest (no backup_id) in place over the source database + const latestRestore = await restoreBackupOffline(DB_NAME); + assert.strictEqual(latestRestore.backupId, 2); + assert.strictEqual(checkRestoreState(databaseDir), 'clear'); + const inPlaceDb = RocksDatabase.open(databaseDir); + try { + assert.deepStrictEqual(inPlaceDb.getSync('gamma'), { n: 3 }); + } finally { + inPlaceDb.close(); + } + + const deleted = await deleteBackupOffline(DB_NAME, 1); + assert.deepStrictEqual(deleted, { ok: true }); + assert.deepStrictEqual( + (await listBackupsInDir(backupDir)).map((backup) => backup.backupId), + [2] + ); + + const purged = await purgeBackupsOffline(DB_NAME, 0); + assert.deepStrictEqual(purged, { deleted: 1, remaining: 0 }); + }); + + it('errors with 404 for unknown backup ids and empty repositories', async function () { + await assert.rejects(verifyBackupOffline(DB_NAME, 999, false), (error) => error.statusCode === 404); + await assert.rejects(deleteBackupOffline(DB_NAME, 999), (error) => error.statusCode === 404); + await assert.rejects(purgeBackupsOffline(DB_NAME, 1), (error) => error.statusCode === 404); + await assert.rejects(restoreBackupOffline(DB_NAME, 999), (error) => error.statusCode === 404); + }); + + it('rejects invalid backup ids and keep counts', async function () { + await assert.rejects(deleteBackupOffline(DB_NAME, 'one'), (error) => error.statusCode === 400); + await assert.rejects(purgeBackupsOffline(DB_NAME, -1), (error) => error.statusCode === 400); + }); + + it('rejects a non-boolean verify_checksum instead of silently skipping the checksum pass', async function () { + await assert.rejects( + verifyBackupOffline(DB_NAME, 1, 'true'), + (error) => error.statusCode === 400 && /verify_checksum/.test(error.message) + ); + }); + + it('refuses to restore into an existing, non-empty target_database', async function () { + this.timeout(30000); + writeRecords([['alpha', { n: 1 }]]); + await createBackupOffline(DB_NAME); + // first restore into a fresh target succeeds and leaves a non-empty database there… + await restoreBackupOffline(DB_NAME, undefined, `${DB_NAME}-occupied`); + // …so a second restore into the same target must be rejected, not purge it + await assert.rejects( + restoreBackupOffline(DB_NAME, undefined, `${DB_NAME}-occupied`), + (error) => error.statusCode === 400 && /already exists/.test(error.message) + ); + await purgeBackupsOffline(DB_NAME, 0); + }); + + it('refuses to back up a database with an incomplete restore pending', async function () { + this.timeout(30000); + writeRecords([['alpha', { n: 1 }]]); + const token = beginRestore(databaseDir); + try { + await assert.rejects(createBackupOffline(DB_NAME), (error) => error.statusCode === 409); + } finally { + completeRestore(databaseDir, token); + } + }); + }); + + describe('online restore_backup validation', function () { + it('rejects database=system with a pointer at running it offline', async function () { + for (const fn of [validateRestoreBackup, restoreBackup]) { + await assert.rejects( + fn({ database: 'system' }), + (error) => error.statusCode === 400 && /restore_backup database=system/.test(error.message) + ); + } + }); + + it('rejects target_database instead of silently restoring in place', async function () { + for (const fn of [validateRestoreBackup, restoreBackup]) { + await assert.rejects( + fn({ database: DB_NAME, target_database: 'copy' }), + (error) => error.statusCode === 400 && /target_database/.test(error.message) + ); + } + }); + }); + + describe('createBackupStream', function () { + it('streams a tar of the database with download headers and no server compression', async function () { + this.timeout(30000); + writeRecords([['alpha', { n: 1 }]]); + const database = RocksDatabase.open(databaseDir); + try { + const stream = createBackupStream(database, DB_NAME, false); + assert.strictEqual(stream.noCompression, true); + assert.strictEqual(stream.headers.get('content-type'), 'application/x-tar'); + assert.strictEqual(stream.headers.get('content-disposition'), `attachment; filename="${DB_NAME}.tar"`); + let bytes = 0; + for await (const chunk of stream) { + bytes += chunk.length; + } + // a tar stream ends with the two-zero-block end-of-archive marker, so any complete + // archive is at least 1024 bytes and 512-byte aligned + assert.ok(bytes >= 1024, `expected a complete tar, got ${bytes} bytes`); + assert.strictEqual(bytes % 512, 0, 'tar streams are 512-byte aligned'); + } finally { + database.close(); + } + }); + + it('sanitizes quotes and backslashes in the content-disposition filename', async function () { + this.timeout(30000); + const database = RocksDatabase.open(databaseDir); + try { + const stream = createBackupStream(database, 'we"ird\\name', false); + assert.strictEqual(stream.headers.get('content-disposition'), 'attachment; filename="we_ird_name.tar"'); + stream.destroy(); + } finally { + database.close(); + } + }); + + it('labels a gzipped stream as application/gzip', async function () { + this.timeout(30000); + const database = RocksDatabase.open(databaseDir); + try { + const stream = createBackupStream(database, DB_NAME, true); + assert.strictEqual(stream.headers.get('content-type'), 'application/gzip'); + assert.strictEqual(stream.headers.get('content-disposition'), `attachment; filename="${DB_NAME}.tar.gz"`); + const chunks = []; + for await (const chunk of stream) { + chunks.push(chunk); + } + const body = Buffer.concat(chunks); + // gzip magic bytes + assert.strictEqual(body[0], 0x1f); + assert.strictEqual(body[1], 0x8b); + } finally { + database.close(); + } + }); + }); + + describe('backup directory hygiene', function () { + it('backups land under the configured backups root, not next to the database', async function () { + this.timeout(30000); + writeRecords([['alpha', { n: 1 }]]); + await createBackupOffline(DB_NAME); + assert.ok(existsSync(backupDirForDatabase(DB_NAME))); + assert.ok(!existsSync(join(databaseDir, 'backups'))); + await purgeBackupsOffline(DB_NAME, 0); + }); + }); +}); diff --git a/unitTests/resources/closeLoadedDatabases.test.js b/unitTests/resources/closeLoadedDatabases.test.js new file mode 100644 index 0000000000..176395b7c2 --- /dev/null +++ b/unitTests/resources/closeLoadedDatabases.test.js @@ -0,0 +1,72 @@ +'use strict'; + +// Regression test for the job-worker RocksDB handle leak that blocked online restore_backup: +// rocksdb-js's registry is process-global across worker threads, and a thread that exits without +// closing leaks its handles (the process-global refCount never drops to zero). Job workers open +// the database graph via getDatabases() and exit per job, so they must release their handles +// explicitly. closeLoadedDatabases() is what jobProcess calls on exit to do that. + +require('../testUtils'); +const assert = require('node:assert'); +const { setupTestDBPath } = require('../testUtils'); +const { table, database, getDatabases, closeDatabase, closeLoadedDatabases } = require('#src/resources/databases'); +const { registryStatus, RocksDatabase } = require('@harperfast/rocksdb-js'); + +describe('RocksDB handle release', function () { + before(function () { + setupTestDBPath(); + }); + + function openRocksDb(databaseName) { + const T = table({ + table: 'pkg', + database: databaseName, + attributes: [{ attribute: 'id', isPrimaryKey: true }, { attribute: 'name' }], + }); + getDatabases(); + return T.primaryStore.rootStore; + } + + function refCountFor(dbPath) { + return registryStatus().find((e) => e.path === dbPath)?.refCount ?? 0; + } + + it('closeDatabase releases all of a database’s native handles (refCount → 0)', async function () { + this.timeout(30000); + const rootStore = openRocksDb('closerelease1'); + if (!(rootStore instanceof RocksDatabase)) return this.skip(); + const dbPath = rootStore.path; + assert.ok(refCountFor(dbPath) > 0, 'database should be open before close'); + + closeDatabase('closerelease1'); + + assert.strictEqual(refCountFor(dbPath), 0, 'no native handles should remain after closeDatabase'); + }); + + it('closeLoadedDatabases releases every loaded user database (what a job worker does on exit)', async function () { + this.timeout(30000); + const a = openRocksDb('closerelease2a'); + const b = openRocksDb('closerelease2b'); + if (!(a instanceof RocksDatabase)) return this.skip(); + assert.ok(refCountFor(a.path) > 0 && refCountFor(b.path) > 0, 'both databases should be open'); + + closeLoadedDatabases(); + + assert.strictEqual(refCountFor(a.path), 0, 'database a should be released'); + assert.strictEqual(refCountFor(b.path), 0, 'database b should be released'); + }); + + it('closeLoadedDatabases releases a tableless database (root store not reachable via any table)', async function () { + this.timeout(30000); + // open a database with no tables: its root store is tracked only on the defined-database + // entry, so the table-based detection alone would miss it and leak the handle + const rootStore = database({ database: 'closerelease3' }); + if (!(rootStore instanceof RocksDatabase)) return this.skip(); + const dbPath = rootStore.path; + assert.ok(refCountFor(dbPath) > 0, 'tableless database should be open'); + + closeLoadedDatabases(); + + assert.strictEqual(refCountFor(dbPath), 0, 'tableless database should be released'); + }); +}); diff --git a/utility/hdbTerms.ts b/utility/hdbTerms.ts index cc708f15f7..422c31e883 100644 --- a/utility/hdbTerms.ts +++ b/utility/hdbTerms.ts @@ -133,7 +133,7 @@ export const DATABASES_DIR_NAME = 'database'; export const LEGACY_DATABASES_DIR_NAME = 'schema'; /** Transaction directory */ export const TRANSACTIONS_DIR_NAME = 'transactions'; -/** Backup directory */ +/** Backup directory (config-file backups and managed RocksDB database backups) */ export const BACKUP_DIR_NAME = 'backup'; /** Key for specifying process specific environment variables */ @@ -297,6 +297,12 @@ export const OPERATIONS_ENUM = { AUDIT_NODE_MODULES: 'audit_node_modules', PURGE_STREAM: 'purge_stream', GET_BACKUP: 'get_backup', + CREATE_BACKUP: 'create_backup', + LIST_BACKUPS: 'list_backups', + VERIFY_BACKUP: 'verify_backup', + DELETE_BACKUP: 'delete_backup', + PURGE_BACKUPS: 'purge_backups', + RESTORE_BACKUP: 'restore_backup', CLEANUP_ORPHAN_BLOBS: 'cleanup_orphan_blobs', GET_ANALYTICS: 'get_analytics', LIST_METRICS: 'list_metrics', @@ -639,6 +645,7 @@ export const CONFIG_PARAMS = { STORAGE_MAX_READ_TRANSACTION_OPEN_TIME: 'storage_maxReadTransactionOpenTime', STORAGE_DEBUGLONGTRANSACTIONS: 'storage_debugLongTransactions', STORAGE_PATH: 'storage_path', + STORAGE_BACKUPPATH: 'storage_backupPath', STORAGE_BLOBPATHS: 'storage_blobPaths', STORAGE_BLOBCLEANUPSPEED: 'storage_blobCleanupSpeed', STORAGE_BLOBREADTIMEOUT: 'storage_blobReadTimeout', @@ -783,6 +790,9 @@ export const JOB_TYPE_ENUM = { export_to_s3: 'export_to_s3', import_from_s3: 'import_from_s3', restart_service: 'restart_service', + create_backup: OPERATIONS_ENUM.CREATE_BACKUP, + verify_backup: OPERATIONS_ENUM.VERIFY_BACKUP, + restore_backup: OPERATIONS_ENUM.RESTORE_BACKUP, } as const; /** Specifies values for licenses */ diff --git a/utility/operation_authorization.ts b/utility/operation_authorization.ts index 112ab277e2..0e56d7aed0 100644 --- a/utility/operation_authorization.ts +++ b/utility/operation_authorization.ts @@ -18,6 +18,7 @@ import * as schemaDescribe from '../dataLayer/schemaDescribe.ts'; import * as delete_ from '../dataLayer/delete.ts'; import readAuditLog from '../dataLayer/readAuditLog.ts'; import getBackup from '../dataLayer/getBackup.ts'; +import * as rocksdbBackup from '../dataLayer/rocksdbBackup.ts'; import * as user from '../security/user.ts'; import * as role from '../security/role.ts'; import harperLogger from '../utility/logging/harper_logger.ts'; @@ -205,6 +206,30 @@ requiredPermissions.set(restart.restart.name, new (permission as any)(true, [], requiredPermissions.set(restart.restartService.name, new (permission as any)(true, [])); requiredPermissions.set(readAuditLog.name, new (permission as any)(true, [], terms.OPERATIONS_ENUM.READ_AUDIT_LOG)); requiredPermissions.set(getBackup.name, new (permission as any)(true, [READ_PERM])); +requiredPermissions.set( + rocksdbBackup.createBackup.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.CREATE_BACKUP) +); +requiredPermissions.set( + rocksdbBackup.listBackups.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.LIST_BACKUPS) +); +requiredPermissions.set( + rocksdbBackup.verifyBackup.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.VERIFY_BACKUP) +); +requiredPermissions.set( + rocksdbBackup.deleteBackup.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.DELETE_BACKUP) +); +requiredPermissions.set( + rocksdbBackup.purgeBackups.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.PURGE_BACKUPS) +); +requiredPermissions.set( + rocksdbBackup.restoreBackup.name, + new (permission as any)(true, [READ_PERM], terms.OPERATIONS_ENUM.RESTORE_BACKUP) +); requiredPermissions.set(schema.cleanupOrphanBlobs.name, new (permission as any)(true, [])); requiredPermissions.set( systemInformation.name, From b38d9ce255f1c6907618584d86c5d31ba67b8b8d Mon Sep 17 00:00:00 2001 From: Chris Barber Date: Thu, 16 Jul 2026 02:10:19 -0500 Subject: [PATCH 2/3] Address PR review: load .env, empty-db restore, Windows fsync, error wrapping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - bin/backup.ts: load `.env` at the top of runBackupCommand so useOperationApi / resolveRequestOptions see HARPER_CLI_TARGET/CLI_TARGET — otherwise a .env-configured remote target was invisible and a destructive backup op could run against local. - rocksdbBackup.ts: restore into an empty/tableless database no longer fails with "no tables to back up" — guard requireRocksRootStore on the database actually having tables (validate + online restore paths). - restoreMarker.ts: directory fsync is now best-effort via a shared fsyncDir helper that ignores EPERM/EISDIR/ENOTSUP, so beginRestore/completeRestore work on Windows. - rocksdbBackup.ts: wrap restore-failure errors in a new Error({cause}) instead of mutating error.message (a frozen/library error's message can be non-writable in strict mode). Co-Authored-By: Claude Opus 4.8 (1M context) --- bin/backup.ts | 5 ++++ dataLayer/restoreMarker.ts | 37 +++++++++++++++++----------- dataLayer/rocksdbBackup.ts | 50 +++++++++++++++++++++++--------------- 3 files changed, 58 insertions(+), 34 deletions(-) diff --git a/bin/backup.ts b/bin/backup.ts index 05faf93b0b..b149c48775 100644 --- a/bin/backup.ts +++ b/bin/backup.ts @@ -33,6 +33,11 @@ const { OPERATIONS_ENUM } = terms; * exception — it streams a live snapshot from a running server and has no offline form. */ export async function runBackupCommand(command: string): Promise { + // Load .env before anything reads process.env — useOperationApi and resolveRequestOptions key + // the local-vs-remote routing off HARPER_CLI_TARGET/CLI_TARGET, and without this a `.env`- + // configured remote target would be invisible, silently running a destructive op locally + // (cliOperations does the same on its first line). + require('dotenv').config(); const request = buildRequest(); const databaseName = request.database || 'data'; diff --git a/dataLayer/restoreMarker.ts b/dataLayer/restoreMarker.ts index a51e09e80d..9a6369faa9 100644 --- a/dataLayer/restoreMarker.ts +++ b/dataLayer/restoreMarker.ts @@ -61,6 +61,26 @@ export function checkRestoreState(dbPath: string): RestoreState { return 'incomplete'; } +/** + * fsync a directory so a create/unlink of an entry within it is durable. Best-effort: Windows (and + * some filesystems) reject opening a directory for fsync with EPERM/EISDIR/ENOTSUP — the durability + * flush is a POSIX nicety, so treat those as a no-op rather than failing the restore. + */ +function fsyncDir(dir: string): void { + let dirFd: number; + try { + dirFd = openSync(dir, 'r'); + } catch (error: any) { + if (error.code === 'EPERM' || error.code === 'EISDIR' || error.code === 'ENOTSUP') return; + throw error; + } + try { + fsyncSync(dirFd); + } finally { + closeSync(dirFd); + } +} + /** * Acquire the per-database restore lock and write the restoring marker. Call before any * destructive step. Returns the lock token for `completeRestore`/`abandonRestore`. @@ -83,12 +103,7 @@ export function beginRestore(dbPath: string): number { } // fsync the parent directory so the marker's directory entry is durable — without this a // power loss can lose the entry, and a half-purged database would load as healthy - const dirFd = openSync(dirname(dbPath), 'r'); - try { - fsyncSync(dirFd); - } finally { - closeSync(dirFd); - } + fsyncDir(dirname(dbPath)); } catch (error) { fileLockRelease(token); throw error; @@ -105,14 +120,8 @@ export function completeRestore(dbPath: string, token: number): void { unlinkSync(restoringMarkerPath(dbPath)); // fsync the parent directory so the marker's *removal* is durable — symmetric with the // creation fsync in beginRestore. Without it, a power loss could resurrect the marker's - // directory entry and misclassify a fully-restored database as incomplete (blocking its - // load until a needless rerun). - const dirFd = openSync(dirname(dbPath), 'r'); - try { - fsyncSync(dirFd); - } finally { - closeSync(dirFd); - } + // directory entry and misclassify a fully-restored database as incomplete. + fsyncDir(dirname(dbPath)); } finally { fileLockRelease(token); } diff --git a/dataLayer/rocksdbBackup.ts b/dataLayer/rocksdbBackup.ts index 8c8f2a698f..18485b35b9 100644 --- a/dataLayer/rocksdbBackup.ts +++ b/dataLayer/rocksdbBackup.ts @@ -276,10 +276,12 @@ export async function validateRestoreBackup(request: any) { } else if ((await listBackupsInDir(backupDir)).length === 0) { throw new BackupNotFoundError(`No backups found for database '${databaseName}'`); } - // The database is normally loaded and RocksDB; a database left unloaded by an interrupted - // restore is also restorable (rerunning the restore is how it is recovered). - const databaseDir = resolveDatabasePath(databaseName); - if (checkRestoreState(databaseDir) === 'clear' || getDatabases()[databaseName]) { + // Only a loaded database that actually has tables can be validated as a single-root RocksDB + // store here (an empty/tableless database has no table to resolve a root store from, and an + // unloaded one recovering an interrupted restore isn't open yet); those cases are validated when + // the restore job runs. `Object.keys` skips the DEFINED_TABLES symbol, so an empty database is 0. + const loaded = getDatabases()[databaseName]; + if (loaded != null && Object.keys(loaded).length > 0) { requireRocksRootStore(databaseName, OPERATIONS_ENUM.RESTORE_BACKUP); } } @@ -313,12 +315,15 @@ export async function restoreBackup(request: any) { } else { backupId = available.reduce((latest, backup) => Math.max(latest, backup.backupId), 0); } - // a loaded database's root store knows its real directory (which can differ from the - // computed default, e.g. legacy layouts); fall back to the computed path only when the - // database is unloaded (recovering an interrupted restore) - const databaseDir = getDatabases()[databaseName] - ? requireRocksRootStore(databaseName, OPERATIONS_ENUM.RESTORE_BACKUP).path - : resolveDatabasePath(databaseName); + // a loaded database *with tables* knows its real directory via its root store (which can differ + // from the computed default, e.g. legacy layouts); fall back to the computed path when the + // database is unloaded (recovering an interrupted restore) or empty (no table to resolve a root + // store from — Object.keys skips the DEFINED_TABLES symbol) + const loaded = getDatabases()[databaseName]; + const databaseDir = + loaded != null && Object.keys(loaded).length > 0 + ? requireRocksRootStore(databaseName, OPERATIONS_ENUM.RESTORE_BACKUP).path + : resolveDatabasePath(databaseName); const lockToken = beginRestoreForDatabase(databaseDir, databaseName); let destructionStarted = false; try { @@ -332,18 +337,20 @@ export async function restoreBackup(request: any) { await verifyDatabaseClosed(databaseDir, databaseName); destructionStarted = true; await backups.restore(backupDir, databaseDir, { backupId, mode: 'purgeAllFiles' }); - } catch (error) { + } catch (error: any) { if (destructionStarted) { // leave the marker: startup/rescan detection reports the incomplete restore until a rerun succeeds abandonRestore(lockToken); - error.message = `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`; - } else { - // nothing destructive happened — clear the marker and let every thread reload the intact database - completeRestore(databaseDir, lockToken); - await signalling.signalSchemaChange( - new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName) + // wrap rather than mutate error.message: a frozen/library error can have a non-writable + // message (assigning it throws TypeError under 'use strict') + throw new Error( + `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`, + { cause: error } ); } + // nothing destructive happened — clear the marker and let every thread reload the intact database + completeRestore(databaseDir, lockToken); + await signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName)); throw error; } completeRestore(databaseDir, lockToken); @@ -506,10 +513,13 @@ export async function restoreBackupOffline(databaseName: string, backupId?: numb const lockToken = beginRestoreForDatabase(databaseDir, targetDatabase ?? databaseName); try { await backups.restore(backupDir, databaseDir, { backupId, mode: 'purgeAllFiles' }); - } catch (error) { + } catch (error: any) { abandonRestore(lockToken); - error.message = `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`; - throw error; + // wrap rather than mutate error.message (a frozen/library error's message may be non-writable) + throw new Error( + `Restore of database '${databaseName}' from backup ${backupId} failed (rerun restore_backup to recover): ${error.message}`, + { cause: error } + ); } completeRestore(databaseDir, lockToken); return { database: databaseName, backupId, restoredTo: databaseDir }; From 2218a5737ecf4e4b72648ca00bb7a28031384546 Mon Sep 17 00:00:00 2001 From: Chris Barber Date: Thu, 16 Jul 2026 10:07:11 -0500 Subject: [PATCH 3/3] Return snake_case fields from backup operations Backup operation responses exposed rocksdb-js's camelCase BackupInfo fields (backupId, numberFiles) directly, inconsistent with the snake_case Operations API convention and these operations' own snake_case inputs. Map at the API boundary: list_backups returns { backup_id, timestamp, size, file_count } (dropping the internal appMetadata), and create/verify/restore return backup_id (offline restore: restored_to). listBackupsInDir stays camelCase for internal use; a new listBackupsOffline maps the CLI offline list to match. Tests updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- bin/backup.ts | 5 ++-- dataLayer/rocksdbBackup.ts | 33 ++++++++++++++++++----- unitTests/dataLayer/rocksdbBackup.test.js | 19 ++++++++----- 3 files changed, 41 insertions(+), 16 deletions(-) diff --git a/bin/backup.ts b/bin/backup.ts index b149c48775..67b328b046 100644 --- a/bin/backup.ts +++ b/bin/backup.ts @@ -13,10 +13,9 @@ import { httpRequest } from '../utility/common_utils.ts'; import { initConfig } from '../config/configUtils.ts'; import { getHdbPid } from '../utility/processManagement/processManagement.js'; import { - backupDirForDatabase, createBackupOffline, deleteBackupOffline, - listBackupsInDir, + listBackupsOffline, purgeBackupsOffline, restoreBackupOffline, verifyBackupOffline, @@ -65,7 +64,7 @@ export async function runBackupCommand(command: string): Promise { result = await createBackupOffline(databaseName); break; case OPERATIONS_ENUM.LIST_BACKUPS: - result = await listBackupsInDir(backupDirForDatabase(databaseName)); + result = await listBackupsOffline(databaseName); break; case OPERATIONS_ENUM.VERIFY_BACKUP: result = await verifyBackupOffline(databaseName, request.backup_id, request.verify_checksum); diff --git a/dataLayer/rocksdbBackup.ts b/dataLayer/rocksdbBackup.ts index 18485b35b9..b9e89c8409 100644 --- a/dataLayer/rocksdbBackup.ts +++ b/dataLayer/rocksdbBackup.ts @@ -157,13 +157,26 @@ async function findBackup(backupDir: string, backupId: number, databaseName: str return info; } +/** + * Shape a rocksdb-js BackupInfo into the snake_case response the Operations API exposes — the + * binding's fields are camelCase, and `appMetadata` is internal, so it is not passed through. + */ +function toBackupResponse(info: BackupInfo): { + backup_id: number; + timestamp: number; + size: number; + file_count: number; +} { + return { backup_id: info.backupId, timestamp: info.timestamp, size: info.size, file_count: info.numberFiles }; +} + // --- synchronous operations --- export async function listBackups(request: any) { const databaseName = getDatabaseName(request); logger.info(`Listing backups for database '${databaseName}'`); requireRocksRootStore(databaseName, OPERATIONS_ENUM.LIST_BACKUPS); - return listBackupsInDir(backupDirForDatabase(databaseName)); + return (await listBackupsInDir(backupDirForDatabase(databaseName))).map(toBackupResponse); } export async function deleteBackup(request: any) { @@ -221,7 +234,7 @@ export async function createBackup(request: any) { } catch (error) { throw mapLockedError(error, databaseName); } - return { database: databaseName, backupId, ...(await describeBackup(backupDir, backupId)) }; + return { database: databaseName, backup_id: backupId, ...(await describeBackup(backupDir, backupId)) }; } /** @@ -253,7 +266,7 @@ export async function verifyBackup(request: any) { const backupDir = backupDirForDatabase(databaseName); await findBackup(backupDir, backupId, databaseName); await backups.verify(backupDir, backupId, { verifyWithChecksum }); - return { database: databaseName, backupId, ok: true }; + return { database: databaseName, backup_id: backupId, ok: true }; } export async function validateRestoreBackup(request: any) { @@ -356,7 +369,7 @@ export async function restoreBackup(request: any) { completeRestore(databaseDir, lockToken); // signal again: with the marker gone, every thread's rescan reloads the restored database await signalling.signalSchemaChange(new SchemaEventMsg(process.pid, OPERATIONS_ENUM.RESTORE_BACKUP, databaseName)); - return { database: databaseName, backupId }; + return { database: databaseName, backup_id: backupId }; } // After the close broadcast is acknowledged, every worker thread has released its Harper-managed @@ -460,7 +473,7 @@ export async function createBackupOffline(databaseName: string) { } catch (error) { throw mapLockedError(error, databaseName); } - return { database: databaseName, backupId, ...(await describeBackup(backupDir, backupId)) }; + return { database: databaseName, backup_id: backupId, ...(await describeBackup(backupDir, backupId)) }; } finally { database.close(); } @@ -522,7 +535,7 @@ export async function restoreBackupOffline(databaseName: string, backupId?: numb ); } completeRestore(databaseDir, lockToken); - return { database: databaseName, backupId, restoredTo: databaseDir }; + return { database: databaseName, backup_id: backupId, restored_to: databaseDir }; } function isMissingOrEmptyDir(path: string): boolean { @@ -536,13 +549,19 @@ function isMissingOrEmptyDir(path: string): boolean { // --- offline management wrappers (no engine validation: they operate on the directory only) --- +export async function listBackupsOffline(databaseName: string) { + validateDatabaseName(databaseName); + // map to the same snake_case response shape as the online list_backups operation + return (await listBackupsInDir(backupDirForDatabase(databaseName))).map(toBackupResponse); +} + export async function verifyBackupOffline(databaseName: string, backupId: number, verifyChecksum?: boolean) { validateDatabaseName(databaseName); const verifyWithChecksum = requireBooleanOption(verifyChecksum, 'verify_checksum'); const backupDir = backupDirForDatabase(databaseName); await findBackup(backupDir, backupId, databaseName); await backups.verify(backupDir, backupId, { verifyWithChecksum }); - return { database: databaseName, backupId, ok: true }; + return { database: databaseName, backup_id: backupId, ok: true }; } export async function deleteBackupOffline(databaseName: string, backupId: number) { diff --git a/unitTests/dataLayer/rocksdbBackup.test.js b/unitTests/dataLayer/rocksdbBackup.test.js index d6ddcddc64..250cbe7e1a 100644 --- a/unitTests/dataLayer/rocksdbBackup.test.js +++ b/unitTests/dataLayer/rocksdbBackup.test.js @@ -11,6 +11,7 @@ const { deleteBackupOffline, getBackupsRoot, listBackupsInDir, + listBackupsOffline, purgeBackupsOffline, restoreBackup, restoreBackupOffline, @@ -97,13 +98,13 @@ describe('rocksdbBackup', function () { const first = await createBackupOffline(DB_NAME); assert.strictEqual(first.database, DB_NAME); - assert.strictEqual(first.backupId, 1); + assert.strictEqual(first.backup_id, 1); assert.ok(first.size > 0, 'size should come from the backups.list match'); assert.ok(first.timestamp !== undefined); writeRecords([['gamma', { n: 3 }]]); const second = await createBackupOffline(DB_NAME); - assert.strictEqual(second.backupId, 2); + assert.strictEqual(second.backup_id, 2); const backupDir = backupDirForDatabase(DB_NAME); const listed = await listBackupsInDir(backupDir); @@ -112,14 +113,20 @@ describe('rocksdbBackup', function () { [1, 2] ); + // the exposed list is snake_case (backup_id/file_count), not the binding's camelCase + const exposed = await listBackupsOffline(DB_NAME); + assert.deepStrictEqual(Object.keys(exposed[0]).sort(), ['backup_id', 'file_count', 'size', 'timestamp']); + assert.strictEqual(exposed[0].backup_id, 1); + assert.ok(exposed[0].file_count >= 1, 'file_count should be mapped from numberFiles'); + const verified = await verifyBackupOffline(DB_NAME, 1, true); - assert.deepStrictEqual(verified, { database: DB_NAME, backupId: 1, ok: true }); + assert.deepStrictEqual(verified, { database: DB_NAME, backup_id: 1, ok: true }); // restore the first backup into a separate target database directory const restored = await restoreBackupOffline(DB_NAME, 1, `${DB_NAME}-restored`); - assert.strictEqual(restored.backupId, 1); + assert.strictEqual(restored.backup_id, 1); const restoredDir = join(storageDir, `${DB_NAME}-restored`); - assert.strictEqual(restored.restoredTo, restoredDir); + assert.strictEqual(restored.restored_to, restoredDir); assert.strictEqual(checkRestoreState(restoredDir), 'clear'); const restoredDb = RocksDatabase.open(restoredDir); try { @@ -131,7 +138,7 @@ describe('rocksdbBackup', function () { // restore latest (no backup_id) in place over the source database const latestRestore = await restoreBackupOffline(DB_NAME); - assert.strictEqual(latestRestore.backupId, 2); + assert.strictEqual(latestRestore.backup_id, 2); assert.strictEqual(checkRestoreState(databaseDir), 'clear'); const inPlaceDb = RocksDatabase.open(databaseDir); try {