Reconcile daemon with production runtime state

daemon/registry.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 
 import importlib.util
+import inspect
 import json
 import os
 from pathlib import Path
@@ -55,6 +56,18 @@ class Plugin:
             self.manifest.get("required_config", [])
         )
 
+    @property
+    def optional_config(self):
+        return list(
+            self.manifest.get("optional_config", [])
+        )
+
+    @property
+    def config_keys(self):
+        return list(dict.fromkeys(
+            self.required_config + self.optional_config
+        ))
+
     def validate_config(self, config):
         missing = [
             key
@@ -68,7 +81,7 @@ class Plugin:
                 + ", ".join(missing)
             )
 
-    def execute(self, action, config):
+    def execute(self, action, config, runtime_state):
         if action not in self.actions:
             raise RegistryError(
                 f"plugin {self.name} does not support action {action}"
@@ -82,6 +95,14 @@ class Plugin:
                 "but does not implement it"
             )
 
+        # Backward-compatible generic state injection. Existing one-argument
+        # plugins continue to receive only config; state-aware plugins can
+        # declare a second positional parameter named runtime_state.
+        parameters = list(inspect.signature(handler).parameters.values())
+
+        if len(parameters) >= 2:
+            return handler(config, runtime_state)
+
         return handler(config)
 
 
@@ -91,6 +112,11 @@ class Resource:
         self.plugin = plugin
         self.config = config
 
+        # Generic daemon-lifetime state storage. The daemon does not interpret
+        # plugin keys or values; applications may use this for transition
+        # history, grace windows, caches, or other resource-local state.
+        self.runtime_state = {}
+
         self.plugin.validate_config(config)
 
     def describe(self):
