⟐ Dev Tools/2026-03-29Advanced

Gemini API Production Notes — Quiet Defenses Against 429, 500, and 503 Under Real Traffic

Operational notes from running Gemini API in production on an indie wallpaper app: exponential backoff, jitter, circuit breakers, token buckets, and model cascades — with the pitfalls I actually hit and measured retry success rates.

gemini-api²⁸³ error-handling⁸ rate-limiting⁴ production¹⁴⁰ circuit-breaker² retry-pattern

✦ Premium Article

Setup and context — Why Production AI Apps Fail Silently

The first sign that something was wrong with my wallpaper app didn't come from a crash report. It came from a review: "The descriptions stopped showing up yesterday." The server was healthy. Not a single exception in the logs. When I ran the call by hand, the Gemini API was returning HTTP 200 every time — with an empty response.text.

That is how AI-backed apps break. The process stays up. Nothing pages you. A feature just quietly falls out, and you find out days later from a one-star review. A 200 with an empty body, a stream that dies halfway through, a response blocked by a safety filter — none of it fails the way a traditional web API fails, where an exception flies and something visibly stops.

What follows are the operational notes I keep for naming each of those quiet failures and catching them before they reach a user: error classification, exponential backoff, circuit breakers, token buckets, and model cascades — each one paired with a pitfall I actually hit on an indie app.

The Complete Error Code Taxonomy and Retry Decision Matrix

HTTP errors from the Gemini API fall into two categories: retryable and non-retryable. Getting this classification wrong means either wasting resources retrying permanent errors or unnecessarily shutting down your service for transient failures.

Retryable Errors (Transient)

429 Too Many Requests — Rate limit exceeded. The most common error, triggered when any of the four rate limit dimensions (RPM, TPM, RPD, or IPM) reaches its ceiling
500 Internal Server Error — Temporary Google-side failure. Occurs due to model inference timeouts or infrastructure issues
503 Service Unavailable — Temporary service suspension during maintenance or high load
504 Gateway Timeout — Request processing exceeded the time limit

Non-Retryable Errors (Permanent)

400 Bad Request — Malformed request (invalid JSON, unsupported parameters, etc.)
401 Unauthorized — Invalid or expired API key
403 Forbidden — API key lacks access to the target model, or region restriction
404 Not Found — Specified model name doesn't exist. Not just typos: you get the same 404 when the model you were using is retired out from under you. Covered below in "Model retirements arrive as a 404"

Implementing the Classification Logic

# Python: Retry decision helper
RETRYABLE_STATUS_CODES = {429, 500, 503, 504}
 
def is_retryable(status_code: int) -> bool:
    """Determine whether an error is retryable"""
    return status_code in RETRYABLE_STATUS_CODES
 
def classify_error(status_code: int, error_message: str) -> dict:
    """Classify an error and return the recommended action"""
    if status_code == 429:
        return {
            "type": "rate_limit",
            "retryable": True,
            "action": "exponential_backoff",
            "message": "Rate limit reached. Retrying with backoff"
        }
    elif status_code in (500, 503, 504):
        return {
            "type": "server_error",
            "retryable": True,
            "action": "exponential_backoff",
            "message": "Transient server error. Auto-retrying"
        }
    elif status_code == 400:
        return {
            "type": "client_error",
            "retryable": False,
            "action": "fix_request",
            "message": f"Request format error: {error_message}"
        }
    else:
        return {
            "type": "fatal_error",
            "retryable": False,
            "action": "alert_developer",
            "message": f"Immediate attention required: {status_code} {error_message}"
        }

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦A retry decision matrix that classifies 429/500/503 with measured frequencies from a live wallpaper app

✦Working Python and TypeScript implementations of exponential backoff with jitter, circuit breakers, and token buckets

✦Pro→Flash→Flash-Lite cascade design with the ordering mistake I made on an indie app and the retention numbers that came back

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Implementing Exponential Backoff with Jitter the Right Way

The foundation of any retry strategy is exponential backoff, but naive implementations create the "thundering herd" problem — multiple clients retry simultaneously and trigger the rate limit all over again. Adding jitter (randomized delay) solves this by distributing retry attempts across time.

