diff --git a/apps/data-processing/METADATA_DRIFT_DETECTOR.md b/apps/data-processing/METADATA_DRIFT_DETECTOR.md new file mode 100644 index 00000000..f69ca097 --- /dev/null +++ b/apps/data-processing/METADATA_DRIFT_DETECTOR.md @@ -0,0 +1,85 @@ +# Metadata Drift Detector (#882) + +## Goal +Detect when off-chain backend metadata (`ProjectView`, `ProjectMilestone`) +falls out of sync with on-chain identifiers or status. + +## How it works +`ContractEvent` rows are the raw, immutable record of events ingested from +the Soroban RPC event stream — the closest thing this service has to +ground-truth on-chain state. `ProjectView` and `ProjectMilestone` are +mutable "materialized view" tables that get incrementally updated as events +arrive (see `PostgresService.save_project_view` / `save_project_milestone`). + +`src/metadata_drift_detector.py` recomputes canonical project/milestone +state directly from the `ContractEvent` log (independent of however the +materialized views were last updated) and diffs it against what's currently +persisted. Any mismatch indicates drift — e.g. a missed event, a bug in +incremental update logic, or a partial write. + +Compared fields: +- **Project**: `status`, `total_contributions`, `unique_contributors`, + `last_event_ledger` (staleness only), and whether a `ProjectView` row + exists at all. +- **Milestone**: `status`, and whether a `ProjectMilestone` row exists for + every milestone observed in the event log. + +## Acceptance criteria mapping +- **Compares selected backend records against chain-derived state** — + `MetadataDriftDetector.run_for_project` / `run_all`. +- **Produces actionable drift reports** — findings are returned as a + `DriftReport` and persisted to the `metadata_drift_findings` table (with + `reviewed` / `reviewed_by` / `review_notes` fields, consistent with the + existing `RoundAnomalySignal` review workflow). +- **Supports scheduled and manual execution** — registered as a 6-hourly + APScheduler job (`metadata_drift_detection` in `src/scheduler.py`) and + exposed as a CLI script (`scripts/detect_metadata_drift.py`). +- **Does not mutate source data automatically** — the detector only reads + `ContractEvent`, `ProjectView`, and `ProjectMilestone`; it never writes to + them. Findings are written exclusively to the separate + `metadata_drift_findings` table. See + `tests/test_metadata_drift_detector.py::test_does_not_mutate_source_records`. + +## Usage + +```bash +# Check every project that has at least one ingested contract event +python scripts/detect_metadata_drift.py + +# Check a single project +python scripts/detect_metadata_drift.py --project-id 101 + +# Preview findings without persisting them +python scripts/detect_metadata_drift.py --no-persist --json +``` + +Programmatically: + +```python +from src.metadata_drift_detector import MetadataDriftDetector + +detector = MetadataDriftDetector() +report = detector.run_and_persist() # all known projects +print(report.to_dict()) +``` + +## Reviewing findings + +```python +from src.db.postgres_service import PostgresService + +service = PostgresService() +findings = service.get_metadata_drift_findings(reviewed=False, severity="critical") +service.mark_metadata_drift_finding_reviewed(findings[0].id, reviewed_by="maintainer") +``` + +## Limitations / follow-ups +- Drift is derived from the locally ingested `ContractEvent` log, not a live + Soroban RPC read. If event ingestion itself has gaps, this detector won't + catch them (that's covered by the separate ingestion quality checks in + `src/ingestion/stellar_ingestion_checks.py`). A future iteration could add + an optional live-RPC spot-check mode for higher assurance. +- `last_event_ledger` drift is only flagged when the backend is *behind* the + chain-derived ledger (staleness), not when it's ahead, since a backend + legitimately may have ingested events not yet reflected in a partial + ContractEvent backfill window. diff --git a/apps/data-processing/scripts/detect_metadata_drift.py b/apps/data-processing/scripts/detect_metadata_drift.py new file mode 100644 index 00000000..002bb8c4 --- /dev/null +++ b/apps/data-processing/scripts/detect_metadata_drift.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +""" +Metadata Drift Detector — manual execution entrypoint (#882) + +Compares backend ProjectView/ProjectMilestone records against state +recomputed from the raw ContractEvent log and reports any drift. +Read-only: never mutates ProjectView/ProjectMilestone. + +Usage: + python scripts/detect_metadata_drift.py + python scripts/detect_metadata_drift.py --project-id 101 + python scripts/detect_metadata_drift.py --limit 1000 --no-persist +""" + +import sys +import os +import json +import argparse +import logging + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.metadata_drift_detector import MetadataDriftDetector + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Detect drift between backend records and on-chain-derived state." + ) + parser.add_argument( + "--project-id", + type=int, + default=None, + help="Check a single project instead of all known projects.", + ) + parser.add_argument( + "--limit", + type=int, + default=500, + help="Max number of projects to check when --project-id is not given.", + ) + parser.add_argument( + "--no-persist", + action="store_true", + help="Print findings without writing them to metadata_drift_findings.", + ) + parser.add_argument( + "--json", + action="store_true", + help="Print the report as JSON instead of a human-readable summary.", + ) + args = parser.parse_args() + + detector = MetadataDriftDetector() + + if args.no_persist: + report = ( + detector.run_for_project(args.project_id) + if args.project_id is not None + else detector.run_all(limit=args.limit) + ) + else: + report = detector.run_and_persist(project_id=args.project_id, limit=args.limit) + + if args.json: + print(json.dumps(report.to_dict(), indent=2)) + else: + logger.info("=" * 60) + logger.info("Metadata Drift Detection Report") + logger.info("=" * 60) + logger.info(f"Run ID: {report.run_id}") + logger.info(f"Projects checked: {report.projects_checked}") + logger.info(f"Projects with drift: {report.projects_with_drift}") + logger.info(f"Total findings: {len(report.findings)}") + for finding in report.findings: + logger.info( + " [%s] project=%s %s%s: backend=%r chain_derived=%r", + finding.severity.upper(), + finding.project_id, + finding.scope, + f" milestone={finding.milestone_id}" if finding.milestone_id is not None else "", + finding.backend_value, + finding.chain_derived_value, + ) + + return 1 if report.drift_detected else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/apps/data-processing/src/db/models.py b/apps/data-processing/src/db/models.py index be557f7b..b0bb6da7 100644 --- a/apps/data-processing/src/db/models.py +++ b/apps/data-processing/src/db/models.py @@ -487,6 +487,56 @@ def __repr__(self): return f"" +class MetadataDriftFinding(Base): + """ + Stores individual drift findings produced by the metadata drift detector + (backend ProjectView/ProjectMilestone records vs. chain-derived state + recomputed from the immutable ContractEvent log). + """ + + __tablename__ = "metadata_drift_findings" + + id = Column(Integer, primary_key=True, autoincrement=True) + run_id = Column(String(64), nullable=False, index=True) + project_id = Column(BigInteger, nullable=False, index=True) + scope = Column(String(20), nullable=False, index=True) # "project" or "milestone" + milestone_id = Column(Integer, nullable=True, index=True) + field = Column(String(100), nullable=False, index=True) + backend_value = Column(Text, nullable=True) + chain_derived_value = Column(Text, nullable=True) + severity = Column(String(20), nullable=False, default="warning", index=True) + detected_at = Column(DateTime(timezone=True), nullable=False, index=True) + created_at = Column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + # Review status (consistent with RoundAnomalySignal review workflow) + reviewed = Column(Boolean, nullable=False, default=False) + review_notes = Column(Text, nullable=True) + reviewed_at = Column(DateTime(timezone=True), nullable=True) + reviewed_by = Column(String(255), nullable=True) + + __table_args__ = ( + Index("idx_metadata_drift_findings_run_id", "run_id"), + Index("idx_metadata_drift_findings_project_id", "project_id"), + Index("idx_metadata_drift_findings_scope", "scope"), + Index("idx_metadata_drift_findings_field", "field"), + Index("idx_metadata_drift_findings_severity", "severity"), + Index("idx_metadata_drift_findings_reviewed", "reviewed"), + Index( + "idx_metadata_drift_findings_project_field", + "project_id", + "field", + ), + ) + + def __repr__(self): + return ( + f"" + ) + + class RoundAnomalySignal(Base): """ Stores anomaly signals detected in quadratic funding rounds for maintainer review. diff --git a/apps/data-processing/src/db/postgres_service.py b/apps/data-processing/src/db/postgres_service.py index 304a563c..ef0017ad 100644 --- a/apps/data-processing/src/db/postgres_service.py +++ b/apps/data-processing/src/db/postgres_service.py @@ -29,6 +29,7 @@ NewsInsight, AssetTrend, RoundAnomalySignal, + MetadataDriftFinding, ) from src.analytics.ner_service import NERService from src.analytics.onchain_entity_linker import ( @@ -2067,3 +2068,127 @@ def cleanup_old_data(self, days: int = 30) -> Dict[str, int]: "news_insights": 0, "asset_trends": 0, } + + # Metadata Drift Finding Methods (#882) + + def save_metadata_drift_finding( + self, finding_data: Dict[str, Any] + ) -> Optional[MetadataDriftFinding]: + """ + Persist a single metadata drift finding produced by the drift detector. + + Args: + finding_data: Dictionary matching MetadataDriftFinding fields + (run_id, project_id, scope, milestone_id, field, + backend_value, chain_derived_value, severity, detected_at) + + Returns: + MetadataDriftFinding object if successful, None otherwise + """ + def _save(): + with self.get_session() as session: + finding = MetadataDriftFinding( + run_id=finding_data["run_id"], + project_id=finding_data["project_id"], + scope=finding_data["scope"], + milestone_id=finding_data.get("milestone_id"), + field=finding_data["field"], + backend_value=finding_data.get("backend_value"), + chain_derived_value=finding_data.get("chain_derived_value"), + severity=finding_data.get("severity", "warning"), + detected_at=finding_data.get("detected_at", datetime.utcnow()), + ) + session.add(finding) + session.flush() + session.refresh(finding) + return finding + + try: + return self._retry_operation(_save) + except SQLAlchemyError as e: + logger.error(f"Failed to save metadata drift finding: {e}") + return None + + def save_metadata_drift_findings( + self, findings: List[Dict[str, Any]] + ) -> int: + """ + Persist multiple metadata drift findings in a batch. + + Args: + findings: List of finding data dictionaries + + Returns: + Number of findings saved successfully + """ + saved_count = 0 + for finding_data in findings: + if self.save_metadata_drift_finding(finding_data): + saved_count += 1 + logger.info(f"Saved {saved_count}/{len(findings)} metadata drift findings") + return saved_count + + def get_metadata_drift_findings( + self, + run_id: Optional[str] = None, + project_id: Optional[int] = None, + scope: Optional[str] = None, + severity: Optional[str] = None, + reviewed: Optional[bool] = None, + limit: int = 200, + ) -> List[MetadataDriftFinding]: + """ + Retrieve metadata drift findings with optional filters. + """ + try: + with self.get_session() as session: + query = select(MetadataDriftFinding) + + if run_id is not None: + query = query.where(MetadataDriftFinding.run_id == run_id) + if project_id is not None: + query = query.where(MetadataDriftFinding.project_id == project_id) + if scope is not None: + query = query.where(MetadataDriftFinding.scope == scope) + if severity is not None: + query = query.where(MetadataDriftFinding.severity == severity) + if reviewed is not None: + query = query.where(MetadataDriftFinding.reviewed == reviewed) + + query = query.order_by(desc(MetadataDriftFinding.detected_at)).limit(limit) + return session.execute(query).scalars().all() + except SQLAlchemyError as e: + logger.error(f"Failed to get metadata drift findings: {e}") + return [] + + def mark_metadata_drift_finding_reviewed( + self, + finding_id: int, + reviewed_by: str, + review_notes: Optional[str] = None, + ) -> bool: + """ + Mark a metadata drift finding as reviewed. + """ + try: + with self.get_session() as session: + finding = session.execute( + select(MetadataDriftFinding).where(MetadataDriftFinding.id == finding_id) + ).scalar_one_or_none() + + if not finding: + logger.warning(f"Metadata drift finding {finding_id} not found") + return False + + finding.reviewed = True + finding.reviewed_at = datetime.utcnow() + finding.reviewed_by = reviewed_by + finding.review_notes = review_notes + + logger.info( + f"Marked metadata drift finding {finding_id} as reviewed by {reviewed_by}" + ) + return True + except SQLAlchemyError as e: + logger.error(f"Failed to mark metadata drift finding as reviewed: {e}") + return False diff --git a/apps/data-processing/src/metadata_drift_detector.py b/apps/data-processing/src/metadata_drift_detector.py new file mode 100644 index 00000000..8d93dead --- /dev/null +++ b/apps/data-processing/src/metadata_drift_detector.py @@ -0,0 +1,430 @@ +""" +Metadata Drift Detector (#882) + +Detects when off-chain backend metadata (ProjectView / ProjectMilestone +materialized views) falls out of sync with on-chain identifiers or status. + +How it works +------------ +ContractEvent rows are the raw, immutable record of events ingested from the +Soroban RPC event stream — they are the closest thing this service has to +ground-truth on-chain state. ProjectView and ProjectMilestone are mutable +"materialized view" tables that get incrementally updated as events arrive. + +This detector recomputes canonical project/milestone state directly from the +ContractEvent log (independent of however ProjectView/ProjectMilestone were +last updated) and diffs it against what's currently persisted. Any mismatch +indicates the materialized view has drifted from on-chain reality — e.g. due +to a missed event, a bug in incremental update logic, or a partial write. + +This module is strictly read-only with respect to ProjectView and +ProjectMilestone: it never mutates source data automatically. Findings are +written to a separate `metadata_drift_findings` table for maintainer review. +""" + +from __future__ import annotations + +import math +import uuid +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from sqlalchemy import select + +from src.utils.logger import setup_logger +from src.db.postgres_service import PostgresService +from src.db.models import ContractEvent, ProjectView, ProjectMilestone + +logger = setup_logger(__name__) + +# Event types that increase a project's recorded contribution total. +_POSITIVE_CONTRIBUTION_EVENT_TYPES = { + "depositevent", + "contributionrecordedevent", +} + +# Event types that decrease a project's recorded contribution total. +_NEGATIVE_CONTRIBUTION_EVENT_TYPES = { + "contributionrefundableevent", + "contributionclawbackedevent", +} + +# Numeric drift below this tolerance is treated as floating point noise, +# not a real discrepancy. +_AMOUNT_TOLERANCE = 1e-6 + +# Default severity per field — status/missing-record mismatches are more +# actionable than a momentum score that's merely stale. +_FIELD_SEVERITY = { + "status": "critical", + "total_contributions": "warning", + "unique_contributors": "warning", + "last_event_ledger": "info", + "missing_project_view": "critical", + "missing_milestone": "critical", +} + + +@dataclass +class DriftFinding: + """A single field-level mismatch between backend and chain-derived state.""" + + project_id: int + scope: str # "project" or "milestone" + field: str + backend_value: Any + chain_derived_value: Any + milestone_id: Optional[int] = None + severity: str = "warning" + detected_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + def to_dict(self) -> Dict[str, Any]: + return { + "project_id": self.project_id, + "scope": self.scope, + "milestone_id": self.milestone_id, + "field": self.field, + "backend_value": self.backend_value, + "chain_derived_value": self.chain_derived_value, + "severity": self.severity, + "detected_at": self.detected_at.isoformat(), + } + + +@dataclass +class DriftReport: + """Aggregate result of a drift detection run across one or more projects.""" + + run_id: str + started_at: datetime + completed_at: Optional[datetime] = None + projects_checked: int = 0 + findings: List[DriftFinding] = field(default_factory=list) + + @property + def drift_detected(self) -> bool: + return len(self.findings) > 0 + + @property + def projects_with_drift(self) -> int: + return len({f.project_id for f in self.findings}) + + def to_dict(self) -> Dict[str, Any]: + return { + "run_id": self.run_id, + "started_at": self.started_at.isoformat(), + "completed_at": self.completed_at.isoformat() if self.completed_at else None, + "projects_checked": self.projects_checked, + "projects_with_drift": self.projects_with_drift, + "findings_count": len(self.findings), + "findings": [f.to_dict() for f in self.findings], + } + + +@dataclass +class _ChainDerivedProjectState: + project_id: int + status: Optional[str] + total_contributions: float + unique_contributors: int + last_event_ledger: Optional[int] + milestones: Dict[int, str] # milestone_id -> latest status + + +class MetadataDriftDetector: + """ + Compares backend-persisted project/milestone records against state + recomputed independently from the raw ContractEvent log. + """ + + def __init__(self, db_service: Optional[PostgresService] = None): + self.db_service = db_service or PostgresService() + + # -- Chain-derived state recomputation ------------------------------- + + def _compute_chain_derived_state( + self, project_id: int + ) -> Optional[_ChainDerivedProjectState]: + """ + Recompute canonical project state purely from the ContractEvent log, + ordered by ledger so later events win on conflicting fields. + """ + with self.db_service.get_session() as session: + events = ( + session.execute( + select(ContractEvent) + .where(ContractEvent.project_id == project_id) + .order_by(ContractEvent.ledger.asc()) + ) + .scalars() + .all() + ) + + if not events: + return None + + total_contributions = 0.0 + contributors: set = set() + status: Optional[str] = None + last_event_ledger: Optional[int] = None + milestones: Dict[int, str] = {} + + for event in events: + event_type = (event.event_type or "").lower() + + if event.amount is not None: + amount = float(event.amount) + if event_type in _POSITIVE_CONTRIBUTION_EVENT_TYPES: + total_contributions += amount + elif event_type in _NEGATIVE_CONTRIBUTION_EVENT_TYPES: + total_contributions -= amount + + if event.contributor: + contributors.add(event.contributor) + + if event.status: + if event.milestone_id is not None: + milestones[event.milestone_id] = event.status + else: + status = event.status + + if event.ledger is not None: + last_event_ledger = event.ledger + + return _ChainDerivedProjectState( + project_id=project_id, + status=status, + total_contributions=round(total_contributions, 6), + unique_contributors=len(contributors), + last_event_ledger=last_event_ledger, + milestones=milestones, + ) + + def _get_known_project_ids(self, limit: int) -> List[int]: + """Project IDs that have at least one ingested contract event.""" + with self.db_service.get_session() as session: + rows = session.execute( + select(ContractEvent.project_id) + .where(ContractEvent.project_id.isnot(None)) + .distinct() + .limit(limit) + ).all() + return [row[0] for row in rows] + + # -- Comparison -------------------------------------------------------- + + def _compare_project( + self, chain_state: _ChainDerivedProjectState + ) -> List[DriftFinding]: + findings: List[DriftFinding] = [] + project_id = chain_state.project_id + + with self.db_service.get_session() as session: + backend_view = session.execute( + select(ProjectView).where(ProjectView.project_id == project_id) + ).scalar_one_or_none() + + milestone_rows = session.execute( + select(ProjectMilestone).where( + ProjectMilestone.project_id == project_id + ) + ).scalars().all() + backend_milestones = {m.milestone_id: m.status for m in milestone_rows} + + if backend_view is None: + findings.append( + DriftFinding( + project_id=project_id, + scope="project", + field="missing_project_view", + backend_value=None, + chain_derived_value="exists", + severity=_FIELD_SEVERITY["missing_project_view"], + ) + ) + else: + if chain_state.status is not None and backend_view.status != chain_state.status: + findings.append( + DriftFinding( + project_id=project_id, + scope="project", + field="status", + backend_value=backend_view.status, + chain_derived_value=chain_state.status, + severity=_FIELD_SEVERITY["status"], + ) + ) + + backend_total = float(backend_view.total_contributions or 0.0) + if not math.isclose( + backend_total, + chain_state.total_contributions, + abs_tol=_AMOUNT_TOLERANCE, + ): + findings.append( + DriftFinding( + project_id=project_id, + scope="project", + field="total_contributions", + backend_value=backend_total, + chain_derived_value=chain_state.total_contributions, + severity=_FIELD_SEVERITY["total_contributions"], + ) + ) + + backend_contributors = int(backend_view.unique_contributors or 0) + if backend_contributors != chain_state.unique_contributors: + findings.append( + DriftFinding( + project_id=project_id, + scope="project", + field="unique_contributors", + backend_value=backend_contributors, + chain_derived_value=chain_state.unique_contributors, + severity=_FIELD_SEVERITY["unique_contributors"], + ) + ) + + backend_ledger = backend_view.last_event_ledger + if ( + chain_state.last_event_ledger is not None + and backend_ledger is not None + and backend_ledger < chain_state.last_event_ledger + ): + findings.append( + DriftFinding( + project_id=project_id, + scope="project", + field="last_event_ledger", + backend_value=backend_ledger, + chain_derived_value=chain_state.last_event_ledger, + severity=_FIELD_SEVERITY["last_event_ledger"], + ) + ) + + for milestone_id, chain_status in chain_state.milestones.items(): + backend_status = backend_milestones.get(milestone_id) + if backend_status is None: + findings.append( + DriftFinding( + project_id=project_id, + scope="milestone", + milestone_id=milestone_id, + field="missing_milestone", + backend_value=None, + chain_derived_value=chain_status, + severity=_FIELD_SEVERITY["missing_milestone"], + ) + ) + elif backend_status != chain_status: + findings.append( + DriftFinding( + project_id=project_id, + scope="milestone", + milestone_id=milestone_id, + field="status", + backend_value=backend_status, + chain_derived_value=chain_status, + severity=_FIELD_SEVERITY["status"], + ) + ) + + return findings + + # -- Public entrypoints ------------------------------------------------ + + def run_for_project(self, project_id: int) -> DriftReport: + """Run drift detection for a single project. Read-only; no mutation.""" + report = DriftReport( + run_id=str(uuid.uuid4()), + started_at=datetime.now(timezone.utc), + ) + + chain_state = self._compute_chain_derived_state(project_id) + if chain_state is None: + logger.info( + "Metadata drift check: no contract events found for project %s", + project_id, + ) + report.completed_at = datetime.now(timezone.utc) + return report + + report.projects_checked = 1 + report.findings = self._compare_project(chain_state) + report.completed_at = datetime.now(timezone.utc) + return report + + def run_all(self, limit: int = 500) -> DriftReport: + """ + Run drift detection across all projects that have at least one + ingested contract event. Supports both scheduled and manual + invocation; never mutates ProjectView/ProjectMilestone. + """ + report = DriftReport( + run_id=str(uuid.uuid4()), + started_at=datetime.now(timezone.utc), + ) + + project_ids = self._get_known_project_ids(limit=limit) + for project_id in project_ids: + chain_state = self._compute_chain_derived_state(project_id) + if chain_state is None: + continue + report.projects_checked += 1 + report.findings.extend(self._compare_project(chain_state)) + + report.completed_at = datetime.now(timezone.utc) + return report + + def run_and_persist(self, project_id: Optional[int] = None, limit: int = 500) -> DriftReport: + """Run detection and persist any findings for maintainer review.""" + report = self.run_for_project(project_id) if project_id is not None else self.run_all(limit=limit) + + if report.findings: + finding_dicts = [ + { + "run_id": report.run_id, + "project_id": f.project_id, + "scope": f.scope, + "milestone_id": f.milestone_id, + "field": f.field, + "backend_value": _stringify(f.backend_value), + "chain_derived_value": _stringify(f.chain_derived_value), + "severity": f.severity, + "detected_at": f.detected_at, + } + for f in report.findings + ] + saved = self.db_service.save_metadata_drift_findings(finding_dicts) + logger.warning( + "Metadata drift detected: run_id=%s projects_checked=%d " + "projects_with_drift=%d findings=%d (persisted=%d)", + report.run_id, + report.projects_checked, + report.projects_with_drift, + len(report.findings), + saved, + ) + else: + logger.info( + "Metadata drift check complete: run_id=%s projects_checked=%d, no drift detected", + report.run_id, + report.projects_checked, + ) + + return report + + +def _stringify(value: Any) -> Optional[str]: + if value is None: + return None + return str(value) + + +def run_manual_drift_check(project_id: Optional[int] = None) -> Dict[str, Any]: + """Manually trigger a drift check (e.g. from a CLI script or API route).""" + detector = MetadataDriftDetector() + report = detector.run_and_persist(project_id=project_id) + return report.to_dict() diff --git a/apps/data-processing/src/scheduler.py b/apps/data-processing/src/scheduler.py index 93cac472..1e58166a 100644 --- a/apps/data-processing/src/scheduler.py +++ b/apps/data-processing/src/scheduler.py @@ -26,6 +26,7 @@ from src.db.postgres_service import PostgresService from src.ingestion.rpc_benchmark import RPCProviderBenchmark from src.round_analyzer import _round_analyzer_job +from src.metadata_drift_detector import MetadataDriftDetector logger = setup_logger(__name__) @@ -270,6 +271,29 @@ def _contributor_reputation_snapshot_job() -> None: exc_info=True, ) + +def _metadata_drift_detector_job() -> None: + """Scheduled wrapper for MetadataDriftDetector (#882). + + Recomputes chain-derived project/milestone state from the ContractEvent + log and diffs it against ProjectView/ProjectMilestone. Read-only with + respect to source data — findings are persisted separately for review. + Errors are caught so the scheduler keeps running. + """ + try: + detector = MetadataDriftDetector() + report = detector.run_and_persist( + limit=int(os.getenv("METADATA_DRIFT_PROJECT_LIMIT", "500")) + ) + logger.info( + "Metadata drift detection: projects_checked=%d projects_with_drift=%d findings=%d", + report.projects_checked, + report.projects_with_drift, + len(report.findings), + ) + except Exception as exc: + logger.error(f"Metadata drift detector job failed: {exc}", exc_info=True) + class AnalyticsScheduler: """Manages the APScheduler scheduler for analytics jobs""" @@ -358,6 +382,15 @@ def start(self): replace_existing=True, ) + # ── Metadata Drift Detection: every 6 hours (#882) ─────────── + self.scheduler.add_job( + func=_metadata_drift_detector_job, + trigger=IntervalTrigger(hours=6), + id="metadata_drift_detection", + name="Metadata Drift Detector (backend vs on-chain)", + replace_existing=True, + ) + self.scheduler.start() logger.info("✓ Analytics scheduler started") logger.info(f" - Job: {market_job.name} | Next: {market_job.next_run_time}") diff --git a/apps/data-processing/tests/test_metadata_drift_detector.py b/apps/data-processing/tests/test_metadata_drift_detector.py new file mode 100644 index 00000000..4d7ebbd7 --- /dev/null +++ b/apps/data-processing/tests/test_metadata_drift_detector.py @@ -0,0 +1,260 @@ +"""Tests for the metadata drift detector (#882).""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from src.db.models import Base +from src.db.postgres_service import PostgresService +from src.metadata_drift_detector import MetadataDriftDetector + + +def build_sqlite_service() -> PostgresService: + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(bind=engine) + + service = PostgresService.__new__(PostgresService) + service.database_url = "sqlite:///:memory:" + service.engine = engine + service.SessionLocal = sessionmaker( + autocommit=False, + autoflush=False, + expire_on_commit=False, + bind=engine, + ) + return service + + +def _ts(offset_seconds: int = 0) -> datetime: + return datetime.now(timezone.utc) + timedelta(seconds=offset_seconds) + + +def test_no_drift_when_view_matches_events(): + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT1", + event_id="evt-1", + ledger=100, + event_type="DepositEvent", + project_id=1, + contributor="GALICE", + amount=50.0, + status="active", + timestamp=_ts(0), + ) + service.save_contract_event( + contract_id="CCONTRACT1", + event_id="evt-2", + ledger=101, + event_type="ContributionRecordedEvent", + project_id=1, + contributor="GBOB", + amount=25.0, + timestamp=_ts(1), + ) + + service.save_project_view( + project_id=1, + contract_id="CCONTRACT1", + status="active", + add_total_contributions=75.0, + unique_contributors=2, + last_event_ledger=101, + ) + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_for_project(1) + + assert report.projects_checked == 1 + assert report.drift_detected is False + assert report.findings == [] + + +def test_detects_status_and_amount_drift(): + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT2", + event_id="evt-1", + ledger=200, + event_type="DepositEvent", + project_id=2, + contributor="GALICE", + amount=100.0, + status="completed", + timestamp=_ts(0), + ) + + # Backend view is stale: status still "active", contributions undercounted. + service.save_project_view( + project_id=2, + contract_id="CCONTRACT2", + status="active", + add_total_contributions=10.0, + unique_contributors=1, + last_event_ledger=200, + ) + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_for_project(2) + + assert report.drift_detected is True + fields = {f.field for f in report.findings} + assert "status" in fields + assert "total_contributions" in fields + + status_finding = next(f for f in report.findings if f.field == "status") + assert status_finding.backend_value == "active" + assert status_finding.chain_derived_value == "completed" + assert status_finding.severity == "critical" + + +def test_detects_missing_project_view(): + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT3", + event_id="evt-1", + ledger=300, + event_type="DepositEvent", + project_id=3, + contributor="GALICE", + amount=10.0, + timestamp=_ts(0), + ) + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_for_project(3) + + assert report.drift_detected is True + assert report.findings[0].field == "missing_project_view" + assert report.findings[0].severity == "critical" + + +def test_detects_milestone_status_drift(): + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT4", + event_id="evt-1", + ledger=400, + event_type="MilestoneApprovedEvent", + project_id=4, + milestone_id=1, + status="approved", + timestamp=_ts(0), + ) + service.save_project_view(project_id=4, contract_id="CCONTRACT4", status="active") + service.save_project_milestone(project_id=4, milestone_id=1, status="pending") + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_for_project(4) + + milestone_findings = [f for f in report.findings if f.scope == "milestone"] + assert len(milestone_findings) == 1 + assert milestone_findings[0].field == "status" + assert milestone_findings[0].backend_value == "pending" + assert milestone_findings[0].chain_derived_value == "approved" + + +def test_does_not_mutate_source_records(): + """Acceptance criterion: detector must not mutate backend records.""" + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT5", + event_id="evt-1", + ledger=500, + event_type="DepositEvent", + project_id=5, + contributor="GALICE", + amount=100.0, + status="completed", + timestamp=_ts(0), + ) + service.save_project_view( + project_id=5, + contract_id="CCONTRACT5", + status="active", + add_total_contributions=10.0, + ) + + detector = MetadataDriftDetector(db_service=service) + detector.run_for_project(5) + + backend_view = service.get_project_view(5) + # Still the stale, pre-drift-detection values — detector did not mutate them. + assert backend_view.status == "active" + assert backend_view.total_contributions == 10.0 + + +def test_run_for_project_with_no_events_reports_nothing(): + service = build_sqlite_service() + detector = MetadataDriftDetector(db_service=service) + report = detector.run_for_project(999) + + assert report.projects_checked == 0 + assert report.drift_detected is False + + +def test_run_all_checks_every_project_with_events(): + service = build_sqlite_service() + + for project_id in (10, 11): + service.save_contract_event( + contract_id=f"CCONTRACT{project_id}", + event_id="evt-1", + ledger=100, + event_type="DepositEvent", + project_id=project_id, + contributor="GALICE", + amount=10.0, + status="active", + timestamp=_ts(0), + ) + + # Only one of the two gets a matching backend view. + service.save_project_view( + project_id=10, + contract_id="CCONTRACT10", + status="active", + add_total_contributions=10.0, + unique_contributors=1, + last_event_ledger=100, + ) + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_all(limit=10) + + assert report.projects_checked == 2 + assert report.drift_detected is True + assert any(f.project_id == 11 and f.field == "missing_project_view" for f in report.findings) + + +def test_run_and_persist_writes_findings_for_review(): + service = build_sqlite_service() + + service.save_contract_event( + contract_id="CCONTRACT6", + event_id="evt-1", + ledger=600, + event_type="DepositEvent", + project_id=6, + contributor="GALICE", + amount=100.0, + status="completed", + timestamp=_ts(0), + ) + service.save_project_view(project_id=6, contract_id="CCONTRACT6", status="active") + + detector = MetadataDriftDetector(db_service=service) + report = detector.run_and_persist(project_id=6) + + assert report.drift_detected is True + + persisted = service.get_metadata_drift_findings(run_id=report.run_id) + assert len(persisted) == len(report.findings) + assert all(f.reviewed is False for f in persisted)