fix(svc-bkp-nfs-2-local): leave backup-disabled volumes out of the export snapshot

The NFS backup rsynced the whole export and honoured only one exclude, the
shared backup root. Volumes declared `backup: false` were copied anyway and
then pulled a second time by remote-2-local; in the swarm DR drill of
Magento#0 (run 40625679270) docs_sites alone (a full repo clone plus its
prepared and work trees) took about 15.5 GiB per copy and filled the
runner. The volume backup already skipped these through the backup_volume
lookup; only the NFS path ignored the declaration.

- utils/storage/nfs.py: volume_dir() names the state directory a volume
  binds to (pinned name or key), shared with compose_volumes;
  backup_disabled_dirs() returns every NFS-backed `backup: false` volume
  dir that no backed-up volume also binds to.
- The role-local lookup backup_nfs_excludes feeds them into
  BKP_NFS_2_LOCAL_EXCLUDES next to the backup root; script.sh takes any
  number of excludes. Compose installs still pass only the backup root.
- recover.py protects every live state directory the snapshot does not
  hold, so `rsync --delete` keeps the excluded volumes (model weights,
  the docs tree, caches) and volumes created after the snapshot instead
  of wiping them.
- The swarm DR drill restores BACKUP_NFS_VOLUME, the first NFS volume
  not declared `backup: false`; PRIMARY_NFS_VOLUME stays for the drain
  tests. BookWyrm now drills bookwyrm_media; decidim, fusiondirectory,
  meta, coturn, svc-registry-cache and svc-registry-docker have nothing
  backed up on NFS and skip the drill.

Measured: compose_volumes renders identically before and after for 291
roles across compose/swarm x local/nfs (1940 of 1940); the 15 excluded
directories are exactly the backup-disabled device dirs; the systemd unit
renders shell-safe on ansible-core 2.20 and 2.21 and stays byte-identical
in compose and swarm-local. A new unit test runs the real script.sh over
two generations and the real recover.py back into the export, and fails
without the restore protection.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/lookup/compose_volumes.py

@@ -32,7 +32,7 @@ from utils.roles.applications.services.database import (
     resolve_database_service_key,
 )
 from utils.roles.applications.services.sso import get_sso_config
-from utils.storage.nfs import swarm_nfs_backed
+from utils.storage.nfs import swarm_nfs_backed, volume_dir
 from utils.templating.ansible import _trust_as_template, to_plain
 
 
@@ -267,8 +267,9 @@ def compose_volumes(
             deployment_mode=deployment_mode,
             storage_backend=storage_backend,
         ):
-            named = vol_spec.get("name", vol_name)
-            vol_spec.update(_swarm_nfs_driver_opts(dir_var_lib, str(named)))
+            vol_spec.update(
+                _swarm_nfs_driver_opts(dir_var_lib, volume_dir(vol_name, vol_spec))
+            )
             if isinstance(nfs_meta, dict):
                 vol_spec["x-infinito-nfs"] = dict(nfs_meta)
 

roles/svc-bkp-nfs-2-local/README.md

@@ -18,9 +18,9 @@ flowchart TD
     TIMER["systemd timer<br>SYS_SCHEDULE_BACKUP_NFS_TO_LOCAL (01:30)"] --> UNIT
     PRELOAD["sys-service-loader preload<br>MODE_BACKUP, run_after svc-storage-nfs-server<br>(force_flush_instant + state started)"] --> UNIT
     UNIT["svc-bkp-nfs-2-local.&lt;version&gt;.&lt;domain&gt;.service"] --> LOCK["ExecStartPre: sys-lock against the manipulation group"]
-    LOCK --> SCRIPT["ExecStart: script.sh &lt;export_base&gt; &lt;backups_dir&gt;<br>backup-nfs-to-local &lt;state&gt;/backup"]
+    LOCK --> SCRIPT["ExecStart: script.sh &lt;export_base&gt; &lt;backups_dir&gt;<br>backup-nfs-to-local &lt;state&gt;/backup &lt;state&gt;/&lt;backup-disabled volume&gt;..."]
     SCRIPT -->|export_base missing| ALARM["ERROR exit 1<br>OnFailure alarm"]
-    SCRIPT --> RSYNC["rsync -aHAX --delete<br>--exclude /&lt;state&gt;/backup (no backups-in-backups)<br>--link-dest previous generation (hard links)"]
+    SCRIPT --> RSYNC["rsync -aHAX --delete<br>--exclude /&lt;state&gt;/backup (no backups-in-backups)<br>--exclude /&lt;state&gt;/&lt;volume&gt; per backup: false volume<br>--link-dest previous generation (hard links)"]
     RSYNC -->|exit 24: live export churn| WARN["WARN, snapshot kept"]
     RSYNC -->|any other failure| CLEAN["generation removed, exit != 0"]
     RSYNC --> TREE["&lt;backups_dir&gt;/&lt;sha256(machine-id)&gt;/<br>backup-nfs-to-local/&lt;YYYYmmddHHMMSS&gt;/files/..."]
@@ -30,6 +30,7 @@ flowchart TD
 ## Features
 
 - **Differential snapshots:** rsync `--link-dest` against the previous generation deduplicates unchanged files.
