refactor(client): Remove GapFillFailedError (#1579)

When an error occurs in message ordering, the `OrderedMsgChain` calls an `onError` callback which is registered by the `OrderMessages` class. There is a parameter in that call, but `OrderMessages` doesn't use the value.

See https://github.com/streamr-dev/network/blob/73b83c0cc2af716a3b09c71f13cb05665cee25fe/packages/client/src/subscribe/OrderMessages.ts#L51

Refactored error handler to be just a callback without arguments and removed the custom `GapFillFailedError` class.

### Open questions

In PR #1570 we changed the error handling functionality so that the pipeline can continue to process messages after error. There we consider the error case as an empty result. But if we want we could notify subscription somehow if there is an error.

If we want to notify, should we emit error only when there is a storage node error, or maybe also when we can't fill the gap (i.e. don't find the missing messages from the storage node). See: https://github.com/streamr-dev/network/blob/73b83c0cc2af716a3b09c71f13cb05665cee25fe/packages/client/test/unit/OrderedMsgChain.test.ts#L429

- If we want to notify the subscription, we should use `StreamrClientError` instead of custom `GapFillFailedError`
- It currently not possible to add error listeners for resend streams (i.e. when user calls `StreamrClient#resend()`)

Author: teogeb

packages/client/src/subscribe/ordering/GapFillFailedError.ts

@@ -1,7 +1,6 @@
 import { MessageRef, StreamMessage, StreamPartID } from '@streamr/protocol'
 import { EthereumAddress, Logger } from '@streamr/utils'
 import Heap from 'heap'
-import GapFillFailedError from './GapFillFailedError'
 
 function toMsgRefId(streamMessage: StreamMessage): MsgRefId {
     return streamMessage.getMessageRef().serialize()
@@ -128,7 +127,7 @@ class MsgChainQueue {
 export type MessageHandler = (msg: StreamMessage) => void
 export type GapHandler = (from: MessageRef, to: MessageRef, context: MsgChainContext) => void | Promise<void>
 export type OnDrain = (numMessages: number) => void
-export type OnError = (error: Error) => void
+export type OnError = () => void
 
 const logger = new Logger(module)
 
@@ -322,7 +321,7 @@ export class OrderedMsgChain {
 
             this.inOrderHandler(msg)
         } catch (err: any) {
-            this.onError(err)
+            this.onError()
         }
         return msg
     }
@@ -384,7 +383,7 @@ export class OrderedMsgChain {
             try {
                 await this.gapHandler(from, to, this.context)
             } catch (err: any) {
-                this.onError(err)
+                this.onError()
             }
         } else {
             this.onGapFillsExhausted()
@@ -411,7 +410,7 @@ export class OrderedMsgChain {
         // skip gap, allow queue processing to continue
         this.lastOrderedMsgRef = msg.getPreviousMessageRef()
         if (this.isGapHandlingEnabled()) {
-            this.onError(new GapFillFailedError(from, to, this.context, maxGapRequests))
+            this.onError()
         }
 
         this.clearGap()

packages/client/src/subscribe/ordering/OrderedMsgChain.ts

@@ -3,7 +3,6 @@ import { Defer, EthereumAddress, toEthereumAddress } from '@streamr/utils'
 import assert from 'assert'
 import { shuffle } from 'lodash'
 import { createSignedMessage } from '../../src/publish/MessageFactory'
-import GapFillFailedError from '../../src/subscribe/ordering/GapFillFailedError'
 import { MsgChainContext, OrderedMsgChain } from '../../src/subscribe/ordering/OrderedMsgChain'
 import { createRandomAuthentication } from '../test-utils/utils'
 
@@ -398,17 +397,11 @@ describe('OrderedMsgChain', () => {
     })
 
     describe('maxGapRequests', () => {
-        it('call the gap handler maxGapRequests times and then fails with GapFillFailedError', (done) => {
+        it('call the gap handler maxGapRequests times and then fails with error', (done) => {
             let counter = 0
-            const onError = (err: Error) => {
-                expect(err).toBeInstanceOf(GapFillFailedError)
-                if (err instanceof GapFillFailedError) {
-                    expect(err.from.serialize()).toEqual('[1,1]')
-                    expect(err.to.serialize()).toEqual('[2,0]')
-                    expect(err.context).toEqual(CONTEXT)
-                    // @ts-expect-error private method
-                    expect(counter).toBe(util.maxGapRequests)
-                }
+            const onError = () => {
+                // @ts-expect-error private method
+                expect(counter).toBe(util.maxGapRequests)
                 done()
             }
             util = new OrderedMsgChain(
@@ -426,12 +419,11 @@ describe('OrderedMsgChain', () => {
             util.add(msg3)
         })
 
-        it('after maxGapRequests OrderingUtil gives up on filling gap with GapFillFailedError "error" event', (done) => {
+        it('after maxGapRequests OrderingUtil gives up on filling gap with "error" event', (done) => {
             const received: StreamMessage[] = []
             const onGap = jest.fn()
             let onErrorCallCount = 0
-            const onError = (err: Error) => {
-                if (!(err instanceof GapFillFailedError)) { throw err }
+            const onError = () => {
                 onGap()
                 if (onErrorCallCount === 0) {
                     setImmediate(() => {