Polling GTFS-Realtime Feeds Efficiently

Poll no faster than the feed actually refreshes — usually every 15 to 30 seconds — and on each request echo the previous response’s ETag in an If-None-Match header and its Last-Modified value in If-Modified-Since. When the server answers 304 Not Modified you skip the download and the protobuf decode entirely. When it does return a fresh body, compare the decoded FeedHeader.timestamp against the last one you processed and drop the snapshot if it has not advanced. Together these three guards mean you fetch bytes only when they changed and do work only when the content is genuinely new.

Root Cause Analysis

A naive poller re-downloads and re-decodes the full feed on a fixed timer regardless of whether anything changed. Against a large agency emitting tens of thousands of entities every 20 seconds, that is a lot of wasted bandwidth and CPU — and, more importantly, a fast path to getting your client rate-limited or blocked. The waste comes from three distinct layers, each with its own fix.

The cadence mismatch is the coarsest. Producers rebuild their snapshot on a schedule; the FeedHeader.timestamp only advances that often. Polling every 2 seconds against a feed that refreshes every 20 fetches the identical payload ten times before it changes. The fix is simply to align your interval with the producer’s cadence, ideally read from a Cache-Control: max-age response header rather than guessed.

The transfer waste is the next layer. Even at the right interval, a plain GET always transfers the whole body. HTTP conditional requests solve this: the server stores a validator (an ETag and/or a Last-Modified date) with each version of the resource, and if you send that validator back on the next request, the server can answer 304 Not Modified with empty body when nothing changed. You pay one round-trip and zero bytes of payload.

The redundant-work waste is the subtlest. Some producers do not set validators, or set an ETag that changes on every rebuild even when the FeedHeader.timestamp is identical — for instance when the body is regenerated but carries the same underlying data. A conditional request cannot catch that; only decoding the header and comparing its timestamp can. This page combines all three guards into one loop.

Production-Ready Python Implementation

The poller below stores the last ETag, Last-Modified, and processed header timestamp between iterations, sends conditional headers, honors Cache-Control: max-age, and applies exponential backoff on transport errors. The process_snapshot callback is where your protobuf parsing work goes.

import time
import logging
from dataclasses import dataclass
from typing import Callable, Optional

import requests
from google.transit import gtfs_realtime_pb2
from google.protobuf.message import DecodeError

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

FEED_URL = "https://example-agency.gov/gtfs-realtime/trip-updates.pb"


@dataclass
class PollState:
    """Validators and dedup cursor carried across poll iterations."""
    etag: Optional[str] = None
    last_modified: Optional[str] = None
    last_header_ts: int = 0


def poll_once(
    session: requests.Session,
    url: str,
    state: PollState,
    process_snapshot: Callable[[gtfs_realtime_pb2.FeedMessage], None],
    default_interval: float = 20.0,
) -> float:
    """
    Perform one conditional poll. Returns the number of seconds to wait
    before the next poll (from Cache-Control max-age when available).
    """
    headers = {}
    if state.etag:
        headers["If-None-Match"] = state.etag
    if state.last_modified:
        headers["If-Modified-Since"] = state.last_modified

    response = session.get(url, headers=headers, timeout=15.0)

    # Derive the next interval from the producer's own cache directive
    interval = default_interval
    cache_control = response.headers.get("Cache-Control", "")
    if "max-age=" in cache_control:
        try:
            interval = max(5.0, float(cache_control.split("max-age=")[1].split(",")[0]))
        except (IndexError, ValueError):
            interval = default_interval

    if response.status_code == 304:
        logging.info("304 Not Modified — skipping download and decode")
        return interval

    response.raise_for_status()

    # Persist validators for the next conditional request
    state.etag = response.headers.get("ETag", state.etag)
    state.last_modified = response.headers.get("Last-Modified", state.last_modified)

    # Decode and apply the content-level dedup guard
    feed = gtfs_realtime_pb2.FeedMessage()
    try:
        feed.ParseFromString(response.content)
    except DecodeError:
        logging.error("Body was not valid protobuf — treating as transient, will retry")
        return interval

    header_ts = feed.header.timestamp
    if header_ts <= state.last_header_ts:
        logging.info("Header timestamp %d not newer than last %d — dropping duplicate",
                     header_ts, state.last_header_ts)
        return interval

    state.last_header_ts = header_ts
    logging.info("New snapshot: header_ts=%d, entities=%d", header_ts, len(feed.entity))
    process_snapshot(feed)
    return interval


def run_poller(
    url: str,
    process_snapshot: Callable[[gtfs_realtime_pb2.FeedMessage], None],
    max_backoff: float = 300.0,
) -> None:
    """Poll forever with conditional requests and exponential error backoff."""
    session = requests.Session()
    session.headers["User-Agent"] = "transit-data-poller/1.0"
    state = PollState()
    backoff = 0.0

    while True:
        try:
            interval = poll_once(session, url, state, process_snapshot)
            backoff = 0.0  # reset on any successful round-trip
            time.sleep(interval)
        except requests.RequestException as exc:
            backoff = min(max(backoff * 2, 5.0), max_backoff)
            logging.warning("Transport error (%s) — backing off %.0f s", exc, backoff)
            time.sleep(backoff)


