chore: drop external payload not found errors

internal/msgqueue/rabbitmq/permanent_preack_test.go

@@ -7,6 +7,8 @@ import (
 
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/hatchet-dev/hatchet/pkg/repository"
 )
 
 func TestIsPermanentPreAckError_PgError22P02(t *testing.T) {
@@ -25,6 +27,22 @@ func TestIsPermanentPreAckError_StringFallback(t *testing.T) {
 	}
 }
 
+func TestIsPermanentPreAckError_ExternalPayloadNotFound(t *testing.T) {
+	err := fmt.Errorf("wrap: %w", &repository.ExternalPayloadNotFoundError{
+		Kind: repository.ExternalPayloadNotFoundKindIndexFile,
+		Key:  "index/2026-06-11/example.index",
+		Err:  errors.New("key not found"),
+	})
+
+	if !isPermanentPreAckError(err) {
+		t.Fatalf("expected true for external payload not found error")
+	}
+
+	if got := permanentPreAckErrorReason(err); got != "external_payload_not_found" {
+		t.Fatalf("expected external payload reason, got %q", got)
+	}
+}
+
 func TestIsPermanentPreAckError_OtherError(t *testing.T) {
 	err := errors.New("some transient error")
 	if isPermanentPreAckError(err) {

internal/msgqueue/rabbitmq/rabbitmq.go

@@ -25,6 +25,7 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/queueutils"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/random"
+	"github.com/hatchet-dev/hatchet/pkg/repository"
 	"github.com/hatchet-dev/hatchet/pkg/telemetry"
 )
 
@@ -832,11 +833,13 @@ func (t *MessageQueueImpl) subscribe(
 
 				if err := preAck(msg); err != nil {
 					if isPermanentPreAckError(err) {
-						t.l.Error().
+						event := t.l.Error().
 							Err(err).
 							Str("message_id", msg.ID).
 							Str("tenant_id", msg.TenantID.String()).
-							Int("num_payloads", len(msg.Payloads)).
+							Int("num_payloads", len(msg.Payloads))
+
+						addPermanentPreAckErrorFields(event, err).
 							Msg("dropping message due to permanent pre-ack error")
 
 						if ackErr := rabbitMsg.Ack(false); ackErr != nil {
@@ -912,28 +915,49 @@ func (t *MessageQueueImpl) subscribe(
 }
 
 func isPermanentPreAckError(err error) bool {
+	return permanentPreAckErrorReason(err) != ""
+}
+
+func permanentPreAckErrorReason(err error) string {
 	if err == nil {
-		return false
+		return ""
+	}
+
+	if errors.Is(err, repository.ErrExternalPayloadNotFound) {
+		return "external_payload_not_found"
 	}
 
 	var pgErr *pgconn.PgError
 	if errors.As(err, &pgErr) {
 		// invalid input syntax for type json / jsonb
 		if pgErr.Code == pgerrcode.InvalidTextRepresentation {
-			return true
+			return "invalid_json"
 		}
 	}
 
 	// Fallback: some error paths may lose pg error type info.
 	errStr := err.Error()
 	if strings.Contains(errStr, fmt.Sprintf("SQLSTATE %s", pgerrcode.InvalidTextRepresentation)) {
-		return true
+		return "invalid_json"
 	}
 	if strings.Contains(errStr, "invalid input syntax for type json") {
-		return true
+		return "invalid_json"
+	}
+
+	return ""
+}
+
+func addPermanentPreAckErrorFields(event *zerolog.Event, err error) *zerolog.Event {
+	event.Str("permanent_pre_ack_reason", permanentPreAckErrorReason(err))
+
+	var payloadErr *repository.ExternalPayloadNotFoundError
+	if errors.As(err, &payloadErr) {
+		event.
+			Str("external_payload_kind", string(payloadErr.Kind)).
+			Str("external_payload_key", payloadErr.Key)
 	}
 
-	return false
+	return event
 }
 
 // identity returns the same host/process unique string for the lifetime of

pkg/repository/payloadstore.go

@@ -89,6 +89,42 @@ type PayloadLocation string
 type ExternalPayloadLocationKey string
 type ExternalIndexFileLocationKey string
 
+var ErrExternalPayloadNotFound = errors.New("external payload not found")
+
+type ExternalPayloadNotFoundKind string
+
+const (
+	ExternalPayloadNotFoundKindIndexFile ExternalPayloadNotFoundKind = "index_file"
+)
+
+type ExternalPayloadNotFoundError struct {
+	Err  error
+	Kind ExternalPayloadNotFoundKind
+	Key  string
+}
+
+func (e *ExternalPayloadNotFoundError) Error() string {
+	msg := ErrExternalPayloadNotFound.Error()
+	if e.Kind != "" {
+		msg = fmt.Sprintf("%s: %s", msg, e.Kind)
+	}
+	if e.Key != "" {
+		msg = fmt.Sprintf("%s %s", msg, e.Key)
+	}
+	if e.Err != nil {
+		msg = fmt.Sprintf("%s: %v", msg, e.Err)
+	}
+	return msg
+}
+
+func (e *ExternalPayloadNotFoundError) Unwrap() error {
+	return e.Err
+}
+
+func (e *ExternalPayloadNotFoundError) Is(target error) bool {
+	return target == ErrExternalPayloadNotFound
+}
+
 type CreateIndexBlockOpts struct {
 	PartitionDate             PartitionDate
 	BlockLowerExternalIdBound uuid.UUID