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

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) {