watchpr: detect merge/close in poll loop (was hanging after baseline)

watchpr only reported meaningful changes as transitions from the poll baseline. A PR already merged or closed when the watch started was captured as the baseline and never produced a transition, so the loop polled the dead PR forever (process alive, never exiting) -- the single-PR --interval case observed in production. Add a terminal-state check applied to the baseline snapshot: a PR that is already merged or closed the moment watchpr starts is reported and exits immediately, since it can never change again. Extract the baseline+poll loop into agent.Watch behind a StateFetcher interface so the loop, its open->merged/close detection, and its poll-error resilience are unit-testable with a fake client.
2026-08-12 23:32:28 +10:00
parent 7d9ae0fcc6
commit f915b5ba3b
3 changed files with 237 additions and 27 deletions
@@ -117,36 +117,23 @@ func runOnce(c *agent.GiteaClient, refs []agent.PRRef, jsonMode bool) error {
 func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration, jsonMode bool) error {
 	login := agent.AgentLogin()

-	prev := make(map[string]agent.PRState, len(refs))
-	for _, ref := range refs {
-		st, err := agent.FetchState(c, ref, login)
-		if err != nil {
-			return err
-		}
-		prev[ref.String()] = st
-	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+
+	onBaseline := func() {
 		if !jsonMode {
 			fmt.Fprintf(os.Stderr, "watching %d PR(s) every %s; baseline established\n", len(refs), interval)
 		}
+	}
+	onError := func(ref agent.PRRef, err error) {
+		fmt.Fprintf(os.Stderr, "warning: polling %s: %v\n", ref.String(), err)
+	}

-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
-	for range ticker.C {
-		for _, ref := range refs {
-			key := ref.String()
-			cur, err := agent.FetchState(c, ref, login)
+	res, err := agent.Watch(c, refs, login, ticker.C, onBaseline, onError)
 	if err != nil {
-				fmt.Fprintf(os.Stderr, "warning: polling %s: %v\n", key, err)
-				continue
-			}
-			changed, reason := agent.MeaningfulChange(prev[key], cur)
-			if changed {
-				report(key, reason, cur, jsonMode)
-				return nil
-			}
-			prev[key] = cur
-		}
+		return err
 	}
+	report(res.Ref.String(), res.Reason, res.State, jsonMode)
 	return nil
 }

@@ -1,5 +1,7 @@
 package agent

+import "time"
+
 // PRState is a point-in-time snapshot of the PR attributes watchpr tracks.
 type PRState struct {
 	Ref              PRRef  `json:"ref"`
@@ -41,6 +43,78 @@ func FetchState(c *GiteaClient, ref PRRef, agentLogin string) (PRState, error) {
 	}, nil
 }

+// StateFetcher fetches the current PRState for a ref. *GiteaClient satisfies it
+// via its FetchState method; tests inject fakes.
+type StateFetcher interface {
+	FetchState(ref PRRef, agentLogin string) (PRState, error)
+}
+
+// FetchState makes *GiteaClient a StateFetcher.
+func (c *GiteaClient) FetchState(ref PRRef, agentLogin string) (PRState, error) {
+	return FetchState(c, ref, agentLogin)
+}
+
+// WatchResult is the change that ended a watch.
+type WatchResult struct {
+	Ref    PRRef
+	Reason string
+	State  PRState
+}
+
+// terminalState reports whether a PR has reached a final state from which no
+// further meaningful change is possible, with a human-readable reason. Unlike a
+// transition (see MeaningfulChange) this holds for a single snapshot, so it also
+// catches a PR that is already merged/closed the moment watchpr starts.
+func terminalState(st PRState) (bool, string) {
+	if st.Merged {
+		return true, "PR merged"
+	}
+	if st.State == "closed" {
+		return true, "PR closed without merging"
+	}
+	return false, ""
+}
+
+// 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; only a baseline fetch error aborts. 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 {
+		st, err := f.FetchState(ref, agentLogin)
+		if err != nil {
+			return WatchResult{}, err
+		}
+		if terminal, reason := terminalState(st); terminal {
+			return WatchResult{Ref: ref, Reason: reason, State: st}, nil
+		}
+		prev[ref.String()] = st
+	}
+	if onBaseline != nil {
+		onBaseline()
+	}
+	for range ticks {
+		for _, ref := range refs {
+			key := ref.String()
+			cur, err := f.FetchState(ref, agentLogin)
+			if err != nil {
+				if onError != nil {
+					onError(ref, err)
+				}
+				continue
+			}
+			if changed, reason := MeaningfulChange(prev[key], cur); changed {
+				return WatchResult{Ref: ref, Reason: reason, State: cur}, nil
+			}
+			prev[key] = cur
+		}
+	}
+	return WatchResult{}, nil
+}
+
 // countNonAgentComments counts comments authored by anyone other than agentLogin.
 func countNonAgentComments(comments []Comment, agentLogin string) int {
 	n := 0
@@ -1,6 +1,10 @@
 package agent

-import "testing"
+import (
+	"errors"
+	"testing"
+	"time"
+)

 func base() PRState {
 	return PRState{
@@ -124,6 +128,151 @@ func TestMeaningfulChangeStaysFailed(t *testing.T) {
 	}
 }

+// fakeFetcher returns a scripted sequence of (state, error) results per call,
+// so tests can drive Watch across baseline and successive polls.
+type fakeFetcher struct {
+	states []PRState
+	errs   []error
+	calls  int
+}
+
+func (f *fakeFetcher) FetchState(ref PRRef, agentLogin string) (PRState, error) {
+	i := f.calls
+	if i >= len(f.states) {
+		i = len(f.states) - 1
+	}
+	f.calls++
+	var err error
+	if f.calls-1 < len(f.errs) {
+		err = f.errs[f.calls-1]
+	}
+	return f.states[i], err
+}
+
+func TestTerminalState(t *testing.T) {
+	open := base()
+	open.State = "open"
+	if term, _ := terminalState(open); term {
+		t.Errorf("open PR should not be terminal")
+	}
+
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	if term, reason := terminalState(merged); !term || reason != "PR merged" {
+		t.Errorf("merged PR: got (%v, %q), want (true, %q)", term, reason, "PR merged")
+	}
+
+	closed := base()
+	closed.State = "closed"
+	if term, reason := terminalState(closed); !term || reason != "PR closed without merging" {
+		t.Errorf("closed PR: got (%v, %q), want (true, %q)", term, reason, "PR closed without merging")
+	}
+}
+
+// The production hang: a PR that is already merged when watchpr starts must be
+// reported at baseline and exit, without ever consuming a tick. Before the fix,
+// Watch only reported transitions, so a terminal baseline was polled forever.
+func TestWatchExitsWhenAlreadyMergedAtBaseline(t *testing.T) {
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	f := &fakeFetcher{states: []PRState{merged}}
+
+	ticks := make(chan time.Time) // never fires; a hang would block here
+	res, err := Watch(f, []PRRef{merged.Ref}, "unkin-agent", ticks, nil, nil)
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if res.Reason != "PR merged" {
+		t.Errorf("reason = %q, want %q", res.Reason, "PR merged")
+	}
+	if f.calls != 1 {
+		t.Errorf("fetch calls = %d, want 1 (baseline only)", f.calls)
+	}
+}
+
+// A PR already closed-without-merge at baseline must also exit immediately.
+func TestWatchExitsWhenAlreadyClosedAtBaseline(t *testing.T) {
+	closed := base()
+	closed.State = "closed"
+	f := &fakeFetcher{states: []PRState{closed}}
+
+	ticks := make(chan time.Time)
+	res, err := Watch(f, []PRRef{closed.Ref}, "unkin-agent", ticks, nil, nil)
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if res.Reason != "PR closed without merging" {
+		t.Errorf("reason = %q, want %q", res.Reason, "PR closed without merging")
+	}
+}
+
+// An open→merged transition observed during polling must be detected and end
+// the watch.
+func TestWatchDetectsMergeAfterBaseline(t *testing.T) {
+	open := base()
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	f := &fakeFetcher{states: []PRState{open, merged}} // baseline open, then merged
+
+	baselineFired := false
+	ticks := make(chan time.Time, 1)
+	ticks <- time.Now()
+	res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent",
+		ticks, func() { baselineFired = true }, nil)
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if !baselineFired {
+		t.Errorf("onBaseline should fire for an open baseline")
+	}
+	if res.Reason != "PR merged" {
+		t.Errorf("reason = %q, want %q", res.Reason, "PR merged")
+	}
+}
+
+// A transient poll error must be reported and the loop must keep polling; a
+// merge on the following tick still ends the watch.
+func TestWatchContinuesPastPollError(t *testing.T) {
+	open := base()
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	// baseline ok, first poll errors, second poll sees the merge.
+	f := &fakeFetcher{
+		states: []PRState{open, open, merged},
+		errs:   []error{nil, errors.New("HTTP 502"), nil},
+	}
+
+	var gotErr error
+	ticks := make(chan time.Time, 2)
+	ticks <- time.Now()
+	ticks <- time.Now()
+	res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil,
+		func(_ PRRef, e error) { gotErr = e })
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if gotErr == nil {
+		t.Errorf("onError should have received the transient poll error")
+	}
+	if res.Reason != "PR merged" {
+		t.Errorf("reason = %q, want %q (loop must survive the error)", res.Reason, "PR merged")
+	}
+}
+
+// A baseline fetch error aborts the watch (nothing to establish a baseline
+// from), unlike a mid-loop poll error.
+func TestWatchBaselineErrorAborts(t *testing.T) {
+	f := &fakeFetcher{states: []PRState{base()}, errs: []error{errors.New("HTTP 500")}}
+	ticks := make(chan time.Time)
+	if _, err := Watch(f, []PRRef{base().Ref}, "unkin-agent", ticks, nil, nil); err == nil {
+		t.Fatal("Watch should return the baseline fetch error")
+	}
+}
+
 func TestCountNonAgentComments(t *testing.T) {
 	comments := []Comment{
 		{User: User{Login: "unkin-agent"}},