Add implementation of returnable last_seq

cloudant/features/changesFollower.ts

@@ -187,6 +187,31 @@ export class ChangesFollower {
     }
   }
 
+  /**
+   * Return the most recent sequence ID that is safe to use as a checkpoint
+   * after the given sequence ID.
+   *
+   * Call this after fully processing a {@link ChangesResultItem} to obtain
+   * a safe value to persist as {@link CloudantV1.PostChangesParams.since}
+   * for the next run.
+   *
+   * @param lastPersistedSeqId - the `seq` of the last {@link ChangesResultItem}
+   * you have fully processed
+   * @throws {Error} if the provided sequence ID is null or empty
+   * @return {string} the most recent safe sequence ID to persist. Returns the
+   * supplied ID unchanged if no newer checkpoint is available, the feed has
+   * not started yet, or the supplied ID was not seen by this {@link ChangesFollower} instance.
+   */
+  getLastSeqNewerThan(lastPersistedSeqId: string): string {
+    if (!lastPersistedSeqId) {
+      throw new Error('The provided sequence ID cannot be null or empty');
+    }
+    if (!this.changesResultIterator) {
+      return lastPersistedSeqId;
+    }
+    return this.changesResultIterator.lastSeqSince(lastPersistedSeqId);
+  }
+
   /**
    *
    * @param mode the mode in which to run the ChangesFollower

cloudant/features/changesResultIterator.ts

@@ -25,6 +25,11 @@ enum TransientErrorSuppression {
   TIMER,
 }
 
+type SeqEntry = {
+  type: 'row' | 'page';
+  seq: string | null;
+};
+
 export class ChangesResultIterableIterator implements AsyncIterableIterator<CloudantV1.ChangesResult> {
   private readonly timeoutPromise = promisify(setTimeout);
   private readonly cancelToken = 'CloudantChangesIteratorCancel';
@@ -42,6 +47,10 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
   private readonly expRetryGate: number = Math.floor(
     Math.log2(ChangesParamsHelper.LONGPOLL_TIMEOUT / this.baseDelay)
   );
+  private readonly seqMarkers: SeqEntry[] = [];
+  private static readonly SEQ_MARKERS_CAPACITY = 200;
+  private static readonly SEQ_MARKERS_EVICTION_COUNT =
+    ChangesResultIterableIterator.SEQ_MARKERS_CAPACITY / 10;
   private cancel: (error?: Error) => void;
   private countDown: number;
   private inflight: Promise<any> = null;
@@ -124,6 +133,25 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
     return this;
   }
 
+  lastSeqSince(lastPersistedSeqId: string): string {
+    let found = false;
+    let result: string | null = null;
+
+    this.seqMarkers.every((entry) => {
+      if (found) {
+        if (entry.type === 'row') return false;
+        result = entry.seq;
+      }
+      if (!found && entry.seq === lastPersistedSeqId) {
+        found = true;
+        result = entry.seq;
+      }
+      return true;
+    });
+
+    return found ? result : lastPersistedSeqId;
+  }
+
   async return(value?: any): Promise<IteratorResult<CloudantV1.ChangesResult>> {
     this.logger.debug('Iterator return entry.');
     if (!this.stopped) {
@@ -195,6 +223,28 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
         }
 
         this.since = response.result.lastSeq;
+
+        const { results }: CloudantV1.ChangesResult = response.result;
+        if (
+          this.seqMarkers.length >=
+          ChangesResultIterableIterator.SEQ_MARKERS_CAPACITY
+        ) {
+          this.seqMarkers.splice(
+            0,
+            ChangesResultIterableIterator.SEQ_MARKERS_EVICTION_COUNT
+          );
+        }
+        if (results.length > 0) {
+          this.seqMarkers.push({
+            type: 'row',
+            seq: results.at(-1).seq,
+          });
+        }
+        this.seqMarkers.push({
+          type: 'page',
+          seq: response.result.lastSeq,
+        });
+
         this.pending = response.result.pending;
 
         if (this.mode === Mode.FINITE && this.pending === 0) {

test/unit/features/changesFollower.test.js

@@ -726,4 +726,111 @@ describe('Test ChangesFollower', () => {
       }
     });
   });
+  describe('getLastSeqNewerThan', () => {
+    /**
+     * Throws when passed null.
+     */
+    it('testGetLastSeqNewerThanWithNull', () => {
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      expect(() => changesFollower.getLastSeqNewerThan(null)).toThrow(
+        'The provided sequence ID cannot be null or empty'
+      );
+    });
+
+    /**
+     * Throws when passed an empty string.
+     */
+    it('testGetLastSeqNewerThanWithEmptyString', () => {
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      expect(() => changesFollower.getLastSeqNewerThan('')).toThrow(
+        'The provided sequence ID cannot be null or empty'
+      );
+    });
+
+    /**
+     * Returns the input seq when the feed has not started yet.
+     */
+    it('testGetLastSeqNewerThanBeforeFeedStarts', () => {
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      expect(changesFollower.getLastSeqNewerThan('seq-a')).toBe('seq-a');
+    });
+
+    /**
+     * Returns the input seq unchanged when the seq was never seen by this follower.
+     */
+    it('testGetLastSeqNewerThanUnknownSeq', (done) => {
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [{ id: 'a', seq: 'seq-a', changes: [] }],
+          pending: 0,
+          lastSeq: 'seq-a',
+        },
+      });
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          expect(changesFollower.getLastSeqNewerThan('seq-unknown')).toBe(
+            'seq-unknown'
+          );
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * Returns the input seq unchanged when querying with a seq from the middle
+     * of a batch — only the last item's seq is stored in seqMarkers.
+     */
+    it('testGetLastSeqNewerThanMiddleOfBatch', (done) => {
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [
+            { id: 'a', seq: 'seq-a', changes: [] },
+            { id: 'b', seq: 'seq-b', changes: [] },
+            { id: 'c', seq: 'seq-c', changes: [] },
+          ],
+          pending: 0,
+          lastSeq: 'seq-c',
+        },
+      });
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          // seq-a and seq-b are middle items — not stored in seqMarkers
+          expect(changesFollower.getLastSeqNewerThan('seq-a')).toBe('seq-a');
+          expect(changesFollower.getLastSeqNewerThan('seq-b')).toBe('seq-b');
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * End-to-end: returns the correct last_seq through a full stream.
+     */
+    it('testGetLastSeqNewerThanEndToEnd', (done) => {
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [{ id: 'a', seq: 'seq-a', changes: [] }],
+          pending: 0,
+          lastSeq: 'seq-b',
+        },
+      });
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          expect(changesFollower.getLastSeqNewerThan('seq-a')).toBe('seq-b');
+        } finally {
+          done();
+        }
+      });
+    });
+  });
 });