refactor: stub coordinator CLI handlers

Author: sjfhsjfh

Cargo.lock

@@ -80,17 +80,20 @@ arrow-data = { version = "54.2.1" }
 arrow-array = { version = "54.2.1" }
 arrow-json = { version = "54.2.1" }
 async-trait = "0.1.89"
+eyre = "0.6.12"
+git2 = { version = "0.18.0", features = ["vendored-openssl"] }
 parquet = { version = "54.2.1" }
 pyo3 = { version = "0.23", features = [
     "eyre",
     "abi3-py37",
     "multiple-pymethods",
 ] }
 pythonize = "0.23"
-git2 = { version = "0.18.0", features = ["vendored-openssl"] }
+serde = "1.0.228"
+serde_json = "1.0.145"
 serde_yaml = "0.9.33"
+tokio = { version = "1.49.0", features = ["full"] }
 zenoh = "1.1.1"
-serde_json = "1.0.145"
 
 [package]
 name = "dora-examples"

Cargo.toml

@@ -1,49 +1,128 @@
-use std::sync::Arc;
+use std::{collections::BTreeSet, time::Duration};
 
 use async_trait::async_trait;
-use dora_core::uhlc::HLC;
 use dora_message::{
     BuildId,
-    cli_to_coordinator::{BuildReq, CliToCoordinator},
+    cli_to_coordinator::{
+        BuildReq, CliAndDefaultDaemonIps, CliToCoordinator, StartReq, WaitForBuildResp,
+    },
+    common::DaemonId,
+    coordinator_to_cli::{DataflowListEntry, NodeInfo},
+    id::{NodeId, OperatorId},
 };
+use eyre::Result;
+use uuid::Uuid;
 
