Durable World/Entity Persistence — Final Design¶
Version: 3.1
Status: Final
Author: Systems Architecture Team
Date: 2025-07-18
Priority: P0 — Critical Path
1. Executive Summary¶
MAID currently loses all world state on server restart. While player characters survive through the CharacterManager persistence in maid-classic-rpg, NPCs, items, room modifications, bank balances, crafting progress, quest state, and builder-created content are all volatile. Every restart resets the world to the content pack's on_load() baseline.
This document presents the final design for a Durable Entity Persistence layer that integrates with MAID's existing DocumentStore infrastructure. The design introduces:
- ComponentRegistry: Type-safe deserialization mapping with pack ownership tracking and overwrite protection
- DirtyTracker: Event-driven change detection with automatic dirty marking for component mutations
- EntityPersistenceManager: Batched async saves with per-operation timeouts, tombstone queue for reliable deletes, and independent per-entity result aggregation
- SaveScheduler: asyncio.Lock-based coordination with atomic drain-then-re-mark strategy
- Deterministic startup flow: Load-then-supplement pattern that preserves existing content pack behavior
- Schema evolution support: Explicit key-set pre-filtering and registered migration functions (v1→v2→v3)
- Corruption detection: SHA-256 entity checksums verified on load; quarantine for unresolvable components
- Comprehensive backup/restore: Full snapshots with snapshot_id manifests, incremental backups, and point-in-time recovery
The implementation is phased across four milestones spanning 10 weeks. Critical prerequisites identified in design review include adding event emission to the ECS core, extending DocumentStore with upsert capability and paginated queries, adding a __setattr__ hook to the Component base class (with _suppress_dirty flag and observable container wrappers for mutable-container mutations), extending the ContentPack protocol with register_component_types(), and migrating the documents table to a composite primary key — these are Phase 0 tasks that must complete before persistence implementation begins.
Key Design Decisions¶
- Event-driven dirty tracking: Leverages existing
EventBusinfrastructure after adding missing event emissions - Upsert-based saves: Eliminates race conditions via
INSERT ... ON CONFLICT DO UPDATEin PostgreSQL - Component-centric serialization: Uses
component.get_type()consistently for registry keys - Separate entity/room collections: Maintains clear separation between entities and room state
- Atomic drain with failure re-marking: Dirty set is drained before save; failed entity IDs are re-marked (§5.1)
- asyncio.Lock for save coordination: Eliminates TOCTOU race in concurrent save triggers (§5.6)
- Explicit schema pre-filtering: Compares field keys before Pydantic validation instead of catching exceptions (§10.8)
- Quarantine over silent drop: Unresolvable components are preserved as raw data for round-trip safety (§5.2)
- Entity checksums: SHA-256 integrity verification covering components and entity metadata on save/load (§5.4)
- Observable containers +
notify_mutation(): Addresses the__setattr__container-mutation gap forlist/dict/setfields (§5.8.1) _suppress_dirtyduring deserialization: Prevents false-positive dirty marks from Pydantic field assignment duringmodel_validate()(§5.8)- Save window between ticks: Serialization/checksum computation runs during a quiescent point between ticks, with main-thread snapshot before executor handoff (§5.9)
- Paginated entity loading: Bounded-memory streaming via
limit/offsetqueries instead of loading all documents at once (§4.6) - Crash-safe tombstones: Tombstone queue persisted to
persistence_metacollection to survive crashes (§5.3)
2. Problem Statement & Current State¶
2.1 Core Problems¶
Data Loss on Restart: Every server restart resets the world to baseline content pack state. Hours of player progress, builder work, and world evolution are lost.
Inconsistent Persistence: Different content packs implement ad-hoc persistence (CharacterManager stores characters, WorldPersistenceHelper saves grid state), creating maintenance burden and gaps.
No Change Detection: Without dirty tracking, full-world saves would require O(N) writes per cycle, making frequent saves prohibitively expensive.
Missing Deserialization: Entity.to_dict() exists but no Entity.from_dict(), preventing entity reconstruction from stored data.
2.2 Existing Infrastructure Analysis¶
| Component | File | Capability | Limitation |
|---|---|---|---|
| Entity serialization | core/ecs/entity.py:202-213 |
Entity.to_dict() via Pydantic model_dump() |
No deserialization path |
| Component base | core/ecs/component.py |
Pydantic BaseModel with component_type ClassVar |
No registry for string→class mapping |
| DocumentStore | storage/document_store.py |
CRUD operations on JSONB collections | No upsert method |
| Entity events | core/events.py:103-129 |
Defined but never emitted by ECS operations | DirtyTracker cannot function |
| ContentPack protocol | plugins/protocol.py |
on_load() creates entities every startup |
No "load if persisted" pattern |
2.3 Prerequisites Identified in Review¶
Before persistence implementation can begin, two blocking issues must be resolved:
P0-A: Add Event Emission to ECS Core
- Entity.add() must emit ComponentAddedEvent
- Entity.remove() must emit ComponentRemovedEvent
- EntityManager.create() must emit EntityCreatedEvent
- Entity destruction must emit EntityDestroyedEvent
P0-B: Add Upsert to DocumentStore
- DocumentCollection.upsert(id, data) using INSERT ... ON CONFLICT DO UPDATE
- Required for atomic save operations without race conditions
P0-C: Add __setattr__ Hook to Component Base Class
- Add _dirty_callback, _owner_id, and _suppress_dirty private attributes to Component base class
- Implement __setattr__ override for automatic dirty notification (see §5.8)
- Add notify_mutation() convenience method for container-mutation scenarios (see §5.8.1)
- Without this, every system must manually call mark_dirty(), which is error-prone
P0-D: Extend ContentPack Protocol with register_component_types()
- Add register_component_types(registry: ComponentRegistry) -> None to the ContentPack protocol
- Content packs must register their component classes before persistence can serialize/deserialize them
- Called during GameEngine.load_content_pack(), before on_load()
P0-E: Add Pagination Support to DocumentStore Queries
- DocumentCollection.query() must accept limit and offset (or cursor) parameters
- Required for loading 10 K+ entities without unbounded memory allocation (see load_all_entities)
- Also needed for batched delete_pack_entities()
P0-F: Composite Primary Key on Documents Table
- The documents table must use PRIMARY KEY (collection, id) — the current schema has id as sole PK, but document IDs are only unique within a collection
- Required for ON CONFLICT (collection, id) in the upsert path (see §4.4)
3. Architecture Overview¶
3.1 High-Level Design¶
┌──────────────────────────────────────────────────────────────────────┐
│ GameEngine │
│ ┌─────────┐ ┌──────────┐ ┌──────────────────────────────────────┐ │
│ │ World │ │ EventBus │ │ ContentPack(s) │ │
│ │ │ │ │ │ register_component_types() │ │
│ └─────────┘ └──────────┘ └──────────────────────────────────────┘ │
│ │ │ │ │
│ │ │ │ │
│ ┌────▼─────────────▼──────────────────────────▼──────────────────┐ │
│ │ EntityPersistenceManager │ │
│ │ │ │
│ │ ┌────────────┐ ┌──────────┐ ┌─────────┐ ┌───────────────────┐ │ │
│ │ │ComponentR. │ │DirtyTrack│ │SaveSched│ │EntitySerializer │ │ │
│ │ │ │ │er │ │uler │ │ │ │ │
│ │ │string→class│ │EventSub │ │Async │ │dict↔Entity │ │ │
│ │ │mapping │ │dirty_set │ │batching │ │via registry │ │ │
│ │ └────────────┘ └──────────┘ └─────────┘ └───────────────────┘ │ │
│ └─────────────────────────────┬───────────────────────────────────┘ │
└────────────────────────────────┼────────────────────────────────────────┘
│
┌────────────────────────────▼────────────────────────────┐
│ DocumentStore │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Collection │ │ Collection │ │ Collection │ │
│ │ "entities" │ │ "rooms" │ │ "persistence_meta" │ │
│ │ │ │ │ │ │ │
│ │EntityDoc[] │ │RoomState[] │ │SaveEvent[] │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└─────────────────────────────────────────────────────────┘
3.2 Component Interaction Flow¶
Save Cycle:
1. Component mutation → __setattr__ hook / notify_mutation() on Component (§5.8) → DirtyTracker.mark_dirty() marks entity dirty
2. SaveScheduler triggers → DirtyTracker.drain() → snapshot component data on main thread (§5.9)
3. EntitySerializer.serialize() (off-thread for large batches) → ComponentRegistry.resolve() for type validation
4. DocumentCollection.upsert() to persist changes
Load Flow (Startup):
1. GameEngine starts → _load_tombstones() restores pending deletes → Load persisted entities in pages via paginated query() → Reconstruct via EntitySerializer.deserialize()
2. ComponentRegistry.resolve(type_string) → Python class → Pydantic validation (with _suppress_dirty = True)
3. ContentPack.on_load() → Check entity existence (idempotency guard) → Create only missing defaults
4. Validate migration chain completeness for all registered component types
5. Subscribe DirtyTracker → Begin normal operation
4. Detailed Design¶
4.1 Entity Serialization/Deserialization¶
EntityDocument Schema¶
# packages/maid-engine/src/maid_engine/persistence/models.py
from datetime import UTC, datetime
from typing import Any
from uuid import UUID
from pydantic import BaseModel, Field
class ComponentData(BaseModel):
"""Serialized component within an entity document."""
component_type: str # Result of component.get_type()
schema_version: int = 1
data: dict[str, Any] # Component.model_dump()
class QuarantinedComponent(BaseModel):
"""Raw data for components whose type is not currently registered.
Preserves round-trip safety when a content pack is unloaded: its
component data is kept verbatim so it can be restored if the pack
is re-loaded later, rather than being silently discarded.
"""
component_type: str
schema_version: int
raw_data: dict[str, Any]
class QuarantineComponent(Component):
"""ECS Component that stores quarantined data on an Entity.
Because Entity uses ``__slots__``, we cannot monkey-patch arbitrary
attributes onto it. Instead, quarantined component data is stored
as a first-class component that the serializer knows to preserve.
"""
component_type: ClassVar[str] = "persistence:quarantine"
items: list[QuarantinedComponent] = Field(default_factory=list)
class EntityDocument(BaseModel):
"""Persistence model for a single ECS entity.
Stored in DocumentStore collection "entities".
"""
entity_id: UUID
world_id: str = "default"
tags: set[str] = Field(default_factory=set) # Use set to avoid duplicates
components: list[ComponentData]
quarantined_components: list[QuarantinedComponent] = Field(default_factory=list)
# Integrity
checksum: str = "" # SHA-256 of serialized components (see §5.4)
snapshot_id: str = "" # Save-cycle identifier for transactional grouping
# Metadata
source_pack: str = "" # Content pack that created this entity
created_at: datetime # Copy from entity.created_at property
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
model_config = ConfigDict(
json_encoders={
UUID: str,
datetime: lambda dt: dt.isoformat(),
set: list, # Convert set to list for JSON serialization
}
)
class RoomStateDocument(BaseModel):
"""Room-to-entity mapping for spatial queries.
**Note:** This model is defined for future use by room-level
persistence (e.g., persisting which entities are in each room for
fast spatial queries on load). It is not yet consumed by any
persistence code path. Integration is planned for Phase 3.
"""
room_id: UUID
world_id: str = "default"
entity_ids: set[UUID] = Field(default_factory=set)
updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
EntitySerializer Implementation¶
# packages/maid-engine/src/maid_engine/persistence/serializer.py
from typing import TYPE_CHECKING, Any
from uuid import UUID
if TYPE_CHECKING:
from maid_engine.core.ecs import Entity
from maid_engine.persistence.registry import ComponentRegistry
from .models import ComponentData, EntityDocument
class EntitySerializer:
"""Handles entity serialization/deserialization with ComponentRegistry."""
def __init__(self, registry: "ComponentRegistry") -> None:
self.registry = registry
def serialize(self, entity: "Entity") -> EntityDocument:
"""Convert ECS Entity to EntityDocument for persistence."""
components = []
for component in entity.components: # entity.components yields Iterator[Component]
# Skip the persistence-internal QuarantineComponent; it is
# handled separately below.
if component.get_type() == "persistence:quarantine":
continue
# Use get_type() consistently for registry keys
component_type = component.get_type()
# Verify component is registered (fail fast on missing types)
if not self.registry.is_registered(component_type):
raise ValueError(f"Component type '{component_type}' not registered for serialization")
component_data = ComponentData(
component_type=component_type,
schema_version=self.registry.get_schema_version(component_type),
data=component.model_dump()
)
components.append(component_data)
return EntityDocument(
entity_id=entity.id,
tags=entity.tags, # Convert set to set (validation ensures set type)
components=components,
created_at=entity.created_at, # Use public property (not _created_at)
# source_pack filled by persistence manager based on context
)
def deserialize(self, doc: EntityDocument, world: "World") -> "Entity":
"""Reconstruct ECS Entity from EntityDocument.
Verifies entity checksum on load when present (see §5.4).
Unknown component types are quarantined, not dropped.
"""
from maid_engine.core.ecs import Entity
# Integrity check: verify checksum if present (includes quarantined)
if doc.checksum:
actual = self._compute_checksum(
doc.components, doc.quarantined_components,
tags=doc.tags, world_id=doc.world_id,
)
if actual != doc.checksum:
raise IntegrityError(
f"Entity {doc.entity_id} checksum mismatch "
f"(expected {doc.checksum[:12]}…, got {actual[:12]}…)"
)
# Create entity with preserved ID via post-construction setup.
# Entity.__init__ only accepts (entity_id, manager), so tags and
# created_at must be set after construction.
entity = Entity(entity_id=doc.entity_id)
for tag in doc.tags:
entity.add_tag(tag)
entity._created_at = doc.created_at # Restore original creation time
# Carry forward any previously quarantined components
quarantined: list["QuarantinedComponent"] = list(doc.quarantined_components)
# Deserialize components with schema evolution support
for comp_data in doc.components:
component = self._deserialize_component(comp_data)
if component is not None:
entity.add(component)
else:
# Quarantine unresolvable components for round-trip safety
quarantined.append(QuarantinedComponent(
component_type=comp_data.component_type,
schema_version=comp_data.schema_version,
raw_data=comp_data.data,
))
# Attach quarantined data via a QuarantineComponent so it survives
# re-serialization. We cannot monkey-patch a bare attribute onto
# Entity because Entity uses __slots__.
if quarantined:
from maid_engine.persistence.models import QuarantineComponent
entity.add(QuarantineComponent(items=quarantined))
return entity
def _deserialize_component(self, comp_data: ComponentData) -> Any | None:
"""Deserialize single component with explicit schema pre-filtering.
Instead of catching validation exceptions and inspecting error
strings, we compare ``model_fields.keys()`` against the stored
data keys *before* validation. This is both more reliable and
more efficient than the previous try/except string-matching
approach.
"""
component_type = comp_data.component_type
component_class = self.registry.resolve(component_type)
if component_class is None:
# Component type not registered (pack uninstalled) — caller
# will quarantine this data.
return None
data = comp_data.data
expected_keys = set(component_class.model_fields.keys())
actual_keys = set(data.keys())
# Pre-filter: strip fields removed from the current schema
extra_keys = actual_keys - expected_keys
if extra_keys:
logger.warning(
"Stripping %d unknown field(s) from %s: %s",
len(extra_keys), component_type, extra_keys,
)
data = {k: v for k, v in data.items() if k in expected_keys}
# Check for migration hooks before plain validation.
# If the migration chain has a gap, migrate() raises
# MigrationGapError and the caller quarantines the component.
stored_version = comp_data.schema_version
current_version = self.registry.get_schema_version(component_type)
if stored_version < current_version:
try:
data = self.registry.migrate(component_type, data, stored_version)
except MigrationGapError:
return None # Caller will quarantine
return component_class.model_validate(data)
@staticmethod
def _compute_checksum(
components: list[ComponentData],
quarantined: list["QuarantinedComponent"] | None = None,
*,
tags: set[str] | None = None,
world_id: str = "",
) -> str:
"""Deterministic SHA-256 over entity metadata and component data.
Covers entity metadata (tags, world_id) in addition to
component and quarantined data, so tag mutations and world
reassignments are also detected as corruption.
Uses ``model_dump_json()`` instead of ``json.dumps(model_dump())``
for deterministic field ordering — Pydantic serializes fields in
model-definition order, removing dependence on Python dict
iteration stability.
Note: Uses a plain SHA-256 hash. For environments requiring
tamper resistance (e.g., untrusted storage backends), consider
replacing with HMAC-SHA-256 keyed by a server secret.
"""
import hashlib
# Include entity metadata in the checksum
meta = f"world_id={world_id};tags={','.join(sorted(tags or set()))}"
# Use model_dump_json() for stable, deterministic field ordering
parts: list[str] = [
c.model_dump_json()
for c in sorted(components, key=lambda c: c.component_type)
]
if quarantined:
parts.extend(
q.model_dump_json()
for q in sorted(quarantined, key=lambda q: q.component_type)
)
canonical = meta + "|" + "|".join(parts)
return hashlib.sha256(canonical.encode()).hexdigest()
4.2 Component Registry¶
# packages/maid-engine/src/maid_engine/persistence/registry.py
from typing import Any, Dict, Type
from maid_engine.core.ecs import Component
class ComponentRegistry:
"""Registry mapping component type strings to Python classes.
Enables type-safe deserialization of persisted entities.
Content packs register their component types during load.
By default, re-registration of an already-registered type raises
``ComponentConflictError`` to prevent accidental component hijacking.
Pass ``allow_overwrite=True`` only during hot-reload scenarios.
"""
def __init__(self) -> None:
self._registry: Dict[str, Type[Component]] = {}
self._schema_versions: Dict[str, int] = {}
self._aliases: Dict[str, str] = {} # Support for renamed components
self._pack_ownership: Dict[str, str] = {} # component_type → pack_name
self._migrations: Dict[str, list[tuple[int, int, Callable]]] = {} # type → [(from, to, fn)]
def register(
self,
component_class: Type[Component],
schema_version: int = 1,
aliases: list[str] | None = None,
pack_name: str = "",
allow_overwrite: bool = False,
) -> None:
"""Register a component type for serialization/deserialization.
Args:
component_class: The component class to register
schema_version: Schema version for migration tracking
aliases: Alternative names for backward compatibility
pack_name: Owning content pack name (used by unregister_pack)
allow_overwrite: If False, raise on duplicate registration
Raises:
ComponentConflictError: If type already registered and
allow_overwrite is False.
"""
component_type = component_class.get_type()
if component_type in self._registry and not allow_overwrite:
raise ComponentConflictError(
f"Component type '{component_type}' is already registered "
f"by pack '{self._pack_ownership.get(component_type, '<unknown>')}'. "
f"Pass allow_overwrite=True for hot-reload."
)
self._registry[component_type] = component_class
self._schema_versions[component_type] = schema_version
if pack_name:
self._pack_ownership[component_type] = pack_name
# Register aliases for renamed components
for alias in aliases or []:
self._aliases[alias] = component_type
def register_migration(
self,
component_type: str,
from_version: int,
to_version: int,
migrate_fn: Callable[[dict[str, Any]], dict[str, Any]],
) -> None:
"""Register a schema migration function.
Migration functions are chained: v1→v2→v3. Each function
receives the component data dict and returns the transformed dict.
"""
self._migrations.setdefault(component_type, []).append(
(from_version, to_version, migrate_fn)
)
# Keep sorted by from_version for sequential application
self._migrations[component_type].sort(key=lambda m: m[0])
def migrate(
self, component_type: str, data: dict[str, Any], from_version: int
) -> dict[str, Any]:
"""Apply chained migrations from from_version to current version."""
target = self._schema_versions.get(component_type, 1)
current = from_version
for src, dst, fn in self._migrations.get(component_type, []):
if src == current and dst <= target:
data = fn(data)
current = dst
if current != target:
logger.warning(
"Incomplete migration for %s: reached v%d, target v%d — "
"component will be quarantined",
component_type, current, target,
)
raise MigrationGapError(
f"Cannot migrate {component_type} from v{current} to v{target}: "
f"no registered migration for v{current}→v{current + 1}"
)
return data
def unregister_pack(self, pack_name: str) -> None:
"""Remove all components registered by a content pack.
Uses the ``_pack_ownership`` mapping populated at registration
time, rather than inspecting class attributes.
"""
to_remove = [
ctype for ctype, owner in self._pack_ownership.items()
if owner == pack_name
]
for component_type in to_remove:
del self._registry[component_type]
del self._schema_versions[component_type]
del self._pack_ownership[component_type]
# Clean up any migrations for this type
self._migrations.pop(component_type, None)
# Remove aliases that pointed to removed types
alias_removals = [
alias for alias, target in self._aliases.items()
if target in to_remove
]
for alias in alias_removals:
del self._aliases[alias]
def resolve(self, component_type: str) -> Type[Component] | None:
"""Get component class from type string.
Returns None if component type not registered (graceful degradation).
"""
# Check aliases first for renamed components
actual_type = self._aliases.get(component_type, component_type)
return self._registry.get(actual_type)
def is_registered(self, component_type: str) -> bool:
"""Check if a component type is registered."""
return component_type in self._registry or component_type in self._aliases
def get_schema_version(self, component_type: str) -> int:
"""Get schema version for a component type."""
return self._schema_versions.get(component_type, 1)
def get_registered_types(self) -> list[str]:
"""Get all registered component type names."""
return list(self._registry.keys())
4.3 Dirty Tracking Mechanism¶
# packages/maid-engine/src/maid_engine/persistence/dirty_tracker.py
from typing import Set
from uuid import UUID
from maid_engine.core.events import (
ComponentAddedEvent,
ComponentRemovedEvent,
EntityCreatedEvent,
EntityDestroyedEvent,
EventBus
)
class DirtyTracker:
"""Tracks entities that have changed since last save.
Subscribes to ECS events to automatically detect mutations.
Provides atomic drain operation for save cycles.
"""
def __init__(self, event_bus: EventBus) -> None:
self.event_bus = event_bus
self._dirty_entities: Set[UUID] = set()
self._destroyed_entities: Set[UUID] = set()
self._subscribed = False
def start_tracking(self) -> None:
"""Begin tracking entity changes via event subscriptions."""
if self._subscribed:
return
# Subscribe to ECS events for automatic dirty detection
self.event_bus.subscribe(EntityCreatedEvent, self._on_entity_created)
self.event_bus.subscribe(EntityDestroyedEvent, self._on_entity_destroyed)
self.event_bus.subscribe(ComponentAddedEvent, self._on_component_added)
self.event_bus.subscribe(ComponentRemovedEvent, self._on_component_removed)
self._subscribed = True
def stop_tracking(self) -> None:
"""Stop tracking (cleanup on shutdown)."""
if not self._subscribed:
return
self.event_bus.unsubscribe(EntityCreatedEvent, self._on_entity_created)
self.event_bus.unsubscribe(EntityDestroyedEvent, self._on_entity_destroyed)
self.event_bus.unsubscribe(ComponentAddedEvent, self._on_component_added)
self.event_bus.unsubscribe(ComponentRemovedEvent, self._on_component_removed)
self._subscribed = False
def mark_dirty(self, entity_id: UUID) -> None:
"""Manually mark an entity as dirty.
Used by systems that perform in-place component mutations
(e.g., health.current -= 10) that don't trigger events, and
by the ``__setattr__`` hook / ``notify_mutation()`` callback.
If the dirty set exceeds ``MAX_DIRTY_SIZE``, a warning is
logged. The set is not capped — dropping dirty IDs would
cause silent data loss — but the warning provides backpressure
visibility so operators can tune save intervals.
"""
MAX_DIRTY_SIZE = 50_000
if entity_id not in self._destroyed_entities:
self._dirty_entities.add(entity_id)
if len(self._dirty_entities) > MAX_DIRTY_SIZE:
logger.warning(
"Dirty set size %d exceeds %d — consider reducing save_interval",
len(self._dirty_entities), MAX_DIRTY_SIZE,
)
def mark_clean(self, entity_id: UUID) -> None:
"""Mark entity as clean (used after successful save)."""
self._dirty_entities.discard(entity_id)
def is_dirty(self, entity_id: UUID) -> bool:
"""Check if entity is dirty."""
return entity_id in self._dirty_entities
def get_dirty_count(self) -> int:
"""Get number of dirty entities (for metrics)."""
return len(self._dirty_entities)
def drain_dirty(self) -> tuple[Set[UUID], Set[UUID]]:
"""Atomically get and clear dirty/destroyed entity sets.
Returns:
Tuple of (dirty_entity_ids, destroyed_entity_ids)
Note: This method clears the internal sets. If save fails,
the caller must re-mark entities as dirty.
"""
dirty = self._dirty_entities.copy()
destroyed = self._destroyed_entities.copy()
self._dirty_entities.clear()
self._destroyed_entities.clear()
return dirty, destroyed
def peek_dirty(self) -> tuple[Set[UUID], Set[UUID]]:
"""Get dirty/destroyed sets without clearing (for inspection)."""
return self._dirty_entities.copy(), self._destroyed_entities.copy()
# Event handlers for automatic dirty tracking.
# Note: EventBus.subscribe() auto-wraps sync handlers in an async
# wrapper, so these methods are intentionally synchronous.
def _on_entity_created(self, event: EntityCreatedEvent) -> None:
"""Handle entity creation."""
self._dirty_entities.add(event.entity_id)
def _on_entity_destroyed(self, event: EntityDestroyedEvent) -> None:
"""Handle entity destruction."""
self._dirty_entities.discard(event.entity_id)
self._destroyed_entities.add(event.entity_id)
def _on_component_added(self, event: ComponentAddedEvent) -> None:
"""Handle component addition."""
self._dirty_entities.add(event.entity_id)
def _on_component_removed(self, event: ComponentRemovedEvent) -> None:
"""Handle component removal."""
self._dirty_entities.add(event.entity_id)
4.4 DocumentStore Extension with Upsert¶
# packages/maid-engine/src/maid_engine/storage/document_store.py (additions)
class DocumentCollection(Generic[T]):
"""Abstract collection interface with upsert support."""
async def upsert(self, id: UUID, data: T) -> bool:
"""Insert or update document atomically.
Args:
id: Document identifier (UUID, consistent with get/create/update)
data: Typed document model to store
Returns:
True if successful, False if failed
"""
raise NotImplementedError
class PostgresDocumentCollection(DocumentCollection[T]):
"""PostgreSQL implementation with UPSERT support."""
async def upsert(self, id: UUID, data: T) -> bool:
"""Atomic upsert using INSERT ... ON CONFLICT DO UPDATE.
Requires a composite UNIQUE constraint on (collection, id)::
ALTER TABLE documents
DROP CONSTRAINT IF EXISTS documents_pkey,
ADD PRIMARY KEY (collection, id);
The default schema's single-column ``id`` PRIMARY KEY is
insufficient because ``id`` values are only unique *within*
a collection, not across all collections.
"""
query = """
INSERT INTO documents (id, collection, data, created_at, updated_at)
VALUES ($1, $2, $3, NOW(), NOW())
ON CONFLICT (collection, id)
DO UPDATE SET
data = EXCLUDED.data,
updated_at = NOW()
"""
try:
json_data = json.dumps(data.model_dump(), cls=DocumentEncoder)
await asyncio.wait_for(
self.db.execute(query, str(id), self.collection_name, json_data),
timeout=5.0, # Per-upsert timeout
)
return True
except asyncio.TimeoutError:
logger.error(f"Upsert timed out for {self.collection_name}.{id}")
return False
except Exception as e:
logger.error(f"Upsert failed for {self.collection_name}.{id}: {e}")
return False
class InMemoryDocumentCollection(DocumentCollection[T]):
"""In-memory implementation for testing."""
async def upsert(self, id: UUID, data: T) -> bool:
"""Upsert for in-memory collection."""
try:
now = datetime.now(UTC)
str_id = str(id)
self.documents[str_id] = {
'id': str_id,
'data': data.model_dump(),
'created_at': self.documents.get(str_id, {}).get('created_at', now),
'updated_at': now
}
return True
except Exception as e:
logger.error(f"In-memory upsert failed for {id}: {e}")
return False
4.5 Save Scheduling with Crash Recovery¶
# packages/maid-engine/src/maid_engine/persistence/save_scheduler.py
import asyncio
from datetime import UTC, datetime
from typing import Set
from uuid import UUID, uuid4
from maid_engine.core.events import Event, EventBus
from maid_engine.persistence.dirty_tracker import DirtyTracker
from maid_engine.persistence.persistence_manager import EntityPersistenceManager
class WorldSaveStartedEvent(Event):
"""Emitted when a save cycle begins."""
started_at: datetime
snapshot_id: str
class WorldSaveCompletedEvent(Event):
"""Emitted when a save cycle completes (success or failure)."""
started_at: datetime
completed_at: datetime
duration: float
entities_saved: int
entities_failed: int
success: bool
snapshot_id: str
class SaveScheduler:
"""Manages periodic and manual entity saves with crash recovery.
Handles:
- Periodic save cycles (configurable interval)
- Manual save triggers (@save command)
- Crash recovery via dirty tracking
- Save operation coordination via asyncio.Lock
"""
CYCLE_TIMEOUT: float = 60.0 # Max duration of a single save cycle
def __init__(
self,
persistence_manager: EntityPersistenceManager,
dirty_tracker: DirtyTracker,
event_bus: EventBus,
save_interval: float = 300.0 # 5 minutes default
) -> None:
self.persistence_manager = persistence_manager
self.dirty_tracker = dirty_tracker
self.event_bus = event_bus
self.save_interval = save_interval
self._save_task: asyncio.Task | None = None
self._save_lock = asyncio.Lock() # Replaces boolean flag (TOCTOU-safe)
self._last_save_time: datetime | None = None
self._shutdown = False
def start(self) -> None:
"""Start periodic save scheduling."""
if self._save_task is None or self._save_task.done():
self._save_task = asyncio.create_task(self._save_loop())
async def stop(self) -> None:
"""Stop scheduling and perform final save."""
self._shutdown = True
if self._save_task and not self._save_task.done():
self._save_task.cancel()
try:
await self._save_task
except asyncio.CancelledError:
pass
# Final save on shutdown
await self.trigger_save()
async def trigger_save(self) -> bool:
"""Manually trigger a save cycle.
Returns:
True if save completed successfully, False if failed.
If another save is already in progress, waits for the
lock rather than silently returning.
"""
return await self._perform_save()
async def _save_loop(self) -> None:
"""Main periodic save loop.
The first save fires after a short initial delay (10 s) rather
than waiting the full ``save_interval``, so early mutations
(e.g., content-pack ``on_load`` entity creation) are persisted
promptly.
"""
first_run = True
while not self._shutdown:
try:
delay = 10.0 if first_run else self.save_interval
first_run = False
await asyncio.sleep(delay)
if not self._shutdown:
await self._perform_save()
except asyncio.CancelledError:
break
except Exception as e:
# Log error but continue scheduling
logger.error(f"Save scheduler error: {e}")
await asyncio.sleep(5.0) # Brief pause before retry
async def _perform_save(self) -> bool:
"""Execute a single save cycle with proper async coordination.
Key design choices:
- Uses asyncio.Lock instead of a boolean flag to eliminate
the TOCTOU race between check and set.
- Drains the dirty set atomically *before* persisting, then
re-marks any entities that failed to save. The previous
peek-then-drain approach lost entities that became dirty
between the peek snapshot and the post-save drain.
- Enforces a per-cycle timeout to prevent unbounded saves.
"""
async with self._save_lock:
save_start_time = datetime.now(UTC)
snapshot_id = f"default:{save_start_time.isoformat()}:{uuid4().hex[:8]}"
try:
# Emit save started event for observability
await self.event_bus.emit(WorldSaveStartedEvent(
started_at=save_start_time,
snapshot_id=snapshot_id,
))
# Atomically drain dirty set *before* persisting.
# If the save fails, we re-mark the failed entities below.
dirty_ids, destroyed_ids = self.dirty_tracker.drain_dirty()
if not dirty_ids and not destroyed_ids:
return True # Nothing to save
# Perform the save operation with cycle-level timeout
result = await asyncio.wait_for(
self.persistence_manager.save_entities(
dirty_ids, destroyed_ids, snapshot_id=snapshot_id,
),
timeout=self.CYCLE_TIMEOUT,
)
# Re-mark any entities that failed to save so they are
# retried on the next cycle.
for entity_id in result.failed_entities:
self.dirty_tracker.mark_dirty(entity_id)
if result.success:
self._last_save_time = save_start_time
# Emit completion event
save_end_time = datetime.now(UTC)
duration = (save_end_time - save_start_time).total_seconds()
await self.event_bus.emit(WorldSaveCompletedEvent(
started_at=save_start_time,
completed_at=save_end_time,
duration=duration,
entities_saved=result.entities_saved,
entities_failed=result.entities_failed,
success=result.success,
snapshot_id=snapshot_id,
))
return result.success
except asyncio.TimeoutError:
logger.error(
"Save cycle exceeded %.0fs timeout — re-marking all entities as dirty",
self.CYCLE_TIMEOUT,
)
# Re-mark everything; we don't know what succeeded
for eid in dirty_ids:
self.dirty_tracker.mark_dirty(eid)
# Re-queue destroyed IDs so deletes are retried next cycle;
# without this, drained destroyed_ids are silently lost on
# timeout and the entities reappear as ghosts on restart.
for eid in destroyed_ids:
self.dirty_tracker._destroyed_entities.add(eid)
return False
except Exception as e:
logger.error(f"Save operation failed: {e}")
# Re-mark drained entities so they are retried
for eid in dirty_ids:
self.dirty_tracker.mark_dirty(eid)
# Re-queue destroyed IDs (same rationale as timeout case)
for eid in destroyed_ids:
self.dirty_tracker._destroyed_entities.add(eid)
return False
def get_next_save_time(self) -> datetime | None:
"""Get estimated time of next periodic save."""
if self._last_save_time is None:
return None
next_save = self._last_save_time + timedelta(seconds=self.save_interval)
return next_save
def get_stats(self) -> dict[str, Any]:
"""Get save scheduler statistics for admin interface."""
return {
'save_interval': self.save_interval,
'last_save_time': self._last_save_time.isoformat() if self._last_save_time else None,
'next_save_time': self.get_next_save_time().isoformat() if self.get_next_save_time() else None,
'save_in_progress': self._save_lock.locked(),
'dirty_entity_count': self.dirty_tracker.get_dirty_count()
}
4.6 Entity Persistence Manager¶
# packages/maid-engine/src/maid_engine/persistence/persistence_manager.py
import asyncio
from datetime import UTC, datetime
from typing import Iterator, Protocol, Set
from uuid import UUID
from maid_engine.core.ecs import Entity
from maid_engine.persistence.models import EntityDocument, QuarantineComponent
from maid_engine.persistence.registry import ComponentRegistry
from maid_engine.persistence.serializer import EntitySerializer
from maid_engine.storage import DocumentStore
class EntitySource(Protocol):
"""Protocol decoupling persistence from World for testability.
World satisfies this protocol. Tests can provide a lightweight
stub instead.
"""
def get_entity(self, entity_id: UUID) -> Entity | None: ...
def get_all_entities(self) -> Iterator[Entity]: ...
def add_entity(self, entity: Entity) -> None: ...
def remove_entity(self, entity_id: UUID) -> None: ...
class PersistenceSettings(BaseModel):
"""Configurable persistence tuning knobs.
Sourced from ``MAID_PERSISTENCE__*`` environment variables via the
settings framework (see §8).
"""
upsert_timeout: float = 5.0
batch_timeout: float = 30.0
batch_size: int = 50
save_interval: float = 300.0 # 5 minutes
cycle_timeout: float = 60.0
max_tombstone_retries: int = 10
max_entity_doc_bytes: int = 1_048_576 # 1 MB
class SaveResult:
"""Result of a save operation."""
def __init__(
self,
entities_saved: int = 0,
entities_failed: int = 0,
failed_entities: list[UUID] | None = None,
errors: list[str] | None = None
):
self.entities_saved = entities_saved
self.entities_failed = entities_failed
self.failed_entities = failed_entities or []
self.errors = errors or []
@property
def success(self) -> bool:
"""True if all entities saved successfully."""
return self.entities_failed == 0
@property
def total_entities(self) -> int:
"""Total entities processed."""
return self.entities_saved + self.entities_failed
class LoadResult:
"""Result of a load operation."""
def __init__(
self,
entities_loaded: int = 0,
entities_failed: int = 0,
failed_entities: list[UUID] | None = None,
errors: list[str] | None = None
):
self.entities_loaded = entities_loaded
self.entities_failed = entities_failed
self.failed_entities = failed_entities or []
self.errors = errors or []
@property
def success(self) -> bool:
"""True if all entities loaded successfully."""
return self.entities_failed == 0
class EntityPersistenceManager:
"""Core persistence manager for ECS entities.
Handles:
- Saving entities to DocumentStore
- Loading entities from DocumentStore
- Batch operations with per-operation timeouts
- Tombstone queue for reliable delete retries
- Error handling and recovery
Architecture note: This class accepts an ``EntitySource`` protocol
(satisfied by ``World``) rather than coupling to ``World`` directly,
improving testability. See §5.5 for further decoupling options.
"""
MAX_ENTITY_DOC_BYTES: int = 1_048_576 # 1 MB hard limit per entity
UPSERT_TIMEOUT: float = 5.0 # Per-entity upsert timeout
BATCH_TIMEOUT: float = 30.0 # Per-batch timeout
MAX_TOMBSTONE_RETRIES: int = 10 # Drop tombstones after N failed cycles
def __init__(
self,
entity_source: "EntitySource",
document_store: DocumentStore,
component_registry: ComponentRegistry,
settings: "PersistenceSettings | None" = None,
) -> None:
self.entity_source = entity_source
self.document_store = document_store
self.component_registry = component_registry
self.serializer = EntitySerializer(component_registry)
# Apply settings unconditionally via a defaults instance so that
# every code path reads from the same attributes without the
# instance-attribute-shadows-class-attribute anti-pattern.
s = settings or PersistenceSettings()
self.UPSERT_TIMEOUT = s.upsert_timeout
self.BATCH_TIMEOUT = s.batch_timeout
self.BATCH_SIZE = s.batch_size
self.MAX_TOMBSTONE_RETRIES = s.max_tombstone_retries
self.MAX_ENTITY_DOC_BYTES = s.max_entity_doc_bytes
# Get collections for entities and persistence metadata
self.entity_collection = document_store.get_collection("entities", EntityDocument)
self._meta_collection = document_store.get_collection("persistence_meta")
# Tombstone queue for failed deletes (see §5.3).
# Each entry tracks (entity_id → retry_count). Tombstones are
# persisted to the ``persistence_meta`` collection so they
# survive crashes — an in-memory-only queue would lose pending
# deletes on restart, causing ghost entities to reappear.
self._delete_tombstones: dict[UUID, int] = {}
async def save_entities(
self,
dirty_entity_ids: Set[UUID],
destroyed_entity_ids: Set[UUID],
snapshot_id: str = "",
world_id: str = "default",
) -> SaveResult:
"""Save a batch of entities and handle destroyed entities.
Each entity is saved independently and returns its own result.
Results are aggregated after all concurrent tasks complete,
avoiding the previous bug where multiple coroutines mutated a
shared ``SaveResult`` object under ``asyncio.gather``.
Args:
dirty_entity_ids: Entities that need to be saved
destroyed_entity_ids: Entities that should be deleted
snapshot_id: Opaque identifier for this save cycle
world_id: World identifier stamped on each EntityDocument
Returns:
SaveResult with operation statistics
"""
entities_saved = 0
entities_failed = 0
failed_entities: list[UUID] = []
errors: list[str] = []
# Merge in any previously failed deletes from the tombstone queue
all_destroys = destroyed_entity_ids | set(self._delete_tombstones.keys())
# Process destroyed entities first
surviving_tombstones: dict[UUID, int] = {}
for entity_id in all_destroys:
try:
await asyncio.wait_for(
self.entity_collection.delete(entity_id),
timeout=self.UPSERT_TIMEOUT,
)
entities_saved += 1
except Exception as e:
retry_count = self._delete_tombstones.get(entity_id, 0) + 1
if retry_count >= self.MAX_TOMBSTONE_RETRIES:
logger.error(
"Dropping tombstone for %s after %d retries: %s",
entity_id, retry_count, e,
)
else:
entities_failed += 1
failed_entities.append(entity_id)
errors.append(f"Delete {entity_id}: {e}")
surviving_tombstones[entity_id] = retry_count
self._delete_tombstones = surviving_tombstones
# Build save tasks — each returns an individual result tuple.
# Track entity IDs alongside tasks so batch timeouts can report
# which entities failed (fix: previously lost IDs on exception).
save_tasks: list[asyncio.Task] = []
save_task_entity_ids: list[UUID] = []
for entity_id in dirty_entity_ids:
entity = self.entity_source.get_entity(entity_id)
if entity is None:
continue
# Skip transient entities (player session state, temporary combat effects)
if "transient" in entity.tags:
continue
save_tasks.append(self._save_single_entity(entity, snapshot_id, world_id))
save_task_entity_ids.append(entity_id)
# Execute saves concurrently with batch-level timeout.
# Each task returns (entity_id, success, error_msg | None).
if save_tasks:
individual_results = await self._execute_batch(
save_tasks, save_task_entity_ids, batch_size=self.BATCH_SIZE,
)
for entity_id, success, error_msg in individual_results:
if success:
entities_saved += 1
else:
entities_failed += 1
failed_entities.append(entity_id)
if error_msg:
errors.append(error_msg)
return SaveResult(
entities_saved=entities_saved,
entities_failed=entities_failed,
failed_entities=failed_entities,
errors=errors,
)
async def _save_single_entity(
self, entity: "Entity", snapshot_id: str = "", world_id: str = "default",
) -> tuple[UUID, bool, str | None]:
"""Save a single entity and return an independent result tuple.
Returns:
(entity_id, success, error_message_or_None)
"""
try:
# Snapshot component data on the main thread BEFORE handing to
# the executor to avoid torn reads (see §5.9).
entity_doc = self.serializer.serialize(entity)
entity_doc.updated_at = datetime.now(UTC)
entity_doc.snapshot_id = snapshot_id
entity_doc.world_id = world_id
# Preserve quarantined components through round-trips — MUST
# happen before checksum computation so the hash covers the
# complete document including quarantined data.
quarantine = entity.try_get(QuarantineComponent)
if quarantine is not None:
entity_doc.quarantined_components = quarantine.items
# Compute integrity checksum over components, quarantined data,
# and entity metadata (tags, world_id) — see §5.4.
entity_doc.checksum = EntitySerializer._compute_checksum(
entity_doc.components,
entity_doc.quarantined_components,
tags=entity_doc.tags,
world_id=entity_doc.world_id,
)
# Enforce entity size limit to prevent oversized documents (DoS)
doc_json = entity_doc.model_dump_json()
if len(doc_json) > self.MAX_ENTITY_DOC_BYTES:
return (
entity.id,
False,
f"Entity {entity.id} exceeds {self.MAX_ENTITY_DOC_BYTES} byte "
f"size limit ({len(doc_json)} bytes)",
)
# Upsert to database with per-operation timeout
success = await asyncio.wait_for(
self.entity_collection.upsert(entity.id, entity_doc),
timeout=self.UPSERT_TIMEOUT,
)
if success:
return (entity.id, True, None)
else:
return (entity.id, False, f"Upsert failed for entity {entity.id}")
except asyncio.TimeoutError:
return (entity.id, False, f"Upsert timed out for entity {entity.id}")
except Exception as e:
return (entity.id, False, f"Serialize {entity.id}: {e}")
async def _execute_batch(
self,
tasks: list,
entity_ids: list[UUID],
batch_size: int = 50,
) -> list[tuple[UUID, bool, str | None]]:
"""Execute tasks in batches with per-batch timeout.
``entity_ids`` parallels ``tasks`` so that on batch-level timeout
the correct entity IDs are reported as failed.
"""
all_results: list[tuple[UUID, bool, str | None]] = []
for i in range(0, len(tasks), batch_size):
batch = tasks[i:i + batch_size]
batch_ids = entity_ids[i:i + batch_size]
try:
batch_results = await asyncio.wait_for(
asyncio.gather(*batch, return_exceptions=True),
timeout=self.BATCH_TIMEOUT,
)
for idx, r in enumerate(batch_results):
if isinstance(r, Exception):
all_results.append((batch_ids[idx], False, str(r)))
else:
all_results.append(r)
except asyncio.TimeoutError:
logger.error("Batch %d–%d timed out", i, i + len(batch))
for eid in batch_ids:
all_results.append((eid, False, "Batch timeout"))
return all_results
async def load_all_entities(
self, world_id: str = "default", *, page_size: int = 1000,
) -> LoadResult:
"""Load all persisted entities for a world.
Streams results in pages of ``page_size`` to bound memory for
worlds with 10 K+ entities. Each page is loaded, deserialized,
and injected into the World before the next page is fetched.
An overall timeout prevents unbounded startup stalls.
Includes an idempotency guard: entities that already exist in
the World (e.g., from a prior partial load) are skipped rather
than duplicated.
Args:
world_id: World identifier to load entities for
page_size: Number of entity documents per query page
Returns:
LoadResult with operation statistics
"""
LOAD_TIMEOUT: float = 120.0 # Overall load timeout
result = LoadResult()
try:
offset = 0
load_start = datetime.now(UTC)
while True:
# Check overall timeout
elapsed = (datetime.now(UTC) - load_start).total_seconds()
if elapsed > LOAD_TIMEOUT:
result.errors.append(
f"Load timed out after {elapsed:.1f}s "
f"({result.entities_loaded} loaded so far)"
)
break
# Paginated query via existing DocumentStore.query() API
entity_docs = await self.entity_collection.query(
{"world_id": world_id},
limit=page_size,
offset=offset,
)
if not entity_docs:
break # No more pages
logger.info(
"Loading entities page %d (%d docs, %d loaded so far)",
offset // page_size + 1,
len(entity_docs),
result.entities_loaded,
)
for doc_data in entity_docs:
try:
entity_doc = EntityDocument.model_validate(doc_data)
# Idempotency guard: skip if entity already exists
if self.entity_source.get_entity(entity_doc.entity_id) is not None:
logger.debug(
"Skipping already-loaded entity %s", entity_doc.entity_id,
)
continue
entity = self.serializer.deserialize(entity_doc, self.entity_source)
self.entity_source.add_entity(entity)
result.entities_loaded += 1
except Exception as e:
result.entities_failed += 1
entity_id = doc_data.get('entity_id', 'unknown')
result.failed_entities.append(entity_id)
result.errors.append(f"Deserialize {entity_id}: {e}")
offset += page_size
# Short yield to keep event loop responsive during large loads
await asyncio.sleep(0)
except Exception as e:
result.errors.append(f"Query failed: {e}")
logger.info(
"Entity load complete: %d loaded, %d failed",
result.entities_loaded, result.entities_failed,
)
return result
async def _load_tombstones(self) -> None:
"""Restore persisted tombstones from the persistence_meta collection.
Called during startup before the first save cycle so that
pending deletes from a previous run are not lost.
"""
try:
docs = await self._meta_collection.query({"type": "tombstone"})
for doc in docs:
entity_id = UUID(doc["entity_id"])
retry_count = doc.get("retry_count", 0)
self._delete_tombstones[entity_id] = retry_count
if self._delete_tombstones:
logger.info(
"Restored %d tombstones from persistence_meta",
len(self._delete_tombstones),
)
except Exception as e:
logger.warning("Failed to load tombstones: %s", e)
async def _persist_tombstones(self) -> None:
"""Flush current tombstone queue to the persistence_meta collection.
Called after each save cycle so pending deletes survive crashes.
"""
try:
# Upsert each tombstone; remove entries that have been resolved
for entity_id, retry_count in self._delete_tombstones.items():
await self._meta_collection.upsert(
entity_id,
{"type": "tombstone", "entity_id": str(entity_id),
"retry_count": retry_count},
)
except Exception as e:
logger.warning("Failed to persist tombstones: %s", e)
async def save_all_entities(self, world_id: str = "default") -> SaveResult:
"""Force save all entities in a world (used by @save admin command).
Passes ``world_id`` through to ``save_entities`` so each
``EntityDocument`` is stamped correctly for per-world queries.
"""
all_entity_ids = {entity.id for entity in self.entity_source.get_all_entities()}
snapshot_id = f"{world_id}:{datetime.now(UTC).isoformat()}:{uuid4().hex[:8]}"
return await self.save_entities(
all_entity_ids, set(), snapshot_id=snapshot_id, world_id=world_id,
)
async def get_entity_count(self, world_id: str = "default") -> int:
"""Get count of persisted entities for a world."""
docs = await self.entity_collection.query({"world_id": world_id}, limit=0)
return len(docs)
async def has_pack_entities(self, pack_name: str, world_id: str = "default") -> bool:
"""Check if any entities exist for a specific content pack."""
docs = await self.entity_collection.query(
{"world_id": world_id, "source_pack": pack_name}, limit=1,
)
return len(docs) > 0
async def delete_pack_entities(self, pack_name: str, world_id: str = "default") -> int:
"""Delete all entities created by a content pack.
Uses the existing ``query()`` + per-entity ``delete()`` API
(DocumentStore does not expose ``delete_many()``).
"""
entity_docs = await self.entity_collection.query(
{"world_id": world_id, "source_pack": pack_name},
)
entity_ids = [doc["entity_id"] for doc in entity_docs]
# Delete from database one at a time
deleted_count = 0
for entity_id in entity_ids:
try:
await self.entity_collection.delete(entity_id)
deleted_count += 1
except Exception as e:
logger.error("Failed to delete pack entity %s: %s", entity_id, e)
# Remove from world
for entity_id in entity_ids:
entity = self.entity_source.get_entity(entity_id)
if entity:
self.entity_source.remove_entity(entity_id)
return deleted_count
5. Error Handling, Reliability & Security¶
5.1 Save Failure Recovery¶
Partial Save Failures: The SaveScheduler drains the dirty set atomically before persisting, then re-marks any entity IDs that appear in SaveResult.failed_entities. This ensures failed entities are retried on the next cycle without losing entities that became dirty between a peek and a drain.
Database Connection Loss: Save operations catch connection errors and log failures without crashing. Failed entity IDs are re-marked dirty so the next cycle retries them.
Serialization Errors: Components that fail serialization are logged. The entity's ID is included in the failed set so it is re-marked dirty. Other entities in the batch are unaffected because each _save_single_entity call returns an independent result tuple rather than mutating a shared SaveResult.
5.2 Load Failure Scenarios¶
Component Type Missing: If a component type is not registered (content pack uninstalled), that component's raw data is preserved in quarantined_components on the EntityDocument. If the pack is re-installed later, the quarantined data can be promoted back to a live component. This avoids silent data loss on round-trips.
Schema Evolution: Component fields added or removed between versions are handled via explicit key-set comparison (model_fields.keys() vs stored data keys) before Pydantic validation. Unknown fields are stripped with a logged warning. Missing fields rely on Pydantic defaults. Registered migration functions are applied when stored_version < current_version (see ComponentRegistry.migrate()). If the migration chain has a gap (e.g., no v2→v3 migration registered), the component is quarantined rather than partially migrated.
Corrupted Data: Invalid JSON or malformed component data logs errors but doesn't crash the load process. Per-entity checksum verification catches corruption early (see §5.4).
5.3 Ghost Entity Prevention (Tombstone Queue)¶
Failed deletes are a silent source of data corruption: the entity is removed from the in-memory World but its database record persists, causing "ghost" entities to reappear on next load.
EntityPersistenceManager maintains a _delete_tombstones: dict[UUID, int] mapping entity IDs whose deletes have failed to their retry count. On each save cycle the tombstone set is merged with the incoming destroyed_entity_ids and retried. Successful deletes are removed from the map; failures increment the retry counter. Tombstones that exceed MAX_TOMBSTONE_RETRIES (default 10) are dropped with an ERROR log to prevent unbounded growth. Non-empty tombstones are logged at WARNING level so operators can investigate persistent failures.
Crash-safe tombstone persistence: The in-memory tombstone queue is flushed to the persistence_meta DocumentStore collection after each save cycle (see _persist_tombstones()). On startup, _load_tombstones() restores pending deletes before the first save fires. Without this, a crash between a failed delete and the next retry would lose the tombstone, allowing ghost entities to reappear silently.
5.4 Corruption Detection (Entity Checksums)¶
Each EntityDocument carries a checksum field containing the SHA-256 hex digest of the deterministically serialized component list — including quarantined components — sorted by component_type, with model_dump_json() for stable field ordering. The checksum also covers entity metadata (tags, world_id) so that tag mutations and world reassignments are detected. Including quarantined data ensures the checksum remains stable across round-trips even when some packs are unloaded.
- On save:
EntitySerializer._compute_checksum()is called and the result stored in the document. - On load: If a checksum is present,
EntitySerializer.deserialize()recomputes the hash and raisesIntegrityErroron mismatch. This detects bit-rot, truncated writes, and external tampering at load time rather than allowing corrupt state to propagate.
Entities that fail the checksum check are excluded from the World and reported in LoadResult.errors for operator triage.
Security note: The current checksum uses plain SHA-256. For deployments where the storage backend is untrusted, consider replacing with HMAC-SHA-256 keyed by a server secret to provide tamper resistance.
5.5 Architectural Decoupling¶
EntityPersistenceManager accepts an EntitySource protocol (see §4.6) instead of a direct World reference. World satisfies this protocol, but tests can provide a lightweight stub. For larger deployments that need a thinner I/O boundary, the class can be further split:
| Layer | Responsibility |
|---|---|
| EntityIOManager | Pure I/O: serialize, upsert, delete, load. No World reference. Accepts and returns EntityDocument objects. |
| WorldPersistenceCoordinator | Bridge: owns the World ref, resolves entity IDs to live entities, delegates to EntityIOManager, handles re-marking dirty on failure. |
This split is not required for initial implementation but should be revisited if persistence needs to be unit-tested without a full World instance.
5.6 Concurrent Save Coordination¶
Save coordination uses asyncio.Lock instead of a plain boolean flag. The previous _save_in_progress boolean suffered from a TOCTOU race: two trigger_save() calls could both read False before either set it to True. asyncio.Lock provides proper async mutual exclusion — if a save is already running, the second caller waits for the lock rather than silently returning.
5.7 Timeout Budget¶
All I/O operations enforce timeouts to prevent unbounded waits:
| Scope | Default | Rationale |
|---|---|---|
| Per-upsert | 5 s | Single DB round-trip should complete quickly |
| Per-batch (50 entities) | 30 s | Allows retries within the batch window |
| Per-save-cycle | 60 s | Hard ceiling; on timeout all entities are re-marked dirty |
Timeouts are implemented via asyncio.wait_for() and are sourced from PersistenceSettings (see §4.6), which can be overridden per deployment via MAID_PERSISTENCE__* environment variables.
5.8 Auto-Detecting Component Mutations (Phase 0 Prerequisite)¶
This hook is a Phase 0 prerequisite, not optional. Without it, DirtyTracker relies solely on ComponentAddedEvent/ComponentRemovedEvent, which do not fire for in-place mutations (e.g., health.current -= 10). Every system that mutates component fields would need manual mark_dirty() calls — an error-prone requirement that will silently drop changes when forgotten.
Add a __setattr__ hook on the Component base class:
class Component(BaseModel):
"""ECS Component base with automatic dirty notification."""
_dirty_callback: Callable[[UUID], None] | None = PrivateAttr(default=None)
_owner_id: UUID | None = PrivateAttr(default=None)
_suppress_dirty: bool = PrivateAttr(default=False)
def __setattr__(self, name: str, value: Any) -> None:
super().__setattr__(name, value)
# Notify DirtyTracker on any public field mutation
if (
not name.startswith("_")
and not self._suppress_dirty
and self._dirty_callback is not None
and self._owner_id is not None
):
self._dirty_callback(self._owner_id)
When an entity is added to a tracked World, the persistence layer binds _dirty_callback to DirtyTracker.mark_dirty and sets _owner_id to the entity's UUID. In-place mutations like health.current -= 10 then automatically mark the entity dirty without any action from the authoring system.
_suppress_dirty flag for deserialization. During EntitySerializer.deserialize() and model_validate(), Pydantic's internal field assignment triggers __setattr__ for every field, flooding the dirty set with false positives for entities that have not actually changed. The serializer sets _suppress_dirty = True before constructing the component and clears it afterward:
component._suppress_dirty = True
entity.add(component) # add triggers __setattr__ for _owner_id binding
component._suppress_dirty = False
5.8.1 Container Mutation Gap¶
Known limitation:
__setattr__fires only on field reassignment (entity.inventory = new_list), not on in-place mutations of mutable containers (entity.inventory.items.append(sword),entity.equipment.slots["head"] = helmet,entity.tags_set.add("poisoned")). These are the primary mutation patterns forInventoryComponent,EquipmentComponent, andDialogueComponent.
Two complementary strategies address this:
Strategy A — Observable container wrappers (recommended for stdlib components):
Provide ObservableList, ObservableDict, and ObservableSet wrappers that invoke a notification callback on mutation. Components that hold mutable containers use these as field types:
class ObservableList(list, Generic[T]):
"""List subclass that notifies on mutation."""
_notify: Callable[[], None] | None
def append(self, item: T) -> None:
super().append(item)
if self._notify:
self._notify()
def __setitem__(self, index: int, value: T) -> None:
super().__setitem__(index, value)
if self._notify:
self._notify()
# Override extend, remove, pop, __delitem__, insert, clear, sort, reverse …
class InventoryComponent(Component):
"""Example: items field uses ObservableList for auto-dirty."""
items: ObservableList[UUID] = Field(default_factory=ObservableList)
When the persistence layer binds _dirty_callback on a component, it also walks the component's fields and binds _notify on any Observable* containers. A Pydantic model_validator(mode="after") can automate this wiring.
Strategy B — Explicit mark_dirty() for third-party / ad-hoc containers:
Content-pack authors whose components use plain list/dict/set fields (or nested Pydantic models with their own mutable state) must call mark_dirty() after container mutations:
# In a combat system:
entity.get(InventoryComponent).items.append(sword_id)
dirty_tracker.mark_dirty(entity.id) # Required — list.append bypasses __setattr__
Document this requirement prominently in the content-pack authoring guide. Provide a Component.notify_mutation() convenience method that calls the bound _dirty_callback if present, reducing the ceremony for pack authors:
class Component(BaseModel):
...
def notify_mutation(self) -> None:
"""Explicitly mark the owning entity dirty after a container mutation."""
if self._dirty_callback is not None and self._owner_id is not None:
self._dirty_callback(self._owner_id)
Implementation plan: Phase 1 ships __setattr__ + _suppress_dirty + notify_mutation(). Phase 2 adds ObservableList/Dict/Set and retrofits stdlib components. This lets the system work correctly from day one (via explicit notify_mutation()) while the ergonomic wrappers are developed.
5.9 Off-Thread Serialization (Save Window)¶
Pydantic model_dump() and model_validate() are CPU-bound. For large save batches (>100 entities) these calls can measurably block the event loop. However, offloading serialization to a background thread introduces torn-read risk: Entity objects are mutable and the tick loop may mutate components concurrently.
Concrete protection: main-thread snapshot, then off-thread serialization.
All mutable state must be captured on the main thread before any executor work begins. The recommended approach is a two-phase save:
-
Snapshot phase (main thread, between ticks): For each dirty entity, call
component.model_dump()on every component to produce an immutabledictsnapshot. Collect these snapshots into a list of(entity_id, tags_copy, [dict, …])tuples. This runs during a save window — a brief pause between ticks where no systems are executing, analogous to a GC safe-point. -
Serialization phase (executor thread): Hand the immutable snapshots to
run_in_executor()for JSON encoding, checksum computation, andEntityDocumentconstruction. Because the executor only touches the frozen dicts, there is no torn-read risk.
# Phase 1: snapshot on main thread (inside save window)
snapshots: list[tuple[UUID, set[str], list[dict]]] = []
for entity_id in dirty_ids:
entity = self.entity_source.get_entity(entity_id)
if entity is None:
continue
comp_dicts = [c.model_dump() for c in entity.components]
snapshots.append((entity.id, set(entity.tags), comp_dicts))
# Phase 2: serialize off-thread
loop = asyncio.get_running_loop()
entity_docs = await loop.run_in_executor(
None, self._build_documents, snapshots, snapshot_id,
)
The save window is implemented by the GameEngine tick loop: after all systems have processed a tick and before the next tick begins, the engine invokes SaveScheduler.run_snapshot_phase() if a save is pending. This ensures the snapshot sees a consistent world state without requiring any locking.
Fallback for small saves: When the dirty set contains fewer than ~100 entities, the overhead of
run_in_executorexceeds the serialization cost. In that case, serialize synchronously on the main thread. Gate the executor path onlen(dirty_ids) > EXECUTOR_THRESHOLD.
5.10 Snapshot Manifest¶
Each save cycle generates a snapshot_id (e.g., a ULID or f"{world_id}:{utc_timestamp}"). The ID is stamped on every EntityDocument written during that cycle. This enables:
- Consistency queries: "Show me all entities from the last complete save" by filtering on
snapshot_id. - Partial-failure forensics: If a cycle fails halfway, operators can identify which entities belong to the incomplete snapshot.
- Point-in-time rollback: Combined with the backup system, snapshot IDs let operators restore to the boundary of a known-good save.
6. Migration Strategy¶
6.1 Phase 0: Prerequisites (Week 1)¶
Task 0-A: Add ECS Event Emission
- Modify Entity.add() to emit ComponentAddedEvent
- Modify Entity.remove() to emit ComponentRemovedEvent
- Modify EntityManager.create() to emit EntityCreatedEvent
- Add entity destruction event emission
Task 0-B: DocumentStore Upsert Extension
- Add upsert() method to DocumentCollection interface
- Implement PostgreSQL INSERT ... ON CONFLICT DO UPDATE
- Implement in-memory upsert for testing
- Update DocumentStore tests
Task 0-C: Component __setattr__ Hook
- Add _dirty_callback, _owner_id, and _suppress_dirty private attributes to Component base class
- Implement __setattr__ override for automatic dirty notification (see §5.8)
- Add notify_mutation() convenience method (see §5.8.1)
Task 0-D: Extend ContentPack Protocol
- Add register_component_types(registry: ComponentRegistry) -> None to ContentPack protocol
- Update GameEngine.load_content_pack() to call it before on_load()
Task 0-E: DocumentStore Pagination
- Add limit and offset parameters to DocumentCollection.query()
- Required before load_all_entities can be implemented safely
Task 0-F: Composite Primary Key Migration
- Alter documents table: PRIMARY KEY (collection, id)
- Add B-tree indexes on world_id, source_pack, snapshot_id (see §8.2)
6.2 Phase 1: Core Infrastructure (Weeks 2-3)¶
Deliverables: - ComponentRegistry implementation - EntitySerializer with schema evolution support - EntityDocument and database schema - Basic persistence manager with save/load - Unit tests for all components
Validation Criteria:
- Can save and load simple entities with basic components
- Component type registration works across content packs
- Schema evolution handles added/removed fields
- Migration chain completeness is validated at startup — missing intermediate steps (e.g., no v2→v3 when v3 is current) raise an error during GameEngine.start(), not lazily on first deserialization
6.3 Phase 2: Change Tracking (Weeks 4-5)¶
Deliverables:
- DirtyTracker with event subscriptions
- SaveScheduler with periodic saves (initial short-delay save + configurable interval)
- Observable container wrappers (ObservableList, ObservableDict, ObservableSet) for stdlib components
- Save/load result reporting
- Integration with existing EventBus
Validation Criteria:
- Only dirty entities are saved per cycle
- Manual dirty marking works for in-place mutations
- Container mutations (list.append, dict.setitem) auto-mark dirty via observable wrappers
- _suppress_dirty flag prevents false dirties during deserialization
- Save failures don't crash the server
6.4 Phase 3: Production Features (Weeks 6-7)¶
Deliverables: - ContentPack integration (register_component_types) - Builder command persistence (@create, @dig tags entities) - Admin commands (@save, @persist status) - World load/save startup integration
Validation Criteria: - Builder-created content survives restart - Content packs can detect first-run vs restart - Manual saves work via admin commands
6.5 Phase 4: Advanced Features (Weeks 8-10)¶
Deliverables: - Backup/restore system with snapshots - Corruption detection and repair tools - Performance optimization and monitoring - Documentation and operational runbooks
Validation Criteria: - Full world snapshots work correctly - Corruption detection catches common issues - Performance meets NFR targets
7. Testing Strategy¶
7.1 Unit Tests¶
Component Registry Tests: - Registration and resolution of component types - Alias support for renamed components - Pack unloading removes correct components - Error handling for duplicate registrations
EntitySerializer Tests: - Round-trip serialization/deserialization - Schema evolution with added/removed fields - Unknown component type handling - Validation error scenarios
DirtyTracker Tests: - Event-based dirty marking - Manual dirty marking - Drain operations preserve state on failure - Memory leak prevention
7.2 Integration Tests¶
Save/Load Cycles: - Full world save/load preserves all data - Incremental saves only write changed entities - Failed saves don't corrupt state - Multiple content packs work together
Content Pack Integration: - Packs can detect persisted entities - Component registration during pack load - Entity creation skip logic works correctly
7.3 Performance Tests¶
Save Performance: - 500 dirty entities save in <500ms - 10,000 entity full save completes reasonably - Memory usage remains bounded - No performance degradation over time
Load Performance: - 50,000 entity load in <5 seconds - Memory usage is reasonable - ComponentRegistry resolution is fast
7.4 Crash Recovery Tests¶
Mid-Save Crashes: - Kill server during save cycle - Verify no corruption on restart - Dirty tracking preserves unsaved entities
Database Failure Recovery: - Simulate connection loss during saves - Verify graceful error handling - Automatic retry on reconnection
8. Performance Considerations¶
8.1 Save Performance Targets¶
- Incremental saves: 500 dirty entities in <500ms (NFR-1)
- Full saves: 10,000 entities in <5 seconds
- Memory overhead: <100 bytes per entity for dirty tracking (NFR-5)
- Tick impact: Save operations don't block the tick loop (async)
8.2 Optimization Strategies¶
Batch Operations: Entity saves use asyncio.gather with configurable batch sizes to balance concurrency and resource usage.
Bulk Upserts: For large save batches, replace per-entity INSERT … ON CONFLICT with multi-row upserts (batches of 50–100 rows per statement) or PostgreSQL COPY for full-world saves. This reduces round-trips from O(N) to O(N/batch_size) and can improve throughput by 5–10×. The _execute_batch method should construct multi-row VALUES clauses rather than issuing individual upserts.
Connection Pooling: DocumentStore connection pooling prevents connection exhaustion during large save operations.
Index Strategy: PostgreSQL GIN index on JSONB data enables fast component-type queries without full table scans. Additionally, add B-tree indexes on frequently queried columns:
CREATE INDEX IF NOT EXISTS idx_documents_world_id
ON documents ((data->>'world_id'));
CREATE INDEX IF NOT EXISTS idx_documents_source_pack
ON documents ((data->>'source_pack'));
CREATE INDEX IF NOT EXISTS idx_documents_snapshot_id
ON documents ((data->>'snapshot_id'));
These support load_all_entities (world_id), has_pack_entities/delete_pack_entities (source_pack), and snapshot-based consistency queries (snapshot_id).
Memory Management: DirtyTracker uses sets for O(1) dirty checks and atomic drain operations to minimize memory footprint.
8.3 Monitoring Points¶
Save Metrics: - Save cycle duration and frequency - Entity save success/failure rates - Dirty entity count over time - Database operation latencies
Load Metrics: - Startup load duration - Entity load success/failure rates - Component type resolution performance - Memory usage during load
9. Open Questions¶
9.1 Multi-World Persistence Scope¶
Question: How should entity persistence interact with the existing WorldManager multi-world support?
Context: MAID supports multiple independent worlds via WorldManager. Portal transitions deep-copy entities between worlds.
Proposed Resolution: Each world has independent persistence. Portal transitions create new persistent entities in the destination world and mark originals for deletion. Add world_id scoping to all queries.
9.1.1 Cross-World Entity Deep-Copy Strategy¶
Portal transitions must handle entity hierarchies — a player entity carries inventory items, each of which is itself an entity with components. A naïve shallow copy loses child references.
Recursive deep-copy algorithm:
- Serialize the source entity via
EntitySerializer.serialize(). - Walk the component list looking for entity-reference fields (components whose fields contain
UUIDvalues that map to other entities — e.g.,InventoryComponent.item_ids). - For each referenced child entity, recursively serialize it, tracking visited entity IDs in a
visited: set[UUID]to prevent infinite cycles from circular references (e.g., mutual parent/child pointers). Enforce a maximum recursion depth (default 32) and raiseDeepCopyDepthErrorif exceeded. - Assign fresh UUIDs to all entities in the copy set to avoid ID collisions in the destination world.
- Rewrite intra-set entity references to use the new UUIDs.
- Deserialize the entire set into the destination world atomically.
- Mark all source entities for deletion (or flag as "transferred" if the source world should retain a tombstone).
This strategy keeps portal transitions transactional: either the full hierarchy copies or none of it does.
9.2 Player Entity Persistence Overlap¶
Question: Should player character entities be persisted via this system or continue using CharacterManager?
Context: CharacterManager already persists Character models. The ECS entities created from characters are currently ephemeral.
Proposed Resolution: Keep CharacterManager for account-level data. Use entity persistence for runtime ECS state (position, health, temporary effects). Ensure no ID conflicts.
9.3 Large Binary Component Data¶
Question: How should components with large binary data (e.g., generated maps, images) be handled?
Context: PostgreSQL JSONB has practical size limits. Serializing large binary data to JSON is inefficient.
Resolution: Introduce a BlobStore abstraction alongside DocumentStore:
class BlobReference(BaseModel):
"""Lightweight reference stored inside a component's JSONB data."""
blob_id: UUID
content_type: str # e.g. "image/png", "application/octet-stream"
size_bytes: int
checksum: str # SHA-256 of the blob content
class BlobStore(Protocol):
"""Abstract interface for large binary object storage."""
async def put(self, blob_id: UUID, data: bytes, content_type: str) -> BlobReference: ...
async def get(self, blob_id: UUID) -> bytes: ...
async def delete(self, blob_id: UUID) -> bool: ...
async def exists(self, blob_id: UUID) -> bool: ...
Components that hold large data store a BlobReference in their model fields instead of embedding raw bytes. The persistence layer resolves references transparently during serialization and deserialization. Initial implementations can back BlobStore with the filesystem (data/blobs/) or PostgreSQL BYTEA; S3-compatible stores can be added later.
10. Design Decisions Log¶
10.1 Event-Driven vs Polling for Dirty Tracking¶
Decision: Use event-driven dirty tracking via EventBus subscription.
Rationale: Leverages existing event infrastructure. More efficient than polling. Provides immediate change detection.
Trade-offs: Requires adding event emission to ECS core (prerequisite work). Manual mark_dirty() / notify_mutation() still needed for in-place container mutations (list.append, dict.__setitem__) that bypass __setattr__; observable container wrappers address this for stdlib components (see §5.8.1).
10.2 Upsert vs Create/Update Pattern¶
Decision: Add upsert capability to DocumentStore via database-specific implementations.
Rationale: Eliminates race conditions in save operations. Simpler save flow without existence checks. Better performance with single round-trip.
Trade-offs: Requires extending DocumentStore interface. Database-specific SQL for PostgreSQL.
10.3 Component Type Keys: name vs get_type()¶
Decision: Use component.get_type() consistently for serialization and registry keys.
Rationale: get_type() is the documented API. Allows custom component type names. Consistent with existing component protocol.
Trade-offs: Requires updating Entity.to_dict() to use get_type(). Slightly more complex than name.
10.4 Schema Evolution Strategy¶
Decision: Explicit key-set pre-filtering with a registered migration function chain.
Rationale: Before Pydantic validation, compare model_fields.keys() against stored data keys to detect and strip removed fields. When stored_version < current_version, apply chained migration functions registered via ComponentRegistry.register_migration(). This replaces the fragile try/except string-matching approach and adds forward migration capability.
Trade-offs: Requires content packs to register migration functions for non-trivial schema changes. Simple field additions/removals are handled automatically via Pydantic defaults and key filtering.
10.5 Save Coordination Approach¶
Decision: Use asyncio.Lock for save coordination instead of a boolean flag.
Rationale: The previous boolean _save_in_progress flag had a TOCTOU race condition — two concurrent trigger_save() calls could both read False before either set it to True, leading to overlapping saves. asyncio.Lock provides proper mutual exclusion in the async context. Callers that arrive while a save is running wait for the lock rather than silently returning.
Trade-offs: Slightly more complex than a boolean flag. Manual trigger_save() calls may block briefly if a periodic save is running. This is acceptable because the alternative (dropped saves) is worse.
10.6 Atomic Drain vs Peek-Then-Drain for Dirty Sets¶
Decision: Drain the dirty set atomically before persisting, then re-mark failures.
Rationale: The previous peek-then-drain approach created a data-loss window: entities that became dirty between the peek snapshot and the post-save drain were silently discarded when drain_dirty() cleared the entire set. Draining upfront and re-marking failures on the back end is safe because the set of failed IDs is always a subset of what was drained.
Trade-offs: On total save failure (e.g., database down), every entity in the batch must be re-marked individually. This is O(N) in the failure case but avoids data loss in the success case, which is the common path.
10.7 Independent Save Results vs Shared Mutable Object¶
Decision: Each _save_single_entity returns a (UUID, bool, str | None) result tuple; the caller aggregates after asyncio.gather completes.
Rationale: The previous design passed a shared SaveResult object into concurrent coroutines. While CPython's GIL prevents true data races, mutating shared state from asyncio.gather tasks is fragile and violates the principle of least surprise. Independent return values are easier to reason about and test.
Trade-offs: Slight increase in allocation (one tuple per entity) which is negligible compared to I/O costs.
10.8 Schema Evolution: Pre-Filtering vs Exception String Matching¶
Decision: Compare model_fields.keys() against stored data keys before Pydantic validation, rather than catching ValueError and inspecting the exception message string.
Rationale: The previous approach (if "extra" in str(e).lower()) was fragile — it could match unrelated validation errors whose messages happened to contain the word "extra", and it would break silently if Pydantic changed its error wording. Explicit key-set comparison is deterministic and independent of Pydantic internals.
Trade-offs: Adds one set operation per component on every deserialization. Cost is negligible.
10.9 Component Registry Overwrite Protection¶
Decision: ComponentRegistry.register() raises ComponentConflictError by default when a type is already registered. An explicit allow_overwrite=True parameter is required for hot-reload.
Rationale: Silent re-registration opens the door to component hijacking — a malicious or buggy content pack could replace a core component class and alter serialization behavior. Failing loudly by default makes this a conscious opt-in decision.
Trade-offs: Hot-reload code paths must pass allow_overwrite=True. This is a minor inconvenience that prevents a class of subtle bugs.
10.10 Pack Ownership Tracking¶
Decision: Track component-to-pack ownership via a _pack_ownership: dict[str, str] in ComponentRegistry rather than relying on a _pack_name attribute on component classes.
Rationale: The previous unregister_pack() implementation checked getattr(component_class, '_pack_name', None), but no code path ever set this attribute on component classes. Tracking ownership in the registry itself is self-contained and doesn't require modifying the Component base class.
Trade-offs: Requires pack_name to be passed at registration time. This is already natural since registration happens inside ContentPack.on_load().
10.11 Quarantining Unknown Components¶
Decision: When a component type cannot be resolved during deserialization, preserve its raw data in quarantined_components on the EntityDocument rather than silently dropping it.
Rationale: Silent drops cause permanent data loss on round-trips. If a content pack is temporarily unloaded (e.g., during development or A/B testing), its component data should survive so it can be restored when the pack is re-loaded.
Trade-offs: Increases document size by carrying raw JSON for unresolvable components. This is bounded by the number of unloaded packs and is preferable to data loss.
This design document incorporates feedback from architecture, reliability, API design, security, and completeness reviews. Key changes from the review cycle include: atomic drain-then-re-mark for dirty tracking (including destroyed IDs on timeout), asyncio.Lock for save coordination, per-operation timeouts, tombstone queues for reliable deletes (bounded by retry count, crash-safe via persistence_meta), independent per-entity save results with entity ID tracking, explicit schema pre-filtering, entity checksums covering both components and metadata (tags, world_id) using model_dump_json() for deterministic serialization, component quarantining via QuarantineComponent (attached before checksum computation), registry overwrite protection, pack ownership tracking, migration function support with gap detection and startup validation, observable container wrappers for mutable-container dirty tracking, _suppress_dirty flag to prevent false dirties during deserialization, notify_mutation() convenience method, save-window architecture with main-thread snapshot before executor handoff, paginated entity loading with idempotency guards and overall load timeout, paginated DocumentStore queries (P0-E prerequisite), composite primary key on documents table (P0-F prerequisite), B-tree indexes on world_id/source_pack/snapshot_id, bulk upsert optimization path, initial short-delay first save, dirty-set backpressure warnings, world_id propagation through the save pipeline, a BlobStore interface, snapshot manifests with generated IDs, EntitySource protocol for testability, PersistenceSettings for configurable tuning (with corrected initialization pattern), WorldSaveStartedEvent/WorldSaveCompletedEvent definitions, entity size limits, ContentPack protocol extension with register_component_types(), and __setattr__-based auto-dirty detection as a Phase 0 prerequisite. The phased approach allows for incremental delivery while ensuring production readiness.