Max
2026-08-22 19:44:43 -04:00
parent c9a26d31f2
commit db4bf9d33e
7 changed files with 253 additions and 29 deletions

View File

@@ -11,6 +11,7 @@ import {
mergeLiveChannelWindowEvent,
mergeLiveThreadSummary,
replaceNewestChannelWindow,
stageOlderChannelWindow,
} from "./channelWindowStore.ts";
function event(id, createdAt, kind = 9) {
@@ -113,23 +114,57 @@ test("rejects inconsistent exhaustion and cursor facts", () => {
);
});
test("newest refresh drops a stale tail when its boundary moves", () => {
test("newest refresh drops a stale tail and its staged successor", () => {
const first = page(null, [event("a", 100)]);
const loaded = appendOlderChannelWindow(
const staged = page(first.nextCursor, [event("z", 90)], { hasMore: false });
const loaded = stageOlderChannelWindow(
replaceNewestChannelWindow(emptyChannelWindowStore(), first),
page(first.nextCursor, [event("z", 90)], { hasMore: false }),
staged,
);
const refreshed = replaceNewestChannelWindow(
loaded,
page(null, [event("n", 110), event("a", 100)]),
);
assert.equal(refreshed.pages.length, 1);
assert.equal(refreshed.stagedPage, null);
assert.deepEqual(
flattenChannelWindowEvents(refreshed).map((item) => item.content),
["a", "n"],
);
});
test("staging replaces at most one matching successor without rendering it", () => {
const first = page(null, [event("a", 100)]);
const store = replaceNewestChannelWindow(emptyChannelWindowStore(), first);
const staged = page(first.nextCursor, [event("b", 90)]);
const replacement = page(first.nextCursor, [event("c", 80)], {
hasMore: false,
});
const once = stageOlderChannelWindow(store, staged);
const twice = stageOlderChannelWindow(once, replacement);
assert.equal(twice.stagedPage, replacement);
assert.equal(twice.pages.length, 1);
assert.deepEqual(
flattenChannelWindowEvents(twice).map((item) => item.content),
["a"],
);
});
test("staging discards a page outside the retained tail cursor", () => {
const first = page(null, [event("a", 100)]);
const store = replaceNewestChannelWindow(emptyChannelWindowStore(), first);
assert.equal(
stageOlderChannelWindow(
store,
page(cursor(event("wrong", 70)), [event("older", 60)]),
),
store,
);
});
test("live rows arriving before page zero enter the overlay", () => {
const live = event("n", 110);
const store = mergeLiveChannelWindowEvent(emptyChannelWindowStore(), live);

View File

@@ -25,6 +25,8 @@ export type ChannelWindowPage = {
};
export type ChannelWindowStore = {
pages: ChannelWindowPage[];
/** One fetched successor held outside the rendered page chain. */
stagedPage: ChannelWindowPage | null;
/** Top-level live events not represented in an authoritative relay page. */
liveOverlay: RelayEvent[];
/** Live structural events retained independently from frozen page closure. */
@@ -38,6 +40,7 @@ export type ChannelWindowStore = {
export const emptyChannelWindowStore = (): ChannelWindowStore => ({
pages: [],
stagedPage: null,
liveOverlay: [],
liveAux: [],
liveSummaries: {},
@@ -112,6 +115,7 @@ export function replaceNewestChannelWindow(
const auxIds = new Set(page.aux.map((event) => event.id));
return {
pages: [page],
stagedPage: null,
liveOverlay: current.liveOverlay.filter((event) => !ids.has(event.id)),
liveAux: current.liveAux.filter((event) => !auxIds.has(event.id)),
// A head refetch is the authoritative resync moment (subscribe/reconnect
@@ -151,10 +155,23 @@ export function appendOlderChannelWindow(
return {
...current,
pages: [...current.pages, page],
stagedPage: null,
liveOverlay: current.liveOverlay.filter((event) => !pageIds.has(event.id)),
};
}
/** Hold one cursor-matched successor outside the rendered page chain. */
export function stageOlderChannelWindow(
current: ChannelWindowStore,
page: ChannelWindowPage,
): ChannelWindowStore {
assertValidPage(page);
const tail = current.pages[current.pages.length - 1];
if (!tail?.hasMore || !tail.nextCursor) return current;
if (!cursorsEqual(page.startCursor, tail.nextCursor)) return current;
return { ...current, stagedPage: page };
}
/**
* Record a relay-pushed live `39005` summary. Newest `created_at` wins per
* root: the relay pushes a full recount on every thread mutation, so the

View File

@@ -0,0 +1,116 @@
import assert from "node:assert/strict";
import test from "node:test";
import { QueryClient } from "@tanstack/react-query";
import {
emptyChannelWindowStore,
replaceNewestChannelWindow,
} from "./channelWindowStore.ts";
import { channelMessagesKey, channelWindowKey } from "./messageQueryKeys.ts";
import { pageOlderMessagesUntilRowFloor } from "./pageOlderMessages.ts";
function event(id, createdAt) {
return {
id: id.padEnd(64, "0"),
pubkey: "a".repeat(64),
created_at: createdAt,
kind: 9,
tags: [["h", "channel"]],
content: id,
sig: "b".repeat(128),
};
}
const cursor = (item) => ({ createdAt: item.created_at, eventId: item.id });
function page(startCursor, rows, hasMore = true) {
return {
startCursor,
rows: rows.map((item) => ({ event: item, thread: null })),
aux: [],
nextCursor: hasMore ? cursor(rows.at(-1)) : null,
hasMore,
};
}
function harness(channelId) {
const queryClient = new QueryClient({
defaultOptions: { queries: { retry: false } },
});
const head = page(null, [event(`${channelId}-head`, 300)]);
queryClient.setQueryData(
channelWindowKey(channelId),
replaceNewestChannelWindow(emptyChannelWindowStore(), head),
);
queryClient.setQueryData(channelMessagesKey(channelId), [head.rows[0].event]);
return { queryClient, head };
}
const flush = () => new Promise((resolve) => setImmediate(resolve));
test("commits one page and stages exactly one successor outside projection", async () => {
const channelId = "stage-successor";
const { queryClient, head } = harness(channelId);
const older = page(head.nextCursor, [event("older", 200)]);
const successor = page(older.nextCursor, [event("successor", 100)], false);
const calls = [];
const fetchPage = async (_channelId, requestCursor) => {
calls.push(requestCursor);
return calls.length === 1 ? older : successor;
};
await pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
await flush();
const store = queryClient.getQueryData(channelWindowKey(channelId));
assert.deepEqual(calls, [head.nextCursor, older.nextCursor]);
assert.equal(store.pages.length, 2);
assert.equal(store.stagedPage, successor);
assert.deepEqual(
queryClient
.getQueryData(channelMessagesKey(channelId))
.map((item) => item.content),
["older", `${channelId}-head`],
);
});
test("next paging pass consumes the staged page without network wait", async () => {
const channelId = "consume-staged";
const { queryClient, head } = harness(channelId);
const older = page(head.nextCursor, [event("older", 200)]);
const successor = page(older.nextCursor, [event("successor", 100)], false);
let calls = 0;
const fetchPage = async () => {
calls += 1;
return calls === 1 ? older : successor;
};
await pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
await flush();
assert.equal(calls, 2);
await pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
() => {
throw new Error("staged page should make this pass network-free");
},
);
const store = queryClient.getQueryData(channelWindowKey(channelId));
assert.equal(store.pages.length, 3);
assert.equal(store.stagedPage, null);
assert.deepEqual(
queryClient
.getQueryData(channelMessagesKey(channelId))
.map((item) => item.content),
["successor", "older", `${channelId}-head`],
);
});

View File

@@ -2,7 +2,10 @@ import type { QueryClient } from "@tanstack/react-query";
import {
appendOlderChannelWindow,
type ChannelWindowCursor,
type ChannelWindowPage,
type ChannelWindowStore,
stageOlderChannelWindow,
} from "@/features/messages/lib/channelWindowStore";
import { projectChannelWindowMessages } from "@/features/messages/lib/projectChannelWindow";
import { parseChannelWindowResponse } from "@/features/messages/lib/channelWindowResponse";
@@ -12,26 +15,53 @@ import { getChannelWindowEvents } from "@/shared/api/channelWindow";
const CHANNEL_WINDOW_PAGE_SIZE = 50;
export type PageOlderResult = { hasOlderMessages: boolean };
const inFlightPasses = new Map<string, Promise<PageOlderResult>>();
type FetchPage = (
channelId: string,
requestCursor: ChannelWindowCursor,
) => Promise<ChannelWindowPage>;
/** Fetch exactly one server-defined older window and append it atomically. */
/** Append one older window, then stage its successor without projection. */
export function pageOlderMessagesUntilRowFloor(
queryClient: QueryClient,
channelId: string,
shouldContinue: () => boolean,
fetchPageFn: FetchPage = fetchPage,
): Promise<PageOlderResult> {
const running = inFlightPasses.get(channelId);
if (running) return running;
const pass = runPage(queryClient, channelId, shouldContinue).finally(() => {
const pass = runPage(
queryClient,
channelId,
shouldContinue,
fetchPageFn,
).finally(() => {
inFlightPasses.delete(channelId);
});
inFlightPasses.set(channelId, pass);
return pass;
}
async function fetchPage(
channelId: string,
requestCursor: ChannelWindowCursor,
) {
const events = await getChannelWindowEvents(
channelId,
requestCursor,
CHANNEL_WINDOW_PAGE_SIZE,
);
return parseChannelWindowResponse(events, channelId, requestCursor);
}
function cursorsEqual(left: ChannelWindowCursor, right: ChannelWindowCursor) {
return left.createdAt === right.createdAt && left.eventId === right.eventId;
}
async function runPage(
queryClient: QueryClient,
channelId: string,
shouldContinue: () => boolean,
fetchPageFn: FetchPage,
): Promise<PageOlderResult> {
const store = queryClient.getQueryData<ChannelWindowStore>(
channelWindowKey(channelId),
@@ -42,13 +72,13 @@ async function runPage(
}
const requestCursor = tail.nextCursor;
const events = await getChannelWindowEvents(
channelId,
requestCursor,
CHANNEL_WINDOW_PAGE_SIZE,
);
const matchingStagedPage = store.stagedPage;
const page =
matchingStagedPage?.startCursor &&
cursorsEqual(matchingStagedPage.startCursor, requestCursor)
? matchingStagedPage
: await fetchPageFn(channelId, requestCursor);
if (!shouldContinue()) return { hasOlderMessages: true };
const page = parseChannelWindowResponse(events, channelId, requestCursor);
const retained = queryClient.getQueryData<ChannelWindowStore>(
channelWindowKey(channelId),
);
@@ -56,5 +86,21 @@ async function runPage(
const next = appendOlderChannelWindow(retained, page);
queryClient.setQueryData(channelWindowKey(channelId), next);
projectChannelWindowMessages(queryClient, channelId);
return { hasOlderMessages: page.hasMore };
if (!page.hasMore || !page.nextCursor || !shouldContinue()) {
return { hasOlderMessages: false };
}
void fetchPageFn(channelId, page.nextCursor)
.then((staged) => {
if (!shouldContinue()) return;
queryClient.setQueryData<ChannelWindowStore>(
channelWindowKey(channelId),
(current) =>
current ? stageOlderChannelWindow(current, staged) : current,
);
})
.catch((error) => {
console.error("Failed to stage older messages", channelId, error);
});
return { hasOlderMessages: true };
}

View File

@@ -23,7 +23,6 @@ import { TimelineSkeleton, useTimelineSkeletonRows } from "./TimelineSkeleton";
import { TimelineMessageList } from "./TimelineMessageList";
import type { TimelineVirtualizerApi } from "./TimelineMessageList";
import { useAnchoredScroll } from "./useAnchoredScroll";
import { useLoadOlderOnScroll } from "./useLoadOlderOnScroll";
import { useBufferedTimelineMessages } from "./useBufferedTimelineMessages";
import {
DirectMessageIntroAvatarStack,
@@ -600,14 +599,6 @@ const MessageTimelineBase = React.forwardRef<
showTimelineSkeleton,
]);
useLoadOlderOnScroll({
fetchOlder: useTimelineVirtualizer ? undefined : fetchOlder,
hasOlderMessages,
isLoading: showTimelineSkeleton,
scrollContainerRef: activeScrollContainerRef,
sentinelRef: topSentinelRef,
});
const timelineSkeletonRows = useTimelineSkeletonRows({
channelId,
isLoading: showTimelineSkeleton,

View File

@@ -718,8 +718,8 @@ function VirtualizedTimelineRows({
// touch, and key listeners are the authoritative user-interaction gate.
onAtBottomStateChange?.(distanceFromBottom <= 32);
updatePinnedDayLabel(offset);
if (offset <= 200) {
// Layout scrolls near the top must not poison the reader's next input.
if (offset <= list.viewportSize * 1.5) {
// Begin loading before momentum reaches the retained history boundary.
armUpwardMomentum(onStartReached?.() ?? false);
}
},

View File

@@ -1995,13 +1995,32 @@ test("one scroll-up gesture pages older history once, not to the channel top", a
const pagesFetched = await fetchCount();
const deepest = await oldestRenderedIndex();
// One gesture, one page. `pageOlderMessagesUntilRowFloor` fetches exactly
// one server-defined window per call (#1500), and the trigger refuses to
// re-fire while a landed page is still unpainted, so anything above one is a
// cascade off the resolve→commit gap.
expect(pagesFetched).toBeLessThanOrEqual(1);
// And it must NOT have reached the oldest seeded root on its own.
// One gesture commits one page and starts exactly one successor fetch for the
// staged slot. The successor remains outside projection, so a second request
// is expected but a third would still be a resolve→commit cascade.
expect(pagesFetched).toBeLessThanOrEqual(2);
// The staged successor must NOT have rendered on its own.
expect(deepest ?? Number.POSITIVE_INFINITY).toBeGreaterThan(50);
// Once staging is complete, the next gesture consumes that page locally.
// Make any accidental network request visibly too slow, then require the next
// page to paint within 500ms without incrementing the request counter.
await expect.poll(fetchCount).toBe(2);
await page.evaluate(() => {
window.__BUZZ_E2E__ = {
...window.__BUZZ_E2E__,
mock: { ...window.__BUZZ_E2E__?.mock, channelWindowDelayMs: 2_000 },
};
});
await page.mouse.wheel(0, 1_000);
await page.waitForTimeout(50);
await page.mouse.wheel(0, -4_000);
await expect
.poll(oldestRenderedIndex, { timeout: 500 })
.toBeLessThan(deepest ?? Number.POSITIVE_INFINITY);
// The only new request is the background successor stage; rendering did not
// wait for its 2s delay.
expect(await fetchCount()).toBe(3);
});
// Regression for Wes's "after a page loads I'm yanked to the oldest of the new