(control) Fix so that sideload actions show up in Mixed profile nodes

The commit removes attempts at probing the root document, feed URLs, and favicon if we are not permitted to do so via robots.txt

code/common/config/java/nu/marginalia/nodecfg/NodeConfigurationService.java

@@ -45,7 +45,7 @@ public class NodeConfigurationService {
     public List<NodeConfiguration> getAll() {
         try (var conn = dataSource.getConnection();
              var qs = conn.prepareStatement("""
-                     SELECT ID, DESCRIPTION, ACCEPT_QUERIES, AUTO_CLEAN, PRECESSION, KEEP_WARCS, NODE_PROFILE, DISABLED
+                     SELECT ID, DESCRIPTION, ACCEPT_QUERIES, AUTO_CLEAN, PRECESSION, AUTO_ASSIGN_DOMAINS, KEEP_WARCS, NODE_PROFILE, DISABLED
                      FROM NODE_CONFIGURATION
                      """)) {
             var rs = qs.executeQuery();
@@ -59,6 +59,7 @@ public class NodeConfigurationService {
                         rs.getBoolean("ACCEPT_QUERIES"),
                         rs.getBoolean("AUTO_CLEAN"),
                         rs.getBoolean("PRECESSION"),
+                        rs.getBoolean("AUTO_ASSIGN_DOMAINS"),
                         rs.getBoolean("KEEP_WARCS"),
                         NodeProfile.valueOf(rs.getString("NODE_PROFILE")),
                         rs.getBoolean("DISABLED")
@@ -75,7 +76,7 @@ public class NodeConfigurationService {
     public NodeConfiguration get(int nodeId) throws SQLException {
         try (var conn = dataSource.getConnection();
              var qs = conn.prepareStatement("""
-                     SELECT ID, DESCRIPTION, ACCEPT_QUERIES, AUTO_CLEAN, PRECESSION, KEEP_WARCS, NODE_PROFILE, DISABLED
+                     SELECT ID, DESCRIPTION, ACCEPT_QUERIES, AUTO_CLEAN, PRECESSION, AUTO_ASSIGN_DOMAINS, KEEP_WARCS, NODE_PROFILE, DISABLED
                      FROM NODE_CONFIGURATION
                      WHERE ID=?
                      """)) {
@@ -88,6 +89,7 @@ public class NodeConfigurationService {
                         rs.getBoolean("ACCEPT_QUERIES"),
                         rs.getBoolean("AUTO_CLEAN"),
                         rs.getBoolean("PRECESSION"),
+                        rs.getBoolean("AUTO_ASSIGN_DOMAINS"),
                         rs.getBoolean("KEEP_WARCS"),
                         NodeProfile.valueOf(rs.getString("NODE_PROFILE")),
                         rs.getBoolean("DISABLED")
@@ -102,7 +104,7 @@ public class NodeConfigurationService {
         try (var conn = dataSource.getConnection();
              var us = conn.prepareStatement("""
                      UPDATE NODE_CONFIGURATION
-                     SET DESCRIPTION=?, ACCEPT_QUERIES=?,  AUTO_CLEAN=?, PRECESSION=?, KEEP_WARCS=?, DISABLED=?, NODE_PROFILE=?
+                     SET DESCRIPTION=?, ACCEPT_QUERIES=?,  AUTO_CLEAN=?, PRECESSION=?, AUTO_ASSIGN_DOMAINS=?, KEEP_WARCS=?, DISABLED=?, NODE_PROFILE=?
                      WHERE ID=?
                      """))
         {
@@ -110,10 +112,11 @@ public class NodeConfigurationService {
             us.setBoolean(2, config.acceptQueries());
             us.setBoolean(3, config.autoClean());
             us.setBoolean(4, config.includeInPrecession());
-            us.setBoolean(5, config.keepWarcs());
-            us.setBoolean(6, config.disabled());
-            us.setString(7, config.profile().name());
-            us.setInt(8, config.node());
+            us.setBoolean(5, config.autoAssignDomains());
+            us.setBoolean(6, config.keepWarcs());
+            us.setBoolean(7, config.disabled());
+            us.setString(8, config.profile().name());
+            us.setInt(9, config.node());
 
             if (us.executeUpdate() <= 0)
                 throw new IllegalStateException("Failed to update configuration");

code/common/config/java/nu/marginalia/nodecfg/model/NodeConfiguration.java

@@ -5,6 +5,7 @@ public record NodeConfiguration(int node,
                                 boolean acceptQueries,
                                 boolean autoClean,
                                 boolean includeInPrecession,
+                                boolean autoAssignDomains,
                                 boolean keepWarcs,
                                 NodeProfile profile,
                                 boolean disabled

code/common/config/java/nu/marginalia/nodecfg/model/NodeProfile.java

@@ -20,9 +20,7 @@ public enum NodeProfile {
     }
 
     public boolean permitBatchCrawl() {
-        return isBatchCrawl() ||isMixed();
-    }
-    public boolean permitSideload() {
-        return isMixed() || isSideload();
+        return isBatchCrawl() || isMixed();
     }
+    public boolean permitSideload() {  return isSideload() || isMixed(); }
 }

code/common/config/test/nu/marginalia/nodecfg/NodeConfigurationServiceTest.java

@@ -2,6 +2,7 @@ package nu.marginalia.nodecfg;
 
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import nu.marginalia.nodecfg.model.NodeConfiguration;
 import nu.marginalia.nodecfg.model.NodeProfile;
 import nu.marginalia.test.TestMigrationLoader;
 import org.junit.jupiter.api.BeforeAll;
@@ -62,6 +63,63 @@ public class NodeConfigurationServiceTest {
         assertEquals(2, list.size());
         assertEquals(a, list.get(0));
         assertEquals(b, list.get(1));
+    }
 
+
+    // Test all the fields that are only exposed via save()
+    @Test
+    public void testSaveChanges() throws SQLException {
+        var original = nodeConfigurationService.create(1, "Test", false, false, NodeProfile.MIXED);
+
+        assertEquals(1, original.node());
+        assertEquals("Test", original.description());
+        assertFalse(original.acceptQueries());
+
+        var precession = new NodeConfiguration(
+                original.node(),
+                "Foo",
+                true,
+                original.autoClean(),
+                original.includeInPrecession(),
+                !original.autoAssignDomains(),
+                original.keepWarcs(),
+                original.profile(),
+                original.disabled()
+        );
+
+        nodeConfigurationService.save(precession);
+        precession = nodeConfigurationService.get(original.node());
+        assertNotEquals(original.autoAssignDomains(), precession.autoAssignDomains());
+
+        var autoClean = new NodeConfiguration(
+                original.node(),
+                "Foo",
+                true,
+                !original.autoClean(),
+                original.includeInPrecession(),
+                original.autoAssignDomains(),
+                original.keepWarcs(),
+                original.profile(),
+                original.disabled()
+        );
+
+        nodeConfigurationService.save(autoClean);
+        autoClean = nodeConfigurationService.get(original.node());
+        assertNotEquals(original.autoClean(), autoClean.autoClean());
+
+        var disabled = new NodeConfiguration(
+                original.node(),
+                "Foo",
+                true,
+                autoClean.autoClean(),
+                autoClean.includeInPrecession(),
+                autoClean.autoAssignDomains(),
+                autoClean.keepWarcs(),
+                autoClean.profile(),
+                !autoClean.disabled()
+        );
+        nodeConfigurationService.save(disabled);
+        disabled = nodeConfigurationService.get(original.node());
+        assertNotEquals(autoClean.disabled(), disabled.disabled());
     }
 }

code/common/db/resources/db/migration/V25_06_0_006__ndp__node_configuration.sql

@@ -0,0 +1,3 @@
+-- Migration script to add AUTO_ASSIGN_DOMAINS column to NODE_CONFIGURATION table
+
+ALTER TABLE NODE_CONFIGURATION ADD COLUMN AUTO_ASSIGN_DOMAINS BOOLEAN NOT NULL DEFAULT TRUE;

code/processes/crawling-process/java/nu/marginalia/crawl/retreival/CrawlerRetreiver.java

@@ -115,9 +115,13 @@ public class CrawlerRetreiver implements AutoCloseable {
                     final SimpleRobotRules robotsRules = fetcher.fetchRobotRules(probedUrl.domain, warcRecorder);
                     final CrawlDelayTimer delayTimer = new CrawlDelayTimer(robotsRules.getCrawlDelay());
 
+                    if (!robotsRules.isAllowed(probedUrl.toString())) {
+                        warcRecorder.flagAsRobotsTxtError(probedUrl);
+                        yield 1; // Nothing we can do here, we aren't allowed to fetch the root URL
+                    }
                     delayTimer.waitFetchDelay(0); // initial delay after robots.txt
 
-                    DomainStateDb.SummaryRecord summaryRecord = sniffRootDocument(probedUrl, delayTimer);
+                    DomainStateDb.SummaryRecord summaryRecord = sniffRootDocument(probedUrl, robotsRules, delayTimer);
                     domainStateDb.save(summaryRecord);
 
                     if (Thread.interrupted()) {
@@ -270,11 +274,11 @@ public class CrawlerRetreiver implements AutoCloseable {
 
 
 
-    private DomainStateDb.SummaryRecord sniffRootDocument(EdgeUrl rootUrl, CrawlDelayTimer timer) {
+    private DomainStateDb.SummaryRecord sniffRootDocument(EdgeUrl rootUrl, SimpleRobotRules robotsRules, CrawlDelayTimer timer) {
         Optional<String> feedLink = Optional.empty();
 
         try {
-            var url = rootUrl.withPathAndParam("/", null);
+            EdgeUrl url = rootUrl.withPathAndParam("/", null);
 
             HttpFetchResult result = fetcher.fetchContent(url, warcRecorder, cookies, timer, ContentTags.empty(), HttpFetcher.ProbeType.DISABLED);
             timer.waitFetchDelay(0);
@@ -331,7 +335,7 @@ public class CrawlerRetreiver implements AutoCloseable {
 
 
             if (feedLink.isEmpty()) {
-                feedLink = guessFeedUrl(timer);
+                feedLink = guessFeedUrl(timer, robotsRules);
             }
 
             // Download the sitemap if available
@@ -339,14 +343,18 @@ public class CrawlerRetreiver implements AutoCloseable {
 
             // Grab the favicon if it exists
 
-            if (fetcher.fetchContent(faviconUrl, warcRecorder, cookies, timer, ContentTags.empty(), HttpFetcher.ProbeType.DISABLED) instanceof HttpFetchResult.ResultOk iconResult) {
-                String contentType = iconResult.header("Content-Type");
-                byte[] iconData = iconResult.getBodyBytes();
+            if (robotsRules.isAllowed(faviconUrl.toString())) {
+                if (fetcher.fetchContent(faviconUrl, warcRecorder, cookies, timer, ContentTags.empty(), HttpFetcher.ProbeType.DISABLED)
+                        instanceof HttpFetchResult.ResultOk iconResult)
+                {
+                    String contentType = iconResult.header("Content-Type");
+                    byte[] iconData = iconResult.getBodyBytes();
 
-                domainStateDb.saveIcon(
-                        domain,
-                        new DomainStateDb.FaviconRecord(contentType, iconData)
-                );
+                    domainStateDb.saveIcon(
+                            domain,
+                            new DomainStateDb.FaviconRecord(contentType, iconData)
+                    );
+                }
             }
             timer.waitFetchDelay(0);
 
@@ -383,7 +391,7 @@ public class CrawlerRetreiver implements AutoCloseable {
             "blog/rss"
     );
 
-    private Optional<String> guessFeedUrl(CrawlDelayTimer timer) throws InterruptedException {
+    private Optional<String> guessFeedUrl(CrawlDelayTimer timer, SimpleRobotRules robotsRules) throws InterruptedException {
         var oldDomainStateRecord = domainStateDb.getSummary(domain);
 
         // If we are already aware of an old feed URL, then we can just revalidate it
@@ -396,6 +404,9 @@ public class CrawlerRetreiver implements AutoCloseable {
 
         for (String endpoint : likelyFeedEndpoints) {
             String url = "https://" + domain + "/" + endpoint;
+            if (!robotsRules.isAllowed(url)) {
+                continue;
+            }
             if (validateFeedUrl(url, timer)) {
                 return Optional.of(url);
             }

code/processes/new-domain-process/README.md

@@ -0,0 +1,12 @@
+The new domain process (NDP) is a process that evaluates new domains for
+inclusion in the search engine index.   
+
+It visits the root document of each candidate domain, ensures that it's reachable,
+verifies that the response is valid HTML, and checks for a few factors such as length
+and links before deciding whether to assign the domain to a node.
+
+The NDP process will assign new domains to the node with the fewest assigned domains.
+
+The NDP process is triggered with a goal target number of domains to process, and
+will find domains until that target is reached.  If e.g. a goal of 100 is set,
+and 50 are in the index, it will find 50 more domains.

code/processes/new-domain-process/build.gradle

@@ -31,6 +31,8 @@ dependencies {
     implementation project(':code:libraries:geo-ip')
     implementation project(':code:libraries:message-queue')
     implementation project(':code:libraries:blocking-thread-pool')
+    
+    implementation project(':code:functions:link-graph:api')
 
     implementation project(':code:processes:process-mq-api')
     implementation project(':code:processes:crawling-process:ft-content-type')

code/processes/new-domain-process/java/nu/marginalia/ndp/DomainEvaluator.java

@@ -2,40 +2,41 @@ package nu.marginalia.ndp;
 
 
 import com.google.inject.Inject;
+import com.google.inject.Singleton;
 import nu.marginalia.WmsaHome;
 import nu.marginalia.contenttype.ContentType;
 import nu.marginalia.contenttype.DocumentBodyToString;
 import nu.marginalia.coordination.DomainCoordinator;
 import nu.marginalia.link_parser.LinkParser;
 import nu.marginalia.model.EdgeDomain;
+import nu.marginalia.model.EdgeUrl;
 import nu.marginalia.ndp.io.HttpClientProvider;
-import nu.marginalia.ndp.model.DomainToTest;
 import org.apache.hc.client5.http.classic.HttpClient;
-import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
 import org.jsoup.Jsoup;
 import org.jsoup.nodes.Document;
+import org.jsoup.nodes.Element;
 
-import java.net.URI;
-import java.net.URISyntaxException;
+import java.io.InputStream;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
-import java.time.Instant;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
-
+/**  Evaluates a domain to determine if it is worth indexing.
+ *  This class fetches the root document, checks the response code, content type,
+ *  and parses the HTML to ensure it smells alright.
+ */
+@Singleton
 public class DomainEvaluator {
     private final HttpClient client;
     private final String userAgentString = WmsaHome.getUserAgent().uaString();
 
     private final LinkParser linkParser = new LinkParser();
     private final DomainCoordinator domainCoordinator;
-    sealed interface FetchResult permits FetchSuccess, FetchFailure {}
-    record FetchSuccess(Document content) implements FetchResult {}
-    record FetchFailure(String reason) implements FetchResult {}
 
     @Inject
     public DomainEvaluator(DomainCoordinator domainCoordinator) throws NoSuchAlgorithmException, KeyManagementException {
@@ -43,100 +44,103 @@ public class DomainEvaluator {
         client = HttpClientProvider.createClient();
     }
 
-    public boolean evaluateDomain(DomainToTest domain) throws Exception {
-        var edgeDomain = new EdgeDomain(domain.domainName());
+    public boolean evaluateDomain(String domainName) {
+        var edgeDomain = new EdgeDomain(domainName);
+
+        // Grab a lock on the domain to prevent concurrent evaluations between processes
         try (var lock = domainCoordinator.lockDomain(edgeDomain)) {
-            var result = fetch(domain.domainName());
+            var rootUrl = edgeDomain.toRootUrlHttps();
 
-            Instant start = Instant.now();
+            var request = ClassicRequestBuilder.get(rootUrl.asURI())
+                    .addHeader("User-Agent", userAgentString)
+                    .addHeader("Accept-Encoding", "gzip")
+                    .addHeader("Accept", "text/html,application/xhtml+xml;q=0.9")
+                    .build();
 
-            var ret = switch(result) {
-                case FetchSuccess(Document content) -> validateHtml(content, edgeDomain);
-                case FetchFailure failure -> false;
-            };
+            return client.execute(request, (rsp) -> {
+                if (rsp.getEntity() == null)
+                    return false;
 
-            // Sleep for up to 1 second before we yield the lock to respect rate limits reasonably well
-            Instant end = Instant.now();
-            Duration sleepDuration = Duration.ofSeconds(1).minus(Duration.between(start, end));
+                try {
+                    // Check if the response code indicates a successful fetch
+                    if (200 != rsp.getCode()) {
+                        return false;
+                    }
 
-            if (sleepDuration.isPositive()) {
-                TimeUnit.MILLISECONDS.sleep(sleepDuration.toMillis());
-            }
+                    byte[] content;
+                    // Read the content from the response entity
+                    try (InputStream contentStream = rsp.getEntity().getContent()) {
+                        content = contentStream.readNBytes(8192);
+                    }
 
-            return ret;
+                    // Parse the content (if it's valid)
+                    ContentType contentType = ContentType.parse(rsp.getEntity().getContentType());
+
+                    // Validate the content type
+                    if (!contentType.contentType().startsWith("text/html") && !contentType.contentType().startsWith("application/xhtml+xml"))
+                        return false;
+
+                    // Parse the document body to a Jsoup Document
+                    final Document document = Jsoup.parse(DocumentBodyToString.getStringData(contentType, content));
+                    final String text = document.body().text();
+
+                    if (text.length() < 100)
+                        return false;
+                    if (text.contains("404 Not Found") || text.contains("Page not found"))
+                        return false;
+                    if (hasMetaRefresh(document))
+                        return false; // This almost always indicates a parked domain
+                    if (!hasInternalLink(document, edgeDomain, rootUrl))
+                        return false; // No internal links means it's not worth indexing
+
+                    return true;
+                }
+                catch (Exception e) {
+                    return false;
+                }
+                finally {
+                    // May or may not be necessary, but let's ensure we clean up the response entity
+                    // to avoid resource leaks
+                    EntityUtils.consumeQuietly(rsp.getEntity());
+
+                    // Sleep for a while before yielding the lock, to avoid immediately hammering the domain
+                    // from another process
+                    sleepQuietly(Duration.ofSeconds(1));
+                }
+            });
+        }
+        catch (Exception ex) {
+            return false; // If we fail to fetch or parse the domain, we consider it invalid
         }
     }
 
-    private boolean validateHtml(Document content, EdgeDomain domain) {
-        var rootUrl = domain.toRootUrlHttps();
-        var text = content.body().text();
+    private boolean hasInternalLink(Document document, EdgeDomain currentDomain, EdgeUrl rootUrl) {
+        for (Element atag : document.select("a")) {
+            Optional<EdgeDomain> destDomain = linkParser
+                    .parseLink(rootUrl, atag)
+                    .map(EdgeUrl::getDomain);
 
-        if (text.length() < 100) {
-            return false; // Too short to be a valid page
+            if (destDomain.isPresent() && Objects.equals(currentDomain, destDomain.get()))
+                return true;
         }
-
-        if (text.contains("404 Not Found") || text.contains("Page not found")) {
-            return false; // Common indicators of a 404 page
-        }
-
-        for (var metaTag : content.select("meta")) {
-            if ("refresh".equalsIgnoreCase(metaTag.attr("http-equiv"))) {
-                return false; // Page has a refresh tag, very likely a parked domain
-            }
-        }
-
-        boolean hasInternalLink = false;
-
-        for (var atag : content.select("a")) {
-            var link = linkParser.parseLink(rootUrl, atag);
-            if (link.isEmpty()) {
-                continue; // Skip invalid links
-            }
-            var edgeUrl = link.get();
-            if (Objects.equals(domain, edgeUrl.getDomain())) {
-                hasInternalLink = true;
-            }
-        }
-
-        return hasInternalLink;
+        return false;
     }
 
-    private FetchResult fetch(String domain) throws URISyntaxException {
-        var uri = new URI("https://" + domain + "/");
-
-        var request = ClassicRequestBuilder.get(uri)
-                .addHeader("User-Agent", userAgentString)
-                .addHeader("Accept-Encoding", "gzip")
-                .addHeader("Accept", "text/html,application/xhtml+xml;q=0.9")
-                .build();
+    private boolean hasMetaRefresh(Document document) {
+        for (Element metaTag : document.select("meta")) {
+            if ("refresh".equalsIgnoreCase(metaTag.attr("http-equiv")))
+                return true;
+        }
+        return false;
+    }
 
+    private void sleepQuietly(Duration duration) {
         try {
-            return client.execute(request, (rsp) -> responseHandler(rsp, domain));
-        } catch (Exception e) {
-            return new FetchFailure("Failed to fetch domain: " + e.getMessage());
+            TimeUnit.MILLISECONDS.sleep(duration.toMillis());
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
         }
     }
 
-    private FetchResult responseHandler(ClassicHttpResponse rsp, String domain) {
-        if (rsp.getEntity() == null)
-            return new FetchFailure("No content returned from " + domain);
-
-        try {
-            int code = rsp.getCode();
-            byte[] content = rsp.getEntity().getContent().readAllBytes();
-
-            if (code >= 300) {
-                return new FetchFailure("Received HTTP " + code + " from " + domain);
-            }
-
-            ContentType contentType = ContentType.parse(rsp.getEntity().getContentType());
-            var html = DocumentBodyToString.getStringData(contentType, content);
-            return new FetchSuccess(Jsoup.parse(html));
-        }
-        catch (Exception e) {
-            EntityUtils.consumeQuietly(rsp.getEntity());
-            return new FetchFailure("Failed to read content from " + domain + ": " + e.getMessage());
-        }
-    }
 
 }

code/processes/new-domain-process/java/nu/marginalia/ndp/DomainNodeAllocator.java

@@ -16,6 +16,9 @@ public class DomainNodeAllocator {
 
     private final NodeConfigurationService nodeConfigurationService;
     private final HikariDataSource dataSource;
+    private final PriorityQueue<NodeCount> countPerNode = new PriorityQueue<>();
+
+    private volatile boolean initialized = false;
 
     private record NodeCount(int nodeId, int count)
         implements Comparable<NodeCount>
@@ -30,8 +33,6 @@ public class DomainNodeAllocator {
         }
     }
 
-    private final PriorityQueue<NodeCount> countPerNode = new PriorityQueue<>();
-    volatile boolean initialized = false;
 
     @Inject
     public DomainNodeAllocator(NodeConfigurationService nodeConfigurationService, HikariDataSource dataSource) {
@@ -43,6 +44,43 @@ public class DomainNodeAllocator {
                 .start(this::initialize);
     }
 
+    public synchronized int totalCount() {
+        ensureInitialized();
+        return countPerNode.stream().mapToInt(NodeCount::count).sum();
+    }
+
+    /** Returns the next node ID to assign a domain to.
+     * This method is synchronized to ensure thread safety when multiple threads are allocating domains.
+     * The node ID returned is guaranteed to be one of the viable nodes configured in the system.
+     */
+    public synchronized int nextNodeId() {
+        ensureInitialized();
+
+        // Synchronized is fine here as this is not a hot path
+        // (and PriorityBlockingQueue won't help since we're re-adding the same element with a new count all the time)
+
+        NodeCount allocation = countPerNode.remove();
+        countPerNode.add(allocation.incrementCount());
+        return allocation.nodeId();
+    }
+
+
+    private void ensureInitialized() {
+        if (initialized) return;
+
+        synchronized (this) {
+            while (!initialized) {
+                try {
+                    // Wait until the initialization is complete
+                    this.wait(1000);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException("DomainAllocator initialization interrupted", e);
+                }
+            }
+        }
+    }
+
 
     public void initialize() {
         if (initialized) return;
@@ -53,6 +91,9 @@ public class DomainNodeAllocator {
         for (var node : nodeConfigurationService.getAll()) {
             if (node.disabled())
                 continue;
+            if (!node.autoAssignDomains())
+                continue;
+
             if (node.profile().permitBatchCrawl())
                 viableNodes.add(node.node());
         }
@@ -89,39 +130,5 @@ public class DomainNodeAllocator {
         initialized = true;
     }
 
-    private void ensureInitialized() {
-        if (initialized) return;
 
-        synchronized (this) {
-            while (!initialized) {
-                try {
-                    // Wait until the initialization is complete
-                    this.wait(1000);
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    throw new RuntimeException("DomainAllocator initialization interrupted", e);
-                }
-            }
-        }
-    }
-
-    public synchronized int totalCount() {
-        ensureInitialized();
-        return countPerNode.stream().mapToInt(NodeCount::count).sum();
-    }
-
-    /** Returns the next node ID to assign a domain to.
-     * This method is synchronized to ensure thread safety when multiple threads are allocating domains.
-     * The node ID returned is guaranteed to be one of the viable nodes configured in the system.
-     */
-    public synchronized int nextNodeId() {
-        ensureInitialized();
-
-        // Synchronized is fine here as this is not a hot path
-        // (and PriorityBlockingQueue won't help since we're re-adding the same element with a new count all the time)
-
-        NodeCount allocation = countPerNode.remove();
-        countPerNode.add(allocation.incrementCount());
-        return allocation.nodeId();
-    }
 }

code/processes/new-domain-process/java/nu/marginalia/ndp/DomainTestingQueue.java

@@ -2,30 +2,41 @@ package nu.marginalia.ndp;
 
 import com.google.inject.Inject;
 import com.zaxxer.hikari.HikariDataSource;
+import it.unimi.dsi.fastutil.ints.Int2IntMap;
+import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import nu.marginalia.api.linkgraph.AggregateLinkGraphClient;
 import nu.marginalia.ndp.model.DomainToTest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.Connection;
+import java.sql.ResultSet;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class DomainTestingQueue {
-    private final ArrayBlockingQueue<DomainToTest> queue = new ArrayBlockingQueue<>(1000);
+    private static Logger logger = LoggerFactory.getLogger(DomainTestingQueue.class);
+
+    private final ArrayBlockingQueue<DomainToTest> queue = new ArrayBlockingQueue<>(2);
 
     // This will grow quite large, but should be manageable in memory, as theoretical maximum is around 100M domains,
     // order of 2 GB in memory.
     private final ConcurrentHashMap<String, Boolean> takenDomains = new ConcurrentHashMap<>();
 
     private final HikariDataSource dataSource;
+    private final AggregateLinkGraphClient linkGraphClient;
 
-    private static Logger logger = LoggerFactory.getLogger(DomainTestingQueue.class);
 
     @Inject
-    public DomainTestingQueue(HikariDataSource dataSource) {
+    public DomainTestingQueue(HikariDataSource dataSource,
+                              AggregateLinkGraphClient linkGraphClient
+                              ) {
         this.dataSource = dataSource;
+        this.linkGraphClient = linkGraphClient;
 
         Thread.ofPlatform()
                 .name("DomainTestingQueue::fetch()")
@@ -43,9 +54,10 @@ public class DomainTestingQueue {
                 SET STATE='ACCEPTED'
                 WHERE DOMAIN_ID=?
                 """);
-             var assigNodeStmt = conn.prepareStatement("""
+             var assignNodeStmt = conn.prepareStatement("""
                 UPDATE EC_DOMAIN SET NODE_AFFINITY=?
                 WHERE ID=?
+                AND EC_DOMAIN.NODE_AFFINITY < 0
                 """)
              )
         {
@@ -53,9 +65,9 @@ public class DomainTestingQueue {
             flagOkStmt.setInt(1, domain.domainId());
             flagOkStmt.executeUpdate();
 
-            assigNodeStmt.setInt(1, nodeId);
-            assigNodeStmt.setInt(2, domain.domainId());
-            assigNodeStmt.executeUpdate();
+            assignNodeStmt.setInt(1, nodeId);
+            assignNodeStmt.setInt(2, domain.domainId());
+            assignNodeStmt.executeUpdate();
             conn.commit();
         } catch (Exception e) {
             throw new RuntimeException("Failed to accept domain in database", e);
@@ -105,9 +117,14 @@ public class DomainTestingQueue {
                 }
 
                 if (domains.isEmpty()) {
-                    refreshQueue(conn);
+                    if (!refreshQueue(conn)) {
+                        throw new RuntimeException("No new domains found, aborting!");
+                    }
                 }
             }
+            catch (RuntimeException e) {
+                throw e; // Rethrow runtime exceptions to avoid wrapping them in another runtime exception
+            }
             catch (Exception e) {
                 throw new RuntimeException("Failed to fetch domains from database", e);
             }
@@ -124,25 +141,100 @@ public class DomainTestingQueue {
         }
     }
 
-    private void refreshQueue(Connection conn) {
+    private boolean refreshQueue(Connection conn) {
         logger.info("Refreshing domain queue in database");
-        try (var stmt = conn.createStatement()) {
-            conn.setAutoCommit(false);
-            logger.info("Revitalizing rejected domains");
 
-            // Revitalize rejected domains
-            stmt.executeUpdate("""
-                UPDATE NDP_NEW_DOMAINS
-                SET STATE='NEW'
-                WHERE NDP_NEW_DOMAINS.STATE = 'REJECTED'
-                AND DATE_ADD(TS_CHANGE, INTERVAL CHECK_COUNT DAY) > NOW()
-                """);
-            conn.commit();
+        Int2IntMap domainIdToCount = new Int2IntOpenHashMap();
+
+        // Load known domain IDs from the database to avoid inserting duplicates from NDP_NEW_DOMAINS
+        // or domains that are already assigned to a node
+        {
+            IntOpenHashSet knownIds = new IntOpenHashSet();
+
+            try (var stmt = conn.createStatement()) {
+                ResultSet rs = stmt.executeQuery("SELECT DOMAIN_ID FROM NDP_NEW_DOMAINS");
+                rs.setFetchSize(10_000);
+                while (rs.next()) {
+                    int domainId = rs.getInt("DOMAIN_ID");
+                    knownIds.add(domainId);
+                }
+
+                rs = stmt.executeQuery("SELECT ID FROM EC_DOMAIN WHERE NODE_AFFINITY>=0");
+                rs.setFetchSize(10_000);
+                while (rs.next()) {
+                    int domainId = rs.getInt("ID");
+                    knownIds.add(domainId);
+                }
+            } catch (Exception e) {
+                throw new RuntimeException("Failed to load known domain IDs from database", e);
+            }
+
+            // Ensure the link graph is ready before proceeding.  This is mainly necessary in a cold reboot
+            // of the entire system.
+            try {
+                logger.info("Waiting for link graph client to be ready...");
+                linkGraphClient.waitReady(Duration.ofHours(1));
+                logger.info("Link graph client is ready, fetching domain links...");
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+
+            // Fetch all domain links from the link graph and count by how many sources each dest domain is linked from
+            var iter = linkGraphClient.getAllDomainLinks().iterator();
+            while (iter.advance()) {
+                int dest = iter.dest();
+                if (!knownIds.contains(dest)) {
+                    domainIdToCount.mergeInt(dest, 1, (i, j) -> i + j);
+                }
+            }
+        }
+
+        boolean didInsert = false;
+
+        /* Insert new domains into NDP_NEW_DOMAINS table */
+        try (var insertStmt = conn.prepareStatement("""
+                INSERT IGNORE INTO NDP_NEW_DOMAINS (DOMAIN_ID, PRIORITY) VALUES (?, ?)
+                """)) {
+            conn.setAutoCommit(false);
+
+            int cnt = 0;
+            for (var entry : domainIdToCount.int2IntEntrySet()) {
+                int domainId = entry.getIntKey();
+                int count = entry.getIntValue();
+
+                insertStmt.setInt(1, domainId);
+                insertStmt.setInt(2, count);
+                insertStmt.addBatch();
+
+                if (++cnt >= 1000) {
+                    cnt = 0;
+                    insertStmt.executeBatch(); // Execute in batches to avoid memory issues
+                    conn.commit();
+                    didInsert = true;
+                }
+            }
+            if (cnt != 0) {
+                insertStmt.executeBatch(); // Execute any remaining batch
+                conn.commit();
+                didInsert = true;
+            }
 
             logger.info("Queue refreshed successfully");
         } catch (Exception e) {
             throw new RuntimeException("Failed to refresh queue in database", e);
         }
+
+        // Clean up NDP_NEW_DOMAINS table to remove any domains that are already in EC_DOMAIN
+        // This acts not only to clean up domains that we've flagged as ACCEPTED, but also to
+        // repair inconsistent states where domains might have incorrectly been added to NDP_NEW_DOMAINS
+        try (var stmt = conn.createStatement()) {
+            stmt.executeUpdate("DELETE FROM NDP_NEW_DOMAINS WHERE DOMAIN_ID IN (SELECT ID FROM EC_DOMAIN WHERE NODE_AFFINITY>=0)");
+        }
+        catch (Exception e) {
+            throw new RuntimeException("Failed to clean up NDP_NEW_DOMAINS", e);
+        }
+
+        return didInsert;
     }
 
 }

code/processes/new-domain-process/java/nu/marginalia/ndp/NdpMain.java

@@ -84,8 +84,23 @@ public class NdpMain extends ProcessMainClass {
                     hb.progress("Discovery Process", cnt, toInsertCount);
                 }
 
-                var nextDomain = domainTestingQueue.next();
-                threadPool.submit(() -> evaluateDomain(nextDomain));
+                final DomainToTest nextDomain = domainTestingQueue.next();
+                threadPool.submit(() -> {
+                    try {
+                        if (domainEvaluator.evaluateDomain(nextDomain.domainName())) {
+                            logger.info("Accepting: {}", nextDomain.domainName());
+                            domainCount.incrementAndGet();
+                            domainTestingQueue.accept(nextDomain, domainNodeAllocator.nextNodeId());
+                        } else {
+                            logger.info("Rejecting: {}", nextDomain.domainName());
+                            domainTestingQueue.reject(nextDomain);
+                        }
+                    }
+                    catch (Exception e) {
+                        domainTestingQueue.reject(nextDomain);
+                        logger.error("Error evaluating domain: " + nextDomain.domainId(), e);
+                    }
+                });
             }
         }
 
@@ -97,24 +112,6 @@ public class NdpMain extends ProcessMainClass {
 
     }
 
-
-    private void evaluateDomain(DomainToTest nextDomain) {
-        try {
-            if (domainEvaluator.evaluateDomain(nextDomain)) {
-                logger.info("Accepting: {}", nextDomain.domainName());
-                domainCount.incrementAndGet();
-                domainTestingQueue.accept(nextDomain, domainNodeAllocator.nextNodeId());
-            } else {
-                logger.info("Rejecting: {}", nextDomain.domainName());
-                domainTestingQueue.reject(nextDomain);
-            }
-        }
-        catch (Exception e) {
-            domainTestingQueue.reject(nextDomain);
-            logger.error("Error evaluating domain: " + nextDomain.domainId(), e);
-        }
-    }
-
     public static void main(String[] args) throws Exception {
         // Prevent Java from caching DNS lookups forever (filling up the system RAM as a result)
         Security.setProperty("networkaddress.cache.ttl" , "3600");

code/processes/new-domain-process/test/nu/marginalia/ndp/DomainEvaluatorTest.java

@@ -0,0 +1,29 @@
+package nu.marginalia.ndp;
+
+import nu.marginalia.coordination.LocalDomainCoordinator;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+import java.security.KeyManagementException;
+import java.security.NoSuchAlgorithmException;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class DomainEvaluatorTest {
+
+    @Tag("flaky") // Exclude from CI runs due to potential network issues
+    @Test
+    public void testSunnyDay() throws NoSuchAlgorithmException, KeyManagementException {
+        DomainEvaluator evaluator = new DomainEvaluator(new LocalDomainCoordinator());
+
+        // Should be a valid domain
+        assertTrue(evaluator.evaluateDomain("www.marginalia.nu"));
+
+        // Should be a redirect to www.marginalia.nu
+        assertFalse(evaluator.evaluateDomain("memex.marginalia.nu"));
+
+        // Should fail on Anubis
+        assertFalse(evaluator.evaluateDomain("marginalia-search.com"));
+    }
+}

code/processes/readme.md

@@ -28,6 +28,7 @@ the data generated by the loader.
 ## 5. Other Processes
 
 * Ping Process: The [ping-process](ping-process/) keeps track of the aliveness of websites, gathering fingerprint information about the security posture of the website, as well as DNS information.
+* New Domain Process (NDP): The [new-domain-process](new-domain-process/) evaluates new domains for inclusion in the search engine index.
 * Live-Crawling Process: The [live-crawling-process](live-crawling-process/) is a process that crawls websites in real-time based on RSS feeds, updating a smaller index with the latest content.
 
 ## Overview 

code/services-core/control-service/java/nu/marginalia/control/node/svc/ControlNodeService.java

@@ -280,6 +280,7 @@ public class ControlNodeService {
                     "on".equalsIgnoreCase(request.queryParams("autoClean")),
                     "on".equalsIgnoreCase(request.queryParams("includeInPrecession")),
                     "on".equalsIgnoreCase(request.queryParams("keepWarcs")),
+                    "on".equalsIgnoreCase(request.queryParams("autoAssignDomains")),
                     NodeProfile.valueOf(request.queryParams("profile")),
                     "on".equalsIgnoreCase(request.queryParams("disabled"))
             );

code/services-core/control-service/resources/templates/control/node/node-config.hdb

@@ -66,13 +66,23 @@
                 </div>
             </div>
 
+            <div class="form-check form-switch">
+                <input class="form-check-input" type="checkbox" role="switch" name="autoAssignDomains" {{#if config.autoAssignDomains}}checked{{/if}}>
+                <label class="form-check-label" for="autoClean">Auto-Assign Domains</label>
+
+                <div class="form-text">If true, the New Domain Process will assign new domains to this node and all other nodes with this setting enabled.
+                                       This is the default behavior, but can be overridden if you want one node with a specific manual domain assignment.
+                                       </div>
+            </div>
+
+<!--  This is not currently used, but may be in the future
             <div class="form-check form-switch">
                 <input class="form-check-input" type="checkbox" role="switch" name="includeInPrecession" {{#if config.includeInPrecession}}checked{{/if}}>
                 <label class="form-check-label" for="includeInPrecession">Include in crawling precession</label>
 
                 <div class="form-text">If true, this node will be included in the crawling precession.</div>
             </div>
-
+-->
             <div class="form-check form-switch">
                 <input class="form-check-input" type="checkbox" role="switch" name="keepWarcs" {{#if config.keepWarcs}}checked{{/if}}>
                 <label class="form-check-label" for="includeInPrecession">Keep WARC files during crawling</label>

code/services-core/control-service/resources/templates/control/node/partial-node-nav.hdb

@@ -13,14 +13,23 @@
     {{#unless node.profile.realtime}}
     <li class="nav-item dropdown">
         <a class="nav-link dropdown-toggle {{#if tab.actions}}active{{/if}}" data-bs-toggle="dropdown" href="#" role="button" aria-expanded="false">Actions</a>
-        {{#if node.profile.permitBatchCrawl}}
         <ul class="dropdown-menu">
+        {{#if node.profile.permitBatchCrawl}}
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=new-crawl">New Crawl</a></li>
             <li><hr class="dropdown-divider"></li>
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=process">Process Crawl Data</a></li>
+        {{/if}}
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=load">Load Processed Data</a></li>
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=repartition">Repartition Index</a></li>
             <li><hr class="dropdown-divider"></li>
+        {{#if node.profile.permitSideload}}
+            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-encyclopedia">Sideload Encyclopedia</a></li>
+            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-stackexchange">Sideload Stackexchange</a></li>
+            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-warc">Sideload WARC Files</a></li>
+            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-dirtree">Sideload Dirtree</a></li>
+            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-reddit">Sideload Reddit</a></li>
+            <li><hr class="dropdown-divider"></li>
+        {{/if}}
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=download-sample-data">Download Sample Crawl Data</a></li>
             <li><hr class="dropdown-divider"></li>
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=export-db-data">Export Database Data</a></li>
@@ -30,19 +39,6 @@
             <li><hr class="dropdown-divider"></li>
             <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=restore-backup">Restore Index Backup</a></li>
         </ul>
-        {{/if}}
-        {{#if node.profile.permitSideload}}
-        <ul class="dropdown-menu">
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-encyclopedia">Sideload Encyclopedia</a></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-stackexchange">Sideload Stackexchange</a></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-warc">Sideload WARC Files</a></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-dirtree">Sideload Dirtree</a></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=sideload-reddit">Sideload Reddit</a></li>
-            <li><hr class="dropdown-divider"></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=load">Load Processed Data</a></li>
-            <li><a class="dropdown-item" href="/nodes/{{node.id}}/actions?view=restore-backup">Restore Index Backup</a></li>
-        </ul>
-        {{/if}}
     </li>
     {{/unless}}
     <li class="nav-item">