Merge pull request #210 from MarginaliaSearch/ads-fingerprinting

Implement advertisement and popover identification based on DOM sample data

ROADMAP.md

@@ -48,10 +48,6 @@ filter for any API consumer.
 
 I've talked to the stract dev and he does not think it's a good idea to mimic their optics language, which is quite ad-hoc, but instead to work together to find some new common description language for this. 
 
-## Show favicons next to search results
-
-This is expected from search engines.  Basic proof of concept sketch of fetching this data has been done, but the feature is some way from being reality. 
-
 ## Specialized crawler for github
 
 One of the search engine's biggest limitations right now is that it does not index github at all.   A specialized crawler that fetches at least the readme.md would go a long way toward providing search capabilities in this domain.
@@ -66,6 +62,10 @@ The documents database probably should have some sort of flag indicating it's a
 PDF parsing is known to be a bit of a security liability so some thought needs to be put in
 that direction as well.
 
+## Show favicons next to search results (COMPLETED 2025-03)
+
+This is expected from search engines.  Basic proof of concept sketch of fetching this data has been done, but the feature is some way from being reality.
+
 ## Web Design Overhaul (COMPLETED 2025-01)
 
 The design is kinda clunky and hard to maintain, and needlessly outdated-looking.  

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(5, config.autoAssignDomains());
-            us.setBoolean(6, config.disabled());
+            us.setBoolean(6, config.keepWarcs());
-            us.setString(7, config.profile().name());
+            us.setBoolean(7, config.disabled());
-            us.setInt(8, config.node());
+            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();
+        return isBatchCrawl() || isMixed();
-    }
-    public boolean permitSideload() {
-        return isMixed() || isSideload();
     }
+    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_003__ping_domains__dsi_cert_details.sql

@@ -0,0 +1,7 @@
+-- Add additional summary columns to DOMAIN_SECURITY_INFORMATION table
+-- to make it easier to get more information about the SSL certificate's validity
+
+ALTER TABLE DOMAIN_SECURITY_INFORMATION ADD COLUMN SSL_CHAIN_VALID BOOLEAN DEFAULT NULL;
+ALTER TABLE DOMAIN_SECURITY_INFORMATION ADD COLUMN SSL_HOST_VALID BOOLEAN DEFAULT NULL;
+ALTER TABLE DOMAIN_SECURITY_INFORMATION ADD COLUMN SSL_DATE_VALID BOOLEAN DEFAULT NULL;
+OPTIMIZE TABLE DOMAIN_SECURITY_INFORMATION;

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

@@ -0,0 +1,5 @@
+-- Add additional summary columns to DOMAIN_SECURITY_EVENTS table
+-- to make it easier to make sense of certificate changes
+
+ALTER TABLE DOMAIN_SECURITY_EVENTS ADD COLUMN CHANGE_SCHEMA ENUM('NONE', 'HTTP_TO_HTTPS', 'HTTPS_TO_HTTP', 'UNKNOWN') NOT NULL DEFAULT 'UNKNOWN';
+OPTIMIZE TABLE DOMAIN_SECURITY_EVENTS;

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

@@ -0,0 +1,12 @@
+-- Table holding domains to be processed by the NDP in order to figure out whether to add them to
+-- be crawled.
+
+CREATE TABLE IF NOT EXISTS NDP_NEW_DOMAINS(
+    DOMAIN_ID INT NOT NULL PRIMARY KEY,
+    STATE ENUM ('NEW', 'ACCEPTED', 'REJECTED') NOT NULL DEFAULT 'NEW',
+    PRIORITY INT NOT NULL DEFAULT 0,
+    TS_CHANGE TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    CHECK_COUNT INT NOT NULL DEFAULT 0
+);
+
+CREATE INDEX IF NOT EXISTS NDP_NEW_DOMAINS__STATE_PRIORITY ON NDP_NEW_DOMAINS (STATE, PRIORITY DESC);

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/common/model/java/nu/marginalia/model/crawl/HtmlFeature.java

