You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by dp...@apache.org on 2026/07/15 12:43:00 UTC
(stormcrawler) branch feat/1979-robots-delay created (now 31598b63)
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a change to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
at 31598b63 docs: document URLFrontier robots crawl-delay pacing
This branch includes the following new commits:
new 83e44354 feat: report the forced robots crawl-delay in the status metadata
new bbf317b5 feat: report the forced robots crawl-delay from SimpleFetcherBolt too
new 9ec07cff feat: add CrawlDelayPolicy for forwarding robots delays to the frontier
new 47ea808a feat: pace host queues on robots crawl-delay via setDelay
new 9f7b8de6 test: prove the frontier actually paces a queue on robots crawl-delay
new a827ca96 fix: preserve robots crawl-delay metadata semantics
new 2de0aa26 feat: enforce safe robots crawl-delay pacing
new 31598b63 docs: document URLFrontier robots crawl-delay pacing
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(stormcrawler) 06/08: fix: preserve robots crawl-delay metadata semantics
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit a827ca965a8b9c9671a76943e1172e08ae44c738
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 15 13:55:27 2026 +0200
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.
---
.../org/apache/stormcrawler/bolt/FetcherBolt.java | 16 ++++++-
.../stormcrawler/bolt/SimpleFetcherBolt.java | 17 +++++--
.../apache/stormcrawler/bolt/FetcherBoltTest.java | 36 +++++++++++++-
.../stormcrawler/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 2b0b263f..ac141a4a 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 class FetcherBolt extends StatusEmitterBolt {
// 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 class FetcherBolt extends StatusEmitterBolt {
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 class FetcherBolt extends StatusEmitterBolt {
// 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 2c47dee1..6c2dae6e 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 class SimpleFetcherBolt extends StatusEmitterBolt {
// 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 class SimpleFetcherBolt extends StatusEmitterBolt {
// 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 class SimpleFetcherBolt extends StatusEmitterBolt {
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 class SimpleFetcherBolt extends StatusEmitterBolt {
// 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 3f872c14..e7f592e1 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 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest {
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<String, Object> config = new HashMap<>();
config.put("http.agent.name", "this_is_only_a_test");
@@ -62,6 +68,26 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest {
assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
}
+ @Test
+ void fractionalLongCrawlDelayIsRoundedUp(WireMockRuntimeInfo wmRuntimeInfo)
+ throws ReflectiveOperationException {
+ stubFor(
+ get(urlEqualTo("/robots.txt"))
+ .willReturn(
+ aResponse()
+ .withStatus(200)
+ .withBody("User-agent: *\nCrawl-delay: 30.5\n")));
+ stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello")));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertEquals("31", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
+
@Test
void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo)
throws ReflectiveOperationException {
@@ -71,7 +97,13 @@ public class FetcherBoltTest extends AbstractFetcherBoltTest {
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<String, Object> 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 91565228..085ef805 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 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest {
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<String, Object> config = new HashMap<>();
config.put("http.agent.name", "this_is_only_a_test");
@@ -60,6 +66,26 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest {
assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
}
+ @Test
+ void fractionalLongCrawlDelayIsRoundedUp(WireMockRuntimeInfo wmRuntimeInfo)
+ throws ReflectiveOperationException {
+ stubFor(
+ get(urlEqualTo("/robots.txt"))
+ .willReturn(
+ aResponse()
+ .withStatus(200)
+ .withBody("User-agent: *\nCrawl-delay: 30.5\n")));
+ stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello")));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertEquals("31", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
+
@Test
void shortCrawlDelayIsNotReported(WireMockRuntimeInfo wmRuntimeInfo)
throws ReflectiveOperationException {
@@ -69,7 +95,13 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest {
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<String, Object> config = new HashMap<>();
config.put("http.agent.name", "this_is_only_a_test");
@@ -79,4 +111,24 @@ public class SimpleFetcherBoltTest extends AbstractFetcherBoltTest {
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<String, Object> 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));
+ }
}
(stormcrawler) 03/08: feat: add CrawlDelayPolicy for forwarding robots delays to the frontier
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 9ec07cff245a9e9006acceba251f0928f99c5dd6
Author: Davide Polato <dp...@apache.org>
AuthorDate: Thu Jul 9 07:52:33 2026 +0200
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.
---
.../stormcrawler/urlfrontier/CrawlDelayPolicy.java | 105 +++++++++++++++++++
.../urlfrontier/CrawlDelayPolicyTest.java | 111 +++++++++++++++++++++
2 files changed, 216 insertions(+)
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 00000000..c7c1ea41
--- /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.
+ *
+ * <p>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.
+ *
+ * <p>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<String, Integer> lastSent;
+
+ private final long maxSecs;
+
+ CrawlDelayPolicy(Map<String, Object> conf) {
+ this(conf, Ticker.systemTicker());
+ }
+
+ /** Visible for tests: a controllable ticker drives the dedupe window. */
+ CrawlDelayPolicy(Map<String, Object> 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 00000000..aed6abf5
--- /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<String, Object> 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")));
+ }
+}
(stormcrawler) 04/08: feat: pace host queues on robots crawl-delay via setDelay
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 47ea808a530b2f08b8d6d40b1b57b051718b3c23
Author: Davide Polato <dp...@apache.org>
AuthorDate: Thu Jul 9 09:18:07 2026 +0200
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 ae403b12..139ca02d 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;
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 @@ import org.slf4j.LoggerFactory;
* 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.
*
+ * <p>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.
+ *
* <p>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 @@ import org.slf4j.LoggerFactory;
* metadata.persist:
* - fetch.statusCode
* - protocol.retry-after
+ * - robots.crawl.delay
* </pre>
*
* (the second entry depends on {@code protocol.md.prefix}), and {@code fetch.exception} as well
@@ -124,6 +137,20 @@ public class QueueRegulatorBolt extends BaseRichBolt {
public void onCompleted() {}
};
+ private static final StreamObserver<Empty> 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 class QueueRegulatorBolt extends BaseRichBolt {
/** 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<String, Object> 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 class QueueRegulatorBolt extends BaseRichBolt {
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 @@ public class QueueRegulatorBolt extends BaseRichBolt {
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<String> missing = new LinkedHashSet<>();
if (filtered.getFirstValue(HostBackoff.STATUS_CODE_KEY) == null) {
@@ -216,6 +261,10 @@ public class QueueRegulatorBolt extends BaseRichBolt {
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 8f068ca2..7ca59b1d 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<String> 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<String, Object> 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 @@ class QueueRegulatorBoltDecisionTest {
@Test
void missingQueueStreamKeysHonoursWildcards() {
Map<String, Object> 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 @@ class QueueRegulatorBoltDecisionTest {
@Test
void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() {
Map<String, Object> 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(
(stormcrawler) 01/08: feat: report the forced robots crawl-delay in the status metadata
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 83e44354bf397aabbd3820521cc26540288a4976
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 8 14:27:41 2026 +0200
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.
---
.../java/org/apache/stormcrawler/Constants.java | 7 ++
.../org/apache/stormcrawler/bolt/FetcherBolt.java | 5 ++
.../stormcrawler/bolt/AbstractFetcherBoltTest.java | 54 ++++++++++++++-
.../apache/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 e06552a6..882f2d96 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 2439bd76..2b0b263f 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 class FetcherBolt extends StatusEmitterBolt {
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 4f4aa521..35cd7b5f 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 @@ abstract class AbstractFetcherBoltTest {
Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size());
}
- private void resetProtocolFactory() throws ReflectiveOperationException {
+ TestOutputCollector fetch(
+ WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+ throws ReflectiveOperationException {
+ resetProtocolFactory();
+ TestOutputCollector output = new TestOutputCollector();
+ bolt.prepare(config, TestUtil.getMockedTopologyContext(), new OutputCollector(output));
+
+ Tuple tuple = mock(Tuple.class);
+ when(tuple.getSourceComponent()).thenReturn("source");
+ when(tuple.getStringByField("url"))
+ .thenReturn("http://localhost:" + wmRuntimeInfo.getHttpPort() + path);
+ when(tuple.getValueByField("metadata")).thenReturn(null);
+ bolt.execute(tuple);
+
+ await().atMost(30, TimeUnit.SECONDS)
+ .until(
+ () ->
+ output.getEmitted(Utils.DEFAULT_STREAM_ID).size() > 0
+ || output.getEmitted(Constants.StatusStreamName).size()
+ > 0);
+ return output;
+ }
+
+ /**
+ * Fetches a page that is expected to be retrieved successfully (HTTP 200, non-304): the fetcher
+ * emits it on the default stream (url, content, metadata) for downstream parsing.
+ */
+ Metadata fetchAndGetContentMetadata(
+ WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+ throws ReflectiveOperationException {
+ TestOutputCollector output = fetch(wmRuntimeInfo, config, path);
+ List<List<Object>> contentTuples = output.getEmitted(Utils.DEFAULT_STREAM_ID);
+ Assertions.assertEquals(1, contentTuples.size());
+ Assertions.assertEquals(0, output.getEmitted(Constants.StatusStreamName).size());
+ return (Metadata) contentTuples.get(0).get(2);
+ }
+
+ /**
+ * Fetches a page that is expected to be rejected before any HTTP request is made (e.g. the
+ * crawl-delay-too-long guard): the fetcher emits directly on the status stream (url, metadata,
+ * status).
+ */
+ List<Object> fetchAndGetStatusTuple(
+ WireMockRuntimeInfo wmRuntimeInfo, Map<String, Object> config, String path)
+ throws ReflectiveOperationException {
+ TestOutputCollector output = fetch(wmRuntimeInfo, config, path);
+ List<List<Object>> statusTuples = output.getEmitted(Constants.StatusStreamName);
+ Assertions.assertEquals(1, statusTuples.size());
+ Assertions.assertEquals(0, output.getEmitted(Utils.DEFAULT_STREAM_ID).size());
+ return statusTuples.get(0);
+ }
+
+ static void resetProtocolFactory() throws ReflectiveOperationException {
Field instance = ProtocolFactory.class.getDeclaredField("single_instance");
instance.setAccessible(true);
instance.set(null, null);
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 8019bcf3..3f872c14 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<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
+
+ @Test
+ void 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<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
+
+ @Test
+ void unforcedLongCrawlDelayStillEmitsCrawlDelayErrorWithoutMetadata(
+ WireMockRuntimeInfo wmRuntimeInfo) throws ReflectiveOperationException {
+ // regression guard: force=false + long delay must keep today's behaviour, i.e. a
+ // Status.ERROR/crawl_delay emission and no robots.crawl.delay metadata key
+ stubFor(
+ get(urlEqualTo("/robots.txt"))
+ .willReturn(
+ aResponse()
+ .withStatus(200)
+ .withBody("User-agent: *\nCrawl-delay: 120\n")));
+ stubFor(get(urlEqualTo("/page")).willReturn(aResponse().withStatus(200).withBody("hello")));
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", false);
+
+ List<Object> statusTuple = fetchAndGetStatusTuple(wmRuntimeInfo, config, "/page");
+ assertEquals(Status.ERROR, statusTuple.get(2));
+ Metadata md = (Metadata) statusTuple.get(1);
+ assertEquals("crawl_delay", md.getFirstValue(Constants.STATUS_ERROR_CAUSE));
+ assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
}
(stormcrawler) 07/08: feat: enforce safe robots crawl-delay pacing
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 2de0aa26b3cfd630fdaafb81a95f5d05521658f3
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 15 13:55:52 2026 +0200
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.
---
.../apache/stormcrawler/urlfrontier/Constants.java | 7 +
.../stormcrawler/urlfrontier/CrawlDelayPolicy.java | 127 ++++++++++---
.../urlfrontier/QueueRegulatorBolt.java | 209 ++++++++++++++++-----
.../org/apache/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 8fd7f4de..976cb739 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 @@ public final class 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 @@ public final class 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 c7c1ea41..b33ba93d 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.Cache;
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 @@ 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.
+ * 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}.
*
* <p>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.
*
- * <p>Not thread-safe; a bolt task calls {@link #delaySecsFor} from a single thread.
+ * <p>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<String, Integer> lastSent;
+ /** Last delay applied successfully per host; expiry bounds the dedupe window and memory. */
+ private final Cache<String, Integer> lastApplied;
+
+ /** At most one unary RPC per host, plus the latest value observed while it is active. */
+ private final ConcurrentMap<String, Flight> 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<String, Object> conf) {
this(conf, Ticker.systemTicker());
}
/** Visible for tests: a controllable ticker drives the dedupe window. */
CrawlDelayPolicy(Map<String, Object> 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<DelayRequest> 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<DelayRequest> 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<DelayRequest> completed(DelayRequest completed, boolean success) {
+ AtomicReference<DelayRequest> 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 139ca02d..6e9d1418 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 @@ import org.slf4j.LoggerFactory;
/**
* 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.
*
* <p>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 @@ import org.slf4j.LoggerFactory;
* 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.
*
- * <p>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
+ * <p>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 @@ import org.slf4j.LoggerFactory;
* metadata.persist:
* - fetch.statusCode
* - protocol.retry-after
- * - robots.crawl.delay
* </pre>
*
* (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.
*
- * <p>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.
+ * <p>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 <b>not</b> 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.
+ *
+ * <p>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.
*
* <p>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.
*
- * <p>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.
+ * <p>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 class QueueRegulatorBolt extends BaseRichBolt {
public void onCompleted() {}
};
- private static final StreamObserver<Empty> 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 class QueueRegulatorBolt extends BaseRichBolt {
/** 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<String, Object> 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 class QueueRegulatorBolt extends BaseRichBolt {
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<Empty> 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 @@ public class QueueRegulatorBolt extends BaseRichBolt {
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<String> missing = new LinkedHashSet<>();
if (filtered.getFirstValue(HostBackoff.STATUS_CODE_KEY) == null) {
@@ -261,13 +293,90 @@ public class QueueRegulatorBolt extends BaseRichBolt {
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<String, Object> 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 class QueueRegulatorBolt extends BaseRichBolt {
@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 6aa9ef7f..9763051e 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_DEFAULT_
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 class Spout extends AbstractQueryingSpout {
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 aed6abf5..10436c64 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 @@ class CrawlDelayPolicyTest {
private CrawlDelayPolicy policy(Object... keysAndValues) {
Map<String, Object> 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 @@ class CrawlDelayPolicyTest {
@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 7ca59b1d..760bd5e6 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 @@ class QueueRegulatorBoltDecisionTest {
// with the default metadata.persist none of the keys survives the filter
Set<String> 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<String, Object> 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 @@ class QueueRegulatorBoltDecisionTest {
@Test
void missingQueueStreamKeysHonoursWildcards() {
Map<String, Object> 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 @@ class QueueRegulatorBoltDecisionTest {
@Test
void exceptionKeyOnlyRequiredWhenBackoffOnExceptionsIsEnabled() {
Map<String, Object> 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 @@ class QueueRegulatorBoltDecisionTest {
Set.of("fetch.exception"),
QueueRegulatorBolt.missingQueueStreamKeys(conf, RETRY_AFTER_KEY, true));
}
+
+ @Test
+ void robotsValidationIsSkippedWhenFrontierPacingIsDisabled() {
+ assertDoesNotThrow(
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(new HashMap<>()));
+ }
+
+ @Test
+ void legacyFetcherForceDoesNotEnableFrontierPacing() {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put("fetcher.max.crawl.delay.force", true);
+ assertDoesNotThrow(() -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+
+ @Test
+ void enabledRobotsPacingRequiresFetcherForce() {
+ Map<String, Object> 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<String, Object> conf = validRobotsConf();
+ conf.remove("metadata.persist");
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+
+ @Test
+ void robotsSignalMustNotTransferToOutlinks() {
+ Map<String, Object> conf = validRobotsConf();
+ conf.put("metadata.transfer", List.of("robots.*"));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+
+ @Test
+ void robotsPersistenceWildcardIsAccepted() {
+ Map<String, Object> 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<String, Object> conf = validRobotsConf();
+ conf.put(org.apache.stormcrawler.Constants.PARTITION_MODEParamName, mode);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+ }
+
+ @Test
+ void robotsPacingRequiresSingleUrlHandouts() {
+ Map<String, Object> defaultBatch = validRobotsConf();
+ defaultBatch.remove(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(defaultBatch));
+
+ Map<String, Object> multiUrlBatch = validRobotsConf();
+ multiUrlBatch.put(Constants.URLFRONTIER_MAX_URLS_PER_BUCKET_KEY, 2);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(multiUrlBatch));
+ }
+
+ @Test
+ void robotsPacingRequiresAnEnabledCrawlDelayLimit() {
+ Map<String, Object> conf = validRobotsConf();
+ conf.put("fetcher.max.crawl.delay", -1);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+
+ @Test
+ void robotsPacingRequiresAPositiveFrontierDelayCap() {
+ Map<String, Object> conf = validRobotsConf();
+ conf.put(Constants.URLFRONTIER_BACKOFF_MAX_KEY, 0);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> QueueRegulatorBolt.validateRobotsPacingConfiguration(conf));
+ }
+
+ private static Map<String, Object> validRobotsConf() {
+ Map<String, Object> 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 ec71bcf6..04183442 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.time.Duration;
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 @@ class QueueRegulatorBoltTest {
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<String, Object> 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 @@ class QueueRegulatorBoltTest {
// 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
(stormcrawler) 02/08: feat: report the forced robots crawl-delay from SimpleFetcherBolt too
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit bbf317b51064fa8f8f2de059a2097c3dd967d219
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 8 15:43:16 2026 +0200
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 ++
.../stormcrawler/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 da40b069..2c47dee1 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 class SimpleFetcherBolt extends StatusEmitterBolt {
// 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 da212022..91565228 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<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertEquals("120", md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
+
+ @Test
+ void 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<String, Object> config = new HashMap<>();
+ config.put("http.agent.name", "this_is_only_a_test");
+ config.put("fetcher.max.crawl.delay", 30);
+ config.put("fetcher.max.crawl.delay.force", true);
+
+ Metadata md = fetchAndGetContentMetadata(wmRuntimeInfo, config, "/page");
+ assertNull(md.getFirstValue(Constants.ROBOTS_CRAWL_DELAY_KEY));
+ }
}
(stormcrawler) 05/08: test: prove the frontier actually paces a queue on robots crawl-delay
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 9f7b8de659bfa64908d4aad4fdfa2008e0371d5c
Author: Davide Polato <dp...@apache.org>
AuthorDate: Thu Jul 9 10:11:54 2026 +0200
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 87a88dcc..ec71bcf6 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 @@ class QueueRegulatorBoltTest {
/** 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<String, Object> config = frontierConfig();
@@ -101,18 +107,30 @@ class QueueRegulatorBoltTest {
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<String> 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 @@ class QueueRegulatorBoltTest {
bolt.cleanup();
}
+
+ @Test
+ @Timeout(value = 2, unit = TimeUnit.MINUTES)
+ void pacesHostQueueOnRobotsCrawlDelay() {
+ seedUrls("/1", "/2");
+
+ QueueRegulatorBolt bolt = new QueueRegulatorBolt();
+ Map<String, Object> 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<URLInfo> it = blocking.getURLs(params);
+ while (it.hasNext()) {
+ it.next();
+ served++;
+ }
+ return served;
+ }
}
(stormcrawler) 08/08: docs: document URLFrontier robots crawl-delay pacing
Posted by dp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dpol1 pushed a commit to branch feat/1979-robots-delay
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 31598b63019f5ba5d7941dad1210e437c982fa2e
Author: Davide Polato <dp...@apache.org>
AuthorDate: Wed Jul 15 13:57:43 2026 +0200
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 9961320b..9b176084 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 fd9e52b6..f938b14f 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 7277f88b..5d1d7479 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 8647fc04..869ed043 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)
```