fix(LakeFS-mount): treat a FUSE mount that cannot be listed as dead

bin/mounter/mounter.py

@@ -154,9 +154,15 @@ def is_mounted(path):
 
 
 def _responds(path):
-    """True if `path` can be stat()ed — false for a mount whose FUSE server is gone."""
+    """True if `path` can be listed — false for a mount whose FUSE server is gone.
+
+    stat() alone is not enough: the kernel can answer it from attributes cached while the
+    server was alive, so a dead mount may stat fine and fail with ENOTCONN only once it is
+    opened. Reporting that mount as live would hand it back as "already mounted" for ever.
+    """
     try:
-        os.stat(path)
+        with os.scandir(path) as entries:
+            next(entries, None)
         return True
     except OSError:
         return False

bin/mounter/tests/test_mounter.py

@@ -95,6 +95,32 @@ def test_a_dead_mount_is_still_reported_as_mounted(mounter, cu_dir):
     assert mounter.is_mounted(target)  # what it should have asked
 
 
+# ─────────────────── _responds() ───────────────────
+
+def test_a_listable_directory_responds(mounter, tmp_path):
+    assert mounter._responds(str(tmp_path))
+
+
+def test_a_missing_path_does_not_respond(mounter, tmp_path):
+    assert not mounter._responds(str(tmp_path / "gone"))
+
+
+def test_a_mount_that_stats_but_cannot_be_listed_does_not_respond(mounter, tmp_path, monkeypatch):
+    """A dead FUSE mount can still stat() from attributes the kernel cached, and fail only
+    when opened. Such a mount must not count as live, or it is never remounted."""
+    real_scandir = os.scandir
+
+    def dead_mount(path):
+        if os.fspath(path) == str(tmp_path):
+            raise OSError(107, "Transport endpoint is not connected")
+        return real_scandir(path)
+
+    monkeypatch.setattr(os, "scandir", dead_mount)
+    os.stat(tmp_path)  # stat still succeeds
+
+    assert not mounter._responds(str(tmp_path))
+
+
 # ─────────────────── _cuid_of() ───────────────────
 
 @pytest.mark.parametrize(

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala

@@ -27,7 +27,7 @@ import java.net.{HttpURLConnection, URI}
 import java.nio.charset.StandardCharsets
 import java.nio.file.{Files, Path, Paths}
 import scala.io.Source
-import scala.util.Using
+import scala.util.{Try, Using}
 
 /**
   * Makes a versioned LakeFS repository readable inside this computing unit's pod.
@@ -141,7 +141,7 @@ object RepositoryMountManager
       35000
     )
 
-private object InPodMount {
+private[amber] object InPodMount {
 
   def postJson(url: String, body: String, jwt: String): Unit = {
     val connection = URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection]
@@ -165,9 +165,20 @@ private object InPodMount {
     }
   }
 
+  /**
+    * True if a live FUSE mount is at `mountPoint`.
+    *
+    * Being listed is not enough. A mount whose GeeseFS process has died keeps its entry, and
+    * the kernel can still answer stat() from attributes it cached, but every read fails with
+    * "Transport endpoint is not connected". Counted as mounted, it would never be requested
+    * again -- and the mounter, which replaces a dead mount when asked, would never be asked.
+    */
+  def isFuseMounted(mountPoint: Path): Boolean =
+    isListedAsFuse(mountPoint) && responds(mountPoint)
+
   // /proc/mounts is the kernel's read-only view of this pod's mount table: the kernel adds the
   // entry when a propagated GeeseFS mount arrives and removes it on unmount, so nothing writes it.
-  def isFuseMounted(mountPoint: Path): Boolean = {
+  private def isListedAsFuse(mountPoint: Path): Boolean = {
     if (!Files.exists(mountPoint)) {
       return false
     }
@@ -179,4 +190,8 @@ private object InPodMount {
       }
     }.getOrElse(false)
   }
+
+  /** True if `directory` can be listed, which a dead FUSE mount cannot. */
+  def responds(directory: Path): Boolean =
+    Try(Using.resource(Files.newDirectoryStream(directory))(_.iterator().hasNext)).isSuccess
 }

common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala

@@ -19,12 +19,12 @@
 
 package org.apache.texera.amber.storage
 
-import org.apache.texera.amber.core.storage.RepositoryMountManager
+import org.apache.texera.amber.core.storage.{InPodMount, RepositoryMountManager}
 import org.apache.texera.common.config.EnvironmentalVariable
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
-import java.nio.file.{Path, Paths}
+import java.nio.file.{Files, Path, Paths}
 import scala.collection.mutable
 
 class RepositoryMountManagerSpec extends AnyFlatSpec with Matchers {
@@ -115,4 +115,32 @@ class RepositoryMountManagerSpec extends AnyFlatSpec with Matchers {
       fixture.requests shouldBe empty
     }
   }
+
+  // A mount whose GeeseFS process died stays in /proc/mounts and can even stat, but cannot be
+  // listed. Telling it apart is what lets ensureMounted ask again instead of handing the
+  // operator an unreadable directory. A real dead FUSE mount cannot be made in a unit test, so
+  // these pin the rule itself: whatever cannot be listed is not live.
+  "InPodMount.responds" should "accept a directory that can be listed" in {
+    val dir = Files.createTempDirectory("mount-live")
+    try InPodMount.responds(dir) shouldBe true
+    finally Files.delete(dir)
+  }
+
+  it should "reject a path that cannot be listed" in {
+    val dir = Files.createTempDirectory("mount-dead")
+    val file = Files.createFile(dir.resolve("not-a-directory"))
+    try {
+      InPodMount.responds(dir.resolve("missing")) shouldBe false
+      InPodMount.responds(file) shouldBe false
+    } finally {
+      Files.delete(file)
+      Files.delete(dir)
+    }
+  }
+
+  "InPodMount.isFuseMounted" should "not count a plain directory as a mount" in {
+    val dir = Files.createTempDirectory("mount-none")
+    try InPodMount.isFuseMounted(dir) shouldBe false
+    finally Files.delete(dir)
+  }
 }