From 19a21da7bf35740c68cfa447e667a4a21161a048 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 10:28:34 +0200 Subject: [PATCH] feat(web): restore, accept and merge endpoints for drifted and moved services (#10) --- internal/master/web/drift_test.go | 287 +++++++++++++++++++++++++++++ internal/master/web/server.go | 171 ++++++++++++++++- internal/master/web/server_test.go | 10 + 3 files changed, 466 insertions(+), 2 deletions(-) create mode 100644 internal/master/web/drift_test.go diff --git a/internal/master/web/drift_test.go b/internal/master/web/drift_test.go new file mode 100644 index 0000000..3153889 --- /dev/null +++ b/internal/master/web/drift_test.go @@ -0,0 +1,287 @@ +package web + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "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/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +const ( + driftDesired = "sha256:1111111111111111111111111111111111111111111111111111111111111111" + driftActual = "sha256:2222222222222222222222222222222222222222222222222222222222222222" +) + +type acceptCall struct { + id int64 + actualRef, actualDigest, desiredRef string +} + +func (f *fakeStore) AcceptRunning(_ context.Context, id int64, actualRef, actualDigest, desired string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.acceptCalls = append(f.acceptCalls, acceptCall{id, actualRef, actualDigest, desired}) + return f.acceptErr +} + +func (f *fakeStore) MoveCandidates(context.Context) ([]model.Move, error) { return f.moves, nil } + +func (f *fakeStore) MergeSources(context.Context, int64) ([]model.Service, error) { + return f.sources, nil +} + +func (f *fakeStore) MergeService(_ context.Context, from, to int64) error { + f.mu.Lock() + defer f.mu.Unlock() + f.merged = append(f.merged, [2]int64{from, to}) + return f.mergeErr +} + +func (f *fakeJobs) Restorable(model.Service) error { return f.restorable } + +func (f *fakeJobs) RestoreDesired(_ context.Context, serviceID int64, by string) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.requestedBy = by + f.restored = append(f.restored, serviceID) + return f.job, f.err +} + +// driftService makes the first fixture service a drifted one: updated to +// 1.4.2 at driftDesired, now running 1.4.2 at driftActual. +func driftService(store *fakeStore) { + store.services[0].DesiredRef = "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired + store.services[0].ActualDigest = driftActual +} + +func postTo(t *testing.T, srv *httptest.Server, path string, form url.Values) (*http.Response, string) { + t.Helper() + form.Set("csrf_token", "csrf") + resp, err := noRedirectClient().PostForm(srv.URL+path, form) + require.NoError(t, err) + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return resp, string(body) +} + +func TestRestoreDesired(t *testing.T) { + tests := []struct { + name string + job model.Job + err error + path string + wantCode int + wantLocation string + wantBody string + }{ + {name: "started", job: model.Job{ID: 42}, path: "/services/1/restore", + wantCode: http.StatusSeeOther, wantLocation: "/jobs/42"}, + {name: "already running", job: model.Job{ID: 41}, err: model.ErrJobActive, path: "/services/1/restore", + wantCode: http.StatusSeeOther, wantLocation: "/jobs/41"}, + {name: "refused", err: &model.NotUpdatableError{Reason: model.ReasonNotDrifted}, path: "/services/1/restore", + wantCode: http.StatusConflict, wantBody: "Cannot restore: runs its desired version."}, + {name: "unknown service", err: model.ErrNotFound, path: "/services/1/restore", wantCode: http.StatusNotFound}, + {name: "invalid id", path: "/services/x/restore", wantCode: http.StatusBadRequest}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv, _, _, _, jobs := newTestEnv(t) + jobs.job, jobs.err = tt.job, tt.err + + resp, body := postTo(t, srv, tt.path, url.Values{}) + + assert.Equal(t, tt.wantCode, resp.StatusCode) + assert.Equal(t, tt.wantLocation, resp.Header.Get("Location")) + assert.Contains(t, body, tt.wantBody) + if tt.wantCode != http.StatusBadRequest { + assert.Equal(t, "alice", jobs.triggeredBy()) + } + }) + } +} + +func TestAcceptRunning_WritesRunningRefAndPublishes(t *testing.T) { + srv, store, _, broker, _ := newTestEnv(t) + driftService(store) + ch, cancel := broker.Subscribe() + defer cancel() + + resp, _ := postTo(t, srv, "/services/1/accept", url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}}) + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/services/1", resp.Header.Get("Location")) + assert.Equal(t, []acceptCall{{ + id: 1, actualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2", actualDigest: driftActual, + desiredRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual, + }}, store.acceptCalls, "the value written comes from the row, not the form") + select { + case topic := <-ch: + assert.Equal(t, events.Services, topic) + case <-time.After(time.Second): + t.Fatal("no services signal") + } +} + +func TestAcceptRunning_ReturnToOverview(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + driftService(store) + + resp, _ := postTo(t, srv, "/services/1/accept", + url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}, "return": {"overview"}}) + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/", resp.Header.Get("Location")) +} + +func TestAcceptRunning_Refusals(t *testing.T) { + tests := []struct { + name string + path string + form url.Values + notDrifted bool + acceptErr error + wantCode int + wantBody string + }{ + {name: "missing actual", path: "/services/1/accept", form: url.Values{}, wantCode: http.StatusBadRequest}, + {name: "stale actual", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.3.0"}}, + wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."}, + {name: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}}, + notDrifted: true, wantCode: http.StatusConflict, wantBody: "Cannot accept: the service runs its desired version."}, + {name: "changed under our feet", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}}, + acceptErr: model.ErrConflict, wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."}, + {name: "unknown service", path: "/services/9/accept", form: url.Values{"actual": {"x"}}, wantCode: http.StatusNotFound}, + {name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}}, + acceptErr: model.ErrNotFound, wantCode: http.StatusNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + if !tt.notDrifted { + driftService(store) + } + store.acceptErr = tt.acceptErr + + resp, body := postTo(t, srv, tt.path, tt.form) + + assert.Equal(t, tt.wantCode, resp.StatusCode) + assert.Contains(t, body, tt.wantBody) + if tt.acceptErr == nil { + assert.Empty(t, store.acceptCalls) + } + }) + } +} + +func TestMergeService_RedirectsToTheSurvivorAndPublishes(t *testing.T) { + srv, store, _, broker, _ := newTestEnv(t) + ch, cancel := broker.Subscribe() + defer cancel() + + resp, _ := postTo(t, srv, "/services/1/merge", url.Values{"from": {"17"}}) + + assert.Equal(t, http.StatusSeeOther, resp.StatusCode) + assert.Equal(t, "/services/17", resp.Header.Get("Location")) + assert.Equal(t, [][2]int64{{17, 1}}, store.merged) + var got []events.Topic + for len(got) < 2 { + select { + case topic := <-ch: + got = append(got, topic) + case <-time.After(time.Second): + t.Fatalf("signals so far: %v", got) + } + } + assert.ElementsMatch(t, []events.Topic{events.Services, events.Jobs}, got) +} + +func TestMergeService_Refusals(t *testing.T) { + tests := []struct { + name string + id, from string + err error + wantCode int + wantBody string + }{ + {name: "invalid from", id: "1", from: "x", wantCode: http.StatusBadRequest}, + {name: "empty from", id: "1", from: "", wantCode: http.StatusBadRequest}, + {name: "refused", id: "1", from: "17", err: &model.MergeRefusedError{Reason: model.ReasonMergeLiveOld}, + wantCode: http.StatusConflict, wantBody: "Cannot merge: the old service still exists."}, + {name: "job active", id: "1", from: "17", err: model.ErrJobActive, + wantCode: http.StatusConflict, wantBody: "Cannot merge: an update is already in progress for this service."}, + {name: "agent revoked", id: "1", from: "17", err: model.ErrRevoked, wantCode: http.StatusConflict, wantBody: "Cannot merge:"}, + {name: "not found", id: "1", from: "17", err: model.ErrNotFound, wantCode: http.StatusNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + srv, store, _, _, _ := newTestEnv(t) + store.mergeErr = tt.err + + resp, body := postTo(t, srv, "/services/"+tt.id+"/merge", url.Values{"from": {tt.from}}) + + assert.Equal(t, tt.wantCode, resp.StatusCode) + assert.Contains(t, body, tt.wantBody) + assert.Empty(t, resp.Header.Get("Location")) + }) + } +} + +// serviceDetail's data is what the service template reads. +func TestServiceDetail_DriftAndMoveData(t *testing.T) { + removed := time.Now() + live := model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1"} + gone := model.Service{ID: 2, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", ActualRef: "slusk:1", RemovedAt: &removed} + swarm := model.Service{ID: 3, Kind: protocol.KindSwarm, SwarmService: "gitea", ActualRef: "gitea:1"} + container := model.Service{ID: 4, Kind: protocol.KindContainer, ContainerName: "adhoc", ActualRef: "redis:7"} + self := model.Service{ID: 5, Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", ActualRef: "pu:1", Self: true} + move := model.Move{FromID: 2, ToID: 1, FromAgent: "old", ToAgent: "new"} + sources := []model.Service{gone} + + tests := []struct { + name string + id int64 + restorable error + wantBlocked string + wantFrom *model.Move + wantTo *model.Move + wantSources []model.Service + }{ + {name: "live row gets the suggestion and merge choices", id: 1, wantFrom: &move, wantSources: sources}, + {name: "removed row gets where it went", id: 2, wantTo: &move}, + {name: "swarm service can merge, no suggestion", id: 3, wantSources: sources}, + {name: "container cannot merge", id: 4}, + {name: "self cannot merge", id: 5}, + {name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonSelf}, + wantBlocked: model.ReasonSelf, wantFrom: &move, wantSources: sources}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + store := &fakeStore{services: []model.Service{live, gone, swarm, container, self}, + moves: []model.Move{move}, sources: sources} + jobs := &fakeJobs{restorable: tt.restorable} + s, err := New(store, &fakeHub{}, jobs, events.NewBroker(), Config{}, slog.New(slog.NewTextHandler(io.Discard, nil))) + require.NoError(t, err) + req := httptest.NewRequest(http.MethodGet, "/services/"+strconv.FormatInt(tt.id, 10), nil) + + data, err := s.serviceDetail(req, tt.id) + + require.NoError(t, err) + assert.Equal(t, tt.wantBlocked, data.RestoreBlocked) + assert.Equal(t, tt.wantFrom, data.MoveFrom) + assert.Equal(t, tt.wantTo, data.MoveTo) + assert.Equal(t, tt.wantSources, data.MergeSources) + }) + } +} diff --git a/internal/master/web/server.go b/internal/master/web/server.go index 4bf6a39..ce16b5d 100644 --- a/internal/master/web/server.go +++ b/internal/master/web/server.go @@ -18,6 +18,7 @@ import ( "github.com/shcizo/package-updater/internal/master/events" "github.com/shcizo/package-updater/internal/master/model" "github.com/shcizo/package-updater/internal/master/webauth" + "github.com/shcizo/package-updater/internal/protocol" ) //go:embed templates/*.html static/* @@ -40,6 +41,16 @@ type Store interface { RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error) // SetUpdateMode returns model.ErrNotFound for an unknown or removed service. SetUpdateMode(ctx context.Context, id int64, mode string) error + // AcceptRunning sets the desired ref of a drifted service; model.ErrNotFound + // for an unknown or removed service, model.ErrConflict when the running + // version is no longer actualRef@actualDigest. + AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error + // MoveCandidates returns the suggested moves of removed services. + MoveCandidates(ctx context.Context) ([]model.Move, error) + // MergeSources returns the removed services that could be merged into toID. + MergeSources(ctx context.Context, toID int64) ([]model.Service, error) + // MergeService moves the live service toID onto the removed fromID. + MergeService(ctx context.Context, fromID, toID int64) error GetJob(ctx context.Context, id int64) (model.Job, error) ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error) ListAgents(ctx context.Context) ([]model.Agent, error) @@ -63,6 +74,10 @@ type Jobs interface { // TriggerVersion updates to exactly the image version versionID of the // service's repo; model.ErrNotFound for an unknown service or version. TriggerVersion(ctx context.Context, serviceID, versionID int64, requestedBy string) (model.Job, error) + // Restorable explains why a drifted service cannot be restored. + Restorable(svc model.Service) error + // RestoreDesired deploys the service's desired ref again. + RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) } // Broker delivers change signals to open streams and accepts signals from @@ -118,6 +133,9 @@ func (s *Server) Register(mux *http.ServeMux, protect func(http.Handler) http.Ha route("GET /fragments/services/{id}/jobs", s.serviceJobsFragment) route("POST /services/{id}/update", s.updateService) route("POST /services/{id}/mode", s.setMode) + route("POST /services/{id}/restore", s.restoreDesired) + route("POST /services/{id}/accept", s.acceptRunning) + route("POST /services/{id}/merge", s.mergeService) route("GET /jobs", s.jobsPage) route("GET /fragments/jobs", s.jobsFragment) route("GET /jobs/{id}", s.jobPage) @@ -145,8 +163,14 @@ type pageData struct { Versions []versionRow Polling string // what the service page says about registry polling PollFailed bool // Polling is the last registry error - Jobs []model.Job - Job model.Job + // RestoreBlocked is why a drifted Service cannot be restored; "" when it + // can. Only meaningful when Service.Drifted(). + RestoreBlocked string + MoveFrom *model.Move // a live Service: the removed service it probably replaces + MoveTo *model.Move // a removed Service: the live service it probably moved to + MergeSources []model.Service // a live Compose or Swarm Service: removed ones to merge in + Jobs []model.Job + Job model.Job } func (s *Server) base(r *http.Request, title string) pageData { @@ -235,6 +259,14 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) { return data, err } data.AutoBlocked = s.autoBlocked(svc) + if err := s.jobs.Restorable(svc); errors.As(err, &nu) { + data.RestoreBlocked = nu.Reason + } else if err != nil { + return data, err + } + if err := s.moveData(r, &data); err != nil { + return data, err + } versions, err := s.store.RecentVersions(r.Context(), svc.ImageRepo, versionsLimit) if err != nil { return data, err @@ -245,6 +277,27 @@ func (s *Server) serviceDetail(r *http.Request, id int64) (pageData, error) { return data, err } +// moveData fills the move suggestion and the merge choices of the service page. +func (s *Server) moveData(r *http.Request, data *pageData) error { + svc := data.Service + moves, err := s.store.MoveCandidates(r.Context()) + if err != nil { + return err + } + for i, m := range moves { + switch svc.ID { + case m.ToID: + data.MoveFrom = &moves[i] + case m.FromID: + data.MoveTo = &moves[i] + } + } + if svc.RemovedAt == nil && !svc.Self && svc.Kind != protocol.KindContainer { + data.MergeSources, err = s.store.MergeSources(r.Context(), svc.ID) + } + return err +} + func (s *Server) servicePage(w http.ResponseWriter, r *http.Request) { id, ok := pathID(w, r) if !ok { @@ -308,6 +361,120 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) { } } +// restoreDesired deploys the desired ref of a drifted service again. +func (s *Server) restoreDesired(w http.ResponseWriter, r *http.Request) { + id, ok := pathID(w, r) + if !ok { + return + } + sess, _ := webauth.SessionFrom(r.Context()) + job, err := s.jobs.RestoreDesired(r.Context(), id, sess.Subject) + var nu *model.NotUpdatableError + switch { + case err == nil || errors.Is(err, model.ErrJobActive): + http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther) + case errors.As(err, &nu): + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot restore: "+nu.Reason+".") + default: + s.failLookup(w, err) + } +} + +// acceptRunning makes the running version of a drifted service the desired +// one. The form's actual field is the ref the user saw; it only guards against +// the service having changed since, the value written comes from the row. +func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) { + id, ok := pathID(w, r) + if !ok { + return + } + actual := r.PostFormValue("actual") + if actual == "" { + http.Error(w, "missing actual", http.StatusBadRequest) + return + } + svc, err := s.store.GetService(r.Context(), id) + if err != nil { + s.failLookup(w, err) + return + } + if !svc.Drifted() { + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot accept: the service runs its desired version.") + return + } + if actual != svc.ActualRef { + s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.") + return + } + desired := model.RunningRef(svc.ActualRef, svc.ActualDigest) + err = s.store.AcceptRunning(r.Context(), id, svc.ActualRef, svc.ActualDigest, desired) + switch { + case errors.Is(err, model.ErrConflict): + s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.") + return + case err != nil: + s.failLookup(w, err) + return + } + sess, _ := webauth.SessionFrom(r.Context()) + s.log.Info("drift_accepted", "service_id", id, "subject", sess.Subject, "desired", desired) + s.broker.Publish(events.Services) + if r.PostFormValue("return") == "overview" { + http.Redirect(w, r, "/", http.StatusSeeOther) + return + } + http.Redirect(w, r, "/services/"+strconv.FormatInt(id, 10), http.StatusSeeOther) +} + +// mergeService moves the live service {id} onto the removed service in the +// form field "from", which keeps its history, mode and desired ref. +func (s *Server) mergeService(w http.ResponseWriter, r *http.Request) { + id, ok := pathID(w, r) + if !ok { + return + } + from, err := strconv.ParseInt(r.PostFormValue("from"), 10, 64) + if err != nil { + http.Error(w, "invalid service", http.StatusBadRequest) + return + } + err = s.store.MergeService(r.Context(), from, id) + var refused *model.MergeRefusedError + switch { + case errors.As(err, &refused): + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+refused.Reason+".") + return + case errors.Is(err, model.ErrJobActive): + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: "+model.ErrJobActive.Error()+".") + return + case errors.Is(err, model.ErrRevoked): + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the agent is revoked.") + return + case errors.Is(err, model.ErrConflict): + s.renderServiceError(w, r, id, http.StatusConflict, "Cannot merge: the service changed; check it again.") + return + case err != nil: + s.failLookup(w, err) + return + } + sess, _ := webauth.SessionFrom(r.Context()) + s.log.Info("service_merged", "from", from, "to", id, "subject", sess.Subject) + s.broker.Publish(events.Services) + s.broker.Publish(events.Jobs) + http.Redirect(w, r, "/services/"+strconv.FormatInt(from, 10), http.StatusSeeOther) +} + +// renderServiceError renders the service page of id with msg as its error. +func (s *Server) renderServiceError(w http.ResponseWriter, r *http.Request, id int64, status int, msg string) { + data, err := s.serviceDetail(r, id) + if err != nil { + s.failLookup(w, err) + return + } + data.Error = msg + s.render(w, status, "service.html", "layout", data) +} + // setMode switches a service between manual and auto. Manual is always // allowed; auto only when jobs.AutoAllowed says so. func (s *Server) setMode(w http.ResponseWriter, r *http.Request) { diff --git a/internal/master/web/server_test.go b/internal/master/web/server_test.go index 8be45c6..6b6e31e 100644 --- a/internal/master/web/server_test.go +++ b/internal/master/web/server_test.go @@ -35,6 +35,13 @@ type fakeStore struct { revokedID int64 revokeFailed []int64 // job IDs RevokeAgent reports as failed versions []model.ImageVersion + + acceptCalls []acceptCall + acceptErr error + moves []model.Move + sources []model.Service + merged [][2]int64 // from, to + mergeErr error } func (f *fakeStore) RecentVersions(_ context.Context, repo string, limit int) ([]model.ImageVersion, error) { @@ -142,6 +149,9 @@ type fakeJobs struct { triggered []int64 // version IDs passed to TriggerVersion triggerSvc []int64 // service IDs passed to TriggerVersion plainCalls int // Trigger calls + + restorable error // returned by Restorable + restored []int64 // service IDs passed to RestoreDesired } func (f *fakeJobs) Updatable(model.Service) error { return f.updatable }