executor: IndexJoin OOM panic during inner result collection leaks the inner executor and worker

./tools/check/failpoint-go-test.sh pkg/executor/test/distsqltest \
  -run '^TestIndexJoinOOMInnerOOMLifecycle$' -count=3 -v -timeout=120s

IndexJoinOOM_CONTROL=normal ./tools/check/failpoint-go-test.sh \
  pkg/executor/test/distsqltest \
  -run '^TestIndexJoinOOMInnerOOMLifecycle$' -count=1 -v -timeout=120s

diff --git a/pkg/executor/distsql.go b/pkg/executor/distsql.go
index 4c04323332..4ae58b65ce 100644
--- a/pkg/executor/distsql.go
+++ b/pkg/executor/distsql.go
@@ -29,6 +29,7 @@ import (
 	"unsafe"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/distsql"
 	distsqlctx "github.com/pingcap/tidb/pkg/distsql/context"
 	"github.com/pingcap/tidb/pkg/executor/internal/builder"
@@ -734,6 +735,7 @@ func (e *IndexLookUpExecutor) startWorkers(ctx context.Context, initBatchSize in
 		return err
 	}
 	e.workerStarted = true
+	failpoint.InjectCall("indexJoinOOMLookupStarted", e)
 	return nil
 }
 
@@ -1213,6 +1215,7 @@ func (e *IndexLookUpExecutor) Close() error {
 	e.finished = nil
 	e.workerStarted = false
 	e.resultCurr = nil
+	failpoint.InjectCall("indexJoinOOMLookupClosed", e)
 	return nil
 }
 
diff --git a/pkg/executor/join/index_lookup_join.go b/pkg/executor/join/index_lookup_join.go
index e77bbc0e28..d0a8cba931 100644
--- a/pkg/executor/join/index_lookup_join.go
+++ b/pkg/executor/join/index_lookup_join.go
@@ -779,6 +779,7 @@ func (iw *innerWorker) fetchInnerResults(ctx context.Context, task *lookUpJoinTa
 			needClose = true
 			break
 		}
