Client: Fetcher Small Bugfixes and Log Improvements (#3024)

* Client -> Execution: less frequent cache stats

* Client -> Fetcher: log improvements

* Client -> Fetcher: fix unwanted job execution on skip case, added some no-result guards

* Client -> Fetcher: more explicit push to Readable stream

* Client -> Fetcher: add initialization debug msg

* Client -> Fetcher: tie job skipping to finished jobs instead of processed to avoid to large bulk import chunks

* Client: a bit less frequent memory stats

* Client -> Fetcher: fix occasional re-enqueue bug when length of an object is not defined, minor log improvements

* Lighsync hot fix for fetcher reenqueue logic switches

* add explicit boolean checks

---------

Co-authored-by: Scotty <[email protected]>
Co-authored-by: acolytec3 <[email protected]>

Author: 3 people

packages/client/src/service/service.ts

@@ -57,7 +57,7 @@ export class Service {
    *
    * (for info there will be somewhat reduced output)
    */
-  private STATS_INTERVAL = 1000 * 20 // 20 seconds
+  private STATS_INTERVAL = 1000 * 30 // 30 seconds
 
   /**
    * Shutdown the client when memory threshold is reached (in percent)

packages/client/src/sync/fetcher/blockfetcher.ts

@@ -104,12 +104,14 @@ export class BlockFetcher extends BlockFetcherBase<Block[], Block> {
   process(job: Job<JobTask, Block[], Block>, result: Block[]) {
     result = (job.partialResult ?? []).concat(result)
     job.partialResult = undefined
-    if (result.length === job.task.count) {
-      return result
-    } else if (result.length > 0 && result.length < job.task.count) {
-      // Save partial result to re-request missing items.
-      job.partialResult = result
-      this.debug(`Partial result received=${result.length} expected=${job.task.count}`)
+    if (result !== undefined) {
+      if (result.length === job.task.count) {
+        return result
+      } else if (result.length > 0 && result.length < job.task.count) {
+        // Save partial result to re-request missing items.
+        job.partialResult = result
+        this.debug(`Partial result received=${result.length} expected=${job.task.count}`)
+      }
     }
     return
   }

packages/client/src/sync/fetcher/blockfetcherbase.ts

@@ -114,7 +114,7 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
       this.debug(`Enqueued num=${tasks.length} tasks`)
     } else {
       this.debug(
-        `Fetcher skipping nextTasks in=${this.in.length} count=${this.count} processed=${this.processed} finished=${this.finished}`
+        `No new tasks enqueued in=${this.in.length} count=${this.count} processed=${this.processed} finished=${this.finished}`
       )
     }
   }
@@ -245,6 +245,7 @@ export abstract class BlockFetcherBase<JobResult, StorageItem> extends Fetcher<
     if ('reverse' in this) {
       str += ` reverse=${this.reverse}`
     }
+    str += ` state=${job.state}`
     return str
   }
 }

packages/client/src/sync/fetcher/fetcher.ts

@@ -88,6 +88,10 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
     this.banTime = options.banTime ?? 60000
     this.maxQueue = options.maxQueue ?? 4
 
+    this.debug(
+      `Fetcher initialized timeout=${this.timeout} interval=${this.interval} banTime=${this.banTime} maxQueue=${this.maxQueue}`
+    )
+
     this.in = new Heap({
       comparBefore: (
         a: Job<JobTask, JobResult, StorageItem>,
@@ -183,7 +187,9 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
     for (let f = this.out.peek(); f && f.index <= this.processed; ) {
       this.processed++
       const job = this.out.remove()
-      if (!this.push(job)) {
+      // Push the job to the Readable stream
+      const success = this.push(job)
+      if (!success) {
         return
       }
       f = this.out.peek()
@@ -226,17 +232,30 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
    * @param result job result
    */
   private success(job: Job<JobTask, JobResult, StorageItem>, result?: JobResult) {
-    if (job.state !== 'active') return
     let jobStr = this.jobStr(job, true)
+    if (job.state !== 'active') return
+
     let reenqueue = false
     let resultSet = ''
     if (result === undefined) {
       resultSet = 'undefined'
       reenqueue = true
     }
-    if (result !== undefined && (result as any).length === 0) {
-      resultSet = 'empty'
-      reenqueue = true
+    if (result !== undefined) {
+      if ('length' in (result as any)) {
+        if ((result as any).length === 0) {
+          resultSet = 'empty'
+          reenqueue = true
+        }
+      } else {
+        // Hot-Fix for lightsync, 2023-12-29
+        // (delete (only the if clause) in case lightsync code
+        // has been removed at some point)
+        if (!('reqId' in (result as any))) {
+          resultSet = 'unknown'
+          reenqueue = true
+        }
+      }
     }
     if (reenqueue) {
       this.debug(
@@ -253,7 +272,8 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
       job.peer!.idle = true
       job.result = this.process(job, result)
       jobStr = this.jobStr(job, true)
-      if (job.result) {
+      if (job.result !== undefined) {
+        this.debug(`Successful job completion job ${jobStr}, writing to out and dequeue`)
         this.out.insert(job)
         this.dequeue()
       } else {
@@ -317,11 +337,11 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
   next() {
     this.nextTasks()
     const job = this.in.peek()
-    if (!job) {
+    if (job === undefined) {
       if (this.finished !== this.total) {
         // There are still jobs waiting to be processed out in the writer pipe
         this.debug(
-          `No job found on next task, skip next job execution processed=${this.processed} finished=${this.finished} total=${this.total}`
+          `No job found as next task, skip next job execution processed=${this.processed} finished=${this.finished} total=${this.total}`
         )
       } else {
         // There are no more jobs in the fetcher, so its better to resolve
@@ -331,19 +351,21 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
       }
       return false
     }
-    if (this._readableState!.length > this.maxQueue) {
+    const jobStr = this.jobStr(job)
+    if (this._readableState === undefined || this._readableState!.length > this.maxQueue) {
       this.debug(
         `Readable state length=${this._readableState!.length} exceeds max queue size=${
           this.maxQueue
-        }, skip next job execution.`
+        }, skip job ${jobStr} execution.`
       )
       return false
     }
-    if (job.index > this.processed + this.maxQueue) {
-      this.debug(`Job index greater than processed + max queue size, skip next job execution.`)
+    if (job.index > this.finished + this.maxQueue) {
+      this.debug(`Job index greater than finished + max queue size, skip job ${jobStr} execution.`)
+      return false
     }
     if (this.processed === this.total) {
-      this.debug(`Total number of tasks reached, skip next job execution.`)
+      this.debug(`Total number of tasks reached, skip job ${jobStr} execution.`)
       return false
     }
     const peer = this.peer()
@@ -355,16 +377,19 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
       const timeout = setTimeout(() => {
         this.expire(job)
       }, this.timeout)
+      this.debug(`All requirements met for job ${jobStr}, start requesting.`)
       this.request(job, peer)
-        .then((result?: JobResult) => this.success(job, result))
+        .then((result?: JobResult) => {
+          this.success(job, result)
+        })
         .catch((error: Error) => {
           const { banPeer } = this.processStoreError(error, job.task)
           this.failure(job, error, false, false, banPeer)
         })
         .finally(() => clearTimeout(timeout))
       return job
     } else {
-      this.debug(`No idle peer available, skip next job execution.`)
+      this.debug(`No idle peer available, skip execution for job ${jobStr}.`)
       return false
     }
   }
@@ -415,6 +440,7 @@ export abstract class Fetcher<JobTask, JobResult, StorageItem> extends Readable
       cb: Function
     ) => {
       const jobItems = job instanceof Array ? job : [job]
+      this.debug(`Starting write for ${jobItems.length} jobs...`)
       try {
         for (const jobItem of jobItems) {
           await this.store(jobItem.result as StorageItem[])