Performance improvements in external files handling (#253)

Author: oskarszoon

settings.conf

@@ -1511,6 +1511,12 @@ utxostore_prunerMaxConcurrentOperations = 0
 # and it is required for large blocks which load in the same tx multiple times, e.g. 814337
 utxostore_useExternalTxCache = true
 
+# Maximum concurrent external storage operations (multi-record transactions)
+# Set to 0 for unlimited. Lower values prevent overwhelming Aerospike connection pool.
+# Should be less than or equal to Aerospike ConnectionQueueSize.
+utxostore_externalStoreConcurrency          = 16
+utxostore_externalStoreConcurrency.docker.m = 4
+
 # utxos per record, if larger it will externalize the transaction
 # ! do not change this after starting a node !
 utxostore_utxoBatchSize          = 128

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", 16, 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

@@ -830,6 +830,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 {

stores/utxo/aerospike/get.go

@@ -1316,6 +1316,17 @@ func (s *Store) GetTxFromExternalStore(ctx context.Context, previousTxHash chain
 		tracing.WithHistogram(prometheusTxMetaAerospikeMapGetExternal),
 	)
 
+	if s.externalTxCache != nil {
+		return s.externalTxCache.GetOrSet(previousTxHash, func() (*bt.Tx, bool, error) {
+			tx, err := s.getExternalTransaction(ctx, previousTxHash)
+			if err != nil {
+				return nil, false, err
+			}
+
+			return tx, true, nil
+		})
+	}
+
 	return s.getExternalTransaction(ctx, previousTxHash)
 }