From 5b55319d271cca94b798ce5db5faecffc4d34eef Mon Sep 17 00:00:00 2001 From: pr0vieh Date: Tue, 20 Jan 2026 22:17:14 +0100 Subject: feat: Add two-layer DHT error propagation to prevent broken URL redistribution Problem: Failed URLs (404, DNS errors, timeouts) are continuously redistributed via DHT, causing infinite recrawl loops and network-wide index pollution. No mechanism exists in YaCy to communicate error status across peers. Solution - Layer 1 (Proactive Rejection): - Receiver checks local Solr index for httpstatus_i != 200 BEFORE accepting RWI entries - Rejects URL immediately if marked as failed previously - Adds rejected URL hash to errorURL response list - Prevents index pollution at ingestion time - Works even if sender doesn't support errorURL protocol (backward compatible) Solution - Layer 2 (Error Feedback): - Receiver reports rejected error URLs back to sender via errorURL response parameter - Sender receives errorURL list and marks those URLs locally as failed - Sender stops re-distributing these URLs to other peers - Network-wide error propagation prevents repeated distribution cycles Implementation Details: - transferURL.java: Implements proactive rejection + error reporting * Checks incoming URL against Solr error status before storing * Collects rejected URL hashes in errorURLs StringBuilder * Returns errorURL list to sender in response - Protocol.java: Processes error URL feedback from receiver * Extracts errorURL from response * Marks reported URLs locally via crawlQueues.errorURL.push() * Logs DHT error reports for monitoring Benefits: - Dramatically reduces network traffic of broken URLs - Prevents wasted crawl resources on unreachable targets - Maintains clean, usable index across distributed network - Defense-in-depth: two independent layers work together - Backward compatible: old peers ignore errorURL parameter Testing: - Log monitoring shows 'DHT: Received X rejected error URL reports from peer Y' - Proactive rejection shows 'blocked X URLs' in transfer logs - Error URLs automatically removed from circulation --- source/net/yacy/htroot/yacy/transferURL.java | 31 +++++++- source/net/yacy/peers/Protocol.java | 113 ++++++++++++++++++++++----- 2 files changed, 123 insertions(+), 21 deletions(-) diff --git a/source/net/yacy/htroot/yacy/transferURL.java b/source/net/yacy/htroot/yacy/transferURL.java index b9c9e2f82..f9b95e270 100644 --- a/source/net/yacy/htroot/yacy/transferURL.java +++ b/source/net/yacy/htroot/yacy/transferURL.java @@ -76,6 +76,7 @@ public final class transferURL { // response values String result = ""; String doublevalues = "0"; + final StringBuilder errorURLs = new StringBuilder(); final Seed otherPeer = sb.peers.get(iam); final String otherPeerName = iam + ":" + ((otherPeer == null) ? "NULL" : (otherPeer.getName() + "/" + otherPeer.getVersion())); @@ -150,9 +151,29 @@ public final class transferURL { for (final String id : lEm.keySet()) { if (sb.index.exists(id)) { doublecheck++; - } else { - lEntry = lEm.get(id); - + // Check if entry we already have is marked as error - if so, reject incoming replacement + try { + final URIMetadataNode meta = sb.index.fulltext().getMetadata(ASCII.getBytes(id)); + if (meta != null && meta.getFieldValue("httpstatus_i") != null) { + final int httpstatus = (meta.getFieldValue("httpstatus_i") instanceof Integer) ? + (Integer) meta.getFieldValue("httpstatus_i") : + Integer.parseInt(meta.getFieldValue("httpstatus_i").toString()); + final Object failreason = meta.getFieldValue("failreason_s"); + if (httpstatus != 200 && failreason != null && failreason.toString().length() > 0) { + if (Network.log.isFine()) Network.log.fine("transferURL: rejected URL hash '" + id + "' (known error, httpstatus=" + httpstatus + ") from peer " + otherPeerName); + errorURLs.append(id).append(','); + blocked++; + continue; + } + } + } catch (final Exception e) { + // Ignore errors during error status check + } + } + + lEntry = lEm.get(id); + + if (lEntry != null) { // write entry to database if (Network.log.isFine()) Network.log.fine("Accepting URL from peer " + otherPeerName + ": " + lEntry.url().toNormalform(true)); try { @@ -182,6 +203,10 @@ public final class transferURL { prop.put("double", doublevalues); prop.put("result", result); + if (errorURLs.length() > 0) { + errorURLs.setLength(errorURLs.length() - 1); // remove trailing comma + prop.put("errorURL", errorURLs.toString()); + } return prop; } } diff --git a/source/net/yacy/peers/Protocol.java b/source/net/yacy/peers/Protocol.java index 569dfccc9..f81caaa57 100644 --- a/source/net/yacy/peers/Protocol.java +++ b/source/net/yacy/peers/Protocol.java @@ -88,6 +88,7 @@ import net.yacy.cora.document.feed.RSSFeed; import net.yacy.cora.document.feed.RSSMessage; import net.yacy.cora.document.feed.RSSReader; import net.yacy.cora.document.id.MultiProtocolURL; +import net.yacy.cora.federate.solr.FailCategory; import net.yacy.cora.federate.solr.connector.RemoteSolrConnector; import net.yacy.cora.federate.solr.connector.SolrConnector; import net.yacy.cora.federate.solr.instance.RemoteInstance; @@ -789,24 +790,24 @@ public final class Protocol { } WriteMetadataNodeToLocalIndexThread writerToLocalIndex = new WriteMetadataNodeToLocalIndexThread(event.query.getSegment(), storeDocs); writerToLocalIndex.start(); - try { - writerToLocalIndex.join(); - } catch(InterruptedException e) { - /* - * Current thread interruption might happen while waiting - * for writeToLocalIndexThread. - */ - writerToLocalIndex.stopWriting(); - throw new InterruptedException("remoteProcess stopped!"); - } - /* Ensure freshly stored metadata is visible to queries before adding results. */ - event.query.getSegment().fulltext().commit(true); - if (storeDocs != null && !storeDocs.isEmpty()) { - event.addNodes(storeDocs, null, snip, false, target.getName() + "/" + target.hash, result.totalCount, true); - } else { - event.addRWIs(container.get(0), false, target.getName() + "/" + target.hash, result.totalCount, time); - } - } else { + try { + writerToLocalIndex.join(); + } catch(InterruptedException e) { + /* + * Current thread interruption might happen while waiting + * for writeToLocalIndexThread. + */ + writerToLocalIndex.stopWriting(); + throw new InterruptedException("remoteProcess stopped!"); + } + /* Ensure freshly stored metadata is visible to queries before adding results. */ + event.query.getSegment().fulltext().commit(true); + if (storeDocs != null && !storeDocs.isEmpty()) { + event.addNodes(storeDocs, null, snip, false, target.getName() + "/" + target.hash, result.totalCount, true); + } else { + event.addRWIs(container.get(0), false, target.getName() + "/" + target.hash, result.totalCount, time); + } + } else { // feed results as nodes (SolrQuery results) which carry metadata, // to prevent a call to getMetaData for RWI results, which would fail (if no metadata in index and no display of these results) event.addNodes(storeDocs, null, snip, false, target.getName() + "/" + target.hash, count, true); @@ -1737,6 +1738,52 @@ public final class Protocol { return result; } + // DHT error propagation: process error URLs reported by remote peer + String errorURLs = in.get("errorURL"); + if ( errorURLs != null && !errorURLs.isEmpty() && !errorURLs.equals(",") ) { + final String[] euhs = CommonPattern.COMMA.split(errorURLs.trim()); + if ( euhs.length > 0 ) { + Network.log.info("DHT: Received " + euhs.length + " error URL reports from peer " + targetSeed.getName() + "/[" + targetSeed.hash + "]"); + for ( final String errorHash : euhs ) { + if ( errorHash == null || errorHash.length() != 12 ) continue; + try { + // Check if we have this URL locally without error status + final URIMetadataNode metadata = segment.fulltext().getMetadata(ASCII.getBytes(errorHash)); + if ( metadata != null ) { + // Extract crawl depth if available; default to 0 when missing + int crawldepth = 0; + final Object cd = metadata.getFieldValue(CollectionSchema.crawldepth_i.getSolrFieldName()); + if (cd instanceof Integer) { + crawldepth = ((Integer) cd).intValue(); + } else if (cd instanceof Long) { + crawldepth = ((Long) cd).intValue(); + } + + // Mark as error to prevent re-distribution via DHT + sb.crawlQueues.errorURL.push( + metadata.url(), + crawldepth, + null, + net.yacy.cora.federate.solr.FailCategory.FINAL_LOAD_CONTEXT, + "DHT error propagation from peer " + targetSeed.getName(), + -1 + ); + if (Network.log.isFine()) Network.log.fine("DHT: Marked URL hash '" + errorHash + "' as error based on peer report"); + } + } catch ( final Exception e ) { + Network.log.warn("DHT: Failed to process error URL hash '" + errorHash + "': " + e.getMessage()); + } + } + EventChannel.channels(EventChannel.DHTRECEIVE).addMessage( + new RSSMessage( + "Received " + euhs.length + " error URL reports from peer " + targetSeed.getName() + "/[" + targetSeed.hash + "]", + "", + targetSeed.hash + ) + ); + } + } + // in now contains a list of unknown hashes String uhss = in.get("unknownURL"); if ( uhss == null ) { @@ -1773,6 +1820,36 @@ public final class Protocol { sb.peers.addConnected(targetSeed); // update the peer return result; } + + // Process error URLs reported back by receiver + final String rejectedURLs = in.get("errorURL"); + if (rejectedURLs != null && !rejectedURLs.isEmpty() && !rejectedURLs.equals(",")) { + final String[] ruhs = CommonPattern.COMMA.split(rejectedURLs.trim()); + if (ruhs.length > 0) { + Network.log.info("DHT: Received " + ruhs.length + " rejected error URL reports from peer " + targetSeed.getName()); + for (final String errorHash : ruhs) { + if (errorHash == null || errorHash.length() != 12) continue; + try { + final URIMetadataNode metadata = segment.fulltext().getMetadata(ASCII.getBytes(errorHash)); + if (metadata != null) { + // Extract crawl depth safely + int crawldepth = 0; + final Object cd = metadata.getFieldValue("crawldepth_i"); + if (cd instanceof Integer) crawldepth = ((Integer) cd).intValue(); + else if (cd instanceof Long) crawldepth = ((Long) cd).intValue(); + + // Mark as error locally + sb.crawlQueues.errorURL.push(metadata.url(), crawldepth, null, + FailCategory.FINAL_LOAD_CONTEXT, + "DHT error propagation from peer " + targetSeed.getName(), -1); + } + } catch (final Exception e) { + Network.log.warn("DHT: Failed to process rejected error URL hash '" + errorHash + "': " + e.getMessage()); + } + } + } + } + EventChannel.channels(EventChannel.DHTSEND).addMessage( new RSSMessage( "Sent " + uhs.length + " URLs to peer " + targetSeed.getName()+ "/[" + targetSeed.hash + "]", -- cgit v1.2.3