feat: add cache for providers and stac requests

geoengine/Settings-default.toml

@@ -188,6 +188,18 @@ size_in_mb = 1_000 # 1 GB
 # storage limit for collecting query results before insertion into the cache in 
 landing_zone_ratio = 0.1 # 10% of total cache size
 
+[stac_cache]
+# maximum number of TileFile entries for STAC query cache (per provider)
+max_tile_files = 10_000
+# TTL in seconds for STAC query cache entries
+ttl_secs = 3600
+
+[provider_cache]
+# maximum number of provider instances in the provider registry
+max_entries = 256
+# maximum idle time in seconds before evicting a cached provider
+max_idle_secs = 1800
+
 [wildlive]
 api_endpoint = "https://wildlive.senckenberg.de/api/"
 

geoengine/Settings-test.toml

@@ -40,3 +40,15 @@ enabled = false
 size_in_mb = 1_000 # 1 GB
 # storage limit for collecting query results before insertion into the cache in 
 landing_zone_ratio = 0.1 # 10% of total cache size
+
+[stac_cache]
+# maximum number of TileFile entries for STAC query cache (per provider)
+max_tile_files = 10_000
+# TTL in seconds for STAC query cache entries
+ttl_secs = 3600
+
+[provider_cache]
+# maximum number of provider instances in the provider registry
+max_entries = 256
+# maximum idle time in seconds before evicting a cached provider
+max_idle_secs = 1800

geoengine/services/src/api/model/services.rs

@@ -1054,6 +1054,13 @@ pub struct StacDataProviderDefinition {
     pub s3_config: Option<StacProviderS3Config>,
     pub time_dimension: TimeDimension,
     pub datasets: Vec<StacProviderDataset>,
+    /// Timeout in seconds for outgoing STAC API HTTP requests.
+    #[serde(default = "default_query_timeout")]
+    pub query_timeout_secs: i64,
+}
+
+fn default_query_timeout() -> i64 {
+    60
 }
 
 impl From<StacDataProviderDefinition>
@@ -1070,6 +1077,7 @@ impl From<StacDataProviderDefinition>
             s3_config: value.s3_config.map(Into::into),
             time_dimension: api_time_dimension_to_datatypes(value.time_dimension),
             datasets: value.datasets.into_iter().map(Into::into).collect(),
+            query_timeout_secs: value.query_timeout_secs,
         }
     }
 }
@@ -1089,6 +1097,7 @@ impl From<crate::datasets::external::stac::StacDataProviderDefinition>
             s3_config: value.s3_config.map(Into::into),
             time_dimension: datatypes_time_dimension_to_api(value.time_dimension),
             datasets: value.datasets.into_iter().map(Into::into).collect(),
+            query_timeout_secs: value.query_timeout_secs,
         }
     }
 }

geoengine/services/src/config.rs

@@ -486,6 +486,31 @@ impl ConfigElement for Cache {
     const KEY: &'static str = "cache";
 }
 
+#[derive(Debug, Deserialize, Clone, Copy)]
+pub struct StacCache {
+    /// Maximum total number of `TileFile` entries held in the STAC query cache
+    /// across all datasets.
+    pub max_tile_files: usize,
+    /// TTL in seconds for STAC query cache entries.
+    pub ttl_secs: u64,
+}
+
+impl ConfigElement for StacCache {
+    const KEY: &'static str = "stac_cache";
+}
+
+#[derive(Debug, Deserialize, Clone, Copy)]
+pub struct ProviderCache {
+    /// Maximum number of cached provider instances in the provider registry.
+    pub max_entries: usize,
+    /// Maximum idle time in seconds before a cached provider is evicted.
+    pub max_idle_secs: u64,
+}
+
+impl ConfigElement for ProviderCache {
+    const KEY: &'static str = "provider_cache";
+}
+
 #[derive(Clone, Debug, Deserialize)]
 pub struct Wildlive {
     pub api_endpoint: Url,

geoengine/services/src/contexts/migrations/current_schema.sql

@@ -904,7 +904,8 @@ CREATE TYPE "StacDataProviderDefinition" AS (
     collection_name text,
     s3_config "StacProviderS3Config",
     time_dimension "TimeDimension",
-    datasets "StacProviderDataset" []
+    datasets "StacProviderDataset" [],
+    query_timeout_secs bigint
 );
 
 CREATE TYPE "DataProviderDefinition" AS (

geoengine/services/src/contexts/migrations/database_migration.rs

@@ -206,6 +206,7 @@ pub mod tests {
     };
     use bb8_postgres::{PostgresConnectionManager, bb8::Pool};
     use geoengine_datatypes::{primitives::DateTime, test_data};
+    use std::sync::Arc;
     use tokio_postgres::NoTls;
 
     pub async fn create_migration_0015_snapshot<Tls>(
@@ -383,6 +384,7 @@ pub mod tests {
                 view: None,
                 roles: vec![RoleId::from_u128(0xb589a590_9c0c_4b55_9aa2_d178a5f42a78)],
             },
+            Arc::new(crate::layers::provider_registry::DataProviderRegistry::default()),
         );
 
         let projects = db

geoengine/services/src/contexts/migrations/migration_0028_stac_provider.sql

@@ -28,7 +28,8 @@ CREATE TYPE "StacDataProviderDefinition" AS (
     collection_name text,
     s3_config "StacProviderS3Config",
     time_dimension "TimeDimension",
-    datasets "StacProviderDataset" []
+    datasets "StacProviderDataset" [],
+    query_timeout_secs bigint
 );
 
 ALTER TYPE "DataProviderDefinition" ADD ATTRIBUTE

geoengine/services/src/contexts/postgres.rs

@@ -14,6 +14,7 @@ use crate::layers::add_from_directory::{
     add_datasets_from_directory, add_layer_collections_from_directory, add_layers_from_directory,
     add_providers_from_directory,
 };
+use crate::layers::provider_registry::DataProviderRegistry;
 use crate::machine_learning::error::MachineLearningError;
 use crate::quota::{QuotaTrackingFactory, initialize_quota_tracking};
 use crate::tasks::SimpleTaskManagerContext;
@@ -62,6 +63,7 @@ where
     pub(crate) pool: Pool<PostgresConnectionManager<Tls>>,
     volumes: Volumes,
     tile_cache: Arc<SharedCache>,
+    provider_registry: Arc<DataProviderRegistry>,
 }
 
 impl<Tls> PostgresContext<Tls>
@@ -85,7 +87,12 @@ where
 
         Self::create_pro_database(pool.get().await?).await?;
 
-        let db = PostgresDb::new(pool.clone(), UserSession::admin_session());
+        let provider_registry = Arc::new(DataProviderRegistry::default());
+        let db = PostgresDb::new(
+            pool.clone(),
+            UserSession::admin_session(),
+            provider_registry.clone(),
+        );
         let quota = initialize_quota_tracking(
             quota_config.mode,
             db,
@@ -103,6 +110,7 @@ where
             pool,
             volumes: Default::default(),
             tile_cache: Arc::new(SharedCache::test_default()),
+            provider_registry,
         })
     }
 
