11package keystore
22
33import (
4+ "context"
45 "fmt"
56 "sync"
67
8+ "github.com/onflow/flow-evm-gateway/config"
79 flowsdk "github.com/onflow/flow-go-sdk"
10+ "github.com/onflow/flow-go-sdk/access"
811 "github.com/onflow/flow-go/model/flow"
12+ "github.com/rs/zerolog"
13+ "google.golang.org/grpc/codes"
14+ "google.golang.org/grpc/status"
915)
1016
1117var ErrNoKeysAvailable = fmt .Errorf ("no signing keys available" )
1218
1319const accountKeyBlockExpiration = flow .DefaultTransactionExpiry
1420
1521type KeyLock interface {
22+ // This method is intended for the happy path of valid EVM transactions.
23+ // The event subscriber module only subscribes to EVM-related events:
24+ // - `EVM.TransactionExecuted`
25+ // - `EVM.BlockExecuted`
26+ //
27+ // Valid EVM transactions do emit `EVM.TransactionExecuted` events, so we
28+ // release the account key that was used by the Flow tx which emitted
29+ // the above EVM event.
1630 NotifyTransaction (txID flowsdk.Identifier )
17- NotifyBlock (blockHeight uint64 )
31+ // This method is intended for the unhappy path of invalid EVM transactions.
32+ // For each new Flow block, we check the result status of all included Flow
33+ // transactions, and we release the account keys which they used. This also
34+ // handles the release of expired transactions, that weren't even included
35+ // in a Flow block.
36+ NotifyBlock (blockHeader flowsdk.BlockHeader )
1837}
1938
2039type KeyStore struct {
40+ client access.Client
41+ config config.Config
2142 availableKeys chan * AccountKey
2243 usedKeys map [flowsdk.Identifier ]* AccountKey
2344 size int
2445 keyMu sync.Mutex
46+ blockChan chan flowsdk.BlockHeader
47+ logger zerolog.Logger
48+
49+ // Signal channel used to prevent blocking writes
50+ // on `blockChan` when the node is shutting down.
51+ done chan struct {}
2552}
2653
2754var _ KeyLock = (* KeyStore )(nil )
2855
29- func New (keys []* AccountKey ) * KeyStore {
56+ func New (
57+ ctx context.Context ,
58+ keys []* AccountKey ,
59+ client access.Client ,
60+ config config.Config ,
61+ logger zerolog.Logger ,
62+ ) * KeyStore {
63+ totalKeys := len (keys )
64+
3065 ks := & KeyStore {
31- usedKeys : map [flowsdk.Identifier ]* AccountKey {},
66+ client : client ,
67+ config : config ,
68+ availableKeys : make (chan * AccountKey , totalKeys ),
69+ usedKeys : map [flowsdk.Identifier ]* AccountKey {},
70+ size : totalKeys ,
71+ // `KeyStore.NotifyBlock` is called for each new Flow block,
72+ // so we use a buffered channel to write the new block headers
73+ // to the `blockChan`, and read them through `processLockedKeys`.
74+ blockChan : make (chan flowsdk.BlockHeader , 200 ),
75+ logger : logger ,
76+ done : make (chan struct {}),
3277 }
3378
34- availableKeys := make (chan * AccountKey , len (keys ))
3579 for _ , key := range keys {
3680 key .ks = ks
37- availableKeys <- key
81+ ks .availableKeys <- key
82+ }
83+
84+ // For cases where the EVM Gateway is run in an index-mode,
85+ // there is no need to release any keys, since transaction
86+ // submission is not allowed.
87+ if ! ks .config .IndexOnly {
88+ go ks .processLockedKeys (ctx )
3889 }
39- ks .size = len (keys )
40- ks .availableKeys = availableKeys
4190
4291 return ks
4392}
@@ -47,6 +96,11 @@ func (k *KeyStore) AvailableKeys() int {
4796 return len (k .availableKeys )
4897}
4998
99+ // HasKeysInUse returns whether any of the keys are currently being used.
100+ func (k * KeyStore ) HasKeysInUse () bool {
101+ return k .AvailableKeys () != k .size
102+ }
103+
50104// Take reserves a key for use in a transaction.
51105func (k * KeyStore ) Take () (* AccountKey , error ) {
52106 select {
@@ -63,22 +117,50 @@ func (k *KeyStore) Take() (*AccountKey, error) {
63117
64118// NotifyTransaction unlocks a key after use and puts it back into the pool.
65119func (k * KeyStore ) NotifyTransaction (txID flowsdk.Identifier ) {
120+ // For cases where the EVM Gateway is run in an index-mode,
121+ // there is no need to release any keys, since transaction
122+ // submission is not allowed. We return early here, to avoid
123+ // any unnecessary steps such as lock acquisition and unlocking
124+ // keys.
125+ if k .config .IndexOnly {
126+ return
127+ }
128+
66129 k .keyMu .Lock ()
67130 defer k .keyMu .Unlock ()
68131
69132 k .unsafeUnlockKey (txID )
70133}
71134
72- // NotifyBlock is called to notify the KeyStore of a new ingested block height .
135+ // NotifyBlock is called to notify the KeyStore of a newly ingested block.
73136// Pending transactions older than a threshold number of blocks are removed.
74- func (k * KeyStore ) NotifyBlock (blockHeight uint64 ) {
75- k .keyMu .Lock ()
76- defer k .keyMu .Unlock ()
137+ func (k * KeyStore ) NotifyBlock (blockHeader flowsdk.BlockHeader ) {
138+ // For cases where the EVM Gateway is run in an index-mode,
139+ // there is no need to release any keys, since transaction
140+ // submission is not allowed. We return early here, to avoid
141+ // blocking forever on writes to `k.blockChan`, because the
142+ // `k.processLockedKeys()` function won't perform any reads
143+ // from `k.blockChan`.
144+ if k .config .IndexOnly {
145+ return
146+ }
77147
78- for txID , key := range k .usedKeys {
79- if blockHeight - key .lastLockedBlock .Load () >= accountKeyBlockExpiration {
80- k .unsafeUnlockKey (txID )
81- }
148+ select {
149+ case <- k .done :
150+ k .logger .Warn ().Msg (
151+ "received `NotifyBlock` while the server is shutting down" ,
152+ )
153+ case k .blockChan <- blockHeader :
154+ k .logger .Info ().Msgf (
155+ "received `NotifyBlock` for block with ID: %s" ,
156+ blockHeader .ID ,
157+ )
158+ default :
159+ // In this case, we only release the account keys which were last
160+ // locked more than or equal to `accountKeyBlockExpiration` blocks
161+ // in the past, in order to avoid slowing down the EVM event
162+ // ingestion engine.
163+ k .releasekeys (blockHeader .Height , nil )
82164 }
83165}
84166
@@ -106,3 +188,58 @@ func (k *KeyStore) setLockMetadata(
106188 defer k .keyMu .Unlock ()
107189 k .usedKeys [txID ] = key
108190}
191+
192+ // processLockedKeys reads from the `blockChan` channel, and for each new
193+ // Flow block, it fetches the transaction results of the given block and
194+ // releases the account keys associated with those transactions.
195+ func (k * KeyStore ) processLockedKeys (ctx context.Context ) {
196+ for {
197+ select {
198+ case <- ctx .Done ():
199+ close (k .done )
200+ return
201+ case blockHeader := <- k .blockChan :
202+ // Optimization to avoid AN calls when no signing keys have
203+ // been used. For example, when back-filling the EVM GW state,
204+ // we don't care about releasing signing keys.
205+ if ! k .HasKeysInUse () {
206+ continue
207+ }
208+
209+ var txResults []* flowsdk.TransactionResult
210+ var err error
211+ if k .config .COATxLookupEnabled {
212+ txResults , err = k .client .GetTransactionResultsByBlockID (ctx , blockHeader .ID )
213+ if err != nil && status .Code (err ) != codes .Canceled {
214+ k .logger .Error ().Err (err ).Msgf (
215+ "failed to get transaction results for block ID: %s" ,
216+ blockHeader .ID .Hex (),
217+ )
218+ continue
219+ }
220+ }
221+
222+ k .releasekeys (blockHeader .Height , txResults )
223+ }
224+ }
225+ }
226+
227+ // releasekeys accepts a block height and a slice of `TransactionResult`
228+ // objects and releases the account keys used for signing the given
229+ // transactions.
230+ // It also releases the account keys which were last locked more than
231+ // or equal to `accountKeyBlockExpiration` blocks in the past.
232+ func (k * KeyStore ) releasekeys (blockHeight uint64 , txResults []* flowsdk.TransactionResult ) {
233+ k .keyMu .Lock ()
234+ defer k .keyMu .Unlock ()
235+
236+ for _ , txResult := range txResults {
237+ k .unsafeUnlockKey (txResult .TransactionID )
238+ }
239+
240+ for txID , key := range k .usedKeys {
241+ if blockHeight - key .lastLockedBlock .Load () >= accountKeyBlockExpiration {
242+ k .unsafeUnlockKey (txID )
243+ }
244+ }
245+ }
0 commit comments