Add shard explain info to ReactiveReason about unassigned shards (#88590)

Add two new fields into the reactive autoscaling policy decision response.

 *   `unassigned_node_decisions` - can_allocate and can_remain decisions for unassigned shards (limited by 5)
 *   `assigned_node_decisions` - can_allocate and can_remain decisions decisions for assigned shards (limited by 5)

```
"unassigned_node_decisions": {
    "[MCVEOGORUQ][0]": {
      "can_allocate_decisions": [
        {
          "node_id": "node1",
          "node_name": "",
          "transport_address": "0.0.0.0:1",
          "deciders": [
            {
              "decider": "no_label",
              "decision": "NO",
              "explanation": "No space to allocate"
            }
          ]
        }
      ],
      "can_remain_decision": null
    }
  },
  "assigned_node_decisions": {
    "[MCVEOGORUQ][0]": {
      "can_allocate_decisions": [],
      "can_remain_decision": {
        "node_id": "node1",
        "node_name": "",
        "transport_address": "0.0.0.0:1",
        "deciders": [
          {
            "decider": "multi_throttle",
            "decision": "THROTTLE",
            "explanation": "is not active yet"
          },
          {
            "decider": "multi_no",
            "decision": "NO",
            "explanation": "No multi decision"
          },
          {
            "decider": "multi_yes",
            "decision": "YES",
            "explanation": "Yes multi decision"
          }
        ]
      }
    }
  }
```

docs/changelog/88590.yaml

@@ -0,0 +1,6 @@
+pr: 88590
+summary: Add shard explain info to `ReactiveReason` about unassigned shards
+area: Autoscaling
+type: enhancement
+issues:
+ - 85243

x-pack/plugin/autoscaling/src/internalClusterTest/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageIT.java

@@ -17,6 +17,7 @@ import org.elasticsearch.cluster.TestShardRoutingRoleStrategies;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
 import org.elasticsearch.cluster.routing.allocation.DataTier;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
@@ -39,6 +40,8 @@ import java.util.stream.IntStream;
 
 import static org.elasticsearch.index.store.Store.INDEX_STORE_STATS_REFRESH_INTERVAL_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.xpack.autoscaling.storage.ReactiveStorageDeciderService.AllocationState.MAX_AMOUNT_OF_SHARD_DECISIONS;
+import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
@@ -105,6 +108,30 @@ public class ReactiveStorageIT extends AutoscalingStorageIntegTestCase {
             response.results().get(policyName).requiredCapacity().node().storage().getBytes(),
             equalTo(maxShardSize + ReactiveStorageDeciderService.NODE_DISK_OVERHEAD + LOW_WATERMARK_BYTES)
         );
+        var reactiveReason = (ReactiveStorageDeciderService.ReactiveReason) response.results()
+            .get(policyName)
+            .results()
+            .get("reactive_storage")
+            .reason();
+        assertEquals(
+            reactiveReason.assignedShardIds().stream().limit(MAX_AMOUNT_OF_SHARD_DECISIONS).collect(Collectors.toSet()),
+            reactiveReason.assignedNodeDecisions().keySet()
+        );
+        NodeDecision canRemainNodeDecision = reactiveReason.assignedNodeDecisions()
+            .get(reactiveReason.assignedShardIds().first())
+            .canRemainDecision();
+        Decision decision = canRemainNodeDecision.decision()
+            .getDecisions()
+            .stream()
+            .filter(d -> d.type() == Decision.Type.NO)
+            .findFirst()
+            .orElseThrow(() -> new IllegalStateException("Unable to find NO can_remain decision"));
+        assertEquals(Decision.Type.NO, decision.type());
+        assertEquals("disk_threshold", decision.label());
+        assertThat(
+            decision.getExplanation(),
+            startsWith("the shard cannot remain on this node because it is above the high watermark cluster setting")
+        );
     }
 
     public void testScaleFromEmptyWarmMove() throws Exception {

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/NodeDecision.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.elasticsearch.cluster.routing.allocation.AbstractAllocationDecision.discoveryNodeToXContent;
+
+class NodeDecision implements ToXContentObject, Writeable {
+
+    private final DiscoveryNode node;
+    private final Decision decision;
+
+    NodeDecision(DiscoveryNode node, Decision decision) {
+        this.node = node;
+        this.decision = decision;
+    }
+
+    NodeDecision(StreamInput in) throws IOException {
+        node = new DiscoveryNode(in);
+        decision = Decision.readFrom(in);
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        node.writeTo(out);
+        decision.writeTo(out);
+    }
+
+    DiscoveryNode node() {
+        return node;
+    }
+
+    Decision decision() {
+        return decision;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        return o instanceof NodeDecision that && Objects.equals(node, that.node) && Objects.equals(decision, that.decision);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(node, decision);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        discoveryNodeToXContent(node, false, builder);
+        {
+            builder.startArray("deciders");
+            decision.toXContent(builder, params);
+            builder.endArray();
+        }
+        builder.endObject();
+        return builder;
+    }
+
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/NodeDecisions.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+class NodeDecisions implements ToXContentObject, Writeable {
+
+    private final List<NodeDecision> canAllocateDecisions;
+    @Nullable
+    private final NodeDecision canRemainDecision;
+
+    NodeDecisions(List<NodeDecision> canAllocateDecisions, @Nullable NodeDecision canRemainDecision) {
+        this.canAllocateDecisions = canAllocateDecisions;
+        this.canRemainDecision = canRemainDecision;
+    }
+
+    NodeDecisions(StreamInput in) throws IOException {
+        canAllocateDecisions = in.readList(NodeDecision::new);
+        canRemainDecision = in.readOptionalWriteable(NodeDecision::new);
+    }
+
+    List<NodeDecision> canAllocateDecisions() {
+        return canAllocateDecisions;
+    }
+
+    @Nullable
+    NodeDecision canRemainDecision() {
+        return canRemainDecision;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeList(canAllocateDecisions);
+        out.writeOptionalWriteable(canRemainDecision);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.xContentList("can_allocate_decisions", canAllocateDecisions);
+        if (canRemainDecision != null) {
+            builder.field("can_remain_decision", canRemainDecision);
+        }
+        builder.endObject();
+        return null;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        NodeDecisions that = (NodeDecisions) o;
+        return Objects.equals(canAllocateDecisions, that.canAllocateDecisions) && Objects.equals(canRemainDecision, that.canRemainDecision);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(canAllocateDecisions, canRemainDecision);
+    }
+}

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderService.java

@@ -67,13 +67,20 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.SortedMap;
 import java.util.SortedSet;
+import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collector;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import static java.util.stream.Collectors.toMap;
+
 public class ReactiveStorageDeciderService implements AutoscalingDeciderService {
     public static final String NAME = "reactive_storage";
     /**
@@ -169,7 +176,9 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 unassignedBytes,
                 unassignedBytesUnassignedShards.shardIds(),
                 assignedBytes,
-                assignedBytesUnmovableShards.shardIds()
+                assignedBytesUnmovableShards.shardIds(),
+                unassignedBytesUnassignedShards.shardNodeDecisions(),
+                assignedBytesUnmovableShards.shardNodeDecisions()
             )
         );
     }
@@ -233,6 +242,9 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
 
     // todo: move this to top level class.
     public static class AllocationState {
+
+        static final int MAX_AMOUNT_OF_SHARD_DECISIONS = 5;
+
         private final ClusterState state;
         private final ClusterState originalState;
         private final AllocationDeciders allocationDeciders;
@@ -286,21 +298,40 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             this.roles = roles;
         }
 
-        public ShardsSize storagePreventsAllocation() {
+        private interface ShardNodeDecision {
+            ShardRouting shard();
+        }
+
+        record ShardNodeAllocationDecision(ShardRouting shard, List<NodeDecision> nodeDecisions) implements ShardNodeDecision {}
+
+        record ShardNodeRemainDecision(ShardRouting shard, NodeDecision nodeDecision) implements ShardNodeDecision {}
+
+        static <T extends ShardNodeDecision> T preferPrimary(T snd1, T snd2) {
+            return snd1.shard().primary() ? snd1 : snd2;
+        }
+
+        public ShardsAllocationResults storagePreventsAllocation() {
             RoutingAllocation allocation = new RoutingAllocation(allocationDeciders, state, info, shardSizeInfo, System.nanoTime());
-            List<ShardRouting> unassignedShards = state.getRoutingNodes()
+            List<ShardNodeAllocationDecision> unassignedShards = state.getRoutingNodes()
                 .unassigned()
                 .stream()
                 .filter(shard -> canAllocate(shard, allocation) == false)
-                .filter(shard -> cannotAllocateDueToStorage(shard, allocation))
+                .flatMap(shard -> cannotAllocateDueToStorage(shard, allocation).stream())
                 .toList();
-            return new ShardsSize(
-                unassignedShards.stream().mapToLong(this::sizeOf).sum(),
-                unassignedShards.stream().map(ShardRouting::shardId).collect(Collectors.toCollection(TreeSet::new))
+            return new ShardsAllocationResults(
+                unassignedShards.stream().map(e -> e.shard).mapToLong(this::sizeOf).sum(),
+                unassignedShards.stream().map(e -> e.shard.shardId()).collect(Collectors.toCollection(TreeSet::new)),
+                unassignedShards.stream()
+                    .filter(shardNodeDecisions -> shardNodeDecisions.nodeDecisions.size() > 0)
+                    .collect(toSortedMap(snd -> snd.shard.shardId(), snd -> new NodeDecisions(snd.nodeDecisions, null)))
+                    .entrySet()
+                    .stream()
+                    .limit(MAX_AMOUNT_OF_SHARD_DECISIONS)
+                    .collect(toSortedMap(Map.Entry::getKey, Map.Entry::getValue))
             );
         }
 
-        public ShardsSize storagePreventsRemainOrMove() {
+        public ShardsAllocationResults storagePreventsRemainOrMove() {
             RoutingAllocation allocation = new RoutingAllocation(allocationDeciders, state, info, shardSizeInfo, System.nanoTime());
 
             List<ShardRouting> candidates = new LinkedList<>();
@@ -315,10 +346,23 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             }
 
             // track these to ensure we do not double account if they both cannot remain and allocated due to storage.
-            Set<ShardRouting> unmovableShards = candidates.stream()
+            List<ShardNodeRemainDecision> unmovableShardNodeDecisions = candidates.stream()
                 .filter(s -> allocatedToTier(s, allocation))
-                .filter(s -> cannotRemainDueToStorage(s, allocation))
-                .collect(Collectors.toSet());
+                .flatMap(shard -> cannotRemainDueToStorage(shard, allocation).stream())
+                .toList();
+            Map<ShardId, ShardNodeRemainDecision> perShardIdUnmovable = unmovableShardNodeDecisions.stream()
+                .collect(toMap(snd -> snd.shard().shardId(), Function.identity(), AllocationState::preferPrimary, TreeMap::new));
+            Map<ShardId, NodeDecisions> otherNodesOnTierAllocationDecisions = perShardIdUnmovable.values()
+                .stream()
+                .limit(MAX_AMOUNT_OF_SHARD_DECISIONS)
+                .collect(
+                    toMap(
+                        snd -> snd.shard().shardId(),
+                        snd -> new NodeDecisions(canAllocateDecisions(allocation, snd.shard()), snd.nodeDecision())
+                    )
+                );
+            Set<ShardRouting> unmovableShards = unmovableShardNodeDecisions.stream().map(e -> e.shard).collect(Collectors.toSet());
+
             long unmovableBytes = unmovableShards.stream()
                 .collect(Collectors.groupingBy(ShardRouting::currentNodeId))
                 .entrySet()
@@ -326,20 +370,82 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 .mapToLong(e -> unmovableSize(e.getKey(), e.getValue()))
                 .sum();
 
-            List<ShardRouting> unallocatedShards = candidates.stream()
+            List<ShardNodeAllocationDecision> unallocatedShardNodeDecisions = candidates.stream()
                 .filter(Predicate.not(unmovableShards::contains))
-                .filter(s1 -> cannotAllocateDueToStorage(s1, allocation))
+                .flatMap(shard -> cannotAllocateDueToStorage(shard, allocation).stream())
                 .toList();
-            long unallocatableBytes = unallocatedShards.stream().mapToLong(this::sizeOf).sum();
+            Map<ShardId, ShardNodeAllocationDecision> perShardIdUnallocated = unallocatedShardNodeDecisions.stream()
+                .collect(toMap(snd -> snd.shard().shardId(), Function.identity(), AllocationState::preferPrimary, TreeMap::new));
+            Map<ShardId, NodeDecisions> canRemainDecisionsForUnallocatedShards = perShardIdUnallocated.values()
+                .stream()
+                .limit(MAX_AMOUNT_OF_SHARD_DECISIONS)
+                .collect(
+                    toMap(
+                        snd -> snd.shard().shardId(),
+                        snd -> new NodeDecisions(snd.nodeDecisions(), canRemainDecision(allocation, snd.shard()))
+                    )
+                );
+            long unallocatableBytes = unallocatedShardNodeDecisions.stream().map(e -> e.shard).mapToLong(this::sizeOf).sum();
 
-            return new ShardsSize(
+            Map<ShardId, NodeDecisions> shardDecisions = Stream.concat(
+                canRemainDecisionsForUnallocatedShards.entrySet().stream(),
+                otherNodesOnTierAllocationDecisions.entrySet().stream()
+            )
+                .collect(toSortedMap(Map.Entry::getKey, Map.Entry::getValue))
+                .entrySet()
+                .stream()
+                .limit(MAX_AMOUNT_OF_SHARD_DECISIONS)
+                .collect(toSortedMap(Map.Entry::getKey, Map.Entry::getValue));
+            return new ShardsAllocationResults(
                 unallocatableBytes + unmovableBytes,
-                Stream.concat(unmovableShards.stream(), unallocatedShards.stream())
-                    .map(ShardRouting::shardId)
-                    .collect(Collectors.toCollection(TreeSet::new))
+                Stream.concat(unmovableShardNodeDecisions.stream(), unallocatedShardNodeDecisions.stream())
+                    .map(e -> e.shard().shardId())
+                    .collect(Collectors.toCollection(TreeSet::new)),
+                shardDecisions
             );
         }
 
+        private List<NodeDecision> canAllocateDecisions(RoutingAllocation allocation, ShardRouting shardRouting) {
+            return state.getRoutingNodes()
+                .stream()
+                .filter(n -> nodeTierPredicate.test(n.node()))
+                .filter(n -> shardRouting.currentNodeId().equals(n.nodeId()) == false)
+                .map(
+                    n -> new NodeDecision(
+                        n.node(),
+                        withAllocationDebugEnabled(allocation, () -> allocationDeciders.canAllocate(shardRouting, n, allocation))
+                    )
+                )
+                .toList();
+        }
+
+        private NodeDecision canRemainDecision(RoutingAllocation allocation, ShardRouting shard) {
+            return new NodeDecision(
+                allocation.routingNodes().node(shard.currentNodeId()).node(),
+                withAllocationDebugEnabled(
+                    allocation,
+                    () -> allocationDeciders.canRemain(shard, allocation.routingNodes().node(shard.currentNodeId()), allocation)
+                )
+            );
+        }
+
+        private static <T extends Decision> T withAllocationDebugEnabled(RoutingAllocation allocation, Supplier<T> supplier) {
+            assert allocation.debugDecision() == false;
+            allocation.debugDecision(true);
+            try {
+                return supplier.get();
+            } finally {
+                allocation.debugDecision(false);
+            }
+        }
+
+        private static <T> Collector<T, ?, SortedMap<ShardId, NodeDecisions>> toSortedMap(
+            Function<T, ShardId> keyMapper,
+            Function<T, NodeDecisions> valueMapper
+        ) {
+            return toMap(keyMapper, valueMapper, (a, b) -> b, TreeMap::new);
+        }
+
         /**
          * Check if shard can remain where it is, with the additional check that the DataTierAllocationDecider did not allow it to stay
          * on a node in a lower preference tier.
@@ -369,26 +475,34 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
 
         /**
          * Check that disk decider is only decider for a node preventing allocation of the shard.
-         * @return true if and only if a node exists in the tier where only disk decider prevents allocation
+         * @return Non-empty {@link ShardNodeAllocationDecision} if and only if a node exists in the tier
+         *         where only disk decider prevents allocation
          */
-        private boolean cannotAllocateDueToStorage(ShardRouting shard, RoutingAllocation allocation) {
+        private Optional<ShardNodeAllocationDecision> cannotAllocateDueToStorage(ShardRouting shard, RoutingAllocation allocation) {
             if (nodeIds.isEmpty() && needsThisTier(shard, allocation)) {
-                return true;
+                return Optional.of(new ShardNodeAllocationDecision(shard, List.of()));
             }
             assert allocation.debugDecision() == false;
             // enable debug decisions to see all decisions and preserve the allocation decision label
             allocation.debugDecision(true);
             try {
-                boolean diskOnly = nodesInTier(allocation.routingNodes()).map(
-                    node -> allocationDeciders.canAllocate(shard, node, allocation)
-                ).anyMatch(ReactiveStorageDeciderService::isDiskOnlyNoDecision);
-                if (diskOnly && shard.unassigned() && shard.recoverySource().getType() == RecoverySource.Type.LOCAL_SHARDS) {
+                List<NodeDecision> nodeDecisions = nodesInTier(allocation.routingNodes()).map(
+                    node -> new NodeDecision(node.node(), allocationDeciders.canAllocate(shard, node, allocation))
+                ).toList();
+                List<NodeDecision> diskOnly = nodeDecisions.stream()
+                    .filter(nar -> ReactiveStorageDeciderService.isDiskOnlyNoDecision(nar.decision()))
+                    .toList();
+                if (diskOnly.size() > 0 && shard.unassigned() && shard.recoverySource().getType() == RecoverySource.Type.LOCAL_SHARDS) {
                     // For resize shards only allow autoscaling if there is no other node where the shard could fit had it not been
                     // a resize shard. Notice that we already removed any initial_recovery filters.
-                    diskOnly = nodesInTier(allocation.routingNodes()).map(node -> allocationDeciders.canAllocate(shard, node, allocation))
-                        .anyMatch(ReactiveStorageDeciderService::isResizeOnlyNoDecision) == false;
+                    boolean hasResizeOnly = nodesInTier(allocation.routingNodes()).map(
+                        node -> allocationDeciders.canAllocate(shard, node, allocation)
+                    ).anyMatch(ReactiveStorageDeciderService::isResizeOnlyNoDecision);
+                    if (hasResizeOnly) {
+                        return Optional.empty();
+                    }
                 }
-                return diskOnly;
+                return diskOnly.size() > 0 ? Optional.of(new ShardNodeAllocationDecision(shard, nodeDecisions)) : Optional.empty();
             } finally {
                 allocation.debugDecision(false);
             }
@@ -396,16 +510,18 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
 
         /**
          * Check that the disk decider is only decider that says NO to let shard remain on current node.
-         * @return true if and only if disk decider is only decider that says NO to canRemain.
+         * @return Non-empty {@link ShardNodeAllocationDecision} if and only if disk decider is only decider that says NO to canRemain.
          */
-        private boolean cannotRemainDueToStorage(ShardRouting shard, RoutingAllocation allocation) {
+        private Optional<ShardNodeRemainDecision> cannotRemainDueToStorage(ShardRouting shard, RoutingAllocation allocation) {
             assert allocation.debugDecision() == false;
             // enable debug decisions to see all decisions and preserve the allocation decision label
             allocation.debugDecision(true);
             try {
-                return isDiskOnlyNoDecision(
-                    allocationDeciders.canRemain(shard, allocation.routingNodes().node(shard.currentNodeId()), allocation)
-                );
+                RoutingNode node = allocation.routingNodes().node(shard.currentNodeId());
+                Decision decision = allocationDeciders.canRemain(shard, node, allocation);
+                return isDiskOnlyNoDecision(decision)
+                    ? Optional.of(new ShardNodeRemainDecision(shard, new NodeDecision(node.node(), decision)))
+                    : Optional.empty();
             } finally {
                 allocation.debugDecision(false);
             }
@@ -840,21 +956,25 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 throw new UnsupportedOperationException();
             }
         }
+
     }
 
     public static class ReactiveReason implements AutoscalingDeciderResult.Reason {
 
         static final int MAX_AMOUNT_OF_SHARDS = 512;
         private static final TransportVersion SHARD_IDS_OUTPUT_VERSION = TransportVersion.V_8_4_0;
+        private static final TransportVersion UNASSIGNED_NODE_DECISIONS_OUTPUT_VERSION = TransportVersion.V_8_9_0;
 
         private final String reason;
         private final long unassigned;
         private final long assigned;
         private final SortedSet<ShardId> unassignedShardIds;
         private final SortedSet<ShardId> assignedShardIds;
+        private final Map<ShardId, NodeDecisions> unassignedNodeDecisions;
+        private final Map<ShardId, NodeDecisions> assignedNodeDecisions;
 
         public ReactiveReason(String reason, long unassigned, long assigned) {
-            this(reason, unassigned, Collections.emptySortedSet(), assigned, Collections.emptySortedSet());
+            this(reason, unassigned, Collections.emptySortedSet(), assigned, Collections.emptySortedSet(), Map.of(), Map.of());
         }
 
         ReactiveReason(
@@ -862,13 +982,17 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             long unassigned,
             SortedSet<ShardId> unassignedShardIds,
             long assigned,
-            SortedSet<ShardId> assignedShardIds
+            SortedSet<ShardId> assignedShardIds,
+            Map<ShardId, NodeDecisions> unassignedNodeDecisions,
+            Map<ShardId, NodeDecisions> assignedNodeDecisions
         ) {
             this.reason = reason;
             this.unassigned = unassigned;
             this.assigned = assigned;
             this.unassignedShardIds = unassignedShardIds;
             this.assignedShardIds = assignedShardIds;
+            this.unassignedNodeDecisions = Objects.requireNonNull(unassignedNodeDecisions);
+            this.assignedNodeDecisions = Objects.requireNonNull(assignedNodeDecisions);
         }
 
         public ReactiveReason(StreamInput in) throws IOException {
@@ -882,6 +1006,13 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 unassignedShardIds = Collections.emptySortedSet();
                 assignedShardIds = Collections.emptySortedSet();
             }
+            if (in.getTransportVersion().onOrAfter(UNASSIGNED_NODE_DECISIONS_OUTPUT_VERSION)) {
+                unassignedNodeDecisions = in.readMap(ShardId::new, NodeDecisions::new);
+                assignedNodeDecisions = in.readMap(ShardId::new, NodeDecisions::new);
+            } else {
+                unassignedNodeDecisions = Map.of();
+                assignedNodeDecisions = Map.of();
+            }
         }
 
         @Override
@@ -905,6 +1036,14 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             return assignedShardIds;
         }
 
+        public Map<ShardId, NodeDecisions> unassignedNodeDecisions() {
+            return unassignedNodeDecisions;
+        }
+
+        public Map<ShardId, NodeDecisions> assignedNodeDecisions() {
+            return assignedNodeDecisions;
+        }
+
         @Override
         public String getWriteableName() {
             return ReactiveStorageDeciderService.NAME;
@@ -919,6 +1058,10 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 out.writeCollection(unassignedShardIds);
                 out.writeCollection(assignedShardIds);
             }
+            if (out.getTransportVersion().onOrAfter(UNASSIGNED_NODE_DECISIONS_OUTPUT_VERSION)) {
+                out.writeMap(unassignedNodeDecisions);
+                out.writeMap(assignedNodeDecisions);
+            }
         }
 
         @Override
@@ -931,6 +1074,14 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
             builder.field("assigned", assigned);
             builder.field("assigned_shards", assignedShardIds.stream().limit(MAX_AMOUNT_OF_SHARDS).toList());
             builder.field("assigned_shards_count", assignedShardIds.size());
+            builder.xContentValuesMap(
+                "unassigned_node_decisions",
+                unassignedNodeDecisions.entrySet().stream().collect(toMap(e -> e.getKey().toString(), Map.Entry::getValue))
+            );
+            builder.xContentValuesMap(
+                "assigned_node_decisions",
+                assignedNodeDecisions.entrySet().stream().collect(toMap(e -> e.getKey().toString(), Map.Entry::getValue))
+            );
             builder.endObject();
             return builder;
         }
@@ -944,12 +1095,22 @@ public class ReactiveStorageDeciderService implements AutoscalingDeciderService
                 && assigned == that.assigned
                 && reason.equals(that.reason)
                 && unassignedShardIds.equals(that.unassignedShardIds)
-                && assignedShardIds.equals(that.assignedShardIds);
+                && assignedShardIds.equals(that.assignedShardIds)
+                && Objects.equals(unassignedNodeDecisions, that.unassignedNodeDecisions)
+                && Objects.equals(assignedNodeDecisions, that.assignedNodeDecisions);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(reason, unassigned, assigned, unassignedShardIds, assignedShardIds);
+            return Objects.hash(
+                reason,
+                unassigned,
+                assigned,
+                unassignedShardIds,
+                assignedShardIds,
+                unassignedNodeDecisions,
+                assignedNodeDecisions
+            );
         }
     }
 }

