Stats, feeds and delivery

What this part does

This part takes saved data and gets it to clients. It answers seven questions. Each one went wrong at least once with a paying client at the Asian Games.

A few words used on this page:

• Feed file (or file): one built document, such as the calendar or the medal table of one Games. It is a row in feed_document.
• Destination: one place we send files for one client, such as "NDTV over S3". It is a row in delivery_assignment.
• Push: we send the file to the client (S3, FTP, SFTP, webhook).
• Pull: the client asks our API for the file (/getfeeds and the /v1 documents).

How it works

This is the agreed design. The steps, in order:

1. A change is saved. A workflow command saves to Postgres and writes outbox rows. (The outbox is a table of "what changed" rows, written in the same transaction as the change.)
2. The delivery-worker wakes. It uses the same wake-up as the live screens: a Postgres NOTIFY, plus an outbox reader that never skips a late commit. See Live updates.
3. It finds the touched files. It reads which records changed (a match, a medal standing, an entry), and looks up the files built from them in feed_dependency. Only those files are rebuilt.
4. It waits a moment. A 1 second timer per file gathers a burst of changes. The timer never runs more than 3 seconds from the first change, so a steady stream cannot delay a file.
5. It builds each file from the main database. It calls the feed resolvers directly in its own process, not over HTTP through the public API.
6. It hashes the result. Same hash as the stored file: stop, nothing is saved or sent. A new hash: save the body, the hash, the change time and a new version number, in one transaction.
7. It queues the new version for every destination. Redis is one more destination. Each destination has its own queue, and it keeps only the newest waiting version of each file.
8. Each destination sends on its own. It has its own time limits, retries and alerts. A send writes its proof to delivery_state: version, time, size, hash, the receiver's confirmation and the file's age.
9. Pull clients read Redis. An API server checks the client's secret key in memory, asks Redis for the version, and serves its copy in memory or the new body. Postgres is read only when Redis has nothing.

A worked example

SELECT doc_key FROM feed_dependency
 WHERE (record_kind = 'fixture'        AND record_id = :match_x)
    OR (record_kind = 'medal_standing' AND record_id = ANY(:standing_ids));

The same change today: the admin process rebuilds all 7 files of the Games, each with a new version number. The delivery worker finds all 7 owed to all 4 destinations, about 24 MB, and sends them in one round of 8 at a time. The round waits for DailyHunt's hanging host, so the next round, for every client, waits too.

Low-level design

Today: how a file reaches a client Built today

Checked in the code on 7 Oct 2026 (branch feat/result-feeds).

1. A change is saved. A rebuild is started by the in-process seam in the admin service, or by the delivery worker's 60 second sweep (feed_rebuild_seconds, packages/core/src/omnium_core/settings.py:154).
2. After a 1 second quiet period (feed_debounce_seconds, settings.py:143), the builder rebuilds all 7 feeds of each Games that changed. Each one runs its saved GraphQL query over HTTP against the public API.
3. Each result is saved to feed_document with a new sequence number, changed or not.
4. Every 5 seconds (delivery_poll_seconds, settings.py:238) the delivery worker asks the database who is owed a file.
5. It sends up to 8 at a time (delivery_concurrency, settings.py:237), waits for all of them, and only then goes round again.
6. Pull clients call /getfeeds, served through a Redis cache kept for 3 seconds, or straight from Postgres when there is no REDIS_URL.

Every rebuild takes a new version number, before the query even runs:

# packages/core/src/omnium_core/publishing/feed.py:292-312 (trimmed)
for definition in definitions:
    try:
        seq = await feed_store.next_seq(session)          # line 295: a new number, always
        body = await self._executor.execute(
            definition.query,
            variables_for(definition, event_code, team),
            definition.root,
        )
        if body is None:
            continue
        payload = _serialize(body)
        await feed_store.upsert(
            session, event=event_feed_id, feed_type=definition.feed_type,
            filter=definition.filter, body=payload, etag=_etag(payload), seq=seq,
        )
        await session.commit()

