fix(master): keep rollbacks of auto services until a new digest (#9)

Adds services.last_auto_digest so the poller starts one automatic update per registry digest, and re-reads the service before starting it. Also records a bare image's running version as tag latest, only records well-formed digests, and shows a digest-less version unless a digest row of its tag was first seen at or after it.
2026-10-13 08:28:50 +02:00
parent f169eddc4c
commit 7e4b927293
9 changed files with 325 additions and 45 deletions
@@ -25,7 +25,9 @@ const maxErrorBytes = 500
 // Store is the storage the poller needs. *store.Store satisfies it.
 type Store interface {
 	ListServices(ctx context.Context) ([]model.Service, error)
+	GetService(ctx context.Context, id int64) (model.Service, error)
 	SetAvailable(ctx context.Context, id int64, a model.Available) error
+	SetLastAutoDigest(ctx context.Context, id int64, digest string) error
 	SetPollError(ctx context.Context, ids []int64, msg string) error
 }

@@ -74,7 +76,7 @@ func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Lo
 func SkipReason(svc model.Service) string {
 	switch {
 	case jobs.FollowedRef(svc.ActualRef) == "":
-		return "follows no tag"
+		return model.ReasonNoTag
 	case svc.Built:
 		return "built from source"
 	case svc.ActualDigest == "":
@@ -208,35 +210,74 @@ func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, sta
 	stats.ok++
 	p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" })
 	for _, svc := range svcs {
-		if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
+		if !p.suggest(ctx, svc, l, stats) {
 			continue
 		}
-		err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()})
-		if err != nil {
-			p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err)
-			continue
+		if autoWanted(svc, l.ref, l.digest) {
+			p.startAuto(ctx, svc.ID, l.ref, l.digest, stats)
 		}
-		stats.wrote = true
-		stats.suggested++
-		if svc.UpdateMode != model.ModeAuto {
-			continue
-		}
-		p.startAuto(ctx, svc, l.ref, stats)
 	}
 }

-// startAuto starts the update job of an auto service to ref. A service that
-// already has a job or cannot be updated is not an error for the cycle.
-func (p *Poller) startAuto(ctx context.Context, svc model.Service, ref string, stats *cycleStats) {
-	_, err := p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
+// suggest marks l.digest as the available version of svc unless the service
+// already runs it or already has exactly that suggestion. It returns false
+// only when writing failed.
+func (p *Poller) suggest(ctx context.Context, svc model.Service, l lookup, stats *cycleStats) bool {
+	if svc.ActualDigest == l.digest || (svc.AvailableRef == l.ref && svc.AvailableDigest == l.digest) {
+		return true
+	}
+	err := p.store.SetAvailable(ctx, svc.ID, model.Available{Ref: l.ref, Digest: l.digest, Source: model.SourcePoll, At: p.now()})
+	if err != nil {
+		p.log.Error("poll_set_available_failed", "service_id", svc.ID, "ref", l.ref, "error", err)
+		return false
+	}
+	stats.wrote = true
+	stats.suggested++
+	return true
+}
+
+// autoWanted reports whether an update of svc to ref at digest should be
+// started automatically: the service is in auto mode, not removed, does not
+// already run that digest, was not already updated automatically to it
+// (otherwise a rollback to an older version would be undone at the next
+// cycle), and its last job did not already deploy it (desired_ref holds the
+// digest until the agent reports the new inventory).
+func autoWanted(svc model.Service, ref, digest string) bool {
+	return svc.RemovedAt == nil &&
+		svc.UpdateMode == model.ModeAuto &&
+		svc.ActualDigest != digest &&
+		svc.LastAutoDigest != digest &&
+		svc.DesiredRef != ref+"@"+digest
+}
+
+// startAuto starts the update job of an auto service to ref and remembers
+// digest as services.last_auto_digest, so the same digest is not started
+// again, for instance after the user rolled back. It re-reads the service
+// first, because a cycle can run for a long time and the user may have
+// switched the service to manual meanwhile. A service that already has a
+// job or cannot be updated is not an error for the cycle, and last_auto_digest
+// is left alone so the next cycle tries again.
+func (p *Poller) startAuto(ctx context.Context, id int64, ref, digest string, stats *cycleStats) {
+	svc, err := p.store.GetService(ctx, id)
+	if err != nil {
+		p.log.Warn("poll_get_service_failed", "service_id", id, "error", err)
+		return
+	}
+	if !autoWanted(svc, ref, digest) {
+		return
+	}
+	_, err = p.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref})
 	var notUpdatable *model.NotUpdatableError
 	switch {
 	case err == nil:
 		stats.jobs++
+		if err := p.store.SetLastAutoDigest(ctx, id, digest); err != nil {
+			p.log.Error("poll_set_last_auto_digest_failed", "service_id", id, "error", err)
+		}
 	case errors.Is(err, model.ErrJobActive), errors.As(err, &notUpdatable):
-		p.log.Debug("poll_job_not_started", "service_id", svc.ID, "ref", ref, "reason", err.Error())
+		p.log.Debug("poll_job_not_started", "service_id", id, "ref", ref, "reason", err.Error())
 	default:
-		p.log.Warn("poll_job_failed", "service_id", svc.ID, "ref", ref, "error", err)
+		p.log.Warn("poll_job_failed", "service_id", id, "ref", ref, "error", err)
 	}
 }