x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ShardsSize.java → x-pack/plugin/autoscaling/src/main/java/org/elasticsearch/xpack/autoscaling/storage/ShardsAllocationResults.java

@@ -9,6 +9,7 @@ package org.elasticsearch.xpack.autoscaling.storage;
 
 import org.elasticsearch.index.shard.ShardId;
 
+import java.util.Map;
 import java.util.SortedSet;
 
-record ShardsSize(long sizeInBytes, SortedSet<ShardId> shardIds) {}
+record ShardsAllocationResults(long sizeInBytes, SortedSet<ShardId> shardIds, Map<ShardId, NodeDecisions> shardNodeDecisions) {}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/NodeDecisionTestUtils.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
+import org.elasticsearch.cluster.routing.allocation.decider.DiskThresholdDecider;
+import org.elasticsearch.cluster.routing.allocation.decider.FilterAllocationDecider;
+
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
+import static org.elasticsearch.test.ESTestCase.buildNewFakeTransportAddress;
+import static org.elasticsearch.test.ESTestCase.randomAlphaOfLength;
+import static org.elasticsearch.test.ESTestCase.randomFrom;
+
+class NodeDecisionTestUtils {
+
+    static NodeDecision randomNodeDecision() {
+        return new NodeDecision(randomDiscoveryNode(), randomDecision());
+    }
+
+    static DiscoveryNode randomDiscoveryNode() {
+        return new DiscoveryNode(randomAlphaOfLength(6), buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
+    }
+
+    static Decision randomDecision() {
+        return randomFrom(
+            new Decision.Single(Decision.Type.NO, DiskThresholdDecider.NAME, "Unable to allocate on disk"),
+            new Decision.Single(Decision.Type.YES, FilterAllocationDecider.NAME, "Filter allows allocation"),
+            new Decision.Single(Decision.Type.THROTTLE, "throttle label", "Throttling the consumer"),
+            new Decision.Multi().add(randomFrom(Decision.NO, Decision.YES, Decision.THROTTLE))
+                .add(new Decision.Single(Decision.Type.NO, "multi_no", "No multi decision"))
+                .add(new Decision.Single(Decision.Type.YES, "multi_yes", "Yes multi decision"))
+        );
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/NodeDecisionWireSerializationTests.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+
+public class NodeDecisionWireSerializationTests extends AbstractWireSerializingTestCase<NodeDecision> {
+
+    @Override
+    protected NodeDecision mutateInstance(NodeDecision instance) throws IOException {
+        if (randomBoolean()) {
+            return new NodeDecision(NodeDecisionTestUtils.randomDiscoveryNode(), instance.decision());
+        } else if (randomBoolean()) {
+            return new NodeDecision(instance.node(), randomValueOtherThan(instance.decision(), NodeDecisionTestUtils::randomDecision));
+        } else {
+            return randomValueOtherThan(instance, this::createTestInstance);
+        }
+    }
+
+    @Override
+    protected Writeable.Reader<NodeDecision> instanceReader() {
+        return NodeDecision::new;
+    }
+
+    @Override
+    protected NodeDecision createTestInstance() {
+        return NodeDecisionTestUtils.randomNodeDecision();
+    }
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/NodeDecisionsWireSerializationTests.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+package org.elasticsearch.xpack.autoscaling.storage;
+
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.test.AbstractWireSerializingTestCase;
+
+import java.io.IOException;
+import java.util.List;
+
+public class NodeDecisionsWireSerializationTests extends AbstractWireSerializingTestCase<NodeDecisions> {
+
+    @Override
+    protected NodeDecisions mutateInstance(NodeDecisions instance) throws IOException {
+        if (randomBoolean()) {
+            return new NodeDecisions(
+                randomValueOtherThan(instance.canAllocateDecisions(), () -> randomNodeDecisions()),
+                instance.canRemainDecision()
+            );
+        } else if (randomBoolean()) {
+            return new NodeDecisions(
+                instance.canAllocateDecisions(),
+                randomValueOtherThan(instance.canRemainDecision(), () -> NodeDecisionTestUtils.randomNodeDecision())
+            );
+        } else {
+            return randomValueOtherThan(instance, this::createTestInstance);
+        }
+    }
+
+    @Override
+    protected Writeable.Reader<NodeDecisions> instanceReader() {
+        return NodeDecisions::new;
+    }
+
+    @Override
+    protected NodeDecisions createTestInstance() {
+        return new NodeDecisions(randomNodeDecisions(), randomBoolean() ? NodeDecisionTestUtils.randomNodeDecision() : null);
+    }
+
+    private static List<NodeDecision> randomNodeDecisions() {
+        return randomList(8, NodeDecisionTestUtils::randomNodeDecision);
+    }
+
+}

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveReasonTests.java

@@ -7,6 +7,9 @@
 
 package org.elasticsearch.xpack.autoscaling.storage;
 
+import org.elasticsearch.Version;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.index.shard.ShardId;
@@ -18,12 +21,15 @@ import org.elasticsearch.xcontent.json.JsonXContent;
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.stream.Collectors;
 
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.emptySet;
 import static org.hamcrest.Matchers.greaterThan;
 
 public class ReactiveReasonTests extends ESTestCase {
@@ -37,12 +43,43 @@ public class ReactiveReasonTests extends ESTestCase {
         String indexName = randomAlphaOfLength(10);
         SortedSet<ShardId> unassignedShardIds = new TreeSet<>(randomUnique(() -> new ShardId(indexName, indexUUID, randomInt(1000)), 600));
         SortedSet<ShardId> assignedShardIds = new TreeSet<>(randomUnique(() -> new ShardId(indexName, indexUUID, randomInt(1000)), 600));
+        DiscoveryNode discoveryNode = new DiscoveryNode("node1", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
+        DiscoveryNode discoveryNode2 = new DiscoveryNode("node2", buildNewFakeTransportAddress(), emptyMap(), emptySet(), Version.CURRENT);
+        Map<ShardId, NodeDecisions> unassignedShardAllocationDecision = Map.of(
+            unassignedShardIds.first(),
+            new NodeDecisions(
+                List.of(
+                    new NodeDecision(discoveryNode, Decision.single(Decision.Type.NO, "no_label", "No space to allocate")),
+                    new NodeDecision(
+                        discoveryNode2,
+                        new Decision.Multi().add(
+                            Decision.single(Decision.Type.YES, "data_tier", "Enough disk on this node for the shard to remain")
+                        ).add(Decision.single(Decision.Type.NO, "shards_limit", "Disallowed because of shard limits"))
+                    )
+                ),
+                null
+            )
+        );
+        Map<ShardId, NodeDecisions> assignedShardAllocateDecision = Map.of(
+            assignedShardIds.first(),
+            new NodeDecisions(
+                List.of(),
+                new NodeDecision(
+                    discoveryNode,
+                    new Decision.Multi().add(Decision.single(Decision.Type.THROTTLE, "multi_throttle", "is not active yet"))
+                        .add(new Decision.Single(Decision.Type.NO, "multi_no", "No multi decision"))
+                        .add(new Decision.Single(Decision.Type.YES, "multi_yes", "Yes multi decision"))
+                )
+            )
+        );
         var reactiveReason = new ReactiveStorageDeciderService.ReactiveReason(
             reason,
             unassigned,
             unassignedShardIds,
             assigned,
-            assignedShardIds
+            assignedShardIds,
+            unassignedShardAllocationDecision,
+            assignedShardAllocateDecision
         );
 
         try (
@@ -77,6 +114,63 @@ public class ReactiveReasonTests extends ESTestCase {
             );
             assertSorted(xContentAssignedShardIds.stream().map(ShardId::fromString).toList());
             assertEquals(assignedShardIds.size(), map.get("assigned_shards_count"));
+
+            Map<String, Object> unassignedNodeDecisions = (Map<String, Object>) ((Map<String, Object>) map.get("unassigned_node_decisions"))
+                .get(unassignedShardIds.first().toString());
+            List<Map<String, Object>> canAllocateDecisions = (List<Map<String, Object>>) unassignedNodeDecisions.get(
+                "can_allocate_decisions"
+            );
+            assertEquals(2, canAllocateDecisions.size());
+            assertEquals("node1", canAllocateDecisions.get(0).get("node_id"));
+            assertEquals(
+                List.of(Map.of("decision", "NO", "decider", "no_label", "explanation", "No space to allocate")),
+                canAllocateDecisions.get(0).get("deciders")
+            );
+            assertEquals("node2", canAllocateDecisions.get(1).get("node_id"));
+            assertEquals(
+                List.of(
+                    Map.of("decision", "YES", "decider", "data_tier", "explanation", "Enough disk on this node for the shard to remain"),
+                    Map.of("decision", "NO", "decider", "shards_limit", "explanation", "Disallowed because of shard limits")
+                ),
+                canAllocateDecisions.get(1).get("deciders")
+            );
+            assertFalse(unassignedNodeDecisions.containsKey("can_remain_decision"));
+
+            Map<String, Object> assignedNodeDecisions = (Map<String, Object>) ((Map<String, Object>) map.get("assigned_node_decisions"))
+                .get(assignedShardIds.first().toString());
+            var canRemainDecision = (Map<String, Object>) assignedNodeDecisions.get("can_remain_decision");
+            assertEquals("node1", canRemainDecision.get("node_id"));
+            assertEquals(
+                List.of(
+                    Map.of("decision", "THROTTLE", "decider", "multi_throttle", "explanation", "is not active yet"),
+                    Map.of("decision", "NO", "decider", "multi_no", "explanation", "No multi decision"),
+                    Map.of("decision", "YES", "decider", "multi_yes", "explanation", "Yes multi decision")
+                ),
+                canRemainDecision.get("deciders")
+            );
+            assertEquals(List.of(), assignedNodeDecisions.get("can_allocate_decisions"));
+        }
+    }
+
+    public void testEmptyNodeDecisions() throws IOException {
+        var reactiveReason = new ReactiveStorageDeciderService.ReactiveReason(
+            randomAlphaOfLength(10),
+            randomNonNegativeLong(),
+            Collections.emptySortedSet(),
+            randomNonNegativeLong(),
+            Collections.emptySortedSet(),
+            Map.of(),
+            Map.of()
+        );
+        try (
+            XContentParser parser = createParser(
+                JsonXContent.jsonXContent,
+                BytesReference.bytes(reactiveReason.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS))
+            )
+        ) {
+            Map<String, Object> map = parser.map();
+            assertEquals(Map.of(), map.get("unassigned_node_decisions"));
+            assertEquals(Map.of(), map.get("assigned_node_decisions"));
         }
     }
 

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderDecisionTests.java

@@ -60,17 +60,21 @@ import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeSet;
 import java.util.function.Consumer;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
+import static java.util.Collections.emptySortedSet;
 import static java.util.function.Predicate.not;
+import static java.util.stream.Collectors.toSet;
 import static java.util.stream.Collectors.toUnmodifiableMap;
 import static org.elasticsearch.cluster.node.DiscoveryNodeRole.DATA_HOT_NODE_ROLE;
 import static org.elasticsearch.cluster.node.DiscoveryNodeRole.DATA_WARM_NODE_ROLE;
 import static org.elasticsearch.cluster.routing.ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE;
 import static org.elasticsearch.common.util.set.Sets.haveNonEmptyIntersection;
+import static org.elasticsearch.xpack.autoscaling.storage.ReactiveStorageDeciderService.AllocationState.MAX_AMOUNT_OF_SHARD_DECISIONS;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -149,18 +153,35 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
             long numPrevents = allocatableShards.numOfShards();
             assert round != 0 || numPrevents > 0 : "must have shards that can be allocated on first round";
 
+            SortedSet<ShardId> expectedShardIds = allocatableShards.shardIds();
             verify(
                 ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation,
-                new ShardsSize(numPrevents, allocatableShards.shardIds()),
+                numPrevents,
+                expectedShardIds,
+                shardNodeDecisions -> {
+                    if (numPrevents > 0) {
+                        assertEquals(cappedShardIds(expectedShardIds), shardNodeDecisions.keySet());
+                        assertEquals(
+                            Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                            firstNoDecision(shardNodeDecisions.get(expectedShardIds.first()).canAllocateDecisions().get(0))
+                        );
+                        assertNull(shardNodeDecisions.get(expectedShardIds.first()).canRemainDecision());
+                    } else {
+                        assertTrue(shardNodeDecisions.isEmpty());
+                    }
+                    return true;
+                },
                 mockCanAllocateDiskDecider
             );
             verify(
                 ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation,
-                emptyShardsSize(),
+                0,
+                emptySortedSet(),
+                Map::isEmpty,
                 mockCanAllocateDiskDecider,
                 CAN_ALLOCATE_NO_DECIDER
             );
-            verify(ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation, emptyShardsSize());
+            verify(ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation, 0, emptySortedSet(), Map::isEmpty);
             // verify empty tier (no cold nodes) are always assumed a storage reason.
             SortedSet<ShardId> unassignedShardIds = state.getRoutingNodes()
                 .unassigned()
@@ -170,22 +191,41 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
             verify(
                 moveToCold(allIndices()),
                 ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation,
-                new ShardsSize(state.getRoutingNodes().unassigned().size(), unassignedShardIds),
+                state.getRoutingNodes().unassigned().size(),
+                unassignedShardIds,
+                Map::isEmpty,
                 DiscoveryNodeRole.DATA_COLD_NODE_ROLE
             );
             verify(
                 ReactiveStorageDeciderService.AllocationState::storagePreventsAllocation,
-                emptyShardsSize(),
+                0,
+                emptySortedSet(),
+                Map::isEmpty,
                 DiscoveryNodeRole.DATA_COLD_NODE_ROLE
             );
             if (numPrevents > 0) {
-                verifyScale(numPrevents, "not enough storage available, needs " + numPrevents + "b", mockCanAllocateDiskDecider);
+                verifyScale(numPrevents, "not enough storage available, needs " + numPrevents + "b", Map::isEmpty, shardIdNodeDecisions -> {
+                    assertEquals(cappedShardIds(expectedShardIds), shardIdNodeDecisions.keySet());
+                    assertEquals(
+                        Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                        firstNoDecision(shardIdNodeDecisions.get(expectedShardIds.first()).canAllocateDecisions().get(0))
+                    );
+                    assertNull(shardIdNodeDecisions.get(expectedShardIds.first()).canRemainDecision());
+                    return true;
+                }, mockCanAllocateDiskDecider);
             } else {
-                verifyScale(0, "storage ok", mockCanAllocateDiskDecider);
+                verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty, mockCanAllocateDiskDecider);
             }
-            verifyScale(0, "storage ok", mockCanAllocateDiskDecider, CAN_ALLOCATE_NO_DECIDER);
-            verifyScale(0, "storage ok");
-            verifyScale(addDataNodes(DATA_HOT_NODE_ROLE, "additional", state, hotNodes), 0, "storage ok", mockCanAllocateDiskDecider);
+            verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty, mockCanAllocateDiskDecider, CAN_ALLOCATE_NO_DECIDER);
+            verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty);
+            verifyScale(
+                addDataNodes(DATA_HOT_NODE_ROLE, "additional", state, hotNodes),
+                0,
+                "storage ok",
+                Map::isEmpty,
+                Map::isEmpty,
+                mockCanAllocateDiskDecider
+            );
             lastState = state;
             startRandomShards();
             ++round;
@@ -248,30 +288,55 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
                 )
         );
 
