Merge pull request 'watchpr: detect merge/close in poll loop (was hanging after baseline)' (#3) from benvin/watchpr-poll-loop-fix into main

Reviewed-on: #3

AGENTS.md

@@ -53,7 +53,7 @@ Config via env (all have defaults):
 | Variable | Default | Purpose |
 |---|---|---|
 | `VAULT_ADDR` | `https://vault.service.consul:8200` | Vault/OpenBao address |
-| `AGENT_APPROLE_ROLE_ID` | `ababbcd3-9c77-5c6a-be2d-287fce9214a6` | AppRole role_id |
+| `AGENT_APPROLE_ROLE_ID` | built-in default | AppRole role_id (overridable) |
 | `GITEA_URL` | `https://git.unkin.net` | Gitea base URL |
 | `AGENT_LOGIN` | `unkin-agent` | login whose comments watchpr ignores |
 

README.md

@@ -20,7 +20,7 @@ Everything is configured by environment variables, all with defaults:
 | Variable | Default | Purpose |
 |---|---|---|
 | `VAULT_ADDR` | `https://vault.service.consul:8200` | Vault/OpenBao address |
-| `AGENT_APPROLE_ROLE_ID` | `ababbcd3-9c77-5c6a-be2d-287fce9214a6` | AppRole role_id |
+| `AGENT_APPROLE_ROLE_ID` | built-in default | AppRole role_id (overridable) |
 | `GITEA_URL` | `https://git.unkin.net` | Gitea base URL |
 | `AGENT_LOGIN` | `unkin-agent` | login whose comments `watchpr` ignores |
 

cmd/agentpr/main.go

@@ -20,6 +20,17 @@ import (
 var version = "dev"
 
 func main() {
+	// cobra prints the error itself (SilenceErrors stays off); we only need to
+	// turn any command error into a non-zero exit.
+	if err := newRootCmd().Execute(); err != nil {
+		os.Exit(1)
+	}
+}
+
+// newRootCmd builds the agentpr command tree. It is separated from main so
+// tests can invoke Execute and assert the exit behaviour without spawning a
+// process.
+func newRootCmd() *cobra.Command {
 	root := &cobra.Command{
 		Use:          "agentpr",
 		Short:        "Manage Gitea PRs and comments as the unkin-agent user.",
@@ -30,10 +41,7 @@ func main() {
 	root.SetVersionTemplate("{{.Version}}\n")
 
 	root.AddCommand(newPRCmd(), newWhoamiCmd(), newVersionCmd())
-
-	if err := root.Execute(); err != nil {
-		os.Exit(1)
-	}
+	return root
 }
 
 // client mints a Gitea token via Vault and returns a ready client.

cmd/agentpr/main_test.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"io"
+	"testing"
+)
+
+// A malformed --repo must fail the command (so main exits non-zero). ParseRepo
+// rejects it before any Vault/Gitea call, so this stays hermetic.
+func TestExecuteBadRepoErrors(t *testing.T) {
+	cmd := newRootCmd()
+	cmd.SetArgs([]string{"pr", "create", "--repo", "not-a-repo", "--base", "main", "--head", "x", "--title", "t"})
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	if err := cmd.Execute(); err == nil {
+		t.Fatal("Execute() = nil, want error for a malformed --repo")
+	}
+}

cmd/watchpr/main.go

@@ -23,6 +23,17 @@ import (
 var version = "dev"
 
 func main() {
+	// cobra prints the error itself (SilenceErrors stays off); we only need to
+	// turn any command error into a non-zero exit.
+	if err := newRootCmd().Execute(); err != nil {
+		os.Exit(1)
+	}
+}
+
+// newRootCmd builds the watchpr command tree. It is separated from main so
+// tests can invoke Execute and assert the exit behaviour without spawning a
+// process.
+func newRootCmd() *cobra.Command {
 	var interval time.Duration
 	var once, jsonMode bool
 
@@ -70,10 +81,7 @@ func main() {
 		Run:          func(cmd *cobra.Command, args []string) { fmt.Println(version) },
 		SilenceUsage: true,
 	})
-
-	if err := root.Execute(); err != nil {
-		os.Exit(1)
-	}
+	return root
 }
 
 func clientFor() (*agent.GiteaClient, error) {
@@ -109,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
 }
 

cmd/watchpr/main_test.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"io"
+	"testing"
+)
+
+// A bad PR reference must fail the command (so main exits non-zero) rather than
+// return nil. Parsing rejects the ref before any Vault/Gitea call, so this stays
+// hermetic.
+func TestExecuteBadRefErrors(t *testing.T) {
+	cmd := newRootCmd()
+	cmd.SetArgs([]string{"--once", "not-a-ref"})
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	if err := cmd.Execute(); err == nil {
+		t.Fatal("Execute() = nil, want error for a bad PR reference")
+	}
+}
+
+// No arguments is also an error (nothing to watch).
+func TestExecuteNoArgsErrors(t *testing.T) {
+	cmd := newRootCmd()
+	cmd.SetArgs(nil)
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	if err := cmd.Execute(); err == nil {
+		t.Fatal("Execute() = nil, want error when no PR references are given")
+	}
+}

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
@@ -67,7 +141,7 @@ func isFailedCI(state string) bool {
 //   - the PR closed without merging
 //   - a new comment from someone other than the agent
 //   - CI transitioned into failure/error
-//   - the PR lost mergeability (a conflict appeared)
+//   - the PR lost mergeability (a conflict appeared) for two consecutive polls
 func MeaningfulChange(prev, cur PRState) (bool, string) {
 	if !prev.Merged && cur.Merged {
 		return true, "PR merged"
@@ -82,7 +156,11 @@ func MeaningfulChange(prev, cur PRState) (bool, string) {
 	if isFailedCI(cur.CIStatus) && !isFailedCI(prev.CIStatus) {
 		return true, "CI failed (" + cur.CIStatus + ")"
 	}
-	if prev.Mergeable && !cur.Mergeable && cur.State == "open" {
+	// Gitea computes mergeability asynchronously, so a PR can briefly report
+	// mergeable=false right after a push. Require the loss to persist across two
+	// consecutive polls (both prev and cur false, still open) before treating it
+	// as a real conflict; a single false poll is debounced.
+	if !prev.Mergeable && !cur.Mergeable && cur.State == "open" {
 		return true, "PR lost mergeability (conflict)"
 	}
 	return false, ""

internal/agent/watch_test.go

@@ -1,6 +1,10 @@
 package agent
 
-import "testing"
+import (
+	"errors"
+	"testing"
+	"time"
+)
 
 func base() PRState {
 	return PRState{
@@ -17,6 +21,7 @@ func base() PRState {
 func TestMeaningfulChange(t *testing.T) {
 	tests := []struct {
 		name       string
+		mutatePrev func(s *PRState)
 		mutate     func(s *PRState)
 		wantChange bool
 	}{
@@ -56,10 +61,27 @@ func TestMeaningfulChange(t *testing.T) {
 			wantChange: true,
 		},
 		{
-			name:       "lost mergeability alerts",
+			// A single mergeable=false poll is debounced: Gitea often reports
+			// this transiently right after a push.
+			name:       "mergeable true to false for one poll is benign",
+			mutate:     func(s *PRState) { s.Mergeable = false },
+			wantChange: false,
+		},
+		{
+			// mergeable=false persisting into a second consecutive poll is a
+			// real conflict and alerts.
+			name:       "mergeable false persisting a second poll alerts",
+			mutatePrev: func(s *PRState) { s.Mergeable = false },
 			mutate:     func(s *PRState) { s.Mergeable = false },
 			wantChange: true,
 		},
+		{
+			// mergeable recovered (false then true) must not alert.
+			name:       "mergeable recovered false to true is benign",
+			mutatePrev: func(s *PRState) { s.Mergeable = false },
+			mutate:     func(s *PRState) {},
+			wantChange: false,
+		},
 		{
 			name:       "new head sha alone is benign",
 			mutate:     func(s *PRState) { s.HeadSHA = "def456" },
@@ -69,6 +91,9 @@ func TestMeaningfulChange(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			prev := base()
+			if tt.mutatePrev != nil {
+				tt.mutatePrev(&prev)
+			}
 			cur := base()
 			tt.mutate(&cur)
 			got, reason := MeaningfulChange(prev, cur)
@@ -103,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"}},