MCP Tools for Google Pub/Sub — pull subscriptions, backlog metrics, and seek

Topic and subscription model: one topic, many independent subscriptions

Pub/Sub's fanout model means every subscription on a topic receives every message independently. This is different from Kafka consumer groups (all consumers in a group share messages) and from RabbitMQ queues (a single message is delivered to one consumer per queue). For MCP tooling, this matters because:

• A backlog on Subscription A does not affect Subscription B — each subscription's cursor is independent.
• Deleting a subscription does not affect other subscriptions — the topic retains messages for other subscriptions.
• Creating a new subscription on a topic does not deliver historical messages — it starts from the point of creation (or from a snapshot if you seek backward).

import { PubSub } from '@google-cloud/pubsub';

const pubsub = new PubSub({ projectId: process.env.GOOGLE_CLOUD_PROJECT });

// List all subscriptions on a topic
async function getTopicSubscriptions(topicName) {
  const topic = pubsub.topic(topicName);
  const [subscriptions] = await topic.getSubscriptions();
  return subscriptions.map(s => s.name);
}

// Get subscription metadata including backlog indicators
async function getSubscriptionMetadata(subscriptionName) {
  const subscription = pubsub.subscription(subscriptionName);
  const [metadata] = await subscription.getMetadata();

  return {
    name: metadata.name,
    topic: metadata.topic,
    push_config: metadata.pushConfig, // null for pull subscriptions
    ack_deadline_seconds: metadata.ackDeadlineSeconds, // 10-600s, default 10
    retain_acked_messages: metadata.retainAckedMessages,
    message_retention_duration: metadata.messageRetentionDuration,
    expiration_policy: metadata.expirationPolicy, // null = never expires
    // Dead letter policy
    dead_letter_policy: metadata.deadLetterPolicy,
    // Filter expression (if set, only matching messages are delivered)
    filter: metadata.filter,
    // Ordering
    enable_message_ordering: metadata.enableMessageOrdering,
    // State
    state: metadata.state, // ACTIVE, RESOURCE_ERROR
  };
}

The ackDeadlineSeconds is the Pub/Sub equivalent of SQS visibility timeout — the subscriber must acknowledge each message within this window or it becomes redeliverable. The default is 10 seconds, which is short for any non-trivial processing. Unlike SQS, Pub/Sub's streaming pull automatically extends ack deadlines via keep-alive signals — but the synchronous pull API (pull RPC) requires explicit modifyAckDeadline calls for long-running processing.

Pull vs push delivery: choosing the right mode for MCP tools

Pull delivery: the subscriber calls the pull RPC or uses streaming pull, receives up to maxMessages messages per call, and sends acknowledge or modifyAckDeadline per message. The subscriber controls flow — it requests messages at its own pace. This is the preferred mode for MCP tools because it provides natural backpressure and does not require the subscriber to be reachable via HTTP.

Push delivery: Pub/Sub sends HTTP POST requests to a configured push endpoint when messages arrive. The push endpoint must be publicly reachable, respond with HTTP 2xx within the ack deadline, and for non-public URLs, must authenticate Pub/Sub's service account. Push is appropriate for stateless, HTTP-triggered workloads like Cloud Run services or Cloud Functions. For MCP servers running in your infrastructure, pull is almost always the right choice.

// Pull subscription: correct pattern for MCP tools
async function pullAndProcess(subscriptionName, maxMessages = 10) {
  const subscription = pubsub.subscription(subscriptionName, {
    flowControl: {
      maxMessages: 100,          // max outstanding unacked messages per client
      allowExcessMessages: false // block if outstanding > maxMessages
    }
  });

  // Streaming pull is more efficient than synchronous pull for continuous consumption
  subscription.on('message', async (message) => {
    try {
      const data = JSON.parse(message.data.toString());
      await processMessage(data, message.attributes, message.orderingKey);
      message.ack();
    } catch (err) {
      // nack = immediately redeliver; useful for transient errors
      // For permanent errors, nack will cause DLT after maxDeliveryAttempts
      message.nack();
    }
  });

  subscription.on('error', (err) => {
    console.error('Pub/Sub subscription error:', err);
    // Subscription errors are often transient (stream resets) — the client reconnects automatically
  });
}

// Synchronous pull: for MCP tools that process in controlled batches
async function pullBatch(subscriptionName, maxMessages = 10) {
  const subscription = pubsub.subscription(subscriptionName);
  const [messages] = await subscription.pull({ maxMessages });

  const ackIds = [];
  const nackIds = [];

  for (const message of messages) {
    try {
      await processMessage(JSON.parse(message.data.toString()), message.attributes);
      ackIds.push(message.ackId);
    } catch {
      nackIds.push(message.ackId);
    }
  }

  // Batch acknowledge/nack
  if (ackIds.length) await subscription.ack(ackIds);
  if (nackIds.length) await subscription.modifyAckDeadline(nackIds, 0); // immediate redeliver
}

Subscription expiration: the silent message loss trap

