[RelayMiner] Supplier rate limiting

config.yml

@@ -141,13 +141,13 @@ genesis:
             denom: upokt
     bank:
       supply:
-        - amount: "1102000204"
+        - amount: "1202000272"
           denom: upokt
       balances:
         # Application module
         - address: pokt1rl3gjgzexmplmds3tq3r3yk84zlwdl6djzgsvm
           coins:
-            - amount: "100000068" # MUST BE equal to the total of all app stakes below
+            - amount: "200000136" # MUST BE equal to the total of all app stakes below
               denom: upokt
         # Supplier module
         - address: pokt1j40dzzmn6cn9kxku7a5tjnud6hv37vesr5ccaa

pkg/relayer/cmd/cmd.go

@@ -198,10 +198,11 @@ func setupRelayerDependencies(
 		config.NewSupplyDelegationClientFn(),                              // leaf
 		config.NewSupplySharedQueryClientFn(),                             // leaf
 		config.NewSupplyServiceQueryClientFn(),
+		config.NewSupplyApplicationQuerierFn(),
+		supplyRelayMeter,
 		supplyMiner,
 		config.NewSupplyAccountQuerierFn(),
 		config.NewSupplyBankQuerierFn(),
-		config.NewSupplyApplicationQuerierFn(),
 		config.NewSupplySupplierQuerierFn(),
 		config.NewSupplySessionQuerierFn(),
 		config.NewSupplyProofQueryClientFn(),
@@ -231,6 +232,21 @@ func supplyMiner(
 	return depinject.Configs(deps, depinject.Supply(mnr)), nil
 }
 
+// supplyRelayMeter constructs a RelayMeter instance and returns a new
+// depinject.Config which is supplied with the given deps and the new RelayMeter.
+func supplyRelayMeter(
+	_ context.Context,
+	deps depinject.Config,
+	_ *cobra.Command,
+) (depinject.Config, error) {
+	rm, err := proxy.NewRelayMeter(deps)
+	if err != nil {
+		return nil, err
+	}
+
+	return depinject.Configs(deps, depinject.Supply(rm)), nil
+}
+
 // supplyTxFactory constructs a cosmostx.Factory instance and returns a new
 // depinject.Config which is supplied with the given deps and the new
 // cosmostx.Factory.

pkg/relayer/interface.go

@@ -1,6 +1,7 @@
 //go:generate mockgen -destination=../../testutil/mockrelayer/relayer_proxy_mock.go -package=mockrelayer . RelayerProxy
 //go:generate mockgen -destination=../../testutil/mockrelayer/miner_mock.go -package=mockrelayer . Miner
 //go:generate mockgen -destination=../../testutil/mockrelayer/relayer_sessions_manager_mock.go -package=mockrelayer . RelayerSessionsManager
+//go:generate mockgen -destination=../../testutil/mockrelayer/relay_meter_mock.go -package=mockrelayer . RelayMeter
 
 package relayer
 
@@ -162,3 +163,26 @@ type SessionTree interface {
 	// GetTrieSpec returns the trie spec of the SMST.
 	GetTrieSpec() smt.TrieSpec
 }
+
+// RelayMeter is an interface that keeps track of the amount of stake consumed between
+// a single onchain Application and a single onchain Supplier over the course of a single session.
+// It enables the RelayMiner t orate limit the number of requests handled offchain as a function
+// of the optimistic onchain rate limits.
+type RelayMeter interface {
+	// Start starts the relay meter.
+	Start(ctx context.Context) error
+
+	// AccumulateRelayReward adds the relay reward from the incoming request to session's accumulator.
+	// The relay cost is added optimistically, assuming that the relay WILL be volume / reward applicable.
+	// The reason why optimistic AccumulateRelayReward + SetNonApplicableRelayReward is used instead of
+	// a simpler AccumulateVolumeApplicableRelayReward is that when the relay is first seen
+	// we don't know if it will be volume / reward applicable until it is served.
+	// Since the check of whether we allow the relay to be served or not is done before.
+	// To rate limit or not the current relay, we need to optimistically account
+	// for it as volume / reward applicable.
+	AccumulateRelayReward(ctx context.Context, relayRequestMeta servicetypes.RelayRequestMetadata) error
+
+	// SetNonApplicableRelayReward updates the relay meter for the given relay request as
+	// non-applicable between a single Application and a single Supplier for a single session.
+	SetNonApplicableRelayReward(ctx context.Context, relayRequestMeta servicetypes.RelayRequestMetadata) error
+}

pkg/relayer/miner/miner.go

@@ -27,6 +27,7 @@ type miner struct {
 	// serviceQueryClient is used to query for the relay difficulty target hash of a service.
 	// relay_difficulty is the target hash which a relay hash must be less than to be volume/reward applicable.
 	serviceQueryClient client.ServiceQueryClient
+	relayMeter         relayer.RelayMeter
 }
 
 // NewMiner creates a new miner from the given dependencies and options. It
@@ -43,7 +44,7 @@ func NewMiner(
 ) (*miner, error) {
 	mnr := &miner{}
 
-	if err := depinject.Inject(deps, &mnr.serviceQueryClient); err != nil {
+	if err := depinject.Inject(deps, &mnr.serviceQueryClient, &mnr.relayMeter); err != nil {
 		return nil, err
 	}
 
@@ -88,18 +89,30 @@ func (mnr *miner) mapMineRelay(
 ) (_ either.Either[*relayer.MinedRelay], skip bool) {
 	relayBz, err := relay.Marshal()
 	if err != nil {
+		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
+		if meteringError != nil {
+			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		}
 		return either.Error[*relayer.MinedRelay](err), false
 	}
 	relayHashArr := protocol.GetRelayHashFromBytes(relayBz)
 	relayHash := relayHashArr[:]
 
 	relayDifficultyTargetHash, err := mnr.getServiceRelayDifficultyTargetHash(ctx, relay.Req)
 	if err != nil {
+		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
+		if meteringError != nil {
+			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		}
 		return either.Error[*relayer.MinedRelay](err), false
 	}
 
 	// The relay IS NOT volume / reward applicable
 	if !protocol.IsRelayVolumeApplicable(relayHash, relayDifficultyTargetHash) {
+		meteringError := mnr.relayMeter.SetNonApplicableRelayReward(ctx, relay.GetReq().GetMeta())
+		if meteringError != nil {
+			return either.Error[*relayer.MinedRelay](fmt.Errorf("failed to unclaim relay price: %w", meteringError)), false
+		}
 		return either.Success[*relayer.MinedRelay](nil), true
 	}
 

pkg/relayer/miner/miner_test.go

@@ -13,12 +13,14 @@ import (
 	"time"
 
 	"cosmossdk.io/depinject"
+	gomock "github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/pokt-network/poktroll/pkg/crypto/protocol"
 	"github.com/pokt-network/poktroll/pkg/observable/channel"
 	"github.com/pokt-network/poktroll/pkg/relayer"
 	"github.com/pokt-network/poktroll/pkg/relayer/miner"
+	"github.com/pokt-network/poktroll/testutil/mockrelayer"
 	"github.com/pokt-network/poktroll/testutil/testclient/testqueryclients"
 	servicetypes "github.com/pokt-network/poktroll/x/service/types"
 )
@@ -48,8 +50,9 @@ func TestMiner_MinedRelays(t *testing.T) {
 
 	testqueryclients.SetServiceRelayDifficultyTargetHash(t, testSvcId, testRelayMiningTargetHash)
 	serviceQueryClientMock := testqueryclients.NewTestServiceQueryClient(t)
+	testRelayMeter := mockRelayMeter(t)
 
-	deps := depinject.Supply(serviceQueryClientMock)
+	deps := depinject.Supply(serviceQueryClientMock, testRelayMeter)
 	mnr, err := miner.NewMiner(deps)
 	require.NoError(t, err)
 
@@ -154,3 +157,17 @@ func unmarshalHexMinedRelay(
 		Hash:  relayHash,
 	}
 }
+
+// mockRelayMeter returns a mock RelayMeter that is used by the relay miner to claim and unclaim relays.
+func mockRelayMeter(t *testing.T) relayer.RelayMeter {
+	t.Helper()
+
+	ctrl := gomock.NewController(t)
+	relayMeter := mockrelayer.NewMockRelayMeter(ctrl)
+
+	relayMeter.EXPECT().Start(gomock.Any()).Return(nil).AnyTimes()
+	relayMeter.EXPECT().AccumulateRelayReward(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+	relayMeter.EXPECT().SetNonApplicableRelayReward(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
+
+	return relayMeter
+}

pkg/relayer/proxy/errors.go

@@ -17,4 +17,6 @@ var (
 	ErrRelayerProxyUnsupportedTransportType       = sdkerrors.Register(codespace, 9, "unsupported proxy transport type")
 	ErrRelayerProxyInternalError                  = sdkerrors.Register(codespace, 10, "internal error")
 	ErrRelayerProxyMissingSupplierOperatorAddress = sdkerrors.Register(codespace, 11, "supplier operator address is missing")
+	ErrRelayerProxyUnknownApplication             = sdkerrors.Register(codespace, 12, "relayer proxy encountered unknown application")
+	ErrRelayerProxyRateLimited                    = sdkerrors.Register(codespace, 13, "offchain rate limit hit by relayer proxy")
 )

pkg/relayer/proxy/proxy.go

@@ -78,6 +78,11 @@ type relayerProxy struct {
 	// 1. Relay verification to check if the incoming relay matches the supplier hosted by the relay miner;
 	// 2. Relay signing to resolve which keyring key name to use for signing;
 	OperatorAddressToSigningKeyNameMap map[string]string
+
+	// relayMeter keeps track of the total amount of stake an onchhain Application
+	// will owe an onchain Supplier (backed by this RelayMiner) once the session settles.
+	// It also configures over servicing allowance.
+	relayMeter relayer.RelayMeter
 }
 
 // NewRelayerProxy creates a new relayer proxy with the given dependencies or returns
@@ -86,9 +91,15 @@ type relayerProxy struct {
 // Required dependencies:
 //   - cosmosclient.Context
 //   - client.BlockClient
+//   - crypto.RingCache
+//   - client.SupplierQueryClient
 //   - client.SessionQueryClient
 //   - client.SharedQueryClient
-//   - client.SupplierQueryClient
+//   - keyring.Keyring
+//   - client.SharedQueryClient
+//   - client.ApplicationQueryClient
+//   - client.ServiceQueryClient
+//   - client.EventsQueryClient
 //
 // Available options:
 //   - WithSigningKeyNames
@@ -108,6 +119,7 @@ func NewRelayerProxy(
 		&rp.sessionQuerier,
 		&rp.sharedQuerier,
 		&rp.keyring,
+		&rp.relayMeter,
 	); err != nil {
 		return nil, err
 	}
@@ -143,6 +155,13 @@ func (rp *relayerProxy) Start(ctx context.Context) error {
 	// Start the ring cache.
 	rp.ringCache.Start(ctx)
 
+	// Start the relay meter by subscribing to the on-chain events.
+	// This function is non-blocking and the subscription cancellation is handled
+	// by the context passed to the Start method.
+	if err := rp.relayMeter.Start(ctx); err != nil {
+		return err
+	}
+
 	startGroup, ctx := errgroup.WithContext(ctx)
 
 	for _, relayServer := range rp.servers {