Zero aux files on any aux file ingest.

Ignore aux files ingest and listing in basebackup for one tenant.
As one off hack to make it run right now.
2026-08-12 09:19:39 +00:00 · 2024-02-22 14:59:29 +03:00 · 2024-02-22 09:50:14 +03:00
8 changed files with 119 additions and 200 deletions
@@ -45,6 +45,7 @@ use std::{thread, time::Duration};
 use anyhow::{Context, Result};
 use chrono::Utc;
 use clap::Arg;
+use nix::sys::signal::{kill, Signal};
 use signal_hook::consts::{SIGQUIT, SIGTERM};
 use signal_hook::{consts::SIGINT, iterator::Signals};
 use tracing::{error, info};
@@ -52,9 +53,7 @@ use url::Url;

 use compute_api::responses::ComputeStatus;

-use compute_tools::compute::{
-    forward_termination_signal, ComputeNode, ComputeState, ParsedSpec, PG_PID,
-};
+use compute_tools::compute::{ComputeNode, ComputeState, ParsedSpec, PG_PID, SYNC_SAFEKEEPERS_PID};
 use compute_tools::configurator::launch_configurator;
 use compute_tools::extension_server::get_pg_version;
 use compute_tools::http::api::launch_http_server;
@@ -395,15 +394,6 @@ fn main() -> Result<()> {
        info!("synced safekeepers at lsn {lsn}");
    }

-    let mut state = compute.state.lock().unwrap();
-    if state.status == ComputeStatus::TerminationPending {
-        state.status = ComputeStatus::Terminated;
-        compute.state_changed.notify_all();
-        // we were asked to terminate gracefully, don't exit to avoid restart
-        delay_exit = true
-    }
-    drop(state);
-
    if let Err(err) = compute.check_for_core_dumps() {
        error!("error while checking for core dumps: {err:?}");
    }
@@ -533,7 +523,16 @@ fn cli() -> clap::Command {
 /// wait for termination which would be easy then.
 fn handle_exit_signal(sig: i32) {
    info!("received {sig} termination signal");
-    forward_termination_signal();
+    let ss_pid = SYNC_SAFEKEEPERS_PID.load(Ordering::SeqCst);
+    if ss_pid != 0 {
+        let ss_pid = nix::unistd::Pid::from_raw(ss_pid as i32);
+        kill(ss_pid, Signal::SIGTERM).ok();
+    }
+    let pg_pid = PG_PID.load(Ordering::SeqCst);
+    if pg_pid != 0 {
+        let pg_pid = nix::unistd::Pid::from_raw(pg_pid as i32);
+        kill(pg_pid, Signal::SIGTERM).ok();
+    }
    exit(1);
 }

@@ -28,8 +28,6 @@ use compute_api::responses::{ComputeMetrics, ComputeStatus};
 use compute_api::spec::{ComputeFeature, ComputeMode, ComputeSpec};
 use utils::measured_stream::MeasuredReader;

-use nix::sys::signal::{kill, Signal};
-
 use remote_storage::{DownloadError, RemotePath};

 use crate::checker::create_availability_check_data;
@@ -1324,17 +1322,3 @@ LIMIT 100",
        Ok(remote_ext_metrics)
    }
 }
-
-pub fn forward_termination_signal() {
-    let ss_pid = SYNC_SAFEKEEPERS_PID.load(Ordering::SeqCst);
-    if ss_pid != 0 {
-        let ss_pid = nix::unistd::Pid::from_raw(ss_pid as i32);
-        kill(ss_pid, Signal::SIGTERM).ok();
-    }
-    let pg_pid = PG_PID.load(Ordering::SeqCst);
-    if pg_pid != 0 {
-        let pg_pid = nix::unistd::Pid::from_raw(pg_pid as i32);
-        // use 'immediate' shutdown (SIGQUIT): https://www.postgresql.org/docs/current/server-shutdown.html
-        kill(pg_pid, Signal::SIGQUIT).ok();
-    }
-}
@@ -5,7 +5,6 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::thread;

-use crate::compute::forward_termination_signal;
 use crate::compute::{ComputeNode, ComputeState, ParsedSpec};
 use compute_api::requests::ConfigurationRequest;
 use compute_api::responses::{ComputeStatus, ComputeStatusResponse, GenericAPIError};
