Configurable option to skip ads with 500 error on first entries block

A new boolean config option `Ingest.Skip500EntriesError` allows the indexer to skip advertisements when the publisher returns a 500 response with "failed to sync first entry". This value is run-time reloadable.

The purpose of this is to skip ads that to publisher will never deliver content for. There is an issue with providers where they will not return entries from some advertisements, and permanently return a 500 result when trying to sync the entries.

Author: gammazero

command/daemon.go

@@ -612,6 +612,7 @@ func reloadConfig(cfgPath string, ingester *ingest.Ingester, reg *registry.Regis
 
 	if ingester != nil {
 		ingester.RunWorkers(cfg.Ingest.IngestWorkerCount)
+		ingester.Skip500EntriesError(cfg.Ingest.Skip500EntriesError)
 	}
 
 	err = setLoggingConfig(cfg.Logging)

config/ingest.go

@@ -52,8 +52,9 @@ type Ingest struct {
 	// IngestWorkerCount sets how many ingest worker goroutines to spawn. This
 	// controls how many concurrent ingest from different providers we can handle.
 	IngestWorkerCount int
-	// MaxAsyncConcurrency sets the maximum number of concurrent asynchrouous syncs
-	// (started by announce messages). Set -1 for unlimited, 0 for default.
+	// MaxAsyncConcurrency sets the maximum number of concurrent asynchrouous
+	// syncs (started by announce messages). Set -1 for unlimited, 0 for
+	// default. This value is reloadable.
 	MaxAsyncConcurrency int
 	// MinimumKeyLengt causes any multihash, that has a digest length less than
 	// this, to be ignored.
@@ -69,6 +70,10 @@ type Ingest struct {
 	// the announce so that other indexers can also receive it. This is always
 	// false if configured to use an assigner.
 	ResendDirectAnnounce bool
+	// Skip500EntriesError, when true, skips advertisements for which the
+	// publisher returns a 500 status code and an error message "failed to sync
+	// first entry". This value is reloadable.
+	Skip500EntriesError bool
 	// SyncSegmentDepthLimit is the depth limit of a single sync in a series of
 	// calls that collectively sync advertisements or their entries. The value
 	// -1 disables the segmentation where the sync will be done in a single call

doc/config.md

@@ -13,5 +13,6 @@ The storetheindex daemon can reload some portions of its config without restarti
 - `Indexer.ConfigCheckInterval`
 - `Indexer.ShutdownTimeout`
 - `Ingest.IngestWorkerCount`
+- `Ingest.Skip500EntriesError`
 - `Logging`
 - `Peering`

internal/ingest/ingest.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io/fs"
 	"path"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -166,6 +167,8 @@ type Ingester struct {
 
 	// Ingest rates
 	ingestRates *rate.Map
+
+	skip500EntsErr atomic.Bool
 }
 
 // NewIngester creates a new Ingester that uses a dagsync Subscriber to handle
@@ -208,6 +211,8 @@ func NewIngester(cfg config.Ingest, h host.Host, idxr indexer.Interface, reg *re
 		return nil, err
 	}
 
+	ing.skip500EntsErr.Store(cfg.Skip500EntriesError)
+
 	// Create and start subscriber. This also registers the storage hook to
 	// index data as it is received.
 	sub, err := dagsync.NewSubscriber(h, ing.dsTmp, ing.lsys, cfg.PubSubTopic,
@@ -258,6 +263,10 @@ func (ing *Ingester) MultihashesFromMirror() uint64 {
 	return ing.mhsFromMirror.Load()
 }
 
+func (ing *Ingester) Skip500EntriesError(skip bool) {
+	ing.skip500EntsErr.Store(skip)
+}
+
 func (ing *Ingester) generalDagsyncBlockHook(_ peer.ID, c cid.Cid, actions dagsync.SegmentSyncActions) {
 	// The only kind of block we should get by loading CIDs here should be
 	// Advertisement.
@@ -1114,6 +1123,8 @@ func (ing *Ingester) ingestWorkerLogic(ctx context.Context, provider peer.ID, as
 		}
 	}
 
+	skip500EntsErr := ing.skip500EntsErr.Load()
+
 	total := len(assignment.adInfos)
 	log.Infow("Running worker on ad stack", "headAdCid", headAdCid, "numAdsToProcess", total)
 	var count int
@@ -1181,6 +1192,12 @@ func (ing *Ingester) ingestWorkerLogic(ctx context.Context, provider peer.ID, as
 					log.Errorw("Skipping ad because of a permanent error", "adCid", ai.cid, "err", err, "errKind", adIngestErr.state)
 					stats.Record(context.Background(), metrics.AdIngestSkippedCount.M(1))
 					err = nil
+				case adIngestSyncEntriesErr:
+					if skip500EntsErr && strings.Contains(err.Error(), "failed to sync first entry") && strings.Contains(err.Error(), ": 500") {
+						log.Errorw("Skipping ad because of a permanent 500 error", "adCid", ai.cid, "err", err, "errKind", adIngestErr.state)
+						stats.Record(context.Background(), metrics.AdIngestSkippedCount.M(1))
+						err = nil
+					}
 				}
 				stats.RecordWithOptions(context.Background(),
 					stats.WithMeasurements(metrics.AdIngestErrorCount.M(1)),