fix(transform): return an error on non-2xx in send_http_post

CODEOWNERS

@@ -1 +1 @@
-*       @brexhq/substation
+*       @LiveRamp/seceng

cmd/gcp/function/substation/pubsub.go

@@ -78,7 +78,7 @@ func (ps *processingState) log() {
 	}
 }
 
-// nolint: gocognit // Ignore cognitive complexity.
+//nolint:gocyclo,cyclop,gocognit // Ignore cyclomatic and cognitive complexity.
 func pubSubHandler(ctx context.Context, e cloudevents.Event) error {
 	// Set up signal handling for graceful shutdown
 	var state processingState

transform/send_http_post.go

@@ -1,6 +1,7 @@
 package transform
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -19,6 +20,10 @@ import (
 	"github.com/brexhq/substation/v2/internal/secrets"
 )
 
+// errorBodyLimit bounds how much of a non-2xx response body is read for error
+// context. Servers may echo request data, so this is deliberately small.
+const errorBodyLimit = 512
+
 type sendHTTPPostConfig struct {
 	// URL is the HTTP(S) endpoint that data is sent to.
 	URL string `json:"url"`
@@ -188,7 +193,11 @@ func (tf *sendHTTPPost) send(ctx context.Context, key string) error {
 			return err
 		}
 
-		//nolint:errcheck // Response body is discarded to avoid resource leaks.
+		// A bounded prefix of the body is retained so that non-2xx responses can
+		// explain themselves. The limit keeps payload data echoed by the server
+		// out of logs and errors.
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit))
+		//nolint:errcheck // Remainder is discarded to avoid resource leaks.
 		io.Copy(io.Discard, resp.Body)
 		resp.Body.Close()
 
@@ -200,6 +209,16 @@ func (tf *sendHTTPPost) send(ctx context.Context, key string) error {
 			WithField("event_count", eventCount).
 			WithField("duration_ms", duration.Milliseconds()).
 			Debug("Sent HTTP POST request")
+
+		// Responses that the HTTP client does not retry (any 4xx except 429) are
+		// returned with a nil error, so the status must be checked explicitly.
+		// Without this the batch is silently discarded.
+		//
+		// The URL is deliberately omitted: it is interpolated from secrets and may
+		// carry credentials, and errors propagate further than the debug log above.
+		if resp.StatusCode < 200 || resp.StatusCode > 299 {
+			return fmt.Errorf("transform %s: http post: status %d: %s", tf.conf.ID, resp.StatusCode, bytes.TrimSpace(body))
+		}
 	}
 
 	return nil

transform/send_http_post_test.go

@@ -0,0 +1,76 @@
+package transform
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/brexhq/substation/v2/config"
+	"github.com/brexhq/substation/v2/message"
+)
+
+var _ Transformer = &sendHTTPPost{}
+
+// Statuses that the HTTP client does not retry are returned with a nil error,
+// so the transform must inspect the status code itself. Retried statuses (429
+// and 5xx) are deliberately excluded here because exercising them would incur
+// the client's full backoff and exceed the test timeout.
+var sendHTTPPostTests = []struct {
+	name       string
+	statusCode int
+	body       string
+	expectErr  bool
+}{
+	{"200 succeeds", http.StatusOK, "", false},
+	{"201 succeeds", http.StatusCreated, "", false},
+	{"400 errors", http.StatusBadRequest, "invalid JSON at offset 12", true},
+	{"401 errors", http.StatusUnauthorized, "invalid ingest token", true},
+	{"404 errors", http.StatusNotFound, "no such endpoint", true},
+}
+
+func TestSendHTTPPost(t *testing.T) {
+	ctx := context.TODO()
+
+	for _, test := range sendHTTPPostTests {
+		t.Run(test.name, func(t *testing.T) {
+			serv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.WriteHeader(test.statusCode)
+				//nolint:errcheck // Test server write.
+				w.Write([]byte(test.body))
+			}))
+			defer serv.Close()
+
+			tf, err := newSendHTTPPost(ctx, config.Config{
+				Settings: map[string]interface{}{"url": serv.URL},
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if _, err := tf.Transform(ctx, message.New().SetData([]byte(`{"a":1}`))); err != nil {
+				t.Fatal(err)
+			}
+
+			// The batch is sent when the control message is received.
+			_, err = tf.Transform(ctx, message.New().AsControl())
+
+			if !test.expectErr {
+				if err != nil {
+					t.Errorf("expected no error, got %v", err)
+				}
+
+				return
+			}
+
+			if err == nil {
+				t.Fatalf("expected an error for status %d, got nil", test.statusCode)
+			}
+
+			if !strings.Contains(err.Error(), test.body) {
+				t.Errorf("expected error to contain response body %q, got %q", test.body, err)
+			}
+		})
+	}
+}