A Pub/Sub subscription with expirationPolicy.ttl set will be automatically deleted after the TTL period if no subscriber connects. The default is 31 days. For subscriptions that process periodic or low-frequency messages (batch jobs that run monthly, event-driven workflows with infrequent triggers), 31 days can be surprisingly short.

When a subscription is deleted due to expiration, all its unacknowledged messages are lost. The topic continues to exist — new subscriptions can be created — but historical messages are not retroactively delivered to new subscriptions (unless you seek to a snapshot taken before the expiration).

async function auditSubscriptionExpiration(projectId) {
  const [subscriptions] = await pubsub.getSubscriptions();
  const now = Date.now();
  const warningThresholdMs = 7 * 24 * 60 * 60 * 1000; // 7 days

  const results = subscriptions.map(s => {
    const meta = s.metadata;
    const expirationTtl = meta.expirationPolicy?.ttl;
    // TTL is a Duration proto: { seconds: '2678400', nanos: 0 } for 31 days
    const ttlSeconds = expirationTtl ? parseInt(expirationTtl.seconds) : null;

    return {
      name: meta.name,
      topic: meta.topic,
      expiration_ttl_days: ttlSeconds ? (ttlSeconds / 86400).toFixed(1) : 'never',
      never_expires: ttlSeconds == null,
      // Subscriptions without active subscribers expire based on last active time
      // The API does not expose "last active time" directly
      state: meta.state,
      warning: ttlSeconds != null && ttlSeconds < 30 * 86400,
    };
  });

  const expiringSoon = results.filter(s => s.warning);
  if (expiringSoon.length > 0) {
    console.warn('Subscriptions with short expiration TTL:', expiringSoon.map(s => s.name));
  }
  return results;
}

// Set expiration policy to never expire for critical subscriptions
async function disableSubscriptionExpiration(subscriptionName) {
  const subscription = pubsub.subscription(subscriptionName);
  await subscription.setMetadata({
    expirationPolicy: {} // empty object = never expire
  });
}

Backlog metrics: num_undelivered_messages vs oldest_unacked_message_age

The key health metrics for a Pub/Sub subscription live in Cloud Monitoring, not the Pub/Sub Admin API. The Admin API subscription metadata does not include backlog depth — that requires either Cloud Monitoring or the subscription's getSnapshot data. For MCP tools monitoring Pub/Sub health, query Cloud Monitoring:

import { MetricServiceClient } from '@google-cloud/monitoring';

const monitoring = new MetricServiceClient();

async function getSubscriptionBacklog(projectId, subscriptionName) {
  const subShortName = subscriptionName.includes('/')
    ? subscriptionName.split('/').pop()
    : subscriptionName;

  const [timeSeries] = await monitoring.listTimeSeries({
    name: `projects/${projectId}`,
    filter: [
      `metric.type="pubsub.googleapis.com/subscription/num_undelivered_messages"`,
      `resource.labels.subscription_id="${subShortName}"`
    ].join(' AND '),
    interval: {
      startTime: { seconds: Math.floor(Date.now() / 1000) - 300 }, // last 5 min
      endTime: { seconds: Math.floor(Date.now() / 1000) }
    }
  });

  const [ageSeries] = await monitoring.listTimeSeries({
    name: `projects/${projectId}`,
    filter: [
      `metric.type="pubsub.googleapis.com/subscription/oldest_unacked_message_age"`,
      `resource.labels.subscription_id="${subShortName}"`
    ].join(' AND '),
    interval: {
      startTime: { seconds: Math.floor(Date.now() / 1000) - 300 },
      endTime: { seconds: Math.floor(Date.now() / 1000) }
    }
  });

  const latestBacklog = timeSeries[0]?.points?.[0]?.value?.int64Value ?? 0;
  const latestAgeSeconds = parseInt(ageSeries[0]?.points?.[0]?.value?.int64Value ?? '0');

  return {
    subscription: subscriptionName,
    backlog_messages: Number(latestBacklog),
    oldest_unacked_age_seconds: latestAgeSeconds,
    oldest_unacked_age_minutes: (latestAgeSeconds / 60).toFixed(1),
    // oldest_unacked_age is the SLA signal: how far behind are we?
    sla_breached: latestAgeSeconds > 300, // alert if any message is older than 5 minutes
  };
}

num_undelivered_messages is the total count of unacknowledged messages — the backlog. oldest_unacked_message_age is the age in seconds of the oldest unacknowledged message — this is the SLA signal. A subscription with 1 million messages but oldest_unacked_age of 30 seconds is processing faster than messages arrive. A subscription with 100 messages but oldest_unacked_age of 3600 seconds is severely behind and has been for an hour.

Dead letter topics: configuration and monitoring

Pub/Sub dead letter topics (DLT) are separate topics (not subscriptions) that receive messages after they exhaust their maximum delivery attempts. Configuration requires IAM grants — Pub/Sub's service account needs pubsub.subscriber on the source subscription's topic and pubsub.publisher on the dead letter topic:

