wip recurring snapshots

.github/workflows/protocol-build-and-push-snapshot.yml

@@ -3,6 +3,7 @@ name: Protocol Build & Push Image to AWS ECR
 on:  # yamllint disable-line rule:truthy
   push:
     branches:
+      - jonfung/**
       - main
       - 'release/protocol/v[0-9]+.[0-9]+.x'  # e.g. release/protocol/v0.1.x
       - 'release/protocol/v[0-9]+.x'  # e.g. release/protocol/v1.x

.github/workflows/protocol-build-and-push.yml

@@ -3,6 +3,7 @@ name: Protocol Build & Push Image to AWS ECR
 on:  # yamllint disable-line rule:truthy
   push:
     branches:
+      - jonfung/**
       - main
       - 'release/protocol/v[0-9]+.[0-9]+.x'  # e.g. release/protocol/v0.1.x
       - 'release/protocol/v[0-9]+.x'  # e.g. release/protocol/v1.x

protocol/app/app.go

@@ -2030,6 +2030,8 @@ func getFullNodeStreamingManagerFromOptions(
 			appFlags.GrpcStreamingFlushIntervalMs,
 			appFlags.GrpcStreamingMaxBatchSize,
 			appFlags.GrpcStreamingMaxChannelBufferSize,
+			// TODO appflag this.
+			50,
 		)
 	}
 	return streaming.NewNoopGrpcStreamingManager()

protocol/streaming/full_node_streaming_manager.go

@@ -2,10 +2,11 @@ package streaming
 
 import (
 	"fmt"
-	satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types"
 	"sync"
 	"time"
 
+	satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types"
+
 	"cosmossdk.io/log"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/dydxprotocol/v4-chain/protocol/lib/metrics"
@@ -44,14 +45,18 @@ type FullNodeStreamingManagerImpl struct {
 
 	maxUpdatesInCache          uint32
 	maxSubscriptionChannelSize uint32
+
+	// Block interval in which snapshot info should be sent out in.
+	// Defaults to 0, which means only one snapshot will be sent out.
+	snapshotBlockInterval uint32
 }
 
 // OrderbookSubscription represents a active subscription to the orderbook updates stream.
 type OrderbookSubscription struct {
 	subscriptionId uint32
 
 	// Initialize the subscription with orderbook snapshots.
-	initialize sync.Once
+	initialize *sync.Once
 
 	// Clob pair ids to subscribe to.
 	clobPairIds []uint32
@@ -64,13 +69,18 @@ type OrderbookSubscription struct {
 
 	// Channel to buffer writes before the stream
 	updatesChannel chan []clobtypes.StreamUpdate
+
+	// Block height initial snapshot was emitted at modulo snapshotBlockInterval.
+	// Used to keep position of subsequent snapshots.
+	snapshotDelayBlock uint32
 }
 
 func NewFullNodeStreamingManager(
 	logger log.Logger,
 	flushIntervalMs uint32,
 	maxUpdatesInCache uint32,
 	maxSubscriptionChannelSize uint32,
+	snapshotBlockInterval uint32,
 ) *FullNodeStreamingManagerImpl {
 	logger = logger.With(log.ModuleKey, "full-node-streaming")
 	fullNodeStreamingManager := &FullNodeStreamingManagerImpl{
@@ -87,6 +97,7 @@ func NewFullNodeStreamingManager(
 
 		maxUpdatesInCache:          maxUpdatesInCache,
 		maxSubscriptionChannelSize: maxSubscriptionChannelSize,
+		snapshotBlockInterval:      snapshotBlockInterval,
 	}
 
 	// Start the goroutine for pushing order updates through.
@@ -149,6 +160,7 @@ func (sm *FullNodeStreamingManagerImpl) Subscribe(
 	}
 	subscription := &OrderbookSubscription{
 		subscriptionId: sm.nextSubscriptionId,
+		initialize:     &sync.Once{},
 		clobPairIds:    clobPairIds,
 		subaccountIds:  sIds,
 		messageSender:  messageSender,
@@ -674,7 +686,16 @@ func (sm *FullNodeStreamingManagerImpl) InitializeNewStreams(
 	sm.FlushStreamUpdatesWithLock()
 
 	updatesByClobPairId := make(map[uint32]*clobtypes.OffchainUpdates)
+
 	for subscriptionId, subscription := range sm.orderbookSubscriptions {
+		// If the snapshot block interval is enabled, reset the sync.Once in order to
+		// re-send snapshots out.
+		if sm.snapshotBlockInterval > 0 &&
+			(blockHeight+subscription.snapshotDelayBlock)%uint32(sm.snapshotBlockInterval) == 0 {
+			subscription.initialize = &sync.Once{}
+			sm.logger.Info(fmt.Sprintf("resetting sync once for sub id %+v", subscriptionId))
+		}
+
 		subscription.initialize.Do(
 			func() {
 				allUpdates := clobtypes.NewOffchainUpdates()
@@ -688,8 +709,11 @@ func (sm *FullNodeStreamingManagerImpl) InitializeNewStreams(
 				for _, subaccountId := range subscription.subaccountIds {
 					saUpdates = append(saUpdates, getSubaccountSnapshot(subaccountId))
 				}
-
+				sm.logger.Info(fmt.Sprintf("Sending out snapshot for %+v", subscriptionId))
 				sm.SendCombinedSnapshot(allUpdates, saUpdates, subscriptionId, blockHeight, execMode)
+				if sm.snapshotBlockInterval != 0 {
+					subscription.snapshotDelayBlock = blockHeight % uint32(sm.snapshotBlockInterval)
+				}
 			},
 		)
 	}