feat(utils/throttler): throttler revamped

node/node.go

@@ -301,7 +301,7 @@
 			logger,
 		),
 		cfg.MaxConcurrentCompilations,
-		int32(cfg.MaxCompilationQueue),
+		uint64(cfg.MaxCompilationQueue),
 	)
 
 	if cfg.Sequencer {
@@ -320,7 +320,7 @@
 			FeeTokenAddresses: feeTokens,
 		}
 		nodeVM = vm.New(&chainInfo, false, logger)
-		throttledVM = NewThrottledVM(nodeVM, cfg.MaxVMs, int32(cfg.MaxVMQueue))
+		throttledVM = NewThrottledVM(nodeVM, cfg.MaxVMs, uint64(cfg.MaxVMQueue))
 		mempool := mempool.New(database, chain, mempoolLimit, logger)
 		executor := builder.NewExecutor(chain, nodeVM, logger, cfg.SeqDisableFees, false)
 		builder := builder.New(chain, executor)
@@ -380,7 +380,7 @@
 			FeeTokenAddresses: feeTokens,
 		}
 		nodeVM = vm.New(&chainInfo, false, logger)
-		throttledVM = NewThrottledVM(nodeVM, cfg.MaxVMs, int32(cfg.MaxVMQueue))
+		throttledVM = NewThrottledVM(nodeVM, cfg.MaxVMs, uint64(cfg.MaxVMQueue))
 
 		feederGatewayDataSource := sync.NewFeederGatewayDataSource(chain, adaptfeeder.New(client))
 		synchronizer = sync.New(

node/throttled_compiler.go

@@ -5,28 +5,30 @@ import (
 
 	"github.com/NethermindEth/juno/starknet"
 	"github.com/NethermindEth/juno/starknet/compiler"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 )
 
 var _ compiler.Compiler = (*ThrottledCompiler)(nil)
 
 type ThrottledCompiler struct {
-	*utils.Throttler[compiler.Compiler]
+	*throttler.Throttler[compiler.Compiler]
 }
 
 func NewThrottledCompiler(
-	res compiler.Compiler, concurrencyBudget uint, maxQueueLen int32,
+	res compiler.Compiler, concurrencyBudget uint, maxQueueLen uint64,
 ) *ThrottledCompiler {
 	return &ThrottledCompiler{
-		Throttler: utils.NewThrottler(concurrencyBudget, &res).WithMaxQueueLen(maxQueueLen),
+		Throttler: throttler.NewThrottler(
+			concurrencyBudget, &res, throttler.WithMaxQueueLen(maxQueueLen),
+		),
 	}
 }
 
 func (tc *ThrottledCompiler) Compile(
 	ctx context.Context, sierra *starknet.SierraClass,
 ) (*starknet.CasmClass, error) {
 	var result *starknet.CasmClass
-	err := tc.Do(func(c *compiler.Compiler) error {
+	err := tc.Do(ctx, func(c *compiler.Compiler) error {
 		var cErr error
 		result, cErr = (*c).Compile(ctx, sierra)
 		return cErr

node/throttled_compiler_test.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/NethermindEth/juno/node"
 	"github.com/NethermindEth/juno/starknet"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -52,7 +52,7 @@ func TestThrottledCompiler(t *testing.T) {
 
 		// The queue is full, so the next request is rejected.
 		_, err := throttled.Compile(t.Context(), &starknet.SierraClass{})
-		require.ErrorIs(t, err, utils.ErrResourceBusy)
+		require.ErrorIs(t, err, throttler.ErrResourceBusy)
 
 		// Release the four running/queued jobs and let them finish.
 		for range 4 {

node/throttled_vm.go

@@ -1,21 +1,23 @@
 package node
 
 import (
+	"context"
+
 	"github.com/NethermindEth/juno/core"
 	"github.com/NethermindEth/juno/core/felt"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 )
 
 var _ vm.VM = (*ThrottledVM)(nil)
 
 type ThrottledVM struct {
-	*utils.Throttler[vm.VM]
+	*throttler.Throttler[vm.VM]
 }
 
-func NewThrottledVM(res vm.VM, concurrenyBudget uint, maxQueueLen int32) *ThrottledVM {
+func NewThrottledVM(res vm.VM, concurrenyBudget uint, maxQueueLen uint64) *ThrottledVM {
 	return &ThrottledVM{
-		Throttler: utils.NewThrottler(concurrenyBudget, &res).WithMaxQueueLen(maxQueueLen),
+		Throttler: throttler.NewThrottler(concurrenyBudget, &res, throttler.WithMaxQueueLen(maxQueueLen)),
 	}
 }
 
@@ -28,7 +30,8 @@ func (tvm *ThrottledVM) Call(
 	errStack, returnStateDiff bool,
 ) (vm.CallResult, error) {
 	ret := vm.CallResult{}
-	return ret, tvm.Do(func(vm *vm.VM) error {
+	// vm.VM carries no ctx; queued VM calls aren't cancellable yet.
+	return ret, tvm.Do(context.Background(), func(vm *vm.VM) error {
 		var err error
 		ret, err = (*vm).Call(
 			callInfo,
@@ -47,7 +50,8 @@ func (tvm *ThrottledVM) runExec(
 	fn func(inner vm.VM) (vm.ExecutionResults, error),
 ) (vm.ExecutionResults, error) {
 	var result vm.ExecutionResults
-	return result, tvm.Do(func(inner *vm.VM) error {
+	// vm.VM carries no ctx; queued VM calls aren't cancellable yet.
+	return result, tvm.Do(context.Background(), func(inner *vm.VM) error {
 		var err error
 		result, err = fn(*inner)
 		return err

rpc/v10/simulation.go

@@ -14,7 +14,7 @@
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 )
 
@@ -319,7 +319,7 @@
 			network,
 		)
 		if aErr != nil {
-			if errors.Is(aErr, utils.ErrResourceBusy) {
+			if errors.Is(aErr, throttler.ErrResourceBusy) {
 				return nil, nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 			}
 			return nil, nil, jsonrpc.Err(jsonrpc.InvalidParams, aErr.Error())
@@ -335,7 +335,7 @@
 }
 
 func handleExecutionError(err error) *jsonrpc.Error {
-	if errors.Is(err, utils.ErrResourceBusy) {
+	if errors.Is(err, throttler.ErrResourceBusy) {
 		return rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 	}
 	var txnExecutionError vm.TransactionExecutionError

rpc/v10/simulation_pkg_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/stretchr/testify/require"
 )
@@ -130,7 +130,7 @@ func TestHandleExecutionError(t *testing.T) {
 	}{
 		{
 			name:         "Resource Busy Error",
-			err:          utils.ErrResourceBusy,
+			err:          throttler.ErrResourceBusy,
 			jsonRPCError: rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr),
 		},
 		{

rpc/v10/trace.go

@@ -18,6 +18,7 @@
 	"github.com/NethermindEth/juno/rpc/rpccore"
 	"github.com/NethermindEth/juno/sync"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 )
 
@@ -100,7 +101,7 @@
 		false,
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		return nil, MakeContractError(json.RawMessage(err.Error()))
@@ -192,7 +193,7 @@
 	httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10))
 
 	if vmErr != nil {
-		if errors.Is(vmErr, utils.ErrResourceBusy) {
+		if errors.Is(vmErr, throttler.ErrResourceBusy) {
 			return nil, nil, httpHeader, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		return nil, nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(vmErr.Error())

rpc/v10/transaction.go

@@ -17,6 +17,7 @@
 	"github.com/NethermindEth/juno/starknet"
 	"github.com/NethermindEth/juno/starknet/compiler"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"go.uber.org/zap"
 )
 
@@ -200,7 +201,7 @@
 		ctx, h.compiler, tx, h.bcReader.Network(),
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 		}
 		return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(err.Error())

rpc/v8/simulation.go

@@ -14,7 +14,7 @@
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/consensys/gnark-crypto/ecc/stark-curve/fp"
 )
@@ -197,7 +197,7 @@
 			network,
 		)
 		if aErr != nil {
-			if errors.Is(aErr, utils.ErrResourceBusy) {
+			if errors.Is(aErr, throttler.ErrResourceBusy) {
 				return nil, nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 			}
 			return nil, nil, jsonrpc.Err(jsonrpc.InvalidParams, aErr.Error())
@@ -213,7 +213,7 @@
 }
 
 func handleExecutionError(err error) *jsonrpc.Error {
-	if errors.Is(err, utils.ErrResourceBusy) {
+	if errors.Is(err, throttler.ErrResourceBusy) {
 		return rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 	}
 	var txnExecutionError vm.TransactionExecutionError

rpc/v8/simulation_pkg_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/stretchr/testify/require"
 )
@@ -123,7 +123,7 @@ func TestHandleExecutionError(t *testing.T) {
 	}{
 		{
 			name:         "Resource Busy Error",
-			err:          utils.ErrResourceBusy,
+			err:          throttler.ErrResourceBusy,
 			jsonRPCError: rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr),
 		},
 		{

rpc/v8/trace.go

@@ -16,6 +16,7 @@ import (
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 )
 
@@ -239,7 +240,7 @@ func (h *Handler) traceBlockTransactionWithVM(block *core.Block) (
 	httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10))
 
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return nil, httpHeader, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		// Since we are tracing an existing block, we know that there should be no errors during execution. If we encounter any,
@@ -387,7 +388,7 @@ func (h *Handler) Call(funcCall *FunctionCall, id *BlockID) ([]*felt.Felt, *json
 		false,
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		return nil, MakeContractError(json.RawMessage(err.Error()))

rpc/v8/transaction.go

@@ -21,6 +21,7 @@
 	"github.com/NethermindEth/juno/starknet"
 	"github.com/NethermindEth/juno/starknet/compiler"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"go.uber.org/zap"
 )
 
@@ -664,7 +665,7 @@
 		ctx, h.compiler, tx, h.bcReader.Network(),
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 		}
 		return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(err.Error())

rpc/v9/simulation.go

@@ -14,7 +14,7 @@
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/consensys/gnark-crypto/ecc/stark-curve/fp"
 )
@@ -207,7 +207,7 @@
 			network,
 		)
 		if aErr != nil {
-			if errors.Is(aErr, utils.ErrResourceBusy) {
+			if errors.Is(aErr, throttler.ErrResourceBusy) {
 				return nil, nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 			}
 			return nil, nil, jsonrpc.Err(jsonrpc.InvalidParams, aErr.Error())
@@ -223,7 +223,7 @@
 }
 
 func handleExecutionError(err error) *jsonrpc.Error {
-	if errors.Is(err, utils.ErrResourceBusy) {
+	if errors.Is(err, throttler.ErrResourceBusy) {
 		return rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 	}
 	var txnExecutionError vm.TransactionExecutionError

rpc/v9/simulation_pkg_test.go

@@ -9,7 +9,7 @@ import (
 	"github.com/NethermindEth/juno/core/felt"
 	"github.com/NethermindEth/juno/jsonrpc"
 	"github.com/NethermindEth/juno/rpc/rpccore"
-	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 	"github.com/stretchr/testify/require"
 )
@@ -123,7 +123,7 @@ func TestHandleExecutionError(t *testing.T) {
 	}{
 		{
 			name:         "Resource Busy Error",
-			err:          utils.ErrResourceBusy,
+			err:          throttler.ErrResourceBusy,
 			jsonRPCError: rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr),
 		},
 		{

rpc/v9/trace.go

@@ -18,6 +18,7 @@ import (
 	"github.com/NethermindEth/juno/rpc/rpccore"
 	"github.com/NethermindEth/juno/sync"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"github.com/NethermindEth/juno/vm"
 )
 
@@ -158,7 +159,7 @@ func (h *Handler) Call(funcCall *FunctionCall, id *BlockID) ([]*felt.Felt, *json
 		false,
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return nil, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		return nil, MakeContractError(json.RawMessage(err.Error()))
@@ -227,7 +228,7 @@ func traceTransactionsWithState(
 	httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10))
 
 	if vmErr != nil {
-		if errors.Is(vmErr, utils.ErrResourceBusy) {
+		if errors.Is(vmErr, throttler.ErrResourceBusy) {
 			return nil, httpHeader, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledVMErr)
 		}
 		return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(vmErr.Error())

rpc/v9/transaction.go

@@ -21,6 +21,7 @@
 	"github.com/NethermindEth/juno/starknet"
 	"github.com/NethermindEth/juno/starknet/compiler"
 	"github.com/NethermindEth/juno/utils"
+	"github.com/NethermindEth/juno/utils/throttler"
 	"go.uber.org/zap"
 )
 
@@ -723,7 +724,7 @@
 		ctx, h.compiler, tx, h.bcReader.Network(),
 	)
 	if err != nil {
-		if errors.Is(err, utils.ErrResourceBusy) {
+		if errors.Is(err, throttler.ErrResourceBusy) {
 			return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(rpccore.ThrottledCompilerErr)
 		}
 		return AddTxResponse{}, rpccore.ErrInternal.CloneWithData(err.Error())