From ba32d80bf010c3be327b6423c3763f06c5a9d24f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 29 Aug 2026 02:31:04 +0000 Subject: [PATCH 1/3] fix(changes): stop two different filters sharing one resume buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A resume buffer opens exactly one ScopedStack.subscribe(), carrying the filter of whichever connection minted it, so every connection that shares a buffer shares that filter. Two things could put connections that mean different things onto one buffer. resumeBufferKey() built its key with a positional JSON.stringify over an array, where an undefined element renders as null. That collapsed "no parentId filter" onto `parentId: null` — root records only, a filter the wire format defines. Whichever connected first then decided what the other received: an unfiltered connection silently never heard about child records, with no reset frame to announce the gap, or a roots-only connection received changes outside its filter. Both are contract violations, and the silent one is the failure wire-format.md § Change feed calls untrustworthy. Keying on an object fixes it directly: JSON.stringify drops an undefined property but not an undefined array element, so absent is now encoded as an absent key. acquire() also registered a buffer before awaiting its subscription, so a subscription that rejected left a buffer in the map that nothing was feeding and nothing would retry — every later connection on that key got ready followed by permanent silence. The buffer now lands in the map only once its subscription resolves, with in-flight attempts tracked separately so two connections racing on one key still open one subscription and share however it settles. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FH88yDv3Fk2RmLsU7DU4Rz --- src/lib/resumeBuffer.ts | 98 +++++++++++++++++++++++--------- tests/lib/resumeBuffer.test.ts | 100 +++++++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 27 deletions(-) diff --git a/src/lib/resumeBuffer.ts b/src/lib/resumeBuffer.ts index c6ca5c1..7f3568f 100644 --- a/src/lib/resumeBuffer.ts +++ b/src/lib/resumeBuffer.ts @@ -111,21 +111,37 @@ export type ResumeBufferKeyParts = { * order-independent on the parts of `filter` that don't carry order of * their own (`typeId`/`kinds` are matched as sets, not sequences, so * re-sending them in a different order must key the same buffer). + * + * Two filters that mean different things must never key the same buffer. + * A buffer opens exactly one `ScopedStack.subscribe()`, carrying the + * filter of whichever connection minted it, so a collision hands one of + * the two colliding connections the *other's* filter: either a silent gap + * (events it asked for and never hears about, with no `reset` to tell it + * so) or frames outside its filter, which "filtering is exact, not + * advisory" forbids. See docs/spec/wire-format.md § Change feed. + * + * Hence an object rather than a positional array: `JSON.stringify` drops + * an undefined *property* but renders an undefined *array element* as + * `null`, which collapsed "no parentId filter" onto `parentId: null` + * ("root records only" — a filter the wire format defines) when both were + * array slots. Absent is now encoded as an absent key, which no present + * value can imitate. Key order is fixed by construction below. */ export function resumeBufferKey(parts: ResumeBufferKeyParts): string { - const typeId = parts.filter.typeId - ? [...(Array.isArray(parts.filter.typeId) ? parts.filter.typeId : [parts.filter.typeId])].sort() + const { filter } = parts; + const typeId = filter.typeId + ? [...(Array.isArray(filter.typeId) ? filter.typeId : [filter.typeId])].sort() : undefined; - const kinds = parts.filter.kinds ? [...parts.filter.kinds].sort() : undefined; - return JSON.stringify([ - parts.principalId, - parts.subjectId, - typeId, - parts.filter.parentId, - parts.filter.entityId, - kinds, - parts.includeRecords, - ]); + const kinds = filter.kinds ? [...filter.kinds].sort() : undefined; + return JSON.stringify({ + principalId: parts.principalId, + subjectId: parts.subjectId, + includeRecords: parts.includeRecords, + ...(typeId !== undefined && { typeId }), + ...(filter.parentId !== undefined && { parentId: filter.parentId }), + ...(filter.entityId !== undefined && { entityId: filter.entityId }), + ...(kinds !== undefined && { kinds }), + }); } export type ResumeBufferRegistryOptions = { @@ -145,6 +161,14 @@ export class ResumeBufferRegistry { private readonly buffers = new Map(); private readonly unsubscribes = new Map(); private readonly evictionTimers = new Map(); + /** + * Buffers whose subscription is still opening. A key lands in `buffers` + * only once its `ScopedStack.subscribe()` has resolved, so this is what + * keeps two connections arriving in the same tick from opening two + * subscriptions for one key — they await the same attempt and share + * whichever way it settles. + */ + private readonly acquiring = new Map>(); constructor(private readonly opts: ResumeBufferRegistryOptions) {} @@ -152,6 +176,12 @@ export class ResumeBufferRegistry { * Get this key's buffer, creating one and opening its scoped subscription * if none is currently retained. `subscribeToStack` is called at most * once per buffer instance. + * + * A failed subscription registers nothing: the next connection on this + * key retries it. Registering a buffer that nothing is feeding would + * hand every later connection on that key a `ready` frame followed by + * permanent silence — a gap with no `reset` to announce it, which is + * the one failure the feed contract calls untrustworthy. */ async acquire( key: string, @@ -165,21 +195,34 @@ export class ResumeBufferRegistry { return existing; } - const buffer = new ResumeBuffer(this.opts.depth); - this.buffers.set(key, buffer); - buffer.liveCount = 1; - // Registered before the promise settles is fine either way — the - // handler closes over `buffer`, not over the unsubscribe function this - // call eventually returns, so nothing here depends on ordering between - // the two. - const unsubscribe = await subscribeToStack((change) => { - buffer.append(change); - }); - // A concurrent release() + eviction could in principle have already - // dropped this key before subscribeToStack() resolved; re-store to be - // sure the unsubscribe we now hold is reachable for cleanup either way. - this.unsubscribes.set(key, unsubscribe); - return buffer; + const inFlight = this.acquiring.get(key); + if (inFlight) { + const buffer = await inFlight; + buffer.liveCount += 1; + return buffer; + } + + const attempt = (async () => { + const buffer = new ResumeBuffer(this.opts.depth); + // Awaited before either map is touched — the handler closes over + // `buffer`, so anything the stack emits while this is in flight is + // already being recorded by the time the buffer becomes reachable. + const unsubscribe = await subscribeToStack((change) => { + buffer.append(change); + }); + this.buffers.set(key, buffer); + this.unsubscribes.set(key, unsubscribe); + return buffer; + })(); + this.acquiring.set(key, attempt); + + try { + const buffer = await attempt; + buffer.liveCount += 1; + return buffer; + } finally { + this.acquiring.delete(key); + } } /** Detach one connection from this key's buffer, starting its retention countdown once none remain. */ @@ -215,6 +258,7 @@ export class ResumeBufferRegistry { closeAll(): void { for (const timer of this.evictionTimers.values()) clearTimeout(timer); this.evictionTimers.clear(); + this.acquiring.clear(); for (const unsubscribe of this.unsubscribes.values()) unsubscribe(); this.unsubscribes.clear(); this.buffers.clear(); diff --git a/tests/lib/resumeBuffer.test.ts b/tests/lib/resumeBuffer.test.ts index 8eebccb..799b612 100644 --- a/tests/lib/resumeBuffer.test.ts +++ b/tests/lib/resumeBuffer.test.ts @@ -119,6 +119,42 @@ describe('resumeBufferKey', () => { const keys = new Set([anon, authed, filtered, withRecords]); expect(keys.size).toBe(4); }); + + // A buffer opens one subscription carrying the filter of whichever + // connection minted it, so a collision here hands one of the two the + // other's filter — a silent gap in one direction, frames outside the + // filter in the other. `parentId` is the case a positional + // JSON.stringify got wrong: `undefined` renders as `null` inside an + // array, so "no parentId filter" and `parentId: null` collapsed onto + // one key. + it('distinguishes an absent parentId filter from parentId: null', () => { + const base = { principalId: null, subjectId: null, includeRecords: false }; + const unfiltered = resumeBufferKey({ ...base, filter: {} }); + const rootsOnly = resumeBufferKey({ ...base, filter: { parentId: null } }); + const oneParent = resumeBufferKey({ ...base, filter: { parentId: 'rec-1' } }); + + expect(new Set([unfiltered, rootsOnly, oneParent]).size).toBe(3); + }); + + it('distinguishes an absent filter field from every value that field can hold', () => { + const base = { principalId: null, subjectId: null, includeRecords: false }; + expect(resumeBufferKey({ ...base, filter: {} })).not.toBe( + resumeBufferKey({ ...base, filter: { entityId: 'e1' } }), + ); + expect(resumeBufferKey({ ...base, filter: {} })).not.toBe( + resumeBufferKey({ ...base, filter: { typeId: [] } }), + ); + expect(resumeBufferKey({ ...base, filter: {} })).not.toBe( + resumeBufferKey({ ...base, filter: { kinds: [] } }), + ); + }); + + it("does not let one field's value imitate another's", () => { + const base = { principalId: null, subjectId: null, includeRecords: false }; + expect(resumeBufferKey({ ...base, filter: { parentId: 'x' } })).not.toBe( + resumeBufferKey({ ...base, filter: { entityId: 'x' } }), + ); + }); }); describe('ResumeBufferRegistry', () => { @@ -214,4 +250,68 @@ describe('ResumeBufferRegistry', () => { expect(unsubscribeA).toHaveBeenCalledTimes(1); expect(unsubscribeB).toHaveBeenCalledTimes(1); }); + + // Registering a buffer nothing is feeding would answer every later + // connection on that key with `ready` and then permanent silence — a gap + // with no `reset` to announce it. + it('registers nothing when the subscription fails, and retries on the next acquire', async () => { + const registry = new ResumeBufferRegistry({ depth: 10, retentionMs: 10_000 }); + const failing = vi.fn(async () => { + throw new Error('adapter unavailable'); + }); + + await expect(registry.acquire('k', failing)).rejects.toThrow('adapter unavailable'); + + const unsubscribe = vi.fn(); + const healthy = subscribeStub(unsubscribe); + const buffer = await registry.acquire('k', healthy); + + expect(healthy).toHaveBeenCalledTimes(1); + expect(buffer.liveCount).toBe(1); + + // And the retried buffer is genuinely fed and genuinely releasable. + registry.release('k'); + expect(buffer.liveCount).toBe(0); + }); + + it('opens one subscription for two acquires that race on the same key', async () => { + const registry = new ResumeBufferRegistry({ depth: 10, retentionMs: 10_000 }); + let resolveSubscribe!: (u: Unsubscribe) => void; + const subscribe = vi.fn( + () => + new Promise((resolve) => { + resolveSubscribe = resolve; + }), + ); + + const first = registry.acquire('k', subscribe); + const second = registry.acquire('k', subscribe); + resolveSubscribe(() => {}); + const [a, b] = await Promise.all([first, second]); + + expect(subscribe).toHaveBeenCalledTimes(1); + expect(a).toBe(b); + expect(a.liveCount).toBe(2); + }); + + it('fails both racing acquires when the shared subscription fails, then retries', async () => { + const registry = new ResumeBufferRegistry({ depth: 10, retentionMs: 10_000 }); + let rejectSubscribe!: (err: Error) => void; + const failing = vi.fn( + () => + new Promise((_resolve, reject) => { + rejectSubscribe = reject; + }), + ); + + const first = registry.acquire('k', failing); + const second = registry.acquire('k', failing); + rejectSubscribe(new Error('adapter unavailable')); + await expect(first).rejects.toThrow('adapter unavailable'); + await expect(second).rejects.toThrow('adapter unavailable'); + + const healthy = subscribeStub(); + await registry.acquire('k', healthy); + expect(healthy).toHaveBeenCalledTimes(1); + }); }); From 1b623b386ce6f15f076e20f9833003bd7b17562c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 29 Aug 2026 02:31:16 +0000 Subject: [PATCH 2/3] fix(changes): release a change-feed connection that ends by throwing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The streamSSE callback had no try/finally and passed no onError, so a throw anywhere in it — the backlog replay awaits ScopedStack.get() per entry against the adapter — skipped clearInterval on both timers, unsubscribe(), and unregister(). That leaked a live core subscription, a listener on the buffer, and a refcount that never returned to zero, which meant the buffer was never evicted. Hono catches the throw itself, so nothing surfaced beyond a bare console.error. Cleanup now runs in a finally. The throw is caught rather than left to hono, which answers one by writing the raw error message to the client as an error frame — internal detail on a stream any anonymous caller can open. A closed connection is already a repair the client knows how to make, and the failure is logged with its request id like every other. Three smaller things in the same path: - The session re-check's lookupToken() had no rejection handler, so an unreachable token store would raise an unhandled rejection rather than closing the stream. It now closes, which is the answer the client already handles via a 401 on reconnect. - The owner-token comparison here was a plain ===, where authMiddleware deliberately uses timingSafeEqual. safeCompare is exported and reused so the same secret is compared the same way in both places. - Keepalives went out as raw writes outside FrameGate, and the backlog replay kept running permission checks after the gate had tripped. Both now go through the gate, so "in-flight frames are bounded" holds for the whole stream rather than for record frames alone. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FH88yDv3Fk2RmLsU7DU4Rz --- src/app.ts | 2 +- src/lib/frameGate.ts | 2 +- src/middleware/auth.ts | 8 +- src/routes/changes.ts | 242 ++++++++++++++++++++--------------- tests/conformance.test.ts | 8 +- tests/routes/changes.test.ts | 113 +++++++++++++++- 6 files changed, 267 insertions(+), 108 deletions(-) diff --git a/src/app.ts b/src/app.ts index 3126040..9f39078 100644 --- a/src/app.ts +++ b/src/app.ts @@ -77,7 +77,7 @@ export function createApp(ctx: StackContext, config: Config, logger: Logger): Ho app.route('/.well-known', wellknownRoutes(ctx, config)); app.route('/health', healthRoutes()); app.route('/records', recordRoutes(ctx, config.queryTimeoutMs)); - app.route('/changes', changeRoutes(ctx, config)); + app.route('/changes', changeRoutes(ctx, config, logger)); app.route('/types', typeRoutes(ctx)); app.route('/attachments', attachmentRoutes(ctx, config.maxAttachmentBytes)); app.route('/entity', entityRoutes(ctx)); diff --git a/src/lib/frameGate.ts b/src/lib/frameGate.ts index ab33721..399d2e6 100644 --- a/src/lib/frameGate.ts +++ b/src/lib/frameGate.ts @@ -25,7 +25,7 @@ export class FrameGate { * means this call (and every one after it) triggered — or already * followed — overflow, and `onOverflow` has fired at most once. */ - send(write: () => Promise): boolean { + send(write: () => Promise): boolean { if (this.overflowed) return false; if (this.pending >= this.maxPending) { this.overflowed = true; diff --git a/src/middleware/auth.ts b/src/middleware/auth.ts index a3fc071..57b400d 100644 --- a/src/middleware/auth.ts +++ b/src/middleware/auth.ts @@ -5,7 +5,13 @@ import type { AppEnv } from '../types.js'; import type { StackContext } from '../stack.js'; import { wireError } from '../wireError.js'; -function safeCompare(a: string, b: string): boolean { +/** + * Constant-time string equality, for comparing a presented credential + * against a configured secret. Exported so every comparison of the owner + * token uses the same one — `GET /changes` re-derives "is this the owner" + * to decide whether to arm its session re-check. + */ +export function safeCompare(a: string, b: string): boolean { const ba = Buffer.from(a); const bb = Buffer.from(b); return ba.length === bb.length && timingSafeEqual(ba, bb); diff --git a/src/routes/changes.ts b/src/routes/changes.ts index efe2d85..fc8ae24 100644 --- a/src/routes/changes.ts +++ b/src/routes/changes.ts @@ -14,6 +14,8 @@ import type { import { StackQueryError } from '@haverstack/core'; import { serializeChange, isValidSeq } from '@haverstack/wire-types'; import type { ChangeResetReason } from '@haverstack/wire-types'; +import type { Logger } from 'pino'; +import { safeCompare } from '../middleware/auth.js'; import { FrameGate } from '../lib/frameGate.js'; import { decodeCursor } from '../lib/resumeCursor.js'; import { ResumeBufferRegistry, resumeBufferKey, type ResumeEntry } from '../lib/resumeBuffer.js'; @@ -103,6 +105,7 @@ function presentedCursorRaw(c: Context, url: URL): string | undefined { export function changeRoutes( ctx: StackContext, config: Config, + logger: Logger, opts: ChangeRouteOptions = {}, ): Hono { const app = new Hono(); @@ -158,125 +161,160 @@ export function changeRoutes( const unregister = ctx.changeStreams.add(() => resolveDone()); const gate = new FrameGate(maxPendingFrames, () => resolveDone()); - function send(event: string, data: unknown, id?: string): void { - gate.send(() => + /** Returns false once the gate has tripped — this connection is closing. */ + function send(event: string, data: unknown, id?: string): boolean { + return gate.send(() => stream.writeSSE({ event, data: JSON.stringify(data), ...(id !== undefined && { id }) }), ); } - let unsubscribe: () => void; + // Assigned inside the try below; torn down in the finally whether the + // connection ended by abort, by a revoked session, or by a throw. A + // throw is caught rather than left to hono, which would answer it by + // writing the raw error message to the client as an `error` frame. + let unsubscribe: (() => void) | undefined; + let keepalive: NodeJS.Timeout | undefined; + let sessionCheck: NodeJS.Timeout | undefined; - if (!resumeEnabled) { - // `resume: false` — a discovery-conformant server that mints no - // cursors at all. Exists only for WellknownRouteOptions' - // matching test toggle; every connection here is answered exactly - // as #82 shipped it. - send('ready', {}); - if (presentedRaw !== undefined) { - const reason: ChangeResetReason = 'not_supported'; - send('reset', { reason }); - } - unsubscribe = await scopeFor(auth).subscribe( - (change: RecordChange) => { - send('record', serializeChange(change)); - }, - { filter, includeRecords }, - ); - } else { - const key = resumeBufferKey({ - principalId: auth?.principalId ?? null, - subjectId: auth?.subjectId ?? null, - filter, - includeRecords, - }); - const scoped = scopeFor(auth); - const buffer = await resumeBuffers.acquire(key, (onChange) => - scoped.subscribe(onChange, { filter, includeRecords }), - ); + try { + if (!resumeEnabled) { + // `resume: false` — a discovery-conformant server that mints no + // cursors at all. Exists only for WellknownRouteOptions' + // matching test toggle; every connection here is answered exactly + // as #82 shipped it. + send('ready', {}); + if (presentedRaw !== undefined) { + const reason: ChangeResetReason = 'not_supported'; + send('reset', { reason }); + } + unsubscribe = await scopeFor(auth).subscribe( + (change: RecordChange) => { + send('record', serializeChange(change)); + }, + { filter, includeRecords }, + ); + } else { + const key = resumeBufferKey({ + principalId: auth?.principalId ?? null, + subjectId: auth?.subjectId ?? null, + filter, + includeRecords, + }); + const scoped = scopeFor(auth); + const buffer = await resumeBuffers.acquire(key, (onChange) => + scoped.subscribe(onChange, { filter, includeRecords }), + ); - // Attached before anything below reads or awaits, so nothing this - // connection is owed can land in the gap between "what the backlog - // snapshot covers" and "what the live buffer starts reporting" — - // there isn't one. `replaying` buffers what arrives live while the - // (possibly awaited, permission-rechecking) backlog replay below is - // still in flight, so a change appended mid-replay can never be - // sent out of order ahead of an older, still-pending backlog frame. - let replaying = true; - const queued: ResumeEntry[] = []; - const detachLive = buffer.subscribeLive((entry) => { - if (replaying) queued.push(entry); - else send('record', entry.frame, entry.frame.seq); - }); + // Attached before anything below reads or awaits, so nothing this + // connection is owed can land in the gap between "what the backlog + // snapshot covers" and "what the live buffer starts reporting" — + // there isn't one. `replaying` buffers what arrives live while the + // (possibly awaited, permission-rechecking) backlog replay below is + // still in flight, so a change appended mid-replay can never be + // sent out of order ahead of an older, still-pending backlog frame. + let replaying = true; + const queued: ResumeEntry[] = []; + const detachLive = buffer.subscribeLive((entry) => { + if (replaying) queued.push(entry); + else send('record', entry.frame, entry.frame.seq); + }); + // Registered before the first `await` below, so a subscription + // opened above is always released — including when that await + // throws. Everything after this point runs under the finally. + unsubscribe = () => { + detachLive(); + resumeBuffers.release(key); + }; - let resetReason: ChangeResetReason | undefined; - let backlog: ResumeEntry[] = []; - if (presentedRaw !== undefined) { - const decoded = decodeCursor(presentedRaw); - if (!decoded || decoded.bufferId !== buffer.id) { - // Unrecognized outright, or names a buffer instance that isn't - // this key's current one (evicted past its retention window, - // or minted for a filter this cursor doesn't actually match). - resetReason = 'cursor_expired'; - } else { - const outcome = buffer.entriesAfter(decoded.n); - if (outcome.status === 'ok') backlog = outcome.entries; - else resetReason = outcome.status; + let resetReason: ChangeResetReason | undefined; + let backlog: ResumeEntry[] = []; + if (presentedRaw !== undefined) { + const decoded = decodeCursor(presentedRaw); + if (!decoded || decoded.bufferId !== buffer.id) { + // Unrecognized outright, or names a buffer instance that isn't + // this key's current one (evicted past its retention window, + // or minted for a filter this cursor doesn't actually match). + resetReason = 'cursor_expired'; + } else { + const outcome = buffer.entriesAfter(decoded.n); + if (outcome.status === 'ok') backlog = outcome.entries; + else resetReason = outcome.status; + } } - } - send('ready', { seq: buffer.headCursor() }); + send('ready', { seq: buffer.headCursor() }); - if (resetReason) { - send('reset', { reason: resetReason }); - } else { - for (const entry of backlog) { - // Purge frames aren't re-checkable — the mutation-time decision - // is the only one that will ever exist (docs/spec/events.md, - // quoted on issue #84). A non-purge frame's record ID is still - // in hand, so a grant revoked during the gap is caught here. - if (!entry.isPurge) { - const stillReadable = await scoped.get(entry.recordId); - if (!stillReadable) continue; + if (resetReason) { + send('reset', { reason: resetReason }); + } else { + for (const entry of backlog) { + // Purge frames aren't re-checkable — the mutation-time decision + // is the only one that will ever exist (docs/spec/events.md, + // quoted on issue #84). A non-purge frame's record ID is still + // in hand, so a grant revoked during the gap is caught here. + if (!entry.isPurge) { + const stillReadable = await scoped.get(entry.recordId); + if (!stillReadable) continue; + } + // Once the gate trips this connection is closing, so the + // remaining entries are permission checks nobody will read. + if (!send('record', entry.frame, entry.frame.seq)) break; } - send('record', entry.frame, entry.frame.seq); } - } - replaying = false; - for (const entry of queued) { - send('record', entry.frame, entry.frame.seq); + replaying = false; + for (const entry of queued) { + if (!send('record', entry.frame, entry.frame.seq)) break; + } } - unsubscribe = () => { - detachLive(); - resumeBuffers.release(key); - }; - } - - const keepalive = setInterval(() => { - void stream.write(': keepalive\n\n'); - }, keepaliveMs); - - // A revoked or expired token must stop delivering. The owner's - // static bearer token never expires (authMiddleware compares it - // directly, not via the token store), so only a minted session needs - // re-checking; an anonymous connection has no token to revoke. - const isOwnerToken = bearerToken === config.ownerToken; - const sessionCheck = - auth && !isOwnerToken && bearerToken - ? setInterval(() => { - void ctx.tokens.lookupToken(bearerToken).then((session) => { - if (!session) resolveDone(); - }); - }, sessionCheckMs) - : undefined; + // Through the gate like any other write: a client not draining its + // keepalives isn't draining anything, and counting them is what + // keeps "in-flight frames are bounded" true of the whole stream + // rather than of `record` frames alone. + keepalive = setInterval(() => { + gate.send(() => stream.write(': keepalive\n\n')); + }, keepaliveMs); - await done; + // A revoked or expired token must stop delivering. The owner's + // static bearer token never expires (authMiddleware compares it + // directly, not via the token store), so only a minted session needs + // re-checking; an anonymous connection has no token to revoke. + const isOwnerToken = + bearerToken !== undefined && safeCompare(bearerToken, config.ownerToken); + sessionCheck = + auth && !isOwnerToken && bearerToken + ? setInterval(() => { + void ctx.tokens.lookupToken(bearerToken).then( + (session) => { + if (!session) resolveDone(); + }, + // The token store is unreachable. Closing is the honest + // answer — the client reconnects and re-authenticates + // through the path that already handles a 401 — and it + // keeps a rejection here from going unhandled. + () => resolveDone(), + ); + }, sessionCheckMs) + : undefined; - clearInterval(keepalive); - if (sessionCheck) clearInterval(sessionCheck); - unsubscribe(); - unregister(); + await done; + } catch (err) { + // Hono answers a throw from this callback by writing the raw error + // message to the client as an `error` frame; catching it here keeps + // internal detail off a stream any anonymous caller can open, and + // the closed connection is already a repair the client knows how to + // make (reconnect, present the cursor, take frames or a `reset`). + logger.error( + { err, requestId: c.get('requestId') }, + 'Change feed connection ended with an error', + ); + } finally { + if (keepalive) clearInterval(keepalive); + if (sessionCheck) clearInterval(sessionCheck); + unsubscribe?.(); + unregister(); + } }); }); diff --git a/tests/conformance.test.ts b/tests/conformance.test.ts index cba79ca..9a7f973 100644 --- a/tests/conformance.test.ts +++ b/tests/conformance.test.ts @@ -55,6 +55,7 @@ import { testConfig, TEST_TOKEN, TEST_ENTITY_ID, + logger, type TestApp, } from './setup.js'; import { openChangeFeed, type DecodedFrame } from './changeFeedClient.js'; @@ -1292,7 +1293,10 @@ describe('changeFeed fixtures', () => { // `changeFeedResume: false` app above. const resumeDisabledApp = new Hono(); resumeDisabledApp.use(authMiddleware(testConfig(t.dbPath).ownerToken, t.ctx)); - resumeDisabledApp.route('/', changeRoutes(t.ctx, testConfig(t.dbPath), { resume: false })); + resumeDisabledApp.route( + '/', + changeRoutes(t.ctx, testConfig(t.dbPath), logger, { resume: false }), + ); const conn = await openChangeFeed(resumeDisabledApp, '/', { token: TEST_TOKEN, headers: { 'Last-Event-ID': 'AA3f1R' }, @@ -1395,7 +1399,7 @@ describe('changeFeed sequence fixtures', () => { resumeApp.use(authMiddleware(config.ownerToken, t.ctx)); resumeApp.route( '/changes', - changeRoutes(t.ctx, config, { resume: true, resumeRetentionMs: 10 }), + changeRoutes(t.ctx, config, logger, { resume: true, resumeRetentionMs: 10 }), ); const first = await openChangeFeed(resumeApp, fixture.steps[0]!.path, { token: TEST_TOKEN }); diff --git a/tests/routes/changes.test.ts b/tests/routes/changes.test.ts index 8a1867f..5115111 100644 --- a/tests/routes/changes.test.ts +++ b/tests/routes/changes.test.ts @@ -7,6 +7,7 @@ import { testConfig, tempDbPath, TEST_TOKEN, + logger, type TestApp, } from '../setup.js'; import { changeRoutes, type ChangeRouteOptions } from '../../src/routes/changes.js'; @@ -30,7 +31,7 @@ const CONTRIBUTOR_ID = 'entity-contributor-789'; function testChangesApp(ctx: StackContext, config: Config, opts: ChangeRouteOptions = {}) { const app = new Hono(); app.use(authMiddleware(config.ownerToken, ctx)); - app.route('/', changeRoutes(ctx, config, opts)); + app.route('/', changeRoutes(ctx, config, logger, opts)); return app; } @@ -321,4 +322,114 @@ describe('GET /changes', () => { } }); }); + + // Two connections sharing a resume buffer share its single subscription, + // and so its single filter. `?parentId=null` ("root records only") and no + // parentId filter at all mean different things and must never share one: + // whichever connected first would otherwise decide what the other + // receives — a silent gap in one direction, frames outside the filter in + // the other, and no `reset` either way to announce it. + describe('buffer keying across distinct filters', () => { + async function childWrite(): Promise { + const parent = await t.ctx.stack.create(NOTE_TYPE, { title: 'parent' }); + const child = await t.ctx.stack.create( + NOTE_TYPE, + { title: 'child' }, + { parentId: parent.id }, + ); + return child.id; + } + + it('delivers a child change to an unfiltered connection opened after a ?parentId=null one', async () => { + const rootsOnly = await openChangeFeed(t.app, '/changes?parentId=null', { + token: TEST_TOKEN, + }); + await rootsOnly.waitForFrames(1); + const unfiltered = await openChangeFeed(t.app, '/changes', { token: TEST_TOKEN }); + await unfiltered.waitForFrames(1); + + try { + const childId = await childWrite(); + const frames = await unfiltered.waitForFrames(3); + const child = frames.find( + (f) => f.event === 'record' && (f.data as { recordId: string }).recordId === childId, + ); + expect(child).toBeDefined(); + expect(unfiltered.frames.some((f) => f.event === 'reset')).toBe(false); + } finally { + await unfiltered.close(); + await rootsOnly.close(); + } + }); + + it('withholds a child change from a ?parentId=null connection opened after an unfiltered one', async () => { + const unfiltered = await openChangeFeed(t.app, '/changes', { token: TEST_TOKEN }); + await unfiltered.waitForFrames(1); + const rootsOnly = await openChangeFeed(t.app, '/changes?parentId=null', { + token: TEST_TOKEN, + }); + await rootsOnly.waitForFrames(1); + + try { + const childId = await childWrite(); + // The parent is a root record, so it does arrive — waiting on it + // proves the connection is live and that the child's absence below + // is a filter decision rather than a race. + await rootsOnly.waitForFrames(2); + const leaked = rootsOnly.frames.filter( + (f) => f.event === 'record' && (f.data as { recordId: string }).recordId === childId, + ); + expect(leaked).toHaveLength(0); + } finally { + await rootsOnly.close(); + await unfiltered.close(); + } + }); + }); + + it('closes the connection when the token store cannot answer a session re-check', async () => { + const dbPath = tempDbPath(); + const ctx = await createTestContext(dbPath); + await ctx.stack.defineType(NOTE_TYPE, 'Note', { title: { kind: 'string' } }); + const config = testConfig(dbPath); + const { token } = await ctx.adapter.createToken(CONTRIBUTOR_ID); + const app = testChangesApp(ctx, config, { sessionCheckMs: 20, keepaliveMs: 60_000 }); + + // The store is reachable at connect (auth succeeds) and unreachable by + // the time the re-check runs. Closing is the honest answer: the client + // reconnects through the path that already handles a 401. Left + // unhandled, the rejection would take the process down instead. + const lookupToken = ctx.tokens.lookupToken.bind(ctx.tokens); + let authenticated = false; + ctx.tokens.lookupToken = async (t: string) => { + if (authenticated) throw new Error('token store unavailable'); + authenticated = true; + return lookupToken(t); + }; + + try { + const res = await app.request('/', { + headers: { Accept: 'text/event-stream', Authorization: `Bearer ${token}` }, + }); + expect(res.status).toBe(200); + + const reader = res.body!.getReader(); + const drained = await Promise.race([ + (async () => { + while (true) { + const { done } = await reader.read(); + if (done) return true; + } + })(), + new Promise((resolve) => setTimeout(() => resolve(false), 2000)), + ]); + expect(drained).toBe(true); + } finally { + ctx.tokens.lookupToken = lookupToken; + await ctx.queryWorker.close(); + await ctx.stack.close(); + await ctx.tokens.close(); + ctx.nonces.close(); + } + }); }); From 1f774eb366359c82554392b8df927cf38ee4a0e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 29 Aug 2026 12:07:57 +0000 Subject: [PATCH 3/3] fix(discovery): stop letting discovery contradict the change feed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WellknownRouteOptions` carried two flags, `changeFeedResume` and `changeFeedRecords`, whose only effect was to print a different literal in the discovery response. Neither was a deployer lever, and `createApp()` passed neither, so production was consistent by two defaults happening to line up rather than by construction. `changeFeedRecords` was the worse of the two: there is no route-side counterpart at all. `parseIncludeRecord()` honors `?include=record` unconditionally, so the flag could only ever make discovery advertise something this server does not do. That is not cosmetic — a client is entitled to act on discovery without asking again, since `APIAdapter.subscribeChanges()` against a server advertising no feed throws locally without sending a request. What the flags bought was one conformance fixture, discovery-advertises-a-feed-that-neither-resumes-nor-includes-records, which describes a conformant server with neither capability. This server is not that server, and satisfying a fixture about a different implementation is not worth an option that lets this one misreport itself. The fixture is skipped with that reason, using the SKIPPED set assertCoverage() already takes and the createRecord block already uses. Both fields are now literals. `ChangeRouteOptions.resume` stays: unlike these, it drives real spec-defined behavior — `ready` with no `seq`, then `reset` with reason `not_supported` — that a behavioral fixture exercises and that needs a way to be reached. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01FH88yDv3Fk2RmLsU7DU4Rz --- src/routes/changes.ts | 10 ++++--- src/routes/wellknown.ts | 52 +++++++++++----------------------- tests/conformance.test.ts | 36 +++++++---------------- tests/routes/wellknown.test.ts | 27 +----------------- 4 files changed, 33 insertions(+), 92 deletions(-) diff --git a/src/routes/changes.ts b/src/routes/changes.ts index fc8ae24..61d53a2 100644 --- a/src/routes/changes.ts +++ b/src/routes/changes.ts @@ -49,10 +49,12 @@ export type ChangeRouteOptions = { sessionCheckMs?: number; maxPendingFrames?: number; /** - * Whether a presented cursor is honored at all. Default true. Exists - * only so a test can exercise the `resume: false` branch of the wire - * contract (see WellknownRouteOptions.changeFeedResume) — there's no - * real deployer lever here once #84 lands. + * Whether a presented cursor is honored at all. Default true, and never + * false in production — #84 made this server a resuming one, and + * discovery says so unconditionally. It exists because `resume: false` + * is real, spec-defined behavior (`ready` with no `seq`, then `reset` + * with reason `not_supported`) that a conformance fixture exercises, and + * that behavior needs a way to be reached. */ resume?: boolean; resumeBufferDepth?: number; diff --git a/src/routes/wellknown.ts b/src/routes/wellknown.ts index 109ac3f..dc0046d 100644 --- a/src/routes/wellknown.ts +++ b/src/routes/wellknown.ts @@ -9,32 +9,7 @@ import type { AppEnv } from '../types.js'; import type { StackContext } from '../stack.js'; import type { Config } from '../config.js'; -export type WellknownRouteOptions = { - /** - * Whether `?include=record` is honored on `GET /changes`. Always `true` - * in production — `src/routes/changes.ts` honors it unconditionally, so - * there's no real deployer lever here. This exists only so a test can - * exercise the `records: false` branch of the wire contract (both flags - * false is fully conformant — see docs/spec/wire-format.md § Change - * feed) without `src/routes/changes.ts` growing a matching toggle it - * doesn't otherwise need. - */ - changeFeedRecords?: boolean; - /** - * Whether `GET /changes` honors a resume cursor. Always `true` in - * production (#84) — mirrors `changeRoutes()`'s own `resume` option - * (`ChangeRouteOptions.resume`), which a test threads through alongside - * this one so the discovery response and the route's actual behavior - * never disagree. - */ - changeFeedResume?: boolean; -}; - -export function wellknownRoutes( - ctx: StackContext, - config: Config, - opts: WellknownRouteOptions = {}, -): Hono { +export function wellknownRoutes(ctx: StackContext, config: Config): Hono { const app = new Hono(); app.get('/stack', (c) => { @@ -61,18 +36,23 @@ export function wellknownRoutes( // adapter's own capabilities do, so it's added explicitly. An object // rather than a boolean for the same reason `auth` is: the surface // grows entries (another transport, batched frames) rather than - // gaining a second and third boolean alongside it. Advertise what's - // true, not what's aspirational — a client that calls - // subscribeChanges() against a server advertising no feed fails - // locally at open(), which is strictly better than discovering a 404 - // partway through a connection. `resume: true` since #84 mints - // cursors and GET /changes honors Last-Event-ID/?since=; `records: - // true` because `GET /changes` already honors `?include=record` - // unconditionally (#82). + // gaining a second and third boolean alongside it. + // + // Both literals, with no override: this server resumes (#84 mints + // cursors and GET /changes honors Last-Event-ID/?since=) and honors + // `?include=record` (#82), so those are the only true answers it can + // give. A client is entitled to act on this without asking again — + // `APIAdapter.subscribeChanges()` against a server advertising no + // feed throws locally, without sending a request — which makes an + // override that could report otherwise a way to make this response + // lie about the route next to it. The conformant both-false shape is + // a different server's discovery response, not a mode of this one; + // the fixture describing it is skipped in tests/conformance.test.ts + // for exactly that reason. changes: { transports: [CHANGE_TRANSPORT_SSE], - resume: opts.changeFeedResume ?? true, - records: opts.changeFeedRecords ?? true, + resume: true, + records: true, }, }; return c.json(body); diff --git a/tests/conformance.test.ts b/tests/conformance.test.ts index 9a7f973..670a006 100644 --- a/tests/conformance.test.ts +++ b/tests/conformance.test.ts @@ -59,7 +59,6 @@ import { type TestApp, } from './setup.js'; import { openChangeFeed, type DecodedFrame } from './changeFeedClient.js'; -import { wellknownRoutes } from '../src/routes/wellknown.js'; import { changeRoutes } from '../src/routes/changes.js'; import { authMiddleware } from '../src/middleware/auth.js'; import type { AppEnv } from '../src/types.js'; @@ -127,6 +126,15 @@ function assertCoverage(names: string[], handled: Set, skipped: Set { + const SKIPPED = new Set([ + // Describes a conformant server that neither resumes nor includes + // records. This server does both — #84 mints and honors cursors, #82 + // honors `?include=record` — so the only way to produce this response + // would be an override that lets discovery contradict the route beside + // it, on a field clients are entitled to act on without asking again. + // The shape is exercised by whichever implementation actually has it. + 'discovery-advertises-a-feed-that-neither-resumes-nor-includes-records', + ]); const handled = new Set(); test('discovery-declares-protocol-version-and-capabilities', async () => { @@ -187,35 +195,11 @@ describe('discovery fixtures', () => { expect(changes!.records).toBe(true); }); - test('discovery-advertises-a-feed-that-neither-resumes-nor-includes-records', async () => { - const fixture = discoveryFixtures.find( - (f) => f.name === 'discovery-advertises-a-feed-that-neither-resumes-nor-includes-records', - )!; - handled.add(fixture.name); - // Both flags false is fully conformant, but this server always honors - // both resume and `?include=record` — it has no real deployer toggle - // for either (see WellknownRouteOptions in src/routes/wellknown.ts). - // Build a standalone app that passes the test-only overrides directly, - // the same way tests/routes/changes.test.ts bypasses createApp() to - // pass ChangeRouteOptions createApp() has no way to thread through. - const app = new Hono(); - app.route( - '/.well-known', - wellknownRoutes(t.ctx, testConfig(t.dbPath), { - changeFeedRecords: false, - changeFeedResume: false, - }), - ); - const { status, data } = await req(app, fixture.method, fixture.path); - expect(status).toBe(fixture.responseStatus); - expect((data as { changes?: unknown }).changes).toEqual(fixture.responseBody!.changes); - }); - test('coverage', () => { assertCoverage( discoveryFixtures.map((f) => f.name), handled, - new Set(), + SKIPPED, ); }); }); diff --git a/tests/routes/wellknown.test.ts b/tests/routes/wellknown.test.ts index 51270e8..983aefb 100644 --- a/tests/routes/wellknown.test.ts +++ b/tests/routes/wellknown.test.ts @@ -1,9 +1,6 @@ -import { Hono } from 'hono'; import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import { WIRE_PROTOCOL_VERSION } from '@haverstack/wire-types'; -import { buildTestApp, req, testConfig, TEST_ENTITY_ID, type TestApp } from '../setup.js'; -import { wellknownRoutes } from '../../src/routes/wellknown.js'; -import type { AppEnv } from '../../src/types.js'; +import { buildTestApp, req, TEST_ENTITY_ID, type TestApp } from '../setup.js'; describe('GET /.well-known/stack', () => { let t: TestApp; @@ -63,26 +60,4 @@ describe('GET /.well-known/stack', () => { records: true, }); }); - - it('can advertise a feed that neither resumes nor includes records', async () => { - // This server has no real deployer-facing toggle for either flag — GET - // /changes always resumes and always honors ?include=record. - // changeFeedResume/changeFeedRecords are test-only overrides on - // wellknownRoutes() itself for exercising the (fully conformant) - // both-false branch of the wire contract. - const app = new Hono(); - app.route( - '/.well-known', - wellknownRoutes(t.ctx, testConfig(t.dbPath), { - changeFeedRecords: false, - changeFeedResume: false, - }), - ); - const { data } = await req(app, 'GET', '/.well-known/stack'); - expect((data as { changes?: unknown }).changes).toEqual({ - transports: ['sse'], - resume: false, - records: false, - }); - }); });