RFC: move decision-making of desired VM size to VM monitor

docs/rfcs/033-compute-controlled-autoscaling.md

@@ -0,0 +1,181 @@
+# Compute controlled autoscaling
+
+## Summary
+
+The proposal is to move the responsibility for deciding what the
+desired size of a VM is, from the autoscaler agent to the VM monitor.
+
+## Motivation
+
+Currently, the decision to upscale or downscale a VM is made outside
+the VM, in the autoscaler-agent, based on the load average and memory
+consumption. In addition to that, the VM can override that decision by
+making an explicit upscale request, which will bump up the VM size for
+a certain period of time.
+
+Moving the decision-making to the compute has several benefits:
+
+- It allows the compute to explicitly request a certain size, which
+  can be useful for testing and debugging purposes. We can expose that
+  as SQL-callable functions to provide an "escape hatch" for the cases
+  where the automatic algorithm doesn't work well.
+
+- It makes it possible for the compute to upscale _before_ making a
+  large memory allocation, for example, avoiding OOM. We battled with
+  this problem with pgvector, which makes one giant dynamic shared
+  memory allocation at CREATE INDEX, which would immediately OOM if
+  the instance wasn't already scaled up to accommodate it. We mostly
+  solved it by enabling swap and setting dynamic_shared_memory=mmap,
+  but it's fiddly. Explicitly scaling up would be more robust.
+
+- It's simpler. Even with no change to the actual algorithm used, it
+  is more straightforward to measure the CPU and memory usage directly
+  in the VM. The vm-monitor can read the statistics directly from the
+  OS, while the agent needs to receive them through Vector. Also, we
+  already had the "emergency upscale" request codepath that the
+  vm-monitor could use to request immediate upscale.  This proposal
+  eliminates that case as a separate thing, all downscale/upscale
+  requests follow the same path.
+
+- We can improve the scaling algorithm to take into account more
+  information about query plans and resource usage, without having to
+  expose all the information to the agent outside the VM.
+
+- Organizationally, the compute team has the PostgreSQL expertise to
+  improve the algorithm for choosing the optimal compute size in the
+  long run.
+
+
+## Scope / Non Goals
+
+This proposal is only about choosing the *desired* size of the
+VM. There can be reasons that the agent / scheduler cannot grant the
+desired resources, e.g if the host is overbooked. Downscaling to the
+desired size might also not be possible, because of an inbalance in
+Linux memory zones, or if there is a sudden spike in memory usage
+after the desired size was last calculated. This proposal doesn't
+change what happens in those cases.
+
+What happens between the autoscaler agent, neonvm controller, and to
+resize the VM is out of scope. Scheduling VMs on nodes is also out of
+scope. Choosing the desired size of a VM is a purely VM-local
+decision, and doesn't take into account overall system load.
+
+## Impacted components (e.g. pageserver, safekeeper, console, etc)
+
+VM monitor (libs/vm_monitor) and autoscaling agent.
+
+## Proposed implementation
+
+The current VM monitor <-> autoscaling agent protocol includes an
+UpscaleRequest message. The VM monitor sends that message to request
+upscaling when memory is running low. The proposal is to add a new
+ScaleRequest message, which is similar to UpscaleRequest, but tells
+the agent directly what the desired CPU and memory size of the VM
+is. When the autoscaler agent receives that message, it tries to
+upscale/downscale the VM to the size specified in the message, and
+disables theusual metrics-based algorithm.
+
+### Backwards compatibility
+
+This proposal adds a new ScaleRequest message, but doesn't modify the
+old message types. The migration path is to deploy the support for the
+new message in the autoscaler agent first. After that, new computes
+can start using a new version of the VM monitor, which issues
+ScaleRequest messages. After all old computes have expired, the
+metrics-based algorithm and the code to handle the old UpscaleRequest
+message can be removed from the autocaler agent.
+
+### Reliability, failure modes and corner cases
+
+The VM monitor's and autoscaler agent's idea of what the desired and
+current VM size is might go out of sync, e.g. if the autoscaler agent
+is restarted or some messages are lost. Or if upscaling or downscaling
+fails for some reason.  The VM monitor will periodically resend the
+ScaleRequest if the actual size of the VM doesn't match the desired
+size.
+
+### Interaction/Sequence diagram
+
+Currently, before this proposal, there are two ways that upscaling /
+downscaling can be initiated. It can be initiated by an algorithm that
+runs in the autoscaler agent, which makes the decision based on metrics
+received from Vector running inside the VM:
+
+Current metrics-based scaling:
+
+
++--------+   (1) metrics    +------------+
+| Vector |  ------------->  | Autoscaler |
++--------+                  | agent      |
+                            +------------+
+			           |
+			           |
+           (2) UpscaleNotification |
+               / DownscaleRequest  |
+			           |
++------------+		           |
+| VM monitor |  <------------------/
++------------+
+                                   |
+                                   |
++------+      (3) QMP plug/unplug  |
+| QEMU | <-------------------------/
++------+
+
+The second way to initiate upscaling (but not downscaling!), is that
+the VM monitor requests upscaling by sending an UpscaleRequest message
+to the agent. The upscale request overrides the metrics-based algorithm's
+decision for some period of time:
+
++------------+   (1) UpscaleRequest
+| VM monitor |  <------------------\
++------------+                     |
+                                   V
+                            +------------+
+                            | Autoscaler |
+                            | agent      |
+                            +------------+
+                                   |
++------+      (2) QMP hotplug      |
+| QEMU | <-------------------------/
++------+
+
+The proposal is to replace the above two mechanisms with a new
+mechanism that is very similar to the old UpscaleRequest message. The
+difference is that the new ScaleRequest message can initiate immediate
+upscaling but also downscaling, and when the new mechanism is used,
+the metrics-based algorithm in the agent is disabled for the VM, and
+upscaling/downscaling is only initiated when a new ScaleRequest is
+received from the VM monitor.
+
+
++------------+   (1) ScaleRequest
+| VM monitor |  <------------------\
++------------+                     |
+                                   V
+                            +------------+
+                            | Autoscaler |
+                            | agent      |
+                            +------------+
+                                   |
++------+      (2) QMP hotplug      |
+| QEMU | <-------------------------/
++------+
+
+
+### Scalability (if relevant)
+
+n/a
+
+### Security implications (if relevant)
+
+This doesn't open any new communication paths.
+
+### Unresolved questions (if relevant)
+
+## Alternative implementation (if relevant)
+
+## Pros/cons of proposed approaches (if relevant)
+
+## Definition of Done (if relevant)

