diff --git a/.github/workflows/bump_version.yml b/.github/workflows/bump_version.yml
index 67ebdaab..76bdaa37 100644
--- a/.github/workflows/bump_version.yml
+++ b/.github/workflows/bump_version.yml
@@ -6,7 +6,7 @@ on:
     # Inputs the workflow accepts.
     inputs:
       pcluster-version:
-        description: 'The target version of ParallelCluster CLI'
+        description: 'The target version of ParallelCluster Node'
         required: true
         type: string
       branch:
diff --git a/.github/workflows/cherry-pick-to-branch.yml b/.github/workflows/cherry-pick-to-branch.yml
new file mode 100644
index 00000000..8bbb8a4c
--- /dev/null
+++ b/.github/workflows/cherry-pick-to-branch.yml
@@ -0,0 +1,201 @@
+name: Cherry-pick to branch
+
+on:
+  workflow_dispatch:
+    inputs:
+      pr_numbers:
+        description: "Space-separated list of PR numbers to cherry-pick"
+        required: true
+        type: string
+      target_branches:
+        description: "Space-separated list of target branches to cherry-pick into"
+        required: true
+        type: string
+
+jobs:
+  cherry-pick:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+      pull-requests: write
+    steps:
+      - name: Get PR details
+        id: pr-info
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          PR_NUMBERS: ${{ inputs.pr_numbers }}
+          TARGET_BRANCHES: ${{ inputs.target_branches }}
+          REPO: ${{ github.repository }}
+        run: |
+          # workflow_dispatch has no PR event payload, so fetch everything we need
+          # about each PR from the API.
+
+          # Space-separated list of PR numbers. PRs are processed in the order
+          # they are provided, so commits are cherry-picked in that same order.
+          PR_LIST=$(echo "${PR_NUMBERS}" | tr -s ' ' | sed 's/^ *//;s/ *$//')
+          for PR in ${PR_LIST}; do
+            if ! echo "${PR}" | grep -qE '^[0-9]+$'; then
+              echo "::error::Invalid PR number '${PR}' — must be numeric"
+              exit 1
+            fi
+          done
+
+          if [ -z "${PR_LIST// }" ]; then
+            echo "::error::No PR numbers provided"
+            exit 1
+          fi
+
+          # Validate target branches contain only safe characters
+          for BRANCH in ${TARGET_BRANCHES}; do
+            if ! echo "${BRANCH}" | grep -qE '^[a-zA-Z0-9._/-]+$'; then
+              echo "::error::Invalid branch name '${BRANCH}' — must match [a-zA-Z0-9._/-]+"
+              exit 1
+            fi
+          done
+
+          ALL_COMMITS=""
+          PR_DETAILS=""
+          PR_DISPLAY=""
+          FIRST_TITLE=""
+          for PR in ${PR_LIST}; do
+            PR_JSON=$(gh pr view "${PR}" --repo "${REPO}" \
+              --json baseRefName,title,author)
+            BASE_BRANCH=$(echo "${PR_JSON}" | jq -r '.baseRefName')
+            PR_TITLE=$(echo "${PR_JSON}" | jq -r '.title')
+            PR_AUTHOR=$(echo "${PR_JSON}" | jq -r '.author.login')
+
+            # A PR can contain merges from its base branch. Those commits should not be cherry-picked.
+            # The types of commit are distinguished by the number of parents:
+            # 1 parent = non-merge commits, 2 parents = merge commit
+            COMMITS=$(gh api --paginate \
+              "repos/${REPO}/pulls/${PR}/commits?per_page=100" \
+              --jq '.[] | select((.parents | length) < 2) | .sha' | tr '\n' ' ')
+
+            if [ -z "${COMMITS// }" ]; then
+              echo "::error::PR #${PR} contains no non-merge commits to cherry-pick"
+              exit 1
+            fi
+
+            ALL_COMMITS="${ALL_COMMITS} ${COMMITS}"
+            PR_DETAILS="${PR_DETAILS}- #${PR} \"${PR_TITLE}\" by @${PR_AUTHOR} (from \`${BASE_BRANCH}\`)"$'\n'
+
+            if [ -z "${PR_DISPLAY}" ]; then
+              PR_DISPLAY="#${PR}"
+              FIRST_TITLE="${PR_TITLE}"
+            else
+              PR_DISPLAY="${PR_DISPLAY}, #${PR}"
+            fi
+          done
+
+          # Trim leading whitespace from the accumulated commit list
+          ALL_COMMITS=$(echo "${ALL_COMMITS}" | sed 's/^ *//')
+
+          # Drop the trailing newline so the rendered body has no blank last line
+          PR_DETAILS="${PR_DETAILS%$'\n'}"
+
+          # A single PR (no space in the list) keeps its title in the PR name;
+          # multiple PRs reference the first one followed by "and more".
+          FIRST_PR="${PR_LIST%% *}"
+          case "${PR_LIST}" in
+            *" "*) NEW_PR_TITLE="Cherry-pick #${FIRST_PR} and more" ;;
+            *)     NEW_PR_TITLE="Cherry-pick ${PR_DISPLAY}: ${FIRST_TITLE}" ;;
+          esac
+
+          echo "commits=${ALL_COMMITS}" >> "$GITHUB_OUTPUT"
+          echo "pr_numbers=${PR_LIST}" >> "$GITHUB_OUTPUT"
+          echo "pr_display=${PR_DISPLAY}" >> "$GITHUB_OUTPUT"
+          echo "pr_title=${NEW_PR_TITLE}" >> "$GITHUB_OUTPUT"
+          {
+            echo "pr_details<<CHERRY_PICK_EOF"
+            echo "${PR_DETAILS}"
+            echo "CHERRY_PICK_EOF"
+          } >> "$GITHUB_OUTPUT"
+
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Configure git
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+
+      - name: Cherry-pick and create PR(s)
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          COMMITS: ${{ steps.pr-info.outputs.commits }}
+          PR_NUMBERS: ${{ steps.pr-info.outputs.pr_numbers }}
+          PR_DISPLAY: ${{ steps.pr-info.outputs.pr_display }}
+          PR_TITLE: ${{ steps.pr-info.outputs.pr_title }}
+          PR_DETAILS: ${{ steps.pr-info.outputs.pr_details }}
+          TARGET_BRANCHES: ${{ inputs.target_branches }}
+          REPO: ${{ github.repository }}
+        run: |
+          # Fetch each PR's commits. They may live in a fork (different account),
+          # so they are not present in the base repo checkout. The refs/pull/*/head
+          # ref makes them available regardless of which fork the PR came from.
+          for PR in ${PR_NUMBERS}; do
+            git fetch origin "refs/pull/${PR}/head"
+          done
+
+          # Name the branch after the first and last PR in the set (a single PR
+          # keeps the plain pr-<n> form) so it stays within git's ref length limit
+          # regardless of how many PRs are cherry-picked.
+          FIRST_PR="${PR_NUMBERS%% *}"
+          LAST_PR="${PR_NUMBERS##* }"
+          case "${PR_NUMBERS}" in
+            *" "*) BRANCH_PR="pr-${FIRST_PR}-and-${LAST_PR}" ;;
+            *)     BRANCH_PR="pr-${FIRST_PR}" ;;
+          esac
+
+          HAS_FAILURE=false
+          for TARGET_BRANCH in ${TARGET_BRANCHES}; do
+            CHERRY_PICK_BRANCH="cherry-pick/${BRANCH_PR}-to-${TARGET_BRANCH}"
+
+            echo "Cherry-picking commits from PR(s) ${PR_DISPLAY} to ${TARGET_BRANCH}"
+            echo "Commits: ${COMMITS}"
+
+            git fetch origin "${TARGET_BRANCH}"
+            git checkout -b "${CHERRY_PICK_BRANCH}" "origin/${TARGET_BRANCH}"
+
+            # Cherry-pick all commits in order (works for both squash and rebase)
+            CHERRY_PICK_FAILED=false
+            for COMMIT in ${COMMITS}; do
+              if ! git cherry-pick "${COMMIT}" --no-edit; then
+                CHERRY_PICK_FAILED=true
+                git cherry-pick --abort
+                break
+              fi
+            done
+
+            if [ "${CHERRY_PICK_FAILED}" = "false" ]; then
+              git push origin "${CHERRY_PICK_BRANCH}"
+              PR_BODY=$(printf '%s\n\nIncluded PRs:\n%s' \
+                "Automated cherry-pick of ${PR_DISPLAY} to \`${TARGET_BRANCH}\`." \
+                "${PR_DETAILS}")
+              NEW_PR_URL=$(gh pr create \
+                --base "${TARGET_BRANCH}" \
+                --head "${CHERRY_PICK_BRANCH}" \
+                --title "${PR_TITLE}" \
+                --body "${PR_BODY}" \
+                --label "cherry-pick")
+              for PR in ${PR_NUMBERS}; do
+                gh pr comment "${PR}" --repo "${REPO}" \
+                  --body "✅ Cherry-pick to \`${TARGET_BRANCH}\` opened: ${NEW_PR_URL}"
+              done
+              echo "✅ Cherry-pick PR to ${TARGET_BRANCH} created successfully: ${NEW_PR_URL}"
+            else
+              HAS_FAILURE=true
+              echo "::error::Cherry-pick to ${TARGET_BRANCH} failed due to conflicts."
+            fi
+
+            # Clean up for next iteration
+            git checkout --detach 2>/dev/null || true
+            git branch -D "${CHERRY_PICK_BRANCH}" 2>/dev/null || true
+          done
+
+          if [ "${HAS_FAILURE}" = "true" ]; then
+            exit 1
+          fi
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 138ae4e4..2f161ff9 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -26,40 +26,20 @@ jobs:
       matrix:
         os: [ubuntu-latest]
         name:
