Merge pull request #87 from stafiprotocol/main

feat: add query IDs filter

.env.example

@@ -20,6 +20,7 @@ RELAYER_TARGET_CHAIN_DEBUG=true
 RELAYER_TARGET_CHAIN_OUTPUT_FORMAT=json
 
 RELAYER_REGISTRY_ADDRESSES=neutron14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9s5c2epq
+RELAYER_REGISTRY_QUERY_IDS=
 
 RELAYER_ALLOW_TX_QUERIES=true
 RELAYER_ALLOW_KV_CALLBACKS=true

.env.example.dev

@@ -34,6 +34,7 @@ RELAYER_TARGET_CHAIN_OUTPUT_FORMAT=json
 RELAYER_TARGET_CHAIN_SIGN_MODE_STR=direct
 
 RELAYER_REGISTRY_ADDRESSES=
+RELAYER_REGISTRY_QUERY_IDS=
 
 RELAYER_ALLOW_TX_QUERIES=true
 RELAYER_ALLOW_KV_CALLBACKS=true

README.md

@@ -98,6 +98,7 @@ Relayer:
 | `RELAYER_TARGET_CHAIN_DEBUG `                    | `bool`            | flag to run target chain provider in debug mode                                                                                                                            | optional |
 | `RELAYER_TARGET_CHAIN_OUTPUT_FORMAT`             | `json`  or `yaml` | target chain provider output format                                                                                                                                        | optional |
 | `RELAYER_REGISTRY_ADDRESSES`                     | `string`          | a list of comma-separated smart-contract addresses for which the relayer processes interchain queries                                                                      | required |
+| `RELAYER_REGISTRY_QUERY_IDS`                     | `string`          | a list of comma-separated query IDs which complements to `RELAYER_REGISTRY_ADDRESSES` to further filter out interchain queries being processed                                                                     | optional |
 | `RELAYER_ALLOW_TX_QUERIES`                       | `bool`            | if true relayer will process tx queries  (if `false`, relayer will drop them)                                                                                              | required |
 | `RELAYER_ALLOW_KV_CALLBACKS`                     | `bool`            | if `true`, will pass proofs as sudo callbacks to contracts                                                                                                                 | required |
 | `RELAYER_MIN_KV_UPDATE_PERIOD`                   | `uint`            | minimal period of queries execution and submission (not less than `n` blocks)                                                                                              | optional |

internal/registry/registry.go

@@ -3,36 +3,53 @@ package registry
 // RegistryConfig represents the config structure for the Registry.
 type RegistryConfig struct {
 	Addresses []string
+	QueryIDs  []uint64 `envconfig:"QUERY_IDS"`
 }
 
 // New instantiates a new *Registry based on the cfg.
 func New(cfg *RegistryConfig) *Registry {
 	r := &Registry{
 		addresses: make(map[string]struct{}, len(cfg.Addresses)),
+		queryIDs:  make(map[uint64]struct{}, len(cfg.QueryIDs)),
 	}
 	for _, addr := range cfg.Addresses {
 		r.addresses[addr] = struct{}{}
 	}
+	for _, queryID := range cfg.QueryIDs {
+		r.queryIDs[queryID] = struct{}{}
+	}
 	return r
 }
 
-// Registry is the relayer's watch list registry. It contains a list of addresses, and the relayer
-// only works with interchain queries that are under these addresses' ownership.
+// Registry is the relayer's watch list registry. It contains a list of addresses and a list of queryIDs,
+// and the relayer only works with interchain queries that are under these addresses' ownership and match the queryIDs.
 type Registry struct {
 	addresses map[string]struct{}
+	queryIDs  map[uint64]struct{}
 }
 
