feat(firestore-bigquery-export): reinstate the Cloud Tasks write buffer

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

kits/firestore-bigquery-export/src/config.ts

@@ -103,7 +103,8 @@ export interface ConfigExpressions {
   datasetId: ConfigExpression<string>;
   tableId: ConfigExpression<string>;
   database: ConfigExpression<string>;
-  maxDispatchesPerSecond: ConfigExpression<number>;
+  /** 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", {

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 extends string | number | boolean | string[]> =
   | 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,
   };

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<void>;
+  /**
+   * 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<void>;
 }
 
 /**
@@ -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<void> {
   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<SerializedDocumentChange>,
+  ctx: HandlerContext
+): Promise<void> {
+  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();

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.<projectId>`), 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,13 +65,20 @@ 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<Role> = [
   "roles/bigquery.dataEditor",
   "roles/datastore.user",
   "roles/bigquery.user",
   // 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<SerializedDocumentChange>(
+  {
+    ...(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<void> {
   try {
     await getHandlerContext().ensureInitialized();

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

kits/firestore-bigquery-export/src/lib.ts

@@ -53,4 +53,5 @@ export {
   type HandlerContext,
   type SerializedDocumentChange,
   handleDocumentWrite,
+  handleSyncBigQueryTask,
 } from "./handlers";

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 }),
   });
 };