diff --git a/kits/bigquery-firestore-export/src/dts.ts b/kits/bigquery-firestore-export/src/dts.ts index 5f509b360..66c04f96c 100644 --- a/kits/bigquery-firestore-export/src/dts.ts +++ b/kits/bigquery-firestore-export/src/dts.ts @@ -15,6 +15,7 @@ */ import * as bigqueryDataTransfer from "@google-cloud/bigquery-data-transfer"; +import { PermanentConfigurationError } from "./errors"; import type { ResolvedBigqueryFirestoreExportConfig } from "./export-config"; import * as logs from "./logs"; @@ -31,6 +32,13 @@ export const PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX = PACKAGE_PARTITIONING_ERROR; export const PARTITIONING_FIELD_REMOVAL_ERROR = `${PACKAGE_PARTITIONING_ERROR}. The BigQuery Data Transfer API does not support clearing this parameter once it has been set. To change partitioning, create a new transfer config with the desired setting.`; +const STRUCTURE_ERROR_PREFIX = "Transfer config has invalid structure:"; +// This check is only reached from the stored-document branch of the upsert, +// which runs when TRANSFER_CONFIG_NAME is unset, so that param is never the +// remedy here. +const STRUCTURE_REMEDIATION = + "Only scheduled queries are supported. Delete this deployment's document from the configs collection so a new scheduled query is created, then redeploy."; + function isNotFoundError(err: unknown): boolean { return ( typeof err === "object" && @@ -43,18 +51,18 @@ function isNotFoundError(err: unknown): boolean { function transferConfigFields(config: TransferConfig) { const fields = config.params?.fields; if (!fields) { - throw new Error( - "Transfer config has invalid structure: missing params.fields" + throw new PermanentConfigurationError( + `${STRUCTURE_ERROR_PREFIX} missing params.fields. ${STRUCTURE_REMEDIATION}` ); } if (!fields.query) { - throw new Error( - "Transfer config has invalid structure: missing params.fields.query" + throw new PermanentConfigurationError( + `${STRUCTURE_ERROR_PREFIX} missing params.fields.query. ${STRUCTURE_REMEDIATION}` ); } if (!fields.destination_table_name_template) { - throw new Error( - "Transfer config has invalid structure: missing params.fields.destination_table_name_template" + throw new PermanentConfigurationError( + `${STRUCTURE_ERROR_PREFIX} missing params.fields.destination_table_name_template. ${STRUCTURE_REMEDIATION}` ); } @@ -136,7 +144,11 @@ export async function constructUpdateTransferConfigRequest( config: ResolvedBigqueryFirestoreExportConfig ): Promise { const transferConfig = await getTransferConfig(client, transferConfigName); - if (!transferConfig) throw new Error("Transfer config not found"); + if (!transferConfig) { + throw new PermanentConfigurationError( + `Transfer config not found: ${transferConfigName}. The scheduled query recorded for this deployment no longer exists in BigQuery. Delete its document from the configs collection so a new query is created, then redeploy.` + ); + } const fields = transferConfigFields(transferConfig); const updatedConfig = JSON.parse( @@ -169,7 +181,7 @@ export async function constructUpdateTransferConfigRequest( transferConfigName, existingPartitioningField ); - throw new Error(PARTITIONING_FIELD_REMOVAL_ERROR); + throw new PermanentConfigurationError(PARTITIONING_FIELD_REMOVAL_ERROR); } updateMask.push("params"); updatedFields.partitioning_field ??= {}; diff --git a/kits/bigquery-firestore-export/src/errors.ts b/kits/bigquery-firestore-export/src/errors.ts new file mode 100644 index 000000000..fd6ec8a43 --- /dev/null +++ b/kits/bigquery-firestore-export/src/errors.ts @@ -0,0 +1,28 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** + * A deploy-time misconfiguration that no retry can resolve. Cloud Tasks retries + * every non-2xx response, so the upsert task handler reports these and returns + * rather than throwing. The message must name the misconfiguration and the + * action the user has to take. + */ +export class PermanentConfigurationError extends Error { + constructor(message: string) { + super(message); + this.name = "PermanentConfigurationError"; + } +} diff --git a/kits/bigquery-firestore-export/src/export-config.ts b/kits/bigquery-firestore-export/src/export-config.ts index c39f434d3..c08233a70 100644 --- a/kits/bigquery-firestore-export/src/export-config.ts +++ b/kits/bigquery-firestore-export/src/export-config.ts @@ -16,6 +16,8 @@ import type { Expression } from "firebase-functions/params"; +import { PermanentConfigurationError } from "./errors"; + /** Log levels supported by the original extension. */ export type LogLevel = "debug" | "info" | "warn" | "error" | "silent"; @@ -77,7 +79,9 @@ export interface DeployTimeOptions { function required(value: string, field: string): string { const normalized = value.trim(); if (!normalized) { - throw new Error(`${field} must be a non-empty string.`); + throw new PermanentConfigurationError( + `${field} must be a non-empty string. Set it in the deployment configuration, then redeploy.` + ); } return normalized; } @@ -96,7 +100,9 @@ export function resolveConfig( const logLevel = config.logLevel ?? "info"; if (!["debug", "info", "warn", "error", "silent"].includes(logLevel)) { - throw new Error(`Unsupported logLevel: ${logLevel}`); + throw new PermanentConfigurationError( + `Unsupported logLevel: ${logLevel}. Use one of debug, info, warn, error, silent, then redeploy.` + ); } return { diff --git a/kits/bigquery-firestore-export/src/handlers.ts b/kits/bigquery-firestore-export/src/handlers.ts index 13ddddd53..19646b102 100644 --- a/kits/bigquery-firestore-export/src/handlers.ts +++ b/kits/bigquery-firestore-export/src/handlers.ts @@ -25,6 +25,7 @@ import { getTransferConfig, updateTransferConfig, } from "./dts"; +import { PermanentConfigurationError } from "./errors"; import type { ResolvedBigqueryFirestoreExportConfig } from "./export-config"; import { handleTransferRunMessage, parseTransferConfigName } from "./helper"; import * as logs from "./logs"; @@ -93,10 +94,30 @@ export async function handleMessagePublished( } } -/** Idempotently creates, links, or updates this deployment's DTS config. */ +/** + * Runs the upsert and stops permanently on a misconfiguration. + * + * Cloud Tasks retries every non-2xx response and the enqueued lifecycle task + * has no channel for reporting deploy status, so a failure that no retry can + * resolve is logged at error level and the task returns successfully. + * + * The context arrives as a factory so that config resolution runs inside the + * try. Resolving it in the caller would put param validation outside this + * catch, which is the one failure most certain that no retry can fix. + */ export async function handleUpsertTransferConfig( - ctx: HandlerContext + getCtx: () => HandlerContext ): Promise { + try { + await upsertTransferConfig(getCtx()); + } catch (err) { + if (!(err instanceof PermanentConfigurationError)) throw err; + logs.upsertTransferConfigAborted(err); + } +} + +/** Idempotently creates, links, or updates this deployment's DTS config. */ +async function upsertTransferConfig(ctx: HandlerContext): Promise { await ensureNotificationTopic(ctx); if (ctx.config.transferConfigName) { @@ -105,8 +126,8 @@ export async function handleUpsertTransferConfig( ctx.config.transferConfigName ); if (!linked) { - throw new Error( - `Transfer config not found: ${ctx.config.transferConfigName}` + throw new PermanentConfigurationError( + `Transfer config not found: ${ctx.config.transferConfigName}. Set TRANSFER_CONFIG_NAME to a scheduled query that exists in this project, then redeploy. To have this deployment create its own scheduled query instead, clear TRANSFER_CONFIG_NAME and delete this instance's document from the ${ctx.config.firestoreCollection} collection, otherwise a previously linked query is updated in place.` ); } await storeTransferConfig(ctx, linked); @@ -127,8 +148,8 @@ 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.` + throw new PermanentConfigurationError( + `Existing transfer config document ${existing.docs[0].id} in ${ctx.config.firestoreCollection} is missing required 'name' field. Delete that document so this deployment creates a new scheduled query, or restore its 'name' field, then redeploy.` ); } diff --git a/kits/bigquery-firestore-export/src/index.ts b/kits/bigquery-firestore-export/src/index.ts index e3a57260d..3e01c05f5 100644 --- a/kits/bigquery-firestore-export/src/index.ts +++ b/kits/bigquery-firestore-export/src/index.ts @@ -122,5 +122,5 @@ export const upsertTransferConfig = onTaskDispatched( memory: "1GiB", retryConfig: { maxAttempts: 5, minBackoffSeconds: 30 }, }, - () => handleUpsertTransferConfig(getContext()) + () => handleUpsertTransferConfig(getContext) ); diff --git a/kits/bigquery-firestore-export/src/lib.ts b/kits/bigquery-firestore-export/src/lib.ts index 234faf45e..20a72f291 100644 --- a/kits/bigquery-firestore-export/src/lib.ts +++ b/kits/bigquery-firestore-export/src/lib.ts @@ -30,6 +30,7 @@ export { type TransferConfig, updateTransferConfig, } from "./dts"; +export { PermanentConfigurationError } from "./errors"; export { type BigqueryFirestoreExportConfig, type DeployTimeOptions, diff --git a/kits/bigquery-firestore-export/src/logs.ts b/kits/bigquery-firestore-export/src/logs.ts index 084869113..401cfe5ba 100644 --- a/kits/bigquery-firestore-export/src/logs.ts +++ b/kits/bigquery-firestore-export/src/logs.ts @@ -140,6 +140,13 @@ export function partitioningFieldRemovalAttempted( }); } +export function upsertTransferConfigAborted(err: Error): void { + logger.error( + "Could not set up the scheduled query. This is a configuration problem that a retry cannot fix, so the deployment task has stopped. Fix the configuration and deploy again.", + { reason: err.message } + ); +} + export function topicCreated(name: string): void { logger.info("Created Pub/Sub topic for transfer notifications", { name }); } diff --git a/kits/bigquery-firestore-export/tests/dts.test.ts b/kits/bigquery-firestore-export/tests/dts.test.ts index 060ff3ae3..83c217f8a 100644 --- a/kits/bigquery-firestore-export/tests/dts.test.ts +++ b/kits/bigquery-firestore-export/tests/dts.test.ts @@ -21,6 +21,7 @@ import { type DataTransferClient, PARTITIONING_FIELD_REMOVAL_ERROR, } from "../src/dts"; +import { PermanentConfigurationError } from "../src/errors"; import { resolveConfig } from "../src/export-config"; const config = resolveConfig({ @@ -120,5 +121,65 @@ describe("constructUpdateTransferConfigRequest", () => { { ...config, partitioningField: undefined } ) ).rejects.toThrow(PARTITIONING_FIELD_REMOVAL_ERROR); + + await expect( + constructUpdateTransferConfigRequest( + client, + "projects/p/locations/us/transferConfigs/c", + { ...config, partitioningField: undefined } + ) + ).rejects.toBeInstanceOf(PermanentConfigurationError); + }); + + test.for([ + ["missing params.fields", {}], + [ + "missing params.fields.query", + { params: { fields: { destination_table_name_template: {} } } }, + ], + [ + "missing params.fields.destination_table_name_template", + { params: { fields: { query: { stringValue: "SELECT 1" } } } }, + ], + ] as const)( + "reports a config the kit cannot update as permanent: %s", + async ([expectedMessage, shape]) => { + const client = clientWithTransferConfig({ + name: "projects/p/locations/us/transferConfigs/c", + ...shape, + }); + + const rejects = expect( + constructUpdateTransferConfigRequest( + client, + "projects/p/locations/us/transferConfigs/c", + config + ) + ).rejects; + + await rejects.toBeInstanceOf(PermanentConfigurationError); + await rejects.toThrow( + `Transfer config has invalid structure: ${expectedMessage}` + ); + await rejects.toThrow("Only scheduled queries are supported"); + await rejects.toThrow( + "Delete this deployment's document from the configs collection" + ); + } + ); + + test("reports a vanished transfer config as permanent", async () => { + const notFound = Object.assign(new Error("5 NOT_FOUND"), { code: 5 }); + const client = { + getTransferConfig: vi.fn().mockRejectedValue(notFound), + } as unknown as DataTransferClient; + + await expect( + constructUpdateTransferConfigRequest( + client, + "projects/p/locations/us/transferConfigs/gone", + config + ) + ).rejects.toThrow(PermanentConfigurationError); }); }); diff --git a/kits/bigquery-firestore-export/tests/export-config.test.ts b/kits/bigquery-firestore-export/tests/export-config.test.ts index 728ae2883..ac289afdc 100644 --- a/kits/bigquery-firestore-export/tests/export-config.test.ts +++ b/kits/bigquery-firestore-export/tests/export-config.test.ts @@ -15,6 +15,7 @@ */ import { describe, expect, test } from "vitest"; +import { PermanentConfigurationError } from "../src/errors"; import { type BigqueryFirestoreExportConfig, resolveConfig, @@ -62,7 +63,7 @@ describe("resolveConfig", () => { "rejects an empty %s", (field) => { expect(() => resolveConfig({ ...minimal, [field]: " " })).toThrow( - `${field} must be a non-empty string.` + `${field} must be a non-empty string. Set it in the deployment configuration, then redeploy.` ); } ); @@ -75,4 +76,16 @@ describe("resolveConfig", () => { }) ).toThrow("Unsupported logLevel: verbose"); }); + + test("reports validation failures as permanent", () => { + expect(() => resolveConfig({ ...minimal, datasetId: " " })).toThrow( + PermanentConfigurationError + ); + expect(() => + resolveConfig({ + ...minimal, + logLevel: "verbose" as BigqueryFirestoreExportConfig["logLevel"], + }) + ).toThrow(PermanentConfigurationError); + }); }); diff --git a/kits/bigquery-firestore-export/tests/handlers.test.ts b/kits/bigquery-firestore-export/tests/handlers.test.ts index 6e9cfcc1a..87545c3b0 100644 --- a/kits/bigquery-firestore-export/tests/handlers.test.ts +++ b/kits/bigquery-firestore-export/tests/handlers.test.ts @@ -15,8 +15,11 @@ */ import { beforeEach, describe, expect, test, vi } from "vitest"; -import type { HandlerContext } from "../src/handlers"; +import { PARTITIONING_FIELD_REMOVAL_ERROR } from "../src/dts"; +import { PermanentConfigurationError } from "../src/errors"; import { resolveConfig } from "../src/export-config"; +import type { HandlerContext } from "../src/handlers"; +import * as logs from "../src/logs"; const mocks = vi.hoisted(() => ({ createTransferConfig: vi.fn(), @@ -25,7 +28,8 @@ const mocks = vi.hoisted(() => ({ handleTransferRunMessage: vi.fn(), })); -vi.mock("../src/dts", () => ({ +vi.mock("../src/dts", async (importOriginal) => ({ + ...(await importOriginal()), createTransferConfig: mocks.createTransferConfig, getTransferConfig: mocks.getTransferConfig, updateTransferConfig: mocks.updateTransferConfig, @@ -43,6 +47,7 @@ vi.mock("../src/logs", () => ({ error: vi.fn(), start: vi.fn(), topicCreated: vi.fn(), + upsertTransferConfigAborted: vi.fn(), })); import { handleUpsertTransferConfig } from "../src/handlers"; @@ -59,16 +64,19 @@ const config = resolveConfig({ }); function makeContext(options: { - existing?: { empty: boolean; docs: Array<{ data(): object }> }; + existing?: { empty: boolean; docs: Array<{ id?: string; data(): object }> }; + existingError?: Error; transferConfigName?: string; }) { const set = vi.fn(); - const get = vi.fn().mockResolvedValue( - options.existing ?? { - empty: true, - docs: [], - } - ); + const get = options.existingError + ? vi.fn().mockRejectedValue(options.existingError) + : vi.fn().mockResolvedValue( + options.existing ?? { + empty: true, + docs: [], + } + ); const collection = vi.fn(() => ({ doc: vi.fn(() => ({ set })), where: vi.fn(() => ({ @@ -105,7 +113,7 @@ describe("handleUpsertTransferConfig", () => { mocks.createTransferConfig.mockResolvedValue(created); const { ctx, set } = makeContext({}); - await handleUpsertTransferConfig(ctx); + await handleUpsertTransferConfig(() => ctx); expect(mocks.createTransferConfig).toHaveBeenCalledWith( ctx.dataTransfer, @@ -136,7 +144,7 @@ describe("handleUpsertTransferConfig", () => { }, }); - await handleUpsertTransferConfig(ctx); + await handleUpsertTransferConfig(() => ctx); expect(mocks.updateTransferConfig).toHaveBeenCalledOnce(); expect(mocks.getTransferConfig).not.toHaveBeenCalled(); @@ -155,7 +163,7 @@ describe("handleUpsertTransferConfig", () => { transferConfigName: linked.name, }); - await handleUpsertTransferConfig(ctx); + await handleUpsertTransferConfig(() => ctx); expect(mocks.getTransferConfig).toHaveBeenCalledWith( ctx.dataTransfer, @@ -168,3 +176,170 @@ describe("handleUpsertTransferConfig", () => { }); }); }); + +describe("handleUpsertTransferConfig permanent failures", () => { + const aborted = vi.mocked(logs.upsertTransferConfigAborted); + + test("stops when the linked transfer config does not exist", async () => { + mocks.getTransferConfig.mockResolvedValue(null); + const { ctx, set } = makeContext({ + transferConfigName: "projects/p/locations/us/transferConfigs/missing", + }); + + await expect( + handleUpsertTransferConfig(() => ctx) + ).resolves.toBeUndefined(); + + expect(set).not.toHaveBeenCalled(); + expect(aborted).toHaveBeenCalledOnce(); + expect(aborted.mock.calls[0][0].message).toContain( + "Transfer config not found: projects/p/locations/us/transferConfigs/missing" + ); + expect(aborted.mock.calls[0][0].message).toContain( + "Set TRANSFER_CONFIG_NAME" + ); + expect(aborted.mock.calls[0][0].message).toContain( + "clear TRANSFER_CONFIG_NAME and delete this instance's document from the transferConfigs collection" + ); + }); + + test("stops when the partitioning field is cleared on an existing config", async () => { + mocks.updateTransferConfig.mockRejectedValue( + new PermanentConfigurationError(PARTITIONING_FIELD_REMOVAL_ERROR) + ); + const { ctx, set } = makeContext({ + existing: { + empty: false, + docs: [ + { + id: "config-1", + data: () => ({ + name: "projects/p/locations/us/transferConfigs/config-1", + }), + }, + ], + }, + }); + + await expect( + handleUpsertTransferConfig(() => ctx) + ).resolves.toBeUndefined(); + + expect(set).not.toHaveBeenCalled(); + expect(aborted).toHaveBeenCalledOnce(); + expect(aborted.mock.calls[0][0].message).toBe( + PARTITIONING_FIELD_REMOVAL_ERROR + ); + }); + + test("stops when the stored transfer config document has no name", async () => { + const { ctx, set } = makeContext({ + existing: { + empty: false, + docs: [ + { id: "config-1", data: () => ({ extInstanceId: "users-export" }) }, + ], + }, + }); + + await expect( + handleUpsertTransferConfig(() => ctx) + ).resolves.toBeUndefined(); + + expect(mocks.updateTransferConfig).not.toHaveBeenCalled(); + expect(set).not.toHaveBeenCalled(); + expect(aborted).toHaveBeenCalledOnce(); + expect(aborted.mock.calls[0][0].message).toContain( + "Existing transfer config document config-1 in transferConfigs is missing required 'name' field" + ); + }); + + test("stops when the transfer config has an unsupported structure", async () => { + mocks.updateTransferConfig.mockRejectedValue( + new PermanentConfigurationError( + "Transfer config has invalid structure: missing params.fields. Only scheduled queries are supported." + ) + ); + const { ctx, set } = makeContext({ + existing: { + empty: false, + docs: [ + { + id: "config-1", + data: () => ({ + name: "projects/p/locations/us/transferConfigs/config-1", + }), + }, + ], + }, + }); + + await expect( + handleUpsertTransferConfig(() => ctx) + ).resolves.toBeUndefined(); + + expect(set).not.toHaveBeenCalled(); + expect(aborted).toHaveBeenCalledOnce(); + expect(aborted.mock.calls[0][0].message).toContain( + "Transfer config has invalid structure" + ); + }); +}); + +describe("handleUpsertTransferConfig transient failures", () => { + const aborted = vi.mocked(logs.upsertTransferConfigAborted); + + test("rethrows a BigQuery error so the task is retried", async () => { + const unavailable = Object.assign( + new Error("14 UNAVAILABLE: no healthy upstream"), + { + code: 14, + } + ); + mocks.createTransferConfig.mockRejectedValue(unavailable); + const { ctx } = makeContext({}); + + await expect(handleUpsertTransferConfig(() => ctx)).rejects.toBe( + unavailable + ); + expect(aborted).not.toHaveBeenCalled(); + }); + + test("rethrows a Firestore error so the task is retried", async () => { + const unavailable = new Error("5 DEADLINE_EXCEEDED: Deadline exceeded"); + const { ctx } = makeContext({ existingError: unavailable }); + + await expect(handleUpsertTransferConfig(() => ctx)).rejects.toBe( + unavailable + ); + expect(aborted).not.toHaveBeenCalled(); + }); +}); + +describe("handleUpsertTransferConfig context resolution", () => { + const aborted = vi.mocked(logs.upsertTransferConfigAborted); + + test("aborts when building the context hits a misconfiguration", async () => { + const invalid = new PermanentConfigurationError( + "datasetId must be a non-empty string. Set it in the deployment configuration, then redeploy." + ); + + await expect( + handleUpsertTransferConfig(() => { + throw invalid; + }) + ).resolves.toBeUndefined(); + expect(aborted).toHaveBeenCalledWith(invalid); + }); + + test("rethrows a transient context failure so the task is retried", async () => { + const unavailable = new Error("14 UNAVAILABLE: metadata server"); + + await expect( + handleUpsertTransferConfig(() => { + throw unavailable; + }) + ).rejects.toBe(unavailable); + expect(aborted).not.toHaveBeenCalled(); + }); +});