@@ -124,17 +123,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
            }
        }

-        (&Method::POST, "/terminate") => {
-            info!("serving /terminate POST request");
-            match handle_terminate_request(compute).await {
-                Ok(()) => Response::new(Body::empty()),
-                Err((msg, code)) => {
-                    error!("error handling /terminate request: {msg}");
-                    render_json_error(&msg, code)
-                }
-            }
-        }
-
        // download extension files from remote extension storage on demand
        (&Method::POST, route) if route.starts_with("/extension_server/") => {
            info!("serving {:?} POST request", route);
@@ -309,49 +297,6 @@ fn render_json_error(e: &str, status: StatusCode) -> Response<Body> {
        .unwrap()
 }

-async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (String, StatusCode)> {
-    {
-        let mut state = compute.state.lock().unwrap();
-        if state.status == ComputeStatus::Terminated {
-            return Ok(());
-        }
-        if state.status != ComputeStatus::Empty && state.status != ComputeStatus::Running {
-            let msg = format!(
-                "invalid compute status for termination request: {:?}",
-                state.status.clone()
-            );
-            return Err((msg, StatusCode::PRECONDITION_FAILED));
-        }
-        state.status = ComputeStatus::TerminationPending;
-        compute.state_changed.notify_all();
-        drop(state);
-    }
-    forward_termination_signal();
-    info!("sent signal and notified waiters");
-
-    // Spawn a blocking thread to wait for compute to become Terminated.
-    // This is needed to do not block the main pool of workers and
-    // be able to serve other requests while some particular request
-    // is waiting for compute to finish configuration.
-    let c = compute.clone();
-    task::spawn_blocking(move || {
-        let mut state = c.state.lock().unwrap();
-        while state.status != ComputeStatus::Terminated {
-            state = c.state_changed.wait(state).unwrap();
-            info!(
-                "waiting for compute to become Terminated, current status: {:?}",
-                state.status
-            );
-        }
-
-        Ok(())
-    })
-    .await
-    .unwrap()?;
-    info!("terminated Postgres");
-    Ok(())
-}
-
 // Main Hyper HTTP server function that runs it and blocks waiting on it forever.
 #[tokio::main]
 async fn serve(port: u16, state: Arc<ComputeNode>) {
@@ -168,29 +168,6 @@ paths:
              schema:
                $ref: "#/components/schemas/GenericError"

-  /terminate:
-    post:
-      tags:
-      - Terminate
-      summary: Terminate Postgres and wait for it to exit
-      description: ""
-      operationId: terminate
-      responses:
-        200:
-          description: Result
-        412:
-          description: "wrong state"
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-        500:
-          description: "Unexpected error"
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-
 components:
  securitySchemes:
    JWT:
@@ -652,9 +652,7 @@ impl Endpoint {
                        }
                        ComputeStatus::Empty
                        | ComputeStatus::ConfigurationPending
-                        | ComputeStatus::Configuration
-                        | ComputeStatus::TerminationPending
-                        | ComputeStatus::Terminated => {
+                        | ComputeStatus::Configuration => {
                            bail!("unexpected compute status: {:?}", state.status)
                        }
                    }
@@ -52,10 +52,6 @@ pub enum ComputeStatus {
    // compute will exit soon or is waiting for
    // control-plane to terminate it.
    Failed,
-    // Termination requested
-    TerminationPending,
-    // Terminated Postgres
-    Terminated,
 }

 fn rfc3339_serialize<S>(x: &Option<DateTime<Utc>>, s: S) -> Result<S::Ok, S::Error>
@@ -13,8 +13,9 @@
 use anyhow::{anyhow, bail, ensure, Context};
 use bytes::{BufMut, Bytes, BytesMut};
 use fail::fail_point;
+
 use pageserver_api::key::{key_to_slru_block, Key};
-use postgres_ffi::pg_constants;
+
 use std::fmt::Write as FmtWrite;
 use std::time::SystemTime;
 use tokio::io;
@@ -273,7 +274,7 @@ where
            slru_builder.finish().await?;
        }

-        let mut min_restart_lsn: Lsn = Lsn::MAX;
+        let min_restart_lsn: Lsn = Lsn::MAX;
        // Create tablespace directories
        for ((spcnode, dbnode), has_relmap_file) in
            self.timeline.list_dbdirs(self.lsn, self.ctx).await?
@@ -308,21 +309,24 @@ where
                }
            }

