From 46dfe48adc0b21ba033320c8d456e877b4e94726 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Wed, 9 Sep 2026 22:41:27 +1000 Subject: [PATCH] Abort watchpr when a poll can no longer see the PR - treat a mid-run 404 on a tracked PR as terminal - cap consecutive transient poll failures at 20 per PR - reset the failure count on a successful poll - export IsNotFound for callers to classify the abort --- cmd/watchpr/main.go | 3 + internal/agent/gitea.go | 6 +- internal/agent/watch.go | 27 +++-- internal/agent/watch_test.go | 193 +++++++++++++++++++++++++++++++++++ 4 files changed, 220 insertions(+), 9 deletions(-) diff --git a/cmd/watchpr/main.go b/cmd/watchpr/main.go index f26db99..8ba1f4d 100644 --- a/cmd/watchpr/main.go +++ b/cmd/watchpr/main.go @@ -135,6 +135,9 @@ func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration, if agent.IsAuthError(err) { return fmt.Errorf("gitea authentication failed after re-minting the token, watch aborted: %w", err) } + if agent.IsNotFound(err) { + return fmt.Errorf("PR no longer visible (repo deleted, renamed, or made private), watch aborted: %w", err) + } return err } report(res.Ref.String(), res.Reason, res.State, jsonMode) diff --git a/internal/agent/gitea.go b/internal/agent/gitea.go index 3dcf518..840417d 100644 --- a/internal/agent/gitea.go +++ b/internal/agent/gitea.go @@ -24,8 +24,10 @@ func (e *APIError) Error() string { return fmt.Sprintf("gitea %s %s: HTTP %d: %s", e.Method, e.Path, e.StatusCode, e.Body) } -// isNotFound reports whether err is a Gitea 404. -func isNotFound(err error) bool { +// IsNotFound reports whether err is a Gitea 404. Gitea hides repositories a +// caller may not see behind a 404 rather than a 403, so this also covers a repo +// that was renamed, deleted, or made private. +func IsNotFound(err error) bool { var apiErr *APIError return errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound } diff --git a/internal/agent/watch.go b/internal/agent/watch.go index 72ed33b..c194e71 100644 --- a/internal/agent/watch.go +++ b/internal/agent/watch.go @@ -29,7 +29,7 @@ func FetchState(c *GiteaClient, ref PRRef, agentLogin string) (PRState, error) { // rebase merge); the PR object is still authoritative, so treat CI as absent // rather than discarding the merge signal and hanging the watch loop. ci, err := c.CommitStatus(ref.RepoPath(), pr.Head.Sha) - if err != nil && !isNotFound(err) { + if err != nil && !IsNotFound(err) { return PRState{}, err } comments, err := c.ListComments(ref.RepoPath(), ref.Number) @@ -81,14 +81,21 @@ func terminalState(st PRState) (bool, string) { return false, "" } +// MaxPollFailures is how many consecutive failed polls of the same PR are +// tolerated before Watch gives up. At watchpr's default 60s interval that rides +// out a 20-minute outage. +const MaxPollFailures = 20 + // Watch establishes a baseline for each ref, then polls on every tick until a // tracked PR changes meaningfully, returning the first such change. A PR that is // already terminal (merged/closed) at baseline is reported immediately rather -// than polled forever. Poll errors are handed to onError and never stop the -// loop; a baseline fetch error and an authentication failure (the token was -// rejected and re-minting it did not help) abort instead, because a watcher that -// cannot authenticate sees nothing. onBaseline, if set, fires once after all -// baselines are captured and before the first tick. +// than polled forever. Transient poll errors are handed to onError and the loop +// continues, but never blindly: a baseline fetch error, an authentication +// failure surviving a token re-mint, a 404 on a tracked PR (the repo is gone, +// renamed, or no longer visible), and MaxPollFailures consecutive failures of +// one PR all abort, because a watcher that sees nothing must not look healthy. +// onBaseline, if set, fires once after all baselines are captured and before the +// first tick. func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Time, onBaseline func(), onError func(PRRef, error)) (WatchResult, error) { prev := make(map[string]PRState, len(refs)) for _, ref := range refs { @@ -104,19 +111,25 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti if onBaseline != nil { onBaseline() } + fails := make(map[string]int, len(refs)) for range ticks { for _, ref := range refs { key := ref.String() cur, err := f.FetchState(ref, agentLogin) if err != nil { - if IsAuthError(err) { + if IsAuthError(err) || IsNotFound(err) { return WatchResult{}, fmt.Errorf("polling %s: %w", key, err) } + fails[key]++ if onError != nil { onError(ref, err) } + if fails[key] >= MaxPollFailures { + return WatchResult{}, fmt.Errorf("polling %s: giving up after %d consecutive failures: %w", key, fails[key], err) + } continue } + fails[key] = 0 if changed, reason := MeaningfulChange(prev[key], cur); changed { return WatchResult{Ref: ref, Reason: reason, State: cur}, nil } diff --git a/internal/agent/watch_test.go b/internal/agent/watch_test.go index f4178a2..6230920 100644 --- a/internal/agent/watch_test.go +++ b/internal/agent/watch_test.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" "time" @@ -421,6 +422,198 @@ func TestWatchContinuesPastServerError(t *testing.T) { } } +// The production failure: a watched repo was renamed mid-watch, so every poll +// 404'd (Gitea hides a repo the caller may not see rather than 403ing) and the +// loop warned past it forever while reporting nothing. A 404 on a tracked PR +// must end the watch with an error naming that PR. +func TestWatchAbortsOnMidRunNotFound(t *testing.T) { + const sha = "deadbeefdeadbeef" + var polls atomic.Int32 + + mux := http.NewServeMux() + mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) { + if polls.Add(1) > 1 { // repo renamed/made private after the baseline + w.WriteHeader(http.StatusNotFound) + _, _ = fmt.Fprint(w, `{"message":"Not Found"}`) + return + } + _, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha) + }) + mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprint(w, `{"state":"success"}`) + }) + mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprint(w, `[]`) + }) + srv := httptest.NewServer(mux) + defer srv.Close() + + c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()} + ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7} + + tk := time.NewTicker(5 * time.Millisecond) + defer tk.Stop() + var warned atomic.Int32 + done := make(chan error, 1) + go func() { + _, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil, + func(PRRef, error) { warned.Add(1) }) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatal("Watch should abort on a mid-run 404, not keep polling") + } + if !IsNotFound(err) { + t.Errorf("Watch error = %v, want a 404", err) + } + if !strings.Contains(err.Error(), ref.String()) { + t.Errorf("Watch error = %v, want it to name %s", err, ref.String()) + } + if n := warned.Load(); n != 0 { + t.Errorf("404 was logged as a warning %d time(s); it must abort", n) + } + case <-time.After(3 * time.Second): + t.Fatal("Watch hung: a vanished repo was warned past instead of aborting") + } +} + +// A 5xx blip must not kill a long watch: it warns, keeps polling, and still +// catches the merge that lands afterwards. +func TestWatchSurvivesTransientServerError(t *testing.T) { + const sha = "feedfacefeedface" + var polls atomic.Int32 + + mux := http.NewServeMux() + mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) { + switch n := polls.Add(1); { + case n == 2 || n == 3: // gateway blip across two polls + w.WriteHeader(http.StatusBadGateway) + _, _ = fmt.Fprint(w, `bad gateway`) + case n >= 4: + _, _ = fmt.Fprintf(w, `{"number":7,"state":"closed","merged":true,"mergeable":true,"head":{"sha":%q}}`, sha) + default: + _, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha) + } + }) + mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprint(w, `{"state":"success"}`) + }) + mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprint(w, `[]`) + }) + srv := httptest.NewServer(mux) + defer srv.Close() + + c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()} + ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7} + + tk := time.NewTicker(5 * time.Millisecond) + defer tk.Stop() + var warned atomic.Int32 + type outcome struct { + res WatchResult + err error + } + done := make(chan outcome, 1) + go func() { + res, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil, + func(PRRef, error) { warned.Add(1) }) + done <- outcome{res, err} + }() + + select { + case o := <-done: + if o.err != nil { + t.Fatalf("Watch: %v", o.err) + } + if o.res.Reason != "PR merged" { + t.Errorf("reason = %q, want %q", o.res.Reason, "PR merged") + } + if n := warned.Load(); n != 2 { + t.Errorf("warnings = %d, want 2", n) + } + case <-time.After(3 * time.Second): + t.Fatal("Watch hung: a transient 5xx must not stop the watch") + } +} + +// pollFailures scripts a fetcher whose polls fail with a 502 at the given call +// indexes (0 is the baseline); the final call returns merged. +func pollFailures(calls int, failAt map[int]bool) *fakeFetcher { + open, merged := base(), base() + merged.State = "closed" + merged.Merged = true + f := &fakeFetcher{states: make([]PRState, calls), errs: make([]error, calls)} + for i := range calls { + f.states[i] = open + if failAt[i] { + f.errs[i] = &APIError{Method: "GET", Path: "/p", StatusCode: 502, Body: "bad gateway"} + } + } + f.states[calls-1] = merged + return f +} + +// A permanently wedged endpoint (5xx forever) must eventually give up instead of +// warning on every tick for the life of the process. +func TestWatchAbortsAfterConsecutiveFailures(t *testing.T) { + failAt := map[int]bool{} + for i := 1; i <= MaxPollFailures; i++ { + failAt[i] = true + } + f := pollFailures(MaxPollFailures+1, failAt) + + warned := 0 + ticks := make(chan time.Time, MaxPollFailures) + for range MaxPollFailures { + ticks <- time.Now() + } + close(ticks) + _, err := Watch(f, []PRRef{base().Ref}, "unkin-agent", ticks, nil, func(PRRef, error) { warned++ }) + if err == nil { + t.Fatal("Watch should give up once the failures stop being transient") + } + if !strings.Contains(err.Error(), "consecutive failures") { + t.Errorf("Watch error = %v, want it to report the failure cap", err) + } + if warned != MaxPollFailures { + t.Errorf("warnings = %d, want %d", warned, MaxPollFailures) + } + if f.calls != MaxPollFailures+1 { + t.Errorf("fetch calls = %d, want %d", f.calls, MaxPollFailures+1) + } +} + +// The cap counts consecutive failures only: a single successful poll clears it, +// so an intermittent endpoint is watched indefinitely and the merge is caught. +func TestWatchFailureCountResetsOnSuccess(t *testing.T) { + const runs = MaxPollFailures - 1 + failAt := map[int]bool{} + for i := 1; i <= runs; i++ { // first run of failures + failAt[i] = true + } + for i := runs + 2; i <= 2*runs+1; i++ { // second run, after one good poll + failAt[i] = true + } + f := pollFailures(2*runs+3, failAt) + + ticks := make(chan time.Time, 2*runs+2) + for range 2*runs + 2 { + ticks <- time.Now() + } + close(ticks) + res, err := Watch(f, []PRRef{base().Ref}, "unkin-agent", ticks, nil, func(PRRef, error) {}) + if err != nil { + t.Fatalf("Watch: %v (a successful poll must reset the failure count)", err) + } + if res.Reason != "PR merged" { + t.Errorf("reason = %q, want %q", res.Reason, "PR merged") + } +} + // Anonymous watching of a public repo must poll on without a credential in // sight: no token, no mint, no exit until something actually changes. func TestWatchAnonymousKeepsPolling(t *testing.T) {