NET-193: Storage config propagation (#227)

Modified addToStorageNode to wait until the information has been propagated to the Broker. At that point, we can be sure that any published data will be stored. To enable the propagation check, new client options were added: storageNode.address and storageNode.url.

Updated also all relevant integration tests: if a test does a resend for some stream, that stream must have been added to a StorageNode.

Author: teogeb

package-lock.json

@@ -117,7 +117,6 @@
     "pretty-bytes": "^5.6.0",
     "process": "^0.11.10",
     "sinon": "^9.2.4",
-    "streamr-test-utils": "^1.3.1",
     "terser-webpack-plugin": "^5.1.1",
     "ts-jest": "^26.5.1",
     "ts-loader": "^8.0.17",
@@ -160,6 +159,7 @@
     "quick-lru": "^6.0.0",
     "readable-stream": "^3.6.0",
     "streamr-client-protocol": "^8.0.0-beta.2",
+    "streamr-test-utils": "^1.3.1",
     "ts-toolbelt": "^9.3.12",
     "uuid": "^8.3.2",
     "webpack-node-externals": "^2.5.2",

package.json

@@ -71,6 +71,10 @@ export type StrictStreamrClientOptions = {
         templateMainnetAddress: EthereumAddress
         templateSidechainAddress: EthereumAddress
     },
+    storageNode: {
+        address: EthereumAddress
+        url: string
+    }
     cache: {
         maxSize: number,
         maxAge: number
@@ -139,6 +143,10 @@ export const STREAM_CLIENT_DEFAULTS: StrictStreamrClientOptions = {
         templateMainnetAddress: '0x5FE790E3751dd775Cb92e9086Acd34a2adeB8C7b',
         templateSidechainAddress: '0xf1E9d6E254BeA3f0129018AcA1A50AEcb7D528be',
     },
+    storageNode: {
+        address: '0x31546eEA76F2B2b3C5cC06B1c93601dc35c9D916',
+        url: 'https://corea1.streamr.network:8001'
+    },
     cache: {
         maxSize: 10000,
         maxAge: 30 * 60 * 1000, // 30 minutes
@@ -160,7 +168,8 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
         'dataUnion.factoryMainnetAddress',
         'dataUnion.factorySidechainAddress',
         'dataUnion.templateMainnetAddress',
-        'dataUnion.templateSidechainAddress'
+        'dataUnion.templateSidechainAddress',
+        'storageNode.address'
     ])
 
     const options: StrictStreamrClientOptions = {
@@ -174,7 +183,7 @@ export default function ClientConfig(opts: StreamrClientOptions = {}) {
             ...STREAM_CLIENT_DEFAULTS.cache,
             ...opts.cache,
         }
-        // NOTE: sidechain is not merged with the defaults
+        // NOTE: sidechain and storageNode settings are not merged with the defaults
     }
 
     const parts = options.url!.split('?')

src/Config.ts

@@ -7,7 +7,7 @@ import debugFactory from 'debug'
 import { getEndpointUrl } from '../utils'
 import { validateOptions } from '../stream/utils'
 import { Stream, StreamOperation, StreamProperties } from '../stream'
-import StreamPart from '../stream/StreamPart'
+import { StreamPart } from '../stream/StreamPart'
 import { isKeyExchangeStream } from '../stream/KeyExchange'
 
 import authFetch, { ErrorCode, NotFoundError } from './authFetch'

src/rest/StreamEndpoints.ts

@@ -1,6 +1,6 @@
 import { EthereumAddress } from '../types'
 
-export default class StorageNode {
+export class StorageNode {
 
     private _address: EthereumAddress
 

src/stream/StorageNode.ts

@@ -1,4 +1,4 @@
-export default class StreamPart {
+export class StreamPart {
 
     _streamId: string
     _streamPartition: number

src/stream/StreamPart.ts

@@ -1,7 +1,10 @@
+import fetch from 'node-fetch'
+import { getAddress } from '@ethersproject/address'
+import { waitForCondition } from 'streamr-test-utils'
 import { getEndpointUrl } from '../utils'
 import authFetch from '../rest/authFetch'
 
-import StorageNode from './StorageNode'
+import { StorageNode } from './StorageNode'
 import { StreamrClient } from '../StreamrClient'
 
 // TODO explicit types: e.g. we never provide both streamId and id, or both streamPartition and partition
@@ -221,6 +224,12 @@ export class Stream {
     }
 
     async addToStorageNode(address: string) {
+        // currently we support only one storage node
+        // -> we can validate that the given address is that address
+        // -> remove this comparison when we start to support multiple storage nodes
+        if (getAddress(address) !== this._client.options.storageNode.address) {
+            throw new Error('Unknown storage node: ' + address)
+        }
         await authFetch(
             getEndpointUrl(this._client.options.restUrl, 'streams', this.id, 'storageNodes'),
             this._client.session,
@@ -231,6 +240,24 @@ export class Stream {
                 })
             },
         )
+        // wait for propagation: the storage node sees the database change in E&E and
+        // is ready to store the any stream data which we publish
+        const TIMEOUT = 30 * 1000
+        const POLL_INTERVAL = 500
+        await waitForCondition(() => this.isStreamStoredInStorageNode(this.id), TIMEOUT, POLL_INTERVAL,
+            () => `Propagation timeout when adding stream to a storage node: ${this.id}`)
+    }
+
+    private async isStreamStoredInStorageNode(streamId: string) {
+        const url = `${this._client.options.storageNode.url}/api/v1/streams/${encodeURIComponent(streamId)}/storage/partitions/0`
+        const response = await fetch(url)
+        if (response.status === 200) {
+            return true
+        }
+        if (response.status === 404) { // eslint-disable-line padding-line-between-statements
+            return false
+        }
+        throw new Error(`Unexpected response code ${response.status} when fetching stream storage status`)
     }
 
     async removeFromStorageNode(address: string) {

src/stream/index.ts

@@ -106,6 +106,8 @@ describe('decryption', () => {
             requireEncryptedData: true,
         })
 
+        await stream.addToStorageNode(config.clientOptions.storageNode.address)
+
         publishTestMessages = getPublishTestMessages(client, {
             stream
         })

test/integration/Encryption.test.js

@@ -47,6 +47,7 @@ describeRepeats('GapFill with resends', () => {
             requireSignedData: true,
             name: uid('stream')
         })
+        await stream.addToStorageNode(config.clientOptions.storageNode.address)
 
         client.debug('connecting before test <<')
         publishTestMessages = getPublishTestMessages(client, stream.id)

test/integration/GapFill.test.ts

@@ -50,6 +50,7 @@ describeRepeats('PubSub with multiple clients', () => {
         stream = await mainClient.createStream({
             name: uid('stream')
         })
+        await stream.addToStorageNode(config.clientOptions.storageNode.address)
     })
 
     afterEach(async () => {