summaryrefslogtreecommitdiffhomepage
path: root/src/features/conversation-cache
diff options
context:
space:
mode:
authorAdam Malczewski <[email protected]>2026-10-21 14:46:06 +0900
committerAdam Malczewski <[email protected]>2026-10-21 14:46:06 +0900
commit0bdb04b0aa0c0500e68e2e985b1a2880730ad7e8 (patch)
tree0768ddd1be8828738ee80770dd723236b91e32e8 /src/features/conversation-cache
parent6a2a40d64737b3cb678740c8f85027c3be0000bc (diff)
downloaddispatch-web-0bdb04b0aa0c0500e68e2e985b1a2880730ad7e8.tar.gz
dispatch-web-0bdb04b0aa0c0500e68e2e985b1a2880730ad7e8.zip
fix(review-followup-fec): reconcile epoch rollback and retain reconnect backoff
Diffstat (limited to 'src/features/conversation-cache')
-rw-r--r--src/features/conversation-cache/cache.test.ts35
-rw-r--r--src/features/conversation-cache/cache.ts24
2 files changed, 58 insertions, 1 deletions
diff --git a/src/features/conversation-cache/cache.test.ts b/src/features/conversation-cache/cache.test.ts
index 4306d51..72b4e20 100644
--- a/src/features/conversation-cache/cache.test.ts
+++ b/src/features/conversation-cache/cache.test.ts
@@ -214,3 +214,38 @@ it("serializes invalidation and rejects stale epoch commits", async () => {
expect(await cache.commit("c", [chunk(103)], 1)).toEqual([]);
expect(await store.historyEpoch("c")).toBe(2);
});
+
+it("replaces a confirmed full history at a lower epoch, without trusting stale commits", async () => {
+ const store = createFakeStore();
+ const cache = createConversationCache(store);
+ await cache.commit("c", [chunk(9)], 5);
+ expect(await cache.replace("c", [chunk(1)], 1, 5)).toBe(true);
+ expect(await cache.historyEpoch("c")).toBe(1);
+ expect(await cache.load("c")).toEqual([chunk(1)]);
+ expect(await cache.commit("c", [chunk(9)], 5)).toEqual([chunk(1)]);
+ // A delayed full response started against epoch 5 must not restore it.
+ expect(await cache.replace("c", [chunk(9)], 5, 5)).toBe(false);
+ expect(await cache.load("c")).toEqual([chunk(1)]);
+});
+
+it("serializes a guarded replacement behind newer invalidation", async () => {
+ const store = createFakeStore();
+ const cache = createConversationCache(store);
+ await cache.commit("c", [chunk(9)], 5);
+ const resetting = cache.reset("c", 6);
+ const replacing = cache.replace("c", [chunk(1)], 1, 5);
+ await resetting;
+ expect(await replacing).toBe(false);
+ expect(await cache.historyEpoch("c")).toBe(6);
+ expect(await cache.load("c")).toEqual([]);
+});
+
+it("allows an empty replacement and a legacy cache only if its observed epoch still matches", async () => {
+ const store = createFakeStore();
+ const cache = createConversationCache(store);
+ await store.append("c", [chunk(9)]);
+ expect(await cache.replace("c", [], 0, undefined)).toBe(true);
+ expect(await cache.sinceSeq("c")).toBe(0);
+ expect(await cache.historyEpoch("c")).toBe(0);
+ expect(await cache.replace("c", [chunk(9)], 2, undefined)).toBe(false);
+});
diff --git a/src/features/conversation-cache/cache.ts b/src/features/conversation-cache/cache.ts
index 431a2a7..600a6e6 100644
--- a/src/features/conversation-cache/cache.ts
+++ b/src/features/conversation-cache/cache.ts
@@ -6,6 +6,19 @@ export interface ConversationCache {
historyEpoch(conversationId: string): Promise<number | undefined>;
reset(conversationId: string, historyEpoch: number): Promise<void>;
+ /**
+ * Replace from a fresh full-history response, including an epoch decrease.
+ * Only succeeds if the cache still has the epoch observed BEFORE that request.
+ * Unlike monotonic reset/commit, this is for confirmed server authority, not
+ * broadcasts or cursor/backfill responses. False means the response is stale.
+ */
+ replace(
+ conversationId: string,
+ incoming: readonly StoredChunk[],
+ historyEpoch: number,
+ expectedEpoch: number | undefined,
+ ): Promise<boolean>;
+
/** Load all cached chunks for a conversation. */
load(conversationId: string): Promise<readonly StoredChunk[]>;
@@ -73,6 +86,13 @@ export function createConversationCache(
const current = await store.historyEpoch(id);
if (current === undefined || current < epoch) await store.reset(id, epoch);
}),
+ replace: (id, incoming, epoch, expectedEpoch) =>
+ exclusive(id, async () => {
+ if ((await store.historyEpoch(id)) !== expectedEpoch) return false;
+ await store.reset(id, epoch);
+ if (incoming.length > 0) await store.append(id, incoming);
+ return true;
+ }),
async load(conversationId) {
await pending.get(conversationId);
return store.load(conversationId);
@@ -82,7 +102,9 @@ export function createConversationCache(
return exclusive(conversationId, async () => {
if (epoch !== undefined) {
const current = await store.historyEpoch(conversationId);
- if (current !== undefined && current > epoch) return store.load(conversationId);
+ // Once an epoch is known, only explicit invalidation or guarded
+ // full-history replacement may change it (in either direction).
+ if (current !== undefined && current !== epoch) return store.load(conversationId);
if (current !== epoch) await store.reset(conversationId, epoch);
}
const cached = await store.load(conversationId);