diff --git a/kits/bigquery-firestore-export/src/handlers.ts b/kits/bigquery-firestore-export/src/handlers.ts index 13ddddd535..3c5ad0bb30 100644 --- a/kits/bigquery-firestore-export/src/handlers.ts +++ b/kits/bigquery-firestore-export/src/handlers.ts @@ -23,6 +23,7 @@ import { createTransferConfig, type DataTransferClient, getTransferConfig, + PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX, updateTransferConfig, } from "./dts"; import type { ResolvedBigqueryFirestoreExportConfig } from "./export-config"; @@ -64,6 +65,13 @@ async function ensureNotificationTopic(ctx: HandlerContext): Promise { } } +function isPartitioningFieldRemovalError(err: unknown): err is Error { + return ( + err instanceof Error && + err.message.includes(PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX) + ); +} + async function storeTransferConfig( ctx: HandlerContext, transferConfig: Awaited> @@ -105,9 +113,10 @@ export async function handleUpsertTransferConfig( ctx.config.transferConfigName ); if (!linked) { - throw new Error( - `Transfer config not found: ${ctx.config.transferConfigName}` - ); + // Only a redeploy with a corrected TRANSFER_CONFIG_NAME can resolve this, + // so retrying the task cannot help. + logs.linkedTransferConfigMissing(ctx.config.transferConfigName); + return; } await storeTransferConfig(ctx, linked); return; @@ -128,14 +137,21 @@ export async function handleUpsertTransferConfig( const transferConfigName = existing.docs[0].data().name; if (typeof transferConfigName !== "string" || !transferConfigName) { throw new Error( - `Existing transfer config document in ${ctx.config.firestoreCollection} is missing required 'name' field.` + `Existing transfer config document in ${ctx.config.firestoreCollection} is missing required 'name' field. Delete the document so a new scheduled query is created, then redeploy.` ); } - const updated = await updateTransferConfig( - ctx.dataTransfer, - transferConfigName, - ctx.config - ); - await storeTransferConfig(ctx, updated); + try { + const updated = await updateTransferConfig( + ctx.dataTransfer, + transferConfigName, + ctx.config + ); + await storeTransferConfig(ctx, updated); + } catch (err) { + if (!isPartitioningFieldRemovalError(err)) throw err; + // The guard rejects the update while building the request, so nothing was + // sent and a retry hits the same guard. + logs.partitioningFieldRemovalAborted(err.message); + } } diff --git a/kits/bigquery-firestore-export/src/logs.ts b/kits/bigquery-firestore-export/src/logs.ts index 0848691139..96e93ac1e7 100644 --- a/kits/bigquery-firestore-export/src/logs.ts +++ b/kits/bigquery-firestore-export/src/logs.ts @@ -140,6 +140,19 @@ export function partitioningFieldRemovalAttempted( }); } +export function linkedTransferConfigMissing(name: string): void { + logger.error( + "The scheduled query named by TRANSFER_CONFIG_NAME does not exist, so nothing was linked. Set it to a scheduled query that exists in this project and redeploy, or clear it to have this deployment create its own.", + { name } + ); +} + +// The reason carries the remediation, and an Error passed as structured data +// serialises to {}, so it goes in the message. +export function partitioningFieldRemovalAborted(reason: string): void { + logger.error(`Stopped without updating the scheduled query. ${reason}`); +} + export function topicCreated(name: string): void { logger.info("Created Pub/Sub topic for transfer notifications", { name }); } diff --git a/kits/bigquery-firestore-export/tests/handlers.test.ts b/kits/bigquery-firestore-export/tests/handlers.test.ts index 6e9cfcc1ac..e33115f91d 100644 --- a/kits/bigquery-firestore-export/tests/handlers.test.ts +++ b/kits/bigquery-firestore-export/tests/handlers.test.ts @@ -23,13 +23,23 @@ const mocks = vi.hoisted(() => ({ getTransferConfig: vi.fn(), updateTransferConfig: vi.fn(), handleTransferRunMessage: vi.fn(), + linkedTransferConfigMissing: vi.fn(), + partitioningFieldRemovalAborted: vi.fn(), })); -vi.mock("../src/dts", () => ({ - createTransferConfig: mocks.createTransferConfig, - getTransferConfig: mocks.getTransferConfig, - updateTransferConfig: mocks.updateTransferConfig, -})); +// The error constants come from the real module so the handler's prefix match +// is tested against the message the guard actually throws. +vi.mock("../src/dts", async (importOriginal) => { + const actual = await importOriginal(); + return { + createTransferConfig: mocks.createTransferConfig, + getTransferConfig: mocks.getTransferConfig, + updateTransferConfig: mocks.updateTransferConfig, + PARTITIONING_FIELD_REMOVAL_ERROR: actual.PARTITIONING_FIELD_REMOVAL_ERROR, + PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX: + actual.PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX, + }; +}); vi.mock("../src/helper", () => ({ handleTransferRunMessage: mocks.handleTransferRunMessage, @@ -41,10 +51,13 @@ vi.mock("../src/helper", () => ({ vi.mock("../src/logs", () => ({ complete: vi.fn(), error: vi.fn(), + linkedTransferConfigMissing: mocks.linkedTransferConfigMissing, + partitioningFieldRemovalAborted: mocks.partitioningFieldRemovalAborted, start: vi.fn(), topicCreated: vi.fn(), })); +import { PARTITIONING_FIELD_REMOVAL_ERROR } from "../src/dts"; import { handleUpsertTransferConfig } from "../src/handlers"; const config = resolveConfig({ @@ -167,4 +180,70 @@ describe("handleUpsertTransferConfig", () => { ...linked, }); }); + + test("reports a missing linked transfer config without retrying", async () => { + mocks.getTransferConfig.mockResolvedValue(null); + const { ctx, set } = makeContext({ + transferConfigName: "projects/p/locations/us/transferConfigs/gone", + }); + + await expect(handleUpsertTransferConfig(ctx)).resolves.toBeUndefined(); + + expect(mocks.linkedTransferConfigMissing).toHaveBeenCalledWith( + "projects/p/locations/us/transferConfigs/gone" + ); + expect(set).not.toHaveBeenCalled(); + }); + + test("reports a rejected partitioning-field removal without retrying", async () => { + mocks.updateTransferConfig.mockRejectedValue( + new Error(PARTITIONING_FIELD_REMOVAL_ERROR) + ); + const { ctx, set } = makeContext({ + existing: { + empty: false, + docs: [ + { + data: () => ({ + name: "projects/p/locations/us/transferConfigs/config-1", + }), + }, + ], + }, + }); + + await expect(handleUpsertTransferConfig(ctx)).resolves.toBeUndefined(); + + expect(mocks.partitioningFieldRemovalAborted).toHaveBeenCalledWith( + PARTITIONING_FIELD_REMOVAL_ERROR + ); + expect(set).not.toHaveBeenCalled(); + }); + + test("rethrows any other update failure so the task retries", async () => { + mocks.updateTransferConfig.mockRejectedValue( + new Error( + "bigquerydatatransfer.googleapis.com is temporarily unavailable" + ) + ); + const { ctx, set } = makeContext({ + existing: { + empty: false, + docs: [ + { + data: () => ({ + name: "projects/p/locations/us/transferConfigs/config-1", + }), + }, + ], + }, + }); + + await expect(handleUpsertTransferConfig(ctx)).rejects.toThrow( + "temporarily unavailable" + ); + + expect(mocks.partitioningFieldRemovalAborted).not.toHaveBeenCalled(); + expect(set).not.toHaveBeenCalled(); + }); });