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

.env.example

@@ -9,6 +9,26 @@ JWT_SECRET=change-me-in-production
 # logs a startup warning when this is unset. See docs/api-reference.md.
 OPENSHIELD_AUTHORIZED_SUBSCRIPTIONS=
 
+# Optional - durable scan worker tuning. A claim is held for SCAN_LEASE_SECONDS
+# and renewed every SCAN_HEARTBEAT_SECONDS; the heartbeat MUST be shorter than
+# the lease or the worker loses its own claim mid-scan. A non-numeric or
+# non-positive value logs a warning and uses the default; a heartbeat >= the
+# lease logs a warning and uses one third of the lease. Defaults: 900 and 300.
+SCAN_LEASE_SECONDS=900
+SCAN_HEARTBEAT_SECONDS=300
+
+# Optional - explicit per-subscription hourly admission quota. Unset or 0 keeps
+# the historical no-time-window policy (a non-numeric value logs a warning and
+# disables the quota); one active (pending/running) scan per subscription is
+# always enforced regardless of this value.
+OPENSHIELD_MAX_SCANS_PER_SUBSCRIPTION_PER_HOUR=0
+
+# Optional - how long a retired worker's heartbeat row is kept before it is
+# pruned. Worker identities are per-process, so this bounds worker_heartbeats
+# across restarts. Must stay far above SCAN_HEARTBEAT_SECONDS so that a live
+# worker is never pruned. Default: 604800 (7 days).
+WORKER_HEARTBEAT_RETENTION_SECONDS=604800
+
 # AI providers - add at least one
 ANTHROPIC_API_KEY=
 GROQ_API_KEY=

alembic/versions/a7c5e9d2f1b4_scan_admission_idempotency.py

@@ -0,0 +1,101 @@
+"""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
+
+_ACTIVE_INDEX = "uq_scans_one_active_per_subscription"
+_KEY_INDEX = "uq_scans_subscription_idempotency_key"
+
+
+def _assert_one_active_scan_per_subscription() -> None:
+    """Fail with an actionable error instead of an unusable index.
+
+    ``CREATE UNIQUE INDEX CONCURRENTLY`` on a table that already violates the
+    constraint fails *and* leaves an INVALID index behind.  A deployment that
+    predates the one-active-scan rule can legitimately hold several
+    ``pending``/``running`` rows for one subscription, so this checks first and
+    reports exactly which subscriptions block the upgrade.  Choosing which of
+    those scans is authoritative is an operator decision -- deleting or
+    completing production scan history automatically is never this migration's
+    call.
+    """
+    rows = (
+        op.get_bind()
+        .execute(
+            sa.text(
+                """
+                SELECT subscription_id, COUNT(*) AS active
+                FROM scans
+                WHERE status IN ('pending', 'running')
+                GROUP BY subscription_id
+                HAVING COUNT(*) > 1
+                ORDER BY active DESC, subscription_id
+                """
+            )
+        )
+        .fetchall()
+    )
+    if not rows:
+        return
+    detail = ", ".join(f"{subscription_id} ({active} active)" for subscription_id, active in rows)
+    raise RuntimeError(
+        "Cannot enforce one active scan per subscription: "
+        f"{len(rows)} subscription(s) already have more than one pending/running scan: {detail}. "
+        "Resolve them first (let the scans finish, or mark the superseded rows "
+        "'failed'), then re-run this migration. "
+        "See docs/async-scan-architecture.md for the documented cleanup order."
+    )
+
+
+def upgrade() -> None:
+    """Persist idempotency semantics and prevent more than one active scan."""
+    # Added before the preflight and the concurrent index builds, both of
+    # which can fail; autocommit_block() has already committed this column by
+    # then, so a retry must tolerate it already being there.
+    op.execute("ALTER TABLE scans ADD COLUMN IF NOT EXISTS idempotency_key TEXT")
+
+    # Checked before either index is built so a blocked upgrade leaves the
+    # schema exactly as it was, with the added columns unused and harmless.
+    _assert_one_active_scan_per_subscription()
+
+    with op.get_context().autocommit_block():
+        # An earlier interrupted or failed CONCURRENTLY build leaves an INVALID
+        # index that cannot serve queries but does occupy the name. Drop both
+        # names first so retrying this migration is deterministic.
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_KEY_INDEX}")
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_ACTIVE_INDEX}")
+        op.execute(
+            f"""
+            CREATE UNIQUE INDEX CONCURRENTLY {_KEY_INDEX}
+            ON scans (subscription_id, idempotency_key)
+            WHERE idempotency_key IS NOT NULL
+            """
+        )
+        op.execute(
+            f"""
+            CREATE UNIQUE INDEX CONCURRENTLY {_ACTIVE_INDEX}
+            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(f"DROP INDEX CONCURRENTLY IF EXISTS {_ACTIVE_INDEX}")
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_KEY_INDEX}")
+    op.drop_column("scans", "idempotency_key")

