fix(core): harden scan durability and idempotency (#303)

alembic/versions/a7c5e9d2f1b4_scan_admission_idempotency.py

@@ -0,0 +1,47 @@
+"""Enforce durable scan admission and idempotency.
+
+Revision ID: a7c5e9d2f1b4
+Revises: f2b6d8e1a4c9
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "a7c5e9d2f1b4"
+down_revision: Union[str, Sequence[str], None] = "f2b6d8e1a4c9"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Persist idempotency semantics and prevent more than one active scan."""
+    op.add_column("scans", sa.Column("idempotency_key", sa.Text(), nullable=True))
+    op.add_column("scans", sa.Column("request_fingerprint", sa.Text(), nullable=True))
+    with op.get_context().autocommit_block():
+        op.execute(
+            """
+            CREATE UNIQUE INDEX CONCURRENTLY uq_scans_subscription_idempotency_key
+            ON scans (subscription_id, idempotency_key)
+            WHERE idempotency_key IS NOT NULL
+            """
+        )
+        op.execute(
+            """
+            CREATE UNIQUE INDEX CONCURRENTLY uq_scans_one_active_per_subscription
+            ON scans (subscription_id)
+            WHERE status IN ('pending', 'running')
+            """
+        )
+
+
+def downgrade() -> None:
+    """Remove scan admission metadata and constraints."""
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS uq_scans_one_active_per_subscription")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS uq_scans_subscription_idempotency_key")
+    op.drop_column("scans", "request_fingerprint")
+    op.drop_column("scans", "idempotency_key")

alembic/versions/c9e1a5b7d3f2_durable_enrichment_jobs.py

@@ -0,0 +1,69 @@
+"""Add durable, fenced CVE enrichment jobs.
+
+Revision ID: c9e1a5b7d3f2
+Revises: a7c5e9d2f1b4
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "c9e1a5b7d3f2"
+down_revision: Union[str, Sequence[str], None] = "a7c5e9d2f1b4"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create one resumable enrichment job per scan."""
+    op.create_table(
+        "enrichment_jobs",
+        sa.Column("job_id", postgresql.UUID(), nullable=False),
+        sa.Column("scan_id", postgresql.UUID(), nullable=False),
+        sa.Column("status", sa.Text(), nullable=False, server_default=sa.text("'pending'")),
+        sa.Column("lease_owner", sa.Text(), nullable=True),
+        sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("last_heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("fencing_token", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+        sa.Column("attempt_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column(
+            "next_retry_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP")
+        ),
+        sa.Column("checkpoint", sa.Integer(), nullable=False, server_default=sa.text("0")),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP")
+        ),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.ForeignKeyConstraint(["scan_id"], ["scans.scan_id"], name="enrichment_jobs_scan_id_fkey"),
+        sa.PrimaryKeyConstraint("job_id", name="enrichment_jobs_pkey"),
+        sa.UniqueConstraint("scan_id", name="uq_enrichment_jobs_scan_id"),
+        sa.CheckConstraint("status IN ('pending', 'running', 'completed', 'failed')", name="ck_enrichment_jobs_status"),
+    )
+    with op.get_context().autocommit_block():
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_enrichment_jobs_pending_retry
+            ON enrichment_jobs (next_retry_at ASC)
+            WHERE status = 'pending'
+            """
+        )
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_enrichment_jobs_running_lease
+            ON enrichment_jobs (lease_expires_at ASC)
+            WHERE status = 'running'
+            """
+        )
+
+
+def downgrade() -> None:
+    """Remove durable enrichment work state."""
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_enrichment_jobs_running_lease")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_enrichment_jobs_pending_retry")
+    op.drop_table("enrichment_jobs")

alembic/versions/d4a8c1e6b2f9_operational_worker_metrics.py

@@ -0,0 +1,38 @@
+"""Persist worker liveness used by bounded operational metrics.
+
+Revision ID: d4a8c1e6b2f9
+Revises: c9e1a5b7d3f2
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "d4a8c1e6b2f9"
+down_revision: Union[str, Sequence[str], None] = "c9e1a5b7d3f2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Store one liveness timestamp per worker process."""
+    op.create_table(
+        "worker_heartbeats",
+        sa.Column("worker_id", sa.Text(), nullable=False),
+        sa.Column("worker_type", sa.Text(), nullable=False),
+        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("worker_id", "worker_type", name="worker_heartbeats_pkey"),
+        sa.CheckConstraint("worker_type IN ('scan', 'enrichment')", name="ck_worker_heartbeats_type"),
+    )
+    op.create_index(
+        "idx_worker_heartbeats_type_seen", "worker_heartbeats", ["worker_type", "last_seen_at"], unique=False
+    )
+
+
+def downgrade() -> None:
+    """Remove durable worker heartbeat state."""
+    op.drop_index("idx_worker_heartbeats_type_seen", table_name="worker_heartbeats")
+    op.drop_table("worker_heartbeats")

alembic/versions/e4f7a9b2c6d8_scan_leases_and_fencing.py

