Metadata-Version: 2.4
Name: tuladb-iceberg-exporter
Version: 0.0.1rc0
Summary: TulaDB committed cursor exporter to Parquet/Iceberg-ready layouts
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyarrow>=15.0.0
Provides-Extra: iceberg
Requires-Dist: pyiceberg[pyarrow,sql-sqlite]<0.12,>=0.11.1; extra == "iceberg"
Provides-Extra: s3tables
Requires-Dist: pyiceberg[glue,pyarrow]<0.12,>=0.11.1; extra == "s3tables"
Requires-Dist: botocore[crt]; extra == "s3tables"
Requires-Dist: s3fs>=2025.3.0; extra == "s3tables"
Provides-Extra: biglake
Requires-Dist: pyiceberg[pyarrow]<0.12,>=0.11.1; extra == "biglake"
Requires-Dist: google-auth; extra == "biglake"
Provides-Extra: duckdb
Requires-Dist: duckdb>=1.0.0; extra == "duckdb"
Provides-Extra: all
Requires-Dist: tuladb-iceberg-exporter[biglake,iceberg,s3tables]; extra == "all"
Dynamic: license-file

# tuladb-iceberg-exporter

Python exporter for mirroring committed TulaDB transaction records into an
Iceberg-ready Parquet lake.

TulaDB Iceberg exporter is a checkpointed, replay-safe, committed-log exporter
for reconciliation and audit workloads. It is downstream of TulaDB commit and
never participates in the ledger write path.

The exporter is designed to run outside `tuladbd` and keep export work away
from the write hot path:

- polls `MSG_EXPORT_CURSOR` from TulaDB's query/export listener (a read-only
  port; the exporter's principal needs the `exporter` role for its lease and acks);
- reads only committed WAL / Raft-applied records;
- forms bounded exporter micro-batches;
- stages committed transaction rows as deterministic Parquet files;
- commits those files through an atomic Iceberg append;
- advances the local durable checkpoint only after the Iceberg commit succeeds.

The current sink is local/object-store-style Parquet layout. In that local mode,
the checkpoint advances after a deterministic file write, and the file manifest
provides crash-restart recovery plus idempotency/overlap fail-closed so a
crash between write and checkpoint cannot duplicate committed transfers.
Production Iceberg mode uses the stricter contract: staged Parquet files become
visible through an atomic Iceberg append before the checkpoint moves forward.

The source connection must point at TulaDB's read-only `query_port`
(`7800` by default), not the `node_addr` write/client port (`7700` by default).

## Clusters: list every node, export from the leader

Give the exporter every node of the cluster; it exports from whichever one
currently leads and follows leadership changes on its own:

```toml
[source]
hosts = ["tuladb-1.internal:7800", "tuladb-2.internal:7800", "tuladb-3.internal:7800"]
# or bare hosts plus a shared port:
# hosts = ["tuladb-1.internal", "tuladb-2.internal", "tuladb-3.internal"]
# query_port = 7800
leader_check_interval_ms = 1000        # re-ask "do you still lead?" this often; 0 = strict
leader_discovery_max_backoff_ms = 5000 # cap on retries while no leader exists
leader_probe_timeout_ms = 2000         # leadership probes (discovery, sweeper)
leader_term_floor_grace_ms = 3000      # max wait on a higher term no majority confirms
```

`host = "..."` is still accepted as a one-node `hosts` list (set one or the
other, not both).

> **A single `host` must be the cluster's leader, or the exporter waits.** In
> leader mode (the default) a node that is not the leader is never read. An
> old config whose one `host` points at a **follower** -- or at a **load
> balancer / shared DNS name** in front of several nodes -- now stalls in
> leader discovery (or flaps between nodes) instead of quietly exporting from
> whatever answers. The exporter says which at startup: "the only configured
> address ... answered as node N, a follower", or "address ... answered as
> node A earlier and as node B now: it is a load balancer". Fix it by listing
> every node's own address in `hosts`. Every address must belong to ONE
> cluster: the wire carries no cluster id, so the exporter refuses only what
> it can prove -- a standalone node (peer_count 0) listed with any other node,
> or two nodes leading the same term.

How it works:

- **Finding the leader.** The exporter sends `MSG_CLUSTER_STATUS` (allowed on
  the query port) to every listed address. Each node answers with its own node
  id, the id it believes leads, its role and term -- ids, not addresses -- so
  the exporter reads from the address that answers "I lead" (role leader,
  naming itself; the highest term wins if a deposed leader still claims it).
  The list order and node ids do not have to line up with anything.
