Repository: influxdata/influxdb-client-js
Pull request #547 (2022-08-23 05:06:27)

feat(core): improve write retry buffer
This PR improves write retry buffer so that
- the calling code be informed about items that were removed in order to keep the size of the retry buffer under the configured limit (32_000 lines, maxBufferLines options) using a new optional `writeRetrySkipped` callback
- when the buffer becomes full, the oldest items (according to the time of the first failures) are always removed, previously oldest items according to the time they were added/re-added to the retry buffer were removed; the new strategy in this PR is fair
- the code now remembers scheduled retry time for each failed write, it always follows the order of retry execution regarding the scheduled retry time

## Checklist

<!-- Checkboxes below this note can be erased if not applicable to your Pull Request. -->

- [x] CHANGELOG.md updated
- [x] Rebased/mergeable
- [x] A test has been added if appropriate
- [x] `yarn test` completes successfully
- [x] Commit messages are [conventional](https://www.conventionalcommits.org/en/v1.0.0/)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7dc585ba..26e25940 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
 ## 1.29.0 [unreleased]
 
+### Features
+
+1. [#547](https://github.com/influxdata/influxdb-client-js/pull/547): Improve write retry buffer.
+
 ### Other
 
 1. [#511](https://github.com/influxdata/influxdb-client-js/pull/511): Replace rollup with tsup.
diff --git a/packages/core/src/impl/RetryBuffer.ts b/packages/core/src/impl/RetryBuffer.ts
index 3bb4ae0e..678e2494 100644
--- a/packages/core/src/impl/RetryBuffer.ts
+++ b/packages/core/src/impl/RetryBuffer.ts
@@ -1,23 +1,35 @@
 import {Log} from '../util/logger'
 
-/* interval between successful retries */
-const RETRY_INTERVAL = 1
-
 interface RetryItem {
   lines: string[]
   retryCount: number
+  retryTime: number
   expires: number
   next?: RetryItem
 }
 
+type FindShrinkCandidateResult = [found: RetryItem, parent?: RetryItem]
+
+function findShrinkCandidate(first: RetryItem): FindShrinkCandidateResult {
+  let parent = undefined
+  let found = first
+  let currentParent = first
+  while (currentParent.next) {
+    if (currentParent.next.expires < found.expires) {
+      parent = currentParent
+      found = currentParent.next
+    }
+    currentParent = currentParent.next
+  }
+  return [found, parent]
+}
+
 /**
  * Retries lines up to a limit of max buffer size.
  */
 export default class RetryBuffer {
   first?: RetryItem
-  last?: RetryItem
   size = 0
-  nextRetryTime = 0
   closed = false
   private _timeoutHandle: any = undefined
 
@@ -27,7 +39,12 @@ export default class RetryBuffer {
       lines: string[],
       retryCountdown: number,
       started: number
-    ) => Promise<void>
+    ) => Promise<void>,
+    private onShrink: (entry: {
+      lines: string[]
+      retryCount: number
+      expires: number
+    }) => void = () => undefined
   ) {}
 
   addLines(
@@ -40,43 +57,57 @@ export default class RetryBuffer {
     if (!lines.length) return
     let retryTime = Date.now() + delay
     if (expires < retryTime) {
-      delay = expires - Date.now()
       retryTime = expires
     }
-    if (retryTime > this.nextRetryTime) this.nextRetryTime = retryTime
     // ensure at most maxLines are in the Buffer
     if (this.first && this.size + lines.length > this.maxLines) {
       const origSize = this.size
       const newSize = origSize * 0.7 // reduce to 70 %
       do {
-        const newFirst = this.first.next as RetryItem
-        this.size -= this.first.lines.length
-        this.first.next = undefined
-        this.first = newFirst
-        if (!this.first) {
-          this.last = undefined
+        // remove "oldest" item
+        const [found, parent] = findShrinkCandidate(this.first)
+        this.size -= found.lines.length
+        if (parent) {
+          parent.next = found.next
+        } else {
+          this.first = found.next
+          if (this.first) {
+            this.scheduleRetry(this.first.retryTime - Date.now())
+          }
         }
+        found.next = undefined
+        this.onShrink(found)
       } while (this.first && this.size + lines.length > newSize)
       Log.error(
         `RetryBuffer: ${
           origSize - this.size
         } oldest lines removed to keep buffer size under the limit of ${
           this.maxLines
-        } lines`
+        } lines.`
       )
     }
     const toAdd: RetryItem = {
       lines,
       retryCount,
+      retryTime,
       expires,
     }
-    if (this.last) {
-      this.last.next = toAdd
-      this.last = toAdd
-    } else {
-      this.first = toAdd
-      this.last = toAdd
-      this.scheduleRetry(delay)
+    // insert sorted according to retryTime
+    let current: RetryItem | undefined = this.first
+    let parent = undefined
+    for (;;) {
+      if (!current || current.retryTime > retryTime) {
+        toAdd.next = current
+        if (parent) {
+          parent.next = toAdd
+        } else {
+          this.first = toAdd
+          this.scheduleRetry(retryTime - Date.now())
+        }
+        break
+      }
+      parent = current
+      current = current.next
     }
     this.size += lines.length
   }
@@ -87,25 +118,28 @@ export default class RetryBuffer {
       this.first = this.first.next
       toRetry.next = undefined
       this.size -= toRetry.lines.length
-      if (!this.first) this.last = undefined
       return toRetry
     }
     return undefined
   }
 
   scheduleRetry(delay: number): void {
+    if (this._timeoutHandle) {
+      clearTimeout(this._timeoutHandle)
+    }
     this._timeoutHandle = setTimeout(() => {
       const toRetry = this.removeLines()
       if (toRetry) {
-        this.retryLines(toRetry.lines, toRetry.retryCount, toRetry.expires)
-          .then(() => {
-            // continue with successfull retry
-            this.scheduleRetry(RETRY_INTERVAL)
-          })
-          .catch((_e) => {
-            // already logged
-            this.scheduleRetry(this.nextRetryTime - Date.now())
-          })
+        this.retryLines(
+          toRetry.lines,
+          toRetry.retryCount,
+          toRetry.expires
+        ).finally(() => {
+          // schedule next retry execution
+          if (this.first) {
+            this.scheduleRetry(this.first.retryTime - Date.now())
+          }
+        })
       } else {
         this._timeoutHandle = undefined
       }
diff --git a/packages/core/src/impl/WriteApiImpl.ts b/packages/core/src/impl/WriteApiImpl.ts
index afb1f233..f2e2aebe 100644
--- a/packages/core/src/impl/WriteApiImpl.ts
+++ b/packages/core/src/impl/WriteApiImpl.ts
@@ -146,7 +146,8 @@ export default class WriteApiImpl implements WriteApi {
     this.retryStrategy = createRetryDelayStrategy(this.writeOptions)
     this.retryBuffer = new RetryBuffer(
       this.writeOptions.maxBufferLines,
-      this.sendBatch
+      this.sendBatch,
+      this.writeOptions.writeRetrySkipped
     )
   }
 
diff --git a/packages/core/src/options.ts b/packages/core/src/options.ts
index cff2198e..cfac6b42 100644
--- a/packages/core/src/options.ts
+++ b/packages/core/src/options.ts
@@ -93,6 +93,13 @@ export interface WriteRetryOptions extends RetryDelayStrategyOptions {
    */
   writeSuccess(this: WriteApi, lines: Array<string>): void
 
+  /**
+   * WriteRetrySkipped is informed about lines that were removed from the retry buffer
+   * to keep the size of the retry buffer under the configured limit (maxBufferLines).
+   * @param entry - lines that were skipped
+   */
+  writeRetrySkipped(entry: {lines: Array<string>; expires: number}): void
+
   /** max count of retries after the first write fails */
   maxRetries: number
   /** max time (millis) that can be spent with retries */
@@ -137,6 +144,7 @@ export const DEFAULT_WriteOptions: WriteOptions = {
   flushInterval: 60000,
   writeFailed: function () {},
   writeSuccess: function () {},
+  writeRetrySkipped: function () {},
   maxRetries: 5,
   maxRetryTime: 180_000,
   maxBufferLines: 32_000,


Test patch:
diff --git a/packages/core/test/unit/WriteApi.test.ts b/packages/core/test/unit/WriteApi.test.ts
index 1f80934b..a32b22bd 100644
--- a/packages/core/test/unit/WriteApi.test.ts
+++ b/packages/core/test/unit/WriteApi.test.ts
@@ -242,6 +242,45 @@ describe('WriteApi', () => {
         ])
       })
     })
+    const withWriteRetryOptions = [true, false]
+    for (const withWriteRetry of withWriteRetryOptions) {
+      it(`informs about skipped retry ${
+        withWriteRetry ? 'with' : 'without'
+      } callback`, async () => {
+        const skippedLines: string[][] = []
+        let failedAttempts = 0
+        const writeOptions: Partial<WriteOptions> = {
+          flushInterval: 0,
+          batchSize: 1,
+          maxBufferLines: 1,
+          writeFailed: () => {
+            failedAttempts++
+          },
+        }
+        if (withWriteRetry) {
+          writeOptions.writeRetrySkipped = ({lines}) => {
+            skippedLines.push(lines)
+          }
+        }
+        useSubject(writeOptions)
+        subject.writeRecord('test value=1')
+        subject.writeRecord('test value=2')
+        // wait for http calls to finish
+        await waitForCondition(() => failedAttempts == 2)
+        await subject.close().then(() => {
+          expect(logs.error).to.length(2)
+          expect(logs.error[0][0]).equals(
+            'RetryBuffer: 1 oldest lines removed to keep buffer size under the limit of 1 lines.'
+          )
+          expect(logs.error[1][0]).equals(
+            'Retry buffer closed with 1 items that were not written to InfluxDB!'
+          )
+          if (withWriteRetry) {
+            expect(skippedLines).deep.equals([['test value=1']])
+          }
+        })
+      })
+    }
     it('uses the pre-configured batchSize', async () => {
       useSubject({flushInterval: 0, maxRetries: 0, batchSize: 2})
       subject.writeRecords(['test value=1', 'test value=2', 'test value=3'])
diff --git a/packages/core/test/unit/impl/RetryBuffer.test.ts b/packages/core/test/unit/impl/RetryBuffer.test.ts
index e3748773..5fc314bd 100644
--- a/packages/core/test/unit/impl/RetryBuffer.test.ts
+++ b/packages/core/test/unit/impl/RetryBuffer.test.ts
@@ -25,22 +25,58 @@ describe('RetryBuffer', () => {
     }
     await waitForCondition(() => output.length >= 10)
     expect(output).length.is.greaterThan(0)
-    subject.addLines([], 1, 1, Date.now() + 1000) // shall be ignored
+    subject.addLines([], 1, 1, Date.now() + 1000) // empty lines are ignored
     subject.close()
-    subject.addLines(['x'], 1, 1, Date.now() + 1000) // shall be ignored
+    subject.addLines(['x'], 1, 1, Date.now() + 1000) // ignored after calling flush
     await subject.flush()
     expect(input).deep.equals(output)
   })
   it('ignores lines on heavy load', async () => {
-    const input = [] as Array<[string[], number]>
     const output = [] as Array<[string[], number]>
-    const subject = new RetryBuffer(5, (lines, countdown) => {
-      output.push([lines, countdown])
-      return Promise.resolve()
-    })
+    const subject = new RetryBuffer(
+      5,
+      (lines, countdown) => {
+        output.push([lines, countdown])
+        return Promise.resolve()
+      },
+      ({expires}) => {
+        for (let actual = subject.first; actual; actual = actual?.next) {
+          if (actual.expires < expires) {
+            expect.fail(
+              `${actual.expires} entry was not removed from the retry buffer, but ${expires} was!`
+            )
+          }
+        }
+      }
+    )
+    for (let i = 0; i < 10; i++) {
+      subject.addLines(['a' + i], i, 100 + i, Date.now() + 1000 + i)
+    }
+    await subject.flush()
+    expect(subject.close()).equals(0)
+    expect(logs.error).length.is.greaterThan(0) // 5 entries over limit
+    expect(output).length.is.lessThan(6) // at most 5 items will be written
+  })
+  it('ignores lines on heavy load with descending expiration', async () => {
+    const output = [] as Array<[string[], number]>
+    const subject = new RetryBuffer(
+      5,
+      (lines, countdown) => {
+        output.push([lines, countdown])
+        return Promise.resolve()
+      },
+      ({expires}) => {
+        for (let actual = subject.first; actual; actual = actual?.next) {
+          if (actual.expires < expires) {
+            expect.fail(
+              `${actual.expires} entry was not removed from the retry buffer, but ${expires} was!`
+            )
+          }
+        }
+      }
+    )
     for (let i = 0; i < 10; i++) {
-      if (i >= 5) input.push([['a' + i], i])
-      subject.addLines(['a' + i], i, 100, Date.now() + 1000)
+      subject.addLines(['a' + i], i, 100 + i, Date.now() + 1000 - i)
     }
     await subject.flush()
     expect(subject.close()).equals(0)