Skip to content

Commit 017f1bb

Browse files
committed
feat: latestSequenceFrom for sparse changes feeds
1 parent 7251940 commit 017f1bb

5 files changed

Lines changed: 559 additions & 1 deletion

File tree

‎modules/cloudant/src/main/java/com/ibm/cloud/cloudant/features/ChangesFollower.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -274,4 +274,34 @@ private synchronized Stream<ChangesResultItem> run(Mode mode) throws IllegalStat
274274
return changesStream;
275275
}
276276

277+
/**
278+
* Returns the most recent sequence ID that is safe to use as a checkpoint,
279+
* advancing beyond the supplied checkpoint sequence ID where possible.
280+
* <p>
281+
* With highly filtered changes feeds, multiple pages can pass through the
282+
* follower without returning any changes. Using only the
283+
* {@code seq} of the last processed {@link ChangesResultItem} in those cases
284+
* causes a long changes feed rewind on the next run. To avoid this, call this
285+
* method after fully processing each {@link ChangesResultItem} with a
286+
* non-null {@code seq} and persist the returned value to use as the
287+
* {@code since} parameter for the next run.
288+
*
289+
* @param checkpointSequenceId the last checkpoint sequence ID — either the
290+
* non-null {@code seq} of the last {@link ChangesResultItem} fully
291+
* processed, or a value previously returned by this method.
292+
* @return the most recent safe sequence ID to use as a checkpoint, or the
293+
* supplied value if no newer sequence is available
294+
* @throws IllegalArgumentException if {@code checkpointSequenceId} is null or empty
295+
*/
296+
public String latestSequenceFrom(String checkpointSequenceId) {
297+
if (checkpointSequenceId == null || checkpointSequenceId.isEmpty()) {
298+
throw new IllegalArgumentException("Provided sequence ID must be a non-empty string.");
299+
}
300+
// If we have a spliterator, ask it for the last sequence ID (if any
301+
ChangesResultSpliterator spliterator = this.changesResultSpliterator.get();
302+
if (spliterator != null) {
303+
return spliterator.lastSeqSince(checkpointSequenceId);
304+
}
305+
return checkpointSequenceId;
306+
}
277307
}

‎modules/cloudant/src/main/java/com/ibm/cloud/cloudant/features/ChangesResultSpliterator.java‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,9 @@
1616
import java.time.Duration;
1717
import java.time.Instant;
1818
import java.time.temporal.ChronoUnit;
19+
import java.util.ArrayList;
1920
import java.util.Collections;
21+
import java.util.List;
2022
import java.util.Random;
2123
import java.util.Spliterator;
2224
import java.util.Spliterators.AbstractSpliterator;
@@ -28,6 +30,7 @@
2830
import com.ibm.cloud.cloudant.features.ChangesFollower.Mode;
2931
import com.ibm.cloud.cloudant.v1.Cloudant;
3032
import com.ibm.cloud.cloudant.v1.model.ChangesResult;
33+
import com.ibm.cloud.cloudant.v1.model.ChangesResultItem;
3134
import com.ibm.cloud.cloudant.v1.model.PostChangesOptions;
3235
import com.ibm.cloud.sdk.core.http.Response;
3336
import com.ibm.cloud.sdk.core.http.ServiceCall;
@@ -68,6 +71,7 @@ public java.util.List<com.ibm.cloud.cloudant.v1.model.ChangesResultItem> getResu
6871
private final Duration errorTolerance;
6972
private final TransientErrorSuppression transientSuppression;
7073
private final Object requestLock = new Object();
74+
private final SeqMarkers seqMarkers = new SeqMarkers();
7175
private volatile String since;
7276
// Default to "infinite"
7377
private volatile Long pending = Long.MAX_VALUE;
@@ -152,8 +156,11 @@ ChangesResult next() {
152156
if (this.transientSuppression == TransientErrorSuppression.TIMER) {
153157
this.successTimestamp = Instant.now();
154158
}
159+
155160
this.since = result.getLastSeq();
156161
this.pending = result.getPending();
162+
this.seqMarkers.put(result);
163+
157164
if (this.mode == Mode.FINITE && this.pending == 0L) {
158165
this.hasNext = false;
159166
}
@@ -246,4 +253,88 @@ void stop() {
246253
}
247254
}
248255
}
256+
257+
String lastSeqSince(String lastPersistedSeq) {
258+
return this.seqMarkers.get(lastPersistedSeq);
259+
}
260+
261+
private static final class SeqMarkers extends ArrayList<SeqMarkerEntry> {
262+
private static final int CAPACITY = 200;
263+
private static final int EVICTION_COUNT = CAPACITY / 10;
264+
265+
@Override
266+
public synchronized boolean add(SeqMarkerEntry entry) {
267+
if (size() >= CAPACITY) {
268+
// Bulk-evict oldest 10% to amortise the cost of array shifting
269+
subList(0, EVICTION_COUNT).clear();
270+
}
271+
return super.add(entry);
272+
}
273+
274+
synchronized void put(ChangesResult pageResult) {
275+
// Get the last_seq from the page
276+
String lastSeq = pageResult.getLastSeq();
277+
// Find the last row seq from these results
278+
List<ChangesResultItem> changes = pageResult.getResults();
279+
if (changes.size() > 0) {
280+
add(new SeqMarkerEntry(changes.get(changes.size() - 1).getSeq(), SeqMarkerType.ROW));
281+
}
282+
add(new SeqMarkerEntry(lastSeq, SeqMarkerType.LAST));
283+
}
284+
285+
synchronized String get(String lastPersistedSeq) {
286+
// List of last_seq values that the caller could checkpoint
287+
// The most recent sequence ID that is safe to checkpoint beyond the user-provided one
288+
// (i.e. the last_seq of the last page after the user page or row for which there are no other rows inbetween
289+
String returnSeq = null;
290+
291+
// Note entries are in insertion order
292+
// Find the index of the caller's sequence entry
293+
int startIdx = indexOf(lastPersistedSeq);
294+
if (startIdx >= 0) {
295+
// Always collect the matched entry
296+
returnSeq = get(startIdx).seq;
297+
// Then continue while subsequent entries are LAST type,
298+
// stopping when a ROW entry is encountered
299+
for (SeqMarkerEntry entry : subList(startIdx + 1, size())) {
300+
if (entry.seqMarkerType == SeqMarkerType.LAST) {
301+
String s;
302+
if ((s = entry.seq) != null) {
303+
returnSeq = s;
304+
}
305+
} else {
306+
break;
307+
}
308+
}
309+
}
310+
// Return a suitable lastSeq if there is one, otherwise return the user's seq
311+
return (returnSeq == null) ? lastPersistedSeq : returnSeq;
312+
}
313+
314+
// Find the index of the entry whose seq matches the given seqId
315+
private int indexOf(String seqId) {
316+
for (int i = 0; i < size(); i++) {
317+
if (seqId.equals(get(i).seq)) {
318+
return i;
319+
}
320+
}
321+
return -1;
322+
}
323+
}
324+
325+
private enum SeqMarkerType {
326+
LAST,
327+
ROW
328+
}
329+
330+
private static final class SeqMarkerEntry {
331+
332+
private final String seq;
333+
private final SeqMarkerType seqMarkerType;
334+
335+
SeqMarkerEntry(String seq, SeqMarkerType seqMarkerType) {
336+
this.seq = seq;
337+
this.seqMarkerType = seqMarkerType;
338+
}
339+
}
249340
}

