Repository: denoland/deno
Pull request: feat: Send reject-ack message through cron socket


Changed files:
filename: ext/cron/socket.rs
patch: @@ -39,13 +39,15 @@ pub(crate) enum SocketTaskCommand {
     name: String,
     success: bool,
   },
+  AckReject,
 }
 
 #[derive(Serialize)]
-#[serde(tag = "kind", rename_all = "lowercase")]
+#[serde(tag = "kind", rename_all = "kebab-case")]
 enum OutboundMessage<'a> {
   Register { crons: &'a [CronRegistration<'a>] },
   Result { name: &'a str, success: bool },
+  RejectAck,
 }
 
 #[derive(Serialize)]
@@ -75,6 +77,7 @@ impl SocketCronHandler {
     let reject_reason = Rc::new(OnceCell::new());
     let socket_task_handle = spawn(Self::socket_task(
       socket_addr.clone(),
+      socket_task_tx.clone(),
       socket_task_rx,
       socket_task_exit_error.clone(),
       reject_reason.clone(),
@@ -95,6 +98,7 @@ impl SocketCronHandler {
 
   async fn socket_task(
     socket_addr: String,
+    socket_task_tx: mpsc::Sender<SocketTaskCommand>,
     mut socket_task_rx: mpsc::Receiver<SocketTaskCommand>,
     exit_error: Rc<OnceCell<CronError>>,
     reject_reason: Rc<OnceCell<String>>,
@@ -139,13 +143,22 @@ impl SocketCronHandler {
                 return;
               }
             }
+            SocketTaskCommand::AckReject => {
+              if let Err(e) = send_reject_ack(&mut socket_writer).await {
+                let _ = exit_error.set(CronError::SocketError(format!(
+                  "Failed to send reject acknowledgment: {}",
+                  e
+                )));
+                return;
+              }
+            }
           }
         }
 
         result = socket_reader.next_line() => {
           match result {
             Ok(Some(line)) => {
-              let _ = handle_inbound_messages(&line, &invocation_txs, &reject_reason);
+              let _ = handle_inbound_messages(&line, &invocation_txs, &reject_reason, &socket_task_tx);
             }
             Ok(None) => {
               let _ = exit_error.set(CronError::SocketError(
@@ -432,10 +445,24 @@ async fn send_execution_result(
   Ok(())
 }
 
+async fn send_reject_ack(
+  socket_writer: &mut BufWriter<impl tokio::io::AsyncWrite + Unpin>,
+) -> Result<(), std::io::Error> {
+  let msg = OutboundMessage::RejectAck;
+
+  let json = serde_json::to_string(&msg).map_err(std::io::Error::other)?;
+  socket_writer.write_all(json.as_bytes()).await?;
+  socket_writer.write_all(b"\n").await?;
+  socket_writer.flush().await?;
+
+  Ok(())
+}
+
 fn handle_inbound_messages(
   line: &str,
   invocation_txs: &HashMap<String, mpsc::Sender<Traceparent>>,
   reject_reason: &Rc<OnceCell<String>>,
+  socket_task_tx: &mpsc::Sender<SocketTaskCommand>,
 ) -> Result<(), Box<dyn std::error::Error>> {
   let msg: InboundMessage = serde_json::from_str(line)?;
 
@@ -446,7 +473,9 @@ fn handle_inbound_messages(
       }
     }
     InboundMessage::RejectNewCrons { reason } => {
-      let _ = reject_reason.set(reason);
+      if reject_reason.set(reason).is_ok() {
+        let _ = socket_task_tx.try_send(SocketTaskCommand::AckReject);
+      }
     }
   }
 

filename: tests/specs/run/unstable_cron_socket_serve_reject/test_runner.js
patch: @@ -58,6 +58,8 @@ const registeredCrons = [];
 
 // 5. Process messages concurrently
 async function readCronMessages() {
+  let receivedRejectAck = false;
+
   while (true) {
     const { value: chunk, done } = await cronReader.read();
     if (done) break;
@@ -84,9 +86,14 @@ async function readCronMessages() {
           );
           Deno.exit(1);
         }
+      } else if (msg.kind === "reject-ack") {
+        console.error("[CRON SERVER] Received reject-ack");
+        receivedRejectAck = true;
       }
     }
   }
+
+  return receivedRejectAck;
 }
 
 async function readControlMessages() {
@@ -123,7 +130,15 @@ async function readControlMessages() {
   }
 }
 
-await Promise.all([readCronMessages(), readControlMessages()]);
+const [receivedRejectAck] = await Promise.all([
+  readCronMessages(),
+  readControlMessages(),
+]);
+
+if (!receivedRejectAck) {
+  console.error("[CRON SERVER] ERROR: Did not receive reject-ack");
+  Deno.exit(1);
+}
 
 console.error("[CRON SERVER] Registered crons:", registeredCrons.join(", "));
 if (registeredCrons.length === 1 && registeredCrons[0] === "early-cron") {

filename: tests/specs/run/unstable_cron_socket_serve_reject/test_runner.out
patch: @@ -6,6 +6,7 @@ Listening on http://0.0.0.0:[WILDCARD]/ (http://localhost:[WILDCARD]/)
 [CRON SERVER] Registered: early-cron
 [CONTROL CLIENT] Received Serving signal from Deno
 [CRON SERVER] Sent rejection
+[CRON SERVER] Received reject-ack
 [ISOLATE] late-cron rejected as expected with error: Error registering cron: No crons after serve
 [CRON SERVER] Registered crons: early-cron
 [CRON SERVER] SUCCESS: Only early cron was registered