From 345bf884cab7fa6ccc7cecb50109bccf1f52e15f Mon Sep 17 00:00:00 2001 From: Jacob Cable Date: Mon, 7 Sep 2026 15:20:33 +0100 Subject: [PATCH] feat(firestore-bigquery-export): reinstate the Cloud Tasks write buffer A failed inline write now enqueues onto the syncBigQuery task queue instead of self-healing once and rethrowing to the trigger retry policy. The queue retries five times with a 60s minimum backoff, throttled by MAX_DISPATCHES_PER_SECOND, and the tracker parks terminal failures in BACKUP_COLLECTION. The trigger drops its retry policy and a failed enqueue is logged, published as onError, and dropped, as the extension did. The success event is published after the insert and swallowed on failure, so a task retry cannot duplicate a row that already landed. --- kits/firestore-bigquery-export/CHANGELOG.md | 1 + kits/firestore-bigquery-export/src/config.ts | 5 +- .../src/export-config.ts | 6 +- .../firestore-bigquery-export/src/handlers.ts | 121 +++++++++++--- kits/firestore-bigquery-export/src/index.ts | 59 ++++++- kits/firestore-bigquery-export/src/init.ts | 2 +- kits/firestore-bigquery-export/src/lib.ts | 1 + kits/firestore-bigquery-export/src/logs.ts | 4 +- .../tests/handlers.test.ts | 149 ++++++++++++++++-- .../tests/index.test.ts | 40 ++++- .../tests/required-apis.test.ts | 55 +++++++ 11 files changed, 393 insertions(+), 50 deletions(-) create mode 100644 kits/firestore-bigquery-export/tests/required-apis.test.ts diff --git a/kits/firestore-bigquery-export/CHANGELOG.md b/kits/firestore-bigquery-export/CHANGELOG.md index d4a4a7407a..a3a12579e5 100644 --- a/kits/firestore-bigquery-export/CHANGELOG.md +++ b/kits/firestore-bigquery-export/CHANGELOG.md @@ -1,3 +1,4 @@ +- feat: reinstate the extension's Cloud Tasks write buffer. A failed inline BigQuery write now enqueues onto a new `syncBigQuery` task queue (5 attempts, 60s minimum backoff, throttled by the restored `MAX_DISPATCHES_PER_SECOND` param, default 100) instead of replaying the Firestore event through Eventarc redelivery for up to 24 hours; `MAX_ENQUEUE_ATTEMPTS` (default 3) is also back. The `onSuccess` event returns with the queue handler. Two behavior changes against earlier release candidates: a row that exhausts the queue is dropped unless `BACKUP_COLLECTION` is set (extension parity - the tracker backs the row up on every terminal insert failure, so configure a backup collection), and deleting or moving the functions can leave the Cloud Tasks queue behind. A failed enqueue is logged at error level, published as an `onError` event, and dropped, as in the extension; the trigger no longer declares `retry: true`, so nothing is redelivered through Eventarc. Export the new `syncBigQuery` function from your codebase entry, and deploy with Firebase CLI 15.28.0+ so the trigger can address its own queue (`FIREBASE_KIT_INSTANCE_ID`); requires firebase-admin 14.2.0+. - fix: restore explicit function placement from `DATABASE_REGION`, now with the Firestore-location-to-Cloud-Run-region mapping. The `DATABASE_REGION` parameter is back and all three functions deploy to the region derived from it: regional locations pass through unchanged, and the multi-region locations map to a Cloud Run region (`nam5`/`nam7` to `us-central1`, `eur3` to `europe-west1`) instead of failing the deploy. With the parameter unset the functions still declare no region and the CLI falls back as before (`us-central1` by default, `FIREBASE_FUNCTIONS_DEFAULT_REGION` to override). Placement requires firebase-tools >= 15.28.0 (older CLIs do not load `.env` at discovery and keep the fallback). If your `.env` already carries `DATABASE_REGION` from an extension migration, upgrading to this version moves the functions to the mapped region on your next deploy, which deletes and recreates them. - fix: stop deploying functions to the `DATABASE_REGION` value. Firestore multi-region locations (`eur3`, `nam5`, `nam7`) are not Cloud Run regions, so any multi-region database made every deploy fail. The `DATABASE_REGION` parameter is removed; the functions now declare no region and deploy to `us-central1` by default (set `FIREBASE_FUNCTIONS_DEFAULT_REGION` when deploying to choose another region), while the Firestore trigger is always pinned to the database's own region. `ExportConfig.location` is removed from the library surface. - Initial release of kit, see README for differences between the legacy extension and this kit diff --git a/kits/firestore-bigquery-export/src/config.ts b/kits/firestore-bigquery-export/src/config.ts index 4715b03464..cd6f958d3f 100644 --- a/kits/firestore-bigquery-export/src/config.ts +++ b/kits/firestore-bigquery-export/src/config.ts @@ -103,7 +103,8 @@ export interface ConfigExpressions { datasetId: ConfigExpression; tableId: ConfigExpression; database: ConfigExpression; - maxDispatchesPerSecond: ConfigExpression; + /** An `IntParam`, not a bare expression: the queue's `rateLimits` guards it with a CEL comparison. */ + maxDispatchesPerSecond: IntParam; } /** @@ -289,7 +290,7 @@ const params = { backupCollection: defineString("BACKUP_COLLECTION", { label: "Backup Collection Name", description: - "This (optional) parameter will allow you to specify a collection for which failed BigQuery updates will be written to.", + "Strongly recommended. The Firestore collection where rows whose BigQuery insert is rejected are written, on the inline attempt and on each queue attempt; without it, those rows are dropped once the queue gives up. A change that cannot be enqueued at all is not backed up. See the README for how to reconcile backed-up rows into BigQuery.", default: "", }), maxDispatchesPerSecond: defineInt("MAX_DISPATCHES_PER_SECOND", { diff --git a/kits/firestore-bigquery-export/src/export-config.ts b/kits/firestore-bigquery-export/src/export-config.ts index e8af202881..a3e8bba0a8 100644 --- a/kits/firestore-bigquery-export/src/export-config.ts +++ b/kits/firestore-bigquery-export/src/export-config.ts @@ -19,6 +19,9 @@ import type { } from "@firebaseextensions/firestore-bigquery-change-tracker"; import type { Expression } from "firebase-functions/params"; +/** Dispatch rate of the `syncBigQuery` queue when `MAX_DISPATCHES_PER_SECOND` is unset. */ +export const DEFAULT_MAX_DISPATCHES_PER_SECOND = 100; + type TrackerLogLevel = "debug" | "info" | "warn" | "error" | "silent"; type ConfigValue = | T @@ -179,7 +182,8 @@ export function resolveExportConfig( kmsKeyName: resolveOptionalConfigValue(config.kmsKeyName), logLevel: (logLevel as TrackerLogLevel) ?? "info", maxDispatchesPerSecond: - resolveOptionalConfigValue(config.maxDispatchesPerSecond) ?? 100, + resolveOptionalConfigValue(config.maxDispatchesPerSecond) ?? + DEFAULT_MAX_DISPATCHES_PER_SECOND, maxEnqueueAttempts: resolveOptionalConfigValue(config.maxEnqueueAttempts) ?? 3, }; diff --git a/kits/firestore-bigquery-export/src/handlers.ts b/kits/firestore-bigquery-export/src/handlers.ts index daf7cd254e..7bdfb4e07b 100644 --- a/kits/firestore-bigquery-export/src/handlers.ts +++ b/kits/firestore-bigquery-export/src/handlers.ts @@ -24,12 +24,17 @@ import type { DocumentSnapshot, FirestoreEvent, } from "firebase-functions/firestore"; +import type { Request } from "firebase-functions/tasks"; import * as events from "./events"; import type { ResolvedExportConfig } from "./export-config"; import * as logs from "./logs"; import { getChangeType, getDocumentId } from "./util"; -/** Serialized Firestore change ready to write to BigQuery. */ +/** + * Serialized Firestore change ready to write to BigQuery. Also the + * `syncBigQuery` task payload: it is built from already-serialized data, so it + * survives the JSON round trip through Cloud Tasks unchanged. + */ export interface SerializedDocumentChange { timestamp: string; eventId: string; @@ -55,11 +60,15 @@ export interface HandlerContext { tracker: FirestoreBigQueryEventHistoryTracker; config: ResolvedExportConfig; /** - * Provisions the BigQuery dataset/table/views once per instance. Only called - * after an inline write failure as a self-heal; the hot path relies on - * out-of-band provisioning (`initBigQuerySync` / `setupBigQuerySync`). + * Provisions the BigQuery resources. Used by the lifecycle tasks only; the + * write paths never call it. */ ensureInitialized: () => Promise; + /** + * Enqueues a failed change onto the `syncBigQuery` task queue. Rejects with + * the enqueue error once its own retry budget is exhausted. + */ + enqueue: (change: SerializedDocumentChange) => Promise; } /** @@ -87,38 +96,38 @@ async function recordEventToBigQuery( } /** - * Gives a failed inline write one self-heal attempt before surfacing it to the - * Firestore trigger retry policy. + * Buffers a failed inline write through the `syncBigQuery` task queue. A + * terminal enqueue failure is logged and published as an `onError` event, then + * dropped, exactly as the extension did: with no retry policy on the trigger a + * rethrow would only fail the execution once and drop it anyway. * - * @param change - The serialized change to write. + * @param change - The serialized change to enqueue. * @param ctx - The handler context. */ -async function retryAfterSelfHeal( +async function enqueueForSync( change: SerializedDocumentChange, ctx: HandlerContext ): Promise { try { - await ctx.ensureInitialized(); - await recordEventToBigQuery(change, ctx.tracker); - } catch (retryErr) { - await events.recordErrorEvent(retryErr as Error); - + await ctx.enqueue(change); + } catch (enqueueErr) { + // Log before publishing: the log line is the only trace of the dropped + // row, and the event publish can itself reject. logs.logFailedEventAction( - "Failed to write event to BigQuery from onWrite handler after self-heal", + "Failed to enqueue event to Cloud Tasks from onWrite handler", change.fullResourceName, change.eventId, change.changeType, - retryErr as Error + enqueueErr as Error ); - - throw retryErr; + await events.recordErrorEvent(enqueueErr as Error); } } /** * Handles a Firestore document write: serializes the change and writes it to - * BigQuery. Failed writes are surfaced to the trigger retry policy after one - * self-heal attempt. + * BigQuery. A failed inline write is buffered through the `syncBigQuery` task + * queue; a failed enqueue is logged and dropped. * * @param event - The Firestore document-write event. * @param ctx - The handler context. @@ -130,12 +139,10 @@ export async function handleDocumentWrite( const { data, ...context } = event; if (!data) return; - logs.start(); - - // No provisioning on the hot path: BigQuery resources are provisioned - // out-of-band (afterFirstDeploy / afterRedeploy tasks). If they are missing, - // the inline write fails, self-heals once, then falls back to the trigger - // retry policy. + logs.start(); // No provisioning on the hot path, and none on the queue path either: only + // the lifecycle tasks create BigQuery resources. A missing dataset or table + // fails the inline write and every queue attempt the same way until a + // redeploy runs the lifecycle task; BACKUP_COLLECTION is the only net. const { config, tracker } = ctx; const changeType = getChangeType(data); const documentId = getDocumentId(data); @@ -204,7 +211,69 @@ export async function handleDocumentWrite( await recordEventToBigQuery(change, tracker); } catch (err) { logs.failedToWriteToBigQueryImmediately(err as Error); - await retryAfterSelfHeal(change, ctx); + await enqueueForSync(change, ctx); + } + + logs.complete(); +} + +/** + * Handles a `syncBigQuery` task: re-attempts a buffered write. No provisioning + * runs here, as in the extension: that stays in the lifecycle tasks, so a + * recovery burst does not fan `initialize()` out across every cold instance. + * A failed write rethrows so Cloud Tasks retries on the queue's schedule; the + * tracker parks the row in the backup collection before each terminal + * rethrow. + * + * @param req - The dispatched task request carrying the serialized change. + * @param ctx - The handler context. + */ +export async function handleSyncBigQueryTask( + req: Request, + ctx: HandlerContext +): Promise { + const change = req.data; + + logs.logEventAction( + "Firestore event received by onDispatch trigger", + change.fullResourceName, + change.eventId, + change.changeType + ); + + try { + await recordEventToBigQuery(change, ctx.tracker); + } catch (err) { + logs.logFailedEventAction( + "Failed to write event to BigQuery from onDispatch handler", + change.fullResourceName, + change.eventId, + change.changeType, + err as Error, + req.retryCount + ); + + throw err; + } + + try { + await events.recordSuccessEvent({ + subject: change.documentId, + data: { + timestamp: change.timestamp, + operation: change.changeType, + documentName: change.fullResourceName, + documentId: change.documentId, + pathParams: change.params, + eventId: change.eventId, + data: change.data, + oldData: change.oldData, + }, + }); + } catch (err) { + // The row is already in BigQuery. Rethrowing would have Cloud Tasks retry + // the insert past the dedupe window and duplicate it. + logs.error(false, "Failed to record success event", err as Error); } logs.complete(); diff --git a/kits/firestore-bigquery-export/src/index.ts b/kits/firestore-bigquery-export/src/index.ts index 268a43b2ba..d74383ae35 100644 --- a/kits/firestore-bigquery-export/src/index.ts +++ b/kits/firestore-bigquery-export/src/index.ts @@ -17,8 +17,8 @@ /** * Main entry point. Exports the wired functions with deploy-time param * expressions, then resolves concrete config lazily at runtime. Re-export - * `fsexportbigquery` and `initBigQuerySync` from your own functions codebase - * entry; configuration comes from a `.env` (or + * `fsexportbigquery`, `syncBigQuery`, and `initBigQuerySync` from your own + * functions codebase entry; configuration comes from a `.env` (or * `.env.`), which the Firebase CLI loads at deploy. * * Because this module initializes runtime dependencies lazily, deploy discovery @@ -40,11 +40,21 @@ import { } from "firebase-functions/v2/lifecycle"; import { CONFIG_EXPRESSIONS, configFromEnv } from "./config"; import * as events from "./events"; -import { resolveExportConfig, toTrackerConfig } from "./export-config"; -import { type HandlerContext, handleDocumentWrite } from "./handlers"; +import { + DEFAULT_MAX_DISPATCHES_PER_SECOND, + resolveExportConfig, + toTrackerConfig, +} from "./export-config"; +import { + type HandlerContext, + type SerializedDocumentChange, + handleDocumentWrite, + handleSyncBigQueryTask, +} from "./handlers"; import { createEnsureInitialized } from "./init"; import * as logs from "./logs"; import { firestoreLocationToFunctionRegion } from "./region"; +import { enqueueSyncTask } from "./tasks"; // Re-export the side-effect-free library surface (handlers and config types). export * from "./lib"; @@ -55,6 +65,11 @@ const LIFECYCLE_RETRY_CONFIG = { maxAttempts: 15, minBackoffSeconds: 60, } as const; +const SYNC_RETRY_CONFIG = { + maxAttempts: 5, + minBackoffSeconds: 60, +} as const; +const SYNC_MAX_CONCURRENT_DISPATCHES = 500; const REQUIRED_ROLES: ReadonlyArray = [ "roles/bigquery.dataEditor", "roles/datastore.user", @@ -62,6 +77,8 @@ const REQUIRED_ROLES: ReadonlyArray = [ // Gen2 Firestore triggers need Eventarc receive and run.invoker on the function SA. "roles/eventarc.eventReceiver", "roles/run.invoker", + // The trigger enqueues failed writes onto its own syncBigQuery task queue. + "roles/cloudtasks.enqueuer", ]; const REQUIRED_APIS = [ { @@ -116,6 +133,8 @@ function getHandlerContext(): HandlerContext { tracker, config, ensureInitialized, + enqueue: (change: SerializedDocumentChange) => + enqueueSyncTask(change, config.maxEnqueueAttempts), }; return ctx; @@ -135,19 +154,45 @@ const functionRegion = firestoreLocationToFunctionRegion( /** * Firestore trigger: streams document writes on the watched collection into the - * BigQuery changelog table. Failed executions are retried by the Firebase - * Functions runtime. + * BigQuery changelog table. A failed inline write buffers through the + * `syncBigQuery` queue and the execution still succeeds. No runtime retry + * policy, as in the extension: a failure before the write is attempted fails + * the execution once, and a failed enqueue is logged and dropped. */ export const fsexportbigquery = onDocumentWritten( { ...(functionRegion ? { region: functionRegion } : {}), document: expr`${CONFIG_EXPRESSIONS.collectionPath}/{documentId}`, database: CONFIG_EXPRESSIONS.database, - retry: true, }, (event) => handleDocumentWrite(event, getHandlerContext()) ); +/** + * Write-buffer task queue: re-attempts writes that failed inline, on Cloud + * Tasks' schedule (5 attempts, 60s minimum backoff, dispatch-throttled by + * `MAX_DISPATCHES_PER_SECOND`). After the last attempt the task is dropped; + * by then the tracker has written the row to `BACKUP_COLLECTION` on every + * terminal insert failure, when that collection is configured. + */ +export const syncBigQuery = onTaskDispatched( + { + ...(functionRegion ? { region: functionRegion } : {}), + retryConfig: SYNC_RETRY_CONFIG, + rateLimits: { + maxConcurrentDispatches: SYNC_MAX_CONCURRENT_DISPATCHES, // A blank .env value reaches this deploy-time expression as 0, which + // Cloud Tasks would not accept; runtime falls back to the same default. + maxDispatchesPerSecond: CONFIG_EXPRESSIONS.maxDispatchesPerSecond + .lessThan(1) + .thenElse( + DEFAULT_MAX_DISPATCHES_PER_SECOND, + CONFIG_EXPRESSIONS.maxDispatchesPerSecond + ), + }, + }, + (req) => handleSyncBigQueryTask(req, getHandlerContext()) +); + async function handleBigQuerySyncInitialization(): Promise { try { await getHandlerContext().ensureInitialized(); diff --git a/kits/firestore-bigquery-export/src/init.ts b/kits/firestore-bigquery-export/src/init.ts index 530b2b3543..06c57f1b66 100644 --- a/kits/firestore-bigquery-export/src/init.ts +++ b/kits/firestore-bigquery-export/src/init.ts @@ -18,7 +18,7 @@ import type { FirestoreBigQueryEventHistoryTracker } from "@firebaseextensions/f /** * Builds the provisioning guard used by the `initBigQuerySync` endpoint and the - * retry-path self-heal. The hot write path never calls it. + * lifecycle tasks. The write paths never call it. * * The returned function runs `tracker.initialize()` at most once per instance: * concurrent invocations on a cold instance share a single in-flight promise. A diff --git a/kits/firestore-bigquery-export/src/lib.ts b/kits/firestore-bigquery-export/src/lib.ts index 3c83dd304c..8e5939f514 100644 --- a/kits/firestore-bigquery-export/src/lib.ts +++ b/kits/firestore-bigquery-export/src/lib.ts @@ -53,4 +53,5 @@ export { type HandlerContext, type SerializedDocumentChange, handleDocumentWrite, + handleSyncBigQueryTask, } from "./handlers"; diff --git a/kits/firestore-bigquery-export/src/logs.ts b/kits/firestore-bigquery-export/src/logs.ts index 12ee2c4885..bb17b72c14 100644 --- a/kits/firestore-bigquery-export/src/logs.ts +++ b/kits/firestore-bigquery-export/src/logs.ts @@ -208,7 +208,8 @@ export const logFailedEventAction = ( document_name: string, event_id: string, operation: ChangeType, - error: Error + error: Error, + retry_count?: number ) => { const changeTypeMap = { 0: "CREATE", @@ -222,6 +223,7 @@ export const logFailedEventAction = ( event_id, operation: changeTypeMap[operation], error, + ...(retry_count === undefined ? {} : { retry_count }), }); }; diff --git a/kits/firestore-bigquery-export/tests/handlers.test.ts b/kits/firestore-bigquery-export/tests/handlers.test.ts index b37f675090..8e88f0fcc9 100644 --- a/kits/firestore-bigquery-export/tests/handlers.test.ts +++ b/kits/firestore-bigquery-export/tests/handlers.test.ts @@ -15,12 +15,15 @@ */ import { ChangeType } from "@firebaseextensions/firestore-bigquery-change-tracker"; +import type { Request } from "firebase-functions/tasks"; import { beforeEach, describe, expect, test, vi } from "vitest"; import { resolveExportConfig } from "../src/export-config"; import { type DocumentWriteEvent, type HandlerContext, + type SerializedDocumentChange, handleDocumentWrite, + handleSyncBigQueryTask, } from "../src/handlers"; vi.mock("../src/events"); @@ -70,6 +73,33 @@ function makeCtx( tracker: tracker as unknown as HandlerContext["tracker"], config: config as HandlerContext["config"], ensureInitialized: vi.fn().mockResolvedValue(undefined), + enqueue: vi.fn().mockResolvedValue(undefined), + }; +} + +/** Fake dispatched task request carrying a serialized change. */ +function taskRequest( + change: SerializedDocumentChange, + retryCount = 0 +): Request { + return { data: change, retryCount } as Request; +} + +/** A serialized change as it would arrive in a task payload. */ +function serializedChange( + overrides: Partial = {} +): SerializedDocumentChange { + return { + timestamp: "2026-01-01T00:00:00Z", + eventId: "evt-1", + fullResourceName: + "projects/test-project/databases/(default)/documents/users/doc1", + changeType: ChangeType.CREATE, + documentId: "doc1", + params: null, + data: { a: 1 }, + oldData: undefined, + ...overrides, }; } @@ -170,7 +200,7 @@ describe("handleDocumentWrite", () => { expect(recordedWithout[0].pathParams).toBeNull(); }); - test("self-heals and retries the write when the inline write fails", async () => { + test("a failed inline write buffers through the queue and the execution succeeds", async () => { const ctx = makeCtx(); (ctx.tracker.record as ReturnType).mockRejectedValueOnce( new Error("bq down") @@ -181,24 +211,68 @@ describe("handleDocumentWrite", () => { ctx ); - expect(ctx.ensureInitialized).toHaveBeenCalledTimes(1); - expect(ctx.tracker.record).toHaveBeenCalledTimes(2); + expect(ctx.tracker.record).toHaveBeenCalledTimes(1); + expect(ctx.enqueue).toHaveBeenCalledTimes(1); + expect(ctx.ensureInitialized).not.toHaveBeenCalled(); }); - test("rethrows when self-heal retry fails so runtime retry can replay", async () => { + test("a successful inline write enqueues nothing", async () => { const ctx = makeCtx(); - (ctx.tracker.record as ReturnType) - .mockRejectedValueOnce(new Error("bq down")) - .mockRejectedValueOnce(new Error("still down")); + + await handleDocumentWrite( + writeEvent(snap(false, "doc1"), snap(true, "doc1", { a: 1 })), + ctx + ); + + expect(ctx.enqueue).not.toHaveBeenCalled(); + }); + + test("the enqueued change equals what the inline path tried to write", async () => { + const ctx = makeCtx(); + (ctx.tracker.record as ReturnType).mockRejectedValueOnce( + new Error("bq down") + ); + + await handleDocumentWrite( + writeEvent(snap(true, "doc1", { a: 1 }), snap(true, "doc1", { a: 2 })), + ctx + ); + + const [[recorded]] = (ctx.tracker.record as ReturnType).mock + .calls; + const [[enqueued]] = (ctx.enqueue as ReturnType).mock.calls; + expect(enqueued).toMatchObject({ + timestamp: recorded[0].timestamp, + eventId: recorded[0].eventId, + fullResourceName: recorded[0].documentName, + changeType: recorded[0].operation, + documentId: recorded[0].documentId, + params: recorded[0].pathParams, + data: recorded[0].data, + oldData: recorded[0].oldData, + }); + // The payload must survive the JSON round trip through Cloud Tasks. + expect(JSON.parse(JSON.stringify(enqueued))).toEqual(enqueued); + }); + + test("a failed enqueue is recorded and logged, then the execution succeeds", async () => { + // Extension parity: no retry policy on the trigger, so a rethrow would + // only fail the execution once and drop the event anyway. + const ctx = makeCtx(); + (ctx.tracker.record as ReturnType).mockRejectedValueOnce( + new Error("bq down") + ); + (ctx.enqueue as ReturnType).mockRejectedValueOnce( + new Error("tasks down") + ); await expect( handleDocumentWrite( writeEvent(snap(false, "doc1"), snap(true, "doc1", { a: 1 })), ctx ) - ).rejects.toThrow("still down"); - expect(ctx.ensureInitialized).toHaveBeenCalledTimes(1); - expect(events.recordErrorEvent).toHaveBeenCalled(); + ).resolves.toBeUndefined(); + expect(events.recordErrorEvent).toHaveBeenCalledTimes(1); }); test("rethrows when serialization fails", async () => { @@ -218,3 +292,58 @@ describe("handleDocumentWrite", () => { expect(ctx.tracker.record).not.toHaveBeenCalled(); }); }); + +describe("handleSyncBigQueryTask", () => { + beforeEach(() => vi.clearAllMocks()); + + test("records the buffered change and emits a success event", async () => { + const ctx = makeCtx(); + const change = serializedChange(); + + await handleSyncBigQueryTask(taskRequest(change), ctx); + + // Extension parity: no provisioning on the write path. + expect(ctx.ensureInitialized).not.toHaveBeenCalled(); + const [[recorded]] = (ctx.tracker.record as ReturnType).mock + .calls; + expect(recorded[0]).toMatchObject({ + timestamp: change.timestamp, + operation: change.changeType, + documentName: change.fullResourceName, + documentId: change.documentId, + eventId: change.eventId, + data: change.data, + }); + expect(events.recordSuccessEvent).toHaveBeenCalledTimes(1); + expect(ctx.enqueue).not.toHaveBeenCalled(); + }); + + test("rethrows a failed write so Cloud Tasks retries", async () => { + const ctx = makeCtx(); + (ctx.tracker.record as ReturnType).mockRejectedValueOnce( + new Error("still down") + ); + + await expect( + handleSyncBigQueryTask(taskRequest(serializedChange(), 2), ctx) + ).rejects.toThrow("still down"); + expect(events.recordSuccessEvent).not.toHaveBeenCalled(); + // Re-enqueueing from the task would seed a trigger-queue loop; retries + // belong to Cloud Tasks alone. + expect(ctx.enqueue).not.toHaveBeenCalled(); + }); + + test("does not rethrow when the success event fails after the row lands", async () => { + // The row is already in BigQuery; a Cloud Tasks retry would land past the + // insertId dedupe window and duplicate it. + const ctx = makeCtx(); + ( + events.recordSuccessEvent as ReturnType + ).mockRejectedValueOnce(new Error("channel down")); + + await expect( + handleSyncBigQueryTask(taskRequest(serializedChange()), ctx) + ).resolves.toBeUndefined(); + expect(ctx.tracker.record).toHaveBeenCalledTimes(1); + }); +}); diff --git a/kits/firestore-bigquery-export/tests/index.test.ts b/kits/firestore-bigquery-export/tests/index.test.ts index f5748c4280..4a96a0576f 100644 --- a/kits/firestore-bigquery-export/tests/index.test.ts +++ b/kits/firestore-bigquery-export/tests/index.test.ts @@ -35,6 +35,7 @@ type FunctionOptions = Record; interface ExportedOptions { trigger: FunctionOptions; + /** Options of syncBigQuery, initBigQuerySync, setupBigQuerySync, in order. */ tasks: FunctionOptions[]; } @@ -66,10 +67,10 @@ async function loadExportedOptions( const taskCalls = vi.mocked(onTaskDispatched).mock.calls; const trigger = triggerCalls[triggerCalls.length - 1][0] as FunctionOptions; const tasks = taskCalls - .slice(-2) + .slice(-3) .map((call) => call[0] as unknown as FunctionOptions); - expect(tasks).toHaveLength(2); + expect(tasks).toHaveLength(3); return { trigger, tasks }; } @@ -123,4 +124,39 @@ describe("exported function options", () => { const document = trigger.document as { toCEL(): string }; expect(document.toCEL()).toContain("params.COLLECTION_PATH"); }); + + test("the trigger declares no retry policy, matching the extension", async () => { + const { trigger } = await loadExportedOptions(); + expect(trigger.retry).toBeUndefined(); + }); + + test("syncBigQuery pins the extension's queue shape", async () => { + const { tasks } = await loadExportedOptions(); + const [syncTask] = tasks; + + expect(syncTask.retryConfig).toEqual({ + maxAttempts: 5, + minBackoffSeconds: 60, + }); + + const rateLimits = syncTask.rateLimits as Record; + expect(rateLimits.maxConcurrentDispatches).toBe(500); // Concurrency comes from the gen2 defaults (80 per instance), so no + // instance cap is declared. + expect(syncTask.maxInstances).toBeUndefined(); + // A blank .env value is 0 at deploy; the ternary restores the default. + expect(String(rateLimits.maxDispatchesPerSecond)).toBe( + "params.MAX_DISPATCHES_PER_SECOND < 1 ? 100 : params.MAX_DISPATCHES_PER_SECOND" + ); + }); + + test("the lifecycle tasks keep their own retry config", async () => { + const { tasks } = await loadExportedOptions(); + for (const opts of tasks.slice(1)) { + expect(opts.retryConfig).toEqual({ + maxAttempts: 15, + minBackoffSeconds: 60, + }); + expect(opts).not.toHaveProperty("rateLimits"); + } + }); }); diff --git a/kits/firestore-bigquery-export/tests/required-apis.test.ts b/kits/firestore-bigquery-export/tests/required-apis.test.ts new file mode 100644 index 0000000000..73fc9bbfe2 --- /dev/null +++ b/kits/firestore-bigquery-export/tests/required-apis.test.ts @@ -0,0 +1,55 @@ +/** + * 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. + */ + +import { beforeAll, expect, test, vi } from "vitest"; + +const { requiresAPI } = vi.hoisted(() => ({ requiresAPI: vi.fn() })); + +vi.mock("firebase-functions/firestore", () => ({ + onDocumentWritten: vi.fn(() => ({})), +})); +vi.mock("firebase-functions/tasks", () => ({ + onTaskDispatched: vi.fn(() => ({})), +})); +vi.mock("firebase-functions/v2", () => ({ + requiresAPI, + requiresRole: vi.fn(), +})); +vi.mock("firebase-functions/v2/lifecycle", () => ({ + afterFirstDeploy: vi.fn(), + afterRedeploy: vi.fn(), +})); + +beforeAll(async () => { + await import("../src/index"); +}); + +test.each([ + [ + "firestore.googleapis.com", + "Receives document change events from Cloud Firestore.", + ], + [ + "bigquery.googleapis.com", + "Mirrors data from your Cloud Firestore collection in BigQuery.", + ], +])("declares %s", (api, reason) => { + expect(requiresAPI).toHaveBeenCalledWith(api, reason); +}); + +test("declares exactly the two APIs", () => { + expect(requiresAPI).toHaveBeenCalledTimes(2); +});