testutils: add reusable channel receive assertions

## Summary

Add reusable generic channel assertion helpers to `pkg/testutils`, tentatively:

```go
func ReceiveSoon[T any](t TestFataler, ch <-chan T, description string) T
func ReceiveWithin[T any](t TestFataler, ch <-chan T, timeout time.Duration, description string) T
```

Tests currently implement this pattern through direct `select` statements, local helpers, `testutils.SucceedsSoon`, and `require.Eventually`. There is no shared channel-receive helper in `pkg/testutils`.

This was noticed while reviewing cockroachlabs/cockroach#3952. It is follow-up cleanup, not a blocker for that PR.

## Audit results

The audit used `master` at commit `61e8d82a116c6abe7d91e47e27b8cd874c60184f`. A conservative AST scan found:

- 176 straightforward positive two-case receive-or-timeout sites across 91 test files.
- 11 simple channel polls wrapped in `testutils.SucceedsSoon`/`SucceedsWithin` across 5 files.
- 7 simple channel polls wrapped in `require.Eventually` across 3 files.
- 97 distinct candidate files after overlap, or roughly 100 files and 190–200 call sites after accounting for named-timer helpers.

The scan deliberately excludes 64 negative waits (where receiving is the failure), multi-channel selects, and other cases requiring semantic judgment.

Representative existing patterns:

- `pkg/ccl/changefeedccl/cdcbatcher/buffer_test.go` defines a file-local generic `mustReceive[T]` that is nearly the proposed helper.
- `pkg/sql/stats/create_stats_job_test.go` uses a direct receive with `time.After(testutils.SucceedsSoonDuration())`.
- `pkg/ccl/changefeedccl/changefeed_dist_test.go` performs a non-blocking receive inside `testutils.SucceedsSoon`.
- `pkg/cmd/roachprod-centralized/services/clusters/internal/scheduler/scheduler_test.go` performs a non-blocking receive inside `require.Eventually`.

A repo-wide migration should be net-negative by approximately 800–1,200 lines after adding the helper and its tests.

## Related issue

- #100796 proposes linting against `require.Eventually` / `assert.Eventually` because their callbacks run in separate goroutines and can misuse `testing.T` or leak. This issue is not a duplicate: `ReceiveSoon` would provide a direct, synchronous replacement for the channel-only cases found by this audit and complements that lint proposal.

Searches for open issues containing `ReceiveSoon`, `mustReceive`, channel test helpers, and `SucceedsSoon` channel cleanup found no closer match.

## Proposed behavior

- `ReceiveSoon` delegates to `ReceiveWithin` with `SucceedsSoonDuration()`, preserving the race/deadlock-enabled timeout adjustment.
- Return the received value so it works for both notification and value channels.
- Fail clearly on timeout and unexpected channel closure.
- Produce a goroutine dump on timeout, consistent with `SucceedsWithin`.
- Keep negative assertions and selects involving multiple result/error channels explicit.

## Suggested rollout

1. Add `ReceiveSoon` and `ReceiveWithin` with focused unit tests.
2. Migrate mechanically safe positive waits.
3. Review custom timeout normalization separately.
4. Leave negative and multi-channel selects unchanged unless a separate helper is justified.

## Reproduction

Save the following as `receive_candidates.go`, then run:

```sh
go run receive_candidates.go /path/to/cockroach
```