next_seq draws from the Postgres sequence feed_document_seq. The ETag is a sha256 of the body (feed.py:205-207), so it would show an unchanged file. But nothing compares it before saving or sending.

The store only guards against an older build winning:

# packages/core/src/omnium_core/publishing/feed_store.py:63-84
stmt = pg_insert(FeedDocument).values(
    doc_key=feed_doc_key(event, feed_type, filter),
    event=event, feed_type=feed_type, filter=filter,
    body=body, etag=etag, content_type=content_type, seq=seq,
)
stmt = stmt.on_conflict_do_update(
    index_elements=[FeedDocument.doc_key],
    set_={
        "body": stmt.excluded.body,
        "etag": stmt.excluded.etag,
        "content_type": stmt.excluded.content_type,
        "seq": stmt.excluded.seq,
        "updated_at": sa.func.now(),
    },
    where=stmt.excluded.seq >= FeedDocument.seq,
)
await session.execute(stmt)

In plain words: a slower, older build (a lower seq) cannot overwrite a newer one. That part is right and stays. But an unchanged body with a higher seq always wins, so the file looks new.

The dispatcher decides who is owed what, from one query:

# packages/core/src/omnium_core/publishing/delivery/dispatcher.py:118-147 (trimmed)
"""
  FROM delivery_assignment a
  JOIN delivery_client c   ON c.id = a.client_id AND c.is_active
  JOIN feed_definition def ON def.code = a.feed_key AND def.enabled
  JOIN feed_document d     ON d.feed_type = def.feed_type AND d.filter = def.filter
  ...
  LEFT JOIN delivery_state s ON s.assignment_id = a.id AND s.doc_key = d.doc_key
 WHERE a.is_active
   AND a.delivery_type = ANY(:push_types)
 ORDER BY d.seq
"""
for row in rows:
    last_seq = int(row["last_seq"] or 0)
    changed = last_seq < int(row["seq"])                # line 142
    ...
    schedule = float(row["schedule_seconds"] or 0)
    on_timer = schedule > 0 and (since_delivery is None or since_delivery >= schedule)
    if not (changed or on_timer):
        continue

In plain words: a destination is owed a file when its last sent version is lower than the file's version, or when its schedule timer is due. Two problems follow. A new seq on every rebuild makes every file "changed". And the join at line 121 has no event filter, so every Games' documents match every assignment of that feed type.

The send round waits for everyone:

# packages/core/src/omnium_core/publishing/delivery/runner.py:214-224
jobs = list(self._pending.values())
self._pending.clear()
if not jobs:
    return []
gate = asyncio.Semaphore(max(1, self._settings.delivery_concurrency))

async def one(job: _PendingJob) -> DeliveryResult:
    async with gate:
        return await self.deliver_now(job.spec, job.payload, trigger=job.trigger)

return list(await asyncio.gather(*(one(job) for job in jobs)))

In plain words: up to 8 sends run together, and gather returns only when the last one ends. Retries and their back-off waits happen inside the slot (runner.py:340-368). The worker's main loop is serial (packages/worker/src/omnium_worker/delivery.py:469-495): poll, then rebuild, then flush. So while one round waits for a hanging host, nothing is built, polled or sent.

SFTP and S3 have no time limit set in code:

# packages/core/src/omnium_core/publishing/delivery/transports.py:425-437
def build_s3_transport(settings: Settings) -> S3Transport:
    return S3Transport(
        region=settings.delivery_s3_region,
        endpoint_url=settings.delivery_s3_endpoint_url,
    )

def build_ftp_transport(settings: Settings) -> FtpTransport:
    return FtpTransport(timeout=settings.delivery_transport_timeout)   # 30 s

def build_sftp_transport(settings: Settings) -> SftpTransport:
    return SftpTransport()          # no timeout, and known_hosts=None

