From 94a9f815c9e68d2f206cbd608e4899d614f77a3d Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Sat, 3 Oct 2026 21:04:33 +0800 Subject: [PATCH] fix(database): batch database saves for bulk host writes (#1554) * fix(database): yield to the event loop between database saves Each SQLite save serializes and encrypts the whole database into new buffers, which V8 only releases once the event loop turns. Boot migrations and bulk host import write plugin settings per host and per key in an awaited loop, so hundreds of full copies piled up and the container ran out of memory. * fix(database): save once per bulk host write instead of once per row On SQLite every repository write force-saves the whole database. Boot plugin data migrations, host defaults materialization and the bulk host routes write one row per host per setting, so they serialized and encrypted the full database hundreds of times in a row. DatabaseSaveTrigger.batched wraps a function so force saves made while it runs collapse into a single save when it finishes. Nested scopes fold into the outer one, and work that outlives its scope saves normally. --- .../database/routes/host-bulk-routes.ts | 13 +- src/backend/hosts/defaults/materialize.ts | 149 +++++++++--------- src/backend/tests/sync/two-way.test.ts | 1 + .../tests/utils/database-save-trigger.test.ts | 86 ++++++++++ src/backend/upgrade/boot-migrations.ts | 27 ++-- src/backend/utils/database-save-trigger.ts | 41 ++++- 6 files changed, 225 insertions(+), 92 deletions(-) diff --git a/src/backend/database/routes/host-bulk-routes.ts b/src/backend/database/routes/host-bulk-routes.ts index aa3fb77f6..37c9d2506 100644 --- a/src/backend/database/routes/host-bulk-routes.ts +++ b/src/backend/database/routes/host-bulk-routes.ts @@ -22,6 +22,7 @@ import { getErrorMessage } from "../../utils/error-message.js"; import type { AuthenticatedRequest } from "../../../types/index.js"; import type { Request, RequestHandler, Response, Router } from "express"; import { sshLogger } from "../../utils/logger.js"; +import { DatabaseSaveTrigger } from "../../utils/database-save-trigger.js"; import { createCurrentCredentialRepository, createCurrentHostRepository, @@ -237,7 +238,7 @@ export function registerHostBulkRoutes( authenticateJWT, requireEditPermission, requireDataAccess, - async (req: Request, res: Response) => { + DatabaseSaveTrigger.batched(async (req: Request, res: Response) => { const userId = (req as AuthenticatedRequest).userId; const { hostIds, updates } = req.body; @@ -366,7 +367,7 @@ export function registerHostBulkRoutes( sshLogger.error("Failed to bulk update hosts:", error); return res.status(500).json({ error: "Failed to bulk update hosts" }); } - }, + }), ); /** @@ -455,7 +456,7 @@ export function registerHostBulkRoutes( requireCreatePermission, requireEditPermission, requireDataAccess, - async (req: Request, res: Response) => { + DatabaseSaveTrigger.batched(async (req: Request, res: Response) => { const userId = (req as AuthenticatedRequest).userId; const { hosts: hostsToImport, @@ -825,7 +826,7 @@ export function registerHostBulkRoutes( failed: results.failed, errors: results.errors, }); - }, + }), ); /** @@ -862,7 +863,7 @@ export function registerHostBulkRoutes( requireCreatePermission, requireEditPermission, requireDataAccess, - async (req: Request, res: Response) => { + DatabaseSaveTrigger.batched(async (req: Request, res: Response) => { const userId = (req as AuthenticatedRequest).userId; const { content, overwrite } = req.body; @@ -1028,7 +1029,7 @@ export function registerHostBulkRoutes( failed: results.failed, errors: results.errors, }); - }, + }), ); } diff --git a/src/backend/hosts/defaults/materialize.ts b/src/backend/hosts/defaults/materialize.ts index f178a0514..e0edf7280 100644 --- a/src/backend/hosts/defaults/materialize.ts +++ b/src/backend/hosts/defaults/materialize.ts @@ -30,6 +30,7 @@ import { import { hostSettingsPlugins } from "../../database/routes/host-plugin-settings.js"; import { notifySettingChange } from "../../plugins/settings.js"; import { databaseLogger } from "../../utils/logger.js"; +import { DatabaseSaveTrigger } from "../../utils/database-save-trigger.js"; import { buildCatalog, catalogNamespaces, @@ -342,83 +343,85 @@ async function afterHostsChanged( * Classifies and rewrites a set of hosts. Returns the hosts whose values or * overrides changed; with `dryRun` nothing is written. */ -export async function materializeHosts( - target: HostDefaultsTarget, - options: MaterializeOptions = {}, -): Promise { - const repository = createCurrentHostDefaultsRepository(); - const rows = (await repository.listHosts(target)).filter( - // A desktop's read-only copy of a shared host follows its owner's server. - (row) => !row.sharedSource, - ); - if (rows.length === 0) return { changedHostIds: [] }; - - const { catalog } = currentCatalog(); - const userIds = [...new Set(rows.map((row) => row.userId))]; - const [index, placement, pluginRows] = await Promise.all([ - loadDefaultsIndex(userIds), - loadPlacement(userIds), - loadPluginRows(rows.map((row) => row.id)), - ]); - const withOverlay = applyOverlay(index, options.overlay); - - const changed: Array<{ row: HostRow; plan: HostPlan }> = []; - for (const row of rows) { - const levels = levelsForHost( - withOverlay, - row.userId, - effectiveFolderPath(row.id, placement), +export const materializeHosts = DatabaseSaveTrigger.batched( + async ( + target: HostDefaultsTarget, + options: MaterializeOptions = {}, + ): Promise => { + const repository = createCurrentHostDefaultsRepository(); + const rows = (await repository.listHosts(target)).filter( + // A desktop's read-only copy of a shared host follows its owner's server. + (row) => !row.sharedSource, ); - const plan = planHost(row, levels, catalog, pluginRows.get(row.id), { - keys: options.keys, - }); - const valuesChanged = - Object.keys(plan.columns).some((key) => key !== "defaultOverrides") || - plan.pluginWrites.length > 0; - if (options.dryRun) { - if (valuesChanged) changed.push({ row, plan }); - continue; + if (rows.length === 0) return { changedHostIds: [] }; + + const { catalog } = currentCatalog(); + const userIds = [...new Set(rows.map((row) => row.userId))]; + const [index, placement, pluginRows] = await Promise.all([ + loadDefaultsIndex(userIds), + loadPlacement(userIds), + loadPluginRows(rows.map((row) => row.id)), + ]); + const withOverlay = applyOverlay(index, options.overlay); + + const changed: Array<{ row: HostRow; plan: HostPlan }> = []; + for (const row of rows) { + const levels = levelsForHost( + withOverlay, + row.userId, + effectiveFolderPath(row.id, placement), + ); + const plan = planHost(row, levels, catalog, pluginRows.get(row.id), { + keys: options.keys, + }); + const valuesChanged = + Object.keys(plan.columns).some((key) => key !== "defaultOverrides") || + plan.pluginWrites.length > 0; + if (options.dryRun) { + if (valuesChanged) changed.push({ row, plan }); + continue; + } + if (valuesChanged || plan.overridesChanged) changed.push({ row, plan }); } - if (valuesChanged || plan.overridesChanged) changed.push({ row, plan }); - } - if (options.dryRun) { - return { changedHostIds: changed.map(({ row }) => row.id) }; - } + if (options.dryRun) { + return { changedHostIds: changed.map(({ row }) => row.id) }; + } - await repository.updateHosts( - changed.map(({ row, plan }) => ({ - id: row.id, - values: plan.columns as Partial, - })), - ); - const settings = createCurrentPluginSettingsRepository(); - for (const { plan } of changed) { - for (const write of plan.pluginWrites) { - if (write.value === undefined) { - await settings.delete( - write.pluginId, - "host", - String(write.hostId), - write.key, - ); - } else { - await settings.set( - write.pluginId, - "host", - String(write.hostId), - write.key, - JSON.stringify(write.value), - ); + await repository.updateHosts( + changed.map(({ row, plan }) => ({ + id: row.id, + values: plan.columns as Partial, + })), + ); + const settings = createCurrentPluginSettingsRepository(); + for (const { plan } of changed) { + for (const write of plan.pluginWrites) { + if (write.value === undefined) { + await settings.delete( + write.pluginId, + "host", + String(write.hostId), + write.key, + ); + } else { + await settings.set( + write.pluginId, + "host", + String(write.hostId), + write.key, + JSON.stringify(write.value), + ); + } } } - } - const valueChanges = changed.filter( - ({ plan }) => - plan.pluginWrites.length > 0 || - Object.keys(plan.columns).some((key) => key !== "defaultOverrides"), - ); - await afterHostsChanged(valueChanges); - return { changedHostIds: valueChanges.map(({ row }) => row.id) }; -} + const valueChanges = changed.filter( + ({ plan }) => + plan.pluginWrites.length > 0 || + Object.keys(plan.columns).some((key) => key !== "defaultOverrides"), + ); + await afterHostsChanged(valueChanges); + return { changedHostIds: valueChanges.map(({ row }) => row.id) }; + }, +); diff --git a/src/backend/tests/sync/two-way.test.ts b/src/backend/tests/sync/two-way.test.ts index 2aa19d0ac..ded0f0ab2 100644 --- a/src/backend/tests/sync/two-way.test.ts +++ b/src/backend/tests/sync/two-way.test.ts @@ -26,6 +26,7 @@ vi.mock("../../utils/database-save-trigger.js", () => ({ DatabaseSaveTrigger: { triggerSave: vi.fn(), forceSave: vi.fn(async () => {}), + batched: (work: T) => work, }, })); vi.mock("../../utils/system-crypto.js", () => ({ diff --git a/src/backend/tests/utils/database-save-trigger.test.ts b/src/backend/tests/utils/database-save-trigger.test.ts index edf0c5d17..45d8858b2 100644 --- a/src/backend/tests/utils/database-save-trigger.test.ts +++ b/src/backend/tests/utils/database-save-trigger.test.ts @@ -37,6 +37,18 @@ describe("DatabaseSaveTrigger", () => { }); }); + it("lets the event loop turn before a force save resolves", async () => { + DatabaseSaveTrigger.initialize(vi.fn().mockResolvedValue(undefined)); + let turned = false; + setImmediate(() => { + turned = true; + }); + + await DatabaseSaveTrigger.forceSave("bulk_write"); + + expect(turned).toBe(true); + }); + it("debounces dirty saves and marks the database clean after saving", async () => { vi.useFakeTimers(); const save = vi.fn().mockResolvedValue(undefined); @@ -49,6 +61,7 @@ describe("DatabaseSaveTrigger", () => { expect(DatabaseSaveTrigger.getStatus().hasPendingTimeout).toBe(true); await vi.advanceTimersByTimeAsync(2000); + await vi.runOnlyPendingTimersAsync(); expect(save).toHaveBeenCalledTimes(1); expect(DatabaseSaveTrigger.isDirty).toBe(false); @@ -78,4 +91,77 @@ describe("DatabaseSaveTrigger", () => { expect(save).toHaveBeenCalledTimes(2); expect(DatabaseSaveTrigger.getStatus().pendingSave).toBe(false); }); + + it("collapses force saves inside a batch into one save", async () => { + const save = vi.fn().mockResolvedValue(undefined); + DatabaseSaveTrigger.initialize(save); + + const result = await DatabaseSaveTrigger.batched(async (rows: number) => { + for (let i = 0; i < rows; i++) { + await DatabaseSaveTrigger.forceSave("row_write"); + } + expect(save).not.toHaveBeenCalled(); + return rows; + })(50); + + expect(result).toBe(50); + expect(save).toHaveBeenCalledTimes(1); + expect(DatabaseSaveTrigger.isDirty).toBe(false); + }); + + it("folds a nested batch into the outer one", async () => { + const save = vi.fn().mockResolvedValue(undefined); + DatabaseSaveTrigger.initialize(save); + const inner = DatabaseSaveTrigger.batched(() => + DatabaseSaveTrigger.forceSave("inner_write"), + ); + + await DatabaseSaveTrigger.batched(async () => { + await inner(); + await inner(); + })(); + + expect(save).toHaveBeenCalledTimes(1); + }); + + it("skips the save when a batch wrote nothing", async () => { + const save = vi.fn().mockResolvedValue(undefined); + DatabaseSaveTrigger.initialize(save); + + await DatabaseSaveTrigger.batched(async () => {})(); + + expect(save).not.toHaveBeenCalled(); + }); + + it("still saves what a failed batch wrote", async () => { + const save = vi.fn().mockResolvedValue(undefined); + DatabaseSaveTrigger.initialize(save); + + await expect( + DatabaseSaveTrigger.batched(async () => { + await DatabaseSaveTrigger.forceSave("row_write"); + throw new Error("boom"); + })(), + ).rejects.toThrow("boom"); + + expect(save).toHaveBeenCalledTimes(1); + }); + + it("saves immediately for work that outlives its batch", async () => { + const save = vi.fn().mockResolvedValue(undefined); + DatabaseSaveTrigger.initialize(save); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + let detached: Promise | undefined; + + await DatabaseSaveTrigger.batched(async () => { + detached = gate.then(() => DatabaseSaveTrigger.forceSave("late_write")); + })(); + release?.(); + await detached; + + expect(save).toHaveBeenCalledTimes(1); + }); }); diff --git a/src/backend/upgrade/boot-migrations.ts b/src/backend/upgrade/boot-migrations.ts index e5f9cd3ab..cac8c7cd2 100644 --- a/src/backend/upgrade/boot-migrations.ts +++ b/src/backend/upgrade/boot-migrations.ts @@ -8,6 +8,7 @@ */ import { databaseLogger } from "../utils/logger.js"; +import { DatabaseSaveTrigger } from "../utils/database-save-trigger.js"; import { getErrorMessage } from "../utils/error-message.js"; import { runTailscaleSettingsMigration } from "./tailscale-settings-migration.js"; import { runProxmoxSettingsMigration } from "./proxmox-settings-migration.js"; @@ -73,19 +74,21 @@ const MIGRATIONS: Array<[string, () => Promise]> = [ ["runHostDefaultsMigration", runHostDefaultsMigration], ]; -export async function runPluginDataMigrations(): Promise { - for (const [name, run] of MIGRATIONS) { - try { - await run(); - } catch (error) { - databaseLogger.warn(`Plugin data migration ${name} failed`, { - operation: "plugin_data_migration", - error: getErrorMessage(error), - }); +export const runPluginDataMigrations = DatabaseSaveTrigger.batched( + async (): Promise => { + for (const [name, run] of MIGRATIONS) { + try { + await run(); + } catch (error) { + databaseLogger.warn(`Plugin data migration ${name} failed`, { + operation: "plugin_data_migration", + error: getErrorMessage(error), + }); + } } - } - await applyHostDefaults(); -} + await applyHostDefaults(); + }, +); /** * Classifies hosts nobody classified yet (from an older release, or for a diff --git a/src/backend/utils/database-save-trigger.ts b/src/backend/utils/database-save-trigger.ts index 4630b5c1c..c2d8d0ddf 100644 --- a/src/backend/utils/database-save-trigger.ts +++ b/src/backend/utils/database-save-trigger.ts @@ -1,3 +1,4 @@ +import { AsyncLocalStorage } from "node:async_hooks"; import { getErrorMessage } from "./error-message.js"; import { databaseLogger } from "./logger.js"; @@ -8,6 +9,10 @@ export class DatabaseSaveTrigger { private static activeSave: Promise | null = null; private static saveTimeout: NodeJS.Timeout | null = null; private static _dirty = false; + private static batch = new AsyncLocalStorage<{ + open: boolean; + pending: boolean; + }>(); static initialize(saveFunction: () => Promise): void { this.saveFunction = saveFunction; @@ -60,6 +65,27 @@ export class DatabaseSaveTrigger { }, 2000); } + /** + * Wraps `work` so the force saves made while it runs collapse into one save + * when it finishes. Every save serializes and encrypts the whole database, + * so a loop writing one row per host per setting otherwise rewrites the + * file once per row. Nested scopes fold into the outer one; work that + * outlives its scope (fire-and-forget) saves normally again. + */ + static batched( + work: (...args: A) => Promise, + ): (...args: A) => Promise { + return async (...args) => { + const scope = { open: true, pending: false }; + try { + return await this.batch.run(scope, () => work(...args)); + } finally { + scope.open = false; + if (scope.pending) await this.forceSave("batched_writes"); + } + }; + } + static async forceSave(reason: string = "critical_operation"): Promise { if (!this.isInitialized || !this.saveFunction) { databaseLogger.warn( @@ -72,6 +98,13 @@ export class DatabaseSaveTrigger { return; } + const scope = this.batch.getStore(); + if (scope?.open) { + scope.pending = true; + this._dirty = true; + return; + } + if (this.saveTimeout) { clearTimeout(this.saveTimeout); this.saveTimeout = null; @@ -99,7 +132,13 @@ export class DatabaseSaveTrigger { } } - const save = Promise.resolve().then(() => this.saveFunction!()); + // Each save serializes the whole database into fresh buffers, and V8 only + // frees them once the event loop turns. A loop of awaited writes (boot + // migrations, bulk import) never yields, so every save's copies piled up + // until the process ran out of memory. + const save = Promise.resolve() + .then(() => this.saveFunction!()) + .finally(() => new Promise((resolve) => setImmediate(resolve))); this.activeSave = save; this.pendingSave = true;