From 16fcbea76bd060f8e8c4dcfb39885c4a9a621c7a Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Fri, 28 Aug 2026 08:24:20 +0200 Subject: [PATCH 1/6] bugfix(#1212) Return empty TimeRange for null start and end times --- .../reitti/service/processing/LocationPointStagingService.java | 1 + 1 file changed, 1 insertion(+) diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java index 3e4da308..6616e724 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java @@ -18,6 +18,7 @@ import java.sql.*; import java.util.List; import java.util.Set; +import java.util.Timer; import java.util.concurrent.ConcurrentHashMap; @Service From e6eac1a7030f16683120b9582ebbb60735dd9eb5 Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Fri, 28 Aug 2026 08:24:52 +0200 Subject: [PATCH 2/6] bugfix(#1212) Remove unused Timer import from LocationPointStagingService --- .../reitti/service/processing/LocationPointStagingService.java | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java index 6616e724..3e4da308 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java @@ -18,7 +18,6 @@ import java.sql.*; import java.util.List; import java.util.Set; -import java.util.Timer; import java.util.concurrent.ConcurrentHashMap; @Service From c49e10980b3bed177f5eaa29ac9988424ce04b2b Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Sat, 29 Aug 2026 05:18:54 +0200 Subject: [PATCH 3/6] bugfix(#1212) Fix incorrect timeRange comparison in PromotionJobHandler --- .../reitti/service/importer/PromotionJobHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java index 2dd91e27..5d1ef1a8 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java +++ b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java @@ -65,7 +65,7 @@ public void execute(JobExecutionContext context) throws JobExecutionException { if (user.getUserType() == UserType.LIVE_DATA_ONLY) { metadataRepository.updateProgress(jobId, 2, 3, "Live data only, skipping cleanup"); - if (timeRange != null) { + if (!timeRange.equals(TimeRange.empty())) { this.jobSchedulingService.enqueueTask(liveModeOnlyUpdateTask, new LiveModeOnlyUpdateTask.TaskData(user, data.getDevice(), timeRange.start(), timeRange.end()).withParentJobId(data.getParentJobId()), JobSchedulingService.Metadata.builder() From babf2da0b6d113587f3a7ce8e398fab0725e01a6 Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Sat, 29 Aug 2026 07:00:12 +0200 Subject: [PATCH 4/6] bugfix(#1212) Add tests for job metadata cleanup, inflight promotion guard, and promotion time range fixes - Add `JobMetadataCleanupServiceTest` to verify orphan trigger fix. - Add `LocationBatchingServiceTest` to enforce single promotion inflight per partition. - Add `PromotionCleanupChurnTest` to demonstrate and verify promotion cleanup bottlenecks. - Enhance `LocationPointStagingServiceTest` to check correct promotion ranges and concurrent flush handling. - Refactor `PromotionJobHandler` to ensure inflight guard and transactional integrity during promotion. --- .../repository/JobMetadataRepository.java | 2 +- .../service/LocationBatchingService.java | 30 +++- .../service/importer/PromotionJobHandler.java | 95 +++++++---- .../service/jobs/PromotionInflightGuard.java | 20 +++ .../LocationPointStagingService.java | 75 ++++----- .../service/LocationBatchingServiceTest.java | 98 +++++++++++ .../importer/PromotionJobHandlerTest.java | 147 ++++++++++++++++ .../jobs/JobMetadataCleanupServiceTest.java | 61 +++++++ .../LocationPointStagingServiceTest.java | 121 +++++++++++++ .../processing/PromotionCleanupChurnTest.java | 159 ++++++++++++++++++ 10 files changed, 725 insertions(+), 83 deletions(-) create mode 100644 src/main/java/com/dedicatedcode/reitti/service/jobs/PromotionInflightGuard.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/LocationBatchingServiceTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandlerTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/jobs/JobMetadataCleanupServiceTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingServiceTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java diff --git a/src/main/java/com/dedicatedcode/reitti/repository/JobMetadataRepository.java b/src/main/java/com/dedicatedcode/reitti/repository/JobMetadataRepository.java index 8b5bfdae..e735cbd8 100644 --- a/src/main/java/com/dedicatedcode/reitti/repository/JobMetadataRepository.java +++ b/src/main/java/com/dedicatedcode/reitti/repository/JobMetadataRepository.java @@ -164,7 +164,7 @@ public void delete(UUID jobId) { } public int deleteOlderThan(Instant cutoff) { - String sql = "DELETE FROM job_meta_data WHERE enqueued_at < ?"; + String sql = "DELETE FROM job_meta_data WHERE enqueued_at < ? AND status IN ('COMPLETED', 'FAILED', 'CANCELLED')"; return jdbcTemplate.update(sql, Timestamp.from(cutoff)); } diff --git a/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java b/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java index 2176de0e..e1edd675 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java @@ -6,6 +6,7 @@ import com.dedicatedcode.reitti.service.importer.PromotionJobHandler; import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; import com.dedicatedcode.reitti.service.jobs.JobType; +import com.dedicatedcode.reitti.service.jobs.PromotionInflightGuard; import com.dedicatedcode.reitti.service.processing.LocationPointStagingService; import jakarta.annotation.PreDestroy; import org.quartz.JobDetail; @@ -33,19 +34,22 @@ public class LocationBatchingService { private final LocationPointStagingService locationPointStagingService; private final JobDetail promotionTask; private final JobSchedulingService jobScheduler; + private final PromotionInflightGuard promotionInflightGuard; private final int maxBatchSize; private final long maxWaitTimeMs; - + @Autowired public LocationBatchingService(LocationPointStagingService locationPointStagingService, @Qualifier("promotionJob") JobDetail promotionTask, JobSchedulingService jobScheduler, + PromotionInflightGuard promotionInflightGuard, @Value("${reitti.batching.max-batch-size:100}") int maxBatchSize, @Value("${reitti.batching.max-wait-time:5}") long maxWaitTime) { this.locationPointStagingService = locationPointStagingService; this.promotionTask = promotionTask; this.jobScheduler = jobScheduler; + this.promotionInflightGuard = promotionInflightGuard; this.maxBatchSize = maxBatchSize; this.maxWaitTimeMs = maxWaitTime * 1000; } @@ -91,12 +95,24 @@ private void executeFlush(UserBatch batch) { batch.getLocationPoints() ); batch.clear(); - this.jobScheduler.enqueueTask(promotionTask, - new PromotionJobHandler.TaskData(batch.user, batch.device, pKey, false), - JobSchedulingService.Metadata.builder() - .user(batch.user) - .jobType(JobType.GPS_INGESTION) - .friendlyName("GPS Data Promotion").build()); + // coalesce: the rows are staged, an already scheduled or running promotion for this + // partition will pick them up. This keeps us from enqueueing a new promotion trigger + // every few seconds, which lets the Quartz queue grow faster than it drains (#1212). + if (promotionInflightGuard.tryAcquire(pKey)) { + try { + this.jobScheduler.enqueueTask(promotionTask, + new PromotionJobHandler.TaskData(batch.user, batch.device, pKey, false), + JobSchedulingService.Metadata.builder() + .user(batch.user) + .jobType(JobType.GPS_INGESTION) + .friendlyName("GPS Data Promotion").build()); + } catch (RuntimeException e) { + promotionInflightGuard.release(pKey); + throw e; + } + } else { + logger.debug("Promotion already in flight for partition {}, skipping enqueue", pKey); + } } catch (Exception e) { logger.error("Failed to flush batch for partition {}", batch.getPartitionKey(), e); } diff --git a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java index 5d1ef1a8..9a8274a4 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java +++ b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java @@ -8,15 +8,18 @@ import com.dedicatedcode.reitti.service.UserNotificationService; import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; import com.dedicatedcode.reitti.service.jobs.JobType; +import com.dedicatedcode.reitti.service.jobs.PromotionInflightGuard; import com.dedicatedcode.reitti.service.processing.LiveModeOnlyUpdateTask; import com.dedicatedcode.reitti.service.processing.LocationDataCleanupTask; import com.dedicatedcode.reitti.service.processing.LocationPointStagingService; +import com.dedicatedcode.reitti.service.processing.LocationPointStagingService.PromotionResult; import com.dedicatedcode.reitti.service.processing.TimeRange; import org.quartz.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.stereotype.Component; +import org.springframework.transaction.annotation.Transactional; import java.util.UUID; @@ -28,76 +31,96 @@ public class PromotionJobHandler implements Job { private final JobSchedulingService jobSchedulingService; private final JobMetadataRepository metadataRepository; private final UserNotificationService userNotificationService; + private final PromotionInflightGuard promotionInflightGuard; private final JobDetail locationDataCleanupTask; private final JobDetail liveModeOnlyUpdateTask; + private final JobDetail promotionTask; public PromotionJobHandler(LocationPointStagingService stagingService, JobSchedulingService jobSchedulingService, JobMetadataRepository metadataRepository, UserNotificationService userNotificationService, + PromotionInflightGuard promotionInflightGuard, @Qualifier("locationDataCleanupJob") JobDetail locationDataCleanupTask, - @Qualifier("liveModeUserUpdateJob") JobDetail liveModeOnlyUpdateTask) { + @Qualifier("liveModeUserUpdateJob") JobDetail liveModeOnlyUpdateTask, + @Qualifier("promotionJob") JobDetail promotionTask) { this.stagingService = stagingService; this.jobSchedulingService = jobSchedulingService; this.metadataRepository = metadataRepository; this.userNotificationService = userNotificationService; + this.promotionInflightGuard = promotionInflightGuard; this.locationDataCleanupTask = locationDataCleanupTask; this.liveModeOnlyUpdateTask = liveModeOnlyUpdateTask; + this.promotionTask = promotionTask; } @Override + @Transactional public void execute(JobExecutionContext context) throws JobExecutionException { JobDataMap dataMap = context.getMergedJobDataMap(); TaskData data = (TaskData) dataMap.get("data"); UUID jobId = data.getJobId(); User user = data.getUser(); String partitionKey = data.getPartitionKey(); - TimeRange timeRange = this.stagingService.getTimeRange(partitionKey); - metadataRepository.updateProgress(jobId, 0, 3, "Promoting points"); - int promote = this.stagingService.promote(user, partitionKey); - metadataRepository.updateProgress(jobId, 1, 3, "Dropping partition"); - - log.debug("Promoted [{}] points into live table", promote); - if (data.isManual()) { - this.stagingService.dropPartition(partitionKey); - } - - if (user.getUserType() == UserType.LIVE_DATA_ONLY) { - metadataRepository.updateProgress(jobId, 2, 3, "Live data only, skipping cleanup"); - - if (!timeRange.equals(TimeRange.empty())) { - this.jobSchedulingService.enqueueTask(liveModeOnlyUpdateTask, - new LiveModeOnlyUpdateTask.TaskData(user, data.getDevice(), timeRange.start(), timeRange.end()).withParentJobId(data.getParentJobId()), - JobSchedulingService.Metadata.builder() - .user(user) - .jobType(JobType.LOCATION_PROCESSING) - .friendlyName("Location Data Cleanup") - .build()); + try { + metadataRepository.updateProgress(jobId, 0, 3, "Promoting points"); + PromotionResult promotionResult = this.stagingService.promote(user, partitionKey); + metadataRepository.updateProgress(jobId, 1, 3, "Dropping partition"); + + log.debug("Promoted [{}] points into live table", promotionResult.promotedCount()); + if (data.isManual()) { + this.stagingService.dropPartition(partitionKey); } - } else { - metadataRepository.updateProgress(jobId, 2, 3, "Scheduling cleanup job"); - if (promote > 0) { - if (timeRange.equals(TimeRange.empty())) { - log.debug("No timerange found for partitionKey [{}], recalculating", partitionKey); - timeRange = this.stagingService.getWholeTimeRange(partitionKey); + if (user.getUserType() == UserType.LIVE_DATA_ONLY) { + metadataRepository.updateProgress(jobId, 2, 3, "Live data only, skipping cleanup"); + + if (promotionResult.hasPromoted()) { + TimeRange promotedRange = promotionResult.promotedRange(); + this.jobSchedulingService.enqueueTask(liveModeOnlyUpdateTask, + new LiveModeOnlyUpdateTask.TaskData(user, data.getDevice(), promotedRange.start(), promotedRange.end()).withParentJobId(data.getParentJobId()), + JobSchedulingService.Metadata.builder() + .user(user) + .jobType(JobType.LOCATION_PROCESSING) + .friendlyName("Location Data Cleanup") + .build()); } - if (timeRange.equals(TimeRange.empty())) { - log.warn("Still no timerange found for partitionKey [{}], skipping cleanup", partitionKey); - } else { - this.userNotificationService.newLocationData(user, data.device, timeRange); + } else { + metadataRepository.updateProgress(jobId, 2, 3, "Scheduling cleanup job"); + + if (promotionResult.hasPromoted()) { + TimeRange promotedRange = promotionResult.promotedRange(); + this.userNotificationService.newLocationData(user, data.device, promotedRange); this.jobSchedulingService.enqueueTask(locationDataCleanupTask, - new LocationDataCleanupTask.TaskData(user, data.getDevice(), timeRange.start(), timeRange.end()).withParentJobId(data.getParentJobId()), + new LocationDataCleanupTask.TaskData(user, data.getDevice(), promotedRange.start(), promotedRange.end()).withParentJobId(data.getParentJobId()), JobSchedulingService.Metadata.builder() .user(user) .jobType(JobType.LOCATION_DATA_CLEANUP) .friendlyName("Location Data Cleanup") .build()); + } else { + log.debug("No points to promote for partitionKey [{}]", partitionKey); } - } else { - log.debug("No points to promote, timerange was [{}]", timeRange); + metadataRepository.updateProgress(jobId, 3, 3, "Done"); + } + } finally { + promotionInflightGuard.release(partitionKey); + } + + // points flushed by the ingest thread while this promotion was running are still + // unpromoted - make sure they get promoted instead of waiting for the next flush + if (this.stagingService.hasUnpromotedPoints(partitionKey) && promotionInflightGuard.tryAcquire(partitionKey)) { + try { + this.jobSchedulingService.enqueueTask(promotionTask, + new TaskData(user, data.getDevice(), partitionKey, false), + JobSchedulingService.Metadata.builder() + .user(user) + .jobType(JobType.GPS_INGESTION) + .friendlyName("GPS Data Promotion").build()); + } catch (RuntimeException e) { + promotionInflightGuard.release(partitionKey); + throw e; } - metadataRepository.updateProgress(jobId, 3, 3, "Done"); } } diff --git a/src/main/java/com/dedicatedcode/reitti/service/jobs/PromotionInflightGuard.java b/src/main/java/com/dedicatedcode/reitti/service/jobs/PromotionInflightGuard.java new file mode 100644 index 00000000..d0dcdd92 --- /dev/null +++ b/src/main/java/com/dedicatedcode/reitti/service/jobs/PromotionInflightGuard.java @@ -0,0 +1,20 @@ +package com.dedicatedcode.reitti.service.jobs; + +import org.springframework.stereotype.Service; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +@Service +public class PromotionInflightGuard { + + private final Set inFlight = ConcurrentHashMap.newKeySet(); + + public boolean tryAcquire(String partitionKey) { + return this.inFlight.add(partitionKey); + } + + public void release(String partitionKey) { + this.inFlight.remove(partitionKey); + } +} diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java index 3e4da308..ae5a6fd9 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingService.java @@ -16,6 +16,7 @@ import org.springframework.transaction.annotation.Transactional; import java.sql.*; +import java.time.Instant; import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -115,55 +116,51 @@ public int getBatchSize() { } @Transactional - public int promote(User user, String partitionKey) { + public PromotionResult promote(User user, String partitionKey) { String sql = """ - INSERT INTO raw_source_points ( - user_id, device_id, timestamp, accuracy_meters, elevation_meters, - geom, invalid, status, h3_cell + WITH promoted_rows AS ( + UPDATE staging_location_points SET promoted = TRUE + WHERE partition_key = ? AND promoted = FALSE + RETURNING user_id, device_id, timestamp, accuracy_meters, elevation_meters, geom, h3_cell + ), inserted AS ( + INSERT INTO raw_source_points ( + user_id, device_id, timestamp, accuracy_meters, elevation_meters, + geom, invalid, status, h3_cell + ) + SELECT + user_id, device_id, timestamp, accuracy_meters, elevation_meters, + geom, false, 0, h3_cell + FROM promoted_rows + ON CONFLICT (user_id, device_id, timestamp) DO NOTHING + RETURNING id, timestamp ) - SELECT - user_id, device_id, timestamp, accuracy_meters, elevation_meters, - geom, false, 0, h3_cell - FROM staging_location_points - WHERE partition_key = ? AND promoted = FALSE - ON CONFLICT (user_id, device_id, timestamp) DO NOTHING - RETURNING id; - """; + SELECT id, timestamp FROM inserted + """; - List insertedIds = jdbcTemplate.queryForList(sql, Long.class, partitionKey); + List inserted = this.jdbcTemplate.query(sql, (rs, rowNum) -> new PromotedPoint(rs.getLong("id"), rs.getTimestamp("timestamp").toInstant()), partitionKey); if (user.getUserType() == UserType.NORMAL) { - spatialCoverageService.postPromotion(insertedIds); + spatialCoverageService.postPromotion(inserted.stream().map(PromotedPoint::id).toList()); + } + if (inserted.isEmpty()) { + return new PromotionResult(0, TimeRange.empty()); } - this.jdbcTemplate.update("UPDATE staging_location_points SET promoted = TRUE WHERE partition_key = ? AND promoted = FALSE", partitionKey); - return insertedIds.size(); + Instant min = inserted.stream().map(PromotedPoint::timestamp).min(Instant::compareTo).orElseThrow(); + Instant max = inserted.stream().map(PromotedPoint::timestamp).max(Instant::compareTo).orElseThrow(); + return new PromotionResult(inserted.size(), new TimeRange(min, max)); } - public TimeRange getWholeTimeRange(String partitionKey) { - String sql = "SELECT MIN(timestamp) as start_time, MAX(timestamp) as end_time FROM staging_location_points WHERE partition_key = ?"; - return this.jdbcTemplate.queryForObject(sql, (rs, rowNum) -> { - Timestamp start = rs.getTimestamp("start_time"); - Timestamp end = rs.getTimestamp("end_time"); - - if (start == null || end == null) { - return TimeRange.empty(); - } - - return new TimeRange(start.toInstant(), end.toInstant()); - }, partitionKey); + public boolean hasUnpromotedPoints(String partitionKey) { + String sql = "SELECT EXISTS(SELECT 1 FROM staging_location_points WHERE partition_key = ? AND promoted = FALSE)"; + return Boolean.TRUE.equals(this.jdbcTemplate.queryForObject(sql, Boolean.class, partitionKey)); } - public TimeRange getTimeRange(String partitionKey) { - String sql = "SELECT MIN(timestamp) as start_time, MAX(timestamp) as end_time FROM staging_location_points WHERE partition_key = ? AND promoted = FALSE"; - return this.jdbcTemplate.queryForObject(sql, (rs, rowNum) -> { - Timestamp start = rs.getTimestamp("start_time"); - Timestamp end = rs.getTimestamp("end_time"); - - if (start == null || end == null) { - return TimeRange.empty(); - } + public record PromotionResult(int promotedCount, TimeRange promotedRange) { + public boolean hasPromoted() { + return promotedCount > 0 && promotedRange != null && !promotedRange.equals(TimeRange.empty()); + } + } - return new TimeRange(start.toInstant(), end.toInstant()); - }, partitionKey); + private record PromotedPoint(long id, Instant timestamp) { } @Scheduled(cron = "${reitti.import.staging.cleanup.cron}") diff --git a/src/test/java/com/dedicatedcode/reitti/service/LocationBatchingServiceTest.java b/src/test/java/com/dedicatedcode/reitti/service/LocationBatchingServiceTest.java new file mode 100644 index 00000000..f8008143 --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/LocationBatchingServiceTest.java @@ -0,0 +1,98 @@ +package com.dedicatedcode.reitti.service; + +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.model.Role; +import com.dedicatedcode.reitti.model.UserType; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.service.importer.PromotionJobHandler; +import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; +import com.dedicatedcode.reitti.service.jobs.PromotionInflightGuard; +import com.dedicatedcode.reitti.service.processing.LocationPointStagingService; +import org.junit.jupiter.api.Test; +import org.quartz.JobDetail; + +import java.time.Instant; +import java.time.LocalDate; +import java.time.format.DateTimeFormatter; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * Verifies the #1212 trigger flood fix: while a promotion for a partition is scheduled or + * running, further flushes of the same partition must not enqueue additional promotion + * triggers - the queued trigger count has to stay bounded, otherwise Quartz starves. + */ +class LocationBatchingServiceTest { + + private final LocationPointStagingService stagingService = mock(LocationPointStagingService.class); + private final JobDetail promotionTask = mock(JobDetail.class); + private final JobSchedulingService jobScheduler = mock(JobSchedulingService.class); + private final PromotionInflightGuard promotionInflightGuard = new PromotionInflightGuard(); + + private final User user = new User(1L, "tester", null, "Tester", null, null, Role.USER, UserType.NORMAL, 0L); + private final Device device = new Device(3L, "phone", true, true, true, "#ffffff", true, Instant.now(), Instant.now(), 0L); + + private String partitionKey() { + return String.format("stream_%d_%s_%s", user.getId(), device.id(), LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)).toLowerCase(); + } + + @Test + void enqueuesOnlyOnePromotionWhileAnotherIsInFlight() { + LocationBatchingService service = new LocationBatchingService(stagingService, promotionTask, jobScheduler, promotionInflightGuard, 1, 5); + + // first point fills the batch (maxBatchSize=1) -> flush -> promotion enqueued + service.addLocationPoint(user, device, point("2026-08-27T10:00:00Z")); + verify(jobScheduler, times(1)).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + + // further flushes while that promotion is scheduled/running must not create new triggers + service.addLocationPoint(user, device, point("2026-08-27T10:00:05Z")); + service.addLocationPoint(user, device, point("2026-08-27T10:00:10Z")); + verify(jobScheduler, times(1)).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + + // the promoted rows are staged nevertheless, so the pending promotion picks them up + verify(stagingService, times(3)).insertBatch(eq(partitionKey()), eq(user), eq(device), any()); + } + + @Test + void enqueuesAgainAfterTheRunningPromotionReleasedThePartition() { + LocationBatchingService service = new LocationBatchingService(stagingService, promotionTask, jobScheduler, promotionInflightGuard, 1, 5); + + service.addLocationPoint(user, device, point("2026-08-27T10:00:00Z")); + verify(jobScheduler, times(1)).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + + // PromotionJobHandler releases the partition after execution + promotionInflightGuard.release(partitionKey()); + + service.addLocationPoint(user, device, point("2026-08-27T10:00:05Z")); + verify(jobScheduler, times(2)).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + } + + @Test + void releaseOnFailedEnqueueDoesNotBlockFurtherPromotions() { + LocationBatchingService service = new LocationBatchingService(stagingService, promotionTask, jobScheduler, promotionInflightGuard, 1, 5); + + org.mockito.Mockito.doThrow(new IllegalStateException("scheduler down")) + .when(jobScheduler).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + service.addLocationPoint(user, device, point("2026-08-27T10:00:00Z")); + + // enqueue failed -> guard must have been released again, otherwise the partition + // would never be promoted until restart + service.addLocationPoint(user, device, point("2026-08-27T10:00:05Z")); + verify(jobScheduler, times(2)).enqueueTask(eq(promotionTask), any(PromotionJobHandler.TaskData.class), any()); + } + + private LocationPoint point(String timestamp) { + LocationPoint locationPoint = new LocationPoint(); + locationPoint.setTimestamp(Instant.parse(timestamp)); + locationPoint.setLatitude(53.551086); + locationPoint.setLongitude(9.993682); + locationPoint.setAccuracyMeters(10.0); + return locationPoint; + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandlerTest.java b/src/test/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandlerTest.java new file mode 100644 index 00000000..9325fdff --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandlerTest.java @@ -0,0 +1,147 @@ +package com.dedicatedcode.reitti.service.importer; + +import com.dedicatedcode.reitti.model.Role; +import com.dedicatedcode.reitti.model.UserType; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.JobMetadataRepository; +import com.dedicatedcode.reitti.service.UserNotificationService; +import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; +import com.dedicatedcode.reitti.service.jobs.PromotionInflightGuard; +import com.dedicatedcode.reitti.service.processing.LocationDataCleanupTask; +import com.dedicatedcode.reitti.service.processing.LocationPointStagingService; +import com.dedicatedcode.reitti.service.processing.LocationPointStagingService.PromotionResult; +import com.dedicatedcode.reitti.service.processing.TimeRange; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.quartz.JobDataMap; +import org.quartz.JobDetail; +import org.quartz.JobExecutionContext; +import org.quartz.JobExecutionException; + +import java.time.Instant; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.*; + +/** + * Documents the promotion/cleanup hand-off fixed for #1212. + *

+ * promote() returns the exact time range of the rows it actually inserted, so the cleanup job + * is always scheduled with the range of the data that was just promoted - independent of + * duplicate queued promotion jobs or points arriving from the ingest thread while the + * promotion runs. In v5.3.0 the range was looked up in the staging table before promoting, + * which could return null (NPE at TimeRange.start()) or a too-wide range. + */ +@ExtendWith(MockitoExtension.class) +class PromotionJobHandlerTest { + + private static final Instant FLUSH_START = Instant.parse("2026-08-27T22:59:55Z"); + private static final Instant FLUSH_END = Instant.parse("2026-08-27T23:00:00Z"); + private static final String PARTITION_KEY = "stream_2_3_20260827"; + private static final UUID JOB_ID = UUID.randomUUID(); + + @Mock + private LocationPointStagingService stagingService; + @Mock + private JobSchedulingService jobSchedulingService; + @Mock + private JobMetadataRepository metadataRepository; + @Mock + private UserNotificationService userNotificationService; + @Mock + private JobDetail locationDataCleanupTask; + @Mock + private JobDetail liveModeOnlyUpdateTask; + @Mock + private JobDetail promotionTask; + @Mock + private JobExecutionContext context; + + private final PromotionInflightGuard promotionInflightGuard = new PromotionInflightGuard(); + + private PromotionJobHandler handler; + + private final User user = new User(1L, "tester", null, "Tester", null, null, Role.USER, UserType.NORMAL, 0L); + private final Device device = new Device(3L, "phone", true, true, true, "#ffffff", true, Instant.now(), Instant.now(), 0L); + + @BeforeEach + void setUp() { + this.handler = new PromotionJobHandler(stagingService, + jobSchedulingService, + metadataRepository, + userNotificationService, + promotionInflightGuard, + locationDataCleanupTask, + liveModeOnlyUpdateTask, + promotionTask); + } + + @Test + void shouldScheduleCleanupWithActuallyPromotedRange() throws JobExecutionException { + when(stagingService.promote(user, PARTITION_KEY)).thenReturn(new PromotionResult(5, new TimeRange(FLUSH_START, FLUSH_END))); + when(stagingService.hasUnpromotedPoints(PARTITION_KEY)).thenReturn(false); + + runHandler(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(LocationDataCleanupTask.TaskData.class); + verify(jobSchedulingService).enqueueTask(eq(locationDataCleanupTask), captor.capture(), any()); + assertEquals(FLUSH_START, captor.getValue().getStart()); + assertEquals(FLUSH_END, captor.getValue().getEnd()); + verify(userNotificationService).newLocationData(eq(user), eq(device), any(TimeRange.class)); + verify(jobSchedulingService, never()).enqueueTask(eq(promotionTask), any(), any()); + } + + @Test + void shouldNotScheduleCleanupWhenNothingWasPromoted() throws JobExecutionException { + when(stagingService.promote(user, PARTITION_KEY)).thenReturn(new PromotionResult(0, TimeRange.empty())); + when(stagingService.hasUnpromotedPoints(PARTITION_KEY)).thenReturn(false); + + runHandler(); + + verify(jobSchedulingService, never()).enqueueTask(any(), any(), any()); + verify(userNotificationService, never()).newLocationData(any(), any(), any()); + } + + @Test + void shouldReDrivePromotionWhenPointsRemainUnpromoted() throws JobExecutionException { + // points were flushed by the ingest thread while this promotion was running + when(stagingService.promote(user, PARTITION_KEY)).thenReturn(new PromotionResult(0, TimeRange.empty())); + when(stagingService.hasUnpromotedPoints(PARTITION_KEY)).thenReturn(true); + + runHandler(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(PromotionJobHandler.TaskData.class); + verify(jobSchedulingService).enqueueTask(eq(promotionTask), captor.capture(), any()); + assertEquals(PARTITION_KEY, captor.getValue().getPartitionKey()); + verify(jobSchedulingService, never()).enqueueTask(eq(locationDataCleanupTask), any(), any()); + } + + @Test + void shouldReleaseInflightGuardAfterExecution() throws JobExecutionException { + // the batching service acquired the guard before enqueueing this promotion + assertTrue(promotionInflightGuard.tryAcquire(PARTITION_KEY)); + when(stagingService.promote(user, PARTITION_KEY)).thenReturn(new PromotionResult(0, TimeRange.empty())); + when(stagingService.hasUnpromotedPoints(PARTITION_KEY)).thenReturn(false); + + runHandler(); + + // the guard was released, so the next flush may enqueue a promotion again + assertTrue(promotionInflightGuard.tryAcquire(PARTITION_KEY)); + } + + private void runHandler() throws JobExecutionException { + JobDataMap dataMap = new JobDataMap(); + dataMap.put("data", new PromotionJobHandler.TaskData(user, device, PARTITION_KEY, false, JOB_ID, null)); + when(context.getMergedJobDataMap()).thenReturn(dataMap); + handler.execute(context); + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/jobs/JobMetadataCleanupServiceTest.java b/src/test/java/com/dedicatedcode/reitti/service/jobs/JobMetadataCleanupServiceTest.java new file mode 100644 index 00000000..b7ff14cc --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/jobs/JobMetadataCleanupServiceTest.java @@ -0,0 +1,61 @@ +package com.dedicatedcode.reitti.service.jobs; + +import com.dedicatedcode.reitti.IntegrationTest; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the #1212 orphan-trigger fix: the metadata cleanup must not delete the metadata of + * jobs whose Quartz triggers are still pending. Deleting that metadata left live triggers + * behind without any tracking row (8,683 orphaned promotion-job triggers in the reported + * instance), which let qrtz_triggers grow unbounded. + */ +@IntegrationTest +class JobMetadataCleanupServiceTest { + + @Autowired + private JobMetadataCleanupService cleanupService; + @Autowired + private JdbcTemplate jdbcTemplate; + + @Test + void keepsMetadataOfPendingJobsAndDeletesTerminalOnes() { + UUID pendingId = insertJob("AWAITING", Instant.now().minus(48, java.time.temporal.ChronoUnit.HOURS)); + UUID runningId = insertJob("RUNNING", Instant.now().minus(48, java.time.temporal.ChronoUnit.HOURS)); + UUID completedId = insertJob("COMPLETED", Instant.now().minus(48, java.time.temporal.ChronoUnit.HOURS)); + UUID failedId = insertJob("FAILED", Instant.now().minus(48, java.time.temporal.ChronoUnit.HOURS)); + UUID recentPendingId = insertJob("AWAITING", Instant.now().minus(1, java.time.temporal.ChronoUnit.HOURS)); + + cleanupService.cleanUpOldJobs(); + + assertTrue(exists(pendingId), "pending trigger would become an orphan if its metadata was deleted"); + assertTrue(exists(runningId)); + assertTrue(exists(recentPendingId)); + assertFalse(exists(completedId)); + assertFalse(exists(failedId)); + } + + private UUID insertJob(String status, Instant enqueuedAt) { + UUID id = UUID.randomUUID(); + jdbcTemplate.update(""" + INSERT INTO job_meta_data (id, user_id, task_id, type, friendly_name, status, enqueued_at, scheduled_at, parent_job_id, created_at, updated_at) + VALUES (?, null, 'promotion-job', 'GPS_INGESTION', 'JobMetadataCleanupServiceTest', ?, ?, ?, null, NOW(), NOW()) + """, id, status, Timestamp.from(enqueuedAt), Timestamp.from(enqueuedAt)); + return id; + } + + private boolean exists(UUID id) { + List result = jdbcTemplate.queryForList("SELECT status FROM job_meta_data WHERE id = ?", String.class, id); + return !result.isEmpty(); + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingServiceTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingServiceTest.java new file mode 100644 index 00000000..843afc1b --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/LocationPointStagingServiceTest.java @@ -0,0 +1,121 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.IntegrationTest; +import com.dedicatedcode.reitti.TestingService; +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.service.processing.LocationPointStagingService.PromotionResult; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies the #1212 fix: promote() returns the exact time range of the rows it actually + * inserted. A promotion that races with concurrent ingest (or a duplicate queued promotion + * job that finds no unpromoted points) therefore schedules its cleanup job with the range of + * the few new points only - not with the whole partition range as the previous fallback + * (getWholeTimeRange) did, which re-opened the whole day for reprocessing. + */ +@IntegrationTest +class LocationPointStagingServiceTest { + + private static final Instant MORNING_START = Instant.parse("2026-08-27T06:00:00Z"); + private static final Instant MORNING_END = Instant.parse("2026-08-27T06:05:00Z"); + private static final Instant EVENING_FLUSH = Instant.parse("2026-08-27T23:59:55Z"); + + @Autowired + private LocationPointStagingService stagingService; + @Autowired + private TestingService testingService; + @Autowired + private JdbcTemplate jdbcTemplate; + + @Test + void promotedRangeCoversOnlyTheActuallyInsertedRows() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + String partitionKey = "stream_" + user.getId() + "_" + device.id() + "_20260827"; + + stagingService.ensurePartitionExists(partitionKey); + + // first flush of the day, promoted in the morning + stagingService.insertBatch(partitionKey, user, device, points(MORNING_START, MORNING_END, Duration.ofSeconds(30))); + PromotionResult firstPromotion = stagingService.promote(user, partitionKey); + assertEquals(11, firstPromotion.promotedCount()); + assertEquals(MORNING_START, firstPromotion.promotedRange().start()); + assertEquals(MORNING_END, firstPromotion.promotedRange().end()); + + // 18 hours later a duplicate promotion job for the same partition runs: nothing to do + assertFalse(stagingService.hasUnpromotedPoints(partitionKey)); + PromotionResult emptyPromotion = stagingService.promote(user, partitionKey); + assertEquals(0, emptyPromotion.promotedCount()); + assertFalse(emptyPromotion.hasPromoted()); + + // ... meanwhile the ingest thread stages a single new point + stagingService.insertBatch(partitionKey, user, device, List.of(point(EVENING_FLUSH))); + assertTrue(stagingService.hasUnpromotedPoints(partitionKey)); + + // the promotion picks it up and returns exactly its range - not the whole partition + PromotionResult result = stagingService.promote(user, partitionKey); + assertEquals(1, result.promotedCount()); + assertEquals(EVENING_FLUSH, result.promotedRange().start()); + assertEquals(EVENING_FLUSH, result.promotedRange().end()); + assertFalse(stagingService.hasUnpromotedPoints(partitionKey)); + + // and the promoted rows are still sitting in staging - they are never deleted in live mode + String tableName = "staged_" + partitionKey.toLowerCase().replace("-", "_").replace(".", "_"); + Long rowsInStaging = jdbcTemplate.queryForObject("SELECT count(*) FROM " + tableName + " WHERE partition_key = ?", Long.class, partitionKey); + assertEquals(12, rowsInStaging); + } + + @Test + void concurrentFlushDuringPromotionIsNotLost() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + String partitionKey = "stream_" + user.getId() + "_" + device.id() + "_20260827"; + + stagingService.ensurePartitionExists(partitionKey); + + // the promotion claims and inserts all unpromoted rows atomically ... + stagingService.insertBatch(partitionKey, user, device, List.of(point(MORNING_START))); + PromotionResult first = stagingService.promote(user, partitionKey); + assertEquals(1, first.promotedCount()); + + // ... so a point inserted afterwards stays unpromoted and is picked up by the next + // promotion instead of being marked promoted without ever reaching raw_source_points + stagingService.insertBatch(partitionKey, user, device, List.of(point(EVENING_FLUSH))); + assertTrue(stagingService.hasUnpromotedPoints(partitionKey)); + + PromotionResult second = stagingService.promote(user, partitionKey); + assertEquals(1, second.promotedCount()); + assertEquals(EVENING_FLUSH, second.promotedRange().start()); + assertFalse(stagingService.hasUnpromotedPoints(partitionKey)); + } + + private List points(Instant start, Instant end, Duration step) { + List result = new ArrayList<>(); + for (Instant t = start; !t.isAfter(end); t = t.plus(step)) { + result.add(point(t)); + } + return result; + } + + private LocationPoint point(Instant timestamp) { + LocationPoint locationPoint = new LocationPoint(); + locationPoint.setTimestamp(timestamp); + locationPoint.setLatitude(53.551086); + locationPoint.setLongitude(9.993682); + locationPoint.setAccuracyMeters(10.0); + return locationPoint; + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java new file mode 100644 index 00000000..99b66891 --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java @@ -0,0 +1,159 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.IntegrationTest; +import com.dedicatedcode.reitti.TestingService; +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.RawLocationPointJdbcService; +import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; +import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; + +/** + * Demonstrates the throughput problem behind #1212 ("promotion advances at ~1/8 real-time"). + *

+ * Every promotion - no matter how small the flush - schedules a LocationDataCleanupTask. + * That task widens the range by the anomaly lookback (reitti.geo-point-filter.history-lookback-hours, + * default 24h) on both sides and hands that range to UpdateCuratedTimelineTask, which deletes + * every raw_location_points row in the window (dropForReSeeding) and re-inserts it as + * processed = false (updateFromDevices). The pipeline then re-detects visits/trips for all of them. + *

+ * So promoting a few seconds of new data re-opens ~48h of already processed history. With the + * default batching config a promotion job is enqueued every ~5s per stream, the chain + * (promotion -> cleanup -> reseed -> pipeline) can never keep up with real time, jobs pile up + * faster than they drain and the whole Quartz machinery saturates. + */ +@IntegrationTest +class PromotionCleanupChurnTest { + + private static final Instant HISTORY_START = Instant.parse("2026-08-25T00:00:00Z"); + private static final Instant HISTORY_END = Instant.parse("2026-08-26T23:55:00Z"); + private static final UUID JOB_ID = UUID.randomUUID(); + + @Autowired + private LocationDataCleanupTask locationDataCleanupTask; + @Autowired + private SourceLocationPointJdbcService sourceLocationPointJdbcService; + @Autowired + private RawLocationPointJdbcService rawLocationPointJdbcService; + @Autowired + private TestingService testingService; + @Autowired + private JdbcTemplate jdbcTemplate; + + @MockitoBean + private JobSchedulingService jobSchedulingService; + + @Test + void singlePromotedPointReopensTwoDaysOfAlreadyProcessedTimeline() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + + // The state as promotion sees it: two days of points already promoted into raw_source_points + List history = points(HISTORY_START, HISTORY_END, Duration.ofMinutes(5)); + assertEquals(576, sourceLocationPointJdbcService.bulkInsert(user, device, history)); + + // --- promotion cycle 1: the flush [00:00, 00:05] was promoted --- + Instant flushStart = HISTORY_START; + Instant flushEnd = HISTORY_START.plus(Duration.ofMinutes(5)); + TimeRange reseedRange1 = runCleanupAndCaptureReseedRange(user, device, flushStart, flushEnd); + + // LocationDataCleanupTask hands the ±24h lookback window to the timeline update, + // not the 5 minutes that were actually promoted + assertEquals(Duration.ofHours(24), Duration.between(reseedRange1.start(), flushStart)); + assertTrue(Duration.between(reseedRange1.start(), reseedRange1.end()).compareTo(Duration.ofHours(48)) >= 0); + + reseedFromView(user, reseedRange1); + + // 289 timeline points were seeded although the flush contained only 2 points + assertEquals(289, timelineCount(user)); + assertEquals(289, unprocessedCount(user)); + + // the pipeline finished - all points are processed now + markAllProcessed(user); + assertEquals(289, processedCount(user)); + + // --- promotion cycle 2: a single new point arrives --- + Instant newPoint = HISTORY_END.plus(Duration.ofMinutes(1)); + assertEquals(1, sourceLocationPointJdbcService.bulkInsert(user, device, List.of(point(newPoint)))); + + reset(jobSchedulingService); + TimeRange reseedRange2 = runCleanupAndCaptureReseedRange(user, device, newPoint, newPoint); + reseedFromView(user, reseedRange2); + + // the single new point wiped and re-opened the whole lookback window: + // 288 already processed points are unprocessed again and will be re-detected + assertEquals(288, processedCount(user)); + assertEquals(289, unprocessedCount(user), "one new point was promoted, but the whole window was re-opened"); + assertEquals(577, timelineCount(user)); + } + + private TimeRange runCleanupAndCaptureReseedRange(User user, Device device, Instant start, Instant end) { + locationDataCleanupTask.execute(new LocationDataCleanupTask.TaskData(user, device, start, end, JOB_ID, null)); + + ArgumentCaptor captor = ArgumentCaptor.forClass(UpdateCuratedTimelineTask.TaskData.class); + verify(jobSchedulingService).enqueueTask(any(), captor.capture(), any()); + return (TimeRange) ReflectionTestUtils.getField(captor.getValue(), "timeRange"); + } + + /** + * Steps 1-2 of UpdateCuratedTimelineTask.execute(). fillGaps() (step 3) and the + * processing pipeline (step 4) are skipped to keep this test deterministic - they only + * add more work on top. + */ + private void reseedFromView(User user, TimeRange reseedRange) { + rawLocationPointJdbcService.dropForReSeeding(user, reseedRange); + rawLocationPointJdbcService.updateFromDevices(user, reseedRange); + } + + private void markAllProcessed(User user) { + jdbcTemplate.update("UPDATE raw_location_points SET processed = true WHERE user_id = ?", user.getId()); + } + + private long timelineCount(User user) { + return jdbcTemplate.queryForObject("SELECT count(*) FROM raw_location_points WHERE user_id = ?", Long.class, user.getId()); + } + + private long processedCount(User user) { + return jdbcTemplate.queryForObject("SELECT count(*) FROM raw_location_points WHERE user_id = ? AND processed", Long.class, user.getId()); + } + + private long unprocessedCount(User user) { + return jdbcTemplate.queryForObject("SELECT count(*) FROM raw_location_points WHERE user_id = ? AND NOT processed", Long.class, user.getId()); + } + + private List points(Instant start, Instant end, Duration step) { + List result = new ArrayList<>(); + for (Instant t = start; !t.isAfter(end); t = t.plus(step)) { + result.add(point(t)); + } + return result; + } + + private LocationPoint point(Instant timestamp) { + LocationPoint locationPoint = new LocationPoint(); + locationPoint.setTimestamp(timestamp); + locationPoint.setLatitude(53.551086); + locationPoint.setLongitude(9.993682); + locationPoint.setAccuracyMeters(10.0); + return locationPoint; + } +} From effb106b16264c390a9c1c838bb86af4099aa75b Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Sat, 29 Aug 2026 13:00:20 +0200 Subject: [PATCH 5/6] bugfix(#1212) Add tests for gap handling, anomaly processing, and promotion fixes - Add `DataGapProcessingTest` to verify outage gap bridging. - Add `PipelineVisitDetectionScopeTest` to validate margin-based visit detection. - Add `ProcessingWindowResolverTest` for time range resolution scenarios. - Update `SyntheticPointInserterTest` for dynamic and accurate time ranges. - Update `AnomalyProcessingService` to rely on context-based time ranges. - Optimize promotion cleanup and re-seeding logic to prevent unnecessary timeline re-processing. - Add SQL migration to clean stale Quartz triggers and orphan metadata. --- .../SourceLocationPointJdbcService.java | 46 ++++++- .../processing/AnomalyProcessingService.java | 19 ++- .../processing/ExcessDensityHandler.java | 22 +--- .../GeoPointAnomalyFilterConfig.java | 7 -- .../processing/LocationDataCleanupTask.java | 10 +- .../processing/ProcessingWindowResolver.java | 63 ++++++++++ .../processing/SyntheticPointInserter.java | 28 ++--- .../UnifiedLocationProcessingService.java | 20 ++- .../service/workbench/WorkbenchService.java | 2 +- src/main/resources/application.properties | 1 - .../V124__cleanup_stale_quartz_triggers.sql | 50 ++++++++ .../AnomalyProcessingServiceTest.java | 7 +- .../processing/DataGapProcessingTest.java | 117 ++++++++++++++++++ .../PipelineVisitDetectionScopeTest.java | 73 +++++++++++ .../ProcessingWindowResolverTest.java | 117 ++++++++++++++++++ .../processing/PromotionCleanupChurnTest.java | 47 ++++--- .../SyntheticPointInserterTest.java | 10 +- 17 files changed, 541 insertions(+), 98 deletions(-) create mode 100644 src/main/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolver.java create mode 100644 src/main/resources/db/migration/V124__cleanup_stale_quartz_triggers.sql create mode 100644 src/test/java/com/dedicatedcode/reitti/service/processing/DataGapProcessingTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/processing/PipelineVisitDetectionScopeTest.java create mode 100644 src/test/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolverTest.java diff --git a/src/main/java/com/dedicatedcode/reitti/repository/SourceLocationPointJdbcService.java b/src/main/java/com/dedicatedcode/reitti/repository/SourceLocationPointJdbcService.java index c2c2a33d..951724ce 100644 --- a/src/main/java/com/dedicatedcode/reitti/repository/SourceLocationPointJdbcService.java +++ b/src/main/java/com/dedicatedcode/reitti/repository/SourceLocationPointJdbcService.java @@ -78,6 +78,7 @@ public long countByUserAndTimestampBetween(User user, Device device, Instant sta if (!includeIgnored) { sql.append("AND rlp.status = 0 "); } + if (!includeInvalid) { sql.append("AND rlp.invalid = false "); } @@ -89,6 +90,46 @@ public List findByUserAndTimestampBetweenOrderByTimestampAs return findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, startTime, endTime, includeIgnored, includeInvalid, NO_PAGING, NO_PAGING); } + /** + * @return the timestamp of the n-th valid (non-ignored) point before the given instant, + * or the oldest point before it if fewer than n exist. Empty if no point exists before. + */ + public Optional findNthPointTimestampBefore(User user, Device device, Instant when, int n) { + String sql = """ + SELECT MIN(timestamp) FROM ( + SELECT timestamp FROM raw_source_points + WHERE user_id = ? AND device_id IS NOT DISTINCT FROM ? AND status = 0 AND timestamp < ? + ORDER BY timestamp DESC + LIMIT ? + ) context_points + """; + List result = jdbcTemplate.query(sql, (rs, _) -> { + Timestamp ts = rs.getTimestamp(1); + return ts == null ? null : ts.toInstant(); + }, user.getId(), device != null ? device.id() : null, Timestamp.from(when), n); + return result.stream().filter(java.util.Objects::nonNull).findFirst(); + } + + /** + * @return the timestamp of the n-th valid (non-ignored) point at or after the given instant, + * or the newest point from there on if fewer than n exist. Empty if no point exists after. + */ + public Optional findNthPointTimestampAfter(User user, Device device, Instant when, int n) { + String sql = """ + SELECT MAX(timestamp) FROM ( + SELECT timestamp FROM raw_source_points + WHERE user_id = ? AND device_id IS NOT DISTINCT FROM ? AND status = 0 AND timestamp > ? + ORDER BY timestamp ASC + LIMIT ? + ) context_points + """; + List result = jdbcTemplate.query(sql, (rs, _) -> { + Timestamp ts = rs.getTimestamp(1); + return ts == null ? null : ts.toInstant(); + }, user.getId(), device != null ? device.id() : null, Timestamp.from(when), n); + return result.stream().filter(java.util.Objects::nonNull).findFirst(); + } + public SourceLocationPoint create(User user, Device device, SourceLocationPoint rawLocationPoint) { String sql = "INSERT INTO raw_source_points (user_id, device_id, timestamp, accuracy_meters, elevation_meters, geom, invalid, status) " + "VALUES (?, ?, ?, ?, ?, ST_GeomFromText(?, '4326'), ?, ?) ON CONFLICT DO NOTHING RETURNING id"; @@ -181,11 +222,6 @@ private void updateBulkStatus(User user, List pointIds, SourceLocationPoin jdbcTemplate.batchUpdate(sql, batchArgs); } - - public int updateLocation(User user, Long id, double lat, double lng) { - return updateLocation(user, id, lat, lng, null); - } - public int updateLocation(User user, Long id, double lat, double lng, Long h3Cell) { if (h3Cell != null) { return this.jdbcTemplate.update("UPDATE raw_source_points SET geom = CAST(? AS geometry), h3_cell = ? WHERE id = ? AND user_id = ?", diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java index d7ff13ed..cc4d6438 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java @@ -10,33 +10,32 @@ import org.springframework.stereotype.Service; import java.time.Instant; -import java.time.temporal.ChronoUnit; import java.util.List; +/** + * Re-evaluates the anomaly status of the source points within the given time range and marks + * outliers as invalid. The caller is responsible for passing a range that includes enough + * neighbor context for the boundary points (see {@link ProcessingWindowResolver}). + */ @Service public class AnomalyProcessingService { private static final Logger logger = LoggerFactory.getLogger(AnomalyProcessingService.class); private final GeoPointAnomalyFilter detector; - private final GeoPointAnomalyFilterConfig config; private final SourceLocationPointJdbcService repository; - public AnomalyProcessingService(GeoPointAnomalyFilter geoPointAnomalyFilter, GeoPointAnomalyFilterConfig config, SourceLocationPointJdbcService repository) { + public AnomalyProcessingService(GeoPointAnomalyFilter geoPointAnomalyFilter, SourceLocationPointJdbcService repository) { this.detector = geoPointAnomalyFilter; - this.config = config; this.repository = repository; } public TimeRange processAndMarkAnomalies(User user, Device device, Instant start, Instant end) { - Instant startTime = start.minus(config.getHistoryLookback(), ChronoUnit.HOURS); - Instant endTime = end.plus(config.getHistoryLookback(), ChronoUnit.HOURS); - - repository.resetInvalidStatus(user, startTime, endTime); - List pointsToCheck = repository.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, startTime, endTime, false, true); + repository.resetInvalidStatus(user, start, end); + List pointsToCheck = repository.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, start, end, false, true); logger.debug("Found {} points to check for user {}", pointsToCheck.size(), user.getUsername()); List anomalousPoints = detector.detectAnomalies(pointsToCheck); repository.bulkUpdateInvalidStatus(anomalousPoints); logger.info("Marked {} points as invalid for user {}", anomalousPoints.size(), user.getUsername()); - return new TimeRange(startTime, endTime); + return new TimeRange(start, end); } } diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java b/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java index 33010827..1f349631 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java @@ -3,10 +3,8 @@ import com.dedicatedcode.reitti.config.LocationDensityConfig; import com.dedicatedcode.reitti.model.devices.Device; import com.dedicatedcode.reitti.model.geo.SourceLocationPoint; -import com.dedicatedcode.reitti.model.processing.DetectionParameter; import com.dedicatedcode.reitti.model.security.User; import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; -import com.dedicatedcode.reitti.service.VisitDetectionParametersService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Service; @@ -20,28 +18,18 @@ public class ExcessDensityHandler { private static final Logger logger = LoggerFactory.getLogger(ExcessDensityHandler.class); private final LocationDensityConfig config; - private final VisitDetectionParametersService visitDetectionParametersService; private final SourceLocationPointJdbcService rawLocationPointService; - public ExcessDensityHandler(LocationDensityConfig config, VisitDetectionParametersService visitDetectionParametersService, + public ExcessDensityHandler(LocationDensityConfig config, SourceLocationPointJdbcService rawLocationPointService) { this.config = config; - this.visitDetectionParametersService = visitDetectionParametersService; this.rawLocationPointService = rawLocationPointService; } public TimeRange handleExcess(User user, Device device, TimeRange inputRange) { - DetectionParameter detectionParams = visitDetectionParametersService.getCurrentConfiguration(user, inputRange.start()); - DetectionParameter.LocationDensity densityConfig = detectionParams.getLocationDensity(); - - // Step 2: Expand the time range by the interpolation window to catch boundary gaps - long maxInterpolationGapMinutes = densityConfig.getMaxInterpolationGapMinutes(); - Duration window = Duration.ofMinutes(maxInterpolationGapMinutes); - TimeRange expandedRange = new TimeRange( - inputRange.start().minus(window), - inputRange.end().plus(window) - ); - List points = rawLocationPointService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, expandedRange.start(), expandedRange.end(), false, true); + // The caller provides a range that already includes the boundary context (see + // ProcessingWindowResolver), so adjacent points on both range edges are covered. + List points = rawLocationPointService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, inputRange.start(), inputRange.end(), false, true); if (points.size() < 2) { return TimeRange.empty(); } @@ -74,7 +62,7 @@ public TimeRange handleExcess(User user, Device device, TimeRange inputRange) { rawLocationPointService.bulkUpdateIgnoredStatus(user, new ArrayList<>(pointsToIgnore)); logger.debug("Marked {} points as ignored for user {}", pointsToIgnore.size(), user.getUsername()); } - return expandedRange; + return inputRange; } // The selection logic is unchanged from the original, kept here for completeness. diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/GeoPointAnomalyFilterConfig.java b/src/main/java/com/dedicatedcode/reitti/service/processing/GeoPointAnomalyFilterConfig.java index 9a29dc46..d2cf665d 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/GeoPointAnomalyFilterConfig.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/GeoPointAnomalyFilterConfig.java @@ -9,17 +9,14 @@ public class GeoPointAnomalyFilterConfig { private final double maxSpeedKmh; private final double maxAccuracyMeters; - private final int historyLookback; private final int windowSize; public GeoPointAnomalyFilterConfig( @Value("${reitti.geo-point-filter.max-speed-kmh:1000}") double maxSpeedKmh, @Value("${reitti.geo-point-filter.max-accuracy-meters:100}") double maxAccuracyMeters, - @Value("${reitti.geo-point-filter.history-lookback-hours:2}") int historyLookback, @Value("${reitti.geo-point-filter.window-size:5}") int windowSize) { this.maxSpeedKmh = maxSpeedKmh; this.maxAccuracyMeters = maxAccuracyMeters; - this.historyLookback = historyLookback; this.windowSize = windowSize; } @@ -31,10 +28,6 @@ public double getMaxAccuracyMeters() { return maxAccuracyMeters; } - public int getHistoryLookback() { - return historyLookback; - } - public int getWindowSize() { return windowSize; } diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java index 1186ffba..2c9cd4b5 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java @@ -24,6 +24,7 @@ public class LocationDataCleanupTask implements Job { private static final Logger log = LoggerFactory.getLogger(LocationDataCleanupTask.class); private final ExcessDensityHandler excessDensityHandler; private final AnomalyProcessingService anomalyProcessingService; + private final ProcessingWindowResolver processingWindowResolver; private final UserSettingsJdbcService userSettingsJdbcService; private final UserJdbcService userJdbcService; private final JobSchedulingService jobScheduler; @@ -32,6 +33,7 @@ public class LocationDataCleanupTask implements Job { public LocationDataCleanupTask(ExcessDensityHandler excessDensityHandler, AnomalyProcessingService anomalyProcessingService, + ProcessingWindowResolver processingWindowResolver, UserSettingsJdbcService userSettingsJdbcService, UserJdbcService userJdbcService, JobSchedulingService jobScheduler, @@ -39,6 +41,7 @@ public LocationDataCleanupTask(ExcessDensityHandler excessDensityHandler, JobMetadataRepository metadataRepository) { this.excessDensityHandler = excessDensityHandler; this.anomalyProcessingService = anomalyProcessingService; + this.processingWindowResolver = processingWindowResolver; this.userSettingsJdbcService = userSettingsJdbcService; this.userJdbcService = userJdbcService; this.jobScheduler = jobScheduler; @@ -61,9 +64,12 @@ public void execute(TaskData data) { Instant end = data.getEnd(); log.debug("Starting LocationDataCleanupJob for user [{}] and device [{}] between {} and {}", user, device, start, end); this.metadataRepository.updateProgress(jobId, 0,4, "Anomaly processing started ..."); - TimeRange processedTimeRange = anomalyProcessingService.processAndMarkAnomalies(user, device, start, end); + // resolve how far around the promoted range we actually have to process: context + // margin for the filters, closed gaps and backfill - instead of a static ±24h window + TimeRange window = processingWindowResolver.resolve(user, device, new TimeRange(start, end)); + TimeRange processedTimeRange = anomalyProcessingService.processAndMarkAnomalies(user, device, window.start(), window.end()); this.metadataRepository.updateProgress(jobId, 1,4, "Density normalization started ..."); - TimeRange densityTimeRange = excessDensityHandler.handleExcess(user, device, new TimeRange(start, end)); + TimeRange densityTimeRange = excessDensityHandler.handleExcess(user, device, window); this.metadataRepository.updateProgress(jobId, 2,4, "Update user data started ..."); this.userSettingsJdbcService.updateNewestData(user, end); this.userJdbcService.setLastDataModificationAt(user, Instant.now()); diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolver.java b/src/main/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolver.java new file mode 100644 index 00000000..f80d5d0c --- /dev/null +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolver.java @@ -0,0 +1,63 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.processing.DetectionParameter; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; +import com.dedicatedcode.reitti.service.VisitDetectionParametersService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.Optional; + +@Service +public class ProcessingWindowResolver { + + private static final Logger log = LoggerFactory.getLogger(ProcessingWindowResolver.class); + + private final SourceLocationPointJdbcService sourceLocationPointJdbcService; + private final VisitDetectionParametersService visitDetectionParametersService; + private final GeoPointAnomalyFilterConfig filterConfig; + + public ProcessingWindowResolver(SourceLocationPointJdbcService sourceLocationPointJdbcService, + VisitDetectionParametersService visitDetectionParametersService, + GeoPointAnomalyFilterConfig filterConfig) { + this.sourceLocationPointJdbcService = sourceLocationPointJdbcService; + this.visitDetectionParametersService = visitDetectionParametersService; + this.filterConfig = filterConfig; + } + + public TimeRange resolve(User user, Device device, TimeRange promotedRange) { + DetectionParameter params = visitDetectionParametersService.getCurrentConfiguration(user, promotedRange.start()); + Duration cap = Duration.ofMinutes(params.getLocationDensity().getMaxInterpolationGapMinutes()); + int contextPoints = this.filterConfig.getWindowSize(); + + Instant windowStart = promotedRange.start(); + Instant windowEnd = promotedRange.end(); + + Optional before = this.sourceLocationPointJdbcService + .findNthPointTimestampBefore(user, device, windowStart, contextPoints); + if (before.isPresent()) { + Duration back = Duration.between(before.get(), windowStart); + windowStart = back.compareTo(cap) > 0 ? windowStart.minus(cap) : before.get(); + } + + Optional after = this.sourceLocationPointJdbcService + .findNthPointTimestampAfter(user, device, windowEnd, contextPoints); + if (after.isPresent()) { + Duration forward = Duration.between(windowEnd, after.get()); + windowEnd = forward.compareTo(cap) > 0 ? windowEnd.plus(cap) : after.get(); + } + + if (!windowEnd.isAfter(promotedRange.end())) { + windowEnd = promotedRange.end().plus(1, ChronoUnit.MILLIS); + } + + log.debug("Resolved processing window [{}] for promoted range [{}]", new TimeRange(windowStart, windowEnd), promotedRange); + return new TimeRange(windowStart, windowEnd); + } +} diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java b/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java index bd1409f9..82dd61c0 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java @@ -46,6 +46,10 @@ public SyntheticPointInserter(LocationDensityConfig config, /** * Processes the given time range: deletes old synthetic points, then * inserts new synthetic points where real-point gaps are too large. + *

+ * The caller provides a range that already includes the boundary context (see + * {@link ProcessingWindowResolver}), so gaps closing at the range edges are bridged as + * long as they do not exceed the configured interpolation limit. * * @param user the owning user * @param inputRange the time range that covers the newly arrived points @@ -56,38 +60,30 @@ public void fillGaps(User user, TimeRange inputRange) { user, inputRange.start()); DetectionParameter.LocationDensity densityConfig = detectionParams.getLocationDensity(); - // 2. Expand range to catch boundary gaps - long maxInterpolationGapMinutes = densityConfig.getMaxInterpolationGapMinutes(); - Duration window = Duration.ofMinutes(maxInterpolationGapMinutes); - TimeRange expandedRange = new TimeRange( - inputRange.start().minus(window), - inputRange.end().plus(window) - ); + // 2. Delete all existing synthetic points in the range + rawLocationPointService.deleteSyntheticPointsInRange(user, inputRange.start(), inputRange.end()); - // 3. Delete all existing synthetic points in the expanded range - rawLocationPointService.deleteSyntheticPointsInRange(user, expandedRange.start(), expandedRange.end()); - - Instant currentStart = expandedRange.start(); - while (currentStart.isBefore(expandedRange.end())) { - // 4. Fetch all real points in the expanded range + Instant currentStart = inputRange.start(); + while (currentStart.isBefore(inputRange.end())) { + // 3. Fetch all real points in the range List realPoints = rawLocationPointService .findByUserAndTimestampBetweenOrderByTimestampAsc( user, currentStart, - expandedRange.end(), + inputRange.end(), false, 0, maxBatchSize ); - // 5. Sort deterministically (same logic as original) + // 4. Sort deterministically (same logic as original) realPoints.sort(Comparator .comparing(RawLocationPoint::getTimestamp) .thenComparing(p -> p.getGeom().latitude()) .thenComparing(p -> p.getGeom().longitude()) .thenComparing(RawLocationPoint::isSynthetic)); - // 6. Process gaps + // 5. Process gaps processGaps(user, realPoints, densityConfig); if (realPoints.isEmpty()) break; currentStart = realPoints.getLast().getTimestamp().plus(1, ChronoUnit.MILLIS); diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java index 979c0135..b43ead68 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java @@ -231,16 +231,24 @@ private VisitDetectionResult detectVisits(User user, LocationProcessEvent event) long start = System.currentTimeMillis(); String previewId = event.getPreviewId(); - Instant windowStart = event.getEarliest().minus(1, ChronoUnit.DAYS); - Instant windowEnd = event.getLatest().plus(1, ChronoUnit.DAYS); DetectionParameter currentConfiguration; if (previewId == null) { - currentConfiguration = visitDetectionParametersService.getCurrentConfiguration(user, windowStart); + currentConfiguration = visitDetectionParametersService.getCurrentConfiguration(user, event.getEarliest()); } else { currentConfiguration = previewVisitDetectionParametersJdbcService.findCurrent(user, previewId); } + // The window only spans the batch plus enough margin for stays that continue across + // its edges (a stay becomes visible as processed visit and pulls the window out to + // its boundaries below). A static ±1 day window here made every pipeline run + // re-detect visits over days of already processed data (#1212). + Duration boundaryMargin = Duration.ofSeconds( + currentConfiguration.getVisitDetection().getMinimumStayTimeInSeconds() + + 2 * currentConfiguration.getVisitDetection().getMaxMergeTimeBetweenSameStayPoints()); + Instant windowStart = event.getEarliest().minus(boundaryMargin); + Instant windowEnd = event.getLatest().plus(boundaryMargin); + List existingProcessedVisits; if (previewId == null) { existingProcessedVisits = processedVisitJdbcService @@ -254,8 +262,10 @@ private VisitDetectionResult detectVisits(User user, LocationProcessEvent event) if (existingProcessedVisits.getFirst().getStartTime().isBefore(windowStart)) { windowStart = existingProcessedVisits.getFirst().getStartTime(); } - if (existingProcessedVisits.getLast().getEndTime().isAfter(windowEnd)) { - windowEnd = existingProcessedVisits.getLast().getEndTime(); + if (!existingProcessedVisits.getLast().getEndTime().isBefore(windowEnd)) { + // the point queries load with an exclusive end, so the visit's defining last + // point must be covered by extending the window beyond the visit end + windowEnd = existingProcessedVisits.getLast().getEndTime().plus(1, ChronoUnit.MILLIS); } } diff --git a/src/main/java/com/dedicatedcode/reitti/service/workbench/WorkbenchService.java b/src/main/java/com/dedicatedcode/reitti/service/workbench/WorkbenchService.java index fa8253ad..49d3af75 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/workbench/WorkbenchService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/workbench/WorkbenchService.java @@ -102,7 +102,7 @@ private void handleMove(User user, EditStoreDto editStore, UUID parentJob) { this.spatialCoverageService.preMove(movedPoints); List affectedTimeRange = this.sourceLocationPointJdbcService.findAffectedTimeRange(user, movedPointIds); for (MovedPointDto movedPoint : movedPoints) { - long newH3Cell = this.spatialCoverageService.getLevelCellForPoint(movedPoint.getLat(), movedPoint.getLng(), 12); + Long newH3Cell = this.spatialCoverageService.getLevelCellForPoint(movedPoint.getLat(), movedPoint.getLng(), 12); this.sourceLocationPointJdbcService.updateLocation(user, movedPoint.getSourceId(), movedPoint.getLat(), movedPoint.getLng(), newH3Cell); } scheduleUpdateJob(user, parentJob, affectedTimeRange); diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 272f5b2b..eb311c0a 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -88,7 +88,6 @@ reitti.db-janitor.schedule=0 0 4 * * ? reitti.geo-point-filter.max-speed-kmh=1000 reitti.geo-point-filter.max-accuracy-meters=100 -reitti.geo-point-filter.history-lookback-hours=24 reitti.geo-point-filter.window-size=50 reitti.process-data.refresh-views.schedule=0 0 4 * * * diff --git a/src/main/resources/db/migration/V124__cleanup_stale_quartz_triggers.sql b/src/main/resources/db/migration/V124__cleanup_stale_quartz_triggers.sql new file mode 100644 index 00000000..ec6b8056 --- /dev/null +++ b/src/main/resources/db/migration/V124__cleanup_stale_quartz_triggers.sql @@ -0,0 +1,50 @@ +CREATE TEMP TABLE stale_triggers ON COMMIT DROP AS +SELECT t.trigger_name, t.trigger_group +FROM qrtz_triggers t +WHERE t.trigger_group = 'reitti-tasks' + AND ( + NOT EXISTS (SELECT 1 FROM job_meta_data jmd WHERE jmd.id::text = t.trigger_name) + OR EXISTS (SELECT 1 + FROM job_meta_data jmd + WHERE jmd.id::text = t.trigger_name + AND jmd.status IN ('COMPLETED', 'FAILED', 'CANCELLED')) + OR (position(':' IN t.trigger_name) > 0 + AND EXISTS (SELECT 1 + FROM job_meta_data jmd + WHERE jmd.id::text = split_part(t.trigger_name, ':', 1) + AND jmd.status IN ('COMPLETED', 'FAILED', 'CANCELLED'))) + ); + +DELETE FROM qrtz_fired_triggers f +USING stale_triggers s +WHERE f.trigger_name = s.trigger_name AND f.trigger_group = s.trigger_group; + +DELETE FROM qrtz_simple_triggers st +USING stale_triggers s +WHERE st.trigger_name = s.trigger_name AND st.trigger_group = s.trigger_group; + +DELETE FROM qrtz_simprop_triggers sp +USING stale_triggers s +WHERE sp.trigger_name = s.trigger_name AND sp.trigger_group = s.trigger_group; + +DELETE FROM qrtz_cron_triggers ct +USING stale_triggers s +WHERE ct.trigger_name = s.trigger_name AND ct.trigger_group = s.trigger_group; + +DELETE FROM qrtz_blob_triggers bt +USING stale_triggers s +WHERE bt.trigger_name = s.trigger_name AND bt.trigger_group = s.trigger_group; + +DELETE FROM qrtz_triggers t +USING stale_triggers s +WHERE t.trigger_name = s.trigger_name AND t.trigger_group = s.trigger_group; + +-- Dead job metadata: jobs that never started (AWAITING/CREATED) and have no trigger left +-- that could ever run them. These would linger forever since the periodic metadata cleanup +-- only removes metadata of jobs that reached a terminal state. +DELETE FROM job_meta_data jmd +WHERE jmd.status IN ('AWAITING', 'CREATED') + AND NOT EXISTS (SELECT 1 + FROM qrtz_triggers t + WHERE t.trigger_group = 'reitti-tasks' + AND (t.trigger_name = jmd.id::text OR t.trigger_name LIKE jmd.id::text || ':%')); diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingServiceTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingServiceTest.java index 6a077797..ca207dd8 100644 --- a/src/test/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingServiceTest.java +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingServiceTest.java @@ -51,7 +51,8 @@ void shouldFilterOutSinglePoint() { assertEquals(14, this.rawLocationPointJdbcService.bulkInsert(user, null, points)); - this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:26:24Z"), Instant.parse("2022-01-01T09:26:24Z")); + // the caller passes the range to process - including enough context for the boundary + this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:03:03Z"), Instant.parse("2022-01-01T09:26:25Z")); List storedPoints = this.rawLocationPointJdbcService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, null, Instant.parse("2022-01-01T09:03:03Z"), Instant.parse("2022-01-01T09:27:24Z"), true, true); assertEquals(14, storedPoints.size(), "Point should be stored"); @@ -83,7 +84,7 @@ void shouldFilterOutSinglePointButRetainOne() { assertEquals(15, this.rawLocationPointJdbcService.bulkInsert(user, null, points)); - this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:26:24Z"), Instant.parse("2022-01-01T09:26:24Z")); + this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:03:03Z"), Instant.parse("2022-01-01T09:26:49Z")); List storedPoints = this.rawLocationPointJdbcService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, null, Instant.parse("2022-01-01T09:03:03Z"), Instant.parse("2022-01-01T09:27:24Z"), true, true); assertEquals(15, storedPoints.size(), "Point should be stored"); @@ -121,7 +122,7 @@ void shouldFilter() { //now store new points this.rawLocationPointJdbcService.bulkInsert(user, null, points); - this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:03:03Z"), Instant.parse("2022-01-01T09:17:59Z")); + this.anomalyFilter.processAndMarkAnomalies(user, null, Instant.parse("2022-01-01T09:02:24Z"), Instant.parse("2022-01-01T09:17:59Z")); storedPoints = this.rawLocationPointJdbcService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, null, Instant.parse("2022-01-01T09:00:00Z"), Instant.parse("2022-01-01T09:27:24Z"), true, true); diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/DataGapProcessingTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/DataGapProcessingTest.java new file mode 100644 index 00000000..7fb9d0e5 --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/DataGapProcessingTest.java @@ -0,0 +1,117 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.IntegrationTest; +import com.dedicatedcode.reitti.TestingService; +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; +import com.dedicatedcode.reitti.service.jobs.JobSchedulingService; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import java.sql.Timestamp; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.verify; + +@IntegrationTest +class DataGapProcessingTest { + + private static final Instant FIRST_POINT = Instant.parse("2026-08-25T10:00:00Z"); + private static final Instant LAST_POINT_BEFORE_OUTAGE = Instant.parse("2026-08-25T10:08:00Z"); + private static final Instant RESUME = Instant.parse("2026-08-25T14:00:00Z"); + private static final Instant LAST_POINT = Instant.parse("2026-08-25T14:08:00Z"); + + @Autowired + private LocationDataCleanupTask locationDataCleanupTask; + @Autowired + private UpdateCuratedTimelineTask updateCuratedTimelineTask; + @Autowired + private ProcessingPipelineTask processingPipelineTask; + @Autowired + private SourceLocationPointJdbcService sourceLocationPointJdbcService; + @Autowired + private TestingService testingService; + @Autowired + private JdbcTemplate jdbcTemplate; + + @MockitoBean + private JobSchedulingService jobSchedulingService; + + @Test + void resumedDataBridgesTheOutageGapAndYieldsOneVisitAcrossIt() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + + // 3 points before the outage and 3 after the resume, all at the same place so the + // gap is interpolatable (max interpolation distance 50m) + List points = new ArrayList<>(); + points.addAll(pointsAt(FIRST_POINT, 3)); + points.addAll(pointsAt(RESUME, 3)); + assertEquals(6, sourceLocationPointJdbcService.bulkInsert(user, device, points)); + + // the resume flush [14:00, 14:08] is promoted and cleaned up + locationDataCleanupTask.execute(new LocationDataCleanupTask.TaskData(user, device, RESUME, LAST_POINT, UUID.randomUUID(), null)); + + // real reseed + gap filling, exactly what UpdateCuratedTimelineTask does + updateCuratedTimelineTask.execute(capturedEnqueueOf(UpdateCuratedTimelineTask.TaskData.class)); + + Long syntheticBridges = jdbcTemplate.queryForObject( + "SELECT count(*) FROM raw_location_points WHERE user_id = ? AND synthetic AND timestamp > ? AND timestamp < ?", + Long.class, user.getId(), Timestamp.from(LAST_POINT_BEFORE_OUTAGE), Timestamp.from(RESUME)); + assertTrue(syntheticBridges > 500, "expected the outage gap to be bridged with synthetic points, got " + syntheticBridges); + + // the pipeline processes the reseeded and bridged timeline (second enqueue) + processingPipelineTask.execute(capturedEnqueueOf(ProcessingPipelineTask.TaskData.class)); + + Long visits = jdbcTemplate.queryForObject( + "SELECT count(*) FROM processed_visits WHERE user_id = ?", Long.class, user.getId()); + assertEquals(1, visits, "the whole span including the outage should be detected as one visit"); + + var visit = jdbcTemplate.queryForMap( + "SELECT start_time, end_time FROM processed_visits WHERE user_id = ?", user.getId()); + assertEquals(FIRST_POINT, ((Timestamp) visit.get("start_time")).toInstant()); + assertEquals(LAST_POINT, ((Timestamp) visit.get("end_time")).toInstant()); + } + + /** + * Extracts the single enqueued task of the wanted type from the mocked scheduler, + * ignoring unrelated enqueues (e.g. the async h3-indexing job during context startup). + */ + @SuppressWarnings("unchecked") + private > T capturedEnqueueOf(Class type) { + ArgumentCaptor captor = ArgumentCaptor.forClass(com.dedicatedcode.reitti.service.JobContext.class); + verify(jobSchedulingService, atLeast(1)).enqueueTask(any(), captor.capture(), any()); + List matches = captor.getAllValues().stream() + .filter(value -> value != null && type.isAssignableFrom(value.getClass())) + .map(value -> (T) value) + .toList(); + assertEquals(1, matches.size(), "expected exactly one enqueued " + type.getSimpleName() + ", got " + matches.size()); + return matches.getFirst(); + } + + private List pointsAt(Instant start, int count) { + List points = new ArrayList<>(); + for (int i = 0; i < count; i++) { + LocationPoint point = new LocationPoint(); + point.setTimestamp(start.plus(i * 4L, ChronoUnit.MINUTES)); + point.setLatitude(53.551086); + point.setLongitude(9.993682); + point.setAccuracyMeters(10.0); + points.add(point); + } + return points; + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/PipelineVisitDetectionScopeTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/PipelineVisitDetectionScopeTest.java new file mode 100644 index 00000000..a08b72d0 --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/PipelineVisitDetectionScopeTest.java @@ -0,0 +1,73 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.IntegrationTest; +import com.dedicatedcode.reitti.TestingService; +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.event.LocationProcessEvent; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.RawLocationPointJdbcService; +import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@IntegrationTest +class PipelineVisitDetectionScopeTest { + + private static final Instant T0 = Instant.parse("2026-08-25T00:00:00Z"); + + @Autowired + private UnifiedLocationProcessingService processingService; + @Autowired + private SourceLocationPointJdbcService sourceLocationPointJdbcService; + @Autowired + private RawLocationPointJdbcService rawLocationPointJdbcService; + @Autowired + private TestingService testingService; + @Autowired + private JdbcTemplate jdbcTemplate; + + @Test + void visitDetectionStaysWithinTheMarginAroundTheBatch() { + User user = testingService.randomUser(); + + // 4 hours of points at one place, 15s apart + List points = new ArrayList<>(); + for (int i = 0; i < 4 * 240; i++) { + LocationPoint point = new LocationPoint(); + point.setTimestamp(T0.plus(i * 15L, ChronoUnit.SECONDS)); + point.setLatitude(53.551086); + point.setLongitude(9.993682); + point.setAccuracyMeters(10.0); + points.add(point); + } + assertEquals(960, sourceLocationPointJdbcService.bulkInsert(user, testingService.findDefaultDevice(user), points)); + + // populate the timeline table from the source points, like UpdateCuratedTimelineTask does + rawLocationPointJdbcService.dropForReSeeding(user, TimeRange.of(T0, T0.plus(4, ChronoUnit.HOURS))); + rawLocationPointJdbcService.updateFromDevices(user, TimeRange.of(T0, T0.plus(4, ChronoUnit.HOURS))); + + // the pipeline batch only covers one minute in the middle of the stay + Instant batchStart = T0.plus(2, ChronoUnit.HOURS); + processingService.processLocationEvent(new LocationProcessEvent(user.getUsername(), batchStart, batchStart.plus(1, ChronoUnit.MINUTES), null, null, null)); + + var visit = jdbcTemplate.queryForMap("SELECT start_time, end_time FROM processed_visits WHERE user_id = ?", user.getId()); + Instant start = ((java.sql.Timestamp) visit.get("start_time")).toInstant(); + Instant end = ((java.sql.Timestamp) visit.get("end_time")).toInstant(); + + // margin = minStay (300s) + 2 * maxMerge (300s) = 15min: the detection window is + // [earliest - 15min, latest + 15min], not ±1 day around the batch + assertTrue(start.isAfter(batchStart.minus(16, ChronoUnit.MINUTES)), + "visit detection must not reach days before the batch, but started at " + start); + assertTrue(end.isBefore(batchStart.plus(17, ChronoUnit.MINUTES)), + "visit detection must not reach days after the batch, but ended at " + end); + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolverTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolverTest.java new file mode 100644 index 00000000..e0fdece4 --- /dev/null +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/ProcessingWindowResolverTest.java @@ -0,0 +1,117 @@ +package com.dedicatedcode.reitti.service.processing; + +import com.dedicatedcode.reitti.IntegrationTest; +import com.dedicatedcode.reitti.TestingService; +import com.dedicatedcode.reitti.dto.LocationPoint; +import com.dedicatedcode.reitti.model.devices.Device; +import com.dedicatedcode.reitti.model.security.User; +import com.dedicatedcode.reitti.repository.SourceLocationPointJdbcService; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; + +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@IntegrationTest +class ProcessingWindowResolverTest { + + private static final Instant T0 = Instant.parse("2026-08-25T00:00:00Z"); + private static final Duration STEP = Duration.ofMinutes(5); + + @Autowired + private ProcessingWindowResolver resolver; + @Autowired + private SourceLocationPointJdbcService sourceLocationPointJdbcService; + @Autowired + private TestingService testingService; + + @Test + void continuousStreamingGetsAContextMarginInsteadOfDays() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + // 100 points, 5 minutes apart: T0 .. T0+495min + insert(user, device, T0, 100); + + // the last flush of the stream is promoted + TimeRange window = resolver.resolve(user, device, TimeRange.of(T0.plus(490, java.time.temporal.ChronoUnit.MINUTES), T0.plus(495, java.time.temporal.ChronoUnit.MINUTES))); + + // 50 points of context backwards: index 48 .. 97 -> oldest at T0+240min + assertEquals(T0.plus(240, java.time.temporal.ChronoUnit.MINUTES), window.start()); + // nothing follows the promoted range, but the exclusive end must still cover the last point + assertEquals(T0.plus(495, java.time.temporal.ChronoUnit.MINUTES).plusMillis(1), window.end()); + } + + @Test + void closedGapIsBridgedByReachingBackToThePreOutagePoint() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + // phone reports until 10:04, then is offline for ~4h, resumes at 14:00 + insert(user, device, Instant.parse("2026-08-25T10:00:00Z"), 3); + Instant resume = Instant.parse("2026-08-25T14:00:00Z"); + insert(user, device, resume, 2); + + TimeRange window = resolver.resolve(user, device, TimeRange.of(resume, resume.plus(STEP))); + + // the window reaches back over the whole gap to the last pre-outage point, + // so the synthetic interpolation sees both gap endpoints + assertEquals(Instant.parse("2026-08-25T10:00:00Z"), window.start()); + assertEquals(resume.plus(STEP).plusMillis(1), window.end()); + } + + @Test + void firstPromotionHasNoBackwardExtension() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + insert(user, device, T0, 100); + + TimeRange window = resolver.resolve(user, device, TimeRange.of(T0, T0.plus(STEP))); + + assertEquals(T0, window.start()); + // backfill: 50 points follow the promoted range -> T0+10min .. T0+255min + assertEquals(T0.plus(255, java.time.temporal.ChronoUnit.MINUTES), window.end()); + } + + @Test + void sparseDataIsCappedAtMaxInterpolationGap() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + insert(user, device, Instant.parse("2026-08-23T10:00:00Z"), 1); + Instant resume = Instant.parse("2026-08-24T11:00:00Z"); // 25h after the single pre-outage point + insert(user, device, resume, 2); + + TimeRange window = resolver.resolve(user, device, TimeRange.of(resume, resume.plus(STEP))); + + // the context point lies 25h back; the widening is capped at maxInterpolationGapMinutes (24h) + // so a longer outage is never partially bridged + assertEquals(resume.minus(Duration.ofHours(24)), window.start()); + } + + @Test + void noPriorPointsKeepsThePromotedStart() { + User user = testingService.randomUser(); + Device device = testingService.findDefaultDevice(user); + TimeRange window = resolver.resolve(user, device, TimeRange.of(T0, T0.plus(STEP))); + + assertEquals(T0, window.start()); + assertEquals(T0.plus(STEP).plusMillis(1), window.end()); + } + + private void insert(User user, Device device, Instant start, int count) { + List points = new ArrayList<>(); + for (int i = 0; i < count; i++) { + LocationPoint point = new LocationPoint(); + point.setTimestamp(start.plus(STEP.multipliedBy(i))); + point.setLatitude(53.551086); + point.setLongitude(9.993682); + point.setAccuracyMeters(10.0); + points.add(point); + } + int inserted = sourceLocationPointJdbcService.bulkInsert(user, device, points); + assertTrue(inserted == count, "expected all points to be inserted"); + } +} diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java index 99b66891..d8c517c6 100644 --- a/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/PromotionCleanupChurnTest.java @@ -22,24 +22,18 @@ import java.util.UUID; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.verify; /** - * Demonstrates the throughput problem behind #1212 ("promotion advances at ~1/8 real-time"). + * Verifies the throughput fix for #1212 ("promotion advances at ~1/8 real-time"). *

- * Every promotion - no matter how small the flush - schedules a LocationDataCleanupTask. - * That task widens the range by the anomaly lookback (reitti.geo-point-filter.history-lookback-hours, - * default 24h) on both sides and hands that range to UpdateCuratedTimelineTask, which deletes - * every raw_location_points row in the window (dropForReSeeding) and re-inserts it as - * processed = false (updateFromDevices). The pipeline then re-detects visits/trips for all of them. - *

- * So promoting a few seconds of new data re-opens ~48h of already processed history. With the - * default batching config a promotion job is enqueued every ~5s per stream, the chain - * (promotion -> cleanup -> reseed -> pipeline) can never keep up with real time, jobs pile up - * faster than they drain and the whole Quartz machinery saturates. + * Each promotion cycle used to widen the promoted range by a static ±24h lookback and reseed + * (delete + re-insert as unprocessed) that whole window, so promoting a few seconds of data + * re-opened ~48h of already processed history. The window is now resolved from the actual + * boundary conditions (context margin, capped at maxInterpolationGapMinutes), so a tiny + * promotion only re-opens the points within that margin. */ @IntegrationTest class PromotionCleanupChurnTest { @@ -63,7 +57,7 @@ class PromotionCleanupChurnTest { private JobSchedulingService jobSchedulingService; @Test - void singlePromotedPointReopensTwoDaysOfAlreadyProcessedTimeline() { + void tinyPromotionOnlyReopensTheContextMarginInsteadOfTwoDays() { User user = testingService.randomUser(); Device device = testingService.findDefaultDevice(user); @@ -76,20 +70,20 @@ void singlePromotedPointReopensTwoDaysOfAlreadyProcessedTimeline() { Instant flushEnd = HISTORY_START.plus(Duration.ofMinutes(5)); TimeRange reseedRange1 = runCleanupAndCaptureReseedRange(user, device, flushStart, flushEnd); - // LocationDataCleanupTask hands the ±24h lookback window to the timeline update, - // not the 5 minutes that were actually promoted - assertEquals(Duration.ofHours(24), Duration.between(reseedRange1.start(), flushStart)); - assertTrue(Duration.between(reseedRange1.start(), reseedRange1.end()).compareTo(Duration.ofHours(48)) >= 0); + // no prior points -> no backward extension; forward only up to 50 points of context, + // not the static +24h of the old lookback + assertEquals(flushStart, reseedRange1.start()); + assertEquals(flushEnd.plus(Duration.ofMinutes(250)), reseedRange1.end()); reseedFromView(user, reseedRange1); - // 289 timeline points were seeded although the flush contained only 2 points - assertEquals(289, timelineCount(user)); - assertEquals(289, unprocessedCount(user)); + // 51 timeline points were seeded (50 points of context + the flush), not 289 + assertEquals(51, timelineCount(user)); + assertEquals(51, unprocessedCount(user)); // the pipeline finished - all points are processed now markAllProcessed(user); - assertEquals(289, processedCount(user)); + assertEquals(51, processedCount(user)); // --- promotion cycle 2: a single new point arrives --- Instant newPoint = HISTORY_END.plus(Duration.ofMinutes(1)); @@ -99,11 +93,12 @@ void singlePromotedPointReopensTwoDaysOfAlreadyProcessedTimeline() { TimeRange reseedRange2 = runCleanupAndCaptureReseedRange(user, device, newPoint, newPoint); reseedFromView(user, reseedRange2); - // the single new point wiped and re-opened the whole lookback window: - // 288 already processed points are unprocessed again and will be re-detected - assertEquals(288, processedCount(user)); - assertEquals(289, unprocessedCount(user), "one new point was promoted, but the whole window was re-opened"); - assertEquals(577, timelineCount(user)); + // the single new point re-opens only the 50-point context margin before it, + // not the ~48h window the old lookback produced + assertEquals(HISTORY_END.minus(Duration.ofMinutes(49 * 5)), reseedRange2.start()); + assertEquals(51, processedCount(user), "already processed points must stay processed"); + assertEquals(51, unprocessedCount(user), "50 context points + the new point"); + assertEquals(102, timelineCount(user)); } private TimeRange runCleanupAndCaptureReseedRange(User user, Device device, Instant start, Instant end) { diff --git a/src/test/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserterTest.java b/src/test/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserterTest.java index 3ed23ddd..4decb572 100644 --- a/src/test/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserterTest.java +++ b/src/test/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserterTest.java @@ -50,8 +50,8 @@ void shouldGenerateSyntheticPointsForLargeGaps() { // When: we simulate a new point arriving in between and trigger gap filling Instant newPointTime = start.plus(1, ChronoUnit.MINUTES); - // Determine the time range of new points (just one point) - TimeRange range = new TimeRange(newPointTime, newPointTime); + // the caller passes a window covering the gap (see ProcessingWindowResolver) + TimeRange range = new TimeRange(start, end.plusMillis(1)); syntheticPointInserter.fillGaps(testUser, range); // Then: synthetic points should be inserted @@ -70,7 +70,7 @@ void shouldRespectMaxInterpolationDistance() { createAndSaveRawPoint(start, 50.0, 8.0); createAndSaveRawPoint(end, 50.01, 8.01); - TimeRange range = new TimeRange(start.plus(1, ChronoUnit.MINUTES), start.plus(1, ChronoUnit.MINUTES)); + TimeRange range = new TimeRange(start, end.plusMillis(1)); syntheticPointInserter.fillGaps(testUser, range); List all = rawLocationPointService @@ -88,7 +88,7 @@ void shouldRespectMaxInterpolationTimeGap() { createAndSaveRawPoint(start, 50.0, 8.0); createAndSaveRawPoint(end, 50.001, 8.001); - TimeRange range = new TimeRange(start.plus(90, ChronoUnit.MINUTES), start.plus(90, ChronoUnit.MINUTES)); + TimeRange range = new TimeRange(start, end.plusMillis(1)); syntheticPointInserter.fillGaps(testUser, range); List all = rawLocationPointService @@ -126,7 +126,7 @@ void shouldGenerateExpectedNumberOfSyntheticPointsForGivenRealPoints() { TimeRange range = new TimeRange( Instant.parse("2013-04-15T06:31:26.860000Z"), - Instant.parse("2013-04-15T06:36:32.566000Z")); + Instant.parse("2013-04-15T06:36:32.566000Z").plusMillis(1)); syntheticPointInserter.fillGaps(testUser, range); List stored = rawLocationPointService From 09253198f09d1a57abfd842c229eff7cc0a9f03b Mon Sep 17 00:00:00 2001 From: Daniel Graf Date: Sat, 29 Aug 2026 13:09:51 +0200 Subject: [PATCH 6/6] bugfix(#1212) Remove outdated comments from processing and service classes - Clean up obsolete and redundant comments in `UnifiedLocationProcessingService`. - Remove outdated annotations and documentation in `AnomalyProcessingService` and `SyntheticPointInserter`. - Simplify promotion logic by removing unnecessary comments in `PromotionJobHandler` and `LocationBatchingService`. - Streamline range handling by eliminating redundant notes in `ExcessDensityHandler` and `LocationDataCleanupTask`. --- .../reitti/service/LocationBatchingService.java | 3 --- .../reitti/service/importer/PromotionJobHandler.java | 2 -- .../service/processing/AnomalyProcessingService.java | 5 ----- .../service/processing/ExcessDensityHandler.java | 2 -- .../service/processing/LocationDataCleanupTask.java | 2 -- .../service/processing/SyntheticPointInserter.java | 11 ----------- .../processing/UnifiedLocationProcessingService.java | 6 ------ 7 files changed, 31 deletions(-) diff --git a/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java b/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java index e1edd675..3cebf401 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/LocationBatchingService.java @@ -95,9 +95,6 @@ private void executeFlush(UserBatch batch) { batch.getLocationPoints() ); batch.clear(); - // coalesce: the rows are staged, an already scheduled or running promotion for this - // partition will pick them up. This keeps us from enqueueing a new promotion trigger - // every few seconds, which lets the Quartz queue grow faster than it drains (#1212). if (promotionInflightGuard.tryAcquire(pKey)) { try { this.jobScheduler.enqueueTask(promotionTask, diff --git a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java index 9a8274a4..331d12a9 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java +++ b/src/main/java/com/dedicatedcode/reitti/service/importer/PromotionJobHandler.java @@ -107,8 +107,6 @@ public void execute(JobExecutionContext context) throws JobExecutionException { promotionInflightGuard.release(partitionKey); } - // points flushed by the ingest thread while this promotion was running are still - // unpromoted - make sure they get promoted instead of waiting for the next flush if (this.stagingService.hasUnpromotedPoints(partitionKey) && promotionInflightGuard.tryAcquire(partitionKey)) { try { this.jobSchedulingService.enqueueTask(promotionTask, diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java index cc4d6438..25e3772c 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java @@ -12,11 +12,6 @@ import java.time.Instant; import java.util.List; -/** - * Re-evaluates the anomaly status of the source points within the given time range and marks - * outliers as invalid. The caller is responsible for passing a range that includes enough - * neighbor context for the boundary points (see {@link ProcessingWindowResolver}). - */ @Service public class AnomalyProcessingService { private static final Logger logger = LoggerFactory.getLogger(AnomalyProcessingService.class); diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java b/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java index 1f349631..1dfa7847 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/ExcessDensityHandler.java @@ -27,8 +27,6 @@ public ExcessDensityHandler(LocationDensityConfig config, } public TimeRange handleExcess(User user, Device device, TimeRange inputRange) { - // The caller provides a range that already includes the boundary context (see - // ProcessingWindowResolver), so adjacent points on both range edges are covered. List points = rawLocationPointService.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, inputRange.start(), inputRange.end(), false, true); if (points.size() < 2) { return TimeRange.empty(); diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java index 2c9cd4b5..1e179b14 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/LocationDataCleanupTask.java @@ -64,8 +64,6 @@ public void execute(TaskData data) { Instant end = data.getEnd(); log.debug("Starting LocationDataCleanupJob for user [{}] and device [{}] between {} and {}", user, device, start, end); this.metadataRepository.updateProgress(jobId, 0,4, "Anomaly processing started ..."); - // resolve how far around the promoted range we actually have to process: context - // margin for the filters, closed gaps and backfill - instead of a static ±24h window TimeRange window = processingWindowResolver.resolve(user, device, new TimeRange(start, end)); TimeRange processedTimeRange = anomalyProcessingService.processAndMarkAnomalies(user, device, window.start(), window.end()); this.metadataRepository.updateProgress(jobId, 1,4, "Density normalization started ..."); diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java b/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java index 82dd61c0..b5513266 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/SyntheticPointInserter.java @@ -43,17 +43,6 @@ public SyntheticPointInserter(LocationDensityConfig config, this.maxBatchSize = maxBatchSize; } - /** - * Processes the given time range: deletes old synthetic points, then - * inserts new synthetic points where real-point gaps are too large. - *

- * The caller provides a range that already includes the boundary context (see - * {@link ProcessingWindowResolver}), so gaps closing at the range edges are bridged as - * long as they do not exceed the configured interpolation limit. - * - * @param user the owning user - * @param inputRange the time range that covers the newly arrived points - */ public void fillGaps(User user, TimeRange inputRange) { // 1. Fetch density configuration (using the earliest point time) DetectionParameter detectionParams = visitDetectionParametersService.getCurrentConfiguration( diff --git a/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java b/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java index b43ead68..c83a57ab 100644 --- a/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java +++ b/src/main/java/com/dedicatedcode/reitti/service/processing/UnifiedLocationProcessingService.java @@ -239,10 +239,6 @@ private VisitDetectionResult detectVisits(User user, LocationProcessEvent event) currentConfiguration = previewVisitDetectionParametersJdbcService.findCurrent(user, previewId); } - // The window only spans the batch plus enough margin for stays that continue across - // its edges (a stay becomes visible as processed visit and pulls the window out to - // its boundaries below). A static ±1 day window here made every pipeline run - // re-detect visits over days of already processed data (#1212). Duration boundaryMargin = Duration.ofSeconds( currentConfiguration.getVisitDetection().getMinimumStayTimeInSeconds() + 2 * currentConfiguration.getVisitDetection().getMaxMergeTimeBetweenSameStayPoints()); @@ -263,8 +259,6 @@ private VisitDetectionResult detectVisits(User user, LocationProcessEvent event) windowStart = existingProcessedVisits.getFirst().getStartTime(); } if (!existingProcessedVisits.getLast().getEndTime().isBefore(windowEnd)) { - // the point queries load with an exclusive end, so the visit's defining last - // point must be covered by extending the window beyond the visit end windowEnd = existingProcessedVisits.getLast().getEndTime().plus(1, ChronoUnit.MILLIS); } }