DEV Community: Humza Tareen

The Firestore Default Database Trap: Why Your Data Is Going to the Wrong Place

Humza Tareen — Mon, 23 Mar 2026 11:30:00 +0000

Firestore has a (default) database. If you don't explicitly specify which database to use, everything routes there. We had multiple Firestore databases in production, but several code paths were accidentally hitting the default.

This guide covers how Firestore's default database works, how to detect misrouting, and how to fix it in Python and JavaScript/React.

The Problem: Silent Data Misrouting

We use multiple Firestore databases for tenant isolation. Each evaluation tenant has its own database:

evaluations-db-prod-tenant-1
evaluations-db-prod-tenant-2
evaluations-db-prod-tenant-3
...
evaluations-db-prod-tenant-12

But some code paths were missing explicit database references:

# ❌ BAD: Routes to (default) database
from google.cloud import firestore

db = firestore.Client()  # No database specified!
db.collection("evaluations").add({"score": 0.95})

This code writes to (default), not the tenant-specific database. The bug was silent — no errors, just wrong data location.

How Firestore Defaults Work

Firestore has two ways to specify a database:

1. Explicit Database Reference (Correct)

from google.cloud import firestore

# Specify database ID explicitly
db = firestore.Client(database="evaluations-db-prod-tenant-1")
db.collection("evaluations").add({"score": 0.95})

