Event Sourcing: Audit, Lineage Tracking, and State Replay
Most systems use the CRUD model: Create, Read, Update, Delete. This works perfectly for simple scenarios. But when you face these requirements, CRUD falls short:
CoomiaPublished on December 17, 202512 min read
Share this articleTwitter / X
Event Sourcing: Audit, Lineage Tracking, and State Replay
“Series: S10 Design Patterns · Article 3 | Level: Advanced | Reading Time: 18 min
#TL;DR
- Event Sourcing does not store current state — it stores all events that led to the current state. Current state is derived by replaying the event sequence.
- In coomia-dip, Event Sourcing is the infrastructure for audit compliance, data lineage tracking, and state replay — every Ontology object change, every Action execution, every reasoning chain trigger is recorded as an immutable event.
- By combining Nessie (Git-like version control) and Apache Iceberg (time-travel queries), coomia-dip achieves full-chain traceability from metadata to data.
#Introduction: Why CRUD Is Not Enough
Most systems use the CRUD model: Create, Read, Update, Delete. This works perfectly for simple scenarios. But when you face these requirements, CRUD falls short:
Code
Regulatory audit: How was this transaction's risk score calculated 3 years ago?
Incident investigation: Why did the system reject this approval yesterday at 15:30?
Compliance retrospection: What rule version was in effect at that time?
Data correction: A month-old data error was found — can we precisely fix it and recalculate impact?
CRUD only tells you "what is now," not "why it became this way." You need to dig through logs, check backups, ask colleagues, and you still may not figure it out.
Event Sourcing solves this fundamentally: store the process, not the result.
#Part 1: Core Concepts of Event Sourcing
#1.1 Events
An event is a fact that has occurred — immutable and undeletable:
Python
@dataclass(frozen=True)
class DomainEvent:
"""Immutable domain event."""
event_id: str # Globally unique ID
event_type: str # Event type
aggregate_id: str # Aggregate root ID
aggregate_type: str # Aggregate root type
sequence_number: int # Sequence within aggregate
timestamp: datetime # When the event occurred
payload: dict # Event data
metadata: EventMetadata # Metadata
causation_id: str | None = None # Causation chain ID
correlation_id: str | None = None # Correlation ID
@dataclass(frozen=True)
class EventMetadata:
"""Event metadata for audit and lineage."""
actor_id: str # Who performed the action
actor_type: str # user / system / rule / agent
tenant_id: str # Tenant
world_id: str # World context
source_plane: str # Source Layer
trace_id: str # Distributed trace ID
rule_version: str | None = None # Rule version that triggered
model_version: str | None = None # Inference model version
#1.2 Event Store
The Event Store is an append-only log. coomia-dip uses Apache Kafka as the event bus and Apache Iceberg for persistent storage:
Python
class EventStore:
"""Append-only event store backed by Kafka + Iceberg."""
async def append(
self,
events: list[DomainEvent],
expected_version: int | None = None,
) -> None:
"""Append events with optimistic concurrency control."""
if expected_version is not None:
current = await self._get_current_version(
events[0].aggregate_id
)
if current != expected_version:
raise ConcurrencyConflict(
f"Expected version {expected_version}, "
f"got {current}"
)
for event in events:
# Write to Kafka (real-time consumption)
await self.kafka_producer.send(
topic=f"events.{event.aggregate_type}",
key=event.aggregate_id,
value=self.serializer.serialize(event),
)
# Synchronously write to Iceberg (persistence + time travel)
await self.iceberg_writer.write_events(events)
async def load_events(
self,
aggregate_id: str,
from_version: int = 0,
to_version: int | None = None,
as_of: datetime | None = None,
) -> list[DomainEvent]:
"""Load events for an aggregate, optionally at a point in time."""
if as_of:
# Time travel query via Iceberg
return await self.iceberg_reader.read_events(
aggregate_id=aggregate_id,
snapshot_time=as_of,
)
return await self.iceberg_reader.read_events(
aggregate_id=aggregate_id,
from_version=from_version,
to_version=to_version,
)
#1.3 Aggregate Reconstruction
Reconstruct current state from an event sequence:
Python
class AggregateRepository:
"""Reconstruct aggregates from event streams."""
async def load(
self,
aggregate_id: str,
as_of: datetime | None = None,
) -> Aggregate:
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
as_of=as_of,
)
if not events:
raise AggregateNotFound(aggregate_id)
aggregate = self.aggregate_factory.create_empty(
events[0].aggregate_type
)
for event in events:
aggregate.apply(event)
return aggregate
#Part 2: Event Sourcing in coomia-dip
#2.1 Ontology Change Events
Every operation on an Ontology object produces events:
Python
class OntologyEventTypes:
"""All ontology-level events."""
# Schema changes
OBJECT_TYPE_CREATED = "ontology.object_type.created"
OBJECT_TYPE_PROPERTY_ADDED = "ontology.object_type.property_added"
OBJECT_TYPE_PROPERTY_MODIFIED = "ontology.object_type.property_modified"
RELATION_TYPE_CREATED = "ontology.relation_type.created"
ACTION_TYPE_REGISTERED = "ontology.action_type.registered"
# Instance changes
OBJECT_CREATED = "ontology.object.created"
OBJECT_UPDATED = "ontology.object.updated"
OBJECT_DELETED = "ontology.object.deleted"
RELATION_CREATED = "ontology.relation.created"
RELATION_DELETED = "ontology.relation.deleted"
# Action execution
ACTION_SUBMITTED = "ontology.action.submitted"
ACTION_APPROVED = "ontology.action.approved"
ACTION_EXECUTED = "ontology.action.executed"
ACTION_ROLLED_BACK = "ontology.action.rolled_back"
# Reasoning and decisions
REASONING_TRIGGERED = "reasoning.chain.triggered"
REASONING_COMPLETED = "reasoning.chain.completed"
DECISION_MADE = "decision.made"
RULE_FIRED = "rule.fired"
#2.2 Complete Event Flow for Action Execution
A full lifecycle of an Action (e.g., "approve loan application") under event sourcing:
Python
class LoanApprovalActionHandler:
"""Handle loan approval with full event sourcing."""
async def handle(self, command: ApproveLoanCommand) -> ActionResult:
events = []
# 1. Submission event
events.append(DomainEvent(
event_type="action.loan_approval.submitted",
aggregate_id=command.loan_id,
payload={
"applicant_id": command.applicant_id,
"amount": command.amount,
"officer_id": command.officer_id,
},
metadata=self._build_metadata(command),
))
# 2. Rule evaluation event
risk_result = await self.risk_engine.evaluate(command.loan_id)
events.append(DomainEvent(
event_type="rule.credit_risk.evaluated",
aggregate_id=command.loan_id,
payload={
"risk_score": risk_result.score,
"risk_factors": risk_result.factors,
"rule_version": risk_result.rule_version,
"model_version": risk_result.model_version,
},
))
# 3. Reasoning chain event
reasoning = await self.reasoning_engine.run_chain(
chain="loan_assessment",
context={"loan_id": command.loan_id},
)
events.append(DomainEvent(
event_type="reasoning.loan_assessment.completed",
aggregate_id=command.loan_id,
payload={
"recommendation": reasoning.recommendation,
"confidence": reasoning.confidence,
"reasoning_trace": reasoning.trace,
},
))
# 4. Decision event
decision = self._make_decision(risk_result, reasoning)
events.append(DomainEvent(
event_type="decision.loan_approval.made",
aggregate_id=command.loan_id,
payload={
"decision": decision.outcome,
"approved_amount": decision.approved_amount,
"conditions": decision.conditions,
},
))
# Atomically append all events
await self.event_store.append(events)
return ActionResult(decision=decision, events=events)
#Part 3: Audit and Compliance
#3.1 Audit Queries
Event sourcing makes audit queries straightforward:
Python
class AuditService:
"""Audit service powered by event sourcing."""
async def get_audit_trail(
self,
entity_id: str,
start_time: datetime | None = None,
end_time: datetime | None = None,
actor_filter: str | None = None,
) -> list[AuditEntry]:
"""Get complete audit trail for any entity."""
events = await self.event_store.load_events(
aggregate_id=entity_id,
)
entries = []
for event in events:
if start_time and event.timestamp < start_time:
continue
if end_time and event.timestamp > end_time:
continue
if actor_filter and event.metadata.actor_id != actor_filter:
continue
entries.append(AuditEntry(
timestamp=event.timestamp,
action=event.event_type,
actor=event.metadata.actor_id,
actor_type=event.metadata.actor_type,
details=event.payload,
trace_id=event.metadata.trace_id,
))
return entries
async def reconstruct_state_at(
self,
entity_id: str,
point_in_time: datetime,
) -> dict:
"""Reconstruct exact state at any point in time."""
aggregate = await self.aggregate_repo.load(
aggregate_id=entity_id,
as_of=point_in_time,
)
return aggregate.to_dict()
#3.2 Compliance Report Generation
Python
class ComplianceReporter:
"""Generate compliance reports from event streams."""
async def generate_decision_report(
self,
decision_id: str,
) -> ComplianceReport:
"""Generate full decision audit for regulators."""
events = await self.event_store.load_events(
aggregate_id=decision_id
)
causal_chain = self._build_causal_chain(events)
steps = []
for event in causal_chain:
step = ComplianceStep(
timestamp=event.timestamp,
action=event.event_type,
rule_version=event.metadata.rule_version,
model_version=event.metadata.model_version,
input_data=event.payload.get("input"),
output_data=event.payload.get("output"),
actor=event.metadata.actor_id,
)
steps.append(step)
return ComplianceReport(
decision_id=decision_id,
steps=steps,
generated_at=datetime.utcnow(),
reproducible=True,
)
#Part 4: Data Lineage Tracking
#4.1 Building the Lineage Graph
Event sourcing naturally provides data lineage information:
Python
class LineageTracker:
"""Track data lineage from event streams."""
async def build_lineage_graph(
self,
entity_id: str,
depth: int = 5,
) -> LineageGraph:
"""Build lineage graph showing data provenance."""
graph = LineageGraph()
visited = set()
await self._traverse_lineage(
entity_id, graph, visited, depth
)
return graph
async def _traverse_lineage(
self,
entity_id: str,
graph: LineageGraph,
visited: set,
remaining_depth: int,
) -> None:
if entity_id in visited or remaining_depth <= 0:
return
visited.add(entity_id)
events = await self.event_store.load_events(
aggregate_id=entity_id
)
for event in events:
node = LineageNode(
entity_id=entity_id,
event_type=event.event_type,
timestamp=event.timestamp,
source=event.metadata.source_plane,
)
graph.add_node(node)
if event.causation_id:
causation_event = await self.event_store.get_event(
event.causation_id
)
if causation_event:
graph.add_edge(
from_id=causation_event.aggregate_id,
to_id=entity_id,
relation="caused_by",
event_type=causation_event.event_type,
)
await self._traverse_lineage(
causation_event.aggregate_id,
graph, visited, remaining_depth - 1,
)
#4.2 Impact Analysis
Track downstream impact when a data source changes:
Python
class ImpactAnalyzer:
"""Analyze downstream impact of data changes."""
async def analyze_impact(
self,
source_entity_id: str,
change_type: str,
) -> ImpactReport:
"""Find all entities affected by a change."""
affected = []
queue = [(source_entity_id, 0)]
visited = set()
while queue:
entity_id, depth = queue.pop(0)
if entity_id in visited:
continue
visited.add(entity_id)
downstream = await self.event_store.find_events_referencing(
entity_id=entity_id
)
for event in downstream:
impact = ImpactEntry(
affected_entity=event.aggregate_id,
affected_type=event.aggregate_type,
relationship=event.event_type,
depth=depth,
)
affected.append(impact)
queue.append((event.aggregate_id, depth + 1))
return ImpactReport(
source=source_entity_id,
change_type=change_type,
affected_entities=affected,
total_impact=len(affected),
)
#Part 5: State Replay and Time Travel
#5.1 Full Replay
Rebuild system state at any point in time from scratch:
Python
class StateReplayer:
"""Replay events to reconstruct system state."""
async def replay_to(
self,
target_time: datetime,
entity_filter: str | None = None,
) -> ReplayResult:
"""Replay all events up to a target timestamp."""
stream = self.event_store.stream_all_events(
until=target_time,
entity_filter=entity_filter,
)
state = {}
event_count = 0
async for event in stream:
aggregate_id = event.aggregate_id
if aggregate_id not in state:
state[aggregate_id] = self.aggregate_factory.create_empty(
event.aggregate_type
)
state[aggregate_id].apply(event)
event_count += 1
if event_count % 10000 == 0:
self.progress_reporter.report(
f"Replayed {event_count} events"
)
return ReplayResult(
target_time=target_time,
entities_rebuilt=len(state),
events_replayed=event_count,
state=state,
)
#5.2 Partial Replay (What-If Analysis)
Modify certain events and replay to simulate "what if a different decision was made":
Python
class WhatIfAnalyzer:
"""Perform what-if analysis by modifying and replaying events."""
async def analyze(
self,
entity_id: str,
modifications: list[EventModification],
) -> WhatIfResult:
"""Replay with modified events to see alternative outcomes."""
original_events = await self.event_store.load_events(
aggregate_id=entity_id
)
modified_events = self._apply_modifications(
original_events, modifications
)
original_state = self._replay_events(original_events)
modified_state = self._replay_events(modified_events)
diff = self._compute_diff(original_state, modified_state)
return WhatIfResult(
original_outcome=original_state.to_dict(),
modified_outcome=modified_state.to_dict(),
differences=diff,
affected_downstream=await self._trace_downstream_impact(
entity_id, diff
),
)
#Part 6: Integration with Nessie
#6.1 Schema Versioning
coomia-dip uses Nessie for Git-like Schema version management, complementing event sourcing:
Python
class NessieEventIntegration:
"""Integrate Nessie version control with event sourcing."""
async def record_schema_change(
self,
schema_change: SchemaChange,
) -> None:
# 1. Commit schema change in Nessie
nessie_commit = await self.nessie_client.commit(
branch=schema_change.branch,
operations=[
NessieOp.put(
key=schema_change.schema_key,
content=schema_change.new_schema,
)
],
message=schema_change.description,
)
# 2. Record event with Nessie commit hash
event = DomainEvent(
event_type="ontology.schema.changed",
aggregate_id=schema_change.object_type_id,
payload={
"change_type": schema_change.change_type,
"old_schema": schema_change.old_schema,
"new_schema": schema_change.new_schema,
"nessie_commit": nessie_commit.hash,
"nessie_branch": schema_change.branch,
},
)
await self.event_store.append([event])
#6.2 Cross-Version Queries
Combine Nessie branches with event sourcing for cross-version queries:
Python
class CrossVersionQuery:
"""Query data across different schema versions."""
async def query_at_version(
self,
object_type: str,
nessie_ref: str,
filters: dict,
) -> list[dict]:
"""Query objects using the schema at a specific version."""
schema = await self.nessie_client.get_content(
key=f"schemas/{object_type}",
ref=nessie_ref,
)
events = await self.event_store.load_events(
aggregate_type=object_type,
as_of=await self._ref_to_timestamp(nessie_ref),
)
objects = self._rebuild_with_schema(events, schema)
return self._apply_filters(objects, filters)
#Part 7: Performance Optimization
#7.1 Snapshots
When event counts grow large, replaying from scratch is too slow. Snapshots solve this:
Python
class SnapshotManager:
"""Manage aggregate snapshots for performance."""
SNAPSHOT_INTERVAL = 100
async def load_with_snapshot(
self,
aggregate_id: str,
) -> Aggregate:
snapshot = await self.snapshot_store.get_latest(aggregate_id)
if snapshot:
aggregate = self.aggregate_factory.from_snapshot(snapshot)
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
from_version=snapshot.version + 1,
)
else:
aggregate = self.aggregate_factory.create_empty()
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
)
for event in events:
aggregate.apply(event)
if aggregate.version - (snapshot.version if snapshot else 0) >= self.SNAPSHOT_INTERVAL:
await self._create_snapshot(aggregate)
return aggregate
async def _create_snapshot(self, aggregate: Aggregate) -> None:
snapshot = AggregateSnapshot(
aggregate_id=aggregate.id,
aggregate_type=aggregate.type,
version=aggregate.version,
state=aggregate.serialize(),
created_at=datetime.utcnow(),
)
await self.snapshot_store.save(snapshot)
#7.2 Event Compaction
For high-frequency aggregates, compact event streams while preserving full history:
Python
class EventCompactor:
"""Compact event streams while preserving full history."""
async def compact(
self,
aggregate_id: str,
before: datetime,
) -> CompactionResult:
events = await self.event_store.load_events(
aggregate_id=aggregate_id,
)
to_compact = [e for e in events if e.timestamp < before]
to_keep = [e for e in events if e.timestamp >= before]
if len(to_compact) < 50:
return CompactionResult(compacted=False)
summary = self._create_summary_event(to_compact)
await self.cold_storage.archive(to_compact)
await self.event_store.replace_compacted(
aggregate_id=aggregate_id,
summary_event=summary,
compacted_count=len(to_compact),
)
return CompactionResult(
compacted=True,
original_count=len(to_compact),
archived=True,
)
#Part 8: CQRS Separation
#8.1 Command Side and Query Side
Event sourcing is typically paired with CQRS (Command Query Responsibility Segregation):
Python
class CommandSide:
"""Handle commands by producing events."""
async def handle_command(self, command: Command) -> list[DomainEvent]:
aggregate = await self.repo.load(command.aggregate_id)
events = aggregate.handle(command)
await self.event_store.append(events)
return events
class QuerySide:
"""Handle queries from read-optimized projections."""
async def handle_query(self, query: Query) -> QueryResult:
return await self.read_store.query(
entity_type=query.entity_type,
filters=query.filters,
sort=query.sort,
page=query.page,
)
class ProjectionBuilder:
"""Build read-optimized projections from events."""
async def process_event(self, event: DomainEvent) -> None:
handler = self.handlers.get(event.event_type)
if handler:
await handler(event)
async def _handle_object_created(self, event: DomainEvent) -> None:
await self.read_store.upsert(
entity_type=event.aggregate_type,
entity_id=event.aggregate_id,
data=event.payload,
version=event.sequence_number,
)
#Part 9: Real-World Case Study — Loan Approval Full-Chain Traceability
#9.1 Scenario
A regulator requires a bank to explain the complete decision process for a loan rejection 3 years ago.
#9.2 Traceability Process
Python
# 1. Load the loan's complete event history
events = await audit_service.get_audit_trail(
entity_id="loan-2023-08-15-0042",
start_time=datetime(2023, 8, 14),
end_time=datetime(2023, 8, 16),
)
# 2. Reconstruct state at decision time
state_at_decision = await audit_service.reconstruct_state_at(
entity_id="loan-2023-08-15-0042",
point_in_time=datetime(2023, 8, 15, 14, 30),
)
# 3. Examine the rule and model versions used
# → rule_version: "credit-risk-v2.3.1"
# → model_version: "xgboost-credit-2023Q3"
# 4. Replay with identical rule and model versions
replay = await what_if.analyze(
entity_id="loan-2023-08-15-0042",
modifications=[], # No modifications, pure replay verification
)
# 5. Confirm consistent results → decision is explainable and reproducible
assert replay.original_outcome == replay.modified_outcome
#Part 10: Summary
#Key Takeaways
- Events are facts, state is derived — events are immutable and undeletable; current state is derived by replaying events.
- Audit is built in — event-sourced systems do not need separate audit logs; the event stream itself is the complete audit trail.
- Lineage tracking is a natural extension of causation chains — through causation_id and correlation_id, the data lineage graph builds itself.
- Time travel is the killer feature — combined with Iceberg and Nessie, you can query data and schemas at any point in time.
- CQRS is a necessary companion — event sourcing read performance requires materialized projections for optimization.
- Snapshots and compaction ensure long-term viability — event sourcing without optimization degrades at scale.
#References
- Martin Fowler, Event Sourcing↗
- Greg Young, CQRS and Event Sourcing↗
- Apache Iceberg Time Travel↗
- Project Nessie↗
- coomia-dip Architecture Overview
tags: event-sourcing audit lineage replay cqrs coomia-dip
Next article: S10-04 Saga Pattern in Practice