fix(provider): close stdin when an announce write itself fails

docker-agent found that streamImageTo returned a plain error (channel
intact) when the announce write itself failed -- for the success
announce, and for the error announce on a platform-parse/ImageSave
failure. In both cases the provider never received anything to read
and was left blocked on ReadString, while the caller, seeing a plain
error, never closed stdin either -- a deadlock.

Both paths now return streamAbortedError when the announce write
fails, matching the existing mid-stream/close failure handling so the
caller's errors.As check closes stdin and unblocks the provider.

Pinned by two new regression tests using a writer that always errors,
verified by reverting the fix locally and confirming both fail with a
plain error instead of streamAbortedError.

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

pkg/compose/plugins.go

@@ -344,12 +344,13 @@ 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.
+// streamAbortedError marks a streamImageTo failure that leaves the channel
+// unresynchronizable: either the chunked body was left incomplete after the
+// success announce, or a write to w itself failed (the pipe is broken,
+// whatever was or wasn't delivered). A failure to prepare the image (an
+// invalid platform, an ImageSave error) whose error announce write
+// succeeded is reported as a plain error instead — the channel is intact
+// and the provider has a well-formed error line to read.
 type streamAbortedError struct {
 	err error
 }
@@ -371,20 +372,24 @@ func (s *composeService) streamImageTo(ctx context.Context, w io.Writer, ref, pl
 	if platform != "" {
 		p, err := platforms.Parse(platform)
 		if err != nil {
-			_ = announce(imageStreamAnswer{Type: ImageStreamType, Error: fmt.Sprintf("invalid platform %q: %s", platform, err)})
+			if werr := announce(imageStreamAnswer{Type: ImageStreamType, Error: fmt.Sprintf("invalid platform %q: %s", platform, err)}); werr != nil {
+				return &streamAbortedError{werr}
+			}
 			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()})
+		if werr := announce(imageStreamAnswer{Type: ImageStreamType, Error: err.Error()}); werr != nil {
+			return &streamAbortedError{werr}
+		}
 		return err
 	}
 	defer func() { _ = tar.Close() }()
 
 	if err := announce(imageStreamAnswer{Type: ImageStreamType, Encoding: "chunked", MediaType: "application/x-tar"}); err != nil {
-		return err
+		return &streamAbortedError{err}
 	}
 	cw := httputil.NewChunkedWriter(w)
 	if _, err := io.CopyBuffer(cw, struct{ io.Reader }{tar}, make([]byte, imageStreamChunkSize)); err != nil {

pkg/compose/provider_images_test.go

@@ -137,6 +137,53 @@ func TestStreamImageTo_SaveError(t *testing.T) {
 	assert.Equal(t, announce.Encoding, "")
 }
 
+// TestStreamImageTo_SuccessAnnounceWriteFailureAborts: when the write of the
+// success announce itself fails, the channel is broken regardless of
+// whether the provider received anything — the caller must close stdin
+// (via streamAbortedError) rather than leave the provider blocked reading a
+// line that was never delivered.
+func TestStreamImageTo_SuccessAnnounceWriteFailureAborts(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(fakeImageSaveResult{io.NopCloser(bytes.NewReader([]byte("tar")))}, nil)
+
+	err = s.streamImageTo(t.Context(), failingWriter{errors.New("broken pipe")}, "proj-app", "")
+	var aborted *streamAbortedError
+	assert.Assert(t, errors.As(err, &aborted), "want *streamAbortedError, got %T: %v", err, err)
+}
+
+// TestStreamImageTo_ErrorAnnounceWriteFailureAborts: same reasoning on the
+// error-announce path — if ImageSave fails AND the resulting error line
+// cannot be written either, the provider gets no error to read and must be
+// unblocked by closing stdin.
+func TestStreamImageTo_ErrorAnnounceWriteFailureAborts(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"))
+
+	err = s.streamImageTo(t.Context(), failingWriter{errors.New("broken pipe")}, "proj-app", "")
+	var aborted *streamAbortedError
+	assert.Assert(t, errors.As(err, &aborted), "want *streamAbortedError, got %T: %v", err, err)
+}
+
+type failingWriter struct{ err error }
+
+func (w failingWriter) Write([]byte) (int, error) { return 0, w.err }
+
 type fakeImageSaveResult struct{ io.ReadCloser }
 
 // TestExecutePlugin_GetImage runs the pull command against a fake provider