diff --git a/rpc/acl_boundary_test.go b/rpc/acl_boundary_test.go new file mode 100644 index 00000000..714224ef --- /dev/null +++ b/rpc/acl_boundary_test.go @@ -0,0 +1,131 @@ +package rpc + +import ( + "context" + "encoding/json" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Conflux-Chain/confura/util/acl" + "github.com/Conflux-Chain/confura/util/rate" + "github.com/Conflux-Chain/confura/util/rpc/handlers" + "github.com/Conflux-Chain/confura/util/rpc/middlewares" + "github.com/ethereum/go-ethereum/common" + rpc "github.com/openweb3/go-rpc-provider" + wt "github.com/openweb3/web3go/types" + "github.com/stretchr/testify/require" +) + +type ACLBoundaryAPI struct{} + +func (*ACLBoundaryAPI) GetLogs(context.Context, wt.FilterQuery) ([]wt.Log, error) { + return []wt.Log{{Address: common.HexToAddress("0x2222222222222222222222222222222222222222")}}, nil +} +func (*ACLBoundaryAPI) NewFilter(context.Context, wt.FilterQuery) (string, error) { return "0x1", nil } +func (*ACLBoundaryAPI) GetFilterLogs(context.Context, rpc.ID) ([]wt.Log, error) { + return []wt.Log{{Address: common.HexToAddress("0x1111111111111111111111111111111111111111")}}, nil +} +func (api *ACLBoundaryAPI) GetFilterChanges(ctx context.Context, id rpc.ID) ([]wt.Log, error) { + return api.GetFilterLogs(ctx, id) +} +func (*ACLBoundaryAPI) Echo(ctx context.Context) (string, error) { + id, _ := handlers.GetAuthIdFromContext(ctx) + return id, nil +} +func (*ACLBoundaryAPI) Logs(ctx context.Context, f wt.FilterQuery) (*rpc.Subscription, error) { + n, _ := rpc.NotifierFromContext(ctx) + return n.CreateSubscription(), nil +} + +func TestContractACLHTTPBatchAndWebSocket(t *testing.T) { + // Same production registration order; no config loading, external services, or real credentials. + rpc.HookHandleCallMsg(middlewares.Recover) + rpc.HookHandleCallMsg(middlewares.AntiInjection) + rpc.HookHandleCallMsg(middlewares.Authenticate) + rpc.HookHandleCallMsg(middlewares.Allowlists) + rpc.HookHandleCallMsg(middlewares.DailyMaxReqRateLimit) + rpc.HookHandleCallMsg(middlewares.QpsRateLimit) + rpc.HookHandleCallMsg(middlewares.Metrics) + rpc.HookHandleCallMsg(middlewares.Log) + rpc.HookHandleCallMsg(clientMiddleware) + rpc.HookHandleCallMsg(middlewares.UniformError) + rpc.HookHandleCallMsg(rpc.PreventMessagesWithouID) + const key = "FixtureProvisionedSynthetic0123456789" + loader := rate.NewKeyLoader(func(f *rate.KeysetFilter) ([]*rate.KeyInfo, error) { + if len(f.KeySet) == 0 || f.KeySet[0] == key { + return []*rate.KeyInfo{{Key: key, SID: 1, AclID: 1}}, nil + } + return nil, nil + }) + r := rate.NewRegistry(loader, acl.NewEthValidator) + s := rate.NewStrategy(1, "default") + s.LimitOptions["rpc_all_qps"] = rate.NewTokenBucketOption(10000, 10000) + s.LimitOptions["rpc_all_daily"] = rate.FixedWindowOption{Interval: time.Hour, Quota: 10000} + config := &rate.Config{Strategies: map[uint32]*rate.Strategy{1: s}, AllowLists: map[uint32]*acl.AllowList{1: {ID: 1, Name: "default", ContractAddresses: []string{"0x1111111111111111111111111111111111111111"}}}} + go r.AutoReload(time.Hour, func() (*rate.Config, error) { return config, nil }) + require.Eventually(t, func() bool { + return len(r.Strategies()) == 1 && r.Allow(acl.Context{Context: context.Background(), RpcMethod: "eth_getLogs", ExtractRpcParams: func() ([]interface{}, error) { return []interface{}{wt.FilterQuery{}}, nil }}) != nil + }, time.Second, time.Millisecond) + h := rpc.NewServer() + defer h.Stop() + require.NoError(t, h.RegisterName("eth", &ACLBoundaryAPI{})) + httpH := httpMiddleware("eth", r, nil)(h) + request := func(path, token, body string) string { + q := httptest.NewRequest("POST", "http://fixture.invalid"+path, strings.NewReader(body)) + q.Header.Set("Content-Type", "application/json") + q.Header.Set("Access-Token", token) + w := httptest.NewRecorder() + httpH.ServeHTTP(w, q) + return w.Body.String() + } + for _, tc := range []struct{ name, path, token, body, want string }{ + {"guest", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, `"result":""`}, + {"unknown", "/", "FixtureUnknownSynthetic0123456789", `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, "invalid api key"}, + {"provisioned", "/", key, `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, key}, + {"path beats header", "/FixtureUnknownSynthetic0123456789", key, `{"jsonrpc":"2.0","id":1,"method":"eth_echo","params":[]}`, "invalid api key"}, + {"deny getLogs", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[{"address":"0x2222222222222222222222222222222222222222"}]}`, "invalid contract address"}, + {"malformed params", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[9]}`, "bad RPC parameters"}, + {"filter denied", "/", "", `{"jsonrpc":"2.0","id":1,"method":"eth_newFilter","params":[{"address":"0x2222222222222222222222222222222222222222"}]}`, "invalid contract address"}, + {"batch", "/", "", `[{"jsonrpc":"2.0","id":1,"method":"eth_getLogs","params":[{}]},{"jsonrpc":"2.0","id":2,"method":"eth_newFilter","params":[{}]}]`, "invalid contract address"}, + {"notification", "/", "", `{"jsonrpc":"2.0","method":"eth_echo","params":[]}`, "invalid request"}, + } { + t.Run(tc.name, func(t *testing.T) { + out := request(tc.path, tc.token, tc.body) + require.Contains(t, out, tc.want) + if tc.name == "batch" { + require.NotContains(t, out, `"result"`) + } + }) + } + t.Run("allowed filter remains readable", func(t *testing.T) { + created := request("/", key, `{"jsonrpc":"2.0","id":1,"method":"eth_newFilter","params":[{"address":"0x1111111111111111111111111111111111111111"}]}`) + var response struct { + Result string `json:"result"` + Error json.RawMessage `json:"error"` + } + require.NoError(t, json.Unmarshal([]byte(created), &response)) + require.Empty(t, response.Error, created) + require.Equal(t, "0x1", response.Result) + for _, method := range []string{"eth_getFilterLogs", "eth_getFilterChanges"} { + out := request("/", key, `{"jsonrpc":"2.0","id":2,"method":"`+method+`","params":["`+response.Result+`"]}`) + require.Contains(t, out, `"result":[{"address":"0x1111111111111111111111111111111111111111"`, out) + require.NotContains(t, out, `"error"`, out) + } + }) + // Real WebSocket upgrade and provider subscription dispatch, with identical HTTP context injector. + ws := httptest.NewServer(httpMiddleware("eth", r, nil)(h.WebsocketHandler([]string{"*"}))) + defer ws.Close() + conn, err := rpc.Dial("ws" + strings.TrimPrefix(ws.URL, "http")) + require.NoError(t, err) + defer conn.Close() + logs := make(chan json.RawMessage, 1) + sub, err := conn.Subscribe(context.Background(), "eth", logs, "logs", map[string]interface{}{"address": "0x2222222222222222222222222222222222222222"}) + require.ErrorContains(t, err, "invalid contract address") + require.Nil(t, sub) + sub, err = conn.Subscribe(context.Background(), "eth", logs, "logs", map[string]interface{}{"address": "0x1111111111111111111111111111111111111111"}) + require.NoError(t, err) + sub.Unsubscribe() + +} diff --git a/util/acl/validator.go b/util/acl/validator.go index ada92d7c..6a72e4f2 100644 --- a/util/acl/validator.go +++ b/util/acl/validator.go @@ -2,6 +2,7 @@ package acl import ( "context" + "encoding/json" "errors" "regexp" "strings" @@ -28,6 +29,7 @@ type Context struct { context.Context RpcMethod string + RawParams json.RawMessage ExtractRpcParams func() ([]interface{}, error) } @@ -50,9 +52,9 @@ type validatorBase struct { *AllowList // wildchar pattern matching - allowMethodRules []string // allow methods - disallowMethodRules []string // disallow methods - originRules []string // request origins + allowMethodRules []*regexp.Regexp // allow methods + disallowMethodRules []*regexp.Regexp // disallow methods + originRules []*regexp.Regexp // request origins // contract addresses mapset contractAddressRules map[string]bool @@ -62,25 +64,22 @@ type validatorBase struct { contractAddressParsers map[string]contractAddressParser } -func newValidatorBase(al *AllowList) *validatorBase { - var originRules []string - for _, r := range al.Origins { - regp := util.WildCardToRegexp(strings.ToLower(r)) - originRules = append(originRules, regp) - } - - var allowMethodRules []string - for _, r := range al.AllowMethods { - regp := util.WildCardToRegexp(r) - allowMethodRules = append(allowMethodRules, regp) - } - - var disallowMethodRules []string - for _, r := range al.DisallowMethods { - regp := util.WildCardToRegexp(r) - disallowMethodRules = append(disallowMethodRules, regp) +func compileWildcardRules(patterns []string, lowercase bool) []*regexp.Regexp { + rules := make([]*regexp.Regexp, 0, len(patterns)) + for _, pattern := range patterns { + if lowercase { + pattern = strings.ToLower(pattern) + } + // Invalid patterns never matched in the per-request regexp path. Skip + // them, but retain the original lists for the empty-policy checks below. + if rule, err := regexp.Compile(util.WildCardToRegexp(pattern)); err == nil { + rules = append(rules, rule) + } } + return rules +} +func newValidatorBase(al *AllowList) *validatorBase { contractAddrRules := make(map[string]bool) for _, r := range al.ContractAddresses { contractAddrRules[strings.ToLower(r)] = true @@ -88,9 +87,9 @@ func newValidatorBase(al *AllowList) *validatorBase { return &validatorBase{ AllowList: al, - originRules: originRules, - allowMethodRules: allowMethodRules, - disallowMethodRules: disallowMethodRules, + originRules: compileWildcardRules(al.Origins, true), + allowMethodRules: compileWildcardRules(al.AllowMethods, false), + disallowMethodRules: compileWildcardRules(al.DisallowMethods, false), contractAddressRules: contractAddrRules, } } @@ -129,8 +128,7 @@ func (v *validatorBase) validateOrigin(ctx Context) error { reqOrigin = strings.ToLower(reqOrigin) for _, r := range v.originRules { - matched, err := regexp.MatchString(r, reqOrigin) - if err == nil && matched { + if r.MatchString(reqOrigin) { return nil } } @@ -167,8 +165,7 @@ func (v *validatorBase) validateAllowMethods(ctx Context) error { } for _, r := range v.allowMethodRules { - matched, err := regexp.MatchString(r, ctx.RpcMethod) - if err == nil && matched { + if r.MatchString(ctx.RpcMethod) { return nil } } @@ -178,8 +175,7 @@ func (v *validatorBase) validateAllowMethods(ctx Context) error { } for _, r := range v.disallowMethodRules { - matched, err := regexp.MatchString(r, ctx.RpcMethod) - if err == nil && matched { + if r.MatchString(ctx.RpcMethod) { return errBadRpcMethod } } @@ -194,8 +190,36 @@ func (v *validatorBase) validateContractAddresses(ctx Context) error { return nil } - if ctx.ExtractRpcParams == nil { - return nil + // Check filter criteria at creation. Filter-ID reads retain their existing + // access semantics; they do not carry addresses to validate here. + if ctx.RpcMethod == "eth_subscribe" || ctx.RpcMethod == "cfx_subscribe" { + var args []json.RawMessage + var kind string + if json.Unmarshal(ctx.RawParams, &args) != nil || len(args) == 0 || json.Unmarshal(args[0], &kind) != nil { + return errBadRpcParams + } + if kind != "logs" { + return nil + } + if len(args) != 2 { + return errBadRpcParams + } + raw := args[1] + if ctx.RpcMethod == "eth_subscribe" { + ctx.RpcMethod = "eth_getLogs" + ctx.ExtractRpcParams = func() ([]interface{}, error) { + var f web3Types.FilterQuery + err := json.Unmarshal(raw, &f) + return []interface{}{f}, err + } + } else { + ctx.RpcMethod = "cfx_getLogs" + ctx.ExtractRpcParams = func() ([]interface{}, error) { + var f cfxTypes.LogFilter + err := json.Unmarshal(raw, &f) + return []interface{}{f}, err + } + } } parseContractAddresses, ok := v.contractAddressParsers[ctx.RpcMethod] @@ -203,6 +227,9 @@ func (v *validatorBase) validateContractAddresses(ctx Context) error { return nil } + if ctx.ExtractRpcParams == nil { + return errBadRpcParams + } inputParams, err := ctx.ExtractRpcParams() if err != nil { return errBadRpcParams @@ -242,6 +269,7 @@ func NewEthValidator(al *AllowList) Validator { "eth_call": v.parseCallRequest, "eth_estimateGas": v.parseCallRequest, "eth_getLogs": v.parseFilterQuery, + "eth_newFilter": v.parseFilterQuery, "eth_scanLogs": parseContractAddressProvider, "eth_scanLogsWithPivotAssumption": parseContractAddressProvider, "eth_getBalance": v.parseAddr, @@ -266,7 +294,7 @@ func (v *EthValidator) parseCallRequest(params []interface{}) (res []string, ok } cr, ok := params[0].(web3Types.CallRequest) - if ok { + if ok && cr.To != nil { res = append(res, cr.To.String()) } @@ -329,6 +357,7 @@ func NewCfxValidator(al *AllowList) Validator { "cfx_call": v.parseCallRequest, "cfx_estimateGasAndCollateral": v.parseCallRequest, "cfx_getLogs": v.parseLogFilter, + "cfx_newFilter": v.parseLogFilter, "cfx_scanLogs": parseContractAddressProvider, "cfx_scanLogsWithPivotAssumption": parseContractAddressProvider, "cfx_getBalance": v.parseAddr, @@ -361,7 +390,7 @@ func (v *CfxValidator) parseCallRequest(params []interface{}) (res []string, ok } cr, ok := params[0].(cfxTypes.CallRequest) - if ok { // only non-verbose base32 address format is accepted + if ok && cr.To != nil { // only non-verbose base32 address format is accepted res = append(res, cr.To.MustGetBase32Address()) } @@ -390,7 +419,7 @@ func (v *CfxValidator) parseAddr(params []interface{}) (res []string, ok bool) { ca, ok := params[0].(cfxTypes.Address) if ok { - res = append(res, ca.MustGetVerboseBase32Address()) + res = append(res, ca.MustGetBase32Address()) } return diff --git a/util/acl/validator_boundary_test.go b/util/acl/validator_boundary_test.go new file mode 100644 index 00000000..aaed18a2 --- /dev/null +++ b/util/acl/validator_boundary_test.go @@ -0,0 +1,93 @@ +package acl + +import ( + "context" + "encoding/json" + "testing" + + "github.com/Conflux-Chain/confura/util/rpc/handlers" + cfxTypes "github.com/Conflux-Chain/go-conflux-sdk/types" + "github.com/Conflux-Chain/go-conflux-sdk/types/cfxaddress" + "github.com/ethereum/go-ethereum/common" + web3Types "github.com/openweb3/web3go/types" + "github.com/stretchr/testify/require" +) + +func TestContractACLFilterAndSubscriptionBoundaries(t *testing.T) { + v := NewEthValidator(&AllowList{ContractAddresses: []string{allowedContractAddr}}) + for _, method := range []string{"eth_getLogs", "eth_newFilter"} { + ctx := ethGetLogsContext(web3Types.FilterQuery{Addresses: []common.Address{common.HexToAddress(allowedContractAddr)}}) + ctx.RpcMethod = method + require.NoError(t, v.Validate(ctx)) + ctx.ExtractRpcParams = nil + require.Error(t, v.Validate(ctx)) + } + for _, tc := range []struct { + params string + allowed bool + }{ + {`["logs",{"address":"` + allowedContractAddr + `"}]`, true}, + {`["logs",{"address":"` + notAllowedContractAddr + `"}]`, false}, + {`["logs",{}]`, false}, {`["logs"]`, false}, {`["logs",null]`, false}, + {`["newHeads"]`, true}, {`[null]`, true}, + } { + ctx := Context{Context: context.Background(), RpcMethod: "eth_subscribe", RawParams: json.RawMessage(tc.params)} + err := v.Validate(ctx) + require.Equal(t, tc.allowed, err == nil, tc.params) + } +} + +func TestContractACLRetainsFilterIDReads(t *testing.T) { + for _, tc := range []struct { + prefix string + factory ValidatorFactory + }{ + {"eth", NewEthValidator}, + {"cfx", NewCfxValidator}, + } { + t.Run(tc.prefix, func(t *testing.T) { + for _, suffix := range []string{"getFilterLogs", "getFilterChanges"} { + method := tc.prefix + "_" + suffix + ctx := Context{Context: context.Background(), RpcMethod: method, RawParams: json.RawMessage(`["0x1"]`)} + require.NoError(t, tc.factory(&AllowList{ContractAddresses: []string{allowedContractAddr}}).Validate(ctx)) + require.NoError(t, tc.factory(&AllowList{}).Validate(ctx)) + // Filter-ID reads still enforce method restrictions. + require.ErrorIs(t, tc.factory(&AllowList{ContractAddresses: []string{allowedContractAddr}, DisallowMethods: []string{method}}).Validate(ctx), errBadRpcMethod) + } + }) + } +} + +func TestCoreNewFilterContractACL(t *testing.T) { + allowed := cfxaddress.MustNew(allowedContractAddr, 1) + denied := cfxaddress.MustNew(notAllowedContractAddr, 1) + v := NewCfxValidator(&AllowList{ContractAddresses: []string{allowed.MustGetBase32Address()}}) + for _, tc := range []struct { + addresses []cfxaddress.Address + allowed bool + }{ + {[]cfxaddress.Address{allowed}, true}, + {[]cfxaddress.Address{denied}, false}, + {[]cfxaddress.Address{allowed, denied}, false}, + {nil, false}, + } { + ctx := cfxGetLogsContext(cfxTypes.LogFilter{Address: tc.addresses}) + ctx.RpcMethod = "cfx_newFilter" + require.Equal(t, tc.allowed, v.Validate(ctx) == nil) + } +} + +func TestLiteralOriginAndCoreAddress(t *testing.T) { + v := NewEthValidator(&AllowList{Origins: []string{"https://app.example.com"}}) + for _, origin := range []string{"https://app-example.com", "https://appXexample.com"} { + require.Error(t, v.Validate(Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, origin)})) + } + require.NoError(t, v.Validate(Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, "https://app.example.com")})) + addr := cfxaddress.MustNew(allowedContractAddr, 1) + for _, form := range []string{addr.MustGetBase32Address(), addr.MustGetVerboseBase32Address()} { + v := NewCfxValidator(&AllowList{ContractAddresses: []string{form}}) + for _, method := range []string{"cfx_getBalance", "cfx_getCode", "cfx_getStorageAt", "cfx_getNextNonce"} { + require.NoError(t, v.Validate(Context{RpcMethod: method, ExtractRpcParams: func() ([]interface{}, error) { return []interface{}{addr}, nil }})) + } + } +} diff --git a/util/acl/validator_rules_test.go b/util/acl/validator_rules_test.go new file mode 100644 index 00000000..239428a1 --- /dev/null +++ b/util/acl/validator_rules_test.go @@ -0,0 +1,128 @@ +package acl + +import ( + "context" + "fmt" + "regexp" + "strings" + "sync" + "testing" + + "github.com/Conflux-Chain/confura/util" + "github.com/Conflux-Chain/confura/util/rpc/handlers" + "github.com/stretchr/testify/require" +) + +func TestValidatorRuleSemantics(t *testing.T) { + for _, tc := range []struct { + name, origin, method string + list AllowList + want error + }{ + {name: "empty rules", method: "eth_getLogs"}, + {name: "wildcards", list: AllowList{Origins: []string{"HTTPS://*.EXAMPLE.COM"}, AllowMethods: []string{"eth_*"}}, origin: "https://app.example.com", method: "eth_getLogs"}, + {name: "literal origin", list: AllowList{Origins: []string{"https://app.example.com"}}, origin: "https://appXexample.com", want: errInvalidOrigin}, + {name: "literal method", list: AllowList{AllowMethods: []string{"eth_read.v1"}}, method: "eth_readXv1", want: errBadRpcMethod}, + {name: "regexp characters", list: AllowList{AllowMethods: []string{"eth_[read]+(v1)?$"}}, method: "eth_[read]+(v1)?$"}, + {name: "case sensitive method", list: AllowList{AllowMethods: []string{"eth_*"}}, method: "ETH_getLogs", want: errBadRpcMethod}, + {name: "missing origin", list: AllowList{Origins: []string{"*"}}, want: errInvalidOrigin}, + {name: "allow takes priority", list: AllowList{AllowMethods: []string{"eth_getLogs"}, DisallowMethods: []string{"eth_*"}}, method: "eth_getLogs"}, + {name: "allow miss denies", list: AllowList{AllowMethods: []string{"eth_getLogs"}, DisallowMethods: []string{"cfx_*"}}, method: "eth_call", want: errBadRpcMethod}, + {name: "disallow hit", list: AllowList{DisallowMethods: []string{"eth_*"}}, method: "eth_getLogs", want: errBadRpcMethod}, + {name: "disallow miss", list: AllowList{DisallowMethods: []string{"cfx_*"}}, method: "eth_getLogs"}, + {name: "invalid allow denies", list: AllowList{AllowMethods: []string{"eth_\xff"}}, method: "eth_getLogs", want: errBadRpcMethod}, + {name: "invalid disallow ignored", list: AllowList{DisallowMethods: []string{"eth_\xff"}}, method: "eth_getLogs"}, + {name: "invalid followed by valid allow", list: AllowList{AllowMethods: []string{"eth_\xff", "eth_*"}}, method: "eth_getLogs"}, + {name: "invalid followed by valid disallow", list: AllowList{DisallowMethods: []string{"eth_\xff", "eth_*"}}, method: "eth_getLogs", want: errBadRpcMethod}, + } { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + if tc.origin != "" { + ctx = context.WithValue(ctx, handlers.CtxKeyReqOrigin, tc.origin) + } + err := NewEthValidator(&tc.list).Validate(Context{Context: ctx, RpcMethod: tc.method}) + require.ErrorIs(t, err, tc.want) + }) + } +} + +// Compare decisions with the prior per-request regexp path, including invalid +// UTF-8 and empty-string patterns, whose behavior must not change on compile. +func TestValidatorRulesMatchOriginalDecisions(t *testing.T) { + ruleSets := [][]string{nil, {""}, {"*"}, {"eth_*", "cfx_*"}, {"eth_\xff"}, {"eth_\xff", "eth_*"}, {"https://*.example.com"}, {"https://\xff.example.com"}, {"[a]+(b)?."}} + inputs := []string{"", "eth_getLogs", "ETH_getLogs", "eth_\xff", "cfx_call", "https://app.example.com", "https://\xff.example.com", "[a]+(b)?."} + matches := func(rules []string, input string, lower bool) bool { + for _, rule := range rules { + if lower { + rule, input = strings.ToLower(rule), strings.ToLower(input) + } + if ok, err := regexp.MatchString(util.WildCardToRegexp(rule), input); err == nil && ok { + return true + } + } + return false + } + for _, rules := range ruleSets { + for _, input := range inputs { + ctx := Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, input), RpcMethod: input} + for _, kind := range []string{"origin", "allow", "disallow"} { + al := &AllowList{} + allowed := true + switch kind { + case "origin": + al.Origins = rules + allowed = len(rules) == 0 || matches(rules, input, true) + case "allow": + al.AllowMethods = rules + allowed = len(rules) == 0 || matches(rules, input, false) + case "disallow": + al.DisallowMethods = rules + allowed = !matches(rules, input, false) + } + require.Equal(t, allowed, NewEthValidator(al).Validate(ctx) == nil, "%s rules=%q input=%q", kind, rules, input) + } + } + } +} + +func TestValidatorConcurrentRuleReads(t *testing.T) { + v := NewEthValidator(&AllowList{Origins: []string{"https://*.example.com"}, AllowMethods: []string{"eth_*"}}) + ctx := Context{Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, "https://app.example.com"), RpcMethod: "eth_getLogs"} + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 32; j++ { + if err := v.Validate(ctx); err != nil { + t.Error(err) + } + } + }() + } + wg.Wait() +} + +func BenchmarkACLValidation(b *testing.B) { + for _, count := range []int{1, 10} { + b.Run(fmt.Sprint(count), func(b *testing.B) { + al := &AllowList{} + for i := 0; i < count; i++ { + al.Origins = append(al.Origins, fmt.Sprintf("https://app%d.*.example.com", i)) + al.AllowMethods = append(al.AllowMethods, fmt.Sprintf("fixture%d_*", i)) + } + v := NewEthValidator(al) + ctx := Context{ + Context: context.WithValue(context.Background(), handlers.CtxKeyReqOrigin, fmt.Sprintf("https://app%d.rpc.example.com", count-1)), + RpcMethod: fmt.Sprintf("fixture%d_read", count-1), + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := v.Validate(ctx); err != nil { + b.Fatal(err) + } + } + }) + } +} diff --git a/util/blockchain.go b/util/blockchain.go index e480b298..e0c897ca 100644 --- a/util/blockchain.go +++ b/util/blockchain.go @@ -171,7 +171,7 @@ func IsEip155Tx(tx *web3goTypes.TransactionDetail) bool { // IsLegacyEthTx check if the EVM transaction is legacy (pre EIP155) func IsLegacyEthTx(tx *web3goTypes.TransactionDetail) bool { - if tx.V != nil && tx.V.Uint64() == 27 || tx.V.Uint64() == 28 { + if tx != nil && tx.V != nil && (tx.V.Uint64() == 27 || tx.V.Uint64() == 28) { return true } @@ -200,6 +200,7 @@ func MatchEthLogTopics(log *web3goTypes.Log, topics [][]common.Hash) bool { for i := range topics { if len(topics[i]) == 0 { + // Conflux eSpace wildcards also match missing topic positions. continue } diff --git a/util/boundary_test.go b/util/boundary_test.go new file mode 100644 index 00000000..8deddd04 --- /dev/null +++ b/util/boundary_test.go @@ -0,0 +1,50 @@ +package util + +import ( + "testing" + + "github.com/ethereum/go-ethereum/common" + web3Types "github.com/openweb3/web3go/types" +) + +// Follow Conflux eSpace's LogFilterParams::matches, including missing wildcard +// positions: https://github.com/Conflux-Chain/conflux-rust/blob/2cb16583060d7a569271adb59cc1a9cd92dfd447/crates/primitives/src/filter.rs#L317-L339 +func TestEthLogTopicsMatchConfluxSemantics(t *testing.T) { + a, b := common.HexToHash("0x1"), common.HexToHash("0x2") + for _, tc := range []struct { + name string + log []common.Hash + filter [][]common.Hash + want bool + }{ + {"no conditions", nil, nil, true}, + {"existing wildcard", []common.Hash{a}, [][]common.Hash{{}}, true}, + {"missing null position", []common.Hash{a}, [][]common.Hash{{a}, nil}, true}, + {"missing empty position", []common.Hash{a}, [][]common.Hash{{a}, {}}, true}, + {"all wildcard positions", []common.Hash{a}, [][]common.Hash{nil, nil}, true}, + {"wildcard without topics", nil, [][]common.Hash{nil}, true}, + {"missing required topic", []common.Hash{a}, [][]common.Hash{{a}, {b}}, false}, + {"wildcard then missing required topic", []common.Hash{a}, [][]common.Hash{nil, {b}}, false}, + {"wildcard hole", []common.Hash{a, b}, [][]common.Hash{nil, {b}}, true}, + {"matching alternative", []common.Hash{a}, [][]common.Hash{{b, a}}, true}, + {"nonmatching topic", []common.Hash{a}, [][]common.Hash{{b}}, false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := MatchEthLogTopics(&web3Types.Log{Topics: tc.log}, tc.filter); got != tc.want { + t.Fatalf("matched %v, want %v", got, tc.want) + } + }) + } +} + +func TestNilBoundaries(t *testing.T) { + if IsInterfaceValNil([1]int{}) { + t.Fatal("array is not nil") + } + if !IsInterfaceValNil((*int)(nil)) { + t.Fatal("typed nil not recognized") + } + if IsLegacyEthTx(nil) || IsLegacyEthTx(&web3Types.TransactionDetail{}) { + t.Fatal("missing signature is not legacy") + } +} diff --git a/util/metrics/service/registry.go b/util/metrics/service/registry.go index 7e3dab50..bc296e0c 100644 --- a/util/metrics/service/registry.go +++ b/util/metrics/service/registry.go @@ -22,35 +22,35 @@ func NewRegistry(updater Updater) metrics.Registry { func (r *Registry) GetOrRegisterCounter(name string, args ...interface{}) metrics.Counter { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.Counter { + return r.Registry.GetOrRegister(metricName, func() metrics.Counter { return &Counter{clientMetric{metricName, r.updater}, metrics.NewCounter()} }).(metrics.Counter) } func (r *Registry) GetOrRegisterGauge(name string, args ...interface{}) metrics.Gauge { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.Gauge { + return r.Registry.GetOrRegister(metricName, func() metrics.Gauge { return &Gauge{clientMetric{metricName, r.updater}, metrics.NewGauge()} }).(metrics.Gauge) } func (r *Registry) GetOrRegisterGaugeFloat64(name string, args ...interface{}) metrics.GaugeFloat64 { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.GaugeFloat64 { + return r.Registry.GetOrRegister(metricName, func() metrics.GaugeFloat64 { return &GaugeFloat64{clientMetric{metricName, r.updater}, metrics.NewGaugeFloat64()} }).(metrics.GaugeFloat64) } func (r *Registry) GetOrRegisterMeter(name string, args ...interface{}) metrics.Meter { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.Meter { + return r.Registry.GetOrRegister(metricName, func() metrics.Meter { return &Meter{clientMetric{metricName, r.updater}, metrics.NewMeter()} }).(metrics.Meter) } func (r *Registry) GetOrRegisterHistogram(name string, args ...interface{}) metrics.Histogram { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.Histogram { + return r.Registry.GetOrRegister(metricName, func() metrics.Histogram { return &Histogram{ clientMetric{metricName, r.updater}, metrics.NewHistogram(metrics.NewExpDecaySample(1028, 0.015)), @@ -60,14 +60,14 @@ func (r *Registry) GetOrRegisterHistogram(name string, args ...interface{}) metr func (r *Registry) GetOrRegisterTimer(name string, args ...interface{}) metrics.Timer { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metrics.Timer { + return r.Registry.GetOrRegister(metricName, func() metrics.Timer { return &Timer{clientMetric{metricName, r.updater}, metrics.NewTimer()} }).(metrics.Timer) } func (r *Registry) GetOrRegisterPercentage(name string, args ...interface{}) metricUtil.Percentage { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metricUtil.Percentage { + return r.Registry.GetOrRegister(metricName, func() metricUtil.Percentage { return &Percentage{clientMetric{metricName, r.updater}, metricUtil.NewPercentage()} }).(metricUtil.Percentage) } @@ -78,7 +78,7 @@ func (r *Registry) GetOrRegisterTimeWindowPercentageDefault(name string, args .. func (r *Registry) GetOrRegisterTimeWindowPercentage(slots int, slotInterval time.Duration, name string, args ...interface{}) metricUtil.Percentage { metricName := fmt.Sprintf(name, args...) - return r.Registry.GetOrRegister(name, func() metricUtil.Percentage { + return r.Registry.GetOrRegister(metricName, func() metricUtil.Percentage { return &TimeWindowPercentage{clientMetric{metricName, r.updater}, metricUtil.NewPercentage(), slots, slotInterval} }).(metricUtil.Percentage) } diff --git a/util/metrics/service/registry_test.go b/util/metrics/service/registry_test.go new file mode 100644 index 00000000..203e2936 --- /dev/null +++ b/util/metrics/service/registry_test.go @@ -0,0 +1,17 @@ +package service + +import "testing" + +func TestFormattedMetricNamesRemainDistinct(t *testing.T) { + r := NewRegistry(nil).(*Registry) + a, b := r.GetOrRegisterGauge("fixture/%s", "a"), r.GetOrRegisterGauge("fixture/%s", "b") + if a == b { + t.Fatal("distinct names alias") + } + if a != r.GetOrRegisterGauge("fixture/%s", "a") { + t.Fatal("same name not reused") + } + if r.Get("fixture/a") != a || r.Get("fixture/b") != b { + t.Fatal("registry contains format instead of name") + } +} diff --git a/util/rate/loader.go b/util/rate/loader.go index 0102169b..1c3227eb 100644 --- a/util/rate/loader.go +++ b/util/rate/loader.go @@ -1,11 +1,13 @@ package rate import ( + "errors" "sync" "time" "github.com/Conflux-Chain/confura/util" "github.com/sirupsen/logrus" + "golang.org/x/sync/singleflight" ) const ( @@ -13,6 +15,8 @@ const ( LimitKeyExpirationTTL = 60 * time.Second ) +var errKeyLoad = errors.New("key lookup failed") + type KeyInfo struct { SID uint32 // bound strategy ID AclID uint32 // bound allowlist ID @@ -30,7 +34,8 @@ type KeysetFilter struct { type ksLoadFunc func(filter *KeysetFilter) ([]*KeyInfo, error) type KeyLoader struct { - mu sync.Mutex + mu sync.Mutex + loads singleflight.Group // raw keyset load function ksload ksLoadFunc // limit key cache: limit key => *KeyInfo (nil if missing) @@ -54,13 +59,23 @@ func NewKeyLoader(ksload ksLoadFunc) *KeyLoader { // Load loads key info from cache or raw loading from somewhere else // if cache missed. func (l *KeyLoader) Load(key string) (*KeyInfo, bool) { - // load from cache at first - if ki, ok := l.cacheLoad(key); ok { // cache hit - return ki, ok + if cv, expired, found := l.keyCache.GetWithoutExp(key); found && !expired { + return cv.(*KeyInfo), true } - - // otherwise, populate the cache - return l.populateCache(key) + result, err, _ := l.loads.Do(key, func() (interface{}, error) { + if ki, ok := l.cacheLoad(key); ok { + return ki, nil + } + ki, ok := l.populateCache(key) + if !ok { + return (*KeyInfo)(nil), errKeyLoad + } + return ki, nil + }) + if err != nil { + return nil, false + } + return result.(*KeyInfo), true } func (l *KeyLoader) cacheLoad(key string) (*KeyInfo, bool) { @@ -98,9 +113,7 @@ func (l *KeyLoader) populateCache(key string) (*KeyInfo, bool) { l.keyCache.Add(key, (*KeyInfo)(nil)) } - logrus.WithField("key", key). - WithError(err). - Error("Key loader failed to load limit key info") + logrus.Error("Key loader failed to load limit key info") return nil, false } @@ -124,7 +137,7 @@ func (kl *KeyLoader) rawLoad(key string) (*KeyInfo, error) { func (kl *KeyLoader) warmUpKeyCache() { kis, err := kl.ksload(&KeysetFilter{Limit: (LimitKeyCacheSize * 3 / 4)}) if err != nil { - logrus.WithError(err).Warn("Failed to load limit keyset to warm up cache") + logrus.Warn("Failed to load limit keyset to warm up cache") return } diff --git a/util/rate/loader_test.go b/util/rate/loader_test.go new file mode 100644 index 00000000..24f6335c --- /dev/null +++ b/util/rate/loader_test.go @@ -0,0 +1,62 @@ +package rate + +import ( + "bytes" + "errors" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/Conflux-Chain/confura/util" + "github.com/sirupsen/logrus" +) + +func TestKeyLoaderCoalescesConcurrentMisses(t *testing.T) { + var calls atomic.Int32 + entered, release := make(chan struct{}), make(chan struct{}) + l := &KeyLoader{keyCache: util.NewExpirableLruCache(LimitKeyCacheSize, LimitKeyExpirationTTL)} + l.ksload = func(*KeysetFilter) ([]*KeyInfo, error) { + if calls.Add(1) == 1 { + close(entered) + } + <-release + return []*KeyInfo{{Key: "fixture", SID: 2}}, nil + } + const n = 32 + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ki, ok := l.Load("fixture") + if !ok || ki.SID != 2 { + t.Error("incorrect lookup") + } + }() + } + <-entered + close(release) + wg.Wait() + if calls.Load() != 1 { + t.Fatalf("got %d DB reads", calls.Load()) + } +} + +func TestKeyLoaderDoesNotLogCredentials(t *testing.T) { + const key = "secret-fixture-not-a-real-key" + var out bytes.Buffer + old := logrus.StandardLogger().Out + logrus.SetOutput(&out) + defer logrus.SetOutput(old) + l := &KeyLoader{keyCache: util.NewExpirableLruCache(1, LimitKeyExpirationTTL), ksload: func(*KeysetFilter) ([]*KeyInfo, error) { return nil, errors.New("query key=" + key) }} + if _, ok := l.Load(key); ok { + t.Fatal("error treated as successful load") + } + if strings.Contains(out.String(), key) { + t.Fatalf("credential logged: %s", out.String()) + } + if ki, ok := l.Load(key); !ok || ki != nil { + t.Fatal("missing negative cache") + } +} diff --git a/util/rate/registry.go b/util/rate/registry.go index 7b74dbd3..97f85aab 100644 --- a/util/rate/registry.go +++ b/util/rate/registry.go @@ -2,7 +2,9 @@ package rate import ( "context" + "math" "sync" + "sync/atomic" "time" "github.com/Conflux-Chain/confura/util/acl" @@ -38,8 +40,9 @@ type Registry struct { *http.Registry *aclRegistry - mu sync.Mutex - kloader *KeyLoader + mu sync.Mutex + limiterStrategies atomic.Value // immutable map[string]*Strategy; Create never takes mu + kloader *KeyLoader // all available strategies strategies map[string]*Strategy // strategy name => *Strategy @@ -62,6 +65,7 @@ func NewRegistry(kloader *KeyLoader, valFactory acl.ValidatorFactory) *Registry ), } + m.publishLimiterStrategies() m.Registry = http.NewRegistry(m) go m.ScheduleGC(GCScheduleInterval) @@ -112,10 +116,8 @@ func (r *Registry) GetGroupAndKey( } func (r *Registry) Create(ctx context.Context, resource, group string) (rate.Limiter, error) { - r.mu.Lock() - defer r.mu.Unlock() - - stg, ok := r.strategies[group] + strategies := r.limiterStrategies.Load().(map[string]*Strategy) + stg, ok := strategies[group] if !ok { return nil, errors.New("strategy not found") } @@ -131,12 +133,35 @@ func (r *Registry) Create(ctx context.Context, resource, group string) (rate.Lim func (r *Registry) createWithOption(option interface{}) (l rate.Limiter, err error) { switch opt := option.(type) { case FixedWindowOption: + if opt.Interval <= 0 || opt.Quota <= 0 { + return nil, errors.New("fixed window interval and quota must be positive") + } l = rate.NewFixedWindow(opt.Interval, opt.Quota) case TokenBucketOption: - l = rate.NewTokenBucket(int(opt.Rate), opt.Burst) + if opt.Rate <= 0 || math.IsNaN(float64(opt.Rate)) || math.IsInf(float64(opt.Rate), 0) { + return nil, errors.New("token bucket rate must be positive and finite") + } + if opt.Burst <= 0 { + return nil, errors.New("token bucket burst must be positive") + } + if float64(opt.Burst)/float64(opt.Rate) >= float64(math.MaxInt64/int64(time.Second)) { + return nil, errors.New("token bucket refill interval is too large") + } + l = rate.NewTokenBucketRate(opt.Rate, opt.Burst) default: err = errors.New("invalid limit option") } return l, err } + +// publishLimiterStrategies is called with mu held (or before publication at construction). +// The dependency invokes Create with its own lock held. Publish an immutable +// snapshot so Create does not reverse reload's mu -> registry lock order. +func (r *Registry) publishLimiterStrategies() { + snapshot := make(map[string]*Strategy, len(r.strategies)) + for name, strategy := range r.strategies { + snapshot[name] = strategy + } + r.limiterStrategies.Store(snapshot) +} diff --git a/util/rate/registry_concurrency_test.go b/util/rate/registry_concurrency_test.go new file mode 100644 index 00000000..08588a9c --- /dev/null +++ b/util/rate/registry_concurrency_test.go @@ -0,0 +1,108 @@ +package rate + +import ( + "context" + "encoding/json" + "math" + "testing" + "time" + + limiter "github.com/Conflux-Chain/go-conflux-util/rate" + httprate "github.com/Conflux-Chain/go-conflux-util/rate/http" + xrate "golang.org/x/time/rate" +) + +func TestCreateDoesNotAcquireStrategyLock(t *testing.T) { + r := &Registry{strategies: map[string]*Strategy{"default": {Name: "default", LimitOptions: map[string]interface{}{"rpc": TokenBucketOption{Rate: 1, Burst: 1}}}}} + r.publishLimiterStrategies() + // Reload holds mu while waiting for the dependency registry lock. The + // dependency holds that lock when it invokes Create: Create must progress. + r.mu.Lock() + defer r.mu.Unlock() + done := make(chan error, 1) + go func() { _, err := r.Create(context.Background(), "rpc", "default"); done <- err }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("Create waits for reload lock") + } +} + +func TestFractionalTokenBucket(t *testing.T) { + r := &Registry{} + l, err := r.createWithOption(TokenBucketOption{Rate: 0.5, Burst: 1}) + if err != nil { + t.Fatal(err) + } + bucket := l.(*limiter.TokenBucket) + // Past times make this test independent of the old dependency's Delay() + // wall-clock implementation as well as its corrected DelayFrom(now). + now := time.Now().Add(-10 * time.Second) + if err := bucket.LimitAt(now, 1); err != nil { + t.Fatal(err) + } + if err := bucket.LimitAt(now.Add(2*time.Second), 1); err != nil { + t.Fatalf("fractional rate did not refill: %v", err) + } +} + +func TestInvalidLimiterOptionsReturnErrors(t *testing.T) { + r := &Registry{} + for _, opt := range []interface{}{TokenBucketOption{Rate: 0, Burst: 1}, TokenBucketOption{Rate: -1, Burst: 1}, TokenBucketOption{Rate: 1, Burst: 0}, TokenBucketOption{Rate: xrate.Limit(math.NaN()), Burst: 1}, TokenBucketOption{Rate: xrate.Limit(math.Inf(1)), Burst: 1}, TokenBucketOption{Rate: 1e-20, Burst: 1}, FixedWindowOption{}, FixedWindowOption{Interval: -1, Quota: 1}} { + if _, err := r.createWithOption(opt); err == nil { + t.Errorf("accepted invalid option %+v", opt) + } + } + var s Strategy + if err := json.Unmarshal([]byte(`{"rpc":null}`), &s); err == nil { + t.Fatal("accepted null rule") + } + if err := json.Unmarshal([]byte(`{"rpc":{"algo":"token_bucket","option":{"rate":0.5,"burst":1}}}`), &s); err != nil { + t.Fatal(err) + } +} + +type gatedLimiterFactory struct { + registry *Registry + entered chan struct{} + release chan struct{} +} + +func (f *gatedLimiterFactory) GetGroupAndKey(context.Context, string) (string, string, error) { + return "default", "fixture", nil +} +func (f *gatedLimiterFactory) Create(ctx context.Context, resource, group string) (limiter.Limiter, error) { + close(f.entered) + <-f.release + return f.registry.Create(ctx, resource, group) +} +func TestReloadAndDependencyCreateMakeProgress(t *testing.T) { + strategy := &Strategy{ID: 1, Name: "default", LimitOptions: map[string]interface{}{"rpc": TokenBucketOption{Rate: 1, Burst: 1}}} + r := &Registry{strategies: map[string]*Strategy{"default": strategy}, id2Strategies: map[uint32]*Strategy{1: strategy}} + r.publishLimiterStrategies() + factory := &gatedLimiterFactory{registry: r, entered: make(chan struct{}), release: make(chan struct{})} + r.Registry = httprate.NewRegistry(factory) + requestDone := make(chan error, 1) + go func() { requestDone <- r.Limit(context.Background(), "rpc") }() + <-factory.entered // dependency now owns its limiter registry lock + r.mu.Lock() // reload owns the strategy lock + defer r.mu.Unlock() + reloadDone := make(chan struct{}) + go func() { r.removeStrategy(strategy); close(reloadDone) }() + close(factory.release) + for _, done := range []<-chan struct{}{reloadDone} { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("reload deadlocked") + } + } + select { + case <-requestDone: + case <-time.After(time.Second): + t.Fatal("request deadlocked") + } +} diff --git a/util/rate/registry_reload.go b/util/rate/registry_reload.go index d83d1475..39f1b7eb 100644 --- a/util/rate/registry_reload.go +++ b/util/rate/registry_reload.go @@ -101,24 +101,29 @@ func (m *Registry) getVipStrategy(tier handlers.VipTier) (*Strategy, bool) { } func (m *Registry) removeStrategy(s *Strategy) { + delete(m.strategies, s.Name) + delete(m.id2Strategies, s.ID) + m.publishLimiterStrategies() // remove all limiters under this strategy for resource := range s.LimitOptions { m.Remove(resource, s.Name) logrus.WithField("resource", resource).Info("RateLimit rule deleted") } - - delete(m.strategies, s.Name) - delete(m.id2Strategies, s.ID) } func (m *Registry) addStrategy(s *Strategy) { m.strategies[s.Name] = s m.id2Strategies[s.ID] = s + m.publishLimiterStrategies() } func (m *Registry) updateStrategy(old, new *Strategy) { m.strategies[new.Name] = new m.id2Strategies[new.ID] = new + if old.Name != new.Name { + delete(m.strategies, old.Name) + } + m.publishLimiterStrategies() if old.Name != new.Name { // strategy name changed? this shall be rare, but we also need to diff --git a/util/rate/resolver.go b/util/rate/resolver.go index 7d0780ee..512e80a4 100644 --- a/util/rate/resolver.go +++ b/util/rate/resolver.go @@ -112,7 +112,7 @@ func (r *ProvisionedResolver) Resolve(ctx context.Context, reg *Registry) (*Stra stg, _ := reg.id2Strategies[ki.SID] if stg == nil { - logrus.WithField("keyInfo", ki).Debug("Custom rate limit strategy not configured") + logrus.WithField("strategyID", ki.SID).Debug("Custom rate limit strategy not configured") } var limitKey string diff --git a/util/rate/strategy.go b/util/rate/strategy.go index bc574bd7..a05c21a9 100644 --- a/util/rate/strategy.go +++ b/util/rate/strategy.go @@ -63,9 +63,14 @@ func (s *Strategy) UnmarshalJSON(data []byte) error { return errors.WithMessage(err, "malformed json format") } + options := make(map[string]interface{}, len(tmpRules)) for resource, rule := range tmpRules { - s.LimitOptions[resource] = rule.Option + if rule == nil { + return errors.New("null rate limit rule") + } + options[resource] = rule.Option } + s.LimitOptions = options return nil } diff --git a/util/rate/util.go b/util/rate/util.go index 803f8e67..d0806be9 100644 --- a/util/rate/util.go +++ b/util/rate/util.go @@ -1,8 +1,7 @@ package rate import ( - "math/rand" - "time" + "crypto/rand" "github.com/btcsuite/btcutil/base58" "github.com/pkg/errors" @@ -13,15 +12,13 @@ const ( ) func GenerateRandomLimitKey(limitType LimitType) (string, error) { - data := make([]byte, LimitKeyLength) - - source := rand.NewSource(time.Now().UnixNano()) - r := rand.New(source) - r.Read(data) - if limitType != LimitTypeByKey && limitType != LimitTypeByIp { return "", errors.New("invalid limit type") } + data := make([]byte, LimitKeyLength) + if _, err := rand.Read(data); err != nil { + return "", errors.Wrap(err, "failed to generate limit key") + } secretb := []byte(base58.Encode(data)) secretb[0] = '0' + uint8(limitType) diff --git a/util/rate/util_test.go b/util/rate/util_test.go index 17ef129b..44951883 100644 --- a/util/rate/util_test.go +++ b/util/rate/util_test.go @@ -1,33 +1,34 @@ package rate import ( + "bytes" + "crypto/rand" + "errors" "testing" - "github.com/stretchr/testify/assert" + "github.com/btcsuite/btcutil/base58" + "github.com/stretchr/testify/require" ) -func TestGenerateRandomLimitKey(t *testing.T) { - // limit type by IP - limitKey, err := GenerateRandomLimitKey(LimitTypeByIp) - assert.Nil(t, err) - - assert.Greater(t, len(limitKey), LimitKeyLength) - assert.Less(t, len(limitKey), LimitKeyLength*2) - - t.Logf("New random limit key of type `byIP`: %v", limitKey) - assert.Equal(t, limitKey[0], uint8('0'+LimitTypeByIp)) - - // limit type by Key - limitKey, err = GenerateRandomLimitKey(LimitTypeByKey) - assert.Nil(t, err) +type failingRandomReader struct{} - assert.Greater(t, len(limitKey), LimitKeyLength) - assert.Less(t, len(limitKey), LimitKeyLength*2) - - t.Logf("New random limit key of type `byKey`: %v", limitKey) - assert.Equal(t, limitKey[0], uint8('0'+LimitTypeByKey)) - - // invalid limit type - _, err = GenerateRandomLimitKey(-1) - assert.NotNil(t, err) +func (failingRandomReader) Read([]byte) (int, error) { return 0, errors.New("entropy unavailable") } +func TestGenerateRandomLimitKey(t *testing.T) { + original := rand.Reader + t.Cleanup(func() { rand.Reader = original }) + for _, kind := range []LimitType{LimitTypeByKey, LimitTypeByIp} { + entropy := bytes.Repeat([]byte{42}, LimitKeyLength) + rand.Reader = bytes.NewReader(entropy) + key, err := GenerateRandomLimitKey(kind) + require.NoError(t, err) + expected := []byte(base58.Encode(entropy)) + expected[0] = '0' + byte(kind) + require.Equal(t, string(expected), key) + } + rand.Reader = failingRandomReader{} + key, err := GenerateRandomLimitKey(LimitTypeByKey) + require.ErrorContains(t, err, "entropy unavailable") + require.Empty(t, key) + _, err = GenerateRandomLimitKey(LimitType(255)) + require.Error(t, err) } diff --git a/util/regex.go b/util/regex.go index 1e6b4da9..c6b9f515 100644 --- a/util/regex.go +++ b/util/regex.go @@ -8,10 +8,6 @@ import ( // WildCardToRegexp converts a wildcard pattern to a regular expression pattern. func WildCardToRegexp(pattern string) string { components := strings.Split(pattern, "*") - if len(components) == 1 { - // if len is 1, there are no *'s, return exact match pattern - return "^" + pattern + "$" - } var result strings.Builder for i, literal := range components { diff --git a/util/relay/relay.go b/util/relay/relay.go index 651c9f2a..79d8258c 100644 --- a/util/relay/relay.go +++ b/util/relay/relay.go @@ -61,12 +61,19 @@ func newTxnRelayer(relayConf *TxnRelayerConfig) *txnRelayer { // Relay relays raw transaction broadcasting asynchronously. func (relayer *txnRelayer) Relay(signedTx hexutil.Bytes) bool { - if len(relayer.txnQueue) == relayer.BufferSize { // queue is full? + // Avoid copying a transaction while a buffered queue is already full. + // This is only an early rejection: select still decides admission atomically. + if capacity := cap(relayer.txnQueue); capacity > 0 && len(relayer.txnQueue) == capacity { + return false + } + // Own queued bytes; callers may reuse their request buffer after return. + raw := append(hexutil.Bytes(nil), signedTx...) + select { + case relayer.txnQueue <- raw: + return true + default: return false } - - relayer.txnQueue <- signedTx - return true } // run starts concurrency worker(s) asynchronously diff --git a/util/relay/relay_test.go b/util/relay/relay_test.go new file mode 100644 index 00000000..70d5359a --- /dev/null +++ b/util/relay/relay_test.go @@ -0,0 +1,115 @@ +package relay + +import ( + "fmt" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestRelayConcurrentProducersNeverWaitForConsumer(t *testing.T) { + r := newTxnRelayer(&TxnRelayerConfig{BufferSize: 2}) + raw := []byte{1, 2} + if !r.Relay(raw) { + t.Fatal("initial enqueue failed") + } + raw[0] = 9 + var accepted atomic.Int32 + var wg sync.WaitGroup + start, done := make(chan struct{}), make(chan struct{}) + for i := 0; i < 64; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if r.Relay([]byte{3}) { + accepted.Add(1) + } + }() + } + close(start) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("producer blocked without consumer") + } + if accepted.Load() != 1 { + t.Fatalf("accepted %d beyond remaining capacity", accepted.Load()) + } + if (<-r.txnQueue)[0] != 1 { + t.Fatal("caller mutated queued bytes") + } +} + +func TestRelayFullQueueDoesNotCopy(t *testing.T) { + r := newTxnRelayer(&TxnRelayerConfig{BufferSize: 1}) + if !r.Relay([]byte{1}) { + t.Fatal("initial admission failed") + } + raw := make([]byte, 128<<10) + if allocs := testing.AllocsPerRun(100, func() { + if r.Relay(raw) { + t.Fatal("accepted into a full queue") + } + }); allocs != 0 { + t.Fatalf("full queue rejection allocated %v times", allocs) + } +} + +func TestRelayUnbufferedAdmission(t *testing.T) { + r := newTxnRelayer(&TxnRelayerConfig{BufferSize: 0}) + if r.Relay([]byte{1}) { + t.Fatal("unbuffered queue accepted without a receiver") + } + received := make(chan []byte, 1) + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + select { + case raw := <-r.txnQueue: + received <- raw + case <-stop: + } + }() + defer func() { close(stop); <-done }() + raw := []byte{2} + deadline := time.Now().Add(time.Second) + for !r.Relay(raw) { + if time.Now().After(deadline) { + t.Fatal("unbuffered queue never admitted with a waiting receiver") + } + runtime.Gosched() + } + raw[0] = 9 + if got := <-received; got[0] != 2 { + t.Fatal("caller mutated unbuffered transaction bytes") + } +} + +func BenchmarkRelayAdmission(b *testing.B) { + for _, full := range []bool{false, true} { + for _, size := range []int{32, 128 << 10} { + b.Run(fmt.Sprintf("full=%t/%d", full, size), func(b *testing.B) { + r := newTxnRelayer(&TxnRelayerConfig{BufferSize: 1}) + raw := make([]byte, size) + if full && !r.Relay(raw) { + b.Fatal("initial admission failed") + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if r.Relay(raw) == full { + b.Fatal("unexpected admission result") + } + if !full { + <-r.txnQueue + } + } + }) + } + } +} diff --git a/util/rpc/cached_client_eth.go b/util/rpc/cached_client_eth.go index 3e0f9ee9..d6e01cc7 100644 --- a/util/rpc/cached_client_eth.go +++ b/util/rpc/cached_client_eth.go @@ -2,8 +2,10 @@ package rpc import ( "context" + "math" "math/big" "sync" + "time" "github.com/Conflux-Chain/confura-data-cache/nearhead" cacheRpc "github.com/Conflux-Chain/confura-data-cache/rpc" @@ -26,6 +28,15 @@ import ( var ( _ util.EthBlockNumberResolver = (*ethBlockTagResolver)(nil) + + // ErrInconsistentCanonicalView means near-head and node log reads could not + // be aligned within the local retry budget. No partial result is returned. + ErrInconsistentCanonicalView = errors.New("inconsistent canonical view") +) + +const ( + nearHeadLogRetryTimeout = time.Second + nearHeadLogRetryBackoff = 10 * time.Millisecond ) const ( @@ -51,9 +62,12 @@ const ( rpcMethodTraceBlock = "trace_block" ) +// EthNearHeadCache uses data-cache's readers unchanged. Cache-internal locking, +// topic filtering and returned-buffer ownership require fixes in that package. type EthNearHeadCache struct { *nearhead.EthCache stopSync context.CancelFunc + syncDone sync.WaitGroup } func newDummyEthNearHeadCache() *EthNearHeadCache { @@ -65,6 +79,7 @@ func newDummyEthNearHeadCache() *EthNearHeadCache { func (c *EthNearHeadCache) Close() { if c.stopSync != nil { c.stopSync() + c.syncDone.Wait() } } @@ -84,17 +99,17 @@ func NewEthNearHeadCacheFromViper(url string) (*EthNearHeadCache, error) { conf.Adapter.URL = url - // Start near head sync. + // Keep polling, reorg writes and adapter ownership in data-cache. Internal + // cache fixes belong there; Confura only coordinates its full-node reads. ctx, cancel := context.WithCancel(context.Background()) - var wg sync.WaitGroup - - cache, err := cacheSync.StartNearHead(ctx, &wg, conf.NearHeadConfig) + cache := &EthNearHeadCache{stopSync: cancel} + ethCache, err := cacheSync.StartNearHead(ctx, &cache.syncDone, conf.NearHeadConfig) if err != nil { cancel() - return nil, errors.WithMessage(err, "Failed to sync near head data") + return nil, errors.WithMessage(err, "failed to sync near head data") } - - return &EthNearHeadCache{EthCache: cache, stopSync: cancel}, nil + cache.EthCache = ethCache + return cache, nil } func MustNewEthDataCacheClientFromViper() cacheRpc.Interface { @@ -662,72 +677,195 @@ func (c ethCacheClient) logsByBlockHash(filter web3Types.FilterQuery) ([]web3Typ return logs, nil } -func (c ethCacheClient) logsByBlockRange(filter web3Types.FilterQuery) (res []web3Types.Log, err error) { +func (c ethCacheClient) logsByBlockRange(filter web3Types.FilterQuery) ([]web3Types.Log, error) { tracker := newRPCCacheTracker(rpcMethodEthLogsByBlockRange).withNearHeadHit(false).withFullNodeHit(false) defer tracker.report() fromBlock, toBlock, err := c.resolveBlockRange(filter) if err != nil { - err = errors.WithMessage(err, "failed to resolve block range") - return nil, tracker.withError(err) + return nil, tracker.withError(errors.WithMessage(err, "failed to resolve block range")) + } + // Resolve moving tags once. Every attempt, including a cache miss after a + // conflict, must query the same heights without changing the caller's filter. + from, to := web3Types.BlockNumber(fromBlock), web3Types.BlockNumber(toBlock) + filter.FromBlock, filter.ToBlock = &from, &to + + // The local window controls admission to another attempt, not RPC duration. + // Keep the configured client context/timeouts. An admitted attempt may finish + // after this window and still return a coherent result or its original error. + var deadline time.Time + started := time.Now() + for attempts := 1; ; attempts++ { + logs, err := c.logsByBlockRangeOnce(filter, fromBlock, toBlock, tracker) + conflict, ok := err.(*nearHeadLogConflict) + if !ok { + if err == nil && attempts > 1 && logrus.IsLevelEnabled(logrus.DebugLevel) { + logrus.WithField("attempts", attempts).Debug("Near-head log consistency recovered") + } + return logs, tracker.withError(err) + } + if deadline.IsZero() { + deadline = time.Now().Add(nearHeadLogRetryTimeout) + } + logConflict := func(level logrus.Level, message string) { + if !logrus.IsLevelEnabled(level) { + return + } + logrus.WithFields(logrus.Fields{ + "fromBlock": fromBlock, + "toBlock": toBlock, + "boundary": conflict.boundary, + "reason": conflict.reason, + "attempts": attempts, + "elapsed": time.Since(started), + }).Log(level, message) + } + if remaining := time.Until(deadline); remaining > 0 { + logConflict(logrus.DebugLevel, "Retrying near-head logs after canonical view conflict") + time.Sleep(min(nearHeadLogRetryBackoff, remaining)) + } + if !time.Now().Before(deadline) { + logConflict(logrus.InfoLevel, "Near-head log consistency retry window exhausted") + return nil, tracker.withError(ErrInconsistentCanonicalView) + } } +} - ethLogs, err := c.nearhead.GetLogsByBlockRange( - fromBlock, toBlock, - nearhead.FilterOpt{ - Addresses: filter.Addresses, - Topics: filter.Topics, - }, - ) - if err != nil { - err = errors.WithMessage(err, "failed to get logs from near head cache") - return nil, tracker.withError(err) - } +// nearHeadLogConflict carries fixed diagnostic reasons only. It is consumed by +// the retry loop; clients receive only ErrInconsistentCanonicalView on exhaustion. +type nearHeadLogConflict struct { + reason string + boundary uint64 +} - if ethLogs == nil { - logs, err := c.RpcEthClient.Logs(filter) - if err != nil { - return nil, tracker.withError(err) - } +func (e *nearHeadLogConflict) Error() string { return e.reason } - tracker.withFullNodeHit(true) - return logs, nil - } +// nearHeadLogReader keeps the metadata/log read sequence explicit and testable. +// Cache-internal snapshot and ownership guarantees remain data-cache's concern. +type nearHeadLogReader interface { + GetBlock(cacheTypes.BlockHashOrNumber, bool) nearhead.EthCacheResultWithCoverage[*web3Types.Block] + GetLogsByBlockRange(uint64, uint64, nearhead.FilterOpt) (*nearhead.EthLogs, error) +} - if ethLogs.FromBlock < fromBlock || ethLogs.ToBlock > toBlock { - err = errors.New("bad event logs from near head cache: unexpected block range") - return nil, tracker.withError(err) +// readNearHeadLogs scans logs at most once. A lightweight metadata lookup +// captures the cached end first, so a partial hit needs no second log scan. +func readNearHeadLogs( + cache nearHeadLogReader, + fromBlock, toBlock uint64, + opt nearhead.FilterOpt, +) (*nearhead.EthLogs, common.Hash, error) { + if fromBlock > toBlock { + return nil, common.Hash{}, errors.New("failed to get logs from near head cache: invalid block range params") + } + // In the pinned cache, true returns the cached block directly; false copies + // it and builds a transaction-hash list. We only read metadata, never mutate. + probe := cache.GetBlock(cacheTypes.BlockHashOrNumberWithNumber(toBlock), true) + start, end := probe.Coverage[0], probe.Coverage[1] // end is exclusive + if start >= end || toBlock < start || fromBlock >= end { + return nil, common.Hash{}, nil + } + end = min(toBlock, end-1) + ref := cacheTypes.BlockHashOrNumberWithNumber(end) + before := probe.Result + if end != toBlock { + before = cache.GetBlock(ref, true).Result + } + // Copy the hash before the scan; the cache returns a shared block pointer. + var hash common.Hash + if before != nil { + hash = before.Hash + } + snapshot, err := cache.GetLogsByBlockRange(fromBlock, toBlock, opt) + if err != nil { + return nil, common.Hash{}, errors.WithMessage(err, "failed to get logs from near head cache") + } + if snapshot == nil { + return nil, common.Hash{}, &nearHeadLogConflict{"near-head coverage disappeared", end} } + if snapshot.FromBlock < fromBlock || snapshot.ToBlock > toBlock || snapshot.FromBlock > snapshot.ToBlock { + return nil, common.Hash{}, errors.New("bad event logs from near head cache: unexpected block range") + } + if snapshot.FromBlock == fromBlock && snapshot.ToBlock == toBlock { + // A complete result has one source and needs no boundary validation. + return snapshot, common.Hash{}, nil + } + if before == nil || hash == (common.Hash{}) { + return nil, common.Hash{}, &nearHeadLogConflict{"near-head boundary unavailable", end} + } + after := cache.GetBlock(ref, true).Result + if snapshot.ToBlock != end || after == nil || after.Hash != hash { + return nil, common.Hash{}, &nearHeadLogConflict{"near-head cache changed while reading logs", end} + } + // Start eviction can extend the node prefix. The actual returned range is + // authoritative as long as its end still matches the captured boundary. + return snapshot, hash, nil +} +// logsByBlockRangeOnce owns all data for one attempt. A conflict discards both +// node segments and the cache result; no logs carry over to the next attempt. +func (c ethCacheClient) logsByBlockRangeOnce( + filter web3Types.FilterQuery, + fromBlock, toBlock uint64, + tracker *rpcCacheTracker, +) ([]web3Types.Log, error) { + opt := nearhead.FilterOpt{Addresses: filter.Addresses, Topics: filter.Topics} + snapshot, hash, err := readNearHeadLogs(c.nearhead, fromBlock, toBlock, opt) + if err != nil { + return nil, err + } + if snapshot == nil { + tracker.withFullNodeHit(true) + return c.fetchLogsFromFullNode(filter, fromBlock, toBlock) + } tracker.withNearHeadHit(true) - - // Stitches together logs from the near head cache and full node. - // Handle the block range before the near head cached range. - if fromBlock < ethLogs.FromBlock { - logs, err := c.fetchLogsFromFullNode(filter, fromBlock, ethLogs.FromBlock-1) + if snapshot.FromBlock == fromBlock && snapshot.ToBlock == toBlock { + return snapshot.Logs, nil + } + end := snapshot.ToBlock + + // A block hash commits to its ancestry. For an internally coherent cache + // and node view, matching the cached end aligns both the older node prefix + // and the newer node suffix. Check that same height around BOTH node reads; + // no start/parent-hash or query-upper-bound lookup is needed here. This is + // optimistic alignment, not a node snapshot or protection against ABA or + // independently inconsistent responses within the node's Logs RPC. + if err := c.checkNearHeadLogBoundary(end, hash); err != nil { + return nil, err + } + logs := make([]web3Types.Log, 0, len(snapshot.Logs)) + if fromBlock < snapshot.FromBlock { + tracker.withFullNodeHit(true) + prefix, err := c.fetchLogsFromFullNode(filter, fromBlock, snapshot.FromBlock-1) if err != nil { - return nil, tracker.withError(err) + return nil, err } - - res = append(logs, res...) - tracker.withFullNodeHit(true) + logs = append(logs, prefix...) } - - // Append logs from the near head cache. - res = append(res, ethLogs.Logs...) - - // Handle the block range after the near head cached range. - if toBlock > ethLogs.ToBlock { - logs, err := c.fetchLogsFromFullNode(filter, ethLogs.ToBlock+1, toBlock) + logs = append(logs, snapshot.Logs...) + if toBlock > end { + tracker.withFullNodeHit(true) + suffix, err := c.fetchLogsFromFullNode(filter, end+1, toBlock) if err != nil { - return nil, tracker.withError(err) + return nil, err } - - res = append(res, logs...) - tracker.withFullNodeHit(true) + logs = append(logs, suffix...) } + if err := c.checkNearHeadLogBoundary(end, hash); err != nil { + return nil, err + } + return logs, nil +} - return res, nil +func (c ethCacheClient) checkNearHeadLogBoundary(number uint64, hash common.Hash) error { + block, err := c.RpcEthClient.BlockByNumber(web3Types.BlockNumber(number), false) + if err != nil { + return err + } + if block == nil || block.Number == nil || !block.Number.IsUint64() || + block.Number.Uint64() != number || block.Hash != hash { + return &nearHeadLogConflict{"node boundary differs from near-head cache", number} + } + return nil } func (c ethCacheClient) fetchLogsFromFullNode( @@ -805,6 +943,9 @@ func (c ethCacheClient) LazyTransactionByBlockHashAndIndex( return res, err }, func() (cacheTypes.Lazy[*web3Types.TransactionDetail], error) { + if uint64(index) > math.MaxUint32 { + return cacheTypes.Lazy[*web3Types.TransactionDetail]{}, nil + } return c.dataCache.GetTransactionByIndex(cacheBlockRef, uint32(index)) }, func() (cacheTypes.Lazy[*web3Types.TransactionDetail], error) { @@ -852,7 +993,7 @@ func (c ethCacheClient) LazyTransactionByBlockNumberAndIndex( return res, err }, func() (res cacheTypes.Lazy[*web3Types.TransactionDetail], err error) { - if blockRef.shouldReadFromDataCache(nearheadCoverage) { + if uint64(index) <= math.MaxUint32 && blockRef.shouldReadFromDataCache(nearheadCoverage) { res, err = c.dataCache.GetTransactionByIndex(blockRef.cacheRef, uint32(index)) } return res, err @@ -889,20 +1030,29 @@ func (c traceCacheClient) LazyTrace( if len(indexes) == 0 { return res, nil } - + // Conflux eSpace uses the first index as an offset in transaction traces. + index := indexes[0] tracker := newRPCCacheTracker(rpcMethodTraceGet) defer tracker.report() return readThroughLazyCacheChain( tracker, func() (res cacheTypes.Lazy[*web3Types.LocalizedTrace], err error) { - if trace := c.nearhead.GetTrace(transactionHash, indexes[0]); trace != nil { + // The pinned cache converts index to int for its bounds check. + if index > uint(math.MaxInt) { + return res, nil + } + if trace := c.nearhead.GetTrace(transactionHash, index); trace != nil { res, err = cacheTypes.NewLazy(trace) } return res, err }, func() (cacheTypes.Lazy[*web3Types.LocalizedTrace], error) { - return c.dataCache.GetTrace(transactionHash, indexes[0]) + // The pinned gRPC client narrows index to uint32 on the wire. + if uint64(index) > math.MaxUint32 { + return cacheTypes.Lazy[*web3Types.LocalizedTrace]{}, nil + } + return c.dataCache.GetTrace(transactionHash, index) }, func() (cacheTypes.Lazy[*web3Types.LocalizedTrace], error) { hexIndexes := make([]hexutil.Uint, len(indexes)) diff --git a/util/rpc/cached_client_eth_test.go b/util/rpc/cached_client_eth_test.go new file mode 100644 index 00000000..0d6fa9ea --- /dev/null +++ b/util/rpc/cached_client_eth_test.go @@ -0,0 +1,199 @@ +package rpc + +import ( + "context" + "encoding/json" + "math" + "math/big" + "testing" + + "github.com/Conflux-Chain/confura-data-cache/nearhead" + cacheRpc "github.com/Conflux-Chain/confura-data-cache/rpc" + ct "github.com/Conflux-Chain/confura-data-cache/types" + "github.com/Conflux-Chain/go-conflux-util/blockchain/sync/evm" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/openweb3/go-rpc-provider/interfaces" + "github.com/openweb3/web3go/client" + wt "github.com/openweb3/web3go/types" + "github.com/stretchr/testify/require" +) + +type fixtureProvider struct { + interfaces.Provider + call func(context.Context, interface{}, string, ...interface{}) error +} + +func (p *fixtureProvider) CallContext(ctx context.Context, out interface{}, method string, args ...interface{}) error { + return p.call(ctx, out, method, args...) +} + +type fixtureResolver struct{} + +func (fixtureResolver) Resolve(n wt.BlockNumber) (uint64, error) { + if n >= 0 { + return uint64(n), nil + } + return 100, nil +} +func fixtureNearhead() *EthNearHeadCache { + return &EthNearHeadCache{EthCache: nearhead.NewEthCache(nearhead.Config{MaxMemory: 1 << 20, MaxLogs: 100})} +} +func fixtureBlock(n int64, h common.Hash, topics []common.Hash) evm.BlockData { + tx := common.HexToHash("0x1") + return evm.BlockData{Block: &wt.Block{Number: big.NewInt(n), Difficulty: big.NewInt(0), Hash: h, Transactions: *wt.NewTxOrHashListByTxs([]wt.TransactionDetail{{Hash: tx}})}, Receipts: []*wt.Receipt{{TransactionHash: tx, BlockHash: h, Logs: []*wt.Log{{BlockNumber: uint64(n), BlockHash: h, TxHash: tx, Topics: topics}}}}} +} + +type fixtureIndexCache struct { + cacheRpc.Interface + calls int +} + +func (c *fixtureIndexCache) GetTransactionByIndex(_ ct.BlockHashOrNumber, i uint32) (ct.Lazy[*wt.TransactionDetail], error) { + c.calls++ + return ct.NewLazy(&wt.TransactionDetail{Hash: common.HexToHash("0xa")}) +} + +func TestLargeTransactionIndexBypassesNarrowCache(t *testing.T) { + d := &fixtureIndexCache{Interface: cacheRpc.NotFoundImpl} + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + require.Equal(t, hexutil.Uint(uint64(1)<<32), args[1]) + return json.Unmarshal([]byte("null"), out) + }} + n := fixtureNearhead() + require.NoError(t, n.Put(fixtureBlock(100, common.Hash{}, nil))) + c := ethCacheClient{RpcEthClient: client.NewRpcEthClient(p), cacheClientContext: cacheClientContext{nearhead: n, dataCache: d, resolver: fixtureResolver{}}} + a, err := c.LazyTransactionByBlockHashAndIndex(common.Hash{}, uint(uint64(1)<<32)) + require.NoError(t, err) + require.True(t, a.IsEmptyOrNull()) + b, err := c.LazyTransactionByBlockNumberAndIndex(1, uint(uint64(1)<<32)) + require.NoError(t, err) + require.True(t, b.IsEmptyOrNull()) + require.Zero(t, d.calls) +} + +type fixtureTraceCache struct { + cacheRpc.Interface + indexes []uint + bulkCalls int +} + +func (c *fixtureTraceCache) GetTrace(_ common.Hash, index uint) (ct.Lazy[*wt.LocalizedTrace], error) { + c.indexes = append(c.indexes, index) + // Model the pinned gRPC client's uint32 wire conversion. The caller must + // skip this API when the original index does not fit. + index = uint(uint32(index)) + traces := fixtureTransactionTraces() + if index >= uint(len(traces)) { + return ct.NewLazy((*wt.LocalizedTrace)(nil)) + } + return ct.NewLazy(&traces[index]) +} + +func (c *fixtureTraceCache) GetTransactionTraces(common.Hash) (ct.Lazy[[]wt.LocalizedTrace], error) { + c.bulkCalls++ + return ct.NewLazy(fixtureTransactionTraces()) +} + +func fixtureTransactionTraces() []wt.LocalizedTrace { + var traces []wt.LocalizedTrace + txHash := common.HexToHash("0x1") + for _, path := range [][]uint{{}, {0}, {0, 1}, {0, 2}} { + var trace wt.LocalizedTrace + _ = json.Unmarshal([]byte(`{"type":"call","action":{"from":"0x0000000000000000000000000000000000000000","to":"0x0000000000000000000000000000000000000000","gas":"0x0","input":"0x","value":"0x0","callType":"call"},"traceAddress":[],"subtraces":0,"blockNumber":100,"blockHash":"0x0000000000000000000000000000000000000000000000000000000000000001"}`), &trace) + trace.TraceAddress = path + trace.TransactionHash = &txHash + traces = append(traces, trace) + } + return traces +} + +// Conflux eSpace selects the first flat index and returns null for empty input: +// https://github.com/Conflux-Chain/conflux-rust/blob/2cb16583060d7a569271adb59cc1a9cd92dfd447/crates/rpc/rpc-eth-impl/src/trace.rs#L305-L319 +func TestTraceGetMatchesConfluxSemantics(t *testing.T) { + traces, txHash := fixtureTransactionTraces(), common.HexToHash("0x1") + for _, source := range []string{"nearhead", "data-cache", "fullnode"} { + t.Run(source, func(t *testing.T) { + for _, tc := range []struct { + name string + indexes []uint + want int // -1 means null + }{ + {"empty", []uint{}, -1}, + {"nil", nil, -1}, + {"first", []uint{0}, 0}, + {"second", []uint{1}, 1}, + {"third", []uint{2}, 2}, + {"ignore suffix one", []uint{0, 1}, 0}, + {"ignore suffix two", []uint{0, 2}, 0}, + {"ignore suffix after second", []uint{1, 0}, 1}, + {"out of range", []uint{99}, -1}, + {"maximum protocol index", []uint{math.MaxUint32}, -1}, + {"large index", []uint{uint(uint64(1) << 32)}, -1}, + {"maximum signed index", []uint{uint(math.MaxInt)}, -1}, + {"signed overflow", []uint{uint(math.MaxInt) + 1}, -1}, + {"maximum index", []uint{^uint(0)}, -1}, + } { + t.Run(tc.name, func(t *testing.T) { + n := fixtureNearhead() + var dataCache cacheRpc.Interface = cacheRpc.NotFoundImpl + var traceCache *fixtureTraceCache + switch source { + case "nearhead": + block := fixtureBlock(100, common.HexToHash("0x1"), nil) + block.Traces = traces + require.NoError(t, n.Put(block)) + case "data-cache": + traceCache = &fixtureTraceCache{Interface: cacheRpc.NotFoundImpl} + dataCache = traceCache + } + calls := 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + calls++ + require.Equal(t, "trace_get", method) + require.Equal(t, txHash, args[0]) + indices := args[1].([]hexutil.Uint) + require.Len(t, indices, len(tc.indexes)) + for i := range indices { + require.Equal(t, hexutil.Uint(tc.indexes[i]), indices[i]) + } + // Model the pinned node implementation, independently of TraceAddress. + var trace *wt.LocalizedTrace + if len(indices) > 0 && uint(indices[0]) < uint(len(traces)) { + trace = &traces[uint(indices[0])] + } + raw, err := json.Marshal(trace) + if err != nil { + return err + } + return json.Unmarshal(raw, out) + }} + c := traceCacheClient{RpcTraceClient: client.NewRpcTraceClient(p), cacheClientContext: cacheClientContext{nearhead: n, dataCache: dataCache, resolver: fixtureResolver{}}} + lazy, err := c.LazyTrace(txHash, tc.indexes) + require.NoError(t, err) + got, err := lazy.Load() + require.NoError(t, err) + if tc.want < 0 { + require.Nil(t, got) + } else { + require.NotNil(t, got) + require.Equal(t, traces[tc.want].TraceAddress, got.TraceAddress) + } + wantCalls := 0 + if len(tc.indexes) > 0 && (source == "fullnode" || tc.want < 0) { + wantCalls = 1 + } + require.Equal(t, wantCalls, calls) + if traceCache != nil { + require.Zero(t, traceCache.bulkCalls, "trace_get must not fetch all transaction traces") + if len(tc.indexes) > 0 && uint64(tc.indexes[0]) <= math.MaxUint32 { + require.Equal(t, []uint{tc.indexes[0]}, traceCache.indexes) + } else { + require.Empty(t, traceCache.indexes, "unsafe index must bypass data-cache") + } + } + }) + } + }) + } +} diff --git a/util/rpc/handlers/ip.go b/util/rpc/handlers/ip.go index 29b72060..082bef79 100644 --- a/util/rpc/handlers/ip.go +++ b/util/rpc/handlers/ip.go @@ -22,7 +22,7 @@ type ipRange struct { // inRange - check to see if a given ip address is within a range given func inRange(r ipRange, ipAddress net.IP) bool { // strcmp type byte comparison - if bytes.Compare(ipAddress, r.start) >= 0 && bytes.Compare(ipAddress, r.end) < 0 { + if bytes.Compare(ipAddress, r.start) >= 0 && bytes.Compare(ipAddress, r.end) <= 0 { return true } return false @@ -84,12 +84,19 @@ func GetIPAddress(r *http.Request) string { // bad address, go to next continue } - return ip + return realIP.String() } } - if idx := strings.Index(r.RemoteAddr, ":"); idx != -1 { - return r.RemoteAddr[:idx] + host := r.RemoteAddr + if h, _, err := net.SplitHostPort(host); err == nil { + host = h + } + if ip := net.ParseIP(host); ip != nil { + return ip.String() + } + if host != r.RemoteAddr { + return host } return r.RemoteAddr diff --git a/util/rpc/handlers/ip_test.go b/util/rpc/handlers/ip_test.go new file mode 100644 index 00000000..229eba4a --- /dev/null +++ b/util/rpc/handlers/ip_test.go @@ -0,0 +1,23 @@ +package handlers + +import ( + "net/http" + "testing" +) + +func TestRemoteIPCanonicalization(t *testing.T) { + for _, tc := range []struct{ remote, forwarded, want string }{ + {"[2001:db8::1]:1234", "", "2001:db8::1"}, + {"[2001:db8::2]:1234", "", "2001:db8::2"}, + {"2001:db8::1", "", "2001:db8::1"}, + {"192.0.2.1:456", "", "192.0.2.1"}, + {"192.0.2.1:456", "2001:4860:0000:0000:0000:0000:0000:0001", "2001:4860::1"}, + {"192.0.2.1:456", "10.255.255.255", "192.0.2.1"}, + } { + r := &http.Request{RemoteAddr: tc.remote, Header: make(http.Header)} + r.Header.Set("X-Forwarded-For", tc.forwarded) + if got := GetIPAddress(r); got != tc.want { + t.Errorf("%+v: %q", tc, got) + } + } +} diff --git a/util/rpc/middlewares/allowlists.go b/util/rpc/middlewares/allowlists.go index 438b41bf..f3822c75 100644 --- a/util/rpc/middlewares/allowlists.go +++ b/util/rpc/middlewares/allowlists.go @@ -20,6 +20,7 @@ func Allowlists(next rpc.HandleCallMsgFunc) rpc.HandleCallMsgFunc { aclCtx := acl.Context{ Context: ctx, RpcMethod: msg.Method, + RawParams: msg.Params, } if connH, ok := ctx.Value("handler").(*rpc.ConnHandler); ok { diff --git a/util/rpc/middlewares/metrics.go b/util/rpc/middlewares/metrics.go index d8ec3afd..aa351a55 100644 --- a/util/rpc/middlewares/metrics.go +++ b/util/rpc/middlewares/metrics.go @@ -2,8 +2,6 @@ package middlewares import ( "context" - "fmt" - "strings" "time" "github.com/Conflux-Chain/confura/util/metrics" @@ -28,10 +26,10 @@ func Metrics(next rpc.HandleCallMsgFunc) rpc.HandleCallMsgFunc { start := time.Now() resp := next(ctx, msg) - metricMethod := msg.Method - if resp.Error != nil && isMethodNotFoundByError(msg.Method, resp.Error) { - metricMethod = "method_not_found" + if resp == nil { + return nil } + metricMethod := registeredMetricMethod(ctx, msg) // collect rpc QPS/latency, response size etc. space, _ := handlers.GetNamespaceFromContext(ctx) @@ -61,11 +59,6 @@ func getTrafficSourceFromContext(ctx context.Context) (source string) { return source } -func isMethodNotFoundByError(method string, err error) bool { - subPattern := fmt.Sprintf("the method %s does not exist/is not available", method) - return strings.Contains(err.Error(), subPattern) -} - func unwrapJsonError(jsErr *rpc.JsonError) error { if jsErr == nil { return nil @@ -77,3 +70,26 @@ func unwrapJsonError(jsErr *rpc.JsonError) error { return jsErr } + +// Classify against registered callbacks, independently of early routing/auth +// errors. Never use arbitrary client-supplied method names as metric labels. +func registeredMetricMethod(ctx context.Context, msg *rpc.JsonRpcMessage) string { + switch msg.Method { + case "eth_subscribe", "eth_unsubscribe", "cfx_subscribe", "cfx_unsubscribe": + return msg.Method + } + if h, ok := ctx.Value("handler").(*rpc.ConnHandler); ok { + // The provider checks registration before parsing arguments. Probe with + // no params so classification never decodes the request payload again. + probe := *msg + probe.Params = nil + _, err := h.ParseCallArguments(&probe) + if err == nil { + return msg.Method + } + if e, ok := err.(rpc.Error); ok && e.ErrorCode() != -32601 { + return msg.Method + } + } + return "method_not_found" +} diff --git a/util/rpc/middlewares/metrics_test.go b/util/rpc/middlewares/metrics_test.go new file mode 100644 index 00000000..263bd71f --- /dev/null +++ b/util/rpc/middlewares/metrics_test.go @@ -0,0 +1,209 @@ +package middlewares + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/ethereum/go-ethereum/common/hexutil" + rpc "github.com/openweb3/go-rpc-provider" + web3Types "github.com/openweb3/web3go/types" + "github.com/stretchr/testify/require" +) + +type MetricsFixture struct{} + +func (*MetricsFixture) Ping() bool { return true } + +var metricArgumentDecodes atomic.Int32 + +type MetricArgument string + +func (a *MetricArgument) UnmarshalJSON(data []byte) error { + metricArgumentDecodes.Add(1) + return json.Unmarshal(data, (*string)(a)) +} + +func (*MetricsFixture) Required(MetricArgument) bool { return true } +func (*MetricsFixture) Optional(*MetricArgument) bool { return true } +func (*MetricsFixture) RawTransaction(hexutil.Bytes) bool { return true } +func (*MetricsFixture) Call(web3Types.CallRequest) bool { return true } + +func newMetricsFixtureServer(tb testing.TB) *rpc.Server { + tb.Helper() + server := rpc.NewServer() + tb.Cleanup(server.Stop) + if err := server.RegisterName("fixture", &MetricsFixture{}); err != nil { + tb.Fatal(err) + } + return server +} + +// Capture a real provider handler without opening a listening socket. Each +// global hook is gated by a unique context key and ignores subsequent tests. +func metricsFixtureContext(tb testing.TB) context.Context { + tb.Helper() + key := new(int) + captured := make(chan context.Context, 1) + rpc.HookHandleCallMsg(func(next rpc.HandleCallMsgFunc) rpc.HandleCallMsgFunc { + return func(ctx context.Context, msg *rpc.JsonRpcMessage) *rpc.JsonRpcMessage { + if ctx.Value(key) != nil { + captured <- ctx + } + return next(ctx, msg) + } + }) + server := newMetricsFixtureServer(tb) + req := httptest.NewRequest("POST", "/", strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"fixture_ping"}`)) + req.Header.Set("Content-Type", "application/json") + server.ServeHTTP(httptest.NewRecorder(), req.WithContext(context.WithValue(req.Context(), key, true))) + return <-captured +} + +func TestMetricMethodUsesRegisteredCallbacks(t *testing.T) { + // The provider exposes global hooks; scope the test hook to a context key. + type fixtureKey struct{} + rpc.HookHandleCallMsg(func(next rpc.HandleCallMsgFunc) rpc.HandleCallMsgFunc { + return func(ctx context.Context, msg *rpc.JsonRpcMessage) *rpc.JsonRpcMessage { + if want, ok := ctx.Value(fixtureKey{}).(string); ok { + if got := registeredMetricMethod(ctx, msg); got != want { + t.Errorf("%s: got %s, want %s", msg.Method, got, want) + } + } + return next(ctx, msg) + } + }) + server := rpc.NewServer() + defer server.Stop() + if err := server.RegisterName("fixture", &MetricsFixture{}); err != nil { + t.Fatal(err) + } + for i := 0; i < 200; i++ { + method, want := fmt.Sprintf("untrusted_%d", i), "method_not_found" + if i == 0 { + method, want = "fixture_ping", "fixture_ping" + } + req := httptest.NewRequest("POST", "/", strings.NewReader(fmt.Sprintf(`{"jsonrpc":"2.0","id":1,"method":%q,"params":[]}`, method))) + req.Header.Set("Content-Type", "application/json") + req = req.WithContext(context.WithValue(req.Context(), fixtureKey{}, want)) + server.ServeHTTP(httptest.NewRecorder(), req) + } +} + +func TestMetricMethodDoesNotDecodeParameters(t *testing.T) { + ctx := metricsFixtureContext(t) + for _, tc := range []struct { + method, params, want string + }{ + {"fixture_required", `["payload"]`, "fixture_required"}, + {"fixture_required", `[123]`, "fixture_required"}, + {"fixture_required", `not JSON`, "fixture_required"}, + {"fixture_required", `[]`, "fixture_required"}, + {"fixture_optional", `["payload"]`, "fixture_optional"}, + {"fixture_optional", `[]`, "fixture_optional"}, + {"fixture_ping", `[]`, "fixture_ping"}, + {"fixture_unknown", `["payload"]`, "method_not_found"}, + {"eth_subscribe", `["logs",{}]`, "eth_subscribe"}, + {"eth_unsubscribe", `["0x1"]`, "eth_unsubscribe"}, + {"cfx_subscribe", `["logs",{}]`, "cfx_subscribe"}, + {"cfx_unsubscribe", `["0x1"]`, "cfx_unsubscribe"}, + } { + t.Run(tc.method+"/"+tc.params, func(t *testing.T) { + msg := &rpc.JsonRpcMessage{Version: "2.0", ID: json.RawMessage(`1`), Method: tc.method, Params: json.RawMessage(tc.params)} + original := *msg + original.Params = bytes.Clone(msg.Params) + metricArgumentDecodes.Store(0) + require.Equal(t, tc.want, registeredMetricMethod(ctx, msg)) + require.Zero(t, metricArgumentDecodes.Load(), "classification must not invoke argument decoders") + require.Equal(t, original, *msg, "classification must leave the message unchanged") + }) + } +} + +func TestMetricsDecodesOnlyForDispatch(t *testing.T) { + key := new(int) + rpc.HookHandleCallMsg(func(next rpc.HandleCallMsgFunc) rpc.HandleCallMsgFunc { + return func(ctx context.Context, msg *rpc.JsonRpcMessage) *rpc.JsonRpcMessage { + mode, active := ctx.Value(key).(string) + if !active { + return next(ctx, msg) + } + callNext := next + if mode == "routing failure" { + callNext = func(_ context.Context, msg *rpc.JsonRpcMessage) *rpc.JsonRpcMessage { + return msg.ErrorResponse(errors.New("fixture route unavailable")) + } + } + return Metrics(callNext)(ctx, msg) + } + }) + server := newMetricsFixtureServer(t) + for _, mode := range []string{"success", "routing failure"} { + t.Run(mode, func(t *testing.T) { + metricArgumentDecodes.Store(0) + payload := strings.Repeat("a", 128<<10) + body := `{"jsonrpc":"2.0","id":1,"method":"fixture_required","params":["` + payload + `"]}` + req := httptest.NewRequest("POST", "/", strings.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + resp := httptest.NewRecorder() + server.ServeHTTP(resp, req.WithContext(context.WithValue(req.Context(), key, mode))) + var msg rpc.JsonRpcMessage + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &msg)) + if mode == "success" { + require.Nil(t, msg.Error) + require.JSONEq(t, `true`, string(msg.Result)) + require.EqualValues(t, 1, metricArgumentDecodes.Load()) + } else { + require.EqualError(t, msg.Error, "fixture route unavailable") + require.Zero(t, metricArgumentDecodes.Load()) + } + }) + } +} + +type metricLookupContext struct { + context.Context + lookups int +} + +func (c *metricLookupContext) Value(key interface{}) interface{} { + if key == "handler" { + c.lookups++ + } + return c.Context.Value(key) +} + +func TestMetricsSkipsClassificationForNilResponse(t *testing.T) { + ctx := &metricLookupContext{Context: context.Background()} + next := func(context.Context, *rpc.JsonRpcMessage) *rpc.JsonRpcMessage { return nil } + require.Nil(t, Metrics(next)(ctx, &rpc.JsonRpcMessage{Method: "fixture_required"})) + require.Zero(t, ctx.lookups) +} + +func BenchmarkRegisteredMetricMethod(b *testing.B) { + ctx := metricsFixtureContext(b) + for _, size := range []int{32, 128 << 10} { + hex := "0x" + strings.Repeat("ab", size) + for _, tc := range []struct{ name, method, params string }{ + {"raw", "fixture_rawTransaction", `["` + hex + `"]`}, + {"call", "fixture_call", `[{"data":"` + hex + `"}]`}, + } { + b.Run(fmt.Sprintf("%s/%d", tc.name, size), func(b *testing.B) { + msg := &rpc.JsonRpcMessage{Method: tc.method, Params: json.RawMessage(tc.params)} + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if registeredMetricMethod(ctx, msg) != tc.method { + b.Fatal("registered method misclassified") + } + } + }) + } + } +} diff --git a/util/rpc/nearhead_test.go b/util/rpc/nearhead_test.go new file mode 100644 index 00000000..bf23ad8c --- /dev/null +++ b/util/rpc/nearhead_test.go @@ -0,0 +1,692 @@ +package rpc + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + "github.com/Conflux-Chain/confura-data-cache/nearhead" + cacheRpc "github.com/Conflux-Chain/confura-data-cache/rpc" + cacheTypes "github.com/Conflux-Chain/confura-data-cache/types" + "github.com/ethereum/go-ethereum/common" + "github.com/openweb3/web3go/client" + wt "github.com/openweb3/web3go/types" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" +) + +func nearHeadTestReply(out, value interface{}) error { + raw, err := json.Marshal(value) + if err != nil { + return err + } + return json.Unmarshal(raw, out) +} + +func nearHeadTestHeader(number int64, hash common.Hash) *wt.Block { + block := fixtureBlock(number, hash, nil).Block + block.Transactions = *wt.NewTxOrHashListByHashes(nil) + return block +} + +func nearHeadTestClient(n *EthNearHeadCache, p *fixtureProvider) ethCacheClient { + return ethCacheClient{RpcEthClient: client.NewRpcEthClient(p), cacheClientContext: cacheClientContext{nearhead: n, dataCache: cacheRpc.NotFoundImpl, resolver: fixtureResolver{}}} +} + +// A complete cache hit must not depend on a live node. Mixed reads check only +// the cached end, even when the node supplies both a prefix and a suffix. +func TestNearHeadLogsCoverage(t *testing.T) { + for _, tc := range []struct { + name string + from, to wt.BlockNumber + empty bool + events []string + blocks []uint64 + }{ + {name: "complete", from: 100, to: 100, blocks: []uint64{100}}, + {name: "complete empty", from: 100, to: 100, empty: true}, + {name: "miss", from: 101, to: 101, events: []string{"logs:101-101"}, blocks: []uint64{101}}, + {name: "left", from: 99, to: 100, events: []string{"header:100", "logs:99-99", "header:100"}, blocks: []uint64{99, 100}}, + {name: "right", from: 100, to: 101, events: []string{"header:100", "logs:101-101", "header:100"}, blocks: []uint64{100, 101}}, + {name: "both", from: 99, to: 101, events: []string{"header:100", "logs:99-99", "logs:101-101", "header:100"}, blocks: []uint64{99, 100, 101}}, + {name: "both empty", from: 99, to: 101, empty: true, events: []string{"header:100", "logs:99-99", "logs:101-101", "header:100"}}, + } { + t.Run(tc.name, func(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + block := fixtureBlock(100, hash, nil) + if tc.empty { + block.Receipts[0].Logs = nil + } + require.NoError(t, n.Put(block)) + var events []string + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + switch method { + case "eth_getBlockByNumber": + number := args[0].(wt.BlockNumber) + require.Equal(t, wt.BlockNumber(100), number, "no start, parent, or request-upper-bound lookup") + events = append(events, fmt.Sprintf("header:%d", number)) + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + case "eth_getLogs": + fq := args[0].(wt.FilterQuery) + events = append(events, fmt.Sprintf("logs:%d-%d", *fq.FromBlock, *fq.ToBlock)) + logs := []wt.Log{} + if !tc.empty { + logs = append(logs, wt.Log{BlockNumber: uint64(*fq.FromBlock)}) + } + return nearHeadTestReply(out, logs) + default: + t.Fatalf("unexpected RPC %s", method) + return nil + } + }} + logs, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &tc.from, ToBlock: &tc.to}) + require.NoError(t, err) + require.NotNil(t, logs) + var blocks []uint64 + for _, entry := range logs { + blocks = append(blocks, entry.BlockNumber) + } + require.Equal(t, tc.blocks, blocks) + require.Equal(t, tc.events, events) + }) + } +} + +// Reorg either node segment between the two boundary checks. A fresh attempt +// must replace every old log, including a prefix already read before the reorg. +func TestNearHeadLogsRetryWholeStitch(t *testing.T) { + for _, reorgAt := range []uint64{99, 101} { + t.Run(fmt.Sprint(reorgAt), func(t *testing.T) { + n := fixtureNearhead() + a, b := common.HexToHash("0xa100"), common.HexToHash("0xb100") + require.NoError(t, n.Put(fixtureBlock(100, a, nil))) + canonical, switched, headers := a, false, 0 + var reads []uint64 + p := &fixtureProvider{call: func(ctx context.Context, out interface{}, method string, args ...interface{}) error { + switch method { + case "eth_getBlockByNumber": + headers++ + require.Equal(t, wt.BlockNumber(100), args[0]) + _, bounded := ctx.Deadline() + require.False(t, bounded, "keep the original client context on every attempt") + return nearHeadTestReply(out, nearHeadTestHeader(100, canonical)) + case "eth_getLogs": + fq := args[0].(wt.FilterQuery) + number := uint64(*fq.FromBlock) + reads = append(reads, number) + servedHash := canonical + if number == reorgAt && !switched { + switched, canonical = true, b + require.Equal(t, 1, n.Pop(100)) + require.NoError(t, n.Put(fixtureBlock(100, b, nil))) + } + return nearHeadTestReply(out, []wt.Log{{BlockNumber: number, BlockHash: servedHash}}) + default: + t.Fatalf("unexpected RPC %s", method) + return nil + } + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(101) + logs, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.NoError(t, err) + require.Len(t, logs, 3) + for i, entry := range logs { + require.Equal(t, uint64(99+i), entry.BlockNumber) + require.Equal(t, b, entry.BlockHash, "discard the failed attempt's logs") + } + require.Equal(t, 4, headers) + require.Equal(t, []uint64{99, 101, 99, 101}, reads) + }) + } +} + +// The retry window admits attempts; it does not cancel an RPC or reject a +// coherent result that completes after the window. Late ordinary errors survive. +func TestNearHeadLogsRetryBudget(t *testing.T) { + for _, outcome := range []string{"conflict", "success", "RPC error"} { + t.Run(outcome, func(t *testing.T) { + logger := logrus.StandardLogger() + output, level := logger.Out, logger.GetLevel() + var diagnostics bytes.Buffer + logger.SetOutput(&diagnostics) + logger.SetLevel(logrus.InfoLevel) + t.Cleanup(func() { logger.SetOutput(output); logger.SetLevel(level) }) + + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + var firstConflict time.Time + headers, reads := 0, 0 + failure := fmt.Errorf("node unavailable") + p := &fixtureProvider{call: func(ctx context.Context, out interface{}, method string, args ...interface{}) error { + _, bounded := ctx.Deadline() + require.False(t, bounded, "do not replace configured RPC context/timeouts") + if method == "eth_getLogs" { + reads++ + return nearHeadTestReply(out, []wt.Log{}) + } + require.Equal(t, "eth_getBlockByNumber", method) + headers++ + if headers == 1 { + firstConflict = time.Now() + return nearHeadTestReply(out, nil) + } + if headers < 4 { + return nearHeadTestReply(out, nil) + } + if headers == 4 { + require.Less(t, time.Since(firstConflict), nearHeadLogRetryTimeout, "admit the round inside the window") + time.Sleep(time.Until(firstConflict.Add(nearHeadLogRetryTimeout + 50*time.Millisecond))) + switch outcome { + case "conflict": + return nearHeadTestReply(out, nil) + case "RPC error": + return failure + } + } + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + c := nearHeadTestClient(n, p) + c.nodeName = "https://secret@node.invalid/private-api-key" + logs, err := c.Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.GreaterOrEqual(t, time.Since(firstConflict), nearHeadLogRetryTimeout) + switch outcome { + case "conflict": + require.Same(t, ErrInconsistentCanonicalView, err) + require.EqualError(t, err, "inconsistent canonical view") + require.Nil(t, logs) + require.Equal(t, 4, headers, "no new round after expiry") + require.Zero(t, reads) + require.Equal(t, 1, bytes.Count(diagnostics.Bytes(), []byte("level=info"))) + require.Contains(t, diagnostics.String(), "node boundary differs from near-head cache") + require.Contains(t, diagnostics.String(), "boundary=100") + case "success": + require.NoError(t, err) + require.Len(t, logs, 1) + require.Equal(t, 5, headers) + require.Equal(t, 1, reads) + require.Empty(t, diagnostics.String()) + case "RPC error": + require.ErrorIs(t, err, failure) + require.Nil(t, logs) + require.Equal(t, 4, headers) + require.Zero(t, reads) + require.Empty(t, diagnostics.String()) + } + require.NotContains(t, diagnostics.String(), "private-api-key") + require.NotContains(t, diagnostics.String(), "secret") + }) + } +} + +func TestNearHeadLogsPreserveRPCErrors(t *testing.T) { + failure := fmt.Errorf("node unavailable") + for _, failAt := range []string{"before", "logs", "after"} { + t.Run(failAt, func(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + headers, reads := 0, 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getLogs" { + reads++ + if failAt == "logs" { + return failure + } + return nearHeadTestReply(out, []wt.Log{{BlockNumber: 99}}) + } + require.Equal(t, "eth_getBlockByNumber", method) + headers++ + if headers == 1 && failAt == "before" || headers == 2 && failAt == "after" { + return failure + } + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + logs, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.ErrorIs(t, err, failure) + require.Nil(t, logs) + if failAt == "before" { + require.Equal(t, 1, headers) + require.Zero(t, reads) + } else { + require.Equal(t, 1, reads) + } + }) + } +} + +// Hold one request inside its retry RPC while a second request uses the same +// original client. Both requests must retain the configured context/deadline. +func TestNearHeadLogsRetryContextIsolation(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + entered, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + headers := 0 + configuredCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + p := &fixtureProvider{call: func(ctx context.Context, out interface{}, method string, args ...interface{}) error { + require.Equal(t, configuredCtx, ctx) + if method == "eth_getLogs" { + fq := args[0].(wt.FilterQuery) + if *fq.FromBlock == 102 { + require.Equal(t, configuredCtx, ctx, "shared client context was mutated") + } + return nearHeadTestReply(out, []wt.Log{}) + } + headers++ + if headers == 1 { + return nearHeadTestReply(out, nil) + } + if headers == 2 { + require.Equal(t, configuredCtx, ctx, "retry context must stay unchanged") + close(entered) + select { + case <-release: + case <-ctx.Done(): + return ctx.Err() + } + } + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + }} + c := nearHeadTestClient(n, p) + c.RpcEthClient.SetContext(configuredCtx) + originalClient := c.RpcEthClient + done := make(chan error, 1) + go func() { + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + _, err := c.Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + done <- err + }() + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("retry did not reach the controlled RPC") + } + from, to := wt.BlockNumber(102), wt.BlockNumber(102) + _, err := c.Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.NoError(t, err) + require.Same(t, originalClient, c.RpcEthClient) + releaseOnce.Do(func() { close(release) }) + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(3 * time.Second): + t.Fatal("retry did not finish after release") + } +} + +func TestNearHeadLogsRetryInvalidBoundary(t *testing.T) { + for _, invalid := range []string{"missing", "hash", "number"} { + t.Run(invalid, func(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + block := fixtureBlock(100, hash, nil) + block.Receipts[0].Logs = nil // Empty cached results still need alignment. + require.NoError(t, n.Put(block)) + headers, reads := 0, 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getLogs" { + reads++ + return nearHeadTestReply(out, []wt.Log{}) + } + require.Equal(t, "eth_getBlockByNumber", method) + headers++ + block := nearHeadTestHeader(100, hash) + if headers == 1 { + switch invalid { + case "missing": + block = nil + case "hash": + block.Hash = common.HexToHash("0xb100") + case "number": + block.Number.SetInt64(101) + } + } + return nearHeadTestReply(out, block) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + logs, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.NoError(t, err) + require.Empty(t, logs) + require.NotNil(t, logs) + require.Equal(t, 3, headers) + require.Equal(t, 1, reads) + }) + } +} + +type nearHeadTestResolver struct { + calls, latest int +} + +func (r *nearHeadTestResolver) Resolve(number wt.BlockNumber) (uint64, error) { + r.calls++ + if number >= 0 { + return uint64(number), nil + } + r.latest++ + return uint64(99 + r.latest), nil +} + +func TestNearHeadLogsRetryFreezesLatest(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + headers := 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getBlockByNumber" { + require.Equal(t, wt.BlockNumber(100), args[0]) + headers++ + if headers == 1 { + return nearHeadTestReply(out, nil) + } + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + } + require.Equal(t, "eth_getLogs", method) + fq := args[0].(wt.FilterQuery) + require.Equal(t, wt.BlockNumber(99), *fq.FromBlock) + require.Equal(t, wt.BlockNumber(99), *fq.ToBlock) + return nearHeadTestReply(out, []wt.Log{}) + }} + c := nearHeadTestClient(n, p) + resolver := &nearHeadTestResolver{} + c.resolver = resolver + from, to := wt.BlockNumber(99), wt.LatestBlockNumber + filter := wt.FilterQuery{FromBlock: &from, ToBlock: &to} + _, err := c.Logs(filter) + require.NoError(t, err) + require.Equal(t, 2, resolver.calls) + require.Equal(t, 1, resolver.latest) + require.Equal(t, wt.LatestBlockNumber, *filter.ToBlock) + require.Equal(t, wt.BlockNumber(99), *filter.FromBlock) +} + +func TestNearHeadLogsRetryCoverageChange(t *testing.T) { + for _, complete := range []bool{false, true} { + t.Run(fmt.Sprint(complete), func(t *testing.T) { + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + headers, reads := 0, 0 + p := &fixtureProvider{call: func(ctx context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getBlockByNumber" { + headers++ + if headers != 1 { + return fmt.Errorf("new single-source attempt must not check headers") + } + require.Equal(t, 1, n.Pop(100)) + if complete { + require.NoError(t, n.Put(fixtureBlock(99, hash, nil))) + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + } + return nearHeadTestReply(out, nil) + } + require.Equal(t, "eth_getLogs", method) + reads++ + fq := args[0].(wt.FilterQuery) + require.Equal(t, wt.BlockNumber(99), *fq.FromBlock) + require.Equal(t, wt.BlockNumber(100), *fq.ToBlock) + _, bounded := ctx.Deadline() + require.False(t, bounded) + return nearHeadTestReply(out, []wt.Log{}) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + logs, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.NoError(t, err) + require.Equal(t, 1, headers) + if complete { + require.Len(t, logs, 2) + require.Zero(t, reads) + } else { + require.Empty(t, logs) + require.Equal(t, 1, reads) + } + }) + } +} + +func TestNearHeadLogsUsesCachedEnd(t *testing.T) { + n := fixtureNearhead() + first, last := common.HexToHash("0xa100"), common.HexToHash("0xa101") + topic := common.HexToHash("0x1234") + address := common.HexToAddress("0xabcd") + for i, hash := range []common.Hash{first, last} { + block := fixtureBlock(int64(100+i), hash, []common.Hash{topic}) + block.Block.ParentHash = common.HexToHash("0xa99") + if i > 0 { + block.Block.ParentHash = first + } + block.Receipts[0].Logs[0].Address = address + require.NoError(t, n.Put(block)) + } + headers := 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getBlockByNumber" { + headers++ + require.Equal(t, wt.BlockNumber(101), args[0], "use the cached end, not its start or the requested upper bound") + return nearHeadTestReply(out, nearHeadTestHeader(101, last)) + } + require.Equal(t, "eth_getLogs", method) + fq := args[0].(wt.FilterQuery) + require.Equal(t, []common.Address{address}, fq.Addresses) + require.Equal(t, [][]common.Hash{{topic}}, fq.Topics) + return nearHeadTestReply(out, []wt.Log{{BlockNumber: uint64(*fq.FromBlock)}}) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(102) + filter := wt.FilterQuery{FromBlock: &from, ToBlock: &to, Addresses: []common.Address{address}, Topics: [][]common.Hash{{topic}}} + logs, err := nearHeadTestClient(n, p).Logs(filter) + require.NoError(t, err) + require.Equal(t, 2, headers) + require.Len(t, logs, 4) + for i, entry := range logs { + require.Equal(t, uint64(99+i), entry.BlockNumber) + } + require.Equal(t, wt.BlockNumber(99), *filter.FromBlock) + require.Equal(t, wt.BlockNumber(102), *filter.ToBlock) +} + +// Wrap the real cache only at call boundaries, so mutations occur at controlled +// points without relying on scheduling or changing the dependency's readers. +type nearHeadCountingReader struct { + *EthNearHeadCache + t *testing.T + blocks int + scans int + beforeScan func() + afterScan func(*nearhead.EthLogs) + missingAt int + scanError error + wantOpt nearhead.FilterOpt +} + +func (r *nearHeadCountingReader) GetBlock(ref cacheTypes.BlockHashOrNumber, full bool) nearhead.EthCacheResultWithCoverage[*wt.Block] { + r.blocks++ + require.True(r.t, full, "metadata lookup must not build transaction hash lists") + res := r.EthNearHeadCache.GetBlock(ref, full) + if r.blocks == r.missingAt { + res.Result = nil + } + return res +} + +func (r *nearHeadCountingReader) GetLogsByBlockRange(from, to uint64, opt nearhead.FilterOpt) (*nearhead.EthLogs, error) { + r.scans++ + require.Equal(r.t, r.wantOpt, opt) + if r.beforeScan != nil { + r.beforeScan() + } + if r.scanError != nil { + return nil, r.scanError + } + res, err := r.EthNearHeadCache.GetLogsByBlockRange(from, to, opt) + if r.afterScan != nil { + r.afterScan(res) + } + return res, err +} + +func TestReadNearHeadLogsSingleScan(t *testing.T) { + a, b := common.HexToHash("0xa101"), common.HexToHash("0xb101") + for _, tc := range []struct { + name string + from, to uint64 + wantBlocks int + wantScans int + wantFrom uint64 + wantTo uint64 + conflict bool + }{ + {name: "complete", from: 100, to: 101, wantBlocks: 1, wantScans: 1, wantFrom: 100, wantTo: 101}, + {name: "empty complete", from: 100, to: 101, wantBlocks: 1, wantScans: 1, wantFrom: 100, wantTo: 101}, + {name: "miss", from: 102, to: 103, wantBlocks: 1}, + {name: "left", from: 99, to: 101, wantBlocks: 2, wantScans: 1, wantFrom: 100, wantTo: 101}, + {name: "right", from: 100, to: 102, wantBlocks: 3, wantScans: 1, wantFrom: 100, wantTo: 101}, + {name: "both", from: 99, to: 102, wantBlocks: 3, wantScans: 1, wantFrom: 100, wantTo: 101}, + {name: "disappeared", from: 99, to: 101, wantBlocks: 1, wantScans: 1, conflict: true}, + {name: "reorg before scan", from: 99, to: 101, wantBlocks: 2, wantScans: 1, conflict: true}, + {name: "reorg after scan", from: 99, to: 101, wantBlocks: 2, wantScans: 1, conflict: true}, + {name: "end expanded", from: 99, to: 102, wantBlocks: 3, wantScans: 1, conflict: true}, + {name: "expanded to complete", from: 100, to: 102, wantBlocks: 2, wantScans: 1, wantFrom: 100, wantTo: 102}, + {name: "start evicted", from: 99, to: 101, wantBlocks: 2, wantScans: 1, wantFrom: 101, wantTo: 101}, + {name: "missing before", from: 99, to: 101, wantBlocks: 1, wantScans: 1, conflict: true}, + {name: "missing after", from: 99, to: 101, wantBlocks: 2, wantScans: 1, conflict: true}, + {name: "complete missing metadata", from: 100, to: 101, wantBlocks: 1, wantScans: 1, wantFrom: 100, wantTo: 101}, + } { + t.Run(tc.name, func(t *testing.T) { + n := fixtureNearhead() + first, last := fixtureBlock(100, common.HexToHash("0xa100"), nil), fixtureBlock(101, a, nil) + if tc.name == "start evicted" { + n.EthCache = nearhead.NewEthCache(nearhead.Config{MaxMemory: uint64(first.Size() + last.Size()), MaxLogs: 100}) + } + if tc.name == "empty complete" { + first.Receipts[0].Logs, last.Receipts[0].Logs = nil, nil + } + require.NoError(t, n.Put(first)) + require.NoError(t, n.Put(last)) + r := &nearHeadCountingReader{EthNearHeadCache: n, t: t} + reorg := func() { + require.Equal(t, 1, n.Pop(101)) + require.NoError(t, n.Put(fixtureBlock(101, b, nil))) + } + switch tc.name { + case "disappeared": + r.beforeScan = func() { require.Equal(t, 2, n.Pop(100)) } + case "reorg before scan": + r.beforeScan = reorg + case "reorg after scan": + r.afterScan = func(*nearhead.EthLogs) { reorg() } + case "end expanded", "expanded to complete", "start evicted": + r.beforeScan = func() { require.NoError(t, n.Put(fixtureBlock(102, common.HexToHash("0xa102"), nil))) } + case "missing before", "complete missing metadata": + r.missingAt = 1 + case "missing after": + r.missingAt = 2 + } + snapshot, hash, err := readNearHeadLogs(r, tc.from, tc.to, nearhead.FilterOpt{}) + require.Equal(t, tc.wantBlocks, r.blocks) + require.Equal(t, tc.wantScans, r.scans) + if tc.conflict { + require.IsType(t, &nearHeadLogConflict{}, err) + require.Nil(t, snapshot, "discard the scanned logs on conflict") + return + } + require.NoError(t, err) + if tc.wantScans == 0 { + require.Nil(t, snapshot) + return + } + require.Equal(t, tc.wantFrom, snapshot.FromBlock) + require.Equal(t, tc.wantTo, snapshot.ToBlock) + if tc.name == "empty complete" { + require.Empty(t, snapshot.Logs) + } + if tc.wantFrom != tc.from || tc.wantTo != tc.to { + require.Equal(t, a, hash) + } else { + require.Zero(t, hash, "complete cache result needs no boundary") + } + }) + } +} + +func TestReadNearHeadLogsErrorsAndFilter(t *testing.T) { + n := fixtureNearhead() + topic := common.HexToHash("0x1234") + address := common.HexToAddress("0xabcd") + block := fixtureBlock(100, common.HexToHash("0xa100"), []common.Hash{topic}) + block.Receipts[0].Logs[0].Address = address + require.NoError(t, n.Put(block)) + opt := nearhead.FilterOpt{Addresses: []common.Address{address}, Topics: [][]common.Hash{{topic}}} + r := &nearHeadCountingReader{EthNearHeadCache: n, t: t, wantOpt: opt} + snapshot, _, err := readNearHeadLogs(r, 99, 100, opt) + require.NoError(t, err) + require.Len(t, snapshot.Logs, 1) + require.Equal(t, 1, r.scans) + + failure := fmt.Errorf("cache unavailable") + r.scanError = failure + snapshot, _, err = readNearHeadLogs(r, 99, 100, opt) + require.ErrorIs(t, err, failure) + require.Nil(t, snapshot) + r.scanError = nil + r.afterScan = func(res *nearhead.EthLogs) { res.FromBlock = 98 } + snapshot, _, err = readNearHeadLogs(r, 99, 100, opt) + require.EqualError(t, err, "bad event logs from near head cache: unexpected block range") + require.Nil(t, snapshot) + scans := r.scans + _, _, err = readNearHeadLogs(r, 101, 100, opt) + require.EqualError(t, err, "failed to get logs from near head cache: invalid block range params") + require.Equal(t, scans, r.scans) +} + +func TestNearHeadLogRetryDiagnosticLevels(t *testing.T) { + for _, level := range []logrus.Level{logrus.DebugLevel, logrus.InfoLevel, logrus.WarnLevel} { + t.Run(level.String(), func(t *testing.T) { + logger := logrus.StandardLogger() + output, originalLevel := logger.Out, logger.GetLevel() + var diagnostics bytes.Buffer + logger.SetOutput(&diagnostics) + logger.SetLevel(level) + t.Cleanup(func() { logger.SetOutput(output); logger.SetLevel(originalLevel) }) + + n := fixtureNearhead() + hash := common.HexToHash("0xa100") + require.NoError(t, n.Put(fixtureBlock(100, hash, nil))) + headers := 0 + p := &fixtureProvider{call: func(_ context.Context, out interface{}, method string, args ...interface{}) error { + if method == "eth_getLogs" { + return nearHeadTestReply(out, []wt.Log{}) + } + require.Equal(t, "eth_getBlockByNumber", method) + headers++ + if headers == 1 { + return nearHeadTestReply(out, nil) + } + return nearHeadTestReply(out, nearHeadTestHeader(100, hash)) + }} + from, to := wt.BlockNumber(99), wt.BlockNumber(100) + _, err := nearHeadTestClient(n, p).Logs(wt.FilterQuery{FromBlock: &from, ToBlock: &to}) + require.NoError(t, err) + require.Equal(t, 3, headers) + if level == logrus.DebugLevel { + require.Equal(t, 2, bytes.Count(diagnostics.Bytes(), []byte("level=debug"))) + require.Contains(t, diagnostics.String(), "Retrying near-head logs after canonical view conflict") + require.Contains(t, diagnostics.String(), "Near-head log consistency recovered") + require.Contains(t, diagnostics.String(), "boundary=100") + require.NotContains(t, diagnostics.String(), "level=info") + } else { + require.Empty(t, diagnostics.String()) + } + }) + } +} diff --git a/util/types.go b/util/types.go index a38e3095..20e3da5b 100644 --- a/util/types.go +++ b/util/types.go @@ -10,7 +10,7 @@ func IsInterfaceValNil(i interface{}) bool { } switch reflect.TypeOf(i).Kind() { - case reflect.Ptr, reflect.Map, reflect.Array, reflect.Chan, reflect.Slice, reflect.Func: + case reflect.Ptr, reflect.Map, reflect.Chan, reflect.Slice, reflect.Func: return reflect.ValueOf(i).IsNil() }