+- **Backup-disabled volumes skipped:** the directory of every NFS-backed volume declared `backup: false` in its role's `meta/volumes.yml` is excluded, the same declaration [svc-bkp-volume-2-local](../svc-bkp-volume-2-local/) honours; a directory a backed-up volume also binds to stays in the snapshot.
 - **Baudolo-compatible layout:** snapshots land in the same `<machine-hash>/<repo>/<generation>` tree as the container backups, so pull and cleanup tooling applies unchanged.
 - **Schedule-coordinated:** the systemd unit is part of the global manipulation group, so it never races backup/cleanup/repair jobs on the same host.
 

roles/svc-bkp-nfs-2-local/files/python/recover.py

@@ -23,16 +23,42 @@ from pathlib import Path
 _REPO_ROOT = Path(__file__).resolve().parents[4]  # nocheck: project-root-import
 sys.path.insert(0, str(_REPO_ROOT))
 
-from utils.recovery.base import DirectoryRecovery  # noqa: E402
+from utils.recovery.base import DirectoryRecovery, is_remote  # noqa: E402
 from utils.storage.nfs import STATE_SUBDIR  # noqa: E402
 
 
 class NfsExportRecovery(DirectoryRecovery):
     unit_pattern = "svc-bkp-nfs-2-local*.service"
-    rsync_extra_args = (
-        f"--filter=protect /{STATE_SUBDIR}/backup",
-        f"--exclude=/{STATE_SUBDIR}/backup",
-    )
+
+    @property
+    def rsync_extra_args(self) -> tuple[str, ...]:
+        return (
+            f"--filter=protect /{STATE_SUBDIR}/backup",
+            f"--exclude=/{STATE_SUBDIR}/backup",
+            *(
+                f"--filter=protect /{STATE_SUBDIR}/{name}"
+                for name in self.volumes_missing_from_snapshot()
+            ),
+        )
+
+    def volumes_missing_from_snapshot(self) -> list[str]:
+        """Volume directories of the live export the snapshot does not hold.
+
+        Returns:
+            The sorted names under the state path; empty when either side is
+            remote and cannot be listed.
+        """
+        if is_remote(str(self.source_dir)) or is_remote(str(self.target_dir)):
+            return []
+        live = self.target_dir / STATE_SUBDIR
+        if not live.is_dir():
+            return []
+        kept = self.source_dir / STATE_SUBDIR
+        return sorted(
+            entry.name
+            for entry in live.iterdir()
+            if entry.is_dir() and not (kept / entry.name).exists()
+        )
 
 
 def main() -> int:

roles/svc-bkp-nfs-2-local/files/shell/script.sh

@@ -6,19 +6,18 @@
 #   $1 SOURCE_DIR   NFS export base to back up
 #   $2 BACKUPS_DIR  local backup root
 #   $3 REPO_NAME    repository directory name inside the machine hash dir
-#   $4 EXCLUDE_REL  optional path relative to SOURCE_DIR to exclude
-#                   (the shared backup root inside the export; without it
-#                   every generation snapshots all previous backups)
+#   $4.. EXCLUDE_REL  optional paths relative to SOURCE_DIR to exclude
 #
 # Environment:
 #   BKP_NFS_2_LOCAL_GENERATION  optional generation name override
 #                               (defaults to the current timestamp)
 set -euo pipefail
 
-SOURCE_DIR="${1:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL]}"
-BACKUPS_DIR="${2:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL]}"
-REPO_NAME="${3:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL]}"
-EXCLUDE_REL="${4:-}"
+SOURCE_DIR="${1:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL...]}"
+BACKUPS_DIR="${2:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL...]}"
+REPO_NAME="${3:?usage: script.sh SOURCE_DIR BACKUPS_DIR REPO_NAME [EXCLUDE_REL...]}"
+shift 3
+EXCLUDE_RELS=("$@")
 
 if [[ ! -d "${SOURCE_DIR}" ]]; then
     echo "ERROR: ${SOURCE_DIR} missing; this host is expected to serve the NFS export" >&2
@@ -46,9 +45,11 @@ trap cleanup_failed_generation ERR
 mkdir -p "${DEST_DIR}"
 
 rsync_args=(-aHAX --numeric-ids --delete)
-if [[ -n "${EXCLUDE_REL}" ]]; then
-    rsync_args+=(--exclude "/${EXCLUDE_REL#/}")
-fi
+for exclude_rel in "${EXCLUDE_RELS[@]}"; do
+    if [[ -n "${exclude_rel}" ]]; then
+        rsync_args+=(--exclude "/${exclude_rel#/}")
+    fi
+done
 if [[ -n "${PREVIOUS_FILES}" ]]; then
     rsync_args+=(--link-dest "${PREVIOUS_FILES}")
 fi