2. Default Database (What Happens If You Don't Specify)

# No database specified → routes to (default)
db = firestore.Client()  # Uses (default) database!

The (default) database is created automatically when you first use Firestore. It's always there, even if you never intended to use it.

Detecting Misrouting

Method 1: Check Database Usage in Console

Go to Firebase Console → Firestore → Databases. Check if (default) has unexpected collections:

(default) database:
  - evaluations (shouldn't be here!)
  - scores (shouldn't be here!)

evaluations-db-prod-tenant-1:
  - evaluations (correct)
  - scores (correct)

Method 2: Query Default Database Explicitly

# Check what's in (default)
default_db = firestore.Client(database="(default)")
evaluations = default_db.collection("evaluations").stream()

for doc in evaluations:
    print(f"Found evaluation in default DB: {doc.id}")
    # This shouldn't exist!

Method 3: Add Logging

import logging

def get_firestore_client(database_id: str):
    """Get Firestore client with explicit database."""
    if not database_id:
        logging.error("Database ID is required!")
        raise ValueError("Database ID must be specified")

    client = firestore.Client(database=database_id)
    logging.info(f"Using Firestore database: {database_id}")
    return client

# Usage
db = get_firestore_client("evaluations-db-prod-tenant-1")

Fixing Python Code

Pattern 1: Environment Variable

import os
from google.cloud import firestore

# Get database ID from environment
DATABASE_ID = os.getenv("FIRESTORE_DATABASE_ID")
if not DATABASE_ID:
    raise ValueError("FIRESTORE_DATABASE_ID environment variable must be set")

db = firestore.Client(database=DATABASE_ID)

Pattern 2: Configuration Class

from dataclasses import dataclass
from google.cloud import firestore

@dataclass
class FirestoreConfig:
    database_id: str
    project_id: str = None

    def get_client(self):
        """Get Firestore client with explicit database."""
        kwargs = {"database": self.database_id}
        if self.project_id:
            kwargs["project"] = self.project_id
        return firestore.Client(**kwargs)

# Usage
config = FirestoreConfig(
    database_id="evaluations-db-prod-tenant-1",
    project_id="my-project"
)
db = config.get_client()

Pattern 3: Factory Function

from google.cloud import firestore
from functools import lru_cache

@lru_cache(maxsize=12)  # Cache clients for each database
def get_firestore_client(database_id: str) -> firestore.Client:
    """Get Firestore client for a specific database."""
    if not database_id:
        raise ValueError("database_id is required")

    return firestore.Client(database=database_id)

# Usage
db = get_firestore_client("evaluations-db-prod-tenant-1")

Fixing JavaScript/React Code

Frontend: Firebase SDK

// ❌ BAD: Uses default database
import { getFirestore } from 'firebase/firestore';

const db = getFirestore(app);  // No database specified!

// ✅ GOOD: Explicit database reference
import { getFirestore } from 'firebase/firestore';

const db = getFirestore(app, 'evaluations-db-prod-tenant-1');

React Hook Pattern

import { useState, useEffect } from 'react';
import { getFirestore, collection, onSnapshot } from 'firebase/firestore';
import { app } from './firebase-config';

function useFirestoreCollection(collectionName, databaseId) {
  const [data, setData] = useState([]);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState(null);

  useEffect(() => {
    if (!databaseId) {
      setError('Database ID is required');
      return;
    }

    // Explicit database reference
    const db = getFirestore(app, databaseId);
    const colRef = collection(db, collectionName);

    const unsubscribe = onSnapshot(
      colRef,
      (snapshot) => {
        const docs = snapshot.docs.map(doc => ({
          id: doc.id,
          ...doc.data()
        }));
        setData(docs);
        setLoading(false);
      },
      (err) => {
        setError(err.message);
        setLoading(false);
      }
    );

    return () => unsubscribe();
  }, [collectionName, databaseId]);

  return { data, loading, error };
}

// Usage
function EvaluationsList({ tenantId }) {
  const databaseId = `evaluations-db-prod-tenant-${tenantId}`;
  const { data, loading, error } = useFirestoreCollection('evaluations', databaseId);

  if (loading) return <div>Loading...</div>;
  if (error) return <div>Error: {error}</div>;

  return (
    <ul>
      {data.map(eval => (
        <li key={eval.id}>{eval.id}</li>
      ))}
    </ul>
  );
}

Backend: Admin SDK

// ❌ BAD: Uses default database
const admin = require('firebase-admin');
const db = admin.firestore();

// ✅ GOOD: Explicit database reference
const admin = require('firebase-admin');
const db = admin.firestore().database('evaluations-db-prod-tenant-1');

Real-Time Listeners on Non-Default Databases

Real-time listeners must also specify the database:

// ❌ BAD: Listens to default database
import { collection, onSnapshot } from 'firebase/firestore';

const db = getFirestore(app);  // Missing database ID!
const colRef = collection(db, 'evaluations');

onSnapshot(colRef, (snapshot) => {
  // This listens to (default) database, not tenant-specific!
});

// ✅ GOOD: Explicit database in listener
const db = getFirestore(app, 'evaluations-db-prod-tenant-1');
const colRef = collection(db, 'evaluations');

onSnapshot(colRef, (snapshot) => {
  // Now listening to the correct database
});

Auto-Export Configuration

If you're using Firestore auto-export (for backups or analytics), make sure it's configured for the correct databases:

# Export specific database (not default)
gcloud firestore export gs://BUCKET_NAME/export \
  --database-ids=evaluations-db-prod-tenant-1 \
  --collection-ids=evaluations,scores

Gotcha: If you don't specify --database-ids, the export includes (default) and all other databases. This can be expensive and slow.

Migration: Moving Data from Default

If you've already written data to (default), you need to migrate it:

from google.cloud import firestore
from google.cloud.firestore_v1 import Query

def migrate_from_default(target_database_id: str, collection_name: str):
    """Migrate data from (default) to target database."""
    default_db = firestore.Client(database="(default)")
    target_db = firestore.Client(database=target_database_id)

    # Read from default
    docs = default_db.collection(collection_name).stream()

    # Write to target
    batch = target_db.batch()
    count = 0

    for doc in docs:
        batch.set(
            target_db.collection(collection_name).document(doc.id),
            doc.to_dict()
        )
        count += 1

        # Commit in batches of 500 (Firestore limit)
        if count % 500 == 0:
            batch.commit()
            batch = target_db.batch()
            print(f"Migrated {count} documents...")

    # Commit remaining
    if count % 500 != 0:
        batch.commit()

    print(f"Migration complete: {count} documents migrated")

# Usage
migrate_from_default("evaluations-db-prod-tenant-1", "evaluations")

Prevention: Code Review Checklist

Add this to your PR template:

## Firestore Database Checklist
- [ ] All Firestore clients specify explicit `database` parameter
- [ ] No `firestore.Client()` calls without database ID
- [ ] Environment variables set for database IDs
- [ ] Frontend real-time listeners specify database
- [ ] Tests use explicit database IDs (not default)

Complete Example: Production-Ready Pattern

import os
from typing import Optional
from google.cloud import firestore
from functools import lru_cache
import logging

class FirestoreManager:
    """Manages Firestore clients with explicit database references."""

    def __init__(self, project_id: Optional[str] = None):
        self.project_id = project_id or os.getenv("GCP_PROJECT_ID")
        if not self.project_id:
            raise ValueError("GCP_PROJECT_ID must be set")

    @lru_cache(maxsize=20)
    def get_client(self, database_id: str) -> firestore.Client:
        """Get Firestore client for a specific database."""
        if not database_id:
            raise ValueError("database_id is required")

        if database_id == "(default)":
            logging.warning("Using (default) database - is this intentional?")

        client = firestore.Client(
            project=self.project_id,
            database=database_id
        )
        logging.info(f"Created Firestore client for database: {database_id}")
        return client

    def get_tenant_database(self, tenant_id: str) -> firestore.Client:
        """Get Firestore client for a tenant-specific database."""
        database_id = f"evaluations-db-prod-tenant-{tenant_id}"
        return self.get_client(database_id)

# Usage
manager = FirestoreManager()

# Get tenant-specific database
db = manager.get_tenant_database("1")
db.collection("evaluations").add({"score": 0.95})

# Explicit database
db = manager.get_client("evaluations-db-prod-tenant-2")

TL;DR

Problem	Solution
Code routes to `(default)` database	Always specify `database` parameter
Silent data misrouting	Check console, add logging, query default DB
Frontend uses wrong database	Pass database ID to `getFirestore()`
Real-time listeners wrong DB	Specify database in listener setup
Auto-export includes default	Use `--database-ids` flag

Key takeaway: Never rely on Firestore defaults. Always specify the database ID explicitly. The (default) database is a trap — it's always there, even when you don't want it.

More production GCP articles on my blog. I write about patterns from real infrastructure — find me at humzakt.github.io.

Redis State Management on Cloud Run: Handling Missing Keys and Silent Failures

Humza Tareen — Fri, 20 Mar 2026 11:30:00 +0000

Redis HSET always returns a value, even when the key doesn't exist. This led to a subtle bug in our production system: state updates were logging success when they were actually failing silently.

This guide covers Redis state management patterns for long-running Cloud Run tasks, including how to detect missing keys, handle socket timeouts, and properly configure VPC networking for Cloud Run + Memorystore.

The Problem: Silent Failures

We use Redis (Memorystore) to track state for long-running evaluation tasks. Each task updates its progress periodically:

# ❌ BAD: This logs success even when the key doesn't exist
import redis
import logging

r = redis.Redis(host='10.x.x.x', port=6379)

def update_task_progress(task_id: str, progress: int):
    result = r.hset(f"task:{task_id}", "progress", progress)
    if result:
        logging.info(f"Updated progress for {task_id}: {progress}%")
    else:
        logging.warning(f"Failed to update progress for {task_id}")

The bug: HSET returns 1 if a new field was added, 0 if the field already existed. But if the key task:{task_id} doesn't exist, HSET creates it and returns 1. So the code logs success even when the key was missing.

Understanding Redis HSET Behavior

Let's clarify what HSET actually does:

# Key exists, field exists → returns 0 (field updated)
r.hset("task:123", "progress", 50)  # Returns 0

# Key exists, field doesn't exist → returns 1 (new field added)
r.hset("task:123", "status", "running")  # Returns 1

# Key doesn't exist → creates key, returns 1 (new field added)
r.hset("task:456", "progress", 25)  # Returns 1 (key didn't exist!)

The return value tells you if a field was added, not if the key existed.

The Fix: Check Key Existence First

For state management, you usually want to ensure the key exists before updating:

# ✅ GOOD: Check key existence first
def update_task_progress(task_id: str, progress: int):
    key = f"task:{task_id}"

    # Check if key exists
    if not r.exists(key):
        logging.error(f"Task {task_id} not found in Redis")
        # Create recovery record or raise exception
        create_recovery_record(task_id, "missing_key")
        return False

    # Update progress
    result = r.hset(key, "progress", progress)
    logging.info(f"Updated progress for {task_id}: {progress}%")
    return True

Creating Recovery Records

When you detect a missing key, log it somewhere persistent (not just Redis):

from datetime import datetime
import json

def create_recovery_record(task_id: str, error_type: str):
    """Log missing keys to Firestore or Cloud SQL for recovery."""
    recovery_record = {
        "task_id": task_id,
        "error_type": error_type,
        "timestamp": datetime.utcnow().isoformat(),
        "status": "needs_recovery"
    }

    # Write to Firestore or Cloud SQL
    db.collection("recovery_records").add(recovery_record)
    logging.error(f"Created recovery record: {recovery_record}")

Handling Socket Timeouts on Startup

Cloud Run services can start cold. If Redis isn't ready (or VPC networking isn't established), you'll get socket timeouts:

# ✅ GOOD: Retry with exponential backoff on startup
import time
from redis.exceptions import ConnectionError, TimeoutError

def get_redis_client(max_retries=5):
    """Get Redis client with retry logic for cold starts."""
    r = redis.Redis(
        host=os.getenv("REDIS_HOST"),
        port=int(os.getenv("REDIS_PORT", 6379)),
        socket_connect_timeout=5,
        socket_timeout=5,
        retry_on_timeout=True,
        health_check_interval=30
    )

    # Retry on connection failure (cold start)
    for attempt in range(max_retries):
        try:
            r.ping()  # Test connection
            return r
        except (ConnectionError, TimeoutError) as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # Exponential backoff
            logging.warning(f"Redis connection failed (attempt {attempt + 1}/{max_retries}), retrying in {wait_time}s...")
            time.sleep(wait_time)

    return r

# Use in your service
r = get_redis_client()

Cloud Run + Memorystore VPC Requirements

Memorystore Redis uses private IPs. Your Cloud Run service needs VPC access:

Option 1: VPC Connector (Recommended)

# Create VPC Connector (if you don't have one)
gcloud compute networks vpc-access connectors create redis-connector \
  --region us-central1 \
  --subnet default \
  --subnet-project PROJECT_ID \
  --min-instances 2 \
  --max-instances 3

# Deploy Cloud Run service with VPC Connector
gcloud run deploy evaluation-service \
  --image gcr.io/PROJECT/service:latest \
  --vpc-connector projects/PROJECT/locations/us-central1/connectors/redis-connector \
  --vpc-egress private-ranges-only \
  --set-env-vars REDIS_HOST=10.x.x.x,REDIS_PORT=6379

Why private-ranges-only? Your service only needs VPC access for Redis. Public internet traffic (API calls, Cloud Tasks) can go through the default route.

Option 2: Direct VPC Egress

# Deploy with Direct VPC Egress
gcloud run deploy evaluation-service \
  --image gcr.io/PROJECT/service:latest \
  --network projects/PROJECT/global/networks/VPC_NAME \
  --subnet projects/PROJECT/regions/us-central1/subnetworks/SUBNET_NAME \
  --set-env-vars REDIS_HOST=10.x.x.x,REDIS_PORT=6379

Note: You can't use both VPC Connector and Direct VPC Egress. Pick one.

Complete Example: State Management with Error Handling

Here's a production-ready pattern:

import redis
import logging
import os
from typing import Optional, Dict
from datetime import datetime, timedelta

class TaskStateManager:
    def __init__(self):
        self.redis = redis.Redis(
            host=os.getenv("REDIS_HOST"),
            port=int(os.getenv("REDIS_PORT", 6379)),
            socket_connect_timeout=5,
            socket_timeout=5,
            retry_on_timeout=True,
            health_check_interval=30,
            decode_responses=True  # Return strings, not bytes
        )
        self.ttl = int(os.getenv("REDIS_TTL_SECONDS", 3600))  # 1 hour default

    def initialize_task(self, task_id: str, initial_state: Dict) -> bool:
        """Create a new task state record."""
        key = f"task:{task_id}"

        # Check if already exists
        if self.redis.exists(key):
            logging.warning(f"Task {task_id} already exists, skipping initialization")
            return False

        # Set initial state
        self.redis.hset(key, mapping=initial_state)
        self.redis.expire(key, self.ttl)  # Set TTL

        logging.info(f"Initialized task {task_id}")
        return True

    def update_progress(self, task_id: str, progress: int) -> bool:
        """Update task progress. Returns False if task doesn't exist."""
        key = f"task:{task_id}"

        # Check existence first
        if not self.redis.exists(key):
            logging.error(f"Task {task_id} not found in Redis")
            self._create_recovery_record(task_id, "missing_key_on_update")
            return False

        # Update progress
        self.redis.hset(key, "progress", progress)
        self.redis.hset(key, "last_updated", datetime.utcnow().isoformat())

        logging.info(f"Updated progress for {task_id}: {progress}%")
        return True

    def get_state(self, task_id: str) -> Optional[Dict]:
        """Get current task state."""
        key = f"task:{task_id}"

        if not self.redis.exists(key):
            return None

        return self.redis.hgetall(key)

    def complete_task(self, task_id: str, final_state: Dict) -> bool:
        """Mark task as complete and store final state."""
        key = f"task:{task_id}"

        if not self.redis.exists(key):
            logging.error(f"Task {task_id} not found when completing")
            self._create_recovery_record(task_id, "missing_key_on_complete")
            return False

        # Update final state
        final_state["status"] = "completed"
        final_state["completed_at"] = datetime.utcnow().isoformat()
        self.redis.hset(key, mapping=final_state)

        # Extend TTL for completed tasks (keep for 24 hours)
        self.redis.expire(key, 86400)

        logging.info(f"Completed task {task_id}")
        return True

    def _create_recovery_record(self, task_id: str, error_type: str):
        """Log missing keys for recovery."""
        # In production, write to Firestore or Cloud SQL
        logging.error(f"Recovery needed: task_id={task_id}, error_type={error_type}")

# Usage
state_manager = TaskStateManager()

# Initialize task
state_manager.initialize_task("eval-123", {
    "status": "running",
    "progress": 0,
    "started_at": datetime.utcnow().isoformat()
})

# Update progress
state_manager.update_progress("eval-123", 50)

# Get state
state = state_manager.get_state("eval-123")
print(state)  # {'status': 'running', 'progress': '50', ...}

# Complete task
state_manager.complete_task("eval-123", {
    "score": 0.95,
    "duration_seconds": 120
})

Testing Redis State Management

import pytest
from unittest.mock import Mock, patch

def test_update_progress_missing_key():
    """Test that missing keys are detected."""
    state_manager = TaskStateManager()
    state_manager.redis = Mock()
    state_manager.redis.exists.return_value = False  # Key doesn't exist

    result = state_manager.update_progress("task-123", 50)

    assert result is False
    state_manager.redis.hset.assert_not_called()  # Should not update

def test_update_progress_existing_key():
    """Test that existing keys are updated."""
    state_manager = TaskStateManager()
    state_manager.redis = Mock()
    state_manager.redis.exists.return_value = True  # Key exists

    result = state_manager.update_progress("task-123", 50)

    assert result is True
    state_manager.redis.hset.assert_called()  # Should update

TL;DR

Problem	Solution
`HSET` returns 1 even when key doesn't exist	Check `EXISTS` before updating
Silent failures in production	Create recovery records for missing keys
Socket timeouts on cold start	Retry with exponential backoff
VPC networking issues	Use VPC Connector or Direct VPC Egress
No visibility into Redis state	Log all state operations, use structured logging

Key takeaway: Always check key existence before updating state. Redis operations are fast, but missing keys indicate a deeper problem (task initialization failed, TTL expired, or data corruption).

More production GCP articles on my blog. I write about patterns from real infrastructure — find me at humzakt.github.io.

Cloud Run VPC Networking: The Two Modes That Can't Coexist

Humza Tareen — Wed, 18 Mar 2026 11:30:00 +0000

If you've ever seen the error "VPC connector and direct VPC can not be used together" when deploying a Cloud Run service, you've hit one of GCP's most confusing networking gotchas. It took us 3 days of debugging to figure out what was happening.

This guide explains the two VPC networking modes on Cloud Run, why they conflict, and how to fix it.

The Two Networking Modes

Cloud Run services can connect to VPC resources (like Cloud SQL, Memorystore Redis, or private GKE clusters) in two ways:

Mode 1: VPC Connector (Serverless VPC Access)

A VPC Connector is a managed service that creates a bridge between your Cloud Run service and your VPC network. It's the recommended approach for most use cases.

# Deploy with VPC Connector
gcloud run deploy my-service \
  --vpc-connector projects/PROJECT_ID/locations/REGION/connectors/CONNECTOR_NAME \
  --vpc-egress all-traffic

When to use:

You need access to private IP resources (Cloud SQL private IP, Memorystore, private GKE)
You want managed networking (Google handles scaling, HA)
You're okay with a small latency overhead (~1-2ms)

Mode 2: Direct VPC Egress

Direct VPC Egress assigns your Cloud Run service a VPC IP address directly. The service becomes part of your VPC network.

# Deploy with Direct VPC Egress
gcloud run deploy my-service \
  --network projects/PROJECT_ID/global/networks/VPC_NAME \
  --subnet projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME

When to use:

You need the lowest possible latency
You require VPC-native features (like custom routes)
You're using Cloud Run (fully managed) with specific VPC requirements

The Conflict

Here's the problem: You cannot use both modes simultaneously. If your Cloud Run service already has a VPC Connector configured, and you try to deploy with Direct VPC Egress flags (or vice versa), you'll get this error:

ERROR: (gcloud.run.deploy) INVALID_ARGUMENT: VPC connector and direct VPC can not be used together

Why This Happens

The error occurs when:

Existing service has one mode — Your service was previously deployed with a VPC Connector
Deploy command specifies the other — Your new deploy command includes Direct VPC Egress flags
GCP rejects the conflict — Cloud Run can't have both networking modes active

This is especially confusing because:

The error message doesn't tell you which mode the service currently uses
You might not remember how the service was originally deployed
The conflict only appears during deployment, not in the console

How to Diagnose

Check Current Configuration

First, check what networking mode your service is currently using:

# Get the current service configuration
gcloud run services describe SERVICE_NAME \
  --region REGION \
  --format="yaml(spec.template.metadata.annotations)"

Look for these annotations:

VPC Connector mode:

run.googleapis.com/vpc-access-connector: projects/.../connectors/...
run.googleapis.com/vpc-access-egress: all-traffic  # or private-ranges-only

Direct VPC Egress mode:

run.googleapis.com/execution-environment: gen2
run.googleapis.com/vpc-access-egress: all-traffic
# Plus network/subnet annotations

Check Your Deploy Command

If you're using Infrastructure as Code (Terraform, Pulumi, or YAML), check your configuration:

# service.yaml - VPC Connector mode
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: my-service
spec:
  template:
    metadata:
      annotations:
        run.googleapis.com/vpc-access-connector: projects/PROJECT/locations/REGION/connectors/CONNECTOR
        run.googleapis.com/vpc-access-egress: all-traffic

vs.

# service.yaml - Direct VPC Egress mode
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: my-service
spec:
  template:
    metadata:
      annotations:
        run.googleapis.com/execution-environment: gen2
        run.googleapis.com/vpc-access-egress: all-traffic
    spec:
      containers:
      - image: gcr.io/...

How to Fix

Option 1: Remove the Conflicting Configuration

If you want to use VPC Connector, remove Direct VPC Egress flags:

# Remove network/subnet flags, keep VPC Connector
gcloud run deploy my-service \
  --image gcr.io/PROJECT/service:latest \
  --vpc-connector projects/PROJECT/locations/REGION/connectors/CONNECTOR \
  --vpc-egress all-traffic
  # NO --network or --subnet flags

If you want to use Direct VPC Egress, remove VPC Connector:

# Remove VPC Connector, use Direct VPC Egress
gcloud run deploy my-service \
  --image gcr.io/PROJECT/service:latest \
  --network projects/PROJECT/global/networks/VPC_NAME \
  --subnet projects/PROJECT/regions/REGION/subnetworks/SUBNET_NAME
  # NO --vpc-connector flag

Option 2: Update via YAML

Export the current service configuration, edit it, and apply:

# Export current config
gcloud run services describe SERVICE_NAME \
  --region REGION \
  --format export > service.yaml

# Edit service.yaml to remove conflicting annotations
# Remove either VPC Connector OR Direct VPC Egress config

# Apply the updated config
gcloud run services replace service.yaml --region REGION

Option 3: Clean Slate (Nuclear Option)

If you're stuck, delete and recreate the service:

# Delete the service
gcloud run services delete SERVICE_NAME --region REGION

# Deploy fresh with your chosen mode
gcloud run deploy SERVICE_NAME \
  --image gcr.io/PROJECT/service:latest \
  --vpc-connector projects/PROJECT/locations/REGION/connectors/CONNECTOR \
  --vpc-egress all-traffic

Warning: This will cause downtime. Only use if the service can be safely recreated.

Real-World Example: Our Debugging Journey

Here's what happened to us:

Initial setup: Service was deployed with VPC Connector for Memorystore Redis access
The change: Someone added --network and --subnet flags to the deploy script (thinking it would improve performance)
The problem: Deployments started failing with the conflict error. We tried various workarounds, thinking it was a transient issue
The solution: Finally checked the existing service config, realized the conflict, removed the Direct VPC Egress flags

Lesson learned: Always check the current service configuration before changing networking settings.

Best Practices

Document your networking mode — Add comments in your deploy scripts or IaC configs
Use consistent patterns — Pick one mode per service and stick with it
Check before changing — Run gcloud run services describe before modifying networking
Prefer VPC Connector — Unless you have specific requirements, VPC Connector is simpler and more flexible
Version control your configs — Keep service YAML files in git so you can see what changed

Quick Reference

Mode	Flag	Use Case
VPC Connector	`--vpc-connector`	Most common, managed service
Direct VPC Egress	`--network` + `--subnet`	Low latency, VPC-native features

Remember: You can only use one mode at a time. If you see the conflict error, check your existing service config and remove the conflicting flags.

More production GCP articles on my blog. I write about patterns from real infrastructure — find me at humzakt.github.io.

How to Audit Production Code: A 5-Layer Bug Hunting Methodology

Humza Tareen — Mon, 09 Mar 2026 11:30:00 +0000

I filed 33 issues in one week across four production services. Not because the code was terrible — production codebases accumulate subtle bugs that only surface under specific conditions. This tutorial shows the 5-layer audit methodology I use, with code you can drop into your own FastAPI/Python services.

The 5-Layer Audit Approach

Audit in layers. Each layer has a different failure mode. Skipping one leaves bugs in production.

Layer	Question	Failure Mode
1. API Surface	Does every endpoint validate input?	OOM, injection, bad data
2. Auth & Ownership	Can user A access user B's resources?	Data leaks, privilege escalation
3. Data Integrity	Can the same input create duplicates?	Duplicate records, race conditions
4. Error Handling	What happens when dependencies fail?	Cascading failures, unbounded loads
5. Security	SSRF, path traversal, CORS?	External attacks, metadata exposure

Let's go through each layer with concrete checks and fixes.

Layer 1: API Surface Validation

The bug: We had limit=-1 accepted. Result: unbounded query returning the entire dataset. OOM risk.

Pydantic validators for query params

from pydantic import BaseModel, field_validator

class ListTasksParams(BaseModel):
    limit: int = 20
    offset: int = 0

    @field_validator("limit")
    @classmethod
    def limit_bounds(cls, v: int) -> int:
        if v < 1 or v > 100:
            raise ValueError("limit must be between 1 and 100")
        return v

    @field_validator("offset")
    @classmethod
    def offset_non_negative(cls, v: int) -> int:
        if v < 0:
            raise ValueError("offset must be >= 0")
        return v

FastAPI dependency for consistent validation

from fastapi import Depends, Query

def paginated(limit: int = Query(20, ge=1, le=100),
              offset: int = Query(0, ge=0)):
    return {"limit": limit, "offset": offset}

@app.get("/tasks")
async def list_tasks(pagination: dict = Depends(paginated)):
    return await db.get_tasks(
        limit=pagination["limit"],
        offset=pagination["offset"]
    )

Audit script: find unvalidated limits

# Grep for limit/offset without validation
rg "limit.*=.*Query|limit.*int" --type py -A 1
# If you see limit: int without ge/le, you have a bug

Layer 2: Auth & Ownership

The bug: Any authenticated user could read/update/cancel any task. We checked "is logged in" but not "owns this resource."

Ownership check pattern

async def get_task_or_403(task_id: str, user: User) -> Task:
    task = await db.get(Task, id=task_id)
    if not task:
        raise HTTPException(404, "Task not found")
    if task.user_id != user.id:
        raise HTTPException(403, "Forbidden")
    return task

@app.get("/tasks/{task_id}")
async def get_task(task_id: str, user: User = Depends(require_auth)):
    task = await get_task_or_403(task_id, user)
    return task

Audit script: find missing ownership checks

# For each endpoint that takes an ID, verify:
# 1. Fetches the resource
# 2. Compares resource.owner_id (or similar) to request.user.id
# 3. Returns 403 if mismatch

# Red flags: endpoints with {id} in path but no ownership comparison

Test: can user A access user B's task?

def test_ownership_enforced():
    user_a = create_user("a")
    user_b = create_user("b")
    task_b = create_task(user_id=user_b.id)

    # User A tries to access User B's task
    response = client.get(
        f"/tasks/{task_b.id}",
        headers=auth_headers(user_a)
    )
    assert response.status_code == 403

Layer 3: Data Integrity

The bug: Same PR URL could create unlimited concurrent tasks. No uniqueness constraint, no deduplication.

Add unique constraints

-- Prevent duplicate tasks for same PR
CREATE UNIQUE INDEX idx_tasks_pr_unique 
ON tasks(pr_url, evaluation_id) 
WHERE status IN ('pending', 'running');

-- Or use a partial unique index for "one active task per PR"

Idempotent creation with upsert

async def create_or_get_task(pr_url: str, eval_id: str, user_id: str):
    task_id = hashlib.sha256(f"{pr_url}:{eval_id}".encode()).hexdigest()[:32]
    result = await db.execute(text("""
        INSERT INTO tasks (id, pr_url, evaluation_id, user_id, status)
        VALUES (:id, :pr_url, :eval_id, :user_id, 'pending')
        ON CONFLICT (id) DO NOTHING
        RETURNING id
    """), {"id": task_id, "pr_url": pr_url, "eval_id": eval_id, "user_id": user_id})
    row = result.fetchone()
    if row:
        return row[0]  # New task
    # Duplicate - return existing
    return task_id

Audit: find endpoints that create without idempotency

# Look for INSERT without ON CONFLICT in handlers that accept user input
rg "INSERT INTO" --type py -B 5

Layer 4: Error Handling

The bug: Analytics service loaded unbounded result sets when the database returned an error. Fallback path had no limit.

Bounded fallbacks

# BAD: unbounded fallback
try:
    data = await analytics.query(...)
except AnalyticsError:
    data = await db.query("SELECT * FROM events")  # OOM!

# GOOD: bounded fallback
except AnalyticsError:
    data = await db.query(
        "SELECT * FROM events ORDER BY created_at DESC LIMIT 1000"
    )

Timeout on all external calls

import httpx

async def call_external_api(url: str) -> dict:
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.get(url)
        return response.json()

Audit: grep for bare except or missing limits

rg "except:" --type py   # Bare except swallows everything
rg "SELECT \*" --type py  # Unbounded queries

Layer 5: Security

The bug: file_path parameter allowed ../../etc/passwd style path traversal. Webhook secrets not enforced in dev.

Path traversal prevention

import os

def safe_path(base_dir: str, user_path: str) -> Path:
    resolved = (Path(base_dir) / user_path).resolve()
    if not str(resolved).startswith(str(Path(base_dir).resolve())):
        raise ValueError("Path traversal attempt")
    return resolved

SSRF prevention (block internal IPs)

BLOCKED_IPS = {"127.0.0.1", "169.254.169.254", "10.0.0.0"}

def validate_url(url: str) -> bool:
    parsed = urlparse(url)
    ip = socket.gethostbyname(parsed.hostname)
    if ip in BLOCKED_IPS or ip.startswith("10."):
        return False
    return True

Webhook secret enforcement (even in dev)

def verify_webhook(request: Request, body: bytes) -> bool:
    secret = os.getenv("WEBHOOK_SECRET")
    if not secret:
        raise SystemExit("WEBHOOK_SECRET required in all environments")
    signature = request.headers.get("X-Webhook-Signature", "")
    expected = hmac.new(secret.encode(), body, "sha256").hexdigest()
    return hmac.compare_digest(signature, expected)

Audit: deprecated datetime.utcnow()

# BAD: timezone bugs
from datetime import datetime
now = datetime.utcnow()  # Deprecated, no timezone info

# GOOD
from datetime import datetime, timezone
now = datetime.now(timezone.utc)

rg "datetime.utcnow" --type py  # Every match is a potential bug

Putting It Together: Audit Checklist

Run this checklist on every service before production:

## Layer 1 — API Surface
- [ ] All list endpoints have limit/offset with bounds (1-100)
- [ ] All IDs in path validated (UUID format, exists)
- [ ] Request body size limits set

## Layer 2 — Auth & Ownership
- [ ] Every {id} endpoint checks resource.owner_id == user.id
- [ ] Test: user A cannot access user B's resources

## Layer 3 — Data Integrity
- [ ] Creation endpoints idempotent (deterministic ID + upsert)
- [ ] Unique constraints on natural keys (pr_url, etc.)
- [ ] No duplicate creation under concurrent requests

## Layer 4 — Error Handling
- [ ] All external calls have timeouts
- [ ] Fallback paths have bounds (LIMIT, max retries)
- [ ] No bare except:

## Layer 5 — Security
- [ ] Path parameters validated for traversal
- [ ] URLs validated for SSRF (block metadata IPs)
- [ ] Webhook/docs secrets enforced in dev
- [ ] datetime.utcnow() replaced with timezone-aware

Quick Wins We Implemented

Of the 33 issues we found, 11 were closed within 3 days. The fastest fixes:

Pydantic validators — Add ge=1, le=100 to limit params. 5 minutes.
Ownership decorator — @require_ownership("task_id") on 12 endpoints. 2 hours.
Path safety — Wrapper for all file operations. 1 hour.
Webhook secret — Fail startup if unset. 15 minutes.

The methodology forces you to ask the right questions. Production bugs hide in the gaps between "it works" and "it's correct."

Read the full article on my blog. I write about production patterns and security — find me at humzakt.github.io.

I Built a Starter Repo That Turns AI Coding Tools Into Senior Engineers

Humza Tareen — Wed, 04 Mar 2026 19:40:48 +0000

I Built a Starter Repo That Turns AI Coding Tools Into Senior Engineers

Every new project starts the same way: create a repo, write some code, realize you have no CI, no linting, no test enforcement, no issue templates. Three weeks in, you're debugging a production incident with no runbook, your AI coding assistant is generating code that doesn't follow your conventions, and half the team is committing directly to main.

I've set up development practices across multiple production projects — from a multi-cluster AI evaluation platform on GCP to Python evaluation pipelines. The same patterns kept working: Ruff for linting with auto-fix in CI, pre-commit hooks that catch problems before they reach the remote, test coverage enforcement that posts PR comments telling you exactly which test files to create, and incident runbooks that turn 2-hour debugging sessions into 15-minute triage flows.

So I extracted all of it into a single, clone-and-go starter repo. But I added something the original projects didn't have: instruction files for every major AI coding tool — Cursor, Claude Code, Codex, and GitHub Copilot — configured to follow the same practices the CI enforces.

The repo: github.com/humzakt/dev-starter-kit

AI Tools Without Context Are Junior Developers

AI coding tools are powerful, but out of the box they don't know your project's conventions. They'll use single quotes when your linter expects double quotes. They'll skip tests. They'll commit to main. They'll generate 200-character lines when your config says 120. They'll catch generic exceptions when your style guide says to be specific.

The fix isn't to stop using AI tools — it's to give them the same onboarding document you'd give a new team member. That's what AGENTS.md, CLAUDE.md, and .cursor/rules/ files are for. They turn your AI assistant from a context-free code generator into something that understands your project's architecture, follows your conventions, and runs the right commands after every change.

What's in the Starter Kit

CI/CD Workflows

Three GitHub Actions workflows that work together:

PR Checks is the main quality gate. It runs Ruff with --fix, commits any auto-fixes back to your branch, then verifies the code is clean. After linting passes, it runs pytest. The interesting part is the test coverage enforcement: it detects which Python source files you changed, checks whether you also added or updated the corresponding test files, and if you didn't, it posts a PR comment with the exact test file names, class names, and method signatures you need to write.

# The CI posts comments like this on your PR:
#
# Test Coverage Required
#
# Source: src/services/auth.py
# Test file to create/update: tests/test_auth.py
# Test class: TestAuth
# Required test methods:
#     def test_validate_token(self):  ...
#     def test_refresh_session(self):  ...

This isn't a coverage percentage check — it's a structural check that ensures you're at least thinking about tests for the code you changed. You can bypass it with a skip-test-check label for config-only PRs.

Lint & Format runs on both pushes and PRs, but only checks files that actually changed. It validates YAML files too (excluding workflow files, which have their own validation).

Merge Readiness is the final gate. It runs strict lint (no auto-fix), Python syntax compilation, YAML validation, and tests. All checks must pass before the merge button lights up. Skipped checks (because no relevant files changed) count as passing.

Pre-commit Hooks

The .pre-commit-config.yaml catches problems before they even reach CI:

Ruff lint and format on every commit
Syntax checks — Python AST verification, YAML, JSON, TOML validation
Hygiene — trailing whitespace, line endings, end-of-file fixers
Security — private key detection, large file prevention
Branch protection — blocks direct commits to main

Issue Templates

Three structured templates that guide reporters through providing the right information:

Bug Report — component dropdown, priority, reproduction steps, error output, environment info
Incident Report — SEV-1 through SEV-4 severity, affected components checklist, triage checklist, timeline table, root cause, action items
Infrastructure Issue — CI/CD failures, dependency issues, environment config, investigation checklist

Incident Runbook

The docs/INCIDENT_RUNBOOK.md is a structured playbook for when things go wrong:

Severity classification table with response time targets
3-phase triage checklist (Identify Scope, Gather Evidence, Communicate)
Component-specific diagnostic commands you can copy-paste
CI/CD troubleshooting section
4 common failure scenarios with diagnosis and fixes
Post-incident review template

The AI Tool Configuration Layer

This is the part that makes the starter kit different from other project templates. Every major AI coding tool reads a different file format for project instructions. The starter kit includes all of them:

File	Tool	Purpose
`AGENTS.md`	Codex, Cursor, Claude Code, Windsurf	Universal agent instructions
`CLAUDE.md`	Claude Code	Session-persistent instructions
`.cursor/rules/*.mdc`	Cursor IDE	Modular, glob-scoped rules
`.claude/rules/*.md`	Claude Code	Scoped rules with patterns
`.github/copilot-instructions.md`	GitHub Copilot	Code generation guidelines

AGENTS.md: The Universal Standard

AGENTS.md is the broadest-reaching file. It's read by Codex, Cursor, Claude Code, and other AI agents. According to recent evaluations, AGENTS.md achieves a 100% pass rate for providing AI agents with project context. It includes build commands, code style conventions, testing rules, the Git workflow, project architecture, and the development loop AI agents should follow.

CLAUDE.md: Session-Persistent Instructions

Claude Code reads CLAUDE.md at the start of every session. It's kept intentionally concise — under 100 lines — because Claude Code already has a large system prompt, and every line competes for attention. It focuses on the commands to run and the rules to follow.

Cursor Rules: Modular and Scoped

Cursor uses .cursor/rules/*.mdc files with YAML frontmatter that controls when each rule applies:

general.mdc (alwaysApply: true) — project conventions that apply everywhere
python.mdc (glob: **/*.py) — Python-specific rules
testing.mdc (glob: tests/**/*.py) — test structure and patterns

This modular approach means the AI only gets the relevant rules for the file it's editing.

What the Instructions Actually Teach

All instruction files converge on the same core behaviors:

Read before editing — understand the existing code and its tests before making changes
Run lint after every edit — ruff check --fix . && ruff format .
Run tests after every edit — pytest tests/ -v
Fix failures before moving on — don't leave broken tests or lint errors
Write tests for new code — the CI will enforce this anyway
Use conventional commits — feat:, fix:, chore:, etc.
Never commit secrets — use environment variables

The result: your AI coding tool follows the same development loop a senior engineer would. It doesn't just generate code — it generates code that passes your CI.

How to Use It

Clone, delete the git history, and start fresh:

git clone https://github.com/humzakt/dev-starter-kit.git my-project
cd my-project
rm -rf .git
git init && git checkout -b main

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
pre-commit install

Customize by updating pyproject.toml (project name, lint rules), issue templates (component dropdowns), and AI tool files (architecture section, project-specific commands).

Design Decisions

Why multiple AI tool files instead of just AGENTS.md? AGENTS.md is the most broadly compatible, but each tool has its own file format with unique capabilities. Cursor's .mdc files support glob-based scoping. Claude Code's hierarchical system supports user-level, project-level, and file-level overrides. By including all formats, the starter kit works regardless of which tool your team uses.

Why structural test coverage instead of coverage percentage? Coverage percentage creates perverse incentives. The structural check is simpler: if you changed source files, did you also change test files? It doesn't check that the tests are good — that's what code review is for. It checks that you at least thought about testing.

Why auto-fix in CI? The alternative is rejecting PRs for formatting issues, which wastes everyone's time. CI fixes formatting and commits it back. Squash merging eliminates the extra commits.

Why pre-commit AND CI? Pre-commit catches issues locally. CI catches issues when pre-commit is bypassed. Belt and suspenders.

What I Learned Building This Across Multiple Projects

After implementing these practices in a multi-service GCP platform and a Python evaluation pipeline:

Auto-fix eliminated 90% of "fix formatting" follow-up commits
PR comment guidance is more effective than just a red X on a check
Incident runbooks pay for themselves the first time someone uses the triage checklist
AI tool instructions compound — every session where the AI follows your conventions saves correction time

The repo is MIT licensed. Clone it, customize it, ship it.

github.com/humzakt/dev-starter-kit

Humza Tareen builds production AI systems at scale. More articles at humzakt.github.io.

Zero-Downtime Embedding Migration: Switching from text-embedding-004 to text-embedding-3-large in Production

Humza Tareen — Fri, 20 Feb 2026 11:30:00 +0000

Our embedding model got deprecated overnight. Every RAG query started returning 404s. Here's the exact playbook we used to migrate to a new model in 48 hours with zero downtime.

The Situation

Service: RAG retrieval service using pgvector on PostgreSQL
Old model: text-embedding-004 (deprecated)
New model: text-embedding-3-large (768 dimensions)
Data volume: Thousands of embedded documents
Constraint: Zero downtime, zero data loss, production traffic must keep flowing

Step 1: Make the Model Configurable

Before anything else, stop hardcoding:

# Before (hardcoded in 6 places)
response = openai.embeddings.create(
    model="text-embedding-004",
    input=text,
)

# After (configured once)
EMBED_MODEL = os.getenv("EMBED_MODEL", "text-embedding-3-large")
EMBED_DIMENSIONS = int(os.getenv("EMBED_DIMENSIONS", "768"))

response = openai.embeddings.create(
    model=EMBED_MODEL,
    input=text,
    dimensions=EMBED_DIMENSIONS,
)

Two environment variables. This is what makes the difference between a 2-day migration and a 2-week one.

Step 2: Add New Columns (Don't Replace)

-- Migration: add new embedding column alongside the old one
ALTER TABLE documents 
ADD COLUMN embedding_v2 vector(768);

CREATE INDEX CONCURRENTLY idx_documents_embedding_v2 
ON documents USING ivfflat (embedding_v2 vector_cosine_ops)
WITH (lists = 100);

Using CONCURRENTLY means the index builds without locking the table. Production reads continue uninterrupted.

Step 3: Batch Re-embedding Script

import asyncio
from tqdm import tqdm

async def re_embed_batch(session, documents, batch_size=50):
    """Re-embed documents in batches with progress tracking."""
    for i in tqdm(range(0, len(documents), batch_size)):
        batch = documents[i:i + batch_size]
        texts = [doc.content for doc in batch]

        # Batch embedding call
        response = await openai.embeddings.create(
            model=EMBED_MODEL,
            input=texts,
            dimensions=EMBED_DIMENSIONS,
        )

        for doc, embedding in zip(batch, response.data):
            doc.embedding_v2 = embedding.embedding

        await session.commit()

        # Rate limiting
        await asyncio.sleep(0.5)

Key features:

Batch processing (don't embed one doc at a time)
Progress bar (you need to know how long this takes)
Rate limiting (embedding APIs have rate limits)
Commits per batch (don't hold a transaction for 10K docs)

Step 4: Dry-Run Validation

Before switching production traffic:

async def validate_migration(session, sample_size=100):
    """Compare search results between old and new embeddings."""
    test_queries = get_random_queries(sample_size)

    for query in test_queries:
        old_results = await search(session, query, column="embedding")
        new_results = await search(session, query, column="embedding_v2")

        # Check overlap
        old_ids = {r.id for r in old_results[:10]}
        new_ids = {r.id for r in new_results[:10]}
        overlap = len(old_ids & new_ids) / len(old_ids)

        if overlap < 0.6:  # Less than 60% overlap is concerning
            logger.warning(f"Low overlap for query: {query[:50]}... ({overlap:.0%})")

    logger.info(f"Validation complete. Average overlap: {avg_overlap:.0%}")

Our average overlap was 82% -- different models produce different embeddings, but the top results were comparable. Good enough.

Step 5: Feature Flag Switch

USE_V2_EMBEDDINGS = os.getenv("USE_V2_EMBEDDINGS", "false") == "true"

async def search(session, query: str, top_k: int = 10):
    column = "embedding_v2" if USE_V2_EMBEDDINGS else "embedding"
    query_embedding = await embed(query)

    results = await session.execute(text(f"""
        SELECT id, content, 1 - ({column} <=> :query_vec) as similarity
        FROM documents
        ORDER BY {column} <=> :query_vec
        LIMIT :top_k
    """), {"query_vec": str(query_embedding), "top_k": top_k})

    return results.fetchall()

Deploy with USE_V2_EMBEDDINGS=false. Verify everything works. Flip to true. If anything breaks, flip back instantly.

Step 6: Cleanup

After running with v2 for a week with no issues:

ALTER TABLE documents DROP COLUMN embedding;
ALTER TABLE documents RENAME COLUMN embedding_v2 TO embedding;
DROP INDEX idx_documents_embedding;
ALTER INDEX idx_documents_embedding_v2 RENAME TO idx_documents_embedding;

Lessons Learned

Always abstract the embedding provider. Two env vars saved us from a multi-file refactor.
Add model version tracking to stored vectors. We didn't. We should have.
Build migration tooling before you need it. The batch script and validation tool are reusable.
Side-by-side columns > in-place replacement. The rollback story is instant.
Dry-run everything. Our validation caught 3 queries with low overlap that needed investigation.

Total impact: 48 hours, zero downtime, zero data loss.

Read the full migration story on my blog. Part of my "Production GCP Patterns" series — find me at humzakt.github.io.

How to Build Idempotent Cloud Tasks Handlers in Python (The Pattern That Eliminated Our Duplicate Record Bugs)

Humza Tareen — Thu, 19 Feb 2026 13:30:00 +0000

Google Cloud Tasks guarantees at-least-once delivery. That means your handler WILL be called multiple times for the same task. If you're not handling this, you have bugs. You might not know it yet.

This guide shows the exact pattern we use in production to handle 1000s of Cloud Tasks per day with zero duplicates.

The Problem

Here's what happens without idempotency:

# ❌ BAD: This creates duplicates on retry
@app.post("/tasks/score")
async def handle_score_task(request: ScoreRequest):
    result = ScoringResult(
        id=uuid4(),  # New ID every time!
        evaluation_id=request.evaluation_id,
        score=compute_score(request),
    )
    db.add(result)
    db.commit()
    return {"status": "ok"}

Cloud Tasks retries if:

Your handler returns 5xx
Your handler takes longer than the timeout
The network hiccups

Every retry creates a new row with a new UUID. Your data is silently corrupted.

The Fix: 3-Step Idempotency Pattern

Step 1: Deterministic IDs

Generate IDs from the task payload, not randomly:

import hashlib

def compute_task_id(evaluation_id: str, model: str, turn: int) -> str:
    """Same inputs → same ID. Always."""
    payload = f"{evaluation_id}:{model}:{turn}"
    return hashlib.sha256(payload.encode()).hexdigest()[:32]

Step 2: Guarded Upserts

Use ON CONFLICT DO NOTHING:

from sqlalchemy import text

@app.post("/tasks/score")
async def handle_score_task(request: ScoreRequest):
    task_id = compute_task_id(
        request.evaluation_id,
        request.model,
        request.turn,
    )

    result = db.execute(text("""
        INSERT INTO scoring_results (id, evaluation_id, score, model, turn)
        VALUES (:id, :eval_id, :score, :model, :turn)
        ON CONFLICT (id) DO NOTHING
        RETURNING id
    """), {
        "id": task_id,
        "eval_id": request.evaluation_id,
        "score": compute_score(request),
        "model": request.model,
        "turn": request.turn,
    })

    was_inserted = result.fetchone() is not None

    if not was_inserted:
        logger.info(f"Duplicate delivery detected: {task_id}")

    db.commit()
    return {"status": "ok", "duplicate": not was_inserted}

Step 3: Return 200 Even on Duplicates

This is critical. If your handler detects a duplicate and returns 409 (Conflict), Cloud Tasks will retry it. Forever.

# ✅ Always return 200 for idempotent operations
return {"status": "ok", "duplicate": not was_inserted}

Handling Multi-Step Tasks

What if your task does 3 things?

@app.post("/tasks/evaluate")
async def handle_evaluation(request: EvalRequest):
    task_id = compute_task_id(request.evaluation_id, request.model)

    # Step 1: Generate chain-of-thought
    cot_id = f"{task_id}:cot"
    db.execute(text("""
        INSERT INTO cot_results (id, task_id, content)
        VALUES (:id, :task_id, :content)
        ON CONFLICT (id) DO NOTHING
    """), {"id": cot_id, "task_id": task_id, "content": generate_cot(request)})

    # Step 2: Generate critique
    critique_id = f"{task_id}:critique"
    db.execute(text("""
        INSERT INTO critiques (id, task_id, content)
        VALUES (:id, :task_id, :content)
        ON CONFLICT (id) DO NOTHING
    """), {"id": critique_id, "task_id": task_id, "content": generate_critique(request)})

    # Step 3: Store final score
    db.execute(text("""
        INSERT INTO scores (id, task_id, score)
        VALUES (:id, :task_id, :score)
        ON CONFLICT (id) DO NOTHING
    """), {"id": f"{task_id}:score", "task_id": task_id, "score": compute_score(request)})

    db.commit()
    return {"status": "ok"}

Each step has its own deterministic ID. If the handler crashes after step 2, the retry will skip steps 1 and 2 (already exist) and only execute step 3.

Cloud Run Configuration

Don't forget the infrastructure side:

# service.yaml for Cloud Run
spec:
  template:
    metadata:
      annotations:
        run.googleapis.com/execution-environment: gen2
    spec:
      containerConcurrency: 10  # Don't overwhelm the DB
      timeoutSeconds: 3600      # Match your longest task
      containers:
        - resources:
            limits:
              memory: 2Gi
              cpu: "2"

We reduced concurrency from 80 to 10 per instance. Why? Each handler holds a database connection. 80 concurrent handlers × 10 instances = 800 connections. Our Cloud SQL instance can't handle that. 10 × 10 = 100 connections. Much better.

Adding Correlation IDs

For debugging duplicate deliveries:

from uuid import uuid4
import structlog

@app.middleware("http")
async def add_correlation_id(request, call_next):
    # Cloud Tasks sends X-CloudTasks-TaskName header
    task_name = request.headers.get("X-CloudTasks-TaskName", str(uuid4()))
    structlog.contextvars.bind_contextvars(correlation_id=task_name)
    response = await call_next(request)
    return response

Now every log line from a Cloud Tasks handler includes the task name. Duplicate deliveries show up as multiple log entries with the same correlation ID.

Testing Idempotency

def test_handler_is_idempotent():
    """Same request twice → same result, no duplicates."""
    request = ScoreRequest(
        evaluation_id="eval-123",
        model="gpt-4o",
        turn=1,
    )

    # First call
    response1 = client.post("/tasks/score", json=request.dict())
    assert response1.status_code == 200
    assert response1.json()["duplicate"] == False

    # Second call (simulating retry)
    response2 = client.post("/tasks/score", json=request.dict())
    assert response2.status_code == 200
    assert response2.json()["duplicate"] == True

    # Verify only one row exists
    count = db.execute(text("SELECT COUNT(*) FROM scoring_results")).scalar()
    assert count == 1

TL;DR

Problem	Solution
Random UUIDs	Deterministic IDs from payload
INSERT	INSERT ... ON CONFLICT DO NOTHING
Return 409 on duplicate	Return 200 always
No logging	Correlation IDs from Cloud Tasks headers
High concurrency	Cap containerConcurrency (10, not 80)

This pattern handles our entire evaluation pipeline -- thousands of tasks per day, zero duplicates, zero data corruption.

Read the full article on my blog. I write about production GCP patterns — find me at humzakt.github.io.

AlloyDB + SQLAlchemy: Connection Pooling Strategies That Actually Work in Production

Humza Tareen — Tue, 17 Feb 2026 19:17:12 +0000

We had a 5% failure rate on long-running tasks. Every error looked the same:

sqlalchemy.exc.OperationalError: (psycopg2.OperationalError)
SSL connection has been closed unexpectedly

This post explains why it happens and the exact configuration that fixed it.

The Setup

Database: AlloyDB (GCP's PostgreSQL-compatible database)
ORM: SQLAlchemy 2.0 with asyncio
Deployment: Cloud Run (auto-scaling, 0-100 instances)
Workload: Long-running AI evaluation tasks (5-30 minutes per task)

Why Connections Die

AlloyDB (and Cloud SQL) terminates idle SSL connections after a timeout. The exact timeout depends on your configuration, but it's typically 10-15 minutes.

SQLAlchemy's connection pool holds connections open for reuse. Here's the problem:

1. Task A grabs Connection 1 from pool → starts 20-min evaluation
2. Connection 2, 3, 4 sit idle in the pool
3. After 10 min: AlloyDB kills Connections 2, 3, 4 (idle too long)
4. Task B grabs Connection 2 from pool → DEAD → SSL error

The pool doesn't know the connection is dead until you try to use it.

The Fix

from sqlalchemy import create_engine

engine = create_engine(
    DATABASE_URL,

    # 1. Recycle connections before AlloyDB kills them
    pool_recycle=180,  # 3 minutes (AlloyDB timeout is ~10 min)

    # 2. Test connection before handing it to your code
    pool_pre_ping=True,  # SELECT 1 before checkout

    # 3. Don't hoard connections
    pool_size=5,         # Base pool size
    max_overflow=10,     # Allow burst to 15 total

    # 4. Don't wait forever for a connection
    pool_timeout=30,     # Fail fast if pool exhausted
)

Why each setting matters:

pool_recycle=180 -- Forces SQLAlchemy to close and recreate connections every 3 minutes, well before AlloyDB's ~10 minute idle timeout. Yes, this means more connection overhead. No, it doesn't matter compared to your query time.

pool_pre_ping=True -- Before giving a connection to your code, SQLAlchemy runs SELECT 1. If it fails, the connection is discarded and a fresh one is created. This catches connections that died between recycles.

pool_size=5, max_overflow=10 -- Total max connections = 15. With Cloud Run auto-scaling to 100 instances, that's 1,500 potential connections. AlloyDB's limit depends on your tier. We use db-custom-4-16384 which handles this fine, but a db-f1-micro would not.

pool_timeout=30 -- If all 15 connections are in use, wait 30 seconds then raise an error. Better than hanging indefinitely.

Cloud Run Gotcha: Concurrency × Instances × Pool Size

This is the formula that will save you:

Max DB connections = containerConcurrency × maxInstances × pool_size

Our initial config:

containerConcurrency: 80
maxInstances: 100
pool_size: 10
Max connections: 80,000 🔥

After fixing:

containerConcurrency: 10
maxInstances: 100
pool_size: 5
Max connections: 5,000 ✅

We also capped max_overflow to prevent connection storms during traffic spikes.

For Async SQLAlchemy

If you're using AsyncSession:

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    DATABASE_URL.replace("postgresql://", "postgresql+asyncpg://"),
    pool_recycle=180,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
)

Same settings, different driver. asyncpg handles connection validation differently than psycopg2, but pool_pre_ping works with both.

Making Tasks Survive Connection Drops

Even with the pool fix, you should handle connection drops in long-running tasks:

from sqlalchemy.exc import OperationalError
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, max=10),
    retry=retry_if_exception_type(OperationalError),
)
async def save_evaluation_result(session, result):
    """Retry on transient connection errors."""
    session.add(result)
    await session.commit()

But crucially, make this idempotent (see my other post on idempotent Cloud Tasks handlers). If the first attempt committed successfully but the response got lost, the retry shouldn't create a duplicate.

Monitoring

Add these metrics to catch connection issues early:

import structlog

logger = structlog.get_logger()

@app.middleware("http")
async def log_db_pool_stats(request, call_next):
    pool = engine.pool
    logger.info(
        "db_pool_stats",
        pool_size=pool.size(),
        checked_out=pool.checkedout(),
        overflow=pool.overflow(),
        checked_in=pool.checkedin(),
    )
    response = await call_next(request)
    return response

When checked_out is consistently at pool_size + max_overflow, you're about to hit connection exhaustion.

TL;DR Config

# Copy-paste this for AlloyDB / Cloud SQL + Cloud Run
engine = create_engine(
    DATABASE_URL,
    pool_recycle=180,
    pool_pre_ping=True,
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
)

And set containerConcurrency in Cloud Run to something reasonable (10-20, not 80).

This took us from 5% random failures to 0.01%. Three lines of configuration.

Read the full AlloyDB debugging story on my blog. Part of my "Production GCP Patterns" series — find me at humzakt.github.io.

PostgreSQL Memory Allocation: A Beginner's Guide

Humza Tareen — Wed, 16 Aug 2023 07:53:45 +0000

Memory allocation plays a crucial role in the operation of any database system, and PostgreSQL is no exception. Just like how a chef needs an organized kitchen to prepare dishes efficiently, PostgreSQL needs memory organized in specific ways to process and manage data effectively. Let's dive into how PostgreSQL uses memory and why it matters.

1. How does PostgreSQL use memory?

a) Shared Buffers:

What: This is a cache of disk pages that PostgreSQL uses to reduce the number of times it must read from or write to the disk.
Why Important: Reading and writing directly to disk is slow. Shared buffers provide a space in memory for faster access, significantly speeding up operations.

b) WAL Buffers:

