feat(jobs): refuse poll and webhook jobs for drifted services, add restore (#10)

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")
+}

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)
+}

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
 	}

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 {