Skip to content

Commit

Permalink
Move helper to util module. Test for streaming util.
Browse files Browse the repository at this point in the history
  • Loading branch information
UnbornAztecKing committed Jan 9, 2025
1 parent 05ff9f3 commit 309f2ec
Show file tree
Hide file tree
Showing 3 changed files with 147 additions and 23 deletions.
33 changes: 10 additions & 23 deletions protocol/streaming/full_node_streaming_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@ import (
ante_types "github.com/dydxprotocol/v4-chain/protocol/app/ante/types"
"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
"github.com/dydxprotocol/v4-chain/protocol/streaming/types"
"github.com/dydxprotocol/v4-chain/protocol/streaming/util"
streaming_util "github.com/dydxprotocol/v4-chain/protocol/streaming/util"
clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types"

Expand Down Expand Up @@ -191,29 +192,11 @@ func (sm *FullNodeStreamingManagerImpl) getNextAvailableSubscriptionId() uint32
return id
}

// TODO best practice for ensuring all cases are handled
// default error? default panic?
func GetOffChannelUpdateV1SubaccountId(update ocutypes.OffChainUpdateV1) uint32 {
var orderSubaccountIdNumber uint32
switch ocu1 := update.UpdateMessage.(type) {
case *ocutypes.OffChainUpdateV1_OrderPlace:
orderSubaccountIdNumber = ocu1.OrderPlace.Order.OrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderRemove:
orderSubaccountIdNumber = ocu1.OrderRemove.RemovedOrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderUpdate:
orderSubaccountIdNumber = ocu1.OrderUpdate.OrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderReplace:
orderSubaccountIdNumber = ocu1.OrderReplace.Order.OrderId.SubaccountId.Number

}
return orderSubaccountIdNumber
}