-          - Python 3.9 Tests
-          - Python 3.10 Tests
-          - Python 3.11 Tests
-          - Python 3.12 Tests
-          - Python 3.13 Tests
-          - Python 3.12 Tests Coverage
+          - Python 3.14 Tests
+          - Python 3.14 Tests Coverage
           - Code Checks
         include:
-          - name: Python 3.9 Tests
-            python: 3.9
+          - name: Python 3.14 Tests
+            python: '3.14'
             toxdir: cli
-            toxenv: py39-nocov
-          - name: Python 3.10 Tests
-            python: '3.10'
+            toxenv: py314-nocov
+          - name: Python 3.14 Tests Coverage
+            python: '3.14'
             toxdir: cli
-            toxenv: py310-nocov
-          - name: Python 3.11 Tests
-            python: '3.11'
-            toxdir: cli
-            toxenv: py311-nocov
-          - name: Python 3.12 Tests
-            python: '3.12'
-            toxdir: cli
-            toxenv: py312-nocov
-          - name: Python 3.13 Tests
-            python: '3.13'
-            toxdir: cli
-            toxenv: py313-nocov
-          - name: Python 3.12 Tests Coverage
-            python: 3.12
-            toxdir: cli
-            toxenv: py312-cov
+            toxenv: py314-cov
           - name: Code Checks
-            python: 3.12
+            python: '3.14'
             toxdir: cli
             toxenv: code-linters
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1c55533c..fe701eab 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,15 @@ aws-parallelcluster-node CHANGELOG
 
 This file is used to list changes made in each version of the aws-parallelcluster-node package.
 
+3.16.2
+------
+
+**BUG FIXES**
+- Fix an issue where EC2 throttling during compute node launch is not retried and is reported as insufficient capacity
+  when using Multiple Instance Types or multiple subnets.
+- Fix an issue where the Comment and Extra fields of Slurm nodes and the Comment field of Slurm reservations
+  could alter the node and reservation information read by ParallelCluster.
+
 3.16.1
 ------
 
diff --git a/setup.py b/setup.py
index f26ccea5..5ab47e6d 100644
--- a/setup.py
+++ b/setup.py
@@ -44,7 +44,7 @@ def read(fname):
     license="Apache License 2.0",
     packages=find_packages("src", exclude=["tests"]),
     package_dir={"": "src"},
-    python_requires=">=3.9",
+    python_requires=">=3.14",
     install_requires=requires,
     entry_points=dict(console_scripts=console_scripts),
     zip_safe=False,
diff --git a/src/common/schedulers/slurm_commands.py b/src/common/schedulers/slurm_commands.py
index a46e4412..dbec3dd8 100644
--- a/src/common/schedulers/slurm_commands.py
+++ b/src/common/schedulers/slurm_commands.py
@@ -75,10 +75,15 @@
     r"|(LastBusyTime=\S+)"
     r"|(ReservationName=\S+)"
     r"|(InstanceId=\S+)"
-    r"|(Reason=.*)",
+    r"|(Reason=.*)"
+    r"|(Comment=.*)"
+    r"|(Extra=.*)",
     re.MULTILINE,
 )
 