+        SortedSet<ShardId> expectedShardIds = RoutingNodesHelper.shardsWithState(state.getRoutingNodes(), ShardRoutingState.STARTED)
+            .stream()
+            .map(ShardRouting::shardId)
+            .filter(subjectShards::contains)
+            .collect(Collectors.toCollection(TreeSet::new));
         verify(
             ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            new ShardsSize(
-                subjectShards.size(),
-                RoutingNodesHelper.shardsWithState(state.getRoutingNodes(), ShardRoutingState.STARTED)
-                    .stream()
-                    .map(ShardRouting::shardId)
-                    .filter(sr -> subjectShards.contains(sr))
-                    .collect(Collectors.toCollection(TreeSet::new))
-            ),
+            subjectShards.size(),
+            expectedShardIds,
+            shardIdNodeDecisions -> {
+                assertEquals(cappedShardIds(expectedShardIds), shardIdNodeDecisions.keySet());
+                assertEquals(
+                    Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                    firstNoDecision(shardIdNodeDecisions.get(expectedShardIds.first()).canAllocateDecisions().get(0))
+                );
+                assertDebugNoDecision(shardIdNodeDecisions.get(expectedShardIds.first()).canRemainDecision().decision());
+                return true;
+            },
             mockCanAllocateDiskDecider
         );
         verify(
             ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            emptyShardsSize(),
+            0,
+            emptySortedSet(),
+            Map::isEmpty,
             mockCanAllocateDiskDecider,
             CAN_ALLOCATE_NO_DECIDER
         );
