CheckpointStore SPI

CheckpointStore is the SPI behind the @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 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.

Interface

public interface CheckpointStore {

    void                     save(Checkpoint checkpoint);
    Optional<Checkpoint>     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 below).

`Checkpoint` record

public record Checkpoint(
        String              streamName,
        String              instanceId,              // nullable
        BsonDocument        lastSeenToken,           // nullable
        Instant             lastSeenTimestamp,       // nullable
        BsonDocument        lastProcessedToken,      // nullable
        Instant             lastProcessedTimestamp,  // nullable
        Map<String, Object> 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 satellite ships RedisCheckpointStore (blocking, auto-configured) and ReactiveRedisCheckpointStore (reactive, opt-in). Both override saveSeen / saveProcessed with targeted HSET operations. See Redis Configuration for the storage layout.

No-op default

CheckpointStore.noOp() returns a shared singleton that silently ignores all calls:

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:

@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:

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:

<dependency>
    <groupId>io.flowwarden</groupId>
    <artifactId>flowwarden-stream-core-testkit</artifactId>
    <scope>test</scope>
</dependency>

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.

@Checkpoint annotation

The user-facing annotation that drives this SPI.

Redis backend

Drop-in Redis implementations of CheckpointStore and LockService.

LockService SPI

The sister SPI that backs DeploymentMode.SINGLE_LEADER.

DlqStore SPI

Failed-event persistence for the @DeadLetterQueue annotation.

Annotations

Runtime

SPI

When the SPI is exercised

Interface

Semantic contract

`Checkpoint` record

Provided implementations

No-op default

Providing a custom implementation

Overriding `saveSeen` / `saveProcessed`

Testing the contract

See Also

@Checkpoint annotation

Redis backend

LockService SPI

DlqStore SPI

​When the SPI is exercised

​Interface

​Semantic contract

​Checkpoint record

​Provided implementations

​No-op default

​Providing a custom implementation

​Overriding saveSeen / saveProcessed

​Testing the contract

​See Also

@Checkpoint annotation

Redis backend

LockService SPI

DlqStore SPI

When the SPI is exercised

Interface

Semantic contract

`Checkpoint` record

Provided implementations

No-op default

Providing a custom implementation

Overriding `saveSeen` / `saveProcessed`

Testing the contract

See Also