Ingest sinks: one HTTP intake per table

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

There are no transforms, no schema evolution, no ordering across batches and no user code. The table's current schema is the contract.

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

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

curl -X POST https://api.lakehousebox.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

A sink is the table's single committer, so a table has one sink at most (409 sink_exists for a second).

• The table must exist. Otherwise the answer is 404 table_not_found, with the namespace's tables.
• Partitioning. A format-version 2 table must be unpartitioned. A format-version 3 table may be partitioned by time: day, hour, month or year of a date or timestamp column. Any other partition is refused (409 partitioned_table_unsupported). A roll writes one file per partition, so an identity or bucket partition would cut every roll into many small files.
• The sink never partitions a table itself. Create it partitioned (lhbox table create garden.garden.events --format-version 3 --partition 'day(ts)' --column ts:timestamptz:required), or partition it with your engine.
• Geography is refused. A geography column, or a geometry column with a CRS of its own, answers 409 unsupported_column_type. A sink writes longitude/latitude geometries ("Positions and shapes" below).
• Creating a sink costs no IAM write and no commit. The answer says iam_writes: []. The store's identity count does not move (every other create route on the platform costs identity writes).
• Describe the table. When the table or some of its columns have no description, the answer says so in docs_hint, with the command that adds them (lhbox table document). Other people's agents will read this table, and an undescribed one leaves them guessing what a row is.

The send key

POST /v1/sinks answers the key once: al_send_<12 hex>_<43 chars>. The sink 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 answers 403 send_only.
• An API key or a token on the send route is 403 send_only too.
• To revoke it, delete the sink: the sink is the key.
• The CLI treats it like an uploader's secret. --save FILE writes it (mode 0600, nothing printed), and --show-key prints it 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. It lands under _sink/<sink_id>/<batch_id>.ndjson in the catalog's file bucket. It is written as the catalog's own identity, in one small PUT, with no IAM write and no catalog call. The object carries:

• Content-Type: application/x-ndjson;
• x-amz-meta-received, the intake's clock, which __ingest_ts is set from;
• x-amz-meta-rows and x-amz-meta-sink.

The roll

The roll policy is set per sink. Whichever limit is reached first rolls:

• 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;
• a fixed count of 1,000 pending batches.

A roll is a commit, and a commit costs more the more snapshots the table holds. A roll every 300–900 s suits a producer sending under about a megabyte per five minutes. The maintenance sweep runs every 6 hours: it expires the table back to 20 snapshots and compacts the small files.

The first roll makes one commit. It adds an optional __ingest_ts (timestamptz) column and sets the metadata-retention properties: write.metadata.delete-after-commit.enabled=true and write.metadata.previous-versions-max=5. A table that already has both is left untouched.

Each roll. The roller ticks every 10 s. For every sink whose policy says so, it runs these steps in order:

1. It reloads the table as the tenant.
2. It re-establishes the invariant. Any batch the current snapshot's summary already names is deleted from the intake before anything new is committed. A delete that fails ends the roll: lag grows rather than a row being appended twice.
3. It makes the first-roll changes, once.
4. It reads the pending batches, at most 1,000.
5. It validates them row by row against the table's current schema, and writes the rejects files.
6. It writes one Parquet file with the Iceberg field ids, and commits one append. The snapshot summary carries lakehousebox.sink.id, lakehousebox.sink.batches (the consumed ids), lakehousebox.sink.rows, lakehousebox.sink.rejected_rows and lakehousebox.sink.rolled_at. An older roll carries the same keys under an earlier prefix, and they still count.
7. It deletes the consumed batches, reports, and sweeps rejects older than 7 days.

A roll that loses a race against the compactor or an engine re-commits the same file against the new base. It tries up to 5 times, with 0.5–8 s of backoff. An empty roll, where every row was rejected, commits nothing.

Exactly once

Exactly-once comes from recovery, not coordination. A batch in the intake is already consumed when two things hold:

• a retained snapshot of this sink (lakehousebox.sink.id) names its id;
• the object was stored before that snapshot, with 1 h of tolerance for clock skew.

Every roll recomputes this from the table and the store. Every retained snapshot counts, not only the latest. So a commit by someone else, such as the tenant's own write or a compaction, cannot hide an earlier roll.

• A roller that dies at any step restarts and finds each of its batches in one of two places. Named in the history, a batch is consumed: it is deleted, and reported so the ledger marks it committed. Still in the intake, it is pending, and it is rolled.
• If another roller committed some of a roll's batches since it read them, the roll commits nothing. The next tick sorts it out.
• Past the retained history (20 snapshots or 168 h), a roll is still known. Right after its commit it writes _sink/<id>/committed/<its data file>.json, naming its batch ids. It deletes that marker after the intake.
• A roll that dies between the two leaves the marker. Step 2 reads it like the history, and the newer record wins. So a roller that is down for a week does not roll its leftovers again.
• A marker that cannot be read ends the tick. Lag grows; nothing is rolled twice.
• A repair is reported again for as long as the ledger still holds its batches pending, and its marker is kept until then. A marker left by a failed delete goes on the sink's next roll.
• Until the repair is reported, those batches are pending in the ledger. The sink_lag alert fires at 3 x roll_seconds.

Duplicate detection at the intake has a bound. Ledger rows are pruned 7 days after commit, so a batch id replayed later than that is accepted again.

A key: one row per key, not one per send

