Code review by Fable 5

.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 :<full-sha>
         #   - 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

.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

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" }

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-<name>`) 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.
+

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:<digest>`
-# and commit the digest. Update both stages together.
+# Both stages are digest-pinned (tag kept for readability). To
+# update: `crane digest <image>:<tag>` 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"]

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: <JSON>` 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: <JSON>` 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.