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;