What: Stands for Write-Ahead Logging buffers. They temporarily hold changes to data before it gets written permanently to storage.
Why Important: This ensures data integrity. If the system crashes, PostgreSQL can use the WAL to restore the database to its last known consistent state.

c) Maintenance Work Mem:

What: Memory used for maintenance operations such as vacuuming (cleaning up and optimizing database storage) and creating indexes.
Why Important: Allocating adequate memory here ensures that maintenance operations, which are critical for long-term database health, run efficiently.

d) Work Mem:

What: Used for internal operations like sorting and hashing during query processing.
Why Important: If this is set too low, PostgreSQL might have to resort to disk-based temporary storage, slowing down query performance.

2. Why is Memory Allocation Important in Performance Tuning and Troubleshooting?

Avoiding Disk Swaps: When memory is inadequately allocated, PostgreSQL might have to swap information to and from the disk more frequently. Disk operations are much slower than memory operations. Proper memory allocation can dramatically reduce these slow disk operations.
Efficient Query Execution: Proper allocation can drastically speed up the execution of queries. For instance, if work_mem is set appropriately, sorts and hashes are done swiftly in memory.
Troubleshooting Bottlenecks: Understanding memory usage can help pinpoint performance issues. If a certain operation is slower than expected, it's possible that there's not enough memory allocated for it, causing the system to revert to slower methods of operation.
Preventing Crashes: Over-allocation can be just as problematic. If PostgreSQL consumes all available system memory, it could cause the system to become unresponsive or even crash.

