From a8b11319239c62ae58e9f0de0c70ff1414e5eff7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:52:06 +0000 Subject: [PATCH 01/11] fix(provider-review): page the Bitbucket inbox with a signed continuation cursor Each inbox load reads one provider page per enumerated repository and continues from the `updated_on` boundary of the last row served. The continuation cursor is an HMAC-signed token bound to the workspace, and a load without usable continuation state starts from the first page. The merge keeps a bounded newest-first window, and the repository fan-out is capped by the same per-load request budget. --- .../provider-review/bitbucket-read.test.ts | 343 ++++++++++-------- .../src/lib/provider-review/bitbucket-read.ts | 253 ++++++++----- 2 files changed, 343 insertions(+), 253 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 1b972bb0d7c..66d8c58a433 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1,4 +1,7 @@ import { describe, expect, it, beforeEach, afterEach } from '@jest/globals'; +import crypto from 'node:crypto'; +import { NEXTAUTH_SECRET } from '@kilocode/web-shared/lib/config.server'; +import { createSignedToken } from '@/lib/signed-token'; import { getMergeRestrictions, getPullRequest, @@ -24,6 +27,7 @@ jest.mock('@/lib/integrations/platforms/bitbucket/workspace-access-token-reposit jest.mock('@kilocode/web-shared/lib/config.server', () => ({ GIT_TOKEN_SERVICE_API_URL: 'https://token-service.example.com', + NEXTAUTH_SECRET: 'bitbucket-inbox-test-secret', })); jest.mock('@kilocode/web-shared/lib/tokens', () => ({ @@ -943,6 +947,50 @@ describe('listInbox', () => { destination: { branch: { name: 'main' }, repository: { full_name: fullName } }, }); + type InboxRow = { updated_on: string }; + type InboxRequest = { slug: string; query: string | null; page: string | null }; + + /** + * The workspace inbox as the provider serves it: each repository answers + * with its rows at or below the requested boundary, newest first, one page. + */ + const serveInbox = (rowsBySlug: Record, requests: InboxRequest[]) => { + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + if (parsed.pathname === '/2.0/repositories/acme') { + return jsonResponse({ + pagelen: 100, + values: Object.keys(rowsBySlug).map(slug => ({ slug })), + next: null, + }); + } + const match = parsed.pathname.match(/^\/2\.0\/repositories\/acme\/([^/]+)\/pullrequests$/); + const rows = match ? rowsBySlug[match[1]] : undefined; + if (!match || !rows) return jsonResponse({ pagelen: 50, values: [], next: null }); + const query = parsed.searchParams.get('q'); + requests.push({ slug: match[1], query, page: parsed.searchParams.get('page') }); + const boundary = query?.match(/^state="OPEN" AND updated_on <= (.+)$/); + const boundaryMs = boundary ? Date.parse(boundary[1]) : null; + const eligible = rows + .filter(row => boundaryMs === null || Date.parse(row.updated_on) <= boundaryMs) + .sort((left, right) => Date.parse(right.updated_on) - Date.parse(left.updated_on)); + return jsonResponse({ + pagelen: 50, + values: eligible.slice(0, 50), + next: eligible.length > 50 ? 'https://api.bitbucket.org/2.0/next?page=2' : null, + }); + }); + }; + + const minutesFixture = (firstId: number, count: number) => + Array.from({ length: count }, (_, index) => + inboxPr(firstId + index, `2026-09-02T00:${String(index).padStart(2, '0')}:00+00:00`) + ); + it('fans out over the workspace repositories and carries full identity', async () => { fetchMock.mockImplementation(async (url: string | URL) => { const full = url.toString(); @@ -1004,193 +1052,182 @@ describe('listInbox', () => { expect(result.nextCursor).toBeNull(); }); - it('merges pages across repositories newest first and continues with a page cursor', async () => { - fetchMock.mockImplementation(async (url: string | URL) => { - const full = url.toString(); - if (full.includes('token-service.example.com')) { - return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); - } - const parsed = new URL(full); - if (parsed.pathname === '/2.0/repositories/acme') { - return jsonResponse({ pagelen: 100, values: [{ slug: 'repo' }], next: null }); - } - if (parsed.pathname === '/2.0/repositories/acme/repo/pullrequests') { - // The provider listing is sorted newest first (the inbox request pins - // sort=-updated_on): page 2 holds only older rows. - const page = parsed.searchParams.get('page'); - if (page === '2') { - return jsonResponse({ - pagelen: 50, - values: [inboxPr(99, '2026-09-01T00:00:00.000000+00:00')], - next: null, - }); - } - return jsonResponse({ - pagelen: 50, - values: Array.from({ length: 50 }, (_, index) => - inboxPr(100 + index, `2026-09-02T00:00:${String(index).padStart(2, '0')}+00:00`) - ), - next: null, - }); - } - return jsonResponse({ pagelen: 50, values: [], next: null }); - }); + it('continues one repository from the signed boundary', async () => { + // 60 open PRs: the first load serves the newest 50 and hands back a signed + // boundary; the next load asks only for rows at or below that boundary. + const requests: InboxRequest[] = []; + serveInbox({ repo: minutesFixture(100, 60) }, requests); const first = await listInbox(ORG_OWNER); expect(first.items).toHaveLength(50); - expect(first.items[0]?.ref).toMatchObject({ prId: 149 }); + expect(first.items[0]?.ref).toMatchObject({ prId: 159 }); expect(first.nextCursor).toBeTruthy(); + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ slug: 'repo', query: 'state="OPEN"', page: '1' }); const second = await listInbox(ORG_OWNER, first.nextCursor!); - // The second page continues the sorted sequence with the rows the first - // page's trim could not serve, instead of re-sorting a fresh fan-out. - expect(second.items.map(item => item.ref)).toEqual([expect.objectContaining({ prId: 99 })]); + // One provider page for the continuation too, carrying the boundary of + // the last row already served. + expect(requests).toHaveLength(2); + expect(requests[1]).toMatchObject({ + slug: 'repo', + query: 'state="OPEN" AND updated_on <= 2026-09-02T00:10:00.000+00:00', + page: '1', + }); + expect(second.items).toHaveLength(10); + expect(second.items[0]?.ref).toMatchObject({ prId: 109 }); expect(second.nextCursor).toBeNull(); + + const served = [...first.items, ...second.items].flatMap(item => + item.ref && 'prId' in item.ref ? [item.ref.prId] : [] + ); + expect(new Set(served).size).toBe(60); }); - it('serves every row across pages: the sorted window refetches earlier provider pages', async () => { - // 60 PRs in one repository: a full first provider page (50) plus a - // 10-row older tail. Page 1 serves the newest 50; page 2 must serve the - // remaining 10 — none dropped, none duplicated. - fetchMock.mockImplementation(async (url: string | URL) => { - const full = url.toString(); - if (full.includes('token-service.example.com')) { - return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); - } - const parsed = new URL(full); - if (parsed.pathname === '/2.0/repositories/acme') { - return jsonResponse({ pagelen: 100, values: [{ slug: 'repo' }], next: null }); - } - if (parsed.pathname === '/2.0/repositories/acme/repo/pullrequests') { - const page = parsed.searchParams.get('page'); - if (page === '2') { - return jsonResponse({ - pagelen: 50, - values: Array.from({ length: 10 }, (_, index) => - inboxPr(150 + index, `2026-09-01T00:00:${String(index).padStart(2, '0')}+00:00`) - ), - next: null, - }); - } - return jsonResponse({ - pagelen: 50, - values: Array.from({ length: 50 }, (_, index) => - inboxPr(100 + index, `2026-09-02T00:00:${String(index).padStart(2, '0')}+00:00`) - ), - next: null, - }); - } - return jsonResponse({ pagelen: 50, values: [], next: null }); - }); + it('continues several repositories across pages', async () => { + // repo-a's rows are all older than repo-b's, so the first load serves + // exactly repo-b's page and the continuation serves repo-a's. + const requests: InboxRequest[] = []; + serveInbox( + { + 'repo-a': minutesFixture(100, 50), + 'repo-b': Array.from({ length: 50 }, (_, index) => + inboxPr(200 + index, `2026-09-02T01:${String(index).padStart(2, '0')}:00+00:00`) + ), + }, + requests + ); const first = await listInbox(ORG_OWNER); expect(first.items).toHaveLength(50); - expect(first.items[0]?.ref).toMatchObject({ prId: 149 }); + expect(first.items[0]?.ref).toMatchObject({ prId: 249 }); + expect(first.nextCursor).toBeTruthy(); const second = await listInbox(ORG_OWNER, first.nextCursor!); - expect(second.items).toHaveLength(10); - expect(second.items[0]?.ref).toMatchObject({ prId: 159 }); + expect(second.items).toHaveLength(50); + expect(second.items[0]?.ref).toMatchObject({ prId: 149 }); expect(second.nextCursor).toBeNull(); + expect(requests).toHaveLength(4); - const servedIds = new Set( - [...first.items, ...second.items].map(item => - item.ref && 'prId' in item.ref ? item.ref.prId : null - ) + const served = [...first.items, ...second.items].flatMap(item => + item.ref && 'prId' in item.ref ? [item.ref.prId] : [] ); - expect(servedIds.size).toBe(60); - for (let id = 100; id <= 159; id += 1) { - expect(servedIds.has(id)).toBe(true); - } + expect(new Set(served).size).toBe(100); }); - it('serves every row of several repositories across pages without drops', async () => { - // Two repositories with 50 open PRs each: page 1 serves the newest 50 of - // the 100 aggregated rows, page 2 the remaining 50 — the old page-N - // fan-out re-fetched provider page 2 (empty here) and dropped 50 rows. - fetchMock.mockImplementation(async (url: string | URL) => { - const full = url.toString(); - if (full.includes('token-service.example.com')) { - return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); - } - const parsed = new URL(full); - if (parsed.pathname === '/2.0/repositories/acme') { - return jsonResponse({ - pagelen: 100, - values: [{ slug: 'repo-a' }, { slug: 'repo-b' }], - next: null, - }); - } - const match = parsed.pathname.match( - /^\/2\.0\/repositories\/acme\/(repo-[ab])\/pullrequests$/ + it('reads one provider page per repository on every inbox page', async () => { + // 12 repositories × 5 open PRs: every load reads one page per repository. + const rowsBySlug: Record = {}; + for (let repo = 0; repo < 12; repo += 1) { + rowsBySlug[`repo-${repo}`] = Array.from({ length: 5 }, (_, index) => + inboxPr( + 1000 + repo * 10 + index, + `2026-09-02T00:${String(repo * 5 + index).padStart(2, '0')}:00+00:00` + ) ); - if (match && parsed.searchParams.get('page') === '1') { - // repo-a rows are older than repo-b rows, so the first inbox page is - // exactly repo-b's page; the single provider page holds all 50 rows. - const base = match[1] === 'repo-a' ? 100 : 200; - const hour = match[1] === 'repo-a' ? '00' : '01'; - return jsonResponse({ - pagelen: 50, - values: Array.from({ length: 50 }, (_, index) => - inboxPr( - base + index, - `2026-09-02T${hour}:00:${String(index).padStart(2, '0')}.000000+00:00` - ) - ), - next: null, - }); - } - return jsonResponse({ pagelen: 50, values: [], next: null }); - }); + } + const requests: InboxRequest[] = []; + serveInbox(rowsBySlug, requests); const first = await listInbox(ORG_OWNER); expect(first.items).toHaveLength(50); - expect(first.items[0]?.ref).toMatchObject({ prId: 249 }); expect(first.nextCursor).toBeTruthy(); + expect(requests).toHaveLength(12); const second = await listInbox(ORG_OWNER, first.nextCursor!); - expect(second.items).toHaveLength(50); - expect(second.items[0]?.ref).toMatchObject({ prId: 149 }); + expect(requests).toHaveLength(24); + expect(second.items).toHaveLength(10); expect(second.nextCursor).toBeNull(); + expect(requests.every(request => request.page === '1')).toBe(true); + expect(requests.filter(request => request.query === 'state="OPEN"')).toHaveLength(12); + }); - const servedIds = new Set( - [...first.items, ...second.items].map(item => - item.ref && 'prId' in item.ref ? item.ref.prId : null - ) - ); - expect(servedIds.size).toBe(100); - for (let id = 100; id <= 249; id += 1) { - if ((id >= 100 && id <= 149) || (id >= 200 && id <= 249)) { - expect(servedIds.has(id)).toBe(true); - } + it('reads one provider page per enumerated repository', async () => { + // The enumeration cap is 300 repositories; a load reads at most one page + // from each of them. + const rowsBySlug: Record = {}; + for (let repo = 0; repo < 300; repo += 1) { + rowsBySlug[`repo-${repo}`] = [ + inboxPr(repo, `2026-09-02T00:${String(repo % 60).padStart(2, '0')}:00+00:00`), + ]; } + const requests: InboxRequest[] = []; + serveInbox(rowsBySlug, requests); + + const page = await listInbox(ORG_OWNER); + expect(requests).toHaveLength(300); + expect(page.items).toHaveLength(50); + expect(page.nextCursor).toBeTruthy(); }); - it('ignores a cursor minted for another workspace', async () => { - fetchMock.mockImplementation(async (url: string | URL) => { - const full = url.toString(); - if (full.includes('token-service.example.com')) { - return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); - } - const parsed = new URL(full); - if (parsed.pathname === '/2.0/repositories/acme') { - return jsonResponse({ pagelen: 100, values: [{ slug: 'repo' }], next: null }); - } - if (parsed.pathname === '/2.0/repositories/acme/repo/pullrequests') { - expect(parsed.searchParams.get('page')).toBe('1'); - return jsonResponse({ - pagelen: 50, - values: [inboxPr(12, '2026-09-02T00:00:00.000000+00:00')], - next: null, - }); - } - return jsonResponse({ pagelen: 50, values: [], next: null }); + it('starts again at the first page when continuation state is not valid', async () => { + const requests: InboxRequest[] = []; + serveInbox({ repo: minutesFixture(100, 60) }, requests); + + const clientValue = Buffer.from( + JSON.stringify({ identity: 'bitbucket-inbox:acme', page: 10 }) + ).toString('base64url'); + const page = await listInbox(ORG_OWNER, clientValue); + + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ query: 'state="OPEN"', page: '1' }); + expect(page.items).toHaveLength(50); + }); + + it('scopes continuation state to the workspace it was issued for', async () => { + const requests: InboxRequest[] = []; + serveInbox({ repo: minutesFixture(100, 60) }, requests); + + const foreign = createSignedToken({ + kind: 'bitbucket-inbox-cursor', + identity: 'bitbucket-inbox:evil', + boundary: '2026-09-02T00:10:00.000+00:00', + servedRefs: [], }); + const page = await listInbox(ORG_OWNER, foreign); - const foreign = Buffer.from( - JSON.stringify({ identity: 'bitbucket-inbox:evil', page: 7 }) + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ query: 'state="OPEN"' }); + expect(page.items).toHaveLength(50); + }); + + it('accepts continuation state only with a matching signature', async () => { + const requests: InboxRequest[] = []; + serveInbox({ repo: minutesFixture(100, 60) }, requests); + const first = await listInbox(ORG_OWNER); + const cursor = first.nextCursor!; + requests.length = 0; + + const invalidSignature = `${cursor.slice(0, -1)}${cursor.endsWith('A') ? 'B' : 'A'}`; + await listInbox(ORG_OWNER, invalidSignature); + + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ query: 'state="OPEN"', page: '1' }); + }); + + it('keeps continuation state usable for one hour', async () => { + const requests: InboxRequest[] = []; + serveInbox({ repo: minutesFixture(100, 60) }, requests); + + const encoded = Buffer.from( + JSON.stringify({ + kind: 'bitbucket-inbox-cursor', + identity: 'bitbucket-inbox:acme', + boundary: '2026-09-02T00:10:00.000+00:00', + servedRefs: [], + iat: Math.floor(Date.now() / 1000) - 2 * 60 * 60, + nonce: 'expired-nonce', + }) ).toString('base64url'); - const result = await listInbox(ORG_OWNER, foreign); - expect(result.items).toHaveLength(1); + const signature = crypto + .createHmac('sha256', NEXTAUTH_SECRET) + .update(encoded) + .digest('base64url'); + const page = await listInbox(ORG_OWNER, `${encoded}.${signature}`); + + expect(requests).toHaveLength(1); + expect(requests[0]).toMatchObject({ query: 'state="OPEN"' }); + expect(page.items).toHaveLength(50); }); }); diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 040f7272c6c..c2a9da83d3e 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -12,16 +12,17 @@ import 'server-only'; import { z } from 'zod'; -import type { - ProviderPrChecksResult, - ProviderPrFile, - ProviderPrFilesPage, - ProviderPrInboxItem, - ProviderPrInboxPage, - ProviderPrMergeBlockedReason, - ProviderPrMergeState, - ProviderPrSummary, - ProviderPrThread, +import { + providerPrRefKey, + type ProviderPrChecksResult, + type ProviderPrFile, + type ProviderPrFilesPage, + type ProviderPrInboxItem, + type ProviderPrInboxPage, + type ProviderPrMergeBlockedReason, + type ProviderPrMergeState, + type ProviderPrSummary, + type ProviderPrThread, } from '@kilocode/app-shared/provider-review'; import { authorizeRepository, @@ -33,6 +34,7 @@ import { type BitbucketReviewOwner, } from './bitbucket-authorization'; import { FILE_LINES_MAX } from '@/lib/github-pr-review/dtos'; +import { createSignedToken, verifySignedToken } from '@/lib/signed-token'; const BITBUCKET_API_ORIGIN = 'https://api.bitbucket.org'; const BITBUCKET_PAGE_SIZE = 50; @@ -50,15 +52,18 @@ const MAX_BUILD_PAGES = 3; /** The inbox enumerates at most this many pages of this size of workspace repositories. */ const INBOX_REPOSITORY_PAGE_SIZE = 100; const INBOX_REPOSITORY_PAGES = 3; +/** The repositories one inbox load enumerates: the enumeration cap. */ +const INBOX_MAX_REPOSITORIES = INBOX_REPOSITORY_PAGE_SIZE * INBOX_REPOSITORY_PAGES; /** How many repository PR collections the inbox fetches at once. */ const INBOX_REPOSITORY_CONCURRENCY = 8; /** - * The provider pages one inbox page may walk per repository: inbox page N - * refetches provider pages 1..N so the sorted windows continue each other, - * and this bound keeps that fan-out capped (a bounded inbox beats an - * unbounded crawl). + * The provider-request budget of one inbox load: the repository enumeration + * plus one PR page per repository. An inbox load continues from a signed + * `updated_on` boundary, so this aggregate bound holds for every inbox page. */ -const MAX_INBOX_PROVIDER_PAGES = 10; +const MAX_INBOX_PROVIDER_REQUESTS = INBOX_REPOSITORY_PAGES + INBOX_MAX_REPOSITORIES; +/** How long a server-issued inbox cursor stays usable. */ +const INBOX_CURSOR_TTL_SECONDS = 60 * 60; /** * The task-collection page bound for the discussion task walk: the same * bounded walk the write layer's thread resolution uses, so one discussion @@ -951,12 +956,13 @@ export async function listChecks( * PR of the workspace's repositories, newest first. * * The fan-out sorts across repositories, so one inbox page cannot map onto a - * single provider page: page N refetches provider pages 1..N of every - * repository (each listing is pinned to `sort=-updated_on`, the inbox's own - * sort key) and serves the sorted window [(N-1)·size, N·size). Refetching - * the earlier provider pages is what keeps the windows from drifting: a - * plain provider-page-N fan-out plus sort-and-trim would silently drop the - * rows page 1's trim left over, and they would never be re-served. + * single provider page. A later page continues from a signed `updated_on` + * boundary — the timestamp of the last row already served — and every + * repository answers with one page of its rows at or below that boundary. The + * newest BITBUCKET_PAGE_SIZE rows of those per-repository pages are exactly + * the next slice of the merged order (the global top-K of a union is + * contained in the union of the per-repository top-K), so one inbox load + * reads one provider page per repository. */ export async function listInbox( owner: BitbucketReviewOwner, @@ -965,73 +971,67 @@ export async function listInbox( const access = await authorizeWorkspace(owner); try { const identity = `bitbucket-inbox:${access.workspace.slug}`; - const page = decodeInboxPageCursor(cursor, identity); - if (page > MAX_INBOX_PROVIDER_PAGES) { - // Past the walk bound the pagination ends instead of crawling. - return { items: [], nextCursor: null }; - } - const slugs = await listWorkspaceRepositorySlugs(access, access.workspace.slug); - const values: unknown[] = []; - let deeperProviderPage = false; - for (let providerPage = 1; providerPage <= page; providerPage += 1) { - let lastRequestedPageFull = false; - for (let offset = 0; offset < slugs.length; offset += INBOX_REPOSITORY_CONCURRENCY) { - const batch = slugs.slice(offset, offset + INBOX_REPOSITORY_CONCURRENCY); - const pages = await Promise.all( - batch.map(slug => - requestBitbucketJson( - access, - `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(slug)}/pullrequests`, - { - query: { - pagelen: BITBUCKET_PAGE_SIZE, - page: providerPage, - q: 'state="OPEN"', - sort: '-updated_on', - }, - } - ) + const position = decodeInboxCursor(cursor, identity); + // The enumeration counts toward the same per-load budget as the fan-out. + const repositoryBudget = MAX_INBOX_PROVIDER_REQUESTS - INBOX_REPOSITORY_PAGES; + const slugs = (await listWorkspaceRepositorySlugs(access, access.workspace.slug)).slice( + 0, + repositoryBudget + ); + const providerQuery = position + ? `state="OPEN" AND updated_on <= ${position.boundary}` + : 'state="OPEN"'; + let repositoryPageFull = false; + let window: ProviderPrInboxItem[] = []; + let beyondWindow = false; + for (let offset = 0; offset < slugs.length; offset += INBOX_REPOSITORY_CONCURRENCY) { + const batch = slugs.slice(offset, offset + INBOX_REPOSITORY_CONCURRENCY); + const pages = await Promise.all( + batch.map(slug => + requestBitbucketJson( + access, + `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(slug)}/pullrequests`, + { + query: { + pagelen: BITBUCKET_PAGE_SIZE, + page: 1, + q: providerQuery, + sort: '-updated_on', + }, + } ) - ); - for (const payload of pages) { - const parsedPage = BitbucketPageSchema.safeParse(payload); - if (!parsedPage.success) { - throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); - } - values.push(...parsedPage.data.values); - if (parsedPage.data.values.length >= BITBUCKET_PAGE_SIZE) lastRequestedPageFull = true; + ) + ); + for (const payload of pages) { + const parsedPage = BitbucketPageSchema.safeParse(payload); + if (!parsedPage.success) { + throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); } + if (parsedPage.data.next) repositoryPageFull = true; + const pageItems = inboxItemsFrom(parsedPage.data.values, access.workspace.slug, position); + const merged = [...window, ...pageItems].sort( + (left, right) => inboxUpdatedMs(right) - inboxUpdatedMs(left) + ); + if (merged.length > BITBUCKET_PAGE_SIZE) beyondWindow = true; + window = merged.slice(0, BITBUCKET_PAGE_SIZE); } - if (providerPage === page) deeperProviderPage = lastRequestedPageFull; } - const items: ProviderPrInboxItem[] = []; - for (const value of values) { - const parsed = BitbucketInboxPullRequestSchema.safeParse(value); - if (!parsed.success) continue; - const ref = inboxRefFrom(parsed.data, access.workspace.slug); - if (!ref) continue; - items.push({ - ref, - title: parsed.data.title, - author: mapUser(parsed.data.author ?? null), - state: mapPullRequestState(parsed.data.state), - draft: parsed.data.draft === true, - updatedAt: parsed.data.updated_on ?? '', - }); - } - items.sort((left, right) => inboxUpdatedMs(right) - inboxUpdatedMs(left)); - const skip = (page - 1) * BITBUCKET_PAGE_SIZE; - const window = items.slice(skip, skip + BITBUCKET_PAGE_SIZE); - // More follows when a repository still holds a page after the deepest - // one read, or when the sorted aggregate itself already reaches past - // this window. - const hasMore = deeperProviderPage || items.length > skip + window.length; + // More follows when a repository page came back full — its collection + // continues below the boundary — or when the merged order already reaches + // past this window. + const hasMore = repositoryPageFull || beyondWindow; + const boundary = window.length > 0 ? inboxBoundaryValue(window[window.length - 1]) : null; + // The rows at the boundary value are recorded by ref alongside it. + const servedRefs = + boundary === null + ? [] + : window + .filter(item => inboxBoundaryValue(item) === boundary) + .map(item => providerPrRefKey(item.ref)); return { items: window, nextCursor: - hasMore && page < MAX_INBOX_PROVIDER_PAGES - ? encodeInboxPageCursor(identity, page + 1) - : null, + hasMore && boundary !== null ? encodeInboxCursor(identity, boundary, servedRefs) : null, }; } catch (error) { throw classifyBitbucketError(error); @@ -1044,29 +1044,82 @@ function inboxUpdatedMs(item: ProviderPrInboxItem): number { } /** - * The inbox cursor is a plain page counter, not a provider `next` URL: one - * inbox page fans out over the workspace's repositories, so no single next - * link can represent it. The page number is the sorted window index — page N - * serves rows [(N-1)·size, N·size) of the merged newest-first order. A - * cursor minted for another workspace, or in the old next-URL shape, decodes - * to page 1 — a cursor can never switch the workspace a request reads. + * The provider filter value of a row's `updated_on`: millisecond ISO-8601 in + * UTC, the form the provider's query language documents. A row without a + * parseable timestamp has no boundary and can only be served last. */ -function encodeInboxPageCursor(identity: string, page: number): string { - return Buffer.from(JSON.stringify({ identity, page })).toString('base64url'); +function inboxBoundaryValue(item: ProviderPrInboxItem): string | null { + const ms = Date.parse(item.updatedAt); + if (Number.isNaN(ms)) return null; + return new Date(ms).toISOString().replace('Z', '+00:00'); } -function decodeInboxPageCursor(cursor: string | undefined, identity: string): number { - if (!cursor) return 1; - try { - const parsed = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as { - identity?: unknown; - page?: unknown; - }; - if (parsed.identity !== identity || !Number.isInteger(parsed.page)) return 1; - return Math.max(1, parsed.page as number); - } catch { - return 1; +/** The inbox rows of one provider page, minus the rows an earlier page served. */ +function inboxItemsFrom( + values: unknown[], + workspaceSlug: string, + position: InboxCursorPosition | null +): ProviderPrInboxItem[] { + const items: ProviderPrInboxItem[] = []; + for (const value of values) { + const parsed = BitbucketInboxPullRequestSchema.safeParse(value); + if (!parsed.success) continue; + const ref = inboxRefFrom(parsed.data, workspaceSlug); + if (!ref) continue; + if (position?.servedRefs.has(providerPrRefKey(ref))) continue; + items.push({ + ref, + title: parsed.data.title, + author: mapUser(parsed.data.author ?? null), + state: mapPullRequestState(parsed.data.state), + draft: parsed.data.draft === true, + updatedAt: parsed.data.updated_on ?? '', + }); } + return items; +} + +const INBOX_CURSOR_KIND = 'bitbucket-inbox-cursor'; + +const inboxCursorPayloadSchema = z.object({ + kind: z.literal(INBOX_CURSOR_KIND), + identity: z.string().min(1), + boundary: z.string().min(1), + servedRefs: z.array(z.string().min(1)).max(BITBUCKET_PAGE_SIZE), +}); + +type InboxCursorPosition = { boundary: string; servedRefs: ReadonlySet }; + +/** + * The inbox cursor is a signed continuation token: it carries the identity it + * was minted for and the `updated_on` boundary of the last row served, and it + * stays usable for INBOX_CURSOR_TTL_SECONDS. A load without usable + * continuation state starts from the first page. + */ +function encodeInboxCursor(identity: string, boundary: string, servedRefs: string[]): string { + return createSignedToken({ + kind: INBOX_CURSOR_KIND, + identity, + boundary, + servedRefs: servedRefs.slice(0, BITBUCKET_PAGE_SIZE), + }); +} + +function decodeInboxCursor( + cursor: string | undefined, + identity: string +): InboxCursorPosition | null { + if (!cursor) return null; + const payload = verifySignedToken(cursor, { + ttlSeconds: INBOX_CURSOR_TTL_SECONDS, + parse: value => { + const parsed = inboxCursorPayloadSchema.safeParse(value); + if (!parsed.success || parsed.data.identity !== identity) return null; + return { boundary: parsed.data.boundary, servedRefs: parsed.data.servedRefs }; + }, + }); + if (payload === null) return null; + return { boundary: payload.boundary, servedRefs: new Set(payload.servedRefs) }; } /** From 56194759ea47f8d6207dabc10c6c7799d33b2916 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Tue, 20 Oct 2026 02:09:45 +0000 Subject: [PATCH 02/11] fix(provider-review): keep the inbox boundary at provider precision The continuation cursor carries the provider's own `updated_on` string, so a later inbox page asks for rows at or below that exact value. The merge orders rows by the same sub-millisecond value, and the boundary comes from the last row in the window that has a parseable timestamp. --- .../provider-review/bitbucket-read.test.ts | 50 +++++++++++++++++-- .../src/lib/provider-review/bitbucket-read.ts | 22 +++++--- 2 files changed, 60 insertions(+), 12 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 66d8c58a433..1fdf5cf5c31 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -950,6 +950,13 @@ describe('listInbox', () => { type InboxRow = { updated_on: string }; type InboxRequest = { slug: string; query: string | null; page: string | null }; + /** A provider timestamp in milliseconds, including sub-millisecond digits. */ + const inboxTime = (value: string) => { + const ms = Date.parse(value); + const fraction = /\.(\d+)/.exec(value)?.[1] ?? ''; + return fraction.length > 3 ? ms + Number(fraction.slice(3)) / 10 ** (fraction.length - 3) : ms; + }; + /** * The workspace inbox as the provider serves it: each repository answers * with its rows at or below the requested boundary, newest first, one page. @@ -974,10 +981,10 @@ describe('listInbox', () => { const query = parsed.searchParams.get('q'); requests.push({ slug: match[1], query, page: parsed.searchParams.get('page') }); const boundary = query?.match(/^state="OPEN" AND updated_on <= (.+)$/); - const boundaryMs = boundary ? Date.parse(boundary[1]) : null; + const boundaryMs = boundary ? inboxTime(boundary[1]) : null; const eligible = rows - .filter(row => boundaryMs === null || Date.parse(row.updated_on) <= boundaryMs) - .sort((left, right) => Date.parse(right.updated_on) - Date.parse(left.updated_on)); + .filter(row => boundaryMs === null || inboxTime(row.updated_on) <= boundaryMs) + .sort((left, right) => inboxTime(right.updated_on) - inboxTime(left.updated_on)); return jsonResponse({ pagelen: 50, values: eligible.slice(0, 50), @@ -1071,7 +1078,7 @@ describe('listInbox', () => { expect(requests).toHaveLength(2); expect(requests[1]).toMatchObject({ slug: 'repo', - query: 'state="OPEN" AND updated_on <= 2026-09-02T00:10:00.000+00:00', + query: 'state="OPEN" AND updated_on <= 2026-09-02T00:10:00+00:00', page: '1', }); expect(second.items).toHaveLength(10); @@ -1084,6 +1091,41 @@ describe('listInbox', () => { expect(new Set(served).size).toBe(60); }); + it('continues past a boundary that carries microseconds', async () => { + // The window ends on a timestamp inside the second's millisecond group; the + // row in the same millisecond from another repository still follows. + const requests: InboxRequest[] = []; + serveInbox( + { + 'repo-a': [ + inboxPr(300, '2026-09-02T00:00:02.400902+00:00', 'acme/repo-a'), + inboxPr(301, '2026-09-02T00:00:02.400500+00:00', 'acme/repo-a'), + ], + 'repo-b': Array.from({ length: 49 }, (_, index) => + inboxPr(400 + index, `2026-09-02T00:02:${String(index).padStart(2, '0')}.900100+00:00`) + ), + }, + requests + ); + + const first = await listInbox(ORG_OWNER); + expect(first.items).toHaveLength(50); + expect(first.items[49]?.ref).toMatchObject({ prId: 300 }); + + const second = await listInbox(ORG_OWNER, first.nextCursor!); + expect(requests).toHaveLength(4); + expect( + requests.filter( + request => + request.query === 'state="OPEN" AND updated_on <= 2026-09-02T00:00:02.400902+00:00' + ) + ).toHaveLength(2); + expect( + second.items.map(item => (item.ref && 'prId' in item.ref ? item.ref.prId : null)) + ).toEqual([301]); + expect(second.nextCursor).toBeNull(); + }); + it('continues several repositories across pages', async () => { // repo-a's rows are all older than repo-b's, so the first load serves // exactly repo-b's page and the continuation serves repo-a's. diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index c2a9da83d3e..9b3fe73d7d7 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -1020,7 +1020,11 @@ export async function listInbox( // continues below the boundary — or when the merged order already reaches // past this window. const hasMore = repositoryPageFull || beyondWindow; - const boundary = window.length > 0 ? inboxBoundaryValue(window[window.length - 1]) : null; + let boundary: string | null = null; + for (let index = window.length - 1; index >= 0; index -= 1) { + boundary = inboxBoundaryValue(window[index]); + if (boundary !== null) break; + } // The rows at the boundary value are recorded by ref alongside it. const servedRefs = boundary === null @@ -1038,20 +1042,22 @@ export async function listInbox( } } +/** The row's `updated_on` in milliseconds, including any sub-millisecond digits. */ function inboxUpdatedMs(item: ProviderPrInboxItem): number { const ms = Date.parse(item.updatedAt); - return Number.isNaN(ms) ? 0 : ms; + if (Number.isNaN(ms)) return 0; + const fraction = /\.(\d+)/.exec(item.updatedAt)?.[1] ?? ''; + return fraction.length > 3 ? ms + Number(fraction.slice(3)) / 10 ** (fraction.length - 3) : ms; } /** - * The provider filter value of a row's `updated_on`: millisecond ISO-8601 in - * UTC, the form the provider's query language documents. A row without a - * parseable timestamp has no boundary and can only be served last. + * The provider filter value of a row's `updated_on`: the provider's own + * timestamp string, so its full precision carries into the continuation. A + * row without a parseable timestamp has no boundary. */ function inboxBoundaryValue(item: ProviderPrInboxItem): string | null { - const ms = Date.parse(item.updatedAt); - if (Number.isNaN(ms)) return null; - return new Date(ms).toISOString().replace('Z', '+00:00'); + if (Number.isNaN(Date.parse(item.updatedAt))) return null; + return item.updatedAt; } /** The inbox rows of one provider page, minus the rows an earlier page served. */ From f46f4de838175a31877147dff6336b41bea96497 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 04:35:47 +0000 Subject: [PATCH 03/11] fix(provider-review): advance the Bitbucket inbox past tied provider pages Every inbox load reads one provider page per integration repository. A continuation now advances a repository's provider page when the page holds only rows already served at the continuation boundary, so rows that share a timestamp across provider pages stay reachable. The merged order breaks timestamp ties by repository then pull-request id, so the continuation carries only the last served ref at its boundary instead of every ref at that boundary, and the per-repository page positions are packed two bytes each. A three-hundred-repository continuation is checked against the provider-review router's page-cursor limit. --- .../provider-review/bitbucket-read.test.ts | 92 ++++++++++++-- .../src/lib/provider-review/bitbucket-read.ts | 119 +++++++++++------- 2 files changed, 157 insertions(+), 54 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 4530551d173..bb676448496 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -11,6 +11,7 @@ import { listChecks, listDiscussions, listInbox, + INBOX_CURSOR_MAX_CHARS, } from './bitbucket-read'; import { BitbucketReviewError } from './bitbucket-authorization'; import { FILE_LINES_MAX } from '@/lib/github-pr-review/dtos'; @@ -1150,6 +1151,41 @@ describe('listInbox', () => { expect(third.nextCursor).toBeNull(); expect(requests.map(request => request.page)).toEqual(['1', '1', '2']); }); + it('keeps equal-time rows in a stable order across inbox pages', async () => { + const requests: InboxRequest[] = []; + const tiedAt = '2026-09-02T00:00:00.000000+00:00'; + serveInbox( + { + 'repo-a': Array.from({ length: 40 }, (_, index) => + inboxPr(1000 + index, tiedAt, 'acme/repo-a') + ), + 'repo-b': Array.from({ length: 40 }, (_, index) => + inboxPr(2000 + index, tiedAt, 'acme/repo-b') + ), + 'repo-c': Array.from({ length: 40 }, (_, index) => + inboxPr(3000 + index, tiedAt, 'acme/repo-c') + ), + }, + requests + ); + + const first = await listInbox(ORG_OWNER); + const second = await listInbox(ORG_OWNER, first.nextCursor!); + const third = await listInbox(ORG_OWNER, second.nextCursor!); + const ids = [...first.items, ...second.items, ...third.items].map(item => + item.ref && 'prId' in item.ref ? item.ref.prId : null + ); + + expect(ids).toHaveLength(120); + expect(new Set(ids).size).toBe(120); + expect(ids).toEqual([ + ...Array.from({ length: 40 }, (_, index) => 1000 + index), + ...Array.from({ length: 40 }, (_, index) => 2000 + index), + ...Array.from({ length: 40 }, (_, index) => 3000 + index), + ]); + expect(requests.every(request => request.page === '1')).toBe(true); + expect(third.nextCursor).toBeNull(); + }); it('continues several repositories across pages', async () => { // repo-a's rows are all older than repo-b's, so the first load serves @@ -1209,24 +1245,64 @@ describe('listInbox', () => { expect(requests.filter(request => request.query === 'state="OPEN"')).toHaveLength(12); }); - it('reads one provider page per enumerated repository', async () => { - // The enumeration cap is 300 repositories; a load reads at most one page - // from each of them. + it('caps the workspace fan-out at the enumerated repository set', async () => { + const slugs = Array.from({ length: 301 }, (_, repo) => `repo-${repo}`); const rowsBySlug: Record = {}; - for (let repo = 0; repo < 300; repo += 1) { - rowsBySlug[`repo-${repo}`] = [ - inboxPr(repo, `2026-09-02T00:${String(repo % 60).padStart(2, '0')}:00+00:00`), - ]; + for (const slug of slugs) { + rowsBySlug[slug] = [inboxPr(slugs.indexOf(slug) + 1, '2026-09-02T00:00:00+00:00')]; } const requests: InboxRequest[] = []; serveInbox(rowsBySlug, requests); const page = await listInbox(ORG_OWNER); - expect(requests).toHaveLength(300); + const expected = [...slugs].sort().slice(0, 300); + expect(requests.map(request => request.slug).sort()).toEqual(expected); expect(page.items).toHaveLength(50); expect(page.nextCursor).toBeTruthy(); }); + it('keeps a maximum-size continuation within the receiving wire limit', async () => { + const slugs = Array.from( + { length: 300 }, + (_, index) => `repo-${String(index).padStart(3, '0')}` + ); + const tiedAt = '2026-09-02T00:00:00.000000+00:00'; + const rowsBySlug: Record = {}; + for (const slug of slugs) { + rowsBySlug[slug] = Array.from({ length: 60 }, (_, index) => + inboxPr(index + 1, tiedAt, `acme/${slug}`) + ); + } + const requests: InboxRequest[] = []; + serveInbox(rowsBySlug, requests); + + const fingerprint = crypto + .createHash('sha256') + .update( + [...slugs] + .sort() + .map(slug => `acme/${slug}`) + .join('\0') + ) + .digest('base64url'); + const pages = Buffer.alloc(slugs.length * 2); + for (let index = 0; index < slugs.length; index += 1) pages.writeUInt16BE(2, index * 2); + const cursor = createSignedToken({ + kind: 'bitbucket-inbox-cursor', + identity: `bitbucket-inbox:acme:${ORG_OWNER.organizationId}:${ORG_OWNER.userId}:${fingerprint}`, + boundary: tiedAt, + lastRef: null, + pages: pages.toString('base64url'), + }); + expect(cursor.length).toBeLessThanOrEqual(INBOX_CURSOR_MAX_CHARS); + + const page = await listInbox(ORG_OWNER, cursor); + + expect(requests).toHaveLength(300); + expect(requests.every(request => request.page === '2')).toBe(true); + expect(page.nextCursor?.length).toBeLessThanOrEqual(INBOX_CURSOR_MAX_CHARS); + }); + it('starts again at the first page when continuation state is not valid', async () => { const requests: InboxRequest[] = []; serveInbox({ repo: minutesFixture(100, 60) }, requests); diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 93775e64ed6..c9f588cc880 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -11,9 +11,9 @@ */ import 'server-only'; +import { createHash } from 'node:crypto'; import { z } from 'zod'; import { - providerPrRefKey, type ProviderPrChecksResult, type ProviderPrFile, type ProviderPrFilesPage, @@ -54,6 +54,14 @@ const MAX_BUILD_PAGES = 3; const INBOX_REPOSITORY_CONCURRENCY = 8; /** The maximum repository count served by the inbox on one load. */ const MAX_INBOX_REPOSITORIES = 300; +/** + * The receiving wire limit of an inbox cursor: the `pageCursor` bound the + * provider-review router accepts, so a cursor this layer issues always fits + * the next request. The per-repository page state is two bytes each — 300 + * repositories stay far below the limit — and a test pins the maximum-size + * cursor to it. + */ +export const INBOX_CURSOR_MAX_CHARS = 2048; /** How long a server-issued inbox cursor stays usable. */ const INBOX_CURSOR_TTL_SECONDS = 60 * 60; /** @@ -980,11 +988,22 @@ export async function listInbox( ): Promise { try { const access = await authorizeWorkspace(owner); + const repositories = [...access.repositories] + .sort((left, right) => + left.fullName.toLowerCase() < right.fullName.toLowerCase() + ? -1 + : left.fullName.toLowerCase() > right.fullName.toLowerCase() + ? 1 + : 0 + ) + .slice(0, MAX_INBOX_REPOSITORIES); + const repositoryFingerprint = createHash('sha256') + .update(repositories.map(repository => repository.fullName.toLowerCase()).join('\0')) + .digest('base64url'); const identity = owner.type === 'organization' - ? `bitbucket-inbox:${access.workspace.slug}:${owner.organizationId}:${owner.userId}` - : `bitbucket-inbox:${access.workspace.slug}:user:${owner.userId}`; - const repositories = access.repositories.slice(0, MAX_INBOX_REPOSITORIES); + ? `bitbucket-inbox:${access.workspace.slug}:${owner.organizationId}:${owner.userId}:${repositoryFingerprint}` + : `bitbucket-inbox:${access.workspace.slug}:user:${owner.userId}:${repositoryFingerprint}`; const repositoryNames = new Set( repositories.map(repository => repository.fullName.toLowerCase()) ); @@ -1021,9 +1040,7 @@ export async function listInbox( } if (parsedPage.data.next) repositoryPageFull = true; const pageItems = inboxItemsFrom(parsedPage.data.values, position, repositoryNames); - const merged = [...window, ...pageItems].sort( - (left, right) => inboxUpdatedMs(right) - inboxUpdatedMs(left) - ); + const merged = [...window, ...pageItems].sort(compareInboxItems); if (merged.length > BITBUCKET_PAGE_SIZE) beyondWindow = true; window = merged.slice(0, BITBUCKET_PAGE_SIZE); if ( @@ -1049,13 +1066,24 @@ export async function listInbox( boundary !== null && position?.boundary === boundary ? position.providerPages : new Map(); - const servedRefs = - boundary === null ? [] : window.filter(item => inboxBoundaryValue(item) === boundary); + let lastBoundaryItem: ProviderPrInboxItem | undefined; + for (let index = window.length - 1; index >= 0; index -= 1) { + if (inboxBoundaryValue(window[index]) === boundary) { + lastBoundaryItem = window[index]; + break; + } + } + const lastRef = + lastBoundaryItem?.ref.platform === 'bitbucket' + ? ([lastBoundaryItem.ref.repoSlug, lastBoundaryItem.ref.prId] as const) + : boundary === position?.boundary + ? position.lastRef + : null; return { items: window, nextCursor: hasMore && boundary !== null - ? encodeInboxCursor(identity, boundary, servedRefs, providerPages, repositories) + ? encodeInboxCursor(identity, boundary, lastRef, providerPages, repositories) : null, }; } catch (error) { @@ -1080,7 +1108,7 @@ function inboxBoundaryValue(item: ProviderPrInboxItem): string | null { if (Number.isNaN(Date.parse(item.updatedAt))) return null; return item.updatedAt; } -/** The inbox rows of one provider page, minus the rows an earlier page served. */ +/** Rows in one provider page after applying the stable inbox continuation. */ function inboxItemsFrom( values: unknown[], position: InboxCursorPosition | null, @@ -1091,8 +1119,14 @@ function inboxItemsFrom( const parsed = BitbucketInboxPullRequestSchema.safeParse(value); if (!parsed.success) continue; const ref = inboxRefFrom(parsed.data, allowedFullNames); - if (!ref) continue; - if (position?.servedRefs.has(providerPrRefKey(ref))) continue; + if (ref?.platform !== 'bitbucket') continue; + if ( + position?.lastRef && + parsed.data.updated_on === position.boundary && + compareInboxTie(ref.repoSlug, ref.prId, position.lastRef[0], position.lastRef[1]) <= 0 + ) { + continue; + } items.push({ ref, title: parsed.data.title, @@ -1105,41 +1139,50 @@ function inboxItemsFrom( return items; } +function compareInboxTie( + leftRepo: string, + leftId: number, + rightRepo: string, + rightId: number +): number { + const leftKey = leftRepo.toLowerCase(); + const rightKey = rightRepo.toLowerCase(); + if (leftKey !== rightKey) return leftKey < rightKey ? -1 : 1; + return leftId - rightId; +} + +function compareInboxItems(left: ProviderPrInboxItem, right: ProviderPrInboxItem): number { + const timestampOrder = inboxUpdatedMs(right) - inboxUpdatedMs(left); + if (timestampOrder !== 0) return timestampOrder; + if (left.ref.platform !== 'bitbucket' || right.ref.platform !== 'bitbucket') return 0; + return compareInboxTie(left.ref.repoSlug, left.ref.prId, right.ref.repoSlug, right.ref.prId); +} + const INBOX_CURSOR_KIND = 'bitbucket-inbox-cursor'; const inboxCursorPayloadSchema = z.object({ kind: z.literal(INBOX_CURSOR_KIND), identity: z.string().min(1), boundary: z.string().min(1), - served: z.string().max(400), + lastRef: z.tuple([z.string().min(1).max(512), z.number().int().positive()]).nullable(), pages: z.string().max(800), }); type InboxCursorPosition = { boundary: string; - servedRefs: ReadonlySet; + lastRef: readonly [string, number] | null; providerPages: Map; }; function encodeInboxCursor( identity: string, boundary: string, - servedRefs: ProviderPrInboxItem[], + lastRef: readonly [string, number] | null, providerPages: ReadonlyMap, repositories: BitbucketWorkspaceRepositoryAccess['repositories'] ): string { const repositoryIndexes = new Map( repositories.map((repository, index) => [repository.fullName.toLowerCase(), index]) ); - const served = Buffer.alloc(servedRefs.length * 6); - let servedOffset = 0; - for (const item of servedRefs) { - if (item.ref.platform !== 'bitbucket') continue; - const index = repositoryIndexes.get(`${item.ref.workspace}/${item.ref.repoSlug}`.toLowerCase()); - if (index === undefined || item.ref.prId > 0xffff_ffff) continue; - served.writeUInt16BE(index, servedOffset); - served.writeUInt32BE(item.ref.prId, servedOffset + 2); - servedOffset += 6; - } const pages = Buffer.alloc(repositories.length * 2); for (const [fullName, page] of providerPages) { const index = repositoryIndexes.get(fullName); @@ -1149,7 +1192,7 @@ function encodeInboxCursor( kind: INBOX_CURSOR_KIND, identity, boundary, - served: served.subarray(0, servedOffset).toString('base64url'), + lastRef, pages: pages.toString('base64url'), }); } @@ -1165,36 +1208,20 @@ function decodeInboxCursor( parse: value => { const parsed = inboxCursorPayloadSchema.safeParse(value); if (!parsed.success || parsed.data.identity !== identity) return null; - const servedBytes = Buffer.from(parsed.data.served, 'base64url'); const pageBytes = Buffer.from(parsed.data.pages, 'base64url'); - if (servedBytes.length % 6 !== 0 || pageBytes.length !== repositories.length * 2) { - return null; - } - const servedRefs = new Set(); - for (let offset = 0; offset < servedBytes.length; offset += 6) { - const repository = repositories[servedBytes.readUInt16BE(offset)]; - if (!repository) return null; - servedRefs.add( - providerPrRefKey({ - platform: 'bitbucket', - workspace: repository.fullName.split('/')[0] ?? '', - repoSlug: repository.slug, - prId: servedBytes.readUInt32BE(offset + 2), - }) - ); - } + if (pageBytes.length !== repositories.length * 2) return null; const providerPages = new Map(); for (let index = 0; index < repositories.length; index += 1) { const page = pageBytes.readUInt16BE(index * 2); if (page > 1) providerPages.set(repositories[index].fullName.toLowerCase(), page); } - return { boundary: parsed.data.boundary, servedRefs, providerPages }; + return { boundary: parsed.data.boundary, lastRef: parsed.data.lastRef, providerPages }; }, }); if (payload === null) return null; return { boundary: payload.boundary, - servedRefs: payload.servedRefs, + lastRef: payload.lastRef, providerPages: payload.providerPages, }; } From 26679ed0adc4de9c046762d26db2803cc1365411 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 04:58:53 +0000 Subject: [PATCH 04/11] fix(provider-review): keep the Bitbucket inbox continuation page-local The continuation records, per repository, the provider page it is on and how many of that page's rows are served, instead of one cross-repository tie key. A repository whose page rows are all served moves to the next provider page in the same load, and a repository with nothing left at the boundary stops being read until the boundary changes, so rows that share a timestamp across pages stay reachable no matter which repository the merged order reaches last. The cursor carries two packed bytes per repository, and a 300-repository continuation stays within the provider-review router's page-cursor limit. --- .../provider-review/bitbucket-read.test.ts | 89 ++++--- .../src/lib/provider-review/bitbucket-read.ts | 226 ++++++++++-------- 2 files changed, 182 insertions(+), 133 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index bb676448496..eb71715c4cc 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1013,6 +1013,27 @@ describe('listInbox', () => { inboxPr(firstId + index, `2026-09-02T00:${String(index).padStart(2, '0')}:00+00:00`) ); + /** The identity a continuation is bound to: the caller plus the repository set. */ + const inboxIdentity = (slugs: string[]) => + `bitbucket-inbox:acme:${ORG_OWNER.organizationId}:${ORG_OWNER.userId}:${crypto + .createHash('sha256') + .update( + [...slugs] + .sort() + .map(slug => `acme/${slug}`.toLowerCase()) + .join('\0') + ) + .digest('base64url')}`; + + /** A continuation token in the shape the read layer issues. */ + const inboxCursor = (identity: string, boundary: string, pages = Buffer.alloc(2)) => + createSignedToken({ + kind: 'bitbucket-inbox-cursor', + identity, + boundary, + pages: pages.toString('base64url'), + }); + it('fans out over the workspace repositories and carries full identity', async () => { fetchMock.mockImplementation(async (url: string | URL) => { const full = url.toString(); @@ -1127,7 +1148,7 @@ describe('listInbox', () => { expect(second.nextCursor).toBeNull(); }); - it('advances past a provider page of equal-timestamp rows', async () => { + it('advances past a provider page whose rows are all served', async () => { const requests: InboxRequest[] = []; const tiedAt = '2026-09-02T00:00:00.000000+00:00'; serveInbox( @@ -1141,15 +1162,11 @@ describe('listInbox', () => { expect(first.nextCursor!.length).toBeLessThanOrEqual(2048); const second = await listInbox(ORG_OWNER, first.nextCursor!); - expect(second.items).toHaveLength(0); - expect(second.nextCursor).toBeTruthy(); - - const third = await listInbox(ORG_OWNER, second.nextCursor!); expect( - third.items.map(item => (item.ref && 'prId' in item.ref ? item.ref.prId : null)) + second.items.map(item => (item.ref && 'prId' in item.ref ? item.ref.prId : null)) ).toEqual([550]); - expect(third.nextCursor).toBeNull(); - expect(requests.map(request => request.page)).toEqual(['1', '1', '2']); + expect(second.nextCursor).toBeNull(); + expect(requests.map(request => request.page)).toEqual(['1', '2']); }); it('keeps equal-time rows in a stable order across inbox pages', async () => { const requests: InboxRequest[] = []; @@ -1218,6 +1235,39 @@ describe('listInbox', () => { expect(new Set(served).size).toBe(100); }); + it('serves every tied row when another repository also sits at the boundary', async () => { + // 60 tied rows in one repository and one tied row in a repository that + // sorts after it: the later repository's row must not hide the rows that + // continue onto the first repository's second provider page. + const requests: InboxRequest[] = []; + const tiedAt = '2026-09-02T00:00:00.000000+00:00'; + serveInbox( + { + 'repo-a': Array.from({ length: 60 }, (_, index) => + inboxPr(100 + index, tiedAt, 'acme/repo-a') + ), + 'repo-b': [inboxPr(900, tiedAt, 'acme/repo-b')], + }, + requests + ); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 4; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (cursor === undefined) break; + } + + expect(served).toHaveLength(61); + expect(new Set(served).size).toBe(61); + expect(served).toEqual([...Array.from({ length: 60 }, (_, index) => 100 + index), 900]); + expect(requests).toHaveLength(4); + }); + it('reads one provider page per repository on every inbox page', async () => { // 12 repositories × 5 open PRs: every load reads one page per repository. const rowsBySlug: Record = {}; @@ -1276,24 +1326,9 @@ describe('listInbox', () => { const requests: InboxRequest[] = []; serveInbox(rowsBySlug, requests); - const fingerprint = crypto - .createHash('sha256') - .update( - [...slugs] - .sort() - .map(slug => `acme/${slug}`) - .join('\0') - ) - .digest('base64url'); const pages = Buffer.alloc(slugs.length * 2); for (let index = 0; index < slugs.length; index += 1) pages.writeUInt16BE(2, index * 2); - const cursor = createSignedToken({ - kind: 'bitbucket-inbox-cursor', - identity: `bitbucket-inbox:acme:${ORG_OWNER.organizationId}:${ORG_OWNER.userId}:${fingerprint}`, - boundary: tiedAt, - lastRef: null, - pages: pages.toString('base64url'), - }); + const cursor = inboxCursor(inboxIdentity(slugs), tiedAt, pages); expect(cursor.length).toBeLessThanOrEqual(INBOX_CURSOR_MAX_CHARS); const page = await listInbox(ORG_OWNER, cursor); @@ -1325,7 +1360,7 @@ describe('listInbox', () => { kind: 'bitbucket-inbox-cursor', identity: 'bitbucket-inbox:evil', boundary: '2026-09-02T00:10:00.000+00:00', - servedRefs: [], + pages: Buffer.alloc(2).toString('base64url'), }); const page = await listInbox(ORG_OWNER, foreign); @@ -1355,9 +1390,9 @@ describe('listInbox', () => { const encoded = Buffer.from( JSON.stringify({ kind: 'bitbucket-inbox-cursor', - identity: 'bitbucket-inbox:acme', + identity: inboxIdentity(['repo']), boundary: '2026-09-02T00:10:00.000+00:00', - servedRefs: [], + pages: Buffer.alloc(2).toString('base64url'), iat: Math.floor(Date.now() / 1000) - 2 * 60 * 60, nonce: 'expired-nonce', }) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index c9f588cc880..2e40732617a 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -977,10 +977,11 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * * The fan-out sorts across repositories, so one inbox page cannot map onto a * single provider page. A signed continuation carries the full-precision - * timestamp boundary, the already-served rows at that boundary, and compact - * per-repository page positions. If a provider page contains only rows already - * served at the boundary, its position advances so later equal-time rows - * remain reachable. Repository reads are capped at one page each per load. + * timestamp boundary and, per repository, the provider page it is on and how + * many of that page's rows are served. A page whose rows are all served moves + * to the next page within the same load, and a repository with nothing left at + * the boundary is not read again until the boundary changes. Repository reads + * are therefore capped at one page each per load. */ export async function listInbox( owner: BitbucketReviewOwner, @@ -1012,14 +1013,18 @@ export async function listInbox( ? `state="OPEN" AND updated_on <= ${position.boundary}` : 'state="OPEN"'; let repositoryPageFull = false; - let window: ProviderPrInboxItem[] = []; + let window: InboxCandidate[] = []; let beyondWindow = false; + const pageReads: InboxPageRead[] = []; for (let offset = 0; offset < repositories.length; offset += INBOX_REPOSITORY_CONCURRENCY) { - const batch = repositories.slice(offset, offset + INBOX_REPOSITORY_CONCURRENCY); + const batch = repositories + .slice(offset, offset + INBOX_REPOSITORY_CONCURRENCY) + .filter(repository => position?.pages.get(repository.fullName.toLowerCase())?.page !== 0); const pages = await Promise.all( - batch.map(repository => { - const page = position?.providerPages.get(repository.fullName.toLowerCase()) ?? 1; - return requestBitbucketJson( + batch.map(async repository => { + const key = repository.fullName.toLowerCase(); + const page = position?.pages.get(key)?.page ?? 1; + const payload = await requestBitbucketJson( access, `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(repository.slug)}/pullrequests`, { @@ -1030,60 +1035,57 @@ export async function listInbox( sort: '-updated_on', }, } - ).then(payload => ({ payload, repository, page })); + ); + return { key, page, payload }; }) ); - for (const { payload, repository, page } of pages) { + for (const { key, page, payload } of pages) { const parsedPage = BitbucketPageSchema.safeParse(payload); if (!parsedPage.success) { throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); } if (parsedPage.data.next) repositoryPageFull = true; - const pageItems = inboxItemsFrom(parsedPage.data.values, position, repositoryNames); - const merged = [...window, ...pageItems].sort(compareInboxItems); + const rows = inboxItemsFrom(parsedPage.data.values, repositoryNames).map( + (item, rowIndex) => ({ item, repoKey: key, rowIndex }) + ); + const served = position?.pages.get(key)?.served ?? 0; + pageReads.push({ key, page, rows, hadNext: parsedPage.data.next !== null }); + const merged = [...window, ...rows.slice(served)].sort(compareInboxCandidates); if (merged.length > BITBUCKET_PAGE_SIZE) beyondWindow = true; window = merged.slice(0, BITBUCKET_PAGE_SIZE); - if ( - position && - parsedPage.data.next && - pageItems.length === 0 && - parsedPage.data.values.length > 0 - ) { - position.providerPages.set(repository.fullName.toLowerCase(), page + 1); - } } } - const hasMore = repositoryPageFull || beyondWindow; - let boundary: string | null = position?.boundary ?? null; - for (let index = window.length - 1; index >= 0; index -= 1) { - const itemBoundary = inboxBoundaryValue(window[index]); - if (itemBoundary !== null) { - boundary = itemBoundary; - break; - } + const boundary = oldestInboxTimestamp(window) ?? position?.boundary ?? null; + const sameBoundary = boundary !== null && position?.boundary === boundary; + const pages = new Map(); + for (const repository of repositories) { + const key = repository.fullName.toLowerCase(); + const carried = sameBoundary ? position?.pages.get(key) : undefined; + pages.set(key, carried ? { ...carried } : { page: 1, served: 0 }); } - const providerPages = - boundary !== null && position?.boundary === boundary - ? position.providerPages - : new Map(); - let lastBoundaryItem: ProviderPrInboxItem | undefined; - for (let index = window.length - 1; index >= 0; index -= 1) { - if (inboxBoundaryValue(window[index]) === boundary) { - lastBoundaryItem = window[index]; - break; - } + // Every row of the window is at the boundary, so the rows served from one + // repository are its next rows in the page order: a count, not a set. + for (const candidate of window) { + if (candidate.item.updatedAt !== boundary) continue; + const state = pages.get(candidate.repoKey); + if (state) state.served += 1; } - const lastRef = - lastBoundaryItem?.ref.platform === 'bitbucket' - ? ([lastBoundaryItem.ref.repoSlug, lastBoundaryItem.ref.prId] as const) - : boundary === position?.boundary - ? position.lastRef - : null; + for (const read of pageReads) { + const state = pages.get(read.key); + if (!state || state.served < read.rows.length) continue; + pages.set( + read.key, + read.hadNext && state.page < INBOX_PAGE_MAX + ? { page: state.page + 1, served: 0 } + : { page: 0, served: 0 } + ); + } + const hasMore = repositoryPageFull || beyondWindow; return { - items: window, + items: window.map(candidate => candidate.item), nextCursor: hasMore && boundary !== null - ? encodeInboxCursor(identity, boundary, lastRef, providerPages, repositories) + ? encodeInboxCursor(identity, boundary, pages, repositories) : null, }; } catch (error) { @@ -1099,19 +1101,18 @@ function inboxUpdatedMs(item: ProviderPrInboxItem): number { return fraction.length > 3 ? ms + Number(fraction.slice(3)) / 10 ** (fraction.length - 3) : ms; } -/** - * The provider filter value of a row's `updated_on`: the provider's own - * timestamp string, so its full precision carries into the continuation. A - * row without a parseable timestamp has no boundary. - */ -function inboxBoundaryValue(item: ProviderPrInboxItem): string | null { - if (Number.isNaN(Date.parse(item.updatedAt))) return null; - return item.updatedAt; +/** The oldest parseable timestamp in the window: the next load's provider filter. */ +function oldestInboxTimestamp(window: InboxCandidate[]): string | null { + for (let index = window.length - 1; index >= 0; index -= 1) { + const item = window[index].item; + if (!Number.isNaN(Date.parse(item.updatedAt))) return item.updatedAt; + } + return null; } -/** Rows in one provider page after applying the stable inbox continuation. */ + +/** The inbox rows of one provider page, in provider order. */ function inboxItemsFrom( values: unknown[], - position: InboxCursorPosition | null, allowedFullNames: ReadonlySet ): ProviderPrInboxItem[] { const items: ProviderPrInboxItem[] = []; @@ -1120,13 +1121,6 @@ function inboxItemsFrom( if (!parsed.success) continue; const ref = inboxRefFrom(parsed.data, allowedFullNames); if (ref?.platform !== 'bitbucket') continue; - if ( - position?.lastRef && - parsed.data.updated_on === position.boundary && - compareInboxTie(ref.repoSlug, ref.prId, position.lastRef[0], position.lastRef[1]) <= 0 - ) { - continue; - } items.push({ ref, title: parsed.data.title, @@ -1139,61 +1133,81 @@ function inboxItemsFrom( return items; } -function compareInboxTie( - leftRepo: string, - leftId: number, - rightRepo: string, - rightId: number -): number { - const leftKey = leftRepo.toLowerCase(); - const rightKey = rightRepo.toLowerCase(); - if (leftKey !== rightKey) return leftKey < rightKey ? -1 : 1; - return leftId - rightId; -} - -function compareInboxItems(left: ProviderPrInboxItem, right: ProviderPrInboxItem): number { - const timestampOrder = inboxUpdatedMs(right) - inboxUpdatedMs(left); +/** + * The merged inbox order: newest first, ties broken by repository and then by + * the row's position in that repository's provider page. Page order is kept, + * so the rows a continuation served from a page are always its leading rows. + */ +function compareInboxCandidates(left: InboxCandidate, right: InboxCandidate): number { + const timestampOrder = inboxUpdatedMs(right.item) - inboxUpdatedMs(left.item); if (timestampOrder !== 0) return timestampOrder; - if (left.ref.platform !== 'bitbucket' || right.ref.platform !== 'bitbucket') return 0; - return compareInboxTie(left.ref.repoSlug, left.ref.prId, right.ref.repoSlug, right.ref.prId); + if (left.repoKey !== right.repoKey) return left.repoKey < right.repoKey ? -1 : 1; + return left.rowIndex - right.rowIndex; } const INBOX_CURSOR_KIND = 'bitbucket-inbox-cursor'; +/** Bits of the packed word that hold the page number; the rest hold the count. */ +const INBOX_PAGE_BITS = 10; +const INBOX_PAGE_MAX = (1 << INBOX_PAGE_BITS) - 1; const inboxCursorPayloadSchema = z.object({ kind: z.literal(INBOX_CURSOR_KIND), identity: z.string().min(1), boundary: z.string().min(1), - lastRef: z.tuple([z.string().min(1).max(512), z.number().int().positive()]).nullable(), pages: z.string().max(800), }); +/** One provider row with the repository and page position it was read at. */ +type InboxCandidate = { + item: ProviderPrInboxItem; + repoKey: string; + rowIndex: number; +}; + +type InboxPageRead = { + key: string; + page: number; + rows: InboxCandidate[]; + hadNext: boolean; +}; + +/** + * One repository's position in the continuation: the provider page it is on + * and how many of that page's rows are served. Page 0 means the repository has + * nothing left at this boundary and is not read again until the boundary + * changes. + */ +type InboxPageState = { page: number; served: number }; + type InboxCursorPosition = { boundary: string; - lastRef: readonly [string, number] | null; - providerPages: Map; + pages: ReadonlyMap; }; +/** + * The inbox cursor is a signed continuation token carrying the caller and + * repository-set identity, the provider timestamp boundary, and two packed + * bytes per repository: its page number and how many rows of that page are + * served. The page position is meaningful only while the boundary is + * unchanged, because a different boundary selects a different provider + * collection. + */ function encodeInboxCursor( identity: string, boundary: string, - lastRef: readonly [string, number] | null, - providerPages: ReadonlyMap, + pages: ReadonlyMap, repositories: BitbucketWorkspaceRepositoryAccess['repositories'] ): string { - const repositoryIndexes = new Map( - repositories.map((repository, index) => [repository.fullName.toLowerCase(), index]) - ); - const pages = Buffer.alloc(repositories.length * 2); - for (const [fullName, page] of providerPages) { - const index = repositoryIndexes.get(fullName); - if (index !== undefined && page <= 0xffff) pages.writeUInt16BE(page, index * 2); + const pageState = Buffer.alloc(repositories.length * 2); + for (const [index, repository] of repositories.entries()) { + const state = pages.get(repository.fullName.toLowerCase()); + const page = Math.min(state?.page ?? 1, INBOX_PAGE_MAX); + pageState.writeUInt16BE(((state?.served ?? 0) << INBOX_PAGE_BITS) | page, index * 2); } return createSignedToken({ kind: INBOX_CURSOR_KIND, identity, boundary, - lastRef, - pages: pages.toString('base64url'), + pages: pageState.toString('base64url'), }); } @@ -1208,22 +1222,22 @@ function decodeInboxCursor( parse: value => { const parsed = inboxCursorPayloadSchema.safeParse(value); if (!parsed.success || parsed.data.identity !== identity) return null; - const pageBytes = Buffer.from(parsed.data.pages, 'base64url'); - if (pageBytes.length !== repositories.length * 2) return null; - const providerPages = new Map(); + const pageState = Buffer.from(parsed.data.pages, 'base64url'); + if (pageState.length !== repositories.length * 2) return null; + const pages = new Map(); for (let index = 0; index < repositories.length; index += 1) { - const page = pageBytes.readUInt16BE(index * 2); - if (page > 1) providerPages.set(repositories[index].fullName.toLowerCase(), page); + const packed = pageState.readUInt16BE(index * 2); + const served = packed >> INBOX_PAGE_BITS; + if (served > BITBUCKET_PAGE_SIZE) return null; + pages.set(repositories[index].fullName.toLowerCase(), { + page: packed & INBOX_PAGE_MAX, + served, + }); } - return { boundary: parsed.data.boundary, lastRef: parsed.data.lastRef, providerPages }; + return { boundary: parsed.data.boundary, pages }; }, }); - if (payload === null) return null; - return { - boundary: payload.boundary, - lastRef: payload.lastRef, - providerPages: payload.providerPages, - }; + return payload; } /** * The ref of an inbox row: `workspace/repo-slug` from the destination From 7ff7b435d3e8c1b3a56daf28058581a213c6b4f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 05:09:10 +0000 Subject: [PATCH 05/11] fix(provider-review): end a Bitbucket inbox repository at its last page Bitbucket omits the next link entirely on the last page, so a repository is now paginated only while the link is present. Added a case where one repository's single-row last page carries no next link while another repository keeps paginating: the finished repository is not requested again, and the provider's page-not-found answer never reaches the inbox. --- .../provider-review/bitbucket-read.test.ts | 67 +++++++++++++++++++ .../src/lib/provider-review/bitbucket-read.ts | 2 +- 2 files changed, 68 insertions(+), 1 deletion(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index eb71715c4cc..94d7a985bb6 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1268,6 +1268,73 @@ describe('listInbox', () => { expect(requests).toHaveLength(4); }); + it('ends a repository whose final provider page omits the next link', async () => { + // Both repositories share one timestamp. The first has a single extra row + // on its last page, which Bitbucket answers without a `next` key, while the + // second keeps paginating: the finished repository must not be asked for a + // page that does not exist. + const tiedAt = '2026-09-02T00:00:00.000000+00:00'; + const rowsBySlug: Record = { + 'repo-a': Array.from({ length: 51 }, (_, index) => + inboxPr(100 + index, tiedAt, 'acme/repo-a') + ), + 'repo-b': Array.from({ length: 100 }, (_, index) => + inboxPr(500 + index, tiedAt, 'acme/repo-b') + ), + }; + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: Object.keys(rowsBySlug).map((slug, index) => ({ + id: `87654321-4321-4321-4321-${String(index).padStart(12, '0')}`, + workspaceUuid: WORKSPACE.uuid, + name: slug, + fullName: `acme/${slug}`, + private: true, + defaultBranch: 'main', + })), + syncedAt: '2026-09-06T00:00:00.000Z', + }); + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const match = parsed.pathname.match(/^\/2\.0\/repositories\/acme\/([^/]+)\/pullrequests$/); + const rows = match ? rowsBySlug[match[1]] : undefined; + if (!match || !rows) return jsonResponse({ pagelen: 50, values: [] }); + const page = Number(parsed.searchParams.get('page') ?? '1'); + const values = rows.slice((page - 1) * 50, page * 50); + if (values.length === 0) { + // What Bitbucket answers for a page past the collection. + return jsonResponse( + { type: 'error', error: { message: 'Page number does not exist' } }, + 404 + ); + } + return jsonResponse( + rows.length > page * 50 + ? { pagelen: 50, values, next: 'https://api.bitbucket.org/2.0/next?page=2' } + : { pagelen: 50, values } + ); + }); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 6; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (cursor === undefined) break; + } + + expect(new Set(served).size).toBe(151); + expect(served).toHaveLength(151); + expect(cursor).toBeUndefined(); + }); + it('reads one provider page per repository on every inbox page', async () => { // 12 repositories × 5 open PRs: every load reads one page per repository. const rowsBySlug: Record = {}; diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 2e40732617a..55de3561ce8 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -1049,7 +1049,7 @@ export async function listInbox( (item, rowIndex) => ({ item, repoKey: key, rowIndex }) ); const served = position?.pages.get(key)?.served ?? 0; - pageReads.push({ key, page, rows, hadNext: parsedPage.data.next !== null }); + pageReads.push({ key, page, rows, hadNext: Boolean(parsedPage.data.next) }); const merged = [...window, ...rows.slice(served)].sort(compareInboxCandidates); if (merged.length > BITBUCKET_PAGE_SIZE) beyondWindow = true; window = merged.slice(0, BITBUCKET_PAGE_SIZE); From 83b935baf26334a9e253050172aa1ac7ea979e3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 20:27:37 +0000 Subject: [PATCH 06/11] fix(provider-review): serve the Bitbucket inbox from page positions The inbox continuation now carries a fixed provider ceiling and, per repository, the page it is on plus how many leading rows of that page are served. Each load reads at most one page per repository that still has rows, skips the rows already served, and moves to the next page once the current one is fully served, so a continuation advances without restarting. Tests cover distinct timestamps across repositories where one repository spans two provider pages, a single repository whose first page is fully served, and a continuation that keeps sub-millisecond precision. --- .../provider-review/bitbucket-read.test.ts | 101 +++++++--- .../src/lib/provider-review/bitbucket-read.ts | 183 +++++++++++------- .../routers/provider-review-router.test.ts | 2 +- 3 files changed, 179 insertions(+), 107 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 94d7a985bb6..9fa1cf77ef0 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1026,11 +1026,11 @@ describe('listInbox', () => { .digest('base64url')}`; /** A continuation token in the shape the read layer issues. */ - const inboxCursor = (identity: string, boundary: string, pages = Buffer.alloc(2)) => + const inboxCursor = (identity: string, ceiling: string, pages = Buffer.alloc(2)) => createSignedToken({ kind: 'bitbucket-inbox-cursor', identity, - boundary, + ceiling, pages: pages.toString('base64url'), }); @@ -1081,9 +1081,10 @@ describe('listInbox', () => { expect(result.nextCursor).toBeNull(); }); - it('continues one repository from the signed boundary', async () => { + it('continues one repository from the signed ceiling', async () => { // 60 open PRs: the first load serves the newest 50 and hands back a signed - // boundary; the next load asks only for rows at or below that boundary. + // continuation; the next load re-reads the first page, skips the rows + // already served, and reads the second page in the same load. const requests: InboxRequest[] = []; serveInbox({ repo: minutesFixture(100, 60) }, requests); @@ -1095,13 +1096,13 @@ describe('listInbox', () => { expect(requests[0]).toMatchObject({ slug: 'repo', query: 'state="OPEN"', page: '1' }); const second = await listInbox(ORG_OWNER, first.nextCursor!); - // One provider page for the continuation too, carrying the boundary of - // the last row already served. + // The first load's page was fully served, so the continuation moves to the + // second provider page under the first load's ceiling. expect(requests).toHaveLength(2); expect(requests[1]).toMatchObject({ slug: 'repo', - query: 'state="OPEN" AND updated_on <= 2026-09-02T00:10:00+00:00', - page: '1', + query: 'state="OPEN" AND updated_on <= 2026-09-02T00:59:00+00:00', + page: '2', }); expect(second.items).toHaveLength(10); expect(second.items[0]?.ref).toMatchObject({ prId: 109 }); @@ -1113,18 +1114,18 @@ describe('listInbox', () => { expect(new Set(served).size).toBe(60); }); - it('continues past a boundary that carries microseconds', async () => { - // The window ends on a timestamp inside the second's millisecond group; the - // row in the same millisecond from another repository still follows. + it('keeps the ceiling precise enough for rows inside one millisecond', async () => { + // 51 rows share a millisecond and differ only below it: the window serves + // the newest 50, and the continuation must keep the last one, whose + // provider timestamp sits above a millisecond-truncated ceiling. const requests: InboxRequest[] = []; serveInbox( { - 'repo-a': [ - inboxPr(300, '2026-09-02T00:00:02.400902+00:00', 'acme/repo-a'), - inboxPr(301, '2026-09-02T00:00:02.400500+00:00', 'acme/repo-a'), - ], - 'repo-b': Array.from({ length: 49 }, (_, index) => - inboxPr(400 + index, `2026-09-02T00:02:${String(index).padStart(2, '0')}.900100+00:00`) + repo: Array.from({ length: 51 }, (_, index) => + inboxPr( + 300 + index, + `2026-09-02T00:00:02.${String(410101 + index).padStart(6, '0')}+00:00` + ) ), }, requests @@ -1132,19 +1133,16 @@ describe('listInbox', () => { const first = await listInbox(ORG_OWNER); expect(first.items).toHaveLength(50); - expect(first.items[49]?.ref).toMatchObject({ prId: 300 }); + expect(requests).toHaveLength(1); const second = await listInbox(ORG_OWNER, first.nextCursor!); - expect(requests).toHaveLength(4); - expect( - requests.filter( - request => - request.query === 'state="OPEN" AND updated_on <= 2026-09-02T00:00:02.400902+00:00' - ) - ).toHaveLength(2); + expect(first.nextCursor).toBeTruthy(); + expect(requests[1]).toMatchObject({ + query: 'state="OPEN" AND updated_on <= 2026-09-02T00:00:02.410151+00:00', + }); expect( second.items.map(item => (item.ref && 'prId' in item.ref ? item.ref.prId : null)) - ).toEqual([301]); + ).toEqual([300]); expect(second.nextCursor).toBeNull(); }); @@ -1227,7 +1225,13 @@ describe('listInbox', () => { expect(second.items).toHaveLength(50); expect(second.items[0]?.ref).toMatchObject({ prId: 149 }); expect(second.nextCursor).toBeNull(); - expect(requests).toHaveLength(4); + // repo-b has nothing left, so the continuation reads only repo-a. + expect(requests).toHaveLength(3); + expect(requests[2]).toMatchObject({ + slug: 'repo-a', + query: 'state="OPEN" AND updated_on <= 2026-09-02T01:49:00+00:00', + page: '1', + }); const served = [...first.items, ...second.items].flatMap(item => item.ref && 'prId' in item.ref ? [item.ref.prId] : [] @@ -1235,6 +1239,38 @@ describe('listInbox', () => { expect(new Set(served).size).toBe(100); }); + it('serves every row once when one repository spans provider pages and another is older', async () => { + // repo-a has 100 rows with distinct times, two provider pages of 50; + // repo-b has one row older than every repo-a row. Every load reads one + // provider page per repository, so repo-b's single row must not drag the + // continuation below repo-a's second page. + const base = Date.UTC(2026, 8, 2, 0, 0, 0); + const requests: InboxRequest[] = []; + serveInbox( + { + 'repo-a': Array.from({ length: 100 }, (_, index) => + inboxPr(100 + index, new Date(base - index * 1000).toISOString()) + ), + 'repo-b': [inboxPr(900, new Date(base - 1000 * 1000).toISOString())], + }, + requests + ); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 10; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (cursor === undefined) break; + } + + expect(new Set(served).size).toBe(101); + expect(served).toHaveLength(101); + }); + it('serves every tied row when another repository also sits at the boundary', async () => { // 60 tied rows in one repository and one tied row in a repository that // sorts after it: the later repository's row must not hide the rows that @@ -1335,8 +1371,9 @@ describe('listInbox', () => { expect(cursor).toBeUndefined(); }); - it('reads one provider page per repository on every inbox page', async () => { - // 12 repositories × 5 open PRs: every load reads one page per repository. + it('reads one provider page per repository that still has rows', async () => { + // 12 repositories × 5 open PRs: the first load reads one page per + // repository, and the continuation reads only the two with rows left. const rowsBySlug: Record = {}; for (let repo = 0; repo < 12; repo += 1) { rowsBySlug[`repo-${repo}`] = Array.from({ length: 5 }, (_, index) => @@ -1355,7 +1392,7 @@ describe('listInbox', () => { expect(requests).toHaveLength(12); const second = await listInbox(ORG_OWNER, first.nextCursor!); - expect(requests).toHaveLength(24); + expect(requests).toHaveLength(14); expect(second.items).toHaveLength(10); expect(second.nextCursor).toBeNull(); expect(requests.every(request => request.page === '1')).toBe(true); @@ -1426,7 +1463,7 @@ describe('listInbox', () => { const foreign = createSignedToken({ kind: 'bitbucket-inbox-cursor', identity: 'bitbucket-inbox:evil', - boundary: '2026-09-02T00:10:00.000+00:00', + ceiling: '2026-09-02T00:10:00.000+00:00', pages: Buffer.alloc(2).toString('base64url'), }); const page = await listInbox(ORG_OWNER, foreign); @@ -1458,7 +1495,7 @@ describe('listInbox', () => { JSON.stringify({ kind: 'bitbucket-inbox-cursor', identity: inboxIdentity(['repo']), - boundary: '2026-09-02T00:10:00.000+00:00', + ceiling: '2026-09-02T00:10:00.000+00:00', pages: Buffer.alloc(2).toString('base64url'), iat: Math.floor(Date.now() / 1000) - 2 * 60 * 60, nonce: 'expired-nonce', diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 0f537d46eb7..844924e492e 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -980,12 +980,20 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * PRs of the integration's repositories, newest first. * * The fan-out sorts across repositories, so one inbox page cannot map onto a - * single provider page. A signed continuation carries the full-precision - * timestamp boundary and, per repository, the provider page it is on and how - * many of that page's rows are served. A page whose rows are all served moves - * to the next page within the same load, and a repository with nothing left at - * the boundary is not read again until the boundary changes. Repository reads - * are therefore capped at one page each per load. + * single provider page. A signed continuation carries a fixed provider ceiling + * — the newest `updated_on` the first load saw, so every row that existed then + * stays inside the collection — and, per repository, the provider page it is + * on and how many leading rows of that page are already served. + * + * Rows are emitted in the merged order, so a repository's served rows are + * always the leading rows of its current page: the next load re-reads that + * page, skips them, and moves to the next page once the page is fully served. + * The continuation advances monotonically, never moves back to the first page, + * and reads at most one provider page per repository per load (two when the + * current page is already fully served). A pull request updated after the + * first load sits above the ceiling and appears on a fresh inbox load; one + * that stops being open leaves the collection and shifts the rows after it up, + * which the provider's page-number paging cannot prevent. */ export async function listInbox( owner: BitbucketReviewOwner, @@ -1014,11 +1022,9 @@ export async function listInbox( ); const position = decodeInboxCursor(cursor, identity, repositories); const providerQuery = position - ? `state="OPEN" AND updated_on <= ${position.boundary}` + ? `state="OPEN" AND updated_on <= ${position.ceiling}` : 'state="OPEN"'; - let repositoryPageFull = false; let window: InboxCandidate[] = []; - let beyondWindow = false; const pageReads: InboxPageRead[] = []; for (let offset = 0; offset < repositories.length; offset += INBOX_REPOSITORY_CONCURRENCY) { const batch = repositories @@ -1027,69 +1033,79 @@ export async function listInbox( const pages = await Promise.all( batch.map(async repository => { const key = repository.fullName.toLowerCase(); - const page = position?.pages.get(key)?.page ?? 1; - const payload = await requestBitbucketJson( - access, - `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(repository.slug)}/pullrequests`, - { - query: { - pagelen: BITBUCKET_PAGE_SIZE, - page, - q: providerQuery, - sort: '-updated_on', - }, + let page = position?.pages.get(key)?.page ?? 1; + let served = position?.pages.get(key)?.served ?? 0; + for (let reads = 1; ; reads += 1) { + const payload = await requestBitbucketJson( + access, + `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(repository.slug)}/pullrequests`, + { + query: { + pagelen: BITBUCKET_PAGE_SIZE, + page, + q: providerQuery, + sort: '-updated_on', + }, + } + ); + const parsedPage = BitbucketPageSchema.safeParse(payload); + if (!parsedPage.success) { + throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); } - ); - return { key, page, payload }; + const rows = inboxItemsFrom(parsedPage.data.values, repositoryNames).map( + (item, rowIndex) => ({ item, repoKey: key, rowIndex }) + ); + const hadNext = Boolean(parsedPage.data.next); + // The page was fully served on an earlier load: move on to the next + // page within this load, so the continuation makes progress without + // spending a load on a page that has nothing left to offer. + if ( + served >= rows.length && + hadNext && + page < INBOX_PAGE_MAX && + reads < INBOX_PAGE_READS_PER_LOAD + ) { + page += 1; + served = 0; + continue; + } + return { key, page, rows, fromRow: Math.min(served, rows.length), hadNext }; + } }) ); - for (const { key, page, payload } of pages) { - const parsedPage = BitbucketPageSchema.safeParse(payload); - if (!parsedPage.success) { - throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); - } - if (parsedPage.data.next) repositoryPageFull = true; - const rows = inboxItemsFrom(parsedPage.data.values, repositoryNames).map( - (item, rowIndex) => ({ item, repoKey: key, rowIndex }) - ); - const served = position?.pages.get(key)?.served ?? 0; - pageReads.push({ key, page, rows, hadNext: Boolean(parsedPage.data.next) }); - const merged = [...window, ...rows.slice(served)].sort(compareInboxCandidates); - if (merged.length > BITBUCKET_PAGE_SIZE) beyondWindow = true; + for (const read of pages) { + const merged = [...window, ...read.rows.slice(read.fromRow)].sort(compareInboxCandidates); window = merged.slice(0, BITBUCKET_PAGE_SIZE); + pageReads.push(read); } } - const boundary = oldestInboxTimestamp(window) ?? position?.boundary ?? null; - const sameBoundary = boundary !== null && position?.boundary === boundary; + const ceiling = position?.ceiling ?? inboxCeiling(pageReads); + const emitted = new Map(); + for (const candidate of window) { + emitted.set(candidate.repoKey, (emitted.get(candidate.repoKey) ?? 0) + 1); + } const pages = new Map(); for (const repository of repositories) { - const key = repository.fullName.toLowerCase(); - const carried = sameBoundary ? position?.pages.get(key) : undefined; - pages.set(key, carried ? { ...carried } : { page: 1, served: 0 }); - } - // Every row of the window is at the boundary, so the rows served from one - // repository are its next rows in the page order: a count, not a set. - for (const candidate of window) { - if (candidate.item.updatedAt !== boundary) continue; - const state = pages.get(candidate.repoKey); - if (state) state.served += 1; + const carried = position?.pages.get(repository.fullName.toLowerCase()); + if (!carried || carried.page === 0) pages.set(repository.fullName.toLowerCase(), DONE); } for (const read of pageReads) { - const state = pages.get(read.key); - if (!state || state.served < read.rows.length) continue; + const served = read.fromRow + (emitted.get(read.key) ?? 0); pages.set( read.key, - read.hadNext && state.page < INBOX_PAGE_MAX - ? { page: state.page + 1, served: 0 } - : { page: 0, served: 0 } + served >= read.rows.length + ? read.hadNext && read.page < INBOX_PAGE_MAX + ? { page: read.page + 1, served: 0 } + : DONE + : { page: read.page, served } ); } - const hasMore = repositoryPageFull || beyondWindow; + const hasMore = [...pages.values()].some(state => state.page !== 0); return { items: window.map(candidate => candidate.item), nextCursor: - hasMore && boundary !== null - ? encodeInboxCursor(identity, boundary, pages, repositories) + hasMore && ceiling !== null + ? encodeInboxCursor(identity, ceiling, pages, repositories) : null, }; } catch (error) { @@ -1105,13 +1121,25 @@ function inboxUpdatedMs(item: ProviderPrInboxItem): number { return fraction.length > 3 ? ms + Number(fraction.slice(3)) / 10 ** (fraction.length - 3) : ms; } -/** The oldest parseable timestamp in the window: the next load's provider filter. */ -function oldestInboxTimestamp(window: InboxCandidate[]): string | null { - for (let index = window.length - 1; index >= 0; index -= 1) { - const item = window[index].item; - if (!Number.isNaN(Date.parse(item.updatedAt))) return item.updatedAt; +/** + * The newest provider timestamp the pages of a load carry: the continuation's + * fixed ceiling. Stored as the provider's own string, so the filter keeps the + * full microsecond precision the collection was read at. + */ +function inboxCeiling(reads: InboxPageRead[]): string | null { + let ceiling: string | null = null; + let newest = -Infinity; + for (const read of reads) { + for (const candidate of read.rows) { + if (Number.isNaN(Date.parse(candidate.item.updatedAt))) continue; + const ms = inboxUpdatedMs(candidate.item); + if (ms > newest) { + newest = ms; + ceiling = candidate.item.updatedAt; + } + } } - return null; + return ceiling; } /** The inbox rows of one provider page, in provider order. */ @@ -1153,10 +1181,15 @@ const INBOX_CURSOR_KIND = 'bitbucket-inbox-cursor'; /** Bits of the packed word that hold the page number; the rest hold the count. */ const INBOX_PAGE_BITS = 10; const INBOX_PAGE_MAX = (1 << INBOX_PAGE_BITS) - 1; +/** + * Provider page reads one repository may spend on a single load: its current + * page, plus the next page when the current one is already fully served. + */ +const INBOX_PAGE_READS_PER_LOAD = 2; const inboxCursorPayloadSchema = z.object({ kind: z.literal(INBOX_CURSOR_KIND), identity: z.string().min(1), - boundary: z.string().min(1), + ceiling: z.string().min(1), pages: z.string().max(800), }); @@ -1171,33 +1204,35 @@ type InboxPageRead = { key: string; page: number; rows: InboxCandidate[]; + /** How many of this page's leading rows were already served before this load. */ + fromRow: number; hadNext: boolean; }; /** * One repository's position in the continuation: the provider page it is on - * and how many of that page's rows are served. Page 0 means the repository has - * nothing left at this boundary and is not read again until the boundary - * changes. + * and how many of that page's leading rows are served. Page 0 means the + * repository has nothing left to serve. */ type InboxPageState = { page: number; served: number }; +/** A repository with nothing left to serve. */ +const DONE: InboxPageState = { page: 0, served: 0 }; + type InboxCursorPosition = { - boundary: string; + ceiling: string; pages: ReadonlyMap; }; /** * The inbox cursor is a signed continuation token carrying the caller and - * repository-set identity, the provider timestamp boundary, and two packed - * bytes per repository: its page number and how many rows of that page are - * served. The page position is meaningful only while the boundary is - * unchanged, because a different boundary selects a different provider - * collection. + * repository-set identity, the provider timestamp ceiling the collection was + * read at, and two packed bytes per repository: its page number and how many + * leading rows of that page are served. */ function encodeInboxCursor( identity: string, - boundary: string, + ceiling: string, pages: ReadonlyMap, repositories: BitbucketWorkspaceRepositoryAccess['repositories'] ): string { @@ -1210,7 +1245,7 @@ function encodeInboxCursor( return createSignedToken({ kind: INBOX_CURSOR_KIND, identity, - boundary, + ceiling, pages: pageState.toString('base64url'), }); } @@ -1238,7 +1273,7 @@ function decodeInboxCursor( served, }); } - return { boundary: parsed.data.boundary, pages }; + return { ceiling: parsed.data.ceiling, pages }; }, }); return payload; diff --git a/apps/web/src/routers/provider-review-router.test.ts b/apps/web/src/routers/provider-review-router.test.ts index 77d2c63f074..134c02da74d 100644 --- a/apps/web/src/routers/provider-review-router.test.ts +++ b/apps/web/src/routers/provider-review-router.test.ts @@ -282,7 +282,7 @@ describe('providerReviewRouter inputs', () => { const base = { kind: 'bitbucket-inbox-cursor', identity: 'bitbucket-inbox:acme', - boundary: '2026-09-02T00:00:00.000000+00:00', + ceiling: '2026-09-02T00:00:00.000000+00:00', served: [], pages: [], }; From 55ac218a215e5c1811002c7738aeb8b6c128bf9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 22:30:05 +0000 Subject: [PATCH 07/11] fix(provider-review): end a Bitbucket inbox repository whose page is gone A cached repository the provider no longer serves, and a provider page an earlier load was promised for a collection that has since shrunk, now end that repository's walk instead of the whole inbox load. The load still serves every repository the provider answers for, and the repository stops appearing in later continuations. --- .../provider-review/bitbucket-read.test.ts | 128 ++++++++++++++++++ .../src/lib/provider-review/bitbucket-read.ts | 39 ++++-- 2 files changed, 155 insertions(+), 12 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 9fa1cf77ef0..150ca0fe952 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1371,6 +1371,134 @@ describe('listInbox', () => { expect(cursor).toBeUndefined(); }); + it('serves the rest of the inbox when a cached repository is no longer served', async () => { + // The cached set can name a repository the provider has deleted since its + // last sync: its collection answers 404, which ends that repository only. + const rowsBySlug: Record = { + 'repo-a': minutesFixture(100, 2), + }; + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: ['repo-a', 'repo-gone'].map((slug, index) => ({ + id: `87654321-4321-4321-4321-${String(index).padStart(12, '0')}`, + workspaceUuid: WORKSPACE.uuid, + name: slug, + fullName: `acme/${slug}`, + private: true, + defaultBranch: 'main', + })), + syncedAt: '2026-09-06T00:00:00.000Z', + }); + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const match = parsed.pathname.match(/^\/2\.0\/repositories\/acme\/([^/]+)\/pullrequests$/); + const rows = match ? rowsBySlug[match[1]] : undefined; + if (!match || !rows) { + return jsonResponse({ type: 'error', error: { message: 'Repository not found' } }, 404); + } + return jsonResponse({ + pagelen: 50, + values: rows.map(row => ({ + ...row, + source: { repository: { full_name: `acme/${match[1]}` } }, + destination: { repository: { full_name: `acme/${match[1]}` } }, + })), + next: null, + }); + }); + + const page = await listInbox(ORG_OWNER); + + expect(page.items.map(item => item.ref && 'prId' in item.ref && item.ref.prId)).toEqual([ + 101, 100, + ]); + expect(page.nextCursor).toBeNull(); + }); + + it('ends a repository whose continuation page no longer exists', async () => { + // Load one is promised a second page of repo-b. Twenty of repo-b's rows + // leave the collection before load two, so the page it was promised is now + // past the end and answers 404: the remaining repositories must still page, + // and repo-b must not be asked again. + const rowsBySlug: Record = { + 'repo-a': Array.from({ length: 60 }, (_, index) => + inboxPr(100 + index, `2026-09-02T00:${String(59 - index).padStart(2, '0')}:00+00:00`) + ), + 'repo-b': Array.from({ length: 60 }, (_, index) => + inboxPr(200 + index, `2026-09-02T01:${String(59 - index).padStart(2, '0')}:00+00:00`) + ), + }; + const requests: InboxRequest[] = []; + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: Object.keys(rowsBySlug).map((slug, index) => ({ + id: `87654321-4321-4321-4321-${String(index).padStart(12, '0')}`, + workspaceUuid: WORKSPACE.uuid, + name: slug, + fullName: `acme/${slug}`, + private: true, + defaultBranch: 'main', + })), + syncedAt: '2026-09-06T00:00:00.000Z', + }); + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const match = parsed.pathname.match(/^\/2\.0\/repositories\/acme\/([^/]+)\/pullrequests$/); + const rows = match ? rowsBySlug[match[1]] : undefined; + if (!match || !rows) return jsonResponse({ pagelen: 50, values: [], next: null }); + const page = Number(parsed.searchParams.get('page') ?? '1'); + requests.push({ slug: match[1], query: parsed.searchParams.get('q'), page: String(page) }); + const values = rows.slice((page - 1) * 50, page * 50); + if (values.length === 0) { + return jsonResponse( + { type: 'error', error: { message: 'Page number does not exist' } }, + 404 + ); + } + return jsonResponse({ + pagelen: 50, + values: values.map(row => ({ + ...row, + source: { repository: { full_name: `acme/${match[1]}` } }, + destination: { repository: { full_name: `acme/${match[1]}` } }, + })), + next: rows.length > page * 50 ? 'https://api.bitbucket.org/2.0/next?page=2' : null, + }); + }); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 5; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (load === 0) { + // Load one served repo-b's newest 50 rows and was promised page two. + expect(served).toHaveLength(50); + expect(requests.filter(request => request.slug === 'repo-b')).toHaveLength(1); + rowsBySlug['repo-b'] = rowsBySlug['repo-b'].slice(0, 40); + } + if (cursor === undefined) break; + } + + expect(cursor).toBeUndefined(); + expect(served).toHaveLength(110); + expect(new Set(served).size).toBe(110); + expect(served.slice(0, 50)).toEqual(Array.from({ length: 50 }, (_, index) => 200 + index)); + expect(served.slice(50)).toEqual(Array.from({ length: 60 }, (_, index) => 100 + index)); + expect(requests.filter(request => request.slug === 'repo-b')).toHaveLength(2); + }); + it('reads one provider page per repository that still has rows', async () => { // 12 repositories × 5 open PRs: the first load reads one page per // repository, and the continuation reads only the two with rows left. diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 844924e492e..f3a476720bb 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -993,7 +993,9 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * current page is already fully served). A pull request updated after the * first load sits above the ceiling and appears on a fresh inbox load; one * that stops being open leaves the collection and shifts the rows after it up, - * which the provider's page-number paging cannot prevent. + * which the provider's page-number paging cannot prevent. A repository or a + * page the provider no longer serves ends that repository's walk rather than + * failing the whole inbox. */ export async function listInbox( owner: BitbucketReviewOwner, @@ -1036,18 +1038,31 @@ export async function listInbox( let page = position?.pages.get(key)?.page ?? 1; let served = position?.pages.get(key)?.served ?? 0; for (let reads = 1; ; reads += 1) { - const payload = await requestBitbucketJson( - access, - `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(repository.slug)}/pullrequests`, - { - query: { - pagelen: BITBUCKET_PAGE_SIZE, - page, - q: providerQuery, - sort: '-updated_on', - }, + let payload: unknown; + try { + payload = await requestBitbucketJson( + access, + `/2.0/repositories/${encodeURIComponent(access.workspace.slug)}/${encodeURIComponent(repository.slug)}/pullrequests`, + { + query: { + pagelen: BITBUCKET_PAGE_SIZE, + page, + q: providerQuery, + sort: '-updated_on', + }, + } + ); + } catch (error) { + // A repository the cached set still names but the provider no + // longer serves, and a page an earlier load was promised for a + // collection that has since shrunk, are both that repository's + // end: the rest of the inbox still loads, and the repository + // stops appearing in later continuations. + if (classifyBitbucketError(error).kind === 'not_found') { + return { key, page, rows: [], fromRow: 0, hadNext: false }; } - ); + throw error; + } const parsedPage = BitbucketPageSchema.safeParse(payload); if (!parsedPage.success) { throw new BitbucketReviewError('retryable', 'Bitbucket returned an unexpected page.'); From 82a35e8bfcd5e5df09f26327c05accbc0594470f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 22:58:45 +0000 Subject: [PATCH 08/11] fix(provider-review): walk the Bitbucket inbox in repository windows A workspace whose cached repository set is larger than one cursor carries is now walked one window at a time, so every cached repository is served instead of the walk stopping at the first window. A provider row without a comparable timestamp is left out of the merged order instead of ending the walk before the rows behind it, and the omitted-next regression test now records provider requests so a page past a repository's last page fails it. --- .../provider-review/bitbucket-read.test.ts | 110 ++++++++++++++++-- .../src/lib/provider-review/bitbucket-read.ts | 92 +++++++++++---- 2 files changed, 173 insertions(+), 29 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 150ca0fe952..aa3d927601c 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1026,11 +1026,12 @@ describe('listInbox', () => { .digest('base64url')}`; /** A continuation token in the shape the read layer issues. */ - const inboxCursor = (identity: string, ceiling: string, pages = Buffer.alloc(2)) => + const inboxCursor = (identity: string, ceiling: string, pages = Buffer.alloc(2), window = 0) => createSignedToken({ kind: 'bitbucket-inbox-cursor', identity, ceiling, + window, pages: pages.toString('base64url'), }); @@ -1330,6 +1331,7 @@ describe('listInbox', () => { })), syncedAt: '2026-09-06T00:00:00.000Z', }); + const requests: InboxRequest[] = []; fetchMock.mockImplementation(async (url: string | URL) => { const full = url.toString(); if (full.includes('token-service.example.com')) { @@ -1340,6 +1342,7 @@ describe('listInbox', () => { const rows = match ? rowsBySlug[match[1]] : undefined; if (!match || !rows) return jsonResponse({ pagelen: 50, values: [] }); const page = Number(parsed.searchParams.get('page') ?? '1'); + requests.push({ slug: match[1], query: parsed.searchParams.get('q'), page: String(page) }); const values = rows.slice((page - 1) * 50, page * 50); if (values.length === 0) { // What Bitbucket answers for a page past the collection. @@ -1369,6 +1372,13 @@ describe('listInbox', () => { expect(new Set(served).size).toBe(151); expect(served).toHaveLength(151); expect(cursor).toBeUndefined(); + // A page past either repository's last page answers 404: the walk must + // never ask for one, so the omitted `next` link — not the provider error — + // is what ends a repository. + for (const slug of ['repo-a', 'repo-b']) { + const pages = requests.filter(request => request.slug === slug).map(request => request.page); + expect(new Set(pages)).toEqual(new Set(['1', '2'])); + } }); it('serves the rest of the inbox when a cached repository is no longer served', async () => { @@ -1499,6 +1509,65 @@ describe('listInbox', () => { expect(requests.filter(request => request.slug === 'repo-b')).toHaveLength(2); }); + it('leaves a row without a comparable timestamp out of the walk', async () => { + // The provider can answer a row with no `updated_on`. It has no place in + // the merged newest-first order and no comparable position for a + // continuation, so it is left out while every orderable row is served. + const rows = Array.from({ length: 60 }, (_, index) => + inboxPr(100 + index, `2026-09-02T00:${String(59 - index).padStart(2, '0')}:00+00:00`) + ); + const values = [ + ...rows.slice(0, 50), + { ...inboxPr(900, ''), updated_on: null }, + ...rows.slice(50), + ]; + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: [ + { + id: '87654321-4321-4321-4321-000000000000', + workspaceUuid: WORKSPACE.uuid, + name: 'repo', + fullName: 'acme/repo', + private: true, + defaultBranch: 'main', + }, + ], + syncedAt: '2026-09-06T00:00:00.000Z', + }); + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const page = Number(parsed.searchParams.get('page') ?? '1'); + return jsonResponse( + values.length > page * 50 + ? { + pagelen: 50, + values: values.slice((page - 1) * 50, page * 50), + next: 'https://api.bitbucket.org/2.0/next?page=2', + } + : { pagelen: 50, values: values.slice((page - 1) * 50, page * 50) } + ); + }); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 4; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (cursor === undefined) break; + } + + expect(served).toEqual(Array.from({ length: 60 }, (_, index) => 100 + index)); + expect(cursor).toBeUndefined(); + }); + it('reads one provider page per repository that still has rows', async () => { // 12 repositories × 5 open PRs: the first load reads one page per // repository, and the continuation reads only the two with rows left. @@ -1527,7 +1596,10 @@ describe('listInbox', () => { expect(requests.filter(request => request.query === 'state="OPEN"')).toHaveLength(12); }); - it('caps the workspace fan-out at the enumerated repository set', async () => { + it('walks every cached repository across repository windows', async () => { + // One more repository than a cursor carries: the first window reads the + // first 300, and the continuation moves to the next window instead of the + // inbox stopping at the first slice. const slugs = Array.from({ length: 301 }, (_, repo) => `repo-${repo}`); const rowsBySlug: Record = {}; for (const slug of slugs) { @@ -1536,11 +1608,33 @@ describe('listInbox', () => { const requests: InboxRequest[] = []; serveInbox(rowsBySlug, requests); - const page = await listInbox(ORG_OWNER); - const expected = [...slugs].sort().slice(0, 300); - expect(requests.map(request => request.slug).sort()).toEqual(expected); - expect(page.items).toHaveLength(50); - expect(page.nextCursor).toBeTruthy(); + const first = await listInbox(ORG_OWNER); + const sorted = [...slugs].sort(); + expect([...new Set(requests.map(request => request.slug))].sort()).toEqual( + sorted.slice(0, 300) + ); + expect(first.items).toHaveLength(50); + expect(first.nextCursor).toBeTruthy(); + + const served = first.items.flatMap(item => + item.ref && 'prId' in item.ref ? [item.ref.prId] : [] + ); + let cursor: string | undefined = first.nextCursor!; + for (let load = 0; load < 12 && cursor; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + } + + // Every repository is served, including the one past the first window. + expect(new Set(served)).toEqual(new Set(Array.from({ length: 301 }, (_, index) => index + 1))); + expect(served).toHaveLength(301); + expect(new Set(requests.map(request => request.slug))).toEqual(new Set(sorted)); + // One row per repository: every load re-reads page one of the repositories + // in its window, and no repository is read past its first page. + expect(requests.every(request => request.page === '1')).toBe(true); }); it('keeps a maximum-size continuation within the receiving wire limit', async () => { @@ -1592,6 +1686,7 @@ describe('listInbox', () => { kind: 'bitbucket-inbox-cursor', identity: 'bitbucket-inbox:evil', ceiling: '2026-09-02T00:10:00.000+00:00', + window: 0, pages: Buffer.alloc(2).toString('base64url'), }); const page = await listInbox(ORG_OWNER, foreign); @@ -1624,6 +1719,7 @@ describe('listInbox', () => { kind: 'bitbucket-inbox-cursor', identity: inboxIdentity(['repo']), ceiling: '2026-09-02T00:10:00.000+00:00', + window: 0, pages: Buffer.alloc(2).toString('base64url'), iat: Math.floor(Date.now() / 1000) - 2 * 60 * 60, nonce: 'expired-nonce', diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index f3a476720bb..0934a2cde81 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -990,7 +990,10 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * page, skips them, and moves to the next page once the page is fully served. * The continuation advances monotonically, never moves back to the first page, * and reads at most one provider page per repository per load (two when the - * current page is already fully served). A pull request updated after the + * current page is already fully served). A workspace with more cached + * repositories than one cursor can carry is walked one window of repositories + * at a time, so every cached repository is covered instead of the walk stopping + * at the first window. A pull request updated after the * first load sits above the ceiling and appears on a fresh inbox load; one * that stops being open leaves the collection and shifts the rows after it up, * which the provider's page-number paging cannot prevent. A repository or a @@ -1003,26 +1006,26 @@ export async function listInbox( ): Promise { try { const access = await authorizeWorkspace(owner); - const repositories = [...access.repositories] - .sort((left, right) => - left.fullName.toLowerCase() < right.fullName.toLowerCase() - ? -1 - : left.fullName.toLowerCase() > right.fullName.toLowerCase() - ? 1 - : 0 - ) - .slice(0, MAX_INBOX_REPOSITORIES); + const cachedRepositories = [...access.repositories].sort((left, right) => + left.fullName.toLowerCase() < right.fullName.toLowerCase() + ? -1 + : left.fullName.toLowerCase() > right.fullName.toLowerCase() + ? 1 + : 0 + ); const repositoryFingerprint = createHash('sha256') - .update(repositories.map(repository => repository.fullName.toLowerCase()).join('\0')) + .update(cachedRepositories.map(repository => repository.fullName.toLowerCase()).join('\0')) .digest('base64url'); const identity = owner.type === 'organization' ? `bitbucket-inbox:${access.workspace.slug}:${owner.organizationId}:${owner.userId}:${repositoryFingerprint}` : `bitbucket-inbox:${access.workspace.slug}:user:${owner.userId}:${repositoryFingerprint}`; + const position = decodeInboxCursor(cursor, identity, cachedRepositories); + const windowIndex = position?.window ?? 0; + const repositories = inboxRepositoryWindow(cachedRepositories, windowIndex); const repositoryNames = new Set( repositories.map(repository => repository.fullName.toLowerCase()) ); - const position = decodeInboxCursor(cursor, identity, repositories); const providerQuery = position ? `state="OPEN" AND updated_on <= ${position.ceiling}` : 'state="OPEN"'; @@ -1116,12 +1119,29 @@ export async function listInbox( ); } const hasMore = [...pages.values()].some(state => state.page !== 0); + const nextWindow = windowIndex + 1; + const carriesNextWindow = + !hasMore && inboxRepositoryWindow(cachedRepositories, nextWindow).length > 0; + let nextCursor: string | null = null; + if (ceiling !== null) { + if (hasMore) { + nextCursor = encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); + } else if (carriesNextWindow) { + // This window is served in full: the walk moves to the next slice of + // the cached repositories, so a workspace larger than one window is + // still covered instead of stopping at the first slice. + nextCursor = encodeInboxCursor( + identity, + nextWindow, + ceiling, + new Map(), + inboxRepositoryWindow(cachedRepositories, nextWindow) + ); + } + } return { items: window.map(candidate => candidate.item), - nextCursor: - hasMore && ceiling !== null - ? encodeInboxCursor(identity, ceiling, pages, repositories) - : null, + nextCursor, }; } catch (error) { throw classifyInboxError(error); @@ -1157,7 +1177,12 @@ function inboxCeiling(reads: InboxPageRead[]): string | null { return ceiling; } -/** The inbox rows of one provider page, in provider order. */ +/** + * The inbox rows of one provider page, in provider order. A row the provider + * gives without a parseable `updated_on` has no place in the merged newest-first + * order and no comparable position for a continuation, so it is left out + * instead of sorting as the oldest row. + */ function inboxItemsFrom( values: unknown[], allowedFullNames: ReadonlySet @@ -1168,13 +1193,15 @@ function inboxItemsFrom( if (!parsed.success) continue; const ref = inboxRefFrom(parsed.data, allowedFullNames); if (ref?.platform !== 'bitbucket') continue; + const updatedAt = parsed.data.updated_on ?? ''; + if (Number.isNaN(Date.parse(updatedAt))) continue; items.push({ ref, title: parsed.data.title, author: mapUser(parsed.data.author ?? null), state: mapPullRequestState(parsed.data.state), draft: parsed.data.draft === true, - updatedAt: parsed.data.updated_on ?? '', + updatedAt, }); } return items; @@ -1205,6 +1232,7 @@ const inboxCursorPayloadSchema = z.object({ kind: z.literal(INBOX_CURSOR_KIND), identity: z.string().min(1), ceiling: z.string().min(1), + window: z.number().int().min(0), pages: z.string().max(800), }); @@ -1236,17 +1264,34 @@ const DONE: InboxPageState = { page: 0, served: 0 }; type InboxCursorPosition = { ceiling: string; + window: number; pages: ReadonlyMap; }; +/** + * The slice of the sorted cached repositories one walk covers: the cursor's + * window index times the per-window bound. A workspace with more cached + * repositories than one window is walked window by window, so every cached + * repository is served within the cursor's wire limit. + */ +function inboxRepositoryWindow( + repositories: BitbucketWorkspaceRepositoryAccess['repositories'], + window: number +): BitbucketWorkspaceRepositoryAccess['repositories'] { + const start = window * MAX_INBOX_REPOSITORIES; + return repositories.slice(start, start + MAX_INBOX_REPOSITORIES); +} + /** * The inbox cursor is a signed continuation token carrying the caller and * repository-set identity, the provider timestamp ceiling the collection was - * read at, and two packed bytes per repository: its page number and how many - * leading rows of that page are served. + * read at, the repository window the walk is on, and two packed bytes per + * repository of that window: its page number and how many leading rows of that + * page are served. */ function encodeInboxCursor( identity: string, + window: number, ceiling: string, pages: ReadonlyMap, repositories: BitbucketWorkspaceRepositoryAccess['repositories'] @@ -1261,6 +1306,7 @@ function encodeInboxCursor( kind: INBOX_CURSOR_KIND, identity, ceiling, + window, pages: pageState.toString('base64url'), }); } @@ -1268,7 +1314,7 @@ function encodeInboxCursor( function decodeInboxCursor( cursor: string | undefined, identity: string, - repositories: BitbucketWorkspaceRepositoryAccess['repositories'] + cachedRepositories: BitbucketWorkspaceRepositoryAccess['repositories'] ): InboxCursorPosition | null { if (!cursor) return null; const payload = verifySignedToken(cursor, { @@ -1276,6 +1322,8 @@ function decodeInboxCursor( parse: value => { const parsed = inboxCursorPayloadSchema.safeParse(value); if (!parsed.success || parsed.data.identity !== identity) return null; + const repositories = inboxRepositoryWindow(cachedRepositories, parsed.data.window); + if (repositories.length === 0) return null; const pageState = Buffer.from(parsed.data.pages, 'base64url'); if (pageState.length !== repositories.length * 2) return null; const pages = new Map(); @@ -1288,7 +1336,7 @@ function decodeInboxCursor( served, }); } - return { ceiling: parsed.data.ceiling, pages }; + return { ceiling: parsed.data.ceiling, window: parsed.data.window, pages }; }, }); return payload; From a74138a153672ff9d6732b4f94b3075794216f8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:05:52 +0000 Subject: [PATCH 09/11] fix(provider-review): set each inbox window's ceiling when its walk starts A repository window's rows are read for the first time when the walk reaches that window, so the previous window's newest timestamp cannot bound them: a pull request newer than it stayed outside the collection for the whole walk. Each window now starts without a ceiling, records the newest timestamp of its own first load, and filters later loads of that window with it. --- .../provider-review/bitbucket-read.test.ts | 44 +++++++++++++++++++ .../src/lib/provider-review/bitbucket-read.ts | 41 +++++++++-------- 2 files changed, 67 insertions(+), 18 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index aa3d927601c..6c70602dd70 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1568,6 +1568,50 @@ describe('listInbox', () => { expect(cursor).toBeUndefined(); }); + it('starts each repository window with its own ceiling', async () => { + // The repository in the second window carries the newest row of the + // workspace. Its rows were never read under the first window's ceiling, so + // that ceiling must not filter them out. + const slugs = Array.from( + { length: 301 }, + (_, index) => `repo-${String(index).padStart(3, '0')}` + ); + const rowsBySlug: Record = {}; + for (const slug of slugs) { + rowsBySlug[slug] = [ + inboxPr( + slugs.indexOf(slug) + 1, + slug === 'repo-300' ? '2026-09-03T00:00:00+00:00' : '2026-09-02T00:00:00+00:00', + `acme/${slug}` + ), + ]; + } + const requests: InboxRequest[] = []; + serveInbox(rowsBySlug, requests); + + const served: number[] = []; + let cursor: string | undefined; + for (let load = 0; load < 12; load += 1) { + const page = await listInbox(ORG_OWNER, cursor); + served.push( + ...page.items.flatMap(item => (item.ref && 'prId' in item.ref ? [item.ref.prId] : [])) + ); + cursor = page.nextCursor ?? undefined; + if (cursor === undefined) break; + } + + expect(cursor).toBeUndefined(); + expect(new Set(served)).toEqual(new Set(Array.from({ length: 301 }, (_, index) => index + 1))); + expect(served).toHaveLength(301); + // The second window is read without the first window's ceiling. + const secondWindowRequest = requests.at(-1); + expect(secondWindowRequest).toMatchObject({ + slug: 'repo-300', + query: 'state="OPEN"', + page: '1', + }); + }); + it('reads one provider page per repository that still has rows', async () => { // 12 repositories × 5 open PRs: the first load reads one page per // repository, and the continuation reads only the two with rows left. diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 0934a2cde81..d364f50fdb7 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -980,10 +980,11 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * PRs of the integration's repositories, newest first. * * The fan-out sorts across repositories, so one inbox page cannot map onto a - * single provider page. A signed continuation carries a fixed provider ceiling - * — the newest `updated_on` the first load saw, so every row that existed then - * stays inside the collection — and, per repository, the provider page it is - * on and how many leading rows of that page are already served. + * single provider page. A signed continuation carries the provider ceiling of + * the window it is walking — the newest `updated_on` that window's first load + * saw, so every row of the window that existed then stays inside the + * collection — and, per repository, the provider page it is on and how many + * leading rows of that page are already served. * * Rows are emitted in the merged order, so a repository's served rows are * always the leading rows of its current page: the next load re-reads that @@ -993,12 +994,13 @@ function classifyInboxError(error: unknown): BitbucketReviewError { * current page is already fully served). A workspace with more cached * repositories than one cursor can carry is walked one window of repositories * at a time, so every cached repository is covered instead of the walk stopping - * at the first window. A pull request updated after the - * first load sits above the ceiling and appears on a fresh inbox load; one - * that stops being open leaves the collection and shifts the rows after it up, - * which the provider's page-number paging cannot prevent. A repository or a - * page the provider no longer serves ends that repository's walk rather than - * failing the whole inbox. + * at the first window; each window starts its own ceiling, because its rows + * were never read under the previous window's. A pull request updated after + * the window's first load sits above the ceiling and appears on a fresh inbox + * load; one that stops being open leaves the collection and shifts the rows + * after it up, which the provider's page-number paging cannot prevent. A + * repository or a page the provider no longer serves ends that repository's + * walk rather than failing the whole inbox. */ export async function listInbox( owner: BitbucketReviewOwner, @@ -1026,7 +1028,7 @@ export async function listInbox( const repositoryNames = new Set( repositories.map(repository => repository.fullName.toLowerCase()) ); - const providerQuery = position + const providerQuery = position?.ceiling ? `state="OPEN" AND updated_on <= ${position.ceiling}` : 'state="OPEN"'; let window: InboxCandidate[] = []; @@ -1127,13 +1129,15 @@ export async function listInbox( if (hasMore) { nextCursor = encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); } else if (carriesNextWindow) { - // This window is served in full: the walk moves to the next slice of - // the cached repositories, so a workspace larger than one window is - // still covered instead of stopping at the first slice. + // This window is served in full: the walk moves to the next slice of the + // cached repositories, so a workspace larger than one window is still + // covered. The next window sets its own ceiling from its first load: + // its rows were never read under this window's ceiling, and one newer + // than it would otherwise stay outside the collection for the whole walk. nextCursor = encodeInboxCursor( identity, nextWindow, - ceiling, + null, new Map(), inboxRepositoryWindow(cachedRepositories, nextWindow) ); @@ -1231,7 +1235,8 @@ const INBOX_PAGE_READS_PER_LOAD = 2; const inboxCursorPayloadSchema = z.object({ kind: z.literal(INBOX_CURSOR_KIND), identity: z.string().min(1), - ceiling: z.string().min(1), + /** The current window's ceiling; null starts a window whose walk begins now. */ + ceiling: z.string().min(1).nullable(), window: z.number().int().min(0), pages: z.string().max(800), }); @@ -1263,7 +1268,7 @@ type InboxPageState = { page: number; served: number }; const DONE: InboxPageState = { page: 0, served: 0 }; type InboxCursorPosition = { - ceiling: string; + ceiling: string | null; window: number; pages: ReadonlyMap; }; @@ -1292,7 +1297,7 @@ function inboxRepositoryWindow( function encodeInboxCursor( identity: string, window: number, - ceiling: string, + ceiling: string | null, pages: ReadonlyMap, repositories: BitbucketWorkspaceRepositoryAccess['repositories'] ): string { From 15e6cde8167cfcf008587fcda4766081e7bcb63e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:11:59 +0000 Subject: [PATCH 10/11] fix(provider-review): hand the inbox walk on from a window that served nothing A window whose repositories the provider no longer serves, or whose pages are empty, produces no timestamp and therefore no ceiling. The walk now moves to the next repository window in that case, so the windows behind it are still read instead of the inbox ending at an empty slice. --- .../provider-review/bitbucket-read.test.ts | 54 +++++++++++++++++++ .../src/lib/provider-review/bitbucket-read.ts | 39 ++++++++------ 2 files changed, 76 insertions(+), 17 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 6c70602dd70..164de3ef65b 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1612,6 +1612,60 @@ describe('listInbox', () => { }); }); + it('hands the walk to the next window when a window serves nothing', async () => { + // The first window holds repositories the provider no longer serves, so it + // produces no rows and no ceiling; the repository behind it still has to be + // read. + const slugs = Array.from( + { length: 301 }, + (_, index) => `repo-${String(index).padStart(3, '0')}` + ); + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: slugs.map((slug, index) => ({ + id: `87654321-4321-4321-4321-${String(index).padStart(12, '0')}`, + workspaceUuid: WORKSPACE.uuid, + name: slug, + fullName: `acme/${slug}`, + private: true, + defaultBranch: 'main', + })), + syncedAt: '2026-09-06T00:00:00.000Z', + }); + const requests: InboxRequest[] = []; + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const match = parsed.pathname.match(/^\/2\.0\/repositories\/acme\/([^/]+)\/pullrequests$/); + requests.push({ + slug: match?.[1] ?? '', + query: parsed.searchParams.get('q'), + page: parsed.searchParams.get('page'), + }); + if (match?.[1] !== 'repo-300') { + return jsonResponse({ type: 'error', error: { message: 'Repository not found' } }, 404); + } + return jsonResponse({ + pagelen: 1, + values: [inboxPr(999, '2026-09-02T00:00:00+00:00', 'acme/repo-300')], + next: null, + }); + }); + + const first = await listInbox(ORG_OWNER); + expect(first.items).toEqual([]); + expect(first.nextCursor).toBeTruthy(); + + const second = await listInbox(ORG_OWNER, first.nextCursor!); + expect(second.items).toHaveLength(1); + expect(second.items[0].ref).toMatchObject({ repoSlug: 'repo-300', prId: 999 }); + expect(second.nextCursor).toBeNull(); + expect(requests.at(-1)).toMatchObject({ slug: 'repo-300', query: 'state="OPEN"' }); + }); + it('reads one provider page per repository that still has rows', async () => { // 12 repositories × 5 open PRs: the first load reads one page per // repository, and the continuation reads only the two with rows left. diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index d364f50fdb7..33955933c00 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -1125,23 +1125,28 @@ export async function listInbox( const carriesNextWindow = !hasMore && inboxRepositoryWindow(cachedRepositories, nextWindow).length > 0; let nextCursor: string | null = null; - if (ceiling !== null) { - if (hasMore) { - nextCursor = encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); - } else if (carriesNextWindow) { - // This window is served in full: the walk moves to the next slice of the - // cached repositories, so a workspace larger than one window is still - // covered. The next window sets its own ceiling from its first load: - // its rows were never read under this window's ceiling, and one newer - // than it would otherwise stay outside the collection for the whole walk. - nextCursor = encodeInboxCursor( - identity, - nextWindow, - null, - new Map(), - inboxRepositoryWindow(cachedRepositories, nextWindow) - ); - } + if (hasMore) { + // Every row of this window that can be ordered carries a timestamp, so a + // ceiling exists whenever the window still has rows to serve. + nextCursor = + ceiling === null + ? null + : encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); + } else if (carriesNextWindow) { + // This window is served in full: the walk moves to the next slice of the + // cached repositories, so a workspace larger than one window is still + // covered. The next window sets its own ceiling from its first load: + // its rows were never read under this window's ceiling, and one newer + // than it would otherwise stay outside the collection for the whole walk. + // A window that served nothing (every repository gone or empty) still + // hands the walk on, so the windows behind it are read. + nextCursor = encodeInboxCursor( + identity, + nextWindow, + null, + new Map(), + inboxRepositoryWindow(cachedRepositories, nextWindow) + ); } return { items: window.map(candidate => candidate.item), From 4e051d9dfa9a6e4ed2e7b4d7eaf15a4cb44d4b28 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Igor=20=C5=A0=C4=87eki=C4=87?= <4062236+iscekic@users.noreply.github.com> Date: Sat, 24 Oct 2026 23:19:02 +0000 Subject: [PATCH 11/11] fix(provider-review): keep the inbox walk moving across provider pages The continuation carries the window's provider timestamp bound whenever the window has produced one. Where it has not, the next load reads the window without that bound, so the walk reaches every remaining provider page and every repository window behind it. --- .../provider-review/bitbucket-read.test.ts | 49 +++++++++++++++++++ .../src/lib/provider-review/bitbucket-read.ts | 10 ++-- 2 files changed, 53 insertions(+), 6 deletions(-) diff --git a/apps/web/src/lib/provider-review/bitbucket-read.test.ts b/apps/web/src/lib/provider-review/bitbucket-read.test.ts index 164de3ef65b..b3cec1d6f22 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.test.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.test.ts @@ -1666,6 +1666,55 @@ describe('listInbox', () => { expect(requests.at(-1)).toMatchObject({ slug: 'repo-300', query: 'state="OPEN"' }); }); + it('continues a window whose pages carry no orderable row', async () => { + // Pages one and two answer rows without a timestamp and promise another + // page; page three holds the first orderable row. The walk must continue + // without a ceiling instead of ending at the first unorderable page. + mockReadCachedRepositories.mockResolvedValue({ + status: 'available', + repositories: [ + { + id: '87654321-4321-4321-4321-000000000000', + workspaceUuid: WORKSPACE.uuid, + name: 'repo', + fullName: 'acme/repo', + private: true, + defaultBranch: 'main', + }, + ], + syncedAt: '2026-09-06T00:00:00.000Z', + }); + const requests: InboxRequest[] = []; + fetchMock.mockImplementation(async (url: string | URL) => { + const full = url.toString(); + if (full.includes('token-service.example.com')) { + return jsonResponse({ status: 'available', token: 'at-mock-token', workspace: WORKSPACE }); + } + const parsed = new URL(full); + const page = Number(parsed.searchParams.get('page') ?? '1'); + requests.push({ slug: 'repo', query: parsed.searchParams.get('q'), page: String(page) }); + if (page < 3) { + return jsonResponse({ + pagelen: 1, + values: [{ ...inboxPr(900 + page, ''), updated_on: null }], + next: `https://api.bitbucket.org/2.0/repositories/acme/repo/pullrequests?page=${page + 1}`, + }); + } + return jsonResponse({ pagelen: 1, values: [inboxPr(999, '2026-09-02T00:00:00+00:00')] }); + }); + + const first = await listInbox(ORG_OWNER); + expect(first.items).toEqual([]); + expect(first.nextCursor).toBeTruthy(); + + const second = await listInbox(ORG_OWNER, first.nextCursor!); + expect(second.items.map(item => item.ref)).toEqual([ + { platform: 'bitbucket', workspace: 'acme', repoSlug: 'repo', prId: 999 }, + ]); + expect(second.nextCursor).toBeNull(); + expect(requests.at(-1)).toMatchObject({ slug: 'repo', query: 'state="OPEN"', page: '3' }); + }); + it('reads one provider page per repository that still has rows', async () => { // 12 repositories × 5 open PRs: the first load reads one page per // repository, and the continuation reads only the two with rows left. diff --git a/apps/web/src/lib/provider-review/bitbucket-read.ts b/apps/web/src/lib/provider-review/bitbucket-read.ts index 33955933c00..ef55d5c8005 100644 --- a/apps/web/src/lib/provider-review/bitbucket-read.ts +++ b/apps/web/src/lib/provider-review/bitbucket-read.ts @@ -1126,12 +1126,10 @@ export async function listInbox( !hasMore && inboxRepositoryWindow(cachedRepositories, nextWindow).length > 0; let nextCursor: string | null = null; if (hasMore) { - // Every row of this window that can be ordered carries a timestamp, so a - // ceiling exists whenever the window still has rows to serve. - nextCursor = - ceiling === null - ? null - : encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); + // A page whose rows all lack a timestamp leaves no ceiling yet; the + // continuation then carries none and the next load reads on unfiltered, + // so the rest of the window is still walked instead of ending here. + nextCursor = encodeInboxCursor(identity, windowIndex, ceiling, pages, repositories); } else if (carriesNextWindow) { // This window is served in full: the walk moves to the next slice of the // cached repositories, so a workspace larger than one window is still