1212 bug promotion advances at 18 real time with majority of points marked invalid staged partitions never consumed v530

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));
     }
 

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<SourceLocationPoint> 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<Instant> 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<Instant> 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<Instant> 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<Instant> 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<Long> 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 = ?",

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,21 @@ 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());
+            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);
         }

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,94 @@ 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 != null) {
-                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);
+        }
+
+        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");
         }
     }
 

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<String> inFlight = ConcurrentHashMap.newKeySet();
+
+    public boolean tryAcquire(String partitionKey) {
+        return this.inFlight.add(partitionKey);
+    }
+
+    public void release(String partitionKey) {
+        this.inFlight.remove(partitionKey);
+    }
+}

src/main/java/com/dedicatedcode/reitti/service/processing/AnomalyProcessingService.java

@@ -10,33 +10,27 @@
 import org.springframework.stereotype.Service;
 
 import java.time.Instant;
-import java.time.temporal.ChronoUnit;
 import java.util.List;
 
 @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<SourceLocationPoint> pointsToCheck = repository.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, startTime, endTime, false, true);
+        repository.resetInvalidStatus(user, start, end);
+        List<SourceLocationPoint> pointsToCheck = repository.findByUserAndTimestampBetweenOrderByTimestampAsc(user, device, start, end, false, true);
         logger.debug("Found {} points to check for user {}", pointsToCheck.size(), user.getUsername());
         List<SourceLocationPoint> 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);
     }
 }