Conclusion:

Memory allocation in PostgreSQL isn't just about throwing more RAM at the database. It's about understanding the distinct ways PostgreSQL uses memory and fine-tuning them for your specific use case. As you delve deeper into PostgreSQL, you'll find that tweaking these settings, along with regular monitoring, can lead to significant performance gains and a smoother database experience. Happy tuning! 🐘

PostgreSQL Background Processes: A Simple Guide

Humza Tareen — Wed, 16 Aug 2023 07:41:53 +0000

PostgreSQL, a highly advanced open-source relational database system, is built to be robust and dynamic. Part of what makes it so reliable is the set of background processes that ensure smooth operation, data integrity, and performance optimization. In this guide, we'll delve into these background processes, what they do, and why they are crucial for maintaining database performance.

1. Postmaster Process

What it does: It's the primary process that starts up and manages all other PostgreSQL processes.
Importance: It initializes shared memory, monitors server status, and ensures the database's overall health. If any of the child processes (like the backend process) dies unexpectedly, postmaster intervenes and restores stability.

2. Backend Processes

What it does: These are spawned for every new client connection. Each backend process is isolated from others, ensuring that processes don't interfere with each other.
Importance: This isolation enhances the database's reliability. If a client causes a backend process to crash, it won’t affect others.

3. Autovacuum Process

