puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-11-09 09:54:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(app): keep hidden chats live while a subscription is replaced

Browse Source 
The old timeline subscription was released in the same tick the new one
was requested. The client deletes a route on release and adds the new
route only when the daemon responds, so events sent in between were
dropped. Visible chats recover through the cursor check; hidden chats
skip it and stayed behind until something else fetched them.

Release the old subscription when the new one settles. Accepting the new
route and releasing the old one happen without a socket frame in
between, so each event reaches the client once.
This commit is contained in:
Mohamed Boudra
2026-11-09 10:51:22 +01:00
parent 130dec624d
commit ef19db587e
2 changed files with 60 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/timeline/viewed-timeline-sync.test.ts
+56 -1
View File
@@ -173,6 +173,7 @@ class TimelineWorld {
ready: result.promise,
release: async () => {
this.releasedMemberships.push(agentIds);
result.reject(new Error("Subscription released"));
},
};
},
@@ -1371,7 +1372,6 @@ test("foregrounding a current chat checks over the subscription instead of fetch
const check = await world.nextMembership();
expect(check.agentIds).toEqual(["agent-a"]);
expect(check.cursors).toEqual({ "agent-a": { epoch: "epoch-a", seq: 42 } });
await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]]));
check.succeed({ "agent-a": "current" });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
@@ -1379,6 +1379,61 @@ test("foregrounding a current chat checks over the subscription instead of fetch
world.sync.dispose();
});
test("the old subscription keeps delivering until the daemon accepts its replacement", async () => {
const world = freshnessWorld();
world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 });
world.cursors.set("agent-b", { epoch: "epoch-b", endSeq: 2 });
world.openedEarlier(["agent-b"]);
world.show("pane", ["agent-a"]);
world.sync.setConnected(true);
(await world.nextMembership()).succeed({ "agent-a": "current", "agent-b": "current" });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-b")).toBe("ready"));
world.sync.setActive(false);
world.sync.setActive(true);
const check = await world.nextMembership();
await new Promise((resolve) => setTimeout(resolve, 0));
expect(world.releasedMemberships).toEqual([]);
check.succeed({ "agent-a": "current" });
await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a", "agent-b"]]));
world.sync.dispose();
});
test("losing the connection during a check releases both the check and what it replaced", async () => {
const world = freshnessWorld();
world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 });
world.show("pane", ["agent-a"]);
world.sync.setConnected(true);
(await world.nextMembership()).succeed({ "agent-a": "current" });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.setActive(false);
world.sync.setActive(true);
await world.nextMembership();
world.sync.setConnected(false);
await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"], ["agent-a"]]));
world.sync.dispose();
});
test("a replacement the daemon refuses still releases the subscription it replaced", async () => {
const world = freshnessWorld();
world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 });
world.show("pane", ["agent-a"]);
world.sync.setConnected(true);
(await world.nextMembership()).succeed({ "agent-a": "current" });
await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready"));
world.sync.setActive(false);
world.sync.setActive(true);
const check = await world.nextMembership();
check.fail("refused");
await vi.waitFor(() => expect(world.releasedMemberships).toEqual([["agent-a"]]));
world.sync.dispose();
});
test("foregrounding asks only about visible chats; hidden chats stay caught up", async () => {
const world = freshnessWorld();
world.cursors.set("agent-a", { epoch: "epoch-a", endSeq: 1 });
packages/app/src/timeline/viewed-timeline-sync.ts
+4 -2
View File
@@ -921,8 +921,10 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
const previous = observation;
const handle = ports.observe(requested, cursors);
observation = { handle, agentIds: requested };
void previous?.handle.release().catch(ports.reportError);
const answer = await handle.ready;
// A subscription routes events only once the daemon accepts it, so the old one delivers until then.
const answer = await handle.ready.finally(() => {
void previous?.handle.release().catch(ports.reportError);
});
timelines = answer ? answer.timelines : undefined;
} catch (error) {
if (isReconcileSuperseded(generation)) return;