diff --git a/core/autoflush.js b/core/autoflush.js new file mode 100644 index 0000000..acec5f7 --- /dev/null +++ b/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, + }; + }, + }; +} diff --git a/core/server.js b/core/server.js index 67590ea..c890e2f 100644 --- a/core/server.js +++ b/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(); diff --git a/package.json b/package.json index c03f1eb..c32f3e1 100644 --- a/package.json +++ b/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", diff --git a/test/autoflush.test.js b/test/autoflush.test.js new file mode 100644 index 0000000..582c9b9 --- /dev/null +++ b/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); +}); diff --git a/theseus/addon.json b/theseus/addon.json index 86112c4..1685801 100644 --- a/theseus/addon.json +++ b/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=", diff --git a/ui/app.js b/ui/app.js index 5fd16f8..aca97ee 100644 --- a/ui/app.js +++ b/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', [