‎modules/cloudant/src/test/java/com/ibm/cloud/cloudant/features/ChangesFollowerTest.java‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -580,4 +580,121 @@ void testBatchSizeLimit() {
580580
});
581581
Assert.assertEquals(lee.getRequestLimit(), 1000L, "The changes request should have the expected limit.");
582582
}
583+
584+
/**
585+
* Tests for getLastSeqNewerThan method
586+
*/
587+
588+
/**
589+
* Verify that getLastSeqNewerThan throws IllegalArgumentException when passed null.
590+
*/
591+
@Test
592+
void testGetLastSeqNewerThanWithNull() {
593+
Cloudant mockClient = new ChangesRequestMockClient(new PerpetualSupplier(true));
594+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
595+
IllegalArgumentException iae = Assert.expectThrows(IllegalArgumentException.class, () -> {
596+
testFollower.latestSequenceFrom(null);
597+
});
598+
Assert.assertEquals(iae.getMessage(),
599+
"Provided sequence ID must be a non-empty string.",
600+
"The error message should be correct.");
601+
}
602+
603+
/**
604+
* Verify that getLastSeqNewerThan throws IllegalArgumentException when passed an empty string.
605+
*/
606+
@Test
607+
void testGetLastSeqNewerThanWithEmptyString() {
608+
Cloudant mockClient = new ChangesRequestMockClient(new PerpetualSupplier(true));
609+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
610+
IllegalArgumentException iae = Assert.expectThrows(IllegalArgumentException.class, () -> {
611+
testFollower.latestSequenceFrom("");
612+
});
613+
Assert.assertEquals(iae.getMessage(),
614+
"Provided sequence ID must be a non-empty string.",
615+
"The error message should be correct.");
616+
}
617+
618+
/**
619+
* Verify that getLastSeqNewerThan returns the input sequence unchanged when called before the feed starts.
620+
*/
621+
@Test
622+
void testGetLastSeqNewerThanBeforeStart() {
623+
Cloudant mockClient = new ChangesRequestMockClient(new PerpetualSupplier(true));
624+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
625+
String inputSeq = "100-abc";
626+
String result = testFollower.latestSequenceFrom(inputSeq);
627+
Assert.assertEquals(result, inputSeq,
628+
"Should return the input sequence unchanged when feed hasn't started.");
629+
}
630+
631+
/**
632+
* Verify that getLastSeqNewerThan returns the input sequence when no newer sequence is available.
633+
*/
634+
@Test
635+
void testGetLastSeqNewerThanAfterStartNoNewerSeq() {
636+
// Create a mock client with a single batch
637+
Cloudant mockClient = new ChangesRequestMockClient(ChangesRequestMockClient.makeBatchSupplier(1));
638+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
639+
640+
// Start the feed and consume some changes
641+
Stream<ChangesResultItem> stream = testFollower.startOneOff();
642+
stream.limit(5).forEach(item -> {
643+
// Just consume the items
644+
});
645+
646+
// Query with a sequence that's not in seqMarkers
647+
String unknownSeq = "999-unknown";
648+
String result = testFollower.latestSequenceFrom(unknownSeq);
649+
Assert.assertEquals(result, unknownSeq,
650+
"Should return the input sequence when it's not found in seqMarkers.");
651+
}
652+
653+
/**
654+
* Verify that getLastSeqNewerThan returns the associated last_seq when querying
655+
* with a user-facing sequence. The method returns the last_seq value associated
656+
* with the user sequence, not necessarily advancing through empty pages.
657+
*/
658+
@Test
659+
void testGetLastSeqNewerThanAfterStartWithNewerSeq() {
660+
// Create a mock client with a single batch
661+
Cloudant mockClient = new ChangesRequestMockClient(ChangesRequestMockClient.makeBatchSupplier(1));
662+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
663+
664+
// Start the feed and consume some changes
665+
Stream<ChangesResultItem> stream = testFollower.startOneOff();
666+
long count = stream.limit(5).count();
667+
Assert.assertEquals(count, 5L, "Should have consumed 5 changes.");
668+
669+
// Query with a user-facing sequence that exists in seqMarkers
670+
// The last item in the batch has seq "10000-g" which maps to last_seq "10000-g"
671+
String result = testFollower.latestSequenceFrom("10000-g");
672+
Assert.assertEquals(result, "10000-g",
673+
"Should return the last_seq associated with the user sequence.");
674+
}
675+
676+
/**
677+
* Verify that getLastSeqNewerThan returns the input sequence unchanged when
678+
* the sequence is not found in seqMarkers (e.g., querying with a sequence
679+
* from the middle of a batch that wasn't the last item).
680+
*/
681+
@Test
682+
void testGetLastSeqNewerThanAfterStartWithEmptyPages() {
683+
// Create a mock client with a single batch
684+
Cloudant mockClient = new ChangesRequestMockClient(ChangesRequestMockClient.makeBatchSupplier(1));
685+
ChangesFollower testFollower = new ChangesFollower(mockClient, TestOptions.MINIMUM.getOptions());
686+
687+
// Start the feed and consume all changes
688+
Stream<ChangesResultItem> stream = testFollower.startOneOff();
689+
stream.forEach(item -> {
690+
// Just consume the items
691+
});
692+
693+
// Query with a sequence that's not in seqMarkers (e.g., from middle of batch)
694+
// Only the LAST item's sequence is stored in seqMarkers
695+
String unknownSeq = "5000-g";
696+
String result = testFollower.latestSequenceFrom(unknownSeq);
697+
Assert.assertEquals(result, unknownSeq,
698+
"Should return the input sequence unchanged when not found in seqMarkers.");
699+
}
583700
}

