fix: snapshot IsBusy state to prevent TOCTOU panic in partitionHealthy and release callReq+stream on addCall failure

partitionHealthy: IsBusy() reads a live atomic in-flight counter.
Calling it twice (counting pass + placement pass) creates a TOCTOU
race: if hosts flip from healthy to busy between passes, the placement
indices can overflow past the slice bounds causing an index-out-of-range
panic. Fix by snapshotting the busy state once per host in pass 1 and
reusing it in pass 2, using a small stack buffer for the common case.

exec: when addCall() fails, the callReq obtained from the pool and
the reserved stream ID were leaked. Fix by clearing the stream and
returning the callReq to the pool before returning the error.

Author: mykaul

conn.go

@@ -1226,6 +1226,8 @@ func (c *Conn) exec(ctx context.Context, req frameBuilder, tracer Tracer, reques
 	}
 
 	if err := c.addCall(call); err != nil {
+		c.streams.Clear(stream)
+		putCallReq(call)
 		return nil, &QueryError{err: err, potentiallyExecuted: false}
 	}
 

policies.go

@@ -769,10 +769,22 @@ func partitionHealthy(replicas []*HostInfo, s *Session) {
 		return
 	}
 
-	// Two-pass stable partition: first pass counts, second places.
+	// Snapshot IsBusy state once per host to avoid TOCTOU races between
+	// the counting pass and the placement pass. IsBusy reads a live atomic
+	// in-flight counter, so a host can flip between passes. If healthyCount
+	// becomes stale, the placement indices (hi/ui) can overflow.
+	var busyBuf [8]bool
+	var busy []bool
+	if n <= len(busyBuf) {
+		busy = busyBuf[:n]
+	} else {
+		busy = make([]bool, n)
+	}
+
 	healthyCount := 0
-	for _, h := range replicas {
-		if !h.IsBusy(s) {
+	for i, h := range replicas {
+		busy[i] = h.IsBusy(s)
+		if !busy[i] {
 			healthyCount++
 		}
 	}
@@ -792,8 +804,8 @@ func partitionHealthy(replicas []*HostInfo, s *Session) {
 	copy(tmp, replicas)
 
 	hi, ui := 0, healthyCount
-	for _, h := range tmp {
-		if !h.IsBusy(s) {
+	for i, h := range tmp {
+		if !busy[i] {
 			replicas[hi] = h
 			hi++
 		} else {