Improvements

Author: ardatan

.changeset/funny-rockets-speak.md

@@ -1,7 +1,13 @@
 ---
 'hive-console-sdk-rs': patch
-'hive-apollo-router-plugin': patch
+'hive-apollo-router-plugin': minor
 ---
 
 Extract Hive Console integration implementation into a new package `hive-console-sdk` which can
 be used by any Rust library for Hive Console integration
+
+It also includes a refactor to use less Mutexes like replacing `lru` + `Mutex` with the thread-safe `moka` package.
+Only one place that handles queueing uses `Mutex` now.
+
+
+Fixes a bug when Persisted Operations are enabled by default which should be explicitly enabled

packages/libraries/router/src/persisted_documents.rs

@@ -64,14 +64,14 @@ impl Plugin for PersistedDocumentsPlugin {
     }
 
     fn router_service(&self, service: router::BoxService) -> router::BoxService {
-        let enabled = self.configuration.enabled.unwrap_or(true);
+        let enabled = self.configuration.enabled.unwrap_or(false);
+        if enabled {
         let allow_arbitrary_documents = self
             .configuration
             .allow_arbitrary_documents
             .unwrap_or(false);
         let mgr_ref = self.persisted_documents_manager.clone();
 
-        if enabled {
             ServiceBuilder::new()
                 .checkpoint_async(move |req: router::Request| {
                     let mgr = mgr_ref.clone();

packages/libraries/router/src/usage.rs

@@ -207,11 +207,6 @@ impl Plugin for UsagePlugin {
             .clone()
             .or_else(|| env::var("HIVE_TARGET_ID").ok());
 
-        // In case target ID is specified in configuration, append it to the endpoint
-        if let Some(target_id) = &target_id {
-            endpoint.push_str(&format!("/{}", target_id));
-        }
-
         let default_config = Config::default();
         let user_config = init.config;
         let enabled = user_config
@@ -251,6 +246,7 @@ impl Plugin for UsagePlugin {
             let agent = Arc::new(UsageAgent::new(
                 token.expect("token is set"),
                 endpoint,
+                target_id,
                 buffer_size,
                 connect_timeout,
                 request_timeout,

packages/libraries/sdk-rs/src/agent.rs

@@ -141,6 +141,7 @@ impl UsageAgent {
     pub fn new(
         token: String,
         endpoint: String,
+        target_id: Option<String>,
         buffer_size: usize,
         connect_timeout: u64,
         request_timeout: u64,
@@ -164,6 +165,14 @@ impl UsageAgent {
             .build();
         let buffer = Arc::new(Buffer::new());
 
+        let mut endpoint = endpoint;
+
+        if token.starts_with("hvo1/") {
+            if let Some(target_id) = target_id {
+                endpoint.push_str(&format!("/{}", target_id));
+            }
+        }
+
         UsageAgent {
             buffer,
             processor,
@@ -198,43 +207,51 @@ impl UsageAgent {
                     );
                     continue;
                 }
-                Ok(operation) => {
-                    let hash = operation.hash;
-
-                    let client_name = non_empty_string(op.client_name);
-                    let client_version = non_empty_string(op.client_version);
-
-                    let metadata: Option<Metadata> =
-                        if client_name.is_some() || client_version.is_some() {
-                            Some(Metadata {
-                                client: Some(ClientInfo {
-                                    name: client_name,
-                                    version: client_version,
-                                }),
-                            })
-                        } else {
-                            None
-                        };
-                    report.operations.push(Operation {
-                        operationMapKey: hash.clone(),
-                        timestamp: op.timestamp,
-                        execution: Execution {
-                            ok: op.ok,
-                            duration: op.duration.as_nanos(),
-                            errorsTotal: op.errors,
-                        },
-                        persistedDocumentHash: op.persisted_document_hash,
-                        metadata,
-                    });
-                    if let std::collections::hash_map::Entry::Vacant(e) = report.map.entry(hash) {
-                        e.insert(OperationMapRecord {
-                            operation: operation.operation,
-                            operationName: non_empty_string(op.operation_name),
-                            fields: operation.coordinates,
+                Ok(operation) => match operation {
+                    Some(operation) => {
+                        let hash = operation.hash;
+
+                        let client_name = non_empty_string(op.client_name);
+                        let client_version = non_empty_string(op.client_version);
+
+                        let metadata: Option<Metadata> =
+                            if client_name.is_some() || client_version.is_some() {
+                                Some(Metadata {
+                                    client: Some(ClientInfo {
+                                        name: client_name,
+                                        version: client_version,
+                                    }),
+                                })
+                            } else {
+                                None
+                            };
+                        report.operations.push(Operation {
+                            operationMapKey: hash.clone(),
+                            timestamp: op.timestamp,
+                            execution: Execution {
+                                ok: op.ok,
+                                duration: op.duration.as_nanos(),
+                                errorsTotal: op.errors,
+                            },
+                            persistedDocumentHash: op.persisted_document_hash,
+                            metadata,
                         });
+                        if let std::collections::hash_map::Entry::Vacant(e) = report.map.entry(hash)
+                        {
+                            e.insert(OperationMapRecord {
+                                operation: operation.operation,
+                                operationName: non_empty_string(op.operation_name),
+                                fields: operation.coordinates,
+                            });
+                        }
+                        report.size += 1;
                     }
-                    report.size += 1;
-                }
+                    None => {
+                        tracing::debug!(
+                            "Dropping operation (phase: PROCESSING): probably introspection query"
+                        );
+                    }
+                },
             }
         }
 

packages/libraries/sdk-rs/src/graphql.rs

@@ -849,7 +849,7 @@ pub struct ProcessedOperation {
 }
 
 pub struct OperationProcessor {
-    cache: Cache<String, ProcessedOperation>,
+    cache: Cache<String, Option<ProcessedOperation>>,
 }
 
 impl Default for OperationProcessor {
@@ -869,18 +869,17 @@ impl OperationProcessor {
         &self,
         query: &str,
         schema: &SchemaDocument<'static, String>,
-    ) -> Result<ProcessedOperation, String> {
-        let key = query.to_string();
-        if let Some(cached) = self.cache.get(&key) {
-            Ok(cached)
+    ) -> Result<Option<ProcessedOperation>, String> {
+        if self.cache.contains_key(query) {
+            let entry = self
+                .cache
+                .get(query)
+                .expect("Unable to acquire Cache in OperationProcessor.process");
+            Ok(entry.clone())
         } else {
             let result = self.transform(query, schema)?;
-            if let Some(result) = result {
-                self.cache.insert(key, result.clone());
-                Ok(result)
-            } else {
-                Err("Unable to acquire Cache in OperationProcessor.process".to_string())
-            }
+            self.cache.insert(query.to_string(), result.clone());
+            Ok(result)
         }
     }