From d9195a2d2b26d16b95862d2ed1336c264e769c0f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 09:23:27 +0200 Subject: [PATCH] Reconcile orphaned Campaign artifacts --- README.md | 4 + docs/CAMPAIGN_BUILD_RECOVERY.md | 63 +- docs/CAMPAIGN_DELIVERY_RUNBOOK.md | 7 +- docs/CAMPAIGN_HANDBOOK.md | 6 + .../backend/artifact_reconciliation.py | 734 ++++++++++++++++++ src/govoplan_campaign/backend/manifest.py | 11 +- .../backend/persistence/campaigns.py | 11 + src/govoplan_campaign/backend/router.py | 2 + .../backend/routes/operations.py | 102 +++ src/govoplan_campaign/backend/schemas.py | 49 ++ tests/test_artifact_reconciliation.py | 337 ++++++++ tests/test_route_registration.py | 8 +- 12 files changed, 1320 insertions(+), 14 deletions(-) create mode 100644 src/govoplan_campaign/backend/artifact_reconciliation.py create mode 100644 src/govoplan_campaign/backend/routes/operations.py create mode 100644 tests/test_artifact_reconciliation.py diff --git a/README.md b/README.md index 241d94a..de8d255 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,10 @@ frozen evidence before delivery. Build failure compensates objects written before database commit. Retention uses a fenced forward-recovery operation and independently verifies both artifact absence and the committed locator update; partial or unobservable cleanup remains visible in Ops. +An operator-only, dry-run-first reconciler inventories bounded tenant-prefix +pages and removes only old objects that remain unreferenced after an active +build-fence check. Applied runs are idempotent, fenced, audited, and preserve +database references on every storage failure. ## Dependencies diff --git a/docs/CAMPAIGN_BUILD_RECOVERY.md b/docs/CAMPAIGN_BUILD_RECOVERY.md index 6291f36..d3174a2 100644 --- a/docs/CAMPAIGN_BUILD_RECOVERY.md +++ b/docs/CAMPAIGN_BUILD_RECOVERY.md @@ -7,20 +7,69 @@ opaque object prefix. A repeated idempotency key can replay only a verified successful build; it cannot start a second active build. Generated EML and bounded print output are written to shared storage and checked -for exact size and SHA-256 content. Campaign then commits its jobs and execution -snapshot, compares the stored object and database manifests, and records -verified success. A managed Files output makes the operation forward-recoverable -because Campaign cannot undo a Files-owned artifact; object-only builds use -explicit compensation. +for exact size and SHA-256 content. Campaign renews the build fence after that +check and before changing jobs, then commits its jobs and execution snapshot, +compares the stored object and database manifests, and records verified success. +A managed Files output makes the operation forward-recoverable because Campaign +cannot undo a Files-owned artifact; object-only builds use explicit +compensation. If the Campaign database transaction fails, Campaign deletes every object it recorded and verifies absence before recording recovered state. Failed deletion, an unavailable storage check, process loss, or superseded-object cleanup failure leaves a recovery-required operation visible in Ops. Do not retry such an operation as a normal build. Verify its checkpoint chain and reserved prefix, -then reconcile it through the owning-module procedure. The orphan inventory -reconciler tracked in Campaign issue 91 will automate that bounded inspection. +then reconcile it through the owning-module procedure. + +## Orphan inventory and cleanup + +The operator-only endpoint +`POST /api/v1/campaigns/operations/artifacts/reconcile` requires +`system:settings:write`. It never scans outside +`campaign-artifacts/{tenant_id}/`, and one request reads at most `page_size` +objects. The default request is a dry run: + +```json +{} +``` + +The response reports each eligible key, size, modification time, age, reason, +page totals, and `next_cursor`. Continue with that cursor to inspect the next +bounded page. Objects are not eligible when they are referenced by a committed +EML or print-output row, belong to an actively fenced build, have no trustworthy +modification time, have an invalid build-key shape, or are younger than the +grace period. The minimum and default grace period is 24 hours. + +Apply an inspected page with a new idempotency key: + +```json +{ + "apply": true, + "idempotency_key": "incident-2026-08-03-page-1", + "grace_period_hours": 24, + "page_size": 250 +} +``` + +An applied run rechecks committed references and active build leases before +each bounded deletion batch. A Core distributed lease prevents two nodes from +committing the same tenant cleanup concurrently. Every attempted deletion is +probed afterward. `recovery_required` means an object was verified to remain; +`outcome_unknown` means storage could not prove whether the delete took effect. +Use a new idempotency key to retry after the storage problem is corrected. A +successful repeated request with the same key returns `already_completed` and +does not delete again. + +Cleanup never clears Campaign database references. Audit and recovery evidence +records counts and hashed manifests rather than object keys. Exact keys are +returned only by this privileged endpoint and must stay in restricted incident +records. No recovery checkpoint contains message bodies, recipients, credentials, or resolved provider secrets. Object keys remain restricted diagnostics rather than Campaign business data. + +Database rows, `campaign-artifacts/` objects, and the encryption/key service are +one coordinated backup and recovery boundary. After any partial restore, pause +delivery, run a dry inventory, reconcile Campaign recovery operations in Ops, +and verify referenced object hashes before workers resume. diff --git a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md index 51becff..dce71a0 100644 --- a/docs/CAMPAIGN_DELIVERY_RUNBOOK.md +++ b/docs/CAMPAIGN_DELIVERY_RUNBOOK.md @@ -169,8 +169,11 @@ before attempting delivery. leaves an outcome-unknown operation; a deletion/metadata mismatch becomes recovery-required in Ops. - A hard process loss between object creation and metadata commit can leave an - orphan object. Reconcile only within the Campaign build prefix and verify that - no job references the object before deleting it. + orphan object. Use the operator-only, dry-run-first Campaign artifact + reconciler documented in `CAMPAIGN_BUILD_RECOVERY.md`; it scans one bounded + tenant-prefix page, enforces a minimum 24-hour grace period, protects active + build fences, and rechecks committed EML and print-output references before + deletion. - Restore Campaign rows, object storage, and the encryption key to one coordinated recovery point before resuming workers. diff --git a/docs/CAMPAIGN_HANDBOOK.md b/docs/CAMPAIGN_HANDBOOK.md index 7c6413c..78e7a3f 100644 --- a/docs/CAMPAIGN_HANDBOOK.md +++ b/docs/CAMPAIGN_HANDBOOK.md @@ -397,6 +397,12 @@ After restore, keep outbound delivery paused until queue/attempt state and provider evidence have been reconciled; never let restored accepted jobs send again merely because a queue message was lost. +For unreferenced generated objects after process loss, platform operators first +run the bounded Campaign artifact inventory in dry-run mode. Apply only an +inspected page with a unique incident idempotency key. The cleanup keeps exact +keys out of ordinary Campaign responses, does not clear database references, +and leaves storage failures in the Core recovery ledger for explicit retry. + ### Incident handling 1. Pause new delivery when duplicate or unknown effects are possible. diff --git a/src/govoplan_campaign/backend/artifact_reconciliation.py b/src/govoplan_campaign/backend/artifact_reconciliation.py new file mode 100644 index 0000000..4b4bc61 --- /dev/null +++ b/src/govoplan_campaign/backend/artifact_reconciliation.py @@ -0,0 +1,734 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +import hashlib +import json +from typing import Any, Callable + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from govoplan_campaign.backend.db.models import ( + Campaign, + CampaignJob, + CampaignVersion, +) +from govoplan_core.core.object_storage import ( + StorageBackend, + StorageBackendError, + StorageObjectInfo, +) +from govoplan_core.core.recovery import ( + RecoveryMode, + RecoveryOperation, + RecoveryPlan, + RecoveryStatus, + TERMINAL_RECOVERY_STATUSES, +) +from govoplan_core.core.recovery_runtime import begin_durable_recovery_operation +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, +) + + +CAMPAIGN_ARTIFACT_NAMESPACE = "campaign-artifacts" +MINIMUM_GRACE_HOURS = 24 +MAXIMUM_PAGE_SIZE = 1000 +_CHECKPOINT_BATCH_SIZE = 25 + +SessionFactory = Callable[[], Session] + + +class CampaignArtifactReconciliationError(RuntimeError): + pass + + +@dataclass(slots=True) +class ArtifactCandidate: + key: str + size_bytes: int + modified_at: datetime + age_seconds: int + reason: str = "unreferenced_after_grace_period" + disposition: str = "candidate" + failure_type: str | None = None + + def as_dict(self) -> dict[str, Any]: + return { + "key": self.key, + "size_bytes": self.size_bytes, + "modified_at": self.modified_at.isoformat(), + "age_seconds": self.age_seconds, + "reason": self.reason, + "disposition": self.disposition, + "failure_type": self.failure_type, + } + + +@dataclass(slots=True) +class ArtifactInventory: + tenant_prefix: str + cursor: str | None + next_cursor: str | None + scanned_count: int + scanned_bytes: int + referenced_count: int + active_build_count: int + young_count: int + unknown_age_count: int + invalid_shape_count: int + candidates: list[ArtifactCandidate] + manifest_sha256: str + + @property + def candidate_bytes(self) -> int: + return sum(candidate.size_bytes for candidate in self.candidates) + + def response( + self, + *, + apply: bool, + status: str, + recovery_operation_id: str | None = None, + ) -> dict[str, Any]: + deleted = [ + candidate + for candidate in self.candidates + if candidate.disposition == "deleted" + ] + failures = [ + candidate + for candidate in self.candidates + if candidate.disposition + in {"delete_failed", "delete_outcome_unknown"} + ] + return { + "apply": apply, + "status": status, + "recovery_operation_id": recovery_operation_id, + "tenant_prefix": self.tenant_prefix, + "cursor": self.cursor, + "next_cursor": self.next_cursor, + "scanned_count": self.scanned_count, + "scanned_bytes": self.scanned_bytes, + "referenced_count": self.referenced_count, + "active_build_count": self.active_build_count, + "young_count": self.young_count, + "unknown_age_count": self.unknown_age_count, + "invalid_shape_count": self.invalid_shape_count, + "candidate_count": len(self.candidates), + "candidate_bytes": self.candidate_bytes, + "deleted_count": len(deleted), + "deleted_bytes": sum(candidate.size_bytes for candidate in deleted), + "failure_count": len(failures), + "manifest_sha256": self.manifest_sha256, + "candidates": [candidate.as_dict() for candidate in self.candidates], + } + + +def campaign_artifact_inventory( + session: Session, + *, + storage: StorageBackend, + tenant_id: str, + grace_period: timedelta, + cursor: str | None = None, + page_size: int = 250, + now: datetime | None = None, +) -> ArtifactInventory: + observed_at = _as_utc(now or datetime.now(timezone.utc)) + if grace_period < timedelta(hours=MINIMUM_GRACE_HOURS): + raise ValueError( + f"Campaign artifact grace period must be at least {MINIMUM_GRACE_HOURS} hours" + ) + bounded_page_size = max(1, min(int(page_size), MAXIMUM_PAGE_SIZE)) + prefix = _tenant_artifact_prefix(tenant_id) + if cursor is not None and not cursor.startswith(prefix): + raise ValueError("Campaign artifact cursor is outside the tenant namespace") + + page = storage.list_objects( + prefix=prefix, + after=cursor, + limit=bounded_page_size, + ) + page_keys = {info.key for info in page.objects} + referenced_keys = _referenced_artifact_keys( + session, + tenant_id=tenant_id, + prefix=prefix, + artifact_keys=page_keys, + ) + active_build_ids = _active_build_ids( + session, + tenant_id=tenant_id, + now=observed_at, + ) + + candidates: list[ArtifactCandidate] = [] + referenced_count = 0 + active_build_count = 0 + young_count = 0 + unknown_age_count = 0 + invalid_shape_count = 0 + scanned_bytes = 0 + manifest_rows: list[dict[str, Any]] = [] + for info in page.objects: + scanned_bytes += info.size_bytes + build_id = _build_id_for_key(info.key, prefix=prefix) + modified_at = _object_modified_at(info) + manifest_rows.append( + { + "key_sha256": _sha256(info.key), + "size_bytes": info.size_bytes, + "modified_at": modified_at.isoformat() if modified_at else None, + } + ) + if build_id is None: + invalid_shape_count += 1 + continue + if info.key in referenced_keys: + referenced_count += 1 + continue + if build_id in active_build_ids: + active_build_count += 1 + continue + if modified_at is None: + unknown_age_count += 1 + continue + age = observed_at - modified_at + if age < grace_period: + young_count += 1 + continue + candidates.append( + ArtifactCandidate( + key=info.key, + size_bytes=info.size_bytes, + modified_at=modified_at, + age_seconds=max(0, int(age.total_seconds())), + ) + ) + + return ArtifactInventory( + tenant_prefix=prefix, + cursor=cursor, + next_cursor=page.next_cursor, + scanned_count=len(page.objects), + scanned_bytes=scanned_bytes, + referenced_count=referenced_count, + active_build_count=active_build_count, + young_count=young_count, + unknown_age_count=unknown_age_count, + invalid_shape_count=invalid_shape_count, + candidates=candidates, + manifest_sha256=_canonical_sha256(manifest_rows), + ) + + +def reconcile_campaign_artifacts( + session_factory: SessionFactory, + *, + storage: StorageBackend, + identity: RuntimeIdentity, + tenant_id: str, + apply: bool = False, + idempotency_key: str | None = None, + grace_period_hours: int = MINIMUM_GRACE_HOURS, + cursor: str | None = None, + page_size: int = 250, + now: datetime | None = None, +) -> dict[str, Any]: + if apply and not (idempotency_key or "").strip(): + raise ValueError("Applied Campaign artifact cleanup requires an idempotency key") + if grace_period_hours < MINIMUM_GRACE_HOURS: + raise ValueError( + f"Campaign artifact grace period must be at least {MINIMUM_GRACE_HOURS} hours" + ) + observed_at = _as_utc(now or datetime.now(timezone.utc)) + grace_period = timedelta(hours=grace_period_hours) + if not apply: + with session_factory() as session: + inventory = campaign_artifact_inventory( + session, + storage=storage, + tenant_id=tenant_id, + grace_period=grace_period, + cursor=cursor, + page_size=page_size, + now=observed_at, + ) + return inventory.response(apply=False, status="dry_run") + + prefix = _tenant_artifact_prefix(tenant_id) + request = { + "tenant_id": tenant_id, + "prefix": prefix, + "cursor_sha256": _sha256(cursor) if cursor else None, + "page_size": max(1, min(int(page_size), MAXIMUM_PAGE_SIZE)), + "grace_period_hours": grace_period_hours, + } + recovery_start = begin_durable_recovery_operation( + session_factory, + identity=identity, + module_id="campaigns", + operation_type="artifact-orphan-reconciliation", + idempotency_key=str(idempotency_key).strip(), + request=request, + recovery_plan=RecoveryPlan( + mode=RecoveryMode.FORWARD_RECOVERY, + preconditions=( + "inventory is bounded to one tenant Campaign artifact prefix", + "objects younger than the conservative grace period are excluded", + ), + forward_recovery_steps=( + "retry only objects still unreferenced by committed Campaign state", + ), + verification_steps=( + "probe every attempted object after deletion", + "preserve database references without mutation", + ), + ), + precondition_evidence={ + "tenant_prefix_sha256": _sha256(prefix), + "grace_period_hours": grace_period_hours, + "page_size": request["page_size"], + }, + lease_resource_key=f"campaign:artifact-reconcile:{tenant_id}", + lease_ttl_seconds=15 * 60, + resource_type="campaign_artifact_namespace", + resource_id=tenant_id, + metadata={"tenant_id": tenant_id}, + ) + if recovery_start.replayed: + return _replayed_response( + prefix=prefix, + cursor=cursor, + operation_id=recovery_start.operation_id, + ) + operation = recovery_start.operation + if operation is None: # pragma: no cover - guarded by replay branch + raise CampaignArtifactReconciliationError( + "Campaign artifact cleanup authority was not created" + ) + + try: + try: + with session_factory() as session: + inventory = campaign_artifact_inventory( + session, + storage=storage, + tenant_id=tenant_id, + grace_period=grace_period, + cursor=cursor, + page_size=page_size, + now=observed_at, + ) + except Exception as exc: + operation.fail( + summary="Campaign artifact inventory failed before deletion", + evidence={ + "effect_started": False, + "failure_type": type(exc).__name__, + }, + ) + raise + + operation.checkpoint( + kind="artifact-inventory", + summary="The bounded Campaign artifact inventory was classified", + evidence={ + "manifest_sha256": inventory.manifest_sha256, + "scanned_count": inventory.scanned_count, + "candidate_count": len(inventory.candidates), + "candidate_bytes": inventory.candidate_bytes, + "next_page": inventory.next_cursor is not None, + }, + ) + + if inventory.candidates: + _apply_inventory( + session_factory, + storage=storage, + operation=operation, + inventory=inventory, + tenant_id=tenant_id, + now=observed_at, + ) + + failed = [ + item + for item in inventory.candidates + if item.disposition == "delete_failed" + ] + unknown = [ + item + for item in inventory.candidates + if item.disposition == "delete_outcome_unknown" + ] + evidence = _cleanup_evidence(inventory) + if unknown: + operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="Campaign artifact deletion could not be verified", + evidence=evidence, + failure_summary=( + "One or more Campaign artifact deletion outcomes are unknown" + ), + ) + status = "outcome_unknown" + elif failed: + operation.unresolved( + status=RecoveryStatus.RECOVERY_REQUIRED, + summary="Campaign artifact deletion requires a retry", + evidence=evidence, + failure_summary=( + "One or more unreferenced Campaign artifacts remain" + ), + ) + status = "recovery_required" + else: + operation.succeed(evidence=evidence) + status = "applied" + return inventory.response( + apply=True, + status=status, + recovery_operation_id=recovery_start.operation_id, + ) + except Exception: + if not operation.closed: + try: + operation.release_unresolved() + except Exception: + pass + raise + + +def _apply_inventory( + session_factory: SessionFactory, + *, + storage: StorageBackend, + operation: Any, + inventory: ArtifactInventory, + tenant_id: str, + now: datetime, +) -> None: + for index in range(0, len(inventory.candidates), _CHECKPOINT_BATCH_SIZE): + batch = inventory.candidates[index : index + _CHECKPOINT_BATCH_SIZE] + with session_factory() as session: + referenced = _referenced_artifact_keys( + session, + tenant_id=tenant_id, + prefix=inventory.tenant_prefix, + artifact_keys={candidate.key for candidate in batch}, + ) + active_build_ids = _active_build_ids( + session, + tenant_id=tenant_id, + now=now, + ) + operation.checkpoint( + kind="artifact-delete-batch-authorized", + summary="Deletion authority was renewed for a bounded object batch", + evidence={ + "batch_index": index // _CHECKPOINT_BATCH_SIZE, + "batch_count": len(batch), + "batch_manifest_sha256": _canonical_sha256( + [_sha256(candidate.key) for candidate in batch] + ), + }, + ) + for candidate in batch: + build_id = _build_id_for_key( + candidate.key, + prefix=inventory.tenant_prefix, + ) + if candidate.key in referenced or build_id in active_build_ids: + candidate.disposition = "protected_before_delete" + candidate.reason = "reference_or_active_build_appeared" + continue + _delete_and_verify(storage, candidate) + + +def _delete_and_verify( + storage: StorageBackend, + candidate: ArtifactCandidate, +) -> None: + delete_failure: Exception | None = None + try: + storage.delete(candidate.key) + except (OSError, StorageBackendError) as exc: + delete_failure = exc + try: + remains = storage.exists(candidate.key) + except (OSError, StorageBackendError) as exc: + candidate.disposition = "delete_outcome_unknown" + candidate.failure_type = type(exc).__name__ + return + if not remains: + candidate.disposition = "deleted" + return + candidate.disposition = "delete_failed" + candidate.failure_type = ( + type(delete_failure).__name__ if delete_failure is not None else None + ) + + +def _referenced_artifact_keys( + session: Session, + *, + tenant_id: str, + prefix: str, + artifact_keys: set[str] | None = None, +) -> set[str]: + if artifact_keys == set(): + return set() + keys: set[str] = set() + version_ids = { + identity[1] + for key in artifact_keys or () + if (identity := _artifact_identity(key, prefix=prefix)) is not None + } + job_query = session.query( + CampaignJob.eml_storage_key, + CampaignJob.resolved_print_output, + ).filter(CampaignJob.tenant_id == tenant_id) + if artifact_keys is not None: + filters = [CampaignJob.eml_storage_key.in_(artifact_keys)] + if version_ids: + filters.append(CampaignJob.campaign_version_id.in_(version_ids)) + job_query = job_query.filter(or_(*filters)) + job_rows = job_query.yield_per(1000) + for eml_storage_key, print_output in job_rows: + _add_key( + keys, + eml_storage_key, + prefix=prefix, + allowed_keys=artifact_keys, + ) + _add_key( + keys, + _print_output_storage_key(print_output), + prefix=prefix, + allowed_keys=artifact_keys, + ) + + version_query = session.query(CampaignVersion.build_summary).join( + Campaign, + Campaign.id == CampaignVersion.campaign_id, + ).filter(Campaign.tenant_id == tenant_id) + if artifact_keys is not None: + if not version_ids: + return keys + version_query = version_query.filter(CampaignVersion.id.in_(version_ids)) + version_rows = version_query.yield_per(500) + for (build_summary,) in version_rows: + print_output = ( + build_summary.get("print_output") + if isinstance(build_summary, dict) + else None + ) + _add_key( + keys, + _print_output_storage_key(print_output), + prefix=prefix, + allowed_keys=artifact_keys, + ) + return keys + + +def _artifact_identity( + key: str, + *, + prefix: str, +) -> tuple[str, str, str] | None: + if not key.startswith(prefix): + return None + parts = key[len(prefix) :].split("/") + if len(parts) < 4 or any(not part for part in parts[:4]): + return None + return parts[0], parts[1], parts[2] + + +def _active_build_ids( + session: Session, + *, + tenant_id: str, + now: datetime, +) -> set[str]: + operations = ( + session.query(RecoveryOperation) + .join( + CampaignVersion, + CampaignVersion.id == RecoveryOperation.resource_id, + ) + .join(Campaign, Campaign.id == CampaignVersion.campaign_id) + .filter( + Campaign.tenant_id == tenant_id, + RecoveryOperation.module_id == "campaigns", + RecoveryOperation.operation_type == "build-artifacts", + RecoveryOperation.status.not_in(TERMINAL_RECOVERY_STATUSES), + ) + .all() + ) + resource_keys = { + operation.lease_resource_key + for operation in operations + if operation.lease_resource_key + } + if not resource_keys: + return set() + installation_ids = {operation.installation_id for operation in operations} + leases = ( + session.query(DistributedLease) + .filter( + DistributedLease.installation_id.in_(installation_ids), + DistributedLease.resource_key.in_(resource_keys), + ) + .all() + ) + leases_by_key = { + (lease.installation_id, lease.resource_key): lease for lease in leases + } + active: set[str] = set() + for operation in operations: + lease = leases_by_key.get( + (operation.installation_id, operation.lease_resource_key or "") + ) + if ( + lease is not None + and lease.holder_node_id == operation.holder_node_id + and lease.holder_incarnation == operation.holder_incarnation + and lease.fencing_token == operation.fencing_token + and _as_utc(lease.expires_at) > now + ): + active.add(operation.id) + return active + + +def _cleanup_evidence(inventory: ArtifactInventory) -> dict[str, Any]: + dispositions: dict[str, int] = {} + for candidate in inventory.candidates: + dispositions[candidate.disposition] = ( + dispositions.get(candidate.disposition, 0) + 1 + ) + verified = not any( + key in dispositions for key in ("delete_failed", "delete_outcome_unknown") + ) + return { + "verified": verified, + "checks": { + "candidate_objects": ( + "absent-or-newly-protected" if verified else "incomplete" + ), + "database_references": "unchanged", + "lease_fence": "renewed-before-each-batch", + }, + "inventory_manifest_sha256": inventory.manifest_sha256, + "candidate_manifest_sha256": _canonical_sha256( + [_sha256(candidate.key) for candidate in inventory.candidates] + ), + "candidate_count": len(inventory.candidates), + "candidate_bytes": inventory.candidate_bytes, + "dispositions": dispositions, + "database_references_mutated": False, + } + + +def _replayed_response( + *, + prefix: str, + cursor: str | None, + operation_id: str, +) -> dict[str, Any]: + return { + "apply": True, + "status": "already_completed", + "recovery_operation_id": operation_id, + "tenant_prefix": prefix, + "cursor": cursor, + "next_cursor": None, + "scanned_count": 0, + "scanned_bytes": 0, + "referenced_count": 0, + "active_build_count": 0, + "young_count": 0, + "unknown_age_count": 0, + "invalid_shape_count": 0, + "candidate_count": 0, + "candidate_bytes": 0, + "deleted_count": 0, + "deleted_bytes": 0, + "failure_count": 0, + "manifest_sha256": None, + "candidates": [], + } + + +def _tenant_artifact_prefix(tenant_id: str) -> str: + normalized = str(tenant_id or "").strip() + if not normalized or "/" in normalized or normalized in {".", ".."}: + raise ValueError("Campaign artifact inventory requires a valid tenant id") + return f"{CAMPAIGN_ARTIFACT_NAMESPACE}/{normalized}/" + + +def _build_id_for_key(key: str, *, prefix: str) -> str | None: + identity = _artifact_identity(key, prefix=prefix) + return identity[2] if identity is not None else None + + +def _print_output_storage_key(value: object) -> str | None: + if not isinstance(value, dict): + return None + artifact = value.get("artifact") + if not isinstance(artifact, dict): + return None + key = artifact.get("storage_key") + return str(key) if key else None + + +def _add_key( + keys: set[str], + value: object, + *, + prefix: str, + allowed_keys: set[str] | None = None, +) -> None: + if value is None: + return + key = str(value) + if key.startswith(prefix) and (allowed_keys is None or key in allowed_keys): + keys.add(key) + + +def _object_modified_at(info: StorageObjectInfo) -> datetime | None: + return _as_utc(info.modified_at) if info.modified_at is not None else None + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _sha256(value: object) -> str: + return hashlib.sha256(str(value).encode("utf-8")).hexdigest() + + +def _canonical_sha256(value: object) -> str: + return hashlib.sha256( + json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + default=str, + ).encode("utf-8") + ).hexdigest() + + +__all__ = [ + "CAMPAIGN_ARTIFACT_NAMESPACE", + "CampaignArtifactReconciliationError", + "campaign_artifact_inventory", + "reconcile_campaign_artifacts", +] diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index f051c52..b76cd7e 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -1053,7 +1053,7 @@ manifest = ModuleManifest( id="campaigns.reference.shared-build-artifacts", title="Operate Campaign build artifacts across workers", summary="Generated messages use shared object storage and are verified before delivery.", - body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical build request, validated-version evidence, and reserved object prefix. Before a real Mail, Postbox, or print effect, a separate job-fenced operation records immutable message and recipient digests and later verifies the authoritative Campaign/channel attempt state. Definitive rejection is distinct from accepted, outcome-unknown, and recovery-required work in Ops. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery work. Retention commits locator changes and independently verifies artifact absence; partial cleanup remains recovery-required. Never copy or edit runtime object keys as business data.", + body="Campaign stores generated EML under opaque shared object keys and records expected size, SHA-256 digest, and Message-ID in each job. A worker may run on another node and verifies that evidence before delivery. Before object or Files-owned output effects, a fenced Core recovery operation records the canonical source, validated-version evidence, and reserved object prefix, then renews its fence after object verification and before domain commit. Before a real Mail, Postbox, or print effect, a separate job-fenced operation records immutable message and recipient digests and later verifies the authoritative Campaign/channel attempt state. Definitive rejection is distinct from accepted, outcome-unknown, and recovery-required work in Ops. Object-only failures prove compensation; managed Files output and uncertain cleanup remain explicit forward-recovery work. Retention commits locator changes and independently verifies artifact absence. An operator-only reconciler inventories bounded tenant-prefix pages, protects active builds and a minimum 24-hour grace period, and deletes only objects that remain unreferenced. Never copy or edit runtime object keys as business data.", layer="evidence", documentation_types=("admin",), audience=("campaign_operator", "platform_operator", "release_reviewer"), @@ -1086,9 +1086,9 @@ manifest = ModuleManifest( "kind": "reference", "route": "/campaigns/queue", "screen": "Campaign operator queue", - "verification": "Build on one replica, verify the recovery checkpoint chain and database/object manifests, deliver from another, then exercise process loss, stale fencing, storage failure, and retention cleanup.", + "verification": "Build on one replica, verify the recovery checkpoint chain and database/object manifests, deliver from another, then exercise process loss, dry-run/apply orphan reconciliation, stale fencing, storage failure, and retention cleanup.", "limitations": [ - "A hard process loss between object creation and database commit remains recovery-required under its reserved prefix until the inventory reconciler verifies or removes it.", + "Artifact inventory and cleanup are currently an operator API/runbook action rather than an Ops WebUI control.", "Database, object storage, and encryption keys require a coordinated deployment backup and restore procedure.", ], }, @@ -1178,7 +1178,10 @@ manifest = ModuleManifest( "durable address directory", "file storage", ), - recovery_docs=("docs/CAMPAIGN_DELIVERY_RUNBOOK.md",), + recovery_docs=( + "docs/CAMPAIGN_DELIVERY_RUNBOOK.md", + "docs/CAMPAIGN_BUILD_RECOVERY.md", + ), security_docs=("docs/ACCESS_EXPLANATION_COVERAGE.md",), operations_docs=("docs/CAMPAIGN_DELIVERY_RUNBOOK.md",), ), diff --git a/src/govoplan_campaign/backend/persistence/campaigns.py b/src/govoplan_campaign/backend/persistence/campaigns.py index 1412201..0c3806d 100644 --- a/src/govoplan_campaign/backend/persistence/campaigns.py +++ b/src/govoplan_campaign/backend/persistence/campaigns.py @@ -1736,6 +1736,17 @@ def build_campaign_version( print_outputs_by_index=resolved_print_outputs_by_index, ), ) + if recovery_operation is not None: + recovery_operation.checkpoint( + kind="build-storage-ready", + summary=( + "Generated Campaign objects were verified before domain commit" + ), + evidence={ + "storage_prefix": storage_prefix, + "storage": storage_manifest, + }, + ) report_json = _campaign_build_report(result, files) report_json["built_by_user_id"] = user_id if resolved_print_outputs_by_index: diff --git a/src/govoplan_campaign/backend/router.py b/src/govoplan_campaign/backend/router.py index 4603fee..4e265d8 100644 --- a/src/govoplan_campaign/backend/router.py +++ b/src/govoplan_campaign/backend/router.py @@ -6,6 +6,7 @@ from govoplan_campaign.backend.routes.attachments import router as attachments_r from govoplan_campaign.backend.routes.campaigns import router as campaigns_router from govoplan_campaign.backend.routes.delivery import router as delivery_router from govoplan_campaign.backend.routes.jobs import router as jobs_router +from govoplan_campaign.backend.routes.operations import router as operations_router from govoplan_campaign.backend.routes.reports import router as reports_router from govoplan_campaign.backend.routes.sharing import router as sharing_router from govoplan_campaign.backend.routes.versions import router as versions_router @@ -13,6 +14,7 @@ from govoplan_campaign.backend.routes.versions import router as versions_router router = APIRouter() for workflow_router in ( + operations_router, campaigns_router, versions_router, jobs_router, diff --git a/src/govoplan_campaign/backend/routes/operations.py b/src/govoplan_campaign/backend/routes/operations.py new file mode 100644 index 0000000..50c05cc --- /dev/null +++ b/src/govoplan_campaign/backend/routes/operations.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session + +from govoplan_campaign.backend.artifact_reconciliation import ( + CampaignArtifactReconciliationError, + reconcile_campaign_artifacts, +) +from govoplan_campaign.backend.persistence.campaigns import _object_storage +from govoplan_campaign.backend.schemas import ( + CampaignArtifactReconcileRequest, + CampaignArtifactReconcileResponse, +) +from govoplan_core.auth import ApiPrincipal, require_scope +from govoplan_core.audit.logging import audit_from_principal +from govoplan_core.core.object_storage import StorageBackendError +from govoplan_core.core.recovery import RecoveryGuaranteeError +from govoplan_core.core.recovery_runtime import ( + RecoveryOperationBusy, + RecoveryOperationStateConflict, +) +from govoplan_core.db.session import get_database, get_session +from govoplan_core.server.runtime_agent import application_runtime_identity + + +router = APIRouter(prefix="/campaigns/operations", tags=["campaigns"]) + + +@router.post( + "/artifacts/reconcile", + response_model=CampaignArtifactReconcileResponse, +) +def reconcile_artifacts( + payload: CampaignArtifactReconcileRequest, + request: Request, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(require_scope("system:settings:write")), +) -> CampaignArtifactReconcileResponse: + """Inventory or remove old, unreferenced Campaign-owned build objects.""" + + try: + result = reconcile_campaign_artifacts( + get_database().SessionLocal, + storage=_object_storage(), + identity=application_runtime_identity(request.app), + tenant_id=principal.tenant_id, + apply=payload.apply, + idempotency_key=payload.idempotency_key, + grace_period_hours=payload.grace_period_hours, + cursor=payload.cursor, + page_size=payload.page_size, + ) + except ValueError as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(exc), + ) from exc + except (RecoveryOperationBusy, RecoveryOperationStateConflict) as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(exc), + ) from exc + except ( + CampaignArtifactReconciliationError, + RecoveryGuaranteeError, + StorageBackendError, + SQLAlchemyError, + OSError, + ) as exc: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=( + "Campaign artifact reconciliation is temporarily unavailable " + f"({type(exc).__name__})." + ), + ) from exc + + audit_from_principal( + session, + principal, + action=( + "campaign.artifact_orphans_reconciled" + if payload.apply + else "campaign.artifact_inventory_scanned" + ), + object_type="campaign_artifact_namespace", + object_id=principal.tenant_id, + details={ + "apply": payload.apply, + "status": result["status"], + "scanned_count": result["scanned_count"], + "candidate_count": result["candidate_count"], + "deleted_count": result["deleted_count"], + "failure_count": result["failure_count"], + "manifest_sha256": result["manifest_sha256"], + "recovery_operation_id": result["recovery_operation_id"], + }, + commit=True, + ) + return CampaignArtifactReconcileResponse.model_validate(result) diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index b1a6a81..ac683b3 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -680,6 +680,55 @@ class BuildCampaignRequest(BaseModel): idempotency_key: str | None = Field(default=None, min_length=1, max_length=200) +class CampaignArtifactReconcileRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + apply: bool = False + idempotency_key: str | None = Field(default=None, min_length=1, max_length=200) + grace_period_hours: int = Field(default=24, ge=24, le=24 * 90) + cursor: str | None = Field(default=None, min_length=1, max_length=1000) + page_size: int = Field(default=250, ge=1, le=1000) + + @model_validator(mode="after") + def require_apply_idempotency_key(self) -> "CampaignArtifactReconcileRequest": + if self.apply and not self.idempotency_key: + raise ValueError("Applied artifact cleanup requires an idempotency key") + return self + + +class CampaignArtifactCandidateResponse(BaseModel): + key: str + size_bytes: int + modified_at: datetime + age_seconds: int + reason: str + disposition: str + failure_type: str | None = None + + +class CampaignArtifactReconcileResponse(BaseModel): + apply: bool + status: str + recovery_operation_id: str | None = None + tenant_prefix: str + cursor: str | None = None + next_cursor: str | None = None + scanned_count: int + scanned_bytes: int + referenced_count: int + active_build_count: int + young_count: int + unknown_age_count: int + invalid_shape_count: int + candidate_count: int + candidate_bytes: int + deleted_count: int + deleted_bytes: int + failure_count: int + manifest_sha256: str | None = None + candidates: list[CampaignArtifactCandidateResponse] = Field(default_factory=list) + + class ApiKeyCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/tests/test_artifact_reconciliation.py b/tests/test_artifact_reconciliation.py new file mode 100644 index 0000000..60fe9bb --- /dev/null +++ b/tests/test_artifact_reconciliation.py @@ -0,0 +1,337 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest +from sqlalchemy import create_engine, select +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from govoplan_campaign.backend.artifact_reconciliation import ( + campaign_artifact_inventory, + reconcile_campaign_artifacts, +) +from govoplan_core.core.object_storage import ( + StorageBackendError, + StorageObjectInfo, + StorageObjectMissing, + StorageObjectPage, +) +from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation +from govoplan_core.core.recovery_runtime import RecoveryOperationBusy +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + acquire_lease, +) + + +NOW = datetime(2026, 8, 3, 12, 0, tzinfo=timezone.utc) +PREFIX = "campaign-artifacts/tenant-1/campaign-1/version-1/" + + +class _MemoryStorage: + name = "memory" + + def __init__(self) -> None: + self.objects: dict[str, bytes] = {} + self.modified_at: dict[str, datetime | None] = {} + self.delete_failures: set[str] = set() + self.exists_failures: set[str] = set() + + def add( + self, + key: str, + *, + payload: bytes = b"artifact", + modified_at: datetime | None = NOW - timedelta(days=2), + ) -> None: + self.objects[key] = payload + self.modified_at[key] = modified_at + + def put_bytes(self, key: str, data: bytes, **_kwargs) -> None: + self.add(key, payload=data, modified_at=NOW) + + def get_bytes(self, key: str) -> bytes: + try: + return self.objects[key] + except KeyError as exc: + raise StorageObjectMissing("missing") from exc + + def iter_bytes(self, key: str, *, chunk_size: int = 1024 * 1024): + del chunk_size + yield self.get_bytes(key) + + def delete(self, key: str) -> None: + if key in self.delete_failures: + raise StorageBackendError("delete unavailable") + self.objects.pop(key, None) + self.modified_at.pop(key, None) + + def exists(self, key: str) -> bool: + if key in self.exists_failures: + raise StorageBackendError("probe unavailable") + return key in self.objects + + def stat(self, key: str) -> StorageObjectInfo: + if key not in self.objects: + raise StorageObjectMissing("missing") + return StorageObjectInfo( + key=key, + size_bytes=len(self.objects[key]), + modified_at=self.modified_at[key], + ) + + def list_objects( + self, + *, + prefix: str, + after: str | None = None, + limit: int = 500, + ) -> StorageObjectPage: + keys = [ + key + for key in sorted(self.objects) + if key.startswith(prefix) and (after is None or key > after) + ] + selected = keys[:limit] + return StorageObjectPage( + objects=tuple(self.stat(key) for key in selected), + next_cursor=(selected[-1] if len(keys) > len(selected) else None), + ) + + +@pytest.fixture +def recovery_session_factory(): + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + DistributedLease.__table__.create(engine) + RecoveryOperation.__table__.create(engine) + RecoveryCheckpoint.__table__.create(engine) + factory = sessionmaker(bind=engine, expire_on_commit=False) + try: + yield factory + finally: + engine.dispose() + + +def _identity(*, node_id: str = "node-1", incarnation: str = "run-1"): + return RuntimeIdentity( + installation_id="campaign-artifact-tests", + node_id=node_id, + incarnation=incarnation, + role="worker", + software_version="test", + composition_hash="a" * 64, + ) + + +def _without_domain_references(): + return ( + patch( + "govoplan_campaign.backend.artifact_reconciliation._referenced_artifact_keys", + return_value=set(), + ), + patch( + "govoplan_campaign.backend.artifact_reconciliation._active_build_ids", + return_value=set(), + ), + ) + + +def test_inventory_classifies_reference_grace_active_build_and_unknown_age() -> None: + storage = _MemoryStorage() + orphan = f"{PREFIX}build-orphan/message.eml" + referenced = f"{PREFIX}build-referenced/message.eml" + active = f"{PREFIX}build-active/message.eml" + young = f"{PREFIX}build-young/message.eml" + unknown_age = f"{PREFIX}build-unknown/message.eml" + malformed = "campaign-artifacts/tenant-1/not-a-build-object" + storage.add(orphan) + storage.add(referenced) + storage.add(active) + storage.add(young, modified_at=NOW - timedelta(hours=1)) + storage.add(unknown_age, modified_at=None) + storage.add(malformed) + + with ( + patch( + "govoplan_campaign.backend.artifact_reconciliation._referenced_artifact_keys", + return_value={referenced}, + ), + patch( + "govoplan_campaign.backend.artifact_reconciliation._active_build_ids", + return_value={"build-active"}, + ), + ): + inventory = campaign_artifact_inventory( + object(), # type: ignore[arg-type] + storage=storage, + tenant_id="tenant-1", + grace_period=timedelta(hours=24), + page_size=20, + now=NOW, + ) + + assert [candidate.key for candidate in inventory.candidates] == [orphan] + assert inventory.referenced_count == 1 + assert inventory.active_build_count == 1 + assert inventory.young_count == 1 + assert inventory.unknown_age_count == 1 + assert inventory.invalid_shape_count == 1 + + +def test_process_loss_orphan_is_deleted_once_and_same_request_replays( + recovery_session_factory, +) -> None: + storage = _MemoryStorage() + orphan = f"{PREFIX}lost-build/message.eml" + storage.add(orphan) + reference_patch, active_patch = _without_domain_references() + with reference_patch, active_patch: + dry_run = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + now=NOW, + ) + applied = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="cleanup-lost-build", + now=NOW, + ) + replayed = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="cleanup-lost-build", + now=NOW, + ) + repeated = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="cleanup-empty-page", + now=NOW, + ) + + assert dry_run["candidate_count"] == 1 + assert dry_run["deleted_count"] == 0 + assert applied["status"] == "applied" + assert applied["deleted_count"] == 1 + assert orphan not in storage.objects + assert replayed["status"] == "already_completed" + assert repeated["candidate_count"] == 0 + + +def test_competing_node_cannot_acquire_cleanup_authority( + recovery_session_factory, +) -> None: + with recovery_session_factory() as session: + claim = acquire_lease( + session, + installation_id="campaign-artifact-tests", + resource_key="campaign:artifact-reconcile:tenant-1", + holder_node_id="node-other", + holder_incarnation="run-other", + ttl_seconds=900, + now=NOW, + ) + assert claim is not None + session.commit() + + storage = _MemoryStorage() + reference_patch, active_patch = _without_domain_references() + with reference_patch, active_patch, pytest.raises(RecoveryOperationBusy): + reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="competing-cleanup", + now=NOW, + ) + + +def test_partial_storage_outage_remains_visible_and_retryable( + recovery_session_factory, +) -> None: + storage = _MemoryStorage() + removed = f"{PREFIX}build-a/message.eml" + retained = f"{PREFIX}build-b/message.eml" + storage.add(removed) + storage.add(retained) + storage.delete_failures.add(retained) + reference_patch, active_patch = _without_domain_references() + with reference_patch, active_patch: + partial = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="partial-cleanup", + now=NOW, + ) + + storage.delete_failures.clear() + retry = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="partial-cleanup-retry", + now=NOW, + ) + + assert partial["status"] == "recovery_required" + assert partial["deleted_count"] == 1 + assert partial["failure_count"] == 1 + assert removed not in storage.objects + assert retry["status"] == "applied" + assert retry["deleted_count"] == 1 + assert storage.objects == {} + with recovery_session_factory() as session: + states = session.execute( + select(RecoveryOperation.status).order_by(RecoveryOperation.created_at) + ).scalars().all() + assert "recovery_required" in states + assert states[-1] == "succeeded" + + +def test_unverifiable_delete_is_outcome_unknown( + recovery_session_factory, +) -> None: + storage = _MemoryStorage() + orphan = f"{PREFIX}build-unknown/message.eml" + storage.add(orphan) + storage.exists_failures.add(orphan) + reference_patch, active_patch = _without_domain_references() + with reference_patch, active_patch: + result = reconcile_campaign_artifacts( + recovery_session_factory, + storage=storage, + identity=_identity(), + tenant_id="tenant-1", + apply=True, + idempotency_key="unknown-cleanup", + now=NOW, + ) + + assert result["status"] == "outcome_unknown" + assert result["failure_count"] == 1 diff --git a/tests/test_route_registration.py b/tests/test_route_registration.py index 34b8357..02fac54 100644 --- a/tests/test_route_registration.py +++ b/tests/test_route_registration.py @@ -7,6 +7,7 @@ from govoplan_campaign.backend.routes.attachments import router as attachments_r from govoplan_campaign.backend.routes.campaigns import router as campaigns_router from govoplan_campaign.backend.routes.delivery import router as delivery_router from govoplan_campaign.backend.routes.jobs import router as jobs_router +from govoplan_campaign.backend.routes.operations import router as operations_router from govoplan_campaign.backend.routes.reports import router as reports_router from govoplan_campaign.backend.routes.sharing import router as sharing_router from govoplan_campaign.backend.routes.versions import router as versions_router @@ -22,6 +23,7 @@ def _operation_keys(candidate_router) -> list[tuple[str, str]]: def test_campaign_router_composes_every_workflow_operation_once() -> None: workflow_routers = ( + operations_router, campaigns_router, versions_router, jobs_router, @@ -38,12 +40,16 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None: actual = _operation_keys(router) assert actual == expected - assert len(actual) == 71 + assert len(actual) == 72 assert not [operation for operation, count in Counter(actual).items() if count > 1] def test_key_routes_are_owned_by_their_focused_router() -> None: expectations = ( + ( + operations_router, + ("POST", "/campaigns/operations/artifacts/reconcile"), + ), (campaigns_router, ("GET", "/campaigns/{campaign_id}/workspace")), (versions_router, ("POST", "/campaigns/versions/{version_id}/build")), (jobs_router, ("GET", "/campaigns/{campaign_id}/jobs")),