fix(desktop): preserve live channel timelines (#5662)

## Summary

- restore the post-subscribe channel-window refresh that closes the gap
left by a live subscription starting at the current second
- prevent an unresolved, pageless channel window from replacing a
populated timeline cache with its first live event
- replace the invalid freshness-gate tests with a regression reproducing
the populated cache + pageless window + first live event state from the
report

## Root cause

This was a data-projection bug, not a virtualized-row failure. PR #5577
skipped the post-subscribe refresh for a fresh cache even though
`subscribeToChannelLive` starts at `since: now`, leaving events between
the cached page and subscription establishment undiscovered. A
successful but pageless companion window could then receive one live
event and project that one-row overlay over the populated message cache.
Reload fetched page zero and restored the conversation.

## Validation

Validated exact head `bfbaefe95da5452cdda3a0b5df970eb11e44f6f8`:

- focused `projectChannelWindow.test.mjs`: 9/9 passed
- pre-push: branch skew, desktop check, desktop typecheck, and all 4,715
desktop tests passed
- independent fresh-frame review: 9/10, no blockers

## Authorship disclosure

Carl implemented and is posting this change on Wes's behalf.

Signed-off-by: Wes <wesbillman@users.noreply.github.com>
Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
This commit is contained in:
Wes
2026-08-12 16:19:43 +00:00
committed by GitHub
co-authored by Carl
parent 63f961c7e4
commit 63d14a0e95
4 changed files with 131 additions and 124 deletions
+40 -15
View File
@@ -1,5 +1,10 @@
import { useEffect, useEffectEvent } from "react";
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import {
type QueryClient,
useMutation,
useQuery,
useQueryClient,
} from "@tanstack/react-query";
import { toast } from "sonner";
import {
@@ -17,7 +22,6 @@ import {
import {
projectChannelWindowMessages,
refreshChannelWindowMessages,
shouldRefreshChannelWindowAfterSubscribe,
} from "@/features/messages/lib/projectChannelWindow";
import { reconcileChannelWindowMessages } from "@/features/messages/lib/channelWindowReconciliation";
import {
@@ -235,26 +239,46 @@ export function useChannelWindowQuery(channel: Channel | null) {
});
}
export function reconcileFetchedChannelWindow(
queryClient: QueryClient,
channelId: string,
events: Awaited<ReturnType<typeof getChannelWindowEvents>>,
previousMessages: RelayEvent[],
signal: AbortSignal,
): RelayEvent[] {
// Tauri invokes cannot be canceled after dispatch. A replacement refetch can
// therefore win while this older request is still in flight. Never let that
// canceled request commit its stale page into the authoritative window.
signal.throwIfAborted();
const windowKey = channelWindowKey(channelId);
const page = parseChannelWindowResponse(events, channelId, null);
const current =
queryClient.getQueryData<ChannelWindowStore>(windowKey) ??
emptyChannelWindowStore();
const next = replaceNewestChannelWindow(current, page);
queryClient.setQueryData(windowKey, next);
return reconcileChannelWindowMessages(next, previousMessages);
}
export function useChannelMessagesQuery(channel: Channel | null) {
const queryClient = useQueryClient();
const queryKey = channelMessagesKey(channel?.id ?? "none");
const windowKey = channelWindowKey(channel?.id ?? "none");
return useQuery({
enabled: channel !== null && channel.channelType !== "forum",
queryKey,
queryFn: async () => {
queryFn: async ({ signal }) => {
if (!channel) throw new Error("No channel selected.");
const previousMessages =
queryClient.getQueryData<RelayEvent[]>(queryKey) ?? [];
const events = await getChannelWindowEvents(channel.id);
const page = parseChannelWindowResponse(events, channel.id, null);
const current =
queryClient.getQueryData<ChannelWindowStore>(windowKey) ??
emptyChannelWindowStore();
const next = replaceNewestChannelWindow(current, page);
queryClient.setQueryData(windowKey, next);
return reconcileChannelWindowMessages(next, previousMessages);
return reconcileFetchedChannelWindow(
queryClient,
channel.id,
events,
previousMessages,
signal,
);
},
staleTime: 5 * 60 * 1_000,
gcTime: 60 * 60 * 1_000,
@@ -382,9 +406,10 @@ export function useChannelSubscription(channel: Channel | null) {
}
cleanup = dispose;
if (!shouldRefreshChannelWindowAfterSubscribe(queryClient, channelId)) {
return;
}
// The live subscription starts at "now", so it cannot close the gap
// between the last page snapshot and subscription establishment. Always
// refresh after the subscription is active; freshness alone is not a
// proof that no relay events landed in that interval.
void refreshNewestWindow().catch((error) => {
if (!isDisposed) {
console.error(
@@ -406,7 +431,7 @@ export function useChannelSubscription(channel: Channel | null) {
void cleanup();
}
};
}, [channelId, channelType, queryClient]);
}, [channelId, channelType]);
}
export function useSendMessageMutation(
@@ -28,6 +28,18 @@ export function reconcileChannelWindowMessages(
messages: RelayEvent[],
) {
const windowEvents = flattenChannelWindowEvents(window);
if (window.pages.length === 0) {
// A pageless window is unresolved, not authoritative. This state can exist
// briefly when the companion window query mounts beside an already-cached
// rendered timeline. Preserve that cache while admitting live events;
// otherwise the first live event projects a one-row overlay over the
// entire conversation until reload refetches page zero.
let merged = messages;
for (const event of windowEvents) {
merged = reconcileIncomingMessage(merged, event);
}
return [...merged].sort((left, right) => compareRelayOrder(right, left));
}
const authoritativeIds = new Set(windowEvents.map((event) => event.id));
const retained = retainRefetchReconciliationEvents(messages).filter(
(event) => !authoritativeIds.has(event.id),
@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
import test from "node:test";
import { QueryClient, QueryObserver } from "@tanstack/react-query";
import { reconcileFetchedChannelWindow } from "../hooks.ts";
import { channelMessagesKey, channelWindowKey } from "./messageQueryKeys.ts";
import {
appendOlderChannelWindow,
@@ -11,10 +12,8 @@ import {
replaceNewestChannelWindow,
} from "./channelWindowStore.ts";
import {
CHANNEL_WINDOW_FRESH_MS,
projectChannelWindowMessages,
refreshChannelWindowMessages,
shouldRefreshChannelWindowAfterSubscribe,
} from "./projectChannelWindow.ts";
import { reconcileChannelWindowMessages } from "./channelWindowReconciliation.ts";
@@ -30,6 +29,18 @@ function event(id, createdAt) {
};
}
function wirePage(rows) {
return [
...rows,
{
...event("bounds", 0),
kind: 39006,
tags: [["d", "channel:head"]],
content: JSON.stringify({ has_more: false, next_cursor: null }),
},
];
}
function newestPage(rows) {
return {
startCursor: null,
@@ -276,92 +287,79 @@ test("test_live_projection_retains_pending_send_and_non_broadcast_thread_reply",
]);
});
test("test_subscribe_refresh_skips_fresh_populated_window", () => {
test("test_canceled_stale_fetch_cannot_overwrite_catch_up_window", async () => {
const harness = createHarness();
const updatedAt = harness.client.getQueryState(
harness.windowKey,
).dataUpdatedAt;
assert.equal(
shouldRefreshChannelWindowAfterSubscribe(
harness.client,
harness.channelId,
updatedAt + CHANNEL_WINDOW_FRESH_MS - 1,
),
false,
);
});
test("test_subscribe_refresh_runs_for_stale_window", () => {
const harness = createHarness();
const updatedAt = harness.client.getQueryState(
harness.windowKey,
).dataUpdatedAt;
assert.equal(
shouldRefreshChannelWindowAfterSubscribe(
harness.client,
harness.channelId,
updatedAt + CHANNEL_WINDOW_FRESH_MS,
),
true,
);
});
test("test_live_cache_merge_does_not_extend_window_freshness", () => {
const harness = createHarness();
const windowUpdatedAt = harness.client.getQueryState(
harness.windowKey,
).dataUpdatedAt;
harness.client.setQueryData(harness.messagesKey, (messages) => [
...messages,
event("live-cache-only", 110),
]);
assert.equal(
shouldRefreshChannelWindowAfterSubscribe(
harness.client,
harness.channelId,
windowUpdatedAt + CHANNEL_WINDOW_FRESH_MS,
),
true,
);
});
test("test_subscribe_refresh_runs_without_a_message_query", () => {
const client = new QueryClient({
defaultOptions: { queries: { retry: false } },
const requests = [];
let resolveRequestStarted;
let requestStarted = new Promise((resolve) => {
resolveRequestStarted = resolve;
});
assert.equal(
shouldRefreshChannelWindowAfterSubscribe(client, "missing-channel"),
true,
);
});
test("test_subscribe_refresh_does_not_duplicate_inflight_initial_fetch", async () => {
const client = new QueryClient({
defaultOptions: { queries: { retry: false } },
});
const channelId = "pending-channel";
const queryKey = channelMessagesKey(channelId);
let resolveFetch;
const observer = new QueryObserver(client, {
queryKey,
queryFn: () =>
new Promise((resolve) => {
const observer = new QueryObserver(harness.client, {
queryKey: harness.messagesKey,
queryFn: async ({ signal }) => {
const previousMessages = harness.client.getQueryData(harness.messagesKey);
let resolveFetch;
const fetch = new Promise((resolve) => {
resolveFetch = resolve;
}),
});
requests.push({ resolveFetch, signal });
resolveRequestStarted();
const events = await fetch;
return reconcileFetchedChannelWindow(
harness.client,
harness.channelId,
events,
previousMessages,
signal,
);
},
});
const unsubscribe = observer.subscribe(() => {});
assert.equal(
shouldRefreshChannelWindowAfterSubscribe(client, channelId),
false,
await requestStarted;
requestStarted = new Promise((resolve) => {
resolveRequestStarted = resolve;
});
const catchUp = refreshChannelWindowMessages(
harness.client,
harness.channelId,
);
await requestStarted;
resolveFetch([]);
await client.getQueryCache().find({ queryKey })?.promise;
assert.equal(requests[0].signal.aborted, true);
requests[1].resolveFetch(
wirePage([event("gap", 110), event("initial", 100)]),
);
await catchUp;
assert.deepEqual(contents(harness), ["initial", "gap"]);
requests[0].resolveFetch(wirePage([event("initial", 100)]));
await new Promise((resolve) => setImmediate(resolve));
appendLiveEvent(harness, event("live", 120));
assert.deepEqual(contents(harness), ["initial", "gap", "live"]);
assert.deepEqual(
flattenChannelWindowEvents(
harness.client.getQueryData(harness.windowKey),
).map((item) => item.content),
["initial", "gap", "live"],
);
unsubscribe();
});
test("test_pageless_live_projection_preserves_cached_timeline", () => {
const harness = createHarness();
const cached = harness.client.getQueryData(harness.messagesKey);
const pageless = emptyChannelWindowStore();
harness.client.setQueryData(harness.windowKey, pageless);
const next = mergeLiveChannelWindowEvent(
harness.client.getQueryData(harness.windowKey),
event("live", 110),
);
harness.client.setQueryData(harness.windowKey, next);
projectChannelWindowMessages(harness.client, harness.channelId);
assert.deepEqual(contents(harness), ["initial", "live"]);
assert.equal(harness.client.getQueryData(harness.messagesKey)[0], cached[0]);
});
@@ -8,34 +8,6 @@ import {
} from "./channelWindowStore";
import { reconcileChannelWindowMessages } from "./channelWindowReconciliation";
export const CHANNEL_WINDOW_FRESH_MS = 5 * 60_000;
/**
* Subscription setup closes the gap between the initial page and live events,
* but revisiting a channel with a fresh page has no gap to close. Reconnects
* still refresh unconditionally at their call site.
*/
export function shouldRefreshChannelWindowAfterSubscribe(
queryClient: QueryClient,
channelId: string,
now = Date.now(),
): boolean {
const messagesState = queryClient.getQueryState(
channelMessagesKey(channelId),
);
if (!messagesState) return true;
if (messagesState.fetchStatus === "fetching") return false;
const windowState = queryClient.getQueryState(channelWindowKey(channelId));
if (
messagesState.status !== "success" ||
windowState?.status !== "success" ||
windowState.dataUpdatedAt === 0
) {
return true;
}
return now - windowState.dataUpdatedAt >= CHANNEL_WINDOW_FRESH_MS;
}
/** Keep the rendered timeline cache aligned with its authoritative window. */
export function projectChannelWindowMessages(
queryClient: QueryClient,