feat(server): Implement bandwidth-based rate limits

[lcian]

Introduces an implementation of bandwidth-based rate limiting and bandwidth metrics reporting.

The bandwidth is measured by wrapping our PayloadStream in GET/POST/PUT requests and counting the bytes that flow through it. This allows us to estimate the bandwidth that's being utilized at a given point in time between the server and all clients -- note that we're not concerned with the bandwidth utilization between the server and the storage backends, which should anyways be the same as the client-server one, plus an overhead for tombstone requests and responses.

The counts are summed every 50ms (this might be too frequent, we can tune it) and then contribute to an Exponentially Weighted Moving Average.

This EWMA estimate of the bandwidth is reported to Datadog (widget) and is used to enforce bandwidth-based rate limits.

Differently from throughput-based rate limits, bandwidth-based rate limits only enforce a global limit in the bandwidth that can be used by an instance of the service.
Of course, they can always be extended to support the same configuration flexibility as throughput-based rate limits. That might be unnecessary and ineffective as: - we're more concerned with global bandwidth as it's related to node limits in the infra; - getting more granular would only increase our "measurement error" due to all the approximation we're doing with wrapping the stream and using a moving average.

[matt-codecov]

accept with nits

nice work!

[matt-codecov]

i think PayloadStream is already pinned so this #[pin] (and the pin-project-lite) dep are unneeded?

PayloadStream is a type alias for BoxStream which is a type alias for Pin<Box<dyn Stream>>.

[lcian]

You're right.
In this case we can just use self.get_mut() to get a &mut MeteredPayloadStream which ultimately allows us to get a Pin<&mut dyn Stream<...>> that we can call poll_next on.

However, I don't understand why we're allowed to call self.get_mut in the first place. The requirement to do that is that our MeteredPayloadStream be Unpin, which in my interpretation should be false due to inner being Pin.

I will need to do some more reading about Pin because I don't understand it enough.