feat: add settings sync logic to poll cycle with 30-second interval
- Add SETTINGS_SYNC_INTERVAL = 15 constant (15 poll cycles × 2s ≈ 30s) - Add _settings_sync_counter module-level counter - Add _sync_settings_with_remotes() async function that: - GETs /api/settings/sync from each remote instance - Adopts remote settings if remote timestamp is newer - Pushes local settings via PUT if local timestamp is newer - Silently skips 404/405 (older muxplex without sync endpoints) - Catches all per-remote errors and logs as warnings - Wire sync call into _run_poll_cycle() step 13 (runs outside state_lock) - Add test_settings_sync_poll.py with 10 tests covering all sync behaviours
This commit is contained in:
@@ -79,6 +79,7 @@ from muxplex.ttyd import kill_orphan_ttyd, kill_ttyd, spawn_ttyd, TTYD_PORT
|
|||||||
|
|
||||||
POLL_INTERVAL: float = float(os.environ.get("POLL_INTERVAL", "2.0"))
|
POLL_INTERVAL: float = float(os.environ.get("POLL_INTERVAL", "2.0"))
|
||||||
SERVER_PORT: int = int(os.environ.get("MUXPLEX_PORT", "8088"))
|
SERVER_PORT: int = int(os.environ.get("MUXPLEX_PORT", "8088"))
|
||||||
|
SETTINGS_SYNC_INTERVAL: int = 15 # sync every ~30 seconds (15 * 2s poll interval)
|
||||||
|
|
||||||
_log = logging.getLogger(__name__)
|
_log = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -88,6 +89,74 @@ _log = logging.getLogger(__name__)
|
|||||||
|
|
||||||
_poll_task: asyncio.Task | None = None
|
_poll_task: asyncio.Task | None = None
|
||||||
_federation_client: httpx.AsyncClient | None = None
|
_federation_client: httpx.AsyncClient | None = None
|
||||||
|
_settings_sync_counter: int = 0
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Settings sync
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def _sync_settings_with_remotes(
|
||||||
|
settings: dict, http_client: httpx.AsyncClient
|
||||||
|
) -> None:
|
||||||
|
"""Sync settings with all reachable remote instances.
|
||||||
|
|
||||||
|
For each remote:
|
||||||
|
- GET /api/settings/sync to retrieve remote timestamp.
|
||||||
|
- If remote is newer: adopt remote settings via apply_synced_settings().
|
||||||
|
- If local is newer: push local settings via PUT /api/settings/sync.
|
||||||
|
- If equal: no action.
|
||||||
|
|
||||||
|
Errors are caught per-remote so one unreachable peer doesn't abort others.
|
||||||
|
404/405 responses from older muxplex instances that lack sync endpoints are
|
||||||
|
silently skipped.
|
||||||
|
"""
|
||||||
|
local_sync = get_syncable_settings()
|
||||||
|
local_ts = local_sync.get("settings_updated_at", 0.0)
|
||||||
|
|
||||||
|
for remote in settings.get("remote_instances", []):
|
||||||
|
url = remote.get("url", "").rstrip("/")
|
||||||
|
key = remote.get("key", "")
|
||||||
|
if not url:
|
||||||
|
continue
|
||||||
|
headers = {"Authorization": f"Bearer {key}"} if key else {}
|
||||||
|
try:
|
||||||
|
resp = await http_client.get(
|
||||||
|
f"{url}/api/settings/sync", headers=headers, timeout=5.0
|
||||||
|
)
|
||||||
|
if resp.status_code in (404, 405):
|
||||||
|
# Older muxplex instance without sync endpoint — skip silently.
|
||||||
|
continue
|
||||||
|
resp.raise_for_status()
|
||||||
|
remote_data = resp.json()
|
||||||
|
remote_ts = remote_data.get("settings_updated_at", 0.0)
|
||||||
|
|
||||||
|
if remote_ts > local_ts:
|
||||||
|
# Remote is newer — adopt.
|
||||||
|
apply_synced_settings(remote_data.get("settings", {}), remote_ts)
|
||||||
|
# Refresh local state so subsequent remotes see the updated ts.
|
||||||
|
local_sync = get_syncable_settings()
|
||||||
|
local_ts = local_sync.get("settings_updated_at", 0.0)
|
||||||
|
elif local_ts > remote_ts:
|
||||||
|
# Local is newer — push.
|
||||||
|
payload = {
|
||||||
|
"settings": {
|
||||||
|
k: local_sync[k]
|
||||||
|
for k in local_sync
|
||||||
|
if k != "settings_updated_at"
|
||||||
|
},
|
||||||
|
"settings_updated_at": local_ts,
|
||||||
|
}
|
||||||
|
await http_client.put(
|
||||||
|
f"{url}/api/settings/sync",
|
||||||
|
json=payload,
|
||||||
|
headers=headers,
|
||||||
|
timeout=5.0,
|
||||||
|
)
|
||||||
|
# If equal: no action.
|
||||||
|
except Exception as exc:
|
||||||
|
_log.warning("Settings sync with %s failed: %s", url, exc)
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -97,6 +166,7 @@ _federation_client: httpx.AsyncClient | None = None
|
|||||||
|
|
||||||
async def _run_poll_cycle() -> None:
|
async def _run_poll_cycle() -> None:
|
||||||
"""Perform one full poll cycle, all operations executed under state_lock."""
|
"""Perform one full poll cycle, all operations executed under state_lock."""
|
||||||
|
global _settings_sync_counter
|
||||||
async with state_lock:
|
async with state_lock:
|
||||||
# 1. Enumerate live tmux sessions
|
# 1. Enumerate live tmux sessions
|
||||||
names = await enumerate_sessions()
|
names = await enumerate_sessions()
|
||||||
@@ -184,6 +254,19 @@ async def _run_poll_cycle() -> None:
|
|||||||
# 12. Atomically persist the updated state
|
# 12. Atomically persist the updated state
|
||||||
save_state(state)
|
save_state(state)
|
||||||
|
|
||||||
|
# 13. Periodically sync settings with remote instances (every SETTINGS_SYNC_INTERVAL
|
||||||
|
# poll cycles, ~30 seconds). Runs outside the state_lock to avoid blocking the
|
||||||
|
# poll cycle while waiting on remote HTTP calls.
|
||||||
|
_settings_sync_counter += 1
|
||||||
|
if _settings_sync_counter >= SETTINGS_SYNC_INTERVAL:
|
||||||
|
_settings_sync_counter = 0
|
||||||
|
if _federation_client is not None:
|
||||||
|
settings = load_settings()
|
||||||
|
try:
|
||||||
|
await _sync_settings_with_remotes(settings, _federation_client)
|
||||||
|
except Exception:
|
||||||
|
_log.exception("settings sync cycle error")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Poll loop
|
# Poll loop
|
||||||
|
|||||||
@@ -0,0 +1,250 @@
|
|||||||
|
"""
|
||||||
|
Tests for settings sync logic in the poll cycle (main.py).
|
||||||
|
|
||||||
|
Covers:
|
||||||
|
- SETTINGS_SYNC_INTERVAL constant exists
|
||||||
|
- _settings_sync_counter module-level variable exists
|
||||||
|
- _sync_settings_with_remotes function exists and is callable
|
||||||
|
- Sync adopts newer remote settings (GET, remote_ts > local_ts)
|
||||||
|
- Sync pushes local settings to older remote (PUT, local_ts > remote_ts)
|
||||||
|
- Sync skips when timestamps are equal
|
||||||
|
- Sync gracefully handles 404 from older muxplex instances
|
||||||
|
- Sync gracefully handles connection errors
|
||||||
|
- Sync skips remotes with no URL
|
||||||
|
"""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
import muxplex.main as main_mod
|
||||||
|
import muxplex.settings as settings_mod
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Autouse fixture: redirect settings to tmp_path
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def redirect_settings(tmp_path, monkeypatch):
|
||||||
|
"""Redirect SETTINGS_PATH to a temporary file for all tests."""
|
||||||
|
fake_path = tmp_path / "settings.json"
|
||||||
|
monkeypatch.setattr(settings_mod, "SETTINGS_PATH", fake_path)
|
||||||
|
return fake_path
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Pattern / existence tests
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_sync_interval_constant_exists():
|
||||||
|
"""SETTINGS_SYNC_INTERVAL constant must exist in main module."""
|
||||||
|
assert hasattr(main_mod, "SETTINGS_SYNC_INTERVAL")
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_sync_interval_is_15():
|
||||||
|
"""SETTINGS_SYNC_INTERVAL must equal 15 (15 * 2s = ~30s)."""
|
||||||
|
assert main_mod.SETTINGS_SYNC_INTERVAL == 15
|
||||||
|
|
||||||
|
|
||||||
|
def test_settings_sync_counter_exists():
|
||||||
|
"""_settings_sync_counter module-level variable must exist in main module."""
|
||||||
|
assert hasattr(main_mod, "_settings_sync_counter")
|
||||||
|
|
||||||
|
|
||||||
|
def test_sync_settings_with_remotes_is_callable():
|
||||||
|
"""_sync_settings_with_remotes must be an async callable in main module."""
|
||||||
|
assert hasattr(main_mod, "_sync_settings_with_remotes")
|
||||||
|
assert asyncio.iscoroutinefunction(main_mod._sync_settings_with_remotes)
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Behaviour tests: sync logic with mocked HTTP
|
||||||
|
#
|
||||||
|
# Patch at the main_mod level (not settings_mod) because the functions are
|
||||||
|
# imported into muxplex.main's namespace via "from muxplex.settings import ...".
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_adopts_newer_remote_settings():
|
||||||
|
"""When remote timestamp is newer, apply_synced_settings is called with remote data."""
|
||||||
|
local_ts = 100.0
|
||||||
|
remote_ts = 200.0
|
||||||
|
remote_settings = {"fontSize": 18, "sort_order": "alpha"}
|
||||||
|
|
||||||
|
get_resp = MagicMock()
|
||||||
|
get_resp.status_code = 200
|
||||||
|
get_resp.json.return_value = {
|
||||||
|
"settings": remote_settings,
|
||||||
|
"settings_updated_at": remote_ts,
|
||||||
|
}
|
||||||
|
get_resp.raise_for_status = MagicMock()
|
||||||
|
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock(return_value=get_resp)
|
||||||
|
http_client.put = AsyncMock()
|
||||||
|
|
||||||
|
remote_config = {
|
||||||
|
"remote_instances": [{"url": "http://remote1:8088", "key": "testkey"}]
|
||||||
|
}
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(main_mod, "get_syncable_settings") as mock_get_sync,
|
||||||
|
patch.object(main_mod, "apply_synced_settings") as mock_apply,
|
||||||
|
):
|
||||||
|
mock_get_sync.return_value = {
|
||||||
|
"fontSize": 14,
|
||||||
|
"sort_order": "manual",
|
||||||
|
"settings_updated_at": local_ts,
|
||||||
|
}
|
||||||
|
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
mock_apply.assert_called_once_with(remote_settings, remote_ts)
|
||||||
|
http_client.put.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_pushes_local_to_older_remote():
|
||||||
|
"""When local timestamp is newer, PUT is called to push local settings to remote."""
|
||||||
|
local_ts = 200.0
|
||||||
|
remote_ts = 100.0
|
||||||
|
local_syncable = {
|
||||||
|
"fontSize": 16,
|
||||||
|
"sort_order": "alpha",
|
||||||
|
"settings_updated_at": local_ts,
|
||||||
|
}
|
||||||
|
|
||||||
|
get_resp = MagicMock()
|
||||||
|
get_resp.status_code = 200
|
||||||
|
get_resp.json.return_value = {
|
||||||
|
"settings": {"fontSize": 14},
|
||||||
|
"settings_updated_at": remote_ts,
|
||||||
|
}
|
||||||
|
get_resp.raise_for_status = MagicMock()
|
||||||
|
|
||||||
|
put_resp = MagicMock()
|
||||||
|
put_resp.status_code = 200
|
||||||
|
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock(return_value=get_resp)
|
||||||
|
http_client.put = AsyncMock(return_value=put_resp)
|
||||||
|
|
||||||
|
remote_config = {
|
||||||
|
"remote_instances": [{"url": "http://remote1:8088", "key": "testkey"}]
|
||||||
|
}
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(main_mod, "get_syncable_settings") as mock_get_sync,
|
||||||
|
patch.object(main_mod, "apply_synced_settings") as mock_apply,
|
||||||
|
):
|
||||||
|
mock_get_sync.return_value = local_syncable
|
||||||
|
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
mock_apply.assert_not_called()
|
||||||
|
http_client.put.assert_called_once()
|
||||||
|
call_kwargs = http_client.put.call_args
|
||||||
|
assert "/api/settings/sync" in call_kwargs[0][0]
|
||||||
|
payload = call_kwargs[1]["json"]
|
||||||
|
assert payload["settings_updated_at"] == local_ts
|
||||||
|
assert "fontSize" in payload["settings"]
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_skips_equal_timestamps():
|
||||||
|
"""When timestamps are equal, neither apply_synced_settings nor PUT is called."""
|
||||||
|
equal_ts = 100.0
|
||||||
|
|
||||||
|
get_resp = MagicMock()
|
||||||
|
get_resp.status_code = 200
|
||||||
|
get_resp.json.return_value = {
|
||||||
|
"settings": {"fontSize": 14},
|
||||||
|
"settings_updated_at": equal_ts,
|
||||||
|
}
|
||||||
|
get_resp.raise_for_status = MagicMock()
|
||||||
|
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock(return_value=get_resp)
|
||||||
|
http_client.put = AsyncMock()
|
||||||
|
|
||||||
|
remote_config = {
|
||||||
|
"remote_instances": [{"url": "http://remote1:8088", "key": "testkey"}]
|
||||||
|
}
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(main_mod, "get_syncable_settings") as mock_get_sync,
|
||||||
|
patch.object(main_mod, "apply_synced_settings") as mock_apply,
|
||||||
|
):
|
||||||
|
mock_get_sync.return_value = {"fontSize": 14, "settings_updated_at": equal_ts}
|
||||||
|
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
mock_apply.assert_not_called()
|
||||||
|
http_client.put.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_handles_404_gracefully():
|
||||||
|
"""A 404 from the remote (older muxplex, no sync endpoint) is silently skipped."""
|
||||||
|
get_resp = MagicMock()
|
||||||
|
get_resp.status_code = 404
|
||||||
|
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock(return_value=get_resp)
|
||||||
|
http_client.put = AsyncMock()
|
||||||
|
|
||||||
|
remote_config = {
|
||||||
|
"remote_instances": [{"url": "http://remote1:8088", "key": "testkey"}]
|
||||||
|
}
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(main_mod, "get_syncable_settings") as mock_get_sync,
|
||||||
|
patch.object(main_mod, "apply_synced_settings") as mock_apply,
|
||||||
|
):
|
||||||
|
mock_get_sync.return_value = {"fontSize": 14, "settings_updated_at": 100.0}
|
||||||
|
|
||||||
|
# Must not raise
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
mock_apply.assert_not_called()
|
||||||
|
http_client.put.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_handles_connection_error_gracefully():
|
||||||
|
"""Connection errors during sync are caught and logged, not re-raised."""
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock(side_effect=ConnectionError("timeout"))
|
||||||
|
http_client.put = AsyncMock()
|
||||||
|
|
||||||
|
remote_config = {
|
||||||
|
"remote_instances": [{"url": "http://remote1:8088", "key": "testkey"}]
|
||||||
|
}
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch.object(main_mod, "get_syncable_settings") as mock_get_sync,
|
||||||
|
patch.object(main_mod, "apply_synced_settings") as mock_apply,
|
||||||
|
):
|
||||||
|
mock_get_sync.return_value = {"fontSize": 14, "settings_updated_at": 100.0}
|
||||||
|
|
||||||
|
# Must not raise even with connection error
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
mock_apply.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_sync_skips_remote_with_no_url():
|
||||||
|
"""Remotes without a URL are silently skipped (no HTTP call made)."""
|
||||||
|
http_client = MagicMock()
|
||||||
|
http_client.get = AsyncMock()
|
||||||
|
http_client.put = AsyncMock()
|
||||||
|
|
||||||
|
remote_config = {"remote_instances": [{"url": "", "key": "somekey"}]}
|
||||||
|
|
||||||
|
with patch.object(main_mod, "get_syncable_settings") as mock_get_sync:
|
||||||
|
mock_get_sync.return_value = {"fontSize": 14, "settings_updated_at": 100.0}
|
||||||
|
|
||||||
|
await main_mod._sync_settings_with_remotes(remote_config, http_client)
|
||||||
|
|
||||||
|
http_client.get.assert_not_called()
|
||||||
|
http_client.put.assert_not_called()
|
||||||
Reference in New Issue
Block a user