compute: add non-skippable apply config operations.

apply_config() step of compute start is controlled by skip_pg_catalog_updates flag,
this is a performance optimization to decrease compute startup time, but it introduces extra dependency on cplane.
Introduce small subset of operations that we run always, independent from this flag.

compute_tools/src/compute.rs

@@ -1497,6 +1497,27 @@ impl ComputeNode {
         Ok::<(), anyhow::Error>(())
     }
 
+    /// Apply config operations that are not covered by `skip_pg_catalog_updates`
+    #[instrument(skip_all)]
+    pub fn apply_config_non_skippable(&self, compute_state: &ComputeState) -> Result<()> {
+        let conf = self.get_tokio_conn_conf(Some("compute_ctl:apply_config"));
+
+        let conf = Arc::new(conf);
+        let spec = Arc::new(
+            compute_state
+                .pspec
+                .as_ref()
+                .expect("spec must be set")
+                .spec
+                .clone(),
+        );
+
+        // Merge-apply spec & changes to PostgreSQL state.
+        self.apply_spec_sql_non_skippable(spec.clone(), conf.clone())?;
+
+        Ok::<(), anyhow::Error>(())
+    }
+
     // Wrapped this around `pg_ctl reload`, but right now we don't use
     // `pg_ctl` for start / stop.
     #[instrument(skip_all)]
@@ -1619,8 +1640,24 @@ impl ComputeNode {
                     "updated postgresql.conf to set neon.disable_logical_replication_subscribers=false"
                 );
             }
+            self.pg_reload_conf()?;
+        } else {
+            // We need to run some operations even if skip_pg_catalog_updates is set
+            let pgdata_path = Path::new(&self.params.pgdata);
+            // temporarily reset max_cluster_size in config
+            // to avoid the possibility of hitting the limit, while we are applying config:
+            // creating new extensions, roles, etc...
+            config::with_compute_ctl_tmp_override(pgdata_path, "neon.max_cluster_size=-1", || {
+                self.pg_reload_conf()?;
+
+                self.apply_config_non_skippable(compute_state)?;
+
+                Ok(())
+            })?;
+
             self.pg_reload_conf()?;
         }
+
         self.post_apply_config()?;
 
         Ok(())

compute_tools/src/spec_apply.rs

@@ -308,6 +308,75 @@ impl ComputeNode {
         Ok(())
     }
 
+    // Similar to apply_spec_sql, but for the simplified set of operations
+    // that we perform even when `pg_skip_catalog_updates` is set.
+    //
+    // Keep the list of operations as small as possible,
+    // as it will be run on every spec change and affect compute start time.
+    pub fn apply_spec_sql_non_skippable(
+        &self,
+        spec: Arc<ComputeSpec>,
+        conf: Arc<tokio_postgres::Config>,
+    ) -> Result<()> {
+        info!("Applying non_skippable config",);
+        debug!("Config: {:?}", spec);
+
+        let rt = tokio::runtime::Handle::current();
+        rt.block_on(async {
+            let client = Self::get_maintenance_client(&conf).await?;
+            let spec = spec.clone();
+
+            let jwks_roles = Arc::new(
+                spec.as_ref()
+                    .local_proxy_config
+                    .iter()
+                    .flat_map(|it| &it.jwks)
+                    .flatten()
+                    .flat_map(|setting| &setting.role_names)
+                    .cloned()
+                    .collect::<HashSet<_>>(),
+            );
+
+            // NOTE: Here we assume that operations below don't use ctx
+            // TODO: refactor apply_operations() to accept ctx as option.
+            let ctx = Arc::new(tokio::sync::RwLock::new(MutableApplyContext {
+                roles: HashMap::new(),
+                dbs: HashMap::new(),
+            }));
+
+            let mut phases = vec![];
+
+            match spec.audit_log_level {
+                ComputeAudit::Hipaa => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(CreatePgauditlogtofileExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
+                ComputeAudit::Log => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
+                ComputeAudit::Disabled => {}
+            }
+
+            for phase in phases {
+                debug!("Applying phase {:?}", &phase);
+                apply_operations(
+                    spec.clone(),
+                    ctx.clone(),
+                    jwks_roles.clone(),
+                    phase,
+                    || async { Ok(&client) },
+                )
+                .await?;
+            }
+
+            Ok::<(), anyhow::Error>(())
+        })?;
+
+        Ok(())
+    }
+
     /// Apply SQL migrations of the RunInEachDatabase phase.
     ///
     /// May opt to not connect to databases that don't have any scheduled