Pyth-Lazer periodic consolidation of subscriptions

[hacheigriega]

Motivation

To reduce frequency of incoming messages (frames), we periodically consolidate individual subscriptions into bulk subscriptions. This PR replaces #143 to accommodate base branch changes and add a few improvements.

Explanation of Changes

• We create separate bulk subscriptions for different channels.
• After we send a new bulk subscription request, we wait for bulkConsolidateTimeout for a tick. We only refresh the bulk subscription upon receipt of a tick. If we do not receive a tick, we keep the old bulk subscription or, if there is no existing bulk subscription, we do not create a new one.
• Idle cleanup does not unsubscribe a bulk subscription. Removing the idle feed from the local map stops cache writes immediately, and the next consolidation pass rebuilds the bulk set without it.

This PR also includes a small commit on concurrent execution of symbol -> ID resolution.

Testing

Added some unit tests.
Cannot test locally against the WS endpoint due to lack of Pyth Lazer API key for testing :(

Related PRs and Issues

Related PR: #143

[Thomasvdam]

Before I do a full review I wonder if we considered always doing bulk subscriptions? What are the pros/cons of that approach?

Rough outline:

• I have 1 bulk running for X and Y on channel 200ms.
• Request comes in for Z on 200ms.
• I send a subscribe frame for X+Y+Z on 200ms.
• Success: new sub is leading. Unsubscribe the old X+Y on 200ms.
• Failure: return failure to client for Z. Keep old subscription for X+Y on 200ms.

Maybe it gets trickier with invalid IDs? Docs recommend to not error on them in the bulk request. How does that currently work with single feed subscriptions?

[hacheigriega]

Before I do a full review I wonder if we considered always doing bulk subscriptions? What are the pros/cons of that approach?

I did briefly think about that, but I now see that I didn't consider that approach seriously enough. I just assumed that bulk-only approach would complicate and slow down serving requests containing new symbols.

You are right the doc recommends ignoreInvalidFeedIds: true and this could bring a lot of zombie invalid symbols to the bulk subscription under the bulk-only approach. At least in our current implementation we check the symbols are valid at the time of subscription. I see that the symbols could still be delisted or be renamed later. For that I'm thinking we can stop serving stale values from the cache and cleaning up state symbols.

Talking to AI I realized there is another issue with race condition when there are two in-flight requests:

Starting with [A,B] as bulk subscription:

1. Request 1 sends [A,B,C]
2. Request 2 sends [A,B,D]
3. Both succeed -> The new bulk subscription becomes [A,B,C] or [A,B,D] instead of [A,B,C,D]

I don't think there is a way around this other than serializing the wait for new bulk subscription. This could prevent requests from receiving price response because new symbols in requests are not subscribed until earlier bulk subscriptions start ticking.

Please let me know what you think.

[Thomasvdam]

You are right the doc recommends ignoreInvalidFeedIds: true and this could bring a lot of zombie invalid symbols to the bulk subscription under the bulk-only approach. At least in our current implementation we check the symbols are valid at the time of subscription. I see that the symbols could still be delisted or be renamed later. For that I'm thinking we can stop serving stale values from the cache and cleaning up state symbols.

I don't think this is (always) true. The validation only happens when we convert a symbol to an ID. When I pass an ID directly the module will happily create a subscription and wait for the timeout to tell me there's no price.

---

I'm still in camp "always batch" but not 100% convinced it's correct so here's my thought process. In a way the race condition already exists right? When 2 clients send a request for C we deduplicate it through the lastRequestToPriceFeed map. This implies that we have a central 'store' that keeps track of all the active feeds. If that's true we should be able to resolve your scenario as follows:

Starting with [A,B] as bulk subscription, 2 requests come in for D and C respectively:

1. Request 1 appends D to the pending subscriptions, sends request for [A,B,D]
2. Request 2 sends appends C to the pending subscriptions, sends request for [A,B,D,C]
3. Both succeed:
   a. [A,B,D] responds first, updates the cache for A, B, and D, replaces current bulk subscription.
   a. [A,B,D,C] responds second, updates the cache for A, B, D, and C, replaces the bulk subscription again.
   b. [A,B,D,C] responds first, updates the cache for A, B, D, and C, replaces the current bulk subscription.
   b. [A,B,D] responds second, either ignore completely or allow it to update the cache for A, B, and D.

I think the main downside of this approach is that the housekeeping of which subscription we have to keep is more complex compared to the round up flow in the current PR. The upside is that we have a single flow of managing subscriptions, usually have only 1 subscription, and more predictable behaviour instead of being at the mercy of yet another background process. Oh and it immediately covers subscribing to multiple new symbols in 1 request, that's also nice.

But again, I have a nagging feeling I'm missing something so please challenge me :)

[hacheigriega]

I don't think this is (always) true. The validation only happens when we convert a symbol to an ID. When I pass an ID directly the module will happily create a subscription and wait for the timeout to tell me there's no price.

Right. From the user perspective there is no problem. I was just thinking the bulk subscription could end up with a lot of invalid symbols. Currently we don't clean up based on whether we receive price data or not. We only clean up based on whether there was a recent user request.

Similarly for the second part I was mainly concerned about the bulk subscription state. In this step 3b

b. [A,B,D] responds second, either ignore completely or allow it to update the cache for A, B, and D.

we would end up replacing the bulk subscription state to [A,B,D]. To avoid this we would need some kind of mutex mechanism around bulk state, which I think would inevitably bring latency when there is a series of requests containing new symbols. There is no problem in terms of providing price responses to all valid requested symbols. But the problem is maintaining synchronized bulk subscription state.

I think the main downside of this approach is that the housekeeping of which subscription we have to keep is more complex compared to the round up flow in the current PR.

Maybe I reiterated what you are already aware of. But I was thinking this is the main problem with the bulk-only approach.

[hacheigriega]

See #167 for an alternative approach