// Create subscription with dead letter policy
async function createSubscriptionWithDLT(topicName, subscriptionName, deadLetterTopicName, maxAttempts = 5) {
  const topic = pubsub.topic(topicName);

  // IAM: Pub/Sub service account needs roles for DLT forwarding
  // Service account: service-{project-number}@gcp-sa-pubsub.iam.gserviceaccount.com
  // Roles needed:
  //   - roles/pubsub.subscriber on source subscription
  //   - roles/pubsub.publisher on dead letter topic

  const [subscription] = await topic.createSubscription(subscriptionName, {
    deadLetterPolicy: {
      deadLetterTopic: `projects/${process.env.GOOGLE_CLOUD_PROJECT}/topics/${deadLetterTopicName}`,
      maxDeliveryAttempts: maxAttempts // 5-100; messages delivered > maxAttempts times go to DLT
    },
    retryPolicy: {
      minimumBackoff: { seconds: 10 },
      maximumBackoff: { seconds: 600 }
    }
  });

  return subscription.name;
}

// Monitor DLT for persistent failures
async function checkDeadLetterTopicHealth(projectId, deadLetterTopicName) {
  // Get all subscriptions on the dead letter topic
  const dltTopic = pubsub.topic(deadLetterTopicName);
  const [dltSubscriptions] = await dltTopic.getSubscriptions();

  const backlogs = await Promise.all(
    dltSubscriptions.map(s => getSubscriptionBacklog(projectId, s.name))
  );

  return {
    dead_letter_topic: deadLetterTopicName,
    total_dlt_messages: backlogs.reduce((sum, b) => sum + b.backlog_messages, 0),
    subscriptions: backlogs,
    has_failures: backlogs.some(b => b.backlog_messages > 0),
  };
}

When a message is forwarded to the dead letter topic, Pub/Sub adds attributes to the message: CloudPubSubDeadLetterSourceSubscription (the subscription that failed), CloudPubSubDeadLetterSourceTopicPublishTime (original publish time), and CloudPubSubDeadLetterSourceSubscriptionProject. These allow a dead letter consumer to understand the origin of each failed message and route it appropriately for manual review or automated retry.

Message ordering and snapshot/seek

Ordered delivery in Pub/Sub requires two things: (1) messages published with the same orderingKey will be delivered in publish order to each subscription, and (2) the subscription must have enableMessageOrdering: true. Publishing with ordering keys also requires using a regional endpoint (us-central1-pubsub.googleapis.com rather than pubsub.googleapis.com) to ensure ordering is preserved across the network path.

When ordered delivery is enabled and a message fails (nack or deadline exceeded), Pub/Sub stops delivering subsequent messages with the same ordering key until the failed message is successfully acknowledged — this prevents out-of-order delivery but can cause head-of-line blocking. To resume delivery after a stuck message, call modifyAckDeadline to 0 repeatedly to force exhaustion of maxDeliveryAttempts, or call seek to skip the problematic offset.

// Snapshot and seek: create a point-in-time cursor for replay
async function createAndSeekToSnapshot(subscriptionName, snapshotName) {
  const subscription = pubsub.subscription(subscriptionName);

  // Create snapshot: captures the current subscription cursor state
  const [snapshot] = await pubsub.createSnapshot(snapshotName, subscription);
  console.log('Snapshot created:', snapshot.name, 'expires:', snapshot.metadata.expireTime);

  // Later: seek the subscription back to the snapshot cursor
  // This replays all messages that were unacked at snapshot time
  await subscription.seek(snapshot.name);
  console.log('Subscription seeked to snapshot:', snapshot.name);

  return snapshot.name;
}

// Seek to a specific timestamp (replay from that point in time)
async function seekToTimestamp(subscriptionName, timestampMs) {
  const subscription = pubsub.subscription(subscriptionName);
  await subscription.seek(new Date(timestampMs));
}

// Clean up expired snapshots
async function listSnapshots(projectId) {
  const [snapshots] = await pubsub.getSnapshots();
  return snapshots.map(s => ({
    name: s.name,
    topic: s.metadata.topic,
    expire_time: s.metadata.expireTime,
  }));
}

Snapshots expire after 7 days or when the oldest message in the snapshot's topic is older than the topic's messageRetentionDuration (default 7 days). The seek operation is useful for replaying failed processing batches or rewinding a subscription after a deployment that introduced a bug — without needing to republish messages.

Further reading

• MCP Tools for RabbitMQ — Management API, queue depth, and dead-letter exchange
• MCP Tools for AWS SQS — visibility timeout, long polling, and DLQ redrive
• MCP Tools for Apache Kafka — consumer groups, offsets, and lag monitoring
• MCP Tools for Google Cloud Run — revision lifecycle, traffic splitting, and cold start
• MCP Server Message Queue — patterns for async task processing
• MCP Server Health Checks — implementing and monitoring protocol probes
• MCP Tools for DevOps — patterns across message queues and cloud infrastructure