--- a/server/p2p-hub.js +++ b/server/p2p-hub.js @@ -118,3 +118,28 @@ } return { ok: true, staleDays, cids: out }; } + +// Flow 8 (plan 018): the source is gone and the server evicted its copy, so +// ask ONE online holder that shares to upload it through intake +// ({type:'upload-request', videoId, cid}). At most one request per video per +// 10 minutes; returns true when a device was asked (or recently was). +export function createRehydrator({ p2pDb, hub, hasServerCopy, enabled = () => true, now = () => Date.now() }) { + const asked = new Map(); // videoId -> ms + return async function rehydrate(videoId) { + if (!enabled()) return false; + if (await hasServerCopy(videoId)) return false; + const last = asked.get(videoId); + if (last && now() - last < 10 * 60_000) return true; + for (const c of await p2pDb.listContentForVideo(videoId)) { + for (const h of await p2pDb.listHolders(c.cid, 50)) { + if (Number(h.share) !== 1 || !hub.isOnline(h.device_id)) continue; + if (hub.send(h.device_id, { type: 'upload-request', videoId, cid: c.cid })) { + asked.set(videoId, now()); + if (asked.size > 5000) asked.delete(asked.keys().next().value); + return true; + } + } + } + return false; + }; +} --- a/server/p2p-hub.test.js +++ b/server/p2p-hub.test.js @@ -85,3 +85,21 @@ ]); expect(JSON.stringify(p)).not.toContain('dev_'); // never leak device ids }); + +test('rehydrator asks one online sharing holder, once per 10 min, only when the server has no copy', async () => { + const { createRehydrator } = await import('./p2p-hub.js'); + const sent = []; + let t = NOW; + let serverHas = false; + const hub = { isOnline: (d) => d === 'dev_000000000000000a' || d === 'dev_000000000000000c', send: (d, m) => { sent.push([d, m]); return true; } }; + const rehydrate = createRehydrator({ p2pDb, hub, hasServerCopy: async () => serverHas, now: () => t }); + expect(await rehydrate('dQw4w9WgXcQ')).toBe(true); + // c is online but has sharing off; a is online and shares. + expect(sent).toEqual([['dev_000000000000000a', { type: 'upload-request', videoId: 'dQw4w9WgXcQ', cid: CID }]]); + expect(await rehydrate('dQw4w9WgXcQ')).toBe(true); // recently asked — no second message + expect(sent.length).toBe(1); + t += 11 * 60_000; + serverHas = true; + expect(await rehydrate('dQw4w9WgXcQ')).toBe(false); // server copy is back + expect(await rehydrate('unknownVid1')).toBe(false); // nobody holds it +}); --- a/server/p2p-intake.js +++ b/server/p2p-intake.js @@ -3,8 +3,10 @@ * (docs/p2p-architecture.md flow 7). The ONLY path by which bytes that did * not come from the server's own fetch can become verified content. * - * POST /api/p2p/intake (device) { videoId, cid?, size, title?, channel? } + * POST /api/p2p/intake (device) { videoId, cid?, size, title?, channel?, restore? } * → { ok, known:true } cid already verified — nothing to send + * (unless restore:true and the server + * lost its copy — plan 018) * → { ok, ticket, url, expiresAt } PUT the bytes to url within 30 min * PUT /api/p2p/intake/:ticket (device) raw file body * → { ok, cid, adopted } admitted (+ adopted into the media cache) @@ -30,6 +32,7 @@ export function registerIntakeRoutes(app, deps) { const { cfg, p2pDb, gate, requireDevice, validateMedia, admitFile, adopt = null, + serverHasCid = async () => true, // (cid) → does the server still hold these bytes? now = () => Date.now(), log = console, } = deps; const tickets = new Map(); // ticket -> { deviceId, videoId, cid, size, expiresAt, busy } @@ -48,7 +51,8 @@ if (!(size > 0) || size > cfg.intakeMaxBytes) return c.json({ ok: false, error: 'bad size' }, 413); if (cid) { const known = await p2pDb.getContent(cid); - if (known && known.status === 'verified') return c.json({ ok: true, known: true }); + const restore = body.restore === true && known && known.status === 'verified' && !(await serverHasCid(cid)); + if (known && known.status === 'verified' && !restore) return c.json({ ok: true, known: true }); if (known && known.status === 'revoked') return c.json({ ok: false, error: 'revoked' }, 410); } sweep(); --- a/server/p2p-intake.test.js +++ b/server/p2p-intake.test.js @@ -1,6 +1,6 @@ // Device → server intake: hash, validate, (scan), admit (plan 016). Needs ffmpeg. import { test, expect, beforeAll } from 'bun:test'; -import { mkdtempSync, readFileSync, readdirSync } from 'node:fs'; +import { mkdtempSync, readFileSync, readdirSync, unlinkSync } from 'node:fs'; import { spawnSync } from 'node:child_process'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -64,6 +64,27 @@ expect(await (await req('/api/p2p/intake', { method: 'POST', body: { videoId: 'upAAAAAAAA1', cid: goodCid, size: good.length } })).json()).toEqual({ ok: true, known: true }); }); +test('restore: a known cid is uploaded again only when the server lost its copy', async () => { + let serverHas = true; + const app3 = new Hono(); + const p2p = registerP2pRoutes(app3, { cfg, p2pDb, fileForCid: async () => null, sha256Range: async () => '', log: quiet }); + const got = []; + registerIntakeRoutes(app3, { + cfg, p2pDb, gate: p2p.gate, requireDevice: p2p.requireDevice, validateMedia, admitFile, log: quiet, + serverHasCid: async () => serverHas, + adopt: async (videoId, path, info) => { got.push(info.sha256); unlinkSync(path); return { adopted: true }; }, // a real adopt moves the file + }); + const h = { 'X-Device': dev.deviceId + '.' + dev.secret, 'Content-Type': 'application/json' }; + const open = async () => (await app3.request('/api/p2p/intake', { method: 'POST', headers: h, body: JSON.stringify({ videoId: 'upAAAAAAAA1', cid: goodCid, size: good.length, restore: true }) })).json(); + expect(await open()).toEqual({ ok: true, known: true }); // server still has it + serverHas = false; + const t = await open(); + expect(t.ticket).toMatch(/^[0-9a-f]{32}$/); + const r = await (await app3.request(t.url, { method: 'PUT', headers: { 'X-Device': h['X-Device'] }, body: good })).json(); + expect(r).toEqual({ ok: true, cid: goodCid, adopted: true }); + expect(got).toEqual([goodCid]); +}); + test('claimed cid mismatch, truncated media, and non-media are rejected and deleted', async () => { const cases = [ [{ cid: 'f'.repeat(64), bytes: good }, 400, /hash mismatch/],