Replication stream multi-cursor

[robholland]

What changed?

• Added ReplicationReaderGroup, a type that encapsulates the 3-scope QueueReaderState watermark arithmetic (catchup begin, reader-state construction, failover watermark) that was previously duplicated across recvSyncReplicationState and getSendCatchupBeginInclusiveWatermark. Gated by EnableReplicationReaderGroup.
• Added namespaceIsolationManager, which coordinates per-namespace dedicated LOW readers on the sender. When the receiver signals a namespace is throttled, a dedicated goroutine is created for it and the default LOW reader skips it. When the namespace is un-throttled the dedicated reader drains to the recorded cursor before removing itself. Gated by
  EnableReplicationNamespaceIsolation.
• Added NamespaceThrottler interface and NoopNamespaceThrottler default for the receiver side to report which namespaces exceeded the LOW-lane threshold.
• Added throttled_namespace_ids to SyncReplicationState proto so the receiver can communicate the throttled namespace set to the sender on each ack.

Why?

A large backfill can flood the LOW-priority replication lane and stall catchup for every other namespace behind it. The isolation manager gives those heavily loaded namespaces their own reader so the default LOW reader is unblocked. The ReplicationReaderGroup is a prerequisite refactor, the scope-index arithmetic needs to live in one place before we can extend it to support per-namespace cursors cleanly. Both features are off by default and the stream restarts itself if either flag changes at runtime. The namespace throttler allows per-namespace adjustments of low priority limits.

How did you test it?

• built
• run locally and tested manually
• covered by existing tests
• added new unit test(s)
• added new functional test(s)

Potential risks

Any change is risky. Identify all risks you are aware of. If none, remove this section.

[yux0]

-1 if not found?

[yux0]

how does this work without pass in the task id?

[robholland]

We don't actually care about the task per se, just that we've had a task for the given namespace. Is there better naming here maybe to make that clearer? The standard for throttling interfaces like this has been Allow but we aren't returning yes/no here as the throttling system is after-the-fact in this instance.