watchpr: add --max-wait with distinct timeout exit code #25
@@ -178,6 +178,9 @@ wrapped per stage (login / read denied / write denied) via `ErrVaultDenied`.
|
||||
## Gotchas
|
||||
|
||||
- `watchpr` exits 0 with no output changes on `--once` (just prints state).
|
||||
- Background commands are killed at a 2h cap, indistinguishable from a crash.
|
||||
Orchestrators run `watchpr --max-wait 110m ...` and restart it on exit code 3
|
||||
(no change within max-wait); any other non-zero exit is a real failure.
|
||||
- Gitea tokens expire in ~1h, shorter than a watch: the client re-mints once when
|
||||
the credential it sent was rejected and replays the request. If the fresh token
|
||||
is rejected too, `watchpr` exits non-zero rather than polling blind.
|
||||
|
||||
@@ -131,11 +131,20 @@ watchpr --interval 30 unkin/argocd-apps#42
|
||||
# One-shot: print current state and exit 0 (great for scripts)
|
||||
watchpr --once unkin/argocd-apps#42
|
||||
watchpr --once --json unkin/argocd-apps#42
|
||||
|
||||
# Orchestrators: run inside a background command (2h hard cap) and restart on exit 3
|
||||
watchpr --max-wait 110m unkin/argocd-apps#42
|
||||
```
|
||||
|
||||
On a meaningful change `watchpr` prints the reason and the PR's current state,
|
||||
then exits 0. Use `--json` for machine-readable output.
|
||||
|
||||
`--max-wait` (same forms as `--interval`; default `0` = unlimited) bounds a
|
||||
watch: when it elapses with no change, `watchpr` prints `timeout: no change
|
||||
within <d>` and each PR's current state (`{"timeout":true,...}` under `--json`)
|
||||
and exits **3**, so a timed-out watch is distinguishable from a crash and is
|
||||
safe to restart.
|
||||
|
||||
### Exit behaviour
|
||||
|
||||
A watcher that sees nothing must not look healthy, so every terminal failure
|
||||
|
||||
+95
-7
@@ -19,10 +19,12 @@
|
||||
// watchpr --once --json owner/repo#12
|
||||
// watchpr --interval 30s owner/repo#12
|
||||
// watchpr --interval 30 owner/repo#12
|
||||
// watchpr --max-wait 110m owner/repo#12 # exits 3 if nothing changed
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
@@ -36,19 +38,34 @@ import (
|
||||
|
||||
var version = "dev"
|
||||
|
||||
// exitTimedOut is the exit status when --max-wait elapses with no change, so a
|
||||
// caller can tell a deliberate timeout from a crash and restart the watch.
|
||||
const exitTimedOut = 3
|
||||
|
||||
var errTimedOut = errors.New("max-wait elapsed with no change")
|
||||
|
||||
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)
|
||||
os.Exit(exitCode(newRootCmd().Execute()))
|
||||
}
|
||||
|
||||
// exitCode maps a command result to the process exit status.
|
||||
func exitCode(err error) int {
|
||||
switch {
|
||||
case err == nil:
|
||||
return 0
|
||||
case errors.Is(err, errTimedOut):
|
||||
return exitTimedOut
|
||||
}
|
||||
return 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 intervalFlag string
|
||||
var intervalFlag, maxWaitFlag string
|
||||
var once, jsonMode bool
|
||||
|
||||
root := &cobra.Command{
|
||||
@@ -57,7 +74,9 @@ func newRootCmd() *cobra.Command {
|
||||
Long: "watchpr polls each PR every --interval and exits (reporting what changed)\n" +
|
||||
"when a PR merges/closes, gets a new non-agent comment, its CI fails, or it\n" +
|
||||
"loses mergeability after the baseline. Conditions already true at the\n" +
|
||||
"baseline are printed, not alerted on. Refs take owner/repo#N or owner/repo:N.",
|
||||
"baseline are printed, not alerted on. Refs take owner/repo#N or owner/repo:N.\n\n" +
|
||||
"Exit status: 0 on a change (or --once), 3 when --max-wait elapses with no\n" +
|
||||
"change (current states are printed), 1 on any error.",
|
||||
Version: version,
|
||||
Args: cobra.ArbitraryArgs,
|
||||
SilenceUsage: true,
|
||||
@@ -69,6 +88,13 @@ func newRootCmd() *cobra.Command {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
maxWait, err := parseMaxWait(maxWaitFlag)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if once && maxWait > 0 {
|
||||
return fmt.Errorf("--max-wait has no effect with --once")
|
||||
}
|
||||
refs := make([]agent.PRRef, 0, len(args))
|
||||
for _, a := range args {
|
||||
ref, err := agent.ParsePRRef(a)
|
||||
@@ -81,13 +107,18 @@ func newRootCmd() *cobra.Command {
|
||||
if once {
|
||||
return runOnce(c, refs, jsonMode)
|
||||
}
|
||||
return runWatch(c, refs, interval, jsonMode)
|
||||
err = runWatch(c, refs, interval, maxWait, jsonMode)
|
||||
if errors.Is(err, errTimedOut) {
|
||||
cmd.SilenceErrors = true
|
||||
}
|
||||
return err
|
||||
},
|
||||
}
|
||||
root.SetVersionTemplate("{{.Version}}\n")
|
||||
|
||||
f := root.Flags()
|
||||
f.StringVar(&intervalFlag, "interval", "60s", "Polling interval: a duration (30s, 2m, 1h30m) or a bare number of seconds")
|
||||
f.StringVar(&maxWaitFlag, "max-wait", "0", "Give up after this long with no change and exit 3 (same forms as --interval; 0 = unlimited)")
|
||||
f.BoolVar(&once, "once", false, "Check once, print current state, and exit")
|
||||
f.BoolVar(&jsonMode, "json", false, "Emit JSON")
|
||||
|
||||
@@ -154,11 +185,17 @@ func runOnce(c *agent.GiteaClient, refs []agent.PRRef, jsonMode bool) error {
|
||||
|
||||
// runWatch establishes a baseline then polls until a tracked PR changes
|
||||
// meaningfully, at which point it reports the change and returns.
|
||||
func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration, jsonMode bool) error {
|
||||
func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval, maxWait time.Duration, jsonMode bool) error {
|
||||
login := agent.AgentLogin()
|
||||
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
var deadline <-chan time.Time
|
||||
if maxWait > 0 {
|
||||
timer := time.NewTimer(maxWait)
|
||||
defer timer.Stop()
|
||||
deadline = timer.C
|
||||
}
|
||||
|
||||
onBaseline := func(states []agent.PRState) {
|
||||
emitBaselines(os.Stderr, states, interval, jsonMode)
|
||||
@@ -167,14 +204,65 @@ func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration,
|
||||
warn(os.Stderr, jsonMode, "polling %s: %v", ref.String(), err)
|
||||
}
|
||||
|
||||
res, err := agent.Watch(c, refs, login, ticker.C, onBaseline, onError)
|
||||
res, err := agent.Watch(c, refs, login, untilDeadline(ticker.C, deadline), onBaseline, onError)
|
||||
if err != nil {
|
||||
return describeFailure(err)
|
||||
}
|
||||
if res.TimedOut {
|
||||
reportTimeout(os.Stdout, maxWait, res.States, jsonMode)
|
||||
return errTimedOut
|
||||
}
|
||||
report(res.Ref.String(), res.Reason, res.State, jsonMode)
|
||||
return nil
|
||||
}
|
||||
|
||||
// parseMaxWait reads --max-wait: 0 means unlimited, anything else parses like
|
||||
// --interval.
|
||||
func parseMaxWait(v string) (time.Duration, error) {
|
||||
if d, err := time.ParseDuration(strings.TrimSpace(v)); err == nil && d == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
return agent.ParseDurationFlag("max-wait", v)
|
||||
}
|
||||
|
||||
// untilDeadline forwards ticks until deadline fires, then closes, which ends
|
||||
// Watch without waiting out the rest of an interval. A nil deadline never fires.
|
||||
func untilDeadline(ticks, deadline <-chan time.Time) <-chan time.Time {
|
||||
out := make(chan time.Time)
|
||||
go func() {
|
||||
defer close(out)
|
||||
for {
|
||||
select {
|
||||
case <-deadline:
|
||||
return
|
||||
case t := <-ticks:
|
||||
select {
|
||||
case out <- t:
|
||||
case <-deadline:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return out
|
||||
}
|
||||
|
||||
// reportTimeout emits each PR's current state when --max-wait elapses.
|
||||
func reportTimeout(w io.Writer, maxWait time.Duration, states []agent.PRState, jsonMode bool) {
|
||||
if jsonMode {
|
||||
_ = json.NewEncoder(w).Encode(struct {
|
||||
Timeout bool `json:"timeout"`
|
||||
MaxWait string `json:"max_wait"`
|
||||
States []agent.PRState `json:"states"`
|
||||
}{true, maxWait.String(), states})
|
||||
return
|
||||
}
|
||||
_, _ = fmt.Fprintf(w, "timeout: no change within %s\n", maxWait)
|
||||
for _, st := range states {
|
||||
_, _ = fmt.Fprintln(w, stateLine(st))
|
||||
}
|
||||
}
|
||||
|
||||
// baselineRecord is the --json form of baselineLine. Both go to stderr, leaving
|
||||
// the stdout contract a single result record: a caller automating watchpr is
|
||||
// precisely the one who needs to be told the watch started against a PR that is
|
||||
|
||||
@@ -311,3 +311,96 @@ func TestBaselineNamesAnUnknownMergeability(t *testing.T) {
|
||||
t.Errorf("baselineLine = %q, want the suppression note", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseMaxWait(t *testing.T) {
|
||||
for in, want := range map[string]time.Duration{"0": 0, "0s": 0, "30s": 30 * time.Second, "1h55m": 115 * time.Minute, "90": 90 * time.Second} {
|
||||
got, err := parseMaxWait(in)
|
||||
if err != nil || got != want {
|
||||
t.Errorf("parseMaxWait(%q) = %v, %v; want %v", in, got, err, want)
|
||||
}
|
||||
}
|
||||
for _, in := range []string{"soon", "-1m"} {
|
||||
if _, err := parseMaxWait(in); err == nil || !strings.Contains(err.Error(), "--max-wait") {
|
||||
t.Errorf("parseMaxWait(%q) err = %v, want a --max-wait error", in, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteBadMaxWaitErrors(t *testing.T) {
|
||||
cmd := newRootCmd()
|
||||
cmd.SetArgs([]string{"--max-wait", "soon", "unkin/repo#1"})
|
||||
cmd.SetOut(io.Discard)
|
||||
cmd.SetErr(io.Discard)
|
||||
if err := cmd.Execute(); err == nil || errors.Is(err, errTimedOut) {
|
||||
t.Fatalf("Execute() = %v, want a parse error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteOnceWithMaxWaitErrors(t *testing.T) {
|
||||
cmd := newRootCmd()
|
||||
cmd.SetArgs([]string{"--once", "--max-wait", "5m", "unkin/repo#1"})
|
||||
cmd.SetOut(io.Discard)
|
||||
cmd.SetErr(io.Discard)
|
||||
err := cmd.Execute()
|
||||
if err == nil || !strings.Contains(err.Error(), "--once") {
|
||||
t.Fatalf("Execute() = %v, want a --once/--max-wait conflict error", err)
|
||||
}
|
||||
if got := exitCode(err); got != 1 {
|
||||
t.Fatalf("exitCode = %d, want 1", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExitCode(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
err error
|
||||
want int
|
||||
}{
|
||||
{nil, 0},
|
||||
{errTimedOut, exitTimedOut},
|
||||
{fmt.Errorf("wrapped: %w", errTimedOut), exitTimedOut},
|
||||
{errors.New("boom"), 1},
|
||||
} {
|
||||
if got := exitCode(tc.err); got != tc.want {
|
||||
t.Errorf("exitCode(%v) = %d, want %d", tc.err, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Ticks pass through until the deadline fires; then the channel closes without
|
||||
// waiting for another tick.
|
||||
func TestUntilDeadline(t *testing.T) {
|
||||
ticks := make(chan time.Time)
|
||||
deadline := make(chan time.Time)
|
||||
out := untilDeadline(ticks, deadline)
|
||||
|
||||
now := time.Now()
|
||||
ticks <- now
|
||||
if got := <-out; !got.Equal(now) {
|
||||
t.Fatalf("forwarded %v, want %v", got, now)
|
||||
}
|
||||
close(deadline)
|
||||
if _, ok := <-out; ok {
|
||||
t.Fatal("channel still open after the deadline")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReportTimeout(t *testing.T) {
|
||||
st := agent.PRState{Ref: agent.PRRef{Owner: "unkin", Repo: "repo", Number: 3}, State: "open", Mergeable: agent.MergeYes, CIStatus: "pending"}
|
||||
|
||||
var buf bytes.Buffer
|
||||
reportTimeout(&buf, 110*time.Minute, []agent.PRState{st}, false)
|
||||
lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
|
||||
if len(lines) != 2 || lines[0] != "timeout: no change within 1h50m0s" || lines[1] != stateLine(st) {
|
||||
t.Errorf("text output = %q", buf.String())
|
||||
}
|
||||
|
||||
buf.Reset()
|
||||
reportTimeout(&buf, time.Minute, []agent.PRState{st}, true)
|
||||
var rec struct {
|
||||
Timeout bool `json:"timeout"`
|
||||
States []agent.PRState `json:"states"`
|
||||
}
|
||||
if err := json.Unmarshal(buf.Bytes(), &rec); err != nil || !rec.Timeout || len(rec.States) != 1 {
|
||||
t.Errorf("json output = %q (err %v)", buf.String(), err)
|
||||
}
|
||||
}
|
||||
|
||||
+12
-6
@@ -124,11 +124,14 @@ func (c *GiteaClient) FetchState(ref PRRef, agentLogin string) (PRState, error)
|
||||
return FetchState(c, ref, agentLogin)
|
||||
}
|
||||
|
||||
// WatchResult is the change that ended a watch.
|
||||
// WatchResult is the change that ended a watch. When the tick channel closes
|
||||
// with no change, TimedOut is set and States holds each PR's last seen state.
|
||||
type WatchResult struct {
|
||||
Ref PRRef
|
||||
Reason string
|
||||
State PRState
|
||||
Ref PRRef
|
||||
Reason string
|
||||
State PRState
|
||||
TimedOut bool
|
||||
States []PRState
|
||||
}
|
||||
|
||||
// terminalState reports whether a PR has reached a final state from which no
|
||||
@@ -276,6 +279,7 @@ const MaxPollFailures = 20
|
||||
// look healthy.
|
||||
// onBaseline, if set, receives every captured baseline once, before the first
|
||||
// tick, so a caller can show what state the watch started from.
|
||||
// A closed ticks channel ends the watch with a TimedOut result.
|
||||
func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Time, onBaseline func([]PRState), onError func(PRRef, error)) (WatchResult, error) {
|
||||
watches := make(map[string]*prWatch, len(refs))
|
||||
baselines := make([]PRState, 0, len(refs))
|
||||
@@ -293,11 +297,12 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti
|
||||
if onBaseline != nil {
|
||||
onBaseline(baselines)
|
||||
}
|
||||
latest := append([]PRState(nil), baselines...)
|
||||
fails := make(map[string]int, len(refs))
|
||||
// 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 {
|
||||
for i, ref := range refs {
|
||||
key := ref.String()
|
||||
cur, err := f.FetchState(ref, agentLogin)
|
||||
if err != nil {
|
||||
@@ -315,12 +320,13 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti
|
||||
continue
|
||||
}
|
||||
fails[key] = 0
|
||||
latest[i] = cur
|
||||
if changed, reason := watches[key].observe(cur, now); changed {
|
||||
return WatchResult{Ref: ref, Reason: reason, State: cur}, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return WatchResult{}, nil
|
||||
return WatchResult{TimedOut: true, States: latest}, nil
|
||||
}
|
||||
|
||||
// countNonAgentComments counts comments authored by anyone other than agentLogin.
|
||||
|
||||
@@ -1389,3 +1389,23 @@ func TestWatchConflictWindowCountsFromTheZeroTime(t *testing.T) {
|
||||
t.Fatalf("reason = %q, want the mergeability loss; a run starting at the zero time still counts", res.Reason)
|
||||
}
|
||||
}
|
||||
|
||||
// Closed ticks with no change end the watch as a timeout carrying the latest
|
||||
// state seen, not the baseline.
|
||||
func TestWatchTimesOutWithLatestStates(t *testing.T) {
|
||||
open := base()
|
||||
pushed := base()
|
||||
pushed.HeadSHA = "def456"
|
||||
f := &fakeFetcher{states: []PRState{open, pushed}}
|
||||
|
||||
res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", drainableTicks(2), nil, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Watch: %v", err)
|
||||
}
|
||||
if !res.TimedOut {
|
||||
t.Fatalf("TimedOut = false, want true")
|
||||
}
|
||||
if len(res.States) != 1 || res.States[0].HeadSHA != "def456" {
|
||||
t.Errorf("States = %+v, want the latest polled state", res.States)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user