Add daemon PostgreSQL notify action

plugins/postgresql/manifest.json

@@ -2,7 +2,10 @@
   "name": "postgresql",
   "version": 3,
   "module": "plugin.py",
-  "actions": ["monitor"],
+  "actions": [
+    "monitor",
+    "notify"
+  ],
   "required_config": [
     "PGHOST",
     "PGPORT",

plugins/postgresql/plugin.py

@@ -380,3 +380,58 @@ def monitor(config, runtime_state):
 
     finally:
         conn.close()
+
+
+NOTIFY_FIELDS = (
+    "type",
+    "operation",
+    "promote_uname",
+    "demote_uname",
+    "promoted_uname",
+    "unpromoted_uname",
+)
+
+
+def _notify_value(parameters, field):
+    value = parameters.get(field, "")
+
+    if not isinstance(value, str):
+        raise ValueError(
+            f"notify field {field} must be a string"
+        )
+
+    if any(char in value for char in ("\n", "\r", "\x00")):
+        raise ValueError(
+            f"notify field {field} contains invalid characters"
+        )
+
+    return value.strip()
+
+
+def notify(config, runtime_state, request_parameters):
+    """
+    Record Pacemaker notification metadata in daemon runtime state.
+
+    The daemon remains the local authority for this resource instance.
+    Node-target interpretation is deliberately deferred until the state
+    persistence and demotion action are migrated together.
+    """
+    if not isinstance(request_parameters, dict):
+        raise ValueError(
+            "notify parameters must be a JSON object"
+        )
+
+    state = {
+        field: _notify_value(request_parameters, field)
+        for field in NOTIFY_FIELDS
+    }
+
+    state["received_at"] = time.time()
+
+    postgresql_state = _postgres_runtime_state(runtime_state)
+    postgresql_state["notification"] = state
+
+    return {
+        "ocf_rc": OCF_SUCCESS,
+        "state": dict(state),
+    }

tests/test-notify-action.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+
+fake_psycopg2 = types.ModuleType("psycopg2")
+fake_psycopg2.Error = Exception
+fake_psycopg2.OperationalError = OSError
+
+fake_extras = types.ModuleType("psycopg2.extras")
+fake_extras.RealDictCursor = object
+fake_psycopg2.extras = fake_extras
+
+sys.modules["psycopg2"] = fake_psycopg2
+sys.modules["psycopg2.extras"] = fake_extras
+
+plugin_path = Path("plugins/postgresql/plugin.py")
+
+spec = importlib.util.spec_from_file_location(
+    "postgresql_plugin",
+    plugin_path,
+)
+plugin = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(plugin)
+
+
+def require(condition, message):
+    if not condition:
+        raise SystemExit("FAIL: " + message)
+
+
+print("===== DAEMON NOTIFY ACTION =====")
+
+runtime_state = {}
+
+parameters = {
+    "type": "pre",
+    "operation": "demote",
+    "promote_uname": "david-eve",
+    "demote_uname": "david-prodesk",
+    "promoted_uname": "david-prodesk",
+    "unpromoted_uname": "david-eve david-macbookpro",
+}
+
+response = plugin.notify(
+    {},
+    runtime_state,
+    parameters,
+)
+
+require(
+    response["ocf_rc"] == plugin.OCF_SUCCESS,
+    "notify did not return OCF_SUCCESS",
+)
+require(
+    response["state"]["operation"] == "demote",
+    "notify operation was not preserved",
+)
+require(
+    runtime_state["postgresql"]["notification"]
+    == response["state"],
+    "notify state was not retained",
+)
+print("valid_notify            accepted")
+print("notify_state             retained")
+
+for invalid in (
+    [],
+    {"type": None},
+    {"operation": "demote\n"},
+):
+    try:
+        plugin.notify({}, {}, invalid)
+    except ValueError:
+        continue
+
+    raise SystemExit(
+        "FAIL: invalid notify parameters were accepted"
+    )
+
+print("invalid_notify           rejected")
+print()
+print("NOTIFY_ACTION_TEST_PASS=1")