feat(sdk): Add content delivery layer buffer configurations (#3305)

## Background

There are two different buffers related to this PR. The propagation
buffer and connection buffers. The propagation buffer is used when a
node is broadcasting data but does not have neighbors to send the data
to. The connection buffer is used when a new connection is formed in the
network.

## Summary

We can now configure these two options for the content delivery layer:
- `streamPartitionMaxPropagationBufferSize` can be used to set the size
of the propagation buffer.
- `contentDeliveryBufferWhileConnecting` can be used to stop buffering
while a connection is being formed.

## Related changes

- renamed `doNotBufferWhileConnecting` to `bufferWhileConnecting` in the
`trackerless-network` and `dht` packages.

## Example use case

In a case where a publisher joins the network before subscribers the
publisher stores the data to a buffer as it cannot send it to anyone.
When a subscriber joins the content of the buffer is sent to the
subscriber first and after that the real-time data is sent. If we don't
want that the buffer content is sent then we can use set
`streamPartitionMaxBufferSize` to 0 to avoid propagation buffering.

Additionally if the opening of connections takes long it is possible to
disable buffering on connecting connections to avoid sending of old
messages by setting the `contentDeliveryBufferWhileConnecting` as false.

---------

Co-authored-by: Teo Gebhard <[email protected]>

Author: juslesan

CHANGELOG.md

@@ -17,6 +17,7 @@ Changes before Tatum release are not documented in this file.
 - Add `StreamrClient#publishRaw()` for publishing raw messages (https://github.com/streamr-dev/network/pull/3280)
 - Add new `keys` configuration to the `encryption` section (https://github.com/streamr-dev/network/pull/3284)
 - Add new `validation` configuration section (https://github.com/streamr-dev/network/pull/3302)
+- Add new configuration options for controlling content delivery buffering (https://github.com/streamr-dev/network/pull/3305)
 
 #### Changed
 

packages/dht/src/connection/ConnectionManager.ts

@@ -309,7 +309,7 @@ export class ConnectionManager extends EventEmitter<TransportEvents> implements
 
         if (this.endpoints.get(nodeId)!.connected) {
             (connection as ManagedConnection).send(binary)
-        } else if (!opts.doNotBufferWhileConnecting) {
+        } else if (opts.bufferWhileConnecting) {
             return (this.endpoints.get(nodeId)! as ConnectingEndpoint).buffer.push(binary)
         }
     }

packages/dht/src/rpc-protocol/DhtCallContext.ts

@@ -9,7 +9,7 @@ export class DhtCallContext extends ProtoCallContext implements DhtRpcOptions {
     clientId?: number
     connect?: boolean
     sendIfStopped?: boolean
-    doNotBufferWhileConnecting?: boolean
+    bufferWhileConnecting?: boolean
     //used in incoming calls
     incomingSourceDescriptor?: PeerDescriptor
 }

packages/dht/src/rpc-protocol/DhtRpcOptions.ts

@@ -7,5 +7,5 @@ export interface DhtRpcOptions extends ProtoRpcOptions {
     clientId?: number
     connect?: boolean
     sendIfStopped?: boolean
-    doNotBufferWhileConnecting?: boolean
+    bufferWhileConnecting?: boolean
 }

packages/dht/src/transport/ITransport.ts

@@ -9,13 +9,13 @@ export interface TransportEvents {
 export interface SendOptions {
     connect: boolean
     sendIfStopped: boolean
-    doNotBufferWhileConnecting: boolean
+    bufferWhileConnecting: boolean
 }
 
 export const DEFAULT_SEND_OPTIONS = {
     connect: true,
     sendIfStopped: false,
-    doNotBufferWhileConnecting: false
+    bufferWhileConnecting: true
 }
 
 export interface ITransport {

packages/dht/src/transport/RoutingRpcCommunicator.ts

@@ -47,11 +47,11 @@ export class RoutingRpcCommunicator extends RpcCommunicator<DhtCallContext> {
                     // TODO maybe this options could be removed?
                     sendIfStopped: true,
                     // Responses should be buffered if necessary
-                    doNotBufferWhileConnecting: false
+                    bufferWhileConnecting: true
                 } : {
                     connect: callContext?.connect ?? DEFAULT_SEND_OPTIONS.connect,
                     sendIfStopped: callContext?.sendIfStopped ?? DEFAULT_SEND_OPTIONS.sendIfStopped,
-                    doNotBufferWhileConnecting: callContext?.doNotBufferWhileConnecting ?? DEFAULT_SEND_OPTIONS.doNotBufferWhileConnecting
+                    bufferWhileConnecting: callContext?.bufferWhileConnecting ?? DEFAULT_SEND_OPTIONS.bufferWhileConnecting
                 }
             return sendFn(message, sendOpts)
         })

packages/dht/test/integration/ConnectionManager.test.ts

@@ -523,7 +523,7 @@ describe('ConnectionManager', () => {
 
     })
 
-    it('send doNotBufferWhileConnecting', async () => {
+    it('send bufferWhileConnecting as false', async () => {
         const connectionManager1 = createConnectionManager({
             transport: mockTransport,
             websocketHost: '127.0.0.1',
@@ -572,7 +572,7 @@ describe('ConnectionManager', () => {
         const sendOptions = {
             connect: true,
             sendIfStopped: false,
-            doNotBufferWhileConnecting: true
+            bufferWhileConnecting: false
         }
         await Promise.all([connectedPromise1, connectedPromise2, connectionManager2.send(msg, sendOptions)])
         await wait(1000)

packages/sdk/src/Config.ts

@@ -183,6 +183,16 @@ export interface NetworkNodeConfig {
      */
     streamPartitionMinPropagationTargets?: number
 
+    /**
+     * The maximum number of messages that can be buffered in the propagation buffer.
+     */
+    streamPartitionMaxPropagationBufferSize?: number
+
+    /**
+     * Whether to buffer while connecting in the content delivery layer.
+     */
+    contentDeliveryBufferWhileConnecting?: boolean
+
     /**
      * Whether to accept proxy connections. Enabling this option allows
      * this network node to act as proxy on behalf of other nodes / clients.

packages/sdk/src/NetworkNodeFacade.ts

@@ -17,6 +17,7 @@ import {
 import { Logger, MetricsContext, StreamPartID, StreamPartIDUtils, UserID } from '@streamr/utils'
 import EventEmitter from 'eventemitter3'
 import pull from 'lodash/pull'
+import omit from 'lodash/omit'
 import { Lifecycle, inject, scoped } from 'tsyringe'
 import { Identity, IdentityInjectionToken } from './identity/Identity'
 import { ConfigInjectionToken, NetworkPeerDescriptor, StrictStreamrClientConfig } from './Config'
@@ -140,7 +141,10 @@ export class NetworkNodeFacade {
                     ? this.config.network.controlLayer.websocketPortRange
                     : undefined
             },
-            networkNode: this.config.network.node,
+            networkNode: {
+                ...omit(this.config.network.node, 'contentDeliveryBufferWhileConnecting'),
+                bufferWhileConnecting: this.config.network.node.contentDeliveryBufferWhileConnecting
+            },
             metricsContext: new MetricsContext()
         }
     }

packages/sdk/src/config.schema.json

@@ -275,6 +275,14 @@
                             "type": "number",
                             "default": 2
                         },
+                        "streamPartitionMaxPropagationBufferSize": {
+                            "type": "number",
+                            "default": 150
+                        },
+                        "contentDeliveryBufferWhileConnecting": {
+                            "type": "boolean",
+                            "default": true
+                        },
                         "acceptProxyConnections": {
                             "type": "boolean",
                             "default": false