- **When it looks.** On start; after any connection loss, timeout or
  "must target the leader" refusal; and every `leader_check_interval_ms` on the
  node being read. That last check matters: the query port serves the export
  cursor from any node's applied state and never answers "not the leader", so a
  leader that was deposed but stayed up would otherwise keep being read. A
  background sweeper also asks the OTHER nodes for their term every five check
  intervals (at least every 5 s), so a leader cut off by a partition -- which
  still says "leader" about itself until its check-quorum step-down -- is left
  once another node reports a newer term. The sweeper never runs on the read
  path: its probes (and discovery's) use `leader_probe_timeout_ms` (default
  2000, capped at `timeout_s`), and an address that does not answer is backed
  off (doubling, up to 60 s), so a blackholed node does not slow the export.
  Between checks a just-deposed node may still serve pages: they are committed
  data and correct, but its acks no longer count toward purge.
  `leader_check_interval_ms = 0` is the strict mode: every page is confirmed
  on the same connection after it is served (one extra round trip per page)
  and dropped unless the node still leads the term it led before the request.
  What that proves is that the node *had not yet learned* it was deposed: a
  leader cut off by a partition passes until the sweeper sees the newer term
  or the node steps down on check-quorum. That window is harmless -- the pages
  are committed data, and its acks never reach the real leader, so they cannot
  cause a purge -- but it is not "no page from a node that had lost the
  leadership".
  During an election discovery backs off (100 ms doubling, capped). The
  highest term seen anywhere is a floor: once any node reports term T, a node
  still leading an older term is not read even though it claims "leader" --
  the exporter waits, with that same back-off and one log line per term, for
  the leader of T -- unless a strict majority of the cluster still answers in
  the older term naming that leader. Then no node can have won T (an election
  needs a majority voting in T, and terms never go back): T is a failed or
  isolated candidacy, such as a leadership-transfer target that skipped
  pre-vote, and the leader is read at once; the background sweep re-checks
  the majority every round. Without such a majority, if no node has led T
  after `leader_term_floor_grace_ms` (default 3000, counted from the first
  sighting; the same node bumping its term again does not restart it) and
  none reports anything newer (a cluster rebuilt with reset terms, or one
  node whose term ran ahead), the floor is dropped with an ERROR and the
  highest-term leader is read; a node whose own term goes backwards is
  treated as rebuilt at once. A node that
  is not the leader is never exported from in this mode; if the leader is not
  in `hosts`, the exporter waits and says so in its log.
- **Resume.** Every fetch names the exporter's own position, and Raft gives
  every node the same committed log with the same global batch sequence, so
  the new leader continues exactly where the old one stopped. The durable,
  monotonic checkpoint and the min-committed floor are unchanged: a node that
  has committed less than the exporter has already seen is refused (backed off)
  rather than read. The sink's dedup is position-based and does not care which
  node served a page.
- **Bootstrap stays put.** A full-state seed (`bootstrap`) starts on the
  leader and then stays on that NODE even if leadership moves: its cursor
  walks that node's account directory and means nothing elsewhere, and a seed
  needs no purge credit. A dropped connection (e.g. the server's idle close
  during a slow sink write) is re-opened to the same address and accepted only
  if it answers as the same node id with no sign of a restart. A restart is
  visible only as its process counters going backwards -- the wire carries no
  boot id -- so a restart between two seed pages can go unseen when those
  counters were zero both times; avoid restarting nodes during a bootstrap.
  Two cheap backstops catch most of what that misses: a seed that repeats an
  account (same ledger, currency and id) is refused before the repeat is
  written, and a seed with fewer accounts than the source reported on its
  first page is logged at WARNING. If
  the node is lost before any seed row is written the scan restarts from
  scratch on the current leader. After rows are written the bootstrap stops
  without a checkpoint and leaves a durable unfinished-seed marker beside it
  (written before the first row), so a later `bootstrap` -- including after a
  crash -- refuses instead of seeding a second set beside the partial one; re-run
  into a fresh metadata table, or with `--force` to accept the partial rows.
- **Same principal, same role, on every node.** Every connection the exporter
  opens -- the first discovery, every rediscovery after a failover, a bootstrap
  reconnect -- authenticates afresh with the configured `[source.auth]`
  credential and gets the exporter-role check, so leases and durable acks reach
  a new leader under the same principal. Grant that principal the `exporter`
  role on EVERY node (`role_grant_N` is per-node config): a node that refuses it
  fails the export with the role error rather than being skipped.
- **Same principal on every INSTANCE, too (HA).** The source binds a pipeline's
  lease key to the first principal that pages with it, cluster-wide, and
  refuses any other -- an anonymous caller included -- until that binding
  lapses, 2 x `export_lease_ttl_ms` after the owner's last page. Every instance
  of one pipeline (both halves of an active/passive pair, a replacement pod)
  must therefore authenticate as the SAME principal (same client certificate
  subject or credential); with different ones, the instance that takes over is
  refused, exports nothing and holds no WAL for up to 2 x TTL. The exporter
  reports that refusal as such (not as a missing role), backs off and retries
  until the binding lapses (`export_lease_key_bound` gauge). A lease registered
  anonymously (`[source.auth] type = "none"`, no client certificate) is never
  bound, so any exporter-role principal can claim its key: the exporter WARNs
  at start (`export_lease_anonymous` gauge). Set `source.export_lease_ttl_ms`
  to the source's value; it paces the lease keepalive and the back-off.
- **Why the leader.** History retention with `history_purge_requires_export`
  purges only below what every export pipeline has durably acked -- and only
  acks received by the leader, in its current term, count. After a failover the
  new leader purges nothing past the coverage it inherited until this exporter
  acks it, which it does on its first page there.

`allow_follower_reads = true` reads from a follower instead (offloading the
leader; data is committed but may lag). Its acks then land on the follower and
never reach the purge gate, so with export-gated history retention the source
keeps history for this pipeline indefinitely -- the safe direction, but purge
stops. Leader mode is the default.

