fixes

.github/workflows/force-test-extensions-upgrade.yml

@@ -55,7 +55,7 @@ jobs:
           echo tag=${tag} >> ${GITHUB_OUTPUT}
 
       - name: Test extension upgrade
-        timeout-minutes: 60
+        timeout-minutes: 20
         env:
           NEW_COMPUTE_TAG: latest
           OLD_COMPUTE_TAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}

Cargo.lock

@@ -2248,17 +2248,6 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
-[[package]]
-name = "flag-bearer"
-version = "0.1.0-rc.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e8feaa1b7a5ad6e6dd7791d42d36c1a25004f1c25eae9ab7b904c864109d8260"
-dependencies = [
- "parking_lot 0.12.1",
- "pin-list",
- "pin-project-lite",
-]
-
 [[package]]
 name = "flagset"
 version = "0.4.6"
@@ -4565,16 +4554,6 @@ dependencies = [
  "siphasher",
 ]
 
-[[package]]
-name = "pin-list"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a3c0987a7464afc0d593f13429732ef87e9a6c8e7909a1a22faeff7e1d2159d"
-dependencies = [
- "pin-project-lite",
- "pinned-aliasable",
-]
-
 [[package]]
 name = "pin-project"
 version = "1.1.9"
@@ -4607,12 +4586,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
-[[package]]
-name = "pinned-aliasable"
-version = "0.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d0f9ae89bf0ed03b69ac1f3f7ea2e6e09b4fa5448011df2e67d581c2b850b7b"
-
 [[package]]
 name = "pkcs1"
 version = "0.7.5"
@@ -5106,7 +5079,6 @@ dependencies = [
  "ed25519-dalek",
  "env_logger",
  "fallible-iterator",
- "flag-bearer",
  "flate2",
  "framed-websockets",
  "futures",

libs/safekeeper_api/src/models.rs

@@ -227,6 +227,8 @@ pub struct TimelineDeleteResult {
     pub dir_existed: bool,
 }
 
+pub type TenantDeleteResult = std::collections::HashMap<String, TimelineDeleteResult>;
+
 fn lsn_invalid() -> Lsn {
     Lsn::INVALID
 }

pageserver/src/http/routes.rs

@@ -3774,7 +3774,7 @@ pub fn make_router(
         )
         .put(
             "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/mark_invisible",
-            |r| api_handler( r, timeline_mark_invisible_handler),
+            |r| testing_api_handler("mark timeline invisible", r, timeline_mark_invisible_handler),
         )
         .put(
             "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/checkpoint",

pageserver/src/tenant.rs

@@ -11526,255 +11526,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[cfg(feature = "testing")]
-    #[tokio::test]
-    async fn test_synthetic_size_calculation_with_invisible_branches() -> anyhow::Result<()> {
-        use pageserver_api::models::TimelineVisibilityState;
-
-        use crate::tenant::size::gather_inputs;
-
-        let tenant_conf = pageserver_api::models::TenantConfig {
-            // Ensure that we don't compute gc_cutoffs (which needs reading the layer files)
-            pitr_interval: Some(Duration::ZERO),
-            ..Default::default()
-        };
-        let harness = TenantHarness::create_custom(
-            "test_synthetic_size_calculation_with_invisible_branches",
-            tenant_conf,
-            TenantId::generate(),
-            ShardIdentity::unsharded(),
-            Generation::new(0xdeadbeef),
-        )
-        .await?;
-        let (tenant, ctx) = harness.load().await;
-        let main_tline = tenant
-            .create_test_timeline_with_layers(
-                TIMELINE_ID,
-                Lsn(0x10),
-                DEFAULT_PG_VERSION,
-                &ctx,
-                vec![],
-                vec![],
-                vec![],
-                Lsn(0x100),
-            )
-            .await?;
-
-        let snapshot1 = TimelineId::from_array(hex!("11223344556677881122334455667790"));
-        tenant
-            .branch_timeline_test_with_layers(
-                &main_tline,
-                snapshot1,
-                Some(Lsn(0x20)),
-                &ctx,
-                vec![],
-                vec![],
-                Lsn(0x50),
-            )
-            .await?;
-        let snapshot2 = TimelineId::from_array(hex!("11223344556677881122334455667791"));
-        tenant
-            .branch_timeline_test_with_layers(
-                &main_tline,
-                snapshot2,
-                Some(Lsn(0x30)),
-                &ctx,
-                vec![],
-                vec![],
-                Lsn(0x50),
-            )
-            .await?;
-        let snapshot3 = TimelineId::from_array(hex!("11223344556677881122334455667792"));
-        tenant
-            .branch_timeline_test_with_layers(
-                &main_tline,
-                snapshot3,
-                Some(Lsn(0x40)),
-                &ctx,
-                vec![],
-                vec![],
-                Lsn(0x50),
-            )
-            .await?;
-        let limit = Arc::new(Semaphore::new(1));
-        let max_retention_period = None;
-        let mut logical_size_cache = HashMap::new();
-        let cause = LogicalSizeCalculationCause::EvictionTaskImitation;
-        let cancel = CancellationToken::new();
-
-        let inputs = gather_inputs(
-            &tenant,
-            &limit,
-            max_retention_period,
-            &mut logical_size_cache,
-            cause,
-            &cancel,
-            &ctx,
-        )
-        .instrument(info_span!(
-            "gather_inputs",
-            tenant_id = "unknown",
-            shard_id = "unknown",
-        ))
-        .await?;
-        use crate::tenant::size::{LsnKind, ModelInputs, SegmentMeta};
-        use LsnKind::*;
-        use tenant_size_model::Segment;
-        let ModelInputs { mut segments, .. } = inputs;
-        segments.retain(|s| s.timeline_id == TIMELINE_ID);
-        for segment in segments.iter_mut() {
-            segment.segment.parent = None; // We don't care about the parent for the test
-            segment.segment.size = None; // We don't care about the size for the test
-        }
-        assert_eq!(
-            segments,
-            [
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x10,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchStart,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x20,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x30,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x40,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x100,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: GcCutOff,
-                }, // we need to retain everything above the last branch point
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x100,
-                        size: None,
-                        needed: true,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchEnd,
-                },
-            ]
-        );
-
-        main_tline
-            .remote_client
-            .schedule_index_upload_for_timeline_invisible_state(
-                TimelineVisibilityState::Invisible,
-            )?;
-        main_tline.remote_client.wait_completion().await?;
-        let inputs = gather_inputs(
-            &tenant,
-            &limit,
-            max_retention_period,
-            &mut logical_size_cache,
-            cause,
-            &cancel,
-            &ctx,
-        )
-        .instrument(info_span!(
-            "gather_inputs",
-            tenant_id = "unknown",
-            shard_id = "unknown",
-        ))
-        .await?;
-        let ModelInputs { mut segments, .. } = inputs;
-        segments.retain(|s| s.timeline_id == TIMELINE_ID);
-        for segment in segments.iter_mut() {
-            segment.segment.parent = None; // We don't care about the parent for the test
-            segment.segment.size = None; // We don't care about the size for the test
-        }
-        assert_eq!(
-            segments,
-            [
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x10,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchStart,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x20,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x30,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x40,
-                        size: None,
-                        needed: false,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchPoint,
-                },
-                SegmentMeta {
-                    segment: Segment {
-                        parent: None,
-                        lsn: 0x40, // Branch end LSN == last branch point LSN
-                        size: None,
-                        needed: true,
-                    },
-                    timeline_id: TIMELINE_ID,
-                    kind: BranchEnd,
-                },
-            ]
-        );
-        Ok(())
-    }
 }

