[CT-1050] DeliverTx state change reset for subaccount updates

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

@@ -3,47 +3,12 @@ name: Protocol Build & Push Image to AWS ECR
 on:  # yamllint disable-line rule:truthy
   push:
     branches:
+      - 'wl/sa3'
       - 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
 
 jobs:
-  build-and-push-dev:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./protocol
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: '0'  # without this, ignite fails.
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_VALIDATOR_DEV }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_VALIDATOR_DEV }}
-          aws-region: us-east-2
-
-      - name: Login to Amazon ECR
-        id: login-ecr
-        uses: aws-actions/amazon-ecr-login@v1
-
-      - name: Build, Tag, and Push the Image to Amazon ECR
-        id: build-image
-        env:
-          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          ECR_REPOSITORY: dev-validator
-        run: |
-          make localnet-build-amd64
-          commit_hash=$(git rev-parse --short=7 HEAD)
-          docker build \
-            --platform amd64 \
-            -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
-            -f testing/testnet-dev/Dockerfile .
-          docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags
-
   build-and-push-dev2:
     runs-on: ubuntu-latest
     defaults:
@@ -79,147 +44,3 @@ jobs:
             -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
             -f testing/testnet-dev/Dockerfile .
           docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags
-
-  build-and-push-dev3:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./protocol
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: '0'  # without this, ignite fails.
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_VALIDATOR_DEV3 }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_VALIDATOR_DEV3 }}
-          aws-region: us-east-2
-
-      - name: Login to Amazon ECR
-        id: login-ecr
-        uses: aws-actions/amazon-ecr-login@v1
-
-      - name: Build, Tag, and Push the Image to Amazon ECR
-        id: build-image
-        env:
-          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          ECR_REPOSITORY: dev3-validator
-        run: |
-          make localnet-build-amd64
-          commit_hash=$(git rev-parse --short=7 HEAD)
-          docker build \
-            --platform amd64 \
-            -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
-            -f testing/testnet-dev/Dockerfile .
-          docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags
-
-  build-and-push-dev4:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./protocol
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: '0'  # without this, ignite fails.
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_VALIDATOR_DEV4 }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_VALIDATOR_DEV4 }}
-          aws-region: us-east-2
-
-      - name: Login to Amazon ECR
-        id: login-ecr
-        uses: aws-actions/amazon-ecr-login@v1
-
-      - name: Build, Tag, and Push the Image to Amazon ECR
-        id: build-image
-        env:
-          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          ECR_REPOSITORY: dev4-validator
-        run: |
-          make localnet-build-amd64
-          commit_hash=$(git rev-parse --short=7 HEAD)
-          docker build \
-            --platform amd64 \
-            -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
-            -f testing/testnet-dev/Dockerfile .
-          docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags
-
-  build-and-push-dev5:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./protocol
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: '0'  # without this, ignite fails.
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_VALIDATOR_DEV5 }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_VALIDATOR_DEV5 }}
-          aws-region: us-east-2
-
-      - name: Login to Amazon ECR
-        id: login-ecr
-        uses: aws-actions/amazon-ecr-login@v1
-
-      - name: Build, Tag, and Push the Image to Amazon ECR
-        id: build-image
-        env:
-          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          ECR_REPOSITORY: dev5-validator
-        run: |
-          make localnet-build-amd64
-          commit_hash=$(git rev-parse --short=7 HEAD)
-          docker build \
-            --platform amd64 \
-            -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
-            -f testing/testnet-dev/Dockerfile .
-          docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags
-
-  build-and-push-staging:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: ./protocol
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: '0'  # without this, ignite fails.
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@v1
-        with:
-          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID_VALIDATOR_STAGING }}
-          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY_VALIDATOR_STAGING }}
-          aws-region: us-east-2
-
-      - name: Login to Amazon ECR
-        id: login-ecr
-        uses: aws-actions/amazon-ecr-login@v1
-
-      - name: Build, Tag, and Push the Image to Amazon ECR
-        id: build-image
-        env:
-          ECR_REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          ECR_REPOSITORY: staging-validator
-        run: |
-          make localnet-build-amd64
-          commit_hash=$(git rev-parse --short=7 HEAD)
-          docker build \
-            --platform amd64 \
-            -t $ECR_REGISTRY/$ECR_REPOSITORY:$commit_hash \
-            -f testing/testnet-staging/Dockerfile .
-          docker push $ECR_REGISTRY/$ECR_REPOSITORY --all-tags

protocol/x/clob/keeper/keeper.go

