Exclusive Cache Lock

android-test/src/androidTest/java/okhttp/android/test/OkHttpTest.kt

@@ -859,6 +859,7 @@ class OkHttpTest {
     assertEquals(setOf(OkHttpTest::class.java.name), testHandler.calls.keys)
   }
 
+  @Test
   fun testCachedRequest() {
     enableTls()
 

okcurl/src/test/kotlin/okhttp3/curl/MainTest.kt

@@ -22,7 +22,9 @@ import assertk.assertions.startsWith
 import java.io.IOException
 import kotlin.test.Test
 import okhttp3.RequestBody
+import okhttp3.TestUtil
 import okio.Buffer
+import org.junit.jupiter.api.BeforeAll
 
 class MainTest {
   @Test
@@ -137,5 +139,12 @@ class MainTest {
         throw RuntimeException(e)
       }
     }
+
+    @JvmStatic
+    @BeforeAll
+    fun beforeAll() {
+//    https://github.com/square/okhttp/issues/8342
+      TestUtil.assumeNotWindows()
+    }
   }
 }

okcurl/src/test/kotlin/okhttp3/curl/OkcurlTest.kt

@@ -16,10 +16,21 @@
 package okhttp3.curl
 
 import kotlin.test.Test
+import okhttp3.TestUtil
+import org.junit.jupiter.api.BeforeAll
 
 class OkcurlTest {
   @Test
   fun simple() {
     Main().main(listOf("--help"))
   }
+
+  companion object {
+    @JvmStatic
+    @BeforeAll
+    fun beforeAll() {
+//    https://github.com/square/okhttp/issues/8342
+      TestUtil.assumeNotWindows()
+    }
+  }
 }

okhttp-dnsoverhttps/src/test/java/okhttp3/dnsoverhttps/DnsOverHttpsTest.kt

