Allow subscribing to reflector store updates

[nightkr]

Motivation

This is largely a "shower thoughts" take on #1189 and the discussions from the last office hours.

It still fails controller::tests::applier_must_not_deadlock_if_reschedule_buffer_fills and introduces some type safety regressions that I'm not quite sure how to fix.

Solution

It introduces a new (internal) Broadcaster channel, which has the backpressure and broadcasting properties that we need (async-channel is inappropriate after all since it only sends each message to one receiver).

It then adds an interface to reflector::store::Writer for subscribing to changes.

This all leads to a fair amount of changes to reflector and its downstreams to accomodate for Writer::apply_watcher_event now being async (so that it can send out the appropriate events).

[mateiidavid]

hey, just linking this here since I've been working on the same thing

main...mateiidavid:kube:matei/arc-watcher

I too switched to async-broadcast after our discussion since it's better suited for what we need to do.

[nightkr]

Ah, hadn't seen that one. However, that also looks like it prefers to drop messages rather than apply backpressure (https://docs.rs/async-broadcast/latest/async_broadcast/enum.RecvError.html#variant.Overflowed).

[mateiidavid]

I think overflow is disabled by default. It can also be disabled manually (see https://docs.rs/async-broadcast/latest/async_broadcast/struct.Receiver.html#method.overflow).

The Send future does have annoying lifetimes so what I ended up doing is using try_recv instead of trying to build a state machine to resume after backpressure was applied. The function returns a RecvError that is either Error::Full or Error::Closed (slightly paraphrasing the types).

I actually wanted to send that over to you for a quick sanity check but I guess now's as good a time as any 😂

[nightkr]

I think overflow is disabled by default. It can also be disabled manually (see https://docs.rs/async-broadcast/latest/async_broadcast/struct.Receiver.html#method.overflow).

Ahh: https://docs.rs/async-broadcast/latest/src/async_broadcast/lib.rs.html#156, gotcha, looks reasonable.

The Send future does have annoying lifetimes so what I ended up doing is using try_recv instead of trying to build a state machine to resume after backpressure was applied. The function returns a RecvError that is either Error::Full or Error::Closed (slightly paraphrasing the types).

I assume you mean try_broadcast? Sadly that won't work, since it won't register with the waker (so that it knows when it's time to retry). That'll deadlock until the next time that the future is polled for some other reason.