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

[pan3793]

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.
• makeRoom() re-checks each eviction candidate under the active lock before deleting it, skipping candidates that became active or whose store directory is already gone. It also removes the stale listing entry when the store directory is already gone, and prevents evicting a store that a concurrent openStore() just handed out. The summary log reports the stores actually deleted and the space actually freed.

The lock now covers directory I/O (sizeOf, deletion, listing read/write), the same pattern Lease.commit already uses under this lock. As a trade-off, openStore() from UI requests can block while a concurrent release() deletes a store (e.g., in the cleanLogs loop); 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, 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 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 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 a new test SPARK-58985: release with delete is atomic with openStore in HistoryServerDiskManagerSuite, which fails on the pre-fix code with IllegalStateException: Disk usage tracker went negative and passes with the fix.

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

Generated-by: Qwen 3.8 Max

[dongjoon-hyun]

Thanks for tracking this down. I walked through the four release() cases (delete x active membership) against the old code and the accounting outcome is preserved in each, so the change looks correct to me. Folding the deduction and the store operation under the active lock is also consistent with what Lease.commit and openStore already do in this class.

One thing worth calling out in the description: the makeRoom() re-check also fixes a second bug on its own -- previously a candidate collected before a concurrent openStore() could be deleted while the UI was holding the path.

I also checked the lock ordering. release() now does listing I/O under the active lock, while makeRoom() acquires active while holding an open listing iterator. That crossing already exists on master, and LevelDB writes only take the per-type monitor (not held by iterators), so I don't see a deadlock cycle.

A few minor comments:

1. In release(), the comment // If the app was not actively tracked, its size was not deducted above; do it now. is now stale -- there is no unconditional deduction above it anymore.

2. In the final else if (oldSizeOpt.isDefined) branch, the example in the comment (evicted by a concurrent makeRoom()) is exactly the case this PR eliminates: makeRoom() now skips active apps under the lock. The branch is still worth keeping as a defensive path (out-of-band deletion, a partially failed deleteRecursively), but the justification could be reworded.

3. When makeRoom() skips a candidate because its directory is gone, the ApplicationStoreInfo entry stays in the listing, since deleteStore() is what removes it. The window is narrow, but such an entry is permanent and gets counted against needed on every later makeRoom(), which under-evicts. A listing.delete(classOf[ApplicationStoreInfo], info.path) in that branch would clean it up. Optional -- it does widen the scope a bit.

4. release(delete = true) now holds the global active lock across the recursive store deletion, and FsHistoryProvider.cleanLogs calls it in a loop. openStore() from UI requests blocks for that duration. Not a new class of problem given the existing Lease.commit behavior, but it may be worth a note in the description that this is a deliberate trade-off.

5. In the test, after the fix the main thread blocks on the active lock inside openStore(), so openStoreDone.countDown() in the finally is never reached and the await(2, TimeUnit.SECONDS) inside the sizeOf stub always runs to timeout. The test is correct, but it costs a fixed 2 seconds on every run.

6. Nit: new File(info.path) is constructed twice in makeRoom(); a local val would do.

Finally, since SPARK-56044 shipped in 4.2.0 / 4.1.2 / 4.0.3, should this be backported to those branches as well?

[pan3793]

Thanks for the careful review!

1. Fixed -- reworded to "Use the tracked size if the app was active; otherwise measure it from disk."
2. Reworded to "The store directory is already gone (e.g., deleted out of band)"; kept the branch as a defensive path.
3. Done -- makeRoom() now drops the stale listing entry when the directory is already gone. In-process deleters always remove the entry under the same lock, so this branch only fires for out-of-band deletion; initialize() already sweeps such orphans at startup, this makes it immediate.
4. Added a note in the description (see the updated PR body).
5. Kept as is: the wait is what keeps the repro deterministic against the old code -- release() must be held until openStore() completes. In the fixed code openStore() is serialized behind the lock release() holds, so the latch can only release via the timeout; shortening it would weaken the repro without changing fixed-code coverage.
6. Done -- hoisted to a local val.

Backports: yes, SPARK-56044 shipped in 4.0.3/4.1.2/4.2.0, so the race is live there; will backport to branch-4.0/4.1/4.2/4.3/4.x once merged.

[dongjoon-hyun]

Thanks for tracking this down. The direction looks right to me: putting the whole of release() under the active lock and re-checking each eviction candidate under the same lock is consistent with what Lease.commit already does, and cleaning up the stale listing entry in makeRoom() is a nice addition. I ran the new test locally and confirmed it reproduces the reported failure on the pre-fix code:

- SPARK-58985: release with delete is atomic with openStore *** FAILED *** (2 seconds, 567 ms)
  java.lang.IllegalStateException: Disk usage tracker went negative (now = -2, delta = -2)

Three comments.

1. The new test no longer exercises the interleaving it describes

