> ## 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.

# Transactions

> Use MongoDB transactions with Change Stream handlers for exactly-once processing semantics

By default, FlowWarden provides **at-least-once** delivery: if the application crashes between processing an event and saving the checkpoint, the event is replayed on restart. For most use cases (cache invalidation, notifications, projections), this is sufficient as long as your handler is idempotent.

For critical streams where duplicate processing is unacceptable (e.g., billing, payments, inventory), you can achieve **exactly-once** semantics by wrapping your business writes and checkpoint save in a single MongoDB transaction.

***

## Reading transaction metadata

When the **upstream writer** commits inside a MongoDB transaction, every change event in that transaction carries the session id (`lsid`) and transaction number (`txnNumber`). FlowWarden surfaces them through `ChangeStreamContext.getTransactionInfo()` as an `Optional<TransactionInfo>`, empty for standalone writes.

No extra configuration is required — just the replica set already needed by Change Streams.

```java theme={null}
@ChangeStream(documentType = Order.class)
public class OrderAuditHandler {

    @OnInsert
    void onInsert(Order order, ChangeStreamContext<Order> ctx) {
        ctx.getTransactionInfo().ifPresentOrElse(
            txn -> log.info("Order {} part of txn {}", order.getId(), txn.txnNumber()),
            ()  -> log.info("Order {} (standalone)", order.getId()));
    }
}
```

The reactive twin returns `Mono<Void>` and wraps the body in `Mono.fromRunnable`; the `ctx.getTransactionInfo()` call is identical.

### Common use cases

* **Audit / event grouping** — bucket downstream log entries by `txnNumber` so a single business operation surfaces as one logical record
* **Atomicity-preserving propagation** — collect all events of one upstream transaction before publishing them downstream as a unit (Kafka batch, message envelope, projection rebuild)
* **Correlation** — propagate `txnNumber` into traces / spans so cross-service correlation matches the source-of-truth transaction boundary

