Fix infinite recursion in in replicatefrom tags (#3072)

Besides that:
1. fix problem with is_physical_slot() methods, it was returning false positives for logical slots.
2. Fix a little issue with replicatefrom docs.

Close https://github.com/zalando/patroni/issues/3068

docs/yaml_configuration.rst

@@ -395,7 +395,7 @@ Tags
 ----
 -  **clonefrom**: ``true`` or ``false``. If set to ``true`` other nodes might prefer to use this node for bootstrap (take ``pg_basebackup`` from). If there are several nodes with ``clonefrom`` tag set to ``true`` the node to bootstrap from will be chosen randomly. The default value is ``false``.
 -  **noloadbalance**: ``true`` or ``false``. If set to ``true`` the node will return HTTP Status Code 503 for the ``GET /replica`` REST API health-check and therefore will be excluded from the load-balancing. Defaults to ``false``.
--  **replicatefrom**: The IP address/hostname of another replica. Used to support cascading replication.
+-  **replicatefrom**: The name of another replica to replicate from. Used to support cascading replication.
 -  **nosync**: ``true`` or ``false``. If set to ``true`` the node will never be selected as a synchronous replica.
 -  **nofailover**: ``true`` or ``false``, controls whether this node is allowed to participate in the leader race and become a leader. Defaults to ``false``, meaning this node _can_ participate in leader races. 
 -  **failover_priority**: integer, controls the priority that this node should have during failover. Nodes with higher priority will be preferred over lower priority nodes if they received/replayed the same amount of WAL. However, nodes with higher values of receive/replay LSN are preferred regardless of their priority. If the ``failover_priority`` is 0 or negative - such node is not allowed to participate in the leader race and to become a leader (similar to ``nofailover: true``).

patroni/dcs/__init__.py

@@ -10,7 +10,7 @@ from copy import deepcopy
 from random import randint
 from threading import Event, Lock
 from typing import Any, Callable, Collection, Dict, Iterator, List, \
-    NamedTuple, Optional, Tuple, Type, TYPE_CHECKING, Union
+    NamedTuple, Optional, Set, Tuple, Type, TYPE_CHECKING, Union
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 import dateutil.parser
@@ -913,7 +913,9 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: ``True`` if *value* is a physical replication slot, otherwise ``False``.
         """
-        return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'
+        return not value \
+            or (isinstance(value, dict) and not Cluster.is_logical_slot(value)
+                and value.get('type', 'physical') == 'physical')
 
     @staticmethod
     def is_logical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
@@ -1179,6 +1181,10 @@ class Cluster(NamedTuple('Cluster',
 
         if global_config.use_slots:
             name = member.name if isinstance(member, Member) else postgresql.name
+
+            if not self.get_slot_name_on_primary(name, member):
+                return False
+
             members = [m for m in self.members if m.replicatefrom == name and m.name != self.leader_name]
             return any(self.should_enforce_hot_standby_feedback(postgresql, m) for m in members)
         return False
@@ -1197,11 +1203,18 @@ class Cluster(NamedTuple('Cluster',
 
         :returns: the slot name on the primary that is in use for physical replication on this node.
         """
-        if tags.nostream:
-            return None
-        replicatefrom = self.get_member(tags.replicatefrom, False) if tags.replicatefrom else None
-        return self.get_slot_name_on_primary(replicatefrom.name, replicatefrom) \
-            if isinstance(replicatefrom, Member) else slot_name_from_member_name(name)
+        seen_nodes: Set[str] = set()
+        while True:
+            seen_nodes.add(name)
+            if tags.nostream:
+                return None
+            replicatefrom = self.get_member(tags.replicatefrom, False) \
+                if tags.replicatefrom and tags.replicatefrom != name else None
+            if not isinstance(replicatefrom, Member):
+                return slot_name_from_member_name(name)
+            if replicatefrom.name in seen_nodes:
+                return None
+            name, tags = replicatefrom.name, replicatefrom
 
     @property
     def timeline(self) -> int:

tests/test_slots.py

@@ -186,6 +186,32 @@ class TestSlotsHandler(BaseTestPostgresql):
             cluster.get_slot_name_on_primary(self.p.name, stream_node),
             'test_4')
 
+    def test_get_slot_name_on_primary(self):
+        node1 = Member(0, 'node1', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'node2'}
+        })
+        node2 = Member(0, 'node2', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'node1'}
+        })
+        cluster = Cluster(True, None, self.leader, Status.empty(), [self.leadermem, node1, node2],
+                          None, SyncState.empty(), None, None)
+        self.assertIsNone(cluster.get_slot_name_on_primary('node1', node1))
+
+    def test_should_enforce_hot_standby_feedback(self):
+        node1 = Member(0, 'postgresql0', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'postgresql1'}
+        })
+        node2 = Member(0, 'postgresql1', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'postgresql0'}
+        })
+        cluster = Cluster(True, None, self.leader, Status.empty(), [self.leadermem, node1, node2],
+                          None, SyncState.empty(), None, None)
+        self.assertFalse(cluster.should_enforce_hot_standby_feedback(self.p, node1))
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')