/
batch.go
213 lines (187 loc) · 7.88 KB
/
batch.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
package keeper
import (
"encoding/binary"
"fmt"
"sort"
"strconv"
sdk "github.com/cosmos/cosmos-sdk/types"
sdkerrors "github.com/cosmos/cosmos-sdk/types/errors"
"github.com/ethereum/go-ethereum/common"
"github.com/peggyjv/gravity-bridge/module/v3/x/gravity/types"
)
// TODO: should we make this a parameter or a a call arg?
const BatchTxSize = 100
// CreateBatchTx starts the following process chain:
// - find bridged denominator for given voucher type
// - determine if a an unexecuted batch is already waiting for this token type, if so confirm the new batch would
// have a higher total fees. If not exit withtout creating a batch
// - select available transactions from the unbatched SendToEthereums sorted by fee desc
// - persist an OutgoingTx (BatchTx) object with an incrementing ID = nonce
// - emit an event
func (k Keeper) CreateBatchTx(ctx sdk.Context, contractAddress common.Address, maxElements int) *types.BatchTx {
// if there is a more profitable batch for this token type do not create a new batch
if lastBatch := k.getLastOutgoingBatchByTokenType(ctx, contractAddress); lastBatch != nil {
if lastBatch.GetFees().GTE(k.getBatchFeesByTokenType(ctx, contractAddress, maxElements)) {
return nil
}
}
var selectedStes []*types.SendToEthereum
k.iterateUnbatchedSendToEthereumsByContract(ctx, contractAddress, func(ste *types.SendToEthereum) bool {
selectedStes = append(selectedStes, ste)
k.deleteUnbatchedSendToEthereum(ctx, ste.Id, ste.Erc20Fee)
return len(selectedStes) == maxElements
})
// do not create batches that would contain no transactions, even if they are requested
if len(selectedStes) == 0 {
return nil
}
batch := &types.BatchTx{
BatchNonce: k.incrementLastOutgoingBatchNonce(ctx),
Timeout: k.getTimeoutHeight(ctx),
Transactions: selectedStes,
TokenContract: contractAddress.Hex(),
Height: uint64(ctx.BlockHeight()),
}
k.SetOutgoingTx(ctx, batch)
ctx.EventManager().EmitEvent(sdk.NewEvent(
types.EventTypeOutgoingBatch,
sdk.NewAttribute(sdk.AttributeKeyModule, types.ModuleName),
sdk.NewAttribute(types.AttributeKeyContract, k.getBridgeContractAddress(ctx)),
sdk.NewAttribute(types.AttributeKeyBridgeChainID, strconv.Itoa(int(k.getBridgeChainID(ctx)))),
sdk.NewAttribute(types.AttributeKeyOutgoingBatchID, fmt.Sprint(batch.BatchNonce)),
sdk.NewAttribute(types.AttributeKeyNonce, fmt.Sprint(batch.BatchNonce)),
))
return batch
}
// batchTxExecuted is run when the Cosmos chain detects that a batch has been executed on Ethereum
// It deletes all the transactions in the batch, then cancels all earlier batches
func (k Keeper) batchTxExecuted(ctx sdk.Context, tokenContract common.Address, nonce uint64) {
otx := k.GetOutgoingTx(ctx, types.MakeBatchTxKey(tokenContract, nonce))
if otx == nil {
k.Logger(ctx).Error("Failed to clean batches",
"token contract", tokenContract.Hex(),
"nonce", nonce)
return
}
batchTx, _ := otx.(*types.BatchTx)
k.IterateOutgoingTxsByType(ctx, types.BatchTxPrefixByte, func(key []byte, otx types.OutgoingTx) bool {
// If the iterated batches nonce is lower than the one that was just executed, cancel it
btx, ok := otx.(*types.BatchTx)
if !ok {
panic(sdkerrors.Wrapf(types.ErrInvalid, "couldn't cast to batch tx for %s", otx))
}
if (btx.BatchNonce < batchTx.BatchNonce) && (btx.TokenContract == batchTx.TokenContract) {
k.DeleteEthereumSignatures(ctx, btx.GetStoreIndex())
k.CancelBatchTx(ctx, btx)
}
return false
})
k.CompleteOutgoingTx(ctx, batchTx)
}
// getBatchFeesByTokenType gets the fees the next batch of a given token type would
// have if created. This info is both presented to relayers for the purpose of determining
// when to request batches and also used by the batch creation process to decide not to create
// a new batch
func (k Keeper) getBatchFeesByTokenType(ctx sdk.Context, tokenContractAddr common.Address, maxElements int) sdk.Int {
feeAmount := sdk.ZeroInt()
i := 0
k.iterateUnbatchedSendToEthereumsByContract(ctx, tokenContractAddr, func(tx *types.SendToEthereum) bool {
feeAmount = feeAmount.Add(tx.Erc20Fee.Amount)
i++
return i == maxElements
})
return feeAmount
}
// GetBatchFeesByTokenType gets the fees the next batch of a given token type would
// have if created. This info is both presented to relayers for the purpose of determining
// when to request batches and also used by the batch creation process to decide not to create
// a new batch
func (k Keeper) GetBatchFeesByTokenType(ctx sdk.Context, tokenContractAddr common.Address, maxElements int) sdk.Int {
feeAmount := sdk.ZeroInt()
i := 0
k.iterateUnbatchedSendToEthereumsByContract(ctx, tokenContractAddr, func(tx *types.SendToEthereum) bool {
feeAmount = feeAmount.Add(tx.Erc20Fee.Amount)
i++
return i == maxElements
})
return feeAmount
}
// CancelBatchTx releases all TX in the batch and deletes the batch
func (k Keeper) CancelBatchTx(ctx sdk.Context, batch *types.BatchTx) {
// free transactions from batch and reindex them
for _, tx := range batch.Transactions {
k.setUnbatchedSendToEthereum(ctx, tx)
}
// Delete batch since it is finished
k.DeleteOutgoingTx(ctx, batch.GetStoreIndex())
ctx.EventManager().EmitEvent(
sdk.NewEvent(
types.EventTypeOutgoingBatchCanceled,
sdk.NewAttribute(sdk.AttributeKeyModule, types.ModuleName),
sdk.NewAttribute(types.AttributeKeyContract, k.getBridgeContractAddress(ctx)),
sdk.NewAttribute(types.AttributeKeyBridgeChainID, strconv.Itoa(int(k.getBridgeChainID(ctx)))),
sdk.NewAttribute(types.AttributeKeyOutgoingBatchID, fmt.Sprint(batch.BatchNonce)),
sdk.NewAttribute(types.AttributeKeyNonce, fmt.Sprint(batch.BatchNonce)),
),
)
}
// getLastOutgoingBatchByTokenType gets the latest outgoing tx batch by token type
func (k Keeper) getLastOutgoingBatchByTokenType(ctx sdk.Context, token common.Address) *types.BatchTx {
var lastBatch *types.BatchTx = nil
lastNonce := uint64(0)
k.IterateOutgoingTxsByType(ctx, types.BatchTxPrefixByte, func(key []byte, otx types.OutgoingTx) bool {
btx, _ := otx.(*types.BatchTx)
if common.HexToAddress(btx.TokenContract) == token && btx.BatchNonce > lastNonce {
lastBatch = btx
lastNonce = btx.BatchNonce
}
return false
})
return lastBatch
}
// GetUnsignedBatchTxs returns all batches for which the specified validator has not submitted confirmations in ascending nonce order
func (k Keeper) GetUnsignedBatchTxs(ctx sdk.Context, val sdk.ValAddress) []*types.BatchTx {
var unconfirmed []*types.BatchTx
k.IterateCompletedOutgoingTxsByType(ctx, types.BatchTxPrefixByte, func(_ []byte, cotx types.OutgoingTx) bool {
sig := k.getEthereumSignature(ctx, cotx.GetStoreIndex(), val)
if len(sig) == 0 {
batch, ok := cotx.(*types.BatchTx)
if !ok {
panic(sdkerrors.Wrapf(types.ErrInvalid, "couldn't cast to batch tx for completed tx %s", cotx))
}
unconfirmed = append(unconfirmed, batch)
}
return false
})
k.IterateOutgoingTxsByType(ctx, types.BatchTxPrefixByte, func(_ []byte, otx types.OutgoingTx) bool {
sig := k.getEthereumSignature(ctx, otx.GetStoreIndex(), val)
if len(sig) == 0 {
batch, ok := otx.(*types.BatchTx)
if !ok {
panic(sdkerrors.Wrapf(types.ErrInvalid, "couldn't cast to batch tx for %s", otx))
}
unconfirmed = append(unconfirmed, batch)
}
return false
})
return orderBatchesByNonceAscending(unconfirmed)
}
func (k Keeper) incrementLastOutgoingBatchNonce(ctx sdk.Context) uint64 {
store := ctx.KVStore(k.storeKey)
bz := store.Get([]byte{types.LastOutgoingBatchNonceKey})
var id uint64 = 0
if bz != nil {
id = binary.BigEndian.Uint64(bz)
}
newId := id + 1
bz = sdk.Uint64ToBigEndian(newId)
store.Set([]byte{types.LastOutgoingBatchNonceKey}, bz)
return newId
}
// orderBatchesByNonceAscending orders the batches by their BatchNonce in ascending order
func orderBatchesByNonceAscending(batches []*types.BatchTx) []*types.BatchTx {
sort.Slice(batches, func(i, j int) bool {
return batches[i].BatchNonce < batches[j].BatchNonce
})
return batches
}