From f5b59cc1d4036ce0650d0a87bafdb7fc52d2a62d Mon Sep 17 00:00:00 2001 From: "prompt.ac/@jeffrey" Date: Fri, 9 Oct 2026 21:05:08 -0700 Subject: [PATCH] Whistlegraph turns, slice 1: the queue and the submit route system/backend/whistlegraph-turns.mjs is the Mongo queue from TURNS.md: enqueue (one row per requestID, three waiting per owner), claim (oldest first, one running turn per thread, a lease that went quiet for 90 s is taken over first, checkpoint carried across), heartbeat, complete and fail (lease holder only), read, listOpen, recent, depth. publicTurn never exposes the owner's subject. POST/GET /api/whistlegraph-turn submits for the signed-in owner against a thread they own, refusing a request that builds on a version the thread has moved past. Workers never use the route. Tested against a fake collection. --- system/backend/whistlegraph-turns.mjs | 116 ++++++++++++++++++ system/backend/whistlegraph-turns.test.mjs | 86 +++++++++++++ .../netlify/functions/whistlegraph-turn.mjs | 42 +++++++ 3 files changed, 244 insertions(+) create mode 100644 system/backend/whistlegraph-turns.mjs create mode 100644 system/backend/whistlegraph-turns.test.mjs create mode 100644 system/netlify/functions/whistlegraph-turn.mjs diff --git a/system/backend/whistlegraph-turns.mjs b/system/backend/whistlegraph-turns.mjs new file mode 100644 index 0000000000..1362b79458 --- /dev/null +++ b/system/backend/whistlegraph-turns.mjs @@ -0,0 +1,116 @@ +// The queue of Whistlegraph turns that run off the phone (apple/whistlegraph/TURNS.md). +// One row per submitted request. Workers claim with an atomic update and hold a +// lease by heartbeat; a lease that goes quiet is claimable again. Rows are the +// durable record of what was asked, what happened, and what it cost. +import {randomUUID} from 'node:crypto'; +import {validCode} from './whistlegraph.mjs'; + +export const LEASE_MS = 90_000; // A worker that misses three 30 s heartbeats has lost the job. +export const MAX_QUEUED_PER_OWNER = 3; // Backpressure: a person waits on their own queue, never on everyone's. +export const MAX_TEXT = 1200; // The phone's 96-grapheme line, or the full mixed-input prompt. +const STATUSES = ['queued','running','done','failed']; +const isHex = value => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value); + +export function validateTurnRequest(value) { + if (!value || typeof value !== 'object') throw Error('Send a turn'); + const request = {}; + if (!validCode(value.code)) throw Error('Specify the thread code'); + request.code = value.code; + if (typeof value.text !== 'string' || value.text.length > MAX_TEXT) throw Error(`Text must be a string of at most ${MAX_TEXT} characters`); + request.text = value.text; + if (value.displayText !== undefined) { + if (typeof value.displayText !== 'string' || value.displayText.length > 200) throw Error('displayText must be short'); + request.displayText = value.displayText; + } + if (value.drawing !== undefined && value.drawing !== null) { + const d = value.drawing; + if (d?.schema !== 'whistlegraph-drawing/v1' || !Array.isArray(d.strokes) || !d.strokes.length || d.strokes.length > 32 || JSON.stringify(d).length > 200_000) throw Error('Invalid drawing'); + request.drawing = d; + } + if (!request.text.trim() && !request.drawing) throw Error('Type a few words or add chalk'); + if (!Number.isSafeInteger(value.baseVersion) || value.baseVersion < 0) throw Error('baseVersion must be the version the request builds on'); + request.baseVersion = value.baseVersion; + if (!isHex(value.baseHash)) throw Error('baseHash must be the sha256 of that version\'s source'); + request.baseHash = value.baseHash; + if (value.model !== undefined) { + if (typeof value.model !== 'string' || !/^[a-z0-9./-]{3,80}$/i.test(value.model)) throw Error('Invalid model'); + request.model = value.model; + } + if (value.requestID !== undefined) { + if (typeof value.requestID !== 'string' || !/^[a-f0-9-]{36}$/i.test(value.requestID)) throw Error('requestID must be a UUID'); + request.requestID = value.requestID; + } + return request; +} + +// The row as the phone and the admin see it: never the owner's subject. +export function publicTurn(row) { + if (!row) return null; + const {owner, ...rest} = row; + return {...rest, id: row._id}; +} + +export function mongoTurnQueue(collection, {now = () => new Date()} = {}) { + let indexes; + const ready = () => indexes ??= Promise.all([ + collection.createIndex({status: 1, createdAt: 1}), + collection.createIndex({owner: 1, status: 1}), + collection.createIndex({threadID: 1, status: 1}), + collection.createIndex({requestID: 1}, {unique: true}), + ]); + return { + // One running turn per thread, at most MAX_QUEUED_PER_OWNER waiting per person, + // one row per requestID: a resubmit of the same request returns the same row. + async enqueue(owner, thread, request) { + await ready(); + const requestID = request.requestID || randomUUID(); + const existing = await collection.findOne({requestID}); + if (existing) { if (existing.owner !== owner) throw Object.assign(Error('Request unavailable'), {statusCode: 404}); return existing; } + const waiting = await collection.countDocuments({owner, status: 'queued'}); + if (waiting >= MAX_QUEUED_PER_OWNER) throw Object.assign(Error(`You already have ${waiting} turns waiting. Let one finish first.`), {statusCode: 429}); + const at = now().toISOString(); + const row = {_id: randomUUID(), owner, threadID: thread._id, code: thread.code, requestID, + request: {text: request.text, displayText: request.displayText ?? '', drawing: request.drawing ?? null}, + baseVersion: request.baseVersion, baseHash: request.baseHash, model: request.model ?? null, + status: 'queued', createdAt: at, claimedBy: null, claimedAt: null, heartbeatAt: null, finishedAt: null, + attempts: 0, checkpoint: null, result: null}; + await collection.insertOne(row); + return row; + }, + // Oldest queued job whose thread has nothing running; or a running job whose lease lapsed. + async claim(worker) { + await ready(); + const at = now(); + const stale = new Date(+at - LEASE_MS).toISOString(); + const running = await collection.distinct('threadID', {status: 'running', heartbeatAt: {$gte: stale}}); + const set = {status: 'running', claimedBy: worker, claimedAt: at.toISOString(), heartbeatAt: at.toISOString()}; + const lapsed = await collection.findOneAndUpdate({status: 'running', heartbeatAt: {$lt: stale}}, {$set: set, $inc: {attempts: 1}}, {sort: {heartbeatAt: 1}, returnDocument: 'after'}); + if (lapsed) return lapsed; + return collection.findOneAndUpdate({status: 'queued', threadID: {$nin: running}}, {$set: set, $inc: {attempts: 1}}, {sort: {createdAt: 1}, returnDocument: 'after'}); + }, + async heartbeat(id, worker, checkpoint) { + const update = {$set: {heartbeatAt: now().toISOString()}}; + if (typeof checkpoint === 'string') update.$set.checkpoint = checkpoint; + const result = await collection.updateOne({_id: id, status: 'running', claimedBy: worker}, update); + return result.modifiedCount === 1; + }, + // Only the worker holding the lease can finish a job; a late worker's result is dropped. + async complete(id, worker, result) { + const r = await collection.updateOne({_id: id, status: 'running', claimedBy: worker}, {$set: {status: 'done', finishedAt: now().toISOString(), result}}); + return r.modifiedCount === 1; + }, + async fail(id, worker, error, {draft = null} = {}) { + const r = await collection.updateOne({_id: id, status: 'running', claimedBy: worker}, {$set: {status: 'failed', finishedAt: now().toISOString(), result: {error: String(error).slice(0, 2000), draft}}}); + return r.modifiedCount === 1; + }, + async read(owner, id) { return collection.findOne({_id: id, owner}); }, + async listOpen(owner, threadID) { return collection.find({owner, threadID, status: {$in: ['queued','running']}}).sort({createdAt: 1}).toArray(); }, + async recent(owner, limit = 20) { return collection.find({owner}).sort({createdAt: -1}).limit(Math.min(100, Math.max(1, limit))).toArray(); }, + // Queue depth is the autoscale signal. + async depth() { + const [queued, running] = await Promise.all([collection.countDocuments({status: 'queued'}), collection.countDocuments({status: 'running'})]); + return {queued, running}; + }, + }; +} +export {STATUSES as TURN_STATUSES}; diff --git a/system/backend/whistlegraph-turns.test.mjs b/system/backend/whistlegraph-turns.test.mjs new file mode 100644 index 0000000000..67f95cfa43 --- /dev/null +++ b/system/backend/whistlegraph-turns.test.mjs @@ -0,0 +1,86 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import {mongoTurnQueue, validateTurnRequest, publicTurn, LEASE_MS, MAX_QUEUED_PER_OWNER} from './whistlegraph-turns.mjs'; + +// The slice of Mongo a queue needs, including findOneAndUpdate with sort, $nin, $lt/$gte and distinct. +function collection() { + const rows = []; + const get = (row, key) => key.split('.').reduce((o, k) => o?.[k], row); + const matches = (row, query) => Object.entries(query).every(([key, value]) => { + const actual = get(row, key); + if (value && typeof value === 'object' && !Array.isArray(value)) { + if ('$in' in value && !value.$in.includes(actual)) return false; + if ('$nin' in value && value.$nin.includes(actual)) return false; + if ('$lt' in value && !(actual != null && actual < value.$lt)) return false; + if ('$gte' in value && !(actual != null && actual >= value.$gte)) return false; + return true; + } + return actual === value; + }); + const apply = (row, update) => { Object.assign(row, update.$set || {}); for (const [k, n] of Object.entries(update.$inc || {})) row[k] = (row[k] || 0) + n; }; + const sorted = (list, sort) => { if (!sort) return list; const [[key, dir]] = Object.entries(sort); return list.slice().sort((a, b) => (get(a, key) > get(b, key) ? 1 : get(a, key) < get(b, key) ? -1 : 0) * dir); }; + return { + rows, + createIndex: async () => {}, + insertOne: async row => { if (rows.some(r => r.requestID === row.requestID)) throw Object.assign(Error('dup'), {code: 11000}); rows.push({...row}); }, + findOne: async q => rows.find(r => matches(r, q)) || null, + countDocuments: async q => rows.filter(r => matches(r, q)).length, + distinct: async (key, q) => [...new Set(rows.filter(r => matches(r, q)).map(r => get(r, key)))], + updateOne: async (q, update) => { const row = rows.find(r => matches(r, q)); if (!row) return {modifiedCount: 0}; apply(row, update); return {modifiedCount: 1}; }, + findOneAndUpdate: async (q, update, {sort} = {}) => { const row = sorted(rows.filter(r => matches(r, q)), sort)[0]; if (!row) return null; apply(row, update); return {...row}; }, + find: (q) => { let out = rows.filter(r => matches(r, q)); const chain = {sort(s) { out = sorted(out, s); return chain; }, limit(n) { out = out.slice(0, n); return chain; }, toArray: async () => out.map(r => ({...r}))}; return chain; }, + }; +} +const hash = 'a'.repeat(64); +const thread = (id, code) => ({_id: id, code}); +const good = {code: 'wgFeeda', text: 'make it spin', baseVersion: 2, baseHash: hash}; + +test('a turn request is checked before it is queued', () => { + assert.deepEqual(validateTurnRequest(good), {code: 'wgFeeda', text: 'make it spin', baseVersion: 2, baseHash: hash}); + assert.throws(() => validateTurnRequest({...good, code: 'nope'}), /thread code/); + assert.throws(() => validateTurnRequest({...good, text: ''}), /words or add chalk/); + assert.throws(() => validateTurnRequest({...good, text: 'x'.repeat(1201)}), /at most/); + assert.throws(() => validateTurnRequest({...good, baseHash: 'short'}), /sha256/); + assert.throws(() => validateTurnRequest({...good, baseVersion: -1}), /baseVersion/); + assert.throws(() => validateTurnRequest({...good, model: 'bad model!'}), /model/); + assert.throws(() => validateTurnRequest({...good, drawing: {schema: 'nope'}}), /drawing/); + const chalk = validateTurnRequest({...good, text: '', drawing: {schema: 'whistlegraph-drawing/v1', id: '0'.repeat(8) + '-0000-4000-8000-' + '0'.repeat(12), revision: 1, aspect: 1.33, strokes: [[[1, 2, 0]]]}}); + assert.equal(chalk.text, ''); assert.ok(chalk.drawing); +}); + +test('claims go oldest first, one running turn per thread, and a quiet lease is reclaimable', async () => { + let t = 0; const clock = () => new Date(1_700_000_000_000 + t); + const c = collection(), q = mongoTurnQueue(c, {now: clock}); + const a = await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest(good)); t += 1000; + const b = await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'then bounce'})); t += 1000; + const other = await q.enqueue('bob', thread('t2', 'wgBobbo'), validateTurnRequest({...good, code: 'wgBobbo', text: 'a moon'})); t += 1000; + assert.equal(publicTurn(a).owner, undefined, 'the subject never leaves the server'); + assert.equal((await q.claim('w1'))._id, a._id, 'oldest first'); + assert.equal((await q.claim('w2'))._id, other._id, 'the same thread is skipped while a turn runs on it, so bob goes next'); + assert.equal(await q.claim('w3'), null, 'alice\'s second turn waits for her first'); + assert.ok(await q.heartbeat(a._id, 'w1', 'export function paint(){}')); + assert.equal(!(await q.heartbeat(a._id, 'w9')), true, 'only the lease holder heartbeats'); + t += LEASE_MS + 1; + assert.equal((await q.claim('w3'))._id, a._id, 'w1 went quiet; w3 takes the job over'); + assert.equal(c.rows.find(r => r._id === a._id).attempts, 2); + assert.equal(c.rows.find(r => r._id === a._id).checkpoint, 'export function paint(){}', 'the checkpoint survives the handover'); + assert.equal(await q.complete(a._id, 'w1', {versionID: 3}), false, 'the late worker cannot finish it'); + assert.ok(await q.complete(a._id, 'w3', {versionID: 3})); + assert.equal((await q.claim('w1'))._id, other._id, 'bob\'s lease also lapsed; a lapsed job is taken over before new work'); + assert.equal((await q.claim('w1'))._id, b._id, 'with alice\'s first done, her second runs'); + assert.ok(await q.fail(b._id, 'w1', 'Nothing painted', {draft: 'x'})); + const depth = await q.depth(); assert.deepEqual(depth, {queued: 0, running: 1}, 'bob\'s reclaimed turn is still running'); + assert.equal((await q.listOpen('alice', 't1')).length, 0); + assert.equal((await q.recent('alice')).length, 2); +}); + +test('the same request is one row, and a person cannot stack more than the cap', async () => { + const c = collection(), q = mongoTurnQueue(c); + const id = '12345678-1234-4123-8123-123456789abc'; + const first = await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, requestID: id})); + const again = await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, requestID: id})); + assert.equal(again._id, first._id); + await assert.rejects(q.enqueue('bob', thread('t2', 'wgBobbo'), validateTurnRequest({...good, code: 'wgBobbo', requestID: id})), {statusCode: 404}); + for (let i = 1; i < MAX_QUEUED_PER_OWNER; i++) await q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'more ' + i})); + await assert.rejects(q.enqueue('alice', thread('t1', 'wgFeeda'), validateTurnRequest({...good, text: 'one too many'})), {statusCode: 429}); +}); diff --git a/system/netlify/functions/whistlegraph-turn.mjs b/system/netlify/functions/whistlegraph-turn.mjs new file mode 100644 index 0000000000..8aa8fce70c --- /dev/null +++ b/system/netlify/functions/whistlegraph-turn.mjs @@ -0,0 +1,42 @@ +// Submit a Whistlegraph turn to run off the phone, or read how it is going. +// POST {code, text, displayText?, drawing?, baseVersion, baseHash, model?, requestID?} → the queued turn +// GET ?id= → one turn GET ?code= → that thread's open turns +// GET ?recent=1 → the owner's last turns +// Workers claim and finish turns through the queue module directly, never this route. +import {connect} from '../../backend/database.mjs'; +import {authenticateMusical} from './easel-musical-jev.mjs'; +import {whistlegraphStore} from './whistlegraph.mjs'; +import {mongoTurnQueue, validateTurnRequest, publicTurn} from '../../backend/whistlegraph-turns.mjs'; +let pending; +export const turnQueue = () => pending ??= (async () => { const {db} = await connect(); return mongoTurnQueue(db.collection('walkieware-turns')); })().catch(error => { pending = null; throw error; }); +export async function handler(event) { + const headers = {'Content-Type': 'application/json', 'Cache-Control': 'no-store', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Headers': 'Authorization, Content-Type', 'Access-Control-Allow-Methods': 'GET, POST, OPTIONS'}; + const reply = (statusCode, value) => ({statusCode, headers, body: JSON.stringify(value)}); + if (event.httpMethod === 'OPTIONS') return reply(204, {}); + if (!['GET', 'POST'].includes(event.httpMethod)) return reply(405, {error: 'Use GET or POST'}); + let owner; + try { owner = await authenticateMusical(event.headers || {}); } catch { return reply(503, {error: 'Authentication unavailable'}); } + if (!owner) return reply(401, {error: 'Sign in to your AC account'}); + let queue, store; + try { [queue, store] = await Promise.all([turnQueue(), whistlegraphStore()]); } catch { return reply(503, {error: 'Turn queue unavailable'}); } + try { + if (event.httpMethod === 'GET') { + const q = event.queryStringParameters || {}; + if (q.id) { const row = await queue.read(owner, q.id); return row ? reply(200, publicTurn(row)) : reply(404, {error: 'No such turn'}); } + if (q.code) { + const thread = await store.read(owner, q.code); + if (!thread) return reply(404, {error: 'Thread unavailable'}); + return reply(200, {turns: (await queue.listOpen(owner, thread._id)).map(publicTurn)}); + } + return reply(200, {turns: (await queue.recent(owner, Number(q.limit) || 20)).map(publicTurn)}); + } + let body; try { body = JSON.parse(event.body || '{}'); } catch { return reply(400, {error: 'Send JSON'}); } + let request; try { request = validateTurnRequest(body); } catch (error) { return reply(400, {error: error.message}); } + const thread = await store.read(owner, request.code); + if (!thread) return reply(404, {error: 'Thread unavailable'}); + const head = thread.ledger?.versions.find(v => v.id === thread.ledger.head); + if (!head || head.id !== request.baseVersion) return reply(409, {error: `The piece is at v${head?.id ?? 0}; the request builds on v${request.baseVersion}. Reopen it and try again.`}); + try { return reply(202, publicTurn(await queue.enqueue(owner, thread, request))); } + catch (error) { return reply(error.statusCode || 500, {error: error.message}); } + } catch { return reply(503, {error: 'Turn queue unavailable'}); } +} -- 2.54.0