Chunk liquidation daemon collat check request (#371)

* Chunk liquidation daemon collat check request

* comments

* comment

protocol/daemons/flags/flags.go

@@ -22,6 +22,7 @@ const (
 	FlagLiquidationDaemonEnabled             = "liquidation-daemon-enabled"
 	FlagLiquidationDaemonLoopDelayMs         = "liquidation-daemon-loop-delay-ms"
 	FlagLiquidationDaemonSubaccountPageLimit = "liquidation-daemon-subaccount-page-limit"
+	FlagLiquidationDaemonRequestChunkSize    = "liquidation-daemon-request-chunk-size"
 )
 
 type SharedFlags struct {
@@ -39,6 +40,7 @@ type LiquidationFlags struct {
 	Enabled             bool
 	LoopDelayMs         uint32
 	SubaccountPageLimit uint64
+	RequestChunkSize    uint64
 }
 
 type PriceFlags struct {
@@ -71,6 +73,7 @@ func GetDefaultDaemonFlags() DaemonFlags {
 				Enabled:             true,
 				LoopDelayMs:         1_600,
 				SubaccountPageLimit: 1_000,
+				RequestChunkSize:    500,
 			},
 			Price: PriceFlags{
 				Enabled:     true,
@@ -136,6 +139,11 @@ func AddDaemonFlagsToCmd(
 		df.Liquidation.SubaccountPageLimit,
 		"Limit on the number of subaccounts to fetch per query in the Liquidation Daemon task loop.",
 	)
+	cmd.Flags().Uint64(
+		FlagLiquidationDaemonRequestChunkSize,
+		df.Liquidation.RequestChunkSize,
+		"Limit on the number of subaccounts per collateralization check in the Liquidation Daemon task loop.",
+	)
 
 	// Price Daemon.
 	cmd.Flags().Bool(
@@ -186,6 +194,9 @@ func GetDaemonFlagValuesFromOptions(
 	if v, ok := appOpts.Get(FlagLiquidationDaemonSubaccountPageLimit).(uint64); ok {
 		result.Liquidation.SubaccountPageLimit = v
 	}
+	if v, ok := appOpts.Get(FlagLiquidationDaemonRequestChunkSize).(uint64); ok {
+		result.Liquidation.RequestChunkSize = v
+	}
 
 	// Price Daemon.
 	if v, ok := appOpts.Get(FlagPriceDaemonEnabled).(bool); ok {

protocol/daemons/flags/flags_test.go

@@ -51,6 +51,7 @@ func TestGetDaemonFlagValuesFromOptions_Custom(t *testing.T) {
 	optsMap[flags.FlagLiquidationDaemonEnabled] = true
 	optsMap[flags.FlagLiquidationDaemonLoopDelayMs] = uint32(2222)
 	optsMap[flags.FlagLiquidationDaemonSubaccountPageLimit] = uint64(3333)
+	optsMap[flags.FlagLiquidationDaemonRequestChunkSize] = uint64(4444)
 
 	optsMap[flags.FlagPriceDaemonEnabled] = true
 	optsMap[flags.FlagPriceDaemonLoopDelayMs] = uint32(4444)
@@ -76,6 +77,7 @@ func TestGetDaemonFlagValuesFromOptions_Custom(t *testing.T) {
 	require.Equal(t, optsMap[flags.FlagLiquidationDaemonEnabled], r.Liquidation.Enabled)
 	require.Equal(t, optsMap[flags.FlagLiquidationDaemonLoopDelayMs], r.Liquidation.LoopDelayMs)
 	require.Equal(t, optsMap[flags.FlagLiquidationDaemonSubaccountPageLimit], r.Liquidation.SubaccountPageLimit)
+	require.Equal(t, optsMap[flags.FlagLiquidationDaemonRequestChunkSize], r.Liquidation.RequestChunkSize)
 
 	// Price Daemon.
 	require.Equal(t, optsMap[flags.FlagPriceDaemonEnabled], r.Price.Enabled)

protocol/daemons/liquidation/client/client.go

@@ -87,7 +87,7 @@ func RunLiquidationDaemonTaskLoop(
 		metrics.Latency,
 	)
 
-	// Fetch all subaccounts from query service.
+	// 1. Fetch all subaccounts from query service.
 	subaccounts, err := GetAllSubaccounts(
 		ctx,
 		subaccountQueryClient,
@@ -96,55 +96,19 @@ func RunLiquidationDaemonTaskLoop(
 	if err != nil {
 		return err
 	}
-	telemetry.ModuleSetGauge(
-		metrics.LiquidationDaemon,
-		float32(len(subaccounts)),
-		metrics.GetAllSubaccounts,
-		metrics.Count,
-	)
 
-	// Filter out subaccounts with no open positions.
-	subaccountsWithOpenPositions := make([]satypes.SubaccountId, 0)
-	for _, subaccount := range subaccounts {
-		if len(subaccount.PerpetualPositions) > 0 {
-			subaccountsWithOpenPositions = append(subaccountsWithOpenPositions, *subaccount.Id)
-		}
-	}
-	telemetry.ModuleSetGauge(
-		metrics.LiquidationDaemon,
-		float32(len(subaccountsWithOpenPositions)),
-		metrics.SubaccountsWithOpenPositions,
-		metrics.Count,
+	// 2. Check collateralization statuses of subaccounts with at least one open position.
+	liquidatableSubaccountIds, err := GetLiquidatableSubaccountIds(
+		ctx,
+		clobQueryClient,
+		liqFlags,
+		subaccounts,
 	)
-	liquidatableSubaccountIds := make([]satypes.SubaccountId, 0)
-	if len(subaccountsWithOpenPositions) > 0 {
-		// Check collateralization statuses of subaccounts with at least one open position.
-		collateralizationCheckResults, err :=
-			CheckCollateralizationForSubaccounts(
-				ctx,
-				clobQueryClient,
-				subaccountsWithOpenPositions,
-			)
-		if err != nil {
-			return err
-		}
-
-		// Append all liquidatable subaccount ids to a new slice.
-		for _, result := range collateralizationCheckResults {
-			if result.IsLiquidatable {
-				liquidatableSubaccountIds = append(liquidatableSubaccountIds, result.SubaccountId)
-			}
-		}
+	if err != nil {
+		return err
 	}
 
-	telemetry.ModuleSetGauge(
-		metrics.LiquidationDaemon,
-		float32(len(liquidatableSubaccountIds)),
-		metrics.LiquidatableSubaccountIds,
-		metrics.Count,
-	)
-
-	// Send the list of liquidatable subaccount ids to the daemon server.
+	// 3. Send the list of liquidatable subaccount ids to the daemon server.
 	err = SendLiquidatableSubaccountIds(
 		ctx,
 		liquidationServiceClient,
@@ -189,9 +153,73 @@ func GetAllSubaccounts(
 			break
 		}
 	}
+
+	telemetry.ModuleSetGauge(
+		metrics.LiquidationDaemon,
+		float32(len(subaccounts)),
+		metrics.GetAllSubaccounts,
+		metrics.Count,
+	)
+
 	return subaccounts, nil
 }
 
+// GetLiquidatableSubaccountIds verifies collateralization statuses of subaccounts with
+// at least one open position and returns a list of unique and potentially liquidatable subaccount ids.
+func GetLiquidatableSubaccountIds(
+	ctx context.Context,
+	client clobtypes.QueryClient,
+	liqFlags flags.LiquidationFlags,
+	subaccounts []satypes.Subaccount,
+) (
+	liquidatableSubaccountIds []satypes.SubaccountId,
+	err error,
+) {
+	defer telemetry.ModuleMeasureSince(
+		metrics.LiquidationDaemon,
+		time.Now(),
+		metrics.GetLiquidatableSubaccountIds,
+		metrics.Latency,
+	)
+
+	// Filter out subaccounts with no open positions.
+	subaccountsToCheck := make([]satypes.SubaccountId, 0)
+	for _, subaccount := range subaccounts {
+		if len(subaccount.PerpetualPositions) > 0 {
+			subaccountsToCheck = append(subaccountsToCheck, *subaccount.Id)
+		}
+	}
+
+	telemetry.ModuleSetGauge(
+		metrics.LiquidationDaemon,
+		float32(len(subaccountsToCheck)),
+		metrics.SubaccountsWithOpenPositions,
+		metrics.Count,
+	)
+
+	// Query the gRPC server in chunks of size `liqFlags.RequestChunkSize`.
+	liquidatableSubaccountIds = make([]satypes.SubaccountId, 0)
+	for start := 0; start < len(subaccountsToCheck); start += int(liqFlags.RequestChunkSize) {
+		end := lib.Min(start+int(liqFlags.RequestChunkSize), len(subaccountsToCheck))
+
+		results, err := CheckCollateralizationForSubaccounts(
+			ctx,
+			client,
+			subaccountsToCheck[start:end],
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, result := range results {
+			if result.IsLiquidatable {
+				liquidatableSubaccountIds = append(liquidatableSubaccountIds, result.SubaccountId)
+			}
+		}
+	}
+	return liquidatableSubaccountIds, nil
+}
+
 // CheckCollateralizationForSubaccounts queries a gRPC server using `AreSubaccountsLiquidatable`
 // and returns a list of collateralization statuses for the given list of subaccount ids.
 func CheckCollateralizationForSubaccounts(
@@ -233,6 +261,13 @@ func SendLiquidatableSubaccountIds(
 		metrics.Latency,
 	)
 
+	telemetry.ModuleSetGauge(
+		metrics.LiquidationDaemon,
+		float32(len(subaccountIds)),
+		metrics.LiquidatableSubaccountIds,
+		metrics.Count,
+	)
+
 	request := &api.LiquidateSubaccountsRequest{
 		SubaccountIds: subaccountIds,
 	}

protocol/lib/metrics/constants.go

@@ -242,6 +242,7 @@ const (
 	// Liquidation Daemon.
 	CheckCollateralizationForSubaccounts = "check_collateralization_for_subaccounts"
 	GetAllSubaccounts                    = "get_all_subaccounts"
+	GetLiquidatableSubaccountIds         = "get_liquidatable_subaccount_ids"
 	GetSubaccountsFromKey                = "get_subaccounts_from_key"
 	LiquidatableSubaccountIds            = "liquidatable_subaccount_ids"
 	LiquidationDaemon                    = "liquidation_daemon"