From 0e9f04e139a7cf039a99a8688f82be43563554b4 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 15 Jul 2026 08:23:30 +0300 Subject: [PATCH 1/2] feat(relay): monotonic seq cursor + timing-safe token compare MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses codex's post-merge review of #37: - Every message now carries a monotonic `seq`. GET accepts `after=` for an incremental cursor feed (seq>after, oldest-first, capped) so same-millisecond messages are never skipped — `since` (created_at) alone couldn't cursor exactly. Prerequisite for the client-failover and sync-on-reconnect PRs. - Token comparison is now constant-time (crypto.timingSafeEqual over sha256). +2 tests (no-skip cursor incl. same-ms, forward-feed paging). Full suite 152 green. Co-Authored-By: Claude Opus 4.8 --- docs/groupmind-local.md | 8 +++++++- scripts/local-relay.mjs | 38 ++++++++++++++++++++++++++++---------- test/local-relay.test.mjs | 38 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 73 insertions(+), 11 deletions(-) diff --git a/docs/groupmind-local.md b/docs/groupmind-local.md index 5617a59..0508411 100644 --- a/docs/groupmind-local.md +++ b/docs/groupmind-local.md @@ -31,7 +31,13 @@ If `IAK_RELAY_TOKEN` is set, every request must present it via `X-API-Key` or | `GET` | `/health` | — | always open; reports message count | | `POST` | `/api/v1/messages` | `{ room, body, from?, metadata? }` | create a message | | `POST` | `/api/v1/rooms/:room/messages` | `{ body, from?, metadata? }` | create (room in path) | -| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=` | newest-first, GroupMind shape | +| `GET` | `/api/v1/rooms/:room/messages` | `?limit=&since=&after=` | see below | + +`GET` returns the recent view (**newest-first**) by default. Pass `after=` +for an incremental **cursor feed**: messages with `seq > after`, **oldest-first**, +capped by `limit` — the client advances its cursor and re-fetches, so nothing is +skipped even when messages share a millisecond timestamp. Every message carries a +monotonic `seq` (the exact cursor) alongside its millisecond `created_at`. An agent points its room client at `http://:18790/api/v1` when cloud GroupMind is unreachable — same paths, just a different base URL. diff --git a/scripts/local-relay.mjs b/scripts/local-relay.mjs index ef73265..85c58e8 100644 --- a/scripts/local-relay.mjs +++ b/scripts/local-relay.mjs @@ -35,7 +35,7 @@ import http from 'node:http'; import { appendFileSync, mkdirSync, readFileSync, existsSync } from 'node:fs'; -import { randomUUID } from 'node:crypto'; +import { randomUUID, timingSafeEqual, createHash } from 'node:crypto'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; @@ -60,6 +60,14 @@ function appendMessage(storePath, msg) { appendFileSync(storePath, JSON.stringify(msg) + '\n'); } +// Constant-time token compare. Hash both sides to a fixed 32 bytes first so +// timingSafeEqual never sees a length mismatch (and length isn't leaked). +function safeTokenEqual(provided, expected) { + const a = createHash('sha256').update(String(provided ?? '')).digest(); + const b = createHash('sha256').update(String(expected ?? '')).digest(); + return timingSafeEqual(a, b); +} + function send(res, status, obj) { const payload = JSON.stringify(obj); res.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload) }); @@ -115,10 +123,17 @@ export function startRelay(opts = {}) { if (!token) log('WARNING: no IAK_RELAY_TOKEN set — relay is LAN-open (fine on a trusted home LAN only).'); + // Monotonic per-message sequence, resumed from the store. `created_at` is + // millisecond-resolution so it can't cursor exactly (a same-ms checkpoint + // would skip messages); `seq` is the precise cursor for incremental fetch, + // which the failover + sync layers rely on. + let nextSeq = readMessages(storePath).reduce((mx, m) => Math.max(mx, m.seq || 0), 0) + 1; + const commit = (msg) => { msg.seq = nextSeq++; appendMessage(storePath, msg); return msg; }; + function authorized(req) { if (!token) return true; const bearer = (req.headers['authorization'] || '').replace(/^Bearer\s+/i, ''); - return req.headers['x-api-key'] === token || bearer === token; + return safeTokenEqual(req.headers['x-api-key'], token) || safeTokenEqual(bearer, token); } const server = http.createServer(async (req, res) => { @@ -138,12 +153,17 @@ export function startRelay(opts = {}) { const room = decodeURIComponent(getMatch[1]); const limit = Math.min(Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT), MAX_LIMIT); const since = url.searchParams.get('since'); + const after = url.searchParams.get('after'); // exact seq cursor + // readMessages returns JSONL append order = chronological. let msgs = readMessages(storePath).filter((m) => m.room === room); if (since) msgs = msgs.filter((m) => m.created_at > since); - // JSONL append order IS chronological, so it orders even messages that - // share a millisecond timestamp. Take the last `limit` (newest) and - // reverse to newest-first, matching GroupMind's shape. - const messages = msgs.slice(-limit).reverse(); + const cursoring = after !== null && after !== ''; + if (cursoring) msgs = msgs.filter((m) => (m.seq ?? 0) > Number(after)); + // Cursor feed (`after`): chronological forward from the cursor, capped — + // the client advances its cursor and re-fetches, so nothing is skipped + // even for same-millisecond messages. Recent view (no cursor): + // newest-first, matching GroupMind's shape. + const messages = cursoring ? msgs.slice(0, limit) : msgs.slice(-limit).reverse(); return send(res, 200, { messages, count: messages.length }); } @@ -152,8 +172,7 @@ export function startRelay(opts = {}) { const b = await readJsonBody(req); const { msg, error } = buildMessage({ room: b.room, body: b.body, from: b.from, metadata: b.metadata }); if (error) return send(res, 400, { error }); - appendMessage(storePath, msg); - return send(res, 201, msg); + return send(res, 201, commit(msg)); } // POST /api/v1/rooms/:room/messages { body, ... } @@ -163,8 +182,7 @@ export function startRelay(opts = {}) { const b = await readJsonBody(req); const { msg, error } = buildMessage({ room, body: b.body, from: b.from, metadata: b.metadata }); if (error) return send(res, 400, { error }); - appendMessage(storePath, msg); - return send(res, 201, msg); + return send(res, 201, commit(msg)); } return send(res, 404, { error: 'not found' }); diff --git a/test/local-relay.test.mjs b/test/local-relay.test.mjs index 9dc29ea..df0c0c4 100644 --- a/test/local-relay.test.mjs +++ b/test/local-relay.test.mjs @@ -126,3 +126,41 @@ test('unknown route is 404', async () => { assert.equal(r.status, 404); }); }); + +test('messages get a monotonic seq; after-cursor is a no-skip chronological feed', async () => { + await withRelay({}, async ({ base, headers }) => { + const posted = []; + for (const n of [1, 2, 3, 4]) { + const r = await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + posted.push(await r.json()); + } + // seq strictly increasing even when created_at collides in the same ms + for (let i = 1; i < posted.length; i++) assert.ok(posted[i].seq > posted[i - 1].seq, 'seq increases'); + + // cursor from 0: all four, oldest-first + const all = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0`, { headers })).json(); + assert.deepEqual(all.messages.map((m) => m.body), ['m1', 'm2', 'm3', 'm4']); + + // cursor from m2's seq: exactly m3, m4 — the same-ms skip that `since` alone would cause + const after2 = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[1].seq}`, { headers })).json(); + assert.deepEqual(after2.messages.map((m) => m.body), ['m3', 'm4']); + + // cursor from the last seq: caught up, empty + const afterLast = await (await fetch(`${base}/api/v1/rooms/c/messages?after=${posted[3].seq}`, { headers })).json(); + assert.equal(afterLast.count, 0); + }); +}); + +test('after-cursor honors limit and stays chronological (forward feed pages)', async () => { + await withRelay({}, async ({ base, headers }) => { + for (const n of [1, 2, 3, 4, 5]) { + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + } + const page = await (await fetch(`${base}/api/v1/rooms/c/messages?after=0&limit=2`, { headers })).json(); + assert.deepEqual(page.messages.map((m) => m.body), ['m1', 'm2'], 'oldest-first page from the cursor'); + }); +}); From bc49bff07ea848d5e9608ff8e44b3ad8f98748d7 Mon Sep 17 00:00:00 2001 From: ThinkOff Date: Wed, 15 Jul 2026 08:31:42 +0300 Subject: [PATCH 2/2] fix(relay): make after cursor authoritative + seq for pre-seq rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses codex's two P2 findings on #38: - When both are present, 'after' (seq) is now authoritative and 'since' (millisecond created_at) is ignored, so a same-ms checkpoint can't re-drop the messages the cursor must return. - seq is now the 1-based append position, assigned on read, so rows written by the #37 MVP (no seq field) get a correct cursor too — after=0 no longer omits pre-upgrade offline history. +2 regression tests. Full suite 154 green. Co-Authored-By: Claude Opus 4.8 --- scripts/local-relay.mjs | 32 +++++++++++++++++++--------- test/local-relay.test.mjs | 45 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 66 insertions(+), 11 deletions(-) diff --git a/scripts/local-relay.mjs b/scripts/local-relay.mjs index 85c58e8..307338d 100644 --- a/scripts/local-relay.mjs +++ b/scripts/local-relay.mjs @@ -52,6 +52,10 @@ function readMessages(storePath) { if (!line.trim()) continue; try { out.push(JSON.parse(line)); } catch { /* skip a corrupt line, never crash a read */ } } + // `seq` is the 1-based append position, assigned HERE on read so it is present + // and correct even for rows written before seq existed (e.g. a store from the + // #37 MVP). This is the authoritative cursor value; any stored seq is ignored. + out.forEach((m, i) => { m.seq = i + 1; }); return out; } @@ -123,12 +127,13 @@ export function startRelay(opts = {}) { if (!token) log('WARNING: no IAK_RELAY_TOKEN set — relay is LAN-open (fine on a trusted home LAN only).'); - // Monotonic per-message sequence, resumed from the store. `created_at` is + // `seq` = 1-based append position (see readMessages). `created_at` is // millisecond-resolution so it can't cursor exactly (a same-ms checkpoint - // would skip messages); `seq` is the precise cursor for incremental fetch, - // which the failover + sync layers rely on. - let nextSeq = readMessages(storePath).reduce((mx, m) => Math.max(mx, m.seq || 0), 0) + 1; - const commit = (msg) => { msg.seq = nextSeq++; appendMessage(storePath, msg); return msg; }; + // would skip messages); `seq` is the precise cursor the failover + sync + // layers rely on. `count` is seeded from the (possibly pre-seq) store so old + // rows are counted, and it tracks the next position across appends in-process. + let count = readMessages(storePath).length; + const commit = (msg) => { msg.seq = ++count; appendMessage(storePath, msg); return msg; }; function authorized(req) { if (!token) return true; @@ -153,12 +158,19 @@ export function startRelay(opts = {}) { const room = decodeURIComponent(getMatch[1]); const limit = Math.min(Math.max(1, Number(url.searchParams.get('limit')) || DEFAULT_LIMIT), MAX_LIMIT); const since = url.searchParams.get('since'); - const after = url.searchParams.get('after'); // exact seq cursor - // readMessages returns JSONL append order = chronological. - let msgs = readMessages(storePath).filter((m) => m.room === room); - if (since) msgs = msgs.filter((m) => m.created_at > since); + const after = url.searchParams.get('after'); // exact seq cursor (authoritative) const cursoring = after !== null && after !== ''; - if (cursoring) msgs = msgs.filter((m) => (m.seq ?? 0) > Number(after)); + // readMessages returns JSONL append order = chronological, each with its + // positional seq. + let msgs = readMessages(storePath).filter((m) => m.room === room); + if (cursoring) { + // `after` (seq) is authoritative — never ALSO narrow by the + // millisecond `since`, which would re-drop the same-ms messages the + // cursor is meant to return (PR #38 review). + msgs = msgs.filter((m) => m.seq > Number(after)); + } else if (since) { + msgs = msgs.filter((m) => m.created_at > since); + } // Cursor feed (`after`): chronological forward from the cursor, capped — // the client advances its cursor and re-fetches, so nothing is skipped // even for same-millisecond messages. Recent view (no cursor): diff --git a/test/local-relay.test.mjs b/test/local-relay.test.mjs index df0c0c4..b8ec4da 100644 --- a/test/local-relay.test.mjs +++ b/test/local-relay.test.mjs @@ -3,7 +3,7 @@ import test from 'node:test'; import { once } from 'node:events'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { rmSync } from 'node:fs'; +import { rmSync, writeFileSync } from 'node:fs'; import { randomUUID } from 'node:crypto'; import { startRelay } from '../scripts/local-relay.mjs'; @@ -164,3 +164,46 @@ test('after-cursor honors limit and stays chronological (forward feed pages)', a assert.deepEqual(page.messages.map((m) => m.body), ['m1', 'm2'], 'oldest-first page from the cursor'); }); }); + +test('after is authoritative over since (since must not re-drop same-ms messages)', async () => { + await withRelay({}, async ({ base, headers }) => { + for (const n of [1, 2, 3]) { + await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: j(headers), body: JSON.stringify({ room: 'c', body: `m${n}` }), + }); + } + // A future `since` would exclude everything on its own; with `after=0` + // present it must be ignored, so all three still come back. + const future = new Date(Date.now() + 60_000).toISOString(); + const res = await (await fetch( + `${base}/api/v1/rooms/c/messages?after=0&since=${encodeURIComponent(future)}`, { headers })).json(); + assert.deepEqual(res.messages.map((m) => m.body), ['m1', 'm2', 'm3'], 'after wins; since ignored'); + }); +}); + +test('rows written before seq existed still get a positional seq (upgrade compat)', async () => { + const storePath = join(tmpdir(), `iak-relay-seed-${randomUUID()}.jsonl`); + // Simulate a #37-era store: rows with NO seq field. + const row = (id, body) => JSON.stringify({ + id, room: 'r', from: '@x', body, created_at: '2026-01-01T00:00:00.000Z', origin: 'local-relay', + }); + writeFileSync(storePath, `${row('a', 'old1')}\n${row('b', 'old2')}\n`); + const server = startRelay({ port: 0, storePath, logger: () => {} }); + await once(server, 'listening'); + const base = `http://127.0.0.1:${server.address().port}`; + try { + // after=0 must include BOTH pre-seq rows (not silently skip them) + const feed = await (await fetch(`${base}/api/v1/rooms/r/messages?after=0`)).json(); + assert.deepEqual(feed.messages.map((m) => m.body), ['old1', 'old2']); + assert.deepEqual(feed.messages.map((m) => m.seq), [1, 2], 'positional seq assigned to legacy rows'); + // a new POST continues the sequence past the seeded rows + const posted = await (await fetch(`${base}/api/v1/messages`, { + method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ room: 'r', body: 'new' }), + })).json(); + assert.equal(posted.seq, 3, 'new message continues after the seeded rows'); + } finally { + server.close(); + await once(server, 'close'); + rmSync(storePath, { force: true }); + } +});