sql/flowinfra: flow teardown is not unwind safe

**Summary:**

Flow teardown in `pkg/sql/flowinfra` is not unwind safe: a panic in one processor can leave its peers blocked forever, and two lock-handling sites in the flow registry turn a panic into either a permanent self-deadlock or an unrecoverable fatal error. Part of #174944.

The hang cases are worse than the crash they would replace: the job's goroutine and everything it holds are gone silently, with no diagnostics.

**Findings:**

- `flowinfra.FlowBase.Wait`: the flow context — `Wait` calls `recover()` specifically so it can call `ctxCancel()`, with the comment "otherwise we will wait indefinitely". Any recovery that consumes the panic before `Wait` runs makes that `recover()` return nil, so nothing cancels the context and `Wait` blocks on peers whose only exit was that cancellation.
- `flowinfra.FlowBase.Wait`: the `<-waitChan` receive has no deadline, so there is no escape once the above happens.
- `execinfra.RowChannel.Push`: the send is a bare channel send with no `select` on `ctx.Done()`. The only thing that frees a blocked producer is `RowChannel.ConsumerClosed`, reachable only from `FlowBase.Cleanup`'s `proc.Close` loop — which runs *after* `Wait`. Peer processors and inbound-stream readers park in `Push` forever and their `waitGroup.Done()` never fires.
- `flowinfra.FlowRegistry.Drain`: `fr.Mutex` — the registry lock is held across a caller-supplied `reporter` callback with a non-deferred unlock, while a separate closure deferred at the top of `Drain` unconditionally re-`Lock()`s it. If `reporter` panics, that deferred re-lock deadlocks the goroutine against a mutex it still holds, wedging every future flow on the node. `cancelPendingStreams` in the same file shows the correct shape.
- `flowinfra.FlowRegistry.waitForFlow`: `fr.Mutex` — a `defer fr.Unlock()` is registered, then the mutex is manually unlocked around a blocking `select` and re-locked only on the non-panic path. A panic in that window fires the deferred `Unlock()` on an already-unlocked mutex, which is a runtime `throw()` — fatal, and not recoverable by any `recover()` in the process. The safe self-relocking-closure idiom is already used in `colflow.routers` and `stats.stats_cache`.
- `flowinfra.FlowBase.Cleanup`: `f.Descriptors`, `f.sp`, `f.Mon`, `f.DiskMonitor` — the `proc.Close` loop runs before descriptor release, span finish and monitor teardown, none of which are deferred, and the `recover()` in the body does not catch a panic raised inside `Cleanup` itself. One panicking `proc.Close` skips all of them.
- `execinfra.ProcessorBaseNoHelper.InternalClose`: `pb.span` — `Finish()` is a plain statement, so a processor that panics mid-`Next()` strands its span until the `Cleanup` loop above rescues it, which is the same loop that can be skipped.

**Next Steps:**

- [ ] Cancel the flow context on any recovery path before returning
- [ ] Give `Wait` a deadline on the post-recovery path; on expiry, log a goroutine dump and return so the operation fails rather than hangs
- [ ] Make `RowChannel.Push` select on `ctx.Done()`, or close consumers before waiting
- [ ] Call `reporter` outside the locked region in `Drain`
- [ ] Convert `waitForFlow` to the self-relocking-closure idiom
- [ ] Register monitor teardown and `sp.Finish()` as defers at the top of `Cleanup`; move `ReleaseAll` before the `proc.Close` loop
- [ ] Defer `pb.span.Finish()` in `InternalClose`

Epic: none

Jira issue: CRDB-68120

Start by reading FlowBase.Wait, Cleanup, FlowRegistry.Drain and waitForFlow under pkg/sql/flowinfra, plus RowChannel.Push and ProcessorBaseNoHelper.InternalClose in execinfra; compare the self-relocking patterns in colflow.routers and stats.stats_cache. Done means every listed panic path cancels or exits safely, releases locks and resources, prevents blocked peers, and produces the requested diagnostics instead of hanging or fatally throwing.

Written by the indexing model from the issue text.