Log detection of duplicate layer

Make clippy happy
Add removed BufferedHistoricLayerCoverage.contains method
2026-07-07 22:20:36 +00:00 · 2023-05-16 09:53:45 +03:00 · 2023-05-15 22:59:15 +03:00 · 2023-05-15 18:24:35 +03:00 · 2023-05-15 17:34:30 +03:00 · 2023-05-15 17:25:00 +03:00
17 changed files with 178 additions and 228 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2859,22 +2859,22 @@ dependencies = [

 [[package]]
 name = "pin-project"
-version = "1.1.0"
+version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c95a7476719eab1e366eaf73d0260af3021184f18177925b07f54b30089ceead"
+checksum = "ad29a609b6bcd67fee905812e544992d216af9d755757c05ed2d0e15a74c6ecc"
 dependencies = [
 "pin-project-internal",
 ]

 [[package]]
 name = "pin-project-internal"
-version = "1.1.0"
+version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39407670928234ebc5e6e580247dd567ad73a3578460c5990f9503df207e8f07"
+checksum = "069bdb1e05adc7a8990dce9cc75370895fbe4e3d58b9b73bf1aee56359344a55"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.15",
+ "syn 1.0.109",
 ]

 [[package]]
@@ -4979,7 +4979,6 @@ dependencies = [
 "metrics",
 "nix",
 "once_cell",
- "pin-project",
 "pin-project-lite",
 "pq_proto",
 "rand",
--- a/libs/utils/Cargo.toml
+++ b/libs/utils/Cargo.toml
@@ -41,9 +41,6 @@ pq_proto.workspace = true
 metrics.workspace = true
 workspace_hack.workspace = true

-# needed to have pin-projected with PinnedDrop
-pin-project = "1.1"
-
 [dev-dependencies]
 byteorder.workspace = true
 bytes.workspace = true
--- a/libs/utils/src/lib.rs
+++ b/libs/utils/src/lib.rs
@@ -147,56 +147,3 @@ macro_rules! const_assert {
        const _: () = assert!($($args)*);
    };
 }
-
-pub mod cancel_log {
-    pub trait CancelationLoggingExt {
-        fn log_being_canceled(self, step: &'static str) -> LogCancelation<Self>
-        where
-            Self: Sized;
-    }
-
-    impl<Fut: std::future::Future> CancelationLoggingExt for Fut {
-        fn log_being_canceled(self, step: &'static str) -> LogCancelation<Self>
-        where
-            Self: Sized,
-        {
-            LogCancelation {
-                inner: self,
-                name: Some(step),
-                span: tracing::Span::current(),
-            }
-        }
-    }
-
-    #[pin_project::pin_project(PinnedDrop)]
-    pub struct LogCancelation<Fut> {
-        #[pin]
-        inner: Fut,
-        name: Option<&'static str>,
-        span: tracing::Span,
-    }
-
-    impl<Fut: std::future::Future> std::future::Future for LogCancelation<Fut> {
-        type Output = Fut::Output;
-
-        fn poll(
-            self: std::pin::Pin<&mut Self>,
-            cx: &mut std::task::Context<'_>,
-        ) -> std::task::Poll<Self::Output> {
-            let this = self.project();
-            let ready = std::task::ready!(this.inner.poll(cx));
-            this.name.take();
-            std::task::Poll::Ready(ready)
-        }
-    }
-
-    #[pin_project::pinned_drop]
-    impl<Fut> PinnedDrop for LogCancelation<Fut> {
-        fn drop(self: std::pin::Pin<&mut Self>) {
-            if let Some(name) = self.name {
-                let _g = self.span.enter();
-                tracing::info!("canceled while waiting for {name}");
-            }
-        }
-    }
-}
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -29,7 +29,6 @@ use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError, Timeline}
 use crate::{config::PageServerConf, tenant::mgr};
 use utils::{
    auth::JwtAuth,
-    cancel_log::CancelationLoggingExt,
    http::{
        endpoint::{self, attach_openapi_ui, auth_middleware, check_permission_with},
        error::{ApiError, HttpErrorBody},
@@ -187,7 +186,6 @@ async fn build_timeline_info(
                    CancellationToken::new(),
                    ctx,
                )
-                .log_being_canceled("get_current_logical_size_non_incremental")
                .await?,
        );
    }
@@ -340,9 +338,7 @@ async fn timeline_detail_handler(request: Request<Body>) -> Result<Response<Body
    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download);

    let timeline_info = async {
-        let tenant = mgr::get_tenant(tenant_id, true)
-            .log_being_canceled("get_tenant")
-            .await?;
+        let tenant = mgr::get_tenant(tenant_id, true).await?;

        let timeline = tenant
            .get_timeline(timeline_id, false)
@@ -353,7 +349,6 @@ async fn timeline_detail_handler(request: Request<Body>) -> Result<Response<Body
            include_non_incremental_logical_size.unwrap_or(false),
            &ctx,
        )
