Make sure CHECKPOINT is execute after promote (patroni#3368)

It was possible that `Rewind._checkpoint_task` wasn't reset on demote if CHECKPOINT wasn't yet finished, what resulted in using stale `result` when the next promote is triggered.

It is not easy to reproduce, but steps are the following:
1. failover to node1.
2. while CHECKPOINT after promote is still running, switchover to node2.
3. next failover/switchover to node1 results in missing CHECKPOINT.

To solve the problem we take following measures:
1. call `Rewind.reset_state()` before promote.
2. reset `Rewind._checkpoint_task` from trigger_check_diverged_lsn().

Besides that, we didn't check that CHECKPOINT during 2 actually finished successfully. If check implemented correctly chances to hit the problem would have been much smaller. However, there was still race condition, if switchover was triggered right after CHECKPOINT task finished.

Close patroni#3367

Author: CyberDem0n

patroni/ha.py

@@ -1121,6 +1121,7 @@ def enforce_primary_role(self, message: str, promote_message: str) -> str:
                 self._failsafe.set_is_active(0)
 
                 def before_promote():
+                    self._rewind.reset_state()  # make sure we will trigger checkpoint after promote
                     self.notify_mpp_coordinator('before_promote')
 
                 with self._async_response:

patroni/postgresql/rewind.py

@@ -73,6 +73,8 @@ def can_rewind_or_reinitialize_allowed(self) -> bool:
     def trigger_check_diverged_lsn(self) -> None:
         if self.can_rewind_or_reinitialize_allowed and self._state != REWIND_STATUS.NEED:
             self._state = REWIND_STATUS.CHECK
+        with self._checkpoint_task_lock:
+            self._checkpoint_task = None
 
     @staticmethod
     def check_leader_is_not_in_recovery(conn_kwargs: Dict[str, Any]) -> Optional[bool]:
@@ -306,10 +308,16 @@ def ensure_checkpoint_after_promote(self, wakeup: Callable[..., Any]) -> None:
         if self._state != REWIND_STATUS.CHECKPOINT and self._postgresql.is_primary():
             with self._checkpoint_task_lock:
                 if self._checkpoint_task:
+                    result = None
+
                     with self._checkpoint_task:
-                        if self._checkpoint_task.result is not None:
-                            self._state = REWIND_STATUS.CHECKPOINT
-                            self._checkpoint_task = None
+                        result = self._checkpoint_task.result
+
+                    if result is True:
+                        self._state = REWIND_STATUS.CHECKPOINT
+
+                    if result is not None:
+                        self._checkpoint_task = None
                 elif self._postgresql.get_primary_timeline() == self._postgresql.pg_control_timeline():
                     self._state = REWIND_STATUS.CHECKPOINT
                 else: