feat(provider): providers take part in the image phase of up and pull

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 <[email protected]>

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

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
 	}

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")
 	})
 

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
+//
+//	<provider> compose pull --image=<ref> [--digest=… --created=…] --source=… --policy=… <service>
+//
+// 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
+}

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")
+}

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

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
 	}