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