Merge pull request #278 from MikeFalcon77/feature/gts_resolver

feat(modkit): add GtsPluginSelector and ThrottledLog for lazy plugin resolution

Author: MikeFalcon77

examples/plugin-modules/tenant_resolver/tenant_resolver-gw/src/api/rest/error.rs

@@ -12,10 +12,10 @@ impl From<DomainError> for Problem {
                 "PluginNotFound",
                 format!("No plugin instances found for vendor '{vendor}'"),
             ),
-            DomainError::PluginClientNotFound { gts_id } => Problem::new(
-                StatusCode::INTERNAL_SERVER_ERROR,
-                "PluginClientNotFound",
-                format!("Plugin client not found in ClientHub for '{gts_id}'"),
+            DomainError::PluginUnavailable { gts_id, reason } => Problem::new(
+                StatusCode::SERVICE_UNAVAILABLE,
+                "PluginUnavailable",
+                format!("Plugin '{gts_id}' is not available: {reason}"),
             ),
             DomainError::InvalidPluginInstance { gts_id, reason } => Problem::new(
                 StatusCode::BAD_REQUEST,

examples/plugin-modules/tenant_resolver/tenant_resolver-gw/src/domain/error.rs

@@ -10,8 +10,8 @@ pub enum DomainError {
     #[error("invalid plugin instance content for '{gts_id}': {reason}")]
     InvalidPluginInstance { gts_id: String, reason: String },
 
-    #[error("plugin client not registered in ClientHub for '{gts_id}'")]
-    PluginClientNotFound { gts_id: String },
+    #[error("plugin not available for '{gts_id}': {reason}")]
+    PluginUnavailable { gts_id: String, reason: String },
 
     #[error("tenant not found: {0}")]
     TenantNotFound(String),
@@ -33,6 +33,10 @@ impl From<tenant_resolver_sdk::TenantResolverError> for DomainError {
             TenantResolverError::PermissionDenied(msg) | TenantResolverError::Unauthorized(msg) => {
                 Self::PermissionDenied(msg)
             }
+            TenantResolverError::ServiceUnavailable(msg) => Self::PluginUnavailable {
+                gts_id: "unknown".to_owned(),
+                reason: msg,
+            },
             TenantResolverError::Internal(msg) => Self::Internal(msg),
         }
     }
@@ -65,8 +69,8 @@ impl From<DomainError> for tenant_resolver_sdk::TenantResolverError {
             DomainError::InvalidPluginInstance { gts_id, reason } => {
                 Self::Internal(format!("invalid plugin instance '{gts_id}': {reason}"))
             }
-            DomainError::PluginClientNotFound { gts_id } => {
-                Self::Internal(format!("plugin client not registered for '{gts_id}'"))
+            DomainError::PluginUnavailable { gts_id, reason } => {
+                Self::ServiceUnavailable(format!("plugin not available for '{gts_id}': {reason}"))
             }
             DomainError::TenantNotFound(msg) => Self::NotFound(msg),
             DomainError::PermissionDenied(msg) => Self::PermissionDenied(msg),

examples/plugin-modules/tenant_resolver/tenant_resolver-gw/src/domain/service.rs

@@ -4,16 +4,18 @@
 //! after `types_registry` has switched to ready mode.
 
 use std::sync::Arc;
+use std::time::Duration;
 
 use modkit::client_hub::{ClientHub, ClientScope};
 use modkit::gts::BaseModkitPluginV1;
+use modkit::plugins::GtsPluginSelector;
+use modkit::telemetry::ThrottledLog;
 use modkit_odata::{ODataQuery, Page};
 use modkit_security::SecurityContext;
 use tenant_resolver_sdk::{
     AccessOptions, GetParentsResponse, Tenant, TenantFilter, TenantResolverPluginClient,
     TenantResolverPluginSpecV1,
 };
-use tokio::sync::OnceCell;
 use tracing::info;
 use types_registry_sdk::{GtsEntity, ListQuery, TypesRegistryClient};
 
@@ -22,21 +24,20 @@ use types_registry_sdk::{GtsEntity, ListQuery, TypesRegistryClient};
 
 use crate::domain::error::DomainError;
 
-/// Cached result of plugin resolution.
-struct ResolvedPlugin {
-    gts_id: String,
-    scope: ClientScope,
-}
+/// Throttle interval for unavailable plugin warnings.
+const UNAVAILABLE_LOG_THROTTLE: Duration = Duration::from_secs(10);
 
 /// Tenant Resolver Gateway service.
 ///
 /// Holds a reference to `ClientHub` and the configured vendor.
-/// Plugin discovery is lazy and cached via `OnceCell`.
+/// Plugin discovery is lazy and cached via `GtsPluginSelector`.
 pub struct Service {
     hub: Arc<ClientHub>,
     vendor: String,
-    /// Lazily resolved plugin (cached after first call).
-    resolved: OnceCell<ResolvedPlugin>,
+    /// Shared selector for plugin instance IDs.
+    selector: GtsPluginSelector,
+    /// Throttle for plugin unavailable warnings.
+    unavailable_log_throttle: ThrottledLog,
 }
 
 impl Service {
@@ -46,7 +47,8 @@ impl Service {
         Self {
             hub,
             vendor,
-            resolved: OnceCell::new(),
+            selector: GtsPluginSelector::new(),
+            unavailable_log_throttle: ThrottledLog::new(UNAVAILABLE_LOG_THROTTLE),
         }
     }
 
@@ -55,21 +57,32 @@ impl Service {
     /// On first call, queries `types_registry` to find the plugin instance
     /// matching the configured vendor. Result is cached for subsequent calls.
     async fn get_plugin(&self) -> Result<Arc<dyn TenantResolverPluginClient>, DomainError> {
-        let resolved = self
-            .resolved
-            .get_or_try_init(|| self.resolve_plugin())
-            .await?;
+        let instance_id = self.selector.get_or_init(|| self.resolve_plugin()).await?;
+        let scope = ClientScope::gts_id(instance_id.as_ref());
 
-        self.hub
-            .get_scoped::<dyn TenantResolverPluginClient>(&resolved.scope)
-            .map_err(|_| DomainError::PluginClientNotFound {
-                gts_id: resolved.gts_id.clone(),
+        if let Some(client) = self
+            .hub
+            .try_get_scoped::<dyn TenantResolverPluginClient>(&scope)
+        {
+            Ok(client)
+        } else {
+            if self.unavailable_log_throttle.should_log() {
+                tracing::warn!(
+                    plugin_gts_id = %instance_id,
+                    vendor = %self.vendor,
+                    "Plugin client not registered yet"
+                );
+            }
+            Err(DomainError::PluginUnavailable {
+                gts_id: instance_id.to_string(),
+                reason: "client not registered yet".into(),
             })
+        }
     }
 
     /// Resolves the plugin instance from `types_registry`.
     #[tracing::instrument(skip_all, fields(vendor = %self.vendor))]
-    async fn resolve_plugin(&self) -> Result<ResolvedPlugin, DomainError> {
+    async fn resolve_plugin(&self) -> Result<String, DomainError> {
         info!("Resolving tenant resolver plugin");
 
         let registry = self
@@ -90,8 +103,7 @@ impl Service {
         let gts_id = choose_plugin_instance(&self.vendor, &instances)?;
         info!(plugin_gts_id = %gts_id, "Selected tenant resolver plugin instance");
 
-        let scope = ClientScope::gts_id(&gts_id);
-        Ok(ResolvedPlugin { gts_id, scope })
+        Ok(gts_id)
     }
 
     /// Returns the root tenant.

examples/plugin-modules/tenant_resolver/tenant_resolver-sdk/src/error.rs

@@ -12,6 +12,9 @@ pub enum TenantResolverError {
     #[error("not found: {0}")]
     NotFound(String),
 
+    #[error("service unavailable: {0}")]
+    ServiceUnavailable(String),
+
     #[error("internal error: {0}")]
     Internal(String),
 }

libs/modkit/src/client_hub.rs

@@ -216,6 +216,23 @@ impl ClientHub {
         })
     }
 
+    /// Try to fetch a scoped client by interface type `T` and scope.
+    ///
+    /// Returns `None` if not found or if the stored type doesn't match.
+    pub fn try_get_scoped<T>(&self, scope: &ClientScope) -> Option<Arc<T>>
+    where
+        T: ?Sized + Send + Sync + 'static,
+    {
+        let key = ScopedKey {
+            type_key: TypeKey::of::<T>(),
+            scope: scope.clone(),
+        };
+        let r = self.scoped_map.read();
+        let boxed = r.get(&key)?;
+
+        boxed.downcast_ref::<Arc<T>>().cloned()
+    }
+
     /// Remove a client by interface type; returns the removed client if it was present.
     pub fn remove<T>(&self) -> Option<Arc<T>>
     where
@@ -355,4 +372,23 @@ mod tests {
         assert_eq!(&*hub.get::<str>().unwrap(), "global");
         assert_eq!(&*hub.get_scoped::<str>(&scope).unwrap(), "scoped");
     }
+
+    #[test]
+    fn try_get_scoped_returns_some_on_hit() {
+        let hub = ClientHub::new();
+        let scope = ClientScope::gts_id("gts.x.core.modkit.plugins.v1~x.core.tenant_resolver.plugin.v1~contoso.app._.plugin.v1.0");
+        hub.register_scoped::<str>(scope.clone(), Arc::from("scoped"));
+
+        let got = hub.try_get_scoped::<str>(&scope);
+        assert_eq!(got.as_deref(), Some("scoped"));
+    }
+
+    #[test]
+    fn try_get_scoped_returns_none_on_miss() {
+        let hub = ClientHub::new();
+        let scope = ClientScope::gts_id("gts.x.core.modkit.plugins.v1~x.core.tenant_resolver.plugin.v1~fabrikam.app._.plugin.v1.0");
+
+        let got = hub.try_get_scoped::<str>(&scope);
+        assert!(got.is_none());
+    }
 }

libs/modkit/src/lib.rs

@@ -115,6 +115,7 @@ pub mod telemetry;
 
 pub mod backends;
 pub mod lifecycle;
+pub mod plugins;
 pub mod runtime;
 
 // Error catalog runtime support
@@ -142,6 +143,7 @@ pub use backends::{
     OopModuleConfig, OopSpawnConfig,
 };
 pub use lifecycle::{Lifecycle, Runnable, Status, StopReason, WithLifecycle};
+pub use plugins::GtsPluginSelector;
 pub use runtime::{
     DbOptions, Endpoint, ModuleInstance, ModuleManager, OopModuleSpawnConfig, OopSpawnOptions,
     RunOptions, ShutdownOptions, run,