test(node): OperatorPlugin inspection config options

packages/node/src/plugins/operator/OperatorPlugin.ts

@@ -57,6 +57,11 @@ export interface OperatorPluginConfig {
     }
     inspectRandomNode: {
         intervalInMs: number
+        maxInspectionCount: number
+    }
+    reviewSuspectNode: {
+        maxInspectionCount: number
+        maxDelayBeforeFirstInspectionInMs: number
     }
     closeExpiredFlags: {
         intervalInMs: number
@@ -201,6 +206,7 @@ export class OperatorPlugin extends Plugin<OperatorPluginConfig> {
                         streamPartAssignments,
                         streamrClient,
                         this.pluginConfig.heartbeatTimeoutInMs,
+                        this.pluginConfig.inspectRandomNode.maxInspectionCount,
                         async (targetOperatorContractAddress) => {
                             return streamrClient.getOperator(targetOperatorContractAddress).fetchRedundancyFactor()
                         },
@@ -240,14 +246,14 @@ export class OperatorPlugin extends Plugin<OperatorPluginConfig> {
                                     getRedundancyFactor: async (targetOperatorContractAddress) => {
                                         return streamrClient.getOperator(targetOperatorContractAddress).fetchRedundancyFactor()
                                     },
-                                    maxSleepTime: 5 * 60 * 1000,
+                                    maxDelayBeforeFirstInspectionInMs: this.pluginConfig.reviewSuspectNode.maxDelayBeforeFirstInspectionInMs,
                                     heartbeatTimeoutInMs: this.pluginConfig.heartbeatTimeoutInMs,
                                     votingPeriod: {
                                         startTime: event.votingPeriodStartTimestamp,
                                         endTime: event.votingPeriodEndTimestamp
                                     },
                                     inspectionIntervalInMs: 8 * 60 * 1000,
-                                    maxInspections: 10,
+                                    maxInspectionCount: this.pluginConfig.reviewSuspectNode.maxInspectionCount,
                                     abortSignal: this.abortController.signal
                                 })
                             }

packages/node/src/plugins/operator/config.schema.json

@@ -125,6 +125,32 @@
           "description": "How often to run (in milliseconds)",
           "minimum": 0,
           "default": 900000
+        },
+        "maxInspectionCount": {
+          "type": "integer",
+          "description": "How many rounds of analysis are performed at most",
+          "minimum": 1,
+          "default": 10
+        }
+      }
+    },
+    "reviewSuspectNode": {
+      "type": "object",
+      "description": "Review suspect node settings",
+      "additionalProperties": false,
+      "default": {},
+      "properties": {
+        "maxInspectionCount": {
+          "type": "integer",
+          "description": "How many rounds of analysis are performed at most",
+          "minimum": 1,
+          "default": 10
+        },
+        "maxDelayBeforeFirstInspectionInMs": {
+          "type": "integer",
+          "description": "The maximum time (in milliseconds) to wait before the 1st inspection round",
+          "minimum": 0,
+          "default": 300000
         }
       }
     },

packages/node/src/plugins/operator/inspectOverTime.ts

