Python worker never sends heartbeats, so anti-entropy force-fails every task running longer than ~330s

[ethan-tyler]

Severity: Medium · Category: reliability

Affected code

• python/arco/src/arco_flow/worker/server.py:244-311
• crates/arco-flow/src/orchestration/controllers/anti_entropy.rs:230
• crates/arco-flow/src/orchestration/controllers/anti_entropy.rs:342-348
• crates/arco-flow/src/bin/arco_flow_sweeper.rs:293-322
• crates/arco-flow/src/orchestration/controllers/run_request_processor.rs:38
• python/arco/src/arco_flow/client.py:350-400

Problem

DispatchWorker.handle_dispatch only issues three calls for a task's whole lifetime: self._client.task_started(...) (line 248), self._client.task_completed(...) (line 285), and self._client.upload_logs(...) (line 301). It never calls ArcoFlowApiClient.task_heartbeat — that method exists at python/arco/src/arco_flow/client.py:350 and is exercised only by python/arco/tests/unit/test_worker_heartbeat_contract.py; grep -rn heartbeat python/arco/src/arco_flow/worker/ returns nothing. Meanwhile the orchestrator plans every task with heartbeat_timeout_sec: self.default_heartbeat_timeout_sec where default_heartbeat_timeout_sec: 300 (crates/arco-flow/src/orchestration/controllers/run_request_processor.rs:38,138), and the sweeper's anti-entropy controller treats a RUNNING task as dead using started_at when no heartbeat ever arrived:

fn running_task_is_stale(task: &TaskRow, now: DateTime<Utc>) -> bool {
    let Some(reference_time) = task.last_heartbeat_at.or(task.started_at) else { return false; };
    let timeout = Duration::seconds(i64::from(task.heartbeat_timeout_sec));
    let grace = Duration::seconds(30);
    now - reference_time > timeout + grace
}

(crates/arco-flow/src/orchestration/controllers/anti_entropy.rs:342-348, reached via TaskState::Running => Self::check_running_task(task, now) at line 230). The sweeper then appends a real TaskFinished { outcome: TaskOutcome::Failed, worker_id: "anti-entropy", error_message: Some("heartbeat_timeout_anti_entropy") } event (crates/arco-flow/src/bin/arco_flow_sweeper.rs:293-322). The HeartbeatHandler timer path does degrade gracefully (no_heartbeat_yet → reschedule, timer_handlers.rs:134-139), but anti-entropy does not — it is the started_at fallback that bites.

Why it matters

The only worker runtime Arco ships cannot execute any task that takes more than 5.5 minutes. Every such task is marked FAILED by the control plane while the worker is still computing, then retried (default_max_attempts: 3, run_request_processor.rs:37), producing a second concurrent execution of the same user asset. For a data orchestrator, a hard 5.5-minute ceiling on task duration with silent duplicate execution is a fundamental correctness/availability defect, not a tuning issue.

Failure scenario

Trigger a run whose asset function takes 7 minutes (a normal ETL). t=0 worker POSTs /tasks/{id}/started → task folds to RUNNING with started_at=t0. t=330s the scheduled sweeper (infra/terraform/cloud_run_flow.tf:226-246 Cloud Scheduler job hitting /run) evaluates check_running_task, running_task_is_stale returns true (330 > 300+30), and emits TaskFinished/Failed. Retry policy then creates attempt 2 with a fresh attempt_id and re-dispatches. At t=420s the original worker thread finishes and POSTs /tasks/{id}/completed with attempt_id of attempt 1; handle_task_completed returns 409 Conflict (state.is_terminal(), handlers.rs:630-635, or attempt_id_mismatch at 645-653). The ApiError raised by _raise_for_status propagates out of the finally block in handle_dispatch (it is NOT caught — only upload_logs is wrapped, lines 309-310), do_POST returns 500 (server.py:412-416), and Cloud Tasks re-delivers, executing the asset a third time.

Suggested fix

Start a daemon heartbeat thread in handle_dispatch before executing the asset and stop it in the finally block, posting self._client.task_heartbeat(...) at roughly heartbeat_timeout_sec / 3. The envelope must also carry heartbeat_timeout_sec (add it to WorkerDispatchEnvelope in crates/arco-worker-contract/src/lib.rs:105-140 and populate it in crates/arco-flow/src/bin/arco_flow_dispatcher.rs:208-224) so the worker can pick a safe interval. Also honor the shouldCancel field the heartbeat response already returns (client.py:393-400). Separately, wrap the task_completed call so a 409/410 does not turn into an HTTP 500 that triggers Cloud Tasks re-delivery.

Verification notes

Interaction with other defects: (1) Failed→RetryWait is non-terminal (fold.rs:2141-2148), so run counters (fold.rs:2199) and downstream failure propagation (fold.rs:2256) are never triggered. (2) No component creates a Retry timer, so retry_not_before stays None and anti_entropy.rs:312 never bootstraps a second dispatch; ready_dispatch.rs:186 only dispatches READY. (3) callbacks/handlers.rs:630-653 accepts the late task_completed from RETRY_WAIT (not terminal, attempt/attempt_id unchanged), folding the task to SUCCEEDED — the state error self-heals. (4) The deployed worker image is the Rust smoke worker, which completes in seconds. (5) Sweeper skips entirely when compaction lag >5 min (anti_entropy.rs:150-163).

Found during a staff-level audit of origin/main @ c3c0867. Mechanism confirmed by an independent adversarial verifier. Note this interacts with the RetryWait and zombie-reaper issues filed alongside it.