diff --git a/qubes/api/__init__.py b/qubes/api/__init__.py
index da7516d76..764c98679 100644
--- a/qubes/api/__init__.py
+++ b/qubes/api/__init__.py
@@ -245,15 +245,21 @@ def enforce(predicate):
         if not predicate:
             raise PermissionDenied()
 
-    def validate_size(self, untrusted_size: bytes) -> int:
+    def validate_size(
+        self, untrusted_size: bytes, allow_negative: bool = False
+    ) -> int:
         self.enforce(isinstance(untrusted_size, bytes))
+        coefficient = 1
+        if allow_negative and untrusted_size.startswith(b"-"):
+            coefficient = -1
+            untrusted_size = untrusted_size[1:]
         if not untrusted_size.isdigit():
             raise qubes.exc.ProtocolError("Size must be ASCII digits (only)")
         if len(untrusted_size) >= 20:
             raise qubes.exc.ProtocolError("Sizes limited to 19 decimal digits")
         if untrusted_size[0] == 48 and untrusted_size != b"0":
             raise qubes.exc.ProtocolError("Spurious leading zeros not allowed")
-        return int(untrusted_size)
+        return int(untrusted_size) * coefficient
 
 
 class QubesDaemonProtocol(asyncio.Protocol):
diff --git a/qubes/api/admin.py b/qubes/api/admin.py
index 4543c6525..2e2e781d6 100644
--- a/qubes/api/admin.py
+++ b/qubes/api/admin.py
@@ -502,7 +502,7 @@ async def vm_volume_clone_to(self, untrusted_payload):
             src_volume=src_volume, dst_volume=dst_volume
         )
         self.dest.volumes[self.arg] = await qubes.utils.coro_maybe(
-            dst_volume.import_volume(src_volume)
+            self.dest.storage.import_volume(dst_volume, src_volume)
         )
         self.app.save()
 
@@ -550,11 +550,9 @@ async def vm_volume_clear(self):
     )
     async def vm_volume_set_revisions_to_keep(self, untrusted_payload):
         self.enforce(self.arg in self.dest.volumes.keys())
-        newvalue = self.validate_size(untrusted_payload)
-
+        newvalue = self.validate_size(untrusted_payload, allow_negative=True)
         self.fire_event_for_permission(newvalue=newvalue)
-
-        self.dest.volumes[self.arg].revisions_to_keep = newvalue
+        self.dest.storage.set_revisions_to_keep(self.arg, newvalue)
         self.app.save()
 
     @qubes.api.method("admin.vm.volume.Set.rw", scope="local", write=True)
@@ -820,7 +818,10 @@ async def pool_set_revisions_to_keep(self, untrusted_payload):
         self.enforce(self.dest.name == "dom0")
         self.enforce(self.arg in self.app.pools.keys())
         pool = self.app.pools[self.arg]
-        newvalue = self.validate_size(untrusted_payload)
+        newvalue = self.validate_size(untrusted_payload, allow_negative=True)
+
+        if newvalue < -1:
+            raise qubes.api.ProtocolError("Invalid value for revisions_to_keep")
 
         self.fire_event_for_permission(newvalue=newvalue)
 
diff --git a/qubes/backup.py b/qubes/backup.py
index c92f7f58a..951e66773 100644
--- a/qubes/backup.py
+++ b/qubes/backup.py
@@ -517,10 +517,20 @@ def get_backup_summary(self):
             summary_line += fmt.format(size_to_human(vm_size))
 
             if qid != 0 and vm_info.vm.is_running():
-                summary_line += (
-                    " <-- The VM is running, backup will contain "
-                    "its state from before its start!"
-                )
+                if [
+                    volume
+                    for volume in vm_info.vm.volumes.values()
+                    if volume.snapshots_disabled
+                ]:
+                    summary_line += (
+                        " <-- The VM is running and private volume snapshots "
+                        "are disabled. Backup will fail!"
+                    )
+                else:
+                    summary_line += (
+                        " <-- The VM is running, backup will "
+                        "contain its state from before its start!"
+                    )
 
             summary += summary_line + "\n"
 
@@ -823,6 +833,15 @@ async def backup_do(self):
             backup_app.domains[qid].features["backup-content"] = True
             backup_app.domains[qid].features["backup-path"] = vm_info.subdir
             backup_app.domains[qid].features["backup-size"] = vm_info.size
+
+            # VM running private volumes without snapshoting them
+            # (revision_to_keep = -1) must be powered off to be backup
+            if backup_app.domains[qid].is_running:
+                for volume in backup_app.domains[qid].volumes.values():
+                    if volume.snapshots_disabled:
+                        raise qubes.exc.QubesVMNotHaltedError(
+                            backup_app.domains[qid]
+                        )
         backup_app.save()
         del backup_app
 
diff --git a/qubes/storage/__init__.py b/qubes/storage/__init__.py
index bfef07eeb..8ba975f4a 100644
--- a/qubes/storage/__init__.py
+++ b/qubes/storage/__init__.py
@@ -41,6 +41,8 @@
 from qubes.exc import StoragePoolException
 
 STORAGE_ENTRY_POINT = "qubes.storage"
+VOLUME_STATE_DIR = "/var/run/qubes/"
+VOLUME_STATE_PREFIX = "volume-running-"
 _am_root = os.getuid() == 0
 
 BYTES_TO_ZERO = 1 << 16
