From 63c977cea54eb34188587abff1175b8587dbe675 Mon Sep 17 00:00:00 2001 From: Alex Blank Date: Mon, 27 Jul 2026 23:00:18 +0200 Subject: [PATCH] feat(v2): persist execution event revisions --- .../alembic/versions/0003_execution_events.py | 37 ++++++++++++++++++ backend/src/backup_tool/api/app.py | 39 +++++++++++-------- backend/src/backup_tool/db/models.py | 14 +++++++ backend/src/backup_tool/execution.py | 32 +++++++++++++-- tests/integration/test_migrations.py | 1 + 5 files changed, 104 insertions(+), 19 deletions(-) create mode 100644 backend/alembic/versions/0003_execution_events.py diff --git a/backend/alembic/versions/0003_execution_events.py b/backend/alembic/versions/0003_execution_events.py new file mode 100644 index 0000000..073ee49 --- /dev/null +++ b/backend/alembic/versions/0003_execution_events.py @@ -0,0 +1,37 @@ +"""add durable execution event revision + +Revision ID: 0003_execution_events +Revises: 0002_sessions +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0003_execution_events" +down_revision = "0002_sessions" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("executions") as batch: + batch.add_column(sa.Column("revision", sa.Integer(), nullable=False, server_default="0")) + op.create_table( + "execution_events", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column( + "execution_id", + sa.String(36), + sa.ForeignKey("executions.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("payload", sa.JSON(), nullable=False), + sa.UniqueConstraint("execution_id", "revision", name="uq_execution_event_revision"), + ) + + +def downgrade() -> None: + op.drop_table("execution_events") + with op.batch_alter_table("executions") as batch: + batch.drop_column("revision") diff --git a/backend/src/backup_tool/api/app.py b/backend/src/backup_tool/api/app.py index 5561300..e94e42b 100644 --- a/backend/src/backup_tool/api/app.py +++ b/backend/src/backup_tool/api/app.py @@ -23,6 +23,7 @@ from backup_tool.db.models import ( ApiToken, AuditEvent, Execution, + ExecutionEvent, IdempotencyRecord, Job, Repository, @@ -58,28 +59,34 @@ from backup_tool.security.redaction import redact from backup_tool.security.secrets import EnvelopeCipher -def execution_event_id(execution: Execution) -> str: - """Stable monotonic cursor for an execution's current durable revision.""" - updated = execution.updated_at.astimezone(UTC).isoformat() - return f"{updated}:{execution.state}:{execution.attempt}" - - async def execution_events_stream( sessions: async_sessionmaker[AsyncSession], execution_id: str, last_event_id: str | None ) -> AsyncGenerator[str, None]: - """Poll durable execution state and yield each changed redacted revision as SSE.""" - last_sent = last_event_id + """Replay every durable event after Last-Event-ID, then poll for appended events.""" + try: + last_revision = int(last_event_id or "0") + except ValueError: + last_revision = 0 while True: async with sessions() as stream_db: + events = ( + await stream_db.scalars( + select(ExecutionEvent) + .where( + ExecutionEvent.execution_id == execution_id, + ExecutionEvent.revision > last_revision, + ) + .order_by(ExecutionEvent.revision) + ) + ).all() + for event in events: + yield ( + f"id: {event.revision}\\nevent: execution\\n" + f"data: {json.dumps(event.payload)}\\n\\n" + ) + last_revision = event.revision execution = await stream_db.get(Execution, execution_id) - if execution is None: - return - event_id = execution_event_id(execution) - if event_id != last_sent: - payload = public_event(execution) - yield (f"id: {event_id}\\nevent: execution\\ndata: {json.dumps(payload)}\\n\\n") - last_sent = event_id - if execution.state in TERMINAL_STATES: + if execution is None or (execution.state in TERMINAL_STATES and not events): return await asyncio.sleep(0.1) diff --git a/backend/src/backup_tool/db/models.py b/backend/src/backup_tool/db/models.py index f3a15e3..efcbe07 100644 --- a/backend/src/backup_tool/db/models.py +++ b/backend/src/backup_tool/db/models.py @@ -165,6 +165,7 @@ class Execution(IdentityMixin, TimestampMixin, Base): lease_expires_at: Mapped[datetime | None] = mapped_column(UTCDateTime()) heartbeat_at: Mapped[datetime | None] = mapped_column(UTCDateTime()) progress: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0) reason_code: Mapped[str | None] = mapped_column(String(64)) operator_message: Mapped[str | None] = mapped_column(Text) started_at: Mapped[datetime | None] = mapped_column(UTCDateTime()) @@ -189,6 +190,19 @@ class Execution(IdentityMixin, TimestampMixin, Base): ) +class ExecutionEvent(Base): + __tablename__ = "execution_events" + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + execution_id: Mapped[str] = mapped_column( + ForeignKey("executions.id", ondelete="CASCADE"), nullable=False + ) + revision: Mapped[int] = mapped_column(Integer, nullable=False) + payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + __table_args__ = ( + UniqueConstraint("execution_id", "revision", name="uq_execution_event_revision"), + ) + + class Backup(IdentityMixin, Base): __tablename__ = "backups" execution_id: Mapped[str] = mapped_column( diff --git a/backend/src/backup_tool/execution.py b/backend/src/backup_tool/execution.py index 9e412db..582ca6d 100644 --- a/backend/src/backup_tool/execution.py +++ b/backend/src/backup_tool/execution.py @@ -7,7 +7,7 @@ from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession -from backup_tool.db.models import Execution, Job +from backup_tool.db.models import Execution, ExecutionEvent, Job from backup_tool.security.redaction import redact ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"}) @@ -67,6 +67,8 @@ async def enqueue(db: AsyncSession, job_id: str, trigger: str = "manual") -> Exe "execution_active", "Job already has an active execution.", existing ) from error await db.refresh(execution) + await record_event(db, execution) + await db.commit() return execution @@ -89,7 +91,11 @@ async def claim( await db.rollback() return None await db.commit() - return await db.get(Execution, execution_id) + execution = await db.get(Execution, execution_id) + if execution is not None: + await record_event(db, execution) + await db.commit() + return execution async def heartbeat( @@ -128,7 +134,11 @@ async def request_cancellation(db: AsyncSession, execution_id: str) -> Execution await db.rollback() return None await db.commit() - return await db.get(Execution, execution_id) + execution = await db.get(Execution, execution_id) + if execution is not None: + await record_event(db, execution) + await db.commit() + return execution async def complete_cancellation(db: AsyncSession, execution_id: str, owner: str) -> bool: @@ -177,6 +187,8 @@ async def retry(db: AsyncSession, execution_id: str) -> Execution | None: "execution_active", "Job already has an active execution.", active_id ) from error await db.refresh(execution) + await record_event(db, execution) + await db.commit() return execution @@ -196,6 +208,19 @@ def _redact_progress(value: object) -> object: return redact(value) +async def record_event(db: AsyncSession, execution: Execution) -> ExecutionEvent: + """Append the public representation as the next durable event revision.""" + execution.revision += 1 + event = ExecutionEvent( + execution_id=execution.id, + revision=execution.revision, + payload=public_event(execution), + ) + db.add(event) + await db.flush() + return event + + def public_event(execution: Execution) -> dict[str, object]: """Return redacted progress suitable for polling or SSE.""" progress = _redact_progress(execution.progress) @@ -203,6 +228,7 @@ def public_event(execution: Execution) -> dict[str, object]: "id": execution.id, "state": execution.state, "attempt": execution.attempt, + "revision": execution.revision, "reason_code": execution.reason_code, "progress": progress, } diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 1d2142d..3618ae4 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -20,6 +20,7 @@ EXPECTED_TABLES = { "audit_events", "backups", "executions", + "execution_events", "idempotency_records", "jobs", "notification_deliveries",