From 0d56c068e9f5d4f70aacac471688e9194a28b7ef Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 15:07:53 +0200 Subject: [PATCH 1/7] feat(provider): pull command and get-image protocol for image distribution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Providers opt into image distribution by declaring a pull block in their metadata — the presence of the command block is the declaration of support, generalizing the stop precedent. During its pull command a provider can request the service image from the local daemon with a get-image message (optionally platform-narrowed); compose answers on stdin with one image-stream JSON line then the tar as HTTP/1.1 chunked data (RFC 9112): length-prefixed blocks need no in-band delimiter in binary data, the zero-length chunk marks a COMPLETE transfer, and any stock chunked reader consumes it. There is deliberately no trailer nor final CRLF after the zero chunk — a stock reader stops there without consuming further bytes, so the next stdin answer starts clean for a provider requesting several images. An export failure is announced in-band through the error field so the provider is never left waiting; a failure after the announce closes the answer channel so the truncation is observable as EOF mid-chunk instead of a stream nobody will finish. The stream is exclusive on stdin for its whole duration — the framing invariant behind the lock. get-image is scoped to the pull command: emitted during up, down or stop it is a protocol error rather than served, so an image export can never stall another lifecycle command. Message dispatch moves out of executePlugin into handlePluginMessage, which also keeps the function under the complexity threshold. Signed-off-by: Nicolas De Loof --- pkg/compose/plugins.go | 146 ++++++++++++++++++++++++++++++++++-- pkg/compose/plugins_test.go | 19 +++++ 2 files changed, 158 insertions(+), 7 deletions(-) diff --git a/pkg/compose/plugins.go b/pkg/compose/plugins.go index 65c38c52e74..d82f455c993 100644 --- a/pkg/compose/plugins.go +++ b/pkg/compose/plugins.go @@ -24,6 +24,7 @@ import ( "fmt" "io" "net" + "net/http/httputil" "os" "os/exec" "path/filepath" @@ -33,8 +34,10 @@ import ( "github.com/compose-spec/compose-go/v2/types" "github.com/containerd/errdefs" + "github.com/containerd/platforms" "github.com/docker/cli/cli-plugins/manager" "github.com/docker/cli/cli/config" + "github.com/moby/moby/client" "github.com/sirupsen/logrus" "github.com/spf13/cobra" @@ -44,6 +47,9 @@ import ( type JsonMessage struct { Type string `json:"type"` Message string `json:"message,omitempty"` + // Platform optionally narrows a get-image request to one platform of a + // multi-platform image (e.g. "linux/arm64"). + Platform string `json:"platform,omitempty"` } const ( @@ -70,6 +76,17 @@ const ( // resource has no use for it. GetRelayInfoType = "get-relay-info" + // GetImageType is a message the provider sends during its pull command + // to receive the service image from the local daemon: compose answers on + // the provider's stdin with one ImageStreamType JSON line, then — unless + // that line carries an error — the image tar encoded as an HTTP/1.1 + // chunked body (see streamImageTo). + GetImageType = "get-image" + + // ImageStreamType is the type of the JSON line answering a get-image + // request. + ImageStreamType = "image-stream" + // ComposeProviderMessagesEnv announces to the provider process, as a // comma-separated list, every message type this compose accepts on the // provider's stdout — so a provider can adapt to the compose it runs @@ -91,8 +108,19 @@ var providerMessageTypes = strings.Join([]string{ PublishEndpointType, GetServiceConfigType, GetRelayInfoType, + GetImageType, }, ",") +// imageStreamAnswer is the stdin answer to a get-image request. On success +// Encoding and MediaType describe the byte stream that follows the JSON line; +// on failure Error carries the reason and no stream follows. +type imageStreamAnswer struct { + Type string `json:"type"` + Encoding string `json:"encoding,omitempty"` + MediaType string `json:"media-type,omitempty"` + Error string `json:"error,omitempty"` +} + type pluginVariables struct { prefixed types.Mapping raw types.Mapping @@ -105,7 +133,7 @@ type pluginVariables struct { var mux sync.Mutex -func (s *composeService) runPlugin(ctx context.Context, project *types.Project, service types.ServiceConfig, command string) error { +func (s *composeService) runPlugin(ctx context.Context, project *types.Project, service types.ServiceConfig, command string, extraArgs ...string) error { provider := *service.Provider plugin, err := s.getPluginBinaryPath(provider.Type) @@ -113,7 +141,7 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project, return err } - cmd, err := s.setupPluginCommand(ctx, project, service, plugin, command) + cmd, err := s.setupPluginCommand(ctx, project, service, plugin, command, extraArgs...) if err != nil { return err } @@ -126,7 +154,7 @@ func (s *composeService) runPlugin(ctx context.Context, project *types.Project, return err } - if command == "stop" { + if command == "stop" || command == "pull" { return nil } @@ -203,6 +231,9 @@ func (s *composeService) executePlugin(ctx context.Context, project *types.Proje case "stop": s.events.On(stoppingEvent(service.Name)) action = "stop" + case "pull": + s.events.On(newEvent(service.Name, api.Working, "Pulling")) + action = "pull" default: return pluginVariables{}, fmt.Errorf("unsupported plugin command: %s", command) } @@ -275,7 +306,7 @@ func (s *composeService) executePlugin(ctx context.Context, project *types.Proje if err != nil { return pluginVariables{}, err } - if err := s.handlePluginMessage(ctx, project, service, msg, &variables, stdin, &stdinMu, &answers); err != nil { + if err := s.handlePluginMessage(ctx, project, service, msg, command, &variables, stdin, &stdinMu, &answers); err != nil { return pluginVariables{}, err } } @@ -293,17 +324,77 @@ func (s *composeService) executePlugin(ctx context.Context, project *types.Proje s.events.On(removedEvent(service.Name)) case "stop": s.events.On(stoppedEvent(service.Name)) + case "pull": + s.events.On(newEvent(service.Name, api.Done, "Pulled")) } return variables, nil } +// imageStreamChunkSize is the fixed buffer compose fills from the image +// export before flushing it as one chunk of the stream. +const imageStreamChunkSize = 1024 * 1024 + +// streamImageTo answers a get-image request on the provider's stdin: one +// ImageStreamType JSON line announcing the transfer, then the image tar +// encoded as HTTP/1.1 chunked data (RFC 9112 §7.1) — every block of data +// prefixed by its length, the zero-length chunk marking a COMPLETE +// transfer. Length-prefixed framing needs no in-band delimiter (any +// byte value can appear inside a tar) and is consumable with any language's +// chunked-body reader. A stream that ends without the terminating zero chunk +// was aborted: the provider must discard it. +// When the image cannot be exported, the announce line carries an error +// instead and no stream follows. +func (s *composeService) streamImageTo(ctx context.Context, w io.Writer, ref, platform string) error { + announce := func(a imageStreamAnswer) error { + payload, err := json.Marshal(a) + if err != nil { + return err + } + _, err = w.Write(append(payload, '\n')) + return err + } + + var opts []client.ImageSaveOption + if platform != "" { + p, err := platforms.Parse(platform) + if err != nil { + _ = announce(imageStreamAnswer{Type: ImageStreamType, Error: fmt.Sprintf("invalid platform %q: %s", platform, err)}) + return err + } + opts = append(opts, client.ImageSaveWithPlatforms(p)) + } + tar, err := s.apiClient().ImageSave(ctx, []string{ref}, opts...) + if err != nil { + _ = announce(imageStreamAnswer{Type: ImageStreamType, Error: err.Error()}) + return err + } + defer func() { _ = tar.Close() }() + + if err := announce(imageStreamAnswer{Type: ImageStreamType, Encoding: "chunked", MediaType: "application/x-tar"}); err != nil { + return err + } + cw := httputil.NewChunkedWriter(w) + if _, err := io.CopyBuffer(cw, struct{ io.Reader }{tar}, make([]byte, imageStreamChunkSize)); err != nil { + return err + } + // ChunkedWriter.Close writes the zero-length chunk, which ends the + // stream. Unlike an HTTP message there is deliberately no trailer + // section nor final CRLF: a stock chunked reader stops at the zero + // chunk without consuming either, and leftover bytes would corrupt the + // next JSON answer a provider requesting several images reads. + return cw.Close() +} + // handlePluginMessage processes one provider message, mutating variables in // place; a returned error is terminal for the run (the provider is killed // and the command fails). An unknown message type IS such an error: a // provider requiring a message this compose does not support must fail // loudly, not degrade silently — providers adapt through the -// COMPOSE_PROVIDER_MESSAGES announcement instead. -func (s *composeService) handlePluginMessage(ctx context.Context, project *types.Project, service types.ServiceConfig, msg JsonMessage, +// COMPOSE_PROVIDER_MESSAGES announcement instead. Answer-bearing requests +// (get-service-config, get-relay-info, get-image) are served from their own +// goroutine — see the answers/stdinMu contract in executePlugin. +func (s *composeService) handlePluginMessage( + ctx context.Context, project *types.Project, service types.ServiceConfig, msg JsonMessage, command string, variables *pluginVariables, stdin io.WriteCloser, stdinMu *sync.Mutex, answers *sync.WaitGroup, ) error { switch msg.Type { @@ -339,6 +430,36 @@ func (s *composeService) handlePluginMessage(ctx context.Context, project *types return fmt.Errorf("failed to answer get-relay-info: %w", err) } answerProvider(stdin, stdinMu, answers, payload) + case GetImageType: + // image distribution belongs to the pull command: answering it + // elsewhere would let an image export stall a down or a stop + if command != "pull" { + return fmt.Errorf("invalid message from plugin: %s is only supported during the pull command", GetImageType) + } + ref, platform := msg.Message, msg.Platform + answers.Add(1) + go func() { + defer answers.Done() + // stdinMu is deliberately held for the whole transfer: the + // chunked body must be contiguous on stdin, any concurrent + // answer interleaved into it would corrupt the framing. A + // provider must therefore drain the announced stream before + // expecting any other answer (documented contract); compose + // cannot hang forever on a provider that stops reading — the + // error path kills the process, which EPIPEs the write. + stdinMu.Lock() + defer stdinMu.Unlock() + if err := s.streamImageTo(ctx, stdin, ref, platform); err != nil { + logrus.Warnf("provider %q: get-image %q: %v", service.Name, ref, err) + // A failure after the success announce leaves the stream + // without its terminating chunk, and the channel cannot be + // resynchronized (any byte would read as chunk data). + // Closing stdin makes the truncation observable — the + // provider gets EOF mid-chunk and discards, instead of + // blocking forever on a stream nobody will finish. + _ = stdin.Close() + } + }() case PublishEndpointType: port, upstream, err := parseEndpointMessage(msg.Message) if err != nil { @@ -388,7 +509,7 @@ func (s *composeService) getPluginBinaryPath(provider string) (path string, err return path, err } -func (s *composeService) setupPluginCommand(ctx context.Context, project *types.Project, service types.ServiceConfig, path, command string) (*exec.Cmd, error) { +func (s *composeService) setupPluginCommand(ctx context.Context, project *types.Project, service types.ServiceConfig, path, command string, extraArgs ...string) (*exec.Cmd, error) { cmdOptionsMetadata := s.getPluginMetadata(path, service.Provider.Type, project) var currentCommandMetadata CommandMetadata switch command { @@ -401,6 +522,13 @@ func (s *composeService) setupPluginCommand(ctx context.Context, project *types. return nil, nil } currentCommandMetadata = *cmdOptionsMetadata.Stop + case "pull": + // image distribution is opt-in, declared like stop by the presence + // of the command block in the provider metadata + if cmdOptionsMetadata.Pull == nil { + return nil, nil + } + currentCommandMetadata = *cmdOptionsMetadata.Pull } provider := *service.Provider @@ -417,6 +545,7 @@ func (s *composeService) setupPluginCommand(ctx context.Context, project *types. } } } + args = append(args, extraArgs...) args = append(args, service.Name) cmd := exec.CommandContext(ctx, path, args...) @@ -469,6 +598,9 @@ type ProviderMetadata struct { Up CommandMetadata `json:"up"` Down CommandMetadata `json:"down"` Stop *CommandMetadata `json:"stop,omitempty"` + // Pull declares support for the image-distribution command; like Stop, + // the presence of the block is what opts the provider in. + Pull *CommandMetadata `json:"pull,omitempty"` } func (p ProviderMetadata) IsEmpty() bool { diff --git a/pkg/compose/plugins_test.go b/pkg/compose/plugins_test.go index 9ff2f653c4c..93d2b47bc57 100644 --- a/pkg/compose/plugins_test.go +++ b/pkg/compose/plugins_test.go @@ -97,6 +97,25 @@ func TestProviderMetadata_StopAbsent(t *testing.T) { assert.Assert(t, metadata.Stop == nil, "Stop should be nil when absent from JSON") } +func TestProviderMetadata_PullAbsent(t *testing.T) { + raw := `{"description":"x","up":{"parameters":[]},"down":{"parameters":[]}}` + + var metadata ProviderMetadata + err := json.Unmarshal([]byte(raw), &metadata) + assert.NilError(t, err) + assert.Assert(t, metadata.Pull == nil, "Pull should be nil when absent from JSON") +} + +func TestProviderMetadata_PullPresent(t *testing.T) { + raw := `{"pull":{"parameters":[{"name":"registry"}]}}` + + var metadata ProviderMetadata + err := json.Unmarshal([]byte(raw), &metadata) + assert.NilError(t, err) + assert.Assert(t, metadata.Pull != nil, "Pull should be non-nil when key present") + assert.Equal(t, metadata.Pull.Parameters[0].Name, "registry") +} + func TestProviderMetadata_StopAdvertisedWithoutParameters(t *testing.T) { raw := `{"stop":{"parameters":null}}` From 7a110586c8e2c641dc4d104915782a7a9e80229e Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 15:28:20 +0200 Subject: [PATCH 2/7] feat(provider): providers take part in the image phase of up and pull MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ensureImagesExists and docker compose pull now give provider-backed services their turn: compose invokes the provider's pull command with the image identity and the state of the local daemon cache (--image/--digest/--created), plus two verdicts compose alone can compute so providers never re-implement that arbitration: - --source: the authority for this invocation — local (the daemon's image is the desired state: build-only service, pull_policy build, or an image the current run just built) or registry (resolve the reference upstream — including the CI workflow where build is only the recipe used to publish the image consumers pull); - --policy: missing on the up path (a usable version suffices) or always on compose pull (ensure freshness of the authority). digest/created describe the cache, they are not instructions: the provider persists them as the bookkeeping keys of what it ingested — digest as identity test, created as the ordering fallback for backends that cannot preserve digests. Providers without the metadata block are skipped: behavior unchanged. Providers are independent, so their pulls run without fail-fast cancellation — the first failure no longer kills its siblings mid-run and mangles their reports into context-canceled noise; every provider runs to completion and every failure is reported, prefixed by service. On the pull path the provider phase runs unconditionally — skipping it because an unrelated service failed to pull would silently leave provider runtimes stale — and its error is a per-service pull error, following their exact regime: reported with them, suppressed by IgnoreFailures. A fatal errgroup error is reported joined with the accumulated per-service context instead of dropping it. The provider runs under the root context: the errgroup context is canceled once Wait returns and would kill the provider process on the spot. Signed-off-by: Nicolas De Loof --- pkg/compose/build.go | 18 +- pkg/compose/create.go | 2 +- pkg/compose/images_test.go | 2 +- pkg/compose/provider_images.go | 132 ++++++++++++++ pkg/compose/provider_images_test.go | 266 ++++++++++++++++++++++++++++ pkg/compose/pull.go | 15 +- pkg/compose/run.go | 2 +- 7 files changed, 432 insertions(+), 5 deletions(-) create mode 100644 pkg/compose/provider_images.go create mode 100644 pkg/compose/provider_images_test.go diff --git a/pkg/compose/build.go b/pkg/compose/build.go index b422b50790e..e70b16d5e01 100644 --- a/pkg/compose/build.go +++ b/pkg/compose/build.go @@ -106,7 +106,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti return s.doBuildClassic(ctx, project, serviceToBuild, options) } -func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool) error { +func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, buildOpts *api.BuildOptions, quietPull bool, skipProviders bool) error { for name, service := range project.Services { if service.Provider == nil && service.Image == "" && service.Build == nil { return fmt.Errorf("invalid service %q. Must specify either image or build", name) @@ -127,6 +127,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. return err } + built := map[string]string{} if buildOpts != nil { err = tracing.SpanWrapFunc("project/build", tracing.ProjectOptions(ctx, project), func(ctx context.Context) error { @@ -136,6 +137,7 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. } for name, digest := range builtImages { + built[name] = digest images[name] = api.ImageSummary{ Repository: name, ID: digest, @@ -150,6 +152,20 @@ func (s *composeService) ensureImagesExists(ctx context.Context, project *types. } } + // provider-backed services get their turn in the image phase: the + // provider makes the image available to ITS runtime (weak contract on + // the up path: a usable version present suffices) + if !skipProviders { + err = tracing.SpanWrapFunc("project/provider-pull", tracing.ProjectOptions(ctx, project), + func(ctx context.Context) error { + return s.ensureProviderImages(ctx, project, built, providerPullPolicyMissing) + }, + )(ctx) + if err != nil { + return err + } + } + // set digest as com.docker.compose.image label so we can detect outdated // containers — the single writer of that label, so the platform-pinned // resolution below can't be overwritten by another code path diff --git a/pkg/compose/create.go b/pkg/compose/create.go index 9ac5199db2d..01cc3aacc74 100644 --- a/pkg/compose/create.go +++ b/pkg/compose/create.go @@ -85,7 +85,7 @@ func (s *composeService) create(ctx context.Context, project *types.Project, opt return err } - err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull) + err = s.ensureImagesExists(ctx, project, options.Build, options.QuietPull, options.SkipProviders) if err != nil { return err } diff --git a/pkg/compose/images_test.go b/pkg/compose/images_test.go index a1b53ad477d..53921e97b2d 100644 --- a/pkg/compose/images_test.go +++ b/pkg/compose/images_test.go @@ -423,7 +423,7 @@ func TestPlatformPinnedDigest(t *testing.T) { Return(client.ImageInspectResult{InspectResponse: multiPlatform}, nil) project := newProject() - assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true)) + assert.NilError(t, tested.ensureImagesExists(t.Context(), project, nil, true, false)) assert.Equal(t, project.Services["app"].CustomLabels[compose.ImageDigestLabel], "sha256:s390x") }) diff --git a/pkg/compose/provider_images.go b/pkg/compose/provider_images.go new file mode 100644 index 00000000000..0150474b3da --- /dev/null +++ b/pkg/compose/provider_images.go @@ -0,0 +1,132 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/compose-spec/compose-go/v2/types" + "golang.org/x/sync/errgroup" + + "github.com/docker/compose/v5/pkg/api" +) + +const ( + // providerImageSourceLocal: the desired state is the local daemon's + // image; the provider synchronizes from it (get-image) when its + // bookkeeping says it holds something else. + providerImageSourceLocal = "local" + // providerImageSourceRegistry: the reference is resolvable upstream and + // upstream is the authority; local facts are an optimization, never an + // obligation. + providerImageSourceRegistry = "registry" + + // providerPullPolicyMissing (up path): a usable version present in the + // provider's runtime suffices. + providerPullPolicyMissing = "missing" + // providerPullPolicyAlways (compose pull): the provider must ensure it + // holds the latest version of the authority. + providerPullPolicyAlways = "always" +) + +// ensureProviderImages gives provider-backed services their turn in the image +// phase: for every service whose provider declares the pull command (metadata +// opt-in, like stop), compose invokes +// +// compose pull --image= [--digest=… --created=…] --source=… --policy=… +// +// The provider owns making the image available to ITS runtime: resolving the +// reference upstream, or requesting the local bytes with a get-image message. +// built maps the image names (re)built by the current run, which forces the +// local-source verdict for them. +func (s *composeService) ensureProviderImages(ctx context.Context, project *types.Project, built map[string]string, policy string) error { + // deliberately NOT errgroup.WithContext: providers are independent, and + // fail-fast cancellation would kill the siblings of the first failure + // mid-run, mangling their reports into context-canceled noise. Every + // provider runs to completion and every failure is reported. + var eg errgroup.Group + eg.SetLimit(s.maxConcurrency) + var mu sync.Mutex + var errs []error + for _, service := range project.Services { + if service.Provider == nil { + continue + } + image := api.GetImageNameOrDefault(service, project.Name) + _, justBuilt := built[image] + source := providerImageSource(service, justBuilt) + eg.Go(func() error { + if err := s.runProviderPull(ctx, project, service, image, source, policy); err != nil { + mu.Lock() + errs = append(errs, fmt.Errorf("provider service %q: %w", service.Name, err)) + mu.Unlock() + } + return nil + }) + } + _ = eg.Wait() + return errors.Join(errs...) +} + +// providerImageSource computes the authority VERDICT for one invocation — +// compose owns this arbitration (pull_policy, flags, what this run built) so +// providers never re-implement it: +// - "local": the desired state is the local daemon's image — a declared +// build with no pullable reference, pull_policy build, or an image the +// current run just (re)built; +// - "registry": the provider resolves the reference upstream — including +// the build-in-CI workflow where a build section is only the recipe used +// to publish the image consumers pull. +func providerImageSource(service types.ServiceConfig, justBuilt bool) string { + if service.Build == nil { + return providerImageSourceRegistry + } + if service.Image == "" || justBuilt { + return providerImageSourceLocal + } + if policy, _, err := service.GetPullPolicy(); err == nil && policy == types.PullPolicyBuild { + return providerImageSourceLocal + } + return providerImageSourceRegistry +} + +func (s *composeService) runProviderPull(ctx context.Context, project *types.Project, service types.ServiceConfig, image, source, policy string) error { + args := []string{"--image=" + image, "--source=" + source, "--policy=" + policy} + if digest, created, ok := s.localImageFacts(ctx, image); ok { + args = append(args, "--digest="+digest) + if created != "" { + args = append(args, "--created="+created) + } + } + return s.runPlugin(ctx, project, service, "pull", args...) +} + +// localImageFacts reports the state of the local daemon cache for ref: the +// image ID and its creation time. These describe the cache, they are not +// instructions — the provider persists them as the bookkeeping keys of what +// it ingested (digest as identity test, created as the ordering fallback for +// backends that cannot preserve digests). +func (s *composeService) localImageFacts(ctx context.Context, ref string) (digest, created string, ok bool) { + inspected, err := s.apiClient().ImageInspect(ctx, ref) + if err != nil { + return "", "", false + } + return inspected.ID, inspected.Created, true +} diff --git a/pkg/compose/provider_images_test.go b/pkg/compose/provider_images_test.go new file mode 100644 index 00000000000..578f37ba6ff --- /dev/null +++ b/pkg/compose/provider_images_test.go @@ -0,0 +1,266 @@ +/* + Copyright 2020 Docker Compose CLI authors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package compose + +import ( + "bufio" + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http/httputil" + "os" + "os/exec" + "strconv" + "testing" + + "github.com/compose-spec/compose-go/v2/types" + "go.uber.org/mock/gomock" + "gotest.tools/v3/assert" + + "github.com/docker/compose/v5/pkg/mocks" +) + +// providerSourceService builds a ServiceConfig by assignment: Image and +// PullPolicy are fields promoted from the embedded ContainerSpec, which +// struct literals cannot set before go1.27. +func providerSourceService(build *types.BuildConfig, image, pullPolicy string) types.ServiceConfig { + var s types.ServiceConfig + s.Build = build + s.Image = image + s.PullPolicy = pullPolicy + return s +} + +func TestProviderImageSource(t *testing.T) { + build := &types.BuildConfig{Context: "."} + tests := []struct { + name string + service types.ServiceConfig + justBuilt bool + want string + }{ + {"no build: registry authority", providerSourceService(nil, "nginx", ""), false, providerImageSourceRegistry}, + {"build without image: local-only name", providerSourceService(build, "", ""), false, providerImageSourceLocal}, + {"build just run: local wins for this invocation", providerSourceService(build, "repo/app", ""), true, providerImageSourceLocal}, + {"build as CI recipe, default policy: registry authority", providerSourceService(build, "repo/app", ""), false, providerImageSourceRegistry}, + {"pull_policy build: local authority", providerSourceService(build, "repo/app", types.PullPolicyBuild), false, providerImageSourceLocal}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, providerImageSource(tc.service, tc.justBuilt), tc.want) + }) + } +} + +// TestStreamImageTo locks the wire format of the get-image answer: one JSON +// announce line, then the tar bytes as an HTTP/1.1 chunked body terminated by +// the zero chunk and a final CRLF — decodable with a stock chunked reader. +func TestStreamImageTo(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + s := svc.(*composeService) + + payload := bytes.Repeat([]byte("compose-image-bytes\x00\x17\xff"), 100_000) // > one chunk, binary-hostile values included + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}). + Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil) + + var out bytes.Buffer + assert.NilError(t, s.streamImageTo(t.Context(), &out, "proj-app", "")) + + rd := bufio.NewReader(&out) + line, err := rd.ReadString('\n') + assert.NilError(t, err) + var announce imageStreamAnswer + assert.NilError(t, json.Unmarshal([]byte(line), &announce)) + assert.Equal(t, announce.Type, ImageStreamType) + assert.Equal(t, announce.Encoding, "chunked") + assert.Equal(t, announce.Error, "") + + body, err := io.ReadAll(httputil.NewChunkedReader(rd)) + assert.NilError(t, err) + assert.Assert(t, bytes.Equal(body, payload), "chunked round-trip must be byte-identical (%d vs %d bytes)", len(body), len(payload)) + + // nothing may follow the zero chunk: a stock chunked reader stops there + // without consuming further bytes, and any leftover would corrupt the + // next answer of a provider requesting several images + rest, err := io.ReadAll(rd) + assert.NilError(t, err) + assert.Equal(t, string(rest), "", "no byte may follow the terminating zero chunk") +} + +// TestStreamImageTo_SaveError: when the daemon cannot export the image, the +// announce line carries the error and no stream follows — the provider is +// never left waiting. +func TestStreamImageTo_SaveError(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + s := svc.(*composeService) + + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-app"}). + Return(nil, errors.New("no such image")) + + var out bytes.Buffer + assert.ErrorContains(t, s.streamImageTo(t.Context(), &out, "proj-app", ""), "no such image") + + var announce imageStreamAnswer + assert.NilError(t, json.Unmarshal(out.Bytes(), &announce)) + assert.Equal(t, announce.Type, ImageStreamType) + assert.ErrorContains(t, errors.New(announce.Error), "no such image") + assert.Equal(t, announce.Encoding, "") +} + +type fakeImageSaveResult struct{ io.ReadCloser } + +// TestExecutePlugin_GetImage runs the pull command against a fake provider +// (this test binary re-executed, see TestHelperProviderPull): the get-image +// request must be answered with the announced chunked stream, which the +// provider consumes and acknowledges by hashing. +func TestExecutePlugin_GetImage(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + payload := bytes.Repeat([]byte{0x17, 0x00, 0xAB}, 500_000) // ETB and NUL laden + sum := sha256.Sum256(payload) + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}). + Return(fakeImageSaveResult{io.NopCloser(bytes.NewReader(payload))}, nil) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + variables, err := svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "pull", service) + assert.NilError(t, err) + assert.Equal(t, variables.prefixed["SHA"], hex.EncodeToString(sum[:])) + assert.Equal(t, variables.prefixed["LEN"], strconv.Itoa(len(payload))) +} + +// TestHelperProviderPull is not a test: it is the fake provider process +// spawned by TestExecutePlugin_GetImage. It requests the image, decodes the +// chunked stream with the stock reader, and reports what it received. +func TestHelperProviderPull(t *testing.T) { + if os.Getenv("GO_WANT_HELPER_PROCESS") != "1" { + t.Skip("helper process for TestExecutePlugin_GetImage") + } + emit := func(msg JsonMessage) { + if err := json.NewEncoder(os.Stdout).Encode(msg); err != nil { + os.Exit(1) + } + } + emit(JsonMessage{Type: GetImageType, Message: "proj-db"}) + + stdin := bufio.NewReader(os.Stdin) + line, err := stdin.ReadString('\n') + if err != nil { + emit(JsonMessage{Type: ErrorType, Message: "reading announce: " + err.Error()}) + os.Exit(1) + } + var announce imageStreamAnswer + if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" { + emit(JsonMessage{Type: ErrorType, Message: fmt.Sprintf("bad announce %q (err %v)", line, err)}) + os.Exit(1) + } + h := sha256.New() + n, err := io.Copy(h, httputil.NewChunkedReader(stdin)) + if err != nil { + emit(JsonMessage{Type: ErrorType, Message: "reading stream: " + err.Error()}) + os.Exit(1) + } + emit(JsonMessage{Type: SetEnvType, Message: "SHA=" + hex.EncodeToString(h.Sum(nil))}) + emit(JsonMessage{Type: SetEnvType, Message: fmt.Sprintf("LEN=%d", n)}) + os.Exit(0) +} + +// TestExecutePlugin_GetImageRefusedOutsidePull: image distribution belongs to +// the pull command — a get-image emitted during up is a protocol error, so an +// image export can never stall another lifecycle command. +func TestExecutePlugin_GetImageRefusedOutsidePull(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + _, err = svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "up", service) + assert.ErrorContains(t, err, "get-image is only supported during the pull command") +} + +// failingReader errors after serving its prefix, simulating a daemon export +// dying mid-stream. +type failingReader struct{ data []byte } + +func (f *failingReader) Read(p []byte) (int, error) { + if len(f.data) == 0 { + return 0, errors.New("export died mid-stream") + } + n := copy(p, f.data) + f.data = f.data[n:] + return n, nil +} +func (f *failingReader) Close() error { return nil } + +// TestExecutePlugin_GetImageAbortedMidStream: a transfer failing after the +// success announce must end observably — compose closes the answer channel, +// the provider reads EOF mid-chunk and reports, instead of blocking forever +// on a stream nobody will finish. +func TestExecutePlugin_GetImageAbortedMidStream(t *testing.T) { + mockCtrl := gomock.NewController(t) + cli := mocks.NewMockCli(mockCtrl) + apiClient := mocks.NewMockAPIClient(mockCtrl) + cli.EXPECT().Client().Return(apiClient).AnyTimes() + svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{})) + assert.NilError(t, err) + + apiClient.EXPECT().ImageSave(gomock.Any(), []string{"proj-db"}). + Return(&failingReader{data: bytes.Repeat([]byte{0xAB}, 4096)}, nil) + + cmd := exec.Command(os.Args[0], "-test.run=TestHelperProviderPull") + cmd.Env = append(os.Environ(), "GO_WANT_HELPER_PROCESS=1") + + service := types.ServiceConfig{ + Name: "db", + Provider: &types.ServiceProviderConfig{Type: "fake"}, + } + _, err = svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "pull", service) + assert.ErrorContains(t, err, "reading stream") +} diff --git a/pkg/compose/pull.go b/pkg/compose/pull.go index 45ba070efa7..2456cbeb06a 100644 --- a/pkg/compose/pull.go +++ b/pkg/compose/pull.go @@ -64,6 +64,7 @@ type imagePuller struct { } func (s *composeService) pull(ctx context.Context, project *types.Project, opts api.PullOptions) error { + rootCtx := ctx images, _, err := s.getLocalImagesDigests(ctx, project) if err != nil { return err @@ -97,8 +98,20 @@ func (s *composeService) pull(ctx context.Context, project *types.Project, opts logrus.Warnf("WARNING: Some service image(s) must be built from source by running:\n docker compose build %s", strings.Join(p.mustBuild, " ")) } + // provider-backed services: pull is the strong contract — the provider + // must ensure its runtime holds the latest version of the authority. + // It runs even when a regular pull failed: skipping the provider phase + // because an unrelated service failed would silently leave provider + // runtimes stale. Its error is a per-service pull error and follows + // their exact regime — reported below, suppressed by IgnoreFailures. + // rootCtx, not ctx: the errgroup context above is canceled once Wait + // returns, and would kill the provider on the spot. + p.pullErrors = append(p.pullErrors, s.ensureProviderImages(rootCtx, project, nil, providerPullPolicyAlways)) + if err != nil { - return err + // fatal errgroup error: report it with whatever per-service context + // accumulated (provider phase included) instead of dropping it + return errors.Join(append(p.pullErrors, err)...) } if opts.IgnoreFailures { return nil diff --git a/pkg/compose/run.go b/pkg/compose/run.go index 1f233fe32f5..081a908290c 100644 --- a/pkg/compose/run.go +++ b/pkg/compose/run.go @@ -160,7 +160,7 @@ func (s *composeService) prepareRun(ctx context.Context, project *types.Project, // Only ensure image exists for the target service, dependencies were already handled by startDependencies buildOpts := prepareBuildOptions(options) - if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull); err != nil { // all dependencies already checked, but might miss service img + if err := s.ensureImagesExists(ctx, project, buildOpts, options.QuietPull, false); err != nil { // all dependencies already checked, but might miss service img return prepareRunResult{}, err } From baab4b2c9a06e24e44ba5032dec02891c6372949 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 14:36:07 +0200 Subject: [PATCH 3/7] fix(start): a project made only of provider services could not start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit startService errored with 'no container to start' whenever the project-wide container listing came back empty, before the per-service filtering that lets provider services pass through. A project whose services are all provider-backed legitimately reaches the start phase with zero containers. The empty-list error is now softened for provider services only — a relay container deployed for published endpoints still goes through the regular start path. Signed-off-by: Nicolas De Loof --- pkg/compose/service_containers.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pkg/compose/service_containers.go b/pkg/compose/service_containers.go index 2337765adba..0ac19ee9b7a 100644 --- a/pkg/compose/service_containers.go +++ b/pkg/compose/service_containers.go @@ -591,7 +591,6 @@ func (s *composeService) startService(ctx context.Context, if service.Deploy != nil && service.Deploy.Replicas != nil && *service.Deploy.Replicas == 0 { return nil } - err := s.waitDependencies(ctx, project, service.Name, service.DependsOn, containers, timeout) if err != nil { return err @@ -601,6 +600,13 @@ func (s *composeService) startService(ctx context.Context, if service.GetScale() == 0 { return nil } + if service.Provider != nil { + // a provider-backed service usually has no container of its own + // (it gets one — the relay — only when the provider published + // endpoints), so a project made only of provider services + // legitimately reaches the start phase with no container at all + return nil + } return errNoContainerToStart(service.Name) } From b322e4ab04f25de8a353cb13a5cd5f7e726de7b2 Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Tue, 22 Sep 2026 15:09:57 +0200 Subject: [PATCH 4/7] docs(provider), e2e: image distribution documented and exercised MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit extension.md gains the Image distribution section: invocation contract (--image/--digest/--created/--source/--policy), the get-image request and its chunked image-stream answer (zero-chunk terminated, stdin exclusivity and drain-before-next-answer contract), the two-branch decision guidance (digest as identity, created as ordering fallback, reproducible builds caveat), and the support-by-presence metadata convention is now stated as the general rule. The example provider implements pull end to end — request, stock chunked reader, hash recorded at PROVIDER_PULL_MARKER — and the e2e scenario locks the whole path: up builds the service image and streams it to the provider under the local verdict, compose pull re-invokes it under the freshness contract. Signed-off-by: Nicolas De Loof --- docs/examples/provider.go | 95 ++++++++++++++++++- docs/extension.md | 73 +++++++++++++- pkg/e2e/providers_test.go | 18 ++++ .../testdata/TestProviderImagePull/Dockerfile | 2 + .../TestProviderImagePull/compose.yaml | 9 ++ 5 files changed, 191 insertions(+), 6 deletions(-) create mode 100644 pkg/e2e/testdata/TestProviderImagePull/Dockerfile create mode 100644 pkg/e2e/testdata/TestProviderImagePull/compose.yaml diff --git a/docs/examples/provider.go b/docs/examples/provider.go index 28dcffe229e..59f939e3bbc 100644 --- a/docs/examples/provider.go +++ b/docs/examples/provider.go @@ -18,9 +18,12 @@ package main import ( "bufio" + "crypto/sha256" "encoding/json" "fmt" + "io" "net" + "net/http/httputil" "os" "os/exec" "slices" @@ -86,11 +89,88 @@ func composeCommand() *cobra.Command { Args: cobra.ExactArgs(1), } - c.AddCommand(upCmd, downCmd, stopCmd) - c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd)) + // pull is the image-distribution command: compose invokes it during the + // image phase (up) and on `docker compose pull`, passing the identity of + // the service image and the state of the local daemon cache. None of its + // flags is required: they are injected by compose, not declared by the + // user under provider.options. + pullCmd := &cobra.Command{ + Use: "pull", + Run: pull, + Args: cobra.ExactArgs(1), + } + pullCmd.Flags().String("image", "", "Image reference as compose resolved it") + pullCmd.Flags().String("digest", "", "Image ID in the local daemon cache, when present") + pullCmd.Flags().String("created", "", "Creation time of the local cache entry, when present") + pullCmd.Flags().String("source", "", "Authority verdict: local (sync from the daemon) or registry (resolve upstream)") + pullCmd.Flags().String("policy", "", "missing (a usable version suffices) or always (ensure freshness)") + + c.AddCommand(upCmd, downCmd, stopCmd, pullCmd) + c.AddCommand(metadataCommand(upCmd, downCmd, stopCmd, pullCmd)) return c } +// pull demonstrates the image-distribution contract. A real provider would +// compare the announced digest/created with its bookkeeping and either +// resolve the reference upstream (source=registry) or request the local +// bytes (source=local). This demo requests the stream whenever +// PROVIDER_PULL_MARKER is set, hashes it, and records the outcome there. +func pull(cmd *cobra.Command, args []string) { + image, _ := cmd.Flags().GetString("image") + digest, _ := cmd.Flags().GetString("digest") + created, _ := cmd.Flags().GetString("created") + source, _ := cmd.Flags().GetString("source") + policy, _ := cmd.Flags().GetString("policy") + + emit := func(kind, message string) { + payload, _ := json.Marshal(map[string]string{"type": kind, "message": message}) + fmt.Println(string(payload)) + } + emit("info", fmt.Sprintf("ensuring image %s (source=%s, policy=%s)", image, source, policy)) + + marker := os.Getenv("PROVIDER_PULL_MARKER") + if marker == "" { + // nothing to synchronize in the demo: a real registry-sourced provider + // would pull the reference upstream here + return + } + + // Request the image bytes from the local daemon. The answer is one JSON + // line, then — unless it carries an error — the tar as an HTTP/1.1 + // chunked body (readable with any stock chunked reader), whose zero + // chunk marks a COMPLETE transfer. + request, _ := json.Marshal(map[string]string{"type": "get-image", "message": image}) + fmt.Println(string(request)) + + stdin := bufio.NewReader(os.Stdin) + line, err := stdin.ReadString('\n') + if err != nil { + emit("error", "reading image-stream announce: "+err.Error()) + os.Exit(1) + } + var announce struct { + Encoding string `json:"encoding"` + Error string `json:"error"` + } + if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" { + emit("error", fmt.Sprintf("image stream unavailable: %q (err %v)", line, err)) + os.Exit(1) + } + h := sha256.New() + n, err := io.Copy(h, httputil.NewChunkedReader(stdin)) + if err != nil { + emit("error", "image stream truncated: "+err.Error()) + os.Exit(1) + } + record := fmt.Sprintf("image=%s digest=%s created=%s source=%s policy=%s sha256=%x bytes=%d\n", + image, digest, created, source, policy, h.Sum(nil), n) + if err := os.WriteFile(marker, []byte(record), 0o600); err != nil { + emit("error", "writing marker: "+err.Error()) + os.Exit(1) + } + emit("info", fmt.Sprintf("image received (%d bytes)", n)) +} + // serveDemoCommand is the detached helper process behind the // publish-endpoint demonstration: a TCP server on the given address // answering every connection with a fixed HTTP response, exiting on its own @@ -288,23 +368,27 @@ func stop(_ *cobra.Command, _ []string) { } } -func metadataCommand(upCmd, downCmd, stopCmd *cobra.Command) *cobra.Command { +func metadataCommand(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) *cobra.Command { return &cobra.Command{ Use: "metadata", Run: func(cmd *cobra.Command, _ []string) { - metadata(upCmd, downCmd, stopCmd) + metadata(upCmd, downCmd, stopCmd, pullCmd) }, Args: cobra.NoArgs, } } -func metadata(upCmd, downCmd, stopCmd *cobra.Command) { +func metadata(upCmd, downCmd, stopCmd, pullCmd *cobra.Command) { metadata := ProviderMetadata{} metadata.Description = "Manage services on AwesomeCloud" metadata.Up = commandParameters(upCmd) metadata.Down = commandParameters(downCmd) stopParams := commandParameters(stopCmd) metadata.Stop = &stopParams + // like stop, declaring the pull block is what opts the provider into + // image distribution + pullParams := commandParameters(pullCmd) + metadata.Pull = &pullParams jsonMetadata, err := json.Marshal(metadata) if err != nil { panic(err) @@ -332,6 +416,7 @@ type ProviderMetadata struct { Up CommandMetadata `json:"up"` Down CommandMetadata `json:"down"` Stop *CommandMetadata `json:"stop,omitempty"` + Pull *CommandMetadata `json:"pull,omitempty"` } type CommandMetadata struct { diff --git a/docs/extension.md b/docs/extension.md index 175a666d3da..419e54bdd23 100644 --- a/docs/extension.md +++ b/docs/extension.md @@ -30,7 +30,10 @@ the resource(s) needed to run a service. If `provider.type` doesn't resolve into any of those, Compose will report an error and interrupt the `up` command. To be a valid Compose extension, provider command *MUST* accept a `compose` command (which can be hidden) -with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`. +with subcommands `up` and `down`. It *MAY* additionally implement a `stop` subcommand to support `docker compose stop`, +and a `pull` subcommand to take part in image distribution (see [Image distribution](#image-distribution)). +Optional subcommands are declared through the provider metadata: the presence of the command block is what +opts the provider in. ## Up lifecycle @@ -121,6 +124,68 @@ sequenceDiagram Compose-)Shell: service started ``` +## Image distribution + +A provider-backed service can declare `build` or `image` like any other service; the image then has to reach +the provider's runtime, which may be nowhere near the local daemon. Providers opt into image distribution by +declaring a `pull` block in their `metadata` output — like `stop`, the presence of the block is the +declaration of support. Providers without it keep managing images on their own during `up`. + +When the provider declares `pull`, Compose invokes it during the image phase of `up` (after any build) and on +`docker compose pull`: + +```console +awesomecloud compose --project-name pull --image= --source= --policy= [--digest= --created=