-        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, emptyShardsSize());
-
-        verifyScale(subjectShards.size(), "not enough storage available, needs " + subjectShards.size() + "b", mockCanAllocateDiskDecider);
-        verifyScale(0, "storage ok", mockCanAllocateDiskDecider, CAN_ALLOCATE_NO_DECIDER);
-        verifyScale(0, "storage ok");
-        verifyScale(addDataNodes(DATA_HOT_NODE_ROLE, "additional", state, hotNodes), 0, "storage ok", mockCanAllocateDiskDecider);
+        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, 0, emptySortedSet(), Map::isEmpty);
+
+        verifyScale(subjectShards.size(), "not enough storage available, needs " + subjectShards.size() + "b", shardIdNodeDecisions -> {
+            assertEquals(cappedShardIds(expectedShardIds), shardIdNodeDecisions.keySet());
+            assertDebugNoDecision(shardIdNodeDecisions.get(expectedShardIds.first()).canRemainDecision().decision());
+            assertEquals(
+                Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                firstNoDecision(shardIdNodeDecisions.values().iterator().next().canAllocateDecisions().get(0))
+            );
+            return true;
+        }, Map::isEmpty, mockCanAllocateDiskDecider);
+        verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty, mockCanAllocateDiskDecider, CAN_ALLOCATE_NO_DECIDER);
+        verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty);
+        verifyScale(
+            addDataNodes(DATA_HOT_NODE_ROLE, "additional", state, hotNodes),
+            0,
+            "storage ok",
+            Map::isEmpty,
+            Map::isEmpty,
+            mockCanAllocateDiskDecider
+        );
     }
 
     public void testMoveToEmpty() {
@@ -290,7 +355,9 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
 
         verify(
             ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            emptyShardsSize(),
+            0,
+            emptySortedSet(),
+            Map::isEmpty,
             DiscoveryNodeRole.DATA_COLD_NODE_ROLE
         );
 
@@ -305,7 +372,17 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
         verify(
             moveToCold(candidates),
             ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            new ShardsSize(allocatedCandidateShards, allocatedShardIds),
+            allocatedCandidateShards,
+            allocatedShardIds,
+            shardNodeDecisions -> {
+                assertEquals(cappedShardIds(allocatedShardIds), shardNodeDecisions.keySet());
+                if (allocatedShardIds.size() > 0) {
+                    NodeDecisions nodeDecisions = shardNodeDecisions.get(allocatedShardIds.first());
+                    assertTrue(nodeDecisions.canAllocateDecisions().isEmpty());
+                    assertNotNull(nodeDecisions.canRemainDecision());
+                }
+                return true;
+            },
             DiscoveryNodeRole.DATA_COLD_NODE_ROLE
         );
     }
