From 83e44354bf397aabbd3820521cc26540288a4976 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 8 Jul 2026 14:27:41 +0200 Subject: [PATCH 1/8] feat: report the forced robots crawl-delay in the status metadata When a robots.txt Crawl-delay exceeds fetcher.max.crawl.delay and the force flag makes the fetcher keep going anyway, FetcherBolt now writes the requested delay in seconds to the new robots.crawl.delay metadata key. A frontier-side consumer can then enforce the pace at the source instead of losing it, which is the gap #1979 describes for #867. The unforced path is untouched and keeps emitting Status.ERROR with error.cause=crawl_delay, pinned by a regression test. --- .../org/apache/stormcrawler/Constants.java | 7 ++ .../apache/stormcrawler/bolt/FetcherBolt.java | 5 ++ .../bolt/AbstractFetcherBoltTest.java | 54 ++++++++++++- .../stormcrawler/bolt/FetcherBoltTest.java | 81 +++++++++++++++++++ 4 files changed, 146 insertions(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/stormcrawler/Constants.java b/core/src/main/java/org/apache/stormcrawler/Constants.java index e06552a6a..882f2d965 100644 --- a/core/src/main/java/org/apache/stormcrawler/Constants.java +++ b/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"; diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java index 2439bd76c..2b0b263f9 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java @@ -677,6 +677,11 @@ 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) + metadata.setValue( + Constants.ROBOTS_CRAWL_DELAY_KEY, + Long.toString(rules.getCrawlDelay() / 1000L)); } else { // pass the info about crawl delay metadata.setValue(Constants.STATUS_ERROR_CAUSE, "crawl_delay"); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java index 4f4aa521b..35cd7b5f5 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/AbstractFetcherBoltTest.java +++ b/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 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 config, String path) + throws ReflectiveOperationException { + TestOutputCollector output = fetch(wmRuntimeInfo, config, path); + List> 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 fetchAndGetStatusTuple( + WireMockRuntimeInfo wmRuntimeInfo, Map config, String path) + throws ReflectiveOperationException { + TestOutputCollector output = fetch(wmRuntimeInfo, config, path); + List> 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); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java index 8019bcf34..3f872c145 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java @@ -17,7 +17,22 @@ 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 { @@ -25,4 +40,70 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest { 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).withBody("hello"))); + + Map 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 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).withBody("hello"))); + + Map 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 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 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)); + } } From bbf317b51064fa8f8f2de059a2097c3dd967d219 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 8 Jul 2026 15:43:16 +0200 Subject: [PATCH 2/8] feat: report the forced robots crawl-delay from SimpleFetcherBolt too Same signal as in FetcherBolt: when maxCrawlDelayForce caps a long Crawl-delay, the original robots value lands in robots.crawl.delay so the frontier side of #1979 sees both fetchers alike. The shared test harness moves up to AbstractFetcherBoltTest rather than being copied. --- .../stormcrawler/bolt/SimpleFetcherBolt.java | 5 ++ .../bolt/SimpleFetcherBoltTest.java | 54 +++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java index da40b069a..2c47dee12 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java @@ -407,6 +407,11 @@ public void execute(Tuple input) { // 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) + metadata.setValue( + Constants.ROBOTS_CRAWL_DELAY_KEY, + Long.toString(robotsDelay / 1000L)); } else { LOG.debug( "Skipped URL from queue with overlong crawl-delay ({}): {}", diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java index da2120228..91565228e 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java @@ -17,7 +17,20 @@ 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.Map; +import org.apache.stormcrawler.Constants; +import org.apache.stormcrawler.Metadata; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest { @@ -25,4 +38,45 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest { void setUpContext() throws Exception { bolt = new SimpleFetcherBolt(); } + + @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).withBody("hello"))); + + Map 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 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).withBody("hello"))); + + Map 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)); + } } From 9ec07cff245a9e9006acceba251f0928f99c5dd6 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Thu, 9 Jul 2026 07:52:33 +0200 Subject: [PATCH 3/8] feat: add CrawlDelayPolicy for forwarding robots delays to the frontier Package-private decision logic for the queue-stream consumer, same split as HostBackoff: reads robots.crawl.delay from the tuple metadata, caps it with urlfrontier.backoff.max.secs since sites do request absurd values (#1979), and deduplicates per host. An unchanged value is re-sent at most once per urlfrontier.backoff.decay.secs window, which also lets a lost fire-and-forget RPC converge. The shared _DEFAULT_ queue is skipped, as one host must not slow down the catch-all bucket. --- .../urlfrontier/CrawlDelayPolicy.java | 105 +++++++++++++++++ .../urlfrontier/CrawlDelayPolicyTest.java | 111 ++++++++++++++++++ 2 files changed, 216 insertions(+) create mode 100644 external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java create mode 100644 external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java new file mode 100644 index 000000000..c7c1ea41d --- /dev/null +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java @@ -0,0 +1,105 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.urlfrontier; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Ticker; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.util.ConfUtils; + +/** + * Decides whether a queue-stream tuple carries a robots.txt crawl-delay worth forwarding to the + * frontier via {@code setDelay}. The fetcher reports the delay (in seconds, {@code + * robots.crawl.delay} metadata) only when it exceeds {@code fetcher.max.crawl.delay} and {@code + * fetcher.max.crawl.delay.force} lets it keep fetching; the frontier then paces the queue at the + * requested rate, restoring the compliance that force alone would violate. + * + *

The signal is declarative — the last value wins, there is no escalation and no interaction + * with {@link HostBackoff}'s blocks. Values are capped by {@code urlfrontier.backoff.max.secs} + * (robots can request absurd delays; the parse-time cap is disabled in StormCrawler) and + * deduplicated per host: an unchanged value is re-sent at most once per {@code + * urlfrontier.backoff.decay.secs} window, which also makes a lost fire-and-forget call converge. + * + *

