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

exchanges/bybit/bybit_types.go

@@ -38,7 +38,7 @@ type SubscriptionArgument struct {
 	RequestID      string            `json:"req_id"`
 	Operation      string            `json:"op"`
 	Arguments      []string          `json:"args"`
-	associatedSubs subscription.List `json:"-"` // Used to store associated subscriptions
+	associatedSubs subscription.List `json:"-"`
 }
 
 // Fee holds fee information

exchanges/bybit/bybit_websocket.go

@@ -167,32 +167,19 @@ func (by *Bybit) handleSubscriptions(operation string, subs subscription.List) (
 	if err != nil {
 		return
 	}
-	var chans subscription.List
-	var authChans subscription.List
-	for _, s := range subs {
-		if s.Authenticated {
-			authChans = append(authChans, s)
-		} else {
-			chans = append(chans, s)
+
+	for _, list := range []subscription.List{subs.Public(), subs.Private()} {
+		for _, b := range common.Batch(list, 10) {
+			args = append(args, SubscriptionArgument{
+				auth:           b[0].Authenticated,
+				Operation:      operation,
+				RequestID:      strconv.FormatInt(by.Websocket.Conn.GenerateMessageID(false), 10),
+				Arguments:      b.QualifiedChannels(),
+				associatedSubs: b,
+			})
 		}
 	}
-	for _, b := range common.Batch(chans, 10) {
-		args = append(args, SubscriptionArgument{
-			Operation:      operation,
-			RequestID:      strconv.FormatInt(by.Websocket.Conn.GenerateMessageID(false), 10),
-			Arguments:      b.QualifiedChannels(),
-			associatedSubs: b,
-		})
-	}
-	if len(authChans) != 0 {
-		args = append(args, SubscriptionArgument{
-			auth:           true,
-			Operation:      operation,
-			RequestID:      strconv.FormatInt(by.Websocket.Conn.GenerateMessageID(false), 10),
-			Arguments:      authChans.QualifiedChannels(),
-			associatedSubs: authChans,
-		})
-	}
+
 	return
 }
 

exchanges/deribit/deribit_websocket.go

@@ -123,8 +123,6 @@ var (
 	}
 )
 
-var errSubscriptionFailureDetected = errors.New("subscription failure detected")
-
 // WsConnect starts a new connection with the websocket API
 func (d *Deribit) WsConnect() error {
 	if !d.Websocket.IsEnabled() || !d.IsEnabled() {
@@ -831,19 +829,25 @@ func (d *Deribit) handleSubscription(method string, subs subscription.List) erro
 		subAck[c] = true
 	}
 	if len(subAck) != len(subs) {
-		err = errSubscriptionFailureDetected
+		err = stream.ErrSubscriptionFailure
 	}
 	for _, s := range subs {
 		if _, ok := subAck[s.QualifiedChannel]; ok {
+			delete(subAck, s.QualifiedChannel)
 			if !strings.Contains(method, "unsubscribe") {
 				err = common.AppendError(err, d.Websocket.AddSuccessfulSubscriptions(d.Websocket.Conn, s))
 			} else {
 				err = common.AppendError(err, d.Websocket.RemoveSubscriptions(d.Websocket.Conn, s))
 			}
 		} else {
-			err = common.AppendError(err, errors.New(s.String()+"failed to"+method))
+			err = common.AppendError(err, errors.New(s.String()+" failed to "+method))
 		}
 	}
+
+	for key := range subAck {
+		err = common.AppendError(err, fmt.Errorf("unexpected channel `%s` in result", key))
+	}
+
 	if err != nil {
 		return err
 	}

exchanges/stream/websocket.go

@@ -375,8 +375,8 @@ func (w *Websocket) connect() error {
 				return err
 			}
 
-			if w.subscriptions.Len() != len(subs) {
-				return fmt.Errorf("%s %w expecting %d subscribed", w.exchangeName, errSubscriptionsNotAdded, len(subs))
+			if _, unmatched := w.subscriptions.PartitionByPresence(subs); len(unmatched) > 0 {
+				return fmt.Errorf("%v %w `%s`", w.exchangeName, errSubscriptionsNotAdded, unmatched)
 			}
 		}
 		return nil
@@ -461,8 +461,8 @@ func (w *Websocket) connect() error {
 			break
 		}
 
-		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)
+		if _, unmatched := w.connectionManager[i].Subscriptions.PartitionByPresence(subs); len(subs) > 0 && len(unmatched) > 0 {
+			multiConnectFatalError = fmt.Errorf("%v %w `%s`", w.exchangeName, errSubscriptionsNotAdded, unmatched)
 			break
 		}
 
@@ -682,26 +682,23 @@ func (w *Websocket) FlushChannels() error {
 // updateChannelSubscriptions subscribes or unsubscribes from channels and checks that the correct number of channels
 // have been subscribed to or unsubscribed from.
 func (w *Websocket) updateChannelSubscriptions(c Connection, store *subscription.Store, incoming subscription.List) error {
-	if store == nil {
-		return fmt.Errorf("%w for %T", common.ErrNilPointer, store)
-	}
 	subs, unsubs := store.Diff(incoming)
 	if len(unsubs) != 0 {
-		prevState := store.Len()
 		if err := w.UnsubscribeChannels(c, unsubs); err != nil {
 			return err
 		}
-		if diff := prevState - store.Len(); diff != len(unsubs) {
-			return fmt.Errorf("%v %w expected %d unsubscribed", w.exchangeName, errSubscriptionsNotRemoved, len(unsubs))
+
+		if matched, _ := store.PartitionByPresence(unsubs); len(matched) > 0 {
+			return fmt.Errorf("%v %w `%s`", w.exchangeName, errSubscriptionsNotRemoved, matched)
 		}
 	}
 	if len(subs) != 0 {
-		prevState := store.Len()
 		if err := w.SubscribeToChannels(c, subs); err != nil {
 			return err
 		}
-		if diff := store.Len() - prevState; diff != len(subs) {
-			return fmt.Errorf("%v %w expected %d subscribed", w.exchangeName, errSubscriptionsNotAdded, len(subs))
+
+		if _, unmatched := store.PartitionByPresence(subs); len(unmatched) > 0 {
+			return fmt.Errorf("%v %w `%s`", w.exchangeName, errSubscriptionsNotAdded, unmatched)
 		}
 	}
 	return nil

exchanges/stream/websocket_test.go

@@ -252,7 +252,7 @@ func TestConnectionMessageErrors(t *testing.T) {
 	require.ErrorIs(t, err, errDastardlyReason)
 
 	ws.connectionManager[0].Setup.GenerateSubscriptions = func() (subscription.List, error) {
-		return subscription.List{{}}, nil
+		return subscription.List{{Channel: "test"}}, nil
 	}
 	err = ws.Connect()
 	require.ErrorIs(t, err, errNoConnectFunc)
@@ -1585,12 +1585,8 @@ func TestUpdateChannelSubscriptions(t *testing.T) {
 	t.Parallel()
 
 	ws := Websocket{}
-	err := ws.updateChannelSubscriptions(nil, nil, nil)
-	require.ErrorIs(t, err, common.ErrNilPointer)
-
 	store := subscription.NewStore()
-
-	err = ws.updateChannelSubscriptions(nil, store, subscription.List{{Channel: "test"}})
+	err := ws.updateChannelSubscriptions(nil, store, subscription.List{{Channel: "test"}})
 	require.ErrorIs(t, err, common.ErrNilPointer)
 	require.Zero(t, store.Len())
 

exchanges/subscription/store.go

@@ -206,3 +206,21 @@ func (s *Store) Len() int {
 	defer s.mu.RUnlock()
 	return len(s.m)
 }
+
+// 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) {
+	if s == nil || s.m == nil {
+		return nil, compare // All are unmatched
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	for _, sub := range compare {
+		if found := s.get(sub); found != nil {
+			matched = append(matched, found)
+		} else {
+			unmatched = append(unmatched, sub)
+		}
+	}
+	return
+}

exchanges/subscription/store_test.go

@@ -204,3 +204,35 @@ func EqualLists(tb testing.TB, a, b List) {
 		assert.Fail(tb, fail, "Subscriptions should be equal")
 	}
 }
+
+func TestPartitionByPresence(t *testing.T) {
+	t.Parallel()
+
+	var s *Store
+	matched, unmatched := s.PartitionByPresence(nil)
+	assert.Nil(t, matched)
+	assert.Nil(t, unmatched)
+
+	matched, unmatched = s.PartitionByPresence(List{{Channel: TickerChannel}})
+	assert.Nil(t, matched)
+	assert.Len(t, unmatched, 1)
+
+	s = NewStore()
+	matched, unmatched = s.PartitionByPresence(nil)
+	assert.Nil(t, matched)
+	assert.Nil(t, unmatched)
+
+	matched, unmatched = s.PartitionByPresence(List{})
+	assert.Nil(t, matched)
+	assert.Nil(t, unmatched)
+
+	matched, unmatched = s.PartitionByPresence(List{{Channel: TickerChannel}})
+	assert.Nil(t, matched)
+	assert.Len(t, unmatched, 1)
+
+	require.NoError(t, s.add(&Subscription{Channel: TickerChannel}))
+
+	matched, unmatched = s.PartitionByPresence(List{{Channel: TickerChannel}})
+	assert.Len(t, matched, 1)
+	assert.Nil(t, unmatched)
+}