test: repair CI TERM handshakes, port candidates, and TopN coverage

optools/ut_schedule_test.go

@@ -1533,33 +1533,33 @@ read -r observed <&8
 wait "$child_pid"
 printf 'CHILD %s %s\n' "$observed" "$child_pid"
 handled=0
-ack_pending=0
-trap 'fixture_self_pid trap_pid; [[ "$trap_pid" == "$main" ]] || exit 97; handled=$((handled+1)); ack_pending=1' TERM
+pending_ack=0
+trap 'fixture_self_pid trap_pid; [[ "$trap_pid" == "$main" ]] || exit 97; handled=$((handled+1)); pending_ack=1' TERM
 (
+ # Unblock the main loop on protocol failure as well as normal completion.
+ trap ': > "$scratch/done"' EXIT
  for ((j=0; j<200; j++)); do
   kill -TERM "$main"
-  read -r acknowledgment <&7
-  [[ "$acknowledgment" == ack ]] || exit 98
+  read -r acknowledgment sequence pending <&7
+  [[ "$acknowledgment" == ack && "$sequence" == "$((j+1))" && "$pending" == 0 ]] || exit 98
  done
- touch "$scratch/done"
 ) &
 sender=$!
 trap 'kill "$sender" 2>/dev/null || :; wait "$sender" 2>/dev/null || :' EXIT
 while [[ ! -f "$scratch/done" ]]; do
+ # ACK means the previous trap has returned, so the sender can safely issue
+ # the next TERM. Publishing it inside the trap can lose that next signal.
+ if (( pending_ack )); then
+  pending_ack=0
+  printf 'ack %s %s\n' "$handled" "$pending_ack" >&7
+ fi
  # Reenter BASHPID expansion in modern Bash. The legacy fallback spawns
  # a child, whose output can be interrupted by TERM; its PID is checked
  # at startup, in the subshell, and by each TERM handler instead.
  if [[ -n "${BASHPID:-}" ]]; then
   fixture_self_pid observed
   [[ "$observed" == "$main" ]] || exit 99
  fi
- # Acknowledge only after the handler returns. Otherwise the sender can
- # deliver the next TERM while TERM is still blocked inside the handler,
- # and Bash can coalesce it with the signal already being handled.
- if (( ack_pending )); then
-  ack_pending=0
-  printf 'ack\n' >&7
- fi
 done
 wait "$sender"
 trap - EXIT

pkg/logservice/test_ports_test.go

