From a675d27f0345911bd0956730a524a0606d5b462b Mon Sep 17 00:00:00 2001
From: AdaAibaby <8872412+adababys@users.noreply.github.com>
Date: Thu, 22 Oct 2026 11:35:26 +0800
Subject: [PATCH] fix(nbd): downgrade broken-pipe reply errors from ERROR to
 WARN on shutdown

When a sandbox VM is torn down, its NBD reply socket closes while in-flight
response goroutines are still writing to it, so those writes fail with EPIPE.
The first failure is handed to the read loop via d.fatal; the remaining
concurrent ones fall into completeAsync default branch and were logged at
ERROR, producing a burst of ERROR-level noise on every normal shutdown and
masking real failures.

Add isSocketClosed (EPIPE / io.ErrClosedPipe) and log that expected burst at
WARN in the completeAsync default branch; genuinely unexpected reply-path
errors stay at ERROR. This is the reply-socket counterpart of the read-side
ErrObjectNotExist downgrade.

Rebased onto current main: the dispatcher was refactored since the earlier
attempt (#3296), which consolidated the three per-command default branches
(cmdRead/cmdWrite/cmdWriteZeroes) into the single completeAsync reply path, so
the downgrade now lives in one place.
---
 .../orchestrator/pkg/sandbox/nbd/dispatch.go  | 23 +++++++++++--
 .../pkg/sandbox/nbd/dispatch_epipe_test.go    | 34 +++++++++++++++++++
 2 files changed, 55 insertions(+), 2 deletions(-)
 create mode 100644 packages/orchestrator/pkg/sandbox/nbd/dispatch_epipe_test.go

diff --git a/packages/orchestrator/pkg/sandbox/nbd/dispatch.go b/packages/orchestrator/pkg/sandbox/nbd/dispatch.go
index 5e3911068a..a935cd2929 100644
--- a/packages/orchestrator/pkg/sandbox/nbd/dispatch.go
+++ b/packages/orchestrator/pkg/sandbox/nbd/dispatch.go
@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"syscall"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -227,12 +228,30 @@ func (d *Dispatch) completeAsync(ctx context.Context, cmd command, perform func(
 		select {
 		case d.fatal <- err:
 		default:
-			d.logger.Error(ctx, "nbd command failed while the dispatcher was already failing",
-				append(cmd.fields(), zap.Error(err))...)
+			// A failure is already on its way to the read loop. When the sandbox VM
+			// is torn down, its reply socket closes and the remaining in-flight
+			// responses all fail with EPIPE at once; that burst is expected on a
+			// normal shutdown, so log it at WARN. Anything else here is a genuine,
+			// unexpected reply-path failure and stays at ERROR.
+			if isSocketClosed(err) {
+				d.logger.Warn(ctx, "nbd reply dropped: socket already closed while dispatcher was failing",
+					append(cmd.fields(), zap.Error(err))...)
+			} else {
+				d.logger.Error(ctx, "nbd command failed while the dispatcher was already failing",
+					append(cmd.fields(), zap.Error(err))...)
+			}
 		}
 	}()
 }
 
+// isSocketClosed reports whether err is the NBD reply socket being closed
+// by the peer: EPIPE (writing to a closed socket) or io.ErrClosedPipe. Both
+// are the expected outcome when the sandbox VM is torn down with responses
+// still in flight, and are logged at WARN rather than ERROR.
+func isSocketClosed(err error) bool {
+	return errors.Is(err, syscall.EPIPE) || errors.Is(err, io.ErrClosedPipe)
+}
+
 func (d *Dispatch) Drain() {
 	d.shuttingDownLock.Lock()
 	d.shuttingDown = true
diff --git a/packages/orchestrator/pkg/sandbox/nbd/dispatch_epipe_test.go b/packages/orchestrator/pkg/sandbox/nbd/dispatch_epipe_test.go
new file mode 100644
index 0000000000..d2881fd760
--- /dev/null
+++ b/packages/orchestrator/pkg/sandbox/nbd/dispatch_epipe_test.go
@@ -0,0 +1,34 @@
+//go:build linux
+
+package nbd
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"syscall"
+	"testing"
+)
+
+func TestIsSocketClosed(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"epipe", syscall.EPIPE, true},
+		{"epipe wrapped", fmt.Errorf("write unix @-: %w", syscall.EPIPE), true},
+		{"closed pipe", io.ErrClosedPipe, true},
+		{"closed pipe wrapped", fmt.Errorf("writeResponse: %w", io.ErrClosedPipe), true},
+		{"other", errors.New("some backend error"), false},
+		{"econnreset", syscall.ECONNRESET, false},
+		{"nil", nil, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if got := isSocketClosed(c.err); got != c.want {
+				t.Fatalf("isSocketClosed(%v) = %v, want %v", c.err, got, c.want)
+			}
+		})
+	}
+}