diff --git a/services/ingestion/app/partner_file_retention.py b/services/ingestion/app/partner_file_retention.py new file mode 100644 index 000000000..3a8a5f60d --- /dev/null +++ b/services/ingestion/app/partner_file_retention.py @@ -0,0 +1,318 @@ +"""30-day partner-file retention job (SEL-549 first-file). + +Deletes objects whose ``partner_files.delete_after`` has passed, writes an +append-only ``partner_file_deletion_log`` row (hash-chained), marks the +metadata row deleted, and emits an audit event. Synthetic fixtures only +until SEL-590 clears; do not invent partner payloads. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional +from uuid import uuid4 + +from sqlalchemy import text + +from shared.tenant_context import set_tenant_guc + +logger = logging.getLogger("partner_file_retention") + +ACTOR = "retention_job" +REASON = "retention_30d" +BATCH_LIMIT = 100 +_CHAIN_LOCK_KEY_PREFIX = "partner_file_deletion:" + + +def _canonical_entry_payload(fields: Dict[str, Any]) -> str: + return json.dumps(fields, sort_keys=True, separators=(",", ":"), default=str) + + +def compute_entry_hash(*, prev_entry_hash: Optional[str], fields: Dict[str, Any]) -> str: + material = _canonical_entry_payload( + {"prev_entry_hash": prev_entry_hash, **fields} + ).encode("utf-8") + return hashlib.sha256(material).hexdigest() + + +def _lock_tenant_deletion_chain(db, *, tenant_id: str) -> None: + """Serialize deletion-log tip updates for one tenant (multi-replica safe).""" + db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + "SELECT pg_advisory_xact_lock(hashtextextended(:lock_key, 0))" + ), + {"lock_key": f"{_CHAIN_LOCK_KEY_PREFIX}{tenant_id}"}, + ) + + +def _latest_entry_hash(db, *, tenant_id: str) -> Optional[str]: + """Tip of the tenant deletion chain under the advisory lock. + + Do not ``FOR UPDATE`` here: ``regengine`` has SELECT+INSERT only on the + append-only log (UPDATE is revoked). Serialization comes from + ``_lock_tenant_deletion_chain`` (pg_advisory_xact_lock) held for the + whole delete_one_due_file savepoint. + """ + row = db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + SELECT entry_hash + FROM public.partner_file_deletion_log + WHERE tenant_id = CAST(:tenant_id AS uuid) + ORDER BY deleted_at DESC, id DESC + LIMIT 1 + """ + ), + {"tenant_id": tenant_id}, + ).fetchone() + return row[0] if row else None + + +def _due_rows_for_tenant( + db, *, tenant_id: str, limit: int = BATCH_LIMIT +) -> List[Dict[str, Any]]: + """Due partner_files for one tenant. Caller must set the tenant GUC first.""" + rows = db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + SELECT id, tenant_id, engagement_id, storage_bucket, storage_path, + content_sha256, ciphertext_sha256, byte_size, report_delivered_at, + delete_after + FROM public.partner_files + WHERE tenant_id = CAST(:tenant_id AS uuid) + AND deleted_at IS NULL + AND delete_after IS NOT NULL + AND delete_after <= now() + AND status <> 'deleted' + ORDER BY delete_after ASC + LIMIT :limit + FOR UPDATE SKIP LOCKED + """ + ), + {"tenant_id": tenant_id, "limit": limit}, + ).fetchall() + out: List[Dict[str, Any]] = [] + for r in rows: + out.append( + { + "id": str(r[0]), + "tenant_id": str(r[1]), + "engagement_id": str(r[2]) if r[2] is not None else None, + "storage_bucket": r[3], + "storage_path": r[4], + "content_sha256": r[5], + "ciphertext_sha256": r[6], + "byte_size": r[7], + "report_delivered_at": r[8], + "delete_after": r[9], + } + ) + return out + + +def delete_one_due_file(db, row: Dict[str, Any]) -> Dict[str, Any]: + """Delete storage object, append deletion_log, mark metadata deleted, audit. + + Requires object storage. Order: tenant GUC → chain advisory lock → lock tip + → S3 delete (idempotent) → durable log/update/audit. Caller wraps in a + savepoint so a PG error does not abort the whole batch. + """ + from .partner_file_storage import PARTNER_UPLOADS_BUCKET + from .remediation.audit_utils import audit_event + from .s3_utils import delete_object, is_object_storage_configured + + if not is_object_storage_configured(): + raise RuntimeError( + "partner file retention requires object storage; refusing to mark deleted" + ) + + tenant_id = row["tenant_id"] + set_tenant_guc(db, tenant_id) + _lock_tenant_deletion_chain(db, tenant_id=tenant_id) + + # Re-claim under FOR UPDATE: concurrent pass may have won already. + claimed = db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + SELECT id + FROM public.partner_files + WHERE id = CAST(:id AS uuid) + AND tenant_id = CAST(:tenant_id AS uuid) + AND deleted_at IS NULL + AND status <> 'deleted' + FOR UPDATE + """ + ), + {"id": row["id"], "tenant_id": tenant_id}, + ).fetchone() + if claimed is None: + return { + "partner_file_id": row["id"], + "skipped": True, + "reason": "already_deleted_or_missing", + } + + # S3 first (missing object is success on rerun), then durable DB state. + delete_object(row["storage_bucket"] or PARTNER_UPLOADS_BUCKET, row["storage_path"]) + + prev = _latest_entry_hash(db, tenant_id=tenant_id) + deleted_at = datetime.now(timezone.utc) + log_id = str(uuid4()) + fields = { + "partner_file_id": row["id"], + "tenant_id": tenant_id, + "engagement_id": row["engagement_id"], + "storage_bucket": row["storage_bucket"], + "storage_path": row["storage_path"], + "content_sha256": row["content_sha256"], + "ciphertext_sha256": row["ciphertext_sha256"], + "byte_size": row["byte_size"], + "report_delivered_at": row["report_delivered_at"], + "deleted_at": deleted_at.isoformat(), + "actor": ACTOR, + "reason": REASON, + } + entry_hash = compute_entry_hash(prev_entry_hash=prev, fields=fields) + + db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + INSERT INTO public.partner_file_deletion_log ( + id, partner_file_id, tenant_id, engagement_id, + storage_bucket, storage_path, content_sha256, ciphertext_sha256, + byte_size, report_delivered_at, deleted_at, actor, reason, + entry_hash, prev_entry_hash + ) VALUES ( + CAST(:id AS uuid), + CAST(:partner_file_id AS uuid), + CAST(:tenant_id AS uuid), + CAST(NULLIF(:engagement_id, '') AS uuid), + :storage_bucket, + :storage_path, + :content_sha256, + :ciphertext_sha256, + :byte_size, + :report_delivered_at, + :deleted_at, + :actor, + :reason, + :entry_hash, + :prev_entry_hash + ) + """ + ), + { + "id": log_id, + "partner_file_id": row["id"], + "tenant_id": tenant_id, + "engagement_id": row["engagement_id"], + "storage_bucket": row["storage_bucket"], + "storage_path": row["storage_path"], + "content_sha256": row["content_sha256"], + "ciphertext_sha256": row["ciphertext_sha256"], + "byte_size": row["byte_size"], + "report_delivered_at": row["report_delivered_at"], + "deleted_at": deleted_at, + "actor": ACTOR, + "reason": REASON, + "entry_hash": entry_hash, + "prev_entry_hash": prev, + }, + ) + + db.execute( + # nosemgrep: avoid-sqlalchemy-text — parameterized with :param + text( + """ + UPDATE public.partner_files + SET status = 'deleted', + deleted_at = :deleted_at + WHERE id = CAST(:id AS uuid) + AND tenant_id = CAST(:tenant_id AS uuid) + AND deleted_at IS NULL + """ + ), + {"id": row["id"], "tenant_id": tenant_id, "deleted_at": deleted_at}, + ) + + audit_event( + db, + tenant_id=tenant_id, + event_type="partner_file.retention_delete", + action="partner_file.retention_delete", + severity="info", + actor_id=None, + resource_type="partner_file", + resource_id=row["id"], + metadata={ + "actor_label": ACTOR, + "reason": REASON, + "storage_bucket": row["storage_bucket"], + "storage_path": row["storage_path"], + "content_sha256": row["content_sha256"], + "delete_after": str(row["delete_after"]), + "deletion_log_id": log_id, + "entry_hash": entry_hash, + }, + ) + + return { + "partner_file_id": row["id"], + "deletion_log_id": log_id, + "entry_hash": entry_hash, + "skipped": False, + } + + +def run_retention_pass(db, *, limit: int = BATCH_LIMIT) -> Dict[str, int]: + """Process up to ``limit`` due partner files across tenants. Returns counts. + + When object storage is unconfigured the pass is a no-op: no rows are marked + deleted and no retention audits are written. + """ + from .s3_utils import is_object_storage_configured + + if not is_object_storage_configured(): + logger.warning("partner_file_retention_skipped_storage_unconfigured") + return {"due": 0, "deleted": 0, "errors": 0, "skipped_unconfigured": 1} + + deleted = 0 + errors = 0 + due_total = 0 + remaining = limit + + # Cross-tenant discovery needs ADMIN_DATABASE_URL / regengine_sysadmin; + # the ordinary SessionLocal role cannot list tenants under FORCE RLS. + from shared.maintenance_tenants import list_maintenance_tenant_ids + + for tenant_id in list_maintenance_tenant_ids(): + if remaining <= 0: + break + set_tenant_guc(db, tenant_id) + rows = _due_rows_for_tenant(db, tenant_id=tenant_id, limit=remaining) + due_total += len(rows) + for row in rows: + try: + with db.begin_nested(): + result = delete_one_due_file(db, row) + if not result.get("skipped"): + deleted += 1 + except Exception as exc: # pragma: no cover — per-row isolation + errors += 1 + logger.exception( + "partner_file_retention_row_failed file=%s tenant=%s error=%s", + row.get("id"), + row.get("tenant_id"), + exc, + ) + remaining -= 1 + + return {"due": due_total, "deleted": deleted, "errors": errors, "skipped_unconfigured": 0} diff --git a/services/ingestion/app/s3_utils.py b/services/ingestion/app/s3_utils.py index 771f30689..330276c91 100644 --- a/services/ingestion/app/s3_utils.py +++ b/services/ingestion/app/s3_utils.py @@ -401,3 +401,22 @@ def list_raw_documents( status_code=500, detail="Failed to list raw documents from S3", ) from exc + + +def delete_object(bucket: str, key: str) -> None: + """Delete one object from the bucket. Missing keys are treated as success.""" + s3 = _client() + _validate_s3_key(key) + try: + s3.delete_object(Bucket=bucket, Key=key) + logger.info("s3_object_deleted", bucket=bucket, key=key) + except ClientError as exc: + error_code = exc.response["Error"]["Code"] + if error_code in ("NoSuchKey", "404", "NotFound"): + logger.info("s3_object_already_absent", bucket=bucket, key=key) + return + logger.error("s3_delete_failed", bucket=bucket, key=key, error=str(exc)) + raise HTTPException(status_code=500, detail="Failed to delete object from S3") from exc + except BotoCoreError as exc: + logger.error("s3_delete_failed", bucket=bucket, key=key, error=str(exc)) + raise HTTPException(status_code=500, detail="Failed to delete object from S3") from exc diff --git a/services/ingestion/tests/test_partner_file_retention_job.py b/services/ingestion/tests/test_partner_file_retention_job.py new file mode 100644 index 000000000..ebbda0dc7 --- /dev/null +++ b/services/ingestion/tests/test_partner_file_retention_job.py @@ -0,0 +1,315 @@ +"""Unit tests for the 30-day partner-file retention job (synthetic only).""" + +from __future__ import annotations + +from contextlib import nullcontext +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional +from uuid import uuid4 + +import pytest + +from services.ingestion.app import partner_file_retention as retention + + +class _FakeDB: + def __init__( + self, + *, + due_rows: Optional[List[tuple]] = None, + prev_hash: Optional[str] = None, + tenant_ids: Optional[List[str]] = None, + claim_row: bool = True, + ): + self.due_rows = due_rows or [] + self.prev_hash = prev_hash + self.tenant_ids = tenant_ids or [] + self.claim_row = claim_row + self.statements: List[str] = [] + self.params: List[Dict[str, Any]] = [] + self.nested = 0 + + def execute(self, statement, params=None): + sql = str(statement) + self.statements.append(sql) + self.params.append(params or {}) + + class _R: + def __init__(self, outer, sql): + self._outer = outer + self._sql = sql + + def fetchall(self): + if "FROM public.tenants" in self._sql: + return [(tid,) for tid in self._outer.tenant_ids] + if "FROM public.partner_files" in self._sql and "delete_after" in self._sql: + return self._outer.due_rows + return [] + + def fetchone(self): + if "pg_advisory_xact_lock" in self._sql: + return (True,) + if ( + "FROM public.partner_files" in self._sql + and "FOR UPDATE" in self._sql + and "delete_after" not in self._sql + ): + return (self._outer.due_rows[0][0],) if self._outer.claim_row and self._outer.due_rows else None + if "entry_hash" in self._sql and "partner_file_deletion_log" in self._sql: + return (self._outer.prev_hash,) if self._outer.prev_hash else None + return None + + return _R(self, sql) + + def begin_nested(self): + self.nested += 1 + return nullcontext() + + +def _synth_due_tuple(tenant_id: str, file_id: str, engagement_id: str) -> tuple: + return ( + file_id, + tenant_id, + engagement_id, + "partner-uploads", + f"tenants/{tenant_id}/engagements/{engagement_id}/obj-synth.csv", + "a" * 64, + None, + 12, + datetime(2026, 10, 1, tzinfo=timezone.utc), + datetime(2026, 10, 1, tzinfo=timezone.utc) + timedelta(days=30), + ) + + +def test_compute_entry_hash_stable() -> None: + fields = { + "partner_file_id": "11111111-1111-1111-1111-111111111111", + "tenant_id": "22222222-2222-2222-2222-222222222222", + "reason": "retention_30d", + } + h1 = retention.compute_entry_hash(prev_entry_hash=None, fields=fields) + h2 = retention.compute_entry_hash(prev_entry_hash=None, fields=fields) + assert h1 == h2 + assert len(h1) == 64 + h3 = retention.compute_entry_hash(prev_entry_hash=h1, fields=fields) + assert h3 != h1 + + +def test_delete_one_due_file_writes_log_and_audit(monkeypatch) -> None: + tenant_id = str(uuid4()) + file_id = str(uuid4()) + engagement_id = str(uuid4()) + deletes: List[tuple] = [] + audits: List[Dict[str, Any]] = [] + gucs: List[str] = [] + + import services.ingestion.app.s3_utils as s3_utils + import services.ingestion.app.remediation.audit_utils as audit_utils + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: True) + monkeypatch.setattr( + s3_utils, + "delete_object", + lambda bucket, key: deletes.append((bucket, key)), + ) + monkeypatch.setattr(audit_utils, "audit_event", lambda db, **kw: audits.append(kw)) + monkeypatch.setattr(retention, "set_tenant_guc", lambda db, tid: gucs.append(tid)) + + row = { + "id": file_id, + "tenant_id": tenant_id, + "engagement_id": engagement_id, + "storage_bucket": "partner-uploads", + "storage_path": f"tenants/{tenant_id}/engagements/{engagement_id}/obj-synth.csv", + "content_sha256": "a" * 64, + "ciphertext_sha256": None, + "byte_size": 12, + "report_delivered_at": datetime(2026, 10, 1, tzinfo=timezone.utc), + "delete_after": datetime(2026, 10, 1, tzinfo=timezone.utc) + timedelta(days=30), + } + db = _FakeDB(prev_hash=None, due_rows=[_synth_due_tuple(tenant_id, file_id, engagement_id)]) + result = retention.delete_one_due_file(db, row) + + assert deletes == [("partner-uploads", row["storage_path"])] + assert any("pg_advisory_xact_lock" in s for s in db.statements) + tip_selects = [ + s + for s in db.statements + if "partner_file_deletion_log" in s and "entry_hash" in s and "INSERT" not in s + ] + assert tip_selects + assert all("FOR UPDATE" not in s for s in tip_selects) + assert any("INSERT INTO public.partner_file_deletion_log" in s for s in db.statements) + assert any("SET status = 'deleted'" in s for s in db.statements) + assert audits and audits[0]["event_type"] == "partner_file.retention_delete" + assert audits[0]["metadata"]["reason"] == "retention_30d" + assert result["entry_hash"] + assert result["skipped"] is False + assert gucs == [tenant_id] + log_params = next( + p + for p, s in zip(db.params, db.statements) + if "INSERT INTO public.partner_file_deletion_log" in s + ) + assert log_params["actor"] == "retention_job" + assert log_params["reason"] == "retention_30d" + assert log_params["prev_entry_hash"] is None + # S3 delete precedes durable log insert + del_idx = next(i for i, s in enumerate(db.statements) if "pg_advisory_xact_lock" in s) + # delete_object is not SQL; ensure log insert happens after claim + lock + log_idx = next(i for i, s in enumerate(db.statements) if "INSERT INTO public.partner_file_deletion_log" in s) + assert del_idx < log_idx + + +def test_delete_one_refuses_when_storage_unconfigured(monkeypatch) -> None: + import services.ingestion.app.s3_utils as s3_utils + + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + db = _FakeDB() + row = { + "id": str(uuid4()), + "tenant_id": str(uuid4()), + "engagement_id": str(uuid4()), + "storage_bucket": "partner-uploads", + "storage_path": "tenants/t/engagements/e/x.csv", + "content_sha256": "b" * 64, + "ciphertext_sha256": None, + "byte_size": 1, + "report_delivered_at": None, + "delete_after": datetime.now(timezone.utc) - timedelta(days=1), + } + with pytest.raises(RuntimeError, match="object storage"): + retention.delete_one_due_file(db, row) + assert not any("partner_file_deletion_log" in s for s in db.statements) + assert not any("SET status = 'deleted'" in s for s in db.statements) + + +def test_run_retention_pass_noop_when_storage_unconfigured(monkeypatch) -> None: + import services.ingestion.app.s3_utils as s3_utils + + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: False) + called = {"due": False} + + def _boom(*a, **k): + called["due"] = True + raise AssertionError("must not scan when unconfigured") + + import shared.maintenance_tenants as maintenance_tenants + + monkeypatch.setattr(maintenance_tenants, "list_maintenance_tenant_ids", _boom) + totals = retention.run_retention_pass(object()) + assert totals == {"due": 0, "deleted": 0, "errors": 0, "skipped_unconfigured": 1} + assert called["due"] is False + + +def test_run_retention_pass_sets_guc_and_savepoint(monkeypatch) -> None: + tenant_id = str(uuid4()) + file_id = str(uuid4()) + engagement_id = str(uuid4()) + gucs: List[str] = [] + + import services.ingestion.app.s3_utils as s3_utils + import shared.maintenance_tenants as maintenance_tenants + + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: True) + monkeypatch.setattr(retention, "set_tenant_guc", lambda db, tid: gucs.append(tid)) + monkeypatch.setattr(maintenance_tenants, "list_maintenance_tenant_ids", lambda: [tenant_id]) + monkeypatch.setattr( + s3_utils, + "delete_object", + lambda bucket, key: None, + ) + monkeypatch.setattr( + "services.ingestion.app.remediation.audit_utils.audit_event", + lambda db, **kw: None, + ) + + db = _FakeDB( + tenant_ids=[tenant_id], + due_rows=[_synth_due_tuple(tenant_id, file_id, engagement_id)], + prev_hash=None, + ) + totals = retention.run_retention_pass(db) + assert totals["due"] == 1 + assert totals["deleted"] == 1 + assert totals["errors"] == 0 + assert totals["skipped_unconfigured"] == 0 + assert db.nested >= 1 + assert tenant_id in gucs + assert any("FOR UPDATE SKIP LOCKED" in s for s in db.statements) + + +def test_concurrent_chain_uses_advisory_lock_and_distinct_prev(monkeypatch) -> None: + """Two sequential deletes under the lock must chain prev hashes (no fork).""" + tenant_id = str(uuid4()) + engagement_id = str(uuid4()) + file_a = str(uuid4()) + file_b = str(uuid4()) + audits: List[Dict[str, Any]] = [] + tip: Dict[str, Optional[str]] = {"hash": None} + + import services.ingestion.app.s3_utils as s3_utils + import services.ingestion.app.remediation.audit_utils as audit_utils + monkeypatch.setattr(s3_utils, "is_object_storage_configured", lambda: True) + monkeypatch.setattr(s3_utils, "delete_object", lambda bucket, key: None) + monkeypatch.setattr(audit_utils, "audit_event", lambda db, **kw: audits.append(kw)) + monkeypatch.setattr(retention, "set_tenant_guc", lambda db, tid: None) + + class _ChainDB(_FakeDB): + def execute(self, statement, params=None): + sql = str(statement) + self.statements.append(sql) + self.params.append(params or {}) + if "INSERT INTO public.partner_file_deletion_log" in sql: + tip["hash"] = (params or {}).get("entry_hash") + return super().execute(statement, params) + + # Override fetchone path via prev_hash mutation between calls + def __init__(self, **kw): + super().__init__(**kw) + + # First delete + db1 = _FakeDB( + prev_hash=None, + due_rows=[_synth_due_tuple(tenant_id, file_a, engagement_id)], + ) + # Patch _latest_entry_hash to read tip so second call sees first hash + real_latest = retention._latest_entry_hash + + def _latest(db, *, tenant_id): + # Prefer live tip if set; else fall through to DB prev_hash + if tip["hash"] is not None and db is not db1: + return tip["hash"] + return real_latest(db, tenant_id=tenant_id) + + monkeypatch.setattr(retention, "_latest_entry_hash", _latest) + + row_a = { + "id": file_a, + "tenant_id": tenant_id, + "engagement_id": engagement_id, + "storage_bucket": "partner-uploads", + "storage_path": f"tenants/{tenant_id}/engagements/{engagement_id}/a.csv", + "content_sha256": "a" * 64, + "ciphertext_sha256": None, + "byte_size": 1, + "report_delivered_at": datetime(2026, 10, 1, tzinfo=timezone.utc), + "delete_after": datetime(2026, 10, 31, tzinfo=timezone.utc), + } + r1 = retention.delete_one_due_file(db1, row_a) + tip["hash"] = r1["entry_hash"] + assert any("pg_advisory_xact_lock" in s for s in db1.statements) + + db2 = _FakeDB( + prev_hash=tip["hash"], + due_rows=[_synth_due_tuple(tenant_id, file_b, engagement_id)], + ) + row_b = dict(row_a, id=file_b, storage_path=row_a["storage_path"].replace("a.csv", "b.csv")) + r2 = retention.delete_one_due_file(db2, row_b) + assert r2["entry_hash"] != r1["entry_hash"] + log_b = next( + p + for p, s in zip(db2.params, db2.statements) + if "INSERT INTO public.partner_file_deletion_log" in s + ) + assert log_b["prev_entry_hash"] == r1["entry_hash"] + assert any("pg_advisory_xact_lock" in s for s in db2.statements) diff --git a/services/scheduler/app/config.py b/services/scheduler/app/config.py index 2dc6ead38..4771415c5 100644 --- a/services/scheduler/app/config.py +++ b/services/scheduler/app/config.py @@ -112,6 +112,10 @@ class SchedulerSettings(BaseServiceSettings): default=30, description="Interval (seconds) between remediation notification-outbox drain passes", ) + partner_file_retention_interval_seconds: int = Field( + default=3600, + description="Interval (seconds) between partner-file 30-day retention passes", + ) # WS-8 (#2709) — dead-lettered rule evaluations are replayed on this # cadence. 15 min: the rows are already durable, so latency only # affects how quickly the audit trail self-heals. diff --git a/services/scheduler/service.py b/services/scheduler/service.py index f142e505d..d263d081f 100644 --- a/services/scheduler/service.py +++ b/services/scheduler/service.py @@ -870,6 +870,24 @@ def schedule_jobs(self) -> None: interval_seconds=self.settings.remediation_outbox_drain_interval_seconds, ) + # Partner-file 30-day retention (SEL-549 first-file): delete objects + # whose delete_after has passed, append partner_file_deletion_log, + # and audit. Hourly by default. + self._add_tracked_job( + self.run_partner_file_retention, + job_id="partner_file_retention", + trigger=IntervalTrigger( + seconds=self.settings.partner_file_retention_interval_seconds, + ), + name="Partner File 30-Day Retention", + max_instances=1, + ) + logger.info( + "job_scheduled", + job_id="partner_file_retention", + interval_seconds=self.settings.partner_file_retention_interval_seconds, + ) + # WS-8 (#2709): replay dead-lettered rule evaluations into # fsma.rule_evaluations so the audit trail self-heals. self._add_tracked_job( @@ -920,6 +938,33 @@ def drain_remediation_notifications(self) -> None: if totals["claimed"]: logger.info("remediation_notification_drain_tick", **totals) + def run_partner_file_retention(self) -> None: + """Delete due partner uploads (delete_after passed) and audit. + + Discovers tenants via ``list_maintenance_tenant_ids`` (sysadmin), then + sets the tenant GUC per tenant / row so RLS on partner_files and + deletion_log holds. + """ + try: + from shared.database import SessionLocal # noqa: PLC0415 + + from services.ingestion.app.partner_file_retention import ( # noqa: PLC0415 + run_retention_pass, + ) + + db = SessionLocal() + try: + totals = run_retention_pass(db) + db.commit() + finally: + db.close() + except Exception as exc: + logger.exception("partner_file_retention_failed", error=str(exc)) + raise + if totals.get("due"): + logger.info("partner_file_retention_tick", **totals) + + def replay_rule_eval_deadletter(self) -> None: """Replay dead-lettered rule evaluations once (WS-8, #2709).