Full Jitter vs Equal Jitter vs Decorrelated Jitter

Full Jitter is the recommended approach for most production systems. It's endorsed by AWS best practices and provides maximum variance in retry intervals, minimizing collision between multiple clients.

import random
import time
import asyncio
from google import genai
 
# Full Jitter implementation
class GeminiRetryClient:
    def __init__(
        self,
        api_key: str,
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0
    ):
        self.client = genai.Client(api_key=api_key)
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
 
    def _calculate_delay_full_jitter(self, attempt: int) -> float:
        """Full Jitter: sleep = random(0, min(cap, base * 2^attempt))"""
        exp_delay = self.base_delay * (2 ** attempt)
        capped_delay = min(exp_delay, self.max_delay)
        return random.uniform(0, capped_delay)
 
    def _calculate_delay_equal_jitter(self, attempt: int) -> float:
        """Equal Jitter: sleep = base_delay/2 + random(0, base_delay/2)"""
        exp_delay = self.base_delay * (2 ** attempt)
        capped_delay = min(exp_delay, self.max_delay)
        half = capped_delay / 2
        return half + random.uniform(0, half)
 
    def _calculate_delay_decorrelated(self, prev_delay: float) -> float:
        """Decorrelated Jitter: sleep = min(cap, random(base, prev * 3))"""
        return min(self.max_delay, random.uniform(self.base_delay, prev_delay * 3))
 
    async def generate_with_retry(self, model: str, prompt: str) -> str:
        """Generate content with automatic retry"""
        last_exception = None
 
        for attempt in range(self.max_retries + 1):
            try:
                response = self.client.models.generate_content(
                    model=model,
                    contents=prompt
                )
                return response.text
 
            except Exception as e:
                last_exception = e
                status_code = getattr(e, 'status_code', 500)
 
                if not is_retryable(status_code):
                    raise  # Non-retryable errors fail immediately
 
                if attempt == self.max_retries:
                    break  # Max retries reached
 
                delay = self._calculate_delay_full_jitter(attempt)
                print(f"[Retry {attempt + 1}/{self.max_retries}] "
                      f"{status_code} error — retrying in {delay:.2f}s")
                await asyncio.sleep(delay)
 
        raise last_exception  # All retries exhausted
 
# Usage example
# client = GeminiRetryClient(api_key="YOUR_API_KEY")
# result = await client.generate_with_retry(
#     model="gemini-3.7-flash",
#     prompt="Explain the four seasons in Japan"
# )
# print(result)

Leveraging the Retry-After Header for 429 Errors

When Gemini API returns a 429, the response may include a Retry-After header. When present, this server-specified wait time should take priority over your exponential backoff calculation.

def get_retry_delay(error, attempt: int, base_delay: float = 1.0) -> float:
    """Prefer Retry-After header; fall back to Full Jitter"""
    retry_after = getattr(error, 'retry_after', None)
    if retry_after is not None:
        # Server-specified wait time + small jitter
        return float(retry_after) + random.uniform(0, 1.0)
    # Fallback: Full Jitter
    return random.uniform(0, min(60.0, base_delay * (2 ** attempt)))

Implementing the Circuit Breaker Pattern

When errors cascade, continuing to retry only makes things worse. The circuit breaker pattern prevents this cascade failure by temporarily halting requests when the failure rate exceeds a threshold.

Three-State Transition Model

CLOSED (Normal): Requests pass through normally. Failure count is monitored
OPEN (Tripped): Requests are immediately rejected. Waiting for cooldown period
HALF-OPEN (Testing): A limited number of requests are allowed through to test recovery

import time
from enum import Enum
from threading import Lock
 
class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
 
