Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1681,7 +1681,6 @@ async def _check_stale_workers() -> None:
logger.warning("Stale worker check error: %s", exc)
return
now = datetime.now(UTC)
cutoff = now - timedelta(seconds=STALE_WORKER_SECONDS)
purge_cutoff = now - timedelta(hours=1)
global _collector_alerts
for w in workers:
Expand All @@ -1691,6 +1690,7 @@ async def _check_stale_workers() -> None:
continue
cid = w.get("client_id") or w["name"]
last = datetime.fromisoformat(last_hb).replace(tzinfo=UTC)
cutoff = now - timedelta(seconds=_stale_after_seconds(w))
if w["status"] == "online" and last < cutoff:
_worker_online_streak.pop(cid, None)
# Conditional on the heartbeat the decision was based on: a
Expand Down Expand Up @@ -1871,7 +1871,7 @@ async def _raise_worker_offline_alert(w: dict[str, Any]) -> None:
"""
cid = w.get("client_id") or w["name"]
msg = (
f"Worker '{w['name']}': no heartbeat for over {STALE_WORKER_SECONDS // 60} minutes. Its "
f"Worker '{w['name']}': no heartbeat for over {_stale_after_seconds(w) // 60} minutes. Its "
"containers keep running and earning, but CashPilot cannot see or manage them until it reconnects."
)
if alert_id := await database.record_alert("worker", cid, msg):
Expand Down Expand Up @@ -1909,6 +1909,20 @@ async def _raise_worker_offline_alert(w: dict[str, Any]) -> None:
HOSTNAME_PREFIX = os.getenv("CASHPILOT_HOSTNAME_PREFIX", "cashpilot")
COLLECT_INTERVAL_MIN = int(os.getenv("CASHPILOT_COLLECT_INTERVAL", "60"))
STALE_WORKER_SECONDS = 180 # Mark worker offline after 3 missed heartbeats
# A phone reports from Android's WorkManager, whose shortest period is 15
# minutes, so 180 s would show every phone offline between two reports.
# Three missed periods, the same rule as the 60 s Docker workers.
STALE_ANDROID_WORKER_SECONDS = 45 * 60


def _stale_after_seconds(w: dict[str, Any]) -> int:
"""How long this worker may stay silent before it is marked offline."""
info = w.get("system_info") or {}
if not isinstance(info, dict): # raw DB rows carry it as JSON text
info = _safe_json(info, {})
if placement.is_android(info):
return STALE_ANDROID_WORKER_SECONDS
return STALE_WORKER_SECONDS


async def _warm_session_epochs() -> None:
Expand Down
2 changes: 1 addition & 1 deletion docs/fleet.md
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ If the worker and UI are on different subnets (e.g., connected via Tailscale):

## Offline Handling

If a worker goes offline (no heartbeat for 180 seconds):
If a worker goes offline (no heartbeat for 180 seconds; 45 minutes for an Android phone, which reports every 15 minutes):

- The UI marks the worker as offline
- Historical earnings and health data is retained
Expand Down
112 changes: 112 additions & 0 deletions tests/test_android_worker_stale.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
"""A phone is offline after three missed reports, not after three minutes.

The Android client reports from WorkManager, whose shortest period is 15
minutes. With the Docker workers' 180 s threshold every phone would be marked
offline, and alerted on, between two ordinary reports. Phones get three missed
periods (45 min); Docker workers keep 180 s.
"""

import asyncio
import os
from datetime import UTC, datetime, timedelta

os.environ.setdefault("CASHPILOT_API_KEY", "test-fleet-key")

import pytest # noqa: E402

try:
from app.main import _check_stale_workers # noqa: E402
except ImportError:
pytest.skip(
"Requires full app dependencies (fastapi, httpx, etc.) — runs in CI",
allow_module_level=True,
)

from unittest.mock import AsyncMock, patch # noqa: E402

from app import main # noqa: E402

DOCKER = '{"docker_available": true}'
PHONE = '{"os": "Android", "device_type": "android"}'
# CashPilot-android 0.3.0 sends only os, no device_type.
OLD_PHONE = '{"os": "Android"}'


def _silent_for(minutes: float) -> str:
return (datetime.now(UTC) - timedelta(minutes=minutes)).replace(tzinfo=None).isoformat()


def _row(system_info: str, silent_min: float) -> dict:
return {
"id": 7,
"client_id": "cid-pixel",
"name": "pixel",
"url": "",
"status": "online",
"containers": "[]",
"apps": "[]",
"system_info": system_info,
"last_heartbeat": _silent_for(silent_min),
"registered_at": "2026-04-01T00:00:00",
"api_key_enc": "enc",
}


@pytest.fixture(autouse=True)
def _isolate():
before = main._collector_alerts
main._collector_alerts = []
main._worker_online_streak.clear()
main._worker_last_offline_push.clear()
yield
main._collector_alerts = before
main._worker_online_streak.clear()
main._worker_last_offline_push.clear()


def _sweep(row: dict):
record = AsyncMock(return_value=True)
mark = AsyncMock(return_value=True)

async def _send(*a, **kw):
return 1

with (
patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=[row]),
patch("app.main.database.mark_worker_offline_if_unchanged", mark),
patch("app.main.database.delete_worker", new_callable=AsyncMock),
patch("app.main.database.record_alert", record),
patch("app.main.database.clear_alerts", AsyncMock()),
patch("app.main.notify.send", _send),
):
asyncio.run(_check_stale_workers())
return mark, record


@pytest.mark.parametrize("info", [PHONE, OLD_PHONE])
def test_phone_between_two_reports_stays_online(info):
mark, record = _sweep(_row(info, silent_min=16))
mark.assert_not_awaited()
record.assert_not_awaited()


def test_phone_after_three_missed_reports_goes_offline_with_its_own_minutes():
mark, record = _sweep(_row(PHONE, silent_min=46))
mark.assert_awaited_once()
msg = record.await_args.args[2]
assert "over 45 minutes" in msg, msg


def test_docker_worker_keeps_the_three_minute_threshold():
# Control: the same sweep, a Docker worker silent for 4 minutes, still goes
# offline, so the phone tests above are not passing because nothing fires.
mark, record = _sweep(_row(DOCKER, silent_min=4))
mark.assert_awaited_once()
assert "over 3 minutes" in record.await_args.args[2]


def test_threshold_reads_decoded_and_raw_system_info():
assert main._stale_after_seconds({"system_info": PHONE}) == 45 * 60
assert main._stale_after_seconds({"system_info": {"device_type": "android"}}) == 45 * 60
assert main._stale_after_seconds({"system_info": "not json"}) == main.STALE_WORKER_SECONDS
assert main._stale_after_seconds({}) == main.STALE_WORKER_SECONDS
Loading