alembic/versions/c9e1a5b7d3f2_durable_enrichment_jobs.py

@@ -0,0 +1,80 @@
+"""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."""
+    # autocommit_block() below commits the table before the concurrent index
+    # builds run, so a failure there leaves the table behind with
+    # alembic_version unchanged. Skip the create on a retry rather than
+    # failing on "relation already exists" before the index recovery.
+    if not sa.inspect(op.get_bind()).has_table("enrichment_jobs"):
+        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():
+        # Drop first so an INVALID index left by an interrupted build is
+        # rebuilt rather than kept (see the note in e4f7a9b2c6d8).
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_enrichment_jobs_pending_retry")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_enrichment_jobs_pending_retry
+            ON enrichment_jobs (next_retry_at ASC)
+            WHERE status = 'pending'
+            """
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_enrichment_jobs_running_lease")
+        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,59 @@
+"""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."""
+    # autocommit_block() below commits this table before the concurrent index
+    # build, so a failure there leaves it behind with alembic_version
+    # unchanged. Skip on a retry rather than failing before the recovery.
+    if not sa.inspect(op.get_bind()).has_table("worker_heartbeats"):
+        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
+        )
+
+    # /metrics reports the last successful scan on every scrape. Without this
+    # the aggregate degrades into a sequential scan of the whole scans table as
+    # scan history grows; the partial index keeps it an index-only lookup.
+    with op.get_context().autocommit_block():
+        # Drop first so an INVALID index left by an interrupted build is
+        # rebuilt rather than kept (see the note in e4f7a9b2c6d8).
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_completed_completed_at")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_scans_completed_completed_at
+            ON scans (completed_at DESC)
+            WHERE status = 'completed'
+            """
+        )
+
+
+def downgrade() -> None:
+    """Remove durable worker heartbeat state."""
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_completed_completed_at")
+    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,75 @@
+"""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
+
+
+revision: str = "e4f7a9b2c6d8"
+down_revision: Union[str, Sequence[str], None] = "3f59f83a5253"
+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."""
+    # autocommit_block() below commits everything issued before it, so a
+    # failure while building the concurrent indexes leaves these columns in
+    # place with alembic_version still on the previous revision. The retry has
+    # to be able to walk back over them instead of failing on "column already
+    # exists" before it reaches the index recovery.
+    op.execute("ALTER TABLE scans ADD COLUMN IF NOT EXISTS lease_owner TEXT")
+    op.execute("ALTER TABLE scans ADD COLUMN IF NOT EXISTS lease_expires_at TIMESTAMPTZ")
+    op.execute("ALTER TABLE scans ADD COLUMN IF NOT EXISTS last_heartbeat_at TIMESTAMPTZ")
+    op.execute("ALTER TABLE scans ADD COLUMN IF NOT EXISTS fencing_token BIGINT NOT NULL DEFAULT 0")
+
+    # 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():
+        # An interrupted CONCURRENTLY build leaves an INVALID index that still
+        # owns the name and can never serve a query. Dropping first (rather
+        # than CREATE ... IF NOT EXISTS, which would keep the broken one) makes
+        # a retry rebuild it.
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_pending_started_at")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_scans_pending_started_at
+            ON scans (started_at ASC)
+            WHERE status = 'pending'
+            """
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_scans_running_lease_expires_at")
+        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_finding_identities.py

@@ -0,0 +1,43 @@
+"""Add database-enforced identities for scan findings.
+
+Revision ID: f2b6d8e1a4c9
+Revises: e4f7a9b2c6d8
+Create Date: 2026-08-29 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+
+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
+
+_UNIQUE_INDEX = "uq_findings_scan_finding_key"
+
+
+def upgrade() -> None:
+    """Give every finding a stable identity so replayed results upsert."""
+    # Rerunnable after a failed concurrent index build (see the note in
+    # e4f7a9b2c6d8): the backfill and the NOT NULL tightening below are both
+    # idempotent, so the whole revision can be replayed.
+    op.execute("ALTER TABLE findings ADD COLUMN IF NOT EXISTS finding_key TEXT")
+    # 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():
+        # A previous interrupted CONCURRENTLY build leaves an unusable index
+        # behind that would make this statement fail with "already exists".
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_UNIQUE_INDEX}")
+        op.execute(f"CREATE UNIQUE INDEX CONCURRENTLY {_UNIQUE_INDEX} ON findings (scan_id, finding_key)")
+
+
+def downgrade() -> None:
+    """Remove the finding identity introduced by this revision."""
+    with op.get_context().autocommit_block():
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_UNIQUE_INDEX}")
+    op.drop_column("findings", "finding_key")