class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
 
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time = 0.0
        self._half_open_calls = 0
        self._lock = Lock()
 
    @property
    def state(self) -> CircuitState:
        with self._lock:
            if self._state == CircuitState.OPEN:
                # Transition to HALF_OPEN after cooldown period
                if time.time() - self._last_failure_time >= self.recovery_timeout:
                    self._state = CircuitState.HALF_OPEN
                    self._half_open_calls = 0
            return self._state
 
    def allow_request(self) -> bool:
        """Determine whether a request should be allowed"""
        current_state = self.state
        if current_state == CircuitState.CLOSED:
            return True
        elif current_state == CircuitState.HALF_OPEN:
            with self._lock:
                if self._half_open_calls < self.half_open_max_calls:
                    self._half_open_calls += 1
                    return True
            return False
        return False  # OPEN state
 
    def record_success(self):
        """Record a successful request"""
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.half_open_max_calls:
                    # Sufficient successes — return to CLOSED
                    self._state = CircuitState.CLOSED
                    self._failure_count = 0
                    self._success_count = 0
            elif self._state == CircuitState.CLOSED:
                self._failure_count = 0
 
    def record_failure(self):
        """Record a failed request"""
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.time()
            if self._state == CircuitState.HALF_OPEN:
                # Failure in HALF_OPEN — revert to OPEN
                self._state = CircuitState.OPEN
                self._success_count = 0
            elif self._failure_count >= self.failure_threshold:
                self._state = CircuitState.OPEN
 
# Usage example
# breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)
# if breaker.allow_request():
#     try:
#         result = call_gemini_api(prompt)
#         breaker.record_success()
#     except RetryableError:
#         breaker.record_failure()
# else:
#     # Fallback: cached response or default message
#     result = get_fallback_response()

Proactive Rate Limiting with Token Buckets

Rather than reacting to errors after they occur, the ideal approach is to prevent rate limit violations proactively. The token bucket algorithm is the most widely adopted technique for this type of preemptive throttling.

Understanding Gemini API's Four Rate Limit Dimensions

Gemini API enforces rate limits across four dimensions simultaneously.

RPM (Requests Per Minute): Request count per minute. Free Tier allows 5–15 RPM; Tier 1 provides 150–300 RPM
TPM (Tokens Per Minute): Token consumption per minute. Calculated as the sum of input and output tokens
RPD (Requests Per Day): Request count per day. Free Tier allows 100–1,000 RPD
IPM (Images Per Minute): Image generation model exclusive. Image generation count per minute

A critical detail: these limits are enforced per project, not per API key. Using multiple API keys within the same project won't increase your rate limits.

import time
import asyncio
 
class TokenBucket:
    """Rate limiting via the token bucket algorithm"""
 
    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: Maximum bucket capacity (e.g., RPM=15 means 15)
            refill_rate: Tokens added per second (e.g., RPM=15 means 15/60=0.25)
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = float(capacity)
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()
 
    async def acquire(self, tokens: int = 1, timeout: float = 30.0) -> bool:
        """Acquire tokens, waiting until available"""
        deadline = time.monotonic() + timeout
 
        while True:
            async with self._lock:
                self._refill()
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
 
            # Wait for token replenishment
            wait_time = (tokens - self.tokens) / self.refill_rate
            if time.monotonic() + wait_time > deadline:
                return False  # Timeout
            await asyncio.sleep(min(wait_time, 0.1))
 
    def _refill(self):
        """Replenish tokens based on elapsed time"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
 
class GeminiRateLimiter:
    """Manages Gemini API's four-dimensional rate limits"""
 
    def __init__(self, rpm: int = 15, tpm: int = 250000, rpd: int = 1000):
        self.rpm_bucket = TokenBucket(capacity=rpm, refill_rate=rpm / 60.0)
        self.tpm_bucket = TokenBucket(capacity=tpm, refill_rate=tpm / 60.0)
        self.rpd_bucket = TokenBucket(capacity=rpd, refill_rate=rpd / 86400.0)
 
    async def acquire(self, estimated_tokens: int = 1000) -> bool:
        """Check rate limits before sending a request"""
        # RPM check
        rpm_ok = await self.rpm_bucket.acquire(1)
        if not rpm_ok:
            return False
 
        # TPM check (estimated input token count)
        tpm_ok = await self.tpm_bucket.acquire(estimated_tokens)
        if not tpm_ok:
            return False
 
        # RPD check
        rpd_ok = await self.rpd_bucket.acquire(1)
        if not rpd_ok:
            return False
 
        return True
 
