// One s3d per hosted account, each wrapped in the same createPithos() the // desktop runs, so the server speaks exactly the /api the Pithos UI already // knows. Accounts live under /users//: // // account.json who (public key), how they got in (gate), port slot, state // s3d.yml loopback-only addresses from the slot, random admin password // data/ the s3d data folder moved here from the user's PC // rules.json public folders, links and shared keys // usage.json bytes served this month // // An s3d starts on the account's first request and is stopped after // `idleMinutes` without one, flushing leftovers to Sia first. At most // `maxRunning` run at once; the least recently used idle one makes room. import fs from 'node:fs'; import path from 'node:path'; import { createPithos } from '../core/server.js'; import { makeCli } from '../core/cli.js'; import { readConfig, writeConfig, randomPassword } from '../core/config.js'; import { writePrefs } from '../core/prefs.js'; import { S3Client } from '../core/s3.js'; import * as admin from '../core/admin.js'; import { readJson, writeJson } from './store.js'; // reservedKey(akid): true for key ids the gateway itself hands out (grants), // which no account's s3d may also use. export function createSupervisor(cfg, log = console.log, { reservedKey = () => false } = {}) { const usersDir = path.join(cfg.root, 'users'); const slotsFile = path.join(cfg.root, 'slots.json'); const keysFile = path.join(cfg.root, 'keys.json'); const running = new Map(); // id -> instance const loading = new Map(); // id -> Promise const closing = new Map(); // id -> Promise, while an s3d is being stopped const keyIndex = readJson(keysFile, {}); // accessKeyId -> account id (owners' s3d keys) const dir = (id) => path.join(usersDir, id); const accountFile = (id) => path.join(dir(id), 'account.json'); const configFile = (id) => path.join(dir(id), 's3d.yml'); const dataDir = (id) => path.join(dir(id), 'data'); function account(id) { if (!/^[0-9a-f]{32}$/.test(id)) return null; return readJson(accountFile(id), null); } function saveAccount(a) { writeJson(accountFile(a.id), a); nameIndex = null; } // Accounts by the BNS name they signed up with (one slot per name), for // public addresses like /p/alice.x//… instead of the 32-hex id. let nameIndex = null; function accountByName(name) { if (!nameIndex) { nameIndex = new Map(); let ids = []; try { ids = fs.readdirSync(usersDir); } catch {} for (const id of ids) { const a = account(id); if (a?.gate?.type === 'bns' && a.gate.name) nameIndex.set(String(a.gate.name).toLowerCase(), id); } } return nameIndex.get(String(name).toLowerCase()) || null; } function createAccount({ id, pubkey, gate }) { const slots = readJson(slotsFile, { next: 0 }); const slot = slots.next++; writeJson(slotsFile, slots); const a = { id, pubkey, gate, slot, state: 'empty', createdAt: Date.now() }; fs.mkdirSync(dataDir(id), { recursive: true }); const port = cfg.portBase + slot * 2; writeConfig(configFile(id), { apiAddress: `127.0.0.1:${port}`, adminAddress: `127.0.0.1:${port + 1}`, adminPassword: randomPassword(), directory: dataDir(id), 'sia.diskUsageLimit': cfg.bufferLimitBytes, }); // Leftovers always go to Sia in hosted mode: an idle s3d gets stopped. writePrefs(configFile(id), { autostart: false, autoFlush: true, autoFlushMinutes: 10 }); saveAccount(a); return a; } function setState(id, state) { const a = account(id); a.state = state; saveAccount(a); writePrefs(configFile(id), { autostart: state === 'active' }); return a; } async function load(id) { // Never start a second s3d on a folder whose first one is still stopping. while (closing.has(id)) await closing.get(id); if (running.has(id)) return running.get(id); if (loading.has(id)) return loading.get(id); const p = (async () => { await makeRoom(); const pithos = await createPithos({ host: '127.0.0.1', port: 0, hostName: 'hosted', configFile: configFile(id), binary: cfg.s3dBinary, }); const inst = { id, pithos, cli: makeCli(pithos.daemon), cookie: pithos.internalSession(), lastUsed: Date.now(), busy: 0, buckets: new Map() }; running.set(id, inst); pithos.daemon.on('log', (l) => { if (l.stream !== 'out') log(`[${id.slice(0, 8)}] ${l.line}`); }); return inst; })().finally(() => loading.delete(id)); loading.set(id, p); return p; } // The instance with s3d up, for anything that talks S3. async function ensure(id) { const a = account(id); if (!a) throw httpError(404, 'no such account'); if (a.state === 'empty') throw httpError(409, 'nothing is hosted yet: move your drives here from Pithos first'); if (a.state === 'moving-out') throw httpError(409, 'this account is being moved to its owner\'s computer'); const inst = await load(id); // The account may have started moving out while it loaded. if (account(id).state !== 'active') throw httpError(409, 'this account is being moved'); inst.lastUsed = Date.now(); const d = inst.pithos.daemon; if (d.state !== 'running') { if (d.state !== 'starting') d.start(); const until = Date.now() + 30_000; while (d.state === 'starting' && Date.now() < until) await sleep(200); if (d.state !== 'running') throw httpError(503, 'your s3d did not start; try again in a minute'); await indexKeys(inst); } return inst; } // Routes S3 requests: access key id -> account. A key id another account // (or a grant) already uses is never taken over; that account's traffic // would otherwise land in this one's s3d. Such keys are left unrouted and // reported in inst.conflicts. async function indexKeys(inst) { const keys = await inst.cli.listKeys().catch(() => []); let changed = false; for (const [k, v] of Object.entries(keyIndex)) if (v === inst.id && !keys.some((x) => x.accessKeyId === k)) { delete keyIndex[k]; changed = true; } const conflicts = []; for (const k of keys) { const owner = Object.hasOwn(keyIndex, k.accessKeyId) ? keyIndex[k.accessKeyId] : null; if (owner === inst.id) continue; if (owner || reservedKey(k.accessKeyId)) { conflicts.push(k.accessKeyId); continue; } keyIndex[k.accessKeyId] = inst.id; changed = true; } if (changed) writeJson(keysFile, keyIndex); if (conflicts.length) log(`[${inst.id.slice(0, 8)}] access keys already used elsewhere, not routed: ${conflicts.join(', ')}`); inst.keys = keys.filter((k) => !conflicts.includes(k.accessKeyId)); inst.conflicts = conflicts; inst.buckets.clear(); return inst.keys; } const keyTaken = (akid, id) => (Object.hasOwn(keyIndex, akid) && keyIndex[akid] !== id) || reservedKey(akid); // Every drive on the account, mapped to a key of the s3d user owning it. async function mapBuckets(inst) { inst.buckets.clear(); for (const k of inst.keys || await indexKeys(inst)) { const list = await clientFor(inst, k).listBuckets().catch(() => []); for (const b of list) if (!inst.buckets.has(b.name)) inst.buckets.set(b.name, k); } return [...inst.buckets.keys()]; } // An owner key that can act on `bucket`. async function bucketClient(inst, bucket) { if (!inst.buckets.has(bucket)) await mapBuckets(inst); const k = inst.buckets.get(bucket); return k ? clientFor(inst, k) : null; } function clientFor(inst, k) { return new S3Client({ endpoint: `http://${readConfig(configFile(inst.id)).apiAddress}`, accessKeyId: k.accessKeyId, secretKey: k.secretKey }); } async function close(id, reason) { if (closing.has(id)) return closing.get(id); const inst = running.get(id); if (!inst) return; const p = (async () => { log(`[${id.slice(0, 8)}] stopping s3d (${reason})`); await inst.pithos.close().catch((e) => log(`[${id.slice(0, 8)}] close failed: ${e.message}`)); running.delete(id); })().finally(() => closing.delete(id)); closing.set(id, p); return p; } async function flushQuietly(inst, timeoutMs = 120_000) { const c = readConfig(configFile(inst.id)); try { const st = await admin.uploadStats(c); if (!st.pendingObjects) return true; await Promise.race([admin.flush(c), sleep(timeoutMs).then(() => { throw new Error('timed out'); })]); return true; } catch (e) { log(`[${inst.id.slice(0, 8)}] flush before stop failed: ${e.message}`); return false; } } async function makeRoom() { // Loads in flight count too, or parallel first requests overshoot the cap. if (running.size + loading.size < cfg.maxRunning) return; const idle = [...running.values()].filter((i) => !i.busy).sort((a, b) => a.lastUsed - b.lastUsed); if (!idle.length) throw httpError(503, 'the server is busy; try again in a minute'); await flushQuietly(idle[0]); await close(idle[0].id, 'making room'); } async function sweep() { const cutoff = Date.now() - cfg.idleMinutes * 60_000; for (const inst of [...running.values()]) { if (inst.busy || inst.lastUsed > cutoff || closing.has(inst.id)) continue; await flushQuietly(inst); if (!inst.busy && inst.lastUsed <= cutoff) await close(inst.id, 'idle'); } } const sweeper = setInterval(() => { sweep().catch((e) => log('sweep failed:', e.message)); }, 60_000); sweeper.unref?.(); // Holds an instance "in use" for the duration of a streamed request, so the // sweeper does not stop s3d under a long download. function hold(inst) { inst.busy++; let done = false; return () => { if (!done) { done = true; inst.busy--; inst.lastUsed = Date.now(); } }; } return { dir, dataDir, configFile, account, accountByName, createAccount, saveAccount, setState, load, ensure, close, flushQuietly, bucketClient, mapBuckets, indexKeys, hold, keyTaken, ownerOfKey: (akid) => (Object.hasOwn(keyIndex, akid) ? keyIndex[akid] : null), async closeAll() { clearInterval(sweeper); for (const id of [...running.keys()]) { await flushQuietly(running.get(id), 30_000); await close(id, 'shutdown'); } }, runningCount: () => running.size, }; } export function httpError(status, message) { return Object.assign(new Error(message), { status }); } const sleep = (ms) => new Promise((r) => setTimeout(r, ms));