diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index becd8deb34c..a82cbfde95b 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -309,6 +309,7 @@ jobs:
               "Auth/jacoco" \
               "Config/jacoco" \
               "Resource/jacoco" \
+              "Util/jacoco" \
               "PyBuilder/jacoco" \
               "WorkflowCore/jacoco" \
               "WorkflowOperator/jacoco" \
diff --git a/AGENTS.md b/AGENTS.md
index b21ed6d6a1a..7118073a781 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -29,7 +29,7 @@ engine, an Angular UI, and the agent service. JVM modules wired in
 | --- | --- | --- |
 | Workflow execution engine (Amber) | `amber/` | [amber/README.md](amber/README.md) |
 | Backend services | `config-service/`, `access-control-service/`, `file-service/`, `computing-unit-managing-service/`, `workflow-compiling-service/`, `notebook-migration-service/` | `build.sbt` |
-| Shared Scala libs | `common/` (`auth`, `config`, `dao`, `workflow-core`, `workflow-operator`, `pybuilder`) | `build.sbt` |
+| Shared Scala libs | `common/` (`auth`, `config`, `dao`, `util`, `workflow-core`, `workflow-operator`, `pybuilder`) | `build.sbt` |
 | Frontend (Angular) | `frontend/` | [frontend/README.md](frontend/README.md) |
 | Agent service (Bun/TS, LLM agents) | `agent-service/` | `agent-service/package.json` |
 | Pyright language service | `pyright-language-service/` | [pyright-language-service/README.md](pyright-language-service/README.md) |
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
index 5dee2c67546..393e5b343a5 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/Utils.scala
@@ -67,6 +67,11 @@ object Utils extends LazyLogging {
     * for callers on an actor or coordinator thread, where a `Thread.sleep` would also stall
     * unrelated work queued on that thread.
     *
+    * This is the non-blocking half of the pair: for blocking work, use
+    * `org.apache.texera.common.util.RetryUtil.withBackoff` in `common/util`, which takes the same
+    * attempts-and-doubling knobs. It lives in a separate module because this variant needs
+    * `com.twitter:util-core`, which only `amber` declares.
+    *
     * A synchronous throw while evaluating `fn` counts as a failed attempt, same as a failed
     * `Future`. Fatal errors are never retried in either shape: they propagate immediately.
     *
@@ -88,8 +93,8 @@ object Utils extends LazyLogging {
   )(fn: => Future[T]): Future[T] = {
     def attempt(attemptNumber: Int, backoffTimeInMS: Long): Future[T] =
       Future(fn).flatten.rescue {
-        // `NonFatal` so that a fatal handed back as a failed `Future` is not retried either, which
-        // also matches how the blocking backoff loops elsewhere in the repo catch `Exception`.
+        // `NonFatal` so that a fatal handed back as a failed `Future` is not retried either, matching
+        // the blocking sibling `RetryUtil.withBackoff`, which uses the same `NonFatal` predicate.
         case NonFatal(e) if attemptNumber < attempts =>
           onRetry(e, attemptNumber, backoffTimeInMS)
           Future
diff --git a/build.sbt b/build.sbt
index 720fc28b806..b10661fd70e 100644
--- a/build.sbt
+++ b/build.sbt
@@ -117,6 +117,9 @@ val nettyDependencyOverrides = Seq(
 // keep the org.apache.log4j API available at runtime.
 ThisBuild / excludeDependencies += ExclusionRule("log4j", "log4j")
 
+// Dependency-free helpers (retry/backoff, ...) that any module may depend on. Keep it that way:
+// anything added here reaches the classpath of every service that depends on it.
+lazy val Util = (project in file("common/util")).settings(commonModuleSettings)
 lazy val DAO = (project in file("common/dao")).settings(commonModuleSettings)
 lazy val Config = (project in file("common/config")).settings(commonModuleSettings)
 lazy val Resource = (project in file("common/resource")).settings(commonModuleSettings)
@@ -156,7 +159,7 @@ lazy val PyBuilder = (project in file("common/pybuilder"))
 
 lazy val WorkflowCore = (project in file("common/workflow-core"))
   .settings(commonModuleSettings)
-  .dependsOn(DAO, Config, PyBuilder)
+  .dependsOn(DAO, Config, PyBuilder, Util)
   .configs(Test)
   .dependsOn(DAO % "test->test") // test scope dependency
 lazy val ComputingUnitManagingService = (project in file("computing-unit-managing-service"))
@@ -202,7 +205,7 @@ lazy val ComputingUnitManagingService = (project in file("computing-unit-managin
   )
 lazy val FileService = (project in file("file-service"))
   .settings(commonModuleSettings)
-  .dependsOn(WorkflowCore, Auth, Config, Resource)
+  .dependsOn(WorkflowCore, Auth, Config, Resource, Util)
   .configs(Test)
   .dependsOn(DAO % "test->test") // test scope dependency
   .settings(
@@ -278,6 +281,7 @@ lazy val TexeraProject = (project in file("."))
     Auth,
     Config,
     Resource,
+    Util,
     DAO,
     PyBuilder,
     WorkflowCore,
diff --git a/common/util/build.sbt b/common/util/build.sbt
new file mode 100644
index 00000000000..0f4446edccd
--- /dev/null
+++ b/common/util/build.sbt
@@ -0,0 +1,56 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import scala.collection.Seq
+
+name := "util"
+
+enablePlugins(JavaAppPackaging)
+
+// Enable semanticdb for Scalafix
+ThisBuild / semanticdbEnabled := true
+ThisBuild / semanticdbVersion := scalafixSemanticdb.revision
+
+// Manage dependency conflicts by always using the latest revision
+ThisBuild / conflictManager := ConflictManager.latestRevision
+
+// Restrict parallel execution of tests to avoid conflicts
+Global / concurrentRestrictions += Tags.limit(Tags.Test, 1)
+
+/////////////////////////////////////////////////////////////////////////////
+// Compiler Options
+/////////////////////////////////////////////////////////////////////////////
+
+// Scala compiler options
+Compile / scalacOptions ++= Seq(
+  "-Xelide-below", "WARNING",       // Turn on optimizations with "WARNING" as the threshold
+  "-feature",                       // Check feature warnings
+  "-deprecation",                   // Check deprecation warnings
+  "-Ywarn-unused:imports"           // Check for unused imports
+)
+
+/////////////////////////////////////////////////////////////////////////////
+// Dependencies
+/////////////////////////////////////////////////////////////////////////////
+
+// This module is deliberately dependency-free apart from the test framework:
+// any module may depend on it, so anything added here reaches the classpath of
+// every service that does. Callers pass their own logger in through the hooks
+// instead of the module pulling in a logging library.
+libraryDependencies ++= Seq(
+  "org.scalatest" %% "scalatest" % "3.2.15" % Test // ScalaTest (for unit tests)
+)
diff --git a/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala
new file mode 100644
index 00000000000..3ebf094b4e7
--- /dev/null
+++ b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.common.util
+
+import scala.annotation.tailrec
+import scala.util.control.NonFatal
+
+/**
+  * Retry with exponential backoff, for blocking work.
+  *
+  * This module has no dependencies of its own, so any module may depend on it rather than writing
+  * another retry loop by hand. If the work returns a `Future` rather than blocking, use the
+  * non-blocking sibling
+  * `org.apache.texera.amber.engine.common.Utils.retry` in `amber` instead: it takes the same
+  * attempts-and-doubling-backoff knobs but waits on a `Timer`, which matters on an actor or
+  * coordinator thread where `Thread.sleep` would stall unrelated work queued behind it.
+  */
+object RetryUtil {
+
+  /**
+    * One failed attempt that is about to be retried. Carries everything a caller needs to log the
+    * retry itself; `message` is the standard wording, so retries read the same everywhere.
+    */
+  final case class RetryAttempt(
+      description: String,
+      attempt: Int,
+      maxAttempts: Int,
+      delayMillis: Long,
+      cause: Throwable
+  ) {
+    def message: String =
+      s"Failed to $description (attempt $attempt/$maxAttempts): ${cause.getMessage}. " +
+        s"Retrying in ${delayMillis}ms..."
+  }
+
+  /**
+    * Runs `operation`, retrying on failure with exponential backoff (the delay doubles after each
+    * failed attempt) until it succeeds or `maxAttempts` is reached. The final failure is wrapped
+    * with `description` and the last exception as its cause.
+    *
+    * Only `NonFatal` failures are treated as transient, which is the same predicate the
+    * non-blocking sibling uses. Note that `NonFatal` admits non-fatal `Error`s -- `AssertionError`,
+    * `java.io.IOError`, `ServiceConfigurationError` -- so those are retried rather than propagated
+    * straight away. An `InterruptedException` -- raised by the operation or by the wait between
+    * attempts -- fails fast with the interrupt status restored, so a caller shutting the thread
+    * down is never made to sit through the remaining backoff.
+    *
+    * @param description        verb phrase naming the work, e.g. "connect to lake fs server". It is
+    *                           interpolated into every message: "Failed to $description after ...".
+    * @param maxAttempts        total attempts; 1 means no retry at all.
+    * @param initialDelayMillis wait before the first retry; doubled after each failed attempt.
+    * @param onRetry            invoked before each wait. Log `RetryAttempt.message` through the
+    *                           caller's own logger, so retries are attributed to the caller rather
+    *                           than to this util.
+    * @param sleep              how to wait; injectable so tests exercise the backoff without waiting.
+    * @param operation          the work to run, re-evaluated on each attempt.
+    * @tparam T whatever `operation` returns.
+    * @return `operation`'s value from the first attempt that succeeds.
+    */
+  def withBackoff[T](
+      description: String,
+      maxAttempts: Int,
+      initialDelayMillis: Long,
+      onRetry: RetryAttempt => Unit,
+      sleep: Long => Unit = Thread.sleep
+  )(operation: => T): T = {
+    // Restore the interrupt status and fail fast rather than retrying, whether the interrupt
+    // arrives while running `operation` or while waiting between attempts.
+    def failInterrupted(cause: InterruptedException): Nothing = {
+      Thread.currentThread().interrupt()
+      throw new RuntimeException(s"Interrupted while waiting to $description", cause)
+    }
+
+    @tailrec
+    def attemptFrom(attempt: Int, delayMillis: Long): T = {
+      val outcome: Either[Throwable, T] =
+        try Right(operation)
+        catch {
+          case ie: InterruptedException => failInterrupted(ie)
+          case NonFatal(cause)          => Left(cause)
+        }
+
+      outcome match {
+        case Right(value) => value
+        case Left(cause) =>
+          if (attempt >= maxAttempts) {
+            throw new RuntimeException(
+              s"Failed to $description after $maxAttempts attempts: ${cause.getMessage}",
+              cause
+            )
+          }
+          onRetry(RetryAttempt(description, attempt, maxAttempts, delayMillis, cause))
+          try sleep(delayMillis)
+          catch { case ie: InterruptedException => failInterrupted(ie) }
+          attemptFrom(attempt + 1, delayMillis * 2)
+      }
+    }
+
+    attemptFrom(attempt = 1, delayMillis = initialDelayMillis)
+  }
+}
diff --git a/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala
new file mode 100644
index 00000000000..828cb8d01e5
--- /dev/null
+++ b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala
@@ -0,0 +1,206 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.common.util
+
+import org.apache.texera.common.util.RetryUtil.RetryAttempt
+import org.scalatest.flatspec.AnyFlatSpec
+
+import scala.collection.mutable.ListBuffer
+import scala.util.control.ControlThrowable
+
+/**
+  * Contract of the shared blocking backoff retry. `sleep` is injected everywhere so the backoff
+  * progression is asserted exactly without any test waiting.
+  *
+  * Coverage is the full contract both blocking callers rely on: the doubling progression, which
+  * failures count as transient, the give-up wrapping, and interrupt fail-fast during the operation
+  * and during a backoff sleep.
+  */
+class RetryUtilSpec extends AnyFlatSpec {
+
+  private def noopRetryHook: RetryAttempt => Unit = _ => ()
+
+  "RetryUtil.withBackoff" should "return the operation's value on first success without sleeping" in {
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) {
+      attempts += 1
+      "value"
+    }
+    assert(result == "value")
+    assert(attempts == 1)
+    assert(delays.isEmpty)
+  }
+
+  it should "retry until success and double the delay after each failed attempt" in {
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    val result = RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) {
+      attempts += 1
+      if (attempts < 3) throw new RuntimeException("transient")
+      attempts
+    }
+    assert(result == 3)
+    assert(delays.toList == List(200L, 400L))
+  }
+
+  it should "honor a custom initial delay when computing the progression" in {
+    // Guards against a hardcoded base: from 50ms the progression must be 50, 100, 200.
+    val delays = ListBuffer.empty[Long]
+    intercept[RuntimeException] {
+      RetryUtil.withBackoff("reach the store", 4, 50L, noopRetryHook, delays += _) {
+        throw new RuntimeException("down")
+      }
+    }
+    assert(delays.toList == List(50L, 100L, 200L))
+  }
+
+  it should "succeed on the final permitted attempt without giving up one try too early" in {
+    // Boundary for `attempt >= maxAttempts`: success on the very last attempt must still count.
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    RetryUtil.withBackoff("reach the store", 3, 200L, noopRetryHook, delays += _) {
+      attempts += 1
+      if (attempts < 3) throw new RuntimeException("transient")
+    }
+    assert(attempts == 3)
+    assert(delays.toList == List(200L, 400L))
+  }
+
+  it should "give up after maxAttempts, naming the description and preserving the cause" in {
+    val cause = new RuntimeException("still down")
+    var attempts = 0
+    val failure = intercept[RuntimeException] {
+      RetryUtil.withBackoff("connect to lake fs server", 3, 200L, noopRetryHook, _ => ()) {
+        attempts += 1
+        throw cause
+      }
+    }
+    assert(attempts == 3)
+    assert(failure.getMessage == "Failed to connect to lake fs server after 3 attempts: still down")
+    assert(failure.getCause eq cause)
+  }
+
+  it should "give up immediately without sleeping when maxAttempts is one" in {
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    val failure = intercept[RuntimeException] {
+      RetryUtil.withBackoff("reach the store", 1, 200L, noopRetryHook, delays += _) {
+        attempts += 1
+        throw new RuntimeException("still down")
+      }
+    }
+    assert(attempts == 1)
+    assert(delays.isEmpty)
+    assert(failure.getMessage.contains("after 1 attempts"))
+  }
+
+  it should "report each retry to the hook, and never after the last attempt" in {
+    val observed = ListBuffer.empty[RetryAttempt]
+    intercept[RuntimeException] {
+      RetryUtil.withBackoff("reach the store", 3, 200L, observed += _, _ => ()) {
+        throw new RuntimeException("down")
+      }
+    }
+    // 3 attempts buy 2 retries, so the hook fires twice with 1-based attempt numbers.
+    assert(
+      observed.map(a => (a.attempt, a.maxAttempts, a.delayMillis)).toList == List(
+        (1, 3, 200L),
+        (2, 3, 400L)
+      )
+    )
+    assert(observed.forall(_.cause.getMessage == "down"))
+    assert(
+      observed.head.message ==
+        "Failed to reach the store (attempt 1/3): down. Retrying in 200ms..."
+    )
+  }
+
+  it should "fail fast and restore the interrupt status when the operation is interrupted" in {
+    val delays = ListBuffer.empty[Long]
+    val failure = intercept[RuntimeException] {
+      RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) {
+        throw new InterruptedException("interrupted")
+      }
+    }
+    // Thread.interrupted() both reads and clears the flag, so the interrupt was restored.
+    assert(Thread.interrupted())
+    assert(failure.getMessage == "Interrupted while waiting to reach the store")
+    assert(failure.getCause.isInstanceOf[InterruptedException])
+    assert(delays.isEmpty)
+  }
+
+  it should "fail fast and restore the interrupt status when interrupted during a backoff sleep" in {
+    // A `catch` cannot catch what its own body throws, so an interrupt raised by the wait needs
+    // handling of its own; without it the exception escapes raw and the flag stays cleared.
+    var attempts = 0
+    val failure = intercept[RuntimeException] {
+      RetryUtil.withBackoff(
+        "reach the store",
+        5,
+        200L,
+        noopRetryHook,
+        _ => throw new InterruptedException("interrupted")
+      ) {
+        attempts += 1
+        throw new RuntimeException("transient")
+      }
+    }
+    assert(attempts == 1)
+    assert(Thread.interrupted())
+    assert(failure.getMessage == "Interrupted while waiting to reach the store")
+    assert(failure.getCause.isInstanceOf[InterruptedException])
+  }
+
+  it should "not retry a fatal throwable, and let it through unwrapped" in {
+    // Only `NonFatal` failures are transient. A control throwable stands in for a fatal here
+    // because a real `VirtualMachineError` would abort the suite rather than be caught.
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    object Fatal extends ControlThrowable
+    intercept[ControlThrowable] {
+      RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _) {
+        attempts += 1
+        throw Fatal
+      }
+    }
+    assert(attempts == 1)
+    assert(delays.isEmpty)
+  }
+
+  it should "retry a non-fatal Error, which the predicate admits despite it not being an Exception" in {
+    // `NonFatal` is wider than `Exception`: an `AssertionError` (or `java.io.IOError`,
+    // `ServiceConfigurationError`) is transient here, so it is retried and then wrapped like any
+    // other failure. Pinned because it is the one behavior difference from the `case e: Exception`
+    // loops this util replaced.
+    val delays = ListBuffer.empty[Long]
+    var attempts = 0
+    val cause = new AssertionError("assertion blew up")
+    val failure = intercept[RuntimeException] {
+      RetryUtil.withBackoff("reach the store", 3, 200L, noopRetryHook, delays += _) {
+        attempts += 1
+        throw cause
+      }
+    }
+    assert(attempts == 3)
+    assert(delays.toList == List(200L, 400L))
+    assert(failure.getCause eq cause)
+  }
+}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala
index 8e92ecb9415..e79bf63b1df 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala
@@ -24,6 +24,7 @@ import io.lakefs.clients.sdk._
 import io.lakefs.clients.sdk.model.ResetCreation.TypeEnum
 import io.lakefs.clients.sdk.model._
 import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.common.util.RetryUtil
 
 import java.io.{File, FileOutputStream, InputStream}
 import java.net.URI
@@ -75,53 +76,16 @@ object LakeFSStorageClient extends LazyLogging {
   private val branchName: String = "main"
 
   def healthCheck(): Unit = {
-    retryWithBackoff(HealthCheckMaxAttempts, HealthCheckInitialDelayMillis) {
+    RetryUtil.withBackoff(
+      description = "connect to lake fs server",
+      maxAttempts = HealthCheckMaxAttempts,
+      initialDelayMillis = HealthCheckInitialDelayMillis,
+      onRetry = attempt => logger.warn(attempt.message)
+    ) {
       this.healthCheckApi.healthCheck().execute()
     }
   }
 
-  /**
-    * Runs `operation`, retrying on failure with exponential backoff (the delay
-    * doubles after each failed attempt) until it succeeds or `maxAttempts` is
-    * reached. The final failure is rethrown with the last exception as its cause.
-    * If interrupted while waiting, restores the interrupt status and fails fast.
-    *
-    * `sleep` is injectable so the backoff can be exercised in tests without real waiting.
-    */
-  private[util] def retryWithBackoff(
-      maxAttempts: Int,
-      initialDelayMillis: Long,
-      sleep: Long => Unit = Thread.sleep
-  )(operation: => Unit): Unit = {
-    var attempt = 1
-    var delayMillis = initialDelayMillis
-    while (true) {
-      try {
-        operation
-        return
-      } catch {
-        case ie: InterruptedException =>
-          // Restore the interrupt status and fail fast rather than retrying.
-          Thread.currentThread().interrupt()
-          throw new RuntimeException("Interrupted while waiting to retry lake fs health check", ie)
-        case e: Exception =>
-          if (attempt >= maxAttempts) {
-            throw new RuntimeException(
-              s"Failed to connect to lake fs server after $maxAttempts attempts: ${e.getMessage}",
-              e
-            )
-          }
-          logger.warn(
-            s"LakeFS not reachable (attempt $attempt/$maxAttempts): ${e.getMessage}. " +
-              s"Retrying in ${delayMillis}ms..."
-          )
-          sleep(delayMillis)
-          attempt += 1
-          delayMillis *= 2
-      }
-    }
-  }
-
   /**
     * Initializes a new repository in LakeFS.
     *
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
index ff5dca57d99..ad2bb389775 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala
@@ -21,55 +21,10 @@ package org.apache.texera.amber.core.storage.util
 
 import org.scalatest.flatspec.AnyFlatSpec
 
-import scala.collection.mutable.ListBuffer
-
 class LakeFSStorageClientSpec extends AnyFlatSpec {
 
-  "retryWithBackoff" should "run the operation once and not sleep when it succeeds immediately" in {
-    var attempts = 0
-    val delays = ListBuffer.empty[Long]
-    LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) {
-      attempts += 1
-    }
-    assert(attempts == 1)
-    assert(delays.isEmpty)
-  }
-
-  it should "retry until success and double the delay after each failed attempt" in {
-    var attempts = 0
-    val delays = ListBuffer.empty[Long]
-    LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) {
-      attempts += 1
-      if (attempts < 3) throw new RuntimeException("transient")
-    }
-    assert(attempts == 3)
-    assert(delays.toList == List(200L, 400L))
-  }
-
-  it should "give up after maxAttempts and preserve the last failure as the cause" in {
-    var attempts = 0
-    val cause = new RuntimeException("still down")
-    val ex = intercept[RuntimeException] {
-      LakeFSStorageClient.retryWithBackoff(3, 200L, _ => ()) {
-        attempts += 1
-        throw cause
-      }
-    }
-    assert(attempts == 3)
-    assert(ex.getMessage.contains("after 3 attempts"))
-    assert(ex.getCause eq cause)
-  }
-
-  it should "fail fast and restore the interrupt status when interrupted" in {
-    val ex = intercept[RuntimeException] {
-      LakeFSStorageClient.retryWithBackoff(5, 200L, _ => ()) {
-        throw new InterruptedException("interrupted")
-      }
-    }
-    // Thread.interrupted() both reads and clears the flag, so the interrupt was restored.
-    assert(Thread.interrupted())
-    assert(ex.getCause.isInstanceOf[InterruptedException])
-  }
+  // `healthCheck` retries through the shared `RetryUtil.withBackoff`; that contract (progression,
+  // give-up wrapping, interrupt fail-fast) is covered by `RetryUtilSpec` in `common/util`.
 
   "parsePhysicalAddress" should "split a well-formed address into bucket and key" in {
     assert(
diff --git a/file-service/src/main/scala/org/apache/texera/service/FileService.scala b/file-service/src/main/scala/org/apache/texera/service/FileService.scala
index d8e3ff122d5..1bb29f5dab3 100644
--- a/file-service/src/main/scala/org/apache/texera/service/FileService.scala
+++ b/file-service/src/main/scala/org/apache/texera/service/FileService.scala
@@ -26,6 +26,7 @@ import io.dropwizard.configuration.{EnvironmentVariableSubstitutor, Substituting
 import io.dropwizard.core.Application
 import io.dropwizard.core.setup.{Bootstrap, Environment}
 import org.apache.texera.common.config.StorageConfig
+import org.apache.texera.common.util.RetryUtil
 import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
 import org.apache.texera.auth.{AuthFeatures, RequestLoggingFilter, RoleAnnotationEnforcer}
 import org.apache.texera.dao.SqlServer
@@ -70,11 +71,11 @@ class FileService extends Application[FileServiceConfiguration] with LazyLogging
     )
 
     // check if the texera dataset bucket exists, if not create it
-    awaitDependency("texera dataset bucket") {
+    awaitDependency("reach the texera dataset bucket") {
       S3StorageClient.createBucketIfNotExist(StorageConfig.lakefsBucketName)
     }
     // ensure the large-binary S3 bucket exists before any workflow execution attempts to use it
-    awaitDependency("large-binary bucket") {
+    awaitDependency("reach the large-binary bucket") {
       S3StorageClient.createBucketIfNotExist(LargeBinaryManager.DEFAULT_BUCKET)
     }
     // check if we can connect to the lakeFS service
@@ -121,52 +122,24 @@ class FileService extends Application[FileServiceConfiguration] with LazyLogging
         .manage(new StagedFileCleanupJob(retentionHours, intervalMinutes))
 
   /**
-    * Runs `operation`, retrying with exponential backoff until it succeeds or `maxAttempts` is
-    * reached, to tolerate a slow-to-start object store. The last failure is rethrown as the cause.
-    * `sleep` is injectable for tests. Defaults: 6 attempts from 200ms (200, 400, 800, 1600, 3200), ~6s.
+    * Waits for a startup dependency (a slow-to-start object store) via the shared backoff retry,
+    * logging each retry under this service's logger. `description` is a verb phrase, e.g.
+    * "reach the texera dataset bucket". `sleep` is injectable for tests.
+    * Defaults: 6 attempts from 200ms (200, 400, 800, 1600, 3200), ~6s.
     */
   private[service] def awaitDependency(
       description: String,
       maxAttempts: Int = 6,
       initialDelayMillis: Long = 200L,
       sleep: Long => Unit = Thread.sleep
-  )(operation: => Unit): Unit = {
-    // Restore the interrupt status and fail fast rather than retrying, whether the
-    // interrupt arrives while running `operation` or while sleeping between attempts.
-    def failInterrupted(ie: InterruptedException): Nothing = {
-      Thread.currentThread().interrupt()
-      throw new RuntimeException(s"Interrupted while waiting for $description", ie)
-    }
-
-    var attempt = 1
-    var delayMillis = initialDelayMillis
-    while (true) {
-      try {
-        operation
-        return
-      } catch {
-        case ie: InterruptedException => failInterrupted(ie)
-        case e: Exception =>
-          if (attempt >= maxAttempts) {
-            throw new RuntimeException(
-              s"$description not ready after $maxAttempts attempts: ${e.getMessage}",
-              e
-            )
-          }
-          logger.warn(
-            s"$description not ready (attempt $attempt/$maxAttempts): ${e.getMessage}. " +
-              s"Retrying in ${delayMillis}ms..."
-          )
-          try {
-            sleep(delayMillis)
-          } catch {
-            case ie: InterruptedException => failInterrupted(ie)
-          }
-          attempt += 1
-          delayMillis *= 2
-      }
-    }
-  }
+  )(operation: => Unit): Unit =
+    RetryUtil.withBackoff(
+      description,
+      maxAttempts,
+      initialDelayMillis,
+      attempt => logger.warn(attempt.message),
+      sleep
+    )(operation)
 }
 
 object FileService {
diff --git a/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala
index 3deb60b6f76..71606c52360 100644
--- a/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala
+++ b/file-service/src/test/scala/org/apache/texera/service/FileServiceSpec.scala
@@ -34,7 +34,7 @@ class FileServiceSpec extends AnyFlatSpec {
   "awaitDependency" should "run the operation once and not sleep when it succeeds immediately" in {
     var attempts = 0
     val delays = ListBuffer.empty[Long]
-    service.awaitDependency("dep", 6, 200L, delays += _) {
+    service.awaitDependency("reach dep", 6, 200L, delays += _) {
       attempts += 1
     }
     assert(attempts == 1)
@@ -45,7 +45,7 @@ class FileServiceSpec extends AnyFlatSpec {
     // Exercises the default maxAttempts/initialDelay/sleep parameters: a first-try success
     // returns without ever invoking the (real Thread.sleep) default backoff.
     var attempts = 0
-    service.awaitDependency("dep") {
+    service.awaitDependency("reach dep") {
       attempts += 1
     }
     assert(attempts == 1)
@@ -54,7 +54,7 @@ class FileServiceSpec extends AnyFlatSpec {
   it should "retry until success and double the delay after each failed attempt" in {
     var attempts = 0
     val delays = ListBuffer.empty[Long]
-    service.awaitDependency("dep", 6, 200L, delays += _) {
+    service.awaitDependency("reach dep", 6, 200L, delays += _) {
       attempts += 1
       if (attempts < 3) throw new RuntimeException("not reachable yet")
     }
@@ -66,7 +66,7 @@ class FileServiceSpec extends AnyFlatSpec {
     var attempts = 0
     val delays = ListBuffer.empty[Long]
     val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 6, 200L, delays += _) {
+      service.awaitDependency("reach dep", 6, 200L, delays += _) {
         attempts += 1
         throw new RuntimeException("down")
       }
@@ -81,7 +81,7 @@ class FileServiceSpec extends AnyFlatSpec {
     var attempts = 0
     val cause = new RuntimeException("still down")
     val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 3, 200L, _ => ()) {
+      service.awaitDependency("reach dep", 3, 200L, _ => ()) {
         attempts += 1
         throw cause
       }
@@ -92,96 +92,35 @@ class FileServiceSpec extends AnyFlatSpec {
     assert(ex.getCause eq cause)
   }
 
-  it should "give up immediately without sleeping when maxAttempts is 1" in {
-    var attempts = 0
-    val delays = ListBuffer.empty[Long]
-    val cause = new RuntimeException("still down")
-    val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 1, 200L, delays += _) {
-        attempts += 1
-        throw cause
-      }
-    }
-    assert(attempts == 1)
-    assert(delays.isEmpty)
-    assert(ex.getMessage.contains("after 1 attempts"))
-    assert(ex.getCause eq cause)
-  }
-
   it should "fail fast and restore the interrupt status when the operation is interrupted" in {
     val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 6, 200L, _ => ()) {
+      service.awaitDependency("reach dep", 6, 200L, _ => ()) {
         throw new InterruptedException("interrupted")
       }
     }
     // Thread.interrupted() both reads and clears the flag, so the interrupt was restored.
     assert(Thread.interrupted())
-    assert(ex.getMessage.contains("Interrupted while waiting for dep"))
-    assert(ex.getCause.isInstanceOf[InterruptedException])
-  }
-
-  it should "fail fast and restore the interrupt status when interrupted while sleeping between attempts" in {
-    var attempts = 0
-    val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 6, 200L, _ => throw new InterruptedException("interrupted")) {
-        attempts += 1
-        throw new RuntimeException("not reachable yet")
-      }
-    }
-    // The operation failed once, then the interrupt arrived during the backoff sleep.
-    assert(attempts == 1)
-    // Thread.interrupted() both reads and clears the flag, so the interrupt was restored.
-    assert(Thread.interrupted())
-    assert(ex.getMessage.contains("Interrupted while waiting for dep"))
+    assert(ex.getMessage.contains("Interrupted while waiting to reach dep"))
     assert(ex.getCause.isInstanceOf[InterruptedException])
   }
 
-  it should "succeed on the final allowed attempt without giving up one try too early" in {
-    // Boundary for `attempt >= maxAttempts`: the operation only succeeds on the very last
-    // attempt, so the loop must not give up prematurely. Expect maxAttempts - 1 backoff waits.
-    var attempts = 0
-    val delays = ListBuffer.empty[Long]
-    service.awaitDependency("dep", 3, 200L, delays += _) {
-      attempts += 1
-      if (attempts < 3) throw new RuntimeException("not reachable yet")
-    }
-    assert(attempts == 3)
-    assert(delays.toList == List(200L, 400L))
-  }
-
-  it should "honor a custom initial delay when computing the backoff progression" in {
-    // Guards against the initial delay being hardcoded: starting from 50ms the geometric
-    // progression must be 50, 100, 200 rather than the default 200-based sequence.
-    var attempts = 0
-    val delays = ListBuffer.empty[Long]
-    val ex = intercept[RuntimeException] {
-      service.awaitDependency("dep", 4, 50L, delays += _) {
-        attempts += 1
-        throw new RuntimeException("down")
-      }
-    }
-    assert(attempts == 4)
-    assert(delays.toList == List(50L, 100L, 200L))
-    assert(ex.getMessage.contains("after 4 attempts"))
-  }
-
-  it should "include the underlying failure message when giving up" in {
+  it should "include the description and the underlying failure message when giving up" in {
     val ex = intercept[RuntimeException] {
-      service.awaitDependency("dataset bucket", 2, 200L, _ => ()) {
+      service.awaitDependency("reach the dataset bucket", 2, 200L, _ => ()) {
         throw new RuntimeException("connection refused")
       }
     }
-    assert(ex.getMessage.contains("dataset bucket not ready after 2 attempts"))
+    assert(ex.getMessage.contains("Failed to reach the dataset bucket after 2 attempts"))
     assert(ex.getMessage.contains("connection refused"))
   }
 
-  it should "propagate a non-Exception Throwable immediately without retrying or wrapping it" in {
-    // The catch clause only matches Exception, so an Error must escape on the first attempt:
-    // it is neither retried nor wrapped in the \"not ready after N attempts\" RuntimeException.
+  it should "propagate a fatal Throwable immediately without retrying or wrapping it" in {
+    // A fatal throwable is not transient, so it must escape on the first attempt: it is neither
+    // retried nor wrapped in the "Failed to ... after N attempts" RuntimeException.
     var attempts = 0
     val delays = ListBuffer.empty[Long]
     val err = intercept[StackOverflowError] {
-      service.awaitDependency("dep", 6, 200L, delays += _) {
+      service.awaitDependency("reach dep", 6, 200L, delays += _) {
         attempts += 1
         throw new StackOverflowError("boom")
       }