Block the URLFrontier queue on Retry-After via the queue stream

[dpol1]

Closes #784.

On an HTTP 429 or 503 response, the Retry-After header (delta-seconds or IMF-fixdate) already reaches the status stream: the protocol layer stores response headers in the status metadata, along with fetch.statusCode. The URLFrontier StatusUpdaterBolt now forwards (key, metadata) on the generic queue stream introduced in #1974, mirroring the OpenSearch updater. A new HostBlockBolt consumes that stream, parses the header via the standalone RetryAfterParser, caps the delay with the new urlfrontier.max.retry.after setting (default 24h, -1 for no cap) and calls blockQueueUntil, so the frontier stops handing out URLs for that host until the requested time. FetcherBolt is untouched.

This follows the conclusion of the discussion in #1944: the fetch pipeline itself never delays or holds tuples; enforcement lives with the queue owner. It implements the rate-limit case of #867 on top of the queue stream. The header parsing and the cap reuse the groundwork @rzo1 laid in #1944; what changes is where the signal is read and enforced.

Wiring:

  - id: "hostblock"
    className: "org.apache.stormcrawler.urlfrontier.HostBlockBolt"
    parallelism: 1

  - from: "status"
    to: "hostblock"
    grouping:
      type: FIELDS
      args: ["key"]
      streamId: "queue"

A few design decisions worth calling out for review:

• The queue key is the one the status updater computes with URLPartitioner from partition.url.mode, which is the same setting the frontier keys its queues by. A block on a differently derived key would be a silent no-op.
• Only 429 and 503 are considered, where Retry-After has back-off semantics; Retry-After on a 3xx is deliberately ignored. A configurable list of status codes belongs to the adaptive follow-up.
• The HTTP-date form accepts IMF-fixdate only, which is what servers produce in practice; the obsolete RFC 850/asctime forms are rejected.
• The gRPC call is fire-and-forget: a missed block costs one extra fetch and the next 429 re-emits the signal.
• Topologies that don't wire a consumer are unaffected: tuples emitted on a stream without subscribers are simply discarded.
• URLs already sitting in the fetcher's internal queues for the blocked host are still fetched; the frontier block only stops new hand-outs. Discussed with @sebastian-nagel on Send host info on a specific stream  #867: they fail on their own and reschedule via the status stream, since they never received a 429 themselves and no Status encodes a purge.

One scope question: should this solves #867 as well, or should the issue stay open for a future robots-delay signal on the same stream?

For all changes

• Is there a issue associated with this PR? Is it referenced in the commit message?
• Does your PR title start with #XXXX where XXXX is the issue number you are trying to resolve?
• Has your PR been rebased against the latest commit within the target branch (typically main)?
• Is your initial contribution a single, squashed commit? (reworked after review, history kept for the discussion)
• Is the code properly formatted with mvn git-code-format:format-code -Dgcf.globPattern="**/*" -Dskip.format.code=false?

For code changes

• Have you ensured that the full suite of tests is executed via mvn clean verify?
• Have you written or updated unit tests to verify your changes?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0? (no new dependencies)
• If applicable, have you updated the LICENSE file, including the main LICENSE file? (n/a)
• If applicable, have you updated the NOTICE file, including the main NOTICE file? (n/a)

[jnioche]

Thanks @dpol1, 2 very quick comments for now

• This PR adds more content to FetcherBolt, which is already pretty massive as a class and I think we should break it down to make it more manageable. Should that be done first?