libs/postgres_ffi/src/pg_constants.rs

@@ -79,8 +79,6 @@ pub const XLOG_XACT_PREPARE: u8 = 0x10;
 pub const XLOG_XACT_ABORT: u8 = 0x20;
 pub const XLOG_XACT_COMMIT_PREPARED: u8 = 0x30;
 pub const XLOG_XACT_ABORT_PREPARED: u8 = 0x40;
-pub const XLOG_XACT_ASSIGNMENT: u8 = 0x50;
-pub const XLOG_XACT_INVALIDATIONS: u8 = 0x60;
 
 // From standbydefs.h
 pub const XLOG_RUNNING_XACTS: u8 = 0x10;
@@ -108,6 +106,12 @@ pub const XACT_XINFO_HAS_ORIGIN: u32 = 1u32 << 5;
 // pub const XACT_XINFO_HAS_AE_LOCKS: u32 = 1u32 << 6;
 // pub const XACT_XINFO_HAS_GID: u32 = 1u32 << 7;
 
+// From pg_control.h and rmgrlist.h
+pub const XLOG_NEXTOID: u8 = 0x30;
+pub const XLOG_SWITCH: u8 = 0x40;
+pub const XLOG_FPI_FOR_HINT: u8 = 0xA0;
+pub const XLOG_FPI: u8 = 0xB0;
+
 // From multixact.h
 pub const FIRST_MULTIXACT_ID: u32 = 1;
 pub const MAX_MULTIXACT_ID: u32 = 0xFFFFFFFF;
@@ -135,20 +139,12 @@ pub const MULTIXACT_MEMBERS_PER_PAGE: u16 =
 pub const XLOG_HEAP_INSERT: u8 = 0x00;
 pub const XLOG_HEAP_DELETE: u8 = 0x10;
 pub const XLOG_HEAP_UPDATE: u8 = 0x20;
-pub const XLOG_HEAP_TRUNCATE: u8 = 0x30;
 pub const XLOG_HEAP_HOT_UPDATE: u8 = 0x40;
-pub const XLOG_HEAP_CONFIRM: u8 = 0x50;
 pub const XLOG_HEAP_LOCK: u8 = 0x60;
-pub const XLOG_HEAP_INPLACE: u8 = 0x70;
 pub const XLOG_HEAP_INIT_PAGE: u8 = 0x80;
-pub const XLOG_HEAP2_REWRITE: u8 = 0x00;
-pub const XLOG_HEAP2_PRUNE: u8 = 0x10;
-pub const XLOG_HEAP2_VACUUM: u8 = 0x20;
-pub const XLOG_HEAP2_FREEZE_PAGE: u8 = 0x30;
 pub const XLOG_HEAP2_VISIBLE: u8 = 0x40;
 pub const XLOG_HEAP2_MULTI_INSERT: u8 = 0x50;
 pub const XLOG_HEAP2_LOCK_UPDATED: u8 = 0x60;
-pub const XLOG_HEAP2_NEW_CID: u8 = 0x70;
 pub const XLH_LOCK_ALL_FROZEN_CLEARED: u8 = 0x01;
 pub const XLH_INSERT_ALL_FROZEN_SET: u8 = (1 << 5) as u8;
 pub const XLH_INSERT_ALL_VISIBLE_CLEARED: u8 = (1 << 0) as u8;
@@ -171,21 +167,9 @@ pub const RM_RELMAP_ID: u8 = 7;
 pub const RM_STANDBY_ID: u8 = 8;
 pub const RM_HEAP2_ID: u8 = 9;
 pub const RM_HEAP_ID: u8 = 10;
-pub const RM_BTREE_ID: u8 = 11;
-pub const RM_HASH_ID: u8 = 12;
-pub const RM_GIN_ID: u8 = 13;
-pub const RM_GIST_ID: u8 = 14;
-pub const RM_SEQ_ID: u8 = 15;
-pub const RM_SPGIST_ID: u8 = 16;
-pub const RM_BRIN_ID: u8 = 17;
-pub const RM_COMMIT_TS_ID: u8 = 18;
 pub const RM_REPLORIGIN_ID: u8 = 19;
-pub const RM_GENERIC_ID: u8 = 20;
 pub const RM_LOGICALMSG_ID: u8 = 21;
 
-// from relmapper.h
-pub const XLOG_RELMAP_UPDATE: u8 = 0x0;
-
 // from neon_rmgr.h
 pub const RM_NEON_ID: u8 = 134;
 
@@ -235,22 +219,8 @@ pub const INVALID_TRANSACTION_ID: u32 = 0;
 pub const FIRST_BOOTSTRAP_OBJECT_ID: u32 = 12000;
 pub const FIRST_NORMAL_OBJECT_ID: u32 = 16384;
 
-/* pg_control.h */
 pub const XLOG_CHECKPOINT_SHUTDOWN: u8 = 0x00;
 pub const XLOG_CHECKPOINT_ONLINE: u8 = 0x10;
-pub const XLOG_NOOP: u8 = 0x20;
-pub const XLOG_NEXTOID: u8 = 0x30;
-pub const XLOG_SWITCH: u8 = 0x40;
-pub const XLOG_BACKUP_END: u8 = 0x50;
-pub const XLOG_PARAMETER_CHANGE: u8 = 0x60;
-pub const XLOG_RESTORE_POINT: u8 = 0x70;
-pub const XLOG_FPW_CHANGE: u8 = 0x80;
-pub const XLOG_END_OF_RECOVERY: u8 = 0x90;
-pub const XLOG_FPI_FOR_HINT: u8 = 0xA0;
-pub const XLOG_FPI: u8 = 0xB0;
-/* 0xC0 is used in Postgres 9.5-11 */
-pub const XLOG_OVERWRITE_CONTRECORD: u8 = 0xD0;
-
 pub const XLP_FIRST_IS_CONTRECORD: u16 = 0x0001;
 pub const XLP_LONG_HEADER: u16 = 0x0002;
 

pageserver/src/walingest.rs

@@ -55,36 +55,6 @@ pub struct WalIngest {
     checkpoint_modified: bool,
 }
 
