Protect Cassandra snapshot writes on persist (compare-and-set)

docs/kafka-single-writer-design.md

@@ -5,7 +5,9 @@ sidebar_label: Kafka single-writer design
 ---
 
 Design notes for the transactional snapshot mode of `kafka-flow-persistence-kafka`
-(`KafkaPersistenceModuleOf.cachingTransactional`) — the mechanism and the measurements behind it.
+(`KafkaPersistenceModuleOf.cachingTransactional`) — the mechanism and the measurements behind it. The
+Cassandra backend solves the same problem differently — see
+[Cassandra single-writer design: persist only](cassandra-single-writer-design.md).
 
 ## Problem
 

docs/persistence.md

@@ -37,32 +37,105 @@ between the two snapshots even though their offsets were committed. See
 [kafka-flow#732](https://github.com/evolution-gaming/kafka-flow/issues/732); overlaps of tens of
 seconds have been seen in production.
 
-This page is about turning the protection on and running it; for *how* it fences a stale writer, see
-the [Kafka single-writer design](kafka-single-writer-design.md).
+This page is about turning the protection on and running it; for *how* each backend fences a stale
+writer, see the design docs:
+[Cassandra](cassandra-single-writer-design.md), [Kafka](kafka-single-writer-design.md).
 
 Timer settings change how often the window is hit:
 `TimerFlowOf.persistPeriodically(flushOnRevoke = true)` makes it **more** likely (revoked partitions
 flush while the new owner starts up); a higher `persistEvery` makes it **less** likely, at the cost of
 more events to replay on recovery.
 
-For the Kafka snapshot backend the protection is **transactional** snapshot writes — opt-in, off by
-default, enabled with `KafkaPersistenceModuleOf.cachingTransactional`. (A custom `SnapshotDatabase`
-can implement its own protection — see [Custom snapshot storage](#custom-snapshot-storage).)
+The protections are **opt-in and off by default** — pick the one for your snapshot backend:
+
+|                    | Compare-and-set (Cassandra)                  | Transactional (Kafka)                                        |
+| ------------------ | -------------------------------------------- | ------------------------------------------------------------ |
+| **Enable**         | `compareAndSet = true`                       | `KafkaPersistenceModuleOf.cachingTransactional`              |
+| **Rejects with**   | `CassandraSnapshots.SnapshotWriteConflict`   | `CommitFailedException` (the fenced offset commit)           |
+| **Per-write cost** | a Cassandra lightweight transaction (Paxos)  | a Kafka transaction (concurrent writes are group-committed)  |
 
 ### What a rejected write looks like
 
-You do not catch the rejection yourself; it is handled for you:
+You do not catch the rejection yourself; it is handled the same way for both backends:
 
-- **Periodic flush** — the conflict fails the stale instance's flow. That is safe (it no longer owns
-  the partition), unless you set `persistPeriodically(ignorePersistErrors = true)`, in which case it
-  is logged and swallowed.
+- **Periodic flush** — the conflict fails the stale instance's flow, a harmless outcome since it no
+  longer owns the partition. Setting `persistPeriodically(ignorePersistErrors = true)` logs and
+  swallows it instead, so the flow keeps running — still safe either way: the fence already rejected
+  the write, and the flag only decides whether the stale flow tears down or keeps getting rejected.
 - **Flush-on-revoke** — the conflict surfaces as a cache-entry release error that scache prints to
   `System.err` — not via the logging framework — and swallows
   (`scache: failed to release cache entry: ...`), so the partition hands off cleanly.
 
 Either way the rejected write does not land and no offset is committed for it, so the new owner
 replays the affected events.
 
+### Compare-and-set snapshot writes (Cassandra)
+
+Enable with the `compareAndSet` flag:
+
+```scala
+CassandraSnapshots.withSchema[F, State](
+  session,
+  sync,
+  compareAndSet = true,
+)
+// or via the persistence module:
+CassandraPersistence.withSchema[F, State](
+  session,
+  sync,
+  consistencyOverrides,
+  keysSegments,
+  snapshotCompareAndSet = true,
+)
+```
+
+Each snapshot **persist** becomes an offset-guarded conditional write; a stale write is rejected with
+`CassandraSnapshots.SnapshotWriteConflict`. Deletes remain ordinary last-write-wins (offset-gated deletes
+are out of scope for this mode).
+
+- **Cost** — every persist becomes a lightweight transaction (Paxos): several inter-replica
+  round-trips, a few times slower and more coordinator-CPU-intensive than a quorum write. A
+  `persistEvery` wave flushes a partition's whole changed-key population, so the added load scales with
+  that wave.
+- **Consistency** — set `ConsistencyOverrides` read **and** write to a quorum (`QUORUM`, or
+  `LOCAL_QUORUM` single-DC); they are **not** defaulted, and the usual `LOCAL_ONE` default is too weak.
+  Recovery reads at the regular (non-serial) level, so it sees the fenced write only when `R + W > N`; a
+  too-weak read still lets the write-side LWT apply but can miss the newest snapshot, silently
+  reintroducing #732 on the read side. Single-DC ownership (a key contended only within one DC, whatever
+  its replication footprint) also needs `query.serial-consistency = LOCAL_SERIAL` on
+  the scassandra client — the LWT's serial level is separate and defaults to cross-DC `SERIAL`, so a
+  conditional write otherwise pays a cross-DC round-trip.
+- **TTL** — set a `ttl` to bound the live key set; the TTL rides onto each key's Paxos state too
+  (`system.paxos`, written per key by every LWT), so it bounds that internal table's growth as well. (A
+  plain `delete` also leaves a Cassandra row tombstone reclaimed only after `gc_grace_seconds` — the
+  cluster default, not set here; not a tombstone-scan risk since keys are single-row partitions read by
+  point lookup, but it feeds compaction and repair under create/delete churn.)
+- **Rollout** — no migration either direction (the condition reads the `offset` column every version
+  already writes). A rolling deploy is safe, with one caveat while the two modes coexist: a
+  conditional write is timestamped by the Cassandra coordinator, a plain write by the client, so an
+  application clock running ahead of the coordinators can let an old instance's plain write shadow a
+  newer conditional one. Negligible with NTP-synced clocks (application hosts and the Cassandra
+  cluster on one source), and gone once every instance writes conditionally.
+
+Limitations:
+- **Deletes are not fenced.** A delete is a plain last-write-wins `DELETE`, issued when your fold
+  returns `None` for a key whose state was already persisted or recovered (a `None` fold for a
+  never-persisted key touches only the in-memory buffer). During a rebalance overlap a stale writer can
+  then erase a newer owner's snapshot, or resurrect a just-deleted key by writing at a lower offset —
+  #732 for that key. For any key that can be concurrently re-written, **avoid the `None` delete — fold
+  to an empty/"tombstone" state (`Some(empty)`) instead**: the deletion then rides the offset-gated
+  persist path and is protected like any other write, at the cost of the row living until its TTL (which
+  also moves the tombstone from an immediate `DELETE` to a TTL expiry). Plain `None` is safe only for
+  keys never concurrently re-persisted.
+- Offsets must be monotonic per key: after a backward consumer-group offset reset every persist
+  conflicts and the affected flows **stall** until reprocessing passes the stored offsets — to replay
+  from an earlier offset, `truncate` the snapshot table first (`CassandraSnapshots.truncate`).
+- Writes at an *equal* offset are allowed (e.g. a timer-driven state change at the same offset), so a
+  stale writer holding exactly the stored offset is not detected. It is safe: a same-offset write
+  cannot drop committed events — it does not move the recovery point.
+- The guard lives in the row, so it expires with the `ttl`: once a row's TTL lapses a stale write can
+  land a fresh `INSERT`. Harmless when the TTL far exceeds the overlap window (the usual case).
+
 ### Transactional snapshot writes (Kafka)
 
 **EXPERIMENTAL** — use at your own risk: the mechanism is design-verified but not yet proven in

persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala

@@ -3,14 +3,17 @@ package com.evolutiongaming.kafka.flow
 import cats.data.NonEmptyList
 import cats.effect.unsafe.IORuntime
 import cats.effect.{IO, Ref, Resource}
+import cats.syntax.all.*
 import com.evolutiongaming.catshelper.LogOf
 import com.evolutiongaming.kafka.flow.cassandra.{CassandraPersistence, ConsistencyOverrides}
-import com.evolutiongaming.kafka.flow.kafka.Consumer
+import com.evolutiongaming.kafka.flow.kafka.{Consumer, ScheduleCommit}
 import com.evolutiongaming.kafka.flow.key.CassandraKeys
+import com.evolutiongaming.kafka.flow.persistence.PersistenceModule
 import com.evolutiongaming.kafka.flow.registry.EntityRegistry
 import com.evolutiongaming.kafka.flow.snapshot.KafkaSnapshot
 import com.evolutiongaming.kafka.flow.timer.{TimerFlowOf, TimersOf}
 import com.evolutiongaming.retry.Retry
+import com.evolutiongaming.skafka.consumer.ConsumerGroupMetadata
 import com.evolutiongaming.skafka.consumer.{ConsumerRecord, ConsumerRecords, WithSize}
 import com.evolutiongaming.skafka.{Offset, TopicPartition}
 import scodec.bits.ByteVector
