add comment

Before this PR, if walredo failed, we would still update `last_redo_at`.
This means the `maybe_quiesce()` would never kill that process, although
clearly something is wrong.

However, we don't want to kill the process immediately on each redo failure,
because, the redo failure could be caused by corrupted or malicious
data.

So, change `maybe_quiesce()` to determine inactivity based on last
*successful* `last_redo_at`.
The result is that a transient redo failure won't cause a kill.
If there are *only* redo failures, we'll spawn & kill walredo process
at frequency 1/(10*compaction_period).

pageserver/src/walredo.rs

@@ -93,7 +93,7 @@ struct ProcessOutput {
 pub struct PostgresRedoManager {
     tenant_shard_id: TenantShardId,
     conf: &'static PageServerConf,
-    last_redo_at: std::sync::Mutex<Option<Instant>>,
+    last_successful_redo_at: std::sync::Mutex<Option<Instant>>,
     redo_process: RwLock<Option<Arc<WalRedoProcess>>>,
 }
 
@@ -193,7 +193,7 @@ impl PostgresRedoManager {
         PostgresRedoManager {
             tenant_shard_id,
             conf,
-            last_redo_at: std::sync::Mutex::default(),
+            last_successful_redo_at: std::sync::Mutex::default(),
             redo_process: RwLock::new(None),
         }
     }
@@ -202,9 +202,21 @@ impl PostgresRedoManager {
     /// rely on our owner calling this function periodically in its own housekeeping
     /// loops.
     pub(crate) fn maybe_quiesce(&self, idle_timeout: Duration) {
-        if let Ok(g) = self.last_redo_at.try_lock() {
-            if let Some(last_redo_at) = *g {
-                if last_redo_at.elapsed() >= idle_timeout {
+        if let Ok(g) = self.last_successful_redo_at.try_lock() {
+            if let Some(last_successful_redo_at) = *g {
+                // Kill the walredo process if
+                // - it has been unused for `idle_timeout`
+                // - it has been used, but, without success.
+                // The former is just good housekeeping.
+                // The latter adds robustness for the case where something is wrong
+                // with the walredo process.
+                //
+                // Note that we don't want to kill the process immediately on each redo failure.
+                // The reason is that the redo failure could be caused by corrupted or malicious data.
+                // We don't want to get into a kill-respawn loop in that case.
+                // So, we piggy-back on the quiescing mechanism,
+                // resulting in a max kill-respawn frequency of `1/idle_timeout`.
+                if last_successful_redo_at.elapsed() >= idle_timeout {
                     drop(g);
                     let mut guard = self.redo_process.write().unwrap();
                     *guard = None;
@@ -227,8 +239,32 @@ impl PostgresRedoManager {
         wal_redo_timeout: Duration,
         pg_version: u32,
     ) -> anyhow::Result<Bytes> {
-        *(self.last_redo_at.lock().unwrap()) = Some(Instant::now());
+        let res = self.apply_batch_postgres0(
+            key,
+            lsn,
+            base_img,
+            base_img_lsn,
+            records,
+            wal_redo_timeout,
+            pg_version,
+        );
+        if res.is_ok() {
+            *self.last_successful_redo_at.lock().unwrap() = Some(Instant::now());
+        }
+        res
+    }
 
+    #[allow(clippy::too_many_arguments)]
+    fn apply_batch_postgres0(
+        &self,
+        key: Key,
+        lsn: Lsn,
+        base_img: Option<Bytes>,
+        base_img_lsn: Lsn,
+        records: &[(Lsn, NeonWalRecord)],
+        wal_redo_timeout: Duration,
+        pg_version: u32,
+    ) -> anyhow::Result<Bytes> {
         let (rel, blknum) = key_to_rel_block(key).context("invalid record")?;
         const MAX_RETRY_ATTEMPTS: u32 = 1;
         let mut n_attempts = 0u32;