Cassandra design doc: make it self-consistent for the full fence

The doc carried leftovers of the persist-only split: an empty
"Scope: persist-only" heading, the lost "The replay window" heading
(its content sat under the Delete section), and a paragraph arguing
for persist-only against the very mode this doc describes.

- Restore "The replay window" heading; drop the empty Scope heading.
- Replace the persist-only advocacy with a "Compatibility and rollout"
  section (KIP-style) describing the actual API break - the unified
  Stored read/write replacing the get/persist/delete triple - and why
  it is accepted; move persist-only scope to Rejected alternatives.
- Restructure Consistency why-first: the read side reopens evolution-gaming#732 under
  weak levels, then the serial level, then locality-driven guidance.
- Route the intro to the user-facing Persistence page; note the
  fenced-gated events-recovery floor read; align Stored pseudo-notation
  with the ADT constructors.
- Fix stale wording: deleteCompareAndSet Scaladoc still said "get reads
  it back as None"; persistence.md said "journal" for the input topic.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PGxViqVaLYb1xQyveED6Hs

Author: tobiajo

docs/cassandra-single-writer-design.md

@@ -5,15 +5,10 @@ sidebar_label: "Cassandra single-writer design: persist only"
 ---
 
 Design notes for the compare-and-set snapshot mode of `kafka-flow-persistence-cassandra`
