[SPARK-58985][CORE] Fix HistoryServerDiskManager double-counting store size on concurrent release and makeRoom

### What changes were proposed in this pull request?

Make the disk usage accounting in `HistoryServerDiskManager` atomic with the store operation it accompanies:

- `release()` now performs the whole operation -- removing the app from the `active` map, updating usage accounting, and deleting or re-measuring the store -- under the `active` lock. When the store directory is already gone, it also drops the listing entry so the store is not deducted again later.
- `Lease.commit()` now moves the store into place, updates usage accounting, writes the listing entry, and registers the app in the `active` map under the `active` lock, so the store is never on disk without being tracked. A failed rename raises an `IOException` instead of recording a store that is not there. When the previous store directory is already gone but still listed, its listing entry and accounting are dropped before the new store takes its place.
- `openStore()` now writes the listing entry under the `active` lock, so a concurrent `release()` cannot leave a listing entry for a deleted store.
- `makeRoom()` re-checks each eviction candidate under the `active` lock before deleting it: it skips candidates that became active or that a concurrent `release()` or `commit()` already deleted, and deducts the size the listing holds at that point. The summary log reports the stores actually deleted and the space actually freed, and warns when nothing could be freed. On that log line, `NUM_BYTES_EVICTED` now carries the freed bytes and `NUM_BYTES_TO_FREE` the target, and the store count moves to `NUM_APPS`; consumers of these structured-logging fields should note the change.

The lock now covers directory I/O (`sizeOf`, rename, deletion, listing read/write). As a trade-off, `openStore()` from UI requests can block while a concurrent `release()` (e.g., in the `cleanLogs` loop) or `makeRoom()` eviction deletes a store; this is deliberate to keep the accounting accurate.

### Why are the changes needed?

`HistoryServerDiskManager` can deduct the same store size twice, driving the committed usage negative and making the History Server throw `IllegalStateException: Disk usage tracker went negative`.

The race is longstanding: `release()` updates usage and operates on the store directory outside the `active` lock, `commit()` moves the store into place before registering it, and `makeRoom()` deletes eviction candidates without re-checking, so two paths have been able to deduct the same store since the disk manager was introduced by SPARK-20654 (2.3.0).

SPARK-56044 (4.0.3) widened the race. By adding a deduction in `release()` based on the size measured from disk for apps not in the `active` map, a double deduction no longer requires the application to be actively open:

- `release(delete = true)` vs `openStore()`: `release()` deducts the measured size and deletes the store, but a concurrent `openStore()` re-registered the app in `active`, so a subsequent `release()` deducts the size again.
- `release(delete = true)` vs `commit()`: `commit()` has moved the store into place but not yet registered it, so `release()` deducts the measured size and deletes the store; `commit()` then registers the deleted store, and a subsequent `release()` deducts the size again.
- `release(delete = true)` vs `makeRoom()`: both paths deduct the size of the same store.

This makes the race reachable in normal History Server operation, e.g. log cleanup calling `release(delete = true)` for an app never opened after a restart while a concurrent UI request opens, loads, or evicts the same store. The fix also closes two concurrent `makeRoom()` calls evicting the same store twice.

This crash was observed on a production History Server (4.1-based build), in the periodic log cleanup path:

```
java.lang.IllegalStateException: Disk usage tracker went negative (now = -118595158, delta = -151974353)
  at o.a.s.deploy.history.HistoryServerDiskManager.updateUsage(HistoryServerDiskManager.scala:285)
  at o.a.s.deploy.history.HistoryServerDiskManager.release(HistoryServerDiskManager.scala:186)
  at o.a.s.deploy.history.FsHistoryProvider.cleanAppData(FsHistoryProvider.scala:746)
  at o.a.s.deploy.history.FsHistoryProvider.deleteAttemptLogs(FsHistoryProvider.scala:1132)
  at o.a.s.deploy.history.FsHistoryProvider.cleanLogs(FsHistoryProvider.scala:1063)
  at o.a.s.deploy.history.FsHistoryProvider.$anonfun$startPolling$4(FsHistoryProvider.scala:305)
```

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Added five tests to `HistoryServerDiskManagerSuite`. The three race tests (`release with delete is atomic with openStore`, `makeRoom is atomic with release`, `commit is atomic with release`) park one operation mid-way, run the competing operation on another thread, and let the parked one proceed once the competing thread has finished or blocked on the lock, so none relies on a fixed wait. Each fails on the pre-fix code, the first with `openStore()` handing out a store being deleted, the other two with `IllegalStateException: Disk usage tracker went negative`. `makeRoom deducts a store deleted out of band` and `commit deducts a store deleted out of band` pin the accounting for stores removed outside the History Server.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Qwen 3.8 Max

