fix(notifications): report a cluster node only when it stays away

A node that leaves the corosync membership is reported after 60 seconds
if it has not come back. A quicker return, such as a corosync restart
during a package update, is one notice under Node reconnected that says
how long the node was away. Only membership lines name a node, so
placeholders like lost or unknown no longer appear, and the peers a node
sees after restarting its own corosync are not reported as reconnected.

.github/scripts/tests/test_fencing_wording.py

@@ -111,9 +111,12 @@ class FencingWordingTests(unittest.TestCase):
                 rendered = self.render(args[0], args[2], lang)
                 body = catalog['runtime']['notifications']['templates']['split_brain']['body']
                 self.assertEqual(rendered['body'], body.replace('{reason}', message), lang)
-        journal = Journal()
-        method(journal, 'quorum lost', 'syslog')
-        self.assertEqual(journal.events[0][0][0], 'node_disconnect')
+        # Losing quorum is this host's state, not a node that left: the
+        # membership lines name the nodes that left (issue #605).
+        for message in ('quorum lost', '[status] notice: node lost quorum'):
+            journal = Journal()
+            method(journal, message, 'pmxcfs')
+            self.assertEqual(journal.events, [])
 
     def test_english_family_and_settings_label_align_without_changing_key(self):
         en = self.catalogs['en']

AppImage/messages/de/common.json

@@ -6992,6 +6992,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Der Knoten {node_name} hat seinen Clusterdienst neu gestartet und ist nach {seconds} s wieder beigetreten."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} geändert zu {current}{entity_suffix}",

AppImage/messages/es/common.json

@@ -7145,6 +7145,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "El nodo {node_name} reinició su servicio de clúster y volvió a unirse tras {seconds} s."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} cambió a {current}{entity_suffix}",

AppImage/messages/fr/common.json

@@ -6992,6 +6992,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Le nœud {node_name} a redémarré son service de cluster et l'a rejoint après {seconds} s."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname} : {category} remplacé par {current}{entity_suffix}",

AppImage/messages/it/common.json

@@ -6992,6 +6992,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Il nodo {node_name} ha riavviato il servizio del cluster ed è rientrato dopo {seconds} s."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} cambiato in {current}{entity_suffix}",

AppImage/messages/pt/common.json

@@ -6992,6 +6992,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "O nó {node_name} reiniciou o serviço do cluster e voltou a entrar após {seconds} s."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} alterado para {current}{entity_suffix}",

AppImage/messages/ru/common.json

@@ -7144,6 +7144,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Узел {node_name} перезапустил службу кластера и снова присоединился через {seconds} с."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} — новое состояние: {current}{entity_suffix}",

AppImage/messages/sk/common.json

@@ -6991,6 +6991,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Uzol {node_name} reštartoval službu klastra a po {seconds} s sa znova pripojil."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} sa zmenilo na {current}{entity_suffix}",

AppImage/messages/sv/common.json

@@ -6992,6 +6992,9 @@
   },
   "runtime": {
     "notifications": {
+      "cluster": {
+        "rejoinedBody": "Noden {node_name} startade om sin klustertjänst och anslöt igen efter {seconds} s."
+      },
       "templates": {
         "state_change": {
           "title": "{hostname}: {category} har ändrats till {current}{entity_suffix}",

AppImage/scripts/notification_events.py

@@ -35,6 +35,10 @@ from proxmox_known_errors import analyze_oom_event, format_oom_diagnosis
 # This provides a single source of truth for all grace period logic
 import startup_grace
 
+# Serializes JournalWatcher emits: held cluster events fire from a timer.
+_JOURNAL_EMIT_LOCK = threading.RLock()
+
+
 class _SharedState:
     """Wrapper around centralized startup_grace module for backwards compatibility.
     
@@ -1872,14 +1876,18 @@ class JournalWatcher:
         # the surviving nodes when a peer leaves or comes back:
         #   [TOTEM ] A new membership (1.55) was formed. Members left: 2
         #   [TOTEM ] A new membership (1.5a) was formed. Members joined: 2
-        # The line carries node IDs and no "node" keyword, so the generic
-        # match below never fired for a plain node loss in a quorate cluster.
+        # Only these lines identify a node. pmxcfs' "node lost quorum" is this
+        # host losing quorum while a peer is away, never a node of that name.
         if 'a new membership' in msg_lower and 'was formed' in msg_lower:
             names = self._corosync_node_names()
             local_name = socket.gethostname().split('.')[0]
-            for keyword, event_type, severity in (
-                    ('left', 'node_disconnect', 'CRITICAL'),
-                    ('joined', 'node_reconnect', 'INFO')):
+            local_ids = {node_id for node_id, name in names.items() if name == local_name}
+            joined = re.search(r'\bjoined:((?:\s+\d+)+)', msg)
+            if joined and local_ids & set(joined.group(1).split()):
+                # Our own corosync (re)started: the peers it now sees never
+                # left the cluster, so they are not reconnections.
+                self._cluster_own_join_at = time.time()
+            for keyword, event_type in (('left', 'node_disconnect'), ('joined', 'node_reconnect')):
                 id_match = re.search(r'\b%s:((?:\s+\d+)+)' % keyword, msg)
                 if not id_match:
                     continue
@@ -1887,28 +1895,64 @@ class JournalWatcher:
                     node_name = names.get(node_id, node_id)
                     if node_name == local_name:
                         continue  # our own corosync (re)starting
-                    if (event_type == 'node_reconnect'
-                            and startup_grace.is_real_system_boot()
-                            and _shared_state.is_startup_health_grace()):
-                        continue  # this host just booted: it is us joining them
-                    self._emit(event_type, severity, {
-                        'node_name': node_name,
-                        'hostname': self._hostname,
-                    }, entity='cluster', entity_id=node_name)
+                    if event_type == 'node_disconnect':
+                        self._hold_node_disconnect(node_name)
+                    else:
+                        self._node_rejoined(node_name)
             return
 
-        # Node disconnect
-        if (('quorum' in msg_lower and 'lost' in msg_lower) or
-            ('node' in msg_lower and any(w in msg_lower for w in ['left', 'offline', 'lost']))):
-            
-            node_match = re.search(r'[Nn]ode\s+(\S+)', msg)
-            node_name = node_match.group(1) if node_match else 'unknown'
-            
-            self._emit('node_disconnect', 'CRITICAL', {
+    # A corosync restart (package update, maintenance) takes a node out of the
+    # membership for a few seconds. Report a departure only when the node is
+    # still away after this grace; a quick return is reported as one rejoin.
+    CLUSTER_LEAVE_GRACE_SECONDS = 60
+    CLUSTER_OWN_JOIN_WINDOW_SECONDS = 30
+
+    def _cluster_pending(self) -> Dict[str, Any]:
+        if not hasattr(self, '_cluster_pending_nodes'):
+            self._cluster_pending_nodes = {}
+        return self._cluster_pending_nodes
+
+    def _hold_node_disconnect(self, node_name: str):
+        pending = self._cluster_pending()
+        with _JOURNAL_EMIT_LOCK:
+            if node_name in pending:
+                return
+            timer = threading.Timer(self.CLUSTER_LEAVE_GRACE_SECONDS,
+                                    self._report_node_disconnect, args=(node_name,))
+            timer.daemon = True
+            pending[node_name] = (time.time(), timer)
+        timer.start()
+
+    def _report_node_disconnect(self, node_name: str):
+        with _JOURNAL_EMIT_LOCK:
+            if self._cluster_pending().pop(node_name, None) is None:
+                return  # it came back in time
+            self._emit_unlocked('node_disconnect', 'CRITICAL', {
                 'node_name': node_name,
                 'hostname': self._hostname,
             }, entity='cluster', entity_id=node_name)
-    
+
+    def _node_rejoined(self, node_name: str):
+        with _JOURNAL_EMIT_LOCK:
+            held = self._cluster_pending().pop(node_name, None)
+        if held is not None:
+            left_at, timer = held
+            timer.cancel()
+            self._emit('node_reconnect', 'INFO', {
+                'node_name': node_name,
+                'downtime_seconds': max(1, int(round(time.time() - left_at))),
+                'hostname': self._hostname,
+            }, entity='cluster', entity_id=node_name)
+            return
+        if time.time() - getattr(self, '_cluster_own_join_at', 0) < self.CLUSTER_OWN_JOIN_WINDOW_SECONDS:
+            return  # we joined them, they did not come back
+        if startup_grace.is_real_system_boot() and _shared_state.is_startup_health_grace():
+            return  # this host just booted: it is us joining them
+        self._emit('node_reconnect', 'INFO', {
+            'node_name': node_name,
+            'hostname': self._hostname,
+        }, entity='cluster', entity_id=node_name)
+
     def _corosync_node_names(self) -> Dict[str, str]:
         """Map corosync node IDs to node names from corosync.conf."""
         for path in ('/etc/pve/corosync.conf', '/etc/corosync/corosync.conf'):
@@ -2022,6 +2066,12 @@ class JournalWatcher:
     def _emit(self, event_type: str, severity: str, data: Dict,
               entity: str = 'node', entity_id: str = ''):
         """Emit event to queue with short-term deduplication (30s window)."""
+        # A held cluster event is emitted from its timer thread.
+        with _JOURNAL_EMIT_LOCK:
+            self._emit_unlocked(event_type, severity, data, entity, entity_id)
+
+    def _emit_unlocked(self, event_type: str, severity: str, data: Dict,
+                       entity: str = 'node', entity_id: str = ''):
         event = NotificationEvent(
             event_type, severity, data, source='journal',
             entity=entity, entity_id=entity_id,

AppImage/scripts/notification_templates.py

@@ -1967,6 +1967,13 @@ def render_template(event_type: str, data: Dict[str, Any],
         )
         if localized:
             template[field] = localized
+    if event_type == 'node_reconnect' and data.get('downtime_seconds'):
+        # A node that left and came back within the grace: one notice.
+        rejoined = runtime_message('cluster.rejoinedBody', language,
+                                   node_name=data.get('node_name', ''),
+                                   seconds=data['downtime_seconds'])
+        if rejoined:
+            template['body'] = rejoined
     backup_title_target = ''
     if event_type == 'backup_complete':
         template['title'] = runtime_message('backup.unconfirmedTitle', language,

AppImage/scripts/tests/test_cluster_membership_grace.py

@@ -0,0 +1,146 @@
+"""A corosync restart is not a node failure: departures are held for a grace,
+a quick return is one rejoin notice, and only membership lines name a node.
+The journal lines below were captured from a lab cluster during
+`systemctl restart corosync` on one of its nodes."""
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import notification_events as ne  # noqa: E402
+import notification_templates as nt  # noqa: E402
+
+NAMES = {'1': 'pmxlab1', '2': 'pmxlab2', '3': 'pmxlab3'}
+
+
+class Clock:
+    def __init__(self):
+        self.now = 1_000_000.0
+
+    def __call__(self):
+        return self.now
+
+
+class FakeTimer:
+    started = []
+
+    def __init__(self, interval, function, args=()):
+        self.interval, self.function, self.args = interval, function, args
+        self.cancelled = False
+        self.daemon = False
+
+    def start(self):
+        FakeTimer.started.append(self)
+
+    def cancel(self):
+        self.cancelled = True
+
+
+class MembershipGraceTests(unittest.TestCase):
+    def watcher(self, hostname):
+        sub = type('Watcher', (ne.JournalWatcher,), {'_hostname': hostname})
+        watcher = object.__new__(sub)
+        watcher._recent_events = {}
+        watcher._dedup_window = 30
+        watcher.events = []
+
+        class Queue:
+            def put(self, event):
+                watcher.events.append((event.event_type, event.severity, dict(event.data)))
+        watcher._queue = Queue()
+        watcher._corosync_node_names = lambda: NAMES
+        return watcher
+
+    def setUp(self):
+        FakeTimer.started = []
+        self.clock = Clock()
+        for target, value in ((ne.threading, 'Timer'),):
+            patcher = patch.object(target, value, FakeTimer)
+            patcher.start()
+            self.addCleanup(patcher.stop)
+        for patcher in (patch.object(ne.time, 'time', self.clock),
+                        patch.object(ne.startup_grace, 'is_real_system_boot', return_value=False)):
+            patcher.start()
+            self.addCleanup(patcher.stop)
+
+    def feed(self, watcher, host, lines):
+        with patch.object(ne.socket, 'gethostname', return_value=host):
+            for delay, ident, msg in lines:
+                self.clock.now += delay
+                watcher._check_cluster_events(msg, ident)
+
+    def expire_timers(self):
+        self.clock.now += ne.JournalWatcher.CLUSTER_LEAVE_GRACE_SECONDS
+        for timer in list(FakeTimer.started):
+            if not timer.cancelled:
+                timer.function(*timer.args)
+
+    def test_a_peer_corosync_restart_is_one_rejoin_notice(self):
+        watcher = self.watcher('pmxlab1')
+        self.feed(watcher, 'pmxlab1', [
+            (0, 'corosync', '  [CFG   ] Node 2 was shut down by sysadmin'),
+            (0, 'corosync', '  [QUORUM] Sync left[1]: 2'),
+            (0, 'corosync', '  [TOTEM ] A new membership (1.59) was formed. Members left: 2'),
+            (0, 'corosync', '  [QUORUM] This node is within the non-primary component and will NOT provide any services.'),
+            (0, 'pmxcfs', '[status] notice: node lost quorum'),
+            (3, 'corosync', '  [TOTEM ] A new membership (1.5e) was formed. Members joined: 2'),
+            (0, 'pmxcfs', '[status] notice: node has quorum'),
+        ])
+        self.expire_timers()
+        self.assertEqual(watcher.events, [('node_reconnect', 'INFO', {
+            'node_name': 'pmxlab2', 'downtime_seconds': 3, 'hostname': 'pmxlab1'})])
+
+    def test_a_node_that_stays_away_is_reported_after_the_grace(self):
+        watcher = self.watcher('pmxlab1')
+        self.feed(watcher, 'pmxlab1', [
+            (0, 'corosync', '  [TOTEM ] A new membership (1.59) was formed. Members left: 2'),
+        ])
+        self.assertEqual(watcher.events, [])
+        self.expire_timers()
+        self.assertEqual(watcher.events, [('node_disconnect', 'CRITICAL',
+                                           {'node_name': 'pmxlab2', 'hostname': 'pmxlab1'})])
+        self.feed(watcher, 'pmxlab1', [
+            (600, 'corosync', '  [TOTEM ] A new membership (1.5e) was formed. Members joined: 2'),
+        ])
+        self.assertEqual(watcher.events[-1], ('node_reconnect', 'INFO',
+                                              {'node_name': 'pmxlab2', 'hostname': 'pmxlab1'}))
+
+    def test_our_own_restart_does_not_report_the_peers_as_reconnected(self):
+        watcher = self.watcher('pmxlab2')
+        self.feed(watcher, 'pmxlab2', [
+            (0, 'corosync', '  [CFG   ] Node 2 was shut down by sysadmin'),
+            (0, 'corosync', '  [MAIN  ] Node was shut down by a signal'),
+            (0, 'pmxcfs', '[status] notice: node lost quorum'),
+            (1, 'corosync', '  [TOTEM ] A new membership (2.5a) was formed. Members joined: 2'),
+            (2, 'corosync', '  [TOTEM ] A new membership (1.5e) was formed. Members joined: 1'),
+            (4, 'pmxcfs', '[status] notice: node has quorum'),
+        ])
+        self.expire_timers()
+        self.assertEqual(watcher.events, [])
+
+    def test_lines_that_only_mention_a_node_name_nothing(self):
+        watcher = self.watcher('pmxlab1')
+        self.feed(watcher, 'pmxlab1', [
+            (0, 'pmxcfs', '[status] notice: node lost quorum'),
+            (0, 'node_exporter', 'level=warn msg="node offline, lost connection"'),
+            (0, 'app', 'worker node left the pool'),
+            (0, '', 'quorum lost by votequorum'),
+            (0, 'corosync', '  [QUORUM] Sync left[1]: 2'),
+        ])
+        self.expire_timers()
+        self.assertEqual(watcher.events, [])
+
+    def test_the_rejoin_notice_says_how_long_the_node_was_away(self):
+        rendered = nt.render_template('node_reconnect', {
+            'node_name': 'pve', 'downtime_seconds': 3, 'hostname': 'pve1'}, language='es')
+        self.assertIn('pve', rendered['body'])
+        self.assertIn('3 s', rendered['body'])
+        self.assertNotIn('{', rendered['body'])
+        plain = nt.render_template('node_reconnect', {'node_name': 'pve', 'hostname': 'pve1'}, language='en')
+        self.assertIn('rejoined the cluster successfully', plain['body'])
+
+
+if __name__ == '__main__':
+    unittest.main()

AppImage/scripts/tests/test_cluster_node_membership_notifications.py

@@ -35,8 +35,23 @@ nodelist {
 """
 
 
+class GraceTimer:
+    """Departures wait for a grace; these tests let it run out at once."""
+    pending = []
+
+    def __init__(self, interval, function, args=()):
+        self.function, self.args, self.cancelled, self.daemon = function, args, False, False
+
+    def start(self):
+        GraceTimer.pending.append(self)
+
+    def cancel(self):
+        self.cancelled = True
+
+
 class ClusterMembershipTests(unittest.TestCase):
     def setUp(self):
+        GraceTimer.pending = []
         folder = tempfile.TemporaryDirectory()
         self.addCleanup(folder.cleanup)
         conf = Path(folder.name) / "corosync.conf"
@@ -54,7 +69,8 @@ class ClusterMembershipTests(unittest.TestCase):
                      patch.object(notification_events.startup_grace, "is_real_system_boot",
                                   side_effect=lambda: self.booting),
                      patch.object(notification_events._shared_state, "is_startup_health_grace",
-                                  side_effect=lambda: self.booting)):
+                                  side_effect=lambda: self.booting),
+                     patch.object(notification_events.threading, "Timer", GraceTimer)):
             item.start()
             self.addCleanup(item.stop)
         self.queue = Queue()
@@ -62,6 +78,10 @@ class ClusterMembershipTests(unittest.TestCase):
 
     def notified(self, line):
         self.watcher._check_cluster_events(line, "corosync")
+        while GraceTimer.pending:
+            timer = GraceTimer.pending.pop(0)
+            if not timer.cancelled:
+                timer.function(*timer.args)
         events = []
         while True:
             try: