From 3ea0b7775a51fd74cd6dd562cf91531df2e78b84 Mon Sep 17 00:00:00 2001 From: Janez Podhostnik Date: Wed, 5 Aug 2026 17:02:21 +0200 Subject: [PATCH 1/2] harden fee receiver validation: never give up, surface status via /call Previously validateFeeReceivers gave up for good after 5 quick attempts, so a flaky access node at startup plus a stale config could leave the server running with an unvalidated fee set indefinitely, and the only signal was a single log line. - Retry forever: quick backoff for the first 5 attempts, then a slow one-minute poll. Treat malformed script results as retryable instead of giving up. - Re-check every 10 minutes after a definitive result, so receivers added on chain while the server is running are also detected. - Downgrade a config mismatch from a fatal exit to a logged error that is surfaced via the new fee_receiver_validation_status /call method, matching the balance_validation_status pattern. - Replace the stringly-typed validation status with a validationStatus enum shared by balance and fee validation. - Add tests for the fee validation state machine, including a concurrent access test for the race detector. --- README.md | 9 ++- api/api.go | 117 ++++++++++++++++++++++++++++++--- api/call_service.go | 37 ++++++++--- api/validate.go | 150 +++++++++++++++++++++++++++---------------- api/validate_test.go | 131 +++++++++++++++++++++++++++++++++++++ 5 files changed, 367 insertions(+), 77 deletions(-) create mode 100644 api/validate_test.go diff --git a/README.md b/README.md index 8a922f3..9db3d8b 100644 --- a/README.md +++ b/README.md @@ -375,9 +375,12 @@ config file: ``` * The canonical list is returned by `FlowFees.getFeeReceiverAddresses()` on - chain. On startup, the server validates the configured addresses against - that list and exits with a fatal error if any on-chain receiver is - missing from the config. + chain. The server validates the configured addresses against that list in + the background: retrying until an access node responds, and then + re-checking periodically so receivers added on chain while the server is + running are still detected. If any on-chain receiver is missing from the + config, the server logs an error and reports the mismatch via the + `fee_receiver_validation_status` method of the `/call` endpoint. * `data_dir: string` diff --git a/api/api.go b/api/api.go index d49f4b2..c551801 100644 --- a/api/api.go +++ b/api/api.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "fmt" "net/http" + "strings" "sync" "time" @@ -27,6 +28,7 @@ const ( callAccountPublicKeys = "account_public_keys" callBalanceValidationStatus = "balance_validation_status" callEcho = "echo" + callFeeValidationStatus = "fee_receiver_validation_status" callLatestBlock = "latest_block" callListAccounts = "list_accounts" callVerifyAddress = "verify_address" @@ -48,6 +50,7 @@ var ( callAccountPublicKeys, callBalanceValidationStatus, callEcho, + callFeeValidationStatus, callLatestBlock, callListAccounts, callVerifyAddress, @@ -96,13 +99,18 @@ type Server struct { scriptSetContract []byte validation *validation validationMu sync.RWMutex // protects validation + feeValidation *feeValidation + feeValidationMu sync.RWMutex // protects feeValidation } // Run initializes the server and starts serving Rosetta API calls. func (s *Server) Run(ctx context.Context) { s.compileScripts() s.validation = &validation{ - status: "not_started", + status: validationNotStarted, + } + s.feeValidation = &feeValidation{ + status: validationNotStarted, } go s.validateBalances(ctx) s.feeAddrs = s.Chain.Contracts.FeeAddresses() @@ -212,37 +220,95 @@ func (s *Server) setIndexedStateErr(format string, a ...interface{}) { s.mu.Unlock() s.validationMu.Lock() defer s.validationMu.Unlock() - if s.validation.status == "failure" { + if s.validation.status == validationFailure { return } s.validation = &validation{ err: msg, - status: "failure", + status: validationFailure, + } +} + +func (s *Server) getFeeValidationStatus() *feeValidation { + s.feeValidationMu.RLock() + defer s.feeValidationMu.RUnlock() + return s.feeValidation +} + +func (s *Server) setFeeValidationRetrying(format string, a ...interface{}) { + msg := fmt.Sprintf(format, a...) + log.Errorf("%s", msg) + s.feeValidationMu.Lock() + defer s.feeValidationMu.Unlock() + // We only track transient errors while we're still waiting for the first + // definitive result. Once we have one, it stays in place until the next + // definitive result replaces it. + if s.feeValidation.status == validationSuccess || s.feeValidation.status == validationFailure { + return + } + s.feeValidation = &feeValidation{ + err: msg, + status: validationInProgress, + } +} + +func (s *Server) setFeeValidationFailure(onchain []string, missing []string) { + msg := fmt.Sprintf( + "On-chain fee receiver account(s) %s are missing from the configured fee addresses: "+ + "fee deposits to them would be misclassified as transfers; add them to .contracts.fee_receivers", + strings.Join(missing, ", "), + ) + log.Errorf("%s", msg) + s.feeValidationMu.Lock() + defer s.feeValidationMu.Unlock() + s.feeValidation = &feeValidation{ + err: msg, + missing: missing, + onchain: onchain, + status: validationFailure, + } +} + +func (s *Server) setFeeValidationSuccess(onchain []string) { + s.feeValidationMu.Lock() + prev := s.feeValidation.status + s.feeValidation = &feeValidation{ + onchain: onchain, + status: validationSuccess, + } + s.feeValidationMu.Unlock() + // We only log on transitions so that the periodic re-checks don't flood + // the logs. + if prev != validationSuccess { + log.Infof( + "Validated the configured fee addresses against the on-chain fee receivers: %s", + strings.Join(onchain, ", "), + ) } } func (s *Server) setValidationProgress(accounts int, checked int) { s.validationMu.Lock() defer s.validationMu.Unlock() - if s.validation.status == "failure" || s.validation.status == "success" { + if s.validation.status == validationFailure || s.validation.status == validationSuccess { return } s.validation = &validation{ accounts: accounts, checked: checked, - status: "in_progress", + status: validationInProgress, } } func (s *Server) setValidationSuccess(accounts int) { s.validationMu.Lock() defer s.validationMu.Unlock() - if s.validation.status == "failure" { + if s.validation.status == validationFailure { return } s.validation = &validation{ accounts: accounts, - status: "success", + status: validationSuccess, } } @@ -293,9 +359,44 @@ type txnIntent struct { sender []byte } +// validationStatus enumerates the states a background validation process can +// be in. +type validationStatus int + +const ( + validationNotStarted validationStatus = iota + validationInProgress + validationSuccess + validationFailure +) + +// String returns the status in the form reported by the /call endpoint. +func (v validationStatus) String() string { + switch v { + case validationNotStarted: + return "not_started" + case validationInProgress: + return "in_progress" + case validationSuccess: + return "success" + case validationFailure: + return "failure" + default: + log.Fatalf("Unsupported validation status %d", int(v)) + panic("unreachable code") + } +} + type validation struct { accounts int checked int err string - status string + status validationStatus +} + +type feeValidation struct { + err string + missing []string + onchain []string + status validationStatus } diff --git a/api/call_service.go b/api/call_service.go index 4e3c659..e98d254 100644 --- a/api/call_service.go +++ b/api/call_service.go @@ -24,6 +24,8 @@ func (s *Server) Call(ctx context.Context, r *types.CallRequest) (*types.CallRes return s.balanceValidationStatus(ctx) case callEcho: return s.echo(r.Parameters) + case callFeeValidationStatus: + return s.feeReceiverValidationStatus() case callLatestBlock: return s.latestBlock(ctx, r.Parameters) case callListAccounts: @@ -176,40 +178,57 @@ func (s *Server) accountPublicKeys(ctx context.Context, params map[string]interf func (s *Server) balanceValidationStatus(ctx context.Context) (*types.CallResponse, *types.Error) { v := s.getValidationStatus() switch v.status { - case "failure": + case validationFailure: return &types.CallResponse{ Result: map[string]interface{}{ "error": v.err, - "status": v.status, + "status": v.status.String(), }, }, nil - case "in_progress": + case validationInProgress: return &types.CallResponse{ Result: map[string]interface{}{ "accounts": v.accounts, "checked": v.checked, - "status": v.status, + "status": v.status.String(), }, }, nil - case "not_started": + case validationNotStarted: return &types.CallResponse{ Result: map[string]interface{}{ - "status": v.status, + "status": v.status.String(), }, }, nil - case "success": + case validationSuccess: return &types.CallResponse{ Result: map[string]interface{}{ "accounts": v.accounts, - "status": v.status, + "status": v.status.String(), }, }, nil default: - log.Fatalf("Unsupported validation status %q", v.status) + log.Fatalf("Unsupported validation status %d", int(v.status)) panic("unreachable code") } } +func (s *Server) feeReceiverValidationStatus() (*types.CallResponse, *types.Error) { + v := s.getFeeValidationStatus() + result := map[string]interface{}{ + "status": v.status.String(), + } + if v.err != "" { + result["error"] = v.err + } + if v.onchain != nil { + result["fee_receivers"] = v.onchain + } + if v.missing != nil { + result["missing"] = v.missing + } + return &types.CallResponse{Result: result}, nil +} + func (s *Server) echo(params map[string]interface{}) (*types.CallResponse, *types.Error) { return &types.CallResponse{ Idempotent: true, diff --git a/api/validate.go b/api/validate.go index 443b7ef..7184781 100644 --- a/api/validate.go +++ b/api/validate.go @@ -3,79 +3,115 @@ package api import ( "context" "os" - "strings" "time" "github.com/onflow/cadence" "github.com/onflow/rosetta/log" ) -// validateFeeReceivers checks the configured fee addresses (the FlowFees -// contract account plus .contracts.fee_receivers) against the fee receiver -// accounts the FlowFees contract rotates deposits across on chain. If an -// on-chain receiver is missing from the config, fee deposits to it would be -// misclassified as ordinary transfers, so we exit with a fatal error. -// Configured addresses that are no longer on chain are fine — they may be -// needed to classify fees in historical blocks. +const ( + feeValidateQuickAttempts = 5 // short-backoff attempts before the slow poll + feeValidateSlowInterval = time.Minute // retry interval after the quick attempts + feeValidateRecheckInterval = 10 * time.Minute // re-check interval after a definitive result +) + +// validateFeeReceivers runs a background loop that checks the configured fee +// addresses (the FlowFees contract account plus .contracts.fee_receivers) +// against the fee receiver accounts the FlowFees contract rotates deposits +// across on chain. If an on-chain receiver is missing from the config, fee +// deposits to it would be misclassified as ordinary transfers, so we log an +// error and surface the failure via the fee_receiver_validation_status /call +// method. Configured addresses that are no longer on chain are fine — they +// may be needed to classify fees in historical blocks. +// +// Transient failures are retried forever, and the check re-runs periodically +// to catch receivers added on chain at runtime. func (s *Server) validateFeeReceivers(ctx context.Context) { if s.Offline { return } - const attempts = 5 - for attempt := 1; attempt <= attempts; attempt++ { - select { - case <-ctx.Done(): - return - default: - } - if attempt > 1 { - time.Sleep(time.Duration(attempt) * time.Second) - } - // Pick a client on each attempt so a retry can land on a different - // access node if the previously selected one is unavailable. - client := s.DataAccessNodes.Client() - latest, err := client.LatestBlockHeader(ctx) - if err != nil { - log.Errorf("Failed to get the latest block header to validate fee receivers: %s", err) - continue - } - resp, err := client.Execute(ctx, latest.Id, s.scriptGetFeeReceivers, nil) - if err != nil { - log.Errorf("Failed to execute the get_fee_receivers script: %s", err) - continue + attempt := 0 + for { + var delay time.Duration + if s.checkFeeReceivers(ctx) { + attempt = 0 + delay = feeValidateRecheckInterval + } else { + attempt++ + delay = time.Duration(attempt) * time.Second + if attempt >= feeValidateQuickAttempts { + delay = feeValidateSlowInterval + } } - arr, ok := resp.(cadence.Array) - if !ok { - log.Errorf("Failed to convert get_fee_receivers result to an array: got %T", resp) + if !sleepCtx(ctx, delay) { return } - onchain := []string{} - missing := []string{} - for _, val := range arr.Values { - addr, ok := val.(cadence.Address) - if !ok { - log.Errorf("Failed to convert get_fee_receivers element to an address: got %T", val) - return - } - onchain = append(onchain, addr.String()) - if !s.feeAddrs[string(addr.Bytes())] { - missing = append(missing, addr.String()) - } - } - if len(missing) > 0 { - log.Fatalf( - "On-chain fee receiver account(s) %s are missing from the configured fee addresses: "+ - "fee deposits to them would be misclassified as transfers; add them to .contracts.fee_receivers", - strings.Join(missing, ", "), + } +} + +// sleepCtx sleeps for the given duration, returning early with false if the +// context is cancelled first. +func sleepCtx(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + +// checkFeeReceivers makes a single attempt at validating the configured fee +// addresses against the on-chain fee receivers, and records the outcome in +// the server's fee validation state. It returns false if the attempt failed +// and should be retried. +func (s *Server) checkFeeReceivers(ctx context.Context) bool { + // Pick a client on each attempt so a retry can land on a different + // access node if the previously selected one is unavailable. + client := s.DataAccessNodes.Client() + latest, err := client.LatestBlockHeader(ctx) + if err != nil { + s.setFeeValidationRetrying( + "Failed to get the latest block header to validate fee receivers: %s", err, + ) + return false + } + resp, err := client.Execute(ctx, latest.Id, s.scriptGetFeeReceivers, nil) + if err != nil { + s.setFeeValidationRetrying( + "Failed to execute the get_fee_receivers script: %s", err, + ) + return false + } + arr, ok := resp.(cadence.Array) + if !ok { + s.setFeeValidationRetrying( + "Failed to convert get_fee_receivers result to an array: got %T", resp, + ) + return false + } + onchain := []string{} + missing := []string{} + for _, val := range arr.Values { + addr, ok := val.(cadence.Address) + if !ok { + s.setFeeValidationRetrying( + "Failed to convert get_fee_receivers element to an address: got %T", val, ) + return false } - log.Infof( - "Validated the configured fee addresses against the on-chain fee receivers: %s", - strings.Join(onchain, ", "), - ) - return + onchain = append(onchain, addr.String()) + if !s.feeAddrs[string(addr.Bytes())] { + missing = append(missing, addr.String()) + } + } + if len(missing) > 0 { + s.setFeeValidationFailure(onchain, missing) + } else { + s.setFeeValidationSuccess(onchain) } - log.Errorf("Giving up on fee receiver validation after %d attempts", attempts) + return true } // NOTE(tav): We exit with a fatal error if the on-chain state doesn't match diff --git a/api/validate_test.go b/api/validate_test.go new file mode 100644 index 0000000..6d86f28 --- /dev/null +++ b/api/validate_test.go @@ -0,0 +1,131 @@ +package api + +import ( + "sync" + "testing" +) + +func TestValidationStatusString(t *testing.T) { + for status, want := range map[validationStatus]string{ + validationNotStarted: "not_started", + validationInProgress: "in_progress", + validationSuccess: "success", + validationFailure: "failure", + } { + if got := status.String(); got != want { + t.Errorf("validationStatus(%d).String() = %q, want %q", int(status), got, want) + } + } +} + +func newFeeValidationServer() *Server { + return &Server{ + feeValidation: &feeValidation{ + status: validationNotStarted, + }, + } +} + +func TestFeeValidationRetrying(t *testing.T) { + s := newFeeValidationServer() + s.setFeeValidationRetrying("attempt %d failed", 1) + v := s.getFeeValidationStatus() + if v.status != validationInProgress { + t.Fatalf("status = %s, want in_progress", v.status) + } + if v.err != "attempt 1 failed" { + t.Fatalf("err = %q, want %q", v.err, "attempt 1 failed") + } +} + +func TestFeeValidationSuccess(t *testing.T) { + s := newFeeValidationServer() + onchain := []string{"912d5440f7e3769e"} + s.setFeeValidationSuccess(onchain) + v := s.getFeeValidationStatus() + if v.status != validationSuccess { + t.Fatalf("status = %s, want success", v.status) + } + if len(v.onchain) != 1 || v.onchain[0] != onchain[0] { + t.Fatalf("onchain = %v, want %v", v.onchain, onchain) + } +} + +func TestFeeValidationFailure(t *testing.T) { + s := newFeeValidationServer() + missing := []string{"e1ac6b2740d204c2"} + s.setFeeValidationFailure([]string{"912d5440f7e3769e", "e1ac6b2740d204c2"}, missing) + v := s.getFeeValidationStatus() + if v.status != validationFailure { + t.Fatalf("status = %s, want failure", v.status) + } + if len(v.missing) != 1 || v.missing[0] != missing[0] { + t.Fatalf("missing = %v, want %v", v.missing, missing) + } + if v.err == "" { + t.Fatal("err is empty, want mismatch description") + } +} + +// TestFeeValidationRetryingKeepsDefinitiveResult checks that a transient +// error during a periodic re-check does not overwrite the last definitive +// result. +func TestFeeValidationRetryingKeepsDefinitiveResult(t *testing.T) { + s := newFeeValidationServer() + + s.setFeeValidationSuccess([]string{"912d5440f7e3769e"}) + s.setFeeValidationRetrying("access node unavailable") + v := s.getFeeValidationStatus() + if v.status != validationSuccess { + t.Fatalf("status = %s, want success to be preserved", v.status) + } + + s.setFeeValidationFailure([]string{"912d5440f7e3769e"}, []string{"912d5440f7e3769e"}) + s.setFeeValidationRetrying("access node unavailable") + v = s.getFeeValidationStatus() + if v.status != validationFailure { + t.Fatalf("status = %s, want failure to be preserved", v.status) + } +} + +// TestFeeValidationConcurrentAccess exercises the fee validation state from +// multiple goroutines so the race detector can verify the locking. +func TestFeeValidationConcurrentAccess(t *testing.T) { + s := newFeeValidationServer() + wg := sync.WaitGroup{} + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 100; j++ { + s.setFeeValidationRetrying("attempt failed") + s.setFeeValidationSuccess([]string{"912d5440f7e3769e"}) + s.setFeeValidationFailure( + []string{"e1ac6b2740d204c2"}, + []string{"e1ac6b2740d204c2"}, + ) + v := s.getFeeValidationStatus() + _ = v.status.String() + _ = len(v.onchain) + _ = len(v.missing) + } + }() + } + wg.Wait() +} + +// TestFeeValidationFailureRecovery checks that a later successful check +// replaces a previous mismatch, e.g. after the on-chain receiver list +// changes. +func TestFeeValidationFailureRecovery(t *testing.T) { + s := newFeeValidationServer() + s.setFeeValidationFailure([]string{"912d5440f7e3769e"}, []string{"912d5440f7e3769e"}) + s.setFeeValidationSuccess([]string{"912d5440f7e3769e"}) + v := s.getFeeValidationStatus() + if v.status != validationSuccess { + t.Fatalf("status = %s, want success", v.status) + } + if v.err != "" || len(v.missing) != 0 { + t.Fatalf("err = %q, missing = %v, want both cleared", v.err, v.missing) + } +} From a7082580f9bd8a1a6b393287dff540cdd8385f25 Mon Sep 17 00:00:00 2001 From: Janez Podhostnik Date: Tue, 11 Aug 2026 16:46:31 +0200 Subject: [PATCH 2/2] track fee receivers on chain and validate them at the indexed tip --- README.md | 31 +++++++-- api/api.go | 42 ++++++++++++ api/construction_service.go | 6 +- api/validate.go | 72 +++++++++++++++----- api/validate_test.go | 84 ++++++++++++++++++++++++ config/config.go | 18 +++++ config/config_test.go | 22 +++++++ indexdb/indexdb.go | 127 ++++++++++++++++++++++++++++++++---- indexdb/indexdb_test.go | 113 ++++++++++++++++++++++++++++++++ state/process.go | 89 ++++++++++++++++++++++++- state/state.go | 11 ++++ 11 files changed, 580 insertions(+), 35 deletions(-) create mode 100644 indexdb/indexdb_test.go diff --git a/README.md b/README.md index 9db3d8b..8214603 100644 --- a/README.md +++ b/README.md @@ -376,11 +376,32 @@ config file: * The canonical list is returned by `FlowFees.getFeeReceiverAddresses()` on chain. The server validates the configured addresses against that list in - the background: retrying until an access node responds, and then - re-checking periodically so receivers added on chain while the server is - running are still detected. If any on-chain receiver is missing from the - config, the server logs an error and reports the mismatch via the - `fee_receiver_validation_status` method of the `/call` endpoint. + the background: the script runs at the latest indexed block (the genesis + block if nothing has been indexed yet), retrying until an access node + responds, and then re-checking periodically so receivers added on chain + while the server is running are still detected. On networks whose FlowFees + contract predates the concurrent fee collection upgrade (and therefore + doesn't define `getFeeReceiverAddresses`), the FlowFees contract account + is treated as the only receiver, and the server keeps polling so that a + later upgrade is detected. If any on-chain receiver is missing from the + fee addresses used for classification, the server logs an error and + reports the mismatch via the `fee_receiver_validation_status` method of + the `/call` endpoint. + + * The indexer also watches for `FlowFees.ChildFeeAccountsChanged` events and + stores them in the index database. The most recent such event overrides + the configured fee addresses from the block containing it onward, so receivers + added on chain are picked up automatically — no config update or restart + needed. (The configured addresses remain the base for chains where the + child fee accounts were registered without emitting the event, e.g. + testnet.) + + * Note: if your index database contains blocks indexed before the + `fee_receivers` config was available (e.g. testnet blocks at or after + height 309507846 indexed with an older version), fee deposits to the child + fee accounts in those blocks will have been misclassified as ordinary + transfers. Use `resync_from` to reindex from before the on-chain upgrade + if you need those blocks classified correctly. * `data_dir: string` diff --git a/api/api.go b/api/api.go index c551801..2b1f9ae 100644 --- a/api/api.go +++ b/api/api.go @@ -229,6 +229,48 @@ func (s *Server) setIndexedStateErr(format string, a ...interface{}) { } } +// currentFeeAddrs returns the set of fee addresses used to classify fee +// deposits at the given indexed height: the configured fee addresses, +// overridden by the most recent FlowFees.ChildFeeAccountsChanged event +// indexed at or before that height, if any. +func (s *Server) currentFeeAddrs(height uint64) map[string]bool { + children, err := s.Index.FeeReceiversAt(height) + if err != nil { + log.Errorf( + "Failed to get the indexed fee receivers at height %d, falling back to the configured fee addresses: %s", + height, err, + ) + return s.feeAddrs + } + if children == nil { + return s.feeAddrs + } + return s.Chain.Contracts.FeeAddressesWith(children) +} + +// setFeeValidationFallback records a successful validation against a FlowFees +// contract that predates the concurrent fee collection upgrade: the FlowFees +// account is the only fee receiver until the contract is upgraded. +func (s *Server) setFeeValidationFallback() { + onchain := []string{s.Chain.Contracts.FlowFees} + s.feeValidationMu.Lock() + prev := s.feeValidation.status + s.feeValidation = &feeValidation{ + onchain: onchain, + status: validationSuccess, + } + s.feeValidationMu.Unlock() + // We only log on transitions so that the periodic re-checks don't flood + // the logs. + if prev != validationSuccess { + log.Infof( + "The FlowFees contract does not define getFeeReceiverAddresses (pre concurrent fee collection); "+ + "falling back to the FlowFees account %s as the only fee receiver and continuing to poll for an upgrade", + onchain[0], + ) + } +} + func (s *Server) getFeeValidationStatus() *feeValidation { s.feeValidationMu.RLock() defer s.feeValidationMu.RUnlock() diff --git a/api/construction_service.go b/api/construction_service.go index f6bc3c6..b7ec30e 100644 --- a/api/construction_service.go +++ b/api/construction_service.go @@ -525,7 +525,11 @@ func (s *Server) ConstructionPreprocess(ctx context.Context, r *types.Constructi } // NOTE(tav): We explicitly error on transfers to a fee address so as to // simplify our event processing logic. - if s.feeAddrs[string(intent.receiver)] { + feeAddrs := s.feeAddrs + if latest := s.Index.Latest(); latest != nil { + feeAddrs = s.currentFeeAddrs(latest.Height) + } + if feeAddrs[string(intent.receiver)] { return nil, wrapErrorf( errInvalidOpsIntent, "cannot make transfers to the fee address: 0x%x", diff --git a/api/validate.go b/api/validate.go index 7184781..9a62d94 100644 --- a/api/validate.go +++ b/api/validate.go @@ -3,6 +3,7 @@ package api import ( "context" "os" + "strings" "time" "github.com/onflow/cadence" @@ -15,10 +16,11 @@ const ( feeValidateRecheckInterval = 10 * time.Minute // re-check interval after a definitive result ) -// validateFeeReceivers runs a background loop that checks the configured fee -// addresses (the FlowFees contract account plus .contracts.fee_receivers) +// validateFeeReceivers runs a background loop that checks the fee addresses +// used to classify fee deposits (the configured fee addresses, overridden by +// the most recent indexed FlowFees.ChildFeeAccountsChanged event, if any) // against the fee receiver accounts the FlowFees contract rotates deposits -// across on chain. If an on-chain receiver is missing from the config, fee +// across on chain. If an on-chain receiver is missing from that set, fee // deposits to it would be misclassified as ordinary transfers, so we log an // error and surface the failure via the fee_receiver_validation_status /call // method. Configured addresses that are no longer on chain are fine — they @@ -62,25 +64,50 @@ func sleepCtx(ctx context.Context, d time.Duration) bool { } } -// checkFeeReceivers makes a single attempt at validating the configured fee -// addresses against the on-chain fee receivers, and records the outcome in -// the server's fee validation state. It returns false if the attempt failed -// and should be retried. +// checkFeeReceivers makes a single attempt at validating the fee addresses +// used for classification against the on-chain fee receivers, and records the +// outcome in the server's fee validation state. It returns false if the +// attempt failed and should be retried. func (s *Server) checkFeeReceivers(ctx context.Context) bool { - // Pick a client on each attempt so a retry can land on a different - // access node if the previously selected one is unavailable. - client := s.DataAccessNodes.Client() - latest, err := client.LatestBlockHeader(ctx) - if err != nil { + // We validate at the latest indexed block (the genesis block if nothing + // has been indexed yet), rather than the latest block available on the + // Access API: the fee addresses only matter for the blocks the indexer is + // currently classifying, and this keeps the check consistent with the + // indexed fee receiver overrides. + latest := s.Index.Latest() + if latest == nil { + s.setFeeValidationRetrying( + "Failed to validate fee receivers: no block has been indexed yet", + ) + return false + } + // The latest indexed block may belong to a past spork while the indexer + // is catching up, so we use the access nodes of the spork containing it — + // and re-select a client on each attempt so a retry can land on a + // different access node if the previously selected one is unavailable. + spork := s.Chain.SporkFor(latest.Height) + if spork == nil { s.setFeeValidationRetrying( - "Failed to get the latest block header to validate fee receivers: %s", err, + "Failed to validate fee receivers: the latest indexed block at height %d cannot be associated with any sporks in the config", + latest.Height, ) return false } - resp, err := client.Execute(ctx, latest.Id, s.scriptGetFeeReceivers, nil) + client := spork.AccessNodes.Client() + resp, err := client.Execute(ctx, latest.Hash, s.scriptGetFeeReceivers, nil) if err != nil { + if isMissingFeeReceiverFunc(err) { + // The FlowFees contract predates the concurrent fee collection + // upgrade (onflow/flow-core-contracts#575): the FlowFees account + // is the only fee receiver until the contract is upgraded. We + // record this as a successful validation and keep polling so + // that a later upgrade is detected. + s.setFeeValidationFallback() + return true + } s.setFeeValidationRetrying( - "Failed to execute the get_fee_receivers script: %s", err, + "Failed to execute the get_fee_receivers script at the latest indexed block %x (%d): %s", + latest.Hash, latest.Height, err, ) return false } @@ -91,6 +118,7 @@ func (s *Server) checkFeeReceivers(ctx context.Context) bool { ) return false } + feeAddrs := s.currentFeeAddrs(latest.Height) onchain := []string{} missing := []string{} for _, val := range arr.Values { @@ -102,7 +130,7 @@ func (s *Server) checkFeeReceivers(ctx context.Context) bool { return false } onchain = append(onchain, addr.String()) - if !s.feeAddrs[string(addr.Bytes())] { + if !feeAddrs[string(addr.Bytes())] { missing = append(missing, addr.String()) } } @@ -114,6 +142,18 @@ func (s *Server) checkFeeReceivers(ctx context.Context) bool { return true } +// isMissingFeeReceiverFunc returns whether the script execution error +// indicates that the FlowFees contract predates the concurrent fee collection +// upgrade (onflow/flow-core-contracts#575), i.e. it does not define +// getFeeReceiverAddresses. This is a deterministic script type-checking +// failure, so retrying it would never succeed — unlike transient access node +// errors. +func isMissingFeeReceiverFunc(err error) bool { + msg := err.Error() + return strings.Contains(msg, "getFeeReceiverAddresses") && + (strings.Contains(msg, "has no member") || strings.Contains(msg, "cannot find")) +} + // NOTE(tav): We exit with a fatal error if the on-chain state doesn't match // what we expect. This assumes that we can trust the data returned to us by the // Access API servers, which may not necessarily be true. diff --git a/api/validate_test.go b/api/validate_test.go index 6d86f28..45912da 100644 --- a/api/validate_test.go +++ b/api/validate_test.go @@ -1,8 +1,12 @@ package api import ( + "errors" "sync" "testing" + + "github.com/onflow/rosetta/config" + "github.com/onflow/rosetta/indexdb" ) func TestValidationStatusString(t *testing.T) { @@ -114,6 +118,86 @@ func TestFeeValidationConcurrentAccess(t *testing.T) { wg.Wait() } +func TestIsMissingFeeReceiverFunc(t *testing.T) { + for name, tt := range map[string]struct { + err error + want bool + }{ + "missing member": { + err: errors.New("rpc error: code = InvalidArgument desc = failed to execute script: error: value of type `&FlowFees` has no member `getFeeReceiverAddresses`"), + want: true, + }, + "unavailable access node": { + err: errors.New("rpc error: code = Unavailable desc = connection refused"), + want: false, + }, + "unrelated member error": { + err: errors.New("error: value of type `&FlowToken` has no member `getBalance`"), + want: false, + }, + } { + t.Run(name, func(t *testing.T) { + if got := isMissingFeeReceiverFunc(tt.err); got != tt.want { + t.Errorf("isMissingFeeReceiverFunc(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +func TestFeeValidationFallback(t *testing.T) { + s := newFeeValidationServer() + s.Chain = &config.Chain{ + Contracts: &config.Contracts{FlowFees: "912d5440f7e3769e"}, + } + s.setFeeValidationFallback() + v := s.getFeeValidationStatus() + if v.status != validationSuccess { + t.Fatalf("status = %s, want success", v.status) + } + if len(v.onchain) != 1 || v.onchain[0] != "912d5440f7e3769e" { + t.Fatalf("onchain = %v, want [912d5440f7e3769e]", v.onchain) + } +} + +func TestCurrentFeeAddrs(t *testing.T) { + store := indexdb.New(t.TempDir()) + chain := &config.Chain{ + Contracts: &config.Contracts{ + FlowFees: "912d5440f7e3769e", + FeeReceivers: []string{"e1ac6b2740d204c2"}, + }, + } + s := &Server{ + Chain: chain, + Index: store, + feeAddrs: chain.Contracts.FeeAddresses(), + } + flowFees := []byte{0x91, 0x2d, 0x54, 0x40, 0xf7, 0xe3, 0x76, 0x9e} + configured := []byte{0xe1, 0xac, 0x6b, 0x27, 0x40, 0xd2, 0x04, 0xc2} + child := []byte{0x05, 0xcb, 0xd2, 0xfa, 0x51, 0x28, 0x04, 0x1d} + + // Without any indexed event, the configured fee addresses apply. + addrs := s.currentFeeAddrs(100) + if !addrs[string(flowFees)] || !addrs[string(configured)] || addrs[string(child)] { + t.Fatalf("currentFeeAddrs without event = %v, want the configured fee addresses", addrs) + } + + // An indexed event overrides the configured fee addresses. + if err := store.SetFeeReceivers(50, [][]byte{child}); err != nil { + t.Fatalf("SetFeeReceivers: %s", err) + } + addrs = s.currentFeeAddrs(100) + if !addrs[string(flowFees)] || !addrs[string(child)] || addrs[string(configured)] { + t.Fatalf("currentFeeAddrs with event = %v, want the FlowFees account and the event's child account", addrs) + } + + // Events after the given height do not apply. + addrs = s.currentFeeAddrs(49) + if !addrs[string(configured)] || addrs[string(child)] { + t.Fatalf("currentFeeAddrs before the event = %v, want the configured fee addresses", addrs) + } +} + // TestFeeValidationFailureRecovery checks that a later successful check // replaces a previous mismatch, e.g. after the on-chain receiver list // changes. diff --git a/config/config.go b/config/config.go index a41b59b..a7140a9 100644 --- a/config/config.go +++ b/config/config.go @@ -113,6 +113,24 @@ func (c *Contracts) FeeAddresses() map[string]bool { return addrs } +// FeeAddressesWith returns the set of accounts whose FLOW deposits represent +// transaction fees after a FlowFees.ChildFeeAccountsChanged event carrying +// the given child fee accounts: the FlowFees contract account plus the given +// accounts. The map is keyed by the raw 8-byte address string. +func (c *Contracts) FeeAddressesWith(children [][]byte) map[string]bool { + addr, err := hex.DecodeString(c.FlowFees) + if err != nil || len(addr) != 8 { + log.Fatalf("Invalid FlowFees contract address %q", c.FlowFees) + } + addrs := map[string]bool{ + string(addr): true, + } + for _, child := range children { + addrs[string(child)] = true + } + return addrs +} + // Consensus defines the metadata needed to initialize a consensus follower for // a live spork. type Consensus struct { diff --git a/config/config_test.go b/config/config_test.go index dce6fc9..d4c8a62 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -39,3 +39,25 @@ func TestFeeAddresses(t *testing.T) { require.Len(t, contracts.FeeAddresses(), 1) }) } + +func TestFeeAddressesWith(t *testing.T) { + contracts := &Contracts{ + FlowFees: "912d5440f7e3769e", + FeeReceivers: []string{"e1ac6b2740d204c2"}, + } + + t.Run("returns the FlowFees account plus the given child accounts", func(t *testing.T) { + require.Equal(t, map[string]bool{ + "\x91\x2d\x54\x40\xf7\xe3\x76\x9e": true, + "\x05\xcb\xd2\xfa\x51\x28\x04\x1d": true, + }, contracts.FeeAddressesWith([][]byte{ + {0x05, 0xcb, 0xd2, 0xfa, 0x51, 0x28, 0x04, 0x1d}, + })) + }) + + t.Run("an empty list resets to just the FlowFees account", func(t *testing.T) { + require.Equal(t, map[string]bool{ + "\x91\x2d\x54\x40\xf7\xe3\x76\x9e": true, + }, contracts.FeeAddressesWith(nil)) + }) +} diff --git a/indexdb/indexdb.go b/indexdb/indexdb.go index 292526b..adab118 100644 --- a/indexdb/indexdb.go +++ b/indexdb/indexdb.go @@ -26,23 +26,30 @@ var ( ) const ( - accountPrefix byte = 'a' - blockPrefix byte = 'b' - hash2HeightPrefix byte = 'c' - height2HashPrefix byte = 'd' - isProxyPrefix byte = 'p' + accountPrefix byte = 'a' + blockPrefix byte = 'b' + hash2HeightPrefix byte = 'c' + height2HashPrefix byte = 'd' + feeReceiversPrefix byte = 'f' + isProxyPrefix byte = 'p' ) // NOTE(tav): We store the blockchain data within Badger using the following // key/value structure: // -// accountKey a = -// blockKey b = model.IndexedBlock -// blockHash2HeightKey c = -// blockHeight2HashKey d = -// isProxyKey p = 1 -// genesis = model.BlockMeta -// latest = model.BlockMeta +// accountKey a = +// blockKey b = model.IndexedBlock +// blockHash2HeightKey c = +// blockHeight2HashKey d = +// feeReceiversKey f = +// isProxyKey p = 1 +// genesis = model.BlockMeta +// latest = model.BlockMeta +// +// The feeReceiversKey entries record the child fee account addresses carried +// by FlowFees.ChildFeeAccountsChanged events, keyed by the height of the block +// in which the event was emitted. The most recent entry at or before a height +// overrides the configured default fee addresses for that height. // AccountInfo represents all the balance changes (block height, balance) for an // account and whether it's a proxy account or not. @@ -282,6 +289,80 @@ func (s *Store) ExportAccounts(filename string) { } } +// FeeReceiversAt returns the child fee account addresses recorded by the most +// recent FlowFees.ChildFeeAccountsChanged event indexed at or before the given +// height. It returns nil if no such event has been indexed, in which case the +// configured fee addresses apply as-is. +func (s *Store) FeeReceiversAt(height uint64) ([][]byte, error) { + key := make([]byte, 9) + key[0] = feeReceiversPrefix + binary.BigEndian.PutUint64(key[1:], height) + var addrs [][]byte + err := s.db.View(func(txn *badger.Txn) error { + it := txn.NewIterator(badger.IteratorOptions{ + Reverse: true, + }) + defer it.Close() + it.Seek(key) + if !it.ValidForPrefix(key[:1]) { + return nil + } + // NOTE: We initialize the result to a non-nil empty slice so that + // callers can distinguish an event carrying an empty address list (a + // reset to no child fee accounts) from no event having been indexed. + addrs = [][]byte{} + return it.Item().Value(func(val []byte) error { + if len(val)%8 != 0 { + return fmt.Errorf( + "indexdb: found malformed fee receivers value at height %d (length %d)", + binary.BigEndian.Uint64(it.Item().Key()[1:]), len(val), + ) + } + for i := 0; i < len(val); i += 8 { + addr := make([]byte, 8) + copy(addr, val[i:i+8]) + addrs = append(addrs, addr) + } + return nil + }) + }) + if err != nil { + return nil, fmt.Errorf( + "indexdb: failed to get fee receivers at height %d: %s", height, err, + ) + } + return addrs, nil +} + +// SetFeeReceivers records the child fee account addresses carried by a +// FlowFees.ChildFeeAccountsChanged event indexed at the given height. It must +// be called before Index for the same height so that a crash in between is +// recovered by re-processing the block, which rewrites the same value. +func (s *Store) SetFeeReceivers(height uint64, addrs [][]byte) error { + key := make([]byte, 9) + key[0] = feeReceiversPrefix + binary.BigEndian.PutUint64(key[1:], height) + val := make([]byte, 0, len(addrs)*8) + for _, addr := range addrs { + if len(addr) != 8 { + return fmt.Errorf( + "indexdb: invalid fee receiver address %x: expected 8 bytes, got %d", + addr, len(addr), + ) + } + val = append(val, addr...) + } + err := s.db.Update(func(txn *badger.Txn) error { + return txn.Set(key, val) + }) + if err != nil { + return fmt.Errorf( + "indexdb: failed to set fee receivers at height %d: %s", height, err, + ) + } + return nil +} + // Genesis returns the stored genesis block metadata. func (s *Store) Genesis() *model.BlockMeta { s.mu.RLock() @@ -690,6 +771,28 @@ func (s *Store) ResetTo(base uint64) error { if err != nil { return fmt.Errorf("indexdb: failed to get proxy account keys to delete: %s", err) } + err = s.db.View(func(txn *badger.Txn) error { + it := txn.NewIterator(badger.IteratorOptions{}) + prefix := []byte{feeReceiversPrefix} + it.Seek(prefix) + for { + if !it.ValidForPrefix(prefix) { + break + } + item := it.Item() + key := item.Key() + height := binary.BigEndian.Uint64(key[1:]) + if height > base { + delKeys = append(delKeys, item.KeyCopy(nil)) + } + it.Next() + } + it.Close() + return nil + }) + if err != nil { + return fmt.Errorf("indexdb: failed to get fee receiver keys to delete: %s", err) + } last := uint64(0) err = s.db.View(func(txn *badger.Txn) error { it := txn.NewIterator(badger.IteratorOptions{}) diff --git a/indexdb/indexdb_test.go b/indexdb/indexdb_test.go new file mode 100644 index 0000000..853d3a7 --- /dev/null +++ b/indexdb/indexdb_test.go @@ -0,0 +1,113 @@ +package indexdb + +import ( + "context" + "encoding/binary" + "testing" + + "github.com/dgraph-io/badger/v3" + "github.com/stretchr/testify/require" + + "github.com/onflow/rosetta/model" +) + +func newTestStore(t *testing.T) *Store { + t.Helper() + opts := badger.DefaultOptions("").WithInMemory(true).WithLogger(nil) + db, err := badger.Open(opts) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close() }) + return &Store{db: db} +} + +func addr(b byte) []byte { + return []byte{b, b, b, b, b, b, b, b} +} + +func TestFeeReceiversAt(t *testing.T) { + s := newTestStore(t) + + // No events indexed yet. + got, err := s.FeeReceiversAt(100) + require.NoError(t, err) + require.Nil(t, got) + + require.NoError(t, s.SetFeeReceivers(100, [][]byte{addr(1), addr(2)})) + require.NoError(t, s.SetFeeReceivers(200, [][]byte{addr(3)})) + + // The most recent event at or before the given height applies. + got, err = s.FeeReceiversAt(99) + require.NoError(t, err) + require.Nil(t, got) + + got, err = s.FeeReceiversAt(100) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(1), addr(2)}, got) + + got, err = s.FeeReceiversAt(150) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(1), addr(2)}, got) + + got, err = s.FeeReceiversAt(200) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(3)}, got) + + got, err = s.FeeReceiversAt(1000) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(3)}, got) +} + +func TestSetFeeReceiversEmpty(t *testing.T) { + s := newTestStore(t) + + // An event carrying an empty address list resets the child fee accounts. + require.NoError(t, s.SetFeeReceivers(100, [][]byte{addr(1)})) + require.NoError(t, s.SetFeeReceivers(200, [][]byte{})) + + got, err := s.FeeReceiversAt(200) + require.NoError(t, err) + require.NotNil(t, got) + require.Empty(t, got) +} + +func TestSetFeeReceiversInvalidAddress(t *testing.T) { + s := newTestStore(t) + err := s.SetFeeReceivers(100, [][]byte{[]byte("short")}) + require.Error(t, err) +} + +func TestResetToDeletesFeeReceivers(t *testing.T) { + s := newTestStore(t) + require.NoError(t, s.SetGenesis(testBlockMeta(10))) + + ctx := context.Background() + for height := uint64(11); height <= 200; height++ { + if height == 50 { + require.NoError(t, s.SetFeeReceivers(50, [][]byte{addr(1)})) + } + if height == 150 { + require.NoError(t, s.SetFeeReceivers(150, [][]byte{addr(2)})) + } + err := s.Index(ctx, height, testBlockMeta(height).Hash, &model.IndexedBlock{}) + require.NoError(t, err) + } + + require.NoError(t, s.ResetTo(100)) + + got, err := s.FeeReceiversAt(100) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(1)}, got) + + got, err = s.FeeReceiversAt(200) + require.NoError(t, err) + require.Equal(t, [][]byte{addr(1)}, got) +} + +func testBlockMeta(height uint64) *model.BlockMeta { + hash := make([]byte, 8) + binary.BigEndian.PutUint64(hash, height) + return &model.BlockMeta{ + Hash: hash, + Height: height, + } +} diff --git a/state/process.go b/state/process.go index c442345..0077b0f 100644 --- a/state/process.go +++ b/state/process.go @@ -418,6 +418,17 @@ outer: data := &model.IndexedBlock{ Timestamp: uint64(block.Timestamp.AsTime().UnixNano()), } + // feeReceivers tracks the child fee account addresses carried by a + // FlowFees.ChildFeeAccountsChanged event within this block, if any. If + // several events occur within the same block, the last one wins — each + // event carries the complete list of child fee accounts. + var feeReceivers [][]byte + // NOTE: We classify fee deposits against a block-scoped copy of + // the fee addresses so that a retry of the block (continue outer) + // reclassifies every transaction against the set as of the start of + // the block. The indexer's set is only updated once the block has been + // successfully indexed. + feeAddrs := i.feeAddrs newAccounts := map[string]bool{} newCounter := 0 transfers := 0 @@ -648,6 +659,58 @@ outer: if i.isProxy(addr[:], newAccounts) { proxyDeposits[string(addr[:])] += amount } + case i.typFeeAcctsChanged: + // NOTE: The event carries the complete list of child fee + // accounts. We update our set of fee addresses here, in the + // first event loop, so that the fee deposits of this and all + // subsequent transactions are classified with the updated set + // — fee deduction runs after the transaction body that emits + // the event. + event, err := decodeEvent("FlowFees.ChildFeeAccountsChanged", evt, hash, height) + if err != nil { + skipCache = true + continue outer + } + fields := event.FieldsMappedByName() + if len(fields) != 1 { + log.Errorf( + "Found FlowFees.ChildFeeAccountsChanged event with %d fields in transaction %x in block %x at height %d", + len(fields), txnHash, hash, height, + ) + skipCache = true + continue outer + } + // 'addresses' field + arr, ok := cadence.SearchFieldByName( + event, + "addresses", + ).(cadence.Array) + if !ok { + log.Errorf( + "Unable to load addresses from FlowFees.ChildFeeAccountsChanged event in transaction %x in block %x at height %d", + txnHash, hash, height, + ) + skipCache = true + continue outer + } + feeReceivers = [][]byte{} + for _, val := range arr.Values { + addr, ok := val.(cadence.Address) + if !ok { + log.Errorf( + "Unable to convert FlowFees.ChildFeeAccountsChanged element to an address (got %T) in transaction %x in block %x at height %d", + val, txnHash, hash, height, + ) + skipCache = true + continue outer + } + feeReceivers = append(feeReceivers, addr[:]) + } + feeAddrs = i.Chain.Contracts.FeeAddressesWith(feeReceivers) + log.Infof( + "Indexed FlowFees.ChildFeeAccountsChanged event in block %x at height %d: fee receivers are now %x", + hash, height, feeReceivers, + ) case i.typProxyTransferred: // NOTE(tav): For all proxy accounts originated by us, // we will only ever make transfers once we've found the @@ -803,7 +866,7 @@ outer: Receiver: receiver[:], Type: model.TransferType_DEPOSIT, }) - if i.feeAddrs[string(receiver[:])] { + if feeAddrs[string(receiver[:])] { // NOTE(tav): When the deposit is to the fee // address, just increment the fee amount. fees += amount @@ -1042,6 +1105,27 @@ outer: } } } + if feeReceivers != nil { + // NOTE: We store the fee receiver update before indexing the + // block so that a crash in between is recovered by re-processing the + // block, which rewrites the same value. + for { + select { + case <-ctx.Done(): + return + default: + } + err = i.Store.SetFeeReceivers(height, feeReceivers) + if err == nil { + break + } + log.Errorf( + "Failed to store fee receivers from block %x at height %d: %s", + hash, height, err, + ) + time.Sleep(10 * time.Millisecond) + } + } for { select { case <-ctx.Done(): @@ -1069,6 +1153,9 @@ outer: for acct, isProxy := range newAccounts { i.accts[acct] = isProxy } + if feeReceivers != nil { + i.feeAddrs = feeAddrs + } i.mu.Lock() i.lastIndexed.Hash = hash i.lastIndexed.Height = height diff --git a/state/state.go b/state/state.go index 9e9e92d..e02bde5 100644 --- a/state/state.go +++ b/state/state.go @@ -64,6 +64,7 @@ type Indexer struct { root *stateSnapshot sealedResults map[string]string synced bool + typFeeAcctsChanged string typProxyCreated string typProxyDeposited string typProxyTransferred string @@ -542,11 +543,21 @@ func (i *Indexer) initState() { i.accts[string(acct[:])] = isProxy } i.feeAddrs = i.Chain.Contracts.FeeAddresses() + // The set of fee receivers recorded by the most recently indexed + // FlowFees.ChildFeeAccountsChanged event (at or before the last indexed + // height) overrides the configured default, so that fee receivers added on + // chain are picked up without a config update or a restart. + if children, err := i.Store.FeeReceiversAt(i.lastIndexed.Height); err != nil { + log.Fatalf("Failed to load fee receivers from the index database: %s", err) + } else if children != nil { + i.feeAddrs = i.Chain.Contracts.FeeAddressesWith(children) + } i.originators = map[string]bool{} for _, addr := range i.Chain.Originators { i.originators[string(addr)] = true } i.sealedResults = map[string]string{} + i.typFeeAcctsChanged = fmt.Sprintf("A.%s.FlowFees.ChildFeeAccountsChanged", i.Chain.Contracts.FlowFees) i.typProxyCreated = fmt.Sprintf("A.%s.FlowColdStorageProxy.Created", i.Chain.Contracts.FlowColdStorageProxy) i.typProxyDeposited = fmt.Sprintf("A.%s.FlowColdStorageProxy.Deposited", i.Chain.Contracts.FlowColdStorageProxy) i.typProxyTransferred = fmt.Sprintf("A.%s.FlowColdStorageProxy.Transferred", i.Chain.Contracts.FlowColdStorageProxy)