fix(streaming): decouple stream ctx timeout from per-recv timeout to fix ttstream connection reuse

Previously, RecvMsg created a child context from the stream ctx for recv timeout.
This coupled stream-level timeout with per-recv timeout, causing incorrect behavior:
- When DisableCancelRemote=true, stream ctx expiration was misidentified as recv timeout,
  preventing the RST frame from being sent and breaking connection reuse.
- When stream ctx expired before recv timeout, the error was reported as recv timeout
  instead of stream timeout.

Changes:
- Redesign Pipe to support two-level ctx: a pipe-level ctx (stream lifecycle) and a
  per-read ctx (individual recv timeout), each with its own callback
- Separate stream timeout (error 12015) from recv timeout (error 12014) so they trigger
  independent close/cancel logic
- Fix Pipe.ReadCtx to drain remaining items after detecting closed state, preventing
  data loss in Write+Close race
- Move container package to internal to prevent external dependency

Author: DMwangnima

pkg/remote/trans/ttstream/client_handler.go

@@ -18,6 +18,7 @@ package ttstream
 
 import (
 	"context"
+	"time"
 
 	"github.com/bytedance/gopkg/cloud/metainfo"
 	"github.com/cloudwego/gopkg/protocol/ttheader"
@@ -74,6 +75,10 @@ func (c clientTransHandler) NewStream(ctx context.Context, ri rpcinfo.RPCInfo) (
 	}
 	strHeader[ttheader.HeaderIDLServiceName] = invocation.ServiceName()
 	metainfo.SaveMetaInfoToMap(ctx, strHeader)
+	var tm time.Duration
+	if ddl, ok := ctx.Deadline(); ok {
+		tm = time.Until(ddl)
+	}
 
 	trans, err := c.transPool.Get(addr.Network(), addr.String())
 	if err != nil {
@@ -87,6 +92,7 @@ func (c clientTransHandler) NewStream(ctx context.Context, ri rpcinfo.RPCInfo) (
 	cs.setRecvTimeoutConfig(rconfig)
 	cs.setMetaFrameHandler(c.metaHandler)
 	cs.setTraceController(c.traceCtl)
+	cs.setStreamTimeout(tm)
 
 	if err = trans.WriteStream(ctx, cs, intHeader, strHeader); err != nil {
 		return nil, err

pkg/remote/trans/ttstream/client_trans_pool_longconn.go

@@ -19,7 +19,7 @@ package ttstream
 import (
 	"time"
 
-	"github.com/cloudwego/kitex/pkg/remote/trans/ttstream/container"
+	"github.com/cloudwego/kitex/pkg/remote/trans/ttstream/internal/container"
 )
 
 var DefaultLongConnConfig = LongConnConfig{

pkg/remote/trans/ttstream/container/pipe.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/cloudwego/kitex/pkg/kerrors"
 	"github.com/cloudwego/kitex/pkg/streaming"
@@ -49,6 +50,10 @@ func newStreamRecvTimeoutException(cfg streaming.TimeoutConfig) *Exception {
 	return newException(fmt.Sprintf("stream Recv timeout, timeout config=%+v", cfg), kerrors.ErrStreamingTimeout, 12014).withSide(clientSide)
 }
 
+func newStreamTimeoutException(tm time.Duration) *Exception {
+	return newException(fmt.Sprintf("stream timeout, timeout in ctx: %+v", tm), kerrors.ErrStreamingTimeout, 12015).withSide(clientSide)
+}
+
 const (
 	setSide = 1 << iota
 	setCancelPath