mounts: re-arm the trigger when unmount cannot stop the mount

Stopping the automount detaches the whole stack at the path, so an
unmount that then fails to stop the .mount leaves a plain writable
directory behind. Arm a fresh pair before raising, best effort — the
local data repair covers what lands there if that fails too.

The automount stop itself needs no such handling: systemd's
automount_stop() enters dead synchronously and the detach it performs
(MNT_DETACH, no server contact) only logs its errors, so a failure
there means systemd could not be reached at all.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: agners

supervisor/mounts/mount.py

@@ -616,9 +616,10 @@ async def unmount(self) -> None:
         # Stop the .automount first: it disarms the trigger so nothing can
         # re-mount during cleanup, and it lazily detaches the whole stack
         # at the path (systemd's unmount_autofs() uses MNT_DETACH), so the
-        # stop cannot block on an unreachable server. A failure other than
-        # "no such unit" must not pass silently — it would leave an armed
-        # trigger behind at the path of a supposedly removed mount.
+        # stop cannot block on an unreachable server. Stopping an automount
+        # is synchronous and cannot fail on its own — an error here means
+        # systemd could not be reached, which must not pass silently: it
+        # would leave an armed trigger at the path of a removed mount.
         try:
             result = await self._run_systemd_job(
                 "stop_unit",
@@ -661,6 +662,7 @@ async def unmount(self) -> None:
                     # it as done is how a stale mount once survived a
                     # "successful" cleanup (see #6938).
                     if result != "done":
+                        await self._rearm_after_failed_unmount()
                         raise MountError(
                             f"Could not unmount {self.name} (systemd result: {result})",
                             _LOGGER.error,
@@ -669,6 +671,7 @@ async def unmount(self) -> None:
                 # Unit went away with the automount detach — fine.
                 pass
             except DBusError as err:
+                await self._rearm_after_failed_unmount()
                 raise MountError(
                     f"Could not unmount {self.name} due to: {err!s}", _LOGGER.error
                 ) from err
@@ -682,6 +685,28 @@ async def unmount(self) -> None:
         self._unit = None
         self._state = None
 
+    async def _rearm_after_failed_unmount(self) -> None:
+        """Cover the path again after the .mount could not be stopped.
+
+        The automount stop already detached the whole stack, so a failing
+        .mount stop leaves a plain writable directory behind. One attempt
+        to arm a fresh pair, best effort — if that fails too the local
+        data repair picks up whatever lands there.
+        """
+        for unit_name in (self.automount_unit_name, self.unit_name):
+            with suppress(DBusError):
+                await self.sys_dbus.systemd.reset_failed_unit(unit_name)
+
+        try:
+            await self.mount()
+        except (MountError, OSError) as err:
+            _LOGGER.warning(
+                "Could not re-arm automount for %s after a failed unmount, "
+                "its path is a local directory until the next reload: %s",
+                self.name,
+                err,
+            )
+
     async def discard_session(self) -> None:
         """Stop the .mount unit while keeping the automount trigger armed.
 

tests/mounts/test_mount.py

@@ -580,7 +580,10 @@ async def test_mount_arming_failure(
 
 
 async def test_unmount_failure(
-    coresys: CoreSys, all_dbus_services: dict[str, DBusServiceMock], path_extern
+    coresys: CoreSys,
+    all_dbus_services: dict[str, DBusServiceMock],
+    tmp_supervisor_data,
+    path_extern,
 ):
     """Test failure to unmount."""
     systemd_service: SystemdService = all_dbus_services["systemd"]
@@ -608,20 +611,28 @@ async def test_unmount_failure(
         ("mnt-data-supervisor-mounts-test.automount", "fail")
     ]
 
-    # With the .automount stopped, a failure stopping the .mount itself
-    # raises as well.
+    # With the .automount stopped the path is detached, so a failure
+    # stopping the .mount raises but arms a fresh pair first — otherwise
+    # the path stays a plain writable directory.
     systemd_service.StopUnit.calls.clear()
+    systemd_service.StartTransientUnit.calls.clear()
     systemd_service.response_stop_unit = [
         "/org/freedesktop/systemd1/job/7623",
         ERROR_FAILURE,
     ]
-    with pytest.raises(MountError):
+    with (
+        patch("supervisor.mounts.mount._probe_network_mount", return_value=True),
+        pytest.raises(MountError),
+    ):
         await mount.unmount()
 
     assert systemd_service.StopUnit.calls == [
         ("mnt-data-supervisor-mounts-test.automount", "fail"),
         ("mnt-data-supervisor-mounts-test.mount", "fail"),
     ]
+    assert [call[0] for call in systemd_service.StartTransientUnit.calls] == [
+        "mnt-data-supervisor-mounts-test.automount"
+    ]
 
     # If the .mount unit is missing only the .automount stop is attempted —
     # it disarms the trigger and detaches anything left at the path.