feat: master/agent phase 5 — drift and merge, backend (#10) (#30)

## Summary
- Drift detection: `model.Service.Drifted()` compares desired and running version by digest (tag when the digest is unknown). `services.awaiting_inventory` hides drift between a job result and the next inventory write.
- Poll and webhook jobs are refused for a drifted service; `POST /services/{id}/restore` (manual job to `desired_ref`) and `POST /services/{id}/accept` (running version becomes desired, guarded by `actual` = `Service.RunningRef`).
- Move suggestions (`store.MoveCandidates`, Compose only) and `POST /services/{id}/merge`: the old row keeps its id, mode, `desired_ref` and history and takes the new location; the duplicate is deleted.
- Backend only. The buttons and badges come with #28, on top of the Ledger UI (#18).

Closes #10

## Test plan
- [x] `gofmt -l .`, `go vet ./...`, `go test ./...`
- [x] `go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/`
- [x] `go test -tags integration ./internal/master/store/` against postgres:17-alpine
- [ ] On the server: migration 0005 applies; no drift shown after a normal update
- [ ] On the server: manual `docker compose up` with another tag gives drift; webhook for an auto service is refused
- [ ] On the server: restore and accept via POST; merge after moving a stack

Reviewed-on: #30

internal/e2e/agent_master_test.go

@@ -11,7 +11,6 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
-	"regexp"
 	"sort"
 	"strings"
 	"sync"
@@ -25,6 +24,7 @@ import (
 	"github.com/shcizo/package-updater/internal/agent/conn"
 	"github.com/shcizo/package-updater/internal/agent/updater"
 	"github.com/shcizo/package-updater/internal/agenttoken"
+	"github.com/shcizo/package-updater/internal/imageref"
 	"github.com/shcizo/package-updater/internal/master/events"
 	"github.com/shcizo/package-updater/internal/master/hub"
 	"github.com/shcizo/package-updater/internal/master/model"
@@ -32,9 +32,6 @@ import (
 	"github.com/shcizo/package-updater/internal/protocol"
 )
 
-// digestPattern is the digest format the real store records.
-var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
-
 type memStore struct {
 	mu        sync.Mutex
 	agents    map[string]model.Agent
@@ -245,7 +242,8 @@ 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
-		if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) {
+		svc.AwaitingInventory = true
+		if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && imageref.IsDigest(d) {
 			svc.LastTagDigest = d
 		}
 		// Same rule as the SQL: the job was created for the suggested ref after it was reported.

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/imageref/imageref.go

@@ -2,7 +2,10 @@
 // and digest.
 package imageref
 
-import "strings"
+import (
+	"regexp"
+	"strings"
+)
 
 // Ref is an image reference split into its parts. Tag and Digest are empty
 // when the reference has none.
@@ -41,3 +44,8 @@ func (r Ref) String() string {
 	}
 	return s
 }
+
+var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+
+// IsDigest reports whether d is a well-formed sha256 digest.
+func IsDigest(d string) bool { return digestPattern.MatchString(d) }

internal/imageref/imageref_test.go

@@ -1,6 +1,7 @@
 package imageref
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -27,3 +28,19 @@ func TestParse(t *testing.T) {
 		})
 	}
 }
+
+func TestIsDigest(t *testing.T) {
+	cases := []struct {
+		in   string
+		want bool
+	}{
+		{"sha256:" + strings.Repeat("a", 64), true},
+		{"sha256:" + strings.Repeat("a", 63), false},
+		{"sha256:" + strings.Repeat("A", 64), false},
+		{"sha512:" + strings.Repeat("a", 64), false},
+		{"", false},
+	}
+	for _, c := range cases {
+		assert.Equal(t, c.want, IsDigest(c.in), c.in)
+	}
+}

internal/master/jobs/drift_test.go

@@ -0,0 +1,237 @@
+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.ReasonRemoved},
+		{"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},
+		{"last tag digest holds back a newer one", func(s *model.Service) {
+			s.DesiredRef = driftRepo + ":1.4.2@" + driftD1
+			s.LastTagDigest = driftD3
+		}, ""},
+		{"last tag digest equals the desired digest", func(s *model.Service) { s.LastTagDigest = driftD1 }, driftD2},
+		{"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,81 @@ 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.RemovedAt != nil {
+		return &model.NotUpdatableError{Reason: model.ReasonRemoved}
+	}
+	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). It is not
+// written when last_tag_digest holds back a newer digest, that is when it is
+// set and differs from the digest the desired ref names.
+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 && holdsNothingNewer(svc) {
+		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
+}
+
+// holdsNothingNewer reports whether last_tag_digest may be overwritten by a
+// restore: it is empty or equals the digest the desired ref names. Anything
+// else is a digest the poller must keep holding back.
+func holdsNothingNewer(svc model.Service) bool {
+	return svc.LastTagDigest == "" || svc.LastTagDigest == imageref.Parse(svc.DesiredRef).Digest
+}
+
+// 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 +237,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 {

internal/master/model/job.go

@@ -44,6 +44,27 @@ const (
 	ReasonNoTag     = "follows no tag"
 	ReasonRemoved   = "service no longer exists"
 	ReasonBuilt     = "built from source (build:)"
+
+	ReasonDrifted    = "drifted: restore or accept the running version first"
+	ReasonNotDrifted = "runs its desired version"
+)
+
+// MergeRefusedError says why two services cannot be merged. Reason is shown
+// in the UI as is.
+type MergeRefusedError struct {
+	Reason string
+}
+
+func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason }
+
+// Reasons two services cannot be merged. ReasonSelf is reused for the
+// updater itself.
+const (
+	ReasonMergeSame      = "a service cannot be merged into itself"
+	ReasonMergeLiveOld   = "the old service still exists"
+	ReasonMergeGoneNew   = "the new service no longer exists"
+	ReasonMergeKind      = "only services of the same kind can be merged"
+	ReasonMergeContainer = "plain containers cannot be merged"
 )
 
 // JobStatus is the state of an update job.

internal/master/model/model.go

