diff --git a/cmd/master/main.go b/cmd/master/main.go index c942ff4..15cde34 100644 --- a/cmd/master/main.go +++ b/cmd/master/main.go @@ -23,6 +23,7 @@ import ( "github.com/shcizo/package-updater/internal/master/store" "github.com/shcizo/package-updater/internal/master/web" "github.com/shcizo/package-updater/internal/master/webauth" + "github.com/shcizo/package-updater/internal/master/webhook" "github.com/shcizo/package-updater/internal/pin" ) @@ -93,6 +94,14 @@ func run() error { } _, _ = w.Write([]byte("ok")) }) + // The webhook does its own bearer-token check, so it sits outside the + // OIDC login like /healthz. + mux.Handle("POST /update", webhook.New(st, jobService, broker, webhook.Config{ + Token: cfg.UpdaterAPIKey, Timeout: cfg.UpdateTimeout, PublicURL: cfg.PublicURL, + }, logger)) + if cfg.UpdaterAPIKey == "" { + logger.Warn("webhook_disabled", "reason", "UPDATER_API_KEY not set") + } mux.HandleFunc("GET /auth/login", auth.Login) mux.HandleFunc("GET /auth/callback", auth.Callback) mux.Handle("POST /auth/logout", auth.Require(http.HandlerFunc(auth.Logout))) @@ -103,8 +112,8 @@ func run() error { // instead of holding Shutdown until its timeout. baseContext := func(net.Listener) context.Context { return ctx } - // No WriteTimeout on either server: SSE streams and agent connections - // are long-lived by design. + // No WriteTimeout on either server: SSE streams, agent connections and + // POST /update (which waits up to UPDATE_TIMEOUT) are long-lived by design. webSrv := &http.Server{ Addr: ":" + cfg.Port, Handler: mux, ReadHeaderTimeout: 10 * time.Second, BaseContext: baseContext, diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 905b064..74cb2c2 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -11,6 +11,7 @@ import ( "net" "net/http" "net/http/httptest" + "sort" "strings" "sync" "sync/atomic" @@ -65,6 +66,33 @@ func (m *memStore) GetService(_ context.Context, id int64) (model.Service, error return svc, nil } +// ServicesByRepo returns the seeded services whose image repo equals repo, +// ordered by ID. +func (m *memStore) ServicesByRepo(_ context.Context, repo string) ([]model.Service, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []model.Service + for _, svc := range m.svcRows { + if svc.ImageRepo == repo && svc.RemovedAt == nil { + out = append(out, svc) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (m *memStore) SetAvailable(_ context.Context, id int64, a model.Available) error { + m.mu.Lock() + defer m.mu.Unlock() + svc, ok := m.svcRows[id] + if !ok { + return model.ErrNotFound + } + svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Source, &a.At + m.svcRows[id] = svc + return nil +} + func (m *memStore) CreateJob(_ context.Context, nj model.NewJob) (model.Job, error) { m.mu.Lock() defer m.mu.Unlock() @@ -155,6 +183,10 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo if f.Status == model.JobSucceeded { svc := m.svcRows[j.ServiceID] svc.DesiredRef = f.Ref + // Same rule as the SQL: a job created after the suggestion deployed it. + if svc.AvailableAt != nil && !svc.AvailableAt.After(j.CreatedAt) { + svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = "", "", nil + } m.svcRows[j.ServiceID] = svc } m.finishApplied++ diff --git a/internal/e2e/jobs_test.go b/internal/e2e/jobs_test.go index d3f1ae7..912ea93 100644 --- a/internal/e2e/jobs_test.go +++ b/internal/e2e/jobs_test.go @@ -102,7 +102,7 @@ func newJobEnv(t *testing.T) *jobEnv { t.Helper() store := &memStore{agents: map[string]model.Agent{agenttoken.Hash("pua_ok"): {ID: agentID, Name: "node-2"}}} store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", - AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"}) + AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":1.4.2@sha256:old"}) m := startMaster(t, store) exec := newGateExecutor() env := &jobEnv{store: store, m: m, exec: exec, diff --git a/internal/e2e/webhook_test.go b/internal/e2e/webhook_test.go new file mode 100644 index 0000000..5e38c82 --- /dev/null +++ b/internal/e2e/webhook_test.go @@ -0,0 +1,128 @@ +package e2e + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/master/webhook" +) + +// webhookServer mounts the webhook handler on a test server, wired to the +// job environment's store, jobs service and broker. +func (e *jobEnv) webhookServer(t *testing.T) *httptest.Server { + t.Helper() + h := webhook.New(e.store, e.jobs, e.m.broker, webhook.Config{ + Token: "ci-key", Timeout: 5 * time.Second, PublicURL: "https://updater.example.com", + }, slog.New(slog.NewTextHandler(io.Discard, nil))) + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + return srv +} + +func (e *jobEnv) setAuto(id int64) { + e.store.mu.Lock() + defer e.store.mu.Unlock() + svc := e.store.svcRows[id] + svc.UpdateMode = model.ModeAuto + e.store.svcRows[id] = svc +} + +type webhookResponse struct { + Matched int `json:"matched"` + Results []struct { + Status string `json:"status"` + JobID int64 `json:"job_id"` + ToRef string `json:"to_ref"` + Error string `json:"error"` + } `json:"results"` +} + +func callWebhook(t *testing.T, srv *httptest.Server, body string) (int, webhookResponse) { + t.Helper() + req, err := http.NewRequest(http.MethodPost, srv.URL, strings.NewReader(body)) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer ci-key") + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + var out webhookResponse + require.NoError(t, json.NewDecoder(resp.Body).Decode(&out)) + return resp.StatusCode, out +} + +func TestWebhookWaitsForAutoJob(t *testing.T) { + e := newJobEnv(t) + e.setAuto(1) + srv := e.webhookServer(t) + + type reply struct { + code int + body webhookResponse + } + done := make(chan reply, 1) + go func() { + code, body := callWebhook(t, srv, `{"image":"`+slusk+`:ignored","tag":"2.0.0"}`) + done <- reply{code, body} + }() + + e.waitStarted(t) + select { + case <-done: + t.Fatal("the webhook answered before the job finished") + case <-time.After(100 * time.Millisecond): + } + close(e.exec.release) + + var r reply + select { + case r = <-done: + case <-time.After(waitFor): + t.Fatal("the webhook never answered") + } + require.Equal(t, http.StatusOK, r.code) + require.Len(t, r.body.Results, 1) + assert.Equal(t, "succeeded", r.body.Results[0].Status) + assert.Equal(t, slusk+":2.0.0", r.body.Results[0].ToRef) + + job := e.job(t, r.body.Results[0].JobID) + assert.Equal(t, model.TriggerWebhook, job.Trigger) + assert.Equal(t, model.JobSucceeded, job.Status) + assert.Equal(t, slusk+":2.0.0@sha256:e2e", job.ToRef) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Empty(t, svc.AvailableRef, "the job deployed the suggestion, so it is cleared") +} + +func TestWebhookFailedAutoJobIs500(t *testing.T) { + e := newJobEnv(t) + e.setAuto(1) + srv := e.webhookServer(t) + e.agent.stopConn() // no agent connection: the job fails at once with "agent offline" + require.Eventually(t, func() bool { + svc, _ := e.store.GetService(context.Background(), 1) + _, err := e.m.hub.Route(svc) + return err != nil + }, waitFor, pollEvery) + + code, body := callWebhook(t, srv, `{"image":"`+slusk+`","tag":"2.0.0"}`) + + assert.Equal(t, http.StatusInternalServerError, code) + require.Len(t, body.Results, 1) + assert.Equal(t, "failed", body.Results[0].Status) + assert.Equal(t, "agent offline", body.Results[0].Error) + assert.Equal(t, model.JobFailed, e.job(t, body.Results[0].JobID).Status) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Equal(t, slusk+":2.0.0", svc.AvailableRef, "a failed job keeps the suggestion") +} diff --git a/internal/master/config/config.go b/internal/master/config/config.go index 0739ec1..77acd6f 100644 --- a/internal/master/config/config.go +++ b/internal/master/config/config.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "time" ) // Config holds the master's runtime configuration. @@ -20,8 +21,18 @@ type Config struct { Port string AgentPort string LogLevel string + + // UpdaterAPIKey is the bearer token for POST /update (UPDATER_API_KEY). + // Empty disables the webhook. + UpdaterAPIKey string + // UpdateTimeout bounds how long one webhook request waits for its auto + // jobs (UPDATE_TIMEOUT, default 5m). + UpdateTimeout time.Duration } +// defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset. +const defaultUpdateTimeout = 5 * time.Minute + // Load reads and validates the master configuration, failing fast on // anything missing so a half-configured master never starts. func Load() (*Config, error) { @@ -37,6 +48,8 @@ func Load() (*Config, error) { Port: getenvDefault("PORT", "8080"), AgentPort: getenvDefault("AGENT_PORT", "8090"), LogLevel: getenvDefault("LOG_LEVEL", "info"), + UpdaterAPIKey: os.Getenv("UPDATER_API_KEY"), + UpdateTimeout: defaultUpdateTimeout, } for _, r := range []struct{ name, value string }{ {"DATABASE_URL", cfg.DatabaseURL}, @@ -57,6 +70,13 @@ func Load() (*Config, error) { if !strings.HasPrefix(cfg.AgentURL, "wss://") { return nil, fmt.Errorf("AGENT_URL must start with wss://") } + if v := os.Getenv("UPDATE_TIMEOUT"); v != "" { + d, err := time.ParseDuration(v) + if err != nil || d <= 0 { + return nil, fmt.Errorf("UPDATE_TIMEOUT must be a positive duration such as 5m, got %q", v) + } + cfg.UpdateTimeout = d + } return cfg, nil } diff --git a/internal/master/config/config_test.go b/internal/master/config/config_test.go index 3ef5614..977ee67 100644 --- a/internal/master/config/config_test.go +++ b/internal/master/config/config_test.go @@ -2,6 +2,7 @@ package config import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -21,7 +22,7 @@ func setAll(t *testing.T) { for k, v := range required { t.Setenv(k, v) } - for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL"} { + for _, k := range []string{"PORT", "AGENT_PORT", "AGENT_IMAGE", "LOG_LEVEL", "UPDATER_API_KEY", "UPDATE_TIMEOUT"} { t.Setenv(k, "") } } @@ -35,6 +36,29 @@ func TestLoad_Defaults(t *testing.T) { assert.Equal(t, "gitea.shcizo.se/shcizo/package-updater:latest", cfg.AgentImage) assert.Equal(t, "info", cfg.LogLevel) assert.Equal(t, "updater-admins", cfg.AllowedGroup) + assert.Empty(t, cfg.UpdaterAPIKey) + assert.Equal(t, 5*time.Minute, cfg.UpdateTimeout) +} + +func TestLoad_WebhookSettings(t *testing.T) { + setAll(t) + t.Setenv("UPDATER_API_KEY", "s3cret") + t.Setenv("UPDATE_TIMEOUT", "90s") + cfg, err := Load() + require.NoError(t, err) + assert.Equal(t, "s3cret", cfg.UpdaterAPIKey) + assert.Equal(t, 90*time.Second, cfg.UpdateTimeout) +} + +func TestLoad_InvalidUpdateTimeout(t *testing.T) { + for _, v := range []string{"abc", "0s", "-5m", "30"} { + t.Run(v, func(t *testing.T) { + setAll(t) + t.Setenv("UPDATE_TIMEOUT", v) + _, err := Load() + assert.ErrorContains(t, err, "UPDATE_TIMEOUT must be a positive duration") + }) + } } func TestLoad_EachRequiredVariable(t *testing.T) { diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go index 49086c7..6b2fe1b 100644 --- a/internal/master/jobs/jobs.go +++ b/internal/master/jobs/jobs.go @@ -68,9 +68,34 @@ func target(actualRef string) (repo, tag string, ok bool) { return ref.Repo, "", false } +// FollowedRef returns "repo:tag" for the tag a service with this actual ref +// follows (a bare image follows "latest"), or "" when it follows no tag. +func FollowedRef(actualRef string) string { + repo, tag, ok := target(actualRef) + if !ok { + return "" + } + return repo + ":" + tag +} + // Updatable returns nil or a *model.NotUpdatableError. It looks only at the -// service row, so the UI can call it for every page render. +// service row, so the UI can call it for every page render. It answers for an +// update to the service's own followed tag (the Update button). func (s *Service) Updatable(svc model.Service) error { + return s.updatable(svc, false) +} + +// AutoAllowed returns nil or a *model.NotUpdatableError saying why the +// service can never be updated automatically. Unlike Updatable it does not +// refuse a service that follows no tag: a webhook that names a tag can still +// update it. +func (s *Service) AutoAllowed(svc model.Service) error { + return s.updatable(svc, true) +} + +// updatable checks the service row. explicitRef means the caller supplies the +// target ref, so a service that follows no tag is fine. +func (s *Service) updatable(svc model.Service, explicitRef bool) error { switch { case svc.RemovedAt != nil: return &model.NotUpdatableError{Reason: model.ReasonRemoved} @@ -81,32 +106,49 @@ func (s *Service) Updatable(svc model.Service) error { case svc.Kind == protocol.KindContainer: return &model.NotUpdatableError{Reason: model.ReasonContainer} } - if _, _, ok := target(svc.ActualRef); !ok { + if _, _, ok := target(svc.ActualRef); !ok && !explicitRef { return &model.NotUpdatableError{Reason: model.ReasonNoTag} } return nil } +// StartRequest says who asks for an update and what to deploy. +type StartRequest struct { + Trigger model.JobTrigger + RequestedBy string // OIDC subject for manual jobs; "" otherwise + // Ref is the ref to deploy ("repo:tag"). Empty means the service's + // followed tag, derived from its actual ref. + Ref string +} + // Trigger starts a manual update of the service to its current tag. It -// returns the job, which may already be failed ("agent offline", "agent too -// old for updates"); a *model.NotUpdatableError when the service cannot be -// updated (no job created); the active job and model.ErrJobActive when one -// is queued or running; model.ErrNotFound for an unknown service. +// returns what Start returns, and model.ErrNotFound for an unknown service. func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) { svc, err := s.store.GetService(ctx, serviceID) if err != nil { return model.Job{}, err } - if err := s.Updatable(svc); err != nil { + return s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy}) +} + +// Start starts an update of svc. It returns the job, which may already be +// failed ("agent offline", "agent too old for updates"); a +// *model.NotUpdatableError when the service cannot be updated (no job +// created); the active job and model.ErrJobActive when one is queued or +// running. +func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) { + if err := s.updatable(svc, req.Ref != ""); err != nil { return model.Job{}, err } - repo, tag, _ := target(svc.ActualRef) // Updatable checked the tag - ref := repo + ":" + tag + ref := req.Ref + if ref == "" { + ref = FollowedRef(svc.ActualRef) // updatable checked the tag + } r, routeErr := s.hub.Route(svc) nj := model.NewJob{ ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: ref, - Trigger: model.TriggerManual, RequestedBy: requestedBy, CreatedAt: s.now(), + Trigger: req.Trigger, RequestedBy: req.RequestedBy, CreatedAt: s.now(), } if routeErr == nil { nj.AgentID = &r.AgentID diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go index 5e7468a..3918fb2 100644 --- a/internal/master/jobs/jobs_test.go +++ b/internal/master/jobs/jobs_test.go @@ -242,3 +242,102 @@ func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) { assert.Empty(t, h.starts) assert.Empty(t, pub.topics) } + +func TestStart_WebhookTriggerWithExplicitRef(t *testing.T) { + st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + job, err := s.Start(context.Background(), composeSvc, StartRequest{ + Trigger: model.TriggerWebhook, Ref: "gitea.shcizo.se/shcizo/slusk:2.0.0"}) + + require.NoError(t, err) + assert.Equal(t, model.JobQueued, job.Status) + require.Len(t, st.created, 1) + assert.Equal(t, model.TriggerWebhook, st.created[0].Trigger) + assert.Empty(t, st.created[0].RequestedBy) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", st.created[0].ToRef) + require.Len(t, h.starts, 1) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:2.0.0", h.starts[0].start.Ref) +} + +func TestStart_ExplicitRefBypassesOnlyTheNoTagRefusal(t *testing.T) { + tests := []struct { + name string + svc model.Service + reason string // "" means the job starts + }{ + {"digest only", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""}, + {"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved}, + {"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf}, + {"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt}, + {"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tt.svc.ID = 1 + st := &fakeStore{services: map[int64]model.Service{1: tt.svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + _, err := s.Start(context.Background(), tt.svc, StartRequest{Trigger: model.TriggerWebhook, Ref: "x/y:2"}) + + if tt.reason == "" { + require.NoError(t, err) + assert.Len(t, st.created, 1) + return + } + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, tt.reason, nu.Reason) + assert.Empty(t, st.created) + }) + } +} + +func TestStart_WithoutRefStillRefusesNoTag(t *testing.T) { + svc := model.Service{ID: 1, Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"} + s, _ := newService(&fakeStore{jobs: map[int64]model.Job{}}, &fakeHub{}) + + _, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerWebhook}) + + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, model.ReasonNoTag, nu.Reason) +} + +func TestAutoAllowed(t *testing.T) { + tests := []struct { + name string + svc model.Service + reason string + }{ + {"removed", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", RemovedAt: ptr(time.Now())}, model.ReasonRemoved}, + {"self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf}, + {"agent image repo", model.Service{Kind: protocol.KindCompose, ActualRef: selfImage}, model.ReasonSelf}, + {"built", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Built: true}, model.ReasonBuilt}, + {"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer}, + {"digest without tag is allowed", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, ""}, + {"tag", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:16"}, ""}, + } + s, _ := newService(&fakeStore{}, &fakeHub{}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := s.AutoAllowed(tt.svc) + if tt.reason == "" { + assert.NoError(t, err) + return + } + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Equal(t, tt.reason, nu.Reason) + }) + } +} + +func TestFollowedRef(t *testing.T) { + assert.Equal(t, "postgres:16", FollowedRef("postgres:16@sha256:abc")) + assert.Equal(t, "postgres:latest", FollowedRef("postgres")) + assert.Equal(t, "localhost:5000/foo:latest", FollowedRef("localhost:5000/foo")) + assert.Empty(t, FollowedRef("redis@sha256:abc")) +} diff --git a/internal/master/webhook/webhook.go b/internal/master/webhook/webhook.go new file mode 100644 index 0000000..f061374 --- /dev/null +++ b/internal/master/webhook/webhook.go @@ -0,0 +1,331 @@ +// Package webhook implements POST /update, the endpoint CI calls after +// pushing an image. It records the pushed tag as "available" on every service +// that uses the image, starts an update for the services in auto mode and +// waits for those updates to finish. +package webhook + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "errors" + "log/slog" + "net/http" + "regexp" + "strconv" + "strings" + "time" + + "github.com/shcizo/package-updater/internal/imageref" + "github.com/shcizo/package-updater/internal/logging" + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" +) + +// maxBodyBytes caps the request body; a valid request is a few hundred bytes. +const maxBodyBytes = 64 << 10 + +// defaultPollEvery is how often pending jobs are re-read from the store when +// no broker signal arrives (signals to a slow subscriber can be dropped). +const defaultPollEvery = 5 * time.Second + +// tagPattern is Docker's tag grammar. It keeps spaces, colons and "@" out of +// the ref the agent is asked to deploy. +var tagPattern = regexp.MustCompile(`^[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}$`) + +// Result statuses. +const ( + statusSucceeded = "succeeded" + statusFailed = "failed" + statusTimeout = "timeout" + statusMarked = "marked" + statusBusy = "busy" + statusRefused = "refused" +) + +// Store is the storage the webhook needs. *store.Store satisfies it. +type Store interface { + ServicesByRepo(ctx context.Context, repo string) ([]model.Service, error) + SetAvailable(ctx context.Context, id int64, a model.Available) error + GetJob(ctx context.Context, id int64) (model.Job, error) +} + +// Jobs starts updates. *jobs.Service satisfies it. +type Jobs interface { + Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) +} + +// Broker delivers change signals. *events.Broker satisfies it. +type Broker interface { + Subscribe() (<-chan events.Topic, func()) + Publish(t events.Topic) +} + +// Config holds the webhook settings. +type Config struct { + Token string // bearer token; empty disables the webhook + Timeout time.Duration // bounds the whole wait of one request + PublicURL string // base of job_url, without trailing slash +} + +// Handler serves POST /update. +type Handler struct { + store Store + jobs Jobs + broker Broker + cfg Config + log *slog.Logger + + now func() time.Time + pollEvery time.Duration +} + +// New returns the handler for POST /update. It does its own bearer-token +// check, so mount it outside the OIDC login. +func New(store Store, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) *Handler { + return &Handler{store: store, jobs: jobs, broker: broker, cfg: cfg, log: log, + now: time.Now, pollEvery: defaultPollEvery} +} + +type request struct { + Image string `json:"image"` + Tag string `json:"tag"` +} + +type result struct { + ServiceID int64 `json:"service_id"` + Service string `json:"service"` + Kind string `json:"kind"` + Mode string `json:"mode"` + Status string `json:"status"` + JobID int64 `json:"job_id,omitempty"` + JobURL string `json:"job_url,omitempty"` + ToRef string `json:"to_ref,omitempty"` + Error string `json:"error,omitempty"` + DurationMS int64 `json:"duration_ms"` +} + +type response struct { + RequestID string `json:"request_id"` + Image string `json:"image"` + Tag string `json:"tag,omitempty"` + Matched int `json:"matched"` + Results []result `json:"results"` +} + +// ServeHTTP handles one webhook call. See the README for the contract. +func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + started := h.now() + requestID := r.Header.Get("X-Request-ID") + if requestID == "" { + requestID = newRequestID() + } + w.Header().Set("X-Request-ID", requestID) + + if h.cfg.Token == "" { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "webhook disabled: UPDATER_API_KEY not set"}) + return + } + if !h.authorized(r) { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + + var req request + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"}) + return + } + if req.Image == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "image is required"}) + return + } + if req.Tag != "" && !tagPattern.MatchString(req.Tag) { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid tag"}) + return + } + + log := logging.FromContext(logging.WithRequestID(r.Context(), requestID), h.log) + repo := imageref.Parse(req.Image).Repo + // A client that disconnects must not abort marking or starting: that + // could leave a queued job no agent knows about. + work := context.WithoutCancel(r.Context()) + + services, err := h.store.ServicesByRepo(work, repo) + if err != nil { + log.Error("webhook_lookup_failed", "image", req.Image, "error", err) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "internal error"}) + return + } + if len(services) == 0 { + writeJSON(w, http.StatusNotFound, map[string]string{ + "error": "no service uses image " + repo, "request_id": requestID}) + return + } + + // Subscribe before any job exists so no completion signal is missed. + sub, cancel := h.broker.Subscribe() + defer cancel() + + resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)} + pending := map[int]int64{} // index into resp.Results -> job ID + for _, svc := range services { + res, job := h.handle(work, svc, req.Tag, started) + resp.Results = append(resp.Results, res) + if job != nil { + pending[len(resp.Results)-1] = job.ID + } + } + h.broker.Publish(events.Services) + + h.wait(r.Context(), sub, &resp, pending, started, log) + + status := http.StatusOK + counts := map[string]int{} + for _, res := range resp.Results { + counts[res.Status]++ + if res.Status == statusFailed || res.Status == statusTimeout { + status = http.StatusInternalServerError + } + } + log.Info("webhook_request", "image", req.Image, "tag", req.Tag, "matched", resp.Matched, + "succeeded", counts[statusSucceeded], "failed", counts[statusFailed], "timeout", counts[statusTimeout], + "marked", counts[statusMarked], "busy", counts[statusBusy], "refused", counts[statusRefused]) + writeJSON(w, status, resp) +} + +// handle records the available ref on svc and, in auto mode, starts an update. +// It returns the result and, if the job is still active, that job. +func (h *Handler) handle(ctx context.Context, svc model.Service, tag string, started time.Time) (result, *model.Job) { + res := result{ServiceID: svc.ID, Service: svc.DisplayName(), Kind: string(svc.Kind), Mode: svc.UpdateMode} + + // The ref is built from the service's own repo, never from the request + // string, so a request cannot point a service at another registry. + var ref string + if tag != "" { + ref = svc.ImageRepo + ":" + tag + } else { + ref = jobs.FollowedRef(svc.ActualRef) + } + res.ToRef = ref + if ref != "" { + err := h.store.SetAvailable(ctx, svc.ID, model.Available{Ref: ref, Source: model.SourceWebhook, At: h.now()}) + if err != nil { + h.log.Error("webhook_set_available_failed", "service_id", svc.ID, "error", err) + return h.finish(res, statusFailed, "could not record available version: "+err.Error(), started), nil + } + } + if svc.UpdateMode != model.ModeAuto { + return h.finish(res, statusMarked, "", started), nil + } + + job, err := h.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerWebhook, Ref: ref}) + var notUpdatable *model.NotUpdatableError + switch { + case errors.As(err, ¬Updatable): + return h.finish(res, statusRefused, notUpdatable.Reason, started), nil + case errors.Is(err, model.ErrJobActive): + h.setJob(&res, job) + return h.finish(res, statusBusy, err.Error(), started), nil + case err != nil: + h.log.Error("webhook_start_failed", "service_id", svc.ID, "error", err) + return h.finish(res, statusFailed, err.Error(), started), nil + } + h.setJob(&res, job) + if job.Active() { + return res, &job + } + h.resolveJob(&res, job, started) + return res, nil +} + +func (h *Handler) setJob(res *result, job model.Job) { + res.JobID = job.ID + res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10) +} + +func (h *Handler) finish(res result, status, msg string, started time.Time) result { + res.Status, res.Error = status, msg + res.DurationMS = h.now().Sub(started).Milliseconds() + return res +} + +// resolveJob sets the result from a job that is no longer active. +func (h *Handler) resolveJob(res *result, job model.Job, started time.Time) { + if job.Status == model.JobSucceeded { + *res = h.finish(*res, statusSucceeded, "", started) + return + } + *res = h.finish(*res, statusFailed, job.Error, started) +} + +// wait blocks until every pending job has finished, the timeout passes or +// the client goes away. Jobs still pending then get status "timeout"; the +// master never cancels or fails them. +func (h *Handler) wait(parent context.Context, sub <-chan events.Topic, resp *response, + pending map[int]int64, started time.Time, log *slog.Logger) { + if len(pending) == 0 { + return + } + ctx, cancel := context.WithTimeout(parent, h.cfg.Timeout) + defer cancel() + ticker := time.NewTicker(h.pollEvery) + defer ticker.Stop() + + for len(pending) > 0 { + select { + case <-ctx.Done(): + for i := range pending { + resp.Results[i] = h.finish(resp.Results[i], statusTimeout, "still running when the request timed out", started) + } + return + case t := <-sub: + if t != events.Jobs { + continue + } + case <-ticker.C: + } + for i, id := range pending { + job, err := h.store.GetJob(ctx, id) + if err != nil { + log.Warn("webhook_get_job_failed", "job_id", id, "error", err) + continue + } + if !job.Active() { + h.resolveJob(&resp.Results[i], job, started) + delete(pending, i) + } + } + } +} + +// authorized reports whether the request carries the configured bearer token. +// Both sides are hashed so the comparison takes the same time whatever the +// length of the supplied token. +func (h *Handler) authorized(r *http.Request) bool { + const prefix = "Bearer " + got, ok := strings.CutPrefix(r.Header.Get("Authorization"), prefix) + if !ok || got == "" { + return false + } + a, b := sha256.Sum256([]byte(got)), sha256.Sum256([]byte(h.cfg.Token)) + return subtle.ConstantTimeCompare(a[:], b[:]) == 1 +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} + +// newRequestID returns a random 128-bit hex ID. +func newRequestID() string { + var b [16]byte + _, _ = rand.Read(b[:]) + return hex.EncodeToString(b[:]) +} diff --git a/internal/master/webhook/webhook_test.go b/internal/master/webhook/webhook_test.go new file mode 100644 index 0000000..b2b5890 --- /dev/null +++ b/internal/master/webhook/webhook_test.go @@ -0,0 +1,475 @@ +package webhook + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const token = "s3cret" + +type availCall struct { + id int64 + a model.Available +} + +type fakeStore struct { + mu sync.Mutex + services []model.Service // returned for any repo that equals their ImageRepo + jobs map[int64]model.Job + avail []availCall + repos []string +} + +func (f *fakeStore) ServicesByRepo(_ context.Context, repo string) ([]model.Service, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.repos = append(f.repos, repo) + var out []model.Service + for _, s := range f.services { + if s.ImageRepo == repo { + out = append(out, s) + } + } + return out, nil +} + +func (f *fakeStore) SetAvailable(_ context.Context, id int64, a model.Available) error { + f.mu.Lock() + defer f.mu.Unlock() + f.avail = append(f.avail, availCall{id, a}) + return nil +} + +func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.jobs[id], nil +} + +func (f *fakeStore) setJob(j model.Job) { + f.mu.Lock() + defer f.mu.Unlock() + f.jobs[j.ID] = j +} + +type startCall struct { + svc model.Service + req jobs.StartRequest + ctx context.Context +} + +// fakeJobs hands out job IDs and applies per-service behaviour. +type fakeJobs struct { + mu sync.Mutex + store *fakeStore + calls []startCall + errFor map[int64]error // returned together with errJob[id] if set + jobFor map[int64]model.Job // status of the job returned by Start + next int64 +} + +func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, startCall{svc, req, ctx}) + if err, ok := f.errFor[svc.ID]; ok { + return f.jobFor[svc.ID], err + } + f.next++ + j := model.Job{ID: f.next, ServiceID: svc.ID, Status: model.JobQueued, Trigger: req.Trigger, ToRef: req.Ref} + if o, ok := f.jobFor[svc.ID]; ok { + o.ID = j.ID + j = o + } + f.store.mu.Lock() + f.store.jobs[j.ID] = j + f.store.mu.Unlock() + return j, nil +} + +func (f *fakeJobs) startCalls() []startCall { + f.mu.Lock() + defer f.mu.Unlock() + return append([]startCall(nil), f.calls...) +} + +type env struct { + store *fakeStore + jobs *fakeJobs + broker *events.Broker + h *Handler +} + +func newEnv(t *testing.T, services ...model.Service) *env { + t.Helper() + st := &fakeStore{services: services, jobs: map[int64]model.Job{}} + fj := &fakeJobs{store: st, errFor: map[int64]error{}, jobFor: map[int64]model.Job{}} + b := events.NewBroker() + h := New(st, fj, b, Config{Token: token, Timeout: 2 * time.Second, PublicURL: "https://updater.example.com"}, + slog.New(slog.NewTextHandler(io.Discard, nil))) + h.now = func() time.Time { return time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC) } + return &env{store: st, jobs: fj, broker: b, h: h} +} + +func (e *env) post(body string, headers ...string) *httptest.ResponseRecorder { + return e.postCtx(context.Background(), body, headers...) +} + +func (e *env) postCtx(ctx context.Context, body string, headers ...string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(body)).WithContext(ctx) + req.Header.Set("Authorization", "Bearer "+token) + for i := 0; i+1 < len(headers); i += 2 { + req.Header.Set(headers[i], headers[i+1]) + } + rec := httptest.NewRecorder() + e.h.ServeHTTP(rec, req) + return rec +} + +func decode(t *testing.T, rec *httptest.ResponseRecorder) response { + t.Helper() + var r response + require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &r), rec.Body.String()) + return r +} + +func compose(id int64, mode, actualRef string) model.Service { + return model.Service{ID: id, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web" + string(rune('a'+id)), + ImageRepo: "gitea.shcizo.se/shcizo/slusk", ActualRef: actualRef, UpdateMode: mode} +} + +const slusk = "gitea.shcizo.se/shcizo/slusk" + +func TestAuth(t *testing.T) { + tests := []struct { + name string + serverKey string + header string + wantStatus int + wantBody string + }{ + {"missing header", token, "", 401, `{"error":"unauthorized"}`}, + {"wrong scheme", token, "Basic " + token, 401, `{"error":"unauthorized"}`}, + {"wrong token", token, "Bearer nope", 401, `{"error":"unauthorized"}`}, + {"empty bearer", token, "Bearer ", 401, `{"error":"unauthorized"}`}, + {"key unset", "", "Bearer " + token, 503, `{"error":"webhook disabled: UPDATER_API_KEY not set"}`}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + e.h.cfg.Token = tt.serverKey + req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"`+slusk+`"}`)) + if tt.header != "" { + req.Header.Set("Authorization", tt.header) + } + rec := httptest.NewRecorder() + e.h.ServeHTTP(rec, req) + assert.Equal(t, tt.wantStatus, rec.Code) + assert.JSONEq(t, tt.wantBody, rec.Body.String()) + assert.Empty(t, e.store.repos, "nothing may be looked up before auth") + }) + } +} + +func TestBadRequest(t *testing.T) { + tests := []struct { + name, body, wantError string + }{ + {"invalid JSON", `{`, "invalid JSON body"}, + {"empty body", ``, "invalid JSON body"}, + {"empty image", `{"image":""}`, "image is required"}, + {"missing image", `{"tag":"v1"}`, "image is required"}, + {"oversized", `{"image":"` + strings.Repeat("a", 70<<10) + `"}`, "invalid JSON body"}, + {"tag with digest", `{"image":"` + slusk + `","tag":"v1@sha256:abc"}`, "invalid tag"}, + {"tag with space", `{"image":"` + slusk + `","tag":"v 1"}`, "invalid tag"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1")) + rec := e.post(tt.body) + assert.Equal(t, 400, rec.Code) + assert.JSONEq(t, `{"error":"`+tt.wantError+`"}`, rec.Body.String()) + assert.Empty(t, e.jobs.startCalls()) + }) + } +} + +func TestNoMatch(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1")) + for _, image := range []string{"gitea.shcizo.se/shcizo/other", "gitea.shcizo.se/shcizo/Slusk"} { + rec := e.post(`{"image":"` + image + `"}`) + assert.Equal(t, 404, rec.Code, image) + assert.Contains(t, rec.Body.String(), "no service uses image "+image) + } + assert.Empty(t, e.jobs.startCalls()) + assert.Empty(t, e.store.avail) +} + +func TestImageIsNormalisedBeforeMatching(t *testing.T) { + svc := compose(1, model.ModeManual, slusk+":1") + local := compose(2, model.ModeManual, "localhost:5000/foo:v1") + local.ImageRepo = "localhost:5000/foo" + for _, tt := range []struct{ image, wantRepo string }{ + {slusk, slusk}, + {slusk + ":2.0", slusk}, + {slusk + "@sha256:abc", slusk}, + {slusk + ":2.0@sha256:abc", slusk}, + {"localhost:5000/foo:v1", "localhost:5000/foo"}, + } { + e := newEnv(t, svc, local) + rec := e.post(`{"image":"` + tt.image + `"}`) + assert.Equal(t, 200, rec.Code, tt.image) + assert.Equal(t, []string{tt.wantRepo}, e.store.repos, tt.image) + assert.Equal(t, 1, decode(t, rec).Matched, tt.image) + } +} + +func TestManualServiceIsMarkedNotUpdated(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1.4.2@sha256:old")) + sub, cancel := e.broker.Subscribe() + defer cancel() + + rec := e.post(`{"image":"` + slusk + `:ignored-in-ref","tag":"1.5.0"}`) + + assert.Equal(t, 200, rec.Code) + assert.Empty(t, e.jobs.startCalls()) + require.Len(t, e.store.avail, 1) + assert.Equal(t, int64(1), e.store.avail[0].id) + assert.Equal(t, model.Available{Ref: slusk + ":1.5.0", Source: model.SourceWebhook, At: e.h.now()}, e.store.avail[0].a) + resp := decode(t, rec) + assert.Equal(t, 1, resp.Matched) + require.Len(t, resp.Results, 1) + r := resp.Results[0] + assert.Equal(t, statusMarked, r.Status) + assert.Equal(t, "slusk/webb", r.Service) + assert.Equal(t, "compose", r.Kind) + assert.Equal(t, "manual", r.Mode) + assert.Equal(t, slusk+":1.5.0", r.ToRef) + assert.Zero(t, r.JobID) + assert.NotEmpty(t, resp.RequestID) + assert.Equal(t, resp.RequestID, rec.Header().Get("X-Request-ID")) + assert.Equal(t, events.Services, <-sub) +} + +func TestRequestIDIsEchoed(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, slusk+":1")) + rec := e.post(`{"image":"`+slusk+`"}`, "X-Request-ID", "ci-123") + assert.Equal(t, "ci-123", decode(t, rec).RequestID) + assert.Equal(t, "ci-123", rec.Header().Get("X-Request-ID")) +} + +func TestAvailableRefWithoutTag(t *testing.T) { + tests := []struct { + name string + actualRef string + wantRef string // "" means SetAvailable must not be called + }{ + {"followed tag", slusk + ":1.4.2@sha256:old", slusk + ":1.4.2"}, + {"bare image follows latest", slusk, slusk + ":latest"}, + {"digest only", slusk + "@sha256:old", ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + e := newEnv(t, compose(1, model.ModeManual, tt.actualRef)) + rec := e.post(`{"image":"` + slusk + `"}`) + assert.Equal(t, 200, rec.Code) + if tt.wantRef == "" { + assert.Empty(t, e.store.avail) + return + } + require.Len(t, e.store.avail, 1) + assert.Equal(t, tt.wantRef, e.store.avail[0].a.Ref) + }) + } +} + +func TestAutoDigestOnlyServiceWithoutTagIsRefused(t *testing.T) { + svc := compose(1, model.ModeAuto, slusk+"@sha256:old") + e := newEnv(t, svc) + e.jobs.errFor[1] = &model.NotUpdatableError{Reason: model.ReasonNoTag} + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + assert.Empty(t, e.store.avail) + calls := e.jobs.startCalls() + require.Len(t, calls, 1) + assert.Empty(t, calls[0].req.Ref, "no request tag and nothing followed: the job layer decides") + r := decode(t, rec).Results[0] + assert.Equal(t, statusRefused, r.Status) + assert.Equal(t, model.ReasonNoTag, r.Error) +} + +func TestAutoJobTargetIsBuiltFromServiceRepo(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2@sha256:old")) + e.jobs.jobFor[1] = model.Job{Status: model.JobSucceeded} + + rec := e.post(`{"image":"` + slusk + `:other","tag":"2.0.0"}`) + + assert.Equal(t, 200, rec.Code) + calls := e.jobs.startCalls() + require.Len(t, calls, 1) + assert.Equal(t, slusk+":2.0.0", calls[0].req.Ref) + assert.Equal(t, model.TriggerWebhook, calls[0].req.Trigger) + assert.Empty(t, calls[0].req.RequestedBy) + require.Len(t, e.store.avail, 1) + assert.Equal(t, slusk+":2.0.0", e.store.avail[0].a.Ref) +} + +func TestAutoJobSucceeded(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) + e.broker.Publish(events.JobTopic(1)) // progress topic is ignored + e.broker.Publish(events.Jobs) + }() + rec := e.post(`{"image":"` + slusk + `","tag":"1.5.0"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusSucceeded, r.Status) + assert.Equal(t, int64(1), r.JobID) + assert.Equal(t, "https://updater.example.com/jobs/1", r.JobURL) + assert.Equal(t, "auto", r.Mode) +} + +func TestAutoJobFailedIs500(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobFailed, Error: "pull access denied"}) + e.broker.Publish(events.Jobs) + }() + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusFailed, r.Status) + assert.Equal(t, "pull access denied", r.Error) +} + +func TestJobFailedAtStartIsNotWaitedFor(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.jobFor[1] = model.Job{Status: model.JobFailed, Error: "agent offline"} + e.h.cfg.Timeout = time.Minute + + begin := time.Now() + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Less(t, time.Since(begin), 5*time.Second) + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusFailed, r.Status) + assert.Equal(t, "agent offline", r.Error) + assert.NotZero(t, r.JobID) +} + +func TestTimeout(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.cfg.Timeout = 50 * time.Millisecond + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 500, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusTimeout, r.Status) + assert.Equal(t, int64(1), r.JobID) + assert.Equal(t, model.JobQueued, e.store.jobs[1].Status, "the master must not touch a timed-out job") +} + +func TestDroppedSignalIsRecoveredByPolling(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.pollEvery = 5 * time.Millisecond + go func() { + assert.Eventually(t, func() bool { return len(e.jobs.startCalls()) == 1 }, time.Second, time.Millisecond) + e.store.setJob(model.Job{ID: 1, Status: model.JobSucceeded}) // no Publish + }() + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + assert.Equal(t, statusSucceeded, decode(t, rec).Results[0].Status) +} + +func TestBusyIsNotAFailure(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.errFor[1] = model.ErrJobActive + e.jobs.jobFor[1] = model.Job{ID: 42, Status: model.JobRunning} + e.h.cfg.Timeout = time.Minute + + rec := e.post(`{"image":"` + slusk + `"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusBusy, r.Status) + assert.Equal(t, int64(42), r.JobID) + assert.Equal(t, "https://updater.example.com/jobs/42", r.JobURL) + assert.NotEmpty(t, e.store.avail, "the available ref is still recorded") +} + +func TestRefusedIsNotAFailure(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.jobs.errFor[1] = &model.NotUpdatableError{Reason: model.ReasonSelf} + + rec := e.post(`{"image":"` + slusk + `","tag":"2"}`) + + assert.Equal(t, 200, rec.Code) + r := decode(t, rec).Results[0] + assert.Equal(t, statusRefused, r.Status) + assert.Equal(t, model.ReasonSelf, r.Error) + assert.Zero(t, r.JobID) + assert.NotEmpty(t, e.store.avail) +} + +func TestMixedAutoAndManual(t *testing.T) { + e := newEnv(t, + compose(1, model.ModeManual, slusk+":1"), + compose(2, model.ModeAuto, slusk+":1"), + compose(3, model.ModeAuto, slusk+":1")) + e.jobs.jobFor[2] = model.Job{Status: model.JobSucceeded} + e.jobs.jobFor[3] = model.Job{Status: model.JobFailed, Error: "boom"} + + rec := e.post(`{"image":"` + slusk + `","tag":"2"}`) + + assert.Equal(t, 500, rec.Code) + resp := decode(t, rec) + assert.Equal(t, 3, resp.Matched) + got := []string{resp.Results[0].Status, resp.Results[1].Status, resp.Results[2].Status} + assert.Equal(t, []string{statusMarked, statusSucceeded, statusFailed}, got) + assert.Len(t, e.store.avail, 3) +} + +func TestClientCancelStopsWaitingButNotStarting(t *testing.T) { + e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) + e.h.cfg.Timeout = time.Minute + ctx, cancel := context.WithCancel(context.Background()) + cancel() // the client is already gone when the handler runs + + begin := time.Now() + rec := e.postCtx(ctx, `{"image":"`+slusk+`"}`) + + assert.Less(t, time.Since(begin), 5*time.Second) + assert.Equal(t, statusTimeout, decode(t, rec).Results[0].Status) + calls := e.jobs.startCalls() + require.Len(t, calls, 1, "the job must still be started") + assert.NoError(t, calls[0].ctx.Err(), "Start must run on a context the client cannot cancel") +}