# Checkpoint fencing

> Stop a node that lost its lease from overwriting a checkpoint.

[Layer 0: CloudEvents](/docs/layer-0/) · Blocking · Added in 0.33.0

A node that has lost its competing-consumer lock can still try to write a checkpoint after another node has already taken over and written a later one. Without a fence, that write moves the checkpoint backward, and the new holder redelivers events it already processed. Since 0.33.0, `CompetingConsumerStrategy` can hand a subscription model a fencing token, a number that only goes up as the lock changes hands, and the model refuses a checkpoint write whose token is older than the one already stored.

To fence, pass `strategy::fencingToken` as an extra constructor argument to whichever model writes the checkpoints, `DurableSubscriptionModel` in the example below:

* Java

  ```java
  CheckpointStorage checkpointStorage = new NativeMongoCheckpointStorage(mongoDatabase, "position-storage");
  NativeMongoLeaseCompetingConsumerStrategy competingConsumerStrategy = NativeMongoLeaseCompetingConsumerStrategy.withDefaults(mongoDatabase);
  SubscriptionModel wrappedSubscriptionModel = new DurableSubscriptionModel(
          new NativeMongoSubscriptionModel(mongoDatabase, "events", TimeRepresentation.DATE),
          checkpointStorage,
          competingConsumerStrategy::fencingToken);


  CompetingConsumerSubscriptionModel competingConsumerSubscriptionModel = new CompetingConsumerSubscriptionModel(wrappedSubscriptionModel, competingConsumerStrategy);
  ```

* Kotlin

  ```kotlin
  val checkpointStorage = NativeMongoCheckpointStorage(mongoDatabase, "position-storage")
  val competingConsumerStrategy = NativeMongoLeaseCompetingConsumerStrategy.withDefaults(mongoDatabase)
  val wrappedSubscriptionModel = DurableSubscriptionModel(
      NativeMongoSubscriptionModel(mongoDatabase, "events", TimeRepresentation.DATE),
      checkpointStorage,
      competingConsumerStrategy::fencingToken
  )


  val competingConsumerSubscriptionModel = CompetingConsumerSubscriptionModel(wrappedSubscriptionModel, competingConsumerStrategy)
  ```

`StreamCatchupSubscriptionModel`, `DcbCatchupSubscriptionModel` and `CatchupThenPushSubscriptionModel` take the same extra argument. `CompetingConsumerStrategy` does not implement `CheckpointWriteVersionSource` itself, the method reference is what connects the two, so a lock never has to know what a checkpoint is and a checkpoint store never has to know what a lease is.

The model calls that function right before every checkpoint write. When a number comes back, the write goes out stamped `notOlderThan(number)`. When the answer is empty, or when you passed no function at all, the write goes out unconditional. It is the same three-argument `save` either way, so the model has one path rather than a choice between two.

There are two things it deliberately leaves alone. It never asks your `CheckpointStorage` whether it can evaluate a condition at all, since that check runs at startup in the Spring Boot starter and nothing runs it when you wire the models yourself. And it neither catches nor retries a refusal, so the exception comes straight back out of the event handler, and finding out that this node lost its lock is left to the strategy’s own lease refresh.

A refused checkpoint write does not stop the application. `save` throws `CheckpointWriteConditionNotFulfilledException` and the position of the event that triggered the write is never recorded. Neither Mongo subscription model retries it, since a lease that has already moved on will never succeed on a later attempt. Within one lease period the strategy’s own lease refresh finds that this node no longer holds the lock, and pauses the consumer here. The node that took the lock over reads the checkpoint the refused write never got to update, and gets every event from there again. That is the at-least-once delivery every subscription model in this library already gives you. The fence keeps a node that lost its lease from moving the checkpoint backward, and that is all it does. Delivery is not exactly-once, so a handler still has to cope with running twice.

`CheckpointStorage.save` takes a `CheckpointWriteCondition` as its third argument, and a subscription model holding a fencing token stamps every write `notOlderThan(token)`. There are three conditions to choose from:

| Condition               | Accepted when                                                              | What it is for                                                                                                                                |
| :---------------------- | :------------------------------------------------------------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------- |
| `any()`                 | always                                                                     | The plain unconditional write, which is what the two-argument `save` has always meant. It leaves whatever version is stored exactly as it was |
| `notOlderThan(version)` | nothing is stored yet, or the stored version is not greater than `version` | The fence                                                                                                                                     |
| `ifAbsent()`            | no checkpoint is stored yet for that subscription id                       | Recording a subscription’s very first checkpoint, its start position, without another writer overwriting it through `any()`                   |

`CheckpointStorage.writeVersion(subscriptionId)` reads back the version a condition is judged against.

If you implement `CheckpointStorage` yourself, on the blocking interface or its reactor twin, your implementation stops compiling against 0.33.0, because `save` gains a third argument and the interface gains a `writeVersion` method. If you implement in Java, run the `org.occurrent.UpgradeToOccurrent_0_33` OpenRewrite recipe first. It adds the three-argument `save` and `writeVersion` to every Java implementation it finds missing them, each marked with a `TODO [Occurrent 0.33 upgrade]` comment, so the module compiles again without a manual pass. The recipe is Java only, so a Kotlin implementer adds the same two members by hand, against the interface’s own signatures. What it generates for a Java class with its own two-argument `save`, the shape every genuine 0.32.0 implementer has, and what a Kotlin implementer writes by hand, is exactly the snippet below. `save` delegates `any()` to that existing two-argument write and refuses anything stronger, and `writeVersion` answers empty. A class with no two-argument `save` of its own gets a `save` that refuses every write instead, so the generated code can never call back into itself. A store that only ever wrote unconditionally can leave the delegating shape exactly as generated:

```java
@Override
public Checkpoint save(String subscriptionId, Checkpoint checkpoint, CheckpointWriteCondition condition) {
    if (!(condition instanceof CheckpointWriteCondition.Any)) {
        throw new UnsupportedOperationException("This storage cannot evaluate " + condition + ", only any() is supported.");
    }
    return save(subscriptionId, checkpoint); // your existing unconditional write
}


@Override
public OptionalLong writeVersion(String subscriptionId) {
    return OptionalLong.empty();
}
```

Leave that exactly as generated. It is the right permanent answer for a store that genuinely cannot evaluate a condition, not something to come back and finish later. Such a store also leaves `evaluatesWriteConditions()` alone, because the default already answers `false` for it.

Occurrent’s own Mongo and Redis checkpoint storages do not need this recipe. They already evaluate every condition for real, Redis Cluster included, covered below. If your store can too, the conformance suite in `occurrent-tck-subscription-blocking` checks the two rules that matter for it. `any()` has to leave whatever version is stored untouched, rather than clearing it or working one out from the write. `notOlderThan(version)` has to accept when nothing is stored, since that is a checkpoint written before this condition existed, and every checkpoint saved by an earlier release has to stay readable.

The fence works on Redis Cluster too. `SpringRedisCheckpointStorage` keeps the checkpoint and its stored version in two differently named keys, and Cluster refuses a script that touches keys in different slots. The version key’s name includes a hash tag built from the checkpoint key’s own hash, so Cluster places both in the same slot and `notOlderThan` and `ifAbsent` work there exactly as they do on a standalone or replicated server. After that tag the version key’s name also includes a SHA-256 digest of the subscription id, so two ids that happen to share a hash tag, two tenant-scoped ids under the same `{tenant}` say, still get their own version key rather than silently sharing one fencing version.

One subscription id shape is still refused outright, an `IllegalArgumentException` from `save` for `notOlderThan` or `ifAbsent`. It is the shape Cluster itself falls back to hashing the whole id for (no brace pair, an unmatched brace, or an empty pair like `{}`), where that whole id is also either empty or has a closing brace somewhere in it, for example `""`, `"{}orders"` or `"a}b{c"`. `any()` and `delete` never refuse an id like this, since `any()` writes only the checkpoint key. In Cluster-safe mode a conditional write refuses an id of this shape before touching Redis, so that mode never leaves a version key behind for such an id. A version key for such an id can still be there from `forStandalone(RedisOperations)`, which has no such refusal and does write one, the case of a standalone deployment later migrated into Cluster with the same subscription ids. `delete`’s fallback removes that stray version key when there is one and does nothing when there isn’t. A Redis Cluster deployment with no `CompetingConsumerStrategy` wired in is unaffected either way.

Some storages answer per subscription id rather than once for the whole store. `evaluatesWriteConditionsFor(subscriptionId)` is the per-id version of `evaluatesWriteConditions()`, defaulting to whatever that answers, and it exists for a storage whose answer depends on the id it is asked about. `SpringRedisCheckpointStorage`’s Cluster-safe constructors above are exactly that case, answering `true` overall while refusing the one subscription id shape just described. `SpringRedisCheckpointStorage.forStandalone(RedisOperations)`, for a deployment that’s standalone or replicated rather than Cluster, accepts that shape too, and every other id outside the version key’s own reserved namespace, since a server that isn’t Cluster has no slots to align. Don’t build a Cluster deployment’s storage with it though, because a conditional write then fails with Redis’s own `CROSSSLOT` error for an id this constructor accepts and Cluster cannot align a slot for.

A storage of your own can also override `resolveFirstCheckpointRace(subscriptionId, candidate)`. `DurableSubscriptionModel` and `ManualStartSubscriptionModel` store a new subscription’s first position with an `ifAbsent()` write, and they call it when that write is refused because a position is already stored. The reactor `CheckpointStorage` has the same method returning a `Mono`, and `ReactorDurableSubscriptionModel` calls it the same way.

The default implementation returns an empty `Optional` (an empty `Mono` on the reactor side), and the model may then refuse the registration with `StartPositionAlreadyPinnedException`.

Override it only if your storage can compare the two positions and write the earlier one in one atomic step. Return whichever checkpoint is stored once that step is done. The MongoDB checkpoint storages do this while no delivery has moved the stored position yet.

Read this before rolling the upgrade out on a cluster already running competing consumers. During the rolling upgrade from 0.32.0 to 0.33.0, a node still on 0.32.0 releases its lock by deleting the lock document, so the next node to take it over starts at version 0 again. A 0.33.0 node’s checkpoint write then offers `notOlderThan(0)` against a checkpoint already stamped with a higher version and is refused. That refusal repeats, once per unit of the stored version, each cycle costing one lease period and one re-run of whatever the handler did before the refusal, until every node in the deployment runs 0.33.0. From then on the version keeps climbing instead of resetting, and the fence holds. If a subscription is stuck cycling and you want it to stop sooner, `CheckpointStorage.delete(subscriptionId)` clears the checkpoint and its stored version together, at the cost of replaying everything since. [ADR 116](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0116-a-checkpoint-write-from-a-lease-that-has-moved-on-is-refused.md) has the full design, and the [upgrade guide](https://github.com/johanhaleby/occurrent/blob/main/doc/migration/upgrading-to-0.33.0.md) covers implementing `CheckpointStorage` yourself in more detail.

> **Using Spring Boot?**
>
> The MongoDB starter fences every checkpoint for you, and checks at startup that your `CheckpointStorage` can be fenced. See [What the starter configures](/docs/spring-boot/what-the-starter-configures/#checkpoint-fencing).