feat(client): Abort resends when OrderMessages stops (#1571)

When `OrderMessages` stops, it aborts any ongoing resend HTTP `fetch` requests. Before this PR it stopped the iteration of resend streams but didn't stop the `fetch` requests.

## Changes

The `OrderMessages` now uses `AbortController` to store the `done` state. That signal is provided to the `fetchHttpStream` as an argument, and therefore the all `fetch` operations are stopped when `OrderMessages.stop()` is called.

Manual handling the list of ongoing resends is no longer needed in `OrderMessages`.

Added also `this.outBuffer.endWrite()` call to `OrderMessages.stop()` and `this.isDone()` check to `OrderMessages#addToOrderingUtil`. These are just optimizations and not strictly needed. In the current use case where `OrderMessages.stop()` is called only for stopped pipelines (see`messageStream.onBeforeFinally`) these optimizations don't have any effect.

### ComposedAbortSignal

Enhanced `composeAbortSignals` utility function so that the returned object contains a `destroy()` function. To avoid memory leaks it would be good to call the method when the composed signal is no longer used so that we can remove the listeners associated to the source signals. (If all source signals have short lifespan or there are just few listeners, the cleanup is not needed in practice.)

### Future improvements

- Is `stream.once('close', () => { abortController.abort() })` actually needed, or could we remove it?

Author: teogeb

packages/client/src/subscribe/OrderMessages.ts

@@ -17,10 +17,9 @@ import OrderingUtil from './ordering/OrderingUtil'
  */
 export class OrderMessages {
 
-    private done = false
+    private abortController: AbortController = new AbortController()
     private inputClosed = false
     private enabled = true
-    private readonly resendStreams = new Set<PushPipeline<StreamMessage, StreamMessage>>() // holds outstanding resends for cleanup
     private readonly outBuffer = new PushBuffer<StreamMessage>()
     private readonly orderingUtil: OrderingUtil
     private readonly resends: Resends
@@ -57,7 +56,7 @@ export class OrderMessages {
     }
 
     async onGap(from: MessageRef, to: MessageRef, context: MsgChainContext): Promise<void> {
-        if (this.done || !this.enabled) { return }
+        if (this.isDone() || !this.enabled) { return }
         this.logger.debug('Encountered gap', {
             streamPartId: this.streamPartId,
             context,
@@ -74,19 +73,15 @@ export class OrderMessages {
                 publisherId: context.publisherId,
                 msgChainId: context.msgChainId,
                 raw: true
-            }, this.getStorageNodes)
-            resendMessageStream.onFinally.listen(() => {
-                this.resendStreams.delete(resendMessageStream)
-            })
-            this.resendStreams.add(resendMessageStream)
-            if (this.done) { return }
+            }, this.getStorageNodes, this.abortController.signal)
+            if (this.isDone()) { return }
 
             for await (const streamMessage of resendMessageStream) {
-                if (this.done) { return }
+                if (this.isDone()) { return }
                 this.orderingUtil.add(streamMessage)
             }
         } catch (err) {
-            if (this.done) { return }
+            if (this.isDone()) { return }
 
             if (err.code === 'NO_STORAGE_NODES') {
                 // ignore NO_STORAGE_NODES errors
@@ -96,23 +91,24 @@ export class OrderMessages {
             } else {
                 throw err
             }
-        } finally {
-            if (resendMessageStream != null) {
-                this.resendStreams.delete(resendMessageStream)
-            }
         }
     }
 
     onOrdered(orderedMessage: StreamMessage): void {
-        if (this.outBuffer.isDone() || this.done) {
+        if (this.outBuffer.isDone() || this.isDone()) {
             return
         }
 
         this.outBuffer.push(orderedMessage)
     }
 
     stop(): void {
-        this.done = true
+        this.outBuffer.endWrite()
+        this.abortController.abort()
+    }
+
+    private isDone() {
+        return this.abortController.signal.aborted
     }
 
     private maybeClose(): void {
@@ -131,7 +127,9 @@ export class OrderMessages {
     private async addToOrderingUtil(src: AsyncGenerator<StreamMessage>): Promise<void> {
         try {
             for await (const msg of src) {
-                this.orderingUtil.add(msg)
+                if (!this.isDone()) {
+                    this.orderingUtil.add(msg)
+                }
             }
             this.inputClosed = true
             this.maybeClose()
@@ -146,10 +144,9 @@ export class OrderMessages {
                 this.addToOrderingUtil(src)
                 yield* this.outBuffer
             } finally {
-                this.done = true
+                this.stop()
                 this.orderingUtil.clearGaps()
-                this.resendStreams.forEach((s) => s.end())
-                this.resendStreams.clear()
+                // TODO why there are two clearGaps() calls?
                 this.orderingUtil.clearGaps()
             }
         }.bind(this)

packages/client/src/subscribe/Resends.ts

@@ -121,7 +121,8 @@ export class Resends {
     async resend(
         streamPartId: StreamPartID,
         options: ResendOptions & { raw?: boolean },
-        getStorageNodes?: (streamId: StreamID) => Promise<EthereumAddress[]>
+        getStorageNodes?: (streamId: StreamID) => Promise<EthereumAddress[]>,
+        abortSignal?: AbortSignal
     ): Promise<PushPipeline<StreamMessage, StreamMessage>> {
         const raw = options.raw ?? false
         if (isResendLast(options)) {
@@ -132,7 +133,7 @@ export class Resends {
             }
             return this.fetchStream('last', streamPartId, {
                 count: options.last
-            }, raw, getStorageNodes)
+            }, raw, getStorageNodes, abortSignal)
         } else if (isResendRange(options)) {
             return this.fetchStream('range', streamPartId, {
                 fromTimestamp: new Date(options.from.timestamp).getTime(),
@@ -141,13 +142,13 @@ export class Resends {
                 toSequenceNumber: options.to.sequenceNumber,
                 publisherId: options.publisherId !== undefined ? toEthereumAddress(options.publisherId) : undefined,
                 msgChainId: options.msgChainId
-            }, raw, getStorageNodes)
+            }, raw, getStorageNodes, abortSignal)
         } else if (isResendFrom(options)) {
             return this.fetchStream('from', streamPartId, {
                 fromTimestamp: new Date(options.from.timestamp).getTime(),
                 fromSequenceNumber: options.from.sequenceNumber,
                 publisherId: options.publisherId !== undefined ? toEthereumAddress(options.publisherId) : undefined
-            }, raw, getStorageNodes)
+            }, raw, getStorageNodes, abortSignal)
         } else {
             throw new StreamrClientError(
                 `can not resend without valid resend options: ${JSON.stringify({ streamPartId, options })}`,
@@ -161,7 +162,8 @@ export class Resends {
         streamPartId: StreamPartID,
         query: QueryDict,
         raw: boolean,
-        getStorageNodes?: (streamId: StreamID) => Promise<EthereumAddress[]>
+        getStorageNodes?: (streamId: StreamID) => Promise<EthereumAddress[]>,
+        abortSignal?: AbortSignal
     ): Promise<PushPipeline<StreamMessage, StreamMessage>> {
         const traceId = randomString(5)
         this.logger.debug('Fetch resend data', {
@@ -191,7 +193,7 @@ export class Resends {
             getStorageNodes: async () => without(nodeAddresses, nodeAddress),
             config: (nodeAddresses.length === 1) ? { ...this.config, orderMessages: false } : this.config
         })
-        const lines = transformError(fetchHttpStream(url), getHttpErrorTransform())
+        const lines = transformError(fetchHttpStream(url, abortSignal), getHttpErrorTransform())
         setImmediate(async () => {
             let count = 0
             const messages = map(lines, (line: string) => StreamMessage.deserialize(line))

packages/client/src/utils/utils.ts

@@ -1,8 +1,9 @@
 import { ContractReceipt } from '@ethersproject/contracts'
 import { StreamID, toStreamID } from '@streamr/protocol'
-import { Logger, TheGraphClient, merge, randomString, toEthereumAddress } from '@streamr/utils'
+import { Logger, TheGraphClient, composeAbortSignals, merge, randomString, toEthereumAddress } from '@streamr/utils'
+import compact from 'lodash/compact'
 import fetch, { Response } from 'node-fetch'
-import { AbortSignal } from 'node-fetch/externals'
+import { AbortSignal as FetchAbortSignal } from 'node-fetch/externals'
 import split2 from 'split2'
 import { Readable } from 'stream'
 import LRU from '../../vendor/quick-lru'
@@ -160,12 +161,14 @@ export class FetchHttpStreamResponseError extends Error {
 
 export const fetchHttpStream = async function*(
     url: string,
-    abortController = new AbortController()
+    abortSignal?: AbortSignal
 ): AsyncGenerator<string, void, undefined> {
-    logger.debug('Send HTTP request', { url })
+    logger.debug('Send HTTP request', { url }) 
+    const abortController = new AbortController()
+    const fetchAbortSignal = composeAbortSignals(...compact([abortController.signal, abortSignal]))
     const response: Response = await fetch(url, {
         // cast is needed until this is fixed: https://github.com/node-fetch/node-fetch/issues/1652
-        signal: abortController.signal as AbortSignal
+        signal: fetchAbortSignal as FetchAbortSignal
     })
     logger.debug('Received HTTP response', {
         url,
@@ -193,5 +196,6 @@ export const fetchHttpStream = async function*(
         throw err
     } finally {
         stream?.destroy()
+        fetchAbortSignal.destroy()
     }
 }

packages/client/test/unit/OrderMessages.test.ts

@@ -1,13 +1,13 @@
-import { MessageID, StreamMessage, StreamPartIDUtils, toStreamID } from '@streamr/protocol'
+import { MessageID, StreamID, StreamMessage, StreamPartID, StreamPartIDUtils, toStreamID } from '@streamr/protocol'
 import { randomEthereumAddress } from '@streamr/test-utils'
-import { collect } from '@streamr/utils'
+import { EthereumAddress, collect } from '@streamr/utils'
 import range from 'lodash/range'
 import without from 'lodash/without'
-import { OrderMessages } from '../../src/subscribe/OrderMessages'
-import { Resends } from '../../src/subscribe/Resends'
-import { fromArray } from '../../src/utils/GeneratorUtils'
-import { PushPipeline } from '../../src/utils/PushPipeline'
-import { mockLoggerFactory } from '../test-utils/utils'
+import { OrderMessages } from './../../src/subscribe/OrderMessages'
+import { ResendOptions, Resends } from './../../src/subscribe/Resends'
+import { fromArray } from './../../src/utils/GeneratorUtils'
+import { PushPipeline } from './../../src/utils/PushPipeline'
+import { mockLoggerFactory } from './../test-utils/utils'
 
 const MESSAGE_COUNT = 7
 const STREAM_PART_ID = StreamPartIDUtils.parse('stream#0')
@@ -22,13 +22,13 @@ const CONFIG = {
     gapFillTimeout: 50
 }
 
-const createTransform = (resends: Pick<Resends, 'resend'>, config = CONFIG) => {
+const createOrderMessages = (resends: Pick<Resends, 'resend'>, config = CONFIG) => {
     return new OrderMessages(
         config as any,
         resends as any,
         STREAM_PART_ID,
         mockLoggerFactory()
-    ).transform()
+    )
 }
 
 const createMockMessages = async (): Promise<StreamMessage[]> => {
@@ -58,7 +58,7 @@ describe('OrderMessages', () => {
 
     it('no gaps', async () => {
         const msgs = await createMockMessages()
-        const transform = createTransform(undefined as any)
+        const transform = createOrderMessages(undefined as any).transform()
         const output = transform(fromArray(msgs))
         expect(await collect(output)).toEqual(msgs)
     })
@@ -69,7 +69,7 @@ describe('OrderMessages', () => {
         const resends = {
             resend: jest.fn().mockResolvedValue(createMessageStream(...missing))
         }
-        const transform = createTransform(resends)
+        const transform = createOrderMessages(resends).transform()
         const output = transform(fromArray(without(msgs, ...missing)))
         expect(await collect(output)).toEqual(msgs)
         expect(resends.resend).toBeCalledWith(
@@ -87,7 +87,8 @@ describe('OrderMessages', () => {
                 msgChainId: MSG_CHAIN_ID,
                 raw: true
             },
-            undefined
+            undefined,
+            expect.anything()
         )
     })
 
@@ -97,7 +98,7 @@ describe('OrderMessages', () => {
         const resends = {
             resend: jest.fn().mockResolvedValue(createMessageStream(...missing))
         }
-        const transform = createTransform(resends)
+        const transform = createOrderMessages(resends).transform()
         const output = transform(fromArray(without(msgs, ...missing)))
         expect(await collect(output)).toEqual(msgs)
         expect(resends.resend).toBeCalledWith(
@@ -115,7 +116,8 @@ describe('OrderMessages', () => {
                 msgChainId: MSG_CHAIN_ID,
                 raw: true
             },
-            undefined
+            undefined,
+            expect.anything()
         )
     })
 
@@ -128,7 +130,7 @@ describe('OrderMessages', () => {
                 .mockResolvedValueOnce(createMessageStream(...missing1))
                 .mockResolvedValueOnce(createMessageStream(...missing2))
         }
-        const transform = createTransform(resends)
+        const transform = createOrderMessages(resends).transform()
         const output = transform(fromArray(without(msgs, ...missing1.concat(missing2))))
         expect(await collect(output)).toEqual(msgs)
         expect(resends.resend).toHaveBeenNthCalledWith(1,
@@ -146,7 +148,8 @@ describe('OrderMessages', () => {
                 msgChainId: MSG_CHAIN_ID,
                 raw: true
             },
-            undefined
+            undefined,
+            expect.anything()
         )
         expect(resends.resend).toHaveBeenNthCalledWith(2,
             STREAM_PART_ID,
@@ -163,7 +166,8 @@ describe('OrderMessages', () => {
                 msgChainId: MSG_CHAIN_ID,
                 raw: true
             },
-            undefined
+            undefined,
+            expect.anything()
         )
     })
 
@@ -173,7 +177,7 @@ describe('OrderMessages', () => {
         const resends = {
             resend: jest.fn().mockImplementation(() => createMessageStream())
         }
-        const transform = createTransform(resends)
+        const transform = createOrderMessages(resends).transform()
         const output = transform(fromArray(without(msgs, ...missing)))
         expect(await collect(output)).toEqual(without(msgs, ...missing))
         expect(resends.resend).toBeCalledTimes(CONFIG.maxGapRequests)
@@ -185,9 +189,9 @@ describe('OrderMessages', () => {
         const resends = {
             resend: jest.fn()
         }
-        const transform = createTransform(resends, {
+        const transform = createOrderMessages(resends, {
             orderMessages: false
-        } as any)
+        } as any).transform()
         const output = transform(fromArray(without(msgs, ...missing)))
         expect(await collect(output)).toEqual(without(msgs, ...missing))
         expect(resends.resend).toBeCalledTimes(0)
@@ -209,9 +213,33 @@ describe('OrderMessages', () => {
                 .mockRejectedValueOnce(new Error('mock-error'))
                 .mockResolvedValueOnce(createMessageStream(...missing3))
         }
-        const transform = createTransform(resends)
+        const transform = createOrderMessages(resends).transform()
         const output = transform(fromArray(without(msgs, ...missing1.concat(missing2).concat(missing3))))
         expect(await collect(output)).toEqual(without(msgs, ...missing2))
         expect(resends.resend).toBeCalledTimes(2 + CONFIG.maxGapRequests)
     })
+
+    it('aborts resends when stopped', async () => {
+        const msgs = await createMockMessages()
+        const missing = msgs.filter((m) => m.getTimestamp() === 3000)
+        let orderedMessages: OrderMessages | undefined = undefined
+        let resendAborted = false
+        const resends = {
+            resend: jest.fn().mockImplementation((
+                _streamPartId: StreamPartID,
+                _options: ResendOptions & { raw?: boolean },
+                _getStorageNodes?: (streamId: StreamID) => Promise<EthereumAddress[]>,
+                abortSignal?: AbortSignal
+            ) => {
+                abortSignal!.addEventListener('abort', () => resendAborted = true)
+                orderedMessages!.stop()
+                return createMessageStream(...missing)
+            })
+        }
+        orderedMessages = createOrderMessages(resends)
+        const transform = orderedMessages.transform()
+        const output = transform(fromArray(without(msgs, ...missing)))
+        expect(await collect(output)).toEqual(msgs.filter((msg) => msg.getTimestamp() < missing[0].getTimestamp()))
+        expect(resendAborted).toBe(true)
+    })
 })

packages/client/test/unit/utils.test.ts

@@ -50,7 +50,7 @@ describe('utils', () => {
                 res.write(`foobar\n`)
             })
             const abortController = new AbortController()
-            const iterator = fetchHttpStream(server.url, abortController)[Symbol.asyncIterator]()
+            const iterator = fetchHttpStream(server.url, abortController.signal)[Symbol.asyncIterator]()
             const line = await nextValue(iterator)
             expect(line).toBe('foobar')
             abortController.abort()