Fix foreground stdio deadlock when the internal logging process stops consuming

pkg/cioutil/container_io.go

@@ -32,6 +32,7 @@ import (
 	"github.com/containerd/containerd/v2/cmd/containerd-shim-runc-v2/process"
 	"github.com/containerd/containerd/v2/defaults"
 	"github.com/containerd/containerd/v2/pkg/cio"
+	"github.com/containerd/log"
 )
 
 const binaryIOProcTermTimeout = 12 * time.Second // Give logger process 10 seconds for cleanup
@@ -52,6 +53,44 @@ var bufPool = sync.Pool{
 	},
 }
 
+// closeOnce wraps f's Close so that extra calls return the first result
+// instead of "file already closed". The logger pipe ends below are closed
+// individually on the success path, but also sit in the error-path closers
+// list of NewContainerIO.
+func closeOnce(f *os.File) func() error {
+	var once sync.Once
+	var err error
+	return func() error {
+		once.Do(func() { err = f.Close() })
+		return err
+	}
+}
+
+// bestEffortWriter forwards writes to w until one fails, then silently
+// discards all further writes. The pipe feeding the logging binary is wrapped
+// in this before it joins the stdio tee of a foreground container: logging is
+// best-effort there, and a dead logging binary (EPIPE once our read ends are
+// closed, see NewContainerIO) must not error the whole tee — that would stop
+// the copier that drains the container's stdio FIFO and deadlock the
+// container. The attach keeps streaming; the log is what goes incomplete.
+// https://github.com/containerd/nerdctl/issues/5137
+type bestEffortWriter struct {
+	w    io.Writer
+	dead bool
+}
+
+func (b *bestEffortWriter) Write(p []byte) (int, error) {
+	// Only ever called from the single stdio copy goroutine of its stream, so
+	// no locking is needed.
+	if !b.dead {
+		if _, err := b.w.Write(p); err != nil {
+			b.dead = true
+			log.L.WithError(err).Warn("writing container output to the logging binary failed; further output will not be logged")
+		}
+	}
+	return len(p), nil
+}
+
 func (c *ncio) Config() cio.Config {
 	return c.config
 }