By default a roll appends. A producer that sends the same row twice (a republished feed, a retry with a new batch id) stores it twice. A sink with a key has each roll replace the rows that share the key instead. Set the key at creation or by PATCH (key: ["observation_id"]):

lhbox sink create garden --table garden.observations --name observations_in --key observation_id --save ~/obs-sink.json
lhbox sink update observations_in --key observation_id   # an existing sink

• The latest wins. A keyed sink rolls its batches in the order they arrived. Inside a roll, the row received last wins; among the rows of one batch, the last one. A row already in the table with the same key is replaced.
• A row with a null in a key column is a row reject (key_null).
• One commit per roll, as for an append. The snapshot drops the files whose rows are all replaced, rewrites those partly replaced, and adds the roll's rows. It names the batches in the same lakehousebox.sink.* summary, so exactly-once by recovery is unchanged.
• Cost. Each roll reads the key columns of the table's files, except those whose bounds rule the keys out (string and integer keys). It rewrites only the files that hold a replaced row.
  • A key that follows arrival time (a feed republishing today's rows) rewrites nothing: the last run's file is dropped whole.
  • A key spread over every file would rewrite the table on every roll.
  • Past 256 MiB of rewritten files in one roll, nothing is committed and the sink is disabled (upsert_rewrite_too_large).
• What a key needs: a format-version 2, unpartitioned table, and columns of a key type. The key types are string, integer, date, time, timestamp, uuid, decimal and boolean, never a float.
• A key the roller cannot apply disables the sink with an upsert_* reason and leaves the batches waiting. Changing the key makes the sink active again: PATCH {key}, or null to append again (lhbox sink update <sink> --no-key).
• Rows stored before the key was set keep their duplicates, except where a later roll carries the same key again.

To clean those up, lhbox sink dedupe <sink> (POST /v1/sinks/{id}/dedupe) asks the roller to keep only the latest row per key in the whole table, once.

• The latest is by __ingest_ts: the row received last wins, and a tie goes to the later commit.
• It is one commit, at the next tick. It is refused past 1 GiB of rewritten files or 10 million rows.
• GET /v1/sinks/{id} shows dedupe.last when it is done, or why not. --wait waits for it.
• A request already waiting is answered as it is. After a dedupe, the next one waits an hour.

Rejects

A row is rejected, never a batch, with one of these reasons:

Three cases reject a whole batch, and the sink goes on:

• batch_unwritable, on a format-version 3 table: the writer refuses the batch's rows even after these checks. The rejects keep the writer's error. The roll's other batches are committed one by one first, so one batch never holds the others back.
• batch_unreadable: the roller cannot parse the batch at all.
• batch_too_large: an intake object larger than its writer may send, unread. The limit is 16 MiB for a producer's batch and 256 MiB for a scheduled SQL run's Parquet.

The rejects files are _sink/<sink_id>/rejects/<batch_id>.ndjson. Each holds 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, on s3://<blob bucket>/_sink/<sink_id>/rejects/*.ndjson. GET /v1/sinks/{id} shows rejects {rows_total, batches_with_rejects, last_at, prefix}.

How to send values:

• timestamps as RFC 3339 strings, and dates as YYYY-MM-DD;
• decimals as strings or numbers, and binary as base64;
• nested structs, lists and maps as JSON.

Sink geometry, a page of its own: Positions and shapes sent to an ingest sink: point and shape mappings, the geometry column, its format version and how the roller writes it.

Files to rows, a page of its own: An index rule turns the files dropped in a catalog's file bucket into rows of a table: the rule, the columns it writes, how often it runs.

What it costs and where it is counted

Pending intake is metered twice:

• The storage figure reads the store's own per-bucket figure, every few minutes, or at once with POST /v1/warehouses/{id}/measure. It counts every byte in the file bucket, _sink/ included.
• ingest_bytes counts the bytes accepted at the intake, per project, in GET /v1/usage: month_to_date, batches, rows_accepted, rows_rejected and 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 also counts every byte in the file 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).

Sink devices, a page of its own: A device that cannot set headers posts to a sink's device route; a mapping turns its payload into the table's columns, with presets for common devices.

Scheduled SQL: a sink that runs a query

LakehouseBox itself can feed a sink: one DuckDB SELECT over http(s) URLs, run on a schedule, its rows committed to the table. It has its own page: Scheduled SQL.

Routes and commands

POST   /v1/sinks                  {name, table: "<ns>.<table>", warehouse, roll_seconds?, roll_bytes?, inactivity_seconds?, key?}   write level
PATCH  /v1/sinks/{id|name}        {device_require?, device_strip?, mapping?, key?}   write level; key null appends again
POST   /v1/sinks/{id|name}/dedupe write level; a keyed, active sink; 202 {dedupe{requested_at, last}}
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

The sink commands are create, update, dedupe, list, get, delete and send. Every flag is in the CLI reference. A send takes its key from --send-key-file FILE, --send-key K or LHBOX_SEND_KEY, and its batch id from --batch-id ID.

lhbox connect mentions sinks in its notes: sinks_note, and ingest_url in GET /v1/connection. The audit log records sink.create, sink.delete and sink.roll.

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: create the table partitioned instead.
• 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.

Numbers and dates

Status: enabled on the service.

Measured (2026-09-21, the local stack, the control plane of PR 1 + the roller of PR 2)

The run taught two facts, 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. The table's own metadata files go through vended credentials, exactly as every engine's do.
• The intake's boto3 client sends a signed payload (request_checksum_calculation="when_required"). The store refuses the streaming-unsigned-trailer shape under that grant.