+# Fields matched up to the end of the line and discarded, to prevent their free-text content from being parsed as fields
+SCONTROL_DISCARDED_FIELDS = ("Comment", "Extra")
+
 # Fields extracted from `scontrol show partitions` output. `(?<!Next)` ensures `State` is matched but
 # `NextState` is not.
 SCONTROL_PARTITION_INFO_FIELD_REGEX = re.compile(
@@ -426,6 +431,8 @@ def _extract_scontrol_records(raw_output: str, field_regex) -> List[Dict[str, st
         fields = {}
         for match in field_regex.finditer(record):
             key, _, value = match.group(0).partition("=")
+            if key in SCONTROL_DISCARDED_FIELDS:
+                continue
             fields[key] = value
         if fields:
             records.append(fields)
diff --git a/src/common/schedulers/slurm_reservation_commands.py b/src/common/schedulers/slurm_reservation_commands.py
index afd71681..645c9c56 100644
--- a/src/common/schedulers/slurm_reservation_commands.py
+++ b/src/common/schedulers/slurm_reservation_commands.py
@@ -39,7 +39,7 @@
 # Fields extracted from raw `scontrol show reservations` output. Only `ReservationName` is anchored at the
 # start of a line; `(?<!Next)` ensures `State` is matched but `NextState` is not.
 SCONTROL_RESERVATION_INFO_FIELD_REGEX = re.compile(
-    r"^(ReservationName=\S+)" r"|(?<!Next)(State=\S+)" r"|(Users=\S+)" r"|(Nodes=\S+)",
+    r"^(ReservationName=\S+)" r"|(?<!Next)(State=\S+)" r"|(Users=\S+)" r"|(Nodes=\S+)" r"|(Comment=.*)",
     re.MULTILINE,
 )
 
diff --git a/src/slurm_plugin/fleet_manager.py b/src/slurm_plugin/fleet_manager.py
index 37fb37ab..83bfd3a4 100644
--- a/src/slurm_plugin/fleet_manager.py
+++ b/src/slurm_plugin/fleet_manager.py
@@ -21,6 +21,7 @@
 from common.utils import setup_logging_filter
 from retrying import retry
 from slurm_plugin.common import print_with_count
+from slurm_plugin.slurm_resources import SlurmNode
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +31,32 @@
 INSTANCE_INFO_RETRIEVAL_TIMEOUT_DEFAULT = 90
 INSTANCE_INFO_RETRIEVAL_MAX_BACKOFF = 30
 
+# The only error code EC2 returns when a launch request is throttled, see
+# https://docs.aws.amazon.com/ec2/latest/devguide/ec2-api-throttling.html
+LAUNCH_THROTTLING_ERROR_CODE = "RequestLimitExceeded"
+
+# An override that CreateFleet did not fulfill because another override failed is reported with this exact
+# code and message pair, which points at the real error rather than being one.
+UNFULFILLED_OVERRIDE_ERROR = (
+    "UnfulfillableCapacity",
+    "Failed to fulfill capacity. Please review errors in the response.",
+)
+
+# Errors confined to one instance type and subnet pool. Reported alongside throttling, they leave the throttled
+# pools able to serve the batch once the rate limit refills, so the throttling is still retried. Any other error
+# alongside throttling would fail the retry on every pool alike and is reported instead.
+POOL_LEVEL_ERROR_CODES = frozenset(
+    SlurmNode.EC2_ICE_ERROR_CODES
+    | {
+        "InsufficientFreeAddressesInSubnet",
+        "InvalidSubnetID.NotFound",
+        "InvalidSubnet",
+        "InsufficientVolumeCapacity",
+        "VolumeTypeNotAvailableInZone",
+        "ServiceUnavailable",
+    }
+)
+
 
 class EC2Instance:
     def __init__(self, id, private_ip, hostname, all_private_ips, launch_time):
@@ -430,7 +457,40 @@ def _launch_instances(self, launch_params):
             if partial_instance_ids:
                 logger.error("Unable to retrieve instance info for instances: %s", partial_instance_ids)
 
-            if not instances and len(err_list) == 1:
+            if not instances:
+                # Drop the entries that only point at the real error, unless the response carries nothing else.
+                real_errors = [
+                    err
+                    for err in err_list
+                    if (err.get("ErrorCode"), err.get("ErrorMessage")) != UNFULFILLED_OVERRIDE_ERROR
+                ]
+                if real_errors:
+                    err_list = real_errors
+                # A single cause is normally left. Should there be several, throttling is retried only when every
+                # other cause is confined to a pool, since the throttled pools can then still serve the batch once the
+                # rate limit refills. Any other cause would fail the retry on every pool alike and is reported instead.
+                # For example:
+                # - [RequestLimitExceeded, InsufficientInstanceCapacity] -> retry, other pools can still serve the batch
+                # - [RequestLimitExceeded, VcpuLimitExceeded] -> report VcpuLimitExceeded, a retry hits the same limit
+                throttling = next(
+                    (err for err in err_list if err.get("ErrorCode") == LAUNCH_THROTTLING_ERROR_CODE), None
+                )
+                if throttling:
+                    blocking = next(
+                        (
+                            err
+                            for err in err_list
+                            if err.get("ErrorCode") != LAUNCH_THROTTLING_ERROR_CODE
+                            and err.get("ErrorCode") not in POOL_LEVEL_ERROR_CODES
+                        ),
+                        None,
+                    )
+                    chosen = blocking or throttling
+                    raise LaunchInstancesError(chosen.get("ErrorCode"), chosen.get("ErrorMessage"))
+            # Normally a single cause is left. Reporting the first one of several is a second safety net: the
+            # caller otherwise records a hardcoded InsufficientInstanceCapacity, and any code EC2 actually
+            # returned is more useful than an invented one, whichever of them the response happens to list first.
+            if not instances and err_list:
                 raise LaunchInstancesError(err_list[0].get("ErrorCode"), err_list[0].get("ErrorMessage"))
             return {"Instances": instances}
         except ClientError as e:
diff --git a/src/slurm_plugin/slurm_resources.py b/src/slurm_plugin/slurm_resources.py
index 19ed0c50..77d9b472 100644
--- a/src/slurm_plugin/slurm_resources.py
+++ b/src/slurm_plugin/slurm_resources.py
@@ -126,8 +126,16 @@ def __init__(
         reservation="",
         features="",
         extra="",
+        exclusive=None,
+        priority=None,
+        **unknown_fields,
     ):
-        """Initialize Slurm job with attributes."""
+        """
+        Initialize Slurm job with attributes.
+
+        Slurm adds fields to the jobs of the Resume File over time (exclusive and priority in 26.05), so fields this
+        class does not know are ignored rather than rejected: rejecting them fails the resume of every node.
+        """
         super().__init__(
             job_id,
         )
@@ -139,9 +147,10 @@ def __init__(
         self.reservation = reservation
         self.features = features
         self.extra = extra
-
-    def is_exclusive(self):
-        return self.oversubscribe == JobOversubscribe.NO
+        self.exclusive = exclusive
+        self.priority = priority
+        if unknown_fields:
+            logger.debug("Ignoring unknown fields of job %s in Slurm Resume File: %s", job_id, unknown_fields)
 
     def __eq__(self, other):
         """Compare 2 SlurmJob objects."""
diff --git a/tests/common/schedulers/test_slurm_commands.py b/tests/common/schedulers/test_slurm_commands.py
index e07e1877..3343cc6b 100644
--- a/tests/common/schedulers/test_slurm_commands.py
+++ b/tests/common/schedulers/test_slurm_commands.py
@@ -1404,7 +1404,127 @@ def test_get_nodes_info_argument_validation(nodes, cmd_timeout, expected_scontro
                     },
                 ]
             ),
-        )
+        ),
+        pytest.param(
+            (
+                "NodeName=queue1-st-compute-resource-1-1 Arch=x86_64 CoresPerSocket=1\n"
+                "   NodeAddr=1.2.3.4 NodeHostName=queue1-st-compute-resource-1-1 Version=25.11.4\n"
+                "   State=IDLE+CLOUD ThreadsPerCore=1 TmpDisk=0 Weight=1 Owner=N/A MCS_label=N/A\n"
+                "   Partitions=queue1\n"
+                "   BootTime=2023-01-26T09:56:30 SlurmdStartTime=2023-01-26T09:57:15\n"
+                "   LastBusyTime=2023-01-26T09:57:15\n"
+                "   Reason=someReason InstanceId=someInstanceId\n"
+                "   Comment=someComment State=DOWN+DRAIN NodeAddr=someNodeAddr "
+                "InstanceId=someInstanceId Reason=someReason\n"
+                "   Extra=someExtra State=DOWN+DRAIN NodeAddr=someNodeAddr "
+                "InstanceId=someInstanceId Reason=someReason\n"
+                "   InstanceId=i-0abc123def456 InstanceType=t2.medium\n"
+            ),
+            [
+                {
+                    "NodeName": "queue1-st-compute-resource-1-1",
+                    "NodeAddr": "1.2.3.4",
+                    "NodeHostName": "queue1-st-compute-resource-1-1",
+                    "State": "IDLE+CLOUD",
+                    "Partitions": "queue1",
+                    "SlurmdStartTime": "2023-01-26T09:57:15",
+                    "LastBusyTime": "2023-01-26T09:57:15",
+                    "Reason": "someReason InstanceId=someInstanceId",
+                    "InstanceId": "i-0abc123def456",
+                }
+            ],
+            id="Free-text fields with key=value pairs are ignored",
+        ),
+        pytest.param(
+            (
+                "NodeName=queue1-st-compute-resource-1-1 Arch=x86_64 CoresPerSocket=1\n"
+                "   NodeAddr=1.2.3.4 NodeHostName=queue1-st-compute-resource-1-1 Version=25.11.4\n"
+                "   State=IDLE+CLOUD ThreadsPerCore=1 TmpDisk=0 Weight=1 Owner=N/A MCS_label=N/A\n"
+                "   Partitions=queue1 Comment=someComment State=DOWN+DRAIN InstanceId=someInstanceId\n"
+            ),
+            [
+                {
+                    "NodeName": "queue1-st-compute-resource-1-1",
+                    "NodeAddr": "1.2.3.4",
+                    "NodeHostName": "queue1-st-compute-resource-1-1",
+                    "State": "IDLE+CLOUD",
+                    "Partitions": "queue1",
+                }
+            ],
+            id="Free-text field sharing a line with other fields is ignored up to the end of the line",
+        ),
+        pytest.param(
+            (
+                "NodeName=q1-st-cr1-1 Arch=x86_64 CoresPerSocket=1 \n"
+                "   CPUAlloc=0 CPUEfctv=2 CPUTot=2 CPULoad=0.08\n"
+                "   AvailableFeatures=static,t3.medium,cr1\n"
+                "   ActiveFeatures=static,t3.medium,cr1\n"
+                "   Gres=(null)\n"
+                "   NodeAddr=1.2.3.4 NodeHostName=q1-st-cr1-1 Version=25.11.8\n"
+                "   OS=Linux 6.12.110-135.202.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Sep 25 01:45:58 UTC 2026 \n"
+                "   RealMemory=3891 AllocMem=0 FreeMem=2452 Sockets=2 Boards=1\n"
+                "   State=IDLE+CLOUD ThreadsPerCore=1 TmpDisk=0 Weight=1 Owner=N/A MCS_label=N/A\n"
+                "   Partitions=q1 \n"
+                "   BootTime=2026-10-08T19:57:35 SlurmdStartTime=2026-10-08T20:00:08\n"
+                "   LastBusyTime=2026-10-08T20:00:11 ResumeAfterTime=None\n"
+                "   CfgTRES=cpu=2,mem=3891M,billing=2\n"
+                "   AllocTRES=\n"
+                "   CurrentWatts=0 AveWatts=0\n"
+                "   Reason=Node start up [root@2026-10-08T20:00:11]\n"
+                "   Comment=someComment\n"
+                "   InstanceId=i-003d87dc15788298b InstanceType=t3.medium\n"
+            ),
+            [
+                {
+                    "NodeName": "q1-st-cr1-1",
+                    "NodeAddr": "1.2.3.4",
+                    "NodeHostName": "q1-st-cr1-1",
+                    "State": "IDLE+CLOUD",
+                    "Partitions": "q1",
+                    "SlurmdStartTime": "2026-10-08T20:00:08",
+                    "LastBusyTime": "2026-10-08T20:00:11",
+                    "Reason": "Node start up [root@2026-10-08T20:00:11]",
+                    "InstanceId": "i-003d87dc15788298b",
+                }
+            ],
+            id="Real Slurm 25.11.8 output with Comment",
+        ),
+        pytest.param(
+            (
+                "NodeName=q1-st-cr1-1 Arch=x86_64 CoresPerSocket=1 \n"
+                "   CPUAlloc=0 CPUEfctv=2 CPUTot=2 CPULoad=0.08\n"
+                "   AvailableFeatures=static,t3.medium,cr1\n"
+                "   ActiveFeatures=static,t3.medium,cr1\n"
+                "   Gres=(null)\n"
+                "   NodeAddr=1.2.3.4 NodeHostName=q1-st-cr1-1 Version=25.11.8\n"
+                "   OS=Linux 6.12.110-135.202.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Sep 25 01:45:58 UTC 2026 \n"
+                "   RealMemory=3891 AllocMem=0 FreeMem=2452 Sockets=2 Boards=1\n"
+                "   State=IDLE+CLOUD ThreadsPerCore=1 TmpDisk=0 Weight=1 Owner=N/A MCS_label=N/A\n"
+                "   Partitions=q1 \n"
+                "   BootTime=2026-10-08T19:57:35 SlurmdStartTime=2026-10-08T20:00:08\n"
+                "   LastBusyTime=2026-10-08T20:00:11 ResumeAfterTime=None\n"
+                "   CfgTRES=cpu=2,mem=3891M,billing=2\n"
+                "   AllocTRES=\n"
+                "   CurrentWatts=0 AveWatts=0\n"
+                "   Reason=Node start up [root@2026-10-08T20:00:11]\n"
+                "   Comment=someComment State=DOWN+DRAIN InstanceId=someInstanceId NodeAddr=someNodeAddr\n"
+                "   InstanceId=i-003d87dc15788298b InstanceType=t3.medium\n"
+            ),
+            [
+                {
+                    "NodeName": "q1-st-cr1-1",
+                    "NodeAddr": "1.2.3.4",
+                    "NodeHostName": "q1-st-cr1-1",
+                    "State": "IDLE+CLOUD",
+                    "Partitions": "q1",
+                    "SlurmdStartTime": "2026-10-08T20:00:08",
+                    "LastBusyTime": "2026-10-08T20:00:11",
+                    "Reason": "Node start up [root@2026-10-08T20:00:11]",
+                    "InstanceId": "i-003d87dc15788298b",
+                }
+            ],
+            id="Real Slurm 25.11.8 output with Comment containing fields",
+        ),
     ],
 )
 def test_extract_scontrol_records(scontrol_output, expected_parsed_output):
