From 02de0b8cf3ff10cf14d209cfd4d5a88b98740d9c Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Mon, 12 Oct 2026 18:43:11 +0200 Subject: [PATCH] feat(master): start manual update jobs with self-protection (#7) --- internal/master/jobs/jobs.go | 122 +++++++++++++++++ internal/master/jobs/jobs_test.go | 211 ++++++++++++++++++++++++++++++ 2 files changed, 333 insertions(+) create mode 100644 internal/master/jobs/jobs.go create mode 100644 internal/master/jobs/jobs_test.go diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go new file mode 100644 index 0000000..f32e1ee --- /dev/null +++ b/internal/master/jobs/jobs.go @@ -0,0 +1,122 @@ +// Package jobs decides whether a service may be updated and starts update +// jobs on the right agent. +package jobs + +import ( + "context" + "log/slog" + "time" + + "github.com/shcizo/package-updater/internal/imageref" + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/model" + "github.com/shcizo/package-updater/internal/protocol" +) + +// Store is the storage jobs needs. *store.Store satisfies it. +type Store interface { + GetService(ctx context.Context, id int64) (model.Service, error) + CreateJob(ctx context.Context, j model.NewJob) (model.Job, error) + FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error) + GetJob(ctx context.Context, id int64) (model.Job, error) +} + +// Hub routes jobs to agent connections. *hub.Hub satisfies it. +type Hub interface { + // Route picks the connection for svc: model.ErrAgentOffline or + // model.ErrAgentTooOld when there is none. + Route(svc model.Service) (model.Route, error) + // StartJob sends job.start over exactly the routed connection. + // model.ErrAgentOffline if that connection is gone or was replaced. + StartJob(r model.Route, jobID int64, start protocol.JobStart) error +} + +// Publisher receives change signals. *events.Broker satisfies it. +type Publisher interface{ Publish(t events.Topic) } + +// Service starts update jobs. +type Service struct { + store Store + hub Hub + pub Publisher + selfRepo string + log *slog.Logger + now func() time.Time +} + +// New returns a Service. agentImage is the master's AGENT_IMAGE; services +// whose image repo equals its repo are the updater itself and are refused. +func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service { + return &Service{ + store: store, hub: hub, pub: pub, log: log, now: time.Now, + selfRepo: imageref.Parse(agentImage).Repo, + } +} + +// Updatable returns nil or a *model.NotUpdatableError. It looks only at the +// service row, so the UI can call it for every page render. +func (s *Service) Updatable(svc model.Service) error { + switch { + case svc.RemovedAt != nil: + return &model.NotUpdatableError{Reason: model.ReasonRemoved} + case svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo: + return &model.NotUpdatableError{Reason: model.ReasonSelf} + case svc.Kind == protocol.KindContainer: + return &model.NotUpdatableError{Reason: model.ReasonContainer} + case imageref.Parse(svc.ActualRef).Tag == "": + return &model.NotUpdatableError{Reason: model.ReasonNoTag} + } + return nil +} + +// 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. +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 model.Job{}, err + } + ref := imageref.Parse(svc.ActualRef) + target := ref.Repo + ":" + ref.Tag + + r, routeErr := s.hub.Route(svc) + nj := model.NewJob{ + ServiceID: svc.ID, FromRef: svc.ActualRef, ToRef: target, + Trigger: model.TriggerManual, RequestedBy: requestedBy, CreatedAt: s.now(), + } + if routeErr == nil { + nj.AgentID = &r.AgentID + } + job, err := s.store.CreateJob(ctx, nj) + if err != nil { + return job, err + } + + failure := routeErr + if failure == nil { + start := protocol.JobStart{Kind: svc.Kind, Ref: target} + if svc.Kind == protocol.KindSwarm { + start.SwarmClusterID, start.SwarmService = svc.SwarmClusterID, svc.SwarmService + } else { + start.ComposeProject, start.ComposeService = svc.ComposeProject, svc.ComposeService + } + if err := s.hub.StartJob(r, job.ID, start); err != nil { + s.log.Warn("job_start_failed", "job_id", job.ID, "error", err) + failure = model.ErrAgentOffline + } + } + if failure != nil { + if _, err := s.store.FailJob(ctx, job.ID, failure.Error(), s.now()); err != nil { + return model.Job{}, err + } + } + s.pub.Publish(events.Jobs) + s.pub.Publish(events.JobTopic(job.ID)) + return s.store.GetJob(ctx, job.ID) +} diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go new file mode 100644 index 0000000..c996276 --- /dev/null +++ b/internal/master/jobs/jobs_test.go @@ -0,0 +1,211 @@ +package jobs + +import ( + "context" + "errors" + "io" + "log/slog" + "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" +) + +type fakeStore struct { + services map[int64]model.Service + jobs map[int64]model.Job + created []model.NewJob + active *model.Job // returned with ErrJobActive when set +} + +func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) { + svc, ok := f.services[id] + if !ok { + return model.Service{}, model.ErrNotFound + } + return svc, nil +} + +func (f *fakeStore) CreateJob(_ context.Context, nj model.NewJob) (model.Job, error) { + if f.active != nil { + return *f.active, model.ErrJobActive + } + f.created = append(f.created, nj) + j := model.Job{ID: int64(len(f.created)), ServiceID: nj.ServiceID, AgentID: nj.AgentID, FromRef: nj.FromRef, + ToRef: nj.ToRef, Trigger: nj.Trigger, RequestedBy: nj.RequestedBy, Status: model.JobQueued, CreatedAt: nj.CreatedAt} + f.jobs[j.ID] = j + return j, nil +} + +func (f *fakeStore) FailJob(_ context.Context, id int64, msg string, at time.Time) (bool, error) { + j := f.jobs[id] + j.Status, j.Error, j.FinishedAt = model.JobFailed, msg, &at + f.jobs[id] = j + return true, nil +} + +func (f *fakeStore) GetJob(_ context.Context, id int64) (model.Job, error) { return f.jobs[id], nil } + +type startCall struct { + route model.Route + id int64 + start protocol.JobStart +} + +type fakeHub struct { + route model.Route + routeErr error + startErr error + starts []startCall +} + +func (h *fakeHub) Route(model.Service) (model.Route, error) { return h.route, h.routeErr } + +func (h *fakeHub) StartJob(r model.Route, id int64, s protocol.JobStart) error { + h.starts = append(h.starts, startCall{r, id, s}) + return h.startErr +} + +type fakePub struct{ topics []events.Topic } + +func (p *fakePub) Publish(t events.Topic) { p.topics = append(p.topics, t) } + +const selfImage = "gitea.shcizo.se/shcizo/package-updater:latest" + +var ( + composeSvc = model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", + AgentID: ptr(int64(7)), ActualRef: "gitea.shcizo.se/shcizo/slusk:1.4.2@sha256:d1"} + swarmSvc = model.Service{ID: 2, Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", + ActualRef: "gitea/gitea:1.27.3@sha256:87a6"} +) + +func ptr[T any](v T) *T { return &v } + +func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) { + pub := &fakePub{} + return New(st, h, pub, selfImage, slog.New(slog.NewTextHandler(io.Discard, nil))), pub +} + +func TestUpdatable(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}, + {"flagged self", model.Service{Kind: protocol.KindCompose, ActualRef: "a:1", Self: true}, model.ReasonSelf}, + {"same repo as the agent image", model.Service{Kind: protocol.KindCompose, + ActualRef: "gitea.shcizo.se/shcizo/package-updater:1.2.0@sha256:abc"}, model.ReasonSelf}, + {"container", model.Service{Kind: protocol.KindContainer, ActualRef: "redis:7"}, model.ReasonContainer}, + {"no tag", model.Service{Kind: protocol.KindCompose, ActualRef: "redis@sha256:abc"}, model.ReasonNoTag}, + {"plain", model.Service{Kind: protocol.KindCompose, ActualRef: "postgres:17.10"}, ""}, + {"registry with port", model.Service{Kind: protocol.KindSwarm, ActualRef: "localhost:5000/foo:v1"}, ""}, + } + s, _ := newService(&fakeStore{}, &fakeHub{}) + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := s.Updatable(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 TestTrigger(t *testing.T) { + ok := model.Route{AgentID: 7, ConnID: 70} + activeJob := model.Job{ID: 99, Status: model.JobRunning} + + tests := []struct { + name string + serviceID int64 + hub *fakeHub + active *model.Job + wantErr error + wantStatus model.JobStatus + wantError string + wantStart *protocol.JobStart + wantCreate bool + }{ + {name: "compose happy path", serviceID: 1, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true, + wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}}, + {name: "swarm", serviceID: 2, hub: &fakeHub{route: ok}, wantStatus: model.JobQueued, wantCreate: true, + wantStart: &protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27.3"}}, + {name: "active job", serviceID: 1, hub: &fakeHub{route: ok}, active: &activeJob, wantErr: model.ErrJobActive, wantStatus: model.JobRunning}, + {name: "agent offline", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentOffline}, wantStatus: model.JobFailed, + wantError: "agent offline", wantCreate: true}, + {name: "agent too old", serviceID: 1, hub: &fakeHub{routeErr: model.ErrAgentTooOld}, wantStatus: model.JobFailed, + wantError: "agent too old for updates", wantCreate: true}, + {name: "start fails", serviceID: 1, hub: &fakeHub{route: ok, startErr: model.ErrAgentOffline}, wantStatus: model.JobFailed, + wantError: "agent offline", wantCreate: true, + wantStart: &protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4.2"}}, + {name: "unknown service", serviceID: 5, hub: &fakeHub{route: ok}, wantErr: model.ErrNotFound}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + st := &fakeStore{ + services: map[int64]model.Service{1: composeSvc, 2: swarmSvc}, + jobs: map[int64]model.Job{99: activeJob}, + active: tt.active, + } + s, pub := newService(st, tt.hub) + + job, err := s.Trigger(context.Background(), tt.serviceID, "sub-1") + + if tt.wantErr != nil { + assert.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + if tt.wantErr == nil || tt.active != nil { + assert.Equal(t, tt.wantStatus, job.Status) + assert.Equal(t, tt.wantError, job.Error) + } + if tt.wantStart == nil { + assert.Empty(t, tt.hub.starts) + } else { + require.Len(t, tt.hub.starts, 1) + assert.Equal(t, *tt.wantStart, tt.hub.starts[0].start) + assert.Equal(t, ok, tt.hub.starts[0].route) + assert.Equal(t, job.ID, tt.hub.starts[0].id) + } + if tt.wantCreate { + require.Len(t, st.created, 1) + assert.Equal(t, "sub-1", st.created[0].RequestedBy) + assert.Equal(t, model.TriggerManual, st.created[0].Trigger) + assert.Equal(t, st.services[tt.serviceID].ActualRef, st.created[0].FromRef) + assert.Contains(t, pub.topics, events.Jobs) + assert.Contains(t, pub.topics, events.JobTopic(job.ID)) + } else { + assert.Empty(t, st.created) + assert.Empty(t, pub.topics) + } + }) + } +} + +func TestTrigger_NotUpdatableCreatesNoJob(t *testing.T) { + st := &fakeStore{ + services: map[int64]model.Service{1: {ID: 1, Kind: protocol.KindContainer, ActualRef: "redis:7"}}, + jobs: map[int64]model.Job{}, + } + h := &fakeHub{} + s, pub := newService(st, h) + + _, err := s.Trigger(context.Background(), 1, "sub-1") + + var nu *model.NotUpdatableError + require.True(t, errors.As(err, &nu)) + assert.Empty(t, st.created) + assert.Empty(t, h.starts) + assert.Empty(t, pub.topics) +}