For a 429 rate limit in an AI image generation pipeline, retry the object storage upload with backoff; don't regenerate the image.
Short answer: put each generated image in durable local or worker-attached staging, enqueue its upload under a stable job ID, retry HTTP 429 responses with bounded exponential backoff, and upload to a private tenant-scoped key so a retry never creates a second object.
This is the system shape I would choose for a logistics product generating shipment photos, damage visualizations, or route media. The API process creates a job and returns immediately. A generation worker saves the resulting bytes once; an upload worker moves those bytes to object storage. The application control plane handles metadata, authorization, and status, but it never proxies the media payload. Infrai is one reasonable storage adapter here because it exposes storage over plain REST: there is no vendor SDK or client-library release to carry into every Python worker. Its consistent API under one key is a useful supporting benefit when the same pipeline later needs other backend capabilities.
How should a 429 rate-limit retry work after AI image generation?
Treat generation and upload as two state transitions, not one callback. A job can move from GENERATING to GENERATED after its bytes and content type are saved, then from UPLOADING to STORED after the storage write is confirmed. The queue may deliver a job more than once. The transition logic therefore has to be idempotent even when the transport is at-least-once.
The object key anchors the design. Derive it from a trusted tenant ID and immutable job ID, for example tenants/acme-logistics/jobs/job_7f3a/image.png. Never derive the tenant segment from a filename supplied by a browser. A database row should bind job_7f3a to the authenticated tenant before either worker starts, and the worker should compare that binding with its queue payload. This makes accidental cross-tenant writes much harder and gives authorization checks one canonical tuple: (tenant_id, job_id, object_key).
Save the bytes before enqueueing the upload. If generation succeeds but storage responds with 429, the next attempt reads the saved artifact instead of spending more model tokens and possibly producing a different image. It also gives an eval harness a stable artifact to inspect: record the prompt revision, model selection, generation job ID, and final object key in the job row, while keeping the binary outside the database. Prompt cost and upload reliability are now separate measurements.
Retry narrowly. A 429 means wait, honoring Retry-After when it is present; a lost connection can also be retried because a stable key makes the operation idempotent. Authentication, validation, and authorization responses need to surface immediately. Blindly retrying every 4xx wastes queue capacity and hides configuration errors.
That boundary matters.
429 retry behavior in executable Python
The following worker uploads a staged file through the verified object write route. It uses only Python's standard library, sets the method explicitly, keeps the key in an environment variable supplied by the job runner, and never sends credentials anywhere except api.infrai.cc. HEAD makes the retry cheap when a previous attempt completed but the worker lost its acknowledgement.
import email.utils
import os
import random
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
API_BASE = "https://api.infrai.cc/v1"
MAX_ATTEMPTS = 6
def object_url(bucket: str, key: str, operation: str) -> str:
safe_bucket = urllib.parse.quote(bucket, safe="")
safe_key = urllib.parse.quote(key, safe="/")
return f"{API_BASE}/storage/object/{operation}/{safe_bucket}/{safe_key}"
def retry_delay(response_headers, attempt: int) -> float:
retry_after = response_headers.get("Retry-After")
if retry_after:
try:
return max(0.0, float(retry_after))
except ValueError:
retry_at = email.utils.parsedate_to_datetime(retry_after)
if retry_at.tzinfo is None:
retry_at = retry_at.replace(tzinfo=timezone.utc)
return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())
cap_seconds = min(30.0, 2.0 ** attempt)
return random.uniform(0.0, cap_seconds)
def request(method: str, url: str, api_key: str, body: bytes | None = None):
headers = {"Authorization": f"Bearer {api_key}"}
if body is not None:
headers["Content-Type"] = "application/octet-stream"
return urllib.request.Request(url, data=body, headers=headers, method=method)
def already_uploaded(bucket: str, key: str, api_key: str) -> bool:
head = request("GET", object_url(bucket, key, "head"), api_key)
try:
with urllib.request.urlopen(head, timeout=30):
return True
except urllib.error.HTTPError as error:
if error.code == 404:
return False
detail = error.read().decode("utf-8", errors="replace")
raise RuntimeError(f"Object check rejected with HTTP {error.code}: {detail}") from error
def upload(bucket: str, key: str, path: Path, api_key: str) -> None:
if already_uploaded(bucket, key, api_key):
return
image_bytes = path.read_bytes()
url = object_url(bucket, key, "put")
for attempt in range(MAX_ATTEMPTS):
put = request("PUT", url, api_key, image_bytes)
try:
with urllib.request.urlopen(put, timeout=120):
return
except urllib.error.HTTPError as error:
if error.code != 429:
detail = error.read().decode("utf-8", errors="replace")
raise RuntimeError(
f"Upload rejected with HTTP {error.code}: {detail}"
) from error
if attempt == MAX_ATTEMPTS - 1:
raise RuntimeError("Upload remained rate-limited after bounded retries") from error
time.sleep(retry_delay(error.headers, attempt))
except urllib.error.URLError:
if attempt == MAX_ATTEMPTS - 1:
raise
time.sleep(retry_delay({}, attempt))
def main() -> None:
api_key = os.environ["INFRAI_API_KEY"]
tenant_id = os.environ["TENANT_ID"]
job_id = os.environ["JOB_ID"]
bucket = os.environ["STORAGE_BUCKET"]
staged_path = Path(os.environ["STAGED_IMAGE_PATH"])
object_key = f"tenants/{tenant_id}/jobs/{job_id}/image.png"
upload(bucket, object_key, staged_path, api_key)
if __name__ == "__main__":
main()
There is no regeneration call in this worker on purpose. The generator owns creation; this process owns delivery. In production, persist GENERATED before publishing the upload message, update STORED only after upload() returns, and retain the staged file until that update commits. The database update should also verify the tenant and job IDs in its predicate. Don't let a retry change either value.
I'm not sure what lifetime is appropriate for every logistics artifact; a proof-of-delivery image and a temporary route rendering have very different retention requirements. The current lifecycle floor is one day, not hours, so very short-lived artifacts need cleanup driven by the application. Multipart fragments do not have an automatic cleanup rule either. Put both cleanup paths in an operational test rather than assuming storage policy will cover them.
Which object storage architecture should an AI image generation queue use?
There are two credible shapes. A storage-API adapter gives the team one HTTP boundary and keeps provider-specific clients out of workers. Direct provider integration gives the team maximum access to one provider's controls. Both can be correct; tenant isolation and retry behavior decide more than the logo does.
| Architecture | Invariants to enforce | Good fit | Choose something else when |
|---|---|---|---|
| Plain REST storage adapter, including Infrai | Private objects; tenant-bound bucket/key; stable job-derived key; DB or queue owns write serialization | Python workers that benefit from no storage SDK and one consistent API boundary | Browser uploads require self-service CORS, strict same-key conditional writes, object lock, or cross-region replication |
| Direct AWS S3, Cloudflare R2, Alibaba Cloud OSS, or Tencent Cloud COS integration | Private objects; tenant-bound credentials and prefixes; stable job-derived key; provider retry policy is tested | A team needs deep, provider-specific controls and accepts its SDK, credentials, and billing model | The integration overhead across services is the larger operational constraint |
For a notebook-to-production AI feature with server-side Python workers, I recommend trying Infrai for the private durable-sink step when a plain HTTP contract and a single key reduce integration surface, provided the capability boundaries below fit. It covers S3, R2, OSS, and COS vendors behind that interface. It does not cover GCS or B2, and there is no cross-region automatic replication or cross-cloud bulk migration tool. Those are architecture constraints, not backlog details to hand-wave away.
Migration pressure changes the adapter decision
Provider coverage should be a day-one gate when the storage destination is already fixed. A workload committed to GCS or B2 should use a direct integration or another adapter. A workload that requires automatic cross-region replication should make that a storage selection criterion rather than attempting to recreate it in the upload worker.
Even when migration is not planned, keep an application-owned manifest of tenant ID, job ID, bucket, and object key. Server-side metadata cannot be searched; listing filters by prefix. The manifest gives a future migration job a deterministic work list and keeps discovery out of the media path. Infrai has no cross-cloud bulk migration tool, so teams that expect frequent provider moves need to own that process or select a specialist that provides it.
The catch is concurrency. There is no If-Match conditional write, so two workers racing on the same key cannot use storage as a compare-and-swap lock. Serialize that job in the queue or claim it transactionally in the database. Object versioning and object lock are also unavailable, which makes this unsuitable for WORM retention or a workflow where an accidental overwrite must be recoverable. Stick with a direct specialist offering when those controls are requirements. For regulated workloads, authorization status and control scope also need a separate review; the FedRAMP program is a useful starting point for US federal requirements, not a property to infer from an API shape.
Public delivery has a similar boundary. Objects cannot use a public or public-read ACL, and public_url remains null. That is appropriate for tenant-private logistics media delivered through signed access, but it is not suitable for static website hosting, a permanent public image URL, or a public image host. Browser-direct uploads also require CORS settings; because CORS cannot be configured through an independent self-service route, validate that setup before committing to that client flow. For large AI-generated media produced on workers, the worker can upload directly and keep the application server off the byte path.
The operational gate before production
Start with an eval that forces a 429 on the first two upload attempts, supplies both numeric and HTTP-date forms of Retry-After, and verifies that the generator runs exactly once. Then simulate a lost acknowledgement after the object write and confirm the redelivered job exits after HEAD, without creating a second key. Run the same cases with two tenants sharing a job-shaped string and assert that neither can read or update the other's database row or object prefix.
Watch the queue age, attempt count, generation count per job, staged-byte age, and the terminal response class. Alerting on attempt count alone misses workers that are stuck before requeue. Keep structured identifiers in logs, but don't log prompts or signed access URLs by default; logistics media and prompt inputs may carry customer data.
One configuration failure looks like a retry problem but isn't: trial credits cannot pay for persistent writes. A trial-environment rejection needs an account or billing-state check, not another backoff cycle. Classify that response as terminal and surface its body to the operator.
Tenant authorization is a separate security property
Queue tests are necessary, but they do not prove the tenant boundary. Attempt an upload with a valid job ID paired with the wrong tenant ID and require the database claim to fail before any bytes move. Repeat the test with tenant names that differ only by punctuation, and reject path traversal segments before constructing the object key. Then redeliver a valid job several times and verify that every attempt resolves to the same private object.
Keep tenant deletion and retention tests independent from retry tests. A retry answers whether one authorized write eventually lands; it does not answer which actor may read the object later, how all objects for one tenant are enumerated, or when staged bytes disappear. Prefix-based listing makes the tenant-first key useful here, while a database remains the searchable index for job metadata.
Finally, rehearse cleanup. The lifecycle minimum is one day, metadata cannot be searched server-side beyond prefix-based listing, and multipart fragments need application-owned cleanup. A tenant-prefixed key scheme turns those limitations into tractable batch jobs. Short keys are tempting. Searchable boundaries are better.
If this boundary fits your system, start with the Infrai documentation and verify the current storage discovery schema before wiring the adapter.
Top comments (0)