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

shazbert · 2024-11-18T04:43:59Z

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 · 2024-11-18T04:54:06Z

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 · 2024-11-22T05:27:29Z

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.

exchanges/stream/websocket.go

shazbert · 2024-11-23T00:34:15Z

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 · 2024-11-25T01:04:54Z

@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 · 2025-01-15T02:44:36Z

@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 · 2025-01-15T03:26:37Z

exchanges/stream/websocket.go

+		if err := w.UnsubscribeChannels(c, unsubs); err != nil {
+			return err
+		}
+		if diff := prevState - store.Len(); diff != len(unsubs) {


This doesn't feel right ...

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

Shouldn't the state of all subs in unsubs be changing ?
That said, it's not too bad either.

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 🤤.

Sorry for late reply on this.
I think I'd want see that ranging the subs and unsubs have changed State, and that store contains (or doesn't) each one. ContainsAll and ContainsNone or something.
I really don't like checking len as a catchall, because it could false positive.
We're not locking store when we do any of this, afterall.

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.

gloriousCode

A strict rule, but it makes sense

exchanges/bybit/bybit_websocket.go

gloriousCode · 2025-01-28T21:33:47Z

exchanges/deribit/deribit_websocket.go

+	if method == "unsubscribe" {
+		return d.Websocket.RemoveSubscriptions(d.Websocket.Conn, subs...)
+	}
+	return d.Websocket.AddSubscriptions(d.Websocket.Conn, subs...)


Is there something I am missing with this?

This inclusion does not appear to make much sense to me, given that by this point all subscriptions have been set to the Subscribed state due the for loop above this

Why aren't you handling the unsubscribe in the for loop above?

Nice catch, didn't even notice that. None of those subs were being added to the subscription store and my changes complained. I will check it out.

exchanges/stream/websocket_test.go

exchanges/stream/websocket.go

…ifference method as its only used locally and Diff method can be accessed directly

…mplate for edge case perps with settlement

gloriousCode · 2025-01-29T02:11:45Z

