test: ensure abort signals are respected when consuming lists of events (#3094)

Adds tests to ensure we can consume lists of events with a delay.

---------

Co-authored-by: Daniel N <[email protected]>
Co-authored-by: Alex Potsides <[email protected]>

Author: 2color

packages/kad-dht/test/content-routing.spec.ts

@@ -6,7 +6,7 @@ import delay from 'delay'
 import all from 'it-all'
 import drain from 'it-drain'
 import sinon from 'sinon'
-import { MessageType } from '../src/index.js'
+import { MessageType, type QueryEvent } from '../src/index.js'
 import * as kadUtils from '../src/utils.js'
 import { createValues } from './utils/create-values.js'
 import { sortDHTs } from './utils/sort-closest-peers.js'
@@ -253,4 +253,105 @@ describe('content routing', () => {
     }, {}))
     expect(provs).to.have.length(1)
   })
+
+  it('aborts provide operation when abort signal is triggered before starting', async function () {
+    this.timeout(20 * 1000)
+
+    const dhts = await sortDHTs(await Promise.all([
+      testDHT.spawn(),
+      testDHT.spawn(),
+      testDHT.spawn(),
+      testDHT.spawn()
+    ]), await kadUtils.convertBuffer(cid.multihash.bytes))
+
+    // Spy on network.sendMessage to verify it's not called after abort
+    const sendMessageSpy = sinon.spy(dhts[3].network, 'sendMessage')
+
+    // Connect peers
+    await Promise.all([
+      testDHT.connect(dhts[0], dhts[1]),
+      testDHT.connect(dhts[1], dhts[2]),
+      testDHT.connect(dhts[2], dhts[3])
+    ])
+
+    const controller = new AbortController()
+    controller.abort()
+
+    const generator = dhts[3].provide(cid, { signal: controller.signal })
+    await expect(all(generator)).to.eventually.be.rejected
+      .with.property('message').that.include('aborted')
+
+    expect(sendMessageSpy.called).to.be.false('sendMessage should not be called when aborted')
+  })
+
+  it('properly terminates generator when a non-immediate abort signal is triggered', async function () {
+    this.timeout(20 * 1000)
+
+    const dhts = await sortDHTs(await Promise.all([
+      testDHT.spawn(),
+      testDHT.spawn(),
+      testDHT.spawn(),
+      testDHT.spawn()
+    ]), await kadUtils.convertBuffer(cid.multihash.bytes))
+
+    // Connect peers
+    await Promise.all([
+      testDHT.connect(dhts[0], dhts[1]),
+      testDHT.connect(dhts[1], dhts[2]),
+      testDHT.connect(dhts[2], dhts[3])
+    ])
+
+    const sendMessageSpy = sinon.spy(dhts[3].network, 'sendMessage')
+
+    const controller = new AbortController()
+
+    // Start the provide operation
+    const generator = dhts[3].provide(cid, { signal: controller.signal })
+
+    // We want to push the generator manually to control timing
+    const reader = async (): Promise<{ results: QueryEvent[], aborted: boolean }> => {
+      const results = []
+      try {
+        for await (const event of generator) {
+          results.push(event)
+          // After we get the first few results, abort the operation
+          if (results.length === 2) {
+            controller.abort()
+            // This delay simulates an onward consumer performing work before
+            // accepting another value from the generator
+            await delay(50)
+          }
+        }
+      } catch (err) {
+        // We expect an abort error here
+        expect(err).to.have.property('message').that.include('abort')
+        return { results, aborted: true }
+      }
+      return { results, aborted: false }
+    }
+
+    const { results, aborted } = await reader()
+
+    // We should have aborted
+    expect(aborted).to.be.true('Generator should have thrown an abort error')
+
+    // We should have received some events before the abort
+    expect(results.length).to.be.greaterThan(0, 'Should have received some events before abort')
+
+    // After aborting, if we try to get more from the generator, it should be
+    // done. Testing this requires using the original generator reference, but
+    // we've already drained it. So instead we check side effects to confirm the
+    // operation stopped.
+
+    // Wait a reasonable time for any pending operations to complete
+    await delay(500)
+
+    // Check that no new network calls were made after the abort
+    const initialMessageCalls = sendMessageSpy.callCount
+    await delay(200)
+
+    // The number of calls should not have increased
+    expect(sendMessageSpy.callCount).to.equal(initialMessageCalls,
+      'No new network calls should be made after abort')
+  })
 })