Closes #58312 from pan3793/SPARK-58985.

Authored-by: Cheng Pan <pan3793@gmail.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>
(cherry picked from commit ff9d3e3)
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: pan3793

core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy.history
 
-import java.io.File
+import java.io.{File, IOException}
 import java.util.concurrent.atomic.AtomicLong
 
 import scala.collection.mutable.{HashMap, ListBuffer}
@@ -67,6 +67,8 @@ private class HistoryServerDiskManager(
   private val maxUsage = conf.get(MAX_LOCAL_DISK_USAGE)
   private val currentUsage = new AtomicLong(0L)
   private val committedUsage = new AtomicLong(0L)
+  // Besides the map itself, this lock guards the store directories, their listing entries and
+  // the committed usage: whether a store is still counted is decided and acted on in one hold.
   private val active = new HashMap[(String, Option[String]), Long]()
 
   def initialize(): Unit = {
@@ -139,60 +141,64 @@ private class HistoryServerDiskManager(
    * being used so that it's not evicted when running out of designated space.
    */
   def openStore(appId: String, attemptId: Option[String]): Option[File] = {
-    var newSize: Long = 0
-    val storePath = active.synchronized {
+    active.synchronized {
       val path = appStorePath(appId, attemptId)
       if (path.isDirectory()) {
-        newSize = sizeOf(path)
+        val newSize = sizeOf(path)
         active(appId -> attemptId) = newSize
+        updateApplicationStoreInfo(appId, attemptId, newSize)
         Some(path)
       } else {
         None
       }
     }
-
-    storePath.foreach { path =>
-      updateApplicationStoreInfo(appId, attemptId, newSize)
-    }
-
-    storePath
   }
 
   /**
    * Tell the disk manager that the store for the given application is not being used anymore.
+   * A tracked store whose directory is already gone is dropped from the listing regardless of
+   * `delete`.
    *
    * @param delete Whether to delete the store from disk.
    */
   def release(appId: String, attemptId: Option[String], delete: Boolean = false): Unit = {
     // Because disk-based stores may modify the structure of the store files even when just reading,
-    // update the accounting for this application when it's closed.
-    val oldSizeOpt = active.synchronized {
-      active.remove(appId -> attemptId)
-    }
-
-    oldSizeOpt.foreach { oldSize =>
-      updateUsage(-oldSize, committed = true)
-    }
+    // update the accounting for this application when it's closed. Do the whole operation under
+    // the active lock so a concurrent openStore(), commit() or makeRoom() cannot deduct the same
+    // store size twice.
+    active.synchronized {
+      val oldSizeOpt = active.remove(appId -> attemptId)
+
+      oldSizeOpt.foreach { oldSize =>
+        updateUsage(-oldSize, committed = true)
+      }
 
-    // Apply the store operation regardless of whether the app was in the active map, since
-    // the store directory may still exist on disk (e.g., when the app was never opened after
-    // a restart).
-    val path = appStorePath(appId, attemptId)
-    if (path.isDirectory()) {
-      if (delete) {
-        // If the app was not actively tracked, its size was not deducted above; do it now.
-        if (oldSizeOpt.isEmpty) {
-          val size = sizeOf(path)
-          updateUsage(-size, committed = true)
+      // Apply the store operation regardless of whether the app was in the active map, since
+      // the store directory may still exist on disk (e.g., when the app was never opened after
+      // a restart).
+      val path = appStorePath(appId, attemptId)
+      if (path.isDirectory()) {
+        if (delete) {
+          // If the app was not actively tracked, its size was not deducted above; do it now.
+          // Measure it from disk rather than trusting the listing: a store directory without a
+          // listing entry is still counted, since initialize() measures the whole directory.
+          if (oldSizeOpt.isEmpty) {
+            val size = sizeOf(path)
+            updateUsage(-size, committed = true)
+          }
+          deleteStore(path)
+        } else if (oldSizeOpt.isDefined) {
+          // Re-measure the size since the store may have changed while it was open.
+          val newSize = sizeOf(path)
+          val newInfo = listing.read(classOf[ApplicationStoreInfo], path.getAbsolutePath())
+            .copy(size = newSize)
+          listing.write(newInfo)
+          updateUsage(newSize, committed = true)
         }
-        deleteStore(path)
       } else if (oldSizeOpt.isDefined) {
-        // Re-measure the size since the store may have changed while it was open.
-        val newSize = sizeOf(path)
-        val newInfo = listing.read(classOf[ApplicationStoreInfo], path.getAbsolutePath())
-          .copy(size = newSize)
-        listing.write(newInfo)
-        updateUsage(newSize, committed = true)
+        // The store directory is gone (e.g., deleted out of band) and its size was deducted
+        // above; drop its listing entry too, so makeRoom() does not deduct it again.
+        deleteStore(path)
       }
     }
   }
@@ -226,6 +232,15 @@ private class HistoryServerDiskManager(
     listing.delete(classOf[ApplicationStoreInfo], path.getAbsolutePath())
   }
 
+  /** Returns the listing entry of the store at `path`, if any. */
+  private def readStoreInfo(path: String): Option[ApplicationStoreInfo] = {
+    try {
+      Some(listing.read(classOf[ApplicationStoreInfo], path))
+    } catch {
+      case _: NoSuchElementException => None
+    }
+  }
+
   private def makeRoom(size: Long): Unit = {
     if (free() < size) {
       logDebug(s"Not enough free space, looking at candidates for deletion...")
@@ -246,18 +261,30 @@ private class HistoryServerDiskManager(
         }
       }
 
-      if (evicted.nonEmpty) {
-        val freed = evicted.map { info =>
-          logInfo(log"Deleting store for" +
-            log" ${MDC(APP_ID, info.appId)}/${MDC(APP_ATTEMPT_ID, info.attemptId)}.")
-          deleteStore(new File(info.path))
-          updateUsage(-info.size, committed = true)
-          info.size
-        }.sum
-
-        logInfo(log"Deleted ${MDC(NUM_BYTES_EVICTED, evicted.size)} store(s)" +
-          log" to free ${MDC(NUM_BYTES_TO_FREE, Utils.bytesToString(freed))}" +
-          log" (target = ${MDC(NUM_BYTES, Utils.bytesToString(size))}).")
+      var freedCount = 0
+      var freedBytes = 0L
+      evicted.foreach { candidate =>
+        active.synchronized {
+          // Re-check under the lock: since the candidate was collected, it may have become
+          // active, or a concurrent release() or commit() may have deleted it along with its
+          // listing entry. Deduct the size the listing holds now, which may have changed as well.
+          if (!active.contains(candidate.appId -> candidate.attemptId)) {
+            readStoreInfo(candidate.path).foreach { info =>
+              logInfo(log"Deleting store for" +
+                log" ${MDC(APP_ID, info.appId)}/${MDC(APP_ATTEMPT_ID, info.attemptId)}.")
+              deleteStore(new File(info.path))
+              updateUsage(-info.size, committed = true)
+              freedCount += 1
+              freedBytes += info.size
+            }
+          }
+        }
+      }
+
+      if (freedCount > 0) {
+        logInfo(log"Deleted ${MDC(NUM_APPS, freedCount)} store(s)" +
+          log" to free ${MDC(NUM_BYTES_EVICTED, Utils.bytesToString(freedBytes))}" +
+          log" (target = ${MDC(NUM_BYTES_TO_FREE, Utils.bytesToString(size))}).")
       } else {
         logWarning(log"Unable to free any space to make room for " +
           log"${MDC(NUM_BYTES, Utils.bytesToString(size))}.")
@@ -320,29 +347,39 @@ private class HistoryServerDiskManager(
           val size = sizeOf(dst)
           deleteStore(dst)
           updateUsage(-size, committed = true)
+        } else {
+          // The store directory is gone (e.g., deleted out of band) but may still be listed and
+          // counted; drop it before the new store takes over its listing entry.
+          readStoreInfo(dst.getAbsolutePath()).foreach { info =>
+            deleteStore(dst)
+            updateUsage(-info.size, committed = true)
+          }
         }
       }
 
       updateUsage(-leased)
 
       val newSize = sizeOf(tmpPath)
       makeRoom(newSize)
-      tmpPath.renameTo(dst)
 
-      updateUsage(newSize, committed = true)
+      // Move the store into place, account for it and mark the app active in one step, so a
+      // concurrent release() or makeRoom() cannot delete the store before it is tracked.
+      active.synchronized {
+        if (!tmpPath.renameTo(dst)) {
+          throw new IOException(s"Failed to move the store from $tmpPath to $dst")
+        }
+        updateUsage(newSize, committed = true)
+        updateApplicationStoreInfo(appId, attemptId, newSize)
+        active(appId -> attemptId) = newSize
+      }
+
       if (committedUsage.get() > maxUsage) {
         val current = Utils.bytesToString(committedUsage.get())
         val max = Utils.bytesToString(maxUsage)
         logWarning(log"Commit of application ${MDC(APP_ID, appId)} / " +
           log"${MDC(APP_ATTEMPT_ID, attemptId)} causes maximum disk usage to be " +
           log"exceeded (${MDC(NUM_BYTES, current)} > ${MDC(NUM_BYTES_MAX, max)}")
       }
-
-      updateApplicationStoreInfo(appId, attemptId, newSize)
-
-      active.synchronized {
-        active(appId -> attemptId) = newSize
-      }
       dst
     }