From c63946db5698eb6b35706217bf28e6b9eb295a30 Mon Sep 17 00:00:00 2001 From: Local Dev Date: Tue, 6 Oct 2026 19:14:43 +0200 Subject: [PATCH] Pithos: stop s3d gently and never in the middle of an upload On Windows s3d was stopped with TerminateProcess. One killed during an upload (a Theseus restart, a cut-off flush) had deleted four buffered files without recording them as uploaded; every flush afterwards failed on them and the files were gone. A stop now waits while a flush Pithos started is running, or while s3d has begun an upload batch in the last two minutes (up to five minutes; twenty seconds when the host is quitting), then asks s3d to shut down cleanly: Ctrl+Break sent to its own hidden console on Windows (Windows disables Ctrl+C in a new process group; Go reads both as an interrupt), SIGTERM elsewhere. It is ended by force only if still there a minute later. Checked against s3d v0.2.0: 'shutdown complete', exit code 0. --- core/daemon.js | 47 +++++++++++++++++++++++++++++++++++++++++++---- core/server.js | 34 ++++++++++++++++++++++++++++++---- core/winctrl.js | 33 +++++++++++++++++++++++++++++++++ 3 files changed, 106 insertions(+), 8 deletions(-) create mode 100644 core/winctrl.js diff --git a/core/daemon.js b/core/daemon.js index 2c1c334..a1d6a17 100644 --- a/core/daemon.js +++ b/core/daemon.js @@ -4,6 +4,7 @@ import { spawn, execFile } from 'node:child_process'; import { EventEmitter } from 'node:events'; import path from 'node:path'; +import { sendBreak } from './winctrl.js'; const LOG_LINES = 2000; const ANSI = /\x1b\[[0-9;]*m/g; @@ -19,6 +20,8 @@ export class Daemon extends EventEmitter { this.startedAt = null; this.logs = []; this.seq = 0; + this.busy = 0; // flushes Pithos started and is waiting on + this.lastUploadAt = 0; // s3d's last "uploading object group" } env() { @@ -54,6 +57,7 @@ export class Daemon extends EventEmitter { // s3d logs "server started" once every listener is up. if (this.state === 'starting' && /server started/.test(line)) this.setState('running'); if (/No app key found/.test(line)) this.needsLogin = true; + if (/uploading object group/.test(line)) this.lastUploadAt = Date.now(); this.logs.push(entry); if (this.logs.length > LOG_LINES) this.logs.shift(); this.emit('log', entry); @@ -101,14 +105,49 @@ export class Daemon extends EventEmitter { return this.status(); } - async stop() { + // Work s3d must not be stopped in the middle of (Upload now, autoflush). + async whileBusy(fn) { + this.busy++; + try { return await fn(); } finally { this.busy--; } + } + + // True once no flush of ours is running and s3d has not started an upload + // batch for a while (it logs the start of one, not the end). + quiet() { return this.busy === 0 && Date.now() - this.lastUploadAt > 120_000; } + + async waitForQuiet(maxMs) { + const until = Date.now() + maxMs; + let told = false; + while (!this.quiet() && Date.now() < until && this.child) { + if (!told) { this.pushLog('sys', 'waiting for the upload in progress to finish before stopping s3d'); told = true; } + await new Promise((r) => setTimeout(r, 1000)); + } + return this.quiet(); + } + + // An s3d stopped in the middle of an upload has lost buffered files whose + // rows still said "waiting", so a stop first lets uploads finish, then + // asks s3d to shut down cleanly (Ctrl+Break on Windows, SIGTERM elsewhere) + // and only ends it by force if it is still there a minute later. + // waitMs: how long to wait for uploads (short when the host is quitting). + async stop({ waitMs = 5 * 60_000 } = {}) { const child = this.child; if (!child && this.pendingStart) { this.pendingStart = false; this.setState('stopped'); return this.status(); } if (!child) return this.status(); this.setState('stopping'); - const exited = new Promise((r) => child.once('exit', r)); - child.kill(); // SIGTERM on unix; TerminateProcess on Windows - const timer = setTimeout(() => { try { child.kill('SIGKILL'); } catch {} }, 10_000); + const exited = new Promise((r) => { if (this.child !== child) r(); else child.once('exit', r); }); + if (!(await this.waitForQuiet(waitMs)) && this.child === child) this.pushLog('sys', 'an upload is still running; stopping s3d anyway'); + if (this.child === child) { + let asked = false; + if (process.platform === 'win32') asked = await sendBreak(child.pid); + else { try { asked = child.kill('SIGTERM'); } catch {} } + if (!asked) { try { child.kill(); } catch {} } + } + const timer = setTimeout(() => { + if (this.child !== child) return; + this.pushLog('sys', 's3d did not stop within a minute; ending it'); + try { child.kill('SIGKILL'); } catch {} + }, 60_000); await exited; clearTimeout(timer); return this.status(); diff --git a/core/server.js b/core/server.js index 97b6a4e..7bf05d3 100644 --- a/core/server.js +++ b/core/server.js @@ -29,6 +29,7 @@ import { accountInfo, readConnection, fingerprint } from './sia-account.js'; import { guessType } from './hosted.js'; import { accountSpace } from './space.js'; import { createShareUrl } from './sia-share.js'; +import { stuckUploads } from './stuck.js'; const HERE = path.dirname(fileURLToPath(import.meta.url)); const UI_DIR = path.join(HERE, '..', 'ui'); @@ -73,6 +74,8 @@ export async function createPithos(opts = {}) { const configFile = resolveConfigPath(configOpt); const daemon = new Daemon({ configFile, binary: resolveBinary(binaryOpt, binDir) }); + // a flush s3d is in the middle of keeps any stop waiting (see Daemon.stop) + const flushNow = () => daemon.whileBusy(() => admin.flush(cfg())); const cli = makeCli(daemon); const login = new LoginSession(daemon, { registration: () => cli.registration(), @@ -366,7 +369,30 @@ export async function createPithos(opts = {}) { }); route('GET', '/api/stats', () => admin.uploadStats(cfg())); - route('POST', '/api/flush', async () => { await admin.flush(cfg()); return { ok: true }; }); + route('POST', '/api/flush', async () => { + try { await flushNow(); } catch (e) { + if (/failed to open upload|cannot find the file|no such file/i.test(e.message)) { + throw new HttpError(409, 'Some files waiting for Sia lost their copy on this computer, so the upload stopped. Pithos lists them on the Drives page: upload them again or remove them.'); + } + throw e; + } + return { ok: true }; + }); + + // Waiting files whose buffered copy is gone (see core/stuck.js). + route('GET', '/api/uploads/stuck', async () => { + if (hosted.isHosted()) return { items: [] }; + if (!daemon.quiet()) return { items: [], busy: true }; + return { items: await stuckUploads(cfg().directory) }; + }); + route('POST', '/api/uploads/stuck/remove', async (req) => { + const { bucket, key } = await jsonBody(req); + const item = (await stuckUploads(cfg().directory)).find((i) => i.bucket === bucket && i.key === key); + if (!item) throw new HttpError(409, 'This file is no longer stuck (it may have been uploaded again).'); + if (!item.user) throw new HttpError(409, 'No S3 user owns this drive.'); + await (await s3For(item.user)).deleteObject(bucket, key); + return { ok: true }; + }); const autoFlush = createAutoFlush({ settings: () => { @@ -378,7 +404,7 @@ export async function createPithos(opts = {}) { // Only an s3d that is up can be asked; a stopped one just means no flush. stats: () => (daemon.state === 'starting' || daemon.state === 'stopping' ? Promise.reject(new Error('busy')) : admin.uploadStats(cfg())), - flush: () => admin.flush(cfg()), + flush: () => flushNow(), log: (line) => daemon.pushLog('sys', line), }); route('GET', '/api/autoflush', () => autoFlush.status()); @@ -511,7 +537,7 @@ export async function createPithos(opts = {}) { try { return await createShareUrl(dir, bucket, key, until); } catch (e) { if (e.code !== 'pending') throw new HttpError(e.status || 502, e.message); - await admin.flush(cfg()); // send what is waiting, then try once more + await flushNow(); // send what is waiting, then try once more try { return await createShareUrl(dir, bucket, key, until); } catch (e2) { throw new HttpError(e2.status || 502, e2.code === 'pending' ? 'This file is still uploading to Sia. Try again in a minute.' : e2.message); } } @@ -949,7 +975,7 @@ export async function createPithos(opts = {}) { login.cancel(); for (const s of subscribers) s.end(); handoff?.close(); - await daemon.stop(); + await daemon.stop({ waitMs: 20_000 }); // the host is quitting: a short wait, then a clean stop // Idle keep-alive sockets would hold close() for seconds; nobody is left to answer. const closed = new Promise((r) => server.close(r)); server.closeAllConnections?.(); diff --git a/core/winctrl.js b/core/winctrl.js new file mode 100644 index 0000000..5eaac0a --- /dev/null +++ b/core/winctrl.js @@ -0,0 +1,33 @@ +// A gentle stop for s3d on Windows. child.kill() there is TerminateProcess: +// s3d dies wherever it is, and one stopped in the middle of an upload had +// lost buffered files whose database rows still said "waiting". s3d shuts +// down cleanly on an interrupt, so this attaches a short PowerShell to s3d's +// own (hidden) console and sends Ctrl+Break there. +// +// Ctrl+Break, not Ctrl+C: Windows disables Ctrl+C for a process started in +// a new process group, and Go reads both as os.Interrupt. A spawned console +// program gets a console of its own (Node passes CREATE_NO_WINDOW), so the +// event reaches s3d and not Pithos. It ends the PowerShell too, which is +// why only "could not attach" (2) and "could not send" (3) count as failure. + +import { execFile } from 'node:child_process'; + +const SCRIPT = (pid) => [ + "Add-Type -Name K -Namespace PithosCtrl -MemberDefinition '", + '[DllImport("kernel32.dll")] public static extern bool FreeConsole();', + '[DllImport("kernel32.dll")] public static extern bool AttachConsole(uint pid);', + '[DllImport("kernel32.dll")] public static extern bool GenerateConsoleCtrlEvent(uint ev, uint group);', + "';", + '[PithosCtrl.K]::FreeConsole() | Out-Null;', + `if (-not [PithosCtrl.K]::AttachConsole(${Number(pid) >>> 0})) { exit 2 };`, + 'if ([PithosCtrl.K]::GenerateConsoleCtrlEvent(1, 0)) { exit 0 } else { exit 3 }', +].join(' '); + +// Resolves true when Ctrl+Break was sent to the console of `pid`. +export function sendBreak(pid) { + if (process.platform !== 'win32' || !pid) return Promise.resolve(false); + return new Promise((resolve) => { + execFile('powershell.exe', ['-NoLogo', '-NoProfile', '-NonInteractive', '-ExecutionPolicy', 'Bypass', '-Command', SCRIPT(pid)], + { windowsHide: true, timeout: 20_000 }, (err) => resolve(!err || (err.code !== 2 && err.code !== 3 && !err.killed))); + }); +}