Retry the registry record on transient failures

cmd/record-release/main.go

@@ -313,11 +313,6 @@ func main() {
 		os.Exit(1)
 	}
 	endpoint := baseURL.JoinPath("/api/v1/ingest/release")
-	httpReq, err := http.NewRequest(http.MethodPost, endpoint.String(), bytes.NewReader(bodyBytes))
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "record-release: error: creating HTTP request: %v\n", err)
-		os.Exit(1)
-	}
 
 	client := &http.Client{
 		Timeout: 30 * time.Second,
@@ -326,16 +321,9 @@ func main() {
 			base:  http.DefaultTransport,
 		},
 	}
-	resp, err := client.Do(httpReq)
+	resp, respBody, err := postWithRetry(client, endpoint.String(), bodyBytes)
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "record-release: error: HTTP request failed: %v\n", err)
-		os.Exit(1)
-	}
-	defer resp.Body.Close()
-
-	respBody, err := io.ReadAll(resp.Body)
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "record-release: error: reading response body: %v\n", err)
+		fmt.Fprintf(os.Stderr, "record-release: error: %v\n", err)
 		os.Exit(1)
 	}
 
@@ -363,6 +351,58 @@ func main() {
 	}
 }
 
+// retryDelays are the waits between the attempts of postWithRetry; tests
+// shorten them.
+var retryDelays = []time.Duration{10 * time.Second, 20 * time.Second, 40 * time.Second, 80 * time.Second}
+
+// retryable reports whether a response is worth another attempt: a 5xx, a
+// 429, or a 400 the registry raised because the CDN answered 5xx while it was
+// fetching an asset's signature files -- a transient upstream failure, not a
+// bad record. Any other 4xx is final.
+func retryable(status int, body []byte) bool {
+	return status >= 500 || status == http.StatusTooManyRequests ||
+		(status == http.StatusBadRequest && bytes.Contains(body, []byte("returned HTTP 5")))
+}
+
+// postWithRetry POSTs body to endpoint, re-sending it after each retryable
+// failure until the delays run out. It returns the last response and its
+// body, or the transport error of the last attempt.
+func postWithRetry(client *http.Client, endpoint string, body []byte) (*http.Response, []byte, error) {
+	for attempt := 0; ; attempt++ {
+		resp, err := client.Do(mustRequest(endpoint, body))
+		var respBody []byte
+		if err == nil {
+			// A read failure is the connection dying mid-response: retried
+			// like a failed request, not reported as a bad answer.
+			respBody, err = io.ReadAll(resp.Body)
+			resp.Body.Close()
+			if err == nil && !retryable(resp.StatusCode, respBody) {
+				return resp, respBody, nil
+			}
+		}
+		if attempt == len(retryDelays) {
+			if err != nil {
+				return nil, nil, fmt.Errorf("HTTP request failed after %d attempts: %w", attempt+1, err)
+			}
+			return resp, respBody, nil
+		}
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "record-release: attempt %d failed: %v; retrying in %s\n", attempt+1, err, retryDelays[attempt])
+		} else {
+			fmt.Fprintf(os.Stderr, "record-release: attempt %d failed: HTTP %d; retrying in %s\n", attempt+1, resp.StatusCode, retryDelays[attempt])
+		}
+		time.Sleep(retryDelays[attempt])
+	}
+}
+
+func mustRequest(endpoint string, body []byte) *http.Request {
+	req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewReader(body))
+	if err != nil {
+		panic(err) // the endpoint was parsed by url.Parse already
+	}
+	return req
+}
+
 func transformAssets(manifest *pb.Manifest) map[string]*ReleaseAsset {
 	assets := make(map[string]*ReleaseAsset)
 	for platform, asset := range manifest.GetAssets() {

cmd/record-release/main_test.go

@@ -3,9 +3,13 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 	"unicode/utf8"
 
 	pb "github.com/ConductorOne/github-workflows/pb/artifacts/v1"
@@ -305,3 +309,109 @@ func attestation(predicateType, bundleHref string) *pb.AttestationDescriptor {
 func strPtr(s string) *string {
 	return &s
 }
+
+// A transient failure -- including the registry's 400 for a CDN 5xx -- is
+// retried with the same body; a real 400 is not.
+func TestPostWithRetry(t *testing.T) {
+	saved := retryDelays
+	retryDelays = []time.Duration{0, 0, 0, 0}
+	t.Cleanup(func() { retryDelays = saved })
+
+	cases := []struct {
+		name      string
+		responses []int
+		body      string
+		wantCalls int
+		wantCode  int
+	}{
+		{"503 then 200", []int{503, 200}, "", 2, 200},
+		{"400 for an upstream 503 then 200", []int{400, 200}, `{"code":3,"message":"fetch certificate: GET https://cdn.example/x.cert returned HTTP 503"}`, 2, 200},
+		{"plain 400 is final", []int{400}, `{"code":3,"message":"version is required"}`, 1, 400},
+		{"gives up after five", []int{503, 503, 503, 503, 503}, "", 5, 503},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			calls := 0
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if got, _ := io.ReadAll(r.Body); string(got) != "payload" {
+					t.Errorf("attempt %d body = %q, want payload", calls+1, got)
+				}
+				w.WriteHeader(tc.responses[calls])
+				_, _ = w.Write([]byte(tc.body))
+				calls++
+			}))
+			defer srv.Close()
+
+			resp, _, err := postWithRetry(srv.Client(), srv.URL, []byte("payload"))
+			if err != nil {
+				t.Fatalf("postWithRetry: %v", err)
+			}
+			if calls != tc.wantCalls || resp.StatusCode != tc.wantCode {
+				t.Fatalf("calls = %d, status = %d; want %d, %d", calls, resp.StatusCode, tc.wantCalls, tc.wantCode)
+			}
+		})
+	}
+}
+
+// A connection dropped before or during the response is retried the same way;
+// five drops in a row are the final error.
+func TestPostWithRetryTransportErrors(t *testing.T) {
+	saved := retryDelays
+	retryDelays = []time.Duration{0, 0, 0, 0}
+	t.Cleanup(func() { retryDelays = saved })
+
+	drops := 0
+	drop := func(w http.ResponseWriter) {
+		drops++
+		conn, _, err := w.(http.Hijacker).Hijack()
+		if err != nil {
+			t.Fatalf("hijack: %v", err)
+		}
+		conn.Close()
+	}
+
+	t.Run("dropped then 200", func(t *testing.T) {
+		drops = 0
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if drops == 0 {
+				drop(w)
+				return
+			}
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer srv.Close()
+		resp, _, err := postWithRetry(srv.Client(), srv.URL, []byte("payload"))
+		if err != nil || resp.StatusCode != http.StatusOK || drops != 1 {
+			t.Fatalf("err = %v, status = %v, drops = %d; want nil, 200, 1", err, resp, drops)
+		}
+	})
+
+	t.Run("body cut after the headers then 200", func(t *testing.T) {
+		drops = 0
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if drops == 0 {
+				w.Header().Set("Content-Length", "100")
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write([]byte("part"))
+				drop(w) // flushes the partial body, then closes: the client sees an unexpected EOF
+				return
+			}
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer srv.Close()
+		resp, _, err := postWithRetry(srv.Client(), srv.URL, []byte("payload"))
+		if err != nil || resp.StatusCode != http.StatusOK || drops != 1 {
+			t.Fatalf("err = %v, status = %v, drops = %d; want nil, 200, 1", err, resp, drops)
+		}
+	})
+
+	t.Run("dropped five times", func(t *testing.T) {
+		drops = 0
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { drop(w) }))
+		defer srv.Close()
+		_, _, err := postWithRetry(srv.Client(), srv.URL, []byte("payload"))
+		if err == nil || !strings.Contains(err.Error(), "after 5 attempts") || drops != 5 {
+			t.Fatalf("err = %v, drops = %d; want 'after 5 attempts', 5", err, drops)
+		}
+	})
+}

docs/release-workflow.md

@@ -108,6 +108,9 @@ Records release metadata in the connector registry API:
   folded into one line with a count before the changelog is recorded
 - Includes `config_schema.json` and `baton_capabilities.json` when present
 - Sends release timestamp, commit SHA, and workflow run metadata
+- Retries the record up to five times with doubling waits (10s to 80s) on
+  transport errors, 5xx responses, 429, or a 400 whose message reports an
+  upstream 5xx from the CDN; other 4xx responses fail immediately
 
 ### verify-release