@@ -37,7 +37,7 @@ func TestTestPortAllocation(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			a := allocatedPorts{ports: make(map[int]struct{})}
 			calls := 0
-			port, err := a.allocate(func(int) error {
+			port, err := a.allocate(firstTestPort, 32768, func(int) error {
 				calls++
 				if calls <= tc.failures {
 					return &net.OpError{Op: "listen", Err: tc.cause}
@@ -123,3 +123,48 @@ func listenUDPOnTCPAvailablePort(t *testing.T) (net.PacketConn, int) {
 	require.FailNow(t, "could not reserve a UDP port that is also available to TCP")
 	return nil, 0
 }
+
+func TestTestPortRangeOutsideEphemeral(t *testing.T) {
+	for _, tc := range []struct {
+		name, value  string
+		first, limit int
+	}{
+		{"linux default", "32768\t60999\n", 21024, 32768},
+		{"macos default", "49152 65535", 21024, 49152},
+		{"only above", "20000 40000", 40001, 65535},
+		{"only below", "25000 65535", 21024, 25000},
+		{"below candidates", "1024 20000", 21024, 65535},
+		{"last port", "65535 65535", 21024, 65535},
+		{"no candidates", "1024 65535", 0, 0},
+		{"reversed", "61000 32768", 0, 0},
+		{"missing", "32768", 0, 0},
+		{"extra", "32768 60999 0", 0, 0},
+		{"not numeric", "low high", 0, 0},
+		{"zero", "0 60999", 0, 0},
+		{"too high", "32768 65536", 0, 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			first, limit, err := testPortRangeOutsideEphemeral(tc.value)
+			if tc.first == 0 {
+				require.Error(t, err)
+				require.Zero(t, first)
+				require.Zero(t, limit)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.first, first)
+			require.Equal(t, tc.limit, limit)
+			a := allocatedPorts{ports: make(map[int]struct{})}
+			for range 100 {
+				port, err := a.allocate(first, limit, func(candidate int) error {
+					require.GreaterOrEqual(t, candidate, tc.first)
+					require.Less(t, candidate, tc.limit)
+					return nil
+				})
+				require.NoError(t, err)
+				require.GreaterOrEqual(t, port, first)
+				require.Less(t, port, limit)
+			}
+		})
+	}
+}

pkg/logservice/test_utils.go

@@ -20,6 +20,10 @@ import (
 	"fmt"
 	"math/rand"
 	"net"
+	"os"
+	stdRuntime "runtime"
+	"strconv"
+	"strings"
 	"sync"
 	"syscall"
 	"testing"
@@ -61,20 +65,67 @@ var randomPorts = allocatedPorts{
 }
 
 func getAvailablePort() int {
-	port, err := randomPorts.allocate(probeTestPort)
+	first, limit, err := testPortRange()
+	if err != nil {
+		panic(err)
+	}
+	port, err := randomPorts.allocate(first, limit, probeTestPort)
 	if err != nil {
 		panic(err) // Preserve the existing fixture API, but fail promptly with the cause.
 	}
 	return port
 }
 
-const maxPortAllocationAttempts = 128
+const (
+	maxPortAllocationAttempts = 128
+	firstTestPort             = 21024
+	testPortLimit             = 65535
+)
+
+// Probing a port does not reserve it. Keep candidate listener ports outside
+// the automatic client-port range so outbound traffic cannot claim them in
+// the gap between the probe and service startup. Explicit binds can still
+// race that gap; they are not coordinated by the process-local used-port map.
+func testPortRange() (int, int, error) {
+	if stdRuntime.GOOS != "linux" {
+		// Conservative for the default macOS dynamic-port range (49152+).
+		// Unlike Linux, customized non-Linux ranges are not inspected here.
+		return firstTestPort, 32768, nil
+	}
+	data, err := os.ReadFile("/proc/sys/net/ipv4/ip_local_port_range")
+	if err != nil {
+		return 0, 0, &testPortAllocationError{message: fmt.Sprintf("read automatic client-port range: %v", err), cause: err}
+	}
+	return testPortRangeOutsideEphemeral(string(data))
+}
+
+func testPortRangeOutsideEphemeral(value string) (int, int, error) {
+	fields := strings.Fields(value)
+	if len(fields) == 2 {
+		low, lowErr := strconv.Atoi(fields[0])
+		high, highErr := strconv.Atoi(fields[1])
+		if lowErr == nil && highErr == nil && low > 0 && low <= high && high <= 65535 {
+			belowLimit := max(firstTestPort, min(testPortLimit, low))
+			aboveFirst := min(testPortLimit, max(firstTestPort, high+1))
+			// Choose the larger contiguous interval; neither includes the
+			// configured inclusive ephemeral range.
+			if belowLimit-firstTestPort >= testPortLimit-aboveFirst && belowLimit > firstTestPort {
+				return firstTestPort, belowLimit, nil
+			}
+			if aboveFirst < testPortLimit {
+				return aboveFirst, testPortLimit, nil
+			}
+			return 0, 0, &testPortAllocationError{message: "automatic client-port range leaves no test listener ports"}
+		}
+	}
+	return 0, 0, &testPortAllocationError{message: fmt.Sprintf("invalid automatic client-port range %q", value)}
+}
 
-func (a *allocatedPorts) allocate(probe func(int) error) (int, error) {
+func (a *allocatedPorts) allocate(first, limit int, probe func(int) error) (int, error) {
 	a.Lock()
 	defer a.Unlock()
 	for range maxPortAllocationAttempts {
-		port := rand.Intn(65535-21024) + 21024
+		port := rand.Intn(limit-first) + first
 		if _, exists := a.ports[port]; exists {
 			continue
 		}

pkg/sql/colexec/partition/topn_test.go

@@ -771,12 +771,10 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) {
 	t.Cleanup(func() { arg.Free(proc, false, nil) })
 
 	for generation := 0; generation < 2; generation++ {
-		batches := []*batch.Batch{
-			makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false),
-			makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true),
-		}
-		child := colexec.NewMockOperator().WithBatchs(batches)
+		child := colexec.NewMockOperator()
 		t.Cleanup(func() { child.Free(proc, false, nil) })
+		child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{10, 20}, false)})
+		child.WithBatchs([]*batch.Batch{makeAnyNullTopNBatch(t, proc, []int32{1, 2}, []int64{11, 21}, true)})
 		arg.Children = nil
 		arg.AppendChild(child)
 
@@ -805,10 +803,15 @@ func TestPartitionTopNNonLiteralAnyNullOrder(t *testing.T) {
 	}
 }
 
-func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) *batch.Batch {
+func makeAnyNullTopNBatch(t *testing.T, proc *process.Process, groups []int32, payloads []int64, constant bool) (result *batch.Batch) {
 	t.Helper()
 	require.Len(t, groups, len(payloads))
 	bat := batch.NewWithSize(3)
+	defer func() {
+		if result == nil {
+			bat.Clean(proc.Mp())
+		}
+	}()
 	bat.Vecs[0] = vector.NewVec(types.T_int32.ToType())
 	require.NoError(t, vector.AppendFixedList(bat.Vecs[0], groups, nil, proc.Mp()))
 	if constant {