@@ -96,7 +98,7 @@ def __init__(
         snap_on_start=False,
         source=None,
         ephemeral=None,
-        **kwargs
+        **kwargs,
     ):
         """Initialize a volume.
 
@@ -513,6 +515,27 @@ def _not_implemented(self, method_name):
         msg = msg.format(str(self.__class__.__name__), method_name)
         return NotImplementedError(msg)
 
+    @property
+    def snapshots_disabled(self) -> bool:
+        return (
+            self.revisions_to_keep == -1
+            and not self.snap_on_start
+            and self.save_on_stop
+        )
+
+    @property
+    def state_file(self) -> str:
+        return os.path.join(
+            VOLUME_STATE_DIR,
+            VOLUME_STATE_PREFIX
+            + f"{self.pool.name}:{self.vid}".replace("-", "--").replace(
+                "/", "-"
+            ),
+        )
+
+    def is_running(self) -> bool:
+        return os.path.exists(self.state_file)
+
 
 class Storage:
     """Class for handling VM virtual disks.
@@ -786,8 +809,37 @@ def block_devices(self):
                 else:
                     yield block_dev
 
+    def set_revisions_to_keep(self, volume, value):
+        if value < -1:
+            raise qubes.exc.QubesValueError(
+                "Invalid value for revisions_to_keep"
+            )
+
+        currentvalue = self.vm.volumes[volume].revisions_to_keep
+        enabling_disabling_snapshots = value != currentvalue and (
+            currentvalue == -1 or value == -1
+        )
+        if self.vm.is_running() and enabling_disabling_snapshots:
+            raise qubes.exc.QubesVMNotHaltedError(self.vm)
+
+        if self.vm.klass == "AppVM" and enabling_disabling_snapshots:
+            for vm in self.vm.dispvms:
+                if vm.is_running():
+                    raise qubes.exc.QubesVMNotHaltedError(vm)
+
+        self.vm.volumes[volume].revisions_to_keep = value
+
     async def start(self):
         """Execute the start method on each volume"""
+        for vol in self.vm.volumes.values():
+            if (
+                vol.source
+                and vol.source.snapshots_disabled
+                and vol.source.is_running()
+            ):
+                raise qubes.exc.QubesVMError(
+                    self.vm, f"Volume {vol.source.vid} is running"
+                )
         await qubes.utils.void_coros_maybe(
             # pylint: disable=line-too-long
             (
@@ -800,6 +852,10 @@ async def start(self):
             for name, vol in self.vm.volumes.items()
         )
 
+        for vol in self.vm.volumes.values():
+            with open(vol.state_file, "w", encoding="ascii"):
+                pass
+
     async def stop(self):
         """Execute the stop method on each volume"""
         await qubes.utils.void_coros_maybe(
@@ -813,6 +869,8 @@ async def stop(self):
             )
             for name, vol in self.vm.volumes.items()
         )
+        for vol in self.vm.volumes.values():
+            qubes.utils.remove_file(vol.state_file)
 
     def unused_frontend(self):
         """Find an unused device name"""
@@ -863,6 +921,18 @@ async def import_data_end(self, volume, success):
             self.get_volume(volume).import_data_end(success=success)
         )
 
+    async def import_volume(self, dst_volume: Volume, src_volume: Volume):
+        """Helper function to import data from another volume"""
+        if src_volume.is_running() and src_volume.snapshots_disabled:
+            raise StoragePoolException(
+                f"Volume {src_volume.vid} must be stopped before importing its "
+                f"data"
+            )
+
+        return await qubes.utils.coro_maybe(
+            dst_volume.import_volume(src_volume)
+        )
+
 
 class VolumesCollection:
     """Convenient collection wrapper for pool.get_volume and
diff --git a/qubes/storage/file.py b/qubes/storage/file.py
index 8b0bd727f..0099924f2 100644
--- a/qubes/storage/file.py
+++ b/qubes/storage/file.py
@@ -272,7 +272,7 @@ def revisions_to_keep(self):
 
     @revisions_to_keep.setter
     def revisions_to_keep(self, value):
-        if not isinstance(value, int) or value > 1 or value < 0:
+        if not isinstance(value, int) or value > 1 or value < -1:
             raise NotImplementedError(
                 "FileVolume supports maximum 1 volume revision to keep"
             )
@@ -490,9 +490,14 @@ async def start(self):
             # shutdown routine wasn't called (power interrupt or so)
             _remove_if_exists(self.path_cow)
         if not os.path.exists(self.path_cow):
-            create_sparse_file(self.path_cow, self.size)
+            if not self.snapshots_disabled:
+                create_sparse_file(self.path_cow, self.size)
         if not self.snap_on_start:
             _check_path(self.path)
+
+        if self.snapshots_disabled:
+            return self
+
         if hasattr(self, "path_source_cow"):
             if not os.path.exists(self.path_source_cow):
                 create_sparse_file(self.path_source_cow, self.size)
@@ -524,10 +529,12 @@ async def stop(self):
             self._export_lock is not FileVolume._marker_exported
         ), "trying to stop exported file volume?"
         if self.save_on_stop or self.snap_on_start:
-            await self._destroy_blockdev()
+            if not self.snapshots_disabled:
+                await self._destroy_blockdev()
         if self.save_on_stop:
             if self.rw:
-                self.commit()
+                if not self.snapshots_disabled:
+                    self.commit()
             else:
                 _remove_if_exists(self.path_cow)
         elif self.snap_on_start:
@@ -569,7 +576,7 @@ def script(self):
 
     def _block_device_path(self):
         if not self.snap_on_start:
-            if not self.save_on_stop:
+            if not self.save_on_stop or self.snapshots_disabled:
                 return self.path
             return "-".join(
                 [
diff --git a/qubes/storage/lvm.py b/qubes/storage/lvm.py
index d0b4a0cef..b866b15c7 100644
--- a/qubes/storage/lvm.py
+++ b/qubes/storage/lvm.py
@@ -780,7 +780,10 @@ async def start(self):
         try:
             if self.snap_on_start or self.save_on_stop:
                 if not self.save_on_stop or not self.is_dirty():
-                    await self._snapshot()
+                    if self.snapshots_disabled and self.revisions:
+                        await self._remove_revisions(self.revisions)
+                    if not self.snapshots_disabled:
+                        await self._snapshot()
                 else:
                     await self._activate()
             else:
@@ -795,7 +798,8 @@ async def stop(self):
         changed = True
         try:
             if self.save_on_stop:
-                changed = await self._commit()
+                if not self.snapshots_disabled:
+                    changed = await self._commit()
             elif self.snap_on_start:
                 changed = await self._remove_if_exists(self._vid_snap)
             else:
@@ -828,9 +832,10 @@ def block_device(self):
         the libvirt XML template as <disk>.
         """
         if self.snap_on_start or self.save_on_stop:
