drpcwire,drpcstream: return the real error from a preempted send

A send blocked on backpressure used to return an opaque "interrupted"
sentinel from WriteFrame, which the data path then remapped through
CheckCancelError. WriteFrame lives in drpcwire and cannot know why a send was
stopped, so it could only return a generic error and leave the stream to
recover the cause.

Pass the stream's send signal to WriteFrame directly and have it return that
signal's error when the signal fires. The send signal is the write side's own
"stop sending" signal, and terminate() always sets it, so a parked send is
woken on every termination and returns send.Err(), which is io.EOF in the
cancel and error cases. That is the same error the write loop's guard already
returns when send is set before a frame is sent, so the parked and unparked
paths now return the same thing.

This is why the CheckCancelError remap on the write path is removed. It
existed to turn the sentinel into the cancellation cause, but the write path
now returns io.EOF, which is what gRPC reports on send; the real cause reaches
the caller through the receive side, not the send side. Every cancel sets the
send signal under the same lock, so the loop guard returns io.EOF before
WriteFrame is even reached, and a send woken while parked returns it too. The
only WriteFrame errors left are genuine transport failures, which happen only
when the stream was not terminated, so there is no cancel cause to substitute
and the raw error is correct.

Passing a signal instead of a channel also lets WriteFrame resolve its channel
lazily, only when a producer actually parks, so a stream that never hits
backpressure no longer allocates that channel on every send.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: shubhamdhama

drpcstream/stream.go

@@ -405,8 +405,13 @@ func (s *Stream) rawWriteLocked(kind drpcwire.Kind, data []byte) (err error) {
 		drpcopts.GetStreamStats(&s.opts.Internal).AddWritten(uint64(len(fr.Data)))
 		s.log("SEND", fr.String)
 
-		if err := s.wr.WriteFrame(fr, s.sigs.term.Signal()); err != nil {
-			return s.CheckCancelError(errs.Wrap(err))
+		// Pass the send signal, the write side's own stop signal, so a parked
+		// WriteFrame is woken when sending ends (cancel, error, close) and
+		// returns that signal's error directly. That is io.EOF in the cancel and
+		// error cases, the same error the loop guard above returns, so the parked
+		// and unparked paths agree and there is nothing to remap here.
+		if err := s.wr.WriteFrame(fr, &s.sigs.send); err != nil {
+			return err
 		} else if fr.Done {
 			return nil
 		}

drpcstream/stream_test.go

@@ -487,7 +487,10 @@ func TestStream_SendCancelPreemptsBlockedSend(t *testing.T) {
 	}
 	select {
 	case err := <-sendDone:
-		assert.Error(t, err)
+		// The preempted send reports the send signal's error (io.EOF), the same
+		// as a send attempted after termination, per the gRPC convention. The
+		// cancellation cause is available from the receive side.
+		assert.Equal(t, err, io.EOF)
 	case <-time.After(5 * time.Second):
 		t.Fatal("blocked send was not interrupted by SendCancel")
 	}

drpcwire/mux_writer.go

@@ -7,18 +7,13 @@ import (
 	"io"
 	"sync"
 
-	"github.com/zeebo/errs"
+	"storj.io/drpc/drpcsignal"
 )
 
 // defaultBufferCapacity is the initial capacity of the pending and in-flight
 // write buffers.
 var defaultBufferCapacity = 4096
 
-// errInterrupted is returned by WriteFrame when its cancel channel fires while
-// it is blocked on backpressure. The caller maps it to the appropriate stream
-// error (e.g. via CheckCancelError).
-var errInterrupted = errs.New("sending frames interrupted")
-
 // WriterOptions controls configuration settings for a MuxWriter.
 type WriterOptions struct {
 	// MaximumBufferSize is the high-water mark for the pending write buffer.
@@ -135,11 +130,13 @@ func (mw *MuxWriter) run() {
 
 // WriteFrame appends fr to the pending buffer. If the buffer is at its
 // high-water mark it blocks until run frees space, cancel fires, or the writer
-// is stopped. cancel is the caller's termination channel (e.g. a stream's term
-// signal); when it fires WriteFrame returns errInterrupted. A control-bit frame
-// is appended immediately even past the high-water mark, so an abortive cancel
-// is never delayed by backpressure.
-func (mw *MuxWriter) WriteFrame(fr Frame, cancel <-chan struct{}) (err error) {
+// is stopped. cancel is the caller's termination signal (e.g. a stream's send
+// signal); when it fires WriteFrame stops waiting and returns cancel.Err(), so
+// the caller gets the termination cause directly without interpreting a
+// sentinel. cancel may be nil to wait indefinitely for space. A control-bit
+// frame is appended immediately even past the high-water mark, so an abortive
+// cancel is never delayed by backpressure.
+func (mw *MuxWriter) WriteFrame(fr Frame, cancel *drpcsignal.Signal) (err error) {
 	for {
 		mw.mu.Lock()
 		if mw.closed {
@@ -164,17 +161,25 @@ func (mw *MuxWriter) WriteFrame(fr Frame, cancel <-chan struct{}) (err error) {
 		mw.blocked++
 		mw.mu.Unlock()
 
+		// Resolve the cancel channel lazily, only now that we are parking, so the
+		// common non-blocking path never forces the signal's channel to be
+		// allocated.
+		var cancelCh <-chan struct{}
+		if cancel != nil {
+			cancelCh = cancel.Signal()
+		}
+
 		select {
 		case <-ch:
 			// Space may be available now; loop and re-check.
 			mw.mu.Lock()
 			mw.blocked--
 			mw.mu.Unlock()
-		case <-cancel:
+		case <-cancelCh:
 			mw.mu.Lock()
 			mw.blocked--
 			mw.mu.Unlock()
-			return errInterrupted
+			return cancel.Err()
 		case <-mw.done:
 			mw.mu.Lock()
 			mw.blocked--

drpcwire/mux_writer_test.go

@@ -12,6 +12,8 @@ import (
 	"time"
 
 	"github.com/zeebo/assert"
+
+	"storj.io/drpc/drpcsignal"
 )
 
 // blockingWriter blocks in Write until unblock is closed, then returns err.
@@ -399,22 +401,22 @@ func TestMuxWriter_WriteFrameBlocksUntilDrain(t *testing.T) {
 	<-mw.Done()
 }
 
-// A parked WriteFrame returns errInterrupted when its cancel channel fires.
+// A parked WriteFrame returns the cancel signal's error when it fires.
 func TestMuxWriter_WriteFrameCanceledWhileBlocked(t *testing.T) {
 	bw := newBlockingWriter()
 	mw := newTinyMuxWriter(bw)
 	blockUntilFull(t, mw, bw)
 
-	cancel := make(chan struct{})
+	var cancel drpcsignal.Signal
 	done := make(chan error, 1)
-	go func() { done <- mw.WriteFrame(RandFrame(), cancel) }()
+	go func() { done <- mw.WriteFrame(RandFrame(), &cancel) }()
 	assertBlocked(t, done)
 
-	close(cancel)
+	cancelErr := errors.New("canceled")
+	cancel.Set(cancelErr)
 	select {
 	case err := <-done:
-		assert.Error(t, err)
-		assert.Equal(t, err.Error(), errInterrupted.Error())
+		assert.Equal(t, err, cancelErr)
 	case <-time.After(5 * time.Second):
 		t.Fatal("cancel did not unblock WriteFrame")
 	}