fix(v2): enforce M1 runtime persistence invariants

This commit is contained in:
2026-07-27 19:42:41 +02:00
parent 45a526a1e9
commit 15f8794029
7 changed files with 188 additions and 39 deletions
+25 -3
View File
@@ -3,17 +3,19 @@ from __future__ import annotations
import importlib import importlib
import os import os
from logging.config import fileConfig from logging.config import fileConfig
from typing import Any
from alembic import context from alembic import context
from sqlalchemy import create_engine, pool from sqlalchemy import create_engine, event, pool
from sqlalchemy.engine import URL, make_url from sqlalchemy.engine import URL, make_url
Base = importlib.import_module("backup_tool.db.models").Base models = importlib.import_module("backup_tool.db.models")
sqlite_runtime = importlib.import_module("backup_tool.db.sqlite")
config = context.config config = context.config
if config.config_file_name is not None: if config.config_file_name is not None:
fileConfig(config.config_file_name) fileConfig(config.config_file_name)
target_metadata = Base.metadata target_metadata = models.Base.metadata
def configured_url() -> str: def configured_url() -> str:
@@ -23,6 +25,19 @@ def configured_url() -> str:
return raw_url return raw_url
def busy_timeout_ms() -> int:
raw_value = os.environ.get(
"BACKUP_TOOL_SQLITE_BUSY_TIMEOUT_MS", str(sqlite_runtime.DEFAULT_BUSY_TIMEOUT_MS)
)
try:
value = int(raw_value)
except ValueError as error:
raise RuntimeError("SQLite busy timeout must be an integer") from error
if value < 1_000 or value > 120_000:
raise RuntimeError("SQLite busy timeout must be between 1000 and 120000 ms")
return value
def synchronous_url(raw_url: str) -> URL: def synchronous_url(raw_url: str) -> URL:
url = make_url(raw_url) url = make_url(raw_url)
if url.drivername != "sqlite+aiosqlite": if url.drivername != "sqlite+aiosqlite":
@@ -44,10 +59,17 @@ def run_migrations_offline() -> None:
def run_migrations_online() -> None: def run_migrations_online() -> None:
timeout = busy_timeout_ms()
engine = create_engine( engine = create_engine(
synchronous_url(configured_url()), synchronous_url(configured_url()),
poolclass=pool.NullPool, poolclass=pool.NullPool,
connect_args=sqlite_runtime.sqlite_connect_args(timeout),
) )
@event.listens_for(engine, "connect")
def configure_sqlite(dbapi_connection: Any, _connection_record: Any) -> None:
sqlite_runtime.configure_sqlite_connection(dbapi_connection)
with engine.connect() as connection: with engine.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata, compare_type=True) context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
with context.begin_transaction(): with context.begin_transaction():
+60 -5
View File
@@ -3,26 +3,81 @@
from __future__ import annotations from __future__ import annotations
import argparse import argparse
from collections.abc import Sequence import asyncio
from collections.abc import Callable, Sequence
from pathlib import Path
from alembic import command
from alembic.config import Config
from backup_tool import __version__ from backup_tool import __version__
from backup_tool.config import Settings
ROLES = ("web", "scheduler", "worker", "migrate", "admin") from .db.engine import assert_schema_current, create_engine
RoleHandler = Callable[[Settings], int]
DATABASE_ROLES = ("web", "scheduler", "worker", "admin")
def _placeholder_role(_settings: Settings) -> int:
return 0
ROLE_HANDLERS: dict[str, RoleHandler] = {role: _placeholder_role for role in DATABASE_ROLES}
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="backup-tool") parser = argparse.ArgumentParser(prog="backup-tool")
parser.add_argument("--version", action="version", version=__version__) parser.add_argument("--version", action="version", version=__version__)
subparsers = parser.add_subparsers(dest="role", required=True) subparsers = parser.add_subparsers(dest="role", required=True)
for role in ROLES: for role in DATABASE_ROLES:
subparsers.add_parser(role, help=f"run the {role} role") subparsers.add_parser(role, help=f"run the {role} role")
migrate = subparsers.add_parser("migrate", help="manage the metadata schema")
migrate.add_argument("action", choices=("upgrade", "downgrade", "current"))
return parser return parser
def main(argv: Sequence[str] | None = None) -> int: def build_alembic_config(settings: Settings) -> Config:
build_parser().parse_args(argv) backend_root = Path(__file__).resolve().parents[2]
migration = Config(backend_root / "alembic.ini")
migration.set_main_option("sqlalchemy.url", settings.database_url)
return migration
def require_current_schema(settings: Settings) -> None:
async def check() -> None:
engine = create_engine(settings)
try:
await assert_schema_current(engine, build_alembic_config(settings))
finally:
await engine.dispose()
asyncio.run(check())
def run_migration(action: str, settings: Settings) -> int:
migration = build_alembic_config(settings)
if action == "upgrade":
command.upgrade(migration, "head")
elif action == "downgrade":
command.downgrade(migration, "base")
else:
command.current(migration)
return 0 return 0
def load_settings() -> Settings:
return Settings() # type: ignore[call-arg]
def main(argv: Sequence[str] | None = None, *, settings: Settings | None = None) -> int:
args = build_parser().parse_args(argv)
runtime_settings = settings or load_settings()
if args.role == "migrate":
return run_migration(args.action, runtime_settings)
require_current_schema(runtime_settings)
return ROLE_HANDLERS[args.role](runtime_settings)
if __name__ == "__main__": if __name__ == "__main__":
raise SystemExit(main()) raise SystemExit(main())
+8 -12
View File
@@ -1,6 +1,5 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable
from typing import Any from typing import Any
from alembic.config import Config from alembic.config import Config
@@ -12,23 +11,23 @@ from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from backup_tool.config import Settings from backup_tool.config import Settings
from . import sqlite as sqlite_runtime
class SchemaNotCurrentError(RuntimeError): class SchemaNotCurrentError(RuntimeError):
pass pass
def create_engine(settings: Settings) -> AsyncEngine: def create_engine(settings: Settings) -> AsyncEngine:
engine = create_async_engine(settings.database_url, pool_pre_ping=True) engine = create_async_engine(
settings.database_url,
pool_pre_ping=True,
connect_args=sqlite_runtime.sqlite_connect_args(settings.sqlite_busy_timeout_ms),
)
@event.listens_for(engine.sync_engine, "connect") @event.listens_for(engine.sync_engine, "connect")
def configure_sqlite(dbapi_connection: Any, _connection_record: Any) -> None: def configure_sqlite(dbapi_connection: Any, _connection_record: Any) -> None:
cursor = dbapi_connection.cursor() sqlite_runtime.configure_sqlite_connection(dbapi_connection)
try:
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute(f"PRAGMA busy_timeout={settings.sqlite_busy_timeout_ms}")
cursor.execute("PRAGMA journal_mode=WAL")
finally:
cursor.close()
return engine return engine
@@ -45,6 +44,3 @@ async def assert_schema_current(engine: AsyncEngine, alembic_config: Config) ->
raise SchemaNotCurrentError( raise SchemaNotCurrentError(
f"database schema is not current: expected {expected!r}, found {current!r}" f"database schema is not current: expected {expected!r}, found {current!r}"
) )
SchemaCheck = Callable[[AsyncEngine, Config], Any]
+20 -17
View File
@@ -7,7 +7,6 @@ from sqlalchemy import (
JSON, JSON,
Boolean, Boolean,
CheckConstraint, CheckConstraint,
DateTime,
ForeignKey, ForeignKey,
Index, Index,
Integer, Integer,
@@ -21,6 +20,10 @@ from sqlalchemy import (
) )
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from backup_tool.ids import new_uuid7
from .types import UTCDateTime
NAMING_CONVENTION = { NAMING_CONVENTION = {
"ix": "ix_%(column_0_label)s", "ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s", "uq": "uq_%(table_name)s_%(column_0_name)s",
@@ -35,15 +38,15 @@ class Base(DeclarativeBase):
class IdentityMixin: class IdentityMixin:
id: Mapped[str] = mapped_column(String(36), primary_key=True) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(new_uuid7()))
class TimestampMixin: class TimestampMixin:
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.current_timestamp() UTCDateTime(), nullable=False, server_default=func.current_timestamp()
) )
updated_at: Mapped[datetime] = mapped_column( updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), UTCDateTime(),
nullable=False, nullable=False,
server_default=func.current_timestamp(), server_default=func.current_timestamp(),
onupdate=func.current_timestamp(), onupdate=func.current_timestamp(),
@@ -63,8 +66,8 @@ class ApiToken(IdentityMixin, TimestampMixin, Base):
owner_id: Mapped[str] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT")) owner_id: Mapped[str] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT"))
token_hash: Mapped[str] = mapped_column(Text, nullable=False, unique=True) token_hash: Mapped[str] = mapped_column(Text, nullable=False, unique=True)
scopes: Mapped[list[str]] = mapped_column(JSON, nullable=False) scopes: Mapped[list[str]] = mapped_column(JSON, nullable=False)
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) expires_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) revoked_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
__table_args__ = (Index("ix_api_tokens_owner_id", "owner_id"),) __table_args__ = (Index("ix_api_tokens_owner_id", "owner_id"),)
@@ -134,7 +137,7 @@ class Schedule(IdentityMixin, TimestampMixin, Base):
misfire_grace_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=900) misfire_grace_seconds: Mapped[int] = mapped_column(Integer, nullable=False, default=900)
overlap_policy: Mapped[str] = mapped_column(String(32), nullable=False, default="prohibit") overlap_policy: Mapped[str] = mapped_column(String(32), nullable=False, default="prohibit")
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
next_nominal_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) next_nominal_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
last_enqueue_outcome: Mapped[str | None] = mapped_column(String(64)) last_enqueue_outcome: Mapped[str | None] = mapped_column(String(64))
__table_args__ = ( __table_args__ = (
CheckConstraint("misfire_grace_seconds >= 0", name="misfire_nonnegative"), CheckConstraint("misfire_grace_seconds >= 0", name="misfire_nonnegative"),
@@ -146,18 +149,18 @@ class Execution(IdentityMixin, TimestampMixin, Base):
__tablename__ = "executions" __tablename__ = "executions"
job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id", ondelete="RESTRICT")) job_id: Mapped[str] = mapped_column(ForeignKey("jobs.id", ondelete="RESTRICT"))
schedule_id: Mapped[str | None] = mapped_column(ForeignKey("schedules.id", ondelete="RESTRICT")) schedule_id: Mapped[str | None] = mapped_column(ForeignKey("schedules.id", ondelete="RESTRICT"))
nominal_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) nominal_run_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
trigger: Mapped[str] = mapped_column(String(32), nullable=False) trigger: Mapped[str] = mapped_column(String(32), nullable=False)
state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued") state: Mapped[str] = mapped_column(String(32), nullable=False, default="queued")
attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=1) attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
lease_owner: Mapped[str | None] = mapped_column(String(255)) lease_owner: Mapped[str | None] = mapped_column(String(255))
lease_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) lease_expires_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
heartbeat_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) heartbeat_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
progress: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) progress: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
reason_code: Mapped[str | None] = mapped_column(String(64)) reason_code: Mapped[str | None] = mapped_column(String(64))
operator_message: Mapped[str | None] = mapped_column(Text) operator_message: Mapped[str | None] = mapped_column(Text)
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) started_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) completed_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
__table_args__ = ( __table_args__ = (
CheckConstraint("attempt > 0", name="attempt_positive"), CheckConstraint("attempt > 0", name="attempt_positive"),
CheckConstraint( CheckConstraint(
@@ -192,9 +195,9 @@ class Backup(IdentityMixin, Base):
stored_bytes: Mapped[int] = mapped_column(Integer, nullable=False) stored_bytes: Mapped[int] = mapped_column(Integer, nullable=False)
integrity: Mapped[str] = mapped_column(String(32), nullable=False, default="unverified") integrity: Mapped[str] = mapped_column(String(32), nullable=False, default="unverified")
pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) pinned: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
tombstoned_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) tombstoned_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.current_timestamp() UTCDateTime(), nullable=False, server_default=func.current_timestamp()
) )
__table_args__ = ( __table_args__ = (
CheckConstraint("logical_bytes >= 0", name="logical_bytes_nonnegative"), CheckConstraint("logical_bytes >= 0", name="logical_bytes_nonnegative"),
@@ -233,7 +236,7 @@ class AuditEvent(IdentityMixin, Base):
request_id: Mapped[str] = mapped_column(String(36), nullable=False) request_id: Mapped[str] = mapped_column(String(36), nullable=False)
details: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) details: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.current_timestamp() UTCDateTime(), nullable=False, server_default=func.current_timestamp()
) )
__table_args__ = ( __table_args__ = (
CheckConstraint("outcome IN ('success','failure','denied')", name="outcome"), CheckConstraint("outcome IN ('success','failure','denied')", name="outcome"),
@@ -264,7 +267,7 @@ class NotificationDelivery(IdentityMixin, TimestampMixin, Base):
attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=1) attempt: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
state: Mapped[str] = mapped_column(String(32), nullable=False) state: Mapped[str] = mapped_column(String(32), nullable=False)
response_class: Mapped[str | None] = mapped_column(String(64)) response_class: Mapped[str | None] = mapped_column(String(64))
next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) next_attempt_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
__table_args__ = ( __table_args__ = (
CheckConstraint("attempt > 0", name="attempt_positive"), CheckConstraint("attempt > 0", name="attempt_positive"),
CheckConstraint("state IN ('pending','delivered','retry','failed')", name="state"), CheckConstraint("state IN ('pending','delivered','retry','failed')", name="state"),
@@ -282,7 +285,7 @@ class IdempotencyRecord(IdentityMixin, Base):
response_resource_type: Mapped[str] = mapped_column(String(64), nullable=False) response_resource_type: Mapped[str] = mapped_column(String(64), nullable=False)
response_resource_id: Mapped[str] = mapped_column(String(36), nullable=False) response_resource_id: Mapped[str] = mapped_column(String(36), nullable=False)
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.current_timestamp() UTCDateTime(), nullable=False, server_default=func.current_timestamp()
) )
__table_args__ = ( __table_args__ = (
UniqueConstraint("actor_id", "key", "operation", name="uq_idempotency_actor_key_operation"), UniqueConstraint("actor_id", "key", "operation", name="uq_idempotency_actor_key_operation"),
+20
View File
@@ -0,0 +1,20 @@
from __future__ import annotations
from typing import Any
DEFAULT_BUSY_TIMEOUT_MS = 5_000
def sqlite_connect_args(busy_timeout_ms: int) -> dict[str, float]:
"""Configure SQLite lock waiting without dynamic PRAGMA SQL."""
return {"timeout": busy_timeout_ms / 1_000}
def configure_sqlite_connection(dbapi_connection: Any) -> None:
"""Apply connection-local safety and persistent WAL mode."""
cursor = dbapi_connection.cursor()
try:
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
finally:
cursor.close()
+34
View File
@@ -0,0 +1,34 @@
from __future__ import annotations
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import DateTime
from sqlalchemy.engine import Dialect
from sqlalchemy.types import TypeDecorator
class UTCDateTime(TypeDecorator[datetime]):
"""Store instants and always return timezone-aware UTC datetimes."""
impl = DateTime
cache_ok = True
def load_dialect_impl(self, dialect: Dialect) -> Any:
return dialect.type_descriptor(DateTime(timezone=True))
def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
del dialect
if value is None:
return None
if value.tzinfo is None:
raise ValueError("datetime values must be timezone-aware")
return value.astimezone(UTC)
def process_result_value(self, value: datetime | None, dialect: Dialect) -> datetime | None:
del dialect
if value is None:
return None
if value.tzinfo is None:
return value.replace(tzinfo=UTC)
return value.astimezone(UTC)
+21 -2
View File
@@ -35,9 +35,28 @@ Observed results:
- Frontend typecheck/build: passed. - Frontend typecheck/build: passed.
- Disposable database removed after verification. - Disposable database removed after verification.
## Review-Fix RED/GREEN
RED commit: `45a526a test(v2): prove runtime dispatch and persistence invariants`
The focused M1 command exited 1 with five intended failures: role dispatch was absent, migrate did not dispatch Alembic, database roles accepted an unmigrated DB, persisted ORM IDs had no default, and migration connections had no shared SQLite configurator.
GREEN verification after implementing the missing behavior:
- CLI dispatch and `migrate upgrade` tests passed.
- Unmigrated worker startup was rejected before its handler ran.
- Persisted User and Repository IDs were UUIDv7.
- Generated and non-UTC supplied timestamps reloaded as aware UTC.
- Alembic connection observation reported `foreign_keys=1`, `journal_mode=wal`, and `busy_timeout=7000`.
- Baseline active-execution index, schedule uniqueness, and execution checks were present.
- Focused M1 suite: `19 passed`.
- Fast unit/contract suite: `30 passed`.
- Alembic upgrade/downgrade/upgrade, Ruff, strict mypy, frontend typecheck/build, forbidden-v1 scan, and `git diff --check`: passed.
## Hand Review ## Hand Review
- Baseline revision contains explicit `op.create_table`, constraints, foreign keys, and indexes for all 14 v2 entities; it does not call `create_all`. - Baseline revision contains explicit `op.create_table`, constraints, foreign keys, and indexes for all 14 v2 entities; it does not call `create_all`.
- SQLite connections enable foreign keys, WAL, and a bounded busy timeout. - Runtime and Alembic share one SQLite connection configurator; lock waiting uses the driver timeout rather than interpolated PRAGMA SQL.
- Startup compares the database Alembic revision with the current script head and rejects unmigrated databases. - Startup compares the database Alembic revision with the current script head and rejects unmigrated databases.
- Runtime roles are explicit CLI subcommands; no role starts another role in M1. - Every ORM identity has a UUIDv7 default; every persisted datetime uses the aware-UTC normalizing type.
- Runtime roles are explicit CLI subcommands and dispatch only after the schema gate; `migrate` invokes Alembic directly.