What it does: It reclaims storage occupied by "dead tuples" (rows which have been deleted or made obsolete by updates) and updates statistics used by PostgreSQL's query planner.
Importance: Ensures that the database doesn't become bloated with obsolete data. It also helps in keeping the query planner's statistics up-to-date, leading to optimized query execution.

4. Background Writer

What it does: Periodically flushes "dirty" data pages (pages that have been modified in-memory but not yet written to disk) to the database on the disk.
Importance: Reduces the work needed to be done during checkpoints, leading to smoother database operations and reducing the I/O load during checkpoint operations.

5. WAL Writer

What it does: Writes the Write-Ahead Log (WAL) to disk. PostgreSQL uses WAL to record changes made to its data for durability.
Importance: Helps in achieving data durability without having to write every database change to the main data files immediately. This allows for improved performance.

6. Archiver Process

What it does: Transfers WAL files to a long-term storage location once they are filled up.
Importance: Ensures that the WAL is stored safely, which is crucial for Point-In-Time Recovery and certain replication setups.

7. Stats Collector

What it does: Gathers statistics about the PostgreSQL server activity.
Importance: Provides insights for the query planner to optimize queries and for administrators to monitor the database.

8. Logger Process

What it does: Handles the server's log messages.
Importance: Logging is crucial for diagnostics, troubleshooting, and monitoring. Proper handling of logs ensures that administrators have the necessary information when needed.

