Method OnWorkerCrash

controlplane/session_mgr.go:848–913

OnWorkerCrash handles a worker crash by marking all affected executors as dead and notifying sessions. Executors are marked dead BEFORE the shared gRPC client is closed to prevent nil-pointer panics from concurrent RPCs. errorFn is called for each affected session to send an error to the client.

(workerID int, errorFn func(pid int32))

func (sm *SessionManager) OnWorkerCrash(workerID int, errorFn func(pid int32)) {
	sm.mu.Lock()
	pids := make([]int32, len(sm.byWorker[workerID]))
	copy(pids, sm.byWorker[workerID])

	// Mark all executors as dead first (under lock) so any concurrent RPC
	// sees the dead flag before the gRPC client is closed.
	for _, pid := range pids {
		if s, ok := sm.sessions[pid]; ok && s.Executor != nil {
			s.Executor.MarkDead()
		}
	}
	sm.mu.Unlock()

	sm.log.Warn("Worker crashed, notifying sessions.", "worker", workerID, "sessions", len(pids), "pids", pids)

	for _, pid := range pids {
		cleanupStart := time.Now()
		errorFn(pid)
		var executor *flightclient.FlightExecutor
		var connCloser io.Closer
		var session *ManagedSession
		var finishCleanup func()
		sm.mu.Lock()
		detached, remainingSessions, _, ok := sm.detachSessionLocked(pid)
		if ok {
			session = detached
			executor = detached.Executor
			connCloser = detached.connCloser
			finishCleanup = sm.lifecycle.beginCleanup()
		}
		sm.mu.Unlock()
		if executor != nil {
			_ = executor.Close()
		}
		// Close the TCP connection to unblock the message loop's read.
		// This causes the session goroutine to exit instead of looping
		// with ErrWorkerDead on every query. The deferred close in
		// handleConnection will also call Close() on the same conn;
		// that's harmless (net.Conn.Close on a closed socket returns
		// an error which is discarded).
		if connCloser != nil {
			_ = connCloser.Close()
		}
		sm.releaseSessionLease(session, "pid", pid)
		sm.log.Info("Worker crash session cleanup completed.",
			"pid", pid,
			"worker", workerID,
			"session_found", ok,
			"duration", time.Since(cleanupStart),
			"remaining_sessions", remainingSessions,
		)
		if ok {
			finishCleanup()
		}
	}

	sm.mu.Lock()
	delete(sm.byWorker, workerID)
	sm.mu.Unlock()