Keep walproposer alive until shutdown checkpoint is safe on safekepeers

libs/walproposer/src/api_bindings.rs

@@ -324,11 +324,11 @@ extern "C" fn finish_sync_safekeepers(wp: *mut WalProposer, lsn: XLogRecPtr) {
  }
 }
 
-extern "C" fn process_safekeeper_feedback(wp: *mut WalProposer, commit_lsn: XLogRecPtr) {
+extern "C" fn process_safekeeper_feedback(wp: *mut WalProposer) {
  unsafe {
  let callback_data = (*(*wp).config).callback_data;
  let api = callback_data as *mut Box<dyn ApiImpl>;
- (*api).process_safekeeper_feedback(&mut (*wp), commit_lsn)
+ (*api).process_safekeeper_feedback(&mut (*wp))
  }
 }
 

libs/walproposer/src/walproposer.rs

@@ -142,7 +142,7 @@ pub trait ApiImpl {
  todo!()
  }
 
- fn process_safekeeper_feedback(&self, _wp: &mut WalProposer, _commit_lsn: u64) {
+ fn process_safekeeper_feedback(&mut self, _wp: &mut WalProposer) {
  todo!()
  }
 

pgxn/neon/walproposer.c

@@ -1220,7 +1220,7 @@ PrepareAppendRequest(WalProposer *wp, AppendRequestHeader *req, XLogRecPtr begin
  req->epochStartLsn = wp->propEpochStartLsn;
  req->beginLsn = beginLsn;
  req->endLsn = endLsn;
- req->commitLsn = GetAcknowledgedByQuorumWALPosition(wp);
+ req->commitLsn = wp->commitLsn;
  req->truncateLsn = wp->truncateLsn;
  req->proposerId = wp->greetRequest.proposerId;
 }
@@ -1405,7 +1405,7 @@ static bool
 RecvAppendResponses(Safekeeper *sk)
 {
  WalProposer *wp = sk->wp;
- XLogRecPtr minQuorumLsn;
+ XLogRecPtr newCommitLsn;
  bool readAnything = false;
 
  while (true)
@@ -1444,18 +1444,19 @@ RecvAppendResponses(Safekeeper *sk)
  if (!readAnything)
  return sk->state == SS_ACTIVE;
 
- HandleSafekeeperResponse(wp);
-
+ /* update commit_lsn */
+ newCommitLsn = GetAcknowledgedByQuorumWALPosition(wp);
  /*
- * Also send the new commit lsn to all the safekeepers.
+ * Send the new value to all safekeepers.
  */
- minQuorumLsn = GetAcknowledgedByQuorumWALPosition(wp);
- if (minQuorumLsn > wp->lastSentCommitLsn)
+ if (newCommitLsn > wp->commitLsn)
  {
+ wp->commitLsn = newCommitLsn;
  BroadcastAppendRequest(wp);
- wp->lastSentCommitLsn = minQuorumLsn;
  }
 
+ HandleSafekeeperResponse(wp);
+
  return sk->state == SS_ACTIVE;
 }
 
@@ -1632,11 +1633,9 @@ GetDonor(WalProposer *wp, XLogRecPtr *donor_lsn)
 static void
 HandleSafekeeperResponse(WalProposer *wp)
 {
- XLogRecPtr minQuorumLsn;
  XLogRecPtr candidateTruncateLsn;
 
- minQuorumLsn = GetAcknowledgedByQuorumWALPosition(wp);
- wp->api.process_safekeeper_feedback(wp, minQuorumLsn);
+ wp->api.process_safekeeper_feedback(wp);
 
  /*
  * Try to advance truncateLsn -- the last record flushed to all
@@ -1649,7 +1648,7 @@ HandleSafekeeperResponse(WalProposer *wp)
  * can't commit entries from previous term' in Raft); 2)
  */
  candidateTruncateLsn = CalculateMinFlushLsn(wp);
- candidateTruncateLsn = Min(candidateTruncateLsn, minQuorumLsn);
+ candidateTruncateLsn = Min(candidateTruncateLsn, wp->commitLsn);
  if (candidateTruncateLsn > wp->truncateLsn)
  {
  wp->truncateLsn = candidateTruncateLsn;

pgxn/neon/walproposer.h

@@ -564,7 +564,7 @@ typedef struct walproposer_api
  * backpressure feedback and to confirm WAL persistence (has been commited
  * on the quorum of safekeepers).
  */
- void (*process_safekeeper_feedback) (WalProposer *wp, XLogRecPtr commitLsn);
+ void (*process_safekeeper_feedback) (WalProposer *wp);
 
  /*
  * Write a log message to the internal log processor. This is used only
@@ -646,8 +646,8 @@ typedef struct WalProposer
  /* WAL has been generated up to this point */
  XLogRecPtr availableLsn;
 
- /* last commitLsn broadcasted to safekeepers */
- XLogRecPtr lastSentCommitLsn;
+ /* cached GetAcknowledgedByQuorumWALPosition result */
+ XLogRecPtr commitLsn;
 
  ProposerGreeting greetRequest;
 

pgxn/neon/walproposer_pg.c

@@ -68,6 +68,8 @@ static WalproposerShmemState *walprop_shared;
 static WalProposerConfig walprop_config;
 static XLogRecPtr sentPtr = InvalidXLogRecPtr;
 static const walproposer_api walprop_pg;
+static volatile sig_atomic_t got_SIGUSR2 = false;
+static bool reported_sigusr2 = false;
 
 static void nwp_shmem_startup_hook(void);
 static void nwp_register_gucs(void);
@@ -101,6 +103,8 @@ static void add_nwr_event_set(Safekeeper *sk, uint32 events);
 static void update_nwr_event_set(Safekeeper *sk, uint32 events);
 static void rm_safekeeper_event_set(Safekeeper *to_remove, bool is_sk);
 
+static void CheckGracefulShutdown(WalProposer *wp);
+
 static XLogRecPtr GetLogRepRestartLSN(WalProposer *wp);
 
 static void
@@ -492,6 +496,24 @@ walprop_pg_init_standalone_sync_safekeepers(void)
  BackgroundWorkerUnblockSignals();
 }
 
+/*
+ * We pretend to be a walsender process, and the lifecycle of a walsender is
+ * slightly different than other procesess. At shutdown, walsender processes
+ * stay alive until the very end, after the checkpointer has written the
+ * shutdown checkpoint. When the checkpointer exits, the postmaster sends all
+ * remaining walsender processes SIGUSR2. On receiving SIGUSR2, we try to send
+ * the remaining WAL, and then exit. This ensures that the checkpoint record
+ * reaches durable storage (in safekeepers), before the server shuts down
+ * completely.
+ */
+static void
+walprop_sigusr2(SIGNAL_ARGS)
+{
+ got_SIGUSR2 = true;
+
+ SetLatch(MyLatch);
+}
+
 static void
 walprop_pg_init_bgworker(void)
 {
@@ -503,6 +525,7 @@ walprop_pg_init_bgworker(void)
  pqsignal(SIGUSR1, procsignal_sigusr1_handler);
  pqsignal(SIGHUP, SignalHandlerForConfigReload);
  pqsignal(SIGTERM, die);
+ pqsignal(SIGUSR2, walprop_sigusr2);
 
  BackgroundWorkerUnblockSignals();
 
@@ -1075,14 +1098,26 @@ StartProposerReplication(WalProposer *wp, StartReplicationCmd *cmd)
 #endif
 
  /*
- * When we first start replication the standby will be behind the primary.
- * For some applications, for example synchronous replication, it is
- * important to have a clear state for this initial catchup mode, so we
- * can trigger actions when we change streaming state later. We may stay
- * in this state for a long time, which is exactly why we want to be able
- * to monitor whether or not we are still here.
+ * XXX: Move straight to STOPPING state, skipping the STREAMING state.
+ *
+ * This is a bit weird. Normal walsenders stay in STREAMING state, until
+ * the checkpointer signals them that it is about to start writing the
+ * shutdown checkpoint. The walsenders acknowledge that they have received
+ * that signal by switching to STOPPING state. That tells the walsenders
+ * that they must not write any new WAL.
+ *
+ * However, we cannot easily intercept that signal from the checkpointer.
+ * It's sent by WalSndInitStopping(), using
+ * SendProcSignal(PROCSIGNAL_WALSND_INIT_STOPPING). It's received by
+ * HandleWalSndInitStopping, which sets a process-local got_STOPPING flag.
+ * However, that's all private to walsender.c.
+ *
+ * We don't need to do anything special upon receiving the signal, the
+ * walproposer doesn't write any WAL anyway, so we skip the STREAMING
+ * state and go directly to STOPPING mode. That way, the checkpointer
+ * won't wait for us.
  */
- WalSndSetState(WALSNDSTATE_CATCHUP);
+ WalSndSetState(WALSNDSTATE_STOPPING);
 
  /*
  * Don't allow a request to stream from a future point in WAL that hasn't
@@ -1122,6 +1157,8 @@ StartProposerReplication(WalProposer *wp, StartReplicationCmd *cmd)
 static void
 WalSndLoop(WalProposer *wp)
 {
+ XLogRecPtr flushPtr;
+
  /* Clear any already-pending wakeups */
  ResetLatch(MyLatch);
 
@@ -1130,9 +1167,6 @@ WalSndLoop(WalProposer *wp)
  CHECK_FOR_INTERRUPTS();
 
  XLogBroadcastWalProposer(wp);
-
- if (MyWalSnd->state == WALSNDSTATE_CATCHUP)
- WalSndSetState(WALSNDSTATE_STREAMING);
  WalProposerPoll(wp);
  }
 }
@@ -1745,6 +1779,9 @@ walprop_pg_wait_event_set(WalProposer *wp, long timeout, Safekeeper **sk, uint32
  {
  ConditionVariableCancelSleep();
  ResetLatch(MyLatch);
+
+ CheckGracefulShutdown(wp);
+
  *events = WL_LATCH_SET;
  return 1;
  }
@@ -1798,6 +1835,41 @@ walprop_pg_finish_sync_safekeepers(WalProposer *wp, XLogRecPtr lsn)
  exit(0);
 }
 
+/*
+ * Like vanilla walsender, on sigusr2 send all remaining WAL and exit.
+ *
+ * Note that unlike sync-safekeepers waiting here is not reliable: we
+ * don't check that majority of safekeepers received and persisted
+ * commit_lsn -- only that walproposer reached it (which immediately
+ * broadcasts new value). Doing that without incurring redundant control
+ * file syncing would need wp -> sk protocol change. OTOH unlike
+ * sync-safekeepers which must bump commit_lsn or basebackup will fail,
+ * this catchup is important only for tests where safekeepers/network
+ * don't crash on their own.
+ */
+static void
+CheckGracefulShutdown(WalProposer *wp)
+{
+ if (got_SIGUSR2)
+ {
+ if (!reported_sigusr2)
+ {
+ XLogRecPtr flushPtr = walprop_pg_get_flush_rec_ptr(wp);
+
+ wpg_log(LOG, "walproposer will send and wait for remaining WAL between %X/%X and %X/%X",
+ LSN_FORMAT_ARGS(wp->commitLsn), LSN_FORMAT_ARGS(flushPtr));
+ reported_sigusr2 = true;
+ }
+
+ if (wp->commitLsn >= walprop_pg_get_flush_rec_ptr(wp))
+ {
+ wpg_log(LOG, "walproposer sent all WAL up to %X/%X, exiting",
+ LSN_FORMAT_ARGS(wp->commitLsn));
+ proc_exit(0);
+ }
+ }
+}
+
 /*
  * Choose most advanced PageserverFeedback and set it to *rf.
  */
@@ -1878,7 +1950,7 @@ CombineHotStanbyFeedbacks(HotStandbyFeedback *hs, WalProposer *wp)
  * None of that is functional in sync-safekeepers.
  */
 static void
-walprop_pg_process_safekeeper_feedback(WalProposer *wp, XLogRecPtr commitLsn)
+walprop_pg_process_safekeeper_feedback(WalProposer *wp)
 {
  HotStandbyFeedback hsFeedback;
  XLogRecPtr oldDiskConsistentLsn;
@@ -1893,10 +1965,10 @@ walprop_pg_process_safekeeper_feedback(WalProposer *wp, XLogRecPtr commitLsn)
  replication_feedback_set(&quorumFeedback.rf);
  SetZenithCurrentClusterSize(quorumFeedback.rf.currentClusterSize);
 
- if (commitLsn > quorumFeedback.flushLsn || oldDiskConsistentLsn != quorumFeedback.rf.disk_consistent_lsn)
+ if (wp->commitLsn > quorumFeedback.flushLsn || oldDiskConsistentLsn != quorumFeedback.rf.disk_consistent_lsn)
  {
- if (commitLsn > quorumFeedback.flushLsn)
- quorumFeedback.flushLsn = commitLsn;
+ if (wp->commitLsn > quorumFeedback.flushLsn)
+ quorumFeedback.flushLsn = wp->commitLsn;
 
  /*
  * Advance the replication slot to commitLsn. WAL before it is
@@ -1929,6 +2001,8 @@ walprop_pg_process_safekeeper_feedback(WalProposer *wp, XLogRecPtr commitLsn)
  XidFromFullTransactionId(hsFeedback.catalog_xmin),
  EpochFromFullTransactionId(hsFeedback.catalog_xmin));
  }
+
+ CheckGracefulShutdown(wp);
 }
 
 static XLogRecPtr

safekeeper/tests/walproposer_sim/walproposer_api.rs

@@ -196,6 +196,7 @@ pub struct SimulationApi {
  safekeepers: RefCell<Vec<SafekeeperConn>>,
  disk: Arc<DiskWalProposer>,
  redo_start_lsn: Option<Lsn>,
+ last_logged_commit_lsn: u64,
  shmem: UnsafeCell<walproposer::bindings::WalproposerShmemState>,
  config: Config,
  event_set: RefCell<Option<EventSet>>,
@@ -228,6 +229,7 @@ impl SimulationApi {
  safekeepers: RefCell::new(sk_conns),
  disk: args.disk,
  redo_start_lsn: args.redo_start_lsn,
+ last_logged_commit_lsn: 0,
  shmem: UnsafeCell::new(walproposer::bindings::WalproposerShmemState {
  mutex: 0,
  feedback: PageserverFeedback {
@@ -596,14 +598,11 @@ impl ApiImpl for SimulationApi {
  }
  }
 
- fn process_safekeeper_feedback(
- &self,
- wp: &mut walproposer::bindings::WalProposer,
- commit_lsn: u64,
- ) {
- debug!("process_safekeeper_feedback, commit_lsn={}", commit_lsn);
- if commit_lsn > wp.lastSentCommitLsn {
- self.os.log_event(format!("commit_lsn;{}", commit_lsn));
+ fn process_safekeeper_feedback(&mut self, wp: &mut walproposer::bindings::WalProposer) {
+ debug!("process_safekeeper_feedback, commit_lsn={}", wp.commitLsn);
+ if wp.commitLsn > self.last_logged_commit_lsn {
+ self.os.log_event(format!("commit_lsn;{}", wp.commitLsn));
+ self.last_logged_commit_lsn = wp.commitLsn;
  }
  }