Imports from Scout

What this part does

An import answers one question again and again: what did the site change, and should our data change with it? It is the main way data gets into omnium for a multi-sport event.

Words used on this page:

The design section was built around seven questions. Each one went wrong at least once at the Asian Games:

How it works

The agreed run has six steps. What runs today is described step by step in the low-level design below.

1. Fetch outside the database. A job reads the Scout run over HTTP with no database transaction open. A failed call is tried 3 times, with longer waits each time. A Scout run that failed its own checks is saved as "Scout failed" and alerted, not skipped quietly.
2. Record the run first. In its own short transaction, the run is saved with status received and its row count. From then on, no failure can erase the record that the run happened.
3. Map. Today's no-code steps, unchanged. Each row gets a key and a hash.
4. Compare with our data, not with the last read. For each row, the engine reads the current values of that match (or entry, or medal row), leaves out every pinned field, and keeps only fields that really differ. A fast path skips the read when nothing has touched the record since this integration last wrote it.
5. Save in batches of 25 matches. Each batch is one import command with the key run:batch. It takes match locks, has import time limits, and is one transaction. A failed batch is tried again on its own; the others carry on.
6. Mark missing. After a full run, keys the run no longer listed are counted. Missing from two full runs in a row opens a review item. Nothing is cancelled by an import on its own.

Three more rules sit around these steps:

• Unknown ids are held. A row naming a team, person or match we do not know waits with a clear question. It is tried again by itself on every later run. A held row is never reported as saved.
• Finished matches are read again. After the source marks a result official, its page is read again for a settle window: hourly, then every 6 hours, then daily, up to 72 hours.
• A person is told. A failed run, 2 Scout runs in a row that failed their checks, a batch that failed 3 times, a guard hold, or no new run for three intervals: each raises an alert and shows on the integration's card.

A worked example

Low-level design

This section has two parts: the code as it runs today, quoted from the repo, and the agreed design, which is not in the code yet. Each code block says which one it is. How commands themselves are written in the flows plugin is on Writing a workflow in code.

The tables today Built today

All six tables live in packages/contract/src/omnium_contract/models/integrations.py.

Today's run statuses (integrations.py:119-120): queued, running, written, nothing_new, older, paused_by_guard, failed, tried, cancelled. Row outcomes (integrations.py:175-176): accepted, unchanged, kept_by_hand, refused, not_placed, held, skipped, older, problem.

How a run starts today Built today

Scout's webhook is only a doorbell: it says a run passed and carries ten sample rows, never the full list (scout.py:10-23). omnium checks the signature and queues a pull job. It never fetches inside the web request.

packages/admin/src/omnium_admin/routers/integrations.py:2252-2284 (trimmed)

@doorbell_router.post("/doorbell/{code}")
async def doorbell(code, request, session, x_scout_signature=Header(default=None)):
    integration = (await session.execute(
        sa.select(Integration).where(Integration.code == code))).scalar_one_or_none()
    body = await request.body()
    if integration is None or not scout.verify_signature(
        body, x_scout_signature, integration.secret
    ):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "the signature did not match")
    ...
    if integration.state != "on":
        return {"queued": False, "reason": f"the integration is {integration.state}"}
    job = await jobs.enqueue(session, integration, "pull", {"via": "doorbell", "run": run.run_id})
    return {"queued": True, "job": job.id}

• The signature is HMAC-SHA256 over the exact body, in the X-Scout-Signature header with a sha256= prefix.
• An integration that is not on does not queue a pull.
• If no doorbell arrives, a loop every 60 seconds (CHECK_SECONDS, scheduler/__main__.py:43 and admin/integration_queue.py:42) queues a check for any integration not heard from for check_seconds (jobs.py:339-368).

