[IND-288] send final message per order id for order fills, with aggregated fill sfrom the entire block #658
Conversation
WalkthroughThe changes primarily focus on improving the handling, sorting, and aggregation of subaccount messages and fill events in the Kafka publisher. A new interface Changes
TipsChat with CodeRabbit Bot (
|
![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: 0
Files selected for processing (6)
- indexer/services/ender/tests/lib/kafka-publisher.test.ts (3 hunks)
- indexer/services/ender/src/handlers/handler.ts (3 hunks)
- indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts (1 hunks)
- indexer/services/ender/src/lib/candles-generator.ts (2 hunks)
- indexer/services/ender/src/lib/kafka-publisher.ts (4 hunks)
- indexer/services/ender/src/lib/types.ts (1 hunks)
Files skipped from review due to trivial changes (2)
- indexer/services/ender/tests/lib/kafka-publisher.test.ts
- indexer/services/ender/src/lib/candles-generator.ts
Additional comments (Suppressed): 12
indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts (1)
- 407-412: The
generateConsolidatedSubaccountKafkaEventmethod now accepts two additional parameters:orderIdandisFill. Ensure that all calls to this method throughout the codebase have been updated to match the new signature. Also, verify that theorderobject is guaranteed to be non-null when accessingorder.id, as this could potentially cause a runtime error iforderis null or undefined.indexer/services/ender/src/lib/types.ts (2)
191-194: The new
AnnotatedSubaccountMessageinterface extendsSubaccountMessagewith two optional properties:orderIdandisFill. This change seems to be in line with the PR's goal of consolidating multiple fill events for a single order into a single websocket message. However, ensure that the rest of the codebase that usesSubaccountMessagecan handle the new properties without any issues.201-206: The
ConsolidatedKafkaEventtype now usesAnnotatedSubaccountMessageinstead ofSubaccountMessage. This change is consistent with the introduction of theAnnotatedSubaccountMessageinterface. However, make sure that all parts of the codebase that useConsolidatedKafkaEventare updated to handleAnnotatedSubaccountMessage.indexer/services/ender/src/handlers/handler.ts (3)
25-26: The import statement now includes
AnnotatedSubaccountMessagewhich extends the previousSubaccountMessagewith two additional properties:orderIdandisFill. Ensure that this new type is correctly defined and used throughout the codebase.105-108: The
generateConsolidatedSubaccountKafkaEventmethod now accepts two additional optional parameters:orderIdandisFill. Make sure that all calls to this method have been updated accordingly.117-124: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [111-122]
The
subaccountMessageis now of typeAnnotatedSubaccountMessageand includes the neworderIdandisFillproperties. Ensure that these properties are correctly populated and used in the rest of the code.indexer/services/ender/src/lib/kafka-publisher.ts (6)
28-31: The type of
subaccountMessageshas been changed fromSubaccountMessage[]toAnnotatedSubaccountMessage[]. Ensure that all usages ofsubaccountMessagesthroughout the codebase have been updated to handleAnnotatedSubaccountMessageobjects.49-50: The
addEventmethod has been updated to use thegetMessageshelper function to add an event to the appropriate message array based on its topic. This is a more flexible approach than the previous switch-case structure.81-104: The
retainLastFillEventsForSubaccountMessagesmethod has been added to retain only the last fill event per order ID insubaccountMessages. This is a significant change that affects how fill events are handled and sent to the websocket. Ensure that this change does not negatively impact any systems that rely on receiving all fill events for an order.109-117: The
compareMessageshelper function has been added to compare twoAnnotatedSubaccountMessageobjects based on block height, transaction index, and event index. This function is used inretainLastFillEventsForSubaccountMessagesto determine which fill event is the latest for a given order ID.46-139: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [123-149]
The
sortEventsmethod has been added to sort events based on block height, transaction index, and event index. This method replaces the previoussortTradeEventsmethod and is more flexible as it can sort both subaccount and trade events. However, it throws an error if the topic is not subaccounts or trades. Ensure that this method is only called with the appropriate topics.
- 175-176: The
retainLastFillEventsForSubaccountMessagesmethod is now called ingenerateAllTopicKafkaMessagesbefore generating Kafka messages forsubaccountMessages. This ensures that only the last fill event per order ID is included in the Kafka messages.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 2
Files selected for processing (4)
- indexer/services/ender/tests/helpers/kafka-publisher-helpers.ts (2 hunks)
- indexer/services/ender/tests/lib/kafka-publisher.test.ts (2 hunks)
- indexer/services/ender/src/lib/kafka-publisher.ts (4 hunks)
- indexer/services/ender/src/lib/types.ts (2 hunks)
Additional comments (Suppressed): 13
indexer/services/ender/__tests__/helpers/kafka-publisher-helpers.ts (2)
5-5: The import of
AnnotatedSubaccountMessageis new. Ensure that this type is correctly defined and used in the code.44-51: The new function
createConsolidatedKafkaEventFromSubaccountis introduced to create aConsolidatedKafkaEventfrom aAnnotatedSubaccountMessage. This function seems to be correctly implemented, but ensure that it is used properly in the rest of the codebase.indexer/services/ender/src/lib/types.ts (4)
31-37: The import of lodash (
_) is new in this hunk. Ensure that it is used in the codebase and that it doesn't introduce unnecessary overhead if its functionalities are not fully utilized. If only specific methods from lodash are used, consider importing only those methods to reduce the bundle size.192-195: The
AnnotatedSubaccountMessageinterface extendsSubaccountMessagewith two optional properties:orderIdandisFill. This is a good practice as it maintains the original structure ofSubaccountMessagewhile adding new properties, thus ensuring backward compatibility.197-202: The
convertToSubaccountMessagefunction is used to convert anAnnotatedSubaccountMessageback to aSubaccountMessageby omitting theorderIdandisFillproperties. This function uses lodash'somitmethod, which is a good choice for this operation. However, be aware that lodash'somitfunction can be slow for large objects because it creates a new object. If performance becomes an issue, consider using a more performant way to omit properties.209-214: The
ConsolidatedKafkaEventtype now usesAnnotatedSubaccountMessageinstead ofSubaccountMessage. This change should be compatible with the existing codebase as long as the new properties (orderIdandisFill) are optional and are handled correctly in the code that usesConsolidatedKafkaEvent.indexer/services/ender/__tests__/lib/kafka-publisher.test.ts (3)
1-30: The new hunk introduces additional imports from the
@dydxprotocol-indexer/kafkaand@dydxprotocol-indexer/postgrespackages, as well as from the localtypesandhelpersmodules. TheAnnotatedSubaccountMessageandconvertToSubaccountMessagetypes are now being imported, which suggests that theKafkaPublisherclass may now be handlingAnnotatedSubaccountMessageobjects. ThecreateConsolidatedKafkaEventFromSubaccounthelper function is also being imported, which suggests that theKafkaPublisherclass may now be creatingConsolidatedKafkaEventobjects fromAnnotatedSubaccountMessageobjects.149-218: The new hunk introduces a new test case for the
sortSubaccountEventsmethod of theKafkaPublisherclass. This method appears to sortAnnotatedSubaccountMessageobjects based on theblockHeight,transactionIndex, andeventIndexproperties. The test case uses theit.eachfunction to test the sorting ofAnnotatedSubaccountMessageobjects based on each of these properties. TheAnnotatedSubaccountMessageobjects are created with different values for the property being tested, and thesortSubaccountEventsmethod is called with these objects. The test case then asserts that theAnnotatedSubaccountMessageobjects are sorted in the expected order.221-308: The new hunk introduces a new test case for the
retainLastFillEventsForSubaccountMessagesmethod of theKafkaPublisherclass. This method appears to retain only the last fill event per order ID in thesubaccountMessagesarray. The test case creates severalAnnotatedSubaccountMessageobjects with differentblockHeight,transactionIndex,eventIndex,contents,subaccountId,version,orderId, andisFillproperties. Some of these objects represent fill events, while others do not. TheretainLastFillEventsForSubaccountMessagesmethod is called with these objects, and the test case then asserts that only the last fill event per order ID and the non-fill events are retained in thesubaccountMessagesarray. The test case also asserts that thepublishmethod calls theproducer.sendfunction with the expected arguments.indexer/services/ender/src/lib/kafka-publisher.ts (4)
30-36: The
KafkaPublisherclass now usesAnnotatedSubaccountMessageinstead ofSubaccountMessage. Ensure that all instances ofKafkaPublisherhave been updated to useAnnotatedSubaccountMessageand that the additional propertiesorderIdandisFillare being correctly set and used where necessary.54-55: The
addEventmethod now uses a helper functiongetMessagesto retrieve the appropriate message array based on the topic. This is a good refactor as it reduces code duplication and improves readability.51-143: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [130-156]
The
sortEventsmethod sorts events based on block height, transaction index, and event index. This is a good approach to ensure that events are processed in the correct order. However, the method throws an error if the topic is notKafkaTopics.TO_WEBSOCKETS_SUBACCOUNTSorKafkaTopics.TO_WEBSOCKETS_TRADES. Ensure that this method is only called with these topics, or consider handling other topics as well.
- 182-185: The
generateAllTopicKafkaMessagesmethod now callsretainLastFillEventsForSubaccountMessagesbefore generating all topic Kafka messages. This is a good approach to ensure that only the most recent fill event for each order ID is included in the Kafka messages. However, ensure that this does not cause any issues with the order or timing of the Kafka messages.
| private getMessages(kafkaTopic: KafkaTopics): any[] | undefined { | ||
| switch (kafkaTopic) { | ||
| case KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS: | ||
| this.subaccountMessages.push(event.message); | ||
| break; | ||
| return this.subaccountMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_TRADES: | ||
| this.tradeMessages.push(event.message); | ||
| break; | ||
| return this.tradeMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_MARKETS: | ||
| this.marketMessages.push(event.message); | ||
| break; | ||
| return this.marketMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_CANDLES: | ||
| this.candleMessages.push(event.message); | ||
| break; | ||
| return this.candleMessages; | ||
| case KafkaTopics.TO_VULCAN: | ||
| this.vulcanMessages.push(event.message); | ||
| break; | ||
| return this.vulcanMessages; | ||
| default: | ||
| throw new Error('Invalid Topic'); | ||
| } |
There was a problem hiding this comment.
The getMessages method returns an array of messages for a given Kafka topic. It uses a switch statement to determine which message array to return based on the topic. This is a good use of a switch statement as it makes the code more readable and maintainable. However, the method returns any[] | undefined, which is not type-safe. Consider using a more specific type or a union of specific types instead of any.
- private getMessages(kafkaTopic: KafkaTopics): any[] | undefined {
+ private getMessages(kafkaTopic: KafkaTopics): (AnnotatedSubaccountMessage | SingleTradeMessage | MarketMessage | CandleMessage)[] | undefined {| public retainLastFillEventsForSubaccountMessages() { | ||
| // Create a map to store the last fill event per order ID | ||
| const lastFillEvents: Record<string, AnnotatedSubaccountMessage> = {}; | ||
| const nonFillEvents: AnnotatedSubaccountMessage[] = []; | ||
|
|
||
| this.subaccountMessages.forEach((message) => { | ||
| if (message.isFill && message.orderId) { | ||
| const orderId = message.orderId; | ||
| // If we haven't seen this order ID before or if the current message | ||
| // has a higher block height, update the lastFillEvents for this order ID | ||
| if ( | ||
| !lastFillEvents[orderId] || | ||
| this.compareMessages(message, lastFillEvents[orderId]) > 0 | ||
| ) { | ||
| lastFillEvents[orderId] = message; | ||
| } | ||
| } else { | ||
| nonFillEvents.push(message); | ||
| } | ||
| }); | ||
|
|
||
| this.subaccountMessages = Object.values(lastFillEvents) | ||
| .concat(nonFillEvents) | ||
| .map((annotatedMessage) => convertToSubaccountMessage(annotatedMessage)); | ||
| this.sortEvents(KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS); | ||
| } |
There was a problem hiding this comment.
The retainLastFillEventsForSubaccountMessages method sorts and retains only the last fill event per order ID in the subaccountMessages array. This is a good approach to ensure that only the most recent fill event for each order ID is kept. However, the method mutates the subaccountMessages array directly, which can lead to unexpected side effects. Consider creating a new array instead of mutating the existing one.
- this.subaccountMessages = Object.values(lastFillEvents)
+ const updatedSubaccountMessages = Object.values(lastFillEvents)| * and event index in ascending order per order id. Only keep the subaccount message if | ||
| * it represents the last fill event per order id. | ||
| */ | ||
| public retainLastFillEventsForSubaccountMessages() { |
There was a problem hiding this comment.
This would lead to all prior fills for an order (if there were multiple) to also be omitted, and that would be incorrect.
E.g. Imagine order A has 2 fills:
- message 1: {fills: [ .... 40% of order size], order: [ .... status is OPEN, total filled is 40%]}
- message 2: {fills: [ .... 60% of order size], order: [..... status is FILLED, total filled is 100%]}
The websocket message I expect to see as a client is:
- message: {fills: [.... 40% of order size, .... 60% of order size], order: [....status is FILLED, total filled is 100%]}
However with this logic what the client sees is:
- message: {fills: [....60% of order size], order: [....status is FILLED, total filled is 100%]}
| if (message.isFill && message.orderId) { | ||
| const orderId = message.orderId; | ||
| // If we haven't seen this order ID before or if the current message | ||
| // has a higher block height, update the lastFillEvents for this order ID |
There was a problem hiding this comment.
| // has a higher block height, update the lastFillEvents for this order ID | |
| // was associated with a later event, update the lastFillEvents for this order ID |
| public sortTradeEvents() { | ||
| this.tradeMessages = this.tradeMessages.sort( | ||
| (a: SingleTradeMessage, b: SingleTradeMessage) => { | ||
| public sortEvents(kafkaTopic: KafkaTopics) { |
There was a problem hiding this comment.
nit: can this be made a generic function that takes in OrderedMessage and sorts them and returns a sorted list rather than checking if the topic is sortable?
…wl/send_final_fill_data
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Files selected for processing (3)
- indexer/services/ender/tests/lib/kafka-publisher.test.ts (2 hunks)
- indexer/services/ender/src/lib/candles-generator.ts (2 hunks)
- indexer/services/ender/src/lib/kafka-publisher.ts (4 hunks)
Files skipped from review due to trivial changes (1)
- indexer/services/ender/src/lib/candles-generator.ts
Additional comments (Suppressed): 10
indexer/services/ender/__tests__/lib/kafka-publisher.test.ts (3)
1-50: The new hunk introduces additional imports that are necessary for the updated tests. The
AnnotatedSubaccountMessagetype is imported, which extends theSubaccountMessagetype with two additional properties:orderIdandisFill. This new type is used in theKafkaPublisherclass, where thesubaccountMessagesproperty now holdsAnnotatedSubaccountMessageobjects. TheconvertToSubaccountMessagefunction is also imported, which is used to convertAnnotatedSubaccountMessageobjects toSubaccountMessageobjects. ThedefaultWalletAddressconstant is imported, which is used in the tests to simulate a wallet address. ThecreateConsolidatedKafkaEventFromSubaccountfunction is imported, which is used to createConsolidatedKafkaEventobjects fromAnnotatedSubaccountMessageobjects. ThegenerateFillSubaccountMessage,generateOrderSubaccountMessage, andgenerateTransferContentsfunctions are imported, which are used to generate different types of subaccount messages.169-238: The new hunk introduces a new test suite for the
sortSubaccountEventsmethod of theKafkaPublisherclass. This method is used to sort subaccount events based on different fields such asblockHeight,transactionIndex, andeventIndex. The test suite includes test cases that create aKafkaPublisherinstance, add subaccount events with different values for the specified field, call thesortSubaccountEventsmethod, and assert that the subaccount events are sorted in the expected order.241-410: The new hunk introduces a new test suite for the
aggregateFillEventsForSubaccountMessagesmethod of theKafkaPublisherclass. This method is used to aggregate all fill events per order ID and sort the messages. The test suite includes a test case that creates aKafkaPublisherinstance, adds several subaccount messages with different fill events and order IDs, calls theaggregateFillEventsForSubaccountMessagesmethod, and asserts that the subaccount messages are aggregated and sorted as expected. The test case also calls thepublishmethod and asserts that theproducer.sendfunction is called with the expected arguments.indexer/services/ender/src/lib/kafka-publisher.ts (7)
34-40: The
subaccountMessagesproperty of theKafkaPublisherclass has been changed fromSubaccountMessage[]toAnnotatedSubaccountMessage[]. This change is in line with the introduction of theAnnotatedSubaccountMessagetype, which extends theSubaccountMessagetype with two additional properties:orderIdandisFill. This change allows for more detailed tracking of subaccount messages, particularly those related to order fills.57-83: The
addEventmethod has been simplified by using a helper functiongetMessagesto retrieve the appropriate message array based on the topic. This change improves the readability and maintainability of the code by reducing the complexity of theaddEventmethod and encapsulating the logic for retrieving the message array in a separate function.91-110: Two new helper methods
extractFillsandreplaceFillshave been added to handle the extraction and replacement of fill events in the subaccount message contents. These methods improve the modularity of the code by encapsulating specific tasks in separate functions. They also improve the readability of the code by providing descriptive names for these tasks.121-165: A new method
aggregateFillEventsForSubaccountMessageshas been added to sort and retain only the last fill event per order ID in thesubaccountMessagesarray. This method addresses the issue of multiple fill events for an Immediate-Or-Cancel (IOC) order causing an invalid state transition fromCANCELEDtoFILLEDin the front-end. The method improves the accuracy of the state transitions by ensuring that only the last fill event per order ID is retained.170-178: A new helper method
compareMessageshas been added to compare twoAnnotatedSubaccountMessageobjects based on block height, transaction index, and event index. This method improves the modularity of the code by encapsulating the comparison logic in a separate function. It also improves the readability of the code by providing a descriptive name for this task.55-189: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [184-202]
The
sortTradeEventsmethod has been replaced with a more generalsortEventsmethod that can sort any array ofOrderedMessageobjects. This change improves the flexibility of the code by allowing any array ofOrderedMessageobjects to be sorted, not just trade events.
- 225-231: The
aggregateFillEventsForSubaccountMessagesmethod is now called in thegenerateAllTopicKafkaMessagesmethod if there are any subaccount messages. This change ensures that the fill events for subaccount messages are aggregated before the Kafka messages are generated.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 1
Files selected for processing (1)
- indexer/services/ender/src/lib/kafka-publisher.ts (4 hunks)
Additional comments (Suppressed): 6
indexer/services/ender/src/lib/kafka-publisher.ts (6)
34-40: The
subaccountMessagesproperty of theKafkaPublisherclass has been updated to holdAnnotatedSubaccountMessageobjects instead ofSubaccountMessageobjects. This change is in line with the introduction of theAnnotatedSubaccountMessagetype, which extendsSubaccountMessageand addsorderIdandisFillproperties. This change will allow theKafkaPublisherclass to handle messages that are annotated with additional information about the order and fill status.91-110: The
extractFillsandreplaceFillsmethods have been added to extract fill events fromSubaccountMessageContentsand replace the fills inSubaccountMessageContents, respectively. These methods are used in theaggregateFillEventsForSubaccountMessagesmethod to handle multiple fills in a single block for an Immediate-Or-Cancel (IOC) order.121-165: The
aggregateFillEventsForSubaccountMessagesmethod has been added to sort and retain only the last fill event per order ID in thesubaccountMessagesarray. This method is called in thegenerateAllTopicKafkaMessagesmethod before generating all topic Kafka messages. This change addresses the issue of multiple fills in a single block for an IOC order, which can lead to an invalid state transition fromCANCELEDtoFILLED.170-178: The
compareMessagesmethod has been added to compare twoAnnotatedSubaccountMessageobjects based on block height, transaction index, and event index. This method is used in theaggregateFillEventsForSubaccountMessagesmethod to determine the order of fill events.55-189: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [184-202]
The
sortEventsmethod has been added to sort events based on block height, transaction index, and event index. This method is called in theaggregateFillEventsForSubaccountMessagesmethod after aggregating fill events. This change ensures that the events are sorted in the correct order before being published.
- 225-231: The
publishmethod now calls theaggregateFillEventsForSubaccountMessagesmethod before generating all topic Kafka messages. This change ensures that only the last fill event per order ID is retained in thesubaccountMessagesarray before the messages are published.
| public addEvent(event: ConsolidatedKafkaEvent) { | ||
| switch (event.topic) { | ||
| this.getMessages(event.topic)!.push(event.message); | ||
| } | ||
|
|
||
| /** | ||
| * Helper function to get messages for a given topic. | ||
| * | ||
| * @param kafkaTopic | ||
| * @private | ||
| */ | ||
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | ||
| private getMessages(kafkaTopic: KafkaTopics): any[] | undefined { | ||
| switch (kafkaTopic) { | ||
| case KafkaTopics.TO_WEBSOCKETS_SUBACCOUNTS: | ||
| this.subaccountMessages.push(event.message); | ||
| break; | ||
| return this.subaccountMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_TRADES: | ||
| this.tradeMessages.push(event.message); | ||
| break; | ||
| return this.tradeMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_MARKETS: | ||
| this.marketMessages.push(event.message); | ||
| break; | ||
| return this.marketMessages; | ||
| case KafkaTopics.TO_WEBSOCKETS_CANDLES: | ||
| this.candleMessages.push(event.message); | ||
| break; | ||
| return this.candleMessages; | ||
| case KafkaTopics.TO_VULCAN: | ||
| this.vulcanMessages.push(event.message); | ||
| break; | ||
| return this.vulcanMessages; | ||
| default: | ||
| throw new Error('Invalid Topic'); | ||
| } | ||
| } |
There was a problem hiding this comment.
The addEvent method has been updated to use a helper function getMessages to retrieve the appropriate message array based on the topic. This change simplifies the addEvent method and makes it easier to add new topics in the future. However, the getMessages method returns any[] | undefined, which could lead to potential type safety issues. Consider refining the return type to be more specific.
- private getMessages(kafkaTopic: KafkaTopics): any[] | undefined {
+ private getMessages(kafkaTopic: KafkaTopics): (AnnotatedSubaccountMessage | SingleTradeMessage)[] | undefined {
dydxwill
changed the title
| * Due to separate handlers for order fills, we can be sure that if a message is annotated | ||
| * with a fill, it should only contain data about a fill / order and not transfers. |
There was a problem hiding this comment.
nit: Change this to note that the message should only contain a single fill / order.
| private extractFills(contents: string): FillSubaccountMessageContents[] | undefined { | ||
| try { | ||
| const subaccountContents: | ||
| SubaccountMessageContents = JSON.parse(contents) as SubaccountMessageContents; |
There was a problem hiding this comment.
nit: Let's add a TODO + linear ticket for adding a better way to aggregate messages. In addition to orders, we also want to make it so perpetual positions / asset positions only have a single update per block, so a more generic solution would be better. I think the current fix is fine for the short-term.
Having to do an extra JSON.parse per order message is un-ideal, and the JSON.stringify in the order handler is now redundant. It'd be better to add more generic logic to aggregate different types of subaccount messages rather than do post-processing.
There was a problem hiding this comment.
Actually, what do you think about updating the annotated messages such that it also contains a contents property to avoid re-parsing all the message contents? I think that's a small enough change that can be scoped into this fix, rather than waiting for the ticket to allow more generic aggregation of subaccount events.
| } | ||
| }); | ||
|
|
||
| // Replace the fills in the last event for fills with all the fills for that order ID |
There was a problem hiding this comment.
| // Replace the fills in the last event for fills with all the fills for that order ID | |
| // Update the last event for the order ID such that it has all the fills that occurred for the order ID |
| * where the first event should be the earliest event in the block. | ||
| */ | ||
| public sortEvents(msgs: OrderedMessage[]) { | ||
| if (msgs) { |
There was a problem hiding this comment.
Is this needed? Would calling .sort on an empty array lead to an error / undesirable outcome?
| * @private | ||
| */ | ||
| // eslint-disable-next-line @typescript-eslint/no-explicit-any | ||
| private getMessages(kafkaTopic: KafkaTopics): any[] | undefined { |
There was a problem hiding this comment.
nit: Is it possible to create a type other than any for this?
| eventIndex: 3, | ||
| contents: JSON.stringify(msg3Contents), | ||
| subaccountId: { | ||
| owner: 'owner3', |
There was a problem hiding this comment.
nit: For the sake of correct-ness, this should match msg1.
indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts
Show resolved
Hide resolved
| export function convertToSubaccountMessage( | ||
| annotatedMessage: AnnotatedSubaccountMessage, | ||
| ): SubaccountMessage { | ||
| const subaccountMessage: SubaccountMessage = _.omit(annotatedMessage, ['orderId', 'isFill']); | ||
| return subaccountMessage; | ||
| } | ||
|
|
There was a problem hiding this comment.
nit: Is there a helpers file this can go into?
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Files selected for processing (6)
- indexer/services/ender/tests/lib/kafka-publisher.test.ts (2 hunks)
- indexer/services/ender/src/handlers/handler.ts (3 hunks)
- indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts (1 hunks)
- indexer/services/ender/src/lib/helper.ts (2 hunks)
- indexer/services/ender/src/lib/kafka-publisher.ts (3 hunks)
- indexer/services/ender/src/lib/types.ts (3 hunks)
Additional comments (Suppressed): 17
indexer/services/ender/src/handlers/order-fills/abstract-order-fill-handler.ts (1)
- 407-413: The
generateConsolidatedSubaccountKafkaEventmethod now includes two additional parameters:order?.idandtrue. Ensure that all calls to this method throughout the codebase have been updated to match the new signature. Also, verify that theorderobject is always defined when this method is called, asorder?.idcould potentially beundefined.indexer/services/ender/src/handlers/handler.ts (3)
10-30: The import statements have been updated to include the new
AnnotatedSubaccountMessagetype andSubaccountMessageContentsfrom the respective modules. Ensure that these new types are correctly defined and exported in their respective modules.105-111: The
generateConsolidatedSubaccountKafkaEventmethod now accepts three additional optional parameters:orderId,isFill, andsubaccountMessageContents. Make sure that all calls to this method throughout the codebase have been updated to pass these parameters where necessary.119-127: > Note: This review was outside of the patch, so it was mapped to the patch with the greatest overlap. Original lines [113-125]
The
subaccountMessageobject now includes the new propertiesorderId,isFill, andsubaccountMessageContents. Ensure that these properties are correctly handled in the downstream code where this object is consumed.indexer/services/ender/src/lib/helper.ts (2)
21-38: The import statements have been updated to include
SubaccountMessagefrom@dydxprotocol-indexer/v4-protosandAnnotatedSubaccountMessagefrom./types. Ensure that these new imports are used appropriately in the code.58-69: A new function
convertToSubaccountMessagehas been added. This function takes anAnnotatedSubaccountMessageas input and returns aSubaccountMessageby omitting certain properties. This function seems to be used for converting anAnnotatedSubaccountMessageto aSubaccountMessage. Ensure that this function is used correctly and that the omitted properties are not required in theSubaccountMessage.indexer/services/ender/src/lib/types.ts (3)
3-12: The import statement has been updated to include
SubaccountMessageContentsfrom@dydxprotocol-indexer/postgres. Ensure that this new import is used appropriately in the codebase.32-37: The import of
DateTimefromluxonand theEventHandlerDatainterface have been removed. Verify that these changes do not affect other parts of the codebase that might be using them.180-199: A new interface
AnnotatedSubaccountMessagehas been introduced, extendingSubaccountMessagewith additional propertiesorderId,isFill, andsubaccountMessageContents. TheConsolidatedKafkaEventtype has been updated to useAnnotatedSubaccountMessageinstead ofSubaccountMessage. Ensure that these changes are reflected in the rest of the codebase where these types and interfaces are used.indexer/services/ender/__tests__/lib/kafka-publisher.test.ts (3)
1-50: The new hunk introduces additional imports from
@dydxprotocol-indexer/postgresand../../src/lib/typesto support the new functionality being added. It also includes new helper functions for generating subaccount and transfer contents. The changes are consistent with the PR summary and seem to be necessary for the new functionality.170-240: The
sortEventsmethod is now being used to sort both trade and subaccount events. This is a good change as it reduces code duplication and makes the code more maintainable. However, it's important to ensure that thesortEventsmethod can handle both types of events correctly.242-403: The
aggregateFillEventsForSubaccountMessagestest case is introduced to verify the new functionality of aggregating multiple fills in a block for an IOC order into a single websocket message. The test case seems to cover the main scenarios and edge cases. However, it's important to ensure that theaggregateFillEventsForSubaccountMessagesmethod handles all possible edge cases correctly.indexer/services/ender/src/lib/kafka-publisher.ts (5)
6-40: The new hunk introduces a new type
AnnotatedSubaccountMessageand replaces the type ofsubaccountMessagesfromSubaccountMessage[]toAnnotatedSubaccountMessage[]. This change is likely to affect the rest of the code wheresubaccountMessagesis used. Ensure that all such instances are updated to handleAnnotatedSubaccountMessageinstead ofSubaccountMessage.57-82: The
addEventmethod has been refactored to use a new helper functiongetMessageswhich returns the appropriate message array based on the providedkafkaTopic. This change improves the readability and maintainability of the code.93-152: The new method
aggregateFillEventsForSubaccountMessagesis introduced to aggregate multiple fill events for an order into a single message. This is a significant change and should be thoroughly tested to ensure it works as expected and does not introduce any unintended side effects.158-174: The
sortTradeEventsmethod has been renamed tosortEventsand updated to handle anyOrderedMessage, not just trade events. This change makes the method more versatile and reusable.197-203: The
publishmethod now calls theaggregateFillEventsForSubaccountMessagesmethod before generating all topic Kafka messages. This change is part of the main feature introduced in this PR, which is to aggregate multiple fill events for an order into a single message.
| * contents contains all individual fills from that block. | ||
| * | ||
| * Due to separate handlers for order fills, we can be sure that if a message is annotated | ||
| * with a fill, it should only contain data about a single fill / order and not transfers. |
There was a problem hiding this comment.
nit: not transfers or positions.
| convertToSubaccountMessage(message4), | ||
| convertToSubaccountMessage(message2), | ||
| convertToSubaccountMessage(mergedMessage3), |
There was a problem hiding this comment.
is this the right ordering? Shouldn't message 4 come after message 3?
There was a problem hiding this comment.
yes.
message 4 has txn index 1/event index 4.
msg 2 has txn index 2/ event index1.
msg 3 has txn index 3/event index 1.
| if (message.isFill && message.orderId) { | ||
| const fills: | ||
| FillSubaccountMessageContents[] | undefined = message.subaccountMessageContents?.fills; | ||
| const orderId = message.orderId; |
| const lastEvent = lastEventForFills[orderId]; | ||
| const fills = allFillEvents[orderId]; |
| return -1; | ||
| } else if (Big(a.blockHeight).gt(b.blockHeight)) { | ||
| return 1; | ||
| public aggregateFillEventsForSubaccountMessages() { |
There was a problem hiding this comment.
nit: I think it still makes sense to add the TODO, to genericize so we can do a similar thing for positions / create a linear ticket.
There was a problem hiding this comment.
Review Status
Actionable comments generated: 0
Files selected for processing (1)
- indexer/services/ender/src/lib/kafka-publisher.ts (3 hunks)
Additional comments (Suppressed): 6
indexer/services/ender/src/lib/kafka-publisher.ts (6)
26-40: The
KafkaPublisherclass now usesAnnotatedSubaccountMessage[]instead ofSubaccountMessage[]forsubaccountMessages. This change is in line with the introduction of theAnnotatedSubaccountMessageinterface, which extendsSubaccountMessagewith additional properties. Ensure that all usages ofsubaccountMessagesin the codebase are updated to handleAnnotatedSubaccountMessageobjects.57-82: The
addEventmethod has been refactored to use the newgetMessageshelper function, which returns the appropriate message array based on the topic. This change improves the readability and maintainability of the code.94-141: The
sortTradeEventsmethod has been renamed toaggregateFillEventsForSubaccountMessagesand updated to aggregate fill events for subaccount messages. This change is in line with the PR's goal of aggregating multiple fills in a block for an IOC order into a single websocket message. The logic seems sound, but it's important to ensure that the aggregation of fill events doesn't inadvertently omit any important information.146-154: The
compareMessageshelper function is introduced to compare twoAnnotatedSubaccountMessageobjects based on block height, transaction index, and event index. This function is used in theaggregateFillEventsForSubaccountMessagesmethod to determine the order of fill events.160-176: The
sortEventsmethod is introduced to sort events by block height, transaction index, and event index in ascending order. This method is used in theaggregateFillEventsForSubaccountMessagesmethod to sort the aggregated fill events.199-205: The
generateAllTopicKafkaMessagesmethod now callsaggregateFillEventsForSubaccountMessagesbefore generating Kafka messages forsubaccountMessages. This ensures that the fill events are properly aggregated before they are published to Kafka.
…gated fill sfrom the entire block (#658)
Changelist
For order fills, the state transition of
CANCELED->FILLEDisn't considered valid by the FE. This can happen if there are multiple fills in a given block for an IOC order. Because the totalFilled < order amount on the first fill, Indexer will publish a websocket message withCANCELEDstatus. In subsequent fills, when totalFilled=order amount, Indexer will publish a follow-up message withFILLEDstatus.Rather than send individual messages per fill per order, send a single websocket message containing the order status of the last fill and all aggregated fills.
Test Plan
Unit tests
Author/Reviewer Checklist
state-breakinglabel.PrepareProposalorProcessProposal, manually add the labelproposal-breaking.feature:[feature-name].backport/[branch-name].refactor,chore,bug.