diff --git a/docs/yaml_configuration.rst b/docs/yaml_configuration.rst
index e837f5f0..be3852c8 100644
--- a/docs/yaml_configuration.rst
+++ b/docs/yaml_configuration.rst
@@ -401,8 +401,9 @@ Tags
 -  **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 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.
+-  **sync_priority**: integer, controls the priority this node should have during synchronous replica selection when ``synchronous_mode`` is set to ``on``. Nodes with higher priority will be preferred over lower-priority nodes. If the ``sync_priority`` is 0 or negative - such node is not allowed to be written to ``synchronous_standby_names`` PostgreSQL parameter (similar to ``nosync: true``). Keep in mind, that this parameter has the opposite meaning to ``sync_priority`` value reported in ``pg_stat_replication`` view.
 -  **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``).
+-  **failover_priority**: integer, controls the priority 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``).
 -  **nostream**: ``true`` or ``false``. If set to ``true`` the node will not use replication protocol to stream WAL. It will rely instead on archive recovery (if ``restore_command`` is configured) and ``pg_wal``/``pg_xlog`` polling. It also disables copying and synchronization of permanent logical replication slots on the node itself and all its cascading replicas. Setting this tag on primary node has no effect.
 
 .. warning::
diff --git a/features/priority_sync.feature b/features/priority_sync.feature
new file mode 100644
index 00000000..a84fe7de
--- /dev/null
+++ b/features/priority_sync.feature
@@ -0,0 +1,36 @@
+Feature: synchronous replicas priority
+    We should check that we can give nodes priority for becoming synchronous replicas
+
+  Scenario: check replica with sync_priority=0 does not become a synchronous replica
+     Given I start postgres-0
+     When I issue a PATCH request to http://127.0.0.1:8008/config with {"ttl": 20, "synchronous_mode": true}
+     Then I receive a response code 200
+     When I configure and start postgres-1 with a tag sync_priority 0
+     Then sync key in DCS has leader=postgres-0 after 20 seconds
+     And sync key in DCS has sync_standby=None after 5 seconds
+
+  Scenario: check higher synchronous replicas priority is respected
+    Given I configure and start postgres-2 with a tag sync_priority 1
+    And I configure and start postgres-3 with a tag sync_priority 2
+    Then replication works from postgres-0 to postgres-2 after 20 seconds
+    And replication works from postgres-0 to postgres-3 after 20 seconds
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"synchronous_node_count": 1}
+    Then I receive a response code 200
+    And sync key in DCS has sync_standby=postgres-3 after 10 seconds
+
+
+ Scenario: check conflicting configuration handling
+    When I set nosync tag in postgres-3 config
+    And I issue an empty POST request to http://127.0.0.1:8011/reload
+    Then I receive a response code 202
+    And there is one of ["Conflicting configuration between nosync: True and sync_priority: 2. Defaulting to nosync: True"] WARNING in the postgres-3 patroni log after 5 seconds
+    And "members/postgres-3" key in DCS has tags={'nosync': True, 'sync_priority': '2'} after 10 seconds
+    And "sync" key in DCS has sync_standby=postgres-2 after 10 seconds
+    When I reset nosync tag in postgres-1 config
+    And I issue an empty POST request to http://127.0.0.1:8009/reload
+    Then I receive a response code 202
+    And there is one of ["Conflicting configuration between nosync: False and sync_priority: 0. Defaulting to nosync: False"] WARNING in the postgres-1 patroni log after 5 seconds
+    And "members/postgres-1" key in DCS has tags={'nosync': False, 'sync_priority': '0'} after 10 seconds
+    When I shut down postgres-2
+    And "sync" key in DCS has sync_standby=postgres-1 after 3 seconds
+
diff --git a/patroni/config.py b/patroni/config.py
index 05ed4ad5..401285b0 100644
--- a/patroni/config.py
+++ b/patroni/config.py
@@ -147,7 +147,7 @@ class Config(object):
         self._cache_file = os.path.join(self._data_dir, self.__CACHE_FILENAME)
         if validator:  # patronictl uses validator=None
             self._load_cache()  # we don't want to load anything from local cache for ctl