9. Checkpointer

What it does: At specific intervals, this process ensures that all dirty pages and transaction logs are flushed to disk.
Importance: Helps maintain the durability of the database and ensures data consistency.

10. WAL Receiver & WAL Sender

What they do: These processes are responsible for replication. WAL Receiver receives WAL records from a primary server, and WAL Sender sends them from a primary to standby servers.
Importance: Ensures data consistency and availability in replication setups.

Conclusion

The myriad background processes of PostgreSQL, from the postmaster to the WAL sender, play pivotal roles in ensuring the database's reliability, performance, and durability. Understanding them provides insights into the internal workings of PostgreSQL, which can be beneficial for administrators and developers alike.

Cracking the Code: How PostgreSQL Parses SQL Queries

Humza Tareen — Fri, 04 Aug 2023 12:34:58 +0000

In the vast world of SQL, understanding how your queries are parsed and processed can be a game-changer when it comes to performance optimization. Today, we're taking a behind-the-scenes look at PostgreSQL, one of the world's most popular open-source SQL database systems. So, buckle up, and let's demystify the process of PostgreSQL's SQL query parsing.

The Journey of a SQL Query in PostgreSQL

Every SQL query you write embarks on a fascinating journey through PostgreSQL's inner workings. It's not just about executing the query; several crucial stages are involved in transforming your query into actionable instructions for the database.

Stage 1: Lexical Analysis

Imagine you're building a puzzle. The first step is to lay out all the pieces. That's what lexical analysis does with your query. It breaks down the query into "tokens" - these can be keywords, operators, or identifiers.

Performance Implications:

This stage is generally lightweight. However, larger and more complex queries might require more time for tokenization.

Stage 2: Syntactic Analysis

Now you start connecting those puzzle pieces. PostgreSQL uses a parser to arrange the tokens into a "parse tree" based on the SQL grammar rules.

Performance Implications:

Again, complex or poorly formed queries can make this stage more time-consuming.

Stage 3: Semantic Analysis

This is where PostgreSQL validates whether the pieces are fitting correctly. It verifies data types, table and column existence, and permissions.

Performance Implications:

Extensive schema checks or permissions validations can slow down the process here.

Stage 4: Optimization

This is where PostgreSQL's real magic shines. The query optimizer takes over, analyzing the parse tree and devising the most efficient way to execute your query.

Performance Implications:

Complex queries may take longer to optimize. However, this might lead to a more efficient and faster execution in the end.

Stage 5: Execution

Finally, PostgreSQL's execution engine steps in. It takes the optimized plan and performs the requested operation.

Performance Implications:

Execution speed greatly depends on the specific query, the underlying data, system resources, and more.

