diff --git a/docs/decisions/implemented/2026-10-28-add-log-accept-default.md b/docs/decisions/implemented/2026-10-28-add-log-accept-default.md index d75fc17a..3f6264e8 100644 --- a/docs/decisions/implemented/2026-10-28-add-log-accept-default.md +++ b/docs/decisions/implemented/2026-10-28-add-log-accept-default.md @@ -93,3 +93,17 @@ them is far larger than the run-to-run spread within a mode. unreadable to an older deployment. - Folding happens on the `Add` path when pending reaches `FOLD_TRIGGER`; it never happens on the read path, so search latency does not inherit write contention. + +## Follow-up: the read window takes the newest entries + +The first version of this path read the _oldest_ pending entries into the search window, which +is the right order for folding and the wrong order for reading. Measured on the deployed +service with twenty sequential Adds to one user (below the fold trigger, so nothing folded): +markers 0, 3 and 7 were found and markers 8, 12 and 19 were missing - the window held the +first eight entries and hid everything after them. Nothing recovers those entries, because a +search never folds and the fold is triggered by the counter. + +`listPending` gained a `newest` option that takes the tail of the arrival order by name before +reading any body, and the search path passes it. The fold still reads the head, so entries are +applied in the order they arrived. A test pins both directions, because either one alone would +look correct. diff --git a/docs/decisions/implemented/2026-10-28-add-log-accept-default.zh-CN.md b/docs/decisions/implemented/2026-10-28-add-log-accept-default.zh-CN.md index ccb31bfa..94d16de3 100644 --- a/docs/decisions/implemented/2026-10-28-add-log-accept-default.zh-CN.md +++ b/docs/decisions/implemented/2026-10-28-add-log-accept-default.zh-CN.md @@ -73,3 +73,13 @@ Add,因此每次运行只有一轮。跨过折叠阈值的轮被排除;`roun 而 v4 的 manifest 对旧部署不可读。 - 折叠只发生在 `Add` 路径(pending 达到 `FOLD_TRIGGER` 时),读路径永不折叠,因此检索延迟不 继承写竞争。 + +## 后续修正:检索窗口取最新的条目 + +这条路径的第一版把**最老**的 pending 条目放进检索窗口——那是折叠该有的顺序,不是读取该有的 +顺序。在已部署服务上用同一用户连续 20 次 Add 实测(低于折叠阈值,因此没有发生折叠): +marker 0、3、7 命中,marker 8、12、19 缺失——窗口装的是最老的 8 条,之后的全被挡住。而且没有 +任何机制会把它们捞回来:检索从不折叠,折叠由计数器触发。 + +`listPending` 增加 `newest` 选项:先按名字取到达顺序的尾部,再读 body;检索路径传它。折叠仍读 +头部,因此条目依旧按到达顺序被应用。一条测试同时钉住两个方向,因为只钉一边会看起来"都对"。 diff --git a/onlion/backend/append-log.mjs b/onlion/backend/append-log.mjs index 0fd7a1f1..34b9135d 100644 --- a/onlion/backend/append-log.mjs +++ b/onlion/backend/append-log.mjs @@ -161,6 +161,11 @@ export async function findLogEntry(bucket, userKey, requestHash, options = {}) { export async function listPending(bucket, userKey, options = {}) { const limit = readLimit(options?.limit); const generation = options?.generation; + // `newest` reads from the tail of the arrival order. The fold wants the oldest entries + // (they are the ones a reader has waited longest for), while a search wants the newest: + // a streaming update is invisible if the window is filled with the entries that were + // already there. + const newest = options?.newest === true; const skipRequestHash = options?.skipRequestHash; if (skipRequestHash !== undefined && typeof skipRequestHash !== 'function') throw new TypeError('skipRequestHash must be a function'); @@ -173,9 +178,13 @@ export async function listPending(bucket, userKey, options = {}) { ? Math.max(MAX_PENDING_READ, limit * SCAN_FACTOR) : limit; const objects = await listObjects(bucket, userKey, scan); + // A tail window is taken by name before any body is read: names sort in arrival order, + // so the last `scan` keys are the newest entries without paying for the ones skipped. + const candidates = newest && Number.isFinite(scan) ? objects.slice(-scan) : objects; const pending = []; - for (let index = 0; index < objects.length && pending.length < limit; index += READ_CONCURRENCY) { - const chunk = objects.slice(index, index + READ_CONCURRENCY); + const ceiling = newest ? Infinity : limit; + for (let index = 0; index < candidates.length && pending.length < ceiling; index += READ_CONCURRENCY) { + const chunk = candidates.slice(index, index + READ_CONCURRENCY); const read = await Promise.all( chunk.map(async (object) => { const name = object.key.slice(prefix.length); @@ -205,7 +214,9 @@ export async function listPending(bucket, userKey, options = {}) { ); pending.push(...read.filter(Boolean)); } - return pending.slice(0, limit); + // Oldest-first callers take the head, newest-first callers the tail: both return at most + // `limit` entries, in arrival order, so the fold still applies them the way they arrived. + return newest ? pending.slice(-limit) : pending.slice(0, limit); } /** Batch delete. Deletion is always after the partition commit, so repeating it diff --git a/onlion/backend/worker.mjs b/onlion/backend/worker.mjs index bcbddf28..ed8288c0 100644 --- a/onlion/backend/worker.mjs +++ b/onlion/backend/worker.mjs @@ -866,6 +866,9 @@ async function pendingForSearch(env, ctx, limit) { const waiting = await listPending(env.MEMORY, ctx.key, { limit: Math.min(limit, MAX_PENDING_READ), generation: ctx.manifest.generation ?? null, + // Newest first: a streaming update has to be visible before it is folded, and taking + // the oldest entries meant the most recent ones fell outside the window. + newest: true, }); return waiting.flatMap(({ entry }) => entry.messages.map((message, index) => ({ diff --git a/onlion/scripts/test-append-log.mjs b/onlion/scripts/test-append-log.mjs index 5d3ce986..af59dca1 100644 --- a/onlion/scripts/test-append-log.mjs +++ b/onlion/scripts/test-append-log.mjs @@ -528,3 +528,40 @@ console.log( 'PASS: a generation filter fills the window from current entries even when every older key sorts first.', ); } + +// A read window that takes the oldest entries hides the newest ones - the entry a +// streaming update just added is exactly what a search is asking about. +{ + const bucket = memoryBucket(); + const names = []; + for (let index = 0; index < 12; index += 1) { + const entry = { + v: 1, + generation: 'g1', + user_id: 'u', + session_id: 's', + request_id: `entry-${index}`, + request_hash: createHash('sha256').update(`entry-${index}`).digest('hex'), + fingerprint: 'f', + received_at: 1700000000000 + index, + messages: [], + }; + await putLogEntry(bucket, 'window-key', entry); + names.push(logEntryName(entry.received_at, entry.request_hash)); + } + const oldest = await listPending(bucket, 'window-key', { limit: 4, generation: 'g1' }); + assert.deepEqual( + oldest.map((item) => item.name), + names.slice(0, 4), + 'the default window is arrival order, which the fold needs', + ); + const newest = await listPending(bucket, 'window-key', { limit: 4, generation: 'g1', newest: true }); + assert.deepEqual( + newest.map((item) => item.name), + names.slice(-4), + 'a newest window must hold the entries that arrived last', + ); + console.log( + 'PASS: the pending window reads from the head for the fold and from the tail for search.', + ); +}