-            self._validate_failover_tags()  # irrelevant for ctl
+            self._validate_contradictory_tags()  # irrelevant for ctl
         self._cache_needs_saving = False
 
     @property
@@ -359,7 +359,7 @@ class Config(object):
                     new_configuration = self._build_effective_configuration(self._dynamic_configuration, configuration)
                     self._local_configuration = configuration
                     self.__effective_configuration = new_configuration
-                    self._validate_failover_tags()
+                    self._validate_contradictory_tags()
                     return True
                 else:
                     logger.info('No local configuration items changed.')
@@ -798,25 +798,31 @@ class Config(object):
         """
         return deepcopy(self.__effective_configuration)
 
-    def _validate_failover_tags(self) -> None:
-        """Check ``nofailover``/``failover_priority`` config and warn user if it's contradictory.
+    def _validate_contradictory_tags(self) -> None:
+        """Check boolean/priority tags' config and warn user if it's contradictory.
 
         .. note::
-          To preserve sanity (and backwards compatibility) the ``nofailover`` tag will still exist. A contradictory
-          configuration is one where ``nofailover`` is ``True`` but ``failover_priority > 0``, or where
-          ``nofailover`` is ``False``, but ``failover_priority <= 0``. Essentially, ``nofailover`` and
-          ``failover_priority`` are communicating different things.
+          To preserve sanity (and backwards compatibility) the ``nofailover``/``nosync`` tag will still exist.
+          A contradictory configuration is one where ``nofailover``/``nosync`` is ``True`` but
+          ``failover_priority > 0``/``sync_priority > 0``, or where ``nofailover``/``nosync`` is ``False``,
+          but ``failover_priority <= 0``/``sync_priority <= 0``. Essentially, ``nofailover``/``nosync`` and
+          ``failover_priority``/``sync_priority`` are communicating different things.
           This checks for this edge case (which is a misconfiguration on the part of the user) and warns them.
-          The behaviour is as if ``failover_priority`` were not provided (i.e ``nofailover`` is the
-          bedrock source of truth)
+          The behaviour is as if ``failover_priority``/``sync_priority`` were not provided
+          (i.e ``nofailover``/``nosync`` is the bedrock source of truth).
         """
         tags = self.get('tags', {})
