From fa6fca8f1f3ca0f7512b18aa44d66ca7002ba64f Mon Sep 17 00:00:00 2001 From: Phil Leggetter Date: Mon, 2 Nov 2026 16:39:25 +0000 Subject: [PATCH] test(listen): attribute missing binary deliveries and stop one late delivery failing later steps The binary-body listen tests failed intermittently with "Connected. Waiting for events..." then "no request reached the local app within 45s", which could not say where the event went. Reproduced locally: the event was created for the listening CLI and stayed QUEUED with no delivery attempt for minutes while the CLI's websocket was connected and answering pings. The same stall failed the JSON forwarding test, so it is neither binary-specific nor in the CLI. The harness made it worse: localApp.next took whatever arrived next, so a late delivery failed the following steps with a wrong Content-Type and file. - Wait for listen's "Connected. Waiting for events..." line instead of sleeping 12 seconds. - Tag each send with X-Acceptance-Delivery-Id; each step waits for its own delivery and skips others. - On timeout, look the request and its events up by that header and report event status and attempts, plus the websocket lines from listen's debug log. - Drop the duplicate connection delete that logged a 410 warning. Co-Authored-By: Claude Opus 5.5 --- test/acceptance/listen_binary_test.go | 232 +++++++++++++++++++++----- 1 file changed, 188 insertions(+), 44 deletions(-) diff --git a/test/acceptance/listen_binary_test.go b/test/acceptance/listen_binary_test.go index 116f509f..6a7c6d0d 100644 --- a/test/acceptance/listen_binary_test.go +++ b/test/acceptance/listen_binary_test.go @@ -26,6 +26,7 @@ import ( "path/filepath" "runtime" "strconv" + "strings" "sync" "testing" "time" @@ -89,10 +90,17 @@ func downloadLegacyCLI(t *testing.T) string { // Local app // --------------------------------------------------------------------------- +// deliveryIDHeader tags every request these tests send with a value unique to +// that send. Hookdeck forwards it to the local app, so a step waits for its own +// delivery rather than whatever arrives next, and a failure can find the +// request and its event in the API. +const deliveryIDHeader = "X-Acceptance-Delivery-Id" + // receivedRequest is one request as the local app saw it. Files holds what the // app wrote to disk: the raw body for single-file requests, or every file part // of a multipart form, keyed by form field name. type receivedRequest struct { + deliveryID string contentType string eventID string attemptCount int @@ -126,6 +134,7 @@ func startLocalApp(t *testing.T) *localApp { app.mu.Unlock() req := receivedRequest{ + deliveryID: r.Header.Get(deliveryIDHeader), contentType: r.Header.Get("Content-Type"), eventID: r.Header.Get("X-Hookdeck-EventID"), files: map[string]string{}, @@ -184,17 +193,25 @@ func (a *localApp) setStatus(status func(n int) int) { a.status = status } -func (a *localApp) next(t *testing.T, within time.Duration, outputs ...*syncBuffer) receivedRequest { +// next waits for the delivery of the request sent with deliveryID. A delivery +// of an earlier step's request is logged and skipped: one arriving late used +// to be taken as the next step's, so a slow first delivery failed the step +// after it instead. On timeout it reports how far the request got. +func (a *localApp) next(t *testing.T, s binaryListenSetup, deliveryID string, within time.Duration, l listenOutput) receivedRequest { t.Helper() - select { - case r := <-a.received: - return r - case <-time.After(within): - for _, o := range outputs { - t.Logf("listen output: %s", o.String()) + timeout := time.After(within) + for { + select { + case r := <-a.received: + if r.deliveryID == deliveryID { + return r + } + t.Logf("local app: skipping a delivery of %q while waiting for %q", r.deliveryID, deliveryID) + case <-timeout: + l.log(t) + t.Fatalf("no request reached the local app within %s.\n%s", within, s.explainMissingDelivery(t, deliveryID)) + return receivedRequest{} } - t.Fatalf("no request reached the local app within %s; if the event failed with CLI_BINARY_UNSUPPORTED, the server does not deliver binary bodies to this CLI", within) - return receivedRequest{} } } @@ -348,6 +365,7 @@ func assertSavedFile(t *testing.T, path string, want fixture) { type binaryListenSetup struct { cli *CLIRunner sourceName string + sourceID string sourceURL string connID string } @@ -369,38 +387,163 @@ func newBinaryListenSetup(t *testing.T, prefix string) binaryListenSetup { "--destination-cli-path", "/", )) require.NotEmpty(t, conn.ID) - t.Cleanup(func() { deleteConnection(t, cli, conn.ID) }) + // No cleanup of its own: listen's cleanup deletes the connection by source + // name, and a second delete here only logged a 410 warning into every + // failure's output. var src Source require.NoError(t, cli.RunJSON(&src, "gateway", "source", "get", conn.Source.ID)) require.NotEmpty(t, src.URL) - return binaryListenSetup{cli: cli, sourceName: sourceName, sourceURL: src.URL, connID: conn.ID} + return binaryListenSetup{cli: cli, sourceName: sourceName, sourceID: conn.Source.ID, sourceURL: src.URL, connID: conn.ID} +} + +// listenOutput is a running `hookdeck listen` started by these tests. +type listenOutput struct { + stdout, stderr *syncBuffer + done chan error +} + +// listenReadyLine is what listen prints in compact mode once its websocket is +// connected. Both this CLI and v3.0.3 print it. +const listenReadyLine = "Connected. Waiting for events..." + +// startBinaryListen runs `listen` for the setup's source, forwarding to port, +// from binary or, when binary is empty, from a fresh build of this CLI. Debug +// logging is on so a failure can show whether any message reached the +// session's websocket. +func startBinaryListen(t *testing.T, cli *CLIRunner, binary, port, sourceName string) listenOutput { + t.Helper() + args := []string{"listen", port, sourceName, "--output", "compact", "--log-level", "debug"} + var l listenOutput + if binary == "" { + _, l.stdout, l.stderr, l.done = startListenCapturingOutput(t, cli, args...) + } else { + _, l.stdout, l.stderr, l.done = startListenBinaryCapturingOutput(t, cli, binary, args...) + } + return l } -// waitForTunnel gives listen time to connect and fails fast if it exited. -func waitForTunnel(t *testing.T, done chan error, outputs ...*syncBuffer) { +// waitReady waits for listen to report its websocket connected, rather than +// sleeping a fixed time, and fails fast if listen exits first. +func (l listenOutput) waitReady(t *testing.T) { t.Helper() - t.Log("Waiting 12 seconds for the tunnel to connect...") - time.Sleep(12 * time.Second) - select { - case err := <-done: - for _, o := range outputs { - t.Logf("listen output: %s", o.String()) + const within = 60 * time.Second + deadline := time.Now().Add(within) + for !strings.Contains(l.stdout.String(), listenReadyLine) { + select { + case err := <-l.done: + l.log(t) + t.Fatalf("listen exited before connecting: %v", err) + case <-time.After(100 * time.Millisecond): + } + if time.Now().After(deadline) { + l.log(t) + t.Fatalf("listen did not print %q within %s", listenReadyLine, within) + } + } +} + +// log writes listen's stdout, and the lines of its debug log that say what +// happened on the websocket: connects, closes, and every incoming message. +// The rest of the debug log is API traffic and would bury those lines. +func (l listenOutput) log(t *testing.T) { + t.Helper() + t.Logf("listen stdout:\n%s", l.stdout.String()) + var lines []string + for _, line := range strings.Split(l.stderr.String(), "\n") { + if !strings.Contains(line, "prefix=websocket.") && !strings.Contains(line, "level=error") && !strings.Contains(line, "level=warn") { + continue } - t.Fatalf("listen exited before forwarding anything: %v", err) - default: + if len(line) > 300 { + line = line[:300] + "..." + } + lines = append(lines, line) } + t.Logf("listen websocket log:\n%s", strings.Join(lines, "\n")) } -// postRaw sends body to a source URL with the given Content-Type, unmodified. -func postRaw(t *testing.T, sourceURL, contentType string, body []byte) { +// postRaw sends body to the setup's source URL with the given Content-Type, +// unmodified, and returns the delivery id it tagged the request with. +func (s binaryListenSetup) postRaw(t *testing.T, contentType string, body []byte) string { t.Helper() + deliveryID := fmt.Sprintf("%s-%d", s.sourceName, time.Now().UnixNano()) + req, err := http.NewRequest(http.MethodPost, s.sourceURL, bytes.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Content-Type", contentType) + req.Header.Set(deliveryIDHeader, deliveryID) client := &http.Client{Timeout: 10 * time.Second} - resp, err := client.Post(sourceURL, contentType, bytes.NewReader(body)) + resp, err := client.Do(req) require.NoError(t, err, "POST to source URL failed") defer resp.Body.Close() require.True(t, resp.StatusCode >= 200 && resp.StatusCode < 300, "POST to source URL returned %d", resp.StatusCode) + return deliveryID +} + +// explainMissingDelivery says how far the request tagged deliveryID got, as +// the API reports it: whether the source recorded it, whether an event was +// created for a CLI, and whether that event was ever attempted. Together with +// the websocket log it separates "never sent to this listen session" from +// "sent, but not forwarded by the CLI". +func (s binaryListenSetup) explainMissingDelivery(t *testing.T, deliveryID string) string { + t.Helper() + var requests struct { + Models []struct { + ID string `json:"id"` + EventsCount int `json:"events_count"` + CLIEventsCount int `json:"cli_events_count"` + IgnoredCount int `json:"ignored_count"` + RejectionCause any `json:"rejection_cause"` + } `json:"models"` + } + // Search lags a few seconds behind ingestion, so a request sent moments ago + // can be missing from it; give it a short while before saying so. + for i := 0; i < 10; i++ { + if err := s.cli.RunJSON(&requests, "gateway", "request", "list", "--source-id", s.sourceID, "--search-term", deliveryID); err != nil { + return fmt.Sprintf("Could not look up the request: %v", err) + } + if len(requests.Models) > 0 { + break + } + time.Sleep(2 * time.Second) + } + if len(requests.Models) == 0 { + return fmt.Sprintf("Source %s has no request tagged %s=%s.", s.sourceID, deliveryIDHeader, deliveryID) + } + r := requests.Models[0] + report := []string{fmt.Sprintf("Request %s: %d event(s), %d CLI event(s), %d ignored, rejection cause %v.", + r.ID, r.EventsCount, r.CLIEventsCount, r.IgnoredCount, r.RejectionCause)} + + var events struct { + Models []struct { + ID string `json:"id"` + CLIID string `json:"cli_id"` + Status string `json:"status"` + Attempts int `json:"attempts"` + ErrorCode string `json:"error_code"` + } `json:"models"` + } + if err := s.cli.RunJSON(&events, "gateway", "event", "list", "--connection-id", s.connID, "--search-term", deliveryID); err != nil { + return strings.Join(append(report, fmt.Sprintf("Could not look up its events: %v", err)), "\n") + } + if len(events.Models) == 0 { + report = append(report, "No event on connection "+s.connID+" carries this request.") + } + neverAttempted := false + for _, e := range events.Models { + report = append(report, fmt.Sprintf("Event %s for CLI %s: status %s, %d attempt(s), error code %q.", + e.ID, e.CLIID, e.Status, e.Attempts, e.ErrorCode)) + if e.Attempts == 0 && (e.Status == "QUEUED" || e.Status == "SCHEDULED") { + neverAttempted = true + } + if e.ErrorCode == "CLI_BINARY_UNSUPPORTED" { + report = append(report, "CLI_BINARY_UNSUPPORTED: the server does not deliver binary bodies to this CLI.") + } + } + if neverAttempted { + report = append(report, "The event was created for a CLI but never attempted, so nothing was sent to the listen session: the delivery stopped before it reached the CLI.") + } + return strings.Join(report, "\n") } // --------------------------------------------------------------------------- @@ -418,15 +561,15 @@ func TestListenForwardsBinaryFilesThatTheAppCanSave(t *testing.T) { s := newBinaryListenSetup(t, "test-bin") app := startLocalApp(t) - _, stdout, stderr, done := startListenCapturingOutput(t, s.cli, "listen", app.port, s.sourceName, "--output", "compact") - waitForTunnel(t, done, stdout, stderr) + listen := startBinaryListen(t, s.cli, "", app.port, s.sourceName) + listen.waitReady(t) // Raw bodies: the content types ingestion stores as binary. for _, f := range []fixture{allBytesFixture(), pngFixture(t), jpegFixture(t), pdfFixture()} { f := f t.Run("raw "+f.contentType, func(t *testing.T) { - postRaw(t, s.sourceURL, f.contentType, f.data) - got := app.next(t, 45*time.Second, stdout, stderr) + id := s.postRaw(t, f.contentType, f.data) + got := app.next(t, s, id, 45*time.Second, listen) assert.Equal(t, f.contentType, got.contentType, "original Content-Type must be forwarded") assertSavedFile(t, got.files["body"], f) }) @@ -436,8 +579,8 @@ func TestListenForwardsBinaryFilesThatTheAppCanSave(t *testing.T) { files := []fixture{pngFixture(t), wavFixture(), allBytesFixture()} contentType, body := multipartUpload(t, files...) - postRaw(t, s.sourceURL, contentType, body) - got := app.next(t, 45*time.Second, stdout, stderr) + id := s.postRaw(t, contentType, body) + got := app.next(t, s, id, 45*time.Second, listen) assert.Equal(t, contentType, got.contentType, "boundary travels in the original Content-Type") require.Len(t, got.files, len(files), "the app should save every file part") @@ -463,20 +606,21 @@ func TestListenRetriesBinaryBodyByteExact(t *testing.T) { } return http.StatusOK }) - _, stdout, stderr, done := startListenCapturingOutput(t, s.cli, "listen", app.port, s.sourceName, "--output", "compact") - waitForTunnel(t, done, stdout, stderr) + listen := startBinaryListen(t, s.cli, "", app.port, s.sourceName) + listen.waitReady(t) f := pngFixture(t) - postRaw(t, s.sourceURL, f.contentType, f.data) + id := s.postRaw(t, f.contentType, f.data) - first := app.next(t, 45*time.Second, stdout, stderr) + first := app.next(t, s, id, 45*time.Second, listen) require.NotEmpty(t, first.eventID, "delivery should carry X-Hookdeck-EventID") assert.Equal(t, 1, first.attemptCount) assertSavedFile(t, first.files["body"], f) s.cli.RunExpectSuccess("gateway", "event", "retry", first.eventID) - second := app.next(t, 45*time.Second, stdout, stderr) + // The retry delivers the same request, so it carries the same delivery id. + second := app.next(t, s, id, 45*time.Second, listen) assert.Equal(t, first.eventID, second.eventID, "the retry should deliver the same event") assert.Equal(t, 2, second.attemptCount) assert.Equal(t, f.contentType, second.contentType) @@ -498,16 +642,16 @@ func TestListenBinaryWithOldAndNewCLIListening(t *testing.T) { legacyCLI := newSeparateCLIClient(t, s.cli) newApp := startLocalApp(t) oldApp := startLocalApp(t) - _, newOut, newErr, newDone := startListenCapturingOutput(t, s.cli, "listen", newApp.port, s.sourceName, "--output", "compact") - _, oldOut, oldErr, oldDone := startListenBinaryCapturingOutput(t, legacyCLI, legacyBinary, "listen", oldApp.port, s.sourceName, "--output", "compact") - waitForTunnel(t, newDone, newOut, newErr) - waitForTunnel(t, oldDone, oldOut, oldErr) + newListen := startBinaryListen(t, s.cli, "", newApp.port, s.sourceName) + oldListen := startBinaryListen(t, legacyCLI, legacyBinary, oldApp.port, s.sourceName) + newListen.waitReady(t) + oldListen.waitReady(t) t.Run("raw image", func(t *testing.T) { f := pngFixture(t) - postRaw(t, s.sourceURL, f.contentType, f.data) + id := s.postRaw(t, f.contentType, f.data) - got := newApp.next(t, 45*time.Second, newOut, newErr) + got := newApp.next(t, s, id, 45*time.Second, newListen) assertSavedFile(t, got.files["body"], f) oldApp.expectNothing(t, 20*time.Second) @@ -517,16 +661,16 @@ func TestListenBinaryWithOldAndNewCLIListening(t *testing.T) { t.Run("multipart upload", func(t *testing.T) { files := []fixture{pngFixture(t), wavFixture()} contentType, body := multipartUpload(t, files...) - postRaw(t, s.sourceURL, contentType, body) + id := s.postRaw(t, contentType, body) - got := newApp.next(t, 45*time.Second, newOut, newErr) + got := newApp.next(t, s, id, 45*time.Second, newListen) for i, f := range files { assertSavedFile(t, got.files[fmt.Sprintf("file%d", i)], f) } // The old CLI still gets the upload, decoded as UTF-8 text as before // multipart moved to binary, so its file parts are not byte-exact. - legacy := oldApp.next(t, 45*time.Second, oldOut, oldErr) + legacy := oldApp.next(t, s, id, 45*time.Second, oldListen) assert.Equal(t, contentType, legacy.contentType) legacyPNG, err := os.ReadFile(legacy.files["file0"]) require.NoError(t, err)