certusone
/
wormhole
mirror of https://github.com/certusone/wormhole.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

CCQ: eth_call_by_timestamp (#3449) 

* CCQ: eth_call_by_timestamp

* CI: add CCQ sdk tests

* SDK test changes

* Add block parsing tests

* Code review rework

* More code review rework

* More rework

* Allow two blocks to have the same timestamps

* Restore timestamp check

* Restore timestamp check

* Minor code rework

---------

Co-authored-by: Evan Gray <battledingo@gmail.com>
This commit is contained in:
bruce-riley 2023-10-19 10:32:23 -05:00 committed by GitHub
parent c991d991db
commit d2db1616c5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 1764 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
Tiltfile
View File

@ -563,6 +563,16 @@ if ci_tests:
sync("./testing", "/app/testing"),
],
)
docker_build(
ref = "query-sdk-test-image",
context = ".",
dockerfile = "testing/Dockerfile.querysdk.test",
only = [],
live_update = [
sync("./sdk/js/src", "/app/sdk/js-query/src"),
sync("./testing", "/app/testing"),
],
)
k8s_yaml_with_ns(encode_yaml_stream(set_env_in_jobs(read_yaml_stream("devnet/tests.yaml"), "NUM_GUARDIANS", str(num_guardians))))
@ -591,6 +601,12 @@ if ci_tests:
trigger_mode = trigger_mode,
resource_deps = [], # node/hack/query/test/test_query.sh handles waiting for guardian, not having deps gets the build earlier
)
k8s_resource(
"query-sdk-ci-tests",
labels = ["ci"],
trigger_mode = trigger_mode,
resource_deps = [], # testing/querysdk.sh handles waiting for query-server, not having deps gets the build earlier
)
if terra_classic:
docker_build(
@ -912,4 +928,4 @@ if query_server:
],
labels = ["query-server"],
trigger_mode = trigger_mode
)
)

27
devnet/tests.yaml
View File

@ -97,3 +97,30 @@ spec:
- "/app/node/hack/query/test/success"
initialDelaySeconds: 5
periodSeconds: 5
---
kind: Job
apiVersion: batch/v1
metadata:
name: query-sdk-ci-tests
spec:
backoffLimit: 0
template:
spec:
restartPolicy: Never
containers:
- name: query-sdk-ci-tests
image: query-sdk-test-image
command:
- /bin/sh
- -c
- "bash /app/testing/querysdk.sh && touch /app/testing/success"
readinessProbe:
exec:
command:
- test
- -e
- "/app/testing/success"
initialDelaySeconds: 5
periodSeconds: 5
---

32
node/cmd/ccq/devnet.permissions.json
View File

@ -35,6 +35,38 @@
"contractAddress": "0xDDb64fE46a91D46ee29420539FC25FD07c5FEa3E",
"call": "0x18160ddd"
}
},
{
"ethCallByTimestamp": {
"note:": "Name of WETH on Goerli",
"chain": 2,
"contractAddress": "B4FBF271143F4FBf7B91A5ded31805e42b2208d6",
"call": "0x06fdde03"
}
},
{
"ethCallByTimestamp": {
"note:": "Total supply of WETH on Goerli",
"chain": 2,
"contractAddress": "0xB4FBF271143F4FBf7B91A5ded31805e42b2208d6",
"call": "0x18160ddd"
}
},
{
"ethCallByTimestamp": {
"note:": "Name of WETH on Devnet",
"chain": 2,
"contractAddress": "0xDDb64fE46a91D46ee29420539FC25FD07c5FEa3E",
"call": "0x06fdde03"
}
},
{
"ethCallByTimestamp": {
"note:": "Total supply of WETH on Devnet",
"chain": 2,
"contractAddress": "0xDDb64fE46a91D46ee29420539FC25FD07c5FEa3E",
"call": "0x18160ddd"
}
}
]
},

23
node/cmd/ccq/http.go
View File

