fix(processor): dead-letter pubsub messages whose blob is persistently missing

pkg/handler/processor/process/process.go

@@ -24,7 +24,12 @@ import (
 	"io"
 	"path/filepath"
 	"strings"
+	"sync"
 
+	"gocloud.dev/gcerrors"
+	"gocloud.dev/pubsub"
+
+	"github.com/nats-io/nats.go/jetstream"
 	"go.uber.org/zap"
 
 	uuid "github.com/gofrs/uuid"
@@ -44,16 +49,31 @@ import (
 	"github.com/guacsec/guac/pkg/handler/processor/scorecard"
 	"github.com/guacsec/guac/pkg/handler/processor/spdx"
 	"github.com/guacsec/guac/pkg/logging"
+	"github.com/guacsec/guac/pkg/metrics"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/klauspost/compress/zstd"
-	"gocloud.dev/pubsub"
 )
 
 var (
 	documentProcessors = map[processor.DocumentType]processor.DocumentProcessor{}
 	json               = jsoniter.ConfigCompatibleWithStandardLibrary
 )
 
+// maxBlobNotFoundRetries caps how many times a single pubsub message will be
+// retried when the blob store reports the object is missing. After this many
+// consecutive NotFound responses for the same message, it is acked (dropped)
+// so it stops being redelivered indefinitely. Guards against orphaned events
+// whose underlying blob has been deleted or never landed.
+const maxBlobNotFoundRetries = 3
+
+// metricsCollectorName is the namespace used for processor metrics.
+const metricsCollectorName = "guac_processor"
+
+// blobNotFoundDroppedCounter is the metric name (sans namespace) for the
+// counter tracking pubsub messages dropped because their referenced blob is
+// persistently missing.
+const blobNotFoundDroppedCounter = "blob_notfound_dropped_total"
+
 func init() {
 	_ = RegisterDocumentProcessor(&ite6.ITE6Processor{}, processor.DocumentITE6Generic)
 	_ = RegisterDocumentProcessor(&ite6.ITE6Processor{}, processor.DocumentITE6SLSA)
@@ -80,6 +100,72 @@ func RegisterDocumentProcessor(p processor.DocumentProcessor, d processor.Docume
 	return nil
 }
 
+// deliveryCountSource returns a stable label describing where the delivery
+// count was sourced from, for log/metric clarity.
+func deliveryCountSource(fromBroker bool) string {
+	if fromBroker {
+		return "broker"
+	}
+	return "in_process"
+}
+
+// notFoundTracker counts consecutive NotFound responses per blob key for
+// pubsub backends that don't expose a native delivery count (e.g. mempubsub
+// in tests). NATS JetStream tracks NumDelivered server-side, so we prefer
+// that path when available.
+type notFoundTracker struct {
+	mu       sync.Mutex
+	attempts map[string]int
+}
+
+func newNotFoundTracker() *notFoundTracker {
+	return &notFoundTracker{attempts: make(map[string]int)}
+}
+
+func (t *notFoundTracker) increment(key string) int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	t.attempts[key]++
+	return t.attempts[key]
+}
+
+func (t *notFoundTracker) clear(key string) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	delete(t.attempts, key)
+}
+
+// deliveryAttempt returns the broker-reported delivery count for msg if the
+// underlying transport exposes one (currently NATS JetStream). The boolean
+// indicates whether a server-side count was available; callers fall back to
+// in-process tracking when it is not.
+func deliveryAttempt(msg *pubsub.Message) (int, bool) {
+	var jsMsg jetstream.Msg
+	if !msg.As(&jsMsg) {
+		return 0, false
+	}
+	md, err := jsMsg.Metadata()
+	if err != nil || md == nil {
+		return 0, false
+	}
+	return int(md.NumDelivered), true
+}
+
+// registerProcessorMetrics registers (or recovers an existing) counter for
+// dropped-message accounting. Returns nil for the counter if registration
+// fails so the caller can degrade gracefully without aborting ingest.
+func registerProcessorMetrics(ctx context.Context, logger *zap.SugaredLogger) metrics.Counter {
+	collector := metrics.FromContext(ctx, metricsCollectorName)
+	counter, err := collector.RegisterCounter(ctx, blobNotFoundDroppedCounter, "processor_id")
+	if err != nil {
+		// Most commonly this fires in tests where the counter was registered
+		// by a prior Subscribe; the metric is still usable via AddCounter.
+		logger.Debugf("blob-notfound counter registration: %v", err)
+		return nil
+	}
+	return counter
+}
+
 // Subscribe receives the CD event and decodes the event to obtain the blob store key.
 // The key is used to retrieve the "document" from the blob store to be processed and ingested.
 func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobStore, emPubSub *emitter.EmitterPubSub) error {
@@ -98,6 +184,11 @@ func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobSt
 
 	retries := 0
 
+	// In-process fallback for backends without a native delivery count.
+	tracker := newNotFoundTracker()
+	droppedCounter := registerProcessorMetrics(ctx, logger)
+	metricsCollector := metrics.FromContext(ctx, metricsCollectorName)
+
 	// should still continue if there are errors since problem is with individual documents
 	processFunc := func(d *pubsub.Message) error {
 		retries++
@@ -116,10 +207,44 @@ func Subscribe(ctx context.Context, em collector.Emitter, blobStore *blob.BlobSt
 
 		documentBytes, err := blobStore.Read(ctx, blobStoreKey)
 		if err != nil {
+			if gcerrors.Code(err) == gcerrors.NotFound {
+				attempts, fromBroker := deliveryAttempt(d)
+				if !fromBroker {
+					attempts = tracker.increment(blobStoreKey)
+				}
+
+				childLogger.Errorw("failed read document to blob store",
+					"processor_id", uuidString,
+					"attempt", attempts,
+					"max_attempts", maxBlobNotFoundRetries,
+					"delivery_count_source", deliveryCountSource(fromBroker),
+					"error", err.Error(),
+				)
+
+				if attempts >= maxBlobNotFoundRetries {
+					childLogger.Warnw("blob persistently missing — dropping pubsub message to stop redelivery",
+						"processor_id", uuidString,
+						"attempts", attempts,
+						"blob_key", blobStoreKey,
+						"delivery_count_source", deliveryCountSource(fromBroker),
+					)
+					if droppedCounter != nil {
+						droppedCounter.Inc()
+					} else if err := metricsCollector.AddCounter(ctx, blobNotFoundDroppedCounter, 1, uuidString); err != nil {
+						childLogger.Debugf("[processor: %s] AddCounter blob-notfound: %v", uuidString, err)
+					}
+					tracker.clear(blobStoreKey)
+					d.Ack()
+				}
+				return nil
+			}
 			childLogger.Errorf("[processor: %s] failed read document to blob store: %v", uuidString, err)
 			return nil
 		}
 
+		// successful read — clear any prior NotFound bookkeeping for this key
+		tracker.clear(blobStoreKey)
+
 		doc := processor.Document{}
 		if err = json.Unmarshal(documentBytes, &doc); err != nil {
 			childLogger.Errorf("[processor: %s] failed unmarshal the document bytes: %v", uuidString, err)

pkg/handler/processor/process/process_test.go

@@ -19,12 +19,13 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"strings"
 	"testing"
 	"time"
 
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+
 	"github.com/guacsec/guac/internal/testing/dochelper"
 	nats_test "github.com/guacsec/guac/internal/testing/nats"
 	"github.com/guacsec/guac/internal/testing/simpledoc"
@@ -852,3 +853,35 @@ func testPublish(ctx context.Context, d *processor.Document, blobStore *blob.Blo
 	logger.Debugf("doc published: %+v", d.SourceInformation.Source)
 	return nil
 }
+
+func TestNotFoundTracker(t *testing.T) {
+	tracker := newNotFoundTracker()
+
+	if got := tracker.increment("a"); got != 1 {
+		t.Fatalf("first increment for 'a' = %d, want 1", got)
+	}
+	if got := tracker.increment("a"); got != 2 {
+		t.Fatalf("second increment for 'a' = %d, want 2", got)
+	}
+	if got := tracker.increment("b"); got != 1 {
+		t.Fatalf("first increment for 'b' = %d, want 1", got)
+	}
+
+	tracker.clear("a")
+	if got := tracker.increment("a"); got != 1 {
+		t.Fatalf("increment after clear for 'a' = %d, want 1", got)
+	}
+	// 'b' should be untouched by clearing 'a'
+	if got := tracker.increment("b"); got != 2 {
+		t.Fatalf("increment for 'b' after clearing 'a' = %d, want 2", got)
+	}
+}
+
+func TestDeliveryCountSource(t *testing.T) {
+	if got := deliveryCountSource(true); got != "broker" {
+		t.Errorf("deliveryCountSource(true) = %q, want %q", got, "broker")
+	}
+	if got := deliveryCountSource(false); got != "in_process" {
+		t.Errorf("deliveryCountSource(false) = %q, want %q", got, "in_process")
+	}
+}