Ingest Pipeline

Ingest Pipeline

What happens between an agent's POST /api/v1/events and the rows in PostgreSQL. This is the only write path a device has.

Payload shape

{
  "metadata": {
    "deviceId": "11111111-2222-3333-4444-555555555555",
    "serialNumber": "WS-0001",
    "collectedAt": "2026-01-15T08:54:40+00:00",
    "clientVersion": "2026.01.09.1420",
    "platform": "Windows",
    "collectionType": "Full",
    "enabledModules": ["system", "hardware", "inventory"]
  },
  "events": [
    { "eventType": "success", "message": "Managed install run completed", "details": {} }
  ],
  "system": {},
  "hardware": {},
  "inventory": {}
}

Modules may sit at the top level as above, or nested under a modules key. Both camelCase and snake_case metadata keys are accepted, because the two agents differ. The server does not try to be clever about module contents — the client decides what to send, and the payload is stored verbatim.

Stage 1: read the body

The handler reads raw bytes rather than calling request.json(), because that collapses "the client hung up mid-upload" and "the client sent garbage" into one exception, and those have opposite fixes.

Situation	Reason code
Client disconnected before the body arrived	upload_aborted
Body could not be read at all	body_unreadable
Body was empty	empty_body

Every rejection detail carries a transport note — declared Content-Length, bytes actually received, Content-Encoding, decompressed size, and the exception class — so a truncated upload can be told from a client serializing garbage at a glance.

Stage 2: decompress

Content-Encoding of gzip, deflate, or identity is honoured; a full collection is a megabyte or more of repetitive JSON and compresses about eightfold. Anything else is body_unreadable.

The inflate is bounded during decompression against MAX_DECOMPRESSED_BODY_BYTES (64 MiB by default), not after — an unbounded one-shot decompress would exhaust the container before any size check could run. A stream that never reaches its end marker is a truncated upload and is rejected as such.

Stage 3: parse and sanitize

malformed_json when the body will not parse.

PostgreSQL jsonb cannot represent a NUL code point anywhere in a string. The raw body is cheaply prechecked for the six-character escape json.dumps always emits for one; when present, every string in the decoded payload is stripped and a nul_in_payload row is recorded at status 200. The check-in is accepted and the data kept — but a client shipping unstorable characters is a defect that has to stay visible rather than be silently absorbed.

Stage 4: validate structure

Pydantic validates the top level: metadata.deviceId and metadata.serialNumber must be present and non-empty; platform must be Windows, macOS, Linux, or Unknown; collectionType must be Full or Single. Failure is invalid_payload at HTTP 422.

Stage 5: validate the serial number

The serial number is the device's identity for the life of the record, so a bad one pollutes the database permanently. Five checks, in order:

Check	Reason code
Empty or whitespace-only	empty_serial
A known sentinel — -1, 0, unknown, none, null, n/a, (empty), default string, system serial number, to be filled by o.e.m., 00000000 and friends	sentinel_serial
Pure-numeric and shorter than four characters	short_serial
Matches a machine-generated Windows default hostname pattern (DESKTOP-*, LAPTOP-*, WIN-*, each with its fixed-length random suffix)	hostname_serial
Equals the hostname the payload itself reports	serial_equals_hostname

Sentinel matching squashes non-alphanumerics first, so the spaced OEM strings also catch the run-together forms real BIOSes report. The last check is the one that catches the real failure — a client sending its hostname where the hardware serial belongs — which no shape heuristic can do without also rejecting genuine serials.

Stage 6: idempotency

If the request carries an Idempotency-Key, it is inserted into idempotency_keys with ON CONFLICT DO NOTHING RETURNING key. Losing the conflict means this is a replay: the transaction is rolled back and the client gets {"status": "duplicate", "idempotent": true, ...} without reprocessing.

The key row is committed together with the device upsert, which is the first commit of the request. A submission that fails before that point rolls its key back and the retry processes normally. Module writes commit one by one after it, so a submission that fails later keeps its key, and a retry carrying the same key is acknowledged as a duplicate.

Stage 7: upsert the device

Existing devices get device_id, last_seen, updated_at, client_version, and platform refreshed. New devices are inserted with id and serial_number both set to the serial, status online, and a name taken from metadata.additional.deviceName if the client supplied one, otherwise Unknown.

Stage 8: write the modules

For each of the ten module names present in the payload — system, hardware, network, installs, security, applications, inventory, management, peripherals, identity — the row for that device is written to the table of the same name.

The write takes an unchanged-payload fast path. Most check-ins resend identical module data, and rewriting the JSONB every time re-TOASTs multi-MB values on every ingest. So the first statement updates only the timestamp columns WHERE device_id = %s AND data = %s::jsonb; jsonb equality is content-based, so a match means nothing changed and the TOAST data is reused. Only when that matches no row is the payload actually rewritten.

Each module commits independently. A module that fails to store is logged and skipped — the rest of the check-in still lands.

Two modules have side effects:

• installs — classified before it is stored. See the next section.
• system — devices.os_name, os_version, and os are updated from operatingSystem.

Classifying managed-software items

install_status.py holds the one definition of an install item's state, and it applies identically to Munki (macOS) and Cimian (Windows). Ingest runs it before the installs module is written, stores the answer on each item as reportmateStatus, and, when the data changed, writes the four dashboard counters computed from the same answer in the same transaction. Nothing downstream re-derives the state.

First, transient network failures are removed. A Mac that sleeps or leaves the network mid-run cannot fetch its catalog or manifest and reports an error that clears on the next run. Run-level catalog and manifest fetch failures, and the Error during updatecheck: line that repeats their reason, are suppressed when they carry one of the six transient NSURLError codes (-1001, -1003, -1004, -1005, -1006, -1009) or their localized descriptions. A bare Could not download catalog line is suppressed only when no HTTP-error line for the same catalog accompanies it, so a catalog that is genuinely missing still reports. An item's own download failure stays with the item. What is removed is kept under transientProblems on the munki section and transient_items on the session, and the run's status is re-derived from what remains.

Then each item is placed on one ladder:

1. currentStatus / mappedStatus is the tool's verdict. Error, Warning, Installed, and Removed settle it.
2. Pending is the exception: an install is still owed, so lastError / lastWarning still count.
3. With no verdict, lastAttemptStatus and then lastError / lastWarning decide.

lastError and lastWarning count only when the run itself reported them — the item has lastSeenInSession — or when the payload has no sessions at all. A status of Install Loop is an error; the bare install-loop flag is a warning. Not Installed and Not Available are warnings. Pending is never counted as a warning. Anything the ladder cannot place is left unclassified.

Counting per platform: item-level errors and warnings win, and the run's errorItems / warningItems (or the legacy strings) stand in only when no item carried one. A package that the latest session's run events show failing, and that no item already reports a problem for, adds an error; Cimian's item list does not reliably mark a failed install, and the events feed already shows it.

When the classifier changes, POST /api/v1/admin/installs/reclassify applies it to rows already stored.

Stage 9: accumulate usage

If the applications module carries dailyUsageHistory, each entry is upserted into usage_history on (device_id, date, app_name).

A client whose dated build is older than its platform's usage floor (USAGE_MIN_CLIENT_VERSION, or USAGE_MIN_CLIENT_VERSION_WINDOWS for Windows) has its dailyUsageHistory dropped and recorded as usage_client_too_old at status 200. Those builds predate counting fixes, and their rows would be written into the table permanently. The rest of the check-in is stored. See Configuration.

Both clients send window deltas, not cumulative-today totals, so the upsert accumulates rather than replaces. The second POST of the day for the same device and application contains only what is new.

Per-entry validation drops what cannot be trusted:

Condition	Effect
Non-object entry, or missing date / appName	Skipped
Date before 2020-01-01, or more than two days in the future	Row rejected
Non-numeric or non-finite value	Row rejected
Negative value	Clamped to zero, noted

