diff --git a/internal/gh/client.go b/internal/gh/client.go index d7e9b27..1641bb8 100644 --- a/internal/gh/client.go +++ b/internal/gh/client.go @@ -66,6 +66,8 @@ type TokenSource interface { } type Client struct { + // freshHC opens a new connection for every request. + freshHC *http.Client rateReset time.Time blockedUntil time.Time auth TokenSource @@ -88,7 +90,7 @@ type cachedResponse struct { } func NewClient(baseURL string, auth TokenSource, hc *http.Client, maxItems int) *Client { - c := &Client{base: strings.TrimRight(baseURL, "/"), hc: hc, auth: auth, maxItems: maxItems, sleep: sleepCtx, now: time.Now, cache: map[string]cachedResponse{}} + c := &Client{base: strings.TrimRight(baseURL, "/"), hc: hc, freshHC: freshClient(hc), auth: auth, maxItems: maxItems, sleep: sleepCtx, now: time.Now, cache: map[string]cachedResponse{}} c.rateRemaining.Store(-1) return c } @@ -111,7 +113,9 @@ func sleepCtx(ctx context.Context, d time.Duration) error { } // do performs one authenticated GET, retrying transient failures, and returns the body and Link header. -func (c *Client) do(ctx context.Context, repo, target string) ([]byte, string, error) { +// do performs one authenticated GET, retrying transient failures. fresh sends it on a new connection +// and skips the response cache, so a stale GitHub replica behind a reused connection cannot answer it. +func (c *Client) do(ctx context.Context, repo, target string, fresh bool) ([]byte, string, error) { var lastErr error for attempt := 0; attempt <= maxRetries; attempt++ { if attempt > 0 { @@ -119,7 +123,7 @@ func (c *Client) do(ctx context.Context, repo, target string) ([]byte, string, e return nil, "", err } } - body, link, retry, err := c.once(ctx, repo, target) + body, link, retry, err := c.once(ctx, repo, target, fresh) if err == nil { return body, link, nil } @@ -131,7 +135,7 @@ func (c *Client) do(ctx context.Context, repo, target string) ([]byte, string, e return nil, "", lastErr } -func (c *Client) once(ctx context.Context, repo, target string) (body []byte, link string, retry bool, err error) { +func (c *Client) once(ctx context.Context, repo, target string, fresh bool) (body []byte, link string, retry bool, err error) { if gerr := c.guard(); gerr != nil { return nil, "", false, gerr } @@ -150,10 +154,15 @@ func (c *Client) once(ctx context.Context, repo, target string) (body []byte, li c.mu.Lock() cached, haveCache := c.cache[target] c.mu.Unlock() + haveCache = haveCache && !fresh if haveCache { req.Header.Set("If-None-Match", cached.etag) } - resp, err := c.hc.Do(req) + hc := c.hc + if fresh { + hc = c.freshHC + } + resp, err := hc.Do(req) if err != nil { return nil, "", ctx.Err() == nil, err } @@ -174,7 +183,7 @@ func (c *Client) once(ctx context.Context, repo, target string) (body []byte, li return nil, "", resp.StatusCode >= 500, ae } link = resp.Header.Get("Link") - if etag := resp.Header.Get("ETag"); etag != "" { + if etag := resp.Header.Get("ETag"); etag != "" && !fresh { c.mu.Lock() c.cache[target] = cachedResponse{etag: etag, body: data, link: link} c.mu.Unlock() @@ -241,7 +250,16 @@ func (c *Client) url(path string, q url.Values) string { } func (c *Client) getJSON(ctx context.Context, repo, path string, q url.Values, out any) error { - body, _, err := c.do(ctx, repo, c.url(path, q)) + return c.getJSONMode(ctx, repo, path, q, out, false) +} + +// getJSONFresh is getJSON on a new connection without the response cache. +func (c *Client) getJSONFresh(ctx context.Context, repo, path string, q url.Values, out any) error { + return c.getJSONMode(ctx, repo, path, q, out, true) +} + +func (c *Client) getJSONMode(ctx context.Context, repo, path string, q url.Values, out any, fresh bool) error { + body, _, err := c.do(ctx, repo, c.url(path, q), fresh) if err != nil { return err } @@ -257,7 +275,7 @@ func listAll[T any](ctx context.Context, c *Client, repo, path string, q url.Val next := c.url(path, q) var all []T for next != "" && len(all) < limit { - body, link, err := c.do(ctx, repo, next) + body, link, err := c.do(ctx, repo, next, false) if err != nil { return all, err } @@ -289,3 +307,18 @@ func nextLink(h string) string { } return "" } + +// freshClient copies hc but turns off connection reuse. Custom transports that are not an +// *http.Transport are used as they are. +func freshClient(hc *http.Client) *http.Client { + tr, ok := hc.Transport.(*http.Transport) + if !ok { + if hc.Transport != nil { + return hc + } + tr, _ = http.DefaultTransport.(*http.Transport) + } + clone := tr.Clone() + clone.DisableKeepAlives = true + return &http.Client{Transport: clone, Timeout: hc.Timeout} +} diff --git a/internal/gh/fetch.go b/internal/gh/fetch.go index 90195dd..e0bc6e0 100644 --- a/internal/gh/fetch.go +++ b/internal/gh/fetch.go @@ -262,24 +262,28 @@ func (c *Client) workflows(ctx context.Context, full, branch string) ([]model.Wo if w.State != "active" || len(out) >= maxWorkflows { continue } - var runs struct { - WorkflowRuns []rawRun `json:"workflow_runs"` - } - q := url.Values{ - "branch": {branch}, - "exclude_pull_requests": {"true"}, - "per_page": {fmt.Sprint(runsPerWorkflow)}, - } - path := fmt.Sprintf("/repos/%s/actions/workflows/%d/runs", full, w.ID) - if err := c.getJSON(ctx, full, path, q, &runs); err != nil { + runs, err := c.workflowRuns(ctx, full, w.ID, branch, false) + if err != nil { if first == nil { first = fmt.Errorf("%s: %w", w.Name, err) } continue } - if wf, ok := buildWorkflow(w.Name, w.HTMLURL, runs.WorkflowRuns); ok { - out = append(out, wf) + wf, ok := buildWorkflow(w.Name, w.HTMLURL, runs) + if !ok { + continue } + // GitHub sometimes answers a run listing from a lagging replica, and a reused connection + // keeps hitting it. A failure is what raises an alert, so confirm it on a new connection + // and keep the union of both answers. + if wf.LastDone != nil && wf.LastDone.Failed() { + if again, err := c.workflowRuns(ctx, full, w.ID, branch, true); err == nil { + if merged, ok := buildWorkflow(w.Name, w.HTMLURL, mergeRuns(runs, again)); ok { + wf = merged + } + } + } + out = append(out, wf) } return out, first } @@ -343,3 +347,36 @@ func (c *Client) latestRelease(ctx context.Context, full, repoURL string) (*mode FromTag: true, }, nil } + +func (c *Client) workflowRuns(ctx context.Context, full string, id int64, branch string, fresh bool) ([]rawRun, error) { + var out struct { + WorkflowRuns []rawRun `json:"workflow_runs"` + } + q := url.Values{ + "branch": {branch}, + "exclude_pull_requests": {"true"}, + "per_page": {fmt.Sprint(runsPerWorkflow)}, + } + path := fmt.Sprintf("/repos/%s/actions/workflows/%d/runs", full, id) + get := c.getJSON + if fresh { + get = c.getJSONFresh + } + if err := get(ctx, full, path, q, &out); err != nil { + return nil, err + } + return out.WorkflowRuns, nil +} + +// mergeRuns returns the runs of a and b once each, keyed by run ID. +func mergeRuns(a, b []rawRun) []rawRun { + seen := make(map[int64]bool, len(a)+len(b)) + out := make([]rawRun, 0, len(a)+len(b)) + for _, r := range slices.Concat(a, b) { + if !seen[r.ID] { + seen[r.ID] = true + out = append(out, r) + } + } + return out +} diff --git a/internal/gh/gh_test.go b/internal/gh/gh_test.go index 56ae426..b7ef89c 100644 --- a/internal/gh/gh_test.go +++ b/internal/gh/gh_test.go @@ -459,3 +459,97 @@ func TestAPIErrorMessageParsing(t *testing.T) { t.Fatalf("len=%d", len(got)) } } + +// staleThenFresh serves an old failing run on the first request and the current runs afterwards, +// like a lagging GitHub replica followed by a healthy one. +func staleThenFresh(f *fakeGitHub, path string) *atomic.Int64 { + var calls atomic.Int64 + f.routes[path] = func(w http.ResponseWriter, r *http.Request) { + n := calls.Add(1) + if n == 1 { + _, _ = w.Write([]byte(`{"workflow_runs":[ + {"id":1,"html_url":"old","event":"schedule","head_branch":"main","status":"completed","conclusion":"failure","created_at":"2026-09-11T03:02:00Z"}]}`)) + return + } + if r.Header.Get("If-None-Match") != "" { + http.Error(w, "fresh requests must not be conditional", http.StatusBadRequest) + return + } + _, _ = w.Write([]byte(`{"workflow_runs":[ + {"id":3,"html_url":"new","event":"schedule","head_branch":"main","status":"completed","conclusion":"success","created_at":"2026-10-02T03:03:00Z"}, + {"id":1,"html_url":"old","event":"schedule","head_branch":"main","status":"completed","conclusion":"failure","created_at":"2026-09-11T03:02:00Z"}]}`)) + } + return &calls +} + +func TestStaleFailureIsConfirmedOnANewConnection(t *testing.T) { + f := newFake(t) + c := f.client(t, 10) + f.json("/repos/o/r/actions/workflows", `{"workflows":[{"id":7,"name":"Autoupdate","html_url":"w","state":"active"}]}`) + calls := staleThenFresh(f, "/repos/o/r/actions/workflows/7/runs") + wfs, err := c.workflows(context.Background(), "o/r", "main") + if err != nil { + t.Fatal(err) + } + if len(wfs) != 1 || wfs[0].Latest.ID != 3 || wfs[0].LastDone == nil || wfs[0].LastDone.Failed() { + t.Fatalf("the newer success must supersede the stale failure: %+v", wfs) + } + if calls.Load() != 2 { + t.Fatalf("runs requested %d times, want 2", calls.Load()) + } +} + +func TestHealthyWorkflowIsNotRequestedTwice(t *testing.T) { + f := newFake(t) + c := f.client(t, 10) + f.json("/repos/o/r/actions/workflows", `{"workflows":[{"id":7,"name":"CI","html_url":"w","state":"active"}]}`) + var calls atomic.Int64 + f.routes["/repos/o/r/actions/workflows/7/runs"] = func(w http.ResponseWriter, _ *http.Request) { + calls.Add(1) + _, _ = w.Write([]byte(`{"workflow_runs":[{"id":3,"event":"push","head_branch":"main","status":"completed","conclusion":"success","created_at":"2026-10-02T03:03:00Z"}]}`)) + } + if _, err := c.workflows(context.Background(), "o/r", "main"); err != nil { + t.Fatal(err) + } + if calls.Load() != 1 { + t.Fatalf("a passing workflow needs one request, got %d", calls.Load()) + } +} + +func TestConfirmedFailureStaysFailing(t *testing.T) { + f := newFake(t) + c := f.client(t, 10) + f.json("/repos/o/r/actions/workflows", `{"workflows":[{"id":7,"name":"CI","html_url":"w","state":"active"}]}`) + f.json("/repos/o/r/actions/workflows/7/runs", `{"workflow_runs":[{"id":3,"event":"push","head_branch":"main","status":"completed","conclusion":"failure","created_at":"2026-10-02T03:03:00Z"}]}`) + wfs, err := c.workflows(context.Background(), "o/r", "main") + if err != nil || len(wfs) != 1 || wfs[0].LastDone == nil || !wfs[0].LastDone.Failed() { + t.Fatalf("a failure that both answers agree on must stay: %+v %v", wfs, err) + } +} + +func TestMergeRunsKeepsEachRunOnce(t *testing.T) { + got := mergeRuns([]rawRun{{ID: 1}, {ID: 2}}, []rawRun{{ID: 2}, {ID: 3}}) + if len(got) != 3 || got[0].ID != 1 || got[1].ID != 2 || got[2].ID != 3 { + t.Fatalf("%+v", got) + } +} + +func TestFreshClientDisablesConnectionReuse(t *testing.T) { + base := &http.Client{Transport: &http.Transport{}, Timeout: 5 * time.Second} + fc := freshClient(base) + tr, ok := fc.Transport.(*http.Transport) + if !ok || !tr.DisableKeepAlives || fc.Timeout != 5*time.Second { + t.Fatalf("fresh client: %+v", fc) + } + if base.Transport.(*http.Transport).DisableKeepAlives { + t.Fatal("the original client must keep its connections") + } + custom := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { return nil, errors.New("x") })} + if freshClient(custom) != custom { + t.Fatal("non-standard transports are used as they are") + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }