flush updates before snapshot

protocol/streaming/grpc/grpc_streaming_manager.go

@@ -205,9 +205,9 @@ func (sm *GrpcStreamingManagerImpl) Stop() {
 	sm.done <- true
 }
 
-// SendSnapshot groups updates by their clob pair ids and
-// sends messages to the subscribers. It groups out updates differently
-// and bypasses the buffer.
+// SendSnapshot sends messages to a particular subscriber without buffering.
+// Note this method requires the lock and assumes that the lock has already been
+// acquired by the caller.
 func (sm *GrpcStreamingManagerImpl) SendSnapshot(
 	offchainUpdates *clobtypes.OffchainUpdates,
 	subscriptionId uint32,
@@ -387,17 +387,22 @@ func (sm *GrpcStreamingManagerImpl) AddUpdatesToCache(
 	sm.EmitMetrics()
 }
 
-// FlushStreamUpdates takes in a map of clob pair id to stream updates and emits them to subscribers.
 func (sm *GrpcStreamingManagerImpl) FlushStreamUpdates() {
+	sm.Lock()
+	defer sm.Unlock()
+	sm.FlushStreamUpdatesWithLock()
+}
+
+// FlushStreamUpdatesWithLock takes in a map of clob pair id to stream updates and emits them to subscribers.
+// Note this method requires the lock and assumes that the lock has already been
+// acquired by the caller.
+func (sm *GrpcStreamingManagerImpl) FlushStreamUpdatesWithLock() {
 	defer metrics.ModuleMeasureSince(
 		metrics.FullNodeGrpc,
 		metrics.GrpcFlushUpdatesLatency,
 		time.Now(),
 	)
 
-	sm.Lock()
-	defer sm.Unlock()
-
 	// Non-blocking send updates through subscriber's buffered channel.
 	// If the buffer is full, drop the subscription.
 	idsToRemove := make([]uint32, 0)
@@ -446,6 +451,10 @@ func (sm *GrpcStreamingManagerImpl) InitializeNewGrpcStreams(
 	sm.Lock()
 	defer sm.Unlock()
 
+	// Flush any pending updates before sending the snapshot to avoid
+	// race conditions with the snapshot.
+	sm.FlushStreamUpdatesWithLock()
+
 	updatesByClobPairId := make(map[uint32]*clobtypes.OffchainUpdates)
 	for subscriptionId, subscription := range sm.orderbookSubscriptions {
 		subscription.initialize.Do(