-            snap_path = "/dev/mapper/" + self._vid_snap.replace(
-                "-", "--"
-            ).replace("/", "-")
+            vid = self.vid if self.snapshots_disabled else self._vid_snap
+            snap_path = "/dev/mapper/" + vid.replace("-", "--").replace(
+                "/", "-"
+            )
             return qubes.storage.BlockDevice(
                 snap_path, self.name, None, self.rw, self.domain, self.devtype
             )
diff --git a/qubes/storage/reflink.py b/qubes/storage/reflink.py
index 92470cd5d..5d495e8d0 100644
--- a/qubes/storage/reflink.py
+++ b/qubes/storage/reflink.py
@@ -182,6 +182,8 @@ def __init__(self, *args, **kwargs):
     def _update_precache(self):
         _remove_file(self._path_precache)
         yield
+        if self.snapshots_disabled:
+            return
         _copy_file(self._path_clean, self._path_precache, copy_mtime=True)
 
     def _remove_stale_precache(self):
@@ -263,6 +265,13 @@ def is_dirty(self):
     @_coroutinized
     def start(self):  # pylint: disable=invalid-overridden-method
         self._remove_incomplete_images()
+        if self.snapshots_disabled:
+            self._prune_revisions(keep=0)
+            _remove_file(self._path_precache)
+            if not self.is_dirty():
+                _rename_file(self._path_clean, self._path_dirty)
+
+            return self
         if not self.is_dirty():
             if self.snap_on_start:
                 _remove_file(self._path_clean)
@@ -283,7 +292,10 @@ def start(self):  # pylint: disable=invalid-overridden-method
     @_coroutinized
     def stop(self):  # pylint: disable=invalid-overridden-method
         if self.is_dirty():
-            self._commit(self._path_dirty)
+            if self.snapshots_disabled:
+                _rename_file(self._path_dirty, self._path_clean)
+            else:
+                self._commit(self._path_dirty)
         elif not self.save_on_stop:
             if not self.snap_on_start:
                 self._size = self.size  # preserve manual resize of image
diff --git a/qubes/storage/zfs.py b/qubes/storage/zfs.py
index 00e9eac73..d0b16068c 100644
--- a/qubes/storage/zfs.py
+++ b/qubes/storage/zfs.py
@@ -1892,7 +1892,7 @@ async def _purge_old_revisions(self) -> None:
                 )
             )
         )
-        num = self.revisions_to_keep
+        num = max(0, self.revisions_to_keep)
         for snapshot, _ in revs[num:]:
             vsn = VolumeSnapshot.make(self.vid, snapshot)
             self.log.debug("Pruning %s", vsn)
@@ -1908,8 +1908,9 @@ async def _mark_clean(self):
             )
             if s.name.is_clean_snapshot()
         ]
-        new = self.volume.clean_snapshot()
-        await self.pool.accessor.snapshot_volume_async(new, log=self.log)
+        if not self.snapshots_disabled:
+            new = self.volume.clean_snapshot()
+            await self.pool.accessor.snapshot_volume_async(new, log=self.log)
         for old in existing_cleans:
             await self.pool.accessor.remove_volume_async(
                 old,
@@ -2006,6 +2007,19 @@ async def _wipe_and_clone_from(self, source: qubes.storage.Volume) -> None:
             # then clone it from there.
             samepoolsource = cast(ZFSVolume, source)
             self.log.debug("Source shares pool with me")
+            if samepoolsource.snapshots_disabled:
+                if samepoolsource.is_running():
+                    raise qubes.storage.StoragePoolException(
+                        f"Volume {samepoolsource.vid} must be stopped before "
+                        "being cloned."
+                    )
+                assert (
+                    not samepoolsource.is_running()
+                ), f"Volume {samepoolsource.vid} is running"
+                await samepoolsource.pool.accessor.snapshot_volume_async(
+                    samepoolsource.volume.clean_snapshot(),
+                    log=samepoolsource.log,
+                )
             try:
                 src = samepoolsource.latest_clean_snapshot[0]
             except DatasetDoesNotExist:
@@ -2016,6 +2030,11 @@ async def _wipe_and_clone_from(self, source: qubes.storage.Volume) -> None:
             async with self._clone_volume_2phase(src):
                 # Do nothing.  The context manager takes care of everything.
                 pass
+
+            if samepoolsource.snapshots_disabled:
+                # This will schedule removing of created clean snapshot
+                # pylint: disable=protected-access
+                await samepoolsource._mark_clean()
         else:
             # Source is not a ZFS one;
             # create the dataset with the size of the
@@ -2499,6 +2518,14 @@ async def export(self) -> str:
             )
         exported = self.exported_volume_name(get_random_string(8))
         dest_dset = Volume.make(exported)