@@ -1558,6 +1678,45 @@ def test_run_scontrol_command_oserror_raises_and_logs(mocker, caplog):
     assert_that(caplog.text).contains("Unable to execute scontrol command")
 
 
+@pytest.mark.parametrize(
+    "free_text_fields",
+    [
+        pytest.param(["Comment"], id="Comment"),
+        pytest.param(["Extra"], id="Extra"),
+        pytest.param(["Comment", "Extra"], id="Comment and Extra"),
+    ],
+)
+def test_get_nodes_info_ignores_fields_in_free_text_fields(free_text_fields, mocker):
+    # A Comment or Extra containing key=value pairs must not alter the node information, e.g. making a healthy node
+    # drained
+    mocker.patch(
+        "common.schedulers.slurm_commands._run_scontrol_command",
+        return_value=(
+            "NodeName=q1-st-cr1-1 Arch=x86_64 CoresPerSocket=1 \n"
+            "   NodeAddr=1.2.3.4 NodeHostName=q1-st-cr1-1 Version=1.2.3\n"
+            "   State=IDLE+CLOUD ThreadsPerCore=1 TmpDisk=0 Weight=1 Owner=N/A MCS_label=N/A\n"
+            "   Partitions=q1 \n"
+            "   BootTime=2026-10-08T19:57:35 SlurmdStartTime=2026-10-08T20:00:08\n"
+            "   LastBusyTime=2026-10-08T20:00:11 ResumeAfterTime=None\n"
+            + "".join(
+                f"   {field}=some{field} State=DOWN+DRAIN NodeAddr=someNodeAddr InstanceId=someInstanceId\n"
+                for field in free_text_fields
+            )
+            + "   InstanceId=i-0123456789abcdefg InstanceType=t3.medium\n\n"
+        ),
+        autospec=True,
+    )
+
+    nodes = get_nodes_info("q1-st-cr1-1")
+
+    assert_that(nodes).is_length(1)
+    assert_that(nodes[0]).is_instance_of(StaticNode)
+    assert_that(nodes[0].state_string).is_equal_to("IDLE+CLOUD")
+    assert_that(nodes[0].is_drained()).is_false()
+    assert_that(nodes[0].nodeaddr).is_equal_to("1.2.3.4")
+    assert_that(nodes[0].instance_id).is_equal_to("i-0123456789abcdefg")
+
+
 def test_get_nodes_info_parsing_failure_propagates(mocker):
     # If scontrol returns a field the parser cannot interpret (e.g. an unparseable date), the error surfaces
     # instead of being silently masked by the (former) shell pipeline.
