diff --git a/kits/bigquery-firestore-export/README.md b/kits/bigquery-firestore-export/README.md index 884da7b8d..bec7b9ba4 100644 --- a/kits/bigquery-firestore-export/README.md +++ b/kits/bigquery-firestore-export/README.md @@ -152,11 +152,40 @@ query on first deploy, then reconciles supported query, table, schedule, dataset, partitioning, and topic changes on later deploys. It retries transient failures up to five times with at least 30 seconds of backoff. -Set `TRANSFER_CONFIG_NAME` to link an existing scheduled-query config without -changing it. Otherwise the kit stores `extInstanceId` on the Firestore config -document and uses that value to find the config on later deploys. BigQuery DTS -does not support clearing a partitioning field once set; create a new transfer -config to remove partitioning. +Without `TRANSFER_CONFIG_NAME` the kit stores `extInstanceId` on the Firestore +config document and uses that value to find its own config on later deploys. +BigQuery DTS does not support clearing a partitioning field once set; create a +new transfer config to remove partitioning. + +### Linking an existing scheduled query + +Set `TRANSFER_CONFIG_NAME` to link an existing scheduled-query config. The kit +needs that config's completion notifications to read its run output, so it +repoints them at this instance's topic. The update is masked to +`notification_pubsub_topic` alone, so the query, schedule, destination, and +display name stay exactly as the config's owner set them. + +Repointing the topic is a takeover, though. A config notifies one topic, so +whatever consumed the previous one stops receiving runs: no error on either +side, and nothing in the config's run history looks different. Removing this +instance does not put the old topic back either. The deploy that changes it logs +the previous value at warn, which is the only record of what to restore. Two +instances linking the same config fail the same way, with the last deploy +winning and the earlier instance going quiet. + +Linking also assumes the config is shaped the way the kit builds its own: + +- Its destination table template is `_{run_time|"%H%M%S"}`. `processMessages` + substitutes that exact placeholder to work out which table a finished run + wrote to. +- It lives in this project. The results query is built from `PROJECT_ID`, not + from the run notification. +- Its destination dataset is in `BIGQUERY_DATASET_LOCATION`. + +A config outside that shape surfaces as a BigQuery table-not-found inside +`processMessages` rather than as anything naming the mismatch, and +`processMessages` retries, so it repeats instead of failing once. That shape is +the only supported one. ## Firestore layout diff --git a/kits/bigquery-firestore-export/src/dts.ts b/kits/bigquery-firestore-export/src/dts.ts index 5f509b360..3ccd771d0 100644 --- a/kits/bigquery-firestore-export/src/dts.ts +++ b/kits/bigquery-firestore-export/src/dts.ts @@ -65,6 +65,12 @@ function stringField(value: string | undefined): { stringValue: string } { return { stringValue: value ?? "" }; } +function notificationTopicName( + config: ResolvedBigqueryFirestoreExportConfig +): string { + return `projects/${config.projectId}/topics/${config.pubSubTopic}`; +} + /** Creates the protobuf-shaped request used for a scheduled query. */ export function createTransferConfigRequest( config: ResolvedBigqueryFirestoreExportConfig, @@ -87,7 +93,7 @@ export function createTransferConfigRequest( }, }, schedule: config.schedule, - notificationPubsubTopic: `projects/${config.projectId}/topics/${config.pubSubTopic}`, + notificationPubsubTopic: notificationTopicName(config), ...(serviceAccountEmail ? { serviceAccountName: serviceAccountEmail } : {}), @@ -129,6 +135,52 @@ export async function createTransferConfig( return created; } +/** + * Points an adopted config at this deployment's notification topic. The mask + * covers only the topic: everything else, including the query, schedule and + * destination, stays as the config's owner set it. + * + * A config notifies exactly one topic, so this claims it. The previous value is + * logged at warn because nothing else records it, and it is the only way back. + */ +export async function updateNotificationTopic( + client: DataTransferClient, + transferConfig: TransferConfig, + config: ResolvedBigqueryFirestoreExportConfig +): Promise { + const expectedTopic = notificationTopicName(config); + if (transferConfig.notificationPubsubTopic === expectedTopic) { + return transferConfig; + } + if (!transferConfig.name) { + throw new Error("BigQuery transfer config is missing its resource name"); + } + + const previousTopic = transferConfig.notificationPubsubTopic; + if (previousTopic) { + logs.notificationTopicTakeover( + transferConfig.name, + previousTopic, + expectedTopic + ); + } + + logs.updateNotificationTopic(transferConfig.name, expectedTopic); + const converted = + bigqueryDataTransfer.protos.google.cloud.bigquery.datatransfer.v1.UpdateTransferConfigRequest.fromObject( + { + transferConfig: { + name: transferConfig.name, + notificationPubsubTopic: expectedTopic, + }, + updateMask: { paths: ["notification_pubsub_topic"] }, + } + ); + const [updated] = await client.updateTransferConfig(converted); + logs.notificationTopicUpdated(transferConfig.name, expectedTopic); + return updated; +} + /** Builds a minimal update mask while retaining unsupported immutable fields. */ export async function constructUpdateTransferConfigRequest( client: DataTransferClient, @@ -181,7 +233,7 @@ export async function constructUpdateTransferConfigRequest( updatedConfig.schedule = config.schedule; } - const expectedTopic = `projects/${config.projectId}/topics/${config.pubSubTopic}`; + const expectedTopic = notificationTopicName(config); if (expectedTopic !== transferConfig.notificationPubsubTopic) { updateMask.push("notification_pubsub_topic"); updatedConfig.notificationPubsubTopic = expectedTopic; diff --git a/kits/bigquery-firestore-export/src/handlers.ts b/kits/bigquery-firestore-export/src/handlers.ts index 13ddddd53..80bc128d3 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, + updateNotificationTopic, updateTransferConfig, } from "./dts"; import type { ResolvedBigqueryFirestoreExportConfig } from "./export-config"; @@ -109,7 +110,12 @@ export async function handleUpsertTransferConfig( `Transfer config not found: ${ctx.config.transferConfigName}` ); } - await storeTransferConfig(ctx, linked); + const notifying = await updateNotificationTopic( + ctx.dataTransfer, + linked, + ctx.config + ); + await storeTransferConfig(ctx, notifying); return; } diff --git a/kits/bigquery-firestore-export/src/lib.ts b/kits/bigquery-firestore-export/src/lib.ts index 234faf45e..a2dd848f4 100644 --- a/kits/bigquery-firestore-export/src/lib.ts +++ b/kits/bigquery-firestore-export/src/lib.ts @@ -28,6 +28,7 @@ export { PARTITIONING_FIELD_REMOVAL_ERROR, PARTITIONING_FIELD_REMOVAL_ERROR_PREFIX, type TransferConfig, + updateNotificationTopic, updateTransferConfig, } from "./dts"; export { diff --git a/kits/bigquery-firestore-export/src/logs.ts b/kits/bigquery-firestore-export/src/logs.ts index 084869113..1f521b2e2 100644 --- a/kits/bigquery-firestore-export/src/logs.ts +++ b/kits/bigquery-firestore-export/src/logs.ts @@ -122,6 +122,33 @@ export function transferConfigUpdated(name: string): void { logger.info("Updated BigQuery Data Transfer config", { name }); } +export function notificationTopicTakeover( + name: string, + previousTopic: string, + topic: string +): void { + logger.warn( + "Taking over the notification topic of a linked transfer config. A config " + + "notifies one topic, so anything consuming the previous topic stops " + + "receiving runs, and removing this instance does not restore it.", + { name, previousTopic, topic } + ); +} + +export function updateNotificationTopic(name: string, topic: string): void { + logger.info("Pointing a linked transfer config at this instance's topic", { + name, + topic, + }); +} + +export function notificationTopicUpdated(name: string, topic: string): void { + logger.info("Updated the notification topic of a linked transfer config", { + name, + topic, + }); +} + export function transferConfigNotFound(name: string): void { logger.error("BigQuery Data Transfer config not found", { name }); } diff --git a/kits/bigquery-firestore-export/tests/dts.test.ts b/kits/bigquery-firestore-export/tests/dts.test.ts index 060ff3ae3..bd48fd5c4 100644 --- a/kits/bigquery-firestore-export/tests/dts.test.ts +++ b/kits/bigquery-firestore-export/tests/dts.test.ts @@ -14,12 +14,14 @@ * limitations under the License. */ -import { describe, expect, test, vi } from "vitest"; +import { afterEach, describe, expect, test, vi } from "vitest"; +import { logger } from "firebase-functions"; import { constructUpdateTransferConfigRequest, createTransferConfigRequest, type DataTransferClient, PARTITIONING_FIELD_REMOVAL_ERROR, + updateNotificationTopic, } from "../src/dts"; import { resolveConfig } from "../src/export-config"; @@ -122,3 +124,93 @@ describe("constructUpdateTransferConfigRequest", () => { ).rejects.toThrow(PARTITIONING_FIELD_REMOVAL_ERROR); }); }); + +describe("updateNotificationTopic", () => { + const linked = { + name: "projects/p/locations/us/transferConfigs/c", + destinationDatasetId: "adopted_dataset", + schedule: "every 6 hours", + params: { + fields: { + query: { stringValue: "SELECT * FROM adopted.rows" }, + destination_table_name_template: { + stringValue: 'adopted_{run_time|"%H%M%S"}', + }, + }, + }, + }; + + afterEach(() => { + vi.restoreAllMocks(); + }); + + test("updates only the notification topic when it differs", async () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined); + const updateTransferConfig = vi + .fn() + .mockResolvedValue([{ ...linked, notificationPubsubTopic: "updated" }]); + const client = { + updateTransferConfig, + } as unknown as DataTransferClient; + + const updated = await updateNotificationTopic( + client, + { ...linked, notificationPubsubTopic: "projects/p/topics/ext-old-topic" }, + config + ); + + expect(warn).toHaveBeenCalledWith(expect.any(String), { + name: linked.name, + previousTopic: "projects/p/topics/ext-old-topic", + topic: "projects/test-project/topics/kit-users-export-processMessages", + }); + const request = updateTransferConfig.mock.calls[0][0]; + expect(request.updateMask.paths).toEqual(["notification_pubsub_topic"]); + expect(request.transferConfig.notificationPubsubTopic).toBe( + "projects/test-project/topics/kit-users-export-processMessages" + ); + expect(request.transferConfig.name).toBe(linked.name); + expect(request.transferConfig.destinationDatasetId).toBeFalsy(); + expect(request.transferConfig.schedule).toBeFalsy(); + expect(request.transferConfig.params).toBeFalsy(); + expect(updated.notificationPubsubTopic).toBe("updated"); + }); + + test("leaves a config already notifying this instance untouched", async () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined); + const updateTransferConfig = vi.fn(); + const client = { + updateTransferConfig, + } as unknown as DataTransferClient; + const alreadyLinked = { + ...linked, + notificationPubsubTopic: + "projects/test-project/topics/kit-users-export-processMessages", + }; + + const returned = await updateNotificationTopic( + client, + alreadyLinked, + config + ); + + expect(updateTransferConfig).not.toHaveBeenCalled(); + expect(returned).toBe(alreadyLinked); + expect(warn).not.toHaveBeenCalled(); + }); + + test("does not warn about a takeover when the config had no topic", async () => { + const warn = vi.spyOn(logger, "warn").mockImplementation(() => undefined); + const updateTransferConfig = vi + .fn() + .mockResolvedValue([{ ...linked, notificationPubsubTopic: "updated" }]); + const client = { + updateTransferConfig, + } as unknown as DataTransferClient; + + await updateNotificationTopic(client, linked, config); + + expect(updateTransferConfig).toHaveBeenCalledTimes(1); + expect(warn).not.toHaveBeenCalled(); + }); +}); diff --git a/kits/bigquery-firestore-export/tests/handlers.test.ts b/kits/bigquery-firestore-export/tests/handlers.test.ts index 6e9cfcc1a..2a865bf19 100644 --- a/kits/bigquery-firestore-export/tests/handlers.test.ts +++ b/kits/bigquery-firestore-export/tests/handlers.test.ts @@ -21,6 +21,7 @@ import { resolveConfig } from "../src/export-config"; const mocks = vi.hoisted(() => ({ createTransferConfig: vi.fn(), getTransferConfig: vi.fn(), + updateNotificationTopic: vi.fn(), updateTransferConfig: vi.fn(), handleTransferRunMessage: vi.fn(), })); @@ -28,6 +29,7 @@ const mocks = vi.hoisted(() => ({ vi.mock("../src/dts", () => ({ createTransferConfig: mocks.createTransferConfig, getTransferConfig: mocks.getTransferConfig, + updateNotificationTopic: mocks.updateNotificationTopic, updateTransferConfig: mocks.updateTransferConfig, })); @@ -146,11 +148,18 @@ describe("handleUpsertTransferConfig", () => { }); }); - test("links an explicitly named transfer config without updating it", async () => { + test("links an explicitly named transfer config and repoints its topic", async () => { const linked = { name: "projects/p/locations/us/transferConfigs/config-2", + notificationPubsubTopic: "projects/p/topics/ext-old-topic", + }; + const notifying = { + ...linked, + notificationPubsubTopic: + "projects/test-project/topics/kit-users-export-processMessages", }; mocks.getTransferConfig.mockResolvedValue(linked); + mocks.updateNotificationTopic.mockResolvedValue(notifying); const { ctx, set } = makeContext({ transferConfigName: linked.name, }); @@ -161,10 +170,15 @@ describe("handleUpsertTransferConfig", () => { ctx.dataTransfer, linked.name ); + expect(mocks.updateNotificationTopic).toHaveBeenCalledWith( + ctx.dataTransfer, + linked, + ctx.config + ); expect(mocks.updateTransferConfig).not.toHaveBeenCalled(); expect(set).toHaveBeenCalledWith({ extInstanceId: "users-export", - ...linked, + ...notifying, }); }); });