-        .log_being_canceled("build_timeline_info")
        .await
        .context("get local timeline info")
        .map_err(ApiError::InternalServerError)?;
@@ -361,7 +356,6 @@ async fn timeline_detail_handler(request: Request<Body>) -> Result<Response<Body
        Ok::<_, ApiError>(timeline_info)
    }
    .instrument(info_span!("timeline_detail", tenant = %tenant_id, timeline = %timeline_id))
-    .log_being_canceled("whole response")
    .await?;

    json_response(StatusCode::OK, timeline_info)
--- a/pageserver/src/tenant/layer_map.rs
+++ b/pageserver/src/tenant/layer_map.rs
@@ -285,6 +285,11 @@ where
        }
    }

+    pub fn contains(&self, key_range: &Range<Key>, lsn_range: &Range<Lsn>, is_image: bool) -> bool {
+        let key = historic_layer_coverage::LayerKey::from_ranges(key_range, lsn_range, is_image);
+        self.historic.contains(&key)
+    }
+
    ///
    /// Remove an on-disk layer from the map.
    ///
--- a/pageserver/src/tenant/layer_map/historic_layer_coverage.rs
+++ b/pageserver/src/tenant/layer_map/historic_layer_coverage.rs
@@ -53,6 +53,20 @@ impl<'a, L: crate::tenant::storage_layer::Layer + ?Sized> From<&'a L> for LayerK
    }
 }

+impl LayerKey {
+    pub fn from_ranges(
+        kr: &Range<crate::tenant::layer_map::Key>,
+        lr: &Range<utils::lsn::Lsn>,
+        is_image: bool,
+    ) -> Self {
+        LayerKey {
+            key: kr.start.to_i128()..kr.end.to_i128(),
+            lsn: lr.start.0..lr.end.0,
+            is_image,
+        }
+    }
+}
+
 /// Efficiently queryable layer coverage for each LSN.
 ///
 /// Allows answering layer map queries very efficiently,
@@ -417,6 +431,14 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
        }
    }

+    pub fn contains(&self, layer_key: &LayerKey) -> bool {
+        match self.buffer.get(layer_key) {
+            Some(None) => false,                         // layer remove was buffered
+            Some(_) => true,                             // layer insert was buffered
+            None => self.layers.contains_key(layer_key), // no buffered ops for this layer
+        }
+    }
+
    pub fn insert(&mut self, layer_key: LayerKey, value: Value) {
        self.buffer.insert(layer_key, Some(value));
    }