diff --git a/tests/common/schedulers/test_slurm_reservation_commands.py b/tests/common/schedulers/test_slurm_reservation_commands.py
index e600056b..fc2291f1 100644
--- a/tests/common/schedulers/test_slurm_reservation_commands.py
+++ b/tests/common/schedulers/test_slurm_reservation_commands.py
@@ -480,6 +480,7 @@ def test_get_slurm_reservations_info(mocker):
         "   TRES=cpu=480\n"
         "   Users=root Groups=(null) Accounts=(null) Licenses=(null) State=ACTIVE BurstBuffer=(null) Watts=n/a\n"
         "   MaxStartDelay=(null)\n"
+        "   Comment=someComment Users=someUser State=someState Nodes=someNodes\n"
     )
     run_scontrol_command_mocked = mocker.patch(
         "common.schedulers.slurm_reservation_commands._run_scontrol_command",
diff --git a/tests/slurm_plugin/slurm_resources/test_slurm_resources.py b/tests/slurm_plugin/slurm_resources/test_slurm_resources.py
index 13b1a41a..73454fd1 100644
--- a/tests/slurm_plugin/slurm_resources/test_slurm_resources.py
+++ b/tests/slurm_plugin/slurm_resources/test_slurm_resources.py
@@ -18,6 +18,7 @@
     DynamicNode,
     EC2InstanceHealthState,
     InvalidNodenameError,
+    JobOversubscribe,
     MissingInstance,
     SlurmPartition,
     SlurmResumeJob,
@@ -1528,7 +1529,7 @@ def test_get_node_list(nodenames, expected_node_list, expected_exception):
 
 class TestSlurmJob:
     @pytest.mark.parametrize(
-        "job_json, expected_nodes_alloc, expected_nodes_resume, expected_is_exclusive, expected_exception",
+        "job_json, expected_nodes_alloc, expected_nodes_resume, expected_oversubscribe, expected_exception",
         [
             (
                 {
@@ -1543,7 +1544,7 @@ class TestSlurmJob:
                 },
                 ["queue1-st-c5xlarge-1", "queue1-st-c5xlarge-3", "queue1-st-c5xlarge-4", "queue1-st-c5xlarge-5"],
                 ["queue1-st-c5xlarge-7", "queue1-st-c5xlarge-8"],
-                True,
+                JobOversubscribe.NO,
                 False,
             ),
             (
@@ -1559,7 +1560,7 @@ class TestSlurmJob:
                 },
                 ["queue1-st-c5xlarge-1", "queue1-st-c5xlarge-3", "queue1-st-c5xlarge-4", "queue1-st-c5xlarge-5"],
                 ["queue1-st-c5xlarge-7", "queue1-st-c5xlarge-8"],
-                False,
+                JobOversubscribe.YES,
                 False,
             ),
             (
@@ -1607,13 +1608,51 @@ class TestSlurmJob:
                 },
                 ["queue1-st-c5xlarge-1"],
                 ["queue1-st-c5xlarge-1"],
+                JobOversubscribe.OK,
+                False,
+            ),
+            # Slurm 26.05 adds exclusive and priority
+            (
+                {
+                    "exclusive": "NODE",
+                    "extra": None,
+                    "job_id": 140816,
+                    "priority": 1,
+                    "features": None,
+                    "nodes_alloc": "queue1-st-c5xlarge-1",
+                    "nodes_resume": "queue1-st-c5xlarge-1",
+                    "oversubscribe": "NO",
+                    "partition": "cloud_exclusive",
+                    "reservation": None,
+                },
+                ["queue1-st-c5xlarge-1"],
+                ["queue1-st-c5xlarge-1"],
+                JobOversubscribe.NO,
                 False,
+            ),
+            # Fields a future Slurm adds to the Resume File must not break the resume
+            (
+                {
+                    "exclusive": "TOPO",
+                    "extra": None,
+                    "job_id": 140816,
+                    "features": None,
+                    "nodes_alloc": "queue1-st-c5xlarge-1",
+                    "nodes_resume": "queue1-st-c5xlarge-1",
+                    "oversubscribe": "NO",
+                    "partition": "cloud",
+                    "reservation": None,
+                    "field_added_by_a_future_slurm": "value",
+                },
+                ["queue1-st-c5xlarge-1"],
+                ["queue1-st-c5xlarge-1"],
+                JobOversubscribe.NO,
                 False,
             ),
         ],
     )
     def test_slurm_job(
-        self, job_json, expected_nodes_alloc, expected_nodes_resume, expected_is_exclusive, expected_exception
+        self, job_json, expected_nodes_alloc, expected_nodes_resume, expected_oversubscribe, expected_exception
     ):
         if expected_exception:
             with pytest.raises(InvalidNodenameError):
