[CT-646] stream offchain updates through stream manager #1138
Conversation
WalkthroughThe overall change introduces enhancements to the gRPC streaming functionality within a blockchain protocol, focusing on order book updates. Modifications include the addition of a new parameter for better state management, improvements in the gRPC streaming manager's synchronization and subscription handling, and the introduction of a specific error for disabled streaming managers. These updates aim to streamline the process of sending order book updates to subscribers and ensure robust management of gRPC streaming connections. Changes
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configration File (
|
| return streaming.NewGrpcStreamingManager() | ||
| } | ||
| return streaming.NewNoopGrpcStreamingManager() | ||
| return streaming.NewGrpcStreamingManager() |
There was a problem hiding this comment.
will add the if statement back in in the next PR
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Status
Actionable comments generated: 2
Configuration used: CodeRabbit UI
Files selected for processing (5)
- protocol/app/app.go (2 hunks)
- protocol/streaming/grpc/grpc_streaming_manager.go (3 hunks)
- protocol/x/clob/keeper/grpc_stream_orderbook.go (1 hunks)
- protocol/x/clob/keeper/orders.go (1 hunks)
- protocol/x/clob/types/errors.go (1 hunks)
Additional comments: 8
protocol/x/clob/keeper/grpc_stream_orderbook.go (3)
- 11-13: The addition of a check to ensure the gRPC streaming manager is enabled before proceeding with subscriptions is a good practice for robust error handling. This prevents attempts to use the streaming functionality when it's not available, which could lead to runtime errors.
- 16-18: Proper error handling is implemented for the subscription process. If an error occurs during subscription, it is returned immediately, which is a good practice for early error detection and handling.
- 21-29: The loop for managing the stream's lifecycle based on the
finishedchannel and the stream's context is well-implemented. It ensures that the function remains active as long as the stream is open or until it's explicitly finished, which is crucial for maintaining a live data stream.protocol/streaming/grpc/grpc_streaming_manager.go (4)
- 16-26: The introduction of a
sync.Mutexfor synchronization and theOrderbookSubscriptionstruct is a solid approach to managing concurrent access to subscription data. This ensures thread safety when modifying theorderbookSubscriptionsmap and handling individual subscriptions.- 47-60: The
Subscribemethod correctly initializes a new subscription and adds it to theorderbookSubscriptionsmap with proper locking to ensure thread safety. The use ofdeferto unlock is a good practice for avoiding potential deadlocks.- 66-110: The
SendOrderbookUpdatesmethod demonstrates a comprehensive approach to grouping updates by clob pair ids, converting messages to a different version, and sending updates to subscribers. The cleanup of subscriptions that have been closed is also handled well. However, it's important to ensure that thefinishedchannel is properly closed to avoid potential leaks.Ensure that the
finishedchannel is closed appropriately to prevent leaks.
- 112-120: The
GetOffchainUpdatesV1method correctly unmarshals messages in offchain updates toOffChainUpdateV1. Using a loop to append each unmarshaled update to thev1updatesslice is a straightforward and effective approach.protocol/x/clob/types/errors.go (1)
- 522-525: The addition of the
ErrGrpcStreamingManagerNotEnablederror constant with a unique error code (11000) is a good practice for providing clear and specific error messages. This will help in debugging and handling specific scenarios where the gRPC Streaming Manager is not enabled.
| return streaming.NewGrpcStreamingManager() | ||
| } | ||
| return streaming.NewNoopGrpcStreamingManager() | ||
| return streaming.NewGrpcStreamingManager() |
There was a problem hiding this comment.
The gRPC server is initialized without specifying credentials, which could lead to insecure communication. It's important to ensure that all gRPC communication is encrypted to prevent eavesdropping and tampering by malicious actors. Consider adding SSL/TLS credentials to the gRPC server initialization. You can create credentials using credentials.NewServerTLSFromFile("cert.pem", "cert.key") and include them in the server options.
grpc.NewServer(grpc.Creds(credentials.NewServerTLSFromFile("cert.pem", "cert.key")))
There was a problem hiding this comment.
Unfortunately the ec2 instance I am running this on does not support filesystem reads and writes. Can I paste the contents of my .pem file in the code?
There was a problem hiding this comment.
@jonfung-dydx Embedding the contents of your .pem file directly in the code is not recommended due to security concerns. A more secure approach would be to use environment variables or a secrets management service provided by your cloud provider. This way, you can securely manage the .pem file contents without exposing them in your source code. For example, you could load the .pem file contents into an environment variable and then access it in your code like so:
cert := os.Getenv("CERT_PEM_CONTENTS")
key := os.Getenv("KEY_PEM_CONTENTS")
creds, err := credentials.NewServerTLSFromString(cert, key)
if err != nil {
// Handle error
}
grpcServer := grpc.NewServer(grpc.Creds(creds))Make sure to securely set the environment variables CERT_PEM_CONTENTS and KEY_PEM_CONTENTS with the contents of your .pem files. If you're using AWS, you might consider using AWS Secrets Manager or Parameter Store to securely manage and access these values.
There was a problem hiding this comment.
Unfortunately I have no money to use AWS secrets manager. What should I do here?
There was a problem hiding this comment.
@jonfung-dydx If AWS Secrets Manager is not an option due to cost concerns, you might consider the following approaches to securely manage your .pem file contents for the gRPC server:
-
Encrypted Environment Variables: You can encrypt the .pem file contents and store them as environment variables. During your application startup, you can decrypt these variables to use them. This approach requires you to manage encryption keys securely, but it avoids storing sensitive information in plain text.
-
Open-Source Secrets Management Tools: There are several open-source secrets management tools that you can run without significant cost. Tools like HashiCorp Vault or Bitnami Sealed Secrets can help you manage secrets securely. These tools provide mechanisms to store, access, and manage secrets, including certificates, without needing to store them directly in your code or as plain environment variables.
-
In-Memory Storage: Since your environment does not support filesystem reads/writes, consider loading the .pem contents into memory at the start of your application (e.g., from an encrypted source) and using them directly from there. This approach keeps the sensitive information out of the filesystem and environment variables, reducing the risk of accidental exposure.
Each of these methods has its own set of considerations, especially regarding the secure handling of encryption keys and secret values. It's important to evaluate them in the context of your application's security requirements and operational environment.
| @@ -1241,6 +1241,8 @@ func (k Keeper) SendOffchainMessages( | |||
| } | |||
| k.GetIndexerEventManager().SendOffchainData(update) | |||
| } | |||
|
|
|||
| k.GetGrpcStreamingManager().SendOrderbookUpdates(offchainUpdates) | |||
There was a problem hiding this comment.
The addition of SendOrderbookUpdates to the Keeper struct is a critical part of enhancing the system's capability to stream offchain updates. This function leverages the GrpcStreamingManager to send order book updates, aligning with the PR's objectives. However, it's crucial to ensure that offchainUpdates passed to this function are validated and sanitized before use to prevent any potential security issues. Additionally, consider error handling for scenarios where the gRPC streaming manager might not be available or fails to send updates, to ensure the system's robustness.
Ensure proper validation and error handling within SendOrderbookUpdates to enhance security and robustness.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Configuration used: CodeRabbit UI
Files selected for processing (1)
- protocol/streaming/grpc/grpc_streaming_manager.go (2 hunks)
Files skipped from review as they are similar to previous changes (1)
- protocol/streaming/grpc/grpc_streaming_manager.go
| defer sm.Unlock() | ||
|
|
||
| sm.orderbookSubscriptions[sm.nextId] = subscription | ||
| sm.nextId++ |
There was a problem hiding this comment.
for my understanding, what is the point of the nextId variable if we are just using this as a unique key variable for a map? seems like we don't expose it anywhere here, where are we gonna expose subscription key? functionally this is the same as an append only array
EDIT: i see, it is for removal
|
|
||
| // orderbookSubscriptions maps subscription IDs to their respective orderbook subscriptions. | ||
| orderbookSubscriptions map[uint32]*OrderbookSubscription | ||
| nextId uint32 |
There was a problem hiding this comment.
nit: nextSubscriptionId
| updates[clobPairId] = clobtypes.NewOffchainUpdates() | ||
| } | ||
| updates[clobPairId].Messages = append(updates[clobPairId].Messages, message) | ||
| } |
There was a problem hiding this comment.
Doesn't have to be now, but do we want to condense the messages somehow? We have this condensing function for indexer, but i don't think it works in this case. these events might be condensed differently.
func (om *OffchainUpdates) CondenseMessagesForReplay() {
There was a problem hiding this comment.
I think this gets called before SendOffchainMessages is called? i.e. the message is already condensed
There was a problem hiding this comment.
It gets called during Replay Operations but i feel it might not be used half the time due to this code:
if replayUpdates != nil {
offchainUpdates = replayUpdates
}
I think it should be fine, unless you can think of any edge case where condensing messages drops anything vital for orderbook updates
There was a problem hiding this comment.
yeah should be fine to send for example (place + remove) for example
|
|
||
| // Clean up subscriptions that have been closed. | ||
| for _, id := range idsToRemove { | ||
| sm.orderbookSubscriptions[id].finished <- true |
There was a problem hiding this comment.
close the channel if it's done and this is the only goroutine that sends stuff into this channel?
There was a problem hiding this comment.
actually I think we can just use ctx.Done()? no need to have this extra channel
| for clobPairId, update := range updates { | ||
| v1update, err := GetOffchainUpdatesV1(update) | ||
| if err != nil { | ||
| panic(err) |
There was a problem hiding this comment.
i believe golang kills the whole process on a panic. Would this lead to the whole validator going bye bye? Do we want to have a recover?
There was a problem hiding this comment.
yeah I was trying to see what makes sense here -
based on my understanding, MMs will run their own full nodes and these full nodes are only used for streaming information to their trading strategies. in this case, if something happens, probably not important to recover?
There was a problem hiding this comment.
let me know your thoughts
There was a problem hiding this comment.
hm that sounds good to me. but we should make this extra clear in the grpcstream enabled flag description and in the docs that this feature is intended to be used on full nodes?
Would it even make sense to make sense to add validation that grpcstream can only be enabled on full nodes? Might be overkill. I would be a little suspicious if a consensus-participating validator had this stream turned on of mev.
There was a problem hiding this comment.
yup - planning to add that in my next PR
| // Clean up subscriptions that have been closed. | ||
| for _, id := range idsToRemove { | ||
| sm.orderbookSubscriptions[id].finished <- true | ||
| delete(sm.orderbookSubscriptions, id) |
There was a problem hiding this comment.
Are we sure that this delete will properly clean up the nested inner data structures? grpc.ServerStream might have a close function?
There was a problem hiding this comment.
Seems like there's no close function 🆗
There was a problem hiding this comment.
yeah wasn't able to find a close function either
not really sure if we need to do ctx, cancelCtx := context.WithCancel(ctx) to cancel the context, since the connection is likely already broken and the context is already done 🤔
There was a problem hiding this comment.
https://github.com/grpc/grpc-go/blob/v1.60.x/stream.go#L262-L278
// Possible context leak:
// The cancel function for the child context we create will only be called
// when RecvMsg returns a non-nil error, if the ClientConn is closed, or if
// an error is generated by SendMsg.
I'm assuming an error will be generated by SendMsg in this case so the cancel function will be called.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Configuration used: CodeRabbit UI
Files selected for processing (5)
- protocol/streaming/grpc/grpc_streaming_manager.go (2 hunks)
- protocol/streaming/grpc/noop_streaming_manager.go (1 hunks)
- protocol/streaming/grpc/types/manager.go (1 hunks)
- protocol/x/clob/keeper/grpc_stream_orderbook.go (1 hunks)
- protocol/x/clob/types/errors.go (1 hunks)
Files skipped from review as they are similar to previous changes (3)
- protocol/streaming/grpc/grpc_streaming_manager.go
- protocol/x/clob/keeper/grpc_stream_orderbook.go
- protocol/x/clob/types/errors.go
Additional comments: 2
protocol/streaming/grpc/types/manager.go (1)
- 12-17: > 📝 NOTE
This review was outside the diff hunks and was mapped to the diff hunk with the greatest overlap. Original lines [1-14]
The interface
GrpcStreamingManageris well-defined with clear responsibilities for managing gRPC streaming of order book updates. It's important to ensure that all implementations of this interface correctly handle the absence of thefinishedchannel, as mentioned in the summary. This change simplifies the interface, potentially making it easier to implement and use.protocol/streaming/grpc/noop_streaming_manager.go (1)
- 26-26: The update to the
Subscribemethod in theNoopGrpcStreamingManagerto return an error indicating that gRPC streaming is not enabled is a good practice. It clearly communicates to the caller that the functionality is intentionally not available, which is crucial for error handling and debugging.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Configuration used: CodeRabbit UI
Files selected for processing (1)
- protocol/streaming/grpc/grpc_streaming_manager.go (3 hunks)
Files skipped from review as they are similar to previous changes (1)
- protocol/streaming/grpc/grpc_streaming_manager.go
…#1138) * [CT-646] stream offchain updates through stream manager * comments * fix lint * get rid of finished * comments * comments Signed-off-by: Eric <eric.warehime@gmail.com>
commit d98f859 Author: Eric <eric.warehime@gmail.com> Date: Mon Mar 11 12:46:53 2024 -0700 Update sample pregenesis Signed-off-by: Eric <eric.warehime@gmail.com> commit 7f178fe Author: Mohammed Affan <affanmd@nyu.edu> Date: Mon Mar 11 13:46:08 2024 -0400 [OTE-209] Emit metrics gated through execution mode (dydxprotocol#1157) Signed-off-by: Eric <eric.warehime@gmail.com> commit 47e365d Author: dydxwill <119354122+dydxwill@users.noreply.github.com> Date: Mon Mar 11 13:43:16 2024 -0400 add aggregate comlink response code stats (dydxprotocol#1162) Signed-off-by: Eric <eric.warehime@gmail.com> commit 7774ad9 Author: shrenujb <98204323+shrenujb@users.noreply.github.com> Date: Fri Mar 8 17:30:49 2024 -0500 [TRA-70] Add state migrations for isolated markets (dydxprotocol#1155) Signed-off-by: Shrenuj Bansal <shrenuj@dydx.exchange> Signed-off-by: Eric <eric.warehime@gmail.com> commit 89c7405 Author: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com> Date: Thu Mar 7 17:28:06 2024 -0500 [CT-517] E2E tests batch cancel (dydxprotocol#1149) * more end to end test * extraprint * more e2e test Signed-off-by: Eric <eric.warehime@gmail.com> commit 41a3a41 Author: Teddy Ding <teddy@dydx.exchange> Date: Thu Mar 7 15:42:30 2024 -0500 [OTE-200] OIMF protos (dydxprotocol#1125) * OIMF protos * add default genesis value, modify methods interface * fix genesis file * fix integration test * lint Signed-off-by: Eric <eric.warehime@gmail.com> commit 2a062b1 Author: Teddy Ding <teddy@dydx.exchange> Date: Thu Mar 7 15:24:15 2024 -0500 Add `v5` upgrade handler and set up container upgrade test (dydxprotocol#1153) * wip * update preupgrade_genesis * fix preupgrade_genesis.json * nit * setupUpgradeStoreLoaders for v5.0.0 Signed-off-by: Eric <eric.warehime@gmail.com> commit b7942b3 Author: jayy04 <103467857+jayy04@users.noreply.github.com> Date: Thu Mar 7 13:43:48 2024 -0500 [CT-647] construct the initial orderbook snapshot (dydxprotocol#1147) * [CT-647] construct the initial orderbook snapshot * [CT-647] initialize new streams and send orderbook snapshot (dydxprotocol#1152) * [CT-647] initialize new streams and send orderbook snapshot * use sync once * comments Signed-off-by: Eric <eric.warehime@gmail.com> commit c67a3c6 Author: shrenujb <98204323+shrenujb@users.noreply.github.com> Date: Thu Mar 7 12:40:37 2024 -0500 [TRA-84] Move SA module address transfers to use perpetual based SA accounts (dydxprotocol#1146) Signed-off-by: Shrenuj Bansal <shrenuj@dydx.exchange Signed-off-by: Eric <eric.warehime@gmail.com> commit dba23e0 Author: Mohammed Affan <affanmd@nyu.edu> Date: Thu Mar 7 10:34:11 2024 -0500 update readme link to point to the right page (dydxprotocol#1151) Signed-off-by: Eric <eric.warehime@gmail.com> commit b5870d5 Author: Tian <tian@dydx.exchange> Date: Wed Mar 6 16:43:04 2024 -0500 [TRA-86] scaffold x/vault (dydxprotocol#1148) * scaffold x/vault Signed-off-by: Eric <eric.warehime@gmail.com> commit 0eca041 Author: jayy04 <103467857+jayy04@users.noreply.github.com> Date: Wed Mar 6 10:48:42 2024 -0500 [CT-652] add command line flag for full node streaming (dydxprotocol#1145) Signed-off-by: Eric <eric.warehime@gmail.com> commit b319cb8 Author: jayy04 <103467857+jayy04@users.noreply.github.com> Date: Tue Mar 5 21:58:35 2024 -0500 [CT-646] stream offchain updates through stream manager (dydxprotocol#1138) * [CT-646] stream offchain updates through stream manager * comments * fix lint * get rid of finished * comments * comments Signed-off-by: Eric <eric.warehime@gmail.com> commit 1c54620 Author: shrenujb <98204323+shrenujb@users.noreply.github.com> Date: Tue Mar 5 16:34:19 2024 -0500 [TRA-78] Add function to retrieve collateral pool addr for a subaccount (dydxprotocol#1142) Signed-off-by: Shrenuj Bansal <shrenuj@dydx.exchange> Signed-off-by: Eric <eric.warehime@gmail.com> commit b8c1d62 Author: dydxwill <119354122+dydxwill@users.noreply.github.com> Date: Tue Mar 5 15:03:28 2024 -0500 [OTE-141] implement post /compliance/geoblock (dydxprotocol#1129) Signed-off-by: Eric <eric.warehime@gmail.com> commit ab8c570 Author: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com> Date: Tue Mar 5 11:19:53 2024 -0500 Fix mock-gen dydxprotocol#1140 Signed-off-by: Eric <eric.warehime@gmail.com> commit 12506a1 Author: shrenujb <98204323+shrenujb@users.noreply.github.com> Date: Mon Mar 4 21:33:28 2024 -0500 [TRA-64] Use market specific insurance fund for cross or isolated markets (dydxprotocol#1132) Signed-off-by: Shrenuj Bansal <shrenuj@dydx.exchange> Signed-off-by: Eric <eric.warehime@gmail.com> commit 929f09e Author: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com> Date: Mon Mar 4 13:48:04 2024 -0800 [CT-514] Clob `MsgBatchCancel` functionality (dydxprotocol#1110) * wip implementation * use new cometbft * Revert "use new cometbft" This reverts commit e5b8a03. * go mod tidy * basic e2e test * more msgBatchCancels in code * repeated fixed32 -> uint32 * remove debug prints * update cometbft replace go.mod sha * one more debug print * typo * regen indexer protos * update comment on proto * proto comment changes * extract stateful validation into own fn * pr format comments * clean up test file * new return type with success and failure Signed-off-by: Eric <eric.warehime@gmail.com> commit 41de83e Author: dydxwill <119354122+dydxwill@users.noreply.github.com> Date: Mon Mar 4 12:22:16 2024 -0500 add index to address read replica lag (dydxprotocol#1137) Signed-off-by: Eric <eric.warehime@gmail.com> commit 735d9a8 Author: dydxwill <119354122+dydxwill@users.noreply.github.com> Date: Mon Mar 4 11:56:59 2024 -0500 rename (dydxprotocol#1136) Signed-off-by: Eric <eric.warehime@gmail.com> commit 86617dd Author: jayy04 <103467857+jayy04@users.noreply.github.com> Date: Mon Mar 4 10:43:31 2024 -0500 [CT-644] instantiate grpc stream manager (dydxprotocol#1134) * [CT-644] instantiate grpc stream manager * update type * update channel type Signed-off-by: Eric <eric.warehime@gmail.com> commit 32afd64 Author: Eric <eric.warehime@gmail.com> Date: Mon Mar 11 12:41:06 2024 -0700 Update go version in Dockerfile Signed-off-by: Eric <eric.warehime@gmail.com> commit ba27204 Author: Eric <eric.warehime@gmail.com> Date: Fri Mar 8 09:44:04 2024 -0800 Add slinky utils, use that to convert between market and currency pair commit 667a804 Author: Eric <eric.warehime@gmail.com> Date: Wed Mar 6 20:43:40 2024 -0800 Update error messages commit d53292c Author: Eric <eric.warehime@gmail.com> Date: Wed Mar 6 20:16:01 2024 -0800 Update docstrings, rename OracleClient commit daad125 Author: Eric <eric.warehime@gmail.com> Date: Mon Mar 4 10:51:23 2024 -0800 VoteExtension slinky logic
* [CT-646] stream offchain updates through stream manager * comments * fix lint * get rid of finished * comments * comments
* [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
* [CT-646] stream offchain updates through stream manager * comments * fix lint * get rid of finished * comments * comments
* [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 * [CT-700] separate indexer and grpc streaming events (#1209) * [CT-700] separate indexer and grpc streaming events * fix tests * comments * update * [CT-700] only send response when there is at least one update (#1216) * [CT-712] send order update when short term order state fill amounts are pruned (#1241) * [CT-712] send fill amount updates for reverted operations (#1240) * [CT-723] add block number + stage to grpc updates (#1252) * [CT-723] add block number + stage to grpc updates * add indexer changes * [CT-727] avoid state reads when sending updates (#1261)
* [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 * [CT-700] separate indexer and grpc streaming events (#1209) * [CT-700] separate indexer and grpc streaming events * fix tests * comments * update * [CT-700] only send response when there is at least one update (#1216) * [CT-712] send order update when short term order state fill amounts are pruned (#1241) * [CT-712] send fill amount updates for reverted operations (#1240) * [CT-723] add block number + stage to grpc updates (#1252) * [CT-723] add block number + stage to grpc updates * add indexer changes * [CT-727] avoid state reads when sending updates (#1261)
* [OTE-221] Add query for PendingSendPacket (backport #1176) (#1221) --------- Co-authored-by: Teddy Ding <teddy@dydx.exchange> (cherry picked from commit e545bbf) # Conflicts: # indexer/packages/v4-protos/src/codegen/dydxprotocol/bundle.ts # indexer/packages/v4-protos/src/codegen/gogoproto/bundle.ts # indexer/packages/v4-protos/src/codegen/google/bundle.ts # protocol/go.mod * fix protos * update go.mod --------- Co-authored-by: Mohammed Affan <affanmd@nyu.edu> Co-authored-by: affan <affan@dydx.exchange> * [Backport v4.x] backport full node streaming to v4.x branch (#1270) * [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 * [CT-700] separate indexer and grpc streaming events (#1209) * [CT-700] separate indexer and grpc streaming events * fix tests * comments * update * [CT-700] only send response when there is at least one update (#1216) * [CT-712] send order update when short term order state fill amounts are pruned (#1241) * [CT-712] send fill amount updates for reverted operations (#1240) * [CT-723] add block number + stage to grpc updates (#1252) * [CT-723] add block number + stage to grpc updates * add indexer changes * [CT-727] avoid state reads when sending updates (#1261) * [CT-712] send updates for both normal order matches and liquidation (#1280) (#1281) * Fix lib.ErrorLogWithError: properly pass in args (#1306) (#1310) (cherry picked from commit a91c1ca) Co-authored-by: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com> * fix broken tests (#1312) (#1316) (cherry picked from commit 5ec37d2) Co-authored-by: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com> --------- Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com> Co-authored-by: Mohammed Affan <affanmd@nyu.edu> Co-authored-by: affan <affan@dydx.exchange> Co-authored-by: jayy04 <103467857+jayy04@users.noreply.github.com> Co-authored-by: Jonathan Fung <121899091+jonfung-dydx@users.noreply.github.com>
Changelist
[Describe or list the changes made in this PR]
Test Plan
[Describe how this PR was tested (if applicable)]
Author/Reviewer Checklist
state-breakinglabel.indexer-postgres-breakinglabel.PrepareProposalorProcessProposal, manually add the labelproposal-breaking.feature:[feature-name].backport/[branch-name].refactor,chore,bug.