Database and the read-only copy

What this part does

This part answers one big question: how does the database stay fast and correct as data piles up, event after event? It splits into eight smaller questions.

How it works

The rule is simple: code that writes, locks or needs exact data uses the main database. Code that can be a few seconds old uses the copy. Each piece of code says in code which kind of session it wants. A URL setting never decides it.

1. A command saves on the main database. Scorer taps, console fixes and import rows all become commands. A command takes a lock, writes and commits on the main database. Only the main database can do this.
2. The main database streams every change to the copy. Postgres sends its change log (the WAL, see the note) to the copy, which replays it. On a laptop this took 0.3 ms when idle.
3. The delivery-worker builds live files from the main database. It must see the change it was woken for, so it never reads the copy. It writes the file to Redis for pull clients and sends it to push clients.
4. Fans and apps read the copy. The public API, reports, exports, history screens and dashboards may see data a few seconds old. They read the copy, so they add no load to the main database during a match.
5. Pull clients read Redis first. A live match's file comes from Redis. On a Redis miss, a live match goes to the main database, never the copy. A finished or scheduled match may fall back to the copy.
6. A heartbeat measures the lag. One row on the main database is updated every second. Each API server reads that row on the copy every second. Now minus the row's time is the lag.
7. If the copy is over 5 s behind, reads move. Reads that set a freshness limit move to the main database through a small fixed pool, at most 4 connections per server. If the lag stays over 10 s for 60 s, an alert fires.
8. Old data leaves by the month. A daily job creates partitions 3 months ahead and drops months whose rows are all past their keep time. Nothing deletes millions of rows one by one.

What happens when the copy lags

Lag is normal. It is milliseconds most of the time. The design says exactly what happens at each level, so nobody has to decide in the middle of a match.

How much lag is normal?

This was asked during the design review. The answer, from measurements on one laptop (one machine, no network, 7 Oct):

On AWS: not measured. AWS publishes no typical number. Lag grows with heavy writes, long locks and a smaller copy. This is why routing uses our own heartbeat with a 5 s limit, and why we measure real AWS lag once the copy exists (Section 14, parked).

How others do it

These teams faced the same problem. Each page below was opened and read on 7 Oct.

A worked example

Low-level design

Today: one engine factory Built today

Every service builds its database engine through one function. Here it is, as it is today.

# packages/core/src/omnium_core/db.py:18-37
def build_engine(settings: Settings, *, url: str | None = None) -> AsyncEngine:
    """Create an async engine from settings (no connection is opened here). ..."""
    return create_async_engine(
        url or settings.database_url,
        echo=settings.echo_sql,
        pool_pre_ping=True,
        pool_size=settings.db_pool_size,
        max_overflow=settings.db_max_overflow,
        pool_timeout=settings.db_pool_timeout,
        pool_recycle=settings.db_pool_recycle,
        connect_args={
            "server_settings": {"statement_timeout": str(settings.db_statement_timeout_ms)}
        },
    )

# packages/core/src/omnium_core/settings.py:270-275
db_pool_size: int = 10  # persistent connections per engine
db_max_overflow: int = 5  # extra connections allowed under burst
db_pool_timeout: float = 10.0  # seconds to wait for a pooled connection
db_pool_recycle: int = 1_800  # recycle connections older than this (seconds)
# Server-side statement timeout (ms): a pathological query can't pile up.
db_statement_timeout_ms: int = 15_000

What this means:

• Each engine keeps 10 connections and may open 5 more under load.
• Every query stops after 15 s (statement_timeout). That is the only server setting.
• There is no lock_timeout (how long to wait for a lock), no idle_in_transaction_session_timeout (how long a transaction may sit open and idle) and no application_name (a name that tells you which service owns a connection).
• With one copy of each service, the design doc counts about 117 connections at most: api 15, admin 32, scheduler 34, stats 21, delivery 15 (from the 30 Sep plan; not re-counted for this page).

Today: the public API sends all reads to the copy Built today

