Add mutex locks to prevent db transaction conflicts

internal/start/private.go

@@ -181,10 +181,10 @@ func PrivateMode() {
 		g.Status(http.StatusOK)
 	})
 	handlers := []gin.HandlerFunc{
-		relayer.FilterByClientID(),
+		// relayer.FilterByClientID(),
 		jsonrpc.Controller(client.NewRpcAdapter(c, d)),
 		jsonrpc.WithOTELTracerAttributes(),
-		relayer.MapUserOpHashToClientID(),
+		// relayer.MapUserOpHashToClientID(),
 	}
 	r.POST("/", handlers...)
 	r.POST("/rpc", handlers...)

pkg/modules/compose.go

@@ -1,12 +1,14 @@
 package modules
 
+import "fmt"
+
 // ComposeBatchHandlerFunc combines many BatchHandlers into one.
 func ComposeBatchHandlerFunc(fns ...BatchHandlerFunc) BatchHandlerFunc {
 	return func(ctx *BatchHandlerCtx) error {
-		for _, fn := range fns {
+		for i, fn := range fns {
 			err := fn(ctx)
 			if err != nil {
-				return err
+				return fmt.Errorf("error at batch handler %d: %w", i, err)
 			}
 		}
 

pkg/modules/paymaster/reputation.go

@@ -4,6 +4,7 @@ package paymaster
 
 import (
 	"errors"
+	"sync"
 
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/dgraph-io/badger/v3"
@@ -15,12 +16,14 @@ import (
 // UserOperation.
 type Reputation struct {
 	db *badger.DB
+	mu *sync.Mutex
 }
 
 // New returns an instance of a Reputation object to track and appropriately process userOps by paymaster
 // status.
 func New(db *badger.DB) *Reputation {
-	return &Reputation{db}
+	mu := &sync.Mutex{}
+	return &Reputation{db, mu}
 }
 
 // CheckStatus returns a UserOpHandler that is used by the Client to determine if the userOp is allowed based
@@ -37,7 +40,9 @@ func (r *Reputation) CheckStatus() modules.UserOpHandlerFunc {
 				return nil
 			}
 
+			// r.mu.Lock()
 			status, err := getStatus(txn, paymaster)
+			// r.mu.Unlock()
 			if err != nil {
 				return err
 			}
@@ -62,6 +67,8 @@ func (r *Reputation) IncOpsSeen() modules.UserOpHandlerFunc {
 				return nil
 			}
 
+			// r.mu.Lock()
+			// defer r.mu.Unlock()
 			return incrementOpsSeenByPaymaster(txn, paymaster)
 		})
 	}
@@ -87,6 +94,8 @@ func (r *Reputation) IncOpsIncluded() modules.BatchHandlerFunc {
 				}
 			}
 
+			// r.mu.Lock()
+			// defer r.mu.Unlock()
 			return incrementOpsIncludedByPaymasters(txn, c, ps.ToSlice()...)
 		})
 	}

pkg/modules/relay/relayer.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"math/big"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/dgraph-io/badger/v3"
@@ -45,6 +46,7 @@ type Relayer struct {
 	bannedThreshold  int
 	bannedTimeWindow time.Duration
 	waitTimeout      time.Duration
+	mu               *sync.Mutex
 }
 
 // New initializes a new EOA relayer for sending batches to the EntryPoint with IP throttling protection.
@@ -66,6 +68,7 @@ func New(
 		bannedThreshold:  DefaultBanThreshold,
 		bannedTimeWindow: DefaultBanTimeWindow,
 		waitTimeout:      DefaultWaitTimeout,
+		mu:               &sync.Mutex{},
 	}
 }
 
@@ -180,11 +183,15 @@ func (r *Relayer) MapUserOpHashToClientID() gin.HandlerFunc {
 			WithValues("userop_hash", hash).
 			WithValues("client_id", cid)
 		err = r.db.Update(func(txn *badger.Txn) error {
+			r.mu.Lock()
 			err := mapUserOpHashToClientID(txn, hash, cid)
+			r.mu.Unlock()
 			if err != nil {
 				return err
 			}
 
+			r.mu.Lock()
+			defer r.mu.Unlock()
 			return incrementOpsSeenByClientID(txn, cid, r.bannedTimeWindow)
 		})
 		if err != nil {
@@ -221,8 +228,11 @@ func (r *Relayer) SendUserOperation() modules.BatchHandlerFunc {
 			// Delete any userOpHash entries from dropped userOps.
 			if len(ctx.PendingRemoval) > 0 {
 				hashes := getUserOpHashesFromOps(ctx.EntryPoint, ctx.ChainID, ctx.PendingRemoval...)
-				if err := removeUserOpHashEntries(txn, hashes...); err != nil {
-					return err
+				r.mu.Lock()
+				err := removeUserOpHashEntries(txn, hashes...)
+				r.mu.Unlock()
+				if err != nil {
+					return fmt.Errorf("checkpoint 1: %w", err)
 				}
 			}
 
@@ -238,8 +248,11 @@ func (r *Relayer) SendUserOperation() modules.BatchHandlerFunc {
 					estRev = append(estRev, revert.Reason)
 
 					hashes := getUserOpHashesFromOps(ctx.EntryPoint, ctx.ChainID, ctx.PendingRemoval...)
-					if err := removeUserOpHashEntries(txn, hashes...); err != nil {
-						return err
+					r.mu.Lock()
+					err := removeUserOpHashEntries(txn, hashes...)
+					r.mu.Unlock()
+					if err != nil {
+						return fmt.Errorf("checkpoint 2: %w", err)
 					}
 				} else {
 					opts.GasLimit = est
@@ -260,19 +273,23 @@ func (r *Relayer) SendUserOperation() modules.BatchHandlerFunc {
 
 			hashes := getUserOpHashesFromOps(ctx.EntryPoint, ctx.ChainID, ctx.Batch...)
 			del = append([]string{}, hashes...)
+			r.mu.Lock()
+			defer r.mu.Unlock()
 			return incrementOpsIncludedByUserOpHashes(txn, r.bannedTimeWindow, hashes...)
 		})
 		if err != nil {
-			return err
+			return fmt.Errorf("relayer batchHandler error 1: %w", err)
 		}
 
 		// Delete remaining userOpHash entries from submitted userOps.
 		// Perform update in new txn to avoid db conflicts.
 		err = r.db.Update(func(txn *badger.Txn) error {
+			r.mu.Lock()
+			defer r.mu.Unlock()
 			return removeUserOpHashEntries(txn, del...)
 		})
 		if err != nil {
-			return err
+			return fmt.Errorf("relayer batchHandler error 2: %w", err)
 		}
 
 		return nil

pkg/modules/relay/utils.go

@@ -1,6 +1,7 @@
 package relay
 
 import (
+	"fmt"
 	"math/big"
 	"strconv"
 	"time"
@@ -87,7 +88,7 @@ func incrementOpsIncludedByUserOpHashes(txn *badger.Txn, ttl time.Duration, user
 			return nil
 		}
 		if err != nil {
-			return err
+			return fmt.Errorf("checkpoint 3: %w", err)
 		}
 
 		var value []byte
@@ -96,13 +97,13 @@ func incrementOpsIncludedByUserOpHashes(txn *badger.Txn, ttl time.Duration, user
 			return nil
 		})
 		if err != nil {
-			return err
+			return fmt.Errorf("checkpoint 4: %w", err)
 		}
 
 		cid := string(value)
 		opsSeen, opsIncluded, err := getOpsCountByClientID(txn, cid)
 		if err != nil {
-			return err
+			return fmt.Errorf("checkpoint 5: %w", err)
 		}
 
 		var opsCountValue string
@@ -115,7 +116,7 @@ func incrementOpsIncludedByUserOpHashes(txn *badger.Txn, ttl time.Duration, user
 
 		e := badger.NewEntry(getOpsCountKey(cid), []byte(opsCountValue)).WithTTL(ttl)
 		if err := txn.SetEntry(e); err != nil {
-			return err
+			return fmt.Errorf("checkpoint 6: %w", err)
 		}
 	}