Processing GCS Uploads with Pub/Sub Notifications #

Create a Pub/Sub notification on the bucket for OBJECT_FINALIZE events with payload_format=JSON_API_V1 and an object prefix, attach a push subscription that delivers to an authenticated Cloud Run endpoint (or a pull subscription for workers), acknowledge with a 2xx only after processing succeeds, and deduplicate on bucket, name and generation, because Pub/Sub delivers at least once.

Google Cloud Storage reports new objects through Pub/Sub notifications: every finalised object (a completed single upload, a finished resumable upload, a composed object) publishes a message with the object’s metadata. Everything after that is ordinary Pub/Sub — subscriptions, acknowledgement deadlines, retry policies and dead-letter topics — which is where most integrations go wrong. This page is part of upload completion events in backend validation and cloud storage architecture. The AWS equivalent is routing S3 upload events with EventBridge, and the upload side is uploading to GCS with Node.js client libraries.

When to use this approach #

• Uploads land in a Cloud Storage bucket — directly from browsers via signed URLs or resumable sessions — and processing must start automatically.
• Consumers run on Cloud Run, GKE or Compute Engine, and you want Google-managed delivery with retries.
• You prefer Pub/Sub’s explicit subscriptions and dead-lettering over Eventarc’s abstraction (Eventarc uses the same notifications underneath).

Prerequisites #

1. A bucket and a Pub/Sub topic in the same project, and the Cloud Storage service agent (service-PROJECT_NUMBER@gs-project-accounts.iam.gserviceaccount.com) granted roles/pubsub.publisher on the topic.
2. gcloud CLI, or @google-cloud/storage and @google-cloud/pubsub for Node.
3. A Cloud Run service for push delivery, with a service account the subscription uses to sign requests (roles/run.invoker).
4. A dead-letter topic and subscription for messages that repeatedly fail.

From finalised object to processed #

Implementation #

Set up the notification and subscription:

# Allow Cloud Storage to publish to the topic, then attach the notification.
gcloud pubsub topics create uploads-finalized
gcloud storage buckets notifications create gs://uploads-prod \
  --topic=uploads-finalized \
  --event-types=OBJECT_FINALIZE \
  --object-prefix=originals/ \
  --payload-format=json

# Dead-letter topic, then a push subscription with retry and DLQ policy.
gcloud pubsub topics create uploads-finalized-dlq
gcloud pubsub subscriptions create uploads-to-processor \
  --topic=uploads-finalized \
  --push-endpoint="https://processor-abc123-ew.a.run.app/events" \
  --push-auth-service-account=pubsub-push@my-project.iam.gserviceaccount.com \
  --ack-deadline=300 \
  --min-retry-delay=10s --max-retry-delay=600s \
  --dead-letter-topic=uploads-finalized-dlq --max-delivery-attempts=5

The Cloud Run handler:

import { createServer, type IncomingMessage } from "node:http";
import { OAuth2Client } from "google-auth-library";

const auth = new OAuth2Client();
const AUDIENCE = process.env.PUSH_AUDIENCE!;               // the service URL
const PUSH_SA = "pubsub-push@my-project.iam.gserviceaccount.com";

interface PushEnvelope {
  message: { data: string; attributes: Record<string, string>; messageId: string; publishTime: string };
  subscription: string;
  deliveryAttempt?: number;
}

interface GcsObject { bucket: string; name: string; generation: string; size: string; contentType?: string; md5Hash?: string; crc32c: string }

async function verifyPush(req: IncomingMessage): Promise<void> {
  const token = (req.headers.authorization ?? "").replace(/^Bearer /, "");
  const ticket = await auth.verifyIdToken({ idToken: token, audience: AUDIENCE });
  if (ticket.getPayload()?.email !== PUSH_SA) throw new Error("unexpected push identity");
}

async function readJson<T>(req: IncomingMessage): Promise<T> {
  const chunks: Buffer[] = [];
  for await (const c of req) chunks.push(c as Buffer);
  return JSON.parse(Buffer.concat(chunks).toString("utf8")) as T;
}

const processed = new Set<string>();                         // replace with a durable store

createServer(async (req, res) => {
  if (req.method !== "POST" || req.url !== "/events") { res.writeHead(404).end(); return; }
  try {
    await verifyPush(req);
  } catch {
    res.writeHead(401).end();                                // not from our subscription
    return;
  }
  const env = await readJson<PushEnvelope>(req);
  const { eventType, objectGeneration } = env.message.attributes;
  if (eventType !== "OBJECT_FINALIZE") { res.writeHead(204).end(); return; }   // ack and ignore

  const obj = JSON.parse(Buffer.from(env.message.data, "base64").toString("utf8")) as GcsObject;
  const key = `${obj.bucket}/${obj.name}#${objectGeneration ?? obj.generation}`;

  try {
    if (!processed.has(key)) {
      console.log(JSON.stringify({ msg: "processing", name: obj.name, size: Number(obj.size),
        attempt: env.deliveryAttempt ?? 1 }));
      // do the work here: download, validate, transcode, record
      processed.add(key);
    }
    res.writeHead(204).end();                                // ack
  } catch (err) {
    console.error(JSON.stringify({ msg: "failed", key, err: String(err) }));
    res.writeHead(500).end();                                // nack: Pub/Sub will retry
  }
}).listen(Number(process.env.PORT ?? 8080));