@@ -360,54 +437,118 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
             .filter(o -> subjectShards.contains(o))
             .collect(Collectors.toCollection(TreeSet::new));
 
+        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, nodes, shardIds, shardNodeDecisions -> {
+            assertEquals(cappedShardIds(shardIds), shardNodeDecisions.keySet());
+            List<NodeDecision> canAllocateDecisions = shardNodeDecisions.get(shardIds.first()).canAllocateDecisions();
+            assertEquals(hotNodes - 1, canAllocateDecisions.size());
+            if (canAllocateDecisions.size() > 0) {
+                assertDebugNoDecision(canAllocateDecisions.get(0).decision());
+            }
+
+            assertEquals(
+                Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                firstNoDecision(shardNodeDecisions.get(shardIds.first()).canRemainDecision())
+            );
+            return true;
+        }, mockCanRemainDiskDecider, CAN_ALLOCATE_NO_DECIDER);
         verify(
             ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            new ShardsSize(nodes, shardIds),
-            mockCanRemainDiskDecider,
-            CAN_ALLOCATE_NO_DECIDER
-        );
-        verify(
-            ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            emptyShardsSize(),
+            0,
+            emptySortedSet(),
+            Map::isEmpty,
             mockCanRemainDiskDecider,
             CAN_REMAIN_NO_DECIDER,
             CAN_ALLOCATE_NO_DECIDER
         );
