From af61f17dee4009dcef16caafb8e053eb6436d530 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:05:21 +0200 Subject: [PATCH] feat(master): record image versions and track poll suggestions in the store (#9) --- internal/e2e/agent_master_test.go | 54 +++++- internal/master/model/model.go | 31 +++- internal/master/model/model_test.go | 18 ++ internal/master/store/jobs.go | 19 ++- .../master/store/migrations/0004_polling.sql | 19 +++ internal/master/store/services.go | 103 ++++++++++- .../master/store/versions_integration_test.go | 160 ++++++++++++++++++ .../master/store/webhook_integration_test.go | 35 ++-- 8 files changed, 403 insertions(+), 36 deletions(-) create mode 100644 internal/master/store/migrations/0004_polling.sql create mode 100644 internal/master/store/versions_integration_test.go diff --git a/internal/e2e/agent_master_test.go b/internal/e2e/agent_master_test.go index 9036eb1..2974cf8 100644 --- a/internal/e2e/agent_master_test.go +++ b/internal/e2e/agent_master_test.go @@ -44,6 +44,51 @@ type memStore struct { jobs map[int64]*model.Job nextJob int64 finishApplied int // FinishJob calls that returned true + versions map[int64]model.ImageVersion +} + +// seedVersion adds an image_versions row that GetImageVersion returns. +func (m *memStore) seedVersion(v model.ImageVersion) { + m.mu.Lock() + defer m.mu.Unlock() + if m.versions == nil { + m.versions = map[int64]model.ImageVersion{} + } + m.versions[v.ID] = v +} + +func (m *memStore) GetImageVersion(_ context.Context, id int64) (model.ImageVersion, error) { + m.mu.Lock() + defer m.mu.Unlock() + v, ok := m.versions[id] + if !ok { + return model.ImageVersion{}, model.ErrNotFound + } + return v, nil +} + +func (m *memStore) ListServices(_ context.Context) ([]model.Service, error) { + m.mu.Lock() + defer m.mu.Unlock() + var out []model.Service + for _, svc := range m.svcRows { + if svc.RemovedAt == nil { + out = append(out, svc) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, nil +} + +func (m *memStore) SetPollError(_ context.Context, ids []int64, msg string) error { + m.mu.Lock() + defer m.mu.Unlock() + for _, id := range ids { + svc := m.svcRows[id] + svc.LastPollError = msg + m.svcRows[id] = svc + } + return nil } // seedService adds a service row that jobs can be triggered for. @@ -88,7 +133,7 @@ func (m *memStore) SetAvailable(_ context.Context, id int64, a model.Available) if !ok { return model.ErrNotFound } - svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Source, &a.At + svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = a.Ref, a.Digest, a.Source, &a.At m.svcRows[id] = svc return nil } @@ -185,9 +230,10 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo svc := m.svcRows[j.ServiceID] svc.DesiredRef = f.Ref // Same rule as the SQL: the job was created for the suggested ref after it was reported. - requested, _, _ := strings.Cut(requestedRef, "@") - if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) { - svc.AvailableRef, svc.AvailableSource, svc.AvailableAt = "", "", nil + requested, digest, _ := strings.Cut(requestedRef, "@") + if svc.AvailableAt != nil && svc.AvailableRef == requested && !svc.AvailableAt.After(j.CreatedAt) && + (digest == "" || svc.AvailableDigest == digest) { + svc.AvailableRef, svc.AvailableDigest, svc.AvailableSource, svc.AvailableAt = "", "", "", nil } m.svcRows[j.ServiceID] = svc } diff --git a/internal/master/model/model.go b/internal/master/model/model.go index 3b4a8ff..975538c 100644 --- a/internal/master/model/model.go +++ b/internal/master/model/model.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "github.com/shcizo/package-updater/internal/imageref" "github.com/shcizo/package-updater/internal/protocol" ) @@ -56,8 +57,10 @@ type Service struct { // AvailableRef is the newest ref a webhook or poll reported for this // service; "" if none, or if a later job already deployed it. AvailableRef string - AvailableSource string // SourceWebhook; "" when AvailableRef is empty + AvailableDigest string // digest of AvailableRef when known (poll); "" from the webhook + AvailableSource string // SourceWebhook or SourcePoll; "" when AvailableRef is empty AvailableAt *time.Time + LastPollError string // last registry error for the followed ref; "" when fine FirstSeenAt time.Time UpdatedAt time.Time RemovedAt *time.Time @@ -69,18 +72,40 @@ const ( ModeAuto = "auto" ) -// Sources of an available ref. +// Sources of an available ref and of an ImageVersion. const ( SourceWebhook = "webhook" + SourcePoll = "poll" + // SourceRunning is used by image_versions only: an agent reported the + // version as running. + SourceRunning = "running" ) // Available is a newer ref reported for a service. type Available struct { Ref string - Source string // SourceWebhook + Digest string // "" when unknown (webhook) + Source string // SourceWebhook or SourcePoll At time.Time } +// ImageVersion is one version of an image repo the master has seen. Tag and +// Digest are "" when the version has no tag or its digest is unknown; they +// are never both empty. +type ImageVersion struct { + ID int64 + Repo string + Tag string + Digest string + Source string // SourceWebhook, SourcePoll or SourceRunning: the first source that saw it + FirstSeenAt time.Time +} + +// Ref returns "repo:tag@digest", "repo:tag" or "repo@digest". +func (v ImageVersion) Ref() string { + return imageref.Ref{Repo: v.Repo, Tag: v.Tag, Digest: v.Digest}.String() +} + // DisplayName is the human-facing name of the service. func (s Service) DisplayName() string { switch s.Kind { diff --git a/internal/master/model/model_test.go b/internal/master/model/model_test.go index 48c7947..96cb8af 100644 --- a/internal/master/model/model_test.go +++ b/internal/master/model/model_test.go @@ -40,6 +40,24 @@ func TestDisplayName(t *testing.T) { } } +func TestImageVersionRef(t *testing.T) { + cases := []struct { + name string + v ImageVersion + want string + }{ + {"tag and digest", ImageVersion{Repo: "r/x", Tag: "1", Digest: "sha256:a"}, "r/x:1@sha256:a"}, + {"tag only", ImageVersion{Repo: "r/x", Tag: "1"}, "r/x:1"}, + {"digest only", ImageVersion{Repo: "r/x", Digest: "sha256:a"}, "r/x@sha256:a"}, + {"registry port", ImageVersion{Repo: "localhost:5000/x", Tag: "1"}, "localhost:5000/x:1"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + assert.Equal(t, c.want, c.v.Ref()) + }) + } +} + func TestJob_Active(t *testing.T) { cases := []struct { status JobStatus diff --git a/internal/master/store/jobs.go b/internal/master/store/jobs.go index 31a9da9..1789060 100644 --- a/internal/master/store/jobs.go +++ b/internal/master/store/jobs.go @@ -189,9 +189,11 @@ func (s *Store) JobProgress(ctx context.Context, agentID, jobID int64, chunk str // started_at (the agent's, or the existing one if the agent sent none), // finished_at; to_ref = f.Ref when set. On success it also sets the // service's desired_ref = f.Ref and clears its available ref when that is -// the ref the job was created for (ignoring any @digest) and was reported no -// later than the job was created (that version is now running), in the same -// transaction. Returns false if no +// the ref the job was created for and was reported no later than the job was +// created (that version is now running), in the same transaction. A job +// created for "repo:tag@digest" clears only a suggestion with that same +// digest, so rolling back to an older digest keeps a newer suggestion (and a +// suggestion without a digest, such as a webhook's). Returns false if no // such active job belongs to agentID. func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error) { tx, err := s.pool.Begin(ctx) @@ -226,13 +228,14 @@ func (s *Store) FinishJob(ctx context.Context, agentID, jobID int64, f model.Job return false, fmt.Errorf("finish job: %w", err) } if f.Status == model.JobSucceeded { - requestedRef, _, _ = strings.Cut(requestedRef, "@") + requestedRef, requestedDigest, _ := strings.Cut(requestedRef, "@") if _, err := tx.Exec(ctx, ` UPDATE services SET desired_ref = $2, - available_ref = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_ref END, - available_source = CASE WHEN available_ref = $4 AND available_at <= $3 THEN '' ELSE available_source END, - available_at = CASE WHEN available_ref = $4 AND available_at <= $3 THEN NULL ELSE available_at END - WHERE id = $1`, serviceID, f.Ref, createdAt, requestedRef); err != nil { + 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 { return false, fmt.Errorf("finish job: desired ref: %w", err) } } diff --git a/internal/master/store/migrations/0004_polling.sql b/internal/master/store/migrations/0004_polling.sql new file mode 100644 index 0000000..a1da48d --- /dev/null +++ b/internal/master/store/migrations/0004_polling.sql @@ -0,0 +1,19 @@ +ALTER TABLE services DROP CONSTRAINT services_available_source_check; +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 ''; + +-- Every version the master has seen. '' means "no tag" / "digest unknown"; +-- never NULL, so the unique constraint deduplicates. +CREATE TABLE image_versions ( + id bigserial PRIMARY KEY, + repo text NOT NULL, + tag text NOT NULL DEFAULT '', + digest text NOT NULL DEFAULT '', + source text NOT NULL CHECK (source IN ('webhook', 'poll', 'running')), + first_seen_at timestamptz NOT NULL, + CONSTRAINT image_versions_tag_or_digest CHECK (tag <> '' OR digest <> ''), + CONSTRAINT image_versions_unique UNIQUE (repo, tag, digest) +); +CREATE INDEX image_versions_recent ON image_versions (repo, first_seen_at DESC, id DESC); diff --git a/internal/master/store/services.go b/internal/master/store/services.go index a71f6d5..0649086 100644 --- a/internal/master/store/services.go +++ b/internal/master/store/services.go @@ -70,6 +70,25 @@ 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 svc.Digest != "" { + ref := imageref.Parse(svc.Image) + if err := insertVersion(ctx, db, ref.Repo, ref.Tag, svc.Digest, model.SourceRunning, at); err != nil { + return err + } + } + return nil +} + +// insertVersion records a version in image_versions unless the same +// (repo, tag, digest) is already there; the first sighting keeps its source +// and first_seen_at. +func insertVersion(ctx context.Context, db execer, repo, tag, digest, source string, at time.Time) error { + if _, err := db.Exec(ctx, ` + INSERT INTO image_versions (repo, tag, digest, source, first_seen_at) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`, + repo, tag, digest, source, at); err != nil { + return fmt.Errorf("record version %s:%s: %w", repo, tag, err) + } return nil } @@ -149,8 +168,8 @@ func (s *Store) RemoveService(ctx context.Context, agentID int64, clusterID, age const serviceColumns = `s.id, s.kind, s.location_key, s.agent_id, COALESCE(a.name, ''), 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_source, s.available_at, - s.first_seen_at, s.updated_at, s.removed_at` + s.status, s.self, s.update_mode, s.available_ref, s.available_digest, s.available_source, s.available_at, + s.last_poll_error, s.first_seen_at, s.updated_at, s.removed_at` func scanService(row pgx.Row) (model.Service, error) { var m model.Service @@ -158,8 +177,8 @@ func scanService(row pgx.Row) (model.Service, error) { err := row.Scan(&m.ID, &kind, &m.LocationKey, &m.AgentID, &m.AgentName, &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.AvailableSource, &m.AvailableAt, - &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) + &m.Status, &m.Self, &m.UpdateMode, &m.AvailableRef, &m.AvailableDigest, &m.AvailableSource, &m.AvailableAt, + &m.LastPollError, &m.FirstSeenAt, &m.UpdatedAt, &m.RemovedAt) m.Kind = protocol.Kind(kind) return m, err } @@ -198,16 +217,84 @@ func (s *Store) queryServices(ctx context.Context, what, extraWhere string, args } // SetAvailable records a newer ref for the service, replacing any earlier -// suggestion. +// suggestion, and records the version in image_versions in the same +// transaction (the tag is parsed from a.Ref). func (s *Store) SetAvailable(ctx context.Context, id int64, a model.Available) error { - if _, err := s.pool.Exec(ctx, ` - UPDATE services SET available_ref = $2, available_source = $3, available_at = $4 - WHERE id = $1`, id, a.Ref, a.Source, a.At); err != nil { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("set available: begin: %w", err) + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, ` + UPDATE services SET available_ref = $2, available_digest = $3, available_source = $4, available_at = $5 + WHERE id = $1`, id, a.Ref, a.Digest, a.Source, a.At); err != nil { return fmt.Errorf("set available: %w", err) } + ref := imageref.Parse(a.Ref) + if err := insertVersion(ctx, tx, ref.Repo, ref.Tag, a.Digest, a.Source, a.At); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("set available: commit: %w", err) + } return nil } +// SetPollError sets last_poll_error on the given services; msg "" clears it. +func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error { + if len(ids) == 0 { + return nil + } + if _, err := s.pool.Exec(ctx, + `UPDATE services SET last_poll_error = $2 WHERE id = ANY($1)`, ids, msg); err != nil { + return fmt.Errorf("set poll error: %w", err) + } + return nil +} + +// RecentVersions returns up to limit versions of repo, most recently first +// seen first. A version without a digest is left out when a version with the +// same tag and a digest exists. +func (s *Store) RecentVersions(ctx context.Context, repo string, limit int) ([]model.ImageVersion, error) { + rows, err := s.pool.Query(ctx, ` + SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions v + WHERE v.repo = $1 + AND NOT (v.digest = '' AND EXISTS ( + SELECT 1 FROM image_versions d + WHERE d.repo = v.repo AND d.tag = v.tag AND d.digest <> '')) + ORDER BY v.first_seen_at DESC, v.id DESC + LIMIT $2`, repo, limit) + if err != nil { + return nil, fmt.Errorf("recent versions: %w", err) + } + defer rows.Close() + var out []model.ImageVersion + for rows.Next() { + var v model.ImageVersion + if err := rows.Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt); err != nil { + return nil, fmt.Errorf("scan version: %w", err) + } + out = append(out, v) + } + return out, rows.Err() +} + +// GetImageVersion returns one version. Returns model.ErrNotFound if the ID +// does not exist. +func (s *Store) GetImageVersion(ctx context.Context, id int64) (model.ImageVersion, error) { + var v model.ImageVersion + err := s.pool.QueryRow(ctx, ` + SELECT id, repo, tag, digest, source, first_seen_at FROM image_versions WHERE id = $1`, id, + ).Scan(&v.ID, &v.Repo, &v.Tag, &v.Digest, &v.Source, &v.FirstSeenAt) + if errors.Is(err, pgx.ErrNoRows) { + return model.ImageVersion{}, model.ErrNotFound + } + if err != nil { + return model.ImageVersion{}, fmt.Errorf("get image version: %w", err) + } + return v, nil +} + // SetUpdateMode sets the service's update mode (model.ModeManual or // model.ModeAuto). Returns model.ErrNotFound if the service does not exist // or is marked removed. diff --git a/internal/master/store/versions_integration_test.go b/internal/master/store/versions_integration_test.go new file mode 100644 index 0000000..5d86c98 --- /dev/null +++ b/internal/master/store/versions_integration_test.go @@ -0,0 +1,160 @@ +//go:build integration + +package store + +import ( + "context" + "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 sluskRepo = "gitea.shcizo.se/shcizo/slusk" + +func versionsOf(t *testing.T, s *Store, repo string) []model.ImageVersion { + t.Helper() + got, err := s.RecentVersions(context.Background(), repo, 100) + require.NoError(t, err) + return got +} + +func TestSetAvailable_RecordsVersion(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + _, svc := jobFixture(t, s, "a") + + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0})) + + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, sluskRepo+":1.5.0", got.AvailableRef) + assert.Equal(t, "sha256:n1", got.AvailableDigest) + assert.Equal(t, model.SourcePoll, got.AvailableSource) + vs := versionsOf(t, s, sluskRepo) + var found model.ImageVersion + for _, v := range vs { + if v.Digest == "sha256:n1" { + found = v + } + } + require.NotZero(t, found.ID) + assert.Equal(t, "1.5.0", found.Tag) + assert.Equal(t, model.SourcePoll, found.Source) + + // Seeing the same version again keeps the first sighting. + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourceWebhook, At: t0.Add(time.Hour)})) + again, err := s.GetImageVersion(ctx, found.ID) + require.NoError(t, err) + assert.Equal(t, model.SourcePoll, again.Source) + assert.True(t, t0.Equal(again.FirstSeenAt)) + assert.Len(t, versionsOf(t, s, sluskRepo), len(vs)) +} + +func TestSnapshotRecordsRunningVersions(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + id := newAgent(t, s, "node-2") + tagless := protocol.Service{Kind: protocol.KindSwarm, SwarmService: "g_g", Image: "gitea/gitea@sha256:87a6", Digest: "sha256:87a6", Status: "running"} + bare := protocol.Service{Kind: protocol.KindContainer, ContainerName: "pg", Image: "postgres", Digest: "sha256:pg1", Status: "running"} + nodigest := protocol.Service{Kind: protocol.KindContainer, ContainerName: "nd", Image: "nodigest:1", Status: "running"} + hello := protocol.Hello{Hostname: "node-2", SwarmManager: true, SwarmClusterID: "c1"} + svcs := []protocol.Service{slusk, tagless, bare, nodigest} + + require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0)) + require.NoError(t, s.ApplySnapshot(ctx, id, hello, svcs, t0.Add(time.Hour))) + + got := versionsOf(t, s, sluskRepo) + require.Len(t, got, 1) + assert.Equal(t, "1.4.2", got[0].Tag) + assert.Equal(t, "sha256:d1", got[0].Digest) + assert.Equal(t, model.SourceRunning, got[0].Source) + assert.True(t, t0.Equal(got[0].FirstSeenAt), "re-snapshot does not bump first_seen_at") + tl := versionsOf(t, s, "gitea/gitea") + require.Len(t, tl, 1) + assert.Equal(t, "", tl[0].Tag) + assert.Equal(t, "sha256:87a6", tl[0].Digest) + b := versionsOf(t, s, "postgres") + require.Len(t, b, 1) + assert.Equal(t, "", b[0].Tag, "a bare image is recorded without a tag, not as latest") + assert.Empty(t, versionsOf(t, s, "nodigest")) +} + +func TestRecentVersions(t *testing.T) { + ctx := context.Background() + t.Run("order, limit and repo filter", func(t *testing.T) { + s := openTestStore(t) + _, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: "other/repo:x", Digest: "sha256:ox", Source: model.SourcePoll, At: t0})) + // 12 versions of the service's repo, one per minute. + for i := 0; i < 12; i++ { + tag := "t" + string(rune('a'+i)) + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: "sha256:" + tag, Source: model.SourcePoll, At: t0.Add(time.Duration(i+1) * time.Minute)})) + } + + got, err := s.RecentVersions(ctx, sluskRepo, 10) + + require.NoError(t, err) + require.Len(t, got, 10) + assert.Equal(t, "tl", got[0].Tag, "newest first") + assert.Equal(t, "tc", got[9].Tag) + for _, v := range got { + assert.Equal(t, sluskRepo, v.Repo) + } + }) + + t.Run("digest-less row is hidden once its tag has a digest", func(t *testing.T) { + s := openTestStore(t) + _, svc := jobFixture(t, s, "a") + add := func(tag, digest string, at time.Time) { + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":" + tag, Digest: digest, Source: model.SourceWebhook, At: at})) + } + add("main", "", t0.Add(time.Minute)) + add("solo", "", t0.Add(2*time.Minute)) + require.Len(t, versionsOf(t, s, sluskRepo), 3, "running 1.4.2 plus two digest-less rows") + + add("main", "sha256:m1", t0.Add(3*time.Minute)) + + var refs []string + for _, v := range versionsOf(t, s, sluskRepo) { + refs = append(refs, v.Ref()) + } + assert.Equal(t, []string{sluskRepo + ":main@sha256:m1", sluskRepo + ":solo", sluskRepo + ":1.4.2@sha256:d1"}, refs) + }) +} + +func TestGetImageVersion_NotFound(t *testing.T) { + s := openTestStore(t) + + _, err := s.GetImageVersion(context.Background(), 99999) + + assert.ErrorIs(t, err, model.ErrNotFound) +} + +func TestSetPollError(t *testing.T) { + s := openTestStore(t) + ctx := context.Background() + agent, svc := jobFixture(t, s, "a") + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: sluskRepo + ":1.5.0", Digest: "sha256:n1", Source: model.SourcePoll, At: t0})) + + require.NoError(t, s.SetPollError(ctx, []int64{svc}, "boom")) + require.NoError(t, s.SetPollError(ctx, nil, "ignored")) + got, err := s.GetService(ctx, svc) + require.NoError(t, err) + assert.Equal(t, "boom", got.LastPollError) + + 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, "boom", got.LastPollError, "inventory writes do not touch the poll error") + assert.Equal(t, "sha256:n1", got.AvailableDigest, "nor the suggestion") + + require.NoError(t, s.SetPollError(ctx, []int64{svc}, "")) + got, err = s.GetService(ctx, svc) + require.NoError(t, err) + assert.Empty(t, got.LastPollError) +} diff --git a/internal/master/store/webhook_integration_test.go b/internal/master/store/webhook_integration_test.go index e047cec..3139647 100644 --- a/internal/master/store/webhook_integration_test.go +++ b/internal/master/store/webhook_integration_test.go @@ -91,27 +91,34 @@ func TestUpsertKeepsUpdateMode(t *testing.T) { func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) { succeed := model.JobFinish{Status: model.JobSucceeded, Ref: "r:2@sha256:d", StartedAt: t0, FinishedAt: t0} tests := []struct { - name string - availableRef string - availableAt time.Time // job is created at t0 for r:2 - finish model.JobFinish - wantCleared bool + name string + availableRef string + availableDigest string + availableAt time.Time // the job is created at t0 + toRef string // the job's to_ref as created + finish model.JobFinish + wantCleared bool }{ - {"same ref, suggestion before the job", "r:2", t0.Add(-time.Minute), succeed, true}, - {"same ref, suggestion at the same instant", "r:2", t0, succeed, true}, - {"same ref, suggestion after the job was created", "r:2", t0.Add(time.Minute), succeed, false}, - {"different ref keeps the suggestion", "r:3", t0.Add(-time.Minute), succeed, false}, - {"failed job keeps the suggestion", "r:2", t0.Add(-time.Minute), + {"same ref, suggestion before the job", "r:2", "", t0.Add(-time.Minute), "r:2", succeed, true}, + {"same ref, suggestion at the same instant", "r:2", "", t0, "r:2", succeed, true}, + {"same ref, suggestion after the job was created", "r:2", "", t0.Add(time.Minute), "r:2", succeed, false}, + {"different ref keeps the suggestion", "r:3", "", t0.Add(-time.Minute), "r:2", succeed, false}, + {"failed job keeps the suggestion", "r:2", "", t0.Add(-time.Minute), "r:2", model.JobFinish{Status: model.JobFailed, Error: "boom", StartedAt: t0, FinishedAt: t0}, false}, + {"tag job clears a poll suggestion with a digest", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2", succeed, true}, + {"pick of the suggested digest clears it", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:new", succeed, true}, + {"rollback to an older digest keeps the newer suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false}, + {"digest pick keeps a suggestion without digest", "r:2", "", t0.Add(-time.Minute), "r:2@sha256:old", succeed, false}, + {"tagless pick keeps the suggestion", "r:2", "sha256:new", t0.Add(-time.Minute), "r@sha256:old", succeed, false}, } 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.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Source: model.SourceWebhook, At: tt.availableAt})) + require.NoError(t, s.SetAvailable(ctx, svc, model.Available{Ref: tt.availableRef, Digest: tt.availableDigest, Source: model.SourcePoll, At: tt.availableAt})) nj := newJob(svc, agent) - nj.ToRef = "r:2" + nj.ToRef = tt.toRef j, err := s.CreateJob(ctx, nj) require.NoError(t, err) @@ -123,11 +130,13 @@ func TestFinishJob_ClearsAvailableOnlyWhenThatVersionWasDeployed(t *testing.T) { require.NoError(t, err) if tt.wantCleared { assert.Empty(t, got.AvailableRef) + assert.Empty(t, got.AvailableDigest) assert.Empty(t, got.AvailableSource) assert.Nil(t, got.AvailableAt) } else { assert.Equal(t, tt.availableRef, got.AvailableRef) - assert.Equal(t, model.SourceWebhook, got.AvailableSource) + assert.Equal(t, tt.availableDigest, got.AvailableDigest) + assert.Equal(t, model.SourcePoll, got.AvailableSource) assert.NotNil(t, got.AvailableAt) } })