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

fix(master): review fixes for drift restore and accept (#10)

Browse Source
This commit is contained in:
shcizo
2026-10-13 10:36:01 +02:00
parent b7d105e841
commit df8f621683
11 changed files with 77 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/e2e/agent_master_test.go
+2 -5
View File
@@ -11,7 +11,6 @@ import (
"net"
"net/http"
"net/http/httptest"
"regexp"
"sort"
"strings"
"sync"
@@ -25,6 +24,7 @@ import (
"github.com/shcizo/package-updater/internal/agent/conn"
"github.com/shcizo/package-updater/internal/agent/updater"
"github.com/shcizo/package-updater/internal/agenttoken"
"github.com/shcizo/package-updater/internal/imageref"
"github.com/shcizo/package-updater/internal/master/events"
"github.com/shcizo/package-updater/internal/master/hub"
"github.com/shcizo/package-updater/internal/master/model"
@@ -32,9 +32,6 @@ import (
"github.com/shcizo/package-updater/internal/protocol"
)
// digestPattern is the digest format the real store records.
var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
type memStore struct {
mu sync.Mutex
agents map[string]model.Agent
@@ -246,7 +243,7 @@ func (m *memStore) FinishJob(_ context.Context, agentID, jobID int64, f model.Jo
svc := m.svcRows[j.ServiceID]
svc.DesiredRef = f.Ref
svc.AwaitingInventory = true
if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && digestPattern.MatchString(d) {
if _, d, _ := strings.Cut(f.Ref, "@"); !strings.Contains(requestedRef, "@") && imageref.IsDigest(d) {
svc.LastTagDigest = d
}
// Same rule as the SQL: the job was created for the suggested ref after it was reported.
internal/master/jobs/drift_test.go
+6 -1
View File
@@ -50,7 +50,7 @@ func TestRestorable(t *testing.T) {
{"not drifted", func(s *model.Service) { s.ActualDigest = driftD1 }, model.ReasonNotDrifted},
{"never updated", func(s *model.Service) { s.DesiredRef = "" }, model.ReasonNotDrifted},
{"awaiting inventory", func(s *model.Service) { s.AwaitingInventory = true }, model.ReasonNotDrifted},
{"removed", func(s *model.Service) { s.RemovedAt = ptr(time.Now()) }, model.ReasonNotDrifted},
{"removed", func(s *model.Service) { s.RemovedAt = ptr(time.Now()) }, model.ReasonRemoved},
{"flagged self", func(s *model.Service) { s.Self = true }, model.ReasonSelf},
{"built", func(s *model.Service) { s.Built = true }, model.ReasonBuilt},
{"container", func(s *model.Service) { s.Kind = protocol.KindContainer }, model.ReasonContainer},
@@ -166,6 +166,11 @@ func TestRestoreDesired_LastTagDigest(t *testing.T) {
{"digest in the actual ref", func(s *model.Service) {
s.ActualRef, s.ActualDigest = driftRepo+":1.4.2@"+driftD3, ""
}, driftD3},
{"last tag digest holds back a newer one", func(s *model.Service) {
s.DesiredRef = driftRepo + ":1.4.2@" + driftD1
s.LastTagDigest = driftD3
}, ""},
{"last tag digest equals the desired digest", func(s *model.Service) { s.LastTagDigest = driftD1 }, driftD2},
{"already remembered", func(s *model.Service) { s.LastTagDigest = driftD2 }, ""},
{"different tag", func(s *model.Service) { s.ActualRef = driftRepo + ":1.5.0" }, ""},
{"actual digest malformed", func(s *model.Service) { s.ActualDigest = "sha256:abc" }, ""},
internal/master/jobs/jobs.go
+14 -2
View File
@@ -120,6 +120,9 @@ func (s *Service) updatable(svc model.Service, explicitRef bool) error {
// desired version is deployed like an explicit ref, so a service that follows
// no tag is fine. The service page calls it for every render.
func (s *Service) Restorable(svc model.Service) error {
if svc.RemovedAt != nil {
return &model.NotUpdatableError{Reason: model.ReasonRemoved}
}
if !svc.Drifted() {
return &model.NotUpdatableError{Reason: model.ReasonNotDrifted}
}
@@ -140,7 +143,9 @@ func (s *Service) Restorable(svc model.Service) error {
//
// When the service had drifted to another digest of the same tag, that digest
// becomes last_tag_digest, so the poller does not deploy it again before the
// registry shows a newer one (the same rule as for any rollback).
// registry shows a newer one (the same rule as for any rollback). It is not
// written when last_tag_digest holds back a newer digest, that is when it is
// set and differs from the digest the desired ref names.
func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error) {
svc, err := s.store.GetService(ctx, serviceID)
if err != nil {
@@ -153,7 +158,7 @@ func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requested
if err != nil {
return job, err
}
if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest {
if d := driftedTagDigest(svc); d != "" && d != svc.LastTagDigest && holdsNothingNewer(svc) {
if err := s.store.SetLastTagDigest(ctx, svc.ID, d); err != nil {
s.log.Warn("last_tag_digest_not_set", "service_id", svc.ID, "error", err)
}
@@ -161,6 +166,13 @@ func (s *Service) RestoreDesired(ctx context.Context, serviceID int64, requested
return job, nil
}
// holdsNothingNewer reports whether last_tag_digest may be overwritten by a
// restore: it is empty or equals the digest the desired ref names. Anything
// else is a digest the poller must keep holding back.
func holdsNothingNewer(svc model.Service) bool {
return svc.LastTagDigest == "" || svc.LastTagDigest == imageref.Parse(svc.DesiredRef).Digest
}
// driftedTagDigest returns the digest a drifted service runs when it runs the
// same tag as desired at another digest, else "".
func driftedTagDigest(svc model.Service) string {
internal/master/model/model.go
+7
View File
@@ -15,6 +15,8 @@ import (
// Sentinel errors returned by storage.
var (
ErrNotFound = errors.New("not found")
// ErrConflict means the row already exists, or that it changed meanwhile
// (AcceptRunning, MergeService).
ErrConflict = errors.New("already exists")
ErrRevoked = errors.New("agent revoked")
)
@@ -166,6 +168,11 @@ func RunningRef(actualRef, actualDigest string) string {
return r.String()
}
// RunningRef returns the ref accepting the running version records as the
// desired one; see the function RunningRef. Templates post it as the guard of
// the accept form.
func (s Service) RunningRef() string { return RunningRef(s.ActualRef, s.ActualDigest) }
// Move says that the removed service FromID has probably moved to the live
// service ToID on another agent. The names are the agents' own; a revoked
// agent keeps its name.
internal/master/store/migrations/0005_drift_merge.sql
+2 -2
View File
@@ -1,7 +1,7 @@
-- true from a successful job until the next inventory write for the row: the
-- 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
-- Move suggestions look up Compose rows by project and service across agents.
CREATE INDEX services_by_compose_name ON services (compose_project, compose_service) WHERE kind = 'compose';
internal/master/store/services.go
+2 -2
View File
@@ -259,8 +259,8 @@ func (s *Store) SetPollError(ctx context.Context, ids []int64, msg string) error
}
// SetLastTagDigest sets last_tag_digest; see model.Service.LastTagDigest. The
// poller calls it, and FinishJob sets the same column after a tag job.
// Inventory writes never change it.
// poller and jobs.RestoreDesired call it, and FinishJob sets the same column
// after a tag job. Inventory writes never change it.
func (s *Store) SetLastTagDigest(ctx context.Context, id int64, digest string) error {
if _, err := s.pool.Exec(ctx,
`UPDATE services SET last_tag_digest = $2 WHERE id = $1`, id, digest); err != nil {
internal/master/web/drift_test.go
+15 -5
View File
@@ -118,7 +118,7 @@ func TestAcceptRunning_WritesRunningRefAndPublishes(t *testing.T) {
ch, cancel := broker.Subscribe()
defer cancel()
resp, _ := postTo(t, srv, "/services/1/accept", url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}})
resp, _ := postTo(t, srv, "/services/1/accept", url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}})
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
assert.Equal(t, "/services/1", resp.Header.Get("Location"))
@@ -139,7 +139,7 @@ func TestAcceptRunning_ReturnToOverview(t *testing.T) {
driftService(store)
resp, _ := postTo(t, srv, "/services/1/accept",
url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}, "return": {"overview"}})
url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}, "return": {"overview"}})
assert.Equal(t, http.StatusSeeOther, resp.StatusCode)
assert.Equal(t, "/", resp.Header.Get("Location"))
@@ -151,6 +151,7 @@ func TestAcceptRunning_Refusals(t *testing.T) {
path string
form url.Values
notDrifted bool
removed bool
acceptErr error
wantCode int
wantBody string
@@ -158,12 +159,17 @@ func TestAcceptRunning_Refusals(t *testing.T) {
{name: "missing actual", path: "/services/1/accept", form: url.Values{}, wantCode: http.StatusBadRequest},
{name: "stale actual", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.3.0"}},
wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
{name: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}},
{name: "same ref, other digest", path: "/services/1/accept",
form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftDesired}},
wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
{name: "not drifted", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
notDrifted: true, wantCode: http.StatusConflict, wantBody: "Cannot accept: the service runs its desired version."},
{name: "changed under our feet", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}},
{name: "changed under our feet", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
acceptErr: model.ErrConflict, wantCode: http.StatusConflict, wantBody: "The running version changed; check it again."},
{name: "unknown service", path: "/services/9/accept", form: url.Values{"actual": {"x"}}, wantCode: http.StatusNotFound},
{name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2"}},
{name: "removed", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
removed: true, wantCode: http.StatusNotFound},
{name: "removed meanwhile", path: "/services/1/accept", form: url.Values{"actual": {"gitea.shcizo.se/shcizo/slusk:1.4.2@" + driftActual}},
acceptErr: model.ErrNotFound, wantCode: http.StatusNotFound},
}
for _, tt := range tests {
@@ -172,6 +178,10 @@ func TestAcceptRunning_Refusals(t *testing.T) {
if !tt.notDrifted {
driftService(store)
}
if tt.removed {
removedAt := time.Now()
store.services[0].RemovedAt = &removedAt
}
store.acceptErr = tt.acceptErr
resp, body := postTo(t, srv, tt.path, tt.form)
internal/master/web/server.go
+11 -19
View File
@@ -349,13 +349,7 @@ func (s *Server) updateService(w http.ResponseWriter, r *http.Request) {
case err == nil || errors.Is(err, model.ErrJobActive):
http.Redirect(w, r, "/jobs/"+strconv.FormatInt(job.ID, 10), http.StatusSeeOther)
case errors.As(err, &nu):
data, derr := s.serviceDetail(r, id)
if derr != nil {
s.failLookup(w, derr)
return
}
data.Error = "Cannot update: " + nu.Reason + "."
s.render(w, http.StatusConflict, "service.html", "layout", data)
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot update: "+nu.Reason+".")
default:
s.failLookup(w, err)
}
@@ -381,8 +375,9 @@ func (s *Server) restoreDesired(w http.ResponseWriter, r *http.Request) {
}
// acceptRunning makes the running version of a drifted service the desired
// one. The form's actual field is the ref the user saw; it only guards against
// the service having changed since, the value written comes from the row.
// one. The form's actual field is the running ref the user saw (Service.RunningRef);
// it only guards against the service having changed since, the value written
// comes from the row.
func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) {
id, ok := pathID(w, r)
if !ok {
@@ -394,6 +389,9 @@ func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) {
return
}
svc, err := s.store.GetService(r.Context(), id)
if err == nil && svc.RemovedAt != nil {
err = model.ErrNotFound
}
if err != nil {
s.failLookup(w, err)
return
@@ -402,11 +400,11 @@ func (s *Server) acceptRunning(w http.ResponseWriter, r *http.Request) {
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot accept: the service runs its desired version.")
return
}
if actual != svc.ActualRef {
if actual != svc.RunningRef() {
s.renderServiceError(w, r, id, http.StatusConflict, "The running version changed; check it again.")
return
}
desired := model.RunningRef(svc.ActualRef, svc.ActualDigest)
desired := svc.RunningRef()
err = s.store.AcceptRunning(r.Context(), id, svc.ActualRef, svc.ActualDigest, desired)
switch {
case errors.Is(err, model.ErrConflict):
@@ -458,7 +456,7 @@ func (s *Server) mergeService(w http.ResponseWriter, r *http.Request) {
return
}
sess, _ := webauth.SessionFrom(r.Context())
s.log.Info("service_merged", "from", from, "to", id, "subject", sess.Subject)
s.log.Info("service_merged", "from_id", from, "to_id", id, "subject", sess.Subject)
s.broker.Publish(events.Services)
s.broker.Publish(events.Jobs)
http.Redirect(w, r, "/services/"+strconv.FormatInt(from, 10), http.StatusSeeOther)
@@ -498,13 +496,7 @@ func (s *Server) setMode(w http.ResponseWriter, r *http.Request) {
}
var nu *model.NotUpdatableError
if err := s.jobs.AutoAllowed(svc); errors.As(err, &nu) {
data, derr := s.serviceDetail(r, id)
if derr != nil {
s.failLookup(w, derr)
return
}
data.Error = "Cannot enable auto: " + nu.Reason + "."
s.render(w, http.StatusConflict, "service.html", "layout", data)
s.renderServiceError(w, r, id, http.StatusConflict, "Cannot enable auto: "+nu.Reason+".")
return
} else if err != nil {
s.fail(w, err)