Revert "colrpc: fix tracing when an inbox receives an error"

This reverts commit bd21647.

pkg/sql/colflow/BUILD.bazel

@@ -60,7 +60,6 @@ go_test(
     srcs = [
         "colbatch_scan_test.go",
         "dep_test.go",
-        "draining_test.go",
         "main_test.go",
         "routers_test.go",
         "stats_test.go",

pkg/sql/colflow/colrpc/colrpc_test.go

@@ -643,15 +643,6 @@ func TestOutboxInboxMetadataPropagation(t *testing.T) {
 				return len(meta) == 1 && errors.Is(meta[0].Err, expectedError)
 			},
 			test: func(ctx context.Context, inbox *Inbox) []execinfrapb.ProducerMetadata {
-				defer func() {
-					// Make sure that the error is not propagated for the second
-					// time.
-					//
-					// We still need to drain to simulate what happens in
-					// production - there, the consumer of the inbox would
-					// transition into draining upon receiving the error.
-					require.True(t, len(inbox.DrainMeta()) == 0)
-				}()
 				for {
 					var b coldata.Batch
 					if err := colexecerror.CatchVectorizedRuntimeError(func() {

pkg/sql/colflow/colrpc/inbox.go

@@ -328,17 +328,13 @@ func (i *Inbox) Next() coldata.Batch {
 		return coldata.ZeroBatch
 	}
 
-	var ungracefulStreamTermination bool
 	defer func() {
 		// Catch any panics that occur and close the Inbox in order to not leak
 		// the goroutine listening for context cancellation. The Inbox must
 		// still be closed during normal termination.
 		if panicObj := recover(); panicObj != nil {
-			if ungracefulStreamTermination {
-				// Only close the Inbox here in case of an ungraceful
-				// termination.
-				i.close()
-			}
+			// Only close the Inbox here in case of an ungraceful termination.
+			i.close()
 			err := logcrash.PanicAsError(0, panicObj)
 			log.VEventf(i.Ctx, 1, "Inbox encountered an error in Next: %v", err)
 			// Note that here we use InternalError to propagate the error
@@ -367,37 +363,21 @@ func (i *Inbox) Next() coldata.Batch {
 			// to handle it.
 			err = pgerror.Newf(pgcode.InternalConnectionFailure, "inbox communication error: %s", err)
 			i.errCh <- err
-			ungracefulStreamTermination = true
 			colexecerror.ExpectedError(err)
 		}
 		if len(m.Data.Metadata) != 0 {
-			// If an error was encountered, it needs to be propagated
-			// immediately. All other metadata will simply be buffered and
-			// returned in DrainMeta.
-			var receivedErr error
 			for _, rpm := range m.Data.Metadata {
 				meta, ok := execinfrapb.RemoteProducerMetaToLocalMeta(i.Ctx, rpm)
 				if !ok {
 					continue
 				}
-				if meta.Err != nil && receivedErr == nil {
-					receivedErr = meta.Err
-				} else {
-					// Note that if multiple errors are sent in a single
-					// message, then we'll propagate the first one right away
-					// (via a panic below) and will buffer the rest to be
-					// returned in DrainMeta. The caller will catch the panic
-					// and will transition to draining, so this all works out.
-					//
-					// We choose this way of handling multiple errors rather
-					// than something like errors.CombineErrors() since we want
-					// to keep errors unchanged (e.g. roachpb.ErrPriority() will
-					// be called on each error in the DistSQLReceiver).
-					i.bufferedMeta = append(i.bufferedMeta, meta)
+				if meta.Err != nil {
+					// If an error was encountered, it needs to be propagated
+					// immediately. All other metadata will simply be buffered
+					// and returned in DrainMeta.
+					colexecerror.ExpectedError(meta.Err)
 				}
-			}
-			if receivedErr != nil {
-				colexecerror.ExpectedError(receivedErr)
+				i.bufferedMeta = append(i.bufferedMeta, meta)
 			}
 			// Continue until we get the next batch or EOF.
 			continue

pkg/sql/colflow/colrpc/inbox_test.go

@@ -139,10 +139,8 @@ func TestInboxNextPanicDoesntLeakGoroutines(t *testing.T) {
 	m := &execinfrapb.ProducerMessage{}
 	m.Data.RawBytes = []byte("garbage")
 
-	// Simulate the client (outbox) that sends only a single piece of metadata.
 	go func() {
 		_ = rpcLayer.client.Send(m)
-		_ = rpcLayer.client.CloseSend()
 	}()
 
 	// inbox.Next should panic given that the deserializer will encounter garbage
@@ -152,10 +150,6 @@ func TestInboxNextPanicDoesntLeakGoroutines(t *testing.T) {
 		inbox.Next()
 	})
 
-	// Upon catching the panic and converting it into an error, the caller
-	// transitions to draining.
-	inbox.DrainMeta()
-
 	// We require no error from the stream handler as nothing was canceled. The
 	// panic is bubbled up through the Next chain on the Inbox's host.
 	require.NoError(t, <-streamHandlerErrCh)