[CI] SnapshotStressTestsIT testRandomActivities failing

[DaveCTurner]

Found when investigating #108907 but looks like a different failure - a shard ends up being marked as PAUSED_FOR_NODE_REMOVAL in two separate snapshots:

java.lang.AssertionError: Tuple[v1=index-1, v2=1] is assigned twice in [{"repository":"repo-0","snapshot":"snapshot-partial-5","uuid":"cdvmwF6sTkm7j0fVy4fA7g","include_global_state":true,"partial":true,"state":"SUCCESS","indices":[{"name":"index-3","id":"iiD7ALDGTUKF0pUx7ryBNA"},{"name":"index-1","id":"iR2oYK6JQCOk7VjrDnQaOA"}],"start_time_millis":17169382...
        at __randomizedtesting.SeedInfo.seed([4899850D0D379B5B]:0)
        at org.elasticsearch.cluster.SnapshotsInProgress.assertShardStateConsistent(SnapshotsInProgress.java:421)
        at org.elasticsearch.cluster.SnapshotsInProgress.assertConsistentEntries(SnapshotsInProgress.java:390)
        at org.elasticsearch.cluster.SnapshotsInProgress.<init>(SnapshotsInProgress.java:119)
        at org.elasticsearch.cluster.SnapshotsInProgress.withUpdatedEntriesForRepo(SnapshotsInProgress.java:135)
        at org.elasticsearch.snapshots.SnapshotsService$SnapshotShardsUpdateContext.computeUpdatedState(SnapshotsService.java:3213)
        at org.elasticsearch.snapshots.SnapshotsService$SnapshotTaskExecutor.execute(SnapshotsService.java:3935)
        at org.elasticsearch.cluster.service.MasterService.innerExecuteTasks(MasterService.java:1084)
        at org.elasticsearch.cluster.service.MasterService.executeTasks(MasterService.java:1049)
        at org.elasticsearch.cluster.service.MasterService.executeAndPublishBatch(MasterService.java:244)
        at org.elasticsearch.cluster.service.MasterService$BatchingTaskQueue$Processor.lambda$run$2(MasterService.java:1690)
        at org.elasticsearch.action.ActionListener.run(ActionListener.java:433)
        at org.elasticsearch.cluster.service.MasterService$BatchingTaskQueue$Processor.run(MasterService.java:1687)
        at org.elasticsearch.cluster.service.MasterService$5.lambda$doRun$0(MasterService.java:1282)
        at org.elasticsearch.action.ActionListener.run(ActionListener.java:433)
        at org.elasticsearch.cluster.service.MasterService$5.doRun(MasterService.java:1261)
        at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:984)
        at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
        at java.base/java.lang.Thread.run(Thread.java:1570)

Copious logs:

testoutput-2024-05-28T23:17:03.715Z-fail.tar.gz

[elasticsearchmachine]

Pinging @elastic/es-distributed (Team:Distributed)

[DaveCTurner]

Repro:

diff --git a/server/src/test/java/org/elasticsearch/snapshots/SnapshotsServiceTests.java b/server/src/test/java/org/elasticsearch/snapshots/SnapshotsServiceTests.java
index 56a28b11edf..bcc7a23bbec 100644
--- a/server/src/test/java/org/elasticsearch/snapshots/SnapshotsServiceTests.java
+++ b/server/src/test/java/org/elasticsearch/snapshots/SnapshotsServiceTests.java
@@ -401,6 +401,70 @@ public class SnapshotsServiceTests extends ESTestCase {
         assertIsNoop(updatedClusterState, completeShardClone);
     }

+    public void testPauseForNodeRemovalWithQueuedShards() throws Exception {
+        final var repoName = "test-repo";
+        final var snapshot1 = snapshot(repoName, "snap-1");
+        final var snapshot2 = snapshot(repoName, "snap-2");
+        final var indexName = "index-1";
+        final var shardId = new ShardId(index(indexName), 0);
+        final var repositoryShardId = new RepositoryShardId(indexId(indexName), 0);
+        final var nodeId = uuid();
+
+        final var runningEntry = snapshotEntry(
+            snapshot1,
+            Collections.singletonMap(indexName, repositoryShardId.index()),
+            Map.of(shardId, initShardStatus(nodeId))
+        );
+
+        final var queuedEntry = snapshotEntry(
+            snapshot2,
+            Collections.singletonMap(indexName, repositoryShardId.index()),
+            Map.of(shardId, SnapshotsInProgress.ShardSnapshotStatus.UNASSIGNED_QUEUED)
+        );
+
+        final var initialState = stateWithSnapshots(
+            ClusterState.builder(ClusterState.EMPTY_STATE)
+                .nodes(DiscoveryNodes.builder().add(DiscoveryNodeUtils.create(nodeId)).localNodeId(nodeId).masterNodeId(nodeId).build())
+                .routingTable(
+                    RoutingTable.builder()
+                        .add(
+                            IndexRoutingTable.builder(shardId.getIndex())
+                                .addShard(TestShardRouting.newShardRouting(shardId, nodeId, true, ShardRoutingState.STARTED))
+                        )
+                        .build()
+                )
+                .build(),
+            repoName,
+            runningEntry,
+            queuedEntry
+        );
+
+        final var updatedState = applyUpdates(
+            initialState,
+            new SnapshotsService.ShardSnapshotUpdate(
+                snapshot1,
+                shardId,
+                null,
+                new SnapshotsInProgress.ShardSnapshotStatus(
+                    nodeId,
+                    SnapshotsInProgress.ShardState.PAUSED_FOR_NODE_REMOVAL,
+                    runningEntry.shards().get(shardId).generation()
+                ),
+                ActionTestUtils.assertNoFailureListener(t -> {})
+            )
+        );
+
+        assertEquals(
+            SnapshotsInProgress.ShardState.PAUSED_FOR_NODE_REMOVAL,
+            SnapshotsInProgress.get(updatedState).snapshot(snapshot1).shards().get(shardId).state()
+        );
+
+        assertEquals(
+            SnapshotsInProgress.ShardState.QUEUED,
+            SnapshotsInProgress.get(updatedState).snapshot(snapshot2).shards().get(shardId).state()
+        );
+    }
+
     public void testSnapshottingIndicesExcludesClones() {
         final String repoName = "test-repo";
         final String indexName = "index";