-        if 'nofailover' not in tags:
-            return
-        nofailover_tag = tags.get('nofailover')
-        failover_priority_tag = parse_int(tags.get('failover_priority'))
-        if failover_priority_tag is not None \
-                and (bool(nofailover_tag) is True and failover_priority_tag > 0
-                     or bool(nofailover_tag) is False and failover_priority_tag <= 0):
-            logger.warning('Conflicting configuration between nofailover: %s and failover_priority: %s. '
-                           'Defaulting to nofailover: %s', nofailover_tag, failover_priority_tag, nofailover_tag)
+
+        def validate_tag(bool_name: str, priority_name: str) -> None:
+            if bool_name not in tags:
+                return
+            bool_tag = tags.get(bool_name)
+            priority_tag = parse_int(tags.get(priority_name))
+            if priority_tag is not None \
+                    and (bool(bool_tag) is True and priority_tag > 0
+                         or bool(bool_tag) is False and priority_tag <= 0):
+                logger.warning('Conflicting configuration between %s: %s and %s: %s. Defaulting to %s: %s',
+                               bool_name, bool_tag, priority_name, priority_tag, bool_name, bool_tag)
+
+        validate_tag('nofailover', 'failover_priority')
+        validate_tag('nosync', 'sync_priority')
diff --git a/patroni/config_generator.py b/patroni/config_generator.py
index 358d656e..6cf8a90f 100644
--- a/patroni/config_generator.py
+++ b/patroni/config_generator.py
@@ -126,6 +126,7 @@ class AbstractConfigGenerator(abc.ABC):
             },
             'tags': {
                 'failover_priority': 1,
+                'sync_priority': 1,
                 'noloadbalance': False,
                 'clonefrom': True,
                 'nosync': False,
diff --git a/patroni/ha.py b/patroni/ha.py
index 3b2c7d6e..f12501b2 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -410,6 +410,7 @@ class Ha(object):
         # _disable_sync could be modified concurrently, but we don't care as attribute get and set are atomic.
         if self._disable_sync > 0:
             tags['nosync'] = True
+            tags['sync_priority'] = 0
         return tags
 
     def notify_mpp_coordinator(self, event: str) -> None:
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
index 14778e95..1dfe5728 100644
--- a/patroni/postgresql/sync.py
+++ b/patroni/postgresql/sync.py
@@ -196,6 +196,7 @@ class _Replica(NamedTuple):
     sync_state: str
     lsn: int
     nofailover: bool
+    sync_priority: int
 
 
 class _ReplicaList(List[_Replica]):
@@ -238,10 +239,12 @@ class _ReplicaList(List[_Replica]):
             #   b. PostgreSQL on the member is known to be running and accepting client connections.
             if member and row[sort_col] is not None and member.is_running and not member.nosync:
                 self.append(_Replica(row['pid'], row['application_name'],
-                                     row['sync_state'], row[sort_col], bool(member.nofailover)))
+                                     row['sync_state'], row[sort_col],
+                                     bool(member.nofailover), member.sync_priority))
 
-        # Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN.
-        self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True)
+        # Prefer replicas with higher ``sync_priority`` value, in state ``sync``,
+        # and higher values of ``write``/``flush``/``replay`` LSN.
+        self.sort(key=lambda r: (r.sync_priority, r.sync_state, r.lsn), reverse=True)
 
         # When checking ``maximum_lag_on_syncnode`` we want to compare with the most
         # up-to-date replica otherwise with cluster LSN if there is only one replica.
diff --git a/patroni/tags.py b/patroni/tags.py
index 844a1761..c0f38f31 100644
--- a/patroni/tags.py
+++ b/patroni/tags.py
@@ -29,6 +29,7 @@ class Tags(abc.ABC):
             they all are boolean values that default to disabled.
             However ``nofailover`` tag is always returned if ``failover_priority`` tag is defined. In this case, we need
             both values to see if they are contradictory and the ``nofailover`` value should be used.
+            The same rule applies for ``nosync`` and ``sync_priority`` tags.
 
         :returns: a dictionary of tags set for this node. The key is the tag name, and the value is the corresponding
             tag value.
@@ -36,7 +37,8 @@ class Tags(abc.ABC):
         return {tag: value for tag, value in tags.items()
                 if any((tag not in ('clonefrom', 'nofailover', 'noloadbalance', 'nosync', 'nostream'),
                         value,
-                        tag == 'nofailover' and 'failover_priority' in tags))}
+                        tag == 'nofailover' and 'failover_priority' in tags,
+                        tag == 'nosync' and 'sync_priority' in tags))}
 
     @property
     @abc.abstractmethod
@@ -52,31 +54,49 @@ class Tags(abc.ABC):
         """``True`` if ``clonefrom`` tag is ``True``, else ``False``."""
         return self.tags.get('clonefrom', False)
 