+        if self.snapshots_disabled:
+            if self.is_running():
+                raise qubes.storage.StoragePoolException(
+                    f"Volume {self.vid} must be stopped before being cloned"
+                )
+            await self.pool.accessor.snapshot_volume_async(
+                self.volume.clean_snapshot(), log=self.log
+            )
         try:
             src = self.latest_clean_snapshot[0]
         except DatasetDoesNotExist:
@@ -2514,6 +2541,8 @@ async def export(self) -> str:
             NO_AUTO_SNAPSHOT,
             log=self.log,
         )
+        if self.snapshots_disabled:
+            await self._mark_clean()
         return os.path.join(ZVOL_DIR, exported)
 
     async def export_end(self, path: str) -> None:
diff --git a/qubes/tests/api_admin.py b/qubes/tests/api_admin.py
index b4abb78c3..0b5615e67 100644
--- a/qubes/tests/api_admin.py
+++ b/qubes/tests/api_admin.py
@@ -3144,6 +3144,11 @@ def cleanup_for_clone(self):
 
     def test_520_vm_volume_clone(self):
         self.setup_for_clone()
+
+        self.vm.volumes["private"].is_running = lambda: False
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
         token = self.call_mgmt_func(
             b"admin.vm.volume.CloneFrom", b"test-vm1", b"private", b""
         )
@@ -3165,6 +3170,10 @@ def test_520_vm_volume_clone(self):
     def test_521_vm_volume_clone_invalid_volume(self):
         self.setup_for_clone()
 
+        self.vm.volumes["private"].is_running = lambda: False
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
         with self.assertRaises(qubes.exc.PermissionDenied):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneFrom", b"test-vm1", b"private123", b""
@@ -3178,6 +3187,10 @@ def test_521_vm_volume_clone_invalid_volume(self):
     def test_522_vm_volume_clone_invalid_volume2(self):
         self.setup_for_clone()
 
+        self.vm.volumes["private"].is_running = lambda: False
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
         token = self.call_mgmt_func(
             b"admin.vm.volume.CloneFrom", b"test-vm1", b"private", b""
         )
@@ -3197,6 +3210,10 @@ def test_522_vm_volume_clone_invalid_volume2(self):
     def test_523_vm_volume_clone_removed_volume(self):
         self.setup_for_clone()
 
+        self.vm.volumes["private"].is_running = lambda: False
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
         token = self.call_mgmt_func(
             b"admin.vm.volume.CloneFrom", b"test-vm1", b"private", b""
         )
@@ -3224,6 +3241,10 @@ def get_volume(vid):
     def test_524_vm_volume_clone_invlid_token(self):
         self.setup_for_clone()
 
+        self.vm.volumes["private"].is_running = lambda: False
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
         with self.assertRaises(qubes.exc.PermissionDenied):
             self.call_mgmt_func(
                 b"admin.vm.volume.CloneTo",
@@ -3237,6 +3258,26 @@ def test_524_vm_volume_clone_invlid_token(self):
         )
         self.assertFalse(self.app.save.called)
 
+    def test_525_vm_volume_clone_snapshots_disabled(self):
+        self.setup_for_clone()
+
+        self.vm.volumes["private"].revisions_to_keep = -1
+        self.vm.volumes["private"].is_running = lambda: True
+        self.vm.storage.get_volume = lambda x: x
+        self.vm2.storage.get_volume = lambda x: x
+
+        token = self.call_mgmt_func(
+            b"admin.vm.volume.CloneFrom", b"test-vm1", b"private", b""
+        )
+
+        with self.assertRaises(qubes.storage.StoragePoolException):
+            self.call_mgmt_func(
+                b"admin.vm.volume.CloneTo",
+                b"test-vm2",
+                b"private",
+                token.encode(),
+            )
+
     def test_530_tag_list(self):
         self.vm.tags.add("tag1")
         self.vm.tags.add("tag2")
@@ -3521,6 +3562,42 @@ def test_611_backup_already_running(self):
                         b"admin.backup.Execute", b"dom0", b"testprofile"
                     )
 