<Tip>
  The full record signature is `TransactionInfo(BsonDocument lsid, long txnNumber)`. See [`ChangeStreamContext` → TransactionInfo](/reference/change-stream-context#transactioninfo) for the exact contract.
</Tip>

A runnable demo with a generator that mixes transactional + standalone inserts lives in the [`flowwarden-examples` repository](https://github.com/flowwarden-io/flowwarden-examples) — see [`imperative/10-transactions`](https://github.com/flowwarden-io/flowwarden-examples/tree/main/imperative/10-transactions) and [`reactive/10-transactions`](https://github.com/flowwarden-io/flowwarden-examples/tree/main/reactive/10-transactions).

***

## Prerequisites for exactly-once

MongoDB transactions require:

* A **replica set** (already required for Change Streams)
* A `MongoTransactionManager` Spring bean in your application

```java theme={null}
@Configuration
public class MongoConfig {

    @Bean
    MongoTransactionManager transactionManager(MongoDatabaseFactory dbFactory) {
        return new MongoTransactionManager(dbFactory);
    }
}
```

***

## Exactly-once pattern

Use Spring's `@Transactional` on your handler method combined with `ctx.saveCheckpointNow()`:

```java theme={null}
@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
public class OrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
    }
}
```

**On success:** both the business write and the checkpoint are committed atomically.

**On exception:** Spring rolls back the transaction — neither the business write nor the checkpoint are persisted. By default, FlowWarden logs the error and moves on to the next event. The failed event is **lost** unless you configure retry and DLQ:

```java theme={null}
@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
@RetryPolicy(maxAttempts = 3)
@DeadLetterQueue
public class OrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
        // On failure: @RetryPolicy retries in a new transaction,
        // then @DeadLetterQueue captures the event if all retries fail
    }
}
```

<Tip>
  For critical streams using transactions, always combine `@RetryPolicy` and `@DeadLetterQueue` to avoid losing events on failure. See the [Retry & DLQ guide](/guides/retry-and-dlq) for configuration details.
</Tip>

### How it works

1. Spring's `@Transactional` opens a MongoDB session and binds it to the current thread
2. All `MongoTemplate` operations on the same thread automatically use the bound session
3. `ctx.saveCheckpointNow()` saves the checkpoint through `MongoTemplate` — it joins the same transaction
4. On success, Spring commits both the business write and the checkpoint atomically
5. On failure, Spring rolls back everything — the event will be redelivered on the next attempt

<Note>
  `ctx.saveCheckpointNow()` is required inside the handler. Without it, FlowWarden saves the checkpoint **after** the handler returns — outside the transaction boundary.
</Note>

***

## When to use transactions

<CardGroup cols={2}>
  <Card title="Use transactions" icon="check">
    * All handler writes go to MongoDB
    * Duplicate processing has real consequences (double charges, wrong inventory)
    * You need exactly-once guarantees
    * You use `SINGLE_LEADER` deployment mode
  </Card>

  <Card title="Skip transactions" icon="xmark">
    * Handler calls external APIs (HTTP, Kafka, email)
    * Handler is naturally idempotent
    * At-least-once is acceptable
    * High-throughput streams where transaction overhead matters
  </Card>
</CardGroup>

<Warning>
  MongoDB transactions cannot roll back external side-effects (HTTP calls, message queue publishes, emails). If your handler performs external calls, transactions only protect the MongoDB writes — the side-effects will not be undone on rollback. Design these handlers to be idempotent instead.
</Warning>

***

## Interaction with retry and DLQ

Transactions work naturally with `@RetryPolicy` and `@DeadLetterQueue`:

```java theme={null}
@ChangeStream(documentType = Order.class)
@Checkpoint(saveEveryN = 1)
@RetryPolicy(maxAttempts = 3)
@DeadLetterQueue(retentionDays = 90)
public class CriticalOrderProcessor {

    @Autowired
    private MongoTemplate mongoTemplate;

    @OnInsert
    @Transactional
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        mongoTemplate.save(new ProcessedOrder(order), "processed_orders");
        ctx.saveCheckpointNow();
    }
}
```

* Each retry attempt runs the handler in a **new transaction**
* If the handler throws, the transaction is rolled back (business writes + checkpoint)
* After all retries are exhausted, the event is sent to the DLQ **outside** any transaction

***

## Important constraints

### `mongoTemplateRef` breaks transaction atomicity

<Warning>
  If your `@ChangeStream` uses `mongoTemplateRef` to point to a different `MongoTemplate`, the exactly-once guarantee **does not apply**.
</Warning>

Spring binds the transaction session to a specific `MongoDatabaseFactory`. The checkpoint store uses the **default** `MongoTemplate` from auto-configuration. If your handler uses a different template via `mongoTemplateRef`, the checkpoint save and the handler's writes use different database factories — they cannot participate in the same transaction.

```java theme={null}
// This does NOT provide exactly-once:
@ChangeStream(collection = "orders", mongoTemplateRef = "secondaryMongoTemplate")
@Checkpoint(saveEveryN = 1)
public class OrderProcessor {

    @Autowired
    @Qualifier("secondaryMongoTemplate")
    private MongoTemplate secondaryMongoTemplate;

    @OnInsert
    @Transactional  // transaction on secondary factory
    void handle(Order order, ChangeStreamContext<Order> ctx) {
        secondaryMongoTemplate.save(...);  // secondary factory
        ctx.saveCheckpointNow();           // default factory — NOT in the same transaction
    }
}
```

### `saveEveryN` must be 1

For transactional exactly-once, use `@Checkpoint(saveEveryN = 1)`. With `saveEveryN > 1`, FlowWarden skips checkpoint saves between batches — events processed between checkpoints would not have transactional guarantees on crash recovery.

### Handler must call `saveCheckpointNow()`

FlowWarden's automatic checkpoint save happens **after** the handler returns — outside the `@Transactional` boundary. You must explicitly call `ctx.saveCheckpointNow()` inside the handler to include the checkpoint in the transaction. If you forget, FlowWarden falls back to at-least-once (the checkpoint saves outside the transaction).

***

## Comparison with at-least-once

| Aspect                    | At-least-once (default)            | Exactly-once (transactional)                                                         |
| ------------------------- | ---------------------------------- | ------------------------------------------------------------------------------------ |
| **Delivery guarantee**    | Events may be redelivered on crash | Each event processed exactly once                                                    |
| **Handler requirement**   | Must be idempotent                 | No idempotency needed (for MongoDB writes)                                           |
| **Performance**           | No transaction overhead            | Slight overhead from MongoDB sessions                                                |
| **External side-effects** | Handled naturally                  | Not protected by transaction                                                         |
| **Configuration**         | `@Checkpoint` only                 | `@Checkpoint` + `MongoTransactionManager` + `@Transactional` + `saveCheckpointNow()` |