feat: inherit abort signal on piping (google#992)

* feat: inherit abort signal on piping

closes google#968

* test: add delay for nc init

* test: get free port to avoid net collisions

* test: fix flaky test

* test(cli): replace `nc` with `node:net`

Author: antongolub

package-lock.json

@@ -114,6 +114,7 @@
     "esbuild-plugin-transform-hook": "^0.1.1",
     "esbuild-plugin-utils": "^0.1.0",
     "fs-extra": "^11.2.0",
+    "get-port": "^7.1.0",
     "globby": "^14.0.2",
     "jsr": "^0.13.2",
     "madge": "^8.0.0",

package.json

@@ -337,7 +337,14 @@ export class ProcessPromise extends Promise<ProcessOutput> {
     ...args: any[]
   ): (Writable & PromiseLike<ProcessPromise & Writable>) | ProcessPromise {
     if (isStringLiteral(dest, ...args))
-      return this.pipe($({ halt: true })(dest as TemplateStringsArray, ...args))
+      return this.pipe(
+        $({
+          halt: true,
+          ac: this._snapshot.ac,
+          signal: this._snapshot.signal,
+        })(dest as TemplateStringsArray, ...args)
+      )
+
     if (isString(dest))
       throw new Error('The pipe() method does not take strings. Forgot $?')
 

src/core.ts

@@ -15,13 +15,27 @@
 import assert from 'node:assert'
 import { test, describe, before, after } from 'node:test'
 import { fileURLToPath } from 'node:url'
+import net from 'node:net'
+import getPort from 'get-port'
 import '../build/globals.js'
 import { isMain, normalizeExt } from '../build/cli.js'
 
 const __filename = fileURLToPath(import.meta.url)
 const spawn = $.spawn
 const nodeMajor = +process.versions?.node?.split('.')[0]
 const test22 = nodeMajor >= 22 ? test : test.skip
+const getServer = (resp = [], log = console.log) => {
+  const server = net.createServer()
+  server.on('connection', (conn) => {
+    conn.on('data', (d) => {
+      conn.write(resp.shift() || 'pong')
+    })
+  })
+  server.stop = () => new Promise((resolve) => server.close(() => resolve()))
+  server.start = (port) =>
+    new Promise((resolve) => server.listen(port, () => resolve(server)))
+  return server
+}
 
 describe('cli', () => {
   // Helps detect unresolved ProcessPromise.
@@ -123,19 +137,22 @@ describe('cli', () => {
     assert.ok(p.stderr.endsWith(cwd + '\n'))
   })
 
-  test('scripts from https', async () => {
-    const server = $`cat ${path.resolve('test/fixtures/echo.http')} | nc -l 8080`
+  test('scripts from https 200', async () => {
+    const resp = await fs.readFile(path.resolve('test/fixtures/echo.http'))
+    const port = await getPort()
+    const server = await getServer([resp]).start(port)
     const out =
-      await $`node build/cli.js --verbose http://127.0.0.1:8080/echo.mjs`
+      await $`node build/cli.js --verbose http://127.0.0.1:${port}/echo.mjs`
     assert.match(out.stderr, /test/)
-    await server.kill()
+    await server.stop()
   })
 
-  test('scripts from https not ok', async () => {
-    const server = $`echo $'HTTP/1.1 500\n\n' | nc -l 8081`
-    const out = await $`node build/cli.js http://127.0.0.1:8081`.nothrow()
+  test('scripts from https 500', async () => {
+    const port = await getPort()
+    const server = await getServer(['HTTP/1.1 500\n\n']).listen(port)
+    const out = await $`node build/cli.js http://127.0.0.1:${port}`.nothrow()
     assert.match(out.stderr, /Error: Can't get/)
-    await server.kill()
+    await server.stop()
   })
 
   test('scripts with no extension', async () => {

test/cli.test.js

@@ -677,6 +677,20 @@ describe('core', () => {
           assert.match(message, /The operation was aborted/)
         }
       })
+
+      test('abort signal is transmittable through pipe', async () => {
+        const ac = new AbortController()
+        const { signal } = ac
+        const p1 = $({ signal, nothrow: true })`echo test`
+        const p2 = p1.pipe`sleep 999`
+        setTimeout(() => ac.abort(), 50)
+
+        try {
+          await p2
+        } catch ({ message }) {
+          assert.match(message, /The operation was aborted/)
+        }
+      })
     })
 
     describe('kill()', () => {