Metrics: `source_node_id`, `source_node_is_leader`, `source_leader_term`,
`source_leader_switches_total`, `source_leader_lost_total`,
`source_leader_discovery_failures_total`.

## Source Security

> For the full security envelope across all three legs — source TLS/mTLS,
> catalog/S3 transport, and **encryption at rest** (what is native vs. delegated
> to the store vs. deferred) — see [docs/security-posture.md](docs/security-posture.md).

The exporter supports the same native security envelope as TulaDB's raw query
listener:

- TLS server verification with a CA bundle and optional `server_name` override;
- mTLS client certificates for certificate-principal grants;
- `MSG_AUTH` API keys and OIDC/JWT bearer tokens;
- fail-closed token auth over plaintext by default.

Grant the exporter's TulaDB principal the `exporter` role, in the `tuladbd`
config:

```toml
role_grant_1 = "iceberg-exporter -> exporter"   # N = the next index after your existing grants
```

Grant indices must be contiguous from 0 (tuladbd refuses gaps), so use the
index after your last existing grant. The principal is the mTLS client
certificate's principal (CN, or a SPIFFE/DNS
SAN), or the principal an API key / bearer token resolves to. `exporter` sits
between `read-only` and `ledger-operator` and moves no money; it does not need
`ledger-operator` or `cluster-admin`. `read-only` is no longer enough: with
`source.register_export_lease = true` (the default) every page carries an export
lease key and a durable ack, and tuladbd refuses both below `exporter` with
`ERR_PERMISSION` (35) -- a lease pins the source's WAL reclaim floor and an ack
lets history retention purge, so neither is open to a reader. The exporter then
stops on the first page with `ExportPermissionDenied`, naming the principal and
the `role_grant` to add; it never retries and never falls back to lease-less
reads on its own. With `MSG_AUTH` it also warns at connect when the principal
reports a role below `exporter`. Lease-less reads (range repair, or
`register_export_lease = false`) still need only `read-only`, at the cost of the
source neither holding WAL for the pipeline nor purging history on its behalf.

A tuladbd from before the `exporter` role (core 8651fdca) does not gate leases
or acks, so a `read-only` principal -- API key, bearer, mTLS or anonymous --
keeps working against it; the connect-time check only warns. Upgrade the
exporter first, then add the grant in the same config change that upgrades
tuladbd (an older tuladbd rejects the role name). Never grant `ledger-operator`
for this: it moves money. If the grant cannot be added before tuladbd is
upgraded, set `register_export_lease = false` until it is.

Example mTLS source config:

```toml
[source]
hosts = ["tuladb-query-1.internal:7800", "tuladb-query-2.internal:7800",
         "tuladb-query-3.internal:7800"]

[source.tls]
enabled = true
verify_server = true
ca_file = "/etc/tuladb/ca.pem"
cert_file = "/etc/tuladb/exporter-client.pem"
key_file = "/etc/tuladb/exporter-client.key"
server_name = "tuladb-query.internal"

[source.auth]
type = "none"
```

With several `hosts`, `server_name` (when set) is the name every node's
certificate is verified against; leave it unset to verify each node against
its own host name from `hosts`.

Example API-key source config:

```toml
[source]
hosts = ["tuladb-query-1.internal:7800", "tuladb-query-2.internal:7800",
         "tuladb-query-3.internal:7800"]

[source.tls]
enabled = true
verify_server = true
ca_file = "/etc/tuladb/ca.pem"
server_name = "tuladb-query.internal"

[source.auth]
type = "api_key"
credential_file = "/etc/tuladb/exporter.api-key"
```

## Quick Start

```bash
python3.11 -m venv .venv
. .venv/bin/activate
pip install -e .

cp config/tuladb-iceberg.toml config/local.toml
python -m tuladb_iceberg_exporter doctor --config config/local.toml
python -m tuladb_iceberg_exporter once --config config/local.toml
```

Run continuously:

```bash
python -m tuladb_iceberg_exporter run --config config/local.toml
```

Run as a locally managed background service:

```bash
python -m tuladb_iceberg_exporter start --config config/local.toml
python -m tuladb_iceberg_exporter status --config config/local.toml
python -m tuladb_iceberg_exporter stop --config config/local.toml
```

`start` launches the same foreground `run` command in the background and writes
a PID file. `stop` sends `SIGTERM`; the exporter exits after the current safe
point, so a page is either fully committed and checkpointed or not advanced.

For production service managers, run the foreground command and let the manager
own restart policy:

```bash
tuladb-iceberg-exporter run \
  --config /etc/tuladb/iceberg-exporter.toml \
  --pid-file /run/tuladb/iceberg-exporter.pid
```

A hardened systemd template is provided at
`deploy/systemd/tuladb-iceberg-exporter.service`.
The metrics sidecar template is provided at
`deploy/systemd/tuladb-iceberg-exporter-metrics.service`.
Plan-only compactor units are provided at
`deploy/systemd/tuladb-iceberg-compactor.service` and
`deploy/systemd/tuladb-iceberg-compactor.timer`; add `--execute` only after the
table's compaction policy is enabled and tested.
Plan-only object-store cleanup units are provided at
`deploy/systemd/tuladb-iceberg-cleanup.service` and
`deploy/systemd/tuladb-iceberg-cleanup.timer`; add `--execute` only after the
snapshot retention policy is enabled and reviewed.