@@ -15,6 +15,8 @@ import (
 // Sentinel errors returned by storage.
 var (
 	ErrNotFound = errors.New("not found")
+	// ErrConflict means the row already exists, or that it changed meanwhile
+	// (AcceptRunning, MergeService).
 	ErrConflict = errors.New("already exists")
 	ErrRevoked  = errors.New("agent revoked")
 )
@@ -68,9 +70,13 @@ type Service struct {
 	// not start another job for the same digest, so a rollback sticks until the
 	// registry shows a different one.
 	LastTagDigest string
-	FirstSeenAt   time.Time
-	UpdatedAt     time.Time
-	RemovedAt     *time.Time
+	// AwaitingInventory is true from a successful job until the next inventory
+	// write for the row. Until then DesiredRef is newer than ActualRef without
+	// the service having drifted.
+	AwaitingInventory bool
+	FirstSeenAt       time.Time
+	UpdatedAt         time.Time
+	RemovedAt         *time.Time
 }
 
 // Update modes of a service.
@@ -113,6 +119,68 @@ func (v ImageVersion) Ref() string {
 	return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String()
 }
 
+// Drifted reports whether the service runs something other than the version
+// the last job deployed. A service the updater never touched, a removed one
+// and one whose inventory has not caught up with the last job do not drift.
+func (s Service) Drifted() bool {
+	return s.DesiredRef != "" && s.RemovedAt == nil && !s.AwaitingInventory &&
+		!RunsRef(s.DesiredRef, s.ActualRef, s.ActualDigest)
+}
+
+// RunsRef reports whether a service reporting actualRef and actualDigest runs
+// the desired ref. The digest decides when both sides know one, so
+// "repo:tag@D" is run by "repo:tag" reporting digest D, and by "repo@D".
+// Otherwise the repo (compared exactly, without registry normalisation) and
+// the followed tag must be equal.
+func RunsRef(desired, actualRef, actualDigest string) bool {
+	d, a := imageref.Parse(desired), imageref.Parse(actualRef)
+	if d.Repo != a.Repo {
+		return false
+	}
+	ad := a.Digest
+	if ad == "" {
+		ad = actualDigest
+	}
+	if d.Digest != "" && ad != "" {
+		return d.Digest == ad
+	}
+	return followedTag(d) == followedTag(a)
+}
+
+// followedTag is the tag a ref follows: a bare image follows "latest" and a
+// ref with a digest but no tag follows nothing.
+func followedTag(r imageref.Ref) string {
+	if r.Tag == "" && r.Digest == "" {
+		return "latest"
+	}
+	return r.Tag
+}
+
+// RunningRef is the ref to record as the desired one when the running version
+// is accepted: actualRef itself if it carries a digest, "repo:tag@digest" if
+// only actualDigest is known (a bare image is "latest"), else actualRef.
+func RunningRef(actualRef, actualDigest string) string {
+	r := imageref.Parse(actualRef)
+	if r.Digest != "" || !imageref.IsDigest(actualDigest) {
+		return actualRef
+	}
+	r.Tag, r.Digest = followedTag(r), actualDigest
+	return r.String()
+}
+
+// RunningRef returns the ref accepting the running version records as the
+// desired one; see the function RunningRef. Templates post it as the guard of
+// the accept form.
+func (s Service) RunningRef() string { return RunningRef(s.ActualRef, s.ActualDigest) }
+
+// Move says that the removed service FromID has probably moved to the live
+// service ToID on another agent. The names are the agents' own; a revoked
+// agent keeps its name.
+type Move struct {
+	FromID, ToID       int64
+	FromAgent, ToAgent string
+}
+
 // DisplayName is the human-facing name of the service.
 func (s Service) DisplayName() string {
 	switch s.Kind {

internal/master/model/model_test.go

@@ -2,6 +2,7 @@ package model
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -58,6 +59,84 @@ func TestImageVersionRef(t *testing.T) {
 	}
 }
 
+const (
+	dA = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+	dB = "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+)
+
+func TestRunsRef(t *testing.T) {
+	cases := []struct {
+		name                          string
+		desired, actualRef, actualDig string
+		want                          bool
+	}{
+		{"same string", "repo:v1@" + dA, "repo:v1@" + dA, "", true},
+		{"tag and digest, actual reports it separately", "repo:tag@" + dA, "repo:tag", dA, true},
+		{"bare image reports latest digest", "postgres:latest@" + dA, "postgres", dA, true},
+		{"tagless actual with the same digest", "repo:tag@" + dA, "repo@" + dA, "", true},
+		{"same digest under another tag", "repo:v1@" + dA, "repo:v2@" + dA, "", true},
+		{"same digest, actual tag differs, digest reported", "repo:v1@" + dA, "repo:v2", dA, true},
+		{"same tag, actual digest unknown", "repo:tag@" + dA, "repo:tag", "", true},
+		{"desired without digest, same tag", "repo:tag", "repo:tag@" + dA, "", true},
+		{"bare desired, latest actual", "repo", "repo:latest", "", true},
+		{"registry with port", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dA, true},
+		{"registry with port, other digest", "localhost:5000/foo:v1@" + dA, "localhost:5000/foo:v1", dB, false},
+		{"same tag, other digest", "repo:tag@" + dA, "repo:tag", dB, false},
+		{"other digest in the ref", "repo:tag@" + dA, "repo:tag@" + dB, "", false},
+		{"other tag, digest unknown", "repo:v1@" + dA, "repo:v2", "", false},
+		{"tagless desired, tagged actual, digest unknown", "repo@" + dA, "repo:v1", "", false},
+		{"other repo", "repo:tag@" + dA, "other:tag", dA, false},
+		{"no docker.io normalisation", "nginx:1@" + dA, "docker.io/library/nginx:1", dA, false},
+		{"case-sensitive repo", "Repo:1@" + dA, "repo:1", dA, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, RunsRef(c.desired, c.actualRef, c.actualDig))
+		})
+	}
+}
+
+func TestDrifted(t *testing.T) {
+	now := time.Now()
+	drifted := Service{DesiredRef: "repo:tag@" + dA, ActualRef: "repo:tag", ActualDigest: dB}
+	cases := []struct {
+		name   string
+		mutate func(*Service)
+		want   bool
+	}{
+		{"runs another digest", func(*Service) {}, true},
+		{"never updated", func(s *Service) { s.DesiredRef = "" }, false},
+		{"removed", func(s *Service) { s.RemovedAt = &now }, false},
+		{"awaiting inventory", func(s *Service) { s.AwaitingInventory = true }, false},
+		{"runs the desired digest", func(s *Service) { s.ActualDigest = dA }, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			svc := drifted
+			c.mutate(&svc)
+			assert.Equal(t, c.want, svc.Drifted())
+		})
+	}
+}
+
+func TestRunningRef(t *testing.T) {
+	cases := []struct {
+		name, actualRef, actualDigest, want string
+	}{
+		{"ref with digest is kept", "repo:v1@" + dB, dA, "repo:v1@" + dB},
+		{"tag plus reported digest", "repo:v1", dA, "repo:v1@" + dA},
+		{"bare image becomes latest", "postgres", dA, "postgres:latest@" + dA},
+		{"registry with port", "localhost:5000/foo:v1", dA, "localhost:5000/foo:v1@" + dA},
+		{"no digest known", "repo:v1", "", "repo:v1"},
+		{"malformed digest", "repo:v1", "sha256:abc", "repo:v1"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			assert.Equal(t, c.want, RunningRef(c.actualRef, c.actualDigest))
+		})
+	}
+}
+
 func TestJob_Active(t *testing.T) {
 	cases := []struct {
 		status JobStatus

internal/master/store/drift_merge_integration_test.go

@@ -0,0 +1,550 @@
+//go:build integration
+
+package store
+
+import (
+	"context"
+	"errors"
+	"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/protocol"
+)
+
+var (
+	digNew     = "sha256:" + strings.Repeat("e", 64)
+	digOld     = "sha256:" + strings.Repeat("f", 64)
+	sluskDrift = "gitea.shcizo.se/shcizo/slusk"
+)
+
+// serviceID returns the ID of the service the agent reports under svc's key.
+func serviceID(t *testing.T, s *Store, agentID int64, svc protocol.Service) int64 {
+	t.Helper()
+	var id int64
+	require.NoError(t, s.pool.QueryRow(context.Background(),
+		`SELECT id FROM services WHERE location_key = $1`, model.LocationKey(agentID, "", svc.Key())).Scan(&id))
+	return id
+}
+
+// finishJob runs a job for the service to success with the deployed ref.
+func finishJob(t *testing.T, s *Store, agentID, svcID int64, toRef, deployed string) {
+	t.Helper()
+	ctx := context.Background()
+	nj := newJob(svcID, agentID)
+	nj.ToRef = toRef
+	j, err := s.CreateJob(ctx, nj)
+	require.NoError(t, err)
+	ok, err := s.FinishJob(ctx, agentID, j.ID, model.JobFinish{Status: model.JobSucceeded, Ref: deployed, StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	require.True(t, ok)
+}
+
+func mustExec(t *testing.T, s *Store, sql string, args ...any) {
+	t.Helper()
+	_, err := s.pool.Exec(context.Background(), sql, args...)
+	require.NoError(t, err)
+}
+
+func TestFinishJob_SetsAwaitingInventoryUntilTheNextUpsert(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agent, svc := jobFixture(t, s, "a")
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.False(t, got.AwaitingInventory)
+
+	finishJob(t, s, agent, svc, "r:1", "r:1@"+digNew)
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.True(t, got.AwaitingInventory, "set by the successful job")
+	assert.False(t, got.Drifted(), "desired is ahead of actual only until the inventory arrives")
+
+	require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Minute)))
+	got, err = s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.False(t, got.AwaitingInventory, "cleared by the agent's upsert")
+	assert.True(t, got.Drifted(), "the agent still reports the old version")
+}
+
+func TestFinishJob_FailedJobDoesNotSetAwaitingInventory(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	agent, svc := jobFixture(t, s, "a")
+	j, err := s.CreateJob(ctx, newJob(svc, agent))
+	require.NoError(t, err)
+
+	ok, err := s.FinishJob(ctx, agent, j.ID, model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0})
+	require.NoError(t, err)
+	require.True(t, ok)
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.False(t, got.AwaitingInventory)
+}
+
+func TestCreateJob_UnknownServiceIsNotFound(t *testing.T) {
+	s := openTestStore(t)
+	agent := newAgent(t, s, "a")
+
+	_, err := s.CreateJob(context.Background(), newJob(9999, agent))
+
+	assert.ErrorIs(t, err, model.ErrNotFound)
+}
+
+// driftedFixture is a service updated to digNew whose agent now reports digD1.
+func driftedFixture(t *testing.T, s *Store) (agent, svc int64) {
+	t.Helper()
+	agent, svc = jobFixture(t, s, "a")
+	finishJob(t, s, agent, svc, "r:1", sluskDrift+":1.4.2@"+digNew)
+	require.NoError(t, s.UpsertService(context.Background(), agent, "", slusk, t0.Add(time.Minute)))
+	return agent, svc
+}
+
+func TestAcceptRunning_WritesTheDesiredRef(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	_, svc := driftedFixture(t, s)
+	require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:ltd"))
+	require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskDrift + ":1.5.0", Source: model.SourcePoll, At: t0}))
+	require.NoError(t, s.SetUpdateMode(ctx, svc, model.ModeAuto))
+	before, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	require.True(t, before.Drifted())
+
+	running := model.RunningRef(before.ActualRef, before.ActualDigest)
+	require.NoError(t, s.AcceptRunning(ctx, svc, before.ActualRef, before.ActualDigest, running))
+
+	got, err := s.GetService(ctx, svc)
+	require.NoError(t, err)
+	assert.Equal(t, sluskDrift+":1.4.2@"+digD1, got.DesiredRef)
+	assert.False(t, got.Drifted())
+	assert.False(t, got.AwaitingInventory)
+	assert.Equal(t, "sha256:ltd", got.LastTagDigest, "last_tag_digest untouched")
+	assert.Equal(t, sluskDrift+":1.5.0", got.AvailableRef, "available suggestion untouched")
+	assert.Equal(t, model.ModeAuto, got.UpdateMode)
+}
+
+func TestAcceptRunning_Refusals(t *testing.T) {
+	tests := []struct {
+		name      string
+		prepare   func(t *testing.T, s *Store, agent, svc int64) (id int64, actualRef, actualDigest string)
+		wantError error
+	}{
+		{"running version changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return svc, sluskDrift + ":1.4.2", digOld
+		}, model.ErrConflict},
+		{"actual ref changed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return svc, sluskDrift + ":1.3.0", digD1
+		}, model.ErrConflict},
+		{"never updated", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, svc)
+			return svc, sluskDrift + ":1.4.2", digD1
+		}, model.ErrConflict},
+		{"removed", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			require.NoError(t, s.RemoveService(context.Background(), agent, "", slusk.Key(), t0.Add(time.Hour)))
+			return svc, sluskDrift + ":1.4.2", digD1
+		}, model.ErrNotFound},
+		{"unknown", func(t *testing.T, s *Store, agent, svc int64) (int64, string, string) {
+			return 9999, sluskDrift + ":1.4.2", digD1
+		}, model.ErrNotFound},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			agent, svc := driftedFixture(t, s)
+			id, ref, digest := tt.prepare(t, s, agent, svc)
+
+			err := s.AcceptRunning(context.Background(), id, ref, digest, "x:1@"+digOld)
+
+			assert.ErrorIs(t, err, tt.wantError)
+			got, gerr := s.GetService(context.Background(), svc)
+			require.NoError(t, gerr)
+			assert.NotEqual(t, "x:1@"+digOld, got.DesiredRef, "nothing written")
+		})
+	}
+}
+
+// moveFixture has agent "a" with slusk removed after it was updated (history,
+// auto mode) and agent "b" reporting a new slusk row.
+type moveFixture struct {
+	a, b     int64
+	from, to int64
+}
+
+func newMoveFixture(t *testing.T, s *Store) moveFixture {
+	t.Helper()
+	ctx := context.Background()
+	f := moveFixture{a: newAgent(t, s, "a"), b: newAgent(t, s, "b")}
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0))
+	f.from = serviceID(t, s, f.a, slusk)
+	finishJob(t, s, f.a, f.from, "r:1", sluskDrift+":1.4.2@"+digNew)
+	require.NoError(t, s.SetUpdateMode(ctx, f.from, model.ModeAuto))
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(time.Hour)))
+	require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(2*time.Hour)))
+	f.to = serviceID(t, s, f.b, slusk)
+	return f
+}
+
+func TestMoveCandidates(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(t *testing.T, s *Store, f moveFixture)
+		want   bool
+		from   string // agent name of the removed row
+	}{
+		{"one removed and one live", func(*testing.T, *Store, moveFixture) {}, true, "a"},
+		{"removed row's agent was revoked", func(t *testing.T, s *Store, f moveFixture) {
+			_, err := s.RevokeAgent(context.Background(), f.a, t0.Add(3*time.Hour))
+			require.NoError(t, err)
+		}, true, "a"},
+		{"two live targets", func(t *testing.T, s *Store, f moveFixture) {
+			c := newAgent(t, s, "c")
+			require.NoError(t, s.ApplySnapshot(context.Background(), c, worker, []protocol.Service{slusk}, t0.Add(3*time.Hour)))
+		}, false, ""},
+		{"two removed sources", func(t *testing.T, s *Store, f moveFixture) {
+			ctx := context.Background()
+			d := newAgent(t, s, "d")
+			require.NoError(t, s.ApplySnapshot(ctx, d, worker, []protocol.Service{slusk}, t0))
+			require.NoError(t, s.ApplySnapshot(ctx, d, worker, nil, t0.Add(time.Hour)))
+		}, false, ""},
+		{"target has history", func(t *testing.T, s *Store, f moveFixture) {
+			finishJob(t, s, f.b, f.to, "r:1", sluskDrift+":1.4.2@"+digNew)
+		}, false, ""},
+		{"target has a desired ref", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET desired_ref = 'r:1@x' WHERE id = $1`, f.to)
+		}, false, ""},
+		{"source has no history", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from)
+			mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from)
+		}, false, ""},
+		{"source only has auto mode", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `DELETE FROM update_jobs WHERE service_id = $1`, f.from)
+			mustExec(t, s, `UPDATE services SET desired_ref = '' WHERE id = $1`, f.from)
+		}, true, "a"},
+		{"source only has jobs", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET desired_ref = '', update_mode = 'manual' WHERE id = $1`, f.from)
+		}, true, "a"},
+		{"target is older than the source", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET first_seen_at = $2 WHERE id = $1`, f.to, t0.Add(-time.Hour))
+		}, false, ""},
+		{"plain containers", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to)
+		}, false, ""},
+		{"swarm services", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id IN ($1, $2)`, f.from, f.to)
+		}, false, ""},
+		{"the updater itself (target)", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to)
+		}, false, ""},
+		{"the updater itself (source)", func(t *testing.T, s *Store, f moveFixture) {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from)
+		}, false, ""},
+		{"live row also removed", func(t *testing.T, s *Store, f moveFixture) {
+			require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour)))
+		}, false, ""},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			f := newMoveFixture(t, s)
+			tt.mutate(t, s, f)
+
+			got, err := s.MoveCandidates(context.Background())
+
+			require.NoError(t, err)
+			if !tt.want {
+				assert.Empty(t, got)
+				return
+			}
+			assert.Equal(t, []model.Move{{FromID: f.from, ToID: f.to, FromAgent: tt.from, ToAgent: "b"}}, got)
+		})
+	}
+}
+
+func TestMergeService_MovesHistoryModeAndDesiredToTheNewLocation(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	finishJob(t, s, f.b, f.to, "r:2", sluskDrift+":2.0.0@"+digOld)
+	require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto"))
+	require.NoError(t, s.SetAvailable(ctx, f.to, model.Available{Ref: sluskDrift + ":3", Digest: digNew, Source: model.SourcePoll, At: t0}))
+	require.NoError(t, s.SetPollError(ctx, []int64{f.to}, "registry down"))
+	// A job on the old row is finished, and awaiting_inventory is set on both.
+	before, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+
+	require.NoError(t, s.MergeService(ctx, f.from, f.to))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, f.from, got.ID, "the old id survives")
+	assert.Nil(t, got.RemovedAt)
+	assert.False(t, got.AwaitingInventory)
+	assert.Equal(t, "b", got.AgentName)
+	assert.Equal(t, model.LocationKey(f.b, "", slusk.Key()), got.LocationKey)
+	assert.Equal(t, before.FirstSeenAt, got.FirstSeenAt, "first_seen_at kept")
+	assert.Equal(t, model.ModeAuto, got.UpdateMode, "mode kept")
+	assert.Equal(t, before.DesiredRef, got.DesiredRef, "desired ref kept")
+	assert.Equal(t, slusk.Image, got.ActualRef, "inventory from the duplicate")
+	assert.Equal(t, digD1, got.ActualDigest)
+	assert.Equal(t, "running", got.Status)
+	assert.Equal(t, sluskDrift+":3", got.AvailableRef)
+	assert.Equal(t, digNew, got.AvailableDigest)
+	assert.Equal(t, "registry down", got.LastPollError)
+	assert.Equal(t, before.LastTagDigest, got.LastTagDigest, "an existing last_tag_digest wins")
+
+	_, err = s.GetService(ctx, f.to)
+	assert.ErrorIs(t, err, model.ErrNotFound, "the duplicate is deleted")
+	jobs, err := s.ListJobs(ctx, model.JobFilter{ServiceID: f.from})
+	require.NoError(t, err)
+	assert.Len(t, jobs, 2, "the duplicate's job moved to the survivor")
+	live, err := s.ListServices(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, []string{"slusk/web"}, names(live))
+}
+
+func TestMergeService_TakesLastTagDigestOfTheDuplicateWhenTheOldRowHasNone(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	mustExec(t, s, `UPDATE services SET last_tag_digest = '' WHERE id = $1`, f.from)
+	require.NoError(t, s.SetLastTagDigest(ctx, f.to, "sha256:fromto"))
+
+	require.NoError(t, s.MergeService(ctx, f.from, f.to))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "sha256:fromto", got.LastTagDigest)
+}
+
+func TestMergeService_ManualMergeOfAnyRemovedServiceOfTheSameKind(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.b, "", other, t0.Add(3*time.Hour)))
+	target := serviceID(t, s, f.b, other)
+
+	require.NoError(t, s.MergeService(ctx, f.from, target))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "renamed/app", got.DisplayName(), "names may differ")
+}
+
+func TestMergeService_SameAgentIsAllowed(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0.Add(3*time.Hour)))
+
+	require.NoError(t, s.MergeService(ctx, f.from, serviceID(t, s, f.a, other)))
+
+	got, err := s.GetService(ctx, f.from)
+	require.NoError(t, err)
+	assert.Equal(t, "a", got.AgentName)
+}
+
+func TestMergeService_Swarm(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	m1 := newAgent(t, s, "m1")
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{gitea}, t0))
+	oldID := serviceIDByName(t, s, "gitea_gitea")
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, nil, t0.Add(time.Hour)))
+	renamed := gitea
+	renamed.SwarmService = "gitea2_gitea"
+	require.NoError(t, s.ApplySnapshot(ctx, m1, manager, []protocol.Service{renamed}, t0.Add(2*time.Hour)))
+	newID := serviceIDByName(t, s, "gitea2_gitea")
+
+	require.NoError(t, s.MergeService(ctx, oldID, newID))
+
+	got, err := s.GetService(ctx, oldID)
+	require.NoError(t, err)
+	assert.Equal(t, "gitea2_gitea", got.SwarmService)
+	assert.Nil(t, got.RemovedAt)
+}
+
+func serviceIDByName(t *testing.T, s *Store, swarmService string) int64 {
+	t.Helper()
+	var id int64
+	require.NoError(t, s.pool.QueryRow(context.Background(),
+		`SELECT id FROM services WHERE swarm_service = $1`, swarmService).Scan(&id))
+	return id
+}
+
+func TestMergeService_Refusals(t *testing.T) {
+	refused := func(reason string) func(*testing.T, error) {
+		return func(t *testing.T, err error) {
+			var mr *model.MergeRefusedError
+			require.True(t, errors.As(err, &mr), "got %v", err)
+			assert.Equal(t, reason, mr.Reason)
+		}
+	}
+	is := func(want error) func(*testing.T, error) {
+		return func(t *testing.T, err error) { assert.ErrorIs(t, err, want) }
+	}
+	tests := []struct {
+		name   string
+		run    func(t *testing.T, s *Store, f moveFixture) error
+		assert func(*testing.T, error)
+	}{
+		{"into itself", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), f.to, f.to)
+		}, refused(model.ReasonMergeSame)},
+		{"the old service still exists", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET removed_at = NULL WHERE id = $1`, f.from)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeLiveOld)},
+		{"the new service was removed", func(t *testing.T, s *Store, f moveFixture) error {
+			require.NoError(t, s.RemoveService(context.Background(), f.b, "", slusk.Key(), t0.Add(3*time.Hour)))
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeGoneNew)},
+		{"different kinds", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET kind = 'swarm' WHERE id = $1`, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeKind)},
+		{"plain containers", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET kind = 'container' WHERE id IN ($1, $2)`, f.from, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonMergeContainer)},
+		{"the updater itself (new)", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonSelf)},
+		{"the updater itself (old)", func(t *testing.T, s *Store, f moveFixture) error {
+			mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, refused(model.ReasonSelf)},
+		{"active job on the old service", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.CreateJob(context.Background(), newJob(f.from, f.a))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrJobActive)},
+		{"active job on the new service", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.CreateJob(context.Background(), newJob(f.to, f.b))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrJobActive)},
+		{"unknown old service", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), 9999, f.to)
+		}, is(model.ErrNotFound)},
+		{"unknown new service", func(t *testing.T, s *Store, f moveFixture) error {
+			return s.MergeService(context.Background(), f.from, 9999)
+		}, is(model.ErrNotFound)},
+		{"the new service's agent is revoked", func(t *testing.T, s *Store, f moveFixture) error {
+			_, err := s.RevokeAgent(context.Background(), f.b, t0.Add(3*time.Hour))
+			require.NoError(t, err)
+			return s.MergeService(context.Background(), f.from, f.to)
+		}, is(model.ErrRevoked)},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := openTestStore(t)
+			f := newMoveFixture(t, s)
+			var rowsBefore int
+			require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsBefore))
+
+			err := tt.run(t, s, f)
+
+			tt.assert(t, err)
+			var rowsAfter int
+			require.NoError(t, s.pool.QueryRow(context.Background(), `SELECT count(*) FROM services`).Scan(&rowsAfter))
+			assert.Equal(t, rowsBefore, rowsAfter, "nothing is deleted when the merge is refused")
+		})
+	}
+}
+
+func TestMergeService_InventoryAfterTheMerge(t *testing.T) {
+	setup := func(t *testing.T) (*Store, moveFixture) {
+		s := openTestStore(t)
+		f := newMoveFixture(t, s)
+		require.NoError(t, s.MergeService(context.Background(), f.from, f.to))
+		return s, f
+	}
+
+	t.Run("an upsert for the new location updates the survivor", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		moved := slusk
+		moved.Digest = digOld
+		require.NoError(t, s.UpsertService(ctx, f.b, "", moved, t0.Add(4*time.Hour)))
+
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		require.Len(t, live, 1)
+		assert.Equal(t, f.from, live[0].ID)
+		assert.Equal(t, digOld, live[0].ActualDigest)
+	})
+
+	t.Run("a snapshot keeps the survivor", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		require.NoError(t, s.ApplySnapshot(ctx, f.b, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour)))
+
+		got, err := s.GetService(ctx, f.from)
+		require.NoError(t, err)
+		assert.Nil(t, got.RemovedAt)
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		assert.Len(t, live, 1)
+	})
+
+	t.Run("the old agent returning with the stack creates a new row", func(t *testing.T) {
+		s, f := setup(t)
+		ctx := context.Background()
+		require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{slusk}, t0.Add(4*time.Hour)))
+
+		returned := serviceID(t, s, f.a, slusk)
+		assert.NotEqual(t, f.from, returned)
+		got, err := s.GetService(ctx, returned)
+		require.NoError(t, err)
+		assert.Equal(t, model.ModeManual, got.UpdateMode)
+		assert.Empty(t, got.DesiredRef)
+		live, err := s.ListServices(ctx)
+		require.NoError(t, err)
+		assert.Len(t, live, 2)
+	})
+}
+
+func TestMergeSources(t *testing.T) {
+	s := openTestStore(t)
+	ctx := context.Background()
+	f := newMoveFixture(t, s)
+	// A removed service with another name, removed later than the matching one.
+	other := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "renamed", ComposeService: "app", Image: "x/y:1", Status: "running"}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", other, t0))
+	require.NoError(t, s.RemoveService(ctx, f.a, "", other.Key(), t0.Add(5*time.Hour)))
+	otherID := serviceID(t, s, f.a, other)
+	// Other kinds, the updater itself and a live service never show up.
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, []protocol.Service{adhoc}, t0.Add(6*time.Hour)))
+	require.NoError(t, s.ApplySnapshot(ctx, f.a, worker, nil, t0.Add(7*time.Hour)))
+	selfSvc := protocol.Service{Kind: protocol.KindCompose, ComposeProject: "pu", ComposeService: "agent", Image: "pu:1", Status: "running", Self: true}
+	require.NoError(t, s.UpsertService(ctx, f.a, "", selfSvc, t0))
+	require.NoError(t, s.RemoveService(ctx, f.a, "", selfSvc.Key(), t0.Add(8*time.Hour)))
+
+	got, err := s.MergeSources(ctx, f.to)
+
+	require.NoError(t, err)
+	var ids []int64
+	for _, m := range got {
+		ids = append(ids, m.ID)
+	}
+	assert.Equal(t, []int64{f.from, otherID}, ids, "same name first, then the most recently removed")
+	assert.Equal(t, "a", got[0].AgentName)
+
+	_, err = s.MergeSources(ctx, 9999)
+	assert.ErrorIs(t, err, model.ErrNotFound)
+
+	containerID := serviceID(t, s, f.a, adhoc)
+	got, err = s.MergeSources(ctx, containerID)
+	require.NoError(t, err)
+	assert.Empty(t, got, "plain containers cannot be merged")
+}

internal/master/store/jobs.go

@@ -10,6 +10,7 @@ import (
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
 
+	"github.com/shcizo/package-updater/internal/imageref"
 	"github.com/shcizo/package-updater/internal/master/model"
 	"github.com/shcizo/package-updater/internal/protocol"
 )
@@ -90,6 +91,9 @@ func (s *Store) CreateJob(ctx context.Context, j model.NewJob) (model.Job, error
 			}
 			return active, model.ErrJobActive
 		}
+		if errors.As(err, &pgErr) && pgErr.Code == "23503" && pgErr.ConstraintName == "update_jobs_service_id_fkey" {
+			return model.Job{}, model.ErrNotFound
+		}
 		if err != nil {
 			return model.Job{}, fmt.Errorf("create job: %w", err)
 		}
@@ -232,11 +236,11 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
 		// A plain tag job, whoever started it, deployed the tag at the digest
 		// the agent pinned; remember it. Pinned and tagless picks do not.
 		tagDigest := ""
-		if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && digestPattern.MatchString(d) {
+		if _, d, _ := strings.Cut(f.Ref, "@"); requestedDigest == "" && imageref.IsDigest(d) {
 			tagDigest = d
 		}
 		if _, err := tx.Exec(ctx, `
-			UPDATE services SET desired_ref = $2,
+			UPDATE services SET desired_ref = $2, awaiting_inventory = true,
 				last_tag_digest  = CASE WHEN $6 <> '' THEN $6 ELSE last_tag_digest END,
 				available_ref    = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_ref END,
 				available_digest = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_digest END,

internal/master/store/merge.go

@@ -0,0 +1,249 @@
+package store
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/shcizo/package-updater/internal/master/model"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// moveCandidatesSQL pairs a removed Compose service with a live one of the
+// same project and service name on another agent. A pair counts only when it
+// is the single one for both rows, and the live row is new (newer than the
+// removed one, never updated, no jobs) while the removed one has something
+// worth moving (a desired ref, auto mode or jobs).
+const moveCandidatesSQL = `
+WITH pairs AS (
+	SELECT f.id AS from_id, t.id AS to_id
+	FROM services f
+	JOIN services t ON t.kind = 'compose' AND t.removed_at IS NULL AND NOT t.self
+	 AND t.compose_project = f.compose_project AND t.compose_service = f.compose_service
+	 AND t.agent_id IS DISTINCT FROM f.agent_id
+	WHERE f.kind = 'compose' AND f.removed_at IS NOT NULL AND NOT f.self
+), uniq AS (
+	SELECT from_id, to_id FROM pairs p
+	WHERE (SELECT count(*) FROM pairs q WHERE q.from_id = p.from_id) = 1
+	  AND (SELECT count(*) FROM pairs q WHERE q.to_id   = p.to_id)   = 1
+)
+SELECT u.from_id, u.to_id, COALESCE(fa.name, ''), COALESCE(ta.name, '')
+FROM uniq u
+JOIN services f ON f.id = u.from_id
+JOIN services t ON t.id = u.to_id
+LEFT JOIN agents fa ON fa.id = f.agent_id
+LEFT JOIN agents ta ON ta.id = t.agent_id
+WHERE t.first_seen_at > f.first_seen_at
+  AND t.desired_ref = ''
+  AND NOT EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = t.id)
+  AND (f.desired_ref <> '' OR f.update_mode = 'auto'
+       OR EXISTS (SELECT 1 FROM update_jobs j WHERE j.service_id = f.id))
+ORDER BY u.to_id`
+
+// MoveCandidates returns the removed Compose services that look like they
+// moved to another agent; see model.Move. It only suggests: nothing is merged.
+func (s *Store) MoveCandidates(ctx context.Context) ([]model.Move, error) {
+	rows, err := s.pool.Query(ctx, moveCandidatesSQL)
+	if err != nil {
+		return nil, fmt.Errorf("move candidates: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Move
+	for rows.Next() {
+		var m model.Move
+		if err := rows.Scan(&m.FromID, &m.ToID, &m.FromAgent, &m.ToAgent); err != nil {
+			return nil, fmt.Errorf("scan move: %w", err)
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// MergeSources returns up to 50 removed services that could be merged into
+// the service toID: the same kind (Compose or Swarm), not the updater itself,
+// those with the same name first, then the most recently removed. Returns
+// model.ErrNotFound if toID does not exist.
+func (s *Store) MergeSources(ctx context.Context, toID int64) ([]model.Service, error) {
+	to, err := s.GetService(ctx, toID)
+	if err != nil {
+		return nil, err
+	}
+	if to.Kind == protocol.KindContainer {
+		return nil, nil
+	}
+	rows, err := s.pool.Query(ctx, `
+		SELECT `+serviceColumns+`
+		FROM services s LEFT JOIN agents a ON a.id = s.agent_id
+		WHERE s.removed_at IS NOT NULL AND s.kind = $1 AND NOT s.self AND s.id <> $2
+		ORDER BY (s.compose_project, s.compose_service, s.swarm_service) = ($3, $4, $5) DESC,
+		         s.removed_at DESC, s.id DESC
+		LIMIT 50`,
+		string(to.Kind), toID, to.ComposeProject, to.ComposeService, to.SwarmService)
+	if err != nil {
+		return nil, fmt.Errorf("merge sources: %w", err)
+	}
+	defer rows.Close()
+	var out []model.Service
+	for rows.Next() {
+		m, err := scanService(rows)
+		if err != nil {
+			return nil, fmt.Errorf("scan service: %w", err)
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}
+
+// mergeRow is what MergeService needs to know about one of the two rows.
+type mergeRow struct {
+	kind    string
+	agentID *int64
+	removed bool
+	self    bool
+}
+
+// MergeService moves the live service toID onto the removed service fromID:
+// the old row keeps its id, first_seen_at, update mode and desired ref, takes
+// the location and inventory of the live row (and its available suggestion),
+// becomes live, and the live row is deleted. The jobs of the live row move to
+// the old one. This is the only place a services row is deleted.
+//
+// It returns *model.MergeRefusedError when the pair cannot be merged,
+// model.ErrJobActive when either row has a queued or running job,
+// model.ErrRevoked when the live row's agent is revoked, model.ErrNotFound
+// when a row does not exist, and model.ErrConflict when the live row moved to
+// another agent while the merge started.
+//
+// Locks are taken in the order inventory writes use: the agent row first,
+// then the service rows by id. The live row is deleted before the old one is
+// updated because location_key is unique.
+func (s *Store) MergeService(ctx context.Context, fromID, toID int64) error {
+	if fromID == toID {
+		return &model.MergeRefusedError{Reason: model.ReasonMergeSame}
+	}
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("begin merge: %w", err)
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	var toAgent *int64
+	err = tx.QueryRow(ctx, `SELECT agent_id FROM services WHERE id = $1`, toID).Scan(&toAgent)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return model.ErrNotFound
+	}
+	if err != nil {
+		return fmt.Errorf("merge: find new service: %w", err)
+	}
+	if toAgent != nil {
+		if err := lockActiveAgent(ctx, tx, *toAgent); err != nil {
+			return err
+		}
+	}
+
+	rows, err := tx.Query(ctx, `
+		SELECT id, kind, agent_id, removed_at IS NOT NULL, self FROM services
+		WHERE id IN ($1, $2) ORDER BY id FOR UPDATE`, fromID, toID)
+	if err != nil {
+		return fmt.Errorf("merge: lock services: %w", err)
+	}
+	locked := map[int64]mergeRow{}
+	for rows.Next() {
+		var id int64
+		var r mergeRow
+		if err := rows.Scan(&id, &r.kind, &r.agentID, &r.removed, &r.self); err != nil {
+			rows.Close()
+			return fmt.Errorf("merge: scan service: %w", err)
+		}
+		locked[id] = r
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("merge: lock services: %w", err)
+	}
+	from, okFrom := locked[fromID]
+	to, okTo := locked[toID]
+	if !okFrom || !okTo {
+		return model.ErrNotFound
+	}
+	if (to.agentID == nil) != (toAgent == nil) || (toAgent != nil && *to.agentID != *toAgent) {
+		return model.ErrConflict
+	}
+	if reason := mergeRefusal(from, to); reason != "" {
+		return &model.MergeRefusedError{Reason: reason}
+	}
+
+	var active bool
+	if err := tx.QueryRow(ctx, `
+		SELECT EXISTS (SELECT 1 FROM update_jobs
+		               WHERE service_id IN ($1, $2) AND status IN ('queued', 'running'))`,
+		fromID, toID).Scan(&active); err != nil {
+		return fmt.Errorf("merge: check active jobs: %w", err)
+	}
+	if active {
+		return model.ErrJobActive
+	}
+
+	if _, err := tx.Exec(ctx, `UPDATE update_jobs SET service_id = $1 WHERE service_id = $2`, fromID, toID); err != nil {
+		return fmt.Errorf("merge: move jobs: %w", err)
+	}
+	var t model.Service
+	var lastTagDigest string
+	err = tx.QueryRow(ctx, `
+		SELECT agent_id, location_key, compose_project, compose_service, working_dir, config_files, built,
+		       swarm_cluster_id, swarm_service, container_name, image_repo, actual_ref, actual_digest, status, self,
+		       updated_at, available_ref, available_digest, available_source, available_at, last_poll_error,
+		       last_tag_digest
+		FROM services WHERE id = $1`, toID).Scan(
+		&t.AgentID, &t.LocationKey, &t.ComposeProject, &t.ComposeService, &t.WorkingDir, &t.ConfigFiles, &t.Built,
+		&t.SwarmClusterID, &t.SwarmService, &t.ContainerName, &t.ImageRepo, &t.ActualRef, &t.ActualDigest, &t.Status, &t.Self,
+		&t.UpdatedAt, &t.AvailableRef, &t.AvailableDigest, &t.AvailableSource, &t.AvailableAt, &t.LastPollError,
+		&lastTagDigest)
+	if err != nil {
+		return fmt.Errorf("merge: read new service: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `DELETE FROM services WHERE id = $1`, toID); err != nil {
+		return fmt.Errorf("merge: delete new service: %w", err)
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE services SET
+			agent_id = $2, location_key = $3, compose_project = $4, compose_service = $5, working_dir = $6,
+			config_files = $7, built = $8, swarm_cluster_id = $9, swarm_service = $10, container_name = $11,
+			image_repo = $12, actual_ref = $13, actual_digest = $14, status = $15, self = $16, updated_at = $17,
+			available_ref = $18, available_digest = $19, available_source = $20, available_at = $21,
+			last_poll_error = $22,
+			last_tag_digest = CASE WHEN last_tag_digest <> '' THEN last_tag_digest ELSE $23 END,
+			removed_at = NULL, awaiting_inventory = false
+		WHERE id = $1`,
+		fromID, t.AgentID, t.LocationKey, t.ComposeProject, t.ComposeService, t.WorkingDir,
+		t.ConfigFiles, t.Built, t.SwarmClusterID, t.SwarmService, t.ContainerName,
+		t.ImageRepo, t.ActualRef, t.ActualDigest, t.Status, t.Self, t.UpdatedAt,
+		t.AvailableRef, t.AvailableDigest, t.AvailableSource, t.AvailableAt,
+		t.LastPollError, lastTagDigest); err != nil {
+		return fmt.Errorf("merge: update old service: %w", err)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return fmt.Errorf("commit merge: %w", err)
+	}
+	return nil
+}
+
+// mergeRefusal returns why from (the old row) cannot be merged with to (the
+// live one), or "".
+func mergeRefusal(from, to mergeRow) string {
+	switch {
+	case !from.removed:
+		return model.ReasonMergeLiveOld
+	case to.removed:
+		return model.ReasonMergeGoneNew
+	case from.kind != to.kind:
+		return model.ReasonMergeKind
+	case from.kind == string(protocol.KindContainer):
+		return model.ReasonMergeContainer
+	case from.self || to.self:
+		return model.ReasonSelf
+	}
+	return ""
+}

internal/master/store/migrations/0005_drift_merge.sql

@@ -0,0 +1,7 @@
+-- True from a successful job until the next inventory write for the row. The
+-- agent's upsert arrives a debounce interval after job.result, and until then
+-- desired_ref is newer than actual_ref without the service having drifted.
+ALTER TABLE services ADD COLUMN awaiting_inventory boolean NOT NULL DEFAULT false;
+
+-- Move suggestions look up Compose rows by project and service across agents.
+CREATE INDEX services_by_compose_name ON services (compose_project, compose_service) WHERE kind = 'compose';

internal/master/store/services.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"regexp"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -55,10 +54,8 @@ ON CONFLICT (location_key) DO UPDATE SET
 	status        = EXCLUDED.status,
 	self          = EXCLUDED.self,
 	updated_at    = EXCLUDED.updated_at,
-	removed_at    = NULL`
-
-// digestPattern is the form of a digest worth recording as a version.
-var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
+	removed_at    = NULL,
+	awaiting_inventory = false`
 
 func upsertService(ctx context.Context, db execer, agentID int64, clusterID string, svc protocol.Service, at time.Time) error {
 	image := imageref.Parse(svc.Image)
@@ -75,7 +72,7 @@ func upsertService(ctx context.Context, db execer, agentID int64, clusterID stri
 	if err != nil {
 		return fmt.Errorf("upsert service %s: %w", svc.Key(), err)
 	}
-	if digestPattern.MatchString(svc.Digest) {
+	if imageref.IsDigest(svc.Digest) {
 		// A bare image runs "latest", as jobs.FollowedRef sees it, so it
 		// shares its row with the polled one. repo@digest has no tag.
 		if image.Tag == "" && image.Digest == "" {
@@ -178,7 +175,7 @@ const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.nam
 	s.compose_project, s.compose_service, s.working_dir, s.config_files, s.built, s.swarm_cluster_id,
 	s.swarm_service, s.container_name, s.image_repo, s.desired_ref, s.actual_ref, s.actual_digest,
 	s.status, s.self, s.update_mode, s.available_ref, s.available_digest, s.available_source, s.available_at,
-	s.last_poll_error, s.last_tag_digest, s.first_seen_at, s.updated_at, s.removed_at`
+	s.last_poll_error, s.last_tag_digest, s.awaiting_inventory, s.first_seen_at, s.updated_at, s.removed_at`
 
 func scanService(row pgx.Row) (model.Service, error) {
 	var m model.Service
@@ -187,7 +184,7 @@ func scanService(row pgx.Row) (model.Service, error) {
 		&m.ComposeProject, &m.ComposeService, &m.WorkingDir, &m.ConfigFiles, &m.Built, &m.SwarmClusterID,
 		&m.SwarmService, &m.ContainerName, &m.ImageRepo, &m.DesiredRef, &m.ActualRef, &m.ActualDigest,
 		&m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableDigest, &m.AvailableSource, &m.AvailableAt,
-		&m.LastPollError, &m.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
+		&m.LastPollError, &m.LastTagDigest, &m.AwaitingInventory, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
 	m.Kind = protocol.Kind(kind)
 	return m, err
 }
@@ -262,8 +259,8 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
 }
 
 // SetLastTagDigest sets last_tag_digest; see model.Service.LastTagDigest. The
-// poller calls it, and FinishJob sets the same column after a tag job.
-// Inventory writes never change it.
+// poller and jobs.RestoreDesired call it, and FinishJob sets the same column
+// after a tag job. Inventory writes never change it.
 func (s *Store) SetLastTagDigest(ctx context.Context, id int64, digest string) error {
 	if _, err := s.pool.Exec(ctx,
 		`UPDATE services SET last_tag_digest = $2 WHERE id = $1`, id, digest); err != nil {
@@ -331,6 +328,34 @@ func (s *Store) SetUpdateMode(ctx context.Context, id int64, mode string) error
 	return nil
 }
 
+// AcceptRunning makes the running version the desired one: desired_ref =
+// desired, provided the service is live, has been updated before and still
+// reports actualRef and actualDigest (what the caller saw when it decided).
+// It starts no job and leaves last_tag_digest, the available suggestion and
+// the update mode alone. Returns model.ErrNotFound for an unknown or removed
+// service and model.ErrConflict when the service was never updated or the
+// running version changed meanwhile.
+func (s *Store) AcceptRunning(ctx context.Context, id int64, actualRef, actualDigest, desired string) error {
+	tag, err := s.pool.Exec(ctx, `
+		UPDATE services SET desired_ref = $4, awaiting_inventory = false
+		WHERE id = $1 AND removed_at IS NULL AND desired_ref <> ''
+		  AND actual_ref = $2 AND actual_digest = $3`, id, actualRef, actualDigest, desired)
+	if err != nil {
+		return fmt.Errorf("accept running: %w", err)
+	}
+	if tag.RowsAffected() == 1 {
+		return nil
+	}
+	svc, err := s.GetService(ctx, id)
+	if err != nil {
+		return err
+	}
+	if svc.RemovedAt != nil {
+		return model.ErrNotFound
+	}
+	return model.ErrConflict
+}
+
 // GetService returns one service, including a removed one (RemovedAt set).
 // Returns model.ErrNotFound if the ID does not exist.
 func (s *Store) GetService(ctx context.Context, id int64) (model.Service, error) {

internal/master/web/drift_test.go

@@ -0,0 +1,297 @@
+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@" + driftActual}})
+
+	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@" + driftActual}, "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
+		removed    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: "same ref, other digest", path: "/services/1/accept",
+			form:     url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired}},
+			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@" + driftActual}},
+			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@" + driftActual}},
+			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", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			removed: true, wantCode: http.StatusNotFound},
+		{name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
+			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)
+			}
+			if tt.removed {
+				removedAt := time.Now()
+				store.services[0].RemovedAt = &removedAt
+			}
+			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)
+		})
+	}
+}

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 {
@@ -296,18 +349,130 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
 	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):
-		data, derr := s.serviceDetail(r, id)
-		if derr != nil {
-			s.failLookup(w, derr)
-			return
-		}
-		data.Error = "Cannot update: " + nu.Reason + "."
-		s.render(w, http.StatusConflict, "service.html", "layout", data)
+		s.renderServiceError(w, r, id, http.StatusConflict, "Cannot update: "+nu.Reason+".")
 	default:
 		s.failLookup(w, err)
 	}
 }
 
+// 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 running ref the user saw (Service.RunningRef);
+// 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 && svc.RemovedAt != nil {
+		err = model.ErrNotFound
+	}
+	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.RunningRef() {
+		s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
+		return
+	}
+	desired := svc.RunningRef()
+	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_id", from, "to_id", 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) {
@@ -331,13 +496,7 @@ func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
 		}
 		var nu *model.NotUpdatableError
 		if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) {
-			data, derr := s.serviceDetail(r, id)
-			if derr != nil {
-				s.failLookup(w, derr)
-				return
-			}
-			data.Error = "Cannot enable auto: " + nu.Reason + "."
-			s.render(w, http.StatusConflict, "service.html", "layout", data)
+			s.renderServiceError(w, r, id, http.StatusConflict, "Cannot enable auto: "+nu.Reason+".")
 			return
 		} else if err != nil {
 			s.fail(w, err)

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 }