@@ -5,13 +5,15 @@ import java.util.Collection;
 public enum HtmlFeature {
     // Note, the first 32 of these features are bit encoded in the database
     // so be sure to keep anything that's potentially important toward the top
-    // of the list
+    // of the list; but adding new values will shift the encoded values and break
+    // binary compatibility!  Scroll down for a marker where you should add new values
+    // if they need to be accessible from IndexResultScoreCalculator!
 
     MEDIA( "special:media"),
     JS("special:scripts"),
     AFFILIATE_LINK( "special:affiliate"),
     TRACKING("special:tracking"),
-    TRACKING_ADTECH("special:ads"), // We'll call this ads for now
+    TRACKING_ADTECH("special:adtech"),
 
     KEBAB_CASE_URL("special:kcurl"), // https://www.example.com/urls-that-look-like-this/
     LONG_URL("special:longurl"),
@@ -30,6 +32,15 @@ public enum HtmlFeature {
 
     PDF("format:pdf"),
 
+    POPOVER("special:popover"),
+    CONSENT("special:consent"),
+    SHORT_DOCUMENT("special:shorty"),
+    THIRD_PARTY_REQUESTS("special:3pr"),
+
+    // Here!  It is generally safe to add additional values here without
+    // disrupting the encoded values used by the DocumentValuator
+    // class in the index!
+
     /** For fingerprinting and ranking */
     OPENGRAPH("special:opengraph"),
     OPENGRAPH_IMAGE("special:opengraph:image"),
@@ -67,6 +78,7 @@ public enum HtmlFeature {
 
     S3_FEATURE("special:s3"),
 
+    MISSING_DOM_SAMPLE("special:nosample"),
     UNKNOWN("special:uncategorized");
 
 
@@ -93,6 +105,8 @@ public enum HtmlFeature {
     }
 
     public int getFeatureBit() {
+        if (getClass().desiredAssertionStatus() && ordinal() >= 32)
+            throw new IllegalStateException("Attempting to extract feature bit of " + name() + ", with ordinal " + ordinal());
         return (1<< ordinal());
     }
 }

code/common/service/java/nu/marginalia/service/client/GrpcChannelPoolFactory.java

@@ -13,6 +13,7 @@ import nu.marginalia.service.discovery.property.ServicePartition;
 import nu.marginalia.util.NamedExecutorFactory;
 
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 import java.util.function.Function;
 
 @Singleton
@@ -20,10 +21,15 @@ public class GrpcChannelPoolFactory {
 
     private final NodeConfigurationWatcher nodeConfigurationWatcher;
     private final ServiceRegistryIf serviceRegistryIf;
-    private static final Executor executor = NamedExecutorFactory.createFixed("gRPC-Channel-Pool",
+
-            Math.clamp(Runtime.getRuntime().availableProcessors() / 2, 2, 32));
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
-    private static final Executor offloadExecutor = NamedExecutorFactory.createFixed("gRPC-Offload-Pool",
+
-            Math.clamp(Runtime.getRuntime().availableProcessors() / 2, 2, 32));
+    private static final Executor executor = useLoom
+            ? Executors.newVirtualThreadPerTaskExecutor()
+            : NamedExecutorFactory.createFixed("gRPC-Channel-Pool", Math.clamp(Runtime.getRuntime().availableProcessors() / 2, 2, 32));
+    private static final Executor offloadExecutor =  useLoom
+            ? Executors.newVirtualThreadPerTaskExecutor()
+            : NamedExecutorFactory.createFixed("gRPC-Offload-Pool", Math.clamp(Runtime.getRuntime().availableProcessors() / 2, 2, 32));
 
     @Inject
     public GrpcChannelPoolFactory(NodeConfigurationWatcher nodeConfigurationWatcher,

code/common/service/java/nu/marginalia/service/client/GrpcSingleNodeChannelPool.java

@@ -2,6 +2,7 @@ package nu.marginalia.service.client;
 
 import com.google.common.collect.Sets;
 import io.grpc.ManagedChannel;
+import io.grpc.StatusRuntimeException;
 import nu.marginalia.service.discovery.ServiceRegistryIf;
 import nu.marginalia.service.discovery.monitor.ServiceChangeMonitor;
 import nu.marginalia.service.discovery.property.PartitionTraits;
@@ -206,6 +207,11 @@ public class GrpcSingleNodeChannelPool<STUB> extends ServiceChangeMonitor {
         }
 
         for (var e : exceptions) {
+            if (e instanceof StatusRuntimeException se) {
+                throw se; // Re-throw SRE as-is
+            }
+
+            // If there are other exceptions, log them
             logger.error(grpcMarker, "Failed to call service {}", serviceKey, e);
         }
 

code/common/service/java/nu/marginalia/service/discovery/ServiceRegistryIf.java

@@ -5,12 +5,10 @@ import nu.marginalia.service.discovery.monitor.ServiceChangeMonitor;
 import nu.marginalia.service.discovery.monitor.ServiceMonitorIf;
 import nu.marginalia.service.discovery.property.ServiceEndpoint;
 import nu.marginalia.service.discovery.property.ServiceKey;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
 
-import java.util.Collection;
 import java.util.List;
 import java.util.UUID;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 
 import static nu.marginalia.service.discovery.property.ServiceEndpoint.InstanceAddress;
 
@@ -66,6 +64,6 @@ public interface ServiceRegistryIf {
 
     void registerProcess(String processName, int nodeId);
     void deregisterProcess(String processName, int nodeId);
-    void watchProcess(String processName, int nodeId, Consumer<Boolean> callback) throws Exception;
+
-    void watchProcessAnyNode(String processName, Collection<Integer> nodes, BiConsumer<Boolean, Integer> callback) throws Exception;
+    InterProcessSemaphoreV2 getSemaphore(String name, int permits) throws Exception;
 }

code/common/service/java/nu/marginalia/service/discovery/ZkServiceRegistry.java

@@ -6,6 +6,7 @@ import nu.marginalia.service.discovery.monitor.ServiceMonitorIf;
 import nu.marginalia.service.discovery.property.ServiceEndpoint;
 import nu.marginalia.service.discovery.property.ServiceKey;
 import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
 import org.apache.curator.utils.ZKPaths;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Watcher;
@@ -13,10 +14,11 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.charset.StandardCharsets;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.UUID;
 import java.util.concurrent.TimeUnit;
-import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 
 import static nu.marginalia.service.discovery.property.ServiceEndpoint.InstanceAddress;
 
@@ -283,60 +285,12 @@ public class ZkServiceRegistry implements ServiceRegistryIf {
     }
 
     @Override
-    public void watchProcess(String processName, int nodeId, Consumer<Boolean> callback) throws Exception {
+    public InterProcessSemaphoreV2 getSemaphore(String name, int permits) {
-        String path = "/process-locks/" + processName + "/" + nodeId;
+        if (stopped)
+            throw new IllegalStateException("Service registry is stopped, cannot get semaphore " + name);
 
-        // first check if the path exists and call the callback accordingly
+        String path = "/semaphores/" + name;
-
+        return new InterProcessSemaphoreV2(curatorFramework, path, permits);
-        if (curatorFramework.checkExists().forPath(path) != null) {
-            callback.accept(true);
-        }
-        else {
-            callback.accept(false);
-        }
-
-        curatorFramework.watchers().add()
-                .usingWatcher((Watcher) change -> {
-                    Watcher.Event.EventType type = change.getType();
-
-                    if (type == Watcher.Event.EventType.NodeCreated) {
-                        callback.accept(true);
-                    }
-                    if (type == Watcher.Event.EventType.NodeDeleted) {
-                        callback.accept(false);
-                    }
-                })
-                .forPath(path);
-
-    }
-
-    @Override
-    public void watchProcessAnyNode(String processName, Collection<Integer> nodes, BiConsumer<Boolean, Integer> callback) throws Exception {
-
-        for (int node : nodes) {
-            String path = "/process-locks/" + processName + "/" + node;
-
-            // first check if the path exists and call the callback accordingly
-            if (curatorFramework.checkExists().forPath(path) != null) {
-                callback.accept(true, node);
-            }
-            else {
-                callback.accept(false, node);
-            }
-
-            curatorFramework.watchers().add()
-                    .usingWatcher((Watcher) change -> {
-                        Watcher.Event.EventType type = change.getType();
-
-                        if (type == Watcher.Event.EventType.NodeCreated) {
-                            callback.accept(true, node);
-                        }
-                        if (type == Watcher.Event.EventType.NodeDeleted) {
-                            callback.accept(false, node);
-                        }
-                    })
-                    .forPath(path);
-        }
     }
 
     /* Exposed for tests */

code/common/service/java/nu/marginalia/service/server/GrpcServer.java

@@ -1,9 +1,9 @@
 package nu.marginalia.service.server;
 
 import io.grpc.Server;
-import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.netty.NettyServerBuilder;
-import io.grpc.netty.shaded.io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
-import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
 import nu.marginalia.service.discovery.ServiceRegistryIf;
 import nu.marginalia.service.discovery.property.ServiceKey;
 import nu.marginalia.service.discovery.property.ServicePartition;
@@ -13,9 +13,14 @@ import nu.marginalia.util.NamedExecutorFactory;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 public class GrpcServer {
     private final Server server;
+
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
+
     public GrpcServer(ServiceConfiguration config,
                       ServiceRegistryIf serviceRegistry,
                       ServicePartition partition,
@@ -26,13 +31,19 @@ public class GrpcServer {
         int nThreads = Math.clamp(Runtime.getRuntime().availableProcessors() / 2, 2, 16);
 
         // Start the gRPC server
+
+        ExecutorService workExecutor = useLoom ?
+                Executors.newVirtualThreadPerTaskExecutor() :
+                NamedExecutorFactory.createFixed("nettyExecutor", nThreads);
+
         var grpcServerBuilder = NettyServerBuilder.forAddress(new InetSocketAddress(config.bindAddress(), port))
-                .executor(NamedExecutorFactory.createFixed("nettyExecutor", nThreads))
+                .executor(workExecutor)
                 .workerEventLoopGroup(new NioEventLoopGroup(nThreads, NamedExecutorFactory.createFixed("Worker-ELG", nThreads)))
                 .bossEventLoopGroup(new NioEventLoopGroup(nThreads, NamedExecutorFactory.createFixed("Boss-ELG", nThreads)))
                 .channelType(NioServerSocketChannel.class);
 
         for (var grpcService : grpcServices) {
+
             if (!grpcService.shouldRegisterService()) {
                 continue;
             }

code/common/service/java/nu/marginalia/service/server/JoobyService.java

@@ -125,8 +125,7 @@ public class JoobyService {
         // Set a cap on the number of worker threads, as Jooby's default value does not seem to consider
         // multi-tenant servers with high thread counts, and spins up an exorbitant number of threads in that
         // scenario
-        options.setWorkerThreads(Math.min(128, options.getWorkerThreads()));
+        options.setWorkerThreads(Math.min(16, options.getWorkerThreads()));
-
 
         jooby.setServerOptions(options);
 

code/common/service/resources/log4j2-json.xml

@@ -7,6 +7,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </Console>
         <Console name="ProcessConsole" target="SYSTEM_OUT">
@@ -23,6 +24,7 @@
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="PROCESS" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
             <SizeBasedTriggeringPolicy size="10MB" />
         </RollingFile>
@@ -36,6 +38,16 @@
                 <MarkerFilter marker="CRAWLER" onMatch="ALLOW" onMismatch="DENY" />
             </Filters>
         </RollingFile>
+        <RollingFile name="LogToFile" fileName="${env:WMSA_LOG_DIR:-/var/log/wmsa}/converter-audit-${env:WMSA_SERVICE_NODE:-0}.log" filePattern="/var/log/wmsa/converter-audit-${env:WMSA_SERVICE_NODE:-0}-log-%d{MM-dd-yy-HH-mm-ss}-%i.log.gz"
+                     ignoreExceptions="false">
+            <PatternLayout>
+                <Pattern>%d{yyyy-MM-dd HH:mm:ss,SSS}: %msg{nolookups}%n</Pattern>
+            </PatternLayout>
+            <SizeBasedTriggeringPolicy size="100MB" />
+            <Filters>
+                <MarkerFilter marker="CONVERTER" onMatch="ALLOW" onMismatch="DENY" />
+            </Filters>
+        </RollingFile>
     </Appenders>
     <Loggers>
         <Logger name="org.apache.zookeeper" level="WARN" />

code/common/service/resources/log4j2-prod.xml

@@ -8,6 +8,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </Console>
         <Console name="ConsoleWarn" target="SYSTEM_OUT">
@@ -18,6 +19,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </Console>
         <Console name="ConsoleError" target="SYSTEM_OUT">
@@ -28,6 +30,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </Console>
         <Console name="ConsoleFatal" target="SYSTEM_OUT">
@@ -38,6 +41,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </Console>
         <Console name="ProcessConsole" target="SYSTEM_OUT">
@@ -57,6 +61,7 @@
                 <MarkerFilter marker="QUERY" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="HTTP" onMatch="DENY" onMismatch="NEUTRAL" />
                 <MarkerFilter marker="CRAWLER" onMatch="DENY" onMismatch="NEUTRAL" />
+                <MarkerFilter marker="CONVERTER" onMatch="DENY" onMismatch="NEUTRAL" />
             </Filters>
         </RollingFile>
         <RollingFile name="LogToFile" fileName="${env:WMSA_LOG_DIR:-/var/log/wmsa}/crawler-audit-${env:WMSA_SERVICE_NODE:-0}.log" filePattern="/var/log/wmsa/crawler-audit-${env:WMSA_SERVICE_NODE:-0}-log-%d{MM-dd-yy-HH-mm-ss}-%i.log.gz"
@@ -69,6 +74,16 @@
                 <MarkerFilter marker="CRAWLER" onMatch="ALLOW" onMismatch="DENY" />
             </Filters>
         </RollingFile>
+        <RollingFile name="LogToFile" fileName="${env:WMSA_LOG_DIR:-/var/log/wmsa}/converter-audit-${env:WMSA_SERVICE_NODE:-0}.log" filePattern="/var/log/wmsa/converter-audit-${env:WMSA_SERVICE_NODE:-0}-log-%d{MM-dd-yy-HH-mm-ss}-%i.log.gz"
+                     ignoreExceptions="false">
+            <PatternLayout>
+                <Pattern>%d{yyyy-MM-dd HH:mm:ss,SSS}: %msg{nolookups}%n</Pattern>
+            </PatternLayout>
+            <SizeBasedTriggeringPolicy size="100MB" />
+            <Filters>
+                <MarkerFilter marker="CONVERTER" onMatch="ALLOW" onMismatch="DENY" />
+            </Filters>
+        </RollingFile>
     </Appenders>
     <Loggers>
         <Logger name="org.apache.zookeeper" level="WARN" />

code/execution/api/java/nu/marginalia/executor/client/ExecutorClient.java

@@ -9,6 +9,7 @@ import nu.marginalia.executor.storage.FileStorageFile;
 import nu.marginalia.executor.upload.UploadDirContents;
 import nu.marginalia.executor.upload.UploadDirItem;
 import nu.marginalia.functions.execution.api.*;
+import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.service.ServiceId;
 import nu.marginalia.service.client.GrpcChannelPoolFactory;
 import nu.marginalia.service.client.GrpcMultiNodeChannelPool;
@@ -25,27 +26,37 @@ import java.net.URISyntaxException;
 import java.net.URL;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.List;
 
 import static nu.marginalia.functions.execution.api.ExecutorApiGrpc.ExecutorApiBlockingStub;
 
 @Singleton
 public class ExecutorClient {
+    private final MqPersistence persistence;
     private final GrpcMultiNodeChannelPool<ExecutorApiBlockingStub> channelPool;
     private static final Logger logger = LoggerFactory.getLogger(ExecutorClient.class);
     private final ServiceRegistryIf registry;
 
     @Inject
     public ExecutorClient(ServiceRegistryIf registry,
+                          MqPersistence persistence,
                           GrpcChannelPoolFactory grpcChannelPoolFactory)
     {
         this.registry = registry;
+        this.persistence = persistence;
         this.channelPool = grpcChannelPoolFactory
                 .createMulti(
                         ServiceKey.forGrpcApi(ExecutorApiGrpc.class, ServicePartition.multi()),
                         ExecutorApiGrpc::newBlockingStub);
     }
 
+    private long createTrackingTokenMsg(String task, int node, Duration ttl) throws Exception {
+        return persistence.sendNewMessage("task-tracking[" + node + "]", "export-client", null, task, "", ttl);
+    }
+
+
+
     public void startFsm(int node, String actorName) {
         channelPool.call(ExecutorApiBlockingStub::startFsm)
                 .forNode(node)
@@ -96,6 +107,16 @@ public class ExecutorClient {
                         .build());
     }
 
+    public long updateNsfwFilters() throws Exception {
+        long msgId = createTrackingTokenMsg("nsfw-filters", 1, Duration.ofHours(6));
+
+        channelPool.call(ExecutorApiBlockingStub::updateNsfwFilters)
+                .forNode(1)
+                .run(RpcUpdateNsfwFilters.newBuilder().setMsgId(msgId).build());
+
+        return msgId;
+    }
+
     public ActorRunStates getActorStates(int node) {
         try {
             var rs = channelPool.call(ExecutorApiBlockingStub::getActorStates)

code/execution/api/src/main/protobuf/executor-api.proto

@@ -18,6 +18,8 @@ service ExecutorApi {
   rpc calculateAdjacencies(Empty) returns (Empty) {}
   rpc restoreBackup(RpcFileStorageId) returns (Empty) {}
 
+  rpc updateNsfwFilters(RpcUpdateNsfwFilters) returns (Empty) {}
+
   rpc restartExecutorService(Empty) returns (Empty) {}
 }
 
@@ -66,6 +68,9 @@ message RpcExportRequest {
   int64 fileStorageId = 1;
   int64 msgId = 2;
 }
+message RpcUpdateNsfwFilters {
+  int64 msgId = 1;
+}
 message RpcFileStorageIdWithDomainName {
   int64 fileStorageId = 1;
   string targetDomainName = 2;

code/execution/build.gradle

@@ -20,6 +20,7 @@ dependencies {
     implementation project(':code:processes:live-crawling-process')
     implementation project(':code:processes:loading-process')
     implementation project(':code:processes:ping-process')
+    implementation project(':code:processes:new-domain-process')
     implementation project(':code:processes:converting-process')
     implementation project(':code:processes:index-constructor-process')
 
@@ -41,7 +42,6 @@ dependencies {
     implementation project(':code:functions:nsfw-domain-filter')
     implementation project(':code:execution:api')
 
-    implementation project(':code:processes:crawling-process:model')
     implementation project(':code:processes:crawling-process:model')
     implementation project(':code:processes:crawling-process:ft-link-parser')
     implementation project(':code:index:index-journal')

code/execution/java/nu/marginalia/actor/ActorApi.java

@@ -2,10 +2,11 @@ package nu.marginalia.actor;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
-import nu.marginalia.functions.execution.api.*;
+import nu.marginalia.functions.execution.api.RpcFsmName;
+import nu.marginalia.functions.execution.api.RpcProcessId;
 import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.persistence.MqPersistence;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -14,18 +15,18 @@ import spark.Spark;
 @Singleton
 public class ActorApi {
     private final ExecutorActorControlService actors;
-    private final ProcessService processService;
+    private final ProcessSpawnerService processSpawnerService;
     private final MqPersistence mqPersistence;
     private final ServiceConfiguration serviceConfiguration;
     private final Logger logger = LoggerFactory.getLogger(getClass());
     @Inject
     public ActorApi(ExecutorActorControlService actors,
-                    ProcessService processService,
+                    ProcessSpawnerService processSpawnerService,
                     MqPersistence mqPersistence,
                     ServiceConfiguration serviceConfiguration)
     {
         this.actors = actors;
-        this.processService = processService;
+        this.processSpawnerService = processSpawnerService;
         this.mqPersistence = mqPersistence;
         this.serviceConfiguration = serviceConfiguration;
     }
@@ -43,7 +44,7 @@ public class ActorApi {
     }
 
     public Object stopProcess(RpcProcessId processId) {
-        ProcessService.ProcessId id = ProcessService.translateExternalIdBase(processId.getProcessId());
+        ProcessSpawnerService.ProcessId id = ProcessSpawnerService.translateExternalIdBase(processId.getProcessId());
 
         try {
             String inbox = id.name().toLowerCase() + ":" + serviceConfiguration.node();
@@ -60,7 +61,7 @@ public class ActorApi {
                 }
 
             }
-            processService.kill(id);
+            processSpawnerService.kill(id);
         }
         catch (Exception ex) {
             logger.error("Failed to stop process {}", id, ex);

code/execution/java/nu/marginalia/actor/ExecutorActor.java

@@ -6,7 +6,7 @@ import java.util.Set;
 
 public enum ExecutorActor {
     PREC_EXPORT_ALL(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
-    SYNC_NSFW_LISTS(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
+    UPDATE_NSFW_LISTS(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED, NodeProfile.SIDELOAD, NodeProfile.REALTIME),
 
     CRAWL(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
     RECRAWL(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
@@ -14,6 +14,7 @@ public enum ExecutorActor {
     PROC_CRAWLER_SPAWNER(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
     PROC_PING_SPAWNER(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED, NodeProfile.REALTIME),
     PROC_EXPORT_TASKS_SPAWNER(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
+    PROC_NDP_SPAWNER(NodeProfile.MIXED, NodeProfile.REALTIME),
     ADJACENCY_CALCULATION(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
     EXPORT_DATA(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),
     EXPORT_SEGMENTATION_MODEL(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED),

code/execution/java/nu/marginalia/actor/ExecutorActorControlService.java

@@ -49,6 +49,7 @@ public class ExecutorActorControlService {
                                        RecrawlSingleDomainActor recrawlSingleDomainActor,
                                        RestoreBackupActor restoreBackupActor,
                                        ConverterMonitorActor converterMonitorFSM,
+                                       NdpMonitorActor ndpMonitorActor,
                                        PingMonitorActor pingMonitorActor,
                                        CrawlerMonitorActor crawlerMonitorActor,
                                        LiveCrawlerMonitorActor liveCrawlerMonitorActor,
@@ -93,7 +94,7 @@ public class ExecutorActorControlService {
         register(ExecutorActor.PROC_PING_SPAWNER, pingMonitorActor);
         register(ExecutorActor.PROC_LIVE_CRAWL_SPAWNER, liveCrawlerMonitorActor);
         register(ExecutorActor.PROC_EXPORT_TASKS_SPAWNER, exportTasksMonitorActor);
-
+        register(ExecutorActor.PROC_NDP_SPAWNER, ndpMonitorActor);
         register(ExecutorActor.MONITOR_PROCESS_LIVENESS, processMonitorFSM);
         register(ExecutorActor.MONITOR_FILE_STORAGE, fileStorageMonitorActor);
 
@@ -112,7 +113,7 @@ public class ExecutorActorControlService {
         register(ExecutorActor.UPDATE_RSS, updateRssActor);
 
         register(ExecutorActor.MIGRATE_CRAWL_DATA, migrateCrawlDataActor);
-        register(ExecutorActor.SYNC_NSFW_LISTS, updateNsfwFiltersActor);
+        register(ExecutorActor.UPDATE_NSFW_LISTS, updateNsfwFiltersActor);
 
         if (serviceConfiguration.node() == 1) {
             register(ExecutorActor.PREC_EXPORT_ALL, exportAllPrecessionActor);

code/execution/java/nu/marginalia/actor/monitor/AbstractProcessSpawnerActor.java

@@ -4,11 +4,14 @@ import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.prototype.RecordActorPrototype;
-import nu.marginalia.actor.state.*;
+import nu.marginalia.actor.state.ActorResumeBehavior;
-import nu.marginalia.mq.persistence.MqMessageHandlerRegistry;
+import nu.marginalia.actor.state.ActorStep;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.actor.state.Resume;
+import nu.marginalia.actor.state.Terminal;
 import nu.marginalia.mq.MqMessageState;
+import nu.marginalia.mq.persistence.MqMessageHandlerRegistry;
 import nu.marginalia.mq.persistence.MqPersistence;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -24,13 +27,13 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class AbstractProcessSpawnerActor extends RecordActorPrototype {
 
     private final MqPersistence persistence;
-    private final ProcessService processService;
+    private final ProcessSpawnerService processSpawnerService;
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     public static final int MAX_ATTEMPTS = 3;
     private final String inboxName;
-    private final ProcessService.ProcessId processId;
+    private final ProcessSpawnerService.ProcessId processId;
     private final ExecutorService executorService = Executors.newSingleThreadExecutor();
     private final int node;
 
@@ -50,7 +53,7 @@ public class AbstractProcessSpawnerActor extends RecordActorPrototype {
                 for (;;) {
                     var messages = persistence.eavesdrop(inboxName, 1);
 
-                    if (messages.isEmpty() && !processService.isRunning(processId)) {
+                    if (messages.isEmpty() && !processSpawnerService.isRunning(processId)) {
                         synchronized (processId) {
                             processId.wait(5000);
                         }
@@ -92,7 +95,7 @@ public class AbstractProcessSpawnerActor extends RecordActorPrototype {
                 catch (InterruptedException ex) {
                     // We get this exception when the process is cancelled by the user
 
-                    processService.kill(processId);
+                    processSpawnerService.kill(processId);
                     setCurrentMessageToDead();
 
                     yield new Aborted();
@@ -112,13 +115,13 @@ public class AbstractProcessSpawnerActor extends RecordActorPrototype {
     public AbstractProcessSpawnerActor(Gson gson,
                                        ServiceConfiguration configuration,
                                        MqPersistence persistence,
-                                       ProcessService processService,
+                                       ProcessSpawnerService processSpawnerService,
                                        String inboxName,
-                                       ProcessService.ProcessId processId) {
+                                       ProcessSpawnerService.ProcessId processId) {
         super(gson);
         this.node = configuration.node();
         this.persistence = persistence;
-        this.processService = processService;
+        this.processSpawnerService = processSpawnerService;
         this.inboxName = inboxName + ":" + node;
         this.processId = processId;
     }
@@ -149,7 +152,7 @@ public class AbstractProcessSpawnerActor extends RecordActorPrototype {
             // Run this call in a separate thread so that this thread can be interrupted waiting for it
             executorService.submit(() -> {
                 try {
-                    processService.trigger(processId);
+                    processSpawnerService.trigger(processId);
                 } catch (Exception e) {
                     logger.warn("Error in triggering process", e);
                     error.set(true);

code/execution/java/nu/marginalia/actor/proc/ConverterMonitorActor.java

@@ -4,9 +4,9 @@ import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -17,13 +17,13 @@ public class ConverterMonitorActor extends AbstractProcessSpawnerActor {
     public ConverterMonitorActor(Gson gson,
                                  ServiceConfiguration configuration,
                                  MqPersistence persistence,
-                                 ProcessService processService) {
+                                 ProcessSpawnerService processSpawnerService) {
         super(gson,
                 configuration,
                 persistence,
-                processService,
+                processSpawnerService,
                 ProcessInboxNames.CONVERTER_INBOX,
-                ProcessService.ProcessId.CONVERTER);
+                ProcessSpawnerService.ProcessId.CONVERTER);
     }
 
 

code/execution/java/nu/marginalia/actor/proc/CrawlerMonitorActor.java

@@ -4,9 +4,9 @@ import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -16,13 +16,13 @@ public class CrawlerMonitorActor extends AbstractProcessSpawnerActor {
     public CrawlerMonitorActor(Gson gson,
                                ServiceConfiguration configuration,
                                MqPersistence persistence,
-                               ProcessService processService) {
+                               ProcessSpawnerService processSpawnerService) {
         super(gson,
                 configuration,
                 persistence,
-                processService,
+                processSpawnerService,
                 ProcessInboxNames.CRAWLER_INBOX,
-                ProcessService.ProcessId.CRAWLER);
+                ProcessSpawnerService.ProcessId.CRAWLER);
     }
 
 

code/execution/java/nu/marginalia/actor/proc/ExportTaskMonitorActor.java

@@ -6,7 +6,7 @@ import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -16,13 +16,13 @@ public class ExportTaskMonitorActor extends AbstractProcessSpawnerActor {
     public ExportTaskMonitorActor(Gson gson,
                                   ServiceConfiguration configuration,
                                   MqPersistence persistence,
-                                  ProcessService processService) {
+                                  ProcessSpawnerService processSpawnerService) {
         super(gson,
                 configuration,
                 persistence,
-                processService,
+                processSpawnerService,
                 ProcessInboxNames.EXPORT_TASK_INBOX,
-                ProcessService.ProcessId.EXPORT_TASKS);
+                ProcessSpawnerService.ProcessId.EXPORT_TASKS);
     }
 
 

code/execution/java/nu/marginalia/actor/proc/IndexConstructorMonitorActor.java

@@ -4,9 +4,9 @@ import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -17,13 +17,13 @@ public class IndexConstructorMonitorActor extends AbstractProcessSpawnerActor {
     public IndexConstructorMonitorActor(Gson gson,
                                         ServiceConfiguration configuration,
                                         MqPersistence persistence,
-                                        ProcessService processService) {
+                                        ProcessSpawnerService processSpawnerService) {
         super(gson,
                 configuration,
                 persistence,
-                processService,
+                processSpawnerService,
                 ProcessInboxNames.INDEX_CONSTRUCTOR_INBOX,
-                ProcessService.ProcessId.INDEX_CONSTRUCTOR);
+                ProcessSpawnerService.ProcessId.INDEX_CONSTRUCTOR);
     }
 
 

code/execution/java/nu/marginalia/actor/proc/LiveCrawlerMonitorActor.java

@@ -6,7 +6,7 @@ import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -16,13 +16,13 @@ public class LiveCrawlerMonitorActor extends AbstractProcessSpawnerActor {
     public LiveCrawlerMonitorActor(Gson gson,
                                    ServiceConfiguration configuration,
                                    MqPersistence persistence,
-                                   ProcessService processService) {
+                                   ProcessSpawnerService processSpawnerService) {
         super(gson,
                 configuration,
                 persistence,
-                processService,
+                processSpawnerService,
                 ProcessInboxNames.LIVE_CRAWLER_INBOX,
-                ProcessService.ProcessId.LIVE_CRAWLER);
+                ProcessSpawnerService.ProcessId.LIVE_CRAWLER);
     }
 
 

code/execution/java/nu/marginalia/actor/proc/LoaderMonitorActor.java

@@ -4,9 +4,9 @@ import com.google.gson.Gson;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 @Singleton
@@ -17,13 +17,13 @@ public class LoaderMonitorActor extends AbstractProcessSpawnerActor {
     public LoaderMonitorActor(Gson gson,
                               ServiceConfiguration configuration,
                               MqPersistence persistence,
-                              ProcessService processService) {
+                              ProcessSpawnerService processSpawnerService) {
 
         super(gson,
                 configuration,
-                persistence, processService,
+                persistence, processSpawnerService,
                 ProcessInboxNames.LOADER_INBOX,
-                ProcessService.ProcessId.LOADER);
+                ProcessSpawnerService.ProcessId.LOADER);
     }
 
 }

code/execution/java/nu/marginalia/actor/proc/NdpMonitorActor.java

@@ -0,0 +1,29 @@
+package nu.marginalia.actor.proc;
+
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import nu.marginalia.actor.monitor.AbstractProcessSpawnerActor;
+import nu.marginalia.mq.persistence.MqPersistence;
+import nu.marginalia.mqapi.ProcessInboxNames;
+import nu.marginalia.process.ProcessSpawnerService;
+import nu.marginalia.service.module.ServiceConfiguration;
+
+@Singleton
+public class NdpMonitorActor extends AbstractProcessSpawnerActor {
+
+    @Inject
+    public NdpMonitorActor(Gson gson,
+                                   ServiceConfiguration configuration,
+                                   MqPersistence persistence,
+                                   ProcessSpawnerService processSpawnerService) {
+        super(gson,
+                configuration,
+                persistence,
+                processSpawnerService,
+                ProcessInboxNames.NDP_INBOX,
+                ProcessSpawnerService.ProcessId.NDP);
+    }
+
+
+}

code/execution/java/nu/marginalia/actor/proc/PingMonitorActor.java

@@ -13,35 +13,35 @@ import nu.marginalia.mq.persistence.MqMessageHandlerRegistry;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.ProcessInboxNames;
 import nu.marginalia.mqapi.ping.PingRequest;
-import nu.marginalia.nodecfg.NodeConfigurationService;
+import nu.marginalia.process.ProcessSpawnerService;
-import nu.marginalia.nodecfg.model.NodeProfile;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.service.module.ServiceConfiguration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.sql.SQLException;
-import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+
+// Unlike other monitor actors, the ping monitor will not merely wait for a request
+// to be sent, but send one itself,  hence we can't extend AbstractProcessSpawnerActor
+// but have to reimplement a lot of the same logic ourselves.
 @Singleton
 public class PingMonitorActor extends RecordActorPrototype {
 
     private final MqPersistence persistence;
-    private final ProcessService processService;
+    private final ProcessSpawnerService processSpawnerService;
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     public static final int MAX_ATTEMPTS = 3;
     private final String inboxName;
-    private final ProcessService.ProcessId processId;
+    private final ProcessSpawnerService.ProcessId processId;
     private final ExecutorService executorService = Executors.newSingleThreadExecutor();
     private final int node;
-    private final boolean isPrimaryNode;
     private final Gson gson;
 
     public record Initial() implements ActorStep {}
@@ -56,8 +56,7 @@ public class PingMonitorActor extends RecordActorPrototype {
     public ActorStep transition(ActorStep self) throws Exception {
         return switch (self) {
             case Initial i -> {
-                PingRequest request = new PingRequest(isPrimaryNode ? "primary": "secondary");
+                PingRequest request = new PingRequest();
-
                 persistence.sendNewMessage(inboxName, null, null,
                         "PingRequest",
                         gson.toJson(request),
@@ -69,7 +68,7 @@ public class PingMonitorActor extends RecordActorPrototype {
                 for (;;) {
                     var messages = persistence.eavesdrop(inboxName, 1);
 
-                    if (messages.isEmpty() && !processService.isRunning(processId)) {
+                    if (messages.isEmpty() && !processSpawnerService.isRunning(processId)) {
                         synchronized (processId) {
                             processId.wait(5000);
                         }
@@ -111,7 +110,7 @@ public class PingMonitorActor extends RecordActorPrototype {
                 catch (InterruptedException ex) {
                     // We get this exception when the process is cancelled by the user
 
-                    processService.kill(processId);
+                    processSpawnerService.kill(processId);
                     setCurrentMessageToDead();
 
                     yield new Aborted();
@@ -129,20 +128,16 @@ public class PingMonitorActor extends RecordActorPrototype {
 
     @Inject
     public PingMonitorActor(Gson gson,
-                                       NodeConfigurationService nodeConfigurationService,
                                        ServiceConfiguration configuration,
                                        MqPersistence persistence,
-                                       ProcessService processService) throws SQLException {
+                                       ProcessSpawnerService processSpawnerService) throws SQLException {
         super(gson);
         this.gson = gson;
         this.node = configuration.node();
         this.persistence = persistence;
-        this.processService = processService;
+        this.processSpawnerService = processSpawnerService;
         this.inboxName = ProcessInboxNames.PING_INBOX + ":" + node;
-        this.processId = ProcessService.ProcessId.PING;
+        this.processId = ProcessSpawnerService.ProcessId.PING;
-
-        this.isPrimaryNode = Set.of(NodeProfile.BATCH_CRAWL, NodeProfile.MIXED)
-                .contains(nodeConfigurationService.get(node).profile());
     }
 
     /** Sets the message to dead in the database to avoid
@@ -171,7 +166,7 @@ public class PingMonitorActor extends RecordActorPrototype {
             // Run this call in a separate thread so that this thread can be interrupted waiting for it
             executorService.submit(() -> {
                 try {
-                    processService.trigger(processId);
+                    processSpawnerService.trigger(processId);
                 } catch (Exception e) {
                     logger.warn("Error in triggering process", e);
                     error.set(true);

code/execution/java/nu/marginalia/actor/proc/ProcessLivenessMonitorActor.java

@@ -8,7 +8,7 @@ import nu.marginalia.actor.prototype.RecordActorPrototype;
 import nu.marginalia.actor.state.ActorResumeBehavior;
 import nu.marginalia.actor.state.ActorStep;
 import nu.marginalia.actor.state.Resume;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.control.ServiceEventLog;
 import nu.marginalia.service.module.ServiceConfiguration;
 
@@ -21,7 +21,7 @@ import java.util.concurrent.TimeUnit;
 public class ProcessLivenessMonitorActor extends RecordActorPrototype {
 
     private final ServiceEventLog eventLogService;
-    private final ProcessService processService;
+    private final ProcessSpawnerService processSpawnerService;
     private final HikariDataSource dataSource;
 
     private final int node;
@@ -49,7 +49,7 @@ public class ProcessLivenessMonitorActor extends RecordActorPrototype {
                         var processId = heartbeat.getProcessId();
                         if (null == processId) continue;
 
-                        if (processService.isRunning(processId) && heartbeat.lastSeenMillis() < 10_000)
+                        if (processSpawnerService.isRunning(processId) && heartbeat.lastSeenMillis() < 10_000)
                             continue;
 
                         flagProcessAsStopped(heartbeat);
@@ -72,12 +72,12 @@ public class ProcessLivenessMonitorActor extends RecordActorPrototype {
     public ProcessLivenessMonitorActor(Gson gson,
                                        ServiceEventLog eventLogService,
                                        ServiceConfiguration configuration,
-                                       ProcessService processService,
+                                       ProcessSpawnerService processSpawnerService,
                                        HikariDataSource dataSource) {
         super(gson);
         this.node = configuration.node();
         this.eventLogService = eventLogService;
-        this.processService = processService;
+        this.processSpawnerService = processSpawnerService;
         this.dataSource = dataSource;
     }
 
@@ -208,8 +208,8 @@ public class ProcessLivenessMonitorActor extends RecordActorPrototype {
         public boolean isRunning() {
             return "RUNNING".equals(status);
         }
-        public ProcessService.ProcessId getProcessId() {
+        public ProcessSpawnerService.ProcessId getProcessId() {
-            return ProcessService.translateExternalIdBase(processBase);
+            return ProcessSpawnerService.translateExternalIdBase(processBase);
         }
     }
 

code/execution/java/nu/marginalia/actor/proc/ScrapeFeedsActor.java

@@ -47,6 +47,8 @@ public class ScrapeFeedsActor extends RecordActorPrototype {
 
     private final Path feedPath = WmsaHome.getHomePath().resolve("data/scrape-urls.txt");
 
+    private static boolean insertFoundDomains = Boolean.getBoolean("loader.insertFoundDomains");
+
     public record Initial() implements ActorStep {}
     @Resume(behavior = ActorResumeBehavior.RETRY)
     public record Wait(String ts) implements ActorStep {}
@@ -57,6 +59,8 @@ public class ScrapeFeedsActor extends RecordActorPrototype {
     public ActorStep transition(ActorStep self) throws Exception {
         return switch(self) {
             case Initial() -> {
+                if (!insertFoundDomains) yield new Error("Domain insertion prohibited, aborting");
+
                 if (nodeConfigurationService.get(nodeId).profile() != NodeProfile.REALTIME) {
                     yield new Error("Invalid node profile for RSS update");
                 }

code/execution/java/nu/marginalia/actor/task/ActorProcessWatcher.java

@@ -3,11 +3,11 @@ package nu.marginalia.actor.task;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.state.ActorControlFlowException;
-import nu.marginalia.mq.MqMessageState;
-import nu.marginalia.mq.persistence.MqPersistence;
-import nu.marginalia.process.ProcessService;
 import nu.marginalia.mq.MqMessage;
+import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
+import nu.marginalia.mq.persistence.MqPersistence;
+import nu.marginalia.process.ProcessSpawnerService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -20,13 +20,13 @@ public class ActorProcessWatcher {
 
     private static final Logger logger = LoggerFactory.getLogger(ActorProcessWatcher.class);
     private final MqPersistence persistence;
-    private final ProcessService processService;
+    private final ProcessSpawnerService processSpawnerService;
 
     @Inject
     public ActorProcessWatcher(MqPersistence persistence,
-                               ProcessService processService) {
+                               ProcessSpawnerService processSpawnerService) {
         this.persistence = persistence;
-        this.processService = processService;
+        this.processSpawnerService = processSpawnerService;
     }
 
     /** Wait for a process to start, and then wait for a response from the process,
@@ -36,7 +36,7 @@ public class ActorProcessWatcher {
      * <p>
      * When interrupted, the process is killed and the message is marked as dead.
      */
-    public MqMessage waitResponse(MqOutbox outbox, ProcessService.ProcessId processId, long msgId)
+    public MqMessage waitResponse(MqOutbox outbox, ProcessSpawnerService.ProcessId processId, long msgId)
             throws ActorControlFlowException, InterruptedException, SQLException
     {
         // enums values only have a single instance,
@@ -65,7 +65,7 @@ public class ActorProcessWatcher {
                 // This will prevent the monitor process from attempting to respawn the process as we kill it
 
                 outbox.flagAsDead(msgId);
-                processService.kill(processId);
+                processSpawnerService.kill(processId);
 
                 logger.info("Process {} killed due to interrupt", processId);
             }
@@ -94,12 +94,12 @@ public class ActorProcessWatcher {
     }
 
     /** Wait the specified time for the specified process to start running (does not start the process) */
-    private boolean waitForProcess(ProcessService.ProcessId processId, TimeUnit unit, int duration) throws InterruptedException {
+    private boolean waitForProcess(ProcessSpawnerService.ProcessId processId, TimeUnit unit, int duration) throws InterruptedException {
 
         // Wait for process to start
         long deadline = System.currentTimeMillis() + unit.toMillis(duration);
         while (System.currentTimeMillis() < deadline) {
-            if (processService.isRunning(processId))
+            if (processSpawnerService.isRunning(processId))
                 return true;
 
             TimeUnit.MILLISECONDS.sleep(100);

code/execution/java/nu/marginalia/actor/task/ConvertActor.java

@@ -12,7 +12,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.converting.ConvertRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.sideload.RedditSideloadHelper;
 import nu.marginalia.sideload.SideloadHelper;
 import nu.marginalia.sideload.StackExchangeSideloadHelper;
@@ -218,7 +218,7 @@ public class ConvertActor extends RecordActorPrototype {
                 );
             }
             case ConvertWait(FileStorageId destFid, long msgId) -> {
-                var rsp = processWatcher.waitResponse(mqConverterOutbox, ProcessService.ProcessId.CONVERTER, msgId);
+                var rsp = processWatcher.waitResponse(mqConverterOutbox, ProcessSpawnerService.ProcessId.CONVERTER, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     yield new Error("Converter failed");

code/execution/java/nu/marginalia/actor/task/ConvertAndLoadActor.java

@@ -18,7 +18,7 @@ import nu.marginalia.mqapi.index.IndexName;
 import nu.marginalia.mqapi.loading.LoadRequest;
 import nu.marginalia.nodecfg.NodeConfigurationService;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.service.module.ServiceConfiguration;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
@@ -95,7 +95,7 @@ public class ConvertAndLoadActor extends RecordActorPrototype {
             case Convert(FileStorageId crawlId, FileStorageId processedId, long msgId) when msgId < 0 ->
                     new Convert(crawlId, processedId, mqConverterOutbox.sendAsync(ConvertRequest.forCrawlData(crawlId, processedId)));
             case Convert(FileStorageId crawlId, FileStorageId processedId, long msgId) -> {
-                var rsp = processWatcher.waitResponse(mqConverterOutbox, ProcessService.ProcessId.CONVERTER, msgId);
+                var rsp = processWatcher.waitResponse(mqConverterOutbox, ProcessSpawnerService.ProcessId.CONVERTER, msgId);
 
                 if (rsp.state() != MqMessageState.OK)
                     yield new Error("Converter failed");
@@ -129,7 +129,7 @@ public class ConvertAndLoadActor extends RecordActorPrototype {
                 yield new Load(processedIds, id);
             }
             case Load(List<FileStorageId> processedIds, long msgId) -> {
-                var rsp = processWatcher.waitResponse(mqLoaderOutbox, ProcessService.ProcessId.LOADER, msgId);
+                var rsp = processWatcher.waitResponse(mqLoaderOutbox, ProcessSpawnerService.ProcessId.LOADER, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     yield new Error("Loader failed");
@@ -165,7 +165,7 @@ public class ConvertAndLoadActor extends RecordActorPrototype {
             }
             case ReindexFwd(long id) when id < 0 -> new ReindexFwd(createIndex(IndexName.FORWARD));
             case ReindexFwd(long id) -> {
-                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessService.ProcessId.INDEX_CONSTRUCTOR, id);
+                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessSpawnerService.ProcessId.INDEX_CONSTRUCTOR, id);
 
                 if (rsp.state() != MqMessageState.OK)
                     yield new Error("Forward index construction failed");
@@ -174,7 +174,7 @@ public class ConvertAndLoadActor extends RecordActorPrototype {
             }
             case ReindexFull(long id) when id < 0 -> new ReindexFull(createIndex(IndexName.REVERSE_FULL));
             case ReindexFull(long id) -> {
-                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessService.ProcessId.INDEX_CONSTRUCTOR, id);
+                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessSpawnerService.ProcessId.INDEX_CONSTRUCTOR, id);
 
                 if (rsp.state() != MqMessageState.OK)
                     yield new Error("Full index construction failed");
@@ -183,7 +183,7 @@ public class ConvertAndLoadActor extends RecordActorPrototype {
             }
             case ReindexPrio(long id) when id < 0 -> new ReindexPrio(createIndex(IndexName.REVERSE_PRIO));
             case ReindexPrio(long id) -> {
-                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessService.ProcessId.INDEX_CONSTRUCTOR, id);
+                var rsp = processWatcher.waitResponse(mqIndexConstructorOutbox, ProcessSpawnerService.ProcessId.INDEX_CONSTRUCTOR, id);
 
                 if (rsp.state() != MqMessageState.OK)
                     yield new Error("Prio index construction failed");

code/execution/java/nu/marginalia/actor/task/CrawlActor.java

@@ -13,7 +13,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.crawling.CrawlRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageType;
@@ -76,7 +76,7 @@ public class CrawlActor extends RecordActorPrototype {
             case Crawl (long msgId, FileStorageId fid, boolean cascadeLoad) -> {
                 var rsp = processWatcher.waitResponse(
                         mqCrawlerOutbox,
-                        ProcessService.ProcessId.CRAWLER,
+                        ProcessSpawnerService.ProcessId.CRAWLER,
                         msgId);
 
                 if (rsp.state() != MqMessageState.OK) {

code/execution/java/nu/marginalia/actor/task/ExportAtagsActor.java

@@ -10,7 +10,7 @@ import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.tasks.ExportTaskRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageState;
@@ -55,7 +55,7 @@ public class ExportAtagsActor extends RecordActorPrototype {
                 yield new Run(responseMsgId, crawlId, destId, newMsgId);
             }
             case Run(long responseMsgId, FileStorageId crawlId, FileStorageId destId, long msgId) -> {
-                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessService.ProcessId.EXPORT_TASKS, msgId);
+                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessSpawnerService.ProcessId.EXPORT_TASKS, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     storageService.flagFileForDeletion(destId);

code/execution/java/nu/marginalia/actor/task/ExportFeedsActor.java

@@ -10,7 +10,7 @@ import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.tasks.ExportTaskRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageState;
@@ -54,7 +54,7 @@ public class ExportFeedsActor extends RecordActorPrototype {
                 yield new Run(responseMsgId, crawlId, destId, newMsgId);
             }
             case Run(long responseMsgId, _, FileStorageId destId, long msgId) -> {
-                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessService.ProcessId.EXPORT_TASKS, msgId);
+                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessSpawnerService.ProcessId.EXPORT_TASKS, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     storageService.flagFileForDeletion(destId);

code/execution/java/nu/marginalia/actor/task/ExportSampleDataActor.java

@@ -9,7 +9,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.tasks.ExportTaskRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageState;
@@ -52,7 +52,7 @@ public class ExportSampleDataActor extends RecordActorPrototype {
                 yield new Run(crawlId, destId, size, ctFilter, name, newMsgId);
             }
             case Run(_, FileStorageId destId, _, _, _, long msgId) -> {
-                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessService.ProcessId.EXPORT_TASKS, msgId);
+                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessSpawnerService.ProcessId.EXPORT_TASKS, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     storageService.flagFileForDeletion(destId);

code/execution/java/nu/marginalia/actor/task/ExportTermFreqActor.java

@@ -10,7 +10,7 @@ import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.mqapi.tasks.ExportTaskRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageState;
@@ -52,7 +52,7 @@ public class ExportTermFreqActor extends RecordActorPrototype {
                 yield new Run(responseMsgId, crawlId, destId, newMsgId);
             }
             case Run(long responseMsgId, _, FileStorageId destId, long msgId) -> {
-                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessService.ProcessId.EXPORT_TASKS, msgId);
+                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessSpawnerService.ProcessId.EXPORT_TASKS, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     storageService.flagFileForDeletion(destId);

code/execution/java/nu/marginalia/actor/task/LiveCrawlActor.java

@@ -13,7 +13,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.crawling.LiveCrawlRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import org.apache.commons.io.FileUtils;
 import org.slf4j.Logger;
@@ -44,7 +44,6 @@ public class LiveCrawlActor extends RecordActorPrototype {
 
     @Override
     public ActorStep transition(ActorStep self) throws Exception {
-        logger.info("{}", self);
         return switch (self) {
             case Initial() -> {
                 yield new Monitor("-");
@@ -75,7 +74,7 @@ public class LiveCrawlActor extends RecordActorPrototype {
                 yield new LiveCrawl(feedsHash, id);
             }
             case LiveCrawl(String feedsHash, long msgId) -> {
-                var rsp = processWatcher.waitResponse(mqLiveCrawlerOutbox, ProcessService.ProcessId.LIVE_CRAWLER, msgId);
+                var rsp = processWatcher.waitResponse(mqLiveCrawlerOutbox, ProcessSpawnerService.ProcessId.LIVE_CRAWLER, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     yield new Error("Crawler failed");

code/execution/java/nu/marginalia/actor/task/RecrawlSingleDomainActor.java

@@ -11,7 +11,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.crawling.CrawlRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorageId;
 import nu.marginalia.storage.model.FileStorageType;
@@ -51,7 +51,7 @@ public class RecrawlSingleDomainActor extends RecordActorPrototype {
             case Crawl (long msgId) -> {
                 var rsp = processWatcher.waitResponse(
                         mqCrawlerOutbox,
-                        ProcessService.ProcessId.CRAWLER,
+                        ProcessSpawnerService.ProcessId.CRAWLER,
                         msgId);
 
                 if (rsp.state() != MqMessageState.OK) {

code/execution/java/nu/marginalia/actor/task/TriggerAdjacencyCalculationActor.java

@@ -9,7 +9,7 @@ import nu.marginalia.mq.MqMessageState;
 import nu.marginalia.mq.outbox.MqOutbox;
 import nu.marginalia.mqapi.tasks.ExportTaskRequest;
 import nu.marginalia.process.ProcessOutboxes;
-import nu.marginalia.process.ProcessService;
+import nu.marginalia.process.ProcessSpawnerService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,7 +34,7 @@ public class TriggerAdjacencyCalculationActor extends RecordActorPrototype {
                 yield new Run(newMsgId);
             }
             case Run(long msgId) -> {
-                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessService.ProcessId.EXPORT_TASKS, msgId);
+                var rsp = processWatcher.waitResponse(exportTasksOutbox, ProcessSpawnerService.ProcessId.EXPORT_TASKS, msgId);
 
                 if (rsp.state() != MqMessageState.OK) {
                     yield new Error("Exporter failed");

code/execution/java/nu/marginalia/actor/task/UpdateNsfwFiltersActor.java

@@ -5,6 +5,8 @@ import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import nu.marginalia.actor.prototype.RecordActorPrototype;
 import nu.marginalia.actor.state.ActorStep;
+import nu.marginalia.mq.MqMessageState;
+import nu.marginalia.mq.persistence.MqPersistence;
 import nu.marginalia.nsfw.NsfwDomainFilter;
 import nu.marginalia.service.module.ServiceConfiguration;
 
@@ -12,23 +14,26 @@ import nu.marginalia.service.module.ServiceConfiguration;
 public class UpdateNsfwFiltersActor extends RecordActorPrototype {
     private final ServiceConfiguration serviceConfiguration;
     private final NsfwDomainFilter nsfwDomainFilter;
+    private final MqPersistence persistence;
 
-    public record Initial() implements ActorStep {}
+    public record Initial(long respondMsgId) implements ActorStep {}
-    public record Run() implements ActorStep {}
+    public record Run(long respondMsgId) implements ActorStep {}
 
     @Override
     public ActorStep transition(ActorStep self) throws Exception {
         return switch(self) {
-            case Initial() -> {
+            case Initial(long respondMsgId) -> {
                 if (serviceConfiguration.node() != 1) {
+                    persistence.updateMessageState(respondMsgId, MqMessageState.ERR);
                     yield new Error("This actor can only run on node 1");
                 }
                 else {
-                    yield new Run();
+                    yield new Run(respondMsgId);
                 }
             }
-            case Run() -> {
+            case Run(long respondMsgId) -> {
                 nsfwDomainFilter.fetchLists();
+                persistence.updateMessageState(respondMsgId, MqMessageState.OK);
                 yield new End();
             }
             default -> new Error();
@@ -43,11 +48,13 @@ public class UpdateNsfwFiltersActor extends RecordActorPrototype {
     @Inject
     public UpdateNsfwFiltersActor(Gson gson,
                                   ServiceConfiguration serviceConfiguration,
-                                  NsfwDomainFilter nsfwDomainFilter)
+                                  NsfwDomainFilter nsfwDomainFilter,
+                                  MqPersistence persistence)
     {
         super(gson);
         this.serviceConfiguration = serviceConfiguration;
         this.nsfwDomainFilter = nsfwDomainFilter;
+        this.persistence = persistence;
     }
 
 }

code/execution/java/nu/marginalia/execution/ExecutorCrawlGrpcService.java

@@ -1,6 +1,7 @@
 package nu.marginalia.execution;
 
 import com.google.inject.Inject;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import nu.marginalia.actor.ExecutorActor;
 import nu.marginalia.actor.ExecutorActorControlService;
@@ -36,7 +37,7 @@ public class ExecutorCrawlGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -52,7 +53,7 @@ public class ExecutorCrawlGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -66,7 +67,7 @@ public class ExecutorCrawlGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -80,7 +81,7 @@ public class ExecutorCrawlGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -98,7 +99,7 @@ public class ExecutorCrawlGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 

code/execution/java/nu/marginalia/execution/ExecutorExportGrpcService.java

@@ -2,6 +2,7 @@ package nu.marginalia.execution;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import nu.marginalia.actor.ExecutorActor;
 import nu.marginalia.actor.ExecutorActorControlService;
@@ -38,7 +39,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -57,7 +58,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -73,7 +74,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -87,7 +88,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -99,7 +100,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -114,14 +115,14 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
     @Override
     public void exportAllAtags(Empty request, StreamObserver<Empty> responseObserver) {
         if (serviceConfiguration.node() != 1) {
-            responseObserver.onError(new IllegalArgumentException("Export all atags is only available on node 1"));
+            responseObserver.onError(Status.UNAVAILABLE.withDescription("Export all atags is only available on node 1").asRuntimeException());
         }
         try {
             actorControlService.startFrom(ExecutorActor.PREC_EXPORT_ALL,
@@ -131,7 +132,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -145,7 +146,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -159,7 +160,7 @@ public class ExecutorExportGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 }

code/execution/java/nu/marginalia/execution/ExecutorGrpcService.java

@@ -1,6 +1,7 @@
 package nu.marginalia.execution;
 
 import com.google.inject.Inject;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import nu.marginalia.WmsaHome;
 import nu.marginalia.actor.ActorApi;
@@ -10,6 +11,7 @@ import nu.marginalia.actor.state.ActorStateInstance;
 import nu.marginalia.actor.task.DownloadSampleActor;
 import nu.marginalia.actor.task.RestoreBackupActor;
 import nu.marginalia.actor.task.TriggerAdjacencyCalculationActor;
+import nu.marginalia.actor.task.UpdateNsfwFiltersActor;
 import nu.marginalia.functions.execution.api.*;
 import nu.marginalia.service.module.ServiceConfiguration;
 import nu.marginalia.service.server.DiscoverableService;
@@ -57,7 +59,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -69,7 +71,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -81,7 +83,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -95,7 +97,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -111,7 +113,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -127,7 +129,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -202,7 +204,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -228,7 +230,7 @@ public class ExecutorGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -263,4 +265,19 @@ public class ExecutorGrpcService
         System.exit(0);
     }
 
+    @Override
+    public void updateNsfwFilters(RpcUpdateNsfwFilters request, StreamObserver<Empty> responseObserver) {
+        logger.info("Got request {}", request);
+        try {
+            actorControlService.startFrom(ExecutorActor.UPDATE_NSFW_LISTS,
+                    new UpdateNsfwFiltersActor.Initial(request.getMsgId()));
+
+            responseObserver.onNext(Empty.getDefaultInstance());
+            responseObserver.onCompleted();
+        }
+        catch (Exception e) {
+            logger.error("Failed to update nsfw filters", e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
+        }
+    }
 }

code/execution/java/nu/marginalia/execution/ExecutorSideloadGrpcService.java

@@ -1,6 +1,7 @@
 package nu.marginalia.execution;
 
 import com.google.inject.Inject;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import nu.marginalia.actor.ExecutorActor;
 import nu.marginalia.actor.ExecutorActorControlService;
@@ -33,7 +34,7 @@ public class ExecutorSideloadGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -48,7 +49,7 @@ public class ExecutorSideloadGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -63,7 +64,7 @@ public class ExecutorSideloadGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -78,7 +79,7 @@ public class ExecutorSideloadGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -93,7 +94,7 @@ public class ExecutorSideloadGrpcService
             responseObserver.onCompleted();
         }
         catch (Exception e) {
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 

code/execution/java/nu/marginalia/process/ProcessSpawnerService.java

@@ -8,6 +8,7 @@ import nu.marginalia.crawl.CrawlerMain;
 import nu.marginalia.index.IndexConstructorMain;
 import nu.marginalia.livecrawler.LiveCrawlerMain;
 import nu.marginalia.loading.LoaderMain;
+import nu.marginalia.ndp.NdpMain;
 import nu.marginalia.ping.PingMain;
 import nu.marginalia.service.control.ServiceEventLog;
 import nu.marginalia.service.server.BaseServiceParams;
@@ -28,7 +29,7 @@ import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 
 @Singleton
-public class ProcessService {
+public class ProcessSpawnerService {
     private final Logger logger = LoggerFactory.getLogger(getClass());
     private final Marker processMarker = MarkerFactory.getMarker("PROCESS");
 
@@ -57,6 +58,7 @@ public class ProcessService {
         CONVERTER(ConverterMain.class),
         LOADER(LoaderMain.class),
         INDEX_CONSTRUCTOR(IndexConstructorMain.class),
+        NDP(NdpMain.class),
         EXPORT_TASKS(ExportTasksMain.class),
         ;
 
@@ -72,6 +74,7 @@ public class ProcessService {
                 case CONVERTER -> "CONVERTER_PROCESS_OPTS";
                 case LOADER -> "LOADER_PROCESS_OPTS";
                 case PING -> "PING_PROCESS_OPTS";
+                case NDP -> "NDP_PROCESS_OPTS";
                 case INDEX_CONSTRUCTOR -> "INDEX_CONSTRUCTION_PROCESS_OPTS";
                 case EXPORT_TASKS -> "EXPORT_TASKS_PROCESS_OPTS";
             };
@@ -85,7 +88,7 @@ public class ProcessService {
     }
 
     @Inject
-    public ProcessService(BaseServiceParams params) {
+    public ProcessSpawnerService(BaseServiceParams params) {
         this.eventLog = params.eventLog;
         this.node = params.configuration.node();
     }

code/functions/domain-info/api/java/nu/marginalia/api/domains/DomainInfoClient.java

@@ -2,6 +2,8 @@ package nu.marginalia.api.domains;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import nu.marginalia.api.domains.model.DomainInformation;
+import nu.marginalia.api.domains.model.SimilarDomain;
 import nu.marginalia.service.client.GrpcChannelPoolFactory;
 import nu.marginalia.service.client.GrpcSingleNodeChannelPool;
 import nu.marginalia.service.discovery.property.ServiceKey;
@@ -10,16 +12,19 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
-import java.util.concurrent.*;
+import java.util.concurrent.ExecutorService;
-
+import java.util.concurrent.Executors;
-import nu.marginalia.api.domains.model.*;
+import java.util.concurrent.Future;
 
 @Singleton
 public class DomainInfoClient {
     private static final Logger logger = LoggerFactory.getLogger(DomainInfoClient.class);
 
     private final GrpcSingleNodeChannelPool<DomainInfoAPIGrpc.DomainInfoAPIBlockingStub> channelPool;
-    private final ExecutorService executor = Executors.newWorkStealingPool(8);
+
+
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
+    private static final ExecutorService executor = useLoom ? Executors.newVirtualThreadPerTaskExecutor() : Executors.newWorkStealingPool(8);
 
     @Inject
     public DomainInfoClient(GrpcChannelPoolFactory factory) {

code/functions/live-capture/api/java/nu/marginalia/api/domsample/DomSampleClient.java

@@ -0,0 +1,99 @@
+package nu.marginalia.api.domsample;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import nu.marginalia.service.client.GrpcChannelPoolFactory;
+import nu.marginalia.service.client.GrpcSingleNodeChannelPool;
+import nu.marginalia.service.discovery.property.ServiceKey;
+import nu.marginalia.service.discovery.property.ServicePartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+
+@Singleton
+public class DomSampleClient {
+    private final GrpcSingleNodeChannelPool<DomSampleApiGrpc.DomSampleApiBlockingStub> channelPool;
+    private static final Logger logger = LoggerFactory.getLogger(DomSampleClient.class);
+
+    @Inject
+    public DomSampleClient(GrpcChannelPoolFactory factory) {
+
+        // The client is only interested in the primary node
+        var key = ServiceKey.forGrpcApi(DomSampleApiGrpc.class, ServicePartition.any());
+        this.channelPool = factory.createSingle(key, DomSampleApiGrpc::newBlockingStub);
+    }
+
+    public Optional<RpcDomainSample> getSample(String domainName) {
+        try {
+            var val = channelPool.call(DomSampleApiGrpc.DomSampleApiBlockingStub::getSample)
+                    .run(RpcDomainName.newBuilder().setDomainName(domainName).build());
+
+            return Optional.of(val);
+        }
+        catch (StatusRuntimeException sre) {
+            if (sre.getStatus() != Status.NOT_FOUND) {
+                logger.error("Failed to fetch DOM sample");
+            }
+            return Optional.empty();
+        }
+    }
+
+    public boolean hasSample(String domainName) {
+        try {
+            return channelPool.call(DomSampleApiGrpc.DomSampleApiBlockingStub::hasSample)
+                    .run(RpcDomainName.newBuilder().setDomainName(domainName).build())
+                    .getAnswer();
+        }
+        catch (StatusRuntimeException sre) {
+            return false;
+        }
+    }
+
+    public CompletableFuture<Boolean> hasSample(String domainName, ExecutorService executor) {
+        try {
+            return channelPool.call(DomSampleApiGrpc.DomSampleApiBlockingStub::hasSample)
+                    .async(executor)
+                    .run(RpcDomainName.newBuilder().setDomainName(domainName).build())
+                    .thenApply(RpcBooleanRsp::getAnswer);
+        }
+        catch (StatusRuntimeException sre) {
+            return CompletableFuture.completedFuture(false);
+        }
+    }
+
+    public CompletableFuture<RpcDomainSample> getSampleAsync(String domainName, ExecutorService executorService) {
+        return channelPool.call(DomSampleApiGrpc.DomSampleApiBlockingStub::getSample)
+                .async(executorService)
+                .run(RpcDomainName.newBuilder().setDomainName(domainName).build());
+    }
+
+    public List<RpcDomainSample> getAllSamples(String domainName) {
+        try {
+            Iterator<RpcDomainSample> val = channelPool.call(DomSampleApiGrpc.DomSampleApiBlockingStub::getAllSamples)
+                    .run(RpcDomainName.newBuilder().setDomainName(domainName).build());
+
+            List<RpcDomainSample> ret = new ArrayList<>();
+            val.forEachRemaining(ret::add);
+            return ret;
+        }
+        catch (StatusRuntimeException sre) {
+            logger.error("Failed to fetch DOM sample");
+            return List.of();
+        }
+    }
+
+    public boolean waitReady(Duration duration) throws InterruptedException {
+        return channelPool.awaitChannel(duration);
+    }
+
+
+}

code/functions/live-capture/api/java/nu/marginalia/api/feeds/FeedsClient.java

@@ -11,6 +11,7 @@ import nu.marginalia.service.discovery.property.ServicePartition;
 import nu.marginalia.service.module.ServiceConfiguration;
 
 import javax.annotation.CheckReturnValue;
+import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
@@ -23,7 +24,9 @@ import java.util.function.BiConsumer;
 
 @Singleton
 public class FeedsClient {
-    private final ExecutorService executorService = Executors.newCachedThreadPool();
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
+    private static final ExecutorService executorService = useLoom ? Executors.newVirtualThreadPerTaskExecutor() : Executors.newCachedThreadPool();
+
     private final GrpcSingleNodeChannelPool<FeedApiGrpc.FeedApiBlockingStub> channelPool;
     private final MqOutbox updateFeedsOutbox;
 
@@ -59,6 +62,11 @@ public class FeedsClient {
                 .forEachRemaining(rsp -> consumer.accept(rsp.getDomain(), new ArrayList<>(rsp.getUrlList())));
     }
 
+    public boolean waitReady(Duration duration) throws InterruptedException {
+        return channelPool.awaitChannel(duration);
+    }
+
+
     /** Get the hash of the feed data, for identifying when the data has been updated */
     public String getFeedDataHash() {
         return channelPool.call(FeedApiGrpc.FeedApiBlockingStub::getFeedDataHash)

code/functions/live-capture/api/src/main/protobuf/dom-sample.proto

@@ -0,0 +1,40 @@
+syntax="proto3";
+package nu.marginalia.api.domsample;
+
+option java_package="nu.marginalia.api.domsample";
+option java_multiple_files=true;
+
+
+service DomSampleApi {
+  rpc getSample(RpcDomainName) returns (RpcDomainSample) {}
+  rpc hasSample(RpcDomainName) returns (RpcBooleanRsp) {}
+  rpc getAllSamples(RpcDomainName) returns (stream RpcDomainSample) {}
+}
+
+message RpcDomainName {
+  string domainName = 1;
+}
+
+message RpcBooleanRsp {
+  bool answer = 1;
+}
+
+message RpcDomainSample {
+  string domainName = 1;
+  string url = 2;
+  string htmlSample = 3;
+  bool accepted_popover = 4;
+  repeated RpcOutgoingRequest outgoingRequests = 5;
+}
+
+message RpcOutgoingRequest {
+  RequestMethod method = 1;
+  int64 timestamp = 2;
+  string url = 3;
+
+  enum RequestMethod {
+    GET = 0;
+    POST = 1;
+    OTHER = 2;
+  };
+}

code/functions/live-capture/build.gradle

@@ -22,6 +22,7 @@ dependencies {
     implementation project(':code:common:db')
     implementation project(':code:libraries:blocking-thread-pool')
     implementation project(':code:libraries:message-queue')
+    implementation project(':code:libraries:domain-lock')
 
     implementation project(':code:execution:api')
     implementation project(':code:processes:crawling-process:ft-content-type')
@@ -34,6 +35,7 @@ dependencies {
     implementation libs.bundles.slf4j
     implementation libs.commons.lang3
     implementation libs.commons.io
+    implementation libs.httpclient
     implementation libs.wiremock
 
     implementation libs.prometheus

code/functions/live-capture/java/nu/marginalia/domsample/DomSampleGrpcService.java

@@ -0,0 +1,123 @@
+package nu.marginalia.domsample;
+
+import com.google.inject.Inject;
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import nu.marginalia.api.domsample.*;
+import nu.marginalia.domsample.db.DomSampleDb;
+import nu.marginalia.service.server.DiscoverableService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+public class DomSampleGrpcService
+        extends DomSampleApiGrpc.DomSampleApiImplBase
+        implements DiscoverableService
+{
+    private static final Logger logger = LoggerFactory.getLogger(DomSampleGrpcService.class);
+
+    private final DomSampleDb domSampleDb;
+
+    @Inject
+    public DomSampleGrpcService(DomSampleDb domSampleDb) {
+        this.domSampleDb = domSampleDb;
+    }
+
+    @Override
+    public void getSample(RpcDomainName request, StreamObserver<RpcDomainSample> responseObserver) {
+        String domainName = request.getDomainName();
+        if (domainName.isBlank()) {
+            responseObserver.onError(Status.INVALID_ARGUMENT
+                            .withDescription("Invalid domain name")
+                            .asRuntimeException());
+            return;
+        }
+
+        try {
+            List<DomSampleDb.Sample> dbRecords = domSampleDb.getSamples(domainName);
+            if (dbRecords.isEmpty()) {
+                responseObserver.onError(Status.NOT_FOUND.withDescription("No sample found").asRuntimeException());
+                return;
+            }
+
+            // Grab the first sample
+            RpcDomainSample.Builder response = convert(dbRecords.getFirst());
+
+            responseObserver.onNext(response.build());
+            responseObserver.onCompleted();
+        }
+        catch (Exception e) {
+            logger.error("Error in getSample()", e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
+        }
+    }
+
+    @Override
+    public void hasSample(RpcDomainName request, StreamObserver<RpcBooleanRsp> responseObserver) {
+        String domainName = request.getDomainName();
+        if (domainName.isBlank()) {
+            responseObserver.onError(Status.INVALID_ARGUMENT
+                    .withDescription("Invalid domain name")
+                    .asRuntimeException());
+            return;
+        }
+
+        try {
+            responseObserver.onNext(RpcBooleanRsp.newBuilder()
+                    .setAnswer(domSampleDb.hasSample(domainName)).build());
+            responseObserver.onCompleted();
+        }
+        catch (Exception e) {
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
+        }
+    }
+
+    @Override
+    public void getAllSamples(RpcDomainName request, StreamObserver<RpcDomainSample> responseObserver) {
+        String domainName = request.getDomainName();
+        if (domainName.isBlank()) {
+            responseObserver.onError(Status.INVALID_ARGUMENT
+                    .withDescription("Invalid domain name")
+                    .asRuntimeException());
+            return;
+        }
+
+        try {
+            List<DomSampleDb.Sample> dbRecords = domSampleDb.getSamples(domainName);
+
+            for (var record : dbRecords) {
+                responseObserver.onNext(convert(record).build());
+            }
+
+            responseObserver.onCompleted();
+        }
+        catch (Exception e) {
+            logger.error("Error in getSample()", e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
+        }
+    }
+
+    private RpcDomainSample.Builder convert(DomSampleDb.Sample dbSample) {
+
+        var sampleBuilder = RpcDomainSample.newBuilder()
+                .setDomainName(dbSample.domain())
+                .setAcceptedPopover(dbSample.acceptedPopover())
+                .setHtmlSample(dbSample.sample());
+
+        for (var req : dbSample.parseRequests()) {
+            sampleBuilder.addOutgoingRequestsBuilder()
+                    .setUrl(req.uri().toString())
+                    .setMethod(switch (req.method().toUpperCase())
+                    {
+                        case "GET" -> RpcOutgoingRequest.RequestMethod.GET;
+                        case "POST" -> RpcOutgoingRequest.RequestMethod.POST;
+                        default -> RpcOutgoingRequest.RequestMethod.OTHER;
+                    })
+                    .setTimestamp(req.timestamp());
+        }
+
+        return sampleBuilder;
+    }
+
+}

code/functions/live-capture/java/nu/marginalia/domsample/db/DomSampleDb.java

@@ -1,17 +1,28 @@
 package nu.marginalia.domsample.db;
 
 import nu.marginalia.WmsaHome;
+import nu.marginalia.model.EdgeUrl;
+import org.apache.commons.lang3.StringUtils;
 import org.jsoup.Jsoup;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.nio.file.Path;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
 
 public class DomSampleDb implements AutoCloseable {
     private static final String dbFileName = "dom-sample.db";
     private final Connection connection;
+    private static final Logger logger = LoggerFactory.getLogger(DomSampleDb.class);
 
     public DomSampleDb() throws SQLException{
         this(WmsaHome.getDataPath().resolve(dbFileName));
@@ -88,14 +99,78 @@ public class DomSampleDb implements AutoCloseable {
     }
 
 
-    public record Sample(String url, String domain, String sample, String requests, boolean acceptedPopover) {}
+    public record Sample(String url, String domain, String sample, String requests, boolean acceptedPopover) {
+
+        public List<SampleRequest> parseRequests() {
+            List<SampleRequest> requests = new ArrayList<>();
+
+            // Request format is METHOD\tTIMESTAMP\tURI\n
+
+            for (var line : StringUtils.split(this.requests, '\n')) {
+                String[] parts = StringUtils.split(line, "\t", 3);
+                if (parts.length != 3) continue;
+
+                try {
+                    String method =  parts[0];
+                    long ts = Long.parseLong(parts[1]);
+                    String linkUrl = parts[2];
+
+                    URI uri = parseURI(linkUrl);
+
+                    requests.add(new SampleRequest(method, ts, uri));
+                }
+                catch (Exception e) {
+                    logger.warn("Failed to parse requests", e);
+                }
+            }
+
+            return requests;
+        }
+
+
+        private static URI parseURI(String uri) throws URISyntaxException {
+            try {
+                return new URI(uri);
+            }
+            catch (URISyntaxException ex) {
+                return new EdgeUrl(uri).asURI();
+            }
+        }
+    }
+
+    public record SampleRequest(String method, long timestamp, URI uri) {}
+
+    /**
+     * @param consumer - consume the sample, return true to continue consumption
+     * @throws SQLException
+     */
+    public void forEachSample(Predicate<Sample> consumer) throws SQLException {
+        try (var stmt = connection.prepareStatement("""
+                SELECT url, domain, sample, requests, accepted_popover
+                FROM samples
+                """))
+        {
+            var rs = stmt.executeQuery();
+            while (rs.next()) {
+                var sample = new Sample(
+                        rs.getString("url"),
+                        rs.getString("domain"),
+                        rs.getString("sample"),
+                        rs.getString("requests"),
+                        rs.getBoolean("accepted_popover")
+                );
+
+                if (!consumer.test(sample)) break;
+            }
+        }
+    }
 
     public List<Sample> getSamples(String domain) throws SQLException {
         List<Sample> samples = new ArrayList<>();
 
         try (var stmt = connection.prepareStatement("""
                 SELECT url, sample, requests, accepted_popover
-                FROM samples 
+                FROM samples
                 WHERE domain = ?
                 """))
         {
@@ -116,6 +191,21 @@ public class DomSampleDb implements AutoCloseable {
         return samples;
     }
 
+
+    public boolean hasSample(String domain) throws SQLException {
+
+        try (var stmt = connection.prepareStatement("""
+                SELECT 1
+                FROM samples
+                WHERE domain = ?
+                """))
+        {
+            stmt.setString(1, domain);
+            var rs = stmt.executeQuery();
+            return rs.next();
+        }
+    }
+
     public void saveSample(String domain, String url, String rawContent) throws SQLException {
         var doc = Jsoup.parse(rawContent);
 

code/functions/live-capture/java/nu/marginalia/rss/svc/DomainLocks.java

@@ -1,66 +0,0 @@
-package nu.marginalia.rss.svc;
-
-import nu.marginalia.model.EdgeDomain;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Semaphore;
-
-/** Holds lock objects for each domain, to prevent multiple threads from
- * crawling the same domain at the same time.
- */
-public class DomainLocks {
-    // 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, Semaphore> locks = new ConcurrentHashMap<>();
-
-    /** Returns a lock object corresponding to the given domain.  The object is returned as-is,
-     * and may be held by another thread.  The caller is responsible for locking and  releasing the lock.
-     */
-    public DomainLock lockDomain(EdgeDomain domain) throws InterruptedException {
-        return new DomainLock(domain.toString(),
-                locks.computeIfAbsent(domain.topDomain.toLowerCase(), this::defaultPermits));
-    }
-
-    private Semaphore defaultPermits(String topDomain) {
-        if (topDomain.equals("wordpress.com"))
-            return new Semaphore(16);
-        if (topDomain.equals("blogspot.com"))
-            return new Semaphore(8);
-
-        if (topDomain.equals("neocities.org"))
-            return new Semaphore(4);
-        if (topDomain.equals("github.io"))
-            return new Semaphore(4);
-
-        if (topDomain.equals("substack.com")) {
-            return new Semaphore(1);
-        }
-        if (topDomain.endsWith(".edu")) {
-            return new Semaphore(1);
-        }
-
-        return new Semaphore(2);
-    }
-
-    public static class DomainLock implements AutoCloseable {
-        private final String domainName;
-        private final Semaphore semaphore;
-
-        DomainLock(String domainName, Semaphore semaphore) throws InterruptedException {
-            this.domainName = domainName;
-            this.semaphore = semaphore;
-
-            Thread.currentThread().setName("fetching:" + domainName + " [await domain lock]");
-            semaphore.acquire();
-            Thread.currentThread().setName("fetching:" + domainName);
-        }
-
-        @Override
-        public void close() {
-            semaphore.release();
-            Thread.currentThread().setName("fetching:" + domainName + " [wrapping up]");
-        }
-    }
-}

code/functions/live-capture/java/nu/marginalia/rss/svc/FeedFetcherService.java

@@ -5,6 +5,8 @@ import com.opencsv.CSVReader;
 import nu.marginalia.WmsaHome;
 import nu.marginalia.contenttype.ContentType;
 import nu.marginalia.contenttype.DocumentBodyToString;
+import nu.marginalia.coordination.DomainCoordinator;
+import nu.marginalia.coordination.DomainLock;
 import nu.marginalia.executor.client.ExecutorClient;
 import nu.marginalia.model.EdgeDomain;
 import nu.marginalia.nodecfg.NodeConfigurationService;
@@ -18,19 +20,36 @@ import nu.marginalia.storage.FileStorageService;
 import nu.marginalia.storage.model.FileStorage;
 import nu.marginalia.storage.model.FileStorageType;
 import nu.marginalia.util.SimpleBlockingThreadPool;
+import org.apache.hc.client5.http.ConnectionKeepAliveStrategy;
+import org.apache.hc.client5.http.classic.HttpClient;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.config.RequestConfig;
+import org.apache.hc.client5.http.cookie.StandardCookieSpec;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManagerBuilder;
+import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.HeaderElement;
+import org.apache.hc.core5.http.HeaderElements;
+import org.apache.hc.core5.http.HttpResponse;
+import org.apache.hc.core5.http.io.SocketConfig;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
+import org.apache.hc.core5.http.message.MessageSupport;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.util.TimeValue;
+import org.apache.hc.core5.util.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
-import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
-import java.net.http.HttpClient;
-import java.net.http.HttpRequest;
-import java.net.http.HttpResponse;
 import java.sql.SQLException;
-import java.time.*;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
@@ -51,12 +70,15 @@ public class FeedFetcherService {
     private final ServiceHeartbeat serviceHeartbeat;
     private final ExecutorClient executorClient;
 
-    private final DomainLocks domainLocks = new DomainLocks();
+    private final DomainCoordinator domainCoordinator;
+
+    private final HttpClient httpClient;
 
     private volatile boolean updating;
 
     @Inject
     public FeedFetcherService(FeedDb feedDb,
+                              DomainCoordinator domainCoordinator,
                               FileStorageService fileStorageService,
                               NodeConfigurationService nodeConfigurationService,
                               ServiceHeartbeat serviceHeartbeat,
@@ -67,6 +89,84 @@ public class FeedFetcherService {
         this.nodeConfigurationService = nodeConfigurationService;
         this.serviceHeartbeat = serviceHeartbeat;
         this.executorClient = executorClient;
+        this.domainCoordinator = domainCoordinator;
+
+        final ConnectionConfig connectionConfig = ConnectionConfig.custom()
+                .setSocketTimeout(15, TimeUnit.SECONDS)
+                .setConnectTimeout(15, TimeUnit.SECONDS)
+                .setValidateAfterInactivity(TimeValue.ofSeconds(5))
+                .build();
+
+
+        var connectionManager = PoolingHttpClientConnectionManagerBuilder.create()
+                .setMaxConnPerRoute(2)
+                .setMaxConnTotal(50)
+                .setDefaultConnectionConfig(connectionConfig)
+                .build();
+
+        connectionManager.setDefaultSocketConfig(SocketConfig.custom()
+                .setSoLinger(TimeValue.ofSeconds(-1))
+                .setSoTimeout(Timeout.ofSeconds(10))
+                .build()
+        );
+
+        Thread.ofPlatform().daemon(true).start(() -> {
+            try {
+                for (;;) {
+                    TimeUnit.SECONDS.sleep(15);
+                    logger.info("Connection pool stats: {}", connectionManager.getTotalStats());
+                }
+            }
+            catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        });
+
+        final RequestConfig defaultRequestConfig = RequestConfig.custom()
+                .setCookieSpec(StandardCookieSpec.IGNORE)
+                .setResponseTimeout(10, TimeUnit.SECONDS)
+                .setConnectionRequestTimeout(5, TimeUnit.MINUTES)
+                .build();
+
+        httpClient = HttpClients.custom()
+                .setDefaultRequestConfig(defaultRequestConfig)
+                .setConnectionManager(connectionManager)
+                .setUserAgent(WmsaHome.getUserAgent().uaIdentifier())
+                .setConnectionManager(connectionManager)
+                .setKeepAliveStrategy(new ConnectionKeepAliveStrategy() {
+                    // Default keep-alive duration is 3 minutes, but this is too long for us,
+                    // as we are either going to re-use it fairly quickly or close it for a long time.
+                    //
+                    // So we set it to 30 seconds or clamp the server-provided value to a minimum of 10 seconds.
+                    private static final TimeValue defaultValue = TimeValue.ofSeconds(30);
+
+                    @Override
+                    public TimeValue getKeepAliveDuration(HttpResponse response, HttpContext context) {
+                        final Iterator<HeaderElement> it = MessageSupport.iterate(response, HeaderElements.KEEP_ALIVE);
+
+                        while (it.hasNext()) {
+                            final HeaderElement he = it.next();
+                            final String param = he.getName();
+                            final String value = he.getValue();
+
+                            if (value == null)
+                                continue;
+                            if (!"timeout".equalsIgnoreCase(param))
+                                continue;
+
+                            try {
+                                long timeout = Long.parseLong(value);
+                                timeout = Math.clamp(timeout, 30, defaultValue.toSeconds());
+                                return TimeValue.ofSeconds(timeout);
+                            } catch (final NumberFormatException ignore) {
+                                break;
+                            }
+                        }
+                        return defaultValue;
+                    }
+                })
+                .build();
+
     }
 
     public enum UpdateMode {
@@ -82,13 +182,7 @@ public class FeedFetcherService {
 
 
         try (FeedDbWriter writer = feedDb.createWriter();
-             HttpClient client = HttpClient.newBuilder()
+             ExecutorService fetchExecutor = Executors.newVirtualThreadPerTaskExecutor();
-                .connectTimeout(Duration.ofSeconds(15))
-                .executor(Executors.newCachedThreadPool())
-                .followRedirects(HttpClient.Redirect.NORMAL)
-                .version(HttpClient.Version.HTTP_2)
-                .build();
-             ExecutorService fetchExecutor = Executors.newCachedThreadPool();
              FeedJournal feedJournal = FeedJournal.create();
              var heartbeat = serviceHeartbeat.createServiceAdHocTaskHeartbeat("Update Rss Feeds")
         ) {
@@ -132,8 +226,9 @@ public class FeedFetcherService {
                         };
 
                         FetchResult feedData;
-                        try (DomainLocks.DomainLock domainLock = domainLocks.lockDomain(new EdgeDomain(feed.domain()))) {
+                        try (DomainLock domainLock = domainCoordinator.lockDomain(new EdgeDomain(feed.domain()))) {
-                            feedData = fetchFeedData(feed, client, fetchExecutor, ifModifiedSinceDate, ifNoneMatchTag);
+                            feedData = fetchFeedData(feed, fetchExecutor, ifModifiedSinceDate, ifNoneMatchTag);
+                            TimeUnit.SECONDS.sleep(1); // Sleep before we yield the lock to avoid hammering the server from multiple processes
                         } catch (Exception ex) {
                             feedData = new FetchResult.TransientError();
                         }
@@ -212,7 +307,6 @@ public class FeedFetcherService {
     }
 
     private FetchResult fetchFeedData(FeedDefinition feed,
-                                      HttpClient client,
                                       ExecutorService executorService,
                                       @Nullable String ifModifiedSinceDate,
                                       @Nullable String ifNoneMatchTag)
@@ -220,59 +314,63 @@ public class FeedFetcherService {
         try {
             URI uri = new URI(feed.feedUrl());
 
-            HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
+            var requestBuilder = ClassicRequestBuilder.get(uri)
-                    .GET()
+                    .setHeader("User-Agent", WmsaHome.getUserAgent().uaIdentifier())
-                    .uri(uri)
+                    .setHeader("Accept-Encoding", "gzip")
-                    .header("User-Agent", WmsaHome.getUserAgent().uaIdentifier())
+                    .setHeader("Accept", "text/*, */*;q=0.9");
-                    .header("Accept-Encoding", "gzip")
-                    .header("Accept", "text/*, */*;q=0.9")
-                    .timeout(Duration.ofSeconds(15))
-                    ;
 
             // Set the If-Modified-Since or If-None-Match headers if we have them
             // though since there are certain idiosyncrasies in server implementations,
             // we avoid setting both at the same time as that may turn a 304 into a 200.
             if (ifNoneMatchTag != null) {
-                requestBuilder.header("If-None-Match", ifNoneMatchTag);
+                requestBuilder.addHeader("If-None-Match", ifNoneMatchTag);
             } else if (ifModifiedSinceDate != null) {
-                requestBuilder.header("If-Modified-Since", ifModifiedSinceDate);
+                requestBuilder.addHeader("If-Modified-Since", ifModifiedSinceDate);
             }
 
+            return httpClient.execute(requestBuilder.build(), rsp -> {
+                try {
+                    logger.info("Code: {}, URL: {}", rsp.getCode(), uri);
 
-            HttpRequest getRequest = requestBuilder.build();
+                    switch (rsp.getCode()) {
+                        case 200 -> {
+                            if (rsp.getEntity() == null) {
+                                return new FetchResult.TransientError(); // No content to read, treat as transient error
+                            }
+                            byte[] responseData = EntityUtils.toByteArray(rsp.getEntity());
 
-            for (int i = 0; i < 3; i++) {
+                            // Decode the response body based on the Content-Type header
+                            Header contentTypeHeader = rsp.getFirstHeader("Content-Type");
+                            if (contentTypeHeader == null) {
+                                return new FetchResult.TransientError();
+                            }
+                            String contentType = contentTypeHeader.getValue();
+                            String bodyText = DocumentBodyToString.getStringData(ContentType.parse(contentType), responseData);
 
-                /* Note we need to use an executor to time-limit the send() method in HttpClient, as
+                            // Grab the ETag header if it exists
-                 * its support for timeouts only applies to the time until response starts to be received,
+                            Header etagHeader = rsp.getFirstHeader("ETag");
-                 * and does not catch the case when the server starts to send data but then hangs.
+                            String newEtagValue = etagHeader == null ? null : etagHeader.getValue();
-                 */
-                HttpResponse<byte[]> rs = executorService.submit(
-                        () -> client.send(getRequest, HttpResponse.BodyHandlers.ofByteArray()))
-                                .get(15, TimeUnit.SECONDS);
 
-                if (rs.statusCode() == 429) { // Too Many Requests
+                            return new FetchResult.Success(bodyText, newEtagValue);
-                    int retryAfter = Integer.parseInt(rs.headers().firstValue("Retry-After").orElse("2"));
+                        }
-                    Thread.sleep(Duration.ofSeconds(Math.clamp(retryAfter, 1, 5)));
+                        case 304 -> {
-                    continue;
+                            return new FetchResult.NotModified(); // via If-Modified-Since semantics
-                }
+                        }
-
+                        case 404 -> {
-                String newEtagValue = rs.headers().firstValue("ETag").orElse("");
+                            return new FetchResult.PermanentError(); // never try again
-
+                        }
-                return switch (rs.statusCode()) {
+                        default -> {
-                    case 200 -> {
+                            return new FetchResult.TransientError(); // we try again later
-                        byte[] responseData = getResponseData(rs);
+                        }
-
-                        String contentType = rs.headers().firstValue("Content-Type").orElse("");
-                        String bodyText = DocumentBodyToString.getStringData(ContentType.parse(contentType), responseData);
-
-                        yield new FetchResult.Success(bodyText, newEtagValue);
                     }
-                    case 304 -> new FetchResult.NotModified(); // via If-Modified-Since semantics
+                }
-                    case 404 -> new FetchResult.PermanentError(); // never try again
+                catch (Exception ex) {
-                    default -> new FetchResult.TransientError(); // we try again later
+                    return new FetchResult.PermanentError(); // treat as permanent error
-                };
+                }
-            }
+                finally {
+                    EntityUtils.consumeQuietly(rsp.getEntity());
+                }
+            });
         }
         catch (Exception ex) {
             logger.debug("Error fetching feed", ex);
@@ -281,19 +379,6 @@ public class FeedFetcherService {
         return new FetchResult.TransientError();
     }
 
-    private byte[] getResponseData(HttpResponse<byte[]> response) throws IOException {
-        String encoding = response.headers().firstValue("Content-Encoding").orElse("");
-
-        if ("gzip".equals(encoding)) {
-            try (var stream = new GZIPInputStream(new ByteArrayInputStream(response.body()))) {
-                return stream.readAllBytes();
-            }
-        }
-        else {
-            return response.body();
-        }
-    }
-
     public sealed interface FetchResult {
         record Success(String value, String etag) implements FetchResult {}
         record NotModified() implements FetchResult {}

code/functions/live-capture/java/nu/marginalia/rss/svc/FeedsGrpcService.java

@@ -1,6 +1,7 @@
 package nu.marginalia.rss.svc;
 
 import com.google.inject.Inject;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import nu.marginalia.api.feeds.*;
 import nu.marginalia.db.DbDomainQueries;
@@ -69,7 +70,7 @@ public class FeedsGrpcService extends FeedApiGrpc.FeedApiImplBase implements Dis
     @Override
     public void getFeedDataHash(Empty request, StreamObserver<RpcFeedDataHash> responseObserver) {
         if (!feedDb.isEnabled()) {
-            responseObserver.onError(new IllegalStateException("Feed database is disabled on this node"));
+            responseObserver.onError(Status.INTERNAL.withDescription("Feed database is disabled on this node").asRuntimeException());
             return;
         }
 
@@ -80,7 +81,7 @@ public class FeedsGrpcService extends FeedApiGrpc.FeedApiImplBase implements Dis
         }
         catch (Exception e) {
             logger.error("Error getting feed data hash", e);
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -101,7 +102,7 @@ public class FeedsGrpcService extends FeedApiGrpc.FeedApiImplBase implements Dis
         }
         catch (Exception e) {
             logger.error("Error getting updated links", e);
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 
@@ -109,13 +110,13 @@ public class FeedsGrpcService extends FeedApiGrpc.FeedApiImplBase implements Dis
     public void getFeed(RpcDomainId request,
                         StreamObserver<RpcFeed> responseObserver) {
         if (!feedDb.isEnabled()) {
-            responseObserver.onError(new IllegalStateException("Feed database is disabled on this node"));
+            responseObserver.onError(Status.INTERNAL.withDescription("Feed database is disabled on this node").asRuntimeException());
             return;
         }
 
         Optional<EdgeDomain> domainName = domainQueries.getDomain(request.getDomainId());
         if (domainName.isEmpty()) {
-            responseObserver.onError(new IllegalArgumentException("Domain not found"));
+            responseObserver.onError(Status.NOT_FOUND.withDescription("Domain not found").asRuntimeException());
             return;
         }
 

code/functions/live-capture/test/nu/marginalia/rss/svc/FeedFetcherServiceTest.java

@@ -5,6 +5,8 @@ import com.google.inject.Guice;
 import com.google.inject.name.Names;
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
+import nu.marginalia.coordination.DomainCoordinator;
+import nu.marginalia.coordination.LocalDomainCoordinator;
 import nu.marginalia.model.EdgeDomain;
 import nu.marginalia.rss.db.FeedDb;
 import nu.marginalia.rss.model.FeedItems;
@@ -82,6 +84,7 @@ class FeedFetcherServiceTest extends AbstractModule {
     }
 
     public void configure() {
+        bind(DomainCoordinator.class).to(LocalDomainCoordinator.class);
         bind(HikariDataSource.class).toInstance(dataSource);
         bind(ServiceRegistryIf.class).toInstance(Mockito.mock(ServiceRegistryIf.class));
         bind(ServiceConfiguration.class).toInstance(new ServiceConfiguration(ServiceId.Executor, 1, "", "", 0, UUID.randomUUID()));

code/functions/math/api/java/nu/marginalia/api/math/MathClient.java

@@ -26,7 +26,9 @@ public class MathClient {
     private static final Logger logger = LoggerFactory.getLogger(MathClient.class);
 
     private final GrpcSingleNodeChannelPool<MathApiGrpc.MathApiBlockingStub> channelPool;
-    private final ExecutorService executor = Executors.newWorkStealingPool(8);
+
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
+    private static final ExecutorService executor = useLoom ? Executors.newVirtualThreadPerTaskExecutor() : Executors.newWorkStealingPool(8);
 
     @Inject
     public MathClient(GrpcChannelPoolFactory factory) {

code/functions/search-query/java/nu/marginalia/functions/searchquery/QueryGRPCService.java

@@ -3,6 +3,7 @@ package nu.marginalia.functions.searchquery;
 import com.google.common.collect.Lists;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.prometheus.client.Histogram;
 import nu.marginalia.api.searchquery.*;
@@ -93,7 +94,7 @@ public class QueryGRPCService
             });
         } catch (Exception e) {
             logger.error("Exception", e);
-            responseObserver.onError(e);
+            responseObserver.onError(Status.INTERNAL.withCause(e).asRuntimeException());
         }
     }
 

code/index/api/java/nu/marginalia/index/api/IndexClient.java

@@ -38,7 +38,9 @@ public class IndexClient {
             .help("Count of results filtered by NSFW tier")
             .register();
 
-    private static final ExecutorService executor = Executors.newCachedThreadPool();
+
+    private static final boolean useLoom = Boolean.getBoolean("system.experimentalUseLoom");
+    private static final ExecutorService executor = useLoom ? Executors.newVirtualThreadPerTaskExecutor() : Executors.newCachedThreadPool();
 
     @Inject
     public IndexClient(GrpcChannelPoolFactory channelPoolFactory,

code/index/java/nu/marginalia/domainranking/PageRankDomainRanker.java

@@ -1,10 +1,10 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
 import gnu.trove.list.TIntList;
 import gnu.trove.list.array.TIntArrayList;
-import nu.marginalia.ranking.domains.accumulator.RankingResultAccumulator;
+import nu.marginalia.domainranking.accumulator.RankingResultAccumulator;
-import nu.marginalia.ranking.domains.data.GraphSource;
+import nu.marginalia.domainranking.data.GraphSource;
-import nu.marginalia.ranking.domains.jgrapht.PersonalizedPageRank;
+import nu.marginalia.domainranking.jgrapht.PersonalizedPageRank;
 import org.jgrapht.Graph;
 import org.jgrapht.alg.interfaces.VertexScoringAlgorithm;
 import org.jgrapht.alg.scoring.PageRank;

code/index/java/nu/marginalia/domainranking/RankingAlgorithm.java

@@ -1,6 +1,6 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
-import nu.marginalia.ranking.domains.accumulator.RankingResultAccumulator;
+import nu.marginalia.domainranking.accumulator.RankingResultAccumulator;
 
 import java.util.function.Supplier;
 

code/index/java/nu/marginalia/domainranking/accumulator/RankingResultAccumulator.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.accumulator;
+package nu.marginalia.domainranking.accumulator;
 
 public interface RankingResultAccumulator<T> {
     void add(int domainId, int rank);

code/index/java/nu/marginalia/domainranking/accumulator/RankingResultBitSetAccumulator.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.accumulator;
+package nu.marginalia.domainranking.accumulator;
 
 import org.roaringbitmap.RoaringBitmap;
 

code/index/java/nu/marginalia/domainranking/accumulator/RankingResultHashMapAccumulator.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.accumulator;
+package nu.marginalia.domainranking.accumulator;
 
 import it.unimi.dsi.fastutil.ints.Int2IntOpenHashMap;
 

code/index/java/nu/marginalia/domainranking/accumulator/RankingResultHashSetAccumulator.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.accumulator;
+package nu.marginalia.domainranking.accumulator;
 
 import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
 

code/index/java/nu/marginalia/domainranking/accumulator/RankingResultListAccumulator.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.accumulator;
+package nu.marginalia.domainranking.accumulator;
 
 import gnu.trove.list.array.TIntArrayList;
 

code/index/java/nu/marginalia/domainranking/data/AbstractGraphSource.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.data;
+package nu.marginalia.domainranking.data;
 
 import com.zaxxer.hikari.HikariDataSource;
 import org.jgrapht.Graph;

code/index/java/nu/marginalia/domainranking/data/GraphSource.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.data;
+package nu.marginalia.domainranking.data;
 
 import org.jgrapht.Graph;
 

code/index/java/nu/marginalia/domainranking/data/InvertedLinkGraphSource.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.data;
+package nu.marginalia.domainranking.data;
 
 import com.google.inject.Inject;
 import com.zaxxer.hikari.HikariDataSource;

code/index/java/nu/marginalia/domainranking/data/LinkGraphSource.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.data;
+package nu.marginalia.domainranking.data;
 
 import com.google.inject.Inject;
 import com.zaxxer.hikari.HikariDataSource;

code/index/java/nu/marginalia/domainranking/data/SimilarityGraphSource.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.data;
+package nu.marginalia.domainranking.data;
 
 import com.google.inject.Inject;
 import com.zaxxer.hikari.HikariDataSource;

code/index/java/nu/marginalia/domainranking/jgrapht/PersonalizedPageRank.java

@@ -1,4 +1,4 @@
-package nu.marginalia.ranking.domains.jgrapht;
+package nu.marginalia.domainranking.jgrapht;
 
 /*
  * (C) Copyright 2016-2023, by Dimitrios Michail and Contributors.
@@ -21,8 +21,9 @@ package nu.marginalia.ranking.domains.jgrapht;
 
 /* (modified by @vlofgren to add personalization) */
 
-import org.jgrapht.*;
+import org.jgrapht.Graph;
-import org.jgrapht.alg.interfaces.*;
+import org.jgrapht.Graphs;
+import org.jgrapht.alg.interfaces.VertexScoringAlgorithm;
 
 import java.util.*;
 

code/index/java/nu/marginalia/index/IndexGrpcService.java

@@ -2,6 +2,7 @@ package nu.marginalia.index;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import io.grpc.Status;
 import io.grpc.stub.StreamObserver;
 import io.prometheus.client.Counter;
 import io.prometheus.client.Gauge;
@@ -148,7 +149,7 @@ public class IndexGrpcService
         }
         catch (Exception ex) {
             logger.error("Error in handling request", ex);
-            responseObserver.onError(ex);
+            responseObserver.onError(Status.INTERNAL.withCause(ex).asRuntimeException());
         }
     }
 

code/index/java/nu/marginalia/index/results/IndexResultScoreCalculator.java

@@ -551,9 +551,18 @@ public class IndexResultScoreCalculator {
             largeSiteFactor = 2;
         }
 
-        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.TRACKING_ADTECH.getFeatureBit()))
+        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.ADVERTISEMENT.getFeatureBit()))
             penalty += 7.5 * largeSiteFactor;
 
+        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.CONSENT.getFeatureBit()))
+            penalty += 2.5 * largeSiteFactor;
+
+        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.POPOVER.getFeatureBit()))
+            penalty += 2.5 * largeSiteFactor;
+
+        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.TRACKING_ADTECH.getFeatureBit()))
+            penalty += 5.0 * largeSiteFactor;
+
         if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.AFFILIATE_LINK.getFeatureBit()))
             penalty += 5.0 * largeSiteFactor;
 
@@ -563,6 +572,9 @@ public class IndexResultScoreCalculator {
         if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.TRACKING.getFeatureBit()))
             penalty += 2.5 * largeSiteFactor;
 
+        if (DocumentMetadata.hasFlags(featureFlags, HtmlFeature.SHORT_DOCUMENT.getFeatureBit()))
+            penalty += 2.5  * largeSiteFactor;
+
         if (isForum || isWiki) {
             penalty = Math.min(0, penalty - 2);
         }

code/index/java/nu/marginalia/index/searchset/SearchSetsService.java

@@ -6,14 +6,14 @@ import gnu.trove.list.TIntList;
 import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
 import nu.marginalia.db.DomainRankingSetsService;
 import nu.marginalia.db.DomainTypes;
+import nu.marginalia.domainranking.PageRankDomainRanker;
+import nu.marginalia.domainranking.accumulator.RankingResultHashMapAccumulator;
+import nu.marginalia.domainranking.accumulator.RankingResultHashSetAccumulator;
+import nu.marginalia.domainranking.data.GraphSource;
+import nu.marginalia.domainranking.data.LinkGraphSource;
+import nu.marginalia.domainranking.data.SimilarityGraphSource;
 import nu.marginalia.index.IndexFactory;
 import nu.marginalia.index.domainrankings.DomainRankings;
-import nu.marginalia.ranking.domains.PageRankDomainRanker;
-import nu.marginalia.ranking.domains.accumulator.RankingResultHashMapAccumulator;
-import nu.marginalia.ranking.domains.accumulator.RankingResultHashSetAccumulator;
-import nu.marginalia.ranking.domains.data.GraphSource;
-import nu.marginalia.ranking.domains.data.LinkGraphSource;
-import nu.marginalia.ranking.domains.data.SimilarityGraphSource;
 import nu.marginalia.service.control.ServiceEventLog;
 import nu.marginalia.service.module.ServiceConfiguration;
 import org.slf4j.Logger;

code/index/test/nu/marginalia/domainranking/RankingAlgorithmWithRealDataTest.java

@@ -1,6 +1,6 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
-import nu.marginalia.ranking.domains.accumulator.RankingResultListAccumulator;
+import nu.marginalia.domainranking.accumulator.RankingResultListAccumulator;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 

code/index/test/nu/marginalia/domainranking/RankingAlgorithmsContainerTest.java

@@ -1,12 +1,12 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
 
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import nu.marginalia.api.linkgraph.AggregateLinkGraphClient;
-import nu.marginalia.ranking.domains.data.InvertedLinkGraphSource;
+import nu.marginalia.domainranking.data.InvertedLinkGraphSource;
-import nu.marginalia.ranking.domains.data.LinkGraphSource;
+import nu.marginalia.domainranking.data.LinkGraphSource;
-import nu.marginalia.ranking.domains.data.SimilarityGraphSource;
+import nu.marginalia.domainranking.data.SimilarityGraphSource;
 import nu.marginalia.test.TestMigrationLoader;
 import org.jgrapht.Graph;
 import org.jgrapht.graph.DefaultWeightedEdge;

code/index/test/nu/marginalia/domainranking/TestGraphSourceForInvertedLinkData.java

@@ -1,7 +1,7 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
 import nu.marginalia.array.LongArrayFactory;
-import nu.marginalia.ranking.domains.data.GraphSource;
+import nu.marginalia.domainranking.data.GraphSource;
 import org.apache.commons.lang3.StringUtils;
 import org.jgrapht.Graph;
 import org.jgrapht.graph.DefaultDirectedGraph;

code/index/test/nu/marginalia/domainranking/TestGraphSourceForLinkData.java

@@ -1,7 +1,7 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
 import nu.marginalia.array.LongArrayFactory;
-import nu.marginalia.ranking.domains.data.GraphSource;
+import nu.marginalia.domainranking.data.GraphSource;
 import org.apache.commons.lang3.StringUtils;
 import org.jgrapht.Graph;
 import org.jgrapht.graph.DefaultDirectedGraph;

code/index/test/nu/marginalia/domainranking/TestGraphSourceForSimilarityData.java

@@ -1,6 +1,6 @@
-package nu.marginalia.ranking.domains;
+package nu.marginalia.domainranking;
 
-import nu.marginalia.ranking.domains.data.GraphSource;
+import nu.marginalia.domainranking.data.GraphSource;
 import org.apache.commons.lang3.StringUtils;
 import org.jgrapht.Graph;
 import org.jgrapht.graph.DefaultUndirectedWeightedGraph;

code/libraries/domain-lock/build.gradle

@@ -0,0 +1,32 @@
+plugins {
+    id 'java'
+}
+
+java {
+    toolchain {
+        languageVersion.set(JavaLanguageVersion.of(rootProject.ext.jvmVersion))
+    }
+}
+
+apply from: "$rootProject.projectDir/srcsets.gradle"
+
+dependencies {
+    implementation libs.bundles.slf4j
+    implementation project(':code:common:model')
+    implementation project(':code:common:config')
+    implementation project(':code:common:service')
+
+    implementation libs.bundles.curator
+
+    implementation libs.guava
+    implementation dependencies.create(libs.guice.get()) {
+        exclude group: 'com.google.guava'
+    }
+    testImplementation libs.bundles.slf4j.test
+    testImplementation libs.bundles.junit
+    testImplementation libs.mockito
+}
+
+test {
+    useJUnitPlatform()
+}

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

@@ -0,0 +1,32 @@
+package nu.marginalia.coordination;
+
+import nu.marginalia.model.EdgeDomain;
+
+public class DefaultDomainPermits {
+
+    public static int defaultPermits(EdgeDomain domain) {
+        return defaultPermits(domain.topDomain.toLowerCase());
+    }
+
+    public static int defaultPermits(String topDomain) {
+
+        if (topDomain.equals("wordpress.com"))
+            return 16;
+        if (topDomain.equals("blogspot.com"))
+            return 8;
+        if (topDomain.equals("tumblr.com"))
+            return 8;
+        if (topDomain.equals("neocities.org"))
+            return 8;
+        if (topDomain.equals("github.io"))
+            return 8;
+        // Substack really dislikes broad-scale crawlers, so we need to be careful
+        // to not get blocked.
+        if (topDomain.equals("substack.com")) {
+            return 1;
+        }
+
+        return 2;
+    }
+
+}

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

@@ -0,0 +1,17 @@
+package nu.marginalia.coordination;
+
+import com.google.inject.AbstractModule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DomainCoordinationModule extends AbstractModule {
+
+    private static final Logger logger = LoggerFactory.getLogger(DomainCoordinationModule.class);
+
+    public DomainCoordinationModule() {
+    }
+
+    public void configure() {
+        bind(DomainCoordinator.class).to(ZookeeperDomainCoordinator.class);
+    }
+}

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

@@ -0,0 +1,13 @@
+package nu.marginalia.coordination;
+
+import nu.marginalia.model.EdgeDomain;
+
+import java.time.Duration;
+import java.util.Optional;
+
+public interface DomainCoordinator {
+    DomainLock lockDomain(EdgeDomain domain) throws InterruptedException;
+    Optional<DomainLock> tryLockDomain(EdgeDomain domain, Duration timeout) throws InterruptedException;
+    Optional<DomainLock> tryLockDomain(EdgeDomain domain)  throws InterruptedException;
+    boolean isLockableHint(EdgeDomain domain);
+}

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

@@ -0,0 +1,5 @@
+package nu.marginalia.coordination;
+
+public interface DomainLock extends AutoCloseable {
+    void close();
+}

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

@@ -1,16 +1,17 @@
-package nu.marginalia.crawl.logic;
+package nu.marginalia.coordination;
 
+import com.google.inject.Singleton;
 import nu.marginalia.model.EdgeDomain;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
-/** Holds lock objects for each domain, to prevent multiple threads from
+@Singleton
- * crawling the same domain at the same time.
+public class LocalDomainCoordinator implements DomainCoordinator {
- */
-public class DomainLocks {
     // 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.
@@ -24,13 +25,25 @@ public class DomainLocks {
 
         sem.acquire();
 
-        return new DomainLock(sem);
+        return new LocalDomainLock(sem);
     }
 
     public Optional<DomainLock> tryLockDomain(EdgeDomain domain) {
         var sem = locks.computeIfAbsent(domain.topDomain.toLowerCase(), this::defaultPermits);
         if (sem.tryAcquire(1)) {
-            return Optional.of(new DomainLock(sem));
+            return Optional.of(new LocalDomainLock(sem));
+        }
+        else {
+            // We don't have a lock, so we return an empty optional
+            return Optional.empty();
+        }
+    }
+
+
+    public Optional<DomainLock> tryLockDomain(EdgeDomain domain, Duration timeout) throws InterruptedException {
+        var sem = locks.computeIfAbsent(domain.topDomain.toLowerCase(), this::defaultPermits);
+        if (sem.tryAcquire(1, timeout.toMillis(), TimeUnit.MILLISECONDS)) {
+            return Optional.of(new LocalDomainLock(sem));
         }
         else {
             // We don't have a lock, so we return an empty optional
@@ -39,24 +52,7 @@ public class DomainLocks {
     }
 
     private Semaphore defaultPermits(String topDomain) {
-        if (topDomain.equals("wordpress.com"))
+        return new Semaphore(DefaultDomainPermits.defaultPermits(topDomain));
-            return new Semaphore(16);
-        if (topDomain.equals("blogspot.com"))
-            return new Semaphore(8);
-        if (topDomain.equals("tumblr.com"))
-            return new Semaphore(8);
-        if (topDomain.equals("neocities.org"))
-            return new Semaphore(8);
-        if (topDomain.equals("github.io"))
-            return new Semaphore(8);
-
-        // Substack really dislikes broad-scale crawlers, so we need to be careful
-        // to not get blocked.
-        if (topDomain.equals("substack.com")) {
-            return new Semaphore(1);
-        }
-
-        return new Semaphore(2);
     }
 
     /** Returns true if the domain is lockable, i.e. if it is not already locked by another thread.
@@ -71,15 +67,15 @@ public class DomainLocks {
             return sem.availablePermits() > 0;
     }
 
-    public static class DomainLock implements AutoCloseable {
+    public static class LocalDomainLock implements DomainLock {
         private final Semaphore semaphore;
 
-        DomainLock(Semaphore semaphore) {
+        LocalDomainLock(Semaphore semaphore) {
             this.semaphore = semaphore;
         }
 
         @Override
-        public void close() throws Exception {
+        public void close() {
             semaphore.release();
             Thread.currentThread().setName("[idle]");
         }

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

@@ -0,0 +1,116 @@
+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;
+import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2;
+import org.apache.curator.framework.recipes.locks.Lease;
+
+import java.time.Duration;
+import java.util.Map;
+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;
+
+    @Inject
+    public ZookeeperDomainCoordinator(ServiceRegistryIf serviceRegistry, @Named("wmsa-system-node") int nodeId) {
+        // Zookeeper-specific initialization can be done here if needed
+        this.serviceRegistry = serviceRegistry;
+        this.nodeId = nodeId;
+    }
+
+    /** Returns a lock object corresponding to the given domain.  The object is returned as-is,
+     * and may be held by another thread.  The caller is responsible for locking and  releasing the lock.
+     */
+    public DomainLock lockDomain(EdgeDomain domain) throws InterruptedException {
+        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 {
+            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);
+
+        // 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) {
+                return Optional.of(new ZkDomainLock(sem, lease));
+            }
+            else {
+                return Optional.empty(); // If we fail to acquire the lease, we return an empty optional
+            }
+        }
+        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){
+        try {
+            return serviceRegistry.getSemaphore(topDomain + ":" + nodeId, DefaultDomainPermits.defaultPermits(topDomain));
+        }
+        catch (Exception e) {
+            throw new RuntimeException("Failed to get semaphore for domain: " + topDomain, e);
+        }
+    }
+
+    /** Returns true if the domain is lockable, i.e. if it is not already locked by another thread.
+     * (this is just a hint, and does not guarantee that the domain is actually lockable any time
+     * after this method returns true)
+     */
+    public boolean isLockableHint(EdgeDomain domain) {
+        return !waitCounts.containsKey(domain.topDomain.toLowerCase());
+    }
+
+    public static class ZkDomainLock implements DomainLock {
+        private final InterProcessSemaphoreV2 semaphore;
+        private final Lease lease;
+
+        ZkDomainLock(InterProcessSemaphoreV2 semaphore, Lease lease) {
+            this.semaphore = semaphore;
+            this.lease = lease;
+        }
+
+        @Override
+        public void close() {
+            semaphore.returnLease(lease);
+        }
+    }
+}

code/processes/converting-process/build.gradle

@@ -47,11 +47,14 @@ dependencies {
 
     implementation project(':code:processes:converting-process:ft-anchor-keywords')
     implementation project(':code:processes:converting-process:ft-keyword-extraction')
+    implementation project(':code:processes:converting-process:ft-dom-classifier')
 
     implementation project(':code:processes:crawling-process:ft-crawl-blocklist')
     implementation project(':code:processes:crawling-process:ft-link-parser')
     implementation project(':code:processes:crawling-process:ft-content-type')
 
+    implementation project(':code:functions:live-capture:api')
+
     testImplementation project(':code:libraries:term-frequency-dict')
     testImplementation project(':code:processes:crawling-process:model')
 

code/processes/converting-process/ft-dom-classifier/build.gradle

@@ -0,0 +1,40 @@
+plugins {
+    id 'java'
+    id "de.undercouch.download" version "5.1.0"
+    id 'jvm-test-suite'
+}
+
+java {
+    toolchain {
+        languageVersion.set(JavaLanguageVersion.of(rootProject.ext.jvmVersion))
+    }
+}
+
+apply from: "$rootProject.projectDir/srcsets.gradle"
+
+dependencies {
+    implementation project(':code:common:config')
+    implementation project(':code:common:service')
+    implementation project(':code:common:model')
+    implementation project(':code:common:db')
+    implementation project(':code:functions:live-capture:api')
+
+
+    implementation libs.bundles.slf4j
+    implementation libs.guava
+    implementation dependencies.create(libs.guice.get()) {
+        exclude group: 'com.google.guava'
+    }
+    implementation libs.trove
+    implementation libs.gson
+    implementation libs.bundles.protobuf
+    implementation libs.bundles.mariadb
+    implementation libs.duckdb
+    implementation libs.notnull
+    implementation libs.jsoup
+
+    testImplementation libs.bundles.slf4j.test
+    testImplementation libs.bundles.junit
+    testImplementation libs.mockito
+}
+

code/processes/converting-process/ft-dom-classifier/java/nu/marginalia/ddtrackergradar/DDGTrackerData.java

@@ -0,0 +1,99 @@
+package nu.marginalia.ddtrackergradar;
+
+import com.google.gson.Gson;
+import nu.marginalia.WmsaHome;
+import nu.marginalia.ddtrackergradar.model.DDGTDomain;
+import nu.marginalia.model.gson.GsonFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.*;
+
+/** Holds tracker metadata from DuckDuckGo's Tracker Radar
+ *  data itself CC-BY-NC-SA 4.0
+ * */
+public class DDGTrackerData {
+    private final Map<String, DDGTDomain> topDomains = new HashMap<>();
+    private final Map<String, DDGTDomain> domains = new HashMap<>();
+
+    private final Gson gson = GsonFactory.get();
+
+    private static final Logger logger = LoggerFactory.getLogger(DDGTrackerData.class);
+
+    public DDGTrackerData() {
+
+        // Data is assumed to be in ${WMSA_HOME}/data/tracker-radar
+        // ... do a shallow clone of the repo
+        // https://github.com/duckduckgo/tracker-radar/
+
+        Path dataDir = WmsaHome.getDataPath().resolve("tracker-radar");
+        if (!Files.exists(dataDir)) {
+            logger.info("tracker-radar data absent from expected path {}, loading nothing", dataDir);
+            return;
+        }
+
+        try (var sources = Files.list(dataDir.resolve("domains"))) {
+            sources.filter(Files::isDirectory).forEach(this::loadDomainDir);
+        }
+        catch (IOException e) {
+            logger.error("Failed to read tracker radar data dir", e);
+        }
+    }
+
+    /** Tries to fetch available information about tracking coming from the specified domain
+     */
+    public Optional<DDGTDomain> getDomainInfo(String domain) {
+        return Optional
+                .ofNullable(topDomains.get(domain))
+                .or(() -> Optional.ofNullable(domains.get(domain)));
+    }
+
+    /** public for testing */
+    public void loadDomainDir(Path dir) {
+        try (var dirContent = Files.list(dir)) {
+            dirContent
+                    .filter(Files::isRegularFile)
+                    .filter(path -> path.toString().endsWith(".json"))
+                    .forEach(this::loadDomainModel);
+        }
+        catch (IOException e) {
+            logger.error("Error while loading DDGT tracker data", e);
+        }
+    }
+
+    void loadDomainModel(Path jsonFile) {
+        try {
+            var model = gson.fromJson(Files.readString(jsonFile), DDGTDomain.class);
+
+            if (model.domain() == null)
+                return;
+            if ((model.owner() == null || model.owner().isEmpty())
+             && (model.categories() == null || model.categories().isEmpty()))
+                return;
+
+            topDomains.put(model.domain(), model);
+            domains.put(model.domain(), model);
+
+            if (model.subdomains() != null) {
+                for (String subdomain : model.subdomains()) {
+                    domains.put(subdomain + "." + model.domain(), model);
+                }
+            }
+        }
+        catch (Exception e) {
+            logger.error("Error while loading DDGT tracker data", e);
+        }
+    }
+
+    // Export all classifications in the data set
+    public Set<String> getAllClassifications() {
+        Set<String> ret = new HashSet<>();
+        for (var domain: domains.values()) {
+            ret.addAll(domain.categories());
+        }
+        return ret;
+    }
+}