fix(compute_ctl): Use coalesce without schema as it's not a catalog func

compute_tools/src/monitor.rs

@@ -407,8 +407,8 @@ fn get_database_stats(cli: &mut Client) -> anyhow::Result<(f64, i64)> {
     // like `postgres_exporter` use it to query Postgres statistics.
     // Use explicit 8 bytes type casts to match Rust types.
     let stats = cli.query_one(
-        "SELECT pg_catalog.coalesce(pg_catalog.sum(active_time), 0.0)::pg_catalog.float8 AS total_active_time,
-            pg_catalog.coalesce(pg_catalog.sum(sessions), 0)::pg_catalog.bigint AS total_sessions
+        "SELECT COALESCE(pg_catalog.sum(active_time), 0.0)::pg_catalog.float8 AS total_active_time,
+            COALESCE(pg_catalog.sum(sessions), 0)::pg_catalog.int8 AS total_sessions
         FROM pg_catalog.pg_stat_database
         WHERE datname NOT IN (
                 'postgres',

control_plane/src/endpoint.rs

@@ -241,7 +241,7 @@ impl ComputeControlPlane {
             drop_subscriptions_before_start,
             grpc,
             reconfigure_concurrency: 1,
-            features: vec![],
+            features: vec![ComputeFeature::ActivityMonitorExperimental],
             cluster: None,
             compute_ctl_config: compute_ctl_config.clone(),
             privileged_role_name: privileged_role_name.clone(),
@@ -263,7 +263,7 @@ impl ComputeControlPlane {
                 skip_pg_catalog_updates,
                 drop_subscriptions_before_start,
                 reconfigure_concurrency: 1,
-                features: vec![],
+                features: vec![ComputeFeature::ActivityMonitorExperimental],
                 cluster: None,
                 compute_ctl_config,
                 privileged_role_name,

pageserver/src/tenant.rs

@@ -1580,7 +1580,7 @@ impl TenantShard {
     }
 
     #[instrument(skip_all)]
-    pub(crate) async fn  preload(
+    pub(crate) async fn preload(
         self: &Arc<Self>,
         remote_storage: &GenericRemoteStorage,
         cancel: CancellationToken,

pageserver/src/tenant/timeline.rs

@@ -88,7 +88,7 @@ use self::eviction_task::EvictionTaskTimelineState;
 use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};
 use super::remote_timeline_client::RemoteTimelineClient;
-use super::remote_timeline_client::index::{GcCompactionState, IndexPart, LayerFileMetadata};
+use super::remote_timeline_client::index::{GcCompactionState, IndexPart};
 use super::secondary::heatmap::HeatMapLayer;
 use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
 use super::tasks::log_compaction_error;
@@ -4205,12 +4205,6 @@ impl Timeline {
                 let desc: PersistentLayerDesc = hl.name.clone().into();
                 let layer = guard.try_get_from_key(&desc.key())?;
 
-                // Make sure the layer in the old heatmap is the same generation one as in the layer
-                // map otherwise we can in some edge case keep old obsolete layers in the heatmap.
-                if layer.metadata().generation != hl.metadata.generation {
-                    return None;
-                }
-
                 if layer.visibility() == LayerVisibilityHint::Covered {
                     return None;
                 }
@@ -6480,7 +6474,6 @@ pub struct DeltaLayerTestDesc {
     pub lsn_range: Range<Lsn>,
     pub key_range: Range<Key>,
     pub data: Vec<(Key, Lsn, Value)>,
-    pub resident: bool,
 }
 
 #[cfg(test)]
@@ -6498,22 +6491,12 @@ impl DeltaLayerTestDesc {
             lsn_range,
             key_range,
             data,
-            // Default test code creates resident layers.
-            resident: true,
         }
     }
 
     pub fn new_with_inferred_key_range(
         lsn_range: Range<Lsn>,
         data: Vec<(Key, Lsn, Value)>,
-    ) -> Self {
-        Self::new_with_inferred_key_range_and_resident_state(lsn_range, data, true)
-    }
-
-    pub fn new_with_inferred_key_range_and_resident_state(
-        lsn_range: Range<Lsn>,
-        data: Vec<(Key, Lsn, Value)>,
-        resident: bool,
     ) -> Self {
         let key_min = data.iter().map(|(key, _, _)| key).min().unwrap();
         let key_max = data.iter().map(|(key, _, _)| key).max().unwrap();
@@ -6521,7 +6504,6 @@ impl DeltaLayerTestDesc {
             key_range: (*key_min)..(key_max.next()),
             lsn_range,
             data,
-            resident
         }
     }
 
@@ -7523,30 +7505,6 @@ impl Timeline {
         check_start_lsn: Option<Lsn>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-
-        if !deltas.resident {
-            // Don't need to bother creating an on-disk file, we just want the metadata for a non-resident layer.
-            let delta_layer = Layer::for_evicted(
-                self.conf,
-                self,
-                deltas.layer_name(),
-                LayerFileMetadata {
-                    generation: self.generation,
-                    shard: self.get_shard_index(),
-                    file_size: 1024,
-                },
-            );
-            info!("force created non-resident delta layer {}", deltas.layer_name());
-            {
-                let mut guard = self.layers.write(LayerManagerLockHolder::Testing).await;
-                guard
-                    .open_mut()
-                    .unwrap()
-                    .force_insert_optionally_resident_layer(delta_layer);
-            }
-            return Ok(());
-        }
-
         let last_record_lsn = self.get_last_record_lsn();
         deltas
             .data
@@ -8305,148 +8263,6 @@ mod tests {
         ));
     }
 
-    #[tokio::test]
-    async fn test_heatmap_generation_removes_layers_from_old_generation() {
-        use std::time::SystemTime;
-        use utils::generation::Generation;
-        use crate::tenant::remote_timeline_client::index::LayerFileMetadata;
-        use crate::tenant::secondary::heatmap::{HeatMapLayer, HeatMapTimeline as HeatMapTimelineStruct};
-
-        let harness = TenantHarness::create("heatmaheatmap_genereation_removes_layers_from_old_generationp_generation").await.unwrap();
-
-        // Create your existing layer descriptions
-        let covered_delta = DeltaLayerTestDesc::new_with_inferred_key_range(
-            Lsn(0x10)..Lsn(0x20),
-            vec![(
-                Key::from_hex("620000000033333333444444445500000000").unwrap(),
-                Lsn(0x11),
-                Value::Image(test_img("foo")),
-            )],
-        );
-        // This visible layer is non-resident on disk. This is important to reproduce the failure as
-        // a resident file will take priority over the previous heatmap even without the fix for
-        // this issue.
-        let visible_delta = DeltaLayerTestDesc::new_with_inferred_key_range_and_resident_state(
-            Lsn(0x10)..Lsn(0x20),
-            vec![(
-                Key::from_hex("720000000033333333444444445500000000").unwrap(),
-                Lsn(0x11),
-                Value::Image(test_img("foo")),
-            )],
-            false, // Non-resident
-        );
-        let l0_delta = DeltaLayerTestDesc::new(
-            Lsn(0x20)..Lsn(0x30),
-            Key::from_hex("000000000000000000000000000000000000").unwrap()
-                ..Key::from_hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF").unwrap(),
-            vec![(
-                Key::from_hex("720000000033333333444444445500000000").unwrap(),
-                Lsn(0x25),
-                Value::Image(test_img("foo")),
-            )],
-        );
-        let delta_layers = vec![
-            covered_delta.clone(),
-            visible_delta.clone(),
-            l0_delta.clone(),
-        ];
-
-        let image_layer = (
-            Lsn(0x40),
-            vec![(
-                Key::from_hex("620000000033333333444444445500000000").unwrap(),
-                test_img("bar"),
-            )],
-        );
-        let image_layers = vec![image_layer];
-
-        let (tenant, ctx) = harness.load().await;
-
-        // Create timeline with current generation (0xdeadbeef by default)
-        let timeline = tenant
-            .create_test_timeline_with_layers(
-                TimelineId::generate(),
-                Lsn(0x10),
-                PgMajorVersion::PG14,
-                &ctx,
-                Vec::new(), // in-memory layers
-                delta_layers,
-                image_layers,
-                Lsn(0x100),
-            )
-            .await
-            .unwrap();
-
-        // Now create a previous heatmap with the visible_delta layer from an older generation
-        let current_layer_metadata = LayerFileMetadata {
-            generation: timeline.generation,
-            shard: timeline.get_shard_index(),
-            file_size: 1024,
-        };
-        let old_generation = Generation::new(0x12345678); // Older than 0xdeadbeef
-        let old_layer_metadata = LayerFileMetadata {
-            generation: old_generation,
-            shard: timeline.get_shard_index(),
-            file_size: 1024,
-        };
-
-        // Create heatmap layers that reference the same keys but with old generation
-        let prev_heatmap_layers = vec![
-            HeatMapLayer::new(
-                covered_delta.layer_name(),
-                current_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-            HeatMapLayer::new(
-                visible_delta.layer_name(),
-                // Visible delta layer is from an older generation in heatmap
-                old_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-            HeatMapLayer::new(
-                l0_delta.layer_name(),
-                current_layer_metadata.clone(),
-                SystemTime::now(),
-                false, // not cold
-            ),
-        ];
-
-        // Create the previous heatmap with old generation layers
-        let prev_heatmap = HeatMapTimelineStruct::new(timeline.timeline_id, prev_heatmap_layers);
-
-        // Set the previous heatmap on the timeline
-        timeline
-            .previous_heatmap
-            .store(Some(Arc::new(PreviousHeatmap::Active {
-                heatmap: prev_heatmap,
-                read_at: std::time::Instant::now(),
-                end_lsn: None,
-            })));
-
-        // Layer visibility is an input to heatmap generation, so refresh it first
-        timeline.update_layer_visibility().await.unwrap();
-
-        // Generate a new heatmap - this should filter out the old generation layers
-        let heatmap = timeline
-            .generate_heatmap()
-            .await
-            .expect("Infallible while timeline is not shut down");
-
-        assert_eq!(heatmap.timeline_id, timeline.timeline_id);
-
-        // Verify that layers exist but they should be the current generation ones,
-        // not the old generation ones from previous_heatmap
-        for layer in heatmap.all_layers() {
-            assert_eq!(
-                layer.metadata.generation,
-                timeline.generation,
-                "Heatmap should only contain current generation layers, not old ones"
-            );
-        }
-    }
-
     #[tokio::test]
     async fn two_layer_eviction_attempts_at_the_same_time() {
         let harness = TenantHarness::create("two_layer_eviction_attempts_at_the_same_time")

pageserver/src/tenant/timeline/init.rs

@@ -129,16 +129,6 @@ pub(super) fn reconcile(
     // Construct Decisions for layers that are found locally, if they're in remote metadata.  Otherwise
     // construct DismissedLayers to get rid of them.
     for (layer_name, local_metadata) in local_layers {
-        // FIXME: This should probably take generation into account. Currently it's possible to have
-        // an old generation file on disk while a newer one with same name is in index (because
-        // primary just split shard) and we miss that here. We are saved by the check below because
-        // the file size is very likely to be different (and if it isn't then the file contents will
-        // probably be the same anyway in case of shard split), but it's confusing that this logic
-        // doesn't account for name collisions from older generations. Ideally, we should consider a
-        // local file from an older generation than the one in the index to be a different file and
-        // return `DismissedLayer::LocalOnly` if generations don't match. Right now though,
-        // layer_name has the generation part stripped so we'd need to re-parse the generation from
-        // the file name here or back in scan_timeline_dir and add it to LocalLayerFileMetadata.
         let Some(remote_metadata) = index_part.layer_metadata.get(&layer_name) else {
             result.push((layer_name, Err(DismissedLayer::LocalOnly(local_metadata))));
             continue;

pageserver/src/tenant/timeline/layer_manager.rs

@@ -645,13 +645,8 @@ impl OpenLayerManager {
 
     #[cfg(test)]
     pub(crate) fn force_insert_layer(&mut self, layer: ResidentLayer) {
-        self.force_insert_optionally_resident_layer(layer.as_ref().clone());
-    }
-
-    #[cfg(test)]
-    pub(crate) fn force_insert_optionally_resident_layer(&mut self, layer: Layer) {
         let mut updates = self.layer_map.batch_update();
-        Self::insert_historic_layer(layer, &mut updates, &mut self.layer_fmgr);
+        Self::insert_historic_layer(layer.as_ref().clone(), &mut updates, &mut self.layer_fmgr);
         updates.flush()
     }
 

proxy/src/serverless/rest.rs

@@ -5,17 +5,12 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use http::Method;
-use http::header::{
-    ACCESS_CONTROL_ALLOW_HEADERS, ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN,
-    ACCESS_CONTROL_EXPOSE_HEADERS, ACCESS_CONTROL_MAX_AGE, ACCESS_CONTROL_REQUEST_HEADERS, ALLOW,
-    AUTHORIZATION, CONTENT_TYPE, HOST, ORIGIN,
-};
+use http::header::{AUTHORIZATION, CONTENT_TYPE, HOST};
 use http_body_util::combinators::BoxBody;
-use http_body_util::{BodyExt, Empty, Full};
+use http_body_util::{BodyExt, Full};
 use http_utils::error::ApiError;
 use hyper::body::Incoming;
-use hyper::http::response::Builder;
-use hyper::http::{HeaderMap, HeaderName, HeaderValue};
+use hyper::http::{HeaderName, HeaderValue};
 use hyper::{Request, Response, StatusCode};
 use indexmap::IndexMap;
 use moka::sync::Cache;
@@ -72,15 +67,6 @@ use crate::util::deserialize_json_string;
 
 static EMPTY_JSON_SCHEMA: &str = r#"{"schemas":[]}"#;
 const INTROSPECTION_SQL: &str = POSTGRESQL_INTROSPECTION_SQL;
-const HEADER_VALUE_ALLOW_ALL_ORIGINS: HeaderValue = HeaderValue::from_static("*");
-// CORS headers values
-const ACCESS_CONTROL_ALLOW_METHODS_VALUE: HeaderValue =
-    HeaderValue::from_static("GET, POST, PATCH, PUT, DELETE, OPTIONS");
-const ACCESS_CONTROL_MAX_AGE_VALUE: HeaderValue = HeaderValue::from_static("86400");
-const ACCESS_CONTROL_EXPOSE_HEADERS_VALUE: HeaderValue = HeaderValue::from_static(
-    "Content-Encoding, Content-Location, Content-Range, Content-Type, Date, Location, Server, Transfer-Encoding, Range-Unit",
-);
-const ACCESS_CONTROL_ALLOW_HEADERS_VALUE: HeaderValue = HeaderValue::from_static("Authorization");
 
 // A wrapper around the DbSchema that allows for self-referencing
 #[self_referencing]
@@ -151,8 +137,6 @@ pub struct ApiConfig {
     pub role_claim_key: String,
     #[serde(default, deserialize_with = "deserialize_comma_separated_option")]
     pub db_extra_search_path: Option<Vec<String>>,
-    #[serde(default, deserialize_with = "deserialize_comma_separated_option")]
-    pub server_cors_allowed_origins: Option<Vec<String>>,
 }
 
 // The DbSchemaCache is a cache of the ApiConfig and DbSchemaOwned for each endpoint
@@ -181,13 +165,7 @@ impl DbSchemaCache {
         }
     }
 
-    pub fn get_cached(
-        &self,
-        endpoint_id: &EndpointCacheKey,
-    ) -> Option<Arc<(ApiConfig, DbSchemaOwned)>> {
-        count_cache_outcome(CacheKind::Schema, self.0.get(endpoint_id))
-    }
-    pub async fn get_remote(
+    pub async fn get_cached_or_remote(
         &self,
         endpoint_id: &EndpointCacheKey,
         auth_header: &HeaderValue,
@@ -196,42 +174,47 @@ impl DbSchemaCache {
         ctx: &RequestContext,
         config: &'static ProxyConfig,
     ) -> Result<Arc<(ApiConfig, DbSchemaOwned)>, RestError> {
-        info!("db_schema cache miss for endpoint: {:?}", endpoint_id);
-        let remote_value = self
-            .internal_get_remote(auth_header, connection_string, client, ctx, config)
-            .await;
-        let (api_config, schema_owned) = match remote_value {
-            Ok((api_config, schema_owned)) => (api_config, schema_owned),
-            Err(e @ RestError::SchemaTooLarge) => {
-                // for the case where the schema is too large, we cache an empty dummy value
-                // all the other requests will fail without triggering the introspection query
-                let schema_owned = serde_json::from_str::<DbSchemaOwned>(EMPTY_JSON_SCHEMA)
-                    .map_err(|e| JsonDeserialize { source: e })?;
+        let cache_result = count_cache_outcome(CacheKind::Schema, self.0.get(endpoint_id));
+        match cache_result {
+            Some(v) => Ok(v),
+            None => {
+                info!("db_schema cache miss for endpoint: {:?}", endpoint_id);
+                let remote_value = self
+                    .get_remote(auth_header, connection_string, client, ctx, config)
+                    .await;
+                let (api_config, schema_owned) = match remote_value {
+                    Ok((api_config, schema_owned)) => (api_config, schema_owned),
+                    Err(e @ RestError::SchemaTooLarge) => {
+                        // for the case where the schema is too large, we cache an empty dummy value
+                        // all the other requests will fail without triggering the introspection query
+                        let schema_owned = serde_json::from_str::<DbSchemaOwned>(EMPTY_JSON_SCHEMA)
+                            .map_err(|e| JsonDeserialize { source: e })?;
 
-                let api_config = ApiConfig {
-                    db_schemas: vec![],
-                    db_anon_role: None,
-                    db_max_rows: None,
-                    db_allowed_select_functions: vec![],
-                    role_claim_key: String::new(),
-                    db_extra_search_path: None,
-                    server_cors_allowed_origins: None,
+                        let api_config = ApiConfig {
+                            db_schemas: vec![],
+                            db_anon_role: None,
+                            db_max_rows: None,
+                            db_allowed_select_functions: vec![],
+                            role_claim_key: String::new(),
+                            db_extra_search_path: None,
+                        };
+                        let value = Arc::new((api_config, schema_owned));
+                        count_cache_insert(CacheKind::Schema);
+                        self.0.insert(endpoint_id.clone(), value);
+                        return Err(e);
+                    }
+                    Err(e) => {
+                        return Err(e);
+                    }
                 };
                 let value = Arc::new((api_config, schema_owned));
                 count_cache_insert(CacheKind::Schema);
-                self.0.insert(endpoint_id.clone(), value);
-                return Err(e);
+                self.0.insert(endpoint_id.clone(), value.clone());
+                Ok(value)
             }
-            Err(e) => {
-                return Err(e);
-            }
-        };
-        let value = Arc::new((api_config, schema_owned));
-        count_cache_insert(CacheKind::Schema);
-        self.0.insert(endpoint_id.clone(), value.clone());
-        Ok(value)
+        }
     }
-    async fn internal_get_remote(
+    pub async fn get_remote(
         &self,
         auth_header: &HeaderValue,
         connection_string: &str,
@@ -548,7 +531,7 @@ pub(crate) async fn handle(
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, ApiError> {
     let result = handle_inner(cancel, config, &ctx, request, backend).await;
 
-    let response = match result {
+    let mut response = match result {
         Ok(r) => {
             ctx.set_success();
 
@@ -657,6 +640,9 @@ pub(crate) async fn handle(
         }
     };
 
+    response
+        .headers_mut()
+        .insert("Access-Control-Allow-Origin", HeaderValue::from_static("*"));
     Ok(response)
 }
 
@@ -736,37 +722,6 @@ async fn handle_inner(
     }
 }
 
-fn apply_common_cors_headers(
-    response: &mut Builder,
-    request_headers: &HeaderMap,
-    allowed_origins: Option<&Vec<String>>,
-) {
-    let request_origin = request_headers
-        .get(ORIGIN)
-        .map(|v| v.to_str().unwrap_or(""));
-
-    let response_allow_origin = match (request_origin, allowed_origins) {
-        (Some(or), Some(allowed_origins)) => {
-            if allowed_origins.iter().any(|o| o == or) {
-                Some(HeaderValue::from_str(or).unwrap_or(HEADER_VALUE_ALLOW_ALL_ORIGINS))
-            } else {
-                None
-            }
-        }
-        (Some(_), None) => Some(HEADER_VALUE_ALLOW_ALL_ORIGINS),
-        _ => None,
-    };
-    if let Some(h) = response.headers_mut() {
-        h.insert(
-            ACCESS_CONTROL_EXPOSE_HEADERS,
-            ACCESS_CONTROL_EXPOSE_HEADERS_VALUE,
-        );
-        if let Some(origin) = response_allow_origin {
-            h.insert(ACCESS_CONTROL_ALLOW_ORIGIN, origin);
-        }
-    }
-}
-
 #[allow(clippy::too_many_arguments)]
 async fn handle_rest_inner(
     config: &'static ProxyConfig,
@@ -778,6 +733,12 @@ async fn handle_rest_inner(
     jwt: String,
     backend: Arc<PoolingBackend>,
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, RestError> {
+    // validate the jwt token
+    let jwt_parsed = backend
+        .authenticate_with_jwt(ctx, &conn_info.user_info, jwt)
+        .await
+        .map_err(HttpConnError::from)?;
+
     let db_schema_cache =
         config
             .rest_config
@@ -793,83 +754,28 @@ async fn handle_rest_inner(
             message: "Failed to get endpoint cache key".to_string(),
         }))?;
 
+    let mut client = backend.connect_to_local_proxy(ctx, conn_info).await?;
+
     let (parts, originial_body) = request.into_parts();
 
-    // try and get the cached entry for this endpoint
-    // it contains the api config and the introspected db schema
-    let cached_entry = db_schema_cache.get_cached(&endpoint_cache_key);
-
-    let allowed_origins = cached_entry
-        .as_ref()
-        .and_then(|arc| arc.0.server_cors_allowed_origins.as_ref());
-
-    let mut response = Response::builder();
-    apply_common_cors_headers(&mut response, &parts.headers, allowed_origins);
-
-    // handle the OPTIONS request
-    if parts.method == Method::OPTIONS {
-        let allowed_headers = parts
-            .headers
-            .get(ACCESS_CONTROL_REQUEST_HEADERS)
-            .and_then(|a| a.to_str().ok())
-            .filter(|v| !v.is_empty())
-            .map_or_else(
-                || "Authorization".to_string(),
-                |v| format!("{v}, Authorization"),
-            );
-        return response
-            .status(StatusCode::OK)
-            .header(
-                ACCESS_CONTROL_ALLOW_METHODS,
-                ACCESS_CONTROL_ALLOW_METHODS_VALUE,
-            )
-            .header(ACCESS_CONTROL_MAX_AGE, ACCESS_CONTROL_MAX_AGE_VALUE)
-            .header(
-                ACCESS_CONTROL_ALLOW_HEADERS,
-                HeaderValue::from_str(&allowed_headers)
-                    .unwrap_or(ACCESS_CONTROL_ALLOW_HEADERS_VALUE),
-            )
-            .header(ALLOW, ACCESS_CONTROL_ALLOW_METHODS_VALUE)
-            .body(Empty::new().map_err(|x| match x {}).boxed())
-            .map_err(|e| {
-                RestError::SubzeroCore(InternalError {
-                    message: e.to_string(),
-                })
-            });
-    }
-
-    // validate the jwt token
-    let jwt_parsed = backend
-        .authenticate_with_jwt(ctx, &conn_info.user_info, jwt)
-        .await
-        .map_err(HttpConnError::from)?;
-
     let auth_header = parts
         .headers
         .get(AUTHORIZATION)
         .ok_or(RestError::SubzeroCore(InternalError {
             message: "Authorization header is required".to_string(),
         }))?;
-    let mut client = backend.connect_to_local_proxy(ctx, conn_info).await?;
 
-    let entry = match cached_entry {
-        Some(e) => e,
-        None => {
-            // if not cached, get the remote entry (will run the introspection query)
-            db_schema_cache
-                .get_remote(
-                    &endpoint_cache_key,
-                    auth_header,
-                    connection_string,
-                    &mut client,
-                    ctx,
-                    config,
-                )
-                .await?
-        }
-    };
+    let entry = db_schema_cache
+        .get_cached_or_remote(
+            &endpoint_cache_key,
+            auth_header,
+            connection_string,
+            &mut client,
+            ctx,
+            config,
+        )
+        .await?;
     let (api_config, db_schema_owned) = entry.as_ref();
-
     let db_schema = db_schema_owned.borrow_schema();
 
     let db_schemas = &api_config.db_schemas; // list of schemas available for the api
@@ -1093,8 +999,8 @@ async fn handle_rest_inner(
     let _metrics = client.metrics(ctx); // FIXME: is everything in the context set correctly?
 
     // send the request to the local proxy
-    let proxy_response = make_raw_local_proxy_request(&mut client, headers, req_body).await?;
-    let (response_parts, body) = proxy_response.into_parts();
+    let response = make_raw_local_proxy_request(&mut client, headers, req_body).await?;
+    let (parts, body) = response.into_parts();
 
     let max_response = config.http_config.max_response_size_bytes;
     let bytes = read_body_with_limit(body, max_response)
@@ -1103,7 +1009,7 @@ async fn handle_rest_inner(
 
     // if the response status is greater than 399, then it is an error
     // FIXME: check if there are other error codes or shapes of the response
-    if response_parts.status.as_u16() > 399 {
+    if parts.status.as_u16() > 399 {
         // turn this postgres error from the json into PostgresError
         let postgres_error = serde_json::from_slice(&bytes)
             .map_err(|e| RestError::SubzeroCore(JsonDeserialize { source: e }))?;
@@ -1269,7 +1175,7 @@ async fn handle_rest_inner(
         .boxed();
 
     // build the response
-    response = response
+    let mut response = Response::builder()
         .status(StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR))
         .header(CONTENT_TYPE, http_content_type);
 

test_runner/regress/test_compute_monitor.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import time
 from typing import TYPE_CHECKING
 
 from fixtures.metrics import parse_metrics
@@ -9,13 +10,13 @@ if TYPE_CHECKING:
     from fixtures.neon_fixtures import NeonEnv
 
 
-def test_compute_monitor(neon_simple_env: NeonEnv):
+def test_compute_monitor_downtime_calculation(neon_simple_env: NeonEnv):
     """
     Test that compute_ctl can detect Postgres going down (unresponsive) and
     reconnect when it comes back online. Also check that the downtime metrics
     are properly emitted.
     """
-    TEST_DB = "test_compute_monitor"
+    TEST_DB = "test_compute_monitor_downtime_calculation"
 
     env = neon_simple_env
     endpoint = env.endpoints.create_start("main")
@@ -68,3 +69,56 @@ def test_compute_monitor(neon_simple_env: NeonEnv):
 
     # Just a sanity check that we log the downtime info
     endpoint.log_contains("downtime_info")
+
+
+def test_compute_monitor_activity(neon_simple_env: NeonEnv):
+    """
+    Test compute monitor correctly detects user activity inside Postgres
+    and updates last_active timestamp in the /status response.
+    """
+    TEST_DB = "test_compute_monitor_activity_db"
+
+    env = neon_simple_env
+    endpoint = env.endpoints.create_start("main")
+
+    with endpoint.cursor() as cursor:
+        # Create a new database because `postgres` DB is excluded
+        # from activity monitoring.
+        cursor.execute(f"CREATE DATABASE {TEST_DB}")
+
+    client = endpoint.http_client()
+
+    prev_last_active = None
+
+    def check_last_active():
+        nonlocal prev_last_active
+
+        with endpoint.cursor(dbname=TEST_DB) as cursor:
+            # Execute some dummy query to generate 'activity'.
+            cursor.execute("SELECT * FROM generate_series(1, 10000)")
+
+        status = client.status()
+        assert status["last_active"] is not None
+        prev_last_active = status["last_active"]
+
+    wait_until(check_last_active)
+
+    assert prev_last_active is not None
+
+    # Sleep for everything to settle down. It's not strictly necessary,
+    # but should still remove any potential noise and/or prevent test from passing
+    # even if compute monitor is not working.
+    time.sleep(3)
+
+    with endpoint.cursor(dbname=TEST_DB) as cursor:
+        cursor.execute("SELECT * FROM generate_series(1, 10000)")
+
+    def check_last_active_updated():
+        nonlocal prev_last_active
+
+        status = client.status()
+        assert status["last_active"] is not None
+        assert status["last_active"] != prev_last_active
+        assert status["last_active"] > prev_last_active
+
+    wait_until(check_last_active_updated)