diff --git a/docs/cassandra-single-writer-design.md b/docs/cassandra-single-writer-design.md new file mode 100644 index 000000000..48adc812a --- /dev/null +++ b/docs/cassandra-single-writer-design.md @@ -0,0 +1,246 @@ +--- +id: cassandra-single-writer-design +title: "Cassandra single-writer design: persist only" +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 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). + +## Problem + +[kafka-flow#732](https://github.com/evolution-gaming/kafka-flow/issues/732): consumer-group ownership +of the input topic does not extend to the snapshot store. During a rebalance a previous owner that has +not yet observed the revocation keeps flushing snapshots alongside the new owner; the snapshots table +is last-write-wins, so a stale snapshot can overwrite a newer one, and the next recovery loads stale +state while resuming from the committed offset — the events in between are lost from the state. The +[Kafka design doc](kafka-single-writer-design.md) covers the failure in full, with a diagram. + +Kafka's fix does not transfer. There, the snapshot write is bound to the input-offset commit in one +broker transaction, and the consumer generation — authoritative for ownership — fences the commit. +Cassandra offers no transaction to bind a Kafka offset commit into, and holds no notion of who owns a +partition. What it does offer is a conditional write: a lightweight transaction (LWT), linearizable +per partition key. So the fence must be built from something the snapshot itself carries, and checked +per write, per key. + +## Design + +The stored offset is the per-key +[fencing token](https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html). Every +snapshot already carries the offset it was folded up to, and #732 corruption is stale *by offset* — a +lower-offset snapshot replacing a higher-offset one. So each persist asserts the stored offset is not +greater than the one being written, and the newest-by-offset snapshot wins, whoever writes it: + +```sql +UPDATE snapshots_v2 SET ..., offset = :offset WHERE IF offset <= :offset +``` + +A key's first persist finds no row, so the `UPDATE` does not apply and falls back to +`INSERT ... IF NOT EXISTS`, retried once through the gated `UPDATE` if it loses an insert race. This +compound of separate LWTs is the one non-atomic path, and it is safe by construction: both `UPDATE`s +are offset-gated and the `INSERT` only writes an absent cell, so no interleaving overwrites a newer +snapshot. Its worst case is an over-rejection, never corruption: a delete slipping between the +`INSERT` and its retry surfaces as a spurious conflict, cleared on the next flush. A rejected write +raises `SnapshotWriteConflict`. + +Three choices in the predicate carry most of the design: + +- **Per key, not per partition.** #732 corruption is per key and keys are independent, so per-key + monotonic durability is exactly the guarantee needed. Anything per-partition would require an + ownership authority Cassandra does not have (see *Rejected alternatives* for what importing one + would cost). +- **`<=`, not `<`.** The legitimate owner can write at an offset it has already stored — a + timer-driven re-flush of the buffered high-water snapshot — and a strict `<` would fence the owner + against itself. Admitting an equal offset is safe because a same-offset write cannot move the + recovery point, so it cannot drop committed events. It also means a stale writer holding *exactly* + the stored offset goes undetected — accepted for the same reason. +- **Ordered by data, not by identity.** The token orders writes by what they contain, not by who sent + them, so the fence needs no clock synchronisation, no lease and no liveness protocol — but it can + never say "you are not the owner", only "you are behind". Closing that gap needs a generation in + the token (see *Rejected alternatives* and *Forward-looking*). + +Two snapshots at the same offset have folded the same records, so admitting equal offsets is sound +only when folds are deterministic and replayable — already a precondition of recovery generally, but +this mode leans on it harder. + +### Why the store is the whole change + +The mode deliberately changes nothing outside `CassandraSnapshots`: no buffer change, no recovery +change, no new read. That is possible because core already guarantees the legitimate owner never +presents an offset below what it recovered: `SnapshotFold` drops replayed records at or below the +recovered snapshot's offset before they reach the fold, so nothing is re-derived — or re-persisted — +below a key's recovered high-water even while the partition replays from a lower committed offset. +The fence therefore only ever rejects genuinely stale writers, and liveness comes for free. + +That property is load-bearing, not incidental. The deferred design below is the story of what happens +when it stops holding: fencing deletes breaks it, and repairing it pulls the buffer and both recovery +paths into the change. + +## Scope: deletes are not fenced + +`delete` stays a plain last-write-wins row `DELETE`. During an ownership overlap a stale writer can +therefore still erase a newer snapshot, or resurrect a just-deleted key by writing at a lower +offset — #732 residual for exactly the keys a fold deletes concurrently with re-writes. + +Drawing the line here is a choice, not an oversight: + +- **A fenced deletion is still expressible.** Instead of folding to `None`, fold to an + offset-carrying *empty* state (`Some(empty)`): the deletion then travels the fenced persist path + and a lower-offset zombie is rejected exactly as for any persist. The store's row lives until its + TTL — the cost of the workaround. +- **Fencing `delete` itself costs out of all proportion to the predicate above** — a breaking API and + new recovery machinery, with a liveness trap on each recovery path. That is the deferred design + below. + +## The expired guard (TTL reconfiguration) + +Cassandra TTLs are per **cell** and fixed at write time, and a row stays visible while any live cell — +or the first write's `INSERT` row marker (immortal only when that first write ran without a `ttl` and +the table has no `default_time_to_live`) — survives. Under a uniform `ttl` from the first deployment +it is unreachable: the delete co-writes `offset` with `value`, so a visible row always carries a live +guard. It is a **reconfiguration / pre-TTL-legacy** hazard — enabling (or shortening) the `ttl` between writes of a key can leave a +visible row whose `offset` guard cell has expired while an earlier, longer-lived cell or marker +survives: e.g. a no-TTL deployment's first write (immortal marker), TTL'd re-persists, then idleness +past the TTL — every column null, nothing to fence. `get` already reads such a row as absent (its `value` cell is expired) +and a plain delete removes it entirely, but no regular persist could ever claim it: the null guard +fails `IF offset <= :offset`, the not-applied result looks like "row absent" (it is not — Cassandra +returns the condition column, null, exactly when the row exists), `INSERT ... IF NOT EXISTS` loses to +the still-visible row, and the retry conflicts — every write of the key raising +`SnapshotWriteConflict`, forever. So the write path distinguishes the guard-expired result and repairs +it with a Paxos-safe `IF offset = null` persist that reinstates the guard +(`Statements.prepareRepairPersist`; racing repairs serialize, the loser conflicts as usual). The +repair re-arms the guard but (being an `UPDATE`) cannot remove an immortal marker, so such a row +re-poisons after each `ttl` until an owner deletes/reaps it — palliative, not curative. `SnapshotTtlEdgeSpec` pins the state and the repair against real +Cassandra. Prefer configuring the `ttl` from the first deployment anyway. + +## Compatibility + +- **No public API change**, hence no major version: the mode is an opt-in flag + (`compareAndSet = true`, default `false`) on the existing entry points. +- **No schema migration in either direction**: the condition reads the `offset` column every version + already writes. Rolling deploys are safe; the mixed-mode clock-skew caveat is in + [Persistence](persistence.md). +- **The fence is write-side only.** Recovery reads at the regular (non-serial) consistency level, so + the guarantee reaches the reader only when `R + W > N` — with weaker levels a recovery read can + miss the newest committed snapshot and reintroduce #732 on the read side. Required levels and the + serial-consistency setting are in [Persistence](persistence.md). + +## Testing + +Store-level tests against real Cassandra (`SnapshotSpec`) cover monotonic persists, stale-write +rejection, equal-offset re-persist, TTL on both write paths, and concurrent first-writers racing a +fresh key (exercising the insert-retry compound). Flow-level tests (`FlowSpec`) run through the real +recovery and flush machinery: the #732 reproduction shows corruption under last-write-wins, and its +counterpart shows the stale flush rejected under compare-and-set. + +## The full solution, and why it was deferred + +Persist-only was carved out of a full design that also fenced deletes. That full design was +implemented and reviewed, and none of it ships. It is summarised here because its trajectory is +the main learning of this work: **a fence that looked like one more predicate turned into a chain of +core changes, each forced by the previous one.** + +1. **A fenced delete cannot remove the row.** The offset guard *is* the row: remove it and a lagging + zombie's `INSERT ... IF NOT EXISTS` at a lower offset succeeds, resurrecting a stale snapshot. So + a fenced delete must be a logical **tombstone** — the row kept, value nulled, offset gated on the + same predicate as a persist (`SET value = null ... IF offset <= :offset`). + +2. **The tombstone forces a breaking API.** The gate needs the delete's offset, but + `SnapshotWriteDatabase.delete(key)` does not carry one — fencing deletes means a source- and + binary-breaking `delete(key, offset)` through every delete path and every custom store. A major + version, for this alone. + +3. **The buffer must become monotonic, or the owner fences itself.** After recovery a key can hold a + durable snapshot at offset `X` while the partition replays from a lower committed offset `C` (a + slow *other* key held `C` back). A timer-driven delete in that window would be gated on the + processing offset `< X` and rejected — the fence crashing the *legitimate* owner. The fix keeps + the per-key buffer monotonic in offset and gates a delete on the key's high-water `X`, which the + true owner presents and a genuinely stale writer never reached. Note the persist case needs no + such fix (`SnapshotFold` filters replayed records); it is the timer-driven *delete*, which + bypasses that filter, that forces the buffer change. + +4. **Recovery must learn to see tombstones, or the owner livelocks.** The base read is `get`, typed + `Option[S]`: a value or nothing. A tombstone comes back as `None`, indistinguishable from a + never-written key — so recovery seeds no high-water, the buffer climbs from replayed offsets + `< X`, and the offset-`X` tombstone rejects the owner's own flush. The flow tears down, + re-recovers the same floorless tombstone, and repeats: + + ```mermaid + sequenceDiagram + participant O as Owner (recovering) + participant DB as snapshots table + Note over DB: tombstone: value = null, offset = X + O->>DB: recover: get(key) → None (offset X invisible) + O->>O: replay events from committed offset C < X + O->>DB: flush re-derived snapshot at offset < X + DB-->>O: IF offset <= X fails: SnapshotWriteConflict + Note over O: tear down, re-recover the same None,
repeat — livelock + ``` + + The fix is a new recovery read that surfaces `Deleted(offset)` where `get` says `None`, seeding + the buffer's floor. + +5. **Events-recovery re-opens the livelock on its own.** `restoreEvents` rebuilds state by folding + the *journal*, not by reading the snapshot store — and a delete clears the journal, so the fold + yields nothing and the floor is lost again, untouched by the fix above. Events-recovery must read + the snapshot store purely for the tombstone floor before folding. A delete fence that seeds the + floor only on snapshot recovery still livelocks here; the two seedings are independent. + +**What the analysis showed.** Safety was never at risk: a rejected write changes nothing and the +durable offset never regresses, in every configuration. Every defect was a **liveness** failure — the +self-fence and the livelocks above — and each fix proved necessary: with the monotonic buffer or the +tombstone floor removed, liveness fails while safety still holds, reached through a live snapshot in +one configuration and through a deleted key in another. With both in place, a single writer's durable +offsets stay monotonic for both safety and liveness. + +**The deferral decision.** On the benefit side, fencing `delete` protects only folds that return +`None` — a need already coverable by the empty-state persist above. On the cost side: a major version +for every custom snapshot store; invasive changes to the buffer and both recovery paths — the hottest +code in core; and a liveness surface subtle enough to be a maintenance risk in its own right, not +just implementation effort. Timing sharpened the call: +[KIP-939](https://cwiki.apache.org/confluence/display/KAFKA/KIP-939:+Support+Participation+in+2PC) +(participation in 2PC) may eventually let a Cassandra snapshot write bind to a generation-fenced +Kafka offset commit, giving per-partition ownership that supersedes a per-key delete fence — making +the deferred machinery potentially throwaway. So the full design was parked where it can be resumed, +and the persist-only subset shipped. + +The learning generalises: in a last-write-wins store, fencing *writes* is cheap — the token rides the +data. It is *deletion* that is expensive, because deletion removes the very cell that carries the +fence, and every fix for that (tombstones, floors, monotonic buffers) leaks upward into recovery. + +## Rejected alternatives + +- **Offset-as-write-timestamp (LWW register)**: write each snapshot `USING TIMESTAMP ` 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. +- **Composite `(offset, generation)` token**: gate on the consumer generation as well as the offset, + closing the equal-offset gap and giving per-partition ownership. Rejected as the default because it + couples the self-contained Cassandra module to the live consumer generation; reasonable as a future + strict mode. +- **Recovery-side reconciliation** (store the offset, reconcile on read): does not prevent the stale + overwrite — last-write-wins still corrupts the store — so strictly weaker than fencing the write. + +## Forward-looking + +- **Offset-gated deletes** — the deferred design above; a candidate for a future major + version if the empty-state workaround proves insufficient in practice. +- **Per-partition ownership** — a composite `(offset, generation)` strict mode, or, further out, + [KIP-939](https://cwiki.apache.org/confluence/display/KAFKA/KIP-939:+Support+Participation+in+2PC): + an externally-coordinated two-phase commit binding the Cassandra snapshot write to a + generation-fenced Kafka offset commit — per-partition ownership without per-key CAS. Not actionable + now; see the Kafka design doc's forward-looking note. diff --git a/docs/kafka-single-writer-design.md b/docs/kafka-single-writer-design.md index 317978489..e7778fb79 100644 --- a/docs/kafka-single-writer-design.md +++ b/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 diff --git a/docs/persistence.md b/docs/persistence.md index f22d6e566..27c0417dd 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -37,25 +37,31 @@ 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. @@ -63,6 +69,73 @@ You do not catch the rejection yourself; it is handled for you: 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 diff --git a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala index 0910b017f..7d88d512d 100644 --- a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/FlowSpec.scala +++ b/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) } diff --git a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotSpec.scala b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotSpec.scala index c85dbeb97..83b5e9bbe 100644 --- a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotSpec.scala +++ b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotSpec.scala @@ -32,6 +32,136 @@ class SnapshotSpec extends CassandraSpec { test.unsafeRunSync() } + test("compare-and-set: writes with monotonically increasing offsets are applied") { + val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "cas-monotonic") + val test: IO[Unit] = for { + snapshots <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + compareAndSet = true + ) + // first write of a key goes through the `INSERT ... IF NOT EXISTS` path + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(5), value = "state-5")) + five <- snapshots.get(key) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "state-10")) + ten <- snapshots.get(key) + // a snapshot can be replaced at the same offset, e.g. when state was changed by a timer + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "state-10-updated")) + tenUp <- snapshots.get(key) + } yield { + assertEquals(clue(five.map(_.value)), Some("state-5")) + assertEquals(clue(ten.map(_.value)), Some("state-10")) + assertEquals(clue(tenUp.map(_.value)), Some("state-10-updated")) + } + + test.unsafeRunSync() + } + + test("compare-and-set: stale write is rejected") { + val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "cas-stale") + val test: IO[Unit] = for { + snapshots <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + compareAndSet = true + ) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "state-10")) + result <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(7), value = "state-7-stale")).attempt + stored <- snapshots.get(key) + } yield { + result match { + case Left(conflict: CassandraSnapshots.SnapshotWriteConflict) => + assertEquals(clue(conflict.key), key) + assertEquals(clue(conflict.attemptedOffset), Offset.unsafe(7)) + assertEquals(clue(conflict.persistedOffset), Some(Offset.unsafe(10))) + case other => fail(s"expected SnapshotWriteConflict, got $other") + } + assertEquals(clue(stored.map(_.value)), Some("state-10")) + } + + test.unsafeRunSync() + } + + test("compare-and-set: a delete is unguarded, so a lower-offset write resurrects a deleted key (persist-only gap)") { + // Pins the accepted persist-only residual gap (see docs/cassandra-single-writer-design.md, "Scope: + // deletes are not fenced"): a delete is a plain last-write-wins DELETE, so it removes the row and its offset + // guard. A lagging zombie's lower-offset write then finds an absent row, takes the + // `INSERT ... IF NOT EXISTS` first-write path, and resurrects the key below the deleted offset. This + // is the one residual #732 case the persist fence does not cover; if a future mode offset-gates + // deletes, this assertion should flip (the resurrection becomes a SnapshotWriteConflict). + val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "cas-delete-resurrect") + val test: IO[Unit] = for { + snapshots <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + compareAndSet = true + ) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "state-10")) + _ <- snapshots.delete(key) + deleted <- snapshots.get(key) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(7), value = "state-7-stale")) + resurrected <- snapshots.get(key) + } yield { + assert(clue(deleted.isEmpty)) + // the unguarded delete let the lower offset (7) win, resurrecting the key below the delete + assertEquals(clue(resurrected), Some(KafkaSnapshot(offset = Offset.unsafe(7), value = "state-7-stale"))) + } + + test.unsafeRunSync() + } + + test("compare-and-set: concurrent first-writers race on a fresh key; the highest offset wins, no corruption") { + // Exercises persistCompareAndSet's first-write compound under real contention: every writer hits + // UPDATE-absent then `INSERT ... IF NOT EXISTS` for the same new key; one INSERT wins and the losers + // take the retry-`UPDATE` path (the single-threaded tests never reach it). The offset guard keeps it + // safe -- the durable snapshot ends at the highest offset, never clobbered by a lower one, and any + // rejected writer fails cleanly with SnapshotWriteConflict. + val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "cas-first-write-race") + val offsets = (1 to 8).toList + val test: IO[Unit] = for { + snapshots <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + compareAndSet = true + ) + results <- offsets.parTraverse(o => + snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(o.toLong), value = s"state-$o")).attempt + ) + stored <- snapshots.get(key) + } yield { + assertEquals(clue(stored.map(_.value)), Some(s"state-${offsets.max}")) // highest wins, no stale overwrite + results.collect { case Left(e) => e }.foreach { + case _: CassandraSnapshots.SnapshotWriteConflict => () + case other => fail(s"unexpected failure (not a conflict): $other") + } + } + + test.unsafeRunSync() + } + + test("compare-and-set: ttl is set on both insert and update paths") { + val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "cas-ttl") + val test: IO[Unit] = for { + snapshots <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + ttl = 1.hour.some, + compareAndSet = true, + ) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(5), value = "state-5")) + insertTtls <- getTtls(key) + _ <- snapshots.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "state-10")) + updateTtls <- getTtls(key) + } yield { + assertEquals(clue(insertTtls.size), 1) + assert(clue(insertTtls.head.isDefined)) + assertEquals(clue(updateTtls.size), 1) + assert(clue(updateTtls.head.isDefined)) + } + + test.unsafeRunSync() + } + test("failures") { val key = KafkaKey("SnapshotSpec", "integration-tests-1", TopicPartition.empty, "queries") val test: IO[Unit] = for { diff --git a/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotTtlEdgeSpec.scala b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotTtlEdgeSpec.scala new file mode 100644 index 000000000..5d83be776 --- /dev/null +++ b/persistence-cassandra-it-tests/src/test/scala/com/evolutiongaming/kafka/flow/snapshot/SnapshotTtlEdgeSpec.scala @@ -0,0 +1,106 @@ +package com.evolutiongaming.kafka.flow.snapshot + +import cats.effect.IO +import cats.syntax.all.* +import com.evolutiongaming.kafka.flow.cassandra.CassandraCodecs.* +import com.evolutiongaming.kafka.flow.{CassandraSpec, KafkaKey} +import com.evolutiongaming.scassandra.syntax.* +import com.evolutiongaming.skafka.{Offset, TopicPartition} +import scodec.bits.ByteVector + +import scala.concurrent.duration.* +import scala.jdk.CollectionConverters.* + +/** The guard-expired ("poison") row: a TTL reconfiguration artefact, and its repair. + * + * Cassandra TTLs are per cell, and a row stays visible while any live cell - or the first write's `INSERT` row marker, + * which nothing can remove - survives. A no-TTL deployment's first write of a key leaves that immortal marker; enable + * the `ttl` and let the key's TTL'd cells expire, and the row stays visible with every column null: `offset = null`, + * so nothing fences it. `get` already reads it as absent (the `value` cell is expired), but without a repair path no + * persist could ever claim it again: the null guard fails `IF offset <= :offset`, the not-applied result was mistaken + * for "row absent", `INSERT ... IF NOT EXISTS` lost to the still-visible row, and the retry conflicted - every write + * of the key raising `SnapshotWriteConflict`, forever (established empirically against real Cassandra). + * + * The fix: the not-applied result distinguishes "row present, guard null" (Cassandra returns the condition column, + * null, exactly when the row exists) from "row absent", and a persist claims the guard-expired row through the + * Paxos-safe `IF offset = null` repair write, which also reinstates the guard. + */ +class SnapshotTtlEdgeSpec extends CassandraSpec { + + private val table = "snapshots_ttl_edge" + + test("a guard-expired row reads as absent and is repaired by the next persist") { + val key = KafkaKey("SnapshotTtlEdgeSpec", "integration-tests-1", TopicPartition.empty, "poison") + val test: IO[Unit] = for { + // two deployments over one table: before and after `ttl` was enabled + noTtl <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + tableName = table, + compareAndSet = true, + ) + withTtl <- CassandraSnapshots.withSchema[IO, String]( + cassandra().session, + cassandra().sync, + tableName = table, + ttl = 1.second.some, + compareAndSet = true, + ) + // 1. the no-TTL deployment's first write: an INSERT, so the row marker is immortal + _ <- noTtl.persist(key, KafkaSnapshot(offset = Offset.unsafe(10), value = "v10")) + // 2. the TTL-enabled deployment re-persists: all four cells rewritten with TTL 1s (the marker keeps none) + _ <- withTtl.persist(key, KafkaSnapshot(offset = Offset.unsafe(11), value = "v11")) + // 3. the cells expire; the immortal marker keeps the row visible with every column null + _ <- IO.sleep(3.seconds) + row <- selectRaw(key) + // 4. the guard is gone, so the row fences nothing - and reads as absent + readPoison <- noTtl.get(key) + // 5. a legitimate re-creation claims the row through the `IF offset = null` repair write + _ <- noTtl.persist(key, KafkaSnapshot(offset = Offset.unsafe(12), value = "v12")) + afterRepair <- noTtl.get(key) + // 6. the repair reinstated the guard: a stale write is fenced again + staleAfterRepair <- noTtl.persist(key, KafkaSnapshot(offset = Offset.unsafe(5), value = "v5-stale")).attempt + } yield { + row match { + case Some((value, offset)) => + assertEquals(clue(value), None: Option[ByteVector]) // expired + assertEquals(clue(offset), None: Option[Offset]) // expired: the guard is gone + case None => + fail("expected the poison row to stay visible via the immortal row marker") + } + assertEquals(clue(readPoison), None: Option[KafkaSnapshot[String]]) + assertEquals(clue(afterRepair.map(_.value)), Some("v12")) + staleAfterRepair match { + case Left(conflict: CassandraSnapshots.SnapshotWriteConflict) => + assertEquals(clue(conflict.persistedOffset), Some(Offset.unsafe(12))) + case other => fail(s"expected SnapshotWriteConflict after the repair reinstated the guard, got $other") + } + } + + test.unsafeRunSync() + } + + private def selectRaw(key: KafkaKey): IO[Option[(Option[ByteVector], Option[Offset])]] = { + val session = cassandra().session + for { + prepared <- session.prepare( + s"""SELECT value, offset FROM $table WHERE + | application_id = :application_id + | AND group_id = :group_id + | AND topic = :topic + | AND partition = :partition + | AND key = :key""".stripMargin + ) + bound = prepared + .bind() + .encode("application_id", key.applicationId) + .encode("group_id", key.groupId) + .encode("topic", key.topicPartition.topic) + .encode("partition", key.topicPartition.partition.value) + .encode("key", key.key) + rows <- session.execute(bound) + } yield rows.all().asScala.headOption.map { row => + (row.decode[Option[ByteVector]]("value"), row.decode[Option[Offset]]("offset")) + } + } +} diff --git a/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala b/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala index 196e6c1d9..2adf9a1a0 100644 --- a/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala +++ b/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/cassandra/CassandraPersistence.scala @@ -21,6 +21,10 @@ object CassandraPersistence { * - for keys see [[com.evolutiongaming.kafka.flow.key.CassandraKeys.DefaultTableName]] * - for snapshots see [[com.evolutiongaming.kafka.flow.snapshot.CassandraSnapshots.DefaultTableName]] * - for journals see [[com.evolutiongaming.kafka.flow.journal.CassandraJournals.DefaultTableName]] + * + * @param snapshotCompareAndSet + * enables conditional snapshot writes protecting from stale writers overwriting newer snapshots during partition + * ownership transitions, see [[com.evolutiongaming.kafka.flow.snapshot.CassandraSnapshots.withSchema]] */ def withSchemaF[F[_]: Async, S]( session: scassandra.CassandraSession[F], @@ -28,6 +32,7 @@ object CassandraPersistence { consistencyOverrides: ConsistencyOverrides, keysSegments: KeySegments, recordExpiration: RecordExpiration = RecordExpiration.default, + snapshotCompareAndSet: Boolean = false, )(implicit fromBytes: skafka.FromBytes[F, S], toBytes: skafka.ToBytes[F, S]): F[PersistenceModule[F, S]] = for { _keys <- CassandraKeys.withSchema(session, sync, consistencyOverrides, keysSegments, ttl = recordExpiration.keys) _journals <- CassandraJournals.withSchema(session, sync, consistencyOverrides, ttl = recordExpiration.journals) @@ -35,7 +40,8 @@ object CassandraPersistence { session, sync, consistencyOverrides, - ttl = recordExpiration.snapshots + ttl = recordExpiration.snapshots, + compareAndSet = snapshotCompareAndSet, ) } yield new CassandraPersistence[F, S] { def keys = _keys @@ -50,7 +56,7 @@ object CassandraPersistence { * * This method uses the same `JsonCodec[Try]` as `JournalParser` does to simplify defining the basic application. if * \@consistencyConfig is present then applies ConsistencyConfig.Read for all read queries and - * ConsistencyConfig.Write for all the mutations + * ConsistencyConfig.Write for all the mutations. For `snapshotCompareAndSet` see `withSchemaF`. */ def withSchema[F[_]: Async, S]( session: scassandra.CassandraSession[F], @@ -58,11 +64,12 @@ object CassandraPersistence { consistencyOverrides: ConsistencyOverrides, keysSegments: KeySegments, recordExpiration: RecordExpiration = RecordExpiration.default, + snapshotCompareAndSet: Boolean = false, )(implicit fromBytes: skafka.FromBytes[Try, S], toBytes: skafka.ToBytes[Try, S]): F[PersistenceModule[F, S]] = { val fromTry = FunctionK.liftFunction[Try, F](MonadThrow[F].fromTry) implicit val _fromBytes: FromBytes[F, S] = fromBytes mapK fromTry implicit val _toBytes: ToBytes[F, S] = toBytes mapK fromTry - withSchemaF(session, sync, consistencyOverrides, keysSegments, recordExpiration) + withSchemaF(session, sync, consistencyOverrides, keysSegments, recordExpiration, snapshotCompareAndSet) } // This exists for the sake of binary compatibility, to be removed in next major version diff --git a/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/snapshot/CassandraSnapshots.scala b/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/snapshot/CassandraSnapshots.scala index 2ac0c4b5a..98f55a701 100644 --- a/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/snapshot/CassandraSnapshots.scala +++ b/persistence-cassandra/src/main/scala/com/evolutiongaming/kafka/flow/snapshot/CassandraSnapshots.scala @@ -13,38 +13,125 @@ import com.evolutiongaming.kafka.flow.cassandra.StatementHelper.StatementOps import com.evolutiongaming.scassandra.CassandraSession import com.evolutiongaming.scassandra.StreamingCassandraSession.* import com.evolutiongaming.scassandra.syntax.* -import com.evolutiongaming.skafka.{FromBytes, Offset, ToBytes} +import com.evolutiongaming.skafka.{Bytes, FromBytes, Offset, ToBytes} import scodec.bits.ByteVector import CassandraSnapshots.* +import java.time.Instant import scala.concurrent.duration.FiniteDuration +import scala.util.control.NoStackTrace +/** Cassandra-backed implementation of `SnapshotDatabase`. + * + * In `WriteMode.CompareAndSet` mode a snapshot is persisted only if the stored snapshot's offset is not greater than + * the offset of the new snapshot. See [[CassandraSnapshots.withSchema]] for details. + */ class CassandraSnapshots[F[_]: Async, T]( session: CassandraSession[F], getStatement: PreparedStatement, persistStatement: PreparedStatement, deleteStatement: PreparedStatement, consistencyOverrides: ConsistencyOverrides = ConsistencyOverrides.none, + writeMode: WriteMode = WriteMode.LastWriteWins, )(implicit fromBytes: FromBytes[F, T], toBytes: ToBytes[F, T]) extends SnapshotDatabase[F, KafkaKey, KafkaSnapshot[T]] { def persist(key: KafkaKey, snapshot: KafkaSnapshot[T]): F[Unit] = + writeMode match { + case WriteMode.LastWriteWins => persistUnconditional(key, snapshot) + case WriteMode.CompareAndSet(insert, repair) => persistCompareAndSet(insert, repair, key, snapshot) + } + + private def persistUnconditional(key: KafkaKey, snapshot: KafkaSnapshot[T]): F[Unit] = for { boundStatement <- Statements.bindPersist(persistStatement, key, snapshot) statement = boundStatement.withConsistencyLevel(consistencyOverrides.write) _ <- session.execute(statement).void } yield () + /** Persists the snapshot only if the stored one is not newer. + * + * The conditional update is not applied when the stored row has a higher offset (a concurrent writer persisted a + * newer snapshot) or when the row does not exist yet. The latter is retried as `INSERT ... IF NOT EXISTS`; if that + * loses to a concurrent insert, the conditional update is retried once, so the writer with the newest snapshot wins + * a first-write race. A row whose `offset` guard cell has expired (a TTL reconfiguration artefact, see + * `Statements.prepareRepairPersist`) is repaired through an `IF offset = null` write - without it the key would + * conflict forever: the null guard fails the conditional update, and `INSERT ... IF NOT EXISTS` loses to the + * still-visible row. + */ + private def persistCompareAndSet( + insertStatement: PreparedStatement, + repairStatement: PreparedStatement, + key: KafkaKey, + snapshot: KafkaSnapshot[T], + ): F[Unit] = + for { + created <- Clock[F].instant + value <- toBytes.apply(snapshot.value, key.topicPartition.topic) + bind = (statement: PreparedStatement) => Statements.bindPersist(statement, key, snapshot, created, value) + // claims a guard-expired row (null `offset`): applies only while the guard is still gone, so a raced + // repair serializes via Paxos - the loser resolves to a plain conflict and the flow recovers as usual + repair = executeWrite(bind(repairStatement)).flatMap { repairRow => + val conflict = SnapshotWriteConflict(key, snapshot.offset, none).raiseError[F, Unit] + resolveConditional(key, repairRow, snapshot.offset)(onAbsent = conflict, onGuardExpired = conflict) + } + updateRow <- executeWrite(bind(persistStatement)) + _ <- resolveConditional(key, updateRow, snapshot.offset)( + // the row does not exist yet: first write for the key + onAbsent = executeWrite(bind(insertStatement)).flatMap { insertRow => + if (insertRow.getBool("[applied]")) ().pure[F] + else + // lost the insert race to a concurrent writer: retry the conditional update once + executeWrite(bind(persistStatement)).flatMap { retryRow => + // a row deleted between the insert and the retry surfaces as a (spurious) conflict; the flow + // recovers from it on the next flush + resolveConditional(key, retryRow, snapshot.offset)( + onAbsent = SnapshotWriteConflict(key, snapshot.offset, none).raiseError[F, Unit], + onGuardExpired = repair, + ) + } + }, + onGuardExpired = repair, + ) + } yield () + + private def executeWrite(boundStatement: BoundStatement): F[Row] = + session.execute(boundStatement.withConsistencyLevel(consistencyOverrides.write)).map(_.one()) + + /** Interprets a conditional-write (lightweight transaction) result: unit if it applied, [[SnapshotWriteConflict]] if + * a newer stored offset rejected it, `onGuardExpired` when the row exists but its `offset` guard cell is null + * (Cassandra returns the condition column - with a null value - exactly when the row exists; an absent row's result + * carries no such column), or `onAbsent` when the row is absent - the first-write path inserts there, and its retry + * treats a still-absent row as a (spurious) conflict. + */ + private def resolveConditional( + key: KafkaKey, + row: Row, + attemptedOffset: Offset, + )(onAbsent: => F[Unit], onGuardExpired: => F[Unit]): F[Unit] = + if (row.getBool("[applied]")) ().pure[F] + else if (row.getColumnDefinitions.contains("offset")) + row.decode[Option[Offset]]("offset") match { + // the stored snapshot is newer: this writer is stale + case Some(persistedOffset) => + SnapshotWriteConflict(key, attemptedOffset, persistedOffset.some).raiseError[F, Unit] + // the row exists but its offset cell expired (a TTL reconfiguration artefact): the guard is gone + case None => onGuardExpired + } + else onAbsent + def get(key: KafkaKey): F[Option[KafkaSnapshot[T]]] = { val boundStatement = Statements.bindGet(getStatement, key).withConsistencyLevel(consistencyOverrides.read) for { row <- session.executeStream(boundStatement).first - snapshot <- row.map(row => decode(row)).sequence + snapshot <- row.flatTraverse(row => decode(row)) } yield snapshot } + // persist-only mode: a delete is an ordinary last-write-wins DELETE (the offset guard protects persists, not + // deletes). Gating deletes on an offset is out of scope here (it would need a delete(key, offset) signature). def delete(key: KafkaKey): F[Unit] = { val boundStatement = Statements.bindDelete(deleteStatement, key).withConsistencyLevel(consistencyOverrides.write) session.execute(boundStatement).void @@ -56,6 +143,32 @@ object CassandraSnapshots { val DefaultTableName = "snapshots_v2" + /** How a snapshot write is performed. `WriteMode.CompareAndSet` carries the extra statements used for the first write + * of a key and for repairing an expired guard, so they exist exactly when the database is in compare-and-set of a + * key, so the insert statement exists exactly when the database is in compare-and-set mode. + */ + sealed trait WriteMode + object WriteMode { + case object LastWriteWins extends WriteMode + final case class CompareAndSet(insertStatement: PreparedStatement, repairStatement: PreparedStatement) + extends WriteMode + } + + /** Raised in compare-and-set mode (see [[CassandraSnapshots.withSchema]]) when the store already contains a newer + * snapshot for the key - another writer (likely the new partition owner after a rebalance) persisted in parallel, so + * this writer is stale. `persistedOffset` is the stored offset, if it could be determined. + */ + final case class SnapshotWriteConflict( + key: KafkaKey, + attemptedOffset: Offset, + persistedOffset: Option[Offset], + ) extends RuntimeException( + s"snapshot write conflict for key $key: attempted to write with offset $attemptedOffset " + + s"while the store contains offset ${persistedOffset.fold("unknown")(_.toString)}, " + + "another writer is likely owning the key now" + ) + with NoStackTrace + /** Create table for storing snapshots. If table already exists it will not be recreated. * * @param session @@ -68,6 +181,11 @@ object CassandraSnapshots { * name of the table to create. The default value is "snapshots_v2" * @param ttl * optional TTL to set on inserted records + * @param compareAndSet + * if `true`, each snapshot *persist* is a Cassandra lightweight transaction asserting the stored offset is not + * greater than the new one, protecting from stale writers; a rejected write fails with [[SnapshotWriteConflict]]. + * Deletes remain ordinary last-write-wins (gating deletes on an offset is out of scope for this mode). See the + * persistence docs' "Protecting against stale snapshot writes" for limitations and costs. Default `false`. * @param fromBytes * deserializer function to convert array of bytes to the snapshot type T * @param toBytes @@ -79,11 +197,19 @@ object CassandraSnapshots { consistencyOverrides: ConsistencyOverrides = ConsistencyOverrides.none, tableName: String = DefaultTableName, ttl: Option[FiniteDuration] = None, + compareAndSet: Boolean = false, )( implicit fromBytes: FromBytes[F, T], toBytes: ToBytes[F, T] ): F[SnapshotDatabase[F, KafkaKey, KafkaSnapshot[T]]] = - withCustomSchema(SnapshotSchema.of(session, sync, tableName), session, consistencyOverrides, tableName, ttl) + withCustomSchema( + SnapshotSchema.of(session, sync, tableName), + session, + consistencyOverrides, + tableName, + ttl, + compareAndSet + ) /** Create table with a user defined schema for storing snapshots. If table already exists it will not be recreated. * Note that the table schema must be compatible with predefined queries for storing and retrieving snapshots data. @@ -98,6 +224,8 @@ object CassandraSnapshots { * name of the table to create. The default value is "snapshots_v2" * @param ttl * optional TTL to set on inserted records + * @param compareAndSet + * enables conditional writes protecting from stale writers, see [[CassandraSnapshots.withSchema]] * @param fromBytes * deserializer function to convert array of bytes to the snapshot type T * @param toBytes @@ -109,6 +237,7 @@ object CassandraSnapshots { consistencyOverrides: ConsistencyOverrides = ConsistencyOverrides.none, tableName: String = DefaultTableName, ttl: Option[FiniteDuration] = None, + compareAndSet: Boolean = false, )( implicit fromBytes: FromBytes[F, T], toBytes: ToBytes[F, T] @@ -116,14 +245,22 @@ object CassandraSnapshots { for { _ <- snapshotSchema.create getStatement <- Statements.prepareGet(session, tableName) - persistStatement <- Statements.preparePersist(session, tableName, ttl) + persistStatement <- Statements.preparePersist(session, tableName, ttl, compareAndSet) deleteStatement <- Statements.prepareDelete(session, tableName) + writeMode <- + if (compareAndSet) + for { + insert <- Statements.prepareInsertIfNotExists(session, tableName, ttl) + repair <- Statements.prepareRepairPersist(session, tableName, ttl) + } yield WriteMode.CompareAndSet(insert, repair): WriteMode + else (WriteMode.LastWriteWins: WriteMode).pure[F] } yield new CassandraSnapshots( session = session, getStatement = getStatement, persistStatement = persistStatement, deleteStatement = deleteStatement, consistencyOverrides = consistencyOverrides, + writeMode = writeMode, ) def truncate[F[_]: Monad]( @@ -133,16 +270,21 @@ object CassandraSnapshots { ): F[Unit] = SnapshotSchema.of(session, sync, tableName).truncate // we cannot use DecodeRow here because Code[T].decode is effectful - protected def decode[F[_]: Monad, T](row: Row)(implicit fromBytes: FromBytes[F, T]): F[KafkaSnapshot[T]] = { - val value = row.decode[ByteVector]("value") - fromBytes.apply(value.toArray, "").map { value => - KafkaSnapshot[T]( - offset = row.decode[Offset]("offset"), - metadata = row.decode[String]("metadata"), - value = value - ) + // a visible row can carry a null `value`: its cell expired while older cells or the first write's row marker keep + // the row alive (the guard-expired row, see Statements.prepareRepairPersist). Nothing distinguishes it from a + // reaped key, so it reads as absent - decoding it non-optionally would instead fail every recovery of the key. + protected def decode[F[_]: Monad, T](row: Row)(implicit fromBytes: FromBytes[F, T]): F[Option[KafkaSnapshot[T]]] = + row.decode[Option[ByteVector]]("value") match { + case None => none[KafkaSnapshot[T]].pure[F] + case Some(value) => + fromBytes.apply(value.toArray, "").map { value => + KafkaSnapshot[T]( + offset = row.decode[Offset]("offset"), + metadata = row.decode[String]("metadata"), + value = value + ).some + } } - } protected object Statements { @@ -150,23 +292,82 @@ object CassandraSnapshots { session: CassandraSession[F], tableName: String, ttl: Option[FiniteDuration], + compareAndSet: Boolean = false, + ): F[PreparedStatement] = + session.prepare(persistCql(tableName, ttl, condition = if (compareAndSet) "IF offset <= :offset" else "")) + + /** Used in compare-and-set mode to repair a row whose `offset` guard cell has expired while other cells (or the + * first write's `INSERT` row marker, immortal only when that first write ran without a `ttl` and the table has no + * `default_time_to_live`) keep the row visible. Cassandra TTLs are per cell, so the state arises when the `ttl` is + * enabled or shortened between writes of a key - e.g. a no-TTL deployment's first write (immortal row marker) + * followed by TTL'd persists that have since expired. Such a row fences nothing, and the regular conditional write + * can never claim it (the null guard fails `IF offset <= :offset`, and `INSERT ... IF NOT EXISTS` loses to the + * still-visible row - without this statement the key would conflict forever). `IF offset = null` claims exactly + * that state through Paxos: racing writers serialize, the loser sees a live offset and conflicts. The repair + * re-arms the guard but (being an `UPDATE`) cannot remove an immortal marker, so such a row re-poisons after each + * `ttl` until an owner deletes/reaps it - palliative, not curative. (`get` already reads such a row as absent - + * its `value` cell is expired - and a plain delete removes it entirely, so the read and delete paths need no + * counterpart.) + */ + def prepareRepairPersist[F[_]]( + session: CassandraSession[F], + tableName: String, + ttl: Option[FiniteDuration], + ): F[PreparedStatement] = + session.prepare(persistCql(tableName, ttl, condition = "IF offset = null")) + + private def persistCql(tableName: String, ttl: Option[FiniteDuration], condition: String): String = + s""" + |UPDATE + | $tableName + | ${StatementHelper.ttlFragment(ttl)} + |SET + | created = :created, + | metadata = :metadata, + | value = :value, + | offset = :offset + |WHERE + | application_id = :application_id + | AND group_id = :group_id + | AND topic = :topic + | AND partition = :partition + | AND key = :key + | $condition + """.stripMargin + + /** Used in compare-and-set mode for the first write of a key, when the conditional update of `preparePersist` + * cannot be applied because the row does not exist yet. + */ + def prepareInsertIfNotExists[F[_]]( + session: CassandraSession[F], + tableName: String, + ttl: Option[FiniteDuration], ): F[PreparedStatement] = session.prepare( s""" - |UPDATE - | $tableName - | ${StatementHelper.ttlFragment(ttl)} - |SET - | created = :created, - | metadata = :metadata, - | value = :value, - | offset = :offset - |WHERE - | application_id = :application_id - | AND group_id = :group_id - | AND topic = :topic - | AND partition = :partition - | AND key = :key + |INSERT INTO $tableName ( + | application_id, + | group_id, + | topic, + | partition, + | key, + | created, + | metadata, + | value, + | offset + |) VALUES ( + | :application_id, + | :group_id, + | :topic, + | :partition, + | :key, + | :created, + | :metadata, + | :value, + | :offset + |) + |IF NOT EXISTS + |${StatementHelper.ttlFragment(ttl)} """.stripMargin ) @@ -178,19 +379,26 @@ object CassandraSnapshots { for { created <- Clock[F].instant value <- toBytes.apply(snapshot.value, key.topicPartition.topic) - } yield { - statement - .bind() - .encode("application_id", key.applicationId) - .encode("group_id", key.groupId) - .encode("topic", key.topicPartition.topic) - .encode("partition", key.topicPartition.partition) - .encode("key", key.key) - .encode("offset", snapshot.offset) - .encode("created", created) - .encode("metadata", snapshot.metadata) - .encode("value", value) - } + } yield bindPersist(statement, key, snapshot, created, value) + + def bindPersist[T]( + statement: PreparedStatement, + key: KafkaKey, + snapshot: KafkaSnapshot[T], + created: Instant, + value: Bytes, + ): BoundStatement = + statement + .bind() + .encode("application_id", key.applicationId) + .encode("group_id", key.groupId) + .encode("topic", key.topicPartition.topic) + .encode("partition", key.topicPartition.partition) + .encode("key", key.key) + .encode("offset", snapshot.offset) + .encode("created", created) + .encode("metadata", snapshot.metadata) + .encode("value", value) def prepareGet[F[_]](session: CassandraSession[F], tableName: String): F[PreparedStatement] = session diff --git a/website/sidebars.js b/website/sidebars.js index 5be2a5e1b..367364a01 100644 --- a/website/sidebars.js +++ b/website/sidebars.js @@ -13,7 +13,7 @@ const sidebars = { type: "category", label: "Design notes", collapsed: false, - items: ["kafka-single-writer-design"], + items: ["kafka-single-writer-design", "cassandra-single-writer-design"], }, ], };