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

Author: janezpodhostnik

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`
 

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()

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

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.

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.

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 {

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))
+	})
+}