better logging

Author: chris-olszewski

packages/core-bridge/src/client.rs

@@ -55,33 +55,36 @@ pub fn client_new(
     let runtime = runtime.borrow()?.core_runtime.clone();
     let config: CoreClientOptions = config.try_into()?;
 
-    runtime.clone().future_to_promise(async move {
-        let metric_meter = runtime.clone().telemetry().get_temporal_metric_meter();
-        let res = config.connect_no_namespace(metric_meter).await;
-
-        let core_client = match res {
-            Ok(core_client) => core_client,
-            Err(ClientInitError::InvalidHeaders(e)) => Err(BridgeError::TypeError {
-                message: format!("Invalid metadata key: {e}"),
-                field: None,
-            })?,
-            Err(ClientInitError::SystemInfoCallError(e)) => Err(BridgeError::TransportError(
-                format!("Failed to call GetSystemInfo: {e}"),
-            ))?,
-            Err(ClientInitError::TonicTransportError(e)) => {
-                Err(BridgeError::TransportError(format!("{e:?}")))?
-            }
-            Err(ClientInitError::InvalidUri(e)) => Err(BridgeError::TypeError {
-                message: e.to_string(),
-                field: None,
-            })?,
-        };
-
-        Ok(OpaqueOutboundHandle::new(Client {
-            core_runtime: runtime,
-            core_client,
-        }))
-    })
+    runtime.clone().future_to_promise_named(
+        async move {
+            let metric_meter = runtime.clone().telemetry().get_temporal_metric_meter();
+            let res = config.connect_no_namespace(metric_meter).await;
+
+            let core_client = match res {
+                Ok(core_client) => core_client,
+                Err(ClientInitError::InvalidHeaders(e)) => Err(BridgeError::TypeError {
+                    message: format!("Invalid metadata key: {e}"),
+                    field: None,
+                })?,
+                Err(ClientInitError::SystemInfoCallError(e)) => Err(BridgeError::TransportError(
+                    format!("Failed to call GetSystemInfo: {e}"),
+                ))?,
+                Err(ClientInitError::TonicTransportError(e)) => {
+                    Err(BridgeError::TransportError(format!("{e:?}")))?
+                }
+                Err(ClientInitError::InvalidUri(e)) => Err(BridgeError::TypeError {
+                    message: e.to_string(),
+                    field: None,
+                })?,
+            };
+
+            Ok(OpaqueOutboundHandle::new(Client {
+                core_runtime: runtime,
+                core_client,
+            }))
+        },
+        "client_new",
+    )
 }
 
 /// Update a Client's HTTP request headers

packages/core-bridge/src/runtime.rs

@@ -167,6 +167,15 @@ pub trait RuntimeExt {
         F: Future<Output = Result<R, BridgeError>> + Send + 'static,
         R: TryIntoJs + Send + 'static;
 
+    fn future_to_promise_named<F, R>(
+        &self,
+        future: F,
+        caller: &'static str,
+    ) -> BridgeResult<BridgeFuture<R>>
+    where
+        F: Future<Output = Result<R, BridgeError>> + Send + 'static,
+        R: TryIntoJs + Send + 'static;
+
     /// Spawn a future on the Tokio runtime, and let it run to completion without waiting for it to
     /// complete. Should any error occur, we'll try to send them to this Runtime's logger, but may
     /// end up or silently dropping entries in some extreme cases.
@@ -187,6 +196,21 @@ impl RuntimeExt for CoreRuntime {
         )))
     }
 
+    fn future_to_promise_named<F, R>(
+        &self,
+        future: F,
+        caller: &'static str,
+    ) -> BridgeResult<BridgeFuture<R>>
+    where
+        F: Future<Output = Result<R, BridgeError>> + Send + 'static,
+        R: TryIntoJs + Send + 'static,
+    {
+        enter_sync!(self);
+        Ok(BridgeFuture::new(Box::pin(future.instrument(
+            tracing::info_span!("future_to_promise_named", caller),
+        ))))
+    }
+
     fn spawn_and_forget<F>(&self, future: F)
     where
         F: Future<Output = Result<(), BridgeError>> + Send + 'static,
