From 42f48072bbf92ee9b0e43f9fdf2008d03546e7ca Mon Sep 17 00:00:00 2001 From: Nicolas De Loof Date: Fri, 2 Oct 2026 15:49:08 +0200 Subject: [PATCH] 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 --- pkg/compose/plugins.go | 23 ++++++++------ pkg/compose/provider_images_test.go | 47 +++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 9 deletions(-) diff --git a/pkg/compose/plugins.go b/pkg/compose/plugins.go index 368a477d1..50aef6d71 100644 --- a/pkg/compose/plugins.go +++ b/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 { diff --git a/pkg/compose/provider_images_test.go b/pkg/compose/provider_images_test.go index 1f194c467..31545c104 100644 --- a/pkg/compose/provider_images_test.go +++ b/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