Ingest sinks: one HTTP intake per table

Status: enabled on the service since the fourth pull request of the series (research/24a §4); AL_SINKS_ENABLED=false turns it off (every /v1/sinks route answers 404 not_found, the roller idles). Designed in research/24-ingest-sink.md, sized in research/24-ingest-sink-estimate.md; the founder's defaults (2026-09-21): the ingest host is the API host, consumed batches are deleted at commit, the roller is its own service. The integration check is Phase 10 of platform/scripts/test.sh (CI's stack job and the deploy gate run it).

A device, a webhook, a cron job or an agent with no query engine posts JSON arrays to one URL with a send-only key. The platform stores each batch unchanged, appends what is waiting to the Iceberg table on a rolling policy, stamps when each row arrived, and puts the rows that do not fit the table's schema where the producer can read them. No transforms, no schema evolution, no ordering across batches, no user code: the table's current schema is the contract.

tablemere table create garden --namespace garden --name events --column x:long:required --column who:string
tablemere sink create garden --table garden.events --name events_in --save ~/events-sink.json
tablemere sink send events_in batch.json --send-key-file ~/events-sink.json     # or any HTTP client, below
tablemere sink get events_in                                                    # lag, last roll, rejects

The same with curl, the way a device does it:

curl -X POST https://api.tablemere.com/v1/sinks/<sink_id> \
     -H "Authorization: Bearer al_send_…" -H "Idempotency-Key: 2026-09-21T12:00:00Z-0001" \
     -H "Content-Type: application/json" \
     -d '[{"x": 1, "who": "north"}, {"x": 2}]'
-> 202 {"accepted": 2, "rejected": 0, "reasons": [], "batch_id": "…", "bytes": 33, "received_at": "…", "state": "pending"}

What a sink is

One row per table (409 sink_exists for a second): the table's single committer. The table must exist and be unpartitioned (404 table_not_found with the namespace's tables, 409 partitioned_table_unsupported; our maintenance worker does not compact partitioned tables, so the sink does not create one). Creating a sink costs no IAM write and no commit: the answer says iam_writes: [], and the store's identity count does not move (every other create route on the platform costs identity writes; research/25).

The roll policy, per sink: roll_seconds (default 300, floor 60), roll_bytes (32–64 MiB, default 64), inactivity_seconds (optional: a producer that posts a burst and goes quiet gets its rows committed early), and a fixed count of 1,000 pending batches. Any of them rolls. A roll is a commit, and a commit costs more the more snapshots the table holds (research/24 §4 "Interplay"): 300–900 s suits a producer under about a megabyte per five minutes; the hourly maintenance keeps the table at 20 snapshots and compacts the small files.

The first roll adds an optional __ingest_ts (timestamptz) column and sets the metadata-retention properties (write.metadata.delete-after-commit.enabled=true, write.metadata.previous-versions-max=5), one commit; a table that has both is untouched. Nothing is partitioned in v1; the owner partitions with their engine if they want to.

The send key

POST /v1/sinks answers the key once (al_send_<12 hex>_<43 chars>); the row keeps its SHA-256. It is not a principal: it can never become a token, hold a grant, or reach /v1/connection. It is accepted on exactly one route, POST /v1/sinks/{sink_id}; anywhere else it is 403 send_only, and an API key or a token on the send route is 403 send_only too. Revocation is the sink's deletion (the row is the key). The CLI treats it like the uploader's secret: --save FILE (mode 0600, nothing printed) or --show-key (printed once); without either flag sink create stops before calling the API.

The send route

POST /v1/sinks/{sink_id}
  Authorization: Bearer <send key>
  Idempotency-Key: <batch id>        required; 1–128 chars of A-Z a-z 0-9 . _ : -
  Content-Type: application/json
  [ {…}, {…}, … ]                    a JSON array of objects, <= 16 MiB

Each element's bytes are stored unchanged, one per line, as one NDJSON object under _sink/<sink_id>/<batch_id>.ndjson in the catalog's blob bucket, written as the catalog's own identity in one small PUT (no IAM write, no catalog call). The object carries Content-Type: application/x-ndjson, x-amz-meta-received (the intake's clock; what __ingest_ts is set from), x-amz-meta-rows, x-amz-meta-sink.

The roll

