Formal and empirical verification of the single-writer designs

.codacy.yml

@@ -0,0 +1,7 @@
+---
+# lib/skafka is a vendored source fork of skafka; keep it verbatim (upstream
+# style) so the fork stays a minimal diff. This mirrors the same exclusion in
+# .scalafmt.conf so Codacy does not flag upstream code we intentionally do not
+# reformat.
+exclude_paths:
+  - "lib/skafka/**"

.github/workflows/models.yml

@@ -0,0 +1,37 @@
+# Runs the TLA+ model suite (models/run.sh) in CI — the models were previously verified only
+# locally. run.sh fetches the pinned tla2tools release v1.7.4 (jar self-reports TLC 2.19
+# rev 5a47802; the suite is verified against it) and checks every MC_*.tla wrapper's declared outcome
+# (HOLDS / VIOLATES*) by TLC exit code — no output parsing, so a future TLC bump needs only a suite
+# re-run. Guarded to branches carrying models/ (see below), so it is inert on the code-only branches
+# lower in the stack.
+name: Models (TLA+)
+
+on:
+  push:
+    branches: [ master ]
+  pull_request:
+
+jobs:
+  tlc:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+
+    - name: Skip if this branch has no models
+      id: guard
+      run: |
+        if [ -f models/run.sh ]; then echo "run=true" >> "$GITHUB_OUTPUT"; else echo "run=false" >> "$GITHUB_OUTPUT"; echo "no models/ on this branch — nothing to check"; fi
+
+    - name: setup Java 21
+      if: steps.guard.outputs.run == 'true'
+      uses: actions/setup-java@v4
+      with:
+        java-version: '21'
+        distribution: 'temurin'
+
+    - name: Run TLC suite (models/run.sh — fetches tla2tools release v1.7.4 / TLC 2.19)
+      if: steps.guard.outputs.run == 'true'
+      working-directory: ./models
+      run: |
+        java -version
+        ./run.sh

.scalafmt.conf

@@ -1,6 +1,8 @@
 // Trial version of the config
 // https://scalameta.org/scalafmt/docs/configuration.html
 project.git = true
+// lib/skafka is a vendored source fork of skafka; keep it verbatim (upstream style) so the fork stays a minimal diff
+project.excludePaths = ["glob:**/lib/skafka/**"]
 
 runner.dialect = scala213source3
 

build.sbt

@@ -60,7 +60,32 @@ lazy val root = (project in file("."))
     scalaVersion := Scala2Version,
   )
 
+// Vendored fork of skafka 20.2.0 (sources under lib/skafka), used instead of the published artifact so the
+// experimental `group.protocol=consumer` (KIP-848) support added to ConsumerConfig is reproducible from source
+// without an upstream release. Only the fork differs from upstream; the rest of skafka is verbatim.
+lazy val skafka = (project in file("lib/skafka"))
+  .settings(commonSettings)
+  .settings(
+    name := "skafka",
+    // skafka's own build does the same: its sources carry deprecations (e.g. a kafka-clients 4.2-deprecated
+    // constructor) that upstream intentionally does not fail the build on
+    scalacOptsFailOnWarn := Some(false),
+    libraryDependencies ++= Seq(
+      Cats.core,
+      Cats.effect,
+      "org.typelevel"          %% "cats-effect-std"         % "3.7.0",
+      "com.evolutiongaming"    %% "config-tools"            % "1.0.5",
+      "org.apache.kafka"        % "kafka-clients"           % "4.3.0",
+      "com.evolutiongaming"    %% "future-helper"           % "1.0.7",
+      catsHelper,
+      smetrics,
+      "org.scala-lang.modules" %% "scala-java8-compat"      % "1.0.2",
+      "org.scala-lang.modules" %% "scala-collection-compat" % "2.14.0",
+    ),
+  )
+
 lazy val core = (project in file("core"))
