Introduce starting state and master start timeout. (#295)

Previously pg_ctl waited for a timeout and then happily trodded on considering PostgreSQL to be running. This caused PostgreSQL to show up in listings as running when it was actually not and caused a race condition that resulted in either a failover or a crash recovery or a crash recovery interrupted by failover and a missed rewind.

This change adds a master_start_timeout parameter and introduces a new state for the main run_cycle loop: starting. When master_start_timeout is zero we will fail over as soon as there is a failover candidate. Otherwise PostgreSQL will be started, but once master_start_timeout expires we will stop and release leader lock if failover is possible. Once failover succeeds or fails (no leader and no one to take the role) we continue with normal processing. While we are waiting for the master timeout we handle manual failover requests.

* Introduce timeout parameter to restart.

When restart timeout is set master becomes eligible for failover after that timeout expires regardless of master_start_time. Immediate restart calls will wait for this timeout to pass, even when node is a standby.

docs/SETTINGS.rst

@@ -14,6 +14,7 @@ Bootstrap configuration
     -  **loop\_wait**: the number of seconds the loop will sleep. Default value: 10
     -  **ttl**: the TTL to acquire the leader lock. Think of it as the length of time before initiation of the automatic failover process. Default value: 30
     -  **maximum\_lag\_on\_failover**: the maximum bytes a follower may lag to be able to participate in leader election.
+    -  **master\_start\_timeout**: the amount of time a master is allowed to recover from failures before failover is triggered. Default is 300 seconds. When set to 0 failover is done immediately after a crash is detected if possible. When using asynchronous replication a failover can cause lost transactions. Best worst case failover time for master failure is: ttl + master\_start\_timeout + ttl, unless master\_start\_timeout is zero, in which case it's just ttl. Set the value according to your durability/availability tradeoff.
     -  **synchronous\_mode**: turns on synchronous replication mode. In this mode a replica will be chosen as synchronous and only the latest leader and synchronous replica are able to participate in leader election. Synchronous mode makes sure that succesfully committed transactions will not be lost at failover, at the cost of losing availability for writes when Patroni cannot ensure transaction durability. See `replication modes documentation <https://github.com/zalando/patroni/blob/master/docs/replication_modes.rst>`__ for details.
     -  **postgresql**:
         -  **use\_pg\_rewind**:whether or not to use pg_rewind

patroni/__init__.py

@@ -143,8 +143,30 @@ def patroni_main():
             patroni.dcs.delete_leader()
 
 
+def pg_ctl_start(args):
+    import subprocess
+    postmaster = subprocess.Popen(args)
+    print(postmaster.pid)
+
+
+def call_self(args, **kwargs):
+    """This function executes Patroni once again with provided arguments.
+
+    :args: list of arguments to call Patroni with.
+    :returns: `Popen` object"""
+
+    exe = [sys.executable]
+    if not getattr(sys, 'frozen', False):  # Binary distribution?
+        exe.append(sys.argv[0])
+
+    import subprocess
+    return subprocess.Popen(exe + args, **kwargs)
+
+
 def main():
     if os.getpid() != 1:
+        if len(sys.argv) > 5 and sys.argv[1] == 'pg_ctl_start':
+            return pg_ctl_start(sys.argv[2:])
         return patroni_main()
 
     pid = 0
@@ -173,7 +195,6 @@ def main():
     signal.signal(signal.SIGQUIT, passtochild)
     signal.signal(signal.SIGTERM, passtochild)
 
-    import subprocess
-    patroni = subprocess.Popen([sys.executable] + sys.argv)
+    patroni = call_self(sys.argv[1:])
     pid = patroni.pid
     patroni.wait()

patroni/api.py

@@ -8,7 +8,7 @@ import dateutil.parser
 import datetime
 
 from patroni.exceptions import PostgresConnectionException
-from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError, is_valid_pg_version, tzutc
+from patroni.utils import deep_compare, patch_config, Retry, RetryFailedError, is_valid_pg_version, parse_int, tzutc
 from six.moves.BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
 from six.moves.socketserver import ThreadingMixIn
 from threading import Thread
@@ -223,6 +223,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     status_code = 400
                     data = "PostgreSQL version should be in the first.major.minor format"
                     break
+            elif k == 'timeout':
+                request[k] = parse_int(request[k], 's')
+                if request[k] is None or request[k] <= 0:
+                    status_code = 400
+                    data = "Timeout should be a positive number of seconds"
+                    break
             elif k != 'restart_pending':
                 status_code = 400
                 data = "Unknown filter for the scheduled restart: {0}".format(k)
@@ -236,7 +242,6 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     logger.exception('Exception during restart')
                     status_code = 400
             else:
-                request['postmaster_start_time'] = self.server.patroni.ha.state_handler.postmaster_start_time()
                 if self.server.patroni.ha.schedule_future_restart(request):
                     data = "Restart scheduled"
                     status_code = 202
@@ -379,7 +384,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
                                             THEN 0
                                             ELSE pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint
                                        END,
-                                       pg_xlog_location_diff(pg_last_xlog_receive_location(), '0/0')::bigint,
+                                       pg_xlog_location_diff(COALESCE(pg_last_xlog_receive_location(),
+                                                                      pg_last_xlog_replay_location()), '0/0')::bigint,
                                        pg_xlog_location_diff(pg_last_xlog_replay_location(), '0/0')::bigint,
                                        to_char(pg_last_xact_replay_timestamp(), 'YYYY-MM-DD HH24:MI:SS.MS TZ'),
                                        pg_is_in_recovery() AND pg_is_xlog_replay_paused(),

patroni/config.py

@@ -41,6 +41,7 @@ class Config(object):
     __DEFAULT_CONFIG = {
         'ttl': 30, 'loop_wait': 10, 'retry_timeout': 10,
         'maximum_lag_on_failover': 1048576,
+        'master_start_timeout': 300,
         'synchronous_mode': False,
         'postgresql': {
             'bin_dir': '',

patroni/ctl.py

@@ -439,9 +439,11 @@ def parse_scheduled(scheduled):
 @click.option('--pg-version', 'version', help='Restart if the PostgreSQL version is less than provided (e.g. 9.5.2)',
               default=None)
 @click.option('--pending', help='Restart if pending', is_flag=True)
+@click.option('--timeout',
+              help='Return error and fail over if necessary when restarting takes longer than this.')
 @option_force
 @click.pass_obj
-def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, version, pending):
+def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, version, pending, timeout):
     cluster = get_dcs(obj, cluster_name).get_cluster()
 
     members = get_members(cluster, cluster_name, member_names, role, force, 'restart')
@@ -473,6 +475,9 @@ def restart(obj, cluster_name, member_names, force, role, p_any, scheduled, vers
             raise PatroniCtlException("Can't schedule restart in the paused state")
         content['schedule'] = scheduled_at.isoformat()
 
+    if timeout is not None:
+        content['timeout'] = timeout
+
     for member in members:
         if 'schedule' in content:
             if force and member.data.get('scheduled_restart'):

patroni/ha.py

@@ -7,6 +7,7 @@ import requests
 import sys
 import time
 
+from collections import namedtuple
 from multiprocessing.pool import ThreadPool
 from patroni.async_executor import AsyncExecutor
 from patroni.exceptions import DCSError, PostgresConnectionException
@@ -17,6 +18,35 @@ from threading import RLock
 logger = logging.getLogger(__name__)
 
 
+class _MemberStatus(namedtuple('_MemberStatus', 'member,reachable,in_recovery,xlog_location,tags')):
+    """Node status distilled from API response:
+
+        member - dcs.Member object of the node
+        reachable - `!False` if the node is not reachable or is not responding with correct JSON
+        in_recovery - `!True` if pg_is_in_recovery() == true
+        xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.
+        is_lagging - `True` if node considers itself too far behind to promote
+        tags - dictionary with values of different tags (i.e. nofailover)
+    """
+    @classmethod
+    def from_api_response(cls, member, json):
+        is_master = json['role'] == 'master'
+        xlog_location = None if is_master else json['xlog']['received_location']
+        return cls(member, True, not is_master, xlog_location, json.get('tags', {}))
+
+    @classmethod
+    def unknown(cls, member):
+        return cls(member, False, None, 0, {})
+
+    def failover_limitation(self):
+        """Returns reason why this node can't promote or None if everything is ok."""
+        if not self.reachable:
+            return 'not reachable'
+        if self.tags.get('nofailover', False):
+            return 'not allowed to promote'
+        return None
+
+
 class Ha(object):
 
     def __init__(self, patroni):
@@ -26,6 +56,7 @@ class Ha(object):
         self.cluster = None
         self.old_cluster = None
         self.recovering = False
+        self._start_timeout = None
         self._async_executor = AsyncExecutor(self.wakeup)
 
         # Each member publishes various pieces of information to the DCS using touch_member. This lock protects
@@ -140,8 +171,21 @@ class Ha(object):
             return 'waiting for leader to bootstrap'
 
     def recover(self):
+        if self.has_lock() and self.update_lock():
+            timeout = self.patroni.config['master_start_timeout']
+            if timeout == 0:
+                # We are requested to prefer failing over to restarting master. But see first if there
+                # is anyone to fail over to.
+                if self.is_failover_possible(self.cluster.members):
+                    logger.info("Master crashed. Failing over.")
+                    self.demote('immediate')
+                    return 'stopped PostgreSQL to fail over after a crash'
+        else:
+            timeout = None
+
         self.recovering = True
-        return self.follow("starting as readonly because i had the session lock", "starting as a secondary", True, True)
+        return self.follow("starting as readonly because i had the session lock",
+                           "starting as a secondary", True, True, None, timeout)
 
     def _get_node_to_follow(self, cluster):
         # determine the node to follow. If replicatefrom tag is set,
@@ -153,7 +197,7 @@ class Ha(object):
 
         return node_to_follow if node_to_follow and node_to_follow.name != self.state_handler.name else None
 
-    def follow(self, demote_reason, follow_reason, refresh=True, recovery=False, need_rewind=None):
+    def follow(self, demote_reason, follow_reason, refresh=True, recovery=False, need_rewind=None, timeout=None):
         if refresh:
             self.load_cluster_from_dcs()
 
@@ -172,7 +216,8 @@ class Ha(object):
             elif not node_to_follow:
                 return 'no action'
 
-        self.state_handler.follow(node_to_follow, self.cluster.leader, recovery, self._async_executor, need_rewind)
+        self.state_handler.follow(node_to_follow, self.cluster.leader, recovery,
+                                  self._async_executor, need_rewind, timeout)
 
         return ret
 
@@ -282,24 +327,16 @@ class Ha(object):
     @staticmethod
     def fetch_node_status(member):
         """This function perform http get request on member.api_url and fetches its status
-        :returns: tuple(`member`, reachable, in_recovery, xlog_location)
-
-        reachable - `!False` if the node is not reachable or is not responding with correct JSON
-        in_recovery - `!True` if pg_is_in_recovery() == true
-        xlog_location - value of `replayed_location` or `location` from JSON, dependin on its role.
-        tags - dictionary with values of different tags (i.e. nofailover)
+        :returns: `_MemberStatus` object
         """
 
         try:
             response = requests.get(member.api_url, timeout=2, verify=False)
             logger.info('Got response from %s %s: %s', member.name, member.api_url, response.content)
-            json = response.json()
-            is_master = json['role'] == 'master'
-            xlog_location = None if is_master else json['xlog']['replayed_location']
-            return (member, True, not is_master, xlog_location, json.get('tags', {}))
+            return _MemberStatus.from_api_response(member, response.json())
         except Exception as e:
             logger.warning("request failed: GET %s (%s)", member.api_url, e)
-        return (member, False, None, 0, {})
+        return _MemberStatus.unknown(member)
 
     def fetch_nodes_statuses(self, members):
         pool = ThreadPool(len(members))
@@ -308,23 +345,32 @@ class Ha(object):
         pool.join()
         return results
 
+    def is_lagging(self, xlog_location):
+        """Returns if instance with an xlog should consider itself unhealthy to be promoted due to replication lag.
+
+        :param xlog_location: Current xlog location.
+        :returns True when node is lagging
+        """
+        lag = (self.cluster.last_leader_operation or 0) - xlog_location
+        return lag > self.state_handler.config.get('maximum_lag_on_failover', 0)
+
     def _is_healthiest_node(self, members, check_replication_lag=True):
         """This method tries to determine whether I am healthy enough to became a new leader candidate or not."""
 
-        if check_replication_lag and not self.state_handler.check_replication_lag(self.cluster.last_leader_operation):
+        my_xlog_location = self.state_handler.xlog_position()
+        if check_replication_lag and self.is_lagging(my_xlog_location):
             return False  # Too far behind last reported xlog location on master
 
         # Prepare list of nodes to run check against
         members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
 
         if members:
-            my_xlog_location = self.state_handler.xlog_position()
-            for member, reachable, in_recovery, xlog_location, tags in self.fetch_nodes_statuses(members):
-                if reachable and not tags.get('nofailover', False):  # If the node is unreachable it's not healhy
-                    if not in_recovery:
-                        logger.warning('Master (%s) is still alive', member.name)
+            for st in self.fetch_nodes_statuses(members):
+                if st.failover_limitation() is None:
+                    if not st.in_recovery:
+                        logger.warning('Master (%s) is still alive', st.member.name)
                         return False
-                    if my_xlog_location < xlog_location:
+                    if my_xlog_location < st.xlog_location:
                         return False
         return True
 
@@ -332,13 +378,14 @@ class Ha(object):
         ret = False
         members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
         if members:
-            for member, reachable, _, _, tags in self.fetch_nodes_statuses(members):
-                if reachable and not tags.get('nofailover', False):
-                    ret = True  # TODO: check xlog_location
-                elif not reachable:
-                    logger.info('Member %s is not reachable', member.name)
-                elif tags.get('nofailover', False):
-                    logger.info('Member %s is not allowed to promote', member.name)
+            for st in self.fetch_nodes_statuses(members):
+                not_allowed_reason = st.failover_limitation()
+                if not_allowed_reason:
+                    logger.info('Member %s is %s', st.member.name, not_allowed_reason)
+                elif self.is_lagging(st.xlog_location):
+                    logger.info('Member %s exceeds maximum replication lag', st.member.name)
+                else:
+                    ret = True
         else:
             logger.warning('manual failover: members list is empty')
         return ret
@@ -361,15 +408,13 @@ class Ha(object):
             # find specific node and check that it is healthy
             member = self.cluster.get_member(failover.candidate, fallback_to_leader=False)
             if member:
-                member, reachable, _, _, tags = self.fetch_node_status(member)
-                if reachable and not tags.get('nofailover', False):  # node is healthy
-                    logger.info('manual failover: to %s, i am %s', member.name, self.state_handler.name)
+                st = self.fetch_node_status(member)
+                not_allowed_reason = st.failover_limitation()
+                if not_allowed_reason is None:  # node is healthy
+                    logger.info('manual failover: to %s, i am %s', st.member.name, self.state_handler.name)
                     return False
                 # we wanted to failover to specific member but it is not healthy
-                if not reachable:
-                    logger.warning('manual failover: member %s is unhealthy', member.name)
-                elif tags.get('nofailover', False):
-                    logger.warning('manual failover: member %s is not allowed to promote', member.name)
+                logger.warning('manual failover: member %s is %s', st.member.name, not_allowed_reason)
 
             # at this point we should consider all members as a candidates for failover
             # i.e. we assume that failover.candidate is None
@@ -399,6 +444,9 @@ class Ha(object):
             if ret is not None:  # continue if we just deleted the stale failover key as a master
                 return ret
 
+        if self.state_handler.is_starting():  # postgresql still starting up is unhealthy
+            return False
+
         if self.state_handler.is_leader():  # leader is always the healthiest
             return True
 
@@ -424,18 +472,45 @@ class Ha(object):
 
         return self._is_healthiest_node(members.values())
 
-    def demote(self, delete_leader=True):
-        if delete_leader:
+    def release_leader_key_voluntarily(self):
+        self.dcs.delete_leader()
+        self.touch_member()
+        self.dcs.reset_cluster()
+        logger.info("Leader key released")
+
+    def demote(self, mode):
+        """Demote PostgreSQL running as master.
+
+        :param mode: One of offline, graceful or immediate.
+            offline is used when connection to DCS is not available.
+            graceful is used when failing over to another node due to user request. May only be called running async.
+            immediate is used when we determine that we are not suitable for master and want to failover quickly
+                without regard for data durability. May only be called synchronously.
+        """
+        assert mode in ['offline', 'graceful', 'immediate']
+        if mode != 'offline':
+            if mode == 'immediate':
+                self.state_handler.stop('immediate', checkpoint=False)
+            else:
                 self.state_handler.stop()
             self.state_handler.set_role('demoted')
-            self.dcs.delete_leader()
-            self.dcs.reset_cluster()
+            self.release_leader_key_voluntarily()
             time.sleep(2)  # Give a time to somebody to take the leader lock
             cluster = self.dcs.get_cluster()
             node_to_follow = self._get_node_to_follow(cluster)
+            if mode == 'immediate':
+                # We will try to start up as a standby now. If no one takes the leader lock before we finish
+                # recovery we will try to promote ourselves.
+                self._async_executor.schedule('waiting for failover to complete')
+                self._async_executor.run_async(self.state_handler.follow,
+                                               (node_to_follow, cluster.leader, True, None, True))
+            else:
                 return self.state_handler.follow(node_to_follow, cluster.leader, recovery=True, need_rewind=True)
         else:
-            self.state_handler.follow(None, None)
+            # Need to become unavailable as soon as possible, so initiate a stop here. However as we can't release
+            # the leader key we don't care about confirming the shutdown quickly and can use a regular stop.
+            self.state_handler.stop(checkpoint=False)
+            self.state_handler.follow(None, None, recovery=True)
 
     def should_run_scheduled_action(self, action_name, scheduled_at, cleanup_fn):
         if scheduled_at and not self.is_paused():
@@ -454,6 +529,9 @@ class Ha(object):
                                 action_name, scheduled_at.isoformat(), delta)
                     return False
                 elif delta < - int(self.dcs.loop_wait * 1.5):
+                    # This means that if run_cycle gets delayed for 2.5x loop_wait we skip the
+                    # scheduled action. Probably not a problem, if things are that bad we don't
+                    # want to be restarting or failing over anyway.
                     logger.warning('Found a stale %s value, cleaning up: %s',
                                    action_name, scheduled_at.isoformat())
                     cleanup_fn()
@@ -492,7 +570,7 @@ class Ha(object):
                                if not failover.candidate or m.name == failover.candidate]
                     if self.is_failover_possible(members):  # check that there are healthy members
                         self._async_executor.schedule('manual failover: demote')
