Delete at height safely

services/blockassembly/BlockAssembler.go

@@ -149,6 +149,7 @@ type BlockAssembler struct {
 
 	// unminedCleanupTicker manages periodic cleanup of old unmined transactions
 	unminedCleanupTicker *time.Ticker
+
 	// cachedCandidate stores the cached mining candidate
 	cachedCandidate *CachedMiningCandidate
 

services/p2p/peer_registry_cache.go

@@ -75,8 +75,8 @@ func getPeerRegistryCacheFilePath(configuredDir string) string {
 
 // SavePeerRegistryCache saves the peer registry data to a JSON file
 func (pr *PeerRegistry) SavePeerRegistryCache(cacheDir string) error {
-	pr.mu.RLock()
-	defer pr.mu.RUnlock()
+	pr.mu.Lock()
+	defer pr.mu.Unlock()
 
 	cache := &PeerRegistryCache{
 		Version:     PeerRegistryCacheVersion,

services/pruner/server.go

@@ -133,7 +133,6 @@ func (s *Server) Init(ctx context.Context) error {
 								case s.prunerCh <- height32:
 									s.logger.Debugf("Queued pruning for height %d from BlockPersisted notification", height32)
 								default:
-									s.logger.Debugf("Pruning already in progress for height %d", height32)
 								}
 							}
 						}
@@ -145,7 +144,6 @@ func (s *Server) Init(ctx context.Context) error {
 				persistedHeight := s.lastPersistedHeight.Load()
 				if persistedHeight > 0 {
 					// Block persister is running - BlockPersisted notifications will handle pruning
-					s.logger.Debugf("Block notification received but block persister is active (persisted height: %d), skipping", persistedHeight)
 					continue
 				}
 
@@ -186,7 +184,6 @@ func (s *Server) Init(ctx context.Context) error {
 					case s.prunerCh <- state.CurrentHeight:
 						s.logger.Debugf("Queued pruning for height %d from Block notification (mined_set=true)", state.CurrentHeight)
 					default:
-						s.logger.Debugf("Pruning already in progress for height %d", state.CurrentHeight)
 					}
 				}
 			}

services/pruner/worker.go

@@ -104,37 +104,65 @@ func (s *Server) prunerProcessor(ctx context.Context) {
 					continue
 				}
 
-				// Wait for pruner to complete with timeout
+				// Wait for pruner to complete with optional timeout
 				prunerTimeout := s.settings.Pruner.JobTimeout
-				timeoutTimer := time.NewTimer(prunerTimeout)
-				defer timeoutTimer.Stop()
 
-				select {
-				case status := <-doneCh:
-					if status != "completed" {
-						s.logger.Warnf("Pruner for height %d finished with status: %s", latestHeight, status)
-						prunerErrors.WithLabelValues("dah_pruner").Inc()
-					} else {
-						s.logger.Infof("Pruner for height %d completed successfully", latestHeight)
-						prunerDuration.WithLabelValues("dah_pruner").Observe(time.Since(startTime).Seconds())
-						prunerProcessed.Inc()
+				if prunerTimeout == 0 {
+					// Wait indefinitely (no timeout)
+					select {
+					case status := <-doneCh:
+						if status != "completed" {
+							s.logger.Warnf("Pruner for height %d finished with status: %s", latestHeight, status)
+							prunerErrors.WithLabelValues("dah_pruner").Inc()
+						} else {
+							// Get job info to log records processed
+							recordsProcessed := int64(0)
+							if job := s.prunerService.GetJobByHeight(latestHeight); job != nil {
+								recordsProcessed = job.RecordsProcessed.Load()
+							}
+							s.logger.Infof("Pruner for height %d completed successfully, pruned %d records", latestHeight, recordsProcessed)
+							prunerDuration.WithLabelValues("dah_pruner").Observe(time.Since(startTime).Seconds())
+							prunerProcessed.Inc()
+						}
+					case <-ctx.Done():
+						return
 					}
-				case <-timeoutTimer.C:
-					s.logger.Infof("Pruner for height %d exceeded coordinator timeout of %v - pruner continues in background, re-queuing immediately", latestHeight, prunerTimeout)
-					// Note: This is not an error - the pruner job continues processing in the background.
-					// The coordinator re-queues immediately to check again.
-					// Very large pruners may take longer than the timeout and require multiple iterations.
+				} else {
+					// Use timeout
+					timeoutTimer := time.NewTimer(prunerTimeout)
+					defer timeoutTimer.Stop()
 
-					// Immediately re-queue to check again (non-blocking)
 					select {
-					case s.prunerCh <- latestHeight:
-						s.logger.Debugf("Re-queued pruner for height %d after timeout", latestHeight)
-					default:
-						// Channel full, will be retried when notifications trigger again
-						s.logger.Debugf("Pruner channel full, will retry on next notification")
+					case status := <-doneCh:
+						if status != "completed" {
+							s.logger.Warnf("Pruner for height %d finished with status: %s", latestHeight, status)
+							prunerErrors.WithLabelValues("dah_pruner").Inc()
+						} else {
+							// Get job info to log records processed
+							recordsProcessed := int64(0)
+							if job := s.prunerService.GetJobByHeight(latestHeight); job != nil {
+								recordsProcessed = job.RecordsProcessed.Load()
+							}
+							s.logger.Infof("Pruner for height %d completed successfully, pruned %d records", latestHeight, recordsProcessed)
+							prunerDuration.WithLabelValues("dah_pruner").Observe(time.Since(startTime).Seconds())
+							prunerProcessed.Inc()
+						}
+					case <-timeoutTimer.C:
+						s.logger.Infof("Pruner for height %d exceeded coordinator timeout of %v - pruner continues in background", latestHeight, prunerTimeout)
+						// Note: This is not an error - the pruner job continues processing in the background.
+						// Only re-queue if channel is empty (next trigger will catch up if channel has pending jobs).
+
+						// Only re-queue if channel is empty (non-blocking check)
+						select {
+						case s.prunerCh <- latestHeight:
+							s.logger.Debugf("Re-queued pruner for height %d (channel was empty)", latestHeight)
+						default:
+							// Channel has pending jobs, no need to re-queue
+							s.logger.Debugf("Pruner channel has pending jobs, skipping re-queue")
+						}
+					case <-ctx.Done():
+						return
 					}
-				case <-ctx.Done():
-					return
 				}
 			}
 

settings.conf

@@ -583,6 +583,29 @@ pruner_grpcListenAddress.docker.host = localhost:${PORT_PREFIX}${PRUNER_GRPC_POR
 # Default: 10m
 pruner_jobTimeout = 10m
 
+# UTXO-specific pruning settings
+# Enable defensive checks before deleting UTXO transactions
+# When enabled, verifies that all spending children are mined > BlockHeightRetention blocks ago
+# Default: false
+pruner_utxoDefensiveEnabled = false
+
+# UTXO batcher settings - optimized for multi-million record pruning operations
+# These settings control batching of operations during delete-at-height (DAH) UTXO pruning
+# Larger batch sizes = fewer round-trips to storage = faster pruning
+pruner_utxoParentUpdateBatcherSize = 2000
+pruner_utxoParentUpdateBatcherDurationMillis = 100
+pruner_utxoDeleteBatcherSize = 5000
+pruner_utxoDeleteBatcherDurationMillis = 100
+
+# Maximum concurrent operations during UTXO pruning
+# Increase this if you have a large connection pool and want faster pruning
+# Default: 1 (auto-detect from connection pool)
+pruner_utxoMaxConcurrentOperations = 1
+
+# Batch size for reading child transactions during defensive UTXO pruning
+# Default: 1024
+pruner_utxoDefensiveBatchReadSize = 1024
+
 # @group: dashboard
 # Vite dev server ports (comma-separated)
 # dashboard_devServerPorts = 5173,4173
@@ -1493,19 +1516,6 @@ utxostore_storeBatcherDurationMillis = 10
 
 utxostore_storeBatcherSize = 2048
 
-# Pruner batcher settings - optimized for multi-million record pruning operations
-# These settings control batching of operations during delete-at-height (DAH) pruning
-# Larger batch sizes = fewer round-trips to Aerospike = faster pruning
-utxostore_prunerParentUpdateBatcherSize = 2000
-utxostore_prunerParentUpdateBatcherDurationMillis = 100
-utxostore_prunerDeleteBatcherSize = 5000
-utxostore_prunerDeleteBatcherDurationMillis = 100
-
-# Maximum concurrent operations during pruning (0 = use Aerospike connection queue size)
-# Increase this if you have a large connection pool and want faster pruning
-# Default: 0 (auto-detect from connection pool)
-utxostore_prunerMaxConcurrentOperations = 0
-
 # this value determines if a caching mechanism will be used for external transactions
 # if you have the memory available, it will speed up your IBD
 # and it is required for large blocks which load in the same tx multiple times, e.g. 814337

settings/interface.go

@@ -381,12 +381,6 @@ type UtxoStoreSettings struct {
 	MaxMinedBatchSize                 int
 	BlockHeightRetentionAdjustment    int32 // Adjustment to GlobalBlockHeightRetention (can be positive or negative)
 	DisableDAHCleaner                 bool  // Disable the DAH cleaner process completely
-	// Pruner-specific settings
-	PrunerParentUpdateBatcherSize           int // Batch size for parent record updates during pruning
-	PrunerParentUpdateBatcherDurationMillis int // Batch duration for parent record updates during pruning (ms)
-	PrunerDeleteBatcherSize                 int // Batch size for record deletions during pruning
-	PrunerDeleteBatcherDurationMillis       int // Batch duration for record deletions during pruning (ms)
-	PrunerMaxConcurrentOperations           int // Maximum concurrent operations during pruning (0 = use connection queue size)
 }
 
 type P2PSettings struct {
@@ -489,10 +483,15 @@ type CoinbaseSettings struct {
 }
 
 type PrunerSettings struct {
-	GRPCListenAddress string
-	GRPCAddress       string
-	WorkerCount       int
-	JobTimeout        time.Duration // Timeout for waiting for pruner job completion
+	GRPCListenAddress          string
+	GRPCAddress                string
+	WorkerCount                int           // Number of worker goroutines (default: 1)
+	JobTimeout                 time.Duration // Timeout for waiting for pruner job completion (0 = wait indefinitely)
+	UTXODefensiveEnabled       bool          // Enable defensive checks before deleting UTXO transactions (verify children are mined > BlockHeightRetention blocks ago)
+	UTXODefensiveBatchReadSize int           // Batch size for reading child transactions during defensive UTXO pruning (default: 10000)
+	UTXOChunkSize              int           // Number of records to process in each chunk before batch flushing (default: 1000)
+	UTXOChunkGroupLimit        int           // Maximum parallel chunk processing during UTXO pruning (default: 10)
+	UTXOProgressLogInterval    time.Duration // Interval for logging progress during UTXO pruning (default: 30s)
 }
 
 type SubtreeValidationSettings struct {

settings/settings.go

@@ -366,12 +366,6 @@ func NewSettings(alternativeContext ...string) *Settings {
 			MaxMinedBatchSize:                 getInt("utxostore_maxMinedBatchSize", 1024, alternativeContext...),
 			BlockHeightRetentionAdjustment:    getInt32("utxostore_blockHeightRetentionAdjustment", 0, alternativeContext...),
 			DisableDAHCleaner:                 getBool("utxostore_disableDAHCleaner", false, alternativeContext...),
-			// Pruner-specific settings optimized for multi-million record pruning operations
-			PrunerParentUpdateBatcherSize:           getInt("utxostore_prunerParentUpdateBatcherSize", 2000, alternativeContext...),
-			PrunerParentUpdateBatcherDurationMillis: getInt("utxostore_prunerParentUpdateBatcherDurationMillis", 100, alternativeContext...),
-			PrunerDeleteBatcherSize:                 getInt("utxostore_prunerDeleteBatcherSize", 5000, alternativeContext...),
-			PrunerDeleteBatcherDurationMillis:       getInt("utxostore_prunerDeleteBatcherDurationMillis", 100, alternativeContext...),
-			PrunerMaxConcurrentOperations:           getInt("utxostore_prunerMaxConcurrentOperations", 0, alternativeContext...),
 		},
 		P2P: P2PSettings{
 			BlockTopic:         getString("p2p_block_topic", "", alternativeContext...),
@@ -433,10 +427,15 @@ func NewSettings(alternativeContext ...string) *Settings {
 			DistributorTimeout:          getDuration("distributor_timeout", 30*time.Second, alternativeContext...),
 		},
 		Pruner: PrunerSettings{
-			GRPCAddress:       getString("pruner_grpcAddress", "localhost:8096", alternativeContext...),
-			GRPCListenAddress: getString("pruner_grpcListenAddress", ":8096", alternativeContext...),
-			WorkerCount:       getInt("pruner_workerCount", 4, alternativeContext...), // Default to 4 workers
-			JobTimeout:        getDuration("pruner_jobTimeout", 10*time.Minute, alternativeContext...),
+			GRPCAddress:                getString("pruner_grpcAddress", "localhost:8096", alternativeContext...),
+			GRPCListenAddress:          getString("pruner_grpcListenAddress", ":8096", alternativeContext...),
+			WorkerCount:                getInt("pruner_workerCount", 1, alternativeContext...),                               // Single job at a time
+			JobTimeout:                 getDuration("pruner_jobTimeout", 0, alternativeContext...),                           // 0 = wait indefinitely
+			UTXODefensiveEnabled:       getBool("pruner_utxoDefensiveEnabled", false, alternativeContext...),                 // Defensive mode off by default (production)
+			UTXODefensiveBatchReadSize: getInt("pruner_utxoDefensiveBatchReadSize", 10000, alternativeContext...),            // Batch size for child verification
+			UTXOChunkSize:              getInt("pruner_utxoChunkSize", 1000, alternativeContext...),                          // Chunk size for batch operations
+			UTXOChunkGroupLimit:        getInt("pruner_utxoChunkGroupLimit", 10, alternativeContext...),                      // Process 10 chunks in parallel
+			UTXOProgressLogInterval:    getDuration("pruner_utxoProgressLogInterval", 30*time.Second, alternativeContext...), // Progress every 30s
 		},
 		SubtreeValidation: SubtreeValidationSettings{
 			QuorumAbsoluteTimeout:                     getDuration("subtree_quorum_absolute_timeout", 30*time.Second, alternativeContext...),

stores/pruner/interfaces.go

@@ -15,6 +15,9 @@ type Service interface {
 	// SetPersistedHeightGetter sets the function used to get block persister progress.
 	// This allows pruner to coordinate with block persister to avoid premature deletion.
 	SetPersistedHeightGetter(getter func() uint32)
+
+	// GetJobByHeight returns a job for the specified block height
+	GetJobByHeight(blockHeight uint32) *Job
 }
 
 // PrunerServiceProvider defines an interface for stores that can provide a pruner service.

stores/pruner/job.go

@@ -38,15 +38,16 @@ func (s JobStatus) String() string {
 
 // Job represents a pruner job
 type Job struct {
-	BlockHeight uint32
-	status      atomic.Int32 // Using atomic for thread-safe access
-	Error       error
-	Created     time.Time
-	Started     time.Time
-	Ended       time.Time
-	ctx         context.Context
-	cancel      context.CancelFunc
-	DoneCh      chan string // Channel to signal when the job is complete (for testing purposes)
+	BlockHeight      uint32
+	status           atomic.Int32 // Using atomic for thread-safe access
+	RecordsProcessed atomic.Int64 // Number of records processed/pruned
+	Error            error
+	Created          time.Time
+	Started          time.Time
+	Ended            time.Time
+	ctx              context.Context
+	cancel           context.CancelFunc
+	DoneCh           chan string // Channel to signal when the job is complete (for testing purposes)
 }
 
 // GetStatus returns the current status of the job

stores/pruner/job_processor.go

@@ -286,6 +286,20 @@ func (m *JobManager) GetJobs() []*Job {
 	return clone
 }
 
+// GetJobByHeight returns a job for the specified block height
+func (m *JobManager) GetJobByHeight(blockHeight uint32) *Job {
+	m.jobsMutex.RLock()
+	defer m.jobsMutex.RUnlock()
+
+	for _, job := range m.jobs {
+		if job.BlockHeight == blockHeight {
+			return job
+		}
+	}
+
+	return nil
+}
+
 // worker processes jobs from the queue
 func (m *JobManager) worker(workerID int) {
 	m.logger.Debugf("[JobManager] Worker %d started", workerID)