roles/svc-bkp-nfs-2-local/lookup_plugins/backup_nfs_excludes.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import contextlib
+from collections.abc import Mapping
+from typing import Any
+
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from utils.storage.nfs import STATE_SUBDIR, backup_disabled_dirs
+
+
+class LookupModule(LookupBase):
+    """Export-relative directories of the NFS-backed volumes declared
+    ``backup: false``, for the export rsync's ``--exclude`` list.
+
+    Args:
+        terms: none.
+        variables: the play variables; ``DEPLOYMENT_MODE`` and ``storage``
+            decide which volumes the swarm NFS rewrite puts on the export, and
+            the live merged ``applications`` config is read through them.
+
+    Returns:
+        One list: the sorted ``<state subdir>/<volume dir>`` paths.
+    """
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[list[str]]:
+        vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
+        templar = getattr(self, "_templar", None)
+
+        applications = lookup_loader.get(
+            "applications", loader=self._loader, templar=templar
+        ).run([], variables=vars_)[0]
+
+        deployment_mode = vars_.get("DEPLOYMENT_MODE", "")
+        storage = vars_.get("storage")
+        if templar is not None:
+            with contextlib.suppress(Exception):
+                deployment_mode = templar.template(deployment_mode)
+            with contextlib.suppress(Exception):
+                storage = templar.template(storage)
+        storage_backend = ""
+        if isinstance(storage, Mapping):
+            storage_backend = str(storage.get("backend") or "")
+
+        if not isinstance(applications, Mapping):
+            return [[]]
+        dirs = backup_disabled_dirs(
+            applications,
+            deployment_mode=str(deployment_mode or ""),
+            storage_backend=storage_backend,
+        )
+        return [[f"{STATE_SUBDIR}/{name}" for name in dirs]]

roles/svc-bkp-nfs-2-local/templates/systemctl.service.j2

