-
Notifications
You must be signed in to change notification settings - Fork 104
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Move FinalizeBlock event staging logic into a generic EventStager #2435
Changes from 1 commit
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,81 @@ | ||
package finalizeblock | ||
|
||
import ( | ||
"encoding/binary" | ||
|
||
"cosmossdk.io/store/prefix" | ||
storetypes "cosmossdk.io/store/types" | ||
"github.com/cosmos/cosmos-sdk/codec" | ||
sdk "github.com/cosmos/cosmos-sdk/types" | ||
"github.com/cosmos/gogoproto/proto" | ||
ante_types "github.com/dydxprotocol/v4-chain/protocol/app/ante/types" | ||
"github.com/dydxprotocol/v4-chain/protocol/lib" | ||
) | ||
|
||
// EventStager supports staging and retrieval of events (of type T) from FinalizeBlock. | ||
type EventStager[T proto.Message] struct { | ||
transientStoreKey storetypes.StoreKey | ||
cdc codec.BinaryCodec | ||
stagedEventCountKey string | ||
stagedEventKeyPrefix string | ||
} | ||
|
||
// NewEventStager creates a new EventStager. | ||
func NewEventStager[T proto.Message]( | ||
transientStoreKey storetypes.StoreKey, | ||
cdc codec.BinaryCodec, | ||
stagedEventCountKey string, | ||
stagedEventKeyPrefix string, | ||
) EventStager[T] { | ||
return EventStager[T]{ | ||
transientStoreKey: transientStoreKey, | ||
cdc: cdc, | ||
stagedEventCountKey: stagedEventCountKey, | ||
stagedEventKeyPrefix: stagedEventKeyPrefix, | ||
} | ||
} | ||
|
||
// GetStagedFinalizeBlockEventsFromStore retrieves all staged events from the store. | ||
func (s EventStager[T]) GetStagedFinalizeBlockEventsFromStore( | ||
store storetypes.KVStore, | ||
newStagedEvent func() T, | ||
) []T { | ||
count := s.getStagedEventsCount(store) | ||
events := make([]T, count) | ||
store = prefix.NewStore(store, []byte(s.stagedEventKeyPrefix)) | ||
for i := uint32(0); i < count; i++ { | ||
event := newStagedEvent() | ||
bytes := store.Get(lib.Uint32ToKey(i)) | ||
s.cdc.MustUnmarshal(bytes, event) | ||
events[i] = event | ||
teddyding marked this conversation as resolved.
Show resolved
Hide resolved
|
||
} | ||
return events | ||
} | ||
|
||
func (s EventStager[T]) getStagedEventsCount( | ||
store storetypes.KVStore, | ||
) uint32 { | ||
countsBytes := store.Get([]byte(s.stagedEventCountKey)) | ||
if countsBytes == nil { | ||
return 0 | ||
} | ||
return binary.BigEndian.Uint32(countsBytes) | ||
teddyding marked this conversation as resolved.
Show resolved
Hide resolved
|
||
} | ||
|
||
// StageFinalizeBlockEvent stages an event in the transient store. | ||
func (s EventStager[T]) StageFinalizeBlockEvent( | ||
ctx sdk.Context, | ||
eventBytes []byte, | ||
) { | ||
noGasCtx := ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter()) | ||
|
||
store := noGasCtx.TransientStore(s.transientStoreKey) | ||
teddyding marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
// Increment events count. | ||
count := s.getStagedEventsCount(store) | ||
store.Set([]byte(s.stagedEventCountKey), lib.Uint32ToKey(count+1)) | ||
|
||
// Store events keyed by index. | ||
store = prefix.NewStore(store, []byte(s.stagedEventKeyPrefix)) | ||
store.Set(lib.Uint32ToKey(count), eventBytes) | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,7 +1,6 @@ | ||
package streaming | ||
|
||
import ( | ||
"encoding/binary" | ||
"fmt" | ||
"sync" | ||
"sync/atomic" | ||
|
@@ -11,7 +10,6 @@ import ( | |
satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types" | ||
|
||
"cosmossdk.io/log" | ||
"cosmossdk.io/store/prefix" | ||
storetypes "cosmossdk.io/store/types" | ||
"github.com/cosmos/cosmos-sdk/codec" | ||
sdk "github.com/cosmos/cosmos-sdk/types" | ||
|
@@ -22,6 +20,8 @@ import ( | |
clobtypes "github.com/dydxprotocol/v4-chain/protocol/x/clob/types" | ||
|
||
ocutypes "github.com/dydxprotocol/v4-chain/protocol/indexer/off_chain_updates/types" | ||
|
||
"github.com/dydxprotocol/v4-chain/protocol/finalizeblock" | ||
) | ||
|
||
var _ types.FullNodeStreamingManager = (*FullNodeStreamingManagerImpl)(nil) | ||
|
@@ -62,6 +62,8 @@ type FullNodeStreamingManagerImpl struct { | |
|
||
// stores the staged FinalizeBlock events for full node streaming. | ||
streamingManagerTransientStoreKey storetypes.StoreKey | ||
|
||
finalizeBlockStager finalizeblock.EventStager[*clobtypes.StagedFinalizeBlockEvent] | ||
} | ||
|
||
// OrderbookSubscription represents a active subscription to the orderbook updates stream. | ||
|
@@ -119,6 +121,12 @@ func NewFullNodeStreamingManager( | |
|
||
streamingManagerTransientStoreKey: streamingManagerTransientStoreKey, | ||
cdc: cdc, | ||
finalizeBlockStager: finalizeblock.NewEventStager[*clobtypes.StagedFinalizeBlockEvent]( | ||
streamingManagerTransientStoreKey, | ||
cdc, | ||
StagedEventsCountKey, | ||
StagedEventsKeyPrefix, | ||
), | ||
} | ||
|
||
// Start the goroutine for pushing order updates through. | ||
|
@@ -381,14 +389,6 @@ func (sm *FullNodeStreamingManagerImpl) sendStreamUpdates( | |
} | ||
} | ||
|
||
func getStagedEventsCount(store storetypes.KVStore) uint32 { | ||
countsBytes := store.Get([]byte(StagedEventsCountKey)) | ||
if countsBytes == nil { | ||
return 0 | ||
} | ||
return binary.BigEndian.Uint32(countsBytes) | ||
} | ||
|
||
// Send a subaccount update event. | ||
func (sm *FullNodeStreamingManagerImpl) SendSubaccountUpdate( | ||
ctx sdk.Context, | ||
|
@@ -405,51 +405,32 @@ func (sm *FullNodeStreamingManagerImpl) SendSubaccountUpdate( | |
SubaccountUpdate: &subaccountUpdate, | ||
}, | ||
} | ||
sm.stageFinalizeBlockEvent( | ||
sm.finalizeBlockStager.StageFinalizeBlockEvent( | ||
ctx, | ||
sm.cdc.MustMarshal(&stagedEvent), | ||
) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. meganit: the Getter returns |
||
} | ||
|
||
func getStagedFinalizeBlockEventsFromStore( | ||
store storetypes.KVStore, | ||
cdc codec.BinaryCodec, | ||
) []clobtypes.StagedFinalizeBlockEvent { | ||
count := getStagedEventsCount(store) | ||
events := make([]clobtypes.StagedFinalizeBlockEvent, count) | ||
store = prefix.NewStore(store, []byte(StagedEventsKeyPrefix)) | ||
for i := uint32(0); i < count; i++ { | ||
var event clobtypes.StagedFinalizeBlockEvent | ||
bytes := store.Get(lib.Uint32ToKey(i)) | ||
cdc.MustUnmarshal(bytes, &event) | ||
events[i] = event | ||
} | ||
return events | ||
} | ||
|
||
// Retrieve all events staged during `FinalizeBlock`. | ||
func (sm *FullNodeStreamingManagerImpl) GetStagedFinalizeBlockEvents( | ||
ctx sdk.Context, | ||
) []clobtypes.StagedFinalizeBlockEvent { | ||
noGasCtx := ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter()) | ||
store := noGasCtx.TransientStore(sm.streamingManagerTransientStoreKey) | ||
return getStagedFinalizeBlockEventsFromStore(store, sm.cdc) | ||
} | ||
|
||
func (sm *FullNodeStreamingManagerImpl) stageFinalizeBlockEvent( | ||
ctx sdk.Context, | ||
eventBytes []byte, | ||
) { | ||
noGasCtx := ctx.WithGasMeter(ante_types.NewFreeInfiniteGasMeter()) | ||
store := noGasCtx.TransientStore(sm.streamingManagerTransientStoreKey) | ||
|
||
// Increment events count. | ||
count := getStagedEventsCount(store) | ||
store.Set([]byte(StagedEventsCountKey), lib.Uint32ToKey(count+1)) | ||
|
||
// Store events keyed by index. | ||
store = prefix.NewStore(store, []byte(StagedEventsKeyPrefix)) | ||
store.Set(lib.Uint32ToKey(count), eventBytes) | ||
events := sm.finalizeBlockStager.GetStagedFinalizeBlockEventsFromStore( | ||
store, | ||
func() *clobtypes.StagedFinalizeBlockEvent { | ||
return &clobtypes.StagedFinalizeBlockEvent{} | ||
}, | ||
) | ||
results := make([]clobtypes.StagedFinalizeBlockEvent, len(events)) | ||
for i, event := range events { | ||
if event == nil { | ||
panic("Got nil event from finalizeBlockStager") | ||
} | ||
results[i] = *event | ||
} | ||
return results | ||
teddyding marked this conversation as resolved.
Show resolved
Hide resolved
|
||
} | ||
|
||
// SendCombinedSnapshot sends messages to a particular subscriber without buffering. | ||
|
@@ -574,7 +555,7 @@ func (sm *FullNodeStreamingManagerImpl) SendOrderbookUpdates( | |
}, | ||
}, | ||
} | ||
sm.stageFinalizeBlockEvent( | ||
sm.finalizeBlockStager.StageFinalizeBlockEvent( | ||
ctx, | ||
sm.cdc.MustMarshal(&stagedEvent), | ||
) | ||
|
@@ -649,7 +630,7 @@ func (sm *FullNodeStreamingManagerImpl) SendOrderbookFillUpdate( | |
}, | ||
} | ||
|
||
sm.stageFinalizeBlockEvent( | ||
sm.finalizeBlockStager.StageFinalizeBlockEvent( | ||
ctx, | ||
sm.cdc.MustMarshal(&stagedEvent), | ||
) | ||
|
@@ -710,32 +691,6 @@ func getStreamUpdatesForSubaccountUpdates( | |
return streamUpdates, subaccountIds | ||
} | ||
|
||
// SendFinalizedSubaccountUpdates groups subaccount updates by their subaccount ids and | ||
// sends messages to the subscribers. | ||
func (sm *FullNodeStreamingManagerImpl) SendFinalizedSubaccountUpdates( | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This method and below interface are unused and should've been deleted (as part of internalize PR) |
||
subaccountUpdates []satypes.StreamSubaccountUpdate, | ||
blockHeight uint32, | ||
execMode sdk.ExecMode, | ||
) { | ||
defer metrics.ModuleMeasureSince( | ||
metrics.FullNodeGrpc, | ||
metrics.GrpcSendFinalizedSubaccountUpdatesLatency, | ||
time.Now(), | ||
) | ||
|
||
if execMode != sdk.ExecModeFinalize { | ||
panic("SendFinalizedSubaccountUpdates should only be called in ExecModeFinalize") | ||
} | ||
|
||
streamUpdates, subaccountIds := getStreamUpdatesForSubaccountUpdates( | ||
subaccountUpdates, | ||
blockHeight, | ||
execMode, | ||
) | ||
|
||
sm.AddSubaccountUpdatesToCache(streamUpdates, subaccountIds) | ||
} | ||
|
||
// AddOrderUpdatesToCache adds a series of updates to the full node streaming cache. | ||
// Clob pair ids are the clob pair id each update is relevant to. | ||
func (sm *FullNodeStreamingManagerImpl) AddOrderUpdatesToCache( | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
also meganit: take in
ctx
and move the free gas meter here for consistency w setter method