• We should probably first add a basic mechanism for handling host info as part of the StatusUpdaterBolt (i.e. add hosts if they don't already exist). Branch 990 has code to that effect. We can then move on to more advanced uses of the queue / host stream like this one.

What do you think?

[dpol1]

Taking the second point first, since it changes the answer to the first one.

We should probably first add a basic mechanism for handling host info as part of the StatusUpdaterBolt (i.e. add hosts if they don't already exist). Branch 990 has code to that effect. We can then move on to more advanced uses of the queue / host stream like this one.

Agreed on the sequencing. I had a look at branch 990 and have one question about the stream schema: it emits just (key). For a use case like this one the consumer needs a bit more context. With the okhttp protocol the response headers already end up in the status metadata (protocol.retry-after), and fetch.statusCode is there too. If the stream carried (key, metadata) instead, the whole Retry-After handling could move out of FetcherBolt: the URLFrontier-side bolt would read status code and header from the metadata and call BlockQueueUntil, with the parsing in a small standalone util class (the tests for it are already written). The stream being per-URL is not a problem here, since the consumer only acts on a 429/503 with a Retry-After present.

Would you be open to (key, metadata), or would you rather keep the basic stream minimal and add the extra info in a follow-up?

This PR adds more content to FetcherBolt, which is already pretty massive as a class and I think we should break it down to make it more manageable. Should that be done first?

If we go the route above, this PR stops touching FetcherBolt entirely, so the two things become independent. I agree the class needs breaking down, but that feels like its own issue rather than a blocker here. Happy to help with it separately.

On branch 990: do you plan to open a PR for it yourself, or should I pick the core parts (the stream constant, the declaration in AbstractStatusUpdaterBolt, the OpenSearch emit and QueueBolt) onto a fresh branch off main and open a PR from there? I'd then rework this one on top once the basic mechanism is in.

[jnioche]

Would you be open to (key, metadata), or would you rather keep the basic stream minimal and add the extra info in a follow-up?

It totally makes sense to have key + metadata. The code in the 990 branch was a rather minimal starting point. Whatever bolt receives the stream can choose to use metadata or not, based on what it needs to do.

This PR adds more content to FetcherBolt, which is already pretty massive as a class and I think we should break it down to make it more manageable. Should that be done first?

If we go the route above, this PR stops touching FetcherBolt entirely, so the two things become independent. I agree the class needs breaking down, but that feels like its own issue rather than a blocker here. Happy to help with it separately.

100%, breaking down the FetcherBolt is a separate task

On branch 990: do you plan to open a PR for it yourself, or should I pick the core parts (the stream constant, the declaration in AbstractStatusUpdaterBolt, the OpenSearch emit and QueueBolt) onto a fresh branch off main and open a PR from there? I'd then rework this one on top once the basic mechanism is in.

I'll have a go at writing a PR for 990 today

[jnioche]

@dpol1 see #1974

[dpol1]

Ready for review now - Fancy leave a comment on Stage 2 in #867 discussion? Just like Sebastian did. Even just a thumbs-up is fine :-)

One scope question: should this solves #867 as well, or should the issue stay open for a future robots-delay signal on the same stream?

and on this as well thx Never mind, I don't think so

[jnioche]

thanks @dpol1, I like this very much.

Instead of having a new bolt, just for handling block events, what about a generic QueueBolt, just like the one in the OpenSearch module but with a pluggable mechanism to handle the blocking logic within it?

URLFrontier handles the queues itself and does not need to be told about them like OpenSearch, so we could do without the QueueBolt but we probably don't want the duplication of HostBlocks implementations for every single type of backend. Instead we would have a single implementation in core and let each QueueBolt convert it into the appropriate action.

Make things a bit more complex now, but easier to handle later.

[jnioche]

thanks @dpol1, I like this very much.

Instead of having a new bolt, just for handling block events, what about a generic QueueBolt, just like the one in the OpenSearch module but with a pluggable mechanism to handle the blocking logic within it?

URLFrontier handles the queues itself and does not need to be told about them like OpenSearch, so we could do without the QueueBolt but we probably don't want the duplication of HostBlocks implementations for every single type of backend. Instead we would have a single implementation in core and let each QueueBolt convert it into the appropriate action.

Make things a bit more complex now, but easier to handle later.

Maybe I am making it more complicated than necessary. We can have some duplication initially, learn from it and make things more generic afterwards.

[dpol1]

Maybe I am making it more complicated than necessary. We can have some duplication initially, learn from it and make things more generic afterwards.

yes, I'd wait too. URLFrontier is the only backend with a real blocking action today (blockQueueUntil); for OpenSearch or Solr I wouldn't even know what the action should be yet, so a shared base class now would end up designed around a single case. The pieces that would move to core are already isolated anyway: the parsing is in RetryAfterParser, and the 429/503 decision in HostBlockBolt is a static function with no gRPC in it. Whenever a second backend shows up it's a mechanical lift.

[rzo1]

Thanks @dpol1! As long as Fable 5 is still available, here is its review — I will look at it later myself in a human review too. So take the robot's word with the appropriate grain of salt, but it did read the whole thing so I don't have to pretend I did (yet). 🤖

Overall it likes the design: parser in core, enforcement with the queue owner, good unit coverage. It found one broken link in the end-to-end chain and a config gap, though:

1. Silent no-op with default config — the big one. AbstractStatusUpdaterBolt.execute() runs metadata = mdTransfer.filter(metadata) before calling store(), and the default MetadataTransfer only keeps url.path, depth, max.depth and the fetch error count. So fetch.statusCode and <prefix>retry-after are stripped before the queue-stream emit unless both are added to metadata.persist. A user wiring HostBlockBolt exactly as the javadoc says gets blockUntilFor() == -1 for every tuple, with no log hint. The tests don't catch it because they hand-craft the metadata instead of going through the status updater. Either document the two required metadata.persist entries prominently (like AdaptiveScheduler does), or emit the pre-filter metadata on the queue stream.

2. HostBlockBolt ignores urlfrontier.address. It only reads urlfrontier.host/port, while Spout and StatusUpdaterBolt resolve urlfrontier.address first. In a topology configured via urlfrontier.address (even a single node — not just the multi-node case the javadoc disclaims), the bolt silently connects to localhost:7071, and with withWaitForReady() the RPCs never fail, so nothing is ever logged. Since that address-resolution block is already duplicated between Spout and StatusUpdaterBolt, extracting it (e.g. into ManagedChannelUtil) and reusing it here would fix the bug and delete two copies.

3. Overflow when the cap is disabled (fail-open). With urlfrontier.max.retry.after: -1, a server sending Retry-After: 9223372036854775 passes the parser's multiplyExact guard (~9.2e18 ms), then nowMs + retryAfterMs in blockUntilFor wraps negative and the block is skipped — exactly when the server asked loudest for a back-off. The parser was careful with multiplyExact; the addition wants Math.addExact + clamp too. Masked by the default 24h cap.

4. _DEFAULT_ queue can be collateral damage. A 429 on a URL whose partition key falls back to "_DEFAULT_" makes the bolt call blockQueueUntil(key="_DEFAULT_"), stalling every unrelated URL in the shared catch-all queue for up to 24h. Cheap guard: skip the block for the sentinel key.

5. gRPC robustness. withWaitForReady() + no deadline on a fire-and-forget RPC means a frontier outage during a 429 storm accumulates pending RPCs unboundedly and flushes stale blocks on reconnect — for a best-effort call, drop waitForReady or add a short withDeadlineAfter. Also, a burst of 429s from one host issues one RPC per tuple for the same key; a tiny per-key blockedUntil cache would make it one per back-off window.

6. The "while blocked" assertion in HostBlockBoltTest is vacuous. The first getURLs poll hands the seeded URL out and it goes in-flight, so await(!keys.contains(HOST)) passes within a couple of polls even if the block silently failed. Asserting the URL is never handed out (first call empty) would test the block and remove the race the unblock phase has with the in-flight timeout.

7. Minor / cleanup:

• The queue-stream emit is now copy-pasted verbatim from the OpenSearch updater, while AbstractStatusUpdaterBolt declares the stream for all seven subclasses but only two emit on it. A protected helper in the abstract class next to the declareStream call would keep the contract uniform.
• The bespoke HTTP-date formatter is declared twice (parser + test), so the round-trip test can't catch a wrong pattern. IMF-fixdate-only strictness is a fair call per the PR description, but formatting the test inputs with the JDK's RFC_1123_DATE_TIME would make the test an actual cross-check.

Findings 1 and 2 are the ones the robot would block on; the rest are hardening. Human review to follow — if it disagrees with the machine, the human wins. 😄

[dpol1]

Checked all seven against the code, the robot did its homework. Replies inline.

So fetch.statusCode and <prefix>retry-after are stripped before the queue-stream emit unless both are added to metadata.persist.

Confirmed, my tests hand-crafted the metadata so they never hit the filter. I'll document the two metadata.persist entries and make prepare() warn when they don't survive the filter. Emitting pre-filter metadata changes the #1974 contract in core, @jnioche's call.

extracting it (e.g. into ManagedChannelUtil) and reusing it here would fix the bug and delete two copies.

Bug confirmed, fixing. Not deleting the copies though: Spout and updater validate differently on multi-node (== vs multiple), and the bolt sends local=false so any node works. It'll just resolve urlfrontier.address itself.

the addition wants Math.addExact + clamp too.

Yep, taking it.

Cheap guard: skip the block for the sentinel key.

Taking it.

drop waitForReady or add a short withDeadlineAfter

Adding the deadline. The per-key cache I'd leave for the adaptive back-off follow-up (#867 phase 2), duplicate blocks are idempotent frontier-side anyway.

Asserting the URL is never handed out (first call empty) would test the block

Right, the current assertion is vacuous once the URL goes in-flight. Inverting: block first, then seed.

A protected helper in the abstract class next to the declareStream call would keep the contract uniform.

Follow-up material for me, not touching core + OpenSearch for two call sites. Happy to do it when a third consumer shows up.

formatting the test inputs with the JDK's RFC_1123_DATE_TIME would make the test an actual cross-check.

Take it easy Claude: it prints single-digit days unpadded, which the strict dd pattern rejects on purpose. Swap would break for a third of the month. I'll use it for two-digit days and add a rejection test for the unpadded form.

Fixes incoming. No rush on the human review, the robot already gave me enough homework 😄

[dpol1]

Merged - thx @jnioche, @mvolikas and @rzo1 for the review. I ran it locally just to be safe and everything's in order