# Usage example
# limiter = GeminiRateLimiter(rpm=15, tpm=250000, rpd=1000)
#
# async def safe_generate(prompt: str):
#     if await limiter.acquire(estimated_tokens=len(prompt) // 4):
#         return await client.generate_with_retry("gemini-3.7-flash", prompt)
#     else:
#         raise Exception("Rate limit acquisition timed out")

Model Cascade — Designing Fallback Strategies

In production, relying on a single model is a recipe for downtime. A model cascade progressively falls back to lighter models when the primary model is unavailable due to rate limits or errors, maintaining service availability throughout.

from dataclasses import dataclass
 
@dataclass
class ModelConfig:
    name: str
    priority: int  # Lower = higher priority
    max_rpm: int
    cost_per_1m_input: float
    cost_per_1m_output: float
 
# Model cascade definition (priority order).
# Prices are Standard tier, per 1M tokens, as listed on the pricing page in Aug 2026.
# Don't treat these as permanent - reconcile them against the pricing page monthly.
MODEL_CASCADE = [
    ModelConfig(
        name="gemini-3.1-pro",
        priority=1,
        max_rpm=150,
        cost_per_1m_input=2.00,   # prompts <= 200k tokens; 4.00 above that
        cost_per_1m_output=12.00  # likewise, 18.00 above 200k
    ),
    ModelConfig(
        name="gemini-3.7-flash",
        priority=2,
        max_rpm=300,
        cost_per_1m_input=0.75,   # through 2026-12-31; 1.50 from 2027-01-01
        cost_per_1m_output=3.75   # through 2026-12-31; 7.50 from 2027-01-01
    ),
    ModelConfig(
        name="gemini-3.1-flash-lite",
        priority=3,
        max_rpm=300,
        cost_per_1m_input=0.25,   # 0.50 for audio input
        cost_per_1m_output=1.50   # shutdown announced 2027-05-07 -> gemini-3.5-flash-lite
    ),
]
 
class ModelCascadeClient:
    def __init__(self, api_key: str, cascade: list[ModelConfig]):
        self.client = genai.Client(api_key=api_key)
        self.cascade = sorted(cascade, key=lambda m: m.priority)
        self.breakers = {
            model.name: CircuitBreaker(failure_threshold=3, recovery_timeout=60.0)
            for model in cascade
        }
 
    async def generate(self, prompt: str) -> dict:
        """Generate content using the cascade approach"""
        errors = []
 
        for model_config in self.cascade:
            breaker = self.breakers[model_config.name]
 
            if not breaker.allow_request():
                errors.append(f"{model_config.name}: circuit breaker OPEN")
                continue
 
            try:
                response = self.client.models.generate_content(
                    model=model_config.name,
                    contents=prompt
                )
                breaker.record_success()
                return {
                    "text": response.text,
                    "model_used": model_config.name,
                    "fallback": model_config.priority > 1
                }
 
            except Exception as e:
                breaker.record_failure()
                errors.append(f"{model_config.name}: {e}")
                continue
 
        # All models failed — return cached or default response
        return {
            "text": "We're experiencing high demand. Please try again shortly.",
            "model_used": "fallback_cache",
            "fallback": True,
            "errors": errors
        }
 
# Usage example
# cascade_client = ModelCascadeClient(
#     api_key="YOUR_API_KEY",
#     cascade=MODEL_CASCADE
# )
# result = await cascade_client.generate("Explain async programming in Python")
# print(f"Model used: {result['model_used']}")
# print(f"Fallback: {result['fallback']}")
# print(result['text'])

Model Retirements Arrive as a 404 — Designing a Cascade That Outlives Its Models

I left a cascade untouched for about six months and got a cold feeling when I finally looked at it again. All three tiers were the same model generation. Stack Pro, Flash, and Flash-Lite and it feels like redundancy — but if that generation retires together, all three vanish at once. Every fallback was standing on the same cliff.

Worse, that failure shows up in the decision matrix above as a non-retryable 404. Backoff won't help. The circuit breaker won't help. Every model lands in errors, and the user keeps seeing "we're experiencing heavy traffic right now" — which isn't true.

