Synchronous mode strict (#435)

If synchronous_mode_strict==true then '*' will be written as synchronous_standby_names when that last replication host dies.

patroni/config.py

@@ -43,6 +43,7 @@ class Config(object):
         'maximum_lag_on_failover': 1048576,
         'master_start_timeout': 300,
         'synchronous_mode': False,
+        'synchronous_mode_strict': False,
         'postgresql': {
             'bin_dir': '',
             'use_slots': True,
@@ -174,7 +175,7 @@ class Config(object):
                     elif name not in ('connect_address', 'listen', 'data_dir', 'pgpass', 'authentication'):
                         config['postgresql'][name] = deepcopy(value)
             elif name in config:  # only variables present in __DEFAULT_CONFIG allowed to be overriden from DCS
-                if name == 'synchronous_mode':
+                if name in ('synchronous_mode', 'synchronous_mode_strict'):
                     config[name] = value
                 else:
                     config[name] = int(value)

patroni/ha.py

@@ -223,6 +223,9 @@ class Ha(object):
     def is_synchronous_mode(self):
         return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode'))
 
+    def is_synchronous_mode_strict(self):
+        return bool(self.cluster and self.cluster.config and self.cluster.config.data.get('synchronous_mode_strict'))
+
     def process_sync_replication(self):
         """Process synchronous standby beahvior.
 
@@ -242,10 +245,15 @@ class Ha(object):
                     if not self.dcs.write_sync_state(self.state_handler.name, None, index=self.cluster.sync.index):
                         logger.info('Synchronous replication key updated by someone else.')
                         return
+
+                if self.is_synchronous_mode_strict() and picked is None:
+                    picked = '*'
+                    logger.warning("No standbys available!")
+
                 logger.info("Assigning synchronous standby status to %s", picked)
                 self.state_handler.set_synchronous_standby(picked)
 
-                if picked and not allow_promote:
+                if picked and picked != '*' and not allow_promote:
                     # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
                     time.sleep(2)
                     picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster)

patroni/postgresql.py

@@ -175,7 +175,10 @@ class Postgresql(object):
         parameters.update({'cluster_name': self.scope, 'listen_addresses': listen_addresses, 'port': port})
         if config.get('synchronous_mode', False):
             if self._synchronous_standby_names is None:
-                parameters.pop('synchronous_standby_names', None)
+                if config.get('synchronous_mode_strict', False):
+                    parameters['synchronous_standby_names'] = '*'
+                else:
+                    parameters.pop('synchronous_standby_names', None)
             else:
                 parameters['synchronous_standby_names'] = self._synchronous_standby_names
         if self._major_version >= 9.6 and parameters['wal_level'] == 'hot_standby':

tests/test_api.py

@@ -104,6 +104,7 @@ class MockRequest(object):
     def sendall(self, *args, **kwargs):
         pass
 
+
 class MockRestApiServer(RestApiServer):
 
     def __init__(self, Handler, request):

tests/test_ha.py

@@ -663,6 +663,13 @@ class TestHa(unittest.TestCase):
         self.ha.run_cycle()
         self.assertEquals(self.ha.dcs.write_sync_state.call_count, 1)
 
+        # Test sync set to '*' when synchronous_mode_strict is enabled
+        mock_set_sync.reset_mock()
+        self.ha.is_synchronous_mode_strict = true
+        self.p.pick_synchronous_standby = Mock(return_value=(None, False))
+        self.ha.run_cycle()
+        mock_set_sync.assert_called_once_with('*')
+
     def test_sync_replication_become_master(self):
         self.ha.is_synchronous_mode = true
 

tests/test_postgresql.py

@@ -768,5 +768,7 @@ class TestPostgresql(unittest.TestCase):
     def test_get_server_parameters(self):
         config = {'synchronous_mode': True, 'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'}
         self.p.get_server_parameters(config)
+        config['synchronous_mode_strict'] = True
+        self.p.get_server_parameters(config)
         self.p.set_synchronous_standby('foo')
         self.p.get_server_parameters(config)