Where the queue is worked. Today two processes work this queue: admin-api, on a thread of its own (admin/integration_queue.py; the integration_queue_in_admin setting is on by default, settings.py:257), and the scheduler (scheduler/__main__.py:131-132). A job is taken under a row lock, so each job runs in only one of them (jobs.py:65-78). In the plan the command service works the queue, in an import process of its own beside the process that answers scorers, at lower CPU priority; admin-api only receives the doorbell (decided 8 Oct 2026; Section 4, L15). A big pull then never shares a process with the console, and the scorer's process always gets the CPU first.

A restart must not block imports for an hour Proposed. Today a job cut off by a restart blocks its integration until it is an hour old (jobs.py:40, 65, 372-377). With imports in the command service, every deploy can cut a run. Proposed: the running job renews a timestamp every 10 s, and a job not renewed for 60 s is freed. The next run continues from the last batch done; batches carry the key {run_id}:{batch_no}, so nothing is written twice.

Fetching from Scout today Built today

packages/core/src/omnium_core/ingest/scout.py:152-181 (trimmed)

async def fetch_items(base_url: str, run_id: str, *, limit: int = 50_000) -> list[Any]:
    out: list[Any] = []
    offset = 0
    async with httpx.AsyncClient(timeout=_TIMEOUT) as client:   # 20 s, connect 8 s
        while True:
            response = await client.get(
                f"{_base(base_url)}/v1/runs/{run_id}/items",
                params={"offset": str(offset), "limit": str(_PAGE)},   # _PAGE = 500
            )
            ...
            out.extend(items)
            if len(items) < _PAGE:
                return out
            offset += len(items)
            if len(out) >= limit:
                raise ScoutError(f"run {run_id} has more than {limit} rows; refusing to read on")

• Rows come 500 a page, up to 50,000 rows.
• There is no retry. One failed HTTP call fails the job.
• A Scout run that did not pass is refused with an error when it is named (service.py:306-312). The scheduled check just picks the newest passed run (scout.py:104-150), so failing runs are skipped quietly.

One transaction for the whole job today Built today

packages/core/src/omnium_core/integrations/jobs.py:277-306 (trimmed)

async def work(factory, job_id: int) -> str:
    async with factory() as session:                       # one session, one transaction
        job = await session.get(IntegrationJob, job_id)
        integration = await session.get(Integration, job.integration_id) if job else None
        ...
        try:
            outcome = await perform(session, integration, kind, params)   # fetch + map + send + log
            status, error = "done", None
        except IntegrationError as exc:
            outcome, status, error = "failed", "failed", str(exc)
        except Exception as exc:
            log.exception("integration job %s failed", job_id)
            outcome, status, error = "failed", "failed", f"something went wrong: {exc}"
        if error is not None:
            await session.rollback()                        # the run record goes too
            integration = await session.get(Integration, integration_id)
            if integration is not None:
                integration.last_outcome = "failed"
                integration.last_error = error[:2000]
        ...
        await session.commit()

• The session's first read opens a transaction. The HTTP fetch then happens inside it.
• On any error the rollback removes the integration_run row. Only the job's error and the integration's last_error stay.
• There is no alert: only a log line and a screen update.

Map: key and hash today Built today

packages/core/src/omnium_core/integrations/mapping.py:342-346

parts = [read_path(row.value, key) for key in settings.match.key_fields]
row.key = ":".join(str(part) for part in parts) if all(p is not None for p in parts) else None
row.row_hash = hashlib.sha256(
    json.dumps(row.value, sort_keys=True, default=str).encode()
).hexdigest()

The key is built from the version's match.key_fields (default ["externalId"]). The hash is sha256 over the mapped value with sorted keys, so the same value always gives the same hash. The agreed design keeps this step as it is.

Guards today Built today

Two kinds of guard run today. The first looks at size, the second at how many rows failed to place.

packages/core/src/omnium_core/integrations/service.py:805-832 (trimmed)

_COLLAPSE_SHARE = 0.1          # service.py:120

async def _too_few(session, integration, fetched, settings) -> tuple[bool, str | None]:
    usual = await _usual_rows(session, integration)         # median of the last 5 written runs
    if not usual or fetched.rows_collected >= settings.safety.min_share * usual:   # min_share 0.5
        return False, None
    why = f"Scout sent {fetched.rows_collected} rows; it usually sends about {round(usual)}."
    return fetched.rows_collected < _COLLAPSE_SHARE * usual, why

