revert to a working state with concurrency

load-testing/loadtest_manifest_localnet_single_supplier.yaml

@@ -44,7 +44,7 @@ gateways:
 
   # Gateway 1; http://localhost:10350/r/gateway1/overview
   - address: pokt15vzxjqklzjtlz7lahe8z2dfe9nm5vxwwmscne4
-    exposed_url: http://localhost:3069/v1/ # The gateway url that the user sends relays to (e.g. curl)
+    exposed_url:  https://devnet-red0ne-path-1.poktroll.com/v1/ # The gateway url that the user sends relays to (e.g. curl)
 
   ## Gateway 2; http://localhost:10350/r/gateway2/overview
   #- address: pokt15w3fhfyc0lttv7r585e2ncpf6t2kl9uh8rsnyz

makefiles/tests.mk

@@ -59,7 +59,7 @@ test_load_relays_stress_localnet: test_e2e_env warn_message_local_stress_test ##
 .PHONY: test_load_relays_stress_localnet_single_supplier
 test_load_relays_stress_localnet_single_supplier: test_e2e_env warn_message_local_stress_test ## Run the stress test for E2E relays on LocalNet using exclusively one supplier.
 	go test -v -count=1 ./load-testing/tests/... \
-	-tags=load,test -run TestSingleSupplierLoadRelays --log-level=debug --timeout=30m \
+	-tags=load,test -run TestSingleSupplierLoadRelays --log-level=debug --timeout=50m \
 	--manifest ./load-testing/loadtest_manifest_localnet_single_supplier.yaml
 
 .PHONY: test_verbose

pkg/client/query/appquerier.go

@@ -24,7 +24,7 @@ type appQuerier struct {
 	blockClient    client.BlockClient
 	appCache       map[string]*apptypes.Application
 	appParamsCache *apptypes.Params
-	appCacheMu     sync.Mutex
+	appCacheMu     *sync.Mutex
 }
 
 // NewApplicationQuerier returns a new instance of a client.ApplicationQueryClient
@@ -33,7 +33,7 @@ type appQuerier struct {
 // Required dependencies:
 // - clientCtx
 func NewApplicationQuerier(ctx context.Context, deps depinject.Config) (client.ApplicationQueryClient, error) {
-	aq := &appQuerier{}
+	aq := &appQuerier{appCacheMu: &sync.Mutex{}}
 
 	if err := depinject.Inject(
 		deps,

pkg/crypto/protocol/proof_path.go

@@ -2,29 +2,27 @@ package protocol
 
 import (
 	"crypto/sha256"
-
-	"github.com/pokt-network/smt"
 )
 
 // SMT specification used for the proof verification.
 var (
-	newHasher = sha256.New
-	SmtSpec   smt.TrieSpec
+	NewHasher = sha256.New
+	//SmtSpec   smt.TrieSpec
 )
 
 func init() {
 	// Use a spec that does not prehash values in the smst. This returns a nil value
 	// hasher for the proof verification in order to avoid hashing the value twice.
-	SmtSpec = smt.NewTrieSpec(
-		newHasher(), true,
-		smt.WithValueHasher(nil),
-	)
+	//SmtSpec = smt.NewTrieSpec(
+	//	newHasher(), true,
+	//	smt.WithValueHasher(nil),
+	//)
 }
 
 // GetPathForProof computes the path to be used for proof validation by hashing
 // the block hash and session id.
 func GetPathForProof(blockHash []byte, sessionId string) []byte {
-	hasher := newHasher()
+	hasher := NewHasher()
 	if _, err := hasher.Write(append(blockHash, []byte(sessionId)...)); err != nil {
 		panic(err)
 	}

pkg/crypto/rings/client.go

@@ -21,9 +21,7 @@ import (
 
 var _ crypto.RingClient = (*ringClient)(nil)
 
-// ringClient is an implementation of the RingClient interface that uses the
-// client.ApplicationQueryClient to get application's delegation information
-// needed to construct the ring for signing relay requests.
+// ringClient implements the RingClient interface.
 type ringClient struct {
 	logger polylog.Logger
 

proto/poktroll/proof/event.proto

@@ -44,10 +44,13 @@ message EventProofUpdated {
   cosmos.base.v1beta1.Coin claimed_upokt = 6 [(gogoproto.jsontag) = "claimed_upokt"];
 }
 
-// Event emitted after a proof has been checked for validity.
+// Event emitted after a proof has been checked for validity in the proof module's
+// EndBlocker.
 message EventProofValidityChecked {
   poktroll.proof.Proof proof = 1 [(gogoproto.jsontag) = "proof"];
   uint64 block_height = 2 [(gogoproto.jsontag) = "block_height"];
   poktroll.proof.ClaimProofStatus proof_status = 3 [(gogoproto.jsontag) = "proof_status"];
+  // reason is the string representation of the error that led to the proof being
+  // marked as invalid (e.g. "invalid closest merkle proof", "invalid relay request signature")
   string reason = 4 [(gogoproto.jsontag) = "reason"];
 }

x/proof/keeper/proof_validation.go

@@ -105,15 +105,20 @@ func (k Keeper) EnsureWellFormedProof(ctx context.Context, proof *types.Proof) e
 		return types.ErrProofInvalidProof.Wrapf("failed to unmarshal sparse compact merkle closest proof: %s", err)
 	}
 
+	smtSpec := smt.NewTrieSpec(
+		protocol.NewHasher(), true,
+		smt.WithValueHasher(nil),
+	)
+
 	// SparseCompactMerkeClosestProof does not implement GetValueHash, so we need to decompact it.
-	sparseMerkleClosestProof, err := smt.DecompactClosestProof(sparseCompactMerkleClosestProof, &protocol.SmtSpec)
+	sparseMerkleClosestProof, err := smt.DecompactClosestProof(sparseCompactMerkleClosestProof, &smtSpec)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to decompact sparse merkle closest proof due to error: %v", err))
 		return types.ErrProofInvalidProof.Wrapf("failed to decompact sparse erkle closest proof: %s", err)
 	}
 
 	// Get the relay request and response from the proof.GetClosestMerkleProof.
-	relayBz := sparseMerkleClosestProof.GetValueHash(&protocol.SmtSpec)
+	relayBz := sparseMerkleClosestProof.GetValueHash(&smtSpec)
 	relay := &servicetypes.Relay{}
 	if err = k.cdc.Unmarshal(relayBz, relay); err != nil {
 		logger.Error(fmt.Sprintf("failed to unmarshal relay due to error: %v", err))
@@ -229,16 +234,21 @@ func (k Keeper) EnsureValidProofSignaturesAndClosestPath(
 		return types.ErrProofInvalidProof.Wrapf("failed to unmarshal sparse compact merkle closest proof: %s", err)
 	}
 
+	smtSpec := smt.NewTrieSpec(
+		protocol.NewHasher(), true,
+		smt.WithValueHasher(nil),
+	)
+
 	// SparseCompactMerkeClosestProof was intentionally compacted to reduce its onchain state size
 	// so it must be decompacted rather than just retrieving the value via GetValueHash (not implemented).
-	sparseMerkleClosestProof, err := smt.DecompactClosestProof(sparseCompactMerkleClosestProof, &protocol.SmtSpec)
+	sparseMerkleClosestProof, err := smt.DecompactClosestProof(sparseCompactMerkleClosestProof, &smtSpec)
 	if err != nil {
 		logger.Error(fmt.Sprintf("failed to decompact sparse merkle closest proof due to error: %v", err))
 		return types.ErrProofInvalidProof.Wrapf("failed to decompact sparse merkle closest proof: %s", err)
 	}
 
 	// Get the relay request and response from the proof.GetClosestMerkleProof.
-	relayBz := sparseMerkleClosestProof.GetValueHash(&protocol.SmtSpec)
+	relayBz := sparseMerkleClosestProof.GetValueHash(&smtSpec)
 	relay := &servicetypes.Relay{}
 	if err = k.cdc.Unmarshal(relayBz, relay); err != nil {
 		logger.Error(fmt.Sprintf("failed to unmarshal relay due to error: %v", err))
@@ -451,7 +461,11 @@ func verifyClosestProof(
 	proof *smt.SparseMerkleClosestProof,
 	claimRootHash []byte,
 ) error {
-	valid, err := smt.VerifyClosestProof(proof, claimRootHash, &protocol.SmtSpec)
+	smtSpec := smt.NewTrieSpec(
+		protocol.NewHasher(), true,
+		smt.WithValueHasher(nil),
+	)
+	valid, err := smt.VerifyClosestProof(proof, claimRootHash, &smtSpec)
 	if err != nil {
 		return err
 	}

x/proof/types/account_query_client.go

@@ -3,12 +3,12 @@ package types
 import (
 	"context"
 	fmt "fmt"
-	"sync"
 
 	cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types"
 	"github.com/cosmos/cosmos-sdk/types"
 
 	"github.com/pokt-network/poktroll/pkg/client"
+	sharedtypes "github.com/pokt-network/poktroll/x/shared/types"
 )
 
 var _ client.AccountQueryClient = (*AccountKeeperQueryClient)(nil)
@@ -18,8 +18,7 @@ var _ client.AccountQueryClient = (*AccountKeeperQueryClient)(nil)
 // network requests as in the offchain implementation.
 type AccountKeeperQueryClient struct {
 	keeper             AccountKeeper
-	accountPubKeyCache map[string]cryptotypes.PubKey
-	CacheMu            *sync.RWMutex
+	accountPubKeyCache *sharedtypes.Cache[string, cryptotypes.PubKey]
 }
 
 // NewAccountKeeperQueryClient returns a new AccountQueryClient that is backed
@@ -30,8 +29,7 @@ type AccountKeeperQueryClient struct {
 func NewAccountKeeperQueryClient(accountKeeper AccountKeeper) client.AccountQueryClient {
 	return &AccountKeeperQueryClient{
 		keeper:             accountKeeper,
-		accountPubKeyCache: make(map[string]cryptotypes.PubKey),
-		CacheMu:            &sync.RWMutex{},
+		accountPubKeyCache: sharedtypes.NewCache[string, cryptotypes.PubKey](),
 	}
 }
 
@@ -66,11 +64,9 @@ func (accountQueryClient *AccountKeeperQueryClient) GetPubKeyFromAddress(
 	ctx context.Context,
 	address string,
 ) (cryptotypes.PubKey, error) {
-	accountQueryClient.CacheMu.RLock()
-	defer accountQueryClient.CacheMu.RUnlock()
-	if acc, found := accountQueryClient.accountPubKeyCache[address]; found {
+	if pubkey, found := accountQueryClient.accountPubKeyCache.Get(address); found {
 		fmt.Println("-----PubKey cache hit-----")
-		return acc, nil
+		return pubkey, nil
 	}
 
 	acc, err := accountQueryClient.GetAccount(ctx, address)
@@ -87,13 +83,11 @@ func (accountQueryClient *AccountKeeperQueryClient) GetPubKeyFromAddress(
 		return nil, ErrProofPubKeyNotFound
 	}
 
-	accountQueryClient.accountPubKeyCache[address] = pubKey
+	accountQueryClient.accountPubKeyCache.Set(address, pubKey)
 
 	return pubKey, nil
 }
 
 func (accountQueryClient *AccountKeeperQueryClient) ClearCache() {
-	accountQueryClient.CacheMu.Lock()
-	defer accountQueryClient.CacheMu.Unlock()
-	clear(accountQueryClient.accountPubKeyCache)
+	accountQueryClient.accountPubKeyCache.Clear()
 }