stream: force subscription store count check as stop gap for wrapper side implementation

[shazbert]

PR Description

Noticed in Bybit that subscriptions weren't being added when I was flushing the connection every hour, so abstracted adding and removing to websocket package then I can merge this to the open PR chain. This is so we don't need to toil in the wrappers as well but there might be some edge cases I am not considering.

This only impacts exchanges that are upgraded to the websocket multi connection.

Added in checks in stream package so that the outbound subscriptions are added or removed from store.

Fixes bybit and Deribit

Type of change

Please delete options that are not relevant and add an x in [] as item is complete.

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• This change requires a documentation update

How has this been tested

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration and
also consider improving test coverage whilst working on a certain feature or package.

• go test ./... -race
• golangci-lint run
• Test X

Checklist

• My code follows the style guidelines of this project
• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation and regenerated documentation via the documentation tool
• My changes generate no new warnings
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally and on Github Actions with my changes
• Any dependent changes have been merged and published in downstream modules

[codecov]

Codecov Report

Attention: Patch coverage is 15.38462% with 11 lines in your changes missing coverage. Please review.

Project coverage is 37.13%. Comparing base (85ecd0d) to head (f1ca7bd).
Report is 4 commits behind head on master.

Files with missing lines	Patch %	Lines
exchanges/stream/websocket.go	10.00%	6 Missing and 3 partials ⚠️
exchanges/gateio/gateio_websocket.go	33.33%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1717      +/-   ##
==========================================
- Coverage   37.15%   37.13%   -0.02%     
==========================================
  Files         414      414              
  Lines      180198   180199       +1     
==========================================
- Hits        66950    66916      -34     
- Misses     105389   105428      +39     
+ Partials     7859     7855       -4     

Files with missing lines	Coverage Δ
exchanges/gateio/gateio_websocket.go	60.90% <33.33%> (+0.07%)	⬆️
exchanges/stream/websocket.go	85.22% <10.00%> (-0.87%)	⬇️

... and 12 files with indirect coverage changes

---- 🚨 Try these New Features:

• Flaky Tests Detection - Detect and resolve failed and flaky tests
• JS Bundle Analysis - Avoid shipping oversized bundles

[gbjk]

I have concerns about building this out further:

• Some exchanges set PendingState and rely upon it
• Some exchanges add temporary subs until they have a better known sub key
• Fan out mechanics mean that sometimes the subs we add are't the ones we expected to
• We often need to get the sub added in wsHandleData directly, to ensure that the next message gets parsed for orderbook, etc
• I think this doesn't allow for Subscribe to be called directly

[gbjk]

Given comment, I don't see a way to move this forward right now.
I feel like the manageSubs is the only place we can properly manage sub state.
One edge case in the back of my mind is resubscribing, as well, particularly on orderbook fail.

[shazbert]

These are all fair points, an option I will investigate is check the connection subscription status after subscriptions to see if they have been added or removed. So that at least it complains that it needs to be added in the exchange packages themselves.

[gbjk]

@shazbert Just realised that I haven't mentioned:

I completely agree about abstracting from exchange implementations. I'm just not yet ready to see how to do it.
My intention was to finish merging all the exchange sub changes, then handle how multi-asset sub confs should work (which is GateIO + multi-sockets + assets ), and then review what hinge points stop us moving stuff up.

I've been looking to abstract this, but I don't think we're ready yet because too much is in flight and unmerged.
My hope is that they all get ParallelChanOp and GenerateSubs for free, at least, and then maybe some more.

[shazbert]

@shazbert Just realised that I haven't mentioned:

I completely agree about abstracting from exchange implementations. I'm just not yet ready to see how to do it. My intention was to finish merging all the exchange sub changes, then handle how multi-asset sub confs should work (which is GateIO + multi-sockets + assets ), and then review what hinge points stop us moving stuff up.

I've been looking to abstract this, but I don't think we're ready yet because too much is in flight and unmerged. My hope is that they all get ParallelChanOp and GenerateSubs for free, at least, and then maybe some more.

@gbjk I have added in new code now with fixes, when you have time could you please check them out.

[gbjk]

This doesn't feel right ...

1. Shouldn't Unsubscribe be erroring if it wasn't successful ?
2. Shouldn't the state of all subs in unsubs be changing ?
   That said, it's not too bad either.

[shazbert]

Shouldn't Unsubscribe be erroring if it wasn't successful ?

I think it should, I just added it as a catch all in the event we forgot to remove it from the store when it was successful. Then it should complain. should 😬

Shouldn't the state of all subs in unsubs be changing

Now you are making me think this is all completely wrong 😆. This specifically didn't catch any issues. Can you suggest a better way as a back up check? Cause I am drooling at my screen trying to figure it out 🤤.

[gbjk]

Looks good.
It's not too invasive, and doesn't change too much, so I'm happy.

I'm working on sub management and routing separately for gateio, but this doesn't tread on any toes.