fix(provider): address review findings on image distribution

Three fixes from glours' review:

- ensureProviderImages now skips a provider service declaring neither
  image: nor build: (legal for a pure resource provider, e.g. one
  backing a remote Amazon RDS instance). Without the guard, such a
  service reached pull with a fabricated "<project>-<service>"
  reference (GetImageNameOrDefault) under source=registry
  (providerImageSource treats Build == nil as registry-authoritative)
  -- a name that was never real. Already reachable through existing
  e2e fixtures (TestDependsOnMultipleProviders' provider1/provider2),
  only a no-op today because the demo pull() short-circuits on an
  unset env var, not because of any guard here.

- localImageFacts now resolves --digest through the manifest-aware
  localContentDigest instead of the raw ImageInspect ID. Under
  BuildKit provenance attestations, the raw ID covers the attestation
  manifest too and changes on every build even when the runnable image
  content didn't (#13636), which would make the provider re-sync on
  every up for no reason -- defeating docs/extension.md's own
  "identity test" contract for --digest. Threads service.Platform
  through from runProviderPull so the resolved manifest matches what
  the provider will actually run.

- streamImageTo now distinguishes a failure before the success
  announce (invalid platform, ImageSave error -- a clean, well-formed
  error line, channel intact) from one after it (the chunked body
  truncated, channel unresynchronizable) via a new streamAbortedError
  type. Only the latter closes stdin; the provider protocol supports
  several sequential get-image requests per pull invocation, and
  closing stdin on a clean pre-stream error was killing any other
  independent request in the same invocation for no reason.

Each fix is pinned by a regression test, verified by reverting the
fix locally and confirming the new test catches it.

Signed-off-by: Nicolas De Loof <[email protected]>

pkg/compose/plugins.go

@@ -344,6 +344,19 @@ const imageStreamChunkSize = 1024 * 1024
 // was aborted: the provider must discard it.
 // When the image cannot be exported, the announce line carries an error
 // instead and no stream follows.
+// streamAbortedError marks a streamImageTo failure that happened after the
+// success announce was already sent: the chunked body is now incomplete and
+// the channel cannot be resynchronized. A failure before the announce (an
+// invalid platform, an ImageSave error) is reported as a plain error
+// instead — the channel is still intact, nothing has been promised to the
+// provider yet.
+type streamAbortedError struct {
+	err error
+}
+
+func (e *streamAbortedError) Error() string { return e.err.Error() }
+func (e *streamAbortedError) Unwrap() error { return e.err }
+
 func (s *composeService) streamImageTo(ctx context.Context, w io.Writer, ref, platform string) error {
 	announce := func(a imageStreamAnswer) error {
 		payload, err := json.Marshal(a)
@@ -375,14 +388,17 @@ func (s *composeService) streamImageTo(ctx context.Context, w io.Writer, ref, pl
 	}
 	cw := httputil.NewChunkedWriter(w)
 	if _, err := io.CopyBuffer(cw, struct{ io.Reader }{tar}, make([]byte, imageStreamChunkSize)); err != nil {
-		return err
+		return &streamAbortedError{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()
+	if err := cw.Close(); err != nil {
+		return &streamAbortedError{err}
+	}
+	return nil
 }
 
 // handlePluginMessage processes one provider message, mutating variables in
@@ -462,13 +478,21 @@ func (s *composeService) handlePluginMessage(
 			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()
+				// A failure before the success announce (invalid platform,
+				// ImageSave error) is a clean, well-formed error line: the
+				// channel is intact, and closing stdin here would needlessly
+				// kill any other independent get-image request in the same
+				// pull invocation. Only a failure after the success
+				// announce leaves the stream without its terminating chunk,
+				// with the channel unable to be resynchronized (any byte
+				// would read as chunk data) — streamImageTo reports that
+				// case as a *streamAbortedError so closing stdin here makes
+				// the truncation observable, instead of blocking forever on
+				// a stream nobody will finish.
+				var aborted *streamAbortedError
+				if errors.As(err, &aborted) {
+					_ = stdin.Close()
+				}
 			}
 		}()
 	case PublishEndpointType:

pkg/compose/provider_images.go

@@ -69,6 +69,9 @@ func (s *composeService) ensureProviderImages(ctx context.Context, project *type
 		if service.Provider == nil {
 			continue
 		}
+		if service.Image == "" && service.Build == nil {
+			continue // pure resource provider: nothing to distribute
+		}
 		image := api.GetImageNameOrDefault(service, project.Name)
 		_, justBuilt := built[image]
 		source := providerImageSource(service, justBuilt)
@@ -109,7 +112,7 @@ func providerImageSource(service types.ServiceConfig, justBuilt bool) string {
 
 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 {
+	if digest, created, ok := s.localImageFacts(ctx, image, service.Platform); ok {
 		args = append(args, "--digest="+digest)
 		if created != "" {
 			args = append(args, "--created="+created)
@@ -119,14 +122,29 @@ func (s *composeService) runProviderPull(ctx context.Context, project *types.Pro
 }
 
 // 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)
+// image's canonical content digest 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).
+//
+// The digest goes through the manifest-aware localContentDigest, not the raw
+// inspect ID: under BuildKit provenance attestations, the raw ID changes on
+// every build even when the runnable image content didn't (see
+// images.go:279-291, #13636), which would make the provider re-sync on every
+// up for no reason — defeating the "identity test" docs/extension.md
+// documents --digest as.
+func (s *composeService) localImageFacts(ctx context.Context, ref, platform string) (digest, created string, ok bool) {
+	opts, err := s.imageInspectOptions(ctx)
 	if err != nil {
 		return "", "", false
 	}
-	return inspected.ID, inspected.Created, true
+	inspected, err := s.apiClient().ImageInspect(ctx, ref, opts...)
+	if err != nil {
+		return "", "", false
+	}
+	id, _, err := localContentDigest(inspected, platform)
+	if err != nil {
+		return "", "", false
+	}
+	return id, inspected.Created, true
 }

pkg/compose/provider_images_test.go

@@ -32,6 +32,8 @@ import (
 	"testing"
 
 	"github.com/compose-spec/compose-go/v2/types"
+	"github.com/moby/moby/api/types/image"
+	"github.com/moby/moby/client"
 	"go.uber.org/mock/gomock"
 	"gotest.tools/v3/assert"
 
@@ -41,10 +43,10 @@ import (
 // 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 {
+func providerSourceService(build *types.BuildConfig, imageRef, pullPolicy string) types.ServiceConfig {
 	var s types.ServiceConfig
 	s.Build = build
-	s.Image = image
+	s.Image = imageRef
 	s.PullPolicy = pullPolicy
 	return s
 }
@@ -264,3 +266,144 @@ func TestExecutePlugin_GetImageAbortedMidStream(t *testing.T) {
 	_, err = svc.(*composeService).executePlugin(t.Context(), &types.Project{Name: "proj"}, cmd, "pull", service)
 	assert.ErrorContains(t, err, "reading stream")
 }
+
+// TestExecutePlugin_GetImageSaveErrorDoesNotAbortChannel: a failure BEFORE
+// the success announce (ImageSave itself failing) is a clean, well-formed
+// error line — the channel is intact. A provider making a second,
+// independent get-image request in the same pull invocation must still get
+// served, proving compose did not close stdin on it.
+func TestExecutePlugin_GetImageSaveErrorDoesNotAbortChannel(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-missing"}).
+		Return(nil, errors.New("no such image"))
+	payload := []byte("second request succeeds")
+	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=TestHelperProviderPullTwice")
+	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["FIRST_ERROR"], "no such image")
+	assert.Equal(t, variables.prefixed["SHA"], hex.EncodeToString(sum[:]))
+}
+
+// TestHelperProviderPullTwice is not a test: it is the fake provider process
+// spawned by TestExecutePlugin_GetImageSaveErrorDoesNotAbortChannel. It
+// requests an image that fails to export, then — on the SAME stdin, proving
+// compose never closed it — requests a second image that succeeds.
+func TestHelperProviderPullTwice(t *testing.T) {
+	if os.Getenv("GO_WANT_HELPER_PROCESS") != "1" {
+		t.Skip("helper process for TestExecutePlugin_GetImageSaveErrorDoesNotAbortChannel")
+	}
+	emit := func(msg JsonMessage) {
+		if err := json.NewEncoder(os.Stdout).Encode(msg); err != nil {
+			os.Exit(1)
+		}
+	}
+	stdin := bufio.NewReader(os.Stdin)
+
+	emit(JsonMessage{Type: GetImageType, Message: "proj-missing"})
+	line, err := stdin.ReadString('\n')
+	if err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: "reading first announce: " + err.Error()})
+		os.Exit(1)
+	}
+	var announce imageStreamAnswer
+	if err := json.Unmarshal([]byte(line), &announce); err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: fmt.Sprintf("bad first announce %q (err %v)", line, err)})
+		os.Exit(1)
+	}
+	emit(JsonMessage{Type: SetEnvType, Message: "FIRST_ERROR=" + announce.Error})
+
+	emit(JsonMessage{Type: GetImageType, Message: "proj-db"})
+	line, err = stdin.ReadString('\n')
+	if err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: "reading second announce: " + err.Error()})
+		os.Exit(1)
+	}
+	announce = imageStreamAnswer{} // JSON unmarshal only overwrites fields present in the new line
+	if err := json.Unmarshal([]byte(line), &announce); err != nil || announce.Error != "" || announce.Encoding != "chunked" {
+		emit(JsonMessage{Type: ErrorType, Message: fmt.Sprintf("bad second announce %q (err %v)", line, err)})
+		os.Exit(1)
+	}
+	h := sha256.New()
+	if _, err := io.Copy(h, httputil.NewChunkedReader(stdin)); err != nil {
+		emit(JsonMessage{Type: ErrorType, Message: "reading second stream: " + err.Error()})
+		os.Exit(1)
+	}
+	emit(JsonMessage{Type: SetEnvType, Message: "SHA=" + hex.EncodeToString(h.Sum(nil))})
+	os.Exit(0)
+}
+
+// TestEnsureProviderImages_SkipsPureResourceProvider: a provider service
+// declaring neither image: nor build: (legal — e.g. a remote-resource
+// provider like an RDS instance) has nothing to distribute. It must never
+// reach the pull command at all — reaching it would hand the provider a
+// fabricated "<project>-<service>" reference (GetImageNameOrDefault) under
+// source=registry (providerImageSource with Build == nil), a name that was
+// never real.
+func TestEnsureProviderImages_SkipsPureResourceProvider(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)
+
+	project := &types.Project{
+		Name: "proj",
+		Services: types.Services{
+			"db": {
+				Name: "db",
+				// a provider type that resolves to no plugin binary at all:
+				// if the skip guard didn't fire, runProviderPull would reach
+				// getPluginBinaryPath and fail, surfacing as a non-nil error
+				// here instead of this test's expected nil.
+				Provider: &types.ServiceProviderConfig{Type: "no-such-provider-plugin"},
+			},
+		},
+	}
+	err = svc.(*composeService).ensureProviderImages(t.Context(), project, nil, providerPullPolicyMissing)
+	assert.NilError(t, err, "a provider with neither image nor build has nothing to distribute and must not invoke pull")
+}
+
+// TestLocalImageFacts_UsesManifestAwareDigest: --digest must be the
+// manifest-scoped content digest (localContentDigest), not the raw inspect
+// ID. Under BuildKit provenance attestations the raw ID covers the
+// attestation manifest too and changes on every build even when the runnable
+// image content didn't (#13636) — which would make the provider re-sync on
+// every up for no reason, defeating docs/extension.md's "identity test"
+// contract for --digest.
+func TestLocalImageFacts_UsesManifestAwareDigest(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	defer mockCtrl.Finish()
+	api, tested := newTestComposeService(t, mockCtrl, "1.48")
+
+	api.EXPECT().
+		ImageInspect(anyCancellableContext(), "proj-db", gomock.Any()).
+		Return(client.ImageInspectResult{InspectResponse: image.InspectResponse{
+			ID:       "sha256:index",
+			Metadata: image.Metadata{},
+			Manifests: []image.ManifestSummary{
+				imageManifest("sha256:image", "amd64", true),
+				attestationManifest(),
+			},
+		}}, nil)
+
+	digest, _, ok := tested.localImageFacts(t.Context(), "proj-db", "")
+	assert.Assert(t, ok)
+	assert.Equal(t, digest, "sha256:image", "--digest must be the image-kind manifest digest, not the attestation-covering index ID")
+}