Expose Prometheus/OpenMetrics metrics:

```bash
tuladb-iceberg-exporter metrics --config config/local.toml
curl http://127.0.0.1:9146/healthz
curl http://127.0.0.1:9146/metrics
```

Render once for CI/evidence:

```bash
tuladb-iceberg-exporter metrics --config config/local.toml --once
```

Run the full local quality gate:

```bash
make quality-gate
```

The gate runs syntax checks, unit tests, config/CLI smoke checks, dependency
checks, and release package artifact validation. GitHub Actions runs the same
gate on supported Python versions.

Run reconciliation:

```bash
tuladb-iceberg-exporter reconcile --config config/local.toml --json
```

Reconciliation compares committed `export_batches`, retained Iceberg export
snapshot metadata, local checkpoint state, and actual Iceberg table row
counts/sums.

It checks every transactions table the layout writes: the one table under
`shared`, and each `<table>_<type>` (by_type) or `<table>_<mode>` (by_mode)
table against its own `export_batches_<table>.jsonl` control file. An unset
`sink.transactions_layout` is resolved as the writer resolves it (the layout
the checkpoint recorded, else the by_type default). Errors name the table they
are about, and `--json` adds `transactions_layout` and a per-table `tables`
object beside the existing totals. A fan-out table that was never created (no
rejections yet, say) is skipped; one missing while its control file holds
commits is a failure. Totals count source records: a wide (tiered) table's
row is two merged legs and counts twice.

The table check streams: it reads only the `amount` column, one data file at a
time, so its memory does not grow with the table's row count (it used to load
the whole table into one Arrow table and was OOM-killed on a 30M-row worker
table). What it still holds is per commit, not per row: the `export_batches`
records, the table's snapshot metadata and the scan plan's file list.

Plan Iceberg compaction:

```bash
tuladb-iceberg-exporter compact --config config/local.toml --json
```

Execute a cold, explicit range after reviewing the plan:

```bash
tuladb-iceberg-exporter compact \
  --config config/local.toml \
  --execute \
  --from-global-seq 1000000 \
  --to-global-seq 1250000
```

Compaction is separate from the exporter service. It rewrites old Iceberg data
files through an Iceberg overwrite commit, records a maintenance snapshot, and
verifies row counts plus signed/absolute amount sums before and after the
rewrite. It never advances the export checkpoint.

Plan Iceberg snapshot cleanup:

```bash
tuladb-iceberg-exporter cleanup --config config/local.toml --json
```

Execute snapshot cleanup after reviewing the plan:

```bash
tuladb-iceberg-exporter cleanup --config config/local.toml --execute
```

Cleanup is separate from export and compaction. It expires old Iceberg snapshots
through PyIceberg's native table maintenance API and never advances the export
checkpoint. Physical orphan-file deletion is disabled and fails closed unless a
certified native Iceberg adapter is available; do not use ad-hoc object-store
prefix deletes for financial ledger data.

Enable Iceberg append mode:

```bash
pip install -e '.[iceberg]'
```

Then set:

```toml
[iceberg]
enabled = true
catalog_name = "default"
table_identifier = "default.transactions"
create_table_if_missing = true
cluster_id = "prod-cluster-a"
pipeline_id = "ledger-transactions"
writer_epoch = 1
expected_record_schema_version = 1

[iceberg.catalog]
type = "sql"
uri = "sqlite:///warehouse/pyiceberg_catalog.db"
warehouse = "file:///absolute/path/to/warehouse"
```

Service lifecycle defaults live in the same config file:

```toml
[service]
pid_file = "/run/tuladb/iceberg-exporter.pid"
log_file = "/var/log/tuladb-iceberg-exporter/exporter.log"
start_timeout_s = 10.0
stop_timeout_s = 30.0
```

### Offline install

Every release publishes offline wheelhouses at
`https://downloads.tuladb.com/exporter/<VERSION>/`: one tarball per extra
(`base`, `iceberg`, `biglake`, `s3tables`, `all`), CPython (3.11, 3.12) and
Linux arch (x86_64, aarch64), holding every wheel that install needs, the
exporter's included. Verify `SHA256SUMS` first (docs/release.md,
"Signature"), then, with no network:

```bash
tar -xzf tuladb-iceberg-exporter-<VERSION>-wheelhouse-s3tables-cp311-manylinux_x86_64.tar.gz
cd tuladb-iceberg-exporter-<VERSION>-wheelhouse-s3tables-cp311-manylinux_x86_64
python3.11 -m venv /opt/tuladb-iceberg-exporter
/opt/tuladb-iceberg-exporter/bin/pip install --no-index --find-links wheelhouse \
    'tuladb-iceberg-exporter[s3tables]'
```