+    def _priority_tag(self, bool_name: str, priority_name: str) -> int:
+        """Common logic for obtaining the value of a priority tag from ``tags`` if defined.
+
+        If boolean tag is defined as ``True``, this will return ``0``. Otherwise, it will return the value of
+        the respective priority tag, defaulting to ``1`` if it's not defined or invalid.
+
+        :param bool_name: name of the boolean tag (``nofailover``. ``nosync``).
+        :param priority_name: name of the priority tag (``failover_priority``, ``sync_priority``).
+
+        :returns: integer value based on the defined tags.
+        """
+        from_tags = self.tags.get(bool_name)
+        priority = parse_int(self.tags.get(priority_name))
+        priority = 1 if priority is None else priority
+        return 0 if from_tags else priority
+
+    def _bool_tag(self, bool_name: str, priority_name: str) -> bool:
+        """Common logic for obtaining the value of a boolean tag from ``tags`` if defined.
+
+        If boolean tag is not defined, this methods returns ``True`` if priority tag is non-positive,
+        ``False`` otherwise.
+
+        :param bool_name: name of the boolean tag (``nofailover``. ``nosync``).
+        :param priority_name: name of the priority tag (``failover_priority``, ``sync_priority``).
+
+        :returns: boolean value based on the defined tags.
+        """
+        from_tags = self.tags.get(bool_name)
+        if from_tags is not None:
+            # Value of bool tag takes precedence over priority tag
+            return bool(from_tags)
+        priority = parse_int(self.tags.get(priority_name))
+        return priority is not None and priority <= 0
+
     @property
     def nofailover(self) -> bool:
-        """Common logic for obtaining the value of ``nofailover`` from ``tags`` if defined.
-
-        If ``nofailover`` is not defined, this methods returns ``True`` if ``failover_priority`` is non-positive,
-        ``False`` otherwise.
-        """
-        from_tags = self.tags.get('nofailover')
-        if from_tags is not None:
-            # Value of `nofailover` takes precedence over `failover_priority`
-            return bool(from_tags)
-        failover_priority = parse_int(self.tags.get('failover_priority'))
-        return failover_priority is not None and failover_priority <= 0
+        """``True`` if node configuration doesn't allow it to become primary, ``False`` otherwise."""
+        return self._bool_tag('nofailover', 'failover_priority')
 
     @property
     def failover_priority(self) -> int:
-        """Common logic for obtaining the value of ``failover_priority`` from ``tags`` if defined.
-
-        If ``nofailover`` is defined as ``True``, this will return ``0``. Otherwise, it will return the value of
-        ``failover_priority``, defaulting to ``1`` if it's not defined or invalid.
-        """
-        from_tags = self.tags.get('nofailover')
-        failover_priority = parse_int(self.tags.get('failover_priority'))
-        failover_priority = 1 if failover_priority is None else failover_priority
-        return 0 if from_tags else failover_priority
+        """Value of ``failover_priority`` from ``tags`` if defined, otherwise derived from ``nofailover``."""
+        return self._priority_tag('nofailover', 'failover_priority')
 
     @property
     def noloadbalance(self) -> bool:
@@ -85,8 +105,13 @@ class Tags(abc.ABC):
 
     @property
     def nosync(self) -> bool:
-        """``True`` if ``nosync`` is ``True``, else ``False``."""
-        return bool(self.tags.get('nosync', False))
+        """``True`` if node configuration doesn't allow it to become synchronous, ``False`` otherwise."""
+        return self._bool_tag('nosync', 'sync_priority')
+
+    @property
+    def sync_priority(self) -> int:
+        """Value of ``sync_priority`` from ``tags`` if defined, otherwise derived from ``nosync``."""
+        return self._priority_tag('nosync', 'sync_priority')
 
     @property
     def replicatefrom(self) -> Optional[str]:
diff --git a/patroni/validator.py b/patroni/validator.py
index cb7b6452..ae924c2b 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -1182,7 +1182,10 @@ schema = Schema({
         Optional("clonefrom"): bool,
         Optional("noloadbalance"): bool,
         Optional("replicatefrom"): str,
-        Optional("nosync"): bool,
+        AtMostOne("nosync", "sync_priority"): Case({
+            "nosync": bool,
+            "sync_priority": IntValidator(min=0, expected_type=int, raise_assert=True),
+        }),
         Optional("nostream"): bool
     }
 })
