diff --git a/deploy/continuous-sync/README.md b/deploy/continuous-sync/README.md index 204a0c0912..4be3c25388 100644 --- a/deploy/continuous-sync/README.md +++ b/deploy/continuous-sync/README.md @@ -39,6 +39,33 @@ v2 stack. The same commit may be tested repeatedly. That is intentional: the fleet is a continuous sync canary, not a once-per-SHA CI job. +For Zakura and dual-stack nodes, the controller distinguishes an idle chain from +a stalled node. It compares the exact committed block height with the exact +local header-chain height. A header height at or below the committed height +means that the node has no local header backlog, so a long Mainnet block +interval does not start the stall deadline. A higher header height starts the +deadline because the node has a local backlog that it can process. New +committed progress restarts the deadline. + +A dual-stack node that hands block sync to legacy fallback stops the run +immediately, naming the handoff as the reason. The canary exists to catch v2 +stalls before legacy masks them, so an active fallback is the failure it is +looking for, not a recovery window to wait out. Measuring legacy progress after +the handoff would report a healthy node while the v2 stack stayed stalled. + +The controller records a metrics error or missing exact height as unavailable +status evidence. It does not classify that sample as a sync stall. Continuous +status unavailability has its own deadline and failure reason. Wall-clock tip +estimates remain available for diagnostics, but they do not supply stall +evidence. + +The legacy-only node does not run the Zakura header chain. It retains its +1800-second height-only deadline. The cluster monitor independently requires a +healthy peer with an exact committed height to advance before it reports a local +sync stall. It never uses a wall-clock estimate as peer evidence. That peer +evidence covers a header-sync failure where the local node never learns the +newer header. + ## Failure Semantics Any build, install, cleanup, startup, sync, stall, timeout, disk, metrics, or diff --git a/deploy/continuous-sync/alert-monitor.py b/deploy/continuous-sync/alert-monitor.py index 5dde2b827b..0aed28e696 100644 --- a/deploy/continuous-sync/alert-monitor.py +++ b/deploy/continuous-sync/alert-monitor.py @@ -299,16 +299,17 @@ def update_progress_state(state: dict[str, Any], statuses: list[dict[str, Any]], record["controller_run"] = run_id height = status.get("height") - if height is None: + if height is None or status.get("height_is_exact") is not True: continue previous_height = record.get("height") record["height"] = height - if ( - record.pop("progress_reset_pending", False) - or previous_height is None - or height > previous_height - ): + reset = record.pop("progress_reset_pending", False) + if reset or previous_height is None: record["last_progress"] = ts + record.pop("last_advance", None) + elif height > previous_height: + record["last_progress"] = ts + record["last_advance"] = ts else: record.setdefault("last_progress", ts) @@ -459,16 +460,24 @@ def run_once(config: dict[str, Any]) -> int: last_progress = int(record.get("last_progress", ts)) age = ts - last_progress peer_evidence = [] - if node_healthy(local_status) and height is not None: + if ( + node_healthy(local_status) + and height is not None + and local_status.get("height_is_exact") is True + ): for peer in statuses: - if peer["hostname"] == local_host or not node_healthy(peer): + if ( + peer["hostname"] == local_host + or not node_healthy(peer) + or peer.get("height_is_exact") is not True + ): continue peer_height = peer.get("height") peer_record = state.get("nodes", {}).get(peer["hostname"], {}) if ( peer_height is not None and peer_height > height - and int(peer_record.get("last_progress", 0)) > last_progress + and int(peer_record.get("last_advance", 0)) > last_progress ): peer_evidence.append(f"{peer['hostname']} advanced to height {peer_height}") stalled = ( @@ -490,6 +499,7 @@ def run_once(config: dict[str, Any]) -> int: local_progressed = ( isinstance(height, int) and isinstance(previous_local_height, int) + and local_status.get("height_is_exact") is True and height > previous_local_height ) if controller_owns_lifecycle: diff --git a/deploy/continuous-sync/alert-status.py b/deploy/continuous-sync/alert-status.py index 959cd5f058..eab756ff31 100644 --- a/deploy/continuous-sync/alert-status.py +++ b/deploy/continuous-sync/alert-status.py @@ -52,23 +52,20 @@ def service_active(service: str) -> bool: return active_state in {"active", "reloading", "refreshing"} -def metric_height(text: str) -> int | None: - # Prefer finalized/verified block progress over header-only metrics. - priority = [ +def metric_height_observation(text: str) -> tuple[int | None, str | None]: + # Finalized and verifier-only gauges can trail or lead the best committed + # tip, so they cannot prove node progress. + exact = [ "state_memory_best_committed_block_height", "state_memory_committed_block_height", - "state_finalized_block_height", - "state_checkpoint_finalized_block_height", "zcash_chain_verified_block_height", "sync_block_verified_tip_height", - "checkpoint_verified_height", - "checkpoint_processing_next_height", ] estimated = [ "sync_estimated_network_tip_height", "sync_estimated_distance_to_tip", ] - values = {name: [] for name in priority + estimated} + values = {name: [] for name in exact + estimated} for line in text.splitlines(): if not line or line.startswith("#"): continue @@ -83,15 +80,19 @@ def metric_height(text: str) -> int | None: values[dotted_base].append(int(float(parts[1]))) except ValueError: continue - for name in priority: + for name in exact: if values[name]: - return max(values[name]) + return max(values[name]), name if all(values[name] for name in estimated): tip = max(values["sync_estimated_network_tip_height"]) distance = min(values["sync_estimated_distance_to_tip"]) if 0 <= distance <= tip: - return tip - distance - return None + return tip - distance, "estimated_tip_minus_distance" + return None, None + + +def metric_height(text: str) -> int | None: + return metric_height_observation(text)[0] def node_info(config: dict[str, Any], hostname: str) -> dict[str, Any]: @@ -131,11 +132,12 @@ def status(config: dict[str, Any]) -> dict[str, Any]: metrics_status = "unavailable" height = None + height_source = None try: with urllib.request.urlopen(metrics_url, timeout=METRICS_TIMEOUT_SECONDS) as response: metrics = response.read().decode("utf-8", "replace") metrics_status = "ok" - height = metric_height(metrics) + height, height_source = metric_height_observation(metrics) except Exception as exc: metrics_status = f"unavailable: {type(exc).__name__}" @@ -147,6 +149,8 @@ def status(config: dict[str, Any]) -> dict[str, Any]: "service_active": service_active(service), "metrics_status": metrics_status, "height": height, + "height_source": height_source, + "height_is_exact": height_source not in (None, "estimated_tip_minus_distance"), "controller_state": controller_state(controller_state_path), "connection": node.get("ssh_string", f"root@{node.get('public_ip', 'unknown')}"), "alias_connection": f"ssh {node.get('alias', hostname)}", diff --git a/deploy/continuous-sync/continuous-sync.py b/deploy/continuous-sync/continuous-sync.py index 2791719282..c79e92306d 100644 --- a/deploy/continuous-sync/continuous-sync.py +++ b/deploy/continuous-sync/continuous-sync.py @@ -27,6 +27,39 @@ STATE_VERSION = 1 +# These gauges track the best committed tip. Finalized and verifier-only gauges +# can trail or lead that tip, so they remain diagnostics. +COMMITTED_HEIGHT_METRICS = ( + "state.memory.best.committed.block.height", + "state.memory.committed.block.height", + "zcash_chain_verified_block_height", + "sync.block.verified_tip.height", +) + +HEADER_HEIGHT_METRICS = ( + "sync.header_chain.frontier.header_best_height", + "sync.block.best_header_tip.height", +) + +LEGACY_FALLBACK_ACTIVE_METRIC = "sync.zakura.legacy_fallback.active" + +DIAGNOSTIC_METRICS = ( + "state_finalized_block_height", + "state_checkpoint_finalized_block_height", + "checkpoint_verified_height", + "checkpoint_processing_next_height", + "sync.estimated_network_tip_height", + "sync.estimated_distance_to_tip", + "sync.prospective_tips.len", + "sync.reserve.depth", + "sync.downloads.in_flight", + "sync.downloads.waiting_network", + "sync.downloads.downloading", + "sync.downloads.response_received", + "sync.downloads.waiting_verifier", + "sync.downloads.verifying", +) + class ControllerError(Exception): """Operator-facing failure that should halt the sync loop.""" @@ -65,6 +98,7 @@ class Policy: poll_interval_seconds: int = 30 startup_timeout_seconds: int = 600 stall_seconds: int = 600 + status_unavailable_seconds: int = 600 max_run_seconds: int = 172800 ready_samples: int = 6 ready_sample_interval_seconds: int = 30 @@ -83,6 +117,88 @@ class Config: policy: Policy = field(default_factory=Policy) +@dataclass +class SyncProgress: + started_at: int + last_height: int | None = None + highest_height: int | None = None + last_progress_at: int = field(init=False) + backlog_since: int | None = None + status_unavailable_since: int | None = None + + def __post_init__(self) -> None: + self.last_progress_at = self.started_at + + def observe( + self, + sample: dict[str, Any], + policy: Policy, + observed_at: int, + ) -> tuple[bool, str | None]: + committed_height = sample.get("committed_height") + progressed = False + if isinstance(committed_height, int): + self.last_height = committed_height + if self.highest_height is None or committed_height > self.highest_height: + self.highest_height = committed_height + self.last_progress_at = observed_at + progressed = True + + evidence, detail = classify_sync_evidence(sample, policy.p2p_stack) + sample["stall_evidence"] = evidence + sample["stall_evidence_detail"] = detail + + if evidence == "local_header_backlog": + if self.status_unavailable_since is not None and self.backlog_since is not None: + self.backlog_since += observed_at - self.status_unavailable_since + self.status_unavailable_since = None + if progressed or self.backlog_since is None: + self.backlog_since = observed_at + stalled_for = observed_at - self.backlog_since + if stalled_for >= policy.stall_seconds: + return progressed, ( + f"committed height {self.last_height} has not progressed while " + f"{detail} for {stalled_for}s (threshold {policy.stall_seconds}s)" + ) + return progressed, None + + if evidence == "no_local_header_backlog": + self.backlog_since = None + self.status_unavailable_since = None + return progressed, None + + if evidence == "legacy_fallback": + self.backlog_since = None + self.status_unavailable_since = None + return progressed, ( + f"Zakura block sync handed off to legacy fallback at committed height " + f"{self.last_height}; {detail}" + ) + + if evidence == "legacy_height_only": + self.backlog_since = None + self.status_unavailable_since = None + stalled_for = observed_at - self.last_progress_at + if self.last_height is not None and stalled_for >= policy.stall_seconds: + return progressed, ( + f"legacy committed height {self.last_height} has not progressed for " + f"{stalled_for}s (threshold {policy.stall_seconds}s)" + ) + return progressed, None + + if progressed: + self.backlog_since = None + if self.status_unavailable_since is None: + self.status_unavailable_since = observed_at + unavailable_for = observed_at - self.status_unavailable_since + if unavailable_for >= policy.status_unavailable_seconds: + return progressed, ( + f"sync status evidence unavailable for {unavailable_for}s " + f"(threshold {policy.status_unavailable_seconds}s): {detail}" + ) + return progressed, None + + def now() -> int: return int(time.time()) @@ -448,49 +564,77 @@ def metric_value(metrics: str, name: str) -> float | None: return float(match.group(1)) if match else None +def first_metric(status: dict[str, Any], names: tuple[str, ...]) -> tuple[int | None, str | None]: + for name in names: + value = status.get(name) + if isinstance(value, int): + return value, name + return None, None + + +def classify_sync_evidence(sample: dict[str, Any], p2p_stack: str) -> tuple[str, str]: + metrics_status = sample.get("metrics_status") + if metrics_status != "ok": + return "unknown", f"metrics={metrics_status}" + + committed_height = sample.get("committed_height") + if not isinstance(committed_height, int): + return "unknown", "committed block height is missing" + + if p2p_stack in ("legacy", "zebra"): + return "legacy_height_only", "Zakura header state is disabled" + + if sample.get(LEGACY_FALLBACK_ACTIVE_METRIC) == 1: + return "legacy_fallback", "legacy fallback is the active block-sync driver" + + header_height = sample.get("header_height") + if not isinstance(header_height, int): + return "unknown", "authoritative local header height is missing" + if header_height < committed_height: + return ( + "no_local_header_backlog", + f"local header height {header_height} has no backlog above committed height " + f"{committed_height}", + ) + if header_height == committed_height: + return ( + "no_local_header_backlog", + f"local header height equals committed height {committed_height}", + ) + return ( + "local_header_backlog", + f"local header height {header_height} is ahead of committed height {committed_height}", + ) + + def sample_status(config: Config) -> dict[str, Any]: status: dict[str, Any] = {"service_active": service_active(config)} try: metrics = fetch_text(config.policy.metrics_url) status["metrics_status"] = "ok" for key in ( - "state.memory.best.committed.block.height", - "state.memory.committed.block.height", - "state_finalized_block_height", - "state_checkpoint_finalized_block_height", - "zcash_chain_verified_block_height", - "sync_block_verified_tip_height", - "checkpoint_verified_height", - "checkpoint_processing_next_height", - "sync.estimated_network_tip_height", - "sync.estimated_distance_to_tip", - "sync.prospective_tips.len", - "sync.reserve.depth", - "sync.downloads.in_flight", - "sync.downloads.waiting_network", - "sync.downloads.downloading", - "sync.downloads.response_received", - "sync.downloads.waiting_verifier", - "sync.downloads.verifying", + *COMMITTED_HEIGHT_METRICS, + *HEADER_HEIGHT_METRICS, + LEGACY_FALLBACK_ACTIVE_METRIC, + *DIAGNOSTIC_METRICS, ): value = metric_value(metrics, key) if value is not None: status[key] = int(value) - status["height"] = None - for key in ( - "state.memory.best.committed.block.height", - "state.memory.committed.block.height", - "state_finalized_block_height", - "state_checkpoint_finalized_block_height", - "zcash_chain_verified_block_height", - "sync_block_verified_tip_height", - "checkpoint_verified_height", - "checkpoint_processing_next_height", - ): - if status.get(key) is not None: - status["height"] = status[key] - status["height_source"] = key - break + + committed_height, committed_source = first_metric(status, COMMITTED_HEIGHT_METRICS) + status["committed_height"] = committed_height + if committed_source is not None: + status["committed_height_source"] = committed_source + + header_height, header_source = first_metric(status, HEADER_HEIGHT_METRICS) + status["header_height"] = header_height + if header_source is not None: + status["header_height_source"] = header_source + + status["height"] = committed_height + if committed_source is not None: + status["height_source"] = committed_source if status["height"] is None: tip = status.get("sync.estimated_network_tip_height") distance = status.get("sync.estimated_distance_to_tip") @@ -503,13 +647,15 @@ def sample_status(config: Config) -> dict[str, Any]: ready, ready_detail = fetch_ready(config) status["ready"] = ready status["ready_detail"] = ready_detail + evidence, detail = classify_sync_evidence(status, config.policy.p2p_stack) + status["stall_evidence"] = evidence + status["stall_evidence_detail"] = detail return status def wait_for_completion(config: Config, run_dir: Path, run_state: dict[str, Any]) -> None: started = now() - last_height: int | None = None - last_progress = started + progress = SyncProgress(started) ready_samples = 0 samples_path = run_dir / "samples.jsonl" @@ -523,29 +669,24 @@ def wait_for_completion(config: Config, run_dir: Path, run_state: dict[str, Any] sample = sample_status(config) sample["time"] = utc_stamp(ts) + progressed, failure = progress.observe(sample, config.policy, ts) with samples_path.open("a", encoding="utf-8") as samples: samples.write(json.dumps(sample, sort_keys=True) + "\n") - height = sample.get("height") - if isinstance(height, int) and height != last_height: - last_height = height - last_progress = ts - run_state["height"] = height + if progressed: + run_state["height"] = progress.last_height run_state["last_progress_at"] = utc_stamp(ts) write_run_json(run_dir, run_state) - if last_height is None and ts - started >= config.policy.startup_timeout_seconds: + if progress.last_height is None and ts - started >= config.policy.startup_timeout_seconds: raise ControllerError( - f"no height observed within startup timeout " + f"no committed height observed within startup timeout " f"{config.policy.startup_timeout_seconds}s; metrics={sample.get('metrics_status')}, " f"ready={sample.get('ready_detail')}" ) - if ts - last_progress >= config.policy.stall_seconds: - raise ControllerError( - f"height {last_height} has not progressed for {ts - last_progress}s " - f"(threshold {config.policy.stall_seconds}s)" - ) + if failure is not None: + raise ControllerError(failure) if sample.get("ready") is True: ready_samples += 1 diff --git a/deploy/continuous-sync/deploy.py b/deploy/continuous-sync/deploy.py index f2f6a8bd6f..0c63e86b85 100644 --- a/deploy/continuous-sync/deploy.py +++ b/deploy/continuous-sync/deploy.py @@ -160,6 +160,7 @@ def subst_for(node: Node) -> dict[str, str]: "POLL_INTERVAL_SECONDS": str(raw["poll_interval_seconds"]), "STARTUP_TIMEOUT_SECONDS": str(raw["startup_timeout_seconds"]), "STALL_SECONDS": str(raw["stall_seconds"]), + "STATUS_UNAVAILABLE_SECONDS": str(raw["status_unavailable_seconds"]), "MAX_RUN_SECONDS": str(raw["max_run_seconds"]), "READY_SAMPLES": str(raw["ready_samples"]), "READY_SAMPLE_INTERVAL_SECONDS": str(raw["ready_sample_interval_seconds"]), diff --git a/deploy/continuous-sync/nodes.toml b/deploy/continuous-sync/nodes.toml index e0b71742d3..c0b2a7c37c 100644 --- a/deploy/continuous-sync/nodes.toml +++ b/deploy/continuous-sync/nodes.toml @@ -30,6 +30,7 @@ healthy_url = "http://127.0.0.1:8080/healthy" poll_interval_seconds = 30 startup_timeout_seconds = 600 stall_seconds = 600 +status_unavailable_seconds = 600 max_run_seconds = 172800 ready_samples = 6 ready_sample_interval_seconds = 30 diff --git a/deploy/continuous-sync/templates/controller.toml b/deploy/continuous-sync/templates/controller.toml index 4cfe5dec07..9430f6a2e3 100644 --- a/deploy/continuous-sync/templates/controller.toml +++ b/deploy/continuous-sync/templates/controller.toml @@ -28,6 +28,7 @@ healthy_url = "{{HEALTHY_URL}}" poll_interval_seconds = {{POLL_INTERVAL_SECONDS}} startup_timeout_seconds = {{STARTUP_TIMEOUT_SECONDS}} stall_seconds = {{STALL_SECONDS}} +status_unavailable_seconds = {{STATUS_UNAVAILABLE_SECONDS}} max_run_seconds = {{MAX_RUN_SECONDS}} ready_samples = {{READY_SAMPLES}} ready_sample_interval_seconds = {{READY_SAMPLE_INTERVAL_SECONDS}} diff --git a/deploy/continuous-sync/tests/test_continuous_sync.py b/deploy/continuous-sync/tests/test_continuous_sync.py index 40e54079d1..ee26caa306 100644 --- a/deploy/continuous-sync/tests/test_continuous_sync.py +++ b/deploy/continuous-sync/tests/test_continuous_sync.py @@ -68,9 +68,226 @@ def test_sample_status_falls_back_to_estimated_height(self): self.assertEqual(status["height"], 900) self.assertEqual(status["height_source"], "estimated_tip_minus_distance") + self.assertIsNone(status["committed_height"]) + self.assertIsNone(status["header_height"]) self.assertEqual(status["sync.downloads.in_flight"], 17) self.assertEqual(status["sync.downloads.verifying"], 4) + def test_sample_status_reports_exact_committed_and_header_heights(self): + metrics = "\n".join( + [ + "state_memory_best_committed_block_height 42", + "sync_header_chain_frontier_header_best_height 45", + "sync_zakura_legacy_fallback_active 1", + "sync_estimated_network_tip_height 1000", + "sync_estimated_distance_to_tip 100", + ] + ) + config = make_config(Path("/tmp")) + + with ( + patch.object(sync, "service_active", return_value=True), + patch.object(sync, "fetch_text", return_value=metrics), + patch.object(sync, "fetch_ready", return_value=(False, "syncing")), + ): + status = sync.sample_status(config) + + self.assertEqual(status["committed_height"], 42) + self.assertEqual( + status["committed_height_source"], + "state.memory.best.committed.block.height", + ) + self.assertEqual(status["header_height"], 45) + self.assertEqual( + status["header_height_source"], + "sync.header_chain.frontier.header_best_height", + ) + self.assertEqual(status["height"], 42) + self.assertEqual(status["sync.zakura.legacy_fallback.active"], 1) + self.assertEqual(status["stall_evidence"], "legacy_fallback") + + def test_sample_status_classifies_metrics_timeout_as_unknown(self): + config = make_config(Path("/tmp")) + + with ( + patch.object(sync, "service_active", return_value=True), + patch.object(sync, "fetch_text", side_effect=TimeoutError("timed out")), + patch.object(sync, "fetch_ready", return_value=(False, "syncing")), + ): + status = sync.sample_status(config) + + self.assertEqual(status["metrics_status"], "TimeoutError: timed out") + self.assertEqual(status["stall_evidence"], "unknown") + self.assertIn("TimeoutError", status["stall_evidence_detail"]) + + def test_lagging_height_metrics_are_not_committed_tip_progress(self): + for metric in ( + "state_finalized_block_height", + "state_checkpoint_finalized_block_height", + "checkpoint_verified_height", + ): + with self.subTest(metric=metric): + config = make_config(Path("/tmp")) + with ( + patch.object(sync, "service_active", return_value=True), + patch.object(sync, "fetch_text", return_value=f"{metric} 42"), + patch.object(sync, "fetch_ready", return_value=(False, "syncing")), + ): + status = sync.sample_status(config) + + self.assertIsNone(status["committed_height"]) + self.assertEqual(status["stall_evidence"], "unknown") + + def test_natural_block_gap_does_not_create_stall_evidence(self): + policy = sync.Policy(p2p_stack="zakura", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + sample = exact_sync_sample(42, 42) + + self.assertEqual(progress.observe(sample, policy, 100), (True, None)) + self.assertEqual(progress.observe(sample, policy, 701), (False, None)) + self.assertEqual(sample["stall_evidence"], "no_local_header_backlog") + self.assertIsNone(progress.backlog_since) + + def test_committed_height_above_header_height_has_no_local_backlog(self): + policy = sync.Policy(p2p_stack="zakura", status_unavailable_seconds=600) + progress = sync.SyncProgress(started_at=100) + sample = exact_sync_sample(45, 42) + + self.assertEqual(progress.observe(sample, policy, 100), (True, None)) + self.assertEqual(progress.observe(sample, policy, 701), (False, None)) + self.assertEqual(sample["stall_evidence"], "no_local_header_backlog") + self.assertIsNone(progress.status_unavailable_since) + + def test_active_legacy_fallback_stops_the_run_immediately(self): + policy = sync.Policy(p2p_stack="dual", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + sample = exact_sync_sample(42, 45) + sample[sync.LEGACY_FALLBACK_ACTIVE_METRIC] = 1 + + progressed, failure = progress.observe(sample, policy, 100) + + self.assertTrue(progressed) + self.assertEqual(sample["stall_evidence"], "legacy_fallback") + self.assertIn("handed off to legacy fallback at committed height 42", failure) + + def test_legacy_fallback_stops_the_run_even_while_committed_height_advances(self): + policy = sync.Policy(p2p_stack="dual", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + + def fallback_sample(committed_height: int) -> dict[str, object]: + sample = exact_sync_sample(committed_height, 45) + sample[sync.LEGACY_FALLBACK_ACTIVE_METRIC] = 1 + return sample + + _, failure = progress.observe(fallback_sample(43), policy, 700) + + self.assertIn("handed off to legacy fallback", failure) + + def test_continuous_local_header_backlog_reaches_stall_deadline(self): + policy = sync.Policy(p2p_stack="zakura", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + + self.assertEqual(progress.observe(exact_sync_sample(42, 45), policy, 100), (True, None)) + self.assertEqual(progress.observe(exact_sync_sample(42, 45), policy, 699), (False, None)) + progressed, failure = progress.observe(exact_sync_sample(42, 45), policy, 700) + + self.assertFalse(progressed) + self.assertIn("local header height 45 is ahead", failure) + self.assertIn("for 600s", failure) + + def test_new_backlog_starts_a_fresh_deadline_after_a_long_block_gap(self): + policy = sync.Policy(p2p_stack="zakura", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + + progress.observe(exact_sync_sample(42, 42), policy, 100) + self.assertEqual(progress.observe(exact_sync_sample(42, 42), policy, 1000), (False, None)) + self.assertEqual(progress.observe(exact_sync_sample(42, 43), policy, 1001), (False, None)) + self.assertEqual(progress.observe(exact_sync_sample(42, 43), policy, 1600), (False, None)) + _, failure = progress.observe(exact_sync_sample(42, 43), policy, 1601) + + self.assertIn("for 600s", failure) + + def test_committed_progress_restarts_an_existing_backlog_deadline(self): + policy = sync.Policy(p2p_stack="zakura", stall_seconds=600) + progress = sync.SyncProgress(started_at=100) + + progress.observe(exact_sync_sample(42, 44), policy, 100) + self.assertEqual(progress.observe(exact_sync_sample(43, 44), policy, 600), (True, None)) + self.assertEqual(progress.observe(exact_sync_sample(43, 44), policy, 1199), (False, None)) + _, failure = progress.observe(exact_sync_sample(43, 44), policy, 1200) + + self.assertIn("for 600s", failure) + + def test_query_error_uses_a_separate_status_deadline(self): + policy = sync.Policy(p2p_stack="zakura", status_unavailable_seconds=600) + progress = sync.SyncProgress(started_at=100) + sample = {"metrics_status": "TimeoutError: timed out"} + + self.assertEqual(progress.observe(sample, policy, 100), (False, None)) + self.assertEqual(progress.observe(sample, policy, 699), (False, None)) + _, failure = progress.observe(sample, policy, 700) + + self.assertIn("sync status evidence unavailable for 600s", failure) + self.assertIn("TimeoutError", failure) + + def test_status_recovery_clears_the_unavailable_deadline(self): + policy = sync.Policy(p2p_stack="zakura", status_unavailable_seconds=600) + progress = sync.SyncProgress(started_at=100) + unavailable = {"metrics_status": "TimeoutError: timed out"} + + progress.observe(unavailable, policy, 100) + progress.observe(exact_sync_sample(42, 42), policy, 600) + progress.observe(unavailable, policy, 1000) + self.assertEqual(progress.observe(unavailable, policy, 1599), (False, None)) + _, failure = progress.observe(unavailable, policy, 1600) + + self.assertIn("sync status evidence unavailable for 600s", failure) + + def test_query_error_pauses_but_does_not_erase_the_backlog_deadline(self): + policy = sync.Policy( + p2p_stack="zakura", + stall_seconds=600, + status_unavailable_seconds=600, + ) + progress = sync.SyncProgress(started_at=100) + unavailable = {"metrics_status": "TimeoutError: timed out"} + + progress.observe(exact_sync_sample(42, 45), policy, 100) + progress.observe(unavailable, policy, 650) + self.assertEqual(progress.observe(exact_sync_sample(42, 45), policy, 700), (False, None)) + _, failure = progress.observe(exact_sync_sample(42, 45), policy, 750) + + self.assertIn("for 600s", failure) + + def test_estimated_tip_fields_do_not_supply_stall_evidence(self): + policy = sync.Policy(p2p_stack="zakura", status_unavailable_seconds=600) + progress = sync.SyncProgress(started_at=100) + sample = { + "metrics_status": "ok", + "height": 900, + "height_source": "estimated_tip_minus_distance", + "sync.estimated_network_tip_height": 1000, + "sync.estimated_distance_to_tip": 100, + } + + self.assertEqual(progress.observe(sample, policy, 100), (False, None)) + _, failure = progress.observe(sample, policy, 700) + + self.assertEqual(sample["stall_evidence"], "unknown") + self.assertIn("committed block height is missing", failure) + + def test_legacy_node_keeps_its_height_only_deadline(self): + policy = sync.Policy(p2p_stack="legacy", stall_seconds=1800) + progress = sync.SyncProgress(started_at=100) + sample = {"metrics_status": "ok", "committed_height": 42} + + self.assertEqual(progress.observe(sample, policy, 100), (True, None)) + self.assertEqual(progress.observe(sample, policy, 1899), (False, None)) + _, failure = progress.observe(sample, policy, 1900) + + self.assertIn("legacy committed height 42", failure) + self.assertIn("for 1800s", failure) + def test_alert_status_falls_back_to_estimated_height(self): metrics = "\n".join( [ @@ -80,6 +297,10 @@ def test_alert_status_falls_back_to_estimated_height(self): ) self.assertEqual(alert_status.metric_height(metrics), 900) + self.assertEqual( + alert_status.metric_height_observation(metrics), + (900, "estimated_tip_minus_distance"), + ) def test_alert_status_distinguishes_active_and_inactive_service(self): for active_state, expected in (("active", True), ("inactive", False), ("failed", False)): @@ -234,6 +455,8 @@ def test_deploy_renders_per_node_p2p_config(self): self.assertIn('p2p_stack = "zakura"', rendered["zakurad.toml.template"]) self.assertIn('mode_label = "Zakura/v2-only"', rendered["controller.toml"]) + self.assertIn("stall_seconds = 600", rendered["controller.toml"]) + self.assertIn("status_unavailable_seconds = 600", rendered["controller.toml"]) self.assertIn("[[nodes]]", rendered["alert-monitor.toml"]) self.assertIn('hostname = "temp-zakura-sync-test-1"', rendered["alert-monitor.toml"]) self.assertIn("zakura-monitor.py", rendered["zakura-monitor.service"]) @@ -241,6 +464,19 @@ def test_deploy_renders_per_node_p2p_config(self): self.assertIn("down_confirmation_samples = 2", rendered["alert-monitor.toml"]) self.assertIn("zakura.service", rendered) + def test_deploy_keeps_the_dual_stack_stall_deadline_at_the_fleet_default(self): + nodes = deploy.load_nodes( + ROOT / "deploy" / "continuous-sync" / "nodes.toml", + ["temp-zakura-sync-test-1"], + ) + rendered = deploy.render_files(nodes[0]) + + # The deadline deliberately matches the node's own 600-second fallback + # threshold: PR #732 established that catching a v2 stall before legacy + # takes over is the point of this canary. + self.assertIn('p2p_stack = "dual"', rendered["zakurad.toml.template"]) + self.assertIn("stall_seconds = 600", rendered["controller.toml"]) + def test_deploy_renders_expanded_legacy_alert_inventory(self): nodes = deploy.load_nodes( ROOT / "deploy" / "continuous-sync" / "nodes.toml", @@ -698,6 +934,36 @@ def test_stationary_higher_peer_does_not_prove_local_stall(self): post_alert.assert_not_called() + def test_estimated_height_does_not_supply_cluster_stall_evidence(self): + local = "temp-zakura-sync-test-1" + peer = "temp-zakura-sync-test-2" + statuses = { + local: alert_status_fixture(local, service_active=True, height=10), + peer: alert_status_fixture(peer, service_active=True, height=11), + } + statuses[peer]["height_is_exact"] = False + statuses[peer]["height_source"] = "estimated_tip_minus_distance" + with tempfile.TemporaryDirectory() as tmp: + config = alert_config(Path(tmp), [local, peer], cluster_stall_seconds=10) + with ( + patch.object(alert, "query_node", side_effect=lambda _, node: statuses[node["hostname"]]), + patch.object(alert.socket, "gethostname", return_value=local), + patch.object(alert, "now", side_effect=[100, 111, 112, 113]), + patch.object(alert, "post_alert", return_value=True) as post_alert, + ): + alert.run_once(config) + statuses[peer]["height"] = 12 + alert.run_once(config) + statuses[peer]["height_is_exact"] = True + statuses[peer]["height_source"] = "state_memory_best_committed_block_height" + alert.run_once(config) + post_alert.assert_not_called() + + statuses[peer]["height"] = 13 + alert.run_once(config) + + self.assertEqual([call.args[1] for call in post_alert.call_args_list], ["SYNC STALLED"]) + def test_regressing_higher_peer_does_not_prove_local_stall(self): local = "temp-zakura-sync-test-1" peer = "temp-zakura-sync-test-2" @@ -807,6 +1073,40 @@ def test_failed_stall_recovery_is_retried(self): state["alerts"][f"local-sync-stall:{local}"], ) + def test_estimated_local_progress_does_not_recover_a_stall_alert(self): + local = "temp-zakura-sync-test-1" + peer = "temp-zakura-sync-test-2" + statuses = { + local: alert_status_fixture(local, service_active=True, height=10), + peer: alert_status_fixture(peer, service_active=True, height=11), + } + with tempfile.TemporaryDirectory() as tmp: + config = alert_config(Path(tmp), [local, peer], cluster_stall_seconds=10) + with ( + patch.object(alert, "query_node", side_effect=lambda _, node: statuses[node["hostname"]]), + patch.object(alert.socket, "gethostname", return_value=local), + patch.object(alert, "now", side_effect=[100, 111, 112, 113]), + patch.object(alert, "post_alert", return_value=True) as post_alert, + ): + alert.run_once(config) + statuses[peer]["height"] = 12 + alert.run_once(config) + + statuses[local]["height"] = 11 + statuses[local]["height_is_exact"] = False + statuses[local]["height_source"] = "estimated_tip_minus_distance" + alert.run_once(config) + self.assertEqual(post_alert.call_count, 1) + + statuses[local]["height_is_exact"] = True + statuses[local]["height_source"] = "state_memory_best_committed_block_height" + alert.run_once(config) + + self.assertEqual( + [call.args[1] for call in post_alert.call_args_list], + ["SYNC STALLED", "SYNC RECOVERED"], + ) + def test_legacy_alert_state_migrates_without_recovery(self): hostname = "temp-zakura-sync-test-1" with tempfile.TemporaryDirectory() as tmp: @@ -1029,6 +1329,8 @@ def alert_status_fixture( "service_active": service_active, "metrics_status": "ok" if service_active else "unavailable", "height": height, + "height_source": "state_memory_best_committed_block_height" if height is not None else None, + "height_is_exact": height is not None, "connection": "root@138.68.43.212", "alias_connection": f"ssh {hostname}", "log_path": "/tmp/zebrad.log", @@ -1038,6 +1340,14 @@ def alert_status_fixture( } +def exact_sync_sample(committed_height: int, header_height: int) -> dict[str, object]: + return { + "metrics_status": "ok", + "committed_height": committed_height, + "header_height": header_height, + } + + def alert_config(tmp_path: Path, hostnames: list[str], **default_overrides): defaults = { "alert_state_file": str(tmp_path / "state.json"), diff --git a/docs/changelog/params.md b/docs/changelog/params.md index 0f05504baf..917c2f8902 100644 --- a/docs/changelog/params.md +++ b/docs/changelog/params.md @@ -28,6 +28,7 @@ Keep entries **newest-first**. Each row records: | Parameter | Location | Old → New | PR | Why | | --- | --- | --- | --- | --- | +| `status_unavailable_seconds` | `deploy/continuous-sync/continuous-sync.py` and `deploy/continuous-sync/nodes.toml` | new → `600 s` | [#846](https://github.com/zakura-core/zakura/pull/846) | Stop a canary after ten continuous minutes without exact sync evidence while allowing individual metrics errors and timeouts to recover. | | `LEGACY_FALLBACK_APPLY_DRAIN_DEADLINE` | `crates/zakurad/src/commands/start/zakura/coordinator.rs` | new → `30 min` | [#831](https://github.com/zakura-core/zakura/pull/831) | Terminate the node when native block applies prevent legacy fallback from acquiring exclusive ownership, instead of leaving the fallback handoff pending forever. | | `MAX_CANDIDATE_TIPS_V1` | `crates/zakura-header-chain/src/config.rs` | `10` → `11` | [#831](https://github.com/zakura-core/zakura/pull/831) | Retain ten full-state fork tips plus one independent selected header tip, so header candidate pressure cannot evict a branch that full state still owns. | | `VCT_LOCAL_OPERATION_FATAL_AFTER` | `crates/zakura-network/src/zakura/header_sync/reactor.rs` | new → `30 min` | [#821](https://github.com/zakura-core/zakura/pull/821) | Terminate a node whose local VCT repair prepare or apply operation remains pending, while allowing slow valid operations substantially more time than the existing one-minute stall diagnostic. | diff --git a/docs/changelog/unreleased/846.md b/docs/changelog/unreleased/846.md new file mode 100644 index 0000000000..9f0423af5a --- /dev/null +++ b/docs/changelog/unreleased/846.md @@ -0,0 +1,16 @@ +## Fixed + +- Prevented continuous-sync canaries from treating a valid Mainnet block gap + as a stalled node by requiring an exact local header backlog before applying + the stall deadline + ([#846](https://github.com/zakura-core/zakura/pull/846)). +- Reported continuous metrics failures as unavailable status evidence instead + of misclassifying them as sync stalls + ([#846](https://github.com/zakura-core/zakura/pull/846)). +- Stopped a dual-stack run as soon as legacy fallback takes over block sync, + and named the handoff as the failure reason instead of reporting missing + header evidence + ([#846](https://github.com/zakura-core/zakura/pull/846)). +- Required exact committed heights from both nodes before the cluster monitor + treats peer advancement as sync-stall evidence + ([#846](https://github.com/zakura-core/zakura/pull/846)).