sync: add cross-subsystem event bus for config consistency

Add EventBus with loop guards to keep firewall, dnsmasq, wireguard, and network configs consistent. Handlers emit SyncEvent after mutations; subscribers compute diffs and write JSON without manual cascade loops.
2026-06-29 20:44:36 +00:00
parent 348bbfbca6
commit 9088f34345
15 changed files with 2117 additions and 59 deletions
@@ -26,7 +26,8 @@ from daemon.iface import (
    POST_DNSMASQ_UPSTREAMS,
 )
 from daemon.server import NotFoundError, refresh_state, registry
-from lib.common import deep_merge, ensure_dirs, load_json, run, run_proc, save_json
+from lib.common import deep_merge, ensure_dirs, load_json, run, save_json
+from lib.sync import SyncEvent, bus

 logger = logging.getLogger(__name__)

@@ -156,7 +157,10 @@ def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str,
    if not body:
        raise ValueError("Request body required")
    _save_config(body)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent("dnsmasq", "config_saved", {"action": "config_saved"})
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -172,7 +176,10 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    current = _get_config()
    merged = deep_merge(current, body)
    _save_config(merged)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent("dnsmasq", "config_saved", {"action": "config_patched"})
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -198,8 +205,11 @@ def apply_config(_request: Any, _body: Any) -> dict[str, Any]:
    cfg_after = _get_config()
    cfg_after[_APPLY_HASH_KEY] = _config_hash(cfg_after)
    _save_config(cfg_after)
-    refresh_state(["dnsmasq"])
-    return {"applied": True}
+    sync_result = bus.emit(
+        SyncEvent("dnsmasq", "config_saved", {"action": "config_applied"})
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
+    return {"applied": True, "synced": sync_result.affected_subsystems}


@registry.register(GET_DNSMASQ_STATUS)
@@ -260,7 +270,12 @@ def set_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
            entry["dns"] = body["dns"]
        ranges.append(entry)
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "range_added", "interface": iface}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"interface": iface, "start": start, "end": end}


@@ -295,7 +310,12 @@ def remove_dhcp_range(_request: Any, body: dict[str, Any] | None) -> dict[str, A
            f"DHCP range for interface '{iface}' ({start}-{end}) not found"
        )
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "range_removed", "interface": iface}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"interface": iface, "start": start, "end": end}


@@ -334,14 +354,26 @@ def add_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str, An
            if hostname is not None:
                leases[i]["hostname"] = hostname
            _save_config(cfg)
-            refresh_state(["dnsmasq"])
+            sync_result = bus.emit(
+                SyncEvent(
+                    "dnsmasq",
+                    "config_saved",
+                    {"action": "static_lease_added", "mac": mac},
+                )
+            )
+            refresh_state(["dnsmasq", *sync_result.affected_subsystems])
            return {"mac": mac, "ip": ip, "hostname": hostname}
    entry: dict[str, Any] = {"mac": mac, "ip": ip}
    if hostname:
        entry["hostname"] = hostname
    leases.append(entry)
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "static_lease_added", "mac": mac}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"mac": mac, "ip": ip, "hostname": hostname}


@@ -366,7 +398,12 @@ def remove_static_lease(_request: Any, body: dict[str, Any] | None) -> dict[str,
    if len(cfg["dhcp"]["static_leases"]) == before:
        raise NotFoundError(f"Static lease for MAC '{mac}' not found")
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "static_lease_removed", "mac": mac}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"mac": mac}


@@ -392,14 +429,26 @@ def add_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
            if hostname is not None:
                records[i]["hostname"] = hostname
            _save_config(cfg)
-            refresh_state(["dnsmasq"])
+            sync_result = bus.emit(
+                SyncEvent(
+                    "dnsmasq",
+                    "config_saved",
+                    {"action": "dns_record_added", "name": name},
+                )
+            )
+            refresh_state(["dnsmasq", *sync_result.affected_subsystems])
            return {"name": name, "address": address, "hostname": hostname}
    entry: dict[str, Any] = {"name": name, "address": address}
    if hostname:
        entry["hostname"] = hostname
    records.append(entry)
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "dns_record_added", "name": name}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"name": name, "address": address, "hostname": hostname}