def process_snapshot(feed: gtfs_realtime_pb2.FeedMessage) -> None:
    """Placeholder: hand the decoded FeedMessage to your parser."""
    trip_updates = sum(1 for e in feed.entity if e.HasField("trip_update"))
    logging.info("Processing %d trip_update entities", trip_updates)


if __name__ == "__main__":
    run_poller(FEED_URL, process_snapshot)

Step-by-Step Walkthrough

State carried across iterations. PollState holds the three things a poll must remember: the last ETag, the last Last-Modified string, and the last processed FeedHeader.timestamp. Keeping them in one dataclass makes the loop stateless in the functional sense — each poll_once reads and updates the same cursor.

Conditional request headers. When validators exist, the request carries If-None-Match (the ETag) and If-Modified-Since (the date). The server compares them against its current version and either returns a fresh 200 with a body or a 304 with none. Sending both covers producers that support only one.

Interval from the source of truth. Rather than hard-coding 20 seconds, the code reads Cache-Control: max-age from the response and uses it as the next sleep, clamped to a 5-second floor so a misconfigured max-age=0 cannot spin the loop. When the header is absent it falls back to default_interval.

304 short-circuit. A 304 returns immediately with no download, no decode, and no dedup work — the cheapest possible outcome. This is the single biggest efficiency win against a feed that changes less often than you poll.

Content-level dedup. Even on a 200, the header timestamp is compared against last_header_ts. If it has not advanced, the body is a duplicate — a regenerated payload with the same data — and process_snapshot is never called. Only a strictly newer timestamp advances the cursor and triggers work. Because the timestamp is POSIX UTC, this comparison needs no timezone handling; it is a plain integer compare.

Backoff on failure. Transport errors double the wait up to a ceiling, and any successful round-trip resets it to zero. A DecodeError is treated as transient rather than fatal, because a single truncated response should not kill a long-running poller.

Verification and Output

Confirm the dedup logic with a lightweight simulation that replays the same header timestamp twice and a newer one once, asserting the callback fires only for genuinely new snapshots.

processed_timestamps: list[int] = []


def recording_callback(feed: gtfs_realtime_pb2.FeedMessage) -> None:
    processed_timestamps.append(feed.header.timestamp)


def make_feed(ts: int) -> gtfs_realtime_pb2.FeedMessage:
    feed = gtfs_realtime_pb2.FeedMessage()
    feed.header.gtfs_realtime_version = "2.0"
    feed.header.timestamp = ts
    return feed


state = PollState()
for ts in (1715000000, 1715000000, 1715000020):  # duplicate, then newer
    feed = make_feed(ts)
    if feed.header.timestamp > state.last_header_ts:
        state.last_header_ts = feed.header.timestamp
        recording_callback(feed)

assert processed_timestamps == [1715000000, 1715000020], processed_timestamps
print("Dedup OK — processed:", processed_timestamps)

Expected log output during a healthy run alternates cheap 304s with occasional real snapshots:

INFO: 304 Not Modified — skipping download and decode
INFO: 304 Not Modified — skipping download and decode
INFO: New snapshot: header_ts=1715000020, entities=1843
INFO: Processing 1843 trip_update entities

Gotchas and Edge Cases

• Producers that omit validators. Not every feed sets ETag or Last-Modified. When both are absent, every request returns a full 200 body and the HTTP layer cannot help — the FeedHeader.timestamp dedup becomes your only guard against redundant processing, which is exactly why the loop keeps both mechanisms.

• ETags that change without the data changing. Some CDNs or dynamic generators emit a new ETag on every rebuild even when the payload is identical. You will get a 200 and download the body, but the timestamp comparison still prevents wasted downstream work. Never rely on the ETag alone as a content-identity signal.

• Clock skew and non-monotonic timestamps. A misbehaving producer can briefly emit a FeedHeader.timestamp that goes backward. The strict > comparison drops those, which is the safe choice — processing an older snapshot after a newer one would regress your state. Log the occurrence so a persistently stuck producer is visible.

• Respect rate limits and identify yourself. Set a descriptive User-Agent and honor any Retry-After header on a 429 or 503. Aggressive polling from an anonymous client is the fastest way to get an API key revoked; the cadence and conditional-request discipline here is as much about being a good consumer as about saving your own resources.

Frequently Asked Questions

Related

• Parsing GTFS-RT Protobuf in Python — the decode-and-flatten step that runs inside this loop’s process callback
• Timezone Handling and Schedule Normalization — why the POSIX header timestamp used for dedup needs no timezone conversion
• Up: Decoding GTFS-Realtime Protobuf Feeds — the FeedMessage and header model this poller depends on
• Section: GTFS-Realtime Integration — the end-to-end realtime pipeline · Home