Run some services concurrently

This implements a general way to run some of our services concurrently,
notably _just_ the ones that are supposed to be run on one node manager
specifically, like `Read` and `Write`.

This involved some slightly fiddly async code, but the result isn't too
bad, and allows us to generalize a few things that we're not very
consistent about, like setting `current_node_manager_index` between
calls.

Doing this properly required cloning the `RequestContext`. I'm concerned
about the cost of this clone, which up until now we've been pretty lazy
about, so I moved the shared part into an `inner` field behind an `Arc`.
To avoid this breaking every server implementation I added a
`Deref<RequestContextInner>` to `RequestContext` which is a bad
practice, but does seem alright in this case.

Author: einarmo

TODO.md

@@ -11,5 +11,4 @@ This is a list of things that are known to be missing, or ideas that could be im
  - Write some "bad ideas" servers, it would be nice to showcase how flexible this is.
  - Write a framework for method calls. The foundation for this has been laid with `TryFromVariant`, if we really wanted to we could use clever trait magic to let users simply define a rust method that takes in values that each implement a trait `MethodArg`, with a blanket impl for `TryFromVariant`, and return a tuple of results. Could be really powerful, but methods are a little niche.
  - Implement `Query`. I never got around to this, because the service is just so complex. Currently there is no way to actually implement it, since it won't work unless _all_ node managers implement it, and the core node managers don't.
- - Look into running certain services concurrently. Currently they are sequential because that makes everything much simpler, but the services that don't have any cross node-manager interaction could run on all node managers concurrently.
  - Tracing and detailed logging in the client.

async-opcua-server/src/node_manager/context.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{ops::Deref, sync::Arc};
 
 use crate::{
     authenticator::{AuthManager, UserToken},
@@ -83,9 +83,27 @@ impl<R: RawRwLock, T: TypeTree> TypeTreeReadContext for RwLockReadGuard<'_, R, T
 }
 
 #[derive(Clone)]
-/// Context object passed during writes, contains useful context the node
+/// Context object passed during requests, contains useful context the node
 /// managers can use to execute service calls.
 pub struct RequestContext {
+    /// Index of the current node manager.
+    pub current_node_manager_index: usize,
+    /// Inner request context object, shared between service calls.
+    pub(crate) inner: Arc<RequestContextInner>,
+}
+
+// This isn't ideal, but the breaking change from having every field on
+// RequestContext be private is too big for now.
+impl Deref for RequestContext {
+    type Target = RequestContextInner;
+
+    fn deref(&self) -> &RequestContextInner {
+        &self.inner
+    }
+}
+
+/// Inner request context object, shared between service calls.
+pub struct RequestContextInner {
     /// The full session object for the session responsible for this service call.
     pub session: Arc<RwLock<Session>>,
     /// The session ID for the session responsible for this service call.
@@ -94,8 +112,6 @@ pub struct RequestContext {
     pub authenticator: Arc<dyn AuthManager>,
     /// The current user token.
     pub token: UserToken,
-    /// Index of the current node manager.
-    pub current_node_manager_index: usize,
     /// Global type tree object.
     pub type_tree: Arc<RwLock<DefaultTypeTree>>,
     /// Wrapper to get a type tree
@@ -112,6 +128,42 @@ impl RequestContext {
     pub fn get_type_tree_for_user<'a>(&'a self) -> Box<dyn TypeTreeReadContext + 'a> {
         self.type_tree_getter.get_type_tree_for_user(self)
     }
+
+    /// Get the session object responsible for this service call.
+    pub fn session(&self) -> &RwLock<Session> {
+        &self.session
+    }
+
+    /// Get the session ID for the session responsible for this service call.
+    pub fn session_id(&self) -> u32 {
+        self.session_id
+    }
+
+    /// Get the global `AuthManager` object.
+    pub fn authenticator(&self) -> &dyn AuthManager {
+        self.authenticator.as_ref()
+    }
+
+    /// Get the current user token.
+    pub fn user_token(&self) -> &UserToken {
+        &self.token
+    }
+
+    /// Get the global type tree object. If your server needs per-user type trees,
+    /// use `get_type_tree_for_user` instead.
+    pub fn type_tree(&self) -> &RwLock<DefaultTypeTree> {
+        &self.type_tree
+    }
+
+    /// Get the subscription cache.
+    pub fn subscriptions(&self) -> &SubscriptionCache {
+        &self.subscriptions
+    }
+
+    /// Get the server info object.
+    pub fn info(&self) -> &ServerInfo {
+        &self.info
+    }
 }
 
 /// Resolve a list of references.

async-opcua-server/src/node_manager/history.rs

@@ -4,8 +4,8 @@ use opcua_types::{
     match_extension_object_owned, ByteString, DeleteAtTimeDetails, DeleteEventDetails,
     DeleteRawModifiedDetails, DynEncodable, ExtensionObject, HistoryData, HistoryEvent,
     HistoryModifiedData, HistoryReadResult, HistoryReadValueId, HistoryUpdateResult, NodeId,
-    NumericRange, QualifiedName, ReadAnnotationDataDetails, ReadAtTimeDetails, ReadEventDetails,
-    ReadProcessedDetails, ReadRawModifiedDetails, StatusCode, UpdateDataDetails,
+    NumericRange, ObjectId, QualifiedName, ReadAnnotationDataDetails, ReadAtTimeDetails,
+    ReadEventDetails, ReadProcessedDetails, ReadRawModifiedDetails, StatusCode, UpdateDataDetails,
     UpdateEventDetails, UpdateStructureDataDetails,
 };
 