-            for (path, content) in self.timeline.list_aux_files(self.lsn, self.ctx).await? {
-                if path.starts_with("pg_replslot") {
-                    let offs = pg_constants::REPL_SLOT_ON_DISK_OFFSETOF_RESTART_LSN;
-                    let restart_lsn = Lsn(u64::from_le_bytes(
-                        content[offs..offs + 8].try_into().unwrap(),
-                    ));
-                    info!("Replication slot {} restart LSN={}", path, restart_lsn);
-                    min_restart_lsn = Lsn::min(min_restart_lsn, restart_lsn);
-                }
-                let header = new_tar_header(&path, content.len() as u64)?;
-                self.ar
-                    .append(&header, &*content)
-                    .await
-                    .context("could not add aux file to basebackup tarball")?;
-            }
+            // one-off hack: disable listing aux files
+            // {
+            //     for (path, content) in self.timeline.list_aux_files(self.lsn, self.ctx).await? {
+            //         if path.starts_with("pg_replslot") {
+            //             let offs = pg_constants::REPL_SLOT_ON_DISK_OFFSETOF_RESTART_LSN;
+            //             let restart_lsn = Lsn(u64::from_le_bytes(
+            //                 content[offs..offs + 8].try_into().unwrap(),
+            //             ));
+            //             info!("Replication slot {} restart LSN={}", path, restart_lsn);
+            //             min_restart_lsn = Lsn::min(min_restart_lsn, restart_lsn);
+            //         }
+            //         let header = new_tar_header(&path, content.len() as u64)?;
+            //         self.ar
+            //             .append(&header, &*content)
+            //             .await
+            //             .context("could not add aux file to basebackup tarball")?;
+            //     }
+            // }
        }
        if min_restart_lsn != Lsn::MAX {
            info!(
@@ -671,6 +671,7 @@ impl Timeline {
        self.get(CHECKPOINT_KEY, lsn, ctx).await
    }

+    #[allow(dead_code)]
    pub(crate) async fn list_aux_files(
        &self,
        lsn: Lsn,
@@ -1389,81 +1390,96 @@ impl<'a> DatadirModification<'a> {

    pub async fn put_file(
        &mut self,
-        path: &str,
-        content: &[u8],
-        ctx: &RequestContext,
+        _path: &str,
+        _content: &[u8],
+        _ctx: &RequestContext,
    ) -> anyhow::Result<()> {
-        let file_path = path.to_string();
-        let content = if content.is_empty() {
-            None
-        } else {
-            Some(Bytes::copy_from_slice(content))
+        // one-off hack: zero aux files on any ingest
+        let dir = AuxFilesDirectory {
+            files: HashMap::new(),
        };
-
-        let dir = if let Some(mut dir) = self.pending_aux_files.take() {
-            // We already updated aux files in `self`: emit a delta and update our latest value
-
-            self.put(
-                AUX_FILES_KEY,
-                Value::WalRecord(NeonWalRecord::AuxFile {
-                    file_path: file_path.clone(),
-                    content: content.clone(),
-                }),
-            );
-
-            dir.upsert(file_path, content);
-            dir
-        } else {
-            // Check if the AUX_FILES_KEY is initialized
-            match self.get(AUX_FILES_KEY, ctx).await {
-                Ok(dir_bytes) => {
-                    let mut dir = AuxFilesDirectory::des(&dir_bytes)?;
-                    // Key is already set, we may append a delta
-                    self.put(
-                        AUX_FILES_KEY,
-                        Value::WalRecord(NeonWalRecord::AuxFile {
-                            file_path: file_path.clone(),
-                            content: content.clone(),
-                        }),
-                    );
-                    dir.upsert(file_path, content);
-                    dir
-                }
-                Err(
-                    e @ (PageReconstructError::AncestorStopping(_)
-                    | PageReconstructError::Cancelled
-                    | PageReconstructError::AncestorLsnTimeout(_)),
-                ) => {
-                    // Important that we do not interpret a shutdown error as "not found" and thereby
-                    // reset the map.
-                    return Err(e.into());
-                }
-                // FIXME: PageReconstructError doesn't have an explicit variant for key-not-found, so
-                // we are assuming that all _other_ possible errors represents a missing key.  If some
-                // other error occurs, we may incorrectly reset the map of aux files.
-                Err(PageReconstructError::Other(_) | PageReconstructError::WalRedo(_)) => {
-                    // Key is missing, we must insert an image as the basis for subsequent deltas.
-
-                    let mut dir = AuxFilesDirectory {
-                        files: HashMap::new(),
-                    };
-                    dir.upsert(file_path, content);
-                    self.put(
-                        AUX_FILES_KEY,
-                        Value::Image(Bytes::from(
-                            AuxFilesDirectory::ser(&dir).context("serialize")?,
-                        )),
-                    );
-                    dir
-                }
-            }
-        };
-
+        self.put(
+            AUX_FILES_KEY,
+            Value::Image(Bytes::from(
+                AuxFilesDirectory::ser(&dir).context("serialize")?,
+            )),
+        );
        self.pending_directory_entries
            .push((DirectoryKind::AuxFiles, dir.files.len()));
        self.pending_aux_files = Some(dir);
-
        Ok(())
+
+        // let file_path = path.to_string();
+        // let content = if content.is_empty() {
+        //     None
+        // } else {
+        //     Some(Bytes::copy_from_slice(content))
+        // };
+
+        // let dir = if let Some(mut dir) = self.pending_aux_files.take() {
+        //     // We already updated aux files in `self`: emit a delta and update our latest value
+
+        //     self.put(
+        //         AUX_FILES_KEY,
+        //         Value::WalRecord(NeonWalRecord::AuxFile {
+        //             file_path: file_path.clone(),
+        //             content: content.clone(),
+        //         }),
+        //     );
+
+        //     dir.upsert(file_path, content);
+        //     dir
+        // } else {
+        //     // Check if the AUX_FILES_KEY is initialized
+        //     match self.get(AUX_FILES_KEY, ctx).await {
+        //         Ok(dir_bytes) => {
+        //             let mut dir = AuxFilesDirectory::des(&dir_bytes)?;
+        //             // Key is already set, we may append a delta
+        //             self.put(
+        //                 AUX_FILES_KEY,
+        //                 Value::WalRecord(NeonWalRecord::AuxFile {
+        //                     file_path: file_path.clone(),
+        //                     content: content.clone(),
+        //                 }),
+        //             );
+        //             dir.upsert(file_path, content);
+        //             dir
+        //         }
+        //         Err(
+        //             e @ (PageReconstructError::AncestorStopping(_)
+        //             | PageReconstructError::Cancelled
+        //             | PageReconstructError::AncestorLsnTimeout(_)),
+        //         ) => {
+        //             // Important that we do not interpret a shutdown error as "not found" and thereby
+        //             // reset the map.
+        //             return Err(e.into());
+        //         }
+        //         // FIXME: PageReconstructError doesn't have an explicit variant for key-not-found, so
+        //         // we are assuming that all _other_ possible errors represents a missing key.  If some
+        //         // other error occurs, we may incorrectly reset the map of aux files.
+        //         Err(PageReconstructError::Other(_) | PageReconstructError::WalRedo(_)) => {
+        //             // Key is missing, we must insert an image as the basis for subsequent deltas.
+
+        //             let mut dir = AuxFilesDirectory {
+        //                 files: HashMap::new(),
+        //             };
+        //             dir.upsert(file_path, content);
+        //             self.put(
+        //                 AUX_FILES_KEY,
+        //                 Value::Image(Bytes::from(
+        //                     AuxFilesDirectory::ser(&dir).context("serialize")?,
+        //                 )),
+        //             );
+        //             dir
+        //         }
+        //     }
+        // };
+
+        // self.pending_directory_entries
+        //     .push((DirectoryKind::AuxFiles, dir.files.len()));
+        // self.pending_aux_files = Some(dir);
+
+        // Ok(())
    }

    ///
Author	SHA1	Message	Date
Arseny Sher	22a787ec9c	Zero aux files on any aux file ingest.	2024-02-22 14:59:29 +03:00
Arseny Sher	9c662e65a2	Ignore aux files ingest and listing in basebackup for one tenant. As one off hack to make it run right now.	2024-02-22 09:50:14 +03:00