--- a/pageserver/src/tenant/storage_layer/delta_layer.rs
+++ b/pageserver/src/tenant/storage_layer/delta_layer.rs
@@ -867,6 +867,14 @@ pub struct DeltaLayerWriter {
 }

 impl DeltaLayerWriter {
+    pub fn key_start(&self) -> Key {
+        self.inner.as_ref().unwrap().key_start
+    }
+
+    pub fn lsn_range(&self) -> Range<Lsn> {
+        self.inner.as_ref().unwrap().lsn_range.clone()
+    }
+
    ///
    /// Start building a new delta layer.
    ///
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -3375,7 +3375,54 @@ impl Timeline {
                        || contains_hole
                    {
                        // ... if so, flush previous layer and prepare to write new one
-                        new_layers.push(writer.take().unwrap().finish(prev_key.unwrap().next())?);
+
+                        let end_key = prev_key.unwrap().next();
+
+                        let w = writer.take().unwrap();
+
+                        // If an identical L1 layer already exists, no need to create a new one.
+                        //
+                        // This can happen if compaction is interrupted after it has already
+                        // created some or all of the L1 layers, but has not deleted the L0 layers
+                        // yet, so that on next compaction, we do the same work again.
+                        //
+                        // NOTE: this is racy, if there can be any other task that concurrently
+                        // creates L1 layers. Currently, there can be only one compaction task
+                        // running at any time, so this is fine.
+                        //
+                        // Also we hold `layer_removal_cs` guard which should prevent race condition
+                        // even if there are two or more concurrent compaction tasks.
+                        //
+                        // But there is an opposite issue: we check presence of duplicates under
+                        // `layers` shared lock, but then it is released. So there is a gap between
+                        // this check and adding new layer to layer map. In principle in this gap some
+                        // some other task (i.e. GC) can drop this layer and we already abandon insertion
+                        // of duplicate layer. As a result there will be no such layer at all.
+                        // In other words: we have some state S1 of pageserver where layer L1 can be removed by GC.
+                        // Then we run compaction and it switch pageserver to the state S2 which writes duplicate of
+                        // layer L1 and where it can not be removed. With this patch it is possible that
+                        // we switch pageserver to state S2 but... with L1 lost.
+                        // It is just hypothetical situation and there is no such concrete scenario which
+                        // reproduces this problem. So let's take this risk.
+                        //
+                        if self.layers.read().unwrap().contains(
+                            &(w.key_start()..end_key),
+                            &w.lsn_range(),
+                            false, // not an image layer
+                        ) {
+                            info!(
+                                "Skip generation of duplicate layer {}_{}__{}_{}",
+                                w.key_start(),
+                                end_key,
+                                w.lsn_range().start,
+                                w.lsn_range().end
+                            );
+                            drop(w);
+                        } else {
+                            let new_layer = w.finish(end_key)?;
+
+                            new_layers.push(new_layer);
+                        }
                        writer = None;

                        if contains_hole {
@@ -3432,6 +3479,10 @@ impl Timeline {

        drop(all_keys_iter); // So that deltas_to_compact is no longer borrowed

+        fail_point!("compact-level0-phase1-finish", |_| {
+            Err(anyhow::anyhow!("failpoint compact-level0-phase1-finish").into())
+        });
+
        Ok(CompactLevel0Phase1Result {
            new_layers,
            deltas_to_compact,
--- a/pgxn/neon/libpagestore.c
+++ b/pgxn/neon/libpagestore.c
@@ -192,9 +192,8 @@ retry:
 		{
 			if (!PQconsumeInput(pageserver_conn))
 			{
-				char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
-				neon_log(LOG, "could not get response from pageserver: %s", msg);
-				pfree(msg);
+				neon_log(LOG, "could not get response from pageserver: %s",
+						 PQerrorMessage(pageserver_conn));
 				return -1;
 			}
 		}
@@ -344,7 +343,7 @@ pageserver_receive(void)
 			resp = NULL;
 		}
 		else if (rc == -2)
-			neon_log(ERROR, "could not read COPY data: %s", pchomp(PQerrorMessage(pageserver_conn)));
+			neon_log(ERROR, "could not read COPY data: %s", PQerrorMessage(pageserver_conn));
 		else
 			neon_log(ERROR, "unexpected PQgetCopyData return value: %d", rc);
 	}
@@ -368,7 +367,7 @@ pageserver_flush(void)
 	}
 	else if (PQflush(pageserver_conn))
 	{
-		char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
+		char	   *msg = PQerrorMessage(pageserver_conn);

 		pageserver_disconnect();
 		neon_log(ERROR, "failed to flush page requests: %s", msg);
--- a/proxy/src/auth.rs
+++ b/proxy/src/auth.rs
@@ -7,7 +7,6 @@ mod credentials;
 pub use credentials::ClientCredentials;

 mod password_hack;
-pub use password_hack::parse_endpoint_param;
 use password_hack::PasswordHackPayload;

 mod flow;
@@ -45,10 +44,10 @@ pub enum AuthErrorImpl {
    #[error(
        "Endpoint ID is not specified. \
        Either please upgrade the postgres client library (libpq) for SNI support \
-        or pass the endpoint ID (first part of the domain name) as a parameter: '?options=endpoint%3D<endpoint-id>'. \
+        or pass the endpoint ID (first part of the domain name) as a parameter: '?options=project%3D<endpoint-id>'. \
        See more at https://neon.tech/sni"
    )]
-    MissingEndpointName,
+    MissingProjectName,

    #[error("password authentication failed for user '{0}'")]
    AuthFailed(Box<str>),
@@ -89,7 +88,7 @@ impl UserFacingError for AuthError {
            AuthFailed(_) => self.to_string(),
            BadAuthMethod(_) => self.to_string(),
            MalformedPassword(_) => self.to_string(),
-            MissingEndpointName => self.to_string(),
+            MissingProjectName => self.to_string(),
            Io(_) => "Internal error".to_string(),
        }
    }
--- a/proxy/src/auth/backend/hacks.rs
+++ b/proxy/src/auth/backend/hacks.rs
@@ -52,8 +52,8 @@ pub async fn password_hack(
        .authenticate()
        .await?;

-    info!(project = &payload.endpoint, "received missing parameter");
-    creds.project = Some(payload.endpoint);
+    info!(project = &payload.project, "received missing parameter");
+    creds.project = Some(payload.project);

    let mut node = api.wake_compute(extra, creds).await?;
    node.config.password(payload.password);
--- a/proxy/src/auth/credentials.rs
+++ b/proxy/src/auth/credentials.rs
@@ -1,7 +1,6 @@
 //! User credentials used in authentication.

-use crate::{auth::password_hack::parse_endpoint_param, error::UserFacingError};
-use itertools::Itertools;
+use crate::error::UserFacingError;
 use pq_proto::StartupMessageParams;
 use std::collections::HashSet;
 use thiserror::Error;
@@ -62,15 +61,7 @@ impl<'a> ClientCredentials<'a> {
        // Project name might be passed via PG's command-line options.
        let project_option = params
            .options_raw()
-            .and_then(|options| {
-                // We support both `project` (deprecated) and `endpoint` options for backward compatibility.
-                // However, if both are present, we don't exactly know which one to use.
-                // Therefore we require that only one of them is present.
-                options
-                    .filter_map(parse_endpoint_param)
-                    .at_most_one()
-                    .ok()?
-            })
+            .and_then(|mut options| options.find_map(|opt| opt.strip_prefix("project=")))
            .map(|name| name.to_string());

        let project_from_domain = if let Some(sni_str) = sni {
@@ -186,51 +177,6 @@ mod tests {
        Ok(())
    }

-    #[test]
-    fn parse_endpoint_from_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "-ckey=1 endpoint=bar -c geqo=off"),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert_eq!(creds.project.as_deref(), Some("bar"));
-
-        Ok(())
-    }
-
-    #[test]
-    fn parse_three_endpoints_from_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            (
-                "options",
-                "-ckey=1 endpoint=one endpoint=two endpoint=three -c geqo=off",
-            ),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert!(creds.project.is_none());
-
-        Ok(())
-    }
-
-    #[test]
-    fn parse_when_endpoint_and_project_are_in_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "-ckey=1 endpoint=bar project=foo -c geqo=off"),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert!(creds.project.is_none());
-
-        Ok(())
-    }
-
    #[test]
    fn parse_projects_identical() -> anyhow::Result<()> {
        let options = StartupMessageParams::new([("user", "john_doe"), ("options", "project=baz")]);
--- a/proxy/src/auth/flow.rs
+++ b/proxy/src/auth/flow.rs
@@ -91,7 +91,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, PasswordHack> {
            // the user neither enabled SNI nor resorted to any other method
            // for passing the project name we rely on. We should show them
            // the most helpful error message and point to the documentation.
-            .ok_or(AuthErrorImpl::MissingEndpointName)?;
+            .ok_or(AuthErrorImpl::MissingProjectName)?;

        Ok(payload)
    }
--- a/proxy/src/auth/password_hack.rs
+++ b/proxy/src/auth/password_hack.rs
@@ -6,55 +6,27 @@
 use bstr::ByteSlice;

 pub struct PasswordHackPayload {
-    pub endpoint: String,
+    pub project: String,
    pub password: Vec<u8>,
 }

 impl PasswordHackPayload {
    pub fn parse(bytes: &[u8]) -> Option<Self> {
        // The format is `project=<utf-8>;<password-bytes>`.
-        let mut iter = bytes.splitn_str(2, ";");
-        let endpoint = iter.next()?.to_str().ok()?;
-        let endpoint = parse_endpoint_param(endpoint)?.to_owned();
+        let mut iter = bytes.strip_prefix(b"project=")?.splitn_str(2, ";");
+        let project = iter.next()?.to_str().ok()?.to_owned();
        let password = iter.next()?.to_owned();

-        Some(Self { endpoint, password })
+        Some(Self { project, password })
    }
 }

-pub fn parse_endpoint_param(bytes: &str) -> Option<&str> {
-    bytes
-        .strip_prefix("project=")
-        .or_else(|| bytes.strip_prefix("endpoint="))
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;

    #[test]
-    fn parse_endpoint_param_fn() {
-        let input = "";
-        assert!(parse_endpoint_param(input).is_none());
-
-        let input = "project=";
-        assert_eq!(parse_endpoint_param(input), Some(""));
-
-        let input = "project=foobar";
-        assert_eq!(parse_endpoint_param(input), Some("foobar"));
-
-        let input = "endpoint=";
-        assert_eq!(parse_endpoint_param(input), Some(""));
-
-        let input = "endpoint=foobar";
-        assert_eq!(parse_endpoint_param(input), Some("foobar"));
-
-        let input = "other_option=foobar";
-        assert!(parse_endpoint_param(input).is_none());
-    }
-
-    #[test]
-    fn parse_password_hack_payload_project() {
+    fn parse_password_hack_payload() {
        let bytes = b"";
        assert!(PasswordHackPayload::parse(bytes).is_none());

@@ -62,33 +34,13 @@ mod tests {
        assert!(PasswordHackPayload::parse(bytes).is_none());

        let bytes = b"project=;";
-        let payload: PasswordHackPayload =
-            PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "");
+        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
+        assert_eq!(payload.project, "");
        assert_eq!(payload.password, b"");

        let bytes = b"project=foobar;pass;word";
        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "foobar");
-        assert_eq!(payload.password, b"pass;word");
-    }
-
-    #[test]
-    fn parse_password_hack_payload_endpoint() {
-        let bytes = b"";
-        assert!(PasswordHackPayload::parse(bytes).is_none());
-
-        let bytes = b"endpoint=";
-        assert!(PasswordHackPayload::parse(bytes).is_none());
-
-        let bytes = b"endpoint=;";
-        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "");
-        assert_eq!(payload.password, b"");
-
-        let bytes = b"endpoint=foobar;pass;word";
-        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "foobar");
+        assert_eq!(payload.project, "foobar");
        assert_eq!(payload.password, b"pass;word");
    }
 }
