​

Event Sourcing Implementation

​

What is Event Sourcing?

1. All state changes are captured as events
2. Events are immutable and stored in append-only logs
3. Current state can be reconstructed by replaying events from the beginning
4. Historical state can be reconstructed by replaying events up to any point in time

​

Why Event Sourcing for Workflows?

​

Temporal’s Event Sourcing Model

​

Workflow History Events

• Each workflow execution has its own linear sequence of events
• Events are assigned monotonically increasing EventIDs (starting at 1)
• Events are publicly exposed via APIs (unlike internal state)
• Events contain both metadata and optional payloads

​

Event Structure

message HistoryEvent {
    int64 event_id = 1;                    // Monotonic sequence number
    google.protobuf.Timestamp event_time = 2;  // When event occurred
    EventType event_type = 3;              // Type of event
    int64 task_id = 4;                     // Associated task ID
    
    // Event-specific attributes (one of):
    WorkflowExecutionStartedEventAttributes workflow_execution_started_attributes = 5;
    ActivityTaskScheduledEventAttributes activity_task_scheduled_attributes = 6;
    // ... etc
}

​

Event Types

• WorkflowExecutionStarted
• WorkflowExecutionCompleted
• WorkflowExecutionFailed
• WorkflowExecutionTimedOut
• WorkflowExecutionCanceled
• WorkflowExecutionTerminated

• WorkflowTaskScheduled
• WorkflowTaskStarted
• WorkflowTaskCompleted
• WorkflowTaskFailed
• WorkflowTaskTimedOut

• ActivityTaskScheduled
• ActivityTaskStarted
• ActivityTaskCompleted
• ActivityTaskFailed
• ActivityTaskTimedOut
• ActivityTaskCancelRequested
• ActivityTaskCanceled

• TimerStarted
• TimerFired
• TimerCanceled

• StartChildWorkflowExecutionInitiated
• ChildWorkflowExecutionStarted
• ChildWorkflowExecutionCompleted
• ChildWorkflowExecutionFailed

• WorkflowExecutionSignaled
• WorkflowExecutionUpdateAccepted
• WorkflowExecutionUpdateCompleted

​

Event Generation

​

Who Creates Events?

1. User Application Requests: Start, Signal, Cancel, Terminate, Update
2. Worker Completions: Workflow Task completed, Activity Task completed
3. Timer Expirations: Workflow timers, activity timeouts, workflow timeouts
4. System Operations: Workflow reset, replication

​

Event Generation Flow

​

Event Attributes

message ActivityTaskScheduledEventAttributes {
    string activity_id = 1;
    ActivityType activity_type = 2;
    string task_queue = 3;
    Payloads input = 4;
    Duration schedule_to_close_timeout = 5;
    Duration schedule_to_start_timeout = 6;
    Duration start_to_close_timeout = 7;
    Duration heartbeat_timeout = 8;
    RetryPolicy retry_policy = 9;
    // ... etc
}

​

Event Persistence

​

Storage Structure

• Events are grouped into batches (typically 1-100 events per batch)
• Each batch is a binary blob (protobuf serialized)
• Batches are identified by (namespaceID, workflowID, runID, nodeID)
• NodeID represents the batch number

-- Simplified Cassandra schema
CREATE TABLE history_node (
    tree_id         uuid,
    branch_id       uuid,
    node_id         bigint,
    txn_id          bigint,
    data            blob,      -- Serialized batch of events
    data_encoding   text,
    PRIMARY KEY ((tree_id), branch_id, node_id, txn_id)
);

​

Batching Strategy

• Multiple events per write: Reduces database round-trips
• Configurable batch size: Balance between latency and throughput
• Atomic batch writes: All events in a batch are written together

​

Event Versioning

• Events are organized in a tree structure
• Each branch has a branchToken
• Enables branching history for reset scenarios

​

State Reconstruction

​

Mutable State as Cache

• Summarizes current workflow state
• Avoids replaying entire history on every request
• Persisted alongside events
• Can be regenerated from events if corrupted

​

Consistency Model

