Assert aborted snapshots have no pending shard work (#102621)

When a snapshot is aborted, all shard snapshots must either be complete
(`SUCCESS`, `FAILED` or `MISSING`) or else they must be in state
`ABORTED` while the data node finishes its work. It seems we do not
check this invariant anywhere today, so this commit adds an assertion of
this invariant.

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -343,14 +343,20 @@ public class SnapshotsInProgress extends AbstractNamedDiffable<Custom> implement
                 assert entry.repository().equals(repository) : "mismatched repository " + entry + " tracked under " + repository;
                 for (Map.Entry<RepositoryShardId, ShardSnapshotStatus> shard : entry.shardsByRepoShardId().entrySet()) {
                     final RepositoryShardId sid = shard.getKey();
+                    final ShardSnapshotStatus shardSnapshotStatus = shard.getValue();
                     assert assertShardStateConsistent(
                         entriesForRepository,
                         assignedShards,
                         queuedShards,
                         sid.indexName(),
                         sid.shardId(),
-                        shard.getValue()
+                        shardSnapshotStatus
                     );
+
+                    assert entry.state() != State.ABORTED
+                        || shardSnapshotStatus.state == ShardState.ABORTED
+                        || shardSnapshotStatus.state().completed()
+                        : sid + " is in state " + shardSnapshotStatus.state() + " in aborted snapshot " + entry.snapshot;
                 }
             }
             // make sure in-flight-shard-states can be built cleanly for the entries without tripping assertions

server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java

@@ -523,8 +523,15 @@ public class SnapshotsInProgressSerializationTests extends SimpleDiffableWireSer
     }
 
     public static State randomState(Map<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards) {
-        return SnapshotsInProgress.completed(shards.values())
-            ? randomFrom(State.SUCCESS, State.FAILED)
-            : randomFrom(State.STARTED, State.INIT, State.ABORTED);
+        if (SnapshotsInProgress.completed(shards.values())) {
+            return randomFrom(State.SUCCESS, State.FAILED);
+        }
+        if (shards.values()
+            .stream()
+            .map(SnapshotsInProgress.ShardSnapshotStatus::state)
+            .allMatch(st -> st.completed() || st == ShardState.ABORTED)) {
+            return State.ABORTED;
+        }
+        return randomFrom(State.STARTED, State.INIT);
     }
 }