From d73a21e2ff4718836e5ad431815e45bac28d8d00 Mon Sep 17 00:00:00 2001 From: ali Date: Sat, 3 Oct 2026 14:49:14 -0700 Subject: [PATCH] fix(LakeFS-mount): treat a FUSE mount that cannot be listed as dead A computing unit whose GeeseFS process had died could never mount that repository again. The mount stays in /proc/mounts and can still answer stat() from attributes the kernel cached, but every read fails with "Transport endpoint is not connected". RepositoryMountManager.isFuseMounted only asked whether the mount was listed, so it reported "already mounted", never requested the mount again, and the operator found an unreadable directory. The mounter replaces a dead mount when asked, but its own liveness check, _responds, used stat() too, so a dead mount that still stats would be handed back as live. Both now require the mount's directory to be listable. Co-Authored-By: Claude Opus 5.5 (1M context) --- bin/mounter/mounter.py | 10 ++++-- bin/mounter/tests/test_mounter.py | 26 +++++++++++++++ .../core/storage/RepositoryMountManager.scala | 21 ++++++++++-- .../storage/RepositoryMountManagerSpec.scala | 32 +++++++++++++++++-- 4 files changed, 82 insertions(+), 7 deletions(-) diff --git a/bin/mounter/mounter.py b/bin/mounter/mounter.py index 83c26d7608c..7aa39481a80 100644 --- a/bin/mounter/mounter.py +++ b/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 diff --git a/bin/mounter/tests/test_mounter.py b/bin/mounter/tests/test_mounter.py index c7b84472929..3122acfd55d 100644 --- a/bin/mounter/tests/test_mounter.py +++ b/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( diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala index d24b267bf95..9227e02fad2 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala +++ b/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 } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala index 46273d96ac0..59cda79c41f 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala +++ b/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) + } }