@@ -136,7 +162,7 @@ class Registry:
 
         config = {
             key: os.environ.get(key)
-            for key in plugin.required_config
+            for key in plugin.config_keys
         }
 
         self.resources[resource_id] = Resource(
@@ -163,6 +189,7 @@ class Registry:
         result = resource.plugin.execute(
             action,
             resource.config,
+            resource.runtime_state,
         )
 
         return {

plugins/postgresql/manifest.json

@@ -1,15 +1,17 @@
 {
   "name": "postgresql",
-  "version": 1,
+  "version": 3,
   "module": "plugin.py",
-  "actions": [
+  "actions": ["monitor"],
-    "monitor"
-  ],
   "required_config": [
     "PGHOST",
     "PGPORT",
     "PGUSER",
     "PGDATABASE",
-    "PGPASSWORD_FILE"
+    "PGPASSWORD_FILE",
+    "PG_REPLICATION_PEERS"
+  ],
+  "optional_config": [
+    "PG_RECONNECT_GRACE_SECONDS"
   ]
 }

plugins/postgresql/plugin.py

@@ -1,19 +1,248 @@
 #!/usr/bin/env python3
 
+import time
+
 import psycopg2
 import psycopg2.extras
 
+OCF_SUCCESS = 0
+OCF_ERR_GENERIC = 1
+OCF_NOT_RUNNING = 7
+OCF_RUNNING_PROMOTED = 8
+
 
 def read_secret(path):
     with open(path, "r", encoding="utf-8") as f:
         return f.read().strip()
 
 
-def monitor(config):
+def parse_replication_peers(value):
-    password = read_secret(
+    peers = []
-        config["PGPASSWORD_FILE"]
+    for item in (value or "").split(","):
+        item = item.strip()
+        if not item:
+            continue
+        name, sep, endpoint = item.partition("=")
+        if not sep:
+            raise ValueError(f"invalid replication peer entry: {item!r}")
+        host, sep, port = endpoint.rpartition(":")
+        if not sep or not host or not port:
+            raise ValueError(f"invalid replication peer endpoint: {endpoint!r}")
+        peers.append({
+            "name": name.strip(),
+            "host": host.strip(),
+            "port": int(port),
+        })
+    return peers
+
+
+def probe_replication_peers(config, password):
+    results = []
+    writable_primaries = []
+
+    for peer in parse_replication_peers(config["PG_REPLICATION_PEERS"]):
+        result = dict(peer)
+        try:
+            conn = psycopg2.connect(
+                host=peer["host"],
+                port=peer["port"],
+                user=config["PGUSER"],
+                dbname=config["PGDATABASE"],
+                password=password,
+                connect_timeout=2,
+                application_name="managed-containerd-peer-probe",
+            )
+            try:
+                conn.autocommit = True
+                with conn.cursor() as cur:
+                    cur.execute("SELECT pg_is_in_recovery()")
+                    in_recovery = bool(cur.fetchone()[0])
+                result.update({
+                    "reachable": True,
+                    "in_recovery": in_recovery,
+                    "writable_primary": not in_recovery,
+                })
+                if not in_recovery:
+                    writable_primaries.append(peer["name"])
+            finally:
+                conn.close()
+        except psycopg2.Error as exc:
+            result.update({
+                "reachable": False,
+                "in_recovery": None,
+                "writable_primary": False,
+                "error": str(exc).splitlines()[0],
+            })
+        results.append(result)
+
+    return {
+        "peers": results,
+        "writable_primaries": writable_primaries,
+        "writable_primary_count": len(writable_primaries),
+    }
+
+
+def reconnect_grace_seconds(config):
+    raw = config.get("PG_RECONNECT_GRACE_SECONDS") or "15"
+    try:
+        value = float(raw)
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"invalid PG_RECONNECT_GRACE_SECONDS: {raw!r}"
         )
 
+    if value < 0:
+        raise ValueError(
+            "PG_RECONNECT_GRACE_SECONDS must be >= 0"
+        )
+
+    return value
+
+
+def _postgres_runtime_state(runtime_state):
+    return runtime_state.setdefault("postgresql", {})
+
+
+def _set_transition_state(runtime_state, state):
+    pg_state = _postgres_runtime_state(runtime_state)
+    previous = pg_state.get("state")
+
+    if previous != state:
+        pg_state["state"] = state
+        pg_state["state_changed_at"] = time.monotonic()
+
+    return pg_state
+
+
+def _clear_reconnect_grace(runtime_state):
+    pg_state = _postgres_runtime_state(runtime_state)
+    pg_state.pop("reconnect_started_at", None)
+
+
+def normalize_ha_state(
+    *,
+    config,
+    runtime_state,
+    in_recovery,
+    streaming,
+    receiver,
+    recovery,
+    peer_state=None,
+):
+    """Translate only validated PostgreSQL states into OCF semantics."""
+    if not in_recovery:
+        _clear_reconnect_grace(runtime_state)
+        _set_transition_state(runtime_state, "primary")
+        return {
+            "health": "healthy",
+            "ha_role": "promoted",
+            "state": "primary",
+            "ocf_rc": OCF_RUNNING_PROMOTED,
+            "promotion_eligible": True,
+        }
+
+    replay_paused = bool((recovery or {}).get("replay_paused"))
+    if streaming and not replay_paused:
+        _clear_reconnect_grace(runtime_state)
+        _set_transition_state(runtime_state, "streaming")
+        return {
+            "health": "healthy",
+            "ha_role": "unpromoted",
+            "state": "streaming",
+            "ocf_rc": OCF_SUCCESS,
+            "promotion_eligible": True,
+        }
+
+    receiver_status = (receiver or {}).get("status")
+    state = receiver_status or "not_streaming"
+    writable_primary_count = None
+
+    if peer_state is not None:
+        writable_primary_count = peer_state.get(
+            "writable_primary_count"
+        )
+
+    # Validated against the authoritative RA on 2026-09-17: if the local
+    # server remains a reachable standby, its WAL receiver is not streaming,
+    # and no writable primary is reachable, the RA returns OCF_SUCCESS and
+    # keeps the node as a reduced-priority election candidate.
+    if state == "not_streaming" and writable_primary_count == 0:
+        _clear_reconnect_grace(runtime_state)
+        _set_transition_state(runtime_state, "disconnected_candidate")
+        return {
+            "health": "degraded",
+            "ha_role": "unpromoted",
+            "state": "disconnected_candidate",
+            "ocf_rc": OCF_SUCCESS,
+            "promotion_eligible": True,
+        }
+
+    # Reconnect grace is PostgreSQL-specific policy layered on generic daemon
+    # runtime state. It is entered only after this daemon has actually observed
+    # disconnected_candidate and then sees exactly one writable primary return
+    # while the local WAL receiver has not yet resumed streaming.
+    if state == "not_streaming" and writable_primary_count == 1:
+        pg_state = _postgres_runtime_state(runtime_state)
+        previous_state = pg_state.get("state")
+        started_at = pg_state.get("reconnect_started_at")
+        now = time.monotonic()
+        grace = reconnect_grace_seconds(config)
+
+        if previous_state == "disconnected_candidate":
+            started_at = now
+            pg_state["reconnect_started_at"] = started_at
+
+        if started_at is not None:
+            elapsed = max(0.0, now - float(started_at))
+
+            if elapsed < grace:
+                _set_transition_state(runtime_state, "reconnecting")
+                return {
+                    "health": "degraded",
+                    "ha_role": "unpromoted",
+                    "state": "reconnecting",
+                    "ocf_rc": OCF_SUCCESS,
+                    "promotion_eligible": False,
+                    "grace": {
+                        "seconds": grace,
+                        "elapsed_seconds": round(elapsed, 3),
+                        "remaining_seconds": round(
+                            max(0.0, grace - elapsed), 3
+                        ),
+                    },
+                }
+
+            _clear_reconnect_grace(runtime_state)
+
+        # Already validated for the steady-state failure case: a reachable
+        # standby whose WAL receiver is not streaming while exactly one remote
+        # writable primary is available returns OCF_ERR_GENERIC.
+        _set_transition_state(runtime_state, "not_streaming")
+        return {
+            "health": "degraded",
+            "ha_role": "unpromoted",
+            "state": "not_streaming",
+            "ocf_rc": OCF_ERR_GENERIC,
+            "promotion_eligible": False,
+        }
+
+    # Multiple writable primaries and other degraded states remain deliberately
+    # unclassified until independently validated.
+    _clear_reconnect_grace(runtime_state)
+    _set_transition_state(runtime_state, state)
+    return {
+        "health": "degraded",
+        "ha_role": "unpromoted",
+        "state": state,
+        "ocf_rc": None,
+        "promotion_eligible": False,
+    }
+
+
+def monitor(config, runtime_state):
+    password = read_secret(config["PGPASSWORD_FILE"])
+
+    try:
         conn = psycopg2.connect(
             host=config["PGHOST"],
             port=int(config["PGPORT"]),
@@ -23,6 +252,37 @@ def monitor(config):
             connect_timeout=5,
             application_name="managed-containerd",
         )
+    except psycopg2.OperationalError as exc:
+        message = str(exc)
+
+        # Validated against the authoritative RA on 2026-09-17:
+        # an explicitly refused local PostgreSQL connection corresponds to
+        # OCF_NOT_RUNNING.  Other OperationalError classes remain unclassified.
+        if "connection refused" in message.lower():
+            _clear_reconnect_grace(runtime_state)
+            _set_transition_state(runtime_state, "not_running")
+            return {
+                "ha": {
+                    "health": "down",
+                    "ha_role": "unknown",
+                    "state": "not_running",
+                    "ocf_rc": OCF_NOT_RUNNING,
+                    "promotion_eligible": False,
+                },
+                "role": "unknown",
+                "in_recovery": None,
+                "streaming": False,
+                "wal_receiver": None,
+                "recovery": None,
+                "primary": None,
+                "server": None,
+                "error": {
+                    "kind": "connection_refused",
+                    "message": message,
+                },
+            }
+
+        raise
 
     try:
         conn.autocommit = True
@@ -30,38 +290,27 @@ def monitor(config):
         with conn.cursor(
             cursor_factory=psycopg2.extras.RealDictCursor
         ) as cur:
-
             cur.execute("""
                 SELECT pg_is_in_recovery() AS in_recovery
             """)
-
+            in_recovery = bool(cur.fetchone()["in_recovery"])
-            in_recovery = bool(
-                cur.fetchone()["in_recovery"]
-            )
 
             if not in_recovery:
                 cur.execute("""
                     SELECT
-                        pg_current_wal_lsn()::text
+                        pg_current_wal_lsn()::text AS current_wal_lsn
-                            AS current_wal_lsn
                 """)
-
                 primary = cur.fetchone()
-
             else:
                 primary = None
 
             if in_recovery:
                 cur.execute("""
                     SELECT
-                        pg_last_wal_receive_lsn()::text
+                        pg_last_wal_receive_lsn()::text AS receive_lsn,
-                            AS receive_lsn,
+                        pg_last_wal_replay_lsn()::text AS replay_lsn,
-                        pg_last_wal_replay_lsn()::text
+                        pg_is_wal_replay_paused() AS replay_paused
-                            AS replay_lsn,
-                        pg_is_wal_replay_paused()
-                            AS replay_paused
                 """)
-
                 recovery = cur.fetchone()
 
                 cur.execute("""
@@ -70,68 +319,62 @@ def monitor(config):
                         sender_host,
                         sender_port,
                         slot_name,
-                        latest_end_lsn::text
+                        latest_end_lsn::text AS latest_end_lsn
-                            AS latest_end_lsn
                     FROM pg_stat_wal_receiver
                     LIMIT 1
                 """)
-
                 receiver = cur.fetchone()
-
             else:
                 recovery = {
                     "receive_lsn": None,
                     "replay_lsn": None,
                     "replay_paused": False,
                 }
-
                 receiver = None
 
             cur.execute("""
                 SELECT
-                    current_setting('server_version')
+                    current_setting('server_version') AS server_version,
-                        AS server_version,
+                    current_setting('cluster_name', true) AS cluster_name,
-                    current_setting('cluster_name', true)
-                        AS cluster_name,
                     current_user AS current_user,
-                    current_database()
+                    current_database() AS current_database,
-                        AS current_database,
+                    inet_server_addr()::text AS server_address,
-                    inet_server_addr()::text
+                    inet_server_port() AS server_port
-                        AS server_address,
-                    inet_server_port()
-                        AS server_port
             """)
-
             server = cur.fetchone()
 
         if in_recovery:
             role = "standby"
-
             streaming = bool(
-                receiver
+                receiver and receiver.get("status") == "streaming"
-                and receiver.get("status")
-                    == "streaming"
             )
-
         else:
             role = "primary"
             streaming = False
 
+        peer_state = None
+        if in_recovery and not streaming:
+            peer_state = probe_replication_peers(config, password)
+
+        ha = normalize_ha_state(
+            config=config,
+            runtime_state=runtime_state,
+            in_recovery=in_recovery,
+            streaming=streaming,
+            receiver=receiver,
+            recovery=recovery,
+            peer_state=peer_state,
+        )
+
         return {
+            "ha": ha,
+            "peer_state": peer_state,
             "role": role,
             "in_recovery": in_recovery,
             "streaming": streaming,
-            "wal_receiver": (
+            "wal_receiver": dict(receiver) if receiver else None,
-                dict(receiver)
-                if receiver
-                else None
-            ),
             "recovery": dict(recovery),
-            "primary": (
+            "primary": dict(primary) if primary else None,
-                dict(primary)
-                if primary
-                else None
-            ),
             "server": dict(server),
         }
 

postgres.env

@@ -7,8 +7,10 @@ PLUGIN=postgresql
 MANAGED_CONTAINERD_SOCKET=/run/managed-containerd/managed-containerd.sock
 MANAGED_CONTAINERD_PLUGIN_ROOT=/opt/managed-containerd/plugins
 
-PGHOST=192.168.1.4
+PGHOST=postgres
 PGPORT=5432
 PGUSER=david
 PGDATABASE=david
 PGPASSWORD_FILE=/run/secrets/postgres_password
+PG_REPLICATION_PEERS=david-prodesk=192.168.1.4:5432,david-macbookpro=192.168.1.5:5432,david-eve=192.168.1.3:5432
+PG_RECONNECT_GRACE_SECONDS=15

tests/test-runtime-state.py

@@ -0,0 +1,234 @@
+#!/usr/bin/env python3
+
+import ast
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, "daemon")
+
+from registry import Plugin, Registry, RegistryError, Resource
+
+
+def require(condition, message):
+    if not condition:
+        raise SystemExit("FAIL: " + message)
+
+
+def write_plugin(
+    root,
+    name,
+    manifest,
+    source,
+):
+    plugin_root = Path(root) / name
+    plugin_root.mkdir()
+
+    (plugin_root / "manifest.json").write_text(
+        json.dumps(manifest)
+    )
+    (plugin_root / "plugin.py").write_text(source)
+
+
+print("===== GENERIC PLUGIN RUNTIME STATE =====")
+
+with tempfile.TemporaryDirectory() as temporary_root:
+    stateful_manifest = {
+        "name": "stateful",
+        "version": 1,
+        "module": "plugin.py",
+        "actions": ["monitor"],
+        "required_config": ["REQUIRED_VALUE"],
+        "optional_config": ["OPTIONAL_VALUE"],
+    }
+
+    write_plugin(
+        temporary_root,
+        "stateful",
+        stateful_manifest,
+        """
+def monitor(config, runtime_state):
+    counter = runtime_state.get("counter", 0) + 1
+    runtime_state["counter"] = counter
+
+    return {
+        "counter": counter,
+        "required": config["REQUIRED_VALUE"],
+        "optional": config["OPTIONAL_VALUE"],
+    }
+""",
+    )
+
+    environment = {
+        "MANAGED_CONTAINERD_PLUGIN_ROOT": (
+            temporary_root
+        ),
+        "RESOURCE_ID": "stateful-resource",
+        "PLUGIN": "stateful",
+        "REQUIRED_VALUE": "required",
+        "OPTIONAL_VALUE": "optional",
+    }
+
+    with patch.dict(
+        os.environ,
+        environment,
+        clear=False,
+    ):
+        registry = Registry()
+
+    first = registry.execute(
+        "stateful-resource",
+        "monitor",
+    )
+    second = registry.execute(
+        "stateful-resource",
+        "monitor",
+    )
+
+    require(
+        first["result"]["counter"] == 1,
+        "first stateful invocation has wrong counter",
+    )
+    require(
+        second["result"]["counter"] == 2,
+        "runtime state did not persist",
+    )
+    require(
+        second["result"]["required"] == "required",
+        "required configuration was not injected",
+    )
+    require(
+        second["result"]["optional"] == "optional",
+        "optional configuration was not injected",
+    )
+
+    resource = registry.resources[
+        "stateful-resource"
+    ]
+
+    require(
+        resource.runtime_state == {"counter": 2},
+        "resource runtime state has unexpected contents",
+    )
+
+    print("stateful_handler       persisted")
+    print("optional_config        injected")
+
+    legacy_manifest = {
+        "name": "legacy",
+        "version": 1,
+        "module": "plugin.py",
+        "actions": ["monitor"],
+        "required_config": ["REQUIRED_VALUE"],
+    }
+
+    write_plugin(
+        temporary_root,
+        "legacy",
+        legacy_manifest,
+        """
+def monitor(config):
+    return {
+        "legacy": True,
+        "required": config["REQUIRED_VALUE"],
+    }
+""",
+    )
+
+    legacy_plugin = Plugin(
+        temporary_root,
+        "legacy",
+    )
+
+    legacy_result = legacy_plugin.execute(
+        "monitor",
+        {"REQUIRED_VALUE": "required"},
+        {},
+    )
+
+    require(
+        legacy_result == {
+            "legacy": True,
+            "required": "required",
+        },
+        "one-argument plugin compatibility failed",
+    )
+
+    print("legacy_handler         compatible")
+
+    try:
+        Resource(
+            "invalid-resource",
+            legacy_plugin,
+            {"REQUIRED_VALUE": None},
+        )
+    except RegistryError:
+        print("required_config        enforced")
+    else:
+        raise SystemExit(
+            "FAIL: missing required configuration accepted"
+        )
+
+production_manifest = json.loads(
+    Path(
+        "plugins/postgresql/manifest.json"
+    ).read_text()
+)
+
+require(
+    production_manifest.get("version") == 3,
+    "PostgreSQL manifest version is not 3",
+)
+require(
+    "PG_REPLICATION_PEERS"
+    in production_manifest.get(
+        "required_config",
+        [],
+    ),
+    "replication peers are not required",
+)
+require(
+    "PG_RECONNECT_GRACE_SECONDS"
+    in production_manifest.get(
+        "optional_config",
+        [],
+    ),
+    "reconnect grace is not optional",
+)
+
+plugin_tree = ast.parse(
+    Path(
+        "plugins/postgresql/plugin.py"
+    ).read_text()
+)
+
+monitor_functions = [
+    node
+    for node in plugin_tree.body
+    if isinstance(node, ast.FunctionDef)
+    and node.name == "monitor"
+]
+
+require(
+    len(monitor_functions) == 1,
+    "PostgreSQL monitor entry point is ambiguous",
+)
+
+monitor_arguments = [
+    argument.arg
+    for argument
+    in monitor_functions[0].args.args
+]
+
+require(
+    monitor_arguments[:2]
+    == ["config", "runtime_state"],
+    "PostgreSQL monitor does not accept runtime state",
+)
+
+print("postgresql_contract    validated")
+print()
+print("RUNTIME_STATE_TEST_PASS=1")