@@ -220,7 +220,13 @@ impl HistoryUpdateNode {
         self.operation_results = operation_results;
     }
 
-    pub(crate) fn into_result(self) -> HistoryUpdateResult {
+    pub(crate) fn into_result(mut self) -> HistoryUpdateResult {
+        // Special case reads on the server node, to return a proper error if no node manager supports it...
+        if self.details.node_id() == &ObjectId::Server
+            && matches!(self.status, StatusCode::BadNodeIdUnknown)
+        {
+            self.status = StatusCode::BadHistoryOperationUnsupported;
+        }
         HistoryUpdateResult {
             diagnostic_infos: None,
             status_code: self.status,

async-opcua-server/src/node_manager/mod.rs

@@ -40,7 +40,10 @@ use super::{
 pub use {
     attributes::{ParsedReadValueId, ParsedWriteValue, ReadNode, WriteNode},
     build::NodeManagerBuilder,
-    context::{RequestContext, TypeTreeForUser, TypeTreeForUserStatic, TypeTreeReadContext},
+    context::{
+        RequestContext, RequestContextInner, TypeTreeForUser, TypeTreeForUserStatic,
+        TypeTreeReadContext,
+    },
     history::{HistoryNode, HistoryResult, HistoryUpdateDetails, HistoryUpdateNode},
     method::MethodCall,
     monitored_items::{MonitoredItemRef, MonitoredItemUpdateRef},

async-opcua-server/src/session/message_handler.rs

@@ -9,7 +9,7 @@ use tracing::{debug, warn};
 use crate::{
     authenticator::UserToken,
     info::ServerInfo,
-    node_manager::{get_namespaces_for_user, NodeManagers, RequestContext},
+    node_manager::{get_namespaces_for_user, NodeManagers, RequestContext, RequestContextInner},
     session::services,
     subscriptions::{PendingPublish, SubscriptionCache},
 };
@@ -121,15 +121,17 @@ impl<T> Request<T> {
     /// Get a request context object from this request.
     pub(super) fn context(&self) -> RequestContext {
         RequestContext {
-            session: self.session.clone(),
-            authenticator: self.info.authenticator.clone(),
-            token: self.token.clone(),
             current_node_manager_index: 0,
-            type_tree: self.info.type_tree.clone(),
-            type_tree_getter: self.info.type_tree_getter.clone(),
-            subscriptions: self.subscriptions.clone(),
-            session_id: self.session_id,
-            info: self.info.clone(),
+            inner: Arc::new(RequestContextInner {
+                session: self.session.clone(),
+                authenticator: self.info.authenticator.clone(),
+                token: self.token.clone(),
+                type_tree: self.info.type_tree.clone(),
+                type_tree_getter: self.info.type_tree_getter.clone(),
+                subscriptions: self.subscriptions.clone(),
+                session_id: self.session_id,
+                info: self.info.clone(),
+            }),
         }
     }
 }
@@ -366,15 +368,17 @@ impl MessageHandler {
         }
 
         let mut context = RequestContext {
-            session,
-            session_id,
-            authenticator: self.info.authenticator.clone(),
-            token,
             current_node_manager_index: 0,
-            type_tree: self.info.type_tree.clone(),
-            subscriptions: self.subscriptions.clone(),
-            info: self.info.clone(),
-            type_tree_getter: self.info.type_tree_getter.clone(),
+            inner: Arc::new(RequestContextInner {
+                session,
+                session_id,
+                authenticator: self.info.authenticator.clone(),
+                token,
+                type_tree: self.info.type_tree.clone(),
+                subscriptions: self.subscriptions.clone(),
+                info: self.info.clone(),
+                type_tree_getter: self.info.type_tree_getter.clone(),
+            }),
         };
 
         // Ignore the result
@@ -397,15 +401,17 @@ impl MessageHandler {
         token: UserToken,
     ) -> NamespaceMap {
         let ctx = RequestContext {
-            session,
-            authenticator: self.info.authenticator.clone(),
-            token,
             current_node_manager_index: 0,
-            type_tree: self.info.type_tree.clone(),
-            type_tree_getter: self.info.type_tree_getter.clone(),
-            subscriptions: self.subscriptions.clone(),
-            session_id,
-            info: self.info.clone(),
+            inner: Arc::new(RequestContextInner {
+                session,
+                session_id,
+                authenticator: self.info.authenticator.clone(),
+                token,
+                type_tree: self.info.type_tree.clone(),
+                subscriptions: self.subscriptions.clone(),
+                info: self.info.clone(),
+                type_tree_getter: self.info.type_tree_getter.clone(),
+            }),
         };
         get_namespaces_for_user(&ctx, &self.node_managers)
     }