Versionado y Evolución de Eventos
Compatibilidad hacia atrás, upcasting y estrategias de versión.
Los domain events e integration events almacenados o enviados externamente deben poder evolucionar sin romper consumidores existentes ni el replay de aggregates. La evolución del esquema es uno de los problemas más difíciles en sistemas event-driven: los eventos en el store son inmutables, pero el código que los interpreta sí cambia. La tensión entre inmutabilidad del pasado y evolución del código es el núcleo del problema.
Las estrategias de cambio se dividen en dos categorías. Los cambios aditivos — añadir campos opcionales con valor por defecto — son compatibles hacia atrás y generalmente no requieren versionado: los consumidores existentes ignoran el campo nuevo, los nuevos consumidores lo usan. Los cambios rupturistas — renombrar campos, eliminarlos, añadir campos obligatorios, cambiar el tipo de un campo — requieren versionado explícito y un mecanismo de transformación.
El upcasting es la técnica estándar: una función que transforma el payload de un evento antiguo a la forma actual antes de que el aggregate o handler lo procese. El upcasting se aplica en tiempo de lectura, no en tiempo de escritura — los eventos almacenados permanecen inmutables. Al cargar eventos del store, cada evento pasa por la cadena de upcasters antes de ser deserializado.
Las estrategias de versionado incluyen: tipos de evento versionados (SomethingHappenedV1 → SomethingHappenedV2), versionado en el payload (campo version dentro del evento), y versionado por topic/stream (topics separados en Kafka por versión). Cada estrategia tiene tradeoffs de legibilidad, flexibilidad y coste de migración.
Cuándo simplificar: no todo cambio necesita una nueva versión. Los cambios aditivos con defaults sensatos no requieren upcasters. El over-versioning — crear V2, V3, V4 para cambios menores — crea una cadena de upcasters que nadie recuerda y que aumenta el coste de mantenimiento sin beneficio real. Versionar solo cuando el cambio es genuinamente incompatible hacia atrás.
# infrastructure/upcasters/<event>_upcaster.py — Upcaster
# infrastructure/event_store/<event_store>.py — Applies upcasters on load
@dataclass(frozen=True)
class <SomethingHappenedV1>:
aggregate_id: str
<old_field>: str
@dataclass(frozen=True)
class <SomethingHappenedV2>:
aggregate_id: str
<new_field_a>: str
<new_field_b>: str # New in V2; V1 gets a default
class <SomethingHappened>Upcaster:
def upcast(self, stored_event: StoredEvent) -> StoredEvent:
if stored_event.event_type == '<SomethingHappenedV1>':
payload = dict(stored_event.payload)
# V1 had <old_field>; V2 splits it into <new_field_a> and <new_field_b>
payload['<new_field_a>'] = payload.pop('<old_field>')
payload['<new_field_b>'] = '<default_value>'
return StoredEvent(
aggregate_id=stored_event.aggregate_id,
sequence=stored_event.sequence,
event_type='<SomethingHappenedV2>',
payload=payload,
occurred_at=stored_event.occurred_at,
)
return stored_event # Already current version
class <EventStore>:
def load(self, aggregate_id: str) -> list[<DomainEvent>]:
raw_events = self._db.query_events(aggregate_id)
# Apply upcasters at read time — events stay immutable in storage
upcasted = [self._upcaster.upcast(e) for e in raw_events]
return [self._deserialize(e) for e in upcasted]Debugging lab
Detecta y corrige el error o la violación de diseño.
- 7.6.5.1
# Bug: updating stored event payload directly in the DB to rename a field class EventMigration: def migrate_rename_field(self) -> None: # Bug: directly modifying stored events in the database self._db.execute(""" UPDATE events SET payload = jsonb_set(payload, '{new_field_name}', payload->'old_field_name') - 'old_field_name' WHERE event_type = 'SomethingHappenedV1' """)
- 7.6.5.2
# Bug: integration event consumer with hardcoded field access on a V2 event that no longer has that field class SomethingHappenedIntegrationHandler: def handle(self, event: dict) -> None: # Bug: hardcoded access to old_field that was renamed in V2 target_id = event["old_field"] # KeyError on V2 events value = event["combined_value"] # Also gone in V2
- 7.6.5.3
# Bug: new mandatory field <required_value> added to existing event type without upcaster @dataclass(frozen=True) class SomethingHappenedV2: aggregate_id: str original_field: str required_value: str # New mandatory field — V1 events don't have this # No upcaster written — loading V1 events from store will raise TypeError/KeyError
- 7.6.5.4
# Bug: upcaster modifying the original stored_event object in place class SomethingHappenedUpcaster: def upcast(self, stored_event: StoredEvent) -> StoredEvent: if stored_event.event_type == "SomethingHappenedV1": # Bug: mutating the original object in place stored_event.payload["new_field"] = stored_event.payload.pop("old_field") stored_event.event_type = "SomethingHappenedV2" return stored_event
- 7.6.5.5
# Bug: every minor additive field change creating a new versioned event type @dataclass(frozen=True) class SomethingHappenedV2: # Added optional display_name field aggregate_id: str value: str display_name: Optional[str] = None @dataclass(frozen=True) class SomethingHappenedV3: # Added optional metadata field aggregate_id: str value: str display_name: Optional[str] = None metadata: Optional[dict] = None # Upcasters V1->V2->V3 written for changes that didn't need any versioning