transport Rx: error upcall on proto errors; Tx: abort xaction _or_ call back

alex-aizman · alex-aizman · commit 29c4ed1420ef · 2025-11-03T14:57:23.000-05:00
* transport.RecvObj API change: calling now with protocol error
  - filter clean EOF (opcFin)
  - provide an empty header with only sender ID and a nil reader
* Tx: handle stream error differently:
  - use TermedCB XOR abort xaction
* refactor transport/err.go
--------------
* get-batch:
  - add gfnMaxCount=2, improve logging, prep for per-WID abort
-------------
* dsort:
  - check transport.RecvObj() error

Signed-off-by: Alex Aizman &lt;alex.aizman@gmail.com&gt;
diff --git a/cmd/cli/go.mod b/cmd/cli/go.mod
@@ -3,7 +3,7 @@ module github.com/NVIDIA/aistore/cmd/cli
 go 1.25
 
 require (
-	github.com/NVIDIA/aistore v1.4.1-0.20251103150813-c24776bd0c0f
+	github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899
 	github.com/fatih/color v1.18.0
 	github.com/json-iterator/go v1.1.12
 	github.com/onsi/ginkgo/v2 v2.26.0
diff --git a/cmd/cli/go.sum b/cmd/cli/go.sum
@@ -1,8 +1,8 @@
 github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
 github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
-github.com/NVIDIA/aistore v1.4.1-0.20251103150813-c24776bd0c0f h1:iOIIlciARfjdnKZ9izmRIGwEoE9oOIYX0ixs7IPeBjk=
-github.com/NVIDIA/aistore v1.4.1-0.20251103150813-c24776bd0c0f/go.mod h1:+lmRm0p5uEVUZ5KCGwneVKIEu3LiBV7hzId6Rmc0F+M=
+github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899 h1:5O3UWSTuwiGhaYluU9X8z3vXyj98M2mXTbYhTLEJI6g=
+github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899/go.mod h1:+lmRm0p5uEVUZ5KCGwneVKIEu3LiBV7hzId6Rmc0F+M=
 github.com/OneOfOne/xxhash v1.2.8 h1:31czK/TI9sNkxIKfaUfGlU47BAxQ0ztGgd9vPyqimf8=
 github.com/OneOfOne/xxhash v1.2.8/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
diff --git a/cmd/ishard/go.mod b/cmd/ishard/go.mod
@@ -3,7 +3,7 @@ module github.com/NVIDIA/aistore/cmd/ishard
 go 1.25
 
 require (
-	github.com/NVIDIA/aistore v1.4.1-0.20251024202506-aaa8f8953f96
+	github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899
 	github.com/json-iterator/go v1.1.12
 	github.com/vbauerster/mpb/v4 v4.12.2
 )
diff --git a/cmd/ishard/go.sum b/cmd/ishard/go.sum
@@ -1,7 +1,7 @@
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
 github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=
-github.com/NVIDIA/aistore v1.4.1-0.20251024202506-aaa8f8953f96 h1:glOli5bvdW2DUYQw85o+Vzm15pS5CbkzRjNFS2ZW8bA=
-github.com/NVIDIA/aistore v1.4.1-0.20251024202506-aaa8f8953f96/go.mod h1:+lmRm0p5uEVUZ5KCGwneVKIEu3LiBV7hzId6Rmc0F+M=
+github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899 h1:5O3UWSTuwiGhaYluU9X8z3vXyj98M2mXTbYhTLEJI6g=
+github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899/go.mod h1:+lmRm0p5uEVUZ5KCGwneVKIEu3LiBV7hzId6Rmc0F+M=
 github.com/OneOfOne/xxhash v1.2.8 h1:31czK/TI9sNkxIKfaUfGlU47BAxQ0ztGgd9vPyqimf8=
 github.com/OneOfOne/xxhash v1.2.8/go.mod h1:eZbhyaAYD41SGSSsnmcpxVoRiQ/MPUTjUdIIOT9Um7Q=
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
diff --git a/ext/dsort/dsort_general.go b/ext/dsort/dsort_general.go
@@ -408,16 +408,17 @@ func (ds *dsorterGeneral) recvReq(hdr *transport.ObjHdr, objReader io.Reader, er
 		transport.FreeRecv(objReader)
 	}()
 	req := remoteRequest{}
-	if err := jsoniter.Unmarshal(hdr.Opaque, &req); err != nil {
-		err := fmt.Errorf(cmn.FmtErrUnmarshal, apc.ActDsort, "recv request", cos.BHead(hdr.Opaque), err)
+	if errM := jsoniter.Unmarshal(hdr.Opaque, &req); errM != nil {
+		if err == nil {
+			err = fmt.Errorf(cmn.FmtErrUnmarshal, apc.ActDsort, "recv request", cos.BHead(hdr.Opaque), errM)
+		}
 		ds.m.abort(err)
 		return err
 	}
 
 	fromNode := ds.m.smap.GetTarget(hdr.SID)
 	if fromNode == nil {
-		err := fmt.Errorf("received request (%v) from %q not present in the %s", req.Record, hdr.SID, ds.m.smap)
-		return err
+		return fmt.Errorf("received request (%v) from %q not present in the %s", req.Record, hdr.SID, ds.m.smap)
 	}
 
 	if err != nil {
diff --git a/transport/api.go b/transport/api.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/base.go b/transport/base.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
@@ -183,14 +182,14 @@ func (s *streamBase) startSend(streamable fmt.Stringer) (err error) {
 	return
 }
 
-func (s *streamBase) newErr(detail string) error {
+func (s *streamBase) newErr(ctx string) error {
 	reason, errT := s.TermInfo()
 	return &ErrStreamTerm{
 		err:    errT,
 		dst:    s.dstID,
 		loghdr: s.loghdr,
 		reason: reason,
-		detail: detail,
+		ctx:    ctx,
 	}
 }
 
@@ -306,16 +305,17 @@ func (s *streamBase) sendLoop(config *cmn.Config, dryrun bool) {
 		return
 	}
 
-	// termination is caused by anything other than Fin()
-	// (reasonStopped is, effectively, abort via Stop() - totally legit)
-	var errExt error
-	if reason != reasonStopped {
-		errExt = fmt.Errorf("%s[term-reason: %s, err: %w]", s, reason, err)
-		nlog.Errorln(errExt)
+	// termination is caused by anything other than Fin() ---------------
+	// 1) reasonStopped via Stop(), or
+	// 2) broken pipe, connection reset, etc.
+	// steps:
+	// - abort parent xaction if defined AND the parent's TermedCB is nil
+	// - wait and complete
+	// - call parent's TermedCB if defined ---------- (notice "either/OR")
 
-		// NOTE: aborting grandparent xaction
-		if s.parent != nil && s.parent.Xact != nil {
-			s.parent.Xact.Abort(errExt)
+	if reason != reasonStopped {
+		if s.parent != nil && s.parent.Xact != nil && s.parent.TermedCB == nil {
+			s.parent.Xact.Abort(s.newErr(""))
 		}
 	}
 
@@ -325,10 +325,9 @@ func (s *streamBase) sendLoop(config *cmn.Config, dryrun bool) {
 	// cleanup
 	s.streamer.abortPending(err, false /*completions*/)
 
-	// notify parent if defined
 	if reason != reasonStopped {
 		if s.parent != nil && s.parent.TermedCB != nil {
-			s.parent.TermedCB(s.dstID, errExt)
+			s.parent.TermedCB(s.dstID, err)
 		}
 	}
 
diff --git a/transport/client_fasthttp.go b/transport/client_fasthttp.go
@@ -1,7 +1,6 @@
 //go:build !nethttp
 
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/client_nethttp.go b/transport/client_nethttp.go
@@ -1,7 +1,6 @@
 //go:build nethttp
 
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/collect.go b/transport/collect.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/err.go b/transport/err.go
@@ -1,12 +1,13 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
 package transport
 
 import (
-	"fmt"
+	"strings"
+
+	"github.com/NVIDIA/aistore/cmn/debug"
 )
 
 // ErrSBR codes (receive-side stream breakage)
@@ -25,22 +26,23 @@ const (
 	sbrPDUDataSize    = "sbr_pdu_data_size"   // PDU size mismatch
 )
 
+// (do not wrap %w these errors; use Unwrap)
 type (
 	// Tx
 	ErrStreamTerm struct {
 		err    error
 		loghdr string
-		dst    string // destination node ID
+		dst    string // destination node ID // TODO: needed?
 		reason string
-		detail string
+		ctx    string
 	}
 	// Rx
 	ErrSBR struct {
 		err    error
 		loghdr string
 		sid    string // sender node ID
 		code   string
-		detail string
+		ctx    string
 	}
 )
 
@@ -49,34 +51,61 @@ type (
 ///////////////////
 
 func (e *ErrStreamTerm) Error() string {
-	return fmt.Sprintf("%s terminated [dst: %s, reason: %s, detail: %s, err: %v]", e.loghdr, e.dst, e.reason, e.detail, e.err)
+	debug.Assert(e.err != nil)
+	debug.Assert(strings.Contains(e.loghdr, e.dst), e.loghdr, " vs ", e.dst)
+
+	var sb strings.Builder
+	sb.Grow(256)
+	sb.WriteString(e.loghdr)
+	sb.WriteString(" terminated [reason: '")
+	sb.WriteString(e.reason)
+	sb.WriteString("' ")
+	if e.ctx != "" {
+		sb.WriteString("ctx: ")
+		sb.WriteString(e.ctx)
+		sb.WriteByte(' ')
+	}
+	sb.WriteString("err: ")
+	sb.WriteString(e.err.Error())
+	sb.WriteByte(']')
+	return sb.String()
 }
 
 func (e *ErrStreamTerm) Unwrap() error { return e.err }
 
-// func (e *ErrStreamTerm) DestID() string { return e.dst } // uncomment if needed
-
-func IsErrStreamTerm(err error) bool {
-	_, ok := err.(*ErrStreamTerm)
-	return ok
-}
-
 ////////////
 // ErrSBR //
 ////////////
 
 func (e *ErrSBR) Error() string {
-	if e.err == nil {
-		return fmt.Sprintf("%s %s: [sender: %s, detail: %s]", e.loghdr, e.code, e.sid, e.detail)
+	debug.Assert(strings.Contains(e.loghdr, e.sid), e.loghdr, " vs ", e.sid)
+	var sb strings.Builder
+	sb.Grow(256)
+	sb.WriteString(e.loghdr)
+	sb.WriteByte(' ')
+	sb.WriteString(e.code)
+	sb.WriteString(":[")
+	if e.ctx != "" {
+		sb.WriteString("ctx: ")
+		sb.WriteString(e.ctx)
+	}
+	if e.err != nil {
+		if e.ctx != "" {
+			sb.WriteByte(' ')
+		}
+		sb.WriteString("err: ")
+		sb.WriteString(e.err.Error())
 	}
-	return fmt.Sprintf("%s %s: [sender: %s, detail: %s, err: %v]", e.loghdr, e.code, e.sid, e.detail, e.err)
+	sb.WriteByte(']')
+	return sb.String()
 }
 
 func (e *ErrSBR) Unwrap() error { return e.err }
+func (e *ErrSBR) SID() string   { return e.sid }
 
-// func (e *ErrSBR) SID() string   { return e.sid } // uncomment if needed
-
-func IsErrSBR(err error) bool {
-	_, ok := err.(*ErrSBR)
-	return ok
+func AsErrSBR(err error) *ErrSBR {
+	if e, ok := err.(*ErrSBR); ok {
+		return e
+	}
+	return nil
 }
diff --git a/transport/handlers.go b/transport/handlers.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/header.go b/transport/header.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
diff --git a/transport/obj_test.go b/transport/obj_test.go
@@ -1,5 +1,4 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
  * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
@@ -202,7 +201,7 @@ func sendText(stream *transport.Stream, txt1, txt2 string) {
 
 func Example_obj() {
 	receive := func(hdr *transport.ObjHdr, objReader io.Reader, err error) error {
-		cos.Assert(err == nil) // TODO: (ref 020943)
+		cos.Assert(err == nil)
 		object, err := cos.ReadAll(objReader)
 		if err != nil {
 			panic(err)
@@ -368,7 +367,7 @@ func TestObjAttrs(t *testing.T) {
 
 	var receivedCount atomic.Int64
 	recvFunc := func(hdr *transport.ObjHdr, objReader io.Reader, err error) error {
-		cos.Assert(err == nil) // TODO; (ref 020943)
+		cos.Assert(err == nil)
 		idx := hdr.Opaque[0]
 		cos.AssertMsg(hdr.Bck.IsAIS(), "expecting ais bucket")
 		cos.Assertf(reflect.DeepEqual(testAttrs[idx], hdr.ObjAttrs),
@@ -419,7 +418,7 @@ func TestObjAttrs(t *testing.T) {
 }
 
 func receive10G(hdr *transport.ObjHdr, objReader io.Reader, err error) error {
-	cos.Assert(err == nil || cos.IsAnyEOF(err)) // TODO: (ref 020943)
+	cos.Assert(err == nil || cos.IsAnyEOF(err))
 	written, _ := io.Copy(io.Discard, objReader)
 	cos.Assert(written == hdr.ObjAttrs.Size)
 	return nil
@@ -540,7 +539,7 @@ func TestCompletionCount(t *testing.T) {
 	)
 
 	receive := func(hdr *transport.ObjHdr, objReader io.Reader, err error) error {
-		cos.Assert(err == nil) // TODO: (ref 020943)
+		cos.Assert(err == nil)
 		written, _ := io.Copy(io.Discard, objReader)
 		cos.Assert(written == hdr.ObjAttrs.Size)
 		numReceived.Inc()
@@ -675,7 +674,7 @@ func streamWriteUntil(t *testing.T, ii int, wg *sync.WaitGroup, ts *httptest.Ser
 func makeRecvFunc(t *testing.T) (*int64, transport.RecvObj) {
 	totalReceived := new(int64)
 	return totalReceived, func(hdr *transport.ObjHdr, objReader io.Reader, err error) error {
-		cos.Assert(err == nil || cos.IsAnyEOF(err)) // TODO: (ref 020943)
+		cos.Assert(err == nil || cos.IsAnyEOF(err))
 		written, err := io.Copy(io.Discard, objReader)
 		if err != nil && !cos.IsOkEOF(err) {
 			tassert.CheckFatal(t, err)
diff --git a/transport/pdu.go b/transport/pdu.go
@@ -1,7 +1,6 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
- * Copyright (c) 2018-2024, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
 package transport
 
diff --git a/transport/pool.go b/transport/pool.go
@@ -1,7 +1,6 @@
-// Package transport provides long-lived http/tcp connections for
-// intra-cluster communications (see README for details and usage example).
+// Package transport provides long-lived http/tcp connections for intra-cluster communications
 /*
- * Copyright (c) 2018-2024, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
 package transport
 
diff --git a/transport/recv.go b/transport/recv.go
diff --git a/transport/sendobj.go b/transport/sendobj.go
diff --git a/transport/stream_bundle_test.go b/transport/stream_bundle_test.go
diff --git a/transport/tinit.go b/transport/tinit.go
diff --git a/xact/xs/moss.go b/xact/xs/moss.go

Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@ module github.com/NVIDIA/aistore/cmd/ishard`
`3`	`3`	`go 1.25`
`4`	`4`
`5`	`5`	`require (`
`6`		`- github.com/NVIDIA/aistore v1.4.1-0.20251024202506-aaa8f8953f96`
	`6`	`+ github.com/NVIDIA/aistore v1.4.1-0.20251103154500-92dbe79ff899`
`7`	`7`	`github.com/json-iterator/go v1.1.12`
`8`	`8`	`github.com/vbauerster/mpb/v4 v4.12.2`
`9`	`9`	`)`