--- a/proxy/src/compute.rs
+++ b/proxy/src/compute.rs
@@ -1,4 +1,4 @@
-use crate::{auth::parse_endpoint_param, cancellation::CancelClosure, error::UserFacingError};
+use crate::{cancellation::CancelClosure, error::UserFacingError};
 use futures::{FutureExt, TryFutureExt};
 use itertools::Itertools;
 use pq_proto::StartupMessageParams;
@@ -279,7 +279,7 @@ fn filtered_options(params: &StartupMessageParams) -> Option<String> {
    #[allow(unstable_name_collisions)]
    let options: String = params
        .options_raw()?
-        .filter(|opt| parse_endpoint_param(opt).is_none())
+        .filter(|opt| !opt.starts_with("project="))
        .intersperse(" ") // TODO: use impl from std once it's stabilized
        .collect();

--- a/test_runner/regress/test_duplicate_layers.py
+++ b/test_runner/regress/test_duplicate_layers.py
@@ -0,0 +1,42 @@
+import time
+
+import pytest
+from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
+
+
+# Test duplicate layer detection
+#
+# This test sets fail point at the end of first compaction phase:
+# after flushing new L1 layers but before deletion of L0 layes
+# It should cause generation of duplicate L1 layer by compaction after restart
+@pytest.mark.timeout(600)
+def test_duplicate_layers(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
+    env = neon_env_builder.init_start()
+
+    # These warnings are expected, when the pageserver is restarted abruptly
+    env.pageserver.allowed_errors.append(".*found future image layer.*")
+    env.pageserver.allowed_errors.append(".*found future delta layer.*")
+    env.pageserver.allowed_errors.append(".*duplicate layer.*")
+
+    pageserver_http = env.pageserver.http_client()
+
+    # Use aggressive compaction and checkpoint settings
+    tenant_id, _ = env.neon_cli.create_tenant(
+        conf={
+            "checkpoint_distance": f"{1024 ** 2}",
+            "compaction_target_size": f"{1024 ** 2}",
+            "compaction_period": "1 s",
+            "compaction_threshold": "3",
+        }
+    )
+    endpoint = env.endpoints.create_start("main", tenant_id=tenant_id)
+    connstr = endpoint.connstr(options="-csynchronous_commit=off")
+    pg_bin.run_capture(["pgbench", "-i", "-s10", connstr])
+
+    pageserver_http.configure_failpoints(("compact-level0-phase1-finish", "exit"))
+
+    with pytest.raises(Exception):
+        pg_bin.run_capture(["pgbench", "-P1", "-N", "-c5", "-T500", "-Mprepared", connstr])
+    env.pageserver.stop()
+    env.pageserver.start()
+    time.sleep(10)  # let compaction to be performed
--- a/test_runner/regress/test_proxy.py
+++ b/test_runner/regress/test_proxy.py
@@ -5,18 +5,16 @@ import pytest
 from fixtures.neon_fixtures import PSQL, NeonProxy, VanillaPostgres


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_proxy_select_1(static_proxy: NeonProxy, option_name: str):
+def test_proxy_select_1(static_proxy: NeonProxy):
    """
    A simplest smoke test: check proxy against a local postgres instance.
    """

-    out = static_proxy.safe_psql("select 1", options=f"{option_name}=generic-project-name")
+    out = static_proxy.safe_psql("select 1", options="project=generic-project-name")
    assert out[0][0] == 1


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_password_hack(static_proxy: NeonProxy, option_name: str):
+def test_password_hack(static_proxy: NeonProxy):
    """
    Check the PasswordHack auth flow: an alternative to SCRAM auth for
    clients which can't provide the project/endpoint name via SNI or `options`.
@@ -25,12 +23,11 @@ def test_password_hack(static_proxy: NeonProxy, option_name: str):
    user = "borat"
    password = "password"
    static_proxy.safe_psql(
-        f"create role {user} with login password '{password}'",
-        options=f"{option_name}=irrelevant",
+        f"create role {user} with login password '{password}'", options="project=irrelevant"
    )

    # Note the format of `magic`!
-    magic = f"{option_name}=irrelevant;{password}"
+    magic = f"project=irrelevant;{password}"
    static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)

    # Must also check that invalid magic won't be accepted.
@@ -59,62 +56,55 @@ async def test_link_auth(vanilla_pg: VanillaPostgres, link_proxy: NeonProxy):
    assert out == "42"


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_proxy_options(static_proxy: NeonProxy, option_name: str):
+def test_proxy_options(static_proxy: NeonProxy):
    """
    Check that we pass extra `options` to the PostgreSQL server:
-    * `project=...` and `endpoint=...` shouldn't be passed at all
-    * (otherwise postgres will raise an error).
+    * `project=...` shouldn't be passed at all (otherwise postgres will raise an error).
    * everything else should be passed as-is.
    """

-    options = f"{option_name}=irrelevant -cproxytest.option=value"
+    options = "project=irrelevant -cproxytest.option=value"
    out = static_proxy.safe_psql("show proxytest.option", options=options)
    assert out[0][0] == "value"

-    options = f"-c proxytest.foo=\\ str {option_name}=irrelevant"
+    options = "-c proxytest.foo=\\ str project=irrelevant"
    out = static_proxy.safe_psql("show proxytest.foo", options=options)
    assert out[0][0] == " str"


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_auth_errors(static_proxy: NeonProxy, option_name: str):
+def test_auth_errors(static_proxy: NeonProxy):
    """
    Check that we throw very specific errors in some unsuccessful auth scenarios.
    """

    # User does not exist
    with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", options=f"{option_name}=irrelevant")
+        static_proxy.connect(user="pinocchio", options="project=irrelevant")
    text = str(exprinfo.value).strip()
    assert text.endswith("password authentication failed for user 'pinocchio'")

    static_proxy.safe_psql(
-        "create role pinocchio with login password 'magic'",
-        options=f"{option_name}=irrelevant",
+        "create role pinocchio with login password 'magic'", options="project=irrelevant"
    )

    # User exists, but password is missing
    with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password=None, options=f"{option_name}=irrelevant")
+        static_proxy.connect(user="pinocchio", password=None, options="project=irrelevant")
    text = str(exprinfo.value).strip()
    assert text.endswith("password authentication failed for user 'pinocchio'")

    # User exists, but password is wrong
    with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password="bad", options=f"{option_name}=irrelevant")
+        static_proxy.connect(user="pinocchio", password="bad", options="project=irrelevant")
    text = str(exprinfo.value).strip()
    assert text.endswith("password authentication failed for user 'pinocchio'")

    # Finally, check that the user can connect
-    with static_proxy.connect(
-        user="pinocchio", password="magic", options=f"{option_name}=irrelevant"
-    ):
+    with static_proxy.connect(user="pinocchio", password="magic", options="project=irrelevant"):
        pass


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_forward_params_to_client(static_proxy: NeonProxy, option_name: str):
+def test_forward_params_to_client(static_proxy: NeonProxy):
    """
    Check that we forward all necessary PostgreSQL server params to client.
    """
@@ -140,7 +130,7 @@ def test_forward_params_to_client(static_proxy: NeonProxy, option_name: str):
        where name = any(%s)
    """

-    with static_proxy.connect(options=f"{option_name}=irrelevant") as conn:
+    with static_proxy.connect(options="project=irrelevant") as conn:
        with conn.cursor() as cur:
            cur.execute(query, (reported_params_subset,))
            for name, value in cur.fetchall():
@@ -148,18 +138,17 @@ def test_forward_params_to_client(static_proxy: NeonProxy, option_name: str):
                assert conn.get_parameter_status(name) == value


-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
@pytest.mark.timeout(5)
-def test_close_on_connections_exit(static_proxy: NeonProxy, option_name: str):
+def test_close_on_connections_exit(static_proxy: NeonProxy):
    # Open two connections, send SIGTERM, then ensure that proxy doesn't exit
    # until after connections close.
-    with static_proxy.connect(options=f"{option_name}=irrelevant"), static_proxy.connect(
-        options=f"{option_name}=irrelevant"
+    with static_proxy.connect(options="project=irrelevant"), static_proxy.connect(
+        options="project=irrelevant"
    ):
        static_proxy.terminate()
        with pytest.raises(subprocess.TimeoutExpired):
            static_proxy.wait_for_exit(timeout=2)
        # Ensure we don't accept any more connections
        with pytest.raises(psycopg2.OperationalError):
-            static_proxy.connect(options=f"{option_name}=irrelevant")
+            static_proxy.connect(options="project=irrelevant")
    static_proxy.wait_for_exit()
Author	SHA1	Message	Date
Konstantin Knizhnik	aaf8617fcc	Log detection of duplicate layer	2023-05-16 09:53:45 +03:00
Konstantin Knizhnik	b14e70fb5c	Make clippy happy	2023-05-15 22:59:15 +03:00
Konstantin Knizhnik	64e01c5b87	Add removed BufferedHistoricLayerCoverage.contains method	2023-05-15 18:24:35 +03:00
Konstantin Knizhnik	843e82357f	Fix rust formatting issues	2023-05-15 17:34:30 +03:00
Konstantin Knizhnik	51256890a0	Add comment explainign potential risks about duplicates prevention	2023-05-15 17:25:00 +03:00
Heikki Linnakangas	6837356cc1	Fix the duplicate key problem during compaction. Before finishing the delta file, and possibly overwriting an old perfectly valid file, check if an identical file already exists in the layer map. This is an alternative for https://github.com/neondatabase/neon/pull/4094. Test case is copied from that PR.	2023-05-15 16:58:08 +03:00