diff --git a/postgres0.yml b/postgres0.yml
index 3dc0aa18..1546045c 100644
--- a/postgres0.yml
+++ b/postgres0.yml
@@ -133,7 +133,7 @@ postgresql:
 
 tags:
     # failover_priority: 1
+    # sync_priority: 1
     noloadbalance: false
     clonefrom: false
-    nosync: false
     nostream: false
diff --git a/postgres1.yml b/postgres1.yml
index e45e6e25..751a9761 100644
--- a/postgres1.yml
+++ b/postgres1.yml
@@ -125,5 +125,6 @@ postgresql:
 
 tags:
     # failover_priority: 1
+    # sync_priority: 1
     noloadbalance: false
     clonefrom: false
diff --git a/postgres2.yml b/postgres2.yml
index 39f27771..e64da525 100644
--- a/postgres2.yml
+++ b/postgres2.yml
@@ -115,6 +115,7 @@ postgresql:
     unix_socket_directories: '..'  # parent directory of data_dir
 tags:
     # failover_priority: 1
+    # sync_priority: 1
     noloadbalance: false
     clonefrom: false
 #    replicatefrom: postgresql1
diff --git a/tests/test_config.py b/tests/test_config.py
index ef4ef1ae..5a14eec4 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -160,40 +160,47 @@ class TestConfig(unittest.TestCase):
 
     @patch.object(Config, 'get')
     @patch('patroni.config.logger')
-    def test__validate_failover_tags(self, mock_logger, mock_get):
-        """Ensures that only one of `nofailover` or `failover_priority` can be provided"""
-        # Providing one of `nofailover` or `failover_priority` is fine
-        for single_param in ({"nofailover": True}, {"failover_priority": 1}, {"failover_priority": 0}):
-            mock_get.side_effect = [single_param] * 2
-            self.assertIsNone(self.config._validate_failover_tags())
-            mock_logger.warning.assert_not_called()
+    def test__validate_tags(self, mock_logger, mock_get):
+        """Ensures that only one of `nofailover`/`nosync' or `failover_priority`/`sync_priority` can be provided"""
+        tag_setup = (('nofailover', 'failover_priority'),
+                     ('nosync', 'sync_priority',))
 
-        # Providing both `nofailover` and `failover_priority` is fine if consistent
-        for consistent_state in (
-            {"nofailover": False, "failover_priority": 1},
-            {"nofailover": True, "failover_priority": 0},
-            {"nofailover": "False", "failover_priority": 0}
-        ):
-            mock_get.side_effect = [consistent_state] * 2
-            self.assertIsNone(self.config._validate_failover_tags())
-            mock_logger.warning.assert_not_called()
+        for tag, priority_tag in tag_setup:
+            # Providing one tag is fine
+            for single_param in ({tag: True}, {priority_tag: 1}, {priority_tag: 0}):
+                mock_get.side_effect = [single_param] * 2
+                self.assertIsNone(self.config._validate_contradictory_tags())
+                mock_logger.warning.assert_not_called()
 