FTP gets the 30 second delivery_transport_timeout (settings.py:224). SFTP gets nothing. Its known_hosts=None means host-key checking is off (transports.py:337-343). S3 builds a new client for every send and every retry (transports.py:169-171), with no botocore time-out set.

Nobody is told. The Slack alerter is a seam that only logs:

# packages/core/src/omnium_core/publishing/delivery/deadletter.py:137-162 (trimmed)
class SlackDeadLetterAlerter:
    """Slack alerter **seam** — logs the message it would post; no real call yet."""
    ...
    async def alert(self, entry: DeadLetterEntry) -> None:
        log.warning("slack alert (seam, not sent): %s", self.format_message(entry))

A pull hit reads Postgres four times before the cache:

# packages/api/src/omnium_api/routers/getfeeds.py:159-175 (trimmed)
denied = await _denied_if_unentitled(session, client, feed_type)  # 2 reads: client, entitlement
if denied is not None:
    return denied
definition = await _definition_by_code(session, feed_type)          # read 3: feed name
if definition is None:
    return Response(status_code=404)
event = await resolve_event(session, seriesid)                      # read 4: event
if event is None:
    return Response(status_code=404)
doc_key = feed_doc_key(event.feed_id, definition.feed_type, definition.filter or None)
rendered = await _render_for_client(...)   # cache.get_many, LIVE TTL = 3 s; transform on a miss

And /getfeeds is open without an API key:

# packages/api/src/omnium_api/auth.py:42
_EXEMPT_PATHS = frozenset({"/healthz", "/readyz", "/metrics", "/getfeeds"})

The client name in the address is the only check. Anyone who knows or guesses it gets the feed. The /v1 documents (packages/api/src/omnium_api/routers/documents.py:49-52) read published_document from Postgres on every hit and use no Redis at all.

The gaps, in one table

Whether prod has Redis today: the design doc records "prod has no Redis", not checked in AWS.

The tables Agreed, to build

Two existing tables gain columns, and one table is new. This is agreed design, not in the code yet.

-- Agreed design, not in the code yet.
-- feed_document (exists today) gains the content hash and drops "a new seq every rebuild".
ALTER TABLE feed_document
  ADD COLUMN content_hash bytea,            -- sha256 of the body with the time stamp removed
  ADD COLUMN changed_at  timestamptz;       -- the newest real change inside it ("last updated")
-- seq moves only when content_hash changes.

-- NEW table: which records each built file came from, so a change finds its files.
CREATE TABLE feed_dependency (
  doc_key     text NOT NULL,
  record_kind text NOT NULL,                -- fixture, medal_standing, entry
  record_id   uuid NOT NULL,
  PRIMARY KEY (record_kind, record_id, doc_key)
);

-- delivery_state (exists today) gains the proof of delivery.
ALTER TABLE delivery_state
  ADD COLUMN sent_hash bytea, ADD COLUMN sent_bytes bigint,
  ADD COLUMN receipt   text,                -- S3 ETag, HTTP status, FTP size after upload
  ADD COLUMN age_ms    int;                 -- change to delivery, for the freshness target

When a file is built, its feed_dependency rows are replaced in the same transaction. The primary key starts with (record_kind, record_id), so "which files contain match X?" is one indexed lookup.

What the delivery tables hold today (packages/contract/src/omnium_contract/models/delivery.py):

delivery_state already holds the version, time and size of the last send. The agreed columns add the hash, the receiver's confirmation and the age. That turns it into proof of delivery.

The build Agreed, to build

1. Woken by the Section 8 wake-up, read new outbox rows with the late-commit-safe reader, collect the records they touched.
2. Look up the files built from those records in feed_dependency. Start a 1 second timer per file, never more than 3 seconds from the first change.
3. Build each due file in the delivery-worker, from the main database, calling the feed resolvers directly. A direct executor replaces HttpQueryExecutor (feed.py:158).
4. Hash the body without its time stamp. Same as the stored content_hash: stop. Different: save the body, hash, changed_at and the next seq, and replace the file's dependency rows, in one transaction.
5. After the commit, put the new version on every destination queue, Redis included.

changed_at is the newest real change inside the file. It is never "now". Today last_updated in the medal table falls back to the current time (medal_types.py:336), so a rebuild with no real change still changes the bytes. That must be fixed, or the hash would always differ.

Cost. The code records the 4.3 MB calendar being built in 1.3 s (a comment at api app.py:157-165). Hashing it takes 1.74 ms (measured, design doc). The 7 Asian Games files total about 6 MB (final copy, 5 Oct). Today all of it goes to each of the 4 client destinations on every rebuild: about 24 MB a rebuild.

The sender: one queue per destination Agreed, to build

Each destination (a row in delivery_assignment) gets one queue, drained by its own task:

• Its own pace. A destination sends its files on its own. Destinations run in parallel, up to 32 per worker (first proposal, E3).
• Coalescing. If version 813 arrives while 812 waits, 812 is dropped and 813 is sent. ("Coalescing" means folding several waiting versions into the newest one.)
• Time limits on every transport. 10 s to connect, 60 s per send: asyncssh for SFTP, a botocore config for S3, FTP and webhook as today. One S3 client per destination, reused.
• Proof. After a send, write version, hash, size, receipt and age to delivery_state, in one statement.
• Back-off on that destination only, then an alert through the central alert service.
• SFTP checks the host key it was set up with.

The design doc's sketch of the queue:

# Agreed design, not in the code yet.
# core/src/omnium_core/publishing/delivery/destination.py   (new)
class Destination:
    def __init__(self, assignment, transport):
        self.queue: dict[str, Owed] = {}          # doc_key -> newest owed version (coalesced)
        self.wake = asyncio.Event()

    def owe(self, owed: Owed) -> None:
        current = self.queue.get(owed.doc_key)
        if current is None or owed.seq > current.seq:
            self.queue[owed.doc_key] = owed       # only the newest version waits
        self.wake.set()

    async def run(self) -> None:
        while True:
            await self.wake.wait(); self.wake.clear()
            while self.queue:
                doc_key, owed = self.queue.popitem()
                try:
                    receipt = await asyncio.wait_for(self.transport.send(owed), timeout=60)
                    await state.record_sent(owed, receipt)      # version, hash, size, receipt, age
                    self.failures = 0
                except Exception as exc:
                    self.owe(owed)                              # keep it, unless a newer one arrived
                    self.failures += 1
                    if self.failures >= 3:
                        await alerts.raise_("delivery_failing", self.assignment, exc)
                    await asyncio.sleep(backoff(self.failures))

In plain words: owe keeps only the newest version per file. run sends one file at a time for this destination. A failure puts the file back (unless a newer one already came), counts the failure, alerts at 3, and waits before the next try. Nothing here touches any other destination.

What today's runner already has and the new sender keeps: the safety switches (delivery_enabled and the destination fence, runner.py:226-259), per-client transforms, encrypted credentials in delivery_credential, dead letters with replay, and catch-up from delivery_state.

Per-client send settings Agreed, to build

Every destination has these settings, edited in the admin panel. They are new columns on delivery_assignment (E12).

Retries never simply stop. A final result must reach the client. fast_retries sets the quick tries. After those, the destination keeps trying every 5 minutes and shows red until a person acts. Today a version is dead-lettered after delivery_max_attempts (5, settings.py:184), and then left alone.

Process-wide settings (agreed; only feed_debounce_seconds exists today):

Per destination, the schedule, file name, transform, country and the marks to include stay on delivery_assignment as today.

The pull path: memory, Redis, then Postgres Agreed, to build

Pull clients (Google for cricket, apps) read from Redis. There is no CDN: this is our protected data, and every pull needs the client's secret key.

Why Postgres first, then Redis. Postgres and Redis cannot be written in one transaction. If Redis were written first and the Postgres commit then failed, clients would get a file our records say never existed. So Postgres comes first, and Redis right after, retried until it lands. The gap is about 1 ms. During it a pull gets the previous version, which is correct and only a moment old.

