CheckpointStore SPI

CheckpointStore is the SPI behind the @Checkpoint annotation. It persists the two checkpoint anchors (lastProcessedToken — the last terminally settled event, lastSeenToken — the last server-certified position) 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
An event settles terminally and a policy threshold is reached (`saveEveryN` / `saveIntervalSeconds`), or the handler calls `ctx.saveCheckpointNow()`	`saveProcessed(streamName, token, timestamp, heartbeatTimestamp)` — the anchor write carries its heartbeat confirmation.
The idle heartbeat certifies a fresh position (`idleHeartbeatIntervalSeconds`)	`saveSeen(streamName, token, timestamp, heartbeatTimestamp)` — position and confirmation written together.
The probe re-certifies an unchanged position	`saveHeartbeat(streamName, heartbeatTimestamp)` — only the confirmation timestamp moves.
A history-lost recovery re-establishes a durable position	`resetAfterHistoryLost(streamName, freshSeenToken, expectedDeadProcessed, timestamp)` — installs the fresh position and removes the expired processed pair only while it still equals the value read at detection.
Stream starts	`findByStreamName(streamName)` — feeds the resume cascade.
Administrative / test cleanup	`delete(streamName)` — the runtime itself never deletes a checkpoint; the operation exists for operators and tests.
Bulk / migration paths	`save(checkpoint)` — full-record upsert keyed by `streamName`.

The dual-anchor contract (why the two anchors have different natures and writers) is documented in the @Checkpoint reference.

Interface

public interface CheckpointStore {

    void                     save(Checkpoint checkpoint);
    Optional<Checkpoint>     findByStreamName(String streamName);
    void                     delete(String streamName);

    /** Targeted writes — each advances only its own pair of fields. */
    default void saveSeen(String streamName, BsonDocument token, Instant timestamp) { ... }
    default void saveProcessed(String streamName, BsonDocument token, Instant timestamp) { ... }

    /** Anchor + heartbeat confirmation together — override with ONE atomic write. */
    default void saveSeen(String streamName, BsonDocument token, Instant timestamp,
                          Instant heartbeatTimestamp) { ... }
    default void saveProcessed(String streamName, BsonDocument token, Instant timestamp,
                               Instant heartbeatTimestamp) { ... }

    /** Confirmation only — the position was re-certified unchanged. */
    default void saveHeartbeat(String streamName, Instant heartbeatTimestamp) { ... }

    /** History-lost recovery — the dead-processed guard MUST be evaluated
        atomically with the removal (conditional write). */
    default void resetAfterHistoryLost(String streamName, BsonDocument freshSeenToken,
                                       BsonDocument expectedDeadProcessed, Instant timestamp) { ... }

    /** Always-no-op implementation. */
    static CheckpointStore noOp();
}

Only save, findByStreamName and delete are abstract — every other operation has a working default built on them, so a minimal backend is three methods.

Semantic contract

save must upsert the record keyed by checkpoint.streamName().
Partial operations must preserve the fields they do not target, atomically with respect to concurrent writers. The two anchors have independent writers on different threads: the processed-anchor policy writes lastProcessedToken while the heartbeat probe writes lastSeenToken. A targeted update (partial $set, HSET, UPDATE … SET <pair>) satisfies this by construction. The default implementations are read-modify-write fallbacks kept for source compatibility: under concurrent anchor writers they can resurrect a stale value of the other anchor, and a Java-level lock cannot fix this for a store shared between instances. Production implementations must override them with targeted writes — the shipped Mongo and Redis stores do.
The 4-argument overloads exist so the anchor and its lastHeartbeatTimestamp confirmation land in one atomic write. Their defaults perform two writes (anchor, then heartbeat) — correct but observable in an intermediate state.
resetAfterHistoryLost carries the at-least-once guard of the recovery: the expired processed pair is removed only while lastProcessedToken still equals the value read at history-loss detection. If a fresh anchor was re-acquired concurrently, it must be preserved. The race-free guarantee requires evaluating the guard atomically with the removal (a conditional write) — the Mongo stores put it in the update filter, the Redis stores use a Lua script.
findByStreamName must return Optional.empty() when no record exists for that stream.
delete is a no-op when no record exists.

`Checkpoint` record

public record Checkpoint(
        String              streamName,
        String              instanceId,              // nullable
        BsonDocument        lastSeenToken,           // nullable — server-certified position
        Instant             lastSeenTimestamp,       // nullable
        BsonDocument        lastProcessedToken,      // nullable — last terminally settled event
        Instant             lastProcessedTimestamp,  // nullable — that event's cluster time
        Instant             lastHeartbeatTimestamp,  // nullable — last confirmed recoverable position
        Map<String, Object> metadata
) {}

