[Enhancement] Reject agent-task enqueue on a demoted (non-leader) node

The addTask() demotion guard only checked isLeaderDemoting(), leaving a window AFTER
the demotion completed: a straggling leader-session thread (e.g. a user DDL that
passed its admission checks before the demotion began) could still enqueue on the
now-follower node. The waiter would then sit out its full timeout, and the stale
queue entry would survive into a later re-election where it shadows a same-signature
task (addTask de-dups by signature, so BE finish reports resolve to the stale object
and the new waiter never completes).

Extend the guard to also reject when feType is FOLLOWER / OBSERVER / UNKNOWN (the
demotion target roles). Verified every legitimate enqueue happens with feType ==
LEADER: activation flips feType to LEADER before the leader-only daemons start, DDL
threads pass leader admission first, and the replay paths never enqueue (e.g.
OlapTable.onDrop guards sendDropAutoIncrementMapTask with !replay). INIT stays
admitted so bootstrap, checkpoint-image GlobalStateMgr instances, and plain unit
tests are unaffected; a fully mocked GlobalStateMgr (null feType) is also admitted.

Add AgentTaskTest.testAddTaskRejectedAfterDemotionCompleted.

Signed-off-by: gengjun-git <gengjun@starrocks.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: gengjun-git

fe/fe-core/src/main/java/com/starrocks/task/AgentTaskQueue.java

@@ -41,6 +41,7 @@
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Table;
 import com.starrocks.common.Status;
+import com.starrocks.ha.FrontendNodeType;
 import com.starrocks.memory.estimate.Estimator;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.thrift.TPushType;
@@ -76,13 +77,25 @@ public static synchronized void addTaskList(List<AgentTask> taskList) {
     }
 
     public static synchronized boolean addTask(AgentTask task) {
-        // Source guard for leader demotion: once demoting, refuse to enqueue new agent tasks.
-        // Together with abandonInFlightTasks() (which drains what is already queued) this closes the
-        // TOCTOU where a create-tablet started just after the drain would otherwise wait out its
-        // timeout. The throw fails the doomed operation fast (its journal write would be fenced anyway).
-        if (GlobalStateMgr.getCurrentState().isLeaderDemoting()) {
+        // Source guard for leader demotion: refuse to enqueue new agent tasks once this node is
+        // demoting OR has already finished demoting to a non-leader role. Together with
+        // abandonInFlightTasks() (which drains what is already queued) this closes both windows
+        // where a straggling leader-session thread (e.g. a user DDL that passed its admission
+        // checks before the demotion began) would otherwise enqueue after the drain and wait out
+        // its full timeout, leaving a stale entry in a non-leader's queue that could shadow a
+        // same-signature task after re-election. The throw fails the doomed operation fast (its
+        // journal write would be fenced anyway). Every legitimate enqueue happens with
+        // feType == LEADER (activation flips feType to LEADER before leader-only daemons start,
+        // and the replay paths never enqueue); INIT stays admitted so bootstrap, checkpoint-image
+        // GlobalStateMgr instances, and plain unit tests are unaffected.
+        GlobalStateMgr globalStateMgr = GlobalStateMgr.getCurrentState();
+        FrontendNodeType feType = globalStateMgr.getFeType();
+        if (globalStateMgr.isLeaderDemoting()
+                || feType == FrontendNodeType.FOLLOWER
+                || feType == FrontendNodeType.OBSERVER
+                || feType == FrontendNodeType.UNKNOWN) {
             throw new IllegalStateException(
-                    "leader is demoting, refuse to enqueue agent task: " + task);
+                    "node is demoting or not the leader (" + feType + "), refuse to enqueue agent task: " + task);
         }
         long backendId = task.getBackendId();
         TTaskType type = task.getTaskType();

fe/fe-core/src/test/java/com/starrocks/task/AgentTaskTest.java

@@ -529,6 +529,23 @@ public boolean isLeaderDemoting() {
         Assertions.assertEquals(0, AgentTaskQueue.getTaskNum());
     }
 
+    @Test
+    public void testAddTaskRejectedAfterDemotionCompleted() {
+        AgentTaskQueue.clearAllTasks();
+        new MockUp<GlobalStateMgr>() {
+            @Mock
+            public com.starrocks.ha.FrontendNodeType getFeType() {
+                return com.starrocks.ha.FrontendNodeType.FOLLOWER;
+            }
+        };
+        // A straggling leader-session thread (e.g. a DDL that passed admission before the
+        // demotion) must not enqueue AFTER the demotion completed either - otherwise it waits
+        // out its full timeout and leaves a stale entry in the follower's queue that could
+        // shadow a same-signature task after re-election.
+        Assertions.assertThrows(IllegalStateException.class, () -> AgentTaskQueue.addTask(createReplicaTask));
+        Assertions.assertEquals(0, AgentTaskQueue.getTaskNum());
+    }
+
     @Test
     public void testCancelPendingWaiterReleasesLatchOfEveryLatchHolder() {
         Status demoting = new Status(TStatusCode.CANCELLED, "leader is demoting");