-(`CassandraSnapshots.withSchema(compareAndSet = true)`). What ships is small — one conditional
-predicate on each `persist` — so these notes spend little time on mechanics. Their weight is on the
-design choices: where the persist-only line is drawn and why, and the full solution (offset-gated
-deletes) that was designed, TLA+-modelled and then deferred — because the reasons it was deferred are
-the main thing a future author needs before reopening it.
-
-Operational guidance (enabling, consistency levels, TTL, rollout) is in
-[Persistence](persistence.md). The Kafka backend solves the same problem with a different mechanism —
-see [Kafka single-writer design](kafka-single-writer-design.md).
+(`CassandraSnapshots.withSchema(compareAndSet = true)`) — the mechanism and its subtleties. User-facing
+enablement, costs and rollout guidance are in
+[Persistence](persistence.md#protecting-against-stale-snapshot-writes). The Kafka backend solves the
+same problem differently — see [Kafka single-writer design](kafka-single-writer-design.md).
 
 ## Problem
 
@@ -50,8 +45,6 @@ and treats its absence (or a null) as "row absent".
 
 Three choices in the predicate carry most of the design:
 
-## Scope: persist-only — and what the full fence would cost
-
 ## Delete: offset-carrying tombstone
 
 A delete cannot be a plain `DELETE`. Removing the row removes the `offset` guard with it, so a lagging
@@ -67,16 +60,14 @@ UPDATE snapshots_v2 SET value = null, offset = :offset WHERE <key> IF offset <=
 
 keeping the row (and its `offset`). A stale lower-offset writer is then rejected, not resurrected; a
 replayed delete is a no-op (equal offset) or a conflict (a newer write exists), never a resurrection;
-a read surfaces the null `value` as a `Stored.Tombstone` (no live value). The tombstone is reaped by the TTL, if configured. Keeping
-the row also routes the delete through Paxos, avoiding the well-known hazard of mixing lightweight
-transactions and regular mutations on the same row.
+a read surfaces the null `value` as a `Stored.Tombstone` (no live value). The tombstone is reaped by
+the TTL, if configured. Keeping the row also routes the delete through Paxos, avoiding the well-known
+hazard of mixing lightweight transactions and regular mutations on the same row.
 
-**A breaking API.** Gating deletes means a delete must carry an offset: an offset-carrying logical
-*tombstone* (`SET value = null`, keeping the row and its `offset`) instead of a row removal, so a
-lower-offset writer is rejected rather than allowed to resurrect. But that offset has to reach the store
-through `SnapshotWriteDatabase.delete`, forcing a source/binary-breaking `delete(key, offset)` signature
-and offset plumbing through every delete path. Persist-only makes **no public API change** and needs no
-major-version bump.
+Gating deletes has an API cost — the offset must reach the store through the delete path, a breaking
+change for custom stores; see [Compatibility and rollout](#compatibility-and-rollout).
+
+## The replay window
 
 A delete and a re-persist are fenced on an offset that, just after recovery, can legitimately trail
 the key's own stored snapshot. The partition resumes from the committed offset `C` (the minimum offset
@@ -148,8 +139,8 @@ persisted, so a replay-window tick-delete is dispatched `persist = false` — bu
 store.)
 
 So recovery must surface the tombstone's offset as the floor, not collapse it to "nothing there".
-`SnapshotDatabase.read` returns `Some(Stored(None, X))` for a tombstone — value-less but carrying the
-offset — distinct from `None` for a reaped or never-written key (and from `Some(Stored(Some(v), X))` for a
+`SnapshotDatabase.read` returns `Stored.Tombstone(X)` for a tombstone — value-less but carrying the
+offset — distinct from `None` for a reaped or never-written key (and from `Stored.Live(v, X)` for a
 live snapshot); `Snapshots` holds `X` as the buffer high-water even with no buffered value, so a re-derived
 snapshot below `X` is dropped exactly as for a live snapshot and the owner makes progress. The deleted-key
 recovery is thus symmetric with the live one — same floor, only the value is absent — and the floor is
@@ -161,11 +152,12 @@ The same hazard reaches the **events-recovery** mode (`restoreEvents`), where st
 the journal rather than reading the snapshot. A delete clears the key's journal, so the fold yields `None`
 and the high-water `X` survives only on the snapshot tombstone — the buffer would again start with no floor.
 So events-recovery reads the snapshot store for its tombstone floor (`ReadState` runs `Snapshots.read` for
-that side-effect) before folding the journal; the recovered state still comes from the journal. A *live* key
-needs no floor here — its journal is intact, so the fold reconstructs `X` itself. Note compare-and-set
-protects the snapshot store, which events-recovery does not read for state, so pairing the two buys no
-stale-writer safety; seeding the floor only removes the deleted-key livelock for setups that nonetheless
-enable both.
+that side-effect) before folding the journal; the recovered state still comes from the journal. The read
+runs only for a fencing buffer — an unfenced one never reads back a tombstone, so the per-key round-trip
+is skipped. A *live* key needs no floor here — its journal is intact, so the fold reconstructs `X`
+itself. Note compare-and-set protects the snapshot store, which events-recovery does not read for
+state, so pairing the two buys no stale-writer safety; seeding the floor only removes the deleted-key
+livelock for setups that nonetheless enable both.
 
 ## Equal-offset writes and determinism
 
@@ -183,45 +175,53 @@ drop a lower-offset replay as a no-op.
 
 ## Consistency
 
-The lightweight transaction reaches consensus at the *serial* consistency level, which is distinct
-from the read/write levels in `ConsistencyOverrides` and defaults to `SERIAL` (a cross-datacenter
-quorum). `ConsistencyOverrides.write` governs only the transaction's commit phase, not its consensus.
-For single-datacenter partition ownership (the common case) set the scassandra client's
-`query.serial-consistency = LOCAL_SERIAL` to keep the Paxos rounds in-DC; otherwise every conditional
-write and delete pays a cross-datacenter round-trip. Recovery reads at `ConsistencyOverrides.read`;
-a serial read is not required, because `R + W > N` makes a non-serial read see every committed
+The write-side fence only closes #732 if recovery reads can see what fenced writes committed, so
+compare-and-set requires read **and** write consistency at a quorum (`R + W > N`). The conditional
+write reaches consensus on a serial quorum but *materialises* at `ConsistencyOverrides.write`; with a
+weaker write level a quorum recovery read can miss the newest committed snapshot even with no in-flight
+Paxos — #732 reintroduced on the read side, which the write-side fence does not heal. And this is not
+defaulted: `ConsistencyOverrides` is empty unless you set it, so the snapshot table inherits the
+session's default level (often `LOCAL_ONE`) — you must configure both to a quorum.
+
+A *serial* read is not required: `R + W > N` already makes a non-serial read see every committed
 snapshot, and a still-in-flight write — one whose `persist` has not completed — is safe to miss
 (recovery re-folds from the committed offset).
 
-Compare-and-set does require read and write consistency at `QUORUM` or stronger (so `R + W > N`).
-For single-datacenter ownership, `LOCAL_QUORUM` at both levels satisfies `R + W > N` within the
-local DC and pairs with `LOCAL_SERIAL`. What matters is access locality — a key's writers and its
-recovery read all in one DC — not the replication footprint, which may still span DCs for DR. The
-conditional write reaches consensus on a serial quorum but *materialises* at `ConsistencyOverrides.write`;
-with a weaker write level a (non-serial) `QUORUM` recovery read can miss the newest committed snapshot
-even with no in-flight Paxos, reintroducing #732 on the read side — which the write-side fence does not
-heal. This is not a default: `ConsistencyOverrides` is empty unless you set it, so the snapshot table
-inherits the session's default level (often `LOCAL_ONE`) — you must configure read and write to a
-quorum. Keep `R + W > N` within one consistency domain and a key's Paxos in one DC: the `LOCAL_*` set
-is for single-DC ownership, while ownership that can fail over between DCs (or write a key from two
-DCs) needs the cross-DC `SERIAL` / `QUORUM` levels.
-
-## TTL and rollout
-
-The `offset` guard lives in the snapshot row, so it expires with the row's TTL. After a row's TTL
-lapses the guard is gone and a stale write can land a fresh `INSERT`; this is harmless when the TTL far
-exceeds the rebalance/zombie overlap window (the usual case — a zombie outliving the TTL is not
-realistic), but the monotonicity guarantee only holds within the TTL.
-
-Without a TTL the offset-carrying tombstone is never reaped, so a deleted key leaves its row behind
-indefinitely and the table grows by one row per deleted key. Configure a TTL (comfortably above the
-overlap window) for workloads that delete keys.
-
-Enabling on a running system needs no migration (the condition reads the `offset` column every version
-already writes). The one rolling-deploy caveat: a lightweight transaction uses a coordinator-generated
-write timestamp while a regular write uses a client-side one, so during a mixed deploy an application
-clock running ahead of the coordinators can let an old (plain-write) instance's snapshot shadow a newer
-conditional one. Negligible with NTP-synced clocks, and gone once every instance writes conditionally.
+The Paxos consensus itself runs at the *serial* consistency level, which is separate from
+`ConsistencyOverrides` (whose write level governs only the commit phase) and defaults to `SERIAL` — a
+cross-datacenter quorum. For single-datacenter partition ownership (the common case) set the scassandra
+client's `query.serial-consistency = LOCAL_SERIAL`, or every conditional write and delete pays a
+cross-datacenter round-trip.
+
+Pick the levels by ownership locality, not by the replication footprint (which may still span DCs for
+DR): what matters is that a key's writers and its recovery read share one consistency domain. Single-DC
+ownership pairs `LOCAL_QUORUM` reads/writes with `LOCAL_SERIAL`; ownership that can fail over between
+DCs (or write a key from two DCs) needs the cross-DC `QUORUM` / `SERIAL` levels.
+
+## Compatibility and rollout
+
+**A breaking store API.** Gating deletes means a delete must carry an offset to the store, and recovery
+must read a tombstone's offset back (the replay-window floor above). The store interface therefore
+exchanges one `Stored` unit both ways — `read` distinguishes a live snapshot, an offset-carrying
+tombstone and an absent row; `write` takes a `Stored.Live` or a `Stored.Tombstone` — replacing the
+previous `get` / `persist` / `delete` triple, and the per-key buffer's delete gains the offset
+(`delete(persist, offset)`). Source- and binary-breaking for custom `SnapshotDatabase` implementations,
+hence a major-version bump. The alternative — fencing persists only — needs no API change but leaves
+every deleted key open to resurrection (see Rejected alternatives).
+
+**Rolling deploy.** Enabling on a running system needs no data migration (the condition reads the
+`offset` column every version already writes). The one caveat: a lightweight transaction uses a
+coordinator-generated write timestamp while a regular write uses a client-side one, so during a mixed
+deploy an application clock running ahead of the coordinators can let an old (plain-write) instance's
+snapshot shadow a newer conditional one. Negligible with NTP-synced clocks, and gone once every
+instance writes conditionally.
+
+**TTL.** The `offset` guard lives in the snapshot row, so it expires with the row's TTL: once a row's
+TTL lapses a stale write can land a fresh `INSERT`. Harmless when the TTL far exceeds the
+rebalance/zombie overlap window (the usual case — a zombie outliving the TTL is not realistic), but the
+monotonicity guarantee only holds within the TTL. Without a TTL the offset-carrying tombstone is never
+reaped — the table grows by one row per deleted key — so configure a TTL (comfortably above the overlap
+window) for workloads that delete keys.
 
 ## Implementation
 
@@ -231,18 +231,18 @@ Entry point: `CassandraSnapshots.withSchema(compareAndSet = true)` (or
 - **Unified write / read** — `CassandraSnapshots.write` routes a `Stored`: a present value to
   `persistCompareAndSet`, an absent value (a delete) to `deleteCompareAndSet`, both issuing the offset-gated
   `UPDATE`/`INSERT`; `resolveConditional` classifies the result (`applied` / newer-stored-offset /
-  row-absent), shared by both. `read` returns the stored unit — `Some(Stored(value, offset))`, with an
-  absent value for a tombstone — or `None` for no row.
+  row-absent), shared by both. `read` returns the stored unit — a `Stored.Live` or, for a tombstone, a
+  `Stored.Tombstone` — or `None` for no row.
 - **Monotonic buffer** — `Snapshots` holds one `Stored` cell (a live snapshot or an offset-carrying
   tombstone floor); `append` and `delete` both flow through one monotonic `put` that drops a lower-offset
   re-persist and lifts a lower-offset delete to the high-water. Recovery seeds the cell (and thus the floor)
-  from `SnapshotDatabase.read`: `Some(Stored(None, X))` is a tombstone at `X`, `None` is a reaped or
+  from `SnapshotDatabase.read`: `Stored.Tombstone(X)` is a tombstone at `X`, `None` is a reaped or
   never-written key.
 - **Replay filter** — `SnapshotFold` drops replayed events above the recovered offset; a timer-driven
   `TickToState` delete bypasses that filter, which is why the monotonic buffer is what carries the
   tick-delete case.
 - **Events-recovery floor** — `ReadState` runs `Snapshots.read` for its tombstone-floor side effect
-  before folding the journal.
+  before folding the journal, gated on `Snapshots.fenced` (skipped for an unfenced buffer).
 - **Offset accessor** — the offset-carrying wiring passes `Some(_.offset)` (live fence); other stores
   pass `None` (unfenced, last-write-wins).
 
@@ -278,16 +278,22 @@ core changes, each forced by the previous one.**
 
 ## Rejected alternatives
 
-- **Offset-as-write-timestamp (LWW register)**: write each snapshot `USING TIMESTAMP <offset>` and
-  let Cassandra's own last-write-wins reconciliation keep the highest-offset cell — a plain quorum
-  write, much cheaper than Paxos, and a delete becomes a tombstone ordered by offset for free.
-  Rejected: equal-offset replacement breaks (at equal timestamps Cassandra breaks ties by value, not
-  write order), a rolling deploy inverts catastrophically (old instances' wall-clock timestamps
-  dominate every offset-as-timestamp value), and it discards the real write timestamps.
-- **Lease / ownership table**: a per-partition lease acquired with one LWT, then cheap plain writes.
-  The lease alone does not stop a paused leaseholder's in-flight plain writes (last-write-wins still
-  applies), so a per-write fencing token is still required — at which point the lease adds only
-  liveness and expiry concerns on top of the per-write CAS.
+- **Persist-only scope (deletes unfenced)**: gate `persist` and keep `delete` a plain row removal — no
+  store-API change, no major-version bump. Rejected as the end state: removing the row removes the
+  `offset` guard with it, so a zombie's lower-offset `INSERT ... IF NOT EXISTS` resurrects any deleted
+  key (see Delete above) — #732 stays open exactly where keys are deleted, and the plain delete mixes a
+  regular mutation into a lightweight-transaction row. Viable only as an interim step for workloads
+  that never delete keys.
+- **Offset-as-write-timestamp (LWW register)**: write each snapshot `USING TIMESTAMP <offset>` and let
+  Cassandra's last-write-wins reconciliation keep the highest-offset cell — a plain quorum write, much
+  cheaper than a Paxos round, and a delete becomes a tombstone ordered by offset. Rejected as the
+  default: equal-offset replacement breaks (at equal timestamps Cassandra breaks ties by value, not
+  write order), a rolling deploy inverts catastrophically (old instances write wall-clock timestamps
+  that dominate every offset-as-timestamp value), and it discards the real write timestamps.
+- **Lease / ownership table**: a per-partition lease acquired with one LWT, then cheap writes. The
+  lease alone does not stop a paused leaseholder's plain writes (last-write-wins still applies), so a
+  per-write fencing token is still required — at which point the lease only adds liveness/expiry
+  concerns on top of the per-write CAS.
 - **Composite `(offset, generation)` token**: gate on the consumer generation as well as the offset,
   closing the equal-offset gap and giving per-partition (not just per-key) ownership. Couples the
   self-contained Cassandra module to the live consumer generation; reasonable as a future strict mode,