All token / timestamp fields are nullable — a brand-new stream has neither anchor yet, and an idle stream that has only ever certified through the heartbeat 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. All partial operations use a native `upsert` with a partial `$set` (single atomic write, no hot-path read); `resetAfterHistoryLost` puts the dead-processed guard in the update filter.
`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 the partial operations with targeted HSET operations and resetAfterHistoryLost with a Lua script. 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 the partial operations

For any production backend, override every partial operation with a targeted write that touches only the relevant fields. Two traps the naive version falls into:

Every partial operation must upsert. On a brand-new stream, the bootstrap’s saveSeen is the first write — a plain UPDATE matches zero rows, silently persists nothing, and the core then runs believing its position is durable. Use INSERT … ON CONFLICT DO UPDATE (or MERGE).
resetAfterHistoryLost must evaluate its guard atomically with the removal — a conditional write, not a read-then-write.

public class JdbcCheckpointStore implements CheckpointStore {

    private final JdbcTemplate jdbc;

    // ... save / findByStreamName / delete implementations ...

    @Override
    public void saveProcessed(String streamName, BsonDocument token, Instant timestamp,
                              Instant heartbeatTimestamp) {
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_processed_token, " +
            "  last_processed_ts, last_heartbeat_ts) VALUES (?, ?, ?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET last_processed_token = EXCLUDED.last_processed_token, " +
            "  last_processed_ts = EXCLUDED.last_processed_ts, last_heartbeat_ts = EXCLUDED.last_heartbeat_ts",
            streamName, token.toJson(),
            Timestamp.from(timestamp), Timestamp.from(heartbeatTimestamp)
        );
    }

    @Override
    public void saveSeen(String streamName, BsonDocument token, Instant timestamp,
                         Instant heartbeatTimestamp) {
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_seen_token, " +
            "  last_seen_ts, last_heartbeat_ts) VALUES (?, ?, ?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET last_seen_token = EXCLUDED.last_seen_token, " +
            "  last_seen_ts = EXCLUDED.last_seen_ts, last_heartbeat_ts = EXCLUDED.last_heartbeat_ts",
            streamName, token.toJson(),
            Timestamp.from(timestamp), Timestamp.from(heartbeatTimestamp)
        );
    }

    @Override
    public void saveHeartbeat(String streamName, Instant heartbeatTimestamp) {
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_heartbeat_ts) VALUES (?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET last_heartbeat_ts = EXCLUDED.last_heartbeat_ts",
            streamName, Timestamp.from(heartbeatTimestamp)
        );
    }

    @Override
    public void saveProcessed(String streamName, BsonDocument token, Instant timestamp) {
        // Same upsert, without touching last_heartbeat_ts.
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_processed_token, last_processed_ts) " +
            "VALUES (?, ?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET last_processed_token = EXCLUDED.last_processed_token, " +
            "  last_processed_ts = EXCLUDED.last_processed_ts",
            streamName, token.toJson(), Timestamp.from(timestamp)
        );
    }

    @Override
    public void saveSeen(String streamName, BsonDocument token, Instant timestamp) {
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_seen_token, last_seen_ts) " +
            "VALUES (?, ?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET last_seen_token = EXCLUDED.last_seen_token, " +
            "  last_seen_ts = EXCLUDED.last_seen_ts",
            streamName, token.toJson(), Timestamp.from(timestamp)
        );
    }

    @Override
    public void resetAfterHistoryLost(String streamName, BsonDocument freshSeenToken,
                                      BsonDocument expectedDeadProcessed, Instant timestamp) {
        // Two contract points live INSIDE this one statement:
        // - the dead-processed guard: the processed pair is cleared only
        //   while it still equals the value read at history-loss detection —
        //   a concurrently re-acquired anchor survives;
        // - the null-token branch: without a fresh certified position, the
        //   seen pair AND the heartbeat are cleared (a heartbeat without a
        //   recoverable position would be a lie).
        boolean hasFresh = freshSeenToken != null;
        jdbc.update(
            "INSERT INTO fw_checkpoints (stream_name, last_seen_token, last_seen_ts, " +
            "  last_heartbeat_ts) VALUES (?, ?, ?, ?) " +
            "ON CONFLICT (stream_name) DO UPDATE SET " +
            "  last_seen_token = EXCLUDED.last_seen_token, " +
            "  last_seen_ts = EXCLUDED.last_seen_ts, " +
            "  last_heartbeat_ts = EXCLUDED.last_heartbeat_ts, " +
            "  last_processed_token = CASE WHEN fw_checkpoints.last_processed_token " +
            "    IS NOT DISTINCT FROM ? THEN NULL ELSE fw_checkpoints.last_processed_token END, " +
            "  last_processed_ts = CASE WHEN fw_checkpoints.last_processed_token " +
            "    IS NOT DISTINCT FROM ? THEN NULL ELSE fw_checkpoints.last_processed_ts END",
            streamName,
            hasFresh ? freshSeenToken.toJson() : null,
            hasFresh ? Timestamp.from(timestamp) : null,
            hasFresh ? Timestamp.from(timestamp) : null,
            expectedDeadProcessed != null ? expectedDeadProcessed.toJson() : null,
            expectedDeadProcessed != null ? expectedDeadProcessed.toJson() : null
        );
    }
}

The two anchors write different fields of the same row, so their independent writers never overwrite each other. This is how the built-in Mongo and Redis implementations work natively.

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 field-isolation invariant of every partial operation (a seen write never touches the processed pair and vice versa), the anchor+heartbeat overloads, and the resetAfterHistoryLost guard semantics — 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

Registration API

Runtime

SPI

When the SPI is exercised

Interface

Semantic contract

`Checkpoint` record

Provided implementations

No-op default

Providing a custom implementation

Overriding the partial operations

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 the partial operations

​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 the partial operations

Testing the contract

See Also