diff --git a/internal/agent/watch.go b/internal/agent/watch.go index b5e5282..5cf8488 100644 --- a/internal/agent/watch.go +++ b/internal/agent/watch.go @@ -145,9 +145,14 @@ func terminalState(st PRState) (bool, string) { return false, "" } -// conflictPolls is how many consecutive non-mergeable polls of an unchanged -// merge computation confirm a real conflict. -const conflictPolls = 2 +// conflictWindow is how long an unbroken run of non-mergeable polls must span +// before it is reported as a conflict. What the run has to outlast is Gitea's +// merge recompute, which is a duration, so counting polls cannot express it: +// --interval varies from seconds to hours between callers. Recomputes have been +// observed finishing in ~5-20s with no bound on the tail, so this is set well +// clear of that; the cost of it being too long is only a later conflict alert, +// while too short is a false one. +const conflictWindow = 2 * time.Minute // prWatch tracks one PR across polls, because mergeability needs more memory // than the previous snapshot. Gitea reports mergeable=false while it recomputes @@ -155,11 +160,15 @@ const conflictPolls = 2 // seen a merge computation start: the PR was mergeable at some point, or its // head or base SHA moved. A bare false inherited from the baseline says nothing // -- it is equally a conflict the operator is already waiting on and a recompute -// in flight -- so it arms nothing. +// in flight -- so it arms nothing. Arming and the run of falses are independent: +// movement only arms, because a base branch that moves under the PR on every +// push to main would otherwise restart the run forever. type prWatch struct { - prev PRState - armed bool - conflicts int + prev PRState + armed bool + // conflictSince is when the current unbroken run of non-mergeable polls + // began; zero when no run is in progress. + conflictSince time.Time } func newPRWatch(baseline PRState) *prWatch { @@ -173,39 +182,48 @@ func mergeInputsChanged(prev, cur PRState) bool { } // track folds one snapshot's mergeability into the run of observations. The -// polls that confirm a conflict must be adjacent, so anything but another -// non-mergeable observation breaks the run. -func (w *prWatch) track(st PRState) { +// observations that confirm a conflict must be adjacent, so anything but +// another non-mergeable one breaks the run. +func (w *prWatch) track(st PRState, now time.Time) { switch st.Mergeable { case MergeNo: - w.conflicts++ + if w.conflictSince.IsZero() { + w.conflictSince = now + } case MergeYes: w.armed = true - w.conflicts = 0 + w.conflictSince = time.Time{} default: - w.conflicts = 0 + w.conflictSince = time.Time{} } } // missed records a poll that never produced a snapshot; the run of adjacent // non-mergeable observations does not survive the gap. func (w *prWatch) missed() { - w.conflicts = 0 + w.conflictSince = time.Time{} } -// observe folds in the newest snapshot and reports whether the watch should end. -func (w *prWatch) observe(cur PRState) (bool, string) { +// confirmed reports whether the run of non-mergeable observations has spanned +// the recompute window. The run starts at its first observation, so a lone +// non-mergeable poll never confirms anything whatever the interval. +func (w *prWatch) confirmed(now time.Time) bool { + return !w.conflictSince.IsZero() && now.Sub(w.conflictSince) >= conflictWindow +} + +// observe folds in the newest snapshot, taken at now, and reports whether the +// watch should end. +func (w *prWatch) observe(cur PRState, now time.Time) (bool, string) { changed, reason := MeaningfulChange(w.prev, cur) if mergeInputsChanged(w.prev, cur) { w.armed = true - w.conflicts = 0 } w.prev = cur - w.track(cur) + w.track(cur, now) if changed { return true, reason } - if w.armed && w.conflicts >= conflictPolls && cur.State == "open" { + if w.armed && w.confirmed(now) && cur.State == "open" { return true, "PR lost mergeability (conflict)" } return false, "" @@ -245,7 +263,9 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti onBaseline(baselines) } fails := make(map[string]int, len(refs)) - for range ticks { + // The tick carries the time it fired, which is the clock the conflict + // window is measured on. + for now := range ticks { for _, ref := range refs { key := ref.String() cur, err := f.FetchState(ref, agentLogin) @@ -264,7 +284,7 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti continue } fails[key] = 0 - if changed, reason := watches[key].observe(cur); changed { + if changed, reason := watches[key].observe(cur, now); 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 7820072..630aa21 100644 --- a/internal/agent/watch_test.go +++ b/internal/agent/watch_test.go @@ -791,11 +791,21 @@ func TestWatchAnonymousKeepsPolling(t *testing.T) { } // drainableTicks returns a channel holding n ticks and already closed, so Watch -// polls exactly n times and then returns instead of blocking. +// polls exactly n times and then returns instead of blocking. The ticks are +// spaced far past conflictWindow, so a run of non-mergeable polls confirms on +// its second observation; spacedTicks drives the intervals where it must not. func drainableTicks(n int) <-chan time.Time { + return spacedTicks(n, 10*time.Minute) +} + +// spacedTicks is drainableTicks with the poll interval named. The tick carries +// the time Watch measures the conflict window on, which makes every wall-clock +// assertion in these tests exact and instant. +func spacedTicks(n int, interval time.Duration) <-chan time.Time { + start := time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC) ticks := make(chan time.Time, n) - for i := 0; i < n; i++ { - ticks <- time.Now() + for i := 1; i <= n; i++ { + ticks <- start.Add(time.Duration(i) * interval) } close(ticks) return ticks @@ -911,9 +921,10 @@ func TestPRWatchMergeability(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { w := newPRWatch(snap(tt.baseline)) + start := time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC) got := 0 for i, m := range tt.polls { - changed, reason := w.observe(snap(m)) + changed, reason := w.observe(snap(m), start.Add(time.Duration(i+1)*10*time.Minute)) if !changed { continue } @@ -958,13 +969,12 @@ func TestMergeabilityDecoding(t *testing.T) { } } -// The inverse of the bug this PR fixed: Gitea reports mergeable=false while it -// recomputes the merge base after a push, and watchpr is normally started right -// after the agent pushes, so the baseline can land inside that window. Demanding -// a mergeable observation to arm the rule left a genuinely conflicted PR silent -// forever. The new head is a new merge computation, so the conflict it confirms -// belongs to this watch. -func TestWatchAlertsWhenBaselineLandedInTheRecomputeWindow(t *testing.T) { +// The inverse of the bug this PR fixed: demanding a mergeable observation to arm +// the rule left a PR that was non-mergeable at baseline silent forever, even +// after a push gave Gitea a fresh merge computation to answer for. A head that +// moves during the watch arms the rule, so the conflict the new head keeps +// reporting is attributable to this watch and is reported. +func TestWatchAlertsWhenAPushArmsABaselineConflict(t *testing.T) { baseline := base() baseline.Mergeable = MergeNo // still recomputing; the push is not visible yet @@ -1160,3 +1170,164 @@ func TestWatchReportsBaselineStates(t *testing.T) { t.Errorf("baseline = %+v, want the conflicted, CI-red snapshot the watch started from", got[0]) } } + +// The residual gap, pinned so it is a decision rather than an accident: the +// agent pushes, the push conflicts, watchpr starts inside Gitea's recompute and +// nothing moves again. Every poll answers false and none of them is +// attributable to this watch, so no alert is ever sent -- the baseline line is +// the only notice. Gitea's payload carries no field that separates this from a +// merge check still running. +func TestWatchNeverAlertsOnAConflictLandedByThePushBeforeTheWatch(t *testing.T) { + pushed := base() + pushed.Mergeable = MergeNo + pushed.HeadSHA = "def456" + f := &fakeFetcher{states: []PRState{pushed}} + + res, err := Watch(f, []PRRef{pushed.Ref}, "unkin-agent", drainableTicks(50), nil, nil) + if err != nil { + t.Fatalf("Watch: %v", err) + } + if res.Reason != "" { + t.Fatalf("Watch ended with %q; nothing distinguishes this conflict from a merge check in flight, so it must stay silent", res.Reason) + } +} + +// The whole mergeability rule as a sequence table, driven through Watch at the +// intervals that decide it. The debounce is wall-clock, so the same sequence of +// snapshots must alert or stay silent according to how far apart the polls are. +func TestWatchConflictSequences(t *testing.T) { + snap := func(m Mergeability, head, bse string) PRState { + st := base() + st.Mergeable = m + st.HeadSHA = head + st.BaseSHA = bse + return st + } + repeat := func(st PRState, n int) []PRState { + out := make([]PRState, n) + for i := range out { + out[i] = st + } + return out + } + movingBase := func(m Mergeability, n int) []PRState { + out := make([]PRState, n) + for i := range out { + out[i] = snap(m, "h1", fmt.Sprintf("b%d", i)) + } + return out + } + const ( + quick = 5 * time.Second // well inside conflictWindow + normal = 60 * time.Second // watchpr's default + relaxed = 10 * time.Minute // past conflictWindow in a single gap + ) + const conflict = "PR lost mergeability (conflict)" + + yes := snap(MergeYes, "h1", "b1") + no := snap(MergeNo, "h1", "b1") + pushedNo := snap(MergeNo, "h2", "b1") + movedNo := snap(MergeNo, "h1", "b2") + + tests := []struct { + name string + interval time.Duration + baseline PRState + polls []PRState + errs []error + want string + }{ + { + name: "a push then the recompute's falses is not a conflict", + interval: normal, baseline: yes, polls: repeat(pushedNo, 2), want: "", + }, + { + name: "a fast poller rides out the whole recompute after a push", + interval: quick, baseline: yes, polls: repeat(pushedNo, 20), want: "", + }, + { + name: "a base move then the recompute's falses is not a conflict", + interval: normal, baseline: no, polls: repeat(movedNo, 2), want: "", + }, + { + name: "a base moving under every poll still confirms a conflict", + interval: normal, baseline: yes, polls: movingBase(MergeNo, 20), want: conflict, + }, + { + name: "a conflict that predates the watch stays silent forever", + interval: relaxed, baseline: no, polls: repeat(no, 50), want: "", + }, + { + name: "a sustained loss after a mergeable baseline alerts", + interval: normal, baseline: yes, polls: repeat(no, 3), want: conflict, + }, + { + name: "a mergeable poll breaks the run however long it ran", + interval: relaxed, baseline: yes, polls: []PRState{no, yes, no}, want: "", + }, + { + name: "an unknown poll breaks the run however long it ran", + interval: relaxed, baseline: yes, polls: []PRState{no, snap(MergeUnknown, "h1", "b1"), no}, want: "", + }, + { + name: "a baseline conflict that clears and returns alerts", + interval: relaxed, baseline: no, polls: []PRState{yes, no, no}, want: conflict, + }, + { + name: "a failed poll breaks the run", + interval: relaxed, baseline: yes, polls: []PRState{no, no, no}, + errs: []error{nil, nil, errors.New("HTTP 502"), nil}, want: "", + }, + { + name: "the run restarts after a failed poll and still confirms", + interval: relaxed, baseline: yes, polls: []PRState{no, no, no, no}, + errs: []error{nil, nil, errors.New("HTTP 502"), nil, nil}, want: conflict, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := &fakeFetcher{states: append([]PRState{tt.baseline}, tt.polls...), errs: tt.errs} + res, err := Watch(f, []PRRef{tt.baseline.Ref}, "unkin-agent", + spacedTicks(len(tt.polls), tt.interval), nil, func(PRRef, error) {}) + if err != nil { + t.Fatalf("Watch: %v", err) + } + if res.Reason != tt.want { + t.Fatalf("reason = %q, want %q", res.Reason, tt.want) + } + }) + } +} + +// Two refs watched together keep separate runs and separate clocks: a PR being +// pushed to must neither delay nor trigger the conflict its neighbour is really +// in, and the alert must name the conflicted one. +func TestWatchConflictIsolatedFromANeighboursPushes(t *testing.T) { + conflicted := PRRef{Owner: "unkin", Repo: "repo", Number: 1} + pushing := PRRef{Owner: "unkin", Repo: "repo", Number: 2} + snap := func(ref PRRef, m Mergeability, head string) PRState { + st := base() + st.Ref = ref + st.Mergeable = m + st.HeadSHA = head + return st + } + f := &perRefFetcher{ + states: map[string][]PRState{ + conflicted.String(): {snap(conflicted, MergeYes, "a1"), snap(conflicted, MergeNo, "a1")}, + pushing.String(): {snap(pushing, MergeYes, "b1"), snap(pushing, MergeYes, "b2"), snap(pushing, MergeYes, "b3")}, + }, + calls: map[string]int{}, + } + + res, err := Watch(f, []PRRef{conflicted, pushing}, "unkin-agent", drainableTicks(4), nil, nil) + if err != nil { + t.Fatalf("Watch: %v", err) + } + if res.Reason != "PR lost mergeability (conflict)" { + t.Fatalf("reason = %q, want the conflict on %s", res.Reason, conflicted) + } + if res.Ref != conflicted { + t.Errorf("alert names %s, want %s", res.Ref, conflicted) + } +}