From d0bf3e09d999bcb1361ebe41b74cb33412dec53f Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 10:28:34 +0200 Subject: [PATCH] feat(jobs): refuse poll and webhook jobs for drifted services, add restore (#10) --- internal/e2e/drift_test.go | 113 ++++++++++++++ internal/master/jobs/drift_test.go | 232 +++++++++++++++++++++++++++++ internal/master/jobs/jobs.go | 70 ++++++++- internal/master/jobs/jobs_test.go | 11 +- 4 files changed, 424 insertions(+), 2 deletions(-) create mode 100644 internal/e2e/drift_test.go create mode 100644 internal/master/jobs/drift_test.go diff --git a/internal/e2e/drift_test.go b/internal/e2e/drift_test.go new file mode 100644 index 0000000..50a37ca --- /dev/null +++ b/internal/e2e/drift_test.go @@ -0,0 +1,113 @@ +package e2e + +import ( + "context" + "io" + "log/slog" + "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/poller" + "github.com/shcizo/package-updater/internal/protocol" +) + +const ( + driftDesired = "sha256:" + "d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1" + driftOther = "sha256:" + "0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a" +) + +// seedDrifted replaces the environment's service with an auto service that was +// updated to slusk:1.4.2 at driftDesired but runs driftOther now. +func (e *jobEnv) seedDrifted() { + e.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", + AgentID: ptr(agentID), ImageRepo: slusk, UpdateMode: model.ModeAuto, + DesiredRef: slusk + ":1.4.2@" + driftDesired, + ActualRef: slusk + ":1.4.2", ActualDigest: driftOther}) +} + +// inventoryWrite is what the store does for an agent's upsert of the service. +func (e *jobEnv) inventoryWrite(id int64, actualRef, actualDigest string) { + e.store.mu.Lock() + defer e.store.mu.Unlock() + svc := e.store.svcRows[id] + svc.ActualRef, svc.ActualDigest, svc.AwaitingInventory = actualRef, actualDigest, false + e.store.svcRows[id] = svc +} + +func TestDriftedAutoServiceIsNotUpdatedByPollOrWebhook(t *testing.T) { + e := newJobEnv(t) + e.seedDrifted() + reg := &fixedRegistry{digests: map[string]string{slusk + ":1.4.2": "sha256:new"}} + p := poller.New(e.store, e.jobs, reg, e.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil))) + + p.Cycle(context.Background()) + + assert.Equal(t, 1, reg.calls) + assert.Zero(t, e.store.jobCount(), "polling starts no job for a drifted service") + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Equal(t, slusk+":1.4.2", svc.AvailableRef, "the suggestion is still recorded") + + code, body := callWebhook(t, e.webhookServer(t), `{"image":"`+slusk+`:ignored","tag":"2.0.0"}`) + + assert.Equal(t, 200, code) + require.Len(t, body.Results, 1) + assert.Equal(t, "refused", body.Results[0].Status) + assert.Contains(t, body.Results[0].Error, "drifted") + assert.Zero(t, e.store.jobCount(), "the webhook starts no job either") +} + +func TestRestoreDeploysThePinnedRefOnTheAgent(t *testing.T) { + e := newJobEnv(t) + e.seedDrifted() + close(e.exec.release) + + job, err := e.jobs.RestoreDesired(context.Background(), 1, "alice") + + require.NoError(t, err) + assert.Equal(t, model.TriggerManual, job.Trigger) + select { + case ref := <-e.exec.started: + assert.Equal(t, slusk+":1.4.2@"+driftDesired, ref, "the agent receives exactly the desired ref") + case <-time.After(waitFor): + t.Fatal("the agent never received the job") + } + done := e.waitStatus(t, job.ID, model.JobSucceeded) + assert.Equal(t, slusk+":1.4.2@"+driftDesired, done.ToRef) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Equal(t, driftOther, svc.LastTagDigest, "the drifted digest is remembered so polling does not redeploy it") + assert.False(t, svc.Drifted(), "the inventory has not caught up yet") + + e.inventoryWrite(1, slusk+":1.4.2", driftDesired) + svc, err = e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.False(t, svc.Drifted()) +} + +func TestNoDriftAfterASuccessfulJob(t *testing.T) { + e := newJobEnv(t) + close(e.exec.release) + + job, err := e.jobs.Trigger(context.Background(), 1, "alice") + + require.NoError(t, err) + e.waitStatus(t, job.ID, model.JobSucceeded) + svc, err := e.store.GetService(context.Background(), 1) + require.NoError(t, err) + assert.Equal(t, slusk+":1.4.2@"+e2eDigest, svc.DesiredRef) + assert.Equal(t, slusk+":1.4.2@sha256:old", svc.ActualRef, "the agent's inventory write has not arrived") + assert.False(t, svc.Drifted(), "between job.result and the inventory write the row is not drifted") + + e.inventoryWrite(1, slusk+":1.4.2", e2eDigest) + svc, _ = e.store.GetService(context.Background(), 1) + assert.False(t, svc.Drifted()) + + e.inventoryWrite(1, slusk+":1.4.2", driftOther) + svc, _ = e.store.GetService(context.Background(), 1) + assert.True(t, svc.Drifted(), "a later inventory write of another digest is drift") +} diff --git a/internal/master/jobs/drift_test.go b/internal/master/jobs/drift_test.go new file mode 100644 index 0000000..996429f --- /dev/null +++ b/internal/master/jobs/drift_test.go @@ -0,0 +1,232 @@ +package jobs + +import ( + "context" + "errors" + "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/protocol" +) + +const ( + driftRepo = "gitea.shcizo.se/shcizo/slusk" + driftD1 = "sha256:1111111111111111111111111111111111111111111111111111111111111111" + driftD2 = "sha256:2222222222222222222222222222222222222222222222222222222222222222" + driftD3 = "sha256:3333333333333333333333333333333333333333333333333333333333333333" +) + +// driftedSvc was updated to 1.4.2 at D1 but now runs 1.4.2 at D2. +func driftedSvc() model.Service { + svc := composeSvc + svc.ImageRepo = driftRepo + svc.DesiredRef = driftRepo + ":1.4.2@" + driftD1 + svc.ActualRef = driftRepo + ":1.4.2" + svc.ActualDigest = driftD2 + return svc +} + +func notUpdatableReason(t *testing.T, err error) string { + t.Helper() + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu), "got %v", err) + return nu.Reason +} + +func TestRestorable(t *testing.T) { + tests := []struct { + name string + mutate func(*model.Service) + reason string // "" means restorable + }{ + {"drifted", func(*model.Service) {}, ""}, + {"drifted service that follows no tag", func(s *model.Service) { + s.ActualRef = driftRepo + "@" + driftD2 + }, ""}, + {"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted}, + {"never updated", func(s *model.Service) { s.DesiredRef = "" }, model.ReasonNotDrifted}, + {"awaiting inventory", func(s *model.Service) { s.AwaitingInventory = true }, model.ReasonNotDrifted}, + {"removed", func(s *model.Service) { s.RemovedAt = ptr(time.Now()) }, model.ReasonNotDrifted}, + {"flagged self", func(s *model.Service) { s.Self = true }, model.ReasonSelf}, + {"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt}, + {"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer}, + {"desired ref is the agent image", func(s *model.Service) { + s.DesiredRef = selfImage + "@" + driftD1 + }, model.ReasonSelf}, + } + s, _ := newService(&fakeStore{}, &fakeHub{}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := driftedSvc() + tt.mutate(&svc) + err := s.Restorable(svc) + if tt.reason == "" { + assert.NoError(t, err) + return + } + assert.Equal(t, tt.reason, notUpdatableReason(t, err)) + }) + } +} + +func TestRestoreDesired_StartsAManualJobToTheDesiredRef(t *testing.T) { + svc := driftedSvc() + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + job, err := s.RestoreDesired(context.Background(), 1, "sub-1") + + require.NoError(t, err) + assert.Equal(t, model.JobQueued, job.Status) + require.Len(t, st.created, 1) + assert.Equal(t, svc.DesiredRef, st.created[0].ToRef) + assert.Equal(t, svc.ActualRef, st.created[0].FromRef) + assert.Equal(t, "sub-1", st.created[0].RequestedBy) + assert.Equal(t, model.TriggerManual, st.created[0].Trigger) + assert.Equal(t, []int{protocol.MinPinnedRefVersion}, h.mins) + require.Len(t, h.starts, 1) + assert.Equal(t, svc.DesiredRef, h.starts[0].start.Ref) +} + +func TestRestoreDesired_Refusals(t *testing.T) { + tests := []struct { + name string + mutate func(*model.Service) + reason string + }{ + {"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted}, + {"self", func(s *model.Service) { s.Self = true }, model.ReasonSelf}, + {"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt}, + {"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := driftedSvc() + tt.mutate(&svc) + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, pub := newService(st, h) + + _, err := s.RestoreDesired(context.Background(), 1, "sub-1") + + assert.Equal(t, tt.reason, notUpdatableReason(t, err)) + assert.Empty(t, st.created) + assert.Empty(t, h.starts) + assert.Empty(t, pub.topics) + assert.Empty(t, st.lastTag, "last_tag_digest untouched") + }) + } +} + +func TestRestoreDesired_UnknownService(t *testing.T) { + s, _ := newService(&fakeStore{services: map[int64]model.Service{}}, &fakeHub{}) + + _, err := s.RestoreDesired(context.Background(), 9, "sub-1") + + assert.ErrorIs(t, err, model.ErrNotFound) +} + +func TestRestoreDesired_ActiveJobReturnsItAndMarksNothing(t *testing.T) { + activeJob := model.Job{ID: 99, Status: model.JobRunning} + st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{99: activeJob}, active: &activeJob} + s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}) + + job, err := s.RestoreDesired(context.Background(), 1, "sub-1") + + assert.ErrorIs(t, err, model.ErrJobActive) + assert.Equal(t, int64(99), job.ID) + assert.Empty(t, st.lastTag) +} + +func TestRestoreDesired_OldAgentGivesAFailedJob(t *testing.T) { + st := &fakeStore{services: map[int64]model.Service{1: driftedSvc()}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}, agentProtocol: 2} + s, _ := newService(st, h) + + job, err := s.RestoreDesired(context.Background(), 1, "sub-1") + + require.NoError(t, err) + assert.Equal(t, model.JobFailed, job.Status) + assert.Equal(t, model.ErrAgentTooOldForPinned.Error(), job.Error) + assert.Empty(t, h.starts) +} + +func TestRestoreDesired_LastTagDigest(t *testing.T) { + tests := []struct { + name string + mutate func(*model.Service) + want string // "" means SetLastTagDigest is not called + }{ + {"drifted to another digest of the same tag", func(*model.Service) {}, driftD2}, + {"digest in the actual ref", func(s *model.Service) { + s.ActualRef, s.ActualDigest = driftRepo+":1.4.2@"+driftD3, "" + }, driftD3}, + {"already remembered", func(s *model.Service) { s.LastTagDigest = driftD2 }, ""}, + {"different tag", func(s *model.Service) { s.ActualRef = driftRepo + ":1.5.0" }, ""}, + {"actual digest malformed", func(s *model.Service) { s.ActualDigest = "sha256:abc" }, ""}, + {"desired follows no tag", func(s *model.Service) { s.DesiredRef = driftRepo + "@" + driftD1 }, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + svc := driftedSvc() + tt.mutate(&svc) + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}) + + _, err := s.RestoreDesired(context.Background(), 1, "sub-1") + + require.NoError(t, err) + if tt.want == "" { + assert.Empty(t, st.lastTag) + return + } + assert.Equal(t, map[int64]string{1: tt.want}, st.lastTag) + }) + } +} + +func TestStart_RefusesPollAndWebhookWhileDrifted(t *testing.T) { + tests := []struct { + trigger model.JobTrigger + refused bool + }{ + {model.TriggerPoll, true}, + {model.TriggerWebhook, true}, + {model.TriggerManual, false}, + } + for _, tt := range tests { + t.Run(string(tt.trigger), func(t *testing.T) { + svc := driftedSvc() + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}} + s, _ := newService(st, h) + + _, err := s.Start(context.Background(), svc, StartRequest{Trigger: tt.trigger, Ref: driftRepo + ":2.0.0"}) + + if tt.refused { + assert.Equal(t, model.ReasonDrifted, notUpdatableReason(t, err)) + assert.Empty(t, st.created) + return + } + require.NoError(t, err) + assert.Len(t, st.created, 1) + }) + } +} + +func TestStart_PollAllowedWhenNotDrifted(t *testing.T) { + svc := driftedSvc() + svc.AwaitingInventory = true + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + s, _ := newService(st, &fakeHub{route: model.Route{AgentID: 7, ConnID: 70}}) + + _, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerPoll}) + + require.NoError(t, err) + assert.Len(t, st.created, 1) +} diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go index a1bb514..92e00e7 100644 --- a/internal/master/jobs/jobs.go +++ b/internal/master/jobs/jobs.go @@ -21,6 +21,7 @@ type Store interface { FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) GetJob(ctx context.Context, id int64) (model.Job, error) GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error) + SetLastTagDigest(ctx context.Context, id int64, digest string) error } // Hub routes jobs to agent connections. *hub.Hub satisfies it. @@ -114,6 +115,69 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error { return nil } +// Restorable returns nil or a *model.NotUpdatableError saying why the service +// cannot be restored to its desired version: it must have drifted, and the +// desired version is deployed like an explicit ref, so a service that follows +// no tag is fine. The service page calls it for every render. +func (s *Service) Restorable(svc model.Service) error { + if !svc.Drifted() { + return &model.NotUpdatableError{Reason: model.ReasonNotDrifted} + } + if err := s.updatable(svc, true); err != nil { + return err + } + if imageref.Parse(svc.DesiredRef).Repo == s.selfRepo { + return &model.NotUpdatableError{Reason: model.ReasonSelf} + } + return nil +} + +// RestoreDesired starts a manual job that deploys the drifted service's +// desired ref again. The ref is read from the row, never from the caller. It +// returns model.ErrNotFound for an unknown service and otherwise what Start +// returns, with a *model.NotUpdatableError from Restorable if it cannot be +// restored. +// +// When the service had drifted to another digest of the same tag, that digest +// becomes last_tag_digest, so the poller does not deploy it again before the +// registry shows a newer one (the same rule as for any rollback). +func (s *Service) RestoreDesired(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.Restorable(svc); err != nil { + return model.Job{}, err + } + job, err := s.Start(ctx, svc, StartRequest{Trigger: model.TriggerManual, RequestedBy: requestedBy, Ref: svc.DesiredRef}) + if err != nil { + return job, err + } + if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest { + if err := s.store.SetLastTagDigest(ctx, svc.ID, d); err != nil { + s.log.Warn("last_tag_digest_not_set", "service_id", svc.ID, "error", err) + } + } + return job, nil +} + +// driftedTagDigest returns the digest a drifted service runs when it runs the +// same tag as desired at another digest, else "". +func driftedTagDigest(svc model.Service) string { + tag := FollowedRef(svc.DesiredRef) + if tag == "" || tag != FollowedRef(svc.ActualRef) { + return "" + } + d := imageref.Parse(svc.ActualRef).Digest + if d == "" { + d = svc.ActualDigest + } + if !imageref.IsDigest(d) || d == imageref.Parse(svc.DesiredRef).Digest { + return "" + } + return d +} + // StartRequest says who asks for an update and what to deploy. type StartRequest struct { Trigger model.JobTrigger @@ -161,8 +225,12 @@ func (s *Service) TriggerVersion(ctx context.Context, serviceID, versionID int64 // speaks protocol version 2); a // *model.NotUpdatableError when the service cannot be updated (no job // created); the active job and model.ErrJobActive when one is queued or -// running. +// running. Poll and webhook jobs are refused for a drifted service; only a +// person starts one then. func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest) (model.Job, error) { + if req.Trigger != model.TriggerManual && svc.Drifted() { + return model.Job{}, &model.NotUpdatableError{Reason: model.ReasonDrifted} + } if err := s.updatable(svc, req.Ref != ""); err != nil { return model.Job{}, err } diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go index ed8509b..241a59e 100644 --- a/internal/master/jobs/jobs_test.go +++ b/internal/master/jobs/jobs_test.go @@ -23,7 +23,8 @@ type fakeStore struct { active *model.Job // returned with ErrJobActive when set versions map[int64]model.ImageVersion - revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these + revokedAgents map[int64]bool // CreateJob returns ErrAgentOffline for these + lastTag map[int64]string // SetLastTagDigest calls: service ID to digest } func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) { @@ -63,6 +64,14 @@ func (f *fakeStore) GetImageVersion(_ context.Context, id int64) (model.ImageVer return v, nil } +func (f *fakeStore) SetLastTagDigest(_ context.Context, id int64, digest string) error { + if f.lastTag == nil { + f.lastTag = map[int64]string{} + } + f.lastTag[id] = digest + return nil +} + func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil } type startCall struct {