Functions: tracking subscriptions

core/scripts/functions/templates/oracle.toml

@@ -43,6 +43,13 @@ maxSecretsSizesList = [10_240, 20_480, 51_200, 102_400, 307_200, 512_000, 1_048_
   updateFrequencySec = 30
   updateTimeoutSec = 10
 
+  [pluginConfig.OnchainSubscriptions]
+  blockConfirmations = 1
+  contractAddress = "{{router_contract_address}}"
+  updateFrequencySec = 30
+  updateTimeoutSec = 10
+  updateRangeSize = 100
+
   [pluginConfig.RateLimiter]
   globalBurst = 5
   globalRPS = 10

core/services/functions/connector_handler.go

@@ -5,6 +5,7 @@ import (
 	"crypto/ecdsa"
 	"encoding/json"
 	"fmt"
+	"math/big"
 
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/api"
@@ -21,31 +22,33 @@ import (
 type functionsConnectorHandler struct {
 	utils.StartStopOnce
 
-	connector   connector.GatewayConnector
-	signerKey   *ecdsa.PrivateKey
-	nodeAddress string
-	storage     s4.Storage
-	allowlist   functions.OnchainAllowlist
-	rateLimiter *hc.RateLimiter
-	lggr        logger.Logger
+	connector     connector.GatewayConnector
+	signerKey     *ecdsa.PrivateKey
+	nodeAddress   string
+	storage       s4.Storage
+	allowlist     functions.OnchainAllowlist
+	subscriptions functions.OnchainSubscriptions
+	rateLimiter   *hc.RateLimiter
+	lggr          logger.Logger
 }
 
 var (
 	_ connector.Signer                  = &functionsConnectorHandler{}
 	_ connector.GatewayConnectorHandler = &functionsConnectorHandler{}
 )
 
-func NewFunctionsConnectorHandler(nodeAddress string, signerKey *ecdsa.PrivateKey, storage s4.Storage, allowlist functions.OnchainAllowlist, rateLimiter *hc.RateLimiter, lggr logger.Logger) (*functionsConnectorHandler, error) {
-	if signerKey == nil || storage == nil || allowlist == nil || rateLimiter == nil {
-		return nil, fmt.Errorf("signerKey, storage, allowlist and rateLimiter must be non-nil")
+func NewFunctionsConnectorHandler(nodeAddress string, signerKey *ecdsa.PrivateKey, storage s4.Storage, allowlist functions.OnchainAllowlist, rateLimiter *hc.RateLimiter, subscriptions functions.OnchainSubscriptions, lggr logger.Logger) (*functionsConnectorHandler, error) {
+	if signerKey == nil || storage == nil || allowlist == nil || rateLimiter == nil || subscriptions == nil {
+		return nil, fmt.Errorf("signerKey, storage, allowlist, rateLimiter and subscriptions must be non-nil")
 	}
 	return &functionsConnectorHandler{
-		nodeAddress: nodeAddress,
-		signerKey:   signerKey,
-		storage:     storage,
-		allowlist:   allowlist,
-		rateLimiter: rateLimiter,
-		lggr:        lggr.Named("FunctionsConnectorHandler"),
+		nodeAddress:   nodeAddress,
+		signerKey:     signerKey,
+		storage:       storage,
+		allowlist:     allowlist,
+		rateLimiter:   rateLimiter,
+		subscriptions: subscriptions,
+		lggr:          lggr.Named("FunctionsConnectorHandler"),
 	}, nil
 }
 
@@ -68,6 +71,10 @@ func (h *functionsConnectorHandler) HandleGatewayMessage(ctx context.Context, ga
 		h.lggr.Errorw("request rate-limited", "id", gatewayId, "address", fromAddr)
 		return
 	}
+	if balance, err := h.subscriptions.GetMaxUserBalance(fromAddr); err != nil || balance.Cmp(big.NewInt(0)) == 0 {
+		h.lggr.Errorw("request is not backed with a funded subscription", "id", gatewayId, "address", fromAddr)
+		return
+	}
 
 	h.lggr.Debugw("handling gateway request", "id", gatewayId, "method", body.Method)
 

core/services/functions/connector_handler_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"errors"
+	"math/big"
 	"testing"
 
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils"
@@ -31,10 +32,11 @@ func TestFunctionsConnectorHandler(t *testing.T) {
 	connector := gcmocks.NewGatewayConnector(t)
 	allowlist := gfmocks.NewOnchainAllowlist(t)
 	rateLimiter, err := hc.NewRateLimiter(hc.RateLimiterConfig{GlobalRPS: 100.0, GlobalBurst: 100, PerSenderRPS: 100.0, PerSenderBurst: 100})
+	subscriptions := gfmocks.NewOnchainSubscriptions(t)
 	require.NoError(t, err)
 	allowlist.On("Start", mock.Anything).Return(nil)
 	allowlist.On("Close", mock.Anything).Return(nil)
-	handler, err := functions.NewFunctionsConnectorHandler(addr.Hex(), privateKey, storage, allowlist, rateLimiter, logger)
+	handler, err := functions.NewFunctionsConnectorHandler(addr.Hex(), privateKey, storage, allowlist, rateLimiter, subscriptions, logger)
 	require.NoError(t, err)
 
 	handler.SetConnector(connector)
@@ -73,6 +75,7 @@ func TestFunctionsConnectorHandler(t *testing.T) {
 			}
 			storage.On("List", ctx, addr).Return(snapshot, nil).Once()
 			allowlist.On("Allow", addr).Return(true).Once()
+			subscriptions.On("GetMaxUserBalance", mock.Anything).Return(big.NewInt(10), nil)
 			connector.On("SendToGateway", ctx, "gw1", mock.Anything).Run(func(args mock.Arguments) {
 				msg, ok := args[2].(*api.Message)
 				require.True(t, ok)
@@ -129,6 +132,7 @@ func TestFunctionsConnectorHandler(t *testing.T) {
 
 			storage.On("Put", ctx, &key, &record, signature).Return(nil).Once()
 			allowlist.On("Allow", addr).Return(true).Once()
+			subscriptions.On("GetMaxUserBalance", mock.Anything).Return(big.NewInt(10), nil)
 			connector.On("SendToGateway", ctx, "gw1", mock.Anything).Run(func(args mock.Arguments) {
 				msg, ok := args[2].(*api.Message)
 				require.True(t, ok)