feat(store): drift state, move candidates and service merge (#10)

Adds services.awaiting_inventory, Service.Drifted/RunsRef/RunningRef, AcceptRunning, MoveCandidates, MergeSources and MergeService, and maps a job for a deleted service to ErrNotFound.

internal/e2e/agent_master_test.go

@@ -245,6 +245,7 @@ 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
+		svc.AwaitingInventory = true
 		if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) {
 			svc.LastTagDigest = d
 		}

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/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

@@ -68,6 +68,10 @@ type Service struct {
 	// not start another job for the same digest, so a rollback sticks until the
 	// registry shows a different one.
 	LastTagDigest string
+	// 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
@@ -113,6 +117,63 @@ 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()
+}
+
+// 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 + 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
 }
@@ -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) {