worker: make stale peer cleanup reliable

Author: awlx

PARKER.md

@@ -19,11 +19,41 @@ For Parker the following configuration items are relevant in addition to the def
 - `sticky_worker_tolerance`
 - `broker_signing_key`
 - `parker` section
+- `cleanup` section
 - `netbox` section
 
 
 ## Design considerations
 
+### Offline peer cleanup
+
+The Parker worker defaults to a 300-second stale-handshake timeout and a
+600-second initial-handshake grace. The stale timeout accommodates the default
+120-second config retry and 180-second WireGuard tunnel timeout. Zero or missing
+handshake timestamps mean that a peer has never handshaked; they do not cause
+immediate deletion. Accepted queue updates refresh an in-memory, bounded
+provisioning tracker, and cleanup serializes only with updates for the same
+public key or assigned prefix.
+
+WireGuard does not expose peer creation time. After a worker restart, every
+untracked never-handshaked peer therefore receives one grace period from worker
+startup. Truly abandoned peers are removed on a later sweep rather than retained
+forever. The tracker retains at most 65,536 peer timestamps; if that bound is
+reached before entries age out, new provisioning fails explicitly rather than
+dropping grace state and risking premature cleanup.
+
+For a stale Parker peer, cleanup removes the selected assigned-prefix route
+before removing the WireGuard peer. If another current peer owns the prefix, its
+route is preserved. A real route or peer failure is reported and retried on the
+next sweep; already-absent state is treated as idempotent.
+
+When an accepted queue update reassigns a peer, the worker locks both the old
+and new prefixes, installs the new peer state and route, then removes the old
+route. `Range6` must match the configured Parker IPv6 allocation prefix length.
+If old-route deletion fails after reassignment, the bounded worker-side tracker
+retries it on cleanup sweeps after checking that no current peer owns it. This
+prevents cleanup from preserving a route that the peer immediately abandons.
+
 ### Concentrator selection
 
 Parker nodes call the config endpoint every `retry_interval` (default: 120s), regardless of whether they currently have connectivity or not.

README.md

