Prevent concurrent syncs

src/main/java/net/ripe/rpki/rsyncit/rrdp/RRDPFetcherMetrics.java

@@ -13,21 +13,23 @@ public final class RRDPFetcherMetrics {
     private final Counter failedUpdates;
     private final Counter timeoutUpdates;
     private final Counter objectFailures;
+    private final Counter tooSlow;
 
     public final Timer objectConstructionTimer;
 
     public RRDPFetcherMetrics(MeterRegistry meterRegistry) {
         successfulUpdates = buildCounter("success", meterRegistry);
         failedUpdates = buildCounter("failed", meterRegistry);
         timeoutUpdates = buildCounter("timeout", meterRegistry);
+        tooSlow = buildCounter("slow", meterRegistry);
         objectFailures = Counter.builder("rsyncit.fetcher.objects")
-            .description("Metrics on objects")
-            .tag("status", "failure")
-            .register(meterRegistry);
+                .description("Metrics on objects")
+                .tag("status", "failure")
+                .register(meterRegistry);
 
         Gauge.builder("rsyncit.fetcher.rrdp.serial", rrdpSerial::get)
-            .description("Serial of the RRDP notification.xml at the given URL")
-            .register(meterRegistry);
+                .description("Serial of the RRDP notification.xml at the given URL")
+                .register(meterRegistry);
 
         objectConstructionTimer = Timer.builder("rsyncit.fetcher.parsing")
                 .description("Time spent parsing objects for the last run")
@@ -47,15 +49,19 @@ public void timeout() {
         this.timeoutUpdates.increment();
     }
 
+    public void tooSlow() {
+        this.tooSlow.increment();
+    }
+
     public void badObject() {
         this.objectFailures.increment();
     }
 
     private static Counter buildCounter(String statusTag, MeterRegistry registry) {
         return Counter.builder("rsyncit.fetcher.updated")
-            .description("Number of fetches")
-            .tag("status", statusTag)
-            .register(registry);
+                .description("Number of fetches")
+                .tag("status", statusTag)
+                .register(registry);
     }
 
 }

src/main/java/net/ripe/rpki/rsyncit/service/SyncService.java

@@ -16,6 +16,7 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 @Slf4j
 @Component
@@ -26,6 +27,7 @@ public class SyncService {
     private final AppConfig appConfig;
     private final State state;
     private final RRDPFetcherMetrics metrics;
+    private final AtomicBoolean isRunning = new AtomicBoolean(false);
 
     @Autowired
     public SyncService(WebClient webClient,
@@ -38,6 +40,25 @@ public SyncService(WebClient webClient,
     }
 
     public void sync() {
+        boolean shouldRun = true;
+        try {
+            shouldRun = isRunning.compareAndSet(false, true);
+            if (shouldRun) {
+                doSync();
+            } else {
+                log.info("Sync is already running, skipping this run. Most likely it means that the system is abnormally slow.");
+                metrics.tooSlow();
+            }
+        } finally {
+            if (shouldRun) {
+                // shouldRun is true by default to prevent the service to get stuck in the "running" state.
+                // So if this thread was interrupted before "compareAndSet", we reset it to "not running".
+                isRunning.set(false);
+            }
+        }
+    }
+
+    private void doSync() {
         var config = appConfig.getConfig();
         var rrdpFetcher = new RrdpFetcher(config, webClient, state, metrics);
 
@@ -46,7 +67,7 @@ public void sync() {
         if (fetchResult instanceof RrdpFetcher.NoUpdates noUpdates) {
             metrics.success(noUpdates.serial());
             log.info("Session id {} and serial {} have not changed since the last check, nothing to update",
-                noUpdates.sessionId(), noUpdates.serial());
+                    noUpdates.sessionId(), noUpdates.serial());
         } else if (fetchResult instanceof RrdpFetcher.SuccessfulFetch success) {
             metrics.success(success.serial());
             log.info("Fetched {} objects in {}ms", success.objects().size(), t.getTime());