Pithos 0.3.7: optional automatic upload of leftovers

s3d only sends full ~40 MB blocks to Sia, so a handful of small files can
stay in the local buffer forever (seen: 63 objects, 6 MiB, nothing on Sia).
An opt-in setting now flushes once the pending set has not changed for N
minutes, judged from s3d's upload stats so uploads from any S3 client count.
Failed flushes back off up to an hour instead of hammering a broken indexer.

core/autoflush.js

@@ -0,0 +1,81 @@
+// Optional automatic flush of leftovers. s3d packs uploads into full slabs
+// (about 40 MB) before sending them to Sia, so a small batch can sit in the
+// local buffer indefinitely. When the user opts in, this watches s3d's upload
+// stats and flushes once the pending set has stopped changing for N minutes,
+// i.e. nobody is still uploading into it.
+//
+// The watcher only sees stats, so it also catches uploads made by other S3
+// clients, not just the ones that went through Pithos.
+
+export const DEFAULT_MINUTES = 10;
+export const MIN_MINUTES = 1;
+export const MAX_MINUTES = 24 * 60;
+
+export function clampMinutes(m) {
+  const n = Math.round(Number(m));
+  if (!Number.isFinite(n)) return DEFAULT_MINUTES;
+  return Math.min(MAX_MINUTES, Math.max(MIN_MINUTES, n));
+}
+
+// settings: () => ({ enabled, minutes }); stats: async () => /stats/uploads;
+// flush: async () => void; log: (line) => void.
+export function createAutoFlush({ settings, stats, flush, log = () => {}, now = Date.now }) {
+  let signature = null; // what was pending last time we looked
+  let quietSince = 0;
+  let flushing = false;
+  let lastFlushAt = null;
+  let lastError = null;
+  let failures = 0; // in a row; each one doubles the wait, up to an hour
+
+  function reset() { signature = null; quietSince = 0; }
+
+  function waitMs(minutes) {
+    const base = clampMinutes(minutes) * 60_000;
+    return failures ? Math.max(base, Math.min(60 * 60_000, base * 2 ** failures)) : base;
+  }
+
+  async function tick() {
+    const { enabled, minutes } = settings();
+    if (!enabled || flushing) { if (!enabled) reset(); return; }
+    let x;
+    try { x = await stats(); } catch { reset(); return; } // s3d not reachable
+    if (!x.pendingObjects) { reset(); return; }
+    const sig = `${x.pendingObjects}:${x.pendingSize}`;
+    if (sig !== signature) { signature = sig; quietSince = now(); return; }
+    if (now() - quietSince < waitMs(minutes)) return;
+    flushing = true;
+    log(`uploading ${x.pendingObjects} leftover object(s) to Sia after ${clampMinutes(minutes)} min with no new uploads`);
+    try {
+      await flush();
+      lastFlushAt = now();
+      lastError = null;
+      failures = 0;
+      log('leftovers are on Sia');
+      reset();
+    } catch (e) {
+      lastError = e.message;
+      failures++;
+      log(`automatic flush failed: ${e.message}`);
+      quietSince = now(); // a broken indexer is retried less and less often
+    } finally {
+      flushing = false;
+    }
+  }
+
+  return {
+    tick,
+    status() {
+      const { enabled, minutes } = settings();
+      const m = clampMinutes(minutes);
+      return {
+        enabled: !!enabled,
+        minutes: m,
+        flushing,
+        lastFlushAt,
+        lastError,
+        // When the next automatic flush happens if nothing new arrives.
+        dueAt: enabled && signature && !flushing ? quietSince + waitMs(m) : null,
+      };
+    },
+  };
+}

core/server.js

@@ -23,6 +23,7 @@ import * as admin from './admin.js';
 import { installS3d, S3D_VERSION, assetForPlatform } from './binary.js';
 import * as accounts from './accounts.js';
 import { readPrefs, writePrefs } from './prefs.js';
+import { createAutoFlush, clampMinutes, DEFAULT_MINUTES } from './autoflush.js';
 
 const HERE = path.dirname(fileURLToPath(import.meta.url));
 const UI_DIR = path.join(HERE, '..', 'ui');
@@ -296,6 +297,27 @@ export async function createPithos(opts = {}) {
   route('GET', '/api/stats', () => admin.uploadStats(cfg()));
   route('POST', '/api/flush', async () => { await admin.flush(cfg()); return { ok: true }; });
 
+  const autoFlush = createAutoFlush({
+    settings: () => {
+      const p = readPrefs(configFile);
+      return { enabled: p.autoFlush === true, minutes: p.autoFlushMinutes ?? DEFAULT_MINUTES };
+    },
+    // 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()),
+    log: (line) => daemon.pushLog('sys', line),
+  });
+  route('GET', '/api/autoflush', () => autoFlush.status());
+  route('PUT', '/api/autoflush', async (req) => {
+    const body = await jsonBody(req);
+    const patch = {};
+    if (body.enabled !== undefined) patch.autoFlush = body.enabled === true;
+    if (body.minutes !== undefined) patch.autoFlushMinutes = clampMinutes(body.minutes);
+    writePrefs(configFile, patch);
+    return autoFlush.status();
+  });
+
   // Buckets & objects, acting as one s3d user.
   route('GET', '/api/s3/:user/buckets', async (req, url, p) => (await s3For(p.user)).listBuckets());
   route('POST', '/api/s3/:user/buckets', async (req, url, p) => {
@@ -532,6 +554,8 @@ export async function createPithos(opts = {}) {
     setTimeout(() => autostart('it stopped unexpectedly'), 5000).unref?.();
   });
   setTimeout(() => autostart('it was running before'), 500).unref?.();
+  const autoFlushTimer = setInterval(() => { autoFlush.tick(); }, 30_000);
+  autoFlushTimer.unref?.();
   const displayHost = host === '0.0.0.0' || host === '::' ? '127.0.0.1' : host;
   const baseUrl = `http://${displayHost.includes(':') ? `[${displayHost}]` : displayHost}:${boundPort}/`;
 
@@ -565,6 +589,7 @@ export async function createPithos(opts = {}) {
       return `${baseUrl}dl/${id}`;
     },
     async close() {
+      clearInterval(autoFlushTimer);
       login.cancel();
       for (const s of subscribers) s.end();
       await daemon.stop();

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pithos",
-  "version": "0.3.6",
+  "version": "0.3.7",
   "description": "A control panel for s3d, the Sia S3 gateway: daemon, users and keys, buckets, upload pipeline.",
   "license": "MPL-2.0",
   "type": "module",

test/autoflush.test.js

@@ -0,0 +1,88 @@
+// The leftover-flush watcher, driven with a fake clock and fake s3d stats.
+
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { createAutoFlush, clampMinutes } from '../core/autoflush.js';
+
+function rig({ enabled = true, minutes = 5 } = {}) {
+  const r = { t: 0, pending: { pendingObjects: 0, pendingSize: 0 }, flushes: 0, fail: null, prefs: { enabled, minutes }, down: false };
+  r.af = createAutoFlush({
+    settings: () => r.prefs,
+    stats: async () => { if (r.down) throw new Error('refused'); return { ...r.pending }; },
+    flush: async () => {
+      if (r.fail) throw new Error(r.fail);
+      r.flushes++;
+      r.pending = { pendingObjects: 0, pendingSize: 0 };
+    },
+    now: () => r.t,
+  });
+  r.at = async (min) => { r.t = min * 60_000; await r.af.tick(); };
+  return r;
+}
+
+test('flushes once the pending set has been unchanged for N minutes', async () => {
+  const r = rig({ minutes: 5 });
+  r.pending = { pendingObjects: 3, pendingSize: 1000 };
+  await r.at(0);
+  await r.at(4.9);
+  assert.equal(r.flushes, 0);
+  assert.equal(r.af.status().dueAt, 5 * 60_000);
+  await r.at(5);
+  assert.equal(r.flushes, 1);
+  assert.equal(r.af.status().lastFlushAt, 5 * 60_000);
+  await r.at(20);
+  assert.equal(r.flushes, 1, 'nothing pending, nothing to flush');
+});
+
+test('a new upload restarts the quiet period', async () => {
+  const r = rig({ minutes: 5 });
+  r.pending = { pendingObjects: 1, pendingSize: 10 };
+  await r.at(0);
+  await r.at(4);
+  r.pending = { pendingObjects: 2, pendingSize: 20 };
+  await r.at(4.5);
+  await r.at(9);
+  assert.equal(r.flushes, 0);
+  await r.at(9.5);
+  assert.equal(r.flushes, 1);
+});
+
+test('off means never, and s3d being down is not an error', async () => {
+  const r = rig({ enabled: false });
+  r.pending = { pendingObjects: 5, pendingSize: 50 };
+  for (const m of [0, 10, 100]) await r.at(m);
+  assert.equal(r.flushes, 0);
+  assert.equal(r.af.status().dueAt, null);
+  r.prefs.enabled = true;
+  r.down = true;
+  for (const m of [101, 200]) await r.at(m);
+  assert.equal(r.flushes, 0);
+});
+
+test('failed flushes back off, doubling up to an hour', async () => {
+  const r = rig({ minutes: 2 });
+  r.pending = { pendingObjects: 1, pendingSize: 1 };
+  r.fail = 'indexer unreachable';
+  await r.at(0);
+  await r.at(2); // fails; next try 4 min later
+  assert.equal(r.af.status().lastError, 'indexer unreachable');
+  assert.equal(r.af.status().dueAt, 6 * 60_000);
+  await r.at(6); // fails again; next try 8 min later
+  assert.equal(r.af.status().dueAt, 14 * 60_000);
+  for (let i = 0; i < 6; i++) await r.at(r.af.status().dueAt / 60_000);
+  assert.equal(r.af.status().dueAt - r.t, 60 * 60_000, 'capped at an hour');
+  r.fail = null;
+  await r.at(r.af.status().dueAt / 60_000);
+  assert.equal(r.flushes, 1);
+  assert.equal(r.af.status().lastError, null);
+  r.pending = { pendingObjects: 1, pendingSize: 2 };
+  await r.at(r.t / 60_000);
+  assert.equal(r.af.status().dueAt - r.t, 2 * 60_000, 'a success resets the backoff');
+});
+
+test('minutes are clamped to a sane range', () => {
+  assert.equal(clampMinutes(0), 1);
+  assert.equal(clampMinutes(99999), 1440);
+  assert.equal(clampMinutes('abc'), 10);
+  assert.equal(clampMinutes('7'), 7);
+});

theseus/addon.json

@@ -1,7 +1,7 @@
 {
   "id": "pithos",
   "name": "Pithos",
-  "version": "0.3.6",
+  "version": "0.3.7",
   "description": "Run s3d, the Sia S3 gateway, from the Theseus sidebar: connect it to a Sia indexer, create S3 users and access keys, browse and share buckets, and watch uploads reach Sia.",
   "author": "Silent Mode",
   "icon": "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCA2NCA2NCI+PGRlZnM+PGxpbmVhckdyYWRpZW50IGlkPSJnIiB4MT0iMCIgeTE9IjAiIHgyPSIwIiB5Mj0iMSI+PHN0b3Agb2Zmc2V0PSIwIiBzdG9wLWNvbG9yPSIjM2RkYzk3Ii8+PHN0b3Agb2Zmc2V0PSIxIiBzdG9wLWNvbG9yPSIjMWU4ZjZhIi8+PC9saW5lYXJHcmFkaWVudD48L2RlZnM+PHBhdGggZD0iTTIyIDZoMjB2NWMwIDItMiAzLTIgNSA4IDMgMTQgMTEgMTQgMjEgMCAxMi05IDIxLTIyIDIxUzEwIDQ5IDEwIDM3YzAtMTAgNi0xOCAxNC0yMSAwLTItMi0zLTItNXoiIGZpbGw9InVybCgjZykiLz48cGF0aCBkPSJNMTQgMzBoMzZNMTMgNDBoMzgiIHN0cm9rZT0iIzBiMWYxOCIgc3Ryb2tlLXdpZHRoPSIzIiBzdHJva2UtbGluZWNhcD0icm91bmQiIG9wYWNpdHk9Ii41NSIvPjwvc3ZnPgo=",

ui/app.js

@@ -253,6 +253,7 @@ function overview(main) {
     }
     try {
       const x = await api('GET', '/api/stats');
+      const af = await api('GET', '/api/autoflush').catch(() => null);
       const stat = (v, l, bad) => h('div', { class: `stat ${bad ? 'bad' : ''}` }, h('div', { class: 'v' }, v), h('div', { class: 'l' }, l));
       put(statsCard, 
         h('div', { class: 'row spread' }, h('h2', {}, 'Upload pipeline'),
@@ -266,12 +267,24 @@ function overview(main) {
           stat(x.orphanedObjects, 'orphaned', x.orphanedObjects > 0),
         ),
         h('p', { class: 'small muted', style: 'margin:12px 0 0' }, 'New objects wait in a local buffer and are packed into full slabs before upload, which keeps wasted space down.'),
+        af && h('p', { class: 'small muted', style: 'margin:6px 0 0' }, autoFlushLine(af, x), ' ', h('a', { href: '#/settings' }, af.enabled ? 'Change' : 'Turn on')),
       );
     } catch (e) {
       put(statsCard, h('h2', {}, 'Upload pipeline'), h('p', { class: 'error' }, e.message));
     }
   }
 
+  function autoFlushLine(af, x) {
+    if (!af.enabled) return 'Leftovers upload only when a block fills up or you flush.';
+    if (af.flushing) return 'Uploading leftovers now…';
+    if (af.lastError && x.pendingObjects) return `The last automatic upload failed (${af.lastError}). Pithos will try again later.`;
+    if (x.pendingObjects && af.dueAt) {
+      const min = Math.max(1, Math.ceil((af.dueAt - Date.now()) / 60_000));
+      return `Leftovers upload automatically in about ${min} min if nothing new arrives.`;
+    }
+    return `Leftovers upload automatically after ${af.minutes} min with no new uploads.`;
+  }
+
   async function flushNow(ev) {
     const btn = ev.currentTarget;
     btn.disabled = true;
@@ -1385,7 +1398,30 @@ function objectBrowser(head, body, user, bucket, prefix, userSel) {
 
 function settings(main) {
   const card = h('div', { class: 'card' }, h('p', { class: 'muted' }, 'Loading…'));
-  main.append(h('div', { class: 'page-head' }, h('div', {}, h('h1', {}, 'Settings'), h('p', { class: 'muted' }, 'Edits s3d.yml. s3d reads it at start, so a restart applies the changes.'))), card);
+  const flushCard = h('div', { class: 'card', style: 'margin-top:16px' });
+  main.append(h('div', { class: 'page-head' }, h('div', {}, h('h1', {}, 'Settings'), h('p', { class: 'muted' }, 'Edits s3d.yml. s3d reads it at start, so a restart applies the changes.'))), card, flushCard);
+
+  // Kept by Pithos itself (pithos-prefs.json), applies at once, no restart.
+  (async () => {
+    let af;
+    try { af = await api('GET', '/api/autoflush'); } catch (e) { put(flushCard, h('p', { class: 'error' }, e.message)); return; }
+    const on = h('input', { type: 'checkbox', checked: af.enabled });
+    const minutes = h('input', { type: 'number', min: 1, max: 1440, value: af.minutes, style: 'width:6em', disabled: !af.enabled });
+    on.addEventListener('change', () => { minutes.disabled = !on.checked; });
+    put(flushCard,
+      h('h2', {}, 'Leftover uploads'),
+      h('p', { class: 'small muted' }, 's3d packs uploads into blocks of about 40 MB before sending them to Sia, so a few small files can wait in the local buffer indefinitely. Pithos can upload them for you once uploading has stopped.'),
+      h('label', { class: 'check' }, on, 'Upload leftovers automatically'),
+      h('label', { class: 'field' }, h('span', {}, 'after this many minutes with no new uploads'), minutes,
+        h('small', {}, 'Partly filled blocks use a little more Sia space than full ones. Pithos must be open for this to run.')),
+      h('div', { class: 'row' }, h('button', { class: 'btn primary', onclick: saveFlush }, 'Save')));
+    async function saveFlush() {
+      try {
+        await api('PUT', '/api/autoflush', { enabled: on.checked, minutes: Number(minutes.value) || 10 });
+        toast(on.checked ? 'Leftovers will upload automatically' : 'Automatic upload is off');
+      } catch (e) { fail(e); }
+    }
+  })();
 
   const fields = [
     ['Network', [