diff --git a/optools/ut_schedule_test.go b/optools/ut_schedule_test.go index f59eee57481a7..6d5a6d589bca5 100644 --- a/optools/ut_schedule_test.go +++ b/optools/ut_schedule_test.go @@ -1533,19 +1533,26 @@ 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. @@ -1553,13 +1560,6 @@ while [[ ! -f "$scratch/done" ]]; do 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 diff --git a/pkg/logservice/test_ports_test.go b/pkg/logservice/test_ports_test.go index 26f9ecef42dbe..d2b33ef54553e 100644 --- a/pkg/logservice/test_ports_test.go +++ b/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) + } + }) + } +} diff --git a/pkg/logservice/test_utils.go b/pkg/logservice/test_utils.go index 8ca0c9a902f16..ddfe316e07b2a 100644 --- a/pkg/logservice/test_utils.go +++ b/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 } diff --git a/pkg/sql/colexec/partition/topn_test.go b/pkg/sql/colexec/partition/topn_test.go index 86389fa8ec6a6..2ecd552160554 100644 --- a/pkg/sql/colexec/partition/topn_test.go +++ b/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 {