From c6ea2d371089c5ff723f86de8213952eaf0e7e36 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 08:11:37 +0200 Subject: [PATCH] feat(master): registry digest lookup and polling loop (#9) --- go.mod | 9 +- go.sum | 24 +- internal/master/poller/poller.go | 270 ++++++++++++ internal/master/poller/poller_test.go | 487 ++++++++++++++++++++++ internal/master/registry/registry.go | 55 +++ internal/master/registry/registry_test.go | 251 +++++++++++ 6 files changed, 1088 insertions(+), 8 deletions(-) create mode 100644 internal/master/poller/poller.go create mode 100644 internal/master/poller/poller_test.go create mode 100644 internal/master/registry/registry.go create mode 100644 internal/master/registry/registry_test.go diff --git a/go.mod b/go.mod index 74e3aa7..1c2dfb7 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,7 @@ require ( github.com/coreos/go-oidc/v3 v3.21.0 github.com/distribution/reference v0.6.0 github.com/docker/docker v28.5.2+incompatible + github.com/google/go-containerregistry v0.22.1 github.com/jackc/pgx/v5 v5.10.0 github.com/opencontainers/go-digest v1.0.0 github.com/stretchr/testify v1.11.1 @@ -21,6 +22,8 @@ require ( github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/log v0.1.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect + github.com/docker/cli v29.7.2+incompatible // indirect + github.com/docker/docker-credential-helpers v0.9.3 // indirect github.com/docker/go-connections v0.7.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect @@ -30,6 +33,7 @@ require ( github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/klauspost/compress v1.19.2 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/moby/sys/atomicwriter v0.1.0 // indirect github.com/moby/term v0.5.2 // indirect @@ -37,14 +41,15 @@ require ( github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect go.opentelemetry.io/otel v1.43.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/otel/trace v1.43.0 // indirect - golang.org/x/sync v0.20.0 // indirect - golang.org/x/sys v0.42.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.35.0 // indirect golang.org/x/time v0.15.0 // indirect gotest.tools/v3 v3.5.2 // indirect diff --git a/go.sum b/go.sum index 7d4ff22..1ba542e 100644 --- a/go.sum +++ b/go.sum @@ -21,8 +21,12 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/cli v29.7.2+incompatible h1:dlkwallR8XqfeVnA2ELEhdwvb4lsSwuB4IgsG8Q9cLY= +github.com/docker/cli v29.7.2+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8= github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM= github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/docker-credential-helpers v0.9.3 h1:gAm/VtF9wgqJMoxzT3Gj5p4AqIjCBS4wrsOh9yRqcz8= +github.com/docker/docker-credential-helpers v0.9.3/go.mod h1:x+4Gbw9aGmChi3qTLZj8Dfn0TD20M/fuWy0E5+WDeCo= github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c= github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q= github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= @@ -38,6 +42,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/go-containerregistry v0.22.1 h1:RZuuSYhTvlDvtsK+NkutoCZ//C0X2ebLK8X8l3ULs84= +github.com/google/go-containerregistry v0.22.1/go.mod h1:bJR35SK8XgisYmhg/FMQ/5RK0S/XrOAqLBV5/LR2XE0= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= @@ -50,6 +56,8 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -74,8 +82,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= -github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= -github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= @@ -101,18 +109,22 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= +golang.org/x/mod v0.39.0 h1:UF5zwQdCRRUpHfyPwr7d4UrGiVeldIsogtzWVnczL74= +golang.org/x/mod v0.39.0/go.mod h1:bvIbwjQ0HUFFf5AKukeeYQG4ZBUG9yxQbR9aEweIwYY= golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= -golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= -golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= -golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/tools v0.49.0 h1:3NI7VXzL9+1WZD52Dx2ttoPwD5DWrFGpl9mFZDlmisI= +golang.org/x/tools v0.49.0/go.mod h1:SJNXV9DBKT0UbdttsQjbfJlAE/q+y36++zo3uL3N0Oo= google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M= google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= diff --git a/internal/master/poller/poller.go b/internal/master/poller/poller.go new file mode 100644 index 0000000..83fb16c --- /dev/null +++ b/internal/master/poller/poller.go @@ -0,0 +1,270 @@ +// Package poller checks the registry for new digests of the tags the +// services follow. A new digest becomes the service's available version, and +// a service in auto mode gets an update job. +package poller + +import ( + "context" + "errors" + "log/slog" + "strings" + "sync" + "time" + + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" +) + +// lookupTimeout bounds one registry lookup. +const lookupTimeout = 30 * time.Second + +// maxErrorBytes caps the registry error stored in services.last_poll_error. +const maxErrorBytes = 500 + +// Store is the storage the poller needs. *store.Store satisfies it. +type Store interface { + ListServices(ctx context.Context) ([]model.Service, error) + SetAvailable(ctx context.Context, id int64, a model.Available) error + SetPollError(ctx context.Context, ids []int64, msg string) error +} + +// Jobs starts updates. *jobs.Service satisfies it. +type Jobs interface { + Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) +} + +// Registry looks up the digest a "repo:tag" ref points at. *registry.Client +// satisfies it. +type Registry interface { + Digest(ctx context.Context, ref string) (string, error) +} + +// Publisher receives change signals. *events.Broker satisfies it. +type Publisher interface{ Publish(t events.Topic) } + +// Config holds the poller settings. +type Config struct { + Interval time.Duration // time between the end of a cycle and the start of the next (POLL_INTERVAL) + FirstDelay time.Duration // wait before the first cycle; 1 minute in production + Workers int // registry lookups in flight at once; 4 in production +} + +// Poller runs polling cycles. +type Poller struct { + store Store + jobs Jobs + reg Registry + pub Publisher + cfg Config + log *slog.Logger + now func() time.Time +} + +// New returns a Poller. +func New(st Store, j Jobs, reg Registry, pub Publisher, cfg Config, log *slog.Logger) *Poller { + if cfg.Workers < 1 { + cfg.Workers = 1 + } + return &Poller{store: st, jobs: j, reg: reg, pub: pub, cfg: cfg, log: log, now: time.Now} +} + +// SkipReason says why svc is not polled: "follows no tag", "built from +// source", "no registry digest", or "" when it is polled. +func SkipReason(svc model.Service) string { + switch { + case jobs.FollowedRef(svc.ActualRef) == "": + return "follows no tag" + case svc.Built: + return "built from source" + case svc.ActualDigest == "": + return "no registry digest" + } + return "" +} + +// Run polls until ctx is done. The first cycle starts Config.FirstDelay after +// the call and each following one Config.Interval after the previous cycle +// ended, so cycles never overlap. It blocks. +func (p *Poller) Run(ctx context.Context) { + timer := time.NewTimer(p.cfg.FirstDelay) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + } + p.Cycle(ctx) + timer.Reset(p.cfg.Interval) + } +} + +// lookup is the registry's answer for one followed ref. +type lookup struct { + ref string + digest string + err error +} + +// cycleStats is what one cycle did, for the poll_cycle log line. +type cycleStats struct { + refs, ok, failed, suggested, jobs, skipped int + wrote bool +} + +// Cycle runs one poll over all services. It looks up every distinct followed +// ref once, with at most Config.Workers lookups at a time, then records the +// results. Writing results and starting jobs is not cut short by ctx ending +// during the lookups, so a job that was marked is always started. +func (p *Poller) Cycle(ctx context.Context) { + began := p.now() + svcs, err := p.store.ListServices(ctx) + if err != nil { + p.log.Error("poll_list_failed", "error", err) + return + } + var stats cycleStats + byRef := map[string][]model.Service{} + var refs []string + for _, svc := range svcs { + if svc.RemovedAt != nil { + continue + } + if SkipReason(svc) != "" { + stats.skipped++ + continue + } + ref := jobs.FollowedRef(svc.ActualRef) + if _, seen := byRef[ref]; !seen { + refs = append(refs, ref) + } + byRef[ref] = append(byRef[ref], svc) + } + stats.refs = len(refs) + + write := context.WithoutCancel(ctx) + for _, l := range p.lookupAll(ctx, refs) { + if l.err != nil && ctx.Err() != nil { + continue // cancelled mid-lookup: not the registry's fault + } + p.record(write, l, byRef[l.ref], &stats) + } + if stats.wrote { + p.pub.Publish(events.Services) + } + p.log.Info("poll_cycle", "refs", stats.refs, "ok", stats.ok, "failed", stats.failed, + "suggested", stats.suggested, "jobs", stats.jobs, "skipped", stats.skipped, + "duration_ms", p.now().Sub(began).Milliseconds()) +} + +// lookupAll asks the registry for every ref, Config.Workers at a time. It +// returns the answers in ref order and stops starting new lookups once ctx +// is done; refs never looked up are missing from the result. +func (p *Poller) lookupAll(ctx context.Context, refs []string) []lookup { + results := make([]*lookup, len(refs)) + next := make(chan int) + var wg sync.WaitGroup + for w := 0; w < p.cfg.Workers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := range next { + lctx, cancel := context.WithTimeout(ctx, lookupTimeout) + d, err := p.reg.Digest(lctx, refs[i]) + cancel() + results[i] = &lookup{ref: refs[i], digest: d, err: err} + } + }() + } +feed: + for i := range refs { + select { + case next <- i: + case <-ctx.Done(): + break feed + } + } + close(next) + wg.Wait() + var out []lookup + for _, r := range results { + if r != nil { + out = append(out, *r) + } + } + return out +} + +// record stores one lookup result for the services that follow l.ref. +func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, stats *cycleStats) { + if l.err != nil { + stats.failed++ + msg := truncate(l.err.Error()) + p.log.Warn("poll_failed", "ref", l.ref, "error", msg) + p.setPollError(ctx, svcs, msg, stats, func(s model.Service) bool { return s.LastPollError != msg }) + return + } + 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) { + 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 + } + 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}) + var notUpdatable *model.NotUpdatableError + switch { + case err == nil: + stats.jobs++ + case errors.Is(err, model.ErrJobActive), errors.As(err, ¬Updatable): + p.log.Debug("poll_job_not_started", "service_id", svc.ID, "ref", ref, "reason", err.Error()) + default: + p.log.Warn("poll_job_failed", "service_id", svc.ID, "ref", ref, "error", err) + } +} + +// setPollError sets msg as last_poll_error on the services for which +// changed returns true; with none, it writes nothing. +func (p *Poller) setPollError(ctx context.Context, svcs []model.Service, msg string, stats *cycleStats, changed func(model.Service) bool) { + var ids []int64 + for _, svc := range svcs { + if changed(svc) { + ids = append(ids, svc.ID) + } + } + if len(ids) == 0 { + return + } + if err := p.store.SetPollError(ctx, ids, msg); err != nil { + p.log.Error("poll_set_error_failed", "error", err) + return + } + stats.wrote = true +} + +// truncate cuts msg to maxErrorBytes without leaving a broken UTF-8 +// sequence, and drops NUL bytes, which Postgres text cannot hold. +func truncate(msg string) string { + msg = strings.ReplaceAll(msg, "\x00", "") + if len(msg) > maxErrorBytes { + msg = strings.ToValidUTF8(msg[:maxErrorBytes], "") + } + return msg +} diff --git a/internal/master/poller/poller_test.go b/internal/master/poller/poller_test.go new file mode 100644 index 0000000..23a4597 --- /dev/null +++ b/internal/master/poller/poller_test.go @@ -0,0 +1,487 @@ +package poller + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/master/events" + "github.com/shcizo/package-updater/internal/master/jobs" + "github.com/shcizo/package-updater/internal/master/model" +) + +type availableCall struct { + id int64 + a model.Available +} + +type pollErrorCall struct { + ids []int64 + msg string +} + +type fakeStore struct { + mu sync.Mutex + services []model.Service + listErr error + available []availableCall + pollErrors []pollErrorCall + ctxErrs []error // ctx.Err() seen by every write +} + +func (f *fakeStore) ListServices(context.Context) ([]model.Service, error) { + return f.services, f.listErr +} + +func (f *fakeStore) SetAvailable(ctx context.Context, id int64, a model.Available) error { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.available = append(f.available, availableCall{id, a}) + return nil +} + +func (f *fakeStore) SetPollError(ctx context.Context, ids []int64, msg string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.pollErrors = append(f.pollErrors, pollErrorCall{ids, msg}) + return nil +} + +type startCall struct { + svcID int64 + req jobs.StartRequest +} + +type fakeJobs struct { + mu sync.Mutex + starts []startCall + errFor map[int64]error // by service ID + ctxErrs []error +} + +func (f *fakeJobs) Start(ctx context.Context, svc model.Service, req jobs.StartRequest) (model.Job, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.ctxErrs = append(f.ctxErrs, ctx.Err()) + f.starts = append(f.starts, startCall{svc.ID, req}) + return model.Job{ID: int64(len(f.starts))}, f.errFor[svc.ID] +} + +type fakeRegistry struct { + mu sync.Mutex + digests map[string]string // ref -> digest + errs map[string]error // ref -> error + calls []string + inFlight int + maxSeen int + delay time.Duration + started chan struct{} // receives a value whenever a lookup starts, if set + release chan struct{} // lookups wait for it to be closed, if set + onResult func() // called after a lookup computed its answer +} + +func (f *fakeRegistry) Digest(ctx context.Context, ref string) (string, error) { + f.mu.Lock() + f.calls = append(f.calls, ref) + f.inFlight++ + if f.inFlight > f.maxSeen { + f.maxSeen = f.inFlight + } + f.mu.Unlock() + defer func() { + f.mu.Lock() + f.inFlight-- + f.mu.Unlock() + }() + + if f.started != nil { + f.started <- struct{}{} + } + if f.release != nil { + select { + case <-f.release: + case <-ctx.Done(): + return "", ctx.Err() + } + } + if f.delay > 0 { + time.Sleep(f.delay) + } + if f.onResult != nil { + f.onResult() + } + if err := f.errs[ref]; err != nil { + return "", err + } + return f.digests[ref], nil +} + +func (f *fakeRegistry) sortedCalls() []string { + f.mu.Lock() + defer f.mu.Unlock() + out := append([]string(nil), f.calls...) + sort.Strings(out) + return out +} + +type fakePub struct { + mu sync.Mutex + topics []events.Topic +} + +func (p *fakePub) Publish(t events.Topic) { + p.mu.Lock() + defer p.mu.Unlock() + p.topics = append(p.topics, t) +} + +const ( + repoA = "gitea.shcizo.se/shcizo/slusk" + repoB = "gitea/gitea" +) + +// svc builds a polled service whose running version has digest sha256:old. +func svc(id int64, actualRef string) model.Service { + return model.Service{ID: id, Kind: "compose", ActualRef: actualRef, ActualDigest: "sha256:old", UpdateMode: model.ModeManual} +} + +func newPoller(st *fakeStore, j *fakeJobs, reg *fakeRegistry, cfg Config) (*Poller, *fakePub) { + pub := &fakePub{} + if cfg.Workers == 0 { + cfg.Workers = 4 + } + p := New(st, j, reg, pub, cfg, slog.New(slog.NewTextHandler(io.Discard, nil))) + p.now = func() time.Time { return time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) } + return p, pub +} + +func TestSkipReason(t *testing.T) { + tests := []struct { + name string + svc model.Service + want string + }{ + {"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"}, + {"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"}, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, SkipReason(tt.svc)) + }) + } +} + +func TestOneHeadPerDistinctRef(t *testing.T) { + st := &fakeStore{services: []model.Service{ + svc(1, repoA+":1.4"), + svc(2, repoA+":1.4@sha256:old"), + svc(3, repoB), + svc(4, repoB+":latest"), + svc(5, repoB+":1.27"), + }} + reg := &fakeRegistry{digests: map[string]string{}} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []string{repoA + ":1.4", repoB + ":1.27", repoB + ":latest"}, reg.sortedCalls()) +} + +func TestSkipped(t *testing.T) { + removed := svc(4, "r:1") + removed.RemovedAt = ptr(time.Now()) + st := &fakeStore{services: []model.Service{ + {ID: 1, ActualRef: "r@sha256:a", ActualDigest: "sha256:a"}, + {ID: 2, ActualRef: "r:1", ActualDigest: "sha256:a", Built: true}, + {ID: 3, ActualRef: "r:1"}, + removed, + }} + reg := &fakeRegistry{} + j := &fakeJobs{} + p, pub := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, reg.calls, "no lookups") + assert.Empty(t, st.available) + assert.Empty(t, st.pollErrors) + assert.Empty(t, j.starts) + assert.Empty(t, pub.topics) +} + +func TestNewDigest(t *testing.T) { + ref := repoA + ":1.4" + tests := []struct { + name string + mode string + wantStart bool + }{ + {"auto gets one poll job to repo:tag", model.ModeAuto, true}, + {"manual is only marked", model.ModeManual, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + s := svc(1, ref+"@sha256:old") + s.UpdateMode = tt.mode + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, pub := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.available, 1) + assert.Equal(t, int64(1), st.available[0].id) + assert.Equal(t, model.Available{Ref: ref, Digest: "sha256:new", Source: model.SourcePoll, + 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) + } else { + assert.Empty(t, j.starts) + } + assert.Equal(t, []events.Topic{events.Services}, pub.topics) + }) + } +} + +func TestSameSuggestionIsIdempotent(t *testing.T) { + ref := repoA + ":1.4" + running := svc(1, ref+"@sha256:old") + running.UpdateMode = model.ModeAuto + running.ActualDigest = "sha256:new" // already on the registry's digest + suggested := svc(2, ref) + suggested.UpdateMode = model.ModeAuto + suggested.AvailableRef, suggested.AvailableDigest = ref, "sha256:new" + staleSuggestion := svc(3, ref) + staleSuggestion.AvailableRef, staleSuggestion.AvailableDigest = ref, "sha256:older" + st := &fakeStore{services: []model.Service{running, suggested, staleSuggestion}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.available, 1, "only the service with an older suggestion is updated") + assert.Equal(t, int64(3), st.available[0].id) + assert.Empty(t, j.starts, "no new job for an already suggested digest") +} + +func TestStartErrorsDoNotAbortCycle(t *testing.T) { + ref := repoA + ":1.4" + var svcs []model.Service + for id := int64(1); id <= 3; id++ { + s := svc(id, ref) + s.UpdateMode = model.ModeAuto + svcs = append(svcs, s) + } + st := &fakeStore{services: svcs} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{errFor: map[int64]error{ + 1: model.ErrJobActive, + 2: &model.NotUpdatableError{Reason: model.ReasonSelf}, + 3: errors.New("database down"), + }} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Len(t, st.available, 3) + assert.Len(t, j.starts, 3) +} + +func TestErrorIsPerRef(t *testing.T) { + bad, good := repoA+":1.4", repoB+":1.27" + badOld := svc(1, bad) + badSame := svc(2, bad) + badSame.LastPollError = "head: 401 unauthorized" + goodSvc := svc(3, good) + st := &fakeStore{services: []model.Service{badOld, badSame, goodSvc}} + reg := &fakeRegistry{ + errs: map[string]error{bad: errors.New("head: 401 unauthorized")}, + digests: map[string]string{good: "sha256:new"}, + } + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []pollErrorCall{{[]int64{1}, "head: 401 unauthorized"}}, st.pollErrors, + "only the service whose stored error differs is written") + require.Len(t, st.available, 1, "the other ref is still processed") + assert.Equal(t, int64(3), st.available[0].id) + assert.Equal(t, []events.Topic{events.Services}, pub.topics) +} + +func TestErrorTextIsTruncated(t *testing.T) { + ref := repoA + ":1.4" + st := &fakeStore{services: []model.Service{svc(1, ref)}} + reg := &fakeRegistry{errs: map[string]error{ref: errors.New(strings.Repeat("é", 400) + "\x00")}} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, st.pollErrors, 1) + msg := st.pollErrors[0].msg + assert.LessOrEqual(t, len(msg), maxErrorBytes) + assert.Equal(t, strings.Repeat("é", 250), msg, "cut on a character boundary, NUL removed") +} + +func TestErrorClearedOnSuccess(t *testing.T) { + ref := repoA + ":1.4" + failing := svc(1, ref) + failing.LastPollError = "head: boom" + fine := svc(2, ref) + failing.ActualDigest, fine.ActualDigest = "sha256:new", "sha256:new" + st := &fakeStore{services: []model.Service{failing, fine}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Equal(t, []pollErrorCall{{[]int64{1}, ""}}, st.pollErrors) + assert.Empty(t, st.available) + assert.Equal(t, []events.Topic{events.Services}, pub.topics) +} + +func TestNothingWrittenNothingPublished(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.ActualDigest = "sha256:new" + st := &fakeStore{services: []model.Service{s}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, st.pollErrors) + assert.Empty(t, st.available) + assert.Empty(t, pub.topics) +} + +func TestListErrorEndsTheCycle(t *testing.T) { + st := &fakeStore{listErr: errors.New("db down")} + reg := &fakeRegistry{} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{}) + + p.Cycle(context.Background()) + + assert.Empty(t, reg.calls) +} + +func TestConcurrencyLimit(t *testing.T) { + 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}) + + p.Cycle(context.Background()) + + assert.Len(t, reg.calls, 8) + assert.Equal(t, 3, reg.maxSeen) +} + +func TestCycleStopsOnCancel(t *testing.T) { + var svcs []model.Service + for i := 0; i < 6; i++ { + svcs = append(svcs, svc(int64(i+1), fmt.Sprintf("%s%d:1", repoA, i))) + } + st := &fakeStore{services: svcs} + reg := &fakeRegistry{release: make(chan struct{}), started: make(chan struct{}, 6)} + p, pub := newPoller(st, &fakeJobs{}, reg, Config{Workers: 2}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { p.Cycle(ctx); close(done) }() + <-reg.started + <-reg.started + + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Cycle did not return after cancel") + } + assert.Len(t, reg.calls, 2, "no new lookups after cancel") + assert.Empty(t, st.pollErrors, "cancelled lookups are not recorded as registry errors") + assert.Empty(t, st.available) + assert.Empty(t, pub.topics) +} + +func TestMarkedJobsAreStartedEvenIfCancelledDuringTheCycle(t *testing.T) { + ref := repoA + ":1.4" + s := svc(1, ref) + s.UpdateMode = model.ModeAuto + st := &fakeStore{services: []model.Service{s}} + ctx, cancel := context.WithCancel(context.Background()) + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}, onResult: cancel} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(ctx) + + require.Len(t, st.available, 1) + require.Len(t, j.starts, 1) + assert.NoError(t, st.ctxErrs[0], "writes do not see the cancelled context") + assert.NoError(t, j.ctxErrs[0]) +} + +func TestRun(t *testing.T) { + ref := repoA + ":1.4" + st := &fakeStore{services: []model.Service{svc(1, ref)}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:old"}, delay: 30 * time.Millisecond} + p, _ := newPoller(st, &fakeJobs{}, reg, Config{Interval: 5 * time.Millisecond, FirstDelay: 60 * time.Millisecond, Workers: 1}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + began := time.Now() + go func() { p.Run(ctx); close(done) }() + + require.Eventually(t, func() bool { + reg.mu.Lock() + defer reg.mu.Unlock() + return len(reg.calls) >= 3 + }, 5*time.Second, 5*time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } + assert.GreaterOrEqual(t, time.Since(began), 60*time.Millisecond+2*30*time.Millisecond, "first cycle waits FirstDelay") + assert.Equal(t, 1, reg.maxSeen, "cycles never overlap") +} + +func TestRunReturnsDuringFirstDelay(t *testing.T) { + p, _ := newPoller(&fakeStore{}, &fakeJobs{}, &fakeRegistry{}, Config{Interval: time.Hour, FirstDelay: time.Hour}) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { p.Run(ctx); close(done) }() + + cancel() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Run did not return after cancel") + } +} + +func ptr[T any](v T) *T { return &v } diff --git a/internal/master/registry/registry.go b/internal/master/registry/registry.go new file mode 100644 index 0000000..69a4dd6 --- /dev/null +++ b/internal/master/registry/registry.go @@ -0,0 +1,55 @@ +// Package registry looks up the digest a tag points at in a container +// registry, using HEAD requests only. +package registry + +import ( + "context" + "fmt" + + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" + "github.com/google/go-containerregistry/pkg/v1/remote" + + "github.com/shcizo/package-updater/internal/imageref" +) + +// Client looks up tag digests with HEAD requests only. Docker Hub counts a +// GET of a manifest as a pull against the rate limit and a HEAD as nothing, +// so Client never issues a manifest GET. +type Client struct { + keychain authn.Keychain + userAgent string +} + +// New returns a Client that finds credentials the way the Docker CLI does: +// in $DOCKER_CONFIG/config.json, else ~/.docker/config.json. Registries +// without an entry are accessed anonymously. userAgent is added to the +// User-Agent header of every request. +func New(userAgent string) *Client { + return &Client{keychain: authn.DefaultKeychain, userAgent: userAgent} +} + +// Digest returns the digest that "repo:tag" points at, as the registry +// reports it: for a multi-arch image that is the digest of the index, the +// same one `docker inspect` shows in RepoDigests. A digest suffix in ref is +// ignored. It sends HEAD /v2//manifests/ and nothing else to the +// manifest endpoint; it does not fall back to GET when the registry omits +// the digest header. +func (c *Client) Digest(ctx context.Context, ref string) (string, error) { + p := imageref.Parse(ref) + if p.Tag == "" { + return "", fmt.Errorf("image %q has no tag", ref) + } + r, err := name.ParseReference(p.Repo + ":" + p.Tag) + if err != nil { + return "", fmt.Errorf("parse %s: %w", ref, err) + } + desc, err := remote.Head(r, + remote.WithContext(ctx), + remote.WithAuthFromKeychain(c.keychain), + remote.WithUserAgent(c.userAgent)) + if err != nil { + return "", fmt.Errorf("head %s: %w", r, err) + } + return desc.Digest.String(), nil +} diff --git a/internal/master/registry/registry_test.go b/internal/master/registry/registry_test.go new file mode 100644 index 0000000..888de7a --- /dev/null +++ b/internal/master/registry/registry_test.go @@ -0,0 +1,251 @@ +package registry + +import ( + "context" + "encoding/base64" + "io" + "log" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/go-containerregistry/pkg/name" + ggcrregistry "github.com/google/go-containerregistry/pkg/registry" + "github.com/google/go-containerregistry/pkg/v1/random" + "github.com/google/go-containerregistry/pkg/v1/remote" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// request is one request the test registry received. +type request struct{ method, path string } + +// testRegistry is an in-memory registry that records every request. +type testRegistry struct { + srv *httptest.Server + host string // "127.0.0.1:port", the registry part of a ref + + mu sync.Mutex + requests []request +} + +// newTestRegistry starts a registry. When user is set every request needs +// those basic-auth credentials. +func newTestRegistry(t *testing.T, user, pass string) *testRegistry { + t.Helper() + backend := ggcrregistry.New(ggcrregistry.Logger(log.New(io.Discard, "", 0))) + tr := &testRegistry{} + tr.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + tr.mu.Lock() + tr.requests = append(tr.requests, request{r.Method, r.URL.Path}) + tr.mu.Unlock() + if user != "" { + if u, p, ok := r.BasicAuth(); !ok || u != user || p != pass { + w.Header().Set("WWW-Authenticate", `Basic realm="test"`) + w.WriteHeader(http.StatusUnauthorized) + return + } + } + backend.ServeHTTP(w, r) + })) + t.Cleanup(tr.srv.Close) + tr.host = strings.TrimPrefix(tr.srv.URL, "http://") + return tr +} + +// pushIndex stores a random multi-arch index as repo:tag and returns its +// digest. user and pass are used when the registry requires them. +func (tr *testRegistry) pushIndex(t *testing.T, repo, tag string, opts ...remote.Option) string { + t.Helper() + idx, err := random.Index(64, 1, 2) + require.NoError(t, err) + ref, err := name.ParseReference(tr.host + "/" + repo + ":" + tag) + require.NoError(t, err) + require.NoError(t, remote.WriteIndex(ref, idx, opts...)) + d, err := idx.Digest() + require.NoError(t, err) + return d.String() +} + +// manifestRequests returns the requests that touched a manifest. +func (tr *testRegistry) manifestRequests() []request { + tr.mu.Lock() + defer tr.mu.Unlock() + var out []request + for _, r := range tr.requests { + if strings.Contains(r.path, "/manifests/") { + out = append(out, r) + } + } + return out +} + +// reset forgets the requests received so far. +func (tr *testRegistry) reset() { + tr.mu.Lock() + defer tr.mu.Unlock() + tr.requests = nil +} + +// dockerConfig points DOCKER_CONFIG at a fresh directory holding config.json +// with the given content; "" leaves the directory empty. +func dockerConfig(t *testing.T, configJSON string) { + t.Helper() + dir := t.TempDir() + if configJSON != "" { + require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(configJSON), 0o600)) + } + t.Setenv("DOCKER_CONFIG", dir) +} + +func TestDigestOfIndexIsIndexDigest(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + want := reg.pushIndex(t, "team/app", "1.2") + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestDigestIgnoresDigestSuffix(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + want := reg.pushIndex(t, "team/app", "1.2") + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2@sha256:"+strings.Repeat("0", 64)) + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestDigestUsesHeadOnly(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + reg.reset() + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, []request{{http.MethodHead, "/v2/team/app/manifests/1.2"}}, reg.manifestRequests()) + for _, r := range reg.requests { + if !strings.Contains(r.path, "/manifests/") { + assert.Equal(t, request{http.MethodGet, "/v2/"}, r, "the only other request is the API ping") + } + } +} + +func TestDigestSendsUserAgent(t *testing.T) { + dockerConfig(t, "") + var ua string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ua = r.Header.Get("User-Agent") + w.WriteHeader(http.StatusNotFound) + })) + t.Cleanup(srv.Close) + + _, _ = New("package-updater-test/1").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1") + + assert.Contains(t, ua, "package-updater-test/1") +} + +func TestCredentialsFromDockerConfig(t *testing.T) { + reg := newTestRegistry(t, "bot", "s3cret") + want := reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"})) + auth := base64.StdEncoding.EncodeToString([]byte("bot:s3cret")) + dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`) + + got, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestWrongCredentialsFail(t *testing.T) { + reg := newTestRegistry(t, "bot", "s3cret") + reg.pushIndex(t, "team/app", "1.2", remote.WithAuth(&authn.Basic{Username: "bot", Password: "s3cret"})) + auth := base64.StdEncoding.EncodeToString([]byte("bot:wrong")) + dockerConfig(t, `{"auths":{"`+reg.host+`":{"auth":"`+auth+`"}}}`) + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + require.Error(t, err) + assert.NotContains(t, err.Error(), "wrong", "credentials do not leak into errors") + assert.NotContains(t, err.Error(), auth) +} + +func TestAnonymousWithoutConfig(t *testing.T) { + dockerConfig(t, "") + open := newTestRegistry(t, "", "") + want := open.pushIndex(t, "team/app", "1.2") + protected := newTestRegistry(t, "bot", "s3cret") + + got, err := New("test").Digest(context.Background(), open.host+"/team/app:1.2") + require.NoError(t, err) + assert.Equal(t, want, got, "anonymous access works without a config file") + + _, err = New("test").Digest(context.Background(), protected.host+"/team/app:1.2") + assert.Error(t, err, "a registry that needs credentials fails without them") +} + +func TestUnknownTagIsAnError(t *testing.T) { + dockerConfig(t, "") + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:nope") + + assert.Error(t, err) +} + +func TestRefWithoutTagIsAnError(t *testing.T) { + _, err := New("test").Digest(context.Background(), "team/app@sha256:"+strings.Repeat("a", 64)) + + assert.ErrorContains(t, err, "has no tag") +} + +// A config.json whose credsStore names a helper that is not installed makes +// the lookup fail for every registry, including ones that would work +// anonymously. The master container has no credential helpers, so only +// inline "auths" entries are supported there. +func TestCredsStoreWithoutHelperFailsTheLookup(t *testing.T) { + reg := newTestRegistry(t, "", "") + reg.pushIndex(t, "team/app", "1.2") + dockerConfig(t, `{"credsStore":"no-such-helper-for-package-updater-tests"}`) + + _, err := New("test").Digest(context.Background(), reg.host+"/team/app:1.2") + + assert.ErrorContains(t, err, "error getting credentials") + assert.ErrorContains(t, err, "docker-credential-no-such-helper-for-package-updater-tests") +} + +// A registry that leaves out Docker-Content-Digest must not make the client +// retry with a manifest download, which Docker Hub counts as a pull. +func TestMissingDigestHeaderDoesNotFallBackToDownload(t *testing.T) { + dockerConfig(t, "") + var mu sync.Mutex + var methods []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.Contains(r.URL.Path, "/manifests/") { + return + } + mu.Lock() + methods = append(methods, r.Method) + mu.Unlock() + w.Header().Set("Content-Type", "application/vnd.oci.image.index.v1+json") + w.Header().Set("Content-Length", "10") + })) + t.Cleanup(srv.Close) + + _, err := New("test").Digest(context.Background(), strings.TrimPrefix(srv.URL, "http://")+"/a/b:1") + + assert.Error(t, err) + assert.Equal(t, []string{http.MethodHead}, methods) +}