diff --git a/phone-quickdrop/package.json b/phone-quickdrop/package.json index 2e379e4..bd1dadb 100644 --- a/phone-quickdrop/package.json +++ b/phone-quickdrop/package.json @@ -7,7 +7,8 @@ "type": "commonjs", "scripts": { "start": "node server.js", - "smoke": "node scripts/smoke.js" + "smoke": "node scripts/smoke.js", + "smoke:reconnect": "node scripts/smoke-reconnect.js" }, "engines": { "node": ">=18" diff --git a/phone-quickdrop/public/desktop.html b/phone-quickdrop/public/desktop.html index 09d8c8b..fc7b1b9 100644 --- a/phone-quickdrop/public/desktop.html +++ b/phone-quickdrop/public/desktop.html @@ -76,10 +76,7 @@ .qr-wrap svg { width: 100% !important; height: 100% !important; shape-rendering: crispEdges; - color: #000; } - .qr-wrap svg rect[fill="#ffffff"], - .qr-wrap svg path[fill="#ffffff"] { fill: transparent; } .url-row { display: flex; align-items: center; gap: 8px; diff --git a/phone-quickdrop/public/phone.html b/phone-quickdrop/public/phone.html index 85264c0..88e30f9 100644 --- a/phone-quickdrop/public/phone.html +++ b/phone-quickdrop/public/phone.html @@ -276,6 +276,8 @@

Couldn't start camera

let recorder = null; let recording = false; let sessionId = null; + let sessionMime = null; + let sessionExt = null; let bytesSent = 0; let chunkCount = 0; let startedAt = 0; @@ -284,6 +286,9 @@

Couldn't start camera

let micOn = true; let wakeLock = null; let pendingChunks = []; // queued while WS not ready (rare) + // Outstanding ev.data.arrayBuffer() promises. Must drain before we send + // "end", otherwise the final chunk races the end-marker and gets dropped. + const pendingSends = new Set(); // ---------- mime negotiation ---------- function pickMimeType() { @@ -331,7 +336,21 @@

Couldn't start camera

ws.addEventListener("open", () => { wsReady = true; setConn("connected", "ready"); - // flush anything queued (very rare path) + // If we died mid-recording, ask the server to resume this sessionId + // before we flush queued chunks. WebSocket preserves order within a + // single connection, so chunks sent next will land after the resume. + if (recording && sessionId) { + try { + ws.send(JSON.stringify({ + type: "start", + sessionId, + mime: sessionMime || "application/octet-stream", + ext: sessionExt || "bin", + resume: true, + ts: startedAt, + })); + } catch { /* will close and reconnect */ } + } for (const buf of pendingChunks) ws.send(buf); pendingChunks = []; }); @@ -461,28 +480,39 @@

Couldn't start camera

sessionId = (crypto.randomUUID && crypto.randomUUID()) || ("s" + Math.random().toString(36).slice(2) + Date.now().toString(36)); + sessionMime = recorder.mimeType || mime || "application/octet-stream"; + sessionExt = ext; bytesSent = 0; chunkCount = 0; startedAt = Date.now(); + pendingSends.clear(); // Tell server a recording is starting wsSend(JSON.stringify({ type: "start", sessionId, - mime: recorder.mimeType || mime || "application/octet-stream", - ext, + mime: sessionMime, + ext: sessionExt, ts: startedAt, })); - recorder.ondataavailable = async (ev) => { + recorder.ondataavailable = (ev) => { if (!ev.data || ev.data.size === 0) return; chunkCount += 1; bytesSent += ev.data.size; - const buf = await ev.data.arrayBuffer(); - wsSend(buf); - updateSizePill(); + // Schedule the arrayBuffer conversion and remember the promise so + // onstop can drain it before sending "end". + const p = ev.data.arrayBuffer() + .then((buf) => { wsSend(buf); updateSizePill(); }) + .catch(() => { /* ignore; recorder error handler reports */ }); + pendingSends.add(p); + p.finally(() => pendingSends.delete(p)); }; - recorder.onstop = () => { + recorder.onstop = async () => { + // Drain any in-flight chunk conversions so the last chunk beats "end". + if (pendingSends.size) { + try { await Promise.allSettled([...pendingSends]); } catch {} + } wsSend(JSON.stringify({ type: "end", sessionId, @@ -490,6 +520,8 @@

Couldn't start camera

chunks: chunkCount, })); sessionId = null; + sessionMime = null; + sessionExt = null; recorder = null; }; recorder.onerror = (ev) => { diff --git a/phone-quickdrop/scripts/smoke-reconnect.js b/phone-quickdrop/scripts/smoke-reconnect.js new file mode 100644 index 0000000..7cf7ffe --- /dev/null +++ b/phone-quickdrop/scripts/smoke-reconnect.js @@ -0,0 +1,199 @@ +#!/usr/bin/env node +/** + * Reconnect smoke test. + * + * Simulates a phone whose WebSocket dies in the middle of a recording: + * 1. Open WS, send "start" with sessionId S + * 2. Send N/2 chunks + * 3. Abruptly close the socket (not a graceful "end") + * 4. Open a new WS, send "start" with same sessionId and resume:true + * 5. Expect ack { type:"started", resumed:true, bytes: setTimeout(r, ms)); } + +function connect() { + return new Promise((resolve, reject) => { + const ws = new WebSocket(URL, { rejectUnauthorized: false }); + ws.once("open", () => resolve(ws)); + ws.once("error", reject); + }); +} + +function collectMessages(ws) { + const messages = []; + ws.on("message", (data, isBinary) => { + if (isBinary) return; + try { messages.push(JSON.parse(data.toString())); } catch {} + }); + return messages; +} + +async function waitFor(messages, type, timeoutMs = 2000) { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const m = messages.find((x) => x.type === type); + if (m) return m; + await sleep(10); + } + return null; +} + +async function main() { + console.log(`smoke-reconnect: connecting to ${URL}`); + + // Pre-generate all chunks so we know exactly what bytes should land on disk. + const chunks = []; + for (let i = 0; i < CHUNK_COUNT; i++) { + chunks.push(crypto.randomBytes(CHUNK_BYTES)); + } + const expectedBuffer = Buffer.concat(chunks); + + const sessionId = "reconnect-" + crypto.randomBytes(3).toString("hex"); + const label = "reconnect-test"; + + // ---- phase 1: open, start, send half, kill ---- + let ws = await connect(); + let msgs = collectMessages(ws); + + ws.send(JSON.stringify({ + type: "start", + sessionId, + mime: "video/mp4", + ext: "mp4", + label, + ts: Date.now(), + })); + + const started1 = await waitFor(msgs, "started", 2000); + if (!started1 || started1.resumed) { + console.error(`smoke-reconnect: FAIL — expected fresh 'started' ack, got ${JSON.stringify(started1)}`); + process.exit(2); + } + const fileName = started1.fileName; + console.log(`smoke-reconnect: started ${sessionId} → ${fileName}`); + + for (let i = 0; i < CHUNK_COUNT / 2; i++) { + ws.send(chunks[i]); + await sleep(15); + } + + // Give the server a beat to flush the binary frames to disk before we cut + // the socket — otherwise we're testing TCP close semantics, not resume. + await sleep(80); + + // Abrupt close — terminate, not close() — so server sees an unclean drop. + ws.terminate(); + console.log(`smoke-reconnect: killed WS after ${CHUNK_COUNT / 2} chunks`); + + // Wait briefly for server to register the close and orphan the session. + await sleep(200); + + // ---- phase 2: reconnect with resume ---- + ws = await connect(); + msgs = collectMessages(ws); + + ws.send(JSON.stringify({ + type: "start", + sessionId, + mime: "video/mp4", + ext: "mp4", + resume: true, + ts: Date.now(), + })); + + const started2 = await waitFor(msgs, "started", 2000); + if (!started2) { + console.error("smoke-reconnect: FAIL — no 'started' ack after resume"); + process.exit(3); + } + if (!started2.resumed) { + console.error(`smoke-reconnect: FAIL — expected resumed:true, got ${JSON.stringify(started2)}`); + process.exit(4); + } + const halfBytes = (CHUNK_COUNT / 2) * CHUNK_BYTES; + if (typeof started2.bytes !== "number" || started2.bytes < 1 || started2.bytes > halfBytes) { + console.error( + `smoke-reconnect: FAIL — resume bytes unreasonable: got ${started2.bytes}, expected 1..${halfBytes}` + ); + process.exit(5); + } + console.log( + `smoke-reconnect: resumed ${sessionId} (server already has ${started2.bytes} bytes)` + ); + + // ---- phase 3: send the rest ---- + for (let i = CHUNK_COUNT / 2; i < CHUNK_COUNT; i++) { + ws.send(chunks[i]); + await sleep(15); + } + ws.send(JSON.stringify({ + type: "end", + sessionId, + totalBytes: TOTAL, + })); + + const saved = await waitFor(msgs, "saved", 3000); + ws.close(); + if (!saved) { + console.error("smoke-reconnect: FAIL — no 'saved' ack"); + process.exit(6); + } + + // ---- phase 4: verify on-disk bytes ---- + const filePath = path.join(OUTPUT_DIR, fileName); + if (!fs.existsSync(filePath)) { + console.error(`smoke-reconnect: FAIL — file missing: ${filePath}`); + process.exit(7); + } + const onDisk = fs.readFileSync(filePath); + if (onDisk.length !== TOTAL) { + console.error( + `smoke-reconnect: FAIL — size mismatch: disk=${onDisk.length} expected=${TOTAL}` + ); + process.exit(8); + } + if (!onDisk.equals(expectedBuffer)) { + // Find first divergence for debug + let i = 0; + while (i < onDisk.length && onDisk[i] === expectedBuffer[i]) i++; + console.error( + `smoke-reconnect: FAIL — content mismatch (first diff at byte ${i})` + ); + process.exit(9); + } + + try { fs.unlinkSync(filePath); } catch {} + + console.log( + `smoke-reconnect: OK — ${TOTAL.toLocaleString()} bytes landed correctly across a mid-session disconnect.` + ); +} + +main().catch((err) => { + console.error("smoke-reconnect: error", err); + process.exit(1); +}); diff --git a/phone-quickdrop/server.js b/phone-quickdrop/server.js index 0473827..35d4b6f 100644 --- a/phone-quickdrop/server.js +++ b/phone-quickdrop/server.js @@ -43,6 +43,8 @@ const OUTPUT_DIR = process.env.QUICKDROP_OUT || path.join(os.homedir(), "Desktop", "PhoneCaptures"); const RECENT_LIMIT = 50; +const HEARTBEAT_MS = 20_000; +const ORPHAN_RESUME_MS = 20_000; fs.mkdirSync(OUTPUT_DIR, { recursive: true }); fs.mkdirSync(CERT_DIR, { recursive: true }); @@ -148,6 +150,10 @@ function safeLabel(input) { const observers = new Set(); const phones = new Set(); const recentRecordings = []; // newest first +// Sessions whose phone WS dropped mid-recording; a reconnect with the same +// sessionId + resume:true re-attaches and keeps writing to the same file. +// Key: sessionId → { session, timer } +const orphanedSessions = new Map(); function broadcastToObservers(obj) { const payload = JSON.stringify(obj); @@ -243,12 +249,13 @@ async function start() { const phoneUrl = `https://${phoneHost}:${PORT}/phone`; const dashUrl = `https://localhost:${PORT}/`; - // Pre-render QR as inline SVG. + // Pre-render QR as inline SVG — standard black-on-white so any phone + // camera can read it against the white card background. const qrSvg = await QRCode.toString(phoneUrl, { type: "svg", errorCorrectionLevel: "M", margin: 1, - color: { dark: "#ffffff", light: "#00000000" }, + color: { dark: "#000000", light: "#ffffff" }, }); // Read dashboard template once at boot, substitute. @@ -359,6 +366,38 @@ async function start() { pushPhoneStatus(); console.log(`[${peer}] phone connected`); } + + // Resume path: phone reconnected mid-recording. Re-attach to the + // orphaned session so chunks keep landing in the same file. + if (msg.resume && msg.sessionId) { + const orphan = orphanedSessions.get(msg.sessionId); + if (orphan) { + clearTimeout(orphan.timer); + orphanedSessions.delete(msg.sessionId); + if (session && !session.closed && session !== orphan.session) { + await session.finish("abort"); + } + session = orphan.session; + session.peerLabel = peer; + console.log( + `[${peer}] resume ${session.sessionId} → ${session.fileName} ` + + `(have ${session.bytes} bytes so far)` + ); + ack({ + type: "started", + resumed: true, + sessionId: session.sessionId, + fileName: session.fileName, + bytes: session.bytes, + }); + return; + } + // Orphan expired or never existed — fall through to fresh start. + console.log( + `[${peer}] resume requested but session ${msg.sessionId} unknown; starting fresh` + ); + } + if (session && !session.closed) await session.finish("abort"); session = new RecordingSession({ sessionId: msg.sessionId || crypto.randomUUID(), @@ -377,6 +416,13 @@ async function start() { if (msg.type === "end") { if (!session) { ack({ type: "error", reason: "end_without_start" }); return; } const expected = msg.totalBytes; + // If this session was previously orphaned and is now finishing, make + // sure we aren't leaving a dangling timer. + const orphanEntry = orphanedSessions.get(session.sessionId); + if (orphanEntry) { + clearTimeout(orphanEntry.timer); + orphanedSessions.delete(session.sessionId); + } await session.finish("ok"); ack({ type: "saved", @@ -390,7 +436,15 @@ async function start() { return; } if (msg.type === "abort") { - if (session) { await session.finish("abort"); session = null; } + if (session) { + const orphanEntry = orphanedSessions.get(session.sessionId); + if (orphanEntry) { + clearTimeout(orphanEntry.timer); + orphanedSessions.delete(session.sessionId); + } + await session.finish("abort"); + session = null; + } ack({ type: "aborted" }); return; } @@ -401,7 +455,20 @@ async function start() { ws.on("close", async () => { if (session && !session.closed) { - await session.finish("partial"); + // Keep the file open for a window so a reconnect can resume. If the + // phone doesn't come back, finalize as partial and emit to observers. + const sid = session.sessionId; + const s = session; + const timer = setTimeout(async () => { + if (orphanedSessions.get(sid)?.session === s) { + orphanedSessions.delete(sid); + if (!s.closed) await s.finish("partial"); + } + }, ORPHAN_RESUME_MS); + orphanedSessions.set(sid, { session: s, timer }); + console.log( + `[${peer}] session ${sid} orphaned (resumable for ${ORPHAN_RESUME_MS}ms, ${s.bytes} bytes)` + ); session = null; } if (role === "phone") { @@ -417,9 +484,27 @@ async function start() { console.error(`[${peer}] ws error:`, err.message); }); + // Heartbeat: ping every HEARTBEAT_MS; if a ping is unanswered by the next + // cycle, the socket is terminated and normal close handling kicks in + // (which orphans any active session so reconnect can resume). + ws.isAlive = true; + ws.on("pong", () => { ws.isAlive = true; }); + ack({ type: "hello", outputDir: OUTPUT_DIR, phoneUrl }); }); + const heartbeatTimer = setInterval(() => { + for (const client of wss.clients) { + if (client.isAlive === false) { + try { client.terminate(); } catch {} + continue; + } + client.isAlive = false; + try { client.ping(); } catch {} + } + }, HEARTBEAT_MS); + wss.on("close", () => clearInterval(heartbeatTimer)); + server.on("error", (err) => { if (err.code === "EADDRINUSE") { console.error(`\nPort ${PORT} is already in use. Set PORT= and try again.\n`);