Add semaphore to external tx storing in Aerospike to prevent locking timeout issues

Author: oskarszoon

settings/interface.go

@@ -372,6 +372,7 @@ type UtxoStoreSettings struct {
 	DBTimeout                         time.Duration
 	UseExternalTxCache                bool
 	ExternalizeAllTransactions        bool
+	ExternalStoreConcurrency          int // Maximum concurrent external storage operations (0 = unlimited)
 	PostgresMaxIdleConns              int
 	PostgresMaxOpenConns              int
 	VerboseDebug                      bool

settings/settings.go

@@ -357,6 +357,7 @@ func NewSettings(alternativeContext ...string) *Settings {
 			DBTimeout:                         getDuration("utxostore_dbTimeoutDuration", 5*time.Second, alternativeContext...),
 			UseExternalTxCache:                getBool("utxostore_useExternalTxCache", true, alternativeContext...),
 			ExternalizeAllTransactions:        getBool("utxostore_externalizeAllTransactions", false, alternativeContext...),
+			ExternalStoreConcurrency:          getInt("utxostore_externalStoreConcurrency", 64, alternativeContext...),
 			PostgresMaxIdleConns:              getInt("utxostore_utxo_postgresMaxIdleConns", 10, alternativeContext...),
 			PostgresMaxOpenConns:              getInt("utxostore_utxo_postgresMaxOpenConns", 80, alternativeContext...),
 			VerboseDebug:                      getBool("utxostore_verbose_debug", false, alternativeContext...),

stores/utxo/aerospike/aerospike.go

@@ -133,8 +133,9 @@ type Store struct {
 	externalStore       blob.Store
 	utxoBatchSize       int
 	externalTxCache     *util.ExpiringConcurrentCache[chainhash.Hash, *bt.Tx]
-	indexMutex          sync.Mutex // Mutex for index creation operations
-	indexOnce           sync.Once  // Ensures index creation/wait is only done once per process
+	externalStoreSem    chan struct{} // Semaphore to limit concurrent external storage operations
+	indexMutex          sync.Mutex    // Mutex for index creation operations
+	indexOnce           sync.Once     // Ensures index creation/wait is only done once per process
 }
 
 // New creates a new Aerospike-based UTXO store.
@@ -191,6 +192,12 @@ func New(ctx context.Context, logger ulogger.Logger, tSettings *settings.Setting
 		externalTxCache = util.NewExpiringConcurrentCache[chainhash.Hash, *bt.Tx](10 * time.Second)
 	}
 
+	// Initialize external store semaphore if concurrency limit is set
+	var externalStoreSem chan struct{}
+	if tSettings.UtxoStore.ExternalStoreConcurrency > 0 {
+		externalStoreSem = make(chan struct{}, tSettings.UtxoStore.ExternalStoreConcurrency)
+	}
+
 	s := &Store{
 		ctx:       ctx,
 		url:       aerospikeURL,
@@ -199,10 +206,11 @@ func New(ctx context.Context, logger ulogger.Logger, tSettings *settings.Setting
 		setName:   setName,
 		logger:    logger,
 
-		settings:        tSettings,
-		externalStore:   externalStore,
-		utxoBatchSize:   utxoBatchSize,
-		externalTxCache: externalTxCache,
+		settings:         tSettings,
+		externalStore:    externalStore,
+		utxoBatchSize:    utxoBatchSize,
+		externalTxCache:  externalTxCache,
+		externalStoreSem: externalStoreSem,
 	}
 
 	// Ensure index creation/wait is only done once per process

stores/utxo/aerospike/create.go

@@ -833,6 +833,12 @@ func (s *Store) storeExternallyWithLock(
 	fileType fileformat.FileType,
 	funcName string,
 ) {
+	// Acquire semaphore to limit concurrent external storage operations
+	if s.externalStoreSem != nil {
+		s.externalStoreSem <- struct{}{}
+		defer func() { <-s.externalStoreSem }()
+	}
+
 	// Acquire lock FIRST to prevent duplicate work
 	lockKey, err := s.acquireLock(bItem.txHash, len(binsToStore))
 	if err != nil {