-                        self._async_executor.run_async(self.demote)
+                        self._async_executor.run_async(self.demote, ('graceful',))
                         return 'manual failover: demoting myself'
                     else:
                         logger.warning('manual failover: no healthy members found, failover is not possible')
@@ -559,7 +637,7 @@ class Ha(object):
             else:
                 # Either there is no connection to DCS or someone else acquired the lock
                 logger.error('failed to update leader lock')
-                self.demote(delete_leader=False)
+                self.demote('offline')
                 return 'demoted self because failed to update leader lock in DCS'
         else:
             logger.info('does not have lock')
@@ -614,6 +692,7 @@ class Ha(object):
 
     def schedule_future_restart(self, restart_data):
         with self._async_executor:
+            restart_data['postmaster_start_time'] = self.state_handler.postmaster_start_time()
             if not self.patroni.scheduled_restart:
                 self.patroni.scheduled_restart = restart_data
                 self.touch_member()
@@ -636,10 +715,11 @@ class Ha(object):
     def restart_scheduled(self):
         return self._async_executor.scheduled_action == 'restart'
 
-    def restart(self, restart_data=None, run_async=False):
+    def restart(self, restart_data, run_async=False):
         """ conditional and unconditional restart """
-        if (restart_data and isinstance(restart_data, dict) and
-            not self.restart_matches(restart_data.get('role'),
+        assert isinstance(restart_data, dict)
+
+        if (not self.restart_matches(restart_data.get('role'),
                                      restart_data.get('postgres_version'),
                                      ('restart_pending' in restart_data))):
             return (False, "restart conditions are not satisfied")
@@ -649,15 +729,30 @@ class Ha(object):
             if prev is not None:
                 return (False, prev + ' already in progress')
 
-        do_restart = self.state_handler.restart
+            # Make the main loop to think that we were recovering dead postgres. If we fail
+            # to start postgres after a specified timeout (see below), we need to remove
+            # leader key (if it belong to us) rather than trying to start postgres once again.
+            self.recovering = True
+
+        # No that restart is scheduled we can set timeout for startup, it will get reset
+        # once async executor runs and main loop notices PostgreSQL as up.
+        timeout = restart_data.get('timeout', self.patroni.config['master_start_timeout'])
+        self.set_start_timeout(timeout)
+
+        # For non async cases we want to wait for restart to complete or timeout before returning.
+        do_restart = functools.partial(self.state_handler.restart, timeout)
         if self.is_synchronous_mode() and not self.has_lock():
             do_restart = functools.partial(self.while_not_sync_standby, do_restart)
 
         if run_async:
             self._async_executor.run_async(do_restart)
             return (True, 'restart initiated')
-        elif self._async_executor.run(do_restart):
+        else:
+            res = self._async_executor.run(do_restart)
+            if res:
                 return (True, 'restarted successfully')
+            elif res is None:
+                return (False, 'postgres is still starting')
             else:
                 return (False, 'restart failed')
 
@@ -712,6 +807,49 @@ class Ha(object):
             return 'failed to start postgres'
         return None
 
+    def handle_starting_instance(self):
+        """Starting up PostgreSQL may take a long time. In case we are the leader we may want to
+        fail over to."""
+
+        # Check if we are in startup, when paused defer to main loop for manual failovers.
+        if not self.state_handler.check_for_startup() or self.is_paused():
+            self.set_start_timeout(None)
+            return None
+
+        # state_handler.state == 'starting' here
+        if self.has_lock():
+            if not self.update_lock():
+                logger.info("Lost lock while starting up. Demoting self.")
+                self.demote('immediate')
+                return 'stopped PostgreSQL while starting up because leader key was lost'
+
+            timeout = self._start_timeout or self.patroni.config['master_start_timeout']
+            time_left = timeout - self.state_handler.time_in_state()
+
+            if time_left <= 0:
+                if self.is_failover_possible(self.cluster.members):
+                    logger.info("Demoting self because master startup is taking too long")
+                    self.demote('immediate')
+                    return 'stopped PostgreSQL because of startup timeout'
+                else:
+                    return 'master start has timed out, but continuing to wait because failover is not possible'
+            else:
+                msg = self.process_manual_failover_from_leader()
+                if msg is not None:
+                    return msg
+
+                return 'PostgreSQL is still starting up, {0:.0f} seconds until timeout'.format(time_left)
+        else:
+            # Use normal processing for standbys
+            logger.info("Still starting up as a standby.")
+            return None
+
+    def set_start_timeout(self, value):
+        """Sets timeout for starting as master before eligible for failover.
+
+        Must be called when async_executor is busy or in the main thread."""
+        self._start_timeout = value
+
     def _run_cycle(self):
         dcs_failed = False
         try:
@@ -731,6 +869,10 @@ class Ha(object):
             if self._async_executor.busy:
                 return self.handle_long_action_in_progress()
 
+            msg = self.handle_starting_instance()
+            if msg is not None:
+                return msg
+
             # we've got here, so any async action has finished.
             if self.recovering and not self.state_handler.need_rewind:
                 self.recovering = False
@@ -774,7 +916,7 @@ class Ha(object):
                 # we might not have a valid PostgreSQL connection here if another thread
                 # stops PostgreSQL, therefore, we only reload replication slots if no
                 # asynchronous processes are running (should be always the case for the master)
-                if not self._async_executor.busy:
+                if not self._async_executor.busy and not self.state_handler.is_starting():
                     if not self.state_handler.cb_called:
                         self.state_handler.call_nowait(ACTION_ON_START)
                     self.state_handler.sync_replication_slots(self.cluster)
@@ -782,7 +924,7 @@ class Ha(object):
             dcs_failed = True
             logger.error('Error communicating with DCS')
             if not self.is_paused() and self.state_handler.is_running() and self.state_handler.is_leader():
-                self.demote(delete_leader=False)
+                self.demote('offline')
                 return 'demoted self because DCS is not accessible and i was a leader'
             return 'DCS is not accessible'
         except (psycopg2.Error, PostgresConnectionException):

patroni/postgresql.py

@@ -1,4 +1,3 @@
-from collections import defaultdict
 import logging
 import os
 import psycopg2
@@ -9,10 +8,12 @@ import subprocess
 import tempfile
 import time
 
+from collections import defaultdict
+from patroni import call_self
 from patroni.exceptions import PostgresConnectionException, PostgresException
-from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError
+from patroni.utils import compare_values, parse_bool, parse_int, Retry, RetryFailedError, polling_loop
 from six import string_types
-from threading import Lock
+from threading import current_thread, Lock
 
 logger = logging.getLogger(__name__)
 
@@ -22,6 +23,11 @@ ACTION_ON_RESTART = "on_restart"
 ACTION_ON_RELOAD = "on_reload"
 ACTION_ON_ROLE_CHANGE = "on_role_change"
 
+STATE_RUNNING = 'running'
+STATE_REJECT = 'rejecting connections'
+STATE_NO_RESPONSE = 'not responding'
+STATE_UNKNOWN = 'unknown'
+
 
 def slot_name_from_member_name(member_name):
     """Translate member name to valid PostgreSQL slot name.
@@ -80,6 +86,7 @@ class Postgresql(object):
         self._database = config.get('database', 'postgres')
         self._data_dir = config['data_dir']
         self._pending_restart = False
+        self.__thread_ident = current_thread().ident
 
         self._version_file = os.path.join(self._data_dir, 'PG_VERSION')
         self._major_version = self.get_major_version()
@@ -97,6 +104,7 @@ class Postgresql(object):
 
         self._pgpass = config.get('pgpass') or os.path.join(os.path.expanduser('~'), 'pgpass')
         self.__cb_called = False
+        self.__cb_pending = None
         config_base_name = config.get('config_base_name', 'postgresql')
         self._postgresql_conf = os.path.join(self._data_dir, config_base_name + '.conf')
         self._postgresql_base_conf_name = config_base_name + '.base.conf'
@@ -113,11 +121,17 @@ class Postgresql(object):
         self.retry = Retry(max_tries=-1, deadline=config['retry_timeout']/2.0, max_delay=1,
                            retry_exceptions=PostgresConnectionException)
 
+        # Retry 'pg_is_in_recovery()' only once
+        self._is_leader_retry = Retry(max_tries=1, deadline=config['retry_timeout']/2.0, max_delay=1,
+                                      retry_exceptions=PostgresConnectionException)
+
         self._state_lock = Lock()
         self.set_state('stopped')
         self._role_lock = Lock()
         self.set_role(self.get_postgres_role_from_data_directory())
 
+        self._state_entry_timestamp = None
+
         if self.is_running():
             self.set_state('running')
             self.set_role('master' if self.is_leader() else 'replica')
@@ -179,7 +193,7 @@ class Postgresql(object):
         :returns: `!True` when return_code == 0, otherwise `!False`"""
 
         pg_ctl = [self._pgcommand('pg_ctl'), cmd]
-        if cmd in ('start', 'stop', 'restart'):
+        if cmd == 'stop':
             pg_ctl += ['-w']
             timeout = self.config.get('pg_ctl_timeout')
             if timeout:
@@ -189,11 +203,31 @@ class Postgresql(object):
                     logger.error('Bad value of pg_ctl_timeout: %s', timeout)
         return subprocess.call(pg_ctl + ['-D', self._data_dir] + list(args), **kwargs) == 0
 
+    def pg_isready(self):
+        """Runs pg_isready to see if PostgreSQL is accepting connections.
+
+        :returns: 'ok' if PostgreSQL is up, 'reject' if starting up, 'no_resopnse' if not up."""
+
+        cmd = [self._pgcommand('pg_isready'),
+               '-h', self._local_address['host'],
+               '-p', self._local_address['port'],
+               '-d', self._database]
+        # We only need the username because pg_isready does not try to authenticate
+        if 'username' in self._superuser:
+            cmd.extend(['-U', self._superuser['username']])
+
+        ret = subprocess.call(cmd)
+        return_codes = {0: STATE_RUNNING,
+                        1: STATE_REJECT,
+                        2: STATE_NO_RESPONSE,
+                        3: STATE_UNKNOWN}
+        return return_codes.get(ret, STATE_UNKNOWN)
+
     def reload_config(self, config):
         server_parameters = self.get_server_parameters(config)
 
         listen_address_changed = pending_reload = pending_restart = False
-        if self.is_healthy():
+        if self.state == 'running':
             changes = {p: v for p, v in server_parameters.items() if '.' not in p}
             changes.update({p: None for p, v in self._server_parameters.items() if not ('.' in p or p in changes)})
             if changes:
@@ -250,7 +284,7 @@ class Postgresql(object):
         if pending_reload:
             self._write_postgresql_conf()
             self.reload()
-        self.retry.deadline = config['retry_timeout']/2.0
+        self._is_leader_retry.deadline = self.retry.deadline = config['retry_timeout']/2.0
 
     @property
     def pending_restart(self):
@@ -334,6 +368,9 @@ class Postgresql(object):
         self._cursor_holder = self._connection = None
 
     def _query(self, sql, *params):
+        """We are always using the same cursor, therefore this method is not thread-safe!!!
+        You can call it from different threads only if you are holding explicit `AsyncExecutor` lock,
+        because the main thread is always holding this lock when running HA cycle."""
         cursor = None
         try:
             cursor = self._cursor()
@@ -412,6 +449,9 @@ class Postgresql(object):
                 connstring = 'postgres://{user}@{host}:{port}/{database}'.format(**r)
             else:
                 connstring = 'postgres://{host}:{port}/{database}'.format(**r)
+                if 'password' in r:
+                    import getpass
+                    r.setdefault('user', os.environ.get('PGUSER', getpass.getuser()))
 
             env = self.write_pgpass(r) if 'password' in r else None
             try:
@@ -521,14 +561,34 @@ class Postgresql(object):
         return ret
 
     def is_leader(self):
-        return not self.query('SELECT pg_is_in_recovery()').fetchone()[0]
+        try:
+            return not self._is_leader_retry(self._query, 'SELECT pg_is_in_recovery()').fetchone()[0]
+        except RetryFailedError as e:  # SELECT pg_is_in_recovery() failed two times
+            if not self.is_starting() and self.pg_isready() == STATE_REJECT:
+                self.set_state('starting')
+            raise PostgresConnectionException(str(e))
 
     def is_running(self):
         if not (self._version_file_exists() and os.path.isfile(self._postmaster_pid)):
             return False
+        return self.is_pid_running(self.read_pid_file().get('pid', 0))
+
+    def read_pid_file(self):
+        """Reads and parses postmaster.pid from the data directory
+
+        :returns dictionary of values if successful, empty dictionary otherwise
+        """
+        pid_line_names = ['pid', 'data_dir', 'start_time', 'port', 'socket_dir', 'listen_addr', 'shmem_key']
         try:
             with open(self._postmaster_pid) as f:
-                pid = int(f.readline())
+                return {name: line.rstrip("\n") for name, line in zip(pid_line_names, f)}
+        except IOError:
+            return {}
+
+    @staticmethod
+    def is_pid_running(pid):
+        try:
+            pid = int(pid)
             if pid < 0:
                 pid = -pid
             return pid > 0 and pid != os.getpid() and pid != os.getppid() and (os.kill(pid, 0) or True)
@@ -571,8 +631,48 @@ class Postgresql(object):
     def set_state(self, value):
         with self._state_lock:
             self._state = value
+            self._state_entry_timestamp = time.time()
 
-    def start(self, block_callbacks=False):
+    def time_in_state(self):
+        return time.time() - self._state_entry_timestamp
+
+    def is_starting(self):
+        return self.state == 'starting'
+
+    def wait_for_port_open(self, pid, initiated, timeout):
+        """Waits until PostgreSQL opens ports."""
+        for _ in polling_loop(timeout):
+            pid_file = self.read_pid_file()
+            if len(pid_file) > 5:
+                try:
+                    pmpid = int(pid_file['pid'])
+                    pmstart = int(pid_file['start_time'])
+
+                    if pmstart >= initiated - 2 and pmpid == pid:
+                        isready = self.pg_isready()
+                        if isready != STATE_NO_RESPONSE:
+                            if isready not in [STATE_REJECT, STATE_RUNNING]:
+                                logger.warning("Can't determine PostgreSQL startup status, assuming running")
+                            return True
+                except ValueError:
+                    # Garbage in the pid file
+                    pass
+
+            if not self.is_pid_running(pid):
+                logger.error('postmaster is not running')
+                self.set_state('start failed')
+                return False
+
+        logger.warning("Timed out waiting for PostgreSQL to start")
+        return False
+
+    def start(self, timeout=None, block_callbacks=False):
+        """Start PostgreSQL
+
+        Waits for postmaster to open ports or terminate so pg_isready can be used to check startup completion
+        or failure.
+
+        :returns: True if start was initiated and postmaster ports are open, False if start failed"""
         # make sure we close all connections established against
         # the former node, otherwise, we might get a stalled one
         # after kill -9, which would report incorrect data to
@@ -583,18 +683,13 @@ class Postgresql(object):
             logger.error('Cannot start PostgreSQL because one is already running.')
             return True
 
-        self.set_role(self.get_postgres_role_from_data_directory())
-        if os.path.exists(self._postmaster_pid):
-            os.remove(self._postmaster_pid)
-            logger.info('Removed %s', self._postmaster_pid)
-
         if not block_callbacks:
-            self.set_state('starting')
+            self.__cb_pending = ACTION_ON_START
 
-        env = {'PATH': os.environ.get('PATH')}
-        # pg_ctl will write a FATAL if the username is incorrect. exporting PGUSER if necessary
-        if 'username' in self._superuser and self._superuser['username'] != os.environ.get('USER'):
-            env['PGUSER'] = self._superuser['username']
+        self.set_role(self.get_postgres_role_from_data_directory())
+
+        self.set_state('starting')
+        self._pending_restart = False
 
         self._write_postgresql_conf()
         self.resolve_connection_addresses()
@@ -602,20 +697,45 @@ class Postgresql(object):
         opts = {p: self._server_parameters[p] for p, v in self.CMDLINE_OPTIONS.items() if self._major_version >= v[2]}
         if self._major_version >= 9.6 and opts['wal_level'] == 'hot_standby':
             opts['wal_level'] = 'replica'
-        options = ' '.join("--{0}='{1}'".format(p, v) for p, v in opts.items())
+        options = ['--{0}={1}'.format(p, v) for p, v in opts.items()]
 
-        ret = self.pg_ctl('start', '-o', options, env=env, preexec_fn=os.setsid)
-        self._pending_restart = False
+        start_initiated = time.time()
 
-        self.set_state('running' if ret else 'start failed')
+        # Unfortunately `pg_ctl start` does not return postmaster pid to us. Without this information
+        # it is hard to know the current state of postgres startup, so we had to reimplement pg_ctl start
+        # in python. It will start postgres, wait for port to be open and wait until postgres will start
+        # accepting connections.
+        # Important!!! We can't just start postgres using subprocess.Popen, because in this case it
+        # will be our child for the rest of our live and we will have to take care of it (`waitpid`).
+        # So we will use the same approach as pg_ctl uses: start a new process, which will start postgres.
+        # This process will write postmaster pid to stdout and exit immediately. Now it's responsibility
+        # of init process to take care about postmaster.
+        # In order to make everything portable we can't use fork&exec approach here, so  we will call
+        # ourselves and pass list of arguments which must be used to start postgres.
+        proc = call_self(['pg_ctl_start', self._pgcommand('postgres'), '-D', self._data_dir] + options, close_fds=True,
+                         preexec_fn=os.setsid, stdout=subprocess.PIPE, env={'PATH': os.environ.get('PATH')})
+        pid = int(proc.stdout.readline().strip())
+        proc.wait()
+        logger.info('postmaster pid=%s', pid)
 
-        self._schedule_load_slots = ret and self.use_slots
-        self.save_configuration_files()
-        # block_callbacks is used during restart to avoid
-        # running start/stop callbacks in addition to restart ones
-        if ret and not block_callbacks:
-            self.call_nowait(ACTION_ON_START)
+        start_timeout = timeout
+        if not start_timeout:
+            try:
+                start_timeout = float(self.config.get('pg_ctl_timeout', 60))
+            except ValueError:
+                start_timeout = 60
+
+        # We want postmaster to open ports before we continue
+        if not self.wait_for_port_open(pid, start_initiated, start_timeout):
+            return False
+
+        ret = self.wait_for_startup(start_timeout)
+        if ret is not None:
             return ret
+        elif timeout is not None:
+            return False
+        else:
+            return None
 
     def checkpoint(self, connect_kwargs=None):
         check_not_is_in_recovery = connect_kwargs is not None
@@ -642,7 +762,7 @@ class Postgresql(object):
                 self.set_state('stopped')
             return True
 
-        if checkpoint:
+        if checkpoint and not self.is_starting():
             self.checkpoint()
 
         if not block_callbacks:
@@ -665,12 +785,70 @@ class Postgresql(object):
             self.call_nowait(ACTION_ON_RELOAD)
         return ret
 
-    def restart(self):
-        self.set_state('restarting')
-        ret = self.stop(block_callbacks=True) and self.start(block_callbacks=True)
-        if ret:
-            self.call_nowait(ACTION_ON_RESTART)
+    def check_for_startup(self):
+        """Checks PostgreSQL status and returns if PostgreSQL is in the middle of startup."""
+        return self.is_starting() and not self.check_startup_state_changed()
+
+    def check_startup_state_changed(self):
+        """Checks if PostgreSQL has completed starting up or failed or still starting.
+
+        Should only be called when state == 'starting'
+
+        :returns: True iff state was changed from 'starting'
+        """
+        ready = self.pg_isready()
+
+        if ready == STATE_REJECT:
+            return False
+        elif ready == STATE_NO_RESPONSE:
+            self.set_state('start failed')
+            self._schedule_load_slots = False  # TODO: can remove this?
+            self.save_configuration_files()  # TODO: maybe remove this?
+            return True
         else:
+            if ready != STATE_RUNNING:
+                # Bad configuration or unexpected OS error. No idea of PostgreSQL status.
+                # Let the main loop of run cycle clean up the mess.
+                logger.warning("%s status returned from pg_isready",
+                               "Unknown" if ready == STATE_UNKNOWN else "Invalid")
+            self.set_state('running')
+            self._schedule_load_slots = self.use_slots
+            self.save_configuration_files()
+            # TODO: __cb_pending can be None here after PostgreSQL restarts on its own. Do we want to call the callback?
+            # Previously we didn't even notice.
+            action = self.__cb_pending or ACTION_ON_START
+            self.call_nowait(action)
+            self.__cb_pending = None
+
+            return True
+
+    def wait_for_startup(self, timeout=None):
+        """Waits for PostgreSQL startup to complete or fail.
+
+        :returns: True if start was successful, False otherwise"""
+        if not self.is_starting():
+            # Should not happen
+            logger.warning("wait_for_startup() called when not in starting state")
+
+        while not self.check_startup_state_changed():
+            if timeout and self.time_in_state() > timeout:
+                return None
+            time.sleep(1)
+
+        return self.state == 'running'
+
+    def restart(self, timeout=None):
+        """Restarts PostgreSQL.
+
+        When timeout parameter is set the call will block either until PostgreSQL has started, failed to start or
+        timeout arrives.
+
+        :returns: True when restart was successful and timeout did not expire when waiting.
+        """
+        self.set_state('restarting')
+        self.__cb_pending = ACTION_ON_RESTART
+        ret = self.stop(block_callbacks=True) and self.start(timeout=timeout, block_callbacks=True)
+        if not ret and not self.is_starting():
             self.set_state('restart failed ({0})'.format(self.state))
         return ret
 
@@ -691,9 +869,6 @@ class Postgresql(object):
             return False
         return True
 
-    def check_replication_lag(self, last_leader_operation):
-        return (last_leader_operation or 0) - self.xlog_position() <= self.config.get('maximum_lag_on_failover', 0)
-
     def write_pg_hba(self, config):
         with open(os.path.join(self._data_dir, 'pg_hba.conf'), 'a') as f:
             f.write('\n{}\n'.format('\n'.join(config)))
@@ -804,7 +979,7 @@ class Postgresql(object):
     def need_rewind(self):
         return self._need_rewind
 
-    def follow(self, member, leader, recovery=False, async_executor=None, need_rewind=None):
+    def follow(self, member, leader, recovery=False, async_executor=None, need_rewind=None, timeout=None):
         if need_rewind is not None:
             self._need_rewind = need_rewind
 
@@ -815,11 +990,11 @@ class Postgresql(object):
 
         if async_executor:
             async_executor.schedule('changing primary_conninfo and restarting')
-            async_executor.run_async(self._do_follow, (primary_conninfo, leader, recovery))
+            async_executor.run_async(self._do_follow, (primary_conninfo, leader, recovery, timeout))
         else:
-            return self._do_follow(primary_conninfo, leader, recovery)
+            return self._do_follow(primary_conninfo, leader, recovery, timeout)
 
-    def _do_follow(self, primary_conninfo, leader, recovery=False):
+    def _do_follow(self, primary_conninfo, leader, recovery=False, timeout=None):
         change_role = self.role in ('master', 'demoted')
 
         if leader and leader.name == self.name:
@@ -830,12 +1005,13 @@ class Postgresql(object):
         elif change_role:
             self._need_rewind = True
 
-        self._need_rewind &= bool(leader and leader.conn_url) and self.can_rewind
+        if self._need_rewind and not self.can_rewind:
+            logger.warning("Data directory may be out of sync master, rewind may be needed.")
 
-        if self._need_rewind:
+        if self._need_rewind and leader and leader.conn_url and self.can_rewind:
             logger.info("rewind flag is set")
 
-            if self.is_running() and not self.stop():
+            if self.is_running() and not self.stop(checkpoint=False):
                 return logger.warning('Can not run pg_rewind because postgres is still running')
 
             # prepare pg_rewind connection
@@ -874,12 +1050,13 @@ class Postgresql(object):
         else:
             self.write_recovery_conf(primary_conninfo)
             if recovery:
-                self.start()
+                self.start(timeout=timeout)
             else:
                 self.restart()
             self.set_role('replica')
 
         if change_role:
+            # TODO: postpone this until start completes, or maybe do even earlier
             self.call_nowait(ACTION_ON_ROLE_CHANGE)
         return True
 
@@ -935,11 +1112,20 @@ $$""".format(name, ' '.join(options)), name, password, password)
 
     def xlog_position(self, retry=True):
         stmt = """SELECT pg_xlog_location_diff(CASE WHEN pg_is_in_recovery()
-                                                    THEN pg_last_xlog_replay_location()
+                                                    THEN COALESCE(pg_last_xlog_receive_location(),
+                                                                  pg_last_xlog_replay_location())
                                                     ELSE pg_current_xlog_location()
                                                END, '0/0')::bigint"""
+
+        # This method could be called from different threads (simultaneously with some other `_query` calls).
+        # If it is called not from main thread we will create a new cursor to execute statement.
+        if current_thread().ident == self.__thread_ident:
             return (self.query(stmt) if retry else self._query(stmt)).fetchone()[0]
 
+        with self.connection().cursor() as cursor:
+            cursor.execute(stmt)
+            return cursor.fetchone()[0]
+
     def load_replication_slots(self):
         if self.use_slots and self._schedule_load_slots:
             cursor = self._query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")

postgres0.yml

@@ -22,7 +22,8 @@ bootstrap:
     loop_wait: 10
     retry_timeout: 10
     maximum_lag_on_failover: 1048576
-    synchronous_mode: false
+#    master_start_timeout: 300
+#    synchronous_mode: false
     postgresql:
       use_pg_rewind: true
 #      use_slots: true

tests/test_api.py

@@ -63,6 +63,10 @@ class MockHa(object):
     def schedule_future_restart(data):
         return True
 
+    @staticmethod
+    def is_lagging(xlog):
+        return False
+
     @staticmethod
     def get_effective_tags():
         return {'nosync': True}
@@ -233,6 +237,10 @@ class TestRestApiHandler(unittest.TestCase):
 
         mock_dcs.get_cluster.return_value.is_paused.return_value = True
         MockRestApiServer(RestApiHandler, make_request(schedule='2016-08-42 12:45TZ+1', role='master'))
+        # Valid timeout
+        MockRestApiServer(RestApiHandler, make_request(timeout='60s'))
+        # Invalid timeout
+        MockRestApiServer(RestApiHandler, make_request(timeout='42towels'))
 
     def test_do_DELETE_restart(self):
         for retval in (True, False):

tests/test_config.py

@@ -20,9 +20,10 @@ class TestConfig(unittest.TestCase):
     def test_no_config(self):
         self.assertRaises(SystemExit, Config)
 
-    @patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception))
     def test_set_dynamic_configuration(self):
+        with patch.object(Config, '_build_effective_configuration', Mock(side_effect=Exception)):
             self.assertIsNone(self.config.set_dynamic_configuration({'foo': 'bar'}))
+        self.assertTrue(self.config.set_dynamic_configuration({'synchronous_mode': True}))
 
     def test_reload_local_configuration(self):
         os.environ.update({

tests/test_ctl.py

@@ -237,6 +237,9 @@ class TestCtl(unittest.TestCase):
         assert 'Error: PostgreSQL version' in result.output
         assert result.exit_code == 1
 
+        result = self.runner.invoke(ctl, ['restart', 'alpha', '--pending', '--force', '--timeout', '10min'])
+        assert result.exit_code == 0
+
         with patch('requests.delete', Mock(return_value=MockResponse(500))):
             # normal restart, the schedule is actually parsed, but not validated in patronictl
             result = self.runner.invoke(ctl, ['restart', 'alpha', 'other', '--force',

tests/test_etcd.py

@@ -42,7 +42,7 @@ def requests_get(url, **kwargs):
     if url.startswith('http://local'):
         raise requests.exceptions.RequestException()
     elif ':8011/patroni' in url:
-        response.content = '{"role": "replica", "xlog": {"replayed_location": 0}, "tags": {}}'
+        response.content = '{"role": "replica", "xlog": {"received_location": 0}, "tags": {}}'
     elif url.endswith('/members'):
         response.content = '[{}]' if url.startswith('http://error') else members
     elif url.startswith('http://exhibitor'):

tests/test_ha.py

@@ -8,7 +8,7 @@ from patroni.config import Config
 from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState
 from patroni.dcs.etcd import Client
 from patroni.exceptions import DCSError, PostgresException
-from patroni.ha import Ha
+from patroni.ha import Ha, _MemberStatus
 from patroni.postgresql import Postgresql
 from patroni.utils import tzutc
 from test_etcd import socket_getaddrinfo, etcd_read, etcd_write, requests_get
@@ -53,6 +53,15 @@ def get_cluster_initialized_with_only_leader(failover=None):
     l = get_cluster_initialized_without_leader(leader=True, failover=failover).leader
     return get_cluster(True, l, [l], failover, None)
 
+
+def get_node_status(reachable=True, in_recovery=True, xlog_location=10, nofailover=False):
+    def fetch_node_status(e):
+        tags = {}
+        if nofailover:
+            tags['nofailover'] = True
+        return _MemberStatus(e, reachable, in_recovery, xlog_location, tags)
+    return fetch_node_status
+
 future_restart_time = datetime.datetime.now(tzutc) + datetime.timedelta(days=5)
 postmaster_start_time = datetime.datetime.now(tzutc)
 
@@ -100,7 +109,7 @@ def run_async(self, func, args=()):
 
 @patch.object(Postgresql, 'is_running', Mock(return_value=True))
 @patch.object(Postgresql, 'is_leader', Mock(return_value=True))
-@patch.object(Postgresql, 'xlog_position', Mock(return_value=0))
+@patch.object(Postgresql, 'xlog_position', Mock(return_value=10))
 @patch.object(Postgresql, 'call_nowait', Mock(return_value=True))
 @patch.object(Postgresql, 'data_directory_empty', Mock(return_value=False))
 @patch.object(Postgresql, 'controldata', Mock(return_value={'Database system identifier': '1234567890'}))
@@ -126,6 +135,7 @@ class TestHa(unittest.TestCase):
             mock_machines.__get__ = Mock(return_value=['http://remotehost:2379'])
             self.p = Postgresql({'name': 'postgresql0', 'scope': 'dummy', 'listen': '127.0.0.1:5432',
                                  'data_dir': 'data/postgresql0', 'retry_timeout': 10,
+                                 'maximum_lag_on_failover': 5,
                                  'authentication': {'superuser': {'username': 'foo', 'password': 'bar'},
                                                     'replication': {'username': '', 'password': ''}},
                                  'parameters': {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'foo': 'bar',
@@ -133,7 +143,6 @@ class TestHa(unittest.TestCase):
             self.p.set_state('running')
             self.p.set_role('replica')
             self.p.postmaster_start_time = MagicMock(return_value=str(postmaster_start_time))
-            self.p.check_replication_lag = true
             self.p.can_create_replica_without_replication_connection = MagicMock(return_value=False)
             self.e = get_dcs({'etcd': {'ttl': 30, 'host': 'ok:2379', 'scope': 'test',
                                        'name': 'foo', 'retry_timeout': 10}})
@@ -292,18 +301,20 @@ class TestHa(unittest.TestCase):
         self.assertIsNotNone(self.ha.reinitialize())
 
     def test_restart(self):
-        self.assertEquals(self.ha.restart(), (True, 'restarted successfully'))
+        self.assertEquals(self.ha.restart({}), (True, 'restarted successfully'))
+        self.p.restart = Mock(return_value=None)
+        self.assertEquals(self.ha.restart({}), (False, 'postgres is still starting'))
         self.p.restart = false
-        self.assertEquals(self.ha.restart(), (False, 'restart failed'))
+        self.assertEquals(self.ha.restart({}), (False, 'restart failed'))
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.ha.reinitialize()
-        self.assertEquals(self.ha.restart(), (False, 'reinitialize already in progress'))
+        self.assertEquals(self.ha.restart({}), (False, 'reinitialize already in progress'))
         with patch.object(self.ha, "restart_matches", return_value=False):
             self.assertEquals(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied"))
 
     def test_restart_in_progress(self):
         with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)):
-            self.ha.restart(run_async=True)
+            self.ha.restart({}, run_async=True)
             self.assertTrue(self.ha.restart_scheduled())
             self.assertEquals(self.ha.run_cycle(), 'not healthy enough for leader race')
 
@@ -319,6 +330,7 @@ class TestHa(unittest.TestCase):
     @patch('requests.get', requests_get)
     @patch('time.sleep', Mock())
     def test_manual_failover_from_leader(self):
+        self.ha.fetch_node_status = get_node_status()
         self.ha.has_lock = true
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', '', None))
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
@@ -329,7 +341,9 @@ class TestHa(unittest.TestCase):
         f = Failover(0, self.p.name, '', None)
         self.ha.cluster = get_cluster_initialized_with_leader(f)
         self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
-        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'})
+        self.ha.fetch_node_status = get_node_status(nofailover=True)
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
+        self.ha.fetch_node_status = get_node_status(xlog_location=1)
         self.assertEquals(self.ha.run_cycle(), 'no action.  i am the leader with the lock')
         # manual failover from the previous leader to us won't happen if we hold the nofailover flag
         self.ha.cluster = get_cluster_initialized_with_leader(Failover(0, 'blabla', self.p.name, None))
@@ -375,17 +389,17 @@ class TestHa(unittest.TestCase):
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'leader', None))
         self.p.set_role('replica')
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
-        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})  # accessible, in_recovery
+        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, self.p.name, '', None))
         self.assertEquals(self.ha.run_cycle(), 'following a different leader because i am not the healthiest node')
-        self.ha.fetch_node_status = lambda e: (e, False, True, 0, {})  # inaccessible, in_recovery
+        self.ha.fetch_node_status = get_node_status(reachable=False)  # inaccessible, in_recovery
         self.p.set_role('replica')
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
         # set failover flag to True for all members of the cluster
         # this should elect the current member, as we are not going to call the API for it.
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None))
-        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {'nofailover': 'True'})  # accessible, in_recovery
+        self.ha.fetch_node_status = get_node_status(nofailover=True)  # accessible, in_recovery
         self.p.set_role('replica')
         self.assertEquals(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
         # same as previous, but set the current member to nofailover. In no case it should be elected as a leader
@@ -409,21 +423,23 @@ class TestHa(unittest.TestCase):
     def test_is_healthiest_node(self):
         self.ha.state_handler.is_leader = false
         self.ha.patroni.nofailover = False
-        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})
+        self.ha.fetch_node_status = get_node_status()
         self.assertTrue(self.ha.is_healthiest_node())
+        with patch('patroni.postgresql.Postgresql.is_starting', return_value=True):
+            self.assertFalse(self.ha.is_healthiest_node())
         self.ha.is_paused = true
         self.assertFalse(self.ha.is_healthiest_node())
 
     def test__is_healthiest_node(self):
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.p.is_leader = false
-        self.ha.fetch_node_status = lambda e: (e, True, True, 0, {})  # accessible, in_recovery
+        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
         self.assertTrue(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        self.ha.fetch_node_status = lambda e: (e, True, False, 0, {})  # accessible, not in_recovery
+        self.ha.fetch_node_status = get_node_status(in_recovery=False)  # accessible, not in_recovery
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        self.ha.fetch_node_status = lambda e: (e, True, True, 1, {})  # accessible, in_recovery, xlog location ahead
+        self.ha.fetch_node_status = get_node_status(xlog_location=11)  # accessible, in_recovery, xlog location ahead
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
-        self.p.check_replication_lag = false
+        with patch('patroni.postgresql.Postgresql.xlog_position', return_value=1):
             self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
         self.ha.patroni.nofailover = True
         self.assertFalse(self.ha._is_healthiest_node(self.ha.old_cluster.members))
@@ -514,6 +530,70 @@ class TestHa(unittest.TestCase):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.assertEquals(self.ha.run_cycle(), 'PAUSE: DCS is not accessible')
 
+    @patch('patroni.ha.Ha.update_lock', return_value=True)
+    @patch('patroni.ha.Ha.demote')
+    def test_starting_timeout(self, demote, update_lock):
+        def check_calls(seq):
+            for mock, called in seq:
+                if called:
+                    mock.assert_called_once()
+                else:
+                    mock.assert_not_called()
+                mock.reset_mock()
+        self.ha.has_lock = true
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.p.check_for_startup = true
+        self.p.time_in_state = lambda: 30
+        self.assertEquals(self.ha.run_cycle(), 'PostgreSQL is still starting up, 270 seconds until timeout')
+        check_calls([(update_lock, True), (demote, False)])
+
+        self.p.time_in_state = lambda: 350
+        self.ha.fetch_node_status = get_node_status(reachable=False)  # inaccessible, in_recovery
+        self.assertEquals(self.ha.run_cycle(),
+                          'master start has timed out, but continuing to wait because failover is not possible')
+        check_calls([(update_lock, True), (demote, False)])
+
+        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
+        self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL because of startup timeout')
+        check_calls([(update_lock, True), (demote, True)])
+
+        update_lock.return_value = False
+        self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL while starting up because leader key was lost')
+        check_calls([(update_lock, True), (demote, True)])
+
+        self.ha.has_lock = false
+        self.p.is_leader = false
+        self.assertEquals(self.ha.run_cycle(), 'no action.  i am a secondary and i am following a leader')
+        check_calls([(update_lock, False), (demote, False)])
+
+    @patch('time.sleep', Mock())
+    def test_manual_failover_while_starting(self):
+        self.ha.has_lock = true
+        self.p.check_for_startup = true
+        f = Failover(0, self.p.name, '', None)
+        self.ha.cluster = get_cluster_initialized_with_leader(f)
+        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
+        self.assertEquals(self.ha.run_cycle(), 'manual failover: demoting myself')
+
+    @patch('patroni.ha.Ha.demote')
+    def test_failover_immediately_on_zero_master_start_timeout(self, demote):
+        self.p.is_running = false
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.patroni.config.set_dynamic_configuration({'master_start_timeout': 0})
+        self.ha.has_lock = true
+        self.ha.update_lock = true
+        self.ha.fetch_node_status = get_node_status()  # accessible, in_recovery
+        self.assertEquals(self.ha.run_cycle(), 'stopped PostgreSQL to fail over after a crash')
+        demote.assert_called_once()
+
+    @patch('time.sleep', Mock())
+    @patch('patroni.postgresql.Postgresql.follow')
+    def test_demote_immediate(self, follow):
+        self.ha.has_lock = true
+        self.e.get_cluster = Mock(return_value=get_cluster_initialized_without_leader())
+        self.ha.demote('immediate')
+        follow.assert_called_once_with(None, None, True, None, True)
+
     @patch('time.sleep', Mock())
     def test_process_sync_replication(self):
         self.ha.has_lock = true
@@ -648,7 +728,7 @@ class TestHa(unittest.TestCase):
             get_cluster_initialized_with_leader(sync=('leader', syncstandby))
             for syncstandby in ['other', None]])
 
-        self.ha.restart()
+        self.ha.restart({})
 
         mock_restart.assert_called_once()
         mock_sleep.assert_called()
@@ -656,7 +736,7 @@ class TestHa(unittest.TestCase):
         # Restart is still called when DCS connection fails
         mock_restart.reset_mock()
         self.ha.dcs.get_cluster = Mock(side_effect=DCSError("foo"))
-        self.ha.restart()
+        self.ha.restart({})
 
         mock_restart.assert_called_once()
 
@@ -665,7 +745,7 @@ class TestHa(unittest.TestCase):
         self.ha.dcs.get_cluster.reset_mock()
         self.ha.touch_member = Mock(return_value=False)
 
-        self.ha.restart()
+        self.ha.restart({})
 
         mock_restart.assert_called_once()
         self.ha.dcs.get_cluster.assert_not_called()

tests/test_patroni.py

@@ -4,7 +4,7 @@ import sys
 import time
 import unittest
 
-from mock import Mock, patch
+from mock import Mock, PropertyMock, patch
 from patroni.api import RestApiServer
 from patroni.async_executor import AsyncExecutor
 from patroni.dcs.etcd import Client
@@ -72,6 +72,10 @@ class TestPatroni(unittest.TestCase):
         mock_getpid.return_value = 2
         _main()
 
+        with patch('sys.frozen', Mock(return_value=True), create=True):
+            sys.argv = ['/patroni', 'pg_ctl_start', 'postgres', '-D', '/data', '--max_connections=100']
+            _main()
+
         mock_getpid.return_value = 1
 
         def mock_signal(signo, handler):
@@ -98,6 +102,7 @@ class TestPatroni(unittest.TestCase):
 
     @patch('patroni.config.Config.save_cache', Mock())
     @patch('patroni.config.Config.reload_local_configuration', Mock(return_value=True))
+    @patch.object(Postgresql, 'state', PropertyMock(return_value='running'))
     def test_run(self):
         self.p.sighup_handler()
         self.p.ha.dcs.watch = Mock(side_effect=SleepException)

tests/test_postgresql.py

@@ -8,9 +8,10 @@ import unittest
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 from patroni.dcs import Cluster, Leader, Member, SyncState
 from patroni.exceptions import PostgresException, PostgresConnectionException
-from patroni.postgresql import Postgresql
+from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.utils import RetryFailedError
 from six.moves import builtins
+from threading import Thread
 
 
 class MockCursor(object):
@@ -204,19 +205,64 @@ class TestPostgresql(unittest.TestCase):
     def test_delete_trigger_file(self):
         self.p.delete_trigger_file()
 
+    @patch('subprocess.Popen')
+    @patch.object(Postgresql, 'wait_for_startup')
+    @patch.object(Postgresql, 'wait_for_port_open')
     @patch.object(Postgresql, 'is_running')
-    def test_start(self, mock_is_running):
+    def test_start(self, mock_is_running, mock_wait_for_port_open, mock_wait_for_startup, mock_popen):
         mock_is_running.return_value = True
+        mock_wait_for_port_open.return_value = True
+        mock_wait_for_startup.return_value = False
+        mock_popen.stdout.readline.return_value = '123'
         self.assertTrue(self.p.start())
         mock_is_running.return_value = False
         open(os.path.join(self.data_dir, 'postmaster.pid'), 'w').close()
         pg_conf = os.path.join(self.data_dir, 'postgresql.conf')
         open(pg_conf, 'w').close()
-        self.assertTrue(self.p.start())
+        self.assertFalse(self.p.start())
         with open(pg_conf) as f:
             lines = f.readlines()
             self.assertTrue("f.oo = 'bar'\n" in lines)
 
+        mock_wait_for_startup.return_value = None
+        self.assertFalse(self.p.start(10))
+        self.assertIsNone(self.p.start())
+
+        mock_wait_for_port_open.return_value = False
+        self.assertFalse(self.p.start())
+
+    @patch.object(Postgresql, 'pg_isready')
+    @patch.object(Postgresql, 'read_pid_file')
+    @patch.object(Postgresql, 'is_pid_running')
+    @patch('patroni.postgresql.polling_loop', Mock(return_value=range(1)))
+    def test_wait_for_port_open(self, mock_is_pid_running, mock_read_pid_file, mock_pg_isready):
+        mock_is_pid_running.return_value = False
+        mock_pg_isready.return_value = STATE_NO_RESPONSE
+
+        # No pid file and postmaster death
+        mock_read_pid_file.return_value = {}
+        self.assertFalse(self.p.wait_for_port_open(42, 100., 1))
+
+        mock_is_pid_running.return_value = True
+
+        # timeout
+        mock_read_pid_file.return_value = {'pid', 1}
+        self.assertFalse(self.p.wait_for_port_open(42, 100., 1))
+
+        # Garbage pid
+        mock_read_pid_file.return_value = {'pid': 'garbage', 'start_time': '101', 'data_dir': '',
+                                           'socket_dir': '', 'port': '', 'listen_addr': ''}
+        self.assertFalse(self.p.wait_for_port_open(42, 100., 1))
+
+        # Not ready
+        mock_read_pid_file.return_value = {'pid': '42', 'start_time': '101', 'data_dir': '',
+                                           'socket_dir': '', 'port': '', 'listen_addr': ''}
+        self.assertFalse(self.p.wait_for_port_open(42, 100., 1))
+
+        # pg_isready failure
+        mock_pg_isready.return_value = 'garbage'
+        self.assertTrue(self.p.wait_for_port_open(42, 100., 1))
+
     @patch.object(Postgresql, 'is_running')
     def test_stop(self, mock_is_running):
         mock_is_running.return_value = True
@@ -342,8 +388,11 @@ class TestPostgresql(unittest.TestCase):
         self.assertRaises(PostgresConnectionException, self.p.query, 'RetryFailedError')
         self.assertRaises(psycopg2.OperationalError, self.p.query, 'blabla')
 
+    @patch.object(Postgresql, 'pg_isready', Mock(return_value=STATE_REJECT))
     def test_is_leader(self):
         self.assertTrue(self.p.is_leader())
+        with patch.object(Postgresql, '_query', Mock(side_effect=RetryFailedError(''))):
+            self.assertRaises(PostgresConnectionException, self.p.is_leader)
 
     def test_reload(self):
         self.assertTrue(self.p.reload())
@@ -362,6 +411,7 @@ class TestPostgresql(unittest.TestCase):
 
     def test_last_operation(self):
         self.assertEquals(self.p.last_operation(), '0')
+        Thread(target=self.p.last_operation).start()
 
     @patch('os.path.isfile', Mock(return_value=True))
     @patch('os.kill', Mock(side_effect=Exception))
@@ -385,9 +435,6 @@ class TestPostgresql(unittest.TestCase):
         self.p.query = Mock(side_effect=psycopg2.OperationalError("not supported"))
         self.assertTrue(self.p.stop())
 
-    def test_check_replication_lag(self):
-        self.assertTrue(self.p.check_replication_lag(0))
-
     @patch('os.rename', Mock())
     @patch('os.path.isdir', Mock(return_value=True))
     def test_move_data_directory(self):
@@ -421,12 +468,13 @@ class TestPostgresql(unittest.TestCase):
             self.assertFalse(self.p.run_bootstrap_post_init({'post_init': '/bin/false'}))
 
         with patch('subprocess.call', Mock(return_value=0)) as mock_method:
+            self.p._superuser.pop('username')
             self.assertTrue(self.p.run_bootstrap_post_init({'post_init': '/bin/false'}))
 
         mock_method.assert_called()
         args, kwargs = mock_method.call_args
         assert 'PGPASSFILE' in kwargs['env'].keys()
-        self.assertEquals(args[0], ['/bin/false', 'postgres://test@localhost:5432/postgres'])
+        self.assertEquals(args[0], ['/bin/false', 'postgres://localhost:5432/postgres'])
 
     @patch('patroni.postgresql.Postgresql.create_replica', Mock(return_value=0))
     def test_clone(self):
@@ -573,6 +621,88 @@ class TestPostgresql(unittest.TestCase):
         with patch.object(MockCursor, "execute", side_effect=psycopg2.Error):
             self.assertIsNone(self.p.postmaster_start_time())
 
+    def test_check_for_startup(self):
+        with patch('subprocess.call', return_value=0):
+            self.p._state = 'starting'
+            self.assertFalse(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'running')
+
+        with patch('subprocess.call', return_value=1):
+            self.p._state = 'starting'
+            self.assertTrue(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'starting')
+
+        with patch('subprocess.call', return_value=2):
+            self.p._state = 'starting'
+            self.assertFalse(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'start failed')
+
+        with patch('subprocess.call', return_value=0):
+            self.p._state = 'running'
+            self.assertFalse(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'running')
+
+        with patch('subprocess.call', return_value=127):
+            self.p._state = 'running'
+            self.assertFalse(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'running')
+
+            self.p._state = 'starting'
+            self.assertFalse(self.p.check_for_startup())
+            self.assertEquals(self.p.state, 'running')
+
+    def test_wait_for_startup(self):
+        state = {'sleeps': 0, 'num_rejects': 0, 'final_return': 0}
+
+        def increment_sleeps(*args):
+            print("Sleep")
+            state['sleeps'] += 1
+
+        def isready_return(*args):
+            ret = 1 if state['sleeps'] < state['num_rejects'] else state['final_return']
+            print("Isready {0} {1}".format(ret, state))
+            return ret
+
+        def time_in_state(*args):
+            return state['sleeps']
+
+        with patch('subprocess.call', side_effect=isready_return):
+            with patch('time.sleep', side_effect=increment_sleeps):
+                self.p.time_in_state = Mock(side_effect=time_in_state)
+
+                self.p._state = 'stopped'
+                self.assertTrue(self.p.wait_for_startup())
+                self.assertEquals(state['sleeps'], 0)
+
+                self.p._state = 'starting'
+                state['num_rejects'] = 5
+                self.assertTrue(self.p.wait_for_startup())
+                self.assertEquals(state['sleeps'], 5)
+
+                self.p._state = 'starting'
+                state['sleeps'] = 0
+                state['final_return'] = 2
+                self.assertFalse(self.p.wait_for_startup())
+
+                self.p._state = 'starting'
+                state['sleeps'] = 0
+                state['final_return'] = 0
+                self.assertFalse(self.p.wait_for_startup(timeout=2))
+                self.assertEquals(state['sleeps'], 3)
+
+    def test_read_pid_file(self):
+        pidfile = os.path.join(self.data_dir, 'postmaster.pid')
+        if os.path.exists(pidfile):
+            os.remove(pidfile)
+        self.assertEquals(self.p.read_pid_file(), {})
+
+    @patch('os.kill')
+    def test_is_pid_running(self, mock_kill):
+        mock_kill.return_value = True
+        self.assertTrue(self.p.is_pid_running(-100))
+        self.assertFalse(self.p.is_pid_running(0))
+        self.assertFalse(self.p.is_pid_running(None))
+
     def test_pick_sync_standby(self):
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
                           SyncState(0, self.me.name, self.leadermem.name))

tests/test_utils.py

@@ -2,7 +2,13 @@ import unittest
 
 from mock import Mock, patch
 from patroni.exceptions import PatroniException
-from patroni.utils import Retry, RetryFailedError
+from patroni.utils import Retry, RetryFailedError, polling_loop
+
+
+class TestUtils(unittest.TestCase):
+
+    def test_polling_loop(self):
+        self.assertEquals(list(polling_loop(0.001, interval=0.001)), [0])
 
 
 @patch('time.sleep', Mock())