Merge pull request #200 from streamr-dev/NET-181-resend-gapfill

Fix resend gapfills, improve pipeline error handling

Author: timoxley

jest.config.js

@@ -60,10 +60,10 @@ module.exports = {
     // globals: {},
 
     // An array of directory names to be searched recursively up from the requiring module's location
-    // moduleDirectories: [
-    //   "node_modules"
-    // ],
-
+    moduleDirectories: [
+        'node_modules',
+        path.resolve('./node_modules'), // makes npm link work.
+    ],
     // An array of file extensions your modules use
     // moduleFileExtensions: [
     //   "js",

jest.setup.js

@@ -1,16 +1,16 @@
-const Debug = require('debug')
-const GitRevisionPlugin = require('git-revision-webpack-plugin')
+import GitRevisionPlugin from 'git-revision-webpack-plugin'
+import Debug from 'debug'
 
 const pkg = require('./package.json')
 
-if (process.env.DEBUG_CONSOLE) {
-    // Use debug as console log
-    // This prevents jest messing with console output
-    // Ensuring debug messages are printed alongside console messages, in the correct order
-    console.log = Debug('Streamr::CONSOLE') // eslint-disable-line no-console
-}
+export default async () => {
+    if (process.env.DEBUG_CONSOLE) {
+        // Use debug as console log
+        // This prevents jest messing with console output
+        // Ensuring debug messages are printed alongside console messages, in the correct order
+        console.log = Debug('Streamr::CONSOLE') // eslint-disable-line no-console
+    }
 
-module.exports = async () => {
     if (!process.env.GIT_VERSION) {
         const gitRevisionPlugin = new GitRevisionPlugin()
         const [GIT_VERSION, GIT_COMMITHASH, GIT_BRANCH] = await Promise.all([

package-lock.json

@@ -29,7 +29,7 @@
     "test-unit": "jest test/unit --detectOpenHandles",
     "coverage": "jest --coverage",
     "test-integration": "jest --forceExit test/integration",
-    "test-integration-no-resend": "jest --testTimeout=10000 --testPathIgnorePatterns='resend|Resend' --testNamePattern='^((?!(resend|Resend|resent|Resent)).)*$' test/integration/*.test.js",
+    "test-integration-no-resend": "jest --testTimeout=10000 --testPathIgnorePatterns='resend|Resend' --testNamePattern='^((?!(resend|Resend|resent|Resent|gap|Gap)).)*$' test/integration/*.test.js",
     "test-integration-resend": "jest --testTimeout=15000 --testNamePattern='(resend|Resend|resent|Resent)' test/integration/*.test.js",
     "test-integration-dataunions": "jest --testTimeout=15000 --runInBand test/integration/DataUnionEndpoints",
     "test-flakey": "jest --forceExit test/flakey/*",
@@ -49,7 +49,7 @@
     "@babel/plugin-proposal-class-properties": "^7.12.1",
     "@babel/plugin-transform-classes": "^7.12.1",
     "@babel/plugin-transform-modules-commonjs": "^7.12.1",
-    "@babel/plugin-transform-runtime": "^7.12.10",
+    "@babel/plugin-transform-runtime": "^7.12.15",
     "@babel/preset-env": "^7.12.11",
     "@babel/preset-typescript": "^7.12.13",
     "@types/debug": "^4.1.5",
@@ -87,7 +87,8 @@
   },
   "#IMPORTANT": "babel-runtime must be in dependencies, not devDependencies",
   "dependencies": {
-    "@babel/runtime": "^7.12.5",
+    "@babel/runtime": "^7.12.13",
+    "@babel/runtime-corejs3": "^7.12.13",
     "@ethersproject/address": "^5.0.9",
     "@ethersproject/bignumber": "^5.0.13",
     "@ethersproject/bytes": "^5.0.9",
@@ -112,7 +113,7 @@
     "qs": "^6.9.6",
     "quick-lru": "^5.1.1",
     "readable-stream": "^3.6.0",
-    "streamr-client-protocol": "^7.1.2",
+    "streamr-client-protocol": "^8.0.0-beta.2",
     "typescript": "^4.1.4",
     "uuid": "^8.3.2",
     "webpack-node-externals": "^2.5.2",

package.json

@@ -1,5 +1,4 @@
 import qs from 'qs'
-// @ts-expect-error
 import { ControlLayer, MessageLayer } from 'streamr-client-protocol'
 import Debug from 'debug'
 
@@ -28,6 +27,7 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
         orderMessages: true,
         retryResendAfter: 5000,
         gapFillTimeout: 5000,
+        maxGapRequests: 5,
         maxPublishQueueSize: 10000,
 
         // Encryption options

src/Config.ts

@@ -11,7 +11,7 @@ const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms))
 // add global support for pretty millisecond formatting with %n
 Debug.formatters.n = (v) => Debug.humanize(v)
 
-class ConnectionError extends Error {
+export class ConnectionError extends Error {
     constructor(err, ...args) {
         if (err instanceof ConnectionError) {
             return err

src/Connection.js

@@ -44,15 +44,23 @@ export default class Session extends EventEmitter {
         // TODO: move loginFunction to StreamrClient constructor where "auth type" is checked
         if (typeof this.options.privateKey !== 'undefined') {
             const wallet = new Wallet(this.options.privateKey)
-            this.loginFunction = async () => this._client.loginEndpoints.loginWithChallengeResponse((d: string) => wallet.signMessage(d), wallet.address)
+            this.loginFunction = async () => (
+                this._client.loginEndpoints.loginWithChallengeResponse((d: string) => wallet.signMessage(d), wallet.address)
+            )
         } else if (typeof this.options.ethereum !== 'undefined') {
             const provider = new Web3Provider(this.options.ethereum)
             const signer = provider.getSigner()
-            this.loginFunction = async () => this._client.loginEndpoints.loginWithChallengeResponse((d: string) => signer.signMessage(d), await signer.getAddress())
+            this.loginFunction = async () => (
+                this._client.loginEndpoints.loginWithChallengeResponse((d: string) => signer.signMessage(d), await signer.getAddress())
+            )
         } else if (typeof this.options.apiKey !== 'undefined') {
-            this.loginFunction = async () => this._client.loginEndpoints.loginWithApiKey(this.options.apiKey!)
+            this.loginFunction = async () => (
+                this._client.loginEndpoints.loginWithApiKey(this.options.apiKey!)
+            )
         } else if (typeof this.options.username !== 'undefined' && typeof this.options.password !== 'undefined') {
-            this.loginFunction = async () => this._client.loginEndpoints.loginWithUsernamePassword(this.options.username!, this.options.password!)
+            this.loginFunction = async () => (
+                this._client.loginEndpoints.loginWithUsernamePassword(this.options.username!, this.options.password!)
+            )
         } else {
             if (!this.options.sessionToken) {
                 this.options.unauthenticated = true

src/Session.ts

@@ -1,5 +1,4 @@
 import EventEmitter from 'eventemitter3'
-// @ts-expect-error
 import { ControlLayer } from 'streamr-client-protocol'
 import Debug from 'debug'
 
@@ -8,7 +7,7 @@ import { validateOptions } from './stream/utils'
 import Config from './Config'
 import StreamrEthereum from './Ethereum'
 import Session from './Session'
-import Connection from './Connection'
+import Connection, { ConnectionError } from './Connection'
 import Publisher from './publish'
 import Subscriber from './subscribe'
 import { getUserId } from './user'
@@ -237,12 +236,12 @@ export default class StreamrClient extends EventEmitter {
     }
 
     onConnectionError(err: Todo) {
-        this.emit('error', new Connection.ConnectionError(err))
+        this.emit('error', new ConnectionError(err))
     }
 
     getErrorEmitter(source: Todo) {
         return (err: Todo) => {
-            if (!(err instanceof Connection.ConnectionError || err.reason instanceof Connection.ConnectionError)) {
+            if (!(err instanceof ConnectionError || err.reason instanceof ConnectionError)) {
                 // emit non-connection errors
                 this.emit('error', err)
             } else {

src/StreamrClient.ts

@@ -433,7 +433,12 @@ async function transportSignatures(client: StreamrClient, messageHash: Todo, opt
 
 // template for withdraw functions
 // client could be replaced with AMB (mainnet and sidechain)
-async function untilWithdrawIsComplete(client: StreamrClient, getWithdrawTxFunc: (options: DataUnionOptions) => Todo, getBalanceFunc: (options: DataUnionOptions) => Todo, options: DataUnionOptions = {}) {
+async function untilWithdrawIsComplete(
+    client: StreamrClient,
+    getWithdrawTxFunc: (options: DataUnionOptions) => Todo,
+    getBalanceFunc: (options: DataUnionOptions) => Todo,
+    options: DataUnionOptions = {}
+) {
     const {
         pollingIntervalMs = 1000,
         retryTimeoutMs = 60000,

src/rest/DataUnionEndpoints.ts

@@ -217,11 +217,12 @@ export class StreamEndpoints {
             streamPartition,
             count,
         })
-        const query = {
-            count,
-        }
 
-        const url = getEndpointUrl(this.client.options.restUrl, 'streams', streamId, 'data', 'partitions', streamPartition, 'last') + `?${qs.stringify(query)}`
+        const url = (
+            getEndpointUrl(this.client.options.restUrl, 'streams', streamId, 'data', 'partitions', streamPartition, 'last')
+            + `?${qs.stringify({ count })}`
+        )
+
         const json = await authFetch(url, this.client.session)
         return json
     }