Update core/dependencies and call worker validate (#267)

Fixes #253

Author: cretz

src/Temporalio/Bridge/Cargo.lock

@@ -580,10 +580,10 @@ internal partial struct WorkerOptions
     }
 
     [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
-    internal unsafe delegate void WorkerPollCallback(void* user_data, [NativeTypeName("const struct ByteArray *")] ByteArray* success, [NativeTypeName("const struct ByteArray *")] ByteArray* fail);
+    internal unsafe delegate void WorkerCallback(void* user_data, [NativeTypeName("const struct ByteArray *")] ByteArray* fail);
 
     [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
-    internal unsafe delegate void WorkerCallback(void* user_data, [NativeTypeName("const struct ByteArray *")] ByteArray* fail);
+    internal unsafe delegate void WorkerPollCallback(void* user_data, [NativeTypeName("const struct ByteArray *")] ByteArray* success, [NativeTypeName("const struct ByteArray *")] ByteArray* fail);
 
     internal unsafe partial struct WorkerReplayerOrFail
     {
@@ -726,6 +726,9 @@ internal static unsafe partial class Methods
         [DllImport("temporal_sdk_bridge", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)]
         public static extern void worker_free([NativeTypeName("struct Worker *")] Worker* worker);
 
+        [DllImport("temporal_sdk_bridge", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)]
+        public static extern void worker_validate([NativeTypeName("struct Worker *")] Worker* worker, void* user_data, [NativeTypeName("WorkerCallback")] IntPtr callback);
+
         [DllImport("temporal_sdk_bridge", CallingConvention = CallingConvention.Cdecl, ExactSpelling = true)]
         public static extern void worker_replace_client([NativeTypeName("struct Worker *")] Worker* worker, [NativeTypeName("struct Client *")] Client* new_client);
 

src/Temporalio/Bridge/Interop/Interop.cs

@@ -88,6 +88,38 @@ internal Worker(Worker other)
         /// </summary>
         internal unsafe Interop.Worker* Ptr { get; private set; }
 
+        /// <summary>
+        /// Validate the worker.
+        /// </summary>
+        /// <returns>Validation task.</returns>
+        public async Task ValidateAsync()
+        {
+            using (var scope = new Scope())
+            {
+                var completion = new TaskCompletionSource<bool>();
+                unsafe
+                {
+                    Interop.Methods.worker_validate(
+                        Ptr,
+                        null,
+                        scope.FunctionPointer<Interop.WorkerCallback>(
+                            (userData, fail) =>
+                            {
+                                if (fail != null)
+                                {
+                                    completion.TrySetException(new InvalidOperationException(
+                                        new ByteArray(Runtime, fail).ToUTF8()));
+                                }
+                                else
+                                {
+                                    completion.TrySetResult(true);
+                                }
+                            }));
+                }
+                await completion.Task.ConfigureAwait(false);
+            }
+        }
+
         /// <summary>
         /// Replace the client.
         /// </summary>

src/Temporalio/Bridge/Worker.cs

@@ -382,6 +382,11 @@ typedef struct WorkerOptions {
   struct ByteArrayRefArray nondeterminism_as_workflow_fail_for_types;
 } WorkerOptions;
 
+/**
+ * If fail is present, it must be freed.
+ */
+typedef void (*WorkerCallback)(void *user_data, const struct ByteArray *fail);
+
 /**
  * If success or fail are present, they must be freed. They will both be null
  * if this is a result of a poll shutdown.
@@ -390,11 +395,6 @@ typedef void (*WorkerPollCallback)(void *user_data,
                                    const struct ByteArray *success,
                                    const struct ByteArray *fail);
 
-/**
- * If fail is present, it must be freed.
- */
-typedef void (*WorkerCallback)(void *user_data, const struct ByteArray *fail);
-
 typedef struct WorkerReplayerOrFail {
   struct Worker *worker;
   struct WorkerReplayPusher *worker_replay_pusher;
@@ -521,6 +521,8 @@ struct WorkerOrFail worker_new(struct Client *client, const struct WorkerOptions
 
 void worker_free(struct Worker *worker);
 
+void worker_validate(struct Worker *worker, void *user_data, WorkerCallback callback);
+
 void worker_replace_client(struct Worker *worker, struct Client *new_client);
 
 void worker_poll_workflow_activation(struct Worker *worker,

src/Temporalio/Bridge/include/temporal-sdk-bridge.h

@@ -134,6 +134,34 @@ pub extern "C" fn worker_free(worker: *mut Worker) {
     }
 }
 
+/// If fail is present, it must be freed.
+type WorkerCallback = unsafe extern "C" fn(user_data: *mut libc::c_void, fail: *const ByteArray);
+
+#[no_mangle]
+pub extern "C" fn worker_validate(
+    worker: *mut Worker,
+    user_data: *mut libc::c_void,
+    callback: WorkerCallback,
+) {
+    let worker = unsafe { &*worker };
+    let user_data = UserDataHandle(user_data);
+    let core_worker = worker.worker.as_ref().unwrap().clone();
+    worker.runtime.core.tokio_handle().spawn(async move {
+        let fail = match core_worker.validate().await {
+            Ok(_) => std::ptr::null(),
+            Err(err) => worker
+                .runtime
+                .clone()
+                .alloc_utf8(&format!("Worker validation failed: {}", err))
+                .into_raw()
+                .cast_const(),
+        };
+        unsafe {
+            callback(user_data.into(), fail);
+        }
+    });
+}
+
 #[no_mangle]
 pub extern "C" fn worker_replace_client(worker: *mut Worker, new_client: *mut Client) {
     let worker = unsafe { &*worker };
@@ -218,9 +246,6 @@ pub extern "C" fn worker_poll_activity_task(
     });
 }
 
-/// If fail is present, it must be freed.
-type WorkerCallback = unsafe extern "C" fn(user_data: *mut libc::c_void, fail: *const ByteArray);
-
 #[no_mangle]
 pub extern "C" fn worker_complete_workflow_activation(
     worker: *mut Worker,

src/Temporalio/Bridge/sdk-core

@@ -272,6 +272,9 @@ private async Task ExecuteInternalAsync(
                 throw new InvalidOperationException("Already started");
             }
 
+            // Check that the worker is valid
+            await BridgeWorker.ValidateAsync().ConfigureAwait(false);
+
             var tasks = new List<Task>()
             {
                 // Create a task that will complete on cancellation

src/Temporalio/Bridge/src/worker.rs

@@ -4587,6 +4587,21 @@ await AssertMore.HasEventEventuallyAsync(
             });
     }
 
+    [Fact]
+    public async Task ExecuteAsync_InvalidWorker_Fails()
+    {
+        // Try to run a worker on an invalid namespace
+        var options = (TemporalClientOptions)Client.Options.Clone();
+        options.Namespace = "does-not-exist";
+        var client = new TemporalClient(Client.Connection, options);
+        using var worker = new TemporalWorker(
+            client,
+            new TemporalWorkerOptions("some-task-queue").AddWorkflow<SimpleWorkflow>());
+        var err = await Assert.ThrowsAsync<InvalidOperationException>(
+            () => worker.ExecuteAsync(() => Task.Delay(2000)));
+        Assert.Contains("Worker validation failed", err.Message);
+    }
+
     internal static Task AssertTaskFailureContainsEventuallyAsync(
         WorkflowHandle handle, string messageContains)
     {