fix(ext/node): cancel pending ipc writes on channel close (#26504)

Fixes the issue described in
#23882 (comment).

The parent was starting to send a message right before the process would
exit, and the channel closed in the middle of the write. Unlike with
reads, we weren't cancelling the pending writes, which resulted in a
`Broken pipe` error surfacing to the user.

ext/node/ops/ipc.rs

@@ -216,10 +216,17 @@ mod impl_ {
       queue_ok.set_index(scope, 0, v);
     }
     Ok(async move {
-      stream.clone().write_msg_bytes(&serialized).await?;
+      let cancel = stream.cancel.clone();
+      let result = stream
+        .clone()
+        .write_msg_bytes(&serialized)
+        .or_cancel(cancel)
+        .await;
+      // adjust count even on error
       stream
         .queued_bytes
         .fetch_sub(serialized.len(), std::sync::atomic::Ordering::Relaxed);
+      result??;
       Ok(())
     })
   }

ext/node/polyfills/internal/child_process.ts

@@ -1339,7 +1339,7 @@ export function setupChannel(target: any, ipc: number) {
           }
         }
 
-        process.nextTick(handleMessage, msg);
+        nextTick(handleMessage, msg);
       }
     } catch (err) {
       if (
@@ -1400,7 +1400,7 @@ export function setupChannel(target: any, ipc: number) {
     if (!target.connected) {
       const err = new ERR_IPC_CHANNEL_CLOSED();
       if (typeof callback === "function") {
-        process.nextTick(callback, err);
+        nextTick(callback, err);
       } else {
         nextTick(() => target.emit("error", err));
       }
@@ -1416,7 +1416,18 @@ export function setupChannel(target: any, ipc: number) {
       .then(() => {
         control.unrefCounted();
         if (callback) {
-          process.nextTick(callback, null);
+          nextTick(callback, null);
+        }
+      }, (err: Error) => {
+        control.unrefCounted();
+        if (err instanceof Deno.errors.Interrupted) {
+          // Channel closed on us mid-write.
+        } else {
+          if (typeof callback === "function") {
+            nextTick(callback, err);
+          } else {
+            nextTick(() => target.emit("error", err));
+          }
         }
       });
     return queueOk[0];
@@ -1433,7 +1444,7 @@ export function setupChannel(target: any, ipc: number) {
     target.connected = false;
     target[kCanDisconnect] = false;
     control[kControlDisconnect]();
-    process.nextTick(() => {
+    nextTick(() => {
       target.channel = null;
       core.close(ipc);
       target.emit("disconnect");