+  .dependsOn(skafka)
   .settings(commonSettings)
   .settings(
     name := "kafka-flow",
@@ -73,7 +98,6 @@ lazy val core = (project in file("core"))
       Monocle.core       % Test,
       catsHelper,
       scache,
-      skafka,
       sstream,
       random,
       retry,

core/src/main/scala/com/evolutiongaming/kafka/flow/persistence/Persistence.scala

@@ -11,6 +11,7 @@ import com.evolutiongaming.kafka.flow.key.Keys
 import com.evolutiongaming.kafka.flow.snapshot.SnapshotReader
 import com.evolutiongaming.kafka.flow.snapshot.Snapshots
 import com.evolutiongaming.kafka.flow.timer.Timestamps
+import com.evolutiongaming.skafka.Offset
 
 /** Provides persistence for keys, events and snapshots.
   *
@@ -35,8 +36,10 @@ trait Buffers[F[_], S, E] extends WriteToBuffers[F, S, E] {
     *
     * @param persist
     *   if `true` then also calls underlying database, flushes buffers only otherwise.
+    * @param offset
+    *   offset of the state being deleted, forwarded to the snapshot database for stale-writer protection.
     */
-  def delete(persist: Boolean): F[Unit]
+  def delete(persist: Boolean, offset: Offset): F[Unit]
 
   /** Initialize an already persisted state, used to store a state in buffers that was fetched from the database.
     *
@@ -105,19 +108,23 @@ object Persistence {
     def appendEvent(event: E)  = buffers.appendEvent(event)
     def replaceState(state: S) = buffers.replaceState(state)
 
-    // We avoid persisting `delete` unless the state is in a database, i.e.
-    // we did `flush` or actually read it from the database using `read`.
+    // We pass `persist = false` when the state was never put in a database (no `flush`, no `read`), so a
+    // buffer-only delete causes less calls to the storage and avoids producing unnecessary tombstones.
     //
-    // It causes less calls to the storage and avoid producing unnecessary
-    // tombstones in such databases as Cassandra.
-    def delete = Timestamps[F].persistedAt flatMap { persistedAt =>
-      if (persistedAt.isDefined) {
-        Timestamps[F].onPersisted *>
-          buffers.delete(true)
-      } else {
-        buffers.delete(false)
-      }
-    }
+    // A fencing snapshot store overrides this and writes the tombstone anyway (see `Snapshots.delete`): there the
+    // tombstone is the offset gate that stops a zombie resurrecting a never-persisted, just-deleted key, so it is
+    // not unnecessary. `persist` is honored as-is by the unfenced (last-write-wins) store.
+    def delete = for {
+      persistedAt <- Timestamps[F].persistedAt
+      current     <- Timestamps[F].current
+      _ <-
+        if (persistedAt.isDefined) {
+          Timestamps[F].onPersisted *>
+            buffers.delete(true, current.offset)
+        } else {
+          buffers.delete(false, current.offset)
+        }
+    } yield ()
 
     def flush = Timestamps[F].persistedAt flatMap { persistedAt =>
       val flushAll = if (persistedAt.isEmpty) {
@@ -140,12 +147,12 @@ object Persistence {
 object Buffers {
 
   def empty[F[_]: Applicative, S, E]: Buffers[F, S, E] = new Buffers[F, S, E] {
-    def appendEvent(event: E)        = ().pure[F]
-    def replaceState(state: S)       = ().pure[F]
-    def initPersistedState(state: S) = ().pure[F]
-    def flushKeys                    = ().pure[F]
-    def flushState                   = ().pure[F]
-    def delete(persist: Boolean)     = ().pure[F]
+    def appendEvent(event: E)                    = ().pure[F]
+    def replaceState(state: S)                   = ().pure[F]
+    def initPersistedState(state: S)             = ().pure[F]
+    def flushKeys                                = ().pure[F]
+    def flushState                               = ().pure[F]
+    def delete(persist: Boolean, offset: Offset) = ().pure[F]
   }
 
   def apply[F[_]: Monad, S, E](
@@ -160,8 +167,8 @@ object Buffers {
 
     def initPersistedState(state: S) = snapshots.initPersisted(state)
 
-    def delete(persist: Boolean) =
-      snapshots.delete(persist) *> journals.delete(persist) *> keys.delete(persist)
+    def delete(persist: Boolean, offset: Offset) =
+      snapshots.delete(persist, offset) *> journals.delete(persist) *> keys.delete(persist)
 
     def flushKeys =
       keys.flush
@@ -190,6 +197,50 @@ object ReadState {
 
   }
 
+  /** Restores state from previously saved events, folded onto the fenced snapshot store's view of the key.
+    *
+    * When the buffer fences, `snapshots.read` runs first and seeds the buffer cell from the snapshot store: a live
+    * snapshot becomes the recovery BASE the journal folds onto, a deletion tombstone's offset the replay-window floor
+    * (without it a replayed event below the tombstone would be re-derived and persisted, which a compare-and-set store
+    * rejects as stale though the owner is legitimate -- the deleted-key self-fence). Journal events whose offset (via
+    * `offsetOf`) is at or below the store's floor (`snapshots.floor`) are NOT folded: the journal is unfenced -- a
+    * zombie's replayed appends can land after a delete cleared it, and a journal TTL can reap rows the snapshot already
+    * carries -- so a below-floor row is either already reflected in the base or stale residue of a deleted key, and
+    * folding it would durably resurrect pre-delete state or regress a live key (the design doc's journal revive).
+    * Filtering events by offset at the seam, rather than comparing fold results after the fact, is what makes the guard
+    * hold at EVERY recovery: legitimate appends advance the journal past the store's offset, but the residue stays
+    * below the floor forever. An unfenced buffer has no trustworthy store to compare against, so the read is skipped
+    * (no wasted per-key round-trip) and the fold runs from scratch over the whole journal, exactly as before the fence
+    * existed. See docs/cassandra-single-writer-design.md.
+    */
+  def apply[F[_]: Monad: Log, S, E](
+    journals: JournalReader[F, E],
+    fold: FoldOption[F, S, E],
+    snapshots: Snapshots[F, S],
+    offsetOf: E => Offset,
+  ): ReadState[F, S] = new ReadState[F, S] {
+    def read =
+      if (snapshots.fenced)
+        for {
+          base  <- snapshots.read
+          floor <- snapshots.floor
+          events = floor.fold(journals.read)(floor => journals.read.filter(event => offsetOf(event) > floor))
+          state <- events
+            .foldLeftM(base) { (state, event) =>
+              Log[F].info(s"Restoring: $event") *> fold(state, event)
+            }
+            .last
+            .map {
+              // `last` is None exactly when no event survived the filter: the store's view IS the state (an
+              // all-reaped or residue-only journal must not erase the base). A folded Some(none) is different -
+              // the fold itself cleared the state - and stands.
+              case Some(folded) => folded
+              case None         => base
+            }
+        } yield state
+      else ReadState(journals, fold).read
+  }
+
   /** Restores state using previously saved snapshot */
   def apply[F[_], S](
     snapshots: SnapshotReader[F, S]

core/src/main/scala/com/evolutiongaming/kafka/flow/persistence/PersistenceModule.scala

@@ -7,7 +7,7 @@ import com.evolutiongaming.catshelper.LogOf
 import com.evolutiongaming.kafka.flow.KafkaKey
 import com.evolutiongaming.kafka.flow.journal.JournalDatabase
 import com.evolutiongaming.kafka.flow.key.KeyDatabase
-import com.evolutiongaming.kafka.flow.snapshot.{KafkaSnapshot, SnapshotDatabase}
+import com.evolutiongaming.kafka.flow.snapshot.{KafkaSnapshot, SnapshotDatabase, SnapshotsOf}
 import com.evolutiongaming.skafka.consumer.ConsumerRecord
 import scodec.bits.ByteVector
 
@@ -18,35 +18,51 @@ trait PersistenceModule[F[_], S] {
   def journals: JournalDatabase[F, KafkaKey, ConsumerRecord[String, ByteVector]]
   def snapshots: SnapshotDatabase[F, KafkaKey, KafkaSnapshot[S]]
 
+  /** How `snapshots` wires into the per-key buffer. Fenced on `KafkaSnapshot.offset` by default, which is right for a
+    * store that gates writes on the offset (see `CassandraSnapshots` compare-and-set mode). A module over a
+    * last-write-wins store should override to the unfenced `SnapshotsOf.backedBy(snapshots)`: such a store never
+    * returns a tombstone floor, so the fenced buffer would only change buffering semantics and make events-recovery pay
+    * a per-key floor read it can never profit from.
+    */
+  def snapshotsOf(
+    implicit F: Sync[F],
+    logOf: LogOf[F]
+  ): F[SnapshotsOf[F, KafkaKey, KafkaSnapshot[S]]] = snapshots.snapshotsOf
+
   /** Saves both events and snapshots, restores state from events */
   def restoreEvents(
     implicit F: Sync[F],
     logOf: LogOf[F]
   ): Resource[F, PersistenceOf[F, KafkaKey, KafkaSnapshot[S], ConsumerRecord[String, ByteVector]]] = for {
-    keysOf        <- Resource.eval(keys.toKeysOf)
-    journalsOf    <- Resource.eval(journals.journalsOf)
-    snapshotsOf   <- Resource.eval(snapshots.snapshotsOf)
-    persistenceOf <- PersistenceOf.restoreEvents(keysOf, journalsOf, snapshotsOf)
+    keysOf      <- Resource.eval(keys.toKeysOf)
+    journalsOf  <- Resource.eval(journals.journalsOf)
+    snapshotsOf <- Resource.eval(snapshotsOf)
+    persistenceOf <- PersistenceOf.restoreEvents(
+      keysOf,
+      journalsOf,
+      snapshotsOf,
+      (record: ConsumerRecord[String, ByteVector]) => record.offset
+    )
   } yield persistenceOf
 
   /** Saves both events and snapshots, restores state from snapshots */
   def restoreSnapshots(
     implicit F: Sync[F],
     logOf: LogOf[F]
   ): F[SnapshotPersistenceOf[F, KafkaKey, KafkaSnapshot[S], ConsumerRecord[String, ByteVector]]] = for {
-    keysOf      <- keys.toKeysOf
-    journalsOf  <- journals.journalsOf
-    snapshotsOf <- snapshots.snapshotsOf
-  } yield PersistenceOf.restoreSnapshots(keysOf, journalsOf, snapshotsOf)
+    keysOf       <- keys.toKeysOf
+    journalsOf   <- journals.journalsOf
+    _snapshotsOf <- snapshotsOf
+  } yield PersistenceOf.restoreSnapshots(keysOf, journalsOf, _snapshotsOf)
 
   /** Saves snapshots only, restores state from snapshots */
   def snapshotsOnly(
     implicit F: Sync[F],
     logOf: LogOf[F]
   ): F[SnapshotPersistenceOf[F, KafkaKey, KafkaSnapshot[S], ConsumerRecord[String, ByteVector]]] = for {
-    keysOf      <- keys.toKeysOf
-    snapshotsOf <- snapshots.snapshotsOf
-  } yield PersistenceOf.snapshotsOnly(keysOf, snapshotsOf)
+    keysOf       <- keys.toKeysOf
+    _snapshotsOf <- snapshotsOf
+  } yield PersistenceOf.snapshotsOnly(keysOf, _snapshotsOf)
 
 }
 