# packages/api/src/omnium_api/app.py:50-60
engine = build_engine(settings)
# GraphQL is read-only → route its sessions to the replica when one is
# configured; otherwise reuse the primary engine (no redundant pool).
read_engine = (
    build_engine(settings, url=settings.replica_database_url)
    if (settings.replica_database_url)
    else engine
)
app.state.engine = engine
app.state.read_engine = read_engine
app.state.session_factory = build_serial_session_factory(read_engine)

If REPLICA_DATABASE_URL is set, the public API makes a second engine and gives every request a session on it. That covers GraphQL and also the document GET routes, which read published_document through the same factory (packages/api/src/omnium_api/deps.py:10-21, routers/documents.py). REPLICA_DATABASE_URL is set nowhere in the repo. On 21 Sep, staging's replica URL pointed at the main instance itself, so no real copy has ever been tested.

Today: why a copy would make client files out of date Built today

The feed builder does not read the database directly. It runs a saved GraphQL query over HTTP against the public API:

# packages/core/src/omnium_core/publishing/feed.py:158-194
class HttpQueryExecutor:
    """Run persisted queries over HTTP against omnium's public GraphQL API. ..."""

    async def execute(self, query: str, variables: dict[str, Any], root: str) -> Any:
        try:
            async with httpx.AsyncClient(timeout=self._timeout) as client:
                response = await client.post(
                    self._endpoint, json={"query": query, "variables": variables}
                )
        ...
        return (payload.get("data") or {}).get(root)


def default_executor() -> QueryExecutor:
    """The process query executor: HTTP against ``api_url`` ``/graphql``."""
    base = get_settings().api_url
    return HttpQueryExecutor(f"{base}/graphql")

Put the two snippets together. The file builder is woken by a new goal. It asks the public API. The public API reads the copy. The copy may not have the goal yet. The file goes out without the goal, and nothing rebuilds it until the next change. That is why the agreed order is: first the delivery-worker builds from the main database directly (Section 9, D1), then a copy is switched on (DB5).

Agreed: sessions in code Agreed, to build

Agreed design, not in the code yet. One class becomes the only way code reaches the database. Each route asks for the kind it needs.

# core/src/omnium_core/db.py  (agreed design, not in the code yet)
class Sessions:
    """The only way code reaches the database. Each route says which kind it needs."""

    def write(self) -> AsyncSession:
        """Main database: commands, locks, and every read that must be exact.
        An ordinary SQLAlchemy session: ORM, raw SQL, savepoints, advisory locks and bulk inserts all work."""

    def raw(self) -> AsyncContextManager[asyncpg.Connection]:
        """A driver connection on the main database for what a session does badly: COPY, LISTEN (Section 8)."""

    def read(self, max_age_s: float = 5.0) -> AsyncSession:
        """The copy, if it is at most max_age_s behind; else the main database through a small fixed pool."""
        if self._copy is None or self._lag.seconds() > max_age_s:
            return self._fallback()          # at most 4 connections per server on the main database
        return self._copy_session()          # read-only, 5 s statement limit

What each part does:

• write() gives an ordinary SQLAlchemy session on the main database. Anything SQLAlchemy can do works: the ORM, raw SQL, savepoints, advisory locks, bulk inserts. The user asked for this to stay flexible, and it does (DE1).
• raw() gives a plain asyncpg connection on the main database, for the two things a session does badly: COPY (fast bulk load) and LISTEN (wake-ups, Section 8).
• read(max_age_s) returns a copy session if the copy is at most max_age_s behind. Otherwise it returns a session from the small fallback pool on the main database. With no copy set, it uses the main database, so nothing changes until a copy exists.
• On the copy, every query has a 5 s limit.

A copy can cancel a read when replay needs to remove rows the read still uses. Postgres reports this as error code 40001. The agreed helper retries such a read once on the main database:

# core/src/omnium_core/db.py  (agreed design, not in the code yet)
async def read_with_retry(sessions: Sessions, work: Callable[[AsyncSession], Awaitable[T]]) -> T:
    try:
        async with sessions.read() as s:
            return await work(s)
    except DBAPIError as exc:
        if sqlstate(exc) != "40001":       # measured: the copy cancels with 40001 (conflict with recovery)
            raise
        async with sessions.fallback() as s:  # once, on the main database
            return await work(s)

In our read-only queries, 40001 can only mean the copy cancelled the query. So one retry on the main database is safe. Any other error is raised as normal.

Agreed: who reads where Agreed, to build

Agreed: the heartbeat Agreed, to build

Agreed design, not in the code yet. A new table with exactly one row:

-- new table (agreed design, not in the code yet)
CREATE TABLE replica_heartbeat (id int PRIMARY KEY CHECK (id = 1), beat_at timestamptz NOT NULL);

-- On the main database, every second, by the alert service (one working copy, Section 11):
UPDATE replica_heartbeat SET beat_at = now() WHERE id = 1;

-- On the copy, every second, by each API server:
SELECT extract(epoch FROM now() - beat_at) FROM replica_heartbeat;

Does this table fill up? No. This was asked in review. The CHECK (id = 1) allows only one row, and the beat updates it in place. Measured on a laptop after 10,000 beats: still 1 row in one 8 KB page, and 100% of updates were HOT (see the note). Each beat writes 115 bytes of WAL, about 9.7 MB a day.

Why not use AWS's own lag number? AWS measures lag as "now minus the last replayed change". On a quiet database nothing changes, so the copy looks minutes behind (AWS's read replica docs, as cited in the design doc, say "up to five minutes"). One write a second keeps the measure true. AWS's number is shown on the health page but never used to route.

What if the alert service stops? The beats stop. The copy looks behind, and reads move to the main database. That is the safe direction, and the alert service's own silence is alerted anyway. Clocks on the two servers are kept in sync by AWS; any clock error is far below 5 s (not measured).

# core/src/omnium_core/db_lag.py  (new; agreed design, not in the code yet)
class LagWatcher:
    """One per API server. Reads the heartbeat age on the copy every second."""
    def seconds(self) -> float:
        """The last measured lag. If the copy did not answer, infinity, so reads fall back."""

Today: partitions Partly built

A partition is one piece of a table that holds one range of dates. Postgres treats all pieces as one table for queries. You can remove one piece in one quick statement.

Three stats tables are already split:

# packages/contract/src/omnium_contract/models/stats_engine.py:495-502
#: The partitioned tables and their partition column. ``omnium_stats.partitions``
#: creates partitions ahead of time from this list.
PARTITIONED: dict[str, tuple[str, str]] = {
    # table: (partition column, grain)
    "stat_value": ("occurred_on", "year"),
    "stat_change": ("changed_on", "month"),
    "stat_snapshot": ("taken_on", "month"),
}

New partitions are created by ensure():

# packages/stats/src/omnium_stats/partitions.py:51-71
async def ensure(conn: AsyncConnection, *, today: date | None = None, ahead: int = 2) -> list[str]:
    """Create the missing partitions. Returns the names it created."""
    ...
    for table, name, start, end in plan(today, ahead=ahead):
        if name in existing:
            continue
        await conn.execute(
            sa.text(
                f"CREATE TABLE IF NOT EXISTS {name} PARTITION OF {table} "
                f"FOR VALUES FROM ('{start.isoformat()}') TO ('{end.isoformat()}')"
            )
        )

# packages/stats/src/omnium_stats/worker.py:216-217
async with proc.engine.begin() as conn:
    created = await partitions.ensure(conn)

What this means:

• ensure() creates today's partition and the next 2 periods.
• The only caller found is the stats worker at start-up (worker.py:217). The module's docstring says it also runs "once a day", but no daily caller was found in the code.
• So a stats worker that keeps running across a month boundary can send rows to the default partition. Nothing is lost, but nothing alerts on it either.
• Nothing drops old partitions.

Today: tables that grow forever Built today

integration_row (one row per row of every import run) is not partitioned (packages/contract/src/omnium_contract/models/integrations.py:138-142). On the laptop database it was 82% of everything:

Other tables that are never cleaned today: timeline_item, stat_change, stat_snapshot, stat_period, delivery_deadletter, raw_ingested_result, and finished stats-queue jobs. The outbox has a prune function, but no caller was found:

# packages/core/src/omnium_core/jobs_outbox.py:75-80
async def prune(session: AsyncSession, *, older_than: timedelta = PRUNE_AFTER) -> int:
    """Clear events past any conceivable replay. Returns how many went."""
    result = await session.execute(
        sa.delete(DomainEvent).where(DomainEvent.occurred_at < sa.func.now() - older_than)
    )
    return int(getattr(result, "rowcount", 0) or 0)

Delivery attempts are the one table pruned today (packages/worker/src/omnium_worker/delivery.py:379).

Agreed: monthly partitions and retention Agreed, to build

The rule (DB7): a table that can pass 10 million rows or 10 GB a year is split by month, and old data leaves by dropping whole months. Smaller logs get a nightly delete in batches.

Agreed design, not in the code yet. integration_row gets a monthly twin. The partition column must be part of the primary key:

-- integration_row, split by month (new table; agreed design, not in the code yet)
CREATE TABLE integration_row_v2 (
  id           bigint GENERATED ALWAYS AS IDENTITY,
  created_at   timestamptz NOT NULL DEFAULT now(),
  run_id       uuid NOT NULL, integration_id uuid NOT NULL, position integer NOT NULL,
  subject_key  varchar(256), subject_kind varchar(16), subject_id uuid, row_hash varchar(64),
  raw jsonb, mapped jsonb, outcome varchar(16) NOT NULL, message text, timeline_item_id bigint,
  PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);
CREATE INDEX ON integration_row_v2 (run_id, position);
CREATE INDEX ON integration_row_v2 (integration_id, subject_key, id);

-- Daily job: create 3 months ahead.
CREATE TABLE IF NOT EXISTS integration_row_v2_2027_01 PARTITION OF integration_row_v2
  FOR VALUES FROM ('2027-01-01') TO ('2027-02-01');

-- Daily job: drop a month once its newest row is past retention, unless a hold says no.
ALTER TABLE integration_row_v2 DETACH PARTITION integration_row_v2_2026_09 CONCURRENTLY;
DROP TABLE integration_row_v2_2026_09;

This SQL was also run on Postgres 15.19, because our docs say prod may be on 15.

How "keep 30 days" works with months. A month is dropped only when its newest row is past 30 days. So each row is kept at least 30 days and at most about 60. Example: the September partition holds rows up to 30 Sep. Those reach 30 days old on 30 Oct. The daily job drops September on its first run after that.

Moving a big table without a long lock. We never copy the old table into the new one:

1. Create the partitioned table under a new name.
2. Release code that writes new rows only to it.
3. History screens read both tables through one view for 30 days.
4. After 30 days the old table holds only expired rows. Drop it, which is one quick statement.

The keep rules, as agreed (DB8):

# core/src/omnium_core/retention/rules.py  (new; agreed design, not in the code yet)
RULES = [
    Rule("integration_row", by="month", column="created_at", keep=days(30)),
    Rule("domain_event",    by="month", column="occurred_at", keep=days(30)),  # its real time column
    Rule("command",         by="month", column="created_at", keep=days(30)),     # Section 4
    Rule("stat_change",     by="month", column="changed_on", keep=days(90)),     # already monthly
    Rule("stat_snapshot",   by="month", column="taken_on",   keep=days(90)),     # already monthly
    Rule("procrastinate_jobs", by="delete", keep=days(7),  where="status = 'succeeded'"),
    Rule("procrastinate_jobs", by="delete", keep=days(30), where="status = 'failed'"),
    Rule("delivery_attempt",   by="delete", keep=days(14)),                       # as today
    Rule("alert_signal",       by="delete", keep=days(7)),                        # Section 11
]
# Kept, never in RULES: timeline_item (the match log), stat_value, stat_period.

# core/src/omnium_core/retention/job.py  (new; agreed design, not in the code yet)
async def run_daily(sessions: Sessions, today: date) -> RetentionReport:
    """Create partitions 3 months ahead; detach and drop months past retention unless a hold
    covers them; delete small tables in batches of 10,000; write every action to retention_log."""

