test: drain destroy packets before closing out-of-order proxies

test/leaks.txt

@@ -1,3 +1,4 @@
+leak:js_add_finalizer
 leak:js_create_function
 leak:js_create_typed_function
 leak:js_create_external

test/stream.js

@@ -378,7 +378,7 @@ test('write empty buffer', async function (t) {
 })
 
 test('out of order packets', async function (t) {
-  t.plan(3)
+  t.plan(5)
 
   const u = new UDX()
 
@@ -395,13 +395,21 @@ test('out of order packets', async function (t) {
     .join('')
   let received = ''
 
-  const p = await proxy({ from: a, to: b }, async function (pkt) {
+  const p = await proxy({ from: a, to: b }, async function (pkt, source) {
+    // Consume both destroy packets before releasing the proxy port for reuse.
+    if (pkt.isDestroy && source.peer !== 'unknown') {
+      t.pass(`proxy consumed destroy packet (peer: ${source.peer})`)
+      return true
+    }
+
     // Add a random delay to every packet
     await new Promise((resolve) => setTimeout(resolve, (Math.random() * 1000) | 0))
 
     return false
   })
 
+  t.teardown(() => Promise.all([a.close(), b.close(), p.close()]))
+
   const aStream = u.createStream(1)
   const bStream = u.createStream(2)
 
@@ -418,25 +426,22 @@ test('out of order packets', async function (t) {
     if (received.length === expected.length) {
       t.alike(received, expected, 'received in order')
 
-      p.close()
       aStream.destroy()
       bStream.destroy()
     }
   })
 
   aStream.on('close', function () {
     t.pass('a stream closed')
-    b.close()
   })
 
   bStream.on('close', function () {
     t.pass('b stream closed')
-    a.close()
   })
 })
 
 test('out of order reads but can destroy (memleak test)', async function (t) {
-  t.plan(3)
+  t.plan(5)
 
   const u = new UDX()
 
@@ -448,11 +453,16 @@ test('out of order reads but can destroy (memleak test)', async function (t) {
 
   let processed = 0
 
-  const p = await proxy({ from: a, to: b }, function (pkt) {
+  const p = await proxy({ from: a, to: b }, function (pkt, source) {
+    // Consume both destroy packets before releasing the proxy port for reuse.
+    if (pkt.isDestroy && source.peer !== 'unknown') {
+      t.pass(`proxy consumed destroy packet (peer: ${source.peer})`)
+      return true
+    }
+
     if (pkt.data.toString().startsWith('a') && processed > 0) {
       // destroy with out or order packets delivered
       t.pass('close while streams have out of order state')
-      p.close()
       aStream.destroy()
       bStream.destroy()
       return true
@@ -461,6 +471,8 @@ test('out of order reads but can destroy (memleak test)', async function (t) {
     return processed++ === 0 // drop first packet
   })
 
+  t.teardown(() => Promise.all([a.close(), b.close(), p.close()]))
+
   const aStream = u.createStream(1)
   const bStream = u.createStream(2)
 
@@ -472,12 +484,10 @@ test('out of order reads but can destroy (memleak test)', async function (t) {
 
   aStream.on('close', function () {
     t.pass('a stream closed')
-    b.close()
   })
 
   bStream.on('close', function () {
     t.pass('b stream closed')
-    a.close()
   })
 })