allow pool to be blocked

cmd/run.go

@@ -401,8 +401,9 @@ func startProfilingHttpServer(c metrics.Config) {
 	mux.HandleFunc(metrics.ProfilingSymbolEndpoint, pprof.Symbol)
 	mux.HandleFunc(metrics.ProfilingTraceEndpoint, pprof.Trace)
 	profilingServer := &http.Server{
-		Handler:     mux,
-		ReadTimeout: two * time.Minute,
+		Handler:           mux,
+		ReadHeaderTimeout: two * time.Minute,
+		ReadTimeout:       two * time.Minute,
 	}
 	log.Infof("profiling server listening on port %d", c.ProfilingPort)
 	if err := profilingServer.Serve(lis); err != nil {
@@ -426,8 +427,9 @@ func startMetricsHttpServer(c metrics.Config) {
 	mux.Handle(metrics.Endpoint, promhttp.Handler())
 
 	metricsServer := &http.Server{
-		Handler:     mux,
-		ReadTimeout: ten * time.Second,
+		Handler:           mux,
+		ReadHeaderTimeout: ten * time.Second,
+		ReadTimeout:       ten * time.Second,
 	}
 	log.Infof("metrics server listening on port %d", c.Port)
 	if err := metricsServer.Serve(lis); err != nil {

config/default.go

@@ -20,6 +20,7 @@ MaxConns = 200
 
 [Pool]
 FreeClaimGasLimit = 150000
+IntervalToRefreshBlockedAddresses = "5m"
 MaxTxBytesSize=30132
 MaxTxDataBytesSize=30000
 DefaultMinGasPriceAllowed = 1000000000

config/environments/local/local.node.config.toml

@@ -16,6 +16,7 @@ MaxConns = 200
 
 [Pool]
 FreeClaimGasLimit = 1500000
+IntervalToRefreshBlockedAddresses = "5m"
 MaxTxBytesSize=30132
 MaxTxDataBytesSize=30000
 DefaultMinGasPriceAllowed = 1000000000

config/environments/public/public.node.config.toml

@@ -14,6 +14,7 @@ MaxConns = 200
 
 [Pool]
 FreeClaimGasLimit = 1500000
+IntervalToRefreshBlockedAddresses = "5m"
 MaxTxBytesSize=30132
 MaxTxDataBytesSize=30000
 DefaultMinGasPriceAllowed = 1000000000

db/migrations/pool/0006.sql

@@ -4,4 +4,4 @@ ADD COLUMN deposit_count BIGINT;
 
 -- +migrate Down
 ALTER TABLE pool.transaction
-DROP COLUMN deposit_count;
+DROP COLUMN deposit_count;

db/migrations/pool/0007.sql

@@ -0,0 +1,8 @@
+-- +migrate Up
+CREATE TABLE pool.blocked
+(
+    addr varchar NOT NULL PRIMARY KEY
+);
+
+-- +migrate Down
+DROP TABLE pool.blocked;

jsonrpc/server.go

@@ -123,9 +123,10 @@ func (s *Server) startHTTP() error {
 	mux.Handle("/", tollbooth.LimitFuncHandler(lmt, s.handle))
 
 	s.srv = &http.Server{
-		Handler:      mux,
-		ReadTimeout:  s.config.ReadTimeoutInSec * time.Second,
-		WriteTimeout: s.config.WriteTimeoutInSec * time.Second,
+		Handler:           mux,
+		ReadHeaderTimeout: s.config.ReadTimeoutInSec * time.Second,
+		ReadTimeout:       s.config.ReadTimeoutInSec * time.Second,
+		WriteTimeout:      s.config.WriteTimeoutInSec * time.Second,
 	}
 	log.Infof("http server started: %s", address)
 	if err := s.srv.Serve(lis); err != nil {
@@ -160,9 +161,10 @@ func (s *Server) startWS() {
 	mux.HandleFunc("/", s.handleWs)
 
 	s.wsSrv = &http.Server{
-		Handler:      mux,
-		ReadTimeout:  s.config.ReadTimeoutInSec * time.Second,
-		WriteTimeout: s.config.WriteTimeoutInSec * time.Second,
+		Handler:           mux,
+		ReadHeaderTimeout: s.config.ReadTimeoutInSec * time.Second,
+		ReadTimeout:       s.config.ReadTimeoutInSec * time.Second,
+		WriteTimeout:      s.config.WriteTimeoutInSec * time.Second,
 	}
 	s.wsUpgrader = websocket.Upgrader{
 		ReadBufferSize:  wsBufferSizeLimitInBytes,

pool/config.go

@@ -10,6 +10,10 @@ type Config struct {
 	// FreeClaimGasLimit is the max gas allowed use to do a free claim
 	FreeClaimGasLimit uint64 `mapstructure:"FreeClaimGasLimit"`
 
+	// IntervalToRefreshBlockedAddresses is the time it takes to sync the
+	// blocked address list from db to memory
+	IntervalToRefreshBlockedAddresses types.Duration `mapstructure:"IntervalToRefreshBlockedAddresses"`
+
 	// MaxTxBytesSize is the max size of a transaction in bytes
 	MaxTxBytesSize uint64 `mapstructure:"MaxTxBytesSize"`
 

pool/errors.go

@@ -27,6 +27,9 @@ var (
 	// ErrInvalidSender is returned if the transaction contains an invalid signature.
 	ErrInvalidSender = errors.New("invalid sender")
 
+	// ErrBlockedSender is returned if the transaction is sent by a blocked account.
+	ErrBlockedSender = errors.New("blocked sender")
+
 	// ErrNonceTooLow is returned if the nonce of a transaction is lower than the
 	// one present in the local chain.
 	ErrNonceTooLow = errors.New("nonce too low")

pool/interfaces.go

@@ -32,6 +32,7 @@ type storage interface {
 	GetTxZkCountersByHash(ctx context.Context, hash common.Hash) (*state.ZKCounters, error)
 	DeleteTransactionByHash(ctx context.Context, hash common.Hash) error
 	MarkWIPTxsAsPending(ctx context.Context) error
+	GetAllAddressesBlocked(ctx context.Context) ([]common.Address, error)
 	MinGasPriceSince(ctx context.Context, timestamp time.Time) (uint64, error)
 	DepositCountExists(ctx context.Context, depositCount uint64) (bool, error)
 }

pool/pgpoolstorage/pgpoolstorage.go

@@ -656,6 +656,33 @@ func (p *PostgresPoolStorage) UpdateTxWIPStatus(ctx context.Context, hash common
 	return nil
 }
 
+// GetAllAddressesBlocked get all addresses blocked
+func (p *PostgresPoolStorage) GetAllAddressesBlocked(ctx context.Context) ([]common.Address, error) {
+	sql := `SELECT addr FROM pool.blocked`
+
+	rows, err := p.db.Query(ctx, sql)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, nil
+		} else {
+			return nil, err
+		}
+	}
+	defer rows.Close()
+
+	var addrs []common.Address
+	for rows.Next() {
+		var addr string
+		err := rows.Scan(&addr)
+		if err != nil {
+			return nil, err
+		}
+		addrs = append(addrs, common.HexToAddress(addr))
+	}
+
+	return addrs, nil
+}
+
 // DepositCountExists checks if already exists a transaction in the pool with the
 // provided deposit count
 func (p *PostgresPoolStorage) DepositCountExists(ctx context.Context, depositCount uint64) (bool, error) {

pool/pool.go

@@ -44,6 +44,7 @@ type Pool struct {
 	l2BridgeAddr            common.Address
 	chainID                 uint64
 	cfg                     Config
+	blockedAddresses        sync.Map
 	minSuggestedGasPrice    *big.Int
 	minSuggestedGasPriceMux *sync.RWMutex
 	eventLog                *event.EventLog
@@ -58,15 +59,56 @@ type preExecutionResponse struct {
 
 // NewPool creates and initializes an instance of Pool
 func NewPool(cfg Config, s storage, st stateInterface, l2BridgeAddr common.Address, chainID uint64, eventLog *event.EventLog) *Pool {
-	return &Pool{
+	p := &Pool{
 		cfg:                     cfg,
 		storage:                 s,
 		state:                   st,
 		l2BridgeAddr:            l2BridgeAddr,
 		chainID:                 chainID,
+		blockedAddresses:        sync.Map{},
 		minSuggestedGasPriceMux: new(sync.RWMutex),
 		eventLog:                eventLog,
 	}
+
+	p.refreshBlockedAddresses()
+	go func(cfg *Config, p *Pool) {
+		for {
+			time.Sleep(cfg.IntervalToRefreshBlockedAddresses.Duration)
+			p.refreshBlockedAddresses()
+		}
+	}(&cfg, p)
+	return p
+}
+
+// refreshBlockedAddresses refreshes the list of blocked addresses for the provided instance of pool
+func (p *Pool) refreshBlockedAddresses() {
+	blockedAddresses, err := p.storage.GetAllAddressesBlocked(context.Background())
+	if err != nil {
+		log.Error("failed to load blocked addresses")
+		return
+	}
+
+	blockedAddressesMap := sync.Map{}
+	for _, blockedAddress := range blockedAddresses {
+		blockedAddressesMap.Store(blockedAddress.String(), 1)
+		p.blockedAddresses.Store(blockedAddress.String(), 1)
+	}
+
+	unblockedAddresses := []string{}
+	p.blockedAddresses.Range(func(key, value any) bool {
+		addrHex := key.(string)
+		_, found := blockedAddressesMap.Load(addrHex)
+		if found {
+			return true
+		}
+
+		unblockedAddresses = append(unblockedAddresses, addrHex)
+		return true
+	})
+
+	for _, unblockedAddress := range unblockedAddresses {
+		p.blockedAddresses.Delete(unblockedAddress)
+	}
 }
 
 // StartPollingMinSuggestedGasPrice starts polling the minimum suggested gas price
@@ -300,6 +342,12 @@ func (p *Pool) validateTx(ctx context.Context, poolTx Transaction) error {
 		return ErrInvalidSender
 	}
 
+	// check if sender is blocked
+	_, blocked := p.blockedAddresses.Load(from.String())
+	if blocked {
+		return ErrBlockedSender
+	}
+
 	lastL2Block, err := p.state.GetLastL2Block(ctx, nil)
 	if err != nil {
 		return err