Raw import payloads are kept 30 days and import run summaries 1 year. A person can hold any rule, for example during a dispute. A hold is a recorded command (ops.hold_retention, ops.release_hold), and every drop is written to the new retention_log table.

Store only what changed (DB9, DE6). An import row identical to last time is counted on the run, not stored again. Only new or changed rows, and rows with a problem, keep their payload. Locally that removes 38% of stored import rows.

Agreed: hot rows stay cheap Agreed, to build

The match row's counter changes on every ball. That stays cheap only while updates are HOT. Measured on a laptop: 488,714 updates on 200 hot rows were 99.9% HOT, and the table grew only from 3.6 to 4.3 MB. A lower fill factor (70) did not help (99.8% HOT and a bigger table). So the rules are: no index on a column that changes on every ball, the default fill factor stays, and a test enforces both (DB11, DE10).

Today and agreed: migrations Partly built

Today each migration upgrade runs in one transaction, with the 15 s statement limit and no lock limit:

# packages/core/alembic/env.py:70-80
def do_run_migrations(connection: Connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
        include_object=include_object,
    )

    with context.begin_transaction():
        context.run_migrations()

Migration 059 built an index on the biggest table with the time limit switched off, and not CONCURRENTLY:

# packages/core/alembic/versions/20260921_059_integration_row_latest_index.py:39-47
def upgrade() -> None:
    op.execute("SET LOCAL statement_timeout = 0")
    op.create_index(
        INDEX,
        "integration_row",
        ["integration_id", "subject_key", sa.text("id DESC")],
        postgresql_include=["row_hash"],
        postgresql_where=sa.text(HELD),
    )

A plain CREATE INDEX blocks writes to the table until it ends. On a big table during a match, that stops scoring.

Agreed (DB12, DE7): the migration runner sets a 2 s lock limit with 5 retries. Index steps use CONCURRENTLY in their own autocommit block (CONCURRENTLY builds the index without blocking writes, but cannot run inside a transaction). A new CI check, scripts/check_migrations.py, refuses a plain index or a type change on a big table. Heavy changes wait for a gap between sessions.

Agreed: connections Agreed, to build

DB13 and DE8: one database user per service, each with its own connection limit (30 Sep plan). Every connection sets application_name, so the health page can group connections by service. A transaction left open and idle ends after 60 s. The copy gets its own small pools: 8 connections per API server. The fallback pool on the main database is 4 per server.

Settings

The DB_* names are the environment form of the fields in packages/core/src/omnium_core/settings.py:270-275.

The database health page Agreed, to build

One page in the admin panel shows what matters. It reads system views only. Its heavy queries run on the copy; the lag and lock parts run on the main database.

Order of work

1. Quick wins: application_name and the idle limit, the daily partition job, a daily call to the outbox prune that already exists, cleanup of finished stats-queue jobs, dropping the duplicate index.
2. Sessions in code, every route choosing read or write. With no copy set, read() uses the main database, so behaviour does not change.
3. Store only changed import rows.
4. integration_row and domain_event move to monthly partitions (new table plus view).
5. The migration runner limits and the CI check.
6. The health page and daily size snapshots.
7. The delivery-worker builds from the main database directly (Section 9, D1).
8. The heartbeat, the lag watcher and the fallback, tested against a real copy in Docker.
9. The copy itself on AWS, with Section 14 (parked). No AWS step without asking first.

When things go wrong

The rule for every case: a reader that needs exact data never gets old data, and the main database is never flooded because the copy went away.

Decisions

All fifteen DB decisions and ten DE decisions were agreed on 7 Oct. The last four rows were added in chat after the doc.

Built today, or still to build

Numbers

Read next

• Feeds and delivery: the delivery-worker, which must build from the main database, and how files reach clients.
• Monitoring: the alert service that writes the heartbeat and fires the copy-lag alert.
• Commands: why every write takes the main database, and how a retried command is never applied twice.