The roller (worker/roll.py, the sink-roller service) ticks every 10 s and, for every sink whose policy says so, in order: reloads the table as the tenant; re-establishes the invariant (any batch the current snapshot's summary already names is deleted from intake before anything new is committed; a delete that fails ends the roll, and lag grows rather than a row being appended twice); does the first-roll changes once; reads the pending batches (at most 1,000); validates row by row against the table's current schema; writes the rejects files; writes one Parquet file with the Iceberg field ids and commits one append whose snapshot summary carries tablemere.sink.id, tablemere.sink.batches (the consumed ids), tablemere.sink.rows, tablemere.sink.rejected_rows, tablemere.sink.rolled_at; deletes the consumed batches; reports; sweeps rejects older than 7 days. A lost race against the compactor or an engine re-commits the same file against the new base (up to 5 attempts, 0.5–8 s backoff); an empty roll (every row rejected) commits nothing.

Exactly-once is recovery, not coordination: pending = intake minus the ids in the last summary, recomputed from the store on every roll. A roller that dies at any step restarts and finds its batches either in the summary (consumed; deleted) or in intake (pending; rolled). Duplicate detection at the intake is bounded: ledger rows are pruned 7 days after commit, so a batch id replayed later than that is accepted again.

Rejects

A row is rejected, never a batch, with a reason: missing_required (field), unknown_field (field: dropping a field silently is the one failure a producer cannot see), type_mismatch (field, expected, got), unsupported_type (a geometry or geography column: v1 does not write them), not_an_object. A batch the roller cannot parse at all goes to rejects whole as batch_unreadable and the sink goes on. The files are _sink/<sink_id>/rejects/<batch_id>.ndjson, one JSON object per rejected row with the original row and the batch id; the catalog's read recipe reads them (read_json in DuckDB, s3://<blob bucket>/_sink/<sink_id>/rejects/*.ndjson). GET /v1/sinks/{id} shows rejects {rows_total, batches_with_rejects, last_at, prefix}. Values: timestamps as RFC 3339 strings, dates as YYYY-MM-DD, decimals as strings or numbers, binary as base64, nested structs, lists and maps as JSON.

What it costs and where it is counted

Pending intake is transit, not storage: both storage meters (the hourly sweep and POST /v1/warehouses/{id}/measure) skip _sink/, and the bytes accepted at the intake are counted as ingest_bytes per project in GET /v1/usage (month_to_date, batches, rows_accepted, rows_rejected, by_sink; enforced: false). The same rows are metered once more as Parquet once committed, where they are stored. The store's hard bucket quota counts every byte in the blob bucket, intake included: an organisation at the edge of its allowance is stopped by the bucket going read-only before the meter would say so (409 quota_exceeded).

Routes and commands

POST   /v1/sinks                  {name, table: "<ns>.<table>", warehouse, roll_seconds?, roll_bytes?, inactivity_seconds?}   write level
GET    /v1/sinks[?warehouse=]     read level; ?include_deleted=true
GET    /v1/sinks/{id|name}        read level: state, roll, lag{batches_waiting, bytes_waiting, oldest_waiting_at, seconds_behind},
                                  received{…}, last_roll{at, batches, rows, rejected_rows, snapshot_id, seconds}, rejects{…}
DELETE /v1/sinks/{id|name}        write level; 202 {pending_batches_discarded}; the table is untouched
POST   /v1/sinks/{id}             the send key only (above)
GET    /internal/sinks            the roller's view (bootstrap secret; never served publicly)
POST   /internal/sinks/{id}/roll  the roller's report

CLI: tablemere sink create <catalog> --table <ns>.<table> --name <name> [--roll-seconds N] [--roll-bytes N] [--inactivity-seconds N] (--save FILE | --show-key), sink list [<catalog>] [--all], sink get <sink>, sink delete <sink> [--discard-pending], sink send [<sink>] <file|-> (--send-key-file FILE | --send-key K | TABLEMERE_SEND_KEY) [--batch-id ID]. tablemere connect mentions sinks in its notes (sinks_note, ingest_url in GET /v1/connection) once the flag is on. Audit events: sink.create, sink.delete, sink.roll.

Measured (2026-09-21, the local stack: weed mini 4.47-tm.2, the control plane of PR 1 + the roller of PR 2)

Recorded in platform/STATE.md (the two "ingest sink" entries):

Two facts the run taught, both in the code now: PyIceberg's S3 filesystem uploads every file in parts and the store refuses UploadPart to the tenant's static identity under the bucket-policy data grant, so the roll's data file goes through one signed PUT and the table's own metadata files go through vended credentials, exactly as every engine's do; and the intake's boto3 client sends a signed payload (request_checksum_calculation="when_required") because the store refuses the streaming-unsigned-trailer shape under that grant (STATE 2026-09-21 16:40Z).

Not in v1

Creating the table from the sink request; a rotate-key route (delete and create rotates); NDJSON request bodies; MCP tools (after the REST and CLI have run for a week); a partition at the first roll; a per-table max_snapshots override for sinks at the 60 s floor; the sidecar id list for sinks needing more than 1,000 batches per roll. All named with their reasons in research/24 §6 and research/24a §2.