Track fee receivers on chain and validate them at the indexed tip

README.md

@@ -375,9 +375,33 @@ 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: 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`
 

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,137 @@ 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,
+	}
+}
+
+// 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()
+	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 +401,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
 }

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,

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",