> ## Documentation Index > Fetch the complete documentation index at: https://docs.flowwarden.io/llms.txt > Use this file to discover all available pages before exploring further. # CheckpointStore SPI > Resume-token persistence backing @Checkpoint — contract, dual-token semantics, and how to plug in your own backend `CheckpointStore` is the SPI behind the [`@Checkpoint`](/reference/checkpoint) annotation. It persists the two resume tokens (`lastSeenToken`, `lastProcessedToken`) that drive at-least-once delivery and stream resume on restart. The MongoDB-backed implementation is auto-configured; the [Redis backend](/redis) ships an alternative; any custom backend (JDBC, Cassandra, in-memory, …) plugs in by registering a `CheckpointStore` Spring bean. ## When the SPI is exercised | Path | Calls into the SPI | | --------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | | Handler returns successfully | `saveProcessed(streamName, token, timestamp)` — only `lastProcessedToken` advances. | | Heartbeat timer ticks (`saveIntervalSeconds`) | `saveSeen(streamName, token, timestamp)` — only `lastSeenToken` advances, even when the event was rejected by `@Filter` or no event arrived. | | Stream starts | `findByStreamName(streamName)` — feeds the resume cascade. | | Stream is deleted from a `@ChangeStream` definition | `delete(streamName)`. | | Bulk / migration paths | `save(checkpoint)` — full-record upsert keyed by `streamName`. | The dual-token contract (why one token advances independently of the other) is documented in the [@Checkpoint reference](/reference/checkpoint#how-it-works). ## Interface ```java theme={null} public interface CheckpointStore { void save(Checkpoint checkpoint); Optional findByStreamName(String streamName); /** Targeted write — advances only lastSeenToken + lastSeenTimestamp. */ default void saveSeen(String streamName, BsonDocument token, Instant timestamp) { // default impl: read-modify-write via findByStreamName + save } /** Targeted write — advances only lastProcessedToken + lastProcessedTimestamp. */ default void saveProcessed(String streamName, BsonDocument token, Instant timestamp) { // default impl: read-modify-write via findByStreamName + save } void delete(String streamName); /** Always-no-op implementation. */ static CheckpointStore noOp(); } ``` ### Semantic contract * **`save`** must upsert the record keyed by `checkpoint.streamName()`. Concurrent `save` calls for the same stream may interleave; the framework guarantees that for a given stream at most one writer is active at a time, so implementations don't need to serialize internally. * **`saveSeen`** must update **only** the `lastSeenToken` and `lastSeenTimestamp` fields. `lastProcessedToken` / `lastProcessedTimestamp` must remain untouched — they advance on a different schedule (handler success) and overwriting them would break at-least-once delivery. * **`saveProcessed`** must update **only** the `lastProcessedToken` and `lastProcessedTimestamp` fields. Same invariant in mirror. * **`findByStreamName`** must return `Optional.empty()` when no record exists for that stream — never a half-constructed `Checkpoint` with nulls in unrelated fields. * **`delete`** is a no-op when no record exists. The default implementations of `saveSeen` and `saveProcessed` fall back to `findByStreamName + save`, which preserves the dual-token contract through a read-modify-write — at the cost of one extra read per call. Custom implementations should override both methods with a native targeted write (see [Overriding `saveSeen` / `saveProcessed`](#overriding-savedseen--saveprocessed) below). ## `Checkpoint` record ```java theme={null} public record Checkpoint( String streamName, String instanceId, // nullable BsonDocument lastSeenToken, // nullable Instant lastSeenTimestamp, // nullable BsonDocument lastProcessedToken, // nullable Instant lastProcessedTimestamp, // nullable Map metadata ) {} ``` All token / timestamp fields are nullable — a brand-new stream has neither token yet, and a stream that has only ever ticked through the heartbeat timer has `lastSeenToken` populated but no `lastProcessedToken`. The framework's resume cascade tolerates any nullability pattern. ## Provided implementations | Implementation | When active | Storage | | ------------------------------ | ---------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `MongoCheckpointStore` | Imperative mode (`flowwarden.default-mode=IMPERATIVE`) — default | The `_fw_checkpoints` collection. `saveSeen` / `saveProcessed` use a native MongoDB `upsert` with a partial `$set` that targets only the relevant pair of fields, avoiding a hot-path read. | | `ReactiveMongoCheckpointStore` | Reactive mode (`flowwarden.default-mode=REACTIVE`) | Same `_fw_checkpoints` collection, same document layout. Switching between modes is transparent on the persisted state. | For a Redis-backed deployment, the [`flowwarden-redis`](/redis) satellite ships `RedisCheckpointStore` (blocking, auto-configured) and `ReactiveRedisCheckpointStore` (reactive, opt-in). Both override `saveSeen` / `saveProcessed` with targeted `HSET` operations. See [Redis Configuration](/redis/configuration#storage-layout) for the storage layout. ## No-op default `CheckpointStore.noOp()` returns a shared singleton that silently ignores all calls: ```java theme={null} CheckpointStore store = CheckpointStore.noOp(); store.save(checkpoint); // discarded store.findByStreamName("anything"); // Optional.empty() ``` Suitable for tests that don't need resume semantics and for short-lived processes where checkpoint persistence is irrelevant. **Not** a viable production choice — every restart starts the stream from the latest MongoDB event, silently losing anything that arrived during downtime. ## Providing a custom implementation The auto-configured bean is guarded by `@ConditionalOnMissingBean`, so any `@Bean CheckpointStore` you declare wins: ```java theme={null} @Configuration public class CustomCheckpointConfig { @Bean CheckpointStore checkpointStore(/* your client */) { return new MyCustomCheckpointStore(/* ... */); } } ``` ### Overriding `saveSeen` / `saveProcessed` The default implementations work correctly for any backend, but each call costs an extra read. For write-heavy stores (high-throughput streams with low `saveEveryN` and short `saveIntervalSeconds`), override both methods with a targeted write that touches only the relevant pair of fields: ```java theme={null} public class JdbcCheckpointStore implements CheckpointStore { private final JdbcTemplate jdbc; // ... save / findByStreamName / delete implementations ... @Override public void saveSeen(String streamName, BsonDocument token, Instant timestamp) { jdbc.update( "UPDATE fw_checkpoints SET last_seen_token = ?, last_seen_ts = ? WHERE stream_name = ?", token.toJson(), Timestamp.from(timestamp), streamName ); } @Override public void saveProcessed(String streamName, BsonDocument token, Instant timestamp) { jdbc.update( "UPDATE fw_checkpoints SET last_processed_token = ?, last_processed_ts = ? WHERE stream_name = ?", token.toJson(), Timestamp.from(timestamp), streamName ); } } ``` The two methods write **different fields of the same row**, so they never race or overwrite each other — the heartbeat timer advances only `last_seen_token` while handler success advances only `last_processed_token`. This is how the built-in Mongo and Redis implementations work natively. Overriding only one of the two methods is supported but unusual. The framework relies on both being kept consistent with `save` semantics (same record, same dual-token contract) — if you override `saveSeen`, you almost certainly want to override `saveProcessed` too. ### Testing the contract If you write a custom `CheckpointStore`, run it against the contract tests published in `flowwarden-stream-core-testkit`: ```xml theme={null} io.flowwarden flowwarden-stream-core-testkit test ``` Extend `CheckpointStoreContractTest` and implement the abstract factory method. The contract covers upsert / find / delete semantics, the dual-token invariant (`saveSeen` doesn't touch `lastProcessedToken` and vice versa), and the `findByStreamName + save` fallback for default implementations — the same suite the built-in Mongo and Redis stores are validated against. ## See Also The user-facing annotation that drives this SPI. Drop-in Redis implementations of `CheckpointStore` and `LockService`. The sister SPI that backs `DeploymentMode.SINGLE_LEADER`. Failed-event persistence for the `@DeadLetterQueue` annotation.