fix(blockstm): wake up suspended executors on cancel

CHANGELOG.md

@@ -81,6 +81,7 @@ Ref: https://keepachangelog.com/en/1.0.0/
 
 ### Bug Fixes
 
+* (blockstm) [#25789](https://github.com/cosmos/cosmos-sdk/issues/25789) Wake up suspended executors when scheduler doesn't complete to prevent goroutine leaks.
 * (x/staking) [#25649](https://github.com/cosmos/cosmos-sdk/pull/25649) Add missing `defer iterator.Close()` calls in `IterateDelegatorRedelegations` and `GetRedelegations` to prevent resource leaks.
 * (mempool) [#25563](https://github.com/cosmos/cosmos-sdk/pull/25563) Cleanup sender indices in case of tx replacement.
 * (x/epochs) [#25425](https://github.com/cosmos/cosmos-sdk/pull/25425) Fix `InvokeSetHooks` being called with a nil keeper and `AppModule` containing a copy instead of a pointer (hooks set post creating the `AppModule` like with depinject didn't apply because it's a different instance).

blockstm/scheduler.go

@@ -213,3 +213,11 @@ func (s *Scheduler) Stats() string {
 	return fmt.Sprintf("executed: %d, validated: %d",
 		s.executedTxns.Load(), s.validatedTxns.Load())
 }
+
+// CancelAll wakes up all suspended executors.
+// Called during context cancellation to prevent hanging.
+func (s *Scheduler) CancelAll() {
+	for i := range s.txnStatus {
+		s.txnStatus[i].TryCancel()
+	}
+}

blockstm/status.go

@@ -112,3 +112,14 @@ func (s *StatusEntry) Suspend(cond *Condvar) {
 
 	s.Unlock()
 }
+
+// TryCancel wakes up a suspended executor if it's suspended.
+// Called during context cancellation to prevent hanging.
+func (s *StatusEntry) TryCancel() {
+	s.Lock()
+	if s.status == StatusSuspended && s.cond != nil {
+		s.cond.Notify()
+		s.cond = nil
+	}
+	s.Unlock()
+}

blockstm/stm.go

@@ -59,8 +59,10 @@ func ExecuteBlockWithEstimates(
 	}
 
 	if !scheduler.Done() {
-		if ctx.Err() != nil {
-			// canceled
+		err := ctx.Err()
+		// canceled, wake up all suspended executors to prevent hanging
+		scheduler.CancelAll()
+		if err != nil {
 			return ctx.Err()
 		}
 

blockstm/txnrunner.go

@@ -72,7 +72,7 @@ func (e STMRunner) Run(ctx context.Context, ms storetypes.MultiStore, txs [][]by
 		memTxs, estimates = preEstimates(txs, e.workers, authStore, bankStore, e.coinDenom(ms), e.txDecoder)
 	}
 
-	if err := ExecuteBlockWithEstimates(
+	err := ExecuteBlockWithEstimates(
 		ctx,
 		blockSize,
 		index,
@@ -99,7 +99,18 @@ func (e STMRunner) Run(ctx context.Context, ms storetypes.MultiStore, txs [][]by
 				incarnationCache[txn].Store(v)
 			}
 		},
-	); err != nil {
+	)
+
+	// fallback to sequential execution if parallel execution failed with scheduler error
+	if err != nil && err.Error() == "scheduler did not complete" {
+		results = make([]*abci.ExecTxResult, blockSize)
+		for i := range txs {
+			results[i] = deliverTx(txs[i], ms, i, nil)
+		}
+		return results, nil
+	}
+
+	if err != nil {
 		return nil, err
 	}