ci: clean up stale FUSE mounts before running tests (#12020)

* ci(s3tables): tolerate stale FUSE mounts in disk diagnostics

* ci(s3tables): bound df diagnostics with timeout, explain once

A stale FUSE mount can also hang statfs, not just fail it, so wrap df in
timeout 15; one comment at the first site covers all sixteen.

* ci: tolerate stale FUSE mounts in df diagnostics elsewhere

The same self-hosted runners serve the other workflow files; a bare
df -h there can stall or fail a job the same way.

* ci: unmount disconnected FUSE test mounts before jobs start

* ci: skip FUSE cleanup when the variant checkout is skipped

* ci: reclaim the crashed run's test dir once its stale FUSE mount is detached

* ci: stop tracking python bytecode for the FUSE cleanup action

The helper's __pycache__ slipped into the tree; ignore it so local
test runs cannot re-add it.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* ci: run the FUSE cleanup action's unit tests on action changes

test_cleanup.py was never wired into CI, so a regression in cleanup.py
would ship silently. A path-filtered workflow runs it on every change
under .github/actions/.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* ci: match every FUSE test mount layout in the cleanup action

The candidates list only matched seaweedfs_fuse_test_*/mount, so stale
mounts left by fuse_failover and fuse_p2p runs (mountN leaves under
seaweedfs_fuse_* dirs) and the posix-compliance workflow's fixed
/tmp/seaweedfs/mount were never reclaimed and could still wedge df.

* ci: run the FUSE mount cleanup in the fuse_failover and fuse_p2p workflows

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

.github/actions/cleanup-stale-fuse-mounts/action.yml

@@ -0,0 +1,11 @@
+name: Clean up stale SeaweedFS test mounts
+description: Detach disconnected FUSE test mounts left by earlier jobs on a shared runner.
+runs:
+  using: composite
+  steps:
+    - name: Unmount disconnected SeaweedFS test mounts
+      if: runner.os == 'Linux' && runner.environment == 'self-hosted'
+      shell: bash
+      env:
+        CLEANUP_SCRIPT: ${{ github.action_path }}/cleanup.py
+      run: python3 "$CLEANUP_SCRIPT"

.github/actions/cleanup-stale-fuse-mounts/cleanup.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Detach disconnected SeaweedFS integration-test mounts without touching live ones."""
+
+import errno
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+
+
+def test_mounts(mountinfo):
+    """Read the kernel table directly: traversing a stale mount may block."""
+    roots = {tempfile.gettempdir(), "/tmp", "/var/tmp"}
+    for line in mountinfo.splitlines():
+        fields, filesystem = line.split(" - ", 1)
+        if filesystem.split()[0].split(".")[0] != "fuse":
+            continue
+        fields = fields.split()
+        # mountinfo escapes whitespace and backslashes with octal sequences.
+        target = re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), fields[4])
+        path = Path(target)
+        # Test mounts are a mount* leaf under a seaweedfs_fuse_* scratch dir
+        # (seaweedfs_fuse_test_*/mount, seaweedfs_fuse_failover_*/mount0, ...)
+        # or the posix-compliance workflow's fixed /tmp/seaweedfs/mount.
+        if (not path.name.startswith("mount")
+                or str(path.parent.parent) not in roots
+                or not (path.parent.name.startswith("seaweedfs_fuse")
+                        or path.parent.name == "seaweedfs")):
+            continue
+        yield fields[0], target
+
+
+def mounted(mount_id, target):
+    return (mount_id, target) in test_mounts(Path("/proc/self/mountinfo").read_text())
+
+
+def cleanup():
+    failed = False
+    for mount_id, target in test_mounts(Path("/proc/self/mountinfo").read_text()):
+        try:
+            # A separate, bounded probe prevents a wedged FUSE daemon from
+            # hanging cleanup. Only ENOTCONN proves the daemon disconnected;
+            # a timeout or another error is not enough to detach a live mount.
+            probe = subprocess.run(
+                [sys.executable, __file__, "--probe", target], timeout=5,
+                capture_output=True, check=False,
+            )
+        except subprocess.TimeoutExpired:
+            print(f"::warning::FUSE probe timed out; leaving {target} mounted")
+            continue
+        if probe.returncode != errno.ENOTCONN or not mounted(mount_id, target):
+            continue
+
+        print(f"Unmounting disconnected FUSE test mount: {target}", flush=True)
+        # Lazy unmount detaches a dead mount even when an old process still
+        # holds it open. Prefer the unprivileged FUSE helper; sudo is fallback.
+        for command in (
+            ["fusermount3", "-uz", "--", target],
+            ["fusermount", "-uz", "--", target],
+            ["sudo", "-n", "umount", "-l", "--", target],
+        ):
+            try:
+                subprocess.run(command, timeout=10, check=False)
+            except (OSError, subprocess.TimeoutExpired) as error:
+                print(f"{command[0]}: {error}", file=sys.stderr)
+            if not mounted(mount_id, target):
+                break
+        else:
+            print(f"::error::Could not unmount disconnected FUSE test mount: {target}")
+            failed = True
+            continue
+        # The detached mount's parent still holds the crashed run's data and
+        # logs; test_mounts only yields targets under a SeaweedFS test dir in
+        # a temp root, so the parent is safe to reclaim here.
+        shutil.rmtree(Path(target).parent, ignore_errors=True)
+    return int(failed)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) == 3 and sys.argv[1] == "--probe":
+        try:
+            os.stat(sys.argv[2])
+        except OSError as error:
+            sys.exit(error.errno)
+        sys.exit(0)
+    sys.exit(cleanup())

.github/actions/cleanup-stale-fuse-mounts/test_cleanup.py

@@ -0,0 +1,137 @@
+import errno
+import importlib.util
+from pathlib import Path
+import subprocess
+import unittest
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location("cleanup", Path(__file__).with_name("cleanup.py"))
+cleanup = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(cleanup)
+
+TARGET = "/tmp/seaweedfs_fuse_test_123/mount"
+MOUNTINFO = f"42 1 0:9 / {TARGET} rw - fuse.seaweedfs seaweedfs rw\n"
+
+
+class CleanupTests(unittest.TestCase):
+    def run_cleanup(self, results, mounted=True):
+        with patch.object(cleanup.Path, "read_text", return_value=MOUNTINFO), \
+                patch.object(cleanup, "mounted", side_effect=mounted if isinstance(mounted, list) else None,
+                             return_value=mounted) as check, \
+                patch.object(cleanup.shutil, "rmtree") as rmtree, \
+                patch.object(cleanup.subprocess, "run", side_effect=results) as run:
+            status = cleanup.cleanup()
+        return status, run, check, rmtree
+
+    def result(self, status):
+        return subprocess.CompletedProcess([], status)
+
+    def test_only_fuse_integration_test_mounts_are_candidates(self):
+        rows = MOUNTINFO
+        rows += MOUNTINFO.replace("fuse.seaweedfs", "ext4")
+        rows += MOUNTINFO.replace(TARGET, "/mnt/production")
+        rows += MOUNTINFO.replace(TARGET, "/tmp/seaweedfs_fuse_test_123/data")
+        rows += MOUNTINFO.replace(TARGET, "/home/user/seaweedfs_fuse_test_123/mount")
+        rows += MOUNTINFO.replace(TARGET, "/tmp/seaweedfs/data")
+        self.assertEqual(list(cleanup.test_mounts(rows)), [("42", TARGET)])
+
+    def test_all_test_mount_layouts_are_candidates(self):
+        targets = (
+            "/tmp/seaweedfs_fuse_test_123/mount",
+            "/tmp/seaweedfs_fuse_failover_456/mount0",
+            "/tmp/seaweedfs_fuse_failover_456/mount1",
+            "/var/tmp/seaweedfs_fuse_p2p_test_789/mount2",
+            "/tmp/seaweedfs/mount",
+        )
+        rows = "".join(
+            MOUNTINFO.replace("42", str(40 + i)).replace(TARGET, target)
+            for i, target in enumerate(targets)
+        )
+        self.assertEqual(
+            list(cleanup.test_mounts(rows)),
+            [(str(40 + i), target) for i, target in enumerate(targets)],
+        )
+
+    def test_mountinfo_paths_are_unescaped_once(self):
+        target = "/tmp/seaweedfs_fuse_test_space and\\slash/mount"
+        encoded = target.replace("\\", r"\134").replace(" ", r"\040")
+        rows = MOUNTINFO.replace(TARGET, encoded)
+        self.assertEqual(list(cleanup.test_mounts(rows)), [("42", target)])
+
+    def test_no_mounts_is_a_noop(self):
+        with patch.object(cleanup.Path, "read_text", return_value=""), \
+                patch.object(cleanup.subprocess, "run") as run:
+            self.assertEqual(cleanup.cleanup(), 0)
+        run.assert_not_called()
+
+    def test_healthy_mount_is_untouched(self):
+        status, run, _, _ = self.run_cleanup([self.result(0)])
+        self.assertEqual(status, 0)
+        self.assertEqual(run.call_count, 1)
+
+    def test_other_probe_errors_are_untouched(self):
+        for error in (errno.EACCES, errno.ENOENT, errno.EIO):
+            with self.subTest(error=error):
+                status, run, _, _ = self.run_cleanup([self.result(error)])
+                self.assertEqual(status, 0)
+                self.assertEqual(run.call_count, 1)
+
+    def test_probe_timeout_does_not_detach_a_possibly_live_mount(self):
+        status, run, _, _ = self.run_cleanup([subprocess.TimeoutExpired("probe", 5)])
+        self.assertEqual(status, 0)
+        self.assertEqual(run.call_count, 1)
+
+    def test_disconnected_mount_is_lazily_unmounted(self):
+        status, run, _, _ = self.run_cleanup([self.result(errno.ENOTCONN), self.result(0)], [True, False])
+        self.assertEqual(status, 0)
+        self.assertEqual(run.call_args_list[1].args[0], ["fusermount3", "-uz", "--", TARGET])
+        self.assertEqual(run.call_args_list[0].kwargs["timeout"], 5)
+        self.assertEqual(run.call_args_list[1].kwargs["timeout"], 10)
+
+    def test_replaced_or_removed_mount_is_untouched(self):
+        status, run, _, _ = self.run_cleanup([self.result(errno.ENOTCONN)], False)
+        self.assertEqual(status, 0)
+        self.assertEqual(run.call_count, 1)
+
+    def test_mount_identity_is_checked_without_stat(self):
+        with patch.object(cleanup.Path, "read_text", return_value=MOUNTINFO):
+            self.assertTrue(cleanup.mounted("42", TARGET))
+            self.assertFalse(cleanup.mounted("43", TARGET))
+
+    def test_missing_or_timed_out_helper_uses_fallback(self):
+        for error in (FileNotFoundError(), subprocess.TimeoutExpired("fusermount3", 10)):
+            with self.subTest(error=error):
+                status, run, _, _ = self.run_cleanup(
+                    [self.result(errno.ENOTCONN), error, self.result(0)], [True, True, False])
+                self.assertEqual(status, 0)
+                self.assertEqual(run.call_args.args[0], ["fusermount", "-uz", "--", TARGET])
+
+    def test_failed_helpers_fall_back_to_sudo(self):
+        status, run, _, _ = self.run_cleanup(
+            [self.result(errno.ENOTCONN), self.result(1), self.result(1), self.result(0)],
+            [True, True, True, False])
+        self.assertEqual(status, 0)
+        self.assertEqual(run.call_args.args[0], ["sudo", "-n", "umount", "-l", "--", TARGET])
+
+    def test_cleanup_fails_when_stale_mount_remains(self):
+        status, run, _, rmtree = self.run_cleanup(
+            [self.result(errno.ENOTCONN), self.result(1), self.result(1), self.result(1)])
+        self.assertEqual(status, 1)
+        self.assertEqual(run.call_count, 4)
+        rmtree.assert_not_called()
+
+    def test_unmounted_test_dir_is_reclaimed(self):
+        status, _, _, rmtree = self.run_cleanup(
+            [self.result(errno.ENOTCONN), self.result(0)], [True, False])
+        self.assertEqual(status, 0)
+        rmtree.assert_called_once_with(Path(TARGET).parent, ignore_errors=True)
+
+    def test_probe_process_reports_errno(self):
+        for target, expected in ((str(Path(__file__)), 0), (str(Path(__file__) / "absent"), errno.ENOTDIR)):
+            with self.subTest(target=target):
+                result = subprocess.run([cleanup.sys.executable, cleanup.__file__, "--probe", target], check=False)
+                self.assertEqual(result.returncode, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()