-use crate::build_dataflow;
+pub struct CliRequestHandler;
 
 #[async_trait]
 impl CliToCoordinator for CliRequestHandler {
-    async fn build(self, req: BuildReq) -> eyre::Result<BuildId> {
-        let BuildReq {
-            session_id,
-            dataflow,
-            git_sources,
-            prev_git_sources,
-            local_working_dir,
-            uv,
-        } = req;
+    async fn build(self, req: BuildReq) -> Result<BuildId> {
+        // let BuildReq {
+        //     session_id,
+        //     dataflow,
+        //     git_sources,
+        //     prev_git_sources,
+        //     local_working_dir,
+        //     uv,
+        // } = req;
 
-        // assign a random build id
-        let build_id = BuildId::generate();
+        // // assign a random build id
+        // let build_id = BuildId::generate();
 
-        let result = build_dataflow(
-            build_id,
-            session_id,
-            dataflow,
-            git_sources,
-            prev_git_sources,
-            local_working_dir,
-            &self.clock,
-            uv,
-            &mut daemon_connections,
-        )
-        .await;
-        match result {
-            Ok(build) => {
-                running_builds.insert(build_id, build);
-                Ok(build_id)
-            }
-            Err(err) => {
-                let _ = reply_sender.send(Err(err));
-            }
-        }
+        // let result = build_dataflow(
+        //     build_id,
+        //     session_id,
+        //     dataflow,
+        //     git_sources,
+        //     prev_git_sources,
+        //     local_working_dir,
+        //     &self.clock,
+        //     uv,
+        //     &mut daemon_connections,
+        // )
+        // .await;
+        // match result {
+        //     Ok(build) => {
+        //         running_builds.insert(build_id, build);
+        //         Ok(build_id)
+        //     }
+        //     Err(err) => {
+        //         let _ = reply_sender.send(Err(err));
+        //     }
+        // }
+        todo!()
+    }
+    async fn wait_for_build(self, build_id: BuildId) -> Result<WaitForBuildResp> {
+        todo!()
+    }
+    async fn start(self, req: StartReq) -> Result<Uuid> {
+        todo!()
+    }
+    async fn wait_for_spawn(self, dataflow_id: Uuid) -> Result<Uuid> {
+        todo!()
+    }
+    async fn reload(
+        self,
+        dataflow_id: Uuid,
+        node_id: NodeId,
+        operator_id: Option<OperatorId>,
+    ) -> Result<Uuid> {
+        todo!()
+    }
+    async fn check(self, dataflow_uuid: Uuid) -> Result<Uuid> {
+        todo!()
+    }
+    async fn stop(
+        self,
+        dataflow_uuid: Uuid,
+        grace_duration: Option<Duration>,
+        force: bool,
+    ) -> Result<()> {
+        todo!()
+    }
+    async fn stop_by_name(
+        self,
+        name: String,
+        grace_duration: Option<Duration>,
+        force: bool,
+    ) -> Result<()> {
+        todo!()
+    }
+    async fn logs(
+        self,
+        uuid: Option<Uuid>,
+        name: Option<String>,
+        node: String,
+        tail: Option<usize>,
+    ) -> Result<Vec<u8>> {
+        todo!()
+    }
+    async fn destroy(self) -> Result<()> {
+        todo!()
+    }
+    async fn list(self) -> Result<Vec<DataflowListEntry>> {
+        todo!()
+    }
+    async fn info(self, dataflow_uuid: Uuid) -> Result<()> {
+        todo!()
+    }
+    async fn daemon_connected(self) -> Result<bool> {
+        todo!()
+    }
+    async fn connected_machines(self) -> Result<BTreeSet<DaemonId>> {
+        todo!()
+    }
+    async fn log_subscribe(self, dataflow_id: Uuid, level: log::LevelFilter) -> Result<()> {
+        todo!()
+    }
+    async fn build_log_subscribe(self, build_id: BuildId, level: log::LevelFilter) -> Result<()> {
+        todo!()
+    }
+    async fn cli_and_default_daemon_on_same_machine(self) -> Result<CliAndDefaultDaemonIps> {
+        todo!()
+    }
+    async fn get_node_info(self) -> Result<Vec<NodeInfo>> {
+        todo!()
     }
 }

binaries/coordinator/src/handler.rs

@@ -15,3 +15,8 @@ quote = "1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json.workspace = true
 syn = { version = "2", features = ["full", "parsing"] }
+
+[dev-dependencies]
+async-trait = { workspace = true }
+eyre = { workspace = true }
+tokio = { workspace = true }

libraries/dora-schema-macro/Cargo.toml

@@ -24,7 +24,22 @@ use crate::syntax::SchemaInput;
 /// This will generate:
 /// - Protocol enums `MyProtocolRequest` and `MyProtocolResponse`
 /// - A client struct `MyProtocolClient` with methods `foo` and `bar`
-/// - A server trait `MyProtocol` with handler methods for `foo` and `bar
+/// - A server trait `MyProtocol` with handler methods for `foo` and `bar`
+///
+/// For example, the above trait will generate the following protocol enums:
+///
+/// ```ignore
+/// pub enum MyProtocolRequest {
+///     Foo{ foo: FooRequest },
+///     Bar{ bar: BarRequest },
+/// }
+///
+/// pub enum MyProtocolResponse {
+///     Foo(FooResponse),
+///     Bar(BarResponse),
+/// }
+/// ```
+///
 #[proc_macro_attribute]
 pub fn dora_schema(_attr: TokenStream, input: TokenStream) -> TokenStream {
     let schema = syn::parse_macro_input!(input as SchemaInput);

libraries/dora-schema-macro/src/lib.rs

@@ -35,12 +35,12 @@ pub fn generate_protocol(schema: &SchemaInput) -> proc_macro2::TokenStream {
 
     // TODO: better eyre conversion
     quote! {
-        #[derive(Debug, Clone, ::serde::Serialize, ::serde::Deserialize)]
+        #[derive(::std::fmt::Debug, ::std::clone::Clone, ::serde::Serialize, ::serde::Deserialize)]
         pub enum #request_enum {
             #(#request_variants,)*
         }
 
-        #[derive(Debug, Clone, ::serde::Serialize, ::serde::Deserialize)]
+        #[derive(::std::fmt::Debug, ::std::clone::Clone, ::serde::Serialize, ::serde::Deserialize)]
         pub enum #response_enum {
             #(#response_variants,)*
             Error(::std::string::String),

libraries/dora-schema-macro/src/protocol.rs

@@ -1,5 +1,4 @@
-use proc_macro2::Ident;
-use quote::{format_ident, quote};
+use quote::quote;
 
 use crate::{
     SchemaInput,
@@ -46,26 +45,22 @@ pub fn generate_server_trait(schema: &SchemaInput) -> proc_macro2::TokenStream {
         .collect::<Vec<_>>();
 
     let handle_function = quote! {
-        async fn handle<T>(self, request: #request_enum) -> ::eyre::Result<()>
+        /// Handle a single request, returning a response.
+        ///
+        /// Do the dispatching by default, override this method to implement custom dispatching logic.
+        ///
+        async fn handle(self, request: #request_enum) -> ::eyre::Result<#response_enum>
         {
-            use ::communication_layer_request_reply::AsyncTransport;
-            // TODO: handle transport errors (e.g. serde errors) and return error responses instead of exiting
-            if let Some(req) = transport.receive().await? {
-                let resp = match req {
-                    #(#dispatch_arms)*
-                };
-                transport.send(&resp).await?;
-            } else {
-                ::eyre::bail!("Transport closed while waiting for request");
-            }
-            Ok(())
+            Ok(match request {
+                #(#dispatch_arms)*
+            })
         }
     };
 
     quote! {
         #[::async_trait::async_trait]
         /// Note: the `serve` function will clone the handler for each request, use `Arc<Mutex<State>>` to share state between requests.
-        pub trait #trait_name: ::std::marker::Sized {
+        pub trait #trait_name: ::std::marker::Sized + ::std::marker::Send {
             #(#trait_methods)*
 
             #handle_function

libraries/dora-schema-macro/src/server.rs

@@ -0,0 +1,22 @@
+#[derive(std::fmt::Debug, std::clone::Clone, serde::Serialize, serde::Deserialize)]
+pub struct FooRequest;
+#[derive(std::fmt::Debug, std::clone::Clone, serde::Serialize, serde::Deserialize)]
+pub struct FooResponse;
+
+#[derive(std::fmt::Debug, std::clone::Clone, serde::Serialize, serde::Deserialize)]
+pub struct BarRequest;
+#[derive(std::fmt::Debug, std::clone::Clone, serde::Serialize, serde::Deserialize)]
+pub struct BarResponse;
+
+#[dora_schema_macro::dora_schema]
+pub trait MyProtocol {
+    #[allow(clippy::disallowed_names)]
+    fn foo(foo: FooRequest) -> FooResponse;
+    fn bar(bar: BarRequest) -> BarResponse;
+}
+
+#[allow(unused)]
+async fn compile_check<H: MyProtocol>(handler: H) {
+    let request: MyProtocolRequest = MyProtocolRequest::Foo { foo: FooRequest };
+    let _: MyProtocolResponse = handler.handle(request).await.unwrap();
+}