@@ -30,9 +30,16 @@ type pollErrorCall struct {
 	msg string
 }

+type lastAutoCall struct {
+	id     int64
+	digest string
+}
+
 type fakeStore struct {
 	mu         sync.Mutex
 	services   []model.Service
+	fresh      map[int64]model.Service // what GetService returns instead of the listed row
+	lastAuto   []lastAutoCall
 	listErr    error
 	available  []availableCall
 	pollErrors []pollErrorCall
@@ -43,6 +50,34 @@ func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) {
 	return f.services, f.listErr
 }

+func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if s, ok := f.fresh[id]; ok {
+		return s, nil
+	}
+	for _, s := range f.services {
+		if s.ID == id {
+			return s, nil
+		}
+	}
+	return model.Service{}, model.ErrNotFound
+}
+
+// SetLastAutoDigest records the call and, like the database, changes the
+// row the next cycle lists.
+func (f *fakeStore) SetLastAutoDigest(_ context.Context, id int64, digest string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.lastAuto = append(f.lastAuto, lastAutoCall{id, digest})
+	for i := range f.services {
+		if f.services[i].ID == id {
+			f.services[i].LastAutoDigest = digest
+		}
+	}
+	return nil
+}
+
 func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -175,7 +210,7 @@ func TestSkipReason(t *testing.T) {
 	}{
 		{"tag and digest", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a"}, ""},
 		{"bare image follows latest", model.Service{ActualRef: "r", ActualDigest: "sha256:a"}, ""},
-		{"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, "follows no tag"},
+		{"digest only follows no tag", model.Service{ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, model.ReasonNoTag},
 		{"built", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, "built from source"},
 		{"no digest", model.Service{ActualRef: "r:1"}, "no registry digest"},
 		{"self and container are polled", model.Service{ActualRef: "r:1", ActualDigest: "sha256:a", Self: true, Kind: "container"}, ""},
@@ -252,8 +287,10 @@ func TestNewDigest(t *testing.T) {
 				At: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC)}, st.available[0].a)
 			if tt.wantStart {
 				assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts)
+				assert.Equal(t, []lastAutoCall{{1, "sha256:new"}}, st.lastAuto)
 			} else {
 				assert.Empty(t, j.starts)
+				assert.Empty(t, st.lastAuto)
 			}
 			assert.Equal(t, []events.Topic{events.Services}, pub.topics)
 		})
@@ -268,6 +305,7 @@ func TestSameSuggestionIsIdempotent(t *testing.T) {
 	suggested := svc(2, ref)
 	suggested.UpdateMode = model.ModeAuto
 	suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new"
+	suggested.LastAutoDigest = "sha256:new" // its job was started in an earlier cycle
 	staleSuggestion := svc(3, ref)
 	staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older"
 	st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}}
