mirror of
https://github.com/lukaszraczylo/ghtracker.git
synced 2026-10-13 00:48:42 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0c7c782d86 |
+41
-8
@@ -66,6 +66,8 @@ type TokenSource interface {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type Client struct {
|
type Client struct {
|
||||||
|
// freshHC opens a new connection for every request.
|
||||||
|
freshHC *http.Client
|
||||||
rateReset time.Time
|
rateReset time.Time
|
||||||
blockedUntil time.Time
|
blockedUntil time.Time
|
||||||
auth TokenSource
|
auth TokenSource
|
||||||
@@ -88,7 +90,7 @@ type cachedResponse struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func NewClient(baseURL string, auth TokenSource, hc *http.Client, maxItems int) *Client {
|
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)
|
c.rateRemaining.Store(-1)
|
||||||
return c
|
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.
|
// 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
|
var lastErr error
|
||||||
for attempt := 0; attempt <= maxRetries; attempt++ {
|
for attempt := 0; attempt <= maxRetries; attempt++ {
|
||||||
if attempt > 0 {
|
if attempt > 0 {
|
||||||
@@ -119,7 +123,7 @@ func (c *Client) do(ctx context.Context, repo, target string) ([]byte, string, e
|
|||||||
return nil, "", err
|
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 {
|
if err == nil {
|
||||||
return body, link, 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
|
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 {
|
if gerr := c.guard(); gerr != nil {
|
||||||
return nil, "", false, gerr
|
return nil, "", false, gerr
|
||||||
}
|
}
|
||||||
@@ -150,10 +154,15 @@ func (c *Client) once(ctx context.Context, repo, target string) (body []byte, li
|
|||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
cached, haveCache := c.cache[target]
|
cached, haveCache := c.cache[target]
|
||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
|
haveCache = haveCache && !fresh
|
||||||
if haveCache {
|
if haveCache {
|
||||||
req.Header.Set("If-None-Match", cached.etag)
|
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 {
|
if err != nil {
|
||||||
return nil, "", ctx.Err() == nil, err
|
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
|
return nil, "", resp.StatusCode >= 500, ae
|
||||||
}
|
}
|
||||||
link = resp.Header.Get("Link")
|
link = resp.Header.Get("Link")
|
||||||
if etag := resp.Header.Get("ETag"); etag != "" {
|
if etag := resp.Header.Get("ETag"); etag != "" && !fresh {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
c.cache[target] = cachedResponse{etag: etag, body: data, link: link}
|
c.cache[target] = cachedResponse{etag: etag, body: data, link: link}
|
||||||
c.mu.Unlock()
|
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 {
|
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 {
|
if err != nil {
|
||||||
return err
|
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)
|
next := c.url(path, q)
|
||||||
var all []T
|
var all []T
|
||||||
for next != "" && len(all) < limit {
|
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 {
|
if err != nil {
|
||||||
return all, err
|
return all, err
|
||||||
}
|
}
|
||||||
@@ -289,3 +307,18 @@ func nextLink(h string) string {
|
|||||||
}
|
}
|
||||||
return ""
|
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}
|
||||||
|
}
|
||||||
|
|||||||
+49
-12
@@ -262,24 +262,28 @@ func (c *Client) workflows(ctx context.Context, full, branch string) ([]model.Wo
|
|||||||
if w.State != "active" || len(out) >= maxWorkflows {
|
if w.State != "active" || len(out) >= maxWorkflows {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
var runs struct {
|
runs, err := c.workflowRuns(ctx, full, w.ID, branch, false)
|
||||||
WorkflowRuns []rawRun `json:"workflow_runs"`
|
if err != nil {
|
||||||
}
|
|
||||||
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 {
|
|
||||||
if first == nil {
|
if first == nil {
|
||||||
first = fmt.Errorf("%s: %w", w.Name, err)
|
first = fmt.Errorf("%s: %w", w.Name, err)
|
||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if wf, ok := buildWorkflow(w.Name, w.HTMLURL, runs.WorkflowRuns); ok {
|
wf, ok := buildWorkflow(w.Name, w.HTMLURL, runs)
|
||||||
out = append(out, wf)
|
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
|
return out, first
|
||||||
}
|
}
|
||||||
@@ -343,3 +347,36 @@ func (c *Client) latestRelease(ctx context.Context, full, repoURL string) (*mode
|
|||||||
FromTag: true,
|
FromTag: true,
|
||||||
}, nil
|
}, 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
|
||||||
|
}
|
||||||
|
|||||||
@@ -459,3 +459,97 @@ func TestAPIErrorMessageParsing(t *testing.T) {
|
|||||||
t.Fatalf("len=%d", len(got))
|
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) }
|
||||||
|
|||||||
Reference in New Issue
Block a user