Max
2026-08-22 20:21:51 -04:00
parent db4bf9d33e
commit 856b32b020
6 changed files with 161 additions and 15 deletions

View File

@@ -46,7 +46,7 @@ export const emptyChannelWindowStore = (): ChannelWindowStore => ({
liveSummaries: {},
});
function cursorsEqual(
export function cursorsEqual(
left: ChannelWindowCursor | null,
right: ChannelWindowCursor | null,
) {

View File

@@ -75,6 +75,116 @@ test("commits one page and stages exactly one successor outside projection", asy
);
});
test("trigger during successor staging awaits the same request", async () => {
const channelId = "await-in-flight-stage";
const { queryClient, head } = harness(channelId);
const older = page(head.nextCursor, [event("older", 200)]);
const successor = page(older.nextCursor, [event("successor", 100)], false);
let releaseSuccessor;
const successorPending = new Promise((resolve) => {
releaseSuccessor = () => resolve(successor);
});
const calls = [];
const fetchPage = async (_channelId, requestCursor) => {
calls.push(requestCursor);
return calls.length === 1 ? older : successorPending;
};
await pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
const secondPass = pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
await flush();
assert.deepEqual(calls, [head.nextCursor, older.nextCursor]);
releaseSuccessor();
await secondPass;
assert.equal(
queryClient.getQueryData(channelWindowKey(channelId)).pages.length,
3,
);
});
test("replaced tail does not reuse its stale in-flight stage", async () => {
const channelId = "reject-stale-in-flight-stage";
const { queryClient, head } = harness(channelId);
const older = page(head.nextCursor, [event("older", 200)]);
const staleSuccessor = page(
older.nextCursor,
[event("stale-successor", 100)],
false,
);
const freshSuccessor = page(
older.nextCursor,
[event("fresh-successor", 90)],
false,
);
let releaseStaleSuccessor;
const staleSuccessorPending = new Promise((resolve) => {
releaseStaleSuccessor = () => resolve(staleSuccessor);
});
const calls = [];
const fetchPage = async (_channelId, requestCursor) => {
calls.push(requestCursor);
if (calls.length === 1) return older;
if (calls.length === 2) return staleSuccessorPending;
return freshSuccessor;
};
await pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
const replacementHead = page(null, [older.rows[0].event]);
queryClient.setQueryData(
channelWindowKey(channelId),
replaceNewestChannelWindow(
queryClient.getQueryData(channelWindowKey(channelId)),
replacementHead,
),
);
const replacementPass = pageOlderMessagesUntilRowFloor(
queryClient,
channelId,
() => true,
fetchPage,
);
try {
await flush();
assert.deepEqual(calls, [
head.nextCursor,
older.nextCursor,
replacementHead.nextCursor,
]);
} finally {
releaseStaleSuccessor();
}
await replacementPass;
assert.deepEqual(
queryClient
.getQueryData(channelMessagesKey(channelId))
.map((item) => item.content),
["fresh-successor", "older"],
);
await flush();
assert.equal(
queryClient.getQueryData(channelWindowKey(channelId)).stagedPage,
null,
);
});
test("next paging pass consumes the staged page without network wait", async () => {
const channelId = "consume-staged";
const { queryClient, head } = harness(channelId);

View File

@@ -5,6 +5,7 @@ import {
type ChannelWindowCursor,
type ChannelWindowPage,
type ChannelWindowStore,
cursorsEqual,
stageOlderChannelWindow,
} from "@/features/messages/lib/channelWindowStore";
import { projectChannelWindowMessages } from "@/features/messages/lib/projectChannelWindow";
@@ -15,6 +16,12 @@ 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 StagedFetch = {
cursor: ChannelWindowCursor;
fromPage: ChannelWindowPage;
page: Promise<ChannelWindowPage>;
};
const inFlightStagedPages = new Map<string, StagedFetch>();
type FetchPage = (
channelId: string,
requestCursor: ChannelWindowCursor,
@@ -53,8 +60,15 @@ async function fetchPage(
return parseChannelWindowResponse(events, channelId, requestCursor);
}
function cursorsEqual(left: ChannelWindowCursor, right: ChannelWindowCursor) {
return left.createdAt === right.createdAt && left.eventId === right.eventId;
function retainedTailIs(
queryClient: QueryClient,
channelId: string,
page: ChannelWindowPage,
) {
const store = queryClient.getQueryData<ChannelWindowStore>(
channelWindowKey(channelId),
);
return store?.pages[store.pages.length - 1] === page;
}
async function runPage(
@@ -73,11 +87,20 @@ async function runPage(
const requestCursor = tail.nextCursor;
const matchingStagedPage = store.stagedPage;
const staging = inFlightStagedPages.get(channelId);
const matchingStaging =
staging &&
retainedTailIs(queryClient, channelId, staging.fromPage) &&
cursorsEqual(staging.cursor, requestCursor)
? staging
: null;
const page =
matchingStagedPage?.startCursor &&
cursorsEqual(matchingStagedPage.startCursor, requestCursor)
? matchingStagedPage
: await fetchPageFn(channelId, requestCursor);
: matchingStaging
? await matchingStaging.page
: await fetchPageFn(channelId, requestCursor);
if (!shouldContinue()) return { hasOlderMessages: true };
const retained = queryClient.getQueryData<ChannelWindowStore>(
channelWindowKey(channelId),
@@ -90,7 +113,13 @@ async function runPage(
if (!page.hasMore || !page.nextCursor || !shouldContinue()) {
return { hasOlderMessages: false };
}
void fetchPageFn(channelId, page.nextCursor)
const stagedFetch = fetchPageFn(channelId, page.nextCursor);
inFlightStagedPages.set(channelId, {
cursor: page.nextCursor,
fromPage: page,
page: stagedFetch,
});
void stagedFetch
.then((staged) => {
if (!shouldContinue()) return;
queryClient.setQueryData<ChannelWindowStore>(
@@ -101,6 +130,11 @@ async function runPage(
})
.catch((error) => {
console.error("Failed to stage older messages", channelId, error);
})
.finally(() => {
if (inFlightStagedPages.get(channelId)?.page === stagedFetch) {
inFlightStagedPages.delete(channelId);
}
});
return { hasOlderMessages: true };
}

View File

@@ -214,7 +214,6 @@ const MessageTimelineBase = React.forwardRef<
const internalScrollRef = React.useRef<HTMLDivElement>(null);
const scrollContainerRef = externalScrollRef ?? internalScrollRef;
const contentRef = React.useRef<HTMLDivElement>(null);
const topSentinelRef = React.useRef<HTMLDivElement>(null);
const [virtualizerScrollParent, setVirtualizerScrollParent] =
React.useState<HTMLDivElement | null>(null);
const [virtualizerRenderVersion, bumpVirtualizerRenderVersion] =
@@ -768,9 +767,7 @@ const MessageTimelineBase = React.forwardRef<
)}
ref={contentRef}
>
{omitHistoryLeadIn ? null : (
<div ref={topSentinelRef} aria-hidden className="h-px" />
)}
{omitHistoryLeadIn ? null : <div aria-hidden className="h-px" />}
{/* Fixed-height history slot keeps the virtual spacer's offset
stable across load-older fetches. The intro-only state has no

View File

@@ -720,6 +720,8 @@ function VirtualizedTimelineRows({
updatePinnedDayLabel(offset);
if (offset <= list.viewportSize * 1.5) {
// Begin loading before momentum reaches the retained history boundary.
// Arming here also prevents layout-only near-top scrolls from poisoning
// the reader's next upward wheel input.
armUpwardMomentum(onStartReached?.() ?? false);
}
},

View File

@@ -251,9 +251,9 @@ test.describe("list virtualization", () => {
// reproduce Chromium/WebKit's native wheel → scroll callback ordering. The
// old boundary rollback moved the viewport back down before the fetch
// committed; keep that pre-prepend reversal below the same 5px frame bar.
// A 300ms relay delay leaves the input boundary and prepend commit as two
// distinct phases so this assertion cannot accidentally measure only the
// later anchor correction.
// Start outside the history lookahead and let native wheel input cross it.
// This preserves the pre-prepend rollback measurement whether the page is
// already staged locally or still incurs the configured relay delay.
await installMockBridge(page, {
deepHistoryMessageCount: 1_800,
channelWindowDelayMs: 300,
@@ -284,6 +284,7 @@ test.describe("list virtualization", () => {
id: row.dataset.messageId ?? "",
top: row.getBoundingClientRect().top - scrollerTop,
scrollHeight: s.scrollHeight,
clientHeight: s.clientHeight,
bottomDistance: s.scrollHeight - s.clientHeight - s.scrollTop,
};
}
@@ -306,7 +307,7 @@ test.describe("list virtualization", () => {
});
await page.waitForTimeout(300);
await timeline.evaluate((element) => {
element.scrollTop = 180;
element.scrollTop = element.clientHeight * 2;
});
await page.waitForTimeout(150);
const before = await sampleVisibleAnchor();
@@ -353,12 +354,14 @@ test.describe("list virtualization", () => {
const box = await timeline.boundingBox();
if (!box) throw new Error("timeline has no bounding box");
await page.mouse.move(box.x + box.width / 2, box.y + box.height / 2);
for (const deltaY of [-60, -30, -20, -15]) {
for (const deltaY of [-before.clientHeight * 0.75]) {
await page.mouse.wheel(0, deltaY);
await page.waitForTimeout(12);
}
const wheelTrace = await wheelTracePromise;
expect(wheelTrace.minScrollTop).toBeLessThanOrEqual(350);
expect(wheelTrace.minScrollTop).toBeLessThanOrEqual(
before.clientHeight * 1.5,
);
expect(wheelTrace.maxBoundaryRollback).toBeLessThan(5);
// Linux Chromium delivers CDP wheel input with more latency than macOS,
// so the burst's final delta can land AFTER the anchor baseline sample