@@ -83,6 +86,133 @@ class FlowSpec extends CassandraSpec {
     test.unsafeRunSync()
   }
 
+  // Reproduces the #732 stale-writer corruption through the real kafka-flow machinery (PartitionFlow, eager recovery,
+  // fold, buffered snapshots, flush-on-revoke). The ownership overlap is simulated by two PartitionFlows over one
+  // partition: a real overlap is indistinguishable from the second flow being created while the first is still alive.
+  test("issue #732 reproduction: stale flush-on-revoke overwrites the newer snapshot (last-write-wins)") {
+    val (staleFlush, stored) = staleFlushScenario(compareAndSet = false).unsafeRunSync()
+    assertEquals(clue(staleFlush), Right(()))
+    // recovery now returns the STALE snapshot (events e6..e10 are lost although the new owner persisted them):
+    // this assertion documents the corruption of issue #732, prevented in the paired test below
+    assertEquals(clue(stored.map(_.value)), Some("e1,e2,e3,e4,e5"))
+  }
+
+  test("issue #732 prevention: stale flush-on-revoke is rejected (compareAndSet)") {
+    val (staleFlush, stored) = staleFlushScenario(compareAndSet = true).unsafeRunSync()
+    // the release itself succeeds: the rejected write surfaces as a logged-and-swallowed cache entry release error
+    // ("scache: failed to release cache entry: ... SnapshotWriteConflict"), which is the desired outcome for a
+    // partition that is being given away anyway
+    assertEquals(clue(staleFlush), Right(()))
+    // the protection: the stale write did not land, the new owner's snapshot survived
+    assertEquals(clue(stored.map(_.value)), Some((1 to 10).map(i => s"e$i").mkString(",")))
+  }
+
+  /** The #732 scenario: the previous owner (flow A) folds events e1..e5 without flushing; the new owner (flow B)
+    * recovers (nothing was persisted or committed by A), folds events e1..e10, and flushes on release; then A - unaware
+    * of the handover - flushes its stale state on revoke.
+    *
+    * Returns (result of A's release, stored snapshot after A's release).
+    */
+  private def staleFlushScenario(
+    compareAndSet: Boolean
+  ): IO[(Either[Throwable, Unit], Option[KafkaSnapshot[String]])] = {
+    val appId   = if (compareAndSet) "FlowSpec-732-cas" else "FlowSpec-732-lww"
+    val groupId = "integration-tests-1"
+    val tp      = TopicPartition.empty
+    val key     = "key-732"
+
+    val eventsA = (1 to 5).toList.map(i => s"e$i")
+    val eventsB = (1 to 10).toList.map(i => s"e$i")
+
+    for {
+      storage <- CassandraPersistence.withSchema[IO, String](
+        cassandra().session,
+        cassandra().sync,
+        ConsistencyOverrides.none,
+        CassandraKeys.DefaultSegments,
+        snapshotCompareAndSet = compareAndSet,
+      )
+      // the previous owner: folds events, snapshots stay buffered in memory
+      flowA             <- allocateStaleFlow(storage, appId, groupId, tp)
+      (flowA_, releaseA) = flowA
+      _                 <- flowA_(staleFlowRecords(eventsA, key, tp))
+      // the new owner: eagerly recovers (finds nothing), folds all events, flushes on release
+      flowB             <- allocateStaleFlow(storage, appId, groupId, tp)
+      (flowB_, releaseB) = flowB
+      _                 <- flowB_(staleFlowRecords(eventsB, key, tp))
+      _                 <- releaseB
+      newOwnerWrote     <- storage.snapshots.get(KafkaKey(appId, groupId, tp, key))
+      _                  = assertEquals(clue(newOwnerWrote.map(_.value)), Some(eventsB.mkString(",")))
+      // the previous owner flushes its stale state on revoke
+      staleFlush <- releaseA.attempt
+      stored     <- storage.snapshots.get(KafkaKey(appId, groupId, tp, key))
+    } yield (staleFlush, stored)
+  }
+
+  // state is the comma-joined list of folded events; the snapshot offset is the offset of the folded record
+  private val staleFlowFold: FoldOption[IO, KafkaSnapshot[String], ConsumerRecord[String, ByteVector]] =
+    FoldOption.of { (state, record) =>
+      IO {
+        val event = record.value.flatMap(_.value.decodeUtf8.toOption).getOrElse(sys.error("event payload missing"))
+        val value = state.fold(event)(s => s"${s.value},$event")
+        KafkaSnapshot(offset = record.offset, value = value).some
+      }
+    }
+
+  private def staleFlowRecords(
+    events: List[String],
+    key: String,
+    tp: TopicPartition,
+  ): List[ConsumerRecord[String, ByteVector]] =
+    events.zipWithIndex.map {
+      case (event, offset) =>
+        ConsumerRecord[String, ByteVector](
+          topicPartition   = tp,
+          offset           = Offset.unsafe(offset.toLong),
+          timestampAndType = None,
+          key              = Some(WithSize(key)),
+          value            = Some(WithSize(ByteVector.encodeUtf8(event).toOption.get)),
+        )
+    }
+
+  private def allocateStaleFlow(
+    storage: PersistenceModule[IO, String],
+    appId: String,
+    groupId: String,
+    tp: TopicPartition,
+    fold: FoldOption[IO, KafkaSnapshot[String], ConsumerRecord[String, ByteVector]] = staleFlowFold,
+  ): IO[(PartitionFlow[IO], IO[Unit])] = {
+    val flow = for {
+      timersOf      <- Resource.eval(TimersOf.memory[IO, KafkaKey])
+      keysOf        <- Resource.eval(storage.keys.toKeysOf)
+      persistenceOf <- Resource.eval(storage.snapshotsOnly)
+      keyStateOf = KeyStateOf.eagerRecovery[IO, KafkaSnapshot[String]](
+        applicationId = appId,
+        groupId       = groupId,
+        keysOf        = keysOf,
+        timersOf      = timersOf,
+        persistenceOf = persistenceOf,
+        // snapshots are flushed only when the flow is released (flush-on-revoke), never periodically -
+        // so the moment of the stale write is controlled by the test
+        timerFlowOf = TimerFlowOf.persistPeriodically[IO](
+          fireEvery     = 1.hour,
+          persistEvery  = 1.hour,
+          flushOnRevoke = true,
+        ),
+        fold     = fold,
+        tick     = TickOption.id[IO, KafkaSnapshot[String]],
+        registry = EntityRegistry.empty[IO, KafkaKey, KafkaSnapshot[String]],
+      )
+      partitionFlowOf = PartitionFlowOf(keyStateOf, PartitionFlowConfig(commitOnRevoke = true))
+      // no consumer drives this flow: the group-metadata reader is None (the Cassandra fence does not use it)
+      flow <- partitionFlowOf(
+        PartitionAssignment(tp, Offset.min, IO.pure(none[ConsumerGroupMetadata])),
+        ScheduleCommit.empty[IO],
+      )
+    } yield flow
+    flow.allocated
+  }
+
   implicit val log: LogOf[IO] = LogOf.slf4j[IO].unsafeRunSync()(IORuntime.global)
 
 }