-        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, emptyShardsSize());
+        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, 0, emptySortedSet(), Map::isEmpty);
 
         // only consider it once (move case) if both cannot remain and cannot allocate.
-        verify(
-            ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove,
-            new ShardsSize(nodes, shardIds),
-            mockCanAllocateDiskDecider,
-            mockCanRemainDiskDecider
-        );
+        verify(ReactiveStorageDeciderService.AllocationState::storagePreventsRemainOrMove, nodes, shardIds, shardNodeDecisions -> {
+            assertEquals(cappedShardIds(shardIds), shardNodeDecisions.keySet());
+            List<NodeDecision> canAllocateDecisions = shardNodeDecisions.get(shardIds.first()).canAllocateDecisions();
+            assertEquals(hotNodes - 1, canAllocateDecisions.size());
+            if (canAllocateDecisions.size() > 0) {
+                assertDebugNoDecision(canAllocateDecisions.get(0).decision());
+            }
 
-        verifyScale(nodes, "not enough storage available, needs " + nodes + "b", mockCanRemainDiskDecider, CAN_ALLOCATE_NO_DECIDER);
-        verifyScale(0, "storage ok", mockCanRemainDiskDecider, CAN_REMAIN_NO_DECIDER, CAN_ALLOCATE_NO_DECIDER);
-        verifyScale(0, "storage ok");
+            assertEquals(
+                Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                firstNoDecision(shardNodeDecisions.get(shardIds.first()).canRemainDecision())
+            );
+            return true;
+        }, mockCanAllocateDiskDecider, mockCanRemainDiskDecider);
+
+        verifyScale(nodes, "not enough storage available, needs " + nodes + "b", shardNodeDecisions -> {
+            assertEquals(cappedShardIds(shardIds), shardNodeDecisions.keySet());
+            List<NodeDecision> canAllocateDecisions = shardNodeDecisions.get(shardIds.first()).canAllocateDecisions();
+            assertEquals(hotNodes - 1, canAllocateDecisions.size());
+            if (canAllocateDecisions.size() > 0) {
+                assertDebugNoDecision(canAllocateDecisions.get(0).decision());
+            }
+
+            assertEquals(
+                Decision.single(Decision.Type.NO, "disk_threshold", "test"),
+                firstNoDecision(shardNodeDecisions.get(shardIds.first()).canRemainDecision())
+            );
+            return true;
+        }, Map::isEmpty, mockCanRemainDiskDecider, CAN_ALLOCATE_NO_DECIDER);
+        verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty, mockCanRemainDiskDecider, CAN_REMAIN_NO_DECIDER, CAN_ALLOCATE_NO_DECIDER);
+        verifyScale(0, "storage ok", Map::isEmpty, Map::isEmpty);
+    }
+
+    private static void assertDebugNoDecision(Decision canAllocateDecision) {
+        assertEquals(Decision.Type.NO, canAllocateDecision.type());
+        assertTrue(canAllocateDecision.getDecisions().stream().anyMatch(d -> d.type() == Decision.Type.NO));
+        assertTrue(canAllocateDecision.getDecisions().stream().anyMatch(d -> d.type() == Decision.Type.YES));
+    }
+
+    private static SortedSet<ShardId> cappedShardIds(SortedSet<ShardId> shardIds) {
+        return shardIds.stream().limit(MAX_AMOUNT_OF_SHARD_DECISIONS).collect(Collectors.toCollection(TreeSet::new));
     }
 