launches and total_seconds accumulate without a ceiling — total_seconds is summed process lifetime and multi-process applications legitimately exceed 24 hours of it in a day. active_seconds and foreground_seconds measure human attention, which one machine can supply for at most 86 400 seconds a day, so the accumulated value is held at that ceiling by LEAST(). The statement's RETURNING clause reports rows sitting at the ceiling so the anomaly surfaces rather than being silently absorbed.

Anything rejected, clamped, or capped produces a usage_out_of_bounds row at status 200 — the check-in is accepted, but the client defect stays visible. If the usage write fails outright, a usage_write_failed row is recorded at status 500 even though the response is still 200, because nothing was stored and the measurement is simply gone.

An HTTP retry that re-sends the same payload will double-count. The macOS client marks sessions transmitted on a 200 before deleting them, and the endpoint is rate limited, which is what keeps this bounded today.

Stage 10: resolve the device name

First non-empty wins, in this order: inventory.deviceName, hardware.system.computer_name or .hostname, network.hostname, system.hostname, metadata.additional.deviceName. The update only applies where the stored name is null, Unknown, or the serial number — a name already resolved is not overwritten.

Stage 11: store events

When the payload carries an installs module, it is a fresh managed-software run, and the previous run's events no longer describe the device. Before inserting, the device's events under installs, managedinstalls, or munkireport are deleted — along with older unnamed success / warning / error events, which only the managed-software path emits — as long as they are not newer than this payload's collectedAt. That happens even when the new run sent no events, which is exactly when a stale error would otherwise stay on the device.

Each entry in events then becomes a row with its own details only; module payloads live in the module tables. A success / warning / error event with no moduleId that arrives alongside an installs module is stored under installs, so the next run supersedes it. A run's error or warning event is dropped when the only problems it announced were suppressed as transient. A new os_update event replaces the device's previous one.

When the payload carries no events and no installs module, the server writes one fallback info event summarising the collection. When an installs module is present, no fallback is written — an installs run should always carry its own success/warning/error event, and its absence is logged as a client defect rather than papered over.

Each stored event is broadcast to Web PubSub if EVENTS_CONNECTION is configured. A broadcast failure is logged and ignored.

Immediately before the final commit, the device's row in device_ingest_state is stamped with the server's current time. That is what GET /api/v1/events/failures reads to decide whether a dropped upload was followed by one the server accepted.

Stage 12: retention

On roughly 1% of ingests a daemon thread purges events older than 30 days in batches of 50 000 and idempotency keys older than 7 days. It runs off the request path with its own connection, because an inline multi-second DELETE on this async endpoint would stall the event loop and the health probes.

Caches are deliberately not invalidated on ingest. See Architecture.

Failure recording

Every rejection above is written to ingest_failures before the HTTP error is raised. Failures never break the rejection path — every error in the recorder is swallowed after a log line.

Identity comes from the payload when it parsed, and from the mirrored request headers (X-Device-Serial, X-Device-Uuid, X-Device-Name, X-Platform, X-Client-Version) when it did not. That is precisely when a rejection would otherwise be unattributable, and an unattributable rejection is not actionable.

Three other sources write to the same table:

• Authentication failures. Ingest POSTs are always recorded. Other endpoints are recorded only when the caller presented a device-style credential, so scanner probes of GET endpoints do not flood the table.
• Rate limiting. A throttled ingest POST is recorded as throttle / rate_limited. Throttled dashboard traffic is not — it is not a device problem and would bury the devices.
• Server errors. A 5xx on the ingest path is a check-in the device made correctly and the server lost, so it is recorded as server_error / internal_error with the same reference id the caller received.

Repeats of the same (serial, reason, ip) within INGEST_FAILURES_DEDUP_MINUTES collapse into the first row. Read the results at GET /api/v1/events/failures — see API-Reference for the outcome classification that separates a device whose data never arrived from one whose retry landed.