With the fix in place, the latch handshake cannot work: release() holds the active lock while it is blocked inside the mocked sizeOf(), so the main thread's openStore() blocks on that same lock and never reaches openStoreDone.countDown() in the finally. The release thread only proceeds when openStoreDone.await(2, TimeUnit.SECONDS) times out.

The runtimes show this — 2.588 s with the fix, 2.567 s without; the 2 seconds are pure timeout wait.

As a result release() always completes first and deletes the store, so openStore() always returns None, and this block plus its comment are dead code after the fix:

// If openStore() handed out a path, the subsequent release in FsHistoryProvider must not
// deduct the size again.
opened.foreach { _ =>
  manager.release("app1", None, delete = true)
}

It is still a valid regression test, but it would read much better if it asserted the post-fix behavior explicitly (release() wins the lock, therefore openStore() returns None). Running openStore() on its own thread and counting the latch down from there would also let both suites finish without burning ~2.5 s each on a timeout.

2. Lease.commit() vs release(delete = true) leaves the same crash reachable

commit() does tmpPath.renameTo(dst) and updateUsage(newSize, committed = true) outside the lock, and only registers the app in active in a later synchronized block. In that window the store is on disk but not in active — exactly the shape this PR fixes for openStore():

commit:  rename tmp -> dst;  committed += newSize
release: (lock) not active -> deducts sizeOf(dst), deleteStore(dst)
commit:  active(app) = newSize            // registers a store that is already gone
release: (lock) deducts oldSize again -> committed goes negative

I verified this with a scratch test that blocks inside commit() between the rename and the active insertion (hooking Clock.getTimeMillis(), which updateApplicationStoreInfo calls in that window). It fails identically on this branch and on master:

- commit vs release(delete = true) *** FAILED ***
  java.lang.IllegalStateException: Disk usage tracker went negative (now = -2, delta = -2)
  at ...HistoryServerDiskManager.release(HistoryServerDiskManager.scala:195)

scratch test

test("commit vs release(delete = true)") {
  val conf2 = new SparkConf().set(MAX_LOCAL_DISK_USAGE, MAX_USAGE)
  val armed = new java.util.concurrent.atomic.AtomicBoolean(false)
  val inCommit = new CountDownLatch(1)
  val releaseDone = new CountDownLatch(1)
  val clock = new ManualClock() {
    override def getTimeMillis(): Long = {
      if (armed.get()) {
        inCommit.countDown()
        releaseDone.await(10, TimeUnit.SECONDS)
      }
      super.getTimeMillis()
    }
  }
  val manager = spy[HistoryServerDiskManager](
    new HistoryServerDiskManager(conf2, testDir, store, clock))
  doAnswer(AdditionalAnswers.returnsFirstArg[Long]()).when(manager)
    .approximateSize(anyLong(), anyBoolean())

  val leaseA = manager.lease(2)
  doReturn(2L).when(manager).sizeOf(meq(leaseA.tmpPath))
  val dst = leaseA.commit("app1", None)
  doReturn(2L).when(manager).sizeOf(meq(dst))
  manager.release("app1", None)
  assert(manager.committed() === 2)

  val leaseB = manager.lease(2)
  doReturn(2L).when(manager).sizeOf(meq(leaseB.tmpPath))
  armed.set(true)
  val commitThread = new Thread(() => leaseB.commit("app1", None))
  commitThread.start()
  inCommit.await(10, TimeUnit.SECONDS)
  // commit() has renamed the store into place and added its size, but has not yet
  // registered the app in the active map.
  manager.release("app1", None, delete = true)
  releaseDone.countDown()
  commitThread.join(TimeUnit.SECONDS.toMillis(10))
  assert(manager.committed() === 0)
  // The provider that opened the store eventually releases it.
  manager.release("app1", None, delete = true)
  assert(manager.committed() === 0)
}

This is pre-existing rather than something this PR introduces, and the call sites are real: createDiskStore() serving a UI request can run against cleanLogs() calling release(delete = true) for the same app. makeRoom() vs commit() has the same gap — a store being committed is not in active, so it can still pass the new re-check and be deleted, and the info.size deducted there is a stale snapshot value.

Given the PR title says "double-counting store size on concurrent release and makeRoom", readers will likely assume the whole class of bug is closed. Could you either note the remaining window in the description, or close it here as well by reserving the active entry before the rename?

3. Minor

• In makeRoom(), freed is a ListBuffer[Long] but only freed.size and freed.sum are used; two counters (var freedBytes = 0L, var freedCount = 0) would be simpler.
• openStore() still calls updateApplicationStoreInfo outside the lock, so a release() that deletes the store in between resurrects a listing entry for a path that no longer exists. The accounting stays correct and the new stale-entry cleanup in makeRoom() eventually removes it, so this only skews the needed computation during an eviction scan — not a blocker, just noting it.
• The wider lock now covers deleteStore() (a recursive delete of a potentially large RocksDB directory) in both release() and makeRoom(), so UI-driven openStore() calls can block behind it. The trade-off is already called out in the description; just flagging that makeRoom() inherits it too.