@@ -303,6 +341,89 @@ func TestStartErrorsDoNotAbortCycle(t *testing.T) {

 	assert.Len(t, st.available, 3)
 	assert.Len(t, j.starts, 3)
+	assert.Empty(t, st.lastAuto, "a job that was not created is retried next cycle")
+}
+
+func TestRollbackOfAutoServiceIsNotUndone(t *testing.T) {
+	ref := repoA + ":main"
+	// Updated to sha256:new automatically, then rolled back to sha256:old by hand.
+	s := svc(1, ref+"@sha256:old")
+	s.UpdateMode = model.ModeAuto
+	s.LastAutoDigest = "sha256:new"
+	s.DesiredRef = ref + "@sha256:old"
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+	p.Cycle(context.Background())
+
+	assert.Empty(t, j.starts, "same registry digest: the rollback stays")
+	assert.Equal(t, "sha256:new", st.available[0].a.Digest, "the newer version is still offered")
+
+	reg.digests[ref] = "sha256:newer"
+	p.Cycle(context.Background())
+
+	assert.Equal(t, []startCall{{1, jobs.StartRequest{Trigger: model.TriggerPoll, Ref: ref}}}, j.starts, "a new digest moves it again")
+	assert.Equal(t, []lastAutoCall{{1, "sha256:newer"}}, st.lastAuto)
+}
+
+func TestActiveJobIsRetriedNextCycle(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.UpdateMode = model.ModeAuto
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{errFor: map[int64]error{1: model.ErrJobActive}}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+	require.Len(t, j.starts, 1)
+	assert.Empty(t, st.lastAuto)
+
+	j.errFor = nil
+	p.Cycle(context.Background())
+
+	assert.Len(t, j.starts, 2)
+	assert.Equal(t, []lastAutoCall{{1, "sha256:new"}}, st.lastAuto)
+
+	p.Cycle(context.Background())
+	assert.Len(t, j.starts, 2, "started once the job exists")
+}
+
+func TestJobThatAlreadyDeployedTheDigestIsNotStartedAgain(t *testing.T) {
+	ref := repoA + ":1.4"
+	// FinishJob set desired_ref; the agent has not reported the new digest yet.
+	s := svc(1, ref+"@sha256:old")
+	s.UpdateMode = model.ModeAuto
+	s.DesiredRef = ref + "@sha256:new"
+	st := &fakeStore{services: []model.Service{s}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Empty(t, j.starts)
+}
+
+func TestServiceSwitchedToManualDuringTheCycleIsNotStarted(t *testing.T) {
+	ref := repoA + ":1.4"
+	s := svc(1, ref)
+	s.UpdateMode = model.ModeAuto
+	manual := s
+	manual.UpdateMode = model.ModeManual
+	st := &fakeStore{services: []model.Service{s}, fresh: map[int64]model.Service{1: manual}}
+	reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
+	j := &fakeJobs{}
+	p, _ := newPoller(st, j, reg, Config{})
+
+	p.Cycle(context.Background())
+
+	assert.Len(t, st.available, 1, "still suggested")
+	assert.Empty(t, j.starts)
+	assert.Empty(t, st.lastAuto)
 }

 func TestErrorIsPerRef(t *testing.T) {
@@ -384,18 +505,37 @@ func TestListErrorEndsTheCycle(t *testing.T) {
 }

 func TestConcurrencyLimit(t *testing.T) {
+	const workers = 3
 	var svcs []model.Service
 	for i := 0; i < 8; i++ {
 		svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i)))
 	}
 	st := &fakeStore{services: svcs}
-	reg := &fakeRegistry{delay: 20 * time.Millisecond}
-	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: 3})
+	reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 8)}
+	p, _ := newPoller(st, &fakeJobs{}, reg, Config{Workers: workers})
+	done := make(chan struct{})
+	go func() { p.Cycle(context.Background()); close(done) }()

-	p.Cycle(context.Background())
+	for i := 0; i < workers; i++ {
+		select {
+		case <-reg.started:
+		case <-time.After(2 * time.Second):
+			t.Fatalf("only %d lookups started, want %d", i, workers)
+		}
+	}
+	select {
+	case <-reg.started:
+		t.Fatal("more lookups in flight than Workers")
+	case <-time.After(100 * time.Millisecond):
+	}
+	close(reg.release)

+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("Cycle did not finish")
+	}
 	assert.Len(t, reg.calls, 8)
-	assert.Equal(t, 3, reg.maxSeen)
 }

 func TestCycleStopsOnCancel(t *testing.T) {