From 2014c6e3e0ab549aabad481d2771a20da19f9ac0 Mon Sep 17 00:00:00 2001
From: iximeow <me@iximeow.net>
Date: Sun, 2 Jul 2023 14:17:04 -0700
Subject: client ack to new task is now kind:started, status:started

---
 src/ci_driver.rs | 50 ++++++++++++++++++++++++++++----------------------
 1 file changed, 28 insertions(+), 22 deletions(-)

diff --git a/src/ci_driver.rs b/src/ci_driver.rs
index e3cc4c9..c6bef14 100644
--- a/src/ci_driver.rs
+++ b/src/ci_driver.rs
@@ -93,10 +93,19 @@ async fn activate_run(dbctx: Arc<DbCtx>, run: &PendingRun, clients: &mut mpsc::R
             continue;
         }
 
-        if let Ok(Some(mut client_job)) = candidate.submit(&dbctx, &run, &remote_git_url, &commit_sha).await {
-            break client_job;
-        } else {
-            // failed to submit job, move on for now
+        let res = candidate.submit(&dbctx, &run, &remote_git_url, &commit_sha).await;
+
+        match res {
+            Ok(Some(mut client_job)) => {
+                break client_job;
+            }
+            Ok(None) => {
+                eprintln!("client hung up instead of acking task");
+            }
+            Err(e) => {
+                // failed to submit job, move on for now
+                eprintln!("failed to submit task: {:?}", e);
+            }
         }
     };
 
@@ -303,25 +312,22 @@ impl RunnerClient {
             remote_url: remote_git_url.to_string(),
             build_token: self.build_token.to_string(),
         })).await?;
-        match self.recv().await {
+        match self.recv_typed::<ClientProto>().await {
+            Ok(Some(ClientProto::Started)) => {
+                let task_witness = Arc::new(());
+                ACTIVE_TASKS.lock().unwrap().insert(job.id, Arc::downgrade(&task_witness));
+                Ok(Some(ClientJob {
+                    task: job.clone(),
+                    dbctx: Arc::clone(dbctx),
+                    sha: sha.to_string(),
+                    remote_git_url: remote_git_url.to_string(),
+                    client: self,
+                    task_witness,
+                }))
+            }
             Ok(Some(resp)) => {
-                if resp == serde_json::json!({
-                    "status": "started"
-                }) {
-                    let task_witness = Arc::new(());
-                    ACTIVE_TASKS.lock().unwrap().insert(job.id, Arc::downgrade(&task_witness));
-                    eprintln!("resp: {:?}", resp);
-                    Ok(Some(ClientJob {
-                        task: job.clone(),
-                        dbctx: Arc::clone(dbctx),
-                        sha: sha.to_string(),
-                        remote_git_url: remote_git_url.to_string(),
-                        client: self,
-                        task_witness,
-                    }))
-                } else {
-                    Err("client rejected job".to_string())
-                }
+                eprintln!("invalid response: {:?}", resp);
+                Err("client rejected job".to_string())
             }
             Ok(None) => {
                 Ok(None)
-- 
cgit v1.1