diff --git a/templates/content/actions/_database-source-utils.test.ts b/templates/content/actions/_database-source-utils.test.ts index 3d67522a57..44cf639ce5 100644 --- a/templates/content/actions/_database-source-utils.test.ts +++ b/templates/content/actions/_database-source-utils.test.ts @@ -33,6 +33,7 @@ import { builderAuthoritativeRawBodyHash, bulkChunkSizeForColumnCount, builderCmsEntryAlreadyRepresented, + builderCmsSourceContinuationIsCurrent, builderExecutionIsProvablyLocallyBlockedUnsent, canRefreshLocallyBlockedBuilderReview, buildMockBodyChange, @@ -100,6 +101,41 @@ function item(id: string, title: string): ContentDatabaseItem { } describe("database source helpers", () => { + it("accepts only the persisted Builder continuation offset", () => { + const metadataJson = JSON.stringify({ + sourceFetchState: "fetching", + lastReadHasMore: true, + lastReadNextOffset: 400, + activeReadSourceRowIds: ["entry-1"], + }); + + expect(builderCmsSourceContinuationIsCurrent(metadataJson, 400)).toBe(true); + expect(builderCmsSourceContinuationIsCurrent(metadataJson, 100)).toBe( + false, + ); + expect(builderCmsSourceContinuationIsCurrent(metadataJson, 0)).toBe(false); + expect( + builderCmsSourceContinuationIsCurrent( + JSON.stringify({ + sourceFetchState: "fetching", + lastReadHasMore: true, + lastReadNextOffset: 400, + }), + 400, + ), + ).toBe(false); + expect( + builderCmsSourceContinuationIsCurrent( + JSON.stringify({ + sourceFetchState: "idle", + lastReadHasMore: false, + lastReadNextOffset: 580, + }), + 400, + ), + ).toBe(false); + }); + it("selects document bodies only for explicitly heavy Builder snapshots", () => { expect(sourceSnapshotDocumentSelection(false)).not.toHaveProperty( "content", @@ -291,6 +327,44 @@ describe("database source helpers", () => { }); }); + it("clears the Builder refresh claim when read metadata is finalized", () => { + const metadata = JSON.parse( + serializeBuilderCmsSourceReadMetadataRecord({ + sourceTable: "agent-native-blog-article-test", + readState: "live", + entryCount: 580, + matchedRowCount: 580, + existingMetadataJson: JSON.stringify({ + builderContinuationClaimId: "claim-1", + builderContinuationClaimOffset: 400, + }), + completedBuilderContinuationClaimId: "claim-1", + }), + ); + + expect(metadata).not.toHaveProperty("builderContinuationClaimId"); + expect(metadata).not.toHaveProperty("builderContinuationClaimOffset"); + }); + + it("preserves a Builder refresh claim for unrelated metadata writers", () => { + const metadata = JSON.parse( + serializeBuilderCmsSourceReadMetadataRecord({ + sourceTable: "agent-native-blog-article-test", + readState: "live", + entryCount: 400, + matchedRowCount: 400, + existingMetadataJson: JSON.stringify({ + builderContinuationClaimId: "claim-1", + builderContinuationClaimOffset: 400, + builderContinuationClaimedAt: "2026-01-01T00:00:00.000Z", + }), + }), + ); + + expect(metadata.builderContinuationClaimId).toBe("claim-1"); + expect(metadata.builderContinuationClaimOffset).toBe(400); + }); + it("preserves existing Builder model fields during metadata rewrites", () => { const existingMetadataJson = JSON.stringify({ builderModelFields: [ diff --git a/templates/content/actions/_database-source-utils.ts b/templates/content/actions/_database-source-utils.ts index ae77298c4a..e2eaf8026a 100644 --- a/templates/content/actions/_database-source-utils.ts +++ b/templates/content/actions/_database-source-utils.ts @@ -162,6 +162,9 @@ type SourceMetadataRecord = { lastReadSuspiciousEmpty?: boolean; activeReadSourceRowIds?: string[]; sourceFetchState?: "idle" | "fetching" | "error" | string; + builderContinuationClaimId?: string; + builderContinuationClaimOffset?: number; + builderContinuationClaimedAt?: string; allowDraftWrites?: boolean; allowPublishWrites?: boolean; allowedWriteModes?: ContentDatabaseSourcePushMode[]; @@ -183,6 +186,145 @@ function parseObject( } } +export function builderCmsSourceContinuationIsCurrent( + metadataJson: string | null | undefined, + expectedOffset: number, +) { + const metadata = parseObject(metadataJson); + return ( + metadata?.sourceFetchState === "fetching" && + metadata.lastReadHasMore === true && + expectedOffset > 0 && + metadata.lastReadNextOffset === expectedOffset && + Array.isArray(metadata.activeReadSourceRowIds) && + metadata.activeReadSourceRowIds.length > 0 && + metadata.activeReadSourceRowIds.every((id) => typeof id === "string") + ); +} + +// Content's hosted request wall is 75 seconds. Keep the orphan-recovery lease +// well beyond that so an expired owner cannot still be executing mutations. +export const BUILDER_CMS_REFRESH_CLAIM_LEASE_MS = 30 * 60 * 1000; + +export async function claimBuilderCmsSourceRefresh(args: { + source: ContentDatabaseSourceRowDb; + expectedOffset?: number; + now?: string; +}) { + const metadata = + parseObject(args.source.metadataJson) ?? {}; + const now = args.now ?? new Date().toISOString(); + const claimedAt = metadata.builderContinuationClaimedAt + ? Date.parse(metadata.builderContinuationClaimedAt) + : Number.NaN; + const activeClaim = + !!metadata.builderContinuationClaimId && + Number.isFinite(claimedAt) && + Date.parse(now) - claimedAt < BUILDER_CMS_REFRESH_CLAIM_LEASE_MS; + if ( + activeClaim || + (args.expectedOffset !== undefined && + !builderCmsSourceContinuationIsCurrent( + args.source.metadataJson, + args.expectedOffset, + )) + ) { + return null; + } + + const claimId = nanoid(); + const claimedMetadataJson = JSON.stringify({ + ...metadata, + builderContinuationClaimId: claimId, + builderContinuationClaimOffset: args.expectedOffset, + builderContinuationClaimedAt: now, + }); + const [claimed] = await getDb() + .update(schema.contentDatabaseSources) + .set({ metadataJson: claimedMetadataJson }) + .where( + and( + eq(schema.contentDatabaseSources.id, args.source.id), + args.source.metadataJson === null + ? isNull(schema.contentDatabaseSources.metadataJson) + : eq( + schema.contentDatabaseSources.metadataJson, + args.source.metadataJson, + ), + ), + ) + .returning(); + return claimed ? { source: claimed, claimId } : null; +} + +export async function releaseBuilderCmsSourceRefreshClaim(args: { + sourceId: string; + claimId: string; +}) { + const db = getDb(); + for (let attempt = 0; attempt < 5; attempt += 1) { + const [current] = await db + .select({ metadataJson: schema.contentDatabaseSources.metadataJson }) + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, args.sourceId)) + .limit(1); + const metadata = parseObject(current?.metadataJson); + if (!metadata || metadata.builderContinuationClaimId !== args.claimId) { + return false; + } + delete metadata.builderContinuationClaimId; + delete metadata.builderContinuationClaimOffset; + delete metadata.builderContinuationClaimedAt; + const nextMetadataJson = JSON.stringify(metadata); + const released = await db + .update(schema.contentDatabaseSources) + .set({ metadataJson: nextMetadataJson }) + .where( + and( + eq(schema.contentDatabaseSources.id, args.sourceId), + eq(schema.contentDatabaseSources.metadataJson, current.metadataJson!), + ), + ) + .returning({ id: schema.contentDatabaseSources.id }); + if (released.length > 0) return true; + } + return false; +} + +export async function renewBuilderCmsSourceRefreshClaim(args: { + sourceId: string; + claimId: string; + now?: string; +}) { + const db = getDb(); + const now = args.now ?? new Date().toISOString(); + for (let attempt = 0; attempt < 5; attempt += 1) { + const [current] = await db + .select({ metadataJson: schema.contentDatabaseSources.metadataJson }) + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, args.sourceId)) + .limit(1); + const metadata = parseObject(current?.metadataJson); + if (!metadata || metadata.builderContinuationClaimId !== args.claimId) { + return false; + } + metadata.builderContinuationClaimedAt = now; + const nextMetadataJson = JSON.stringify(metadata); + const renewed = await db + .update(schema.contentDatabaseSources) + .set({ metadataJson: nextMetadataJson }) + .where( + and( + eq(schema.contentDatabaseSources.id, args.sourceId), + eq(schema.contentDatabaseSources.metadataJson, current.metadataJson!), + ), + ) + .returning({ id: schema.contentDatabaseSources.id }); + if (renewed.length > 0) return true; + } + return false; +} + function parseArray(value: string | null | undefined): T[] { if (!value) return []; try { @@ -3941,11 +4083,12 @@ export function serializeBuilderCmsSourceReadMetadataRecord(args: { suspiciousEmpty?: boolean; builderModelFields?: BuilderCmsModelFieldSummary[]; existingMetadataJson?: string | null; + completedBuilderContinuationClaimId?: string; }) { const existingMetadata = parseObject( args.existingMetadataJson, ); - return JSON.stringify({ + const metadata: SourceMetadataRecord = { ...builderCmsSourceMetadata(args.sourceTable), ...existingMetadata, builderModelFields: @@ -3970,7 +4113,17 @@ export function serializeBuilderCmsSourceReadMetadataRecord(args: { : args.readState === "error" ? "error" : "idle"), - }); + }; + if ( + args.completedBuilderContinuationClaimId && + existingMetadata?.builderContinuationClaimId === + args.completedBuilderContinuationClaimId + ) { + delete metadata.builderContinuationClaimId; + delete metadata.builderContinuationClaimOffset; + delete metadata.builderContinuationClaimedAt; + } + return JSON.stringify(metadata); } export function serializeSourceCapabilitiesRecord( @@ -5339,6 +5492,7 @@ export async function resyncBuilderCmsSourceSnapshot(args: { source: ContentDatabaseSourceRowDb; now: string; runFullRefresh?: boolean; + refreshClaimId?: string; }) { let { properties, response } = await sourceSetupPayload(args.database.id); const db = getDb(); @@ -5401,6 +5555,17 @@ export async function resyncBuilderCmsSourceSnapshot(args: { (builderRead.progress?.startOffset ?? 0) > 0); const builderEntries = builderRead.state === "live" ? builderRead.entries : []; + if ( + args.refreshClaimId && + !(await renewBuilderCmsSourceRefreshClaim({ + sourceId: args.source.id, + claimId: args.refreshClaimId, + })) + ) { + throw new Error( + "Builder source refresh claim was lost before snapshot mutation.", + ); + } let existingRows = await db .select() .from(schema.contentDatabaseSourceRows) @@ -5432,6 +5597,7 @@ export async function resyncBuilderCmsSourceSnapshot(args: { sourceFetchState: "error", syncState: "error", suspiciousEmpty: true, + refreshClaimId: args.refreshClaimId, }); return; } @@ -5597,6 +5763,7 @@ export async function resyncBuilderCmsSourceSnapshot(args: { sourceFetchState: hasMore ? "fetching" : "idle", syncState: hasMore ? "refreshing" : "idle", activeReadSourceRowIds: hasMore ? nextActiveReadSourceRowIds : undefined, + refreshClaimId: args.refreshClaimId, }); return; } @@ -5711,6 +5878,7 @@ export async function resyncBuilderCmsSourceSnapshot(args: { sourceFetchState: builderRead.state === "error" ? "error" : "idle", activeReadSourceRowIds: undefined, syncState: "idle", + refreshClaimId: args.refreshClaimId, }); } @@ -6053,58 +6221,87 @@ export async function updateBuilderCmsSourceReadMetadata(args: { syncState?: ContentDatabaseSourceSyncState; builderModelFields?: BuilderCmsModelFieldSummary[]; suspiciousEmpty?: boolean; + refreshClaimId?: string; }) { const db = getDb(); - const [currentSource] = await db - .select({ - capabilitiesJson: schema.contentDatabaseSources.capabilitiesJson, - metadataJson: schema.contentDatabaseSources.metadataJson, - lastSourceUpdatedAt: schema.contentDatabaseSources.lastSourceUpdatedAt, - }) - .from(schema.contentDatabaseSources) - .where(eq(schema.contentDatabaseSources.id, args.sourceId)) - .limit(1); - const nextJson = mergeBuilderCmsWriteSettingsIntoJson({ - sourceTable: args.sourceTable, - currentCapabilitiesJson: currentSource?.capabilitiesJson, - currentMetadataJson: currentSource?.metadataJson, - nextCapabilitiesJson: sourceCapabilitiesForType("builder-cms"), - nextMetadataJson: serializeBuilderCmsSourceReadMetadataRecord({ + for (let attempt = 0; attempt < 5; attempt += 1) { + const [currentSource] = await db + .select({ + capabilitiesJson: schema.contentDatabaseSources.capabilitiesJson, + metadataJson: schema.contentDatabaseSources.metadataJson, + lastSourceUpdatedAt: schema.contentDatabaseSources.lastSourceUpdatedAt, + }) + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, args.sourceId)) + .limit(1); + if (!currentSource) throw new Error("Builder source not found."); + const currentMetadata = parseObject( + currentSource.metadataJson, + ); + if ( + args.refreshClaimId && + currentMetadata?.builderContinuationClaimId !== args.refreshClaimId + ) { + throw new Error( + "Builder source refresh claim was lost before completion.", + ); + } + const nextJson = mergeBuilderCmsWriteSettingsIntoJson({ sourceTable: args.sourceTable, - readState: args.readState, - entryCount: args.entryCount, - matchedRowCount: args.matchedRowCount, - progress: args.progress, - sourceFetchState: args.sourceFetchState, - activeReadSourceRowIds: args.activeReadSourceRowIds, - suspiciousEmpty: args.suspiciousEmpty, - builderModelFields: args.builderModelFields, - existingMetadataJson: currentSource?.metadataJson, - }), - }); - await db - .update(schema.contentDatabaseSources) - .set({ - syncState: args.syncState ?? "linked", - freshness: - args.readState === "error" || - args.progress?.partial || - args.suspiciousEmpty - ? "stale" - : "fresh", - capabilitiesJson: nextJson.capabilitiesJson, - metadataJson: nextJson.metadataJson, - lastRefreshedAt: args.now, - lastSourceUpdatedAt: args.suspiciousEmpty - ? (currentSource?.lastSourceUpdatedAt ?? null) - : args.fetchedAt, - lastError: - args.readState === "error" || args.suspiciousEmpty - ? args.message - : null, - updatedAt: args.now, - }) - .where(eq(schema.contentDatabaseSources.id, args.sourceId)); + currentCapabilitiesJson: currentSource.capabilitiesJson, + currentMetadataJson: currentSource.metadataJson, + nextCapabilitiesJson: sourceCapabilitiesForType("builder-cms"), + nextMetadataJson: serializeBuilderCmsSourceReadMetadataRecord({ + sourceTable: args.sourceTable, + readState: args.readState, + entryCount: args.entryCount, + matchedRowCount: args.matchedRowCount, + progress: args.progress, + sourceFetchState: args.sourceFetchState, + activeReadSourceRowIds: args.activeReadSourceRowIds, + suspiciousEmpty: args.suspiciousEmpty, + builderModelFields: args.builderModelFields, + existingMetadataJson: currentSource.metadataJson, + completedBuilderContinuationClaimId: args.refreshClaimId, + }), + }); + const updated = await db + .update(schema.contentDatabaseSources) + .set({ + syncState: args.syncState ?? "linked", + freshness: + args.readState === "error" || + args.progress?.partial || + args.suspiciousEmpty + ? "stale" + : "fresh", + capabilitiesJson: nextJson.capabilitiesJson, + metadataJson: nextJson.metadataJson, + lastRefreshedAt: args.now, + lastSourceUpdatedAt: args.suspiciousEmpty + ? currentSource.lastSourceUpdatedAt + : args.fetchedAt, + lastError: + args.readState === "error" || args.suspiciousEmpty + ? args.message + : null, + updatedAt: args.now, + }) + .where( + and( + eq(schema.contentDatabaseSources.id, args.sourceId), + currentSource.metadataJson === null + ? isNull(schema.contentDatabaseSources.metadataJson) + : eq( + schema.contentDatabaseSources.metadataJson, + currentSource.metadataJson, + ), + ), + ) + .returning({ id: schema.contentDatabaseSources.id }); + if (updated.length > 0) return; + } + throw new Error("Builder source metadata changed repeatedly during refresh."); } export async function updateReadOnlySourceMetadata(args: { diff --git a/templates/content/actions/content-database-source-actions.test.ts b/templates/content/actions/content-database-source-actions.test.ts index c7cfd2fa11..b25876d6c6 100644 --- a/templates/content/actions/content-database-source-actions.test.ts +++ b/templates/content/actions/content-database-source-actions.test.ts @@ -373,6 +373,18 @@ describe("content database source actions", () => { }); }); + it("accepts a guarded Builder continuation offset", () => { + expect( + refreshSource.schema.parse({ + databaseId: "database", + expectedBuilderContinuationOffset: 400, + }), + ).toEqual({ + databaseId: "database", + expectedBuilderContinuationOffset: 400, + }); + }); + it("caps content database page sizes for bounded responses", () => { expect( normalizeContentDatabasePageOptions({ limit: 1, offset: -5 }), diff --git a/templates/content/actions/refresh-content-database-source.ts b/templates/content/actions/refresh-content-database-source.ts index 236eeff4b1..5d576815be 100644 --- a/templates/content/actions/refresh-content-database-source.ts +++ b/templates/content/actions/refresh-content-database-source.ts @@ -5,9 +5,11 @@ import { z } from "zod"; import type { ContentDatabaseSourceStatusResponse } from "../shared/api.js"; import { getContentDatabaseSourceAdapter } from "./_content-database-source-adapters.js"; import { + claimBuilderCmsSourceRefresh, getContentDatabaseSourceSnapshot, getContentDatabaseSourceSnapshotById, getExistingSourceForWrite, + releaseBuilderCmsSourceRefreshClaim, resyncBuilderCmsSourceSnapshot, resyncMockSourceSnapshot, resolveDatabaseForSourceMutation, @@ -33,6 +35,14 @@ export default defineAction({ .describe( "For paginated Builder CMS or Notion sources, read a bounded multi-page snapshot in this refresh.", ), + expectedBuilderContinuationOffset: z + .number() + .int() + .nonnegative() + .optional() + .describe( + "Internal Builder pagination guard. Continue only when the persisted source still expects this offset.", + ), }), run: async (args): Promise => { const database = await resolveDatabaseForSourceMutation(args); @@ -50,15 +60,30 @@ export default defineAction({ } const now = new Date().toISOString(); + let skippedOverlappingBuilderRefresh = false; if (source.sourceType === "mock-local") { await resyncMockSourceSnapshot({ database, source, now }); } else if (source.sourceType === "builder-cms") { - await resyncBuilderCmsSourceSnapshot({ - database, + const claimedSource = await claimBuilderCmsSourceRefresh({ source, - now, - runFullRefresh: args.fullRefresh === true, + expectedOffset: args.expectedBuilderContinuationOffset, }); + skippedOverlappingBuilderRefresh = !claimedSource; + if (claimedSource) { + await resyncBuilderCmsSourceSnapshot({ + database, + source: claimedSource.source, + now, + runFullRefresh: args.fullRefresh === true, + refreshClaimId: claimedSource.claimId, + }).catch(async (error: unknown) => { + await releaseBuilderCmsSourceRefreshClaim({ + sourceId: source.id, + claimId: claimedSource.claimId, + }); + throw error; + }); + } } else if (source.sourceType === "local-table") { // Read-only federated secondary; its rows are re-read on demand, nothing // to resync against the primary's local snapshot here. @@ -116,11 +141,13 @@ export default defineAction({ database: serializeDatabase(database), mode: "source-backed", summary: snapshot - ? builderFetching - ? `${snapshot.sourceName} fetched ${builderFetched ?? "some"} rows. Run refresh again to continue loading the remaining Builder rows.` - : snapshot.sourceType === "notion-database" - ? `${snapshot.sourceName} refreshed read-only from Notion.` - : `${snapshot.sourceName} resynced locally; field mappings and row identity now reflect the current database snapshot.` + ? skippedOverlappingBuilderRefresh + ? `${snapshot.sourceName} already advanced or has another refresh in progress; the current snapshot was preserved.` + : builderFetching + ? `${snapshot.sourceName} fetched ${builderFetched ?? "some"} rows. Run refresh again to continue loading the remaining Builder rows.` + : snapshot.sourceType === "notion-database" + ? `${snapshot.sourceName} refreshed read-only from Notion.` + : `${snapshot.sourceName} resynced locally; field mappings and row identity now reflect the current database snapshot.` : "Source metadata refreshed.", source: snapshot, }; diff --git a/templates/content/actions/resync-content-database-source.db.test.ts b/templates/content/actions/resync-content-database-source.db.test.ts index 414edf893a..c4e39162ed 100644 --- a/templates/content/actions/resync-content-database-source.db.test.ts +++ b/templates/content/actions/resync-content-database-source.db.test.ts @@ -551,6 +551,10 @@ let seedSourceRows: typeof import("./_database-source-utils.js").seedMockSourceR let materializeSourceFields: typeof import("./_database-source-utils.js").materializeSourceFieldPropertyValues; let getSnapshot: typeof import("./_database-source-utils.js").getContentDatabaseSourceSnapshotById; let getWriteSnapshot: typeof import("./_database-source-utils.js").getContentDatabaseSourceSnapshotForWrite; +let claimSourceRefresh: typeof import("./_database-source-utils.js").claimBuilderCmsSourceRefresh; +let releaseSourceRefreshClaim: typeof import("./_database-source-utils.js").releaseBuilderCmsSourceRefreshClaim; +let renewSourceRefreshClaim: typeof import("./_database-source-utils.js").renewBuilderCmsSourceRefreshClaim; +let updateSourceReadMetadata: typeof import("./_database-source-utils.js").updateBuilderCmsSourceReadMetadata; let hydrateQueuedBodies: typeof import("./_database-source-utils.js").processBuilderBodyHydrationQueue; let withBuilderBodyValues: typeof import("./_database-source-utils.js").withBuilderBodySourceValues; let seedCollabFromText: typeof import("@agent-native/core/collab").seedFromText; @@ -607,6 +611,14 @@ beforeAll(async () => { .getContentDatabaseSourceSnapshotById; getWriteSnapshot = (await import("./_database-source-utils.js")) .getContentDatabaseSourceSnapshotForWrite; + claimSourceRefresh = (await import("./_database-source-utils.js")) + .claimBuilderCmsSourceRefresh; + releaseSourceRefreshClaim = (await import("./_database-source-utils.js")) + .releaseBuilderCmsSourceRefreshClaim; + renewSourceRefreshClaim = (await import("./_database-source-utils.js")) + .renewBuilderCmsSourceRefreshClaim; + updateSourceReadMetadata = (await import("./_database-source-utils.js")) + .updateBuilderCmsSourceReadMetadata; hydrateQueuedBodies = (await import("./_database-source-utils.js")) .processBuilderBodyHydrationQueue; withBuilderBodyValues = (await import("./_database-source-utils.js")) @@ -628,6 +640,145 @@ afterAll(() => { } }); +it("atomically grants one Builder continuation claim per persisted offset", async () => { + const db = getDb(); + const now = new Date().toISOString(); + const metadataJson = JSON.stringify({ + sourceFetchState: "fetching", + lastReadHasMore: true, + lastReadNextOffset: 400, + activeReadSourceRowIds: ["entry-1"], + }); + await db.insert(schema.documents).values({ + id: "doc-continuation-claim", + ownerEmail: OWNER, + title: "Continuation claim", + content: "", + createdAt: now, + updatedAt: now, + }); + await db.insert(schema.contentDatabases).values({ + id: "db-continuation-claim", + ownerEmail: OWNER, + documentId: "doc-continuation-claim", + title: "Continuation claim", + createdAt: now, + updatedAt: now, + }); + await db.insert(schema.contentDatabaseSources).values({ + id: "source-continuation-claim", + ownerEmail: OWNER, + databaseId: "db-continuation-claim", + sourceType: "builder-cms", + sourceName: "Continuation claim", + sourceTable: BUILDER_CMS_SAFE_WRITE_MODEL, + syncState: "refreshing", + freshness: "stale", + metadataJson, + createdAt: now, + updatedAt: now, + }); + const [source] = await db + .select() + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, "source-continuation-claim")); + + const claims = await Promise.all([ + claimSourceRefresh({ source, expectedOffset: 400, now }), + claimSourceRefresh({ source, expectedOffset: 400, now }), + ]); + const granted = claims.filter((claim) => claim !== null); + expect(granted).toHaveLength(1); + expect( + JSON.parse(granted[0]!.source.metadataJson!).builderContinuationClaimOffset, + ).toBe(400); + + const claimedMetadata = JSON.parse(granted[0]!.source.metadataJson!); + claimedMetadata.writeMode = "publish_updates"; + await db + .update(schema.contentDatabaseSources) + .set({ metadataJson: JSON.stringify(claimedMetadata) }) + .where(eq(schema.contentDatabaseSources.id, source.id)); + await releaseSourceRefreshClaim({ + sourceId: source.id, + claimId: granted[0]!.claimId, + }); + const [released] = await db + .select({ metadataJson: schema.contentDatabaseSources.metadataJson }) + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, source.id)); + const releasedMetadata = JSON.parse(released.metadataJson!); + expect(releasedMetadata.writeMode).toBe("publish_updates"); + expect(releasedMetadata).not.toHaveProperty("builderContinuationClaimId"); + + const orphanedMetadataJson = JSON.stringify({ + ...releasedMetadata, + builderContinuationClaimId: "orphaned-claim", + builderContinuationClaimOffset: 400, + builderContinuationClaimedAt: "2026-01-01T00:00:00.000Z", + }); + await db + .update(schema.contentDatabaseSources) + .set({ metadataJson: orphanedMetadataJson }) + .where(eq(schema.contentDatabaseSources.id, source.id)); + const [orphanedSource] = await db + .select() + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, source.id)); + const recovered = await claimSourceRefresh({ + source: orphanedSource, + expectedOffset: 400, + now: "2026-01-01T00:31:00.000Z", + }); + expect(recovered?.claimId).toBeTruthy(); + expect(recovered?.claimId).not.toBe("orphaned-claim"); + expect( + await renewSourceRefreshClaim({ + sourceId: source.id, + claimId: "orphaned-claim", + now: "2026-01-01T00:31:30.000Z", + }), + ).toBe(false); + await releaseSourceRefreshClaim({ + sourceId: source.id, + claimId: recovered!.claimId, + }); + + const [readySource] = await db + .select() + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, source.id)); + const fenced = await claimSourceRefresh({ + source: readySource, + expectedOffset: 400, + now: "2026-01-01T00:12:00.000Z", + }); + await expect( + updateSourceReadMetadata({ + sourceId: source.id, + sourceTable: BUILDER_CMS_SAFE_WRITE_MODEL, + readState: "live", + entryCount: 580, + matchedRowCount: 580, + fetchedAt: "2026-01-01T00:12:30.000Z", + now: "2026-01-01T00:12:30.000Z", + message: null, + refreshClaimId: "not-the-owner", + }), + ).rejects.toThrow("refresh claim was lost"); + const [stillFenced] = await db + .select({ metadataJson: schema.contentDatabaseSources.metadataJson }) + .from(schema.contentDatabaseSources) + .where(eq(schema.contentDatabaseSources.id, source.id)); + expect(JSON.parse(stillFenced.metadataJson!).builderContinuationClaimId).toBe( + fenced!.claimId, + ); + await releaseSourceRefreshClaim({ + sourceId: source.id, + claimId: fenced!.claimId, + }); +}); + it("preserves an established source snapshot when Builder unexpectedly returns zero entries", async () => { builderReadMock.mode = "full"; builderReadMock.calls = []; diff --git a/templates/content/app/components/editor/database/DatabaseView.tsx b/templates/content/app/components/editor/database/DatabaseView.tsx index b5148bcfb5..92fe914899 100644 --- a/templates/content/app/components/editor/database/DatabaseView.tsx +++ b/templates/content/app/components/editor/database/DatabaseView.tsx @@ -1115,12 +1115,20 @@ function DatabaseTable({ ), }; const runSourceRefresh = useCallback( - (sourceId: string, onError?: () => void) => { + ( + sourceId: string, + onError?: () => void, + expectedBuilderContinuationOffset?: number, + ) => { if (!acquireDatabaseSourceOperation(refreshSourceInFlightRef, sourceId)) { return false; } refreshSource.mutate( - { documentId: document.id, sourceId }, + { + documentId: document.id, + sourceId, + expectedBuilderContinuationOffset, + }, { onError, onSettled: () => { @@ -1197,10 +1205,13 @@ function DatabaseTable({ continuationKey, ); if ( - !runSourceRefresh(candidate.id, () => - setBuilderContinuationClientErrorKeys((current) => - addUniqueKey(current, continuationKey), - ), + !runSourceRefresh( + candidate.id, + () => + setBuilderContinuationClientErrorKeys((current) => + addUniqueKey(current, continuationKey), + ), + candidate.metadata.lastReadNextOffset, ) ) { autoContinueBuilderSourceRef.current.delete(continuationKey); @@ -1273,10 +1284,13 @@ function DatabaseTable({ return; } if ( - runSourceRefresh(candidate.id, () => - setBuilderContinuationClientErrorKeys((current) => - addUniqueKey(current, continuationKey), - ), + runSourceRefresh( + candidate.id, + () => + setBuilderContinuationClientErrorKeys((current) => + addUniqueKey(current, continuationKey), + ), + candidate.metadata.lastReadNextOffset, ) ) { builderContinuationWatchdogRef.current.set( @@ -2545,10 +2559,13 @@ function DatabaseTable({ continuationKey, ); builderContinuationWatchdogRef.current.set(continuationKey, 0); - runSourceRefresh(builderSource.id, () => - setBuilderContinuationClientErrorKeys((current) => - addUniqueKey(current, continuationKey), - ), + runSourceRefresh( + builderSource.id, + () => + setBuilderContinuationClientErrorKeys((current) => + addUniqueKey(current, continuationKey), + ), + builderSource.metadata.lastReadNextOffset, ); }} /> diff --git a/templates/content/changelog/2026-07-24-builder-collection-loading-no-longer-restarts-from-the-begin.md b/templates/content/changelog/2026-07-24-builder-collection-loading-no-longer-restarts-from-the-begin.md new file mode 100644 index 0000000000..10e8638870 --- /dev/null +++ b/templates/content/changelog/2026-07-24-builder-collection-loading-no-longer-restarts-from-the-begin.md @@ -0,0 +1,6 @@ +--- +type: fixed +date: 2026-07-24 +--- + +Builder collection loading no longer restarts from the beginning after reaching the final page. diff --git a/templates/content/shared/api.ts b/templates/content/shared/api.ts index 27455077f1..b580f6c405 100644 --- a/templates/content/shared/api.ts +++ b/templates/content/shared/api.ts @@ -957,6 +957,7 @@ export interface RefreshContentDatabaseSourceRequest { documentId?: string; sourceId?: string; fullRefresh?: boolean; + expectedBuilderContinuationOffset?: number; } export interface DisconnectContentDatabaseSourceRequest {