Line-by-line on the settings that matter #

• --event-types=OBJECT_FINALIZE. Without it, the bucket also publishes metadata updates, deletions and archive events, and a consumer that ignores eventType processes deletes as uploads. Filter at the source.
• --object-prefix=originals/. Keeps your own outputs (thumbnails written back to the same bucket) from triggering more work. The prefix is the only object filter GCS notifications support; anything finer goes in the consumer.
• --payload-format=json. Puts the full object resource in the message data. With none, you get only attributes and must call the API to learn the size and type.
• --ack-deadline=300. For push subscriptions, the deadline is how long Pub/Sub waits for the HTTP response. Set it above your slowest processing time, or long jobs are redelivered while still running. For jobs longer than ten minutes, acknowledge quickly and hand off to a queue or workflow.
• --max-delivery-attempts=5 with a dead-letter topic. Without it, a poison message is retried until it expires (seven days by default). The Pub/Sub service agent needs publisher rights on the DLQ topic and subscriber rights on the source subscription for dead-lettering to work.
• Verifying the OIDC token. Push endpoints on Cloud Run can require authentication, but verifying the token’s email in code also protects endpoints reachable by other authenticated callers.
• Dedup key with generation. Overwriting an object creates a new generation; the same generation redelivered is a duplicate. Bucket plus name plus generation is exactly one version of one object.

Push or pull? #

Making the pipeline observable #

Pub/Sub exposes the metrics that tell you whether upload processing is healthy; wire them into alerts before you need them. subscription/oldest_unacked_message_age is the single most useful one: if it grows, consumers are failing or falling behind, and new uploads are waiting. subscription/dead_letter_message_count rising means something is failing repeatedly — inspect the DLQ subscription, fix the cause, and replay with a seek or a small republishing script. For push subscriptions, push_request_count broken down by response_class shows the ratio of acks to nacks and whether failures are 4xx (your handler rejecting messages) or 5xx and timeouts (your handler failing).

Log the messageId, deliveryAttempt and object generation with every processing log line. When an upload “never processed”, those three values let you tell apart a message that was never published (check the bucket’s notification configuration), one that is still being retried (attempt counts rising), and one that was processed but whose result was lost further down the pipeline.

Replaying uploads after a consumer fix #

Sooner or later a consumer bug means a day of uploads were acknowledged but processed wrongly, and they must be processed again. Pub/Sub offers two tools. Seek to a timestamp rewinds a subscription so every message published since that time is redelivered — but only if the subscription retains acknowledged messages (--retain-acked-messages with a retention window), which is off by default and worth enabling on upload subscriptions for exactly this reason. Seek to a snapshot does the same from a point you captured before a risky deploy.

When neither is available, rebuild the events from the bucket itself: list objects under the prefix with timeCreated in the affected window and publish a synthetic message per object, in the same JSON format, to the topic or directly to a dedicated replay subscription. Because the consumer deduplicates on bucket, name and generation, replayed messages for objects that were in fact processed correctly are skipped — unless you deliberately bump a processing version so they are redone.

Whichever route you take, throttle the replay. Thousands of redelivered messages arriving at a push endpoint at once scale Cloud Run to its maximum instances and can exhaust downstream quotas; set the service’s maximum instances or replay through a pull subscription drained by a fixed-size worker pool.

Configuration gotchas #

gcloud storage buckets notifications create fails with a permission error. The Cloud Storage service agent lacks pubsub.topics.publish on the topic. Grant roles/pubsub.publisher to service-PROJECT_NUMBER@gs-project-accounts.iam.gserviceaccount.com.

Messages go to the DLQ immediately. The Pub/Sub service agent (service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com) needs roles/pubsub.publisher on the dead-letter topic and roles/pubsub.subscriber on the source subscription.

Every message is delivered twice. Processing takes longer than the ack deadline; Pub/Sub redelivers while the first attempt is still running. Raise the deadline or acknowledge early and process asynchronously.

Events for composed or rewritten objects surprise the consumer. OBJECT_FINALIZE also fires for compose, rewrite and copies. Check metadata or the key prefix to tell user uploads from system writes.

Where the time goes #

Verification #

# The notification exists with the right filters.
gcloud storage buckets notifications list gs://uploads-prod --format=json | jq '.[] | {topic, event_types, object_name_prefix}'

# Upload a file and watch the subscription drain.
gcloud storage cp photo.jpg gs://uploads-prod/originals/test/photo.jpg
gcloud pubsub subscriptions describe uploads-to-processor --format='value(pushConfig.pushEndpoint)'
gcloud logging read 'resource.type="cloud_run_revision" AND jsonPayload.msg="processing"' --limit=1 --freshness=5m

Frequently Asked Questions