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:03 UTC
(stormcrawler) 03/08: feat: add CrawlDelayPolicy for forwarding robots delays to the frontier
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")));
+ }
+}