Database Migration System — Final Design Document¶
Version: 3.1
Status: Implemented
Author: Infrastructure Architecture Team
Date: 2025-01-14 (updated 2025-07-18)
Package: maid-engine
Executive Summary¶
MAID has a fully implemented database migration system built on asyncpg with PostgreSQL. The maid db CLI provides six primary commands (migrate, status, rollback, history, validate, repair) plus authoring and maintenance helpers (create, baseline, cleanup, init). All migration infrastructure described in this document is implemented in packages/maid-engine/src/maid_engine/migrations/ and packages/maid-engine/src/maid_engine/cli/commands/db_migrate.py.
This document describes the comprehensive migration system and its architectural decisions. The implementation provides:
- Safe transactional migrations with atomic rollback guarantees and explicit batching controls for large datasets
- Content pack integration with namespace isolation and dependency resolution
- JSONB document evolution with type-safe transforms and version stamping
- Production-ready CLI with proper content pack discovery, comprehensive error handling, and operational safety features
- Zero-downtime deployment support through expand-contract patterns and online-safe classification
- Enterprise security with input validation, audit logging, and permission controls
The approach uses a custom migration runner built on asyncpg, avoiding SQLAlchemy dependencies while providing first-class support for MAID's multi-namespace content pack architecture and JSONB document store patterns. The implementation is in packages/maid-engine/src/maid_engine/migrations/.
Key decisions:
- Advisory lock implementation with per-namespace granularity, session-level semantics, int4-safe lock IDs, stale heartbeat detection, and cross-namespace requirement validation under lock
- API-level namespace ownership enforcement through the transform library, with RestrictedContext for semi-trusted packs; ownership map built dynamically from pack manifest components fields
- Checkpoint-based batching with monotonic keyset pagination (doc_seq), atomic mini-transactions, and namespace-scoped checkpoint IDs
- SQL injection prevention through a structured JSONBUpdateSpec builder (no raw SQL interpolation); statement timeouts use SET SESSION for non-transactional modes (with RESET in finally)
- Content pack manifest validation for migration discovery and dependency enforcement, with reserved namespace protection; extends existing ContentPackManifest from maid_engine.plugins.protocol
- BaseMigrationContext / MigrationContext / RestrictedContext hierarchy; Migration protocol with class-based migrations only (no module-level adaptation)
- ExecutionMode enum replacing overlapping REQUIRES_BATCHING / REQUIRES_NO_TRANSACTION booleans; validated for compatibility at load time
- Separated discovery/planning/execution architecture (registry, planner, runner) with pool injection via from_pool() constructors
- Security model restricting community packs to declarative YAML migrations (loaded via yaml.safe_load() only); NO_TRANSACTION mode restricted to trusted namespaces
- DESTRUCTIVE_ROLLBACK declarative attribute replacing regex-based source scanning; rollback uses crash-safe rollback_started status marker
- Rollback dependency resolution via semantic migration_requires map, not temporal ordering
- Two-phase history recording (pending → applied) for non-transactional modes; atomic history within transaction for transactional mode
- Prometheus metrics for migration duration, success/failure counts, batch progress, and lock contention
Problem Statement & Prior State¶
Note: This section describes the state of the system before the migration framework was implemented. It is retained for historical context. All problems listed below have been resolved by the implementation described in this document.
Prior Database Management¶
MAID had two disconnected database initialization systems:
-
DatabaseManager (
packages/maid-engine/src/maid_engine/storage/database.py): Managesaccountsandsessionstables via raw asyncpg. Schema defined asSCHEMA_SQLconstant executed unconditionally: -
PostgresDocumentStore (
packages/maid-engine/src/maid_engine/storage/document_store.py): Createsdocumentstable for JSONB data storage.
Problems (now resolved):
- No version tracking: No way to determine what schema version a database was running → resolved by
_migration_historytable - Silent schema drift:
CREATE TABLE IF NOT EXISTSsucceeded even when existing schema was outdated → resolved by checksum verification - No upgrade path: Schema changes required manual intervention → resolved by
maid db migrate - Disconnected subsystems:
DatabaseManagerwas not used byGameEngine→ resolved by unified migration runner - No rollback capability: Failed upgrades left databases in undefined states → resolved by
maid db rollbackwith crash-safe status markers
Content Pack Architecture Challenge¶
MAID's content pack system created a unique migration challenge:
- Multiple independent codebases: Engine, stdlib, classic-rpg, tutorial-world, community packs
- Dynamic loading: Content packs loaded at runtime based on configuration
- Shared database: All packs store data in the same PostgreSQL instance
- JSONB schema evolution: Component models change over time, requiring data transformations
- Dependency chains: Packs depend on other packs (classic-rpg → stdlib → engine)
No existing migration tool (Alembic, yoyo-migrations) was designed for this multi-namespace, dynamic content pack architecture. This motivated the custom migration runner described below.
Architecture Overview¶
Migration Framework Architecture¶
┌─────────────────────────────────────────────────────┐
│ CLI Layer │
│ maid db migrate | rollback | status | history │
│ maid db validate | repair | create | baseline │
│ --dry-run --online-safe --format=json|text │
│ ↓ Discovers content packs via manifest registry │
├─────────────────────────────────────────────────────┤
│ MigrationRunner (execution only) │
│ - Per-namespace advisory locks │
│ - Transactional safety with non-DDL support │
│ - Per-migration timeouts (statement, lock, total) │
│ - Audit logging & rollback tracking │
│ - PostgreSQL NOTIFY on schema changes │
├──────────────────────┬──────────────────────────────┤
│ PackManifestRegistry │ MigrationPlanner │
│ - Content pack │ - Dependency resolution │
│ discovery │ - Topological ordering │
│ - Entry point scan │ - Checksum verification │
│ - Migration source │ - Requirement validation │
│ validation │ │
├──────────────────────┴──────────────────────────────┤
│ MigrationHistory │ BatchExecutor │
│ - Applied migrations │ - Cursor/keyset pagination │
│ with checksums │ - Atomic mini-transactions │
│ - Rollback audit trail│ - Checkpoint resume │
│ - Integrity checking │ - Adaptive batch sizing │
├────────────────────────┴────────────────────────────┤
│ JSONB Transform Library │
│ - JSONBUpdateSpec builder (no raw SQL) │
│ - API-level ownership enforcement │
│ - Schema version management │
│ - Type-safe field transforms │
├─────────────────────────────────────────────────────┤
│ Migration Files (typed Migration protocol) │
│ engine/0001_initial_schema.py (Python - trusted) │
│ stdlib/0001_initial_collections.py (Python - trusted│
│ community_pack/0001_setup.yaml (YAML - sandboxed) │
├─────────────────────────────────────────────────────┤
│ PostgreSQL with asyncpg │
│ ├─ _migration_history (applied migrations) │
│ ├─ _migration_checkpoints (batched progress) │
│ ├─ _migration_rollback_log (rollback tracking) │
│ └─ Indexes: collection+doc_seq, GIN on JSONB │
└─────────────────────────────────────────────────────┘
Module Layout¶
packages/maid-engine/src/maid_engine/
├── migrations/ # Migration framework
│ ├── __init__.py
│ ├── runner.py # MigrationRunner (execution engine, resource management)
│ ├── planner.py # MigrationPlanner (ordering, dependency resolution)
│ ├── discovery.py # PackManifestRegistry (content pack discovery)
│ ├── protocol.py # Migration protocol, BaseMigrationContext, MigrationContext, RestrictedContext
│ ├── history.py # Migration tracking, rollback log, infrastructure DDL
│ ├── lock.py # Global advisory lock (pg_try_advisory_lock with backoff)
│ ├── checkpoint.py # Batched operation progress tracking & resume
│ ├── batch_executor.py # Generic BatchExecutor with checkpoint support
│ ├── types.py # Core data types (MigrationSource, MigrationPlan, MigrationResult, etc.)
│ ├── jsonb_transforms.py # Safe JSONB manipulation (add/remove/rename/transform fields)
│ ├── exceptions.py # Migration-specific exceptions
│ └── engine/ # Engine's own migrations
│ ├── __init__.py
│ ├── 0001_initial_schema.py
│ ├── 0002_document_store.py
│ └── 0003_migration_history.py
└── cli/
└── commands/
└── db_migrate.py # Migration CLI commands (10 commands)
Schema Versioning Strategy¶
Sequential integers within namespaces:
- Namespace: Content pack's
manifest.name(e.g.,"engine","stdlib","classic-rpg") - Sequence: Zero-padded 4-digit integer starting at
0001 - Cross-namespace ordering: Follows content pack dependency resolution
Namespace ownership:
- engine: Core MAID infrastructure
- stdlib: Standard library content pack
- classic-rpg: Classic MUD gameplay content pack
- tutorial-world: Tutorial/example content pack
- Community packs: Named per their manifest
Migration Protocol¶
Formal Migration Interface¶
All migrations implement a typed protocol. This replaces the previous implicit convention of module-level functions and constants with a discoverable, validated interface.
Migration files are authored as classes that directly satisfy the Migration protocol.
Module-level migration files are not supported — all migrations must be class-based
with a module-level migration instance for the loader to discover.
# packages/maid-engine/src/maid_engine/migrations/protocol.py
from enum import Enum
from typing import Protocol, runtime_checkable
class ExecutionMode(Enum):
"""How the runner should execute the migration.
Replaces the overlapping REQUIRES_BATCHING / REQUIRES_NO_TRANSACTION
booleans with a single discriminator.
"""
TRANSACTIONAL = "transactional" # Default: single transaction wraps upgrade()
BATCHED = "batched" # Checkpoint-based batching, no outer transaction
NO_TRANSACTION = "no_transaction" # No transaction wrapping (DDL, concurrent index)
def validate_execution_modes(
execution_mode: ExecutionMode,
rollback_execution_mode: ExecutionMode,
) -> None:
"""Validate that execution_mode and rollback_execution_mode are compatible.
Raises ValueError if the combination is invalid. Called at migration
load time, not at execution time, to fail fast.
Rules:
- BATCHED upgrade requires BATCHED or TRANSACTIONAL rollback
- NO_TRANSACTION upgrade requires NO_TRANSACTION or TRANSACTIONAL rollback
- TRANSACTIONAL upgrade allows any rollback mode
"""
invalid_combos = {
(ExecutionMode.BATCHED, ExecutionMode.NO_TRANSACTION),
(ExecutionMode.NO_TRANSACTION, ExecutionMode.BATCHED),
}
if (execution_mode, rollback_execution_mode) in invalid_combos:
raise ValueError(
f"Incompatible execution modes: upgrade={execution_mode.value}, "
f"rollback={rollback_execution_mode.value}. "
f"BATCHED↔NO_TRANSACTION combinations are not supported."
)
@runtime_checkable
class Migration(Protocol):
"""Formal protocol that all migration classes must implement.
upgrade() and downgrade() receive BaseMigrationContext, the common
interface shared by MigrationContext (trusted) and RestrictedContext
(semi-trusted). Trusted migrations can isinstance-check or type-narrow
to MigrationContext when they need raw conn access.
"""
# Required attributes
namespace: str # Owning content pack name
sequence: int # Monotonic sequence within namespace
description: str # Human-readable summary
# Optional declarations (with defaults)
ONLINE_SAFE: bool # Default: False — see Migration Safety Guidelines
execution_mode: ExecutionMode # Default: ExecutionMode.TRANSACTIONAL
rollback_execution_mode: ExecutionMode # Default: ExecutionMode.TRANSACTIONAL
ROLLBACK_IDEMPOTENT: bool # Default: False
DESTRUCTIVE_ROLLBACK: bool # Default: False — declarative flag
STATEMENT_TIMEOUT_MS: int | None # Default: None (use global)
LOCK_TIMEOUT_MS: int | None # Default: None (use global)
MIGRATION_TIMEOUT_S: int | None # Default: None (use global)
async def upgrade(self, ctx: "BaseMigrationContext") -> None:
"""Apply this migration."""
...
async def downgrade(self, ctx: "BaseMigrationContext") -> None:
"""Reverse this migration. May raise NotImplementedError."""
...
def get_checksum(self) -> str:
"""Return SHA-256 of entire migration source file.
Hashing only upgrade() misses changes to downgrade(), attributes,
or helper functions. The full file content is the only stable unit.
"""
...
@runtime_checkable
class RestrictedMigration(Protocol):
"""Protocol for semi-trusted pack migrations.
Identical to Migration but receives RestrictedContext (no raw conn).
Semi-trusted packs should implement this protocol instead of Migration.
The runner dispatches to the correct context type based on trust level.
"""
namespace: str
sequence: int
description: str
ONLINE_SAFE: bool
execution_mode: ExecutionMode
rollback_execution_mode: ExecutionMode
async def upgrade(self, ctx: "RestrictedContext") -> None: ...
async def downgrade(self, ctx: "RestrictedContext") -> None: ...
def get_checksum(self) -> str: ...
@dataclass
class MigrationDescriptor:
"""Concrete Migration implementation for programmatic construction.
Used by the YAML migration loader and test harnesses to create
Migration-protocol objects without authoring a full class.
"""
namespace: str
sequence: int
description: str
ONLINE_SAFE: bool = False
execution_mode: ExecutionMode = ExecutionMode.TRANSACTIONAL
rollback_execution_mode: ExecutionMode = ExecutionMode.TRANSACTIONAL
ROLLBACK_IDEMPOTENT: bool = False
DESTRUCTIVE_ROLLBACK: bool = False
STATEMENT_TIMEOUT_MS: int | None = None
LOCK_TIMEOUT_MS: int | None = None
MIGRATION_TIMEOUT_S: int | None = None
_upgrade_fn: Callable[["BaseMigrationContext"], Awaitable[None]] = field(repr=False)
_downgrade_fn: Callable[["BaseMigrationContext"], Awaitable[None]] | None = field(
default=None, repr=False
)
_source_path: Path | None = field(default=None, repr=False)
def __post_init__(self) -> None:
validate_execution_modes(self.execution_mode, self.rollback_execution_mode)
async def upgrade(self, ctx: "BaseMigrationContext") -> None:
await self._upgrade_fn(ctx)
async def downgrade(self, ctx: "BaseMigrationContext") -> None:
if self._downgrade_fn is None:
raise NotImplementedError(
f"Migration {self.namespace}/{self.sequence:04d} has no downgrade"
)
await self._downgrade_fn(ctx)
def get_checksum(self) -> str:
"""SHA-256 of the entire source file, not just upgrade().
Fallback for dynamically-constructed descriptors uses namespace
and sequence only — description is excluded because it may change
without affecting migration behavior, causing false checksum mismatches.
"""
if self._source_path and self._source_path.exists():
content = self._source_path.read_bytes()
else:
content = f"{self.namespace}:{self.sequence}".encode()
return hashlib.sha256(content).hexdigest()
@dataclass
class BaseMigrationContext:
"""Common migration context interface shared by all trust levels.
This is the type used in the Migration protocol's upgrade()/downgrade()
signatures. It provides access to batch execution, checkpoint management,
and namespace-scoped configuration — but NOT raw SQL access.
Trusted migrations receive a MigrationContext (which extends this with
conn/execute/fetch). Semi-trusted packs receive a RestrictedContext
(which provides only transform-library access).
"""
namespace: str
batch_executor: "BatchExecutor"
checkpoint_manager: "CheckpointManager"
ownership_validator: "ComponentOwnershipValidator"
logger: logging.Logger
dry_run: bool = False
@dataclass
class MigrationContext(BaseMigrationContext):
"""Full migration context for trusted packs.
Extends BaseMigrationContext with raw asyncpg.Connection access
and SQL execution helpers. Only provided to TRUSTED namespaces
(engine, stdlib).
"""
conn: asyncpg.Connection = field(default=None) # type: ignore[assignment]
async def execute(self, query: str, *args: Any) -> str:
"""Execute SQL with automatic statement_timeout enforcement."""
return await self.conn.execute(query, *args)
async def fetch(self, query: str, *args: Any) -> list[asyncpg.Record]:
"""Fetch rows with automatic statement_timeout enforcement."""
return await self.conn.fetch(query, *args)
@dataclass
class RestrictedContext(BaseMigrationContext):
"""Migration context for semi-trusted packs.
Extends BaseMigrationContext but does NOT add raw connection access.
Semi-trusted packs can only modify data through the transform library
(batch_executor, ownership_validator), which enforces ownership checks.
This closes the "sandbox illusion" where semi-trusted packs could bypass
ownership enforcement by using ctx.conn directly.
"""
@classmethod
def from_full_context(cls, ctx: MigrationContext) -> "RestrictedContext":
return cls(
namespace=ctx.namespace,
batch_executor=ctx.batch_executor,
checkpoint_manager=ctx.checkpoint_manager,
ownership_validator=ctx.ownership_validator,
logger=ctx.logger,
dry_run=ctx.dry_run,
)
Migration File Template (Class-Based)¶
# stdlib/0004_health_component_evolution.py
"""Add shield field and rename hp → current_health.
All transforms must be idempotent: safe to re-run on partial failure.
This ensures a safe downgrade path if the migration fails partway through.
"""
from pathlib import Path
import hashlib
from maid_engine.migrations.protocol import (
ExecutionMode, BaseMigrationContext,
)
from maid_engine.migrations.jsonb_transforms import (
rename_jsonb_field, add_jsonb_field, remove_jsonb_field,
)
class HealthComponentEvolution:
"""Migration implementing the Migration protocol directly."""
namespace = "stdlib"
sequence = 4
description = "Add shield field and rename hp → current_health"
ONLINE_SAFE = False
execution_mode = ExecutionMode.BATCHED
rollback_execution_mode = ExecutionMode.BATCHED
ROLLBACK_IDEMPOTENT = False
DESTRUCTIVE_ROLLBACK = False
STATEMENT_TIMEOUT_MS = 30_000
MIGRATION_TIMEOUT_S = 3600
async def upgrade(self, ctx: BaseMigrationContext) -> None:
await rename_jsonb_field(ctx, collection="entities",
component="HealthComponent",
old_name="hp", new_name="current_health")
await add_jsonb_field(ctx, collection="entities",
component="HealthComponent",
field_name="shield", default_value=0)
async def downgrade(self, ctx: BaseMigrationContext) -> None:
await remove_jsonb_field(ctx, collection="entities",
component="HealthComponent",
field_name="shield")
await rename_jsonb_field(ctx, collection="entities",
component="HealthComponent",
old_name="current_health", new_name="hp")
def get_checksum(self) -> str:
"""SHA-256 of the entire source file — not just upgrade()."""
source_path = Path(__file__)
return hashlib.sha256(source_path.read_bytes()).hexdigest()
# Module-level instance for the loader to discover
migration = HealthComponentEvolution()
Migration loader: All migrations must be class-based with a module-level migration
instance. Module-level function-based migrations are not supported.
# In the migration loader:
module = importlib.import_module(module_name)
if not hasattr(module, "migration"):
raise MigrationLoadError(
f"Migration module {module_name} must define a module-level "
f"'migration' instance implementing the Migration protocol"
)
mig = module.migration
assert isinstance(mig, Migration) # runtime_checkable
validate_execution_modes(mig.execution_mode, mig.rollback_execution_mode)
Alembic Integration with asyncpg¶
Decision: Custom Runner Instead of Alembic¶
After thorough analysis, we rejected Alembic integration in favor of a custom migration runner:
Alembic limitations for MAID:
-
SQLAlchemy dependency conflict: MAID deliberately uses raw asyncpg. Alembic requires SQLAlchemy, adding ~3MB of dependencies and creating an impedance mismatch with existing queries.
-
Single-app assumption: Alembic's branch/merge model assumes a monolithic application. MAID's multi-namespace content pack architecture doesn't map well to Alembic's revision DAG.
-
No JSONB evolution support: Alembic has no opinions about document schema evolution. We'd need custom operations regardless.
-
Content pack dynamic loading: Alembic expects all migration files to be known at configuration time. MAID loads content packs dynamically based on runtime configuration.
Custom Runner Architecture¶
Separated responsibilities: Discovery, planning, and execution are split into
distinct components. A single connection pool is created by the CLI (or test harness)
and injected into all components — DatabaseManager, DocumentStore, and
MigrationRunner never create their own pools.
Core components:
- PackManifestRegistry: Discovers content packs and their migration sources
- MigrationPlanner: Dependency resolution, topological ordering, checksum verification
- MigrationRunner: Execution only — lock acquisition, transaction management, audit
- BatchExecutor: Generic batched operation runner with checkpoint support (reusable)
- MigrationHistory: Tracks applied migrations with audit log
- JSONBTransformLibrary: Safe document evolution via
JSONBUpdateSpecbuilder - CheckpointManager: Tracks progress for resumable batched operations
Integration with asyncpg (pool injection):
# Connection pool is created ONCE and injected everywhere
pool = await asyncpg.create_pool(dsn=settings.database.dsn)
# All components receive the pool — none create their own.
# Existing pool-creating components (DatabaseManager, DocumentStore)
# gain from_pool() constructors to accept an externally-created pool:
#
# db_manager = DatabaseManager.from_pool(pool)
# doc_store = PostgresDocumentStore.from_pool(pool)
#
# This avoids double-pool creation while preserving backward compat.
registry = PackManifestRegistry()
planner = MigrationPlanner(registry)
history = MigrationHistory(pool)
runner = MigrationRunner(pool, planner, history)
class MigrationRunner:
"""Executes migration plans. Does NOT discover or order migrations."""
def __init__(
self,
connection_pool: asyncpg.Pool,
planner: MigrationPlanner,
history: MigrationHistory,
) -> None:
self._pool = connection_pool
self._planner = planner
self._history = history
async def migrate(
self,
target: str | None = None,
namespace: str | None = None,
dry_run: bool = False,
online_only: bool = False,
) -> MigrationResult:
# Verify all applied migration checksums before proceeding
await self._history.verify_checksums(self._planner.get_all_migrations())
plan = await self._planner.create_plan(
applied=await self._history.get_all_applied(),
target=target,
namespace=namespace,
)
if dry_run:
return self._simulate_migration(plan)
# Acquire per-namespace advisory locks (allows parallel independent packs)
async with self._acquire_namespace_locks(plan.namespaces):
# Validate cross-namespace requirements UNDER LOCK to prevent
# races where a parallel rollback removes a dependency between
# our plan creation and execution.
await self._validate_requirements_under_lock(plan)
return await self._execute_migration_plan(plan, online_only)
async def _validate_requirements_under_lock(
self,
plan: MigrationPlan,
) -> None:
"""Re-check migration_requires after acquiring locks.
Per-namespace parallel locking means another process could roll back
a dependency namespace between plan creation and execution. This
validation closes that race.
"""
applied = await self._history.get_all_applied()
for migration in plan.migrations:
source = self._planner.get_source(migration.namespace)
if source and source.migration_requires:
for req_ns, req_seq in source.migration_requires.items():
applied_seq = applied.get(req_ns, 0)
if applied_seq < req_seq:
raise MigrationDependencyError(
f"Namespace '{migration.namespace}' requires "
f"{req_ns}/{req_seq:04d} but only "
f"{req_ns}/{applied_seq:04d} is applied"
)
async def _execute_single_migration(
self,
migration: Migration,
) -> None:
"""Execute one migration with timeout enforcement and transaction management.
History recording strategy:
- TRANSACTIONAL: history INSERT inside the same transaction (atomic).
- BATCHED/NO_TRANSACTION: two-phase approach — insert 'pending' before
execution, update to 'applied' after. On crash, 'pending' records
signal incomplete migrations that need investigation.
"""
ctx = MigrationContext(
conn=..., # acquired from pool
namespace=migration.namespace,
batch_executor=BatchExecutor(self._pool),
checkpoint_manager=CheckpointManager(self._pool),
ownership_validator=ComponentOwnershipValidator(self._planner.registry),
logger=logger.bind(migration=f"{migration.namespace}/{migration.sequence:04d}"),
)
# Semi-trusted packs receive RestrictedContext (no raw conn access)
trust = get_trust_level(migration.namespace)
effective_ctx: MigrationContext | RestrictedContext
if trust == MigrationTrustLevel.SEMI_TRUSTED:
effective_ctx = RestrictedContext.from_full_context(ctx)
else:
effective_ctx = ctx
# Apply per-migration timeouts — validate as int to prevent injection.
#
# IMPORTANT: SET LOCAL is only effective within a transaction block.
# For BATCHED and NO_TRANSACTION modes, we use SET SESSION and reset
# in a finally block. For TRANSACTIONAL mode, SET LOCAL scopes the
# timeout to the migration's transaction.
timeout = migration.MIGRATION_TIMEOUT_S or self._default_timeout
exec_mode = getattr(
migration, "execution_mode", ExecutionMode.TRANSACTIONAL
)
async with asyncio.timeout(timeout):
if exec_mode == ExecutionMode.TRANSACTIONAL:
# --- TRANSACTIONAL: single atomic transaction ---
# History INSERT is inside the same transaction for atomicity.
# SET LOCAL scopes timeout to this transaction.
async with ctx.conn.transaction():
if migration.STATEMENT_TIMEOUT_MS is not None:
timeout_ms = int(migration.STATEMENT_TIMEOUT_MS)
await ctx.conn.execute(
"SET LOCAL statement_timeout = $1", str(timeout_ms)
)
if migration.LOCK_TIMEOUT_MS is not None:
lock_ms = int(migration.LOCK_TIMEOUT_MS)
await ctx.conn.execute(
"SET LOCAL lock_timeout = $1", str(lock_ms)
)
await migration.upgrade(effective_ctx)
await self._history.record_applied(
ctx.conn, migration, status="applied"
)
# NOTIFY inside the transaction — delivered on commit
await ctx.conn.execute(
"SELECT pg_notify('maid_schema_changes', $1)",
json.dumps({
"namespace": migration.namespace,
"sequence": migration.sequence,
"action": "applied",
}),
)
else:
# --- BATCHED / NO_TRANSACTION: two-phase history ---
# Phase 1: Record 'pending' status before execution.
# On crash, 'pending' records indicate incomplete migrations.
await self._history.record_applied(
ctx.conn, migration, status="pending"
)
# SET SESSION for non-transactional modes (reset in finally)
try:
if migration.STATEMENT_TIMEOUT_MS is not None:
timeout_ms = int(migration.STATEMENT_TIMEOUT_MS)
await ctx.conn.execute(
"SET SESSION statement_timeout = $1", str(timeout_ms)
)
if migration.LOCK_TIMEOUT_MS is not None:
lock_ms = int(migration.LOCK_TIMEOUT_MS)
await ctx.conn.execute(
"SET SESSION lock_timeout = $1", str(lock_ms)
)
await migration.upgrade(effective_ctx)
finally:
# Always reset session-level timeouts
await ctx.conn.execute("RESET statement_timeout")
await ctx.conn.execute("RESET lock_timeout")
# Phase 2: Update status to 'applied' after successful execution
await self._history.update_status(
ctx.conn, migration, status="applied"
)
# NOTIFY after successful execution
await ctx.conn.execute(
"SELECT pg_notify('maid_schema_changes', $1)",
json.dumps({
"namespace": migration.namespace,
"sequence": migration.sequence,
"action": "applied",
}),
)
JSONB Schema Evolution¶
The Document Evolution Challenge¶
MAID stores all game content as JSONB documents in the documents table, discriminated by collection. Component models are Pydantic BaseModel subclasses with extra="forbid", meaning field mismatches between stored data and current model cause ValidationError on load.
Evolution scenarios requiring migrations:
| Change Type | Migration Required? | Strategy | Risk Level |
|---|---|---|---|
| New optional field | No | Pydantic default population | Low |
| New required field | Yes | Backfill with computed value | Medium |
| Field renamed | Yes | JSONB key rename | Medium |
| Field type changed | Yes | Value transformation | High |
| Field removed | Yes (if extra="forbid") |
Key removal | Low |
| Nested structure change | Yes | Path restructuring | High |
| Component ownership transfer | Yes | Cross-namespace coordination | High |
Safe JSONB Transform Library¶
Core principle: No raw SQL interpolation. All JSONB updates use a structured builder.
# jsonb_transforms.py
class TransformFunction(Enum):
"""Predefined safe transform expressions for JSONB field transformations.
Replaces raw SQL expression strings (transform_expr) with an enumerated
set of known-safe transformations. This eliminates the SQL injection risk
of accepting arbitrary expression strings while covering common transform
patterns. If a new transform is needed, add it here with a reviewed SQL
implementation rather than allowing ad-hoc expressions.
"""
TO_INTEGER = "to_integer" # Cast text/float to integer
TO_FLOAT = "to_float" # Cast text/integer to float
TO_STRING = "to_string" # Cast any scalar to text
TO_BOOLEAN = "to_boolean" # Cast to boolean (truthy/falsy)
TO_ARRAY_WRAP = "to_array_wrap" # Wrap scalar value in a JSON array
EPOCH_TO_ISO = "epoch_to_iso" # Convert Unix epoch to ISO 8601 string
NULLIFY = "nullify" # Set field to null
@dataclass(frozen=True)
class JSONBUpdateSpec:
"""Structured specification for a JSONB update operation.
Implements the BatchOperation protocol. Generates parameterized SQL
internally — callers never write raw SQL. This eliminates the SQL
injection vector of raw where_clause/update_clause strings passed
via f-strings.
"""
collection: str
component: str
operation: Literal["rename", "add", "remove", "transform"]
field_name: str
new_field_name: str | None = None
default_value: Any = None
transform_fn: "TransformFunction | None" = None # Type-safe transform selection
@property
def operation_key(self) -> str:
"""Stable identifier for checkpoint hashing."""
return f"{self.collection}:{self.component}:{self.operation}:{self.field_name}"
def to_where_sql(self, param_offset: int) -> tuple[str, list[Any]]:
"""Generate parameterized WHERE clause. Returns (sql, params)."""
path = f"$.components.{self.component}.{self.field_name}"
return (
f"jsonb_path_exists(data, ${param_offset}::jsonpath)",
[path],
)
def to_update_sql(self, param_offset: int) -> tuple[str, list[Any]]:
"""Generate parameterized UPDATE SET clause. Returns (sql, params)."""
if self.operation == "rename":
old_path = f"$.components.{self.component}.{self.field_name}"
remove_path = ["components", self.component, self.field_name]
add_path = ["components", self.component, self.new_field_name]
return (
f"data = jsonb_set("
f"data #- ${param_offset}::text[], "
f"${param_offset + 1}::text[], "
f"jsonb_path_query_first(data, ${param_offset + 2}::jsonpath))",
[remove_path, add_path, old_path],
)
# ... other operations similarly parameterized
async def rename_jsonb_field(
ctx: BaseMigrationContext,
*,
collection: str,
component: str,
old_name: str,
new_name: str,
batch_size: int = 5000,
) -> int:
"""Rename field in component JSONB data with injection protection."""
# Validate component and field names to prevent injection
if not _is_valid_identifier(component):
raise ValueError(f"Invalid component name: {component}")
if not _is_valid_identifier(old_name):
raise ValueError(f"Invalid field name: {old_name}")
if not _is_valid_identifier(new_name):
raise ValueError(f"Invalid field name: {new_name}")
spec = JSONBUpdateSpec(
collection=collection,
component=component,
operation="rename",
field_name=old_name,
new_field_name=new_name,
)
return await ctx.batch_executor.execute(
spec, batch_size=batch_size,
)
def _is_valid_identifier(name: str) -> bool:
"""Validate identifier for JSONB path safety."""
return bool(re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', name))
BatchExecutor¶
The BatchExecutor is a generic, reusable batched operation runner extracted from
the JSONB transform internals. It can drive any migration type that operates on
rows in batches.
Key fixes over the prior design:
- Cursor/keyset pagination (WHERE id > $last_id ORDER BY id LIMIT $batch) instead of OFFSET, which skips rows on crash-resume when the WHERE condition changes on already-modified rows.
- Deterministic checkpoint IDs using hashlib.sha256 instead of Python's hash(), which is randomized per-process and would generate different checkpoint IDs on restart. The hash includes namespace+sequence to avoid collisions across migrations.
- Atomic mini-transactions wrapping each batch's data update and checkpoint update together, so they never drift apart.
- Idempotent transforms required: All batched transform functions must be safe to re-run on a partially-migrated dataset. This ensures that a crash mid-batch followed by resume does not corrupt data and provides a safe downgrade path for partial migration failures.
# batch_executor.py
@runtime_checkable
class BatchOperation(Protocol):
"""Protocol for any operation that BatchExecutor can drive.
Extracted so BatchExecutor is genuinely generic — not coupled to
JSONBUpdateSpec. Implementations include JSONBUpdateSpec (JSONB transforms),
DDLBackfillSpec (DDL backfills), and DataSeedSpec (data seeding).
"""
def to_where_sql(self, param_offset: int) -> tuple[str, list[Any]]:
"""Generate parameterized WHERE clause."""
...
def to_update_sql(self, param_offset: int) -> tuple[str, list[Any]]:
"""Generate parameterized UPDATE SET clause."""
...
@property
def table(self) -> str:
"""Target table for the operation.
Decouples BatchExecutor from a hardcoded 'documents' table.
Implementations return their target table name (e.g., 'documents',
'accounts', or a custom table).
"""
...
@property
def collection(self) -> str | None:
"""Optional collection discriminator within the table.
For the documents table, this filters by the collection column.
For tables without a collection concept, return None.
"""
...
@property
def cursor_column(self) -> str:
"""Column used for keyset pagination.
Must be a monotonically increasing column (e.g., a BIGSERIAL
doc_seq or created_at timestamp), NOT a random UUID. Random UUIDs
have no natural ordering and produce unpredictable pagination
behavior. Defaults to 'doc_seq' for the documents table.
"""
...
@property
def operation_key(self) -> str:
"""Stable string identifying this operation for checkpoint hashing."""
...
class BatchExecutor:
"""Generic batched operation runner with checkpoint support.
Reusable for JSONB transforms, DDL backfills, data seeding, or
any migration that processes rows incrementally.
"""
def __init__(self, pool: asyncpg.Pool) -> None:
self._pool = pool
async def execute(
self,
spec: BatchOperation,
*,
batch_size: int = 5000,
namespace: str = "",
sequence: int = 0,
) -> int:
checkpoint_id = self._deterministic_checkpoint_id(
spec, namespace=namespace, sequence=sequence
)
checkpoint = await CheckpointManager.get_or_create(
self._pool, checkpoint_id
)
total_updated = checkpoint.processed_count
last_seen_cursor = checkpoint.last_seen_cursor # Monotonic cursor value
table = spec.table
cursor_col = spec.cursor_column
while True:
# --- Atomic mini-transaction: data update + checkpoint ---
async with self._pool.acquire() as conn:
async with conn.transaction():
where_sql, where_params = spec.to_where_sql(param_offset=3)
update_sql, update_params = spec.to_update_sql(
param_offset=3 + len(where_params)
)
# Build collection filter if applicable
collection_filter = ""
collection_params: list[Any] = []
if spec.collection is not None:
collection_filter = "AND collection = $1"
collection_params = [spec.collection]
# Keyset pagination: WHERE cursor_col > $last ORDER BY cursor_col
# cursor_col MUST be monotonically increasing (BIGSERIAL or
# timestamp), not a random UUID. Random UUIDs have no natural
# ordering — ORDER BY uuid produces arbitrary results and
# keyset pagination becomes unreliable.
cursor_param_idx = 2 if spec.collection is not None else 1
query = f"""
WITH batch AS (
SELECT {cursor_col} FROM {table}
WHERE {cursor_col} > ${cursor_param_idx}
{collection_filter}
AND ({where_sql})
ORDER BY {cursor_col}
LIMIT ${cursor_param_idx + 1 + len(where_params) + len(update_params)}
)
UPDATE {table} SET
{update_sql},
updated_at = NOW()
FROM batch
WHERE {table}.{cursor_col} = batch.{cursor_col}
RETURNING {table}.{cursor_col}
"""
all_params = [
*collection_params,
last_seen_cursor,
*where_params,
*update_params,
batch_size,
]
rows = await conn.fetch(query, *all_params)
rows_affected = len(rows)
if rows_affected > 0:
last_seen_cursor = rows[-1][cursor_col]
total_updated += rows_affected
# Checkpoint updated IN THE SAME transaction
await CheckpointManager.update_progress(
conn, checkpoint_id,
processed_count=total_updated,
last_seen_cursor=last_seen_cursor,
)
# --- End atomic mini-transaction ---
if rows_affected < batch_size:
break
# Yield control between batches
await asyncio.sleep(0.1)
await CheckpointManager.complete(self._pool, checkpoint_id)
return total_updated
@staticmethod
def _deterministic_checkpoint_id(
spec: BatchOperation,
*,
namespace: str = "",
sequence: int = 0,
) -> str:
"""Derive checkpoint ID from namespace, sequence, and spec content.
Python's built-in hash() is randomized per-process (PYTHONHASHSEED),
so it would produce different checkpoint IDs on restart, breaking
crash-resume. SHA-256 is deterministic and stable.
Namespace and sequence are included to avoid collisions when
two migrations use the same operation on the same collection.
"""
content = f"{namespace}:{sequence}:{spec.operation_key}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
Document Version Stamping¶
Optional but recommended for complex packs:
async def upgrade_with_versioning(ctx: BaseMigrationContext) -> None:
"""Transform HealthComponent with schema version tracking."""
await rename_jsonb_field(
ctx,
collection="entities",
component="HealthComponent",
old_name="hp",
new_name="current_health",
)
await add_jsonb_field(
ctx,
collection="entities",
component="HealthComponent",
field_name="shield",
default_value=0,
)
# Stamp schema version for tracking
await set_schema_version(
ctx,
collection="entities",
version="2.0",
component_filter="HealthComponent",
)
Batched Operations with Transactional Safety¶
Critical design decision: Explicit execution mode control
# For small collections (< 50k documents): Single transaction (default)
execution_mode = ExecutionMode.TRANSACTIONAL
async def upgrade(self, ctx: BaseMigrationContext) -> None:
"""Single-transaction migration (atomic, brief lock)."""
await rename_jsonb_field(ctx, ...) # All-or-nothing
# For large collections: Checkpointed batching
execution_mode = ExecutionMode.BATCHED
# Rollback declarations — batched upgrades need batched rollbacks
rollback_execution_mode = ExecutionMode.BATCHED
ROLLBACK_IDEMPOTENT = False # Set True if downgrade() is safe to re-run
async def upgrade(self, ctx: BaseMigrationContext) -> None:
"""Batched migration with checkpoint recovery."""
# This runs OUTSIDE transaction for resumability
# Each batch commits atomically with its checkpoint update
await rename_jsonb_field(ctx, ..., batch_size=5000)
async def downgrade(self, ctx: BaseMigrationContext) -> None:
"""Batched rollback — mirrors upgrade batching strategy."""
# rollback_execution_mode = BATCHED ensures the runner uses BatchExecutor
await rename_jsonb_field(ctx, ..., batch_size=5000)
Batching trade-offs explicitly documented:
- Single transaction: Atomic rollback, brief table lock, limited to ~50k documents
- Checkpointed batching: Resumable on failure, longer operation time, eventual consistency
Content Pack Migration Support¶
Content Pack Discovery¶
Problem resolution: CLI cannot discover packs without engine startup
The original design assumed CLI commands could discover content packs without starting the full GameEngine. This created a circular dependency — content packs are loaded by the engine, but migrations need to run before engine startup.
Solution: Content Pack Manifest Registry
# packages/maid-engine/src/maid_engine/migrations/discovery.py
from maid_engine.plugins.protocol import ContentPackManifest as BaseManifest
@dataclass
class MigrationManifest(BaseManifest):
"""Extends the existing ContentPackManifest with migration-specific fields.
Does NOT redefine ContentPackManifest — extends the canonical definition
from maid_engine.plugins.protocol to avoid type duplication.
"""
migration_requires: dict[str, int] | None = None
# e.g., {"engine": 3, "stdlib": 2}
components: list[str] = field(default_factory=list)
# Components owned by this pack, used to build ownership map dynamically.
# e.g., ["HealthComponent", "PositionComponent"] for stdlib
class PackManifestRegistry:
"""Discovers content packs without instantiating them."""
def __init__(self) -> None:
self._manifests: dict[str, MigrationManifest] = {}
async def discover_available_packs(self) -> dict[str, MigrationManifest]:
"""Find all installed content packs via entry points."""
manifests = {}
# Discover via setuptools entry points
for entry_point in entry_points(group="maid.content_packs"):
try:
pack_module = entry_point.load()
manifest_data = getattr(pack_module, "__manifest__", None)
if manifest_data:
manifest = MigrationManifest(**manifest_data)
manifests[manifest.name] = manifest
except Exception as e:
logger.warning(f"Failed to load pack {entry_point.name}: {e}")
return manifests
async def discover_migration_sources(
self,
enabled_packs: list[str] | None = None,
) -> list[MigrationSource]:
"""Discover migration sources for enabled packs.
If enabled_packs is None, only first-party packs are enabled by
default. Community packs must be explicitly listed to prevent
accidentally running migrations from all installed packages.
"""
manifests = await self.discover_available_packs()
if enabled_packs is None:
# Default: only first-party packs, not all discovered
enabled_packs = [
name for name, m in manifests.items()
if get_trust_level(name) != MigrationTrustLevel.UNTRUSTED
]
# Resolve load order via topological sort
load_order = self._resolve_dependency_order(
manifests, enabled_packs
)
sources = []
for pack_name in load_order:
manifest = manifests[pack_name]
migration_dir = self._find_migration_directory(pack_name)
if migration_dir and migration_dir.exists():
sources.append(MigrationSource(
namespace=pack_name,
directory=migration_dir,
dependencies=manifest.dependencies,
migration_requires=manifest.migration_requires,
))
return sources
def _find_migration_directory(self, pack_name: str) -> Path | None:
"""Locate migration directory for a pack via entry points.
Uses the 'maid.migrations' entry point group rather than
convention-based path guessing. This decouples migration discovery
from package layout and avoids fragile assumptions about directory
structure (e.g., 'maid_{name}/migrations/').
"""
# Primary: entry point declaration (preferred)
for ep in entry_points(group="maid.migrations"):
if ep.name == pack_name:
try:
migrations_module = ep.load()
return Path(migrations_module.__path__[0])
except Exception:
pass
# Fallback: importlib convention (for first-party packs only)
try:
module_name = f"maid_{pack_name.replace('-', '_')}"
spec = importlib.util.find_spec(module_name)
if not spec or not spec.origin:
return None
package_dir = Path(spec.origin).parent
migration_dir = package_dir / "migrations"
return migration_dir if migration_dir.is_dir() else None
except (ImportError, AttributeError):
return None
# Reserved namespaces that community packs cannot shadow
RESERVED_NAMESPACES = frozenset({
"engine", "stdlib", "classic-rpg", "tutorial-world",
"_internal", "maid", "core", "system",
})
def validate_namespace(self, namespace: str) -> None:
"""Prevent community packs from shadowing reserved namespaces."""
trust = get_trust_level(namespace)
if (
trust == MigrationTrustLevel.UNTRUSTED
and namespace in self.RESERVED_NAMESPACES
):
raise SecurityError(
f"Namespace '{namespace}' is reserved and cannot be used "
f"by community content packs"
)
Content pack integration:
# packages/maid-stdlib/src/maid_stdlib/__init__.py
__manifest__ = {
"name": "stdlib",
"version": "1.0.0",
"dependencies": ["engine"],
"migration_requires": {"engine": 2},
"components": [
"PositionComponent",
"HealthComponent",
"ManaComponent",
"InventoryComponent",
"DialogueComponent",
"ExtendedRoomComponent",
],
}
Namespace Dependency Resolution¶
Strict ordering with validation:
Migrations are executed in two-level order: first by namespace (topological sort based on content pack dependencies), then by sequence within each namespace (ascending integer order). This guarantees that within a single namespace, migrations always execute in sequence order (0001, 0002, 0003...) and never out of order.
class DependencyResolver:
"""Resolves migration execution order across namespaces."""
def resolve_migration_order(
self,
sources: list[MigrationSource],
) -> list[str]:
"""Return namespace execution order via topological sort."""
# Build dependency graph
graph = {}
in_degree = {}
for source in sources:
graph[source.namespace] = source.dependencies
in_degree[source.namespace] = 0
for source in sources:
for dep in source.dependencies:
if dep not in graph:
raise MigrationDependencyError(
f"Namespace '{source.namespace}' depends on "
f"'{dep}' which is not available"
)
in_degree[source.namespace] += 1
# Kahn's algorithm for topological sort
queue = [ns for ns, degree in in_degree.items() if degree == 0]
result = []
while queue:
current = queue.pop(0)
result.append(current)
for neighbor in graph:
if current in graph[neighbor]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(result) != len(sources):
raise MigrationDependencyError("Circular dependency detected")
return result
Component Ownership Enforcement¶
Problem: Packs can modify other packs' components
The previous design used _extract_component_references with regex matching
on raw SQL strings to detect ownership violations. This is fundamentally unsound:
regex cannot reliably parse SQL, and any new SQL pattern or comment would evade it.
Solution: API-level enforcement. The JSONB transform library is the only way
to touch component data in migrations. Ownership is validated at the call site —
every rename_jsonb_field, add_jsonb_field, etc. checks the calling migration's
namespace against the component ownership map before generating any SQL.
class ComponentOwnershipValidator:
"""Enforces component ownership at the transform API boundary."""
def __init__(self, pack_registry: PackManifestRegistry) -> None:
self._registry = pack_registry
self._ownership_map = self._build_ownership_map()
def _build_ownership_map(self) -> dict[str, str]:
"""Map component types to owning namespaces.
Built dynamically from pack manifest `components` fields rather
than hardcoded. Each pack declares which components it owns in
its __manifest__. If two packs claim the same component, the
first one in dependency order wins (with a warning logged).
"""
ownership: dict[str, str] = {}
for name, manifest in self._registry._manifests.items():
for component in getattr(manifest, "components", []):
if component in ownership:
logger.warning(
f"Component '{component}' claimed by both "
f"'{ownership[component]}' and '{name}'; "
f"keeping '{ownership[component]}'"
)
else:
ownership[component] = name
return ownership
def check_access(
self,
namespace: str,
component: str,
) -> None:
"""Called by every JSONB transform function before executing.
Raises OwnershipViolationError if the calling namespace does not
own the target component. This is the ONLY enforcement point —
no regex-based SQL scanning is used.
"""
owner = self._ownership_map.get(component)
if owner and owner != namespace:
raise OwnershipViolationError(
f"Namespace '{namespace}' cannot modify '{component}' "
f"owned by '{owner}'"
)
# Enforcement wired into every transform function:
async def rename_jsonb_field(
ctx: BaseMigrationContext,
*,
collection: str,
component: str,
old_name: str,
new_name: str,
batch_size: int = 5000,
) -> int:
# Ownership check happens HERE, at the API boundary
ctx.ownership_validator.check_access(ctx.namespace, component)
# ... proceed with validated, parameterized operation
spec = JSONBUpdateSpec(
collection=collection,
component=component,
operation="rename",
field_name=old_name,
new_field_name=new_name,
)
return await ctx.batch_executor.execute(spec, batch_size=batch_size)
This design means migrations that bypass the transform library and write raw SQL
against the documents table are prohibited for community packs (see Security Model
below). Trusted namespaces (engine, stdlib) may use raw SQL but are expected
to self-enforce ownership conventions.
CLI Commands¶
Implemented Command Interface¶
All commands are implemented in packages/maid-engine/src/maid_engine/cli/commands/db_migrate.py
as synchronous Typer functions that wrap async internals via asyncio.run().
# Migration execution
maid db migrate [--target=N] [--namespace=NS] [--dry-run] [--online-safe] [--backup] [--enabled-packs=LIST] [--format=json|text]
maid db rollback --namespace=NS [--steps=N] [--dry-run] [--force] [--format=json|text]
# Status and introspection
maid db status [--namespace=NS] [--show-pending] [--show-checkpoints] [--format=json|text]
maid db history [--namespace=NS] [--limit=N] [--include-rollbacks] [--format=json|text]
# Migration authoring
maid db create DESCRIPTION --namespace=NS [--template=jsonb|ddl|seed] [--collection=NAME] [--component=NAME] [--field=NAME]
# Validation and maintenance
maid db validate [--fix-checksums] [--cleanup-checkpoints] [--format=json|text]
maid db repair [--clear-pending] [--cleanup-checkpoints] [--fix-checksums]
maid db baseline --namespace=NS --at-version=N [--force]
maid db cleanup --namespace=NS
maid db init [--force]
Command details:
| Command | Description |
|---|---|
migrate |
Run pending migrations with advisory lock, checksum verification, and TODO-placeholder rejection |
rollback |
Roll back migrations for a namespace with destructive-rollback prompts and --force safety |
status |
Show per-namespace migration status (current version, pending count, optional checkpoint info) |
history |
Show applied migration log with optional rollback entries |
create |
Generate new migration files from ddl, jsonb, or seed templates with auto-sequencing |
validate |
Verify checksum integrity, optionally fix mismatches and clean stale checkpoints |
repair |
Clear stuck pending migrations, clean orphaned checkpoints, recompute checksums |
baseline |
Mark an existing database as migrated to a given version (for adopting the migration system on existing deployments) |
cleanup |
Remove migration history and rollback log entries for a decommissioned namespace |
init |
Create the migration infrastructure tables (_migration_history, _migration_rollback_log, _migration_checkpoints) |
Naming conventions:
- --online-safe — boolean filter for safe-to-run-live migrations
- --dry-run — always available on execution commands, shows plan without executing
- --format=json|text — machine-readable output for CI/CD pipelines (default: text)
- --enabled-packs — comma-separated pack name override (defaults to config/env MAID_ENABLED_CONTENT_PACKS)
Content Pack Discovery in CLI¶
Implementation in db_migrate.py:
The CLI resolves enabled content packs from the MAID_ENABLED_CONTENT_PACKS environment
variable or from settings.migration.enabled_packs. A single asyncpg pool is created per
command invocation and injected into all components. The migrate command additionally
rejects migration files containing unresolved "TODO" placeholders.
# packages/maid-engine/src/maid_engine/cli/commands/db_migrate.py
def migrate(
target: int | None = typer.Option(None, "--target", help="Target sequence"),
namespace: str | None = typer.Option(None, "--namespace", help="Namespace filter"),
dry_run: bool = typer.Option(False, "--dry-run", help="Show plan without executing"),
online_safe: bool = typer.Option(False, "--online-safe", help="Only ONLINE_SAFE migrations"),
enabled_packs: str | None = typer.Option(None, "--enabled-packs", help="Comma-separated pack names"),
fmt: str = typer.Option("text", "--format", help="Output format: text|json"),
backup: bool = typer.Option(False, "--backup", help="Backup reminder before executing"),
) -> None:
"""Run pending database migrations."""
async def _run() -> None:
pool = await _create_pool()
try:
history = MigrationHistory(pool)
await history.ensure_infrastructure()
registry = PackManifestRegistry()
sources = await registry.discover_migration_sources(enabled_packs=packs)
planner = MigrationPlanner(registry)
await planner.load_migrations(sources)
runner = MigrationRunner(pool, planner, history)
result = await runner.migrate(
target=target, namespace=namespace,
dry_run=dry_run, online_only=online_safe,
enabled_packs=packs,
)
_display_migration_result(result, fmt)
finally:
await pool.close()
asyncio.run(_run())
def _get_enabled_packs_from_config() -> list[str]:
"""Read enabled content packs from configuration.
Checks MAID_ENABLED_CONTENT_PACKS env var first, then falls back
to settings.migration.enabled_packs.
"""
env = os.environ.get("MAID_ENABLED_CONTENT_PACKS")
if env:
return [p.strip() for p in env.split(",") if p.strip()]
settings = get_settings()
return list(settings.migration.enabled_packs)
CLI Output Examples¶
maid db status --show-pending
Migration Status
┌─────────────────┬─────────┬──────────┬─────────────────────┬─────────────┐
│ Namespace │ Current │ Pending │ Last Applied │ Status │
├─────────────────┼─────────┼──────────┼─────────────────────┼─────────────┤
│ engine │ 0002 │ - │ 2025-01-14 10:30:15 │ ✓ Up to date│
│ stdlib │ 0003 │ 0004 │ 2025-01-14 10:30:18 │ ⚠ 1 pending │
│ classic-rpg │ 0005 │ 0006,007 │ 2025-01-14 10:25:10 │ ⚠ 2 pending │
│ tutorial-world │ - │ 0001,002 │ Never applied │ ⚠ 2 pending │
└─────────────────┴─────────┴──────────┴─────────────────────┴─────────────┘
Checkpoints in Progress:
classic-rpg/0005: Batched update 45,230/67,890 documents (66% complete)
maid db migrate --dry-run --namespace stdlib
Migration Plan (DRY RUN)
Target: Latest (stdlib namespace only)
Enabled Content Packs: engine, stdlib, classic-rpg, tutorial-world
┌─────────────────────────────────────────────────────────────────────────────┐
│ Migration: stdlib/0004_health_component_evolution │
├─────────────────────────────────────────────────────────────────────────────┤
│ Description: Add shield field and rename hp → current_health │
│ Type: JSONB transformation (requires batching) │
│ Safety: ONLINE_SAFE = False (breaking change) │
│ Estimated affected documents: ~45,230 entities │
│ │
│ Operations: │
│ 1. Rename HealthComponent.hp → current_health │
│ 2. Add HealthComponent.shield = 0 (default) │
│ 3. Set schema version = "2.0" │
│ │
│ Rollback: Available (strips shield, reverts field name) │
│ ⚠ WARNING: Rolling back after deployment will lose shield values │
└─────────────────────────────────────────────────────────────────────────────┘
Execution Plan:
1. Acquire advisory lock (timeout: 30s)
2. Create migration checkpoint for batch tracking
3. Execute batched JSONB transform (5,000 docs/batch)
4. Record migration as applied
5. Clean up checkpoint
Continue? [y/N]:
Rollback Strategy¶
Safe Rollback Implementation¶
Multi-tier rollback safety:
class RollbackStrategy:
"""Coordinates safe migration rollback with dependency validation."""
async def plan_rollback(
self,
namespace: str,
steps: int = 1,
force: bool = False,
) -> RollbackPlan:
"""Create rollback plan with safety validation."""
# Get applied migrations in reverse order
applied = await self._history.get_applied_migrations(
namespace, limit=steps
)
errors = []
# Check for dependent migrations in other namespaces
for migration in applied:
dependent_migrations = await self._find_dependent_migrations(
namespace, migration.sequence
)
if dependent_migrations and not force:
errors.append(
f"Cannot rollback {namespace}/{migration.sequence:04d}: "
f"migrations {dependent_migrations} depend on it"
)
# Validate rollback implementations exist
for migration in applied:
if not migration.has_downgrade and not force:
errors.append(
f"Migration {namespace}/{migration.sequence:04d} "
f"has no downgrade() implementation"
)
# Check for destructive rollbacks
destructive = []
for migration in applied:
if await self._is_destructive_rollback(migration):
destructive.append(migration)
return RollbackPlan(
migrations=applied,
errors=errors,
destructive=destructive,
requires_force=(bool(errors) or bool(destructive)),
)
async def _find_dependent_migrations(
self,
namespace: str,
sequence: int,
) -> list[str]:
"""Find migrations that depend on target migration.
Uses the migration_requires map for semantic dependency checking,
not temporal ordering (applied_at timestamps), which is fragile
when migrations are applied out of order or re-applied.
"""
dependent = []
for other_ns in await self._history.get_namespaces():
if other_ns == namespace:
continue
manifest = await self._registry.get_manifest(other_ns)
# Check semantic dependency via migration_requires
required_seq = (manifest.migration_requires or {}).get(namespace)
if required_seq is not None and required_seq >= sequence:
# This namespace requires our migration at or above the
# sequence being rolled back — check if it has applied migrations
latest = await self._history.get_latest(other_ns)
if latest:
dependent.append(f"{other_ns}/{latest.sequence:04d}")
return dependent
async def _is_destructive_rollback(
self, migration: AppliedMigration
) -> bool:
"""Check if rollback would destroy user data.
Uses the declarative DESTRUCTIVE_ROLLBACK attribute rather than
regex-scanning downgrade() source. Regex cannot reliably detect
destructive operations across SQL dialects, comments, and
dynamically-constructed queries.
"""
migration_obj = await self._load_migration_object(migration)
if not hasattr(migration_obj, 'downgrade'):
return True # No rollback = destructive
# Declarative flag — migration author explicitly marks destructive rollbacks
return getattr(migration_obj, 'DESTRUCTIVE_ROLLBACK', False)
Rollback-then-Reapply Support¶
Resolving UNIQUE constraint issue:
-- Migration history table (corrected design)
CREATE TABLE _migration_history (
namespace VARCHAR(63) NOT NULL,
sequence INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
checksum CHAR(64) NOT NULL,
status VARCHAR(20) NOT NULL DEFAULT 'applied',
-- 'pending': execution started but not yet complete (BATCHED/NO_TRANSACTION)
-- 'applied': migration fully applied
-- 'rollback_started': rollback in progress (crash-safe marker)
applied_at TIMESTAMP WITH TIME ZONE NOT NULL,
execution_ms INTEGER,
applied_by VARCHAR(255),
PRIMARY KEY (namespace, sequence)
-- PK enforces uniqueness per (namespace, sequence). Rollback-then-reapply
-- works by DELETEing from this table (moving the record to
-- _migration_rollback_log) before re-inserting on reapply.
);
CREATE TABLE _migration_rollback_log (
id SERIAL PRIMARY KEY,
namespace VARCHAR(63) NOT NULL,
sequence INTEGER NOT NULL,
rolled_back_at TIMESTAMP WITH TIME ZONE NOT NULL,
rolled_back_by VARCHAR(255),
reason TEXT
);
-- Checkpoint table includes cursor position for keyset pagination
CREATE TABLE _migration_checkpoints (
checkpoint_id VARCHAR(64) PRIMARY KEY,
processed_count INTEGER NOT NULL DEFAULT 0,
last_seen_cursor BIGINT, -- Monotonic cursor for keyset pagination (NOT UUID)
total_expected INTEGER,
started_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);
-- Migration-critical indexes for performance.
-- doc_seq is a BIGSERIAL column used for deterministic keyset pagination.
-- Random UUIDs are unsuitable as pagination cursors because they have no
-- natural ordering — ORDER BY uuid produces arbitrary results.
CREATE INDEX idx_documents_collection_seq ON documents (collection, doc_seq);
CREATE INDEX idx_documents_data_gin ON documents USING GIN (data jsonb_path_ops);
Bootstrap problem: On first run, no migration infrastructure tables exist yet.
The runner uses CREATE TABLE IF NOT EXISTS exclusively for its own infrastructure
tables (_migration_history, _migration_rollback_log, _migration_checkpoints,
_migration_lock_heartbeat). This is the only place IF NOT EXISTS is
acceptable — all application schema changes go through migrations.
async def ensure_infrastructure(self, pool: asyncpg.Pool) -> None:
"""Create migration infrastructure tables if they don't exist.
Called once at startup, before any migration discovery or execution.
Uses CREATE TABLE IF NOT EXISTS because these tables cannot be managed
by the migration system they support (chicken-and-egg).
All tables are created in a single transaction to avoid partial
infrastructure state if the process crashes mid-creation.
"""
async with pool.acquire() as conn:
async with conn.transaction():
await conn.execute("""
CREATE TABLE IF NOT EXISTS _migration_history ( ... );
CREATE TABLE IF NOT EXISTS _migration_rollback_log ( ... );
CREATE TABLE IF NOT EXISTS _migration_checkpoints ( ... );
CREATE TABLE IF NOT EXISTS _migration_lock_heartbeat ( ... );
""")
Rollback implementation:
async def rollback_migration(
self,
namespace: str,
sequence: int,
) -> None:
"""Rollback migration with proper audit trail.
Mirrors the upgrade path: rollback_execution_mode determines whether
the downgrade runs in a single transaction (TRANSACTIONAL) or via
checkpointed batching (BATCHED).
Atomicity strategy:
- Record 'rollback_started' BEFORE executing downgrade(). This ensures
that if the process crashes during downgrade, the state is visible
and the operator can investigate.
- For TRANSACTIONAL mode, the downgrade + history deletion happen in
the same transaction.
- For BATCHED mode, downgrade runs outside a transaction; history
update happens after successful completion.
"""
migration_obj = await self._load_migration_object(namespace, sequence)
rb_mode = getattr(
migration_obj, "rollback_execution_mode", ExecutionMode.TRANSACTIONAL
)
async with self._pool.acquire() as conn:
ctx = MigrationContext(
conn=conn,
namespace=namespace,
batch_executor=BatchExecutor(self._pool),
checkpoint_manager=CheckpointManager(self._pool),
ownership_validator=ComponentOwnershipValidator(
self._planner.registry
),
logger=logger.bind(migration=f"{namespace}/{sequence:04d}"),
)
# Phase 1: Record rollback_started BEFORE executing downgrade.
# This provides crash-safe state: if the process dies during
# downgrade(), the 'rollback_started' status is visible to operators.
await conn.execute("""
UPDATE _migration_history
SET status = 'rollback_started'
WHERE namespace = $1 AND sequence = $2
""", namespace, sequence)
if rb_mode == ExecutionMode.TRANSACTIONAL:
# Transactional rollback: downgrade + history in same transaction
async with conn.transaction():
await migration_obj.downgrade(ctx)
await conn.execute("""
INSERT INTO _migration_rollback_log
(namespace, sequence, rolled_back_at, rolled_back_by, reason)
VALUES ($1, $2, NOW(), $3, $4)
""", namespace, sequence, self._current_user, "Manual rollback")
await conn.execute("""
DELETE FROM _migration_history
WHERE namespace = $1 AND sequence = $2
""", namespace, sequence)
elif rb_mode == ExecutionMode.BATCHED:
# Batched rollback — no outer transaction, each batch commits
# independently with checkpoint (mirrors batched upgrade path)
await migration_obj.downgrade(ctx)
# Phase 2: Record completion after successful batched downgrade
async with conn.transaction():
await conn.execute("""
INSERT INTO _migration_rollback_log
(namespace, sequence, rolled_back_at, rolled_back_by, reason)
VALUES ($1, $2, NOW(), $3, $4)
""", namespace, sequence, self._current_user, "Manual rollback")
await conn.execute("""
DELETE FROM _migration_history
WHERE namespace = $1 AND sequence = $2
""", namespace, sequence)
else: # NO_TRANSACTION
await migration_obj.downgrade(ctx)
async with conn.transaction():
await conn.execute("""
INSERT INTO _migration_rollback_log
(namespace, sequence, rolled_back_at, rolled_back_by, reason)
VALUES ($1, $2, NOW(), $3, $4)
""", namespace, sequence, self._current_user, "Manual rollback")
await conn.execute("""
DELETE FROM _migration_history
WHERE namespace = $1 AND sequence = $2
""", namespace, sequence)
JSONB Rollback Limitations¶
Explicit data loss warnings:
async def validate_rollback_safety(
self, migration: AppliedMigration
) -> list[str]:
"""Check for data loss risks in rollback."""
warnings = []
# Check for field additions that may have been written to
if self._migration_adds_jsonb_fields(migration):
# Query for documents with data in added fields
added_fields = self._get_added_fields(migration)
for component, fields in added_fields.items():
for field in fields:
count = await self._count_documents_with_field(
component, field
)
if count > 0:
warnings.append(
f"⚠ {count} documents have values in {component}.{field} "
f"which will be lost on rollback"
)
# Check for type changes with potential data loss
if self._migration_changes_field_types(migration):
warnings.append(
"⚠ This migration changed field types. Rollback may fail "
"if stored values are not compatible with original type."
)
return warnings
Testing Strategy¶
Comprehensive Test Coverage¶
# packages/maid-engine/tests/migrations/test_jsonb_transforms.py
class TestJSONBTransforms:
"""Test JSONB transformation safety and correctness."""
@pytest.fixture
async def isolated_db(self) -> asyncpg.Connection:
"""Create isolated test database with sample data."""
conn = await asyncpg.connect(TEST_DATABASE_URL)
# Create test schema
await conn.execute("""
CREATE TABLE documents (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
collection VARCHAR(50) NOT NULL,
data JSONB NOT NULL,
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
)
""")
# Insert test documents
await conn.execute("""
INSERT INTO documents (collection, data) VALUES
('entities', $1),
('entities', $2),
('entities', $3)
""",
# Mix of old and new schema documents
json.dumps({
"components": {
"HealthComponent": {"hp": 100, "max_hp": 100}
}
}),
json.dumps({
"components": {
"HealthComponent": {"current_health": 85, "max_health": 100}
}
}),
json.dumps({
"components": {
"PositionComponent": {"x": 10, "y": 20}
}
})
)
yield conn
await conn.close()
async def test_rename_field_idempotent(self, isolated_db):
"""Test field rename is idempotent (handles mixed schema)."""
# Run rename transformation
updated = await rename_jsonb_field(
isolated_db,
collection="entities",
component="HealthComponent",
old_name="hp",
new_name="current_health",
)
assert updated == 1 # Only one document had old field
# Verify old field removed, new field present
docs = await isolated_db.fetch(
"SELECT data FROM documents WHERE collection = 'entities'"
)
for doc in docs:
health = doc['data'].get('components', {}).get('HealthComponent')
if health:
assert 'hp' not in health
assert 'current_health' in health
# Run again - should be idempotent (no changes)
updated_again = await rename_jsonb_field(
isolated_db,
collection="entities",
component="HealthComponent",
old_name="hp",
new_name="current_health",
)
assert updated_again == 0
async def test_component_ownership_validation(self):
"""Test migration cannot modify foreign components via transform API."""
validator = ComponentOwnershipValidator(mock_registry)
# API-level enforcement: check_access raises on ownership violation
with pytest.raises(OwnershipViolationError, match="HealthComponent.*owned by.*stdlib"):
validator.check_access(
namespace="classic-rpg",
component="HealthComponent",
)
# Owning namespace is allowed
validator.check_access(
namespace="stdlib",
component="HealthComponent",
) # Should not raise
# Unknown components are allowed (no ownership claim)
validator.check_access(
namespace="classic-rpg",
component="UnclaimedComponent",
) # Should not raise
async def test_batched_update_checkpoint_recovery(self, isolated_db):
"""Test batched operations recover from checkpoint."""
# Insert large dataset
for i in range(1000):
await isolated_db.execute("""
INSERT INTO documents (collection, data) VALUES
('entities', $1)
""", json.dumps({
"components": {
"HealthComponent": {"hp": 100 + i}
}
}))
# Create checkpoint (simulate interrupted migration)
checkpoint_id = "entities_hp_rename_test"
await CheckpointManager.create(
isolated_db,
checkpoint_id,
total_expected=1000,
processed_count=300, # Simulate 30% complete
)
# Run batched rename (should resume from checkpoint)
updated = await rename_jsonb_field(
isolated_db,
collection="entities",
component="HealthComponent",
old_name="hp",
new_name="current_health",
batch_size=50,
)
# Should process remaining 700 documents
assert updated == 700
# Verify checkpoint was cleaned up
checkpoint = await CheckpointManager.get(isolated_db, checkpoint_id)
assert checkpoint is None
class TestMigrationRunner:
"""Test end-to-end migration orchestration."""
@pytest.fixture
def mock_content_packs(self) -> list[MigrationManifest]:
return [
MigrationManifest(
name="engine",
version="1.0.0",
dependencies=[],
components=[],
),
MigrationManifest(
name="stdlib",
version="1.0.0",
dependencies=["engine"],
components=["PositionComponent", "HealthComponent"],
),
MigrationManifest(
name="classic-rpg",
version="1.0.0",
dependencies=["stdlib"],
migration_requires={"stdlib": 2},
components=["CharacterStatsComponent", "CombatSkillComponent"],
),
]
async def test_dependency_order_resolution(self, mock_content_packs):
"""Test migrations execute in dependency order."""
registry = PackManifestRegistry()
registry._manifests = {m.name: m for m in mock_content_packs}
resolver = DependencyResolver()
order = resolver.resolve_migration_order([
MigrationSource("classic-rpg", Path("/fake"), ["stdlib"]),
MigrationSource("engine", Path("/fake"), []),
MigrationSource("stdlib", Path("/fake"), ["engine"]),
])
assert order == ["engine", "stdlib", "classic-rpg"]
async def test_migration_requirement_validation(self, isolated_db):
"""Test pack migration requirements are enforced."""
# Apply engine migrations 1-2, stdlib 1-2
await self._apply_mock_migrations(isolated_db, [
("engine", 1), ("engine", 2),
("stdlib", 1), ("stdlib", 2),
])
# Try to apply classic-rpg which requires stdlib:2
runner = MigrationRunner(mock_pool(isolated_db), mock_registry)
# Should succeed - requirement met
result = await runner.validate_migration_requirements(
"classic-rpg", {"stdlib": 2}
)
assert result.success
# Rollback stdlib to version 1
await self._rollback_migrations(isolated_db, [("stdlib", 2)])
# Should fail - requirement not met
result = await runner.validate_migration_requirements(
"classic-rpg", {"stdlib": 2}
)
assert not result.success
assert "requires stdlib/0002" in result.error
class TestCLIIntegration:
"""Test CLI command integration and user experience."""
async def test_migrate_dry_run_output(self):
"""Test dry run produces useful output without executing."""
runner = CliRunner()
result = await runner.invoke(db_migrate, [
"--namespace", "stdlib",
"--dry-run"
])
assert result.exit_code == 0
assert "Migration Plan (DRY RUN)" in result.output
assert "stdlib/0004_health_component_evolution" in result.output
assert "Estimated affected documents" in result.output
assert "Continue? [y/N]:" not in result.output # Non-interactive
async def test_status_command_formatting(self):
"""Test status command produces readable table output."""
runner = CliRunner()
result = await runner.invoke(db_status, ["--show-pending"])
assert result.exit_code == 0
assert "Migration Status" in result.output
assert "Namespace" in result.output
assert "Current" in result.output
assert "Pending" in result.output
# Should show unicode status indicators
assert "✓" in result.output or "⚠" in result.output
Integration Test Scenarios¶
class TestProductionScenarios:
"""Test real-world migration scenarios."""
async def test_zero_downtime_expand_contract(self):
"""Test expand-contract pattern for zero-downtime deployment."""
# Phase 1: Expand - add new field alongside old
await self._run_migration("stdlib/0010_health_expand")
# Verify both fields coexist
doc = await self._get_test_document()
health = doc['data']['components']['HealthComponent']
assert 'hp' in health # Old field
assert 'current_health' in health # New field
# Simulate application writing to both fields
await self._update_document_both_fields()
# Phase 2: Contract - remove old field
await self._run_migration("stdlib/0011_health_contract")
# Verify old field removed, data preserved
doc = await self._get_test_document()
health = doc['data']['components']['HealthComponent']
assert 'hp' not in health
assert 'current_health' in health
assert health['current_health'] == 90 # Preserved value
async def test_large_collection_migration(self):
"""Test migration performance on large dataset."""
# Create 100K test documents
await self._create_large_dataset(100_000)
start_time = time.time()
# Run batched migration
result = await self._run_migration(
"stdlib/0012_large_health_migration",
expect_batching=True,
)
duration = time.time() - start_time
assert result.success
assert duration < 300 # Should complete in < 5 minutes
assert result.documents_updated == 100_000
# Verify all documents transformed correctly
count = await self._count_documents_with_old_schema()
assert count == 0
async def test_migration_failure_recovery(self):
"""Test recovery from mid-migration failure."""
# Start migration that will fail partway
with pytest.raises(MigrationExecutionError):
await self._run_migration_with_simulated_failure(
"stdlib/0013_failing_migration",
fail_after_batches=3,
)
# Verify partial state - some docs migrated, some not
migrated = await self._count_migrated_documents()
total = await self._count_total_documents()
assert 0 < migrated < total # Partial migration
# Verify migration not recorded as complete
applied = await self._get_applied_migration("stdlib", 13)
assert applied is None
# Fix underlying issue and re-run
await self._fix_migration_issue()
result = await self._run_migration("stdlib/0013_failing_migration")
# Should complete successfully, processing remaining docs
assert result.success
migrated_after = await self._count_migrated_documents()
assert migrated_after == total
Performance Considerations¶
Batched Migration Performance¶
Sizing and timing guidelines:
| Dataset Size | Recommended Strategy | Expected Duration | Downtime Required |
|---|---|---|---|
| < 10k docs | Single transaction | < 30 seconds | Brief (< 1 minute) |
| 10k - 50k docs | Single transaction | 1-5 minutes | Brief (< 10 minutes) |
| 50k - 500k docs | Checkpointed batching | 10-60 minutes | None (online) |
| > 500k docs | Expand-contract pattern | 2-24 hours | None (online) |
Batch size optimization:
def calculate_optimal_batch_size(
document_count: int,
avg_document_size_kb: float,
target_batch_duration_ms: int = 5000,
) -> int:
"""Calculate optimal batch size for dataset characteristics."""
# Base calculation on target processing time
# Empirical: ~1000 docs/second for typical JSONB transforms
base_batch_size = target_batch_duration_ms
# Adjust for document size (larger docs = smaller batches)
if avg_document_size_kb > 10:
size_factor = 10 / avg_document_size_kb
base_batch_size = int(base_batch_size * size_factor)
# Clamp to reasonable bounds
return max(100, min(10_000, base_batch_size))
Note: The previous
execute_batched_migration_with_monitoring()standalone function has been removed. Its monitoring and adaptive batch sizing logic is now integrated intoBatchExecutor.execute(), which is the single entry point for all batched operations. This eliminates the code duplication between the two implementations.### Memory Management ```python class MigrationResourceManager: """Manages memory and connection resources during migrations.""" def __init__(self, pool: asyncpg.Pool) -> None: self._pool = pool self._memory_monitor = MemoryMonitor() async def execute_with_resource_limits( self, migration_func: Callable, **kwargs, ) -> Any: """Execute migration with memory monitoring.""" initial_memory = self._memory_monitor.get_current_usage() try: return await migration_func(**kwargs) except asyncpg.exceptions.OutOfMemoryError: # Reduce batch size and retry if 'batch_size' in kwargs: kwargs['batch_size'] = max(100, kwargs['batch_size'] // 2) logger.warning( f"Memory limit hit, reducing batch size to " f"{kwargs['batch_size']}" ) return await migration_func(**kwargs) raise finally: final_memory = self._memory_monitor.get_current_usage() memory_delta = final_memory - initial_memory if memory_delta > 100 * 1024 * 1024: # 100MB logger.warning( f"Migration used {memory_delta / 1024 / 1024:.1f}MB memory" ) # Force garbage collection gc.collect()
Security Model¶
Trust Levels for Content Packs¶
Content pack migrations have different trust levels based on their source:
| Trust Level | Namespaces | Migration Format | Capabilities |
|---|---|---|---|
| Trusted | engine, stdlib |
Python (.py) |
Full SQL, raw connection access, arbitrary code |
| Semi-trusted | First-party game packs (classic-rpg, tutorial-world) |
Python (.py) |
JSONB transforms only via RestrictedContext, ownership-enforced |
| Untrusted | Community packs | Declarative YAML (.yaml) |
Predefined operations only, no raw SQL |
Rationale: Python migration files execute arbitrary code with database access. Community content packs should not be able to run arbitrary Python against the production database.
Semi-trusted policy details:
Semi-trusted packs receive a RestrictedContext (extends BaseMigrationContext, no raw
conn access). They may only modify data through the transform library (batch_executor,
ownership_validator), which enforces namespace ownership at every call. Specifically:
- Allowed:
rename_jsonb_field,add_jsonb_field,remove_jsonb_field,set_schema_version— all ownership-validated at the API boundary. - Prohibited: Raw SQL execution (
ctx.conn.execute()), direct table DDL, cross-namespace component modification. - Execution modes: Semi-trusted packs may use
TRANSACTIONALorBATCHEDmode.NO_TRANSACTIONmode is restricted toTRUSTEDnamespaces only, because it bypasses transaction safety and requires careful manual coordination.
Semi-trusted packs should implement the RestrictedMigration protocol (which types
upgrade/downgrade with RestrictedContext) for static type safety.
# security.py
class MigrationTrustLevel(Enum):
TRUSTED = "trusted" # engine, stdlib — full Python
SEMI_TRUSTED = "semi_trusted" # first-party game packs — Python, ownership-enforced
UNTRUSTED = "untrusted" # community packs — YAML only
TRUSTED_NAMESPACES = frozenset({"engine", "stdlib"})
SEMI_TRUSTED_NAMESPACES = frozenset({"classic-rpg", "tutorial-world"})
def get_trust_level(namespace: str) -> MigrationTrustLevel:
if namespace in TRUSTED_NAMESPACES:
return MigrationTrustLevel.TRUSTED
if namespace in SEMI_TRUSTED_NAMESPACES:
return MigrationTrustLevel.SEMI_TRUSTED
return MigrationTrustLevel.UNTRUSTED
def validate_migration_format(
namespace: str,
migration_path: Path,
) -> None:
"""Enforce that untrusted packs only use YAML migrations."""
trust = get_trust_level(namespace)
if trust == MigrationTrustLevel.UNTRUSTED and migration_path.suffix == ".py":
raise SecurityError(
f"Community pack '{namespace}' must use declarative YAML migrations, "
f"not Python. Found: {migration_path.name}"
)
def validate_execution_mode_trust(
namespace: str,
execution_mode: ExecutionMode,
rollback_execution_mode: ExecutionMode,
) -> None:
"""Enforce execution mode restrictions based on trust level.
NO_TRANSACTION mode is restricted to TRUSTED namespaces. Semi-trusted
and untrusted packs may only use TRANSACTIONAL or BATCHED modes.
YAML migrations (untrusted) are additionally restricted to TRANSACTIONAL
only — batched operations require Python for checkpoint coordination.
"""
trust = get_trust_level(namespace)
if trust == MigrationTrustLevel.UNTRUSTED:
for mode, label in [
(execution_mode, "execution_mode"),
(rollback_execution_mode, "rollback_execution_mode"),
]:
if mode != ExecutionMode.TRANSACTIONAL:
raise SecurityError(
f"YAML migration in '{namespace}' cannot use "
f"{label}={mode.value}. Only 'transactional' is allowed "
f"for untrusted packs."
)
if trust != MigrationTrustLevel.TRUSTED:
for mode, label in [
(execution_mode, "execution_mode"),
(rollback_execution_mode, "rollback_execution_mode"),
]:
if mode == ExecutionMode.NO_TRANSACTION:
raise SecurityError(
f"Pack '{namespace}' cannot use {label}=no_transaction. "
f"NO_TRANSACTION mode is restricted to trusted namespaces."
)
Declarative YAML Migration Format (Community Packs)¶
YAML deserialization: All YAML migration files must be loaded via
yaml.safe_load() (never yaml.load()). safe_load restricts deserialization
to basic Python types, preventing arbitrary object instantiation attacks.
Complete YAML schema:
# Required top-level fields
namespace: string # Must match the content pack manifest name
sequence: integer # Monotonic within namespace (≥ 1)
description: string # Human-readable summary
# Optional top-level fields
online_safe: boolean # Default: false
execution_mode: string # YAML migrations: "transactional" ONLY.
# "batched" and "no_transaction" are restricted to
# Python migrations at SEMI_TRUSTED+ trust level.
# Default: "transactional"
rollback_execution_mode: string # Same restriction. Default: "transactional"
statement_timeout_ms: integer # Default: null (use global)
migration_timeout_s: integer # Default: null (use global)
# Required: list of upgrade operations
operations:
- type: string # One of the allowed operation types (see below)
collection: string # Target document collection
component: string # Target component type (must be owned by this pack)
# Additional fields depend on operation type
# Optional: list of rollback operations
rollback:
- type: string
collection: string
component: string
# ...
Allowed operation types:
- add_jsonb_field: Adds a field with a default value
- remove_jsonb_field: Removes a field
- rename_jsonb_field: Renames a field
- set_schema_version: Stamps a schema version on matching documents
Example:
# community_pack/0001_add_morale_field.yaml
namespace: my-community-pack
sequence: 1
description: "Add morale field to NPCBehaviorComponent"
execution_mode: transactional
operations:
- type: add_jsonb_field
collection: entities
component: NPCBehaviorComponent # Must be owned by this pack
field_name: morale
default_value: 100
- type: set_schema_version
collection: entities
component: NPCBehaviorComponent
version: "1.1"
rollback:
- type: remove_jsonb_field
collection: entities
component: NPCBehaviorComponent
field_name: morale
YAML loading implementation:
import yaml
def load_yaml_migration(path: Path) -> MigrationDescriptor:
"""Load a declarative YAML migration file.
Uses yaml.safe_load() exclusively — never yaml.load() — to prevent
arbitrary object instantiation from untrusted community pack files.
"""
with open(path) as f:
data = yaml.safe_load(f) # MANDATORY: safe_load only
_validate_yaml_schema(data)
return MigrationDescriptor(
namespace=data["namespace"],
sequence=data["sequence"],
description=data["description"],
execution_mode=ExecutionMode(
data.get("execution_mode", "transactional")
),
rollback_execution_mode=ExecutionMode(
data.get("rollback_execution_mode", "transactional")
),
STATEMENT_TIMEOUT_MS=data.get("statement_timeout_ms"),
MIGRATION_TIMEOUT_S=data.get("migration_timeout_s"),
_upgrade_fn=_build_yaml_upgrade(data["operations"]),
_downgrade_fn=_build_yaml_downgrade(data.get("rollback")),
_source_path=path,
)
Migration Integrity Checking¶
All applied migration checksums are verified before any new migration runs. This detects tampered or modified migration files that have already been applied.
class ChecksumVerifier:
"""Verifies integrity of applied migrations on startup and before runs."""
async def verify_all(
self,
pool: asyncpg.Pool,
known_migrations: dict[tuple[str, int], str], # (ns, seq) -> checksum
) -> list[ChecksumMismatch]:
"""Compare stored checksums against current migration file checksums."""
mismatches = []
async with pool.acquire() as conn:
applied = await conn.fetch(
"SELECT namespace, sequence, checksum FROM _migration_history"
)
for row in applied:
key = (row["namespace"], row["sequence"])
current_checksum = known_migrations.get(key)
if current_checksum and current_checksum != row["checksum"]:
mismatches.append(ChecksumMismatch(
namespace=row["namespace"],
sequence=row["sequence"],
stored=row["checksum"],
current=current_checksum,
))
return mismatches
The runner aborts with a clear error if any mismatches are found, unless
--fix-checksums is explicitly passed to maid db validate.
Advisory Lock Semantics¶
Per-Namespace Locking¶
The prior design used a single global advisory lock, serializing all migrations regardless of independence. The revised design uses per-namespace advisory locks, allowing independent content packs to migrate in parallel.
# lock.py
class NamespaceAdvisoryLock:
"""Per-namespace PostgreSQL advisory lock with session-level semantics.
Design choices:
- Session-level locks (pg_advisory_lock), not transaction-level, because
migrations may span multiple transactions (batched operations).
- A dedicated lock connection, separate from the migration connection,
so the lock survives transaction commits/rollbacks within the migration.
- A heartbeat mechanism for stale lock detection: the lock holder
periodically updates a timestamp; other waiters can detect if the
holder has died without releasing the lock.
"""
# Lock IDs are derived from namespace to avoid collisions
LOCK_NAMESPACE = 0x4D414944 # 'MAID' in hex
STALE_HEARTBEAT_THRESHOLD_S = 60 # Heartbeat older than this is stale
def __init__(self, pool: asyncpg.Pool) -> None:
self._pool = pool
self._lock_conn: asyncpg.Connection | None = None
self._heartbeat_task: asyncio.Task | None = None
self._heartbeat_healthy = True
def _lock_id(self, namespace: str) -> int:
"""Deterministic lock ID from namespace name.
Masked to signed int4 range (0x7FFFFFFF) because PostgreSQL
pg_advisory_lock accepts two int4 arguments. Without masking,
SHA-256 derived values can overflow int4, causing errors or
silent truncation.
"""
raw = int(hashlib.sha256(namespace.encode()).hexdigest()[:8], 16)
return raw & 0x7FFFFFFF
@asynccontextmanager
async def acquire(
self,
namespace: str,
timeout_seconds: int = 30,
) -> AsyncGenerator[None, None]:
"""Acquire session-level advisory lock for a namespace.
Uses exponential backoff when the lock is held, starting at 0.5s
and capping at 8s between attempts. Also checks for stale heartbeats
from crashed lock holders.
"""
lock_id = self._lock_id(namespace)
# Dedicated connection for lock — survives transaction boundaries
self._lock_conn = await self._pool.acquire()
try:
deadline = asyncio.get_event_loop().time() + timeout_seconds
backoff = 0.5 # Start with 500ms, exponential backoff
while True:
acquired = await self._lock_conn.fetchval(
"SELECT pg_try_advisory_lock($1, $2)",
self.LOCK_NAMESPACE, lock_id,
)
if acquired:
break
if asyncio.get_event_loop().time() > deadline:
raise MigrationLockTimeout(
f"Could not acquire lock for '{namespace}' "
f"within {timeout_seconds}s"
)
# Check for stale heartbeat from a crashed holder
stale = await self._check_stale_heartbeat(namespace)
if stale:
logger.warning(
f"Stale lock detected for '{namespace}' "
f"(heartbeat older than {self.STALE_HEARTBEAT_THRESHOLD_S}s). "
f"Previous holder likely crashed. Consider manual cleanup."
)
await asyncio.sleep(min(backoff, 8.0))
backoff *= 2 # Exponential backoff, capped at 8s
# Start heartbeat for stale lock detection
self._heartbeat_healthy = True
self._heartbeat_task = asyncio.create_task(
self._heartbeat_loop(namespace)
)
yield
finally:
if self._heartbeat_task:
self._heartbeat_task.cancel()
try:
await self._heartbeat_task
except asyncio.CancelledError:
pass
await self._lock_conn.execute(
"SELECT pg_advisory_unlock($1, $2)",
self.LOCK_NAMESPACE, lock_id,
)
await self._pool.release(self._lock_conn)
self._lock_conn = None
async def _check_stale_heartbeat(self, namespace: str) -> bool:
"""Check if the current lock holder's heartbeat is stale.
Returns True if a heartbeat exists but hasn't been updated within
STALE_HEARTBEAT_THRESHOLD_S seconds, indicating the holder likely
crashed without releasing the lock.
"""
async with self._pool.acquire() as conn:
row = await conn.fetchrow("""
SELECT last_seen FROM _migration_lock_heartbeat
WHERE namespace = $1
""", namespace)
if row is None:
return False
age_seconds = await conn.fetchval("""
SELECT EXTRACT(EPOCH FROM (NOW() - $1))
""", row["last_seen"])
return age_seconds > self.STALE_HEARTBEAT_THRESHOLD_S
async def _heartbeat_loop(self, namespace: str) -> None:
"""Periodically update heartbeat timestamp for stale detection.
If the heartbeat write fails, sets a flag so the migration runner
can detect the issue and abort gracefully rather than silently
holding a lock that appears stale to other processes.
"""
try:
while True:
await asyncio.sleep(10)
try:
await self._lock_conn.execute("""
INSERT INTO _migration_lock_heartbeat (namespace, last_seen)
VALUES ($1, NOW())
ON CONFLICT (namespace) DO UPDATE SET last_seen = NOW()
""", namespace)
except Exception as exc:
logger.error(
f"Heartbeat write failed for '{namespace}': {exc}. "
f"Lock may appear stale to other processes."
)
self._heartbeat_healthy = False
raise # Propagate to cancel the task
except asyncio.CancelledError:
pass # Normal shutdown
Application Signaling¶
PostgreSQL NOTIFY for Live Engine Coordination¶
When migrations run while the engine is live (online-safe migrations), the engine needs to know about schema changes to refresh cached models or invalidate stale data.
# After each successful migration, emit a NOTIFY:
await conn.execute(
"SELECT pg_notify('maid_schema_changes', $1)",
json.dumps({
"namespace": migration.namespace,
"sequence": migration.sequence,
"action": "applied",
"timestamp": datetime.utcnow().isoformat(),
}),
)
# Engine-side listener (in GameEngine or a dedicated schema watcher):
async def listen_for_schema_changes(pool: asyncpg.Pool) -> None:
conn = await pool.acquire()
await conn.add_listener("maid_schema_changes", on_schema_change)
async def on_schema_change(
conn: asyncpg.Connection,
pid: int,
channel: str,
payload: str,
) -> None:
event = json.loads(payload)
logger.info(f"Schema change: {event['namespace']}/{event['sequence']:04d}")
# Invalidate component model caches, refresh Pydantic validators, etc.
Migration Safety Guidelines¶
ONLINE_SAFE Criteria¶
A migration may declare ONLINE_SAFE = True only if it meets all of the following
criteria. These guidelines define when a migration is safe to apply while the game
engine is running with active player connections.
| Criterion | Requirement | Example |
|---|---|---|
| No exclusive locks | Must not acquire ACCESS EXCLUSIVE or SHARE ROW EXCLUSIVE locks on tables with active reads/writes |
CREATE INDEX CONCURRENTLY (ok), ALTER TABLE (not ok) |
| No schema-breaking changes | Must not remove or rename columns/fields that running code depends on | Adding a new optional JSONB field (ok), renaming hp → current_health (not ok) |
| Backward-compatible JSONB | Running Pydantic models must still validate against both old and new document shapes | New optional field with default (ok), new required field (not ok) |
| Bounded execution time | Must complete within MIGRATION_TIMEOUT_S without risking connection pool exhaustion |
Small backfill or index creation (ok), full-table rewrite (not ok) |
| Idempotent operations | Must be safe to re-run if the migration is interrupted and resumed | jsonb_set with WHERE NOT exists (ok), unconditional increment (not ok) |
| No transaction-long locks | For BATCHED mode, each mini-transaction must hold locks for < 1 second |
Batch size 5000 with simple JSONB update (ok), batch with foreign key cascade (not ok) |
Expand-contract pattern for breaking changes:
Breaking changes (renames, type changes, removals) should use the expand-contract pattern across two migrations:
- Expand (
ONLINE_SAFE = True): Add new field alongside old, backfill data, deploy application code that reads both fields. - Contract (
ONLINE_SAFE = False): Remove old field after all application code has been updated. Apply during maintenance window.
Declaring ONLINE_SAFE:
class AddShieldField:
ONLINE_SAFE = True # New optional field with default — backward compatible
execution_mode = ExecutionMode.BATCHED
async def upgrade(self, ctx: BaseMigrationContext) -> None:
await add_jsonb_field(ctx, collection="entities",
component="HealthComponent",
field_name="shield", default_value=0)
Observability¶
Prometheus Metrics¶
Migration events are instrumented with Prometheus metrics for operational monitoring,
alerting, and SLO tracking. All metrics use the maid_ prefix.
# packages/maid-engine/src/maid_engine/migrations/metrics.py
from prometheus_client import Counter, Histogram, Gauge
# Duration of individual migration executions
maid_migration_duration_seconds = Histogram(
"maid_migration_duration_seconds",
"Time spent executing a single migration",
labelnames=["namespace", "sequence", "execution_mode", "direction"],
buckets=[0.1, 0.5, 1, 5, 10, 30, 60, 120, 300, 600, 1800, 3600],
)
# Count of migration operations (success/failure)
maid_migration_total = Counter(
"maid_migration_total",
"Total number of migration operations",
labelnames=["namespace", "direction", "status"],
# direction: "upgrade" | "rollback"
# status: "success" | "failure" | "skipped"
)
# Batch progress for long-running migrations
maid_migration_batch_rows_total = Counter(
"maid_migration_batch_rows_total",
"Total rows processed by batched migrations",
labelnames=["namespace", "sequence"],
)
# Current migration state
maid_migration_pending_count = Gauge(
"maid_migration_pending_count",
"Number of pending (unapplied) migrations",
labelnames=["namespace"],
)
# Lock contention
maid_migration_lock_wait_seconds = Histogram(
"maid_migration_lock_wait_seconds",
"Time spent waiting for advisory lock",
labelnames=["namespace"],
buckets=[0.1, 0.5, 1, 2, 5, 10, 30],
)
maid_migration_lock_stale_detected_total = Counter(
"maid_migration_lock_stale_detected_total",
"Number of stale lock detections during acquire",
labelnames=["namespace"],
)
Usage in runner:
async def _execute_single_migration(self, migration: Migration) -> None:
labels = {
"namespace": migration.namespace,
"sequence": str(migration.sequence),
"execution_mode": migration.execution_mode.value,
"direction": "upgrade",
}
with maid_migration_duration_seconds.labels(**labels).time():
# ... execute migration ...
pass
maid_migration_total.labels(
namespace=migration.namespace, direction="upgrade", status="success"
).inc()
Recommended alerts:
| Alert | Condition | Severity |
|---|---|---|
MigrationStuck |
maid_migration_duration_seconds > 2× MIGRATION_TIMEOUT_S |
Critical |
MigrationFailure |
maid_migration_total{status="failure"} > 0 |
Critical |
PendingMigrations |
maid_migration_pending_count > 0 for > 24h |
Warning |
StaleLock |
maid_migration_lock_stale_detected_total increase |
Warning |
Open Questions & Future Considerations¶
Database Schema Evolution¶
Question: Should we support database schema migrations beyond PostgreSQL?
The current design targets PostgreSQL exclusively, using PostgreSQL-specific features (advisory locks, JSONB operators, UPSERT syntax). The maid-registry package uses SQLite, creating an architectural inconsistency.
Options: 1. PostgreSQL-only (current approach): Simpler implementation, leverages advanced PostgreSQL features 2. Multi-database support: Abstract common operations, provide database-specific implementations 3. Registry migration: Move registry to PostgreSQL for consistency
Recommendation: Maintain PostgreSQL focus for core MAID, create separate migration runner for registry package.
maid-registry migration strategy: The maid-registry package uses SQLite and is
architecturally independent of the core engine. Its migration needs are explicitly
out of scope for this design. If maid-registry requires schema evolution, it
should use a lightweight SQLite-specific tool (e.g., yoyo-migrations or manual
user_version pragma tracking). This avoids coupling the registry to PostgreSQL
infrastructure and keeps the migration system focused on a single database backend.
Question: How should we handle migration squashing and history compression?
With active development, namespaces could accumulate 50+ migration files over time. Long migration chains slow down fresh deployments and complicate maintenance.
Options:
1. Manual squashing: Developer-initiated consolidation of migration sequences
2. Automatic squashing: System-driven compression at milestone versions
3. History truncation: Remove old migration files while preserving baseline checkpoints
Recommendation: Defer to v3.2. Design migration file format to support future squashing without breaking deployed systems.
Operational Considerations¶
Question: Should content pack migrations be version-pinned to pack releases?
Current design allows content pack code and migrations to evolve independently. A content pack at v2.1.0 might have migrations up to sequence 0015, but v2.0.0 only shipped with migrations through 0012.
Trade-offs: - Version coupling: Simpler deployment coordination, clearer compatibility - Independent versioning: More flexible development, enables hotfix migrations
Recommendation: Add optional MigrationManifest.migration_version field for explicit coupling when needed.
Question: How should we handle cross-pack data migrations?
Some migrations might need to transform data that spans multiple content packs (e.g., moving character stats from stdlib to classic-rpg ownership).
Options: 1. Prohibit cross-pack migrations: Enforce strict namespace isolation 2. Explicit cooperation: Special migration type that declares cross-pack dependencies 3. Migration unions: Multiple packs contribute to a single logical migration
Recommendation: Start with strict isolation. Add cross-pack migration support in v3.2 if needed.
Design Decisions Log¶
Decision 1: Custom Migration Runner over Alembic¶
Date: 2025-01-14
Context: Need migration system for multi-namespace content pack architecture
Decision: Build custom runner using asyncpg
Rationale:
- No SQLAlchemy dependency conflicts
- First-class content pack namespace support
- Native JSONB document evolution features
- Full control over migration discovery and execution order
Decision 2: Advisory Locks with Per-Namespace Granularity¶
Date: 2025-01-14 (updated 2025-01-15)
Context: Prevent concurrent migration execution in multi-server deployments
Decision: Use PostgreSQL session-level advisory locks with per-namespace granularity, dedicated lock connection, and heartbeat for stale detection
Rationale:
- Per-namespace locks allow independent packs to migrate in parallel (see Decision 13)
- Session-level locks survive transaction boundaries for batched migrations
- Dedicated lock connection prevents accidental release during transaction commits
- Heartbeat mechanism detects stale locks from crashed processes
- Retry loop provides actual timeout behavior
Decision 3: Checkpointed Batching for Large Migrations¶
Date: 2025-01-14 (updated 2025-01-15)
Context: Balance transactional safety with performance for large JSONB transformations
Decision: Explicit batching with cursor/keyset pagination, atomic mini-transactions, and deterministic checkpoint IDs
Rationale:
- Cursor pagination (WHERE id > $last_id) replaces broken OFFSET-based approach (see Decision 9)
- Each batch's data update and checkpoint update wrapped in a single transaction (see Decision 11)
- Checkpoint IDs derived via SHA-256, not Python's non-deterministic hash() (see Decision 10)
- Provides resumable execution for interrupted migrations
- Allows operators to choose appropriate strategy per migration
Decision 4: Content Pack Manifest Registry¶
Date: 2025-01-14
Context: CLI commands need to discover content packs without starting GameEngine
Decision: Separate discovery mechanism using setuptools entry points and manifest files
Rationale:
- Resolves CLI discovery problem (critique P0-3)
- Enables migration execution before engine startup
- Supports dynamic content pack loading
- Provides dependency resolution without instantiation
Decision 5: Component Ownership Validation¶
Date: 2025-01-14
Context: Prevent content packs from modifying other packs' JSONB components
Decision: API-level enforcement via the transform library, not regex-based SQL scanning
Rationale:
- Regex-based _extract_component_references is fundamentally unsound — regex cannot reliably parse SQL
- API-level enforcement makes the transform library the single enforcement point
- Ownership checked at call site before any SQL is generated
- Community packs restricted to declarative YAML, so they cannot bypass the API
Decision 6: SQL Injection Prevention in JSONB Helpers¶
Date: 2025-01-14
Context: JSONB path manipulation requires dynamic SQL construction
Decision: Structured JSONBUpdateSpec builder that generates SQL internally — no raw string interpolation
Rationale:
- Prior _batch_jsonb_update accepted raw where_clause and update_clause strings via f-strings
- JSONBUpdateSpec encapsulates all SQL generation; callers never write SQL
- Parameterized path construction using PostgreSQL jsonpath expressions
- Validates component and field names at the API boundary
Decision 7: Rollback History Separation¶
Date: 2025-01-14
Context: Support rollback-then-reapply workflows
Decision: Separate migration history and rollback audit log tables
Rationale:
- Resolves UNIQUE constraint conflict (critique C1)
- Maintains complete audit trail of rollback operations
- Enables clean re-application of previously rolled back migrations
- Provides operational visibility into rollback patterns
Decision 8: Explicit Data Loss Warnings¶
Date: 2025-01-14
Context: JSONB rollbacks can lose data written after migration
Decision: Pre-rollback validation with explicit data loss warnings
Rationale:
- Addresses inherently lossy nature of JSONB rollbacks (critique P1-3)
- Provides operators with informed consent before destructive operations
- Enables detection of field usage before rollback
- Supports safer production rollback decisions
Decision 9: Cursor/Keyset Pagination for Batching¶
Date: 2025-01-15
Context: OFFSET-based batching skips rows on crash-resume when WHERE condition changes on modified rows
Decision: Replace OFFSET $n with WHERE cursor_col > $last ORDER BY cursor_col LIMIT $batch, using a monotonically increasing column (not random UUID)
Rationale:
- OFFSET+WHERE on mutable rows causes row skipping: after a crash, already-modified rows no longer match the WHERE clause, shifting the OFFSET window
- Keyset pagination is stable: id > $last_id always resumes from the correct position regardless of which rows have been modified
- last_seen_cursor stored in checkpoint enables deterministic resume
- Multiple reviewers flagged this as a critical correctness bug
Decision 10: Deterministic Checkpoint IDs (SHA-256)¶
Date: 2025-01-15
Context: Checkpoint IDs used hash() which is randomized per-process in Python 3.12+
Decision: Derive checkpoint IDs from hashlib.sha256 of spec content
Rationale:
- Python's hash() uses PYTHONHASHSEED randomization by default, producing different values across process restarts
- A crash-resume scenario would generate a different checkpoint ID and lose track of progress
- SHA-256 is deterministic and stable across restarts, platforms, and Python versions
Decision 11: Atomic Mini-Transactions for Batches¶
Date: 2025-01-15
Context: Checkpoint updates and data updates were not wrapped in the same transaction
Decision: Each batch wraps data update + checkpoint update in a single transaction
Rationale:
- Without atomicity, a crash between data commit and checkpoint update causes the batch to be re-applied (duplicating transforms)
- A crash between checkpoint update and data commit causes the batch to be skipped (losing transforms)
- Mini-transactions guarantee checkpoint always reflects actual data state
Decision 12: Separated Discovery/Planning/Execution¶
Date: 2025-01-15
Context: MigrationRunner was a god object handling discovery, dependency resolution, and execution
Decision: Split into PackManifestRegistry (discovery), MigrationPlanner (ordering), MigrationRunner (execution)
Rationale:
- Single responsibility: each component has one clear job
- Testable in isolation: planner can be tested without database, runner without filesystem
- Pool injection: one component creates the pool, all others receive it
Decision 13: Per-Namespace Advisory Locks¶
Date: 2025-01-15
Context: Global lock serialized all migrations, blocking independent content packs
Decision: Use per-namespace advisory locks derived from namespace name via SHA-256
Rationale:
- Independent packs (e.g., classic-rpg and tutorial-world) can migrate in parallel
- Session-level locks survive transaction boundaries within batched migrations
- Dedicated lock connection prevents accidental release during transaction commits
- Heartbeat mechanism enables stale lock detection
Decision 14: Content Pack Security Tiers¶
Date: 2025-01-15
Context: Community content packs could execute arbitrary Python via migration files
Decision: Restrict community packs to declarative YAML migrations; only trusted namespaces get Python
Rationale:
- Python migrations have full database access — unsuitable for untrusted code
- YAML format provides a safe, auditable subset of migration operations
- Trust levels (trusted/semi-trusted/untrusted) provide graduated capabilities
- Aligns with content pack security model for the broader plugin system
Decision 15: Formal Migration Protocol (Class-Based Only)¶
Date: 2025-01-15 (updated 2025-01-16)
Context: Migrations used implicit conventions (module-level constants, functions) with no type checking
Decision: Define a typed Migration protocol plus a MigrationDescriptor dataclass. All migrations must be class-based with a module-level migration instance. The from_module() legacy adapter has been removed.
Rationale:
- This is a greenfield system with no legacy module-level migrations to support
- Removing from_module() eliminates legacy boolean flag mapping and input validation concerns
- BaseMigrationContext / MigrationContext / RestrictedContext hierarchy provides type-safe context dispatch
- RestrictedMigration protocol enables static type checking for semi-trusted packs
- ExecutionMode enum replaces overlapping REQUIRES_BATCHING / REQUIRES_NO_TRANSACTION booleans
Decision 16: BaseMigrationContext Hierarchy¶
Date: 2025-01-16
Context: RestrictedContext was a standalone dataclass with duplicated fields, and Migration.upgrade() accepted MigrationContext — incompatible with RestrictedContext
Decision: Introduce BaseMigrationContext base class; MigrationContext extends it (adds conn/execute/fetch); RestrictedContext extends it (no conn). Migration protocol methods accept BaseMigrationContext.
Rationale:
- RestrictedContext previously violated the Migration protocol (wrong argument type)
- Shared base class eliminates field duplication and ensures protocol compatibility
- Semi-trusted packs can implement RestrictedMigration for stricter typing
Decision 17: SET SESSION for Non-Transactional Modes¶
Date: 2025-01-16
Context: SET LOCAL statement_timeout is scoped to the current transaction — ineffective for BATCHED and NO_TRANSACTION modes which operate outside a single transaction
Decision: Use SET SESSION for BATCHED/NO_TRANSACTION modes (with RESET in finally), SET LOCAL for TRANSACTIONAL mode
Rationale:
- SET LOCAL only affects the current transaction block; BATCHED/NO_TRANSACTION modes use multiple transactions or no transaction
- SET SESSION applies to the entire connection; RESET in finally block ensures cleanup even on failure
- TRANSACTIONAL mode retains SET LOCAL for proper transaction-scoped behavior
Decision 18: Two-Phase History Recording¶
Date: 2025-01-16
Context: Migration execution and history recording were not atomic — a crash between them could leave an applied migration unrecorded, or record an unapplied migration
Decision: For TRANSACTIONAL mode, insert history inside the same transaction. For BATCHED/NO_TRANSACTION, use two-phase recording (pending → applied).
Rationale:
- TRANSACTIONAL: wrapping upgrade + history INSERT in one transaction guarantees atomicity
- BATCHED/NO_TRANSACTION: cannot use a single transaction, so 'pending' status before execution + 'applied' after provides crash-safe visibility
- 'rollback_started' status before downgrade() execution provides the same crash safety for rollbacks
Decision 19: Advisory Lock Hardening¶
Date: 2025-01-16
Context: Multiple issues in advisory lock implementation — int4 overflow, silent heartbeat failures, no stale detection, fixed retry interval
Decision: Mask lock IDs to & 0x7FFFFFFF (signed int4); add heartbeat exception handling with health flag; check stale heartbeats during acquire; use exponential backoff
Rationale:
- PostgreSQL pg_advisory_lock accepts int4 arguments; SHA-256 derived values can overflow
- Silent heartbeat failures cause locks to appear stale to other processes
- Stale heartbeat detection lets waiters surface crashes to operators
- Exponential backoff (0.5s → 8s cap) reduces polling pressure under contention
Decision 20: Monotonic Keyset Pagination¶
Date: 2025-01-16
Context: BatchExecutor used UUID id column for keyset pagination (WHERE id > $last)
Decision: Use a monotonically increasing column (doc_seq BIGSERIAL or created_at) instead of random UUID for keyset pagination
Rationale:
- Random UUIDs (v4) have no natural ordering — ORDER BY uuid produces arbitrary results
- Keyset pagination relies on total ordering to produce consistent, non-overlapping pages
- A BIGSERIAL doc_seq column provides deterministic, gap-free ordering
- The cursor_column property on BatchOperation protocol makes this configurable per operation
End of Design Document