reorganize channels & comms module

Author: robherley

crates/funcgg-runtime/src/http.rs renamed to crates/funcgg-runtime/src/comms.rs

@@ -2,6 +2,7 @@ use anyhow::Result;
 use http::StatusCode;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use tokio::sync::{mpsc, oneshot};
 use uuid::Uuid;
 
 // TODO: see if we can get rid of these "in the middle" types
@@ -34,3 +35,10 @@ impl Response {
             .insert("X-FUNC-GG-REQUEST-ID".into(), request_id.to_string());
     }
 }
+
+#[derive(Debug)]
+pub struct Channels {
+    pub incoming_body_rx: mpsc::Receiver<Result<bytes::Bytes, String>>,
+    pub outgoing_body_tx: mpsc::Sender<bytes::Bytes>,
+    pub response_tx: oneshot::Sender<Response>,
+}

crates/funcgg-runtime/src/ext/mod.rs

@@ -10,8 +10,8 @@ use std::cell::RefCell;
 use std::rc::Rc;
 use std::vec;
 
-use super::http;
-use super::sandbox::State;
+use super::comms;
+use super::runtime::State;
 
 mod permissions;
 use permissions::Permissions;
@@ -74,18 +74,18 @@ pub fn op_tls_peer_certificate(#[smi] _: u32, _: bool) -> Option<deno_core::serd
 
 #[op2]
 #[serde]
-fn op_get_request(state: &mut OpState) -> Option<http::Request> {
+fn op_get_request(state: &mut OpState) -> Option<comms::Request> {
     state.borrow::<Rc<RefCell<State>>>().borrow().req.clone()
 }
 
 #[op2(async)]
 async fn op_set_response(
     state: Rc<RefCell<OpState>>,
-    #[serde] mut res: http::Response,
+    #[serde] mut res: comms::Response,
 ) -> Result<(), JsError> {
     let (sender, request_id) = {
         let state_borrow = state.borrow();
-        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::sandbox::State>>>();
+        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::runtime::State>>>();
         let mut borrowed = sandbox_state.borrow_mut();
         (borrowed.response_tx.take(), borrowed.request_id)
     };
@@ -121,7 +121,7 @@ fn op_get_request_id(state: &mut OpState) -> String {
 async fn op_read_request_chunk(state: Rc<RefCell<OpState>>) -> Result<Vec<u8>, JsError> {
     let receiver = {
         let state_borrow = state.borrow();
-        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::sandbox::State>>>();
+        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::runtime::State>>>();
         sandbox_state.borrow().incoming_body_rx.clone()
     };
 
@@ -141,7 +141,7 @@ async fn op_write_response_chunk(
 ) -> Result<(), JsError> {
     let sender = {
         let state_borrow = state.borrow();
-        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::sandbox::State>>>();
+        let sandbox_state = state_borrow.borrow::<Rc<RefCell<super::runtime::State>>>();
         sandbox_state.borrow().outgoing_body_tx.clone()
     };
 

crates/funcgg-runtime/src/lib.rs

@@ -1,7 +1,7 @@
 mod ext;
 mod loader;
-mod sandbox;
+mod runtime;
 
-pub mod http;
+pub mod comms;
 pub mod snapshot;
-pub use sandbox::Sandbox;
+pub use runtime::Sandbox;

crates/funcgg-runtime/src/mod.rs

@@ -1,4 +1,3 @@
-mod ext;
 mod loader;
 mod sandbox;
 mod snapshot;

crates/funcgg-runtime/src/sandbox.rs renamed to crates/funcgg-runtime/src/runtime.rs

@@ -9,8 +9,8 @@ use std::time::Duration;
 use tokio::sync::{Mutex, mpsc, oneshot};
 use uuid::Uuid;
 
+use super::comms;
 use super::ext;
-use super::http;
 use super::loader;
 
 static HEAP_LIMIT: usize = 64 * 1024 * 1024; // 64MB
@@ -23,12 +23,12 @@ static USER_MOD_SPECIFIER: LazyLock<Url> =
     LazyLock::new(|| "func:user-code".parse().expect("bad module specifier"));
 
 pub struct State {
-    pub req: Option<http::Request>,
-    pub res: Option<http::Response>,
+    pub req: Option<comms::Request>,
+    pub res: Option<comms::Response>,
     pub request_id: Uuid,
     pub incoming_body_rx: Rc<Mutex<mpsc::Receiver<Result<bytes::Bytes, String>>>>,
     pub outgoing_body_tx: mpsc::Sender<bytes::Bytes>,
-    pub response_tx: Option<oneshot::Sender<http::Response>>,
+    pub response_tx: Option<oneshot::Sender<comms::Response>>,
 }
 
 pub struct Sandbox {
@@ -40,19 +40,17 @@ impl Sandbox {
     pub fn new(
         request_id: Uuid,
         startup_snapshot: Option<&'static [u8]>,
-        incoming_body_rx: mpsc::Receiver<Result<bytes::Bytes, String>>,
-        outgoing_body_tx: mpsc::Sender<bytes::Bytes>,
-        response_tx: oneshot::Sender<http::Response>,
+        channels: comms::Channels,
     ) -> Result<Self> {
         _ = CryptoProvider::install_default(aws_lc_rs::default_provider());
 
         let state = Rc::new(RefCell::new(State {
             request_id,
             req: None,
             res: None,
-            incoming_body_rx: Rc::new(Mutex::new(incoming_body_rx)),
-            outgoing_body_tx,
-            response_tx: Some(response_tx),
+            incoming_body_rx: Rc::new(Mutex::new(channels.incoming_body_rx)),
+            outgoing_body_tx: channels.outgoing_body_tx,
+            response_tx: Some(channels.response_tx),
         }));
 
         let extension_transpiler = Rc::new(loader::transpile);
@@ -86,7 +84,7 @@ impl Sandbox {
     pub async fn execute(
         &mut self,
         user_code: String,
-        request: http::Request,
+        request: comms::Request,
         timeout_duration: Duration,
     ) -> Result<()> {
         let execution_result = tokio::time::timeout(timeout_duration, async {

crates/funcgg-runtime/src/snapshot.rs

@@ -2,8 +2,7 @@ use deno_core::snapshot::CreateSnapshotOutput;
 use std::env;
 use std::rc::Rc;
 
-use super::ext;
-use crate::loader;
+use super::{ext, loader};
 
 pub const FILE_NAME: &str = "FUNCGG_RUNTIME_SNAPSHOT.bin";
 

crates/funcgg-worker/src/pool.rs

@@ -5,12 +5,12 @@ use rand::seq::SliceRandom;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::sync::{Mutex, mpsc, oneshot};
+use tokio::sync::{Mutex, mpsc};
 use tokio::time::{Instant, sleep};
 use uuid::Uuid;
 
 use crate::worker::{Worker, WorkerRequest};
-use funcgg_runtime::http;
+use funcgg_runtime::comms;
 
 #[derive(Debug)]
 pub enum StateChange {
@@ -79,10 +79,8 @@ impl Pool {
     pub async fn send_work(
         &self,
         js_code: String,
-        http_request: http::Request,
-        incoming_body_rx: mpsc::Receiver<Result<bytes::Bytes, String>>,
-        outgoing_body_tx: mpsc::Sender<bytes::Bytes>,
-        response_tx: oneshot::Sender<http::Response>,
+        http_request: comms::Request,
+        channels: comms::Channels,
     ) -> Result<(), String> {
         let request_id = Uuid::now_v7();
 
@@ -91,9 +89,7 @@ impl Pool {
             id: request_id,
             js_code,
             http_request,
-            incoming_body_rx,
-            outgoing_body_tx,
-            response_tx,
+            channels,
         };
 
         tracing::debug!(

crates/funcgg-worker/src/routes.rs

@@ -6,7 +6,7 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use deno_core::futures::StreamExt;
-use funcgg_runtime::http;
+use funcgg_runtime::comms;
 use std::sync::Arc;
 use std::{collections::HashMap, convert::Infallible};
 use tokio::sync::{mpsc, oneshot};
@@ -52,25 +52,22 @@ pub async fn invoke(State(pool): State<Arc<Pool>>, request: Request) -> Response
         }
     });
 
-    let req = funcgg_runtime::http::Request {
+    let req = funcgg_runtime::comms::Request {
         method,
         uri,
         headers,
     };
 
     let (response_body_tx, response_body_rx) = mpsc::channel::<bytes::Bytes>(1);
-    let (response_tx, response_rx) = oneshot::channel::<http::Response>();
+    let (response_tx, response_rx) = oneshot::channel::<comms::Response>();
 
-    if let Err(err) = pool
-        .send_work(
-            js_code.to_string(),
-            req,
-            body_rx,
-            response_body_tx,
-            response_tx,
-        )
-        .await
-    {
+    let channels = comms::Channels {
+        incoming_body_rx: body_rx,
+        outgoing_body_tx: response_body_tx,
+        response_tx,
+    };
+
+    if let Err(err) = pool.send_work(js_code.to_string(), req, channels).await {
         tracing::error!("Handler invocation failed: {}", err);
         return (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error").into_response();
     };

crates/funcgg-worker/src/worker.rs

@@ -1,20 +1,18 @@
 use std::time::Duration;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::mpsc;
 use uuid::Uuid;
 
 use super::pool::StateChange;
-use funcgg_runtime::{Sandbox, http};
+use funcgg_runtime::{Sandbox, comms};
 
 static STARTUP_SNAPSHOT: &[u8] = include_bytes!(env!("SNAPSHOT_PATH"));
 
 #[derive(Debug)]
 pub struct WorkerRequest {
     pub id: Uuid,
     pub js_code: String,
-    pub http_request: http::Request,
-    pub incoming_body_rx: mpsc::Receiver<Result<bytes::Bytes, String>>,
-    pub outgoing_body_tx: mpsc::Sender<bytes::Bytes>,
-    pub response_tx: oneshot::Sender<http::Response>,
+    pub http_request: comms::Request,
+    pub channels: comms::Channels,
 }
 
 pub struct Worker {
@@ -58,13 +56,7 @@ impl Worker {
 
     // TODO: make this state change to failure on failure
     async fn process_request(&self, request: WorkerRequest) -> Result<(), String> {
-        let mut sandbox = match Sandbox::new(
-            request.id,
-            Some(STARTUP_SNAPSHOT),
-            request.incoming_body_rx,
-            request.outgoing_body_tx,
-            request.response_tx,
-        ) {
+        let mut sandbox = match Sandbox::new(request.id, Some(STARTUP_SNAPSHOT), request.channels) {
             Ok(rt) => rt,
             Err(e) => {
                 tracing::error!(