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

[dgellman]

Summary

The processor previously retried blob NotFound responses indefinitely. process.Subscribe returned nil from its handler without acking, so NATS redelivered the same orphaned message forever — generating hundreds of thousands of GetObject errors per processor over weeks for blobs that no longer exist.

After maxBlobNotFoundRetries (3) consecutive NotFound responses for a message, the message is now acked to stop redelivery. Other read errors retain their previous redelivery behavior.

How retries are counted

When the underlying transport exposes a delivery count, that is preferred so the threshold is shared across replicas and survives processor restarts:

• NATS JetStream: uses Metadata().NumDelivered via gocloud.dev/pubsub.Message.As(&jetstream.Msg).
• Other backends (incl. mempubsub in tests): falls back to a per-key in-process counter.

A Prometheus counter guac_processor_blob_notfound_dropped_total (label processor_id) is registered for alerting on orphan accumulation, and a structured warn log is emitted on each drop with the blob key and attempt count.

What this does not address

Why orphan events get published in the first place (likely TTL mismatch between blob lifecycle and NATS retention, or out-of-band cleanup). That belongs in a separate investigation.

Test plan

Run on this branch:

• go build ./... clean (exit 0)
• go vet ./... clean (exit 0)
• go test -race -timeout=60s ./... -count=1 — 79 packages OK, 0 fail, 0 data races. Note: at the Makefile's default -timeout=30s under race-detector parallelism, the unrelated pkg/ingestor/parser/common/scanner package occasionally tips over (it runs ~25–30 s in isolation, with the same behaviour on main); raising the timeout to 60 s clears it.
• golangci-lint run ./pkg/handler/processor/process/... — 0 issues
• New unit tests pass: TestNotFoundTracker (per-key increment/clear isolation) and TestDeliveryCountSource (label values). Existing Test_ProcessSubscribe continues to pass unchanged.

[stale]

This pull request has been automatically marked as stale because it has not had recent activity (60 days of inactivity).
It will be closed in 30 days if no further activity occurs.
Thank you for your contribution!

[gaganhr94]

@dgellman Can you please rebase the branch with main

[dgellman]

Will work on it a bit later today - this one appears to have a confict that needs to be resolved now.

[dgellman]

rebased but the failing tests don't seem to be me =/ can we try re-running?

[mihaimaruseac]

The failing test is not blocking, it's just flaky.

[gaganhr94]

Reviewed the blob-NotFound retry/drop logic. Nothing here blocks merge — one item is worth fixing, one is a minor note.

1. droppedCounter and the AddCounter fallback emit different label values (process.go, drop branch)

registerProcessorMetrics calls RegisterCounter(ctx, blobNotFoundDroppedCounter, "processor_id"), and prometheusCollector.RegisterCounter returns counterVec.WithLabelValues(labels...) — it passes the label names through as the label values. So droppedCounter.Inc() is safe (no panic), but it always increments the series processor_id="processor_id", whereas the fallback right beside it does AddCounter(ctx, blobNotFoundDroppedCounter, 1, uuidString) and records the real UUID. Same event, two different series depending on whether registration happened to succeed.

That WithLabelValues quirk is pre-existing and repo-wide, not introduced by this PR; the divergence between the two branches is. Simplest fix is to drop the registered-counter path and always go through AddCounter with uuidString:

if err := metricsCollector.AddCounter(ctx, blobNotFoundDroppedCounter, 1, uuidString); err != nil {
    childLogger.Debugf("[processor: %s] AddCounter blob-notfound: %v", uuidString, err)
}

2. notFoundTracker is keyed by blobStoreKey, not by message — minor

Two distinct pubsub messages referencing the same missing blob share one attempt counter, so the second can hit the cap earlier than maxBlobNotFoundRetries deliveries of its own. Low impact in practice (both messages point at the same absent blob, and the successful-read tracker.clear covers the case where the blob later lands), but a short comment on the field noting the key is per-blob rather than per-message would save the next reader the trip.

For the record, I also checked the ack semantics on the sub-cap NotFound path: returning nil without Ack/Nack is correct here — subscriber.GetDataFromSubscriber ignores a nil return and never acks, so the message is redelivered and the attempt counter does advance to the cap. No issue.

[gaganhr94]

Reviewed. See if these comments can be fixed @dgellman