sql: add safety gates for pausable portal cleanup and flow invalidation

Epic: None
Informs: cockroachlabs/support#3463

When a pausable portal encounters an error during execution, two issues
can lead to panics on subsequent resume attempts:

1. The underlying FlowBase gets reset to nil during cleanup, but the
 portal's flow reference remains non-nil, causing hasFlowForPausablePortal()
 to incorrectly return true.

2. Errored portals are not removed from the portal map because deletion
 only occurs when execStmt() returns a non-nil fsm.Event.

This change adds two defensive checks:
- Nil the whole flow object hanging off the portalInfo while cleaning up the flow.
- Ensure errored portals are properly cleaned up regardless of event state

These gates prevent nil pointer dereferences when resuming portals that
have been partially cleaned up due to errors.

Release note: None

Author: ZhouXing19

pkg/sql/conn_executor.go

@@ -2598,8 +2598,6 @@ func (ex *connExecutor) execCmd() (retErr error) {
 		panic(errors.AssertionFailedf("unsupported command type: %T", cmd))
 	}
 
-	var advInfo advanceInfo
-
 	// We close all pausable portals and cursors when we encounter err payload,
 	// otherwise there will be leftover bytes.
 	shouldClosePausablePortalsAndCursors := func(payload fsm.EventPayload) bool {
@@ -2611,18 +2609,19 @@ func (ex *connExecutor) execCmd() (retErr error) {
 		}
 	}
 
+	if shouldClosePausablePortalsAndCursors(payload) {
+		// We need this as otherwise, there'll be leftover bytes when
+		// txnState.finishSQLTxn() is being called, as the underlying resources of
+		// pausable portals hasn't been cleared yet.
+		ex.extraTxnState.prepStmtsNamespace.closeAllPausablePortals(ctx, &ex.extraTxnState.prepStmtsNamespaceMemAcc)
+		if err := ex.extraTxnState.sqlCursors.closeAll(&ex.planner, cursorCloseForTxnRollback); err != nil {
+			log.Dev.Warningf(ctx, "error closing cursors: %v", err)
+		}
+	}
+
+	var advInfo advanceInfo
 	// If an event was generated, feed it to the state machine.
 	if ev != nil {
-		var err error
-		if shouldClosePausablePortalsAndCursors(payload) {
-			// We need this as otherwise, there'll be leftover bytes when
-			// txnState.finishSQLTxn() is being called, as the underlying resources of
-			// pausable portals hasn't been cleared yet.
-			ex.extraTxnState.prepStmtsNamespace.closeAllPausablePortals(ctx, &ex.extraTxnState.prepStmtsNamespaceMemAcc)
-			if err := ex.extraTxnState.sqlCursors.closeAll(&ex.planner, cursorCloseForTxnRollback); err != nil {
-				log.Dev.Warningf(ctx, "error closing cursors: %v", err)
-			}
-		}
 		advInfo, err = ex.txnStateTransitionsApplyWrapper(ev, payload, res, pos)
 		if err != nil {
 			return err

pkg/sql/distsql_running.go

@@ -1979,7 +1979,10 @@ func (dsp *DistSQLPlanner) PlanAndRunAll(
 		}
 		if !p.resumableFlow.cleanup.isComplete {
 			p.resumableFlow.cleanup.appendFunc(func(ctx context.Context) {
-				p.resumableFlow.flow.Cleanup(ctx)
+				if p.resumableFlow.flow != nil {
+					p.resumableFlow.flow.Cleanup(ctx)
+					p.resumableFlow.flow = nil
+				}
 			})
 		}
 	}