```go
package main

import (
"fmt"
"go/ast"
"go/parser"
"go/printer"
"go/token"
"io/fs"
"os"
"path/filepath"
"strings"
)

type counts struct {
direct, positive, negative, other int
succeeds, eventually int
directFiles, positiveFiles map[string]bool
succeedsFiles, eventuallyFiles map[string]bool
candidates map[string]bool
}

func source(fset *token.FileSet, node any) string {
var buf strings.Builder
_ = printer.Fprint(&buf, fset, node)
return buf.String()
}

func receiveExpr(stmt ast.Stmt) ast.Expr {
switch stmt := stmt.(type) {
case *ast.ExprStmt:
if expr, ok := stmt.X.(*ast.UnaryExpr); ok && expr.Op == token.ARROW {
return expr.X
}
case *ast.AssignStmt:
for _, rhs := range stmt.Rhs {
if expr, ok := rhs.(*ast.UnaryExpr); ok && expr.Op == token.ARROW {
return expr.X
}
}
}
return nil
}

func containsFatal(body []ast.Stmt) bool {
found := false
for _, stmt := range body {
ast.Inspect(stmt, func(node ast.Node) bool {
call, ok := node.(*ast.CallExpr)
if !ok {
return true
}
var name string
switch fun := call.Fun.(type) {
case *ast.SelectorExpr:
name = fun.Sel.Name
case *ast.Ident:
name = fun.Name
}
if strings.HasPrefix(name, "Fatal") || name == "Fail" ||
name == "FailNow" || name == "panic" {
found = true
}
return !found
})
}
return found
}

func simplePollingReceive(fn *ast.FuncLit) bool {
if fn == nil || len(fn.Body.List) != 1 {
return false
}
stmt, ok := fn.Body.List[0].(*ast.SelectStmt)
if !ok || len(stmt.Body.List) != 2 {
return false
}
var receives, defaults int
for _, item := range stmt.Body.List {
clause := item.(*ast.CommClause)
if clause.Comm == nil {
defaults++
} else if receiveExpr(clause.Comm) != nil {
receives++
}
}
return receives == 1 && defaults == 1
}

func main() {
if len(os.Args) != 2 {
fmt.Fprintln(os.Stderr, "usage: go run receive_candidates.go CHECKOUT")
os.Exit(2)
}
c := counts{
directFiles: make(map[string]bool),
positiveFiles: make(map[string]bool),
succeedsFiles: make(map[string]bool),
eventuallyFiles: make(map[string]bool),
candidates: make(map[string]bool),
}
err := filepath.WalkDir(os.Args[1], func(path string, entry fs.DirEntry, err error) error {
if err != nil {
return err
}
if entry.IsDir() {
switch entry.Name() {
case ".git", "vendor", "node_modules", "bazel-bin", "bazel-out", "bazel-testlogs":
return filepath.SkipDir
}
return nil
}
if !strings.HasSuffix(path, "_test.go") {
return nil
}
fset := token.NewFileSet()
file, err := parser.ParseFile(fset, path, nil, 0)
if err != nil {
return err
}
ast.Inspect(file, func(node ast.Node) bool {
switch node := node.(type) {
case *ast.SelectStmt:
if len(node.Body.List) != 2 {
return true
}
var timeout, received *ast.CommClause
for _, item := range node.Body.List {
clause := item.(*ast.CommClause)
expr := receiveExpr(clause.Comm)
if expr == nil {
continue
}
if strings.HasPrefix(source(fset, expr), "time.After(") {
timeout = clause
} else {
received = clause
}
}
if timeout == nil || received == nil {
return true
}
c.direct++
c.directFiles[path] = true
timeoutFatal, receiveFatal := containsFatal(timeout.Body), containsFatal(received.Body)
switch {
case timeoutFatal && !receiveFatal:
c.positive++
c.positiveFiles[path] = true
c.candidates[path] = true
case receiveFatal && !timeoutFatal:
c.negative++
default:
c.other++
}
case *ast.CallExpr:
name := source(fset, node.Fun)
for _, arg := range node.Args {
fn, ok := arg.(*ast.FuncLit)
if !ok || !simplePollingReceive(fn) {
continue
}
switch {
case strings.Contains(name, "SucceedsSoon") || strings.Contains(name, "SucceedsWithin"):
c.succeeds++
c.succeedsFiles[path] = true
c.candidates[path] = true
case strings.Contains(name, "Eventually"):
c.eventually++
c.eventuallyFiles[path] = true
c.candidates[path] = true
}
}
}
return true
})
return nil
})
if err != nil {
panic(err)
}
fmt.Printf("simple direct: %d sites in %d files\n", c.direct, len(c.directFiles))
fmt.Printf(" positive candidates: %d sites in %d files\n", c.positive, len(c.positiveFiles))
fmt.Printf(" negative waits: %d; other/complex: %d\n", c.negative, c.other)
fmt.Printf("simple SucceedsSoon/Within polls: %d sites in %d files\n", c.succeeds, len(c.succeedsFiles))
fmt.Printf("simple Eventually polls: %d sites in %d files\n", c.eventually, len(c.eventuallyFiles))
fmt.Printf("candidate union: %d files\n", len(c.candidates))
}
```

Output on the audited `master` commit:

```text
simple direct: 322 sites in 136 files
positive candidates: 176 sites in 91 files
negative waits: 64; other/complex: 82
simple SucceedsSoon/Within polls: 11 sites in 5 files
simple Eventually polls: 7 sites in 3 files
candidate union: 97 files
```

The walker is intentionally conservative and should be followed by human review before bulk edits.

Jira issue: CRDB-67396

Start in pkg/testutils and review the proposed ReceiveSoon and ReceiveWithin signatures and existing SucceedsWithin behavior. Use the representative patterns in pkg/ccl/changefeedccl/cdcbatcher/buffer_test.go, pkg/sql/stats/create_stats_job_test.go, and the scheduler test to guide focused unit tests and migration review. Done means clear timeout and channel-closure failures, returned received values, and only mechanically safe positive waits migrated.

Written by the indexing model from the issue text.