@@ -422,7 +471,12 @@ def remove_dns_record(_request: Any, body: dict[str, Any] | None) -> dict[str, A
    if len(cfg["dns"]["custom_records"]) == before:
        raise NotFoundError(f"DNS record '{name}' not found")
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "dnsmasq", "config_saved", {"action": "dns_record_removed", "name": name}
+        )
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"name": name}


@@ -438,7 +492,10 @@ def set_upstreams(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    cfg = _get_config()
    cfg["dns"]["upstreams"] = list(body["servers"])
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent("dnsmasq", "config_saved", {"action": "upstreams_set"})
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"upstreams": cfg["dns"]["upstreams"]}


@@ -455,5 +512,8 @@ def set_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    cfg = _get_config()
    cfg["dns"]["domain"] = domain if domain else None
    _save_config(cfg)
-    refresh_state(["dnsmasq"])
+    sync_result = bus.emit(
+        SyncEvent("dnsmasq", "config_saved", {"action": "domain_set"})
+    )
+    refresh_state(["dnsmasq", *sync_result.affected_subsystems])
    return {"domain": cfg["dns"]["domain"]}
@@ -43,6 +43,7 @@ from lib.firewall import (
 from lib.firewall import (
    save_backup as _save_backup,
 )
+from lib.sync import SyncEvent, bus

 logger = logging.getLogger(__name__)

@@ -350,7 +351,10 @@ def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str,
        raise ValueError("'zones' must be a dict")
    _save_config(body)
    logger.info("Firewall config saved (%d zones)", len(body["zones"]))
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent("firewall", "config_saved", {"action": "config_saved"})
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -364,7 +368,10 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    merged = deep_merge(current, body)
    _save_config(merged)
    logger.info("Firewall config patched: %s", sorted(body.keys()))
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent("firewall", "config_saved", {"action": "config_patched"})
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -378,7 +385,11 @@ def config_pending_handler(_request: Any, _body: Any) -> dict[str, Any]:
 def config_apply(_request: Any, _body: Any) -> dict[str, Any]:
    result = _config_apply()
    logger.info("Firewall config applied: %s", result.get("applied_zones", []))
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent("firewall", "config_saved", {"action": "config_applied"})
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
+    result["synced"] = sync_result.affected_subsystems
    return result


@@ -404,7 +415,12 @@ def create_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    )
    _reload()
    logger.info("Zone '%s' created (target=%s)", zone_name, target)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "zone_created", "zone": zone_name}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone_name}


@@ -419,7 +435,10 @@ def delete_zone(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    run(["firewall-cmd", f"--zone={zone}", "--delete", "--permanent"], sudo=True)
    _reload()
    logger.info("Zone '%s' deleted", zone)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent("firewall", "config_saved", {"action": "zone_deleted", "zone": zone})
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone}


@@ -489,7 +508,12 @@ def set_zone_interfaces(_request: Any, body: dict[str, Any] | None) -> dict[str,
    _save_config(cfg)

    logger.info("Zone '%s' interfaces set to %s", zone, interfaces)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "interfaces_set", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "interfaces": interfaces}


@@ -528,7 +552,10 @@ def set_zone_services(_request: Any, body: dict[str, Any] | None) -> dict[str, A
            sudo=True,
        )
    _reload()
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent("firewall", "config_saved", {"action": "services_set", "zone": zone})
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "services": services}


@@ -559,7 +586,12 @@ def add_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    entry = {"id": rule_id, "rule": rule}
    cfg["zones"][zone]["rich_rules"].append(entry)
    _save_config(cfg)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "rich_rule_added", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "id": rule_id, "rule": rule}


