Conversation
Cooldown carve-out mirrors the pyducklake one: 1.1.0 published 2026-09-18, inside the 7-day exclude-newer window.
Restore millpond/sink.py (Sink protocol, SAFE_IDENTIFIER, check_reserved_collision, make_sink) removed with the iceberg/icebox backends. Adaptations against the tagged version: - write() returns the written-record count (grown by the DuckLake backend for the VARIANT companion-collision skip while the seam was gone; main.py feeds it to records_written_total) - SAFE_IDENTIFIER moves to the seam; schema.py re-exports it for its existing importers - check_reserved_collision takes backend_name again; ducklake.py keeps its RESERVED_COLUMNS set and passes "DuckLake", so the message is byte-identical to the inlined version - main.py goes back to protocol dispatch via make_sink(cfg); _write_with_retry / reset_caches semantics unchanged - Config gains the destination field (ducklake-only until the hoglake backend lands)
MILLPOND_DESTINATION=hoglake selects the hoglake backend (default stays
ducklake — zero behavior change for existing deployments). The active
destination's env group is loaded and validated; the inactive one is
nulled so stray vars can never leak across backends.
- HOGLAKE_URL / _CATALOG / _NAMESPACE / _TABLE / _S3_ACCESS_KEY /
_S3_SECRET_KEY required; _DATA_PATH / _S3_ENDPOINT / _S3_REGION
optional
- names validated against the server's identifier rules (catalog:
[a-z][a-z0-9_-]{0,62}; namespace/table: [A-Za-z_][A-Za-z0-9_-]{0,127})
- HOGLAKE_PARTITION_BY parsed at load() into (column, transform, param)
triples against the server transform vocabulary (identity, year,
month, day, hour, bucket); unmappable expressions refuse startup with
the vocabulary in the message, never a per-batch failure. pyhoglake's
client-side truncate is deliberately excluded (server gap).
- table_label / default group_id follow the active destination's table
millpond/hoglake.py writes batches via pyhoglake's footer-shipping
append path. Events land as TEXT (properties stay JSON strings);
VARIANT is explicitly out of scope.
- first-write bootstrap: catalog (created only when HOGLAKE_DATA_PATH
is set) -> namespace -> table, concurrent-creation tolerant at every
level; partition spec + sort order declared in one alter at creation
with field ids resolved from the created schema. Partition columns
missing from the schema are fatal; missing sort fields degrade like
main._apply_sort's skip (the sort spec is writer-advisory).
- _inserted_at stamped once per flush (timestamptz us UTC).
- schema evolution mirrors schema.SchemaManager: add_column per new
batch field, promote_column for int->long / float->double, per-column
degrade on failure (log + errors_total{type=schema}), 409 concurrent
DDL absorbed by re-resolve. Unsafe and _hog-prefixed payload keys are
dropped per column (records_skipped_total{reason=unsafe_field_name})
so one poison key cannot wedge the partition.
- batch aligned for pyhoglake's strict _align_table: refused columns
dropped, missing table columns null-filled (INSERT BY NAME analog).
- is_retryable() classifies OCC/transport/5xx as retryable vs permanent
4xx contracts; main._write_with_retry labels pyhoglake conflicts as
errors_total{type=hoglake_commit_contention} (duck-typed on the
retryable class attr — no pyhoglake import on the ducklake path).
- offset-sequencing contract pinned in tests: kafka.commit only after
sink.write success, +1 next-to-fetch, write-then-commit order.
- MILLPOND_DESTINATION=hoglake + MILLPOND_VARIANT_COLUMNS is a startup error: hoglake has no VARIANT column type and a silently-skipped dual-write config is how mixed fleets rot. Events land as text; the guard is revisitable when hoglake grows a variant path. - millpond_hoglake_files_written_total (pipeline/broker_source): parquet files registered per append commit — with partitioned fanout this is the hoglake compaction-debt feed rate. Additive; ducklake destinations never move it. - Commit author is millpond/<table>/<ordinal> for multi-writer forensics in the snapshot log.
tests/hoglake_stack/: throwaway compose stack (project millpond-hog-it) with hoglake-server (ghcr latest, HOGLAKE_SERVER_IMAGE override) + postgres + minio, every port a high 127.0.0.1 bind (25432/29000/28080; kafka 29092 behind the e2e profile) so it coexists with a live hoglake dev environment. Fixtures skip cleanly without Docker or the image and tear the stack down (volumes included) at session end. 15 tests drive the real HoglakeSink: catalog/namespace/table bootstrap (spec + sort order verified via the API), partitioned fanout (one file per team tuple, partition_values on each), append visibility (files + scan_plan record counts, non-pending stats, parquet readback from MinIO), schema evolution round-trip incl. null-fill of upstream-removed columns, two-writer concurrent create + concurrent evolve/append, reset_caches / new-instance resume / external drop-recreate recovery, column hygiene live, and the at-least-once sequence against a real outage (server stopped: _flush raises, kafka.commit never called; restarted: clean retry lands rows exactly once, offset commits +1). The live suite immediately earned its keep: concurrent-DDL alignment race — another writer's add_column between this sink's alignment and append()'s pre-flight resolve makes pyhoglake's strict _align_table refuse the batch (data is missing table columns). Fix: refresh the live schema, null-fill, re-append once (pinned by unit tests; a second refusal still raises).
Mirrors tests/e2e/test_e2e.py's harness with destination=hoglake: the millpond-hog-it stack's e2e profile adds a single-node KRaft Kafka (127.0.0.1:29092; empty-host listener form — the apache/kafka docker wrapper's storage-format step rejects 0.0.0.0); the test produces 600 events across two teams, runs millpond's real entry point as a host subprocess (destination=hoglake, partition_by=team_id,month(_inserted_at), sort_by=team_id), and asserts the end state: exactly 600 rows registered (no dupes on the clean path), table shape text-only with _inserted_at timestamptz, partition spec applied with per-team fanout, sort order declared, /healthz 200 and /metrics serving millpond_* counters during the run, clean SIGTERM shutdown (exit 0). server.start() now honors MILLPOND_HTTP_PORT (default unchanged at 8000) so the harness can bind the health server on a high port next to other local services.
README: two-destination comparison table (partitioning, sort order, evolution, VARIANT posture, _inserted_at semantics, concurrency), HOGLAKE_* config reference with the partition-expression vocabulary, delivery-semantics note (409 = zero rows, orphaned parquet is hoglake cleanup's problem), hoglake partitioning paragraph (Iceberg-semantics transforms, fanout commits — not Hive directories), and the deferred scope (VARIANT rejected at startup). justfile: test-hoglake-integration / test-hoglake-e2e recipes; test-integration now ignores the docker-gated hoglake suite so the fast in-memory recipe stays fast and CI wiring stays a deliberate decision. AGENT.md: sink-seam section rewritten for make_sink dispatch, project structure and pre-push checklist updated.
Two ways a single payload key could crash-loop a partition forever, both
specific to the hoglake backend:
* RESERVED_COLUMNS carried year/month/day/hour over from the DuckLake
set, where they are load-bearing (Hive partitioning materializes a
derived column per key). Hoglake partitions by Iceberg-semantics
transforms recorded in the catalog, so those names are ordinary
columns and nothing in the server reserves them. A producer key named
'month' raised a fatal ValueError that no retry can clear, while a
worse-formed key ('month-of-year') was merely dropped. The set is now
_inserted_at only. The two backends' sets now differ, which stays safe
for a destination swap because it runs one way: hoglake accepts every
batch DuckLake accepted.
* The server's column-name rule is ^[A-Za-z_][A-Za-z0-9_-]{0,127}$, and
the shared SAFE_IDENTIFIER has no length bound. In steady state an
over-long name degraded (the add_column 422s, the column is dropped),
but bootstrap ships the whole schema in one create_table, so one
129-character key in the first batch wedged the table forever. The cap
now sits on the same drop path as the other unwritable names.
…y destination Three small correctness items that all come down to config knowing what the destination is: * MILLPOND_HTTP_PORT was read inside server.start(), behind config.py's back — invisible to the startup config log and unreachable from a Config a caller already holds. It is a Config field now (validated: integer, 0-65535) and server.start() takes the port as an argument only. The test that claimed to prove the explicit argument wins compared two sources that were both 0; it now uses two different values, so it can fail. * The default GROUP_ID omitted the destination, so a shadow deployment (same topic, same table name, other destination — how a migration is canaried) shared an offset namespace with the pipeline it shadowed. That does not duplicate the work, it SPLITS it. The DuckLake form stays byte-identical on purpose: every deployed pipeline's offsets live under it and a change would replay the retention window. * _is_commit_contention is string-matching on generic Postgres/DuckLake wording, and the hoglake control plane is Postgres-backed too, so a hoglake 500 carrying 'duplicate key value' was labelled ducklake_commit_contention — a DuckLake alert from a pod with no DuckLake. The classifier is now gated by destination. _write_with_retry also grows the seam the next two commits need: a per-sink retry budget, a Retry-After hint, an is_retryable veto, and the per-call write_kwargs the icebox sink used at tag final-iceberg. Both sinks are unaffected until they implement the hooks.
…ry budget is_retryable() was dead code with a false-green test. Nothing called it, its docstring claimed IncarnationChangedError was retryable while the server-raised form (409, carries a status) returned False, and the test constructed the statusless client-side form that never reaches production. 429 and 408 — the two 4xx codes that mean 'not now' — were classified permanent, and a statusless HoglakeError (the request never reached the server) relied on falling off the end of the function. It is now wired in: HoglakeSink.is_retryable is consulted by _write_with_retry, which re-raises immediately rather than burning the ladder on a 422 that cannot become valid by waiting. Every test case is built in the shape pyhoglake actually produces, including both IncarnationChangedError forms. The budget itself was inherited from DuckLake, where millpond's 3 attempts wrap an inner loop that retries 100 times. Hoglake has no inner loop, so 3 was the whole budget against a catalog whose backpressure signal is 503 + 'Retry-After: 1' and which expects to be asked again — and pyhoglake drops that header on the floor. So: * HOGLAKE_MAX_RETRY_COUNT (default 8) and HOGLAKE_REQUEST_TIMEOUT_S (default 30, previously pyhoglake's unreachable hardcoded value). The product bounds how long one flush can sit inside write(), which the 480s liveness deadline caps. * The sink reads Retry-After off the raw response via an httpx hook and offers it to the retry loop, which prefers it to the doubling curve and clamps it to 30s so a hostile header cannot park the consume loop. * The exponential curve is capped at the same 30s.
/alter applies its op list in order, atomically, as a single DDL commit — one snapshot, one schema-version bump, one pass through the per-catalog commit lock. Adding a column per commit multiplied that traffic by the width of the schema drift (a producer rollout that adds six keys paid six commits per pod), and gave every column its own chance to lose the concurrent-DDL race against another pod adding the same set. The batch stays an optimization, never a semantics change: the alter is one transaction, so a refusal applies nothing, and the per-column loop runs as the fallback with the same degrade-and-metric behaviour it always had. A column with no hoglake type mapping is dropped before the op list is built, so it cannot take the other columns' DDL down with it.
…e it on every resolve Table creation is two round trips — create_table, then one alter that declares the partition spec and sort order — and the window between them was unrecoverable. Only CommitConflictError was caught, so any other failure (a 422 from a spec the server refuses, a timeout, a 503) left the table CREATED and UNPARTITIONED while self._table was never set; the retry then took the 'table already exists' path and returned happily. One transient blip at first write and the pipeline wrote to a permanently unpartitioned table, silently, forever. Three changes close it: * _declare_specs verifies its own work. The alter's returned TableInfo must carry the spec config asked for, or the bootstrap raises. * The existing-table path reconciles instead of trusting. Live spec matches config: proceed. Live spec absent, config has one: declare it (this is the create-then-alter window reopening, and declaring is the recovery). Anything else — a changed HOGLAKE_PARTITION_BY, a changed bucket count, a partitioned table under a config that says nothing — raises with both layouts in the message and the live one rendered in HOGLAKE_PARTITION_BY's own grammar. A spec typo (month(team_id), bucket on a double) therefore fails identically on every attempt rather than only the first. * config.load() refuses MILLPOND_DESTINATION=hoglake with DUCKLAKE_PARTITION_BY set and HOGLAKE_PARTITION_BY unset. Stray vars from the inactive destination are otherwise harmless, which is why they are nulled — but this one silently changes the shape of the data on a destination flip, and partitioning is the property that is painful to add afterwards, since existing files keep their vintage. Separately: the catalog is now resolved in __init__. config.py and the README both called a missing catalog a startup error and it was not one — a wrong URL, wrong credentials or an uncreated catalog surfaced on the first flush, after the pod had passed its probes, taken its partitions and built lag. One request at construction makes the claim true. Verified against a real hoglake server (19 integration tests), including the 422 a bad transform actually returns.
… Kafka offsets A commit the server APPLIED whose response never arrives is, at the client, indistinguishable from a commit that never happened: both are a timeout. millpond retried, the retry minted a fresh parquet path (uuid4) and committed again, and hoglake accepted it — there is no unique index on a data file's path, deliberately, so the same rows published twice. The offsets then advanced over both copies. No log, no metric, nothing afterwards to tell the two apart. Reproduced against the real server before fixing it: drop one commit response mid-flight and an 11-row table comes back with 20 rows (tests/integration TestLostCommitResponse, red at the parent commit). The flush is now published in two steps. prepare_append_files uploads the parquet and returns the commit request; Catalog.commit_prepared publishes it. The request is held IN MEMORY for the lifetime of the flush, so a retry re-sends it byte for byte instead of building a second one — no re-upload, the same file set, the same key. The server resolves the replay under its per-catalog commit lock: a receipt for that key returns the original result without writing. The key is a UUIDv5 over (catalog, namespace, table) and the Kafka offset range being flushed, threaded from main.py through the write_kwargs seam the icebox sink used at tag final-iceberg (DuckLakeSink's signature is untouched — its INSERT commits inside a transaction whose outcome the client always learns). A flush only happens with at least one new record buffered, so two flushes of one pipeline can never derive the same key; receipts are catalog-scoped, which is why the table identity is in the hash. Two details worth the reader's attention: * Transport failures are classified UNCERTAIN, not failed. The payload stays cached and the exception propagates precisely so the retry re-sends the same request and lets the server decide whether it already landed. That reasoning lives on _commit_prepared. * read_snapshot is stripped from the payload. prepare_append_files pins it to the catalog head, and the server's conflict scan then 409s the commit if any DDL touched the table since — which for millpond means 'another pod added a column', the most likely thing to happen during a producer rollout, and a frozen payload can never clear it. Appends never conflict with appends, so dropping the field keeps the blind append the non-idempotent path always had; the incarnation guard on the entry remains the real protection. A key replayed with a different payload (the pod died after the commit applied but before the offsets committed, and Kafka replayed the range) is a 422 that means 'already published': the flush is accepted, the rebuilt upload is counted on millpond_hoglake_orphaned_files_total, and millpond_hoglake_commit_replays_total records the duplicate that did not happen. Failing instead would wedge the partition forever on rows that are already in the lake.
… is missing Neither new suite ran in CI. The e2e job named the DuckLake test file directly, so the hoglake e2e was never invoked at all; the integration job globbed tests/integration and did pick the hoglake suite up, but a Docker hiccup or an unpullable image turned it into a silent skip — a green tick on a job whose only real-server coverage never ran. Both are wired in now, and MILLPOND_REQUIRE_DOCKER_STACK=1 (set on both CI jobs, unset locally) makes an unavailable stack a failure. The skip-vs-fail decision lives in one helper so the two suites cannot drift apart on it. The server image is pinned by multi-arch index digest instead of :latest. These suites are a contract test — they assert what the server does with millpond's commits — and a floating tag makes every unrelated PR's CI a hostage to whatever was published that morning. A server-side behaviour change should land here as a reviewed bump, not as a mystery failure on someone else's branch. The bump procedure is written out in stack.py; HOGLAKE_SERVER_IMAGE still overrides for a locally built server.
…ew knobs
README said a refused hoglake commit could orphan a parquet file
'(reclaimed by hoglake's cleanup), never a duplicate row'. Both halves
were wrong. Hoglake's cleanup reclaims only files the SERVER queued for
removal — snapshot expiry, table drop, compaction staging — and
automated cleanup for client uploads is explicitly future work on the
hoglake side, so those objects are billed storage until somebody sweeps
them. And a duplicate row was exactly what a lost commit response
produced.
The section now states the actual guarantee as a failure-by-failure
table (response lost, crash after apply, refusal, crash before commit),
says plainly that orphans are unreclaimed, and points at the metric and
the {idempotency_key}/ prefix that make an operator sweep tractable.
Also documented: the reconciliation rules and what they refuse (with the
two consequences operators have to plan for — a live table's partition
spec cannot be changed by changing config, and a table carrying someone
else's sort order needs MILLPOND_SORT_BY to match); the per-destination
retry budgets and why they differ; HOGLAKE_MAX_RETRY_COUNT and
HOGLAKE_REQUEST_TIMEOUT_S with the liveness-deadline interaction; the
destination-scoped GROUP_ID default; the three classes of dropped
payload key, including that hyphenated keys are dropped by millpond's
own shared identifier gate even though the hoglake server accepts them;
and that the catalog client has no auth or TLS credential surface at
all, so HOGLAKE_URL must be a trusted-network endpoint.
python -O strips asserts, and this file's stated posture is explicit raises for exactly that reason (see the constructor guards).
`_evolve_and_align` null-fills against the columns it resolved; a round trip later `_prepare` adopts a fresh `table.info()`. Another writer's add_column in that window puts a name in the target schema the batch does not carry, and `pa.Table.select` answers a missing name with a KeyError. KeyError is not a ValidationError, so the self-heal in `write()` never fired for its own motivating case; and it is not a pyhoglake type, so the retry loop classed it "unknown, assume transient" and spent all eight attempts (about 91 seconds of blocked consume loop) before crashing the pod. Null-fill against the freshly adopted columns before the select, so the select can only ever narrow, and catch KeyError alongside ValidationError in the self-heal behind it. The self-heal's matcher was also pointed at a string that can no longer reach it: "data is missing table columns" is raised only by `_align_table`, and this sink stopped calling `Table.append` when it moved to prepared commits. It now matches the two refusals `prepare_append_files` really raises, and deliberately not the partition-arity one, which re-aligning columns cannot fix.
`is_retryable` ended in `return True` — assume transient — and the sink
raised plain `RuntimeError`/`ValueError`/`KeyError` for its loudest
safety stops: a live spec that disagrees with HOGLAKE_PARTITION_BY, a
partition column that is not in the table, a declaration the server did
not apply, a source column colliding with `_inserted_at`. Every one of
them spent eight attempts and about 91 seconds of backoff before the
operator saw the message, with the real error buried under seven
identical repeats.
Those refusals are now `HoglakeSinkError`, a sink-private type
`is_retryable` answers False for, and ValueError/KeyError/TypeError
answer False too. `retryable=True` is available for the one stop a
rebuilt flush really does clear.
HoglakeSinkError subclasses RuntimeError, so existing callers and
`pytest.raises(RuntimeError)` are unaffected.
test_retryable's `RuntimeError("unknown")` case pinned the wrong
behaviour; it is replaced with a genuinely unknown exception type.
A key is a name for "these rows, published to this table". This one
named neither end of that reliably:
* it omitted the table INCARNATION. Receipts live per catalog and
survive a table drop, with no cascade — so a dropped-and-recreated
table answered a flush from its predecessor's receipt, millpond
reported the flush done, and Kafka offsets advanced over rows sitting
in a table that no longer exists. Proven live: a writer reported 8
rows with 3 in the lake.
* it omitted the START of the offset range. "Everything up to 41" is not
a row set: after a rewind, a flush of [0, 41] carries the name of an
earlier flush of [30, 41].
The key is now a UUIDv5 over catalog/namespace/table/table_uuid plus one
`topic:partition:first-last` line per partition, and main.py tracks
(first, last) per partition in the pending buffer rather than the high
offset alone.
A retry is now recognized by the Kafka identity the payload was prepared
for, not by re-deriving the key: the key depends on the live
incarnation, and re-resolving that on the replay path would rebase a
frozen payload's name onto a table it was never prepared for.
With a faithful key, a "reused with a different request" 422 is the
crash-restart case and only that: same incarnation, same complete range,
a payload that differs only where it cannot be reproduced (a fresh
_inserted_at stamp, fresh uuid4 object names). It is still accepted —
failing wedges the partition on rows that are already in the lake — but
it now reports ZERO rows written, because this process published
nothing and records_written_total counts what this process wrote.
The marker is also matched against message AND detail. hoglake's error
body is {error, detail}: for a 422 the message is the code
("validation") and the sentence is in the detail, so the old matcher
only worked through __str__ and its unit fixture had the two fields the
wrong way round.
Three things the same commit path needed to make the above true:
1. A prepared payload no longer survives a refusal the server ANSWERED.
One commit is one transaction, so a 409 or 422 in a response means
zero rows were written and means the same to every identical resend.
Holding it made reset_caches() inert — write() short-circuits to the
replay before the table is re-resolved — so an incarnation 409 or a
removal-queue collision repeated for the whole budget with ONE
prepare behind it and orphaned an upload per cycle. Held now only for
transport-uncertain outcomes and 5xx.
2. The destination is re-read immediately before the commit. A
same-arity partition-spec change between prepare and commit registers
a file stamped with the new spec_id while carrying values computed
under the old one: silent, permanent mis-pruning that the server
cannot catch, because it never opens the file.
3. Every path that can leave an upload unreferenced now counts it.
hoglake reclaims only files the SERVER queued for removal, so an
uncounted orphan is storage nobody can find: a prepare that fails
partway, an answered refusal, a payload superseded by the next flush,
and a payload still held at close().
The delivery table claimed the rows publish exactly once when a pod dies after the commit applied and before the offsets committed. That holds only if the rebuilt flush cuts at the SAME boundary, and the boundary is not reproducible: the size trigger accumulates per poll batch, the time trigger is wall-clock, the allowlist is mutable, and every partition in the flush has to coincide. Proven: 16 rows where 10 were expected. The in-process guarantee is real and unchanged — the payload is held, the retry is byte-identical, the receipt resolves it — and it is now stated as the in-process guarantee it is. Across process boundaries the pipeline is at-least-once with the duplicate opportunistically suppressed when the boundary repeats, which is what the README, the metric comment and the module docstring now say. Both halves are pinned by integration tests against the real server: the same boundary from a new process publishes once, a boundary shifted by one offset publishes twice. The delivery table also gains the two rows it was missing: the recreated table (the key names the incarnation, so the new table cannot answer from the old one's receipt) and the zero-rows-written result of an accepted receipt.
Nine mutants survived all 1008 unit tests. Each of these tests kills at least one of them, and each covers a seam that had no coverage at all rather than weak coverage: * the sort-order half of the declaration post-condition — a server that applies the partition spec and drops the sort order left a table whose every file hoglake compaction re-sorts; * `_partition_groups`' input-order grouping — file registration order is row-id assignment order on the server, so regrouping breaks the correspondence between row ids and MILLPOND_SORT_BY; * the Retry-After hook's install, against a REAL client. Every test around it used a MagicMock, on which an install into a private attribute behind a bare except succeeds no matter what it does; * the parquet the sink actually writes. The autouse fixture replaces `pq.write_table` with a no-op, so the cast to the destination schema, the column ORDER the prepared path compares position by position, and the file itself were never exercised; * `_sink_write_kwargs` and the write_kwargs seam in `_write_with_retry`, which nothing tested — including that the identity goes to EVERY attempt, without which each retry becomes an anonymous publication.
Five findings that all live where config meets the retry path: * The client timeout (30s) equalled the server's commit-lock admission bound (30s), so the client gave up at the same instant the server would have answered 503 + Retry-After. Its explicit backpressure signal was nearly unreachable and surfaced as a transport-uncertain failure — the one outcome that has to hold a prepared payload and resend it blind. Default is 45s. * A server-supplied Retry-After REPLACED the exponential curve. Hoglake's hint is a hardcoded "1", so the whole eight-attempt budget was spent in about eight seconds against a convoyed catalog, then the pod crashed and rejoined the convoy cold. The hint is now a floor and the curve is kept, with upward jitter so a fleet refused by one 503 does not wake in lockstep on every rung. * HOGLAKE_MAX_RETRY_COUNT was unbounded while its interaction with the 480s liveness deadline was documented in a comment — a documented trap one values-file edit away, whose spring looks like a pod SIGKILLed mid-flush with nothing in its own logs. The arithmetic that comment describes is now a startup check over both knobs. * HOGLAKE_DATA_PATH was not validated. It is frozen into the catalog row at creation and hoglake has no delete-catalog route, so a typo mints a permanently unusable catalog under a name nobody can reuse. * config.py imported millpond.schema for two string constants, and schema.py imports duckdb — so every hoglake pod loaded DuckDB at startup. The constants move to the sink seam (schema.py re-exports them), and the lazy-import test stops inspecting make_sink's source and asks a fresh interpreter what it loaded. Also: reconciling a table that has its partition spec but not its sort order now declares the sort alone. Re-declaring a live spec is a DDL commit through the per-catalog lock that races every other writer and re-versions a layout nobody asked to change.
HOGLAKE_REQUEST_TIMEOUT_S's new default and why it is not pyhoglake's 30; the liveness budget as a refusal rather than a warning; the data-path validation; the Retry-After hint as a floor with jitter, not a replacement.
Both were written against inputs too small to distinguish the behaviour they assert from the behaviour they forbid, so both mutants walked through them: * two partitions holding the SAME offset range is what a key without the partition in each line collapses, and that is the ordinary state at the head of a fresh topic — not the swapped-ranges case, which the tuple sort already separates. * pyarrow's `group_by` emits in hash order, which coincides with first-occurrence order for three keys and stops coinciding at eight (it returns 0,1,2,3,4,5,7,6). Eight is therefore the size at which dropping the explicit ordering is visible.
The unit fixture for this branch had hoglake's error shape backwards — it put the sentence in `message` and the code in `detail`, which is why a matcher reading `.message` alone survived the whole suite. The branch is now driven end to end against the server that produces the real shape, and what it asserts is the number that used to be wrong: a process that publishes nothing reports zero rows written.
`_ensure_table` caches its resolved handle for the pod's life, and that handle is a NAME, not an incarnation. A drop+recreate underneath it was invisible to every guard at once: `_prepare`'s own `table.info()` adopts whatever the name resolves to now into pyhoglake's `_info`, and `prepare_append_files` then reads its default `expected_table_uuid` off that same refreshed value. The client's pre-flight, the server's `expected_table_uuid` and `_check_destination_still_ours` all compared fresh against fresh and passed, so the flush published to a table this pod had never reconciled, under an idempotency key naming the dead one. The consequences were each silent: the README's "the commit is refused" row was false for that window, cross-restart dedup was off for the straddling flush, and `_reconcile_specs` never ran against the recreated table at all. The sink now holds the incarnation it resolved AND reconciled (`_table_uuid`), derives the key from it, and pins `prepare_append_files` to it. pyhoglake's pre-flight then fires before the first upload, and `IncarnationChangedError` is retryable — so reset_caches re-resolves, `_reconcile_specs` runs against the live table, and the rebuilt flush publishes under a key that names it. The pre-flight refusal gets its own arm in the orphan accounting: it raises before any upload, so counting one there sent an operator sweeping for objects that were never written.
`_is_answered_refusal` said "409 or 422", but its 422 case could never execute: pyhoglake maps every wire 422 to `ValidationError`, which the preceding `except ValidationError` already handled with its own `_discard_prepared`. Two arms stating one rule, one of them dead — a mutant narrowing the tuple to `(409,)` survived the whole suite. Collapsed to a single `except HoglakeError`, with the reused-key check as an isinstance test inside it. The rule about what a prepared payload survives is now written down exactly once, and the parametrized 422 case in test_an_answered_refusal_drops_the_payload_and_re_resolves kills the narrowing mutant.
`_prepare` null-fills against `info.columns` and then selects names from
that same object, so the select can only ever narrow — no KeyError can
come out of it. The self-heal's `except (ValidationError, KeyError)` and
`_is_alignment_refusal`'s KeyError arm were therefore dead, and their
comments claimed the opposite ("the alignment's select reports a missing
column that way"). A mutant deleting the arm survived the suite.
Both come off, and `test_column_added_between_align_and_prepare_is_null_filled`
— which exercises exactly the column-arrives-between-resolves case that
used to raise — now says why the arm is gone.
Written on every prepare and cleared on every resolve, read by nothing. The replay is recognized by `_prepared_offsets` and the key the server judges rides on the payload itself, so this was a third copy of an identity with no reader — the kind that goes stale silently and then gets believed.
Each of these had a mutant that survived the whole suite, i.e. the property the code was written for was not being checked anywhere: * the replay match on `_prepared_offsets` — re-deriving the key instead consults an incarnation the retry path deliberately has not re-resolved, so a drop+recreate under a held payload reads as "a different flush" and uploads a second copy of rows whose first commit may have landed. Pinned by replaying across a recreation and asserting exactly one prepare. * `_live_table`'s "deliberately NOT cached" invariant — pinned both directly and by its consequence: the next ordinary flush still goes through `_ensure_table`'s resolve-and-reconcile. * the partial-upload orphan count `max(0, len(files) - 1)` — the only prepare-refusal test was single-file, which collapses `0`, `N-1` and `N` to the same number. A three-file fanout separates them. * an empty offset tuple as an identity — unreachable from main.py today, which is why it needs a test rather than a caller: under the mutant two consecutive anonymous flushes replay each other's payload.
Three statements the delta left behind: * AGENT.md said the flush identity was "(topic, partition, highest offset) triples" and pointed at `_flush_key` "for why the offset range is a sound identity". Both ends of each range have been in the key since it started naming the rows it publishes, and `_flush_key` now argues the incarnation half as well. Corrected to quadruples, and the pointers now go where the argument actually lives. * The jitter added to `_retry_delay` went into the SHARED helper, so the DuckLake ladder became 1.0-1.25s / 2.0-2.5s while the README still said "1s, 2s". Keeping it shared, and saying so: DuckLake pods contend for the same Postgres catalog commit lock and wake in the same lockstep, the spread is bounded to 25% (a 3.75s ladder at worst, under the same 30s ceiling), and a hoglake-only curve would be a second thing to keep in step for no benefit. The README row and the `_retry_delay` docstring now agree, and a test pins the jitter on a hookless DuckLake-shaped sink. * `_accept_already_published` named only differing FILTERS as the collision residual. The key carries no broker, cluster or consumer-group identity at all, so two pipelines on different Kafka clusters that share a catalog, namespace, table and topic NAME collide identically. Widened, with the reasoning for leaving broker identity out: millpond has no stable cluster id — BROKER_SOURCE is a free-text metrics label, bootstrap servers are a rotated address list — so every candidate is a config string whose edit would re-randomize every in-flight flush's identity, which is the hazard _IDEMPOTENCY_NAMESPACE is pinned against.
The drop-only test could not reach this window: the sink's refusal there comes from the name not resolving at all. Once the name resolves again, `_prepare`'s `table.info()` adopts the new incarnation into the pyhoglake handle and a defaulted `expected_table_uuid` reads off that same refreshed value, so every check compared fresh against fresh and the flush published across the seam. Drops, recreates under the same name, and asserts the live server refuses the stale-handle flush (retryably, with zero rows in the recreated table) rather than publishing into it. Fails without the incarnation pin: "DID NOT RAISE IncarnationChangedError".
The delivery table's recreation row claimed a refusal the code did not deliver for the stale-handle half of the window; it does now, so the row says which guard catches which half and what the retry re-reconciles. The orphan paragraph's partial-upload sentence carried the single-file case as an exception. It is the same rule seen at N=1: pyhoglake validates file i before uploading it, so a validation refusal on N files left at most N-1 behind.
The orphan counter is the whole observability story for parquet the
lake will never reclaim, so a phantom in it costs an operator a sweep
for objects that do not exist. Both prepare-side arms produced them.
The ValidationError arm counted `len(files) - 1`. pyhoglake validates
file i before it uploads file i, and every file in this fanout is a
slice of one aligned table -- one schema, one partition arity, no
empty groups -- so each of its three prepare-side refusals is a
property of the whole set and fires at index 0. Uploaded is always 0
and the count was always N-1. The concurrent add_column race that
`write()` self-heals trips exactly this refusal, so a flush that then
SUCCEEDED still reported orphans.
The generic arm counted `len(files)` for failures with no bytes
written at all: the key's UUID parse, the read-snapshot refresh and
the incarnation pre-flight all precede the upload loop, and the
uploads are pyarrow's (OSError), never HoglakeError or httpx. A
convoyed catalog answering 503 through eight retries of an eight-way
fanout booked up to 64 orphans for zero objects.
What is left is the honest unknown: an OSError partway through the
fanout, where pyhoglake reports no progress. Any count there is partly
invented, so the uncertainty moves to the log -- which can say "up to"
and name the `{idempotency_key}/` prefix that makes the sweep
decidable -- and the metric stays sound. It undercounts that one case
rather than overcounting every other.
The README called it an "upper bound" and said it counts what millpond
"knows it orphaned"; the metrics comment still carried the
pre-correction wording ("except in the single-file case") and had
drifted from the README besides. Both described a counter that reports
storage that was never written.
It is now exact for the paths it covers -- every one of which has a
prepared payload behind it, so the files are known to exist and known
to number len(files) -- and blind on two named paths: an S3 failure
partway through the fanout, and a SIGKILL between prepare and commit.
The rule is stated once, in both places, in the same direction: it may
miss an orphan, it never invents one.
The README's "retries do not add to this" was true only once a payload
is held. A flush that fails inside `prepare` has nothing to hold, so
every attempt re-uploads -- and those uploads are in the blind spot,
which is why that case gets a log naming the prefix to sweep.
`_is_alignment_refusal`'s `isinstance(exc, ValidationError)` check went dead when the call site narrowed to `except ValidationError`, and the signature still advertised a `BaseException` it never receives. A mutation that widened the guard survived both suites, because there is no input that reaches it. The type now says what arrives. Also corrects `_flush_key`'s claim about its own table_uuid. The docstring said deriving the key from `self._table_uuid` is what makes a drop+recreate under the cached handle REFUSED. It is not: the refusal is the `expected_table_uuid` pin in `_prepare`, and any flush whose key reaches the server has already cleared that pre-flight. The derivation is still right -- it keeps the key and the pin named from one source by construction rather than by a lower guard firing -- and it is free, so it stays; only the claim changes.
1.1.1 stamps uploaded_files/uploaded_uris on every exception leaving prepare_append_files. The cooldown carve-out already covered it (published 2026-09-18T16:24Z, before the 09-19 cutoff); the comment named 1.1.0.
pyhoglake now reports what it uploaded before it raised, so _prepare's
three exception arms collapse into one that READS uploaded_files rather
than arguing about where in pyhoglake's loop each failure can fire. Both
of this branch's earlier orphan-counting defects were that argument being
wrong.
The log names the object uris, never the {idempotency_key}/ prefix. Object
names are {uuid4}-{index}.parquet under that prefix and a retry under the
same key writes new names beside the old ones, so sweeping the prefix
after a later attempt succeeds deletes live, committed files. The two
other orphan paths carry their paths in the payload already, so they name
them too — retracting the prefix left them with no way to be found.
The file that failed is in neither number: a failed close can leave a
truncated object, so the count is a lower bound and the log says so.
…p unit
README and the metric comment both described a counter with two blind
spots and told operators to sweep the {idempotency_key}/ prefix. The
counter is exact on every path now, and the prefix advice was actively
dangerous — a retry under the same key lands new object names beside the
old ones, so a prefix sweep after a later success deletes committed files.
Keeps the two caveats that survive: the file a prepare failed ON may be in
storage truncated, and the getattr default books zero when the stamp is
missing.
The integration tests pin the pyhoglake contract against the real library
and real objects in MinIO — a mid-fanout failure's count is checked
against the bucket, not against another fake.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Millpond can write the firehose into a hoglake catalog. The destination seam removed with the Iceberg backend is restored from the
final-icebergtag:make_sink(cfg)dispatches onMILLPOND_DESTINATION, a pod writes to one destination for its lifetime, andducklakestays the default — deployed pipelines see no change beyond the retry note below.Events land as text.
MILLPOND_VARIANT_COLUMNStogether withdestination=hoglakeis a startup refusal rather than a silent half-write: there is no dual-written_variantcompanion on this path. When the variant story is settled, the tables get dropped and recreated with the blob column typed.Delivery semantics
Appends go through pyhoglake's prepared-commit API with an idempotency key naming
catalog/namespace/table/table_uuidplus the fulltopic:partition:first-lastrange per partition. The prepared payload is held across retries, so a retry re-sends the same registration rather than re-uploading, and the server adjudicates from its commit receipt.The guarantee, stated as the code behaves rather than as one would wish:
Kafka offsets commit only after a successful write, unchanged from the DuckLake path.
Also in here
HOGLAKE_PARTITION_BY/MILLPOND_SORT_BYconfig, verified as a post-condition, and reconciled on every resolve — a live/configured mismatch fails loudly and repeatedly instead of silently writing an unpartitioned table. Flipping the destination with a DuckLake partition var set but no hoglake one is refused at startup.Retry-Afterhandling are hoglake-specific and configurable.millpond_hoglake_files_written_total,_orphaned_files_total,_commit_replays_total. The orphan counter is sound in one direction — it may miss an orphan, it never invents one — with the two blind spots named in the README and the metric comment.Tests
1087 unit, 65 in-memory integration, 28 integration against a real hoglake server (bootstrap races, concurrent DDL, OCC retry, incarnation change, lost commit response, new-process replay at matching and shifted boundaries), 6 end-to-end through
main.pyfrom a broker to the catalog. The docker-gated suites are wired into CI withMILLPOND_REQUIRE_DOCKER_STACK=1, so a missing image fails rather than silently skips; the server image is pinned by digest.Six adversarial review passes ran over this branch — including a lead-QE and principal-SWE panel — each executing against a live server. They found, in order: a duplication defect on a lost commit response, a data-loss defect where a reused idempotency key was counted as success, an idempotency key that named a dead table incarnation, and an orphan counter that booked orphans that were never written. All are fixed with red-first evidence and mutation-tested coverage.
One change to the deployed DuckLake path
Retry backoff now carries up to 25% upward jitter, shared by both destinations: the DuckLake ladder becomes 1.0–1.25s then 2.0–2.5s. Pods contending for the same catalog commit lock otherwise wake in lockstep. Bounded, documented, and pinned by a test — but it is a behaviour change to running pipelines and should be an explicit call, not a side effect.
Deploy-side, not in this PR
Chart values for a dev instance, IRSA versus the static
HOGLAKE_S3_*keys the sink requires today, and whether the target catalog should have retention and compaction configured before ingest traffic arrives (partitioned fanout appends feed compaction debt immediately).