fix: warm ACL cache on live _changes miss (v1.7.0)

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

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.

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",

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<boolean> {
+  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).
@@ -126,7 +143,12 @@ export async function ensureDocRows(
   ids: Iterable<string>,
 ): Promise<void> {
   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));

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<Uint8Array>,
+  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,56 @@ 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;
+  // Local docs are not ACL-mapped; keep missing-row deny without ensure.
+  if (docId.startsWith("_local/")) 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<Uint8Array>,
+  cache: AclCache,
   state: DbAclState,
   principal: Principal,
   eventsource: boolean,
@@ -82,6 +135,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 +149,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 +216,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<void> {
+        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 +268,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 +300,7 @@ function filterLineFeed(
 /** Buffer a normal/longpoll JSON `_changes` body, filter `results`, re-encode. */
 function filterJsonChanges(
   upstream: ReadableStream<Uint8Array>,
+  cache: AclCache,
   state: DbAclState,
   principal: Principal,
   maxBytes: number,
@@ -208,10 +324,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 +366,7 @@ function filterJsonChanges(
             upstream: upstreamResults.length,
             kept: results.length,
             dropped: upstreamResults.length - results.length,
+            warmed: missing.length,
           });
         }
         const out = JSON.stringify({ ...body, results });