(ping) Avoid thread starvation scenario in job scheduling

Adjust the queueing strategy to avoid thread starvation from whale domains with many subdomains all locking on the same semaphore and gunking up all threads by implementing a mechanism that returns jobs that can't be executed to the queue.

This will lead to some queue churn, but it should be fairly manageable given the small number of threads involved, and the fairly long job execution times.

code/libraries/domain-lock/java/nu/marginalia/coordination/LocalDomainCoordinator.java

@@ -1,5 +1,6 @@
 package nu.marginalia.coordination;
 
+import com.google.inject.Singleton;
 import nu.marginalia.model.EdgeDomain;
 
 import java.time.Duration;
@@ -9,6 +10,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+@Singleton
 public class LocalDomainCoordinator implements DomainCoordinator {
     // The locks are stored in a map, with the domain name as the key.  This map will grow
     // relatively big, but should be manageable since the number of domains is limited to

code/libraries/domain-lock/java/nu/marginalia/coordination/ZookeeperDomainCoordinator.java

@@ -1,6 +1,7 @@
 package nu.marginalia.coordination;
 
 import com.google.inject.Inject;
+import com.google.inject.Singleton;
 import com.google.inject.name.Named;
 import nu.marginalia.model.EdgeDomain;
 import nu.marginalia.service.discovery.ServiceRegistryIf;
@@ -13,11 +14,14 @@ import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 
+@Singleton
 public class ZookeeperDomainCoordinator implements DomainCoordinator {
     // The locks are stored in a map, with the domain name as the key.  This map will grow
     // relatively big, but should be manageable since the number of domains is limited to
     // a few hundred thousand typically.
     private final Map<String, InterProcessSemaphoreV2> locks = new ConcurrentHashMap<>();
+    private final Map<String, Integer> waitCounts = new ConcurrentHashMap<>();
+
     private final ServiceRegistryIf serviceRegistry;
     private final int nodeId;
 
@@ -32,27 +36,35 @@ public class ZookeeperDomainCoordinator implements DomainCoordinator {
      * and may be held by another thread.  The caller is responsible for locking and  releasing the lock.
      */
     public DomainLock lockDomain(EdgeDomain domain) throws InterruptedException {
-        var sem = locks.computeIfAbsent(domain.topDomain.toLowerCase(), this::createSemapore);
+        final String key = domain.topDomain.toLowerCase();
+        var sem = locks.computeIfAbsent(key, this::createSemapore);
 
+        // Increment or add a wait count for the domain
+        waitCounts.compute(key, (k,value) -> (value == null ? 1 : value + 1));
         try {
-            var lease = sem.acquire();
-
-            return new ZkDomainLock(sem, lease);
+            return new ZkDomainLock(sem, sem.acquire());
         }
         catch (Exception e) {
             throw new RuntimeException("Failed to acquire lock for domain: " + domain.topDomain, e);
         }
+        finally {
+            // Decrement or remove the wait count for the domain
+            waitCounts.compute(key, (k,value) -> (value == null || value <= 1) ? null : value - 1);
+        }
     }
 
+
     public Optional<DomainLock> tryLockDomain(EdgeDomain domain) throws InterruptedException {
         return tryLockDomain(domain, Duration.ofSeconds(1)); // Underlying semaphore doesn't have a tryLock method, so we use a short timeout
     }
 
 
     public Optional<DomainLock> tryLockDomain(EdgeDomain domain, Duration timeout) throws InterruptedException {
+        final String key = domain.topDomain.toLowerCase();
+        var sem = locks.computeIfAbsent(key, this::createSemapore);
 
-        var sem = locks.computeIfAbsent(domain.topDomain.toLowerCase(), this::createSemapore);
-
+        // Increment or add a wait count for the domain
+        waitCounts.compute(key, (k,value) -> (value == null ? 1 : value + 1));
         try {
             var lease = sem.acquire(timeout.toMillis(), TimeUnit.MILLISECONDS); // Acquire with timeout
             if (lease != null) {
@@ -65,6 +77,9 @@ public class ZookeeperDomainCoordinator implements DomainCoordinator {
         catch (Exception e) {
             return Optional.empty(); // If we fail to acquire the lock, we return an empty optional
         }
+        finally {
+            waitCounts.compute(key, (k,value) -> (value == null || value <= 1) ? null : value - 1);
+        }
     }
 
     private InterProcessSemaphoreV2 createSemapore(String topDomain){
@@ -81,7 +96,7 @@ public class ZookeeperDomainCoordinator implements DomainCoordinator {
      * after this method returns true)
      */
     public boolean isLockableHint(EdgeDomain domain) {
-        return true; // Curator does not provide a way to check if a lock is available without acquiring it
+        return !waitCounts.containsKey(domain.topDomain.toLowerCase());
     }
 
     public static class ZkDomainLock implements DomainLock {

code/processes/ping-process/java/nu/marginalia/ping/PingDao.java

@@ -185,12 +185,12 @@ public class PingDao {
         return null;
     }
 
-    public List<UpdateSchedule.UpdateJob<Long, HistoricalAvailabilityData>> getDomainUpdateSchedule(int nodeId) {
-        List<UpdateSchedule.UpdateJob<Long, HistoricalAvailabilityData>> updateJobs = new ArrayList<>();
+    public List<UpdateSchedule.UpdateJob<DomainReference, HistoricalAvailabilityData>> getDomainUpdateSchedule(int nodeId) {
+        List<UpdateSchedule.UpdateJob<DomainReference, HistoricalAvailabilityData>> updateJobs = new ArrayList<>();
 
         try (var conn = dataSource.getConnection();
              var ps = conn.prepareStatement("""
-                SELECT ID, NEXT_SCHEDULED_UPDATE
+                SELECT ID, DOMAIN_NAME, NEXT_SCHEDULED_UPDATE
                 FROM EC_DOMAIN
                 LEFT JOIN DOMAIN_AVAILABILITY_INFORMATION
                 ON EC_DOMAIN.ID = DOMAIN_AVAILABILITY_INFORMATION.DOMAIN_ID
@@ -200,11 +200,13 @@ public class PingDao {
             ps.setInt(1, nodeId);
             ResultSet rs = ps.executeQuery();
             while (rs.next()) {
-                long domainId = rs.getLong("ID");
+                int domainId = rs.getInt("ID");
+                String domainName = rs.getString("DOMAIN_NAME");
                 var ts = rs.getTimestamp("NEXT_SCHEDULED_UPDATE");
                 Instant nextUpdate = ts == null ? Instant.now() : ts.toInstant();
 
-                updateJobs.add(new UpdateSchedule.UpdateJob<>(domainId, nextUpdate));
+                var ref = new DomainReference(domainId, nodeId, domainName.toLowerCase());
+                updateJobs.add(new UpdateSchedule.UpdateJob<>(ref, nextUpdate));
             }
         } catch (SQLException e) {
             throw new RuntimeException("Failed to retrieve domain update schedule", e);
@@ -241,7 +243,7 @@ public class PingDao {
                 else {
                     var record = new DomainDnsRecord(rs);
                     updateJobs.add(new UpdateSchedule.UpdateJob<>(
-                            new RootDomainReference.ById(dnsRootDomainId),
+                            new RootDomainReference.ByIdAndName(dnsRootDomainId, rootDomainName),
                             Objects.requireNonNullElseGet(record.tsNextScheduledUpdate(), Instant::now))
                     );
                 }

code/processes/ping-process/java/nu/marginalia/ping/PingJobScheduler.java

@@ -2,6 +2,7 @@ package nu.marginalia.ping;
 
 import com.google.inject.Inject;
 import nu.marginalia.coordination.DomainCoordinator;
+import nu.marginalia.model.EdgeDomain;
 import nu.marginalia.ping.model.*;
 import nu.marginalia.ping.svc.DnsPingService;
 import nu.marginalia.ping.svc.HttpPingService;
@@ -31,7 +32,7 @@ public class PingJobScheduler {
 
     private static final UpdateSchedule<RootDomainReference, RootDomainReference> dnsUpdateSchedule
             = new UpdateSchedule<>(250_000);
-    private static final UpdateSchedule<Long, HistoricalAvailabilityData> availabilityUpdateSchedule
+    private static final UpdateSchedule<DomainReference, HistoricalAvailabilityData> availabilityUpdateSchedule
             = new UpdateSchedule<>(250_000);
 
     public volatile Instant dnsLastSync = Instant.now();
@@ -138,7 +139,15 @@ public class PingJobScheduler {
                     continue;
                 }
 
-                long nextId = availabilityUpdateSchedule.next();
+                DomainReference ref = availabilityUpdateSchedule.nextIf(domain -> {
+                    EdgeDomain domainObj = new EdgeDomain(domain.domainName());
+                    if (!domainCoordinator.isLockableHint(domainObj)) {
+                        return false; // Skip locked domains
+                    }
+                    return true; // Process this domain
+                });
+
+                long nextId = ref.domainId();
                 var data = pingDao.getHistoricalAvailabilityData(nextId);
                 if (data == null) {
                     logger.warn("No availability data found for ID: {}", nextId);
@@ -163,7 +172,7 @@ public class PingJobScheduler {
                     for (var object : objects) {
                         var ts = object.nextUpdateTime();
                         if (ts != null) {
-                            availabilityUpdateSchedule.add(nextId, ts);
+                            availabilityUpdateSchedule.add(ref, ts);
                             break;
                         }
                     }
@@ -194,7 +203,7 @@ public class PingJobScheduler {
 
                 try {
                     List<WritableModel> objects = switch(ref) {
-                        case RootDomainReference.ById(long id) -> {
+                        case RootDomainReference.ByIdAndName(long id, String name) -> {
                             var oldRecord = Objects.requireNonNull(pingDao.getDomainDnsRecord(id));
                             yield dnsPingService.pingDomain(oldRecord.rootDomainName(), oldRecord);
                         }

code/processes/ping-process/java/nu/marginalia/ping/UpdateSchedule.java

@@ -2,9 +2,8 @@ package nu.marginalia.ping;
 
 import java.time.Duration;
 import java.time.Instant;
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.PriorityQueue;
+import java.util.*;
+import java.util.function.Predicate;
 
 /** In-memory schedule for updates, allowing jobs to be added and processed in order of their scheduled time.
  * This is not a particularly high-performance implementation, but exists to take contention off the database's
@@ -23,6 +22,9 @@ public class UpdateSchedule<T, T2> {
         notifyAll();
     }
 
+    /** Returns the next job in the queue that is due to be processed.
+     * If no jobs are due, it will block until a job is added or a job becomes due.
+     * */
     public synchronized T next() throws InterruptedException {
         while (true) {
             if (updateQueue.isEmpty()) {
@@ -44,6 +46,56 @@ public class UpdateSchedule<T, T2> {
         }
     }
 
+
+    /** Returns the first job in the queue matching the predicate that is not scheduled into the future,
+     * blocking until a job is added or a job becomes due.
+     */
+    public synchronized T nextIf(Predicate<T> predicate) throws InterruptedException {
+        List<UpdateJob<T, T2>> rejectedJobs = new ArrayList<>();
+
+        try {
+            while (true) {
+                if (updateQueue.isEmpty()) {
+                    wait(); // Wait for a new job to be added
+                    continue;
+                }
+
+                UpdateJob<T, T2> job = updateQueue.peek();
+                Instant now = Instant.now();
+
+                if (job.updateTime.isAfter(now)) {
+                    Duration toWait = Duration.between(now, job.updateTime);
+
+                    // Return the rejected jobs to the queue for other threads to process
+                    updateQueue.addAll(rejectedJobs);
+                    if (!rejectedJobs.isEmpty())
+                        notifyAll();
+                    rejectedJobs.clear();
+
+                    wait(Math.max(1, toWait.toMillis()));
+                } else {
+                    var candidate = updateQueue.poll(); // Remove the job from the queue since it's due
+
+                    assert candidate != null : "Update job should not be null at this point, since we just peeked it in a synchronized block";
+
+                    if (!predicate.test(candidate.key())) {
+                        rejectedJobs.add(candidate);
+                    }
+                    else {
+                        return candidate.key();
+                    }
+                }
+            }
+        }
+        finally {
+            // Return the rejected jobs to the queue for other threads to process
+            updateQueue.addAll(rejectedJobs);
+            if (!rejectedJobs.isEmpty())
+                notifyAll();
+        }
+
+    }
+
     public synchronized void clear() {
         updateQueue.clear();
         notifyAll();

code/processes/ping-process/java/nu/marginalia/ping/model/RootDomainReference.java

@@ -1,6 +1,6 @@
 package nu.marginalia.ping.model;
 
 public sealed interface RootDomainReference {
-    record ById(long id) implements RootDomainReference { }
+    record ByIdAndName(long id, String name) implements RootDomainReference { }
     record ByName(String name) implements RootDomainReference { }
 }

code/processes/ping-process/test/nu/marginalia/ping/AvailabilityJobSchedulerTest.java

@@ -86,7 +86,9 @@ class AvailabilityJobSchedulerTest {
         DomainDnsInformationFactory dnsDomainInformationFactory = new DomainDnsInformationFactory(processConfig, pic);
 
         PingJobScheduler pingJobScheduler = new PingJobScheduler(
-                new HttpPingService(pingHttpFetcher,
+                new HttpPingService(
+                        new LocalDomainCoordinator(),
+                        pingHttpFetcher,
                         new DomainAvailabilityInformationFactory(new GeoIpDictionary(), new BackoffStrategy(pic)),
                         new DomainSecurityInformationFactory()),
                 new DnsPingService(new PingDnsFetcher(List.of("8.8.8.8", "8.8.4.4")),

code/processes/ping-process/test/nu/marginalia/ping/PingHttpServiceTest.java

@@ -2,6 +2,7 @@ package nu.marginalia.ping;
 
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import nu.marginalia.coordination.LocalDomainCoordinator;
 import nu.marginalia.geoip.GeoIpDictionary;
 import nu.marginalia.ping.fetcher.PingHttpFetcher;
 import nu.marginalia.ping.io.HttpClientProvider;
@@ -63,6 +64,7 @@ class PingHttpServiceTest {
     public void testGetSslInfo() throws Exception {
         var provider = new HttpClientProvider();
         var pingService = new HttpPingService(
+                new LocalDomainCoordinator(),
                 new PingHttpFetcher(provider.get()),
                 new DomainAvailabilityInformationFactory(new GeoIpDictionary(),
                         new BackoffStrategy(PingModule.createPingIntervalsConfiguration())