@@ -216,6 +240,18 @@ impl RuntimeExt for Arc<CoreRuntime> {
     {
         self.as_ref().spawn_and_forget(future);
     }
+
+    fn future_to_promise_named<F, R>(
+        &self,
+        future: F,
+        caller: &'static str,
+    ) -> BridgeResult<BridgeFuture<R>>
+    where
+        F: Future<Output = Result<R, BridgeError>> + Send + 'static,
+        R: TryIntoJs + Send + 'static,
+    {
+        self.as_ref().future_to_promise_named(future, caller)
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////

packages/core-bridge/src/worker.rs

@@ -103,12 +103,15 @@ pub fn worker_validate(worker: OpaqueInboundHandle<Worker>) -> BridgeResult<Brid
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        worker
-            .validate()
-            .await
-            .map_err(|err| BridgeError::TransportError(err.to_string()))
-    })
+    runtime.future_to_promise_named(
+        async move {
+            worker
+                .validate()
+                .await
+                .map_err(|err| BridgeError::TransportError(err.to_string()))
+        },
+        "worker_validate",
+    )
 }
 
 /// Initiate a single workflow activation poll request.
@@ -121,19 +124,22 @@ pub fn worker_poll_workflow_activation(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        let result = worker.poll_workflow_activation().await;
+    runtime.future_to_promise_named(
+        async move {
+            let result = worker.poll_workflow_activation().await;
 
-        match result {
-            Ok(task) => Ok(task.encode_to_vec()),
-            Err(err) => match err {
-                PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
-                PollError::TonicError(status) => {
-                    Err(BridgeError::TransportError(status.message().to_string()))?
-                }
-            },
-        }
-    })
+            match result {
+                Ok(task) => Ok(task.encode_to_vec()),
+                Err(err) => match err {
+                    PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
+                    PollError::TonicError(status) => {
+                        Err(BridgeError::TransportError(status.message().to_string()))?
+                    }
+                },
+            }
+        },
+        "worker_poll_workflow_activation",
+    )
 }
 
 /// Submit a workflow activation completion to core.
@@ -154,21 +160,24 @@ pub fn worker_complete_workflow_activation(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        worker
-            .complete_workflow_activation(workflow_completion)
-            .await
-            .map_err(|err| match err {
-                CompleteWfError::MalformedWorkflowCompletion { reason, run_id } => {
-                    BridgeError::TypeError {
-                        field: None,
-                        message: format!(
-                            "Malformed Workflow Completion: {reason:?} for RunID={run_id}"
-                        ),
+    runtime.future_to_promise_named(
+        async move {
+            worker
+                .complete_workflow_activation(workflow_completion)
+                .await
+                .map_err(|err| match err {
+                    CompleteWfError::MalformedWorkflowCompletion { reason, run_id } => {
+                        BridgeError::TypeError {
+                            field: None,
+                            message: format!(
+                                "Malformed Workflow Completion: {reason:?} for RunID={run_id}"
+                            ),
+                        }
                     }
-                }
-            })
-    })
+                })
+        },
+        "worker_complete_workflow_activation",
+    )
 }
 
 /// Initiate a single activity task poll request.
@@ -181,19 +190,22 @@ pub fn worker_poll_activity_task(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        let result = worker.poll_activity_task().await;
+    runtime.future_to_promise_named(
+        async move {
+            let result = worker.poll_activity_task().await;
 
-        match result {
-            Ok(task) => Ok(task.encode_to_vec()),
-            Err(err) => match err {
-                PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
-                PollError::TonicError(status) => {
-                    Err(BridgeError::TransportError(status.message().to_string()))?
-                }
-            },
-        }
-    })
+            match result {
+                Ok(task) => Ok(task.encode_to_vec()),
+                Err(err) => match err {
+                    PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
+                    PollError::TonicError(status) => {
+                        Err(BridgeError::TransportError(status.message().to_string()))?
+                    }
+                },
+            }
+        },
+        "worker_poll_activity_task",
+    )
 }
 
 /// Submit an activity task completion to core.
