fix clippy

Signed-off-by: Alex Chi Z <chi@neon.tech>

.github/workflows/build_and_test.yml

@@ -263,9 +263,8 @@ jobs:
           echo "json=$(jq --compact-output '.' /tmp/benchmark_durations.json)" >> $GITHUB_OUTPUT
 
   benchmarks:
-    # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `deploy` in PRs
-    if: github.ref_name == 'main' || (contains(github.event.pull_request.labels.*.name, 'run-benchmarks') && !failure() && !cancelled())
-    needs: [ check-permissions, build-build-tools-image, get-benchmarks-durations, deploy ]
+    if: github.ref_name == 'main' || contains(github.event.pull_request.labels.*.name, 'run-benchmarks')
+    needs: [ check-permissions, build-and-test-locally, build-build-tools-image, get-benchmarks-durations ]
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write

Cargo.lock

@@ -6462,7 +6462,6 @@ dependencies = [
  "pageserver_client",
  "postgres_connection",
  "rand 0.8.5",
- "regex",
  "reqwest",
  "routerify",
  "rustls 0.23.18",

pageserver/src/tenant.rs

@@ -283,6 +283,53 @@ pub(crate) enum SpawnMode {
     Lazy,
 }
 
+/// A notifier that can be used to trigger compaction shared by all timelines of a tenant.
+///
+/// It is used to notify the compaction loop that there is work to be done. It is also used
+/// to balance the load of compaction between timelines. If there are pending L0 compaction in
+/// one of the timeline, it could preempt long-running compaction jobs (e.g., image compaction)
+/// on other timelines.
+pub struct CompactionNotifier {
+    notify: Notify,
+    l0_count: std::sync::Mutex<HashMap<TimelineId, usize>>,
+}
+
+impl CompactionNotifier {
+    #[allow(clippy::new_without_default)]
+    pub fn new() -> Self {
+        Self {
+            notify: Notify::new(),
+            l0_count: std::sync::Mutex::new(HashMap::new()),
+        }
+    }
+
+    pub fn notify_one(&self) {
+        self.notify.notify_one();
+    }
+
+    pub fn notified(&self) -> tokio::sync::futures::Notified<'_> {
+        self.notify.notified()
+    }
+
+    pub fn on_l0_update(&self, timeline_id: TimelineId, l0_count: usize) {
+        let mut guard = self.l0_count.lock().unwrap();
+        guard.insert(timeline_id, l0_count);
+    }
+
+    pub fn on_shutdown(&self, timeline_id: TimelineId) {
+        let mut guard = self.l0_count.lock().unwrap();
+        guard.remove(&timeline_id);
+    }
+
+    pub fn get_max_l0_count(&self) -> Option<(usize, TimelineId)> {
+        let guard = self.l0_count.lock().unwrap();
+        guard
+            .iter()
+            .max_by_key(|(_, count)| *count)
+            .map(|(timeline_id, count)| (*count, *timeline_id))
+    }
+}
+
 ///
 /// Tenant consists of multiple timelines. Keep them in a hash table.
 ///
@@ -358,7 +405,7 @@ pub struct Tenant {
     compaction_circuit_breaker: std::sync::Mutex<CircuitBreaker>,
 
     /// Signals the tenant compaction loop that there is L0 compaction work to be done.
-    pub(crate) l0_compaction_trigger: Arc<Notify>,
+    pub(crate) l0_compaction_trigger: Arc<CompactionNotifier>,
 
     /// Scheduled gc-compaction tasks.
     scheduled_compaction_tasks: std::sync::Mutex<HashMap<TimelineId, Arc<GcCompactionQueue>>>,
@@ -4242,7 +4289,7 @@ impl Tenant {
                 // use an extremely long backoff.
                 Some(Duration::from_secs(3600 * 24)),
             )),
-            l0_compaction_trigger: Arc::new(Notify::new()),
+            l0_compaction_trigger: Arc::new(CompactionNotifier::new()),
             scheduled_compaction_tasks: Mutex::new(Default::default()),
             activate_now_sem: tokio::sync::Semaphore::new(0),
             attach_wal_lag_cooldown: Arc::new(std::sync::OnceLock::new()),

pageserver/src/tenant/timeline.rs

