From f915b5ba3b9a01c1063bfdb224dbd79bafb0f2d6 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Wed, 12 Aug 2026 23:32:28 +1000 Subject: [PATCH] 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. --- cmd/watchpr/main.go | 39 +++------ internal/agent/watch.go | 74 +++++++++++++++++ internal/agent/watch_test.go | 151 ++++++++++++++++++++++++++++++++++- 3 files changed, 237 insertions(+), 27 deletions(-) diff --git a/cmd/watchpr/main.go b/cmd/watchpr/main.go index 68952f1..4c1482f 100644 --- a/cmd/watchpr/main.go +++ b/cmd/watchpr/main.go @@ -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 - } - if !jsonMode { - fmt.Fprintf(os.Stderr, "watching %d PR(s) every %s; baseline established\n", len(refs), interval) - } - 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) - 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 + + 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) + } + + res, err := agent.Watch(c, refs, login, ticker.C, onBaseline, onError) + if err != nil { + return err + } + report(res.Ref.String(), res.Reason, res.State, jsonMode) return nil } diff --git a/internal/agent/watch.go b/internal/agent/watch.go index da320f8..23aa239 100644 --- a/internal/agent/watch.go +++ b/internal/agent/watch.go @@ -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 diff --git a/internal/agent/watch_test.go b/internal/agent/watch_test.go index 280b732..dab44fa 100644 --- a/internal/agent/watch_test.go +++ b/internal/agent/watch_test.go @@ -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"}},