A shutdown date is the earliest possible date, not a guarantee

Open the Gemini deprecations page (as marked updated on August 27, 2026) and the ordering is a little surprising. gemini-2.5-pro, gemini-2.5-flash, and gemini-2.5-flash-lite have no shutdown date announced. Meanwhile gemini-3.1-flash-lite carries May 7, 2027, with gemini-3.5-flash-lite named as its replacement. The newer model has an expiry; the older one doesn't. Version numbers and lifespan don't line up.

The same page adds a note: the dates in the table indicate the earliest possible dates on which a model might be retired. So the table isn't a promise of "safe until this day" — it's a floor, a "no sooner than this." And in practice, the Google AI developer forum carries reports of 2.5 models becoming unavailable ahead of their announced date, plus threads pointing out that Gemini API and Vertex AI advertise different retirement dates for the same model. Put a single date on your calendar and the gap between those two numbers is what trips you.

After reading that, I changed my design criterion for cascades from "a ladder of capability and cost" to "a ladder of capability and cost, spread across generations."

Unit prices don't sort by version number

A second assumption broke while I was rebuilding: that newer always costs more. Line up the Standard tier from the pricing page and it looks like this.

Model	Input / 1M tokens	Output / 1M tokens	Shutdown announced
`gemini-3.7-flash`	$0.75 (through 2026-12-31; $1.50 after)	$3.75 (then $7.50)	None
`gemini-3.5-flash`	$1.50	$9.00	None
`gemini-3.5-flash-lite`	$0.30	$2.50	None
`gemini-3.1-flash-lite`	$0.25 ($0.50 audio)	$1.50	May 7, 2027
`gemini-2.5-flash`	$0.30 ($1.00 audio)	$2.50	None
`gemini-2.5-flash-lite`	$0.10 ($0.30 audio)	$0.40	None

gemini-3.7-flash costs half what the previous-generation gemini-3.5-flash costs, through December 31, 2026. On the Lite side it inverts: the newer gemini-3.5-flash-lite is more expensive than gemini-3.1-flash-lite. Neither "newer is pricier" nor "newer is cheaper" holds.

This lands directly on cascade ordering. My original code had the Flash tier at $0.15 input / $0.60 output. Pulling the pricing page again, the real numbers were $0.30 / $2.50 — the output side was off by 4x. Leave unit prices baked into code for a few months and the more your fallback fires, the further your actual cost drifts from your model of it. The machinery you built to stay up quietly becomes machinery that loses money. Error handling turns into cost accounting, right at that point.

One more scheduling note: 3.7 Flash's $0.75 / $3.75 holds through December 31, 2026 and exactly doubles on January 1, 2027. If your estimate spans the new year, run it twice — once at each price.

Check at startup that the cascade is still alive

The fix is unglamorous. Verify the models exist once per deploy, and refuse to boot if the generations are bunched up. Finding out in 30 seconds at deploy time is much cheaper than finding out at 3 a.m.

# startup_check.py - run once when the app boots
from google import genai
 
def assert_cascade_healthy(api_key: str, cascade: list[ModelConfig]) -> None:
    """Verify cascade models still exist and span more than one generation."""
    client = genai.Client(api_key=api_key)
    available = {m.name.removeprefix("models/") for m in client.models.list()}
 
    missing = [m.name for m in cascade if m.name not in available]
    if missing:
        raise RuntimeError(
            f"Cascade references models that no longer exist: {missing}. "
            "Failing here instead of 404-ing in production"
        )
 
    # gemini-3.7-flash -> "3.7". Retirements tend to arrive per generation.
    generations = {m.name.split("-")[1] for m in cascade}
    if len(generations) == 1:
        raise RuntimeError(
            f"Cascade is concentrated in a single generation ({generations.pop()}). "
            "Every tier will go down together when it retires"
        )

client.models.list() returns names prefixed with models/, so strip that before comparing. I run this at startup and again from a weekly CI job. When CI fails, I get to plan the migration before any user touches it.

The cascade earlier in this article is the rebuilt version of that principle: gemini-3.1-pro / gemini-3.7-flash / gemini-3.1-flash-lite spans 3.1 and 3.7, so it passes assert_cascade_healthy. The bottom tier already has its May 7, 2027 notice, so swapping in gemini-3.5-flash-lite is on the calendar. A model with a named successor is the easy case. The hard case is the one with no date at all, that simply stops answering.

Detecting and Recovering from Streaming Interruptions

When using generateContentStream, you'll encounter a unique failure mode: the connection drops mid-stream while the HTTP status is 200. The response appears normal at first but stops delivering chunks unexpectedly.

import asyncio
 
async def stream_with_recovery(
    client,
    model: str,
    prompt: str,
    max_retries: int = 3,
    chunk_timeout: float = 30.0
):
    """Receive streaming responses with interruption detection"""
    accumulated_text = ""
 
    for attempt in range(max_retries + 1):
        try:
            response = client.models.generate_content_stream(
                model=model,
                contents=prompt
            )
 
            chunk_count = 0
            async for chunk in response:
                if chunk.text:
                    accumulated_text += chunk.text
                    chunk_count += 1
                    yield chunk.text
 
                # Detect safety filter blocks
                if hasattr(chunk, 'candidates') and chunk.candidates:
                    candidate = chunk.candidates[0]
                    if hasattr(candidate, 'finish_reason'):
                        if candidate.finish_reason == 'SAFETY':
                            raise SafetyFilterError(
                                "Response blocked by safety filter"
                            )
 
            # Completed successfully
            return
 
        except SafetyFilterError:
            raise  # Don't retry safety filter errors
 
        except (ConnectionError, TimeoutError) as e:
            if attempt < max_retries:
                delay = random.uniform(0, min(60.0, 1.0 * (2 ** attempt)))
                print(f"[Stream Recovery] Interruption detected — "
                      f"retrying in {delay:.1f}s ({attempt + 1}/{max_retries})")
                await asyncio.sleep(delay)
                # Request continuation from where we left off
                if accumulated_text:
                    prompt = (
                        f"Continue writing from where this text was interrupted. "
                        f"Resume seamlessly:\n\n"
                        f"---interrupted at---\n"
                        f"{accumulated_text[-500:]}\n"
                        f"---continue from here---"
                    )
            else:
                raise
 
class SafetyFilterError(Exception):
    pass

Production Monitoring and Alerting Design

Building error handling mechanisms is necessary but not sufficient. Production environments require real-time monitoring that detects changes in error rates and triggers alerts when thresholds are exceeded.

import time
from collections import deque
from dataclasses import dataclass, field
 
@dataclass
class APIMetrics:
    """Collect Gemini API usage metrics"""
    window_size: int = 300  # 5-minute window
 
    requests: deque = field(default_factory=deque)
    errors: deque = field(default_factory=deque)
    latencies: deque = field(default_factory=deque)
    tokens_used: deque = field(default_factory=deque)
 
    def record_request(self, success: bool, latency_ms: float,
                       tokens: int, model: str, status_code: int = 200):
        """Record request outcome"""
        now = time.time()
        self.requests.append((now, model, status_code))
        self.latencies.append((now, latency_ms))
        self.tokens_used.append((now, tokens))
        if not success:
            self.errors.append((now, model, status_code))
        self._cleanup(now)
 
    def _cleanup(self, now: float):
        """Remove entries outside the time window"""
        cutoff = now - self.window_size
        for q in [self.requests, self.errors, self.latencies, self.tokens_used]:
            while q and q[0][0] < cutoff:
                q.popleft()
 
    def get_error_rate(self) -> float:
        """Return current error rate (0.0 to 1.0)"""
        if not self.requests:
            return 0.0
        return len(self.errors) / len(self.requests)
 
    def get_p95_latency(self) -> float:
        """Return P95 latency in milliseconds"""
        if not self.latencies:
            return 0.0
        sorted_latencies = sorted(l[1] for l in self.latencies)
        idx = int(len(sorted_latencies) * 0.95)
        return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
 
    def should_alert(self) -> list[str]:
        """Check alert conditions and return list of triggered alerts"""
        alerts = []
        error_rate = self.get_error_rate()
        p95 = self.get_p95_latency()
 
        if error_rate > 0.1:  # Error rate exceeds 10%
            alerts.append(
                f"HIGH_ERROR_RATE: {error_rate:.1%} "
                f"(threshold: 10%, last {self.window_size}s)"
            )
        if error_rate > 0.5:  # Error rate exceeds 50%
            alerts.append(
                f"CRITICAL_ERROR_RATE: {error_rate:.1%} "
                f"— consider service suspension"
            )
        if p95 > 10000:  # P95 latency exceeds 10 seconds
            alerts.append(
                f"HIGH_LATENCY: P95={p95:.0f}ms "
                f"(threshold: 10,000ms)"
            )
 
        # Detect spikes in specific error codes
        error_codes = {}
        for _, _, code in self.errors:
            error_codes[code] = error_codes.get(code, 0) + 1
        for code, count in error_codes.items():
            if code == 429 and count > 10:
                alerts.append(
                    f"RATE_LIMIT_SPIKE: {count} 429 errors — "
                    f"review rate limiting strategy"
                )
 
        return alerts
 
