7 Commits

Author SHA1 Message Date
benvin 5c0eb1e899 Merge pull request 'Abort watchpr when a poll can no longer see the PR' (#12) from benvin/watchpr-terminal-errors into main
ci/woodpecker/tag/release Pipeline was successful
Reviewed-on: #12
2026-09-09 23:13:17 +10:00
unkin-agent d04c5aa58d Scope watchpr's terminal 404 to the PR lookup
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was successful
Only a 404 from GetPR means the PR is gone. A 404 from any other call
can be a proxy or ingress blip, so it now warns and counts against the
consecutive-failure cap instead of killing the watch on first sight.
2026-09-09 22:55:56 +10:00
benvin 7c6ec361ae Merge pull request 'Accept a bare integer as seconds for watchpr --interval' (#11) from benvin/watchpr-interval-units into main
Reviewed-on: #11
2026-09-09 22:42:24 +10:00
unkin-agent 46dfe48adc Abort watchpr when a poll can no longer see the PR
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was successful
- treat a mid-run 404 on a tracked PR as terminal
- cap consecutive transient poll failures at 20 per PR
- reset the failure count on a successful poll
- export IsNotFound for callers to classify the abort
2026-09-09 22:41:27 +10:00
unkin-agent 71e42811fb Accept a bare integer as seconds for watchpr --interval
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was successful
- add agent.ParseDurationFlag: bare integer means seconds, duration strings still parse
- take --interval as a string and parse it in the command
- reject unparseable and non-positive values with an error naming the flag
2026-09-09 22:38:09 +10:00
benvin 985b58c406 Merge pull request 'Re-mint watchpr's Gitea token when it expires' (#10) from benvin/watchpr-auth-expiry into main
Reviewed-on: #10
2026-09-09 22:34:15 +10:00
unkin-agent 7ef0e28e96 Re-mint watchpr's Gitea token when it expires
ci/woodpecker/pr/build Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/pre-commit Pipeline was successful
Vault-minted Gitea tokens last ~1h, far less than a watch, and every poll
past expiry 401'd into a warning while watchpr looked healthy.

- retry a rejected request once with a freshly minted token
- abort the watch when the fresh token is rejected too
- poll anonymously when no token can be minted, mint only on a real 401/403
2026-09-09 21:15:46 +10:00
11 changed files with 945 additions and 29 deletions
+7 -2
View File
@@ -174,8 +174,13 @@ wrapped per stage (login / read denied / write denied) via `ErrVaultDenied`.
## Gotchas
- `watchpr` exits 0 with no output changes on `--once` (just prints state).
- The token cache is process-wide (`sync.Once`); tests call the unexported
`fetchGiteaToken` to avoid it.
- Gitea tokens expire in ~1h, shorter than a watch: the client re-mints once on a
401/403 and replays the request. If the fresh token is rejected too, `watchpr`
exits non-zero rather than polling blind.
- `watchpr` polls anonymously when no token can be minted (public repos work
fine); only a real 401/403 reaches for Vault.
- The token cache is process-wide (mutex-guarded); `RefreshGiteaToken` replaces
it. Tests call the unexported `fetchGiteaToken` to avoid the cache.
- `agentvault` never puts a secret in an error string: Vault decode failures and
Authentik `view_key` responses are reported without their bodies, and
`seed-oauth` reports key names only.
+3
View File
@@ -70,6 +70,9 @@ watchpr unkin/argocd-apps#42
# Multiple PRs, custom interval; refs accept #N or :N
watchpr --interval 30s unkin/argocd-apps#42 unkin/terraform-vault:98
# --interval takes a duration (30s, 2m, 1h30m) or a bare number of seconds
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
+21 -9
View File
@@ -7,6 +7,7 @@
// watchpr owner/repo#12 owner/repo:15
// watchpr --once --json owner/repo#12
// watchpr --interval 30s owner/repo#12
// watchpr --interval 30 owner/repo#12
package main
import (
@@ -34,7 +35,7 @@ func main() {
// tests can invoke Execute and assert the exit behaviour without spawning a
// process.
func newRootCmd() *cobra.Command {
var interval time.Duration
var intervalFlag string
var once, jsonMode bool
root := &cobra.Command{
@@ -50,6 +51,10 @@ func newRootCmd() *cobra.Command {
if len(args) == 0 {
return fmt.Errorf("no PR references given (e.g. owner/repo#12)")
}
interval, err := agent.ParseDurationFlag("interval", intervalFlag)
if err != nil {
return err
}
refs := make([]agent.PRRef, 0, len(args))
for _, a := range args {
ref, err := agent.ParsePRRef(a)
@@ -58,10 +63,7 @@ func newRootCmd() *cobra.Command {
}
refs = append(refs, ref)
}
c, err := clientFor()
if err != nil {
return err
}
c := clientFor()
if once {
return runOnce(c, refs, jsonMode)
}
@@ -71,7 +73,7 @@ func newRootCmd() *cobra.Command {
root.SetVersionTemplate("{{.Version}}\n")
f := root.Flags()
f.DurationVar(&interval, "interval", 60*time.Second, "Polling interval")
f.StringVar(&intervalFlag, "interval", "60s", "Polling interval: a duration (30s, 2m, 1h30m) or a bare number of seconds")
f.BoolVar(&once, "once", false, "Check once, print current state, and exit")
f.BoolVar(&jsonMode, "json", false, "Emit JSON")
@@ -84,12 +86,16 @@ func newRootCmd() *cobra.Command {
return root
}
func clientFor() (*agent.GiteaClient, error) {
// clientFor builds the Gitea client. Watching public repos works anonymously,
// so an unavailable token is a warning, not a failure; a poll that is actually
// rejected re-mints then.
func clientFor() *agent.GiteaClient {
token, err := agent.GiteaToken()
if err != nil {
return nil, err
fmt.Fprintf(os.Stderr, "warning: no Gitea token (%v); polling anonymously\n", err)
token = ""
}
return agent.NewGiteaClient(token), nil
return agent.NewGiteaClient(token)
}
// runOnce fetches and prints the current state of each PR, then exits 0.
@@ -131,6 +137,12 @@ func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration,
res, err := agent.Watch(c, refs, login, ticker.C, onBaseline, onError)
if err != nil {
if agent.IsAuthError(err) {
return fmt.Errorf("gitea authentication failed after re-minting the token, watch aborted: %w", err)
}
if agent.IsPRGone(err) {
return fmt.Errorf("PR no longer visible (repo deleted, renamed, or made private), watch aborted: %w", err)
}
return err
}
report(res.Ref.String(), res.Reason, res.State, jsonMode)
+77
View File
@@ -2,6 +2,9 @@ package main
import (
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
@@ -28,3 +31,77 @@ func TestExecuteNoArgsErrors(t *testing.T) {
t.Fatal("Execute() = nil, want error when no PR references are given")
}
}
// Watching a public repo with no credentials available must work: the failed
// mint is a warning, the poll goes out unauthenticated, and the command exits 0.
func TestOnceRunsAnonymouslyWhenNoTokenIsAvailable(t *testing.T) {
vault := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
defer vault.Close()
authHeaders := 0
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "" {
authHeaders++
}
_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/cafebabe/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, `[]`)
})
gitea := httptest.NewServer(mux)
defer gitea.Close()
t.Setenv("VAULT_ADDR", vault.URL)
t.Setenv("GITEA_URL", gitea.URL)
cmd := newRootCmd()
cmd.SetArgs([]string{"--once", "unkin/repo#7"})
cmd.SetOut(io.Discard)
cmd.SetErr(io.Discard)
if err := cmd.Execute(); err != nil {
t.Fatalf("anonymous --once should succeed without a token: %v", err)
}
if authHeaders != 0 {
t.Errorf("sent %d Authorization headers, want none", authHeaders)
}
}
// A bare integer interval means seconds and must survive flag parsing: the
// command should fail on the missing PR reference, not on the flag value.
func TestExecuteBareIntervalIsSeconds(t *testing.T) {
cmd := newRootCmd()
cmd.SetArgs([]string{"--interval", "15"})
cmd.SetOut(io.Discard)
cmd.SetErr(io.Discard)
err := cmd.Execute()
if err == nil {
t.Fatal("Execute() = nil, want the no-references error")
}
if !strings.Contains(err.Error(), "no PR references given") {
t.Fatalf("Execute() error = %v, want the no-references error", err)
}
}
// An unparseable interval is rejected before any Vault/Gitea call, with an
// error naming the flag and showing valid forms.
func TestExecuteBadIntervalErrors(t *testing.T) {
cmd := newRootCmd()
cmd.SetArgs([]string{"--interval", "soon", "unkin/repo#1"})
cmd.SetOut(io.Discard)
cmd.SetErr(io.Discard)
err := cmd.Execute()
if err == nil {
t.Fatal("Execute() = nil, want error for an unparseable --interval")
}
for _, want := range []string{"--interval", "30s"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("Execute() error %q does not mention %q", err, want)
}
}
}
+218
View File
@@ -2,9 +2,11 @@ package agent
import (
"encoding/json"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
@@ -231,3 +233,219 @@ func TestGiteaAPIError(t *testing.T) {
t.Fatal("expected error on 422")
}
}
// expiringGitea serves the PR endpoint, rejecting every token other than
// wantToken with a 401 exactly as Gitea does once a Vault-minted token expires.
// It records the tokens it saw, newest last.
func expiringGitea(t *testing.T, wantToken string, seen *[]string) *httptest.Server {
t.Helper()
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
tok := strings.TrimPrefix(r.Header.Get("Authorization"), "token ")
*seen = append(*seen, tok)
if tok != wantToken {
w.WriteHeader(http.StatusUnauthorized)
_, _ = io.WriteString(w, `{"message":"invalid username, password or token"}`)
return
}
_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
})
return httptest.NewServer(mux)
}
// The production failure: the token expired mid-run. The client must re-mint
// once and replay the request with the fresh token.
func TestExpiredTokenIsRemintedAndRetried(t *testing.T) {
var seen []string
srv := expiringGitea(t, "fresh", &seen)
defer srv.Close()
refreshes := 0
c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
Refresh: func() (string, error) { refreshes++; return "fresh", nil }}
pr, err := c.GetPR("unkin/repo", 7)
if err != nil {
t.Fatalf("GetPR after re-mint: %v", err)
}
if pr.Number != 7 {
t.Errorf("PR number = %d, want 7", pr.Number)
}
if refreshes != 1 {
t.Errorf("refreshes = %d, want 1", refreshes)
}
if len(seen) != 2 || seen[0] != "stale" || seen[1] != "fresh" {
t.Errorf("tokens seen = %v, want [stale fresh]", seen)
}
if c.Token != "fresh" {
t.Errorf("client token = %q, want the refreshed token", c.Token)
}
}
// A fresh token that is also rejected is a real auth failure: report it as one
// rather than re-minting forever.
func TestAuthFailureSurvivesRemint(t *testing.T) {
var seen []string
srv := expiringGitea(t, "never-issued", &seen)
defer srv.Close()
refreshes := 0
c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
Refresh: func() (string, error) { refreshes++; return "still-bad", nil }}
_, err := c.GetPR("unkin/repo", 7)
if err == nil {
t.Fatal("GetPR should fail when the fresh token is rejected too")
}
if !IsAuthError(err) {
t.Errorf("IsAuthError(%v) = false, want true", err)
}
if refreshes != 1 {
t.Errorf("refreshes = %d, want 1 (re-mint exactly once)", refreshes)
}
if len(seen) != 2 {
t.Errorf("requests = %d, want 2", len(seen))
}
}
// A refresh that itself fails must surface as an auth error, not as a silent
// success or a bare Vault error.
func TestRemintErrorIsReportedAsAuthFailure(t *testing.T) {
var seen []string
srv := expiringGitea(t, "fresh", &seen)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
Refresh: func() (string, error) { return "", errors.New("vault approle login: HTTP 503") }}
_, err := c.GetPR("unkin/repo", 7)
if err == nil || !IsAuthError(err) {
t.Fatalf("GetPR error = %v, want an auth error", err)
}
if !strings.Contains(err.Error(), "vault approle login") {
t.Errorf("error %q should name the re-mint failure", err)
}
if len(seen) != 1 {
t.Errorf("requests = %d, want 1 (no replay without a token)", len(seen))
}
}
// A 5xx is transient, not an auth problem: no re-mint, no retry, and the caller
// keeps its existing retry behaviour.
func TestServerErrorDoesNotRemint(t *testing.T) {
requests := 0
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
requests++
w.WriteHeader(http.StatusBadGateway)
})
srv := httptest.NewServer(mux)
defer srv.Close()
refreshes := 0
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client(),
Refresh: func() (string, error) { refreshes++; return "fresh", nil }}
_, err := c.GetPR("unkin/repo", 7)
if err == nil {
t.Fatal("expected error on 502")
}
if IsAuthError(err) {
t.Errorf("502 must not be an auth error")
}
if refreshes != 0 || requests != 1 {
t.Errorf("refreshes = %d, requests = %d, want 0 and 1", refreshes, requests)
}
}
// The replayed request must carry the original body, not an empty one.
func TestRemintReplaysRequestBody(t *testing.T) {
var bodies []CreatePROptions
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls", func(w http.ResponseWriter, r *http.Request) {
var body CreatePROptions
_ = json.NewDecoder(r.Body).Decode(&body)
bodies = append(bodies, body)
if strings.TrimPrefix(r.Header.Get("Authorization"), "token ") != "fresh" {
w.WriteHeader(http.StatusUnauthorized)
return
}
_, _ = io.WriteString(w, `{"number":7}`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
Refresh: func() (string, error) { return "fresh", nil }}
if _, err := c.CreatePR("unkin/repo", CreatePROptions{Base: "main", Head: "feature", Title: "T", Body: "B"}); err != nil {
t.Fatalf("CreatePR: %v", err)
}
if len(bodies) != 2 {
t.Fatalf("requests = %d, want 2", len(bodies))
}
if bodies[1] != bodies[0] {
t.Errorf("replayed body = %+v, want %+v", bodies[1], bodies[0])
}
}
func TestIsAuthError(t *testing.T) {
tests := []struct {
status int
want bool
}{
{http.StatusUnauthorized, true},
{http.StatusForbidden, true},
{http.StatusNotFound, false},
{http.StatusUnprocessableEntity, false},
{http.StatusBadGateway, false},
}
for _, tt := range tests {
err := error(&APIError{Method: "GET", Path: "/p", StatusCode: tt.status})
if got := IsAuthError(err); got != tt.want {
t.Errorf("IsAuthError(HTTP %d) = %v, want %v", tt.status, got, tt.want)
}
}
if IsAuthError(errors.New("dial tcp: timeout")) {
t.Errorf("a network error is not an auth error")
}
}
// Anonymous polling of a public repo is a supported mode: with no token the
// client must send no Authorization header, and must never reach for Vault.
func TestAnonymousPollingNeverMints(t *testing.T) {
authHeaders := 0
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "" {
authHeaders++
}
_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/cafebabe/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
_, _ = io.WriteString(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
refreshes := 0
c := &GiteaClient{BaseURL: srv.URL, HTTP: srv.Client(),
Refresh: func() (string, error) { refreshes++; return "", errors.New("vault unreachable") }}
st, err := FetchState(c, PRRef{Owner: "unkin", Repo: "repo", Number: 7}, "unkin-agent")
if err != nil {
t.Fatalf("anonymous FetchState: %v", err)
}
if st.State != "open" || st.CIStatus != "success" || st.HeadSHA != "cafebabe" {
t.Errorf("state = %+v", st)
}
if refreshes != 0 {
t.Errorf("refreshes = %d, want 0 (a 200 must never trigger a mint)", refreshes)
}
if authHeaders != 0 {
t.Errorf("sent %d Authorization headers, want none", authHeaders)
}
}
+42 -7
View File
@@ -24,40 +24,75 @@ func (e *APIError) Error() string {
return fmt.Sprintf("gitea %s %s: HTTP %d: %s", e.Method, e.Path, e.StatusCode, e.Body)
}
// isNotFound reports whether err is a Gitea 404.
func isNotFound(err error) bool {
// IsNotFound reports whether err is a Gitea 404. Gitea hides repositories a
// caller may not see behind a 404 rather than a 403, so this also covers a repo
// that was renamed, deleted, or made private.
func IsNotFound(err error) bool {
var apiErr *APIError
return errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound
}
// IsAuthError reports whether err is a Gitea 401/403: the token is expired or
// unauthorised, which retrying the same request cannot fix.
func IsAuthError(err error) bool {
var apiErr *APIError
return errors.As(err, &apiErr) &&
(apiErr.StatusCode == http.StatusUnauthorized || apiErr.StatusCode == http.StatusForbidden)
}
// GiteaClient talks to the Gitea REST API as the agent user.
type GiteaClient struct {
BaseURL string
Token string
HTTP *http.Client
// Refresh mints a replacement token when the current one is rejected; Vault's
// Gitea tokens expire in ~1h, far short of a watchpr run.
Refresh func() (string, error)
}
// NewGiteaClient builds a client from the configured base URL and a Vault-minted
// token.
// token, re-minting from Vault when that token expires.
func NewGiteaClient(token string) *GiteaClient {
return &GiteaClient{BaseURL: GiteaURL(), Token: token, HTTP: httpClient}
return &GiteaClient{BaseURL: GiteaURL(), Token: token, HTTP: httpClient, Refresh: RefreshGiteaToken}
}
// do sends the request and, if the token was rejected, re-mints it once and
// replays the request with the fresh token.
func (c *GiteaClient) do(method, path string, body any, out any) error {
var reader io.Reader
var payload []byte
if body != nil {
b, err := json.Marshal(body)
if err != nil {
return err
}
reader = bytes.NewReader(b)
payload = b
}
err := c.attempt(method, path, payload, out)
if !IsAuthError(err) || c.Refresh == nil {
return err
}
token, refreshErr := c.Refresh()
if refreshErr != nil {
return fmt.Errorf("%w; re-minting token: %v", err, refreshErr)
}
c.Token = token
return c.attempt(method, path, payload, out)
}
func (c *GiteaClient) attempt(method, path string, body []byte, out any) error {
var reader io.Reader
if body != nil {
reader = bytes.NewReader(body)
}
url := strings.TrimRight(c.BaseURL, "/") + path
req, err := http.NewRequest(method, url, reader)
if err != nil {
return err
}
req.Header.Set("Authorization", "token "+c.Token)
// An empty token means anonymous access, which public repos serve fine.
if c.Token != "" {
req.Header.Set("Authorization", "token "+c.Token)
}
req.Header.Set("Accept", "application/json")
if body != nil {
req.Header.Set("Content-Type", "application/json")
+20
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"strconv"
"strings"
"time"
)
// PRRef identifies a single pull request by repository and number.
@@ -48,6 +49,25 @@ func ParsePRRef(s string) (PRRef, error) {
return PRRef{Owner: owner, Repo: repo, Number: n}, nil
}
// ParseDurationFlag parses a duration flag value, accepting either a Go
// duration string ("30s", "1h30m") or a bare integer read as seconds ("15").
// flag names the flag so the error says which value was rejected.
func ParseDurationFlag(flag, value string) (time.Duration, error) {
s := strings.TrimSpace(value)
d, err := time.ParseDuration(s)
if err != nil {
n, nerr := strconv.Atoi(s)
if nerr != nil {
return 0, fmt.Errorf("invalid --%s value %q: want a duration such as 30s, 2m or 1h30m, or a bare number of seconds such as 15", flag, value)
}
d = time.Duration(n) * time.Second
}
if d <= 0 {
return 0, fmt.Errorf("invalid --%s value %q: must be greater than zero", flag, value)
}
return d, nil
}
// ParseRepo validates and splits an "owner/repo" string.
func ParseRepo(s string) (owner, repo string, err error) {
s = strings.TrimSpace(s)
+56 -1
View File
@@ -1,6 +1,10 @@
package agent
import "testing"
import (
"strings"
"testing"
"time"
)
func TestParsePRRef(t *testing.T) {
tests := []struct {
@@ -77,3 +81,54 @@ func TestParseRepo(t *testing.T) {
}
}
}
func TestParseDurationFlag(t *testing.T) {
tests := []struct {
in string
want time.Duration
wantErr bool
}{
{"15", 15 * time.Second, false},
{"15s", 15 * time.Second, false},
{"2m", 2 * time.Minute, false},
{"1h30m", 90 * time.Minute, false},
{"500ms", 500 * time.Millisecond, false},
{" 45 ", 45 * time.Second, false},
{"0", 0, true},
{"0s", 0, true},
{"-5", 0, true},
{"-5s", 0, true},
{"15x", 0, true},
{"", 0, true},
}
for _, tt := range tests {
got, err := ParseDurationFlag("interval", tt.in)
if tt.wantErr {
if err == nil {
t.Errorf("ParseDurationFlag(%q): expected error, got %v", tt.in, got)
}
continue
}
if err != nil {
t.Errorf("ParseDurationFlag(%q): unexpected error: %v", tt.in, err)
continue
}
if got != tt.want {
t.Errorf("ParseDurationFlag(%q) = %v, want %v", tt.in, got, tt.want)
}
}
}
// The error must name the flag and show valid forms instead of surfacing
// time.ParseDuration's "missing unit" wording.
func TestParseDurationFlagErrorMessage(t *testing.T) {
_, err := ParseDurationFlag("interval", "soon")
if err == nil {
t.Fatal("ParseDurationFlag(\"soon\"): expected error")
}
for _, want := range []string{"--interval", `"soon"`, "30s", "seconds"} {
if !strings.Contains(err.Error(), want) {
t.Errorf("error %q does not mention %q", err, want)
}
}
}
+19 -5
View File
@@ -86,17 +86,31 @@ func AuthentikURL() string {
}
var (
tokenOnce sync.Once
tokenValue string
tokenErr error
tokenMu sync.Mutex
tokenMinted bool
tokenValue string
tokenErr error
)
// GiteaToken returns a Gitea token, minting it via Vault AppRole on first call
// and caching it in-process for the lifetime of the command.
func GiteaToken() (string, error) {
tokenOnce.Do(func() {
tokenMu.Lock()
defer tokenMu.Unlock()
if !tokenMinted {
tokenValue, tokenErr = fetchGiteaToken(VaultAddr(), RoleID(), GiteaCredsPath())
})
tokenMinted = true
}
return tokenValue, tokenErr
}
// RefreshGiteaToken mints a fresh Gitea token and replaces the cached one, for
// callers that outlive the ~1h token TTL.
func RefreshGiteaToken() (string, error) {
tokenMu.Lock()
defer tokenMu.Unlock()
tokenValue, tokenErr = fetchGiteaToken(VaultAddr(), RoleID(), GiteaCredsPath())
tokenMinted = true
return tokenValue, tokenErr
}
+41 -5
View File
@@ -1,6 +1,20 @@
package agent
import "time"
import (
"errors"
"fmt"
"time"
)
// errPRGone marks a 404 from the PR lookup itself. A 404 from any other endpoint
// can be a proxy or ingress blip and is left to the ordinary failure cap.
var errPRGone = errors.New("PR no longer visible")
// IsPRGone reports whether err is a 404 from the PR lookup, meaning the PR is no
// longer visible rather than one endpoint being briefly unreachable.
func IsPRGone(err error) bool {
return errors.Is(err, errPRGone)
}
// PRState is a point-in-time snapshot of the PR attributes watchpr tracks.
type PRState struct {
@@ -20,13 +34,16 @@ type PRState struct {
func FetchState(c *GiteaClient, ref PRRef, agentLogin string) (PRState, error) {
pr, err := c.GetPR(ref.RepoPath(), ref.Number)
if err != nil {
if IsNotFound(err) {
return PRState{}, fmt.Errorf("%w: %w", errPRGone, err)
}
return PRState{}, err
}
// A 404 here means the head commit is gone (branch deleted after a squash/
// rebase merge); the PR object is still authoritative, so treat CI as absent
// rather than discarding the merge signal and hanging the watch loop.
ci, err := c.CommitStatus(ref.RepoPath(), pr.Head.Sha)
if err != nil && !isNotFound(err) {
if err != nil && !IsNotFound(err) {
return PRState{}, err
}
comments, err := c.ListComments(ref.RepoPath(), ref.Number)
@@ -78,12 +95,22 @@ func terminalState(st PRState) (bool, string) {
return false, ""
}
// MaxPollFailures is how many consecutive failed polls of the same PR are
// tolerated before Watch gives up. The abort fires on the 20th failed tick, so
// at watchpr's default 60s interval a watch rides out ~19 minutes of failure.
const MaxPollFailures = 20
// 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.
// than polled forever. Transient poll errors are handed to onError and the loop
// continues, but never blindly: a baseline fetch error, an authentication
// failure surviving a token re-mint, a 404 from the PR lookup itself (the repo
// is gone, renamed, or no longer visible), and MaxPollFailures consecutive
// failures of one PR all abort, because a watcher that sees nothing must not
// look healthy.
// 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 {
@@ -99,16 +126,25 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti
if onBaseline != nil {
onBaseline()
}
fails := make(map[string]int, len(refs))
for range ticks {
for _, ref := range refs {
key := ref.String()
cur, err := f.FetchState(ref, agentLogin)
if err != nil {
if IsAuthError(err) || IsPRGone(err) {
return WatchResult{}, fmt.Errorf("polling %s: %w", key, err)
}
fails[key]++
if onError != nil {
onError(ref, err)
}
if fails[key] >= MaxPollFailures {
return WatchResult{}, fmt.Errorf("polling %s: giving up after %d consecutive failures: %w", key, fails[key], err)
}
continue
}
fails[key] = 0
if changed, reason := MeaningfulChange(prev[key], cur); changed {
return WatchResult{Ref: ref, Reason: reason, State: cur}, nil
}
+441
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
@@ -360,3 +361,443 @@ func TestCountNonAgentComments(t *testing.T) {
t.Errorf("countNonAgentComments = %d, want 2", n)
}
}
// The production failure: the Vault-minted token expired mid-watch and every
// poll 401'd, which the loop logged as a warning and polled past forever. An
// auth error that survived the client's re-mint must end the watch with an
// error so watchpr exits non-zero instead of watching blind.
func TestWatchAbortsOnAuthError(t *testing.T) {
open := base()
merged := base()
merged.State = "closed"
merged.Merged = true
f := &fakeFetcher{
states: []PRState{open, open, merged},
errs: []error{nil, &APIError{Method: "GET", Path: "/p", StatusCode: 401, Body: "invalid token"}, nil},
}
warned := 0
ticks := make(chan time.Time, 2)
ticks <- time.Now()
ticks <- time.Now()
_, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil, func(PRRef, error) { warned++ })
if err == nil {
t.Fatal("Watch should return the auth failure, not keep polling")
}
if !IsAuthError(err) {
t.Errorf("Watch error = %v, want an auth error", err)
}
if warned != 0 {
t.Errorf("auth failure was logged as a warning %d time(s); it must abort", warned)
}
if f.calls != 2 {
t.Errorf("fetch calls = %d, want 2 (baseline + the failing poll)", f.calls)
}
}
// A 5xx keeps its retry behaviour: warn and poll on.
func TestWatchContinuesPastServerError(t *testing.T) {
open := base()
merged := base()
merged.State = "closed"
merged.Merged = true
f := &fakeFetcher{
states: []PRState{open, open, merged},
errs: []error{nil, &APIError{Method: "GET", Path: "/p", StatusCode: 502, Body: "bad gateway"}, nil},
}
warned := 0
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, error) { warned++ })
if err != nil {
t.Fatalf("Watch: %v", err)
}
if warned != 1 {
t.Errorf("warnings = %d, want 1", warned)
}
if res.Reason != "PR merged" {
t.Errorf("reason = %q, want %q", res.Reason, "PR merged")
}
}
// The production failure: a watched repo was renamed mid-watch, so every poll
// 404'd (Gitea hides a repo the caller may not see rather than 403ing) and the
// loop warned past it forever while reporting nothing. A 404 on a tracked PR
// must end the watch with an error naming that PR.
func TestWatchAbortsOnMidRunNotFound(t *testing.T) {
const sha = "deadbeefdeadbeef"
var polls atomic.Int32
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
if polls.Add(1) > 1 { // repo renamed/made private after the baseline
w.WriteHeader(http.StatusNotFound)
_, _ = fmt.Fprint(w, `{"message":"Not Found"}`)
return
}
_, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()}
ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7}
tk := time.NewTicker(5 * time.Millisecond)
defer tk.Stop()
var warned atomic.Int32
done := make(chan error, 1)
go func() {
_, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil,
func(PRRef, error) { warned.Add(1) })
done <- err
}()
select {
case err := <-done:
if err == nil {
t.Fatal("Watch should abort on a mid-run 404, not keep polling")
}
if !IsNotFound(err) {
t.Errorf("Watch error = %v, want a 404", err)
}
if !strings.Contains(err.Error(), ref.String()) {
t.Errorf("Watch error = %v, want it to name %s", err, ref.String())
}
if n := warned.Load(); n != 0 {
t.Errorf("404 was logged as a warning %d time(s); it must abort", n)
}
case <-time.After(3 * time.Second):
t.Fatal("Watch hung: a vanished repo was warned past instead of aborting")
}
}
// A 404 from a sub-resource is not proof the PR is gone: an ingress can serve
// one during a Gitea rolling restart. Only the PR lookup itself is authoritative,
// so a comments 404 must warn and keep polling like any other transient failure,
// and still catch the merge that lands afterwards.
func TestWatchSurvivesCommentsNotFound(t *testing.T) {
const sha = "0badc0de0badc0de"
var polls atomic.Int32
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
if polls.Add(1) >= 4 {
_, _ = fmt.Fprintf(w, `{"number":7,"state":"closed","merged":true,"mergeable":true,"head":{"sha":%q}}`, sha)
return
}
_, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
if n := polls.Load(); n == 2 || n == 3 { // proxy blip across two polls
w.WriteHeader(http.StatusNotFound)
_, _ = fmt.Fprint(w, `{"message":"Not Found"}`)
return
}
_, _ = fmt.Fprint(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()}
ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7}
tk := time.NewTicker(5 * time.Millisecond)
defer tk.Stop()
var warned atomic.Int32
type outcome struct {
res WatchResult
err error
}
done := make(chan outcome, 1)
go func() {
res, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil,
func(PRRef, error) { warned.Add(1) })
done <- outcome{res, err}
}()
select {
case o := <-done:
if o.err != nil {
t.Fatalf("Watch: %v (a comments 404 must not be terminal)", o.err)
}
if o.res.Reason != "PR merged" {
t.Errorf("reason = %q, want %q", o.res.Reason, "PR merged")
}
if n := warned.Load(); n != 2 {
t.Errorf("warnings = %d, want 2", n)
}
case <-time.After(3 * time.Second):
t.Fatal("Watch hung: a comments 404 must warn and keep polling")
}
}
// A comments 404 costs a poll from the same budget as any other failure: it must
// not be free, and a permanently 404ing sub-resource must still end the watch.
func TestWatchCommentsNotFoundCountsTowardCap(t *testing.T) {
const sha = "1badc0de1badc0de"
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `{"state":"success"}`)
})
var comments atomic.Int32
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
if comments.Add(1) > 1 { // healthy at baseline, gone from the first poll on
w.WriteHeader(http.StatusNotFound)
_, _ = fmt.Fprint(w, `{"message":"Not Found"}`)
return
}
_, _ = fmt.Fprint(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()}
ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7}
tk := time.NewTicker(time.Millisecond)
defer tk.Stop()
var warned atomic.Int32
done := make(chan error, 1)
go func() {
_, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil,
func(PRRef, error) { warned.Add(1) })
done <- err
}()
select {
case err := <-done:
if err == nil {
t.Fatal("Watch should give up once the comments 404 stops being transient")
}
if !strings.Contains(err.Error(), "consecutive failures") {
t.Errorf("Watch error = %v, want it to report the failure cap", err)
}
if n := warned.Load(); n != MaxPollFailures {
t.Errorf("warnings = %d, want %d", n, MaxPollFailures)
}
case <-time.After(3 * time.Second):
t.Fatal("Watch hung: a permanently 404ing comments endpoint must hit the cap")
}
}
// The PR lookup is the call whose 404 means the PR is gone, so it aborts on the
// very first occurrence rather than spending the failure budget.
func TestWatchAbortsOnFirstPRLookupNotFound(t *testing.T) {
const sha = "2badc0de2badc0de"
var polls atomic.Int32
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
if polls.Add(1) > 1 {
w.WriteHeader(http.StatusNotFound)
_, _ = fmt.Fprint(w, `{"message":"Not Found"}`)
return
}
_, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()}
ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7}
tk := time.NewTicker(5 * time.Millisecond)
defer tk.Stop()
done := make(chan error, 1)
go func() {
_, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil,
func(PRRef, error) { t.Errorf("a PR-lookup 404 must abort, not warn") })
done <- err
}()
select {
case err := <-done:
if !IsNotFound(err) {
t.Fatalf("Watch error = %v, want a 404", err)
}
if n := polls.Load(); n != 2 {
t.Errorf("PR fetches = %d, want 2 (baseline + the 404 that aborts)", n)
}
case <-time.After(3 * time.Second):
t.Fatal("Watch hung: a vanished PR must abort")
}
}
// A 5xx blip must not kill a long watch: it warns, keeps polling, and still
// catches the merge that lands afterwards.
func TestWatchSurvivesTransientServerError(t *testing.T) {
const sha = "feedfacefeedface"
var polls atomic.Int32
mux := http.NewServeMux()
mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
switch n := polls.Add(1); {
case n == 2 || n == 3: // gateway blip across two polls
w.WriteHeader(http.StatusBadGateway)
_, _ = fmt.Fprint(w, `bad gateway`)
case n >= 4:
_, _ = fmt.Fprintf(w, `{"number":7,"state":"closed","merged":true,"mergeable":true,"head":{"sha":%q}}`, sha)
default:
_, _ = fmt.Fprintf(w, `{"number":7,"state":"open","merged":false,"mergeable":true,"head":{"sha":%q}}`, sha)
}
})
mux.HandleFunc("/api/v1/repos/unkin/repo/commits/"+sha+"/status", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `{"state":"success"}`)
})
mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
_, _ = fmt.Fprint(w, `[]`)
})
srv := httptest.NewServer(mux)
defer srv.Close()
c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client()}
ref := PRRef{Owner: "unkin", Repo: "repo", Number: 7}
tk := time.NewTicker(5 * time.Millisecond)
defer tk.Stop()
var warned atomic.Int32
type outcome struct {
res WatchResult
err error
}
done := make(chan outcome, 1)
go func() {
res, err := Watch(c, []PRRef{ref}, "unkin-agent", tk.C, nil,
func(PRRef, error) { warned.Add(1) })
done <- outcome{res, err}
}()
select {
case o := <-done:
if o.err != nil {
t.Fatalf("Watch: %v", o.err)
}
if o.res.Reason != "PR merged" {
t.Errorf("reason = %q, want %q", o.res.Reason, "PR merged")
}
if n := warned.Load(); n != 2 {
t.Errorf("warnings = %d, want 2", n)
}
case <-time.After(3 * time.Second):
t.Fatal("Watch hung: a transient 5xx must not stop the watch")
}
}
// pollFailures scripts a fetcher whose polls fail with a 502 at the given call
// indexes (0 is the baseline); the final call returns merged.
func pollFailures(calls int, failAt map[int]bool) *fakeFetcher {
open, merged := base(), base()
merged.State = "closed"
merged.Merged = true
f := &fakeFetcher{states: make([]PRState, calls), errs: make([]error, calls)}
for i := range calls {
f.states[i] = open
if failAt[i] {
f.errs[i] = &APIError{Method: "GET", Path: "/p", StatusCode: 502, Body: "bad gateway"}
}
}
f.states[calls-1] = merged
return f
}
// A permanently wedged endpoint (5xx forever) must eventually give up instead of
// warning on every tick for the life of the process.
func TestWatchAbortsAfterConsecutiveFailures(t *testing.T) {
failAt := map[int]bool{}
for i := 1; i <= MaxPollFailures; i++ {
failAt[i] = true
}
f := pollFailures(MaxPollFailures+1, failAt)
warned := 0
ticks := make(chan time.Time, MaxPollFailures)
for range MaxPollFailures {
ticks <- time.Now()
}
close(ticks)
_, err := Watch(f, []PRRef{base().Ref}, "unkin-agent", ticks, nil, func(PRRef, error) { warned++ })
if err == nil {
t.Fatal("Watch should give up once the failures stop being transient")
}
if !strings.Contains(err.Error(), "consecutive failures") {
t.Errorf("Watch error = %v, want it to report the failure cap", err)
}
if warned != MaxPollFailures {
t.Errorf("warnings = %d, want %d", warned, MaxPollFailures)
}
if f.calls != MaxPollFailures+1 {
t.Errorf("fetch calls = %d, want %d", f.calls, MaxPollFailures+1)
}
}
// The cap counts consecutive failures only: a single successful poll clears it,
// so an intermittent endpoint is watched indefinitely and the merge is caught.
func TestWatchFailureCountResetsOnSuccess(t *testing.T) {
const runs = MaxPollFailures - 1
failAt := map[int]bool{}
for i := 1; i <= runs; i++ { // first run of failures
failAt[i] = true
}
for i := runs + 2; i <= 2*runs+1; i++ { // second run, after one good poll
failAt[i] = true
}
f := pollFailures(2*runs+3, failAt)
ticks := make(chan time.Time, 2*runs+2)
for range 2*runs + 2 {
ticks <- time.Now()
}
close(ticks)
res, err := Watch(f, []PRRef{base().Ref}, "unkin-agent", ticks, nil, func(PRRef, error) {})
if err != nil {
t.Fatalf("Watch: %v (a successful poll must reset the failure count)", err)
}
if res.Reason != "PR merged" {
t.Errorf("reason = %q, want %q", res.Reason, "PR merged")
}
}
// Anonymous watching of a public repo must poll on without a credential in
// sight: no token, no mint, no exit until something actually changes.
func TestWatchAnonymousKeepsPolling(t *testing.T) {
open := base()
f := &fakeFetcher{states: []PRState{open}}
ticks := make(chan time.Time, 2)
ticks <- time.Now()
ticks <- time.Now()
close(ticks)
res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil,
func(_ PRRef, e error) { t.Errorf("unexpected poll error: %v", e) })
if err != nil {
t.Fatalf("Watch: %v", err)
}
if res.Reason != "" {
t.Errorf("reason = %q, want no change reported", res.Reason)
}
if f.calls != 3 {
t.Errorf("fetch calls = %d, want 3 (baseline + two polls)", f.calls)
}
}