Each tarball also carries the hashed lock it was built from
(`requirements.txt`; the repository's `requirements/`) for a
`--require-hashes` install. Build them yourself with
`scripts/build-offline-bundle.sh` (`make offline-bundle`).

Secrets need not be inline in the TOML. `[source.auth]` takes
`credential_file` / `credential_env`; any `[iceberg.catalog]` property (a REST
`token` or `credential`, `s3.secret-access-key`, a SQL catalog `uri` with a
password) can be read from `[iceberg.catalog.secrets_from_env]` (property =
"ENV_VAR") or `[iceberg.catalog.secrets_from_files]` (property = "/path"); and
the fencing database URI from `sql_uri_env` / `sql_uri_file`, or its password
alone from `sql_password_env` / `sql_password_file`. An unset variable or an
empty file refuses the config at load.

Active/passive fencing is optional. When enabled, the exporter must own the
lease before fetching or appending. Passive instances stay alive and poll until
the active lease is released or expires.

```toml
[fencing]
enabled = true
backend = "sql"
sql_uri = "postgresql+psycopg://user@postgres.internal:5432/tuladb"
sql_password_file = "/etc/tuladb/fencing-db-password"   # or sql_password_env
sql_table = "tuladb_iceberg_exporter_leases"
instance_id = "iceberg-exporter-a"
lease_ttl_seconds = 30.0
renew_interval_seconds = 10.0
acquire_timeout_seconds = 0.0
passive_poll_interval_ms = 1000
```

Use the SQL backend for multi-host/Kubernetes active/passive deployments. The
file backend is still useful for one host or a strongly consistent shared
filesystem. `writer_epoch` remains the logical Iceberg fencing token.

Compaction defaults are intentionally conservative:

```toml
[compaction]
enabled = true
lock_file = "/var/lib/tuladb-iceberg-exporter/compaction.lock"
small_file_threshold_bytes = 33554432
target_file_size_bytes = 268435456
min_input_files = 4
max_input_files = 128
max_bytes_per_run = 1073741824
min_lag_global_batches = 1000
require_explicit_range_for_execute = true
```

With `require_explicit_range_for_execute = true`, automatic planning suggests a
safe historical range, but execution requires the operator or scheduler to pass
that exact range. This avoids rewriting the active append tail.

Scheduled compaction deployment templates are included:

- `deploy/systemd/tuladb-iceberg-compactor.service`
- `deploy/systemd/tuladb-iceberg-compactor.timer`
- `deploy/systemd/iceberg-compactor.env.example`
- `deploy/kubernetes/tuladb-iceberg-compactor-cronjob.yaml`

All scheduled compactor templates are plan-only by default. Add `--execute`
only after the compaction policy and cold-range bounds are reviewed.

Cleanup defaults are conservative and disabled:

```toml
[cleanup]
enabled = true
lock_file = "/var/lib/tuladb-iceberg-exporter/cleanup.lock"
min_snapshot_age_seconds = 86400
retain_last_snapshots = 10
max_snapshots_per_run = 100
delete_orphan_files = false
min_orphan_age_seconds = 86400
require_native_orphan_cleanup = true
# shared changelog tables: other writers' snapshots are left to their own
# cleanup until this old (their newest always kept); warn above the count
foreign_snapshot_max_age_seconds = 604800
foreign_snapshot_warn_count = 300
```

Keep `delete_orphan_files = false` until the Iceberg runtime provides a native,
table-aware orphan-file removal action that is certified for the configured
catalog/object store. Snapshot expiration is safe to run periodically once the
retention window matches audit and restore requirements.

Observability follows the same split as `tuladb-core`: the exporter writes a
cheap atomic JSON snapshot and a separate metrics process exposes `/metrics`
and `/healthz`.

```toml
[observability]
enabled = true
metrics_snapshot_path = "/var/lib/tuladb-iceberg-exporter/metrics.json"
metrics_snapshot_interval_ms = 1000
cluster_name = "prod-cluster-a"
node_id = "iceberg-exporter-1"
metrics_listen = "127.0.0.1:9146"
max_snapshot_age_seconds = 120
```

When `iceberg.enabled = true`, the exporter writes through PyIceberg's catalog
append path and advances the local checkpoint only after the append commit
returns successfully.

Production Iceberg mode records every committed batch in a durable
`export_batches.jsonl` control table next to the local checkpoint. Recovery uses
both `export_batches` and Iceberg snapshot metadata, then fails closed if they
disagree.

## Output Layout

For `warehouse_path = "./warehouse"` and `table_name = "transactions"`:

```text
warehouse/
  transactions/
    data/
      batch_0000000000000001_0000000000001024.parquet
    _manifests/
      manifest.jsonl
state/
  checkpoint.json
  export_batches.jsonl
  exporter.pid
  exporter.log
```

A partitioned worker suffixes its state files with `_p<i>` (`checkpoint_p0.json`,
`export_batches_p0.jsonl`, and the default pid file `exporter_p0.pid`), so the
N workers of one lake can share a state directory. An explicit
`[service] pid_file` is used as written: give each worker its own.

Parquet files are deterministic by exported source range, and the per-file
manifest (`_manifests/manifest.jsonl`) is the durable record of every committed
range. If the exporter crashes after writing a file but before checkpointing in
local Parquet mode, the next run reconstructs the resume point from the manifest
(`ParquetSink.latest_checkpoint`) and continues *after* the already-written
range — it does not re-coalesce and re-export it. As defence-in-depth, a write
whose source range exactly matches a committed one is an idempotent no-op (or
fails closed if its digest differs), and a write whose range *overlaps* a
committed one fails closed rather than emitting a second, non-overwriting file.
This gives local Parquet mode the same no-duplicate guarantee as Iceberg mode.

In Iceberg mode, the local checkpoint is a fast-resume cache. Iceberg snapshot
metadata is the reconciliation source of truth when an exporter crashes after a
catalog commit but before the local checkpoint update.

## Production Iceberg Flow

```text
TulaDB committed WAL / Raft committed watermark
  -> query_port MSG_EXPORT_CURSOR
  -> exporter micro-batch
  -> staged Parquet data files
  -> atomic Iceberg append commit
  -> checkpoint after commit
```

Each Iceberg snapshot should include the TulaDB source range and file-set
digest so restarts can detect already-committed ranges and repair the local
checkpoint without duplicating logical rows.

## High-Throughput Pipeline

By default the exporter runs a sequential loop: fetch one cursor page (at most
`max_records`, default 8192 -- the source's cap since tuladb-core e0764daf; an
older source clamps to 1024), commit it, checkpoint, repeat. That is one
Iceberg append per page. During a TulaDB burst (e.g. ~1M TPS) the query
listener may also refuse the export cursor with `QUERY_UNAVAILABLE` to protect
the write path, so the exporter reads nothing during the spike and must drain
the backlog afterward. At one commit per page the drain cannot keep up.

Set `[source].pipeline_enabled = true` to switch to an overlapped, coalescing
loop:

- a single **reader thread** fetches cursor pages ahead of the writer onto a
  bounded queue (`pipeline_queue_depth`), overlapping network round-trips with
  commit latency and capping in-flight memory via backpressure;
- a **decode stage** (`pipeline_decode_stage`, default on) decodes each page's
  rows between the two, so the reader sends the next request as soon as a
  page's header arrives instead of after its rows are parsed (one FIFO, one
  decoder: order and exactly-once are unchanged);