# Usage example
# metrics = APIMetrics(window_size=300)
#
# # Record after each request
# metrics.record_request(
#     success=True, latency_ms=1250.0, tokens=3500,
#     model="gemini-3.7-flash", status_code=200
# )
#
# # Periodic alert check
# alerts = metrics.should_alert()
# for alert in alerts:
#     send_notification(alert)  # Slack / PagerDuty / Email

Tier-Based Operations Strategy and Optimization Roadmap

Gemini API rate limits vary dramatically by tier. Planning your operations strategy around your project's growth stage is essential for avoiding surprises.

Free Tier (Development & Validation)

With RPM limited to 5–15 on the Free Tier, the following measures are essential.

Request queuing: Prevent bursts by routing all requests through a queue, maintaining a pace of 1 request per 4 seconds or slower
Aggressive context caching: When reusing the same system prompt, leverage [context caching]((/articles/gemini-api/gemini-api-context-caching-cost-optimization) to reduce TPM consumption by up to 75%
Response caching: Cache responses for identical prompts at the application layer to eliminate redundant API calls

Tier 1 (Early Production)

At Tier 1 (billing enabled, $0+ spend), RPM expands to 150–300, but production services still require careful management.

Deploy model cascades: Fall back from Pro → Flash → Flash-Lite on errors
Enable circuit breakers: Trip to OPEN state after 5 consecutive 429 errors
Async batch processing: Offload non-real-time tasks to the [batch API]((/articles/gemini-api/gemini-api-production-pipeline-architecture) to minimize RPM consumption

Tier 2+ (Scaling Phase)

At Tier 2 ($250+ cumulative spend + 30 days), RPM jumps to 1,000+, shifting the bottleneck from RPM to TPM.

Optimize input tokens: Compress prompts and remove unnecessary context
Strengthen [security measures]((/articles/gemini-api/gemini-api-production-security): Higher traffic increases prompt injection risks
Multi-project isolation: Separate GCP projects by function to distribute rate limits

Indie Developer Field Notes — What I Actually Hit Running This in a Live Wallpaper App

The rest of this article has been blueprint-style. In this section I share what actually went wrong when I integrated Gemini API into the indie wallpaper and relaxation apps I've been running since 2014. Reading API documentation cannot surface the failure modes that only appear once real traffic hits.

Pitfall 1: Exponential backoff alone amplified a late-night burst

For the first two weeks of production, I ran with only the exponential backoff + jitter shown above — three retries, 1.0 second initial delay, factor 2.0, capped at 60 seconds. Daytime traffic was uneventful and everything looked fine.

Then I adjusted my AdMob mediation waterfall and overseas traffic surged between 02:00 and 04:00 JST. Wallpaper generation requests piled up, 429s came in clusters, and because I had not yet wired in a circuit breaker, accumulated retries dragged the entire worker pool down. About 75% of requests in that window timed out.

Two lessons. First, retry on its own can amplify an outage rather than smooth it out. Second, a circuit breaker is not a "phase two" optimization — it belongs in the first deployment.

Defense layer	Success rate after consecutive 429s	Average response time
Exponential backoff only	25%	8.2 s
+ Circuit breaker	87%	1.4 s
+ Token bucket	96%	1.1 s

These are measured over 60 days and roughly 1.24M requests on the live indie wallpaper app. Adding a circuit breaker on its own changed the feel of late-night bursts completely.

Pitfall 2: I got the model cascade ordering wrong

My first cascade was Flash-Lite → Flash → Pro, because I wanted to "optimize for cost first" and escalate to higher-quality models as a last resort. This is wrong.

When Flash-Lite hits 429, Flash is almost always squeezed at the same time. The result was that complex prompts that needed Pro fell back into Flash-Lite and produced noticeably shorter wallpaper descriptions. Within ten days I had three reviews on Google Play mentioning the issue.

I rebuilt the cascade as Pro → Flash → Flash-Lite, preserving quality while distributing load. Monthly cost went up by about 18%, but the rating recovered (4.7 → 4.8) and D7 retention improved from 22% to 27%, which easily paid for the increase.

Pitfall 3: Retries can starve AdMob requests in the same process

This one surprised me. While Gemini retries were piling up, AdMob ad loads in the same process started to lag. The cause was network thread contention — in Python asyncio or Node.js event loops, runaway retry tasks crowded out the AdMob SDK's own network calls.

The fix was to give the Gemini API client its own connection pool, and to drain the async queue immediately when the circuit breaker is OPEN. As an indie developer, AdMob revenue (still steady above ¥1M / month for me) is the floor that keeps the lights on, so protecting it from AI-feature contention is non-negotiable.

Pre-Launch Checklist for Indie Developers

This is the checklist I actually run before deploying any new Gemini-integrated feature.

Refresh the error classification map: Pin 429/500/503 as retryable and 400/403/404 as non-retryable in Slack or your team doc
Tune backoff initial and cap values under load: Sweep initial delay 0.5–2.0 s and cap 30–120 s across three 10-minute load tests on actual traffic
Set circuit breaker thresholds per tier: 3 consecutive failures for Free Tier, 5 for Tier 1, 8 for Tier 2+
Build the cascade as Pro → Flash → Flash-Lite: Quality-first ordering avoids review-rating regressions
Run the token bucket at 80% of the documented limit: Leaving 20% of headroom catches genuine bursts without false 429s
Wire metrics into Slack or PagerDuty: At minimum, alert on p95 latency > 5 s, error rate > 5%, and circuit breaker OPEN state
Keep 30 days of operations logs: Track retry success rate, breaker OPEN events, and per-tier error ratios in a spreadsheet, review monthly

For me, this checklist takes about three working days from "prototype works" to "ready for production traffic." That feels longer than building the prototype itself, but it is far shorter than the time you lose to incident response if you skip it.

A note on cost expectations

Other indie developers often ask what production really costs. On my wallpaper + relaxation apps, monthly Gemini API spend sits between ¥15,000 and ¥35,000 with Cloud Logging adding roughly ¥4,000 — a total of ¥20,000–¥40,000 per month, or about 4–7% of the apps' AdMob revenue.

That ratio is sustainable if AI is layered on top of an ad-supported revenue floor. If you instead rely on AI features for direct subscription revenue, plan for 2–3× the cost margin to absorb retention risk.

Summary — Building "Bulletproof" the Quiet Way

Running Gemini API in production isn't about hardening any single call — it's about layering error classification, exponential backoff, circuit breakers, token buckets, and a model cascade into a structure where failures stay quiet and contained. Three lessons stayed with me from running it. Don't over-architect on day one. Put the circuit breaker in before the first production deploy. And don't forget that the models you're protecting have a lifespan of their own.

For my next step I keep re-reading the FastAPI backend guide alongside the rate-limiting and 429 field notes, and I refresh the checklist above once a month. If you're running Gemini API in production on your own indie app, I hope these notes save you a few late-night incidents. Thanks for reading to the end.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.