packages/core/src/omnium_core/integrations/service.py:860-883 (trimmed)

def _guard(mapped, to_send, sent, settings):
    held = sum(1 for row in mapped.rows if row.settled == "held")
    considered = len(to_send) + held
    ...
    not_placed = words.count("not_placed") + words.count("held") + held
    refused = words.count("refused")
    if not_placed / considered > settings.safety.max_not_placed:      # default 0.1
        return ("not_placed", f"{not_placed} of {considered} rows were not placed or held. "
                              "Nothing from this run was written.")
    if refused / considered > settings.safety.max_refused:            # default 0.1
        return ("refused", ...)

Defaults are in integrations/settings.py:173-181. Section 3 replaces the size rule with I12 (below).

Change detection today Built today

packages/core/src/omnium_core/integrations/service.py:835-851 (trimmed)

async def _changed(session, integration, target, ready, replay_of=None) -> list[Row]:
    if not target.max_rows or replay_of is not None:
        return ready
    last = await _last_hashes(session, integration)
    return [row for row in ready if not (row.key and last.get(row.key) == row.row_hash)]

• _last_hashes reads the newest row_hash per key from integration_row (service.py:342-365).
• It compares with what this integration last held, not with the database. That is the gap Section 3 closes (I3).
• It applies only to imports whose rows stand alone (max_rows set). Today that is games.import_units, up to 1,000 rows per command (UNIT_ROWS_PER_COMMAND, flows/games/commands.py:52). The medal table and entries go in whole, every run.
• This query hit the 15 s statement limit on 20 Sep and killed ten checks in a row, before an index was added (service.py:347-352).

How rows become commands today Built today

packages/core/src/omnium_core/integrations/service.py:422-478 (trimmed)

size = target.max_rows or max(len(rows), 1)
actor = Actor(kind=INTEGRATION, name=f"integration:{integration.code}", source_code=integration.code)
for index, start in enumerate(range(0, len(rows), size)):
    chunk = rows[start : start + size]
    payload = {
        "rows": [{**row.value, "_position": row.position, ...} for row in chunk],
        "idSource": settings.match.id_source or integration.code,
    }
    outcome = await engine.run(
        session, workflow=target.workflow, event=target.event, fixture=None,
        code=integration.import_command, payload=payload, actor=actor,
        idempotency_key=f"{key}:{index}",          # key = "<scout run id>:v<version>"
        dry_run=dry_run,
    )
    if outcome.duplicate:                           # sent before: every row "unchanged"
        ...
        continue
    if not outcome.accepted:                        # refused: every row "refused"
        ...
        continue
    for row in chunk:
        sent.outcomes[row.position] = outcome.rows.get(row.position) or ("accepted", None, None)

In plain words:

• Rows are cut into chunks of the command's maxItems (1,000 for units). A medal table or entry list goes as one chunk.
• Each chunk is one import command on the event (fixture=None), sent by an actor of kind INTEGRATION.
• The idempotency key is the Scout run id, the version number and the chunk number. Sending the same run twice returns duplicate, and nothing is written twice.
• Every chunk of the run sits inside one savepoint (service.py:615). A guard can take back the whole run with one rollback.
• Because the subject is the event, the engine takes an advisory lock on the whole event (engine.py:116), held until the job commits. Imports also skip the 10 s time limit a person's command gets (engine.py:110-115, scoring_txn_timeout_ms in settings.py:290); only the 15 s statement limit applies (settings.py:275).

The writer and pins today Built today

The import command's writer compares each field with the database and skips pinned ones.

packages/core/src/omnium_core/ingest/fixtures.py:1141-1150

pinned = set(row.pinned or [])
changed = False

def allowed(name: str, same: bool) -> bool:
    if same:
        return False
    if name in pinned:
        report.kept_by_hand += 1
        return False
    return True

fixture.pinned and fixture_competitor.pinned are JSONB lists of field names a person set by hand: name, start, venue, status, result, competitors on a fixture (models/fixtures.py:127-132). So the writer already compares with the database. Only the step before it, _changed, decides on the last read, and drops rows the writer would have fixed.

Missing units today Built today

After a written run with build all or day (a day build only up to its own Games day), units the run no longer lists are set to CANCELLED, unless they hold a result, a medal or a pin. The guard on which day and discipline a run covered is careful (retire.py:1-60), but the action is a direct write, outside any command.

packages/core/src/omnium_core/ingest/retire.py:279-283

await session.execute(
    sa.update(Fixture)
    .where(Fixture.id.in_([unit.fixture_id for unit in report.cancelled]))
    .values(status=FixtureStatus.CANCELLED)
)

The setting retire_missing_units is on by default (settings.py:162). In one test, 132 of 270 units it called gone were still on the site an hour later (retire.py:30-34). The cancel step catches its own error but has no savepoint, so a database error there aborts the transaction, and the whole good run rolls back (service.py:701-712).

The new table Agreed, to build

Agreed design, not in the code yet. One row per integration and key: what this integration last saw and wrote. It replaces the _last_hashes query.

-- NEW (K1)
CREATE TABLE integration_key_state (
  integration_id  uuid    NOT NULL REFERENCES integration(id) ON DELETE CASCADE,
  subject_key     text    NOT NULL,      -- e.g. "ATH:M100METRES---FNL-000100"
  subject_kind    text    NOT NULL,      -- fixture, entry, medal_row ...
  subject_id      uuid,                  -- our record, once known
  last_hash       bytea,                 -- sha256 of the mapped value last seen
  applied_seq     bigint,                -- the record's changed_seq right after this integration wrote it
  last_seen_run   uuid,                  -- the last run that listed this key
  missing_runs    int     NOT NULL DEFAULT 0,
  official_at     timestamptz,           -- when the source first marked it official (I8)
  next_read_at    timestamptz,           -- when to read it again in the settle window (I8)
  PRIMARY KEY (integration_id, subject_key)
);
CREATE INDEX ix_key_state_next_read ON integration_key_state (next_read_at)
  WHERE next_read_at IS NOT NULL;

-- integration_run gains the batch count and Scout's own status.
ALTER TABLE integration_run
  ADD COLUMN batch_total int,
  ADD COLUMN batch_done  int NOT NULL DEFAULT 0,
  ADD COLUMN scout_status text;          -- passed / failed, as Scout reported it
-- status values: received, applying, done, partly_done, failed, scout_failed, older, paused_by_guard

The fast path Agreed, to build

Agreed design, not in the code yet. Keys that can be skipped: same value as last time, and nobody touched the record since.

SELECT k.subject_key
  FROM integration_key_state k
  JOIN fixture f ON f.id = k.subject_id
 WHERE k.integration_id = $integration
   AND k.subject_key = ANY($keys)              -- the run's keys, 1,000 at a time
   AND k.last_hash = ANY($hashes)              -- matched to the same key in code
   AND f.changed_seq = k.applied_seq;

Both conditions must hold (K2). Same hash alone is not enough: that is today's bug. Every key the fast path does not skip is compared with the database inside its batch. Measured on a laptop on 6 Oct with 1 million key rows: 1.69 ms median and 3.85 ms at the 99th percentile for 1,000 keys (from the design doc).

One batch, statement by statement Agreed, to build

Agreed design, not in the code yet. Each batch is one import command, sent through the Section 4 road (see Commands).

BEGIN;                                  -- import pool: lock 5 s, query 30 s, idle 30 s
SELECT id, changed_seq FROM fixture WHERE id = ANY($ids) ORDER BY id FOR NO KEY UPDATE;  -- up to 25 match locks
SELECT outcome, answer FROM command WHERE command_id = $run || ':' || $batch;            -- a repeat? -> duplicate
SELECT ... current values of these matches and their sides ...;
-- in code: diff = site values - current values - pinned fields; empty diff = nothing to write
UPDATE fixture SET <changed fields>, changed_seq = $seq WHERE id = ...;                  -- only real changes
UPDATE fixture_competitor SET <changed fields>, changed_seq = $seq WHERE id = ...;
INSERT INTO timeline_item (...) VALUES (...);            -- one log row for the batch, with the diff
INSERT INTO integration_key_state (...) VALUES (...)
  ON CONFLICT (integration_id, subject_key)
  DO UPDATE SET last_hash = EXCLUDED.last_hash, applied_seq = EXCLUDED.applied_seq,
                last_seen_run = EXCLUDED.last_seen_run, missing_runs = 0;
INSERT INTO integration_row (...) VALUES (...);          -- only changed, held and problem rows
INSERT INTO domain_event (...); INSERT INTO command (...);
UPDATE integration_run SET batch_done = batch_done + 1 WHERE id = $run;
COMMIT;

• Locks sorted by id. Two writers on the same matches always lock in the same order, so they cannot deadlock.
• The batch key run:batch is checked first. A repeat after a crash answers duplicate and writes nothing.
• Key state is written in the same transaction as the data. So the fast path can never be ahead of the data.
• Measured with 25 matches and their sides: 1.75 ms median, 3.95 ms at the 99th percentile (laptop, 6 Oct, from the design doc). A 1,240-row schedule run with 50 changed matches is about 2 batches plus the fast path: under 10 ms of database time (estimate in the design doc).

The pipeline as code Agreed, to build

Agreed design, not in the code yet. A new file, integrations/pipeline.py, cuts the run into short transactions around today's mapping and writers.

# core/src/omnium_core/integrations/pipeline.py   (new)
async def run(sessions: Sessions, integration_id: uuid.UUID, via: str) -> RunResult:
    fetched = await scout.fetch_with_retries(integration)          # no transaction open (K4)
    async with sessions.short() as s:                              # transaction 1
        run = await records.received(s, integration, fetched)      # status "received" (I2)
        if fetched.scout_status == "failed":
            return await records.scout_failed(s, run, fetched.reason)        # alert after 2 in a row (I10)
        if await records.is_older(s, integration, fetched):
            return await records.older(s, run)
    rows = mapping.map_items(fetched.items, version.steps, lookups)          # pure, today's code
    if (stop := guards.too_few(rows, history)) is not None:
        return await records.paused_by_guard(sessions, run, stop)
    async with sessions.short() as s:
        skip = await keystate.read_skippable(s, integration.id, rows)        # K1, K2
    todo = [r for r in rows if r.key not in skip]
    batches = chunk_by_match(todo, size=25)                                  # sorted by our id
    await records.set_batches(sessions, run, len(batches))
    for n, batch in enumerate(batches):
        answer = await road.run(sessions.imports(), subject=integration.event_subject,
                                cmd=Command(key=f"{run.id}:{n}", code=integration.import_command,
                                            input={"rows": batch, "runId": str(run.id)}))
        if answer.outcome == "try_again":
            answer = await retry_batch(...)                                  # 3 tries, growing waits (I5)
        if answer.outcome not in ("accepted", "duplicate"):
            return await records.partly_done(sessions, run, n, answer)       # alert (I10)
    async with sessions.short() as s:                                        # last transaction
        await missing.mark_missing(s, integration, run, full=fetched.is_full_build)   # K5
        await records.done(s, run)
    await followups.queue(sessions, integration, run)        # pages, draws, medallists, settle window (K7)

Reading it top to bottom:

1. The fetch runs before any transaction, with 3 retries.
2. Transaction 1 saves the run as received. A Scout-failed run or an older run ends here, recorded.
3. Mapping is pure code with no database.
4. A short read asks the key-state table which keys can be skipped.
5. The rest are cut into batches of up to 25 matches, sorted by our id. Each batch is a command keyed run id:batch number.
6. A batch that says "try again" is retried on its own. A batch that still fails ends the run as partly_done, with an alert. The next attempt of the same run starts from the first unsaved batch; saved batches come back as duplicate.
7. The last short transaction marks missing keys and closes the run.

Missing rows Agreed, to build

Agreed design, not in the code yet. After the last batch, in its own short transaction:

UPDATE integration_key_state SET missing_runs = missing_runs + 1
 WHERE integration_id = $integration AND last_seen_run <> $run
   AND covered_by_run(subject_key, $run);      -- today's day x discipline box (retire.py), kept
-- keys with missing_runs >= 2 and no open item -> insert a "missing" review item (integration_answer)

• Only full runs count. A day build or a page run never marks anything missing.
• Only runs that passed the size check count (a full run must list at least half its usual keys).
• A key seen again resets to 0 (the batch upsert sets missing_runs = 0).
• Cancelling is a person's command, results.set_status CANCELLED. An integration with auto_cancel_missing on sends the same command itself. It is off by default.

The settle window Agreed, to build

Agreed design, not in the code yet. When a row first says official, next_read_at is set to now plus 1 hour. Each read moves it on:

The follow-up step picks keys with next_read_at in the past, oldest first, at most 250 per Scout run. That limit comes from Scout's 900 s run limit (PAGES_PER_RUN = 250, followups.py:106-107, built today).

Today a result page is read again only within lookahead_days (3 by default, integrations/settings.py:46) of the unit's start, and at most every 6 hours (PAGE_AGAIN_HOURS, followups.py:110). A protest decided later than that was missed (F17).

Settings Agreed, to build

Agreed design. Every number is a setting, so a new event or provider changes settings, not code.

The I12 guard limits (more than 10 records cleared, more than 5 finished matches reopened) are also settings per integration. The design doc does not give them setting names.

Files that change Agreed, to build

Migration order Agreed, to build

1. Fix two bugs: _retire_gone gets a savepoint; skipped unnamed units stop being reported as accepted.
2. Fetch outside the transaction, with retries; keep Scout's run status.
3. Record the run in its own transaction first (I2).
4. Create integration_key_state, filled once from today's integration_row (the newest row per key), then kept up to date by the writers.
5. Add changed_seq on fixture and fixture_competitor (Section 4, C3); the fast path turns on.
6. Batches through road.run with the import pool (Section 4, L7).
7. Missing rows become review items; the direct cancel is removed.
8. The settle window on the key; followups.py reads due keys.
9. Row log trimming and the 30-day delete.
10. The console: two-line health, the run screen, Waiting for you.

Screens

An operator should answer three questions in five seconds: is each source working, what did the last run change, and is anything waiting for me. Today's console answers the first only partly.

Today Built today, in omnium-console:

• The Integrations page shows one card per integration with a health word: On time, Late, Very late, Failing, Paused, Stopped by a check (integrationHealth, omnium-console/src/lib/api.ts:461-491). "On time" means only that omnium ran recently: a Scout run that failed its own checks still shows On time.
• The Runs page shows each run, its row count, time taken and its log. It says "passed with 1,240 rows", but not what changed in our data, what was kept as set by hand, or what was held.
• Held rows and guard pauses have no single place to act on them.

Agreed Agreed, to build (U1 to U3; built in Section 7):

Every screen handles: loading, empty ("no runs yet"), failed to read (with Try again), running (bar moving), partly done, paused by a check (with the reason and Start), and off. Nothing shows green unless both Scout and omnium are fine.

When things go wrong

Every case ends with no good data overwritten, nothing saved twice, and a person told when they need to act. This is the agreed design.

Decisions

All 24 decisions were agreed on 6 Oct 2026. I12 and I13 were added that day after the user's comments.

Built today, or still to build

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

Numbers

Read next

• Truth and ownership: the Section 2 priority list that decides who wins when an import and a person disagree.
• Commands: the one road every batch takes, with match locks and time limits.
• Monitoring: the central alert service that import alerts go through.
• Database: the tables, the pools and the limits an import runs under.