From a038506d80d08eff85ad9d078e9999331d09fc73 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 16 Jun 2026 12:05:14 +0700 Subject: [PATCH 1/4] fix(destregistry): treat event format errors as failed deliveries, not DLQ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Format/key-template failure (e.g. an S3 key_template referencing a field absent from the event) returned a nil delivery, which the registry turned into a nil attempt and the deliverymq handler classified as a PreDeliveryError → nack → Pub/Sub DLQ. The failure was never logged, invisible to the customer, and paged us instead of surfacing as an actionable delivery error. Add destregistry.NewFormatErrorDelivery, returning a non-nil failed Delivery plus an ErrDestinationPublishAttempt, so the registry records a failed attempt, acks the message, and retries via the scheduler. The customer-facing response is a generic message; the raw Go error stays on the error for logs/telemetry and is not persisted on the attempt. Apply it across all providers with a Format step: s3, sqs, azure_servicebus, gcp_pubsub, webhook, webhook_standard (previously `return nil, err`) and kinesis, kafka (previously nil-delivery ErrDestinationPublishAttempt). Co-Authored-By: Claude Fable 5 --- internal/destregistry/error.go | 21 ++++++++ .../format_error_delivery_test.go | 52 +++++++++++++++++++ .../destawskinesis/destawskinesis.go | 9 +--- .../providers/destawss3/destawss3.go | 2 +- .../destawss3/destawss3_format_test.go | 36 +++++++++++++ .../providers/destawssqs/destawssqs.go | 2 +- .../destazureservicebus.go | 2 +- .../providers/destgcppubsub/destgcppubsub.go | 2 +- .../providers/destkafka/destkafka.go | 7 +-- .../providers/destwebhook/destwebhook.go | 2 +- .../destwebhookstandard.go | 2 +- 11 files changed, 117 insertions(+), 20 deletions(-) create mode 100644 internal/destregistry/format_error_delivery_test.go diff --git a/internal/destregistry/error.go b/internal/destregistry/error.go index c824d9b26..115b115c3 100644 --- a/internal/destregistry/error.go +++ b/internal/destregistry/error.go @@ -39,6 +39,27 @@ func NewErrDestinationPublishAttempt(err error, provider string, data map[string return &ErrDestinationPublishAttempt{Err: err, Provider: provider, Data: data} } +// NewFormatErrorDelivery returns the (*Delivery, error) a publisher should return when +// formatting an event fails before it can be sent (e.g. an invalid key/partition +// template or an unparseable payload). It records a failed attempt so the failure +// is visible to the customer and the message is acked, instead of nacking into the DLQ. +// +// message is the customer-facing string persisted on the attempt (ResponseData); +// when empty a generic default is used. The raw err is carried only in the returned +// error (for logs/telemetry) and is not persisted on the attempt. +func NewFormatErrorDelivery(provider, message string, err error) (*Delivery, error) { + if message == "" { + message = "could not format event for delivery" + } + return &Delivery{ + Status: "failed", + Code: "ERR", + Response: map[string]interface{}{"error": message}, + }, NewErrDestinationPublishAttempt(err, provider, map[string]interface{}{ + "error": "format_failed", + }) +} + // NewErrPublishCanceled creates an error for when publish is canceled (e.g., service shutdown). // This should return nil Delivery to trigger nack → requeue for another instance. // See: https://github.com/hookdeck/outpost/issues/571 diff --git a/internal/destregistry/format_error_delivery_test.go b/internal/destregistry/format_error_delivery_test.go new file mode 100644 index 000000000..ebfd155ef --- /dev/null +++ b/internal/destregistry/format_error_delivery_test.go @@ -0,0 +1,52 @@ +package destregistry_test + +import ( + "errors" + "testing" + + "github.com/hookdeck/outpost/internal/destregistry" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestNewFormatErrorDelivery(t *testing.T) { + t.Parallel() + + rawErr := errors.New("failed to evaluate key template: invalid type") + + t.Run("returns a failed delivery and a publish-attempt error", func(t *testing.T) { + t.Parallel() + + delivery, err := destregistry.NewFormatErrorDelivery("aws_s3", "", rawErr) + + // A non-nil failed delivery means the registry records an attempt and acks + // the message instead of nacking it into the DLQ. + require.NotNil(t, delivery) + assert.Equal(t, "failed", delivery.Status) + assert.Equal(t, "ERR", delivery.Code) + + var pubErr *destregistry.ErrDestinationPublishAttempt + require.ErrorAs(t, err, &pubErr) + assert.Equal(t, "aws_s3", pubErr.Provider) + assert.Equal(t, "format_failed", pubErr.Data["error"]) + // Raw Go error is carried on the error for logs/telemetry... + assert.Equal(t, rawErr, pubErr.Err) + }) + + t.Run("uses a generic message when none is given", func(t *testing.T) { + t.Parallel() + + delivery, _ := destregistry.NewFormatErrorDelivery("aws_s3", "", rawErr) + + // ...but is NOT persisted on the attempt; the customer-facing response is generic. + assert.Equal(t, "could not format event for delivery", delivery.Response["error"]) + assert.NotContains(t, delivery.Response["error"], "key template") + }) + + t.Run("uses the provided message when given", func(t *testing.T) { + t.Parallel() + + delivery, _ := destregistry.NewFormatErrorDelivery("aws_s3", "could not build S3 object key", rawErr) + assert.Equal(t, "could not build S3 object key", delivery.Response["error"]) + }) +} diff --git a/internal/destregistry/providers/destawskinesis/destawskinesis.go b/internal/destregistry/providers/destawskinesis/destawskinesis.go index 7d5ec3e5a..fc4ef1531 100644 --- a/internal/destregistry/providers/destawskinesis/destawskinesis.go +++ b/internal/destregistry/providers/destawskinesis/destawskinesis.go @@ -258,14 +258,7 @@ func (p *AWSKinesisPublisher) Publish(ctx context.Context, event *models.Event) // Format the event into a PutRecordInput input, err := p.Format(ctx, event) if err != nil { - return nil, destregistry.NewErrDestinationPublishAttempt( - err, - "aws_kinesis", - map[string]interface{}{ - "error": "format_failed", - "message": err.Error(), - }, - ) + return destregistry.NewFormatErrorDelivery("aws_kinesis", "", err) } // Send the record to Kinesis diff --git a/internal/destregistry/providers/destawss3/destawss3.go b/internal/destregistry/providers/destawss3/destawss3.go index 1d7d80bce..8ed82aa0d 100644 --- a/internal/destregistry/providers/destawss3/destawss3.go +++ b/internal/destregistry/providers/destawss3/destawss3.go @@ -304,7 +304,7 @@ func (p *AWSS3Publisher) Publish(ctx context.Context, event *models.Event) (*des input, err := p.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("aws_s3", "", err) } _, err = p.client.PutObject(ctx, input) diff --git a/internal/destregistry/providers/destawss3/destawss3_format_test.go b/internal/destregistry/providers/destawss3/destawss3_format_test.go index e206c5c59..c197c784e 100644 --- a/internal/destregistry/providers/destawss3/destawss3_format_test.go +++ b/internal/destregistry/providers/destawss3/destawss3_format_test.go @@ -198,6 +198,42 @@ func TestAWSS3Publisher_Format_NilResult(t *testing.T) { assert.Contains(t, err.Error(), "nil result") } +// TestAWSS3Publisher_Publish_FormatError reproduces the production incident where a +// key_template references a field absent from the event (here metadata.operationId on +// an event with no such metadata). The key cannot be built, so Format fails. Publish +// must surface this as a failed delivery + ErrDestinationPublishAttempt (so the registry +// records an attempt and acks) rather than a nil delivery (which nacks into the DLQ). +func TestAWSS3Publisher_Publish_FormatError(t *testing.T) { + event := models.Event{ + ID: "event-123", + Time: time.Now(), + Data: json.RawMessage(`{}`), + } + + // Mirrors the incident template: join over a missing metadata field yields nil, + // which join rejects. Reaches Publish's Format step before any S3 client call, + // so a nil client is safe here. + template := `join('/', ['prefix', metadata.operationId])` + publisher := destawss3.NewAWSS3Publisher( + destregistry.NewBasePublisher(), + nil, + "my-bucket", + template, + "STANDARD", + ) + + delivery, err := publisher.Publish(context.Background(), &event) + + require.NotNil(t, delivery, "format failure must yield a non-nil delivery so the message is acked, not DLQ'd") + assert.Equal(t, "failed", delivery.Status) + assert.Equal(t, "ERR", delivery.Code) + + var pubErr *destregistry.ErrDestinationPublishAttempt + require.ErrorAs(t, err, &pubErr) + assert.Equal(t, "aws_s3", pubErr.Provider) + assert.Equal(t, "format_failed", pubErr.Data["error"]) +} + func TestAWSS3Publisher_Format_EmptyResult(t *testing.T) { event := models.Event{ ID: "event-123", diff --git a/internal/destregistry/providers/destawssqs/destawssqs.go b/internal/destregistry/providers/destawssqs/destawssqs.go index f605e3b2c..dd76ed200 100644 --- a/internal/destregistry/providers/destawssqs/destawssqs.go +++ b/internal/destregistry/providers/destawssqs/destawssqs.go @@ -163,7 +163,7 @@ func (p *AWSSQSPublisher) Publish(ctx context.Context, event *models.Event) (*de msg, err := p.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("aws_sqs", "", err) } if _, err = p.client.SendMessage(ctx, msg); err != nil { diff --git a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go b/internal/destregistry/providers/destazureservicebus/destazureservicebus.go index 65252bfd5..2c8d7b717 100644 --- a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go +++ b/internal/destregistry/providers/destazureservicebus/destazureservicebus.go @@ -149,7 +149,7 @@ func (p *AzureServiceBusPublisher) Publish(ctx context.Context, event *models.Ev message, err := p.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("azure_servicebus", "", err) } sender, err := p.ensureSender() diff --git a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go b/internal/destregistry/providers/destgcppubsub/destgcppubsub.go index f522411b4..8bba01bb9 100644 --- a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go +++ b/internal/destregistry/providers/destgcppubsub/destgcppubsub.go @@ -184,7 +184,7 @@ func (pub *GCPPubSubPublisher) Publish(ctx context.Context, event *models.Event) // Format the message msg, err := pub.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("gcp_pubsub", "", err) } // Publish the message diff --git a/internal/destregistry/providers/destkafka/destkafka.go b/internal/destregistry/providers/destkafka/destkafka.go index a75cdc637..2bd72c5e8 100644 --- a/internal/destregistry/providers/destkafka/destkafka.go +++ b/internal/destregistry/providers/destkafka/destkafka.go @@ -222,12 +222,7 @@ func (p *KafkaPublisher) Publish(ctx context.Context, event *models.Event) (*des // Build parsed payload for partition key JMESPath evaluation dataMap, err := event.ParsedData() if err != nil { - return nil, destregistry.NewErrDestinationPublishAttempt( - err, "kafka", map[string]interface{}{ - "error": "format_failed", - "message": err.Error(), - }, - ) + return destregistry.NewFormatErrorDelivery("kafka", "", err) } if dataMap == nil { dataMap = make(map[string]interface{}) diff --git a/internal/destregistry/providers/destwebhook/destwebhook.go b/internal/destregistry/providers/destwebhook/destwebhook.go index 0162a2e23..b1f77b9ee 100644 --- a/internal/destregistry/providers/destwebhook/destwebhook.go +++ b/internal/destregistry/providers/destwebhook/destwebhook.go @@ -658,7 +658,7 @@ func (p *WebhookPublisher) Publish(ctx context.Context, event *models.Event) (*d httpReq, err := p.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("webhook", "", err) } result := ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook") diff --git a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go b/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go index fc4f098b1..3e92ddbbe 100644 --- a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go +++ b/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go @@ -555,7 +555,7 @@ func (p *StandardWebhookPublisher) Publish(ctx context.Context, event *models.Ev httpReq, err := p.Format(ctx, event) if err != nil { - return nil, err + return destregistry.NewFormatErrorDelivery("webhook_standard", "", err) } result := destwebhook.ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook_standard") From 47b34bb75858af8e9eff5f410d520bd9708787e2 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 16 Jun 2026 12:38:02 +0700 Subject: [PATCH 2/4] test(e2e): regression for format error delivered as failed attempt, not DLQ MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Standalone e2e test reproducing the production incident: an aws_s3 destination whose key_template references a field missing from the event. Asserts the fixed behavior end to end — nothing is written to S3, each delivery is recorded as a failed attempt carrying the format error, and retries run on the normal schedule and exhaust their budget rather than being nacked/dead-lettered. Verified as a real guard: reverting the destawss3 fix makes this test fail (0 attempts logged, message dead-lettered) instead of recording 3 attempts. Co-Authored-By: Claude Fable 5 --- cmd/e2e/regression_format_error_test.go | 184 ++++++++++++++++++++++++ 1 file changed, 184 insertions(+) create mode 100644 cmd/e2e/regression_format_error_test.go diff --git a/cmd/e2e/regression_format_error_test.go b/cmd/e2e/regression_format_error_test.go new file mode 100644 index 000000000..98542be33 --- /dev/null +++ b/cmd/e2e/regression_format_error_test.go @@ -0,0 +1,184 @@ +package e2e_test + +import ( + "context" + "fmt" + "log" + "net/http" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awsconfig "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/gin-gonic/gin" + "github.com/hookdeck/outpost/cmd/e2e/configs" + "github.com/hookdeck/outpost/internal/app" + "github.com/hookdeck/outpost/internal/config" + "github.com/hookdeck/outpost/internal/util/testinfra" + "github.com/stretchr/testify/require" +) + +// TestE2E_Regression_FormatErrorIsDeliveredAttempt is a standalone regression test +// for the production incident where an aws_s3 destination's key_template could not be +// evaluated for a given event (the template referenced metadata.operationId, which the +// event lacked). Before the fix, the per-event formatting failure returned a nil +// delivery, which the pipeline treated as a system error: it was nacked, never logged +// as an attempt, retried blindly by the message queue, and ultimately dead-lettered — +// paging us instead of surfacing as a normal delivery failure. +// +// aws_s3 is the only destination type that can force this: its key_template is JMESPath +// validated for syntax only at creation, so a template that is valid at creation can +// still fail at delivery for an event missing a referenced field. (Kafka/Kinesis use the +// same per-event template mechanism but swallow eval errors and fall back to event.ID; +// the other providers have no per-event formatting step that can fail.) +// +// This test asserts the FIXED behavior, end to end: +// 1. nothing is ever written to S3 (the failure is pre-delivery) +// 2. each attempt is RECORDED as a failed delivery carrying the format error +// 3. it retries on the normal schedule and EXHAUSTS its budget, then stops — +// it is never requeued/dead-lettered (which would log zero attempts and loop) +func TestE2E_Regression_FormatErrorIsDeliveredAttempt(t *testing.T) { + t.Parallel() + if testing.Short() { + t.Skip("skipping e2e test") + } + + testinfraCleanup := testinfra.Start(t) + defer testinfraCleanup() + gin.SetMode(gin.TestMode) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // LocalStack S3: the bucket doubles as the "was anything actually delivered?" sink. + endpoint := testinfra.EnsureLocalStack() + awsCfg, err := awsconfig.LoadDefaultConfig(ctx, + awsconfig.WithRegion("us-east-1"), + awsconfig.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")), + ) + require.NoError(t, err) + s3Client := s3.NewFromConfig(awsCfg, func(o *s3.Options) { + o.UsePathStyle = true // required for LocalStack + o.BaseEndpoint = aws.String(endpoint) + }) + bucket := fmt.Sprintf("regr-format-%d", time.Now().UnixNano()) + _, err = s3Client.CreateBucket(ctx, &s3.CreateBucketInput{Bucket: aws.String(bucket)}) + require.NoError(t, err) + + // Isolated outpost instance with a short, bounded retry schedule so the budget + // exhausts quickly: schedule length 2 => 2 retries => 3 total attempts. + cfg := configs.Basic(t, configs.BasicOpts{LogStorage: configs.LogStorageTypeClickHouse}) + cfg.RetrySchedule = []int{1, 1} + cfg.RetryPollBackoffMs = 50 + cfg.LogBatchThresholdSeconds = 0 // immediate flush so /attempts is reliable + require.NoError(t, cfg.Validate(config.Flags{})) + configs.ApplyMigrations(t, &cfg) + + appDone := make(chan struct{}) + go func() { + defer close(appDone) + application := app.New(&cfg) + if err := application.Run(ctx); err != nil { + log.Println("Application stopped:", err) + } + }() + defer func() { + cancel() + <-appDone + }() + + waitForHealthy(t, cfg.APIPort, 5*time.Second) + + client := newRegressionHTTPClient(cfg.APIKey) + apiURL := fmt.Sprintf("http://localhost:%d/api/v1", cfg.APIPort) + + tenantID := fmt.Sprintf("tenant_format_%d", time.Now().UnixNano()) + destinationID := fmt.Sprintf("dest_format_%d", time.Now().UnixNano()) + eventID := fmt.Sprintf("evt_format_%d", time.Now().UnixNano()) + + // Create tenant. + status := client.doJSON(t, http.MethodPut, apiURL+"/tenants/"+tenantID, nil, nil) + require.Equal(t, 201, status, "failed to create tenant") + + // Create an aws_s3 destination whose key_template is valid at creation but cannot be + // evaluated for an event lacking metadata.operationId (join over a nil value fails). + status = client.doJSON(t, http.MethodPost, apiURL+"/tenants/"+tenantID+"/destinations", map[string]any{ + "id": destinationID, + "type": "aws_s3", + "topics": "*", + "config": map[string]any{ + "bucket": bucket, + "region": "us-east-1", + "endpoint": endpoint, + "storage_class": "STANDARD", + "key_template": `join('/', ['prefix', metadata.operationId])`, + }, + "credentials": map[string]any{ + "key": "test", + "secret": "test", + }, + }, nil) + require.Equal(t, 201, status, "failed to create aws_s3 destination") + + // Publish a retry-eligible event WITHOUT metadata.operationId -> formatting fails at delivery. + status = client.doJSON(t, http.MethodPost, apiURL+"/publish", map[string]any{ + "id": eventID, + "tenant_id": tenantID, + "topic": "user.created", + "eligible_for_retry": true, + "metadata": map[string]any{"foo": "bar"}, + "data": map[string]any{"hello": "world"}, + }, nil) + require.Equal(t, 202, status, "failed to publish event") + + attemptsURL := apiURL + "/attempts?tenant_id=" + tenantID + "&event_id=" + eventID + "&dir=asc&include=response_data" + pollAttempts := func(t *testing.T, minCount int, timeout time.Duration) []map[string]any { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + var resp struct { + Models []map[string]any `json:"models"` + } + s := client.doJSON(t, http.MethodGet, attemptsURL, nil, &resp) + if s == http.StatusOK && len(resp.Models) >= minCount { + return resp.Models + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d attempts", minCount) + return nil + } + + // (2) + (3): retries run on the normal schedule and produce recorded failed attempts. + // 1 initial + 2 scheduled retries = 3 attempts. If the fix regressed, the format error + // would nack/dead-letter the message and ZERO attempts would be logged -> this times out. + attempts := pollAttempts(t, 3, 15*time.Second) + for i, atm := range attempts { + require.Equal(t, "failed", atm["status"], "attempt %d should be a failed delivery", i+1) + // (2) the attempt carries the format error as a normal, customer-facing delivery error. + if rd, ok := atm["response_data"].(map[string]any); ok { + require.Equal(t, "could not format event for delivery", rd["error"], + "attempt %d should record the format error", i+1) + } else { + t.Fatalf("attempt %d missing response_data", i+1) + } + } + + // Budget exhausted: after the schedule completes, no further attempts appear. A + // dead-letter/requeue loop would keep producing attempts (or none at all) instead + // of stopping cleanly at the retry budget. + time.Sleep(2 * time.Second) + var finalResp struct { + Models []map[string]any `json:"models"` + } + client.doJSON(t, http.MethodGet, attemptsURL, nil, &finalResp) + require.Len(t, finalResp.Models, 3, + "should have exactly 3 attempts (1 initial + 2 retries) then stop — not requeue into a DLQ") + + // (1) nothing was ever written to S3 — the failure happened before any PutObject. + listed, err := s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{Bucket: aws.String(bucket)}) + require.NoError(t, err) + require.Empty(t, listed.Contents, "no object should have been written to S3") +} From 55dc0e7c7265559a2eb5076a950dcb0cf93e4d13 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 16 Jun 2026 21:02:12 +0700 Subject: [PATCH 3/4] refactor(destregistry): rename NewFormatErrorDelivery to NewFormatError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The helper returns the (*Delivery, error) pair a publisher returns on a format failure, not just a delivery — name it accordingly. Behavior unchanged. Co-Authored-By: Claude Fable 5 --- internal/destregistry/error.go | 4 ++-- ...format_error_delivery_test.go => format_error_test.go} | 8 ++++---- .../providers/destawskinesis/destawskinesis.go | 2 +- internal/destregistry/providers/destawss3/destawss3.go | 2 +- internal/destregistry/providers/destawssqs/destawssqs.go | 2 +- .../providers/destazureservicebus/destazureservicebus.go | 2 +- .../destregistry/providers/destgcppubsub/destgcppubsub.go | 2 +- internal/destregistry/providers/destkafka/destkafka.go | 2 +- .../destregistry/providers/destwebhook/destwebhook.go | 2 +- .../providers/destwebhookstandard/destwebhookstandard.go | 2 +- 10 files changed, 14 insertions(+), 14 deletions(-) rename internal/destregistry/{format_error_delivery_test.go => format_error_test.go} (82%) diff --git a/internal/destregistry/error.go b/internal/destregistry/error.go index 115b115c3..f07f576d1 100644 --- a/internal/destregistry/error.go +++ b/internal/destregistry/error.go @@ -39,7 +39,7 @@ func NewErrDestinationPublishAttempt(err error, provider string, data map[string return &ErrDestinationPublishAttempt{Err: err, Provider: provider, Data: data} } -// NewFormatErrorDelivery returns the (*Delivery, error) a publisher should return when +// NewFormatError returns the (*Delivery, error) a publisher should return when // formatting an event fails before it can be sent (e.g. an invalid key/partition // template or an unparseable payload). It records a failed attempt so the failure // is visible to the customer and the message is acked, instead of nacking into the DLQ. @@ -47,7 +47,7 @@ func NewErrDestinationPublishAttempt(err error, provider string, data map[string // message is the customer-facing string persisted on the attempt (ResponseData); // when empty a generic default is used. The raw err is carried only in the returned // error (for logs/telemetry) and is not persisted on the attempt. -func NewFormatErrorDelivery(provider, message string, err error) (*Delivery, error) { +func NewFormatError(provider, message string, err error) (*Delivery, error) { if message == "" { message = "could not format event for delivery" } diff --git a/internal/destregistry/format_error_delivery_test.go b/internal/destregistry/format_error_test.go similarity index 82% rename from internal/destregistry/format_error_delivery_test.go rename to internal/destregistry/format_error_test.go index ebfd155ef..b2db7c095 100644 --- a/internal/destregistry/format_error_delivery_test.go +++ b/internal/destregistry/format_error_test.go @@ -9,7 +9,7 @@ import ( "github.com/stretchr/testify/require" ) -func TestNewFormatErrorDelivery(t *testing.T) { +func TestNewFormatError(t *testing.T) { t.Parallel() rawErr := errors.New("failed to evaluate key template: invalid type") @@ -17,7 +17,7 @@ func TestNewFormatErrorDelivery(t *testing.T) { t.Run("returns a failed delivery and a publish-attempt error", func(t *testing.T) { t.Parallel() - delivery, err := destregistry.NewFormatErrorDelivery("aws_s3", "", rawErr) + delivery, err := destregistry.NewFormatError("aws_s3", "", rawErr) // A non-nil failed delivery means the registry records an attempt and acks // the message instead of nacking it into the DLQ. @@ -36,7 +36,7 @@ func TestNewFormatErrorDelivery(t *testing.T) { t.Run("uses a generic message when none is given", func(t *testing.T) { t.Parallel() - delivery, _ := destregistry.NewFormatErrorDelivery("aws_s3", "", rawErr) + delivery, _ := destregistry.NewFormatError("aws_s3", "", rawErr) // ...but is NOT persisted on the attempt; the customer-facing response is generic. assert.Equal(t, "could not format event for delivery", delivery.Response["error"]) @@ -46,7 +46,7 @@ func TestNewFormatErrorDelivery(t *testing.T) { t.Run("uses the provided message when given", func(t *testing.T) { t.Parallel() - delivery, _ := destregistry.NewFormatErrorDelivery("aws_s3", "could not build S3 object key", rawErr) + delivery, _ := destregistry.NewFormatError("aws_s3", "could not build S3 object key", rawErr) assert.Equal(t, "could not build S3 object key", delivery.Response["error"]) }) } diff --git a/internal/destregistry/providers/destawskinesis/destawskinesis.go b/internal/destregistry/providers/destawskinesis/destawskinesis.go index fc4ef1531..0baf29d18 100644 --- a/internal/destregistry/providers/destawskinesis/destawskinesis.go +++ b/internal/destregistry/providers/destawskinesis/destawskinesis.go @@ -258,7 +258,7 @@ func (p *AWSKinesisPublisher) Publish(ctx context.Context, event *models.Event) // Format the event into a PutRecordInput input, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("aws_kinesis", "", err) + return destregistry.NewFormatError("aws_kinesis", "", err) } // Send the record to Kinesis diff --git a/internal/destregistry/providers/destawss3/destawss3.go b/internal/destregistry/providers/destawss3/destawss3.go index 8ed82aa0d..6a30a1267 100644 --- a/internal/destregistry/providers/destawss3/destawss3.go +++ b/internal/destregistry/providers/destawss3/destawss3.go @@ -304,7 +304,7 @@ func (p *AWSS3Publisher) Publish(ctx context.Context, event *models.Event) (*des input, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("aws_s3", "", err) + return destregistry.NewFormatError("aws_s3", "", err) } _, err = p.client.PutObject(ctx, input) diff --git a/internal/destregistry/providers/destawssqs/destawssqs.go b/internal/destregistry/providers/destawssqs/destawssqs.go index dd76ed200..511140b47 100644 --- a/internal/destregistry/providers/destawssqs/destawssqs.go +++ b/internal/destregistry/providers/destawssqs/destawssqs.go @@ -163,7 +163,7 @@ func (p *AWSSQSPublisher) Publish(ctx context.Context, event *models.Event) (*de msg, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("aws_sqs", "", err) + return destregistry.NewFormatError("aws_sqs", "", err) } if _, err = p.client.SendMessage(ctx, msg); err != nil { diff --git a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go b/internal/destregistry/providers/destazureservicebus/destazureservicebus.go index 2c8d7b717..7034d991d 100644 --- a/internal/destregistry/providers/destazureservicebus/destazureservicebus.go +++ b/internal/destregistry/providers/destazureservicebus/destazureservicebus.go @@ -149,7 +149,7 @@ func (p *AzureServiceBusPublisher) Publish(ctx context.Context, event *models.Ev message, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("azure_servicebus", "", err) + return destregistry.NewFormatError("azure_servicebus", "", err) } sender, err := p.ensureSender() diff --git a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go b/internal/destregistry/providers/destgcppubsub/destgcppubsub.go index 8bba01bb9..9ee1e9adc 100644 --- a/internal/destregistry/providers/destgcppubsub/destgcppubsub.go +++ b/internal/destregistry/providers/destgcppubsub/destgcppubsub.go @@ -184,7 +184,7 @@ func (pub *GCPPubSubPublisher) Publish(ctx context.Context, event *models.Event) // Format the message msg, err := pub.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("gcp_pubsub", "", err) + return destregistry.NewFormatError("gcp_pubsub", "", err) } // Publish the message diff --git a/internal/destregistry/providers/destkafka/destkafka.go b/internal/destregistry/providers/destkafka/destkafka.go index 2bd72c5e8..6d7cfafaa 100644 --- a/internal/destregistry/providers/destkafka/destkafka.go +++ b/internal/destregistry/providers/destkafka/destkafka.go @@ -222,7 +222,7 @@ func (p *KafkaPublisher) Publish(ctx context.Context, event *models.Event) (*des // Build parsed payload for partition key JMESPath evaluation dataMap, err := event.ParsedData() if err != nil { - return destregistry.NewFormatErrorDelivery("kafka", "", err) + return destregistry.NewFormatError("kafka", "", err) } if dataMap == nil { dataMap = make(map[string]interface{}) diff --git a/internal/destregistry/providers/destwebhook/destwebhook.go b/internal/destregistry/providers/destwebhook/destwebhook.go index b1f77b9ee..055a4318e 100644 --- a/internal/destregistry/providers/destwebhook/destwebhook.go +++ b/internal/destregistry/providers/destwebhook/destwebhook.go @@ -658,7 +658,7 @@ func (p *WebhookPublisher) Publish(ctx context.Context, event *models.Event) (*d httpReq, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("webhook", "", err) + return destregistry.NewFormatError("webhook", "", err) } result := ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook") diff --git a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go b/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go index 3e92ddbbe..33d47c740 100644 --- a/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go +++ b/internal/destregistry/providers/destwebhookstandard/destwebhookstandard.go @@ -555,7 +555,7 @@ func (p *StandardWebhookPublisher) Publish(ctx context.Context, event *models.Ev httpReq, err := p.Format(ctx, event) if err != nil { - return destregistry.NewFormatErrorDelivery("webhook_standard", "", err) + return destregistry.NewFormatError("webhook_standard", "", err) } result := destwebhook.ExecuteHTTPRequest(ctx, p.httpClient, httpReq, "webhook_standard") From b57de332208dcd2b25c70f418b189a923399c075 Mon Sep 17 00:00:00 2001 From: Alex Luong Date: Tue, 16 Jun 2026 21:06:23 +0700 Subject: [PATCH 4/4] docs(e2e): trim format-error regression test comment Co-Authored-By: Claude Fable 5 --- cmd/e2e/regression_format_error_test.go | 26 +++++++++---------------- 1 file changed, 9 insertions(+), 17 deletions(-) diff --git a/cmd/e2e/regression_format_error_test.go b/cmd/e2e/regression_format_error_test.go index 98542be33..eb2d8861e 100644 --- a/cmd/e2e/regression_format_error_test.go +++ b/cmd/e2e/regression_format_error_test.go @@ -20,25 +20,17 @@ import ( "github.com/stretchr/testify/require" ) -// TestE2E_Regression_FormatErrorIsDeliveredAttempt is a standalone regression test -// for the production incident where an aws_s3 destination's key_template could not be -// evaluated for a given event (the template referenced metadata.operationId, which the -// event lacked). Before the fix, the per-event formatting failure returned a nil -// delivery, which the pipeline treated as a system error: it was nacked, never logged -// as an attempt, retried blindly by the message queue, and ultimately dead-lettered — -// paging us instead of surfacing as a normal delivery failure. +// TestE2E_Regression_FormatErrorIsDeliveredAttempt verifies that a per-event +// formatting failure is handled as a normal failed delivery, not a system error +// that gets nacked and dead-lettered. // -// aws_s3 is the only destination type that can force this: its key_template is JMESPath -// validated for syntax only at creation, so a template that is valid at creation can -// still fail at delivery for an event missing a referenced field. (Kafka/Kinesis use the -// same per-event template mechanism but swallow eval errors and fall back to event.ID; -// the other providers have no per-event formatting step that can fail.) +// We use aws_s3 because its key_template is only syntax-validated at creation, so a +// valid template can still fail at delivery for an event missing a referenced field. // -// This test asserts the FIXED behavior, end to end: -// 1. nothing is ever written to S3 (the failure is pre-delivery) -// 2. each attempt is RECORDED as a failed delivery carrying the format error -// 3. it retries on the normal schedule and EXHAUSTS its budget, then stops — -// it is never requeued/dead-lettered (which would log zero attempts and loop) +// Asserts, end to end: +// 1. nothing is written to S3 (the failure is pre-delivery) +// 2. each attempt is recorded as a failed delivery +// 3. it retries on the normal schedule and exhausts, rather than being dead-lettered func TestE2E_Regression_FormatErrorIsDeliveredAttempt(t *testing.T) { t.Parallel() if testing.Short() {