- the **writer** coalesces contiguous pages into one micro-batch and issues a
  single Iceberg commit per batch;
- a batch flushes when any bound trips — `max_records_per_commit`,
  `max_pages_per_commit`, or `commit_linger_ms` (a latency cap) — and always
  when the reader catches up.

```toml
[source]
pipeline_enabled = true
max_records_per_commit = 262144   # ~4 commits/sec at 1M TPS; ~256k records/batch
max_pages_per_commit   = 512      # >= ceil(max_records_per_commit / max_records)
commit_linger_ms       = 1000     # latency cap before a partial batch flushes
pipeline_queue_depth   = 32       # reader read-ahead / backpressure bound
max_records            = 8192     # page rows (an older source clamps to 1024)
# max_bytes unset: sized per row shape for a full 8192-row page --
# 1.5 MiB fixed-width, 2 MiB varlen (include_metadata), 2.5 MiB FULL_TXN.
# Explicit values up to 4 MiB; 0 = one ~208 KiB source buffer.
```

The safety contract is unchanged: one reader advances a single monotonic cursor
(pages stay ordered and non-overlapping), `merge_pages` fails closed on any
schema-version change or non-increasing boundary, the checkpoint advances only
after a successful coalesced commit, and fencing is enforced at every burst
boundary and every commit. An uncommitted in-flight batch is simply replayed
from the last durable checkpoint on restart.

Sizing note: peak memory is the batch **plus** the in-flight queue, roughly
`(max_records_per_commit + pipeline_queue_depth * max_records)` records (~650 B
each decoded) plus the Arrow table and Parquet buffer at flush (~350-450 MB
with the values above). The raw page bytes held ahead are bounded by
`pipeline_queue_depth * max_bytes`, which the loader refuses above 512 MiB
(at most 128 pages at the 4 MiB cap).
Raise `max_records_per_commit` toward 1,000,000 only if you need <=1-2
commits/sec; a 1M-record batch is ~1 GiB live. Watch the coalescing ratio via
`tuladb_iceberg_exporter_export_coalesced_source_pages_total /
tuladb_iceberg_exporter_export_commits_total`.

Batch sizing only speeds the drain **after** a burst. Sustained real-time export
under load also requires the source to serve the cursor during the burst (no
`QUERY_UNAVAILABLE`), which is a TulaDB-side capacity concern.

## AWS S3 Tables Target

The sink is catalog-agnostic (it uses PyIceberg), so it writes to AWS S3 Tables
through the S3 Tables Iceberg REST catalog with SigV4 signing. Install the AWS
extra and point `[iceberg.catalog]` at the regional endpoint:

```bash
pip install "tuladb-iceberg-exporter[s3tables]"
```

```toml
[iceberg]
enabled = true
table_identifier = "ledger.transactions"   # namespace.table in the table bucket
writer_epoch = 1
commit_max_retries = 6                      # transient-error retry (below)
commit_retry_backoff_ms = 250
commit_retry_max_backoff_ms = 8000

[iceberg.catalog]
type = "rest"
uri = "https://s3tables.ca-central-1.amazonaws.com/iceberg"
warehouse = "arn:aws:s3tables:ca-central-1:123456789012:bucket/my-ledger-bucket"
"rest.sigv4-enabled" = "true"
"rest.signing-name" = "s3tables"
"rest.signing-region" = "ca-central-1"
"s3.region" = "ca-central-1"
py-io-impl = "pyiceberg.io.fsspec.FsspecFileIO"   # required, see below
```