@@ -120,7 +128,12 @@ where
 
         Self::create_pro_database(pool.get().await?).await?;
 
-        let db = PostgresDb::new(pool.clone(), UserSession::admin_session());
+        let provider_registry = Arc::new(DataProviderRegistry::default());
+        let db = PostgresDb::new(
+            pool.clone(),
+            UserSession::admin_session(),
+            provider_registry.clone(),
+        );
         let quota = initialize_quota_tracking(
             quota_config.mode,
             db,
@@ -141,6 +154,7 @@ where
                 SharedCache::new(cache_config.size_in_mb, cache_config.landing_zone_ratio)
                     .expect("tile cache creation should work because the config is valid"),
             ),
+            provider_registry,
         })
     }
 
@@ -165,7 +179,12 @@ where
 
         let created_schema = Self::create_pro_database(pool.get().await?).await?;
 
-        let db = PostgresDb::new(pool.clone(), UserSession::admin_session());
+        let provider_registry = Arc::new(DataProviderRegistry::default());
+        let db = PostgresDb::new(
+            pool.clone(),
+            UserSession::admin_session(),
+            provider_registry.clone(),
+        );
         let quota = initialize_quota_tracking(
             quota_config.mode,
             db,
@@ -186,6 +205,7 @@ where
                 SharedCache::new(cache_config.size_in_mb, cache_config.landing_zone_ratio)
                     .expect("tile cache creation should work because the config is valid"),
             ),
+            provider_registry,
         };
 
         if created_schema {
@@ -342,7 +362,11 @@ where
     type ExecutionContext = ExecutionContextImpl<Self::GeoEngineDB>;
 
     fn db(&self) -> Self::GeoEngineDB {
-        PostgresDb::new(self.context.pool.clone(), self.session.clone())
+        PostgresDb::new(
+            self.context.pool.clone(),
+            self.session.clone(),
+            self.context.provider_registry.clone(),
+        )
     }
 
     fn tasks(&self) -> Self::TaskManager {
@@ -406,6 +430,7 @@ where
 {
     pub(crate) conn_pool: Pool<PostgresConnectionManager<Tls>>,
     pub(crate) session: UserSession,
+    pub(crate) provider_registry: Arc<DataProviderRegistry>,
 }
 
 impl<Tls> PostgresDb<Tls>
@@ -415,8 +440,16 @@ where
     <Tls as MakeTlsConnect<Socket>>::TlsConnect: Send,
     <<Tls as MakeTlsConnect<Socket>>::TlsConnect as TlsConnect<Socket>>::Future: Send,
 {
-    pub fn new(conn_pool: Pool<PostgresConnectionManager<Tls>>, session: UserSession) -> Self {
-        Self { conn_pool, session }
+    pub fn new(
+        conn_pool: Pool<PostgresConnectionManager<Tls>>,
+        session: UserSession,
+        provider_registry: Arc<DataProviderRegistry>,
+    ) -> Self {
+        Self {
+            conn_pool,
+            session,
+            provider_registry,
+        }
     }
 
     /// Check whether the namepsace of the given dataset is allowed for insertion