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

@@ -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();

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?.();

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)));
+  });
+}