Skip to content

Commit fee62e6

Browse files
sven1103-agentsven1103-agent
committed
fix: advance watermark updatedSince to now on last sync page
The partial-page branch of createNextWatermark() was setting updatedSince to the maximum registrationDate of the remote entities, which is a historical timestamp. This caused every subsequent sync run to query with an old date, always returning a full page and keeping morePages == true indefinitely. Fix: use Instant.now() (consistent with the empty-page branch) so the watermark always advances to the current time once the last page is consumed, ensuring future jobs only fetch newly registered datasets. Adds RawDataSyncServiceSpec with 7 Spock tests including a direct regression test for the infinite-pagination symptom.
1 parent 5ba81c3 commit fee62e6

2 files changed

Lines changed: 292 additions & 34 deletions

File tree

‎project-management/src/main/java/life/qbic/projectmanagement/application/sync/RawDataSyncService.java‎

Lines changed: 39 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -78,14 +78,17 @@ private boolean runSync() {
7878
// 1) load watermark (offset/updatedSince) from control table
7979
var currentWatermark = watermarkRepo.fetch(JOB_NAME)
8080
.orElse(new Watermark(JOB_NAME, 0, Instant.EPOCH, Instant.EPOCH));
81-
// 2) poll remote resource
81+
// 2) capture the query time BEFORE issuing the remote call so the watermark never skips
82+
// datasets that are registered on the remote between the query and the watermark save.
83+
Instant queryTime = Instant.now();
84+
// 3) poll remote resource
8285
var result = remoteRawDataService.registeredSince(currentWatermark.updatedSince(),
8386
currentWatermark.syncOffset(), MAX_QUERY_SIZE);
8487

8588
log.debug("Found %s new raw datasets in external resource. Syncing them now...".formatted(
8689
result.size()));
8790

88-
// 3) persist the raw dataset information if available
91+
// 4) persist the raw dataset information if available
8992
if (!result.isEmpty()) {
9093
log.info(
9194
"Persisting %d found external raw datasets in local resource. Syncing them now...".formatted(
@@ -94,55 +97,57 @@ private boolean runSync() {
9497
log.info("%d raw datasets synced.".formatted(result.size()));
9598
}
9699

97-
// 4) Create a new watermark for the next job to pick up at
98-
var nextWatermark = createNextWatermark(currentWatermark, result);
100+
// 5) Create a new watermark for the next job to pick up at
101+
var nextWatermark = createNextWatermark(currentWatermark, result, queryTime);
99102

100-
// 5) persist the new watermark state
103+
// 6) persist the new watermark state
101104
watermarkRepo.save(nextWatermark);
102105

103-
// 6) signal job state
106+
// 7) signal job state
104107
// if there were fewer results than the max query size for the search, this means
105108
// there are no more datasets available.
106109
// We can signal there are more datasets available (== true), else return false if finished
107110
return result.size() == MAX_QUERY_SIZE;
108111
}
109112

110113
/**
111-
* There are only two meaningful updates for the newOffset:
114+
* Computes the next watermark after a sync iteration.
115+
*
116+
* <p>There are two cases:
112117
* <ol>
113-
* <li>the result contained fewer entries then the max query size, then we are finished. So the offset can be set to 0
114-
* and the next job will start with a zero offset and the saved date</li>
115-
* <li> the result contained as many entries as the max query size, which means that there are still more datasets to be excepted
116-
* or zero, if the number of datasets % query size = 0. This will lead to the first condition in the next iteration.</li>
118+
* <li><b>Full page</b> ({@code result.size() == MAX_QUERY_SIZE}): more pages may exist.
119+
* Advance the offset by {@code MAX_QUERY_SIZE} and keep {@code updatedSince} unchanged so the
120+
* next call fetches the following page of the same time window.</li>
121+
* <li><b>Last page</b> (result is smaller than {@code MAX_QUERY_SIZE}, including empty): all
122+
* datasets in the current time window have been consumed. Reset offset to 0 and set
123+
* {@code updatedSince} to {@code queryTime} — the timestamp captured <em>before</em> the
124+
* remote call was issued — so that any dataset registered on the remote between the query and
125+
* the watermark save is included in the next sync window rather than silently skipped.</li>
117126
* </ol>
118127
*
119128
* @param currentWatermark the currently set watermark
120-
* @param result the result list seen from the last query, necessary so the last successful update date can be determined
129+
* @param result the result list from the last query
130+
* @param queryTime the {@link Instant} captured immediately before the remote query was
131+
* issued; must be passed from {@code runSync()} to avoid a race where
132+
* the watermark jumps past datasets registered after the query but
133+
* before {@code Instant.now()} would otherwise be called here
121134
* @return a new {@link Watermark} to continue at for the next job execution
122135
* @since 1.12.0
123136
*/
124-
private static Watermark createNextWatermark(Watermark currentWatermark,
125-
List<RawDataset> result) {
126-
boolean morePages = moreDatasetsToSync(result.size(), MAX_QUERY_SIZE);
127-
128-
if (morePages) {
129-
// Still paginating — advance offset, keep updatedSince unchanged
130-
int newOffset = currentWatermark.syncOffset() + MAX_QUERY_SIZE;
131-
return new Watermark(JOB_NAME, newOffset, currentWatermark.updatedSince(), Instant.now());
132-
}
133-
134-
if (result.isEmpty()) {
135-
// Empty page after an exact full-sized page
136-
// nothing was missed, safe to advance to now
137-
return new Watermark(JOB_NAME, 0, Instant.now(), Instant.now());
138-
}
139-
// Last partial page — advance updatedSince to the latest registration timestamp
140-
Instant lastEntityTimestamp = result.stream()
141-
.map(RawDataset::registrationDate)
142-
.max(Instant::compareTo)
143-
.orElse(currentWatermark.updatedSince());
144-
return new Watermark(JOB_NAME, 0, lastEntityTimestamp, Instant.now());
145-
}
137+
private static Watermark createNextWatermark(Watermark currentWatermark,
138+
List<RawDataset> result, Instant queryTime) {
139+
boolean morePages = moreDatasetsToSync(result.size(), MAX_QUERY_SIZE);
140+
141+
if (morePages) {
142+
// Still paginating — advance offset, keep updatedSince unchanged
143+
int newOffset = currentWatermark.syncOffset() + MAX_QUERY_SIZE;
144+
return new Watermark(JOB_NAME, newOffset, currentWatermark.updatedSince(), Instant.now());
145+
}
146+
// Last page (empty or partial) — all pages consumed. Use queryTime (captured before the
147+
// remote call) as updatedSince so the next scheduled job does not skip datasets that were
148+
// registered on the remote between the query and this watermark save.
149+
return new Watermark(JOB_NAME, 0, queryTime, Instant.now());
150+
}
146151

147152
private static boolean moreDatasetsToSync(int lastResultSize, int maxQuerySize) {
148153
// if the last query returned fewer items than could have been based on the max query

‎project-management/src/test/groovy/life/qbic/projectmanagement/application/sync/RawDataSyncServiceSpec.groovy‎

Lines changed: 253 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,253 @@
1+
package life.qbic.projectmanagement.application.sync
2+
3+
import life.qbic.projectmanagement.application.api.AsyncProjectService.RawDataset
4+
import life.qbic.projectmanagement.application.dataset.LocalRawDatasetCache
5+
import life.qbic.projectmanagement.application.dataset.RemoteRawDataService
6+
import life.qbic.projectmanagement.application.sync.WatermarkRepo.Watermark
7+
import spock.lang.Specification
8+
import spock.lang.Subject
9+
10+
import java.time.Instant
11+
12+
/**
13+
* Unit tests for {@link RawDataSyncService}, focusing on watermark update logic.
14+
*
15+
* <p>The outer {@code sync()} method runs a time-boxed while-loop calling {@code runSync()}.
16+
* Tests that need to observe a single iteration set up the mocks so that the first call returns
17+
* a partial/empty page (causing {@code runSync()} to return {@code false} and the loop to break
18+
* immediately). Tests that need to observe two iterations chain mock responses.</p>
19+
*/
20+
class RawDataSyncServiceSpec extends Specification {
21+
22+
static final int MAX_QUERY_SIZE = 1_000
23+
static final String JOB_NAME = "RAW_DATA_SYNC_EXTERNAL"
24+
25+
RemoteRawDataService remoteRawDataServiceMock = Mock()
26+
WatermarkRepo watermarkRepoMock = Mock()
27+
LocalRawDatasetCache localRawDatasetCacheMock = Mock()
28+
29+
@Subject
30+
RawDataSyncService service = new RawDataSyncService(
31+
remoteRawDataServiceMock,
32+
watermarkRepoMock,
33+
localRawDatasetCacheMock
34+
)
35+
36+
// ------------------------------------------------------------------
37+
// Helpers
38+
// ------------------------------------------------------------------
39+
40+
private static RawDataset dataset(Instant registrationDate) {
41+
new RawDataset("QABCD001", 1024L, 1, Set.of(".fastq"), registrationDate)
42+
}
43+
44+
private static List<RawDataset> fullPage() {
45+
(1..MAX_QUERY_SIZE).collect { dataset(Instant.EPOCH.plusSeconds(it)) }
46+
}
47+
48+
private static List<RawDataset> partialPage(int size) {
49+
assert size > 0 && size < MAX_QUERY_SIZE
50+
(1..size).collect { dataset(Instant.EPOCH.plusSeconds(it)) }
51+
}
52+
53+
// ------------------------------------------------------------------
54+
// Race-condition fix: updatedSince must be captured BEFORE the remote query
55+
// ------------------------------------------------------------------
56+
57+
def "race condition fix: updatedSince is not later than lastSuccessAt, proving it was captured before the query completed"() {
58+
given: "no prior watermark"
59+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.empty()
60+
61+
and: "the remote returns a partial page so the loop terminates after one iteration"
62+
remoteRawDataServiceMock.registeredSince(_, _, _) >> partialPage(3)
63+
64+
Watermark savedWatermark = null
65+
watermarkRepoMock.save(_ as Watermark) >> { Watermark w -> savedWatermark = w }
66+
67+
when: "we capture the time just before the sync runs"
68+
Instant beforeSync = Instant.now()
69+
service.sync()
70+
Instant afterSync = Instant.now()
71+
72+
then: "updatedSince falls within the [beforeSync, afterSync] window"
73+
// updatedSince must be >= beforeSync: it was captured inside runSync(), after entry
74+
!savedWatermark.updatedSince().isBefore(beforeSync)
75+
// updatedSince must be <= afterSync: it was captured before sync() returned
76+
!savedWatermark.updatedSince().isAfter(afterSync)
77+
78+
and: "updatedSince is not after lastSuccessAt, proving it was captured before (or at the same moment as) the save"
79+
// queryTime is captured before the remote call; lastSuccessAt is captured after.
80+
// Therefore updatedSince <= lastSuccessAt always holds when the fix is in place.
81+
!savedWatermark.updatedSince().isAfter(savedWatermark.lastSuccessAt())
82+
}
83+
84+
// ------------------------------------------------------------------
85+
// Bug regression: partial page must update watermark to now
86+
// ------------------------------------------------------------------
87+
88+
def "when the last page is partial, the saved watermark updatedSince is close to now (not a remote entity timestamp)"() {
89+
given: "no prior watermark"
90+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.empty()
91+
92+
and: "the remote returns a partial page (< MAX_QUERY_SIZE) — loop will break after one iteration"
93+
def items = partialPage(5)
94+
remoteRawDataServiceMock.registeredSince(_, _, _) >> items
95+
96+
and: "capture the saved watermark"
97+
Watermark savedWatermark = null
98+
watermarkRepoMock.save(_ as Watermark) >> { Watermark w -> savedWatermark = w }
99+
100+
when:
101+
service.sync()
102+
103+
then: "updatedSince is NOT one of the ancient remote registration timestamps (EPOCH + a few seconds)"
104+
savedWatermark != null
105+
// All item registrationDates are <= EPOCH + 5s (effectively year 1970).
106+
// The saved updatedSince must be close to the current time (within 10s of now).
107+
savedWatermark.updatedSince().isAfter(Instant.now().minusSeconds(10))
108+
109+
and: "the offset is reset to 0 so the next run starts fresh"
110+
savedWatermark.syncOffset() == 0
111+
}
112+
113+
def "bug regression: watermark does NOT stay stuck at the remote entity timestamp after a full sync completes"() {
114+
given: "an initial watermark stuck at EPOCH (simulating the bug state)"
115+
def priorWatermark = new Watermark(JOB_NAME, 0, Instant.EPOCH, Instant.EPOCH)
116+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.of(priorWatermark)
117+
118+
and: "the remote returns 10 items with very old registrationDates (EPOCH + 1..10 s)"
119+
def oldItems = (1..10).collect { dataset(Instant.EPOCH.plusSeconds(it)) }
120+
remoteRawDataServiceMock.registeredSince(_, _, _) >> oldItems
121+
122+
Watermark savedWatermark = null
123+
watermarkRepoMock.save(_ as Watermark) >> { Watermark w -> savedWatermark = w }
124+
125+
when:
126+
service.sync()
127+
128+
then: "the saved watermark's updatedSince is NOT stuck at an ancient remote timestamp"
129+
savedWatermark.updatedSince() != Instant.EPOCH
130+
savedWatermark.updatedSince().isAfter(Instant.now().minusSeconds(10))
131+
132+
and: "offset is 0 so we do not re-paginate the old window"
133+
savedWatermark.syncOffset() == 0
134+
}
135+
136+
// ------------------------------------------------------------------
137+
// Empty page: watermark also advances to now
138+
// ------------------------------------------------------------------
139+
140+
def "when the remote returns an empty page, the saved watermark updatedSince is close to now"() {
141+
given: "a prior watermark with a non-zero offset (previous page was exactly full)"
142+
def priorWatermark = new Watermark(JOB_NAME, MAX_QUERY_SIZE, Instant.EPOCH, Instant.EPOCH)
143+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.of(priorWatermark)
144+
145+
and: "the remote returns an empty page — loop will break after one iteration"
146+
remoteRawDataServiceMock.registeredSince(_, _, _) >> []
147+
148+
Watermark savedWatermark = null
149+
watermarkRepoMock.save(_ as Watermark) >> { Watermark w -> savedWatermark = w }
150+
151+
when:
152+
service.sync()
153+
154+
then:
155+
savedWatermark != null
156+
savedWatermark.updatedSince().isAfter(Instant.now().minusSeconds(10))
157+
savedWatermark.syncOffset() == 0
158+
}
159+
160+
// ------------------------------------------------------------------
161+
// Full page: offset advances, updatedSince stays unchanged
162+
// ------------------------------------------------------------------
163+
164+
def "when a full page is returned, offset advances and updatedSince is preserved until the last page"() {
165+
given: "a prior watermark at offset 0 with a specific updatedSince"
166+
Instant originalUpdatedSince = Instant.parse("2025-01-01T00:00:00Z")
167+
def initialWatermark = new Watermark(JOB_NAME, 0, originalUpdatedSince, Instant.EPOCH)
168+
169+
// First fetch returns the initial watermark; the second fetch (after offset advances)
170+
// simulates the stored watermark being reloaded by Spock's iterator stub behaviour.
171+
// We give the second fetch a watermark with the advanced offset so runSync can terminate.
172+
def advancedWatermark = new Watermark(JOB_NAME, MAX_QUERY_SIZE, originalUpdatedSince, Instant.now())
173+
watermarkRepoMock.fetch(JOB_NAME) >>> [
174+
Optional.of(initialWatermark), // 1st iteration
175+
Optional.of(advancedWatermark) // 2nd iteration
176+
]
177+
178+
and: "first remote call returns a full page; second call (advanced offset) returns a partial page"
179+
remoteRawDataServiceMock.registeredSince(_, 0, _) >> fullPage()
180+
remoteRawDataServiceMock.registeredSince(_, MAX_QUERY_SIZE, _) >> partialPage(3)
181+
182+
List<Watermark> savedWatermarks = []
183+
watermarkRepoMock.save(_ as Watermark) >> { Watermark w -> savedWatermarks << w }
184+
185+
when:
186+
service.sync()
187+
188+
then: "exactly two watermarks were saved"
189+
savedWatermarks.size() == 2
190+
191+
and: "the first saved watermark advances the offset and keeps updatedSince unchanged"
192+
savedWatermarks[0].syncOffset() == MAX_QUERY_SIZE
193+
savedWatermarks[0].updatedSince() == originalUpdatedSince
194+
195+
and: "the second saved watermark (last page) resets the offset and advances updatedSince to now"
196+
savedWatermarks[1].syncOffset() == 0
197+
savedWatermarks[1].updatedSince().isAfter(Instant.now().minusSeconds(10))
198+
}
199+
200+
// ------------------------------------------------------------------
201+
// Dataset persistence
202+
// ------------------------------------------------------------------
203+
204+
def "datasets from a partial page are persisted"() {
205+
given:
206+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.empty()
207+
def items = partialPage(7)
208+
remoteRawDataServiceMock.registeredSince(_, _, _) >> items
209+
watermarkRepoMock.save(_ as Watermark) >> {}
210+
211+
when:
212+
service.sync()
213+
214+
then:
215+
1 * localRawDatasetCacheMock.saveAll(items)
216+
}
217+
218+
def "datasets from both pages are persisted during multi-page sync"() {
219+
given:
220+
def initialWatermark = new Watermark(JOB_NAME, 0, Instant.EPOCH, Instant.EPOCH)
221+
def advancedWatermark = new Watermark(JOB_NAME, MAX_QUERY_SIZE, Instant.EPOCH, Instant.now())
222+
watermarkRepoMock.fetch(JOB_NAME) >>> [
223+
Optional.of(initialWatermark),
224+
Optional.of(advancedWatermark)
225+
]
226+
227+
def page1 = fullPage()
228+
def page2 = partialPage(1)
229+
remoteRawDataServiceMock.registeredSince(_, 0, _) >> page1
230+
remoteRawDataServiceMock.registeredSince(_, MAX_QUERY_SIZE, _) >> page2
231+
watermarkRepoMock.save(_ as Watermark) >> {}
232+
233+
when:
234+
service.sync()
235+
236+
then:
237+
1 * localRawDatasetCacheMock.saveAll(page1)
238+
1 * localRawDatasetCacheMock.saveAll(page2)
239+
}
240+
241+
def "when the remote returns an empty page, no datasets are persisted"() {
242+
given:
243+
watermarkRepoMock.fetch(JOB_NAME) >> Optional.empty()
244+
remoteRawDataServiceMock.registeredSince(_, _, _) >> []
245+
watermarkRepoMock.save(_ as Watermark) >> {}
246+
247+
when:
248+
service.sync()
249+
250+
then:
251+
0 * localRawDatasetCacheMock.saveAll(_)
252+
}
253+
}

0 commit comments

Comments
 (0)