@@ -5,4 +5,4 @@ OnFailure={{ SYS_SERVICE_ON_FAILURE_COMPOSE }} {{ SYS_SERVICE_CLEANUP_BACKUPS_FA
 [Service]
 Type=oneshot
 ExecStartPre={{ BIN_LOCK }} {{ SYS_SERVICE_GROUP_MANIPULATION | join(' ') }} --ignore {{ SYS_SERVICE_BACKUP_NFS_2_LOCAL }} --timeout "{{ SYS_TIMEOUT_BACKUP_SERVICES }}"
-ExecStart={{ system_service_script_exec }} {{ BKP_NFS_2_LOCAL_SOURCE }} {{ DIR_BACKUPS }} {{ BKP_NFS_2_LOCAL_REPO_NAME }} {{ BKP_NFS_2_LOCAL_EXCLUDE }}
+ExecStart={{ system_service_script_exec }} {{ BKP_NFS_2_LOCAL_SOURCE }} {{ DIR_BACKUPS }} {{ BKP_NFS_2_LOCAL_REPO_NAME }} {{ BKP_NFS_2_LOCAL_EXCLUDES | join(' ') }}

roles/svc-bkp-nfs-2-local/vars/main.yml

@@ -4,4 +4,4 @@ system_service_id: "{{ application_id }}"
 
 BKP_NFS_2_LOCAL_SOURCE: "{{ lookup('nfs_export_base') }}"
 BKP_NFS_2_LOCAL_REPO_NAME: backup-nfs-to-local
-BKP_NFS_2_LOCAL_EXCLUDE: "{{ [lookup('nfs_state_path') | basename, DIR_BACKUPS | basename] | path_join }}"
+BKP_NFS_2_LOCAL_EXCLUDES: "{{ [[lookup('nfs_state_path') | basename, DIR_BACKUPS | basename] | path_join] + lookup('backup_nfs_excludes') }}"

scripts/tests/deploy/swarm/routine/backup/README.md

@@ -103,9 +103,9 @@ flowchart TB
 
 ## Applicability
 
-- The drill runs whenever the app declares an NFS-flagged volume
-  (`PRIMARY_NFS_VOLUME`); apps without one skip it, since there is nothing
-  to prove a restore against.
+- The drill runs whenever the app declares an NFS-flagged volume that is
+  not `backup: false` (`BACKUP_NFS_VOLUME`); apps without one skip it,
+  since there is nothing to prove a restore against.
 - The manager legs of steps 2, 4 and 9 run only when the app's include
   closure contains `svc-bkp-volume-2-local` / `svc-bkp-secrets-2-local`.
   `utils/tests/swarm/backup_repos.py` derives both the manager's inventory

scripts/tests/deploy/swarm/routine/backup/base.sh

@@ -50,7 +50,7 @@ DEV_TARGET="$(python3 -c "import sys, yaml; print(yaml.safe_load(open(sys.argv[1
 DEV_DEST="${DEV_MOUNT}${DEV_TARGET}"
 
 if [ "${DRILL_PROBE:-false}" = "true" ]; then
-	if [ -n "${PRIMARY_NFS_VOLUME}" ]; then
+	if [ -n "${BACKUP_NFS_VOLUME}" ]; then
 		echo "DRILL=yes"
 	else
 		echo "DRILL=no"
@@ -58,8 +58,8 @@ if [ "${DRILL_PROBE:-false}" = "true" ]; then
 	exit 0
 fi
 
-if [ -z "${PRIMARY_NFS_VOLUME}" ]; then
-	echo "SKIP drill: ${APP_ID} declares no NFS-flagged volume — nothing to prove a restore against"
+if [ -z "${BACKUP_NFS_VOLUME}" ]; then
+	echo "SKIP drill: ${APP_ID} declares no backed-up NFS-flagged volume — nothing to prove a restore against"
 	exit 0
 fi
 if ! docker inspect "${BACKUP_NODE}" >/dev/null 2>&1; then
@@ -71,10 +71,10 @@ if [ ! -f "${BACKUP_KEY_PATH}" ]; then
 	exit 1
 fi
 
-NFS_VOL_DIR="${NFS_STATE_PATH}/${PRIMARY_NFS_VOLUME}"
+NFS_VOL_DIR="${NFS_STATE_PATH}/${BACKUP_NFS_VOLUME}"
 MGR_MID="$(docker exec "${MGR}" sha256sum /etc/machine-id | cut -c1-64)"
 NFS_MID="$(docker exec "${NFS_SERVER}" sha256sum /etc/machine-id | cut -c1-64)"
-echo "==> DR drill for ${APP_ID} (volume '${PRIMARY_NFS_VOLUME}')"
+echo "==> DR drill for ${APP_ID} (volume '${BACKUP_NFS_VOLUME}')"
 
 TRIGGER_UNITS="${NODE_SRC}/scripts/tests/deploy/swarm/utils/trigger_units.sh"
 UNIT_DUMPS="${INFINITO_RESCUE_DIAGNOSTICS_DIR:?INFINITO_RESCUE_DIAGNOSTICS_DIR is not set - source scripts/meta/env/load.sh first}"
@@ -117,11 +117,11 @@ fi
 
 echo "==> [3/9] locate the backup generation holding the marker"
 SRC_HOST=""
-MARKER_PATH="$(docker exec "${NFS_SERVER}" find "${DIR_BACKUPS}/${NFS_MID}/${NFS_REPO}" -type f -name "${DR_MARKER}" -path "*/files/*/${PRIMARY_NFS_VOLUME}/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
+MARKER_PATH="$(docker exec "${NFS_SERVER}" find "${DIR_BACKUPS}/${NFS_MID}/${NFS_REPO}" -type f -name "${DR_MARKER}" -path "*/files/*/${BACKUP_NFS_VOLUME}/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
 if [ -n "${MARKER_PATH}" ]; then
 	SRC_HOST="${NFS_SERVER}"
 else
-	MARKER_PATH="$(docker exec "${MGR}" find "${DIR_BACKUPS}/${MGR_MID}/${VOLUME_REPO}" -type f -name "${DR_MARKER}" -path "*/${PRIMARY_NFS_VOLUME}/files/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
+	MARKER_PATH="$(docker exec "${MGR}" find "${DIR_BACKUPS}/${MGR_MID}/${VOLUME_REPO}" -type f -name "${DR_MARKER}" -path "*/${BACKUP_NFS_VOLUME}/files/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
 	if [ -n "${MARKER_PATH}" ]; then
 		SRC_HOST="${MGR}"
 	fi
@@ -137,7 +137,7 @@ SRC_REL="$(dirname "${MARKER_REL}")"
 echo "    marker captured on ${SRC_HOST} at ${MARKER_REL}"
 VOL_MARKER_REL=""
 if [ "${SRC_HOST}" != "${MGR}" ]; then
-	_vol_marker="$(docker exec "${MGR}" find "${DIR_BACKUPS}/${MGR_MID}/${VOLUME_REPO}" -type f -name "${DR_MARKER}" -path "*/${PRIMARY_NFS_VOLUME}/files/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
+	_vol_marker="$(docker exec "${MGR}" find "${DIR_BACKUPS}/${MGR_MID}/${VOLUME_REPO}" -type f -name "${DR_MARKER}" -path "*/${BACKUP_NFS_VOLUME}/files/${DR_MARKER}" 2>/dev/null | sort | tail -1 || true)"
 	[ -n "${_vol_marker}" ] && VOL_MARKER_REL="${_vol_marker#"${DIR_BACKUPS}"/}"
 fi
 
@@ -285,7 +285,7 @@ for _node in "${MGR}" "${WRK1}" "${WRK2}"; do
 done
 for _i in $(seq 1 24); do
 	if docker exec "${MGR}" sh -c \
-		"touch '${DIR_VAR_LIB}/${PRIMARY_NFS_VOLUME}/.dr-coherence-probe' && rm -f '${DIR_VAR_LIB}/${PRIMARY_NFS_VOLUME}/.dr-coherence-probe'" 2>/dev/null; then
+		"touch '${DIR_VAR_LIB}/${BACKUP_NFS_VOLUME}/.dr-coherence-probe' && rm -f '${DIR_VAR_LIB}/${BACKUP_NFS_VOLUME}/.dr-coherence-probe'" 2>/dev/null; then
 		break
 	fi
 	if [ "${_i}" -eq 24 ]; then
@@ -317,12 +317,12 @@ if [ -n "${VOL_MARKER_REL}" ]; then
 		docker exec "${MGR}" rm -rf "${DR_VOL_STAGE:?}"
 	fi
 else
-	VOL_PATH="$(docker exec "${MGR}" docker volume inspect --format '{{if .Options}}{{.Options.device}}{{end}}' "${PRIMARY_NFS_VOLUME}" 2>/dev/null || true)"
+	VOL_PATH="$(docker exec "${MGR}" docker volume inspect --format '{{if .Options}}{{.Options.device}}{{end}}' "${BACKUP_NFS_VOLUME}" 2>/dev/null || true)"
 	if [ -z "${VOL_PATH}" ]; then
-		echo "    volume recover skipped: ${MGR} exposes no backing store for '${PRIMARY_NFS_VOLUME}' - the volume object is node-local and lives wherever the task ran, and the [8/9] export restore already carried its data"
+		echo "    volume recover skipped: ${MGR} exposes no backing store for '${BACKUP_NFS_VOLUME}' - the volume object is node-local and lives wherever the task ran, and the [8/9] export restore already carried its data"
 	else
 		echo "    volume repo holds no marker by design; proving the volume recover CLI against the bound backing store instead"
-		VOL_MOUNTPOINT="$(docker exec "${MGR}" docker volume inspect --format '{{.Mountpoint}}' "${PRIMARY_NFS_VOLUME}")"
+		VOL_MOUNTPOINT="$(docker exec "${MGR}" docker volume inspect --format '{{.Mountpoint}}' "${BACKUP_NFS_VOLUME}")"
 		docker exec "${NFS_SERVER}" rm -f "${NFS_VOL_DIR}/${DR_MARKER}"
 		DR_VOL_STAGE="/var/tmp/dr-volume-restore"
 		docker exec "${MGR}" bash -c "rm -rf '${DR_VOL_STAGE}'; mkdir -p '${DR_VOL_STAGE}'"
@@ -337,14 +337,14 @@ else
 			echo "WARNING: the bind mount at ${VOL_MOUNTPOINT} refused the umount and stays behind"
 		docker exec "${MGR}" rm -rf "${DR_VOL_STAGE:?}"
 		if [ "${_vol_rc}" -ne 0 ]; then
-			echo "FAILURE: the volume recover CLI exited ${_vol_rc} restoring '${PRIMARY_NFS_VOLUME}' through its bound backing store"
+			echo "FAILURE: the volume recover CLI exited ${_vol_rc} restoring '${BACKUP_NFS_VOLUME}' through its bound backing store"
 			exit 1
 		fi
 		if ! docker exec "${NFS_SERVER}" test -f "${NFS_VOL_DIR}/${DR_MARKER}"; then
 			echo "FAILURE: the volume recover CLI reported success but the marker did not reach the export on ${NFS_SERVER} (expected ${NFS_VOL_DIR}/${DR_MARKER})"
 			exit 1
 		fi
-		echo "    volume '${PRIMARY_NFS_VOLUME}' recovered through its bound backing store; marker verified server-side"
+		echo "    volume '${BACKUP_NFS_VOLUME}' recovered through its bound backing store; marker verified server-side"
 	fi
 fi
 

scripts/tests/deploy/swarm/utils/_context.sh

@@ -67,8 +67,9 @@ print(port)
 ")"
 
 NFS_VOLUMES=""
+BACKUP_NFS_VOLUMES=""
 if [ -f "${ROLE_DIR}/meta/volumes.yml" ]; then
-	NFS_VOLUMES="$(python3 -c "
+	_nfs_volumes_py="
 import sys, yaml
 with open('${ROLE_DIR}/meta/volumes.yml') as f:
     data = yaml.safe_load(f) or {}
@@ -83,17 +84,22 @@ for v in entries:
         continue
     if v.get('nfs') is False:
         continue
+    if sys.argv[1:] == ['backed-up'] and v.get('backup') is False:
+        continue
     docker_name = v.get('docker_name') or v.get('name')
     if isinstance(docker_name, str):
         print(docker_name)
-")"
+"
+	NFS_VOLUMES="$(python3 -c "${_nfs_volumes_py}")"
+	BACKUP_NFS_VOLUMES="$(python3 -c "${_nfs_volumes_py}" backed-up)"
 fi
 
 PRIMARY_NFS_VOLUME="${NFS_VOLUMES%%$'\n'*}"
+BACKUP_NFS_VOLUME="${BACKUP_NFS_VOLUMES%%$'\n'*}"
 
 NFS_CHECK_MOUNTPOINT="/mnt/nfs-check"
 
-export APP_ID ENTITY STACK_NAME SERVICE_NAME PRIMARY_SERVICE_KEY CUSTOM_IMAGE_REPO DB_DEP NFS_VOLUMES PRIMARY_NFS_VOLUME DEFAULT_PLACEMENT_MANAGER PROBE_PORT NFS_CHECK_MOUNTPOINT
+export APP_ID ENTITY STACK_NAME SERVICE_NAME PRIMARY_SERVICE_KEY CUSTOM_IMAGE_REPO DB_DEP NFS_VOLUMES PRIMARY_NFS_VOLUME BACKUP_NFS_VOLUME DEFAULT_PLACEMENT_MANAGER PROBE_PORT NFS_CHECK_MOUNTPOINT
 
 # Exits the calling chaos step (09/10/11) with success when the role is
 # manager-pinned (no worker task exists to drain).
@@ -172,6 +178,7 @@ if [ "${SWARM_NFS_PILOT_VERBOSE:-0}" = "1" ]; then
 	echo "    CUSTOM_IMAGE_REPO=${CUSTOM_IMAGE_REPO}"
 	echo "    DB_DEP=${DB_DEP}"
 	echo "    PRIMARY_NFS_VOLUME=${PRIMARY_NFS_VOLUME}"
+	echo "    BACKUP_NFS_VOLUME=${BACKUP_NFS_VOLUME}"
 	echo "    DEFAULT_PLACEMENT_MANAGER=${DEFAULT_PLACEMENT_MANAGER}"
 	echo "    PROBE_PORT=${PROBE_PORT}"
 fi

tests/unit/python/roles/svc-bkp-nfs-2-local/files/test_script.py

@@ -6,6 +6,7 @@ into older generations."""
 
 import os
 import subprocess
+import sys
 import tempfile
 import unittest
 from pathlib import Path
@@ -15,6 +16,9 @@ from . import PROJECT_ROOT
 SCRIPT = (
     PROJECT_ROOT / "roles" / "svc-bkp-nfs-2-local" / "files" / "shell" / "script.sh"
 )
+RECOVER = (
+    PROJECT_ROOT / "roles" / "svc-bkp-nfs-2-local" / "files" / "python" / "recover.py"
+)
 REPO_NAME = "backup-nfs-to-local"
 
 
@@ -30,7 +34,7 @@ class NfsBackupScriptTests(unittest.TestCase):
     def tearDown(self):
         self.tmp.cleanup()
 
-    def _run(self, source=None, generation=None, exclude=None):
+    def _run(self, source=None, generation=None, excludes=()):
         env = os.environ.copy()
         if generation is not None:
             env["BKP_NFS_2_LOCAL_GENERATION"] = generation
@@ -40,9 +44,8 @@ class NfsBackupScriptTests(unittest.TestCase):
             str(source or self.source),
             str(self.backups),
             REPO_NAME,
+            *excludes,
         ]
-        if exclude is not None:
-            cmd.append(exclude)
         return subprocess.run(
             cmd,
             capture_output=True,
@@ -132,12 +135,12 @@ class NfsBackupScriptTests(unittest.TestCase):
         (decoy / "decoy.txt").write_text("decoy")
         (self.source / "infinito-state" / "state.txt").write_text("state")
 
-        exclude = "infinito-state/backup"
+        excludes = ["infinito-state/backup"]
         self.assertEqual(
-            self._run(generation="20240101000000", exclude=exclude).returncode, 0
+            self._run(generation="20240101000000", excludes=excludes).returncode, 0
         )
         self.assertEqual(
-            self._run(generation="20240101000001", exclude=exclude).returncode, 0
+            self._run(generation="20240101000001", excludes=excludes).returncode, 0
         )
 
         for generation in self._generations():
@@ -148,6 +151,73 @@ class NfsBackupScriptTests(unittest.TestCase):
                 (files / "app" / "infinito-state" / "backup" / "decoy.txt").is_file()
             )
 
+    def test_every_excluded_volume_dir_stays_out_of_the_snapshot(self):
+        state = self.source / "infinito-state"
+        for name in ("backup", "docs_sites", "nextcloud_whiteboard_tmp"):
+            (state / name).mkdir(parents=True)
+            (state / name / "blob").write_text(name)
+        (state / "nextcloud_data").mkdir()
+        (state / "nextcloud_data" / "user.txt").write_text("user")
+
+        result = self._run(
+            excludes=[
+                "infinito-state/backup",
+                "infinito-state/docs_sites",
+                "infinito-state/nextcloud_whiteboard_tmp",
+            ]
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+        files = self._generations()[0] / "files" / "infinito-state"
+        self.assertEqual(sorted(p.name for p in files.iterdir()), ["nextcloud_data"])
+        self.assertTrue((files / "nextcloud_data" / "user.txt").is_file())
+
+    def test_restoring_a_generation_spares_excluded_volumes_and_the_backup_root(self):
+        state = self.source / "infinito-state"
+        self.backups = state / "backup"
+        self.backups.mkdir(parents=True)
+        db_file = state / "app_data" / "db"
+        site_file = state / "docs_sites" / "site"
+        db_file.parent.mkdir()
+        site_file.parent.mkdir()
+        db_file.write_text("v1")
+        site_file.write_text("regenerable")
+        excludes = ["infinito-state/backup", "infinito-state/docs_sites"]
+        for generation in ("20240101000000", "20240101000001"):
+            result = self._run(generation=generation, excludes=excludes)
+            self.assertEqual(result.returncode, 0, result.stderr)
+        first, second = (g / "files" / "infinito-state" for g in self._generations())
+        self.assertEqual(sorted(p.name for p in second.iterdir()), ["app_data"])
+        self.assertEqual(
+            (first / "app_data" / "db").stat().st_ino,
+            (second / "app_data" / "db").stat().st_ino,
+        )
+
+        db_file.write_text("corrupted")
+        (state / "app_data" / "stale").write_text("stale")
+        site_file.write_text("rebuilt")
+        result = subprocess.run(
+            [
+                sys.executable,
+                str(RECOVER),
+                str(second.parent),
+                str(self.source),
+                "--no-safety-backup",
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+
+        db = db_file.read_text()  # nocheck: cache-read tempdir fixture
+        site = site_file.read_text()  # nocheck: cache-read tempdir fixture
+        self.assertEqual(db, "v1")
+        self.assertEqual(site, "rebuilt")
+        self.assertFalse((state / "app_data" / "stale").exists())
+        self.assertTrue((first / "app_data" / "db").is_file())
+        self.assertTrue((second / "app_data" / "db").is_file())
+
     def test_deleted_file_vanishes_only_from_new_generation(self):
         (self.source / "keep.txt").write_text("keep")
         (self.source / "gone.txt").write_text("gone")

tests/unit/python/roles/svc-bkp-nfs-2-local/lookup_plugins/test_backup_nfs_excludes.py

@@ -0,0 +1,138 @@
+import importlib.util
+import unittest
+import unittest.mock as mock
+from pathlib import Path
+from types import ModuleType
+from typing import ClassVar
+
+PLUGIN = (
+    Path(__file__).resolve().parents[6]
+    / "roles"
+    / "svc-bkp-nfs-2-local"
+    / "lookup_plugins"
+    / "backup_nfs_excludes.py"
+)
+
+
+def _load_module() -> ModuleType:
+    spec = importlib.util.spec_from_file_location("backup_nfs_excludes", PLUGIN)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def _run(
+    module: ModuleType,
+    applications: dict,
+    variables: dict,
+    templar: object | None = None,
+) -> list[str]:
+    loader = mock.MagicMock()
+    loader.get.return_value.run.return_value = [applications]
+    with mock.patch.object(module, "lookup_loader", loader):
+        plugin = module.LookupModule()
+        if templar is not None:
+            plugin._templar = templar
+        return plugin.run(None, variables=variables)[0]
+
+
+class TestBackupNfsExcludes(unittest.TestCase):
+    SWARM_NFS: ClassVar[dict] = {
+        "DEPLOYMENT_MODE": "swarm",
+        "storage": {"backend": "nfs"},
+    }
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls._module = _load_module()
+
+    def _excludes(self, applications, variables=None, templar=None):
+        return _run(
+            type(self)._module,
+            applications,
+            dict(self.SWARM_NFS) if variables is None else variables,
+            templar,
+        )
+
+    def test_backup_disabled_volume_dirs_are_excluded(self) -> None:
+        apps = {
+            "web-app-docs": {
+                "volumes": {"docs_sites": {"type": "volume", "backup": False}}
+            },
+            "web-app-nextcloud": {
+                "volumes": {
+                    "data": {"type": "volume", "name": "nextcloud_data"},
+                    "whiteboard_tmp": {
+                        "type": "volume",
+                        "name": "nextcloud_whiteboard_tmp",
+                        "backup": False,
+                    },
+                }
+            },
+        }
+        self.assertEqual(
+            self._excludes(apps),
+            [
+                "infinito-state/docs_sites",
+                "infinito-state/nextcloud_whiteboard_tmp",
+            ],
+        )
+
+    def test_the_export_dir_is_the_semantic_key_not_the_docker_name(self) -> None:
+        apps = {"web-app-docs": {"volumes": {"docs_sites": {"backup": False}}}}
+        self.assertNotIn("infinito-state/docs_docs_sites", self._excludes(apps))
+
+    def test_compose_mode_excludes_nothing(self) -> None:
+        apps = {"web-app-docs": {"volumes": {"docs_sites": {"backup": False}}}}
+        variables = {"DEPLOYMENT_MODE": "compose", "storage": {"backend": "nfs"}}
+        self.assertEqual(self._excludes(apps, variables), [])
+
+    def test_a_volume_off_the_export_is_not_excluded(self) -> None:
+        apps = {
+            "web-svc-xmpp": {
+                "volumes": {
+                    "home": {"name": "ejabberd_home", "backup": False, "nfs": False}
+                }
+            },
+            "web-svc-seaweedfs": {
+                "volumes": {"data": {"name": "seaweedfs_data", "backup": False}}
+            },
+        }
+        self.assertEqual(self._excludes(apps), [])
+
+    def test_a_dir_a_backed_up_volume_also_binds_stays_captured(self) -> None:
+        apps = {
+            "web-app-docs": {"volumes": {"shared": {"backup": False}}},
+            "web-app-nextcloud": {"volumes": {"shared": {"type": "volume"}}},
+        }
+        self.assertEqual(self._excludes(apps), [])
+
+    def test_a_config_entry_is_never_excluded(self) -> None:
+        apps = {
+            "web-app-nextcloud": {
+                "volumes": {"loader": {"type": "config", "backup": False}}
+            }
+        }
+        self.assertEqual(self._excludes(apps), [])
+
+    def test_the_deployment_mode_expression_is_templated(self) -> None:
+        expression = (
+            "{{ 'swarm' if (groups['svc-swarm-node'] | default([]) | length) > 1"
+            " else 'compose' }}"
+        )
+        templar = mock.Mock()
+        templar.template.side_effect = lambda value: (
+            "swarm" if value == expression else value
+        )
+        apps = {"web-app-docs": {"volumes": {"docs_sites": {"backup": False}}}}
+        self.assertEqual(
+            self._excludes(
+                apps, {**self.SWARM_NFS, "DEPLOYMENT_MODE": expression}, templar
+            ),
+            ["infinito-state/docs_sites"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/utils/recovery/test_role_recover_scripts.py

@@ -188,6 +188,27 @@ class NfsRecoverTests(unittest.TestCase):
             )
             self.assertFalse((target / "stale.txt").exists())
 
+    def test_a_volume_the_snapshot_does_not_hold_survives_the_restore(self):
+        mod = _load("svc-bkp-nfs-2-local", "nfs_recover")
+        state = mod.STATE_SUBDIR
+        with tempfile.TemporaryDirectory() as td:
+            source = Path(td) / "source"
+            target = Path(td) / "target"
+            (source / state / "app_data").mkdir(parents=True)
+            (source / state / "app_data" / "db").write_text("from-snapshot")
+            (target / state / "app_data").mkdir(parents=True)
+            (target / state / "app_data" / "stale").write_text("pre-recover")
+            (target / state / "docs_sites").mkdir(parents=True)
+            (target / state / "docs_sites" / "site").write_text("regenerable")
+            recovery = mod.NfsExportRecovery(
+                str(source), str(target), service_backup=False
+            )
+            self.assertEqual(recovery.volumes_missing_from_snapshot(), ["docs_sites"])
+            recovery.restore()
+            self.assertTrue((target / state / "docs_sites" / "site").exists())
+            self.assertTrue((target / state / "app_data" / "db").exists())
+            self.assertFalse((target / state / "app_data" / "stale").exists())
+
     def test_main_wires_argparse_to_recovery(self):
         mod = _load("svc-bkp-nfs-2-local", "nfs_recover")
         for argv, expect_backup in (

tests/unit/python/utils/storage/test_nfs.py

@@ -6,6 +6,7 @@ from utils.storage.nfs import (
     mount_opts,
     state_path,
     swarm_nfs_backed,
+    volume_dir,
 )
 
 
@@ -62,6 +63,15 @@ class TestNfsHelpers(unittest.TestCase):
             "1.2.3.4:/srv/nfs/infinito-state",
         )
 
+    def test_volume_dir_is_the_pinned_name(self):
+        self.assertEqual(
+            volume_dir("whiteboard_tmp", {"name": "nextcloud_whiteboard_tmp"}),
+            "nextcloud_whiteboard_tmp",
+        )
+
+    def test_volume_dir_falls_back_to_the_key_without_a_pin(self):
+        self.assertEqual(volume_dir("docs_sites", {"backup": False}), "docs_sites")
+
 
 class TestSwarmNfsBacked(unittest.TestCase):
     """The predicate mirrors the compose_volumes rewrite decision: swarm mode

utils/storage/nfs.py

@@ -125,3 +125,50 @@ def swarm_nfs_backed(entry, *, application_id, deployment_mode, storage_backend)
     if str(get_role_placement(application_id) or "").strip().lower() == "manager":
         return False
     return not _role_forces_other_mode(application_id)
+
+
+def volume_dir(semantic_name, entry):
+    """Directory under the NFS state path the swarm rewrite binds a volume to.
+
+    Args:
+        semantic_name: the volume's key in the compose ``volumes:`` block.
+        entry: the volume's spec; a pinned ``name`` wins over the key.
+    """
+    pinned = entry.get("name") if isinstance(entry, Mapping) else None
+    return str(pinned) if pinned else str(semantic_name)
+
+
+def backup_disabled_dirs(applications, *, deployment_mode, storage_backend):
+    """State-path directories of the NFS-backed volumes declared ``backup: false``.
+
+    Args:
+        applications: merged applications mapping whose ``volumes`` carry the
+            canonical ``meta/volumes.yml`` entries.
+        deployment_mode: cluster-wide ``compose`` or ``swarm``.
+        storage_backend: the ``storage.backend`` value, e.g. ``local`` or ``nfs``.
+
+    Returns:
+        The sorted directory names, without any a backed-up volume also binds to.
+    """
+    disabled: set[str] = set()
+    kept: set[str] = set()
+    for app_id, app in applications.items():
+        volumes = app.get("volumes") if isinstance(app, Mapping) else None
+        if not isinstance(volumes, Mapping):
+            continue
+        for semantic_name, entry in volumes.items():
+            if (
+                not isinstance(entry, Mapping)
+                or entry.get("type", "volume") != "volume"
+            ):
+                continue
+            if not swarm_nfs_backed(
+                entry,
+                application_id=str(app_id),
+                deployment_mode=deployment_mode,
+                storage_backend=storage_backend,
+            ):
+                continue
+            target = disabled if entry.get("backup") is False else kept
+            target.add(volume_dir(semantic_name, entry))
+    return sorted(disabled - kept)