From 3fa7a29f48645eebc43af484be912f0746f7cb20 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 3 Aug 2026 06:09:53 +0200 Subject: [PATCH] Fence and reconcile Dataflow runs --- README.md | 12 +- docs/DURABLE_RUN_RECOVERY.md | 48 ++ src/govoplan_dataflow/backend/manifest.py | 11 +- src/govoplan_dataflow/backend/recovery.py | 587 +++++++++++++++++++ src/govoplan_dataflow/backend/router.py | 14 +- src/govoplan_dataflow/backend/run_worker.py | 132 ++++- src/govoplan_dataflow/backend/schemas.py | 7 + src/govoplan_dataflow/backend/service.py | 150 ++++- tests/test_run_worker.py | 191 +++++- tests/test_service.py | 160 +++++ webui/src/api/dataflow.ts | 8 +- webui/src/features/dataflow/DataflowPage.tsx | 26 +- webui/src/styles/dataflow.css | 12 + 13 files changed, 1314 insertions(+), 44 deletions(-) create mode 100644 docs/DURABLE_RUN_RECOVERY.md create mode 100644 src/govoplan_dataflow/backend/recovery.py diff --git a/README.md b/README.md index 95e72f1..cbbe127 100644 --- a/README.md +++ b/README.md @@ -103,9 +103,15 @@ counts, outcomes, and publication references remain as audit evidence. Development runs may use the bounded reference backend. Staging and production runs require the short-lived isolated DuckDB process. A revision must be promoted from development to staging and then from staging to production before -it can run in those environments. Complete results can be published atomically -through `datasources.publication`; publication is rejected when a source or -result was truncated. +it can run in those environments. Complete results can be published through +`datasources.publication`; publication is rejected when a source or result was +truncated. The provider effect is a forward-recovery boundary rather than an +atomic database operation. Dataflow records the source and output digests before +dispatch, commits a conclusive provider result together with the local run +projection, and exposes an `outcome_unknown` run when acknowledgement is lost. +Such a run is never retried until the sink has been reconciled by its stable +idempotency key. See +[`docs/DURABLE_RUN_RECOVERY.md`](docs/DURABLE_RUN_RECOVERY.md). ## Governed Definitions And Automation diff --git a/docs/DURABLE_RUN_RECOVERY.md b/docs/DURABLE_RUN_RECOVERY.md new file mode 100644 index 0000000..c1facf9 --- /dev/null +++ b/docs/DURABLE_RUN_RECOVERY.md @@ -0,0 +1,48 @@ +# Durable Dataflow Run Recovery + +Dataflow uses Core's recovery ledger and distributed leases for every complete +run. The run ID, attempt number, pinned definition hash, canonical request hash, +invocation provenance, and access-context digest are recorded before execution. +Secrets and resolved credentials are never recovery evidence. + +## Database-Only Runs + +Runs without output publication use the `atomic` mode. Source resolution and +typed execution are recomputable and intermediate rows remain ephemeral. The +terminal `DataflowRun` projection and verified recovery checkpoint commit in +one database transaction. A stale transaction is failed by Core's fencing +contract and may be retried under the next attempt number. + +## Published Output + +Runs with a `datasources.publication` target use `forward_recovery`. Before +dispatch, Dataflow commits the bounded calculation result and records an output +digest, source-fingerprint digest, row count, and publication-idempotency digest. +The actual sink effect is not claimed to be atomic. + +A conclusive provider response supplies publication, datasource, and +materialization references. Those references, the local `succeeded` run state, +and the verified recovery checkpoint then commit together. If dispatch may have +started but no conclusive response exists, the run and ledger become +`outcome_unknown`; automatic retry is prohibited. + +## Crash And Retry Rules + +- Before output dispatch, an expired forward-recovery attempt is proven absent, + closed as recovered, and may be retried with a new fenced attempt. +- At or after output dispatch, an expired attempt is not retried. The sink must + be inspected using the stable idempotency key and recorded output digest. +- A failed evidence-chain verification prevents the local success projection + from committing. +- Replaying the same run idempotency key returns the existing run, including an + unresolved run, and never performs the publication again. +- Missing optional Datasources publication capability fails before dispatch and + records that no external output was created. + +## Operator View + +The Dataflow run history shows the Core recovery status and an explicit warning +for unresolved runs. Ops provides the platform-wide ledger projection, but sink +reconciliation belongs to Dataflow and the owning Datasources/provider adapter. +An operator must not infer absence from a timeout, process crash, or missing +local output reference. diff --git a/src/govoplan_dataflow/backend/manifest.py b/src/govoplan_dataflow/backend/manifest.py index 2540fcd..c2ab4ca 100644 --- a/src/govoplan_dataflow/backend/manifest.py +++ b/src/govoplan_dataflow/backend/manifest.py @@ -176,6 +176,11 @@ DOCUMENTATION = ( "aggregate, sort, limit, output, revisioning, and bounded preview." ), "sql_safety": "Constrained AST compilation only; no pass-through execution.", + "run_recovery": ( + "Database-only runs commit atomically with Core recovery evidence. " + "Output publication uses forward recovery and blocks blind retry " + "when provider acknowledgement is uncertain." + ), }, ), ) @@ -432,10 +437,12 @@ manifest = ModuleManifest( maturity="vertical_slice", documentation_ref="README.md", test_ref="tests/test_golden_flows.py", - known_limits=("Execution adapters and operator recovery evidence do not yet cover every declared node family.",), + known_limits=( + "Execution adapters do not yet cover every declared node family.", + ), owned_concepts=("dataflow definition", "dataflow revision", "dataflow run", "transformation graph"), non_owned_concepts=("datasource binding", "connector transport", "report presentation", "workflow task"), - recovery_docs=("README.md",), + recovery_docs=("README.md", "docs/DURABLE_RUN_RECOVERY.md"), security_docs=("README.md",), operations_docs=("README.md",), ), diff --git a/src/govoplan_dataflow/backend/recovery.py b/src/govoplan_dataflow/backend/recovery.py new file mode 100644 index 0000000..3eceb5e --- /dev/null +++ b/src/govoplan_dataflow/backend/recovery.py @@ -0,0 +1,587 @@ +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +from typing import Mapping, Sequence + +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_core.core.recovery import ( + RecoveryGuaranteeError, + RecoveryMode, + RecoveryOperation, + RecoveryPlan, + RecoveryStatus, +) +from govoplan_core.core.recovery_runtime import ( + DurableRecoveryOperation, + RecoveryOperationBusy, + RecoveryOperationStateConflict, + begin_durable_recovery_operation, + claim_durable_recovery_operation, +) +from govoplan_core.core.runtime_coordination import process_runtime_identity +from govoplan_dataflow.backend.db.models import DataflowRun + + +class DataflowRecoveryError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class DataflowRecoveryDeclaration: + operation_type: str + mode: RecoveryMode + boundaries: tuple[Mapping[str, object], ...] + verification: tuple[str, ...] + + +DATAFLOW_RECOVERY_OPERATIONS = ( + DataflowRecoveryDeclaration( + operation_type="run.database-only", + mode=RecoveryMode.ATOMIC, + boundaries=( + { + "name": "source-resolution", + "classification": "recomputable", + "resume": "restart with the pinned definition and verified source fingerprints", + }, + { + "name": "typed-execution", + "classification": "ephemeral", + "resume": "restart; intermediate rows are not persisted", + }, + { + "name": "run-finalization", + "classification": "atomic", + "resume": "the run row and terminal recovery checkpoint commit together", + }, + ), + verification=( + "verify the pinned definition and canonical request hashes", + "verify source fingerprints, diagnostics, and bounded row counts", + "commit the run result and terminal recovery checkpoint atomically", + ), + ), + DataflowRecoveryDeclaration( + operation_type="run.publish-output", + mode=RecoveryMode.FORWARD_RECOVERY, + boundaries=( + { + "name": "source-resolution", + "classification": "recomputable", + "resume": "restart only before output dispatch", + }, + { + "name": "typed-execution", + "classification": "checkpointed", + "resume": "the output digest and source fingerprints identify the computed result", + }, + { + "name": "output-publication", + "classification": "outcome-unknown", + "resume": "verify the sink by idempotency key; never publish blindly after a lost acknowledgement", + }, + ), + verification=( + "verify the pinned definition, request, source, and output hashes", + "verify the publication, datasource, and materialization references", + "reconcile an uncertain sink outcome before any new attempt", + ), + ), +) + + +_DECLARATIONS = { + item.operation_type: item for item in DATAFLOW_RECOVERY_OPERATIONS +} + + +def dataflow_session_factory(session: Session) -> sessionmaker[Session]: + bind = session.get_bind() + if bind is None: + raise DataflowRecoveryError("Dataflow recovery requires a database bind") + return sessionmaker(bind=bind, expire_on_commit=False) + + +def _canonical_sha256(value: object) -> str: + encoded = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + default=str, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _operation_type(run: DataflowRun) -> str: + return ( + "run.publish-output" + if isinstance(run.request_.get("publication"), Mapping) + else "run.database-only" + ) + + +@dataclass(slots=True) +class DataflowRunRecovery: + operation: DurableRecoveryOperation | None + operation_id: str + operation_type: str + mode: RecoveryMode + attempt: int + replayed: bool + publication_started: bool = False + + def prepare_publication( + self, + session: Session, + *, + run: DataflowRun, + rows: Sequence[Mapping[str, object]], + ) -> None: + if self.mode != RecoveryMode.FORWARD_RECOVERY: + raise DataflowRecoveryError( + "Only forward-recovery runs may cross a publication boundary" + ) + if self.operation is None: + raise DataflowRecoveryError( + "A replayed recovery operation cannot publish output again" + ) + output_sha256 = _canonical_sha256([dict(row) for row in rows]) + source_sha256 = _canonical_sha256(run.source_fingerprints) + _set_run_recovery_metadata( + run, + operation_id=self.operation_id, + mode=self.mode, + status=RecoveryStatus.RUNNING.value, + attempt=self.attempt, + boundary="output-publication", + output_sha256=output_sha256, + ) + try: + session.commit() + self.operation.checkpoint( + kind="output-publication-dispatch", + summary="The computed output is durable before sink dispatch", + evidence={ + "effect_started": False, + "output_sha256": output_sha256, + "source_fingerprints_sha256": source_sha256, + "output_row_count": run.output_row_count, + "publication_idempotency_sha256": _canonical_sha256( + f"{run.pipeline_id}:{run.idempotency_key or run.id}" + ), + }, + ) + except Exception as exc: + session.rollback() + raise DataflowRecoveryError( + "Dataflow output could not be checkpointed before publication" + ) from exc + self.publication_started = True + + def finish(self, session: Session, *, run: DataflowRun) -> None: + if self.operation is None: + return + evidence = _terminal_evidence(run, publication_started=self.publication_started) + try: + if self.mode == RecoveryMode.ATOMIC: + _set_run_recovery_metadata( + run, + operation_id=self.operation_id, + mode=self.mode, + status=( + RecoveryStatus.SUCCEEDED.value + if run.status == "succeeded" + else RecoveryStatus.FAILED.value + ), + attempt=self.attempt, + boundary="run-finalization", + ) + if run.status == "succeeded": + self.operation.commit_atomic_success(session, evidence=evidence) + else: + self.operation.commit_atomic_failure( + session, + summary=run.error or f"Dataflow run ended as {run.status}", + evidence=evidence, + ) + return + + if self.publication_started and run.status != "succeeded": + run.status = "outcome_unknown" + run.progress_phase = "outcome_unknown" + run.progress_percent = min(run.progress_percent, 95) + run.error = ( + "The output provider may have accepted the publication; " + "verify the sink before retrying." + ) + _set_run_recovery_metadata( + run, + operation_id=self.operation_id, + mode=self.mode, + status=RecoveryStatus.OUTCOME_UNKNOWN.value, + attempt=self.attempt, + boundary="output-publication", + ) + session.commit() + self.operation.unresolved( + status=RecoveryStatus.OUTCOME_UNKNOWN, + summary="The output acknowledgement was not conclusive", + evidence={ + **evidence, + "effect_started": True, + }, + failure_summary=( + "Inspect the publication target by stable idempotency key " + "before any new output attempt" + ), + ) + return + + terminal_status = ( + RecoveryStatus.SUCCEEDED.value + if run.status == "succeeded" + else RecoveryStatus.RECOVERED.value + ) + _set_run_recovery_metadata( + run, + operation_id=self.operation_id, + mode=self.mode, + status=terminal_status, + attempt=self.attempt, + boundary=( + "output-publication" + if run.status == "succeeded" + else "typed-execution" + ), + ) + if run.status == "succeeded": + self.operation.commit_verified_success( + session, + evidence=evidence, + ) + else: + session.commit() + failure_summary = ( + run.error or f"Dataflow run ended as {run.status}" + ) + self.operation.compensate( + failure_summary=failure_summary, + failure_evidence={ + **evidence, + "effect_started": False, + }, + recovery_evidence={ + "verified": True, + "checks": { + "effect_started": False, + "publication_absent": True, + }, + }, + ) + except Exception as exc: + session.rollback() + raise DataflowRecoveryError( + "Dataflow run state and recovery evidence could not be finalized" + ) from exc + + +def begin_dataflow_run_recovery( + session: Session, + *, + run: DataflowRun, + lease_ttl_seconds: int, +) -> DataflowRunRecovery: + if not run.id or not run.request_hash or not run.definition_hash: + raise DataflowRecoveryError( + "Dataflow recovery requires a run ID, request hash, and definition hash" + ) + operation_type = _operation_type(run) + declaration = _DECLARATIONS[operation_type] + attempt = max(1, int(run.attempts)) + operation_key = f"dataflow-run:{run.id}:attempt:{attempt}" + try: + started = begin_durable_recovery_operation( + dataflow_session_factory(session), + identity=process_runtime_identity(), + module_id="dataflow", + operation_type=operation_type, + idempotency_key=operation_key, + request={ + "tenant_id": run.tenant_id, + "run_id": run.id, + "pipeline_id": run.pipeline_id, + "pipeline_revision_id": run.pipeline_revision_id, + "definition_hash": run.definition_hash, + "request_hash": run.request_hash, + "invocation_kind": run.invocation_kind, + "trigger_delivery_id": run.trigger_delivery_id, + "attempt": attempt, + "publication": operation_type == "run.publish-output", + }, + recovery_plan=RecoveryPlan( + mode=declaration.mode, + preconditions=( + "the pinned pipeline revision and canonical request hashes are present", + "the current authorization and deployment are revalidated", + "the run attempt owns a distributed execution fence", + ), + forward_recovery_steps=( + "inspect the output sink by stable publication idempotency key", + "record whether the expected output digest was accepted", + "resume only after accepted or absent state is proven", + ) + if declaration.mode == RecoveryMode.FORWARD_RECOVERY + else (), + verification_steps=declaration.verification, + ), + precondition_evidence={ + "definition_hash": run.definition_hash, + "request_hash": run.request_hash, + "access_context_sha256": _canonical_sha256(run.authorization_), + "environment": run.environment, + "attempt": attempt, + "publication": operation_type == "run.publish-output", + }, + lease_resource_key=f"dataflow:run:{run.id}", + lease_ttl_seconds=max(60, int(lease_ttl_seconds)), + resource_type="dataflow_run", + resource_id=run.id, + metadata={ + "resources": ( + ["postgresql", "queue", "external-sink"] + if declaration.mode == RecoveryMode.FORWARD_RECOVERY + else ["postgresql"] + ), + "boundaries": [dict(item) for item in declaration.boundaries], + "attempt": attempt, + }, + ) + except RecoveryOperationBusy as exc: + raise DataflowRecoveryError( + "Another runtime owns this Dataflow run attempt" + ) from exc + except RecoveryOperationStateConflict as exc: + raise DataflowRecoveryError( + "This Dataflow run attempt is already active or unresolved" + ) from exc + except (RecoveryGuaranteeError, RuntimeError) as exc: + raise DataflowRecoveryError( + "The recovery ledger is unavailable; Dataflow execution did not start" + ) from exc + _set_run_recovery_metadata( + run, + operation_id=started.operation_id, + mode=declaration.mode, + status=started.status, + attempt=attempt, + boundary="source-resolution", + ) + return DataflowRunRecovery( + operation=started.operation, + operation_id=started.operation_id, + operation_type=operation_type, + mode=declaration.mode, + attempt=attempt, + replayed=started.replayed, + ) + + +def dataflow_run_recovery_state( + session: Session, + *, + run_id: str, +) -> dict[str, object] | None: + return dataflow_run_recovery_states(session, run_ids=(run_id,)).get(run_id) + + +def dataflow_run_recovery_states( + session: Session, + *, + run_ids: Sequence[str], +) -> dict[str, dict[str, object]]: + normalized_ids = tuple(dict.fromkeys(str(run_id) for run_id in run_ids if run_id)) + if not normalized_ids: + return {} + operations = session.scalars( + select(RecoveryOperation) + .where( + RecoveryOperation.module_id == "dataflow", + RecoveryOperation.resource_type == "dataflow_run", + RecoveryOperation.resource_id.in_(normalized_ids), + ) + .order_by(RecoveryOperation.created_at.desc(), RecoveryOperation.id.desc()) + ) + states: dict[str, dict[str, object]] = {} + for operation in operations: + run_id = str(operation.resource_id or "") + if run_id and run_id not in states: + states[run_id] = _recovery_state(operation) + return states + + +def _recovery_state(operation: RecoveryOperation) -> dict[str, object]: + status = str(operation.status) + return { + "operation_id": operation.id, + "operation_type": operation.operation_type, + "mode": operation.mode, + "status": status, + "checkpoint_count": operation.checkpoint_count, + "requires_attention": status + in { + RecoveryStatus.OUTCOME_UNKNOWN.value, + RecoveryStatus.RECOVERY_REQUIRED.value, + RecoveryStatus.MANUAL_INTERVENTION.value, + }, + "explanation": _recovery_explanation(operation), + } + + +def claim_stale_dataflow_recovery( + session: Session, + *, + operation_id: str, + lease_ttl_seconds: int, + effect_started: bool, +) -> str: + try: + handle = claim_durable_recovery_operation( + dataflow_session_factory(session), + identity=process_runtime_identity(), + operation_id=operation_id, + lease_ttl_seconds=max(60, int(lease_ttl_seconds)), + ) + except RecoveryOperationStateConflict as exc: + return exc.status + except RecoveryOperationBusy as exc: + raise DataflowRecoveryError( + "Another runtime still owns the stale Dataflow recovery fence" + ) from exc + except (RecoveryGuaranteeError, RuntimeError) as exc: + raise DataflowRecoveryError( + "The stale Dataflow recovery fence could not be claimed" + ) from exc + if effect_started: + handle.release_unresolved() + else: + handle.resolve_unknown( + effect_occurred=False, + summary="No output dispatch checkpoint was recorded before lease expiry", + evidence={ + "verified": True, + "checks": {"effect_started": False}, + }, + ) + session.expire_all() + state = dataflow_run_recovery_state( + session, + run_id=_operation_run_id(session, operation_id), + ) + return str(state["status"] if state is not None else RecoveryStatus.OUTCOME_UNKNOWN.value) + + +def _operation_run_id(session: Session, operation_id: str) -> str: + operation = session.get(RecoveryOperation, operation_id) + return str(operation.resource_id or "") if operation is not None else "" + + +def _terminal_evidence( + run: DataflowRun, + *, + publication_started: bool, +) -> dict[str, object]: + succeeded = run.status == "succeeded" + publication_expected = isinstance(run.request_.get("publication"), Mapping) + publication_verified = bool( + not publication_expected + or ( + run.output_publication_ref + and run.output_datasource_ref + and run.output_materialization_ref + ) + ) + return { + "verified": bool((succeeded and publication_verified) or not succeeded), + "checks": { + "run_status": run.status, + "definition_hash": run.definition_hash, + "request_hash": run.request_hash, + "source_fingerprints_sha256": _canonical_sha256( + run.source_fingerprints + ), + "result_schema_sha256": _canonical_sha256(run.result_schema), + "input_row_count": run.input_row_count, + "output_row_count": run.output_row_count, + "publication_expected": publication_expected, + "publication_started": publication_started, + "publication_verified": publication_verified, + "output_publication_ref": run.output_publication_ref, + "output_datasource_ref": run.output_datasource_ref, + "output_materialization_ref": run.output_materialization_ref, + }, + } + + +def _set_run_recovery_metadata( + run: DataflowRun, + *, + operation_id: str, + mode: RecoveryMode, + status: str, + attempt: int, + boundary: str, + output_sha256: str | None = None, +) -> None: + authorization = dict(run.authorization_) + recovery = { + "operation_id": operation_id, + "mode": mode.value, + "status": status, + "attempt": attempt, + "boundary": boundary, + } + if output_sha256: + recovery["output_sha256"] = output_sha256 + authorization["recovery"] = recovery + run.authorization_ = authorization + + +def _recovery_explanation(operation: RecoveryOperation) -> str: + status = str(operation.status) + if status == RecoveryStatus.OUTCOME_UNKNOWN.value: + return ( + "The output may have been accepted. Verify the sink using the " + "recorded idempotency key before starting another publication." + ) + if status in { + RecoveryStatus.RECOVERY_REQUIRED.value, + RecoveryStatus.RECOVERING.value, + }: + return operation.failure_summary or "The run requires explicit recovery." + if status == RecoveryStatus.SUCCEEDED.value: + return "The run outcome and declared output references were verified." + if status == RecoveryStatus.FAILED.value: + return operation.failure_summary or "The run failed before an uncertain output." + if status == RecoveryStatus.RECOVERED.value: + return "The run stopped before publication and no external output was created." + if status == RecoveryStatus.REJECTED.value: + return operation.failure_summary or "The run was definitively rejected." + return "The run is fenced and has not reached a verified terminal state." + + +__all__ = [ + "DATAFLOW_RECOVERY_OPERATIONS", + "DataflowRecoveryDeclaration", + "DataflowRecoveryError", + "DataflowRunRecovery", + "begin_dataflow_run_recovery", + "claim_stale_dataflow_recovery", + "dataflow_run_recovery_state", + "dataflow_run_recovery_states", + "dataflow_session_factory", +] diff --git a/src/govoplan_dataflow/backend/router.py b/src/govoplan_dataflow/backend/router.py index 2d61163..275389d 100644 --- a/src/govoplan_dataflow/backend/router.py +++ b/src/govoplan_dataflow/backend/router.py @@ -110,6 +110,7 @@ from govoplan_dataflow.backend.service import ( validate_draft, ) from govoplan_dataflow.backend.run_worker import run_metrics +from govoplan_dataflow.backend.recovery import dataflow_run_recovery_states from govoplan_dataflow.backend.triggers import ( create_trigger, delete_trigger, @@ -982,8 +983,19 @@ def api_list_pipeline_runs( pipeline_id=pipeline_id, limit=limit, ) + recovery_states = dataflow_run_recovery_states( + session, + run_ids=tuple(run.id for run in runs), + ) return PipelineRunListResponse( - runs=[pipeline_run_response(session, run) for run in runs] + runs=[ + pipeline_run_response( + session, + run, + recovery_state=recovery_states.get(run.id, {}), + ) + for run in runs + ] ) except PermissionError as exc: raise _governance_http_error(exc) from exc diff --git a/src/govoplan_dataflow/backend/run_worker.py b/src/govoplan_dataflow/backend/run_worker.py index 64ac86e..e5065ae 100644 --- a/src/govoplan_dataflow/backend/run_worker.py +++ b/src/govoplan_dataflow/backend/run_worker.py @@ -25,6 +25,13 @@ from govoplan_dataflow.backend.db.models import ( DataflowTrigger, ) from govoplan_dataflow.backend.governance import require_definition_action +from govoplan_dataflow.backend.recovery import ( + DataflowRecoveryError, + DataflowRunRecovery, + begin_dataflow_run_recovery, + claim_stale_dataflow_recovery, + dataflow_run_recovery_state, +) from govoplan_dataflow.backend.service import ( _execute_pipeline_run, _require_deployed_revision, @@ -55,7 +62,10 @@ def dispatch_pending_runs( if worker_id and str(worker_id).strip() else socket.gethostname() ) - recovered = _recover_expired_leases(session, now=current) + recovered, recovered_outcome_unknown = _recover_expired_leases( + session, + now=current, + ) session.commit() summary: dict[str, object] = { "claimed": 0, @@ -63,6 +73,7 @@ def dispatch_pending_runs( "retrying": 0, "failed": 0, "cancelled": 0, + "outcome_unknown": recovered_outcome_unknown, "recovered": recovered, "runs": [], } @@ -193,7 +204,7 @@ def _recover_expired_leases( session: Session, *, now: datetime, -) -> int: +) -> tuple[int, int]: runs = list( session.scalars( select(DataflowRun) @@ -205,13 +216,46 @@ def _recover_expired_leases( .with_for_update(skip_locked=True) ) ) + recovered = 0 + outcome_unknown = 0 for run in runs: + recovery = dataflow_run_recovery_state(session, run_id=run.id) + recovery_status: str | None = None + recovery_metadata = dict(run.authorization_).get("recovery") + effect_started = bool( + isinstance(recovery_metadata, Mapping) + and recovery_metadata.get("boundary") == "output-publication" + ) + if recovery is not None and recovery.get("operation_id"): + try: + recovery_status = claim_stale_dataflow_recovery( + session, + operation_id=str(recovery["operation_id"]), + lease_ttl_seconds=DEFAULT_LEASE_SECONDS, + effect_started=effect_started, + ) + except DataflowRecoveryError: + logger.warning( + "Could not claim stale Dataflow recovery operation %s", + recovery["operation_id"], + exc_info=True, + ) + continue run.worker_id = None run.claimed_at = None run.lease_expires_at = None run.heartbeat_at = None if run.cancellation_requested_at is not None: _cancel_run(run, now=now) + elif recovery_status == "outcome_unknown" or effect_started: + run.status = "outcome_unknown" + run.finished_at = now + run.progress_phase = "outcome_unknown" + run.error = ( + "The worker lease expired after output publication may have " + "started. Verify the sink before retrying." + ) + outcome_unknown += 1 elif run.attempts < run.max_attempts: run.status = "retrying" run.available_at = now @@ -223,7 +267,8 @@ def _recover_expired_leases( now=now, message="The worker lease expired after the final attempt.", ) - return len(runs) + recovered += 1 + return recovered, outcome_unknown def _claim_next_run( @@ -291,6 +336,7 @@ def _execute_claimed_run( if run.cancellation_requested_at is not None: _cancel_run(run, now=now) return "cancelled" + recovery: DataflowRunRecovery | None = None try: principal, provenance = _resolve_principal( session, @@ -329,6 +375,16 @@ def _execute_claimed_run( revision=revision, environment=run.environment, ) + session.commit() + recovery = begin_dataflow_run_recovery( + session, + run=run, + lease_ttl_seconds=_lease_seconds(run), + ) + if recovery.replayed: + session.rollback() + run = session.get(DataflowRun, run_id) + return run.status if run is not None else "failed" _notify_run( session, registry=registry, @@ -344,6 +400,7 @@ def _execute_claimed_run( request=pipeline_run_request(run), principal=principal, registry=registry, + recovery=recovery, ) if ( run.status == "failed" @@ -359,7 +416,47 @@ def _execute_claimed_run( run=run, pipeline_name=pipeline.name, ) + recovery.finish(session, run=run) return run.status + except DataflowRecoveryError: + logger.exception( + "Dataflow recovery finalization failed for run %s", + run_id, + ) + session.rollback() + run = session.get(DataflowRun, run_id) + if run is not None: + if recovery is not None and recovery.publication_started: + run.status = "outcome_unknown" + run.finished_at = now + run.progress_phase = "outcome_unknown" + run.error = ( + "Output publication completed without verifiable terminal " + "recovery evidence; reconcile the sink." + ) + _finish_claim(run) + return "outcome_unknown" + if recovery is None: + _fail_run( + run, + now=now, + message=( + "The recovery ledger was unavailable; no Dataflow " + "execution started." + ), + ) + _finish_claim(run) + else: + _fail_run( + run, + now=now, + message=( + "The database-only run could not commit verifiable " + "recovery evidence." + ), + ) + _finish_claim(run) + return "failed" except (DataflowWorkerError, PermissionError, ValueError) as exc: _fail_run(run, now=now, message=str(exc)) _finish_claim(run) @@ -370,6 +467,8 @@ def _execute_claimed_run( run=run, pipeline_name="Dataflow", ) + if recovery is not None: + recovery.finish(session, run=run) return "failed" except Exception as exc: logger.exception( @@ -380,6 +479,24 @@ def _execute_claimed_run( run = session.get(DataflowRun, run_id) if run is None: return "failed" + if recovery is not None and recovery.publication_started: + run.status = "outcome_unknown" + run.finished_at = now + run.error = ( + "The output provider failed after dispatch began; verify the " + "sink before retrying." + ) + run.progress_phase = "outcome_unknown" + _finish_claim(run) + try: + recovery.finish(session, run=run) + except DataflowRecoveryError: + logger.exception( + "Could not persist the uncertain output state for run %s", + run_id, + ) + session.rollback() + return "outcome_unknown" _fail_run( run, now=now, @@ -389,6 +506,15 @@ def _execute_claimed_run( ), ) _finish_claim(run) + if recovery is not None: + try: + recovery.finish(session, run=run) + except DataflowRecoveryError: + logger.exception( + "Could not finalize failed Dataflow recovery for run %s", + run_id, + ) + session.rollback() return "failed" diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index 5d0895c..0a240fe 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -31,6 +31,7 @@ DataflowRunStatus = Literal[ "succeeded", "failed", "cancelled", + "outcome_unknown", ] @@ -335,6 +336,12 @@ class PipelineRunResponse(BaseModel): finished_at: datetime | None created_by: str | None created_at: datetime + recovery_operation_id: str | None = None + recovery_operation_type: str | None = None + recovery_mode: str | None = None + recovery_status: str | None = None + recovery_requires_attention: bool = False + recovery_explanation: str | None = None replayed: bool = False diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index 9b82d4b..5b18db9 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -40,6 +40,7 @@ from govoplan_dataflow.backend.db.models import ( DataflowPipelineDeployment, DataflowPipelineRevision, DataflowRun, + new_uuid, ) from govoplan_dataflow.backend.executor import ( EXECUTOR_VERSION, @@ -79,6 +80,12 @@ from govoplan_dataflow.backend.schemas import ( PipelineValidationResponse, PreviewColumn, ) +from govoplan_dataflow.backend.recovery import ( + DataflowRecoveryError, + DataflowRunRecovery, + begin_dataflow_run_recovery, + dataflow_run_recovery_state, +) from govoplan_dataflow.backend.sql_compiler import ( SqlCompilationError, compile_sql, @@ -1071,14 +1078,27 @@ def start_pipeline_run( principal=principal, defer_execution=defer_execution, ) - session.add(run) - session.flush() run.authorization_ = { **dict(run.authorization_), "authorization_ref": ( request.invocation.trigger_ref or f"dataflow-run:{run.id}" ), } + recovery: DataflowRunRecovery | None = None + if not defer_execution: + try: + recovery = begin_dataflow_run_recovery( + session, + run=run, + lease_ttl_seconds=int( + float(run.resource_budget.get("max_wall_seconds") or 30.0) + ) + + 60, + ) + except DataflowRecoveryError as exc: + raise DataflowConflictError(str(exc)) from exc + session.add(run) + session.flush() if defer_execution: session.flush() return run, False @@ -1090,8 +1110,27 @@ def start_pipeline_run( request=request, principal=principal, registry=registry, + recovery=recovery, ) - session.flush() + if recovery is not None: + try: + recovery.finish(session, run=run) + except DataflowRecoveryError as exc: + session.rollback() + if recovery.publication_started: + persisted = session.get(DataflowRun, run.id) + if persisted is not None: + persisted.status = "outcome_unknown" + persisted.finished_at = utcnow() + persisted.progress_phase = "outcome_unknown" + persisted.error = ( + "Output publication completed without verifiable " + "terminal recovery evidence; reconcile the sink." + ) + session.commit() + raise DataflowConflictError(str(exc)) from exc + else: + session.flush() return run, False @@ -1256,6 +1295,7 @@ def _new_pipeline_run( now = utcnow() budget = _run_resource_budget(request) return DataflowRun( + id=new_uuid(), tenant_id=tenant_id, pipeline_id=pipeline.id, pipeline_revision_id=revision.id, @@ -1360,6 +1400,7 @@ def _execute_pipeline_run( request: DataflowRunRequest, principal: ApiPrincipal, registry: object | None, + recovery: DataflowRunRecovery | None = None, ) -> bool: try: if run.cancellation_requested_at is not None: @@ -1405,6 +1446,7 @@ def _execute_pipeline_run( result=result, principal=principal, registry=registry, + recovery=recovery, ) run.status = "succeeded" run.finished_at = utcnow() @@ -1414,6 +1456,10 @@ def _execute_pipeline_run( return False except (DatasourceError, PipelineExecutionError) as exc: _mark_pipeline_run_failed(run, exc) + if recovery is not None and recovery.publication_started: + run.status = "outcome_unknown" + run.progress_phase = "outcome_unknown" + return False return isinstance(exc, DatasourceUnavailableError) or bool( getattr(exc, "retryable", False) ) @@ -1469,6 +1515,7 @@ def _publish_pipeline_result( result: PipelineExecutionResult, principal: ApiPrincipal, registry: object | None, + recovery: DataflowRunRecovery | None = None, ) -> None: publisher = datasource_publication(registry) if publisher is None: @@ -1479,33 +1526,51 @@ def _publish_pipeline_result( target = request.publication if target is None: return - publication = publisher.publish_rows( + if recovery is None: + raise PipelineExecutionError( + "Publishing Dataflow output requires a durable recovery operation." + ) + recovery.prepare_publication( session, - principal, - request=DatasourcePublicationRequest( - producer_module="dataflow", - producer_run_ref=f"dataflow-run:{run.id}", - idempotency_key=f"{pipeline.id}:{request.idempotency_key.strip()}", - rows=tuple(dict(row) for row in result.rows), - target_datasource_ref=target.target_datasource_ref, - name=target.name or f"{pipeline.name} output", - source_name=target.source_name, - description=target.description, - freeze=target.freeze, - frozen_label=target.frozen_label, - set_current=target.set_current, - provenance={ - "pipeline_ref": f"pipeline:{pipeline.id}", - "pipeline_revision": revision.revision, - "definition_hash": revision.content_hash, - "source_fingerprints": result.source_fingerprints, - }, - metadata={ - **dict(target.metadata), - "dataflow_run_ref": f"dataflow-run:{run.id}", - }, - ), + run=run, + rows=tuple(dict(row) for row in result.rows), ) + try: + publication = publisher.publish_rows( + session, + principal, + request=DatasourcePublicationRequest( + producer_module="dataflow", + producer_run_ref=f"dataflow-run:{run.id}", + idempotency_key=( + f"{pipeline.id}:{request.idempotency_key.strip()}" + ), + rows=tuple(dict(row) for row in result.rows), + target_datasource_ref=target.target_datasource_ref, + name=target.name or f"{pipeline.name} output", + source_name=target.source_name, + description=target.description, + freeze=target.freeze, + frozen_label=target.frozen_label, + set_current=target.set_current, + provenance={ + "pipeline_ref": f"pipeline:{pipeline.id}", + "pipeline_revision": revision.revision, + "definition_hash": revision.content_hash, + "source_fingerprints": result.source_fingerprints, + }, + metadata={ + **dict(target.metadata), + "dataflow_run_ref": f"dataflow-run:{run.id}", + }, + ), + ) + except DatasourceError: + raise + except Exception as exc: + raise PipelineExecutionError( + "The output provider failed after publication dispatch began." + ) from exc run.output_publication_ref = publication.ref run.output_datasource_ref = publication.datasource.ref run.output_materialization_ref = publication.materialization.ref @@ -1573,10 +1638,16 @@ def pipeline_run_response( run: DataflowRun, *, replayed: bool = False, + recovery_state: Mapping[str, object] | None = None, ) -> PipelineRunResponse: revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id) if revision is None: raise DataflowNotFoundError("Dataflow pipeline revision not found") + recovery = ( + dict(recovery_state) or None + if recovery_state is not None + else dataflow_run_recovery_state(session, run_id=run.id) + ) return PipelineRunResponse( ref=f"dataflow-run:{run.id}", pipeline_id=run.pipeline_id, @@ -1622,6 +1693,27 @@ def pipeline_run_response( finished_at=run.finished_at, created_by=run.created_by, created_at=run.created_at, + recovery_operation_id=( + str(recovery["operation_id"]) if recovery is not None else None + ), + recovery_operation_type=( + str(recovery["operation_type"]) if recovery is not None else None + ), + recovery_mode=( + str(recovery["mode"]) if recovery is not None else None + ), + recovery_status=( + str(recovery["status"]) if recovery is not None else None + ), + recovery_requires_attention=( + bool(recovery["requires_attention"]) + or run.status == "outcome_unknown" + if recovery is not None + else run.status == "outcome_unknown" + ), + recovery_explanation=( + str(recovery["explanation"]) if recovery is not None else None + ), replayed=replayed, ) @@ -1635,6 +1727,7 @@ def pipeline_run_descriptor( revision = session.get(DataflowPipelineRevision, run.pipeline_revision_id) if revision is None: raise DataflowNotFoundError("Dataflow pipeline revision not found") + recovery = dataflow_run_recovery_state(session, run_id=run.id) return DataflowRunDescriptor( ref=f"dataflow-run:{run.id}", pipeline_ref=f"pipeline:{run.pipeline_id}", @@ -1670,6 +1763,7 @@ def pipeline_run_descriptor( "progress_phase": run.progress_phase, "source_fingerprints": list(run.source_fingerprints), "diagnostics": list(run.diagnostics), + "recovery": dict(recovery) if recovery is not None else None, }, ) diff --git a/tests/test_run_worker.py b/tests/test_run_worker.py index cc7327a..e7c53db 100644 --- a/tests/test_run_worker.py +++ b/tests/test_run_worker.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import timedelta import unittest -from sqlalchemy import create_engine +from sqlalchemy import create_engine, select from sqlalchemy.orm import Session, sessionmaker from govoplan_core.auth import ApiPrincipal @@ -12,7 +12,26 @@ from govoplan_core.core.access import ( PrincipalRef, ) from govoplan_core.core.automation import AutomationPrincipalResolution -from govoplan_core.core.dataflows import DataflowRunRequest +from govoplan_core.core.dataflows import ( + DataflowPublicationTarget, + DataflowRunRequest, +) +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_PUBLICATION, + DatasourceDescriptor, + DatasourceMaterialization, + DatasourcePublicationResult, +) +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryOperation, + RecoveryStatus, +) +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + bind_process_runtime_identity, +) from govoplan_core.db.base import Base, utcnow from govoplan_dataflow.backend.db.models import ( DataflowPipeline, @@ -24,6 +43,7 @@ from govoplan_dataflow.backend.run_worker import ( dispatch_pending_runs, purge_expired_runs, ) +from govoplan_dataflow.backend.recovery import begin_dataflow_run_recovery from govoplan_dataflow.backend.schemas import ( GraphEdge, GraphNode, @@ -51,6 +71,17 @@ def _principal() -> ApiPrincipal: ) +def _runtime_identity() -> RuntimeIdentity: + return RuntimeIdentity( + installation_id="dataflow-worker-tests", + node_id="worker-node", + incarnation="worker-incarnation", + role="worker", + software_version="test", + composition_hash="b" * 64, + ) + + def _graph() -> PipelineGraph: return PipelineGraph( nodes=[ @@ -92,16 +123,54 @@ class _AutomationProvider: ) -class _Registry: +class _PublicationProvider: def __init__(self) -> None: + self.requests = [] + + def publish_rows(self, _session, _principal, *, request): + self.requests.append(request) + descriptor = DatasourceDescriptor( + ref="datasource:worker-output", + source_name="worker_output", + name="Worker output", + kind="custom", + mode="static", + shape="tabular", + fingerprint="worker-output-fingerprint", + ) + return DatasourcePublicationResult( + ref="publication:worker-output", + status="published", + datasource=descriptor, + materialization=DatasourceMaterialization( + ref="materialization:worker-output", + datasource_ref=descriptor.ref, + revision=1, + state="published", + fingerprint=descriptor.fingerprint, + ), + ) + + +class _Registry: + def __init__( + self, + publication_provider: _PublicationProvider | None = None, + ) -> None: self.provider = _AutomationProvider() + self.publication_provider = publication_provider def has_capability(self, name: str) -> bool: - return name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER + return name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER or ( + name == CAPABILITY_DATASOURCE_PUBLICATION + and self.publication_provider is not None + ) def capability(self, name: str): if not self.has_capability(name): raise KeyError(name) + if name == CAPABILITY_DATASOURCE_PUBLICATION: + return self.publication_provider return self.provider @@ -111,6 +180,9 @@ class DataflowRunWorkerTests(unittest.TestCase): Base.metadata.create_all( self.engine, tables=[ + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, DataflowPipeline.__table__, DataflowPipelineRevision.__table__, DataflowRun.__table__, @@ -119,6 +191,7 @@ class DataflowRunWorkerTests(unittest.TestCase): ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() + bind_process_runtime_identity(_runtime_identity()) self.pipeline = create_pipeline( self.session, tenant_id="tenant-1", @@ -133,6 +206,7 @@ class DataflowRunWorkerTests(unittest.TestCase): self.session.commit() def tearDown(self) -> None: + bind_process_runtime_identity(None) self.session.close() Base.metadata.drop_all( self.engine, @@ -141,11 +215,19 @@ class DataflowRunWorkerTests(unittest.TestCase): DataflowRun.__table__, DataflowPipelineRevision.__table__, DataflowPipeline.__table__, + RecoveryCheckpoint.__table__, + RecoveryOperation.__table__, + DistributedLease.__table__, ], ) self.engine.dispose() - def _queue(self, key: str) -> DataflowRun: + def _queue( + self, + key: str, + *, + publication: bool = False, + ) -> DataflowRun: run, replayed = start_pipeline_run( self.session, tenant_id="tenant-1", @@ -156,6 +238,14 @@ class DataflowRunWorkerTests(unittest.TestCase): pipeline_ref=f"pipeline:{self.pipeline.id}", revision=1, idempotency_key=key, + publication=( + DataflowPublicationTarget( + name="Worker output", + source_name="worker_output", + ) + if publication + else None + ), ), defer_execution=True, ) @@ -238,6 +328,97 @@ class DataflowRunWorkerTests(unittest.TestCase): self.assertEqual(2, run.output_row_count) self.assertIsNotNone(run.purged_at) + def test_stale_prepublication_attempt_is_safely_retried(self) -> None: + provider = _PublicationProvider() + registry = _Registry(provider) + run = self._queue("stale-before-publication", publication=True) + run.status = "running" + run.attempts = 1 + run.worker_id = "lost-worker" + run.lease_expires_at = utcnow() + timedelta(minutes=5) + self.session.commit() + begin_dataflow_run_recovery( + self.session, + run=run, + lease_ttl_seconds=120, + ) + self.session.commit() + lease = self.session.scalar( + select(DistributedLease).where( + DistributedLease.resource_key == f"dataflow:run:{run.id}" + ) + ) + assert lease is not None + lease.expires_at = utcnow() - timedelta(minutes=1) + run.lease_expires_at = utcnow() - timedelta(minutes=1) + self.session.commit() + + result = dispatch_pending_runs(self.session, registry=registry) + + self.session.refresh(run) + self.assertEqual(1, result["recovered"]) + self.assertEqual("succeeded", run.status) + self.assertEqual(2, run.attempts) + self.assertEqual(1, len(provider.requests)) + statuses = set( + self.session.scalars( + select(RecoveryOperation.status).where( + RecoveryOperation.resource_id == run.id + ) + ) + ) + self.assertEqual( + {RecoveryStatus.RECOVERED.value, RecoveryStatus.SUCCEEDED.value}, + statuses, + ) + + def test_stale_publication_attempt_is_not_retried_blindly(self) -> None: + provider = _PublicationProvider() + registry = _Registry(provider) + run = self._queue("stale-after-publication", publication=True) + run.status = "running" + run.attempts = 1 + run.worker_id = "lost-worker" + run.lease_expires_at = utcnow() + timedelta(minutes=5) + run.output_row_count = 1 + self.session.commit() + recovery = begin_dataflow_run_recovery( + self.session, + run=run, + lease_ttl_seconds=120, + ) + recovery.prepare_publication( + self.session, + run=run, + rows=({"id": 1},), + ) + lease = self.session.scalar( + select(DistributedLease).where( + DistributedLease.resource_key == f"dataflow:run:{run.id}" + ) + ) + assert lease is not None + lease.expires_at = utcnow() - timedelta(minutes=1) + run.lease_expires_at = utcnow() - timedelta(minutes=1) + self.session.commit() + + result = dispatch_pending_runs(self.session, registry=registry) + + self.session.refresh(run) + self.assertEqual(1, result["recovered"]) + self.assertEqual(1, result["outcome_unknown"]) + self.assertEqual(0, result["claimed"]) + self.assertEqual("outcome_unknown", run.status) + self.assertEqual(1, run.attempts) + self.assertEqual([], provider.requests) + operation = self.session.scalar( + select(RecoveryOperation).where( + RecoveryOperation.resource_id == run.id + ) + ) + assert operation is not None + self.assertEqual(RecoveryStatus.OUTCOME_UNKNOWN.value, operation.status) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_service.py b/tests/test_service.py index 49ea3c2..31fbc7c 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -17,6 +17,18 @@ from govoplan_core.core.datasources import ( DatasourceMaterialization, DatasourcePublicationResult, ) +from govoplan_core.core.recovery import ( + RecoveryCheckpoint, + RecoveryMode, + RecoveryOperation, + RecoveryStatus, + verify_recovery_evidence_chain, +) +from govoplan_core.core.runtime_coordination import ( + DistributedLease, + RuntimeIdentity, + bind_process_runtime_identity, +) from govoplan_core.db.base import Base from govoplan_dataflow.backend.backends.duckdb import DuckDbExecutionBackend from govoplan_dataflow.backend.db.models import ( @@ -99,6 +111,17 @@ def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: ) +def runtime_identity() -> RuntimeIdentity: + return RuntimeIdentity( + installation_id="dataflow-service-tests", + node_id="service-node", + incarnation="service-incarnation", + role="api", + software_version="test", + composition_hash="a" * 64, + ) + + class FakePublicationProvider: def __init__(self) -> None: self.requests = [] @@ -128,6 +151,25 @@ class FakePublicationProvider: ) +class FailingPublicationProvider(FakePublicationProvider): + def publish_rows(self, _session, _principal, *, request): + self.requests.append(request) + raise RuntimeError("connection closed after dispatch") + + +class TamperingPublicationProvider(FakePublicationProvider): + def publish_rows(self, session, principal, *, request): + checkpoint = session.scalar( + select(RecoveryCheckpoint) + .order_by(RecoveryCheckpoint.sequence) + .limit(1) + ) + assert checkpoint is not None + checkpoint.summary = "tampered" + session.commit() + return super().publish_rows(session, principal, request=request) + + class FakeRegistry: def __init__(self, publication_provider: FakePublicationProvider) -> None: self.publication_provider = publication_provider @@ -159,6 +201,9 @@ class DataflowServiceTests(unittest.TestCase): Base.metadata.create_all( self.engine, tables=[ + DistributedLease.__table__, + RecoveryOperation.__table__, + RecoveryCheckpoint.__table__, DataflowPipeline.__table__, DataflowPipelineRevision.__table__, DataflowRun.__table__, @@ -166,8 +211,10 @@ class DataflowServiceTests(unittest.TestCase): ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() + bind_process_runtime_identity(runtime_identity()) def tearDown(self) -> None: + bind_process_runtime_identity(None) self.session.close() Base.metadata.drop_all( self.engine, @@ -175,6 +222,9 @@ class DataflowServiceTests(unittest.TestCase): DataflowRun.__table__, DataflowPipelineRevision.__table__, DataflowPipeline.__table__, + RecoveryCheckpoint.__table__, + RecoveryOperation.__table__, + DistributedLease.__table__, ], ) self.engine.dispose() @@ -435,6 +485,26 @@ class DataflowServiceTests(unittest.TestCase): list(publication_provider.requests[0].rows), ) self.assertEqual(1, len(publication_provider.requests)) + operation = self.session.scalar( + select(RecoveryOperation).where( + RecoveryOperation.resource_id == first.id + ) + ) + assert operation is not None + self.assertEqual(RecoveryMode.FORWARD_RECOVERY.value, operation.mode) + self.assertEqual(RecoveryStatus.SUCCEEDED.value, operation.status) + self.assertTrue( + verify_recovery_evidence_chain(self.session, operation.id) + ) + checkpoint_kinds = list( + self.session.scalars( + select(RecoveryCheckpoint.kind) + .where(RecoveryCheckpoint.operation_id == operation.id) + .order_by(RecoveryCheckpoint.sequence) + ) + ) + self.assertIn("output-publication-dispatch", checkpoint_kinds) + self.assertIn("verified-success", checkpoint_kinds) def test_run_idempotency_key_rejects_changed_parameters(self) -> None: pipeline = self._create() @@ -490,6 +560,96 @@ class DataflowServiceTests(unittest.TestCase): self.assertEqual("failed", run.status) self.assertIn("Datasources publication capability", run.error) + operation = self.session.scalar( + select(RecoveryOperation).where( + RecoveryOperation.resource_id == run.id + ) + ) + assert operation is not None + self.assertEqual(RecoveryStatus.RECOVERED.value, operation.status) + + def test_provider_failure_after_dispatch_requires_reconciliation(self) -> None: + pipeline = self._create() + provider = FailingPublicationProvider() + request = DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="uncertain-publication", + publication=DataflowPublicationTarget( + name="Uncertain output", + source_name="uncertain_output", + ), + ) + + run, replayed = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(provider), + request=request, + ) + replay, second_replayed = start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(provider), + request=request, + ) + + self.assertFalse(replayed) + self.assertTrue(second_replayed) + self.assertEqual(run.id, replay.id) + self.assertEqual("outcome_unknown", run.status) + self.assertEqual(1, len(provider.requests)) + operation = self.session.scalar( + select(RecoveryOperation).where( + RecoveryOperation.resource_id == run.id + ) + ) + assert operation is not None + self.assertEqual(RecoveryStatus.OUTCOME_UNKNOWN.value, operation.status) + + def test_tampered_recovery_chain_prevents_verified_publication(self) -> None: + pipeline = self._create() + provider = TamperingPublicationProvider() + + with self.assertRaises(DataflowConflictError): + start_pipeline_run( + self.session, + tenant_id="tenant-1", + actor_id="user-1", + principal=principal(), + registry=FakeRegistry(provider), + request=DataflowRunRequest( + pipeline_ref=f"pipeline:{pipeline.id}", + revision=1, + idempotency_key="tampered-publication", + publication=DataflowPublicationTarget( + name="Tampered output", + source_name="tampered_output", + ), + ), + ) + + run = self.session.scalar( + select(DataflowRun).where( + DataflowRun.idempotency_key == "tampered-publication" + ) + ) + assert run is not None + operation = self.session.scalar( + select(RecoveryOperation).where( + RecoveryOperation.resource_id == run.id + ) + ) + assert operation is not None + self.assertEqual("outcome_unknown", run.status) + self.assertIsNone(run.output_publication_ref) + self.assertFalse( + verify_recovery_evidence_chain(self.session, operation.id) + ) if __name__ == "__main__": diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index f63d25b..b1a83ee 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -222,7 +222,7 @@ export type PipelineRun = { pipeline_id: string; revision: number; run_type: string; - status: "queued" | "retrying" | "running" | "succeeded" | "failed" | "cancelled"; + status: "queued" | "retrying" | "running" | "succeeded" | "failed" | "cancelled" | "outcome_unknown"; idempotency_key?: string | null; execution_backend: "auto" | "reference" | "duckdb"; environment: "development" | "staging" | "production"; @@ -256,6 +256,12 @@ export type PipelineRun = { finished_at?: string | null; created_by?: string | null; created_at: string; + recovery_operation_id?: string | null; + recovery_operation_type?: string | null; + recovery_mode?: string | null; + recovery_status?: string | null; + recovery_requires_attention: boolean; + recovery_explanation?: string | null; replayed: boolean; }; diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 1e6e8e7..365409b 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -1653,6 +1653,7 @@ function RunPipelineDialog({ const hasActiveRuns = runs.some((run) => run.status === "queued" || run.status === "retrying" || run.status === "running" ); + const recoveryAttentionRuns = runs.filter((run) => run.recovery_requires_attention); useEffect(() => { if (!open || !pipeline || !hasActiveRuns) return; @@ -1877,6 +1878,15 @@ function RunPipelineDialog({ Recent runs {loadingRuns ? "Loading..." : `${runs.length} shown`} + {recoveryAttentionRuns.length ? ( + `${run.ref}:${run.recovery_status}`).join("|")} + dismissible={false} + > + {recoveryAttentionRuns.length} run(s) require output reconciliation. Verify the recorded sink result before retrying. + + ) : null}
@@ -1885,6 +1895,7 @@ function RunPipelineDialog({ + + ))} {!loadingRuns && !runs.length ? ( - + ) : null}
Revision Status ProgressRecovery Rows Output @@ -1897,6 +1908,19 @@ function RunPipelineDialog({ {run.revision} {run.progress_percent}% ยท {run.progress_phase} + {run.recovery_status ? ( + <> + + {run.recovery_explanation ? {run.recovery_explanation} : null} + + ) : ( + Not recorded + )} + {run.output_row_count.toLocaleString()} {run.output_datasource_ref ?? "Run evidence"} @@ -1916,7 +1940,7 @@ function RunPipelineDialog({
No runs yet
No runs yet
diff --git a/webui/src/styles/dataflow.css b/webui/src/styles/dataflow.css index ebda6ed..a7e25e8 100644 --- a/webui/src/styles/dataflow.css +++ b/webui/src/styles/dataflow.css @@ -832,6 +832,18 @@ background: var(--primary-soft); } +.dataflow-run-history td.dataflow-run-recovery { + min-width: 180px; + white-space: normal; +} + +.dataflow-run-recovery small { + display: block; + margin-top: 4px; + color: var(--muted); + line-height: 1.35; +} + .dataflow-source-dialog-fields { display: grid; grid-template-columns: 1fr 1fr;