Skip to content

Commit 2a47cf5

Browse files
jayy04jayy04
authored
[Backport v3.x] backport full node streaming PRs (#1181)
* [CT-645] Move off chain updates and v1 to a different package (#1131) * [CT-645] Add protos for orderbook stream query service * move removal reasons to a separate package * [CT-645] Add protos for orderbook stream query service (#1133) * [CT-645] Add protos for orderbook stream query service * make update not nullable * fix build * [CT-644] instantiate grpc stream manager (#1134) * [CT-644] instantiate grpc stream manager * update type * update channel type * [CT-646] stream offchain updates through stream manager (#1138) * [CT-646] stream offchain updates through stream manager * comments * fix lint * get rid of finished * comments * comments * [CT-652] add command line flag for full node streaming (#1145) * [CT-647] construct the initial orderbook snapshot (#1147) * [CT-647] construct the initial orderbook snapshot * [CT-647] initialize new streams and send orderbook snapshot (#1152) * [CT-647] initialize new streams and send orderbook snapshot * use sync once * comments * fix test
1 parent 85fb10a commit 2a47cf5

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

51 files changed

+1890
-588
lines changed

‎indexer/packages/v4-protos/src/codegen/dydxprotocol/clob/query.rpc.Query.ts

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
import { Rpc } from "../../helpers";
22
import * as _m0 from "protobufjs/minimal";
33
import { QueryClient, createProtobufRpcClient } from "@cosmjs/stargate";
4-
import { QueryGetClobPairRequest, QueryClobPairResponse, QueryAllClobPairRequest, QueryClobPairAllResponse, MevNodeToNodeCalculationRequest, MevNodeToNodeCalculationResponse, QueryEquityTierLimitConfigurationRequest, QueryEquityTierLimitConfigurationResponse, QueryBlockRateLimitConfigurationRequest, QueryBlockRateLimitConfigurationResponse, QueryLiquidationsConfigurationRequest, QueryLiquidationsConfigurationResponse } from "./query";
4+
import { QueryGetClobPairRequest, QueryClobPairResponse, QueryAllClobPairRequest, QueryClobPairAllResponse, MevNodeToNodeCalculationRequest, MevNodeToNodeCalculationResponse, QueryEquityTierLimitConfigurationRequest, QueryEquityTierLimitConfigurationResponse, QueryBlockRateLimitConfigurationRequest, QueryBlockRateLimitConfigurationResponse, QueryLiquidationsConfigurationRequest, QueryLiquidationsConfigurationResponse, StreamOrderbookUpdatesRequest, StreamOrderbookUpdatesResponse } from "./query";
55
/** Query defines the gRPC querier service. */
66

77
export interface Query {
@@ -22,6 +22,9 @@ export interface Query {
2222
/** Queries LiquidationsConfiguration. */
2323

2424
liquidationsConfiguration(request?: QueryLiquidationsConfigurationRequest): Promise<QueryLiquidationsConfigurationResponse>;
25+
/** Streams orderbook updates. */
26+
27+
streamOrderbookUpdates(request: StreamOrderbookUpdatesRequest): Promise<StreamOrderbookUpdatesResponse>;
2528
}
2629
export class QueryClientImpl implements Query {
2730
private readonly rpc: Rpc;
@@ -34,6 +37,7 @@ export class QueryClientImpl implements Query {
3437
this.equityTierLimitConfiguration = this.equityTierLimitConfiguration.bind(this);
3538
this.blockRateLimitConfiguration = this.blockRateLimitConfiguration.bind(this);
3639
this.liquidationsConfiguration = this.liquidationsConfiguration.bind(this);
40+
this.streamOrderbookUpdates = this.streamOrderbookUpdates.bind(this);
3741
}
3842

3943
clobPair(request: QueryGetClobPairRequest): Promise<QueryClobPairResponse> {
@@ -74,6 +78,12 @@ export class QueryClientImpl implements Query {
7478
return promise.then(data => QueryLiquidationsConfigurationResponse.decode(new _m0.Reader(data)));
7579
}
7680

81+
streamOrderbookUpdates(request: StreamOrderbookUpdatesRequest): Promise<StreamOrderbookUpdatesResponse> {
82+
const data = StreamOrderbookUpdatesRequest.encode(request).finish();
83+
const promise = this.rpc.request("dydxprotocol.clob.Query", "StreamOrderbookUpdates", data);
84+
return promise.then(data => StreamOrderbookUpdatesResponse.decode(new _m0.Reader(data)));
85+
}
86+
7787
}
7888
export const createRpcQueryExtension = (base: QueryClient) => {
7989
const rpc = createProtobufRpcClient(base);
@@ -101,6 +111,10 @@ export const createRpcQueryExtension = (base: QueryClient) => {
101111

102112
liquidationsConfiguration(request?: QueryLiquidationsConfigurationRequest): Promise<QueryLiquidationsConfigurationResponse> {
103113
return queryService.liquidationsConfiguration(request);
114+
},
115+
116+
streamOrderbookUpdates(request: StreamOrderbookUpdatesRequest): Promise<StreamOrderbookUpdatesResponse> {
117+
return queryService.streamOrderbookUpdates(request);
104118
}
105119

106120
};

‎indexer/packages/v4-protos/src/codegen/dydxprotocol/clob/query.ts

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import { ClobPair, ClobPairSDKType } from "./clob_pair";
44
import { EquityTierLimitConfiguration, EquityTierLimitConfigurationSDKType } from "./equity_tier_limit_config";
55
import { BlockRateLimitConfiguration, BlockRateLimitConfigurationSDKType } from "./block_rate_limit_config";
66
import { LiquidationsConfig, LiquidationsConfigSDKType } from "./liquidations_config";
7+
import { OffChainUpdateV1, OffChainUpdateV1SDKType } from "../indexer/off_chain_updates/off_chain_updates";
78
import * as _m0 from "protobufjs/minimal";
89
import { DeepPartial, Long } from "../../helpers";
910
/** QueryGetClobPairRequest is request type for the ClobPair method. */
@@ -198,6 +199,58 @@ export interface QueryLiquidationsConfigurationResponse {
198199
export interface QueryLiquidationsConfigurationResponseSDKType {
199200
liquidations_config?: LiquidationsConfigSDKType;
200201
}
202+
/**
203+
* StreamOrderbookUpdatesRequest is a request message for the
204+
* StreamOrderbookUpdates method.
205+
*/
206+
207+
export interface StreamOrderbookUpdatesRequest {
208+
/** Clob pair ids to stream orderbook updates for. */
209+
clobPairId: number[];
210+
}
211+
/**
212+
* StreamOrderbookUpdatesRequest is a request message for the
213+
* StreamOrderbookUpdates method.
214+
*/
215+
216+
export interface StreamOrderbookUpdatesRequestSDKType {
217+
/** Clob pair ids to stream orderbook updates for. */
218+
clob_pair_id: number[];
219+
}
220+
/**
221+
* StreamOrderbookUpdatesResponse is a response message for the
222+
* StreamOrderbookUpdates method.
223+
*/
224+
225+
export interface StreamOrderbookUpdatesResponse {
226+
/** Orderbook updates for the clob pair. */
227+
updates: OffChainUpdateV1[];
228+
/**
229+
* Snapshot indicates if the response is from a snapshot of the orderbook.
230+
* This is true for the initial response and false for all subsequent updates.
231+
* Note that if the snapshot is true, then all previous entries should be
232+
* discarded and the orderbook should be resynced.
233+
*/
234+
235+
snapshot: boolean;
236+
}
237+
/**
238+
* StreamOrderbookUpdatesResponse is a response message for the
239+
* StreamOrderbookUpdates method.
240+
*/
241+
242+
export interface StreamOrderbookUpdatesResponseSDKType {
243+
/** Orderbook updates for the clob pair. */
244+
updates: OffChainUpdateV1SDKType[];
245+
/**
246+
* Snapshot indicates if the response is from a snapshot of the orderbook.
247+
* This is true for the initial response and false for all subsequent updates.
248+
* Note that if the snapshot is true, then all previous entries should be
249+
* discarded and the orderbook should be resynced.
250+
*/
251+
252+
snapshot: boolean;
253+
}
201254

202255
function createBaseQueryGetClobPairRequest(): QueryGetClobPairRequest {
203256
return {
@@ -789,4 +842,116 @@ export const QueryLiquidationsConfigurationResponse = {
789842
return message;
790843
}
791844

845+
};
846+
847+
function createBaseStreamOrderbookUpdatesRequest(): StreamOrderbookUpdatesRequest {
848+
return {
849+
clobPairId: []
850+
};
851+
}
852+
853+
export const StreamOrderbookUpdatesRequest = {
854+
encode(message: StreamOrderbookUpdatesRequest, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
855+
writer.uint32(10).fork();
856+
857+
for (const v of message.clobPairId) {
858+
writer.uint32(v);
859+
}
860+
861+
writer.ldelim();
862+
return writer;
863+
},
864+
865+
decode(input: _m0.Reader | Uint8Array, length?: number): StreamOrderbookUpdatesRequest {
866+
const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
867+
let end = length === undefined ? reader.len : reader.pos + length;
868+
const message = createBaseStreamOrderbookUpdatesRequest();
869+
870+
while (reader.pos < end) {
871+
const tag = reader.uint32();
872+
873+
switch (tag >>> 3) {
874+
case 1:
875+
if ((tag & 7) === 2) {
876+
const end2 = reader.uint32() + reader.pos;
877+
878+
while (reader.pos < end2) {
879+
message.clobPairId.push(reader.uint32());
880+
}
881+
} else {
882+
message.clobPairId.push(reader.uint32());
883+
}
884+
885+
break;
886+
887+
default:
888+
reader.skipType(tag & 7);
889+
break;
890+
}
891+
}
892+
893+
return message;
894+
},
895+
896+
fromPartial(object: DeepPartial<StreamOrderbookUpdatesRequest>): StreamOrderbookUpdatesRequest {
897+
const message = createBaseStreamOrderbookUpdatesRequest();
898+
message.clobPairId = object.clobPairId?.map(e => e) || [];
899+
return message;
900+
}
901+
902+
};
903+
904+
function createBaseStreamOrderbookUpdatesResponse(): StreamOrderbookUpdatesResponse {
905+
return {
906+
updates: [],
907+
snapshot: false
908+
};
909+
}
910+
911+
export const StreamOrderbookUpdatesResponse = {
912+
encode(message: StreamOrderbookUpdatesResponse, writer: _m0.Writer = _m0.Writer.create()): _m0.Writer {
913+
for (const v of message.updates) {
914+
OffChainUpdateV1.encode(v!, writer.uint32(10).fork()).ldelim();
915+
}
916+
917+
if (message.snapshot === true) {
918+
writer.uint32(16).bool(message.snapshot);
919+
}
920+
921+
return writer;
922+
},
923+
924+
decode(input: _m0.Reader | Uint8Array, length?: number): StreamOrderbookUpdatesResponse {
925+
const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
926+
let end = length === undefined ? reader.len : reader.pos + length;
927+
const message = createBaseStreamOrderbookUpdatesResponse();
928+
929+
while (reader.pos < end) {
930+
const tag = reader.uint32();
931+
932+
switch (tag >>> 3) {
933+
case 1:
934+
message.updates.push(OffChainUpdateV1.decode(reader, reader.uint32()));
935+
break;
936+
937+
case 2:
938+
message.snapshot = reader.bool();
939+
break;
940+
941+
default:
942+
reader.skipType(tag & 7);
943+
break;
944+
}
945+
}
946+
947+
return message;
948+
},
949+
950+
fromPartial(object: DeepPartial<StreamOrderbookUpdatesResponse>): StreamOrderbookUpdatesResponse {
951+
const message = createBaseStreamOrderbookUpdatesResponse();
952+
message.updates = object.updates?.map(e => OffChainUpdateV1.fromPartial(e)) || [];
953+
message.snapshot = object.snapshot ?? false;
954+
return message;
955+
}
956+
792957
};

‎proto/dydxprotocol/clob/query.proto

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import "dydxprotocol/clob/clob_pair.proto";
99
import "dydxprotocol/clob/equity_tier_limit_config.proto";
1010
import "dydxprotocol/clob/liquidations_config.proto";
1111
import "dydxprotocol/clob/mev.proto";
12+
import "dydxprotocol/indexer/off_chain_updates/off_chain_updates.proto";
1213

1314
option go_package = "github.com/dydxprotocol/v4-chain/protocol/x/clob/types";
1415

@@ -50,6 +51,12 @@ service Query {
5051
returns (QueryLiquidationsConfigurationResponse) {
5152
option (google.api.http).get = "/dydxprotocol/clob/liquidations_config";
5253
}
54+
55+
// GRPC Streams
56+
57+
// Streams orderbook updates.
58+
rpc StreamOrderbookUpdates(StreamOrderbookUpdatesRequest)
59+
returns (stream StreamOrderbookUpdatesResponse);
5360
}
5461

5562
// QueryGetClobPairRequest is request type for the ClobPair method.
@@ -126,3 +133,24 @@ message QueryLiquidationsConfigurationRequest {}
126133
message QueryLiquidationsConfigurationResponse {
127134
LiquidationsConfig liquidations_config = 1 [ (gogoproto.nullable) = false ];
128135
}
136+
137+
// StreamOrderbookUpdatesRequest is a request message for the
138+
// StreamOrderbookUpdates method.
139+
message StreamOrderbookUpdatesRequest {
140+
// Clob pair ids to stream orderbook updates for.
141+
repeated uint32 clob_pair_id = 1;
142+
}
143+
144+
// StreamOrderbookUpdatesResponse is a response message for the
145+
// StreamOrderbookUpdates method.
146+
message StreamOrderbookUpdatesResponse {
147+
// Orderbook updates for the clob pair.
148+
repeated dydxprotocol.indexer.off_chain_updates.OffChainUpdateV1 updates = 1
149+
[ (gogoproto.nullable) = false ];
150+
151+
// Snapshot indicates if the response is from a snapshot of the orderbook.
152+
// This is true for the initial response and false for all subsequent updates.
153+
// Note that if the snapshot is true, then all previous entries should be
154+
// discarded and the orderbook should be resynced.
155+
bool snapshot = 2;
156+
}

‎proto/dydxprotocol/indexer/off_chain_updates/off_chain_updates.proto

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ package dydxprotocol.indexer.off_chain_updates;
44
import "dydxprotocol/indexer/shared/removal_reason.proto";
55
import "dydxprotocol/indexer/protocol/v1/clob.proto";
66

7-
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/off_chain_updates";
7+
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/off_chain_updates/types";
88

99
// Do not make any breaking changes to these protos, a new version should be
1010
// created if a breaking change is needed.

‎proto/dydxprotocol/indexer/protocol/v1/clob.proto

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ package dydxprotocol.indexer.protocol.v1;
44
import "dydxprotocol/indexer/protocol/v1/subaccount.proto";
55
import "gogoproto/gogo.proto";
66

7-
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1";
7+
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1/types";
88

99
// Initial copy of protos from dYdX chain application state protos for the clob
1010
// module for use to send Indexer specific messages. Do not make any breaking

‎proto/dydxprotocol/indexer/protocol/v1/subaccount.proto

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ package dydxprotocol.indexer.protocol.v1;
44
import "cosmos_proto/cosmos.proto";
55
import "gogoproto/gogo.proto";
66

7-
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1";
7+
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/protocol/v1/types";
88

99
// Initial copy of protos from dYdX chain application state protos for the
1010
// subaccount module for use to send Indexer specific messages. Do not make any

‎proto/dydxprotocol/indexer/shared/removal_reason.proto

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
syntax = "proto3";
22
package dydxprotocol.indexer.shared;
33

4-
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/shared";
4+
option go_package = "github.com/dydxprotocol/v4-chain/protocol/indexer/shared/types";
55

66
// TODO(DEC-869): Update reasons/statuses for Advanced Orders.
77

‎protocol/app/ante_whitebox_test.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ func newTestHandlerOptions() HandlerOptions {
7171
nil,
7272
nil,
7373
nil,
74+
nil,
7475
flags.GetDefaultClobFlags(),
7576
rate_limit.NewNoOpRateLimiter[*types.MsgPlaceOrder](),
7677
rate_limit.NewNoOpRateLimiter[*types.MsgCancelOrder](),

‎protocol/app/app.go

Lines changed: 27 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,10 @@ import (
186186
"github.com/dydxprotocol/v4-chain/protocol/indexer"
187187
"github.com/dydxprotocol/v4-chain/protocol/indexer/indexer_manager"
188188
"github.com/dydxprotocol/v4-chain/protocol/indexer/msgsender"
189+
190+
// Grpc Streaming
191+
streaming "github.com/dydxprotocol/v4-chain/protocol/streaming/grpc"
192+
streamingtypes "github.com/dydxprotocol/v4-chain/protocol/streaming/grpc/types"
189193
)
190194

191195
var (
@@ -284,8 +288,9 @@ type App struct {
284288
// module configurator
285289
configurator module.Configurator
286290

287-
IndexerEventManager indexer_manager.IndexerEventManager
288-
Server *daemonserver.Server
291+
IndexerEventManager indexer_manager.IndexerEventManager
292+
GrpcStreamingManager streamingtypes.GrpcStreamingManager
293+
Server *daemonserver.Server
289294

290295
// startDaemons encapsulates the logic that starts all daemons and daemon services. This function contains a
291296
// closure of all relevant data structures that are shared with various keepers. Daemon services startup is
@@ -593,6 +598,9 @@ func New(
593598
tkeys[indexer_manager.TransientStoreKey],
594599
indexerFlags.SendOffchainData,
595600
)
601+
602+
app.GrpcStreamingManager = getGrpcStreamingManagerFromOptions(appFlags, logger)
603+
596604
timeProvider := &timelib.TimeProviderImpl{}
597605

598606
app.EpochsKeeper = *epochsmodulekeeper.NewKeeper(
@@ -877,7 +885,9 @@ func New(
877885
clobFlags := clobflags.GetClobFlagValuesFromOptions(appOpts)
878886
logger.Info("Parsed CLOB flags", "Flags", clobFlags)
879887

880-
memClob := clobmodulememclob.NewMemClobPriceTimePriority(app.IndexerEventManager.Enabled())
888+
memClob := clobmodulememclob.NewMemClobPriceTimePriority(
889+
app.IndexerEventManager.Enabled() || app.GrpcStreamingManager.Enabled(),
890+
)
881891

882892
app.ClobKeeper = clobmodulekeeper.NewKeeper(
883893
appCodec,
@@ -899,6 +909,7 @@ func New(
899909
app.StatsKeeper,
900910
app.RewardsKeeper,
901911
app.IndexerEventManager,
912+
app.GrpcStreamingManager,
902913
txConfig.TxDecoder(),
903914
clobFlags,
904915
rate_limit.NewPanicRateLimiter[*clobmoduletypes.MsgPlaceOrder](),
@@ -1592,3 +1603,16 @@ func getIndexerFromOptions(
15921603
}
15931604
return indexerMessageSender, indexerFlags
15941605
}
1606+
1607+
// getGrpcStreamingManagerFromOptions returns an instance of a streamingtypes.GrpcStreamingManager from the specified
1608+
// options. This function will default to returning a no-op instance.
1609+
func getGrpcStreamingManagerFromOptions(
1610+
appFlags flags.Flags,
1611+
logger log.Logger,
1612+
) (manager streamingtypes.GrpcStreamingManager) {
1613+
if appFlags.GrpcStreamingEnabled {
1614+
logger.Info("GRPC streaming is enabled")
1615+
return streaming.NewGrpcStreamingManager()
1616+
}
1617+
return streaming.NewNoopGrpcStreamingManager()
1618+
}

0 commit comments

Comments
 (0)