@@ -1622,7 +1661,9 @@ def test_slurm_job(
             job = SlurmResumeJob(**job_json)
             assert_that(job.nodes_alloc).is_equal_to(expected_nodes_alloc)
             assert_that(job.nodes_resume).is_equal_to(expected_nodes_resume)
-            assert_that(job.is_exclusive()).is_equal_to(expected_is_exclusive)
+            assert_that(job.oversubscribe).is_equal_to(expected_oversubscribe)
+            assert_that(job.exclusive).is_equal_to(job_json.get("exclusive"))
+            assert_that(job.priority).is_equal_to(job_json.get("priority"))
 
     @pytest.mark.parametrize(
         "jobs_json, expected_equal",
@@ -1788,7 +1829,9 @@ def test_slurm_job_hash(
                 "partition='cloud_exclusive', "
                 "reservation=None, "
                 "features=None, "
-                "extra=None)",
+                "extra=None, "
+                "exclusive=None, "
+                "priority=None)",
             ),
         ],
     )
diff --git a/tests/slurm_plugin/test_fleet_manager.py b/tests/slurm_plugin/test_fleet_manager.py
index 988f3f21..5818e1f8 100644
--- a/tests/slurm_plugin/test_fleet_manager.py
+++ b/tests/slurm_plugin/test_fleet_manager.py
@@ -19,6 +19,7 @@
 from slurm_plugin.fleet_manager import (
     INSTANCE_INFO_RETRIEVAL_MAX_BACKOFF,
     INSTANCE_INFO_RETRIEVAL_TIMEOUT_DEFAULT,
+    POOL_LEVEL_ERROR_CODES,
     Ec2CreateFleetManager,
     EC2Instance,
     Ec2RunInstancesManager,
@@ -28,6 +29,41 @@
 
 from tests.common import FLEET_CONFIG, MockedBoto3Request
 
+UNFULFILLED_OVERRIDE = {
+    "ErrorCode": "UnfulfillableCapacity",
+    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+}
+UNSUPPORTED_ERROR = {"ErrorCode": "Unsupported", "ErrorMessage": "Not supported in this AZ."}
+MIN_TARGET_CAPACITY_ERROR = {
+    "ErrorCode": "UnfulfillableCapacity",
+    "ErrorMessage": "Unable to fulfill request due to MinTargetCapacity constraints. Please adjust your request.",
+}
+THROTTLING_ERROR = {"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."}
+VCPU_LIMIT_ERROR = {"ErrorCode": "VcpuLimitExceeded", "ErrorMessage": "vCPU limit"}
+SUBNET_FULL_ERROR = {"ErrorCode": "InsufficientFreeAddressesInSubnet", "ErrorMessage": "Not enough free addresses."}
+
+
+def _raises_launch_error(err_list):
+    """Mirror when _launch_instances turns a CreateFleet response with no instances into an exception."""
+    return bool(err_list)
+
+
+def _expected_launch_error_code(err_list):
+    """Mirror which error _launch_instances reports: a non pool-level cause beats throttling, which beats the rest."""
+    real_errors = [err for err in err_list if err != UNFULFILLED_OVERRIDE] or err_list
+    throttling = next((err for err in real_errors if err["ErrorCode"] == "RequestLimitExceeded"), None)
+    if throttling:
+        blocking = next(
+            (
+                err
+                for err in real_errors
+                if err["ErrorCode"] != "RequestLimitExceeded" and err["ErrorCode"] not in POOL_LEVEL_ERROR_CODES
+            ),
+            None,
+        )
+        return (blocking or throttling)["ErrorCode"]
+    return real_errors[0]["ErrorCode"]
+
 
 def _expected_describe_attempts(timeout):
     """Compute DescribeInstances attempts for a never-converging instance, mirroring _get_instances_info."""
@@ -735,6 +771,31 @@ def test_evaluate_launch_params(
                 ],
                 [],
             ),