+    def test_612_backup_when_vm_with_disabled_snapshots_is_running(self):
+        backup_profile = (
+            "include:\n"
+            " - test-vm1\n"
+            "destination_vm: test-vm1\n"
+            "destination_path: /var/tmp\n"
+            "passphrase_text: test\n"
+        )
+        expected_info = (
+            "------------------+--------------+--------------+\n"
+            "               VM |         type |         size |\n"
+            "------------------+--------------+--------------+\n"
+            "         test-vm1 |           VM |            0 | <-- The VM is \
+running and private volume snapshots are disabled. Backup will fail!\n"
+            "------------------+--------------+--------------+\n"
+            "      Total size: |                           0 |\n"
+            "------------------+--------------+--------------+\n"
+            "VMs not selected for backup:\n"
+            " - dom0\n"
+            " - test-template\n"
+        )
+        with tempfile.TemporaryDirectory() as profile_dir:
+            with open(
+                os.path.join(profile_dir, "testprofile.conf"), "w"
+            ) as profile_file:
+                profile_file.write(backup_profile)
+            self.vm.is_running = lambda: True
+            self.vm.volumes["private"].revisions_to_keep = -1
+            with unittest.mock.patch(
+                "qubes.config.backup_profile_dir", profile_dir
+            ):
+                result = self.call_mgmt_func(
+                    b"admin.backup.Info", b"dom0", b"testprofile"
+                )
+            self.assertEqual(result, expected_info)
+
     @unittest.mock.patch("qubes.backup.Backup")
     def test_620_backup_execute(self, mock_backup):
         backup_profile = (
@@ -3597,6 +3674,41 @@ async def service_passphrase(*args, **kwargs):
             "qubes.BackupPassphrase+testprofile"
         )
 
+    @unittest.mock.patch("qubes.backup.Backup")
+    def test_622_backup_execute_when_disable_snapshot_vm_running(
+        self, mock_backup
+    ):
+        backup_profile = (
+            "include:\n"
+            " - test-vm1\n"
+            "destination_vm: test-vm1\n"
+            "destination_path: /home/user\n"
+            "passphrase_text: test\n"
+        )
+        mock_backup.return_value.backup_do.side_effect = self.dummy_coro
+        with tempfile.TemporaryDirectory() as profile_dir:
+            with open(
+                os.path.join(profile_dir, "testprofile.conf"), "w"
+            ) as profile_file:
+                profile_file.write(backup_profile)
+            with unittest.mock.patch(
+                "qubes.config.backup_profile_dir", profile_dir
+            ):
+                result = self.call_mgmt_func(
+                    b"admin.backup.Execute", b"dom0", b"testprofile"
+                )
+        self.assertIsNone(result)
+        mock_backup.assert_called_once_with(
+            self.app,
+            {self.vm},
+            set(),
+            target_vm=self.vm,
+            target_dir="/home/user",
+            compressed=True,
+            passphrase="test",
+        )
+        mock_backup.return_value.backup_do.assert_called_once_with()
+
     def test_630_vm_stats(self):
         send_event = unittest.mock.Mock(spec=[])
 