@@ -156,19 +195,23 @@ func NewContainerIO(namespace string, logURI string, tty bool, stdin io.Reader,
 			if err != nil {
 				return nil, err
 			}
-			closers = append(closers, stdoutr.Close, stdoutw.Close)
+			closeStdoutR := closeOnce(stdoutr)
+			closers = append(closers, closeStdoutR, stdoutw.Close)
 
 			stderrr, stderrw, err := os.Pipe()
 			if err != nil {
 				return nil, err
 			}
-			closers = append(closers, stderrr.Close, stderrw.Close)
+			closeStderrR := closeOnce(stderrr)
+			closers = append(closers, closeStderrR, stderrw.Close)
 
 			r, w, err := os.Pipe()
 			if err != nil {
 				return nil, err
 			}
-			closers = append(closers, r.Close, w.Close)
+			closeR := closeOnce(r)
+			closeW := closeOnce(w)
+			closers = append(closers, closeR, closeW)
 
 			u, err := url.Parse(logURI)
 			if err != nil {
@@ -184,18 +227,44 @@ func NewContainerIO(namespace string, logURI string, tty bool, stdin io.Reader,
 			closers = append(closers, func() error { return cmd.Process.Kill() })
 
 			// close our side of the pipe after start
-			if err := w.Close(); err != nil {
+			if err := closeW(); err != nil {
 				return nil, fmt.Errorf("failed to close write pipe after start: %w", err)
 			}
 
+			// Close our copies of the stdio read ends that were handed to the
+			// logging binary; the child holds its own duplicates via ExtraFiles.
+			// This is the equivalent of containerd's binaryIO.CloseAfterStart.
+			// If this process kept the read ends open, a logging binary that
+			// stops reading (killed, crashed, ...) would never surface as EPIPE
+			// on the tee writes below: the stdio copy goroutine would block
+			// forever on the full pipe, stop draining the container's stdout
+			// FIFO, and deadlock both the container and `nerdctl run` itself
+			// (including `nerdctl rm -f` of the wedged container).
+			// https://github.com/containerd/nerdctl/issues/5137
+			if err := closeStdoutR(); err != nil {
+				return nil, fmt.Errorf("failed to close stdout pipe read end after start: %w", err)
+			}
+			if err := closeStderrR(); err != nil {
+				return nil, fmt.Errorf("failed to close stderr pipe read end after start: %w", err)
+			}
+
 			// wait for the logging binary to be ready
+			// For binary-v2, readiness requires a byte to be written before close.
+			// For binary, EOF is treated as ready for backward compatibility.
 			b := make([]byte, 1)
-			if _, err := r.Read(b); err != nil && err != io.EOF {
+			n, err := r.Read(b)
+			if err != nil && err != io.EOF {
 				return nil, fmt.Errorf("failed to read from logging binary: %w", err)
 			}
+			if u.Scheme == "binary-v2" && n == 0 {
+				return nil, errors.New("logging binary did not call ready (it may have crashed or exited prematurely)")
+			}
+			if err := closeR(); err != nil {
+				return nil, fmt.Errorf("failed to close ready pipe read end: %w", err)
+			}
 
-			stdoutWriters = append(stdoutWriters, stdoutw)
-			stderrWriters = append(stderrWriters, stderrw)
+			stdoutWriters = append(stdoutWriters, &bestEffortWriter{w: stdoutw})
+			stderrWriters = append(stderrWriters, &bestEffortWriter{w: stderrw})
 		}
 
 		streams.Stdout = io.MultiWriter(stdoutWriters...)

pkg/cioutil/container_io_test.go

@@ -0,0 +1,77 @@
+/*
+   Copyright The containerd 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 cioutil
+
+import (
+	"errors"
+	"os"
+	"testing"
+)
+
+type failingWriter struct {
+	calls int
+}
+
+func (f *failingWriter) Write(p []byte) (int, error) {
+	f.calls++
+	return 0, errors.New("broken pipe")
+}
+
+// TestBestEffortWriter verifies that a failing logger pipe never errors the
+// stdio tee: the first failed write disables the writer and every write still
+// reports full success, so the copier draining the container's stdio keeps
+// running. Regression test for
+// https://github.com/containerd/nerdctl/issues/5137
+func TestBestEffortWriter(t *testing.T) {
+	fw := &failingWriter{}
+	b := &bestEffortWriter{w: fw}
+
+	for i := 0; i < 3; i++ {
+		n, err := b.Write([]byte("data"))
+		if err != nil {
+			t.Fatalf("write %d: best-effort writer must not return an error, got %v", i, err)
+		}
+		if n != 4 {
+			t.Fatalf("write %d: expected n=4, got %d", i, n)
+		}
+	}
+	if fw.calls != 1 {
+		t.Fatalf("expected the underlying writer to be abandoned after the first failure, got %d calls", fw.calls)
+	}
+}
+
+// TestBestEffortWriterClosedPipe exercises the real failure mode: writing to
+// an os.Pipe whose read end is closed (EPIPE), as happens when the logging
+// binary dies after our copies of its read ends were closed.
+func TestBestEffortWriterClosedPipe(t *testing.T) {
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	r.Close()
+	defer w.Close()
+
+	b := &bestEffortWriter{w: w}
+	for i := 0; i < 2; i++ {
+		if n, err := b.Write([]byte("data")); err != nil || n != 4 {
+			t.Fatalf("write %d: expected (4, nil), got (%d, %v)", i, n, err)
+		}
+	}
+	if !b.dead {
+		t.Fatal("expected the writer to be marked dead after EPIPE")
+	}
+}

pkg/logging/logging.go

@@ -200,16 +200,37 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi
 	if err != nil {
 		return nil, err
 	}
+	// closeAfterDelivery forwards the first delivery from ch and then closes
+	// the client, so that callers which re-arm the wait (see the wait loop in
+	// loggingProcessAdapter) do not accumulate open clients.
+	closeAfterDelivery := func(ch <-chan containerd.ExitStatus) <-chan containerd.ExitStatus {
+		out := make(chan containerd.ExitStatus, 1)
+		go func() {
+			defer close(out)
+			defer client.Close()
+			if status, ok := <-ch; ok {
+				out <- status
+			}
+		}()
+		return out
+	}
 	con, err := client.LoadContainer(ctx, config.ID)
 	if err != nil {
+		client.Close()
 		return nil, err
 	}
 
 	task, err := con.Task(ctx, nil)
 	if err == nil {
-		return task.Wait(ctx)
+		exitCh, err := task.Wait(ctx)
+		if err != nil {
+			client.Close()
+			return nil, err
+		}
+		return closeAfterDelivery(exitCh), nil
 	}
 	if !errdefs.IsNotFound(err) {
+		client.Close()
 		return nil, err
 	}
 
@@ -232,16 +253,24 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi
 	for {
 		select {
 		case <-ctx.Done():
+			client.Close()
 			return nil, errors.New("timed out waiting for container task to start")
 		case <-ticker.C:
 			task, err = con.Task(ctx, nil)
 			if err == nil {
-				return task.Wait(ctx)
+				exitCh, err := task.Wait(ctx)
+				if err != nil {
+					client.Close()
+					return nil, err
+				}
+				return closeAfterDelivery(exitCh), nil
 			}
 			if !errdefs.IsNotFound(err) {
+				client.Close()
 				return nil, err
 			}
 			if outputSeen() {
+				client.Close()
 				return alreadyExited(), nil
 			}
 		}
@@ -250,6 +279,10 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi
 
 type ContainerWaitFunc func(ctx context.Context, address string, config *logging.Config, outputSeen func() bool) (<-chan containerd.ExitStatus, error)
 
+// containerWaitRetryDelay is how long the logger waits before re-arming the
+// container wait after the wait channel delivered an error instead of an exit.
+const containerWaitRetryDelay = 1 * time.Second
+
 func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config) error {
 	if err := driver.PreProcess(ctx, dataStore, config); err != nil {
 		return err
@@ -374,15 +407,34 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres
 		// keeps the stdio FIFO write ends open (so the container can be
 		// restarted), so the FIFOs may not reach EOF on exit; without this the
 		// read goroutines, and therefore the logger, could block forever.
-		exitCh, err := getContainerWait(ctx, address, config, outputSeen)
-		if err != nil {
-			// We could not determine when the container exits. Do not cancel the
-			// readers: they will finish on their own when the FIFO reaches EOF.
-			// Cancelling here could truncate a still-running container.
-			log.G(ctx).Errorf("failed to get container task wait channel: %v", err)
-			return
+		for {
+			exitCh, err := getContainerWait(ctx, address, config, outputSeen)
+			if err != nil {
+				// We could not determine when the container exits. Do not cancel the
+				// readers: they will finish on their own when the FIFO reaches EOF.
+				// Cancelling here could truncate a still-running container.
+				log.G(ctx).Errorf("failed to get container task wait channel: %v", err)
+				return
+			}
+			status := <-exitCh
+			if status.Error() == nil {
+				// The container has exited.
+				break
+			}
+			// The channel delivered a Wait RPC error, not a container exit:
+			// containerd's client sends Wait failures through the same channel
+			// as a synthetic ExitStatus (client/task.go). Treating that as an
+			// exit would cancel the readers, and with them all logging, while
+			// the container is still running. Re-arm the wait instead.
+			// https://github.com/containerd/nerdctl/issues/5137
+			log.G(ctx).WithError(status.Error()).Warn("error while waiting for container exit; retrying")
+			select {
+			case <-ctx.Done():
+				// SIGTERM: the goroutine above already cancels the readers.
+				return
+			case <-time.After(containerWaitRetryDelay):
+			}
 		}
-		<-exitCh
 		stdoutR.Cancel()
 		stderrR.Cancel()
 	}()