// Filter StreamUpdates for subaccountIdNumbers
// If a StreamUpdate_OrderUpdate contains no updates for subscribed subaccounts, drop message
// If a StreamUpdate_OrderUpdate contains updates for subscribed subaccounts, construct a new
// StreamUpdate_OrderUpdate with updates only for subscribed subaccounts
func (sub *OrderbookSubscription) FilterSubaccountStreamUpdates(output chan []clobtypes.StreamUpdate) {
func (sub *OrderbookSubscription) FilterSubaccountStreamUpdates(output chan []clobtypes.StreamUpdate, logger log.Logger) {
subaccountIdNumbers := make([]uint32, len(sub.subaccountIds))
for i, subaccountId := range sub.subaccountIds {
subaccountIdNumbers[i] = subaccountId.Number
Expand All @@ -227,9 +210,13 @@ func (sub *OrderbookSubscription) FilterSubaccountStreamUpdates(output chan []cl
case *clobtypes.StreamUpdate_OrderbookUpdate:
orderBookUpdates := []ocutypes.OffChainUpdateV1{}
for _, orderBookUpdate := range updateMessage.OrderbookUpdate.Updates {
orderBookUpdateSubaccountIdNumber := GetOffChannelUpdateV1SubaccountId(orderBookUpdate)
if slices.Contains(subaccountIdNumbers, orderBookUpdateSubaccountIdNumber) {
orderBookUpdates = append(orderBookUpdates, orderBookUpdate)
orderBookUpdateSubaccountIdNumber, err := util.GetOffChainUpdateV1SubaccountIdNumber(orderBookUpdate)
if err != nil {
if slices.Contains(subaccountIdNumbers, orderBookUpdateSubaccountIdNumber) {
orderBookUpdates = append(orderBookUpdates, orderBookUpdate)
}
} else {
logger.Error(err.Error())
}
}
// Drop the StreamUpdate_OrderbookUpdate if all updates inside were dropped
Expand Down Expand Up @@ -347,7 +334,7 @@ func (sm *FullNodeStreamingManagerImpl) Subscribe(
}
filteredUpdateChannel = make(chan []clobtypes.StreamUpdate)
defer close(filteredUpdateChannel)
go subscription.FilterSubaccountStreamUpdates(filteredUpdateChannel)
go subscription.FilterSubaccountStreamUpdates(filteredUpdateChannel, sm.logger)
}

// Use current goroutine to consistently poll subscription channel for updates
Expand Down
19 changes: 19 additions & 0 deletions protocol/streaming/util/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,3 +21,22 @@ func GetOffchainUpdatesV1(offchainUpdates *clobtypes.OffchainUpdates) []ocutypes
}
return v1updates
}

// TODO best practice for ensuring all cases are handled
// default error? default panic?
func GetOffChainUpdateV1SubaccountIdNumber(update ocutypes.OffChainUpdateV1) (uint32, error) {
var orderSubaccountIdNumber uint32
switch ocu1 := update.UpdateMessage.(type) {
case *ocutypes.OffChainUpdateV1_OrderPlace:
orderSubaccountIdNumber = ocu1.OrderPlace.Order.OrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderRemove:
orderSubaccountIdNumber = ocu1.OrderRemove.RemovedOrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderUpdate:
orderSubaccountIdNumber = ocu1.OrderUpdate.OrderId.SubaccountId.Number
case *ocutypes.OffChainUpdateV1_OrderReplace:
orderSubaccountIdNumber = ocu1.OrderReplace.Order.OrderId.SubaccountId.Number
default:
return 0, fmt.Errorf("UpdateMessage type not in {OrderPlace, OrderRemove, OrderUpdate, OrderReplace}")
}
return orderSubaccountIdNumber, nil
}
118 changes: 118 additions & 0 deletions protocol/streaming/util/util_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,118 @@
package util_test

import (
"fmt"
"testing"
"time"

"github.com/stretchr/testify/require"

ocutypes "github.com/dydxprotocol/v4-chain/protocol/indexer/off_chain_updates/types"
pv1types "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1/types"
stypes "github.com/dydxprotocol/v4-chain/protocol/indexer/shared/types"
"github.com/dydxprotocol/v4-chain/protocol/streaming/util"
)

func _ToPtr[V any](v V) *V {
return &v
}

func TestGetOffChainUpdateV1SubaccountIdNumber(t *testing.T) {
subaccountIdNumber := uint32(1337)
orderId := pv1types.IndexerOrderId{
SubaccountId: pv1types.IndexerSubaccountId{
Owner: "foo",
Number: uint32(subaccountIdNumber),
},
ClientId: 0,
OrderFlags: 0,
ClobPairId: 0,
}
order := pv1types.IndexerOrder{
OrderId: orderId,
Side: pv1types.IndexerOrder_SIDE_BUY,
Quantums: uint64(10 ^ 6),
Subticks: 1,
GoodTilOneof: &pv1types.IndexerOrder_GoodTilBlock{
GoodTilBlock: 10 ^ 9,
},
TimeInForce: 10 ^ 9,
ReduceOnly: false,
ClientMetadata: 0,
ConditionType: pv1types.IndexerOrder_CONDITION_TYPE_UNSPECIFIED,
ConditionalOrderTriggerSubticks: 0,
}
newOrder := order
newOrder.Quantums += 10 ^ 6

orderPlaceTime := time.Now()
fillQuantums := uint64(1988)

tests := map[string]struct {
update ocutypes.OffChainUpdateV1
id uint32
err error
}{
"OrderPlace": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: &ocutypes.OffChainUpdateV1_OrderPlace{
OrderPlace: &ocutypes.OrderPlaceV1{
Order: &order,
PlacementStatus: ocutypes.OrderPlaceV1_ORDER_PLACEMENT_STATUS_BEST_EFFORT_OPENED,
TimeStamp: _ToPtr(orderPlaceTime),
},
},
},
id: subaccountIdNumber,
err: nil,
},
"OrderRemove": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: &ocutypes.OffChainUpdateV1_OrderRemove{
OrderRemove: &ocutypes.OrderRemoveV1{
RemovedOrderId: &orderId,
Reason: stypes.OrderRemovalReason_ORDER_REMOVAL_REASON_USER_CANCELED,
RemovalStatus: ocutypes.OrderRemoveV1_ORDER_REMOVAL_STATUS_CANCELED,
TimeStamp: _ToPtr(orderPlaceTime.Add(1 * time.Second)),
},
},
},
id: subaccountIdNumber,
err: nil,
},
"OrderUpdate": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: &ocutypes.OffChainUpdateV1_OrderUpdate{
OrderUpdate: &ocutypes.OrderUpdateV1{
OrderId: &orderId,
TotalFilledQuantums: fillQuantums,
},
},
},
id: subaccountIdNumber,
err: nil,
},
"OrderReplace": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: &ocutypes.OffChainUpdateV1_OrderReplace{
OrderReplace: &ocutypes.OrderReplaceV1{
OldOrderId: &orderId,
Order: &newOrder,
PlacementStatus: ocutypes.OrderPlaceV1_ORDER_PLACEMENT_STATUS_OPENED,
TimeStamp: _ToPtr(orderPlaceTime.Add(3 * time.Second)),
},
},
},
id: subaccountIdNumber,
err: nil,
},
}
for name, testCase := range tests {
t.Run(name, func(t *testing.T) {
id, err := util.GetOffChainUpdateV1SubaccountIdNumber(testCase.update)
fmt.Println("expected", id)
require.Equal(t, err, testCase.err)
require.Equal(t, id, testCase.id)
})
}
}

0 comments on commit 309f2ec

Please sign in to comment.