@@ -0,0 +1,68 @@
+"""Add renewable ownership leases and fencing tokens to scans.
+
+Revision ID: e4f7a9b2c6d8
+Revises: d8e4f6a1b2c3
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "e4f7a9b2c6d8"
+down_revision: Union[str, Sequence[str], None] = "d8e4f6a1b2c3"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add additive lease state and make legacy running work recoverable."""
+    op.add_column("scans", sa.Column("lease_owner", sa.Text(), nullable=True))
+    op.add_column("scans", sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column("scans", sa.Column("last_heartbeat_at", sa.DateTime(timezone=True), nullable=True))
+    op.add_column(
+        "scans",
+        sa.Column("fencing_token", sa.BigInteger(), server_default=sa.text("0"), nullable=False),
+    )
+
+    # A pre-lease running row belongs to an old worker that cannot satisfy the
+    # new fencing contract. Marking its lease expired preserves the row and
+    # lets the new worker recover it under a fresh owner/token.
+    op.execute(
+        """
+        UPDATE scans
+        SET lease_expires_at = CURRENT_TIMESTAMP
+        WHERE status = 'running' AND lease_expires_at IS NULL
+        """
+    )
+
+    # These indexes are additive and are created concurrently so a populated
+    # production scans table remains available while the migration runs.
+    with op.get_context().autocommit_block():
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_scans_pending_started_at
+            ON scans (started_at ASC)
+            WHERE status = 'pending'
+            """
+        )
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_scans_running_lease_expires_at
+            ON scans (lease_expires_at ASC)
+            WHERE status = 'running'
+            """
+        )
+
+
+def downgrade() -> None:
+    """Remove lease metadata; callers must be rolled back first."""
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_running_lease_expires_at")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_pending_started_at")
+    op.drop_column("scans", "fencing_token")
+    op.drop_column("scans", "last_heartbeat_at")
+    op.drop_column("scans", "lease_expires_at")
+    op.drop_column("scans", "lease_owner")

alembic/versions/f2b6d8e1a4c9_idempotent_findings_and_evaluations.py

@@ -0,0 +1,66 @@
+"""Add database-enforced identities for scan results.
+
+Revision ID: f2b6d8e1a4c9
+Revises: e4f7a9b2c6d8
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "f2b6d8e1a4c9"
+down_revision: Union[str, Sequence[str], None] = "e4f7a9b2c6d8"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Add stable finding keys and per-resource evaluation rows."""
+    op.add_column("findings", sa.Column("finding_key", sa.Text(), nullable=True))
+    # Existing records predate the identity contract. Preserve each record as
+    # distinct rather than attempting to infer equivalence from mutable text.
+    op.execute("UPDATE findings SET finding_key = 'legacy:' || id::text WHERE finding_key IS NULL")
+    op.alter_column("findings", "finding_key", nullable=False)
+
+    with op.get_context().autocommit_block():
+        op.execute("CREATE UNIQUE INDEX CONCURRENTLY uq_findings_scan_finding_key ON findings (scan_id, finding_key)")
+
+    op.create_table(
+        "rule_evaluations",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("scan_id", postgresql.UUID(), nullable=False),
+        sa.Column("rule_id", sa.Text(), nullable=False),
+        sa.Column("resource_id", sa.Text(), nullable=False),
+        sa.Column("resource_type", sa.Text(), server_default=sa.text("''"), nullable=False),
+        sa.Column("status", sa.Text(), nullable=False),
+        sa.Column("reason_code", sa.Text(), nullable=True),
+        sa.Column("reason", sa.Text(), nullable=True),
+        sa.Column("evidence", postgresql.JSONB(), server_default=sa.text("'{}'::jsonb"), nullable=True),
+        sa.Column("finding_id", sa.Integer(), nullable=True),
+        sa.Column("evaluated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["scan_id"], ["scans.scan_id"], name="rule_evaluations_scan_id_fkey"),
+        sa.ForeignKeyConstraint(
+            ["finding_id"], ["findings.id"], name="rule_evaluations_finding_id_fkey", ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id", name="rule_evaluations_pkey"),
+        sa.UniqueConstraint("scan_id", "rule_id", "resource_id", name="uq_rule_evaluations_scan_rule_resource"),
+        sa.CheckConstraint(
+            "status IN ('PASS', 'FAIL', 'UNKNOWN', 'ERROR', 'NOT_APPLICABLE')",
+            name="ck_rule_evaluations_status_v1",
+        ),
+        sa.CheckConstraint("resource_id <> ''", name="ck_rule_evaluations_resource_id_not_empty"),
+    )
+    op.create_index("idx_rule_evaluations_scan_id", "rule_evaluations", ["scan_id"], unique=False)
+
+
+def downgrade() -> None:
+    """Remove idempotent-result storage introduced by this revision."""
+    op.drop_index("idx_rule_evaluations_scan_id", table_name="rule_evaluations")
+    op.drop_table("rule_evaluations")
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS uq_findings_scan_finding_key")
+    op.drop_column("findings", "finding_key")