Restart already running rsyslog to update configuration

set to 'Hipaa':

- setup and configure
pgaudit and pgauditlogtofile extensions
in compute_ctl.

- spin up a rsyslog server in the compute VM,
and configure it to send logs to the endpoint
specified in AUDIT_LOGGING_ENDPOINT env.

Change pgaudit.log default to log 'all'.
exclude postgres database from audit logging:
we consider it system database that doesn't contain any sensitive data.

- add pgaudit, pgauditlogtofile to shared_preload_libraries
if audit_log_level Hipaa is enabled

Move rsyslog config to compute_rsyslog_template.conf

Set pgaudit.log_rotation_age

.github/workflows/build_and_test.yml

@@ -1061,7 +1061,7 @@ jobs:
           exit 1
 
   deploy:
-    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
+    needs: [ check-permissions, push-neon-image-prod, push-compute-image-prod, meta, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
     # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `push-neon-image-prod` and `push-compute-image-prod`
     if: ${{ contains(fromJSON('["push-main", "storage-release", "proxy-release", "compute-release"]'), needs.meta.outputs.run-kind) && !failure() && !cancelled() }}
     permissions:

compute/compute-node.Dockerfile

@@ -1933,6 +1933,7 @@ RUN apt update && \
         locales \
         procps \
         ca-certificates \
+        rsyslog \
         $VERSION_INSTALLS && \
     apt clean && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* && \
     localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8
@@ -1978,6 +1979,15 @@ COPY --from=sql_exporter_preprocessor --chmod=0644 /home/nonroot/compute/etc/neo
 # Make the libraries we built available
 RUN echo '/usr/local/lib' >> /etc/ld.so.conf && /sbin/ldconfig
 
+# rsyslog config permissions
+RUN chown postgres:postgres /etc/rsyslog.conf && \
+    touch /etc/compute_rsyslog.conf && \
+    chown -R postgres:postgres /etc/compute_rsyslog.conf && \
+    # directory for rsyslogd pid file
+    mkdir /var/run/rsyslogd && \
+    chown -R postgres:postgres /var/run/rsyslogd
+
+
 ENV LANG=en_US.utf8
 USER postgres
 ENTRYPOINT ["/usr/local/bin/compute_ctl"]

compute/vm-image-spec-bookworm.yaml

@@ -39,6 +39,10 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
+  - name: rsyslogd
+    user: postgres
+    sysvInitAction: respawn
+    shell: '/usr/sbin/rsyslogd -i /var/run/rsyslogd/rsyslogd.pid -f /etc/compute_rsyslog.conf'
 shutdownHook: |
   su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
 files:
@@ -54,7 +58,7 @@ files:
       # regardless of hostname (ALL)
       #
       # Also allow it to shut down the VM. The fast_import job does that when it's finished.
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff
+      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff, /usr/sbin/rsyslogd
   - filename: cgconfig.conf
     content: |
       # Configuration for cgroups in VM compute nodes
@@ -69,6 +73,11 @@ files:
           }
           memory {}
       }
+# Create dummy rsyslog config, because it refuses to start without at least one action configured.
+# compute_ctl will rewrite this file with the actual configuration, if needed.
+  - filename: compute_rsyslog.conf
+    content: |
+      *.*    /dev/null
 build: |
   # Build cgroup-tools
   #
@@ -132,6 +141,10 @@ merge: |
   RUN set -e \
       && chmod 0644 /etc/cgconfig.conf
 
+  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
+  RUN set -e \
+      && chmod 0644 /etc/compute_rsyslog.conf
+
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
   COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
   COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/

compute/vm-image-spec-bullseye.yaml

@@ -39,6 +39,10 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
+  - name: rsyslogd
+    user: postgres
+    sysvInitAction: respawn
+    shell: '/usr/sbin/rsyslogd -i /var/run/rsyslogd/rsyslogd.pid -f /etc/compute_rsyslog.conf'
 shutdownHook: |
   su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
 files:
@@ -54,7 +58,7 @@ files:
       # regardless of hostname (ALL)
       #
       # Also allow it to shut down the VM. The fast_import job does that when it's finished.
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff
+      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff, /usr/sbin/rsyslogd
   - filename: cgconfig.conf
     content: |
       # Configuration for cgroups in VM compute nodes
@@ -69,6 +73,11 @@ files:
           }
           memory {}
       }
+# Create dummy rsyslog config, because it refuses to start without at least one action configured.
+# compute_ctl will rewrite this file with the actual configuration, if needed.
+  - filename: compute_rsyslog.conf
+    content: |
+      *.*    /dev/null
 build: |
   # Build cgroup-tools
   #
@@ -128,6 +137,10 @@ merge: |
   RUN set -e \
       && chmod 0644 /etc/cgconfig.conf
 
+  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
+  RUN set -e \
+      && chmod 0644 /etc/compute_rsyslog.conf
+
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
   COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
   COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/

compute_tools/src/compute.rs

@@ -12,7 +12,9 @@ use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use compute_api::privilege::Privilege;
 use compute_api::responses::{ComputeCtlConfig, ComputeMetrics, ComputeStatus};
-use compute_api::spec::{ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent};
+use compute_api::spec::{
+    ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent,
+};
 use futures::StreamExt;
 use futures::future::join_all;
 use futures::stream::FuturesUnordered;
@@ -35,6 +37,7 @@ use crate::logger::startup_context_from_env;
 use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use crate::monitor::launch_monitor;
 use crate::pg_helpers::*;
+use crate::rsyslog::configure_and_start_rsyslog;
 use crate::spec::*;
 use crate::swap::resize_swap;
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
@@ -676,6 +679,23 @@ impl ComputeNode {
             });
         }
 
+        // Configure and start rsyslog if necessary
+        if let ComputeAudit::Hipaa = pspec.spec.audit_log_level {
+            let remote_endpoint = std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
+            if remote_endpoint.is_empty() {
+                anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
+            }
+
+            let log_directory_path = Path::new(&self.params.pgdata).join("log");
+            // TODO: make this more robust
+            // now rsyslog starts once and there is no monitoring or restart if it fails
+            configure_and_start_rsyslog(
+                log_directory_path.to_str().unwrap(),
+                "hipaa",
+                &remote_endpoint,
+            )?;
+        }
+
         // Launch remaining service threads
         let _monitor_handle = launch_monitor(self);
         let _configurator_handle = launch_configurator(self);