docs/persistence.md

@@ -9,7 +9,7 @@ sidebar_label: Persistence
 kafka-flow keeps the state of each key in memory while it processes a partition. To survive a
 restart or a partition rebalance without replaying the whole input topic from the beginning, that
 state is **persisted** and recovered when the partition is next assigned. Persistence is optional —
-it is only needed when reprocessing the full journal on every recovery is too expensive — and two
+it is only needed when reprocessing the input topic on every recovery is too expensive — and two
 backends are provided:
 
 - **Cassandra** (`kafka-flow-persistence-cassandra`) — stores per-key *journals* (the folded

persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/snapshot/CassandraSnapshots.scala

@@ -162,7 +162,8 @@ class CassandraSnapshots[F[_]: Async, T](
     session.execute(boundStatement).void
   }
 
-  /** Deletes via an offset-gated logical tombstone (see [[Statements.prepareDelete]]); `get` reads it back as `None`.
+  /** Deletes via an offset-gated logical tombstone (see [[Statements.prepareDelete]]); `read` surfaces it as a
+    * [[Stored.Tombstone]] carrying the offset.
     *
     * A not-applied result is a benign no-op when the row is absent (an at-least-once replay, or a key never persisted)
     * and is reported as success; a not-applied result with a higher stored offset means a newer writer owns the key,