Make watch queries easier to cancel

[simolus3]

This refactors our watch query stream implementation slightly:

1. Addressing the FIXME comment in watch(), this refactors the stream implementation there to use stream combinators instead of async*. As that comment explains, cancelling the stream might run the query another time which is not typically what users want.
2. The _throttleStream implementation, also using async*, had a similar issue where it couldn't be cancelled efficiently when it's in the await Future.delayed(timeout); stage. I've rewritten that method with a manual stream subscription to be more explicit about how we deal with downstream listeners pausing / cancelling the stream (personally, I find the exact semantics of async* quite hard to understand).

[rkistner]

Added some minor comments, but looks great overall