Keep failed chat history available for retry (#8278)

* fix(app): keep failed chat history until retry

Automatic retries repeatedly reopen provider sessions after persistent failures. Keep the error visible until the user retries, while retaining reconnect recovery.

* fix(app): catch up chats after subscription failure

A failed replacement subscription leaves chats unsubscribed, so discard its old catch-up state and fetch missed history after retry.

* fix(app): recover gaps during history retry

Keep gap recovery active while a user-initiated retry is running so the chat is not marked current before its follow-up fetch.

* fix(app): show history errors in archived chats

Cached archived chats need the same manual retry affordance when history synchronization fails.
This commit is contained in:
Mohamed Boudra
2026-10-30 17:57:52 +07:00
committed by GitHub
parent 9477fe0c00
commit 385c29b655
7 changed files with 155 additions and 190 deletions
@@ -455,10 +455,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
reportError: (error) => {
console.warn("[Session] viewed timeline synchronization failed", { serverId, error });
},
schedule: (task, delayMs) => {
const timeout = setTimeout(task, delayMs);
return () => clearTimeout(timeout);
},
});
viewedTimelineSyncRef.current = sync;
setViewedTimelineSync(serverId, sync);
@@ -422,6 +422,23 @@ describe("deriveAgentScreenViewState", () => {
expect(result.memory.lastReadyAgent).toBeNull();
});
it.each(["error", "retrying"] as const)("shows archived history sync %s", (status) => {
const result = deriveAgentScreenViewState({
input: {
...createBaseInput(),
agent: createAgent("agent-1"),
isArchived: true,
hasHydratedHistoryBefore: true,
visibilityCatchUpStatus: status,
},
memory: createBaseMemory(),
});
expect(expectReadyState(result.state).sync).toEqual({
status: "sync_error",
isRetrying: status === "retrying",
});
});
it("renders an archived agent before provider history is initialized", () => {
const result = deriveAgentScreenViewState({
input: {
@@ -175,10 +175,14 @@ function resolveAgentScreenSync(args: {
hadInitialSyncFailure: boolean;
}): AgentScreenReadySyncState {
const { input, hadInitialSyncFailure } = args;
if (input.isArchived) {
if (
input.isArchived &&
input.visibilityCatchUpStatus !== "error" &&
input.visibilityCatchUpStatus !== "retrying"
) {
return { status: "idle" };
}
if (!input.isConnected) {
if (!input.isConnected && !input.isArchived) {
return { status: "reconnecting" };
}
if (input.missingAgentState.kind === "error") {
@@ -174,7 +174,6 @@ it("feeds restored layout to timeline sync as a release signal, not a subscripti
fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }),
onCatchUpEnded: () => undefined,
reportError: () => undefined,
schedule: () => () => undefined,
});
// The edge session-context.tsx owns. Pointing it at subscription instead would resume every
// agent behind a restored tab on the daemon.
@@ -57,7 +57,6 @@ function createOwner(storage: TimelineReplicaStorage): ViewedTimelineOwner {
fetchPage: async () => ({ hasNewer: false, endCursor: null }),
fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }),
reportError: () => undefined,
schedule: () => () => undefined,
},
});
}
@@ -617,7 +616,6 @@ describe("create handoff lifetime", () => {
},
fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }),
reportError: () => undefined,
schedule: () => () => undefined,
},
});
@@ -90,7 +90,7 @@ test("visible catch-up settles before hidden open chats start fetching", async (
world.sync.dispose();
});
test("a failed visible catch-up releases hidden chats without bypassing its retry delay", async () => {
test("a failed visible catch-up releases hidden chats without bypassing its error", async () => {
const world = new TimelineWorld();
world.openedEarlier(["a-hidden"]);
world.sync.replaceVisibleAgentIds("pane", ["z-visible"]);
@@ -103,7 +103,7 @@ test("a failed visible catch-up releases hidden chats without bypassing its retr
hidden.respond({ hasNewer: false });
await world.nextError();
world.expectNoPendingFetch();
(await world.nextRetry())();
world.sync.retryVisibleAgentTimeline("z-visible");
(await world.nextFetch("z-visible")).respond({ hasNewer: false });
world.sync.dispose();
});
@@ -200,20 +200,6 @@ class TimelineWorld {
const waiter = this.errorWaiters.shift();
if (waiter) waiter(this.errors.at(-1) ?? "");
},
schedule: (task, delayMs) => {
const scheduled = { task, delayMs };
this.scheduled.push(scheduled);
const waiterIndex = this.retryWaiters.findIndex((waiter) => waiter.delayMs === delayMs);
if (waiterIndex >= 0) {
const [waiter] = this.retryWaiters.splice(waiterIndex, 1);
this.scheduled.splice(this.scheduled.indexOf(scheduled), 1);
waiter.resolve(task);
}
return () => {
const index = this.scheduled.indexOf(scheduled);
if (index >= 0) this.scheduled.splice(index, 1);
};
},
});
show(sourceId: string, agentIds: string[]): void {
@@ -235,12 +221,6 @@ class TimelineWorld {
}> = [];
private readonly errorWaiters: Array<(message: string) => void> = [];
private readonly cacheRequestWaiters: Array<(agentId: string) => void> = [];
private readonly scheduled: Array<{ task: () => void; delayMs: number }> = [];
private readonly retryWaiters: Array<{
delayMs: number;
resolve(retry: () => void): void;
}> = [];
get pendingFetchCount(): number {
return this.fetches.length;
}
@@ -300,22 +280,6 @@ class TimelineWorld {
return new Promise((resolve) => this.errorWaiters.push(resolve));
}
nextRetry(delayMs = 1_000): Promise<() => void> {
const index = this.scheduled.findIndex((entry) => entry.delayMs === delayMs);
if (index >= 0) return Promise.resolve(this.scheduled.splice(index, 1)[0].task);
return new Promise((resolve) => this.retryWaiters.push({ delayMs, resolve }));
}
elapse(elapsedMs: number): void {
const due = this.scheduled.filter((entry) => entry.delayMs <= elapsedMs);
this.scheduled.splice(
0,
this.scheduled.length,
...this.scheduled.filter((entry) => entry.delayMs > elapsedMs),
);
for (const entry of due) entry.task();
}
private releaseMembershipWaiter(): void {
const waiter = this.membershipWaiters.shift();
if (!waiter) return;
@@ -483,6 +447,28 @@ test("reports a chat current only once its parked follow-up page completes", asy
world.sync.dispose();
});
test("a gap during manual retry is recovered before reporting ready", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).succeed();
(await world.nextFetch("agent-a")).fail("unavailable");
await world.nextError();
world.sync.retryVisibleAgentTimeline("agent-a");
const retry = await world.nextFetch("agent-a");
world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 9 });
retry.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.pendingFetchCount).toBe(1));
const recovery = await world.nextFetch("agent-a");
expect(recovery.request).toMatchObject({
direction: "after",
cursor: { epoch: "epoch-agent-a", seq: 9 },
});
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying");
recovery.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("reports a chat current once the latest-tail fallback replaces an overflowing catch-up", async () => {
const world = new TimelineWorld();
world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 });
@@ -780,6 +766,21 @@ test("overlapping sources deduplicate membership and retain open hidden chats",
world.expectNoPendingFetch();
});
test("failed timeline fetch does not schedule a retry", async () => {
vi.useFakeTimers();
const world = new TimelineWorld();
world.sync.setConnected(true);
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).succeed();
(await world.nextFetch("agent-a")).fail("unavailable");
await world.nextError();
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(120_000);
world.expectNoPendingFetch();
world.expectNoPendingMembership();
vi.useRealTimers();
});
test("a failed catch-up reports once and retries through the explicit retry policy", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -788,12 +789,12 @@ test("a failed catch-up reports once and retries through the explicit retry poli
membership.succeed();
const failed = await world.nextFetch("agent-a");
failed.fail("timeline unavailable");
const [error, retryCatchUp] = await Promise.all([world.nextError(), world.nextRetry()]);
const error = await world.nextError();
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryCatchUp();
world.sync.retryVisibleAgentTimeline("agent-a");
const retry = await world.nextFetch("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying");
retry.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
@@ -804,35 +805,6 @@ test("a failed catch-up reports once and retries through the explicit retry poli
world.expectNoPendingMembership();
});
test("a failed catch-up retries with exponential backoff", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const membership = await world.nextMembership();
membership.succeed();
const first = await world.nextFetch("agent-a");
first.fail("timeline unavailable");
const [firstError, retryAfterFirstFailure] = await Promise.all([
world.nextError(),
world.nextRetry(),
]);
expect(firstError).toBe("timeline unavailable");
retryAfterFirstFailure();
const second = await world.nextFetch("agent-a");
second.fail("timeline unavailable");
await world.nextError();
const retryAfterSecondFailure = await world.nextRetry(2_000);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryAfterSecondFailure();
const third = await world.nextFetch("agent-a");
third.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("manual retries can immediately re-attempt a failed catch-up", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
@@ -842,7 +814,7 @@ test("manual retries can immediately re-attempt a failed catch-up", async () =>
const failed = await world.nextFetch("agent-a");
failed.fail("timeline unavailable");
await world.nextRetry();
await world.nextError();
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
world.sync.retryVisibleAgentTimeline("agent-a");
@@ -853,7 +825,7 @@ test("manual retries can immediately re-attempt a failed catch-up", async () =>
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("redeclaring unchanged visibility does not bypass catch-up backoff", async () => {
test("redeclaring unchanged visibility does not bypass catch-up error", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
@@ -862,9 +834,11 @@ test("redeclaring unchanged visibility does not bypass catch-up backoff", async
const failed = await world.nextFetch("agent-a");
failed.fail("timeline unavailable");
await world.nextRetry();
await world.nextError();
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
world.sync.setActive(false);
world.sync.setActive(true);
world.expectNoPendingMembership();
world.expectNoPendingFetch();
@@ -880,7 +854,7 @@ test("a manual retry that fails returns to the error state", async () => {
const failed = await world.nextFetch("agent-a");
failed.fail("timeline unavailable");
await world.nextRetry();
await world.nextError();
world.sync.retryVisibleAgentTimeline("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying");
@@ -943,20 +917,34 @@ test("repeated recovery for the same running gap reuses the in-flight fetch", as
gapPage.respond({ hasNewer: false });
});
test("membership failure autonomously retries without another visibility declaration", async () => {
test("failed membership does not schedule a retry", async () => {
vi.useFakeTimers();
const world = new TimelineWorld();
world.sync.setConnected(true);
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).fail("unavailable");
await world.nextError();
expect(vi.getTimerCount()).toBe(0);
await vi.advanceTimersByTimeAsync(120_000);
world.expectNoPendingFetch();
world.expectNoPendingMembership();
vi.useRealTimers();
});
test("membership failure waits for manual retry", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const failed = await world.nextMembership();
failed.fail("subscription unavailable");
const [error, retryMembership] = await Promise.all([world.nextError(), world.nextRetry()]);
const error = await world.nextError();
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryMembership();
world.sync.retryVisibleAgentTimeline("agent-a");
const retry = await world.nextMembership();
retry.succeed();
const catchUp = await world.nextFetch("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying");
catchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
@@ -967,44 +955,68 @@ test("membership failure autonomously retries without another visibility declara
});
});
test("membership failures retry with exponential backoff", async () => {
test("retrying membership restores a previously loaded chat", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const first = await world.nextMembership();
first.fail("subscription unavailable");
const [firstError, retryAfterFirstFailure] = await Promise.all([
world.nextError(),
world.nextRetry(),
]);
expect(firstError).toBe("subscription unavailable");
retryAfterFirstFailure();
const second = await world.nextMembership();
second.fail("subscription unavailable again");
const retryAfterSecondFailure = await world.nextRetry(2_000);
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
retryAfterSecondFailure();
const retry = await world.nextMembership();
retry.succeed();
const catchUp = await world.nextFetch("agent-a");
catchUp.respond({ hasNewer: false });
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).succeed();
(await world.nextFetch("agent-a")).respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 10 });
world.show("workspace", ["agent-b"]);
(await world.nextMembership()).fail("subscription unavailable");
await world.nextError();
world.sync.retryVisibleAgentTimeline("agent-a");
(await world.nextMembership()).succeed();
(await world.nextFetch("agent-b")).respond({ hasNewer: false });
const resumed = await world.nextFetch("agent-a");
expect(resumed.request).toMatchObject({
direction: "after",
cursor: { epoch: "epoch-agent-a", seq: 10 },
});
resumed.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
});
test("redeclaring unchanged visibility does not bypass membership backoff", async () => {
test("a fetch that finishes after membership failure cannot clear the error", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).succeed();
const inFlight = await world.nextFetch("agent-a");
world.show("workspace", ["agent-b"]);
(await world.nextMembership()).fail("subscription unavailable");
await world.nextError();
inFlight.respond({ hasNewer: false });
await Promise.resolve();
await Promise.resolve();
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error");
});
test("a failed membership retry returns to error", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.show("workspace", ["agent-a"]);
(await world.nextMembership()).fail("unavailable");
await world.nextError();
world.sync.retryVisibleAgentTimeline("agent-a");
expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("retrying");
(await world.nextMembership()).fail("still unavailable");
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("error"));
});
test("redeclaring unchanged visibility does not bypass membership error", async () => {
const world = new TimelineWorld();
world.sync.setConnected(true);
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const failed = await world.nextMembership();
failed.fail("subscription unavailable");
await world.nextRetry();
await world.nextError();
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
world.sync.setActive(false);
world.sync.setActive(true);
world.expectNoPendingMembership();
world.expectNoPendingFetch();
@@ -1050,7 +1062,6 @@ test("backgrounding preserves subscriptions and returning verifies the visible c
world.sync.setActive(false);
world.expectNoPendingMembership();
world.elapse(60_000);
world.sync.setActive(true);
world.expectNoPendingMembership();
@@ -1067,11 +1078,10 @@ test("stale membership retry cannot overwrite a newer effective set", async () =
world.show("workspace", ["agent-a"]);
const failed = await world.nextMembership();
failed.fail("subscription unavailable");
const staleRetry = await world.nextRetry();
await world.nextError();
world.show("workspace", ["agent-b"]);
const current = await world.nextMembership();
staleRetry();
current.succeed();
const agentB = await world.nextFetch("agent-b");
agentB.respond({ hasNewer: false });
@@ -1088,10 +1098,9 @@ test("membership retry cannot run while disconnected", async () => {
world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
const failed = await world.nextMembership();
failed.fail("subscription unavailable");
const disconnectedRetry = await world.nextRetry();
await world.nextError();
world.sync.setConnected(false);
disconnectedRetry();
world.expectNoPendingMembership();
world.sync.setConnected(true);
const restored = await world.nextMembership();
@@ -1118,7 +1127,6 @@ test("returning to an open hidden agent stays live after inactivity", async () =
agentBCatchUp.respond({ hasNewer: false });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
world.elapse(60_000);
world.show("workspace", ["agent-a"]);
world.expectNoPendingMembership();
@@ -329,7 +329,6 @@ export interface ViewedTimelineSyncPorts {
*/
onCatchUpEnded(agentId: string): void;
reportError(error: unknown): void;
schedule(task: () => void, delayMs: number): () => void;
}
export type ViewedTimelineStatus = "ready" | "pending" | "error" | "retrying";
@@ -423,26 +422,14 @@ export function createViewedTimelineOwner(input: {
};
}
const RETRY_DELAY_MS = 1_000;
const MAX_RETRY_DELAY_MS = 30_000;
type CatchUpStatus = "running" | "complete" | "error";
interface CatchUpState {
generation: number;
status: CatchUpStatus;
request?: ProjectedTimelineForwardFetchPlan;
cancelRetry?: () => void;
retryDelayMs?: number;
}
const getNextRetryDelayMs = (previousDelayMs: number | undefined): number => {
if (previousDelayMs == null) {
return RETRY_DELAY_MS;
}
return Math.min(previousDelayMs * 2, MAX_RETRY_DELAY_MS);
};
function isSameCatchUpRequest(
left: ProjectedTimelineForwardFetchPlan | undefined,
right: ProjectedTimelineForwardFetchPlan | undefined,
@@ -472,9 +459,7 @@ function decideCatchUp(input: {
}
return "replace";
}
return input.current.status === "running" || input.current.status === "complete"
? "keep"
: "replace";
return "keep";
}
function normalizeAgentIds(agentIds: string[]): string[] {
@@ -494,8 +479,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const pendingCatchUps = new Map<string, ProjectedTimelineForwardFetchPlan>();
const visibilityCatchUpPending = new Set<string>();
const visibilityCatchUpErrors = new Map<string, string>();
// User-initiated retries only. Background retries stay silent; a retry the user asked for
// owes them a pending state until it settles.
// A user-initiated retry owes a pending state until it settles.
const manualRetries = new Set<string>();
const loadedCache = new Set<string>();
const cacheLoads = new Map<string, Promise<void>>();
@@ -511,8 +495,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
let acknowledged: string[] = [];
let membershipGeneration = 0;
let membershipNeedsRetry = false;
let membershipRetryDelayMs: number | undefined;
let cancelMembershipRetry: (() => void) | null = null;
// Chats this session has opened. Visibility is the only way in, so an agent is never
// subscribed — and never resumed on the daemon — until the user actually opens it.
// Membership then survives hiding, workspace switches, backgrounding and reconnect;
@@ -563,7 +545,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const cancelCatchUp = (agentId: string) => {
catchUpGenerations.set(agentId, (catchUpGenerations.get(agentId) ?? 0) + 1);
catchUps.get(agentId)?.cancelRetry?.();
catchUps.delete(agentId);
pendingCatchUps.delete(agentId);
};
@@ -607,19 +588,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
setVisibilityCatchUpReady(agentId);
} catch (error) {
if (catchUps.get(agentId)?.generation === generation) {
const nextRetryDelayMs = getNextRetryDelayMs(catchUps.get(agentId)?.retryDelayMs);
const cancelRetry = ports.schedule(() => {
const current = catchUps.get(agentId);
if (current?.generation !== generation || current.status !== "error") return;
startCatchUp(agentId);
}, nextRetryDelayMs);
catchUps.set(agentId, {
generation,
status: "error",
request,
cancelRetry,
retryDelayMs: nextRetryDelayMs,
});
catchUps.set(agentId, { generation, status: "error", request });
setVisibilityCatchUpError([agentId], error);
ports.reportError(error);
}
@@ -686,17 +655,9 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
if (decision === "keep") {
return;
}
current?.cancelRetry?.();
const generation = (catchUpGenerations.get(agentId) ?? 0) + 1;
catchUpGenerations.set(agentId, generation);
const retryDelayMs =
supersede || current?.status !== "error" ? undefined : current.retryDelayMs;
catchUps.set(agentId, {
generation,
status: "running",
request: nextRequest,
retryDelayMs,
});
catchUps.set(agentId, { generation, status: "running", request: nextRequest });
pendingCatchUps.delete(agentId);
void fetchUntilCurrent(
agentId,
@@ -715,7 +676,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
for (const agentId of ordered) {
const pendingCatchUp = pendingCatchUps.get(agentId);
const current = catchUps.get(agentId);
// Completion of a different chat must not bypass a failed chat's backoff or
// Completion of a different chat must not bypass a failed chat's error or
// supersede an in-flight tail that owns its parked gap recovery.
if (current && !(current.status === "complete" && pendingCatchUp)) continue;
startCatchUp(agentId, {
@@ -741,42 +702,31 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
} catch (error) {
if (disposed || !connected || generation !== membershipGeneration) return;
membershipNeedsRetry = true;
acknowledged = [];
for (const agentId of requested) cancelCatchUp(agentId);
setVisibilityCatchUpError(requested, error);
cancelMembershipRetry?.();
const nextRetryDelayMs = getNextRetryDelayMs(membershipRetryDelayMs);
cancelMembershipRetry = ports.schedule(() => {
cancelMembershipRetry = null;
if (disposed || !connected || membershipGeneration !== generation) return;
void reconcileMembership();
}, nextRetryDelayMs);
membershipRetryDelayMs = nextRetryDelayMs;
ports.reportError(error);
return;
}
if (disposed || !connected || generation !== membershipGeneration) return;
cancelMembershipRetry?.();
cancelMembershipRetry = null;
membershipRetryDelayMs = undefined;
acknowledged = requested;
startAcknowledgedCatchUps();
};
const retryVisibleAgentTimeline = (agentId: string) => {
if (!isDesired(agentId) || manualRetries.has(agentId)) return;
if (!connected || !isDesired(agentId) || manualRetries.has(agentId)) return;
const catchUp = catchUps.get(agentId);
const membershipRetryable = membershipNeedsRetry && connected;
if (catchUp?.status !== "error" && !membershipRetryable) return;
manualRetries.add(agentId);
notifyListeners();
if (catchUp?.status === "error") {
catchUp.cancelRetry?.();
startCatchUp(agentId, { request: catchUp.request, supersede: true });
if (membershipRetryable) {
void reconcileMembership();
return;
}
cancelMembershipRetry?.();
cancelMembershipRetry = null;
membershipRetryDelayMs = undefined;
void reconcileMembership();
if (catchUp?.status === "error") {
startCatchUp(agentId, { request: catchUp.request, supersede: true });
}
};
const commitDesiredMembership = (
@@ -817,8 +767,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
ensureCacheLoaded(agentId);
}
}
cancelMembershipRetry?.();
cancelMembershipRetry = null;
desired = nextDesired;
ports.replaceDemandedAgentIds(desired);
membershipGeneration += 1;
@@ -869,6 +817,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
return;
}
for (const agentId of visibleAgentIds()) {
if (visibilityCatchUpErrors.has(agentId)) continue;
visibilityCatchUpPending.add(agentId);
visibilityCatchUpErrors.delete(agentId);
startCatchUp(agentId, { supersede: true });
@@ -880,9 +829,6 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
connected = nextConnected;
if (!connected) {
commitDesiredMembership(desired, { resetCatchUpStatus: true });
cancelMembershipRetry?.();
cancelMembershipRetry = null;
membershipRetryDelayMs = undefined;
void observation?.handle.release().catch(ports.reportError);
observation = null;
acknowledged = [];
@@ -894,7 +840,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
void reconcileMembership();
},
recoverGap(agentId, cursor) {
if (!isDesired(agentId)) return;
if (!isDesired(agentId) || catchUps.get(agentId)?.status === "error") return;
startCatchUp(agentId, {
request: planTimelineCatchUpAfter({ epoch: cursor.epoch, seq: cursor.endSeq }),
supersede: true,
@@ -904,10 +850,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
disposed = true;
void observation?.handle.release().catch(ports.reportError);
observation = null;
cancelMembershipRetry?.();
cancelMembershipRetry = null;
membershipNeedsRetry = false;
membershipRetryDelayMs = undefined;
sources.clear();
membershipGeneration += 1;
for (const agentId of desired) cancelCatchUp(agentId);