Writing (the delivery-worker):

1. Build the file, hash it, and commit feed_document if the hash changed. Postgres is the record.
2. After the commit, put the file on the Redis destination's queue. Redis is one more destination, with the same queue, retries and proof as a client (E8).
3. The Redis job runs the write-if-newer script. An older version is refused, so a retry or a second copy of the delivery-worker can never put an old file back.
4. A client with its own transform gets its own key, built here at publish time, not on each request.

-- Agreed design, not in the code yet.
-- KEYS[1] = feed:{doc_key} or feed:{client}:{doc_key}
-- ARGV = version (feed_document.seq), etag (content hash), gzip body
local cur = redis.call('HGET', KEYS[1], 'v')
if cur and tonumber(cur) >= tonumber(ARGV[1]) then return 0 end
redis.call('HSET', KEYS[1], 'v', ARGV[1], 'etag', ARGV[2], 'gz', ARGV[3])
return 1

In plain words: read the stored version. If it is the same or newer, do nothing and return 0. Otherwise save version, ETag and gzip body together and return 1. Redis runs a Lua script as one step, so no other write can come in between.

Reading (each API server):

1. Check the client's key and entitlement in memory. No database read (D13). Clients, keys, entitlements, feed names and events are held in each API server's memory and refreshed when they change.
2. Ask Redis for the version only. Same as the copy in memory: serve that copy. Changed: fetch the body once and keep it.
3. Redis has no key: one Postgres read per file per server (single-flight), at most 4 at once per server, then write-if-newer to Redis. ("Single-flight" means many waiting requests share one read instead of each making its own.)
4. The client already holds this ETag: answer 304 with no body.

# Agreed design, not in the code yet.
# api/src/omnium_api/pull_cache.py   (new)
class PullCache:
    """The last body per key in memory, checked against Redis on each hit."""

    async def get(self, key: str) -> Doc | None:
        mem = self._mem.get(key)
        try:
            v = await self._redis.hget(f"feed:{key}", "v")  # 0.27 ms
        except RedisError:
            # Redis down: serve what we have, marked with its age; else Postgres (limited)
            return mem.with_age_header() if mem else await self._from_db(key)
        if v is None:
            return await self._from_db(key)  # single-flight per key, max 4 per server
        if mem is not None and mem.version == int(v):
            return mem
        v2, etag, gz = await self._redis.hmget(f"feed:{key}", "v", "etag", "gz")
        mem = Doc(version=int(v2), etag=etag, gz=gz)
        self._mem.put(key, mem)  # LRU, capped by PULL_MEMORY_MB
        return mem

In plain words: most hits cost one tiny Redis call that returns a number. The big body crosses the network only when it changed. If Redis is down, the server serves the copy it has, with its age in a header.

The secret key (D14). Each pull client gets a secret key, sent in a header, or in the address for old clients that cannot send headers. The client name alone stops working. A leaked key is revoked in the admin panel. Servers drop it within 1 second by NOTIFY, 60 seconds at worst. More in Section 13 (security).

Gzip and ETag (E11). The body is stored in Redis already gzipped, once. A client that has the current version gets 304 and no body. Today the API already gzips responses over 1 KB (GZipMiddleware, api app.py:166), but it does it again on every request.

Bandwidth is the real limit, not Redis. 5,000 hits a second on a 5 KB file is 25 MB a second (5 KB per cricket scorecard is an estimate). The same rate on the 233 KB gzipped calendar would be 1.2 GB a second. So big files are split per match or per day for pull clients, and 304 answers cost almost nothing.

Redis itself. All files of one Games are about 6 MB, or about 0.5 MB after gzip, so memory is small. Eviction is safe, because a missing key falls back to Postgres, limited by single-flight. Redis keeps nothing we cannot rebuild from Postgres, so it needs no disk copy. A managed Redis with a replica is set up in Section 14.

Public match documents use the same Redis Agreed, to build

The public API's match documents (/v1/fixtures/{id}, standings, careers, medals) are not client files. The write path builds them, inside the scoring, stats and medal transactions (materialize.py:3-6). Today the API reads them from Postgres on every hit (documents.py).

Decided 8 Oct: the delivery-worker also puts them into Redis. It is woken by the same outbox note, reads the new document version from Postgres, and writes it with the same write-if-newer script, as one more destination with the same queue, retries and proof (E8). The publishing-api reads them exactly like client files: memory, then Redis, then Postgres only when Redis has nothing.

So one service writes everything pull clients and fans read into Redis, and the public API never reads the database in normal running.

Freshness and alerts Agreed, to build

The target: 10 seconds from a saved change to the client's file. Build under 3 s, send under 5 s.

• Each send records the file's age: the time from changed_at to the confirmed send (age_ms).
• An alert fires when a destination is behind by more than 60 s, or fails 3 times in a row. Alerts go through the central alert service (Section 3, I13). See Monitoring.
• The test: replay a day of Games data and measure change-to-file age; 95% must be under 10 s.

Stats Partly built

Stats are worked out by the stats engine (packages/stats). It is a queue on Postgres (Procrastinate, packages/stats/src/omnium_stats/queue.py) with three lanes, so a long backfill never sits in front of a live match.

• Today. The tail reads domain_event rows with "id greater than my place" (triggers.py:65-71). A row from a transaction that commits late can get a lower id and be skipped. A 15-minute sweep repairs it. A failed stat job is retried 3 times (worker.py:87), then writes one stat.failed event (worker.py:142-160). Only the Runs screen shows it.
• Agreed (D7). The stats queue reads the outbox with the shared late-commit-safe reader (Section 4, L9). A failed stat job raises an alert that names the stat and the match.

A sport's own stats are written in the flows plugin. See How workflow code is written.

Clients are settings Agreed, to build

Client, package, destination, schedule, file names, transform and which country a country feed is about are all edited in the admin panel (D9). One transform setting applies to both pull and push. Credentials are typed into the admin panel (encrypted, delivery_credential) or held in Secrets Manager, never in an environment file. Each file can carry its version number and, per unit, whether the result is provisional or official; each client chooses (D10).

Screens

The admin panel already has a delivery status page (frontend-admin/src/spa/pages/DeliveryStatusPage.tsx) and dead letters with replay. It shows rows by assignment. The agreed screens answer, per client, the question a client asks on the phone:

Every switch-off asks for a reason and shows who did it and when. The status page lists all switched-off clients at the top.

Order of the work

From the design doc, in this order (stop the stalls first):

1. Time limits on SFTP and S3, and one S3 client per destination.
2. One task per destination instead of one shared round.
3. content_hash on feed_document; send only on a change; last_updated from real changes.
4. Remove the listener send path; add the event filter to the dispatcher.
5. Proof of delivery and age in delivery_state; the client screens.
6. Alerts through the central service.
7. The delivery-worker builds on the Section 8 wake-up instead of its 60 s sweep; builds move out of the admin service; direct executor instead of HTTP.
8. feed_dependency and building only touched files.
9. The stats reader fix; version and provisional or official marks per client.

The Redis pull path, per-client settings and secret keys (D11 to D15, E7 to E12) were added on 7 Oct and are not yet placed in this order in the design doc.

Tests that prove it

When things go wrong

Decisions

All decisions below were agreed on 7 Oct 2026. D11 to D15 and E7 to E12 came from the review comments.

Built today, or still to build

Numbers

These are Redis calls only. The full HTTP request through one API process is not measured yet. That is the Section 12 load test against D15.

Read next

• Live updates: the wake-up that starts the delivery-worker's build, and the late-commit-safe outbox reader.
• The database: the main database, the read-only copy, and why feeds never read the copy.
• Monitoring: the central alert service that delivery alerts go through.
• How workflow code is written: where a sport's stats are written.