+            # create-fleet - throttling reported together with one generic error per override
+            (
+                test_on_demand_params,
+                [
+                    MockedBoto3Request(
+                        method="create_fleet",
+                        response={
+                            "Instances": [],
+                            "Errors": [
+                                {"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."},
+                            ]
+                            + [
+                                {
+                                    "ErrorCode": "UnfulfillableCapacity",
+                                    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                                }
+                            ]
+                            * 35,
+                            "ResponseMetadata": {"RequestId": "37633199-bcc6-4a88-89e3-89d859d76096"},
+                        },
+                        expected_params=test_on_demand_params,
+                    ),
+                ],
+                [],
+            ),
         ],
         ids=[
             "fleet_spot",
@@ -743,6 +804,7 @@ def test_evaluate_launch_params(
             "fleet_capacity_block",
             "fleet_throttling",
             "fleet_multiple_errors",
+            "fleet_throttling_with_generic_errors",
         ],
     )
     def test_launch_instances(
@@ -765,10 +827,12 @@ def test_launch_instances(
             with pytest.raises(Exception) as e:
                 fleet_manager._launch_instances(launch_params)
                 assert isinstance(e, ClientError)
-        elif len(expected_assigned_nodes) == 0 and len(mocked_boto3_request[0].response.get("Errors")) == 1:
+        elif not expected_assigned_nodes and _raises_launch_error(mocked_boto3_request[0].response.get("Errors", [])):
             with pytest.raises(LaunchInstancesError) as e:
                 fleet_manager._launch_instances(launch_params)
-                assert isinstance(e, LaunchInstancesError)
+            assert_that(e.value.code).is_equal_to(
+                _expected_launch_error_code(mocked_boto3_request[0].response.get("Errors"))
+            )
         else:
             assigned_nodes = fleet_manager._launch_instances(launch_params)
             assert_that(assigned_nodes.get("Instances", [])).is_equal_to(expected_assigned_nodes)
@@ -1326,3 +1390,138 @@ def test_launch_ec2_instances(self, mocker, count, job_id):
 
         fleet_manager._evaluate_launch_params.assert_called_once_with(count)
         fleet_manager._launch_instances.assert_called_once()
+
+    def test_launch_ec2_instances_retries_on_throttling(self, mocker):
+        """Verify CreateFleet throttling is retried, also when the response carries one error per override."""
+        mocker.patch("time.sleep")
+        fleet_manager = FleetManagerFactory.get_manager(
+            "hit", "region", "boto3_config", FLEET_CONFIG, "queue2", "fleet-ondemand", True, {}, {}
+        )
+        mocker.patch.object(fleet_manager, "_evaluate_launch_params", return_value={})
+        launched_instance_info = {
+            "InstanceId": "i-12345",
+            "PrivateIpAddress": "ip-1",
+            "PrivateDnsName": "hostname",
+            "LaunchTime": datetime(2020, 1, 1, tzinfo=timezone.utc),
+            "NetworkInterfaces": [
+                {"Attachment": {"DeviceIndex": 0, "NetworkCardIndex": 0}, "PrivateIpAddress": "ip-1"},
+            ],
+        }
+        mocker.patch.object(
+            fleet_manager,
+            "_get_instances_info",
+            side_effect=lambda instance_ids: ([launched_instance_info] if instance_ids else [], []),
+        )
+        throttled_response = {
+            "Instances": [],
+            "Errors": [{"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."}]
+            + [
+                {
+                    "ErrorCode": "UnfulfillableCapacity",
+                    "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                }
+            ]
+            * 35,
+            "ResponseMetadata": {"RequestId": "1234-abcde"},
+        }
+        create_fleet = mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            side_effect=[throttled_response, {"Instances": [{"InstanceIds": ["i-12345"]}]}],
+        )
+
+        launched = fleet_manager.launch_ec2_instances(1)
+
+        assert_that(create_fleet.call_count).is_equal_to(2)
+        assert_that(launched).is_length(1)
+
+    def test_launch_ec2_instances_does_not_retry_throttling_with_a_blocking_error(self, mocker):
+        """A cause that fails every pool alike is reported instead of waiting for the rate limit to refill."""
+        mocker.patch("time.sleep")
+        fleet_manager = FleetManagerFactory.get_manager(
+            "hit", "region", "boto3_config", FLEET_CONFIG, "queue2", "fleet-ondemand", True, {}, {}
+        )
+        mocker.patch.object(fleet_manager, "_evaluate_launch_params", return_value={})
+        mocker.patch.object(fleet_manager, "_get_instances_info", return_value=([], []))
+        create_fleet = mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={
+                "Instances": [],
+                "Errors": [THROTTLING_ERROR, VCPU_LIMIT_ERROR] + [UNFULFILLED_OVERRIDE] * 34,
+                "ResponseMetadata": {"RequestId": "1234-abcde"},
+            },
+        )
+
+        with pytest.raises(LaunchInstancesError) as e:
+            fleet_manager.launch_ec2_instances(1)
+
+        assert_that(e.value.code).is_equal_to("VcpuLimitExceeded")
+        assert_that(create_fleet.call_count).is_equal_to(1)
+
+    @pytest.mark.parametrize(
+        ("err_list", "expected_error_code"),
+        [
+            # The real cause is reported even though every other override adds an entry pointing back at it.
+            ([UNSUPPORTED_ERROR] + [UNFULFILLED_OVERRIDE] * 35, "Unsupported"),
+            # The entries pointing at the real cause carry no order guarantee.
+            ([UNFULFILLED_OVERRIDE] * 35 + [UNSUPPORTED_ERROR], "Unsupported"),
+            # UnfulfillableCapacity messages other than that one do describe a cause and are kept.
+            ([MIN_TARGET_CAPACITY_ERROR] + [UNFULFILLED_OVERRIDE] * 35, "UnfulfillableCapacity"),
+            # A single override is unaffected, whatever the entry says.
+            ([UNFULFILLED_OVERRIDE], "UnfulfillableCapacity"),
+            ([MIN_TARGET_CAPACITY_ERROR], "UnfulfillableCapacity"),
+            # Nothing to prefer: report the first entry rather than let a hardcoded code be recorded.
+            ([UNFULFILLED_OVERRIDE] * 36, "UnfulfillableCapacity"),
+            ([UNSUPPORTED_ERROR, {"ErrorCode": "VcpuLimitExceeded", "ErrorMessage": "vCPU limit"}], "Unsupported"),
+            # Throttling is retried when every other cause is confined to a pool: other pools can serve the batch.
+            ([THROTTLING_ERROR, UNSUPPORTED_ERROR] + [UNFULFILLED_OVERRIDE] * 34, "RequestLimitExceeded"),
+            ([THROTTLING_ERROR, SUBNET_FULL_ERROR], "RequestLimitExceeded"),
+            (
+                [THROTTLING_ERROR, {"ErrorCode": "VolumeTypeNotAvailableInZone", "ErrorMessage": "io2"}],
+                "RequestLimitExceeded",
+            ),
+            # Any other cause would fail the retry on every pool alike, so it is reported instead of the throttling.
+            ([THROTTLING_ERROR, VCPU_LIMIT_ERROR] + [UNFULFILLED_OVERRIDE] * 34, "VcpuLimitExceeded"),
+            ([UNFULFILLED_OVERRIDE] * 34 + [VCPU_LIMIT_ERROR, THROTTLING_ERROR], "VcpuLimitExceeded"),
+            ([THROTTLING_ERROR, UNSUPPORTED_ERROR, VCPU_LIMIT_ERROR], "VcpuLimitExceeded"),
+            ([THROTTLING_ERROR, {"ErrorCode": "SomeFutureCode", "ErrorMessage": "?"}], "SomeFutureCode"),
+            # An empty error list is the only case left to the caller, which records insufficient capacity.
+            ([], None),
+        ],
+        ids=[
+            "real_cause_first",
+            "real_cause_last",
+            "min_target_capacity_kept",
+            "single_override_unfulfilled",
+            "single_override_min_target_capacity",
+            "only_unfulfilled_overrides",
+            "two_real_causes",
+            "throttling_retried_with_capacity_error",
+            "throttling_retried_with_subnet_error",
+            "throttling_retried_with_az_volume_type_error",
+            "quota_error_blocks_retry",
+            "quota_error_blocks_retry_any_order",
+            "first_non_pool_error_reported",
+            "unknown_code_blocks_retry",
+            "no_errors_reported",
+        ],
+    )
+    def test_launch_instances_reports_the_cause_among_unfulfilled_overrides(
+        self, mocker, err_list, expected_error_code
+    ):
+        """An entry is added per override that was not fulfilled; only a real cause is worth reporting."""
+        fleet_manager = FleetManagerFactory.get_manager(
+            "hit", "region", "boto3_config", FLEET_CONFIG, "queue2", "fleet-ondemand", False, {}, {}
+        )
+        mocker.patch.object(fleet_manager, "_evaluate_launch_params", return_value={})
+        mocker.patch.object(fleet_manager, "_get_instances_info", return_value=([], []))
+        mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={"Instances": [], "Errors": err_list, "ResponseMetadata": {"RequestId": "1234-abcde"}},
+        )
+
+        if expected_error_code:
+            with pytest.raises(LaunchInstancesError) as e:
+                fleet_manager._launch_instances({})
+            assert_that(e.value.code).is_equal_to(expected_error_code)
+        else:
+            assert_that(fleet_manager._launch_instances({})).is_equal_to({"Instances": []})
diff --git a/tests/slurm_plugin/test_instance_manager.py b/tests/slurm_plugin/test_instance_manager.py
index d7ae8834..f127e446 100644
--- a/tests/slurm_plugin/test_instance_manager.py
+++ b/tests/slurm_plugin/test_instance_manager.py
@@ -4033,6 +4033,69 @@ def test_launch_instances(
         assert_that(instances_launched).is_equal_to(expected_instances_launched)
         assert_that(instance_manager.failed_nodes).is_equal_to(expected_failed_nodes)
 
+    def test_launch_instances_reports_throttling_not_insufficient_capacity(self, mocker, instance_manager):
+        """A throttled CreateFleet must not be recorded as insufficient capacity, which would fail over.
+
+        CreateFleet reports one error per launch template override, so the throttling entry comes with a
+        generic entry for every other override.
+        """
+        mocker.patch("time.sleep")
+        mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={
+                "Instances": [],
+                "Errors": [{"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."}]
+                + [
+                    {
+                        "ErrorCode": "UnfulfillableCapacity",
+                        "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                    }
+                ]
+                * 29,
+                "ResponseMetadata": {"RequestId": "1234-abcde"},
+            },
+        )
+
+        instance_manager._launch_instances(
+            job=None,
+            nodes_to_launch={"queue2": {"fleet-ondemand": ["queue2-dy-fleet-ondemand-1"]}},
+            launch_batch_size=1,
+            scaling_strategy=ScalingStrategy.BEST_EFFORT,
+        )
+
+        assert_that(instance_manager.failed_nodes).is_equal_to({"RequestLimitExceeded": {"queue2-dy-fleet-ondemand-1"}})
+
+    def test_launch_instances_reports_blocking_error_instead_of_retrying_throttling(self, mocker, instance_manager):
+        """A cause that fails every pool alike is recorded instead of waiting for the rate limit to refill."""
+        mocker.patch("time.sleep")
+        mocker.patch(
+            "slurm_plugin.fleet_manager.create_fleet",
+            return_value={
+                "Instances": [],
+                "Errors": [
+                    {"ErrorCode": "RequestLimitExceeded", "ErrorMessage": "Request limit exceeded."},
+                    {"ErrorCode": "VcpuLimitExceeded", "ErrorMessage": "vCPU limit"},
+                ]
+                + [
+                    {
+                        "ErrorCode": "UnfulfillableCapacity",
+                        "ErrorMessage": "Failed to fulfill capacity. Please review errors in the response.",
+                    }
+                ]
+                * 28,
+                "ResponseMetadata": {"RequestId": "1234-abcde"},
+            },
+        )
+
+        instance_manager._launch_instances(
+            job=None,
+            nodes_to_launch={"queue2": {"fleet-ondemand": ["queue2-dy-fleet-ondemand-1"]}},
+            launch_batch_size=1,
+            scaling_strategy=ScalingStrategy.BEST_EFFORT,
+        )
+
+        assert_that(instance_manager.failed_nodes).is_equal_to({"VcpuLimitExceeded": {"queue2-dy-fleet-ondemand-1"}})
+
     @pytest.mark.parametrize(
         "job_list, launch_batch_size, assign_node_batch_size, update_node_address, "
         "expected_single_nodes_no_oversubscribe, scaling_strategy",
diff --git a/tox.ini b/tox.ini
index 6e19dc84..6199ecd4 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,6 +1,6 @@
 [tox]
 envlist =
-    py{39,310,311,312,313}-cov
+    py{314}-cov
     code-linters
 
 # Default testenv. Used to run tests on all python versions.