shcizo/package-updater
1
0
Fork 0
Code Issues 8 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(master): remember the tag's deployed digest after any tag job (#9)

Browse Source 
Renames last_auto_digest to last_tag_digest. FinishJob sets it to the pinned
digest after any successful job whose to_ref had no digest, so a rollback
holds however the update was started.
This commit is contained in:
shcizo
2026-10-13 08:33:17 +02:00
parent 53a2a23f1b
commit 3166874433
13 changed files with 149 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/e2e/agent_master_test.go
+5 -2
View File
@@ -91,14 +91,14 @@ func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) erro
return nil
}
func (m *memStore) SetLastAutoDigest(_ context.Context, id int64, digest string) error {
func (m *memStore) SetLastTagDigest(_ context.Context, id int64, digest string) error {
m.mu.Lock()
defer m.mu.Unlock()
svc, ok := m.svcRows[id]
if !ok {
return model.ErrNotFound
}
svc.LastAutoDigest = digest
svc.LastTagDigest = digest
m.svcRows[id] = svc
return nil
}
@@ -241,6 +241,9 @@ 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, "@") && d != "" {
svc.LastTagDigest = d
}
// Same rule as the SQL: the job was created for the suggested ref after it was reported.
requested, digest, _ := strings.Cut(requestedRef, "@")
if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) &&
internal/e2e/versions_test.go
+35 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/shcizo/package-updater/internal/master/jobs"
"github.com/shcizo/package-updater/internal/master/model"
"github.com/shcizo/package-updater/internal/master/poller"
"github.com/shcizo/package-updater/internal/protocol"
@@ -88,7 +89,7 @@ func TestPollStartsAutoJob(t *testing.T) {
assert.Equal(t, slusk+":1.4.2", svc.AvailableRef)
assert.Equal(t, "sha256:new", svc.AvailableDigest)
assert.Equal(t, model.SourcePoll, svc.AvailableSource)
assert.Equal(t, "sha256:new", svc.LastAutoDigest)
assert.Equal(t, "sha256:new", svc.LastTagDigest)
p.Cycle(context.Background())
@@ -97,3 +98,36 @@ func TestPollStartsAutoJob(t *testing.T) {
close(env.exec.release)
env.waitStatus(t, job.ID, model.JobSucceeded)
}
// An update started by the webhook, then a rollback by hand: polling must not
// redo the update while the registry still shows the digest that was deployed.
func TestPollKeepsARollbackAfterAWebhookUpdate(t *testing.T) {
env := newJobEnv(t)
ctx := context.Background()
close(env.exec.release)
env.store.seedService(model.Service{ID: 1, Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
AgentID: ptr(agentID), ImageRepo: slusk, ActualRef: slusk + ":main@sha256:old", ActualDigest: "sha256:old",
UpdateMode: model.ModeAuto})
env.store.seedVersion(model.ImageVersion{ID: 5, Repo: slusk, Tag: "main", Digest: "sha256:old", Source: model.SourceRunning})
svc, err := env.store.GetService(ctx, 1)
require.NoError(t, err)
hook, err := env.jobs.Start(ctx, svc, jobs.StartRequest{Trigger: model.TriggerWebhook, Ref: slusk + ":main"})
require.NoError(t, err)
env.waitStatus(t, hook.ID, model.JobSucceeded)
svc, err = env.store.GetService(ctx, 1)
require.NoError(t, err)
assert.Equal(t, "sha256:e2e", svc.LastTagDigest, "the deployed digest of the tag is remembered")
pick, err := env.jobs.TriggerVersion(ctx, 1, 5, "alice")
require.NoError(t, err)
env.waitStatus(t, pick.ID, model.JobSucceeded)
jobsBefore := env.store.jobCount()
reg := &fixedRegistry{digests: map[string]string{slusk + ":main": "sha256:e2e"}}
p := poller.New(env.store, env.jobs, reg, env.m.broker, poller.Config{Workers: 1}, slog.New(slog.NewTextHandler(io.Discard, nil)))
p.Cycle(ctx)
assert.Equal(t, 1, reg.calls)
assert.Equal(t, jobsBefore, env.store.jobCount(), "the rollback stays")
}
internal/master/model/model.go
+9 -7
View File
@@ -61,13 +61,15 @@ type Service struct {
AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty
AvailableAt *time.Time
LastPollError string // last registry error for the followed ref; "" when fine
// LastAutoDigest is the registry digest the poller last started an
// automatic update for; "" if never. The poller does not start another
// one for the same digest, so a rollback sticks until a new digest appears.
LastAutoDigest string
FirstSeenAt time.Time
UpdatedAt time.Time
RemovedAt *time.Time
// LastTagDigest is the newest digest of the followed tag that was deployed
// or started: set when the poller starts an auto job and when any plain tag
// job succeeds (the digest the agent pinned); "" if never. The poller does
// 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
}
// Update modes of a service.
internal/master/poller/poller.go
+6 -6
View File
@@ -27,7 +27,7 @@ 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
SetLastTagDigest(ctx context.Context, id int64, digest string) error
SetPollError(ctx context.Context, ids []int64, msg string) error
}
@@ -246,16 +246,16 @@ func autoWanted(svc model.Service, ref, digest string) bool {
return svc.RemovedAt == nil &&
svc.UpdateMode == model.ModeAuto &&
svc.ActualDigest != digest &&
svc.LastAutoDigest != digest &&
svc.LastTagDigest != 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
// digest as services.last_tag_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
// job or cannot be updated is not an error for the cycle, and last_tag_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)
@@ -271,8 +271,8 @@ func (p *Poller) startAuto(ctx context.Context, id int64, ref, digest string, st
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)
if err := p.store.SetLastTagDigest(ctx, id, digest); err != nil {
p.log.Error("poll_set_last_tag_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", id, "ref", ref, "reason", err.Error())
internal/master/poller/poller_test.go
+5 -5
View File
@@ -64,15 +64,15 @@ func (f *fakeStore) GetService(_ context.Context, id int64) (model.Service, erro
return model.Service{}, model.ErrNotFound
}
// SetLastAutoDigest records the call and, like the database, changes the
// SetLastTagDigest 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 {
func (f *fakeStore) SetLastTagDigest(_ 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
f.services[i].LastTagDigest = digest
}
}
return nil
@@ -305,7 +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
suggested.LastTagDigest = "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}}
@@ -349,7 +349,7 @@ func TestRollbackOfAutoServiceIsNotUndone(t *testing.T) {
// 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.LastTagDigest = "sha256:new"
s.DesiredRef = ref + "@sha256:old"
st := &fakeStore{services: []model.Service{s}}
reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}}
internal/master/store/jobs.go
+8 -1
View File
@@ -229,13 +229,20 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job
}
if f.Status == model.JobSucceeded {
requestedRef, requestedDigest, _ := strings.Cut(requestedRef, "@")
// 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) {
tagDigest = d
}
if _, err := tx.Exec(ctx, `
UPDATE services SET desired_ref = $2,
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,
available_source = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN '' ELSE available_source END,
available_at = CASE WHEN available_ref = $4 AND available_at <= $3 AND ($5 = '' OR available_digest = $5) THEN NULL ELSE available_at END
WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef, requestedDigest); err != nil {
WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef, requestedDigest, tagDigest); err != nil {
return false, fmt.Errorf("finish job: desired ref: %w", err)
}
}
internal/master/store/migrations/0004_polling.sql
+1 -1
View File
@@ -3,7 +3,7 @@ ALTER TABLE services ADD CONSTRAINT services_available_source_check
CHECK (available_source IN ('', 'webhook', 'poll'));
ALTER TABLE services ADD COLUMN available_digest text NOT NULL DEFAULT '';
ALTER TABLE services ADD COLUMN last_poll_error text NOT NULL DEFAULT '';
ALTER TABLE services ADD COLUMN last_auto_digest text NOT NULL DEFAULT '';
ALTER TABLE services ADD COLUMN last_tag_digest text NOT NULL DEFAULT '';
-- Every version the master has seen. '' means "no tag" / "digest unknown";
-- never NULL, so the unique constraint deduplicates.
internal/master/store/services.go
+6 -6
View File
@@ -178,7 +178,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_auto_digest, s.first_seen_at, s.updated_at, s.removed_at`
s.last_poll_error, s.last_tag_digest, s.first_seen_at, s.updated_at, s.removed_at`
func scanService(row pgx.Row) (model.Service, error) {
var m model.Service
@@ -187,7 +187,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.LastAutoDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
&m.LastPollError, &m.LastTagDigest, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt)
m.Kind = protocol.Kind(kind)
return m, err
}
@@ -261,11 +261,11 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
return nil
}
// SetLastAutoDigest records the digest an automatic update was last started
// for; see model.Service.LastAutoDigest. Inventory writes never change it.
func (s *Store) SetLastAutoDigest(ctx context.Context, id int64, digest string) error {
// SetLastTagDigest records the digest of the followed tag that the poller
// started an update for; see model.Service.LastTagDigest. 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_auto_digest = $2 WHERE id = $1`, id, digest); err != nil {
`UPDATE services SET last_tag_digest = $2 WHERE id = $1`, id, digest); err != nil {
return fmt.Errorf("set last auto digest: %w", err)
}
return nil
internal/master/store/versions_integration_test.go
+5 -5
View File
@@ -195,22 +195,22 @@ func TestSetPollError(t *testing.T) {
assert.Empty(t, got.LastPollError)
}
func TestSetLastAutoDigest(t *testing.T) {
func TestSetLastTagDigest(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.Empty(t, got.LastAutoDigest)
assert.Empty(t, got.LastTagDigest)
require.NoError(t, s.SetLastAutoDigest(ctx, svc, "sha256:n1"))
require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:n1"))
require.NoError(t, s.UpsertService(ctx, agent, "", slusk, t0.Add(time.Hour)))
got, err = s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, "sha256:n1", got.LastAutoDigest, "inventory writes do not touch it")
assert.Equal(t, "sha256:n1", got.LastTagDigest, "inventory writes do not touch it")
list, err := s.ListServices(ctx)
require.NoError(t, err)
require.Len(t, list, 1)
assert.Equal(t, "sha256:n1", list[0].LastAutoDigest)
assert.Equal(t, "sha256:n1", list[0].LastTagDigest)
}
internal/master/store/webhook_integration_test.go
+46
View File
@@ -4,6 +4,7 @@ package store
import (
"context"
"strings"
"testing"
"time"
@@ -160,3 +161,48 @@ func TestFinishJob_WithoutAvailableLeavesItEmpty(t *testing.T) {
assert.Nil(t, got.AvailableAt)
assert.Equal(t, "r:1@sha256:d", got.DesiredRef)
}
func TestFinishJob_RemembersTheDeployedDigestOfATagJob(t *testing.T) {
good := "sha256:" + strings.Repeat("a", 64)
old := "sha256:" + strings.Repeat("b", 64)
ok := func(ref string) model.JobFinish {
return model.JobFinish{Status: model.JobSucceeded, Ref: ref, StartedAt: t0, FinishedAt: t0}
}
tests := []struct {
name string
trigger model.JobTrigger
toRef string
finish model.JobFinish
want string
}{
{"webhook tag job", model.TriggerWebhook, "r:2", ok("r:2@" + good), good},
{"manual tag job", model.TriggerManual, "r:2", ok("r:2@" + good), good},
{"poll tag job", model.TriggerPoll, "r:2", ok("r:2@" + good), good},
{"pinned pick leaves it", model.TriggerManual, "r:2@" + old, ok("r:2@" + old), "sha256:before"},
{"tagless pick leaves it", model.TriggerManual, "r@" + old, ok("r@" + old), "sha256:before"},
{"failed job leaves it", model.TriggerWebhook, "r:2",
model.JobFinish{Status: model.JobFailed, Error: "boom", Ref: "r:2@" + good, StartedAt: t0, FinishedAt: t0}, "sha256:before"},
{"malformed digest leaves it", model.TriggerWebhook, "r:2", ok("r:2@sha256:d"), "sha256:before"},
{"no pinned ref leaves it", model.TriggerWebhook, "r:2", ok(""), "sha256:before"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
s := openTestStore(t)
ctx := context.Background()
agent, svc := jobFixture(t, s, "a")
require.NoError(t, s.SetLastTagDigest(ctx, svc, "sha256:before"))
nj := newJob(svc, agent)
nj.ToRef, nj.Trigger = tt.toRef, tt.trigger
j, err := s.CreateJob(ctx, nj)
require.NoError(t, err)
done, err := s.FinishJob(ctx, agent, j.ID, tt.finish)
require.NoError(t, err)
require.True(t, done)
got, err := s.GetService(ctx, svc)
require.NoError(t, err)
assert.Equal(t, tt.want, got.LastTagDigest)
})
}
}