@@ -170,7 +170,7 @@ class DnsOverHttpsTest {
   // 5. unsuccessful response
   @Test
   fun usesCache() {
-    val cache = Cache("cache".toPath(), (100 * 1024).toLong(), cacheFs)
+    val cache = Cache("cache-usesCache".toPath(), (100 * 1024).toLong(), cacheFs)
     val cachedClient = bootstrapClient.newBuilder().cache(cache).build()
     val cachedDns = buildLocalhost(cachedClient, false)
 
@@ -208,7 +208,7 @@ class DnsOverHttpsTest {
 
   @Test
   fun usesCacheEvenForPost() {
-    val cache = Cache("cache".toPath(), (100 * 1024).toLong(), cacheFs)
+    val cache = Cache("cache-usesCacheEvenForPost".toPath(), (100 * 1024).toLong(), cacheFs)
     val cachedClient = bootstrapClient.newBuilder().cache(cache).build()
     val cachedDns = buildLocalhost(cachedClient, false, post = true)
     repeat(2) {
@@ -241,6 +241,8 @@ class DnsOverHttpsTest {
     assertThat(recordedRequest.method).isEqualTo("POST")
     assertThat(recordedRequest.path)
       .isEqualTo("/lookup?ct")
+
+    cache.close()
   }
 
   @Test
@@ -282,6 +284,8 @@ class DnsOverHttpsTest {
     assertThat(recordedRequest!!.method).isEqualTo("GET")
     assertThat(recordedRequest.path)
       .isEqualTo("/lookup?ct&dns=AAABAAABAAAAAAAABmdvb2dsZQNjb20AAAEAAQ")
+
+    cache.close()
   }
 
   private fun dnsResponse(s: String): MockResponse {

okhttp/build.gradle.kts

@@ -131,6 +131,7 @@ dependencies {
   testImplementation(projects.okhttpSse)
   testImplementation(projects.okhttpCoroutines)
   testImplementation(libs.kotlinx.coroutines.core)
+  testImplementation(libs.kotlinx.coroutines.test)
   testImplementation(libs.squareup.moshi)
   testImplementation(libs.squareup.moshi.kotlin)
   testImplementation(libs.squareup.okio.fakefilesystem)

okhttp/src/main/kotlin/okhttp3/internal/cache/CacheLock.kt

@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2024 Block, Inc.
+ *
+ * Licensed 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 okhttp3.internal.cache
+
+import android.annotation.SuppressLint
+import java.io.Closeable
+import java.io.File
+import java.nio.channels.FileChannel
+import java.nio.file.StandardOpenOption
+import java.util.Collections
+import okio.FileSystem
+import okio.Path
+import okio.Path.Companion.toOkioPath
+import org.codehaus.mojo.animal_sniffer.IgnoreJRERequirement
+
+/**
+ * An implementation of directory locking to ensure exclusive access in a Cache instance.
+ * Always applies for the current process, and uses a file system lock if supported.
+ */
+internal object CacheLock {
+  private val openCaches = Collections.synchronizedMap(mutableMapOf<Path, Exception>())
+
+  /**
+   * Open a lock, which if successful remains until the returned [Closeable] is closed.
+   * The lock file will be a file with name "lock" inside directory.
+   *
+   * @param fileSystem the file system containing the lock files.
+   * @param directory the cache directory.
+   */
+  fun openLock(
+    fileSystem: FileSystem,
+    directory: Path,
+  ): Closeable {
+    val memoryLock = inMemoryLock(directory)
+
+    // check if possibly a non System file system
+    if (FileSystem.SYSTEM.exists(directory)) {
+      try {
+        val fileSystemLock = fileSystemLock(directory)
+
+        return Closeable {
+          memoryLock.close()
+          fileSystemLock.close()
+        }
+      } catch (le: LockException) {
+        if (fileSystemSupportsLock(fileSystem)) {
+          memoryLock.close()
+          throw le
+        }
+      }
+    }
+
+    return memoryLock
+  }
+
+  @IgnoreJRERequirement // Not called on legacy Android
+  internal fun fileSystemSupportsLock(fileSystem: FileSystem): Boolean {
+    val tmpLockFile = File.createTempFile("test-", ".lock")
+
+    if (!fileSystem.exists(tmpLockFile.toOkioPath())) {
+      return false
+    }
+
+    val channel = FileChannel.open(tmpLockFile.toPath(), StandardOpenOption.APPEND)
+
+    return channel.tryLock().apply { close() } != null
+  }
+
+  /**
+   * Create an in-memory lock, avoiding two open Cache instances.
+   */
+  @SuppressLint("NewApi")
+  @IgnoreJRERequirement // D8 supports put if absent
+  internal fun inMemoryLock(directory: Path): Closeable {
+    val existing = openCaches.putIfAbsent(directory, LockException("Existing CacheLock($directory) opened at"))
+    if (existing != null) {
+      throw LockException("Cache already open at '$directory' in same process", existing)
+    }
+    return okio.Closeable {
+      openCaches.remove(directory)
+    }
+  }
+
+  /**
+   * Create a file system lock, that excludes other processes. However within the process a
+   * memory lock is also needed, since locks don't work within a single process.
+   */
+  @SuppressLint("NewApi")
+  @IgnoreJRERequirement // Not called on legacy Android
+  internal fun fileSystemLock(directory: Path): Closeable {
+    // update once https://github.com/square/okio/issues/1464 is available
+
+    val lockFile = directory / "lock"
+    lockFile.toFile().createNewFile()
+    val channel = FileChannel.open(lockFile.toNioPath(), StandardOpenOption.APPEND)
+
+    channel.tryLock() ?: throw LockException("Cache already open at '$directory' in another process")
+
+    return okio.Closeable {
+      channel.close()
+    }
+  }
+}
+
+class LockException(message: String, cause: Exception? = null) : Exception(message, cause)

okhttp/src/main/kotlin/okhttp3/internal/cache/DiskLruCache.kt

@@ -20,6 +20,7 @@ import java.io.EOFException
 import java.io.Flushable
 import java.io.IOException
 import okhttp3.internal.assertThreadHoldsLock
+import okhttp3.internal.cache.CacheLock.openLock
 import okhttp3.internal.cache.DiskLruCache.Editor
 import okhttp3.internal.closeQuietly
 import okhttp3.internal.concurrent.Task
@@ -95,6 +96,8 @@ class DiskLruCache(
   /** Used for asynchronous journal rebuilds. */
   taskRunner: TaskRunner,
 ) : Closeable, Flushable {
+  lateinit var cacheLock: Closeable
+
   internal val fileSystem: FileSystem =
     object : ForwardingFileSystem(fileSystem) {
       override fun sink(
@@ -242,33 +245,42 @@ class DiskLruCache(
 
     civilizedFileSystem = fileSystem.isCivilized(journalFileBackup)
 
-    // Prefer to pick up where we left off.
-    if (fileSystem.exists(journalFile)) {
-      try {
-        readJournal()
-        processJournal()
-        initialized = true
-        return
-      } catch (journalIsCorrupt: IOException) {
-        Platform.get().log(
-          "DiskLruCache $directory is corrupt: ${journalIsCorrupt.message}, removing",
-          WARN,
-          journalIsCorrupt,
-        )
-      }
+    cacheLock = openLock(fileSystem, directory)
 
-      // The cache is corrupted, attempt to delete the contents of the directory. This can throw and
-      // we'll let that propagate out as it likely means there is a severe filesystem problem.
-      try {
-        delete()
-      } finally {
-        closed = false
+    try {
+      // Prefer to pick up where we left off.
+      if (fileSystem.exists(journalFile)) {
+        try {
+          readJournal()
+          processJournal()
+          initialized = true
+          return
+        } catch (journalIsCorrupt: IOException) {
+          Platform.get().log(
+            "DiskLruCache $directory is corrupt: ${journalIsCorrupt.message}, removing",
+            WARN,
+            journalIsCorrupt,
+          )
+        }
+
+        // The cache is corrupted, attempt to delete the contents of the directory. This can throw and
+        // we'll let that propagate out as it likely means there is a severe filesystem problem.
+        try {
+          delete()
+        } finally {
+          closed = false
+        }
       }
-    }
 
-    rebuildJournal()
+      rebuildJournal()
 
-    initialized = true
+      initialized = true
+    } finally {
+      // If anything failed, leave without a cache lock open
+      if (!initialized) {
+        cacheLock.close()
+      }
+    }
   }
 
   @Throws(IOException::class)
@@ -705,6 +717,8 @@ class DiskLruCache(
       return
     }
 
+    cacheLock.close()
+
     // Copying for concurrent iteration.
     for (entry in lruEntries.values.toTypedArray()) {
       if (entry.currentEditor != null) {

okhttp/src/main/kotlin/okhttp3/internal/cache2/Relay.kt

@@ -344,7 +344,10 @@ class Relay private constructor(
       val header = Buffer()
       fileOperator.read(0, header, FILE_HEADER_SIZE)
       val prefix = header.readByteString(PREFIX_CLEAN.size.toLong())
-      if (prefix != PREFIX_CLEAN) throw IOException("unreadable cache file")
+      if (prefix != PREFIX_CLEAN) {
+        randomAccessFile.close()
+        throw IOException("unreadable cache file")
+      }
       val upstreamSize = header.readLong()
       val metadataSize = header.readLong()