-// IsEmpty returns true if the registry addresses list is empty.
-func (r *Registry) IsEmpty() bool {
+// IsAddressesEmpty returns true if the registry addresses list is empty.
+func (r *Registry) IsAddressesEmpty() bool {
 	return len(r.addresses) == 0
 }
 
-// Contains returns true if the addr is in the registry.
-func (r *Registry) Contains(addr string) bool {
+// IsQueryIDsEmpty returns true if the registry queryIDs list is empty.
+func (r *Registry) IsQueryIDsEmpty() bool {
+	return len(r.queryIDs) == 0
+}
+
+// ContainsAddress returns true if the addr is in the registry.
+func (r *Registry) ContainsAddress(addr string) bool {
 	_, ex := r.addresses[addr]
 	return ex
 }
 
+// ContainsQueryID returns true if the queryID is in the registry.
+func (r *Registry) ContainsQueryID(queryID uint64) bool {
+	_, ex := r.queryIDs[queryID]
+	return ex
+}
+
 func (r *Registry) GetAddresses() []string {
 	var out []string
 	for addr := range r.addresses {

internal/registry/registry_test.go

@@ -0,0 +1,59 @@
+package registry_test
+
+import (
+	"testing"
+
+	"github.com/neutron-org/neutron-query-relayer/internal/registry"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRegistryWithEmptyAddressesAndEmptyQueryIDs(t *testing.T) {
+	cfg := registry.RegistryConfig{}
+	r := registry.New(&cfg)
+	assert.True(t, r.IsAddressesEmpty())
+	assert.True(t, r.IsQueryIDsEmpty())
+
+	assert.False(t, r.ContainsAddress("not_exist_address"))
+	assert.False(t, r.ContainsQueryID(0))
+	assert.False(t, r.ContainsQueryID(1))
+	assert.Equal(t, []string(nil), r.GetAddresses())
+}
+
+func TestRegistryWithAddressesAndEmptyQueryIDs(t *testing.T) {
+	cfg := registry.RegistryConfig{
+		Addresses: []string{"exist_address", "exist_address2"},
+	}
+	r := registry.New(&cfg)
+	assert.False(t, r.IsAddressesEmpty())
+	assert.True(t, r.ContainsAddress("exist_address"))
+	assert.False(t, r.ContainsAddress("not_exist_address"))
+	assert.ElementsMatch(t, []string{"exist_address", "exist_address2"}, r.GetAddresses())
+}
+
+func TestRegistryWithAddressesAndQueryIDs(t *testing.T) {
+	cfg := registry.RegistryConfig{
+		Addresses: []string{"exist_address", "exist_address2"},
+		QueryIDs:  []uint64{0, 1},
+	}
+	r := registry.New(&cfg)
+	assert.False(t, r.IsAddressesEmpty())
+	assert.False(t, r.IsQueryIDsEmpty())
+	assert.True(t, r.ContainsAddress("exist_address"))
+	assert.False(t, r.ContainsAddress("not_exist_address"))
+	assert.True(t, r.ContainsQueryID(0))
+	assert.True(t, r.ContainsQueryID(1))
+	assert.False(t, r.ContainsQueryID(2))
+}
+
+func TestRegistryWithEmptyAddressesAndQueryIDs(t *testing.T) {
+	cfg := registry.RegistryConfig{
+		QueryIDs: []uint64{0, 1},
+	}
+	r := registry.New(&cfg)
+	assert.True(t, r.IsAddressesEmpty())
+	assert.False(t, r.IsQueryIDsEmpty())
+	assert.False(t, r.ContainsAddress("not_exist_address"))
+	assert.True(t, r.ContainsQueryID(0))
+	assert.True(t, r.ContainsQueryID(1))
+	assert.False(t, r.ContainsQueryID(2))
+}

internal/subscriber/subscriber.go

@@ -3,12 +3,14 @@ package subscriber
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"sync"
+	"time"
+
 	nlogger "github.com/neutron-org/neutron-logger"
 	"github.com/neutron-org/neutron-query-relayer/internal/app"
 	"github.com/neutron-org/neutron-query-relayer/internal/config"
 	"github.com/neutron-org/neutron-query-relayer/internal/relay"
-	"sync"
-	"time"
 
 	"github.com/neutron-org/neutron-query-relayer/internal/registry"
 
@@ -31,8 +33,8 @@ type Config struct {
 	ConnectionID string
 	// WatchedTypes is the list of query types to be observed and handled.
 	WatchedTypes []neutrontypes.InterchainQueryType
-	// Registry is a watch list registry. It contains a list of addresses, and the Subscriber only
-	// works with interchain queries and events that are under these addresses' ownership.
+	// Registry is a watch list registry. It contains a list of addresses and a list of queryIDs, and the Subscriber only
+	// works with interchain queries and events that are under ownership of these addresses and match the queryIDs.
 	Registry *rg.Registry
 }
 
@@ -216,6 +218,16 @@ func (s *Subscriber) processUpdateEvent(ctx context.Context, event tmtypes.Resul
 				zap.String("query_id", queryID))
 			continue
 		}
+		queryIDNumber, err := strconv.ParseUint(queryID, 10, 64)
+		if err != nil {
+			return fmt.Errorf("failed to parse queryID: %w", err)
+		}
+
+		if !s.isWatchedQueryID(queryIDNumber) {
+			s.logger.Debug("Skipping query (wrong queryID)", zap.String("owner", owner),
+				zap.String("query_id", queryID))
+			continue
+		}
 
 		// Load all information about the neutronQuery directly from Neutron.
 		neutronQuery, err := s.getNeutronRegisteredQuery(ctx, queryID)