-macro_rules! special_treatment_check {
-    (needs none) => {{
-        // we acknowledge that this record type needs no special treatment
-    }};
-    (unknown record type, $pg_version:expr, $lsn:expr, $decoded:expr) => {{
-        let pg_version: u32 = $pg_version;
-        let lsn: Lsn = $lsn;
-        let decoded: &DecodedWALRecord = $decoded;
-        use std::sync::atomic;
-        static LOGGED: atomic::AtomicBool = atomic::AtomicBool::new(false);
-        if LOGGED
-            .compare_exchange(
-                false,
-                true,
-                atomic::Ordering::Relaxed,
-                atomic::Ordering::Relaxed,
-            )
-            .is_ok()
-        {
-            warn!(
-                pg_version,
-                %lsn,
-                xl_rmid = %decoded.xl_rmid,
-                xl_info = %decoded.xl_info,
-                "unknown WAL record type, investigate whether it needs special treatment"
-            );
-        }
-    }};
-}
-
 impl WalIngest {
     pub async fn new(
         timeline: &Timeline,
@@ -141,7 +111,6 @@ impl WalIngest {
 
         failpoint_support::sleep_millis_async!("wal-ingest-record-sleep");
 
-        #[allow(clippy::if_same_then_else)]
         match decoded.xl_rmid {
             pg_constants::RM_HEAP_ID | pg_constants::RM_HEAP2_ID => {
                 // Heap AM records need some special handling, because they modify VM pages
@@ -165,8 +134,6 @@ impl WalIngest {
                     let truncate = XlSmgrTruncate::decode(&mut buf);
                     self.ingest_xlog_smgr_truncate(modification, &truncate, ctx)
                         .await?;
-                } else {
-                    special_treatment_check!(needs none);
                 }
             }
             pg_constants::RM_DBASE_ID => {
@@ -188,8 +155,6 @@ impl WalIngest {
                                 .drop_dbdir(tablespace_id, dropdb.db_id, ctx)
                                 .await?;
                         }
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else if pg_version == 15 {
                     if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_WAL_LOG {
@@ -210,8 +175,6 @@ impl WalIngest {
                                 .drop_dbdir(tablespace_id, dropdb.db_id, ctx)
                                 .await?;
                         }
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else if pg_version == 16 {
                     if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_WAL_LOG {
@@ -232,16 +195,11 @@ impl WalIngest {
                                 .drop_dbdir(tablespace_id, dropdb.db_id, ctx)
                                 .await?;
                         }
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_TBLSPC_ID => {
                 trace!("XLOG_TBLSPC_CREATE/DROP is not handled yet");
-                todo!() // should we do: special_treatment_check!(unknown record type, pg_version, lsn, decoded);
             }
             pg_constants::RM_CLOG_ID => {
                 let info = decoded.xl_info & !pg_constants::XLR_INFO_MASK;
@@ -259,12 +217,11 @@ impl WalIngest {
                         ctx,
                     )
                     .await?;
-                } else if info == pg_constants::CLOG_TRUNCATE {
+                } else {
+                    assert!(info == pg_constants::CLOG_TRUNCATE);
                     let xlrec = XlClogTruncate::decode(&mut buf);
                     self.ingest_clog_truncate_record(modification, &xlrec, ctx)
                         .await?;
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_XACT_ID => {
@@ -308,12 +265,6 @@ impl WalIngest {
                     modification
                         .put_twophase_file(decoded.xl_xid, Bytes::copy_from_slice(&buf[..]), ctx)
                         .await?;
-                } else if info == pg_constants::XLOG_XACT_ASSIGNMENT {
-                    special_treatment_check!(needs none);
-                } else if info == pg_constants::XLOG_XACT_INVALIDATIONS {
-                    special_treatment_check!(needs none);
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_MULTIXACT_ID => {
@@ -352,20 +303,12 @@ impl WalIngest {
                     let xlrec = XlMultiXactTruncate::decode(&mut buf);
                     self.ingest_multixact_truncate_record(modification, &xlrec, ctx)
                         .await?;
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_RELMAP_ID => {
-                let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
-
-                if info == pg_constants::XLOG_RELMAP_UPDATE {
-                    let xlrec = XlRelmapUpdate::decode(&mut buf);
-                    self.ingest_relmap_page(modification, &xlrec, decoded, ctx)
-                        .await?;
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
-                }
+                let xlrec = XlRelmapUpdate::decode(&mut buf);
+                self.ingest_relmap_page(modification, &xlrec, decoded, ctx)
+                    .await?;
             }
             pg_constants::RM_XLOG_ID => {
                 let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
@@ -375,8 +318,6 @@ impl WalIngest {
                     if self.checkpoint.nextOid != next_oid {
                         self.checkpoint.nextOid = next_oid;
                         self.checkpoint_modified = true;
-                    } else {
-                        special_treatment_check!(needs none);
                     }
                 } else if info == pg_constants::XLOG_CHECKPOINT_ONLINE
                     || info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN
@@ -409,31 +350,6 @@ impl WalIngest {
                     // have some trace of the checkpoint records in the layer files at the same
                     // LSNs.
                     self.checkpoint_modified = true;
-                } else if info == pg_constants::XLOG_FPI || info == pg_constants::XLOG_FPI_FOR_HINT
-                {
-                    // These records are importan for us, bu they are handled by
-                    // generic ingest_decoded_block() function below. They don't need
-                    // any special handling.
-                    //
-                    // HEIKKI: Is Noop the right code for that case?
-                    special_treatment_check!(needs none);
-                } else if info == pg_constants::XLOG_NOOP
-                    || info == pg_constants::XLOG_NEXTOID
-                    || info == pg_constants::XLOG_SWITCH
-                    || info == pg_constants::XLOG_BACKUP_END
-                    || info == pg_constants::XLOG_PARAMETER_CHANGE
-                    || info == pg_constants::XLOG_RESTORE_POINT
-                    || info == pg_constants::XLOG_FPW_CHANGE
-                    || info == pg_constants::XLOG_END_OF_RECOVERY
-                {
-                    special_treatment_check!(needs none);
-                } else if info == pg_constants::XLOG_OVERWRITE_CONTRECORD {
-                    // HEIKKI: I suspect we're not handling these correctly.
-                    // See https://github.com/neondatabase/neon/issues/934
-                    // Given that, not sure what the right outcome is.
-                    todo!()
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_LOGICALMSG_ID => {
@@ -449,14 +365,9 @@ impl WalIngest {
                         // we could peek into the message and only pause if it contains
                         // a particular string, for example, but this is enough for now.
                         failpoint_support::sleep_millis_async!("wal-ingest-logical-message-sleep");
-                        special_treatment_check!(needs none);
                     } else if let Some(path) = prefix.strip_prefix("neon-file:") {
                         modification.put_file(path, message, ctx).await?;
-                    } else {
-                        special_treatment_check!(needs none);
                     }
-                } else {
-                    special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                 }
             }
             pg_constants::RM_STANDBY_ID => {
@@ -464,9 +375,6 @@ impl WalIngest {
                 if info == pg_constants::XLOG_RUNNING_XACTS {
                     let xlrec = crate::walrecord::XlRunningXacts::decode(&mut buf);
                     self.checkpoint.oldestActiveXid = xlrec.oldest_running_xid;
-                    todo!() // checkpoint_modified=true missing?
-                } else {
-                    todo!()
                 }
             }
             pg_constants::RM_REPLORIGIN_ID => {
@@ -479,27 +387,13 @@ impl WalIngest {
                 } else if info == pg_constants::XLOG_REPLORIGIN_DROP {
                     let xlrec = crate::walrecord::XlReploriginDrop::decode(&mut buf);
                     modification.drop_replorigin(xlrec.node_id).await?
-                } else {
-                    todo!()
                 }
             }
-
-            // All of these are handled by the generic ingest_decoded_block function
-            pg_constants::RM_BTREE_ID => special_treatment_check!(needs none),
-            pg_constants::RM_HASH_ID => special_treatment_check!(needs none),
-            pg_constants::RM_GIN_ID => special_treatment_check!(needs none),
-            pg_constants::RM_GIST_ID => special_treatment_check!(needs none),
-            pg_constants::RM_SEQ_ID => special_treatment_check!(needs none),
-            pg_constants::RM_SPGIST_ID => special_treatment_check!(needs none),
-            pg_constants::RM_BRIN_ID => special_treatment_check!(needs none),
-            pg_constants::RM_GENERIC_ID => special_treatment_check!(needs none),
-
-            // We don't support the commit-ts tracking in neon. No harm if we see
-            // these records though.
-            pg_constants::RM_COMMIT_TS_ID => special_treatment_check!(needs none),
-
-            _x => special_treatment_check!(unknown record type, pg_version, lsn, decoded),
-        };
+            _x => {
+                // TODO: should probably log & fail here instead of blindly
+                // doing something without understanding the protocol
+            }
+        }
 
         // Iterate through all the blocks that the record modifies, and
         // "put" a separate copy of the record for each block.
@@ -647,10 +541,7 @@ impl WalIngest {
         let mut old_heap_blkno: Option<u32> = None;
         let mut flags = pg_constants::VISIBILITYMAP_VALID_BITS;
 
-        let pg_version = modification.tline.pg_version;
-        let lsn = modification.get_lsn();
-        #[allow(clippy::if_same_then_else)]
-        match pg_version {
+        match modification.tline.pg_version {
             14 => {
                 if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
@@ -689,19 +580,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP_TRUNCATE {
-                        // per comment in heap_redo:
-                        // TRUNCATE is a no-op because the actions are already logged as
-                        // SMGR WAL records.  TRUNCATE WAL record only exists for logical
-                        // decoding.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP_CONFIRM
-                        || info == pg_constants::XLOG_HEAP_INPLACE
-                    {
-                        // these don't update the FSM or VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
@@ -726,23 +604,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP2_REWRITE
-                        || info == pg_constants::XLOG_HEAP2_NEW_CID
-                    {
-                        // related to logical replication, we can ignore in storage
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_PRUNE
-                        || info == pg_constants::XLOG_HEAP2_VACUUM
-                        || info == pg_constants::XLOG_HEAP2_FREEZE_PAGE
-                    {
-                        // these don't update the VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_VISIBLE {
-                        // This updates the VM, but the VM page is registered as a normal
-                        // block in the WAL record, so no special handling is needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else {
                     bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
@@ -786,19 +647,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP_TRUNCATE {
-                        // per comment in heap_redo:
-                        // TRUNCATE is a no-op because the actions are already logged as
-                        // SMGR WAL records.  TRUNCATE WAL record only exists for logical
-                        // decoding.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP_CONFIRM
-                        || info == pg_constants::XLOG_HEAP_INPLACE
-                    {
-                        // these don't update the FSM or VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
@@ -823,23 +671,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP2_REWRITE
-                        || info == pg_constants::XLOG_HEAP2_NEW_CID
-                    {
-                        // related to logical replication, we can ignore in storage
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_PRUNE
-                        || info == pg_constants::XLOG_HEAP2_VACUUM
-                        || info == pg_constants::XLOG_HEAP2_FREEZE_PAGE
-                    {
-                        // these don't update the VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_VISIBLE {
-                        // This updates the VM, but the VM page is registered as a normal
-                        // block in the WAL record, so no special handling is needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else {
                     bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
@@ -883,19 +714,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP_TRUNCATE {
-                        // per comment in heap_redo:
-                        // TRUNCATE is a no-op because the actions are already logged as
-                        // SMGR WAL records.  TRUNCATE WAL record only exists for logical
-                        // decoding.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP_CONFIRM
-                        || info == pg_constants::XLOG_HEAP_INPLACE
-                    {
-                        // these don't update the FSM or VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else if decoded.xl_rmid == pg_constants::RM_HEAP2_ID {
                     let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;
@@ -920,23 +738,6 @@ impl WalIngest {
                             old_heap_blkno = Some(decoded.blocks[0].blkno);
                             flags = pg_constants::VISIBILITYMAP_ALL_FROZEN;
                         }
-                    } else if info == pg_constants::XLOG_HEAP2_REWRITE
-                        || info == pg_constants::XLOG_HEAP2_NEW_CID
-                    {
-                        // related to logical replication, we can ignore in storage
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_PRUNE
-                        || info == pg_constants::XLOG_HEAP2_VACUUM
-                        || info == pg_constants::XLOG_HEAP2_FREEZE_PAGE
-                    {
-                        // these don't update the VM, so no special handling needed.
-                        special_treatment_check!(needs none);
-                    } else if info == pg_constants::XLOG_HEAP2_VISIBLE {
-                        // This updates the VM, but the VM page is registered as a normal
-                        // block in the WAL record, so no special handling is needed.
-                        special_treatment_check!(needs none);
-                    } else {
-                        special_treatment_check!(unknown record type, pg_version, lsn, decoded);
                     }
                 } else {
                     bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);