@@ -47,7 +47,7 @@ use serde_with::serde_as;
 use storage_broker::BrokerClientChannel;
 use tokio::runtime::Handle;
 use tokio::sync::mpsc::Sender;
-use tokio::sync::{oneshot, watch, Notify};
+use tokio::sync::{oneshot, watch};
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::critical;
@@ -151,7 +151,8 @@ use super::{
     MaybeOffloaded,
 };
 use super::{
-    debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf, HeatMapTimeline,
+    debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf, CompactionNotifier,
+    HeatMapTimeline,
 };
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
 use super::{
@@ -224,7 +225,7 @@ pub struct TimelineResources {
     pub remote_client: RemoteTimelineClient,
     pub pagestream_throttle: Arc<crate::tenant::throttle::Throttle>,
     pub pagestream_throttle_metrics: Arc<crate::metrics::tenant_throttling::Pagestream>,
-    pub l0_compaction_trigger: Arc<Notify>,
+    pub l0_compaction_trigger: Arc<CompactionNotifier>,
     pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
 }
 
@@ -428,7 +429,7 @@ pub struct Timeline {
     compaction_failed: AtomicBool,
 
     /// Notifies the tenant compaction loop that there is pending L0 compaction work.
-    l0_compaction_trigger: Arc<Notify>,
+    l0_compaction_trigger: Arc<CompactionNotifier>,
 
     /// Make sure we only have one running gc at a time.
     ///
@@ -1965,6 +1966,8 @@ impl Timeline {
         // ... and inform any waiters for newer LSNs that there won't be any.
         self.last_record_lsn.shutdown();
 
+        self.l0_compaction_trigger.on_shutdown(self.timeline_id);
+
         if let ShutdownMode::FreezeAndFlush = mode {
             let do_flush = if let Some((open, frozen)) = self
                 .layers
@@ -4119,6 +4122,8 @@ impl Timeline {
                 if l0_count >= self.get_compaction_threshold() {
                     self.l0_compaction_trigger.notify_one();
                 }
+                self.l0_compaction_trigger
+                    .on_l0_update(self.timeline_id, l0_count);
 
                 // Delay the next flush to backpressure if compaction can't keep up. We delay by the
                 // flush duration such that the flush takes 2x as long. This is propagated up to WAL
@@ -5067,20 +5072,22 @@ impl Timeline {
                     // image layer generation taking too long time and blocking L0 compaction. So in this
                     // mode, we also inspect the current number of L0 layers and skip image layer generation
                     // if there are too many of them.
-                    let num_of_l0_layers = {
-                        let layers = self.layers.read().await;
-                        layers.layer_map()?.level0_deltas().len()
-                    };
-                    let image_preempt_threshold = self.get_image_creation_preempt_threshold()
-                        * self.get_compaction_threshold();
-                    if image_preempt_threshold != 0 && num_of_l0_layers >= image_preempt_threshold {
-                        tracing::info!(
-                        "preempt image layer generation at {lsn} when processing partition {}..{}: too many L0 layers {}",
-                        partition.start().unwrap(), partition.end().unwrap(), num_of_l0_layers
-                    );
-                        last_partition_processed = Some(partition.clone());
-                        all_generated = false;
-                        break;
+                    if let Some((max_num_of_l0_layers, timeline_id)) =
+                        self.l0_compaction_trigger.get_max_l0_count()
+                    {
+                        let image_preempt_threshold = self.get_image_creation_preempt_threshold()
+                            * self.get_compaction_threshold();
+                        if image_preempt_threshold != 0
+                            && max_num_of_l0_layers >= image_preempt_threshold
+                        {
+                            tracing::info!(
+                                "preempt image layer generation at {lsn} when processing partition {}..{}: too many L0 layers {} on timeline {}",
+                                partition.start().unwrap(), partition.end().unwrap(), max_num_of_l0_layers, timeline_id
+                            );
+                            last_partition_processed = Some(partition.clone());
+                            all_generated = false;
+                            break;
+                        }
                     }
                 }
             }
@@ -5459,8 +5466,13 @@ impl Timeline {
         self.remote_client
             .schedule_compaction_update(&remove_layers, new_deltas)?;
 
+        let l0_count = guard.layer_map()?.level0_deltas().len();
+
         drop_wlock(guard);
 
+        self.l0_compaction_trigger
+            .on_l0_update(self.timeline_id, l0_count);
+
         Ok(())
     }
 

pageserver/src/virtual_file.rs

@@ -496,8 +496,7 @@ pub(crate) fn is_fatal_io_error(e: &std::io::Error) -> bool {
 /// bad storage or bad configuration, and we can't fix that from inside
 /// a running process.
 pub(crate) fn on_fatal_io_error(e: &std::io::Error, context: &str) -> ! {
-    let backtrace = std::backtrace::Backtrace::force_capture();
-    tracing::error!("Fatal I/O error: {e}: {context})\n{backtrace}");
+    tracing::error!("Fatal I/O error: {e}: {context})");
     std::process::abort();
 }
 
@@ -948,18 +947,13 @@ impl VirtualFileInner {
     where
         Buf: tokio_epoll_uring::IoBufMut + Send,
     {
-        let file_guard = match self
-            .lock_file()
-            .await
-            .maybe_fatal_err("lock_file inside VirtualFileInner::read_at")
-        {
+        let file_guard = match self.lock_file().await {
             Ok(file_guard) => file_guard,
             Err(e) => return (buf, Err(e)),
         };
 
         observe_duration!(StorageIoOperation::Read, {
             let ((_file_guard, buf), res) = io_engine::get().read_at(file_guard, offset, buf).await;
-            let res = res.maybe_fatal_err("io_engine read_at inside VirtualFileInner::read_at");
             if let Ok(size) = res {
                 STORAGE_IO_SIZE
                     .with_label_values(&[

safekeeper/src/http/routes.rs

@@ -626,7 +626,7 @@ pub fn make_router(
                 failpoints_handler(r, cancel).await
             })
         })
-        .get("/v1/utilization", |r| request_span(r, utilization_handler))
+        .get("/v1/uzilization", |r| request_span(r, utilization_handler))
         .delete("/v1/tenant/:tenant_id", |r| {
             request_span(r, tenant_delete_handler)
         })

storage_controller/Cargo.toml

@@ -34,7 +34,6 @@ reqwest = { workspace = true, features = ["stream"] }
 routerify.workspace = true
 safekeeper_api.workspace = true
 safekeeper_client.workspace = true
-regex.workspace = true
 rustls-native-certs.workspace = true
 serde.workspace = true
 serde_json.workspace = true

storage_controller/src/http.rs

@@ -516,17 +516,6 @@ async fn handle_tenant_timeline_block_unblock_gc(
     json_response(StatusCode::OK, ())
 }
 
-// For metric labels where we would like to include the approximate path, but exclude high-cardinality fields like query parameters
-// and tenant/timeline IDs.  Since we are proxying to arbitrary paths, we don't have routing templates to
-// compare to, so we can just filter out our well known ID format with regexes.
-fn path_without_ids(path: &str) -> String {
-    static ID_REGEX: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
-    ID_REGEX
-        .get_or_init(|| regex::Regex::new(r"([0-9a-fA-F]{32}(-[0-9]{4})?|\?.*)").unwrap())
-        .replace_all(path, "")
-        .to_string()
-}
-
 async fn handle_tenant_timeline_passthrough(
     service: Arc<Service>,
     req: Request<Body>,
@@ -562,7 +551,10 @@ async fn handle_tenant_timeline_passthrough(
         .metrics_group
         .storage_controller_passthrough_request_latency;
 
-    let path_label = path_without_ids(&path)
+    // This is a bit awkward. We remove the param from the request
+    // and join the words by '_' to get a label for the request.
+    let just_path = path.replace(&tenant_shard_str, "");
+    let path_label = just_path
         .split('/')
         .filter(|token| !token.is_empty())
         .collect::<Vec<_>>()
@@ -2097,16 +2089,3 @@ pub fn make_router(
             )
         })
 }
-
-#[cfg(test)]
-mod test {
-
-    use super::path_without_ids;
-
-    #[test]
-    fn test_path_without_ids() {
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788/timeline/AA223344556677881122334455667788"), "/v1/tenant//timeline/");
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788"), "/v1/tenant//timeline/");
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788?parameter=foo"), "/v1/tenant//timeline/");
-    }
-}

storage_controller/src/main.rs

@@ -12,8 +12,7 @@ use storage_controller::persistence::Persistence;
 use storage_controller::service::chaos_injector::ChaosInjector;
 use storage_controller::service::{
     Config, Service, HEARTBEAT_INTERVAL_DEFAULT, LONG_RECONCILE_THRESHOLD_DEFAULT,
-    MAX_OFFLINE_INTERVAL_DEFAULT, MAX_WARMING_UP_INTERVAL_DEFAULT,
-    PRIORITY_RECONCILER_CONCURRENCY_DEFAULT, RECONCILER_CONCURRENCY_DEFAULT,
+    MAX_OFFLINE_INTERVAL_DEFAULT, MAX_WARMING_UP_INTERVAL_DEFAULT, RECONCILER_CONCURRENCY_DEFAULT,
 };
 use tokio::signal::unix::SignalKind;
 use tokio_util::sync::CancellationToken;
@@ -76,14 +75,10 @@ struct Cli {
     #[arg(long)]
     split_threshold: Option<u64>,
 
-    /// Maximum number of normal-priority reconcilers that may run in parallel
+    /// Maximum number of reconcilers that may run in parallel
     #[arg(long)]
     reconciler_concurrency: Option<usize>,
 
-    /// Maximum number of high-priority reconcilers that may run in parallel
-    #[arg(long)]
-    priority_reconciler_concurrency: Option<usize>,
-
     /// How long to wait for the initial database connection to be available.
     #[arg(long, default_value = "5s")]
     db_connect_timeout: humantime::Duration,
@@ -294,9 +289,6 @@ async fn async_main() -> anyhow::Result<()> {
         reconciler_concurrency: args
             .reconciler_concurrency
             .unwrap_or(RECONCILER_CONCURRENCY_DEFAULT),
-        priority_reconciler_concurrency: args
-            .priority_reconciler_concurrency
-            .unwrap_or(PRIORITY_RECONCILER_CONCURRENCY_DEFAULT),
         split_threshold: args.split_threshold,
         neon_local_repo_dir: args.neon_local_repo_dir,
         max_secondary_lag_bytes: args.max_secondary_lag_bytes,

storage_controller/src/reconciler.rs

@@ -91,10 +91,9 @@ pub(crate) struct ReconcilerConfigBuilder {
 }
 
 impl ReconcilerConfigBuilder {
-    /// Priority is special: you must pick one thoughtfully, do not just use 'normal' as the default
-    pub(crate) fn new(priority: ReconcilerPriority) -> Self {
+    pub(crate) fn new() -> Self {
         Self {
-            config: ReconcilerConfig::new(priority),
+            config: ReconcilerConfig::default(),
         }
     }
 
@@ -130,18 +129,8 @@ impl ReconcilerConfigBuilder {
     }
 }
 
-// Higher priorities are used for user-facing tasks, so that a long backlog of housekeeping work (e.g. reconciling on startup, rescheduling
-// things on node changes) does not starve user-facing tasks.
-#[derive(Debug, Copy, Clone)]
-pub(crate) enum ReconcilerPriority {
-    Normal,
-    High,
-}
-
-#[derive(Debug, Copy, Clone)]
+#[derive(Default, Debug, Copy, Clone)]
 pub(crate) struct ReconcilerConfig {
-    pub(crate) priority: ReconcilerPriority,
-
     // During live migration give up on warming-up the secondary
     // after this timeout.
     secondary_warmup_timeout: Option<Duration>,
@@ -156,18 +145,6 @@ pub(crate) struct ReconcilerConfig {
 }
 
 impl ReconcilerConfig {
-    /// Configs are always constructed with an explicit priority, to force callers to think about whether
-    /// the operation they're scheduling is high-priority or not. Normal priority is not a safe default, because
-    /// scheduling something user-facing at normal priority can result in it getting starved out by background work.
-    pub(crate) fn new(priority: ReconcilerPriority) -> Self {
-        Self {
-            priority,
-            secondary_warmup_timeout: None,
-            secondary_download_request_timeout: None,
-            tenant_creation_hint: false,
-        }
-    }
-
     pub(crate) fn get_secondary_warmup_timeout(&self) -> Duration {
         const SECONDARY_WARMUP_TIMEOUT_DEFAULT: Duration = Duration::from_secs(300);
         self.secondary_warmup_timeout
@@ -187,9 +164,7 @@ impl ReconcilerConfig {
 
 impl From<&MigrationConfig> for ReconcilerConfig {
     fn from(value: &MigrationConfig) -> Self {
-        // Run reconciler at high priority because MigrationConfig comes from human requests that should
-        // be presumed urgent.
-        let mut builder = ReconcilerConfigBuilder::new(ReconcilerPriority::High);
+        let mut builder = ReconcilerConfigBuilder::new();
 
         if let Some(timeout) = value.secondary_warmup_timeout {
             builder = builder.secondary_warmup_timeout(timeout)

storage_controller/src/service.rs

@@ -30,10 +30,7 @@ use crate::{
         AbortShardSplitStatus, ControllerPersistence, DatabaseResult, MetadataHealthPersistence,
         ShardGenerationState, TenantFilter,
     },
-    reconciler::{
-        ReconcileError, ReconcileUnits, ReconcilerConfig, ReconcilerConfigBuilder,
-        ReconcilerPriority,
-    },
+    reconciler::{ReconcileError, ReconcileUnits, ReconcilerConfig, ReconcilerConfigBuilder},
     safekeeper::Safekeeper,
     scheduler::{MaySchedule, ScheduleContext, ScheduleError, ScheduleMode},
     tenant_shard::{
@@ -82,7 +79,7 @@ use pageserver_api::{
     },
 };
 use pageserver_client::{mgmt_api, BlockUnblock};
-use tokio::sync::{mpsc::error::TrySendError, TryAcquireError};
+use tokio::sync::mpsc::error::TrySendError;
 use tokio_util::sync::CancellationToken;
 use utils::{
     completion::Barrier,
@@ -198,7 +195,6 @@ pub(crate) enum LeadershipStatus {
 }
 
 pub const RECONCILER_CONCURRENCY_DEFAULT: usize = 128;
-pub const PRIORITY_RECONCILER_CONCURRENCY_DEFAULT: usize = 256;
 
 // Depth of the channel used to enqueue shards for reconciliation when they can't do it immediately.
 // This channel is finite-size to avoid using excessive memory if we get into a state where reconciles are finishing more slowly
@@ -370,12 +366,9 @@ pub struct Config {
     /// and/or upon handling the re-attach request from a node.
     pub max_warming_up_interval: Duration,
 
-    /// How many normal-priority Reconcilers may be spawned concurrently
+    /// How many Reconcilers may be spawned concurrently
     pub reconciler_concurrency: usize,
 
-    /// How many high-priority Reconcilers may be spawned concurrently
-    pub priority_reconciler_concurrency: usize,
-
     /// How large must a shard grow in bytes before we split it?
     /// None disables auto-splitting.
     pub split_threshold: Option<u64>,
@@ -443,14 +436,9 @@ pub struct Service {
     // that transition it to/from Active.
     node_op_locks: IdLockMap<NodeId, NodeOperations>,
 
-    // Limit how many Reconcilers we will spawn concurrently for normal-priority tasks such as background reconciliations
-    // and reconciliation on startup.
+    // Limit how many Reconcilers we will spawn concurrently
     reconciler_concurrency: Arc<tokio::sync::Semaphore>,
 
-    // Limit how many Reconcilers we will spawn concurrently for high-priority tasks such as tenant/timeline CRUD, which
-    // a human user might be waiting for.
-    priority_reconciler_concurrency: Arc<tokio::sync::Semaphore>,
-
     /// Queue of tenants who are waiting for concurrency limits to permit them to reconcile
     /// Send into this queue to promptly attempt to reconcile this shard next time units are available.
     ///
@@ -1275,15 +1263,12 @@ impl Service {
         }
 
         // Maybe some other work can proceed now that this job finished.
-        //
-        // Only bother with this if we have some semaphore units available in the normal-priority semaphore (these
-        // reconciles are scheduled at `[ReconcilerPriority::Normal]`).
         if self.reconciler_concurrency.available_permits() > 0 {
             while let Ok(tenant_shard_id) = locked.delayed_reconcile_rx.try_recv() {
                 let (nodes, tenants, _scheduler) = locked.parts_mut();
                 if let Some(shard) = tenants.get_mut(&tenant_shard_id) {
                     shard.delayed_reconcile = false;
-                    self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::Normal);
+                    self.maybe_reconcile_shard(shard, nodes);
                 }
 
                 if self.reconciler_concurrency.available_permits() == 0 {
@@ -1580,9 +1565,6 @@ impl Service {
             reconciler_concurrency: Arc::new(tokio::sync::Semaphore::new(
                 config.reconciler_concurrency,
             )),
-            priority_reconciler_concurrency: Arc::new(tokio::sync::Semaphore::new(
-                config.priority_reconciler_concurrency,
-            )),
             delayed_reconcile_tx,
             abort_tx,
             startup_complete: startup_complete.clone(),
@@ -2355,7 +2337,7 @@ impl Service {
         let waiters = {
             let mut locked = self.inner.write().unwrap();
             let (nodes, tenants, _scheduler) = locked.parts_mut();
-            let config = ReconcilerConfigBuilder::new(ReconcilerPriority::High)
+            let config = ReconcilerConfigBuilder::new()
                 .tenant_creation_hint(true)
                 .build();
             tenants
@@ -2830,8 +2812,7 @@ impl Service {
 
                         shard.schedule(scheduler, &mut schedule_context)?;
 
-                        let maybe_waiter =
-                            self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High);
+                        let maybe_waiter = self.maybe_reconcile_shard(shard, nodes);
                         if let Some(waiter) = maybe_waiter {
                             waiters.push(waiter);
                         }
@@ -2952,9 +2933,7 @@ impl Service {
             let (nodes, tenants, _scheduler) = locked.parts_mut();
             for (_shard_id, shard) in tenants.range_mut(TenantShardId::tenant_range(tenant_id)) {
                 shard.config = config.clone();
-                if let Some(waiter) =
-                    self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High)
-                {
+                if let Some(waiter) = self.maybe_reconcile_shard(shard, nodes) {
                     waiters.push(waiter);
                 }
             }
@@ -3236,9 +3215,7 @@ impl Service {
                 debug_assert!(shard.intent.get_attached().is_none());
                 debug_assert!(shard.intent.get_secondary().is_empty());
 
-                if let Some(waiter) =
-                    self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High)
-                {
+                if let Some(waiter) = self.maybe_reconcile_shard(shard, nodes) {
                     detach_waiters.push(waiter);
                 }
             }
@@ -3390,7 +3367,7 @@ impl Service {
 
             // In case scheduling is being switched back on, try it now.
             shard.schedule(scheduler, &mut schedule_context).ok();
-            self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High);
+            self.maybe_reconcile_shard(shard, nodes);
         }
 
         Ok(())
@@ -4439,7 +4416,7 @@ impl Service {
                     tracing::warn!("Failed to schedule {tenant_shard_id} during shard abort: {e}")
                 }
 
-                self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High);
+                self.maybe_reconcile_shard(shard, nodes);
             }
 
             // We don't expect any new_shard_count shards to exist here, but drop them just in case
@@ -4605,11 +4582,7 @@ impl Service {
                         tracing::warn!("Failed to schedule child shard {child}: {e}");
                     }
                     // In the background, attach secondary locations for the new shards
-                    if let Some(waiter) = self.maybe_reconcile_shard(
-                        &mut child_state,
-                        nodes,
-                        ReconcilerPriority::High,
-                    ) {
+                    if let Some(waiter) = self.maybe_reconcile_shard(&mut child_state, nodes) {
                         waiters.push(waiter);
                     }
 
@@ -4974,9 +4947,7 @@ impl Service {
                 shard.intent.clear_secondary(scheduler);
 
                 // Run Reconciler to execute detach fo secondary locations.
-                if let Some(waiter) =
-                    self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High)
-                {
+                if let Some(waiter) = self.maybe_reconcile_shard(shard, nodes) {
                     waiters.push(waiter);
                 }
             }
@@ -5244,7 +5215,7 @@ impl Service {
 
             let reconciler_config = match migrate_req.migration_config {
                 Some(cfg) => (&cfg).into(),
-                None => ReconcilerConfig::new(ReconcilerPriority::High),
+                None => ReconcilerConfig::default(),
             };
 
             self.maybe_configured_reconcile_shard(shard, nodes, reconciler_config)
@@ -5310,7 +5281,7 @@ impl Service {
                 );
             }
 
-            self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::High)
+            self.maybe_reconcile_shard(shard, nodes)
         };
 
         if let Some(waiter) = waiter {
@@ -5722,7 +5693,7 @@ impl Service {
                             )
                         }
 
-                        self.maybe_reconcile_shard(shard, nodes, ReconcilerPriority::Normal);
+                        self.maybe_reconcile_shard(shard, nodes);
                     }
 
                     // Here we remove an existing observed location for the node we're removing, and it will
@@ -6091,14 +6062,7 @@ impl Service {
                                     tracing::warn!(%tenant_shard_id, "Scheduling error when marking pageserver {} offline: {e}", node_id);
                                 }
                                 Ok(()) => {
-                                    if self
-                                        .maybe_reconcile_shard(
-                                            tenant_shard,
-                                            nodes,
-                                            ReconcilerPriority::Normal,
-                                        )
-                                        .is_some()
-                                    {
+                                    if self.maybe_reconcile_shard(tenant_shard, nodes).is_some() {
                                         tenants_affected += 1;
                                     };
                                 }
@@ -6129,11 +6093,7 @@ impl Service {
 
                     if let Some(observed_loc) = tenant_shard.observed.locations.get_mut(&node_id) {
                         if observed_loc.conf.is_none() {
-                            self.maybe_reconcile_shard(
-                                tenant_shard,
-                                nodes,
-                                ReconcilerPriority::Normal,
-                            );
+                            self.maybe_reconcile_shard(tenant_shard, nodes);
                         }
                     }
                 }
@@ -6497,36 +6457,8 @@ impl Service {
         &self,
         shard: &mut TenantShard,
         nodes: &Arc<HashMap<NodeId, Node>>,
-        priority: ReconcilerPriority,
     ) -> Option<ReconcilerWaiter> {
-        self.maybe_configured_reconcile_shard(shard, nodes, ReconcilerConfig::new(priority))
-    }
-
-    /// Before constructing a Reconciler, acquire semaphore units from the appropriate concurrency limit (depends on priority)
-    fn get_reconciler_units(
-        &self,
-        priority: ReconcilerPriority,
-    ) -> Result<ReconcileUnits, TryAcquireError> {
-        let units = match priority {
-            ReconcilerPriority::Normal => self.reconciler_concurrency.clone().try_acquire_owned(),
-            ReconcilerPriority::High => {
-                match self
-                    .priority_reconciler_concurrency
-                    .clone()
-                    .try_acquire_owned()
-                {
-                    Ok(u) => Ok(u),
-                    Err(TryAcquireError::NoPermits) => {
-                        // If the high priority semaphore is exhausted, then high priority tasks may steal units from
-                        // the normal priority semaphore.
-                        self.reconciler_concurrency.clone().try_acquire_owned()
-                    }
-                    Err(e) => Err(e),
-                }
-            }
-        };
-
-        units.map(ReconcileUnits::new)
+        self.maybe_configured_reconcile_shard(shard, nodes, ReconcilerConfig::default())
     }
 
     /// Wrap [`TenantShard`] reconciliation methods with acquisition of [`Gate`] and [`ReconcileUnits`],
@@ -6546,8 +6478,8 @@ impl Service {
             }
         };
 
-        let units = match self.get_reconciler_units(reconciler_config.priority) {
-            Ok(u) => u,
+        let units = match self.reconciler_concurrency.clone().try_acquire_owned() {
+            Ok(u) => ReconcileUnits::new(u),
             Err(_) => {
                 tracing::info!(tenant_id=%shard.tenant_shard_id.tenant_id, shard_id=%shard.tenant_shard_id.shard_slug(),
                     "Concurrency limited: enqueued for reconcile later");
@@ -6640,10 +6572,7 @@ impl Service {
 
             // Eventual consistency: if an earlier reconcile job failed, and the shard is still
             // dirty, spawn another rone
-            if self
-                .maybe_reconcile_shard(shard, &pageservers, ReconcilerPriority::Normal)
-                .is_some()
-            {
+            if self.maybe_reconcile_shard(shard, &pageservers).is_some() {
                 reconciles_spawned += 1;
             } else if shard.delayed_reconcile {
                 // Shard wanted to reconcile but for some reason couldn't.
@@ -6729,10 +6658,7 @@ impl Service {
             tracing::info!(tenant_shard_id=%tenant_shard_id, "Applying optimization: {optimization:?}");
             if shard.apply_optimization(scheduler, optimization) {
                 optimizations_applied += 1;
-                if self
-                    .maybe_reconcile_shard(shard, nodes, ReconcilerPriority::Normal)
-                    .is_some()
-                {
+                if self.maybe_reconcile_shard(shard, nodes).is_some() {
                     reconciles_spawned += 1;
                 }
             }
@@ -7282,7 +7208,7 @@ impl Service {
         // to not stall the operation when a cold secondary is encountered.
         const SECONDARY_WARMUP_TIMEOUT: Duration = Duration::from_secs(20);
         const SECONDARY_DOWNLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
-        let reconciler_config = ReconcilerConfigBuilder::new(ReconcilerPriority::Normal)
+        let reconciler_config = ReconcilerConfigBuilder::new()
             .secondary_warmup_timeout(SECONDARY_WARMUP_TIMEOUT)
             .secondary_download_request_timeout(SECONDARY_DOWNLOAD_REQUEST_TIMEOUT)
             .build();
@@ -7615,7 +7541,7 @@ impl Service {
     ) -> Result<(), OperationError> {
         const SECONDARY_WARMUP_TIMEOUT: Duration = Duration::from_secs(20);
         const SECONDARY_DOWNLOAD_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
-        let reconciler_config = ReconcilerConfigBuilder::new(ReconcilerPriority::Normal)
+        let reconciler_config = ReconcilerConfigBuilder::new()
             .secondary_warmup_timeout(SECONDARY_WARMUP_TIMEOUT)
             .secondary_download_request_timeout(SECONDARY_DOWNLOAD_REQUEST_TIMEOUT)
             .build();

storage_controller/src/service/chaos_injector.rs

@@ -88,11 +88,7 @@ impl ChaosInjector {
 
         shard.intent.demote_attached(scheduler, old_location);
         shard.intent.promote_attached(scheduler, new_location);
-        self.service.maybe_reconcile_shard(
-            shard,
-            nodes,
-            crate::reconciler::ReconcilerPriority::Normal,
-        );
+        self.service.maybe_reconcile_shard(shard, nodes);
     }
 
     async fn inject_chaos(&mut self) {

test_runner/regress/test_storage_controller.py

@@ -3189,17 +3189,15 @@ def test_safekeeper_deployment_time_update(neon_env_builder: NeonEnvBuilder):
 
     assert len(target.get_safekeepers()) == 0
 
-    sk_0 = env.safekeepers[0]
-
     body = {
         "active": True,
         "id": fake_id,
         "created_at": "2023-10-25T09:11:25Z",
         "updated_at": "2024-08-28T11:32:43Z",
         "region_id": "aws-us-east-2",
-        "host": "localhost",
-        "port": sk_0.port.pg,
-        "http_port": sk_0.port.http,
+        "host": "safekeeper-333.us-east-2.aws.neon.build",
+        "port": 6401,
+        "http_port": 7676,
         "version": 5957,
         "availability_zone_id": "us-east-2b",
     }
@@ -3245,13 +3243,6 @@ def test_safekeeper_deployment_time_update(neon_env_builder: NeonEnvBuilder):
     # Ensure idempotency
     target.safekeeper_scheduling_policy(inserted["id"], "Decomissioned")
 
-    def storcon_heartbeat():
-        assert env.storage_controller.log_contains(
-            "Heartbeat round complete for 1 safekeepers, 0 offline"
-        )
-
-    wait_until(storcon_heartbeat)
-
 
 def eq_safekeeper_records(a: dict[str, Any], b: dict[str, Any]) -> bool:
     compared = [dict(a), dict(b)]