PostgreSQL's Query Journey: Visualized

Textual descriptions are useful, but sometimes a picture is worth a thousand words. So, here's a simplified visualization of how your SQL query travels within PostgreSQL:

  SQL Query Input
      |
      v
  Lexical Analysis
      |
      v
  Syntactic Analysis ----> Error if syntax is incorrect
      |
      v
  Semantic Analysis ----> Error if semantics are incorrect
      |
      v
  Optimization
      |
      v
  Execution

Performance Optimization: It's all in the Details

There are a few key factors you should be aware of when it comes to PostgreSQL query performance:

Query Complexity: Keep it simple. The complexity of your query affects every stage, especially the optimization process.
Indexes: They're your best friend. Proper use of indexes can massively reduce execution time.
System Resources: It's not just about the code. Available CPU, memory, and IO capacity also affect query execution.
Cache Usage: Reuse your queries. PostgreSQL can use precompiled plans for frequently used queries, saving time.
Statistics: Keep your statistics up-to-date. This helps the optimizer make better decisions, further improving performance.

Conclusion

Understanding how PostgreSQL parses and processes queries is like gaining superpowers. It empowers you to write better, more efficient queries, and equips you to troubleshoot performance bottlenecks with more precision. So, keep exploring, keep optimizing, and let PostgreSQL amaze you with its robust performance.

References:

Docs

Apache AGE: A Deep Dive into Data Import and Export Techniques

Humza Tareen — Sun, 23 Jul 2023 10:23:25 +0000

Apache AGE is a graph database that provides SQL-based graph modelling capabilities for PostgreSQL. It has opened new avenues in data management, enabling both relational and graph data processing within the same database system. Today, we'll explore data import and export techniques with Apache AGE, ensuring we're equipped to handle our graph data effectively.

Introduction to AGE

Apache AGE inherits the powerful features of PostgreSQL and integrates the Apache TinkerPop Gremlin graph model and traversal language. AGE extends PostgreSQL's SQL syntax to support graph pattern matching queries, thus enabling relational and graph data processing.

Data Import in Apache AGE

Let's begin by setting up our environment. Please ensure that you have AGE installed and ready.

AGE Graph Creation

First, we need to create a graph. We can do so using the following SQL query:

SELECT * FROM cypher('CREATE (n:Person {name: $name, age: $age}) RETURN n', 
'{ "name" : "John", "age" : 30 }') AS (n agtype);

The above query creates a node labeled Person with the properties name and age.

Importing Data
To import data, we create a copy from SQL statement in AGE. The CSV file format is commonly used, so we'll use it as our example. Let's assume we have a CSV file named persons.csv with the following data:

name,age
John,30
Jane,25
Bob,20

We can import this data into our AGE graph as follows:

COPY Person(name, age) FROM '/path/to/persons.csv' DELIMITER ',' CSV HEADER;

Data Export in Apache AGE

Exporting data from AGE is also straightforward. We can use the copy to SQL command to export the data to a CSV file.

COPY (SELECT * FROM Person) TO '/path/to/export.csv' WITH CSV HEADER;

This command will export all the data in the Person table to a CSV file.

Conclusion

Apache AGE, with its ability to perform both relational and graph data processing, provides flexibility in data management. This post provided a brief overview of how to import and export data using Apache AGE, which could be the beginning of your journey in handling complex graph data structures with the ease of SQL.

Remember to handle your data carefully and ensure that it is correctly imported and exported to maintain its integrity.

Next time, we'll delve deeper into the graph querying techniques of Apache AGE. Until then, keep exploring and learning!

References

Github
Docs