-    private static ShardsSize emptyShardsSize() {
-        return new ShardsSize(0, Collections.emptySortedSet());
+    private static Decision firstNoDecision(NodeDecision nodeAllocationResult) {
+        List<Decision> noDecisions = nodeAllocationResult.decision()
+            .getDecisions()
+            .stream()
+            .filter(d -> d.type() == Decision.Type.NO)
+            .toList();
+        if (noDecisions.isEmpty()) {
+            throw new IllegalStateException("Unable to find NO can_remain decision");
+        }
+        return noDecisions.get(0);
     }
 
     private interface VerificationSubject {
-        ShardsSize invoke(ReactiveStorageDeciderService.AllocationState state);
+        ShardsAllocationResults invoke(ReactiveStorageDeciderService.AllocationState state);
     }
 
-    private void verify(VerificationSubject subject, ShardsSize expected, AllocationDecider... allocationDeciders) {
-        verify(subject, expected, DATA_HOT_NODE_ROLE, allocationDeciders);
+    private void verify(
+        VerificationSubject subject,
+        long expectedSizeInBytes,
+        SortedSet<ShardId> expectedShardIds,
+        Predicate<Map<ShardId, NodeDecisions>> nodeDecisionsCheck,
+        AllocationDecider... allocationDeciders
+    ) {
+        verify(subject, expectedSizeInBytes, expectedShardIds, nodeDecisionsCheck, DATA_HOT_NODE_ROLE, allocationDeciders);
     }
 
-    private void verify(VerificationSubject subject, ShardsSize expected, DiscoveryNodeRole role, AllocationDecider... allocationDeciders) {
-        verify(this.state, subject, expected, role, allocationDeciders);
+    private void verify(
+        VerificationSubject subject,
+        long expectedSizeInBytes,
+        SortedSet<ShardId> expectedShardIds,
+        Predicate<Map<ShardId, NodeDecisions>> nodeDecisionsCheck,
+        DiscoveryNodeRole role,
+        AllocationDecider... allocationDeciders
+    ) {
+        verify(this.state, subject, expectedSizeInBytes, expectedShardIds, nodeDecisionsCheck, role, allocationDeciders);
     }
 
     private static void verify(
         ClusterState state,
         VerificationSubject subject,
-        ShardsSize expected,
+        long expectedSizeInBytes,
+        SortedSet<ShardId> expectedShardIds,
+        Predicate<Map<ShardId, NodeDecisions>> nodeDecisionsCheck,
         DiscoveryNodeRole role,
         AllocationDecider... allocationDeciders
     ) {
@@ -417,14 +558,30 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
             createAllocationDeciders(allocationDeciders),
             TestShardRoutingRoleStrategies.DEFAULT_ROLE_ONLY
         );
-        assertThat(subject.invoke(allocationState), equalTo(expected));
+        ShardsAllocationResults shardsAllocationResults = subject.invoke(allocationState);
+        assertThat(shardsAllocationResults.sizeInBytes(), equalTo(expectedSizeInBytes));
+        assertThat(shardsAllocationResults.shardIds(), equalTo(expectedShardIds));
+        assertTrue("failed canAllocate decisions check", nodeDecisionsCheck.test(shardsAllocationResults.shardNodeDecisions()));
     }
 
-    private void verifyScale(long expectedDifference, String reason, AllocationDecider... allocationDeciders) {
-        verifyScale(state, expectedDifference, reason, allocationDeciders);
+    private void verifyScale(
+        long expectedDifference,
+        String reason,
+        Predicate<Map<ShardId, NodeDecisions>> assignedNodeDecisions,
+        Predicate<Map<ShardId, NodeDecisions>> unassignedNodeDecisions,
+        AllocationDecider... allocationDeciders
+    ) {
+        verifyScale(state, expectedDifference, reason, assignedNodeDecisions, unassignedNodeDecisions, allocationDeciders);
     }
 
-    private static void verifyScale(ClusterState state, long expectedDifference, String reason, AllocationDecider... allocationDeciders) {
+    private static void verifyScale(
+        ClusterState state,
+        long expectedDifference,
+        String reason,
+        Predicate<Map<ShardId, NodeDecisions>> assignedNodeDecisions,
+        Predicate<Map<ShardId, NodeDecisions>> unassignedNodeDecisions,
+        AllocationDecider... allocationDeciders
+    ) {
         ReactiveStorageDeciderService decider = new ReactiveStorageDeciderService(
             Settings.EMPTY,
             new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS),
@@ -443,11 +600,15 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
             assertThat(resultReason.summary(), equalTo(reason));
             assertThat(resultReason.unassignedShardIds(), equalTo(decider.allocationState(context).storagePreventsAllocation().shardIds()));
             assertThat(resultReason.assignedShardIds(), equalTo(decider.allocationState(context).storagePreventsRemainOrMove().shardIds()));
+            assertTrue("failed assigned decisions check", assignedNodeDecisions.test(resultReason.assignedNodeDecisions()));
+            assertTrue("failed unassigned decisions check", unassignedNodeDecisions.test(resultReason.unassignedNodeDecisions()));
         } else {
             assertThat(result.requiredCapacity(), is(nullValue()));
             assertThat(resultReason.summary(), equalTo("current capacity not available"));
             assertThat(resultReason.unassignedShardIds(), equalTo(Set.of()));
             assertThat(resultReason.assignedShardIds(), equalTo(Set.of()));
+            assertThat(resultReason.unassignedNodeDecisions(), equalTo(Map.of()));
+            assertThat(resultReason.assignedNodeDecisions(), equalTo(Map.of()));
         }
     }
 
@@ -458,16 +619,18 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
         RoutingAllocation allocation = createRoutingAllocation(state, deciders);
         // There could be duplicated of shard ids, and numOfShards is calculated based on them,
         // so we can't just collect to the shard ids to `TreeSet`
-        List<ShardId> allocatableShards = state.getRoutingNodes()
+        List<ShardRouting> allocatableShards = state.getRoutingNodes()
             .unassigned()
             .stream()
             .filter(shard -> subjectShards.contains(shard.shardId()))
             .filter(
                 shard -> allocation.routingNodes().stream().anyMatch(node -> deciders.canAllocate(shard, node, allocation) != Decision.NO)
             )
-            .map(ShardRouting::shardId)
             .toList();
-        return new AllocatableShards(allocatableShards.size(), new TreeSet<>(allocatableShards));
+        return new AllocatableShards(
+            allocatableShards.size(),
+            allocatableShards.stream().map(ShardRouting::shardId).collect(Collectors.toCollection(TreeSet::new))
+        );
     }
 
     private boolean hasStartedSubjectShard() {
@@ -545,7 +708,7 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
                     Set<RoutingNode> candidates = allocation.routingNodes()
                         .stream()
                         .filter(n -> allocation.deciders().canAllocate(toMove, n, allocation) == Decision.YES)
-                        .collect(Collectors.toSet());
+                        .collect(toSet());
                     if (candidates.isEmpty() == false) {
                         allocation.routingNodes().relocateShard(toMove, randomFrom(candidates).nodeId(), 0L, allocation.changes());
                     }
@@ -582,7 +745,7 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
         private TestAutoscalingDeciderContext(ClusterState state, Set<DiscoveryNodeRole> roles, AutoscalingCapacity currentCapacity) {
             this.state = state;
             this.currentCapacity = currentCapacity;
-            this.nodes = state.nodes().stream().filter(n -> roles.stream().anyMatch(n.getRoles()::contains)).collect(Collectors.toSet());
+            this.nodes = state.nodes().stream().filter(n -> roles.stream().anyMatch(n.getRoles()::contains)).collect(toSet());
             this.roles = roles;
             this.info = createClusterInfo(state);
         }
@@ -697,11 +860,10 @@ public class ReactiveStorageDeciderDecisionTests extends AutoscalingTestCase {
     }
 
     private static String randomNodeId(RoutingNodes routingNodes, DiscoveryNodeRole role) {
-        return randomFrom(routingNodes.stream().map(RoutingNode::node).filter(n -> n.getRoles().contains(role)).collect(Collectors.toSet()))
-            .getId();
+        return randomFrom(routingNodes.stream().map(RoutingNode::node).filter(n -> n.getRoles().contains(role)).collect(toSet())).getId();
     }
 
     private static Set<ShardId> shardIds(Iterable<ShardRouting> candidateShards) {
-        return StreamSupport.stream(candidateShards.spliterator(), false).map(ShardRouting::shardId).collect(Collectors.toSet());
+        return StreamSupport.stream(candidateShards.spliterator(), false).map(ShardRouting::shardId).collect(toSet());
     }
 }

x-pack/plugin/autoscaling/src/test/java/org/elasticsearch/xpack/autoscaling/storage/ReactiveStorageDeciderReasonWireSerializationTests.java

@@ -9,14 +9,19 @@ package org.elasticsearch.xpack.autoscaling.storage;
 
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.test.AbstractWireSerializingTestCase;
 import org.elasticsearch.test.ESTestCase;
 
+import java.util.Map;
 import java.util.TreeSet;
 
+import static org.elasticsearch.xpack.autoscaling.storage.NodeDecisionTestUtils.randomNodeDecision;
+
 public class ReactiveStorageDeciderReasonWireSerializationTests extends AbstractWireSerializingTestCase<
     ReactiveStorageDeciderService.ReactiveReason> {
+
     @Override
     protected Writeable.Reader<ReactiveStorageDeciderService.ReactiveReason> instanceReader() {
         return ReactiveStorageDeciderService.ReactiveReason::new;
@@ -24,14 +29,16 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
 
     @Override
     protected ReactiveStorageDeciderService.ReactiveReason mutateInstance(ReactiveStorageDeciderService.ReactiveReason instance) {
-        switch (between(0, 5)) {
+        switch (between(0, 7)) {
             case 0:
                 return new ReactiveStorageDeciderService.ReactiveReason(
                     randomValueOtherThan(instance.summary(), () -> randomAlphaOfLength(10)),
                     instance.unassigned(),
                     instance.unassignedShardIds(),
                     instance.assigned(),
-                    instance.assignedShardIds()
+                    instance.assignedShardIds(),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
                 );
             case 1:
                 return new ReactiveStorageDeciderService.ReactiveReason(
@@ -39,7 +46,9 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
                     randomValueOtherThan(instance.unassigned(), ESTestCase::randomNonNegativeLong),
                     instance.unassignedShardIds(),
                     instance.assigned(),
-                    instance.assignedShardIds()
+                    instance.assignedShardIds(),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
                 );
             case 2:
                 return new ReactiveStorageDeciderService.ReactiveReason(
@@ -47,7 +56,9 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
                     instance.unassigned(),
                     instance.unassignedShardIds(),
                     randomValueOtherThan(instance.assigned(), ESTestCase::randomNonNegativeLong),
-                    instance.assignedShardIds()
+                    instance.assignedShardIds(),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
                 );
             case 3:
                 return new ReactiveStorageDeciderService.ReactiveReason(
@@ -55,7 +66,9 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
                     instance.unassigned(),
                     new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
                     instance.assigned(),
-                    instance.assignedShardIds()
+                    instance.assignedShardIds(),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
                 );
             case 4:
                 return new ReactiveStorageDeciderService.ReactiveReason(
@@ -63,7 +76,9 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
                     instance.unassigned(),
                     instance.unassignedShardIds(),
                     instance.assigned(),
-                    new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8))
+                    new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
                 );
             case 5:
                 return new ReactiveStorageDeciderService.ReactiveReason(
@@ -71,7 +86,29 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
                     instance.unassigned(),
                     new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
                     instance.assigned(),
-                    new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8))
+                    new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
+                    instance.unassignedNodeDecisions(),
+                    instance.assignedNodeDecisions()
+                );
+            case 6:
+                return new ReactiveStorageDeciderService.ReactiveReason(
+                    instance.summary(),
+                    instance.unassigned(),
+                    instance.unassignedShardIds(),
+                    instance.assigned(),
+                    instance.assignedShardIds(),
+                    randomShardNodeDecisions(),
+                    instance.assignedNodeDecisions()
+                );
+            case 7:
+                return new ReactiveStorageDeciderService.ReactiveReason(
+                    instance.summary(),
+                    instance.unassigned(),
+                    instance.unassignedShardIds(),
+                    instance.assigned(),
+                    instance.assignedShardIds(),
+                    instance.unassignedNodeDecisions(),
+                    randomShardNodeDecisions()
                 );
             default:
                 fail("unexpected");
@@ -86,7 +123,20 @@ public class ReactiveStorageDeciderReasonWireSerializationTests extends Abstract
             randomNonNegativeLong(),
             new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
             randomNonNegativeLong(),
-            new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8))
+            new TreeSet<>(randomUnique(() -> new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)), 8)),
+            randomShardNodeDecisions(),
+            randomShardNodeDecisions()
+        );
+    }
+
+    private static Map<ShardId, NodeDecisions> randomShardNodeDecisions() {
+        return randomMap(
+            1,
+            5,
+            () -> Tuple.tuple(
+                new ShardId(randomAlphaOfLength(8), UUIDs.randomBase64UUID(), randomInt(5)),
+                new NodeDecisions(randomList(8, () -> randomNodeDecision()), randomBoolean() ? randomNodeDecision() : null)
+            )
         );
     }
 }