Address partition ownership overlap possibility [SPLIT UP TO 4 PRs; STALE]

[tobiajo]

SPLIT UP TO:

1. Protect Kafka snapshot writes from stale writers (transactional offset binding) #833
2. Protect Cassandra snapshot writes on persist (compare-and-set) #838
3. Protect Cassandra snapshot writes from stale writers (full compare-and-set fence) #834
4. Formal and empirical verification of the single-writer designs #835

Protect snapshot persistence from stale writers (single-writer guarantees)

Problem

During a partition rebalance, the previous owner and the new owner of a partition can briefly
overlap. If the old owner flushes its (now stale) in-memory state after the new owner has already
recovered and advanced, the stale flush overwrites the newer snapshot — the classic
ownership-overlap corruption (reproduced as issue #732: a stale flush-on-revoke silently
overwrites the newer snapshot, and the lost events never come back).

kafka-flow had no mechanism to reject a write from a writer that no longer owns the partition. This
PR adds one for both built-in snapshot backends (behind opt-in flags), and gives a custom backend a
hook to do the same.

Solution

Two backends, two fencing mechanisms, one shared invariant: a write from a writer that no longer
owns the partition must not land.

Cassandra — compare-and-set (compareAndSet = true)

• Each snapshot write is a lightweight transaction guarded by the stored offset
  (IF offset <= :offset): a stale writer carrying a lower offset is rejected with
  SnapshotWriteConflict instead of overwriting the newer snapshot.
• A delete becomes an offset-carrying logical tombstone (SET value = null, keeping the row's
  offset) rather than a hard DELETE. Keeping the row preserves the offset guard across the
  delete, so a stale writer can neither erase a newer snapshot nor resurrect the key at a lower
  offset. get reads the tombstone back as absent. The tombstone is reaped by the optional ttl.
• The in-memory snapshot buffer is kept monotonic in offset so a write inside the replay window
  is not self-fenced by the owner's own higher-water offset (this was a liveness trap — fixed and
  regression-tested).

Kafka — transactional offset binding (cachingTransactional)

• Snapshot writes run as group-committed Kafka transactions that also commit the input offset
  (sendOffsetsToTransaction), so the snapshot and the offset advance atomically.
• Fencing is by consumer generation (KIP-447): a stale generation is rejected by the broker,
  aborting the transaction, so neither the writes nor the offset land. The consumer's
  group metadata is captured on assignment and read live per transaction.
• Recovery reads with read_committed, so records from an aborted (fenced) transaction are never
  recovered as snapshots. Output stays at-least-once.

Custom storage (bring-your-own)

A custom SnapshotDatabase is last-write-wins — exposed to #732 like last-write-wins Cassandra — unless
its own persist/delete reject a write when a newer offset is stored, the way CassandraSnapshots
compare-and-set does. That conditional write is the stale-writer fence; the buffer wiring is not.
Conditional writes then need the buffer's replay fence for liveness — backedBy(db, offsetOf) (your own
offset-carrying type) or KafkaSnapshot[S] via .snapshotsOf — otherwise the owner self-fences during
recovery (its current offset lags a recovered snapshot's) and crashes. See "Custom snapshot storage" in
docs/persistence.md.

Formal models (TLA+)

The designs are backed by a refinement tower (Specifying Systems §5.8): one abstract spec,
SingleWriterStore, defines the safety + liveness contract, and each backend refines it —
Cassandra (offset compare-and-set + tombstone) and Kafka (the consumer-generation fence,
KIP-447). Finer-grained models sit under the backend they belong to: GroupCommit (the Kafka write
orchestration — termination + offset ordering), CasFirstWrite / CasFirstWriteAtomic (the
non-atomic first-write compound refining one atomic CAS), and ReplayFence (the replay-window
self-fence / liveness).

Crucially the suite also encodes rejected designs and removed guards that must fail model
checking. The clearest is Epoch — the producer-epoch variant (a stable transactional.id)
whose refinement theorem is deliberately false: epochs are handed out in initTransactions arrival
order, independent of ownership, so a late-initialising stale owner can win the epoch and its write
lands. That counterexample is precisely why the Kafka path fences on the consumer generation, not
the producer epoch. Alongside it sit the removed-guard variants (unguarded, ungated, unseeded,
not-tombstoned, decoupled…). 13 of the 24 configs are expected failures, so a green run means the
guards are load-bearing — not that only the happy path was checked. models/run.sh is self-checking
and runs the whole suite.

Testing

• Local / fast: GroupCommitSpec (group-commit orchestration against a fake producer),
  SnapshotReplayFencingSpec (offset-gated CAS semantics: resurrection, equal-offset, idempotent
  replay), SnapshotsOfSpec (the wiring fences/doesn't fence as expected), plus Snapshots /
  Persistence unit coverage.
• Integration (real broker / real Cassandra): TransactionalKafkaPersistenceSpec,
  TransactionalWriteThroughputSpec, and the issue-Partition ownership overlap during unideal circumstances #732 reproduction + prevention pair in
  FlowSpec / SnapshotSpec.

Documentation

• docs/cassandra-single-writer-design.md and docs/kafka-single-writer-design.md — design,
  limitations, costs, and rollout for each backend.
• docs/persistence.md — how a stale write is rejected and handled uniformly across backends, plus
  a "Custom snapshot storage" section for bring-your-own backends.

Change breakdown

Of 4,986 added lines across 74 files:

Category	Files	Added	Share
Production code (Scala)	16	895	~18%
Tests (Scala)	17	1,970	~40%
Formal models (TLA+)	35	1,445	~29%
Documentation (Markdown)	3	671	~13%
Build	3	~5	<1%

Roughly 58% code · 29% models · 13% docs — and ~69% of the diff (tests + models) is
verification, reflecting that the value here is the guarantee, not the line count.

Compatibility & rollout

• Breaking (major bump): SnapshotWriteDatabase.delete gains an offset parameter so deletes
  can be offset-gated. This is a source- and binary-incompatible change for anyone implementing a
  custom snapshot backend (the built-in backends are updated here).
• Additive: a new SnapshotsOf.backedBy(db, offsetOf) overload lets a custom offset-carrying
  store opt into the buffer's offset fence; the existing backedBy(db) stays last-write-wins.
• The stale-writer protections themselves are opt-in — default behavior (last-write-wins /
  non-transactional) is unchanged, so deployments on the built-in modules with default settings
  behave as before.
• Rolling deploys are supported: Cassandra CAS tolerates mixed instances (clock-skew caveat in the
  design doc); the Kafka path reaches full protection once all instances are transactional.
• In CAS mode, set a ttl to bound tombstone (and Paxos-partition) growth.

[tobiajo]

The current design for Kafka as snapshot persistence is not sufficient. It neither fully have have the proper "partition ownership" through "transactional snapshot write + offset commit" as described in in #732 nor prevents stale snapshot writes.

For Kafka it is a partial solution right now from my analysis. For Cassandra it should be sufficient.

[tobiajo]

I think we have something now that works with enough confidence - to review the production code part. It is mildly put bloated though with tests, documentation and the TLA+ [1] models. The latter part gives me more confidence, but it is basically me trying to learn formal verification, so take that AI assisted modeling with a huge grain of salt, even if I have tried to refine it.

I think the regular testing, unit tests and integration tests looks good overall. Excluding the models that probably should be dropped, the most controversial I guess is now that SnapshotWriteDatabase.delete takes an offset (so deletes can be offset-gated), which breaks any custom snapshot backend. This is a source- and binary-incompatible API change, so it would demand a major release bump.

[1] https://lamport.azurewebsites.net/tla/book.html?back-link=learning.html#book