From ebced09bca1d1d2abced5b9069444e519e38e343 Mon Sep 17 00:00:00 2001 From: Veronika Al Raheem Date: Sat, 8 Aug 2026 12:52:33 +0200 Subject: [PATCH 1/2] feat: add getLastSeqNewerThan for safe checkpoint management --- cloudant/features/changesFollower.ts | 39 ++++++++++++++++++++++ cloudant/features/changesResultIterator.ts | 22 ++++++++++++ 2 files changed, 61 insertions(+) diff --git a/cloudant/features/changesFollower.ts b/cloudant/features/changesFollower.ts index b45d3350d..c9310ea3f 100644 --- a/cloudant/features/changesFollower.ts +++ b/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 diff --git a/cloudant/features/changesResultIterator.ts b/cloudant/features/changesResultIterator.ts index 160a7d35e..3848f143c 100644 --- a/cloudant/features/changesResultIterator.ts +++ b/cloudant/features/changesResultIterator.ts @@ -25,6 +25,11 @@ enum TransientErrorSuppression { TIMER, } +type SeqEntry = { + type: 'row' | 'page'; + lastSeq: string; +}; + export class ChangesResultIterableIterator implements AsyncIterableIterator { private readonly timeoutPromise = promisify(setTimeout); private readonly cancelToken = 'CloudantChangesIteratorCancel'; @@ -42,6 +47,7 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator(); private cancel: (error?: Error) => void; private countDown: number; private inflight: Promise = null; @@ -124,6 +130,10 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator { + return this.seqMap; + } + async return(value?: any): Promise> { this.logger.debug('Iterator return entry.'); if (!this.stopped) { @@ -195,6 +205,18 @@ export class ChangesResultIterableIterator implements AsyncIterableIterator 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) { From 38cac4f0475c97668520412bea7b4f46b1709966 Mon Sep 17 00:00:00 2001 From: Veronika Al Raheem Date: Sat, 8 Aug 2026 12:53:41 +0200 Subject: [PATCH 2/2] test: add unit tests for getLastSeqNewerThan --- test/unit/features/changesFollower.test.js | 151 +++++++++++++++++++++ 1 file changed, 151 insertions(+) diff --git a/test/unit/features/changesFollower.test.js b/test/unit/features/changesFollower.test.js index 859f86faa..a7c02badb 100644 --- a/test/unit/features/changesFollower.test.js +++ b/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(); + } + }); + }); + }); });