‎modules/cloudant/src/test/java/com/ibm/cloud/cloudant/features/ChangesRequestMockClient.java‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -171,10 +171,27 @@ static final class MockChangesResult extends ChangesResult {
171171
MockChangesResult(List<ChangesResultItem> items, long pending) {
172172
this.pending = pending;
173173
this.results = items;
174-
this.lastSeq = this.results.isEmpty()
174+
this.lastSeq = this.results.isEmpty()
175175
? generateSeqLikeString(77777, 512) // generate a fake seq for the empty result
176176
: this.results.get(this.results.size()-1).getSeq();
177177
}
178+
179+
/**
180+
* Constructor for test scenarios requiring explicit sequence control.
181+
* Allows testing of seq_interval gaps and specific sequence patterns.
182+
*
183+
* @param userSeqs List of user-facing sequences (can contain nulls for gaps)
184+
* @param lastSeq The last_seq value for this page
185+
* @param pending Number of pending changes
186+
*/
187+
MockChangesResult(List<String> userSeqs, String lastSeq, long pending) {
188+
this.pending = pending;
189+
this.results = new ArrayList<>();
190+
for (String seq : userSeqs) {
191+
this.results.add(new MockChangesResultItem(seq));
192+
}
193+
this.lastSeq = lastSeq;
194+
}
178195
}
179196

180197
static final class MockChangesResultItem extends ChangesResultItem {
@@ -185,6 +202,19 @@ static final class MockChangesResultItem extends ChangesResultItem {
185202
this.id = generateAlphanumString(10);
186203
this.seq = generateSeqLikeString(counter, 512);
187204
}
205+
206+
/**
207+
* Constructor for test scenarios requiring explicit sequence control.
208+
*
209+
* @param seq The sequence string (can be null for seq_interval gaps)
210+
*/
211+
MockChangesResultItem(String seq) {
212+
this.changes = Collections.singletonList(new MockChange());
213+
this.deleted = false;
214+
this.doc = null;
215+
this.id = generateAlphanumString(10);
216+
this.seq = seq;
217+
}
188218
}
189219

190220
static final class MockChange extends Change {

0 commit comments

Comments
 (0)