// Pseudo-code for state transition
func UpdateWorkflow(workflow, events, tasks) error {
    tx := db.BeginTransaction()
    
    // 1. Append events to history log
    tx.AppendEvents(workflow.ID, events)
    
    // 2. Update mutable state (includes last event ID)
    workflow.MutableState.LastEventID = events.Last().ID
    tx.UpdateMutableState(workflow.ID, workflow.MutableState)
    
    // 3. Add tasks for processing
    tx.AddTasks(workflow.ID, tasks)
    
    // All or nothing
    return tx.Commit()
}

​

Replay Process

1. Receive History Events: Worker gets events since last Workflow Task
2. Replay Workflow Code: Worker executes workflow function
3. Compare Commands: Worker’s commands must match history exactly
4. Non-Determinism Detection: If commands diverge, workflow fails

​

Deterministic Execution

​

Why Determinism Matters

• Workflow code must produce the same commands when replayed
• Same events in, same commands out
• Enables workers to resume workflows without state serialization

​

Determinism Rules

​

Enforcing Determinism

• Workflow sandbox: Restricts available APIs
• Deterministic time: workflow.Now() returns time from events
• Deterministic random: workflow.Random() uses seeded RNG
• Deterministic UUID: workflow.UUID() generates deterministic UUIDs

​

Event-Driven State Machine

​

Commands as Intent

• ScheduleActivityTask
• StartTimer
• StartChildWorkflowExecution
• CompleteWorkflowExecution
• CancelTimer

​

Command Processing

1. Validate: Check if command is valid given current state
2. Generate Events: Create events representing the command’s effect
3. Update State: Apply events to Mutable State
4. Create Tasks: Schedule follow-up tasks (Transfer, Timer)

​

Querying Event History

​

GetWorkflowExecutionHistory API

resp, err := client.GetWorkflowHistory(
    ctx,
    workflowID,
    runID,
    false,  // waitForNewEvents
    enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT,
)

for resp.HasNext() {
    event, err := resp.Next()
    // Process event
}

​

Event Filtering

• All Events: Complete history
• Close Event: Only workflow completion event

​

Pagination

• Events are paginated (default page size: 1000)
• Use continuation tokens for next page
• Can be streamed efficiently

​

Advanced Topics

​

Workflow Reset

1. Choose a reset point (EventID)
2. Create a new branch from that point
3. Original branch is marked as ancestor
4. New execution continues from reset point

// Reset workflow to Event ID 50
client.ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
    WorkflowId: workflowID,
    RunId:      runID,
    Reason:     "Fixing bug in activity",
    EventId:    50,
})

​

Multi-Cluster Replication

• Events are replicated across clusters
• Each cluster maintains its own event log
• Version vectors track causal relationships
• Conflict resolution merges divergent histories

​

History Archival

• Events moved from primary database to archival storage (S3, GCS)
• Reduces load on primary database
• Archived history can be retrieved on demand
• Configurable retention period before archival

​

History Compression

• History can be compressed (gzip, etc.)
• Reduces storage costs
• Transparent decompression on read

​

Performance Considerations

​

Event Size

​

History Size

• Use Continue-As-New to “reset” a workflow’s history
• Starts a new run with same workflow ID
• Previous run’s history is preserved but not loaded

// Workflow code
if executionCount > 1000 {
    return workflow.NewContinueAsNewError(
        ctx,
        WorkflowFunction,
        newArgs,
    )
}

​

Read Optimization

• Recently accessed workflows are cached in memory
• Avoids reading history from database
• Significantly improves latency
• Cache invalidation on state changes

​

Comparison with Other Patterns

​

vs. CRUD (Mutable State)

• Overwrites state in place
• No history of changes
• Difficult to debug past issues

• Appends immutable events
• Complete audit trail
• Can reconstruct any past state

​

vs. Event Streaming (Kafka)

• Generic event transport
• No built-in state management
• Consumers maintain their own state

• Specialized for workflow execution
• Integrated state management (Mutable State)
• Deterministic replay semantics
• Strongly typed event schemas

​

Further Reading