From 17a32ec654ecfed36e97cceb7d28b2b91bb81111 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 26 Jul 2026 00:45:00 +0000 Subject: [PATCH 1/2] fix: warm ACL cache on _changes miss before deny Continuous/_changes filtering no longer drops readable docs solely because the in-memory ACL row was cold. Misses call ensureDocRows, then authorize; true denials and fail-closed behavior remain after ensure. Bump to 1.7.0. Co-authored-by: Peter Baker --- CHANGELOG.md | 17 ++ README.md | 1 + package.json | 2 +- src/acl/lookup.ts | 23 ++- src/proxy/filterChanges.ts | 197 +++++++++++++++--- src/routes/actors.ts | 18 ++ test/integration/security-edges.test.ts | 153 ++++++++++++++ test/unit/acl-lookup.test.ts | 28 ++- test/unit/filter-changes.test.ts | 260 +++++++++++++++++++++++- 9 files changed, 655 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a062ba2..4c4ebd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [1.7.0] - 2026-07-26 + +### Fixed + +- Live `_changes` filtering no longer denies solely because an ACL row is missing + from the in-memory cache (`missing-row-create-path`). Continuous / eventsource / + normal / longpoll feeds warm cold ids via `ensureDocRows` / `cache.ensureDocs`, + then authorize with `resolveDocAcl` (row + parent + `dbacl`). True denials and + create-path / not-found semantics remain after a successful ensure; view/admin + failures still fail closed. Same warm-before-filter applied to `_all_docs` / + views / `_find` response filtering. + +### Changed + +- `filterChangesStream` now takes `AclCache` so stream filters can warm on miss. +- Added `canReadEnsured` for ensure-then-authorize on single-id read paths. + ## [1.6.0] - 2026-07-25 ### Added diff --git a/README.md b/README.md index 6d3e544..22221fc 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,7 @@ Unmapped endpoints return **404** for non-admins (default-deny). `_list`, `_show - Principal-dependent list responses disable shared validators and caching so an old authorized representation cannot survive an ACL or role change. - Filtered row responses omit unfiltered `total_rows`, `offset`, and `update_seq`; Mango responses omit unfiltered execution statistics for non-admins. - Continuous `_changes` sequences are opaque strings (Couch 2+/3); never treat them as integers. +- Live `_changes` (continuous / longpoll / eventsource) may briefly await an ACL view / `_all_docs` warm when a change id is not yet in the in-memory cache. Cold cache never denies by itself; true denials happen only after ensure. Ensure/view failures still fail closed (**503** / drop). - ACL cache is ~hundreds of bytes per doc per process; preload via `COUCH_PRELOAD_DBS` and/or `COUCH_PRELOAD_DB_INCLUDE`. - Initial ACL view loads are paginated; the in-memory cache still contains one compact row per document. - ACL view/admin failures and a down `_changes` follower fail closed (**503**), never serve on a stale cache. diff --git a/package.json b/package.json index d9571a0..4a6c9b9 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "couch-auth-proxy", - "version": "1.6.0", + "version": "1.7.0", "private": true, "description": "Per-document r/w/d ACL proxy for Apache CouchDB 3.5+", "license": "MIT", diff --git a/src/acl/lookup.ts b/src/acl/lookup.ts index 51da6f6..d8957ac 100644 --- a/src/acl/lookup.ts +++ b/src/acl/lookup.ts @@ -1,9 +1,11 @@ /** * Synchronous ACL lookups against a ready `DbAclState`. * - * Actors call these after `AclCache.requireReady` (and often `ensureDocRow`) - * to decide whether a principal may read/write/delete a document. Fail-closed - * for reads when the row is not yet cached — never briefly open a doc. + * Actors call these after `AclCache.requireReady` and, for ids that may be + * absent from the in-memory map, `ensureDocRow(s)` — then use the sync helpers. + * A missing row is fail-closed for reads (`missing-row-create-path`) so a cold + * cache never briefly opens a doc; live `_changes` / list filters must warm + * via `ensureDocRows` before treating that as a real deny. * * Verbose logs (`LOG_LEVEL=verbose`) explain missing-row create paths, * design-doc denials, and noacl/admin short-circuits. @@ -104,6 +106,21 @@ export function canDelete(state: DbAclState, principal: Principal, docId: string return flagsForDoc(state, principal, docId)._d; } +/** + * Warm the ACL row for `docId` (if missing), then authorize read. + * Use on paths that observe document ids from Couch before the changes + * follower may have written them into the in-memory map. + */ +export async function canReadEnsured( + cache: AclCache, + state: DbAclState, + principal: Principal, + docId: string, +): Promise { + await ensureDocRows(cache, state, [docId]); + return canRead(state, principal, docId); +} + /** * Ensure the ACL row (and parent row, if any) is present before single-doc checks. * Fetches missing rows via the admin ACL view — opaque-seq safe (keyed by id). diff --git a/src/proxy/filterChanges.ts b/src/proxy/filterChanges.ts index 7b4a85e..615463c 100644 --- a/src/proxy/filterChanges.ts +++ b/src/proxy/filterChanges.ts @@ -4,10 +4,15 @@ * Supports continuous (NDJSON), eventsource, and normal/longpoll JSON feeds. * Opaque `seq` / `last_seq` values pass through unchanged. Heartbeats and * non-change control lines are forwarded so clients keep the feed alive. + * + * Cold ACL-cache misses are warmed via `ensureDocRows` before authorize — a + * missing in-memory row must not drop a live change that the principal can + * read once the view/`_all_docs` reconcile catches up. Continuous feeds never + * redeliver a seq, so deny-on-cold would permanently hide the document. */ import type { Principal } from "../auth/types.js"; -import type { DbAclState } from "../acl/cache.js"; -import { canRead } from "../acl/lookup.js"; +import type { AclCache, DbAclState } from "../acl/cache.js"; +import { canRead, ensureDocRows } from "../acl/lookup.js"; import { BodyTooLargeError, limitBytes } from "../util/limitStream.js"; import { createLogger, isLevelEnabled } from "../util/log.js"; import { addProfileMs } from "../util/profile.js"; @@ -32,9 +37,12 @@ export type FilterChangesOptions = { /** * Stream-filter a Couch `_changes` response for the given feed style. + * + * `cache` is required so missing ACL rows can be warmed before sync `canRead`. */ export function filterChangesStream( upstream: ReadableStream, + cache: AclCache, state: DbAclState, principal: Principal, feed: string, @@ -50,9 +58,9 @@ export function filterChangesStream( } const maxBytes = options?.maxBufferBytes ?? 50 * 1024 * 1024; if (mode === "continuous" || mode === "eventsource") { - return filterLineFeed(upstream, state, principal, mode === "eventsource", maxBytes); + return filterLineFeed(upstream, cache, state, principal, mode === "eventsource", maxBytes); } - return filterJsonChanges(upstream, state, principal, maxBytes); + return filterJsonChanges(upstream, cache, state, principal, maxBytes); } function normalizeFeed(feed: string): string { @@ -69,11 +77,54 @@ function normalizeFeed(feed: string): string { return "normal"; } +/** True when sync `canRead` would hit the missing-row create path. */ +function needsAclWarm(state: DbAclState, principal: Principal, docId: string): boolean { + if (principal.admin || state.noacl) return false; + if (!docId || typeof docId !== "string") return false; + return !state.acl.has(docId); +} + +/** Collect document ids from continuous / SSE lines that still need a warm. */ +function collectMissingIdsFromLines( + lines: string[], + eventsource: boolean, + state: DbAclState, + principal: Principal, +): string[] { + const missing: string[] = []; + for (const rawLine of lines) { + const id = changeIdFromLine(rawLine, eventsource); + if (id && needsAclWarm(state, principal, id)) missing.push(id); + } + return missing; +} + +function changeIdFromLine(rawLine: string, eventsource: boolean): string | null { + const line = rawLine.replace(/\r$/, ""); + if (!line.trim()) return null; + let payload = line; + if (eventsource) { + if (!line.startsWith("data:")) return null; + payload = line.slice(5).trim(); + if (!payload) return null; + } + try { + const obj = JSON.parse(payload) as ChangeLine; + if (typeof obj.id === "string" && obj.id) return obj.id; + } catch { + // ignore malformed — processLine drops them + } + return null; +} + /** * Filter continuous NDJSON or Server-Sent Events line-by-line with backpressure. + * Missing ACL rows in each read chunk are batched through `ensureDocRows` before + * sync filtering so a cold cache cannot permanently drop a continuous seq. */ function filterLineFeed( upstream: ReadableStream, + cache: AclCache, state: DbAclState, principal: Principal, eventsource: boolean, @@ -82,6 +133,9 @@ function filterLineFeed( const reader = upstream.getReader(); const decoder = new TextDecoder(); let buffer = ""; + /** Complete lines warmed but not yet filtered (survives backpressure yields). */ + let pendingLines: string[] = []; + let upstreamDone = false; /** For eventsource: only forward `id:` lines after an allowed `data:` line. */ let lastEsDataAllowed = false; @@ -93,34 +147,65 @@ function filterLineFeed( await sleep(1); } - const { done, value } = await reader.read(); - if (done) { - const t0 = performance.now(); - flushLine(buffer); - addProfileMs("filter", performance.now() - t0); - controller.close(); - return; + if (pendingLines.length === 0 && !upstreamDone) { + const { done, value } = await reader.read(); + if (done) { + upstreamDone = true; + if (buffer.length > 0) { + pendingLines = [buffer]; + buffer = ""; + } + } else { + buffer += decoder.decode(value, { stream: true }); + + // Bound incomplete-line buffer (malformed/huge lines). + if (buffer.length > maxBufferBytes && !buffer.includes("\n")) { + controller.error(new BodyTooLargeError(maxBufferBytes)); + void reader.cancel(); + return; + } + + const completeLines: string[] = []; + let newlineIndex: number; + while ((newlineIndex = buffer.indexOf("\n")) >= 0) { + completeLines.push(buffer.slice(0, newlineIndex)); + buffer = buffer.slice(newlineIndex + 1); + } + pendingLines = completeLines; + } + + if (pendingLines.length > 0) { + const tWarm = performance.now(); + try { + await warmMissingForLines(pendingLines, eventsource); + } catch (err) { + addProfileMs("filter", performance.now() - tWarm); + controller.error(err); + void reader.cancel(); + return; + } + addProfileMs("filter", performance.now() - tWarm); + } } - buffer += decoder.decode(value, { stream: true }); - // Bound incomplete-line buffer (malformed/huge lines). - if (buffer.length > maxBufferBytes && !buffer.includes("\n")) { - controller.error(new BodyTooLargeError(maxBufferBytes)); - void reader.cancel(); - return; + if (pendingLines.length === 0) { + if (upstreamDone) { + controller.close(); + return; + } + continue; } const t0 = performance.now(); - let newlineIndex: number; let enqueued = false; - while ((newlineIndex = buffer.indexOf("\n")) >= 0) { - const rawLine = buffer.slice(0, newlineIndex); - buffer = buffer.slice(newlineIndex + 1); + while (pendingLines.length > 0) { + const rawLine = pendingLines.shift()!; const out = processLine(rawLine, eventsource); if (out != null) { controller.enqueue(textEncoder.encode(out + "\n")); enqueued = true; // Yield after an allowed line so desiredSize can apply backpressure. + // Remaining pendingLines stay queued for the next pull. if (controller.desiredSize !== null && controller.desiredSize <= 0) { addProfileMs("filter", performance.now() - t0); return; @@ -129,11 +214,24 @@ function filterLineFeed( } addProfileMs("filter", performance.now() - t0); if (enqueued) return; + if (upstreamDone) { + controller.close(); + return; + } } - function flushLine(line: string) { - const out = processLine(line, eventsource); - if (out != null) controller.enqueue(textEncoder.encode(out + "\n")); + async function warmMissingForLines(lines: string[], es: boolean): Promise { + const missing = collectMissingIdsFromLines(lines, es, state, principal); + if (missing.length === 0) return; + if (isLevelEnabled("verbose")) { + log.verbose("changes-cache-miss-warm", { + db: state.name, + user: principal.name, + count: missing.length, + feed: es ? "eventsource" : "continuous", + }); + } + await ensureDocRows(cache, state, missing); } function processLine(rawLine: string, es: boolean): string | null { @@ -168,7 +266,22 @@ function filterLineFeed( // must pass the ACL check. if (obj.id == null) return obj.last_seq != null ? line : null; if (typeof obj.id !== "string") return null; - if (!canRead(state, principal, obj.id)) return null; + if (!canRead(state, principal, obj.id)) { + if ( + isLevelEnabled("verbose") && + !state.acl.has(obj.id) && + !principal.admin && + !state.noacl + ) { + log.verbose("missing-row-deny-after-ensure", { + db: state.name, + user: principal.name, + docId: obj.id, + stillMissing: true, + }); + } + return null; + } return line; } catch { // Malformed change lines: drop (fail closed). @@ -185,6 +298,7 @@ function filterLineFeed( /** Buffer a normal/longpoll JSON `_changes` body, filter `results`, re-encode. */ function filterJsonChanges( upstream: ReadableStream, + cache: AclCache, state: DbAclState, principal: Principal, maxBytes: number, @@ -208,10 +322,40 @@ function filterJsonChanges( return; } const upstreamResults = body.results ?? []; + const ids = upstreamResults + .map((row) => row.id) + .filter((id): id is string => typeof id === "string" && id.length > 0); + const missing = ids.filter((id) => needsAclWarm(state, principal, id)); + if (missing.length > 0) { + if (isLevelEnabled("verbose")) { + log.verbose("changes-cache-miss-warm", { + db: state.name, + user: principal.name, + count: missing.length, + feed: "normal", + }); + } + await ensureDocRows(cache, state, missing); + } const results = upstreamResults.filter((row) => { // Fail closed: only forward changes with a readable document id. if (!row.id || typeof row.id !== "string") return false; - return canRead(state, principal, row.id); + const allowed = canRead(state, principal, row.id); + if ( + !allowed && + isLevelEnabled("verbose") && + !state.acl.has(row.id) && + !principal.admin && + !state.noacl + ) { + log.verbose("missing-row-deny-after-ensure", { + db: state.name, + user: principal.name, + docId: row.id, + stillMissing: true, + }); + } + return allowed; }); if (isLevelEnabled("verbose")) { log.verbose("filterJsonChanges", { @@ -220,6 +364,7 @@ function filterJsonChanges( upstream: upstreamResults.length, kept: results.length, dropped: upstreamResults.length - results.length, + warmed: missing.length, }); } const out = JSON.stringify({ ...body, results }); diff --git a/src/routes/actors.ts b/src/routes/actors.ts index 7dbc2b6..a708b5f 100644 --- a/src/routes/actors.ts +++ b/src/routes/actors.ts @@ -935,6 +935,15 @@ export const actors: Record = { requestBodyJson?.keys != null || requestBodyJson?.key != null; const preserveDenied = hasKeys && !isView; + // Warm cold ACL rows before sync canRead — same race as `_changes`. + const warmIds: string[] = []; + for (const row of body.rows ?? []) { + const rowId = row.id ?? row.doc?._id; + if (typeof rowId === "string" && rowId) warmIds.push(rowId); + const embeddedId = row.doc?._id; + if (typeof embeddedId === "string" && embeddedId) warmIds.push(embeddedId); + } + await ensureDocRows(c.get("aclCache"), state, warmIds); const filtered = profileSync("filter", () => filterRows(state, principal, body, { preserveDenied, @@ -988,6 +997,7 @@ export const actors: Record = { const filtered = filterChangesStream( upstream.body, + c.get("aclCache"), state, principal, feed === "live" ? "continuous" : feed, @@ -1325,6 +1335,14 @@ export const actors: Record = { throw err; } const principal = c.get("principal"); + // Warm cold ACL rows before sync canRead — same race as `_changes`. + await ensureDocRows( + c.get("aclCache"), + state, + (body.docs ?? []) + .map((doc) => doc._id) + .filter((id): id is string => typeof id === "string" && id.length > 0), + ); const filtered = profileSync("filter", () => filterFindDocs(state, principal, body)); logDecision("find", { decision: "filter", diff --git a/test/integration/security-edges.test.ts b/test/integration/security-edges.test.ts index efa28bb..aa570a1 100644 --- a/test/integration/security-edges.test.ts +++ b/test/integration/security-edges.test.ts @@ -589,6 +589,159 @@ describe("security edge cases", () => { const body = (await res.json()) as { error?: string }; expect(body.error).toBe("bad_request"); }); + + it("live continuous _changes warms cold ACL rows (dbacl reader sees guest create)", async () => { + // Reproduce FAIMS race: contributor feed open before guest write; cold + // in-memory miss must ensure+authorize, not permanently drop the seq. + let prevDbacl: unknown; + let putOk = false; + for (let attempt = 0; attempt < 5; attempt++) { + const get = await fetch(`${PROXY}/${DB}/_design/acl`, { headers: adminHeaders() }); + expect(get.status).toBe(200); + const ddoc = (await get.json()) as Record & { _rev: string }; + prevDbacl = ddoc.dbacl; + ddoc.dbacl = { _r: ["r-writers"], _w: [], _d: [] }; + const put = await fetch(`${PROXY}/${DB}/_design/acl`, { + method: "PUT", + headers: { ...adminHeaders(), "Content-Type": "application/json" }, + body: JSON.stringify(ddoc), + }); + if (put.ok) { + putOk = true; + break; + } + if (put.status !== 409) { + throw new Error(`dbacl put: ${put.status} ${await put.text()}`); + } + await sleep(100); + } + expect(putOk).toBe(true); + + const seen = new Set(); + let feedError: unknown; + const controller = new AbortController(); + let feedDone: Promise | undefined; + try { + // Wait until dbacl is live for bob on an existing private doc. + await waitForReadable(DB, ids.alicePrivate, authHeaders("jwt", bobJwt)); + + const sinceRes = await fetch(`${PROXY}/${DB}/_changes?since=now&feed=normal&timeout=1`, { + headers: authHeaders("jwt", bobJwt), + }); + expect(sinceRes.status).toBe(200); + const sinceBody = (await sinceRes.json()) as { last_seq: string | number }; + const since = sinceBody.last_seq; + + feedDone = (async () => { + const res = await fetch( + `${PROXY}/${DB}/_changes?feed=continuous&heartbeat=1000&include_docs=true&since=${encodeURIComponent(String(since))}`, + { headers: authHeaders("jwt", bobJwt), signal: controller.signal }, + ); + if (!res.ok || !res.body) throw new Error(`live feed ${res.status}`); + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buf = ""; + try { + while (!controller.signal.aborted) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let nl: number; + while ((nl = buf.indexOf("\n")) >= 0) { + const line = buf.slice(0, nl).trim(); + buf = buf.slice(nl + 1); + if (!line) continue; + try { + const obj = JSON.parse(line) as { id?: string }; + if (typeof obj.id === "string") seen.add(obj.id); + } catch { + // heartbeat / partial + } + } + } + } catch (err) { + if ((err as Error).name !== "AbortError") feedError = err; + } finally { + await reader.cancel().catch(() => {}); + } + })(); + + // Give the continuous feed time to connect before guest writes. + await sleep(300); + + const recId = `rec-live-${suiteId}`; + const childId = `frev-live-${suiteId}`; + const putRec = await putDoc( + DB, + recId, + { creator: "alice", kind: "rec", body: "guest-create" }, + authHeaders("jwt", aliceJwt), + ); + expect(putRec.ok, `rec put: ${putRec.status} ${await putRec.text()}`).toBe(true); + const putChild = await putDoc( + DB, + childId, + { creator: "alice", parent: recId, kind: "frev", body: "child" }, + authHeaders("jwt", aliceJwt), + ); + expect(putChild.ok, `child put: ${putChild.status} ${await putChild.text()}`).toBe(true); + + // Stress: rapid creates while feed is open — all must appear. + const rapidIds: string[] = []; + for (let i = 0; i < 8; i++) { + const id = `rec-rapid-${suiteId}-${i}`; + rapidIds.push(id); + const put = await putDoc( + DB, + id, + { creator: "alice", kind: "rec", body: `rapid-${i}` }, + authHeaders("jwt", aliceJwt), + ); + expect(put.ok, `rapid put ${id}: ${put.status}`).toBe(true); + } + + await waitUntil( + "live feed saw guest rec + rapid creates", + async () => seen.has(recId) && seen.has(childId) && rapidIds.every((id) => seen.has(id)), + 25_000, + 100, + ); + + // Guest isolation: carol (readers, no dbacl grant) must not see alice's private rec. + const carolChanges = await fetch( + `${PROXY}/${DB}/_changes?include_docs=true&since=0&limit=500`, + { headers: authHeaders("jwt", carolJwt) }, + ); + expect(carolChanges.status).toBe(200); + const carolBody = (await carolChanges.json()) as { results: Array<{ id: string }> }; + const carolIds = new Set(carolBody.results.map((r) => r.id)); + expect(carolIds.has(recId)).toBe(false); + expect(carolIds.has(ids.alicePrivate)).toBe(false); + } finally { + controller.abort(); + await feedDone?.catch(() => {}); + for (let attempt = 0; attempt < 5; attempt++) { + const again = await fetch(`${PROXY}/${DB}/_design/acl`, { headers: adminHeaders() }); + const cur = (await again.json()) as Record; + if (prevDbacl === undefined) delete cur.dbacl; + else cur.dbacl = prevDbacl; + const restore = await fetch(`${PROXY}/${DB}/_design/acl`, { + method: "PUT", + headers: { ...adminHeaders(), "Content-Type": "application/json" }, + body: JSON.stringify(cur), + }); + if (restore.ok || restore.status !== 409) break; + await sleep(100); + } + await waitUntil( + "dbacl cleared for bob on alicePrivate", + async () => + (await getDoc(DB, ids.alicePrivate, authHeaders("jwt", bobJwt))).status === 404, + 20_000, + ); + } + if (feedError) throw feedError; + }); }); // ── Bulk edges ───────────────────────────────────────────────────────── diff --git a/test/unit/acl-lookup.test.ts b/test/unit/acl-lookup.test.ts index 9af210c..8733c2a 100644 --- a/test/unit/acl-lookup.test.ts +++ b/test/unit/acl-lookup.test.ts @@ -1,10 +1,11 @@ /** * Unit tests for flagsForDoc missing-row / admin short-circuit semantics. */ -import { describe, expect, it } from "vitest"; -import { flagsForDoc } from "../../src/acl/lookup.js"; +import { describe, expect, it, vi } from "vitest"; +import { canReadEnsured, flagsForDoc } from "../../src/acl/lookup.js"; +import { aclRowFromDoc } from "../../src/acl/resolve.js"; import { buildPrincipal } from "../../src/auth/principal.js"; -import type { DbAclState } from "../../src/acl/cache.js"; +import type { AclCache, DbAclState } from "../../src/acl/cache.js"; function principal(name: string, roles: string[] = []) { return buildPrincipal({ @@ -76,3 +77,24 @@ describe("flagsForDoc", () => { }); }); }); + +describe("canReadEnsured", () => { + it("warms a missing row then authorizes with resolveDocAcl + dbacl", async () => { + const state = emptyState(); + state.dbacl = { _r: ["r-writers"], _w: [], _d: [] }; + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + state.acl.set(id, aclRowFromDoc({ _id: id, creator: "guest" })); + } + }); + const cache = { ensureDocs } as unknown as AclCache; + + // Sync path denies cold miss even with dbacl (create-path). + expect(flagsForDoc(state, principal("bob", ["writers"]), "rec-1")._r).toBe(false); + + await expect( + canReadEnsured(cache, state, principal("bob", ["writers"]), "rec-1"), + ).resolves.toBe(true); + expect(ensureDocs).toHaveBeenCalledWith("test", ["rec-1"]); + }); +}); diff --git a/test/unit/filter-changes.test.ts b/test/unit/filter-changes.test.ts index f3e7a72..9311db4 100644 --- a/test/unit/filter-changes.test.ts +++ b/test/unit/filter-changes.test.ts @@ -1,20 +1,20 @@ -import { describe, expect, it } from "vitest"; -import type { DbAclState } from "../../src/acl/cache.js"; +import { describe, expect, it, vi } from "vitest"; +import { AclUnavailableError, type AclCache, type DbAclState } from "../../src/acl/cache.js"; import { aclRowFromDoc } from "../../src/acl/resolve.js"; import { buildPrincipal } from "../../src/auth/principal.js"; import { filterChangesStream } from "../../src/proxy/filterChanges.js"; const encoder = new TextEncoder(); -function principal(name: string) { +function principal(name: string, roles: string[] = []) { return buildPrincipal({ ok: true, - userCtx: { name, roles: [] }, + userCtx: { name, roles }, info: { authenticated: "jwt" }, }); } -function state(): DbAclState { +function state(overrides?: Partial): DbAclState { return { name: "docs", acl: new Map([ @@ -24,9 +24,18 @@ function state(): DbAclState { noacl: false, ready: true, followerUp: true, + ...overrides, }; } +/** Minimal AclCache stub — `ensureDocRows` only needs `ensureDocs`. */ +function mockCache( + dbState: DbAclState, + ensureDocs: AclCache["ensureDocs"] = async () => undefined, +): AclCache { + return { ensureDocs } as unknown as AclCache; +} + function stream(...chunks: string[]): ReadableStream { return new ReadableStream({ start(controller) { @@ -42,6 +51,7 @@ async function text(body: ReadableStream): Promise { describe("filterChangesStream", () => { it("filters normal feeds while preserving opaque sequence metadata", async () => { + const dbState = state(); const upstream = stream( JSON.stringify({ results: [ @@ -55,7 +65,9 @@ describe("filterChangesStream", () => { ); const output = JSON.parse( - await text(filterChangesStream(upstream, state(), principal("bob"), "normal")), + await text( + filterChangesStream(upstream, mockCache(dbState), dbState, principal("bob"), "normal"), + ), ) as { results: Array<{ id: string; seq: string }>; last_seq: string; @@ -73,6 +85,7 @@ describe("filterChangesStream", () => { }); it("filters split continuous-feed lines and preserves heartbeats/control rows", async () => { + const dbState = state(); const upstream = stream( '{"id":"pri', 'vate","seq":"1-a"}\n\n{"id":"shared","seq":"2-b"}\n', @@ -80,7 +93,7 @@ describe("filterChangesStream", () => { ); const output = await text( - filterChangesStream(upstream, state(), principal("bob"), "continuous"), + filterChangesStream(upstream, mockCache(dbState), dbState, principal("bob"), "continuous"), ); expect(output).not.toContain("private"); expect(output).not.toContain("not-json"); @@ -89,6 +102,7 @@ describe("filterChangesStream", () => { }); it("keeps SSE metadata only for allowed data events", async () => { + const dbState = state(); const upstream = stream( 'data: {"id":"private","seq":"1-a"}\nid: 1-a\n\n', 'event: message\ndata: {"id":"shared","seq":"2-b"}\nid: 2-b\n\n', @@ -96,7 +110,7 @@ describe("filterChangesStream", () => { ); const output = await text( - filterChangesStream(upstream, state(), principal("bob"), "eventsource"), + filterChangesStream(upstream, mockCache(dbState), dbState, principal("bob"), "eventsource"), ); expect(output).not.toContain("private"); expect(output).not.toContain("id: 1-a"); @@ -106,6 +120,7 @@ describe("filterChangesStream", () => { }); it("does not let last_seq turn a denied change into control metadata", async () => { + const dbState = state(); const continuous = await text( filterChangesStream( stream( @@ -113,7 +128,8 @@ describe("filterChangesStream", () => { '{"id":"shared","seq":"2-b","last_seq":"2-b"}\n', '{"last_seq":"2-b","pending":0}\n', ), - state(), + mockCache(dbState), + dbState, principal("bob"), "continuous", ), @@ -128,7 +144,8 @@ describe("filterChangesStream", () => { 'data: {"id":"private","seq":"1-a","last_seq":"1-a"}\nid: 1-a\n\n', 'data: {"last_seq":"2-b"}\n\n', ), - state(), + mockCache(dbState), + dbState, principal("bob"), "eventsource", ), @@ -139,13 +156,234 @@ describe("filterChangesStream", () => { }); it("rejects oversized buffered normal feeds", async () => { + const dbState = state(); const filtered = filterChangesStream( stream(JSON.stringify({ results: [], padding: "x".repeat(200) })), - state(), + mockCache(dbState), + dbState, principal("bob"), "longpoll", { maxBufferBytes: 64 }, ); await expect(text(filtered)).rejects.toThrow(/64 bytes/); }); + + describe("cold ACL cache miss warm", () => { + it("normal feed: warms missing row then forwards when readable", async () => { + const dbState = state({ + acl: new Map(), + dbacl: { _r: ["r-writers"], _w: [], _d: [] }, + }); + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + dbState.acl.set(id, aclRowFromDoc({ _id: id, creator: "guest" })); + } + }); + const upstream = stream( + JSON.stringify({ + results: [{ id: "rec-1", seq: "9-a", doc: { _id: "rec-1", creator: "guest" } }], + last_seq: "9-a", + }), + ); + + const output = JSON.parse( + await text( + filterChangesStream( + upstream, + mockCache(dbState, ensureDocs), + dbState, + principal("bob", ["writers"]), + "normal", + ), + ), + ) as { results: Array<{ id: string }> }; + + expect(ensureDocs).toHaveBeenCalledWith("docs", ["rec-1"]); + expect(output.results.map((r) => r.id)).toEqual(["rec-1"]); + }); + + it("continuous feed: warms missing row then forwards when readable", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + dbState.acl.set(id, aclRowFromDoc({ _id: id, creator: "alice", acl: ["u-bob"] })); + } + }); + + const output = await text( + filterChangesStream( + stream('{"id":"rec-1","seq":"1-a"}\n{"last_seq":"1-a"}\n'), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "continuous", + ), + ); + + expect(ensureDocs).toHaveBeenCalledWith("docs", ["rec-1"]); + expect(output).toContain('{"id":"rec-1","seq":"1-a"}'); + expect(output).toContain('{"last_seq":"1-a"}'); + }); + + it("eventsource feed: warms missing row then forwards when readable", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + dbState.acl.set(id, aclRowFromDoc({ _id: id, creator: "alice", acl: ["u-bob"] })); + } + }); + + const output = await text( + filterChangesStream( + stream('data: {"id":"rec-1","seq":"1-a"}\nid: 1-a\n\n'), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "eventsource", + ), + ); + + expect(ensureDocs).toHaveBeenCalledWith("docs", ["rec-1"]); + expect(output).toContain('data: {"id":"rec-1","seq":"1-a"}'); + expect(output).toContain("id: 1-a"); + }); + + it("still drops after ensure when principal cannot read", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + // Creator-only; carol has no grant and no dbacl. + dbState.acl.set(id, aclRowFromDoc({ _id: id, creator: "alice" })); + } + }); + + const normal = JSON.parse( + await text( + filterChangesStream( + stream(JSON.stringify({ results: [{ id: "rec-1", seq: "1-a" }], last_seq: "1-a" })), + mockCache(dbState, ensureDocs), + dbState, + principal("carol"), + "normal", + ), + ), + ) as { results: unknown[] }; + expect(ensureDocs).toHaveBeenCalledWith("docs", ["rec-1"]); + expect(normal.results).toEqual([]); + + const continuousState = state({ acl: new Map() }); + const ensureContinuous = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + continuousState.acl.set(id, aclRowFromDoc({ _id: id, creator: "alice" })); + } + }); + const continuous = await text( + filterChangesStream( + stream('{"id":"rec-1","seq":"1-a"}\n'), + mockCache(continuousState, ensureContinuous), + continuousState, + principal("carol"), + "continuous", + ), + ); + expect(ensureContinuous).toHaveBeenCalledWith("docs", ["rec-1"]); + expect(continuous).not.toContain("rec-1"); + }); + + it("keeps create-path deny when ensure finds no row (doc absent)", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async () => { + // Doc does not exist — leave cache empty (create-path semantics). + }); + + const output = JSON.parse( + await text( + filterChangesStream( + stream(JSON.stringify({ results: [{ id: "ghost", seq: "1-a" }], last_seq: "1-a" })), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "normal", + ), + ), + ) as { results: unknown[] }; + + expect(ensureDocs).toHaveBeenCalledWith("docs", ["ghost"]); + expect(output.results).toEqual([]); + }); + + it("fail-closed: AclUnavailableError does not forward the change", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async () => { + throw new AclUnavailableError("view failed"); + }); + + await expect( + text( + filterChangesStream( + stream(JSON.stringify({ results: [{ id: "rec-1", seq: "1-a" }], last_seq: "1-a" })), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "normal", + ), + ), + ).rejects.toBeInstanceOf(AclUnavailableError); + + await expect( + text( + filterChangesStream( + stream('{"id":"rec-1","seq":"1-a"}\n'), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "continuous", + ), + ), + ).rejects.toBeInstanceOf(AclUnavailableError); + + expect(ensureDocs).toHaveBeenCalled(); + }); + + it("batches multiple missing ids on a continuous chunk", async () => { + const dbState = state({ acl: new Map() }); + const ensureDocs = vi.fn(async (_db: string, ids: readonly string[]) => { + for (const id of ids) { + dbState.acl.set(id, aclRowFromDoc({ _id: id, creator: "alice", acl: ["u-bob"] })); + } + }); + + const output = await text( + filterChangesStream( + stream('{"id":"rec-1","seq":"1-a"}\n{"id":"rec-2","seq":"2-b"}\n'), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "continuous", + ), + ); + + expect(ensureDocs).toHaveBeenCalledTimes(1); + expect(ensureDocs).toHaveBeenCalledWith("docs", ["rec-1", "rec-2"]); + expect(output).toContain("rec-1"); + expect(output).toContain("rec-2"); + }); + + it("does not call ensureDocs when rows are already cached", async () => { + const dbState = state(); + const ensureDocs = vi.fn(async () => undefined); + + await text( + filterChangesStream( + stream('{"id":"shared","seq":"2-b"}\n'), + mockCache(dbState, ensureDocs), + dbState, + principal("bob"), + "continuous", + ), + ); + + expect(ensureDocs).not.toHaveBeenCalled(); + }); + }); }); From 3ab240f86da57955de76d5e93ac25228cc43014b Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 26 Jul 2026 00:47:30 +0000 Subject: [PATCH 2/2] fix: skip _local ids in ensureDocRows warm path Keyed _all_docs can report _local docs as live while the ACL view omits them, which marked the DB unavailable (503) on _local_docs listings. Co-authored-by: Peter Baker --- src/acl/lookup.ts | 7 ++++++- src/proxy/filterChanges.ts | 2 ++ test/unit/acl-lookup.test.ts | 13 ++++++++++++- 3 files changed, 20 insertions(+), 2 deletions(-) diff --git a/src/acl/lookup.ts b/src/acl/lookup.ts index d8957ac..db1f583 100644 --- a/src/acl/lookup.ts +++ b/src/acl/lookup.ts @@ -143,7 +143,12 @@ export async function ensureDocRows( ids: Iterable, ): Promise { if (state.noacl) return; - const unique = [...new Set(ids)].filter((id) => typeof id === "string" && id.length > 0); + // `_local/*` never appears in the ACL view; keyed `_all_docs` can still report + // them as live winners, and reconcile would fail closed. Listings keep the + // missing-row deny; individual `_local` CRUD uses the DB-gated pipe actor. + const unique = [...new Set(ids)].filter( + (id) => typeof id === "string" && id.length > 0 && !id.startsWith("_local/"), + ); if (unique.length === 0) return; const missing = unique.filter((id) => !state.acl.has(id)); diff --git a/src/proxy/filterChanges.ts b/src/proxy/filterChanges.ts index 615463c..864650a 100644 --- a/src/proxy/filterChanges.ts +++ b/src/proxy/filterChanges.ts @@ -81,6 +81,8 @@ function normalizeFeed(feed: string): string { function needsAclWarm(state: DbAclState, principal: Principal, docId: string): boolean { if (principal.admin || state.noacl) return false; if (!docId || typeof docId !== "string") return false; + // Local docs are not ACL-mapped; keep missing-row deny without ensure. + if (docId.startsWith("_local/")) return false; return !state.acl.has(docId); } diff --git a/test/unit/acl-lookup.test.ts b/test/unit/acl-lookup.test.ts index 8733c2a..3368ef1 100644 --- a/test/unit/acl-lookup.test.ts +++ b/test/unit/acl-lookup.test.ts @@ -2,7 +2,7 @@ * Unit tests for flagsForDoc missing-row / admin short-circuit semantics. */ import { describe, expect, it, vi } from "vitest"; -import { canReadEnsured, flagsForDoc } from "../../src/acl/lookup.js"; +import { canReadEnsured, ensureDocRows, flagsForDoc } from "../../src/acl/lookup.js"; import { aclRowFromDoc } from "../../src/acl/resolve.js"; import { buildPrincipal } from "../../src/auth/principal.js"; import type { AclCache, DbAclState } from "../../src/acl/cache.js"; @@ -98,3 +98,14 @@ describe("canReadEnsured", () => { expect(ensureDocs).toHaveBeenCalledWith("test", ["rec-1"]); }); }); + +describe("ensureDocRows", () => { + it("skips _local ids so listings cannot fail closed via view reconcile", async () => { + const state = emptyState(); + const ensureDocs = vi.fn(async () => undefined); + const cache = { ensureDocs } as unknown as AclCache; + + await ensureDocRows(cache, state, ["_local/checkpoint", "rec-1"]); + expect(ensureDocs).toHaveBeenCalledWith("test", ["rec-1"]); + }); +});