@@ -597,7 +629,12 @@ def remove_rich_rule(_request: Any, body: dict[str, Any] | None) -> dict[str, An
        r for r in zone_cfg.get("rich_rules", []) if r.get("id") != rule_id
    ]
    _save_config(cfg)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "rich_rule_removed", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "id": rule_id}


@@ -632,7 +669,12 @@ def set_masquerade(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
    action = "--add-masquerade" if enable else "--remove-masquerade"
    run(["firewall-cmd", f"--zone={zone}", action, "--permanent"], sudo=True)
    _reload()
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "masquerade_set", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "masquerade": bool(enable)}


@@ -675,7 +717,12 @@ def add_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str, An
    cfg.setdefault("zones", {}).setdefault(zone, {}).setdefault("forward_ports", [])
    cfg["zones"][zone]["forward_ports"].append(entry)
    _save_config(cfg)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "forward_port_added", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "id": fp_id, "port": int(port), "proto": proto}


@@ -722,7 +769,12 @@ def remove_forward_port(_request: Any, body: dict[str, Any] | None) -> dict[str,
        fp for fp in fps if not (fp.get("port") == port and fp.get("proto") == proto)
    ]
    _save_config(cfg)
-    refresh_state(["firewall"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "firewall", "config_saved", {"action": "forward_port_removed", "zone": zone}
+        )
+    )
+    refresh_state(["firewall", *sync_result.affected_subsystems])
    return {"zone": zone, "port": int(port), "proto": proto}


@@ -20,7 +20,7 @@ from daemon.iface import (
    POST_NETWORK_INTERFACE_RELOAD,
    POST_NETWORK_SYSCTL_SET,
 )
-from daemon.server import NotFoundError, registry
+from daemon.server import NotFoundError, refresh_state, registry
 from lib.common import run, validate_interface_name
 from lib.dnsmasq import set_upstreams
 from lib.network import (
@@ -34,6 +34,7 @@ from lib.network import (
    render_network_file,
    save_config,
 )
+from lib.sync import SyncEvent, bus

 logger = logging.getLogger(__name__)

@@ -194,7 +195,17 @@ def save_interface(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]
        )

    logger.info("Interface '%s' config saved (applied=%s)", name, deployed)
-    return {"name": name, "applied": deployed}
+    sync_result = bus.emit(
+        SyncEvent(
+            "network", "config_saved", {"action": "interface_saved", "interface": name}
+        )
+    )
+    refresh_state(["network", *sync_result.affected_subsystems])
+    return {
+        "name": name,
+        "applied": deployed,
+        "synced": sync_result.affected_subsystems,
+    }


@registry.register(POST_NETWORK_INTERFACE_RELOAD)
@@ -251,6 +262,11 @@ def apply_all(_request: Any, _body: Any) -> dict[str, Any]:
    except Exception:
        logger.warning("Failed to sync DNS upstreams to dnsmasq", exc_info=True)

+    sync_result = bus.emit(
+        SyncEvent("network", "config_saved", {"action": "config_applied"})
+    )
+    refresh_state(["network", *sync_result.affected_subsystems])
+
    logger.info(
        "Network config applied: %d interfaces, %d stale cleaned",
        len(generated),
@@ -260,6 +276,7 @@ def apply_all(_request: Any, _body: Any) -> dict[str, Any]:
        "applied": len(generated),
        "files": [str(p) for p in generated],
        "cleaned": [str(p) for p in cleaned],
+        "synced": sync_result.affected_subsystems,
    }


@@ -311,4 +328,8 @@ def set_sysctl(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
        )

    logger.info("sysctl %s set to %s", name, value)
+    sync_result = bus.emit(
+        SyncEvent("network", "config_saved", {"action": "sysctl_set", "name": name})
+    )
+    refresh_state(["network", *sync_result.affected_subsystems])
    return {"name": name, "value": value}
@@ -457,7 +457,12 @@ def update_domain(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:

    # Path removal: if body has `path` key (string) but no `paths`/`backend`/`headers`
    path_to_remove = body.get("path")
-    if path_to_remove is not None and "paths" not in body and "backend" not in body and "headers" not in body:
+    if (
+        path_to_remove is not None
+        and "paths" not in body
+        and "backend" not in body
+        and "headers" not in body
+    ):
        paths = entry.get("paths", {})
        if path_to_remove in paths:
            del paths[path_to_remove]
@@ -25,6 +25,7 @@ from daemon.iface import (
 )
 from daemon.server import NotFoundError, refresh_state, registry
 from lib.common import deep_merge, load_json, run, run_proc, save_json
+from lib.sync import SyncEvent, bus

 logger = logging.getLogger(__name__)

@@ -126,7 +127,10 @@ def save_config_handler(_request: Any, body: dict[str, Any] | None) -> dict[str,
    if current_key:
        body.setdefault("interface", {})["private_key"] = current_key
    _save_config(body)
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent("wireguard", "config_saved", {"action": "config_saved"})
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -146,7 +150,10 @@ def patch_config(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    current = _get_config()
    merged = deep_merge(current, body)
    _save_config(merged)
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent("wireguard", "config_saved", {"action": "config_patched"})
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    return {"config_saved": True}


@@ -167,8 +174,11 @@ def apply(_request: Any, _body: Any) -> dict[str, Any]:
    local_tmp.unlink(missing_ok=True)
    run([WG_QUICK_BIN, "up", cfg["interface"]["name"]], sudo=True)
    logger.info("WireGuard tunnel '%s' brought up", cfg["interface"]["name"])
-    refresh_state(["wireguard"])
-    return {"applied": True}
+    sync_result = bus.emit(
+        SyncEvent("wireguard", "config_saved", {"action": "config_applied"})
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
+    return {"applied": True, "synced": sync_result.affected_subsystems}


@registry.register(POST_WIREGUARD_DOWN)
@@ -178,7 +188,10 @@ def down(_request: Any, _body: Any) -> dict[str, Any]:
    name = cfg["interface"]["name"]
    run([WG_QUICK_BIN, "down", name], sudo=True)
    logger.info("WireGuard tunnel '%s' brought down", name)
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent("wireguard", "config_saved", {"action": "tunnel_down"})
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    return {"down": True}


@@ -205,7 +218,10 @@ def initialize(_request: Any, _body: Any) -> dict[str, Any]:
    cfg["interface"]["public_key"] = public_key
    _save_config(cfg)
    logger.info("WireGuard initialised (pubkey=%s...)", public_key[:16])
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent("wireguard", "config_saved", {"action": "initialized"})
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    safe = dict(cfg)
    safe["interface"] = dict(safe["interface"])
    safe["interface"].pop("private_key", None)
@@ -235,6 +251,7 @@ def add_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
        if body.get("preshared_key") is not None:
            peer["preshared_key"] = body["preshared_key"]
        logger.info("WireGuard peer '%s' updated", name)
+        _peer_action = "peer_updated"
    else:
        res = run_proc([WG_BIN, "genkey"], sudo=True)
        priv = res.stdout.strip()
@@ -249,8 +266,14 @@ def add_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
            "preshared_key": body.get("preshared_key"),
        }
        logger.info("WireGuard peer '%s' added", name)
+        _peer_action = "peer_added"
    _save_config(cfg)
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "wireguard", "config_saved", {"action": _peer_action, "peer_name": name}
+        )
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    peer_out = dict(peers[name])
    peer_out.pop("private_key", None)
    return peer_out
@@ -276,7 +299,12 @@ def remove_peer(_request: Any, body: dict[str, Any] | None) -> dict[str, Any]:
    del peers[name]
    _save_config(cfg)
    logger.info("WireGuard peer '%s' removed", name)
-    refresh_state(["wireguard"])
+    sync_result = bus.emit(
+        SyncEvent(
+            "wireguard", "config_saved", {"action": "peer_removed", "peer_name": name}
+        )
+    )
+    refresh_state(["wireguard", *sync_result.affected_subsystems])
    return {"name": name}