compute_tools/src/config.rs

@@ -1,3 +1,4 @@
+use anyhow::Result;
 use std::fmt::Write as FmtWrite;
 use std::fs::{File, OpenOptions};
 use std::io;
@@ -5,10 +6,11 @@ use std::io::Write;
 use std::io::prelude::*;
 use std::path::Path;
 
-use anyhow::Result;
-use compute_api::spec::{ComputeMode, ComputeSpec, GenericOption};
+use compute_api::spec::{ComputeAudit, ComputeMode, ComputeSpec, GenericOption};
 
-use crate::pg_helpers::{GenericOptionExt, PgOptionsSerialize, escape_conf_value};
+use crate::pg_helpers::{
+    GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
+};
 
 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.
@@ -138,6 +140,54 @@ pub fn write_postgres_conf(
         writeln!(file, "# Managed by compute_ctl: end")?;
     }
 
+    // If audit logging is enabled, configure pgaudit.
+    //
+    // Note, that this is called after the settings from spec are written.
+    // This way we always override the settings from the spec
+    // and don't allow the user or the control plane admin to change them.
+    if let ComputeAudit::Hipaa = spec.audit_log_level {
+        writeln!(file, "# Managed by compute_ctl audit settings: begin")?;
+        // This log level is very verbose
+        // but this is necessary for HIPAA compliance.
+        writeln!(file, "pgaudit.log='all'")?;
+        writeln!(file, "pgaudit.log_parameter=on")?;
+        // Disable logging of catalog queries
+        // The catalog doesn't contain sensitive data, so we don't need to audit it.
+        writeln!(file, "pgaudit.log_catalog=off")?;
+        // Set log rotation to 5 minutes
+        // TODO: tune this after performance testing
+        writeln!(file, "pgaudit.log_rotation_age=5")?;
+
+        // Add audit shared_preload_libraries, if they are not present.
+        //
+        // The caller who sets the flag is responsible for ensuring that the necessary
+        // shared_preload_libraries are present in the compute image,
+        // otherwise the compute start will fail.
+        if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
+            let mut extra_shared_preload_libraries = String::new();
+            if !libs.contains("pgaudit") {
+                extra_shared_preload_libraries.push_str(",pgaudit");
+            }
+            if !libs.contains("pgauditlogtofile") {
+                extra_shared_preload_libraries.push_str(",pgauditlogtofile");
+            }
+            writeln!(
+                file,
+                "shared_preload_libraries='{}{}'",
+                libs, extra_shared_preload_libraries
+            )?;
+        } else {
+            // Typically, this should be unreacheable,
+            // because we always set at least some shared_preload_libraries in the spec
+            // but let's handle it explicitly anyway.
+            writeln!(
+                file,
+                "shared_preload_libraries='neon,pgaudit,pgauditlogtofile'"
+            )?;
+        }
+        writeln!(file, "# Managed by compute_ctl audit settings: end")?;
+    }
+
     writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
 
     if spec.drop_subscriptions_before_start {

compute_tools/src/config_template/compute_rsyslog_template.conf

@@ -0,0 +1,10 @@
+# Load imfile module to read log files
+module(load="imfile")
+
+# Input configuration for log files in the specified directory
+# Replace {log_directory} with the directory containing the log files
+input(type="imfile" File="{log_directory}/*.log" Tag="{tag}" Severity="info" Facility="local0")
+global(workDirectory="/var/log")
+
+# Forward logs to remote syslog server
+*.* @@{remote_endpoint}

compute_tools/src/lib.rs

@@ -21,6 +21,7 @@ mod migration;
 pub mod monitor;
 pub mod params;
 pub mod pg_helpers;
+pub mod rsyslog;
 pub mod spec;
 mod spec_apply;
 pub mod swap;

compute_tools/src/rsyslog.rs

@@ -0,0 +1,91 @@
+use std::process::Command;
+use std::{fs::OpenOptions, io::Write};
+
+use anyhow::{Context, Result};
+use tracing::info;
+
+fn get_rsyslog_pid() -> Option<String> {
+    let output = Command::new("pgrep")
+        .arg("rsyslogd")
+        .output()
+        .expect("Failed to execute pgrep");
+
+    if !output.stdout.is_empty() {
+        let pid = std::str::from_utf8(&output.stdout)
+            .expect("Invalid UTF-8 in process output")
+            .trim()
+            .to_string();
+        Some(pid)
+    } else {
+        None
+    }
+}
+
+// Start rsyslogd with the specified configuration file
+// If it is already running - restart it.
+// This is necessary, because there is no other way to reload the rsyslog configuration.
+//
+// Rsyslogd shouldn't lose any messages, because of the restart,
+// because it tracks the last read position in the log files
+// and will continue reading from that position.
+// TODO: test it properly
+//
+fn start_rsyslog(rsyslog_conf_path: &str) -> Result<()> {
+    let old_pid = get_rsyslog_pid();
+    if let Some(pid) = old_pid {
+        info!("rsyslogd is already running with pid: {}, restart it", pid);
+        // kill it to restart
+        let _ = Command::new("pkill")
+            .arg("rsyslogd")
+            .output()
+            .context("Failed to stop rsyslogd")?;
+    }
+
+    let _ = Command::new("/usr/sbin/rsyslogd")
+        .arg("-f")
+        .arg(rsyslog_conf_path)
+        .arg("-i")
+        .arg("/var/run/rsyslogd/rsyslogd.pid")
+        .output()
+        .context("Failed to start rsyslogd")?;
+
+    // Check that rsyslogd is running
+    if let Some(pid) = get_rsyslog_pid() {
+        info!("rsyslogd started successfully with pid: {}", pid);
+    } else {
+        return Err(anyhow::anyhow!("Failed to start rsyslogd"));
+    }
+
+    Ok(())
+}
+
+pub fn configure_and_start_rsyslog(
+    log_directory: &str,
+    tag: &str,
+    remote_endpoint: &str,
+) -> Result<()> {
+    let config_content: String = format!(
+        include_str!("config_template/compute_rsyslog_template.conf"),
+        log_directory = log_directory,
+        tag = tag,
+        remote_endpoint = remote_endpoint
+    );
+
+    info!("rsyslog config_content: {}", config_content);
+
+    let rsyslog_conf_path = "/etc/compute_rsyslog.conf";
+    let mut file = OpenOptions::new()
+        .create(true)
+        .write(true)
+        .truncate(true)
+        .open(rsyslog_conf_path)?;
+
+    file.write_all(config_content.as_bytes())?;
+
+    info!("rsyslog configuration added successfully. Starting rsyslogd");
+
+    // start the service, using the configuration
+    start_rsyslog(rsyslog_conf_path)?;
+
+    Ok(())
+}

compute_tools/src/spec_apply.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use compute_api::responses::ComputeStatus;
-use compute_api::spec::{ComputeFeature, ComputeSpec, Database, PgIdent, Role};
+use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeSpec, Database, PgIdent, Role};
 use futures::future::join_all;
 use tokio::sync::RwLock;
 use tokio_postgres::Client;
@@ -19,10 +19,10 @@ use crate::pg_helpers::{
     get_existing_roles_async,
 };
 use crate::spec_apply::ApplySpecPhase::{
-    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateSchemaNeon,
-    CreateSuperUser, DropInvalidDatabases, DropRoles, FinalizeDropLogicalSubscriptions,
-    HandleNeonExtension, HandleOtherExtensions, RenameAndDeleteDatabases, RenameRoles,
-    RunInEachDatabase,
+    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreatePgauditExtension,
+    CreatePgauditlogtofileExtension, CreateSchemaNeon, CreateSuperUser, DisablePostgresDBPgAudit,
+    DropInvalidDatabases, DropRoles, FinalizeDropLogicalSubscriptions, HandleNeonExtension,
+    HandleOtherExtensions, RenameAndDeleteDatabases, RenameRoles, RunInEachDatabase,
 };
 use crate::spec_apply::PerDatabasePhase::{
     ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions, HandleAnonExtension,
@@ -277,6 +277,19 @@ impl ComputeNode {
                 phases.push(FinalizeDropLogicalSubscriptions);
             }
 
+            // Keep DisablePostgresDBPgAudit phase at the end,
+            // so that all config operations are audit logged.
+            match spec.audit_log_level
+            {
+                ComputeAudit::Hipaa => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(CreatePgauditlogtofileExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
+                ComputeAudit::Log => { /* not implemented yet */ }
+                ComputeAudit::Disabled => {}
+            }
+
             for phase in phases {
                 debug!("Applying phase {:?}", &phase);
                 apply_operations(
@@ -463,6 +476,9 @@ pub enum ApplySpecPhase {
     CreateAndAlterDatabases,
     CreateSchemaNeon,
     RunInEachDatabase { db: DB, subphase: PerDatabasePhase },
+    CreatePgauditExtension,
+    CreatePgauditlogtofileExtension,
+    DisablePostgresDBPgAudit,
     HandleOtherExtensions,
     HandleNeonExtension,
     CreateAvailabilityCheck,
@@ -1098,6 +1114,25 @@ async fn get_operations<'a>(
             }
             Ok(Box::new(empty()))
         }
+        ApplySpecPhase::CreatePgauditExtension => Ok(Box::new(once(Operation {
+            query: String::from("CREATE EXTENSION IF NOT EXISTS pgaudit"),
+            comment: Some(String::from("create pgaudit extensions")),
+        }))),
+        ApplySpecPhase::CreatePgauditlogtofileExtension => Ok(Box::new(once(Operation {
+            query: String::from("CREATE EXTENSION IF NOT EXISTS pgauditlogtofile"),
+            comment: Some(String::from("create pgauditlogtofile extensions")),
+        }))),
+        // Disable pgaudit logging for postgres database.
+        // Postgres is neon system database used by monitors
+        // and compute_ctl tuning functions and thus generates a lot of noise.
+        // We do not consider data stored in this database as sensitive.
+        ApplySpecPhase::DisablePostgresDBPgAudit => {
+            let query = "ALTER DATABASE postgres SET pgaudit.log to 'none'";
+            Ok(Box::new(once(Operation {
+                query: query.to_string(),
+                comment: Some(query.to_string()),
+            })))
+        }
         ApplySpecPhase::HandleNeonExtension => {
             let operations = vec![
                 Operation {

control_plane/src/endpoint.rs

@@ -48,7 +48,8 @@ use anyhow::{Context, Result, anyhow, bail};
 use compute_api::requests::ConfigurationRequest;
 use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
 use compute_api::spec::{
-    Cluster, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent, RemoteExtSpec, Role,
+    Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
+    RemoteExtSpec, Role,
 };
 use nix::sys::signal::{Signal, kill};
 use pageserver_api::shard::ShardStripeSize;
@@ -668,6 +669,7 @@ impl Endpoint {
             local_proxy_config: None,
             reconfigure_concurrency: self.reconfigure_concurrency,
             drop_subscriptions_before_start: self.drop_subscriptions_before_start,
+            audit_log_level: ComputeAudit::Disabled,
         };
 
         // this strange code is needed to support respec() in tests

control_plane/src/local_env.rs

@@ -165,6 +165,8 @@ pub struct NeonStorageControllerConf {
 
     #[serde(with = "humantime_serde")]
     pub long_reconcile_threshold: Option<Duration>,
+
+    pub load_safekeepers: bool,
 }
 
 impl NeonStorageControllerConf {
@@ -188,6 +190,7 @@ impl Default for NeonStorageControllerConf {
             max_secondary_lag_bytes: None,
             heartbeat_interval: Self::DEFAULT_HEARTBEAT_INTERVAL,
             long_reconcile_threshold: None,
+            load_safekeepers: true,
         }
     }
 }

control_plane/src/storage_controller.rs

@@ -537,6 +537,10 @@ impl StorageController {
             args.push("--start-as-candidate".to_string());
         }
 
+        if self.config.load_safekeepers {
+            args.push("--load-safekeepers".to_string());
+        }
+
         if let Some(private_key) = &self.private_key {
             let claims = Claims::new(None, Scope::PageServerApi);
             let jwt_token =

docs/explain-extension.md

@@ -1,72 +0,0 @@
-# Neon changes to Postgres EXPLAIN command
-
-## Why do we need to include more information in EXPLAIN?
-
-Neon contains two components: prefetch and LFC (local file cache) which may have critical impact on query performance.
-Both are trying to solve the problem with relatively large round-trip between compute and page server (much larger than average access time for modern SSDs).
-This is why Neon can provide comparable performance only if all data set is present at local node.
-Certainly the fastest case of accessing data in Postgres is when it is present in Postgres cache (shared buffers).
-Unfortunately size of shared buffer can not be changed on the flight: it requires Postgres restart. It is not acceptable for autoscaling.
-This is why we have relatively small shared buffers and dynamically resized local file cache (LFC).
-It is intended that LFC fits in memory, although it can improve performance even if data is read from local disk.
-See https://neondb.slack.com/archives/C03QLRH7PPD/p1718714926044699
-To minimize in-memory footprint of LFC and to improve sequential scan, LFC uses chunks which size is larger than size of Postgres page (right now chunk size is 1Mb).
-
-
-LFC as any other cache is useless after cool restart. Also some data sets can not fit in local disk.
-This is where another approach can help: prefetching. If we are able to predict which pages will be needed soon,
-compute can send prefetch requests to page server before this page is actually requested by executor.
-Prefetch is also used by Postgres (using `fadvise`), but only for vacuum and bitmap scan. Neon provides prefetch for more execution plan nodes:
-sequential scan, index scan (prefetch of referenced heap pages), index-only scan (prefetch B-Tree leaves).
-
-As far as work of prefetch and LFC may have critical impact on query performance, we need to provide this information to the users.
-The most convenient and natural way is to include it in EXPLAIN. Two new keyword are added by Neon to EXPLAIN options: `prefetch` and `filecache`.
-
-## prefetch
-
-The following information is available about prefetch:
-* `hits` - number of pages which are received from page server before actually requested by executor. Prefetch distance is controlled by `effective_io_concurrency` GUC. The larger it is, the more chances that page server will be able to complete request before it is needed. But it should not be larger than `neon.prefetch_buffer_size`.
-* `misses` - number of accessed pages which were not prefetched. Prefetch is not implemented for all plan nodes. And even for those nodes for which it is implemented (i.e. sequential scan) some mispredictions are possible. Please notice that `hits + misses != accessed pages`. If prefetch request for the page was issued but not yet completed before the page is requested, then such access is not considered as prefetch hit or miss.
-* `expired` - page can be updated by backend since the moment of sending prefetch request to page server. Or result of prefetch just not used because executor doesn't need this page (for example because of presence of `LIMIT` clause in the query).  In both cases such requests are considered as expired.
-* `duplicates` - multiple prefetch requests for the same page. For some nodes predicting next pages is trivial, i.e. for sequential scan. But in case of index scan we need to prefetch referenced heap pages. And definitely index entries can have multiple references to the same heap page. Such non-unique prefetch requests are considered as duplicates.
-
-
-## filecache
-
-The following information is available about file cache (LFC):
-* `hits` - number of accessed pages found in LFC.
-* `misses` - number of accessed pages not found in LFC.
-
-# LFC statistic
-
-While `filecache` option of EXPLAIN command provides information about LFC usage in the particular query, there is also available global statistic about LFC usage.
-It is provided by `neon` extension.
-
-## `neon_lfc_stats` view
-
-This view provides information as key-value pairs (so that new information can be added without changing neon extension interface).
-The following keys are provided:
-* `file_cache_hits` - total number of LFC hits (for all backends and queries since server startup).
-* `file_cache_misses` - total number of LFC misses.
-* `file_cache_used` - number chunks used in LFC
-* `file_cache_writes` - number of pages written to LFC
-* `file_cache_size` - current cache size in chunks (can not be larger than `neon.file_cache_size_limit`)
-* `file_cache_used_pages` - number is used pages. As far as not all pages of the chunk can be filled with data it can be smaller than `file_cache_used*128` (128 is number of 8kB pages in 1MB chunk)
-* `file_cache_evicted_pages` - number of pages evicted from LFC because working set doesn't fit in LFC.
-* `file_cache_limit` - current limit of LFC size (in chunks)
-
-## `local_cache` view
-This view is similar with `pg_buffercache` view and contains the following columns:
-```
-(pageoffs int8,
-relfilenode oid,
-reltablespace oid,
-reldatabase oid,
-relforknumber int2,
-relblocknumber int8,
-accesscount int4)
-```
-
-
-
-

libs/compute_api/src/spec.rs

@@ -155,6 +155,16 @@ pub struct ComputeSpec {
     /// over the same replication content from publisher.
     #[serde(default)] // Default false
     pub drop_subscriptions_before_start: bool,
+
+    /// Log level for audit logging:
+    ///
+    /// Disabled - no audit logging. This is the default.
+    /// log - log masked statements to the postgres log using pgaudit extension
+    /// hipaa - log unmasked statements to the file using pgaudit and pgauditlogtofile extension
+    ///
+    /// Extensions should be present in shared_preload_libraries
+    #[serde(default)]
+    pub audit_log_level: ComputeAudit,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -262,6 +272,17 @@ pub enum ComputeMode {
     Replica,
 }
 
+/// Log level for audit logging
+/// Disabled, log, hipaa
+/// Default is Disabled
+#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
+pub enum ComputeAudit {
+    #[default]
+    Disabled,
+    Log,
+    Hipaa,
+}
+
 #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
 pub struct Cluster {
     pub cluster_id: Option<String>,

libs/pageserver_api/src/config.rs

@@ -123,10 +123,6 @@ pub struct ConfigToml {
     pub enable_read_path_debugging: Option<bool>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub validate_wal_contiguity: Option<bool>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub load_previous_heatmap: Option<bool>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub generate_unarchival_heatmap: Option<bool>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -527,8 +523,6 @@ impl Default for ConfigToml {
                 None
             },
             validate_wal_contiguity: None,
-            load_previous_heatmap: None,
-            generate_unarchival_heatmap: None,
         }
     }
 }

pageserver/src/config.rs

@@ -194,13 +194,6 @@ pub struct PageServerConf {
     /// Interpreted protocol feature: if enabled, validate that the logical WAL received from
     /// safekeepers does not have gaps.
     pub validate_wal_contiguity: bool,
-
-    /// When set, the previously written to disk heatmap is loaded on tenant attach and used
-    /// to avoid clobbering the heatmap from new, cold, attached locations.
-    pub load_previous_heatmap: bool,
-
-    /// When set, include visible layers in the next uploaded heatmaps of an unarchived timeline.
-    pub generate_unarchival_heatmap: bool,
 }
 
 /// Token for authentication to safekeepers
@@ -365,8 +358,6 @@ impl PageServerConf {
             get_vectored_concurrent_io,
             enable_read_path_debugging,
             validate_wal_contiguity,
-            load_previous_heatmap,
-            generate_unarchival_heatmap,
         } = config_toml;
 
         let mut conf = PageServerConf {
@@ -456,8 +447,6 @@ impl PageServerConf {
             no_sync: no_sync.unwrap_or(false),
             enable_read_path_debugging: enable_read_path_debugging.unwrap_or(false),
             validate_wal_contiguity: validate_wal_contiguity.unwrap_or(false),
-            load_previous_heatmap: load_previous_heatmap.unwrap_or(false),
-            generate_unarchival_heatmap: generate_unarchival_heatmap.unwrap_or(false),
         };
 
         // ------------------------------------------------------------
@@ -504,8 +493,6 @@ impl PageServerConf {
             metric_collection_interval: Duration::from_secs(60),
             synthetic_size_calculation_interval: Duration::from_secs(60),
             background_task_maximum_delay: Duration::ZERO,
-            load_previous_heatmap: Some(true),
-            generate_unarchival_heatmap: Some(true),
             ..Default::default()
         };
         PageServerConf::parse_and_validate(NodeId(0), config_toml, &repo_dir).unwrap()

pageserver/src/tenant.rs

@@ -1150,7 +1150,7 @@ impl Tenant {
         // a previous heatmap which contains all visible layers in the layer map.
         // This previous heatmap will be used whenever a fresh heatmap is generated
         // for the timeline.
-        if self.conf.generate_unarchival_heatmap && matches!(cause, LoadTimelineCause::Unoffload) {
+        if matches!(cause, LoadTimelineCause::Unoffload) {
             let mut tline_ending_at = Some((&timeline, timeline.get_last_record_lsn()));
             while let Some((tline, end_lsn)) = tline_ending_at {
                 let unarchival_heatmap = tline.generate_unarchival_heatmap(end_lsn).await;
@@ -1582,10 +1582,6 @@ impl Tenant {
     }
 
     async fn read_on_disk_heatmap(&self) -> Option<(HeatMapTenant, std::time::Instant)> {
-        if !self.conf.load_previous_heatmap {
-            return None;
-        }
-
         let on_disk_heatmap_path = self.conf.tenant_heatmap_path(&self.tenant_shard_id);
         match tokio::fs::read_to_string(on_disk_heatmap_path).await {
             Ok(heatmap) => match serde_json::from_str::<HeatMapTenant>(&heatmap) {

pageserver/src/tenant/timeline/compaction.rs

@@ -15,7 +15,7 @@ use super::{
     Timeline,
 };
 
-use anyhow::{Context, anyhow};
+use anyhow::{Context, anyhow, bail};
 use bytes::Bytes;
 use enumset::EnumSet;
 use fail::fail_point;
@@ -234,12 +234,6 @@ impl GcCompactionQueue {
             // it enough in staging yet.
             return Ok(());
         }
-        if timeline.get_gc_compaction_watermark() == Lsn::INVALID {
-            // If the gc watermark is not set, we don't need to trigger auto compaction.
-            // This check is the same as in `gc_compaction_split_jobs` but we don't log
-            // here and we can also skip the computation of the trigger condition earlier.
-            return Ok(());
-        }
 
         let Ok(permit) = CONCURRENT_GC_COMPACTION_TASKS.clone().try_acquire_owned() else {
             // Only allow one compaction run at a time. TODO: As we do `try_acquire_owned`, we cannot ensure
@@ -363,7 +357,8 @@ impl GcCompactionQueue {
                 GcCompactJob::from_compact_options(options.clone()),
                 options.sub_compaction_max_job_size_mb,
             )
-            .await?;
+            .await
+            .map_err(CompactionError::Other)?;
         if jobs.is_empty() {
             info!("no jobs to run, skipping scheduled compaction task");
             self.notify_and_unblock(id);
@@ -830,7 +825,9 @@ impl Timeline {
             .flags
             .contains(CompactFlags::EnhancedGcBottomMostCompaction)
         {
-            self.compact_with_gc(cancel, options, ctx).await?;
+            self.compact_with_gc(cancel, options, ctx)
+                .await
+                .map_err(CompactionError::Other)?;
             return Ok(CompactionOutcome::Done);
         }
 
@@ -2348,19 +2345,12 @@ impl Timeline {
     async fn check_compaction_space(
         self: &Arc<Self>,
         layer_selection: &[Layer],
-    ) -> Result<(), CompactionError> {
-        let available_space = self
-            .check_available_space()
-            .await
-            .map_err(CompactionError::Other)?;
+    ) -> anyhow::Result<()> {
+        let available_space = self.check_available_space().await?;
         let mut remote_layer_size = 0;
         let mut all_layer_size = 0;
         for layer in layer_selection {
-            let needs_download = layer
-                .needs_download()
-                .await
-                .context("failed to check if layer needs download")
-                .map_err(CompactionError::Other)?;
+            let needs_download = layer.needs_download().await?;
             if needs_download.is_some() {
                 remote_layer_size += layer.layer_desc().file_size;
             }
@@ -2369,14 +2359,14 @@ impl Timeline {
         let allocated_space = (available_space as f64 * 0.8) as u64; /* reserve 20% space for other tasks */
         if all_layer_size /* space needed for newly-generated file */ + remote_layer_size /* space for downloading layers */ > allocated_space
         {
-            return Err(CompactionError::Other(anyhow!(
+            return Err(anyhow!(
                 "not enough space for compaction: available_space={}, allocated_space={}, all_layer_size={}, remote_layer_size={}, required_space={}",
                 available_space,
                 allocated_space,
                 all_layer_size,
                 remote_layer_size,
                 all_layer_size + remote_layer_size
-            )));
+            ));
         }
         Ok(())
     }
@@ -2407,7 +2397,7 @@ impl Timeline {
         self: &Arc<Self>,
         job: GcCompactJob,
         sub_compaction_max_job_size_mb: Option<u64>,
-    ) -> Result<Vec<GcCompactJob>, CompactionError> {
+    ) -> anyhow::Result<Vec<GcCompactJob>> {
         let compact_below_lsn = if job.compact_lsn_range.end != Lsn::MAX {
             job.compact_lsn_range.end
         } else {
@@ -2558,7 +2548,7 @@ impl Timeline {
         cancel: &CancellationToken,
         options: CompactOptions,
         ctx: &RequestContext,
-    ) -> Result<(), CompactionError> {
+    ) -> anyhow::Result<()> {
         let sub_compaction = options.sub_compaction;
         let job = GcCompactJob::from_compact_options(options.clone());
         if sub_compaction {
@@ -2590,7 +2580,7 @@ impl Timeline {
         cancel: &CancellationToken,
         job: GcCompactJob,
         ctx: &RequestContext,
-    ) -> Result<(), CompactionError> {
+    ) -> anyhow::Result<()> {
         // Block other compaction/GC tasks from running for now. GC-compaction could run along
         // with legacy compaction tasks in the future. Always ensure the lock order is compaction -> gc.
         // Note that we already acquired the compaction lock when the outer `compact` function gets called.
@@ -2598,7 +2588,8 @@ impl Timeline {
         let gc_lock = async {
             tokio::select! {
                 guard = self.gc_lock.lock() => Ok(guard),
-                _ = cancel.cancelled() => Err(CompactionError::ShuttingDown),
+                // TODO: refactor to CompactionError to correctly pass cancelled error
+                _ = cancel.cancelled() => Err(anyhow!("cancelled")),
             }
         };
 
@@ -2819,10 +2810,10 @@ impl Timeline {
             .map(|layer| layer.layer_desc().layer_name())
             .collect_vec();
         if let Some(err) = check_valid_layermap(&layer_names) {
-            return Err(CompactionError::Other(anyhow!(
+            bail!(
                 "gc-compaction layer map check failed because {}, cannot proceed with compaction due to potential data loss",
                 err
-            )));
+            );
         }
         // The maximum LSN we are processing in this compaction loop
         let end_lsn = job_desc
@@ -2837,24 +2828,11 @@ impl Timeline {
         let mut total_downloaded_size = 0;
         let mut total_layer_size = 0;
         for layer in &job_desc.selected_layers {
-            if layer
-                .needs_download()
-                .await
-                .context("failed to check if layer needs download")
-                .map_err(CompactionError::Other)?
-                .is_some()
-            {
+            if layer.needs_download().await?.is_some() {
                 total_downloaded_size += layer.layer_desc().file_size;
             }
             total_layer_size += layer.layer_desc().file_size;
-            if cancel.is_cancelled() {
-                return Err(CompactionError::ShuttingDown);
-            }
-            let resident_layer = layer
-                .download_and_keep_resident(ctx)
-                .await
-                .context("failed to download and keep resident layer")
-                .map_err(CompactionError::Other)?;
+            let resident_layer = layer.download_and_keep_resident(ctx).await?;
             downloaded_layers.push(resident_layer);
         }
         info!(
@@ -2865,33 +2843,19 @@ impl Timeline {
         );
         for resident_layer in &downloaded_layers {
             if resident_layer.layer_desc().is_delta() {
-                let layer = resident_layer
-                    .get_as_delta(ctx)
-                    .await
-                    .context("failed to get delta layer")
-                    .map_err(CompactionError::Other)?;
+                let layer = resident_layer.get_as_delta(ctx).await?;
                 delta_layers.push(layer);
             } else {
-                let layer = resident_layer
-                    .get_as_image(ctx)
-                    .await
-                    .context("failed to get image layer")
-                    .map_err(CompactionError::Other)?;
+                let layer = resident_layer.get_as_image(ctx).await?;
                 image_layers.push(layer);
             }
         }
-        let (dense_ks, sparse_ks) = self
-            .collect_gc_compaction_keyspace()
-            .await
-            .context("failed to collect gc compaction keyspace")
-            .map_err(CompactionError::Other)?;
+        let (dense_ks, sparse_ks) = self.collect_gc_compaction_keyspace().await?;
         let mut merge_iter = FilterIterator::create(
             MergeIterator::create(&delta_layers, &image_layers, ctx),
             dense_ks,
             sparse_ks,
-        )
-        .context("failed to create filter iterator")
-        .map_err(CompactionError::Other)?;
+        )?;
 
         // Step 2: Produce images+deltas.
         let mut accumulated_values = Vec::new();
@@ -2910,9 +2874,7 @@ impl Timeline {
                     self.get_compaction_target_size(),
                     ctx,
                 )
-                .await
-                .context("failed to create image layer writer")
-                .map_err(CompactionError::Other)?,
+                .await?,
             )
         } else {
             None
@@ -2925,9 +2887,7 @@ impl Timeline {
             lowest_retain_lsn..end_lsn,
             self.get_compaction_target_size(),
         )
-        .await
-        .context("failed to create delta layer writer")
-        .map_err(CompactionError::Other)?;
+        .await?;
 
         #[derive(Default)]
         struct RewritingLayers {
@@ -2967,14 +2927,9 @@ impl Timeline {
         // the key and LSN range are determined. However, to keep things simple here, we still
         // create this writer, and discard the writer in the end.
 
-        while let Some(((key, lsn, val), desc)) = merge_iter
-            .next_with_trace()
-            .await
-            .context("failed to get next key-value pair")
-            .map_err(CompactionError::Other)?
-        {
+        while let Some(((key, lsn, val), desc)) = merge_iter.next_with_trace().await? {
             if cancel.is_cancelled() {
-                return Err(CompactionError::ShuttingDown);
+                return Err(anyhow!("cancelled")); // TODO: refactor to CompactionError and pass cancel error
             }
             if self.shard_identity.is_key_disposable(&key) {
                 // If this shard does not need to store this key, simply skip it.
@@ -3005,9 +2960,7 @@ impl Timeline {
                                 desc.lsn_range.clone(),
                                 ctx,
                             )
-                            .await
-                            .context("failed to create delta layer writer")
-                            .map_err(CompactionError::Other)?,
+                            .await?,
                         );
                     }
                     rewriter.before.as_mut().unwrap()
@@ -3022,20 +2975,14 @@ impl Timeline {
                                 desc.lsn_range.clone(),
                                 ctx,
                             )
-                            .await
-                            .context("failed to create delta layer writer")
-                            .map_err(CompactionError::Other)?,
+                            .await?,
                         );
                     }
                     rewriter.after.as_mut().unwrap()
                 } else {
                     unreachable!()
                 };
-                rewriter
-                    .put_value(key, lsn, val, ctx)
-                    .await
-                    .context("failed to put value")
-                    .map_err(CompactionError::Other)?;
+                rewriter.put_value(key, lsn, val, ctx).await?;
                 continue;
             }
             match val {
@@ -3058,13 +3005,9 @@ impl Timeline {
                         &job_desc.retain_lsns_below_horizon,
                         COMPACTION_DELTA_THRESHOLD,
                         get_ancestor_image(self, *last_key, ctx, has_data_below, lowest_retain_lsn)
-                            .await
-                            .context("failed to get ancestor image")
-                            .map_err(CompactionError::Other)?,
+                            .await?,
                     )
-                    .await
-                    .context("failed to generate key retention")
-                    .map_err(CompactionError::Other)?;
+                    .await?;
                 retention
                     .pipe_to(
                         *last_key,
@@ -3073,9 +3016,7 @@ impl Timeline {
                         &mut stat,
                         ctx,
                     )
-                    .await
-                    .context("failed to pipe to delta layer writer")
-                    .map_err(CompactionError::Other)?;
+                    .await?;
                 accumulated_values.clear();
                 *last_key = key;
                 accumulated_values.push((key, lsn, val));
@@ -3093,14 +3034,9 @@ impl Timeline {
                 job_desc.gc_cutoff,
                 &job_desc.retain_lsns_below_horizon,
                 COMPACTION_DELTA_THRESHOLD,
-                get_ancestor_image(self, last_key, ctx, has_data_below, lowest_retain_lsn)
-                    .await
-                    .context("failed to get ancestor image")
-                    .map_err(CompactionError::Other)?,
+                get_ancestor_image(self, last_key, ctx, has_data_below, lowest_retain_lsn).await?,
             )
-            .await
-            .context("failed to generate key retention")
-            .map_err(CompactionError::Other)?;
+            .await?;
         retention
             .pipe_to(
                 last_key,
@@ -3109,9 +3045,7 @@ impl Timeline {
                 &mut stat,
                 ctx,
             )
-            .await
-            .context("failed to pipe to delta layer writer")
-            .map_err(CompactionError::Other)?;
+            .await?;
         // end: move the above part to the loop body
 
         let mut rewrote_delta_layers = Vec::new();
@@ -3119,23 +3053,13 @@ impl Timeline {
             if let Some(delta_writer_before) = writers.before {
                 let (desc, path) = delta_writer_before
                     .finish(job_desc.compaction_key_range.start, ctx)
-                    .await
-                    .context("failed to finish delta layer writer")
-                    .map_err(CompactionError::Other)?;
-                let layer = Layer::finish_creating(self.conf, self, desc, &path)
-                    .context("failed to finish creating delta layer")
-                    .map_err(CompactionError::Other)?;
+                    .await?;
+                let layer = Layer::finish_creating(self.conf, self, desc, &path)?;
                 rewrote_delta_layers.push(layer);
             }
             if let Some(delta_writer_after) = writers.after {
-                let (desc, path) = delta_writer_after
-                    .finish(key.key_range.end, ctx)
-                    .await
-                    .context("failed to finish delta layer writer")
-                    .map_err(CompactionError::Other)?;
-                let layer = Layer::finish_creating(self.conf, self, desc, &path)
-                    .context("failed to finish creating delta layer")
-                    .map_err(CompactionError::Other)?;
+                let (desc, path) = delta_writer_after.finish(key.key_range.end, ctx).await?;
+                let layer = Layer::finish_creating(self.conf, self, desc, &path)?;
                 rewrote_delta_layers.push(layer);
             }
         }
@@ -3150,9 +3074,7 @@ impl Timeline {
                 let end_key = job_desc.compaction_key_range.end;
                 writer
                     .finish_with_discard_fn(self, ctx, end_key, discard)
-                    .await
-                    .context("failed to finish image layer writer")
-                    .map_err(CompactionError::Other)?
+                    .await?
             } else {
                 drop(writer);
                 Vec::new()
@@ -3164,9 +3086,7 @@ impl Timeline {
         let produced_delta_layers = if !dry_run {
             delta_layer_writer
                 .finish_with_discard_fn(self, ctx, discard)
-                .await
-                .context("failed to finish delta layer writer")
-                .map_err(CompactionError::Other)?
+                .await?
         } else {
             drop(delta_layer_writer);
             Vec::new()
@@ -3246,9 +3166,7 @@ impl Timeline {
                     &layer.layer_desc().key_range,
                     &job_desc.compaction_key_range,
                 ) {
-                    return Err(CompactionError::Other(anyhow!(
-                        "violated constraint: image layer outside of compaction key range"
-                    )));
+                    bail!("violated constraint: image layer outside of compaction key range");
                 }
                 if !fully_contains(
                     &job_desc.compaction_key_range,
@@ -3263,9 +3181,7 @@ impl Timeline {
 
         info!(
             "gc-compaction statistics: {}",
-            serde_json::to_string(&stat)
-                .context("failed to serialize gc-compaction statistics")
-                .map_err(CompactionError::Other)?
+            serde_json::to_string(&stat)?
         );
 
         if dry_run {
@@ -3304,10 +3220,10 @@ impl Timeline {
         // the writer, so potentially, we will need a function like `ImageLayerBatchWriter::get_all_pending_layer_keys` to get all the keys that are
         // in the writer before finalizing the persistent layers. Now we would leave some dangling layers on the disk if the check fails.
         if let Some(err) = check_valid_layermap(&final_layers) {
-            return Err(CompactionError::Other(anyhow!(
+            bail!(
                 "gc-compaction layer map check failed after compaction because {}, compaction result not applied to the layer map due to potential data loss",
                 err
-            )));
+            );
         }
 
         // Between the sanity check and this compaction update, there could be new layers being flushed, but it should be fine because we only
@@ -3359,9 +3275,7 @@ impl Timeline {
         // find_gc_cutoffs will try accessing things below the cutoff. TODO: ideally, this should
         // be batched into `schedule_compaction_update`.
         let disk_consistent_lsn = self.disk_consistent_lsn.load();
-        self.schedule_uploads(disk_consistent_lsn, None)
-            .context("failed to schedule uploads")
-            .map_err(CompactionError::Other)?;
+        self.schedule_uploads(disk_consistent_lsn, None)?;
         // If a layer gets rewritten throughout gc-compaction, we need to keep that layer only in `compact_to` instead
         // of `compact_from`.
         let compact_from = {

storage_controller/src/main.rs

@@ -143,6 +143,10 @@ struct Cli {
     // Flag to use https for requests to pageserver API.
     #[arg(long, default_value = "false")]
     use_https_pageserver_api: bool,
+
+    /// Whether to load safekeeprs from the database and heartbeat them
+    #[arg(long, default_value = "false")]
+    load_safekeepers: bool,
 }
 
 enum StrictMode {
@@ -356,6 +360,7 @@ async fn async_main() -> anyhow::Result<()> {
         start_as_candidate: args.start_as_candidate,
         http_service_port: args.listen.port() as i32,
         use_https_pageserver_api: args.use_https_pageserver_api,
+        load_safekeepers: args.load_safekeepers,
     };
 
     // Validate that we can connect to the database

storage_controller/src/service.rs

@@ -394,6 +394,8 @@ pub struct Config {
     pub long_reconcile_threshold: Duration,
 
     pub use_https_pageserver_api: bool,
+
+    pub load_safekeepers: bool,
 }
 
 impl From<DatabaseError> for ApiError {
@@ -1410,15 +1412,20 @@ impl Service {
             .set(nodes.len() as i64);
 
         tracing::info!("Loading safekeepers from database...");
-        let safekeepers = persistence
-            .list_safekeepers()
-            .await?
-            .into_iter()
-            .map(|skp| Safekeeper::from_persistence(skp, CancellationToken::new()))
-            .collect::<Vec<_>>();
+        let safekeepers = if config.load_safekeepers {
+            persistence
+                .list_safekeepers()
+                .await?
+                .into_iter()
+                .map(|skp| Safekeeper::from_persistence(skp, CancellationToken::new()))
+                .collect::<Vec<_>>()
+        } else {
+            tracing::info!("Skipping safekeeper loading");
+            Default::default()
+        };
+
         let safekeepers: HashMap<NodeId, Safekeeper> =
             safekeepers.into_iter().map(|n| (n.get_id(), n)).collect();
-        tracing::info!("Loaded {} safekeepers from database.", safekeepers.len());
 
         tracing::info!("Loading shards from database...");
         let mut tenant_shard_persistence = persistence.load_active_tenant_shards().await?;
@@ -8059,7 +8066,8 @@ impl Service {
     ) -> Result<(), DatabaseError> {
         let node_id = NodeId(record.id as u64);
         self.persistence.safekeeper_upsert(record.clone()).await?;
-        {
+
+        if self.config.load_safekeepers {
             let mut locked = self.inner.write().unwrap();
             let mut safekeepers = (*locked.safekeepers).clone();
             match safekeepers.entry(node_id) {
@@ -8091,7 +8099,7 @@ impl Service {
             .await?;
         let node_id = NodeId(id as u64);
         // After the change has been persisted successfully, update the in-memory state
-        {
+        if self.config.load_safekeepers {
             let mut locked = self.inner.write().unwrap();
             let mut safekeepers = (*locked.safekeepers).clone();
             let sk = safekeepers

test_runner/fixtures/neon_fixtures.py

@@ -1133,6 +1133,13 @@ class NeonEnv:
         if self.storage_controller_config is not None:
             cfg["storage_controller"] = self.storage_controller_config
 
+        # Disable new storcon flag in compat tests
+        if config.test_may_use_compatibility_snapshot_binaries:
+            if "storage_controller" in cfg:
+                cfg["storage_controller"]["load_safekeepers"] = False
+            else:
+                cfg["storage_controller"] = {"load_safekeepers": False}
+
         # Create config for pageserver
         http_auth_type = "NeonJWT" if config.auth_enabled else "Trust"
         pg_auth_type = "NeonJWT" if config.auth_enabled else "Trust"
@@ -1162,8 +1169,6 @@ class NeonEnv:
                 # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
                 # the pageserver taking a long time to start up due to syncfs flushing other tests' data
                 "no_sync": True,
-                # Look for gaps in WAL received from safekeepeers
-                "validate_wal_contiguity": True,
             }
 
             # Batching (https://github.com/neondatabase/neon/issues/9377):
@@ -1176,12 +1181,11 @@ class NeonEnv:
 
             if config.test_may_use_compatibility_snapshot_binaries:
                 log.info(
-                    "Skipping prev heatmap settings to avoid forward-compatibility related test failures"
+                    "Skipping WAL contiguity validation to avoid forward-compatibility related test failures"
                 )
             else:
                 # Look for gaps in WAL received from safekeepeers
-                ps_cfg["load_previous_heatmap"] = True
-                ps_cfg["generate_unarchival_heatmap"] = True
+                ps_cfg["validate_wal_contiguity"] = True
 
             get_vectored_concurrent_io = self.pageserver_get_vectored_concurrent_io
             if get_vectored_concurrent_io is not None:

test_runner/regress/test_vm_bits.py

@@ -327,9 +327,9 @@ def test_check_visibility_map(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
         log.info(f"pgbench run {i+1}/{PGBENCH_RUNS}")
         endpoint.safe_psql(f"create database {dbname}")
         connstr = endpoint.connstr(dbname=dbname)
-        # Initialize the data set, but don't vacuum yet.
-        pg_bin.run(["pgbench", "-i", "-s", "8", "-n", connstr])
-        # Vacuum to create the visibility map, and freeze the tuples to set the frozen bit.
+        # pgbench -i will automatically vacuum the tables. This creates the visibility map.
+        pg_bin.run(["pgbench", "-i", "-s", "10", connstr])
+        # Freeze the tuples to set the initial frozen bit.
         endpoint.safe_psql("vacuum freeze", dbname=dbname)
         # Run pgbench.
         pg_bin.run(["pgbench", "-c", "32", "-j", "8", "-T", "10", connstr])
@@ -354,3 +354,19 @@ def test_check_visibility_map(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
             row = cur.fetchone()
             assert row is not None
             assert row[0] == 0, f"{row[0]} inconsistent VM pages (frozen)"
+
+    # Vacuum and freeze the tables, and check that the visibility map is still accurate.
+    for dbname in dbnames:
+        log.info(f"Vacuuming and checking visibility map for {dbname}")
+        with endpoint.cursor(dbname=dbname) as cur:
+            cur.execute("vacuum freeze")
+
+            cur.execute("select count(*) from pg_check_visible('pgbench_accounts')")
+            row = cur.fetchone()
+            assert row is not None
+            assert row[0] == 0, f"{row[0]} inconsistent VM pages (visible)"
+
+            cur.execute("select count(*) from pg_check_frozen('pgbench_accounts')")
+            row = cur.fetchone()
+            assert row is not None
+            assert row[0] == 0, f"{row[0]} inconsistent VM pages (frozen)"