pageserver/src/tenant/size.rs

@@ -33,7 +33,7 @@ pub struct ModelInputs {
 }
 
 /// A [`Segment`], with some extra information for display purposes
-#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
+#[derive(Debug, serde::Serialize, serde::Deserialize)]
 pub struct SegmentMeta {
     pub segment: Segment,
     pub timeline_id: TimelineId,
@@ -248,8 +248,6 @@ pub(super) async fn gather_inputs(
             None
         };
 
-        let branch_is_invisible = timeline.is_invisible() == Some(true);
-
         let lease_points = gc_info
             .leases
             .keys()
@@ -273,10 +271,7 @@ pub(super) async fn gather_inputs(
             .map(|(lsn, _child_id, _is_offloaded)| (lsn, LsnKind::BranchPoint))
             .collect::<Vec<_>>();
 
-        if !branch_is_invisible {
-            // Do not count lease points for invisible branches.
-            lsns.extend(lease_points.iter().map(|&lsn| (lsn, LsnKind::LeasePoint)));
-        }
+        lsns.extend(lease_points.iter().map(|&lsn| (lsn, LsnKind::LeasePoint)));
 
         drop(gc_info);
 
@@ -292,9 +287,7 @@ pub(super) async fn gather_inputs(
 
         // Add a point for the PITR cutoff
         let branch_start_needed = next_pitr_cutoff <= branch_start_lsn;
-        if !branch_start_needed && !branch_is_invisible {
-            // Only add the GcCutOff point when the timeline is visible; otherwise, do not compute the size for the LSN
-            // range from the last branch point to the latest data.
+        if !branch_start_needed {
             lsns.push((next_pitr_cutoff, LsnKind::GcCutOff));
         }
 
@@ -380,19 +373,11 @@ pub(super) async fn gather_inputs(
             }
         }
 
-        let branch_end_lsn = if branch_is_invisible {
-            // If the branch is invisible, the branch end is the last requested LSN (likely a branch cutoff point).
-            segments.last().unwrap().segment.lsn
-        } else {
-            // Otherwise, the branch end is the last record LSN.
-            last_record_lsn.0
-        };
-
         // Current end of the timeline
         segments.push(SegmentMeta {
             segment: Segment {
                 parent: Some(parent),
-                lsn: branch_end_lsn,
+                lsn: last_record_lsn.0,
                 size: None, // Filled in later, if necessary
                 needed: true,
             },
@@ -624,7 +609,6 @@ async fn calculate_logical_size(
     Ok(TimelineAtLsnSizeResult(timeline, lsn, size_res))
 }
 
-#[cfg(test)]
 #[test]
 fn verify_size_for_multiple_branches() {
     // this is generated from integration test test_tenant_size_with_multiple_branches, but this way
@@ -782,7 +766,6 @@ fn verify_size_for_multiple_branches() {
     assert_eq!(inputs.calculate(), 37_851_408);
 }
 
-#[cfg(test)]
 #[test]
 fn verify_size_for_one_branch() {
     let doc = r#"

pageserver/src/tenant/timeline.rs

@@ -2215,10 +2215,6 @@ impl Timeline {
         self.remote_client.is_archived()
     }
 
-    pub(crate) fn is_invisible(&self) -> Option<bool> {
-        self.remote_client.is_invisible()
-    }
-
     pub(crate) fn is_stopping(&self) -> bool {
         self.current_state() == TimelineState::Stopping
     }

proxy/Cargo.toml

@@ -103,7 +103,6 @@ uuid.workspace = true
 x509-cert.workspace = true
 redis.workspace = true
 zerocopy.workspace = true
-flag-bearer = { version = "0.1.0-rc.4" }
 
 # jwt stuff
 jose-jwa = "0.1.2"

proxy/src/rate_limiter/limit_algorithm.rs

@@ -1,8 +1,10 @@
 //! Algorithms for controlling concurrency limits.
+use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use flag_bearer::{Semaphore, SemaphoreState, Uncloseable};
+use parking_lot::Mutex;
+use tokio::sync::Notify;
 use tokio::time::Instant;
 use tokio::time::error::Elapsed;
 
@@ -63,13 +65,23 @@ pub struct RateLimiterConfig {
     pub(crate) initial_limit: usize,
 }
 
-pub(crate) struct LimiterInner<L: ?Sized> {
-    limit: usize,
-    in_flight: usize,
-    alg: L,
+impl RateLimiterConfig {
+    pub(crate) fn create_rate_limit_algorithm(self) -> Box<dyn LimitAlgorithm> {
+        match self.algorithm {
+            RateLimitAlgorithm::Fixed => Box::new(Fixed),
+            RateLimitAlgorithm::Aimd { conf } => Box::new(conf),
+        }
+    }
 }
 
-impl<L: ?Sized + LimitAlgorithm> LimiterInner<L> {
+pub(crate) struct LimiterInner {
+    alg: Box<dyn LimitAlgorithm>,
+    available: usize,
+    limit: usize,
+    in_flight: usize,
+}
+
+impl LimiterInner {
     fn update_limit(&mut self, latency: Duration, outcome: Option<Outcome>) {
         if let Some(outcome) = outcome {
             let sample = Sample {
@@ -80,26 +92,21 @@ impl<L: ?Sized + LimitAlgorithm> LimiterInner<L> {
             self.limit = self.alg.update(self.limit, sample);
         }
     }
-}
 
-impl<L: ?Sized> SemaphoreState for LimiterInner<L> {
-    type Params = ();
-    type Permit = ();
-    type Closeable = Uncloseable;
-
-    fn acquire(&mut self, p: Self::Params) -> Result<Self::Permit, Self::Params> {
-        if self.in_flight < self.limit {
+    fn take(&mut self, ready: &Notify) -> Option<()> {
+        if self.available >= 1 {
+            self.available -= 1;
             self.in_flight += 1;
 
-            Ok(p)
+            // tell the next in the queue that there is a permit ready
+            if self.available >= 1 {
+                ready.notify_one();
+            }
+            Some(())
         } else {
-            Err(p)
+            None
         }
     }
-
-    fn release(&mut self, _p: Self::Permit) {
-        self.in_flight -= 1;
-    }
 }
 
 /// Limits the number of concurrent jobs.
@@ -109,9 +116,11 @@ impl<L: ?Sized> SemaphoreState for LimiterInner<L> {
 ///
 /// The limit will be automatically adjusted based on observed latency (delay) and/or failures
 /// caused by overload (loss).
-pub(crate) struct DynamicLimiter<L: ?Sized = dyn LimitAlgorithm> {
-    disabled: bool,
-    sem: Semaphore<LimiterInner<L>>,
+pub(crate) struct DynamicLimiter {
+    config: RateLimiterConfig,
+    inner: Mutex<LimiterInner>,
+    // to notify when a token is available
+    ready: Notify,
 }
 
 /// A concurrency token, required to run a job.
@@ -131,27 +140,22 @@ struct LimiterState {
     limit: usize,
 }
 
-impl<L> DynamicLimiter<L> {
-    pub(crate) fn new_inner(initial_limit: usize, alg: L) -> Arc<Self> {
-        Arc::new(Self {
-            disabled: initial_limit == 0,
-            sem: Semaphore::new_fifo(LimiterInner {
-                limit: initial_limit,
-                in_flight: 0,
-                alg,
-            }),
-        })
-    }
-}
-
 impl DynamicLimiter {
     /// Create a limiter with a given limit control algorithm.
     pub(crate) fn new(config: RateLimiterConfig) -> Arc<Self> {
-        let initial_limit = config.initial_limit;
-        match config.algorithm {
-            RateLimitAlgorithm::Fixed => DynamicLimiter::new_inner(initial_limit, Fixed),
-            RateLimitAlgorithm::Aimd { conf } => DynamicLimiter::new_inner(initial_limit, conf),
-        }
+        let ready = Notify::new();
+        ready.notify_one();
+
+        Arc::new(Self {
+            inner: Mutex::new(LimiterInner {
+                alg: config.create_rate_limit_algorithm(),
+                available: config.initial_limit,
+                limit: config.initial_limit,
+                in_flight: 0,
+            }),
+            ready,
+            config,
+        })
     }
 
     /// Try to acquire a concurrency [Token], waiting for `duration` if there are none available.
@@ -159,21 +163,27 @@ impl DynamicLimiter {
         self: &Arc<Self>,
         duration: Duration,
     ) -> Result<Token, Elapsed> {
-        tokio::time::timeout(duration, self.acquire()).await
+        tokio::time::timeout(duration, self.acquire()).await?
     }
 
     /// Try to acquire a concurrency [Token].
-    async fn acquire(self: &Arc<Self>) -> Token {
-        if self.disabled {
+    async fn acquire(self: &Arc<Self>) -> Result<Token, Elapsed> {
+        if self.config.initial_limit == 0 {
             // If the rate limiter is disabled, we can always acquire a token.
-            Token::disabled()
+            Ok(Token::disabled())
         } else {
-            match self.sem.acquire(()).await {
-                Err(close) => close.never(),
-                Ok(permit) => {
-                    permit.take();
-                    Token::new(self.clone())
+            let mut notified = pin!(self.ready.notified());
+            let mut ready = notified.as_mut().enable();
+            loop {
+                if ready {
+                    let mut inner = self.inner.lock();
+                    if inner.take(&self.ready).is_some() {
+                        break Ok(Token::new(self.clone()));
+                    }
+                    notified.set(self.ready.notified());
                 }
+                notified.as_mut().await;
+                ready = true;
             }
         }
     }
@@ -190,20 +200,27 @@ impl DynamicLimiter {
         } else {
             tracing::debug!("outcome is {:?}", outcome);
         }
-        if self.disabled {
+        if self.config.initial_limit == 0 {
             return;
         }
 
-        self.sem.with_state(|s| {
-            s.update_limit(start.elapsed(), outcome);
-            s.release(());
-        });
+        let mut inner = self.inner.lock();
+
+        inner.update_limit(start.elapsed(), outcome);
+
+        inner.in_flight -= 1;
+        if inner.in_flight < inner.limit {
+            inner.available = inner.limit - inner.in_flight;
+            // At least 1 permit is now available
+            self.ready.notify_one();
+        }
     }
 
     /// The current state of the limiter.
     #[cfg(test)]
     fn state(&self) -> LimiterState {
-        self.sem.with_state(|s| LimiterState { limit: s.limit })
+        let inner = self.inner.lock();
+        LimiterState { limit: inner.limit }
     }
 }
 

safekeeper/client/src/mgmt_api.rs

@@ -115,13 +115,17 @@ impl Client {
             "{}/v1/tenant/{}/timeline/{}",
             self.mgmt_api_endpoint, tenant_id, timeline_id
         );
-        let resp = self.request(Method::DELETE, &uri, ()).await?;
+        let resp = self
+            .request_maybe_body(Method::DELETE, &uri, None::<()>)
+            .await?;
         resp.json().await.map_err(Error::ReceiveBody)
     }
 
-    pub async fn delete_tenant(&self, tenant_id: TenantId) -> Result<models::TimelineDeleteResult> {
+    pub async fn delete_tenant(&self, tenant_id: TenantId) -> Result<models::TenantDeleteResult> {
         let uri = format!("{}/v1/tenant/{}", self.mgmt_api_endpoint, tenant_id);
-        let resp = self.request(Method::DELETE, &uri, ()).await?;
+        let resp = self
+            .request_maybe_body(Method::DELETE, &uri, None::<()>)
+            .await?;
         resp.json().await.map_err(Error::ReceiveBody)
     }
 
@@ -197,6 +201,16 @@ impl Client {
         method: Method,
         uri: U,
         body: B,
+    ) -> Result<reqwest::Response> {
+        self.request_maybe_body(method, uri, Some(body)).await
+    }
+
+    /// Send the request and check that the status code is good, with an optional body.
+    async fn request_maybe_body<B: serde::Serialize, U: reqwest::IntoUrl>(
+        &self,
+        method: Method,
+        uri: U,
+        body: Option<B>,
     ) -> Result<reqwest::Response> {
         let res = self.request_noerror(method, uri, body).await?;
         let response = res.error_from_body().await?;
@@ -208,12 +222,15 @@ impl Client {
         &self,
         method: Method,
         uri: U,
-        body: B,
+        body: Option<B>,
     ) -> Result<reqwest::Response> {
         let mut req = self.client.request(method, uri);
         if let Some(value) = &self.authorization_header {
             req = req.header(reqwest::header::AUTHORIZATION, value.get_contents())
         }
-        req.json(&body).send().await.map_err(Error::ReceiveBody)
+        if let Some(body) = body {
+            req = req.json(&body);
+        }
+        req.send().await.map_err(Error::ReceiveBody)
     }
 }

safekeeper/src/http/routes.rs

@@ -16,9 +16,9 @@ use http_utils::{RequestExt, RouterBuilder};
 use hyper::{Body, Request, Response, StatusCode};
 use postgres_ffi::WAL_SEGMENT_SIZE;
 use safekeeper_api::models::{
-    AcceptorStateStatus, PullTimelineRequest, SafekeeperStatus, SkTimelineInfo, TermSwitchApiEntry,
-    TimelineCopyRequest, TimelineCreateRequest, TimelineDeleteResult, TimelineStatus,
-    TimelineTermBumpRequest,
+    AcceptorStateStatus, PullTimelineRequest, SafekeeperStatus, SkTimelineInfo, TenantDeleteResult,
+    TermSwitchApiEntry, TimelineCopyRequest, TimelineCreateRequest, TimelineDeleteResult,
+    TimelineStatus, TimelineTermBumpRequest,
 };
 use safekeeper_api::{ServerInfo, membership, models};
 use storage_broker::proto::{SafekeeperTimelineInfo, TenantTimelineId as ProtoTenantTimelineId};
@@ -83,13 +83,11 @@ async fn tenant_delete_handler(mut request: Request<Body>) -> Result<Response<Bo
         .delete_all_for_tenant(&tenant_id, action)
         .await
         .map_err(ApiError::InternalServerError)?;
-    json_response(
-        StatusCode::OK,
-        delete_info
-            .iter()
-            .map(|(ttid, resp)| (format!("{}", ttid.timeline_id), *resp))
-            .collect::<HashMap<String, TimelineDeleteResult>>(),
-    )
+    let response_body: TenantDeleteResult = delete_info
+        .iter()
+        .map(|(ttid, resp)| (format!("{}", ttid.timeline_id), *resp))
+        .collect::<HashMap<String, TimelineDeleteResult>>();
+    json_response(StatusCode::OK, response_body)
 }
 
 async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {

storage_controller/src/persistence.rs

@@ -1524,25 +1524,14 @@ impl Persistence {
     /// Load pending operations from db.
     pub(crate) async fn list_pending_ops(
         &self,
-        filter_for_sk: Option<NodeId>,
     ) -> DatabaseResult<Vec<TimelinePendingOpPersistence>> {
         use crate::schema::safekeeper_timeline_pending_ops::dsl;
 
-        const FILTER_VAL_1: i64 = 1;
-        const FILTER_VAL_2: i64 = 2;
-        let filter_opt = filter_for_sk.map(|id| id.0 as i64);
         let timeline_from_db = self
             .with_measured_conn(DatabaseOperation::ListTimelineReconcile, move |conn| {
                 Box::pin(async move {
                     let from_db: Vec<TimelinePendingOpPersistence> =
-                        dsl::safekeeper_timeline_pending_ops
-                            .filter(
-                                dsl::sk_id
-                                    .eq(filter_opt.unwrap_or(FILTER_VAL_1))
-                                    .and(dsl::sk_id.eq(filter_opt.unwrap_or(FILTER_VAL_2))),
-                            )
-                            .load(conn)
-                            .await?;
+                        dsl::safekeeper_timeline_pending_ops.load(conn).await?;
                     Ok(from_db)
                 })
             })

storage_controller/src/safekeeper_client.rs

@@ -101,7 +101,7 @@ impl SafekeeperClient {
     pub(crate) async fn delete_tenant(
         &self,
         tenant_id: TenantId,
-    ) -> Result<models::TimelineDeleteResult> {
+    ) -> Result<models::TenantDeleteResult> {
         measured_request!(
             "delete_tenant",
             crate::metrics::Method::Delete,

storage_controller/src/service/chaos_injector.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 
 use pageserver_api::controller_api::ShardSchedulingPolicy;
 use rand::seq::SliceRandom;
-use rand::{Rng, thread_rng};
+use rand::thread_rng;
 use tokio_util::sync::CancellationToken;
 use utils::id::NodeId;
 use utils::shard::TenantShardId;
@@ -64,22 +64,17 @@ impl ChaosInjector {
         let mut interval = tokio::time::interval(self.interval);
         #[derive(Debug)]
         enum ChaosEvent {
-            MigrationsToSecondary,
-            ForceKillController,
-            GracefulMigrationsAnywhere,
+            ShuffleTenant,
+            ForceKill,
         }
         loop {
             let cron_interval = self.get_cron_interval_sleep_future();
             let chaos_type = tokio::select! {
                 _ = interval.tick() => {
-                    if thread_rng().gen_bool(0.5) {
-                        ChaosEvent::MigrationsToSecondary
-                    } else {
-                        ChaosEvent::GracefulMigrationsAnywhere
-                    }
+                    ChaosEvent::ShuffleTenant
                 }
                 Some(_) = maybe_sleep(cron_interval) => {
-                    ChaosEvent::ForceKillController
+                    ChaosEvent::ForceKill
                 }
                 _ = cancel.cancelled() => {
                     tracing::info!("Shutting down");
@@ -88,29 +83,16 @@ impl ChaosInjector {
             };
             tracing::info!("Chaos iteration: {chaos_type:?}...");
             match chaos_type {
-                ChaosEvent::MigrationsToSecondary => {
-                    self.inject_migrations_to_secondary();
+                ChaosEvent::ShuffleTenant => {
+                    self.inject_chaos().await;
                 }
-                ChaosEvent::GracefulMigrationsAnywhere => {
-                    self.inject_graceful_migrations_anywhere();
-                }
-                ChaosEvent::ForceKillController => {
+                ChaosEvent::ForceKill => {
                     self.force_kill().await;
                 }
             }
         }
     }
 
-    fn is_shard_eligible_for_chaos(&self, shard: &TenantShard) -> bool {
-        // - Skip non-active scheduling policies, so that a shard with a policy like Pause can
-        //   be pinned without being disrupted by us.
-        // - Skip shards doing a graceful migration already, so that we allow these to run to
-        //   completion rather than only exercising the first part and then cancelling with
-        //   some other chaos.
-        !matches!(shard.get_scheduling_policy(), ShardSchedulingPolicy::Active)
-            && shard.get_preferred_node().is_none()
-    }
-
     /// If a shard has a secondary and attached location, then re-assign the secondary to be
     /// attached and the attached to be secondary.
     ///
@@ -126,7 +108,13 @@ impl ChaosInjector {
             .get_mut(&tenant_shard_id)
             .expect("Held lock between choosing ID and this get");
 
-        if !self.is_shard_eligible_for_chaos(shard) {
+        if !matches!(shard.get_scheduling_policy(), ShardSchedulingPolicy::Active) {
+            // Skip non-active scheduling policies, so that a shard with a policy like Pause can
+            // be pinned without being disrupted by us.
+            tracing::info!(
+                "Skipping shard {tenant_shard_id}: scheduling policy is {:?}",
+                shard.get_scheduling_policy()
+            );
             return;
         }
 
@@ -164,77 +152,7 @@ impl ChaosInjector {
         std::process::exit(1);
     }
 
-    // Unlike [`Self::inject_migrations_to_secondary`], this function will not only cut over to secondary, it
-    // will migrate a tenant to a random node in its home AZ using a graceful migration of the same type
-    // that my be initiated by an API caller using prewarm=true.
-    //
-    // This is a much more expensive operation in terms of I/O and time, as we will fully warm up
-    // some new location in order to migrate the tenant there.  For that reason we do far fewer of these.
-    fn inject_graceful_migrations_anywhere(&mut self) {
-        let batch_size = 1;
-        let mut inner = self.service.inner.write().unwrap();
-        let (nodes, tenants, _scheduler) = inner.parts_mut();
-
-        let mut candidates = tenants
-            .values_mut()
-            .filter(|shard| self.is_shard_eligible_for_chaos(shard))
-            .collect::<Vec<_>>();
-
-        tracing::info!(
-            "Injecting chaos: found {} candidates for graceful migrations anywhere",
-            candidates.len()
-        );
-
-        let mut victims: Vec<&mut TenantShard> = Vec::new();
-
-        // Pick our victims: use a hand-rolled loop rather than choose_multiple() because we want
-        // to take the mutable refs from our candidates rather than ref'ing them.
-        while !candidates.is_empty() && victims.len() < batch_size {
-            let i = thread_rng().gen_range(0..candidates.len());
-            victims.push(candidates.swap_remove(i));
-        }
-
-        for victim in victims.into_iter() {
-            // Find a node in the same AZ as the shard, or if the shard has no AZ preference, which
-            // is not where they are currently attached.
-            let candidate_nodes = nodes
-                .values()
-                .filter(|node| {
-                    if let Some(preferred_az) = victim.preferred_az() {
-                        node.get_availability_zone_id() == preferred_az
-                    } else if let Some(attached) = *victim.intent.get_attached() {
-                        node.get_id() != attached
-                    } else {
-                        true
-                    }
-                })
-                .collect::<Vec<_>>();
-
-            let Some(victim_node) = candidate_nodes.choose(&mut thread_rng()) else {
-                // This can happen if e.g. we are in a small region with only one pageserver per AZ.
-                tracing::info!(
-                    "no candidate nodes found for migrating shard {tenant_shard_id} within its home AZ",
-                    tenant_shard_id = victim.tenant_shard_id
-                );
-                continue;
-            };
-
-            // This doesn't change intent immediately: next iteration of Service::optimize_all should do that.  We avoid
-            // doing it here because applying optimizations requires dropping lock to do some async work to check the optimisation
-            // is valid given remote state, and it would be a shame to duplicate that dance here.
-            tracing::info!(
-                "Injecting chaos: migrate {} to {}",
-                victim.tenant_shard_id,
-                victim_node
-            );
-            victim.set_preferred_node(Some(victim_node.get_id()));
-        }
-    }
-
-    /// Migrations of attached locations to their secondary location.  This exercises reconciliation in general,
-    /// live migration in particular, and the pageserver code for cleanly shutting down and starting up tenants
-    /// during such migrations.
-    fn inject_migrations_to_secondary(&mut self) {
+    async fn inject_chaos(&mut self) {
         // Pick some shards to interfere with
         let batch_size = 128;
         let mut inner = self.service.inner.write().unwrap();

storage_controller/src/service/safekeeper_reconciler.rs

@@ -35,6 +35,10 @@ impl SafekeeperReconcilers {
         service: &Arc<Service>,
         reqs: Vec<ScheduleRequest>,
     ) {
+        tracing::info!(
+            "Scheduling {} pending safekeeper ops loaded from db",
+            reqs.len()
+        );
         for req in reqs {
             self.schedule_request(service, req);
         }
@@ -74,7 +78,7 @@ pub(crate) async fn load_schedule_requests(
     service: &Arc<Service>,
     safekeepers: &HashMap<NodeId, Safekeeper>,
 ) -> anyhow::Result<Vec<ScheduleRequest>> {
-    let pending_ops = service.persistence.list_pending_ops(None).await?;
+    let pending_ops = service.persistence.list_pending_ops().await?;
     let mut res = Vec::with_capacity(pending_ops.len());
     for op_persist in pending_ops {
         let node_id = NodeId(op_persist.sk_id as u64);
@@ -232,12 +236,14 @@ impl SafekeeperReconciler {
             let kind = req.kind;
             let tenant_id = req.tenant_id;
             let timeline_id = req.timeline_id;
+            let node_id = req.safekeeper.skp.id;
             self.reconcile_one(req, req_cancel)
                 .instrument(tracing::info_span!(
                     "reconcile_one",
                     ?kind,
                     %tenant_id,
-                    ?timeline_id
+                    ?timeline_id,
+                    %node_id,
                 ))
                 .await;
         }

test_runner/fixtures/utils.py

@@ -724,20 +724,3 @@ def skip_on_ci(reason: str):
         os.getenv("CI", "false") == "true",
         reason=reason,
     )
-
-
-def shared_buffers_for_max_cu(max_cu: float) -> str:
-    """
-    Returns the string value of shared_buffers for the given max CU.
-    Use shared_buffers size like in production for max CU compute.
-    See https://github.com/neondatabase/cloud/blob/877e33b4289a471b8f0a35c84009846358f3e5a3/goapp/controlplane/internal/pkg/compute/computespec/pg_settings.go#L405
-
-    e.g. // 2 CU: 225mb; 4 CU: 450mb; 8 CU: 900mb
-    """
-    ramBytes = int(4096 * max_cu * 1024 * 1024)
-    maxConnections = max(100, min(int(ramBytes / 9531392), 4000))
-    maxWorkerProcesses = 12 + int(max_cu * 2)
-    maxBackends = 1 + maxConnections + maxWorkerProcesses
-    sharedBuffersMb = int(max(128, (1023 + maxBackends * 256) / 1024))
-    sharedBuffers = int(sharedBuffersMb * 1024 / 8)
-    return str(sharedBuffers)

test_runner/performance/test_bulk_update.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import pytest
 from fixtures.neon_fixtures import NeonEnvBuilder, wait_for_last_flush_lsn
-from fixtures.utils import shared_buffers_for_max_cu
 
 
 #
@@ -21,10 +20,7 @@ def test_bulk_update(neon_env_builder: NeonEnvBuilder, zenbenchmark, fillfactor)
 
     timeline_id = env.create_branch("test_bulk_update")
     tenant_id = env.initial_tenant
-    # use shared_buffers size like in production for 8 CU compute
-    endpoint = env.endpoints.create_start(
-        "test_bulk_update", config_lines=[f"shared_buffers={shared_buffers_for_max_cu(8.0)}"]
-    )
+    endpoint = env.endpoints.create_start("test_bulk_update")
     cur = endpoint.connect().cursor()
     cur.execute("set statement_timeout=0")
 

test_runner/performance/test_ingest_insert_bulk.py

@@ -17,10 +17,9 @@ from fixtures.pageserver.utils import (
     wait_for_upload_queue_empty,
 )
 from fixtures.remote_storage import s3_storage
-from fixtures.utils import shared_buffers_for_max_cu
 
 
-@pytest.mark.timeout(1800)
+@pytest.mark.timeout(900)
 @pytest.mark.parametrize("size", [8, 1024, 8192])
 @pytest.mark.parametrize("s3", [True, False], ids=["s3", "local"])
 @pytest.mark.parametrize("backpressure", [True, False], ids=["backpressure", "nobackpressure"])
@@ -61,8 +60,6 @@ def test_ingest_insert_bulk(
             f"fsync = {fsync}",
             "max_replication_apply_lag = 0",
             f"max_replication_flush_lag = {'10GB' if backpressure else '0'}",
-            # use shared_buffers size like in production for 8 CU compute
-            f"shared_buffers={shared_buffers_for_max_cu(8.0)}",
             # NB: neon_local defaults to 15MB, which is too slow -- production uses 500MB.
             f"max_replication_write_lag = {'500MB' if backpressure else '0'}",
         ],

test_runner/performance/test_ingest_logical_message.py

@@ -12,7 +12,7 @@ from fixtures.neon_fixtures import (
 from fixtures.pageserver.utils import wait_for_last_record_lsn
 
 
-@pytest.mark.timeout(1200)
+@pytest.mark.timeout(600)
 @pytest.mark.parametrize("size", [1024, 8192, 131072])
 @pytest.mark.parametrize("fsync", [True, False], ids=["fsync", "nofsync"])
 def test_ingest_logical_message(

test_runner/performance/test_parallel_copy.py

@@ -7,8 +7,6 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from fixtures.neon_fixtures import Endpoint, NeonEnv
 
-from fixtures.utils import shared_buffers_for_max_cu
-
 
 async def repeat_bytes(buf, repetitions: int):
     for _ in range(repetitions):
@@ -47,10 +45,7 @@ async def parallel_load_same_table(endpoint: Endpoint, n_parallel: int):
 # Load data into one table with COPY TO from 5 parallel connections
 def test_parallel_copy(neon_simple_env: NeonEnv, n_parallel=5):
     env = neon_simple_env
-    # use shared_buffers size like in production for 8 CU compute
-    endpoint = env.endpoints.create_start(
-        "main", config_lines=[f"shared_buffers={shared_buffers_for_max_cu(8.0)}"]
-    )
+    endpoint = env.endpoints.create_start("main")
 
     # Create test table
     conn = endpoint.connect()

test_runner/performance/test_perf_many_relations.py

@@ -6,7 +6,6 @@ from fixtures.benchmark_fixture import NeonBenchmarker
 from fixtures.compare_fixtures import RemoteCompare
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder
-from fixtures.utils import shared_buffers_for_max_cu
 
 
 def get_num_relations(default: int = 1000) -> list[int]:
@@ -79,8 +78,7 @@ def test_perf_simple_many_relations_reldir_v2(
     ep = env.endpoints.create_start(
         "main",
         config_lines=[
-            # use shared_buffers size like in production for 8 CU compute
-            f"shared_buffers={shared_buffers_for_max_cu(8.0)}",
+            "shared_buffers=1000MB",
             "max_locks_per_transaction=16384",
         ],
     )

test_runner/regress/test_import_pgdata.py

@@ -19,7 +19,6 @@ from fixtures.pageserver.http import (
 from fixtures.pg_version import PgVersion
 from fixtures.port_distributor import PortDistributor
 from fixtures.remote_storage import MockS3Server, RemoteStorageKind
-from fixtures.utils import shared_buffers_for_max_cu
 from mypy_boto3_kms import KMSClient
 from mypy_boto3_kms.type_defs import EncryptResponseTypeDef
 from mypy_boto3_s3 import S3Client
@@ -81,8 +80,7 @@ def test_pgdata_import_smoke(
     # doesn't allow any prefetching on v17 and above, where the new streaming
     # read machinery keeps buffers pinned while prefetching them.  Use a higher
     # setting to enable prefetching and speed up the tests
-    # use shared_buffers size like in production for 8 CU compute
-    ep_config = [f"shared_buffers={shared_buffers_for_max_cu(8.0)}"]
+    ep_config = ["shared_buffers=64MB"]
 
     #
     # Put data in vanilla pg

test_runner/regress/test_pageserver_getpage_throttle.py

@@ -16,7 +16,6 @@ if TYPE_CHECKING:
     from fixtures.neon_fixtures import NeonEnvBuilder, PgBin
 
 
-@pytest.mark.skip("See https://github.com/neondatabase/neon/issues/11395")
 def test_pageserver_getpage_throttle(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin):
     env = neon_env_builder.init_start()
 

test_runner/regress/test_physical_replication.py

@@ -7,7 +7,6 @@ from typing import TYPE_CHECKING
 import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import wait_replica_caughtup
-from fixtures.utils import shared_buffers_for_max_cu
 
 if TYPE_CHECKING:
     from fixtures.neon_fixtures import NeonEnv
@@ -181,8 +180,7 @@ def test_physical_replication_config_mismatch_too_many_known_xids(neon_simple_en
         endpoint_id="primary",
         config_lines=[
             "max_connections=1000",
-            # use shared_buffers size like in production for 2 CU compute
-            f"shared_buffers={shared_buffers_for_max_cu(2.0)}",  # prevent "no unpinned buffers available" error
+            "shared_buffers=128MB",  # prevent "no unpinned buffers available" error
         ],
     )
     secondary = env.endpoints.new_replica_start(

test_runner/regress/test_storage_controller.py

@@ -4073,6 +4073,134 @@ def test_storage_controller_location_conf_equivalence(neon_env_builder: NeonEnvB
     assert reconciles_after_restart == 0
 
 
+@pytest.mark.parametrize("restart_storcon", [True, False])
+@pytest.mark.parametrize("delete_only_timeline", [True, False])
+def test_storcon_create_delete_sk_down(
+    neon_env_builder: NeonEnvBuilder, restart_storcon: bool, delete_only_timeline: bool
+):
+    """
+    Test that the storcon can create and delete tenants and timelines with a safekeeper being down.
+    """
+
+    neon_env_builder.num_safekeepers = 3
+    neon_env_builder.storage_controller_config = {
+        "timelines_onto_safekeepers": True,
+    }
+    env = neon_env_builder.init_start()
+
+    env.safekeepers[0].stop()
+
+    # Wait for heartbeater to pick up that the safekeeper is gone
+    # This isn't really neccessary
+    def logged_offline():
+        env.storage_controller.assert_log_contains(
+            "Heartbeat round complete for 3 safekeepers, 1 offline"
+        )
+
+    wait_until(logged_offline)
+
+    tenant_id = TenantId.generate()
+    timeline_id = TimelineId.generate()
+    env.create_tenant(tenant_id, timeline_id)
+    branch_timeline_id = env.create_branch("child_of_main", tenant_id)
+
+    log.info(
+        f"Creating tenant {tenant_id} with main timeline {timeline_id} and branch {branch_timeline_id}"
+    )
+
+    env.storage_controller.allowed_errors.extend(
+        [
+            ".*Call to safekeeper.* management API still failed after.*",
+            f".*reconcile_one.*tenant_id={tenant_id}.*Call to safekeeper.* management API failed, will retry.*",
+            f".*reconcile_one.*tenant_id={tenant_id}.*Call to safekeeper.* management API still failed after.*",
+        ]
+    )
+
+    env.safekeepers[1].assert_log_contains(f"creating new timeline {tenant_id}/{timeline_id}")
+    env.safekeepers[2].assert_log_contains(f"creating new timeline {tenant_id}/{timeline_id}")
+
+    if restart_storcon:
+        # Restart the storcon to check that we persist operations
+        env.storage_controller.stop()
+        env.storage_controller.start()
+
+    config_lines = [
+        "neon.safekeeper_proto_version = 3",
+    ]
+    with env.endpoints.create("main", tenant_id=tenant_id, config_lines=config_lines) as ep:
+        # endpoint should start.
+        ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
+        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
+
+    with env.endpoints.create("child_of_main", tenant_id=tenant_id, config_lines=config_lines) as ep:
+        # endpoint should start.
+        ep.start(safekeeper_generation=1, safekeepers=[1, 2, 3])
+        ep.safe_psql("CREATE TABLE IF NOT EXISTS t(key int, value text)")
+
+    env.storage_controller.assert_log_contains("writing pending op for sk id 0")
+    env.safekeepers[0].start()
+
+    # ensure that we applied the operation also for the safekeeper we just brought down
+    def logged_contains_on_sk():
+        env.safekeepers[0].assert_log_contains(
+            f"pulling timeline {tenant_id}/{timeline_id} from safekeeper"
+        )
+        env.safekeepers[0].assert_log_contains(
+            f"pulling timeline {tenant_id}/{branch_timeline_id} from safekeeper"
+        )
+
+    wait_until(logged_contains_on_sk)
+
+    env.safekeepers[1].stop()
+
+    if delete_only_timeline:
+        env.storage_controller.pageserver_api().timeline_delete(tenant_id, branch_timeline_id)
+    else:
+        env.storage_controller.pageserver_api().tenant_delete(tenant_id)
+
+    def logged_delete_finished():
+        env.safekeepers[0].assert_log_contains(f"method=DELETE.*{tenant_id}.*Request handled")
+        env.safekeepers[2].assert_log_contains(f"method=DELETE.*{tenant_id}.*Request handled")
+
+    wait_until(logged_delete_finished)
+
+    env.safekeepers[0].assert_log_contains(
+        f"deleting timeline {tenant_id}/{branch_timeline_id} from disk"
+    )
+    env.safekeepers[2].assert_log_contains(
+        f"deleting timeline {tenant_id}/{branch_timeline_id} from disk"
+    )
+
+    root_was_deleted_on_0 = env.safekeepers[0].log_contains(
+        f"deleting timeline {tenant_id}/{timeline_id} from disk"
+    )
+    root_was_deleted_on_2 = env.safekeepers[2].log_contains(
+        f"deleting timeline {tenant_id}/{timeline_id} from disk"
+    )
+    assert (root_was_deleted_on_0 is None) == (root_was_deleted_on_2 is None)
+
+    # We only delete the root timeline iff the tenant delete was requested
+    if delete_only_timeline:
+        assert not root_was_deleted_on_0
+    else:
+        assert root_was_deleted_on_0
+
+    if restart_storcon:
+        # Restart the storcon to check that we persist operations
+        env.storage_controller.stop()
+        env.storage_controller.start()
+
+    env.safekeepers[1].start()
+
+    # ensure that there is log msgs for the third safekeeper too
+    def logged_deleted_on_sk():
+        env.safekeepers[1].assert_log_contains(
+            f"deleting timeline {tenant_id}/{branch_timeline_id} from disk"
+        )
+
+    wait_until(logged_deleted_on_sk)
+
+
 @pytest.mark.parametrize("wrong_az", [True, False])
 def test_storage_controller_graceful_migration(neon_env_builder: NeonEnvBuilder, wrong_az: bool):
     """