chore: Initial Job Worker revamp. (#796)

* Decouple worker tasks to run independently of the gateway. Fixes #720.
* Reduce GatewayManager queries in the gateway handlers on request.
  Fixes #723.
* Introduce polling helpers in tests ensuring async name record publication
  occurs before continuing the test.

rust/noosphere-cli/Cargo.toml

@@ -64,7 +64,7 @@ vergen-pretty = { workspace = true }
 cid = { workspace = true }
 symlink = { workspace = true }
 pathdiff = { workspace = true }
-subtext = "0.3.2"
+subtext = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

rust/noosphere-cli/src/native/commands/serve.rs

@@ -28,9 +28,16 @@ pub async fn serve(
         .author()
         .did()
         .await?;
-    let manager = SingleTenantGatewayManager::new(sphere_context, counterpart.clone()).await?;
+    let manager = SingleTenantGatewayManager::new(
+        sphere_context,
+        counterpart.clone(),
+        ipfs_api,
+        name_resolver_api,
+        cors_origin,
+    )
+    .await?;
 
-    let gateway = Gateway::new(manager, ipfs_api, name_resolver_api, cors_origin)?;
+    let gateway = Gateway::new(manager)?;
 
     info!(
         r#"A geist is summoned to manage local sphere {}

rust/noosphere-cli/src/native/helpers/workspace.rs

@@ -56,11 +56,17 @@ async fn start_gateway_for_workspace(
     let client_sphere_identity = client_sphere_identity.to_owned();
     let ns_url = ns_url.clone();
     let ipfs_url = ipfs_url.clone();
-    let manager =
-        SingleTenantGatewayManager::new(gateway_sphere_context, client_sphere_identity).await?;
+    let manager = SingleTenantGatewayManager::new(
+        gateway_sphere_context,
+        client_sphere_identity,
+        ipfs_url,
+        ns_url,
+        None,
+    )
+    .await?;
 
     let join_handle = tokio::spawn(async move {
-        let gateway = Gateway::new(manager, ipfs_url, ns_url, None).unwrap();
+        let gateway = Gateway::new(manager).unwrap();
         gateway.start(gateway_listener).await.unwrap()
     });
 

rust/noosphere-core/src/data/address.rs

@@ -151,7 +151,6 @@ impl LinkRecord {
         let facts = if let Some(facts) = self.0.facts() {
             facts
         } else {
-            warn!("No facts found in the link record!");
             return None;
         };
 

rust/noosphere-gateway/src/extractors/authority.rs

@@ -1,3 +1,7 @@
+use crate::{
+    extractors::{map_bad_request, GatewayScope},
+    GatewayManager,
+};
 use anyhow::Result;
 use async_trait::async_trait;
 use axum::{
@@ -10,14 +14,11 @@ use axum_extra::{
 };
 use noosphere_core::{
     api::headers::{self as noosphere_headers},
-    authority::{SphereAbility, SphereReference, SPHERE_SEMANTICS},
+    authority::{generate_capability, SphereAbility, SPHERE_SEMANTICS},
     context::HasMutableSphereContext,
-    data::Did,
 };
 use noosphere_storage::Storage;
-use noosphere_ucan::capability::CapabilityView;
-
-use crate::extractors::map_bad_request;
+use std::{marker::PhantomData, sync::Arc};
 
 /// Represents the scope of a gateway request's authorization and sphere
 /// access.
@@ -26,29 +27,32 @@ use crate::extractors::map_bad_request;
 /// represented by its `ucan` headers. Any request handler can use a [GatewayAuthority]
 /// to test if a required capability is satisfied by the authorization
 /// presented by the maker of the request.
-pub struct GatewayAuthority {
+pub struct GatewayAuthority<M, C, S> {
     bearer: Bearer,
     ucans: noosphere_headers::Ucan,
+    manager: Arc<M>,
+    sphere_context_marker: PhantomData<C>,
+    storage_marker: PhantomData<S>,
 }
 
-impl GatewayAuthority {
-    pub async fn try_authorize<C, S>(
+impl<M, C, S> GatewayAuthority<M, C, S>
+where
+    M: GatewayManager<C, S> + 'static,
+    C: HasMutableSphereContext<S>,
+    S: Storage + 'static,
+{
+    pub async fn try_authorize(
         &self,
-        sphere_context: &mut C,
-        counterpart: &Did,
-        capability: &CapabilityView<SphereReference, SphereAbility>,
-    ) -> Result<(), StatusCode>
-    where
-        C: HasMutableSphereContext<S>,
-        S: Storage + 'static,
-    {
-        let db = {
-            let sphere_context: C::SphereContext = sphere_context
-                .sphere_context()
-                .await
-                .map_err(map_bad_request)?;
-            sphere_context.db().clone()
-        };
+        gateway_scope: &GatewayScope<C, S>,
+        required_ability: SphereAbility,
+    ) -> Result<C, StatusCode> {
+        let counterpart_str = gateway_scope.counterpart.as_str();
+        let capability = generate_capability(counterpart_str, required_ability);
+        let db = self
+            .manager
+            .ucan_store()
+            .await
+            .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
 
         let proof_chain = self
             .ucans
@@ -60,11 +64,15 @@ impl GatewayAuthority {
 
         for capability_info in capability_infos {
             trace!("Checking capability: {:?}", capability_info.capability);
-            if capability_info.originators.contains(counterpart.as_str())
-                && capability_info.capability.enables(capability)
+            if capability_info.originators.contains(counterpart_str)
+                && capability_info.capability.enables(&capability)
             {
                 debug!("Authorized!");
-                return Ok(());
+                return self
+                    .manager
+                    .sphere_context(&gateway_scope.counterpart)
+                    .await
+                    .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR);
             }
         }
 
@@ -73,13 +81,18 @@ impl GatewayAuthority {
 }
 
 #[async_trait]
-impl<State> FromRequestParts<State> for GatewayAuthority
+impl<M, C, S> FromRequestParts<Arc<M>> for GatewayAuthority<M, C, S>
 where
-    State: Send + Sync,
+    M: GatewayManager<C, S> + 'static,
+    C: HasMutableSphereContext<S>,
+    S: Storage + 'static,
 {
     type Rejection = StatusCode;
 
-    async fn from_request_parts(parts: &mut Parts, state: &State) -> Result<Self, Self::Rejection> {
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &Arc<M>,
+    ) -> Result<Self, Self::Rejection> {
         let TypedHeader(Authorization(bearer)) =
             TypedHeader::<Authorization<Bearer>>::from_request_parts(parts, state)
                 .await
@@ -91,6 +104,13 @@ where
                 .map_err(map_bad_request)?
                 .0;
 
-        Ok(GatewayAuthority { bearer, ucans })
+        let manager = state.to_owned();
+        Ok(GatewayAuthority {
+            bearer,
+            ucans,
+            manager,
+            sphere_context_marker: PhantomData,
+            storage_marker: PhantomData,
+        })
     }
 }

rust/noosphere-gateway/src/extractors/mod.rs

@@ -3,12 +3,10 @@
 mod authority;
 mod cbor;
 mod scope;
-mod sphere_extractor;
 
 pub use authority::*;
 pub use cbor::*;
 pub use scope::*;
-pub use sphere_extractor::*;
 
 pub(crate) fn map_bad_request<E: std::fmt::Debug>(error: E) -> axum::http::StatusCode {
     tracing::error!("{:?}", error);

rust/noosphere-gateway/src/extractors/scope.rs

@@ -13,21 +13,34 @@ use noosphere_storage::Storage;
 #[cfg(doc)]
 use noosphere_core::context::SphereContext;
 
-use super::map_bad_request;
-
 /// Represents the scope of a gateway request as a counterpart [Did],
 /// and the corresponding managed sphere's author/device key,
 /// the gateway identity.
 ///
 /// Extracting a [GatewayScope] is efficient, and does not open
 /// a [SphereContext].
+#[derive(Clone)]
 pub struct GatewayScope<C, S> {
+    /// [Did] of the client counterpart sphere.
     pub counterpart: Did,
+    /// [Did] of the author of the managed gateway sphere.
     pub gateway_identity: Did,
     sphere_context_marker: PhantomData<C>,
     storage_marker: PhantomData<S>,
 }
 
+impl<C, S> GatewayScope<C, S> {
+    /// Creates a new [GatewayScope].
+    pub fn new(gateway_identity: Did, counterpart: Did) -> Self {
+        Self {
+            gateway_identity,
+            counterpart,
+            sphere_context_marker: PhantomData,
+            storage_marker: PhantomData,
+        }
+    }
+}
+
 #[async_trait]
 impl<M, C, S> FromRequestParts<Arc<M>> for GatewayScope<C, S>
 where
@@ -41,17 +54,7 @@ where
         parts: &mut Parts,
         state: &Arc<M>,
     ) -> Result<Self, Self::Rejection> {
-        let counterpart = state.extract_counterpart(parts).await?;
-        let gateway_identity = state
-            .get_gateway_identity(&counterpart)
-            .await
-            .map_err(map_bad_request)?;
-
-        Ok(GatewayScope {
-            counterpart,
-            gateway_identity,
-            sphere_context_marker: PhantomData,
-            storage_marker: PhantomData,
-        })
+        let (gateway_identity, counterpart) = state.gateway_scope(parts).await?;
+        Ok(GatewayScope::new(gateway_identity, counterpart))
     }
 }