container: normalize closed CRI pipe read errors (#2560)

Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>

internal/headers/cri/pipe.go

@@ -159,6 +159,15 @@ func (c *pipeConn) Read(p []byte) (int, error) {
 		}
 	})
 	if err != nil {
+		// Control can reject a closed file before invoking the callback.
+		c.mu.Lock()
+		closed := c.closed
+		c.mu.Unlock()
+
+		if closed {
+			err = os.ErrClosed
+		}
+
 		return 0, err
 	}
 
@@ -200,6 +209,8 @@ func (c *pipeConn) SetDeadline(t time.Time) error { return c.SetWriteDeadline(t)
 type readAheadConn struct {
 	net.Conn
 
+	readWG sync.WaitGroup
+
 	mu   sync.Mutex
 	cond *sync.Cond
 	buf  []byte
@@ -210,7 +221,7 @@ func newReadAheadConn(conn net.Conn) *readAheadConn {
 	c := &readAheadConn{Conn: conn}
 	c.cond = sync.NewCond(&c.mu)
 
-	go c.readLoop()
+	c.readWG.Go(c.readLoop)
 
 	return c
 }
@@ -255,6 +266,14 @@ func (c *readAheadConn) Read(p []byte) (int, error) {
 	return n, nil
 }
 
+// Close cancels the read and waits for the background reader to exit.
+func (c *readAheadConn) Close() error {
+	err := c.Conn.Close()
+	c.readWG.Wait()
+
+	return err
+}
+
 // SetDeadline only applies to writes. A read deadline would stop the background reader for good.
 func (c *readAheadConn) SetDeadline(t time.Time) error {
 	return c.SetWriteDeadline(t)

internal/headers/cri/pipe_internal_test.go

@@ -26,6 +26,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/stretchr/testify/require"
@@ -280,6 +281,11 @@ func TestPipeConnClose(t *testing.T) {
 	case <-time.After(5 * time.Second):
 		t.Fatal("Close did not release the pipe handle")
 	}
+
+	// A read started after Close must return the same error as a canceled read.
+	n, err := server.Read(make([]byte, 1))
+	require.Zero(t, n)
+	require.ErrorIs(t, err, os.ErrClosed)
 }
 
 // Both peers write first. On a pipe without buffer, this blocks both writes
@@ -394,3 +400,64 @@ func TestReadAheadConnClose(t *testing.T) {
 		t.Fatal("pending read was not unblocked by Close")
 	}
 }
+
+// delayedReadConn keeps a canceled read active until the test releases it.
+type delayedReadConn struct {
+	net.Conn
+
+	started  chan struct{}
+	canceled chan struct{}
+	finish   <-chan struct{}
+}
+
+func (c *delayedReadConn) Read(p []byte) (int, error) {
+	close(c.started)
+
+	n, err := c.Conn.Read(p)
+	close(c.canceled)
+	<-c.finish
+
+	return n, err
+}
+
+func TestReadAheadConnCloseWaitsForReader(t *testing.T) {
+	t.Parallel()
+
+	synctest.Test(t, func(t *testing.T) {
+		client, server := net.Pipe()
+		defer server.Close()
+
+		finish := make(chan struct{})
+
+		release := sync.OnceFunc(func() { close(finish) })
+		defer release()
+
+		reader := &delayedReadConn{
+			Conn:     client,
+			started:  make(chan struct{}),
+			canceled: make(chan struct{}),
+			finish:   finish,
+		}
+		conn := newReadAheadConn(reader)
+		<-reader.started
+
+		closed := make(chan error, 1)
+
+		go func() { closed <- conn.Close() }()
+
+		<-reader.canceled
+		synctest.Wait()
+
+		select {
+		case <-closed:
+			t.Fatal("Close returned before the background read finished")
+		default:
+		}
+
+		release()
+		require.NoError(t, <-closed)
+
+		_, err := conn.Read(make([]byte, 1))
+		require.ErrorIs(t, err, io.ErrClosedPipe)
+	})
+}