Skip to content

N
nebula
Commit 4b5422be authored by Felipe Andrade's avatar Felipe Andrade
Browse files
Options

refactor(proxy): cache only immutable and hash-based methods

parent 382d38b7
Hide whitespace changes
Inline Side-by-side
Showing with 189 additions and 909 deletions
proxyd/cache.go
View file @ 4b5422be
...@@ -116,15 +116,17 @@ type rpcCache struct { ...@@ -116,15 +116,17 @@ type rpcCache struct {
handlers map[string]RPCMethodHandler handlers map[string]RPCMethodHandler
} }
func newRPCCache(cache Cache, getLatestBlockNumFn GetLatestBlockNumFn, getLatestGasPriceFn GetLatestGasPriceFn, numBlockConfirmations int) RPCCache { func newRPCCache(cache Cache) RPCCache {
handlers := map[string]RPCMethodHandler{ handlers := map[string]RPCMethodHandler{
"eth_chainId": &StaticMethodHandler{}, "eth_chainId": &StaticMethodHandler{cache: cache},
"net_version": &StaticMethodHandler{}, "net_version": &StaticMethodHandler{cache: cache},
"eth_getBlockByNumber": &EthGetBlockByNumberMethodHandler{cache, getLatestBlockNumFn, numBlockConfirmations}, "eth_getBlockTransactionCountByHash": &StaticMethodHandler{cache: cache},
"eth_getBlockRange": &EthGetBlockRangeMethodHandler{cache, getLatestBlockNumFn, numBlockConfirmations}, "eth_getUncleCountByBlockHash": &StaticMethodHandler{cache: cache},
"eth_blockNumber": &EthBlockNumberMethodHandler{getLatestBlockNumFn}, "eth_getBlockByHash": &StaticMethodHandler{cache: cache},
"eth_gasPrice": &EthGasPriceMethodHandler{getLatestGasPriceFn}, "eth_getTransactionByHash": &StaticMethodHandler{cache: cache},
"eth_call": &EthCallMethodHandler{cache, getLatestBlockNumFn, numBlockConfirmations}, "eth_getTransactionByBlockHashAndIndex": &StaticMethodHandler{cache: cache},
"eth_getUncleByBlockHashAndIndex": &StaticMethodHandler{cache: cache},
"eth_getTransactionReceipt": &StaticMethodHandler{cache: cache},
} }
return &rpcCache{ return &rpcCache{
cache: cache, cache: cache,
...@@ -138,12 +140,14 @@ func (c *rpcCache) GetRPC(ctx context.Context, req *RPCReq) (*RPCRes, error) { ...@@ -138,12 +140,14 @@ func (c *rpcCache) GetRPC(ctx context.Context, req *RPCReq) (*RPCRes, error) {
return nil, nil return nil, nil
} }
res, err := handler.GetRPCMethod(ctx, req) res, err := handler.GetRPCMethod(ctx, req)
if res != nil { if err != nil {
if res == nil { RecordCacheError(req.Method)
RecordCacheMiss(req.Method) return nil, err
} else { }
RecordCacheHit(req.Method) if res == nil {
} RecordCacheMiss(req.Method)
} else {
RecordCacheHit(req.Method)
} }
return res, err return res, err
} }
......
proxyd/cache_test.go
View file @ 4b5422be
...@@ -2,23 +2,16 @@ package proxyd ...@@ -2,23 +2,16 @@ package proxyd
import ( import (
"context" "context"
"math"
"strconv" "strconv"
"testing" "testing"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
const numBlockConfirmations = 10
func TestRPCCacheImmutableRPCs(t *testing.T) { func TestRPCCacheImmutableRPCs(t *testing.T) {
const blockHead = math.MaxUint64
ctx := context.Background() ctx := context.Background()
getBlockNum := func(ctx context.Context) (uint64, error) { cache := newRPCCache(newMemoryCache())
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), getBlockNum, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1)) ID := []byte(strconv.Itoa(1))
rpcs := []struct { rpcs := []struct {
...@@ -55,316 +48,100 @@ func TestRPCCacheImmutableRPCs(t *testing.T) { ...@@ -55,316 +48,100 @@ func TestRPCCacheImmutableRPCs(t *testing.T) {
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getBlockTransactionCountByHash",
Params: []byte(`["0x1", false]`), Params: mustMarshalJSON([]string{"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238"}),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`, Result: `{"eth_getBlockTransactionCountByHash":"!"}`,
ID: ID, ID: ID,
}, },
name: "eth_getBlockByNumber", name: "eth_getBlockTransactionCountByHash",
}, },
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getUncleCountByBlockHash",
Params: []byte(`["earliest", false]`), Params: mustMarshalJSON([]string{"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238"}),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`, Result: `{"eth_getUncleCountByBlockHash":"!"}`,
ID: ID, ID: ID,
}, },
name: "eth_getBlockByNumber earliest", name: "eth_getUncleCountByBlockHash",
}, },
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getBlockByHash",
Params: []byte(`["safe", false]`), Params: mustMarshalJSON([]string{"0xc6ef2fc5426d6ad6fd9e2a26abeab0aa2411b7ab17f30a99d3cb96aed1d1055b", "false"}),
ID: ID,
},
res: nil,
name: "eth_getBlockByNumber safe",
},
{
req: &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockByNumber",
Params: []byte(`["finalized", false]`),
ID: ID, ID: ID,
}, },
res: nil, res: &RPCRes{
name: "eth_getBlockByNumber finalized",
},
{
req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Result: `{"eth_getBlockByHash":"!"}`,
Params: []byte(`["pending", false]`),
ID: ID, ID: ID,
}, },
res: nil, name: "eth_getBlockByHash",
name: "eth_getBlockByNumber pending",
}, },
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getTransactionByHash",
Params: []byte(`["latest", false]`), Params: mustMarshalJSON([]string{"0x88df016429689c079f3b2f6ad39fa052532c56795b733da78a91ebe6a713944b"}),
ID: ID,
},
res: nil,
name: "eth_getBlockByNumber latest",
},
{
req: &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockRange",
Params: []byte(`["0x1", "0x2", false]`),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`, Result: `{"eth_getTransactionByHash":"!"}`,
ID: ID, ID: ID,
}, },
name: "eth_getBlockRange", name: "eth_getTransactionByHash",
}, },
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_getTransactionByBlockHashAndIndex",
Params: []byte(`["earliest", "0x2", false]`), Params: mustMarshalJSON([]string{"0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", "0x55"}),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`, Result: `{"eth_getTransactionByBlockHashAndIndex":"!"}`,
ID: ID, ID: ID,
}, },
name: "eth_getBlockRange earliest", name: "eth_getTransactionByBlockHashAndIndex",
}, },
}
for _, rpc := range rpcs {
t.Run(rpc.name, func(t *testing.T) {
err := cache.PutRPC(ctx, rpc.req, rpc.res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, rpc.req)
require.NoError(t, err)
require.Equal(t, rpc.res, cachedRes)
})
}
}
func TestRPCCacheBlockNumber(t *testing.T) {
var blockHead uint64 = 0x1000
var gasPrice uint64 = 0x100
ctx := context.Background()
ID := []byte(strconv.Itoa(1))
getGasPrice := func(ctx context.Context) (uint64, error) {
return gasPrice, nil
}
getBlockNum := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), getBlockNum, getGasPrice, numBlockConfirmations)
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_blockNumber",
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `0x1000`,
ID: ID,
}
err := cache.PutRPC(ctx, req, res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, res, cachedRes)
blockHead = 0x1001
cachedRes, err = cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, &RPCRes{JSONRPC: "2.0", Result: `0x1001`, ID: ID}, cachedRes)
}
func TestRPCCacheGasPrice(t *testing.T) {
var blockHead uint64 = 0x1000
var gasPrice uint64 = 0x100
ctx := context.Background()
ID := []byte(strconv.Itoa(1))
getGasPrice := func(ctx context.Context) (uint64, error) {
return gasPrice, nil
}
getBlockNum := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), getBlockNum, getGasPrice, numBlockConfirmations)
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_gasPrice",
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `0x100`,
ID: ID,
}
err := cache.PutRPC(ctx, req, res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, res, cachedRes)
gasPrice = 0x101
cachedRes, err = cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, &RPCRes{JSONRPC: "2.0", Result: `0x101`, ID: ID}, cachedRes)
}
func TestRPCCacheUnsupportedMethod(t *testing.T) {
const blockHead = math.MaxUint64
ctx := context.Background()
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1))
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_syncing",
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: false,
ID: ID,
}
err := cache.PutRPC(ctx, req, res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
}
func TestRPCCacheEthGetBlockByNumber(t *testing.T) {
ctx := context.Background()
var blockHead uint64
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
makeCache := func() RPCCache { return newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations) }
ID := []byte(strconv.Itoa(1))
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockByNumber",
Params: []byte(`["0xa", false]`),
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`,
ID: ID,
}
req2 := &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockByNumber",
Params: []byte(`["0xb", false]`),
ID: ID,
}
res2 := &RPCRes{
JSONRPC: "2.0",
Result: `{"difficulty": "0x2", "number": "0x2"}`,
ID: ID,
}
t.Run("set multiple finalized blocks", func(t *testing.T) {
blockHead = 100
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
require.NoError(t, cache.PutRPC(ctx, req2, res2))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, res, cachedRes)
cachedRes, err = cache.GetRPC(ctx, req2)
require.NoError(t, err)
require.Equal(t, res2, cachedRes)
})
t.Run("unconfirmed block", func(t *testing.T) {
blockHead = 0xc
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
}
func TestRPCCacheEthGetBlockByNumberForRecentBlocks(t *testing.T) {
ctx := context.Background()
var blockHead uint64 = 2
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1))
rpcs := []struct {
req *RPCReq
res *RPCRes
name string
}{
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getUncleByBlockHashAndIndex",
Params: []byte(`["latest", false]`), Params: mustMarshalJSON([]string{"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238", "0x90"}),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`, Result: `{"eth_getUncleByBlockHashAndIndex":"!"}`,
ID: ID, ID: ID,
}, },
name: "latest block", name: "eth_getUncleByBlockHashAndIndex",
}, },
{ {
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockByNumber", Method: "eth_getTransactionReceipt",
Params: []byte(`["pending", false]`), Params: mustMarshalJSON([]string{"0x85d995eba9763907fdf35cd2034144dd9d53ce32cbec21349d4b12823c6860c5"}),
ID: ID, ID: ID,
}, },
res: &RPCRes{ res: &RPCRes{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`, Result: `{"eth_getTransactionReceipt":"!"}`,
ID: ID, ID: ID,
}, },
name: "pending block", name: "eth_getTransactionReceipt",
}, },
} }
...@@ -375,288 +152,81 @@ func TestRPCCacheEthGetBlockByNumberForRecentBlocks(t *testing.T) { ...@@ -375,288 +152,81 @@ func TestRPCCacheEthGetBlockByNumberForRecentBlocks(t *testing.T) {
cachedRes, err := cache.GetRPC(ctx, rpc.req) cachedRes, err := cache.GetRPC(ctx, rpc.req)
require.NoError(t, err) require.NoError(t, err)
require.Nil(t, cachedRes) require.Equal(t, rpc.res, cachedRes)
}) })
} }
} }
func TestRPCCacheEthGetBlockByNumberInvalidRequest(t *testing.T) { func TestRPCCacheUnsupportedMethod(t *testing.T) {
ctx := context.Background()
const blockHead = math.MaxUint64
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1))
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockByNumber",
Params: []byte(`["0x1"]`), // missing required boolean param
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `{"difficulty": "0x1", "number": "0x1"}`,
ID: ID,
}
err := cache.PutRPC(ctx, req, res)
require.Error(t, err)
cachedRes, err := cache.GetRPC(ctx, req)
require.Error(t, err)
require.Nil(t, cachedRes)
}
func TestRPCCacheEthGetBlockRange(t *testing.T) {
ctx := context.Background()
var blockHead uint64
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
makeCache := func() RPCCache { return newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations) }
ID := []byte(strconv.Itoa(1))
t.Run("finalized block", func(t *testing.T) {
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockRange",
Params: []byte(`["0x1", "0x10", false]`),
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x10"}]`,
ID: ID,
}
blockHead = 0x1000
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, res, cachedRes)
})
t.Run("unconfirmed block", func(t *testing.T) {
cache := makeCache()
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_getBlockRange",
Params: []byte(`["0x1", "0x1000", false]`),
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`,
ID: ID,
}
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
}
func TestRPCCacheEthGetBlockRangeForRecentBlocks(t *testing.T) {
ctx := context.Background() ctx := context.Background()
var blockHead uint64 = 0x1000 cache := newRPCCache(newMemoryCache())
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1)) ID := []byte(strconv.Itoa(1))
rpcs := []struct { rpcs := []struct {
req *RPCReq req *RPCReq
res *RPCRes
name string name string
}{ }{
{ {
name: "eth_syncing",
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_syncing",
Params: []byte(`["0x1", "latest", false]`),
ID: ID,
},
res: &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`,
ID: ID, ID: ID,
}, },
name: "latest block",
}, },
{ {
name: "eth_blockNumber",
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_blockNumber",
Params: []byte(`["0x1", "pending", false]`),
ID: ID,
},
res: &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`,
ID: ID, ID: ID,
}, },
name: "pending block",
}, },
{ {
name: "eth_getBlockByNumber",
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_getBlockByNumber",
Params: []byte(`["latest", "0x1000", false]`),
ID: ID,
},
res: &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`,
ID: ID, ID: ID,
}, },
name: "latest block 2",
}, },
}
for _, rpc := range rpcs {
t.Run(rpc.name, func(t *testing.T) {
err := cache.PutRPC(ctx, rpc.req, rpc.res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, rpc.req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
}
}
func TestRPCCacheEthGetBlockRangeInvalidRequest(t *testing.T) {
ctx := context.Background()
const blockHead = math.MaxUint64
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
cache := newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations)
ID := []byte(strconv.Itoa(1))
rpcs := []struct {
req *RPCReq
res *RPCRes
name string
}{
{ {
name: "eth_getBlockRange",
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_getBlockRange",
Params: []byte(`["0x1", "0x2"]`), // missing required boolean param
ID: ID, ID: ID,
}, },
res: &RPCRes{
JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`,
ID: ID,
},
name: "missing boolean param",
}, },
{ {
name: "eth_gasPrice",
req: &RPCReq{ req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Method: "eth_getBlockRange", Method: "eth_gasPrice",
Params: []byte(`["abc", "0x2", true]`), // invalid block hex
ID: ID, ID: ID,
}, },
res: &RPCRes{ },
{
name: "eth_call",
req: &RPCReq{
JSONRPC: "2.0", JSONRPC: "2.0",
Result: `[{"number": "0x1"}, {"number": "0x2"}]`, Method: "eth_gasPrice",
ID: ID, ID: ID,
}, },
name: "invalid block hex",
}, },
} }
for _, rpc := range rpcs { for _, rpc := range rpcs {
t.Run(rpc.name, func(t *testing.T) { t.Run(rpc.name, func(t *testing.T) {
err := cache.PutRPC(ctx, rpc.req, rpc.res) fakeval := mustMarshalJSON([]string{rpc.name})
require.Error(t, err) err := cache.PutRPC(ctx, rpc.req, &RPCRes{Result: fakeval})
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, rpc.req) cachedRes, err := cache.GetRPC(ctx, rpc.req)
require.Error(t, err) require.NoError(t, err)
require.Nil(t, cachedRes) require.Nil(t, cachedRes)
}) })
} }
}
func TestRPCCacheEthCall(t *testing.T) {
ctx := context.Background()
var blockHead uint64
fn := func(ctx context.Context) (uint64, error) {
return blockHead, nil
}
makeCache := func() RPCCache { return newRPCCache(newMemoryCache(), fn, nil, numBlockConfirmations) }
ID := []byte(strconv.Itoa(1))
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_call",
Params: []byte(`[{"to": "0xDEADBEEF", "data": "0x1"}, "0x10"]`),
ID: ID,
}
res := &RPCRes{
JSONRPC: "2.0",
Result: `0x0`,
ID: ID,
}
t.Run("finalized block", func(t *testing.T) {
blockHead = 0x100
cache := makeCache()
err := cache.PutRPC(ctx, req, res)
require.NoError(t, err)
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Equal(t, res, cachedRes)
})
t.Run("unconfirmed block", func(t *testing.T) {
blockHead = 0x10
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
t.Run("latest block", func(t *testing.T) {
blockHead = 0x100
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_call",
Params: []byte(`[{"to": "0xDEADBEEF", "data": "0x1"}, "latest"]`),
ID: ID,
}
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
t.Run("pending block", func(t *testing.T) {
blockHead = 0x100
req := &RPCReq{
JSONRPC: "2.0",
Method: "eth_call",
Params: []byte(`[{"to": "0xDEADBEEF", "data": "0x1"}, "pending"]`),
ID: ID,
}
cache := makeCache()
require.NoError(t, cache.PutRPC(ctx, req, res))
cachedRes, err := cache.GetRPC(ctx, req)
require.NoError(t, err)
require.Nil(t, cachedRes)
})
} }
proxyd/integration_tests/caching_test.go
View file @ 4b5422be
...@@ -18,15 +18,19 @@ func TestCaching(t *testing.T) { ...@@ -18,15 +18,19 @@ func TestCaching(t *testing.T) {
defer redis.Close() defer redis.Close()
hdlr := NewBatchRPCResponseRouter() hdlr := NewBatchRPCResponseRouter()
/* cacheable */
hdlr.SetRoute("eth_chainId", "999", "0x420") hdlr.SetRoute("eth_chainId", "999", "0x420")
hdlr.SetRoute("net_version", "999", "0x1234") hdlr.SetRoute("net_version", "999", "0x1234")
hdlr.SetRoute("eth_blockNumber", "999", "0x64") hdlr.SetRoute("eth_getBlockTransactionCountByHash", "999", "eth_getBlockTransactionCountByHash")
hdlr.SetRoute("eth_getBlockByNumber", "999", "dummy_block") hdlr.SetRoute("eth_getBlockByHash", "999", "eth_getBlockByHash")
hdlr.SetRoute("eth_call", "999", "dummy_call") hdlr.SetRoute("eth_getTransactionByHash", "999", "eth_getTransactionByHash")
hdlr.SetRoute("eth_getTransactionByBlockHashAndIndex", "999", "eth_getTransactionByBlockHashAndIndex")
// mock LVC requests hdlr.SetRoute("eth_getUncleByBlockHashAndIndex", "999", "eth_getUncleByBlockHashAndIndex")
hdlr.SetFallbackRoute("eth_blockNumber", "0x64") hdlr.SetRoute("eth_getTransactionReceipt", "999", "eth_getTransactionReceipt")
hdlr.SetFallbackRoute("eth_gasPrice", "0x420") /* not cacheable */
hdlr.SetRoute("eth_getBlockByNumber", "999", "eth_getBlockByNumber")
hdlr.SetRoute("eth_blockNumber", "999", "eth_blockNumber")
hdlr.SetRoute("eth_call", "999", "eth_call")
backend := NewMockBackend(hdlr) backend := NewMockBackend(hdlr)
defer backend.Close() defer backend.Close()
...@@ -48,6 +52,7 @@ func TestCaching(t *testing.T) { ...@@ -48,6 +52,7 @@ func TestCaching(t *testing.T) {
response string response string
backendCalls int backendCalls int
}{ }{
/* cacheable */
{ {
"eth_chainId", "eth_chainId",
nil, nil,
...@@ -60,14 +65,51 @@ func TestCaching(t *testing.T) { ...@@ -60,14 +65,51 @@ func TestCaching(t *testing.T) {
"{\"jsonrpc\": \"2.0\", \"result\": \"0x1234\", \"id\": 999}", "{\"jsonrpc\": \"2.0\", \"result\": \"0x1234\", \"id\": 999}",
1, 1,
}, },
{
"eth_getBlockTransactionCountByHash",
[]interface{}{"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getBlockTransactionCountByHash\", \"id\": 999}",
1,
},
{
"eth_getBlockByHash",
[]interface{}{"0xc6ef2fc5426d6ad6fd9e2a26abeab0aa2411b7ab17f30a99d3cb96aed1d1055b", "false"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getBlockByHash\", \"id\": 999}",
1,
},
{
"eth_getTransactionByHash",
[]interface{}{"0x88df016429689c079f3b2f6ad39fa052532c56795b733da78a91ebe6a713944b"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getTransactionByHash\", \"id\": 999}",
1,
},
{
"eth_getTransactionByBlockHashAndIndex",
[]interface{}{"0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", "0x55"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getTransactionByBlockHashAndIndex\", \"id\": 999}",
1,
},
{
"eth_getUncleByBlockHashAndIndex",
[]interface{}{"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238", "0x90"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getUncleByBlockHashAndIndex\", \"id\": 999}",
1,
},
{
"eth_getTransactionReceipt",
[]interface{}{"0x85d995eba9763907fdf35cd2034144dd9d53ce32cbec21349d4b12823c6860c5"},
"{\"jsonrpc\": \"2.0\", \"result\": \"eth_getTransactionReceipt\", \"id\": 999}",
1,
},
/* not cacheable */
{ {
"eth_getBlockByNumber", "eth_getBlockByNumber",
[]interface{}{ []interface{}{
"0x1", "0x1",
true, true,
}, },
"{\"jsonrpc\": \"2.0\", \"result\": \"dummy_block\", \"id\": 999}", "{\"jsonrpc\": \"2.0\", \"result\": \"eth_getBlockByNumber\", \"id\": 999}",
1, 2,
}, },
{ {
"eth_call", "eth_call",
...@@ -79,14 +121,14 @@ func TestCaching(t *testing.T) { ...@@ -79,14 +121,14 @@ func TestCaching(t *testing.T) {
}, },
"0x60", "0x60",
}, },
"{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":\"dummy_call\"}", "{\"jsonrpc\": \"2.0\", \"result\": \"eth_call\", \"id\": 999}",
1, 2,
}, },
{ {
"eth_blockNumber", "eth_blockNumber",
nil, nil,
"{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":\"0x64\"}", "{\"jsonrpc\": \"2.0\", \"result\": \"eth_blockNumber\", \"id\": 999}",
0, 2,
}, },
{ {
"eth_call", "eth_call",
...@@ -98,7 +140,7 @@ func TestCaching(t *testing.T) { ...@@ -98,7 +140,7 @@ func TestCaching(t *testing.T) {
}, },
"latest", "latest",
}, },
"{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":\"dummy_call\"}", "{\"jsonrpc\": \"2.0\", \"result\": \"eth_call\", \"id\": 999}",
2, 2,
}, },
{ {
...@@ -111,7 +153,7 @@ func TestCaching(t *testing.T) { ...@@ -111,7 +153,7 @@ func TestCaching(t *testing.T) {
}, },
"pending", "pending",
}, },
"{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":\"dummy_call\"}", "{\"jsonrpc\": \"2.0\", \"result\": \"eth_call\", \"id\": 999}",
2, 2,
}, },
} }
...@@ -128,24 +170,15 @@ func TestCaching(t *testing.T) { ...@@ -128,24 +170,15 @@ func TestCaching(t *testing.T) {
}) })
} }
t.Run("block numbers update", func(t *testing.T) {
hdlr.SetFallbackRoute("eth_blockNumber", "0x100")
time.Sleep(1500 * time.Millisecond)
resRaw, _, err := client.SendRPC("eth_blockNumber", nil)
require.NoError(t, err)
RequireEqualJSON(t, []byte("{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":\"0x100\"}"), resRaw)
backend.Reset()
})
t.Run("nil responses should not be cached", func(t *testing.T) { t.Run("nil responses should not be cached", func(t *testing.T) {
hdlr.SetRoute("eth_getBlockByNumber", "999", nil) hdlr.SetRoute("eth_getBlockByHash", "999", nil)
resRaw, _, err := client.SendRPC("eth_getBlockByNumber", []interface{}{"0x123"}) resRaw, _, err := client.SendRPC("eth_getBlockByHash", []interface{}{"0x123"})
require.NoError(t, err) require.NoError(t, err)
resCache, _, err := client.SendRPC("eth_getBlockByNumber", []interface{}{"0x123"}) resCache, _, err := client.SendRPC("eth_getBlockByHash", []interface{}{"0x123"})
require.NoError(t, err) require.NoError(t, err)
RequireEqualJSON(t, []byte("{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":null}"), resRaw) RequireEqualJSON(t, []byte("{\"id\":999,\"jsonrpc\":\"2.0\",\"result\":null}"), resRaw)
RequireEqualJSON(t, resRaw, resCache) RequireEqualJSON(t, resRaw, resCache)
require.Equal(t, 2, countRequests(backend, "eth_getBlockByNumber")) require.Equal(t, 2, countRequests(backend, "eth_getBlockByHash"))
}) })
} }
...@@ -158,10 +191,7 @@ func TestBatchCaching(t *testing.T) { ...@@ -158,10 +191,7 @@ func TestBatchCaching(t *testing.T) {
hdlr.SetRoute("eth_chainId", "1", "0x420") hdlr.SetRoute("eth_chainId", "1", "0x420")
hdlr.SetRoute("net_version", "1", "0x1234") hdlr.SetRoute("net_version", "1", "0x1234")
hdlr.SetRoute("eth_call", "1", "dummy_call") hdlr.SetRoute("eth_call", "1", "dummy_call")
hdlr.SetRoute("eth_getBlockByHash", "1", "eth_getBlockByHash")
// mock LVC requests
hdlr.SetFallbackRoute("eth_blockNumber", "0x64")
hdlr.SetFallbackRoute("eth_gasPrice", "0x420")
backend := NewMockBackend(hdlr) backend := NewMockBackend(hdlr)
defer backend.Close() defer backend.Close()
...@@ -181,26 +211,31 @@ func TestBatchCaching(t *testing.T) { ...@@ -181,26 +211,31 @@ func TestBatchCaching(t *testing.T) {
goodChainIdResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"0x420\", \"id\": 1}" goodChainIdResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"0x420\", \"id\": 1}"
goodNetVersionResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"0x1234\", \"id\": 1}" goodNetVersionResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"0x1234\", \"id\": 1}"
goodEthCallResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"dummy_call\", \"id\": 1}" goodEthCallResponse := "{\"jsonrpc\": \"2.0\", \"result\": \"dummy_call\", \"id\": 1}"
goodEthGetBlockByHash := "{\"jsonrpc\": \"2.0\", \"result\": \"eth_getBlockByHash\", \"id\": 1}"
res, _, err := client.SendBatchRPC( res, _, err := client.SendBatchRPC(
NewRPCReq("1", "eth_chainId", nil), NewRPCReq("1", "eth_chainId", nil),
NewRPCReq("1", "net_version", nil), NewRPCReq("1", "net_version", nil),
NewRPCReq("1", "eth_getBlockByHash", []interface{}{"0xc6ef2fc5426d6ad6fd9e2a26abeab0aa2411b7ab17f30a99d3cb96aed1d1055b", "false"}),
) )
require.NoError(t, err) require.NoError(t, err)
RequireEqualJSON(t, []byte(asArray(goodChainIdResponse, goodNetVersionResponse)), res) RequireEqualJSON(t, []byte(asArray(goodChainIdResponse, goodNetVersionResponse, goodEthGetBlockByHash)), res)
require.Equal(t, 1, countRequests(backend, "eth_chainId")) require.Equal(t, 1, countRequests(backend, "eth_chainId"))
require.Equal(t, 1, countRequests(backend, "net_version")) require.Equal(t, 1, countRequests(backend, "net_version"))
require.Equal(t, 1, countRequests(backend, "eth_getBlockByHash"))
backend.Reset() backend.Reset()
res, _, err = client.SendBatchRPC( res, _, err = client.SendBatchRPC(
NewRPCReq("1", "eth_chainId", nil), NewRPCReq("1", "eth_chainId", nil),
NewRPCReq("1", "eth_call", []interface{}{`{"to":"0x1234"}`, "pending"}), NewRPCReq("1", "eth_call", []interface{}{`{"to":"0x1234"}`, "pending"}),
NewRPCReq("1", "net_version", nil), NewRPCReq("1", "net_version", nil),
NewRPCReq("1", "eth_getBlockByHash", []interface{}{"0xc6ef2fc5426d6ad6fd9e2a26abeab0aa2411b7ab17f30a99d3cb96aed1d1055b", "false"}),
) )
require.NoError(t, err) require.NoError(t, err)
RequireEqualJSON(t, []byte(asArray(goodChainIdResponse, goodEthCallResponse, goodNetVersionResponse)), res) RequireEqualJSON(t, []byte(asArray(goodChainIdResponse, goodEthCallResponse, goodNetVersionResponse, goodEthGetBlockByHash)), res)
require.Equal(t, 0, countRequests(backend, "eth_chainId")) require.Equal(t, 0, countRequests(backend, "eth_chainId"))
require.Equal(t, 0, countRequests(backend, "net_version")) require.Equal(t, 0, countRequests(backend, "net_version"))
require.Equal(t, 0, countRequests(backend, "eth_getBlockByHash"))
require.Equal(t, 1, countRequests(backend, "eth_call")) require.Equal(t, 1, countRequests(backend, "eth_call"))
} }
......
proxyd/integration_tests/testdata/caching.toml
View file @ 4b5422be
...@@ -27,3 +27,10 @@ net_version = "main" ...@@ -27,3 +27,10 @@ net_version = "main"
eth_getBlockByNumber = "main" eth_getBlockByNumber = "main"
eth_blockNumber = "main" eth_blockNumber = "main"
eth_call = "main" eth_call = "main"
eth_getBlockTransactionCountByHash = "main"
eth_getUncleCountByBlockHash = "main"
eth_getBlockByHash = "main"
eth_getTransactionByHash = "main"
eth_getTransactionByBlockHashAndIndex = "main"
eth_getUncleByBlockHashAndIndex = "main"
eth_getTransactionReceipt = "main"
proxyd/methods.go
View file @ 4b5422be
...@@ -3,15 +3,9 @@ package proxyd ...@@ -3,15 +3,9 @@ package proxyd
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "github.com/ethereum/go-ethereum/log"
"fmt" "strings"
"sync" "sync"
"github.com/ethereum/go-ethereum/common/hexutil"
)
var (
errInvalidRPCParams = errors.New("invalid RPC params")
) )
type RPCMethodHandler interface { type RPCMethodHandler interface {
...@@ -20,366 +14,27 @@ type RPCMethodHandler interface { ...@@ -20,366 +14,27 @@ type RPCMethodHandler interface {
} }
type StaticMethodHandler struct { type StaticMethodHandler struct {
cache interface{} cache Cache
m sync.RWMutex m sync.RWMutex
} }
func (e *StaticMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) { func (e *StaticMethodHandler) key(req *RPCReq) string {
e.m.RLock() // signature is a cache-friendly base64-encoded string with json.RawMessage param contents
cache := e.cache signature := string(req.Params)
e.m.RUnlock() return strings.Join([]string{"cache", req.Method, signature}, ":")
if cache == nil {
return nil, nil
}
return &RPCRes{
JSONRPC: req.JSONRPC,
Result: cache,
ID: req.ID,
}, nil
} }
func (e *StaticMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res *RPCRes) error { func (e *StaticMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
e.m.Lock()
if e.cache == nil { if e.cache == nil {
e.cache = res.Result
}
e.m.Unlock()
return nil
}
type EthGetBlockByNumberMethodHandler struct {
cache Cache
getLatestBlockNumFn GetLatestBlockNumFn
numBlockConfirmations int
}
func (e *EthGetBlockByNumberMethodHandler) cacheKey(req *RPCReq) string {
input, includeTx, err := decodeGetBlockByNumberParams(req.Params)
if err != nil {
return ""
}
return fmt.Sprintf("method:eth_getBlockByNumber:%s:%t", input, includeTx)
}
func (e *EthGetBlockByNumberMethodHandler) cacheable(req *RPCReq) (bool, error) {
blockNum, _, err := decodeGetBlockByNumberParams(req.Params)
if err != nil {
return false, err
}
return !isBlockDependentParam(blockNum), nil
}
func (e *EthGetBlockByNumberMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
if ok, err := e.cacheable(req); !ok || err != nil {
return nil, err
}
key := e.cacheKey(req)
return getImmutableRPCResponse(ctx, e.cache, key, req)
}
func (e *EthGetBlockByNumberMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res *RPCRes) error {
if ok, err := e.cacheable(req); !ok || err != nil {
return err
}
blockInput, _, err := decodeGetBlockByNumberParams(req.Params)
if err != nil {
return err
}
if isBlockDependentParam(blockInput) {
return nil
}
if blockInput != "earliest" {
curBlock, err := e.getLatestBlockNumFn(ctx)
if err != nil {
return err
}
blockNum, err := decodeBlockInput(blockInput)
if err != nil {
return err
}
if curBlock <= blockNum+uint64(e.numBlockConfirmations) {
return nil
}
}
key := e.cacheKey(req)
return putImmutableRPCResponse(ctx, e.cache, key, req, res)
}
type EthGetBlockRangeMethodHandler struct {
cache Cache
getLatestBlockNumFn GetLatestBlockNumFn
numBlockConfirmations int
}
func (e *EthGetBlockRangeMethodHandler) cacheKey(req *RPCReq) string {
start, end, includeTx, err := decodeGetBlockRangeParams(req.Params)
if err != nil {
return ""
}
return fmt.Sprintf("method:eth_getBlockRange:%s:%s:%t", start, end, includeTx)
}
func (e *EthGetBlockRangeMethodHandler) cacheable(req *RPCReq) (bool, error) {
start, end, _, err := decodeGetBlockRangeParams(req.Params)
if err != nil {
return false, err
}
return !isBlockDependentParam(start) && !isBlockDependentParam(end), nil
}
func (e *EthGetBlockRangeMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
if ok, err := e.cacheable(req); !ok || err != nil {
return nil, err
}
key := e.cacheKey(req)
return getImmutableRPCResponse(ctx, e.cache, key, req)
}
func (e *EthGetBlockRangeMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res *RPCRes) error {
if ok, err := e.cacheable(req); !ok || err != nil {
return err
}
start, end, _, err := decodeGetBlockRangeParams(req.Params)
if err != nil {
return err
}
curBlock, err := e.getLatestBlockNumFn(ctx)
if err != nil {
return err
}
if start != "earliest" {
startNum, err := decodeBlockInput(start)
if err != nil {
return err
}
if curBlock <= startNum+uint64(e.numBlockConfirmations) {
return nil
}
}
if end != "earliest" {
endNum, err := decodeBlockInput(end)
if err != nil {
return err
}
if curBlock <= endNum+uint64(e.numBlockConfirmations) {
return nil
}
}
key := e.cacheKey(req)
return putImmutableRPCResponse(ctx, e.cache, key, req, res)
}
type EthCallMethodHandler struct {
cache Cache
getLatestBlockNumFn GetLatestBlockNumFn
numBlockConfirmations int
}
func (e *EthCallMethodHandler) cacheable(params *ethCallParams, blockTag string) bool {
if isBlockDependentParam(blockTag) {
return false
}
if params.From != "" || params.Gas != "" {
return false
}
if params.Value != "" && params.Value != "0x0" {
return false
}
return true
}
func (e *EthCallMethodHandler) cacheKey(params *ethCallParams, blockTag string) string {
keyParams := fmt.Sprintf("%s:%s:%s", params.To, params.Data, blockTag)
return fmt.Sprintf("method:eth_call:%s", keyParams)
}
func (e *EthCallMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
params, blockTag, err := decodeEthCallParams(req)
if err != nil {
return nil, err
}
if !e.cacheable(params, blockTag) {
return nil, nil return nil, nil
} }
key := e.cacheKey(params, blockTag) e.m.RLock()
return getImmutableRPCResponse(ctx, e.cache, key, req) defer e.m.RUnlock()
}
func (e *EthCallMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res *RPCRes) error {
params, blockTag, err := decodeEthCallParams(req)
if err != nil {
return err
}
if !e.cacheable(params, blockTag) {
return nil
}
if blockTag != "earliest" {
curBlock, err := e.getLatestBlockNumFn(ctx)
if err != nil {
return err
}
blockNum, err := decodeBlockInput(blockTag)
if err != nil {
return err
}
if curBlock <= blockNum+uint64(e.numBlockConfirmations) {
return nil
}
}
key := e.cacheKey(params, blockTag)
return putImmutableRPCResponse(ctx, e.cache, key, req, res)
}
type EthBlockNumberMethodHandler struct {
getLatestBlockNumFn GetLatestBlockNumFn
}
func (e *EthBlockNumberMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
blockNum, err := e.getLatestBlockNumFn(ctx)
if err != nil {
return nil, err
}
return makeRPCRes(req, hexutil.EncodeUint64(blockNum)), nil
}
func (e *EthBlockNumberMethodHandler) PutRPCMethod(context.Context, *RPCReq, *RPCRes) error {
return nil
}
type EthGasPriceMethodHandler struct {
getLatestGasPrice GetLatestGasPriceFn
}
func (e *EthGasPriceMethodHandler) GetRPCMethod(ctx context.Context, req *RPCReq) (*RPCRes, error) {
gasPrice, err := e.getLatestGasPrice(ctx)
if err != nil {
return nil, err
}
return makeRPCRes(req, hexutil.EncodeUint64(gasPrice)), nil
}
func (e *EthGasPriceMethodHandler) PutRPCMethod(context.Context, *RPCReq, *RPCRes) error {
return nil
}
func isBlockDependentParam(s string) bool {
return s == "latest" ||
s == "pending" ||
s == "finalized" ||
s == "safe"
}
func decodeGetBlockByNumberParams(params json.RawMessage) (string, bool, error) {
var list []interface{}
if err := json.Unmarshal(params, &list); err != nil {
return "", false, err
}
if len(list) != 2 {
return "", false, errInvalidRPCParams
}
blockNum, ok := list[0].(string)
if !ok {
return "", false, errInvalidRPCParams
}
includeTx, ok := list[1].(bool)
if !ok {
return "", false, errInvalidRPCParams
}
if !validBlockInput(blockNum) {
return "", false, errInvalidRPCParams
}
return blockNum, includeTx, nil
}
func decodeGetBlockRangeParams(params json.RawMessage) (string, string, bool, error) {
var list []interface{}
if err := json.Unmarshal(params, &list); err != nil {
return "", "", false, err
}
if len(list) != 3 {
return "", "", false, errInvalidRPCParams
}
startBlockNum, ok := list[0].(string)
if !ok {
return "", "", false, errInvalidRPCParams
}
endBlockNum, ok := list[1].(string)
if !ok {
return "", "", false, errInvalidRPCParams
}
includeTx, ok := list[2].(bool)
if !ok {
return "", "", false, errInvalidRPCParams
}
if !validBlockInput(startBlockNum) || !validBlockInput(endBlockNum) {
return "", "", false, errInvalidRPCParams
}
return startBlockNum, endBlockNum, includeTx, nil
}
func decodeBlockInput(input string) (uint64, error) {
return hexutil.DecodeUint64(input)
}
type ethCallParams struct {
From string `json:"from"`
To string `json:"to"`
Gas string `json:"gas"`
GasPrice string `json:"gasPrice"`
Value string `json:"value"`
Data string `json:"data"`
}
func decodeEthCallParams(req *RPCReq) (*ethCallParams, string, error) {
var input []json.RawMessage
if err := json.Unmarshal(req.Params, &input); err != nil {
return nil, "", err
}
if len(input) != 2 {
return nil, "", fmt.Errorf("invalid eth_call parameters")
}
params := new(ethCallParams)
if err := json.Unmarshal(input[0], params); err != nil {
return nil, "", err
}
var blockTag string
if err := json.Unmarshal(input[1], &blockTag); err != nil {
return nil, "", err
}
return params, blockTag, nil
}
func validBlockInput(input string) bool {
if input == "earliest" ||
input == "latest" ||
input == "pending" ||
input == "finalized" ||
input == "safe" {
return true
}
_, err := decodeBlockInput(input)
return err == nil
}
func makeRPCRes(req *RPCReq, result interface{}) *RPCRes {
return &RPCRes{
JSONRPC: JSONRPCVersion,
ID: req.ID,
Result: result,
}
}
func getImmutableRPCResponse(ctx context.Context, cache Cache, key string, req *RPCReq) (*RPCRes, error) { key := e.key(req)
val, err := cache.Get(ctx, key) val, err := e.cache.Get(ctx, key)
if err != nil { if err != nil {
log.Error("error reading from cache", "key", key, "method", req.Method, "err", err)
return nil, err return nil, err
} }
if val == "" { if val == "" {
...@@ -388,6 +43,7 @@ func getImmutableRPCResponse(ctx context.Context, cache Cache, key string, req * ...@@ -388,6 +43,7 @@ func getImmutableRPCResponse(ctx context.Context, cache Cache, key string, req *
var result interface{} var result interface{}
if err := json.Unmarshal([]byte(val), &result); err != nil { if err := json.Unmarshal([]byte(val), &result); err != nil {
log.Error("error unmarshalling value from cache", "key", key, "method", req.Method, "err", err)
return nil, err return nil, err
} }
return &RPCRes{ return &RPCRes{
...@@ -397,10 +53,21 @@ func getImmutableRPCResponse(ctx context.Context, cache Cache, key string, req * ...@@ -397,10 +53,21 @@ func getImmutableRPCResponse(ctx context.Context, cache Cache, key string, req *
}, nil }, nil
} }
func putImmutableRPCResponse(ctx context.Context, cache Cache, key string, req *RPCReq, res *RPCRes) error { func (e *StaticMethodHandler) PutRPCMethod(ctx context.Context, req *RPCReq, res *RPCRes) error {
if key == "" { if e.cache == nil {
return nil return nil
} }
val := mustMarshalJSON(res.Result)
return cache.Put(ctx, key, string(val)) e.m.Lock()
defer e.m.Unlock()
key := e.key(req)
value := mustMarshalJSON(res.Result)
err := e.cache.Put(ctx, key, string(value))
if err != nil {
log.Error("error marshalling value to put into cache", "key", key, "method", req.Method, "err", err)
return err
}
return nil
} }
proxyd/metrics.go
View file @ 4b5422be
...@@ -182,6 +182,14 @@ var ( ...@@ -182,6 +182,14 @@ var (
"method", "method",
}) })
cacheErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
Namespace: MetricsNamespace,
Name: "cache_errors_total",
Help: "Number of cache errors.",
}, []string{
"method",
})
lvcErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ lvcErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
Namespace: MetricsNamespace, Namespace: MetricsNamespace,
Name: "lvc_errors_total", Name: "lvc_errors_total",
...@@ -374,6 +382,10 @@ func RecordCacheMiss(method string) { ...@@ -374,6 +382,10 @@ func RecordCacheMiss(method string) {
cacheMissesTotal.WithLabelValues(method).Inc() cacheMissesTotal.WithLabelValues(method).Inc()
} }
func RecordCacheError(method string) {
cacheMissesTotal.WithLabelValues(method).Inc()
}
func RecordBatchSize(size int) { func RecordBatchSize(size int) {
batchSizeHistogram.Observe(float64(size)) batchSizeHistogram.Observe(float64(size))
} }
......
proxyd/proxyd.go
View file @ 4b5422be
...@@ -213,17 +213,10 @@ func Start(config *Config) (*Server, func(), error) { ...@@ -213,17 +213,10 @@ func Start(config *Config) (*Server, func(), error) {
} }
var ( var (
rpcCache RPCCache cache Cache
blockNumLVC *EthLastValueCache rpcCache RPCCache
gasPriceLVC *EthLastValueCache
) )
if config.Cache.Enabled { if config.Cache.Enabled {
var (
cache Cache
blockNumFn GetLatestBlockNumFn
gasPriceFn GetLatestGasPriceFn
)
if config.Cache.BlockSyncRPCURL == "" { if config.Cache.BlockSyncRPCURL == "" {
return nil, nil, fmt.Errorf("block sync node required for caching") return nil, nil, fmt.Errorf("block sync node required for caching")
} }
...@@ -245,9 +238,7 @@ func Start(config *Config) (*Server, func(), error) { ...@@ -245,9 +238,7 @@ func Start(config *Config) (*Server, func(), error) {
} }
defer ethClient.Close() defer ethClient.Close()
blockNumLVC, blockNumFn = makeGetLatestBlockNumFn(ethClient, cache) rpcCache = newRPCCache(newCacheWithCompression(cache))
gasPriceLVC, gasPriceFn = makeGetLatestGasPriceFn(ethClient, cache)
rpcCache = newRPCCache(newCacheWithCompression(cache), blockNumFn, gasPriceFn, config.Cache.NumBlockConfirmations)
} }
srv, err := NewServer( srv, err := NewServer(
...@@ -345,12 +336,6 @@ func Start(config *Config) (*Server, func(), error) { ...@@ -345,12 +336,6 @@ func Start(config *Config) (*Server, func(), error) {
shutdownFunc := func() { shutdownFunc := func() {
log.Info("shutting down proxyd") log.Info("shutting down proxyd")
if blockNumLVC != nil {
blockNumLVC.Stop()
}
if gasPriceLVC != nil {
gasPriceLVC.Stop()
}
srv.Shutdown() srv.Shutdown()
if err := lim.FlushBackendWSConns(backendNames); err != nil { if err := lim.FlushBackendWSConns(backendNames); err != nil {
log.Error("error flushing backend ws conns", "err", err) log.Error("error flushing backend ws conns", "err", err)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment