# Durable subscriptions

> Continue where a subscription left off after a restart.

[Layer 0: CloudEvents](/docs/layer-0/) · Blocking · Modules: `occurrent-subscription-durable-blocking`

Storing the checkpoint is useful if you need to resume a subscription from its last known checkpoint when restarting an application. Occurrent provides a utility that implements `SubscriptionModel` and combines a `CheckpointAwareSubscriptionModel` and a `CheckpointStorage` implementation (see [here](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage)) to automatically store the checkpoint, by default,\
after each processed event. If you don’t want the checkpoint to be persisted after *every* event, you can control how often this should happen by supplying a predicate to `DurableSubscriptionModelConfig`. There’s a pre-defined predicate, `org.occurrent.subscription.util.predicate.EveryN`, that allow\
the checkpoint to be stored for *every n* event instead of simply *every* event. There’s also a shortcut, e.g. `new DurableSubscriptionModelConfig(3)` that creates an instance of `EveryN` that stores the checkpoint for every third event.

When the filter of a subscription matches no events for a long time, its stored checkpoint stays at the last event it handled. The MongoDB oplog can drop that position while the subscription is still up to date. A resume after a pause, a lease handover between competing consumers, or a restart would then start from a position the oplog no longer has.

To prevent that, `DurableSubscriptionModel` saves a quiet position when a read of the change stream returns no event for a subscription. That is how far the wrapped model, the `CheckpointAwareSubscriptionModel` it combines with the storage, has read, saved as the subscription’s checkpoint. It saves at most once a minute. The minute starts over whenever a checkpoint is saved for an event, so a subscription that stores a checkpoint for an event at least once a minute gets no extra write.

Only a wrapped model that implements `QuietPositionReportingSubscriptions` reports that position. `NativeMongoSubscriptionModel` and `SpringMongoSubscriptionModel` do.

Change the interval with `saveQuietPositionEvery(Duration)`, and keep it well below how long the oplog keeps history:

```java
var config = new DurableSubscriptionModelConfig(1).saveQuietPositionEvery(Duration.ofSeconds(30));
```

`neverSaveQuietPosition()` turns the save off.

A persist predicate that declines some events, such as `EveryN` with `n` above 1, can delay the save. A subscription that stops receiving events right after an event the predicate declined gets no position saved until the predicate stores one. A subscription from a `StartAt` of your own gets none saved until the predicate has stored the position of an event.

If you want full control, it’s recommended to pick a [checkpoint storage](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage) implementation, and store the checkpoint yourself using its API.

To use it, first we need to add the dependency:

* Maven

  ```xml
  <dependency>
      <groupId>org.occurrent</groupId>
      <artifactId>occurrent-subscription-durable-blocking</artifactId>
      <version>0.34.0</version>
  </dependency>
  ```

* Gradle

  ```groovy
  compile 'org.occurrent:occurrent-subscription-durable-blocking:0.34.0'
  ```

* SBT

  ```scala
  libraryDependencies += "org.occurrent" % "occurrent-subscription-durable-blocking" % "0.34.0"
  ```

* Grape

  ```groovy
  @Grab(group='org.occurrent', module='occurrent-subscription-durable-blocking', version='0.34.0')
  ```

* Leiningen

  ```clojure
  [org.occurrent/occurrent-subscription-durable-blocking "0.34.0"]
  ```

* Buildr

  ```ruby
  'org.occurrent:occurrent-subscription-durable-blocking:jar:0.34.0'
  ```

* Ivy

  ```xml
  <dependency org="org.occurrent" name="occurrent-subscription-durable-blocking" rev="0.34.0" />
  ```

Then we should instantiate a `CheckpointAwareSubscriptionModel`, that subscribes to the events from the event store, and an instance of a `CheckpointStorage`, that stores the checkpoint, and combine them to a `DurableSubscriptionModel`:

* Java

  ```java
  // Create the non-durable blocking subscription instance
  CheckpointAwareSubscriptionModel nonDurableSubscriptionModel = ...
  // Create the storage
  CheckpointStorage storage = ...


  // Now combine the non-durable subscription model and the checkpoint storage
  SubscriptionModel durableSubscriptionModel = new DurableSubscriptionModel(nonDurableSubscriptionModel, storage);


  // Start a subscription
  durableSubscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent));
  ```

* Kotlin

  ```kotlin
  // Create the non-durable blocking subscription instance
  val nonDurableSubscriptionModel : CheckpointAwareSubscriptionModel = ...
  // Create the storage
  val storage : CheckpointStorage = ...


  // Now combine the non-durable subscription model and the checkpoint storage
  val durableSubscriptionModel : SubscriptionModel = DurableSubscriptionModel(nonDurableSubscriptionModel, storage)


  // Start a subscription
  durableSubscriptionModel.subscribe("mySubscriptionId") { cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent) };
  ```

## When No Start Position Can Be Recorded

A new subscription that asks for `StartAt.subscriptionModelDefault()` gets its start position stored before any event is delivered. `DurableSubscriptionModel` gets that position by calling `globalCheckpoint()` on the subscription model it wraps, inside `subscribe(..)`. If the application crashes before the first checkpoint is saved, the subscription resumes from that stored position.

When `globalCheckpoint()` returns `null` and nothing is stored for the subscription, `subscribe(..)` throws `IllegalStateException` and registers nothing. `NativeMongoSubscriptionModel` and `SpringMongoSubscriptionModel` return `null` when the server refuses the `hostInfo` command, which a shared MongoDB Atlas cluster does.

A subscription that already has a checkpoint stored is never refused this way, and neither is one that passes a `StartAt` of its own.

To start such a subscription anyway, set `startWhenNoStartPositionCanBeRecorded(true)` on its config:

```java
DurableSubscriptionModelConfig config = new DurableSubscriptionModelConfig(1).startWhenNoStartPositionCanBeRecorded(true);
DurableSubscriptionModel subscriptionModel = new DurableSubscriptionModel(wrappedSubscriptionModel, checkpointStorage, config);
```

The subscription then starts with no stored position. If the application crashes before the first checkpoint is saved, the subscription starts over from wherever the feed has got to by the time it restarts. An event whose delivery failed before the crash is then not delivered again.

> **Using Spring Boot?**
>
> Set `occurrent.subscription.start-when-no-start-position-can-be-recorded=true` instead, see [What the starter configures](/docs/spring-boot/what-the-starter-configures/#starting-when-no-start-position-can-be-recorded).