@ -53,6 +53,17 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
return
}
// Decode the body first. This is because the library seems to hang if we receive a large body and return without decoding it.
// This could be a slight waste of resources, but should not be a DoS risk because we cap the max body size.
var q queryRequest
err := json.NewDecoder(http.MaxBytesReader(w, r.Body, MAX_BODY_SIZE)).Decode(&q)
if err != nil {
s.logger.Debug("failed to decode body", zap.Error(err))
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
// There should be one and only one API key in the header.
apiKeys, exists := r.Header["X-Api-Key"]
if !exists || len(apiKeys) != 1 {
@ -60,24 +71,16 @@ func (s *httpServer) handleQuery(w http.ResponseWriter, r *http.Request) {
http.Error(w, "api key is missing", http.StatusUnauthorized)
return
}
apiKey := apiKeys[0]
apiKey := strings.ToLower(apiKeys[0])
// Make sure the user is authorized before we go any farther.
_, exists = s.permissions[strings.ToLower(apiKey)]
_, exists = s.permissions[apiKey]
if !exists {
s.logger.Debug("invalid api key", zap.String("apiKey", apiKey))
http.Error(w, "invalid api key", http.StatusForbidden)
return
}
var q queryRequest
err := json.NewDecoder(http.MaxBytesReader(w, r.Body, MAX_BODY_SIZE)).Decode(&q)
if err != nil {
s.logger.Debug("failed to decode body", zap.Error(err))
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
queryRequestBytes, err := hex.DecodeString(q.Bytes)
if err != nil {
s.logger.Debug("failed to decode request bytes", zap.Error(err))

45
node/cmd/ccq/parse_config_test.go
View File

@ -157,7 +157,7 @@ func TestParseConfigUnsupportedCallType(t *testing.T) {
_, err := parseConfig([]byte(str))
require.Error(t, err)
assert.Equal(t, `unsupported call type for user "Test User", must be "ethCall"`, err.Error())
assert.Equal(t, `unsupported call type for user "Test User", must be "ethCall" or "ethCallByTimestamp"`, err.Error())
}
func TestParseConfigInvalidContractAddress(t *testing.T) {
@ -271,3 +271,46 @@ func TestParseConfigDuplicateAllowedCallForUser(t *testing.T) {
require.Error(t, err)
assert.Equal(t, `"ethCall:2:000000000000000000000000b4fbf271143f4fbf7b91a5ded31805e42b2208d6:06fdde03" is a duplicate allowed call for user "Test User"`, err.Error())
}
func TestParseConfigSuccess(t *testing.T) {
str := `
{
"permissions": [
{
"userName": "Test User",
"apiKey": "My_secret_key",
"allowedCalls": [
{
"ethCall": {
"note:": "Name of WETH on Goerli",
"chain": 2,
"contractAddress": "B4FBF271143F4FBf7B91A5ded31805e42b2208d6",
"call": "0x06fdde03"
}
},
{
"ethCallByTimestamp": {
"note:": "Name of WETH on Goerli",
"chain": 2,
"contractAddress": "B4FBF271143F4FBf7B91A5ded31805e42b2208d7",
"call": "0x06fdde03"
}
}
]
}
]
}`
perms, err := parseConfig([]byte(str))
require.NoError(t, err)
assert.Equal(t, 1, len(perms))
perm, exists := perms["my_secret_key"]
require.True(t, exists)
_, exists = perm.allowedCalls["ethCall:2:000000000000000000000000b4fbf271143f4fbf7b91a5ded31805e42b2208d6:06fdde03"]
assert.True(t, exists)
_, exists = perm.allowedCalls["ethCallByTimestamp:2:000000000000000000000000b4fbf271143f4fbf7b91a5ded31805e42b2208d7:06fdde03"]
assert.True(t, exists)
}

73
node/cmd/ccq/utils.go
View File

@ -65,7 +65,8 @@ type User struct {
}
type AllowedCall struct {
EthCall *EthCall `json:"ethCall"`
EthCall *EthCall `json:"ethCall"`
EthCallByTimestamp *EthCallByTimestamp `json:"ethCallByTimestamp"`
}
type EthCall struct {
@ -74,6 +75,12 @@ type EthCall struct {
Call string `json:"call"`
}
type EthCallByTimestamp struct {
Chain int `json:"chain"`
ContractAddress string `json:"contractAddress"`
Call string `json:"call"`
}
type Permissions map[string]*permissionEntry
type permissionEntry struct {
@ -132,28 +139,40 @@ func parseConfig(byteValue []byte) (Permissions, error) {
// Build the list of allowed calls for this API key.
allowedCalls := make(allowedCallsForUser)
for _, ac := range user.AllowedCalls {
var callKey string
if ac.EthCall == nil {
return nil, fmt.Errorf(`unsupported call type for user "%s", must be "ethCall"`, user.UserName)
var chain int
var callType, contractAddressStr, callStr string
// var contractAddressStr string
if ac.EthCall != nil {
callType = "ethCall"
chain = ac.EthCall.Chain
contractAddressStr = ac.EthCall.ContractAddress
callStr = ac.EthCall.Call
} else if ac.EthCallByTimestamp != nil {
callType = "ethCallByTimestamp"
chain = ac.EthCallByTimestamp.Chain
contractAddressStr = ac.EthCallByTimestamp.ContractAddress
callStr = ac.EthCallByTimestamp.Call
} else {
return nil, fmt.Errorf(`unsupported call type for user "%s", must be "ethCall" or "ethCallByTimestamp"`, user.UserName)
}
// Convert the contract address into a standard format like "000000000000000000000000b4fbf271143f4fbf7b91a5ded31805e42b2208d6".
contractAddress, err := vaa.StringToAddress(ac.EthCall.ContractAddress)
contractAddress, err := vaa.StringToAddress(contractAddressStr)
if err != nil {
return nil, fmt.Errorf(`invalid contract address "%s" for user "%s"`, ac.EthCall.ContractAddress, user.UserName)
return nil, fmt.Errorf(`invalid contract address "%s" for user "%s"`, contractAddressStr, user.UserName)
}
// The call should be the ABI four byte hex hash of the function signature. Parse it into a standard form of "06fdde03".
call, err := hex.DecodeString(strings.TrimPrefix(ac.EthCall.Call, "0x"))
call, err := hex.DecodeString(strings.TrimPrefix(callStr, "0x"))
if err != nil {
return nil, fmt.Errorf(`invalid eth call "%s" for user "%s"`, ac.EthCall.Call, user.UserName)
return nil, fmt.Errorf(`invalid eth call "%s" for user "%s"`, callStr, user.UserName)
}
if len(call) != ETH_CALL_SIG_LENGTH {
return nil, fmt.Errorf(`eth call "%s" for user "%s" has an invalid length, must be %d bytes`, ac.EthCall.Call, user.UserName, ETH_CALL_SIG_LENGTH)
return nil, fmt.Errorf(`eth call "%s" for user "%s" has an invalid length, must be %d bytes`, callStr, user.UserName, ETH_CALL_SIG_LENGTH)
}
// The permission key is the chain, contract address and call formatted as a colon separated string.
callKey = fmt.Sprintf("ethCall:%d:%s:%s", ac.EthCall.Chain, contractAddress, hex.EncodeToString(call))
callKey := fmt.Sprintf("%s:%d:%s:%s", callType, chain, contractAddress, hex.EncodeToString(call))
if _, exists := allowedCalls[callKey]; exists {
return nil, fmt.Errorf(`"%s" is a duplicate allowed call for user "%s"`, callKey, user.UserName)
@ -177,8 +196,7 @@ func parseConfig(byteValue []byte) (Permissions, error) {
// validateRequest verifies that this API key is allowed to do all of the calls in this request. In the case of an error, it returns the HTTP status.
func validateRequest(logger *zap.Logger, env common.Environment, perms Permissions, signerKey *ecdsa.PrivateKey, apiKey string, qr *gossipv1.SignedQueryRequest) (int, error) {
apiKey = strings.ToLower(apiKey)
permsForUser, exists := perms[strings.ToLower(apiKey)]
permsForUser, exists := perms[apiKey]
if !exists {
logger.Debug("invalid api key", zap.String("apiKey", apiKey))
return http.StatusForbidden, fmt.Errorf("invalid api key")
@ -240,6 +258,37 @@ func validateRequest(logger *zap.Logger, env common.Environment, perms Permissio
return http.StatusBadRequest, fmt.Errorf(`call "%s" not authorized`, callKey)
}
}
case *query.EthCallByTimestampQueryRequest:
for _, callData := range q.CallData {
if q.TargetTimestamp == 0 {
logger.Debug("eth call by timestamp must have a non-zero timestamp", zap.String("userName", permsForUser.userName))
return http.StatusBadRequest, fmt.Errorf("eth call by timestamp must have a non-zero timestamp")
}
// TODO: For now, the block hints are required!
if q.TargetBlockIdHint == "" {
logger.Debug("eth call by timestamp must have the target block hint", zap.String("userName", permsForUser.userName))
return http.StatusBadRequest, fmt.Errorf("eth call by timestamp must have the target block hint")
}
if q.FollowingBlockIdHint == "" {
logger.Debug("eth call by timestamp must have the following block hint", zap.String("userName", permsForUser.userName))
return http.StatusBadRequest, fmt.Errorf("eth call by timestamp must have the following block hint")
}
contractAddress, err := vaa.BytesToAddress(callData.To)
if err != nil {
logger.Debug("failed to parse contract address", zap.String("userName", permsForUser.userName), zap.String("contract", hex.EncodeToString(callData.To)), zap.Error(err))
return http.StatusBadRequest, fmt.Errorf("failed to parse contract address: %w", err)
}
if len(callData.Data) < ETH_CALL_SIG_LENGTH {
logger.Debug("eth call data must be at least four bytes", zap.String("userName", permsForUser.userName), zap.String("data", hex.EncodeToString(callData.Data)))
return http.StatusBadRequest, fmt.Errorf("eth call data must be at least four bytes")
}
call := hex.EncodeToString(callData.Data[0:ETH_CALL_SIG_LENGTH])
callKey := fmt.Sprintf("ethCallByTimestamp:%d:%s:%s", pcq.ChainId, contractAddress, call)
if _, exists := permsForUser.allowedCalls[callKey]; !exists {
logger.Debug("requested call not authorized", zap.String("userName", permsForUser.userName), zap.String("callKey", callKey))
return http.StatusBadRequest, fmt.Errorf(`call "%s" not authorized`, callKey)
}
}
default:
logger.Debug("unsupported query type", zap.String("userName", permsForUser.userName), zap.Any("type", pcq.Query))
return http.StatusBadRequest, fmt.Errorf("unsupported query type")

3
node/hack/query/test/query_test.go
View File

@ -189,7 +189,8 @@ func TestCrossChainQuery(t *testing.T) {
logger.Info("Waiting for message...")
var success bool
signers := map[int]bool{}
subCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
// The guardians can retry for up to a minute so we have to wait longer than that.
subCtx, cancel := context.WithTimeout(ctx, 75*time.Second)
defer cancel()
for {
envelope, err := sub.Next(subCtx)

117
node/pkg/query/query_test.go
View File

@ -45,17 +45,17 @@ var (
watcherChainsForTest = []vaa.ChainID{vaa.ChainIDPolygon, vaa.ChainIDBSC}
)
// createPerChainQueryForTesting creates a per chain query for use in tests. The To and Data fields are meaningless gibberish, not ABI.
func createPerChainQueryForTesting(
// createPerChainQueryForEthCall creates a per chain query for an eth_call for use in tests. The To and Data fields are meaningless gibberish, not ABI.
func createPerChainQueryForEthCall(
t *testing.T,
chainId vaa.ChainID,
block string,
numCalls int,
) *PerChainQueryRequest {
t.Helper()
callData := []*EthCallData{}
ethCallData := []*EthCallData{}
for count := 0; count < numCalls; count++ {
callData = append(callData, &EthCallData{
ethCallData = append(ethCallData, &EthCallData{
To: []byte(fmt.Sprintf("%-20s", fmt.Sprintf("To for %d:%d", chainId, count))),
Data: []byte(fmt.Sprintf("CallData for %d:%d", chainId, count)),
})
@ -63,7 +63,37 @@ func createPerChainQueryForTesting(
callRequest := &EthCallQueryRequest{
BlockId: block,
CallData: callData,
CallData: ethCallData,
}
return &PerChainQueryRequest{
ChainId: chainId,
Query: callRequest,
}
}
// createPerChainQueryForEthCallByTimestamp creates a per chain query for an eth_call_by_timestamp for use in tests. The To and Data fields are meaningless gibberish, not ABI.
func createPerChainQueryForEthCallByTimestamp(
t *testing.T,
chainId vaa.ChainID,
targetBlock string,
followingBlock string,
numCalls int,
) *PerChainQueryRequest {
t.Helper()
ethCallData := []*EthCallData{}
for count := 0; count < numCalls; count++ {
ethCallData = append(ethCallData, &EthCallData{
To: []byte(fmt.Sprintf("%-20s", fmt.Sprintf("To for %d:%d", chainId, count))),
Data: []byte(fmt.Sprintf("CallData for %d:%d", chainId, count)),
})
}
callRequest := &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
TargetBlockIdHint: targetBlock,
FollowingBlockIdHint: followingBlock,
CallData: ethCallData,
}
return &PerChainQueryRequest{
@ -129,6 +159,28 @@ func createExpectedResultsForTest(t *testing.T, perChainQueries []*PerChainQuery
ChainId: pcq.ChainId,
Response: resp,
})
case *EthCallByTimestampQueryRequest:
now := time.Now()
blockNum, err := strconv.ParseUint(strings.TrimPrefix(req.TargetBlockIdHint, "0x"), 16, 64)
if err != nil {
panic("invalid blockNum!")
}
resp := &EthCallByTimestampQueryResponse{
TargetBlockNumber: blockNum,
TargetBlockHash: ethCommon.HexToHash("0x9999bac44d09a7f69ee7941819b0a19c59ccb1969640cc513be09ef95ed2d8e2"),
TargetBlockTime: timeForTest(t, now),
FollowingBlockNumber: blockNum + 1,
FollowingBlockHash: ethCommon.HexToHash("0x9999bac44d09a7f69ee7941819b0a19c59ccb1969640cc513be09ef95ed2d8e3"),
FollowingBlockTime: timeForTest(t, time.Now().Add(10*time.Second)),
Results: [][]byte{},
}
for _, cd := range req.CallData {
resp.Results = append(resp.Results, []byte(hex.EncodeToString(cd.To)+":"+hex.EncodeToString(cd.Data)))
}
expectedResults = append(expectedResults, PerChainQueryResponse{
ChainId: pcq.ChainId,
Response: resp,
})
default:
panic("Invalid call data type!")
@ -412,7 +464,7 @@ func TestInvalidQueries(t *testing.T) {
// Query with a bad signature should fail.
md.resetState()
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
signedQueryRequest, _ = createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
signedQueryRequest.Signature[0] += 1 // Corrupt the signature.
md.signedQueryReqWriteC <- signedQueryRequest
@ -420,27 +472,50 @@ func TestInvalidQueries(t *testing.T) {
// Query for an unsupported chain should fail. The supported chains are defined in supportedChains in query.go
md.resetState()
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDAlgorand, "0x28d9630", 2)}
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDAlgorand, "0x28d9630", 2)}
signedQueryRequest, _ = createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
md.signedQueryReqWriteC <- signedQueryRequest
require.Nil(t, md.waitForResponse())
// Query for a chain that supports queries but that is not in the watcher channel map should fail.
md.resetState()
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDSepolia, "0x28d9630", 2)}
perChainQueries = []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDSepolia, "0x28d9630", 2)}
signedQueryRequest, _ = createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
md.signedQueryReqWriteC <- signedQueryRequest
require.Nil(t, md.waitForResponse())
}
func TestSingleQueryShouldSucceed(t *testing.T) {
func TestSingleEthCallQueryShouldSucceed(t *testing.T) {
ctx := context.Background()
logger := zap.NewNop()
md := createQueryHandlerForTest(t, ctx, logger, watcherChainsForTest)
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
md.setExpectedResults(expectedResults)
// Submit the query request to the handler.
md.signedQueryReqWriteC <- signedQueryRequest
// Wait until we receive a response or timeout.
queryResponsePublication := md.waitForResponse()
require.NotNil(t, queryResponsePublication)
assert.Equal(t, 1, md.getRequestsPerChain(vaa.ChainIDPolygon))
assert.True(t, validateResponseForTest(t, queryResponsePublication, signedQueryRequest, queryRequest, expectedResults))
}
func TestSingleEthCallByTimestampQueryShouldSucceed(t *testing.T) {
ctx := context.Background()
logger := zap.NewNop()
md := createQueryHandlerForTest(t, ctx, logger, watcherChainsForTest)
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForEthCallByTimestamp(t, vaa.ChainIDPolygon, "0x28d9630", "0x28d9631", 2)}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
md.setExpectedResults(expectedResults)
@ -464,8 +539,8 @@ func TestBatchOfTwoQueriesShouldSucceed(t *testing.T) {
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{
createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForTesting(t, vaa.ChainIDBSC, "0x28d9123", 3),
createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForEthCallByTimestamp(t, vaa.ChainIDBSC, "0x28d9123", "0x28d9124", 3),
}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
@ -490,7 +565,7 @@ func TestQueryWithLimitedRetriesShouldSucceed(t *testing.T) {
md := createQueryHandlerForTest(t, ctx, logger, watcherChainsForTest)
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
md.setExpectedResults(expectedResults)
@ -517,7 +592,7 @@ func TestQueryWithRetryDueToTimeoutShouldSucceed(t *testing.T) {
md := createQueryHandlerForTest(t, ctx, logger, watcherChainsForTest)
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
md.setExpectedResults(expectedResults)
@ -544,8 +619,8 @@ func TestQueryWithTooManyRetriesShouldFail(t *testing.T) {
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{
createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForTesting(t, vaa.ChainIDBSC, "0x28d9123", 3),
createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForEthCall(t, vaa.ChainIDBSC, "0x28d9123", 3),
}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
@ -576,8 +651,8 @@ func TestQueryWithLimitedRetriesOnMultipleChainsShouldSucceed(t *testing.T) {
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{
createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForTesting(t, vaa.ChainIDBSC, "0x28d9123", 3),
createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForEthCall(t, vaa.ChainIDBSC, "0x28d9123", 3),
}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
@ -610,8 +685,8 @@ func TestFatalErrorOnPerChainQueryShouldCauseRequestToFail(t *testing.T) {
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{
createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForTesting(t, vaa.ChainIDBSC, "0x28d9123", 3),
createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2),
createPerChainQueryForEthCall(t, vaa.ChainIDBSC, "0x28d9123", 3),
}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
@ -638,7 +713,7 @@ func TestPublishRetrySucceeds(t *testing.T) {
md := createQueryHandlerForTestWithoutPublisher(t, ctx, logger, watcherChainsForTest)
// Create the request and the expected results. Give the expected results to the mock.
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForTesting(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
perChainQueries := []*PerChainQueryRequest{createPerChainQueryForEthCall(t, vaa.ChainIDPolygon, "0x28d9630", 2)}
signedQueryRequest, queryRequest := createSignedQueryRequestForTesting(t, md.sk, perChainQueries)
expectedResults := createExpectedResultsForTest(t, queryRequest.PerChainQueries)
md.setExpectedResults(expectedResults)

218
node/pkg/query/request.go
View File

@ -51,13 +51,39 @@ const EthCallQueryRequestType ChainSpecificQueryType = 1
// EthCallQueryRequest implements ChainSpecificQuery for an EVM eth_call query request.
type EthCallQueryRequest struct {
// BlockId identifies the block to be queried. It mus be a hex string starting with 0x. It may be a block number or a block hash.
// BlockId identifies the block to be queried. It must be a hex string starting with 0x. It may be a block number or a block hash.
BlockId string
// CallData is an array of specific queries to be performed on the specified block, in a single RPC call.
CallData []*EthCallData
}
func (ecr *EthCallQueryRequest) CallDataList() []*EthCallData {
return ecr.CallData
}
// EthCallByTimestampQueryRequestType is the type of an EVM eth_call_by_timestamp query request.
const EthCallByTimestampQueryRequestType ChainSpecificQueryType = 2
// EthCallByTimestampQueryRequest implements ChainSpecificQuery for an EVM eth_call_by_timestamp query request.
type EthCallByTimestampQueryRequest struct {
// TargetTimeInUs specifies the desired timestamp in microseconds.
TargetTimestamp uint64
// TargetBlockIdHint is optional. If specified, it identifies the block prior to the desired timestamp. It must be a hex string starting with 0x. It may be a block number or a block hash.
TargetBlockIdHint string
// FollowingBlockIdHint is optional. If specified, it identifies the block immediately following the desired timestamp. It must be a hex string starting with 0x. It may be a block number or a block hash.
FollowingBlockIdHint string
// CallData is an array of specific queries to be performed on the specified block, in a single RPC call.
CallData []*EthCallData
}
func (ecr *EthCallByTimestampQueryRequest) CallDataList() []*EthCallData {
return ecr.CallData
}
// EthCallData specifies the parameters to a single EVM eth_call request.
type EthCallData struct {
// To specifies the contract address to be queried.
@ -264,6 +290,12 @@ func (perChainQuery *PerChainQueryRequest) UnmarshalFromReader(reader *bytes.Rea
return fmt.Errorf("failed to unmarshal eth call request: %w", err)
}
perChainQuery.Query = &q
case EthCallByTimestampQueryRequestType:
q := EthCallByTimestampQueryRequest{}
if err := q.UnmarshalFromReader(reader); err != nil {
return fmt.Errorf("failed to unmarshal eth call by timestamp request: %w", err)
}
perChainQuery.Query = &q
default:
return fmt.Errorf("unsupported query type: %d", queryType)
}
@ -317,10 +349,17 @@ func (left *PerChainQueryRequest) Equal(right *PerChainQueryRequest) bool {
case *EthCallQueryRequest:
return leftEcq.Equal(rightEcd)
default:
panic("unsupported query type on right") // We checked this above!
panic("unsupported query type on right, must be eth_call")
}
case *EthCallByTimestampQueryRequest:
switch rightEcd := right.Query.(type) {
case *EthCallByTimestampQueryRequest:
return leftEcq.Equal(rightEcd)
default:
panic("unsupported query type on right, must be eth_call_by_timestamp")
}
default:
panic("unsupported query type on left") // We checked this above!
panic("unsupported query type on left")
}
}
@ -362,12 +401,12 @@ func (ecd *EthCallQueryRequest) Unmarshal(data []byte) error {
func (ecd *EthCallQueryRequest) UnmarshalFromReader(reader *bytes.Reader) error {
blockIdLen := uint32(0)
if err := binary.Read(reader, binary.BigEndian, &blockIdLen); err != nil {
return fmt.Errorf("failed to read call Data len: %w", err)
return fmt.Errorf("failed to read block id len: %w", err)
}
blockId := make([]byte, blockIdLen)
if n, err := reader.Read(blockId[:]); err != nil || n != int(blockIdLen) {
return fmt.Errorf("failed to read call To [%d]: %w", n, err)
return fmt.Errorf("failed to read block id [%d]: %w", n, err)
}
ecd.BlockId = string(blockId[:])
@ -388,7 +427,7 @@ func (ecd *EthCallQueryRequest) UnmarshalFromReader(reader *bytes.Reader) error
}
data := make([]byte, dataLen)
if n, err := reader.Read(data[:]); err != nil || n != int(dataLen) {
return fmt.Errorf("failed to read call To [%d]: %w", n, err)
return fmt.Errorf("failed to read call data [%d]: %w", n, err)
}
callData := &EthCallData{
@ -454,8 +493,173 @@ func (left *EthCallQueryRequest) Equal(right *EthCallQueryRequest) bool {
return true
}
//
// Implementation of EthCallByTimestampQueryRequest, which implements the ChainSpecificQuery interface.
//
func (e *EthCallByTimestampQueryRequest) Type() ChainSpecificQueryType {
return EthCallByTimestampQueryRequestType
}
// Marshal serializes the binary representation of an EVM eth_call_by_timestamp request.
// This method calls Validate() and relies on it to range checks lengths, etc.
func (ecd *EthCallByTimestampQueryRequest) Marshal() ([]byte, error) {
if err := ecd.Validate(); err != nil {
return nil, err
}
buf := new(bytes.Buffer)
vaa.MustWrite(buf, binary.BigEndian, ecd.TargetTimestamp)
vaa.MustWrite(buf, binary.BigEndian, uint32(len(ecd.TargetBlockIdHint)))
buf.Write([]byte(ecd.TargetBlockIdHint))
vaa.MustWrite(buf, binary.BigEndian, uint32(len(ecd.FollowingBlockIdHint)))
buf.Write([]byte(ecd.FollowingBlockIdHint))
vaa.MustWrite(buf, binary.BigEndian, uint8(len(ecd.CallData)))
for _, callData := range ecd.CallData {
buf.Write(callData.To)
vaa.MustWrite(buf, binary.BigEndian, uint32(len(callData.Data)))
buf.Write(callData.Data)
}
return buf.Bytes(), nil
}
// Unmarshal deserializes an EVM eth_call_by_timestamp query from a byte array
func (ecd *EthCallByTimestampQueryRequest) Unmarshal(data []byte) error {
reader := bytes.NewReader(data[:])
return ecd.UnmarshalFromReader(reader)
}
// UnmarshalFromReader deserializes an EVM eth_call_by_timestamp query from a byte array
func (ecd *EthCallByTimestampQueryRequest) UnmarshalFromReader(reader *bytes.Reader) error {
if err := binary.Read(reader, binary.BigEndian, &ecd.TargetTimestamp); err != nil {
return fmt.Errorf("failed to read timestamp: %w", err)
}
blockIdHintLen := uint32(0)
if err := binary.Read(reader, binary.BigEndian, &blockIdHintLen); err != nil {
return fmt.Errorf("failed to read target block id hint len: %w", err)
}
targetBlockIdHint := make([]byte, blockIdHintLen)
if n, err := reader.Read(targetBlockIdHint[:]); err != nil || n != int(blockIdHintLen) {
return fmt.Errorf("failed to read target block id hint [%d]: %w", n, err)
}
ecd.TargetBlockIdHint = string(targetBlockIdHint[:])
blockIdHintLen = uint32(0)
if err := binary.Read(reader, binary.BigEndian, &blockIdHintLen); err != nil {
return fmt.Errorf("failed to read following block id hint len: %w", err)
}
followingBlockIdHint := make([]byte, blockIdHintLen)
if n, err := reader.Read(followingBlockIdHint[:]); err != nil || n != int(blockIdHintLen) {
return fmt.Errorf("failed to read following block id hint [%d]: %w", n, err)
}
ecd.FollowingBlockIdHint = string(followingBlockIdHint[:])
numCallData := uint8(0)
if err := binary.Read(reader, binary.BigEndian, &numCallData); err != nil {
return fmt.Errorf("failed to read number of call data entries: %w", err)
}
for count := 0; count < int(numCallData); count++ {
to := [EvmContractAddressLength]byte{}
if n, err := reader.Read(to[:]); err != nil || n != EvmContractAddressLength {
return fmt.Errorf("failed to read call To [%d]: %w", n, err)
}
dataLen := uint32(0)
if err := binary.Read(reader, binary.BigEndian, &dataLen); err != nil {
return fmt.Errorf("failed to read call Data len: %w", err)
}
data := make([]byte, dataLen)
if n, err := reader.Read(data[:]); err != nil || n != int(dataLen) {
return fmt.Errorf("failed to read call data [%d]: %w", n, err)
}
callData := &EthCallData{
To: to[:],
Data: data[:],
}
ecd.CallData = append(ecd.CallData, callData)
}
return nil
}
// Validate does basic validation on an EVM eth_call_by_timestamp query.
func (ecd *EthCallByTimestampQueryRequest) Validate() error {
if ecd.TargetTimestamp == 0 {
return fmt.Errorf("target timestamp may not be zero")
}
if len(ecd.TargetBlockIdHint) > math.MaxUint32 {
return fmt.Errorf("target block id hint too long")
}
if !strings.HasPrefix(ecd.TargetBlockIdHint, "0x") {
return fmt.Errorf("target block id must be a hex number or hash starting with 0x")
}
if len(ecd.FollowingBlockIdHint) > math.MaxUint32 {
return fmt.Errorf("following block id hint too long")
}
if !strings.HasPrefix(ecd.FollowingBlockIdHint, "0x") {
return fmt.Errorf("following block id must be a hex number or hash starting with 0x")
}
if len(ecd.CallData) <= 0 {
return fmt.Errorf("does not contain any call data")
}
if len(ecd.CallData) > math.MaxUint8 {
return fmt.Errorf("too many call data entries")
}
for _, callData := range ecd.CallData {
if callData.To == nil || len(callData.To) <= 0 {
return fmt.Errorf("no call data to")
}
if len(callData.To) != EvmContractAddressLength {
return fmt.Errorf("invalid length for To contract")
}
if callData.Data == nil || len(callData.Data) <= 0 {
return fmt.Errorf("no call data data")
}
if len(callData.Data) > math.MaxUint32 {
return fmt.Errorf("call data data too long")
}
}
return nil
}
// Equal verifies that two EVM eth_call_by_timestamp queries are equal.
func (left *EthCallByTimestampQueryRequest) Equal(right *EthCallByTimestampQueryRequest) bool {
if left.TargetTimestamp != right.TargetTimestamp {
return false
}
if left.TargetBlockIdHint != right.TargetBlockIdHint {
return false
}
if left.FollowingBlockIdHint != right.FollowingBlockIdHint {
return false
}
if len(left.CallData) != len(right.CallData) {
return false
}
for idx := range left.CallData {
if !bytes.Equal(left.CallData[idx].To, right.CallData[idx].To) {
return false
}
if !bytes.Equal(left.CallData[idx].Data, right.CallData[idx].Data) {
return false
}
}
return true
}
func ValidatePerChainQueryRequestType(qt ChainSpecificQueryType) error {
if qt != EthCallQueryRequestType {
if qt != EthCallQueryRequestType && qt != EthCallByTimestampQueryRequestType {
return fmt.Errorf("invalid query request type: %d", qt)
}
return nil

156
node/pkg/query/request_test.go
View File

@ -45,19 +45,31 @@ func createQueryRequestForTesting(t *testing.T, chainId vaa.ChainID) *QueryReque
Data: data2,
},
}
callRequest := &EthCallQueryRequest{
callRequest1 := &EthCallQueryRequest{
BlockId: block,
CallData: callData,
}
perChainQuery := &PerChainQueryRequest{
perChainQuery1 := &PerChainQueryRequest{
ChainId: chainId,
Query: callRequest,
Query: callRequest1,
}
callRequest2 := &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
TargetBlockIdHint: "0x28d9630",
FollowingBlockIdHint: "0x28d9631",
CallData: callData,
}
perChainQuery2 := &PerChainQueryRequest{
ChainId: chainId,
Query: callRequest2,
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
PerChainQueries: []*PerChainQueryRequest{perChainQuery1, perChainQuery2},
}
return queryRequest
@ -342,6 +354,142 @@ func TestMarshalOfEthCallQueryWithWrongToLengthShouldFail(t *testing.T) {
require.Error(t, err)
}
///////////// EthCallByTimestamp tests ////////////////////////////////////////
func TestMarshalOfEthCallByTimestampQueryWithNilToShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: nil,
Data: []byte("This can't be zero length"),
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
func TestMarshalOfEthCallByTimestampQueryWithEmptyToShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: []byte{},
Data: []byte("This can't be zero length"),
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
func TestMarshalOfEthCallByTimestampQueryWithWrongLengthToShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: []byte("TooShort"),
Data: []byte("This can't be zero length"),
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
func TestMarshalOfEthCallByTimestampQueryWithNilDataShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: []byte(fmt.Sprintf("%-20s", fmt.Sprintf("To for %d", 0))),
Data: nil,
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
func TestMarshalOfEthCallByTimestampQueryWithEmptyDataShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: []byte(fmt.Sprintf("%-20s", fmt.Sprintf("To for %d", 0))),
Data: []byte{},
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
func TestMarshalOfEthCallByTimestampQueryWithWrongToLengthShouldFail(t *testing.T) {
perChainQuery := &PerChainQueryRequest{
ChainId: vaa.ChainIDPolygon,
Query: &EthCallByTimestampQueryRequest{
TargetTimestamp: 1697216322000000,
CallData: []*EthCallData{
{
To: []byte("This is too short!"),
Data: []byte("This can't be zero length"),
},
},
},
}
queryRequest := &QueryRequest{
Nonce: 1,
PerChainQueries: []*PerChainQueryRequest{perChainQuery},
}
_, err := queryRequest.Marshal()
require.Error(t, err)
}
///////////// End of EthCallByTimestamp tests /////////////////////////////////
func TestPostSignedQueryRequestShouldFailIfNoOneIsListening(t *testing.T) {
queryRequest := createQueryRequestForTesting(t, vaa.ChainIDPolygon)
queryRequestBytes, err := queryRequest.Marshal()

188
node/pkg/query/response.go
View File

@ -84,6 +84,19 @@ type EthCallQueryResponse struct {
Results [][]byte
}
// EthCallByTimestampQueryResponse implements ChainSpecificResponse for an EVM eth_call_by_timestamp query response.
type EthCallByTimestampQueryResponse struct {
TargetBlockNumber uint64
TargetBlockHash common.Hash
TargetBlockTime time.Time
FollowingBlockNumber uint64
FollowingBlockHash common.Hash
FollowingBlockTime time.Time
// Results is the array of responses matching CallData in EthCallByTimestampQueryRequest
Results [][]byte
}
//
// Implementation of QueryResponsePublication.
//
@ -328,6 +341,12 @@ func (perChainResponse *PerChainQueryResponse) UnmarshalFromReader(reader *bytes
return fmt.Errorf("failed to unmarshal eth call response: %w", err)
}
perChainResponse.Response = &r
case EthCallByTimestampQueryRequestType:
r := EthCallByTimestampQueryResponse{}
if err := r.UnmarshalFromReader(reader); err != nil {
return fmt.Errorf("failed to unmarshal eth call by timestamp response: %w", err)
}
perChainResponse.Response = &r
default:
return fmt.Errorf("unsupported query type: %d", queryType)
}
@ -383,6 +402,13 @@ func (left *PerChainQueryResponse) Equal(right *PerChainQueryResponse) bool {
default:
panic("unsupported query type on right") // We checked this above!
}
case *EthCallByTimestampQueryResponse:
switch rightEcd := right.Response.(type) {
case *EthCallByTimestampQueryResponse:
return leftEcq.Equal(rightEcd)
default:
panic("unsupported query type on right") // We checked this above!
}
default:
panic("unsupported query type on left") // We checked this above!
}
@ -494,6 +520,168 @@ func (left *EthCallQueryResponse) Equal(right *EthCallQueryResponse) bool {
return false
}
if left.Time != right.Time {
return false
}
if len(left.Results) != len(right.Results) {
return false
}
for idx := range left.Results {
if !bytes.Equal(left.Results[idx], right.Results[idx]) {
return false
}
}
return true
}
//
// Implementation of EthCallByTimestampQueryResponse, which implements the ChainSpecificResponse for an EVM eth_call_by_timestamp query response.
//
func (e *EthCallByTimestampQueryResponse) Type() ChainSpecificQueryType {
return EthCallByTimestampQueryRequestType
}
// Marshal serializes the binary representation of an EVM eth_call response.
// This method calls Validate() and relies on it to range checks lengths, etc.
func (ecr *EthCallByTimestampQueryResponse) Marshal() ([]byte, error) {
if err := ecr.Validate(); err != nil {
return nil, err
}
buf := new(bytes.Buffer)
vaa.MustWrite(buf, binary.BigEndian, ecr.TargetBlockNumber)
buf.Write(ecr.TargetBlockHash[:])
vaa.MustWrite(buf, binary.BigEndian, ecr.TargetBlockTime.UnixMicro())
vaa.MustWrite(buf, binary.BigEndian, ecr.FollowingBlockNumber)
buf.Write(ecr.FollowingBlockHash[:])
vaa.MustWrite(buf, binary.BigEndian, ecr.FollowingBlockTime.UnixMicro())
vaa.MustWrite(buf, binary.BigEndian, uint8(len(ecr.Results)))
for idx := range ecr.Results {
vaa.MustWrite(buf, binary.BigEndian, uint32(len(ecr.Results[idx])))
buf.Write(ecr.Results[idx])
}
return buf.Bytes(), nil
}
// Unmarshal deserializes an EVM eth_call response from a byte array
func (ecr *EthCallByTimestampQueryResponse) Unmarshal(data []byte) error {
reader := bytes.NewReader(data[:])
return ecr.UnmarshalFromReader(reader)
}
// UnmarshalFromReader deserializes an EVM eth_call response from a byte array
func (ecr *EthCallByTimestampQueryResponse) UnmarshalFromReader(reader *bytes.Reader) error {
if err := binary.Read(reader, binary.BigEndian, &ecr.TargetBlockNumber); err != nil {
return fmt.Errorf("failed to read response target block number: %w", err)
}
responseHash := common.Hash{}
if n, err := reader.Read(responseHash[:]); err != nil || n != 32 {
return fmt.Errorf("failed to read response target block hash [%d]: %w", n, err)
}
ecr.TargetBlockHash = responseHash
unixMicros := int64(0)
if err := binary.Read(reader, binary.BigEndian, &unixMicros); err != nil {
return fmt.Errorf("failed to read response target block timestamp: %w", err)
}
ecr.TargetBlockTime = time.UnixMicro(unixMicros)
if err := binary.Read(reader, binary.BigEndian, &ecr.FollowingBlockNumber); err != nil {
return fmt.Errorf("failed to read response following block number: %w", err)
}
responseHash = common.Hash{}
if n, err := reader.Read(responseHash[:]); err != nil || n != 32 {
return fmt.Errorf("failed to read response following block hash [%d]: %w", n, err)
}
ecr.FollowingBlockHash = responseHash
unixMicros = int64(0)
if err := binary.Read(reader, binary.BigEndian, &unixMicros); err != nil {
return fmt.Errorf("failed to read response following block timestamp: %w", err)
}
ecr.FollowingBlockTime = time.UnixMicro(unixMicros)
numResults := uint8(0)
if err := binary.Read(reader, binary.BigEndian, &numResults); err != nil {
return fmt.Errorf("failed to read number of results: %w", err)
}
for count := 0; count < int(numResults); count++ {
resultLen := uint32(0)
if err := binary.Read(reader, binary.BigEndian, &resultLen); err != nil {
return fmt.Errorf("failed to read result len: %w", err)
}
result := make([]byte, resultLen)
if n, err := reader.Read(result[:]); err != nil || n != int(resultLen) {
return fmt.Errorf("failed to read result [%d]: %w", n, err)
}
ecr.Results = append(ecr.Results, result)
}
return nil
}
// Validate does basic validation on an EVM eth_call response.
func (ecr *EthCallByTimestampQueryResponse) Validate() error {
// Not checking for block numbers == 0, because maybe that could happen??
if len(ecr.TargetBlockHash) != 32 {
return fmt.Errorf("invalid length for target block hash")
}
if len(ecr.FollowingBlockHash) != 32 {
return fmt.Errorf("invalid length for following block hash")
}
if len(ecr.Results) <= 0 {
return fmt.Errorf("does not contain any results")
}
if len(ecr.Results) > math.MaxUint8 {
return fmt.Errorf("too many results")
}
for _, result := range ecr.Results {
if len(result) > math.MaxUint32 {
return fmt.Errorf("result too long")
}
}
return nil
}
// Equal verifies that two EVM eth_call responses are equal.
func (left *EthCallByTimestampQueryResponse) Equal(right *EthCallByTimestampQueryResponse) bool {
if left.TargetBlockNumber != right.TargetBlockNumber {
return false
}
if !bytes.Equal(left.TargetBlockHash.Bytes(), right.TargetBlockHash.Bytes()) {
return false
}
if left.TargetBlockTime != right.TargetBlockTime {
return false
}
if left.FollowingBlockNumber != right.FollowingBlockNumber {
return false
}
if !bytes.Equal(left.FollowingBlockHash.Bytes(), right.FollowingBlockHash.Bytes()) {
return false
}
if left.FollowingBlockTime != right.FollowingBlockTime {
return false
}
if len(left.Results) != len(right.Results) {
return false
}

18
node/pkg/query/response_test.go
View File

@ -42,6 +42,24 @@ func createQueryResponseFromRequest(t *testing.T, queryRequest *QueryRequest) *Q
Results: results,
},
})
case *EthCallByTimestampQueryRequest:
results := [][]byte{}
for idx := range req.CallData {
result := []byte([]byte(fmt.Sprintf("Result %d", idx)))
results = append(results, result[:])
}
perChainResponses = append(perChainResponses, &PerChainQueryResponse{
ChainId: pcr.ChainId,
Response: &EthCallByTimestampQueryResponse{
TargetBlockNumber: uint64(1000 + idx),
TargetBlockHash: ethCommon.HexToHash("0x9999bac44d09a7f69ee7941819b0a19c59ccb1969640cc513be09ef95ed2d8e2"),
TargetBlockTime: timeForTest(t, time.Now()),
FollowingBlockNumber: uint64(1000 + idx + 1),
FollowingBlockHash: ethCommon.HexToHash("0x9999bac44d09a7f69ee7941819b0a19c59ccb1969640cc513be09ef95ed2d8e3"),
FollowingBlockTime: timeForTest(t, time.Now().Add(10*time.Second)),
Results: results,
},
})
default:
panic("invalid query type!")
}

46
node/pkg/watchers/evm/ccq_test.go Normal file
View File

@ -0,0 +1,46 @@
package evm
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestCcqCreateBlockRequest(t *testing.T) {
type test struct {
input string
blockMethod string
callBlockArgAsJSON string
errMsg string
}
tests := []test{
// Failure cases:
{input: "", blockMethod: "", callBlockArgAsJSON: "", errMsg: "block id is required"},
{input: "deadbeef", blockMethod: "", callBlockArgAsJSON: "", errMsg: "block id must start with 0x"},
{input: "0xHelloWorld", blockMethod: "", callBlockArgAsJSON: "", errMsg: "block id is not valid hex"},
// Success cases:
{input: "0xb96d7a", blockMethod: "eth_getBlockByNumber", callBlockArgAsJSON: `"0xb96d7a"`, errMsg: ""},
{input: "0xb96d7a4751d4ec70a6278a92d361e52821416bb6966aabeb596b81f92f4a6263", blockMethod: "eth_getBlockByHash", callBlockArgAsJSON: `{"blockHash":"0xb96d7a4751d4ec70a6278a92d361e52821416bb6966aabeb596b81f92f4a6263","requireCanonical":true}`, errMsg: ""},
}
for _, tc := range tests {
t.Run(tc.input, func(t *testing.T) {
blockMethod, callBlockArg, err := ccqCreateBlockRequest(tc.input)
if tc.errMsg == "" {
require.NoError(t, err)
assert.Equal(t, tc.blockMethod, blockMethod)
bytes, err := json.Marshal(callBlockArg)
require.NoError(t, err)
assert.Equal(t, tc.callBlockArgAsJSON, string(bytes))
} else {
assert.EqualError(t, err, tc.errMsg)
}
})
}
}

710
node/pkg/watchers/evm/watcher.go
View File

@ -2,9 +2,11 @@ package evm
import (
"context"
"encoding/hex"
"fmt"
"math"
"math/big"
"strings"
"sync"
"sync/atomic"
"time"
@ -130,6 +132,8 @@ type (
// These parameters are currently only used for Polygon and should be set via SetRootChainParams()
rootChainRpc string
rootChainContract string
ccqMaxBlockNumber *big.Int
}
pendingKey struct {
@ -173,6 +177,7 @@ func NewEthWatcher(
queryResponseC: queryResponseC,
pending: map[pendingKey]*pendingMessage{},
unsafeDevMode: unsafeDevMode,
ccqMaxBlockNumber: big.NewInt(0).SetUint64(math.MaxUint64),
}
}
@ -960,8 +965,10 @@ func (w *Watcher) SetMaxWaitConfirmations(maxWaitConfirmations uint64) {
w.maxWaitConfirmations = maxWaitConfirmations
}
// ccqSendQueryResponse sends an error response back to the query handler.
func (w *Watcher) ccqSendQueryResponse(logger *zap.Logger, req *query.PerChainQueryInternal, status query.QueryStatus, resp *query.EthCallQueryResponse) {
// TODO: Once PR #3449 lands, move all of this to ccq.go.
// ccqSendQueryResponseForEthCall sends an error response back to the query handler.
func (w *Watcher) ccqSendQueryResponseForEthCall(logger *zap.Logger, req *query.PerChainQueryInternal, status query.QueryStatus, resp *query.EthCallQueryResponse) {
queryResponse := query.CreatePerChainQueryResponseInternal(req.RequestID, req.RequestIdx, req.Request.ChainId, status, resp)
select {
case w.queryResponseC <- queryResponse:
@ -971,8 +978,29 @@ func (w *Watcher) ccqSendQueryResponse(logger *zap.Logger, req *query.PerChainQu
}
}
// ccqSendQueryResponseForEthCallByTimestamp sends an error response back to the query handler.
func (w *Watcher) ccqSendQueryResponseForEthCallByTimestamp(logger *zap.Logger, req *query.PerChainQueryInternal, status query.QueryStatus, resp *query.EthCallByTimestampQueryResponse) {
queryResponse := query.CreatePerChainQueryResponseInternal(req.RequestID, req.RequestIdx, req.Request.ChainId, status, resp)
select {
case w.queryResponseC <- queryResponse:
logger.Debug("published query response error to handler", zap.String("component", "ccqevm"))
default:
logger.Error("failed to published query response error to handler", zap.String("component", "ccqevm"))
}
}
// ccqSendQueryResponseForError sends an error response back to the query handler.
func (w *Watcher) ccqSendQueryResponseForError(logger *zap.Logger, req *query.PerChainQueryInternal, status query.QueryStatus) {
queryResponse := query.CreatePerChainQueryResponseInternal(req.RequestID, req.RequestIdx, req.Request.ChainId, status, nil)
select {
case w.queryResponseC <- queryResponse:
logger.Debug("published query response error to handler", zap.String("component", "ccqevm"))
default:
logger.Error("failed to published query response error to handler", zap.String("component", "ccqevm"))
}
}
func (w *Watcher) ccqHandleQuery(logger *zap.Logger, ctx context.Context, queryRequest *query.PerChainQueryInternal) {
ccqMaxBlockNumber := big.NewInt(0).SetUint64(math.MaxUint64)
// This can't happen unless there is a programming error - the caller
// is expected to send us only requests for our chainID.
@ -982,197 +1010,497 @@ func (w *Watcher) ccqHandleQuery(logger *zap.Logger, ctx context.Context, queryR
switch req := queryRequest.Request.Query.(type) {
case *query.EthCallQueryRequest:
block := req.BlockId
logger.Info("received query request",
zap.String("block", block),
zap.Int("numRequests", len(req.CallData)),
)
timeout, cancel := context.WithTimeout(ctx, 5*time.Second)
// like https://github.com/ethereum/go-ethereum/blob/master/ethclient/ethclient.go#L610
var blockMethod string
var callBlockArg interface{}
// TODO: try making these error and see what happens
// 1. 66 chars but not 0x hex
// 2. 64 chars but not hex
// 3. bad blocks
// 4. bad 0x lengths
// 5. strings that aren't "latest", "safe", "finalized"
// 6. "safe" on a chain that doesn't support safe
// etc?
// I would expect this to trip within this scissor (if at all) but maybe this should get more defensive
if len(block) == 66 || len(block) == 64 {
blockMethod = "eth_getBlockByHash"
// looks like a hash which requires the object parameter
// https://eips.ethereum.org/EIPS/eip-1898
// https://docs.alchemy.com/reference/eth-call
hash := eth_common.HexToHash(block)
callBlockArg = rpc.BlockNumberOrHash{
BlockHash: &hash,
RequireCanonical: true,
}
} else {
blockMethod = "eth_getBlockByNumber"
callBlockArg = block
}
// EvmCallData contains the details of a single query in the batch.
type EvmCallData struct {
to eth_common.Address
data string
callTransactionArg map[string]interface{}
callResult *eth_hexutil.Bytes
callErr error
}
// We build two slices. The first is the batch submitted to the RPC call. It contains one entry for each query plus one to query the block.
// The second is the data associated with each request (but not the block request). The index into both is the index into the request call data.
batch := []rpc.BatchElem{}
evmCallData := []EvmCallData{}
// Add each requested query to the batch.
for _, callData := range req.CallData {
// like https://github.com/ethereum/go-ethereum/blob/master/ethclient/ethclient.go#L610
to := eth_common.BytesToAddress(callData.To)
data := eth_hexutil.Encode(callData.Data)
ecd := EvmCallData{
to: to,
data: data,
callTransactionArg: map[string]interface{}{
"to": to,
"data": data,
},
callResult: &eth_hexutil.Bytes{},
}
evmCallData = append(evmCallData, ecd)
batch = append(batch, rpc.BatchElem{
Method: "eth_call",
Args: []interface{}{
ecd.callTransactionArg,
callBlockArg,
},
Result: ecd.callResult,
Error: ecd.callErr,
})
}
// Add the block query to the batch.
var blockResult connectors.BlockMarshaller
var blockError error
batch = append(batch, rpc.BatchElem{
Method: blockMethod,
Args: []interface{}{
block,
false, // no full transaction details
},
Result: &blockResult,
Error: blockError,
})
// Query the RPC.
err := w.ethConn.RawBatchCallContext(timeout, batch)
cancel()
if err != nil {
logger.Error("failed to process query request",
zap.Error(err),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
return
}
if blockError != nil {
logger.Error("failed to process query block request",
zap.Error(blockError),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
return
}
if blockResult.Number == nil {
logger.Error("invalid query block result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
return
}
if blockResult.Number.ToInt().Cmp(ccqMaxBlockNumber) > 0 {
logger.Error("block number too large",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
return
}
resp := query.EthCallQueryResponse{
BlockNumber: blockResult.Number.ToInt().Uint64(),
Hash: blockResult.Hash,
Time: time.Unix(int64(blockResult.Time), 0),
Results: [][]byte{},
}
errFound := false
for idx := range req.CallData {
if evmCallData[idx].callErr != nil {
logger.Error("failed to process query call request",
zap.Error(evmCallData[idx].callErr),
zap.String("block", block),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
errFound = true
break
}
// Nil or Empty results are not valid
// eth_call will return empty when the state doesn't exist for a block
if len(*evmCallData[idx].callResult) == 0 {
logger.Error("invalid call result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryRetryNeeded, nil)
errFound = true
break
}
logger.Info("query result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("blockNumber", blockResult.Number.String()),
zap.String("blockHash", blockResult.Hash.Hex()),
zap.String("blockTime", blockResult.Time.String()),
zap.Int("idx", idx),
zap.String("to", evmCallData[idx].to.Hex()),
zap.Any("data", evmCallData[idx].data),
zap.String("result", evmCallData[idx].callResult.String()),
)
resp.Results = append(resp.Results, *evmCallData[idx].callResult)
}
if !errFound {
w.ccqSendQueryResponse(logger, queryRequest, query.QuerySuccess, &resp)
}
w.ccqHandleEthCallQueryRequest(logger, ctx, queryRequest, req)
case *query.EthCallByTimestampQueryRequest:
w.ccqHandleEthCallByTimestampQueryRequest(logger, ctx, queryRequest, req)
default:
logger.Warn("received unsupported request type",
zap.Uint8("payload", uint8(queryRequest.Request.Query.Type())),
)
w.ccqSendQueryResponse(logger, queryRequest, query.QueryFatalError, nil)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
}
}
// EvmCallData contains the details of a single query in the batch.
type EvmCallData struct {
to eth_common.Address
data string
callTransactionArg map[string]interface{}
callResult *eth_hexutil.Bytes
callErr error
}
func (w *Watcher) ccqHandleEthCallQueryRequest(logger *zap.Logger, ctx context.Context, queryRequest *query.PerChainQueryInternal, req *query.EthCallQueryRequest) {
block := req.BlockId
logger.Info("received eth_call query request",
zap.String("block", block),
zap.Int("numRequests", len(req.CallData)),
)
blockMethod, callBlockArg, err := ccqCreateBlockRequest(block)
if err != nil {
logger.Error("invalid block id in eth_call query request",
zap.Error(err),
zap.String("block", block),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
return
}
// We build two slices. The first is the batch submitted to the RPC call. It contains one entry for each query plus one to query the block.
// The second is the data associated with each request (but not the block request). The index into both is the index into the request call data.
batch, evmCallData := ccqBuildBatchFromCallData(req, callBlockArg)
// Add the block query to the batch.
var blockResult connectors.BlockMarshaller
var blockError error
batch = append(batch, rpc.BatchElem{
Method: blockMethod,
Args: []interface{}{
block,
false, // no full transaction details
},
Result: &blockResult,
Error: blockError,
})
// Query the RPC.
timeout, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
err = w.ethConn.RawBatchCallContext(timeout, batch)
if err != nil {
logger.Error("failed to process eth_call query request",
zap.Error(err),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if blockError != nil {
logger.Error("failed to process eth_call query block request",
zap.Error(blockError),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if blockResult.Number == nil {
logger.Error("invalid eth_call query block result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if blockResult.Number.ToInt().Cmp(w.ccqMaxBlockNumber) > 0 {
logger.Error("block number too large for eth_call",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
resp := query.EthCallQueryResponse{
BlockNumber: blockResult.Number.ToInt().Uint64(),
Hash: blockResult.Hash,
Time: time.Unix(int64(blockResult.Time), 0),
Results: [][]byte{},
}
errFound := false
for idx := range req.CallData {
if evmCallData[idx].callErr != nil {
logger.Error("failed to process eth_call query call request",
zap.Error(evmCallData[idx].callErr),
zap.String("block", block),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
errFound = true
break
}
// Nil or Empty results are not valid
// eth_call will return empty when the state doesn't exist for a block
if len(*evmCallData[idx].callResult) == 0 {
logger.Error("invalid call result for eth_call",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
errFound = true
break
}
logger.Info("query result for eth_call",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("blockNumber", blockResult.Number.String()),
zap.String("blockHash", blockResult.Hash.Hex()),
zap.String("blockTime", blockResult.Time.String()),
zap.Int("idx", idx),
zap.String("to", evmCallData[idx].to.Hex()),
zap.Any("data", evmCallData[idx].data),
zap.String("result", evmCallData[idx].callResult.String()),
)
resp.Results = append(resp.Results, *evmCallData[idx].callResult)
}
if !errFound {
w.ccqSendQueryResponseForEthCall(logger, queryRequest, query.QuerySuccess, &resp)
}
}
func (w *Watcher) ccqHandleEthCallByTimestampQueryRequest(logger *zap.Logger, ctx context.Context, queryRequest *query.PerChainQueryInternal, req *query.EthCallByTimestampQueryRequest) {
block := req.TargetBlockIdHint
nextBlock := req.FollowingBlockIdHint
logger.Info("received eth_call_by_timestamp query request",
zap.Uint64("timestamp", req.TargetTimestamp),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Int("numRequests", len(req.CallData)),
)
blockMethod, callBlockArg, err := ccqCreateBlockRequest(block)
if err != nil {
logger.Error("invalid target block id hint in eth_call_by_timestamp query request",
zap.Error(err),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
return
}
nextBlockMethod, _, err := ccqCreateBlockRequest(nextBlock)
if err != nil {
logger.Error("invalid following block id hint in eth_call_by_timestamp query request",
zap.Error(err),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
return
}
// We build two slices. The first is the batch submitted to the RPC call. It contains one entry for each query plus one to query the block and one for the next block.
// The second is the data associated with each request (but not the block requests). The index into both is the index into the request call data.
batch, evmCallData := ccqBuildBatchFromCallData(req, callBlockArg)
// Add the block query to the batch.
var blockResult connectors.BlockMarshaller
var blockError error
batch = append(batch, rpc.BatchElem{
Method: blockMethod,
Args: []interface{}{
block,
false, // no full transaction details
},
Result: &blockResult,
Error: blockError,
})
// Add the next block query to the batch.
var nextBlockResult connectors.BlockMarshaller
var nextBlockError error
batch = append(batch, rpc.BatchElem{
Method: nextBlockMethod,
Args: []interface{}{
nextBlock,
false, // no full transaction details
},
Result: &nextBlockResult,
Error: nextBlockError,
})
// Query the RPC.
timeout, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
err = w.ethConn.RawBatchCallContext(timeout, batch)
if err != nil {
logger.Error("failed to process eth_call_by_timestamp query request",
zap.Error(err),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
// Checks on the target block.
if blockError != nil {
logger.Error("failed to process eth_call_by_timestamp query target block request",
zap.Error(blockError),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if blockResult.Number == nil {
logger.Error("invalid eth_call_by_timestamp query target block result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if blockResult.Number.ToInt().Cmp(w.ccqMaxBlockNumber) > 0 {
logger.Error("target block number too large for eth_call_by_timestamp",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
// Checks on the following block.
if nextBlockError != nil {
logger.Error("failed to process eth_call_by_timestamp query following block request",
zap.Error(nextBlockError),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if nextBlockResult.Number == nil {
logger.Error("invalid eth_call_by_timestamp query following block result",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
if nextBlockResult.Number.ToInt().Cmp(w.ccqMaxBlockNumber) > 0 {
logger.Error("following block number too large for eth_call_by_timestamp",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
return
}
/*
target_block.timestamp <= target_time < following_block.timestamp
and
following_block_num - 1 == target_block_num
*/
targetBlockNum := blockResult.Number.ToInt().Uint64()
followingBlockNum := nextBlockResult.Number.ToInt().Uint64()
// The req.TargetTimestamp is in microseconds but EVM returns seconds. Convert to microseconds.
targetTimestamp := uint64(blockResult.Time * 1000000)
followingTimestamp := uint64(nextBlockResult.Time * 1000000)
if targetBlockNum+1 != followingBlockNum {
logger.Error(" eth_call_by_timestamp query blocks are not adjacent",
zap.String("eth_network", w.networkName),
zap.Uint64("desiredTimestamp", req.TargetTimestamp),
zap.Uint64("targetTimestamp", targetTimestamp),
zap.Uint64("followingTimestamp", followingTimestamp),
zap.String("targetBlockNumber", blockResult.Number.String()),
zap.String("followingBlockNumber", nextBlockResult.Number.String()),
zap.String("targetBlockHash", blockResult.Hash.Hex()),
zap.String("followingBlockHash", nextBlockResult.Hash.Hex()),
zap.String("targetBlockTime", blockResult.Time.String()),
zap.String("followingBlockTime", nextBlockResult.Time.String()),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
return
}
if req.TargetTimestamp < targetTimestamp || req.TargetTimestamp >= followingTimestamp {
logger.Error(" eth_call_by_timestamp desired timestamp falls outside of block range",
zap.String("eth_network", w.networkName),
zap.Uint64("desiredTimestamp", req.TargetTimestamp),
zap.Uint64("targetTimestamp", targetTimestamp),
zap.Uint64("followingTimestamp", followingTimestamp),
zap.String("targetBlockNumber", blockResult.Number.String()),
zap.String("followingBlockNumber", nextBlockResult.Number.String()),
zap.String("targetBlockHash", blockResult.Hash.Hex()),
zap.String("followingBlockHash", nextBlockResult.Hash.Hex()),
zap.String("targetBlockTime", blockResult.Time.String()),
zap.String("followingBlockTime", nextBlockResult.Time.String()),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryFatalError)
return
}
resp := query.EthCallByTimestampQueryResponse{
TargetBlockNumber: targetBlockNum,
TargetBlockHash: blockResult.Hash,
TargetBlockTime: time.Unix(int64(blockResult.Time), 0),
FollowingBlockNumber: followingBlockNum,
FollowingBlockHash: nextBlockResult.Hash,
FollowingBlockTime: time.Unix(int64(nextBlockResult.Time), 0),
Results: [][]byte{},
}
errFound := false
for idx := range req.CallData {
if evmCallData[idx].callErr != nil {
logger.Error("failed to process eth_call_by_timestamp query call request",
zap.Error(evmCallData[idx].callErr),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
errFound = true
break
}
// Nil or Empty results are not valid
// eth_call will return empty when the state doesn't exist for a block
if len(*evmCallData[idx].callResult) == 0 {
logger.Error("invalid call result for eth_call_by_timestamp",
zap.String("eth_network", w.networkName),
zap.String("block", block),
zap.String("nextBlock", nextBlock),
zap.Int("errorIdx", idx),
zap.Any("batch", batch),
)
w.ccqSendQueryResponseForError(logger, queryRequest, query.QueryRetryNeeded)
errFound = true
break
}
logger.Info(" eth_call_by_timestamp query result",
zap.String("eth_network", w.networkName),
zap.Uint64("desiredTimestamp", req.TargetTimestamp),
zap.Uint64("targetTimestamp", targetTimestamp),
zap.Uint64("followingTimestamp", followingTimestamp),
zap.String("targetBlockNumber", blockResult.Number.String()),
zap.String("followingBlockNumber", nextBlockResult.Number.String()),
zap.String("targetBlockHash", blockResult.Hash.Hex()),
zap.String("followingBlockHash", nextBlockResult.Hash.Hex()),
zap.String("targetBlockTime", blockResult.Time.String()),
zap.String("followingBlockTime", nextBlockResult.Time.String()),
zap.Int("idx", idx),
zap.String("to", evmCallData[idx].to.Hex()),
zap.Any("data", evmCallData[idx].data),
zap.String("result", evmCallData[idx].callResult.String()),
)
resp.Results = append(resp.Results, *evmCallData[idx].callResult)
}
if !errFound {
w.ccqSendQueryResponseForEthCallByTimestamp(logger, queryRequest, query.QuerySuccess, &resp)
}
}
// ccqCreateBlockRequest creates a block query. It parses the block string, allowing for both a block number or a block hash. Note that for now, strings like "latest", "finalized" or "safe"
// are not supported, and the block must be a hex string starting with 0x. The determination of whether it is a block number or a block hash is based on the overall length of the string,
// since a hash is 32 bytes (64 hex digits).
func ccqCreateBlockRequest(block string) (string, interface{}, error) {
// like https://github.com/ethereum/go-ethereum/blob/master/ethclient/ethclient.go#L610
var blockMethod string
var callBlockArg interface{}
if block == "" {
return blockMethod, callBlockArg, fmt.Errorf("block id is required")
}
if !strings.HasPrefix(block, "0x") {
return blockMethod, callBlockArg, fmt.Errorf("block id must start with 0x")
}
blk := strings.Trim(block, "0x")
// Devnet can give us block IDs like this: "0x365".
if len(blk)%2 != 0 {
blk = "0" + blk
}
// Make sure it is valid hex.
if _, err := hex.DecodeString(blk); err != nil {
return blockMethod, callBlockArg, fmt.Errorf("block id is not valid hex")
}
if len(blk) == 64 {
blockMethod = "eth_getBlockByHash"
// looks like a hash which requires the object parameter
// https://eips.ethereum.org/EIPS/eip-1898
// https://docs.alchemy.com/reference/eth-call
hash := eth_common.HexToHash(block)
callBlockArg = rpc.BlockNumberOrHash{
BlockHash: &hash,
RequireCanonical: true,
}
} else {
blockMethod = "eth_getBlockByNumber"
callBlockArg = block
}
return blockMethod, callBlockArg, nil
}
type EthCallDataIntf interface {
CallDataList() []*query.EthCallData
}
func ccqBuildBatchFromCallData(req EthCallDataIntf, callBlockArg interface{}) ([]rpc.BatchElem, []EvmCallData) {
batch := []rpc.BatchElem{}
evmCallData := []EvmCallData{}
// Add each requested query to the batch.
for _, callData := range req.CallDataList() {
// like https://github.com/ethereum/go-ethereum/blob/master/ethclient/ethclient.go#L610
to := eth_common.BytesToAddress(callData.To)
data := eth_hexutil.Encode(callData.Data)
ecd := EvmCallData{
to: to,
data: data,
callTransactionArg: map[string]interface{}{
"to": to,
"data": data,
},
callResult: &eth_hexutil.Bytes{},
}
evmCallData = append(evmCallData, ecd)
batch = append(batch, rpc.BatchElem{
Method: "eth_call",
Args: []interface{}{
ecd.callTransactionArg,
callBlockArg,
},
Result: ecd.callResult,
Error: ecd.callErr,
})
}
return batch, evmCallData
}

3
sdk/js-query/ci-config.js Normal file
View File

@ -0,0 +1,3 @@
process.env.CI = true;
module.exports = {};

1
sdk/js-query/package.json
View File

@ -11,6 +11,7 @@
],
"scripts": {
"test": "jest --verbose",
"test-ci": "jest --verbose --setupFiles ./ci-config.js --forceExit",
"build": "tsc -p tsconfig.json && tsc -p tsconfig.cjs.json && tsc -p tsconfig.types.json"
},
"keywords": [

8
sdk/js-query/src/query/BinaryWriter.ts
View File

@ -43,6 +43,14 @@ export class BinaryWriter {
return this;
}
writeUint64(value: bigint) {
if (value < 0 || value > 18446744073709551615)
throw new Error("Invalid value");
this._ensure(8);
this._offset = this._buffer.writeBigUInt64BE(value, this._offset);
return this;
}
writeUint8Array(value: Uint8Array) {
this._ensure(value.length);
this._buffer.set(value, this._offset);

247
sdk/js-query/src/query/ethCall.test.ts
View File

@ -11,6 +11,7 @@ import axios from "axios";
import {
EthCallData,
EthCallQueryRequest,
EthCallByTimestampQueryRequest,
PerChainQueryRequest,
QueryRequest,
sign,
@ -18,11 +19,13 @@ import {
jest.setTimeout(125000);
const CI = false;
const CI = process.env.CI;
const ENV = "DEVNET";
const ETH_NODE_URL = CI ? "ws://eth-devnet:8545" : "ws://localhost:8545";
const CCQ_SERVER_URL = "http://localhost:6069/v1";
const CCQ_SERVER_URL = CI
? "http://query-server:6069/v1"
: "http://localhost:6069/v1";
const QUERY_URL = CCQ_SERVER_URL + "/query";
const HEALTH_URL = CCQ_SERVER_URL + "/health";
const PRIVATE_KEY =
@ -61,6 +64,27 @@ function createTestEthCallData(
};
}
async function getEthCallByTimestampArgs(): Promise<[bigint, bigint, bigint]> {
let followingBlockNumber = BigInt(
await web3.eth.getBlockNumber(ETH_DATA_FORMAT)
);
let targetBlockNumber = BigInt(0);
let targetBlockTime = BigInt(0);
while (targetBlockNumber === BigInt(0)) {
const followingBlock = await web3.eth.getBlock(followingBlockNumber);
const targetBlock = await web3.eth.getBlock(
(Number(followingBlockNumber) - 1).toString()
);
if (targetBlock.timestamp < followingBlock.timestamp) {
targetBlockTime = targetBlock.timestamp * BigInt(1000000);
targetBlockNumber = targetBlock.number;
} else {
followingBlockNumber = targetBlockNumber;
}
}
return [targetBlockTime, targetBlockNumber, followingBlockNumber];
}
describe("eth call", () => {
test("serialize request", () => {
const toAddress = "0x0d500b1d8e8ef31e21c99d1db9a6444d3adf1270";
@ -112,6 +136,39 @@ describe("eth call", () => {
);
expect(response.status).toBe(200);
});
// TODO: This test works in Goerli testnet but not devnet. Try it again after PR #3395 lands.
test.skip("get block by hash should work", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
WETH_ADDRESS,
"totalSupply",
"uint256"
);
const blockNumber = await web3.eth.getBlockNumber(ETH_DATA_FORMAT);
const block = await web3.eth.getBlock(BigInt(blockNumber));
if (block.hash != undefined) {
const ethCall = new EthCallQueryRequest(block.hash?.toString(), [
nameCallData,
totalSupplyCallData,
]);
const chainId = 2;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
const digest = QueryRequest.digest(ENV, serialized);
const signature = sign(PRIVATE_KEY, digest);
const response = await axios.put(
QUERY_URL,
{
signature,
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "my_secret_key" } }
);
expect(response.status).toBe(200);
}
});
test("missing api-key should fail", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
@ -277,7 +334,7 @@ describe("eth call", () => {
const response = await axios.get(HEALTH_URL);
expect(response.status).toBe(200);
});
test("valid api key but payload too large should fail based on size", async () => {
test("payload too large should fail", async () => {
const serialized = new Uint8Array(6000000); // Buffer should be larger than MAX_BODY_SIZE in node/cmd/ccq/http.go.
const signature = "";
let err = false;
@ -286,7 +343,6 @@ describe("eth call", () => {
QUERY_URL,
{
signature,
// bytes: Buffer.alloc(6000000).toString("hex"),
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "my_secret_key" } }
@ -298,24 +354,191 @@ describe("eth call", () => {
});
expect(err).toBe(true);
});
test("invalid api key with payload too large should fail based on api key", async () => {
const serialized = new Uint8Array(6000000); // Buffer should be larger than MAX_BODY_SIZE in node/cmd/ccq/http.go.
const signature = "";
test("serialize eth_call_by_timestamp request", () => {
const toAddress = "0x0d500b1d8e8ef31e21c99d1db9a6444d3adf1270";
const nameCallData = createTestEthCallData(toAddress, "name", "string");
const totalSupplyCallData = createTestEthCallData(
toAddress,
"totalSupply",
"uint256"
);
const ethCall = new EthCallByTimestampQueryRequest(
BigInt(1697216322000000),
"0x28d9630",
"0x28d9631",
[nameCallData, totalSupplyCallData]
);
const chainId = 5;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
expect(Buffer.from(serialized).toString("hex")).toEqual(
"0100000001010005020000005b0006079bf7fad4800000000930783238643936333000000009307832386439363331020d500b1d8e8ef31e21c99d1db9a6444d3adf12700000000406fdde030d500b1d8e8ef31e21c99d1db9a6444d3adf12700000000418160ddd"
);
});
test("successful eth_call_by_timestamp query", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
WETH_ADDRESS,
"totalSupply",
"uint256"
);
const [targetBlockTime, targetBlockNumber, followingBlockNumber] =
await getEthCallByTimestampArgs();
const ethCall = new EthCallByTimestampQueryRequest(
targetBlockTime,
targetBlockNumber.toString(16),
followingBlockNumber.toString(16),
[nameCallData, totalSupplyCallData]
);
const chainId = 2;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
const digest = QueryRequest.digest(ENV, serialized);
const signature = sign(PRIVATE_KEY, digest);
const response = await axios.put(
QUERY_URL,
{
signature,
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "my_secret_key" } }
);
expect(response.status).toBe(200);
});
test("eth_call_by_timestamp query without target timestamp", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
WETH_ADDRESS,
"totalSupply",
"uint256"
);
const followingBlockNum = await web3.eth.getBlockNumber(ETH_DATA_FORMAT);
const followingBlock = await web3.eth.getBlock(BigInt(followingBlockNum));
const targetBlock = await web3.eth.getBlock(
(Number(followingBlockNum) - 1).toString()
);
const ethCall = new EthCallByTimestampQueryRequest(
BigInt(0),
targetBlock.number.toString(16),
followingBlock.number.toString(16),
[nameCallData, totalSupplyCallData]
);
const chainId = 2;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
const digest = QueryRequest.digest(ENV, serialized);
const signature = sign(PRIVATE_KEY, digest);
let err = false;
await axios
const response = await axios
.put(
QUERY_URL,
{
signature,
// bytes: Buffer.alloc(6000000).toString("hex"),
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "some_junk" } }
{ headers: { "X-API-Key": "my_secret_key" } }
)
.catch(function (error) {
err = true;
expect(error.response.status).toBe(403);
expect(error.response.data).toBe(`invalid api key\n`);
expect(error.response.status).toBe(400);
expect(error.response.data).toBe(
`failed to validate request: failed to validate per chain query 0: chain specific query is invalid: target timestamp may not be zero\n`
);
});
expect(err).toBe(true);
});
test("eth_call_by_timestamp query without target hint should fail for now", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
WETH_ADDRESS,
"totalSupply",
"uint256"
);
const followingBlockNum = await web3.eth.getBlockNumber(ETH_DATA_FORMAT);
const followingBlock = await web3.eth.getBlock(BigInt(followingBlockNum));
const targetBlock = await web3.eth.getBlock(
(Number(followingBlockNum) - 1).toString()
);
const targetBlockTime = targetBlock.timestamp * BigInt(1000000);
const ethCall = new EthCallByTimestampQueryRequest(
targetBlockTime,
"",
followingBlock.number.toString(16),
[nameCallData, totalSupplyCallData]
);
const chainId = 2;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
const digest = QueryRequest.digest(ENV, serialized);
const signature = sign(PRIVATE_KEY, digest);
let err = false;
const response = await axios
.put(
QUERY_URL,
{
signature,
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "my_secret_key" } }
)
.catch(function (error) {
err = true;
expect(error.response.status).toBe(400);
expect(error.response.data).toBe(
`failed to validate request: failed to validate per chain query 0: chain specific query is invalid: target block id must be a hex number or hash starting with 0x\n`
);
});
expect(err).toBe(true);
});
test("eth_call_by_timestamp query without following hint should fail for now", async () => {
const nameCallData = createTestEthCallData(WETH_ADDRESS, "name", "string");
const totalSupplyCallData = createTestEthCallData(
WETH_ADDRESS,
"totalSupply",
"uint256"
);
const followingBlockNum = await web3.eth.getBlockNumber(ETH_DATA_FORMAT);
const targetBlock = await web3.eth.getBlock(
(Number(followingBlockNum) - 1).toString()
);
const targetBlockTime = targetBlock.timestamp * BigInt(1000000);
const ethCall = new EthCallByTimestampQueryRequest(
targetBlockTime,
targetBlock.number.toString(16),
"",
[nameCallData, totalSupplyCallData]
);
const chainId = 2;
const ethQuery = new PerChainQueryRequest(chainId, ethCall);
const nonce = 1;
const request = new QueryRequest(nonce, [ethQuery]);
const serialized = request.serialize();
const digest = QueryRequest.digest(ENV, serialized);
const signature = sign(PRIVATE_KEY, digest);
let err = false;
const response = await axios
.put(
QUERY_URL,
{
signature,
bytes: Buffer.from(serialized).toString("hex"),
},
{ headers: { "X-API-Key": "my_secret_key" } }
)
.catch(function (error) {
err = true;
expect(error.response.status).toBe(400);
expect(error.response.data).toBe(
`failed to validate request: failed to validate per chain query 0: chain specific query is invalid: following block id must be a hex number or hash starting with 0x\n`
);
});
expect(err).toBe(true);
});

61
sdk/js-query/src/query/ethCallByTimestamp.ts Normal file
View File

@ -0,0 +1,61 @@
import { BinaryWriter } from "./BinaryWriter";
import { BlockTag, EthCallData } from "./ethCall";
import { ChainQueryType, ChainSpecificQuery } from "./request";
import { hexToUint8Array, isValidHexString } from "./utils";
export class EthCallByTimestampQueryRequest implements ChainSpecificQuery {
targetTimestamp: bigint;
targetBlockHint: string;
followingBlockHint: string;
constructor(
targetTimestamp: bigint,
targetBlockHint: BlockTag,
followingBlockHint: BlockTag,
public callData: EthCallData[]
) {
this.targetTimestamp = targetTimestamp;
this.targetBlockHint = parseBlockHint(targetBlockHint);
this.followingBlockHint = parseBlockHint(followingBlockHint);
}
type(): ChainQueryType {
return ChainQueryType.EthCallByTimeStamp;
}
serialize(): Uint8Array {
const writer = new BinaryWriter()
.writeUint64(this.targetTimestamp)
.writeUint32(this.targetBlockHint.length)
.writeUint8Array(Buffer.from(this.targetBlockHint))
.writeUint32(this.followingBlockHint.length)
.writeUint8Array(Buffer.from(this.followingBlockHint))
.writeUint8(this.callData.length);
this.callData.forEach(({ to, data }) => {
const dataArray = hexToUint8Array(data);
writer
.writeUint8Array(hexToUint8Array(to))
.writeUint32(dataArray.length)
.writeUint8Array(dataArray);
});
return writer.data();
}
}
function parseBlockHint(blockHint: BlockTag): string {
// Block hints are not required.
if (blockHint != "") {
if (typeof blockHint === "number") {
blockHint = `0x${blockHint.toString(16)}`;
} else if (isValidHexString(blockHint)) {
if (!blockHint.startsWith("0x")) {
blockHint = `0x${blockHint}`;
}
blockHint = blockHint;
} else {
throw new Error(`Invalid block tag: ${blockHint}`);
}
}
return blockHint;
}

1
sdk/js-query/src/query/index.ts
View File

@ -1,4 +1,5 @@
export * from "./request";
export * from "./utils";
export * from "./ethCall";
export * from "./ethCallByTimestamp";
export * from "./consts";

1
sdk/js-query/src/query/request.ts
View File

@ -67,4 +67,5 @@ export interface ChainSpecificQuery {
export enum ChainQueryType {
EthCall = 1,
EthCallByTimeStamp = 2,
}

17
testing/Dockerfile.querysdk.test Normal file
View File

@ -0,0 +1,17 @@
FROM node:19.6.1-slim@sha256:a1ba21bf0c92931d02a8416f0a54daad66cb36a85d2b73af9d73b044f5f57cfc
RUN apt-get update && apt-get -y install \
git python3 make curl netcat
RUN mkdir -p /app
WORKDIR /app
COPY sdk/js-query/package.json sdk/js-query/package-lock.json ./sdk/js-query/
RUN --mount=type=cache,uid=1000,gid=1000,target=/home/node/.npm \
npm ci --prefix sdk/js-query
COPY sdk/js-query ./sdk/js-query
RUN npm run build --prefix sdk/js-query
COPY testing ./testing
WORKDIR /app/testing

8
testing/querysdk.sh Normal file
View File

@ -0,0 +1,8 @@
#!/bin/sh
set -e
num=${NUM_GUARDIANS:-1} # default value for NUM_GUARDIANS = 1
for ((i=0; i<num; i++)); do
while [[ "$(curl -s -o /dev/null -w ''%{http_code}'' guardian-$i.guardian:6060/readyz)" != "200" ]]; do sleep 5; done
done
while [[ "$(curl -s -o /dev/null -w ''%{http_code}'' query-server:6069/v1/health)" != "200" ]]; do sleep 5; done
CI=true npm --prefix ../sdk/js-query run test-ci

4
whitepapers/0013_ccq.md
View File

@ -246,7 +246,9 @@ Currently the only supported query types are `eth_call` and `eth_call_by_timesta
This query type is similar to `eth_call` but targets a timestamp instead of a specific block_id. This can be useful when forming requests based on uncorrelated data, such as requiring data from another chain based on the block timestamp of a given chain.
The request MUST include the target timestamp. The request MAY include block hints for the target block and following block. These fields are reserved for the possibility that guardians can optionally perform the timestamp -> block lookup. The resulting block numbers MUST be `1` different and their timestamps MUST be such that the target block is _before_ the target time (inclusive) and the following block is _after_ (exclusive). In other words,
The request MUST include the target timestamp.
As of October 2023, the request MUST also include the block hints for the target block and following block. (A future release may make these hints optional.) These hints are used by the guardians to look up the blocks. The resulting block numbers MUST be `1` different and their timestamps MUST be such that the target block is _before_ the target time (inclusive) and the following block is _after_ (exclusive). In other words,
```
target_block.timestamp <= target_time < following_block.timestamp