From ae9a02e7aa1b0e20c37f8327dd0c5bb3f416d23a Mon Sep 17 00:00:00 2001 From: Izaak Gough Date: Wed, 19 Aug 2026 21:16:00 +0100 Subject: [PATCH 1/2] fix(bigquery-firestore-export): notify this instance's topic when linking a config A linked config kept publishing completion notifications to whatever topic it already had, so processMessages never ran and no run output reached Firestore. The link path now sends an update masked to notification_pubsub_topic only, and skips it when the topic already matches. --- kits/bigquery-firestore-export/README.md | 13 ++-- kits/bigquery-firestore-export/src/dts.ts | 44 ++++++++++++- .../bigquery-firestore-export/src/handlers.ts | 8 ++- kits/bigquery-firestore-export/src/lib.ts | 1 + kits/bigquery-firestore-export/src/logs.ts | 14 ++++ .../tests/dts.test.ts | 64 +++++++++++++++++++ .../tests/handlers.test.ts | 18 +++++- 7 files changed, 152 insertions(+), 10 deletions(-) diff --git a/kits/bigquery-firestore-export/README.md b/kits/bigquery-firestore-export/README.md index 884da7b8d..121dcfcf8 100644 --- a/kits/bigquery-firestore-export/README.md +++ b/kits/bigquery-firestore-export/README.md @@ -152,11 +152,14 @@ 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. +Set `TRANSFER_CONFIG_NAME` to link an existing scheduled-query config. The kit +points that config's completion notifications at this instance's topic, which +it needs to read the run output, and changes nothing else: the query, schedule, +destination, and display name stay as they are. 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. ## Firestore layout diff --git a/kits/bigquery-firestore-export/src/dts.ts b/kits/bigquery-firestore-export/src/dts.ts index 5f509b360..1b4cc02c6 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,40 @@ 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. + */ +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"); + } + + 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 +221,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..28df12a5b 100644 --- a/kits/bigquery-firestore-export/src/logs.ts +++ b/kits/bigquery-firestore-export/src/logs.ts @@ -122,6 +122,20 @@ export function transferConfigUpdated(name: string): void { logger.info("Updated BigQuery Data Transfer config", { name }); } +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..df2a2a19e 100644 --- a/kits/bigquery-firestore-export/tests/dts.test.ts +++ b/kits/bigquery-firestore-export/tests/dts.test.ts @@ -20,6 +20,7 @@ import { createTransferConfigRequest, type DataTransferClient, PARTITIONING_FIELD_REMOVAL_ERROR, + updateNotificationTopic, } from "../src/dts"; import { resolveConfig } from "../src/export-config"; @@ -122,3 +123,66 @@ 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"}', + }, + }, + }, + }; + + test("updates only the notification topic when it differs", async () => { + 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 + ); + + 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 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); + }); +}); 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, }); }); }); From 0a49b4d36f5b1b54f079557614d21128d34954e5 Mon Sep 17 00:00:00 2001 From: Izaak Gough Date: Thu, 20 Aug 2026 12:51:00 +0100 Subject: [PATCH 2/2] fix(bigquery-firestore-export): record the topic a linked config gave up Repointing an adopted config's notifications is a takeover: a config notifies one topic, so whatever consumed the previous one goes quiet with no error and nothing recording what to restore. Log the previous topic at warn before overwriting it, and say so in the README along with the shape linking assumes. --- kits/bigquery-firestore-export/README.md | 40 +++++++++++++++---- kits/bigquery-firestore-export/src/dts.ts | 12 ++++++ kits/bigquery-firestore-export/src/logs.ts | 13 ++++++ .../tests/dts.test.ts | 30 +++++++++++++- 4 files changed, 87 insertions(+), 8 deletions(-) diff --git a/kits/bigquery-firestore-export/README.md b/kits/bigquery-firestore-export/README.md index 121dcfcf8..bec7b9ba4 100644 --- a/kits/bigquery-firestore-export/README.md +++ b/kits/bigquery-firestore-export/README.md @@ -152,14 +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. +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 -points that config's completion notifications at this instance's topic, which -it needs to read the run output, and changes nothing else: the query, schedule, -destination, and display name stay as they are. 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. +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 1b4cc02c6..3ccd771d0 100644 --- a/kits/bigquery-firestore-export/src/dts.ts +++ b/kits/bigquery-firestore-export/src/dts.ts @@ -139,6 +139,9 @@ export async function createTransferConfig( * 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, @@ -153,6 +156,15 @@ export async function updateNotificationTopic( 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( diff --git a/kits/bigquery-firestore-export/src/logs.ts b/kits/bigquery-firestore-export/src/logs.ts index 28df12a5b..1f521b2e2 100644 --- a/kits/bigquery-firestore-export/src/logs.ts +++ b/kits/bigquery-firestore-export/src/logs.ts @@ -122,6 +122,19 @@ 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, diff --git a/kits/bigquery-firestore-export/tests/dts.test.ts b/kits/bigquery-firestore-export/tests/dts.test.ts index df2a2a19e..bd48fd5c4 100644 --- a/kits/bigquery-firestore-export/tests/dts.test.ts +++ b/kits/bigquery-firestore-export/tests/dts.test.ts @@ -14,7 +14,8 @@ * 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, @@ -139,7 +140,12 @@ describe("updateNotificationTopic", () => { }, }; + 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" }]); @@ -153,6 +159,11 @@ describe("updateNotificationTopic", () => { 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( @@ -166,6 +177,7 @@ describe("updateNotificationTopic", () => { }); 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, @@ -184,5 +196,21 @@ describe("updateNotificationTopic", () => { 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(); }); });