fixes 1162; race condition between delegation record queries and withdrawal acks (#1183)

* fixes 1162; race condition between delegation record queries and withdrawal acks

* lint

Author: Joe Bowman

Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.21.7-alpine3.19 AS builder
+FROM golang:1.21.7-alpine3.18 AS builder
 RUN apk add --no-cache git musl-dev openssl-dev linux-headers ca-certificates build-base
 
 WORKDIR /src/app/
@@ -22,7 +22,7 @@ RUN --mount=type=cache,target=/root/.cache/go-build \
     LINK_STATICALLY=true make build
 
 # Add to a distroless container
-FROM alpine:3.19
+FROM alpine:3.18
 COPY --from=builder /src/app/build/quicksilverd /usr/local/bin/quicksilverd
 RUN adduser -S -h /quicksilver -D quicksilver -u 1000
 USER quicksilver

scripts/registerzone.json

@@ -6,16 +6,17 @@
   "content": {
   "@type": "/quicksilver.interchainstaking.v1.RegisterZoneProposal",
   "title": "register lstest-1 zone",
-  "description": "register lstest-1 zone with multisend and lsm enabled",
+  "description": "register lstest-1 zone with lsm disabled",
   "connection_id": "connection-0",
   "base_denom": "uatom",
   "local_denom": "uqatom",
   "account_prefix": "cosmos",
   "deposits_enabled": true,
   "unbonding_enabled": true,
   "liquidity_module": false,
-  "return_to_sender": true,
-  "decimals": 6
+  "return_to_sender": false,
+  "decimals": 6,
+  "is_118": true
   },
   "authority": "quick10d07y265gmmuvt4z0w9aw880jnsr700j3xrh0p"
 }],

scripts/setup.sh

@@ -477,6 +477,7 @@ fi
 ## set the 'epoch' epoch to 5m interval
 jq '.app_state.epochs.epochs = [{"identifier": "epoch","start_time": "0001-01-01T00:00:00Z","duration": "360s","current_epoch": "0","current_epoch_start_time": "0001-01-01T00:00:00Z","epoch_counting_started": false,"current_epoch_start_height": "0"},{"identifier": "day","start_time": "0001-01-01T00:00:00Z","duration": "120s","current_epoch": "0","current_epoch_start_time": "0001-01-01T00:00:00Z","epoch_counting_started": false,"current_epoch_start_height": "0"}]' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,}
 jq '.app_state.interchainstaking.params.deposit_interval = 25' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,}
+jq '.app_state.interchainstaking.params.unbonding_enabled = true' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,} 
 jq '.app_state.mint.params.epoch_identifier = "epoch"' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,}
 jq '.app_state.gov.deposit_params.min_deposit = [{"denom": "uqck", "amount": "100"}]' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,}
 jq '.app_state.gov.deposit_params.max_deposit_period = "10s"' ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json > ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json.new && mv ./${CHAIN_DIR}/${CHAINID_0}/config/genesis.json{.new,}

x/interchainstaking/keeper/callbacks.go