`py-io-impl` is required: pyarrow's bundled S3 client signs the table bucket
with SigV2 and S3 Tables rejects it mid-commit ("Please use Signature Version
4"). The FsspecFileIO signs through boto3 via s3fs, which the `s3tables` extra
installs.

A full, annotated example is in `deploy/examples/s3tables.toml.example`.
Credentials come from the standard AWS chain (env, shared config, EC2/ECS/EKS
role, or IRSA); the principal needs `s3tables` permissions on the table bucket.
`doctor` detects an S3 Tables target and warns about missing SigV4 settings
or a missing `py-io-impl`.

Production notes:

- **Transient-error resilience.** Commits to a remote catalog retry with jittered
  exponential backoff on throttling, 5xx, network blips, and optimistic-
  concurrency conflicts (`commit_max_retries`, `commit_retry_backoff_ms`,
  `commit_retry_max_backoff_ms`). Logical fail-closed errors
  (gap/overlap/digest/writer-epoch) are **never** retried, and retries are
  idempotent — a retry after an unknown-outcome commit re-checks the catalog and
  returns the existing snapshot instead of double-appending.
- **Let S3 Tables own maintenance.** S3 Tables auto-compacts and auto-expires
  snapshots. Disable the exporter's `[compaction]` and `[cleanup]` so the two do
  not fight over the same table.
- **Snapshot metadata.** Checkpoint repair and reconciliation read the exporter's
  `tuladb.*` keys back out of the Iceberg snapshot summary. Standard Iceberg REST
  preserves these; smoke-test a real write before relying on snapshot-based
  recovery.
- **Live-verified, but not in every CI run.** Verified end to end against a
  real table bucket on 2026-08-10 (`docs/s3tables-testing.md`); `make
  test-s3tables` and the manually dispatched `s3tables-integration` workflow
  re-run it. The per-commit CI covers it with unit tests only (catalog config,
  retry classification, idempotent retry), so validate with real credentials
  before production.

## GCP BigLake Target

The same sink writes to GCP BigLake through its Iceberg REST catalog. Choosing
S3 Tables or BigLake is configuration only — `[iceberg.catalog]` and the
maintenance sections; nothing else in the config changes:

```bash
pip install "tuladb-iceberg-exporter[biglake]"
```

```toml
[iceberg.catalog]
type = "rest"
uri = "https://biglake.googleapis.com/iceberg/v1/restcatalog"
warehouse = "gs://my-ledger-lake"                  # gcs-bucket catalog
"header.x-goog-user-project" = "my-gcp-project"   # billing project, required

[iceberg.catalog.auth]
type = "google"          # Application Default Credentials

[cleanup]                # REQUIRED on BigLake -- see below
enabled = true
min_snapshot_age_seconds = 120
```

A full, annotated example is in `deploy/examples/biglake.toml.example`; `doctor`
detects a BigLake target and warns about missing auth, billing project, or a
cleanup policy the metadata cap cannot absorb.

Production notes (all measured against a live BigLake catalog, 2026-09-27):

- **Run cleanup every minute.** BigLake does not expire snapshots and refuses
  commits once `metadata.json` would exceed 1 MB. At ~1.4 KB per export snapshot
  that is ~700 snapshots — about 12 minutes of sustained load. Schedule
  `cleanup --execute` every minute
  (`deploy/kubernetes/tuladb-iceberg-cleanup-biglake-cronjob.yaml`) with a short
  `min_snapshot_age_seconds`. If it is ever missed the exporter fails closed
  with an explicit "metadata.json is at the catalog's size limit" error, and
  resumes cleanly after cleanup.
- **Run compaction.** BigLake leaves table management off for REST-catalog
  tables; the exporter's compactor works against it.
- **Full snapshot history.** The exporter always asks REST catalogs for every
  snapshot (`snapshot-loading-mode = "all"`); BigLake otherwise returns only the
  current one, which blinds dedup, reconcile and cleanup.
- **Validate end to end** with `make test-biglake` (see `docs/biglake-testing.md`).

## Iceberg Safety Contract

When `iceberg.enabled = true`, the exporter:

- stores full range tokens: start, end, and next resume token, each with
  `global_batch_seq` and `array_index`;
- records `record_count`, `amount_sum`, and `amount_abs_sum` in both Iceberg
  snapshot metadata and `export_batches`;
- writes `STARTED` and `COMMITTED` records to `export_batches.jsonl`;
- checks `writer_epoch` against existing committed batches and snapshots, and
  fails closed if a newer writer has already committed;
- rejects gaps, overlaps, exact ranges with different digests, schema-version
  ambiguity, and control/snapshot reconciliation mismatches;
- uses PyIceberg append-generated data file paths and checks existing committed
  range metadata before append, so retries do not re-add the same logical batch.

Strict schema evolution is the default. Bump `expected_record_schema_version`
only with an explicit exporter release and table schema migration plan.

## Iceberg Compaction Contract

Compaction is downstream Iceberg maintenance:

- runs as `compact`, not inside the exporter checkpoint loop;
- uses a local `compaction.lock` to avoid two local compaction workers;
- derives a cold upper bound from the latest committed export checkpoint minus
  `min_lag_global_batches`;
- plans only files below `small_file_threshold_bytes`;
- refuses execution unless the file count, byte limit, cold range, and optional
  explicit range checks pass;
- commits with `tuladb.snapshot_kind = "compaction"` so checkpoint recovery
  ignores the maintenance snapshot;
- validates `record_count`, `amount_sum`, and `amount_abs_sum` before and after
  the overwrite.

Old files remain governed by the separate `cleanup` command, Iceberg snapshot
retention, and object-store lifecycle policy. Compaction must not manually
delete data files.

## Observability Contract

The exporter does not embed a Prometheus SDK in the checkpoint-critical path.
It writes an atomic snapshot, by default at `state/metrics.json`, and the
`metrics` command converts that snapshot to Prometheus/OpenMetrics text.

Low-cardinality labels:

- `cluster`
- `node`
- `pipeline`

Key metrics:

- `tuladb_iceberg_exporter_up`
- `tuladb_iceberg_exporter_snapshot_age_seconds`
- `tuladb_iceberg_exporter_snapshot_stale`
- `tuladb_iceberg_exporter_running`
- `tuladb_iceberg_exporter_export_pages_total`
- `tuladb_iceberg_exporter_export_records_total`
- `tuladb_iceberg_exporter_export_commits_total`
- `tuladb_iceberg_exporter_export_coalesced_source_pages_total`
- `tuladb_iceberg_exporter_last_commit_records`
- `tuladb_iceberg_exporter_pipeline_queue_size`
- `tuladb_iceberg_exporter_export_source_retries_total`
- `tuladb_iceberg_exporter_export_errors_total`
- `tuladb_iceberg_exporter_export_fail_closed_total`
- `tuladb_iceberg_exporter_checkpoint_global_batch_seq`
- `tuladb_iceberg_exporter_checkpoint_array_index`
- `tuladb_iceberg_exporter_committed_global_batch_seq`
- `tuladb_iceberg_exporter_export_lag_batches`
- `tuladb_iceberg_exporter_fetch_seconds_total`
- `tuladb_iceberg_exporter_write_seconds_total`
- `tuladb_iceberg_exporter_checkpoint_save_seconds_total`
- `tuladb_iceberg_exporter_compaction_plans_total`
- `tuladb_iceberg_exporter_compaction_executes_total`
- `tuladb_iceberg_exporter_compaction_errors_total`
- `tuladb_iceberg_exporter_fencing_active`
- `tuladb_iceberg_exporter_fencing_acquire_total`
- `tuladb_iceberg_exporter_fencing_passive_total`
- `tuladb_iceberg_exporter_fencing_lost_total`
- `tuladb_iceberg_exporter_cleanup_plans_total`
- `tuladb_iceberg_exporter_cleanup_executes_total`
- `tuladb_iceberg_exporter_cleanup_errors_total`
- `tuladb_iceberg_exporter_cleanup_expire_snapshot_count`

The metrics listener is intentionally unauthenticated. Bind it to loopback,
pod-local networking, or a private metrics network, and put cross-host access
behind Prometheus network policy, service mesh, or a proxy that provides TLS
and authentication.

## Current Scope

Implemented:

- committed transaction export over TulaDB wire protocol;
- TLS, mTLS, API-key, and bearer-token source connections;
- resume by `(global_batch_seq, array_index)`;
- local checkpoint with atomic replace;
- Parquet schema for transaction facts;
- optional Iceberg append commits through PyIceberg;
- AWS S3 Tables target via the Iceberg REST catalog (SigV4), with the
  `s3tables` install extra;
- GCP BigLake target via the Iceberg REST catalog (Google ADC), with the
  `biglake` install extra, verified end to end against a live catalog;
- high-throughput coalescing pipeline (reader/writer overlap, one commit per
  micro-batch) for burst drain;
- bounded, idempotent, jittered-backoff retry for transient remote-catalog
  commit errors;
- optional Iceberg compaction command with fail-closed reconciliation;
- optional Iceberg snapshot cleanup command with fail-closed orphan deletion;
- optional active/passive file-lease fencing;
- optional active/passive SQL fencing for multi-host deployments;
- reconciliation command for control, checkpoint, snapshot metadata, and table totals;
- scheduled systemd and Kubernetes compactor templates;
- source/wheel/container release packaging;
- core-style metrics snapshot and Prometheus/OpenMetrics exporter command;
- checkpoint repair from Iceberg snapshot metadata;
- durable `export_batches` control records for committed exports;
- fail-closed range/digest/schema/writer-epoch reconciliation;
- one-shot, foreground service, and managed start/stop/status modes.

Planned:

- retention leases;
- participant/account dimension streams;
- certified native orphan-file cleanup adapter;
- **horizontal export throughput via a partitioned source cursor (IEX-1).** Today
  the source is one full ordered stream with no partition filter, so a table is
  **single active writer** (extra instances are passive HA standbys, not parallel)
  and one Python pipeline peaks at ~100–130K records/sec (single-thread parse +
  Parquet encode). Sustained higher rates (e.g. 200K/sec) need the partitioned
  cursor — see `docs/design.md` "Throughput And Horizontal-Scale Boundary (IEX-1)".
