diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 25f2411..656e43f 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -40,6 +40,11 @@ jobs: with: components: rustfmt, clippy - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1 + with: + # PRs restore from main's cache but don't write their own — + # otherwise every (rustc, Cargo.lock) tuple per PR adds a + # ~500 MB entry and the per-repo cache budget blows up. + save-if: ${{ github.ref == 'refs/heads/main' }} - run: cargo fmt --all -- --check - run: cargo clippy --workspace --all-targets --locked -- -Dwarnings - name: unit + integration tests (no docker) @@ -56,11 +61,15 @@ jobs: run: sudo apt-get update && sudo apt-get install -y --no-install-recommends $LIBRDKAFKA_BUILD_DEPS - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable as of 2026-03-27 - uses: Swatinem/rust-cache@c19371144df3bb44fab255c43d04cbc2ab54d1c4 # v2.9.1 + with: + # See note in the `test` job: PRs restore but don't save, + # so PR runs don't proliferate the per-repo cache budget. + save-if: ${{ github.ref == 'refs/heads/main' }} - name: prepull e2e images run: | - docker pull quay.io/ogunalp/kafka-native:latest & - docker pull docker.io/redpandadata/redpanda:latest & - docker pull docker.io/versity/versitygw:latest & + docker pull quay.io/ogunalp/kafka-native:latest@sha256:9ae667c854cd25a86a2d263e0cd7352cd6ffa6395c2dec5f90ab9056cc60b14f & + docker pull docker.io/redpandadata/redpanda:v26.1.12 & + docker pull docker.io/versity/versitygw:v1.6.0 & docker pull ghcr.io/shopify/toxiproxy:2.12.0 & wait - name: cargo test mirror-e2e @@ -73,6 +82,9 @@ jobs: permissions: packages: write contents: read + # The prune-superseded-caches step at the end of this job + # deletes obsolete buildkit blobs via `gh cache delete`. + actions: write steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 - uses: docker/setup-qemu-action@ce360397dd3f832beb865e1373c09c0e9f86d70a # v4.0.0 @@ -87,7 +99,9 @@ jobs: - id: meta uses: docker/metadata-action@030e881283bb7a6894de51c315a6bfe6a94e05cf # v6.0.0 with: - images: ghcr.io/yolean/mirror-v3 + # repository_owner keeps pushes working from forks (e.g. + # YoleanAgents); metadata-action lowercases the owner. + images: ghcr.io/${{ github.repository_owner }}/mirror-v3 # Tag rules (sha tag is the full 40-char commit SHA, no prefix): # - push to main -> :main and :latest and : # - push to tag vX.Y.Z -> :vX.Y.Z (and :latest if default-branch) @@ -117,3 +131,40 @@ jobs: cache-from: type=gha cache-to: type=gha,mode=max provenance: false + - name: prune superseded main buildkit caches + # Buildx writes content-addressed blobs and an index per build. + # Blobs are NOT re-uploaded when their content is unchanged, so + # a stable base layer keeps the createdAt of the build that + # first uploaded it while still being referenced by every newer + # index. Pruning by creation time would therefore delete the + # most stable, still-referenced blobs first and send the next + # build partially cold. lastAccessedAt is the safe ordering: + # restores touch every blob they read, so anything the latest + # main build actually used sorts to the top and the deletions + # hit genuinely dead entries. Keep the top N most-recently- + # accessed buildkit-* / index-buildkit-* entries on + # refs/heads/main; delete the rest. + # + # N=200 is roughly 2-4 main builds' worth (one multi-arch build + # writes ~30-50 entries). Adjust if a single build ever pushes + # close to the limit — symptom is the *next* main build going + # cold despite this step running. + if: github.ref == 'refs/heads/main' && github.event_name == 'push' && success() + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + to_delete=$(gh cache list -R "$GITHUB_REPOSITORY" -L 1000 \ + --json id,key,ref,lastAccessedAt \ + --jq '[ .[] + | select(.ref == "refs/heads/main") + | select((.key | startswith("buildkit-")) or (.key | startswith("index-buildkit-"))) + ] + | sort_by(.lastAccessedAt) | reverse | .[200:] | .[].id') + if [ -n "${to_delete:-}" ]; then + count=$(echo "$to_delete" | wc -l | tr -d ' ') + echo "Pruning $count superseded buildkit cache entries on main" + echo "$to_delete" | xargs -I {} gh cache delete -R "$GITHUB_REPOSITORY" {} + else + echo "No superseded buildkit caches to prune" + fi diff --git a/.gitignore b/.gitignore index ea8c4bf..7eb055a 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,6 @@ /target + +# Local design memos / shelf notes — useful for human resumption +# of in-flight design work but not stable enough to commit. Files +# matching this pattern stay in the working tree, unversioned. +NOTES_*.md diff --git a/Cargo.toml b/Cargo.toml index 5eb443b..172b930 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,9 +17,9 @@ members = [ [workspace.package] version = "0.1.0" edition = "2021" -rust-version = "1.75" +rust-version = "1.88" license = "Apache-2.0" -repository = "https://github.com/YoleanAgents/mirror-v3" +repository = "https://github.com/Yolean/mirror-v3" [workspace.dependencies] mirror-cache = { path = "crates/mirror-cache" } diff --git a/DELIVERY_SEMANTICS_REVISIT.md b/DELIVERY_SEMANTICS_REVISIT.md new file mode 100644 index 0000000..c23c50c --- /dev/null +++ b/DELIVERY_SEMANTICS_REVISIT.md @@ -0,0 +1,217 @@ +# Delivery semantics revisit + +Re-derives the per-mirror delivery contract after auditing kkv, +Quarkus mirror-v3-worker, and our current Rust binary. Scoped to the +checkit deployment (`mirror-v3-worker` Pod, two mirrors) but the +recommended changes are general. + +## What we removed and why it was wrong + +When we rewrote mirror-v3 in Rust we set `enable.auto.commit=false` +on the source consumer and never call `commit*()`. The `group.id` is +still set (defaults to `mirror-v3-`) but is cosmetic manual +`assign()` doesn't use it for partition coordination and nothing +ever writes to `__consumer_offsets`. + +We thought this was safe because each sink has its own authoritative +"next expected offset": + +- Filesystem/S3: derived from the latest written file's offset. +- Kafka: derived from the destination topic's end offset (or its + own consumer group if running as a tee). +- `NotifyOnlySink`: in-memory `position`, reset to source + low-watermark on every restart. + +For sinks with **durable destination state** (the first two) this is +correct and gives exactly-once via destination idempotence. For the +notify path it isn't there is no durable destination state, so a +restart re-seeks to low-watermark and the suppression PR (`5ef7c9e`) +silently drops any notify for records between previous-shutdown and +new-startup. + +Both kkv and the Quarkus mirror-v3-worker avoided this gap by +letting SmallRye Reactive Messaging commit periodically (default: +throttled, 5 s) in the background. `enable.auto.commit=false` on +the Kafka client doesn't disable SmallRye's framework-level commits. +On restart `consumer.position()` returns the last committed offset +and processing resumes there. kkv's HWM-at-startup suppression then +fires only on the very first deployment of a fresh group. + +Our suppression PR replicates the *fresh-group* behavior on every +restart, because we never advance the persisted high-water mark. + +## checkit mirrors and what each one needs + +`mirror-v3-worker-config.yaml` runs two mirrors against the +`kafka-v2-read-bootstrap` cluster, partition 0: + +| Mirror | Destinations | `http-access` | `notify` | Required delivery semantics | +|-------------|---------------------------|-------------------|--------------|-------------------------------------------------------------| +| operations | kafka-v3, filesystem (GCS) | none | none today | Exactly-once on each destination (already correct) | +| userstate | filesystem (GCS) | `cache-v1` (+main once we migrate) | kkv-v1 to `kkv-userstate` consumers | Exactly-once on GCS; **at-least-once on notify** | + +The production symptom in dev2 (boards-v1 stuck on stale +org-membership after a mirror restart) is the `userstate` notify +path missing its delivery contract. + +## Recommended changes, in order + +### 1. Commit source-consumer offsets from mirror-v3 + +Add a `commit_through(offset: u64)` method to the `Source` trait. +For `KafkaSource` it calls `store_offsets` (per-partition) and a +periodic `commit_consumer_state(CommitMode::Async)` from a +background task (interval: 5 s, override via +`MIRROR_V3_OFFSET_COMMIT_INTERVAL_MS`). + +`enable.auto.commit=false` stays we keep manual control over what +gets committed. We are *not* switching to `subscribe()`; pod churn +is bounded by `replicas: 1 + strategy: Recreate`, so manual +`assign()` is still correct. + +### 2. Commit only after a sink/notify has accepted the offset + +The supervisor tracks one committed offset per mirror: + +- A mirror with a `notify` block: commit only after the notify + dispatcher's drain (source-consume) or flush-event POST + (destination-flush) has returned success. Commit at the highest + offset whose batch was accepted. +- A notify-only mirror: same the in-memory `NotifyOnlySink` is + not authoritative; the *notify dispatch* is. +- A mirror without notify (today: `operations`): commit at the + destination's flushed offset. This is observability / + external-lag-monitoring only; resume position still comes from + destination state. + +Result for `userstate`: a notify that hasn't been ACK'd by the +target consumer pod stays uncommitted; a restart resumes from the +last successfully-delivered batch and re-fires from there. That is +at-least-once. + +### 3. Replace HWM-at-startup suppression with committed-offset suppression + +Today (`CacheState::MirrorSlot.bootstrap_hwm`): captured fresh on +every restart, used as a threshold below which `KkvV1Notifier` +suppresses. + +After this change: +```rust +suppression_threshold = max( + last_committed_offset, // from the broker, on startup + bootstrap_hwm_if_no_commit, // fallback for first-ever deploy +) +``` + +- Existing deployments: `last_committed_offset` wins, and notify + resumes exactly where the previous pod left off. The + between-pods gap is closed. +- Fresh deployments: no commit yet, fall back to HWM-at-startup + matches kkv's "first deployment doesn't backfill historical + records" behavior. + +`cache-v1` view rebuild is unaffected the source consumer still +seeks to `low_watermark` (or to a `compaction: log` snapshot +offset) for any mirror with `http_access.cache_v1`. The committed +offset is only consulted to decide *what to suppress*, not where to +seek. + +### 4. Readiness driven by committed-offset lag + +Replace the current "captured HWM at startup" gauge with a per-mirror +freshness signal: + +``` +lag(mirror) = current_end_offset - last_committed_offset +ready(mirror) = lag(mirror) <= MIRROR_V3_READINESS_LAG (default 0) +``` + +The supervisor polls broker end-offsets every few seconds (cheap; +single `fetch_watermarks` per mirror). For mirrors with no notify +(commits are observability-only) we use the destination's flushed +offset instead of the committed group offset. + +`is_mirror_ready()` flips on the lag condition becoming true and +stays sticky-true (for now the dev2 thread also flagged wanting +re-degradation on source-state regressions, but that is a separate +PR). + +The kkv-compat `/q/health/ready` continues to AND together every +mirror's per-mirror flag. + +### 5. Use the same gate for `/cache/v1/{mirror}/...` + +Already aligned after the `cache-v1` rework (`0905f9d`): the +per-mirror cache routes already call `is_mirror_ready(mirror)`. +After (3) and (4), that flag is the lag-based gate and consumers +get a meaningful "still warming up" 503 across restarts. + +## What stays from the current suppression PR + +- `CacheState::MirrorSlot` and the per-mirror layout. +- The notify-side gate in `KkvV1Notifier::on_record` and + `FlushDispatcher::on_flushed`. +- The `mirror_v3_notify_suppressed_records_total` counter, but now + it fires for "below last-committed" rather than "below HWM at + startup" operator-visible difference: the counter only ticks + during fresh deployments, not on every restart. + +Effectively the suppression PR becomes the *first-deploy fallback* +and the new commit machinery is the *steady-state* mechanism. + +## Files we expect to touch + +- `crates/mirror-core/src/lib.rs`: `Source::commit_through` trait + method; `Sink::flushed_through` for the observability-only + commit path on no-notify mirrors. +- `crates/mirror-kafka/src/lib.rs`: implement + `commit_through` (store_offsets + periodic + `commit_consumer_state`); add a startup + `fetch_committed_offset()` helper. +- `crates/mirror-core/src/cache.rs`: add + `last_committed_offset: AtomicU64` next to `bootstrap_hwm`; + expose `record_committed(mirror, offset)` and update the + readiness predicate to use lag. +- `crates/mirror-notify-kkv/src/lib.rs`: after a successful batch + / flush dispatch, call `source.commit_through(batch.high_offset)` + via a handle the supervisor wires in. +- `crates/mirror-bin/src/main.rs`: at startup, per registered + mirror, `fetch_committed_offset()` and pass into + `register_mirror`. Spawn the periodic end-offset poller used by + readiness. +- `crates/mirror-cache/src/lib.rs`: no surface change; the + per-mirror routes already gate on `is_mirror_ready`. + +## Scope decisions + +In scope for the first PR after this revisit: + +- (1) (4) above, behind no feature flag (always on; fallback + behaves correctly on fresh groups). +- Tests: restart simulation against an in-process Kafka harness + showing zero dropped notify between pods. + +Out of scope (separate work): + +- Per-mirror, per-capability JSON on `/q/health/ready`. +- Re-degradation of the ready flag on source-state regressions. +- Multi-pod / rebalance handling we stay on `assign()` while + the worker is `replicas: 1 + strategy: Recreate`. + +## Rollout for checkit + +1. Land the change behind no flag (a fresh group has the same + semantics as today on first deploy). +2. Cut a `dev2` deploy. The first restart still drops between-pod + notifies (group has no committed offset yet); from the second + restart on, the gap is closed. +3. Verify in dev2: invite a user to the example org, kill the + mirror-v3-worker pod, confirm boards-v1 receives the invalidate + on the new pod within one commit interval after re-fetch. +4. Promote to `prod` once dev2 stays healthy through a deliberate + pod-restart cycle. + +The webhooks branch in `Yolean/mirror-v3` already contains the +prerequisites (per-mirror cache surface, suppression scaffolding); +the commit work is additive on top of that. + diff --git a/Dockerfile b/Dockerfile index fe53ef3..abaa8be 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,10 +6,11 @@ # libgcc + libstdc++ — enough for our dynamically-linked binary and, # in later phases, librdkafka. # -# When a new image is cut, replace `:latest` with `@sha256:` -# and commit the digest. Update both stages together. +# Both stages are digest-pinned (tag kept for readability). To +# update: `crane digest :` and bump both stages +# together. -FROM docker.io/library/rust:1-bookworm AS builder +FROM docker.io/library/rust:1-bookworm@sha256:7d0723df719e7f213b69dc7c8c595985c3f4b060cfbee4f7bc0e347a86fe3b6a AS builder # librdkafka 2.12+ unconditionally pulls in libcurl (OIDC support); # libssl/libsasl2/libzstd/liblz4 give us full feature support. cmake # drives the build, g++/make do the compiling, pkg-config is for @@ -36,7 +37,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ cargo build --release --bin mirror-v3 --locked && \ cp target/release/mirror-v3 /usr/local/bin/mirror-v3 -FROM gcr.io/distroless/cc-debian12:latest +FROM gcr.io/distroless/cc-debian12:latest@sha256:a90cf0f046efb32466b38b0972fef3a95e7c580e392e79ff1b7ac08c15fed0bc COPY --from=builder /usr/local/bin/mirror-v3 /usr/local/bin/mirror-v3 USER nonroot:nonroot ENTRYPOINT ["/usr/local/bin/mirror-v3"] diff --git a/README.md b/README.md index 5ce0a43..cae2eba 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Exactly-once Kafka topic+partition mirroring to **Kafka**, **Filesystem**, or **S3**, in one deployment. -> **Status:** Phase 5 — Kafka source + Kafka/Filesystem/S3 sinks; supervisor for parallel mirrors with graceful SIGINT/SIGTERM shutdown that flushes buffered records. Fault-injection tests deferred to Phase 2b; cutover to `checkit/mirror-v3` is a handoff step. See [AGENTS.md](AGENTS.md) for the phase map. +> **Status:** feature-complete for the `checkit/mirror-v3` cutover: Kafka source + Kafka/Filesystem/S3 sinks, `/cache/v1` HTTP surface, kkv-v1 notify webhooks, committed-offset delivery semantics and non-sticky readiness. See [AGENTS.md](AGENTS.md) for the development history. ## What this gives you @@ -74,7 +74,7 @@ GET /cache/v1/{mirror}/keys → newline-separated keys GET /cache/v1/{mirror}/values → newline-separated raw values ``` -Each mirror owns its own `key → latest-value` view; a key only shows up under the mirror that consumed it. Reads carry `x-kkv-last-seen-offsets: ` and return **503** until that mirror has caught up to its source's high-watermark captured at startup — same readiness contract as KKV, so dependents don't transiently see an older state across reloads. The view updates per-record from the consume loop, decoupled from disk flush cadence (set `flush.max-time-ms` high to save bucket ops without sacrificing freshness). Updates are monotonic; if a future feature ever rewinds source consumption, the cache stays at the highest offset seen. +Each mirror owns its own `key → latest-value` view; a key only shows up under the mirror that consumed it. Reads carry `x-kkv-last-seen-offsets: ` and return **503** until that mirror is `ready` (non-sticky, lag-based; see [Readiness](#readiness)) — same readiness contract as KKV, so dependents don't transiently see an older state across reloads. The view updates per-record from the consume loop, decoupled from disk flush cadence (set `flush.max-time-ms` high to save bucket ops without sacrificing freshness). Updates are monotonic; if a future feature ever rewinds source consumption, the cache stays at the highest offset seen. To keep existing kkv consumers working unmodified during a migration, **one** mirror per process may additionally set `cache-v1-main: {}`. That mounts the unprefixed `/cache/v1/...` paths onto that mirror's view (alias-only — same handlers, no separate data path). The validator rejects more than one `cache-v1-main` in the config. Mirror names that collide with the literal path segments `raw | offset | keys | values` are rejected. diff --git a/REVIEW_TEST_STRATEGY.md b/REVIEW_TEST_STRATEGY.md new file mode 100644 index 0000000..0b76637 --- /dev/null +++ b/REVIEW_TEST_STRATEGY.md @@ -0,0 +1,208 @@ +# Test coverage / strategy critique + +Written after a downstream consumer hit three bugs in a row that all +slipped through the existing suite: + +1. `StreamConsumer.fetch_watermarks` returning `(0, 0)` against a + real multi-broker Kafka cluster (the 7fa70e7 fix). +2. `LogStartOffset = 0` on a `cleanup.policy=compact`-only topic + producing an apparent gap on first delivery that the c2e64c11 + bootstrap branch didn't anticipate. +3. A follow-up: mid-stream gaps under `compaction: log` exhausting + `Sink::align_to_source_low_watermark`'s empty-buffer precondition + on the second delivery (caught in production immediately after + shipping the fix for #2, see the accompanying PR). + +Each of those bugs is a direct consequence of one of three different +ways the harness systematically lies about production conditions. +This document spells those out and proposes specific test additions. +Nothing here requires new infrastructure beyond what's already in the +repo; step 5 is the one that genuinely needs a CI conversation. + +## 1. The mock sink layer doesn't enforce real sink invariants + +`mirror_core::mock::MockSink` and `WriteInspector::InspectorSink` +under `tests/loop_invariants.rs` track a running position but no +buffer/durable split. The real `FilesystemSink`/`S3Sink` have: + +- a `buffer: Vec` that mutates after every `write`, +- a `durable_position` distinct from buffer head, +- an empty-buffer precondition on `align_to_source_low_watermark`, +- `flush_locked` deriving `(from, to)` from buffer state. + +The mid-stream-gap bug never failed `loop_invariants.rs` because +`InspectorSink` had none of that state `cargo test` was green and +production crashed on the second record. The earlier draft of the +compaction-only fix even called `align_to_source_low_watermark` from +the run loop on every mid-stream gap; the mock accepted it, the real +sink didn't. + +**Fix.** Either add a `BufferedMockSink` that models a buffer + +durable split plus the empty-buffer precondition, or better bring +the mirror-core loop-invariant tests up against real `FilesystemSink` +instances using `tempfile::TempDir`. Neither needs network. The +latter has the additional benefit that any future invariant change +on the real sinks (e.g. tightening `align`'s preconditions further) +shows up as a `mirror-core` test failure rather than a production +crash. + +## 2. Single-broker Redpanda production multi-broker Kafka + +The `kafka_source_low_watermark_contract.rs` test added with 7fa70e7 +explicitly notes the harness can't reproduce the bug it set out to +fix; the test exists as a contract guard against future maintainers +silently regressing the fetch path. That's fine as far as it goes +but it leaves the fix itself untested against any reproducer, and +the e2e suite still can't catch any other multi-broker metadata-race +bug in the same class. + +**Fix.** A second e2e stack variant against Apache Kafka with 3+ +brokers, run on PR. Each existing e2e test marked +`#[broker_count = single | any]` so the multi-broker pass becomes a +subset rather than a duplicate run. Cost is real (`testcontainers` +with three brokers is slow) but the bug class is uncatchable any +other way. + +## 3. Stand-in policies embed wrong premises + +The c2e64c11 commit message admits: + +> E2e coverage uses delete-records as a deterministic stand-in for +> broker-side compaction (the consumer sees an identical post-trim +> low watermark). + +That sentence IS the bug. `cleanup.policy=delete` (or `delete-records` +on a `compact,delete` topic) advances `LogStartOffset` the broker +reports the new low. `cleanup.policy=compact` alone does not the +broker reports `low = 0` regardless of how many keys have been +deduplicated, because the segment start hasn't moved. The fix built +on the wrong premise let the next layer of the same bug (gaps mid- +stream) hide too. + +**Fix.** Each Kafka cleanup policy gets its own broker-contract +test, named after the policy. No stand-ins. Suggested: + + - `kafka_source_low_watermark_after_delete_records` (rename of + current `kafka_source_low_watermark_contract.rs` for clarity; + keep it it documents the post-trim contract, which is still + real and correct). + - `kafka_source_low_watermark_after_compaction_only` (new uses + `log_compaction_interval_ms` + small `segment.ms` + forced + segment roll; asserts `low == 0` after a real compaction pass). + - `kafka_source_low_watermark_after_compact_and_delete` (new + asserts the `compact,delete` semantics, which differ from both + above). + +The "compaction:log accepts gaps" mirror-level tests then sit on top +of these broker-contract tests. If a future librdkafka or Redpanda +upgrade changes broker semantics, the contract tests fail loudly +before the mirror tests do. + +## 4. The sink trait surface isn't exercised matrix-style + +`allows_compacted_source()` is a mode flag that gates four other +methods (`write`, `next_expected_offset`, `flush_locked`, +`align_to_source_low_watermark`) plus the run-loop's per-record +gate. The test layout treats it as one feature with a handful of +happy-path tests; the actual matrix is: + +| mode | buffer state | event | outcome | +| --------- | ------------ | ----------------- | -------------------------------------- | +| append | empty | delivered=exp | write OK | +| append | empty | delivered>exp | SourceGapAboveExpected | +| append | empty | deliveredexp | SourceGapAboveExpected | +| append | non-empty | deliveredexp | write OK (bootstrap-time gap) | +| log | empty | deliveredexp | write OK (compaction gap mid-stream) | +| log | non-empty | delivered-.` | +| append | any | flush | filename `-.` | + +The mid-stream gap was the `log non-empty delivered>exp` cell +new code, no test before the fix. A table-driven test in +`mirror-fs/tests/` and `mirror-s3/tests/` walking those rows against +real sinks would have caught the buffer-state mismatch before the +ship, and protects against future regressions where a mode change +touches one of the gated paths but not the others. + +`proptest` is overkill; a `#[rstest]`-style or hand-rolled +table-driven test is fine. Rust enums + match on outcome keep the +table compile-checked. + +## 5. Restart correctness has its own small matrix + +The "destination is the source of truth" invariant is load-bearing +for every fix in this area. There's currently one e2e +(`compacted_source_with_compaction_log.rs`) that exercises one +corner of it. The full matrix is small enough to enumerate: + +| Cleanup policy | Destination state | Expected behaviour | +| ---------------- | -------------------------- | ---------------------------------------- | +| `delete` | empty | seek(0), no gap | +| `delete` | non-empty | seek(next_expected), no gap | +| `compact,delete` | empty, after DeleteRecords | seek(0), bootstrap-align to broker low | +| `compact,delete` | non-empty < broker low | seek(next_expected), bootstrap-align | +| `compact,delete` | non-empty broker low | seek(next_expected), no gap | +| `compact` only | empty | seek(0), gap on first delivery | +| `compact` only | non-empty | seek(next_expected), gap mid-stream | + +Seven rows. Two of them (`compact only`) silently misbehaved until +the current fix landed. A table-driven e2e walking these would +catch any variant of this bug class going forward. + +## Smaller observations + +- **Commit messages doing their own testing.** Several recent + commits ("Why the existing e2e didn't catch this", "the test was + passing for the wrong reason") flag known coverage gaps in prose. + That's good engineering culture, but the gaps then sit in + `git log` instead of the test suite. Convert each such note into + an `#[ignore = ""]` test with a TODO contract so it's + discoverable from `cargo test --list`, not just from `git blame`. + +- **Bench against bigger fixtures.** The production HWM the bug + surfaced against (1.2M offsets, real compaction work) is orders + of magnitude larger than the 12-record seeds in the existing e2e. + A medium-sized stress fixture (10k100k records, multiple keys, + forced compaction) catches buffer-pressure issues and flush- + trigger edge cases that small seeds don't. Doesn't have to run + on every PR keep it gated on a label or schedule. + +- **Don't conflate `delete-records` and "compaction" in test + names.** The `compacted_source_*` e2e tests today are about + `delete-records`, not real compaction. Renaming makes the gap + visible at the file-listing level instead of buried in the + comments. + +## Order of operations if I were the maintainer + +1. **Real-compaction repro in the existing single-broker harness.** + Cheap, low risk; just `log_compaction_interval_ms` + `segment.ms` + + forced roll. Unblocks all the renamed/split tests in 3. + +2. **Convert the prose "we don't cover X" notes into ignored + tests.** Five-minute hygiene with high payoff: each known gap + becomes a discoverable contract. + +3. **Sink-trait matrix table (4).** All in-process, no broker. Use + real `FilesystemSink` instances. Catches the next mode buffer- + state regression for free. + +4. **Restart matrix table (5).** Builds on 1. Touches the + destination/restart story which is the load-bearing invariant + for everything else in this area. + +5. **Multi-broker Apache Kafka stack variant.** Expensive but + recovers the missing third lie. Worth doing once the 14 work + has caught everything cheap to catch. + +Steps 1-4 are all in-repo, no new infrastructure or CI changes. +Step 5 is the one that genuinely needs a CI conversation. + diff --git a/WEBHOOKS.md b/WEBHOOKS.md index ed74e29..ef1b483 100644 --- a/WEBHOOKS.md +++ b/WEBHOOKS.md @@ -144,8 +144,9 @@ Field-level notes: every address that comes back. Headless Kubernetes Services naturally return one A record per pod, so this gives the same fan-out kkv used to do via the Endpoints API; without mirror-v3 - needing K8s API access. Resolutions are cached up to the DNS - record TTL. + needing K8s API access. Resolutions are cached for a fixed 30 s + window (`tokio::net::lookup_host` does not expose record TTLs) + and invalidated early on any dispatch failure. - **`notify.trigger`** decides what internal event causes a POST. See the dedicated section below; default is `source-consume` with small debounce, matching kkv's "as records arrive" behaviour. @@ -536,22 +537,26 @@ coupling. mirror-v3's `fan-out: dns-a` should: -1. Resolve the URL's host on first send. Cache the A/AAAA record set - up to the DNS TTL (default 30 s if no TTL is published). +1. Resolve the URL's host on first send. Cache the A/AAAA record + set for a fixed 30 s window (the system resolver API does not + expose record TTLs). 2. Open one HTTP/1.1 keep-alive connection per address (kept inside a pool, capped at the resolved set size). -3. POST the batch to all addresses concurrently. Aggregate the - results; if any address returns non-2xx after retry, the whole - batch is failed. -4. Re-resolve when the cache TTL expires OR when an address fails - repeatedly (forces an immediate re-resolve to pick up scale-up / - scale-down). +3. POST the batch once per address, concurrently. The retry loop + sits at the endpoint level: any address whose outcome policy + says retry triggers a retry of the whole set, and every retry + round resolves fresh (failure invalidates the cache). Addresses + that already accepted receive the batch again on later rounds; + kkv invalidations are idempotent so this stays within + at-least-once. +4. When the retry budget is exhausted (or the outcome says + `retry: false`), each still-failing address's configured final + action applies; any `fail` fails the whole batch. This handles the rolling-update case: during a Deployment rollout, -the headless Service's A-record set has both old and new pod IPs -for a few seconds; mirror-v3 POSTs to both, the old terminating -pods drain on whatever they got, and the next re-resolve drops -them. Same behaviour kkv had via Endpoints API. +a batch that hits a terminating pod IP fails that round, the retry +re-resolves and delivers to the replacement pod. A dead IP never +consumes the whole retry budget. For non-K8s use (a standalone service behind a single hostname), `fan-out: none` skips all of that and uses a single keep-alive @@ -627,7 +632,7 @@ Adds, alongside the existing `mirror_v3_destination_*` counters: | Metric | Type | Labels | Meaning | |-------------------------------------------------|---------|------------------------------------------|-----------------------------------------------| | `mirror_v3_notify_records_total` | counter | `topic`, `partition` | Records appended to a notify batch | -| `mirror_v3_notify_batches_total` | counter | `topic`, `partition`, `result=ok\|fail` | Batches sent | +| `mirror_v3_notify_batches_total` | counter | `topic`, `partition`, `result=ok\|skip\|fail` | Batches sent | | `mirror_v3_notify_post_duration_seconds` | histogram | `topic`, `partition`, `target_host` | Per-target HTTP latency | | `mirror_v3_notify_inflight_retry` | gauge | `topic`, `partition`, `target_host` | Current retry attempt (1-based, 0 when idle) | | `mirror_v3_notify_buffer_records` | gauge | `topic`, `partition` | Current buffer depth | @@ -641,7 +646,8 @@ latency. - One INFO line at startup per notify-enabled mirror: `notify start mirror= api=kkv-v1 targets=[,host…] fan-out=`. - One INFO line per successful batch: - `notify sent mirror= batch_records= highest_offset= targets= elapsed_ms=`. + `notify sent` with `topic`, `partition`, `batch_keys`, + `high_offset`, `targets` and `elapsed_ms` fields. - One WARN per failed attempt with retry remaining: `notify retry mirror= target= attempt=/ reason=`. - One ERROR on retry exhaustion (mirror-task-fatal): @@ -661,18 +667,21 @@ Per-record DEBUG only; counters cover the operational signal. default table above. Listing all six is allowed and recommended for production configs so the policy is explicit. - `final: accept` on `timeout`/`connrefused`/`5xx` with - `retry: false` is a valid but unusual combination; the validator - warns (operator probably meant `retry: true, final: accept`). + `retry: false` is a valid but unusual combination (operator + probably meant `retry: true, final: accept`); not currently + flagged by the validator. - **Destinations relaxation** (new in this proposal): `destinations` MAY be empty *if and only if* `notify` is set with at least one target. See "Notify-only mirrors" above for the full matrix of which other fields are then forbidden (`format`/`compression`/`compaction`/`flush`/`http-access`) and which trigger modes are required (`trigger.on: source-consume`). -- `notify.targets[].url` parses as a valid URL with http:// or https://. -- Each target's resolved host must produce ≥1 address at startup, - otherwise validation fails (catches typos / missing Services - before the mirror runs). +- `notify.targets[].url` parses as a valid URL with http:// or + https:// and a host. Hostname resolution is NOT checked at + startup (validation must work without cluster DNS); an + unresolvable host surfaces on the first dispatch, where a 0 + address resolution is a transport error subject to the + configured outcome policy. ## Out-of-scope (future) diff --git a/crates/mirror-bin/src/ack_tracker.rs b/crates/mirror-bin/src/ack_tracker.rs index 58da341..d8cc046 100644 --- a/crates/mirror-bin/src/ack_tracker.rs +++ b/crates/mirror-bin/src/ack_tracker.rs @@ -32,6 +32,17 @@ use tokio::sync::watch; const DEFAULT_COMMIT_INTERVAL: Duration = Duration::from_secs(5); +/// How long an unchanged offset may go without a re-commit. The +/// consumer uses `assign()` (no group membership), so the broker +/// sees the group as permanently empty and expires its offset after +/// `offsets.retention.minutes` (default 7 days). An idle mirror that +/// never re-commits would silently regress to fresh-deploy +/// HWM suppression on its next restart, reopening the between-pods +/// notify gap. Re-committing the same value refreshes the broker's +/// retention clock; hourly is cheap (one broker round-trip) and two +/// orders of magnitude inside the default retention window. +const COMMIT_REFRESH_INTERVAL: Duration = Duration::from_secs(3600); + /// Read the commit interval from `MIRROR_V3_OFFSET_COMMIT_INTERVAL_MS`, /// falling back to [`DEFAULT_COMMIT_INTERVAL`]. A value of `0` /// disables the periodic task (the supervisor then never advances @@ -197,6 +208,7 @@ pub fn spawn_periodic_commit_task( // Consume the immediate tick `tokio::time::interval` fires. iv.tick().await; let mut last_committed: u64 = 0; + let mut last_commit_at = std::time::Instant::now(); loop { tokio::select! { biased; @@ -211,7 +223,12 @@ pub fn spawn_periodic_commit_task( } _ = iv.tick() => { let off = tracker.commit_offset(); - if off == 0 || off == last_committed { + if off == 0 { + continue; + } + if off == last_committed + && last_commit_at.elapsed() < COMMIT_REFRESH_INTERVAL + { continue; } if let Err(e) = handle.commit_through(off) { @@ -239,12 +256,54 @@ pub fn spawn_periodic_commit_task( "committed source offset" ); last_committed = off; + last_commit_at = std::time::Instant::now(); } } } }) } +/// One final synchronous commit after the mirror's run loop has +/// completed. The periodic task exits on the shutdown signal before +/// the run loop's final drain acks, so without this the last +/// interval's worth of acks (plus the shutdown drain batch) would +/// replay as duplicate webhooks on the next start. Sync commit mode: +/// an async commit here would race process exit. Best-effort like +/// the periodic task; on error the offsets simply replay. +pub async fn final_commit(handle: KafkaCommitHandle, tracker: &AckTracker, mirror_name: &str) { + let off = tracker.commit_offset(); + if off == 0 { + return; + } + if let Err(e) = handle.commit_through(off) { + tracing::warn!( + mirror = %mirror_name, + offset = off, + error = %e, + "final commit_through failed; offsets will replay on next start" + ); + return; + } + let mirror = mirror_name.to_string(); + let result = tokio::task::spawn_blocking(move || handle.commit_pending_sync()).await; + match result { + Ok(Ok(())) => { + tracing::info!(mirror = %mirror, offset = off, "final source offset commit"); + } + Ok(Err(e)) => { + tracing::warn!( + mirror = %mirror, + offset = off, + error = %e, + "final sync commit failed; offsets will replay on next start" + ); + } + Err(e) => { + tracing::warn!(mirror = %mirror, error = %e, "final commit task join error"); + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/mirror-bin/src/main.rs b/crates/mirror-bin/src/main.rs index c0f31f2..d1c0d22 100644 --- a/crates/mirror-bin/src/main.rs +++ b/crates/mirror-bin/src/main.rs @@ -9,8 +9,8 @@ use mirror_config::{Destination, HttpAccess, Mirror}; mod ack_tracker; mod readiness_poller; use ack_tracker::{ - commit_interval_from_env, spawn_periodic_commit_task, AckTracker, DestAckSlot, FlushAckShim, - WriteAckShim, + commit_interval_from_env, final_commit, spawn_periodic_commit_task, AckTracker, DestAckSlot, + FlushAckShim, WriteAckShim, }; use mirror_core::{ heartbeat_interval_from_env, run_mirror_with_notifier, MetricLabels, NoOpNotifier, Record, @@ -685,6 +685,30 @@ async fn fetch_committed_offset_for_mirror(mirror: &Mirror) -> Result Result { + let bootstrap = mirror.source.bootstrap_servers.clone(); + let topic = mirror.topic.clone(); + let partition = mirror.partition as i32; + let mirror_name = mirror.name.clone(); + let low = tokio::task::spawn_blocking(move || { + mirror_kafka::fetch_low_watermark( + &bootstrap, + &topic, + partition, + std::time::Duration::from_secs(5), + ) + }) + .await + .with_context(|| format!("mirror {mirror_name}: low watermark task join"))? + .with_context(|| format!("mirror {mirror_name}: fetch low watermark"))?; + Ok(low.max(0) as u64) +} + async fn shutdown_signal(mut rx: tokio::sync::watch::Receiver) { if *rx.borrow() { return; @@ -733,12 +757,14 @@ async fn spawn_mirror( ); let source = KafkaSource::open(source_cfg) .with_context(|| format!("opening source for mirror {}", mirror.name))?; - // Snapshot two commit handles before the run loop takes + // Snapshot three commit handles before the run loop takes // ownership of the source. Each `KafkaCommitHandle` clones the // underlying `Arc` (cheap); the periodic commit - // task and the readiness poller each get their own. + // task, the readiness poller and the final shutdown commit each + // get their own. let commit_handle = source.commit_handle(); let commit_handle_for_poller = source.commit_handle(); + let commit_handle_final = source.commit_handle(); let name = mirror.name.clone(); let labels = MetricLabels { @@ -829,6 +855,7 @@ async fn spawn_mirror( // contribute (commit 9 wires `affects-readiness` to filter). let notify_present = mirror.notify.is_some(); let ack_tracker = Arc::new(AckTracker::new(notify_present, dest_ack_slots)); + let ack_tracker_final = Arc::clone(&ack_tracker); // Branch on the notify trigger mode (validated upstream in // mirror-config; see WEBHOOKS.md § Trigger): @@ -844,6 +871,23 @@ async fn spawn_mirror( let trigger_mode = mirror.notify.as_ref().map(|n| n.trigger.on); let ack_sink_for_notifier: Arc = Arc::clone(&ack_tracker) as Arc; + // Notify re-delivery across restarts: the destinations' durable + // heads can be ahead of the broker-committed (= notify-acked) + // offset, e.g. when the webhook receiver was down at shutdown. + // The destination state alone would resume above those records + // and their invalidations would never fire. `[committed, + // min(heads))` is that gap; each trigger mode closes it below. + let notify_committed = if trigger_mode.is_some() { + fetch_committed_offset_for_mirror(&mirror).await? + } else { + None + }; + let min_head = tee + .heads() + .iter() + .map(|(_, h)| *h) + .min() + .expect("tee is non-empty by construction"); let notifier_opt = match trigger_mode { Some(mirror_config::TriggerOn::SourceConsume) => { build_source_consume_notifier(&mirror, cache.as_ref())? @@ -851,13 +895,73 @@ async fn spawn_mirror( } _ => None, }; + if notifier_opt.is_some() { + if let Some(committed) = notify_committed { + if committed < min_head { + // Re-read the gap from the source; the tee skips the + // destination writes and the notifier's suppression + // threshold (== committed) lets exactly these + // records fire. + let low = fetch_low_watermark_for_mirror(&mirror).await?; + if committed < low { + tracing::warn!( + mirror = %name, + committed, + low_watermark = low, + "committed offset is below the broker low watermark; notify re-delivery starts at the earliest available offset and the records below it are lost" + ); + } + let floor = committed.max(low); + tracing::info!( + mirror = %name, + committed, + destination_min_head = min_head, + resume_floor = floor, + "re-reading un-acked notify window from the source" + ); + tee.set_resume_floor(floor); + } + } + } + // The run loop races this watch so a terminal dispatch failure + // errors the mirror even when nothing else would surface it: + // the flush drainer's death is otherwise invisible (its events + // are not regenerated on restart), and the notifier's timer + // error only surfaces on the next record, which an idle topic + // never delivers. + let mut dispatch_error_watch = notifier_opt.as_ref().map(|n| n.terminal_error_watch()); + let mut flush_dispatcher_shutdown: Option> = None; if matches!( trigger_mode, Some(mirror_config::TriggerOn::DestinationFlush) ) { let dispatcher = build_flush_dispatcher(&mirror, cache.as_ref())? .with_ack_sink(Arc::clone(&ack_sink_for_notifier)); - tee.set_flush_observer(std::sync::Arc::new(dispatcher)); + dispatch_error_watch = Some(dispatcher.terminal_error_watch()); + // Destination-flush variant of the un-acked window: a replay + // cannot regenerate flush events for data that is already + // durable, so synthesize one catch-up event covering + // `[committed, min_head)`. Suppression (threshold == + // committed) admits it; a fresh deploy (no commit) skips it. + if let Some(committed) = notify_committed { + if committed < min_head { + tracing::info!( + mirror = %name, + committed, + destination_min_head = min_head, + "dispatching synthetic flush notification for the un-acked window" + ); + use mirror_core::FlushObserver; + dispatcher.on_flushed(committed, min_head - 1); + } + } + // The tee owns the dispatcher as its FlushObserver; the + // supervisor keeps a second handle so the graceful-shutdown + // path can drain queued flush events before the process + // exits (they are not regenerated on restart). + let dispatcher = Arc::new(dispatcher); + flush_dispatcher_shutdown = Some(Arc::clone(&dispatcher)); + tee.set_flush_observer(dispatcher); } // Spawn the periodic source-commit task. It reads @@ -934,7 +1038,10 @@ async fn spawn_mirror( MIRROR_LABELS .scope( labels, - run_mirror_with_notifier(source, tee, n, shutdown, heartbeat), + run_racing_dispatch_error( + run_mirror_with_notifier(source, tee, n, shutdown, heartbeat), + dispatch_error_watch, + ), ) .await } @@ -942,26 +1049,74 @@ async fn spawn_mirror( MIRROR_LABELS .scope( labels, - run_mirror_with_notifier( - source, - tee, - NoOpNotifier, - shutdown, - heartbeat, + run_racing_dispatch_error( + run_mirror_with_notifier( + source, + tee, + NoOpNotifier, + shutdown, + heartbeat, + ), + dispatch_error_watch, ), ) .await } }; - match result { - Ok(()) => Ok(()), - Err(e) => Err(anyhow::anyhow!("mirror {name}: {e}")), + // Drain the flush dispatcher before committing: the run + // loop's final sink.flush() may have queued flush events + // that are not regenerated on restart, and their acks + // belong in the final commit below. + let flush_drain = match flush_dispatcher_shutdown { + Some(d) => d.drain_and_stop().await, + None => Ok(()), + }; + // One final sync commit of whatever the notifier acked: + // the periodic commit task exits on the shutdown signal + // before the run loop's final drain acks, and its async + // commit mode would race process exit anyway. Runs on + // the error path too - acked offsets are delivered + // regardless of why the loop stopped, and committing + // them shrinks the duplicate-webhook replay on restart. + final_commit(commit_handle_final, &ack_tracker_final, &name).await; + match (result, flush_drain) { + (Ok(()), Ok(())) => Ok(()), + (Err(e), _) => Err(anyhow::anyhow!("mirror {name}: {e}")), + (Ok(()), Err(e)) => Err(anyhow::anyhow!( + "mirror {name}: flush notify drain on shutdown: {e}" + )), } } .instrument(span), )) } +/// Race the mirror run loop against the notify pipeline's terminal +/// dispatch error. Whichever resolves first errors (or completes) +/// the mirror; the loser is dropped. With no watch installed the +/// second branch is pending forever and this is exactly the run +/// loop. +async fn run_racing_dispatch_error( + run: Fut, + watch: Option, +) -> Result<(), anyhow::Error> +where + Fut: std::future::Future>, +{ + let watch_error = async move { + match watch { + Some(w) => w.wait().await, + None => std::future::pending().await, + } + }; + tokio::select! { + r = run => r.map_err(anyhow::Error::from), + e = watch_error => Err(anyhow::anyhow!( + "notify dispatch failed terminally: {e}" + )), + } +} + /// Construct the `KkvV1Notifier` for a mirror with /// `trigger.on: source-consume`. Returns `None` when the mirror has /// no notify block or uses a different trigger (the supervisor diff --git a/crates/mirror-core/src/lib.rs b/crates/mirror-core/src/lib.rs index 2feece5..f68582a 100644 --- a/crates/mirror-core/src/lib.rs +++ b/crates/mirror-core/src/lib.rs @@ -361,6 +361,17 @@ pub trait Sink: Send { /// replace earlier ones. fn set_flush_observer(&mut self, _observer: Arc) {} + /// Whether this sink ever fires the [`FlushObserver`] installed + /// via [`Self::set_flush_observer`]. Blob sinks (FS, S3) return + /// true; per-record sinks (Kafka) and mocks keep the default + /// false. Coordinators (TeeSink's min-across-destinations for + /// `trigger.on: destination-flush`) use this to exclude sinks + /// whose flush watermark would otherwise sit at 0 forever and + /// pin the min there. + fn supports_flush_observer(&self) -> bool { + false + } + /// Install a [`WriteObserver`] that fires after every successful /// `write`. Default no-op for sinks where the per-record signal /// is uninteresting or already covered by [`FlushObserver`] diff --git a/crates/mirror-core/src/tee.rs b/crates/mirror-core/src/tee.rs index 82a58a4..dc27968 100644 --- a/crates/mirror-core/src/tee.rs +++ b/crates/mirror-core/src/tee.rs @@ -64,6 +64,17 @@ struct InnerSink { pub struct TeeSink { inners: Vec, cache: Option, + /// Resume cursor for notify re-delivery. When the supervisor + /// sets a floor below the inner sinks' heads (the broker-side + /// committed offset of a notify mirror), the tee reports it as + /// `next_expected_offset` so the source seeks there and the run + /// loop re-presents `[floor, min(heads))` to the notifier; the + /// per-sink head skip keeps the destinations write-idempotent + /// through the replay. The cursor then advances with every + /// `write` call (including fully-skipped replays) so the idle + /// drift re-check, which requires `next_expected_offset ==` the + /// loop's own tracker, stays satisfied mid-replay. + resume_cursor: Option, } impl TeeSink { @@ -89,7 +100,23 @@ impl TeeSink { let head = sink.next_expected_offset().await?; inners.push(InnerSink { name, sink, head }); } - Ok(Self { inners, cache }) + Ok(Self { + inners, + cache, + resume_cursor: None, + }) + } + + /// Lower the tee's reported resume position to `floor` so the + /// run loop re-reads `[floor, min(heads))` from the source. Used + /// by the supervisor for notify mirrors: records that were made + /// durable on the destinations but whose webhook batch was never + /// acked (committed) get re-presented to the notifier, closing + /// the at-least-once gap across restarts. A floor at or above + /// the inner minimum is a no-op. Call before the run loop + /// starts; the cursor is not meant to move backwards mid-run. + pub fn set_resume_floor(&mut self, floor: u64) { + self.resume_cursor = Some(floor); } /// Test-only / mirror-bin-style constructor that skips the open @@ -107,6 +134,18 @@ impl TeeSink { .map(|(name, sink, head)| InnerSink { name, sink, head }) .collect(), cache, + resume_cursor: None, + } + } + + /// Keep the resume cursor in step with the run loop's `expected` + /// tracker: every write call (skipped or fanned out) means the + /// loop is about to expect `record_offset + 1`. The idle drift + /// re-check compares `next_expected_offset()` for equality, so a + /// stale cursor would fail it mid-replay. + fn advance_resume_cursor(&mut self, record_offset: u64) { + if let Some(cursor) = self.resume_cursor { + self.resume_cursor = Some(cursor.max(record_offset + 1)); } } @@ -140,12 +179,19 @@ impl Sink for TeeSink { inner.head = head; } } - Ok(self + let inner_min = self .inners .iter() .map(|i| i.head) .min() - .expect("non-empty by construction")) + .expect("non-empty by construction"); + // The resume cursor can only lower the reported position + // (replay for notify re-delivery); it never holds the tee + // above the inner minimum. + Ok(match self.resume_cursor { + Some(cursor) => inner_min.min(cursor), + None => inner_min, + }) } async fn write(&mut self, record: Record) -> Result<(), SinkError> { @@ -168,9 +214,13 @@ impl Sink for TeeSink { } } if indices.is_empty() { - // Every inner sink is already past this offset (rare: - // only happens during restart with all sinks recovered - // to or beyond `record_offset`). Drop the record. + // Every inner sink is already past this offset: the + // notify-resume replay window, or a restart with all + // sinks recovered beyond `record_offset`. Drop the + // record for the destinations (the run loop still hands + // it to the notifier) but keep the resume cursor in step + // with the loop's `expected` tracker. + self.advance_resume_cursor(record_offset); return Ok(()); } @@ -217,6 +267,7 @@ impl Sink for TeeSink { if let Some((name, e)) = first_err { return Err(SinkError::Transport(format!("inner sink {name}: {e}"))); } + self.advance_resume_cursor(record_offset); Ok(()) } @@ -306,26 +357,59 @@ impl Sink for TeeSink { } fn set_flush_observer(&mut self, observer: Arc) { - if self.inners.len() == 1 { - // Length-1 tee (the common case for single-destination - // mirrors): forward the observer to the only inner sink - // unchanged. `from`/`to` flow through verbatim. - self.inners[0].sink.set_flush_observer(observer); - return; - } - // Multi-destination: wrap the outer observer with a per-sink - // relay + a min-coordinator. The outer observer fires only - // when *every* inner sink has committed past a watermark - - // matching the spec's "fire when ALL destinations have - // committed past the batch's high-water offset". - let coordinator = Arc::new(MinFlushCoordinator::new(self.inners.len(), observer)); - for (sink_index, inner) in self.inners.iter_mut().enumerate() { - inner.sink.set_flush_observer(Arc::new(PerSinkRelay { - sink_index, - coordinator: Arc::clone(&coordinator), - })); + // Only sinks that actually fire flush events participate in + // the min-coordination. A per-record sink (Kafka) never + // calls its observer, so including it would pin the min at + // 0 and the outer observer would never fire on a mixed + // blob+kafka mirror. + let supporting: Vec = self + .inners + .iter() + .enumerate() + .filter(|(_, inner)| inner.sink.supports_flush_observer()) + .map(|(i, _)| i) + .collect(); + match supporting.len() { + 0 => { + // The config validator rejects destination-flush + // without a blob destination, so this is a + // programming error rather than an operator one; + // keep it loud. + tracing::error!( + "set_flush_observer on a tee with no flush-capable inner sink; flush notifications will never fire" + ); + } + 1 => { + // Single flush-capable sink (covers the common + // single-destination mirror): forward the observer + // unchanged. `from`/`to` flow through verbatim. + self.inners[supporting[0]].sink.set_flush_observer(observer); + } + n => { + // Multiple flush-capable sinks: wrap the outer + // observer with a per-sink relay + min-coordinator. + // The outer observer fires only when every + // flush-capable inner sink has committed past a + // watermark - the spec's "fire when ALL destinations + // have committed past the batch's high-water offset". + let coordinator = Arc::new(MinFlushCoordinator::new(n, observer)); + for (relay_index, inner_index) in supporting.into_iter().enumerate() { + self.inners[inner_index] + .sink + .set_flush_observer(Arc::new(PerSinkRelay { + sink_index: relay_index, + coordinator: Arc::clone(&coordinator), + })); + } + } } } + + fn supports_flush_observer(&self) -> bool { + self.inners + .iter() + .any(|inner| inner.sink.supports_flush_observer()) + } } /// Per-sink wrapper that funnels every inner sink's `on_flushed` @@ -428,6 +512,10 @@ mod tests { flush_count: Arc>, fail_on_offset: Option, allow_compacted: bool, + /// Mirrors the FS/S3 (`true`) vs Kafka (`false`) split in + /// `Sink::supports_flush_observer`. Default true so the + /// existing flush-coordination tests model blob sinks. + flush_capable: bool, aligned_to: Arc>>, /// The observer the tee installed via `set_flush_observer`, /// if any. Tests fire it explicitly via [`Self::simulate_flush`] @@ -455,6 +543,7 @@ mod tests { flush_count, fail_on_offset: None, allow_compacted: false, + flush_capable: true, aligned_to, observer, }, @@ -469,6 +558,13 @@ mod tests { self.allow_compacted = allow; self } + /// Model a per-record sink (Kafka): accepts an observer + /// install but never fires it and reports itself + /// flush-incapable. + fn without_flush_support(mut self) -> Self { + self.flush_capable = false; + self + } } #[derive(Clone)] @@ -534,6 +630,9 @@ mod tests { fn set_flush_observer(&mut self, observer: Arc) { *self.observer.lock().unwrap() = Some(observer); } + fn supports_flush_observer(&self) -> bool { + self.flush_capable + } } fn boxed(s: Recording) -> Box { @@ -773,4 +872,78 @@ mod tests { ra.simulate_flush(0, 5); assert_eq!(obs.fires.lock().unwrap().clone(), vec![(0, 5)]); } + + #[tokio::test] + async fn resume_floor_lowers_reported_start_and_tracks_replay() { + let (a, ra) = Recording::new(5); + let mut tee = TeeSink::open(vec![("a".into(), boxed(a))], None) + .await + .unwrap(); + tee.set_resume_floor(2); + assert_eq!( + tee.next_expected_offset().await.unwrap(), + 2, + "floor lowers the reported start below the inner head" + ); + // Replay window 2..=4: dropped for the destination, but the + // reported next-expected must track the loop's tracker or + // the idle drift re-check fails mid-replay. + for offset in 2..=4 { + tee.write(rec(offset)).await.unwrap(); + assert_eq!(tee.next_expected_offset().await.unwrap(), offset + 1); + } + assert_eq!( + ra.writes(), + Vec::::new(), + "replay must not re-write to the destination" + ); + // From the destination head onward, writes fan out normally + // and the cursor converges with the inner heads. + for offset in 5..=7 { + tee.write(rec(offset)).await.unwrap(); + } + assert_eq!(ra.writes(), vec![5, 6, 7]); + assert_eq!(tee.next_expected_offset().await.unwrap(), 8); + } + + #[tokio::test] + async fn resume_floor_at_or_above_inner_min_is_a_noop() { + let (a, _ra) = Recording::new(3); + let mut tee = TeeSink::open(vec![("a".into(), boxed(a))], None) + .await + .unwrap(); + tee.set_resume_floor(9); + assert_eq!( + tee.next_expected_offset().await.unwrap(), + 3, + "the cursor never raises the tee above the inner minimum" + ); + } + + /// The mixed blob+kafka regression: a per-record sink must not + /// pin the flush min at 0. With the kafka-like sink excluded + /// from coordination, the blob sink's flush alone drives the + /// outer observer. + #[tokio::test] + async fn flush_observer_ignores_sinks_that_never_flush() { + let (blob, r_blob) = Recording::new(0); + let (kafka, _r_kafka) = Recording::new(0); + let kafka = kafka.without_flush_support(); + let mut tee = TeeSink::open( + vec![("blob".into(), boxed(blob)), ("kafka".into(), boxed(kafka))], + None, + ) + .await + .unwrap(); + let obs = Arc::new(RecordingObserver::default()); + tee.set_flush_observer(obs.clone() as Arc); + assert!(tee.supports_flush_observer()); + + r_blob.simulate_flush(0, 9); + assert_eq!( + obs.fires.lock().unwrap().clone(), + vec![(0, 9)], + "the blob sink's flush must fire the outer observer even though the kafka sink never flushes" + ); + } } diff --git a/crates/mirror-core/tests/notify_resume_replay.rs b/crates/mirror-core/tests/notify_resume_replay.rs new file mode 100644 index 0000000..83bb4df --- /dev/null +++ b/crates/mirror-core/tests/notify_resume_replay.rs @@ -0,0 +1,86 @@ +//! Loop-level test of the committed-offset resume mechanism: when +//! the tee's resume floor sits below the destinations' durable +//! heads (records were made durable but their notify batch was +//! never acked/committed), the loop re-reads the gap from the +//! source, hands every record to the notifier, skips the +//! destination writes, and keeps the idle drift re-check satisfied +//! throughout the replay. + +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use mirror_core::mock::{rec, MockSink, MockSource, MockSourceEvent}; +use mirror_core::{run_mirror_with_notifier, MirrorError, Notifier, NotifyError, Record, TeeSink}; + +fn drive(future: F) -> Result<(), MirrorError> +where + F: std::future::IntoFuture>, +{ + let rt = tokio::runtime::Builder::new_current_thread() + .enable_time() + .build() + .unwrap(); + rt.block_on(async move { future.into_future().await }) +} + +fn never() -> std::future::Pending<()> { + std::future::pending::<()>() +} + +struct RecordingNotifier { + seen: Arc>>, +} + +#[async_trait] +impl Notifier for RecordingNotifier { + async fn on_record(&mut self, record: &Record) -> Result<(), NotifyError> { + self.seen.lock().unwrap().push(record.source_offset); + Ok(()) + } +} + +#[test] +fn resume_floor_replays_unacked_window_to_notifier_without_rewriting_destination() { + // Destination durable through offset 4 (head 5); committed + // (notify-acked) only through 2. The supervisor sets the floor + // to the committed offset; the source re-delivers 2..=6, with + // idle polls interleaved to exercise the drift re-check both + // inside and after the replay window. + let source = MockSource::new([ + MockSourceEvent::Record(rec(2)), + MockSourceEvent::Idle, + MockSourceEvent::Record(rec(3)), + MockSourceEvent::Record(rec(4)), + MockSourceEvent::Idle, + MockSourceEvent::Record(rec(5)), + MockSourceEvent::Record(rec(6)), + MockSourceEvent::Idle, + MockSourceEvent::Error("end of test".into()), + ]); + let sink = MockSink::starting_at(5); + let mut tee = TeeSink::from_inners_for_test(vec![("dest".into(), Box::new(sink), 5)], None); + tee.set_resume_floor(2); + + let seen = Arc::new(Mutex::new(Vec::new())); + let notifier = RecordingNotifier { + seen: Arc::clone(&seen), + }; + + let result = drive(run_mirror_with_notifier( + source, + tee, + notifier, + never(), + std::time::Duration::ZERO, + )); + assert!( + matches!(result, Err(MirrorError::Source(_))), + "loop must run through the whole script (no drift error mid-replay); got: {result:?}" + ); + + assert_eq!( + seen.lock().unwrap().clone(), + vec![2, 3, 4, 5, 6], + "the notifier must observe the full re-read window plus the new records" + ); +} diff --git a/crates/mirror-fs/src/lib.rs b/crates/mirror-fs/src/lib.rs index ddbd917..060c019 100644 --- a/crates/mirror-fs/src/lib.rs +++ b/crates/mirror-fs/src/lib.rs @@ -163,10 +163,24 @@ impl FilesystemSink { // - compaction:log → fold the loaded snapshot view. // - append → replay the whole chain (cost is linear in the // total record count; documented for large topics). + // + // `apply_record` enforces per-(topic, partition) offset + // monotonicity and silently skips records with + // `source_offset <= last_seen`. The compacted snapshot is a + // `BTreeMap` whose `.values()` iterate in + // lexicographic key order, NOT offset order — without + // sorting, the first replayed record's offset would clamp + // every subsequent record whose key happens to land + // alphabetically later but at a lower offset, leaving the + // in-memory view with a key-order-dependent subset of the + // snapshot. Sort by `source_offset` ascending before + // dispatch so every snapshot record passes the guard. if let Some(binding) = cfg.cache.as_ref() { match &view { Some(v) => { - for r in v.values() { + let mut snapshot_records: Vec<&Record> = v.values().collect(); + snapshot_records.sort_by_key(|r| r.source_offset); + for r in snapshot_records { binding.state.apply_record(&binding.mirror_name, r); } } @@ -546,6 +560,10 @@ impl Sink for FilesystemSink { fn set_flush_observer(&mut self, observer: std::sync::Arc) { self.flush_observer = Some(observer); } + + fn supports_flush_observer(&self) -> bool { + true + } } fn unix_now_seconds() -> u64 { diff --git a/crates/mirror-fs/tests/sink.rs b/crates/mirror-fs/tests/sink.rs index 015f464..f244804 100644 --- a/crates/mirror-fs/tests/sink.rs +++ b/crates/mirror-fs/tests/sink.rs @@ -535,3 +535,70 @@ async fn compaction_idle_flush_now_does_not_emit_file() { "idle flush in compaction mode must not emit a file" ); } + +#[tokio::test] +async fn compaction_snapshot_replay_populates_full_cache_view() { + // Regression: BTreeMap.values() yields snapshot + // records in lexicographic key order. CacheState::apply_record + // enforces per-(topic, partition) offset monotonicity, so if the + // first replayed record's offset clamps `last_seen`, every + // later snapshot record at a lower offset is silently dropped + // and the in-memory cache ends up as a random subset of the + // snapshot. The supervisor's bootstrap path must sort by + // source_offset before dispatch. + // + // Construct keys whose alphabetic order disagrees with their + // source-offset order: "z-first" lands at the highest offset + // but its key sorts first under most ordinal expectations after + // the lower-offset "a-…" keys; pick a layout that exposes the + // bug in EITHER direction. + use mirror_core::CacheBinding; + use mirror_fs::CacheBinding as FsCacheBinding; + let tmp = tempfile::tempdir().unwrap(); + // Write a snapshot by emitting 4 records that flush together. + let mut writer_cfg = cfg_compacted(tmp.path(), 4); + writer_cfg.cache = None; + let mut sink = FilesystemSink::open(writer_cfg).unwrap(); + // alphabetical key order: alpha < beta < gamma < zeta + // source-offset order: zeta=0, gamma=1, beta=2, alpha=3 + sink.write(rec_kv(0, "zeta", Some("vz"))).await.unwrap(); + sink.write(rec_kv(1, "gamma", Some("vg"))).await.unwrap(); + sink.write(rec_kv(2, "beta", Some("vb"))).await.unwrap(); + sink.write(rec_kv(3, "alpha", Some("va"))).await.unwrap(); + drop(sink); // close + flush via Drop is not enough; flush explicitly + let mut sink = FilesystemSink::open(cfg_compacted(tmp.path(), 100)).unwrap(); + sink.flush_now().await.unwrap(); + drop(sink); + // Snapshot file exists; now re-open with a cache binding and + // assert every key is in the bootstrapped view. + let cache_state = std::sync::Arc::new(mirror_core::CacheState::new()); + cache_state.register_mirror("ops", 0, None, false); + let mut reopen_cfg = cfg_compacted(tmp.path(), 100); + reopen_cfg.cache = Some(FsCacheBinding { + state: std::sync::Arc::clone(&cache_state), + mirror_name: "ops".into(), + }); + // Cross-check that the binding alias matches mirror-core's + // shape — both crates re-export the same struct. + let _: CacheBinding = CacheBinding { + state: std::sync::Arc::clone(&cache_state), + mirror_name: "ops".into(), + }; + let _sink = FilesystemSink::open(reopen_cfg).unwrap(); + assert_eq!( + cache_state.get_value_for("ops", "alpha").as_deref(), + Some(b"va".as_ref()) + ); + assert_eq!( + cache_state.get_value_for("ops", "beta").as_deref(), + Some(b"vb".as_ref()) + ); + assert_eq!( + cache_state.get_value_for("ops", "gamma").as_deref(), + Some(b"vg".as_ref()) + ); + assert_eq!( + cache_state.get_value_for("ops", "zeta").as_deref(), + Some(b"vz".as_ref()) + ); +} diff --git a/crates/mirror-kafka/src/lib.rs b/crates/mirror-kafka/src/lib.rs index 5bab1b0..000ccbd 100644 --- a/crates/mirror-kafka/src/lib.rs +++ b/crates/mirror-kafka/src/lib.rs @@ -255,6 +255,18 @@ impl KafkaCommitHandle { .commit_consumer_state(CommitMode::Async) .map_err(|e| SourceError::Transport(format!("commit_consumer_state: {e}"))) } + + /// Like [`Self::commit_pending`] but with `CommitMode::Sync`: + /// returns only after the broker has acknowledged the commit. + /// Used for the one final commit on graceful shutdown, where an + /// async commit would race process exit and usually lose. Blocks + /// the calling thread; call under `spawn_blocking` from async + /// contexts. + pub fn commit_pending_sync(&self) -> Result<(), SourceError> { + self.consumer + .commit_consumer_state(CommitMode::Sync) + .map_err(|e| SourceError::Transport(format!("commit_consumer_state sync: {e}"))) + } } /// Stage `through` as the offset to commit for `(topic, partition)`, diff --git a/crates/mirror-notify-kkv/src/lib.rs b/crates/mirror-notify-kkv/src/lib.rs index 9934559..50309c0 100644 --- a/crates/mirror-notify-kkv/src/lib.rs +++ b/crates/mirror-notify-kkv/src/lib.rs @@ -31,7 +31,7 @@ use indexmap::IndexMap; use mirror_config::{ FanOut, FinalAction, NotifyApi, NotifyOutcome, NotifyOutcomes, NotifyRetry, NotifyTarget, }; -use mirror_core::{current_labels, AckSink, CacheState, Notifier, NotifyError, Record}; +use mirror_core::{AckSink, CacheState, Notifier, NotifyError, Record}; use reqwest::Client; use serde::Serialize; use thiserror::Error; @@ -146,7 +146,24 @@ struct NotifierState { buffer: TokioMutex, new_data: TokioNotify, shutting_down: AtomicBool, - error_state: TokioMutex>, + error_state: Arc>>, + /// Signalled (`notify_one`) whenever a background task stashes a + /// terminal error into `error_state`, so a + /// [`TerminalErrorWatch`] can wake without polling. `on_record` + /// still surfaces the error on the next call; the watch exists + /// for the idle-topic case where no next call ever comes. + error_signal: Arc, + /// Serializes take-dispatch-ack across the inline drain + /// (`on_record` max-records path, shutdown) and the background + /// timer drain. Without it two batches can be in flight at once + /// and a *later* batch's success can ack past an *earlier* batch + /// that is still retrying; `AckTracker::note_through` is + /// `fetch_max`, so the periodic source commit would then advance + /// past undelivered records and a restart would suppress them + /// forever. Holding the lock across the whole dispatch also + /// gives the documented backpressure: the consume loop blocks on + /// the in-flight batch instead of racing it. + dispatch_lock: TokioMutex<()>, /// Set once, before any record is dispatched, via /// [`KkvV1Notifier::with_ack_sink`]. Shared between /// `drain_now` (inline path) and the background timer task so @@ -161,11 +178,12 @@ pub struct KkvV1Notifier { state: Arc, timer_task: Option>, max_records: u64, - /// Per-mirror readiness handle. `on_record` consults - /// `cache_state.is_mirror_ready(&mirror_name)` and drops records - /// whose source offset hasn't crossed the mirror's bootstrap - /// high-watermark yet. Matches the legacy kkv `KafkaCache` Stage - /// gate which suppressed push notifications until `Polling`. + /// Per-mirror suppression handle. `on_record` consults + /// `cache_state.is_record_suppressed(&mirror_name, offset)` and + /// drops records below the mirror's suppression threshold + /// (committed offset, or bootstrap high-watermark on a fresh + /// deploy). Matches the legacy kkv `KafkaCache` Stage gate which + /// suppressed push notifications until `Polling`. cache_state: Arc, mirror_name: String, } @@ -232,7 +250,9 @@ impl KkvV1Notifier { buffer: TokioMutex::new(Buffer::default()), new_data: TokioNotify::new(), shutting_down: AtomicBool::new(false), - error_state: TokioMutex::new(None), + error_state: Arc::new(TokioMutex::new(None)), + error_signal: Arc::new(TokioNotify::new()), + dispatch_lock: TokioMutex::new(()), ack_sink: OnceLock::new(), }); @@ -267,9 +287,30 @@ impl KkvV1Notifier { self } + /// Handle for the supervisor to observe a terminal dispatch + /// error without owning the notifier. `on_record` surfaces + /// timer-task errors on the next record, but an idle topic never + /// produces that next record; racing the run loop against this + /// watch closes that gap. + pub fn terminal_error_watch(&self) -> TerminalErrorWatch { + TerminalErrorWatch { + error_state: Arc::clone(&self.state.error_state), + signal: Arc::clone(&self.state.error_signal), + } + } + /// Drain the current buffer (if any) and dispatch it. Used from /// both the on_record max-records path and shutdown. async fn drain_now(&self) -> Result<(), NotifyError> { + let _dispatch = self.state.dispatch_lock.lock().await; + // The timer task may have failed terminally while we waited + // for the lock. Dispatching (and acking) a later batch after + // an earlier one is known-undelivered would let the source + // commit advance past the failed batch; surface the error + // instead and leave the buffer for the post-restart replay. + if let Some(err) = self.state.error_state.lock().await.take() { + return Err(err); + } let batch = { let mut buf = self.state.buffer.lock().await; buf.take(self.inner.partition) @@ -282,7 +323,8 @@ impl KkvV1Notifier { // Successful dispatch through every endpoint => the batch is // delivered. Tell the supervisor's ack tracker so the // periodic source-commit task can advance the broker-side - // committed offset. + // committed offset. Still under the dispatch lock, so acks + // arrive in batch order. if let Some(ack) = self.state.ack_sink.get() { ack.note_through(high + 1); } @@ -291,9 +333,32 @@ impl KkvV1Notifier { } impl Inner { + /// Metric labels from the construction-time mirror identity. + /// The `MIRROR_LABELS` task-local is not available here: the + /// timer task and the flush drainer are `tokio::spawn`ed at + /// construction time, outside the run loop's scope, so + /// `current_labels()` would report `unknown/0` for every drain + /// they dispatch. + fn labels(&self) -> (String, String) { + (self.topic.clone(), self.partition.to_string()) + } + async fn dispatch_drained(&self, batch: DrainedBatch) -> Result<(), NotifyError> { + let high_offset = batch.high_offset(); let payload = KkvV1Payload::new(&self.topic, batch.offsets, batch.updates); - self.dispatch_batch(&payload).await + let batch_keys = payload.updates.len(); + let start = std::time::Instant::now(); + self.dispatch_batch(&payload).await?; + tracing::info!( + topic = %self.topic, + partition = self.partition, + batch_keys, + high_offset, + targets = self.endpoints.len(), + elapsed_ms = start.elapsed().as_millis() as u64, + "notify sent" + ); + Ok(()) } /// POST a single batch payload to every configured endpoint @@ -332,61 +397,127 @@ impl Inner { } } - /// Fan-out dispatch: resolve, then concurrent POSTs per address. - /// First per-address error wins (subsequent results are still - /// awaited so we don't leak in-flight requests). + /// Fan-out dispatch. The retry loop sits at the endpoint level: + /// each attempt resolves the address set fresh (the cache is + /// invalidated on any failure), POSTs once per address + /// concurrently, and retries the whole set on any retryable + /// per-address failure. Retrying a pinned per-address IP instead + /// would turn every receiver rolling restart into a mirror + /// crash: the dead pod IP eats the full retry budget while the + /// replacement pod is one re-resolve away. Addresses that + /// already accepted the batch get it again on a retry round; + /// kkv invalidations are idempotent and this is within + /// at-least-once. async fn dispatch_dns_a( &self, endpoint: &Endpoint, state: &DnsAState, payload: &KkvV1Payload<'_>, ) -> Result<(), NotifyError> { - let addrs = state.resolve_or_cached(self.resolver.as_ref()).await?; - if addrs.is_empty() { - return Err(NotifyError::Transport(format!( - "dns-a resolution of {} returned 0 addresses", - state.host - ))); - } - let futures = addrs.iter().map(|sa| { - let mut per_addr_url = endpoint.url.clone(); - // Set host to the IP literal; set port to the resolved - // socket's port (matches the URL's port in production, - // but lets test stubs aim at arbitrary axum servers). - // Both setters return `Result<(), …>` for malformed - // inputs; IPs and small ports never fail here so unwrap - // is justified. - per_addr_url - .set_ip_host(sa.ip()) - .expect("set_ip_host on a valid URL always succeeds for an IpAddr"); - per_addr_url - .set_port(Some(sa.port())) - .expect("set_port on a valid URL with an http(s) scheme succeeds"); - let host_label = sa.to_string(); - async move { - self.dispatch_to_address(&endpoint.client, per_addr_url, &host_label, payload) - .await + let body = serde_json::to_vec(payload) + .map_err(|e| NotifyError::Transport(format!("payload serialization failed: {e}")))?; + let offsets_header = serde_json::to_string(&payload.offsets).map_err(|e| { + NotifyError::Transport(format!("offsets header serialization failed: {e}")) + })?; + + let mut attempt: u32 = 1; + loop { + let addrs = state.resolve_or_cached(self.resolver.as_ref()).await?; + if addrs.is_empty() { + return Err(NotifyError::Transport(format!( + "dns-a resolution of {} returned 0 addresses", + state.host + ))); } - }); - let results = join_all(futures).await; - let mut first_err: Option = None; - for r in results { - if let Err(e) = r { - first_err.get_or_insert(e); + let futures = addrs.iter().map(|sa| { + let mut per_addr_url = endpoint.url.clone(); + // Set host to the IP literal; set port to the resolved + // socket's port (matches the URL's port in production, + // but lets test stubs aim at arbitrary axum servers). + // Both setters return `Result<(), …>` for malformed + // inputs; IPs and small ports never fail here so unwrap + // is justified. + per_addr_url + .set_ip_host(sa.ip()) + .expect("set_ip_host on a valid URL always succeeds for an IpAddr"); + per_addr_url + .set_port(Some(sa.port())) + .expect("set_port on a valid URL with an http(s) scheme succeeds"); + let host_label = sa.to_string(); + let body = &body; + let offsets_header = &offsets_header; + async move { + let mut last_error = String::new(); + let outcome = self + .post_once( + &endpoint.client, + &per_addr_url, + &host_label, + body, + offsets_header, + attempt, + &mut last_error, + ) + .await; + (per_addr_url, host_label, outcome, last_error) + } + }); + let results = join_all(futures).await; + + let mut want_retry = false; + let mut failures: Vec<(Url, String, Outcome, NotifyOutcome, String)> = Vec::new(); + for (url, host_label, outcome, last_error) in results { + if matches!(outcome, Outcome::TwoXx) { + continue; + } + let policy = self.outcomes.for_outcome(outcome); + if policy.retry { + want_retry = true; + } + failures.push((url, host_label, outcome, policy, last_error)); } - } - match first_err { - Some(e) => { - // Per-spec: "Re-resolve when the cache TTL expires - // OR when an address fails repeatedly." Failure - // invalidates the cached set immediately so the next - // dispatch (after the supervisor restarts the - // mirror) picks up any K8s scale-down that happened - // mid-batch. - state.invalidate_cache().await; - Err(e) + if failures.is_empty() { + return Ok(()); } - None => Ok(()), + + // Any failure means the cached set may be stale (K8s + // scale-down or rollout mid-batch); the next attempt (or + // the next batch) re-resolves. + state.invalidate_cache().await; + + if want_retry && attempt < self.retry.max_attempts { + let reasons: Vec = failures + .iter() + .map(|(_, host, outcome, _, err)| format!("{host}: {outcome:?} {err}")) + .collect(); + tracing::warn!( + endpoint = %endpoint.url, + attempt, + max_attempts = self.retry.max_attempts, + failed_addresses = %reasons.join("; "), + "notify dns-a retry with fresh resolution" + ); + let backoff = backoff_for_attempt(self.retry.backoff_ms, attempt); + tokio::time::sleep(backoff).await; + attempt += 1; + continue; + } + + // Terminal: apply each failing address's final action; + // any `fail` fails the batch. + let mut first_err: Option = None; + for (url, host_label, outcome, policy, last_error) in failures { + if let Err(e) = self + .apply_final_action(&url, &host_label, outcome, policy, attempt, last_error) + .await + { + first_err.get_or_insert(e); + } + } + return match first_err { + Some(e) => Err(e), + None => Ok(()), + }; } } @@ -414,35 +545,17 @@ impl Inner { let mut attempt: u32 = 1; let mut last_error: String = String::new(); loop { - let (topic_l, partition_l) = current_labels(); - // Per-attempt retry gauge; spec says 1-based, 0 when idle. - metrics::gauge!( - "mirror_v3_notify_inflight_retry", - "topic" => topic_l.clone(), - "partition" => partition_l.clone(), - "target_host" => target_host.to_string(), - ) - .set(attempt as f64); - - let start = std::time::Instant::now(); - let result = client - .post(url.clone()) - .header("content-type", "application/json") - .header("x-kkv-topic", &self.topic) - .header("x-kkv-offsets", &offsets_header) - .body(body.clone()) - .send() + let outcome = self + .post_once( + client, + &url, + target_host, + &body, + &offsets_header, + attempt, + &mut last_error, + ) .await; - - metrics::histogram!( - "mirror_v3_notify_post_duration_seconds", - "topic" => topic_l.clone(), - "partition" => partition_l.clone(), - "target_host" => target_host.to_string(), - ) - .record(start.elapsed().as_secs_f64()); - - let outcome = classify(result, &mut last_error); let policy = self.outcomes.for_outcome(outcome); tracing::debug!( @@ -456,21 +569,6 @@ impl Inner { ); if matches!(outcome, Outcome::TwoXx) { - // Reset retry gauge on success. - metrics::gauge!( - "mirror_v3_notify_inflight_retry", - "topic" => topic_l.clone(), - "partition" => partition_l.clone(), - "target_host" => target_host.to_string(), - ) - .set(0.0); - metrics::counter!( - "mirror_v3_notify_batches_total", - "topic" => topic_l, - "partition" => partition_l, - "result" => "ok", - ) - .increment(1); return Ok(()); } @@ -503,6 +601,70 @@ impl Inner { } } + /// One POST attempt against one address: emits the per-attempt + /// retry gauge and duration histogram, classifies the response, + /// and on 2xx resets the gauge and counts the batch as ok. + /// Shared by the fan-out: none per-address retry loop and the + /// dns-a endpoint-level retry loop. + #[allow(clippy::too_many_arguments)] + async fn post_once( + self: &Inner, + client: &Client, + url: &Url, + target_host: &str, + body: &[u8], + offsets_header: &str, + attempt: u32, + last_error: &mut String, + ) -> Outcome { + let (topic_l, partition_l) = self.labels(); + // Per-attempt retry gauge; spec says 1-based, 0 when idle. + metrics::gauge!( + "mirror_v3_notify_inflight_retry", + "topic" => topic_l.clone(), + "partition" => partition_l.clone(), + "target_host" => target_host.to_string(), + ) + .set(attempt as f64); + + let start = std::time::Instant::now(); + let result = client + .post(url.clone()) + .header("content-type", "application/json") + .header("x-kkv-topic", &self.topic) + .header("x-kkv-offsets", offsets_header) + .body(body.to_vec()) + .send() + .await; + + metrics::histogram!( + "mirror_v3_notify_post_duration_seconds", + "topic" => topic_l.clone(), + "partition" => partition_l.clone(), + "target_host" => target_host.to_string(), + ) + .record(start.elapsed().as_secs_f64()); + + let outcome = classify(result, last_error); + if matches!(outcome, Outcome::TwoXx) { + metrics::gauge!( + "mirror_v3_notify_inflight_retry", + "topic" => topic_l.clone(), + "partition" => partition_l.clone(), + "target_host" => target_host.to_string(), + ) + .set(0.0); + metrics::counter!( + "mirror_v3_notify_batches_total", + "topic" => topic_l, + "partition" => partition_l, + "result" => "ok", + ) + .increment(1); + } + outcome + } + async fn apply_final_action( self: &Inner, url: &Url, @@ -512,7 +674,7 @@ impl Inner { attempts: u32, last_error: String, ) -> Result<(), NotifyError> { - let (topic_l, partition_l) = current_labels(); + let (topic_l, partition_l) = self.labels(); // Reset retry gauge regardless of outcome; the request is // no longer in flight. metrics::gauge!( @@ -615,10 +777,8 @@ impl DnsAState { impl Notifier for KkvV1Notifier { async fn on_record(&mut self, record: &Record) -> Result<(), NotifyError> { // First: surface any terminal error the timer task accumulated - // since the last call. Once an error is observed we still let - // the run loop hand us further records; they'll just keep - // returning the same error until the loop aborts. Take() so - // we only return it once. + // since the last call. Take() so it surfaces exactly once; + // the run loop aborts the mirror on the first Err. if let Some(err) = self.state.error_state.lock().await.take() { return Err(err); } @@ -640,7 +800,7 @@ impl Notifier for KkvV1Notifier { .cache_state .is_record_suppressed(&self.mirror_name, record.source_offset) { - let (topic_l, partition_l) = current_labels(); + let (topic_l, partition_l) = self.inner.labels(); metrics::counter!( "mirror_v3_notify_suppressed_records_total", "topic" => topic_l, @@ -657,7 +817,7 @@ impl Notifier for KkvV1Notifier { // on edge cases instead of crashing. let key_str = render_key(record.key.as_deref()); - let (topic_l, partition_l) = current_labels(); + let (topic_l, partition_l) = self.inner.labels(); metrics::counter!( "mirror_v3_notify_records_total", "topic" => topic_l.clone(), @@ -751,6 +911,11 @@ async fn timer_loop(inner: Arc, state: Arc, max_time: Dura if state.shutting_down.load(Ordering::SeqCst) { return; } + // Serialize with the inline drain path; see the + // `dispatch_lock` field docs. Taken before the buffer take so + // batches dispatch, and therefore ack, in source-offset + // order. + let _dispatch = state.dispatch_lock.lock().await; let batch = { let mut buf = state.buffer.lock().await; buf.take(inner.partition) @@ -760,8 +925,11 @@ async fn timer_loop(inner: Arc, state: Arc, max_time: Dura if let Err(e) = inner.dispatch_drained(batch).await { // Stash for the next on_record / shutdown to surface; // exit so the buffer doesn't grow further behind a - // broken receiver. + // broken receiver. The signal wakes any + // TerminalErrorWatch, which covers the idle-topic + // case where no next on_record ever comes. *state.error_state.lock().await = Some(e); + state.error_signal.notify_one(); return; } // Same ack semantics as `drain_now`: successful POST @@ -828,12 +996,23 @@ pub struct FlushDispatcher { #[allow(dead_code)] inner: Arc, tx: tokio::sync::mpsc::UnboundedSender, - drainer: Option>, + /// Behind a mutex so [`Self::drain_and_stop`] can join the task + /// through `&self`: in production the dispatcher lives inside + /// the tee as an `Arc`, and the supervisor + /// holds a second `Arc` clone for the shutdown drain. + drainer: TokioMutex>>, error_state: Arc>>, - /// Per-mirror readiness handle. `on_flushed` consults - /// `cache_state.is_mirror_ready(&mirror_name)` and drops events - /// arriving before the mirror's bootstrap high-watermark is - /// crossed. Matches the source-consume gate on [`KkvV1Notifier`]. + /// Signalled when the drainer stashes a terminal error; see + /// [`TerminalErrorWatch`]. Without a watcher a drainer death is + /// otherwise invisible in production: nothing calls + /// `last_error`/`shutdown`, later `on_flushed` sends fail + /// silently, and flush events (unlike source records) are not + /// regenerated by a restart. + error_signal: Arc, + /// Per-mirror suppression handle. `on_flushed` consults + /// `cache_state.is_record_suppressed(&mirror_name, to)` and + /// drops events below the mirror's suppression threshold. + /// Matches the source-consume gate on [`KkvV1Notifier`]. cache_state: Arc, mirror_name: String, topic: String, @@ -850,6 +1029,37 @@ enum FlushEvent { Shutdown, } +/// Awaitable handle onto a notifier's / dispatcher's terminal +/// dispatch error. The supervisor races this against the mirror run +/// loop so a dead webhook pipeline errors the mirror (and thereby +/// the process) instead of going silent: the orchestrator restarts, +/// and the post-restart replay-from-committed-offset re-delivers +/// what the dead pipeline dropped. +pub struct TerminalErrorWatch { + error_state: Arc>>, + signal: Arc, +} + +impl TerminalErrorWatch { + /// Resolve once a terminal error is stashed, consuming it. + /// Pending forever if dispatch never fails terminally. If the + /// run loop consumes the error first (the `on_record` path), + /// this stays pending; the run loop's own error wins the race, + /// which is fine because either way the mirror errors exactly + /// once. + pub async fn wait(self) -> NotifyError { + loop { + if let Some(err) = self.error_state.lock().await.take() { + return err; + } + // notify_one stores a permit when there's no waiter yet, + // so a stash happening between the check above and this + // await cannot be missed. + self.signal.notified().await; + } + } +} + impl FlushDispatcher { pub fn from_config( notify: &mirror_config::Notify, @@ -879,18 +1089,21 @@ impl FlushDispatcher { let inner = Arc::new(build_inner(notify, topic.clone(), partition, resolver)?); let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); let error_state = Arc::new(TokioMutex::new(None)); + let error_signal = Arc::new(TokioNotify::new()); let ack_sink: Arc>> = Arc::new(OnceLock::new()); let drainer = tokio::spawn(flush_drainer_loop( Arc::clone(&inner), rx, Arc::clone(&error_state), + Arc::clone(&error_signal), Arc::clone(&ack_sink), )); Ok(Self { inner, tx, - drainer: Some(drainer), + drainer: TokioMutex::new(Some(drainer)), error_state, + error_signal, cache_state, mirror_name, topic, @@ -912,10 +1125,25 @@ impl FlushDispatcher { /// Drain pending events and stop the background task. Returns /// any error the drainer accumulated before exit. Idempotent - /// calling twice is safe (the second call is a no-op). + /// + /// The channel is FIFO, so awaiting the drainer after queueing + /// the Shutdown marker lets every already-queued flush event + /// dispatch first - aborting instead would silently drop the + /// final flush notification of a graceful shutdown, and flush + /// events are not regenerated on restart. A dispatch stuck in + /// retries holds this up for at most the retry budget + /// (max-attempts x (timeout + backoff)); beyond that the + /// orchestrator's termination grace period is the backstop. pub async fn shutdown(&mut self) -> Result<(), NotifyError> { + self.drain_and_stop().await + } + + /// `&self` version of [`Self::shutdown`] for the supervisor, + /// which holds the dispatcher behind an `Arc` (the tee owns it + /// as its `FlushObserver`). Idempotent. + pub async fn drain_and_stop(&self) -> Result<(), NotifyError> { let _ = self.tx.send(FlushEvent::Shutdown); - if let Some(handle) = self.drainer.take() { - handle.abort(); + if let Some(handle) = self.drainer.lock().await.take() { let _ = handle.await; } if let Some(err) = self.error_state.lock().await.take() { @@ -925,12 +1153,22 @@ impl FlushDispatcher { } /// Snapshot the drainer's latest error without consuming the - /// dispatcher. Used by `mirror-bin`'s status / supervision loop - /// to detect a fatal dispatch failure without waiting for - /// shutdown. + /// dispatcher. Prefer [`Self::terminal_error_watch`] for + /// supervision; this polling accessor remains for tests and + /// one-shot status checks. pub async fn last_error(&self) -> Option { self.error_state.lock().await.take() } + + /// Handle for the supervisor to observe the drainer's terminal + /// error while the dispatcher itself is owned by the sink as a + /// `FlushObserver`. See [`TerminalErrorWatch`]. + pub fn terminal_error_watch(&self) -> TerminalErrorWatch { + TerminalErrorWatch { + error_state: Arc::clone(&self.error_state), + signal: Arc::clone(&self.error_signal), + } + } } impl mirror_core::FlushObserver for FlushDispatcher { @@ -970,6 +1208,7 @@ async fn flush_drainer_loop( inner: Arc, mut rx: tokio::sync::mpsc::UnboundedReceiver, error_state: Arc>>, + error_signal: Arc, ack_sink: Arc>>, ) { while let Some(event) = rx.recv().await { @@ -986,6 +1225,7 @@ async fn flush_drainer_loop( let payload = KkvV1Payload::new(&inner.topic, offsets, IndexMap::new()); if let Err(e) = inner.dispatch_batch(&payload).await { *error_state.lock().await = Some(e); + error_signal.notify_one(); return; } // Successful POST => the batch is delivered. The flush event diff --git a/crates/mirror-notify-kkv/tests/debounce.rs b/crates/mirror-notify-kkv/tests/debounce.rs index 96533fd..ba7047a 100644 --- a/crates/mirror-notify-kkv/tests/debounce.rs +++ b/crates/mirror-notify-kkv/tests/debounce.rs @@ -272,3 +272,37 @@ async fn buffer_continues_to_accept_after_inline_drain() { assert_eq!(body0["offsets"], serde_json::json!({"0": 11})); assert_eq!(body1["offsets"], serde_json::json!({"0": 13})); } + +/// Idle-topic variant of timer-drain failure: with no further +/// on_record call there is nothing to surface the stashed error, so +/// the supervisor needs the terminal-error watch to learn the +/// notify pipeline is dead. +#[tokio::test] +async fn terminal_error_watch_fires_on_timer_exhaustion_without_further_records() { + let server = TestServer::start(Reply::Status(500), vec![]).await; + let cfg = notify_pointing_at_debounced( + server.addr, + NotifyOutcomes::default(), + NotifyRetry { + max_attempts: 2, + backoff_ms: 1, + }, + 1000, + NotifyDebounce { + max_records: 100, + max_time_ms: 20, + }, + ); + let mut notifier = + KkvV1Notifier::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()).unwrap(); + let watch = notifier.terminal_error_watch(); + + // One record, then silence: the timer drains it after + // max-time-ms and exhausts retries against the 500-only server. + notifier.on_record(&rec(0, "k0")).await.unwrap(); + + let err = tokio::time::timeout(Duration::from_secs(5), watch.wait()) + .await + .expect("watch must resolve on timer-drain exhaustion"); + assert!(format!("{err}").to_lowercase().contains("exhausted")); +} diff --git a/crates/mirror-notify-kkv/tests/dispatch_serialization.rs b/crates/mirror-notify-kkv/tests/dispatch_serialization.rs new file mode 100644 index 0000000..402a343 --- /dev/null +++ b/crates/mirror-notify-kkv/tests/dispatch_serialization.rs @@ -0,0 +1,180 @@ +//! Pin the dispatch-serialization contract between the background +//! timer drain and the inline (max-records / shutdown) drain. +//! +//! The two paths share one `AckSink`, and the supervisor's tracker +//! aggregates acks with `fetch_max`. If a later batch could dispatch +//! while an earlier batch is still in flight, the later batch's +//! success would ack past the earlier one; the periodic source +//! commit would then advance past records that were never delivered, +//! and a restart would suppress them forever (at-least-once +//! violated). The notifier therefore serializes take-dispatch-ack +//! under one lock and refuses to dispatch after a terminal error. + +mod common; + +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use common::{notify_pointing_at_debounced, Reply, TestServer}; +use mirror_config::{NotifyDebounce, NotifyOutcomes, NotifyRetry}; +use mirror_core::{AckSink, CacheState, Notifier, Record, TimestampType}; +use mirror_notify_kkv::KkvV1Notifier; + +#[derive(Debug, Default)] +struct RecordingAck { + values: Mutex>, +} + +impl AckSink for RecordingAck { + fn note_through(&self, through: u64) { + self.values.lock().unwrap().push(through); + } +} + +fn ready_cache(name: &str) -> Arc { + let s = Arc::new(CacheState::new()); + s.register_mirror(name, 0, None, false); + s +} + +fn rec(offset: u64, key: &str) -> Record { + Record { + topic: "t".into(), + partition: 0, + source_offset: offset, + timestamp_ms: Some(1_700_000_000_000), + timestamp_type: TimestampType::CreateTime, + key: Some(key.as_bytes().to_vec()), + value: Some(b"v".to_vec()), + headers: vec![], + } +} + +/// Poll until the server has seen at least `n` requests. The timer +/// drain runs on its own task, so the test has to wait for its +/// dispatch to be provably in flight before scripting the interleave. +async fn wait_for_requests(server: &TestServer, n: usize, deadline: Duration) { + let started = std::time::Instant::now(); + while server.request_count() < n { + assert!( + started.elapsed() < deadline, + "server never reached {n} requests (got {})", + server.request_count() + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } +} + +/// The dev2 follow-up scenario ("commit-during-debounce", flagged +/// 2026-06-10): timer takes batch A, A's target 5xxes into a long +/// retry cycle; meanwhile the consume loop fills the buffer to +/// max-records and inline-drains batch B. B must neither dispatch +/// nor ack: it blocks behind A, and once A fails terminally the +/// inline drain surfaces A's error instead of dispatching. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn later_inline_drain_cannot_ack_past_failing_timer_batch() { + // Two scripted 500s cover exactly batch A's two attempts + // (max-attempts: 2); anything after that would get the default + // 200 and succeed, which is precisely what must not happen. + let server = TestServer::start( + Reply::Status(200), + vec![Reply::Status(500), Reply::Status(500)], + ) + .await; + let retry = NotifyRetry { + max_attempts: 2, + // Long enough that batch B's inline drain trips while A is + // parked between attempts. + backoff_ms: 300, + }; + let cfg = notify_pointing_at_debounced( + server.addr, + NotifyOutcomes::default(), + retry, + 1000, + NotifyDebounce { + max_records: 3, + max_time_ms: 20, + }, + ); + let ack = Arc::new(RecordingAck::default()); + let mut notifier = + KkvV1Notifier::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()) + .unwrap() + .with_ack_sink(ack.clone() as Arc); + + // Record 0 sits alone in the buffer; the timer takes it as + // batch A after max_time_ms and starts dispatching. + notifier.on_record(&rec(0, "k0")).await.unwrap(); + wait_for_requests(&server, 1, Duration::from_secs(5)).await; + + // Batch B: the third record trips the inline max-records drain + // while A is mid-retry. + notifier.on_record(&rec(1, "k1")).await.unwrap(); + notifier.on_record(&rec(2, "k2")).await.unwrap(); + let err = notifier.on_record(&rec(3, "k3")).await.unwrap_err(); + let msg = format!("{err}"); + assert!( + msg.to_lowercase().contains("exhausted"), + "the inline drain must surface batch A's terminal error, got: {msg}" + ); + + assert_eq!( + server.request_count(), + 2, + "only batch A's two attempts may reach the receiver; batch B must not dispatch" + ); + assert!( + ack.values.lock().unwrap().is_empty(), + "no ack may be recorded once an earlier batch is undelivered" + ); +} + +/// Happy-path ordering: when the timer batch eventually succeeds, +/// the blocked inline drain proceeds and acks arrive in source-offset +/// order. Guards against a future "fix" that unblocks the inline +/// path by skipping the lock. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn inline_drain_blocks_behind_slow_timer_batch_and_acks_in_order() { + let server = TestServer::start( + Reply::Status(200), + vec![Reply::SlowOk(Duration::from_millis(400))], + ) + .await; + let retry = NotifyRetry { + max_attempts: 2, + backoff_ms: 1, + }; + let cfg = notify_pointing_at_debounced( + server.addr, + NotifyOutcomes::default(), + retry, + // Comfortably above SlowOk so the slow reply is a success, + // not a client-side timeout. + 5000, + NotifyDebounce { + max_records: 3, + max_time_ms: 20, + }, + ); + let ack = Arc::new(RecordingAck::default()); + let mut notifier = + KkvV1Notifier::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()) + .unwrap() + .with_ack_sink(ack.clone() as Arc); + + notifier.on_record(&rec(0, "k0")).await.unwrap(); + wait_for_requests(&server, 1, Duration::from_secs(5)).await; + + notifier.on_record(&rec(1, "k1")).await.unwrap(); + notifier.on_record(&rec(2, "k2")).await.unwrap(); + // Blocks behind A's slow POST, then dispatches B. + notifier.on_record(&rec(3, "k3")).await.unwrap(); + + assert_eq!(server.request_count(), 2, "one POST per batch"); + assert_eq!( + ack.values.lock().unwrap().clone(), + vec![1, 4], + "acks must arrive in batch (source-offset) order" + ); +} diff --git a/crates/mirror-notify-kkv/tests/fan_out_dns_a.rs b/crates/mirror-notify-kkv/tests/fan_out_dns_a.rs index a5d2571..efbad9f 100644 --- a/crates/mirror-notify-kkv/tests/fan_out_dns_a.rs +++ b/crates/mirror-notify-kkv/tests/fan_out_dns_a.rs @@ -176,10 +176,12 @@ async fn one_address_failure_fails_the_whole_batch() { let err = n.on_record(&rec(1)).await.unwrap_err(); assert!(matches!(err, NotifyError::Exhausted { .. }), "got {err:?}"); - // A retried (2 attempts), B got one success POST. The - // important thing is the whole batch surfaced as failure. + // Retries happen at the endpoint level with a fresh resolution + // per round, so B receives the (idempotent) batch once per + // round alongside A's failing attempts. The important thing is + // the whole batch surfaced as failure. assert_eq!(server_a.request_count(), 2); - assert_eq!(server_b.request_count(), 1); + assert_eq!(server_b.request_count(), 2); } #[tokio::test] @@ -289,3 +291,71 @@ async fn dispatches_concurrently_to_all_addresses() { assert_eq!(server_a.request_count(), 1); assert_eq!(server_b.request_count(), 1); } + +/// Rolling-restart resolver: returns a dead address on the first +/// resolve and the live server afterwards, modelling a K8s headless +/// service whose pod was replaced between the resolve and the POST. +#[derive(Debug)] +struct RollingResolver { + first: Vec, + then: Vec, + calls: Arc, +} + +#[async_trait] +impl DnsAResolver for RollingResolver { + async fn resolve(&self, _host: &str, _port: u16) -> std::io::Result> { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + Ok(if n == 0 { + self.first.clone() + } else { + self.then.clone() + }) + } +} + +/// The receiver-rollout regression: the first resolution points at a +/// dead pod IP. Retries must re-resolve (the failure invalidates the +/// cache) and deliver to the replacement pod instead of burning the +/// whole retry budget against the dead IP and crashing the mirror. +#[tokio::test] +async fn retry_re_resolves_instead_of_pinning_the_dead_address() { + let live = TestServer::start(Reply::Status(200), vec![]).await; + // A bound-then-dropped listener yields a port that refuses + // connections. + let dead_addr = { + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + l.local_addr().unwrap() + }; + let calls = Arc::new(AtomicUsize::new(0)); + let resolver = Arc::new(RollingResolver { + first: vec![dead_addr], + then: vec![live.addr], + calls: Arc::clone(&calls), + }); + + let cfg = notify_dns_a(); + let mut n = KkvV1Notifier::from_config_with_resolver( + &cfg, + "t".into(), + 0, + ready_cache("m"), + "m".into(), + resolver, + ) + .unwrap(); + + n.on_record(&rec(1)) + .await + .expect("batch must succeed via the re-resolved address"); + + assert_eq!( + live.request_count(), + 1, + "the replacement pod must receive the batch" + ); + assert!( + calls.load(Ordering::SeqCst) >= 2, + "a retry must have re-resolved instead of reusing the dead address" + ); +} diff --git a/crates/mirror-notify-kkv/tests/flush_dispatcher.rs b/crates/mirror-notify-kkv/tests/flush_dispatcher.rs index 330399f..51b4f5b 100644 --- a/crates/mirror-notify-kkv/tests/flush_dispatcher.rs +++ b/crates/mirror-notify-kkv/tests/flush_dispatcher.rs @@ -155,3 +155,68 @@ async fn shutdown_with_no_events_is_a_noop() { .expect("empty shutdown is a noop"); assert_eq!(server.request_count(), 0); } + +/// The supervisor's only production-grade visibility into a dead +/// drainer: the watch resolves when dispatch exhausts retries, so +/// the mirror task can error out instead of running silently with +/// every subsequent flush notification dropped. +#[tokio::test] +async fn terminal_error_watch_fires_when_drainer_exhausts() { + let server = TestServer::start(Reply::Status(500), vec![]).await; + let cfg = notify_dest_flush(server.addr); + let dispatcher = + FlushDispatcher::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()) + .expect("must build"); + let watch = dispatcher.terminal_error_watch(); + + dispatcher.on_flushed(0, 9); + + let err = tokio::time::timeout(Duration::from_secs(5), watch.wait()) + .await + .expect("watch must resolve once the drainer exhausts retries"); + let msg = format!("{err}").to_lowercase(); + assert!(msg.contains("exhausted"), "got: {msg}"); +} + +/// The watch must stay pending while dispatch succeeds; resolving +/// spuriously would crash a healthy mirror. +#[tokio::test] +async fn terminal_error_watch_stays_pending_on_success() { + let server = TestServer::start(Reply::Status(200), vec![]).await; + let cfg = notify_dest_flush(server.addr); + let dispatcher = + FlushDispatcher::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()) + .expect("must build"); + let watch = dispatcher.terminal_error_watch(); + + dispatcher.on_flushed(0, 9); + wait_for_requests(&server, 1, Duration::from_secs(5)).await; + + let pending = tokio::time::timeout(Duration::from_millis(200), watch.wait()).await; + assert!( + pending.is_err(), + "watch must not resolve while dispatch succeeds" + ); +} + +/// Graceful shutdown must dispatch every already-queued flush event +/// before returning: flush events are not regenerated on restart, +/// so aborting the drainer here would lose the final flush +/// notification of every clean shutdown. +#[tokio::test] +async fn shutdown_drains_queued_flush_events_before_stopping() { + let server = TestServer::start(Reply::SlowOk(Duration::from_millis(100)), vec![]).await; + let cfg = notify_dest_flush(server.addr); + let mut dispatcher = + FlushDispatcher::from_config(&cfg, "t".into(), 0, ready_cache("m"), "m".into()) + .expect("must build"); + + dispatcher.on_flushed(0, 9); + dispatcher.on_flushed(10, 19); + dispatcher.shutdown().await.expect("drain must succeed"); + assert_eq!( + server.request_count(), + 2, + "both queued flush events must dispatch before shutdown returns" + ); +} diff --git a/crates/mirror-s3/src/lib.rs b/crates/mirror-s3/src/lib.rs index ac713d1..032b394 100644 --- a/crates/mirror-s3/src/lib.rs +++ b/crates/mirror-s3/src/lib.rs @@ -137,10 +137,19 @@ impl S3Sink { // Cache bootstrap: same shape as mirror-fs; replay durable // state into the shared CacheState. Compaction = read latest // snapshot; append + cache = scan + replay every object. + // + // The snapshot `view` is a `BTreeMap` whose + // `.values()` iterate in key order, not offset order, so + // `CacheState::apply_record`'s monotonic guard would drop + // every snapshot record whose key happens to land after a + // higher-offset key in the alphabet. Sort by `source_offset` + // ascending before dispatch (matches mirror-fs). if let Some(binding) = cfg.cache.as_ref() { match &view { Some(v) => { - for r in v.values() { + let mut snapshot_records: Vec<&Record> = v.values().collect(); + snapshot_records.sort_by_key(|r| r.source_offset); + for r in snapshot_records { binding.state.apply_record(&binding.mirror_name, r); } } @@ -491,6 +500,10 @@ impl Sink for S3Sink { fn set_flush_observer(&mut self, observer: Arc) { self.flush_observer = Some(observer); } + + fn supports_flush_observer(&self) -> bool { + true + } } fn record_byte_size(record: &Record) -> u64 { diff --git a/e2e/src/docker.rs b/e2e/src/docker.rs index f23aa76..beb1474 100644 --- a/e2e/src/docker.rs +++ b/e2e/src/docker.rs @@ -17,11 +17,14 @@ use testcontainers::{ContainerAsync, GenericImage, ImageExt}; use crate::{ProvisionedStack, Provisioner}; const REDPANDA_IMAGE: &str = "docker.io/redpandadata/redpanda"; -const REDPANDA_TAG: &str = "latest"; +const REDPANDA_TAG: &str = "v26.1.12"; const KAFKA_NATIVE_IMAGE: &str = "quay.io/ogunalp/kafka-native"; -const KAFKA_NATIVE_TAG: &str = "latest"; +// No stable release tags exist upstream (only moving main- +// builds), so pin the digest that `latest` resolved to at pin time. +const KAFKA_NATIVE_TAG: &str = + "latest@sha256:9ae667c854cd25a86a2d263e0cd7352cd6ffa6395c2dec5f90ab9056cc60b14f"; const VERSITYGW_IMAGE: &str = "docker.io/versity/versitygw"; -const VERSITYGW_TAG: &str = "latest"; +const VERSITYGW_TAG: &str = "v1.6.0"; const TOXIPROXY_IMAGE: &str = "ghcr.io/shopify/toxiproxy"; const TOXIPROXY_TAG: &str = "2.12.0"; diff --git a/e2e/tests/restart_resumes_notify_from_commit.rs b/e2e/tests/restart_resumes_notify_from_commit.rs index a80c00a..53885f3 100644 --- a/e2e/tests/restart_resumes_notify_from_commit.rs +++ b/e2e/tests/restart_resumes_notify_from_commit.rs @@ -111,8 +111,8 @@ fn fs_spec(root: &std::path::Path) -> FilesystemSinkConfig { } /// Extract the `updates` map keys from a kkv-v1 notify body. The -/// notifier POSTs JSON of shape `{"v":"v1","topic":..., "offsets": -/// {...}, "updates": {"": ""}}`. +/// notifier POSTs JSON of shape `{"v":1,"topic":..., "offsets": +/// {...}, "updates": {"": null}}`. fn keys_in_body(body: &[u8]) -> HashSet { let v: serde_json::Value = serde_json::from_slice(body).expect("notify body is JSON"); v.get("updates") @@ -376,3 +376,232 @@ async fn poll_for_committed(bootstrap: &str, group: &str, timeout: Duration) -> tokio::time::sleep(Duration::from_millis(100)).await; } } + +const TOPIC_UNACKED: &str = "mirror-e2e-unacked-window-replays"; + +/// The other half of the restart contract: records that are already +/// durable on the destination but whose notify batch was never acked +/// (committed) must re-fire after a restart. The destination state +/// alone would resume above them; the supervisor closes the gap by +/// setting the tee's resume floor to the committed offset, which +/// re-reads `[committed, durable)` from the source, skips the +/// destination writes and lets the suppression threshold (== +/// committed) admit exactly the un-acked records. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn unacked_notify_window_replays_after_restart() { + init_tracing(); + let stack = DockerProvisioner.provision().await.expect("provision"); + let source_bootstrap = stack.source_bootstrap(); + let root = tempfile::tempdir().expect("tempdir"); + create_topic(&source_bootstrap, TOPIC_UNACKED, 1) + .await + .expect("topic"); + + let group_id = format!("mirror-e2e-unacked-window-{}", uuid::Uuid::new_v4()); + let receiver = WebhookReceiver::start().await; + let mut notify = notify_pointing_at(receiver.addr); + notify.targets[0].url = format!("http://{}", receiver.addr); + + // Stage A: 5 records consumed and flushed durably, but the + // broker commit only reaches offset 3 - as if the receiver died + // after acking the first batch and the un-acked batch (offsets + // 3-4) never advanced the commit. + let pairs: Vec<(String, String)> = (0..5) + .map(|i| (format!("k{i:03}"), format!("v{i:03}"))) + .collect(); + produce_records(&source_bootstrap, TOPIC_UNACKED, 0, &pairs) + .await + .expect("produce stage A"); + + { + let cache = Arc::new(CacheState::new()); + cache.register_mirror_with_topic("notify", 0, None, false, TOPIC_UNACKED, 0); + let cache_binding = CacheBinding { + state: Arc::clone(&cache), + mirror_name: "notify".into(), + }; + let source = KafkaSource::open(KafkaSourceConfig::new( + source_bootstrap.clone(), + group_id.clone(), + TOPIC_UNACKED, + 0, + )) + .expect("open source A"); + let commit_handle = source.commit_handle(); + let fs_cfg = FilesystemSinkConfig { + cache: Some(mirror_fs::CacheBinding { + state: Arc::clone(&cache), + mirror_name: "notify".into(), + }), + destination_name: "notify".into(), + ..fs_spec(root.path()) + }; + let sink: Box = Box::new(FilesystemSink::open(fs_cfg).expect("open fs sink A")); + let tee = TeeSink::open(vec![("notify".into(), sink)], Some(cache_binding)) + .await + .expect("tee A"); + let notifier = mirror_notify_kkv::KkvV1Notifier::from_config( + ¬ify, + TOPIC_UNACKED.into(), + 0, + Arc::clone(&cache), + "notify".into(), + ) + .expect("notifier A"); + + let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false); + let signal = async move { + let _ = shutdown_rx.changed().await; + }; + let handle = tokio::spawn(async move { + run_mirror_with_notifier( + source, + tee, + notifier, + signal, + mirror_core::DEFAULT_HEARTBEAT_INTERVAL, + ) + .await + }); + + receiver.wait_for(1, Duration::from_secs(15)).await; + let _ = shutdown_tx.send(true); + handle.await.expect("join A").expect("mirror A ok"); + // The graceful shutdown flushed offsets 0-4 to the fs chain + // (durable head = 5); commit only through 3. + commit_handle.commit_through(3).expect("commit_through"); + commit_handle.commit_pending().expect("commit_pending"); + } + + let observed = poll_for_committed_on( + &source_bootstrap, + &group_id, + TOPIC_UNACKED, + Duration::from_secs(10), + ) + .await; + assert_eq!(observed, Some(3), "broker must report committed offset 3"); + + let baseline = receiver.request_count(); + + // Stage B: restart against the same fs root and group. Replicates + // the supervisor's startup recipe: threshold = committed, + // resume floor = max(committed, low watermark). + { + let cache = Arc::new(CacheState::new()); + cache.register_mirror_with_topic("notify", 5, Some(3), false, TOPIC_UNACKED, 0); + let cache_binding = CacheBinding { + state: Arc::clone(&cache), + mirror_name: "notify".into(), + }; + let source = KafkaSource::open(KafkaSourceConfig::new( + source_bootstrap.clone(), + group_id.clone(), + TOPIC_UNACKED, + 0, + )) + .expect("open source B"); + let fs_cfg = FilesystemSinkConfig { + cache: Some(mirror_fs::CacheBinding { + state: Arc::clone(&cache), + mirror_name: "notify".into(), + }), + destination_name: "notify".into(), + ..fs_spec(root.path()) + }; + let sink: Box = Box::new(FilesystemSink::open(fs_cfg).expect("open fs sink B")); + let mut tee = TeeSink::open(vec![("notify".into(), sink)], Some(cache_binding)) + .await + .expect("tee B"); + let durable_head = tee.heads().iter().map(|(_, h)| *h).min().unwrap(); + assert_eq!(durable_head, 5, "stage A must have flushed 0-4 durably"); + tee.set_resume_floor(3); + + let notifier = mirror_notify_kkv::KkvV1Notifier::from_config( + ¬ify, + TOPIC_UNACKED.into(), + 0, + Arc::clone(&cache), + "notify".into(), + ) + .expect("notifier B"); + + let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false); + let signal = async move { + let _ = shutdown_rx.changed().await; + }; + let handle = tokio::spawn(async move { + run_mirror_with_notifier( + source, + tee, + notifier, + signal, + mirror_core::DEFAULT_HEARTBEAT_INTERVAL, + ) + .await + }); + + // No new records are produced: everything that arrives at the + // receiver is the replayed un-acked window. + let deadline = std::time::Instant::now() + Duration::from_secs(15); + loop { + if receiver.request_count() > baseline { + tokio::time::sleep(Duration::from_millis(300)).await; + break; + } + if std::time::Instant::now() >= deadline { + panic!("stage B: no webhook fired for the un-acked window"); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + let all_captured = receiver.captured().await; + let mut replay_keys: HashSet = HashSet::new(); + for req in &all_captured[baseline..] { + replay_keys.extend(keys_in_body(&req.body)); + } + for i in 3..5 { + let want = format!("k{i:03}"); + assert!( + replay_keys.contains(&want), + "un-acked window key {want} must re-fire; got {replay_keys:?}" + ); + } + for i in 0..3 { + let unwanted = format!("k{i:03}"); + assert!( + !replay_keys.contains(&unwanted), + "committed key {unwanted} must stay suppressed; got {replay_keys:?}" + ); + } + + let _ = shutdown_tx.send(true); + handle.await.expect("join B").expect("mirror B ok"); + } + + // The replay must not have duplicated destination data: the + // chain still ends at offset 4. + let sink_check = FilesystemSink::open(fs_spec(root.path())).expect("re-open for check"); + let mut boxed: Box = Box::new(sink_check); + let head = boxed.next_expected_offset().await.expect("head check"); + assert_eq!(head, 5, "destination chain must be unchanged by the replay"); +} + +async fn poll_for_committed_on( + bootstrap: &str, + group: &str, + topic: &str, + timeout: Duration, +) -> Option { + let deadline = std::time::Instant::now() + timeout; + loop { + let cfg = KafkaSourceConfig::new(bootstrap.to_string(), group.to_string(), topic, 0); + let mut s = KafkaSource::open(cfg).expect("re-open"); + if let Ok(Some(off)) = s.fetch_committed_offset().await { + return Some(off); + } + if std::time::Instant::now() >= deadline { + return None; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +}