diff --git a/.env.example b/.env.example index 1d520988d..3320c3ee4 100644 --- a/.env.example +++ b/.env.example @@ -133,6 +133,12 @@ UVICORN_PORT = 8000 # JOB_CHECK_NODE_LIMITS_INTERVAL = 60 # JOB_CLEANUP_SUBSCRIPTION_UPDATES_INTERVAL = 600 +## Node health checks: consecutive failed or good probes before a node is marked error or +## connected again. 1 reacts to a single probe (the old behaviour). A dead node stays in the +## healthy set, and is still polled for usage, for up to fail threshold x health check interval. +# NODE_HEALTH_FAIL_THRESHOLD = 3 +# NODE_HEALTH_RECOVER_THRESHOLD = 2 + ## Developer and test flags. # DOCS = False # DEBUG = False diff --git a/app/db/crud/node.py b/app/db/crud/node.py index 58188f7d2..3ae96b81c 100644 --- a/app/db/crud/node.py +++ b/app/db/crud/node.py @@ -493,7 +493,8 @@ async def update_node_status( message: str = "", xray_version: str = "", node_version: str = "", -) -> Node: + expected_status: NodeStatus | None = None, +) -> Node | None: """ Updates the status of a node. @@ -505,7 +506,7 @@ async def update_node_status( version (str, optional): The version of the node software. Returns: - Node: The updated Node object. + Node: The updated node, or None if a guarded transition was superseded. """ stmt = ( update(Node) @@ -518,8 +519,12 @@ async def update_node_status( last_status_change=datetime.now(UTC), ) ) - await db.execute(stmt) + if expected_status is not None: + stmt = stmt.where(Node.status == expected_status) + result = await db.execute(stmt) await db.commit() + if expected_status is not None and result.rowcount == 0: + return None try: # Prefer refreshing the existing instance to keep relationships loaded diff --git a/app/jobs/node_checker.py b/app/jobs/node_checker.py index e8573972d..2147f71ca 100644 --- a/app/jobs/node_checker.py +++ b/app/jobs/node_checker.py @@ -1,5 +1,6 @@ import asyncio import time +from dataclasses import dataclass from PasarGuardNodeBridge import Health, NodeAPIError, PasarGuardNode from PasarGuardNodeBridge.storage import LifecycleStatus @@ -10,6 +11,7 @@ from app.db.models import Node, NodeStatus from app.models.node import NodeListQuery, NodeNotification from app.nats import is_multi_worker, needs_shared_bridge_memory +from app.nats.leader import is_job_leader, needs_job_leader, set_on_leadership_lost from app.node import node_manager from app.node.nats_memory import ensure_bridge_memory, get_bridge_memory, shutdown_bridge_memory from app.operation import OperatorType @@ -28,6 +30,64 @@ _sync_recovery_deadlines: dict[int, float] = {} +@dataclass +class _HealthStreak: + node: PasarGuardNode + status: NodeStatus + leader: bool + failures: int = 0 + successes: int = 0 + broken: bool = False + + +_health_streaks: dict[int, _HealthStreak] = {} + + +def _owns_health_checks() -> bool: + return not needs_job_leader() or is_job_leader() + + +def _reset_health_streaks() -> None: + _health_streaks.clear() + + +def _accept_health(db_node: Node, node: PasarGuardNode, health: Health) -> bool: + leader = _owns_health_checks() + streak = _health_streaks.get(db_node.id) + if streak is None or streak.node is not node or streak.status != db_node.status or streak.leader != leader: + streak = _health_streaks[db_node.id] = _HealthStreak(node, db_node.status, leader) + if health is Health.BROKEN: + streak.successes = 0 + streak.failures = min(streak.failures + 1, job_settings.node_health_fail_threshold) + accepted = streak.failures >= job_settings.node_health_fail_threshold + streak.broken = streak.broken or accepted + return accepted + streak.failures = 0 + streak.successes = min(streak.successes + 1, job_settings.node_health_recover_threshold) + if (streak.broken or db_node.status != NodeStatus.connected) and ( + streak.successes < job_settings.node_health_recover_threshold + ): + return False + streak.broken = False + return True + + +async def _set_node_health(node: PasarGuardNode, health: Health, node_name: str) -> None: + try: + if await node.get_health() != health: + await node.set_health(health) + except Exception: + # A failed local-state write must never turn a failed probe into success. + logger.exception("[%s] Failed to set node health to %s", node_name, health.name) + + +async def _repair_node(node_id: int) -> None: + if not _owns_health_checks(): + return + async with GetDB() as db: + await node_operator.connect_single_node(db, node_id, health_check=True) + + # pg-node returns these while the HTTP API is up. They are not interchangeable: # - backend gone: keep-alive/crash already called Disconnect; panel must Start again # - core still coming up / Xray API blip: another Start would kill that process @@ -80,82 +140,53 @@ async def verify_node_backend_health(node: PasarGuardNode, node_name: str) -> tu Verify node health by checking backend stats. Returns (health, error_code, error_message) - error_code and error_message are None if no error occurred. """ - current_health = await asyncio.wait_for(node.get_health(), timeout=10) - - # Skip nodes that are not connected or invalid - if current_health in (Health.NOT_CONNECTED, Health.INVALID): - return current_health, None, None - try: + current_health = await asyncio.wait_for(node.get_health(), timeout=10) + if current_health in (Health.NOT_CONNECTED, Health.INVALID): + return current_health, None, None await node.get_backend_stats() - if current_health != Health.HEALTHY: - await node.set_health(Health.HEALTHY) - logger.debug(f"[{node_name}] Node health is HEALTHY") return Health.HEALTHY, None, None - except NodeAPIError as e: - logger.error( - f"[{node_name}] Health check failed, setting health to BROKEN | Error: NodeAPIError(code={e.code}) - {e.detail}" - ) - try: - await node.set_health(Health.BROKEN) - return Health.BROKEN, e.code, e.detail - except Exception as e_set_health: - error_type_set = type(e_set_health).__name__ - logger.error(f"[{node_name}] Failed to set health to BROKEN | Error: {error_type_set} - {e_set_health!s}") - return current_health, e.code, e.detail - except Exception as e: - error_type = type(e).__name__ - error_message = f"{error_type}: {e!s}" - logger.error(f"[{node_name}] Health check failed, setting health to BROKEN | Error: {error_message}") - try: - await node.set_health(Health.BROKEN) - return Health.BROKEN, None, error_message - except Exception as e_set_health: - error_type_set = type(e_set_health).__name__ - logger.error(f"[{node_name}] Failed to set health to BROKEN | Error: {error_type_set} - {e_set_health!s}") - return current_health, None, error_message + except TimeoutError: + return Health.BROKEN, -1, "Health check timeout" + except NodeAPIError as exc: + logger.debug("[%s] Backend health probe failed: %s", node_name, exc.detail) + return Health.BROKEN, exc.code, exc.detail + except Exception as exc: + error_message = f"{type(exc).__name__}: {exc!s}" + logger.debug("[%s] Backend health probe failed: %s", node_name, error_message) + return Health.BROKEN, None, error_message async def process_node_health_check(db_node: Node, node: PasarGuardNode): - """ - Process health check for a single node: - 1. Verify backend health - 2. Recover shared user sync when healthy - 3. Compare with database status - 4. Update status if needed - - Timeout handling: - - For timeout errors (code=-1): Don't reconnect, just wait for recovery - - For other errors (code > -1): Reconnect (connection works but has another issue) - - For NOT_CONNECTED/INVALID: Reconnect immediately - """ - if node is None: + """Probe every worker's attachment; only the job leader changes shared status.""" + if node is None or db_node.status not in ACTIVE_NODE_STATUSES: + _health_streaks.pop(db_node.id, None) return - # Limit concurrent health checks to prevent DB/API overload async with NODE_CHECK_SEM: try: health, error_code, error_message = await verify_node_backend_health(node, db_node.name) except TimeoutError: - # Record timeout error in database but don't reconnect - logger.warning(f"[{db_node.name}] Health check timed out") - async with GetDB() as db: - await NodeOperation._update_single_node_status( - db, db_node.id, NodeStatus.error, message="Health check timeout" - ) - return - except NodeAPIError as e: - # Record error in database - async with GetDB() as db: - await NodeOperation._update_single_node_status(db, db_node.id, NodeStatus.error, message=e.detail) - # For timeout errors (code=-1), don't reconnect - just wait for recovery - if e.code == -1: - logger.warning(f"[{db_node.name}] Health check timed out (NodeAPIError), waiting for recovery") + health, error_code, error_message = Health.BROKEN, -1, "Health check timeout" + except NodeAPIError as exc: + health, error_code, error_message = Health.BROKEN, exc.code, exc.detail + except Exception: + _health_streaks.pop(db_node.id, None) + raise + + if health in (Health.HEALTHY, Health.BROKEN): + if not _accept_health(db_node, node, health): + # An explicitly absent backend needs repair immediately; only + # its downtime reporting waits for the configured threshold. + if health is Health.BROKEN and is_core_dead_error(error_code, error_message): + shared_state = await node.get_lifecycle_state() + if _owns_health_checks() and not await _start_already_in_progress(db_node, shared_state): + await _set_node_health(node, health, db_node.name) + await _repair_node(db_node.id) return - # For other errors, reconnect - async with GetDB() as db: - await node_operator.connect_single_node(db, db_node.id) - return + await _set_node_health(node, health, db_node.name) + else: + _health_streaks.pop(db_node.id, None) # Skip nodes that are already healthy and connected if health == Health.HEALTHY and db_node.status == NodeStatus.connected: @@ -208,63 +239,70 @@ async def process_node_health_check(db_node: Node, node: PasarGuardNode): db_node.name, ) - # Handle NOT_CONNECTED - reconnect immediately - if health is Health.NOT_CONNECTED: - async with GetDB() as db: - await node_operator.connect_single_node(db, db_node.id) + # Followers repair local attachments but never restart a core or publish + # status transitions from independent, potentially conflicting samples. + if not _owns_health_checks(): return - # Handle BROKEN health - if health == Health.BROKEN: - # Record actual error in database - async with GetDB() as db: - await NodeOperation._update_single_node_status(db, db_node.id, NodeStatus.error, message=error_message) - if shared_state is not None: - await node.update_observed_lifecycle(LifecycleStatus.BROKEN, expected_epoch=shared_state.epoch) - # Let pg-node recover transient Xray API/core failures internally. - if should_reconnect_after_health_error(error_code, error_message): + if health is Health.NOT_CONNECTED: + # Initial connection is a lifecycle operation, not a failed probe. + # An already-errored node still needs confirmed recovery afterwards. + if db_node.status == NodeStatus.error: + await _repair_node(db_node.id) + else: async with GetDB() as db: await node_operator.connect_single_node(db, db_node.id) + return + + if health is Health.BROKEN: + async with GetDB() as db: + updated = await NodeOperation._update_single_node_status( + db, + db_node.id, + NodeStatus.error, + message=error_message or "Backend health check failed", + expected_status=db_node.status, + ) + if not updated: + _health_streaks.pop(db_node.id, None) + return + # Check before publishing BROKEN: STARTING is evidence of an operation + # already in progress and must not be overwritten by its health probe. + starting = await _start_already_in_progress(db_node, shared_state) + if not _owns_health_checks(): return - # Keep-alive timeout / crash leaves HTTP up but Xray stopped - # ("backend not initialized"). A second Start while Xray is still - # coming up ("core is not started yet") would kill that process. - if is_core_dead_error(error_code, error_message) and not await _start_already_in_progress( - db_node, shared_state + if shared_state is not None and not starting: + await node.update_observed_lifecycle(LifecycleStatus.BROKEN, expected_epoch=shared_state.epoch) + if not starting and ( + should_reconnect_after_health_error(error_code, error_message) + or is_core_dead_error(error_code, error_message) ): - logger.warning(f"[{db_node.name}] Core is not running; re-applying config") - async with GetDB() as db: - await node_operator.connect_single_node(db, db_node.id) - # For timeout (code=-1 or None) or an in-flight/starting core, wait. + await _repair_node(db_node.id) return - # Update status for recovering nodes - if db_node.status in (NodeStatus.connecting, NodeStatus.error) and health == Health.HEALTHY: + if db_node.status in (NodeStatus.connecting, NodeStatus.error) and health is Health.HEALTHY: + node_version, core_version = await node.get_versions() + if not _owns_health_checks(): + return async with GetDB() as db: - logger.info(f"Node '{db_node.name}' have been recovered") - node_version, core_version = await node.get_versions() - # Connection restored without a hard reset. Suppress the default - # connect notification and send a distinct "recovered" one instead, - # so a self-recovery is visibly different from a full reconnect. - await NodeOperation._update_single_node_status( + updated = await NodeOperation._update_single_node_status( db, db_node.id, NodeStatus.connected, xray_version=core_version, node_version=node_version, send_notification=False, + expected_status=db_node.status, ) + if not updated: + _health_streaks.pop(db_node.id, None) + return if shared_state is not None: await node.update_observed_lifecycle(LifecycleStatus.HEALTHY, expected_epoch=shared_state.epoch) + logger.info("Node '%s' has recovered", db_node.name) await notification.recovered_node( - NodeNotification( - id=db_node.id, - name=db_node.name, - xray_version=core_version, - node_version=node_version, - ) + NodeNotification(id=db_node.id, name=db_node.name, xray_version=core_version, node_version=node_version) ) - return async def check_node_limits(): @@ -303,10 +341,16 @@ async def node_health_check(): db_nodes, _ = await get_nodes(db=db, query=NodeListQuery(status=ACTIVE_NODE_STATUSES), load_usage_logs=False) dict_nodes = await node_manager.get_nodes() - for node_id in _sync_recovery_deadlines.keys() - dict_nodes.keys(): - _sync_recovery_deadlines.pop(node_id, None) + active_ids = {db_node.id for db_node in db_nodes if db_node.id in dict_nodes} + for cache in (_sync_recovery_deadlines, _health_streaks): + for node_id in cache.keys() - active_ids: + cache.pop(node_id, None) check_tasks = [process_node_health_check(db_node, dict_nodes.get(db_node.id)) for db_node in db_nodes] - await asyncio.gather(*check_tasks, return_exceptions=True) + results = await asyncio.gather(*check_tasks, return_exceptions=True) + for db_node, result in zip(db_nodes, results): + if isinstance(result, Exception): + _health_streaks.pop(db_node.id, None) + logger.error("[%s] Health check failed: %s", db_node.name, result, exc_info=result) _node_loop_tasks: list[asyncio.Task] = [] @@ -341,7 +385,7 @@ async def initialize_nodes(): await node_operator.connect_nodes_bulk(db, db_nodes) startup_log("All nodes' cores have been started.") - from app.nats.leader import needs_job_leader + set_on_leadership_lost(_reset_health_streaks) if needs_job_leader(): # Every uvicorn worker must keep local node attachments healthy. @@ -387,6 +431,8 @@ async def _stop_node_loops(): if _node_loop_tasks: await asyncio.gather(*_node_loop_tasks, return_exceptions=True) _node_loop_tasks.clear() + _reset_health_streaks() + _sync_recovery_deadlines.clear() async def shutdown_nodes(): diff --git a/app/operation/node.py b/app/operation/node.py index 0c7ee2954..32306297e 100644 --- a/app/operation/node.py +++ b/app/operation/node.py @@ -162,7 +162,8 @@ async def _update_single_node_status( xray_version: str = "", node_version: str = "", send_notification: bool = True, - ): + expected_status: NodeStatus | None = None, + ) -> bool: """ Update single node status with optional notification. @@ -174,33 +175,44 @@ async def _update_single_node_status( xray_version (str): Xray version. node_version (str): Node version. send_notification (bool): Whether to send notification. + expected_status: Ignore observations made before a concurrent status change. """ db_node = await get_node_by_id(db, node_id, load_usage_logs=False) if not db_node: - return + return False + if expected_status is not None and db_node.status != expected_status: + return False if db_node.status in (NodeStatus.disabled, NodeStatus.limited) and status not in ( NodeStatus.disabled, NodeStatus.limited, ): - return + return False old_status = db_node.status + if expected_status is not None and old_status == status and db_node.message == message: + return True + if status == NodeStatus.error: + if expected_status is not None: + xray_version, node_version = db_node.xray_version, db_node.node_version logger.error(f"Failed to connect node {db_node.name} with id {db_node.id}, Error: {message}") - await update_node_status( + updated = await update_node_status( db=db, db_node=db_node, status=status, message=message, xray_version=xray_version, node_version=node_version, + expected_status=expected_status, ) + if updated is None or db_node.status != status: + return False if not send_notification: - return + return True if status == NodeStatus.connected and old_status != NodeStatus.connected: node_notif = NodeNotification( @@ -221,6 +233,8 @@ async def _update_single_node_status( ) asyncio.create_task(notification.error_node(node_notif)) + return True + @staticmethod async def _get_core_users_map( db: AsyncSession, core_ids: set[int], *, lazy: bool = False @@ -581,7 +595,9 @@ async def connect_nodes_bulk( """ await self._connect_bulk_impl(db, nodes, force_start=force_start) - async def connect_single_node(self, db: AsyncSession, node_id: int, *, force_start: bool = False) -> None: + async def connect_single_node( + self, db: AsyncSession, node_id: int, *, force_start: bool = False, health_check: bool = False + ) -> None: """ Connect a single node and update its status (optimized for single-node operations). @@ -592,7 +608,10 @@ async def connect_single_node(self, db: AsyncSession, node_id: int, *, force_sta db (AsyncSession): Database session. node_id (int): ID of the node to connect. force_start: Push a new Start RPC (admin restart / core apply). + health_check: Defer automatic recovery status until probes confirm it. """ + if health_check: + return await self.recover_node(db, node_id) return await self._connect_single_impl(db, node_id, force_start=force_start) async def disconnect_single_node(self, node_id: int) -> None: @@ -842,7 +861,17 @@ async def _connect_nodes_bulk_remote( {"node_ids": [node.id for node in nodes], "force_start": force_start}, ) - async def _connect_single_node_local(self, db: AsyncSession, node_id: int, *, force_start: bool = False) -> bool: + async def recover_node(self, db: AsyncSession, node_id: int) -> None: + """Repair a node, leaving status/notifications to consecutive health probes.""" + if ( + await self._connect_single_node_local(db, node_id, defer_health_status=True) + and needs_shared_bridge_memory() + ): + await publish_node_sync("connect", node_id) + + async def _connect_single_node_local( + self, db: AsyncSession, node_id: int, *, force_start: bool = False, defer_health_status: bool = False + ) -> bool: db_node = await get_node_by_id(db, node_id, load_usage_logs=False) if db_node is None or db_node.status in (NodeStatus.disabled, NodeStatus.limited): return False @@ -857,6 +886,9 @@ async def _connect_single_node_local(self, db: AsyncSession, node_id: int, *, fo try: await node_manager.update_node(db_node) except NodeAPIError as e: + if defer_health_status: + logger.warning("Node recovery failed for %s: %s", node_id, e.detail) + return False # Update status to error using simple CRUD await update_node_status( db=db, @@ -879,6 +911,10 @@ async def _connect_single_node_local(self, db: AsyncSession, node_id: int, *, fo if not result: return False + if defer_health_status: + # A successful Start/attach is not yet a sustained health recovery. + # A failed repair must not bypass the health failure threshold either. + return result["status"] == NodeStatus.connected # Update status using simple CRUD (NOT bulk!) await update_node_status( diff --git a/config.py b/config.py index 446fc9bfc..8488cf59a 100644 --- a/config.py +++ b/config.py @@ -201,6 +201,8 @@ class UsageSettings(EnvSettings): class JobSettings(EnvSettings): core_health_check_interval: int = Field(default=10, validation_alias="JOB_CORE_HEALTH_CHECK_INTERVAL") + node_health_fail_threshold: int = Field(default=3, ge=1, validation_alias="NODE_HEALTH_FAIL_THRESHOLD") + node_health_recover_threshold: int = Field(default=2, ge=1, validation_alias="NODE_HEALTH_RECOVER_THRESHOLD") record_node_usages_interval: int = Field(default=30, validation_alias="JOB_RECORD_NODE_USAGES_INTERVAL") record_user_usages_interval: int = Field(default=10, validation_alias="JOB_RECORD_USER_USAGES_INTERVAL") review_users_interval: int = Field(default=30, validation_alias="JOB_REVIEW_USERS_INTERVAL") diff --git a/tests/test_node_health_debounce.py b/tests/test_node_health_debounce.py new file mode 100644 index 000000000..39ae134a3 --- /dev/null +++ b/tests/test_node_health_debounce.py @@ -0,0 +1,158 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from PasarGuardNodeBridge import Health + +from app.db.models import NodeStatus +from app.jobs import node_checker +from app.operation.node import NodeOperation + +TIMEOUT = (Health.BROKEN, -1, "Health check timeout") +DEAD_CORE = (Health.BROKEN, 500, "backend not initialized") +HEALTHY = (Health.HEALTHY, None, None) + + +class _FakeDB: + async def __aenter__(self): + return object() + + async def __aexit__(self, *args): + return False + + +def _make_node(): + node = MagicMock() + node.requires_hard_reset.return_value = False + node.get_lifecycle_state = AsyncMock(return_value=None) + node.update_observed_lifecycle = AsyncMock() + node.get_health = AsyncMock(return_value=Health.HEALTHY) + node.set_health = AsyncMock() + node.get_versions = AsyncMock(return_value=("0.5.4", "26.3.27")) + return node + + +@pytest.fixture +def checker(monkeypatch): + monkeypatch.setattr(node_checker, "_health_streaks", {}) + monkeypatch.setattr(node_checker.job_settings, "node_health_fail_threshold", 3) + monkeypatch.setattr(node_checker.job_settings, "node_health_recover_threshold", 2) + monkeypatch.setattr(node_checker, "_owns_health_checks", lambda: True) + monkeypatch.setattr(node_checker, "needs_shared_bridge_memory", lambda: False) + monkeypatch.setattr(node_checker, "GetDB", lambda: _FakeDB()) + monkeypatch.setattr(node_checker, "get_bridge_memory", lambda: (None, None, None)) + monkeypatch.setattr(node_checker.notification, "recovered_node", AsyncMock()) + update_status = AsyncMock(return_value=True) + monkeypatch.setattr(NodeOperation, "_update_single_node_status", update_status) + repair = AsyncMock() + monkeypatch.setattr(node_checker.node_operator, "connect_single_node", repair) + probe = AsyncMock() + monkeypatch.setattr(node_checker, "verify_node_backend_health", probe) + + async def run(db_node, node, *results): + for result in results: + probe.return_value = result + await node_checker.process_node_health_check(db_node, node) + + return SimpleNamespace(run=run, update_status=update_status, repair=repair, monkeypatch=monkeypatch) + + +def _db_node(status=NodeStatus.connected): + return SimpleNamespace(id=7, name="debounced", status=status) + + +def _written_statuses(update_status): + return [call.args[2] for call in update_status.await_args_list] + + +@pytest.mark.asyncio +async def test_failures_below_threshold_keep_the_node_connected(checker): + db_node, node = _db_node(), _make_node() + + await checker.run(db_node, node, TIMEOUT, TIMEOUT) + checker.update_status.assert_not_awaited() + node.set_health.assert_not_awaited() + + await checker.run(db_node, node, TIMEOUT) + assert _written_statuses(checker.update_status) == [NodeStatus.error] + assert checker.update_status.await_args.kwargs["expected_status"] is NodeStatus.connected + node.set_health.assert_awaited_once_with(Health.BROKEN) + + +@pytest.mark.asyncio +async def test_a_good_probe_restarts_the_failure_streak(checker): + db_node, node = _db_node(), _make_node() + + await checker.run(db_node, node, TIMEOUT, TIMEOUT, HEALTHY, TIMEOUT, TIMEOUT) + + checker.update_status.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_errored_node_needs_consecutive_good_probes_to_recover(checker): + db_node, node = _db_node(NodeStatus.error), _make_node() + + await checker.run(db_node, node, HEALTHY) + checker.update_status.assert_not_awaited() + + await checker.run(db_node, node, HEALTHY) + assert _written_statuses(checker.update_status) == [NodeStatus.connected] + assert checker.update_status.await_args.kwargs["expected_status"] is NodeStatus.error + node_checker.notification.recovered_node.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_streak_restarts_for_a_replaced_node_object(checker): + db_node = _db_node() + + await checker.run(db_node, _make_node(), TIMEOUT, TIMEOUT) + await checker.run(db_node, _make_node(), TIMEOUT) + + checker.update_status.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_streak_restarts_when_the_stored_status_changes(checker): + db_node, node = _db_node(), _make_node() + + await checker.run(db_node, node, TIMEOUT, TIMEOUT) + db_node.status = NodeStatus.connecting + await checker.run(db_node, node, TIMEOUT) + + checker.update_status.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_followers_never_write_status_or_repair(checker): + checker.monkeypatch.setattr(node_checker, "_owns_health_checks", lambda: False) + connected, errored = _db_node(), SimpleNamespace(id=8, name="errored", status=NodeStatus.error) + + await checker.run(connected, _make_node(), *[DEAD_CORE] * 5) + await checker.run(errored, _make_node(), *[HEALTHY] * 3) + + checker.update_status.assert_not_awaited() + checker.repair.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_dead_core_is_repaired_before_the_failure_threshold(checker): + db_node, node = _db_node(), _make_node() + + await checker.run(db_node, node, DEAD_CORE) + + checker.repair.assert_awaited_once() + assert checker.repair.await_args.kwargs == {"health_check": True} + checker.update_status.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_thresholds_of_one_react_to_a_single_probe(checker): + checker.monkeypatch.setattr(node_checker.job_settings, "node_health_fail_threshold", 1) + checker.monkeypatch.setattr(node_checker.job_settings, "node_health_recover_threshold", 1) + db_node, node = _db_node(), _make_node() + + await checker.run(db_node, node, TIMEOUT) + db_node.status = NodeStatus.error + await checker.run(db_node, node, HEALTHY) + + assert _written_statuses(checker.update_status) == [NodeStatus.error, NodeStatus.connected]