@@ -4271,7 +4383,6 @@ def test_710_vm_volume_set_revisions_to_keep(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"],
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        self.vm.storage = unittest.mock.Mock()
         value = self.call_mgmt_func(
             b"admin.vm.volume.Set.revisions_to_keep",
             b"test-vm1",
@@ -4279,9 +4390,8 @@ def test_710_vm_volume_set_revisions_to_keep(self):
             b"2",
         )
         self.assertIsNone(value)
-        self.assertEqual(
-            self.vm.volumes.mock_calls,
-            [unittest.mock.call.keys(), ("__getitem__", ("private",), {})],
+        self.assertIn(
+            ("__getitem__", ("private",), {}), self.vm.volumes.mock_calls
         )
         self.assertEqual(self.vm.volumes["private"].revisions_to_keep, 2)
         self.app.save.assert_called_once_with()
@@ -4292,8 +4402,7 @@ def test_711_vm_volume_set_revisions_to_keep_negative(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"],
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        self.vm.storage = unittest.mock.Mock()
-        with self.assertRaises(qubes.exc.ProtocolError):
+        with self.assertRaises(qubes.exc.QubesValueError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Set.revisions_to_keep",
                 b"test-vm1",
@@ -4307,8 +4416,7 @@ def test_712_vm_volume_set_revisions_to_keep_not_a_number(self):
             "keys.return_value": ["root", "private", "volatile", "kernel"],
         }
         self.vm.volumes.configure_mock(**volumes_conf)
-        self.vm.storage = unittest.mock.Mock()
-        with self.assertRaises(qubes.exc.ProtocolError):
+        with self.assertRaises(qubes.api.ProtocolError):
             self.call_mgmt_func(
                 b"admin.vm.volume.Set.revisions_to_keep",
                 b"test-vm1",
@@ -4316,6 +4424,98 @@ def test_712_vm_volume_set_revisions_to_keep_not_a_number(self):
                 b"abc",
             )
 
+    def test_713_vm_volume_disable_snapshots_while_running(self):
+        self.vm.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            "keys.return_value": ["root", "private", "volatile", "kernel"],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        with unittest.mock.patch.object(self.vm, "is_running") as _mock:
+            _mock.return_value = True
+            with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
+                self.call_mgmt_func(
+                    b"admin.vm.volume.Set.revisions_to_keep",
+                    b"test-vm1",
+                    b"private",
+                    b"-1",
+                )
+
+    def test_714_vm_volume_re_enable_snapshots_while_running(self):
+        self.vm.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            "keys.return_value": ["root", "private", "volatile", "kernel"],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm.volumes["private"].revisions_to_keep = -1
+        with unittest.mock.patch.object(self.vm, "is_running") as _mock:
+            _mock.return_value = True
+            with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
+                self.call_mgmt_func(
+                    b"admin.vm.volume.Set.revisions_to_keep",
+                    b"test-vm1",
+                    b"private",
+                    b"2",
+                )
+
+    def test_715_start_volume_sourcing_running_volume_without_snapshot(self):
+        self.vm.volumes = {
+            "root": unittest.mock.Mock(),
+            "private": unittest.mock.Mock(),
+            "volatile": unittest.mock.Mock(),
+            "kernel": unittest.mock.Mock(),
+        }
+        self.vm.volumes["private"].source = unittest.mock.Mock()
+        self.vm.volumes["private"].source.snapshots_disabled = True
+        self.vm.volumes["private"].source.is_running.return_value = True
+        with self.assertRaises(qubes.exc.QubesVMError):
+            self.loop.run_until_complete(
+                asyncio.wait_for(self.vm.storage.start(), 1)
+            )
+
+    def test_716_enable_or_disable_snapshots_while_dispvm_running(self):
+        self.vm.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            "keys.return_value": ["root", "private", "volatile", "kernel"],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm.volumes["private"].revisions_to_keep = 2
+
+        fake_running_dispvm = unittest.mock.MagicMock()
+        fake_running_dispvm.is_running.return_value = True
+        type(self.vm).dispvms = unittest.mock.PropertyMock(
+            return_value=[fake_running_dispvm]
+        )
+
+        # Trying to disable snapshot on a VM having a running DispVM based on
+        # it and currently running
+        with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
+            self.call_mgmt_func(
+                b"admin.vm.volume.Set.revisions_to_keep",
+                b"test-vm1",
+                b"private",
+                b"-1",
+            )
+
+        # Trying to renable snapshot on a VM having a running DispVM based on
+        # it and currently running
+        self.vm.volumes["private"].revisions_to_keep = -1
+        with self.assertRaises(qubes.exc.QubesVMNotHaltedError):
+            self.call_mgmt_func(
+                b"admin.vm.volume.Set.revisions_to_keep",
+                b"test-vm1",
+                b"private",
+                b"1",
+            )
+
+        # Changing the number of revisions should work
+        self.vm.volumes["private"].revisions_to_keep = 1
+        self.call_mgmt_func(
+            b"admin.vm.volume.Set.revisions_to_keep",
+            b"test-vm1",
+            b"private",
+            b"2",
+        )
+
     def test_720_vm_volume_set_rw(self):
         self.vm.volumes = unittest.mock.MagicMock()
         volumes_conf = {
diff --git a/qubes/tests/storage_file.py b/qubes/tests/storage_file.py
index 9fab97ee9..67f5d9e5b 100644
--- a/qubes/tests/storage_file.py
+++ b/qubes/tests/storage_file.py
@@ -529,6 +529,76 @@ def test_024_import_data_with_new_size(self):
         self.assertEqual(volume_data.strip(b"\0"), b"test")
         self.assertEqual(len(volume_data), new_size)
 
+    def test_025_private_snapshots_disabled(self):
+        config = {
+            "name": "private",
+            "pool": self.POOL_NAME,
+            "size": defaults["root_img_size"],
+            "rw": True,
+            "save_on_stop": True,
+            "revisions_to_keep": -1,
+        }
+        vm = qubes.tests.storage.TestVM(self)
+        volume = self.app.get_pool(self.POOL_NAME).init_volume(vm, config)
+        self.assertEqual(volume.name, "private")
+        self.assertEqual(volume.pool, self.POOL_NAME)
+        self.assertEqual(volume.size, defaults["root_img_size"])
+        self.assertFalse(volume.snap_on_start)
+        self.assertTrue(volume.save_on_stop)
+        self.assertTrue(volume.rw)
+
+        volume.commit = unittest.mock.MagicMock()
+
+        volume.create()
+        self.loop.run_until_complete(volume.start())
+
+        self.assertTrue(os.path.exists(volume.path))
+        self.assertFalse(os.path.exists(volume.path_cow))
+        self.assertEqual(
+            volume.block_device().path,
+            os.path.join("/tmp/test-pool/appvms/", vm.name, "private.img"),
+        )
+
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.remove())
+        volume.commit.assert_not_called()
+
+    def test_026_private_snapshots_disabled_pool_level(self):
+        config = {
+            "name": "private",
+            "pool": self.POOL_NAME,
+            "size": defaults["root_img_size"],
+            "rw": True,
+            "save_on_stop": True,
+        }
+        vm = qubes.tests.storage.TestVM(self)
+        pool = self.app.get_pool(self.POOL_NAME)
+        pool.revisions_to_keep = -1
+        volume = pool.init_volume(vm, config)
+        self.assertEqual(volume.revisions_to_keep, -1)
+        self.assertEqual(volume.name, "private")
+        self.assertEqual(volume.pool, self.POOL_NAME)
+        self.assertEqual(volume.size, defaults["root_img_size"])
+        self.assertFalse(volume.snap_on_start)
+        self.assertTrue(volume.save_on_stop)
+        self.assertTrue(volume.rw)
+
+        volume.commit = unittest.mock.MagicMock()
+
+        volume.create()
+        self.loop.run_until_complete(volume.start())
+
+        self.assertTrue(os.path.exists(volume.path))
+        self.assertFalse(os.path.exists(volume.path_cow))
+        self.assertEqual(
+            volume.block_device().path,
+            os.path.join("/tmp/test-pool/appvms/", vm.name, "private.img"),
+        )
+
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.remove())
+        volume.commit.assert_not_called()
+
     def _get_loop_size(self, path):
         try:
             loop_name = (
@@ -716,6 +786,8 @@ def test_005_revisions_to_keep(self):
         with self.assertRaises((NotImplementedError, ValueError)):
             pool.revisions_to_keep = 2
         self.assertEqual(pool.revisions_to_keep, 1)
+        pool.revisions_to_keep = -1
+        self.assertEqual(pool.revisions_to_keep, -1)
 
     def test_011_appvm_file_images(self):
         """Check if all the needed image files are created for an AppVm"""
diff --git a/qubes/tests/storage_lvm.py b/qubes/tests/storage_lvm.py
index 62bfd3971..79c644e14 100644
--- a/qubes/tests/storage_lvm.py
+++ b/qubes/tests/storage_lvm.py
@@ -1362,6 +1362,79 @@ def test_120_only_certain_volumes_ephemeral(self):
             "ephemeral_volatile flag must be ignored for read-only volumes",
         )
 
+    def test_130_private_snapshots_disabled(self):
+        config = {
+            "name": "private",
+            "pool": self.pool.name,
+            "save_on_stop": True,
+            "rw": True,
+            "size": qubes.config.defaults["root_img_size"],
+            "revisions_to_keep": -1,
+        }
+
+        vm = qubes.tests.storage.TestVM(self)
+        volume = self.app.get_pool(self.pool.name).init_volume(vm, config)
+        self.assertIsInstance(volume, self.volume_class)
+        self.assertEqual(volume.name, "private")
+        self.assertEqual(volume.pool, self.pool.name)
+        self.assertEqual(volume.size, qubes.config.defaults["root_img_size"])
+        self.loop.run_until_complete(volume.create())
+
+        volume._commit = unittest.mock.Mock()
+        self.loop.run_until_complete(volume.start())
+
+        self.assertTrue(os.path.exists(f"/dev/{volume.vid}"))
+        self.assertFalse(os.path.exists(f"/dev/{volume._vid_snap}"))
+
+        self.assertEqual(
+            volume.block_device().path,
+            "/dev/mapper/{}-vm--test--inst--appvm--private".format(
+                self.pool.volume_group
+            ),
+        )
+
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.remove())
+
+        volume._commit.assert_not_called()
+
+    def test_131_private_snapshots_disabled_pool_level(self):
+        config = {
+            "name": "private",
+            "pool": self.pool.name,
+            "save_on_stop": True,
+            "rw": True,
+            "size": qubes.config.defaults["root_img_size"],
+        }
+
+        vm = qubes.tests.storage.TestVM(self)
+        pool = self.app.get_pool(self.pool.name)
+        pool.revisions_to_keep = -1
+        volume = pool.init_volume(vm, config)
+        self.assertIsInstance(volume, self.volume_class)
+        self.assertEqual(volume.name, "private")
+        self.assertEqual(volume.pool, self.pool.name)
+        self.assertEqual(volume.size, qubes.config.defaults["root_img_size"])
+        self.loop.run_until_complete(volume.create())
+
+        volume._commit = unittest.mock.Mock()
+        self.loop.run_until_complete(volume.start())
+
+        self.assertTrue(os.path.exists(f"/dev/{volume.vid}"))
+        self.assertFalse(os.path.exists(f"/dev/{volume._vid_snap}"))
+
+        self.assertEqual(
+            volume.block_device().path,
+            "/dev/mapper/{}-vm--test--inst--appvm--private".format(
+                self.pool.volume_group
+            ),
+        )
+
+        self.loop.run_until_complete(volume.stop())
+        self.loop.run_until_complete(volume.remove())
+
+        volume._commit.assert_not_called()
+
 
 @skipUnlessLvmPoolExists
 class TC_01_ThinPool(ThinPoolBase, qubes.tests.SystemTestCase):
diff --git a/qubes/tests/storage_reflink.py b/qubes/tests/storage_reflink.py
index 9eee47bfb..ac4e350bb 100644
--- a/qubes/tests/storage_reflink.py
+++ b/qubes/tests/storage_reflink.py
@@ -26,6 +26,7 @@
 import shutil
 import subprocess
 import sys
+import unittest.mock
 from contextlib import nullcontext
 
 import qubes.tests
@@ -346,6 +347,37 @@ def test_110_remove_stale_precache_stale(self):
     def test_111_remove_stale_precache_stale_orphan(self):
         self._test_remove_stale_precache(stale=True, orphan=True)
 
+    def test_120_private_snapshots_disabled(self):
+        config = {
+            "name": "private",
+            "pool": self.pool.name,
+            "save_on_stop": True,
+            "rw": True,
+            "size": qubes.config.defaults["root_img_size"],
+            "revisions_to_keep": -1,
+        }
+
+        vm = qubes.tests.storage.TestVM(self)
+        volume = self.pool.init_volume(vm, config)
+        self.assertIsInstance(volume, reflink.ReflinkVolume)
+        self.assertEqual(volume.name, "private")
+        self.assertEqual(volume.pool, self.pool.name)
+        volume._copy_file = unittest.mock.Mock()
+        self.loop.run_until_complete(volume.create())
+        self.loop.run_until_complete(volume.start())
+        self.assertEqual(
+            volume.block_device().path,
+            "/var/tmp/test-reflink-units-on-btrfs/appvms/test-inst-appvm/private-dirty.img",
+        )
+        self.assertFalse(os.path.exists(volume._path_clean))
+        self.assertFalse(os.path.exists(volume._path_precache))
+        self.loop.run_until_complete(volume.stop())
+        self.assertFalse(os.path.exists(volume._path_dirty))
+        self.assertTrue(os.path.exists(volume._path_clean))
+        self.assertFalse(os.path.exists(volume._path_precache))
+        self.loop.run_until_complete(volume.remove())
+        volume._copy_file.assert_not_called()
+
 
 def setup_loopdev(img, cleanup_via=None):
     dev = str.strip(cmd("sudo", "losetup", "-f", "--show", img).decode())
diff --git a/qubes/tests/storage_zfs.py b/qubes/tests/storage_zfs.py
index e44fd9772..8c821de3c 100644
--- a/qubes/tests/storage_zfs.py
+++ b/qubes/tests/storage_zfs.py
@@ -828,3 +828,54 @@ def deinit():
 
         # Fin
         deinit()
+
+    def test_026_snapshots_disabled(self) -> None:
+        v = self.get_vol(ONEMEG_SAVE_ON_STOP, name="026", revisions_to_keep=0)
+        self.rc(v.create())
+        self.assertEqual(len(v.revisions), 0)
+        self.rc(v.start())
+        self.assertEqual(len(v.revisions), 0)
+        self.rc(v.stop())
+        self.assertEqual(len(v.revisions), 0)
+
+    def test_027_snapshots_disabled_removes_old_snapshots(self) -> None:
+        v = self.get_vol(ONEMEG_SAVE_ON_STOP, name="027", revisions_to_keep=1)
+        self.rc(v.create())
+        self.rc(v.start())
+        self.rc(v.stop())
+        self.assertEqual(len(v.revisions), 1)
+
+        v.revisions_to_keep = -1
+        self.rc(v.start())
+        self.rc(v.stop())
+        self.assertEqual(len(v.revisions), 0)
+
+    def test_028_prevent_export_when_volume_in_use(self) -> None:
+        v = self.get_vol(ONEMEG_SAVE_ON_STOP, name="028", revisions_to_keep=-1)
+
+        v.is_running = lambda: True
+        self.rc(v.create())
+        self.rc(v.start())
+
+        with self.assertRaises(qubes.storage.StoragePoolException):
+            self.rc(v.export())
+        self.rc(v.stop())
+
+    def test_029_prevent_cloning_running_volume_with_snapshots_disabled(self):
+        v1 = self.get_vol(
+            ONEMEG_SAVE_ON_STOP, name="029_1", revisions_to_keep=-1
+        )
+        v1.is_running = lambda: True
+        self.rc(v1.create())
+        self.rc(v1.start())
+
+        v2 = self.get_vol(
+            ONEMEG_SAVE_ON_STOP, name="029_2", revisions_to_keep=1
+        )
+        self.rc(v2.create())
+        with self.assertRaises(qubes.storage.StoragePoolException):
+            self.rc(v2.import_volume(v1))
+
+        self.rc(v1.stop())
+        v1.is_running = lambda: False
+        self.rc(v2.import_volume(v1))
diff --git a/qubes/vm/dispvm.py b/qubes/vm/dispvm.py
index b683ed3fc..9a1c767db 100644
--- a/qubes/vm/dispvm.py
+++ b/qubes/vm/dispvm.py
@@ -408,9 +408,7 @@ def on_domain_pre_unpaused(
             self.use_preload()
 
     @qubes.events.handler("domain-shutdown")
-    async def on_domain_shutdown(
-        self, _event, **_kwargs
-    ):
+    async def on_domain_shutdown(self, _event, **_kwargs):
         """Do auto cleanup if enabled"""
         await self._auto_cleanup()
 
diff --git a/qubes/vm/mix/dvmtemplate.py b/qubes/vm/mix/dvmtemplate.py
index c6cc1f908..b92cddb39 100644
--- a/qubes/vm/mix/dvmtemplate.py
+++ b/qubes/vm/mix/dvmtemplate.py
@@ -118,6 +118,23 @@ def on_feature_pre_set_preload_dispvm_max(
                 "Invalid preload-dispvm-max value: not a digit"
             )
 
+    @qubes.events.handler("domain-pre-start")
+    def __on_domain_pre_start(self, event, **kwargs):
+        """Prevents startup for domain having a volume with disabled snapshots
+        and a DispVM based on this volume started
+        """
+        # pylint: disable=unused-argument
+        volume_with_disabled_snapshots = False
+        for vol in self.volumes.values():
+            volume_with_disabled_snapshots |= vol.snapshots_disabled
+
+        if not volume_with_disabled_snapshots:
+            return
+
+        for vm in self.dispvms:
+            if vm.is_running():
+                raise qubes.exc.QubesVMNotHaltedError(vm)
+
     @qubes.events.handler("domain-feature-set:preload-dispvm-max")
     def on_feature_set_preload_dispvm_max(
         self, event, feature, value, oldvalue=None