Enforce robots.txt Crawl-delay in URLFrontier queues

core/src/main/java/org/apache/stormcrawler/Constants.java

@@ -29,6 +29,13 @@ public class Constants {
     public static final String STATUS_ERROR_SOURCE = "error.source";
     public static final String STATUS_ERROR_CAUSE = "error.cause";
 
+    /**
+     * Metadata key holding the robots.txt Crawl-delay in seconds, set by the fetcher bolts when the
+     * delay exceeds {@code fetcher.max.crawl.delay} and {@code fetcher.max.crawl.delay.force} is
+     * true, so a frontier-side consumer can enforce it at the source. See #867.
+     */
+    public static final String ROBOTS_CRAWL_DELAY_KEY = "robots.crawl.delay";
+
     public static final String QUEUE_STREAM_NAME = "queue";
 
     public static final String StatusStreamName = "status";

core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java

@@ -583,6 +583,9 @@ public void run() {
 
                 // https://github.com/apache/stormcrawler/issues/813
                 metadata.remove("fetch.exception");
+                metadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY);
+
+                String robotsCrawlDelaySecs = null;
 
                 boolean asap = false;
 
@@ -677,6 +680,12 @@ public void run() {
                                     msg);
                             if (force) {
                                 fiq.crawlDelay = maxCrawlDelay;
+                                // report the delay the fetcher is not holding, so a frontier-side
+                                // consumer can enforce it at the source (#867)
+                                robotsCrawlDelaySecs =
+                                        Long.toString(1L + ((rules.getCrawlDelay() - 1L) / 1000L));
+                                metadata.setValue(
+                                        Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs);
                             } else {
                                 // pass the info about crawl delay
                                 metadata.setValue(Constants.STATUS_ERROR_CAUSE, "crawl_delay");
@@ -792,6 +801,14 @@ public void run() {
                     // metadata persisted or transferred from previous fetches
                     mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix);
 
+                    // Only the locally parsed robots.txt value may populate this control signal.
+                    // A colliding protocol prefix/header must not pace an unrelated queue.
+                    mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY);
+                    if (robotsCrawlDelaySecs != null) {
+                        mergedMetadata.setValue(
+                                Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs);
+                    }
+
                     mergedMetadata.setValue(
                             "fetch.statusCode", Integer.toString(response.getStatusCode()));
 

core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java

@@ -273,6 +273,9 @@ public void execute(Tuple input) {
 
         // https://github.com/apache/stormcrawler/issues/813
         metadata.remove("fetch.exception");
+        metadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY);
+
+        String robotsCrawlDelaySecs = null;
 
         URL url;
 
@@ -401,12 +404,16 @@ public void execute(Tuple input) {
             // value is negative when not set
             long robotsDelay = rules.getCrawlDelay();
             if (robotsDelay > 0) {
-                if (robotsDelay > maxCrawlDelay) {
+                if (robotsDelay > maxCrawlDelay && maxCrawlDelay >= 0) {
                     if (maxCrawlDelayForce) {
                         // cap the value to a maximum
                         // as some sites specify ridiculous values
                         LOG.debug("Delay from robots capped at {} for {}", robotsDelay, url);
                         delay = maxCrawlDelay;
+                        // report the delay the fetcher is not holding, so a frontier-side
+                        // consumer can enforce it at the source (#867)
+                        robotsCrawlDelaySecs = Long.toString(1L + ((robotsDelay - 1L) / 1000L));
+                        metadata.setValue(Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs);
                     } else {
                         LOG.debug(
                                 "Skipped URL from queue with overlong crawl-delay ({}): {}",
@@ -499,6 +506,13 @@ public void execute(Tuple input) {
             // persisted or transferred from previous fetches
             mergedMetadata.putAll(response.getMetadata(), protocolMetadataPrefix);
 
+            // Only the locally parsed robots.txt value may populate this control signal.
+            // A colliding protocol prefix/header must not pace an unrelated queue.
+            mergedMetadata.remove(Constants.ROBOTS_CRAWL_DELAY_KEY);
+            if (robotsCrawlDelaySecs != null) {
+                mergedMetadata.setValue(Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs);
+            }
+
             mergedMetadata.setValue("fetch.statusCode", Integer.toString(response.getStatusCode()));
 
             mergedMetadata.setValue("fetch.loadingTime", Long.toString(timeFetching));

core/src/main/resources/crawler-default.yaml

@@ -40,8 +40,11 @@ config:
   #    skip URLs from this queue to avoid that any overlong
   #    crawl-delay throttles the crawler
   #  (if true)
-  #    set the delay to fetcher.max.crawl.delay,
-  #    making fetcher more aggressive than requested
+  #    cap the local delay at fetcher.max.crawl.delay and emit the
+  #    requested value in whole seconds as robots.crawl.delay;
+  #    URLFrontier's QueueRegulatorBolt can forward a capped value when
+  #    its robots crawl-delay integration is explicitly enabled. Without
+  #    that consumer the fetcher remains more aggressive than requested.
   fetcher.max.crawl.delay.force: false
   # behavior of fetcher when the crawl-delay in the robots.txt
   # is smaller (ev. less than one second) than the default delay:

core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java

@@ -192,7 +192,59 @@ void invalidProxyMetadataEmitsFetchError(WireMockRuntimeInfo wmRuntimeInfo)
         Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size());
     }
 
-    private void resetProtocolFactory() throws ReflectiveOperationException {
+    TestOutputCollector fetch(
+            WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+            throws ReflectiveOperationException {
+        resetProtocolFactory();
+        TestOutputCollector output = new TestOutputCollector();
+        bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+        Tuple tuple = mock(Tuple.class);
+        when(tuple.getSourceComponent()).thenReturn("source");
+        when(tuple.getStringByField("url"))
+                .thenReturn("http://localhost:" + wmRuntimeInfo.getHttpPort() + path);
+        when(tuple.getValueByField("metadata")).thenReturn(null);
+        bolt.execute(tuple);
+
+        await().atMost(30, TimeUnit.SECONDS)
+                .until(
+                        () ->
+                                output.getEmitted(Utils.DEFAULT_STREAM_ID).size() > 0
+                                        || output.getEmitted(Constants.StatusStreamName).size()
+                                                > 0);
+        return output;
+    }
+
+    /**
+     * Fetches a page that is expected to be retrieved successfully (HTTP 200, non-304): the fetcher
+     * emits it on the default stream (url, content, metadata) for downstream parsing.
+     */
+    Metadata fetchAndGetContentMetadata(
+            WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+            throws ReflectiveOperationException {
+        TestOutputCollector output = fetch(wmRuntimeInfo, config, path);
+        List<List<Object>> contentTuples = output.getEmitted(Utils.DEFAULT_STREAM_ID);
+        Assertions.assertEquals(1, contentTuples.size());
+        Assertions.assertEquals(0, output.getEmitted(Constants.StatusStreamName).size());
+        return (Metadata) contentTuples.get(0).get(2);
+    }
+
+    /**
+     * Fetches a page that is expected to be rejected before any HTTP request is made (e.g. the
+     * crawl-delay-too-long guard): the fetcher emits directly on the status stream (url, metadata,
+     * status).
+     */
+    List<Object> fetchAndGetStatusTuple(
+            WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+            throws ReflectiveOperationException {
+        TestOutputCollector output = fetch(wmRuntimeInfo, config, path);
+        List<List<Object>> statusTuples = output.getEmitted(Constants.StatusStreamName);
+        Assertions.assertEquals(1, statusTuples.size());
+        Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size());
+        return statusTuples.get(0);
+    }
+
+    static void resetProtocolFactory() throws ReflectiveOperationException {
         Field instance = ProtocolFactory.class.getDeclaredField("single_instance");
         instance.setAccessible(true);
         instance.set(null, null);

core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java

@@ -17,12 +17,125 @@
 
 package org.apache.stormcrawler.bolt;
 
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.stormcrawler.Constants;
+import org.apache.stormcrawler.Metadata;
+import org.apache.stormcrawler.persistence.Status;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 public class FetcherBoltTest extends AbstractFetcherBoltTest {
 
     @BeforeEach
     void setUpContext() throws Exception {
         bolt = new FetcherBolt();
     }
+
+    @Test
+    void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo)
+            throws ReflectiveOperationException {
+        // robots.txt with a Crawl-delay above fetcher.max.crawl.delay
+        stubFor(
+                get(urlEqualTo("/robots.txt"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withBody("User-agent: *\nCrawl-delay: 120\n")));
+        stubFor(
+                get(urlEqualTo("/page"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withHeader("Robots.Crawl.Delay", "1")
+                                        .withBody("hello")));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("fetcher.max.crawl.delay", 30);
+        config.put("fetcher.max.crawl.delay.force", true);
+
+        Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+        assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+    }
+
+    @Test
+    void fractionalLongCrawlDelayIsRoundedUp(WireMockRuntimeInfo wmRuntimeInfo)
+            throws ReflectiveOperationException {
+        stubFor(
+                get(urlEqualTo("/robots.txt"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withBody("User-agent: *\nCrawl-delay: 30.5\n")));
+        stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello")));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("fetcher.max.crawl.delay", 30);
+        config.put("fetcher.max.crawl.delay.force", true);
+
+        Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+        assertEquals("31", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+    }
+
+    @Test
+    void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo)
+            throws ReflectiveOperationException {
+        stubFor(
+                get(urlEqualTo("/robots.txt"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withBody("User-agent: *\nCrawl-delay: 5\n")));
+        stubFor(
+                get(urlEqualTo("/page"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withHeader("Robots.Crawl.Delay", "120")
+                                        .withBody("hello")));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("fetcher.max.crawl.delay", 30);
+        config.put("fetcher.max.crawl.delay.force", true);
+
+        Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+        assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+    }
+
+    @Test
+    void unforcedLongCrawlDelayStillEmitsCrawlDelayErrorWithoutMetadata(
+            WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException {
+        // regression guard: force=false + long delay must keep today's behaviour, i.e. a
+        // Status.ERROR/crawl_delay emission and no robots.crawl.delay metadata key
+        stubFor(
+                get(urlEqualTo("/robots.txt"))
+                        .willReturn(
+                                aResponse()
+                                        .withStatus(200)
+                                        .withBody("User-agent: *\nCrawl-delay: 120\n")));
+        stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello")));
+
+        Map<String, Object> config = new HashMap<>();
+        config.put("http.agent.name", "this_is_only_a_test");
+        config.put("fetcher.max.crawl.delay", 30);
+        config.put("fetcher.max.crawl.delay.force", false);
+
+        List<Object> statusTuple = fetchAndGetStatusTuple(wmRuntimeInfo, config, "/page");
+        assertEquals(Status.ERROR, statusTuple.get(2));
+        Metadata md = (Metadata) statusTuple.get(1);
+        assertEquals("crawl_delay", md.getFirstValue(Constants.STATUS_ERROR_CAUSE));
+        assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+    }
 }