Add implementation of returnable last_seq

cloudant/features/changesFollower.ts

@@ -187,6 +187,45 @@ 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
+   * @return {string | null} the most recent safe sequence ID to persist, or
+   * `null` if no newer checkpoint is available or the supplied ID was not
+   * seen by this {@link ChangesFollower} instance
+   */
+  getLastSeqNewerThan(lastPersistedSeqId: string): string | null {
+    if (!this.changesResultIterator) {
+      return null;
+    }
+    const seqMap = this.changesResultIterator.getSeqMap();
+    let found = false;
+    let result: string | null = null;
+
+    Array.from(seqMap.entries()).every(([key, entries]) =>
+      entries.every((entry) => {
+        if (found) {
+          if (entry.type === 'row') return false;
+          result = entry.lastSeq;
+        }
+        if (!found && key === lastPersistedSeqId) {
+          found = true;
+          result = entry.lastSeq;
+        }
+        return true;
+      })
+    );
+
+    return found ? result : null;
+  }
+
   /**
    *
    * @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';
+  lastSeq: string;
+};
+
 export class ChangesResultIterableIterator implements AsyncIterableIterator<CloudantV1.ChangesResult> {
   private readonly timeoutPromise = promisify(setTimeout);
   private readonly cancelToken = 'CloudantChangesIteratorCancel';
@@ -42,6 +47,7 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
   private readonly expRetryGate: number = Math.floor(
     Math.log2(ChangesParamsHelper.LONGPOLL_TIMEOUT / this.baseDelay)
   );
+  private readonly seqMap = new Map<string, SeqEntry[]>();
   private cancel: (error?: Error) => void;
   private countDown: number;
   private inflight: Promise<any> = null;
@@ -124,6 +130,10 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
     return this;
   }
 
+  getSeqMap(): Map<string, SeqEntry[]> {
+    return this.seqMap;
+  }
+
   async return(value?: any): Promise<IteratorResult<CloudantV1.ChangesResult>> {
     this.logger.debug('Iterator return entry.');
     if (!this.stopped) {
@@ -195,6 +205,18 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator<Clou
         }
 
         this.since = response.result.lastSeq;
+
+        const { results }: CloudantV1.ChangesResult = response.result;
+        if (results.length > 0 && results.at(-1).seq != null) {
+          const lastItem = results.at(-1);
+          const rowEntries = this.seqMap.get(lastItem.seq) ?? [];
+          rowEntries.push({ type: 'row', lastSeq: response.result.lastSeq });
+          this.seqMap.set(lastItem.seq, rowEntries);
+        }
+        const pageEntries = this.seqMap.get(response.result.lastSeq) ?? [];
+        pageEntries.push({ type: 'page', lastSeq: response.result.lastSeq });
+        this.seqMap.set(response.result.lastSeq, pageEntries);
+
         this.pending = response.result.pending;
 
         if (this.mode === Mode.FINITE && this.pending === 0) {

test/unit/features/changesFollower.test.js

@@ -726,4 +726,155 @@ describe('Test ChangesFollower', () => {
       }
     });
   });
+  describe('getLastSeqNewerThan', () => {
+    /**
+     * Returns null when the feed has not started yet.
+     */
+    it('testGetLastSeqNewerThanBeforeFeedStarts', () => {
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      expect(changesFollower.getLastSeqNewerThan('seq-a')).toBeNull();
+    });
+
+    /**
+     * Returns null for a sequence ID not 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')).toBeNull();
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * Returns the last_seq when the last item seq equals the last_seq.
+     */
+    it('testGetLastSeqNewerThanNormalBatch', (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 {
+          expect(changesFollower.getLastSeqNewerThan('seq-c')).toBe('seq-c');
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * Returns the last_seq when the last item seq differs from the last_seq.
+     */
+    it('testGetLastSeqNewerThanSparseBatch', (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-f',
+        },
+      });
+
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          expect(changesFollower.getLastSeqNewerThan('seq-c')).toBe('seq-f');
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * Advances past empty pages from a heavily filtered feed.
+     */
+    it('testGetLastSeqNewerThanEmptyPages', (done) => {
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [{ id: 'a', seq: 'seq-a', changes: [] }],
+          pending: 1,
+          lastSeq: 'seq-a',
+        },
+      });
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [],
+          pending: 0,
+          lastSeq: 'seq-d',
+        },
+      });
+
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          expect(changesFollower.getLastSeqNewerThan('seq-a')).toBe('seq-d');
+        } finally {
+          done();
+        }
+      });
+    });
+
+    /**
+     * Stops advancing at the next row entry.
+     */
+    it('testGetLastSeqNewerThanStopsAtNextRow', (done) => {
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [{ id: 'a', seq: 'seq-a', changes: [] }],
+          pending: 1,
+          lastSeq: 'seq-a',
+        },
+      });
+      postChangesPromiseMock.mockResolvedValueOnce({
+        result: {
+          results: [{ id: 'e', seq: 'seq-e', changes: [] }],
+          pending: 0,
+          lastSeq: 'seq-f',
+        },
+      });
+
+      const changesFollower = new ChangesFollower(service, minimumTestParams);
+      const stream = changesFollower.startOneOff();
+      stream.on('data', () => {});
+      stream.on('end', () => {
+        try {
+          expect(changesFollower.getLastSeqNewerThan('seq-a')).toBe('seq-a');
+        } finally {
+          done();
+        }
+      });
+    });
+  });
 });