googleapis
diff --git a/Collapse file
‎bigtable/internal/transport/debug_tracer.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/debug_tracer.go
+11Lines changed: 11 additions & 0 deletions b/Collapse file
‎bigtable/internal/transport/debug_tracer.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/debug_tracer.go
+11Lines changed: 11 additions & 0 deletions
diff --git a/Collapse file
‎bigtable/internal/transport/session.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/session.go
+65-9Lines changed: 65 additions & 9 deletions b/Collapse file
‎bigtable/internal/transport/session.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/session.go
+65-9Lines changed: 65 additions & 9 deletions
diff --git a/Collapse file
‎bigtable/internal/transport/session_debug.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/session_debug.go
+18-1Lines changed: 18 additions & 1 deletion b/Collapse file
‎bigtable/internal/transport/session_debug.go‎
Copy file name to clipboardExpand all lines: bigtable/internal/transport/session_debug.go
+18-1Lines changed: 18 additions & 1 deletion
@@ -93,13 +93,24 @@ const (
 	tagSessionHeartbeatMissed        = "session_heartbeat_missed"
 	tagSessionForceCloseNeverStarted = "session_force_close_never_started"
 	tagSessionCloseNoReason          = "session_close_no_reason"
+	// tagSessionReadLoopPanic fires when readLoop's deferred recover
+	// catches a panic from handleSessionResponse (or any downstream
+	// handler). Session is force-closed with REASON_ERROR carrying the
+	// panic value in the description.
+	tagSessionReadLoopPanic = "session_read_loop_panic"
 
 	// vRPC dispatch observations.
 	tagSessionVRPCNil                = "session_vrpc_nil"
 	tagSessionVRPCErrorNil           = "session_vrpc_error_nil"
 	tagSessionVRPCIDMismatch         = "session_vrpc_id_mismatch"
 	tagSessionVRPCResponseWrongState = "session_vrpc_response_wrong_state"
 	tagSessionVRPCDuplicateResult    = "session_vrpc_duplicate_result"
+	// tagSessionVRPCCancelledDrained fires when a server response finally
+	// arrives for an rpc whose caller already returned via ctx.Done: the
+	// drain succeeds, currentCancel != nil, and no one is waiting on
+	// resultChan. Bookkeeping-only — the drain still fires OnSlotDrained
+	// so the pool re-enqueues the session.
+	tagSessionVRPCCancelledDrained = "session_vrpc_cancelled_drained"
 
 	// Pool-scoped anomalies.
 	tagSessionPoolStuckSessionSwept = "session_pool_stuck_session_swept"
 
@@ -61,11 +61,19 @@ type Stream interface {
 
 // SessionHooks holds optional lifecycle callbacks. Nil fields are skipped.
 // Hooks must not block.
+//
+// OnSlotDrained fires on every successful drainSlot on the wire (normal
+// deliver, cancelled-drain, Send-failure Invoke branch). It is the sole
+// "session became free" signal — consumers use it to re-enqueue the
+// session in its AFE idle queue and wake one parked Checkout waiter.
+// cancelActiveRPCs (session teardown) intentionally does NOT fire it —
+// OnClosing/OnClose handle removal from routing structures on that path.
 type SessionHooks struct {
-	OnStart   func(ctx context.Context)
-	OnActive  func(s *Session)
-	OnClosing func(s *Session)
-	OnClose   func(s *Session, err error)
+	OnStart       func(ctx context.Context)
+	OnActive      func(s *Session)
+	OnSlotDrained func()
+	OnClosing     func(s *Session)
+	OnClose       func(s *Session, err error)
 }
 
 func (h SessionHooks) onStart(ctx context.Context) {
@@ -80,6 +88,12 @@ func (h SessionHooks) onActive(s *Session) {
 	}
 }
 
+func (h SessionHooks) onSlotDrained() {
+	if h.OnSlotDrained != nil {
+		h.OnSlotDrained()
+	}
+}
+
 func (h SessionHooks) onClosing(s *Session) {
 	if h.OnClosing != nil {
 		h.OnClosing(s)
@@ -159,6 +173,12 @@ type Session struct {
 	heartbeatIntervalNano     atomic.Int64
 	nextHeartbeatDeadlineNano atomic.Int64
 
+	// heartbeatWake nudges the heartbeat loop to re-evaluate its Timer
+	// after the atomic deadline moves. Cap-1 non-blocking channel so a
+	// burst of resets coalesces into a single wake and hot-path frame
+	// handling stays allocation-free.
+	heartbeatWake chan struct{}
+
 	// quiescent closes when the in-flight vRPC drains after StateClosing,
 	// or when ForceClose runs.
 	quiescent     chan struct{}
@@ -184,11 +204,12 @@ type SessionOption func(*Session)
 // valid.
 func NewSession(logName string, stream Stream, hooks SessionHooks, sessionType SessionType, opts ...SessionOption) *Session {
 	s := &Session{
-		logName:     logName,
-		stream:      stream,
-		hooks:       hooks,
-		quiescent:   make(chan struct{}),
-		sessionType: sessionType,
+		logName:       logName,
+		stream:        stream,
+		hooks:         hooks,
+		quiescent:     make(chan struct{}),
+		sessionType:   sessionType,
+		heartbeatWake: make(chan struct{}, 1),
 	}
 	s.state.Store(int32(StateNew))
 	s.heartbeatIntervalNano.Store(int64(defaultHeartbeatInterval))
@@ -245,3 +266,38 @@ func unavailable(cause error, format string, args ...interface{}) error {
 		cause: cause,
 	}
 }
+
+// Send writes a SessionRequest under sendMu so concurrent producers don't
+// corrupt the underlying stream. grpc.ClientStream.Send is not safe for
+// concurrent use; sendMu is the only serialization point.
+func (s *Session) Send(req *spb.SessionRequest) error {
+	s.sendMu.Lock()
+	err := s.stream.Send(req)
+	s.sendMu.Unlock()
+	if err == nil {
+		s.msgsSent.Add(1)
+		s.msgsSentByType[classifyReq(req)].Add(1)
+	}
+	return err
+}
+
+// resetHeartbeatDeadline pushes out the watchdog to (now + heartbeatInterval).
+// One atomic load + one atomic store on the hot path, plus a non-blocking
+// wake to the heartbeat loop so its Timer picks up the new deadline
+// immediately (otherwise the initial bootstrap arm keeps the loop sleeping
+// past atomic shortenings — SESSION_SPEC.md #7).
+func (s *Session) resetHeartbeatDeadline() {
+	s.nextHeartbeatDeadlineNano.Store(time.Now().Add(time.Duration(s.heartbeatIntervalNano.Load())).UnixNano())
+	s.wakeHeartbeatLoop()
+}
+
+// wakeHeartbeatLoop signals the heartbeat loop to re-evaluate its Timer.
+// Non-blocking send on a cap-1 channel: bursts coalesce into a single
+// pending wake, so hot-path frame handlers stay allocation-free and
+// contention-free even under high frame arrival rates.
+func (s *Session) wakeHeartbeatLoop() {
+	select {
+	case s.heartbeatWake <- struct{}{}:
+	default:
+	}
+}
@@ -150,6 +150,23 @@ const (
 	// prior attempt's gRPC code + err (if the retry interceptor stashed
 	// one on ctx).
 	SessionEventRetry SessionEventKind = "retry"
+	// SessionEventProtocolError fires when routeVRPCFrame observes a
+	// state/frame combination that violates the client-server contract:
+	// a frame arrived in a state readLoop shouldn't be running in
+	// (New/Starting/Closed) OR the server's rpc_id didn't match our
+	// active vRPC. Escalated to session teardown via ForceClose so the
+	// pool's OnClosing/OnClose hooks fire and the session leaves the
+	// AFE routing set. Kept separate from SessionEventLateFrame so
+	// operators filtering sessionz for genuine desyncs aren't swamped
+	// by benign late-after-cancel drops.
+	SessionEventProtocolError SessionEventKind = "protocol-error"
+	// SessionEventLateFrame fires when routeVRPCFrame drops a frame
+	// because there's no active vRPC on this session (activeVRPC()==nil).
+	// This is a documented race — the caller ctx.Done'd and cancelled
+	// the slot, but the server's response arrived before the cancel
+	// landed on the wire. Not a protocol violation; the session stays
+	// healthy and the frame is dropped.
+	SessionEventLateFrame SessionEventKind = "late-frame"
 )
 
 // SessionEvent is one entry in a session's per-session debug ring buffer.
@@ -164,7 +181,7 @@ type SessionEvent struct {
 const maxSessionEvents = 64
 
 // recordEvent appends a SessionEvent to the per-session ring buffer.
-// Safe to call from any goroutine (readLoop, heartBeatLoop, etc.). Uses
+// Safe to call from any goroutine (readLoop, heartbeatLoop, etc.). Uses
 // the same wrap-index scheme as latencySamples so an at-cap append is
 // O(1), not an O(N) shift.
 func (s *Session) recordEvent(kind SessionEventKind, format string, args ...interface{}) {