[multistage] defensive thread scheduling

See [Multistage Threading Model](https://docs.google.com/document/d/1XAMHAlhFbINvX-kK1ANlzbRz4_RkS0map4qhqs1yDtE/edit#heading=h.de4smgkh3bzk) for additional background.

As of #9887, operators no longer block at all. Instead, they are scheduled into one of two queues: `_ready` if there is data ready for them to read, and `_available` if there is no data for them to read but they have not yet completed processing.

While I have a high degree of confidence that the `_ready` queue should never grow unbounded, there are various situations where the `_available` queue will leak operator chains. We should add some defensive mechanisms to the scheduler to ensure that such leaks never happen and are properly cleaned up:

1. it is possible that data did in fact arrive, but the scheduler missed the callback thereby keeping the operators in the `_available` queue. We should introduce a mechanism to attempt to reschedule operators on the `_available` queue after some timeout.
2. it is possible that data will never arrive (upstream network error - so we don't even get notified of an error), in which case we want to cleanup operators on the `_available` queue. In conjunction with (1), which represents a local timeout based on how long an operator is on the queue since last attempt at execution, we should introduce a second timeout that checks globally how long it has been since an operator chain was scheduled. If this exceeds the query timeout, we should cancel the query.

Note that we may get (2) for free if we do (1), because the MailboxReceiveOperator has timeout logic already baked into it. That means if we schedule it after a certain amount of time it will notice it has passed its global timeout and return an EOS block in the form of an error.

Anytime (1) is triggered and it wasn't a false-trigger (e.g. there was actual data to be processed) we should log a warning / record a metric.

We should apply the same logic for the `_seenMail` set.

There should also be some observability against false tight loops (e.g. an operator chain keeps getting scheduled but there wasn't actually any data to process).