Repository navigation
kkv-client-compatibility - #6
Merged
Merged
Conversation
rustup in the user's home for the toolchain rust-toolchain.toml pins, and one apt-get line for the rest: cmake, make and g++ for librdkafka's CMake build, pkg-config, the libcurl headers librdkafka 2.12 compiles in for OIDC, and the Dockerfile builder's libssl, libsasl2, libzstd and liblz4, plus zlib for gzip-compressed topics. Verified on Ubuntu 24.04: with these installed, cargo build --workspace --all-targets and the unit tests pass. AGENTS.md said the builder needs librdkafka-dev and links dynamically; the crate builds librdkafka with cmake-build, so it points at the README now. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Blob encryption on S3 destinations uses Parquet modular encryption, which parquet-rs ships behind its `encryption` feature from 56 on; 60 is the version whose output DuckDB 1.5.5 is proven to read with `PRAGMA add_parquet_key` (the observability compactor's interop). The upgrade needs no source change: the ArrowWriter/ParquetRecordBatchReader API mirror-envelope uses is unchanged, and all workspace tests pass. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
CI runs `cargo clippy -- -Dwarnings` on a floating `stable`, so a new lint fails the build without a code change: clippy 1.99 flags async_trait's expansion (a #[must_use] fn returning a boxed Future) as double_must_use on every #[async_trait] trait (Source, Sink, Notifier, DnsAResolver and the e2e provisioner traits). The lint is about the macro's output, not our code, so it is allowed on those traits only. Pinning the toolchain keeps the next lint from breaking an unchanged tree; bump it deliberately. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
rdkafka was built with default features off and only `cmake-build`, so librdkafka came out with WITH_ZLIB=0 WITH_ZSTD=0. A producer with compression.type=gzip or zstd (the Java and Quarkus defaults on our topics) makes every fetch of its batches fail with "Not implemented"; the mirror exits and fails on the same batch after every restart. Enable `libz-static` and `zstd`. Both link statically (zlib from libz-sys, zstd from zstd-sys), so the binary still needs only glibc and libgcc and runs on distroless/cc; plain `libz` would link libz.so, which that image does not ship. TLS and SASL stay off; add `ssl` when a mirrored topic is behind a TLS or SASL listener. A unit test asserts librdkafka's builtin.features lists gzip, zstd, snappy and lz4. The Dockerfile comment claimed "full feature support" from packages the build never used; it now says what is linked. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The legacy Java worker set enable.idempotence=false on purpose (its commit history cites "we dont retransmit" and old timestamps), so idempotence stays off. What breaks the offset-preserving design is librdkafka's default unlimited retries: a produce whose ack is lost is retried and lands twice; the delivery report then says N+1, the mirror exits, the restart seeks to the destination's high watermark N+2, and source record N+1 is never mirrored while N is duplicated, with nothing to notice it. Set message.send.max.retries=0. Any produce failure now ends the mirror before a duplicate can exist, and the gate on restart is the one retry: the destination high watermark tells whether the record landed. This matches "we dont retransmit" and keeps acks=all, one request in flight and idempotence off as the legacy deployment has them. The settings move into producer_config() with a unit test that reads them back from librdkafka. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ation Every enabled mirror got a CacheBinding, every FS/S3 sink with a binding downloaded and decoded its whole append-mode archive at open (100 blobs gave 100 GETs), and the cache slot then kept a copy of every key's latest value, whether or not the mirror served /cache/v1. An archive of 50 GB meant a 50 GB download and an OOM per restart, and a cache could not start while S3 was down. Now: - Blob sinks have no cache binding and read object names only at open (FilesystemSinkConfig/S3SinkConfig lose `cache`; a test counts GETs). - Only mirrors with http-access keep values. Every other mirror registers a progress-only slot (register_progress_only): it still advances last_applied_offset against the bootstrap watermark, which is what 731d53f registered every mirror for (readiness body), and holds no values. - A cache is built the kkv way: the source is read from its low watermark. On a cache mirror that also has destinations the tee's resume floor is set to the low watermark, the destinations skip what they already hold (per-sink heads), and notify suppression (committed offset, or the bootstrap watermark on a fresh group) keeps the replay from firing webhooks the previous pod sent. This removes the bootstrap-from-snapshot sort fix and its test; that path no longer exists. The e2e harness drops the sink-level binding (it cannot run here: no Docker). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The FS and S3 sinks were near copies (buffer, flush triggers, compaction view, offset gate, three scan functions each, ~400 lines twice), so every delivery fix ahead (offset holes, the idle drift check, the idle time flush, encryption) would have had to be made twice and kept in step. mirror_fs::blob now holds everything that does not depend on where objects live: BlobSink<S: BlobStore> with the buffer, triggers, view, gate, flush bookkeeping and metrics, and validate_chain() over a listing (pure, unit-tested). A BlobStore lists, writes (atomically, refusing an existing name where the store can tell) and reads objects in one partition directory. FsStore keeps the tmp-file + fsync + rename write; S3Store keeps PutMode::Create. FilesystemSink is BlobSink<FsStore> (sync open, as before); S3Sink is a newtype over BlobSink<S3Store>. No behaviour change intended: every existing FS and S3 test passes unchanged. Two small unifications: S3 counts header bytes towards max-bytes as FS did, and both use FS's error wording. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
auto.offset.reset=earliest turned an out-of-range position into a silent jump to the earliest retained offset. The loop always seeks to an explicit position (the destination's next offset, or the low watermark for a cache), so an out-of-range position means retention deleted records the mirror never read: the mirror lagged or was down longer than the topic's retention. That must be an error, and it becomes essential with the next change: blob destinations will accept offset holes (compaction, transaction markers), so a silent jump would look like one more hole and lose the records in between unnoticed. With auto.offset.reset=error librdkafka reports Offset out of range through recv(), and the mirror ends. The legacy Java worker used earliest, but it also failed whenever the destination offset did not match, which a hole-tolerant destination no longer does. Settings move into consumer_config() with a unit test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…uses them Both failure scenarios matter: userstate is a compacted topic and its append-mode blob mirror would crash-loop on the first compaction hole. Append mode accepted no hole: the loop raised SourceGapAboveExpected on any offset above the expected one, and the sinks rejected it too. Kafka leaves holes where compaction removed records and at transaction commit/abort markers (librdkafka never delivers those offsets), so a compacted or transactional topic crash-looped forever. Blob objects now cover consumer positions: `from` is the previous object's `to` + 1 (0 first) and `to` is the last record buffered, so the chain of names stays contiguous and validate_chain is unchanged; each record keeps its true offset. Append and compaction mode now share the head arithmetic. A new Sink::allows_offset_holes() says whether a sink can take a hole: blob sinks and the notify-only sink yes, a tee only if every inner sink can, the Kafka destination no (its next offset is always its high watermark, it cannot reproduce a hole): there a hole ends the mirror with an error that says so. allows_compacted_source keeps its narrower meaning (resume past the low watermark: compaction mode only). Safe only because the previous commit made a lost position an error: librdkafka delivers in order, so a hole is never a lost record. Tests: the "gap must error" cases of the FS/S3 sinks, both sink matrices and the real-sink loop test now assert that holes are kept, named by position, read back with true offsets and resumed after; a matrix row checks that a flushed name covers a hole. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…inute, cheaply The idle drift check is deliberate (AGENTS.md invariant 2, 7e17927, 55dfe14: out-of-band writes and two-writer races must fail loudly; the legacy worker checked on every empty poll), but its cost grew with the archive. Every empty poll (2 s) ran a full paginated LIST and chain validation: 1 flush a minute gives 525k objects in a year, 525 LIST pages every 2 s per mirror, ~23M LISTs a day, and a sustained readdir on versitygw's POSIX backend. Startup also listed three times (sink open, tee open, run loop). This process is the destination's only writer, so after open the position stays in memory. The drift check now runs at most every DRIFT_CHECK_INTERVAL (60 s) on the idle path and asks two cheap questions: is the last object of our chain still there (one HEAD), and does anything sort after it (one LIST with start-after, empty when all is well)? A foreign object or a missing tail ends the mirror with an error naming the object. Startup lists once (open); the tee's and the loop's first calls fall inside the interval. BlobStore gains list_after (S3: list_with_offset = start-after) and exists (S3: HEAD). The Sink::next_expected_offset contract is updated: memory is allowed when verified this way. Request-count tests (a counting store) pin: no LIST or HEAD between checks with 200 objects, one HEAD and one empty LIST per check, and both drift cases. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
should_flush was only evaluated in write(), and the idle path (next_expected_offset on each empty poll) only ticked the daily trigger. README promises "whichever trips first", but a burst of 40 records at 17:00 on a Friday and silence until Monday left those records only in pod memory and Kafka all weekend: the blob backup was not current, and the "seconds since last flush" alert stayed silent. The idle path now flushes when max-time has elapsed, so the trigger holds within one poll timeout (2 s) whether or not records arrive. A test writes two records with max-time 20 ms, sees no object, waits, and sees 0-1 after the next idle call. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
kkv-v1 consumers re-read every notified key from the unprefixed
/cache/v1/raw/{key}; the Node client hard-codes it. Only `cache-v1-main`
mounts those paths. A config with notify and per-mirror `cache-v1` alone
validated, every refetch got a 404, the client's retryOnMissing gave up
after 5 s, and every update was silently dropped.
The validator now requires both cache-v1 and cache-v1-main on a mirror
with notify, with an error that says why. With at most one main per
process this also means at most one notify mirror per process, which is
how kkv was deployed (one kkv per target Service). Test fixtures and the
destination-flush example gain cache-v1-main.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
http-access required a filesystem/S3 destination and was forbidden on
mirrors without destinations, because the cache bootstrapped from the
blob chain ("a future bootstrap cache by replaying from broker mode is
conceivable but adds complexity; defer", WEBHOOKS.md). The cache now
bootstraps from the source's low watermark, so that reason is gone, and
a single-node deployment needs exactly this shape: kafka-keyvalue's
cache plus onupdate webhooks in a mirror with no destination, so that S3
being down or slow cannot stop the cache or the notifications.
Rules now:
- No destinations: the mirror must serve a cache (http-access) or
notify, or both; destination-shaped fields (format, compression, keys,
values, compaction, flush, timestamp-mode) are errors; notify uses
source-consume.
- http-access no longer depends on the destinations (also allowed with
Kafka-only destinations). Bytes keys with http-access stay an error.
- notify with http-access, or with destinations, requires cache-v1 +
cache-v1-main; notify without either (the notify-only feed of
WEBHOOKS.md) stays as it was.
The in-memory sink for such mirrors is renamed NoDestinationSink.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… on shutdown
Two supervisor changes, both needed for one Deployment that hosts a
cache mirror and blob mirrors (for example userstate and ops):
1. A mirror with neither a cache (http-access) nor notify is opened and
run again inside the process after a failure, with a backoff from 1 s
doubling to 60 s, each failure logged as an error and counted in
mirror_v3_mirror_restarts_total{topic,partition}. Such a mirror holds
no state but its destinations, and opening it again re-derives its
position from them, exactly what a process restart does; restarting
it alone keeps an S3 outage or a full disk from taking down the
caches and notifications of the other mirrors in the process. A cache
or notify mirror still ends the process on error: its in-memory state
belongs to the process. Each run's commit task and readiness poller
now end with the run (they held its consumer and would have leaked
across in-process restarts).
2. Graceful shutdown waits for every mirror. The supervisor returned as
soon as the first task ended and dropped the runtime, cancelling the
other mirrors' final flush PUTs and final commits; the README's
"shutdown flushes everything" was false with more than one mirror.
wait_mirrors() returns the first error at once and otherwise waits
for all; unit tests pin both.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
DEFAULT DECISION (decision pending; revertable as this commit): per-target delivery state that retries until acknowledged, replacing the matrix's `skip` drop for addresses that still exist. Why: a kkv-v1 push is one-shot and unacknowledged on the consumer side, and kafka-keyvalue's 2-6 replicas each pushed every update, so a push one replica lost was healed by a sibling's. One mirror-v3 replica has no sibling. With a config that has connrefused and 4xx `skip`, a consumer pod that was restarting, or not Ready yet, lost the update until the key changed again; with the defaults (`fail`) a hung pod blocked the consume loop for ~26 s per batch and then exited the process, taking the cache down. What changes for source-consume (new mirror_notify_kkv::delivery): - on_record and the debounce timer hand batches to a deliverer task; the consume loop never waits for HTTP. - Every address of every target keeps its own undelivered key set (highest offset with it). One POST per address is in flight; a failure keeps the set, merged with newer batches, for the next attempt (backoff from retry.backoff-ms, doubling, capped at 30 s). Any non-2xx is a failure. - `skip` now means: keep for that address and retry until it accepts or has been absent from discovery for 30 s; never fail the mirror. `fail` still ends the mirror after the retry budget, `accept` still counts as delivered: the six-outcome matrix stays the policy. - dns-a re-resolves at most once a second while there is work; a new address gets every batch from its first resolution on (with publishNotReadyAddresses on the target Service: before it is Ready, retried until it listens). A failed resolution keeps the known addresses; an empty one is "no target", not an error. - Bodies are cut below 64 KiB (Node body-parser limit 100 kB). - Nothing is sent before the mirror has caught up to its bootstrap watermark (new CacheState::has_caught_up, sticky): a notified consumer re-reads at once and would get 503. - The ack (committed offset) is the lowest offset some address has not accepted, so a restart re-delivers exactly the undelivered window; a later batch can no longer be acked past an earlier failed one. - Graceful shutdown gives pending keys 10 s. - Batch tails go out on the debounce timer without waiting for another record; a test pins it. New metrics: mirror_v3_notify_undelivered_keys and mirror_v3_notify_target_addresses; batches_total results ok|retry|fail. The notifier refuses to build for a mirror not registered in the cache state. The notify tests are rewritten to wait for asynchronous delivery and pin the new contract (per-address merge and retry, the late listener, scale-up, empty and failed resolution, no ack past an undelivered batch, on_record never waiting). WEBHOOKS.md documents it. Destination-flush keeps its one-shot dispatch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ution follows connrefused In the one-shot dispatch the destination-flush trigger still uses, a dns-a resolution error returned straight out of the retry loop with `?` and an empty answer was an error, so both bypassed the outcome policy (contradicting WEBHOOKS.md's failure table) and ended the mirror. A target Service scaled to zero then made every restart fail again on the same event. Now a failed resolution is the `connrefused` outcome (retry, then its final action) and an empty one means there is nothing to deliver. A test drives a dns-a flush dispatcher with an empty resolver. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…ffsets header
apply_record published the new offset before updating the view, under
two locks, and /raw read the value first and the offsets afterwards. A
probe found about two thirds of reads with an x-kkv-last-seen-offsets
header newer than the value in the body. A writer that produced K at
offset M and then called get(K, {requireOffset: M}) (the Node client's
read-your-write) could get the old value with header M and accept it as
fresh; the client then records M for K and drops the other pushes for
it, so the replica stays stale until K changes again.
The view and the per-partition offsets now share one lock in each slot;
apply_record updates both under it, and CacheState::read_value returns a
key's value, the offset of the record that wrote it, and the offsets of
the same state. /raw/{key} uses it and adds the response header
x-kkv-key-offset (additive; kafka-keyvalue clients ignore it). A unit
test races three readers against a writer of one key and requires every
read's value, its offset and the header to agree; a handler test pins
the headers.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
DEFAULT DECISION (decision pending; revertable as this commit). Non-sticky readiness was a deliberate request (66cb8b9, 4cd7697), so this reverses a documented choice; the merge message for #3 still listed the gate as "currently sticky-true", and what was never decided is this consequence: - every /cache/v1 read was gated on the current status with a lag tolerance of 0, so a record produced but not yet applied when the 2 s poller looked turned reads into 503. Client 1.9.3 counts that as a failure and keeps the value stale 5-60 s; 1.8.3 loses the update in an unhandled rejection. - the same flap drops a single replica from its Service's endpoints, which is connection refused for every reader. - /q/health/live did not exist, but the kkv manifests probe it, so a reused manifest gets its pod killed every ~30 s. Now, as kafka-keyvalue (ConsumeLoopReadiness): /cache/v1 routes and /q/health/ready answer 503 until the mirror has caught up to its source's high watermark at startup, and 200 from then on (CacheState::has_caught_up, sticky because the applied offset only grows). Only mirrors that serve /cache/v1 gate /q/health/ready; blob mirrors are listed (gates_readiness: false) so S3 being down cannot take the cache out of service. The enum status, lag and assignment stay in the body, lag also as mirror_v3_source_lag_offsets{topic,partition}. The aggregate loses its "degraded" value. /q/health/live answers 200 while serving. Tests replace the warming-vs-degraded case with: sticky after catch-up with lag in the body and reads still 200; a blob mirror not gating; the live endpoint. OpenAPI regenerated; README and the drop-in doc updated. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
S3 destinations may be set up with least-privilege identities: the writer identity may only PutObject, and the lister identity may only list keys (ListBucket, ListBucketVersions) and read no object. A HEAD of the chain's last object needs s3:GetObject, so the drift check of 6389ada would get 403 there. The sink now keeps the last two names of its chain and lists once from the second-to-last: the result must contain the last object (else it is gone) and no other blob name (else something else wrote). Same cost bound as before (one request, one entry), and it needs list rights only. BlobStore loses exists(). The request-cost test now pins no HEAD and one LIST of one entry. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…_* discovery
The S3 client came from AmazonS3Builder::from_env(), so any AWS_*
variable in the pod silently changed behaviour
(AWS_CONDITIONAL_PUT=disabled makes every flush fail,
AWS_UNSIGNED_PAYLOAD, AWS_VIRTUAL_HOSTED_STYLE_REQUEST, AWS_ENDPOINT),
and missing keys fell back to instance-metadata probing.
It also could not work with least-privilege bucket identities: the
writer identity may only PutObject and the lister may only list keys.
mirror-v3 needs both: list to resume and for the drift check, put to
write.
An S3 destination now requires `credentials: { write, read }`, each
naming the two environment variables that hold its key id and secret
(keys never go in the config). The client is built with
AmazonS3Builder::new() from region, bucket, endpoint and those keys
only; a missing variable is a startup error naming it. S3SinkConfig and
S3Store take a read store (LIST, and GET of the latest snapshot in
compaction mode) and a write store (PUT). Examples, tests, schema and
README updated; one key for both is configured by naming the same
variables twice.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Chain validation skipped names that did not parse with the configured extension, and the drift check ignored them. That hides a misconfiguration (two mirrors sharing a prefix, a stray upload), and with the encrypted naming ahead it is dangerous: an object this version does not read as a blob would be skipped, the resume position would come out too low, the mirror would rewrite those offsets under another name, and the next restart would crash-loop on overlap. validate_chain now errors on any name that is not a blob name of this mirror's format, naming it; the drift check treats any name after the chain's end as foreign. (The filesystem store still leaves out its own temporary files.) A unit test pins the error. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The TDD sketch for this spec, kept as an #[ignore]d stub in palette_demo.rs since fb322af, is now green under the name it proposed. If the topic is recreated (a broker that lost its data while the bucket kept its objects, a re-provision that keeps one and not the other) or truncated by an unclean leader election, the destination's next offset is above the source's high watermark. With a lower high watermark the source delivered from below the expected offset and the mirror crash-looped on a misleading SourceWentBackwards; once producers refilled the topic past the destination's position, the new topic's records were appended to the old topic's chain with no error at all. run_mirror now reads source.high_watermark() at startup and returns MirrorError::SinkAheadOfSource { sink_offset, source_hwm } with a message that says what probably happened and what to do. KafkaSource implements high_watermark (it had only the trait default). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…es are errors - Every MIRROR_V3_* variable (cache port, metrics port, readiness lag and poll, commit interval, heartbeat) silently fell back to its default when it did not parse, so a typo changed behaviour without a trace. They are now read once at startup into Knobs; unset is the default, an unparseable value is an error naming the variable. - A metrics exporter that could not bind logged a warning and the mirror ran unmonitored; MIRROR_V3_METRICS_PORT=0 disabled metrics. Binding is now a startup error, and the 0 switch is gone. - A cache HTTP server that failed (port in use) requested a graceful shutdown, so the process exited 0. The server task is now waited for with the mirrors: its failure ends the process non-zero. - A SIGTERM handler that could not be installed was skipped silently; now a startup error. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…not a delivery
- CacheState::apply_record silently ignored an unknown mirror name ("so
a future refactor can't crash the consume loop"). That is a wiring
bug; it now panics naming the mirror.
- It silently skipped records whose key was null or not UTF-8. As
kafka-keyvalue does with a null key, the cache still skips them, but
logs a warning and counts mirror_v3_cache_skipped_records_total
{reason=null_key|non_utf8_key}. The offset still advances, so
readiness is not stuck behind such a record.
- The notifier sent a null key as "" and a non-UTF-8 key lossily; no
consumer can re-read either as /raw/{key}. Like kafka-keyvalue, such
records are not notified (mirror_v3_notify_skipped_records_total).
- A 1xx final status counted as success; it is now a failure (5xx
bucket); reqwest consumes interim responses itself.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…gauge Both are items no requirement was found for, and no dependency turned up while reworking delivery: - notify.targets[].path (kkv has the knob; no config we deploy sets it). A target URL that has a path still uses it; a URL without one still gets /kafka-keyvalue/v1/updates. - mirror_v3_destination_offset_inflight_retry (a placeholder "so dashboards can be built ahead of the retry implementation"; it was always 0). The destination-trouble signal it was meant to be now exists as mirror_v3_mirror_restarts_total, which the README's metrics table and PromQL example use instead. (mirror_v3_notify_inflight_retry is live and stays.) flush.daily stays: the optional daily-at-UTC flush trigger (b4e45d8) is a wanted feature, and since bef9ffc it is honoured on the idle path too. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…a head going back is drift
- librdkafka reports an unknown watermark as -1, and every caller made
it offset 0 with .max(0): a bootstrap watermark of 0 makes a cache
ready at once and suppresses no notification, a low watermark of 0
hides a trimmed topic. fetch_*_watermark now return u64 and fail on a
negative value; the destination's high watermark likewise.
- A committed offset that was neither a number nor Invalid ("no commit")
became "no commit" too; now an error naming the group and partition.
- TeeSink kept its in-memory head when an inner sink reported a lower
next offset ("transient, e.g. a partial flush seen mid-rename"). The
blob sinks no longer re-scan on that path and check their destination
themselves, so a lower value means the destination lost what it had
accepted: an error naming the inner sink.
- `status` reports a negative lag when a destination is ahead of its
source instead of clamping it to 0.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A destination's directory is <root or bucket/prefix>/<name>/<partition>; the topic is not in it, and name uniqueness was only checked within one mirror. Two mirrors (topic A and topic B, both `name: archive`, same bucket) wrote one chain: the two-writer corruption that the single-writer invariant exists to prevent, found only at the next restart as CorruptChain. The validator now rejects two blob destinations, across all mirrors, that resolve to the same directory, naming both mirrors and the path. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Blob encryption, first part: what encrypts and what holds the keys,
without wiring it to a destination.
- parquet gains its `encryption` feature (ring). encode_batch_encrypted
and decode_batch_encrypted write and read uniform encryption: footer
and every column under one 32-byte key, AES-GCM, encrypted footer.
That is the layout DuckDB 1.5.5 reads with PRAGMA add_parquet_key and
read_parquet(..., encryption_config = {footer_key: ...}), which the
observability compactor proved with parquet-rs 60. The encrypted
footer also hides key/value min/max statistics, which a plaintext
footer leaks.
- keys::Keyring loads a keys directory in the observability compactor's
layout (PARQUET_KEYS_DIR), so operators handle one key layout: one
file per key, the name the key id [a-z0-9][a-z0-9-]{0,31}, the content
the base64 of 32 bytes; dot entries (a Secret mount's own) skipped,
anything else invalid is an error naming the file, never its content.
Only 32-byte keys, for the compactor's reason (DuckDB takes 24/32-char
base64 of shorter keys verbatim). ParquetKey's Debug is redacted.
Tests: round trip; a plain reader and a wrong key both fail; no key,
value or column name in clear text in the file; keyring loading and
rejection without printing content. A DuckDB interop test (ignored
unless MIRROR_V3_DUCKDB names a binary) passed here with a userspace
DuckDB binary: rows read with the key, refused without it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… Parquet keys
The rule: encryption is explicit per S3 destination, `none` written out,
and a missing `encryption` key is a config error. Blob encryption:
Parquet modular encryption, key id in the object name, the observability
compactor's config shape so operators handle one key layout.
- S3Destination.encryption is required: `none`, or
`{ keys-dir, key-id }`. An encrypted destination needs `format:
parquet` written out (ndjson cannot be encrypted, and a default must
not decide it); key-id must be a key id.
- At startup the keys directory is loaded (Keyring) and the active key
must be in it; errors name the directory and file, never key bytes.
- New blobs are encrypted with key-id and named
<from>-<to>.k-<key id>.parquet (naming::blob_filename /
parse_blob_name; the compactor's `*.k-{id}.parquet` glob). The chain
validation reads plain and keyed names alike, so turning encryption on
and rotating keys keep one contiguous chain without reading content.
A blob's content is read with the key its name carries (compaction
mode's snapshot at open); a missing key, or `encryption: none` over an
encrypted snapshot, is an error that says which key.
Tests: plain then k1 then k2 blobs in one chain with the expected names,
each readable only with its own key; an encrypted compaction snapshot
rebuilt on reopen and refused without the key; a missing active key;
config parsing and validation. README documents the layout, rotation
and the DuckDB read.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…wo topics The ops topic mirrored to blobs next to the userstate backup, in the config shape a single-node deployment uses. Three mirrors in one process: - userstate: /cache/v1 and kkv-v1 notify, no destination, so S3 cannot stop the cache or the notifications; every target failure outcome is `skip` (kept for that address), a redirect fails. - userstate-backup and operations-backup: append-mode encrypted Parquet to S3 with least-privilege writer/lister identities and the compactor's key layout. Append mode, not compaction: it keeps every record, needs no GET (the lister cannot read objects), and accepts the compacted topic's offset holes. validate_accepts_each_example covers it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Found in a deployment: a pod with runAsNonRoot refused the image ("image
has non-numeric user (nonroot), cannot verify user is non-root"), so the
Dockerfile now says USER 65532:65532, distroless's nonroot by number.
And tracing wrote ANSI colour codes to the pod log; colours are now on
only when stderr is a terminal.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…process) The README still said any task failure collapses the whole process and that shutdown flushes; since b49b232 a mirror without cache or notify is reopened in the process with a backoff, and graceful shutdown waits for every mirror. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
c7750ca made the idle check for foreign objects run at most once a minute (DRIFT_CHECK_INTERVAL), but adversarial_fs still waited 15 s for the mirror to notice a dropped file, so the e2e suite failed: the branch's e2e had not been run (0309bc8: "no Docker"). The harness gains spawn_kafka_to_filesystem_with_drift_check; the plain spawn keeps the production interval. The test checks every second, which still goes through the idle path that runs the check. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2e78a14 compared the source high watermark with the sink's next_expected_offset, which for a tee is the minimum over its inner sinks, lowered further by a resume floor: the low watermark on a cache mirror with destinations, the committed offset on a notify mirror that re-delivers its window. On a recreated topic those mirrors never tripped the check; the tee skipped each destination up to its old position and then appended the new topic's records to the old chain without an error, which is the offset mismatch the check exists to stop. A tee with one lagging and one leading destination passed too. The run loop now checks Sink::furthest_next_offset, falling back to next_expected_offset when a sink returns None (the default: a second query could flush or run the drift check). TeeSink returns the highest of its inner heads, ignoring the resume cursor. A palette test drives a tee with heads 40 and 150, a floor at 0 and a source at 100. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…r in process c6af396 reopened a destination-only mirror after any failure, so an offset mismatch, an object the mirror did not write, a corrupt chain, a lost source position or a configuration that cannot work were retried every 60 s forever, with /q/health/live at 200 and only mirror_v3_mirror_restarts_total to show it. The legacy Java worker exited on an offset mismatch; a crash loop is the signal deployments alert on, and mirror-v3 prefers it over retrying what cannot heal. Now only a transient failure is retried in process. Errors say which kind they are: - SinkError::Transport is the destination unreachable or answering an error. New SinkError::Inconsistent is a destination that contradicts the mirror (the drift check's foreign or missing object, an object that already exists, a head that went back, the align preconditions), SinkError::Rejected a record it cannot take (no UTF-8 key in compaction mode, an encode failure). - New SourceError::PositionLost is librdkafka's AutoOffsetReset or OffsetOutOfRange under auto.offset.reset=error (2f4a1cb); it was a Transport string. - The tee kept every inner error as Transport; SinkError::context now keeps the kind and turns an inner UnexpectedPosition (a Kafka destination's offset mismatch) into Inconsistent with the inner sink's name. - The supervisor kept errors as strings (anyhow!("{e}")); the run result and the tee open now keep their types, and is_transient reads them: MirrorError/SinkError/SourceError::is_transient, and BlobError::Store at open. Anything else ends the process. An object that already exists at write is now reported as such instead of an UnexpectedPosition with expected == actual. Tests: the classification in mirror-bin, the tee keeping a mismatch fatal, and the sink matrices' align preconditions as Inconsistent. README's supervision paragraph and invariant follow. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
A process may run two mirrors of one partition, as the kkv cache plus backup example does (510720e): a cache mirror and a backup mirror of the same topic. Every metric was labelled with topic and partition only, so their mirror_v3_destination_offset_verified, mirror_v3_destination_records_total, mirror_v3_source_lag_offsets, mirror_v3_mirror_restarts_total and the other per-mirror series were one series that both wrote, and the gauges showed whichever mirror wrote last. MetricLabels gains `mirror` (the mirror's name), current_labels returns it, and the destination, source, restart and cache metrics carry it. README said the name is deliberately not a label ("operator-chosen metadata, not a data-stream dimension"); that holds for joins, which stay on (topic, partition), and README now says why the label is there. The notify metrics are left as they are: the notify delivery code is about to be replaced (2f3be23 is to be reverted). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Kafka keeps a committed offset per (group, topic, partition). Two mirrors of one partition that name one `source.group-id`, say a cache mirror and a backup mirror of the same topic, each commit their own progress there: consumer-group lag shows whichever committed last, and a notify mirror's re-delivery after a restart resumes from the other mirror's offset, re-sending or skipping notifications. The validator rejects that pair, naming both mirrors, the group and the partition. The same group on another topic stays valid. The default group (mirror-v3-<name>) is now one helper, Mirror::effective_group_id, which the supervisor and the validator share. The group-id doc no longer says it is for monitoring only: it is also where notify re-delivery resumes. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
7eaa0fe's message says "a committed offset that was neither a number nor Invalid ("no commit") became "no commit" too; now an error naming the group and partition", but the commit did not change fetch_committed_offset, which still read every other value as None. That turns a returning deploy into a fresh one: notify suppression starts from the high watermark and the un-acked window is not re-delivered. The match moves into committed_position: a non-negative offset is the position, Invalid is no commit, anything else (a negative number, a logical offset) is an error naming the group, topic and partition. A unit test covers each case. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
02df9ca made every object in a destination directory one of its blobs or an error. object_store lists a key ending in `/` (the placeholder the GCS console, gcsfuse and some sync tools create for a folder, here `<prefix>/<name>/<partition>/`) as the directory's own path, whose filename() is the partition number: "0 is not a blob name" on every open, so one console click stopped the mirror for good. An object in a subdirectory was named by its last segment only, so the error named a file that is not where it says. The S3 store now names listed objects by their path below the partition directory: the directory's own placeholder is skipped, a nested object keeps its relative path (`sub/x`) and still fails the chain as foreign. Every other rule of 02df9ca is unchanged. A unit test covers the three cases; the filesystem store lists one directory and has neither. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…il accepted" This reverts commit 2f3be23, which was marked "DEFAULT DECISION (decision pending; revertable as this commit)". Decided: revert. Its premise is a single mirror-v3 replica replacing kafka-keyvalue's 2-6 for one Service. What it changed and what that cost: - Retries were pinned to one pod IP, kept for 30 s after the address left DNS. With `timeout` or `5xx` set to `fail`, one consumer pod rolling away exhausted the budget and ended the process, cache included; b66d8d3 had fixed exactly that by re-resolving each round. - `skip` meant "retry until accepted" and the committed offset was the lowest one some address had not accepted, so a pod answering 4xx for good (one that carries the target label but serves no POST route) pinned the committed offset and grew the undelivered set. Restored: endpoint-level retry rounds that re-resolve the address set (b66d8d3), serialized dispatch so an ack cannot pass an in-flight batch (f9f8830), terminal dispatch errors surfaced to the supervisor (dc637c1), and `skip` as drop-and-advance. Kept from the commits after it, resolved in the conflicts: - e8e56b7: a record without a UTF-8 key is not notified (counted in mirror_v3_notify_skipped_records_total); 1xx is not a delivery. - e614631: no `path` on a target (a test still set it). - a4240f9: an empty dns-a answer is no target, now for source-consume too, so the empty-address test expects delivery to succeed. - has_caught_up stays: sticky readiness (baaaa52) uses it. The deployment example's comment says what `skip` does again. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…uestion A cache that shares a mirror with destinations reads the source from its low watermark and skips each destination up to its own position (the tee's resume floor), so the destinations no longer set the source position; the cache fails together with its destinations; and the cache holds what the topic retains, not what was archived. The split shape (a cache mirror without destinations plus a backup mirror) avoids all three, and the validator could require it, but that would also rule out destination-flush notify and notify re-delivery through the tee. Recorded as an open question; the /cache/v1 bootstrap paragraph points to it and says that a cache mirror's failure ends the process. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
kafka-keyvalue's replicas healed each other's lost pushes; one mirror-v3 replica cannot. Records how delivery works today (retry rounds that re-resolve, serialized batches, skip drops, fail ends the process), what 2f3be23 tried (per address, off the consume loop, until accepted) and why 40cb69a reverted it, and the two directions left open: more reliable delivery from one replica with bounded retention per address, or several replicas of a destination-less cache and notify mirror, as kkv ran, with the backups deployed apart. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
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
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.
The last nodejs client update was Yolean/kafka-keyvalue#58