diff --git a/components/DESIGN.md b/components/DESIGN.md index c2cf56d2b2..43fa21edae 100644 --- a/components/DESIGN.md +++ b/components/DESIGN.md @@ -770,9 +770,9 @@ Branched databases go last, once nothing holds them, unless `keepBranches` keeps earlier peer ignores it and its restart removes them), or while the pool may hold them. Each node answers `{ node, componentRemoved, workerStopped, branchesRemoved }`; `replicated` (`dropPeerResults`) has its peers', `null` where one did not say, `error` where one failed. A worker the coming stop or restart replaces is answered first, `null` for what is -not done (harper#3091), so the owed removal is a durable `` `removal-pending` `` file beside the branches, written before +not done (harper#3091), so the owed removal is a durable `` `removal-pending` `` file beside the branches (or its temp file, which a crash can leave unrenamed), written before the drop changes anything, cleared only by a completed removal or a `keepBranches` drop. Main retries each under its lock -and marker recheck after every drop, after a start's pool reports, and on capped backoff; redeployed or held ones wait. +and marker recheck after every drop, after a start's pool reports, when a worker exits or reports its load, and on capped backoff. A redeployed one waits for a drop, a held one (or an unreadable root config) a pass, one main serves as the only worker a start. ## A rolling restart job completes only if every peer restarted (`bin/restart.ts` `restartEachPeer`) diff --git a/components/operations.ts b/components/operations.ts index b18431b0d6..e3da41fbfc 100644 --- a/components/operations.ts +++ b/components/operations.ts @@ -481,6 +481,9 @@ const BRANCH_STORAGE_RETAINED = { restart: '. Any branched database storage this application owns was left in place, since the restart did not finish ' + 'replacing every worker; it is discarded once none holds it', + serving: + '. Any branched database storage this application owns was left in place, since the main thread serves it as ' + + "this node's only worker; it is discarded at Harper's next start", }; /** Report a restart outcome the operation's own success message cannot convey. */ @@ -1883,6 +1886,7 @@ async function dropApplication(request, requester?: any, answerEarly?: any) { const running = (await manageThreads.getRunningIsolatedApplications()).includes(project); const isolated = configuredIsolated(project) || running; const poolMayHold = manageThreads.poolMayHold(project); + const mainServes = await mainServesApplication(project); const hadWorker = manageThreads.workersForApplication(project).length > 0; const restartsPool = present && req.restart === true && (!isolated || poolMayHold); const keep = req.keepBranches === true; @@ -1906,6 +1910,7 @@ async function dropApplication(request, requester?: any, answerEarly?: any) { } let retained; if (branched && keep) retained = 'keepBranches'; + else if (branched && mainServes) retained = 'serving'; else if (branched && ((present && !isolated) || poolMayHold) && !restartsPool) retained = 'loaded'; const outcome = nodeOutcome(present); let message = present @@ -1963,14 +1968,23 @@ let branchRemovalRetryTimer: NodeJS.Timeout | undefined; let branchRemovalPass: Promise | undefined; let branchRemovalPassRunning = false; let branchRemovalPassRequested = false; +let workerReleasesWatched = false; /** * Finish the branch removals drops on this node recorded as owed and could not complete: after every whole-component - * drop, once a start has settled its components and its workers have reported what they loaded, and on a backoff timer - * while one that a later pass can finish remains. One application at a time, under its component lock, which a pass - * does not wait for. A request during a pass runs one more, so a debt recorded after the listing is not missed. + * drop, once a start has settled its components and its workers have reported what they loaded, and while one that a + * later pass can finish remains, on a backoff timer and whenever a worker exits or reports what it loaded. One + * application at a time, under its component lock, which a pass does not wait for. A request during a pass runs one + * more, so a debt recorded after the listing is not missed. */ function retryPendingBranchRemovals(): Promise { + if (!workerReleasesWatched) { + workerReleasesWatched = true; + manageThreads.onWorkerRelease(() => { + if (manageThreads.isProcessShuttingDown()) return; + if (branchRemovalRetryTimer || branchRemovalPassRunning) void retryPendingBranchRemovals(); + }); + } branchRemovalPassRequested = true; if (!branchRemovalPassRunning) { branchRemovalPassRunning = true; @@ -2018,9 +2032,10 @@ async function runBranchRemovalPasses(): Promise { } /** - * Whether the debt needs no timer: removed, cancelled by a later drop, or waiting on an application that is deployed - * again (or that a root config it cannot read might name), for that application's next drop. A worker that may still - * hold it, a lock another holds, a directory it cannot check, or a failed removal leave it to the next timed pass. + * Whether the debt needs no timer: removed, cancelled by a later drop, waiting on an application that is deployed + * again, for that application's next drop, or on a main thread that serves it, for the next start. A root config it + * cannot read, a worker that may still hold it, a lock another holds, a directory it cannot check, or a failed removal + * leave it to the next timed pass. */ async function retryBranchRemoval(project: string): Promise { const { hasPendingBranchRemoval, removeBranchesForApplication } = branchDatabaseRuntimeModule; @@ -2030,7 +2045,15 @@ async function retryBranchRemoval(project: string): Promise { componentPath, async () => { if (!hasPendingBranchRemoval(project)) return true; - if (pathPresent(componentPath) || hasRootConfigEntry(project)) return true; + if (pathPresent(componentPath)) return true; + let configured; + try { + configured = readRootConfigEntry(project) !== undefined; + } catch { + // not logged: the parser's message can quote the file + return false; + } + if (configured || (await mainServesApplication(project))) return true; if (manageThreads.poolMayHold(project) || manageThreads.workersForApplication(project).length > 0) { return false; } @@ -2052,6 +2075,14 @@ async function retryBranchRemoval(project: string): Promise { } } +let applicationsPlacedOnMain: (() => string[]) | undefined; +/** With `threads.count: 0` main is the only worker, and holds every application it placed until Harper restarts. */ +async function mainServesApplication(project: string): Promise { + if (!isMainThread || manageThreads.getWorkerIndex() !== 0) return false; + applicationsPlacedOnMain ??= (await import('./componentLoader.ts')).applicationsPlacedOnThisThread; + return applicationsPlacedOnMain().includes(project); +} + /** @internal — testing only */ function setBranchRemovalRetryDelayForTests(firstMs: number = BRANCH_REMOVAL_RETRY_FIRST_MS): void { branchRemovalRetryFirstMs = branchRemovalRetryDelayMs = firstMs; diff --git a/integrationTests/components/branched-database-package.test.ts b/integrationTests/components/branched-database-package.test.ts index a3acb320a2..38b6649138 100644 --- a/integrationTests/components/branched-database-package.test.ts +++ b/integrationTests/components/branched-database-package.test.ts @@ -422,6 +422,48 @@ for (const isolated of [false, true]) { assert.strictEqual((await baseRows(['seed']))[0].note, 'base'); } ); + + test( + 'with the main thread as the only worker, a drop keeps the fork main serves until the next start', + { timeout: 240000, skip: isolated }, + async () => { + const mainOnly = { ...config, threads: { count: 0 } }; + await killHarper(ctx); + await startHarper(ctx, { config: mainOnly }); + const deployed = await operation({ + operation: 'deploy_component', + project, + package: pkg, + host, + branchedDatabases: ['data'], + }); + assert.strictEqual(deployed.status, 200, JSON.stringify(deployed.body)); + await killHarper(ctx); + await startHarper(ctx, { config: mainOnly }); + const written = await appRequest('/PackageProbe/main-write', 'PUT', { note: 'main' }); + assert.ok(written.status >= 200 && written.status < 300, written.body); + + for (const restart of [true, false]) { + const dropped = await operation({ operation: 'drop_component', project, restart }); + assert.strictEqual(dropped.status, 200, JSON.stringify(dropped.body)); + assert.strictEqual(dropped.body.branchesRemoved, false, JSON.stringify(dropped.body)); + assert.match(dropped.body.message, /main thread serves it/); + for (const settled = Date.now() + 2000; Date.now() < settled;) { + assert.ok(existsSync(branchRoot()), 'the fork main serves was removed'); + await new Promise((resolve) => setTimeout(resolve, 100)); + } + assert.strictEqual((await read('/PackageProbe/main-write')).note, 'main'); + } + + await killHarper(ctx); + await startHarper(ctx, { config }); + await waitFor(() => !existsSync(branchRoot()), { + timeout: 60000, + interval: 250, + message: 'the next start left the fork on disk', + }); + } + ); } ); } diff --git a/resources/branchDatabase.ts b/resources/branchDatabase.ts index 98a8abd0b9..1fbd9d1c21 100644 --- a/resources/branchDatabase.ts +++ b/resources/branchDatabase.ts @@ -842,9 +842,15 @@ function branchDirectoriesFor(appName: string): { * In each `` directory of an application's branches: a drop owes their removal. It is written, durably, before * that drop changes anything, so the debt outlives an answer the drop gives before the removal, and the process; only a * removal that completes, or a later drop that keeps them, clears it. A file, so `branchDirectoriesFor` never reads it - * as a branch. + * as a branch. Its temp file counts too: only a drop that discards the branches writes one, and a crash can leave it + * unrenamed after that drop removed the application. */ const REMOVAL_PENDING_MARKER = '`removal-pending`'; +const REMOVAL_PENDING_PARTIAL = `${REMOVAL_PENDING_MARKER}.partial`; + +function marksPendingRemoval(directory: string): boolean { + return pathPresent(join(directory, REMOVAL_PENDING_MARKER)) || pathPresent(join(directory, REMOVAL_PENDING_PARTIAL)); +} /** Refused when a directory the removal would cover cannot be read, since no marker could cover it. */ export function recordPendingBranchRemoval(appName: string): void { @@ -853,11 +859,7 @@ export function recordPendingBranchRemoval(appName: string): void { throw new Error(`Could not record the owed removal of branch storage for '${appName}': ${failures.join('; ')}`); } for (const directory of directories) { - writeFileDurably( - join(directory, REMOVAL_PENDING_MARKER), - new Date().toISOString(), - `${REMOVAL_PENDING_MARKER}.partial` - ); + writeFileDurably(join(directory, REMOVAL_PENDING_MARKER), new Date().toISOString(), REMOVAL_PENDING_PARTIAL); } } @@ -867,6 +869,8 @@ export async function clearPendingBranchRemoval(appName: string): Promise throw new Error(`Could not clear the owed removal of branch storage for '${appName}': ${failures.join('; ')}`); } for (const directory of directories) { + // before the marker, which keeps the debt discoverable if this fails + removeFileDurably(join(directory, REMOVAL_PENDING_PARTIAL)); removeFileDurably(join(directory, REMOVAL_PENDING_MARKER)); await rmdir(directory).catch(() => {}); } @@ -875,7 +879,7 @@ export async function clearPendingBranchRemoval(appName: string): Promise /** Throws when that cannot be established, rather than answering no. */ export function hasPendingBranchRemoval(appName: string): boolean { const { directories, failures } = branchDirectoriesFor(appName); - if (directories.some((directory) => pathPresent(join(directory, REMOVAL_PENDING_MARKER)))) return true; + if (directories.some(marksPendingRemoval)) return true; if (failures.length) throw new Error(`Could not check branch storage for '${appName}': ${failures.join('; ')}`); return false; } @@ -898,11 +902,11 @@ export function applicationsWithPendingBranchRemoval(): { applications: string[] } for (const entry of entries) { if (!entry.isDirectory()) continue; - const marker = join(root, entry.name, REMOVAL_PENDING_MARKER); + const directory = join(root, entry.name); try { - if (pathPresent(marker)) applications.add(entry.name); + if (marksPendingRemoval(directory)) applications.add(entry.name); } catch (error) { - warn(marker, error); + warn(directory, error); } } } diff --git a/server/threads/manageThreads.ts b/server/threads/manageThreads.ts index ea95eecee3..8f36cebe6b 100644 --- a/server/threads/manageThreads.ts +++ b/server/threads/manageThreads.ts @@ -234,6 +234,7 @@ export { requestFromMainThread, answerMainThreadRequests, poolMayHold, + onWorkerRelease, failedClosedInMemory, setCanaryVerdictTimeout, setRootComponentsReload, @@ -1491,6 +1492,7 @@ function startWorker(path?: any, options: any = {}, startOptions: any = {}) { options.onRestartExhausted?.(worker); } } + notifyWorkerRelease(); }); workers.push(worker); startMonitoring(); @@ -1575,7 +1577,25 @@ function withPoolFor(application?: string) { } function recordLoadedApplications(worker: any, message: any) { - if (Array.isArray(message.applications)) worker.loadedApplications = new Set(message.applications); + if (!Array.isArray(message.applications)) return; + worker.loadedApplications = new Set(message.applications); + notifyWorkerRelease(); +} + +const workerReleaseListeners: Array<() => void> = []; +/** Told after a worker exits or reports what it loaded, the only events that narrow what a worker may hold. */ +function onWorkerRelease(listener: () => void) { + workerReleaseListeners.push(listener); +} + +function notifyWorkerRelease() { + for (const listener of workerReleaseListeners) { + try { + listener(); + } catch (error) { + harperLogger.error(error); + } + } } /** `withPool`: a named application's restart also replaces the pool workers, which may still hold that application. */ diff --git a/unitTests/components/branchRemovalRetry.test.js b/unitTests/components/branchRemovalRetry.test.js index 00523ab0d2..ededc85440 100644 --- a/unitTests/components/branchRemovalRetry.test.js +++ b/unitTests/components/branchRemovalRetry.test.js @@ -1,18 +1,18 @@ 'use strict'; const assert = require('node:assert'); -const { chmodSync, existsSync, mkdirSync, rmSync } = require('node:fs'); +const { chmodSync, existsSync, mkdirSync, readFileSync, rmSync, statSync, writeFileSync } = require('node:fs'); const path = require('node:path'); const testUtils = require('../testUtils.js'); testUtils.preTestPrep(); -const { getConfigPath } = require('#src/config/configUtils'); -const { CONFIG_PARAMS } = require('#src/utility/hdbTerms'); +const { getConfigFilePath, getConfigPath } = require('#src/config/configUtils'); +const { CONFIG_PARAMS, ITC_EVENT_TYPES, THREAD_TYPES } = require('#src/utility/hdbTerms'); const operations = require('#js/components/operations'); const { openBranchDatabase, table, resolveBranchPath } = require('#src/resources/databases'); const { getOrCreateBranch, hasPendingBranchRemoval } = require('#src/resources/branchDatabase'); -const { setMainIsWorker, getWorkerIndex } = require('#js/server/threads/manageThreads'); +const { setMainIsWorker, getWorkerIndex, startWorker } = require('#js/server/threads/manageThreads'); const { preserveRootConfig } = require('../rootConfigFixture.js'); const { waitFor } = require('../waitFor.js'); @@ -26,6 +26,23 @@ describe('a branch removal a drop could not finish (harper#3463)', function () { const forksOf = (app) => path.dirname(resolveBranchPath(BASE, app)); const releases = []; + /** A pool worker that reports nothing until told to. */ + function idleHttpWorker() { + // a start outside safe mode resolves threads.preload for the rest of the process + const safeMode = process.env.HARPER_SAFE_MODE; + process.env.HARPER_SAFE_MODE = '1'; + try { + return startWorker(path.join(__dirname, 'fixtures', 'idleWorker.js'), { + name: THREAD_TYPES.HTTP, + workerIndex: 1, + autoRestart: false, + }); + } finally { + if (safeMode === undefined) delete process.env.HARPER_SAFE_MODE; + else process.env.HARPER_SAFE_MODE = safeMode; + } + } + /** * A fork of `app`, as its worker left it, and a reference to it the drop does not own, standing in for a worker that * has not let go yet: the removal refuses before it touches anything, until that is released. @@ -130,4 +147,91 @@ describe('a branch removal a drop could not finish (harper#3463)', function () { await drop({ project: 'back-app' }); assert.strictEqual(existsSync(forksOf('back-app')), false); }); + + for (const [unreadable, spoil] of [ + ['does not parse', (configFile) => writeFileSync(configFile, 'config-app: [unclosed\n')], + ['cannot be read', (configFile) => chmodSync(configFile, 0o000)], + ]) { + it(`keeps retrying while the root config ${unreadable}`, async function () { + if (unreadable === 'cannot be read' && (process.platform === 'win32' || process.getuid?.() === 0)) { + return this.skip(); + } + operations.setBranchRemovalRetryDelayForTests?.(50); + const release = await heldFork('config-app'); + await assert.rejects(drop({ project: 'config-app' }), /open reference\(s\) to its store remain/); + const configFile = getConfigFilePath(); + const saved = readFileSync(configFile, 'utf8'); + const mode = statSync(configFile).mode & 0o777; + spoil(configFile); + try { + await release(); + await operations.retryPendingBranchRemovals(); + assert.ok(existsSync(forksOf('config-app')), 'a root config it cannot read might name it'); + } finally { + chmodSync(configFile, mode); + writeFileSync(configFile, saved); + } + + await waitFor(() => !existsSync(forksOf('config-app')), { + timeout: 10_000, + interval: 50, + message: 'the fork was left on disk', + }); + }); + } + + it('runs again as soon as a worker that may hold it reports it did not load it', async () => { + operations.setBranchRemovalRetryDelayForTests?.(60 * 60_000); + const branch = await getOrCreateBranch(BASE, 'reported-app'); + await branch.tables.Rows.put({ id: 'kept' }); + const worker = idleHttpWorker(); + try { + const dropped = await drop({ project: 'reported-app' }); + assert.match(dropped.message, /workers may still have it loaded/); + await operations.retryPendingBranchRemovals(); + assert.ok(existsSync(forksOf('reported-app')), 'a worker that has not reported may hold it'); + + const report = { type: ITC_EVENT_TYPES.CHILD_COMPONENT_VERDICT, components: [], applications: [] }; + worker.postMessage({ report }); + await waitFor(() => !existsSync(forksOf('reported-app')), { + timeout: 10_000, + interval: 50, + message: 'the fork waited for the timer', + }); + } finally { + await worker.terminate(); + } + }); + + it('runs again as soon as the worker that held it exits', async () => { + operations.setBranchRemovalRetryDelayForTests?.(60 * 60_000); + const branch = await getOrCreateBranch(BASE, 'pooled-app'); + await branch.tables.Rows.put({ id: 'kept' }); + const worker = idleHttpWorker(); + try { + const dropped = await drop({ project: 'pooled-app' }); + assert.match(dropped.message, /workers may still have it loaded/); + await operations.retryPendingBranchRemovals(); + assert.ok(existsSync(forksOf('pooled-app')), 'a worker that has not reported may hold it'); + } finally { + await worker.terminate(); + } + + await waitFor(() => !existsSync(forksOf('pooled-app')), { + timeout: 10_000, + interval: 50, + message: 'the fork waited for the timer', + }); + }); + + it('removes the forks and their directory when a crash cut the marker write short', async () => { + const release = await heldFork('debris-app'); + await assert.rejects(drop({ project: 'debris-app' }), /open reference\(s\) to its store remain/); + rmSync(path.join(forksOf('debris-app'), '`removal-pending`')); + writeFileSync(path.join(forksOf('debris-app'), '`removal-pending`.partial'), ''); + await release(); + await operations.retryPendingBranchRemovals(); + + assert.strictEqual(existsSync(forksOf('debris-app')), false); + }); }); diff --git a/unitTests/components/fixtures/idleWorker.js b/unitTests/components/fixtures/idleWorker.js new file mode 100644 index 0000000000..23942f652f --- /dev/null +++ b/unitTests/components/fixtures/idleWorker.js @@ -0,0 +1,7 @@ +'use strict'; + +const { parentPort } = require('node:worker_threads'); + +parentPort.on('message', (message) => { + if (message?.report) parentPort.postMessage(message.report); +});