Merge #155669

155669: sql: add safety gates for pausable portal cleanup and flow invalidation r=ZhouXing19 a=ZhouXing19

Informs: https://github.com/cockroachlabs/support/issues/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

Co-authored-by: ZhouXing19 <zhouxing@uchicago.edu>

Author: craig[bot]

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

@@ -2019,7 +2019,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
+				}
 			})
 		}
 	}