@@ -73,7 +73,9 @@ func (c Callbacks) RegisterCallbacks() icqtypes.QueryCallbacks {
 		AddCallback("validator", Callback(ValidatorCallback)).
 		AddCallback("rewards", Callback(RewardsCallback)).
 		AddCallback("delegations", Callback(DelegationsCallback)).
+		AddCallback("delegations_epoch", Callback(DelegationsEpochCallback)).
 		AddCallback("delegation", Callback(DelegationCallback)).
+		AddCallback("delegation_epoch", Callback(DelegationEpochCallback)).
 		AddCallback("distributerewards", Callback(DistributeRewardsFromWithdrawAccount)).
 		AddCallback("depositinterval", Callback(DepositIntervalCallback)).
 		AddCallback("deposittx", Callback(DepositTxCallback)).
@@ -124,17 +126,25 @@ func RewardsCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Que
 
 	// decrement waitgroup as we have received back the query
 	// (initially incremented in AfterEpochEnd)
-	err = zone.DecrementWithdrawalWaitgroup(k.Logger(ctx), 1, "rewards callback")
-	if err != nil {
-		return err
+	if err = zone.DecrementWithdrawalWaitgroup(k.Logger(ctx), 1, "rewards callback"); err != nil {
+		// given that there _could_ be a backlog of message, we don't want to bail here, else they will remain undeliverable.
+		k.Logger(ctx).Error(err.Error())
 	}
 
 	k.Logger(ctx).Debug("QueryDelegationRewards callback", "wg", zone.GetWithdrawalWaitgroup(), "delegatorAddress", rewardsQuery.DelegatorAddress, "zone", query.ChainId)
 
 	return k.WithdrawDelegationRewardsForResponse(ctx, &zone, rewardsQuery.DelegatorAddress, args)
 }
 
+func DelegationsEpochCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query) error {
+	return delegationsCallback(k, ctx, args, query, true)
+}
+
 func DelegationsCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query) error {
+	return delegationsCallback(k, ctx, args, query, false)
+}
+
+func delegationsCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query, isEpoch bool) error {
 	zone, found := k.GetZone(ctx, query.GetChainId())
 	if !found {
 		return fmt.Errorf("no registered zone for chain id: %s", query.GetChainId())
@@ -152,10 +162,18 @@ func DelegationsCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes
 
 	k.Logger(ctx).Debug("Delegations callback triggered", "chain", zone.ChainId)
 
-	return k.UpdateDelegationRecordsForAddress(ctx, zone, delegationQuery.DelegatorAddr, args)
+	return k.UpdateDelegationRecordsForAddress(ctx, zone, delegationQuery.DelegatorAddr, args, isEpoch)
+}
+
+func DelegationEpochCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query) error {
+	return delegationCallback(k, ctx, args, query, true)
 }
 
 func DelegationCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query) error {
+	return delegationCallback(k, ctx, args, query, false)
+}
+
+func delegationCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.Query, isEpoch bool) error {
 	zone, found := k.GetZone(ctx, query.GetChainId())
 	if !found {
 		return fmt.Errorf("no registered zone for chain id: %s", query.GetChainId())
@@ -204,7 +222,7 @@ func DelegationCallback(k *Keeper, ctx sdk.Context, args []byte, query icqtypes.
 		return err
 	}
 
-	return k.UpdateDelegationRecordForAddress(ctx, delegation.DelegatorAddress, delegation.ValidatorAddress, sdk.NewCoin(zone.BaseDenom, val.SharesToTokens(delegation.Shares)), &zone, true)
+	return k.UpdateDelegationRecordForAddress(ctx, delegation.DelegatorAddress, delegation.ValidatorAddress, sdk.NewCoin(zone.BaseDenom, val.SharesToTokens(delegation.Shares)), &zone, true, isEpoch)
 }
 
 func PerfBalanceCallback(k *Keeper, ctx sdk.Context, response []byte, query icqtypes.Query) error {
@@ -622,9 +640,9 @@ func DelegationAccountBalanceCallback(k *Keeper, ctx sdk.Context, args []byte, q
 	}
 
 	k.Logger(ctx).Info("Received balance response for denom", "denom", coin.Denom)
-	err = zone.DecrementWithdrawalWaitgroup(k.Logger(ctx), 1, "delegationaccountbalance callback")
-	if err != nil {
-		return err
+	if err = zone.DecrementWithdrawalWaitgroup(k.Logger(ctx), 1, "delegationaccountbalance callback"); err != nil {
+		// given that there _could_ be a backlog of message, we don't want to bail here, else they will remain undeliverable.
+		k.Logger(ctx).Error(err.Error())
 	}
 
 	// set the zone amount.
@@ -645,6 +663,12 @@ func DelegationAccountBalanceCallback(k *Keeper, ctx sdk.Context, args []byte, q
 	// if token is not valid for staking, then send to withdrawal account.
 	if valid, _ := zone.ValidateCoinsForZone(sdk.NewCoins(coin), k.GetValidatorAddressesAsMap(ctx, zone.ChainId)); !valid {
 		k.Logger(ctx).Info("token is not a valid staking token, so sending to withdrawal account for disbursal", "chain", zone.ChainId, "assets", coin)
+		if zone.GetWithdrawalWaitgroup() == 0 {
+			k.Logger(ctx).Info("triggering redemption rate calc in lieu of delegation flush")
+			if err := k.TriggerRedemptionRate(ctx, &zone); err != nil {
+				return err
+			}
+		}
 		return k.SendToWithdrawal(ctx, &zone, zone.DelegationAddress, sdk.NewCoins(coin))
 	}
 
@@ -660,7 +684,8 @@ func DelegationAccountBalancesCallback(k *Keeper, ctx sdk.Context, args []byte,
 	k.cdc.MustUnmarshal(args, &result)
 
 	if err := zone.DecrementWithdrawalWaitgroup(k.Logger(ctx), 1, "delegationaccountbalances callback"); err != nil {
-		return err
+		// given that there _could_ be a backlog of message, we don't want to bail here, else they will remain undeliverable.
+		k.Logger(ctx).Error(err.Error())
 	}
 
 	addressBytes, err := addressutils.AccAddressFromBech32(zone.DelegationAddress.Address, zone.AccountPrefix)

x/interchainstaking/keeper/callbacks_test.go

@@ -2649,7 +2649,7 @@ func (suite *KeeperTestSuite) TestDelegationAccountBalancesCallbackNoWg() {
 			suite.Require().NoError(err)
 
 			err = keeper.DelegationAccountBalancesCallback(app.InterchainstakingKeeper, ctx, respbz, icqtypes.Query{ChainId: suite.chainB.ChainID, Request: reqbz})
-			suite.Require().Error(err)
+			suite.Require().NoError(err) // was previously error but we no longer fail here, just exit with nil.
 		})
 	}
 }

x/interchainstaking/keeper/delegation.go

@@ -329,7 +329,8 @@ func (k *Keeper) FlushOutstandingDelegations(ctx sdk.Context, zone *types.Zone,
 		k.Logger(ctx).Info("delegate account balance negative, or nothing to flush, setting outdated receipts")
 		k.SetReceiptsCompleted(ctx, zone.ChainId, exclusionTime, ctx.BlockTime(), delAddrBalance.Denom)
 		if zone.GetWithdrawalWaitgroup() == 0 {
-			k.Logger(ctx).Info("triggering redemption rate calc in lieu of delegation flush")
+			// we won't be sending any messages when we exit here; so if WG==0, then trigger RR update
+			k.Logger(ctx).Info("triggering redemption rate calc in lieu of delegation flush (non-positive coins)")
 			if err := k.TriggerRedemptionRate(ctx, zone); err != nil {
 				return err
 			}
@@ -352,6 +353,15 @@ func (k *Keeper) FlushOutstandingDelegations(ctx sdk.Context, zone *types.Zone,
 	if err = zone.IncrementWithdrawalWaitgroup(k.Logger(ctx), uint32(numMsgs), "sending flush messages"); err != nil {
 		return err
 	}
+
+	// if we didn't send any messages (thus no acks will happen), and WG==0, then trigger RR update
+	if numMsgs == 0 && zone.GetWithdrawalWaitgroup() == 0 {
+		k.Logger(ctx).Info("triggering redemption rate calc in lieu of delegation flush (no messages to send)")
+		if err := k.TriggerRedemptionRate(ctx, zone); err != nil {
+			return err
+		}
+	}
+
 	k.SetZone(ctx, zone)
 	return nil
 }

x/interchainstaking/keeper/delegation_test.go

@@ -217,7 +217,7 @@ func (suite *KeeperTestSuite) TestUpdateDelegation() {
 			}
 
 			for _, update := range tt.updates {
-				err := qApp.InterchainstakingKeeper.UpdateDelegationRecordForAddress(ctx, update.delegation.DelegationAddress, update.delegation.ValidatorAddress, update.delegation.Amount, &zone, update.absolute)
+				err := qApp.InterchainstakingKeeper.UpdateDelegationRecordForAddress(ctx, update.delegation.DelegationAddress, update.delegation.ValidatorAddress, update.delegation.Amount, &zone, update.absolute, false)
 				suite.NoError(err)
 			}
 

x/interchainstaking/keeper/hooks.go

@@ -101,6 +101,10 @@ func (k *Keeper) AfterEpochEnd(ctx sdk.Context, epochIdentifier string, epochNum
 			return false
 		})
 
+		if zone.GetWithdrawalWaitgroup() > 0 {
+			zone.SetWithdrawalWaitgroup(k.Logger(ctx), 0, "epoch waitgroup was unexpected > 0")
+		}
+
 		if err := k.HandleQueuedUnbondings(ctx, zone, epochNumber); err != nil {
 			// we can and need not panic here; logging the error is sufficient.
 			// an error here is not expected, but also not terminal.
@@ -130,10 +134,6 @@ func (k *Keeper) AfterEpochEnd(ctx sdk.Context, epochIdentifier string, epochNum
 			)
 		}
 
-		if zone.GetWithdrawalWaitgroup() > 0 {
-			zone.SetWithdrawalWaitgroup(k.Logger(ctx), 0, "epoch waitgroup was unexpected > 0")
-		}
-
 		// OnChanOpenAck calls SetWithdrawalAddress (see ibc_module.go)
 		k.Logger(ctx).Info(
 			"withdrawing rewards",
@@ -154,10 +154,12 @@ func (k *Keeper) AfterEpochEnd(ctx sdk.Context, epochIdentifier string, epochNum
 			bz,
 			sdk.NewInt(-1),
 			types.ModuleName,
-			"delegations",
+			"delegations_epoch",
 			0,
 		)
 
+		_ = zone.IncrementWithdrawalWaitgroup(k.Logger(ctx), 1, "delegations trigger")
+
 		balancesQuery := banktypes.QueryAllBalancesRequest{Address: zone.DelegationAddress.Address}
 		bz = k.cdc.MustMarshal(&balancesQuery)
 		k.ICQKeeper.MakeRequest(