exchanges/deribit/deribit_websocket.go

 	}
 	for _, s := range subs {
 		if _, ok := subAck[s.QualifiedChannel]; ok {
-			err = common.AppendError(err, d.Websocket.AddSuccessfulSubscriptions(d.Websocket.Conn, s))
+			if strings.Contains(method, "subscribe") {


I haven't tested yet, but "unsubscribe" contains the string "subscribe" and so both would likely fall in here

gbjk

Good work.

Only nits.

I'm asking for a change to not tear everything down on single sub failure, but if you don't make that change, it's no worse than before, so I'm fine.

exchanges/bybit/bybit_types.go

exchanges/bybit/bybit_websocket.go

gbjk · 2025-01-29T03:06:30Z

exchanges/deribit/deribit_websocket.go

@@ -123,6 +123,8 @@ var (
 	}
 )

+var errSubscriptionFailureDetected = errors.New("subscription failure detected")


This feels like it's stream.ErrSubscriptionFailure in disguise 🤠

gbjk · 2025-01-29T03:07:28Z

exchanges/deribit/deribit_websocket.go

@@ -779,6 +781,7 @@ func (d *Deribit) GetSubscriptionTemplate(_ *subscription.Subscription) (*templa
 		"channelName":     channelName,
 		"interval":        channelInterval,
 		"isSymbolChannel": isSymbolChannel,
+		"fmt":             formatPerpetualPairWithSettlement,


Nit: Don't think we can call this fmt when it's specific to Futures.

It calls it against every pair, are you asking me to only call it against futures?

Nope. But I think it's misleading that it's not really format for every pair, even if it's called for all of them.
So I guess I'd vote for: Rename the template func to fmtFuturePairs or rename the func called to formatChannelPair.
What's catching me is the disconnect between what's in the template and the actual func name.
Can also ignore.

exchanges/deribit/deribit_websocket.go

gbjk · 2025-01-29T03:11:52Z

exchanges/deribit/deribit_websocket.go

@@ -827,16 +831,23 @@ func (d *Deribit) handleSubscription(method string, subs subscription.List) erro
 		subAck[c] = true
 	}
 	if len(subAck) != len(subs) {
-		err = common.ErrUnknownError
+		err = errSubscriptionFailureDetected


So, this would be covering ... any subs that come back that we didn't expect, which won't be reported in the for range below?
It feels like it'd be better to delete from subAck as we range subs, and then error to say "Unexpected channels in Result" or something.
Cos otherwise, this error will be impossible to diagnose and I'd probably say misleading.

gbjk · 2025-01-29T03:21:53Z

exchanges/stream/websocket.go

+		if len(subs) != 0 && w.connectionManager[i].Subscriptions.Len() != len(subs) {
+			multiConnectFatalError = fmt.Errorf("%v %w expecting %d subscribed %v", w.exchangeName, errSubscriptionsNotAdded, len(subs), subs)
+			break


I still have this problem:
We call connect on an exchange, one pair on one asset fails to subscribe (like TRUMPEW the other day) and all connections get torn down for all asset types and subs.
I still don't agree with that.
The biggest problem I think I have is the strong coupling between connections and subs.
If a consumer wants to connect the websocket without subs, and then later call Subscribe manually, we both allow it, and don't support it at the same time.

I think this is a wider topic, because this change doesn't make 454/460 worse, but it doubles down on the concept.

I think we need to say "Connecting is connecting, and Subscribing is separate" especially when new subs or resubs could connect new subs.

I'll continue to work on this, so I'm just highlighting for alignment on direction.

In the meantime, I'd gently request again to log the error, and return it, but not tear down all of the subs and conns for all assets.

We call connect on an exchange, one pair on one asset fails to subscribe (like TRUMPEW the other day) and all connections get torn down for all asset types and subs.

Do you remember what specifically caused this?

The coupling connect() with subs is not great, regarding "Connecting is connecting, and Subscribing is separate" though is I think connect function might be changed to handle connections that don't require subscriptions and spawn a middleware handler that performs a JIT connection based on current/incoming subscriptions which also drops connections when unsubscribing when subs is empty, that way it scales a bit better with respect to a connections max sub capacity. I don't currently have a design that could help you in that respect though.

In the meantime, I'd gently request again to log the error, and return it, but not tear down all of the subs and conns for all assets.

Sure, if it's easier for everyone else I will just log it and I can just modify my own trading branch to be throwing the baby out with the bathwater myself. 🤷

@gloriousCode @thrasher- I pivot off these errors in my own trading branches that retries connect until it establish a clean slate or base line of full subscriptions across all trading pairs that I require. Happy to just log this out like GK suggests if we all agree and I will update tests.

Do you remember what specifically caused this?

A badly listed futures pair erroring on orderbook.

I pivot off these errors in my own trading branches that retries connect until it establish a clean slate or base line of full subscriptions across all trading pairs that I require.

Okay. This might be the crux of our disagreement.
You should not need to disconnect and unsub everything just because one thing fails, if your goal is to ensure it eventually works.
connectionManager should be monitoring and reconnecting, and if it doesn't have requisite granularity we should improve that.
And I think subscriptionManager should manager subscriptions.

What I'm aiming for is smaller responsibilities for functions that have names (and meanings) like connect, so they're more versatile (or composable) and easier to test.

Can you throw me any other requirements you might have like "retry subs and conns until everything is working", if you have any ?

I see exactly the same situation, and also wanna know "When's everything looking good?".

What I'm aiming for is smaller responsibilities for functions that have names (and meanings) like connect, so they're more versatile (or composable) and easier to test.

Good direction to have, I agree.

Can you throw me any other requirements you might have like "retry subs and conns until everything is working", if you have any ? && "When's everything looking good?"

Pretty much just this; Personally, I am still going to rely on ensuring a clean slate of full subscriptions before proceeding. If even one subscription fails, it indicates a potential issue with the connection or the subscription logic, and I prefer to retry until everything is established correctly, or bring the instance offline and fix it.

I see gk's issue with regards to killing all connections, but I don't see it as something huge because we already kill everything when Subscribe is called and return errors. The refactoring you speak of gk will change that section regardless, but I get not wanting to tangle further.

Logging the issue isn't actionable via code and so I don't think that's the best way to go

I propose we move subscribing for multi-connections to the end of the connect() function. It is a small change™️ that satisfies the following criteria:

The default handling of websocket connections via websocketroutine_manager.go will log any errors returned

It won't by default kill all connections when subscriptions have issue

It allows users who call Connect() directly to handle any errors that are returned, and resubscribe/reconnect/panic("help!")

Gist:
https://gist.github.com/gloriousCode/609d25ea1ee9b954ecbb95b505c76609

It comes with the caveat that I haven't tested things thoroughly. I tried things like separating subscribing, but that's way too large a refactor for this PR and comes with many extra considerations that I don't think are what is desired out of this PR - which is to ensure all subscriptions are subscribed to and that the caller knows it

Does this solution make people happy?

Yep, that's a good compromise with minimal intrusion.

Check me please 🙏

gbjk · 2025-01-29T03:27:02Z

exchanges/stream/websocket.go

+		if err := w.UnsubscribeChannels(c, unsubs); err != nil {
+			return err
+		}
+		if diff := prevState - store.Len(); diff != len(unsubs) {


Sorry for late reply on this.
I think I'd want see that ranging the subs and unsubs have changed State, and that store contains (or doesn't) each one. ContainsAll and ContainsNone or something.
I really don't like checking len as a catchall, because it could false positive.
We're not locking store when we do any of this, afterall.

exchanges/stream/websocket.go

gbjk

Great work.
Sorry for change request. Spent a while debating. 😢

gbjk · 2025-01-29T08:12:54Z

exchanges/subscription/store.go

+// PartitionByPresence returns two lists; one with the subscriptions that are present in the store and one with the
+// subscriptions that are not
+func (s *Store) PartitionByPresence(compare List) (matched, unmatched List) {


Thought about this for a while, and I think it's cute, but wrong to have a function where you're always using one or the other return.
I thought about giving it a bool for whether you want it there or not.
Took me a while to work out why "Presence" is wrong; It's a noun. Present is more correct, having an adjective form, but still sounds weird.

I'd vote for 2 funcs:

Contained returns a List of subs in compare which are in the store already

alt. Filter, Matching, Match, Union, Contains (but sounds boolish)

Missing returns a List of subs in compare which are not in the store already

alt. Absent, NotContained, Excluded

Alternative note: FilterMatching is probably the best name for keeping a single func, but adding a "contained or missing bool"

I'm feeling a tingle about how similar this is to Diff, and yet not.
Also around Venn and terminology. Spent a while looking at slices and at wiki on set theory and any precedence on this in stack exchange, but came up with nothing good.

Yeah good suggestion. I will introduce contained and missing as it has a better approach and single responsibility is more efficient even though it increases code base. If you wanted a set theory naming convention could have Intersection and Difference but it becomes less obvious for people with little to no experience with that terminology. (me)

…untering subscription specific error paths

gloriousCode · 2025-01-30T22:36:27Z

exchanges/stream/websocket.go

+	errSubscriptionsNotAdded                = errors.New("subscriptions not added")
+	errSubscriptionsNotRemoved              = errors.New("subscriptions not removed")


Now that these are being returned by Connect(), I think it would be nice to export these so you can help distinguish an error from connecting or an error for subscribing and act accordingly

gbjk · 2025-01-31T03:59:46Z

exchanges/subscription/store.go

 	}
 	s.mu.RLock()
 	defer s.mu.RUnlock()
+	var matched List


Can move this to the signature if you want, and/but still use explicit return with it.

gbjk · 2025-01-31T04:01:06Z

exchanges/subscription/store.go

+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	var unmatched List


Whilst considering maybe moving this to the signature, can also consider calling it missing for consistency.

gbjk

All good from me 👍 Great work.

Comments are entirely optional notes.

gloriousCode

tACK!

shazbert added review me This pull request is ready for review medium priority labels Nov 18, 2024

shazbert requested a review from a team November 18, 2024 04:43

shazbert self-assigned this Nov 18, 2024

gbjk requested changes Nov 22, 2024

View reviewed changes

exchanges/stream/websocket.go Outdated Show resolved Hide resolved

shazbert added blocked and removed medium priority review me This pull request is ready for review labels Nov 25, 2024

shazbert added 3 commits January 15, 2025 12:10

Add check for subscription store insertion and validation with tests

3e7fe77

bybit: fix subscriptions

457cba1

deribit: fix subscriptions

dc72434

shazbert force-pushed the ws_subs_abs branch from f1ca7bd to dc72434 Compare January 15, 2025 02:33

Merge branch 'master' into ws_subs_abs

5ce010d

shazbert added bug review me This pull request is ready for review and removed blocked labels Jan 15, 2025

shazbert changed the title ~~stream/gateio: move adding and removing subscriptions from websocket wrapper to stream package~~ stream: force subscription store count check as stop gap for wrapper side implementation Jan 15, 2025

linter: fix

6ec3204

shazbert requested a review from gbjk January 15, 2025 02:42

gbjk reviewed Jan 15, 2025

View reviewed changes

gbjk approved these changes Jan 15, 2025

View reviewed changes

gloriousCode requested changes Jan 28, 2025

View reviewed changes

Ryan O'Hara-Reid added 3 commits January 29, 2025 09:33

Merge branch 'master' into ws_subs_abs

fadf919

glorious/nits: add test for updateChannelSubscriptions RM GetChannelD…

759e225

…ifference method as its only used locally and Diff method can be accessed directly

glorious/nits: add to store in the loop; add correct formatting to te…

fcf1d23

…mplate for edge case perps with settlement

shazbert requested review from gloriousCode and gbjk January 29, 2025 02:08

spelling: fix

7147815

gloriousCode reviewed Jan 29, 2025

View reviewed changes

glorious/nit: silly billy

7401285

gbjk approved these changes Jan 29, 2025

View reviewed changes

gk: nits

102bfc1

shazbert requested review from gloriousCode and thrasher- January 29, 2025 05:46

gbjk requested changes Jan 29, 2025

View reviewed changes

gk/nits: split PartitionByPresence into Contained and Missing

1e3eeda

shazbert changed the title ~~stream: force subscription store count check as stop gap for wrapper side implementation~~ stream: force subscription store check as stop gap for wrapper side implementation Jan 29, 2025

shazbert added 3 commits January 31, 2025 07:31

gk/nits: formatPerpetualPairWithSettlement -> formatChannelPair

9d7f116

stream/websocket: stop full websocket disconnect on Connect when enco…

ba022ec

…untering subscription specific error paths

stream/websocket: rm nil assignment

06380e0

shazbert requested a review from gbjk January 30, 2025 21:15

gloriousCode reviewed Jan 30, 2025

View reviewed changes

glorious: nits

aca87c3

gbjk reviewed Jan 31, 2025

View reviewed changes

gbjk approved these changes Jan 31, 2025

View reviewed changes

gk: niterinos

6c6fb30

shazbert requested a review from gloriousCode February 2, 2025 23:54

gloriousCode approved these changes Feb 7, 2025

View reviewed changes

gloriousCode added the gcrc GloriousCode Review Complete label Feb 9, 2025

		errSubscriptionsNotAdded = errors.New("subscriptions not added")
		errSubscriptionsNotRemoved = errors.New("subscriptions not removed")

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

Are you sure you want to change the base?

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

Conversation

shazbert commented Nov 18, 2024 • edited Loading

PR Description

Type of change

How has this been tested

Checklist

codecov bot commented Nov 18, 2024 • edited Loading

Codecov Report

gbjk commented Nov 22, 2024 • edited Loading

gbjk left a comment

Choose a reason for hiding this comment

shazbert commented Nov 23, 2024

gbjk commented Nov 25, 2024

shazbert commented Jan 15, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gbjk left a comment

Choose a reason for hiding this comment

gloriousCode left a comment

Choose a reason for hiding this comment

gloriousCode Jan 28, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gbjk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gloriousCode Jan 29, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gbjk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gbjk left a comment

Choose a reason for hiding this comment

gloriousCode left a comment

Choose a reason for hiding this comment

shazbert commented Nov 18, 2024 •

edited

Loading

codecov bot commented Nov 18, 2024 •

edited

Loading

gbjk commented Nov 22, 2024 •

edited

Loading

gloriousCode Jan 28, 2025 •

edited

Loading

gloriousCode Jan 29, 2025 •

edited

Loading