chore: drop external payload not found errors

internal/msgqueue/rabbitmq/rabbitmq.go

@@ -4,18 +4,14 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"os"
 	"slices"
-	"strings"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
 	lru "github.com/hashicorp/golang-lru/v2"
-	"github.com/jackc/pgerrcode"
-	"github.com/jackc/pgx/v5/pgconn"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/rs/zerolog"
 	"go.opentelemetry.io/otel/attribute"
@@ -831,21 +827,6 @@ func (t *MessageQueueImpl) subscribe(
 				t.l.Debug().Msgf("(session: %d) got msg", session)
 
 				if err := preAck(msg); err != nil {
-					if isPermanentPreAckError(err) {
-						t.l.Error().
-							Err(err).
-							Str("message_id", msg.ID).
-							Str("tenant_id", msg.TenantID.String()).
-							Int("num_payloads", len(msg.Payloads)).
-							Msg("dropping message due to permanent pre-ack error")
-
-						if ackErr := rabbitMsg.Ack(false); ackErr != nil {
-							t.l.Error().Err(ackErr).Msg("error acknowledging message after permanent pre-ack error")
-						}
-
-						return
-					}
-
 					t.l.Error().Msgf("error in pre-ack on msg %s: %v", msg.ID, err)
 
 					// nack the message
@@ -911,31 +892,6 @@ func (t *MessageQueueImpl) subscribe(
 	return cleanup, nil
 }
 
-func isPermanentPreAckError(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	var pgErr *pgconn.PgError
-	if errors.As(err, &pgErr) {
-		// invalid input syntax for type json / jsonb
-		if pgErr.Code == pgerrcode.InvalidTextRepresentation {
-			return true
-		}
-	}
-
-	// 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
-	}
-	if strings.Contains(errStr, "invalid input syntax for type json") {
-		return true
-	}
-
-	return false
-}
-
 // identity returns the same host/process unique string for the lifetime of
 // this process so that subscriber reconnections reuse the same queue name.
 func identity() string {

internal/queueutils/permanent_errors.go

@@ -0,0 +1,43 @@
+package queueutils
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/hatchet-dev/hatchet/pkg/repository"
+)
+
+// PermanentConsumerErrorReason returns a stable reason for consumer errors that should be shed instead of retried.
+func PermanentConsumerErrorReason(err error) string {
+	if err == nil {
+		return ""
+	}
+
+	if errors.Is(err, repository.ErrExternalPayloadNotFound) {
+		return "external_payload_not_found"
+	}
+
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.InvalidTextRepresentation {
+		return "invalid_json"
+	}
+
+	errStr := err.Error()
+	if strings.Contains(errStr, fmt.Sprintf("SQLSTATE %s", pgerrcode.InvalidTextRepresentation)) {
+		return "invalid_json"
+	}
+	if strings.Contains(errStr, "invalid input syntax for type json") {
+		return "invalid_json"
+	}
+
+	return ""
+}
+
+// IsPermanentConsumerError returns true when a consumer can drop the offending payload from a buffered batch.
+func IsPermanentConsumerError(err error) bool {
+	return PermanentConsumerErrorReason(err) != ""
+}

internal/queueutils/permanent_errors_test.go

@@ -0,0 +1,51 @@
+package queueutils
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/jackc/pgerrcode"
+	"github.com/jackc/pgx/v5/pgconn"
+
+	"github.com/hatchet-dev/hatchet/pkg/repository"
+)
+
+func TestIsPermanentConsumerError_PgError22P02(t *testing.T) {
+	pgErr := &pgconn.PgError{Code: pgerrcode.InvalidTextRepresentation, Message: "invalid input syntax for type json"}
+	wrapped := fmt.Errorf("wrap: %w", pgErr)
+
+	if !IsPermanentConsumerError(wrapped) {
+		t.Fatalf("expected true for wrapped pg error 22P02")
+	}
+}
+
+func TestIsPermanentConsumerError_StringFallback(t *testing.T) {
+	err := errors.New("ERROR: invalid input syntax for type json (SQLSTATE 22P02)")
+	if !IsPermanentConsumerError(err) {
+		t.Fatalf("expected true for sqlstate 22P02 string fallback")
+	}
+}
+
+func TestIsPermanentConsumerError_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 !IsPermanentConsumerError(err) {
+		t.Fatalf("expected true for external payload not found error")
+	}
+
+	if got := PermanentConsumerErrorReason(err); got != "external_payload_not_found" {
+		t.Fatalf("expected external payload reason, got %q", got)
+	}
+}
+
+func TestIsPermanentConsumerError_OtherError(t *testing.T) {
+	err := errors.New("some transient error")
+	if IsPermanentConsumerError(err) {
+		t.Fatalf("expected false for non-permanent error")
+	}
+}