core/src/main/scala/com/evolutiongaming/kafka/flow/persistence/PersistenceOf.scala

@@ -11,6 +11,7 @@ import com.evolutiongaming.kafka.flow.journal.JournalsOf
 import com.evolutiongaming.kafka.flow.key.KeysOf
 import com.evolutiongaming.kafka.flow.snapshot.SnapshotsOf
 import com.evolutiongaming.kafka.flow.timer.Timestamps
+import com.evolutiongaming.skafka.Offset
 
 /** Creates generic persistence for a key */
 trait PersistenceOf[F[_], K, S, A] {
@@ -55,11 +56,20 @@ trait SnapshotPersistenceOf[F[_], K, S, A] extends PersistenceOf[F, K, S, A] { s
 }
 object PersistenceOf {
 
-  /** Saves both events and snapshots, restores state from events */
+  /** Saves both events and snapshots, restores state from events.
+    *
+    * @param offsetOf
+    *   the offset an event was consumed at. When the snapshot store fences stale writers, recovery folds the journal
+    *   onto the store's view of the key: a live snapshot is the base, a deletion tombstone's offset the floor, and
+    *   journal events at or below the store's offset are skipped -- they are already reflected in the base, or stale
+    *   residue of a deleted key that would otherwise be folded back to life (the journal is unfenced). See `ReadState`
+    *   and docs/cassandra-single-writer-design.md (the journal revive). Unfenced stores ignore it.
+    */
   def restoreEvents[F[_]: Monad: LogOf, K, S, A](
     keysOf: KeysOf[F, K],
     journalsOf: JournalsOf[F, K, A],
-    snapshotsOf: SnapshotsOf[F, K, S]
+    snapshotsOf: SnapshotsOf[F, K, S],
+    offsetOf: A => Offset,
   ): Resource[F, PersistenceOf[F, K, S, A]] = {
     val log = LogOf[F].apply(PersistenceOf.getClass)
     Resource.eval(log) map { implicit log => (key, fold, timestamps) =>
@@ -69,7 +79,7 @@ object PersistenceOf {
         snapshots <- snapshotsOf(key)
         keys       = keysOf(key)
       } yield Persistence(
-        readState = ReadState(journals, fold),
+        readState = ReadState(journals, fold, snapshots, offsetOf),
         buffers   = Buffers(keys, journals, snapshots)
       )
     }