@@ -264,31 +264,11 @@ func (k Keeper) InitializeNewStreams(ctx sdk.Context) {
 			)
 		},
 		func(subaccountId satypes.SubaccountId) *satypes.StreamSubaccountUpdate {
-			subaccount := k.subaccountsKeeper.GetSubaccount(
+			subaccountUpdate := k.subaccountsKeeper.GetStreamSubaccountUpdate(
 				ctx,
 				subaccountId,
 			)
-			assetPositions := make([]*satypes.SubaccountAssetPosition, len(subaccount.AssetPositions))
-			for i, ap := range subaccount.AssetPositions {
-				assetPositions[i] = &satypes.SubaccountAssetPosition{
-					AssetId:  ap.AssetId,
-					Quantums: ap.Quantums.BigInt().Uint64(),
-				}
-			}
-			perpetualPositions := make([]*satypes.SubaccountPerpetualPosition, len(subaccount.PerpetualPositions))
-			for i, pp := range subaccount.PerpetualPositions {
-				perpetualPositions[i] = &satypes.SubaccountPerpetualPosition{
-					PerpetualId: pp.PerpetualId,
-					Quantums:    pp.Quantums.BigInt().Uint64(),
-				}
-			}
-
-			return &satypes.StreamSubaccountUpdate{
-				SubaccountId:              &subaccountId,
-				UpdatedAssetPositions:     assetPositions,
-				UpdatedPerpetualPositions: perpetualPositions,
-				Snapshot:                  true,
-			}
+			return &subaccountUpdate
 		},
 		lib.MustConvertIntegerToUint32(ctx.BlockHeight()),
 		ctx.ExecMode(),

protocol/x/clob/keeper/process_operations.go

@@ -38,6 +38,21 @@ func fetchOrdersInvolvedInOpQueue(
 	return orderIdSet
 }
 
+// fetchSubaccountIdsInvolvedInOpQueue fetches all SubaccountIds involved in an operations
+// queue's matches and returns them as a set.
+func fetchSubaccountIdsInvolvedInOpQueue(
+	operations []types.InternalOperation,
+) (subaccountIdSet map[satypes.SubaccountId]struct{}) {
+	subaccountIdSet = make(map[satypes.SubaccountId]struct{})
+	for _, operation := range operations {
+		if clobMatch := operation.GetMatch(); clobMatch != nil {
+			subaccountIdSetForClobMatch := clobMatch.GetAllSubaccountIds()
+			subaccountIdSet = lib.MergeMaps(subaccountIdSet, subaccountIdSetForClobMatch)
+		}
+	}
+	return subaccountIdSet
+}
+
 // ProcessProposerOperations updates on-chain state given an []OperationRaw operations queue
 // representing matches that occurred in the previous block. It performs validation on an operations
 // queue. If all validation passes, the operations queue is written to state.
@@ -58,6 +73,10 @@ func (k Keeper) ProcessProposerOperations(
 	}
 
 	// If grpc streams are on, send absolute fill amounts from local + proposed opqueue to the grpc stream.
+	// Also send subaccount snapshots for impacted subaccounts.
+	// An impacted subaccount is defined as:
+	// - A subaccount that was involved in any match in the local opqueue.
+	//   Only matches generate subaccount updates.
 	// This must be sent out to account for checkState being discarded and deliverState being used.
 	if streamingManager := k.GetFullNodeStreamingManager(); streamingManager.Enabled() {
 		localValidatorOperationsQueue, _ := k.MemClob.GetOperationsToReplay(ctx)
@@ -75,6 +94,21 @@ func (k Keeper) ProcessProposerOperations(
 			allUpdates.Append(orderbookUpdate)
 		}
 		k.SendOrderbookUpdates(ctx, allUpdates)
+
+		// send subaccount snapshots
+		subaccountIdsFromProposed := fetchSubaccountIdsInvolvedInOpQueue(
+			operations,
+		)
+		subaccountIdsFromLocal := fetchSubaccountIdsInvolvedInOpQueue(
+			localValidatorOperationsQueue,
+		)
+		subaccountIdsToUpdate := lib.MergeMaps(subaccountIdsFromLocal, subaccountIdsFromProposed)
+		allSubaccountUpdates := make([]satypes.StreamSubaccountUpdate, 0)
+		for subaccountId := range subaccountIdsToUpdate {
+			subaccountUpdate := k.subaccountsKeeper.GetStreamSubaccountUpdate(ctx, subaccountId)
+			allSubaccountUpdates = append(allSubaccountUpdates, subaccountUpdate)
+		}
+		k.subaccountsKeeper.SendSubaccountUpdates(ctx, allSubaccountUpdates)
 	}
 
 	log.DebugLog(ctx, "Processing operations queue",

protocol/x/clob/types/expected_keepers.go

@@ -37,6 +37,12 @@ type SubaccountsKeeper interface {
 	) (
 		val satypes.Subaccount,
 	)
+	GetStreamSubaccountUpdate(
+		ctx sdk.Context,
+		id satypes.SubaccountId,
+	) (
+		val satypes.StreamSubaccountUpdate,
+	)
 	GetAllSubaccount(
 		ctx sdk.Context,
 	) (
@@ -78,6 +84,10 @@ type SubaccountsKeeper interface {
 		quantums *big.Int,
 		perpetualId uint32,
 	) error
+	SendSubaccountUpdates(
+		ctx sdk.Context,
+		subaccountUpdates []satypes.StreamSubaccountUpdate,
+	)
 }
 
 type AssetsKeeper interface {

protocol/x/clob/types/message_clob_match.go

@@ -1,5 +1,7 @@
 package types
 
+import satypes "github.com/dydxprotocol/v4-chain/protocol/x/subaccounts/types"
+
 // NewClobMatchFromMatchOrders creates a `ClobMatch` from the provided `MatchOrders`.
 func NewClobMatchFromMatchOrders(
 	msgMatchOrders *MatchOrders,
@@ -40,3 +42,27 @@ func (clobMatch *ClobMatch) GetAllOrderIds() (orderIds map[OrderId]struct{}) {
 	}
 	return orderIds
 }
+
+// GetAllSubaccountIds returns a set of subaccountIds involved in a ClobMatch.
+func (clobMatch *ClobMatch) GetAllSubaccountIds() (subaccountIds map[satypes.SubaccountId]struct{}) {
+	subaccountIds = make(map[satypes.SubaccountId]struct{})
+	if matchOrders := clobMatch.GetMatchOrders(); matchOrders != nil {
+		subaccountIds[matchOrders.GetTakerOrderId().SubaccountId] = struct{}{}
+		for _, makerFill := range matchOrders.GetFills() {
+			subaccountIds[makerFill.GetMakerOrderId().SubaccountId] = struct{}{}
+		}
+	}
+	if matchOrders := clobMatch.GetMatchPerpetualLiquidation(); matchOrders != nil {
+		subaccountIds[matchOrders.GetLiquidated()] = struct{}{}
+		for _, makerFill := range matchOrders.GetFills() {
+			subaccountIds[makerFill.GetMakerOrderId().SubaccountId] = struct{}{}
+		}
+	}
+	if matchOrders := clobMatch.GetMatchPerpetualDeleveraging(); matchOrders != nil {
+		subaccountIds[matchOrders.GetLiquidated()] = struct{}{}
+		for _, makerFill := range matchOrders.GetFills() {
+			subaccountIds[makerFill.GetOffsettingSubaccountId()] = struct{}{}
+		}
+	}
+	return subaccountIds
+}

protocol/x/subaccounts/keeper/subaccount.go

@@ -132,6 +132,34 @@ func (k Keeper) GetSubaccount(
 	return val
 }
 
+func (k Keeper) GetStreamSubaccountUpdate(
+	ctx sdk.Context,
+	id types.SubaccountId,
+) (val types.StreamSubaccountUpdate) {
+	subaccount := k.GetSubaccount(ctx, id)
+	assetPositions := make([]*types.SubaccountAssetPosition, len(subaccount.AssetPositions))
+	for i, ap := range subaccount.AssetPositions {
+		assetPositions[i] = &types.SubaccountAssetPosition{
+			AssetId:  ap.AssetId,
+			Quantums: ap.Quantums.BigInt().Uint64(),
+		}
+	}
+	perpetualPositions := make([]*types.SubaccountPerpetualPosition, len(subaccount.PerpetualPositions))
+	for i, pp := range subaccount.PerpetualPositions {
+		perpetualPositions[i] = &types.SubaccountPerpetualPosition{
+			PerpetualId: pp.PerpetualId,
+			Quantums:    pp.Quantums.BigInt().Uint64(),
+		}
+	}
+
+	return types.StreamSubaccountUpdate{
+		SubaccountId:              &id,
+		UpdatedAssetPositions:     assetPositions,
+		UpdatedPerpetualPositions: perpetualPositions,
+		Snapshot:                  true,
+	}
+}
+
 // GetAllSubaccount returns all subaccount.
 // For more performant searching and iteration, use `ForEachSubaccount`.
 func (k Keeper) GetAllSubaccount(ctx sdk.Context) (list []types.Subaccount) {

protocol/x/subaccounts/types/types.go

@@ -62,6 +62,10 @@ type SubaccountsKeeper interface {
 		ctx sdk.Context,
 		id SubaccountId,
 	) (val Subaccount)
+	GetStreamSubaccountUpdate(
+		ctx sdk.Context,
+		id SubaccountId,
+	) (val StreamSubaccountUpdate)
 	LegacyGetNegativeTncSubaccountSeenAtBlock(ctx sdk.Context) (uint32, bool)
 	GetNegativeTncSubaccountSeenAtBlock(
 		ctx sdk.Context,