> For the complete documentation index, see [llms.txt](https://docs.xplatform.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.xplatform.com/talon/developing-applications/microservice-template/event-sourcing-template.md).

# Event Sourcing Template

## Overview

Event Sourcing is Talon's High Availability model that provides the best performance for latency-sensitive applications. With Event Sourcing, Talon replicates inbound messages rather than state changes. Each instance processes the same sequence of messages deterministically to maintain identical state and produce identical outputs.

### How Event Sourcing Works

With Event Sourcing:

* **Inbound messages are replicated** - Messages are replicated to backup instances in parallel with processing
* **State is application-managed** - Your state is stored in POJOs, opaque to the Talon runtime
* **Deterministic processing ensures consistency** - All instances execute identical code on identical inputs
* **No state modeling required** - State can be any Java objects (ADM modeling optional)
* **Lower replication overhead** - Messages replicated in serialized form without re-serialization

### Key Features

* **Application-Controlled State**: Complete control over state structure and management
* **Opaque State**: Runtime doesn't inspect or manage your state
* **Low Latency**: Messages replicated in parallel with handler execution
* **Efficient Replication**: Received messages already serialized, no re-serialization cost
* **Full Flexibility**: Use any Java objects for state, no ADM constraints
* **Deterministic Recovery**: State reconstructed through message replay

### When to Use Event Sourcing

Event Sourcing is ideal when:

* Ultra-low latency is critical (sub-millisecond requirements)
* You need full control over state management
* State logic is complex or involves custom algorithms
* Your state doesn't fit ADM modeling constraints
* You want to use standard Java POJOs
* Replication overhead must be minimized

### Comparison with State Replication

| Aspect                  | Event Sourcing                          | State Replication                  |
| ----------------------- | --------------------------------------- | ---------------------------------- |
| **State Management**    | Application manages (opaque POJOs)      | Runtime manages (transparent)      |
| **State Modeling**      | No modeling required                    | ADM XML required                   |
| **Replication**         | Inbound messages replicated             | State deltas replicated            |
| **Latency**             | Lower (parallel replication)            | Slightly higher (state tracking)   |
| **Complexity**          | More complex (determinism requirements) | Simpler application code           |
| **Recovery**            | State reconstructed by message replay   | State reconstructed from log       |
| **Discipline Required** | High (must avoid divergence)            | Low (runtime enforces consistency) |

See [Consensus Models](/talon/concepts-and-architecture/consensus-models.md) for detailed comparison.

***

## Building an Event Sourcing Microservice

Creating an Event Sourcing microservice involves five main steps:

1. Model your application messages using ADM
2. Annotate your main class for Event Sourcing
3. Manage microservice state in POJOs
4. Write deterministic message handlers
5. Configure storage (clustering and persistence)

### Step 1: Model Application Messages

Define your microservice's messages using the ADM modeling language in your `model.xml` file. Unlike State Replication, you don't need to model your state in ADM - only your messages.

**Example Message Model:**

```xml
<model xmlns="http://www.neeveresearch.com/schema/x-ddl"
       namespace="com.example.orderprocessor.messages"
       defaultFactoryId="1">

  <messages>
    <!-- Inbound message -->
    <message name="NewOrderRequest" id="1">
      <field name="orderId" type="String"/>
      <field name="symbol" type="String"/>
      <field name="quantity" type="Integer"/>
      <field name="price" type="BigDecimal"/>
    </message>

    <!-- Outbound message -->
    <message name="OrderConfirmation" id="2">
      <field name="orderId" type="String"/>
      <field name="status" type="String"/>
      <field name="timestamp" type="Long"/>
    </message>

    <message name="CancelOrderRequest" id="3">
      <field name="orderId" type="String"/>
    </message>
  </messages>
</model>
```

See [The Modeling Language](/talon/developing-applications/modeling-messages-and-state/the-modeling-language.md) for complete modeling guide.

### Step 2: Annotate Main Class for Event Sourcing

Use the `@AppHAPolicy` annotation to declare that your microservice uses Event Sourcing:

```java
package com.example.orderprocessor;

import com.neeve.aep.AepEngine;
import com.neeve.aep.annotations.AppHAPolicy;

@AppHAPolicy(value = AepEngine.HAPolicy.EventSourcing)
public class OrderProcessorApp {
    // ... application code ...
}
```

### Step 3: Manage Application State

With Event Sourcing, your state is completely private to your application. You can use any Java objects - POJOs, collections, or even third-party data structures:

```java
public class OrderProcessorApp {
    // Private POJO state - opaque to Talon
    private class OrderBook {
        private long totalOrders = 0;
        private BigDecimal totalValue = BigDecimal.ZERO;
        private final Map<String, Order> orders = new HashMap<>();

        public void addOrder(Order order) {
            orders.put(order.getOrderId(), order);
            totalOrders++;
            totalValue = totalValue.add(
                order.getPrice().multiply(BigDecimal.valueOf(order.getQuantity()))
            );
        }

        public Order getOrder(String orderId) {
            return orders.get(orderId);
        }

        public long getTotalOrders() {
            return totalOrders;
        }
    }

    // Simple POJO for order state
    private static class Order {
        private String orderId;
        private String symbol;
        private int quantity;
        private BigDecimal price;
        private String status;

        // Constructor, getters, setters...
    }

    // Application state instance
    private final OrderBook orderBook = new OrderBook();
}
```

**Important Points**:

* State is completely private to your application
* No ADM modeling required for state
* Runtime never inspects or manages your state
* State must be reconstructed deterministically on recovery through message replay

### Step 4: Write Deterministic Message Handlers

Message handlers must be deterministic - they must produce identical state and outputs given identical inputs. Handlers receive only the inbound message (not state, since state is private).

```java
import com.neeve.aep.AepMessageSender;
import com.neeve.aep.AepEngine;
import com.neeve.aep.annotations.EventHandler;
import com.neeve.aep.annotations.AppInjectionPoint;

public class OrderProcessorApp {
    private AepMessageSender messageSender;
    private AepEngine engine;
    private final OrderBook orderBook = new OrderBook();

    // Inject message sender for outbound messages
    @AppInjectionPoint
    final public void setMessageSender(AepMessageSender messageSender) {
        this.messageSender = messageSender;
    }

    // Inject engine for accessing engine time
    @AppInjectionPoint
    final public void initialize(AepEngine engine) {
        this.engine = engine;
        // Register message factory
        engine.registerFactory(new com.example.orderprocessor.messages.MessageFactory());
    }

    @EventHandler
    final public void onNewOrder(NewOrderRequest request) {
        // Create order from request - deterministic
        Order order = new Order();
        order.setOrderId(request.getOrderId());
        order.setSymbol(request.getSymbol());
        order.setQuantity(request.getQuantity());
        order.setPrice(request.getPrice());
        order.setStatus("ACCEPTED");

        // Update private POJO state
        orderBook.addOrder(order);

        // Send outbound message
        OrderConfirmation confirmation = OrderConfirmation.create();
        confirmation.setOrderId(order.getOrderId());
        confirmation.setStatus("ACCEPTED");
        // IMPORTANT: Use engine.getEngineTime() for deterministic time
        confirmation.setTimestamp(engine.getEngineTime());
        messageSender.sendMessage("confirmations", confirmation);
    }

    @EventHandler
    final public void onCancelOrder(CancelOrderRequest request) {
        Order order = orderBook.getOrder(request.getOrderId());
        if (order != null && "ACCEPTED".equals(order.getStatus())) {
            // Update state deterministically
            order.setStatus("CANCELLED");

            // Send confirmation
            OrderConfirmation confirmation = OrderConfirmation.create();
            confirmation.setOrderId(order.getOrderId());
            confirmation.setStatus("CANCELLED");
            confirmation.setTimestamp(engine.getEngineTime());
            messageSender.sendMessage("confirmations", confirmation);
        }
    }
}
```

**Critical: Preventing Divergence**

Event Sourcing requires special discipline to avoid state divergence between primary and backup:

{% hint style="danger" %}
**Never use non-deterministic operations in handlers:**

* `System.currentTimeMillis()` - Use `engine.getEngineTime()` instead
* `System.nanoTime()` - Not replicated, will cause divergence
* Random numbers - Will differ on backup
* External I/O or database calls - Results not replicated
* Environment variables - May differ between instances
* Clock-dependent logic - Clocks may be skewed
  {% endhint %}

**Safe Practices**:

* **Use `engine.getEngineTime()`** - Provides deterministic timestamp from message
* **Process only message data** - All decisions based on message content and existing state
* **Use message injection** - Inject results from external operations as new messages
* **Use environment replication** - Tunnel environment data into replication stream (see below)

### Step 5: Configure Storage

Configure storage in your DDL to enable clustering and persistence.

#### Register Message Factories

Only message factories need to be registered (not state factories, since state is opaque):

**DDL Configuration:**

```xml
<app name="order-processor" mainClass="com.example.orderprocessor.OrderProcessorApp">
  <messaging>
    <factories>
      <factory name="com.example.orderprocessor.messages.MessageFactory"/>
    </factories>
    <buses>
      <bus name="orders-bus">
        <channels>
          <channel name="orders" join="true"/>
          <channel name="confirmations" join="false"/>
        </channels>
      </bus>
    </buses>
  </messaging>

  <storage enabled="true">
    <clustering enabled="true">
      <storeName>order-processor-store</storeName>
      <localPort>10000</localPort>
      <discoveryDescriptor>nvds://localhost:4090</discoveryDescriptor>
    </clustering>
    <persistence enabled="true">
      <flushOnCommit>false</flushOnCommit>
      <storeRoot>rdat</storeRoot>
    </persistence>
  </storage>
</app>
```

**Programmatic Registration Alternative:**

```java
@AppInjectionPoint
public void initialize(AepEngine engine) {
    // Register message factory for deserialization
    engine.registerFactory(new com.example.orderprocessor.messages.MessageFactory());
}
```

#### Enable Clustering

Clustering allows multiple instances to discover each other and form an HA cluster:

```xml
<clustering enabled="true">
  <storeName>order-processor-store</storeName>
  <localPort>10000</localPort>
  <localIfAddr>192.168.1.100</localIfAddr>
  <linkParams>SO_REUSEADDR=true,TCP_NODELAY=true</linkParams>
  <discoveryDescriptor>nvds://localhost:4090</discoveryDescriptor>
  <initWaitTime>5000</initWaitTime>
  <failOnMultiplePrimaries>true</failOnMultiplePrimaries>
  <memberElectionPriority>100</memberElectionPriority>
</clustering>
```

**Key clustering concepts**:

* Instances with same `storeName` form a cluster
* One instance elected as primary via leadership election
* Primary establishes messaging and invokes handlers
* Backups receive replicated messages and process them identically

#### Enable Persistence

Persistence logs the message stream to disk for cold start recovery:

```xml
<persistence enabled="true">
  <flushOnCommit>false</flushOnCommit>
  <autoFlushSize>8192</autoFlushSize>
  <storeRoot>rdat</storeRoot>
  <compaction>
    <compactOnStart>false</compactOnStart>
    <compactionThreshold>1024</compactionThreshold>
  </compaction>
</persistence>
```

{% hint style="info" %}
**Clustering Requires Persistence**: When clustering is enabled, persistence must also be enabled. The transaction log is used to initialize new cluster members that connect to the primary.
{% endhint %}

{% hint style="warning" %}
**Cold Start Recovery**: With Event Sourcing, recovery from a cold start (no backup running) requires replaying the entire inbound message stream from disk. For long-running applications with high message volumes, transaction log compaction and checkpointing strategies are critical.
{% endhint %}

See [Storage Configuration](/talon/reference/configuration.md#storage-configuration) for complete configuration reference.

***

## Advanced Techniques

### Using Engine Time

The `AepEngine.getEngineTime()` method provides deterministic timestamps for Event Sourcing applications:

```java
private volatile AepEngine engine;

@AppInjectionPoint
public void initialize(AepEngine engine) {
    this.engine = engine;
}

@EventHandler
public void onMessage(TradeUpdate message) {
    OrderEvent event = OrderEvent.create();

    // Engine time is identical on primary and backup
    event.setTimeReceivedAsTimestamp(engine.getEngineTime());

    messageSender.sendMessage("order-events", event);
}
```

**How It Works**:

* When called from a handler in Event Sourcing mode, returns timestamp from message event
* Timestamp is captured just before handler dispatch
* Same timestamp replicated to backup, ensuring deterministic processing
* Outside handlers or in State Replication mode, returns `System.currentTimeMillis()`

### Message Injection

Message injection allows you to inject messages into the processing stream for fault-tolerant execution. This is useful for:

* **Deferred processing** - Schedule work to be done later
* **Async operations** - Inject results from other threads
* **External system integration** - Inject data from external sources

**Example: Scheduling Deferred Work**

```java
@EventHandler
public void onNewOrder(NewOrderRequest request) {
    // Process order immediately
    processOrder(request);

    // Schedule timeout check for 30 seconds later
    OrderTimeoutCheck timeoutCheck = OrderTimeoutCheck.create();
    timeoutCheck.setOrderId(request.getOrderId());

    // Inject with 30-second delay - replicated to backup
    engine.injectMessage(timeoutCheck, 30000);
}

@EventHandler
public void onOrderTimeoutCheck(OrderTimeoutCheck check) {
    // This executes 30 seconds later on both primary and backup
    Order order = orderBook.getOrder(check.getOrderId());
    if (order != null && "PENDING".equals(order.getStatus())) {
        // Cancel timed-out order
        cancelOrder(order);
    }
}
```

**Example: Integrating External Operations**

```java
@EventHandler
public void onHostnameRequest(BroadcastHostNameRequest request) {
    // Get hostname from local environment (non-deterministic)
    String hostname = InetAddress.getLocalHost().getHostName();

    // Inject result as a message (now deterministic)
    SendHostNameCommand command = SendHostNameCommand.create();
    command.setHostName(hostname);
    engine.injectMessage(command);
}

@EventHandler
public void onSendHostName(SendHostNameCommand command) {
    // This handler processes the injected message
    // Both primary and backup receive the same hostname
    BroadcastHostNameResponse response = BroadcastHostNameResponse.create();
    response.setHostName(command.getHostName());
    messageSender.sendMessage("hostname-response", response);
}
```

See [Scheduling Messages](/talon/developing-applications/authoring-user-code/message-injection.md) for complete message injection documentation.

### Environment Replication

Environment Replication allows you to tunnel local environment data into the replication stream without creating separate transactions. This is more efficient than message injection for frequently-accessed environment data.

**Example: Hostname Provider**

```java
private HostNameEnvironmentProvider hostNameProvider = new HostNameEnvironmentProvider();
private XString hostName = XString.create(256, true, true);

@AppInjectionPoint
public void initialize(AepEngine engine) {
    // Register environment provider
    hostNameProvider.register(engine);
}

@EventHandler
public void onMessage(BroadcastHostNameRequest message) {
    BroadcastHostNameResponse response = BroadcastHostNameResponse.create();

    // Call environment provider - captures or replays hostname
    hostNameProvider.getHostNameTo(hostName);

    // Set hostname in response and send
    response.setHostName(hostName);
    messageSender.sendMessage("hostname-response", response);
}
```

**How It Works**:

* On primary: Provider records hostname lookup in replication buffer
* On backup: Provider replays recorded hostname from buffer
* All within same transaction - no additional message injection overhead

***

## Deterministic Programming Rules

Event Sourcing requires strict adherence to deterministic programming rules:

### What You MUST Do

1. **Base all decisions on message data and existing state only**
2. **Use `engine.getEngineTime()` for all time-dependent logic**
3. **Keep handlers synchronous and single-threaded**
4. **Use message injection for async operations**
5. **Use environment replication for environment data**
6. **Ensure identical handler execution on all instances**

### What You MUST NOT Do

1. **Never call `System.currentTimeMillis()` or `System.nanoTime()`**
2. **Never use `Random` or any non-deterministic algorithms**
3. **Never read from files, databases, or external systems directly**
4. **Never use thread-local storage or instance variables modified outside handlers**
5. **Never depend on environment variables or system properties in handlers**
6. **Never use object identity (e.g., `System.identityHashCode()`) for logic**

See [Programming Fundamentals](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/programming-fundamentals.md) for complete deterministic programming rules.

***

## Related Documentation

### Core Concepts

* [Consensus Models](/talon/concepts-and-architecture/consensus-models.md) - Understanding Event Sourcing vs State Replication
* [Transactions](/talon/concepts-and-architecture/transactions.md) - How transactions work with Event Sourcing
* [Application Lifecycle](/talon/concepts-and-architecture/microservice-operation/lifecycle.md) - Initialization and recovery

### Development

* [Modeling Messages & State](/talon/developing-applications/modeling-messages-and-state.md) - ADM modeling for messages
* [Programming Fundamentals](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/programming-fundamentals.md) - Deterministic coding rules
* [Handling Messages](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages.md) - Writing message handlers
* [Scheduling Messages](https://github.com/neeveresearch/nvx-docs/blob/master/talon/developing-applications/authoring-user-code/message-processing/scheduling-messages.md) - Message injection

### Configuration

* [Storage Configuration](/talon/reference/configuration.md#storage-configuration) - Complete storage configuration reference
* [Configuring Threading](/talon/developing-applications/configuring-the-runtime/threading.md) - Optimize replication performance

### Operations

* [Transaction Log Tool](/talon/operating-applications/analysis-and-troubleshooting/tools/transaction-log-tool.md) - Browse and analyze transaction logs
* [Querying Transaction Logs](/talon/operating-applications/analysis-and-troubleshooting/querying-transaction-logs.md) - XPQL query language

***

## Next Steps

1. **Start modeling**: Define your messages in `model.xml` following ADM syntax
2. **Create application class**: Annotate with `@AppHAPolicy(EventSourcing)` and design state POJOs
3. **Write handlers**: Implement deterministic message handlers
4. **Configure storage**: Enable clustering and persistence in DDL
5. **Test failover**: Verify state consistency through message replay
6. **Test determinism**: Ensure identical processing on multiple instances
7. **Monitor in production**: Track transaction log size and replay time