# @Saga > Declare a saga with an annotation, including one fed from a broker. [Spring Boot](/docs/spring-boot/) On the [Spring Boot starter](/docs/spring-boot/starter/) you do not wire a `SagaRunner` yourself. Annotate a no-arg factory method returning a `Saga` with `org.occurrent.annotation.Saga`. The framework registers it as a managed saga, subscribing through the same catch-up, durable-resume, and [competing-consumer](/docs/layer-0/competing-consumers/) machinery as [`@Subscription`](/docs/spring-boot/subscription-annotations/): * Java ```java import org.occurrent.annotation.Saga; @Component class OrderFulfillmentSaga { @Saga(id = "order-fulfillment") org.occurrent.dsl.saga.Saga, OrderCommand> orderFulfillment() { return FlowSaga.builder() .correlateAll(OrderEvent::orderId) .startsOn(OrderPlaced.class, order -> List.of(new ReservePayment(order.orderId(), order.amount()))) .step("awaiting-payment", step -> step .on(PaymentReserved.class, Continuation.end(), payment -> List.of(new ShipOrder(payment.orderId()))) .on(PaymentFailed.class, Continuation.end(), failure -> List.of(new CancelOrder(failure.orderId(), failure.reason()))) .timeout(Duration.ofMinutes(30), Continuation.end(), received -> List.of(new CancelOrder(received.initiating(OrderPlaced.class).orderId(), "payment timeout")))) .build(); } } ``` * Kotlin ```kotlin import org.occurrent.annotation.Saga @Component class OrderFulfillmentSaga { @Saga(id = "order-fulfillment") fun orderFulfillment() = saga { correlateAll { it.orderId } startsOn { order -> issue(ReservePayment(order.orderId, order.amount)) } step("awaiting-payment") { on(then = end) { payment -> issue(ShipOrder(payment.orderId)) } on(then = end) { failure -> issue(CancelOrder(failure.orderId, failure.reason)) } timeout(after = Duration.ofMinutes(30), then = end) { received -> issue(CancelOrder(received.initiating().orderId, "payment timeout")) } } } } ``` The annotation and the DSL class share the name `Saga`, so a Java factory method has to qualify one of them, as above. Kotlin has no such clash, since the DSL entry point is the lowercase `saga` function and the return type can be inferred. `@Saga` takes: | Attribute | Description | | :-------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `id` | The durable subscription and checkpoint key (required). | | `startAt` | `StartPosition.BEGINNING`, `NOW`, or `DEFAULT`, the same start-position idea as [`@Subscription`](/docs/spring-boot/subscription-annotations/#subscription-start-position). | | `startAtGlobalPosition` | Start after a specific global position instead, to rewind a saga to a known point. Mutually exclusive with a non-default `startAt`. | | `resumeBehavior` | `DEFAULT` or `SAME_AS_START_AT`, the same [resume-behavior idea](/docs/spring-boot/subscription-annotations/#subscription-start-position) as `@Subscription`. | | `startupMode` | `DEFAULT`, `WAIT_UNTIL_STARTED`, or `BACKGROUND`, the same [startup-mode idea](/docs/spring-boot/subscription-annotations/#subscription-startup-mode) as `@Subscription`. The default is `BACKGROUND`, so a saga replaying its history does not hold up application startup. Before 0.32.0 this attribute was accepted and ignored, and a replaying saga blocked startup, so set `WAIT_UNTIL_STARTED` if you depended on that. Timers wait for the replay to finish either way. | | `capability` | `AGNOSTIC` (both stream and DCB events) or `STREAM` (stream events only). | | `store` / `storeName` | Select the `SagaStateStore` bean by type or name. With both unset the store resolves by convention, the unique `SagaStateStore` bean, otherwise a zero-config MongoDB store in a `saga-` collection. | | `commandDispatcher` / `commandDispatcherName` | Select the `CommandDispatcher` bean by type or name, otherwise the unique `CommandDispatcher` bean. There is no default dispatcher, since it is usually a lambda over your `ApplicationService`. | | `source` | `EVENT_STORE` (the default) reads the event store. `PUSH` feeds the saga from a `PushSubscriptionModel` bean instead, see [Fed from a broker](/docs/spring-boot/saga-annotation/#fed-from-a-broker). | | `catchup` | For a push saga only. `FROM_EVENT_STORE` (the default) replays history once before going live, `NONE` takes live events only and needs no event store. | | `subscriptionModel` / `subscriptionModelName` | Select the `PushSubscriptionModel` bean by type or name when `source = PUSH`. | | `redeliveryDetection` | `REQUIRED` (the default) refuses an event the saga cannot recognize a redelivery of. `BEST_EFFORT` takes it anyway with one warning, see [Forward the Occurrent extensions](/docs/spring-boot/saga-annotation/#fed-from-a-broker). | `@Saga` is blocking-only in this first version, the reactive starter does not register it. ## Fed from a broker (Added in 0.32.0) A saga does not have to read the event store. Set `source = Source.PUSH` and point it at a [`PushSubscriptionModel`](/docs/layer-0/push-subscription/) bean, and it reacts to whatever your listener hands that model, from RabbitMQ, Kafka, or any other source that can deliver an event again: * Java ```java import org.occurrent.annotation.Saga; import org.occurrent.annotation.Source; @Component class OrderFulfillmentSaga { @Saga(id = "order-fulfillment", source = Source.PUSH, subscriptionModelName = "orderEvents") org.occurrent.dsl.saga.Saga, OrderCommand> orderFulfillment() { return FlowSaga.builder() .correlateAll(OrderEvent::orderId) .startsOn(OrderPlaced.class, order -> List.of(new ReservePayment(order.orderId(), order.amount()))) .step("awaiting-payment", step -> step .on(PaymentReserved.class, Continuation.end(), payment -> List.of(new ShipOrder(payment.orderId())))) .build(); } } @Configuration class OrderEventsConfig { @Bean("orderEvents") PushSubscriptionModel orderEvents() { return new PushSubscriptionModel(); } } ``` * Kotlin ```kotlin import org.occurrent.annotation.Saga import org.occurrent.annotation.Source @Component class OrderFulfillmentSaga { @Saga(id = "order-fulfillment", source = Source.PUSH, subscriptionModelName = "orderEvents") fun orderFulfillment() = saga { correlateAll { it.orderId } startsOn { order -> issue(ReservePayment(order.orderId, order.amount)) } step("awaiting-payment") { on(then = end) { payment -> issue(ShipOrder(payment.orderId)) } } } } @Configuration class OrderEventsConfig { @Bean("orderEvents") fun orderEvents() = PushSubscriptionModel() } ``` Your listener calls `acceptOrDefer(cloudEvent)` on that bean and acts on the `RoutingOutcome` it returns, as shown for the [`PushSubscriptionModel`](/docs/layer-0/push-subscription/). The saga takes it from there, with the same `correlateAll`, steps, timeouts and state store it always used. An event the saga does not declare is ignored rather than refused. The saga subscribes with a filter on its declared event types. So when the queue delivers something else, `acceptOrDefer` returns `FILTERED` and your listener acknowledges as usual, with nothing logged. A queue that also delivers event types the saga never declared therefore works fine, and each extra event costs one filter check. The redelivery check described under [Forward the Occurrent extensions](/docs/spring-boot/saga-annotation/#forward-the-occurrent-extensions) below only runs for events the saga declares. So a listener that drops the Occurrent extensions shows up only once a declared event arrives without them. `subscriptionModelName` is only needed when there is something to choose between. [A `PushSubscriptionModel` feeds exactly one saga or projection](/docs/layer-0/push-subscription/), so an application with several push sagas declares several feed beans, and each `@Saga` names its own. With a single feed bean, as here, the name can be dropped and the starter finds the bean on its own. It refuses to guess between several, with a message naming them. Declaring the bean also does not replace the default subscription model the starter contributes. The starter skips models without a start position, checkpoint or catch-up when it decides whether the application brought its own subscription model, and a push feed has none of the three, so your event-store sagas and subscriptions keep the durable default and the feed runs beside it. By default the starter puts a [replay in front of the feed](/docs/layer-0/push-subscription/), so a saga that has never run works through the event store’s history first and only then starts taking live events. That is what you want when this application wrote the events and the broker is only how they reach the saga. ### When the events are not in your event store If another application writes the events, your event store does not hold them, so there is nothing to replay. A replay would either find nothing or, worse, apply unrelated events that happen to live in the same store. Say so with `catchup = Catchup.NONE`: * Java ```java @Saga(id = "shipment-tracking", source = Source.PUSH, subscriptionModelName = "warehouseEvents", catchup = Catchup.NONE) org.occurrent.dsl.saga.Saga, ShipmentCommand> shipmentTracking() { return /* ... */; } ``` * Kotlin ```kotlin @Saga(id = "shipment-tracking", source = Source.PUSH, subscriptionModelName = "warehouseEvents", catchup = Catchup.NONE) fun shipmentTracking() = saga { /* ... */ } ``` This takes live events only and touches no event store at all, so the application needs neither a `PositionOrderedReader` nor a `CheckpointStorage` bean. The bean requirements and failure messages are the same as for a [push projection with `catchup = Catchup.NONE`](/docs/layer-0/push-subscription/). Leave the default in place without those beans and startup fails with a message telling you to set `catchup = Catchup.NONE`, rather than a bare missing-bean error. A saga that has run before picks up where it left off either way, because its per-instance state lives in its `SagaStateStore` and not in the feed. The difference shows on a first run against an existing history. With the default the saga is brought up to date from the event store before it reacts. With `Catchup.NONE` it starts from nothing and reacts only to what arrives from here on. ### Forward the Occurrent extensions A saga recognizes a redelivered event by its `streamid` together with its `streamversion`, or by its `position`. That pair, or the position, is the event’s redelivery key. A broker delivers at least once, so an event with none of those would make the saga react a second time and issue its commands again on every redelivery. The saga therefore refuses such an event by throwing `SagaRedeliveryDetectionException` before the reaction runs, instead of acknowledging it. What happens to the event after that is up to your feed. Whatever the feed does, the listener that dropped the metadata sees the exception, and that listener is where it can be fixed. Occurrent’s own stored events always have the extensions, so this only happens with events your listener forwards without them. Your feed might carry none of that redelivery metadata, another application’s broker for example, while every command the saga issues is still safe to receive more than once. When both are true, opt out with `@Saga(redeliveryDetection = RedeliveryDetection.BEST_EFFORT)`, or `SagaRunnerConfig.withRedeliveryDetection(BEST_EFFORT)` when you drive `SagaRunner` yourself. The saga then takes those events and logs one warning, naming the saga so you can find it. Setting `BEST_EFFORT` on an event-store saga (`source = EVENT_STORE`) is rejected at startup instead, since those events always carry the extensions and there is no metadata gap for it to change. The reasoning is in [ADR 0109](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0109-a-saga-refuses-an-event-it-cannot-recognise-a-redelivery-of.md). This is also why a `@Saga` accepts only a `PushSubscriptionModel` and not a [`DomainEventFeed`](/docs/layer-0/push-subscription/#feeding-domain-events-instead-of-cloudevents), which a `@Projection` does accept. A domain event feed carries no stream metadata, so a saga bound to one would lose its redelivery protection without saying anything. ### What a push saga cannot set `startAt`, `startAtGlobalPosition` and `resumeBehavior` are rejected rather than quietly ignored. A replay always starts at the beginning, and where the live feed resumes after a restart is the broker’s business. Occurrent only records that the catch-up finished, so a restart skips the replay and the broker redelivers whatever the consumer had not acknowledged. `startupMode` does apply, under the default `catchup`, because that replay is real work on the startup path. Use `StartupMode.BACKGROUND` to keep a long history from holding up startup. It is rejected together with `Catchup.NONE`, where there is no replay to wait for. Setting `catchup` on an event-store saga is rejected too, since ignoring it would leave the saga reading the whole history it was asked to skip. ### Starting it yourself A push feed is a bean you supply, so `occurrent.subscription.mode=manual` cannot withhold it the way it withholds a subscription Occurrent owns. Inject `ManualStartPushSources` and start the saga when the application is ready, which is what you want for a saga behind a leader election, where every node would otherwise react to the same event: ```java @Autowired ManualStartPushSources pushSources; void becameLeader() { pushSources.start("order-fulfillment"); } ``` `startAll()` starts every withheld push source, projections and sagas alike, and `pendingIds()` says what is still waiting. The saga’s instances are readable through [`SagaInstances`](/docs/layer-1/observing-sagas/) while it waits, since that reads the state store rather than a running subscription, so you can look at what is in flight before deciding to start it. The design rationale, including why a domain event feed is refused and why the timers wait for the handover, is in [ADR 0096](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0096-a-push-fed-saga-may-have-no-history-to-replay.md). ## Observing @Saga instances The `@Saga` registrar publishes each saga’s [`SagaInstances`](/docs/layer-1/observing-sagas/) under a registry keyed by saga id: * Java ```java @Service class SagaDashboard { private final SagaInstancesRegistry registry; SagaDashboard(SagaInstancesRegistry registry) { this.registry = registry; } List stalled(String sagaId, Duration threshold) { SagaInstances instances = registry.get(sagaId); List stalled = new ArrayList<>( instances.findByStatus(SagaStatus.ACTIVE, Instant.now().minus(threshold), 100)); stalled.addAll(instances.findByStatus(SagaStatus.QUARANTINED, Instant.now(), 100)); return stalled; } } ``` * Kotlin ```kotlin @Service class SagaDashboard(private val registry: SagaInstancesRegistry) { fun stalled(sagaId: String, threshold: Duration): List { val instances = registry.get(sagaId) return instances.findByStatus(SagaStatus.ACTIVE, Instant.now().minus(threshold), 100) + instances.findByStatus(SagaStatus.QUARANTINED, Instant.now(), 100) } fun sagaIds(): Set = registry.sagaIds() } ``` `get(id)` throws and names every id that is registered, which is what you want when the id is a constant in your own code. `find(id)` returns an `Optional` for an id that came from a request or a configuration value. `sagaIds()` lists them so a dashboard does not hardcode ids. Each saga is also published under the bean name `sagaInstances-`, reachable with `getBean` or a `@Qualifier` if you prefer to inject one saga’s view directly. Each saga’s running `SagaSubscription` is published the same way, as `sagaSubscription-`. It’s the handle `SagaRunner` returns on the programmatic path, so it gives you the saga’s event `Subscription` and `waitUntilStarted()` as well as `instances()`. A saga on a push feed that you start through `ManualStartPushSources` gets this bean when you start it, not when the context refreshes. One timing constraint comes with the annotation path. A `@Saga` factory can only run once the beans it collaborates with are wired, which is after the context has refreshed, so the registry holds nothing until that scan has run. Inject it and read it when a request arrives, never from another bean’s constructor. You never build a `SagaRunnerConfig` for a `@Saga`, so the [quarantine](/docs/layer-1/observing-sagas/#quarantined-instances) budget is the property `occurrent.saga.quarantine-after`, one value for every `@Saga` in the application. It has no default. Leave it out, or leave it blank, to keep quarantine off, or set it to a positive duration to turn it on. Zero or a negative value stops an application with at least one `@Saga` from starting. ```properties occurrent.saga.quarantine-after=5m ``` ## Several instances of a @Saga The starter uses a [competing-consumer](/docs/layer-0/competing-consumers/) subscription model by default, so only one instance receives a saga’s events at a time. [Gating the timer poller](/docs/layer-1/saga-delivery/) is on by default too. A `@Saga` runner reuses the same `SpringMongoLeaseCompetingConsumerStrategy` the subscription model already builds, so nothing extra to wire. Both the poll interval and the gating are configurable under `occurrent.saga`: ```yaml occurrent: saga: timer-poll-interval: 15s # the default, lower it only if you rely on short timeouts firing promptly competing-consumer: enabled: true # set to false to let every instance poll (the uncoordinated behavior) ``` Turn `competing-consumer.enabled` off and the poller runs on every instance.