At some point, every developer building AI features hits the same wall. A spike in traffic causes cascading Gemini API timeouts. A processing failure mid-flight leaves data in an unknown state because there's no retry mechanism. Event logs accumulate with nowhere structured to send them. These are the moments when the ceiling of synchronous architectures becomes impossible to ignore.
Working with Antigravity has made implementing async pipelines significantly smoother. I used to delay the Cloud Run + Pub/Sub setup because it felt like a lot of infrastructure plumbing for uncertain payoff. Now I describe the intent — "I need an event-driven AI processing pipeline with reliable retries and BigQuery logging" — and Antigravity proposes a coherent architecture covering Pub/Sub topic definitions, Cloud Run service code, and BigQuery schema design all at once. The scaffolding that used to take a day lands in minutes.
This guide walks through the pipeline configuration I've refined through real projects, complete with working code. Not just "something that runs in development," but a production-ready design with dead-letter queues, idempotency, monitoring, and the pitfalls clearly labeled.
Why Synchronous AI Processing Breaks Down in Production
Before getting into implementation, it's worth being specific about why synchronous designs fail at scale. Understanding the failure modes makes the architectural decisions that follow much easier to justify.
The latency problem. Calls to large language models can take anywhere from a few hundred milliseconds to over ten seconds depending on the model, input length, and server load. Placing those calls inside a user-facing HTTP handler degrades the user experience directly. Worse, it creates timeout cascades: if your app server has a 30-second request timeout and the AI call takes 25 seconds, you're living dangerously close to the edge at all times. Any upstream load balancer with a tighter timeout will start dropping requests before your code even knows there's a problem.
The reliability problem. In a synchronous flow, when the AI API returns an error, that request simply fails. The data the user was trying to process isn't processed. If you want retries, you have to implement them yourself — which means managing state about "what was being processed and where did we leave off?" That state management is where most retry implementations go wrong and produce duplicate data.
The scalability problem. Without a work queue acting as a buffer, every increase in traffic directly increases load on your AI processing layer. There's no damper. A 10x traffic spike means 10x concurrent AI API calls, which typically means you hit rate limits, and the whole system backs up simultaneously.
The event-driven pattern with Pub/Sub at the center solves all three by introducing an explicit queue between the ingestion of work and the execution of work.
User Request
↓
Ingestion API (Cloud Run)
- validates input
- assigns jobId
- publishes to Pub/Sub
- returns 202 immediately
↓
Pub/Sub Topic: ai-processing-jobs
- durable message storage
- configurable retry with backoff
- routes failed messages to DLQ
↓
Worker Service (Cloud Run)
- receives push from Pub/Sub
- calls Gemini API
- stores result in BigQuery
- returns 200 ACK or 500 NACK
↓
BigQuery: ai_pipeline_results
- structured result storage
- latency, token counts, errors
- queryable for dashboards
↓
Dead-Letter Queue (Pub/Sub DLQ)
- receives persistently failing messages
- enables manual inspection and re-injection
The ingestion API returns to the client before any AI processing happens. The worker processes at its own pace. Pub/Sub retries automatically on failure. BigQuery accumulates every outcome — successes, failures, partial results — for analysis.
Architecture Components in Detail
Let me walk through each component's responsibility, because the boundaries matter when things go wrong.
Ingestion Service
The ingestion service has exactly one job: accept a request, validate it, assign a job ID, and publish to Pub/Sub. It should know nothing about how AI processing works. This separation means you can change the processing logic, swap the AI model, or add new job types without touching the ingestion layer.
The job ID generation is more important than it looks. I use a hash-based ID that incorporates the user ID, job type, and a snippet of the input. This means identical requests from the same user within a short window get the same job ID — which the worker uses to detect and skip duplicates. It's not a perfect deduplication system (you'd need Redis for that), but it catches the most common case of a user double-submitting a form.
Worker Service
The worker is where the actual AI work happens. It's called via Pub/Sub push subscription — Pub/Sub sends an HTTP POST to the worker's Cloud Run endpoint whenever a message is available. The worker calls Gemini, stores the result in BigQuery, and returns an HTTP status code that tells Pub/Sub what to do next.
This is the most important thing to internalize about this architecture: the HTTP status code the worker returns is not just a response to the caller — it's an instruction to Pub/Sub. Return 200 and the message is acknowledged and deleted. Return anything in the 4xx or 5xx range and Pub/Sub considers the delivery failed and will redeliver after a backoff period.
BigQuery as Result Store
Using BigQuery as the result store rather than a transactional database like Cloud Spanner or Cloud SQL is a deliberate choice with tradeoffs. BigQuery is append-optimized and terrible for point lookups on single rows, but excellent for aggregate analytics over large datasets. If you need to serve AI results back to end users in real-time (e.g., "show me my result for job ID xyz"), BigQuery alone isn't sufficient — you'd add Firestore as a serving layer that the worker also writes to. But for analytics — success rates, latency percentiles, token costs by user — BigQuery is far cheaper and more powerful than trying to run the same queries against Cloud Spanner.
For this pipeline, I use a pattern where the worker writes to both Firestore (for serving the result back to the user) and BigQuery (for analytics). The Firestore write is synchronous and blocks ACK. The BigQuery write is fire-and-forget — if it fails, we log the error but still ACK the message.
Building the Ingestion Service
Here's the complete ingestion service implementation in TypeScript:
// ingestion-service/src/index.ts
import { PubSub } from "@google-cloud/pubsub";
import express from "express";
import { z } from "zod";
import crypto from "crypto";
const app = express();
app.use(express.json());
const pubsub = new PubSub({ projectId: process.env.GCP_PROJECT_ID });
const TOPIC_NAME = process.env.PUBSUB_TOPIC_NAME ?? "ai-processing-jobs";
const ProcessRequestSchema = z.object({
userId: z.string().min(1),
inputText: z.string().min(1).max(10000),
jobType: z.enum(["summarize", "classify", "extract"]),
callbackUrl: z.string().url().optional(),
});
app.post("/api/process", async (req, res) => {
const parseResult = ProcessRequestSchema.safeParse(req.body);
if (!parseResult.success) {
return res.status(400).json({ error: parseResult.error.issues });
}
const { userId, inputText, jobType, callbackUrl } = parseResult.data;
// Deterministic job ID: same request from same user → same ID within time window
const idempotencyKey = crypto
.createHash("sha256")
.update(`${userId}-${jobType}-${inputText.slice(0, 100)}-${Math.floor(Date.now() / 60000)}`)
.digest("hex")
.slice(0, 32);
const jobId = `job-${idempotencyKey}`;
const message = {
jobId,
userId,
inputText,
jobType,
callbackUrl,
createdAt: new Date().toISOString(),
};
try {
const topic = pubsub.topic(TOPIC_NAME);
const messageId = await topic.publishMessage({
data: Buffer.from(JSON.stringify(message)),
attributes: {
jobType,
userId,
idempotencyKey,
schemaVersion: "1",
},
});
console.log(`[ingestion] Published jobId=${jobId} messageId=${messageId}`);
// 202 Accepted: the request is queued but not yet processed
return res.status(202).json({
jobId,
messageId,
status: "accepted",
estimatedProcessingSeconds: jobType === "summarize" ? 5 : 2,
});
} catch (error) {
console.error("[ingestion] Pub/Sub publish failed:", error);
// 503 is more accurate than 500 here — the upstream (Pub/Sub) is unavailable
return res.status(503).json({
error: "Failed to queue job. Retry after a short delay.",
retryAfterSeconds: 5,
});
}
});
// Status endpoint: check if a job has been processed (reads from Firestore)
app.get("/api/jobs/:jobId", async (req, res) => {
// Implementation depends on your Firestore setup
// This is a placeholder showing the expected response shape
return res.status(200).json({
jobId: req.params.jobId,
status: "pending", // "pending" | "processing" | "completed" | "failed"
result: null,
});
});
app.get("/health", (_req, res) => res.json({ status: "ok", timestamp: new Date().toISOString() }));
const PORT = parseInt(process.env.PORT ?? "8080");
app.listen(PORT, () => console.log(`[ingestion] Listening on :${PORT}`));The Math.floor(Date.now() / 60000) in the idempotency key means the same request submitted twice within the same minute will get the same job ID — useful for catching double-clicks or retried form submissions at the client side.
Building the Worker Service
The worker is the most complex component. Let me walk through it section by section:
// worker-service/src/index.ts
import { VertexAI } from "@google-cloud/vertexai";
import { BigQuery } from "@google-cloud/bigquery";
import { Firestore } from "@google-cloud/firestore";
import express from "express";
const app = express();
app.use(express.json());
const vertexAI = new VertexAI({
project: process.env.GCP_PROJECT_ID!,
location: process.env.GCP_REGION ?? "us-central1",
});
const bigquery = new BigQuery({ projectId: process.env.GCP_PROJECT_ID });
const firestore = new Firestore({ projectId: process.env.GCP_PROJECT_ID });
const DATASET_ID = process.env.BQ_DATASET_ID ?? "ai_pipeline_results";
const TABLE_ID = process.env.BQ_TABLE_ID ?? "processing_jobs";
// Simple in-memory idempotency cache
// In production with multiple Cloud Run instances, replace with Redis SET NX EX
const processedJobs = new Map<string, number>(); // jobId → timestamp
function isProcessed(jobId: string): boolean {
const ts = processedJobs.get(jobId);
if (!ts) return false;
// Expire entries after 2 hours
if (Date.now() - ts > 7200000) {
processedJobs.delete(jobId);
return false;
}
return true;
}
function markProcessed(jobId: string): void {
processedJobs.set(jobId, Date.now());
// Prevent unbounded memory growth
if (processedJobs.size > 5000) {
// Delete oldest entries
const cutoff = Date.now() - 7200000;
for (const [id, ts] of processedJobs.entries()) {
if (ts < cutoff) processedJobs.delete(id);
}
}
}
app.post("/pubsub/push", async (req, res) => {
const startTime = Date.now();
// Validate the Pub/Sub envelope
const pubsubMessage = req.body?.message;
if (!pubsubMessage?.data) {
console.warn("[worker] Received message with no data, ACK-ing to discard");
return res.status(200).json({ status: "discarded_no_data" });
}
// Decode and parse the job payload
let job: {
jobId: string;
userId: string;
inputText: string;
jobType: "summarize" | "classify" | "extract";
callbackUrl?: string;
createdAt: string;
};
try {
const decoded = Buffer.from(pubsubMessage.data, "base64").toString("utf-8");
job = JSON.parse(decoded);
if (!job.jobId || !job.userId || !job.inputText || !job.jobType) {
throw new Error("Missing required fields");
}
} catch (parseError) {
console.error("[worker] Message parse failed:", parseError);
// ACK: malformed messages should not be retried
return res.status(200).json({ status: "discarded_parse_error" });
}
// Idempotency check
if (isProcessed(job.jobId)) {
console.log(`[worker] Duplicate detected for jobId=${job.jobId}, ACK-ing`);
return res.status(200).json({ status: "duplicate_acked" });
}
console.log(`[worker] Processing jobId=${job.jobId} type=${job.jobType}`);
let aiResult: { text: string; tokens: number } | null = null;
let processingStatus: "success" | "error" = "error";
let errorMessage: string | null = null;
try {
aiResult = await processWithGemini(job.inputText, job.jobType);
processingStatus = "success";
markProcessed(job.jobId);
} catch (aiError) {
errorMessage = aiError instanceof Error ? aiError.message : String(aiError);
console.error(`[worker] AI processing failed for jobId=${job.jobId}:`, aiError);
}
const processingTimeMs = Date.now() - startTime;
// Write to Firestore for serving results back to users (synchronous)
if (processingStatus === "success" && aiResult) {
try {
await firestore.collection("ai_jobs").doc(job.jobId).set({
jobId: job.jobId,
userId: job.userId,
jobType: job.jobType,
status: "completed",
result: aiResult.text,
completedAt: new Date().toISOString(),
});
} catch (fsError) {
console.error(`[worker] Firestore write failed for ${job.jobId}:`, fsError);
// Firestore write failure = treat as processing failure, NACK to retry
return res.status(500).json({ status: "error_firestore", jobId: job.jobId });
}
}
// Write to BigQuery for analytics (async, best-effort)
insertToBigQuery({
job_id: job.jobId,
user_id: job.userId,
job_type: job.jobType,
input_length: job.inputText.length,
output_text: aiResult?.text ?? null,
output_tokens: aiResult?.tokens ?? 0,
processing_time_ms: processingTimeMs,
created_at: job.createdAt,
completed_at: new Date().toISOString(),
status: processingStatus,
error_message: errorMessage,
}).catch((bqError) => {
// BigQuery failure does not affect ACK/NACK decision
console.error(`[worker] BigQuery insert failed for ${job.jobId}:`, bqError);
});
// Notify callback URL if provided
if (job.callbackUrl && processingStatus === "success" && aiResult) {
notifyCallback(job.callbackUrl, {
jobId: job.jobId,
status: "completed",
result: aiResult.text,
}).catch((cbError) => console.warn(`[worker] Callback failed for ${job.jobId}:`, cbError));
}
if (processingStatus === "success") {
return res.status(200).json({ status: "success", jobId: job.jobId, processingTimeMs });
} else {
// NACK: Pub/Sub will retry up to max-delivery-attempts, then route to DLQ
return res.status(500).json({ status: "error", jobId: job.jobId, errorMessage });
}
});
async function processWithGemini(
text: string,
jobType: "summarize" | "classify" | "extract"
): Promise<{ text: string; tokens: number }> {
const model = vertexAI.getGenerativeModel({
model: "gemini-2.0-flash",
generationConfig: {
maxOutputTokens: 1024,
temperature: 0.2,
},
});
const systemPrompts: Record<string, string> = {
summarize: "You are a precise summarization assistant. Summarize clearly and concisely.",
classify: "You are a text classification assistant. Return valid JSON only, no explanation.",
extract: "You are a keyword extraction assistant. Return a valid JSON array only, no explanation.",
};
const userPrompts: Record<string, string> = {
summarize: `Summarize the following in 3-5 sentences:\n\n${text}`,
classify: `Classify this text as one of: ["positive", "negative", "neutral", "technical", "other"]. Return: {"category": "<value>"}\n\n${text}`,
extract: `Extract up to 10 key terms from this text. Return: ["term1", "term2", ...]\n\n${text}`,
};
const response = await model.generateContent({
systemInstruction: systemPrompts[jobType],
contents: [{ role: "user", parts: [{ text: userPrompts[jobType] }] }],
});
const candidate = response.response.candidates?.[0];
if (!candidate?.content?.parts?.[0]?.text) {
throw new Error("Gemini API returned empty or invalid response");
}
return {
text: candidate.content.parts[0].text.trim(),
tokens: response.response.usageMetadata?.totalTokenCount ?? 0,
};
}
async function insertToBigQuery(row: Record<string, unknown>): Promise<void> {
const table = bigquery.dataset(DATASET_ID).table(TABLE_ID);
await table.insert([row]);
}
async function notifyCallback(url: string, payload: unknown): Promise<void> {
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), 5000);
try {
const response = await fetch(url, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(payload),
signal: controller.signal,
});
if (!response.ok) {
throw new Error(`Callback returned HTTP ${response.status}`);
}
} finally {
clearTimeout(timer);
}
}
app.get("/health", (_req, res) => res.json({ status: "ok" }));
const PORT = parseInt(process.env.PORT ?? "8080");
app.listen(PORT, () => console.log(`[worker] Listening on :${PORT}`));The key design decisions here are worth calling out explicitly:
The Firestore write is synchronous and blocks the ACK decision. If Firestore fails, we NACK and retry. The BigQuery write is fire-and-forget. If BigQuery fails, we log the error but still ACK. This reflects the different roles of the two stores: Firestore is part of the product's correctness guarantee (users need to retrieve their results), while BigQuery is observability infrastructure.
Pub/Sub Infrastructure Configuration
Infrastructure misconfigurations are a common source of subtle bugs. Here's the complete setup:
export GCP_PROJECT_ID="your-project-id"
export GCP_REGION="us-central1"
# Create main topic
gcloud pubsub topics create ai-processing-jobs --project=$GCP_PROJECT_ID
# Create DLQ topic
gcloud pubsub topics create ai-processing-jobs-dlq --project=$GCP_PROJECT_ID
# Create DLQ subscription (for manual inspection)
gcloud pubsub subscriptions create ai-processing-jobs-dlq-sub \
--topic=ai-processing-jobs-dlq \
--project=$GCP_PROJECT_ID
# Create a dedicated service account for Pub/Sub to use when calling Cloud Run
gcloud iam service-accounts create pubsub-invoker \
--display-name="Pub/Sub Cloud Run Invoker" \
--project=$GCP_PROJECT_ID
# Grant the Pub/Sub service agent permission to create tokens on behalf of pubsub-invoker
PROJECT_NUMBER=$(gcloud projects describe $GCP_PROJECT_ID --format="value(projectNumber)")
gcloud iam service-accounts add-iam-policy-binding \
"pubsub-invoker@$GCP_PROJECT_ID.iam.gserviceaccount.com" \
--member="serviceAccount:service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com" \
--role="roles/iam.serviceAccountTokenCreator" \
--project=$GCP_PROJECT_ID
# Create the push subscription pointing to the worker Cloud Run service
WORKER_URL="https://worker-service-$(gcloud run services describe worker-service --region=$GCP_REGION --format='value(status.url)' | sed 's/https:\/\///')/pubsub/push"
gcloud pubsub subscriptions create ai-processing-worker \
--topic=ai-processing-jobs \
--push-endpoint="$WORKER_URL" \
--push-auth-service-account="pubsub-invoker@$GCP_PROJECT_ID.iam.gserviceaccount.com" \
--ack-deadline=300 \
--max-delivery-attempts=5 \
--dead-letter-topic=ai-processing-jobs-dlq \
--min-retry-delay=10s \
--max-retry-delay=300s \
--project=$GCP_PROJECT_ID
# Grant pubsub-invoker permission to invoke the Cloud Run worker
gcloud run services add-iam-policy-binding worker-service \
--member="serviceAccount:pubsub-invoker@$GCP_PROJECT_ID.iam.gserviceaccount.com" \
--role="roles/run.invoker" \
--region=$GCP_REGION \
--project=$GCP_PROJECT_ID
echo "✅ Pub/Sub infrastructure configured"The --ack-deadline=300 gives the worker 5 minutes to process each message. For AI workloads involving large inputs or multiple model calls, this should comfortably cover your P99 processing time. The --min-retry-delay=10s and --max-retry-delay=300s implement exponential backoff between retries, which is important for avoiding thundering herd situations when the AI API is temporarily degraded.
BigQuery Schema and Useful Analytics Queries
[
{ "name": "job_id", "type": "STRING", "mode": "REQUIRED",
"description": "Unique job identifier" },
{ "name": "user_id", "type": "STRING", "mode": "REQUIRED",
"description": "ID of the user who submitted the job" },
{ "name": "job_type", "type": "STRING", "mode": "REQUIRED",
"description": "summarize | classify | extract" },
{ "name": "input_length", "type": "INTEGER", "mode": "NULLABLE",
"description": "Character count of input text" },
{ "name": "output_text", "type": "STRING", "mode": "NULLABLE",
"description": "AI-generated output" },
{ "name": "output_tokens", "type": "INTEGER", "mode": "NULLABLE",
"description": "Total tokens consumed by Gemini" },
{ "name": "processing_time_ms", "type": "INTEGER", "mode": "NULLABLE",
"description": "End-to-end processing time in milliseconds" },
{ "name": "status", "type": "STRING", "mode": "REQUIRED",
"description": "success | error" },
{ "name": "error_message", "type": "STRING", "mode": "NULLABLE",
"description": "Error detail if status=error" },
{ "name": "created_at", "type": "TIMESTAMP", "mode": "REQUIRED",
"description": "When the job was submitted" },
{ "name": "completed_at", "type": "TIMESTAMP", "mode": "NULLABLE",
"description": "When processing completed" }
]With this schema, the following queries cover most operational needs:
-- Daily success rate and token cost (last 30 days)
SELECT
DATE(created_at) AS date,
job_type,
COUNT(*) AS total,
COUNTIF(status = 'success') AS successes,
ROUND(COUNTIF(status = 'success') * 100.0 / COUNT(*), 1) AS success_rate_pct,
ROUND(AVG(IF(status = 'success', processing_time_ms, NULL)) / 1000.0, 2) AS avg_sec,
SUM(output_tokens) AS total_tokens
FROM `your-project.ai_pipeline_results.processing_jobs`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY 1, 2
ORDER BY 1 DESC, total DESC;
-- Top users by usage (last 7 days)
SELECT
user_id,
COUNT(*) AS job_count,
SUM(output_tokens) AS tokens_used,
ROUND(AVG(processing_time_ms) / 1000.0, 2) AS avg_processing_sec
FROM `your-project.ai_pipeline_results.processing_jobs`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
AND status = 'success'
GROUP BY user_id
ORDER BY tokens_used DESC
LIMIT 20;
-- Error pattern analysis
SELECT
REGEXP_EXTRACT(error_message, r'^[^:]+') AS error_type,
job_type,
COUNT(*) AS occurrences,
MAX(created_at) AS last_seen
FROM `your-project.ai_pipeline_results.processing_jobs`
WHERE status = 'error'
AND created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY 1, 2
ORDER BY occurrences DESC;For connecting BigQuery to Looker Studio dashboards, see Antigravity × Firebase, BigQuery & Looker Studio Revenue Dashboard Guide.
Production Monitoring and Alerting
Three alerts cover the critical failure modes for this pipeline:
DLQ backlog alert. Messages accumulating in the DLQ signal that processing is persistently failing for certain jobs. A Cloud Monitoring alert on pubsub.googleapis.com/subscription/num_undelivered_messages for the DLQ subscription firing above 10 messages is a good starting threshold.
Processing latency alert. A BigQuery scheduled query running every 15 minutes that checks P95 processing_time_ms over the last hour and writes to a metrics table, with a Cloud Monitoring custom metric alert on top, covers this well.
Error rate spike. Cloud Run log-based metrics can surface when the worker is returning 500 at an elevated rate, indicating the AI API is degraded or something changed in the processing logic.
# Terraform: DLQ alert policy
resource "google_monitoring_alert_policy" "dlq_backlog" {
display_name = "AI Pipeline — DLQ Backlog"
combiner = "OR"
conditions {
display_name = "DLQ messages exceed threshold"
condition_threshold {
filter = join(" AND ", [
"resource.type=\"pubsub_subscription\"",
"metric.type=\"pubsub.googleapis.com/subscription/num_undelivered_messages\"",
"resource.label.subscription_id=\"ai-processing-jobs-dlq-sub\""
])
comparison = "COMPARISON_GT"
threshold_value = 10
duration = "60s"
aggregations {
alignment_period = "60s"
per_series_aligner = "ALIGN_MAX"
}
}
}
notification_channels = [var.notification_channel_id]
alert_strategy {
auto_close = "3600s"
}
}For Cloud Run deployment setup, see Antigravity × Cloud Run Serverless Deployment Guide.
Four Pitfalls That Will Catch You in Production
① Push authentication 403 errors. When Cloud Run returns 403 to Pub/Sub push requests, the underlying cause is almost always that the Pub/Sub service agent (service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com) is missing the iam.serviceAccountTokenCreator role on your pubsub-invoker service account. Pub/Sub needs to mint OIDC tokens to authenticate with Cloud Run, and the serviceAccountTokenCreator role is what grants that permission. The fix is the gcloud iam service-accounts add-iam-policy-binding command shown in the infrastructure setup above.
② ack_deadline exceeded causing NACK loops. If Gemini processing takes longer than the ack_deadline, Pub/Sub NACKs the message automatically and redelivers it — before your worker has finished the first attempt. Suddenly you have two or more workers trying to process the same message concurrently. Setting ack_deadline=300 in the subscription configuration (as shown above) handles most cases. For workloads with more variance, you can also call modifyAckDeadline programmatically mid-processing:
// Extend the ack deadline if processing is taking longer than expected
async function extendAckDeadline(
subscriptionName: string,
ackId: string,
extensionSeconds: number
): Promise<void> {
const { PubSub } = await import("@google-cloud/pubsub");
const pubsub = new PubSub({ projectId: process.env.GCP_PROJECT_ID });
const subscription = pubsub.subscription(subscriptionName);
await subscription.modifyAckDeadline(ackId, extensionSeconds);
}Note: ackId is only available with pull subscriptions. With push, set a generous ack_deadline upfront.
③ BigQuery schema mismatch causing NACK loops. If table.insert() fails due to a schema mismatch, the worker returns 500, Pub/Sub redelivers, it fails again, and the cycle continues until the message exhausts its retry attempts and hits the DLQ. The clean solution is schema-strict validation before attempting insertion. If a row doesn't match the schema, log the anomaly and ACK — you don't want a malformed event to block the rest of the pipeline.
④ Pub/Sub message ordering with parallel workers. By default, Pub/Sub doesn't guarantee message ordering across multiple concurrent deliveries. If your use case requires that jobs from the same user are processed in submission order (e.g., conversation context that builds on previous turns), you need to enable message ordering on the subscription and set orderingKey on published messages. This limits parallelism within an ordering key — a deliberate tradeoff.
Production-Ready Idempotency with Redis
As mentioned, the in-memory cache in the sample code breaks with multiple Cloud Run instances. Here's the Redis-based implementation using Upstash (which works well with Cloud Run because it's HTTP-based, not a persistent TCP connection):
import { Redis } from "@upstash/redis";
const redis = new Redis({
url: process.env.UPSTASH_REDIS_REST_URL!,
token: process.env.UPSTASH_REDIS_REST_TOKEN!,
});
const IDEMPOTENCY_TTL_SECONDS = 3600; // 1 hour
async function checkAndMarkProcessed(jobId: string): Promise<boolean> {
const key = `ai-pipeline:processed:${jobId}`;
// SET NX: only set if key doesn't exist
// Returns "OK" if set (first time), null if key existed (duplicate)
const result = await redis.set(key, Date.now().toString(), {
nx: true,
ex: IDEMPOTENCY_TTL_SECONDS,
});
// null means the key already existed → this is a duplicate
return result !== null; // true = first time, false = duplicate
}
// Usage in the worker:
const isFirstProcessing = await checkAndMarkProcessed(job.jobId);
if (!isFirstProcessing) {
console.log(`[worker] Duplicate jobId=${job.jobId} detected via Redis, ACK-ing`);
return res.status(200).json({ status: "duplicate_acked" });
}The Redis SET NX EX is atomic — no race condition between checking and setting. This is the reliable cross-instance idempotency pattern. For development, you can run Redis locally in Docker. For production on Google Cloud, Cloud Memorystore (Redis) is the managed option, though Upstash's serverless pricing often makes more sense for workloads with variable traffic. For setting up development containers, Antigravity × DevContainer Reproducible AI Dev Environment covers the container-based workflow that pairs well with this pipeline.
Tips for Working With Antigravity on This Architecture
Antigravity can generate most of this pipeline from a natural language description. The output quality improves substantially when you provide architectural constraints upfront rather than asking for "a Pub/Sub pipeline" and then correcting it iteratively.
Concretely, tell Antigravity:
- The maximum expected processing time per job (this informs
ack_deadline) - Whether you need cross-instance idempotency (Redis) or single-instance is fine
- Your BigQuery schema if you already have one (otherwise it'll generate one you'll want to modify)
- Which failure modes must never result in lost data (Firestore write path) vs. which are best-effort (BigQuery analytics path)
One technique that's consistently useful: after generating the worker code, ask Antigravity to enumerate every scenario where the code would return a 500 to Pub/Sub. This surfaces edge cases — API timeout, parsing failure, partial success — that belong in your test suite. The resulting test coverage is significantly better than what you'd write from scratch.
The Next Step
Moving from synchronous to async changes something beyond the technical architecture — it changes how you experience production incidents. When something fails, the DLQ catches it, retries run automatically, and BigQuery has the complete record. You move from "scrambling to figure out what happened and reconstruct lost data" to "querying BigQuery to read exactly what happened."
The most practical starting point: identify the single heaviest AI API call in your current system — the one most likely to timeout or fail under load — and extract it into this pattern. The ingestion/worker split takes less than an hour to implement. Start with one job type, validate the pattern in staging, and expand from there.