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/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/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..61d53a2 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'; @@ -47,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; @@ -103,6 +107,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 +163,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/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 cba79ca..670a006 100644 --- a/tests/conformance.test.ts +++ b/tests/conformance.test.ts @@ -55,10 +55,10 @@ import { testConfig, TEST_TOKEN, TEST_ENTITY_ID, + logger, 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'; @@ -126,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 () => { @@ -186,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, ); }); }); @@ -1292,7 +1277,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 +1383,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/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); + }); }); 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(); + } + }); }); 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, - }); - }); });