@@ -16,10 +16,10 @@ interface InspectOverTimeOpts {
     streamrClient: StreamrClient
     createOperatorFleetState: CreateOperatorFleetStateFn
     getRedundancyFactor: (operatorContractAddress: EthereumAddress) => Promise<number | undefined>
-    sleepTimeInMsBeforeFirstInspection: number
+    delayBeforeFirstInspectionInMs: number
     heartbeatTimeoutInMs: number
     inspectionIntervalInMs: number
-    maxInspections: number
+    maxInspectionCount: number
     waitUntilPassOrDone: boolean
     abortSignal: AbortSignal
     traceId: string
@@ -44,10 +44,10 @@ class InspectionOverTimeTask {
     private readonly streamrClient: StreamrClient
     private readonly createOperatorFleetState: CreateOperatorFleetStateFn
     private readonly getRedundancyFactor: (operatorContractAddress: EthereumAddress) => Promise<number | undefined>
-    private readonly sleepTimeInMsBeforeFirstInspection: number
+    private readonly delayBeforeFirstInspectionInMs: number
     private readonly heartbeatTimeoutInMs: number
     private readonly inspectionIntervalInMs: number
-    private readonly maxInspections: number
+    private readonly maxInspectionCount: number
     private readonly abortSignal: AbortSignal
     private readonly findNodesForTargetGivenFleetStateFn: FindNodesForTargetGivenFleetStateFn
     private readonly inspectTargetFn: InspectTargetFn
@@ -64,10 +64,10 @@ class InspectionOverTimeTask {
         streamrClient,
         createOperatorFleetState,
         getRedundancyFactor,
-        sleepTimeInMsBeforeFirstInspection,
+        delayBeforeFirstInspectionInMs,
         heartbeatTimeoutInMs,
         inspectionIntervalInMs,
-        maxInspections,
+        maxInspectionCount,
         abortSignal: userAbortSignal,
         traceId,
         findNodesForTargetGivenFleetStateFn = findNodesForTargetGivenFleetState,
@@ -77,10 +77,10 @@ class InspectionOverTimeTask {
         this.streamrClient = streamrClient
         this.createOperatorFleetState = createOperatorFleetState
         this.getRedundancyFactor = getRedundancyFactor
-        this.sleepTimeInMsBeforeFirstInspection = sleepTimeInMsBeforeFirstInspection
+        this.delayBeforeFirstInspectionInMs = delayBeforeFirstInspectionInMs
         this.heartbeatTimeoutInMs = heartbeatTimeoutInMs
         this.inspectionIntervalInMs = inspectionIntervalInMs
-        this.maxInspections = maxInspections
+        this.maxInspectionCount = maxInspectionCount
         this.abortSignal = composeAbortSignals(userAbortSignal, this.abortController.signal)
         this.findNodesForTargetGivenFleetStateFn = findNodesForTargetGivenFleetStateFn
         this.inspectTargetFn = inspectTargetFn
@@ -124,15 +124,15 @@ class InspectionOverTimeTask {
             target: this.target,
             heartbeatTimeoutInMs: this.heartbeatTimeoutInMs,
             inspectionIntervalInMs: this.inspectionIntervalInMs,
-            maxInspections: this.maxInspections
+            maxInspectionCount: this.maxInspectionCount
         })
 
         await this.initializeNewOperatorFleetState()
 
-        this.logger.info('Sleep', { timeInMs: this.sleepTimeInMsBeforeFirstInspection })
-        await wait(this.sleepTimeInMsBeforeFirstInspection, this.abortSignal)
+        this.logger.debug('Sleep', { timeInMs: this.delayBeforeFirstInspectionInMs })
+        await wait(this.delayBeforeFirstInspectionInMs, this.abortSignal)
 
-        for (const attemptNo of range(1, this.maxInspections + 1)) {
+        for (const attemptNo of range(1, this.maxInspectionCount + 1)) {
             const startTime = Date.now()
             this.logger.info('Inspecting target', { attemptNo, target: this.target })
 
@@ -162,7 +162,7 @@ class InspectionOverTimeTask {
                 target: this.target
             })
 
-            if (attemptNo !== this.maxInspections) {
+            if (attemptNo !== this.maxInspectionCount) {
                 // TODO: remove when NET-1169 landed;
                 //  workaround subscribe bug in @streamr/sdk (sometimes messages don't come thru to heartbeat stream)
                 if (this.fleetState?.getNodeIds().length === 0) {
@@ -176,7 +176,7 @@ class InspectionOverTimeTask {
                 }
 
                 const sleepTime = Math.max(this.inspectionIntervalInMs - timeElapsedInMs, 0)
-                this.logger.info('Sleep', { timeInMs: sleepTime })
+                this.logger.debug('Sleep', { timeInMs: sleepTime })
                 await wait(sleepTime, this.abortSignal)
             }
         }

packages/node/src/plugins/operator/inspectRandomNode.ts

@@ -11,6 +11,7 @@ export async function inspectRandomNode(
     assignments: StreamPartAssignments,
     streamrClient: StreamrClient,
     heartbeatTimeoutInMs: number,
+    maxInspectionCount: number,
     getRedundancyFactor: (operatorContractAddress: EthereumAddress) => Promise<number | undefined>,
     createOperatorFleetState: CreateOperatorFleetStateFn,
     abortSignal: AbortSignal,
@@ -31,10 +32,10 @@ export async function inspectRandomNode(
         streamrClient,
         createOperatorFleetState,
         getRedundancyFactor,
-        sleepTimeInMsBeforeFirstInspection: 0,
+        delayBeforeFirstInspectionInMs: 0,
         heartbeatTimeoutInMs,
         inspectionIntervalInMs: 8 * 60 * 1000,
-        maxInspections: 10,
+        maxInspectionCount,
         waitUntilPassOrDone: true,
         abortSignal,
         traceId

packages/node/src/plugins/operator/reviewSuspectNode.ts

@@ -14,14 +14,14 @@ export interface ReviewProcessOpts {
     streamrClient: StreamrClient
     createOperatorFleetState: CreateOperatorFleetStateFn
     getRedundancyFactor: (operatorContractAddress: EthereumAddress) => Promise<number | undefined>
-    maxSleepTime: number
+    maxDelayBeforeFirstInspectionInMs: number
     heartbeatTimeoutInMs: number
     votingPeriod: {
         startTime: number
         endTime: number
     }
     inspectionIntervalInMs: number
-    maxInspections: number
+    maxInspectionCount: number
     abortSignal: AbortSignal
 }
 
@@ -33,19 +33,19 @@ export const reviewSuspectNode = async ({
     streamrClient,
     createOperatorFleetState,
     getRedundancyFactor,
-    maxSleepTime,
+    maxDelayBeforeFirstInspectionInMs,
     heartbeatTimeoutInMs,
     votingPeriod,
     inspectionIntervalInMs,
-    maxInspections,
+    maxInspectionCount,
     abortSignal
 }: ReviewProcessOpts): Promise<void> => {
-    if (Date.now() + maxSleepTime > votingPeriod.startTime) {
-        throw new Error('Max sleep time overlaps with voting period')
+    if (Date.now() + maxDelayBeforeFirstInspectionInMs > votingPeriod.startTime) {
+        throw new Error('Max delay time overlaps with voting period')
     }
     const streamId = await myOperator.getStreamId(sponsorshipAddress)
-    // random sleep time to make sure multiple instances of voters don't all inspect at the same time
-    const sleepTimeInMsBeforeFirstInspection = random(maxSleepTime)
+    // random wait time to make sure multiple instances of voters don't all inspect at the same time
+    const delayBeforeFirstInspectionInMs = random(maxDelayBeforeFirstInspectionInMs)
     const consumeResults = inspectOverTime({
         target: {
             sponsorshipAddress: sponsorshipAddress,
@@ -55,10 +55,10 @@ export const reviewSuspectNode = async ({
         streamrClient,
         createOperatorFleetState,
         getRedundancyFactor,
-        sleepTimeInMsBeforeFirstInspection,
+        delayBeforeFirstInspectionInMs,
         heartbeatTimeoutInMs,
         inspectionIntervalInMs,
-        maxInspections,
+        maxInspectionCount,
         waitUntilPassOrDone: false,
         abortSignal,
         traceId: randomString(6)