@@ -214,20 +226,23 @@ pub fn worker_complete_activity_task(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        worker
-            .complete_activity_task(activity_completion)
-            .await
-            .map_err(|err| match err {
-                CompleteActivityError::MalformedActivityCompletion {
-                    reason,
-                    completion: _,
-                } => BridgeError::TypeError {
-                    field: None,
-                    message: format!("Malformed Activity Completion: {reason:?}"),
-                },
-            })
-    })
+    runtime.future_to_promise_named(
+        async move {
+            worker
+                .complete_activity_task(activity_completion)
+                .await
+                .map_err(|err| match err {
+                    CompleteActivityError::MalformedActivityCompletion {
+                        reason,
+                        completion: _,
+                    } => BridgeError::TypeError {
+                        field: None,
+                        message: format!("Malformed Activity Completion: {reason:?}"),
+                    },
+                })
+        },
+        "worker_complete_activity_task",
+    )
 }
 
 /// Submit an activity heartbeat to core.
@@ -260,19 +275,22 @@ pub fn worker_poll_nexus_task(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        let result = worker.poll_nexus_task().await;
+    runtime.future_to_promise_named(
+        async move {
+            let result = worker.poll_nexus_task().await;
 
-        match result {
-            Ok(task) => Ok(task.encode_to_vec()),
-            Err(err) => match err {
-                PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
-                PollError::TonicError(status) => {
-                    Err(BridgeError::TransportError(status.message().to_string()))?
-                }
-            },
-        }
-    })
+            match result {
+                Ok(task) => Ok(task.encode_to_vec()),
+                Err(err) => match err {
+                    PollError::ShutDown => Err(BridgeError::WorkerShutdown)?,
+                    PollError::TonicError(status) => {
+                        Err(BridgeError::TransportError(status.message().to_string()))?
+                    }
+                },
+            }
+        },
+        "worker_poll_nexus_task",
+    )
 }
 
 /// Submit an nexus task completion to core.
@@ -291,20 +309,25 @@ pub fn worker_complete_nexus_task(
     let worker = worker_ref.core_worker.clone();
     let runtime = worker_ref.core_runtime.clone();
 
-    runtime.future_to_promise(async move {
-        worker
-            .complete_nexus_task(nexus_completion)
-            .await
-            .map_err(|err| match err {
-                CompleteNexusError::NexusNotEnabled {} => {
-                    BridgeError::UnexpectedError(format!("{err}"))
-                }
-                CompleteNexusError::MalformedNexusCompletion { reason } => BridgeError::TypeError {
-                    field: None,
-                    message: format!("Malformed nexus Completion: {reason:?}"),
-                },
-            })
-    })
+    runtime.future_to_promise_named(
+        async move {
+            worker
+                .complete_nexus_task(nexus_completion)
+                .await
+                .map_err(|err| match err {
+                    CompleteNexusError::NexusNotEnabled {} => {
+                        BridgeError::UnexpectedError(format!("{err}"))
+                    }
+                    CompleteNexusError::MalformedNexusCompletion { reason } => {
+                        BridgeError::TypeError {
+                            field: None,
+                            message: format!("Malformed nexus Completion: {reason:?}"),
+                        }
+                    }
+                })
+        },
+        "worker_complete_nexus_task",
+    )
 }
 
 /// Request shutdown of the worker.
@@ -313,6 +336,7 @@ pub fn worker_complete_nexus_task(
 /// the loop to ensure graceful shutdown.
 #[js_function]
 pub fn worker_initiate_shutdown(worker: OpaqueInboundHandle<Worker>) -> BridgeResult<()> {
+    tracing::info!("Typescript initiate worker shutdown");
     let worker_ref = worker.borrow()?;
     worker_ref.core_worker.initiate_shutdown();
     Ok(())
@@ -337,10 +361,13 @@ pub fn worker_finalize_shutdown(
         }
     })?;
 
-    worker_ref.core_runtime.future_to_promise(async move {
-        worker.finalize_shutdown().await;
-        Ok(())
-    })
+    worker_ref.core_runtime.future_to_promise_named(
+        async move {
+            worker.finalize_shutdown().await;
+            Ok(())
+        },
+        "worker_finalize_shutdown",
+    )
 }
 
 impl MutableFinalize for Worker {