-        # Providing both inconsistently should log a warning
-        for inconsistent_state in (
-            {"nofailover": False, "failover_priority": 0},
-            {"nofailover": True, "failover_priority": 1},
-            {"nofailover": "False", "failover_priority": 1},
-            {"nofailover": "", "failover_priority": 0}
-        ):
-            mock_get.side_effect = [inconsistent_state] * 2
-            self.assertIsNone(self.config._validate_failover_tags())
-            mock_logger.warning.assert_called_once_with(
-                'Conflicting configuration between nofailover: %s and failover_priority: %s.'
-                + ' Defaulting to nofailover: %s',
-                inconsistent_state['nofailover'],
-                inconsistent_state['failover_priority'],
-                inconsistent_state['nofailover'])
-            mock_logger.warning.reset_mock()
+            # Providing both tags is fine if consistent
+            for consistent_state in (
+                {tag: False, priority_tag: 1},
+                {tag: True, priority_tag: 0},
+                {tag: "False", priority_tag: 0}
+            ):
+                mock_get.side_effect = [consistent_state] * 2
+                self.assertIsNone(self.config._validate_contradictory_tags())
+                mock_logger.warning.assert_not_called()
+
+            # Providing both inconsistently should log a warning
+            for inconsistent_state in (
+                {tag: False, priority_tag: 0},
+                {tag: True, priority_tag: 1},
+                {tag: "False", priority_tag: 1},
+                {tag: "", priority_tag: 0}
+            ):
+                mock_get.side_effect = [inconsistent_state] * 2
+                self.assertIsNone(self.config._validate_contradictory_tags())
+                mock_logger.warning.assert_called_once_with(
+                    'Conflicting configuration between %s: %s and %s: %s.'
+                    + ' Defaulting to %s: %s',
+                    tag,
+                    inconsistent_state[tag],
+                    priority_tag,
+                    inconsistent_state[priority_tag],
+                    tag,
+                    inconsistent_state[tag])
+                mock_logger.warning.reset_mock()
 
     def test__process_postgresql_parameters(self):
         expected_params = {
diff --git a/tests/test_config_generator.py b/tests/test_config_generator.py
index 1fab6e04..dd1bd5ef 100644
--- a/tests/test_config_generator.py
+++ b/tests/test_config_generator.py
@@ -140,6 +140,7 @@ class TestGenerateConfig(unittest.TestCase):
             },
             'tags': {
                 'failover_priority': 1,
+                'sync_priority': 1,
                 'noloadbalance': False,
                 'clonefrom': True,
                 'nosync': False,
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 9b6f90b6..2e41f5dd 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -99,12 +99,13 @@ def get_cluster_initialized_with_leader_and_failsafe():
 
 def get_node_status(reachable=True, in_recovery=True, dcs_last_seen=0,
                     timeline=2, wal_position=10, nofailover=False,
-                    watchdog_failed=False, failover_priority=1):
+                    watchdog_failed=False, failover_priority=1, sync_priority=1):
     def fetch_node_status(e):
         tags = {}
         if nofailover:
             tags['nofailover'] = True
         tags['failover_priority'] = failover_priority
+        tags['sync_priority'] = sync_priority
         return _MemberStatus(e, reachable, in_recovery, wal_position,
                              {'tags': tags, 'watchdog_failed': watchdog_failed,
                               'dcs_last_seen': dcs_last_seen, 'timeline': timeline})
@@ -156,6 +157,7 @@ zookeeper:
         self.watchdog = Watchdog(self.config)
         self.request = lambda *args, **kwargs: requests_get(args[0].api_url, *args[1:], **kwargs)
         self.failover_priority = 1
+        self.sync_priority = 1
 
 
 def run_async(self, func, args=()):
@@ -1572,7 +1574,7 @@ class TestHa(PostgresInit):
 
     def test_effective_tags(self):
         self.ha._disable_sync = True
-        self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar', 'nosync': True})
+        self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar', 'nosync': True, 'sync_priority': 0})
         self.ha._disable_sync = False
         self.assertEqual(self.ha.get_effective_tags(), {'foo': 'bar'})
 
diff --git a/tests/test_patroni.py b/tests/test_patroni.py
index 5d68d575..7f528e67 100644
--- a/tests/test_patroni.py
+++ b/tests/test_patroni.py
@@ -202,6 +202,12 @@ class TestPatroni(unittest.TestCase):
         tags = {'nofailover': True, 'failover_priority': 1}
         self.assertEqual(self.p._filter_tags(tags), tags)
 
+        tags = {'nosync': False, 'sync_priority': 0}
+        self.assertEqual(self.p._filter_tags(tags), tags)
+
+        tags = {'nosync': True, 'sync_priority': 1}
+        self.assertEqual(self.p._filter_tags(tags), tags)
+
     def test_noloadbalance(self):
         self.p.tags['noloadbalance'] = True
         self.assertTrue(self.p.noloadbalance)