+		failpoint.InjectCall("indexJoinOOMBeforeInnerAdd", task.innerResult)
 		task.innerResult.Add(executorChk)
 		// If maxFetchSize is set, we need to break the loop when the fetched rows reach the max.
 		if iw.maxFetchSize > 0 && task.innerResult.Len() >= iw.maxFetchSize {
diff --git a/pkg/executor/test/distsqltest/BUILD.bazel b/pkg/executor/test/distsqltest/BUILD.bazel
index 2cf87858bd..e57bb6edf6 100644
--- a/pkg/executor/test/distsqltest/BUILD.bazel
+++ b/pkg/executor/test/distsqltest/BUILD.bazel
@@ -6,17 +6,23 @@ go_test(
     srcs = [
         "distsql_test.go",
         "main_test.go",
+        "indexJoinOOM_repro_test.go",
     ],
     flaky = True,
     race = "on",
-    shard_count = 2,
+    shard_count = 3,
     deps = [
         "//pkg/config",
         "//pkg/config/kerneltype",
+        "//pkg/executor",
         "//pkg/kv",
         "//pkg/meta/autoid",
         "//pkg/sessionctx/vardef",
         "//pkg/testkit",
+        "//pkg/util/chunk",
+        "//pkg/util/memory",
+        "//pkg/util/sqlkiller",
+        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//tikv",
         "@org_uber_go_goleak//:goleak",
diff --git a/pkg/executor/test/distsqltest/indexJoinOOM_repro_test.go b/pkg/executor/test/distsqltest/indexJoinOOM_repro_test.go
new file mode 100644
index 0000000000..889dc987cd
--- /dev/null
+++ b/pkg/executor/test/distsqltest/indexJoinOOM_repro_test.go
@@ -0,0 +1,107 @@
+package distsql_test
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"runtime/pprof"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pingcap/failpoint"
+	"github.com/pingcap/tidb/pkg/executor"
+	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/util/chunk"
+	"github.com/pingcap/tidb/pkg/util/memory"
+	"github.com/pingcap/tidb/pkg/util/sqlkiller"
+	"github.com/stretchr/testify/require"
+)
+
+// Diagnostic reproduction only: lower the real result-list tracker quota at a
+// precise allocation boundary. No mutex or StatementContext fields are modified.
+func TestIndexJoinOOMInnerOOMLifecycle(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table a(id int primary key, customerid int, messageid int)")
+	tk.MustExec("create table m(id int primary key nonclustered, customerid int, body varchar(256), key ic(customerid))")
+	tk.MustExec("insert into a values(1,1,10),(2,1,20),(3,1,30)")
+	values := make([]string, 4096)
+	for i := range values {
+		values[i] = fmt.Sprintf("(%d,1,'%s')", i+1, strings.Repeat("x", 128))
+	}
+	tk.MustExec("insert into m values " + strings.Join(values, ","))
+	tk.MustExec("set tidb_index_lookup_size=32")
+	tk.MustExec("set tidb_index_lookup_concurrency=4")
+	tk.MustExec("set tidb_index_lookup_join_concurrency=1")
+	var mu sync.Mutex
+	active := map[*executor.IndexLookUpExecutor]bool{}
+	started := "github.com/pingcap/tidb/pkg/executor/indexJoinOOMLookupStarted"
+	closed := "github.com/pingcap/tidb/pkg/executor/indexJoinOOMLookupClosed"
+	inject := "github.com/pingcap/tidb/pkg/executor/join/indexJoinOOMBeforeInnerAdd"
+	require.NoError(t, failpoint.EnableCall(started, func(e *executor.IndexLookUpExecutor) { mu.Lock(); active[e] = true; mu.Unlock() }))
+	require.NoError(t, failpoint.EnableCall(closed, func(e *executor.IndexLookUpExecutor) { mu.Lock(); delete(active, e); mu.Unlock() }))
+	defer func() {
+		_ = failpoint.Disable(inject)
+		mu.Lock()
+		es := make([]*executor.IndexLookUpExecutor, 0, len(active))
+		for e := range active {
+			es = append(es, e)
+		}
+		mu.Unlock()
+		for _, e := range es {
+			_ = e.Close()
+		}
+		_ = failpoint.Disable(started)
+		_ = failpoint.Disable(closed)
+	}()
+	var once sync.Once
+	require.NoError(t, failpoint.EnableCall(inject, func(l *chunk.List) {
+		if os.Getenv("IndexJoinOOM_CONTROL") == "normal" {
+			return
+		}
+		once.Do(func() {
+			// Let prefetch fill resultCh while the consumer is at the allocation boundary.
+			time.Sleep(100 * time.Millisecond)
+			tr := l.GetMemTracker()
+			tr.SetActionOnExceed(&memory.PanicOnExceed{Killer: &sqlkiller.SQLKiller{}})
+			tr.SetBytesLimit(1)
+		})
+	}))
+	ctx := context.Background()
+	rs, err := tk.Exec("select /*+ INL_JOIN(m) */ a.id,m.body from a join m use index(ic) on a.customerid=m.customerid and a.messageid=m.id")
+	require.NoError(t, err)
+	sc := tk.Session().GetSessionVars().StmtCtx
+	chk := rs.NewChunk(nil)
+	for {
+		err = rs.Next(ctx, chk)
+		if err != nil || chk.NumRows() == 0 {
+			break
+		}
+		chk.Reset()
+	}
+	t.Logf("Next error: %v", err)
+	if os.Getenv("IndexJoinOOM_CONTROL") == "normal" {
+		require.NoError(t, err)
+	} else {
+		require.Error(t, err)
+	}
+	require.NoError(t, rs.Close())
+	_ = failpoint.Disable(inject)
+	mu.Lock()
+	leaked := len(active)
+	mu.Unlock()
+	t.Logf("inner IndexLookup executors not closed after statement Close: %d", leaked)
+	var stacks bytes.Buffer
+	require.NoError(t, pprof.Lookup("goroutine").WriteTo(&stacks, 2))
+	require.NoError(t, os.WriteFile(t.TempDir()+"/goroutines.txt", stacks.Bytes(), 0600))
+	t.Logf("IndexLookup producer stack still present: %v", strings.Contains(stacks.String(), "(*indexWorker).buildAndDispatchLookupTasks"))
+	tk.MustQuery("select 1").Check(testkit.Rows("1"))
+	tk.MustQuery("select 2").Check(testkit.Rows("2"))
+	reused := sc == tk.Session().GetSessionVars().StmtCtx
+	t.Logf("original StatementContext reused after two subsequent statements: %v", reused)
+	require.Zero(t, leaked, "inner executor survived query error and result-set Close")
+}

flowchart TB
    A["Execute IndexJoin"] --> B["OOM while collecting inner results"]
    B --> C["Query returns error 8175"]
    C --> D["RecordSet.Close returns successfully"]
    D --> E["Expected: inner executor closed<br/>Background workers exited"]
    D --> F["Observed: 1 inner executor remains unclosed<br/>Old worker blocked on channel send"]
    F --> G["Same session executes SELECT 1 and SELECT 2"]
    G --> H["Original StatementContext is reused<br/>Old worker has not been cleaned up"]

    style E fill:#dcfce7,stroke:#16a34a
    style F fill:#fee2e2,stroke:#dc2626,stroke-width:3px
    style H fill:#fee2e2,stroke:#dc2626,stroke-width:3px

goroutine 10516 [chan send]:
  (*indexWorker).buildAndDispatchLookupTasks
  distsql.go:1625

Next error: [executor:8175]Your query has been cancelled due to exceeding the allowed memory limit ...
inner IndexLookup executors not closed after statement Close: 1
IndexLookup producer stack still present: true
original StatementContext reused after two subsequent statements: true
Error: Should be zero, but was 1
Messages: inner executor survived query error and result-set Close