Not thread-safe; a bolt task calls {@link #delaySecsFor} from a single thread. + */ +final class CrawlDelayPolicy { + + /** Last delay sent per host; expiry on write bounds both memory and RPC-loss staleness. */ + private final Cache lastSent; + + private final long maxSecs; + + CrawlDelayPolicy(Map conf) { + this(conf, Ticker.systemTicker()); + } + + /** Visible for tests: a controllable ticker drives the dedupe window. */ + CrawlDelayPolicy(Map conf, Ticker ticker) { + this.maxSecs = + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT); + long decaySecs = + Math.max( + 0L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_DECAY_KEY, + Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT)); + this.lastSent = + Caffeine.newBuilder() + .expireAfterWrite(decaySecs, TimeUnit.SECONDS) + .ticker(ticker) + .build(); + } + + /** + * @return the delay to forward to the frontier in seconds, or {@code -1} when the tuple carries + * none, the value is unchanged, or the shared {@code _DEFAULT_} queue is targeted + */ + int delaySecsFor(String key, Metadata metadata) { + if (metadata == null || HostBackoff.DEFAULT_QUEUE_KEY.equals(key)) { + return -1; + } + final String value = + metadata.getFirstValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY); + if (value == null) { + return -1; + } + long secs; + try { + secs = Long.parseLong(value); + } catch (NumberFormatException e) { + return -1; + } + if (secs <= 0) { + return -1; + } + final int delay = (int) Math.min(secs, Math.min(maxSecs, Integer.MAX_VALUE)); + final Integer previous = lastSent.getIfPresent(key); + if (previous != null && previous == delay) { + return -1; + } + lastSent.put(key, delay); + return delay; + } +} diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java new file mode 100644 index 000000000..aed6abf5b --- /dev/null +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java @@ -0,0 +1,111 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.urlfrontier; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.github.benmanes.caffeine.cache.Ticker; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.stormcrawler.Metadata; +import org.junit.jupiter.api.Test; + +/** + * Unit coverage for {@link CrawlDelayPolicy} - whether a queue-stream tuple carries a robots + * crawl-delay worth forwarding to the frontier, with per-host dedupe bounded by the decay window. + */ +class CrawlDelayPolicyTest { + + private static final String KEY = "example.com"; + + private static final class FakeTicker implements Ticker { + private long nanos; + + @Override + public long read() { + return nanos; + } + + void advanceSecs(long secs) { + nanos += TimeUnit.SECONDS.toNanos(secs); + } + } + + private final FakeTicker ticker = new FakeTicker(); + + private CrawlDelayPolicy policy(Object... keysAndValues) { + Map conf = new HashMap<>(); + for (int i = 0; i < keysAndValues.length; i += 2) { + conf.put((String) keysAndValues[i], keysAndValues[i + 1]); + } + return new CrawlDelayPolicy(conf, ticker); + } + + private static Metadata md(String delay) { + Metadata md = new Metadata(); + if (delay != null) { + md.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, delay); + } + return md; + } + + @Test + void newValueIsForwarded() { + assertEquals(120, policy().delaySecsFor(KEY, md("120"))); + } + + @Test + void unchangedValueIsDeduplicatedWithinTheWindow() { + CrawlDelayPolicy policy = policy(); + assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + assertEquals(-1, policy.delaySecsFor(KEY, md("120"))); + } + + @Test + void changedValueIsForwarded() { + CrawlDelayPolicy policy = policy(); + assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + assertEquals(60, policy.delaySecsFor(KEY, md("60"))); + } + + @Test + void unchangedValueIsResentAfterTheWindow() { + // the dedupe entry expires so a lost RPC converges within one window + CrawlDelayPolicy policy = policy(); + assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + ticker.advanceSecs(Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT + 1); + assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + } + + @Test + void valueIsCappedAtBackoffMax() { + CrawlDelayPolicy policy = policy(Constants.URLFRONTIER_BACKOFF_MAX_KEY, 3600); + assertEquals(3600, policy.delaySecsFor(KEY, md("86400"))); + } + + @Test + void malformedAbsentZeroAndDefaultQueueAreIgnored() { + CrawlDelayPolicy policy = policy(); + assertEquals(-1, policy.delaySecsFor(KEY, md("not-a-number"))); + assertEquals(-1, policy.delaySecsFor(KEY, md("0"))); + assertEquals(-1, policy.delaySecsFor(KEY, md(null))); + assertEquals(-1, policy.delaySecsFor(KEY, null)); + assertEquals(-1, policy.delaySecsFor("_DEFAULT_", md("120"))); + } +} From 47ea808a530b2f08b8d6d40b1b57b051718b3c23 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Thu, 9 Jul 2026 09:18:07 +0200 Subject: [PATCH 4/8] feat: pace host queues on robots crawl-delay via setDelay QueueRegulatorBolt now lives up to its name: besides blocking queues on rate limits, it forwards the robots.crawl.delay carried by the queue stream to URLFrontier through setDelay, closing the loop of #1979. The call is fire-and-forget with the same short deadline as the blocks, and setDelay(key, 0) is deliberately never sent because it would erase a server-side default for the queue: a site that drops its Crawl-delay keeps the old pace, which costs throughput on that host but never politeness. The startup metadata.persist probe now checks robots.crawl.delay too, since the signal dies silently if the filter strips it. --- .../urlfrontier/QueueRegulatorBolt.java | 49 +++++++++++++++++++ .../QueueRegulatorBoltDecisionTest.java | 18 ++++--- 2 files changed, 60 insertions(+), 7 deletions(-) diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java index ae403b124..139ca02d9 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java @@ -22,6 +22,7 @@ import crawlercommons.urlfrontier.URLFrontierGrpc.URLFrontierStub; import crawlercommons.urlfrontier.Urlfrontier.BlockQueueParams; import crawlercommons.urlfrontier.Urlfrontier.Empty; +import crawlercommons.urlfrontier.Urlfrontier.QueueDelayParams; import io.grpc.ManagedChannel; import io.grpc.stub.StreamObserver; import java.util.Collections; @@ -68,6 +69,17 @@ * block so that hosts blocked on the same schedule do not all retry at once, and a block is only * ever extended, never shrunk by a later, shorter signal. * + *

Independently of rate limits, when the tuple's metadata carries {@code robots.crawl.delay} + * (written by the fetcher bolts when a robots.txt {@code Crawl-delay} exceeds {@code + * fetcher.max.crawl.delay} and {@code fetcher.max.crawl.delay.force} is true), the bolt forwards it + * to the frontier via {@code setDelay}, capped by {@code urlfrontier.backoff.max.secs}: the queue + * is then served at the pace robots requested, restoring the compliance that the force setting + * alone would violate. The value is declarative — the last one seen wins, re-sent at most once per + * {@code urlfrontier.backoff.decay.secs} window. If a site later removes its Crawl-delay, no signal + * arrives and the frontier keeps the old pace: throughput loss on that host, never impoliteness; + * {@code setDelay(key, 0)} is deliberately never called as it would erase a server-side default for + * the queue. + * *

Wire it with a fields grouping on {@code "key"} from the status updater's {@code queue} * stream: the per-host state lives in the bolt task, so all signals for a host must reach the same * task. The {@code "key"} is the frontier queue key derived from {@code partition.url.mode} (the @@ -81,6 +93,7 @@ * metadata.persist: * - fetch.statusCode * - protocol.retry-after + * - robots.crawl.delay * * * (the second entry depends on {@code protocol.md.prefix}), and {@code fetch.exception} as well @@ -124,6 +137,20 @@ public void onError(Throwable t) { public void onCompleted() {} }; + private static final StreamObserver SET_DELAY_NOOP_OBSERVER = + new StreamObserver<>() { + @Override + public void onNext(Empty value) {} + + @Override + public void onError(Throwable t) { + LOG.warn("setDelay failed", t); + } + + @Override + public void onCompleted() {} + }; + private OutputCollector collector; private ManagedChannel channel; private URLFrontierStub frontier; @@ -132,12 +159,16 @@ public void onCompleted() {} /** Per-host back-off state and decision logic. */ private HostBackoff backoff; + /** Per-host robots crawl-delay forwarding decision. */ + private CrawlDelayPolicy crawlDelayPolicy; + @Override public void prepare(Map conf, TopologyContext context, OutputCollector c) { this.collector = c; this.globalCrawlID = ConfUtils.getString(conf, Constants.URLFRONTIER_CRAWL_ID_KEY, CrawlID.DEFAULT); this.backoff = new HostBackoff(conf); + this.crawlDelayPolicy = new CrawlDelayPolicy(conf); // the status updater emits the queue stream after the metadata.persist // filter: warn upfront when the keys this bolt relies on would be // stripped, as the bolt would otherwise silently never block anything @@ -191,6 +222,19 @@ public void execute(Tuple t) { frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) .blockQueueUntil(params, NOOP_OBSERVER); } + final int delaySecs = crawlDelayPolicy.delaySecsFor(key, metadata); + if (delaySecs > 0) { + LOG.debug("Setting delay of {}s on queue {}", delaySecs, key); + QueueDelayParams delayParams = + QueueDelayParams.newBuilder() + .setKey(key) + .setCrawlID(globalCrawlID) + .setDelayRequestable(delaySecs) + .setLocal(false) + .build(); + frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) + .setDelay(delayParams, SET_DELAY_NOOP_OBSERVER); + } collector.ack(t); } @@ -205,6 +249,7 @@ static Set missingQueueStreamKeys( probe.setValue(HostBackoff.STATUS_CODE_KEY, "429"); probe.setValue(retryAfterKey, "1"); probe.setValue(HostBackoff.EXCEPTION_KEY, "probe"); + probe.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, "1"); Metadata filtered = MetadataTransfer.getInstance(conf).filter(probe); Set missing = new LinkedHashSet<>(); if (filtered.getFirstValue(HostBackoff.STATUS_CODE_KEY) == null) { @@ -216,6 +261,10 @@ static Set missingQueueStreamKeys( if (onExceptions && filtered.getFirstValue(HostBackoff.EXCEPTION_KEY) == null) { missing.add(HostBackoff.EXCEPTION_KEY); } + if (filtered.getFirstValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY) + == null) { + missing.add(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY); + } return missing; } diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java index 8f068ca2d..7ca59b1dd 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java @@ -37,17 +37,19 @@ class QueueRegulatorBoltDecisionTest { private static final String RETRY_AFTER_KEY = "protocol.retry-after"; @Test - void missingQueueStreamKeysReportsBothWithDefaultConfig() { - // with the default metadata.persist neither key survives the filter + void missingQueueStreamKeysReportsAllWithDefaultConfig() { + // with the default metadata.persist none of the keys survives the filter Set missing = QueueRegulatorBolt.missingQueueStreamKeys(new HashMap<>(), RETRY_AFTER_KEY, false); - assertEquals(Set.of("fetch.statusCode", RETRY_AFTER_KEY), missing); + assertEquals(Set.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay"), missing); } @Test - void missingQueueStreamKeysEmptyWhenBothPersisted() { + void missingQueueStreamKeysEmptyWhenAllPersisted() { Map conf = new HashMap<>(); - conf.put("metadata.persist", List.of("fetch.statusCode", RETRY_AFTER_KEY)); + conf.put( + "metadata.persist", + List.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay")); assertTrue( QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, false).isEmpty()); } @@ -55,7 +57,7 @@ void missingQueueStreamKeysEmptyWhenBothPersisted() { @Test void missingQueueStreamKeysHonoursWildcards() { Map conf = new HashMap<>(); - conf.put("metadata.persist", List.of("fetch.*", "protocol.*")); + conf.put("metadata.persist", List.of("fetch.*", "protocol.*", "robots.*")); assertTrue( QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, false).isEmpty()); } @@ -63,7 +65,9 @@ void missingQueueStreamKeysHonoursWildcards() { @Test void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() { Map conf = new HashMap<>(); - conf.put("metadata.persist", List.of("fetch.statusCode", RETRY_AFTER_KEY)); + conf.put( + "metadata.persist", + List.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay")); // the same configuration is complete without the exceptions gate and // incomplete with it assertTrue( From 9f7b8de659bfa64908d4aad4fdfa2008e0371d5c Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Thu, 9 Jul 2026 10:11:54 +0200 Subject: [PATCH 5/8] test: prove the frontier actually paces a queue on robots crawl-delay Testcontainers regression against a real frontier. The delay does not shrink a batch, it gates the queue between hand-outs (lastProduced + delay, first hand-out always served): so the test takes one URL out, then asserts a later getURLs comes back empty although a second URL is neither served nor in-flight. The pause between the two calls also pins the unit of setDelayRequestable as seconds, a millisecond window would have expired. Seeding is generalised to seedUrls(paths...) with the old seedOneUrl() delegating, existing tests untouched. --- .../urlfrontier/QueueRegulatorBoltTest.java | 86 +++++++++++++++++-- 1 file changed, 78 insertions(+), 8 deletions(-) diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java index 87a88dcc1..ec71bcf6a 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java @@ -18,6 +18,7 @@ package org.apache.stormcrawler.urlfrontier; import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -90,6 +91,11 @@ private Map frontierConfig() { /** Seeds one DISCOVERED url for {@link #HOST} so the queue exists and is fetchable. */ private void seedOneUrl() { + seedUrls("/"); + } + + /** Seeds one DISCOVERED url for {@link #HOST} per path so the queue exists and is fetchable. */ + private void seedUrls(String... paths) { StatusUpdaterBolt seeder = new StatusUpdaterBolt(); TestOutputCollector out = new TestOutputCollector(); Map config = frontierConfig(); @@ -101,18 +107,30 @@ private void seedOneUrl() { config.put("urlfrontier.cache.expireafter.sec", 10); seeder.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(out)); - String url = "http://" + HOST + "/"; - Tuple tuple = mock(Tuple.class); - when(tuple.getValueByField("status")).thenReturn(Status.DISCOVERED); - when(tuple.getStringByField("url")).thenReturn(url); - when(tuple.getValueByField("metadata")).thenReturn(new Metadata()); - seeder.execute(tuple); + Set urls = new HashSet<>(); + for (String path : paths) { + String url = "http://" + HOST + path; + urls.add(url); + Tuple tuple = mock(Tuple.class); + when(tuple.getValueByField("status")).thenReturn(Status.DISCOVERED); + when(tuple.getStringByField("url")).thenReturn(url); + when(tuple.getValueByField("metadata")).thenReturn(new Metadata()); + seeder.execute(tuple); + } await().atMost(20, TimeUnit.SECONDS) .until( () -> - out.getAckedTuples().stream() - .anyMatch(t -> url.equals(t.getStringByField("url")))); + urls.stream() + .allMatch( + url -> + out.getAckedTuples().stream() + .anyMatch( + t -> + url.equals( + t + .getStringByField( + "url"))))); seeder.cleanup(); } @@ -224,4 +242,56 @@ void blocksHostQueueOnHeaderless429() { bolt.cleanup(); } + + @Test + @Timeout(value = 2, unit = TimeUnit.MINUTES) + void pacesHostQueueOnRobotsCrawlDelay() { + seedUrls("/1", "/2"); + + QueueRegulatorBolt bolt = new QueueRegulatorBolt(); + Map conf = frontierConfig(); + conf.put(ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, "protocol."); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), mock(OutputCollector.class)); + Tuple t = mock(Tuple.class); + when(t.getStringByField("key")).thenReturn(HOST); + Metadata md = new Metadata(); + md.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, "300"); + when(t.getValueByField("metadata")).thenReturn(md); + bolt.execute(t); + + // blind delay for the fire-and-forget RPC to land, as in the sibling tests + await().pollDelay(Duration.ofSeconds(2)).atMost(3, TimeUnit.SECONDS).until(() -> true); + + // the queue delay paces successive hand-outs, it does not shrink a + // batch: the first getURLs still serves the queue (never produced + // before), so take a single URL out of it + assertEquals(1, countServed(1), "the first hand-out should serve one URL"); + + // the queue is now not requestable again before 300s: a later call + // must come back empty although /2 is neither in-flight nor served. + // The pause also pins the unit of setDelayRequestable: were it + // milliseconds, the 300ms window would have expired and /2 would be + // handed out + await().pollDelay(Duration.ofSeconds(2)).atMost(3, TimeUnit.SECONDS).until(() -> true); + assertEquals(0, countServed(10), "the queue delay should pace the second URL"); + + bolt.cleanup(); + } + + /** Drains one getURLs call and returns how many URLs it handed out. */ + private int countServed(int maxUrlsPerQueue) { + GetParams params = + GetParams.newBuilder() + .setCrawlID(CrawlID.DEFAULT) + .setMaxQueues(10) + .setMaxUrlsPerQueue(maxUrlsPerQueue) + .build(); + int served = 0; + Iterator it = blocking.getURLs(params); + while (it.hasNext()) { + it.next(); + served++; + } + return served; + } } From a827ca965a8b9c9671a76943e1172e08ae44c738 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 15 Jul 2026 13:55:27 +0200 Subject: [PATCH 6/8] fix: preserve robots crawl-delay metadata semantics Round fractional delays up so frontier pacing never becomes more aggressive than robots.txt. Rebuild the control signal from the locally parsed rules after protocol metadata is merged, preventing stale metadata or a colliding response header from changing it. --- .../apache/stormcrawler/bolt/FetcherBolt.java | 16 +++++- .../stormcrawler/bolt/SimpleFetcherBolt.java | 17 ++++-- .../stormcrawler/bolt/FetcherBoltTest.java | 36 +++++++++++- .../bolt/SimpleFetcherBoltTest.java | 56 ++++++++++++++++++- 4 files changed, 115 insertions(+), 10 deletions(-) diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java index 2b0b263f9..ac141a4ad 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/FetcherBolt.java +++ b/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; @@ -679,9 +682,10 @@ public void run() { 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, - Long.toString(rules.getCrawlDelay() / 1000L)); + Constants.ROBOTS_CRAWL_DELAY_KEY, robotsCrawlDelaySecs); } else { // pass the info about crawl delay metadata.setValue(Constants.STATUS_ERROR_CAUSE, "crawl_delay"); @@ -797,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())); diff --git a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java b/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java index 2c47dee12..6c2dae6e2 100644 --- a/core/src/main/java/org/apache/stormcrawler/bolt/SimpleFetcherBolt.java +++ b/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,7 +404,7 @@ 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 @@ -409,9 +412,8 @@ public void execute(Tuple input) { delay = maxCrawlDelay; // report the delay the fetcher is not holding, so a frontier-side // consumer can enforce it at the source (#867) - metadata.setValue( - Constants.ROBOTS_CRAWL_DELAY_KEY, - Long.toString(robotsDelay / 1000L)); + 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 ({}): {}", @@ -504,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)); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java index 3f872c145..e7f592e14 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/FetcherBoltTest.java @@ -51,7 +51,13 @@ void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) aResponse() .withStatus(200) .withBody("User-agent: *\nCrawl-delay: 120\n"))); - stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "1") + .withBody("hello"))); Map config = new HashMap<>(); config.put("http.agent.name", "this_is_only_a_test"); @@ -62,6 +68,26 @@ void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) 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 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 { @@ -71,7 +97,13 @@ void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo) aResponse() .withStatus(200) .withBody("User-agent: *\nCrawl-delay: 5\n"))); - stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "120") + .withBody("hello"))); Map config = new HashMap<>(); config.put("http.agent.name", "this_is_only_a_test"); diff --git a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java index 91565228e..085ef8056 100644 --- a/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java +++ b/core/src/test/java/org/apache/stormcrawler/bolt/SimpleFetcherBoltTest.java @@ -49,7 +49,13 @@ void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) aResponse() .withStatus(200) .withBody("User-agent: *\nCrawl-delay: 120\n"))); - stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "1") + .withBody("hello"))); Map config = new HashMap<>(); config.put("http.agent.name", "this_is_only_a_test"); @@ -60,6 +66,26 @@ void forcedLongCrawlDelayIsReportedInMetadata(WireMockRuntimeInfo wmRuntimeInfo) 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 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 { @@ -69,7 +95,13 @@ void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo) aResponse() .withStatus(200) .withBody("User-agent: *\nCrawl-delay: 5\n"))); - stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello"))); + stubFor( + get(urlEqualTo("/page")) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Robots.Crawl.Delay", "120") + .withBody("hello"))); Map config = new HashMap<>(); config.put("http.agent.name", "this_is_only_a_test"); @@ -79,4 +111,24 @@ void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo) Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); } + + @Test + void negativeMaximumDisablesTheCrawlDelayLimit(WireMockRuntimeInfo wmRuntimeInfo) + throws ReflectiveOperationException { + 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 config = new HashMap<>(); + config.put("http.agent.name", "this_is_only_a_test"); + config.put("fetcher.max.crawl.delay", -1); + config.put("fetcher.max.crawl.delay.force", true); + + Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page"); + assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY)); + } } From 2de0aa26b3cfd630fdaafb81a95f5d05521658f3 Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 15 Jul 2026 13:55:52 +0200 Subject: [PATCH 7/8] feat: enforce safe robots crawl-delay pacing Keep frontier pacing opt-in and fail fast unless queues are partitioned by host, hand out one URL at a time, and carry robots.crawl.delay as persist-only metadata. Serialize setDelay calls per queue so asynchronous completions cannot violate the last-value-wins contract. --- .../stormcrawler/urlfrontier/Constants.java | 7 + .../urlfrontier/CrawlDelayPolicy.java | 127 ++++++++--- .../urlfrontier/QueueRegulatorBolt.java | 209 ++++++++++++++---- .../stormcrawler/urlfrontier/Spout.java | 7 +- .../urlfrontier/CrawlDelayPolicyTest.java | 110 +++++++-- .../QueueRegulatorBoltDecisionTest.java | 128 ++++++++++- .../urlfrontier/QueueRegulatorBoltTest.java | 27 ++- 7 files changed, 509 insertions(+), 106 deletions(-) diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java index 8fd7f4deb..976cb7399 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Constants.java @@ -30,6 +30,7 @@ private Constants() {} // Spout public static final String URLFRONTIER_MAX_URLS_PER_BUCKET_KEY = "urlfrontier.max.urls.per.bucket"; + public static final int URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT = 10; public static final String URLFRONTIER_MAX_BUCKETS_KEY = "urlfrontier.max.buckets"; public static final String URLFRONTIER_DELAY_REQUESTABLE_KEY = "urlfrontier.delay.requestable"; @@ -97,6 +98,12 @@ private Constants() {} public static final boolean URLFRONTIER_BACKOFF_ON_EXCEPTIONS_DEFAULT = false; + /** Enables forwarding long robots.txt Crawl-delays to URLFrontier. Defaults to false. */ + public static final String URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY = + "urlfrontier.robots.crawl.delay.enabled"; + + public static final boolean URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT = false; + /** * Random extra fraction added to every computed block duration, so that hosts blocked on the * same schedule do not all expire together. Defaults to 0.1; 0 disables jitter. diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java index c7c1ea41d..b33ba93d2 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicy.java @@ -21,7 +21,12 @@ import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.Ticker; import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.apache.stormcrawler.Metadata; import org.apache.stormcrawler.util.ConfUtils; @@ -29,35 +34,56 @@ * Decides whether a queue-stream tuple carries a robots.txt crawl-delay worth forwarding to the * frontier via {@code setDelay}. The fetcher reports the delay (in seconds, {@code * robots.crawl.delay} metadata) only when it exceeds {@code fetcher.max.crawl.delay} and {@code - * fetcher.max.crawl.delay.force} lets it keep fetching; the frontier then paces the queue at the - * requested rate, restoring the compliance that force alone would violate. + * fetcher.max.crawl.delay.force} lets it keep fetching. When frontier pacing is explicitly enabled, + * later hand-outs use the requested interval capped by {@code urlfrontier.backoff.max.secs}. * *

The signal is declarative — the last value wins, there is no escalation and no interaction * with {@link HostBackoff}'s blocks. Values are capped by {@code urlfrontier.backoff.max.secs} * (robots can request absurd delays; the parse-time cap is disabled in StormCrawler) and - * deduplicated per host: an unchanged value is re-sent at most once per {@code - * urlfrontier.backoff.decay.secs} window, which also makes a lost fire-and-forget call converge. + * deduplicated per host after a successful RPC. Only one request per host may be active; changes + * received meanwhile are coalesced to the latest value, preventing older unary RPCs from + * overwriting newer values out of order. * - *

Not thread-safe; a bolt task calls {@link #delaySecsFor} from a single thread. + *

The decision runs on the bolt thread while gRPC callbacks may complete requests concurrently. + * Per-key {@code ConcurrentMap.compute} operations serialize those transitions. */ final class CrawlDelayPolicy { - /** Last delay sent per host; expiry on write bounds both memory and RPC-loss staleness. */ - private final Cache lastSent; + /** Last delay applied successfully per host; expiry bounds the dedupe window and memory. */ + private final Cache lastApplied; + + /** At most one unary RPC per host, plus the latest value observed while it is active. */ + private final ConcurrentMap active = new ConcurrentHashMap<>(); + + private final AtomicLong requestTokens = new AtomicLong(); + + private final boolean enabled; private final long maxSecs; + record DelayRequest(long token, String key, int delaySecs) {} + + private record Flight(DelayRequest request, Integer pendingLatest) {} + CrawlDelayPolicy(Map conf) { this(conf, Ticker.systemTicker()); } /** Visible for tests: a controllable ticker drives the dedupe window. */ CrawlDelayPolicy(Map conf, Ticker ticker) { + this.enabled = + ConfUtils.getBoolean( + conf, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT) + && ConfUtils.getBoolean(conf, "fetcher.max.crawl.delay.force", false); this.maxSecs = - ConfUtils.getLong( - conf, - Constants.URLFRONTIER_BACKOFF_MAX_KEY, - Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT); + Math.max( + 1L, + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT)); long decaySecs = Math.max( 0L, @@ -65,41 +91,86 @@ final class CrawlDelayPolicy { conf, Constants.URLFRONTIER_BACKOFF_DECAY_KEY, Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT)); - this.lastSent = + this.lastApplied = Caffeine.newBuilder() .expireAfterWrite(decaySecs, TimeUnit.SECONDS) .ticker(ticker) .build(); } - /** - * @return the delay to forward to the frontier in seconds, or {@code -1} when the tuple carries - * none, the value is unchanged, or the shared {@code _DEFAULT_} queue is targeted - */ - int delaySecsFor(String key, Metadata metadata) { - if (metadata == null || HostBackoff.DEFAULT_QUEUE_KEY.equals(key)) { - return -1; + /** Returns a request to start now, or empty when it is invalid, deduplicated or coalesced. */ + Optional requestFor(String key, Metadata metadata) { + if (!enabled + || key == null + || metadata == null + || HostBackoff.DEFAULT_QUEUE_KEY.equals(key)) { + return Optional.empty(); } final String value = metadata.getFirstValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY); if (value == null) { - return -1; + return Optional.empty(); } long secs; try { secs = Long.parseLong(value); } catch (NumberFormatException e) { - return -1; + return Optional.empty(); } if (secs <= 0) { - return -1; + return Optional.empty(); } final int delay = (int) Math.min(secs, Math.min(maxSecs, Integer.MAX_VALUE)); - final Integer previous = lastSent.getIfPresent(key); - if (previous != null && previous == delay) { - return -1; - } - lastSent.put(key, delay); - return delay; + AtomicReference next = new AtomicReference<>(); + active.compute( + key, + (ignored, flight) -> { + if (flight != null) { + return new Flight(flight.request(), delay); + } + Integer previous = lastApplied.getIfPresent(key); + if (previous != null && previous == delay) { + return null; + } + DelayRequest request = + new DelayRequest(requestTokens.incrementAndGet(), key, delay); + next.set(request); + return new Flight(request, null); + }); + return Optional.ofNullable(next.get()); + } + + /** Completes one RPC and returns the latest coalesced value, if another RPC must follow. */ + Optional completed(DelayRequest completed, boolean success) { + AtomicReference next = new AtomicReference<>(); + active.computeIfPresent( + completed.key(), + (key, flight) -> { + if (flight.request().token() != completed.token()) { + return flight; + } + if (success) { + lastApplied.put(key, completed.delaySecs()); + } else { + // An error is ambiguous: the server may have applied the request before the + // response was lost. Forget the previous value so a later signal reasserts + // it instead of being incorrectly deduplicated. + lastApplied.invalidate(key); + } + Integer pending = flight.pendingLatest(); + if (pending == null || (success && pending == completed.delaySecs())) { + return null; + } + DelayRequest request = + new DelayRequest(requestTokens.incrementAndGet(), key, pending); + next.set(request); + return new Flight(request, null); + }); + return Optional.ofNullable(next.get()); + } + + /** Drops in-flight bookkeeping during bolt cleanup; terminal callbacks then become no-ops. */ + void clear() { + active.clear(); } } diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java index 139ca02d9..6e9d1418e 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBolt.java @@ -44,8 +44,9 @@ /** * Regulates a URLFrontier queue from the {@code queue} stream emitted by the status updater: blocks - * it via {@code blockQueueUntil} whenever the tuple's metadata reports a rate-limit response. See - * issues #867, #784 and #1106. + * it via {@code blockQueueUntil} when metadata reports a rate-limit response, and paces it via + * {@code setDelay} when the fetcher reports a long robots.txt Crawl-delay. See issues #867, #784, + * #1106 and #1979. * *

Two cases are handled, both gated on the status codes configured with {@code * urlfrontier.backoff.status.codes} (default 429 and 503): @@ -69,13 +70,12 @@ * block so that hosts blocked on the same schedule do not all retry at once, and a block is only * ever extended, never shrunk by a later, shorter signal. * - *

Independently of rate limits, when the tuple's metadata carries {@code robots.crawl.delay} - * (written by the fetcher bolts when a robots.txt {@code Crawl-delay} exceeds {@code - * fetcher.max.crawl.delay} and {@code fetcher.max.crawl.delay.force} is true), the bolt forwards it - * to the frontier via {@code setDelay}, capped by {@code urlfrontier.backoff.max.secs}: the queue - * is then served at the pace robots requested, restoring the compliance that the force setting - * alone would violate. The value is declarative — the last one seen wins, re-sent at most once per - * {@code urlfrontier.backoff.decay.secs} window. If a site later removes its Crawl-delay, no signal + *

Independently of rate limits, when robots pacing is explicitly enabled and the tuple's + * metadata carries {@code robots.crawl.delay} (written by the fetcher bolts when a robots.txt + * {@code Crawl-delay} exceeds {@code fetcher.max.crawl.delay}), the bolt forwards it to the + * frontier via {@code setDelay}, capped by {@code urlfrontier.backoff.max.secs}. The value is + * declarative: the last one seen wins, re-sent at most once per {@code + * urlfrontier.backoff.decay.secs} window. If a site later removes its Crawl-delay, no signal * arrives and the frontier keeps the old pace: throughput loss on that host, never impoliteness; * {@code setDelay(key, 0)} is deliberately never called as it would erase a server-side default for * the queue. @@ -93,25 +93,36 @@ * metadata.persist: * - fetch.statusCode * - protocol.retry-after - * - robots.crawl.delay * * * (the second entry depends on {@code protocol.md.prefix}), and {@code fetch.exception} as well * when the back-off on exceptions is enabled. A warning is logged at startup when the configuration * would strip them. * - *

Blocking stops new hand-outs; it does not recall URLs of the host already prefetched by the - * spout into the topology, which still fail against the rate-limited server and reschedule via the - * status stream. Keep {@code urlfrontier.max.urls.per.bucket} and {@code - * topology.max.spout.pending} small for a block to bite quickly. + *

Robots pacing is enabled by {@code urlfrontier.robots.crawl.delay.enabled=true} together with + * {@code fetcher.max.crawl.delay.force=true}. It is safe only with {@code + * partition.url.mode=byHost}, {@code urlfrontier.max.urls.per.bucket=1}, and {@code + * robots.crawl.delay} in {@code metadata.persist} but not in {@code metadata.transfer} + * (including wildcards such as {@code robots.*}). The bolt probes these conditions at startup and + * fails fast rather than silently pacing a shared queue, transferring a host's delay to an outlink, + * or handing out several URLs from that host in one request. A custom {@code + * metadata.transfer.class} remains responsible for satisfying the same persist-only contract for + * all URLs and values. + * + *

Neither blocking nor pacing recalls URLs already prefetched by the spout. For rate-limit + * blocks, keep {@code urlfrontier.max.urls.per.bucket} and {@code topology.max.spout.pending} small + * for the block to bite quickly. Robots pacing requires the stricter per-queue batch size of one; + * this bounds each frontier hand-out but cannot recall earlier hand-outs or eliminate concurrent + * requests during the transition to the new delay. * *

Connects via {@code urlfrontier.address}, falling back to {@code urlfrontier.host} / {@code * urlfrontier.port}. The block is issued with {@code local=false} and propagates to the whole * cluster, so with several frontier nodes a single connection to any one of them is enough. * - *

The block is fire-and-forget with a short deadline: a failed or expired call is logged but the - * tuple is acked anyway. A missed block heals itself: every further rate-limit signal for the host - * re-asserts the stored block until the frontier stops serving it. + *

The RPCs have a short deadline: a failed or expired call is logged but the tuple is acked + * anyway. A missed block heals itself on further rate-limit signals. Robots delay calls are + * serialized per host and coalesce to the latest value; an unchanged delay is re-sent after the + * configured decay window. */ public class QueueRegulatorBolt extends BaseRichBolt { @@ -137,20 +148,6 @@ public void onError(Throwable t) { public void onCompleted() {} }; - private static final StreamObserver SET_DELAY_NOOP_OBSERVER = - new StreamObserver<>() { - @Override - public void onNext(Empty value) {} - - @Override - public void onError(Throwable t) { - LOG.warn("setDelay failed", t); - } - - @Override - public void onCompleted() {} - }; - private OutputCollector collector; private ManagedChannel channel; private URLFrontierStub frontier; @@ -162,12 +159,17 @@ public void onCompleted() {} /** Per-host robots crawl-delay forwarding decision. */ private CrawlDelayPolicy crawlDelayPolicy; + /** Prevents terminal callbacks from chaining another RPC after cleanup starts. */ + private volatile boolean closed; + @Override public void prepare(Map conf, TopologyContext context, OutputCollector c) { + this.closed = false; this.collector = c; this.globalCrawlID = ConfUtils.getString(conf, Constants.URLFRONTIER_CRAWL_ID_KEY, CrawlID.DEFAULT); this.backoff = new HostBackoff(conf); + validateRobotsPacingConfiguration(conf); this.crawlDelayPolicy = new CrawlDelayPolicy(conf); // the status updater emits the queue stream after the metadata.persist // filter: warn upfront when the keys this bolt relies on would be @@ -222,20 +224,51 @@ public void execute(Tuple t) { frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) .blockQueueUntil(params, NOOP_OBSERVER); } - final int delaySecs = crawlDelayPolicy.delaySecsFor(key, metadata); - if (delaySecs > 0) { - LOG.debug("Setting delay of {}s on queue {}", delaySecs, key); - QueueDelayParams delayParams = - QueueDelayParams.newBuilder() - .setKey(key) - .setCrawlID(globalCrawlID) - .setDelayRequestable(delaySecs) - .setLocal(false) - .build(); + crawlDelayPolicy.requestFor(key, metadata).ifPresent(this::setDelay); + collector.ack(t); + } + + private void setDelay(CrawlDelayPolicy.DelayRequest request) { + if (closed) { + return; + } + LOG.debug("Setting delay of {}s on queue {}", request.delaySecs(), request.key()); + QueueDelayParams delayParams = + QueueDelayParams.newBuilder() + .setKey(request.key()) + .setCrawlID(globalCrawlID) + .setDelayRequestable(request.delaySecs()) + .setLocal(false) + .build(); + try { frontier.withDeadlineAfter(BLOCK_RPC_DEADLINE_SECS, TimeUnit.SECONDS) - .setDelay(delayParams, SET_DELAY_NOOP_OBSERVER); + .setDelay(delayParams, setDelayObserver(request)); + } catch (RuntimeException e) { + LOG.warn("setDelay failed", e); + crawlDelayPolicy.completed(request, false).ifPresent(this::setDelay); } - collector.ack(t); + } + + private StreamObserver setDelayObserver(CrawlDelayPolicy.DelayRequest request) { + return new StreamObserver<>() { + @Override + public void onNext(Empty value) {} + + @Override + public void onError(Throwable t) { + LOG.warn("setDelay failed", t); + crawlDelayPolicy + .completed(request, false) + .ifPresent(QueueRegulatorBolt.this::setDelay); + } + + @Override + public void onCompleted() { + crawlDelayPolicy + .completed(request, true) + .ifPresent(QueueRegulatorBolt.this::setDelay); + } + }; } /** @@ -249,7 +282,6 @@ static Set missingQueueStreamKeys( probe.setValue(HostBackoff.STATUS_CODE_KEY, "429"); probe.setValue(retryAfterKey, "1"); probe.setValue(HostBackoff.EXCEPTION_KEY, "probe"); - probe.setValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY, "1"); Metadata filtered = MetadataTransfer.getInstance(conf).filter(probe); Set missing = new LinkedHashSet<>(); if (filtered.getFirstValue(HostBackoff.STATUS_CODE_KEY) == null) { @@ -261,13 +293,90 @@ static Set missingQueueStreamKeys( if (onExceptions && filtered.getFirstValue(HostBackoff.EXCEPTION_KEY) == null) { missing.add(HostBackoff.EXCEPTION_KEY); } - if (filtered.getFirstValue(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY) - == null) { - missing.add(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY); - } return missing; } + /** + * Fails fast when robots pacing is enabled but the queue key, hand-out size, delay cap or + * metadata routing could apply an unsafe pace or target the wrong host. + */ + static void validateRobotsPacingConfiguration(Map conf) { + if (!ConfUtils.getBoolean( + conf, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, + Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_DEFAULT)) { + return; + } + + if (!ConfUtils.getBoolean(conf, "fetcher.max.crawl.delay.force", false)) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires fetcher.max.crawl.delay.force=true"); + } + + int maxCrawlDelay = ConfUtils.getInt(conf, "fetcher.max.crawl.delay", 30); + if (maxCrawlDelay < 0) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires fetcher.max.crawl.delay >= 0"); + } + + String partitionMode = + ConfUtils.getString( + conf, + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + if (!org.apache.stormcrawler.Constants.PARTITION_MODE_HOST.equals(partitionMode)) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires partition.url.mode=" + + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST + + "; found " + + partitionMode); + } + + int maxURLsPerBucket = + ConfUtils.getInt( + conf, + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT); + if (maxURLsPerBucket != 1) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires " + + Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY + + "=1; found " + + maxURLsPerBucket); + } + + long maxDelaySecs = + ConfUtils.getLong( + conf, + Constants.URLFRONTIER_BACKOFF_MAX_KEY, + Constants.URLFRONTIER_BACKOFF_MAX_DEFAULT); + if (maxDelaySecs <= 0) { + throw new IllegalArgumentException( + "robots crawl-delay pacing requires " + + Constants.URLFRONTIER_BACKOFF_MAX_KEY + + " > 0"); + } + + final String robotsDelayKey = org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY; + MetadataTransfer transfer = MetadataTransfer.getInstance(conf); + Metadata probe = new Metadata(); + probe.setValue(robotsDelayKey, "1"); + Metadata persisted = transfer.filter(probe); + if (persisted == null || persisted.getFirstValue(robotsDelayKey) == null) { + throw new IllegalArgumentException( + "robots.crawl.delay must be included in metadata.persist"); + } + + Metadata outlink = + transfer.getMetaForOutlink( + "http://target.example/", "http://source.example/", probe); + if (outlink == null || outlink.getFirstValue(robotsDelayKey) != null) { + throw new IllegalArgumentException( + "robots.crawl.delay must not appear in metadata.transfer, including via a" + + " wildcard: it is a per-host control signal"); + } + } + @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { // terminal bolt @@ -275,6 +384,10 @@ public void declareOutputFields(OutputFieldsDeclarer declarer) { @Override public void cleanup() { + closed = true; + if (crawlDelayPolicy != null) { + crawlDelayPolicy.clear(); + } if (channel != null) { channel.shutdown(); } diff --git a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java index 6aa9ef7f8..9763051ec 100644 --- a/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java +++ b/external/urlfrontier/src/main/java/org/apache/stormcrawler/urlfrontier/Spout.java @@ -24,6 +24,7 @@ import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_DELAY_REQUESTABLE_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_HOST_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_BUCKETS_KEY; +import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY; import static org.apache.stormcrawler.urlfrontier.Constants.URLFRONTIER_PORT_KEY; @@ -70,7 +71,11 @@ public void open( SpoutOutputCollector collector) { super.open(stormConf, context, collector); - maxURLsPerBucket = ConfUtils.getInt(stormConf, URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 10); + maxURLsPerBucket = + ConfUtils.getInt( + stormConf, + URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, + URLFRONTIER_MAX_URLS_PER_BUCKET_DEFAULT); maxBucketNum = ConfUtils.getInt(stormConf, URLFRONTIER_MAX_BUCKETS_KEY, 10); diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java index aed6abf5b..10436c648 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/CrawlDelayPolicyTest.java @@ -18,6 +18,8 @@ package org.apache.stormcrawler.urlfrontier; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; import com.github.benmanes.caffeine.cache.Ticker; import java.util.HashMap; @@ -51,6 +53,8 @@ void advanceSecs(long secs) { private CrawlDelayPolicy policy(Object... keysAndValues) { Map conf = new HashMap<>(); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); for (int i = 0; i < keysAndValues.length; i += 2) { conf.put((String) keysAndValues[i], keysAndValues[i + 1]); } @@ -67,45 +71,123 @@ private static Metadata md(String delay) { @Test void newValueIsForwarded() { - assertEquals(120, policy().delaySecsFor(KEY, md("120"))); + assertEquals(120, request(policy(), "120").delaySecs()); } @Test void unchangedValueIsDeduplicatedWithinTheWindow() { CrawlDelayPolicy policy = policy(); - assertEquals(120, policy.delaySecsFor(KEY, md("120"))); - assertEquals(-1, policy.delaySecsFor(KEY, md("120"))); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); } @Test - void changedValueIsForwarded() { + void changedValuesAreSerializedAndOnlyTheLatestPendingValueIsSent() { CrawlDelayPolicy policy = policy(); - assertEquals(120, policy.delaySecsFor(KEY, md("120"))); - assertEquals(60, policy.delaySecsFor(KEY, md("60"))); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + assertFalse(policy.requestFor(KEY, md("180")).isPresent()); + + CrawlDelayPolicy.DelayRequest second = policy.completed(first, true).orElseThrow(); + assertEquals(180, second.delaySecs()); + assertFalse(policy.completed(second, true).isPresent()); } @Test void unchangedValueIsResentAfterTheWindow() { // the dedupe entry expires so a lost RPC converges within one window CrawlDelayPolicy policy = policy(); - assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.completed(first, true).isPresent()); ticker.advanceSecs(Constants.URLFRONTIER_BACKOFF_DECAY_DEFAULT + 1); - assertEquals(120, policy.delaySecsFor(KEY, md("120"))); + assertEquals(120, request(policy, "120").delaySecs()); } @Test void valueIsCappedAtBackoffMax() { CrawlDelayPolicy policy = policy(Constants.URLFRONTIER_BACKOFF_MAX_KEY, 3600); - assertEquals(3600, policy.delaySecsFor(KEY, md("86400"))); + assertEquals(3600, request(policy, "86400").delaySecs()); + } + + @Test + void signalIsIgnoredWhenFetcherForceIsDisabled() { + CrawlDelayPolicy policy = policy("fetcher.max.crawl.delay.force", false); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void signalIsIgnoredWhenRobotsPacingIsDisabled() { + CrawlDelayPolicy policy = + policy(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, false); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + } + + @Test + void failedRpcRetriesTheLatestPendingValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + + CrawlDelayPolicy.DelayRequest second = policy.completed(first, false).orElseThrow(); + assertEquals(60, second.delaySecs()); + assertFalse(policy.completed(second, false).isPresent()); + assertEquals(60, request(policy, "60").delaySecs()); + } + + @Test + void failedChangedRpcInvalidatesThePreviouslyAppliedValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest initial = request(policy, "120"); + assertFalse(policy.completed(initial, true).isPresent()); + + CrawlDelayPolicy.DelayRequest changed = request(policy, "60"); + assertFalse(policy.completed(changed, false).isPresent()); + + // The server may have applied 60 before the response was lost. Do not deduplicate the + // previous 120: reasserting it is the only way to converge safely. + assertEquals(120, request(policy, "120").delaySecs()); + } + + @Test + void repeatedInFlightValueRetriesOnFailureButNotOnSuccess() { + CrawlDelayPolicy failed = policy(); + CrawlDelayPolicy.DelayRequest failedRequest = request(failed, "120"); + assertFalse(failed.requestFor(KEY, md("120")).isPresent()); + assertEquals(120, failed.completed(failedRequest, false).orElseThrow().delaySecs()); + + CrawlDelayPolicy succeeded = policy(); + CrawlDelayPolicy.DelayRequest succeededRequest = request(succeeded, "120"); + assertFalse(succeeded.requestFor(KEY, md("120")).isPresent()); + assertFalse(succeeded.completed(succeededRequest, true).isPresent()); + } + + @Test + void staleCallbackCannotCompleteANewerRequestWithTheSameValue() { + CrawlDelayPolicy policy = policy(); + CrawlDelayPolicy.DelayRequest first = request(policy, "120"); + assertFalse(policy.requestFor(KEY, md("120")).isPresent()); + CrawlDelayPolicy.DelayRequest retry = policy.completed(first, false).orElseThrow(); + + assertFalse(policy.completed(first, true).isPresent()); + assertFalse(policy.requestFor(KEY, md("60")).isPresent()); + assertEquals(60, policy.completed(retry, true).orElseThrow().delaySecs()); } @Test void malformedAbsentZeroAndDefaultQueueAreIgnored() { CrawlDelayPolicy policy = policy(); - assertEquals(-1, policy.delaySecsFor(KEY, md("not-a-number"))); - assertEquals(-1, policy.delaySecsFor(KEY, md("0"))); - assertEquals(-1, policy.delaySecsFor(KEY, md(null))); - assertEquals(-1, policy.delaySecsFor(KEY, null)); - assertEquals(-1, policy.delaySecsFor("_DEFAULT_", md("120"))); + assertFalse(policy.requestFor(KEY, md("not-a-number")).isPresent()); + assertFalse(policy.requestFor(KEY, md("0")).isPresent()); + assertFalse(policy.requestFor(KEY, md(null)).isPresent()); + assertFalse(policy.requestFor(KEY, null).isPresent()); + assertFalse(policy.requestFor("_DEFAULT_", md("120")).isPresent()); + } + + private static CrawlDelayPolicy.DelayRequest request(CrawlDelayPolicy policy, String delay) { + var request = policy.requestFor(KEY, md(delay)); + assertTrue(request.isPresent()); + return request.orElseThrow(); } } diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java index 7ca59b1dd..760bd5e66 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltDecisionTest.java @@ -17,7 +17,9 @@ package org.apache.stormcrawler.urlfrontier; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.HashMap; @@ -41,15 +43,13 @@ void missingQueueStreamKeysReportsAllWithDefaultConfig() { // with the default metadata.persist none of the keys survives the filter Set missing = QueueRegulatorBolt.missingQueueStreamKeys(new HashMap<>(), RETRY_AFTER_KEY, false); - assertEquals(Set.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay"), missing); + assertEquals(Set.of("fetch.statusCode", RETRY_AFTER_KEY), missing); } @Test void missingQueueStreamKeysEmptyWhenAllPersisted() { Map conf = new HashMap<>(); - conf.put( - "metadata.persist", - List.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay")); + conf.put("metadata.persist", List.of("fetch.statusCode", RETRY_AFTER_KEY)); assertTrue( QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, false).isEmpty()); } @@ -57,7 +57,7 @@ void missingQueueStreamKeysEmptyWhenAllPersisted() { @Test void missingQueueStreamKeysHonoursWildcards() { Map conf = new HashMap<>(); - conf.put("metadata.persist", List.of("fetch.*", "protocol.*", "robots.*")); + conf.put("metadata.persist", List.of("fetch.*", "protocol.*")); assertTrue( QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, false).isEmpty()); } @@ -65,9 +65,7 @@ void missingQueueStreamKeysHonoursWildcards() { @Test void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() { Map conf = new HashMap<>(); - conf.put( - "metadata.persist", - List.of("fetch.statusCode", RETRY_AFTER_KEY, "robots.crawl.delay")); + conf.put("metadata.persist", List.of("fetch.statusCode", RETRY_AFTER_KEY)); // the same configuration is complete without the exceptions gate and // incomplete with it assertTrue( @@ -76,4 +74,118 @@ void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() { Set.of("fetch.exception"), QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, true)); } + + @Test + void robotsValidationIsSkippedWhenFrontierPacingIsDisabled() { + assertDoesNotThrow( + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(new HashMap<>())); + } + + @Test + void legacyFetcherForceDoesNotEnableFrontierPacing() { + Map conf = new HashMap<>(); + conf.put("fetcher.max.crawl.delay.force", true); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void enabledRobotsPacingRequiresFetcherForce() { + Map conf = validRobotsConf(); + conf.put("fetcher.max.crawl.delay.force", false); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void validPersistOnlyRobotsConfigurationPasses() { + assertDoesNotThrow( + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(validRobotsConf())); + } + + @Test + void robotsSignalMustSurvivePersistence() { + Map conf = validRobotsConf(); + conf.remove("metadata.persist"); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsSignalMustNotTransferToOutlinks() { + Map conf = validRobotsConf(); + conf.put("metadata.transfer", List.of("robots.*")); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPersistenceWildcardIsAccepted() { + Map conf = validRobotsConf(); + conf.put("metadata.persist", List.of("robots.*")); + assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPacingRequiresHostPartitioning() { + for (String mode : + List.of( + org.apache.stormcrawler.Constants.PARTITION_MODE_DOMAIN, + org.apache.stormcrawler.Constants.PARTITION_MODE_IP)) { + Map conf = validRobotsConf(); + conf.put(org.apache.stormcrawler.Constants.PARTITION_MODEParamName, mode); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + } + + @Test + void robotsPacingRequiresSingleUrlHandouts() { + Map defaultBatch = validRobotsConf(); + defaultBatch.remove(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(defaultBatch)); + + Map multiUrlBatch = validRobotsConf(); + multiUrlBatch.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 2); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(multiUrlBatch)); + } + + @Test + void robotsPacingRequiresAnEnabledCrawlDelayLimit() { + Map conf = validRobotsConf(); + conf.put("fetcher.max.crawl.delay", -1); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + @Test + void robotsPacingRequiresAPositiveFrontierDelayCap() { + Map conf = validRobotsConf(); + conf.put(Constants.URLFRONTIER_BACKOFF_MAX_KEY, 0); + assertThrows( + IllegalArgumentException.class, + () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf)); + } + + private static Map validRobotsConf() { + Map conf = new HashMap<>(); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put( + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + conf.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 1); + conf.put( + "metadata.persist", + List.of(org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY)); + return conf; + } } diff --git a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java index ec71bcf6a..041834422 100644 --- a/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java +++ b/external/urlfrontier/src/test/java/org/apache/stormcrawler/urlfrontier/QueueRegulatorBoltTest.java @@ -34,6 +34,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -248,9 +249,26 @@ void blocksHostQueueOnHeaderless429() { void pacesHostQueueOnRobotsCrawlDelay() { seedUrls("/1", "/2"); + // Model one hand-out before the response carrying the robots delay reaches the queue + // stream. Production uses max.urls.per.bucket=1 to prevent a single request from draining + // a larger host batch; setDelay cannot recall earlier or concurrent hand-outs. + assertEquals(1, countServed(1), "the first hand-out should serve one URL"); + QueueRegulatorBolt bolt = new QueueRegulatorBolt(); Map conf = frontierConfig(); conf.put(ProtocolResponse.PROTOCOL_MD_PREFIX_PARAM, "protocol."); + conf.put(Constants.URLFRONTIER_ROBOTS_CRAWL_DELAY_ENABLED_KEY, true); + conf.put("fetcher.max.crawl.delay.force", true); + conf.put( + org.apache.stormcrawler.Constants.PARTITION_MODEParamName, + org.apache.stormcrawler.Constants.PARTITION_MODE_HOST); + conf.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 1); + conf.put( + "metadata.persist", + List.of( + "fetch.statusCode", + "protocol.retry-after", + org.apache.stormcrawler.Constants.ROBOTS_CRAWL_DELAY_KEY)); bolt.prepare(conf, TestUtil.getMockedTopologyContext(), mock(OutputCollector.class)); Tuple t = mock(Tuple.class); when(t.getStringByField("key")).thenReturn(HOST); @@ -262,13 +280,8 @@ void pacesHostQueueOnRobotsCrawlDelay() { // blind delay for the fire-and-forget RPC to land, as in the sibling tests await().pollDelay(Duration.ofSeconds(2)).atMost(3, TimeUnit.SECONDS).until(() -> true); - // the queue delay paces successive hand-outs, it does not shrink a - // batch: the first getURLs still serves the queue (never produced - // before), so take a single URL out of it - assertEquals(1, countServed(1), "the first hand-out should serve one URL"); - - // the queue is now not requestable again before 300s: a later call - // must come back empty although /2 is neither in-flight nor served. + // The queue is now not requestable again before 300s: a later call must come back empty + // although /2 was not part of the earlier hand-out and remains queued. // The pause also pins the unit of setDelayRequestable: were it // milliseconds, the 300ms window would have expired and /2 would be // handed out From 31598b63019f5ba5d7941dad1210e437c982fa2e Mon Sep 17 00:00:00 2001 From: Davide Polato Date: Wed, 15 Jul 2026 13:57:43 +0200 Subject: [PATCH 8/8] docs: document URLFrontier robots crawl-delay pacing Describe the opt-in wiring and the host partitioning, single-URL hand-out, persist-only metadata, and delay-cap requirements for #1979. --- core/src/main/resources/crawler-default.yaml | 7 +++- docs/src/main/asciidoc/configuration.adoc | 18 +++++++-- docs/src/main/asciidoc/extending.adoc | 2 +- external/urlfrontier/README.md | 42 ++++++++++++++++++++ 4 files changed, 63 insertions(+), 6 deletions(-) diff --git a/core/src/main/resources/crawler-default.yaml b/core/src/main/resources/crawler-default.yaml index 9961320bd..9b176084c 100644 --- a/core/src/main/resources/crawler-default.yaml +++ b/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: diff --git a/docs/src/main/asciidoc/configuration.adoc b/docs/src/main/asciidoc/configuration.adoc index fd9e52b61..f938b14fd 100644 --- a/docs/src/main/asciidoc/configuration.adoc +++ b/docs/src/main/asciidoc/configuration.adoc @@ -177,7 +177,7 @@ features you are using in your Apache StormCrawler topology. Some external modul | fetcher.max.crawl.delay | 30 | Maximum accepted Crawl-delay from robots.txt (seconds). If the crawl-delay exceeds this value the behavior depends on the value of fetcher.max.crawl.delay.force. | fetcher.max.crawl.delay.force | false | Configures the behavior of fetcher if the robots.txt crawl-delay exceeds fetcher.max.crawl.delay. If false: the tuple is emitted to the StatusStream -as an ERROR. If true: the queue delay is set to fetcher.max.crawl.delay. +as an ERROR. If true: the fetcher caps its local delay and emits the requested delay in whole seconds as `robots.crawl.delay`; an explicitly enabled URLFrontier `QueueRegulatorBolt` can forward a capped value to the host queue. | fetcher.max.queue.size | -1 | The maximum length of the queue used to store items to be fetched by the FetcherBolt. A setting of -1 sets the length to Integer.MAX_VALUE. | fetcher.max.throttle.sleep | -1 | The maximum amount of time to wait between fetches; if exceeded, the item is sent to the back of the queue. Used in SimpleFetcherBolt. -1 disables it. | fetcher.max.urls.in.queues | -1 | Limits the number of URLs that can be stored in a fetch queue. -1 disables the limit. @@ -472,6 +472,17 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/sql[sql m Integration with link:https://github.com/crawler-commons/url-frontier[URLFrontier], a language-agnostic API for URL management and scheduling. See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfrontier[urlfrontier module] for full details. +`QueueRegulatorBolt` consumes the `queue` stream from the URLFrontier status updater and applies +rate-limit blocks or robots.txt pacing. Robots pacing is enabled when both +`urlfrontier.robots.crawl.delay.enabled` and `fetcher.max.crawl.delay.force` are `true`. It requires `partition.url.mode: byHost`, +`urlfrontier.max.urls.per.bucket: 1`, and `robots.crawl.delay` in `metadata.persist`. The same key +must not match `metadata.transfer`, including wildcards such as `robots.*`, because it is a +per-host control signal. The bolt probes these requirements at startup. A batch size of one +bounds each frontier hand-out but does not recall URLs already emitted or eliminate concurrent +requests while the new delay is reaching the frontier. A custom `metadata.transfer.class` is +responsible for preserving the same persist-only contract for all URLs and values; the startup +probe exercises only representative metadata. + [cols="1,1,3", options="header"] |=== | key | default value | description @@ -480,7 +491,8 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfronti | urlfrontier.port | 7071 | URLFrontier service port. | urlfrontier.address | - | URLFrontier service address (can be used instead of host and port). | urlfrontier.max.buckets | 10 | Number of buckets to request from the frontier. -| urlfrontier.max.urls.per.bucket | 10 | Maximum URLs per bucket. +| urlfrontier.max.urls.per.bucket | 10 | Maximum URLs per bucket. Must be 1 when robots crawl-delay pacing is enabled so one request cannot hand out several URLs from the same host. +| urlfrontier.robots.crawl.delay.enabled | false | Enables forwarding long robots.txt Crawl-delays from the queue stream to URLFrontier. Also requires `fetcher.max.crawl.delay.force: true` and the validated configuration described above. | urlfrontier.delay.requestable | - | Time in seconds for which the URLs are locked in URLFrontier. | urlfrontier.cache.expireafter.sec | - | Time in seconds after which the cache expires. | urlfrontier.max.messages.in.flight | - | Maximum number of messages in flight for the StatusUpdater. @@ -490,7 +502,7 @@ See the link:https://github.com/apache/stormcrawler/tree/main/external/urlfronti | urlfrontier.max.retry.after | 86400 | Maximum delay in seconds honoured when a server requests a back-off via the Retry-After HTTP response header. -1 disables the cap. | urlfrontier.backoff.base.secs | 60 | First block duration in seconds applied when a host rate-limits without a usable Retry-After header. 0 disables the adaptive back-off. | urlfrontier.backoff.factor | 2.0 | Multiplicative increase applied to the block duration on each new pressure event for the same host. -| urlfrontier.backoff.max.secs | 86400 | Upper bound in seconds for the escalated block duration. +| urlfrontier.backoff.max.secs | 86400 | Upper bound in seconds for the escalated block duration and for a robots crawl-delay forwarded to the frontier. | urlfrontier.backoff.decay.secs | 1800 | Quiet window in seconds after which its escalation is forgotten and the next pressure event starts again from the base duration. | urlfrontier.backoff.on.exceptions | false | Whether fetch exceptions count as pressure events and escalate the host's block. | urlfrontier.backoff.jitter | 0.1 | Random extra fraction added to every computed block duration. diff --git a/docs/src/main/asciidoc/extending.adoc b/docs/src/main/asciidoc/extending.adoc index 7277f88b9..5d1d74794 100644 --- a/docs/src/main/asciidoc/extending.adoc +++ b/docs/src/main/asciidoc/extending.adoc @@ -273,7 +273,7 @@ When `robots.txt` specifies a `Crawl-delay`: * If the delay is *smaller* than `fetcher.server.delay` and `fetcher.server.delay.force` is `true`, the larger default delay is used instead * If the delay is *larger* than `fetcher.max.crawl.delay`: ** `fetcher.max.crawl.delay.force: false` (default) — the queue is skipped and the URL gets an ERROR status -** `fetcher.max.crawl.delay.force: true` — the delay is capped at `fetcher.max.crawl.delay` +** `fetcher.max.crawl.delay.force: true` — the fetcher caps its local delay at `fetcher.max.crawl.delay` and reports the requested delay in whole seconds as `robots.crawl.delay`; when `QueueRegulatorBolt` is wired and `urlfrontier.robots.crawl.delay.enabled: true`, the frontier forwards a value capped by `urlfrontier.backoff.max.secs` to the host queue ==== Robots.txt Compliance diff --git a/external/urlfrontier/README.md b/external/urlfrontier/README.md index 8647fc040..869ed043f 100644 --- a/external/urlfrontier/README.md +++ b/external/urlfrontier/README.md @@ -24,6 +24,48 @@ urlfrontier.max.buckets: 10 urlfrontier.max.urls.per.bucket:10 ``` +## Robots crawl-delay pacing + +`QueueRegulatorBolt` can pace host queues when a robots.txt Crawl-delay exceeds the fetcher's local +limit. Wire it to the `queue` stream emitted by `StatusUpdaterBolt`: + +```yaml +bolts: + - id: "queue-regulator" + className: "org.apache.stormcrawler.urlfrontier.QueueRegulatorBolt" + parallelism: 1 + +streams: + - from: "status" + to: "queue-regulator" + grouping: + type: FIELDS + args: ["key"] + streamId: "queue" +``` + +Robots pacing is opt-in. It requires host partitioning, one URL per frontier hand-out, a positive +delay cap, and a persist-only control signal: + +```yaml +partition.url.mode: byHost +fetcher.max.crawl.delay.force: true +urlfrontier.robots.crawl.delay.enabled: true +urlfrontier.max.urls.per.bucket: 1 +urlfrontier.backoff.max.secs: 86400 + +metadata.persist: + - robots.crawl.delay +``` + +Do not include `robots.crawl.delay` in `metadata.transfer`, directly or through a wildcard such as +`robots.*`: an outlink must not inherit its parent's host delay. The bolt rejects an unsafe robots +configuration at startup. A custom `metadata.transfer.class` must preserve this contract for every +URL and value; the startup probe can only exercise representative metadata. `setDelay(key, 0)` is +not sent if a site later removes its Crawl-delay, so that host can remain slower than necessary but +is not made less polite. A batch size of one bounds each hand-out; it does not recall URLs already +emitted or eliminate concurrent requests while the new delay is reaching the frontier. + Your StormCrawler topology requires the following dependency in its pom.xml (just like with any other module) ```