@@ -11,6 +11,7 @@
     - [Backend worker](#backend-worker)
   - [Installation](#installation)
   - [Configuration](#configuration)
+    - [Worker peer cleanup](#worker-peer-cleanup)
   - [Development](#development)
     - [Build using Bazel](#build-using-bazel)
     - [Updating PIP dependencies for Bazel](#updating-pip-dependencies-for-bazel)
@@ -152,6 +153,29 @@ For further information, please see this [presentation on the architecture](http
 The `wgkex` configuration file defaults to `/etc/wgkex.yaml` ([Sample configuration file](wgkex.yaml.example)), however
 can also be overwritten by setting the environment variable `WGKEX_CONFIG_FILE`.
 
+### Worker peer cleanup
+
+Workers periodically remove offline WireGuard peers. A peer with a non-zero
+last-handshake timestamp is stale when its age is greater than or equal to the
+mode-specific timeout. A peer that has never handshaked is retained for
+`cleanup.initial_handshake_grace` after its most recent accepted queue update.
+After a worker restart, the kernel does not expose peer creation time, so all
+untracked never-handshaked peers receive one grace period starting at worker
+startup; abandoned peers are removed by the first later sweep.
+
+Cleanup scans do not block MQTT ingestion. Queue updates and cleanup serialize
+mutations for the same public key or Parker prefix. Parker deletion removes the
+assigned prefix route before the WireGuard peer, unless another current peer
+owns that prefix. Legacy deletion removes the FDB entry, then route, then peer.
+Already-absent dependencies are idempotent; other partial failures are reported
+and retried while the peer remains discoverable. Parker prefix reassignment
+locks both old and new prefixes and removes the old route after installing the
+new peer state and route; failed old-route deletion is retained and retried on
+later cleanup sweeps after an ownership check.
+
+See `wgkex.yaml.example` for the cleanup interval, Parker and legacy stale
+timeouts, and initial-handshake grace defaults.
+
 ## Development
 
 ### Build using [Bazel](https://bazel.build)

wgkex.yaml.example

@@ -42,6 +42,18 @@ workers:
 sticky_worker_tolerance: 15
 # [worker] The external hostname of this worker
 externalName: gw04.ext.ffmuc.net
+# [worker] Offline peer cleanup timing, in seconds. All values must be positive.
+cleanup:
+  # How often each WireGuard interface is scanned.
+  interval: 3600
+  # A Parker peer with an earlier last handshake is stale. The default allows
+  # multiple 120s config retries and the 180s WireGuard tunnel timeout.
+  parker_stale_timeout: 300
+  # Preserve the historical three-hour timeout for legacy peers.
+  legacy_stale_timeout: 10800
+  # Never-handshaked peers are retained this long after provisioning or worker
+  # startup, then treated as abandoned.
+  initial_handshake_grace: 600
 # [broker, worker]
 parker:
   # Enable Parker support. The broker will still handle non-Parker in the same instance, the worker however will disable non-Parker support in this case.

wgkex/config/config.py

@@ -2,6 +2,7 @@
 
 import dataclasses
 import logging
+import math
 import os
 import sys
 from enum import Enum
@@ -18,6 +19,19 @@ class ConfigFileNotFoundError(Error):
     """File could not be found on disk."""
 
 
+def _positive_duration(value: Any, name: str) -> float:
+    """Parse a finite, positive duration in seconds."""
+    if isinstance(value, bool):
+        raise ValueError(f"{name} must be a positive duration in seconds")
+    try:
+        duration = float(value)
+    except (TypeError, ValueError) as error:
+        raise ValueError(f"{name} must be a positive duration in seconds") from error
+    if not math.isfinite(duration) or duration <= 0:
+        raise ValueError(f"{name} must be a positive duration in seconds")
+    return duration
+
+
 WG_CONFIG_OS_ENV = "WGKEX_CONFIG_FILE"
 WG_CONFIG_DEFAULT_LOCATION = "/etc/wgkex.yaml"
 
@@ -173,6 +187,42 @@ def from_dict(cls, mqtt_cfg: Dict[str, str]) -> "MQTT":
         )
 
 
+@dataclasses.dataclass(frozen=True)
+class Cleanup:
+    """Worker peer cleanup timing in seconds."""
+
+    interval: float = 3600
+    parker_stale_timeout: float = 300
+    legacy_stale_timeout: float = 10800
+    initial_handshake_grace: float = 600
+
+    @classmethod
+    def from_dict(cls, cleanup_cfg: Dict[str, Any]) -> "Cleanup":
+        if not isinstance(cleanup_cfg, dict):
+            raise ValueError("cleanup must be a mapping")
+        return cls(
+            interval=_positive_duration(
+                cleanup_cfg.get("interval", cls.interval), "cleanup.interval"
+            ),
+            parker_stale_timeout=_positive_duration(
+                cleanup_cfg.get("parker_stale_timeout", cls.parker_stale_timeout),
+                "cleanup.parker_stale_timeout",
+            ),
+            legacy_stale_timeout=_positive_duration(
+                cleanup_cfg.get("legacy_stale_timeout", cls.legacy_stale_timeout),
+                "cleanup.legacy_stale_timeout",
+            ),
+            initial_handshake_grace=_positive_duration(
+                cleanup_cfg.get("initial_handshake_grace", cls.initial_handshake_grace),
+                "cleanup.initial_handshake_grace",
+            ),
+        )
+
+    def stale_timeout(self, parker: bool) -> float:
+        """Return the stale timeout for the selected worker mode."""
+        return self.parker_stale_timeout if parker else self.legacy_stale_timeout
+
+
 @dataclasses.dataclass
 class Parker:
     """A representation of the 'parker' key in Configuration file.
@@ -389,6 +439,7 @@ class Config:
     domain_prefixes: List[str]
     workers: Workers
     parker: Parker
+    cleanup: Cleanup
     external_name: Optional[str]
     mqtt: MQTT
     broker_listen: BrokerListen
@@ -410,6 +461,7 @@ def from_dict(cls, cfg: Dict[str, Any]) -> "Config":
             cfg.get("workers", {}), cfg.get("sticky_worker_tolerance", 10)
         )
         parker = Parker.from_dict(cfg.get("parker", {}))
+        cleanup = Cleanup.from_dict(cfg.get("cleanup", {}))
         broker_signing_key = cfg.get("broker_signing_key", None)
 
         netbox_cfg = None
@@ -428,6 +480,7 @@ def from_dict(cls, cfg: Dict[str, Any]) -> "Config":
             workers=workers_cfg,
             external_name=cfg.get("externalName"),
             parker=parker,
+            cleanup=cleanup,
             netbox=netbox_cfg,
             broker_signing_key=broker_signing_key,
         )

wgkex/config/config_test.py

@@ -182,6 +182,43 @@ def test_worker_explicit_null_values_use_defaults(self):
         workers = config.Workers.from_dict({"worker": {"weight": 0}}, 10)
         self.assertEqual(workers.get("worker").weight, 0)
 
+    def test_cleanup_defaults_and_mode_timeouts(self):
+        cleanup = config.Cleanup.from_dict({})
+        self.assertEqual(cleanup.interval, 3600)
+        self.assertEqual(cleanup.stale_timeout(True), 300)
+        self.assertEqual(cleanup.stale_timeout(False), 10800)
+        self.assertEqual(cleanup.initial_handshake_grace, 600)
+
+        cfg = _config_dict()
+        cfg["cleanup"] = {
+            "interval": 30,
+            "parker_stale_timeout": 600,
+            "legacy_stale_timeout": 7200,
+            "initial_handshake_grace": 900,
+        }
+        parsed = config.Config.from_dict(cfg)
+        self.assertEqual(parsed.cleanup.interval, 30)
+        self.assertEqual(parsed.cleanup.stale_timeout(True), 600)
+        self.assertEqual(parsed.cleanup.stale_timeout(False), 7200)
+        self.assertEqual(parsed.cleanup.initial_handshake_grace, 900)
+
+    def test_cleanup_durations_are_positive_and_finite(self):
+        fields = (
+            "interval",
+            "parker_stale_timeout",
+            "legacy_stale_timeout",
+            "initial_handshake_grace",
+        )
+        for field in fields:
+            for value in (0, -1, True, "invalid", float("inf"), None):
+                with (
+                    self.subTest(field=field, value=value),
+                    self.assertRaisesRegex(ValueError, field),
+                ):
+                    config.Cleanup.from_dict({field: value})
+        with self.assertRaisesRegex(ValueError, "mapping"):
+            config.Cleanup.from_dict([])  # type: ignore
+
     def test_parker_prefix_structure_validation(self):
         invalid_configs = [
             {"enabled": True, "464xlat": True, "ipam": "json"},

wgkex/worker/BUILD

@@ -2,6 +2,20 @@ load("@rules_python//python:defs.bzl", "py_binary", "py_library", "py_test")
 load("@pip//:requirements.bzl", "requirement")
 
 
+py_library(
+    name = "peer_state",
+    srcs = ["peer_state.py"],
+    visibility = ["//visibility:public"],
+)
+
+py_test(
+    name = "peer_state_test",
+    srcs = ["peer_state_test.py"],
+    deps = [":peer_state"],
+    size = "small",
+)
+
+
 py_library(
     name = "netlink",
     srcs = ["netlink.py"],
@@ -13,6 +27,7 @@ py_library(
        "//wgkex/common:utils",
        "//wgkex/common:logger",
        "//wgkex/config:config",
+       ":peer_state",
     ],
 )
 
@@ -42,6 +57,7 @@ py_library(
        "//wgkex/config:config",
        ":msg_queue",
        ":netlink",
+       ":peer_state",
     ],
 )
 
@@ -62,6 +78,7 @@ py_binary(
     deps = [
        ":mqtt",
        ":msg_queue",
+       ":peer_state",
        "//wgkex/config:config",
        "//wgkex/common:logger",
     ],
@@ -85,6 +102,7 @@ py_library(
     deps = [
        "//wgkex/common:logger",
        ":netlink",
+       ":peer_state",
     ],
 )
 
@@ -93,6 +111,8 @@ py_test(
     srcs = ["msg_queue_test.py"],
     deps = [
        ":msg_queue",
+       ":netlink",
+       ":peer_state",
        requirement("mock"),
     ],
     size="small",