Repository: apollographql/apollo-kotlin
Pull request #5685 (2024-03-05 14:25:23)

Add `ApolloClient.retryOnError(Boolean)`
So that took a while but I'm quite happy with this solution. It's a lot more flexible that the previous WebSocket only and in the end it's just a few lines of code meaning someone wanting to add custom logic shouldn't need copy/pasting too many things.

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *apollo-android-docs* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 297950d86f4e5b21421b323b9a9d4bce26f24e9b |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/apollo-android-docs/deploys/65e83d2114dc8e0009410e1c |

Patch:
diff --git a/build-logic/build.gradle.kts b/build-logic/build.gradle.kts
index 2b1fdaed59..4bdcd432c2 100644
--- a/build-logic/build.gradle.kts
+++ b/build-logic/build.gradle.kts
@@ -43,6 +43,7 @@ dependencies {
   // builds but this seems just overkill for now
   runtimeOnly(libs.kotlin.allopen)
   runtimeOnly(libs.kotlinx.serialization.plugin)
+  runtimeOnly(libs.atomicfu.plugin)
 
   runtimeOnly(libs.sqldelight.plugin)
   runtimeOnly(libs.gradle.publish.plugin)
diff --git a/gradle/libraries.toml b/gradle/libraries.toml
index af1f9cc30c..39029bc19e 100644
--- a/gradle/libraries.toml
+++ b/gradle/libraries.toml
@@ -17,6 +17,7 @@ androidx-sqlite = "2.3.1"
 apollo = "4.0.0-beta.5-SNAPSHOT"
 # Used by the apollo-tooling project which uses a published version of Apollo
 apollo-published = "4.0.0-beta.3"
+atomicfu = "0.23.1"
 cache = "2.0.2"
 # See https://developer.android.com/jetpack/androidx/releases/compose-kotlin
 compose-compiler = "1.5.11-dev-k2.0.0-Beta4-21f5e479a96"
@@ -89,7 +90,9 @@ apollo-testingsupport = { group = "com.apollographql.apollo3", name = "apollo-te
 apollo-testingsupport-published = { group = "com.apollographql.apollo3", name = "apollo-testing-support", version.ref = "apollo-published" }
 apollo-tooling = { group = "com.apollographql.apollo3", name = "apollo-tooling", version.ref = "apollo" }
 assertj = { group = "org.assertj", name = "assertj-core", version = "3.24.2" }
-atomicfu = { group = "org.jetbrains.kotlinx", name = "atomicfu", version = "0.23.1" }
+atomicfu-library = { group = "org.jetbrains.kotlinx", name = "atomicfu", version.ref = "atomicfu" }
+atomicfu-plugin = { group = "org.jetbrains.kotlinx", name = "atomicfu-gradle-plugin", version.ref = "atomicfu" }
+
 benmanes-versions = { group = "com.github.ben-manes", name = "gradle-versions-plugin", version = "0.33.0" }
 clikt = { group = "com.github.ajalt.clikt", name = "clikt", version = "3.5.2" }
 compose-runtime = { group = "androidx.compose.runtime", name = "runtime", version.ref = "compose" }
diff --git a/libraries/apollo-api/api/apollo-api.api b/libraries/apollo-api/api/apollo-api.api
index 12c0a6add4..42c6dbb897 100644
--- a/libraries/apollo-api/api/apollo-api.api
+++ b/libraries/apollo-api/api/apollo-api.api
@@ -46,7 +46,7 @@ public final class com/apollographql/apollo3/api/ApolloOptionalAdapter : com/apo
 }
 
 public final class com/apollographql/apollo3/api/ApolloRequest : com/apollographql/apollo3/api/ExecutionOptions {
-	public synthetic fun <init> (Lcom/apollographql/apollo3/api/Operation;Ljava/util/UUID;Lcom/apollographql/apollo3/api/ExecutionContext;Lcom/apollographql/apollo3/api/http/HttpMethod;Ljava/util/List;Ljava/lang/Boolean;Ljava/lang/Boolean;Ljava/lang/Boolean;Ljava/lang/Boolean;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public synthetic fun <init> (Lcom/apollographql/apollo3/api/Operation;Ljava/util/UUID;Lcom/apollographql/apollo3/api/ExecutionContext;Lcom/apollographql/apollo3/api/http/HttpMethod;Ljava/util/List;Ljava/lang/Boolean;Ljava/lang/Boolean;Ljava/lang/Boolean;Ljava/lang/Boolean;Ljava/lang/Boolean;Lkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun getCanBeBatched ()Ljava/lang/Boolean;
 	public fun getEnableAutoPersistedQueries ()Ljava/lang/Boolean;
 	public fun getExecutionContext ()Lcom/apollographql/apollo3/api/ExecutionContext;
@@ -83,6 +83,7 @@ public final class com/apollographql/apollo3/api/ApolloRequest$Builder : com/apo
 	public fun httpMethod (Lcom/apollographql/apollo3/api/http/HttpMethod;)Lcom/apollographql/apollo3/api/ApolloRequest$Builder;
 	public synthetic fun httpMethod (Lcom/apollographql/apollo3/api/http/HttpMethod;)Ljava/lang/Object;
 	public final fun requestUuid (Ljava/util/UUID;)Lcom/apollographql/apollo3/api/ApolloRequest$Builder;
+	public synthetic fun retryOnError (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendApqExtensions (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/api/ApolloRequest$Builder;
 	public synthetic fun sendApqExtensions (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendDocument (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/api/ApolloRequest$Builder;
diff --git a/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ApolloRequest.kt b/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ApolloRequest.kt
index fce6264932..84250f483a 100644
--- a/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ApolloRequest.kt
+++ b/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ApolloRequest.kt
@@ -1,7 +1,6 @@
 package com.apollographql.apollo3.api
 
 import com.apollographql.apollo3.annotations.ApolloExperimental
-import com.apollographql.apollo3.annotations.ApolloInternal
 import com.apollographql.apollo3.api.http.HttpHeader
 import com.apollographql.apollo3.api.http.HttpMethod
 import com.benasher44.uuid.Uuid
@@ -21,13 +20,14 @@ private constructor(
     override val sendDocument: Boolean?,
     override val enableAutoPersistedQueries: Boolean?,
     override val canBeBatched: Boolean?,
+    @ApolloExperimental
+    override val retryOnError: Boolean?,
 ) : ExecutionOptions {
 
   fun newBuilder(): Builder<D> = newBuilder(operation)
 
   @ApolloExperimental
-  fun <E: Operation.Data> newBuilder(operation: Operation<E>): Builder<E> {
-    @Suppress("DEPRECATION")
+  fun <E : Operation.Data> newBuilder(operation: Operation<E>): Builder<E> {
     return Builder(operation)
         .requestUuid(requestUuid)
         .executionContext(executionContext)
@@ -37,6 +37,7 @@ private constructor(
         .sendDocument(sendDocument)
         .enableAutoPersistedQueries(enableAutoPersistedQueries)
         .canBeBatched(canBeBatched)
+        .retryOnError(retryOnError)
   }
 
   class Builder<D : Operation.Data>(
@@ -48,14 +49,24 @@ private constructor(
 
     override var httpMethod: HttpMethod? = null
       private set
+    override var httpHeaders: List<HttpHeader>? = null
+      private set
+    override var enableAutoPersistedQueries: Boolean? = null
+      private set
+    override var sendApqExtensions: Boolean? = null
+      private set
+    override var sendDocument: Boolean? = null
+      private set
+    override var canBeBatched: Boolean? = null
+      private set
+    @ApolloExperimental
+    override var retryOnError: Boolean? = null
+      private set
 
     override fun httpMethod(httpMethod: HttpMethod?): Builder<D> = apply {
       this.httpMethod = httpMethod
     }
 
-    override var httpHeaders: List<HttpHeader>? = null
-      private set
-
     override fun httpHeaders(httpHeaders: List<HttpHeader>?): Builder<D> = apply {
       this.httpHeaders = httpHeaders
     }
@@ -64,34 +75,27 @@ private constructor(
       this.httpHeaders = (this.httpHeaders ?: emptyList()) + HttpHeader(name, value)
     }
 
-    override var sendApqExtensions: Boolean? = null
-      private set
-
     override fun sendApqExtensions(sendApqExtensions: Boolean?): Builder<D> = apply {
       this.sendApqExtensions = sendApqExtensions
     }
 
-    override var sendDocument: Boolean? = null
-      private set
-
     override fun sendDocument(sendDocument: Boolean?): Builder<D> = apply {
       this.sendDocument = sendDocument
     }
 
-    override var enableAutoPersistedQueries: Boolean? = null
-      private set
-
     override fun enableAutoPersistedQueries(enableAutoPersistedQueries: Boolean?): Builder<D> = apply {
       this.enableAutoPersistedQueries = enableAutoPersistedQueries
     }
 
-    override var canBeBatched: Boolean? = null
-      private set
-
     override fun canBeBatched(canBeBatched: Boolean?): Builder<D> = apply {
       this.canBeBatched = canBeBatched
     }
 
+    @ApolloExperimental
+    override fun retryOnError(retryOnError: Boolean?): Builder<D> = apply {
+      this.retryOnError = retryOnError
+    }
+
     fun requestUuid(requestUuid: Uuid) = apply {
       this.requestUuid = requestUuid
     }
@@ -105,7 +109,6 @@ private constructor(
     }
 
     fun build(): ApolloRequest<D> {
-      @Suppress("DEPRECATION")
       return ApolloRequest(
           operation = operation,
           requestUuid = requestUuid,
@@ -116,6 +119,7 @@ private constructor(
           sendDocument = sendDocument,
           enableAutoPersistedQueries = enableAutoPersistedQueries,
           canBeBatched = canBeBatched,
+          retryOnError = retryOnError
       )
     }
   }
diff --git a/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ExecutionOptions.kt b/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ExecutionOptions.kt
index 93aabfee8f..ca739b513c 100644
--- a/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ExecutionOptions.kt
+++ b/libraries/apollo-api/src/commonMain/kotlin/com/apollographql/apollo3/api/ExecutionOptions.kt
@@ -1,5 +1,6 @@
 package com.apollographql.apollo3.api
 
+import com.apollographql.apollo3.annotations.ApolloExperimental
 import com.apollographql.apollo3.api.http.HttpHeader
 import com.apollographql.apollo3.api.http.HttpMethod
 
@@ -45,6 +46,12 @@ interface ExecutionOptions {
    */
   val canBeBatched: Boolean?
 
+  /**
+   * Whether to enable retrying the request on network errors
+   */
+  @ApolloExperimental
+  val retryOnError: Boolean?
+
   companion object {
     /**
      * Used by [com.apollographql.apollo3.network.http.BatchingHttpInterceptor]
@@ -82,4 +89,7 @@ interface MutableExecutionOptions<T> : ExecutionOptions {
   fun enableAutoPersistedQueries(enableAutoPersistedQueries: Boolean?): T
 
   fun canBeBatched(canBeBatched: Boolean?): T
+
+  @ApolloExperimental
+  fun retryOnError(retryOnError: Boolean?): T
 }
diff --git a/libraries/apollo-execution-incubating/build.gradle.kts b/libraries/apollo-execution-incubating/build.gradle.kts
index a687f771d5..375658fa3e 100644
--- a/libraries/apollo-execution-incubating/build.gradle.kts
+++ b/libraries/apollo-execution-incubating/build.gradle.kts
@@ -13,7 +13,7 @@ kotlin {
       dependencies {
         api(project(":apollo-ast"))
         api(project(":apollo-api"))
-        implementation(libs.atomicfu)
+        implementation(libs.atomicfu.library)
         implementation(libs.kotlinx.coroutines)
       }
     }
diff --git a/libraries/apollo-execution-incubating/src/commonMain/kotlin/com/apollographql/apollo3/execution/websocket/ApolloWebSocketHandler.kt b/libraries/apollo-execution-incubating/src/commonMain/kotlin/com/apollographql/apollo3/execution/websocket/ApolloWebSocketHandler.kt
index ee7f5f40c0..f670595dab 100644
--- a/libraries/apollo-execution-incubating/src/commonMain/kotlin/com/apollographql/apollo3/execution/websocket/ApolloWebSocketHandler.kt
+++ b/libraries/apollo-execution-incubating/src/commonMain/kotlin/com/apollographql/apollo3/execution/websocket/ApolloWebSocketHandler.kt
@@ -39,6 +39,15 @@ class ConnectionInitError(val payload: Optional<Any?> = Optional.absent()): Conn
 
 typealias ConnectionInitHandler = suspend (Any?) -> ConnectionInitResult
 
+private class CurrentSubscription(val id: String) : ExecutionContext.Element {
+
+  override val key: ExecutionContext.Key<CurrentSubscription> = Key
+
+  companion object Key : ExecutionContext.Key<CurrentSubscription>
+}
+
+fun ExecutionContext.subscriptionId(): String = get(CurrentSubscription)?.id ?: error("Apollo: not executing a subscription")
+
 class ApolloWebSocketHandler(
     private val executableSchema: ExecutableSchema,
     private val scope: CoroutineScope,
@@ -82,7 +91,7 @@ class ApolloWebSocketHandler(
           return
         }
 
-        val flow = executableSchema.executeSubscription(clientMessage.request, executionContext)
+        val flow = executableSchema.executeSubscription(clientMessage.request, executionContext + CurrentSubscription(clientMessage.id))
 
         val job = scope.launch {
           flow.collect {
diff --git a/libraries/apollo-ksp-incubating/src/main/kotlin/com/apollographql/apollo3/ksp/ksp-validation.kt b/libraries/apollo-ksp-incubating/src/main/kotlin/com/apollographql/apollo3/ksp/ksp-validation.kt
index f93679216b..a017933e13 100644
--- a/libraries/apollo-ksp-incubating/src/main/kotlin/com/apollographql/apollo3/ksp/ksp-validation.kt
+++ b/libraries/apollo-ksp-incubating/src/main/kotlin/com/apollographql/apollo3/ksp/ksp-validation.kt
@@ -118,7 +118,7 @@ internal class ValidationScope(
             val possibleTypes: Set<String>? = possibleTypes[className]
             if (possibleTypes == null) {
               if (typeDefinition is GQLObjectTypeDefinition) {
-                error("Expected a Kotlin object for type '${typeDefinition.name}' but none found. Did you forget a @ApolloObject? at ${ksTypeReference.location}")
+                error("Expected a Kotlin object for type '${typeDefinition.name}' but none found. Did you forget a @GraphQLObject? at ${ksTypeReference.location}")
               } else {
                 error("Expected that Kotlin '$className' is an object or interface for type '${typeDefinition.name}' but none found.")
               }
diff --git a/libraries/apollo-mockserver/build.gradle.kts b/libraries/apollo-mockserver/build.gradle.kts
index cbeabdf176..ff04beaf97 100644
--- a/libraries/apollo-mockserver/build.gradle.kts
+++ b/libraries/apollo-mockserver/build.gradle.kts
@@ -13,7 +13,7 @@ kotlin {
       dependencies {
         api(project(":apollo-annotations"))
         api(libs.okio)
-        implementation(libs.atomicfu.get().toString()) {
+        implementation(libs.atomicfu.library.get().toString()) {
           because("We need locks for native (we don't use the gradle plugin rewrite)")
         }
         api(libs.kotlinx.coroutines)
diff --git a/libraries/apollo-normalized-cache-api-incubating/build.gradle.kts b/libraries/apollo-normalized-cache-api-incubating/build.gradle.kts
index 7de9e274ac..c783934a39 100644
--- a/libraries/apollo-normalized-cache-api-incubating/build.gradle.kts
+++ b/libraries/apollo-normalized-cache-api-incubating/build.gradle.kts
@@ -1,9 +1,10 @@
 plugins {
   id("org.jetbrains.kotlin.multiplatform")
+  id("kotlinx-atomicfu")
 }
 
 apolloLibrary(
-  javaModuleName = "com.apollographql.apollo3.cache.normalized.api",
+    javaModuleName = "com.apollographql.apollo3.cache.normalized.api",
     withLinux = false,
 )
 
@@ -15,9 +16,13 @@ kotlin {
         api(project(":apollo-mpp-utils"))
         implementation(libs.okio)
         api(libs.uuid)
-        implementation(libs.atomicfu.get().toString()) {
-          because("Use of ReentrantLock for Apple (we don't use the gradle plugin rewrite)")
-        }
+        implementation(libs.atomicfu.library)
+      }
+    }
+
+    findByName("commonTest")?.apply {
+      dependencies {
+        implementation(project(":apollo-testing-support"))
       }
     }
   }
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/appleMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.apple.kt b/libraries/apollo-normalized-cache-api-incubating/src/appleMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.apple.kt
new file mode 100644
index 0000000000..1e035fac8f
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/appleMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.apple.kt
@@ -0,0 +1,3 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+internal actual fun <K, V> ConcurrentMap(): MutableMap<K, V> = CommonConcurrentMap()
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/MemoryCache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/MemoryCache.kt
index 2edfc8487b..35e7250c41 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/MemoryCache.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/MemoryCache.kt
@@ -1,10 +1,8 @@
 package com.apollographql.apollo3.cache.normalized.api
 
-import com.apollographql.apollo3.annotations.ApolloExperimental
 import com.apollographql.apollo3.cache.normalized.api.internal.Lock
 import com.apollographql.apollo3.cache.normalized.api.internal.LruCache
-import com.apollographql.apollo3.mpp.currentTimeMillis
-import okio.internal.commonAsUtf8ToByteArray
+import com.apollographql.apollo3.cache.normalized.api.internal.patternToRegex
 import kotlin.jvm.JvmOverloads
 import kotlin.reflect.KClass
 
@@ -18,30 +16,37 @@ import kotlin.reflect.KClass
  * (there is no any sort of GC that runs in the background).
  */
 class MemoryCache(
+    private val nextCache: NormalizedCache? = null,
     private val maxSizeBytes: Int = Int.MAX_VALUE,
     private val expireAfterMillis: Long = -1,
-) : NormalizedCache() {
-  private val lock = Lock()
+) : NormalizedCache {
+  // A lock is only needed if there is a nextCache
+  private val lock = nextCache?.let { Lock() }
 
-  private val lruCache = LruCache<String, CacheEntry>(maxSize = maxSizeBytes) { key, cacheEntry ->
-    key.commonAsUtf8ToByteArray().size + (cacheEntry?.sizeInBytes ?: 0)
+  private fun <T> lockWrite(block: () -> T): T {
+    return lock?.write { block() } ?: block()
+  }
+
+  private fun <T> lockRead(block: () -> T): T {
+    return lock?.read { block() } ?: block()
+  }
+
+  private val lruCache = LruCache<String, Record>(maxSize = maxSizeBytes, expireAfterMillis = expireAfterMillis) { key, record ->
+    key.length + record.sizeInBytes
   }
 
   val size: Int
-    get() = lock.read { lruCache.size() }
+    get() = lockRead { lruCache.weight() }
 
-  override fun loadRecord(key: String, cacheHeaders: CacheHeaders): Record? = lock.write {
-    val cacheEntry = lruCache[key]?.also { cacheEntry ->
-      if (cacheEntry.isExpired || cacheHeaders.hasHeader(ApolloCacheHeaders.EVICT_AFTER_READ)) {
+  override fun loadRecord(key: String, cacheHeaders: CacheHeaders): Record? = lockRead {
+    val record = lruCache[key]?.also {
+      if (cacheHeaders.hasHeader(ApolloCacheHeaders.EVICT_AFTER_READ)) {
         lruCache.remove(key)
       }
     }
 
-    cacheEntry?.takeUnless { it.isExpired }?.record ?: nextCache?.loadRecord(key, cacheHeaders)?.also { nextCachedRecord ->
-      lruCache[key] = CacheEntry(
-          record = nextCachedRecord,
-          expireAfterMillis = expireAfterMillis
-      )
+    record ?: nextCache?.loadRecord(key, cacheHeaders)?.also { nextCachedRecord ->
+      lruCache[key] = nextCachedRecord
     }
   }
 
@@ -50,109 +55,78 @@ class MemoryCache(
   }
 
   override fun clearAll() {
-    lock.write {
+    lockWrite {
       lruCache.clear()
       nextCache?.clearAll()
     }
   }
 
   override fun remove(cacheKey: CacheKey, cascade: Boolean): Boolean {
-    return lock.write {
-      val cacheEntry = lruCache.remove(cacheKey.key)
+    return lockWrite {
+      val record = lruCache.remove(cacheKey.key)
 
-      if (cascade && cacheEntry != null) {
-        for (cacheReference in cacheEntry.record.referencedFields()) {
+      if (cascade && record != null) {
+        for (cacheReference in record.referencedFields()) {
           remove(CacheKey(cacheReference.key), true)
         }
       }
 
       val chainRemoved = nextCache?.remove(cacheKey, cascade) ?: false
-      cacheEntry != null || chainRemoved
+      record != null || chainRemoved
     }
   }
 
-  override fun remove(pattern: String): Int = lock.write {
+  override fun remove(pattern: String): Int {
     val regex = patternToRegex(pattern)
-    var total = 0
-    val keys = HashSet(lruCache.keys()) // local copy to avoid concurrent modification
-    keys.forEach {
-      if (regex.matches(it)) {
-        lruCache.remove(it)
-        total++
+    return lockWrite {
+      var total = 0
+      val keys = HashSet(lruCache.asMap().keys) // local copy to avoid concurrent modification
+      keys.forEach {
+        if (regex.matches(it)) {
+          lruCache.remove(it)
+          total++
+        }
       }
-    }
-
-    val chainRemoved = nextCache?.remove(pattern) ?: 0
-    total + chainRemoved
-  }
-
-  override fun merge(record: Record, cacheHeaders: CacheHeaders): Set<String> {
-    return merge(record, cacheHeaders, DefaultRecordMerger)
-  }
 
-  override fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders): Set<String> {
-    return merge(records, cacheHeaders, DefaultRecordMerger)
+      val chainRemoved = nextCache?.remove(pattern) ?: 0
+      total + chainRemoved
+    }
   }
 
-  @ApolloExperimental
   override fun merge(record: Record, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> {
     if (cacheHeaders.hasHeader(ApolloCacheHeaders.DO_NOT_STORE)) {
       return emptySet()
     }
-    return lock.write {
+    return lockWrite {
       val oldRecord = loadRecord(record.key, cacheHeaders)
       val changedKeys = if (oldRecord == null) {
-        lruCache[record.key] = CacheEntry(
-            record = record,
-            expireAfterMillis = expireAfterMillis
-        )
+        lruCache[record.key] = record
         record.fieldKeys()
       } else {
         val (mergedRecord, changedKeys) = recordMerger.merge(existing = oldRecord, incoming = record, newDate = null)
-        lruCache[record.key] = CacheEntry(
-            record = mergedRecord,
-            expireAfterMillis = expireAfterMillis
-        )
+        lruCache[record.key] = mergedRecord
         changedKeys
       }
       changedKeys + nextCache?.merge(record, cacheHeaders, recordMerger).orEmpty()
     }
   }
 
-  @ApolloExperimental
   override fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> {
     if (cacheHeaders.hasHeader(ApolloCacheHeaders.DO_NOT_STORE)) {
       return emptySet()
     }
-    return lock.write { records.flatMap { record -> merge(record, cacheHeaders, recordMerger) } }.toSet()
+    return lockWrite { records.flatMap { record -> merge(record, cacheHeaders, recordMerger) } }.toSet()
   }
 
   override fun dump(): Map<KClass<*>, Map<String, Record>> {
-    return lock.read {
-      mapOf(this::class to lruCache.dump().mapValues { (_, entry) -> entry.record }) + nextCache?.dump().orEmpty()
+    return lockRead {
+      mapOf(this::class to lruCache.asMap().mapValues { (_, record) -> record }) +
+          nextCache?.dump().orEmpty()
     }
   }
 
   internal fun clearCurrentCache() {
-    lock.write { lruCache.clear() }
-  }
-
-  private class CacheEntry(
-      val record: Record,
-      val expireAfterMillis: Long,
-  ) {
-    val cachedAtMillis: Long = currentTimeMillis()
-
-    val sizeInBytes: Int = record.sizeInBytes + 8
-
-    val isExpired: Boolean
-      get() {
-        return if (expireAfterMillis < 0) {
-          false
-        } else {
-          currentTimeMillis() - cachedAtMillis >= expireAfterMillis
-        }
-      }
+    lruCache.clear()
   }
 }
 
@@ -161,8 +135,16 @@ class MemoryCacheFactory @JvmOverloads constructor(
     private val expireAfterMillis: Long = -1,
 ) : NormalizedCacheFactory() {
 
+  private var nextCacheFactory: NormalizedCacheFactory? = null
+
+  fun chain(factory: NormalizedCacheFactory): MemoryCacheFactory {
+    nextCacheFactory = factory
+    return this
+  }
+
   override fun create(): MemoryCache {
     return MemoryCache(
+        nextCache = nextCacheFactory?.create(),
         maxSizeBytes = maxSizeBytes,
         expireAfterMillis = expireAfterMillis,
     )
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCache.kt
index acb2440144..692e2f9a60 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCache.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCache.kt
@@ -1,6 +1,5 @@
 package com.apollographql.apollo3.cache.normalized.api
 
-import com.apollographql.apollo3.annotations.ApolloExperimental
 import kotlin.jvm.JvmStatic
 import kotlin.jvm.JvmSuppressWildcards
 import kotlin.reflect.KClass
@@ -14,39 +13,22 @@ import kotlin.reflect.KClass
  * If a [NormalizedCache] cannot return all the records needed to read a response, it will be considered a cache
  * miss.
  *
- * A [NormalizedCache] is recommended to implement support for [CacheHeaders] specified in [ ].
+ * A [NormalizedCache] is recommended to implement support for [CacheHeaders] specified in the `cacheHeaders` of [merge] .
  *
  * A [NormalizedCache] can choose to store records in any manner.
  */
-abstract class NormalizedCache : ReadOnlyNormalizedCache {
-  var nextCache: NormalizedCache? = null
-    private set
-
-  /**
-   * @param record       The [Record] to merge.
-   * @param cacheHeaders The [CacheHeaders] associated with the request which generated this record.
-   * @return A set of record field keys that have changed. This set is returned by [Record.mergeWith].
-   */
-  abstract fun merge(record: Record, cacheHeaders: CacheHeaders): Set<String>
-
-  /**
-   * Calls through to [NormalizedCache.merge]. Implementations should override this method
-   * if the underlying storage technology can offer an optimized manner to store multiple records.
-   *
-   * @param records The collection of Records to merge.
-   * @param cacheHeaders The [CacheHeaders] associated with the request which generated this record.
-   * @return A set of record field keys that have changed. This set is returned by [Record.mergeWith].
-   */
-  abstract fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders): Set<String>
-
+interface NormalizedCache : ReadOnlyNormalizedCache {
   /**
    * @param record       The [Record] to merge.
    * @param cacheHeaders The [CacheHeaders] associated with the request which generated this record.
    * @param recordMerger The [RecordMerger] to use when merging the record.
    * @return A set of record field keys that have changed. This set is returned by [RecordMerger.merge].
    */
-  @ApolloExperimental
-  open fun merge(record: Record, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> = merge(record, cacheHeaders)
+  fun merge(
+      record: Record,
+      cacheHeaders: CacheHeaders,
+      recordMerger: RecordMerger,
+  ): Set<String>
 
   /**
    * Calls through to [NormalizedCache.merge]. Implementations should override this method
@@ -57,27 +39,26 @@ abstract class NormalizedCache : ReadOnlyNormalizedCache {
    * @param recordMerger The [RecordMerger] to use when merging the records.
    * @return A set of record field keys that have changed. This set is returned by [RecordMerger.merge].
    */
-  @ApolloExperimental
-  open fun merge(
+  fun merge(
       records: Collection<Record>,
       cacheHeaders: CacheHeaders,
       recordMerger: RecordMerger,
-  ): Set<String> = merge(records, cacheHeaders)
+  ): Set<String>
 
 
   /**
    * Clears all records from the cache.
    */
-  abstract fun clearAll()
+  fun clearAll()
 
   /**
    * Remove a record and potentially its referenced records from this cache and all chained caches
    *
    * @param cacheKey of record to be removed
    * @param cascade remove referenced records if true
-   * @return `true` if a record with such key was successfully removed, `false` otherwise
+   * @return `true` if at least one record was successfully removed, `false` otherwise
    */
-  abstract fun remove(cacheKey: CacheKey, cascade: Boolean): Boolean
+  fun remove(cacheKey: CacheKey, cascade: Boolean): Boolean
 
   /**
    * Remove records whose key matches a given pattern from this cache and all chained caches
@@ -91,18 +72,10 @@ abstract class NormalizedCache : ReadOnlyNormalizedCache {
    *
    * @return the number of records deleted accross all caches
    */
-  abstract fun remove(pattern: String): Int
+  fun remove(pattern: String): Int
 
-  fun chain(cache: NormalizedCache) = apply {
-    var leafCache = this
-    while (leafCache.nextCache != null) {
-      leafCache = leafCache.nextCache!!
-    }
-    leafCache.nextCache = cache
-  }
 
   companion object {
-
     @JvmStatic
     fun prettifyDump(dump: Map<@JvmSuppressWildcards KClass<*>, Map<String, Record>>): String = dump.prettifyDump()
 
@@ -166,43 +139,6 @@ abstract class NormalizedCache : ReadOnlyNormalizedCache {
     }
 
     private fun StringBuilder.indent(level: Int) = append("  ".repeat(level))
-
-    /**
-     * A tentative to approximate the Sqlite LIKE operator with Regexes
-     */
-    fun patternToRegex(pattern: String): Regex {
-      val regex = buildString {
-        var pendingEscape = false
-        for (i in pattern.indices) {
-          val cur = pattern[i]
-          when {
-            pendingEscape -> {
-              when {
-                cur == '\\' -> append("\\\\") // an escaped backslash is also an escape backslash in a regex
-                cur == '%' -> append("%")
-                cur == '_' -> append("_")
-                else -> error("Invalid escape in pattern: $pattern")
-              }
-            }
-
-            cur == '\\' -> pendingEscape = true
-            cur == '%' -> append(".*")
-            cur == '_' -> append(".")
-            else -> {
-              if (specialChars.contains(cur)) {
-                // this needs to be escaped in the regex
-                append("\\")
-              }
-              append(cur)
-            }
-          }
-        }
-      }
-
-      return Regex(regex, option = RegexOption.IGNORE_CASE)
-    }
-
-    private val specialChars = "()^$.*?+{}"
   }
 }
 
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCacheFactory.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCacheFactory.kt
index a1d8b1d2e3..0ce78f27e5 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCacheFactory.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/NormalizedCacheFactory.kt
@@ -6,28 +6,9 @@ package com.apollographql.apollo3.cache.normalized.api
  */
 abstract class NormalizedCacheFactory {
 
-  private var nextFactory: NormalizedCacheFactory? = null
-
   /**
    * ApolloClient.Builder#addCustomScalarAdapter(ScalarType, CustomScalarAdapter).
    * @return An implementation of [NormalizedCache].
    */
   abstract fun create(): NormalizedCache
-
-  fun createChain(): NormalizedCache {
-    val nextFactory = nextFactory
-    return if (nextFactory != null) {
-      create().chain(nextFactory.createChain())
-    } else {
-      create()
-    }
-  }
-
-  fun chain(factory: NormalizedCacheFactory) = apply {
-    var leafFactory: NormalizedCacheFactory = this
-    while (leafFactory.nextFactory != null) {
-      leafFactory = leafFactory.nextFactory!!
-    }
-    leafFactory.nextFactory = factory
-  }
 }
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/CommonConcurrentMap.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/CommonConcurrentMap.kt
new file mode 100644
index 0000000000..8b08d5faf0
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/CommonConcurrentMap.kt
@@ -0,0 +1,54 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+/**
+ * A naive concurrent map implementation that uses a read-write lock to synchronize access to the underlying map.
+ * This is used for native and JS/Wasm while the JVM uses the built-in `ConcurrentHashMap`.
+ */
+internal class CommonConcurrentMap<K, V> : MutableMap<K, V> {
+  private val lock = Lock()
+  private val map = mutableMapOf<K, V>()
+
+  override val entries: MutableSet<MutableMap.MutableEntry<K, V>>
+    get() = lock.read { map.entries }
+
+  override val keys: MutableSet<K>
+    get() = lock.read { map.keys }
+
+  override val size: Int
+    get() = lock.read { map.size }
+
+  override val values: MutableCollection<V>
+    get() = lock.read { map.values }
+
+  override fun clear() {
+    lock.write { map.clear() }
+  }
+
+  override fun isEmpty(): Boolean {
+    return lock.read { map.isEmpty() }
+  }
+
+  override fun remove(key: K): V? {
+    return lock.write { map.remove(key) }
+  }
+
+  override fun putAll(from: Map<out K, V>) {
+    lock.write { map.putAll(from) }
+  }
+
+  override fun put(key: K, value: V): V? {
+    return lock.write { map.put(key, value) }
+  }
+
+  override fun get(key: K): V? {
+    return lock.read { map[key] }
+  }
+
+  override fun containsValue(value: V): Boolean {
+    return lock.read { map.containsValue(value) }
+  }
+
+  override fun containsKey(key: K): Boolean {
+    return lock.read { map.containsKey(key) }
+  }
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.kt
new file mode 100644
index 0000000000..bf432f0a03
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.kt
@@ -0,0 +1,3 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+internal expect fun <K, V> ConcurrentMap(): MutableMap<K, V>
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/LruCache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/LruCache.kt
index c69ac361da..28d0a0def5 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/LruCache.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/LruCache.kt
@@ -1,160 +1,49 @@
 package com.apollographql.apollo3.cache.normalized.api.internal
 
-internal typealias Weigher<Key, Value> = (Key, Value?) -> Int
+import com.apollographql.apollo3.cache.normalized.api.internal.store.CacheBuilder
+import kotlin.time.Duration.Companion.milliseconds
 
-/**
- * Multiplatform LRU cache implementation.
- *
- * Implementation is based on usage of [LinkedHashMap] as a container for the cache and custom
- * double linked queue to track LRU property.
- *
- * [maxSize] - maximum size of the cache, can be anything bytes, number of entries etc. By default is number o entries.
- * [weigher] - to be called to calculate the estimated size (weight) of the cache entry defined by its [Key] and [Value].
- *             By default it returns 1.
- *
- * Cache trim performed only on new entry insertion.
- */
-internal class LruCache<Key, Value>(
-    private val maxSize: Int,
-    private val weigher: Weigher<Key, Value> = { _, _ -> 1 },
+internal class LruCache<Key : Any, Value : Any>(
+    maxSize: Int,
+    expireAfterMillis: Long,
+    private val weigher: Weigher<Key, Value>,
 ) {
-  private val cache = LinkedHashMap<Key, Node<Key, Value>>(0, 0.75f)
-  private var headNode: Node<Key, Value>? = null
-  private var tailNode: Node<Key, Value>? = null
-  private var size: Int = 0
+  private val cache = CacheBuilder<Key, Value>()
+      .apply {
+        if (maxSize != Int.MAX_VALUE) {
+          weigher(maxSize.toLong(), this@LruCache.weigher)
+        }
+        if (expireAfterMillis >= 0) {
+          expireAfterAccess(expireAfterMillis.milliseconds)
+        }
+      }
+      .build()
 
   operator fun get(key: Key): Value? {
-    val node = cache[key]
-    if (node != null) {
-      moveNodeToHead(node)
-    }
-    return node?.value
+    return cache.getIfPresent(key)
   }
 
   operator fun set(key: Key, value: Value) {
-    val node = cache[key]
-    if (node == null) {
-      cache[key] = addNode(key, value)
-    } else {
-      node.value = value
-      moveNodeToHead(node)
-    }
-
-    trim()
+    cache.put(key, value)
   }
 
   fun remove(key: Key): Value? {
-    return removeUnsafe(key)
-  }
-
-  fun keys() = cache.keys
-
-  private fun removeUnsafe(key: Key): Value? {
-    val nodeToRemove = cache.remove(key)
-    val value = nodeToRemove?.value
-    if (nodeToRemove != null) {
-      unlinkNode(nodeToRemove)
-    }
+    val value = cache.getIfPresent(key)
+    cache.invalidate(key)
     return value
   }
 
-  fun remove(keys: Collection<Key>) {
-    keys.forEach { key -> removeUnsafe(key) }
-  }
-
   fun clear() {
-    cache.clear()
-    headNode = null
-    tailNode = null
-    size = 0
-  }
-
-  fun size(): Int {
-    return size
-  }
-
-  fun dump(): Map<Key, Value> {
-    return cache.mapValues { (_, value) ->
-      @Suppress("UNCHECKED_CAST")
-      value.value as Value
-    }
+    cache.invalidateAll()
   }
 
-  private fun trim() {
-    var nodeToRemove = tailNode
-    while (nodeToRemove != null && size > maxSize) {
-      cache.remove(nodeToRemove.key)
-      unlinkNode(nodeToRemove)
-      nodeToRemove = tailNode
-    }
+  fun weight(): Int {
+    return cache.getAllPresent().entries.sumOf { weigher(it.key, it.value) }
   }
 
-  private fun addNode(key: Key, value: Value?): Node<Key, Value> {
-    val node = Node(
-        key = key,
-        value = value,
-        next = headNode,
-        prev = null,
-    )
-
-    headNode = node
-
-    if (node.next == null) {
-      tailNode = headNode
-    } else {
-      node.next?.prev = headNode
-    }
-
-    size += weigher(key, value)
-
-    return node
+  fun asMap(): Map<Key, Value> {
+    return cache.getAllPresent()
   }
-
-  private fun moveNodeToHead(node: Node<Key, Value>) {
-    if (node.prev == null) {
-      return
-    }
-
-    node.prev?.next = node.next
-
-    if (node.next == null) {
-      tailNode = node.prev
-    } else {
-      node.next?.prev = node.prev
-    }
-
-    node.next = headNode
-    node.prev = null
-
-    headNode?.prev = node
-    headNode = node
-  }
-
-  private fun unlinkNode(node: Node<Key, Value>) {
-    if (node.prev == null) {
-      this.headNode = node.next
-    } else {
-      node.prev?.next = node.next
-    }
-
-    if (node.next == null) {
-      this.tailNode = node.prev
-    } else {
-      node.next?.prev = node.prev
-    }
-
-    size -= weigher(node.key!!, node.value)
-
-    node.key = null
-    node.value = null
-    node.next = null
-    node.prev = null
-  }
-
-  private class Node<Key, Value>(
-      var key: Key?,
-      var value: Value?,
-      var next: Node<Key, Value>?,
-      var prev: Node<Key, Value>?,
-  )
 }
+
+internal typealias Weigher<Key, Value> = (Key, Value) -> Int
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticCache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticNormalizedCache.kt
similarity index 56%
rename from libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticCache.kt
rename to libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticNormalizedCache.kt
index 2c2948d59e..c683f2d3f7 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticCache.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/OptimisticNormalizedCache.kt
@@ -1,6 +1,5 @@
 package com.apollographql.apollo3.cache.normalized.api.internal
 
-import com.apollographql.apollo3.annotations.ApolloExperimental
 import com.apollographql.apollo3.annotations.ApolloInternal
 import com.apollographql.apollo3.cache.normalized.api.CacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheKey
@@ -13,131 +12,97 @@ import kotlin.math.max
 import kotlin.reflect.KClass
 
 @ApolloInternal
-class OptimisticCache : NormalizedCache() {
-  private val lock = Lock()
-  private val recordJournals = mutableMapOf<String, RecordJournal>()
+class OptimisticNormalizedCache(private val wrapped: NormalizedCache) : NormalizedCache {
+  private val recordJournals = ConcurrentMap<String, RecordJournal>()
 
   override fun loadRecord(key: String, cacheHeaders: CacheHeaders): Record? {
-    return lock.read {
-      try {
-        val nonOptimisticRecord = nextCache?.loadRecord(key, cacheHeaders)
-        nonOptimisticRecord.mergeJournalRecord(key)
-      } catch (ignore: Exception) {
-        null
-      }
-    }
+    val nonOptimisticRecord = wrapped.loadRecord(key, cacheHeaders)
+    return nonOptimisticRecord.mergeJournalRecord(key)
   }
 
   override fun loadRecords(keys: Collection<String>, cacheHeaders: CacheHeaders): Collection<Record> {
-    return lock.read {
-      val nonOptimisticRecords = nextCache?.loadRecords(keys, cacheHeaders)?.associateBy { it.key } ?: emptyMap()
-      keys.mapNotNull { key ->
-        nonOptimisticRecords[key].mergeJournalRecord(key)
-      }
+    val nonOptimisticRecords = wrapped.loadRecords(keys, cacheHeaders).associateBy { it.key }
+    return keys.mapNotNull { key ->
+      nonOptimisticRecords[key].mergeJournalRecord(key)
     }
   }
 
-  override fun merge(record: Record, cacheHeaders: CacheHeaders): Set<String> {
-    return lock.write { nextCache?.merge(record, cacheHeaders) } ?: emptySet()
-  }
-
-  override fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders): Set<String> {
-    return lock.write { nextCache?.merge(records, cacheHeaders) } ?: emptySet()
-  }
-
-  @ApolloExperimental
   override fun merge(record: Record, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> {
-    return lock.write { nextCache?.merge(record, cacheHeaders, recordMerger) } ?: emptySet()
+    return wrapped.merge(record, cacheHeaders, recordMerger)
   }
 
-  @ApolloExperimental
   override fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> {
-    return lock.write { nextCache?.merge(records, cacheHeaders, recordMerger) } ?: emptySet()
+    return wrapped.merge(records, cacheHeaders, recordMerger)
   }
 
   override fun clearAll() {
-    lock.write {
-      recordJournals.clear()
-      nextCache?.clearAll()
-    }
+    wrapped.clearAll()
+    recordJournals.clear()
   }
 
   override fun remove(cacheKey: CacheKey, cascade: Boolean): Boolean {
-    return lock.write {
-      var result: Boolean = nextCache?.remove(cacheKey, cascade) ?: false
-
-      val recordJournal = recordJournals[cacheKey.key]
-      if (recordJournal != null) {
-        recordJournals.remove(cacheKey.key)
-        result = true
-        if (cascade) {
-          for (cacheReference in recordJournal.current.referencedFields()) {
-            result = result && remove(CacheKey(cacheReference.key), true)
-          }
+    var removed = wrapped.remove(cacheKey, cascade)
+
+    val recordJournal = recordJournals[cacheKey.key]
+    if (recordJournal != null) {
+      recordJournals.remove(cacheKey.key)
+      removed = true
+      if (cascade) {
+        for (cacheReference in recordJournal.current.referencedFields()) {
+          remove(CacheKey(cacheReference.key), true)
         }
       }
-      result
     }
+    return removed
   }
 
   override fun remove(pattern: String): Int {
+    var removed = wrapped.remove(pattern)
+
     val regex = patternToRegex(pattern)
-    var total = 0
-    return lock.write {
-      val iterator = recordJournals.iterator()
-      while (iterator.hasNext()) {
-        val entry = iterator.next()
-        if (regex.matches(entry.key)) {
-          iterator.remove()
-          total++
-        }
+    val keys = HashSet(recordJournals.keys) // local copy to avoid concurrent modification
+    keys.forEach { key ->
+      if (regex.matches(key)) {
+        recordJournals.remove(key)
+        removed++
       }
-
-      val chainRemoved = nextCache?.remove(pattern) ?: 0
-      total + chainRemoved
     }
+
+    return removed
   }
 
   fun addOptimisticUpdates(recordSet: Collection<Record>): Set<String> {
-    return lock.write {
-      recordSet.flatMap {
-        addOptimisticUpdate(it)
-      }
+    return recordSet.flatMap {
+      addOptimisticUpdate(it)
     }.toSet()
   }
 
   fun addOptimisticUpdate(record: Record): Set<String> {
-    return lock.write {
-      val journal = recordJournals[record.key]
-      if (journal == null) {
-        recordJournals[record.key] = RecordJournal(record)
-        record.fieldKeys()
-      } else {
-        journal.addPatch(record)
-      }
+    val journal = recordJournals[record.key]
+    return if (journal == null) {
+      recordJournals[record.key] = RecordJournal(record)
+      record.fieldKeys()
+    } else {
+      journal.addPatch(record)
     }
   }
 
   fun removeOptimisticUpdates(mutationId: Uuid): Set<String> {
-    return lock.write {
-      val changedCacheKeys = mutableSetOf<String>()
-      val iterator = recordJournals.iterator()
-      while (iterator.hasNext()) {
-        val entry = iterator.next()
-        val result = entry.value.removePatch(mutationId)
-        changedCacheKeys.addAll(result.changedKeys)
-        if (result.isEmpty) {
-          iterator.remove()
-        }
+    val changedCacheKeys = mutableSetOf<String>()
+    val keys = HashSet(recordJournals.keys) // local copy to avoid concurrent modification
+    keys.forEach {
+      val recordJournal = recordJournals[it] ?: return@forEach
+      val result = recordJournal.removePatch(mutationId)
+      changedCacheKeys.addAll(result.changedKeys)
+      if (result.isEmpty) {
+        recordJournals.remove(it)
       }
-      changedCacheKeys
     }
+    return changedCacheKeys
   }
 
   override fun dump(): Map<KClass<*>, Map<String, Record>> {
-    return lock.read {
-      mapOf(this::class to recordJournals.mapValues { (_, journal) -> journal.current }) + nextCache?.dump().orEmpty()
-    }
+    return mapOf(this::class to recordJournals.mapValues { (_, journal) -> journal.current }) + wrapped.dump()
   }
 
   private fun Record?.mergeJournalRecord(key: String): Record? {
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/RecordWeigher.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/RecordWeigher.kt
index 90592e37d9..4f49c07d71 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/RecordWeigher.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/RecordWeigher.kt
@@ -2,7 +2,6 @@ package com.apollographql.apollo3.cache.normalized.api.internal
 
 import com.apollographql.apollo3.cache.normalized.api.CacheKey
 import com.apollographql.apollo3.cache.normalized.api.Record
-import okio.utf8Size
 import kotlin.jvm.JvmStatic
 
 internal object RecordWeigher {
@@ -23,9 +22,9 @@ internal object RecordWeigher {
 
   @JvmStatic
   fun calculateBytes(record: Record): Int {
-    var size = SIZE_OF_RECORD_OVERHEAD + record.key.utf8Size().toInt()
+    var size = SIZE_OF_RECORD_OVERHEAD + record.key.length
     for ((key, value) in record.fields) {
-      size += key.utf8Size().toInt() + weighField(value)
+      size += key.length + weighField(value)
     }
     size += weighField(record.metadata)
     size += weighField(record.dates)
@@ -35,7 +34,7 @@ internal object RecordWeigher {
   private fun weighField(field: Any?): Int {
     return when (field) {
       null -> SIZE_OF_NULL
-      is String -> field.utf8Size().toInt()
+      is String -> field.length
       is Boolean -> SIZE_OF_BOOLEAN
       is Int -> SIZE_OF_INT
       is Long -> SIZE_OF_LONG // Might happen with LongDataAdapter
@@ -45,7 +44,7 @@ internal object RecordWeigher {
       }
 
       is CacheKey -> {
-        SIZE_OF_CACHE_KEY_OVERHEAD + field.key.utf8Size().toInt()
+        SIZE_OF_CACHE_KEY_OVERHEAD + field.key.length
       }
       /**
        * Custom scalars with a json object representation are stored directly in the record
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/patternToRegex.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/patternToRegex.kt
new file mode 100644
index 0000000000..26d2f45e60
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/patternToRegex.kt
@@ -0,0 +1,39 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+
+/**
+ * A tentative to approximate the Sqlite LIKE operator with Regexes
+ */
+internal fun patternToRegex(pattern: String): Regex {
+  val regex = buildString {
+    var pendingEscape = false
+    for (i in pattern.indices) {
+      val cur = pattern[i]
+      when {
+        pendingEscape -> {
+          when {
+            cur == '\\' -> append("\\\\") // an escaped backslash is also an escape backslash in a regex
+            cur == '%' -> append("%")
+            cur == '_' -> append("_")
+            else -> error("Invalid escape in pattern: $pattern")
+          }
+        }
+
+        cur == '\\' -> pendingEscape = true
+        cur == '%' -> append(".*")
+        cur == '_' -> append(".")
+        else -> {
+          if (specialChars.contains(cur)) {
+            // this needs to be escaped in the regex
+            append("\\")
+          }
+          append(cur)
+        }
+      }
+    }
+  }
+
+  return Regex(regex, option = RegexOption.IGNORE_CASE)
+}
+
+private val specialChars = "()^$.*?+{}"
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Cache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Cache.kt
new file mode 100644
index 0000000000..e21c34e2f7
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Cache.kt
@@ -0,0 +1,82 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+interface Cache<Key : Any, Value : Any> {
+    /**
+     * @return [Value] associated with [key] or `null` if there is no cached value for [key].
+     */
+    fun getIfPresent(key: Key): Value?
+
+    /**
+     * @return [Value] associated with [key], obtaining the value from [valueProducer] if necessary.
+     * No observable state associated with this cache is modified until loading completes.
+     * @param [valueProducer] Must not return `null`. It may either return a non-null value or throw an exception.
+     * @throws ExecutionExeption If a checked exception was thrown while loading the value.
+     * @throws UncheckedExecutionException If an unchecked exception was thrown while loading the value.
+     * @throws ExecutionError If an error was thrown while loading the value.
+     */
+    fun getOrPut(key: Key, valueProducer: () -> Value): Value
+
+    /**
+     * @return Map of the [Value] associated with each [Key] in [keys]. Returned map only contains entries already present in the cache.
+     */
+    fun getAllPresent(keys: List<*>): Map<Key, Value>
+
+    /**
+     * @return Map of the [Value] associated with each [Key] in the cache.
+     */
+    fun getAllPresent(): Map<Key, Value>
+
+    /**
+     * Associates [value] with [key].
+     * If the cache previously contained a value associated with [key], the old value is replaced by [value].
+     * Prefer [getOrPut] when using the conventional "If cached, then return. Otherwise create, cache, and then return" pattern.
+     */
+    fun put(key: Key, value: Value)
+
+    /**
+     * Copies all of the mappings from the specified map to the cache. The effect of this call is
+     * equivalent to that of calling [put] on this map once for each mapping from [Key] to [Value] in the specified map.
+     * The behavior of this operation is undefined if the specified map is modified while the operation is in progress.
+     */
+    fun putAll(map: Map<Key, Value>)
+
+    /**
+     * Discards any cached value associated with [key].
+     */
+    fun invalidate(key: Key)
+
+    /**
+     * Discards any cached value associated for [keys].
+     */
+    fun invalidateAll(keys: List<Key>)
+
+    /**
+     * Discards all entries in the cache.
+     */
+    fun invalidateAll()
+
+    /**
+     * @return Approximate number of entries in the cache.
+     */
+    fun size(): Long
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/CacheBuilder.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/CacheBuilder.kt
new file mode 100644
index 0000000000..59ce46b26a
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/CacheBuilder.kt
@@ -0,0 +1,87 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (c) 2010 The Guava Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this library except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+import kotlin.time.Duration
+
+class CacheBuilder<Key : Any, Output : Any> {
+    internal var concurrencyLevel = 4
+        private set
+    internal val initialCapacity = 16
+    internal var maximumSize = UNSET
+        private set
+    internal var maximumWeight = UNSET
+        private set
+    internal var expireAfterAccess: Duration = Duration.INFINITE
+        private set
+    internal var expireAfterWrite: Duration = Duration.INFINITE
+        private set
+    internal var weigher: Weigher<Key, Output>? = null
+        private set
+    internal var ticker: Ticker? = null
+        private set
+
+    fun concurrencyLevel(producer: () -> Int): CacheBuilder<Key, Output> = apply {
+        concurrencyLevel = producer.invoke()
+    }
+
+    fun maximumSize(maximumSize: Long): CacheBuilder<Key, Output> = apply {
+        if (maximumSize < 0) {
+            throw IllegalArgumentException("Maximum size must be non-negative.")
+        }
+        this.maximumSize = maximumSize
+    }
+
+    fun expireAfterAccess(duration: Duration): CacheBuilder<Key, Output> = apply {
+        if (duration.isNegative()) {
+            throw IllegalArgumentException("Duration must be non-negative.")
+        }
+        expireAfterAccess = duration
+    }
+
+    fun expireAfterWrite(duration: Duration): CacheBuilder<Key, Output> = apply {
+        if (duration.isNegative()) {
+            throw IllegalArgumentException("Duration must be non-negative.")
+        }
+        expireAfterWrite = duration
+    }
+
+    fun ticker(ticker: Ticker): CacheBuilder<Key, Output> = apply {
+        this.ticker = ticker
+    }
+
+    fun weigher(maximumWeight: Long, weigher: Weigher<Key, Output>): CacheBuilder<Key, Output> = apply {
+        if (maximumWeight < 0) {
+            throw IllegalArgumentException("Maximum weight must be non-negative.")
+        }
+
+        this.maximumWeight = maximumWeight
+        this.weigher = weigher
+    }
+
+    fun build(): Cache<Key, Output> {
+        if (maximumSize != -1L && weigher != null) {
+            throw IllegalStateException("Maximum size cannot be combined with weigher.")
+        }
+        return LocalCache.LocalManualCache(this)
+    }
+
+    companion object {
+        private const val UNSET = -1L
+    }
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/LocalCache.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/LocalCache.kt
new file mode 100644
index 0000000000..03fc9075b0
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/LocalCache.kt
@@ -0,0 +1,1980 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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.
+ *
+ * KMP conversion
+ * Copyright (C) 2022 André Claßen
+ */
+
+@file:Suppress("UNUSED_PARAMETER")
+
+package com.apollographql.apollo3.cache.normalized.api.internal.store
+
+import kotlinx.atomicfu.AtomicArray
+import kotlinx.atomicfu.AtomicRef
+import kotlinx.atomicfu.atomic
+import kotlinx.atomicfu.atomicArrayOfNulls
+import kotlinx.atomicfu.locks.reentrantLock
+import kotlinx.atomicfu.loop
+import kotlin.math.min
+import kotlin.time.Duration
+
+internal class LocalCache<K : Any, V : Any>(builder: CacheBuilder<K, V>) {
+
+    /**
+     * Mask value for indexing into segments. The upper bits of a key's hash code are used to choose
+     * the segment.
+     */
+    private val segmentMask: Int
+
+    /**
+     * Shift value for indexing within segments. Helps prevent entries that end up in the same segment
+     * from also ending up in the same bucket.
+     */
+    private val segmentShift: Int
+
+    /**
+     * The segments, each of which is a specialized hash table.
+     */
+
+    private val segments: Array<Segment<K, V>?>
+
+    /**
+     * Strategy for referencing values.
+     */
+    private val valueStrength: Strength = Strength.Strong
+
+    /**
+     * The maximum weight of this map. UNSET_LONG if there is no maximum.
+     */
+    private val maxWeight: Long
+
+    /**
+     * Weigher to weigh cache entries.
+     */
+    private val weigher: Weigher<K, V>
+
+    /**
+     * How long after the last access to an entry the map will retain that entry.
+     */
+    private val expireAfterAccessNanos: Long
+
+    /**
+     * How long after the last write to an entry the map will retain that entry.
+     */
+    private val expireAfterWriteNanos: Long
+
+    /**
+     * Measures time in a testable way.
+     */
+    private val ticker: Ticker
+
+    /**
+     * Factory used to create new entries.
+     */
+    private val entryFactory: EntryFactory
+
+    private val evictsBySize: Boolean get() = maxWeight >= 0
+
+    private val customWeigher: Boolean get() = weigher !== OneWeigher
+
+    private val expiresAfterWrite: Boolean get() = expireAfterWriteNanos > 0
+
+    private val expiresAfterAccess: Boolean get() = expireAfterAccessNanos > 0
+
+    private val usesAccessQueue: Boolean get() = expiresAfterAccess || evictsBySize
+
+    private val usesWriteQueue: Boolean get() = expiresAfterWrite
+
+    private val recordsWrite: Boolean get() = expiresAfterWrite
+
+    private val recordsAccess: Boolean get() = expiresAfterAccess
+
+    private val recordsTime: Boolean get() = recordsWrite || recordsAccess
+
+    private val usesWriteEntries: Boolean get() = usesWriteQueue || recordsWrite
+
+    private val usesAccessEntries: Boolean get() = usesAccessQueue || recordsAccess
+
+    private sealed class Strength {
+        /*
+         * TODO(kevinb): If we strongly reference the value and aren't loading, we needn't wrap the
+         * value. This could save ~8 bytes per entry.
+         */
+        object Strong : Strength() {
+            override fun <K : Any, V : Any> referenceValue(
+                segment: Segment<K, V>?,
+                entry: ReferenceEntry<K, V>?,
+                value: V,
+                weight: Int
+            ): ValueReference<K, V> {
+                return if (weight == 1) StrongValueReference(value) else WeightedStrongValueReference(
+                    value,
+                    weight
+                )
+            }
+        }
+
+        /**
+         * Creates a reference for the given value according to this value strength.
+         */
+        abstract fun <K : Any, V : Any> referenceValue(
+            segment: Segment<K, V>?,
+            entry: ReferenceEntry<K, V>?,
+            value: V,
+            weight: Int
+        ): ValueReference<K, V>
+    }
+
+    /**
+     * Creates new entries.
+     */
+    private sealed class EntryFactory {
+        object Strong : EntryFactory() {
+            override fun <K : Any, V : Any> newEntry(
+                segment: Segment<K, V>?,
+                key: K,
+                hash: Int,
+                next: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                return StrongEntry(key, hash, next)
+            }
+        }
+
+        object StrongAccess : EntryFactory() {
+
+            override fun <K : Any, V : Any> newEntry(
+                segment: Segment<K, V>?,
+                key: K,
+                hash: Int,
+                next: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                return StrongAccessEntry(key, hash, next)
+            }
+
+            override fun <K : Any, V : Any> copyEntry(
+                segment: Segment<K, V>?,
+                original: ReferenceEntry<K, V>,
+                newNext: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                val newEntry = super.copyEntry(segment, original, newNext)
+                copyAccessEntry(original, newEntry)
+                return newEntry
+            }
+        }
+
+        object StrongWrite : EntryFactory() {
+
+            override fun <K : Any, V : Any> newEntry(
+                segment: Segment<K, V>?,
+                key: K,
+                hash: Int,
+                next: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                return StrongWriteEntry(key, hash, next)
+            }
+
+            override fun <K : Any, V : Any> copyEntry(
+                segment: Segment<K, V>?,
+                original: ReferenceEntry<K, V>,
+                newNext: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                val newEntry = super.copyEntry(segment, original, newNext)
+                copyWriteEntry(original, newEntry)
+                return newEntry
+            }
+        }
+
+        object StrongAccessWrite : EntryFactory() {
+
+            override fun <K : Any, V : Any> newEntry(
+                segment: Segment<K, V>?,
+                key: K,
+                hash: Int,
+                next: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                return StrongAccessWriteEntry(key, hash, next)
+            }
+
+            override fun <K : Any, V : Any> copyEntry(
+                segment: Segment<K, V>?,
+                original: ReferenceEntry<K, V>,
+                newNext: ReferenceEntry<K, V>?
+            ): ReferenceEntry<K, V> {
+                val newEntry = super.copyEntry(segment, original, newNext)
+                copyAccessEntry(original, newEntry)
+                copyWriteEntry(original, newEntry)
+                return newEntry
+            }
+        }
+
+        /**
+         * Creates a new entry.
+         *
+         * @param segment to create the entry for
+         * @param key     of the entry
+         * @param hash    of the key
+         * @param next    entry in the same bucket
+         */
+        abstract fun <K : Any, V : Any> newEntry(
+            segment: Segment<K, V>?,
+            key: K,
+            hash: Int,
+            next: ReferenceEntry<K, V>?
+        ): ReferenceEntry<K, V>
+
+        /**
+         * Copies an entry, assigning it a new `next` entry.
+         *
+         * @param original the entry to copy
+         * @param newNext  entry in the same bucket
+         */
+        // Guarded By Segment.this
+        open fun <K : Any, V : Any> copyEntry(
+            segment: Segment<K, V>?,
+            original: ReferenceEntry<K, V>,
+            newNext: ReferenceEntry<K, V>?
+        ): ReferenceEntry<K, V> {
+            return newEntry(segment, original.key, original.hash, newNext)
+        }
+
+        // Guarded By Segment.this
+        fun <K : Any, V : Any> copyAccessEntry(
+            original: ReferenceEntry<K, V>,
+            newEntry: ReferenceEntry<K, V>
+        ) {
+            // TODO(fry): when we link values instead of entries this method can go
+            // away, as can connectAccessOrder, nullifyAccessOrder.
+            newEntry.accessTime = original.accessTime
+            connectAccessOrder(original.previousInAccessQueue, newEntry)
+            connectAccessOrder(newEntry, original.nextInAccessQueue)
+            nullifyAccessOrder(original)
+        }
+
+        // Guarded By Segment.this
+        fun <K : Any, V : Any> copyWriteEntry(
+            original: ReferenceEntry<K, V>,
+            newEntry: ReferenceEntry<K, V>
+        ) {
+            // TODO(fry): when we link values instead of entries this method can go
+            // away, as can connectWriteOrder, nullifyWriteOrder.
+            newEntry.writeTime = original.writeTime
+            connectWriteOrder(original.previousInWriteQueue, newEntry)
+            connectWriteOrder(newEntry, original.nextInWriteQueue)
+            nullifyWriteOrder(original)
+        }
+
+        companion object {
+            /**
+             * Masks used to compute indices in the following table.
+             */
+            private const val ACCESS_MASK = 1
+            private const val WRITE_MASK = 2
+
+            /**
+             * Look-up table for factories.
+             */
+            private val factories = arrayOf(Strong, StrongAccess, StrongWrite, StrongAccessWrite)
+            fun getFactory(usesAccessQueue: Boolean, usesWriteQueue: Boolean): EntryFactory {
+                val flags = ((if (usesAccessQueue) ACCESS_MASK else 0) or if (usesWriteQueue) WRITE_MASK else 0)
+                return factories[flags]
+            }
+        }
+    }
+
+    /**
+     * A reference to a value.
+     */
+    private interface ValueReference<K : Any, V : Any> {
+        /**
+         * Returns the value. Does not block or throw exceptions.
+         */
+        fun get(): V?
+
+        /**
+         * Returns the weight of this entry. This is assumed to be static between calls to setValue.
+         */
+        val weight: Int
+
+        /**
+         * Returns the entry associated with this value reference, or `null` if this value
+         * reference is independent of any entry.
+         */
+        val entry: ReferenceEntry<K, V>?
+
+        /**
+         * Creates a copy of this reference for the given entry.
+         *
+         *
+         *
+         * `value` may be null only for a loading reference.
+         */
+
+        fun copyFor(value: V?, entry: ReferenceEntry<K, V>?): ValueReference<K, V>
+
+        /**
+         * Notifify pending loads that a new value was set. This is only relevant to loading
+         * value references.
+         */
+        fun notifyNewValue(newValue: V)
+
+        /**
+         * Returns true if this reference contains an active value, meaning one that is still considered
+         * present in the cache. Active values consist of live values, which are returned by cache
+         * lookups, and dead values, which have been evicted but awaiting removal. Non-active values
+         * consist strictly of loading values, though during refresh a value may be both active and
+         * loading.
+         */
+        val isActive: Boolean
+    }
+
+    /**
+     * An entry in a reference map.
+     *
+     *
+     * Entries in the map can be in the following states:
+     *
+     *
+     * Valid:
+     * - Live: valid key/value are set
+     * - Loading: loading is pending
+     *
+     *
+     * Invalid:
+     * - Expired: time expired (key/value may still be set)
+     * - Collected: key/value was partially collected, but not yet cleaned up
+     * - Unset: marked as unset, awaiting cleanup or reuse
+     */
+    private interface ReferenceEntry<K : Any, V : Any> {
+        /**
+         * Returns the value reference from this entry.
+         */
+        /**
+         * Sets the value reference for this entry.
+         */
+        var valueReference: ValueReference<K, V>?
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+
+        /**
+         * Returns the next entry in the chain.
+         */
+        val next: ReferenceEntry<K, V>?
+            get() = throw UnsupportedOperationException()
+
+        /**
+         * Returns the entry's hash.
+         */
+        val hash: Int
+            get() = throw UnsupportedOperationException()
+
+        /**
+         * Returns the key for this entry.
+         */
+        val key: K
+            get() = throw UnsupportedOperationException()
+        /*
+         * Used by entries that use access order. Access entries are maintained in a doubly-linked list.
+         * New entries are added at the tail of the list at write time; stale entries are expired from
+         * the head of the list.
+         */
+        /**
+         * Returns the time that this entry was last accessed, in ns.
+         */
+        /**
+         * Sets the entry access time in ns.
+         */
+        var accessTime: Long
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+        /**
+         * Returns the next entry in the access queue.
+         */
+        /**
+         * Sets the next entry in the access queue.
+         */
+        var nextInAccessQueue: ReferenceEntry<K, V>
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+        /**
+         * Returns the previous entry in the access queue.
+         */
+        /**
+         * Sets the previous entry in the access queue.
+         */
+        var previousInAccessQueue: ReferenceEntry<K, V>
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+        /*
+         * Implemented by entries that use write order. Write entries are maintained in a
+         * doubly-linked list. New entries are added at the tail of the list at write time and stale
+         * entries are expired from the head of the list.
+         */
+        /**
+         * Returns the time that this entry was last written, in ns.
+         */
+        /**
+         * Sets the entry write time in ns.
+         */
+        var writeTime: Long
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+        /**
+         * Returns the next entry in the write queue.
+         */
+        /**
+         * Sets the next entry in the write queue.
+         */
+        var nextInWriteQueue: ReferenceEntry<K, V>
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+        /**
+         * Returns the previous entry in the write queue.
+         */
+        /**
+         * Sets the previous entry in the write queue.
+         */
+        var previousInWriteQueue: ReferenceEntry<K, V>
+            get() = throw UnsupportedOperationException()
+            set(_) = throw UnsupportedOperationException()
+    }
+
+    private object NullEntry : ReferenceEntry<Any, Any> {
+        override var valueReference: ValueReference<Any, Any>?
+            get() = null
+            set(_) {}
+
+        override val next: ReferenceEntry<Any, Any>?
+            get() = null
+
+        override val hash: Int
+            get() = 0
+
+        override val key: Any
+            get() = Unit
+
+        override var accessTime: Long
+            get() = 0
+            set(_) {}
+
+        override var nextInAccessQueue: ReferenceEntry<Any, Any>
+            get() = this
+            set(_) {}
+
+        override var previousInAccessQueue: ReferenceEntry<Any, Any>
+            get() = this
+            set(_) {}
+
+        override var writeTime: Long
+            get() = 0
+            set(_) {}
+
+        override var nextInWriteQueue: ReferenceEntry<Any, Any>
+            get() = this
+            set(_) {}
+
+        override var previousInWriteQueue: ReferenceEntry<Any, Any>
+            get() = this
+            set(_) {}
+    }
+
+    /*
+     * Note: All of this duplicate code sucks, but it saves a lot of memory. If only Java had mixins!
+     * To maintain this code, make a change for the strong reference type. Then, cut and paste, and
+     * replace "Strong" with "Soft" or "Weak" within the pasted text. The primary difference is that
+     * strong entries store the key reference directly while soft and weak entries delegate to their
+     * respective superclasses.
+     */
+    /**
+     * Used for strongly-referenced keys.
+     */
+    private open class StrongEntry<K : Any, V : Any>(
+        override val key: K, // The code below is exactly the same for each entry type.
+        override val hash: Int,
+        override val next: ReferenceEntry<K, V>?
+    ) : ReferenceEntry<K, V> {
+
+        private val _valueReference = atomic<ValueReference<K, V>?>(unset())
+        override var valueReference: ValueReference<K, V>? = _valueReference.value
+    }
+
+    private class StrongAccessEntry<K : Any, V : Any>(
+        key: K,
+        hash: Int,
+        next: ReferenceEntry<K, V>?
+    ) :
+        StrongEntry<K, V>(key, hash, next) {
+        // The code below is exactly the same for each access entry type.
+
+        private val _accessTime = atomic(Long.MAX_VALUE)
+        override var accessTime = _accessTime.value
+
+        // Guarded By Segment.this
+        override var nextInAccessQueue: ReferenceEntry<K, V> = nullEntry()
+
+        // Guarded By Segment.this
+        override var previousInAccessQueue: ReferenceEntry<K, V> = nullEntry()
+    }
+
+    private class StrongWriteEntry<K : Any, V : Any>(
+        key: K,
+        hash: Int,
+        next: ReferenceEntry<K, V>?
+    ) :
+        StrongEntry<K, V>(key, hash, next) {
+        // The code below is exactly the same for each write entry type.
+        private val _writeTime = atomic(Long.MAX_VALUE)
+        override var writeTime = _writeTime.value
+
+        // Guarded By Segment.this
+        override var nextInWriteQueue: ReferenceEntry<K, V> = nullEntry()
+
+        // Guarded By Segment.this
+        override var previousInWriteQueue: ReferenceEntry<K, V> = nullEntry()
+    }
+
+    private class StrongAccessWriteEntry<K : Any, V : Any>(
+        key: K,
+        hash: Int,
+        next: ReferenceEntry<K, V>?
+    ) :
+        StrongEntry<K, V>(key, hash, next) {
+        // The code below is exactly the same for each access entry type.
+        private val _accessTime = atomic(Long.MAX_VALUE)
+        override var accessTime: Long = _accessTime.value
+
+        // Guarded By Segment.this
+        override var nextInAccessQueue: ReferenceEntry<K, V> = nullEntry()
+
+        // Guarded By Segment.this
+        override var previousInAccessQueue: ReferenceEntry<K, V> = nullEntry()
+
+        // The code below is exactly the same for each write entry type.
+        private val _writeTime = atomic(Long.MAX_VALUE)
+        override var writeTime: Long = _writeTime.value
+
+        // Guarded By Segment.this
+        override var nextInWriteQueue: ReferenceEntry<K, V> = nullEntry()
+
+        // Guarded By Segment.this
+        override var previousInWriteQueue: ReferenceEntry<K, V> = nullEntry()
+    }
+
+    /**
+     * References a strong value.
+     */
+    private open class StrongValueReference<K : Any, V : Any>(private val referent: V) :
+        ValueReference<K, V> {
+        override fun get(): V = referent
+        override val weight: Int = 1
+        override val entry: ReferenceEntry<K, V>? = null
+        override fun copyFor(value: V?, entry: ReferenceEntry<K, V>?): ValueReference<K, V> = this
+        override val isActive: Boolean = true
+        override fun notifyNewValue(newValue: V) {}
+    }
+
+    /**
+     * References a strong value.
+     */
+    private class WeightedStrongValueReference<K : Any, V : Any>(
+        referent: V,
+        override val weight: Int
+    ) :
+        StrongValueReference<K, V>(referent)
+
+    /**
+     * This method is a convenience for testing. Code should call [Segment.newEntry] directly.
+     */
+
+    private fun newEntry(key: K, hash: Int, next: ReferenceEntry<K, V>?): ReferenceEntry<K, V> {
+        val segment = segmentFor(hash)
+        segment.reentrantLock.lock()
+        return try {
+            segment.newEntry(key, hash, next)
+        } finally {
+            segment.reentrantLock.unlock()
+        }
+    }
+
+    /**
+     * This method is a convenience for testing. Code should call [Segment.copyEntry] directly.
+     */
+    // Guarded By Segment.this
+    private fun copyEntry(
+        original: ReferenceEntry<K, V>,
+        newNext: ReferenceEntry<K, V>?
+    ): ReferenceEntry<K, V>? {
+        val hash = original.hash
+        return segmentFor(hash).copyEntry(original, newNext)
+    }
+
+    /**
+     * This method is a convenience for testing. Code should call [Segment.setValue] instead.
+     */
+    // Guarded By Segment.this
+    private fun newValueReference(
+        entry: ReferenceEntry<K, V>,
+        value: V,
+        weight: Int
+    ): ValueReference<K, V> {
+        val hash = entry.hash
+        return valueStrength.referenceValue(segmentFor(hash), entry, value, weight)
+    }
+
+    private fun hash(key: K): Int = rehash(key.hashCode())
+
+    /**
+     * Returns the segment that should be used for a key with the given hash.
+     *
+     * @param hash the hash code for the key
+     * @return the segment
+     */
+    private fun segmentFor(hash: Int): Segment<K, V> = // TODO(fry): Lazily create segments?
+        segments[hash ushr segmentShift and segmentMask] as Segment<K, V>
+
+    private fun createSegment(initialCapacity: Int, maxSegmentWeight: Long): Segment<K, V> =
+        Segment(this, initialCapacity, maxSegmentWeight)
+    // expiration
+    /**
+     * Returns true if the entry has expired.
+     */
+    private fun isExpired(entry: ReferenceEntry<K, V>, now: Long): Boolean =
+        if (expiresAfterAccess && now - entry.accessTime >= expireAfterAccessNanos) {
+            true
+        } else {
+            expiresAfterWrite && now - entry.writeTime >= expireAfterWriteNanos
+        }
+
+    // Inner Classes
+
+    private class SegmentTable<K : Any, V : Any>(val size: Int) {
+        private val table: AtomicArray<ReferenceEntry<K, V>?> = atomicArrayOfNulls(size)
+        operator fun get(idx: Int) = table[idx].value
+        operator fun set(idx: Int, value: ReferenceEntry<K, V>?) {
+            table[idx].value = value
+        }
+    }
+
+    /**
+     * Segments are specialized versions of hash tables.
+     */
+    private class Segment<K : Any, V : Any>(
+        private val map: LocalCache<K, V>,
+        initialCapacity: Int,
+        private val maxSegmentWeight: Long
+    ) {
+        /*
+         * TODO(fry): Consider copying variables (like evictsBySize) from outer class into this class.
+         * It will require more memory but will reduce indirection.
+         */
+        /*
+         * Segments maintain a table of entry lists that are ALWAYS kept in a consistent state, so can
+         * be read without locking. Next fields of nodes are immutable (final). All list additions are
+         * performed at the front of each bin. This makes it easy to check changes, and also fast to
+         * traverse. When nodes would otherwise be changed, new nodes are created to replace them. This
+         * works well for hash tables since the bin lists tend to be short. (The average length is less
+         * than two.)
+         *
+         * Read operations can thus proceed without locking, but rely on selected uses of volatiles to
+         * ensure that completed write operations performed by other threads are noticed. For most
+         * purposes, the "count" field, tracking the number of elements, serves as that volatile
+         * variable ensuring visibility. This is convenient because this field needs to be read in many
+         * read operations anyway:
+         *
+         * - All (unsynchronized) read operations must first read the "count" field, and should not
+         * look at table entries if it is 0.
+         *
+         * - All (synchronized) write operations should write to the "count" field after structurally
+         * changing any bin. The operations must not take any action that could even momentarily
+         * cause a concurrent read operation to see inconsistent data. This is made easier by the
+         * nature of the read operations in Map. For example, no operation can reveal that the table
+         * has grown but the threshold has not yet been updated, so there are no atomicity requirements
+         * for this with respect to reads.
+         *
+         * As a guide, all critical volatile reads and writes to the count field are marked in code
+         * comments.
+         */
+
+        val reentrantLock = reentrantLock()
+
+        /**
+         * The number of live elements in this segment's region.
+         */
+        private val count = atomic(0)
+
+        /**
+         * The weight of the live elements in this segment's region.
+         */
+        private var totalWeight: Long = 0
+
+        /**
+         * Number of updates that alter the size of the table. This is used during bulk-read methods to
+         * make sure they see a consistent snapshot: If modCounts change during a traversal of segments
+         * loading size or checking containsValue, then we might have an inconsistent view of state
+         * so (usually) must retry.
+         */
+        private var modCount = 0
+
+        /**
+         * The table is expanded when its size exceeds this threshold. (The value of this field is
+         * always `(int) (capacity * 0.75)`.)
+         */
+        private var threshold = 0
+
+        /**
+         * The per-segment table.
+         */
+        private val table: AtomicRef<SegmentTable<K, V>>
+
+        /**
+         * The recency queue is used to record which entries were accessed for updating the access
+         * list's ordering. It is drained as a batch operation when either the DRAIN_THRESHOLD is
+         * crossed or a write occurs on the segment.
+         */
+        private val recencyQueue: Queue<ReferenceEntry<K, V>>
+
+        /**
+         * A counter of the number of reads since the last write, used to drain queues on a small
+         * fraction of read operations.
+         */
+        private val readCount = atomic(0)
+
+        /**
+         * A queue of elements currently in the map, ordered by write time. Elements are added to the
+         * tail of the queue on write.
+         */
+        private val writeQueue: MutableQueue<ReferenceEntry<K, V>>
+
+        /**
+         * A queue of elements currently in the map, ordered by access time. Elements are added to the
+         * tail of the queue on access (note that writes count as accesses).
+         */
+        private val accessQueue: MutableQueue<ReferenceEntry<K, V>>
+
+        fun newEntry(key: K, hash: Int, next: ReferenceEntry<K, V>?): ReferenceEntry<K, V> =
+            map.entryFactory.newEntry(this, key, hash, next)
+
+        /**
+         * Copies `original` into a new entry chained to `newNext`. Returns the new entry,
+         * or `null` if `original` was already garbage collected.
+         */
+        fun copyEntry(
+            original: ReferenceEntry<K, V>,
+            newNext: ReferenceEntry<K, V>?
+        ): ReferenceEntry<K, V>? {
+            val valueReference = original.valueReference
+            val value = valueReference!!.get()
+            if (value == null && valueReference.isActive) {
+                // value collected
+                return null
+            }
+            val newEntry = map.entryFactory.copyEntry(this, original, newNext)
+            newEntry.valueReference = valueReference.copyFor(value, newEntry)
+            return newEntry
+        }
+
+        /**
+         * Sets a new value of an entry. Adds newly created entries at the end of the access queue.
+         */
+        fun setValue(entry: ReferenceEntry<K, V>, key: K, value: V, now: Long) {
+            val previous = entry.valueReference
+            val weight = map.weigher(key, value)
+            if (weight < 0) throw IllegalStateException("Weights must be non-negative")
+            entry.valueReference = map.valueStrength.referenceValue(this, entry, value, weight)
+            recordWrite(entry, weight, now)
+            previous?.notifyNewValue(value)
+        }
+
+        // recency queue, shared by expiration and eviction
+        /**
+         * Records the relative order in which this read was performed by adding `entry` to the
+         * recency queue. At write-time, or when the queue is full past the threshold, the queue will
+         * be drained and the entries therein processed.
+         *
+         *
+         *
+         * Note: locked reads should use [.recordLockedRead].
+         */
+        private fun recordRead(entry: ReferenceEntry<K, V>, now: Long) {
+            if (map.recordsAccess) {
+                entry.accessTime = now
+            }
+            recencyQueue.add(entry)
+        }
+
+        /**
+         * Updates the eviction metadata that `entry` was just read. This currently amounts to
+         * adding `entry` to relevant eviction lists.
+         *
+         *
+         *
+         * Note: this method should only be called under lock, as it directly manipulates the
+         * eviction queues. Unlocked reads should use [.recordRead].
+         */
+        private fun recordLockedRead(entry: ReferenceEntry<K, V>, now: Long) {
+            if (map.recordsAccess) {
+                entry.accessTime = now
+            }
+            accessQueue.add(entry)
+        }
+
+        /**
+         * Updates eviction metadata that `entry` was just written. This currently amounts to
+         * adding `entry` to relevant eviction lists.
+         */
+        private fun recordWrite(entry: ReferenceEntry<K, V>, weight: Int, now: Long) {
+            // we are already under lock, so drain the recency queue immediately
+            drainRecencyQueue()
+            totalWeight += weight.toLong()
+            if (map.recordsAccess) {
+                entry.accessTime = now
+            }
+            if (map.recordsWrite) {
+                entry.writeTime = now
+            }
+            accessQueue.add(entry)
+            writeQueue.add(entry)
+        }
+
+        /**
+         * Drains the recency queue, updating eviction metadata that the entries therein were read in
+         * the specified relative order. This currently amounts to adding them to relevant eviction
+         * lists (accounting for the fact that they could have been removed from the map since being
+         * added to the recency queue).
+         */
+        private fun drainRecencyQueue() {
+            while (true) {
+                val e = recencyQueue.poll() ?: break
+                // An entry may be in the recency queue despite it being removed from
+                // the map . This can occur when the entry was concurrently read while a
+                // writer is removing it from the segment or after a clear has removed
+                // all of the segment's entries.
+                if (accessQueue.contains(e)) {
+                    accessQueue.add(e)
+                }
+            }
+        }
+        // expiration
+        /**
+         * Cleanup expired entries when the lock is available.
+         */
+        private fun tryExpireEntries(now: Long) {
+            if (reentrantLock.tryLock()) {
+                try {
+                    expireEntries(now)
+                } finally {
+                    reentrantLock.unlock()
+                    // don't call postWriteCleanup as we're in a read
+                }
+            }
+        }
+
+        private fun expireEntries(now: Long) {
+            drainRecencyQueue()
+            while (true) {
+                val e = writeQueue.peek()?.takeIf { map.isExpired(it, now) } ?: break
+                if (!removeEntry(e, e.hash, RemovalCause.EXPIRED)) {
+                    throw AssertionError()
+                }
+            }
+
+            while (true) {
+                val e = accessQueue.peek()?.takeIf { map.isExpired(it, now) } ?: break
+                if (!removeEntry(e, e.hash, RemovalCause.EXPIRED)) {
+                    throw AssertionError()
+                }
+            }
+        }
+
+        // eviction
+        private fun enqueueNotification(entry: ReferenceEntry<K, V>, cause: RemovalCause?) {
+            enqueueNotification(entry.key, entry.hash, entry.valueReference, cause)
+        }
+
+        private fun enqueueNotification(
+            key: K?,
+            hash: Int,
+            valueReference: ValueReference<K, V>?,
+            cause: RemovalCause?
+        ) {
+            valueReference?.weight?.toLong()?.apply {
+                totalWeight -= this
+            }
+        }
+
+        /**
+         * Performs eviction if the segment is over capacity. Avoids flushing the entire cache if the
+         * newest entry exceeds the maximum weight all on its own.
+         *
+         * @param newest the most recently added entry
+         */
+        private fun evictEntries(newest: ReferenceEntry<K, V>) {
+            if (!map.evictsBySize) {
+                return
+            }
+            drainRecencyQueue()
+
+            // If the newest entry by itself is too heavy for the segment, don't bother evicting
+            // anything else, just that
+            if (newest.valueReference!!.weight > maxSegmentWeight) {
+                if (!removeEntry(newest, newest.hash, RemovalCause.SIZE)) {
+                    throw AssertionError()
+                }
+            }
+            while (totalWeight > maxSegmentWeight) {
+                val e = nextEvictable
+                if (!removeEntry(e, e.hash, RemovalCause.SIZE)) {
+                    throw AssertionError()
+                }
+            }
+        }
+
+        // TODO(fry): instead implement this with an eviction head
+
+        private val nextEvictable: ReferenceEntry<K, V>
+            get() {
+                for (e in accessQueue) {
+                    val weight = e.valueReference!!.weight
+                    if (weight > 0) {
+                        return e
+                    }
+                }
+                throw AssertionError()
+            }
+
+        /**
+         * Returns first entry of bin for given hash.
+         */
+        private fun getFirst(hash: Int): ReferenceEntry<K, V>? {
+            // read this volatile field only once
+            val table = table.value
+            return table[hash and table.size - 1]
+        }
+
+        // Specialized implementations of map methods
+        private fun getEntry(key: K, hash: Int): ReferenceEntry<K, V>? {
+            var e = getFirst(hash)
+            while (e != null) {
+                if (e.hash != hash) {
+                    e = e.next
+                    continue
+                }
+                val entryKey = e.key
+                if (key == entryKey) {
+                    return e
+                }
+                e = e.next
+            }
+            return null
+        }
+
+        private fun getLiveEntry(key: K, hash: Int, now: Long): ReferenceEntry<K, V>? {
+            val e = getEntry(key, hash)
+            if (e == null) {
+                return null
+            } else if (map.isExpired(e, now)) {
+                tryExpireEntries(now)
+                return null
+            }
+            return e
+        }
+
+        /**
+         * Gets the value from an entry. Returns null if the entry is invalid, partially-collected,
+         * loading, or expired.
+         */
+
+        fun get(key: K, hash: Int): V? {
+            return try {
+                if (count.value != 0) { // read-volatile
+                    val now = map.ticker()
+                    val e = getLiveEntry(key, hash, now) ?: return null
+                    val value = e.valueReference?.get()
+                    if (value != null) {
+                        recordRead(e, now)
+                        return value
+                    }
+                }
+                null
+            } finally {
+                postReadCleanup()
+            }
+        }
+
+        fun getOrPut(key: K, hash: Int, defaultValue: () -> V): V {
+            reentrantLock.lock()
+            return try {
+                if (count.value != 0) { // read-volatile
+                    val now = map.ticker()
+                    val e = getLiveEntry(key, hash, now)
+                    val value = e?.valueReference?.get()
+                    if (value != null) {
+                        recordRead(e, now)
+                        return value
+                    }
+                }
+                val default = defaultValue()
+                put(key, hash, default, false)
+                default
+            } finally {
+                reentrantLock.unlock()
+                postReadCleanup()
+            }
+        }
+
+        fun put(key: K, hash: Int, value: V, onlyIfAbsent: Boolean): V? {
+            reentrantLock.lock()
+            return try {
+                val now = map.ticker()
+                preWriteCleanup(now)
+                if (count.value + 1 > threshold) { // ensure capacity
+                    expand()
+                }
+                val table = table.value
+                val index = hash and table.size - 1
+                val first = table[index]
+
+                // Look for an existing entry.
+                var e: ReferenceEntry<K, V>? = first
+                while (e != null) {
+                    val entryKey = e.key
+                    if (e.hash == hash && key == entryKey) {
+                        // We found an existing entry.
+                        val valueReference = e.valueReference
+                        val entryValue = valueReference!!.get()
+                        return when {
+                            entryValue == null -> {
+                                ++modCount
+                                val newCount = if (valueReference.isActive) {
+                                    enqueueNotification(
+                                        key,
+                                        hash,
+                                        valueReference,
+                                        RemovalCause.COLLECTED
+                                    )
+                                    setValue(e, key, value, now)
+                                    count.value // count remains unchanged
+                                } else {
+                                    setValue(e, key, value, now)
+                                    count.value + 1
+                                }
+                                count.value = newCount // write-volatile
+                                evictEntries(e)
+                                null
+                            }
+
+                            onlyIfAbsent -> {
+                                // Mimic
+                                // "if (!map.containsKey(key)) ...
+                                // else return map.get(key);
+                                recordLockedRead(e, now)
+                                entryValue
+                            }
+
+                            else -> {
+                                // clobber existing entry, count remains unchanged
+                                ++modCount
+                                enqueueNotification(
+                                    key,
+                                    hash,
+                                    valueReference,
+                                    RemovalCause.REPLACED
+                                )
+                                setValue(e, key, value, now)
+                                evictEntries(e)
+                                entryValue
+                            }
+                        }
+                    }
+                    e = e.next
+                }
+
+                // Create a new entry.
+                ++modCount
+                val newEntry = newEntry(key, hash, first)
+                setValue(newEntry, key, value, now)
+                table[index] = newEntry
+                count.plusAssign(1)
+                evictEntries(newEntry)
+                null
+            } finally {
+                reentrantLock.unlock()
+                postWriteCleanup()
+            }
+        }
+
+        fun remove(key: K, hash: Int): V? {
+            reentrantLock.lock()
+            return try {
+                val now = map.ticker()
+                preWriteCleanup(now)
+                val table = table.value
+                val index = hash and table.size - 1
+                val first = table[index]
+                var e = first
+                while (e != null) {
+                    val entryKey = e.key
+                    if (e.hash == hash && key == entryKey) {
+                        val valueReference = e.valueReference
+                        val entryValue = valueReference!!.get()
+                        val cause: RemovalCause = when {
+                            entryValue != null -> {
+                                RemovalCause.EXPLICIT
+                            }
+
+                            valueReference.isActive -> {
+                                RemovalCause.COLLECTED
+                            }
+
+                            else -> {
+                                // currently loading
+                                return null
+                            }
+                        }
+                        ++modCount
+                        val newFirst = removeValueFromChain(
+                            first!!, e, entryKey, hash, valueReference, cause
+                        )
+                        val newCount = count.value - 1
+                        table[index] = newFirst
+                        count.value = newCount // write-volatile
+                        return entryValue
+                    }
+                    e = e.next
+                }
+                null
+            } finally {
+                reentrantLock.unlock()
+                postWriteCleanup()
+            }
+        }
+
+        fun clear() {
+            if (count.value != 0) { // read-volatile
+                reentrantLock.lock()
+                try {
+                    val table = table.value
+                    for (i in 0 until table.size) {
+                        var e = table[i]
+                        while (e != null) {
+                            // Loading references aren't actually in the map yet.
+                            if (e.valueReference!!.isActive) {
+                                enqueueNotification(e, RemovalCause.EXPLICIT)
+                            }
+                            e = e.next
+                        }
+                    }
+                    for (i in 0 until table.size) {
+                        table[i] = null
+                    }
+                    writeQueue.clear()
+                    accessQueue.clear()
+                    readCount.value = 0
+                    ++modCount
+                    count.value = 0 // write-volatile
+                } finally {
+                    reentrantLock.unlock()
+                    postWriteCleanup()
+                }
+            }
+        }
+
+        /**
+         * Expands the table if possible.
+         */
+        private fun expand() {
+            val oldTable = table.value
+            val oldCapacity = oldTable.size
+            if (oldCapacity >= MAXIMUM_CAPACITY) {
+                return
+            }
+
+            /*
+             * Reclassify nodes in each list to new Map. Because we are using power-of-two expansion, the
+             * elements from each bin must either stay at same index, or move with a power of two offset.
+             * We eliminate unnecessary node creation by catching cases where old nodes can be reused
+             * because their next fields won't change. Statistically, at the default threshold, only
+             * about one-sixth of them need cloning when a table doubles. The nodes they replace will be
+             * garbage collectable as soon as they are no longer referenced by any reader thread that may
+             * be in the midst of traversing table right now.
+             */
+            var newCount = count.value
+            val newTable = SegmentTable<K, V>(oldCapacity shl 1)
+            threshold = newTable.size * 3 / 4
+            val newMask = newTable.size - 1
+            for (oldIndex in 0 until oldCapacity) {
+                // We need to guarantee that any existing reads of old Map can
+                // proceed. So we cannot yet null out each bin.
+                val head = oldTable[oldIndex] ?: continue
+
+                val next = head.next
+                val headIndex = head.hash and newMask
+
+                // Single node on list
+                if (next == null) {
+                    newTable[headIndex] = head
+                } else {
+                    // Reuse the consecutive sequence of nodes with the same target
+                    // index from the end of the list. tail points to the first
+                    // entry in the reusable list.
+                    var tail = head
+                    var tailIndex = headIndex
+                    var entry = next
+                    while (entry != null) {
+                        val newIndex = entry.hash and newMask
+                        if (newIndex != tailIndex) {
+                            // The index changed. We'll need to copy the previous entry.
+                            tailIndex = newIndex
+                            tail = entry
+                        }
+                        entry = entry.next
+                    }
+                    newTable[tailIndex] = tail
+
+                    // Clone nodes leading up to the tail.
+                    var headEntry = head
+                    while (headEntry !== tail) {
+                        val newIndex = headEntry.hash and newMask
+                        val newNext = newTable[newIndex]
+                        val newFirst = copyEntry(headEntry, newNext)
+                        if (newFirst != null) {
+                            newTable[newIndex] = newFirst
+                        } else {
+                            removeCollectedEntry(headEntry)
+                            newCount--
+                        }
+                        headEntry = headEntry.next ?: break
+                    }
+                }
+            }
+            table.value = newTable
+            count.value = newCount
+        }
+
+        private fun removeValueFromChain(
+            first: ReferenceEntry<K, V>,
+            entry: ReferenceEntry<K, V>,
+            key: K,
+            hash: Int,
+            valueReference: ValueReference<K, V>,
+            cause: RemovalCause?
+        ): ReferenceEntry<K, V>? {
+            enqueueNotification(key, hash, valueReference, cause)
+            writeQueue.remove(entry)
+            accessQueue.remove(entry)
+            return removeEntryFromChain(first, entry)
+        }
+
+        private fun removeEntryFromChain(
+            first: ReferenceEntry<K, V>,
+            entry: ReferenceEntry<K, V>
+        ): ReferenceEntry<K, V>? {
+            var newCount = count.value
+            var newFirst = entry.next
+            var e = first
+            while (e !== entry) {
+                val next = copyEntry(e, newFirst)
+                if (next != null) {
+                    newFirst = next
+                } else {
+                    removeCollectedEntry(e)
+                    newCount--
+                }
+                e = e.next ?: break
+            }
+            count.value = newCount
+            return newFirst
+        }
+
+        private fun removeCollectedEntry(entry: ReferenceEntry<K, V>) {
+            enqueueNotification(entry, RemovalCause.COLLECTED)
+            writeQueue.remove(entry)
+            accessQueue.remove(entry)
+        }
+
+        private fun removeEntry(
+            entry: ReferenceEntry<K, V>,
+            hash: Int,
+            cause: RemovalCause?
+        ): Boolean {
+            val table = table.value
+            val index = hash and table.size - 1
+            val first = table[index]
+            var e = first
+
+            while (e != null) {
+                if (e === entry) {
+                    ++modCount
+                    val newFirst = removeValueFromChain(
+                        first!!, e, e.key, hash, e.valueReference!!, cause
+                    )
+                    val newCount = count.value - 1
+                    table[index] = newFirst
+                    count.value = newCount // write-volatile
+                    return true
+                }
+                e = e.next
+            }
+            return false
+        }
+
+        /**
+         * Performs routine cleanup following a read. Normally cleanup happens during writes. If cleanup
+         * is not observed after a sufficient number of reads, try cleaning up from the read thread.
+         */
+        private fun postReadCleanup() {
+            if (readCount.incrementAndGet() and DRAIN_THRESHOLD == 0) {
+                cleanUp()
+            }
+        }
+
+        /**
+         * Performs routine cleanup prior to executing a write. This should be called every time a
+         * write thread acquires the segment lock, immediately after acquiring the lock.
+         *
+         *
+         *
+         * Post-condition: expireEntries has been run.
+         */
+        private fun preWriteCleanup(now: Long) {
+            runLockedCleanup(now)
+        }
+
+        /**
+         * Performs routine cleanup following a write.
+         */
+        private fun postWriteCleanup() {
+            runUnlockedCleanup()
+        }
+
+        fun cleanUp() {
+            val now = map.ticker()
+            runLockedCleanup(now)
+            runUnlockedCleanup()
+        }
+
+        private fun runLockedCleanup(now: Long) {
+            if (reentrantLock.tryLock()) {
+                try {
+                    expireEntries(now) // calls drainRecencyQueue
+                    readCount.value = 0
+                } finally {
+                    reentrantLock.unlock()
+                }
+            }
+        }
+
+        private fun runUnlockedCleanup() {
+            // locked cleanup may generate notifications we can send unlocked
+            /*if (!isHeldByCurrentThread) {
+                map.processPendingNotifications()
+            }*/
+        }
+
+        fun activeEntries(): Map<K, V> {
+            if (count.value != 0) { // read-volatile
+                reentrantLock.lock()
+                try {
+                    return buildMap {
+                        val table = table.value
+                        for (i in 0 until table.size) {
+                            var e = table[i]
+                            while (e != null) {
+                                if (e.valueReference!!.isActive) {
+                                    put(e.key, e.valueReference!!.get()!!)
+                                }
+                                e = e.next
+                            }
+                        }
+                    }
+                } finally {
+                    reentrantLock.unlock()
+                }
+            }
+            return emptyMap()
+        }
+
+        init {
+            threshold = initialCapacity * 3 / 4 // 0.75
+            if (!map.customWeigher && threshold.toLong() == maxSegmentWeight) {
+                // prevent spurious expansion before eviction
+                threshold++
+            }
+            table = atomic(SegmentTable(initialCapacity))
+            recencyQueue = if (map.usesAccessQueue) AtomicLinkedQueue() else discardingQueue()
+            writeQueue = if (map.usesWriteQueue) WriteQueue() else discardingQueue()
+            accessQueue = if (map.usesAccessQueue) AccessQueue() else discardingQueue()
+        }
+    }
+    // Queues
+
+    private interface Queue<T : Any> {
+        fun poll(): T?
+        fun add(value: T)
+    }
+
+    private interface MutableQueue<E : Any> : Queue<E>, Iterable<E> {
+        fun peek(): E?
+        fun isEmpty(): Boolean
+        val size: Int
+        fun clear()
+        fun remove(element: E): Boolean
+        fun contains(element: E): Boolean
+    }
+
+    private class AtomicLinkedQueue<T : Any> : Queue<T> {
+        private val head: AtomicRef<Node<T?>> = atomic(Node(null))
+        private val tail: AtomicRef<Node<T?>> = atomic(head.value)
+
+        private class Node<T>(val value: T) {
+            val next = atomic<Node<T>?>(null)
+        }
+
+        override fun add(value: T) {
+            val node: Node<T?> = Node(value)
+            tail.loop { curTail ->
+                val curNext = curTail.next.value
+                if (curNext != null) {
+                    tail.compareAndSet(curTail, curNext)
+                    return@loop
+                }
+                if (curTail.next.compareAndSet(null, node)) {
+                    tail.compareAndSet(curTail, node)
+                    return
+                }
+            }
+        }
+
+        override fun poll(): T? {
+            head.loop { curHead ->
+                val next = curHead.next.value ?: return null
+                if (head.compareAndSet(curHead, next)) return next.value
+            }
+        }
+    }
+
+    /**
+     * A custom queue for managing eviction order. Note that this is tightly integrated with `ReferenceEntry`, upon which it relies to perform its linking.
+     *
+     *
+     *
+     * Note that this entire implementation makes the assumption that all elements which are in
+     * the map are also in this queue, and that all elements not in the queue are not in the map.
+     *
+     *
+     *
+     * The benefits of creating our own queue are that (1) we can replace elements in the middle
+     * of the queue as part of copyWriteEntry, and (2) the contains method is highly optimized
+     * for the current model.
+     */
+
+    private class WriteQueue<K : Any, V : Any> : MutableQueue<ReferenceEntry<K, V>> {
+        private val head: ReferenceEntry<K, V> = object : ReferenceEntry<K, V> {
+            override var writeTime: Long
+                get() = Long.MAX_VALUE
+                set(_) {}
+            override var nextInWriteQueue: ReferenceEntry<K, V> = this
+            override var previousInWriteQueue: ReferenceEntry<K, V> = this
+        }
+
+        // implements Queue
+        override fun add(value: ReferenceEntry<K, V>) {
+            // unlink
+            connectWriteOrder(value.previousInWriteQueue, value.nextInWriteQueue)
+
+            // add to tail
+            connectWriteOrder(head.previousInWriteQueue, value)
+            connectWriteOrder(value, head)
+        }
+
+        override fun peek(): ReferenceEntry<K, V>? {
+            val next = head.nextInWriteQueue
+            return if (next === head) null else next
+        }
+
+        override fun poll(): ReferenceEntry<K, V>? {
+            val next = head.nextInWriteQueue
+            if (next === head) {
+                return null
+            }
+            remove(next)
+            return next
+        }
+
+        override fun remove(element: ReferenceEntry<K, V>): Boolean {
+            val previous = element.previousInWriteQueue
+            val next = element.nextInWriteQueue
+            connectWriteOrder(previous, next)
+            nullifyWriteOrder(element)
+            return next !== NullEntry
+        }
+
+        override fun contains(element: ReferenceEntry<K, V>): Boolean =
+            element.nextInWriteQueue !== NullEntry
+
+        override fun isEmpty(): Boolean =
+            head.nextInWriteQueue === head
+
+        override val size: Int
+            get() {
+                var size = 0
+                var e = head.nextInWriteQueue
+                while (e !== head) {
+                    size++
+                    e = e.nextInWriteQueue
+                }
+                return size
+            }
+
+        override fun clear() {
+            var e = head.nextInWriteQueue
+            while (e !== head) {
+                val next = e.nextInWriteQueue
+                nullifyWriteOrder(e)
+                e = next
+            }
+            head.nextInWriteQueue = head
+            head.previousInWriteQueue = head
+        }
+
+        override fun iterator(): Iterator<ReferenceEntry<K, V>> = iterator {
+            var value = peek()
+            while (value != null) {
+                yield(value)
+                val next = value.nextInWriteQueue
+                value = if (next === head) null else next
+            }
+        }
+    }
+
+    /**
+     * A custom queue for managing access order. Note that this is tightly integrated with
+     * `ReferenceEntry`, upon which it reliese to perform its linking.
+     *
+     *
+     *
+     * Note that this entire implementation makes the assumption that all elements which are in
+     * the map are also in this queue, and that all elements not in the queue are not in the map.
+     *
+     *
+     *
+     * The benefits of creating our own queue are that (1) we can replace elements in the middle
+     * of the queue as part of copyWriteEntry, and (2) the contains method is highly optimized
+     * for the current model.
+     */
+    private class AccessQueue<K : Any, V : Any> : MutableQueue<ReferenceEntry<K, V>> {
+        private val head: ReferenceEntry<K, V> = object : ReferenceEntry<K, V> {
+            override var accessTime: Long
+                get() = Long.MAX_VALUE
+                set(_) {}
+            override var nextInAccessQueue: ReferenceEntry<K, V> = this
+            override var previousInAccessQueue: ReferenceEntry<K, V> = this
+        }
+
+        // implements Queue
+        override fun add(value: ReferenceEntry<K, V>) {
+            // unlink
+            connectAccessOrder(value.previousInAccessQueue, value.nextInAccessQueue)
+
+            // add to tail
+            connectAccessOrder(head.previousInAccessQueue, value)
+            connectAccessOrder(value, head)
+        }
+
+        override fun peek(): ReferenceEntry<K, V>? {
+            val next = head.nextInAccessQueue
+            return if (next === head) null else next
+        }
+
+        override fun poll(): ReferenceEntry<K, V>? {
+            val next = head.nextInAccessQueue
+            if (next === head) {
+                return null
+            }
+            remove(next)
+            return next
+        }
+
+        override fun remove(element: ReferenceEntry<K, V>): Boolean {
+            val previous = element.previousInAccessQueue
+            val next = element.nextInAccessQueue
+            connectAccessOrder(previous, next)
+            nullifyAccessOrder(element)
+            return next !== NullEntry
+        }
+
+        override fun contains(element: ReferenceEntry<K, V>): Boolean =
+            element.nextInAccessQueue !== NullEntry
+
+        override fun isEmpty(): Boolean =
+            head.nextInAccessQueue === head
+
+        override val size: Int
+            get() {
+                var size = 0
+                var e = head.nextInAccessQueue
+                while (e !== head) {
+                    size++
+                    e = e.nextInAccessQueue
+                }
+                return size
+            }
+
+        override fun clear() {
+            var e = head.nextInAccessQueue
+            while (e !== head) {
+                val next = e.nextInAccessQueue
+                nullifyAccessOrder(e)
+                e = next
+            }
+            head.nextInAccessQueue = head
+            head.previousInAccessQueue = head
+        }
+
+        override fun iterator(): Iterator<ReferenceEntry<K, V>> = iterator {
+            var value = peek()
+            while (value != null) {
+                yield(value)
+                val next = value.nextInAccessQueue
+                value = if (next === head) null else next
+            }
+        }
+    }
+
+    // Cache support
+    fun cleanUp() {
+        for (segment in segments) {
+            segment?.cleanUp()
+        }
+    }
+
+    // ConcurrentMap methods
+    fun getIfPresent(key: K): V? {
+        val hash = hash(key)
+        return segmentFor(hash).get(key, hash)
+    }
+
+    fun put(key: K, value: V): V? {
+        val hash = hash(key)
+        return segmentFor(hash).put(key, hash, value, false)
+    }
+
+    fun getOrPut(key: K, defaultValue: () -> V): V {
+        val hash = hash(key)
+        return segmentFor(hash).getOrPut(key, hash, defaultValue)
+    }
+
+    fun clear() {
+        for (segment in segments) {
+            segment?.clear()
+        }
+    }
+
+    fun remove(key: K): V? {
+        val hash = hash(key)
+        return segmentFor(hash).remove(key, hash)
+    }
+
+    fun getAllPresent(): Map<K, V> {
+        return buildMap {
+            for (segment in segments) {
+                segment?.let { putAll(it.activeEntries()) }
+            }
+        }
+    }
+
+    // Serialization Support
+    internal class LocalManualCache<K : Any, V : Any> private constructor(private val localCache: LocalCache<K, V>) :
+        Cache<K, V> {
+        constructor(builder: CacheBuilder<K, V>) : this(LocalCache<K, V>(builder))
+
+        // Cache methods
+        override fun getIfPresent(key: K): V? {
+            return localCache.getIfPresent(key)
+        }
+
+        override fun put(key: K, value: V) {
+            localCache.put(key, value)
+        }
+
+        override fun invalidate(key: K) {
+            localCache.remove(key)
+        }
+
+        override fun getOrPut(key: K, valueProducer: () -> V): V {
+            return localCache.getOrPut(key, valueProducer)
+        }
+
+        override fun getAllPresent(keys: List<*>): Map<K, V> {
+            return localCache.getAllPresent().filterKeys { it in keys }
+        }
+
+        override fun getAllPresent(): Map<K, V> {
+            return localCache.getAllPresent()
+        }
+
+        override fun invalidateAll(keys: List<K>) {
+            TODO("Not yet implemented")
+        }
+
+        override fun putAll(map: Map<K, V>) {
+            TODO("Not yet implemented")
+        }
+
+        override fun invalidateAll() {
+            localCache.clear()
+        }
+
+        override fun size(): Long {
+            TODO("Not yet implemented")
+        }
+    }
+
+    companion object {
+        /*
+         * The basic strategy is to subdivide the table among Segments, each of which itself is a
+         * concurrently readable hash table. The map supports non-blocking reads and concurrent writes
+         * across different segments.
+         *
+         * If a maximum size is specified, a best-effort bounding is performed per segment, using a
+         * page-replacement algorithm to determine which entries to evict when the capacity has been
+         * exceeded.
+         *
+         * The page replacement algorithm's data structures are kept casually consistent with the map. The
+         * ordering of writes to a segment is sequentially consistent. An update to the map and recording
+         * of reads may not be immediately reflected on the algorithm's data structures. These structures
+         * are guarded by a lock and operations are applied in batches to avoid lock contention. The
+         * penalty of applying the batches is spread across threads so that the amortized cost is slightly
+         * higher than performing just the operation without enforcing the capacity constraint.
+         *
+         * This implementation uses a per-segment queue to record a memento of the additions, removals,
+         * and accesses that were performed on the map. The queue is drained on writes and when it exceeds
+         * its capacity threshold.
+         *
+         * The Least Recently Used page replacement algorithm was chosen due to its simplicity, high hit
+         * rate, and ability to be implemented with O(1) time complexity. The initial LRU implementation
+         * operates per-segment rather than globally for increased implementation simplicity. We expect
+         * the cache hit rate to be similar to that of a global LRU algorithm.
+         */
+        // Constants
+        private val OneWeigher: Weigher<Any, Any> = { _, _ -> 1 }
+
+        /**
+         * The maximum capacity, used if a higher value is implicitly specified by either of the
+         * constructors with arguments. MUST be a power of two <= 1<<30 to ensure that entries are
+         * indexable using ints.
+         */
+        const val MAXIMUM_CAPACITY = 1 shl 30
+
+        /**
+         * The maximum number of segments to allow; used to bound constructor arguments.
+         */
+        const val MAX_SEGMENTS = 1 shl 16 // slightly conservative
+
+        /**
+         * Number of cache access operations that can be buffered per segment before the cache's recency
+         * ordering information is updated. This is used to avoid lock contention by recording a memento
+         * of reads and delaying a lock acquisition until the threshold is crossed or a mutation occurs.
+         *
+         *
+         *
+         * This must be a (2^n)-1 as it is used as a mask.
+         */
+        const val DRAIN_THRESHOLD = 0x3F
+
+        /**
+         * Placeholder. Indicates that the value hasn't been set yet.
+         */
+        private val UNSET: ValueReference<Any, Any> = object : ValueReference<Any, Any> {
+            override fun get(): Any? {
+                return null
+            }
+
+            override val weight: Int
+                get() = 0
+            override val entry: ReferenceEntry<Any, Any>?
+                get() = null
+
+            override fun copyFor(
+                value: Any?,
+                entry: ReferenceEntry<Any, Any>?
+            ): ValueReference<Any, Any> {
+                return this
+            }
+
+            override val isActive: Boolean
+                get() = false
+
+            override fun notifyNewValue(newValue: Any) {}
+        }
+
+        /**
+         * Singleton placeholder that indicates a value is being loaded.
+         */
+        @Suppress("UNCHECKED_CAST")
+        private fun <K : Any, V : Any> unset() = UNSET as ValueReference<K, V>
+
+        @Suppress("UNCHECKED_CAST")
+        private fun <K : Any, V : Any> nullEntry() = NullEntry as ReferenceEntry<K, V>
+
+        private val DISCARDING_QUEUE: MutableQueue<Any> = object : MutableQueue<Any> {
+            override fun add(value: Any) {}
+
+            override fun peek(): Any? = null
+
+            override fun poll(): Any? = null
+
+            override fun iterator(): MutableIterator<Any> = HashSet<Any>().iterator()
+
+            override val size: Int = 0
+
+            override fun isEmpty(): Boolean = true
+
+            override fun clear() {}
+
+            override fun remove(element: Any): Boolean = false
+
+            override fun contains(element: Any): Boolean = false
+        }
+
+        /**
+         * Queue that discards all elements.
+         */
+        @Suppress("UNCHECKED_CAST")
+        private fun <E : Any> discardingQueue(): MutableQueue<E> =
+            DISCARDING_QUEUE as MutableQueue<E>
+
+        /**
+         * Applies a supplemental hash function to a given hash code, which defends against poor quality
+         * hash functions. This is critical when the concurrent hash map uses power-of-two length hash
+         * tables, that otherwise encounter collisions for hash codes that do not differ in lower or
+         * upper bits.
+         *
+         * @param hash hash code
+         */
+        fun rehash(hash: Int): Int {
+            // Spread bits to regularize both segment and index locations,
+            // using variant of single-word Wang/Jenkins hash.
+            // TODO(kevinb): use Hashing/move this to Hashing?
+            var h = hash
+            h += h shl 15 xor -0x3283
+            h = h xor (h ushr 10)
+            h += h shl 3
+            h = h xor (h ushr 6)
+            h += (h shl 2) + (h shl 14)
+            return h xor (h ushr 16)
+        }
+
+        // queues
+        // Guarded By Segment.this
+        private fun <K : Any, V : Any> connectAccessOrder(
+            previous: ReferenceEntry<K, V>,
+            next: ReferenceEntry<K, V>
+        ) {
+            previous.nextInAccessQueue = next
+            next.previousInAccessQueue = previous
+        }
+
+        // Guarded By Segment.this
+        private fun <K : Any, V : Any> nullifyAccessOrder(nulled: ReferenceEntry<K, V>) {
+            val nullEntry: ReferenceEntry<K, V> = nullEntry()
+            nulled.nextInAccessQueue = nullEntry
+            nulled.previousInAccessQueue = nullEntry
+        }
+
+        // Guarded By Segment.this
+        private fun <K : Any, V : Any> connectWriteOrder(
+            previous: ReferenceEntry<K, V>,
+            next: ReferenceEntry<K, V>
+        ) {
+            previous.nextInWriteQueue = next
+            next.previousInWriteQueue = previous
+        }
+
+        // Guarded By Segment.this
+        private fun <K : Any, V : Any> nullifyWriteOrder(nulled: ReferenceEntry<K, V>) {
+            val nullEntry: ReferenceEntry<K, V> = nullEntry()
+            nulled.nextInWriteQueue = nullEntry
+            nulled.previousInWriteQueue = nullEntry
+        }
+    }
+
+    /**
+     * Creates a new, empty map with the specified strategy, initial capacity and concurrency level.
+     */
+    init {
+        this.maxWeight = when {
+            builder.expireAfterAccess == Duration.ZERO || builder.expireAfterWrite == Duration.ZERO -> 0L
+            builder.weigher != null -> builder.maximumWeight
+            else -> builder.maximumSize
+        }
+        this.weigher = builder.weigher ?: OneWeigher as Weigher<K, V>
+
+        this.expireAfterAccessNanos =
+            (if (builder.expireAfterAccess == Duration.INFINITE) Duration.ZERO else builder.expireAfterAccess)
+                .inWholeNanoseconds
+
+        this.expireAfterWriteNanos =
+            (if (builder.expireAfterWrite == Duration.INFINITE) Duration.ZERO else builder.expireAfterWrite)
+                .inWholeNanoseconds
+
+        this.ticker = if (recordsTime) (builder.ticker ?: MonotonicTicker) else ({ 0L })
+        this.entryFactory = EntryFactory.getFactory(usesAccessEntries, usesWriteEntries)
+        var initialCapacity = builder.initialCapacity.coerceAtMost(MAXIMUM_CAPACITY)
+        if (evictsBySize && !customWeigher) {
+            initialCapacity = min(initialCapacity, maxWeight.toInt())
+        }
+        val concurrencyLevel = builder.concurrencyLevel.coerceAtMost(MAX_SEGMENTS)
+        // Find the lowest power-of-two segmentCount that exceeds concurrencyLevel, unless
+        // maximumSize/Weight is specified in which case ensure that each segment gets at least 10
+        // entries. The special casing for size-based eviction is only necessary because that eviction
+        // happens per segment instead of globally, so too many segments compared to the maximum size
+        // will result in random eviction behavior.
+        var segmentShift = 0
+        var segmentCount = 1
+        while (segmentCount < concurrencyLevel && (!evictsBySize || segmentCount * 20 <= maxWeight)) {
+            ++segmentShift
+            segmentCount = segmentCount shl 1
+        }
+        this.segmentShift = 32 - segmentShift
+        segmentMask = segmentCount - 1
+        segments = arrayOfNulls(segmentCount)
+        var segmentCapacity = initialCapacity / segmentCount
+        if (segmentCapacity * segmentCount < initialCapacity) {
+            ++segmentCapacity
+        }
+        var segmentSize = 1
+        while (segmentSize < segmentCapacity) {
+            segmentSize = segmentSize shl 1
+        }
+        if (evictsBySize) {
+            // Ensure sum of segment max weights = overall max weights
+            var maxSegmentWeight = maxWeight / segmentCount + 1
+            val remainder = maxWeight % segmentCount
+            for (i in segments.indices) {
+                if (i.toLong() == remainder) {
+                    maxSegmentWeight--
+                }
+                segments[i] = createSegment(segmentSize, maxSegmentWeight)
+            }
+        } else {
+            for (i in segments.indices) {
+                segments[i] = createSegment(segmentSize, -1)
+            }
+        }
+    }
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/MonotonicTicker.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/MonotonicTicker.kt
new file mode 100644
index 0000000000..0d2e3b9039
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/MonotonicTicker.kt
@@ -0,0 +1,27 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+import kotlin.time.ExperimentalTime
+import kotlin.time.TimeSource
+
+@OptIn(ExperimentalTime::class)
+internal val MonotonicTicker: Ticker = TimeSource.Monotonic.markNow().let { timeMark -> { timeMark.elapsedNow().inWholeNanoseconds } }
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/RemovalCause.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/RemovalCause.kt
new file mode 100644
index 0000000000..ddf5f10cb0
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/RemovalCause.kt
@@ -0,0 +1,35 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+/**
+ * The reason why a cached entry was removed.
+ * @param wasEvicted True if entry removal was automatic due to eviction. That is, the cause of removal is neither [EXPLICIT] or [REPLACED].
+ * @author Charles Fry
+ * @since 10.0
+ */
+internal enum class RemovalCause(val wasEvicted: Boolean) {
+    EXPLICIT(false),
+    REPLACED(false),
+    COLLECTED(true),
+    EXPIRED(true),
+    SIZE(true);
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Ticker.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Ticker.kt
new file mode 100644
index 0000000000..7a51f5f029
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Ticker.kt
@@ -0,0 +1,26 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+/**
+ * @return Number of nanoseconds elapsed since the ticker's fixed point of reference.
+ */
+typealias Ticker = () -> Long
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Weigher.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Weigher.kt
new file mode 100644
index 0000000000..a553733b53
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/store/Weigher.kt
@@ -0,0 +1,26 @@
+/*
+ * Taken from the Mobile Native Foundation Store project.
+ * https://github.com/MobileNativeFoundation/Store/commit/e25e3c130187d9294ad5b998136b0498bd91d88f
+ *
+ * Copyright (c) 2017 The New York Times Company
+ *
+ * Copyright (C) 2009 The Guava Authors
+ *
+ * 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 com.apollographql.apollo3.cache.normalized.api.internal.store
+
+/**
+ * @return Weight of a cache entry. Must be non-negative. There is no unit for entry weights. Rather, they are simply relative to each other.
+ */
+typealias Weigher <Key, Value> = (key: Key, value: Value) -> Int
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/jsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.js.kt b/libraries/apollo-normalized-cache-api-incubating/src/jsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.js.kt
new file mode 100644
index 0000000000..367c2a471e
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/jsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.js.kt
@@ -0,0 +1,3 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+internal actual fun <K, V> ConcurrentMap(): MutableMap<K, V> = mutableMapOf()
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/jvmMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.jvm.kt b/libraries/apollo-normalized-cache-api-incubating/src/jvmMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.jvm.kt
new file mode 100644
index 0000000000..47295f8003
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/jvmMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.jvm.kt
@@ -0,0 +1,5 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+import java.util.concurrent.ConcurrentHashMap
+
+internal actual fun <K, V> ConcurrentMap(): MutableMap<K, V> = ConcurrentHashMap()
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/wasmJsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.wasmJs.kt b/libraries/apollo-normalized-cache-api-incubating/src/wasmJsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.wasmJs.kt
new file mode 100644
index 0000000000..367c2a471e
--- /dev/null
+++ b/libraries/apollo-normalized-cache-api-incubating/src/wasmJsMain/kotlin/com/apollographql/apollo3/cache/normalized/api/internal/ConcurrentMap.wasmJs.kt
@@ -0,0 +1,3 @@
+package com.apollographql.apollo3.cache.normalized.api.internal
+
+internal actual fun <K, V> ConcurrentMap(): MutableMap<K, V> = mutableMapOf()
diff --git a/libraries/apollo-normalized-cache-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt b/libraries/apollo-normalized-cache-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt
index eb945f21f6..0e137a3db1 100644
--- a/libraries/apollo-normalized-cache-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt
+++ b/libraries/apollo-normalized-cache-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/internal/DefaultApolloStore.kt
@@ -18,7 +18,7 @@ import com.apollographql.apollo3.cache.normalized.api.NormalizedCacheFactory
 import com.apollographql.apollo3.cache.normalized.api.ReadOnlyNormalizedCache
 import com.apollographql.apollo3.cache.normalized.api.Record
 import com.apollographql.apollo3.cache.normalized.api.RecordMerger
-import com.apollographql.apollo3.cache.normalized.api.internal.OptimisticCache
+import com.apollographql.apollo3.cache.normalized.api.internal.OptimisticNormalizedCache
 import com.apollographql.apollo3.cache.normalized.api.normalize
 import com.apollographql.apollo3.cache.normalized.api.readDataFromCacheInternal
 import com.apollographql.apollo3.cache.normalized.api.toData
@@ -47,8 +47,8 @@ internal class DefaultApolloStore(
   override val changedKeys = changedKeysEvents.asSharedFlow()
 
   // Keeping this as lazy to avoid accessing the disk at initialization which usually happens on the main thread
-  private val cache: OptimisticCache by lazy {
-    OptimisticCache().chain(normalizedCacheFactory.createChain()) as OptimisticCache
+  private val cache: OptimisticNormalizedCache by lazy {
+    OptimisticNormalizedCache(normalizedCacheFactory.create())
   }
 
   override fun publish(keys: Set<String>) {
diff --git a/libraries/apollo-normalized-cache-sqlite-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCache.kt b/libraries/apollo-normalized-cache-sqlite-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCache.kt
index 07d8448da8..6c98a76bb7 100644
--- a/libraries/apollo-normalized-cache-sqlite-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCache.kt
+++ b/libraries/apollo-normalized-cache-sqlite-incubating/src/commonMain/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCache.kt
@@ -5,29 +5,16 @@ import com.apollographql.apollo3.cache.normalized.api.ApolloCacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.ApolloCacheHeaders.EVICT_AFTER_READ
 import com.apollographql.apollo3.cache.normalized.api.CacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheKey
-import com.apollographql.apollo3.cache.normalized.api.DefaultRecordMerger
 import com.apollographql.apollo3.cache.normalized.api.NormalizedCache
 import com.apollographql.apollo3.cache.normalized.api.Record
 import com.apollographql.apollo3.cache.normalized.api.RecordMerger
-import com.apollographql.apollo3.cache.normalized.api.internal.Lock
 import com.apollographql.apollo3.cache.normalized.sql.internal.RecordDatabase
 import com.apollographql.apollo3.exception.apolloExceptionHandler
 import kotlin.reflect.KClass
 
 class SqlNormalizedCache internal constructor(
     private val recordDatabase: RecordDatabase,
-) : NormalizedCache() {
-
-  // A lock is only needed if there is a nextCache
-  private val lock = nextCache?.let { Lock() }
-
-  private fun <T> lockWrite(block: () -> T): T {
-    return lock?.write { block() } ?: block()
-  }
-
-  private fun <T> lockRead(block: () -> T): T {
-    return lock?.read { block() } ?: block()
-  }
+) : NormalizedCache {
 
   private fun <T> maybeTransaction(condition: Boolean, block: () -> T): T {
     return if (condition) {
@@ -41,77 +28,57 @@ class SqlNormalizedCache internal constructor(
 
   override fun loadRecord(key: String, cacheHeaders: CacheHeaders): Record? {
     val evictAfterRead = cacheHeaders.hasHeader(EVICT_AFTER_READ)
-    return lockWrite {
-      maybeTransaction(evictAfterRead) {
-        try {
-          recordDatabase.select(key)
-        } catch (e: Exception) {
-          // Unable to read the record from the database, it is possibly corrupted - treat this as a cache miss
-          apolloExceptionHandler(Exception("Unable to read a record from the database", e))
-          null
-        }?.also {
-          if (evictAfterRead) {
-            recordDatabase.delete(key)
-          }
+    return maybeTransaction(evictAfterRead) {
+      try {
+        recordDatabase.select(key)
+      } catch (e: Exception) {
+        // Unable to read the record from the database, it is possibly corrupted - treat this as a cache miss
+        apolloExceptionHandler(Exception("Unable to read a record from the database", e))
+        null
+      }?.also {
+        if (evictAfterRead) {
+          recordDatabase.delete(key)
         }
-      } ?: nextCache?.loadRecord(key, cacheHeaders)
+      }
     }
   }
 
   override fun loadRecords(keys: Collection<String>, cacheHeaders: CacheHeaders): Collection<Record> {
     val evictAfterRead = cacheHeaders.hasHeader(EVICT_AFTER_READ)
-    return lockWrite {
-      val records = maybeTransaction(evictAfterRead) {
-        try {
-          internalGetRecords(keys)
-        } catch (e: Exception) {
-          // Unable to read the records from the database, it is possibly corrupted - treat this as a cache miss
-          apolloExceptionHandler(Exception("Unable to read records from the database", e))
-          emptyList()
-        }.also {
-          if (evictAfterRead) {
-            it.forEach { record ->
-              recordDatabase.delete(record.key)
-            }
+    return maybeTransaction(evictAfterRead) {
+      try {
+        internalGetRecords(keys)
+      } catch (e: Exception) {
+        // Unable to read the records from the database, it is possibly corrupted - treat this as a cache miss
+        apolloExceptionHandler(Exception("Unable to read records from the database", e))
+        emptyList()
+      }.also {
+        if (evictAfterRead) {
+          it.forEach { record ->
+            recordDatabase.delete(record.key)
           }
         }
       }
-      val missRecordKeys = keys - records.map { it.key }.toSet()
-      val missRecords = missRecordKeys.ifEmpty { null }?.let { nextCache?.loadRecords(it, cacheHeaders) }.orEmpty()
-      records + missRecords
     }
   }
 
   override fun clearAll() {
-    lockWrite {
-      nextCache?.clearAll()
-      recordDatabase.deleteAll()
-    }
+    recordDatabase.deleteAll()
   }
 
   override fun remove(cacheKey: CacheKey, cascade: Boolean): Boolean {
-    return lockWrite {
-      val selfRemoved = recordDatabase.transaction {
-        internalDeleteRecord(
-            key = cacheKey.key,
-            cascade = cascade,
-        )
-      }
-      val chainRemoved = nextCache?.remove(cacheKey, cascade) ?: false
-      selfRemoved || chainRemoved
+    return recordDatabase.transaction {
+      internalDeleteRecord(
+          key = cacheKey.key,
+          cascade = cascade,
+      )
     }
   }
 
   override fun remove(pattern: String): Int {
-    return lockWrite {
-      var selfRemoved = 0
-      recordDatabase.transaction {
-        recordDatabase.deleteMatching(pattern)
-        selfRemoved = recordDatabase.changes().toInt()
-      }
-      val chainRemoved = nextCache?.remove(pattern) ?: 0
-
-      selfRemoved + chainRemoved
+    return recordDatabase.transaction {
+      recordDatabase.deleteMatching(pattern)
+      recordDatabase.changes().toInt()
     }
   }
 
@@ -119,27 +86,17 @@ class SqlNormalizedCache internal constructor(
     return headerValue(ApolloCacheHeaders.DATE)?.toLong()
   }
 
-  override fun merge(record: Record, cacheHeaders: CacheHeaders): Set<String> {
-    return merge(record, cacheHeaders, DefaultRecordMerger)
-  }
-
-  override fun merge(records: Collection<Record>, cacheHeaders: CacheHeaders): Set<String> {
-    return merge(records, cacheHeaders, DefaultRecordMerger)
-  }
-
   @ApolloExperimental
   override fun merge(record: Record, cacheHeaders: CacheHeaders, recordMerger: RecordMerger): Set<String> {
     if (cacheHeaders.hasHeader(ApolloCacheHeaders.DO_NOT_STORE)) {
       return emptySet()
     }
-    return lockWrite {
-      try {
-        internalUpdateRecord(record = record, recordMerger = recordMerger, date = cacheHeaders.date())
-      } catch (e: Exception) {
-        // Unable to merge the record in the database, it is possibly corrupted - treat this as a cache miss
-        apolloExceptionHandler(Exception("Unable to merge a record from the database", e))
-        emptySet()
-      } + nextCache?.merge(record, cacheHeaders).orEmpty()
+    return try {
+      internalUpdateRecord(record = record, recordMerger = recordMerger, date = cacheHeaders.date())
+    } catch (e: Exception) {
+      // Unable to merge the record in the database, it is possibly corrupted - treat this as a cache miss
+      apolloExceptionHandler(Exception("Unable to merge a record from the database", e))
+      emptySet()
     }
   }
 
@@ -148,21 +105,17 @@ class SqlNormalizedCache internal constructor(
     if (cacheHeaders.hasHeader(ApolloCacheHeaders.DO_NOT_STORE)) {
       return emptySet()
     }
-    return lockWrite {
-      try {
-        internalUpdateRecords(records = records, recordMerger = recordMerger, date = cacheHeaders.date())
-      } catch (e: Exception) {
-        // Unable to merge the records in the database, it is possibly corrupted - treat this as a cache miss
-        apolloExceptionHandler(Exception("Unable to merge records from the database", e))
-        emptySet()
-      } + nextCache?.merge(records, cacheHeaders).orEmpty()
+    return try {
+      internalUpdateRecords(records = records, recordMerger = recordMerger, date = cacheHeaders.date())
+    } catch (e: Exception) {
+      // Unable to merge the records in the database, it is possibly corrupted - treat this as a cache miss
+      apolloExceptionHandler(Exception("Unable to merge records from the database", e))
+      emptySet()
     }
   }
 
   override fun dump(): Map<KClass<*>, Map<String, Record>> {
-    return lockRead {
-      mapOf(this::class to recordDatabase.selectAll().associateBy { it.key }) + nextCache?.dump().orEmpty()
-    }
+    return mapOf(this::class to recordDatabase.selectAll().associateBy { it.key })
   }
 
   /**
@@ -223,7 +176,6 @@ class SqlNormalizedCache internal constructor(
         date = fields.mapValues { date },
         metadata = metadata
     )
-
   }
 
   /**
diff --git a/libraries/apollo-normalized-cache/build.gradle.kts b/libraries/apollo-normalized-cache/build.gradle.kts
index ea68162d75..02c7976eba 100644
--- a/libraries/apollo-normalized-cache/build.gradle.kts
+++ b/libraries/apollo-normalized-cache/build.gradle.kts
@@ -14,7 +14,7 @@ kotlin {
         api(project(":apollo-runtime"))
         api(project(":apollo-normalized-cache-api"))
         api(libs.kotlinx.coroutines)
-        implementation(libs.atomicfu.get().toString()) {
+        implementation(libs.atomicfu.library.get().toString()) {
           because("Use of ReentrantLock in DefaultApolloStore for Apple (we don't use the gradle plugin rewrite)")
         }
       }
diff --git a/libraries/apollo-runtime-java/api/apollo-runtime-java.api b/libraries/apollo-runtime-java/api/apollo-runtime-java.api
index 25642e752b..9c2f657eff 100644
--- a/libraries/apollo-runtime-java/api/apollo-runtime-java.api
+++ b/libraries/apollo-runtime-java/api/apollo-runtime-java.api
@@ -45,6 +45,7 @@ public class com/apollographql/apollo3/runtime/java/ApolloClient$Builder : com/a
 	public fun getExecutionContext ()Lcom/apollographql/apollo3/api/ExecutionContext;
 	public fun getHttpHeaders ()Ljava/util/List;
 	public fun getHttpMethod ()Lcom/apollographql/apollo3/api/http/HttpMethod;
+	public fun getRetryOnError ()Ljava/lang/Boolean;
 	public fun getSendApqExtensions ()Ljava/lang/Boolean;
 	public fun getSendDocument ()Ljava/lang/Boolean;
 	public fun httpBatching (JIZ)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
@@ -59,6 +60,8 @@ public class com/apollographql/apollo3/runtime/java/ApolloClient$Builder : com/a
 	public fun interceptors (Ljava/util/List;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
 	public fun networkTransport (Lcom/apollographql/apollo3/runtime/java/network/NetworkTransport;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
 	public fun okHttpClient (Lokhttp3/OkHttpClient;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
+	public fun retryOnError (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
+	public synthetic fun retryOnError (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendApqExtensions (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
 	public synthetic fun sendApqExtensions (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendDocument (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/runtime/java/ApolloClient$Builder;
diff --git a/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/ApolloClient.java b/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/ApolloClient.java
index 9a39d60cc8..b607d3d11d 100644
--- a/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/ApolloClient.java
+++ b/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/ApolloClient.java
@@ -188,6 +188,7 @@ public class ApolloClient implements Closeable {
     private Boolean enableAutoPersistedQueries;
     private Boolean canBeBatched;
     private Boolean httpExposeErrorBody;
+    private Boolean retryOnError;
 
     public Builder() {
     }
@@ -196,16 +197,16 @@ public class ApolloClient implements Closeable {
      * The url of the GraphQL server used for HTTP. This is the same as {@link #httpServerUrl(String)}. See also
      * {@link #networkTransport(NetworkTransport)} for more customization
      */
-    public Builder serverUrl(@NotNull String serverUrl) {
-      this.httpServerUrl = checkNotNull(serverUrl, "serverUrl is null");
+    public Builder serverUrl(String serverUrl) {
+      this.httpServerUrl = serverUrl;
       return this;
     }
 
     /**
      * The url of the GraphQL server used for HTTP. See also {@link #networkTransport(NetworkTransport)} for more customization
      */
-    public Builder httpServerUrl(@NotNull String httpServerUrl) {
-      this.httpServerUrl = checkNotNull(httpServerUrl, "httpServerUrl is null");
+    public Builder httpServerUrl(String httpServerUrl) {
+      this.httpServerUrl = httpServerUrl;
       return this;
     }
 
@@ -213,8 +214,8 @@ public class ApolloClient implements Closeable {
      * The url of the GraphQL server used for WebSockets. See also {@link #subscriptionNetworkTransport(NetworkTransport)} for more
      * customization
      */
-    public Builder webSocketServerUrl(@NotNull String webSocketServerUrl) {
-      this.webSocketServerUrl = checkNotNull(webSocketServerUrl, "webSocketServerUrl is null");
+    public Builder webSocketServerUrl(String webSocketServerUrl) {
+      this.webSocketServerUrl = webSocketServerUrl;
       return this;
     }
 
@@ -224,8 +225,8 @@ public class ApolloClient implements Closeable {
      * @param okHttpClient the client to use.
      * @return The {@link Builder} object to be used for chaining method calls
      */
-    public Builder okHttpClient(@NotNull OkHttpClient okHttpClient) {
-      this.callFactory = checkNotNull(okHttpClient, "okHttpClient is null");
+    public Builder okHttpClient(OkHttpClient okHttpClient) {
+      this.callFactory = okHttpClient;
       this.webSocketFactory = okHttpClient;
       return null;
     }
@@ -234,8 +235,8 @@ public class ApolloClient implements Closeable {
      * Set the custom call factory for creating {@link Call} instances. <p> Note: Calling {@link #okHttpClient(OkHttpClient)} automatically
      * sets this value.
      */
-    public Builder callFactory(@NotNull Call.Factory factory) {
-      this.callFactory = checkNotNull(factory, "factory is null");
+    public Builder callFactory(Call.Factory factory) {
+      this.callFactory = factory;
       return this;
     }
 
@@ -243,8 +244,8 @@ public class ApolloClient implements Closeable {
      * Set the custom call factory for creating {@link WebSocket} instances. <p> Note: Calling {@link #okHttpClient(OkHttpClient)}
      * automatically sets this value.
      */
-    public Builder webSocketFactory(@NotNull WebSocket.Factory factory) {
-      this.webSocketFactory = checkNotNull(factory, "factory is null");
+    public Builder webSocketFactory(WebSocket.Factory factory) {
+      this.webSocketFactory = factory;
       return this;
     }
 
@@ -253,13 +254,13 @@ public class ApolloClient implements Closeable {
      *
      * @return The {@link Builder} object to be used for chaining method calls
      */
-    public Builder dispatcher(@NotNull Executor dispatcher) {
-      this.executor = checkNotNull(dispatcher, "dispatcher is null");
+    public Builder dispatcher(Executor dispatcher) {
+      this.executor = dispatcher;
       return this;
     }
 
-    public Builder addInterceptor(@NotNull ApolloInterceptor interceptor) {
-      this.interceptors.add(checkNotNull(interceptor, "interceptor is null"));
+    public Builder addInterceptor(ApolloInterceptor interceptor) {
+      this.interceptors.add(interceptor);
       return this;
     }
 
@@ -306,8 +307,8 @@ public class ApolloClient implements Closeable {
       return this;
     }
 
-    public Builder wsProtocolFactory(@NotNull WsProtocol.Factory wsProtocolFactory) {
-      this.wsProtocolFactory = checkNotNull(wsProtocolFactory, "wsProtocolFactory is null");
+    public Builder wsProtocolFactory(WsProtocol.Factory wsProtocolFactory) {
+      this.wsProtocolFactory = wsProtocolFactory;
       return this;
     }
 
@@ -326,8 +327,8 @@ public class ApolloClient implements Closeable {
       return this;
     }
 
-    public Builder wsReopenWhen(@NotNull WebSocketNetworkTransport.ReopenWhen reopenWhen) {
-      this.wsReopenWhen = checkNotNull(reopenWhen, "reopenWhen is null");
+    public Builder wsReopenWhen(WebSocketNetworkTransport.ReopenWhen reopenWhen) {
+      this.wsReopenWhen = reopenWhen;
       return this;
     }
 
@@ -336,8 +337,8 @@ public class ApolloClient implements Closeable {
       return this;
     }
 
-    public Builder httpEngine(@NotNull HttpEngine httpEngine) {
-      this.httpEngine = checkNotNull(httpEngine, "httpEngine is null");
+    public Builder httpEngine(HttpEngine httpEngine) {
+      this.httpEngine = httpEngine;
       return this;
     }
 
@@ -346,16 +347,23 @@ public class ApolloClient implements Closeable {
       return this;
     }
 
-    public Builder networkTransport(@NotNull NetworkTransport networkTransport) {
-      this.networkTransport = checkNotNull(networkTransport, "networkTransport is null");
+    public Builder networkTransport(NetworkTransport networkTransport) {
+      this.networkTransport = networkTransport;
       return this;
     }
 
-    public Builder subscriptionNetworkTransport(@NotNull NetworkTransport subscriptionNetworkTransport) {
-      this.subscriptionNetworkTransport = checkNotNull(subscriptionNetworkTransport, "subscriptionNetworkTransport is null");
+    public Builder subscriptionNetworkTransport(NetworkTransport subscriptionNetworkTransport) {
+      this.subscriptionNetworkTransport = subscriptionNetworkTransport;
       return this;
     }
 
+    public Builder retryOnError(Boolean retryOnError) {
+      throw new IllegalStateException("Not supported yet");
+//      this.retryOnError = retryOnError;
+//      return this;
+    }
+
+
     public ApolloClient build() {
       if (executor == null) {
         executor = defaultExecutor();
@@ -525,6 +533,10 @@ public class ApolloClient implements Closeable {
       return canBeBatched;
     }
 
+    @Nullable @Override public Boolean getRetryOnError() {
+      return retryOnError;
+    }
+
     @Override public Builder addExecutionContext(@NotNull ExecutionContext executionContext) {
       this.executionContext = this.executionContext.plus(executionContext);
       return this;
diff --git a/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/internal/DefaultApolloCall.java b/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/internal/DefaultApolloCall.java
index 37fcfadd57..60de44ce38 100644
--- a/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/internal/DefaultApolloCall.java
+++ b/libraries/apollo-runtime-java/src/main/java/com/apollographql/apollo3/runtime/java/internal/DefaultApolloCall.java
@@ -25,6 +25,7 @@ public class DefaultApolloCall<D extends Operation.Data> implements ApolloCall<D
   private Boolean sendDocument;
   private Boolean enableAutoPersistedQueries;
   private Boolean canBeBatched;
+  private Boolean retryOnError;
 
   public DefaultApolloCall(ApolloClient apolloClient, Operation<D> operation) {
     this.apolloClient = apolloClient;
@@ -40,6 +41,7 @@ public class DefaultApolloCall<D extends Operation.Data> implements ApolloCall<D
         .sendDocument(sendDocument)
         .enableAutoPersistedQueries(enableAutoPersistedQueries)
         .canBeBatched(canBeBatched)
+        .retryOnError(retryOnError)
         .build();
 
     return apolloClient.execute(apolloRequest, callback);
@@ -73,6 +75,10 @@ public class DefaultApolloCall<D extends Operation.Data> implements ApolloCall<D
     return canBeBatched;
   }
 
+  @Nullable @Override public Boolean getRetryOnError() {
+    return retryOnError;
+  }
+
   @Override public ApolloCall<D> addExecutionContext(@NotNull ExecutionContext executionContext) {
     this.executionContext = this.executionContext.plus(executionContext);
     return this;
@@ -115,4 +121,10 @@ public class DefaultApolloCall<D extends Operation.Data> implements ApolloCall<D
     this.canBeBatched = canBeBatched;
     return this;
   }
+
+  @Override public ApolloCall<D> retryOnError(@Nullable Boolean retryOnError) {
+    throw new IllegalStateException("Not supported yet");
+    //this.retryOnError = retryOnError;
+    //return this;
+  }
 }
diff --git a/libraries/apollo-runtime/api/apollo-runtime.api b/libraries/apollo-runtime/api/apollo-runtime.api
index 166d6c8706..a63ee9d29a 100644
--- a/libraries/apollo-runtime/api/apollo-runtime.api
+++ b/libraries/apollo-runtime/api/apollo-runtime.api
@@ -24,6 +24,7 @@ public final class com/apollographql/apollo3/ApolloCall : com/apollographql/apol
 	public fun httpMethod (Lcom/apollographql/apollo3/api/http/HttpMethod;)Lcom/apollographql/apollo3/ApolloCall;
 	public synthetic fun httpMethod (Lcom/apollographql/apollo3/api/http/HttpMethod;)Ljava/lang/Object;
 	public final fun ignoreApolloClientHttpHeaders (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/ApolloCall;
+	public synthetic fun retryOnError (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendApqExtensions (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/ApolloCall;
 	public synthetic fun sendApqExtensions (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendDocument (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/ApolloCall;
@@ -117,6 +118,7 @@ public final class com/apollographql/apollo3/ApolloClient$Builder : com/apollogr
 	public final fun httpServerUrl (Ljava/lang/String;)Lcom/apollographql/apollo3/ApolloClient$Builder;
 	public final fun interceptors (Ljava/util/List;)Lcom/apollographql/apollo3/ApolloClient$Builder;
 	public final fun networkTransport (Lcom/apollographql/apollo3/network/NetworkTransport;)Lcom/apollographql/apollo3/ApolloClient$Builder;
+	public synthetic fun retryOnError (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendApqExtensions (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/ApolloClient$Builder;
 	public synthetic fun sendApqExtensions (Ljava/lang/Boolean;)Ljava/lang/Object;
 	public fun sendDocument (Ljava/lang/Boolean;)Lcom/apollographql/apollo3/ApolloClient$Builder;
diff --git a/libraries/apollo-runtime/build.gradle.kts b/libraries/apollo-runtime/build.gradle.kts
index 84d8de10bf..b6aee0e94d 100644
--- a/libraries/apollo-runtime/build.gradle.kts
+++ b/libraries/apollo-runtime/build.gradle.kts
@@ -13,7 +13,7 @@ kotlin {
       dependencies {
         api(project(":apollo-api"))
         api(project(":apollo-mpp-utils"))
-        implementation(libs.atomicfu) // https://github.com/Kotlin/kotlinx.coroutines/issues/3968
+        implementation(libs.atomicfu.library) // https://github.com/Kotlin/kotlinx.coroutines/issues/3968
         api(libs.okio)
         api(libs.uuid)
         api(libs.kotlinx.coroutines)
diff --git a/libraries/apollo-runtime/src/appleMain/kotlin/com/apollographql/apollo3/network/http/DefaultHttpEngine.apple.kt b/libraries/apollo-runtime/src/appleMain/kotlin/com/apollographql/apollo3/network/http/DefaultHttpEngine.apple.kt
index 541ffa5692..aa3de371da 100644
--- a/libraries/apollo-runtime/src/appleMain/kotlin/com/apollographql/apollo3/network/http/DefaultHttpEngine.apple.kt
+++ b/libraries/apollo-runtime/src/appleMain/kotlin/com/apollographql/apollo3/network/http/DefaultHttpEngine.apple.kt
@@ -248,6 +248,8 @@ private class StreamingDataDelegate : NSObject(), NSURLSessionDataDelegateProtoc
 
     // Cleanup
     handlers.remove(dataTask)
+
+    completionHandler(willCacheResponse)
   }
 
   fun registerHandlerForTask(task: NSURLSessionTask, handler: Handler) {
diff --git a/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloCall.kt b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloCall.kt
index d21fb6ebf8..c036d7e6c9 100644
--- a/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloCall.kt
+++ b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloCall.kt
@@ -1,6 +1,7 @@
 package com.apollographql.apollo3
 
 import com.apollographql.apollo3.annotations.ApolloDeprecatedSince
+import com.apollographql.apollo3.annotations.ApolloExperimental
 import com.apollographql.apollo3.api.ApolloRequest
 import com.apollographql.apollo3.api.ApolloResponse
 import com.apollographql.apollo3.api.ExecutionContext
@@ -33,7 +34,10 @@ class ApolloCall<D : Operation.Data> internal constructor(
     private set
   override var canBeBatched: Boolean? = null
     private set
-  
+  @ApolloExperimental
+  override var retryOnError: Boolean? = null
+    private set
+
   /**
    * The HTTP headers to be sent with the request.
    * By default, these are *added* on top of any HTTP header previously set on [ApolloClient]. Call [ignoreApolloClientHttpHeaders]`(true)`
@@ -87,6 +91,10 @@ class ApolloCall<D : Operation.Data> internal constructor(
     this.canBeBatched = canBeBatched
   }
 
+  @ApolloExperimental
+  override fun retryOnError(retryOnError: Boolean?): ApolloCall<D> = apply {
+    this.retryOnError = retryOnError
+  }
   /**
    * If set to true, the HTTP headers set on [ApolloClient] will not be used for the call, only the ones set on this [ApolloCall] will be
    * used. If set to false, both sets of headers will be concatenated and used.
@@ -107,6 +115,7 @@ class ApolloCall<D : Operation.Data> internal constructor(
         .sendDocument(sendDocument)
         .enableAutoPersistedQueries(enableAutoPersistedQueries)
         .canBeBatched(canBeBatched)
+        .retryOnError(retryOnError)
   }
 
   /**
@@ -153,6 +162,7 @@ class ApolloCall<D : Operation.Data> internal constructor(
         .sendDocument(sendDocument)
         .enableAutoPersistedQueries(enableAutoPersistedQueries)
         .canBeBatched(canBeBatched)
+        .retryOnError(retryOnError)
         .build()
   }
 
diff --git a/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloClient.kt b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloClient.kt
index 3c46a3358f..43b6516953 100644
--- a/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloClient.kt
+++ b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/ApolloClient.kt
@@ -1,6 +1,7 @@
 package com.apollographql.apollo3
 
 import com.apollographql.apollo3.annotations.ApolloDeprecatedSince
+import com.apollographql.apollo3.annotations.ApolloExperimental
 import com.apollographql.apollo3.api.Adapter
 import com.apollographql.apollo3.api.ApolloRequest
 import com.apollographql.apollo3.api.ApolloResponse
@@ -57,6 +58,10 @@ private constructor(
   override val sendDocument: Boolean? = builder.sendDocument
   override val enableAutoPersistedQueries: Boolean? = builder.enableAutoPersistedQueries
   override val canBeBatched: Boolean? = builder.canBeBatched
+  @ApolloExperimental
+  override val retryOnError: Boolean? = builder.retryOnError
+  @ApolloExperimental
+  val retryOnErrorInterceptor = builder.retryOnErrorInterceptor
 
   init {
     networkTransport = if (builder.networkTransport != null) {
@@ -245,10 +250,20 @@ private constructor(
             // canBeBatched(apolloRequest.canBeBatched)
             addHttpHeader(ExecutionOptions.CAN_BE_BATCHED, canBeBatched.toString())
           }
+          if (apolloRequest.retryOnError != null) {
+            retryOnError(apolloRequest.retryOnError)
+          }
         }
         .build()
 
-    return DefaultInterceptorChain(interceptors + networkInterceptor, 0)
+    val allInterceptors = buildList{
+      addAll(interceptors)
+      if (retryOnErrorInterceptor != null) {
+        add(retryOnErrorInterceptor)
+      }
+      add(networkInterceptor)
+    }
+    return DefaultInterceptorChain(allInterceptors, 0)
         .proceed(request)
         .let {
           if (throwing) {
@@ -263,6 +278,10 @@ private constructor(
         }
   }
 
+  fun newBuilder(): Builder {
+    return builder.copy()
+  }
+
   /**
    * A Builder used to create instances of [ApolloClient]
    */
@@ -276,6 +295,24 @@ private constructor(
     private val _httpInterceptors: MutableList<HttpInterceptor> = mutableListOf()
     val httpInterceptors: List<HttpInterceptor> = _httpInterceptors
 
+    override var executionContext: ExecutionContext = ExecutionContext.Empty
+      private set
+    override var httpMethod: HttpMethod? = null
+      private set
+    override var httpHeaders: List<HttpHeader>? = null
+      private set
+    override var sendApqExtensions: Boolean? = null
+      private set
+    override var sendDocument: Boolean? = null
+      private set
+    override var enableAutoPersistedQueries: Boolean? = null
+      private set
+    override var canBeBatched: Boolean? = null
+      private set
+    @ApolloExperimental
+    override var retryOnError: Boolean? = null
+      private set
+
     var networkTransport: NetworkTransport? = null
       private set
     var subscriptionNetworkTransport: NetworkTransport? = null
@@ -300,21 +337,20 @@ private constructor(
       private set
     var webSocketReopenServerUrl: (suspend () -> String)? = null
       private set
-    override var executionContext: ExecutionContext = ExecutionContext.Empty
-      private set
-    override var httpMethod: HttpMethod? = null
-      private set
-    override var httpHeaders: List<HttpHeader>? = null
-      private set
-    override var sendApqExtensions: Boolean? = null
-      private set
-    override var sendDocument: Boolean? = null
-      private set
-    override var enableAutoPersistedQueries: Boolean? = null
-      private set
-    override var canBeBatched: Boolean? = null
+    @ApolloExperimental
+    var retryOnErrorInterceptor: ApolloInterceptor? = null
       private set
 
+    @ApolloExperimental
+    fun retryOnErrorInterceptor(retryOnErrorInterceptor: ApolloInterceptor?): Builder = apply {
+      this.retryOnErrorInterceptor = retryOnErrorInterceptor
+    }
+
+    @ApolloExperimental
+    override fun retryOnError(retryOnError: Boolean?): Builder = apply {
+      this.retryOnError = retryOnError
+    }
+
     override fun httpMethod(httpMethod: HttpMethod?): Builder = apply {
       this.httpMethod = httpMethod
     }
@@ -612,14 +648,12 @@ private constructor(
           .webSocketReopenWhen(webSocketReopenWhen)
           .webSocketIdleTimeoutMillis(webSocketIdleTimeoutMillis)
           .wsProtocol(wsProtocolFactory)
+          .retryOnError(retryOnError)
+          .retryOnErrorInterceptor(retryOnErrorInterceptor)
       return builder
     }
   }
 
-  fun newBuilder(): Builder {
-    return builder.copy()
-  }
-
   companion object {
     @Deprecated("Used for backward compatibility with 2.x", ReplaceWith("ApolloClient.Builder()"), level = DeprecationLevel.ERROR)
     @ApolloDeprecatedSince(ApolloDeprecatedSince.Version.v3_0_0)
diff --git a/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/interceptor/RetryOnErrorInterceptor.kt b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/interceptor/RetryOnErrorInterceptor.kt
new file mode 100644
index 0000000000..f8db0e39fa
--- /dev/null
+++ b/libraries/apollo-runtime/src/commonMain/kotlin/com/apollographql/apollo3/interceptor/RetryOnErrorInterceptor.kt
@@ -0,0 +1,63 @@
+package com.apollographql.apollo3.interceptor
+
+import com.apollographql.apollo3.annotations.ApolloExperimental
+import com.apollographql.apollo3.api.ApolloRequest
+import com.apollographql.apollo3.api.ApolloResponse
+import com.apollographql.apollo3.api.Operation
+import com.apollographql.apollo3.exception.ApolloException
+import com.apollographql.apollo3.exception.SubscriptionOperationException
+import com.benasher44.uuid.uuid4
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.retryWhen
+import kotlin.math.pow
+import kotlin.time.Duration.Companion.seconds
+
+@ApolloExperimental
+class RetryOnErrorInterceptor(private val defaultRetryOnError: ((ApolloRequest<*>) -> Boolean)): ApolloInterceptor {
+  override fun <D : Operation.Data> intercept(request: ApolloRequest<D>, chain: ApolloInterceptorChain): Flow<ApolloResponse<D>> {
+    val retryOnError = request.retryOnError ?: defaultRetryOnError(request)
+    if (!retryOnError) {
+      return chain.proceed(request)
+    }
+
+    var first = true
+    var attempt = 0
+    return flow {
+      val actualRequest = if (first) {
+        first = false
+        request
+      } else {
+        request.newBuilder().requestUuid(uuid4()).build()
+      }
+      emitAll(chain.proceed(actualRequest))
+    }.onEach {
+      if (it.exception != null && it.exception!!.isTerminalAndRecoverable()) {
+        throw RetryException
+      } else {
+        attempt = 0
+      }
+    }.retryWhen { cause, _ ->
+      attempt++
+
+      delay(2.0.pow(attempt).seconds)
+      cause is RetryException
+    }
+  }
+}
+
+
+private fun ApolloException.isTerminalAndRecoverable(): Boolean {
+  when (this) {
+    is SubscriptionOperationException -> {
+      // This often means a validation error on the subscription. It is unrecoverable
+      return false
+    }
+    else -> return true
+  }
+}
+
+private object RetryException: Exception()
\ No newline at end of file
diff --git a/libraries/apollo-websocket-network-transport-incubating/build.gradle.kts b/libraries/apollo-websocket-network-transport-incubating/build.gradle.kts
index 3a2365cc12..62ce3ecb6e 100644
--- a/libraries/apollo-websocket-network-transport-incubating/build.gradle.kts
+++ b/libraries/apollo-websocket-network-transport-incubating/build.gradle.kts
@@ -5,7 +5,6 @@ plugins {
 apolloLibrary(
     javaModuleName = "com.apollographql.apollo3.network.websocket",
     withLinux = false,
-    publish = false
 )
 
 kotlin {
@@ -14,7 +13,7 @@ kotlin {
       dependencies {
         api(project(":apollo-runtime"))
         api(project(":apollo-mpp-utils"))
-        api(libs.atomicfu)
+        api(libs.atomicfu.library)
       }
     }
 


Test patch:
diff --git a/benchmark/microbenchmark/src/androidTest/java/com/apollographql/apollo3/benchmark/CacheIncubatingTests.kt b/benchmark/microbenchmark/src/androidTest/java/com/apollographql/apollo3/benchmark/CacheIncubatingTests.kt
index 2e91ce8ebf..560c270c8e 100644
--- a/benchmark/microbenchmark/src/androidTest/java/com/apollographql/apollo3/benchmark/CacheIncubatingTests.kt
+++ b/benchmark/microbenchmark/src/androidTest/java/com/apollographql/apollo3/benchmark/CacheIncubatingTests.kt
@@ -19,6 +19,7 @@ import com.apollographql.apollo3.benchmark.test.R
 import com.apollographql.apollo3.cache.normalized.incubating.api.CacheHeaders
 import com.apollographql.apollo3.cache.normalized.incubating.api.CacheKeyGenerator
 import com.apollographql.apollo3.cache.normalized.incubating.api.CacheResolver
+import com.apollographql.apollo3.cache.normalized.incubating.api.DefaultRecordMerger
 import com.apollographql.apollo3.cache.normalized.incubating.api.FieldPolicyCacheResolver
 import com.apollographql.apollo3.cache.normalized.incubating.api.MemoryCacheFactory
 import com.apollographql.apollo3.cache.normalized.incubating.api.ReadOnlyNormalizedCache
@@ -26,9 +27,11 @@ import com.apollographql.apollo3.cache.normalized.incubating.api.Record
 import com.apollographql.apollo3.cache.normalized.incubating.api.TypePolicyCacheKeyGenerator
 import com.apollographql.apollo3.cache.normalized.incubating.sql.SqlNormalizedCacheFactory
 import kotlinx.coroutines.runBlocking
+import org.junit.Assert
 import org.junit.Rule
 import org.junit.Test
 import java.lang.reflect.Method
+import java.util.concurrent.Executors
 
 @Suppress("UNCHECKED_CAST")
 class CacheIncubatingTests {
@@ -55,6 +58,27 @@ class CacheIncubatingTests {
     readFromCache("cacheResponseSql", responseBasedQuery, sql = true, ::checkResponseBased)
   }
 
+  @Test
+  fun concurrentCacheOperationMemory() {
+    concurrentReadWriteFromCache(operationBasedQuery, sql = false)
+  }
+
+  @Test
+  fun concurrentCacheOperationSql() {
+    concurrentReadWriteFromCache(operationBasedQuery, sql = true)
+  }
+
+  @Test
+  fun concurrentCacheResponseMemory() {
+    concurrentReadWriteFromCache(responseBasedQuery, sql = false)
+  }
+
+  @Test
+  fun concurrentCacheResponseSql() {
+    concurrentReadWriteFromCache(responseBasedQuery, sql = true)
+  }
+
+
   private fun <D : Query.Data> readFromCache(testName: String, query: Query<D>, sql: Boolean, check: (D) -> Unit) {
     val cache = if (sql) {
       Utils.dbFile.delete()
@@ -74,7 +98,7 @@ class CacheIncubatingTests {
     ) as Map<String, Record>
 
     runBlocking {
-      cache.merge(records.values.toList(), CacheHeaders.NONE)
+      cache.merge(records.values.toList(), CacheHeaders.NONE, DefaultRecordMerger)
     }
 
     if (sql) {
@@ -93,7 +117,55 @@ class CacheIncubatingTests {
     }
   }
 
+  private fun <D : Query.Data> concurrentReadWriteFromCache(query: Query<D>, sql: Boolean) {
+    val cache = if (sql) {
+      Utils.dbFile.delete()
+      // Pass context explicitly here because androidx.startup fails due to relocation
+      SqlNormalizedCacheFactory(InstrumentationRegistry.getInstrumentation().context, Utils.dbName, withDates = true).create()
+    } else {
+      MemoryCacheFactory().create()
+    }
+    val data = query.parseJsonResponse(resource(R.raw.calendar_response_simple).jsonReader()).data!!
+
+    val records = normalizeMethod.invoke(
+        null,
+        query,
+        data,
+        CustomScalarAdapters.Empty,
+        TypePolicyCacheKeyGenerator,
+    ) as Map<String, Record>
+
+    val threadPool = Executors.newFixedThreadPool(CONCURRENCY)
+    benchmarkRule.measureRepeated {
+      val futures = (1..CONCURRENCY).map {
+        threadPool.submit {
+          // Let each thread execute a few writes/reads
+          repeat(WORK_LOAD) {
+            cache.merge(records.values.toList(), CacheHeaders.NONE, DefaultRecordMerger)
+
+            val data2 = readDataFromCacheMethod.invoke(
+                null,
+                query,
+                CustomScalarAdapters.Empty,
+                cache,
+                FieldPolicyCacheResolver,
+                CacheHeaders.NONE
+            ) as D
+
+            Assert.assertEquals(data, data2)
+          }
+        }
+      }
+      // Wait for all threads to finish
+      futures.forEach { it.get() }
+    }
+  }
+
+
   companion object {
+    private const val CONCURRENCY = 15
+    private const val WORK_LOAD = 15
+
     /**
      * There doesn't seem to be a way to relocate Kotlin metdata and kotlin_module files so we rely on reflection to call top-level
      * methods
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/MemoryCacheTest.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/MemoryCacheTest.kt
index 9bf94cc2e0..8eb3df9708 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/MemoryCacheTest.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/MemoryCacheTest.kt
@@ -3,6 +3,7 @@ package com.apollographql.apollo3.cache.normalized
 import com.apollographql.apollo3.cache.normalized.api.ApolloCacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheKey
+import com.apollographql.apollo3.cache.normalized.api.DefaultRecordMerger
 import com.apollographql.apollo3.cache.normalized.api.MemoryCache
 import com.apollographql.apollo3.cache.normalized.api.NormalizedCache
 import com.apollographql.apollo3.cache.normalized.api.Record
@@ -17,7 +18,7 @@ class MemoryCacheTest {
   fun testSaveAndLoad_singleRecord() {
     val lruCache = createCache()
     val testRecord = createTestRecord("1")
-    lruCache.merge(testRecord, CacheHeaders.NONE)
+    lruCache.merge(testRecord, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertTestRecordPresentAndAccurate(testRecord, lruCache)
   }
@@ -29,7 +30,7 @@ class MemoryCacheTest {
     val testRecord2 = createTestRecord("2")
     val testRecord3 = createTestRecord("3")
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertTestRecordPresentAndAccurate(testRecord1, lruCache)
     assertTestRecordPresentAndAccurate(testRecord2, lruCache)
@@ -43,7 +44,7 @@ class MemoryCacheTest {
     val testRecord2 = createTestRecord("2")
     val testRecord3 = createTestRecord("3")
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     val readRecords = lruCache.loadRecords(listOf("key1", "key2", "key3"), CacheHeaders.NONE)
     assertTrue(readRecords.containsAll(records))
@@ -64,11 +65,11 @@ class MemoryCacheTest {
 
     val lruCache = createCache(
         // all records won't fit as there is timestamp that stored with each record
-        maxSizeBytes = 200
+        maxSizeBytes = 600
     )
 
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     //Cache does not reveal exactly how it handles eviction, but appears
     //to evict more than is strictly necessary. Regardless, any sane eviction
@@ -84,11 +85,11 @@ class MemoryCacheTest {
     val testRecord3 = createTestRecord("3")
 
     val lruCache = createCache(
-        maxSizeBytes = 270
+        maxSizeBytes = 800
     )
 
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertNotNull(lruCache.loadRecord(testRecord1.key, CacheHeaders.NONE))
     assertNotNull(lruCache.loadRecord(testRecord2.key, CacheHeaders.NONE))
@@ -100,7 +101,7 @@ class MemoryCacheTest {
         mutationId = testRecord1.mutationId
     )
 
-    lruCache.merge(updatedRestRecord1, CacheHeaders.NONE)
+    lruCache.merge(updatedRestRecord1, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertNotNull(lruCache.loadRecord(testRecord1.key, CacheHeaders.NONE))
     assertNotNull(lruCache.loadRecord(testRecord2.key, CacheHeaders.NONE))
@@ -108,10 +109,10 @@ class MemoryCacheTest {
   }
 
   @Test
-  fun testExpiresImmediatly() {
+  fun testExpiresImmediately() {
     val testRecord = createTestRecord("")
     val lruCache = createCache(expireAfterMillis = 0)
-    lruCache.merge(testRecord, CacheHeaders.NONE)
+    lruCache.merge(testRecord, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertNull(lruCache.loadRecord(testRecord.key, CacheHeaders.NONE))
   }
@@ -120,10 +121,10 @@ class MemoryCacheTest {
   @Test
   fun testDualCacheSingleRecord() {
     val secondaryCache = createCache()
-    val primaryCache = createCache().chain(secondaryCache)
+    val primaryCache = createCache(nextCache = secondaryCache)
 
     val mockRecord = createTestRecord("")
-    primaryCache.merge(mockRecord, CacheHeaders.NONE)
+    primaryCache.merge(mockRecord, CacheHeaders.NONE, DefaultRecordMerger)
 
     //verify write through behavior
     assertEquals(mockRecord.fields, primaryCache.loadRecord(mockRecord.key, CacheHeaders.NONE)?.fields)
@@ -133,13 +134,13 @@ class MemoryCacheTest {
   @Test
   fun testDualCacheMultipleRecord() {
     val secondaryCache = createCache()
-    val primaryCache = createCache().chain(secondaryCache)
+    val primaryCache = createCache(nextCache = secondaryCache)
 
     val testRecord1 = createTestRecord("1")
     val testRecord2 = createTestRecord("2")
     val testRecord3 = createTestRecord("3")
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    primaryCache.merge(records, CacheHeaders.NONE)
+    primaryCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     val keys = listOf(testRecord1.key, testRecord2.key, testRecord3.key)
     assertEquals(3, primaryCache.loadRecords(keys, CacheHeaders.NONE).size)
@@ -149,7 +150,7 @@ class MemoryCacheTest {
   @Test
   fun testDualCache_recordNotPresent() {
     val secondaryCache = createCache()
-    val primaryCache = createCache().chain(secondaryCache)
+    val primaryCache = createCache(nextCache = secondaryCache)
     assertNull(primaryCache.loadRecord("key", CacheHeaders.NONE))
   }
 
@@ -157,13 +158,13 @@ class MemoryCacheTest {
   @Test
   fun testDualCache_clearAll() {
     val secondaryCache = createCache()
-    val primaryCache = createCache().chain(secondaryCache) as MemoryCache
+    val primaryCache = createCache(nextCache = secondaryCache)
 
     val testRecord1 = createTestRecord("1")
     val testRecord2 = createTestRecord("2")
     val testRecord3 = createTestRecord("3")
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    primaryCache.merge(records, CacheHeaders.NONE)
+    primaryCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     primaryCache.clearAll()
 
@@ -174,10 +175,10 @@ class MemoryCacheTest {
   @Test
   fun testDualCache_readFromNext() {
     val secondaryCache = createCache()
-    val primaryCache = createCache().chain(secondaryCache) as MemoryCache
+    val primaryCache = createCache(nextCache = secondaryCache)
 
     val testRecord = createTestRecord("")
-    primaryCache.merge(testRecord, CacheHeaders.NONE)
+    primaryCache.merge(testRecord, CacheHeaders.NONE, DefaultRecordMerger)
 
     primaryCache.clearCurrentCache()
 
@@ -191,7 +192,7 @@ class MemoryCacheTest {
     val lruCache = createCache()
     val testRecord = createTestRecord("1")
 
-    lruCache.merge(testRecord, CacheHeaders.NONE)
+    lruCache.merge(testRecord, CacheHeaders.NONE, DefaultRecordMerger)
 
     val headers = CacheHeaders.builder().addHeader(ApolloCacheHeaders.EVICT_AFTER_READ, "true").build()
 
@@ -206,7 +207,7 @@ class MemoryCacheTest {
 
     val headers = CacheHeaders.builder().addHeader(ApolloCacheHeaders.DO_NOT_STORE, "true").build()
 
-    lruCache.merge(testRecord, headers)
+    lruCache.merge(testRecord, headers, DefaultRecordMerger)
 
     assertNull(lruCache.loadRecord(testRecord.key, headers))
   }
@@ -219,7 +220,7 @@ class MemoryCacheTest {
     val testRecord2 = createTestRecord("2")
     val testRecord3 = createTestRecord("3")
     val records = listOf(testRecord1, testRecord2, testRecord3)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     with(lruCache.dump()) {
       val cache = this[MemoryCache::class]!!
@@ -252,7 +253,7 @@ class MemoryCacheTest {
     )
 
     val records = listOf(record1, record2)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertTrue(lruCache.remove(CacheKey(record2.key), cascade = false))
     assertNotNull(lruCache.loadRecord(record1.key, CacheHeaders.NONE))
@@ -278,7 +279,7 @@ class MemoryCacheTest {
     )
 
     val records = listOf(record1, record2)
-    lruCache.merge(records, CacheHeaders.NONE)
+    lruCache.merge(records, CacheHeaders.NONE, DefaultRecordMerger)
 
     assertTrue(lruCache.remove(CacheKey(record2.key), cascade = true))
     assertNull(lruCache.loadRecord(record1.key, CacheHeaders.NONE))
@@ -287,8 +288,9 @@ class MemoryCacheTest {
   private fun createCache(
       maxSizeBytes: Int = 10 * 1024,
       expireAfterMillis: Long = -1,
+      nextCache: NormalizedCache? = null,
   ): MemoryCache {
-    return MemoryCache(maxSizeBytes = maxSizeBytes, expireAfterMillis = expireAfterMillis)
+    return MemoryCache(maxSizeBytes = maxSizeBytes, expireAfterMillis = expireAfterMillis, nextCache = nextCache)
   }
 
   private fun assertTestRecordPresentAndAccurate(testRecord: Record, store: NormalizedCache) {
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/PatternTest.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/PatternTest.kt
index e9087c7293..d314a75a91 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/PatternTest.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/PatternTest.kt
@@ -1,6 +1,6 @@
 package com.apollographql.apollo3.cache.normalized
 
-import com.apollographql.apollo3.cache.normalized.api.NormalizedCache.Companion.patternToRegex
+import com.apollographql.apollo3.cache.normalized.api.internal.patternToRegex
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFails
@@ -27,4 +27,4 @@ class PatternTest {
   private fun matches(pattern: String, str: String): Boolean {
     return patternToRegex(pattern).matches(str)
   }
-}
\ No newline at end of file
+}
diff --git a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/internal/LruCacheTest.kt b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/internal/LruCacheTest.kt
index 96ed69f155..1772b9e0a9 100644
--- a/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/internal/LruCacheTest.kt
+++ b/libraries/apollo-normalized-cache-api-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/internal/LruCacheTest.kt
@@ -1,128 +1,130 @@
 package com.apollographql.apollo3.cache.normalized.internal
 
 import com.apollographql.apollo3.cache.normalized.api.internal.LruCache
+import com.apollographql.apollo3.testing.internal.runTest
+import kotlinx.coroutines.delay
 import kotlin.test.Test
 import kotlin.test.assertEquals
 
 class LruCacheTest {
   @Test
   fun emptyCache() {
-    val cache = LruCache<String, String?>(10)
+    val cache = LruCache<String, String>(10, -1) { _, _ -> 1 }
 
-    assertEquals(0, cache.size())
+    assertEquals(0, cache.weight())
     assertEquals(null, cache["key"])
     assertEquals(null, cache.remove("key"))
-    assertEquals(mapOf(), cache.dump())
+    assertEquals(mapOf(), cache.asMap())
 
     cache.clear()
-    assertEquals(0, cache.size())
+    assertEquals(0, cache.weight())
   }
 
   @Test
   fun addNewItemsToCache() {
-    val cache = LruCache<String, String?>(10)
+    val cache = LruCache<String, String>(10, -1) { _, _ -> 1 }
 
     val expectedEntries = mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     )
 
     expectedEntries.forEach { (key, value) ->
       cache[key] = value
     }
 
-    assertEquals(3, cache.size())
-    assertEquals(expectedEntries, cache.dump())
+    assertEquals(3, cache.weight())
+    assertEquals(expectedEntries, cache.asMap())
   }
 
   @Test
   fun removeItemsFromCache() {
-    val cache = LruCache<String, String?>(10)
+    val cache = LruCache<String, String>(10, -1) { _, _ -> 1 }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
 
     assertEquals(null, cache.remove("key"))
     assertEquals("value1", cache.remove("key1"))
-    assertEquals(null, cache.remove("key3"))
+    assertEquals("value3", cache.remove("key3"))
 
-    assertEquals(1, cache.size())
-    assertEquals(mapOf("key2" to "value2"), cache.dump())
+    assertEquals(1, cache.weight())
+    assertEquals(mapOf("key2" to "value2"), cache.asMap())
   }
 
   @Test
   fun clearCache() {
-    val cache = LruCache<String, String?>(10)
+    val cache = LruCache<String, String>(10, -1) { _, _ -> 1 }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
 
     cache.clear()
 
-    assertEquals(0, cache.size())
-    assertEquals(mapOf(), cache.dump())
+    assertEquals(0, cache.weight())
+    assertEquals(mapOf(), cache.asMap())
   }
 
   @Test
   fun trimCache() {
-    val cache = LruCache<String, String?>(2)
+    val cache = LruCache<String, String>(2, -1) { _, _ -> 1 }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
 
-    assertEquals(2, cache.size())
+    assertEquals(2, cache.weight())
     assertEquals(
         mapOf(
             "key2" to "value2",
-            "key3" to null
+            "key3" to "value3"
         ),
-        cache.dump()
+        cache.asMap()
     )
   }
 
   @Test
   fun addItemToCacheWithCustomWeigher() {
-    val cache = LruCache<String, String?>(100) { key, value ->
-      key.length + (value?.length ?: 0)
+    val cache = LruCache<String, String>(100, -1) { key, value ->
+      key.length + value.length
     }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
 
-    assertEquals(24, cache.size())
+    assertEquals(30, cache.weight())
   }
 
   @Test
   fun removeItemFromCacheWithCustomWeigher() {
-    val cache = LruCache<String, String?>(100) { key, value ->
-      key.length + (value?.length ?: 0)
+    val cache = LruCache<String, String>(100, -1) { key, value ->
+      key.length + value.length
     }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
@@ -130,40 +132,40 @@ class LruCacheTest {
     cache.remove("key2")
     cache.remove("key3")
 
-    assertEquals(10, cache.size())
+    assertEquals(10, cache.weight())
   }
 
   @Test
   fun trimCacheWithCustomWeigher() {
-    val cache = LruCache<String, String?>(12) { key, value ->
-      key.length + (value?.length ?: 0)
+    val cache = LruCache<String, String>(12, -1) { key, value ->
+      key.length + value.length
     }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
 
-    assertEquals(4, cache.size())
+    assertEquals(10, cache.weight())
     assertEquals(
         mapOf(
-            "key3" to null
+            "key3" to "value3"
         ),
-        cache.dump()
+        cache.asMap()
     )
   }
 
   @Test
   fun recentUsedItem() {
-    val cache = LruCache<String, String?>(10)
+    val cache = LruCache<String, String>(10, -1) { _, _ -> 1 }
 
     mapOf(
         "key1" to "value1",
         "key2" to "value2",
-        "key3" to null
+        "key3" to "value3"
     ).forEach { (key, value) ->
       cache[key] = value
     }
@@ -174,10 +176,42 @@ class LruCacheTest {
     assertEquals(
         mapOf(
             "key2" to "value2",
-            "key3" to null,
+            "key3" to "value3",
             "key1" to "value1",
         ),
-        cache.dump()
+        cache.asMap()
+    )
+  }
+
+  @Test
+  fun expiration() = runTest {
+    val cache = LruCache<String, String>(10, 100) { _, _ -> 1 }
+
+    mapOf(
+        "key1" to "value1",
+        "key2" to "value2",
+        "key3" to "value3"
+    ).forEach { (key, value) ->
+      cache[key] = value
+    }
+
+    assertEquals(
+        mapOf(
+            "key1" to "value1",
+            "key2" to "value2",
+            "key3" to "value3"
+        ),
+        cache.asMap()
+    )
+
+    delay(200)
+    cache["key4"] = "value4"
+
+    assertEquals(
+        mapOf(
+            "key4" to "value4"
+        ),
+        cache.asMap()
     )
   }
 }
diff --git a/libraries/apollo-normalized-cache-sqlite-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCacheTest.kt b/libraries/apollo-normalized-cache-sqlite-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCacheTest.kt
index b45b5e8efc..9704e167b0 100644
--- a/libraries/apollo-normalized-cache-sqlite-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCacheTest.kt
+++ b/libraries/apollo-normalized-cache-sqlite-incubating/src/commonTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/SqlNormalizedCacheTest.kt
@@ -1,25 +1,21 @@
 package com.apollographql.apollo3.cache.normalized.sql
 
+import app.cash.sqldelight.Query
+import app.cash.sqldelight.Transacter
+import app.cash.sqldelight.db.QueryResult
+import app.cash.sqldelight.db.SqlCursor
+import app.cash.sqldelight.db.SqlDriver
+import app.cash.sqldelight.db.SqlPreparedStatement
 import com.apollographql.apollo3.cache.normalized.api.ApolloCacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheHeaders
 import com.apollographql.apollo3.cache.normalized.api.CacheKey
+import com.apollographql.apollo3.cache.normalized.api.DefaultRecordMerger
 import com.apollographql.apollo3.cache.normalized.api.MemoryCacheFactory
 import com.apollographql.apollo3.cache.normalized.api.NormalizedCache
 import com.apollographql.apollo3.cache.normalized.api.Record
 import com.apollographql.apollo3.cache.normalized.sql.internal.JsonRecordDatabase
 import com.apollographql.apollo3.cache.normalized.sql.internal.json.JsonQueries
-import com.apollographql.apollo3.cache.normalized.sql.internal.json.RecordForKey
-import com.apollographql.apollo3.cache.normalized.sql.internal.json.Records
-import com.apollographql.apollo3.cache.normalized.sql.internal.json.RecordsForKeys
 import com.apollographql.apollo3.exception.apolloExceptionHandler
-import app.cash.sqldelight.Query
-import app.cash.sqldelight.Transacter
-import app.cash.sqldelight.TransactionWithReturn
-import app.cash.sqldelight.TransactionWithoutReturn
-import app.cash.sqldelight.db.QueryResult
-import app.cash.sqldelight.db.SqlCursor
-import app.cash.sqldelight.db.SqlDriver
-import app.cash.sqldelight.db.SqlPreparedStatement
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
@@ -91,6 +87,7 @@ class SqlNormalizedCacheTest {
             ),
         ),
         cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
     val record = cache.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
     assertNotNull(record)
@@ -110,68 +107,13 @@ class SqlNormalizedCacheTest {
             ),
         ),
         cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
     cache.remove(cacheKey = CacheKey(STANDARD_KEY), cascade = false)
     val record = cache.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
     assertNull(record)
   }
 
-  @Test
-  fun testSqlThenMemoryMergeAndDeleteCascade() {
-    val chainedCache = cache.chain(MemoryCacheFactory().create())
-    testChainedMergeAndDeleteCascade(chainedCache)
-  }
-
-  @Test
-  fun testMemoryThenSqlMergeAndDeleteCascade() {
-    val chainedCache = MemoryCacheFactory().create().chain(cache)
-    testChainedMergeAndDeleteCascade(chainedCache)
-  }
-
-  private fun testChainedMergeAndDeleteCascade(chainedCache: NormalizedCache) {
-    chainedCache.merge(
-        record = Record(
-            key = "referencedKey",
-            fields = mapOf(
-                "field1" to true,
-            ),
-        ),
-        cacheHeaders = CacheHeaders.NONE,
-    )
-    chainedCache.merge(
-        record = Record(
-            key = STANDARD_KEY,
-            fields = mapOf(
-                "field1" to "value1",
-                "field2" to CacheKey("referencedKey"),
-            ),
-        ),
-        cacheHeaders = CacheHeaders.NONE,
-    )
-    chainedCache.onEach {
-      val record1 = it.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
-      assertNotNull(record1)
-      val record2 = it.loadRecord("referencedKey", CacheHeaders.NONE)
-      assertNotNull(record2)
-    }
-    chainedCache.remove(cacheKey = CacheKey(STANDARD_KEY), cascade = true)
-    chainedCache.onEach {
-      val record1 = it.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
-      assertNull(record1)
-      val record2 = it.loadRecord("referencedKey", CacheHeaders.NONE)
-      assertNull(record2)
-    }
-  }
-
-  private fun NormalizedCache.onEach(block: (NormalizedCache) -> Unit): NormalizedCache? {
-    var c: NormalizedCache? = this
-    while (c != null) {
-      block(c)
-      c = c.nextCache
-    }
-    return c
-  }
-
   @Test
   fun testClearAll() {
     createRecord(QUERY_ROOT_KEY)
@@ -214,6 +156,7 @@ class SqlNormalizedCacheTest {
             fields = emptyMap(),
         ),
         cacheHeaders = CacheHeaders.builder().addHeader(ApolloCacheHeaders.DO_NOT_STORE, "true").build(),
+        recordMerger = DefaultRecordMerger,
     )
     val record = cache.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
     assertNull(record)
@@ -230,6 +173,7 @@ class SqlNormalizedCacheTest {
             ),
         ),
         cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
     val record = cache.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
     assertNotNull(record)
@@ -249,7 +193,8 @@ class SqlNormalizedCacheTest {
                 "newFieldKey" to true,
             ),
         ),
-        cacheHeaders = CacheHeaders.NONE
+        cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
     val record = cache.loadRecord(STANDARD_KEY, CacheHeaders.NONE)
     assertNotNull(record)
@@ -299,10 +244,12 @@ class SqlNormalizedCacheTest {
             ),
         ),
         cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
     assertEquals("Unable to merge a record from the database", throwable!!.message)
     assertEquals("bad cache", throwable!!.cause!!.message)
   }
+
   private val BadDriver = object : SqlDriver {
     override fun close() {
       throw IllegalStateException("bad cache")
@@ -353,6 +300,7 @@ class SqlNormalizedCacheTest {
             ),
         ),
         cacheHeaders = CacheHeaders.NONE,
+        recordMerger = DefaultRecordMerger,
     )
   }
 
diff --git a/libraries/apollo-normalized-cache-sqlite-incubating/src/jvmTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/TrimTest.kt b/libraries/apollo-normalized-cache-sqlite-incubating/src/jvmTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/TrimTest.kt
index 7563e079ce..660bc16df3 100644
--- a/libraries/apollo-normalized-cache-sqlite-incubating/src/jvmTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/TrimTest.kt
+++ b/libraries/apollo-normalized-cache-sqlite-incubating/src/jvmTest/kotlin/com/apollographql/apollo3/cache/normalized/sql/TrimTest.kt
@@ -1,6 +1,7 @@
 package com.apollographql.apollo3.cache.normalized.sql
 
 import com.apollographql.apollo3.cache.normalized.api.CacheHeaders
+import com.apollographql.apollo3.cache.normalized.api.DefaultRecordMerger
 import com.apollographql.apollo3.cache.normalized.api.Record
 import com.apollographql.apollo3.cache.normalized.sql.TrimmableNormalizedCacheFactory
 import org.junit.Test
@@ -28,7 +29,7 @@ class TrimTest {
         date = mapOf("key" to 0L),
         metadata = emptyMap()
     )
-    cache.merge(oldRecord, CacheHeaders.NONE)
+    cache.merge(oldRecord, CacheHeaders.NONE, recordMerger = DefaultRecordMerger)
 
     val newRecords = 0.until(2*1024).map {
       Record(
@@ -39,7 +40,7 @@ class TrimTest {
           metadata = emptyMap()
       )
     }
-    cache.merge(newRecords, CacheHeaders.NONE)
+    cache.merge(newRecords, CacheHeaders.NONE, recordMerger = DefaultRecordMerger)
 
     assertEquals(9596928, dbFile.length())
 
@@ -50,4 +51,4 @@ class TrimTest {
     // The oldest key must have been removed
     assertNull(trimmedCache.loadRecord("old", CacheHeaders.NONE))
   }
-}
\ No newline at end of file
+}
diff --git a/libraries/apollo-testing-support/build.gradle.kts b/libraries/apollo-testing-support/build.gradle.kts
index 6692fb09ae..ebe442bac7 100644
--- a/libraries/apollo-testing-support/build.gradle.kts
+++ b/libraries/apollo-testing-support/build.gradle.kts
@@ -15,7 +15,7 @@ kotlin {
         api(project(":apollo-runtime"))
         api(project(":apollo-mockserver"))
         api(libs.kotlinx.coroutines)
-        implementation(libs.atomicfu.get().toString()) {
+        implementation(libs.atomicfu.library.get().toString()) {
           because("We need locks in TestNetworkTransportHandler (we don't use the gradle plugin rewrite)")
         }
         implementation(libs.kotlinx.coroutines.test)
diff --git a/libraries/apollo-testing-support/src/commonMain/kotlin/com/apollographql/apollo3/testing/runTest.kt b/libraries/apollo-testing-support/src/commonMain/kotlin/com/apollographql/apollo3/testing/runTest.kt
index d89ac27462..f98088d9c2 100644
--- a/libraries/apollo-testing-support/src/commonMain/kotlin/com/apollographql/apollo3/testing/runTest.kt
+++ b/libraries/apollo-testing-support/src/commonMain/kotlin/com/apollographql/apollo3/testing/runTest.kt
@@ -14,9 +14,10 @@ class MockServerTest(val mockServer: MockServer, val apolloClient: ApolloClient,
  */
 @ApolloExperimental
 fun mockServerTest(
+    skipDelays: Boolean = true,
     clientBuilder: ApolloClient.Builder.() -> Unit = {},
     block: suspend MockServerTest.() -> Unit
-) = com.apollographql.apollo3.testing.internal.runTest(true) {
+) = com.apollographql.apollo3.testing.internal.runTest(skipDelays) {
   val mockServer = MockServer()
 
   val apolloClient = ApolloClient.Builder()
diff --git a/libraries/apollo-websocket-network-transport-incubating/src/commonTest/kotlin/com/apollographql/apollo3/network/ws/incubating/WebSocketEngineTest.kt b/libraries/apollo-websocket-network-transport-incubating/src/commonTest/kotlin/com/apollographql/apollo3/network/ws/incubating/WebSocketEngineTest.kt
index 5f36cab3a6..7f0ee50959 100644
--- a/libraries/apollo-websocket-network-transport-incubating/src/commonTest/kotlin/com/apollographql/apollo3/network/ws/incubating/WebSocketEngineTest.kt
+++ b/libraries/apollo-websocket-network-transport-incubating/src/commonTest/kotlin/com/apollographql/apollo3/network/ws/incubating/WebSocketEngineTest.kt
@@ -34,22 +34,28 @@ private data class Item(
 
 private class Listener(private val channel: Channel<Item>) : WebSocketListener {
   override fun onOpen() {
+    println("onOpen")
     channel.trySend(Item(open = true))
   }
 
   override fun onMessage(text: String) {
+    println("onMessage: $text")
     channel.trySend(TextMessage(text))
   }
 
   override fun onMessage(data: ByteArray) {
+    println("onMessage data")
     channel.trySend(DataMessage(data))
   }
 
   override fun onError(cause: ApolloException) {
+    println("onError $cause")
+    cause.printStackTrace()
     channel.trySend(Item(exception = cause))
   }
 
   override fun onClosed(code: Int?, reason: String?) {
+    println("onClosed $code, $reason")
     channel.trySend(CloseFrame(code, reason))
   }
 }
diff --git a/tests/java-client/src/main/graphql/main/operations.graphql b/tests/java-client/src/main/graphql/main/operations.graphql
index 8d765c2f4d..e4cca05661 100644
--- a/tests/java-client/src/main/graphql/main/operations.graphql
+++ b/tests/java-client/src/main/graphql/main/operations.graphql
@@ -3,7 +3,7 @@ query GetRandom {
 }
 
 query TimeQuery {
-  time
+  zero
 }
 
 
@@ -11,8 +11,8 @@ subscription OperationError {
   operationError
 }
 
-subscription Count($to: Int!, $delayMillis: Int!)  {
-  count(to: $to, delayMillis: $delayMillis)
+subscription Count($to: Int!, $intervalMillis: Int!)  {
+  count(to: $to, intervalMillis: $intervalMillis)
 }
 
 mutation CloseSocketMutation {
diff --git a/tests/pagination/build.gradle.kts b/tests/pagination/build.gradle.kts
index d1336df997..163edf0409 100644
--- a/tests/pagination/build.gradle.kts
+++ b/tests/pagination/build.gradle.kts
@@ -19,7 +19,7 @@ kotlin {
 
     findByName("commonTest")?.apply {
       dependencies {
-        implementation(libs.atomicfu)
+        implementation(libs.atomicfu.library)
         implementation(libs.apollo.testingsupport)
         implementation(libs.apollo.normalizedcache.incubating)
         implementation(libs.apollo.normalizedcache.sqlite.incubating)
diff --git a/tests/pagination/src/commonTest/kotlin/OffsetBasedWithPagePaginationTest.kt b/tests/pagination/src/commonTest/kotlin/OffsetBasedWithPagePaginationTest.kt
index ac3c871638..cecb9acadb 100644
--- a/tests/pagination/src/commonTest/kotlin/OffsetBasedWithPagePaginationTest.kt
+++ b/tests/pagination/src/commonTest/kotlin/OffsetBasedWithPagePaginationTest.kt
@@ -10,10 +10,9 @@ import com.apollographql.apollo3.cache.normalized.api.MetadataGeneratorContext
 import com.apollographql.apollo3.cache.normalized.api.NormalizedCacheFactory
 import com.apollographql.apollo3.cache.normalized.api.Record
 import com.apollographql.apollo3.cache.normalized.api.TypePolicyCacheKeyGenerator
-import com.apollographql.apollo3.cache.normalized.api.internal.OptimisticCache
+import com.apollographql.apollo3.cache.normalized.api.internal.OptimisticNormalizedCache
 import com.apollographql.apollo3.cache.normalized.sql.SqlNormalizedCacheFactory
 import com.apollographql.apollo3.testing.internal.runTest
-import kotlinx.atomicfu.atomic
 import pagination.offsetBasedWithPage.UsersQuery
 import pagination.offsetBasedWithPage.type.buildUser
 import pagination.offsetBasedWithPage.type.buildUserPage
@@ -233,13 +232,11 @@ class OffsetBasedWithPagePaginationTest {
 }
 
 internal fun assertChainedCachesAreEqual(apolloStore: ApolloStore) {
-  val hasNextCache = atomic(false)
-  apolloStore.accessCache { cache ->
-    // First cache is always OptimisticCache
-    hasNextCache.value = cache.nextCache!!.nextCache != null
+  val dump = apolloStore.dump().filterKeys {
+    // Ignore optimistic cache for comparison
+    it != OptimisticNormalizedCache::class
   }
-  if (!hasNextCache.value) return
-  val dump = apolloStore.dump().filterKeys { it != OptimisticCache::class }
+  if (dump.size < 2) return
   val caches = dump.values.toList()
   val cache1: Map<String, Record> = caches[0]
   val cache2: Map<String, Record> = caches[1]
diff --git a/tests/sample-server/build.gradle.kts b/tests/sample-server/build.gradle.kts
index 8bf045f6a9..336f7da02d 100644
--- a/tests/sample-server/build.gradle.kts
+++ b/tests/sample-server/build.gradle.kts
@@ -12,7 +12,7 @@ dependencies {
   implementation(libs.apollo.execution)
   implementation(libs.apollo.api)
   implementation(libs.kotlinx.coroutines)
-  implementation(libs.atomicfu)
+  implementation(libs.atomicfu.library)
 
   implementation(platform(libs.http4k.bom.get()))
   implementation(libs.http4k.core)
diff --git a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/SampleServer.kt b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/SampleServer.kt
index 95d91355a0..2abae3bc8a 100644
--- a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/SampleServer.kt
+++ b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/SampleServer.kt
@@ -1,5 +1,6 @@
 package com.apollographql.apollo.sample.server
 
+import com.apollographql.apollo.sample.server.graphql.SubscriptionRoot
 import com.apollographql.apollo3.api.ExecutionContext
 import com.apollographql.apollo3.ast.toGQLDocument
 import com.apollographql.apollo3.ast.toSchema
@@ -52,7 +53,7 @@ import java.io.Closeable
 import java.time.Duration
 import org.http4k.routing.ws.bind as wsBind
 
-fun ExecutableSchema(): ExecutableSchema {
+fun ExecutableSchema(tag: String): ExecutableSchema {
   val schema = GraphQLHttpHandler::class.java.classLoader
       .getResourceAsStream("schema.graphqls")!!
       .source()
@@ -61,7 +62,7 @@ fun ExecutableSchema(): ExecutableSchema {
       .toGQLDocument()
       .toSchema()
 
-  return SampleserverExecutableSchemaBuilder(schema)
+  return SampleserverExecutableSchemaBuilder(schema, rootSubscriptionObject = { SubscriptionRoot(tag) })
       .build()
 }
 
@@ -129,7 +130,7 @@ class WebSocketRegistry : ExecutionContext.Element {
   companion object Key : ExecutionContext.Key<WebSocketRegistry>
 }
 
-class CurrentWebSocket(val ws: Websocket): ExecutionContext.Element {
+class CurrentWebSocket(val ws: Websocket) : ExecutionContext.Element {
 
   override val key: ExecutionContext.Key<CurrentWebSocket> = Key
 
@@ -208,8 +209,8 @@ private fun WebSocketMessage.toWsMessage(): WsMessage {
   }
 }
 
-fun AppHandler(): PolyHandler {
-  val executableSchema = ExecutableSchema()
+fun AppHandler(tag: String): PolyHandler {
+  val executableSchema = ExecutableSchema(tag)
   val forceClose = WebSocketRegistry()
   val ws = websockets("/subscriptions" wsBind ApolloWebsocketHandler(executableSchema, webSocketRegistry = forceClose))
 
@@ -231,11 +232,11 @@ fun AppHandler(): PolyHandler {
 }
 
 fun runServer() {
-  AppHandler().asServer(Jetty(8000)).start().block()
+  AppHandler("").asServer(Jetty(8000)).start().block()
 }
 
-class SampleServer(val port: Int = 0) : Closeable {
-  private val server = AppHandler().asServer(Jetty(port, stopMode = ServerConfig.StopMode.Graceful(Duration.ZERO)))
+class SampleServer(port: Int = 0, tag: String = "") : Closeable {
+  private val server = AppHandler(tag).asServer(Jetty(port, stopMode = ServerConfig.StopMode.Graceful(Duration.ZERO)))
 
   init {
     server.start()
diff --git a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/QueryRoot.kt b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/QueryRoot.kt
index ef1dd42996..9b49f7dc76 100644
--- a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/QueryRoot.kt
+++ b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/QueryRoot.kt
@@ -6,7 +6,9 @@ import com.apollographql.apollo3.annotations.GraphQLObject
 @GraphQLObject(name = "Query")
 class QueryRoot {
   fun random(): Int = 42
-  fun time(): Int = 0
+  fun zero(): Int = 0
+  fun valueSharedWithSubscriptions(): Int = 0
+  fun secondsSinceEpoch(): Double = System.currentTimeMillis().div(1000).toDouble()
 }
 
 
diff --git a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/SubscriptionRoot.kt b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/SubscriptionRoot.kt
index 0f6667c13f..110d90f575 100644
--- a/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/SubscriptionRoot.kt
+++ b/tests/sample-server/src/main/kotlin/com/apollographql/apollo/sample/server/graphql/SubscriptionRoot.kt
@@ -3,39 +3,40 @@ package com.apollographql.apollo.sample.server.graphql
 import com.apollographql.apollo.sample.server.CurrentWebSocket
 import com.apollographql.apollo3.annotations.GraphQLObject
 import com.apollographql.apollo3.api.ExecutionContext
+import com.apollographql.apollo3.execution.websocket.subscriptionId
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.flow
 import org.http4k.websocket.WsStatus
 
 @GraphQLObject(name = "Subscription")
-class SubscriptionRoot {
-  fun count(to: Int, delayMillis: Int): Flow<Int> = flow {
+class SubscriptionRoot(private val tag: String) {
+  fun count(to: Int, intervalMillis: Int): Flow<Int> = flow {
     repeat(to) {
       emit(it)
-      if (delayMillis > 0) {
-        delay(delayMillis.toLong())
+      if (intervalMillis > 0) {
+        delay(intervalMillis.toLong())
       }
     }
   }
 
-  fun countString(to: Int, delayMillis: Int): Flow<String> = flow {
+  fun countString(to: Int, intervalMillis: Int): Flow<String> = flow {
     repeat(to) {
       emit(it.toString())
-      if (delayMillis > 0) {
-        delay(delayMillis.toLong())
+      if (intervalMillis > 0) {
+        delay(intervalMillis.toLong())
       }
     }
   }
 
-  fun time(): Flow<Int> = flow {
-    repeat(100) {
-      emit(it)
-      delay(100)
+  fun secondsSinceEpoch(intervalMillis: Int): Flow<Double> = flow {
+    while (true) {
+      emit(System.currentTimeMillis().div(1000).toDouble())
+      delay(intervalMillis.toLong())
     }
   }
 
-  fun operationError(): Flow<String> = flow<String> {
+  fun operationError(): Flow<String> = flow {
     throw Exception("Woops")
   }
 
@@ -52,4 +53,24 @@ class SubscriptionRoot {
 
     emit("closed")
   }
-}
\ No newline at end of file
+
+  fun state(executionContext: ExecutionContext, intervalMillis: Int): Flow<State> = flow {
+    while (true) {
+      emit(State(tag, executionContext.subscriptionId()))
+      delay(intervalMillis.toLong())
+    }
+  }
+
+  fun valueSharedWithSubscriptions(): Flow<Int> = flow {
+    repeat(10) {
+      emit(it)
+      delay(100)
+    }
+  }
+}
+
+@GraphQLObject
+class State(
+    val tag: String,
+    val subscriptionId: String
+)
\ No newline at end of file
diff --git a/tests/sample-server/src/main/resources/schema.graphqls b/tests/sample-server/src/main/resources/schema.graphqls
index dfa712d857..8e341a10d8 100644
--- a/tests/sample-server/src/main/resources/schema.graphqls
+++ b/tests/sample-server/src/main/resources/schema.graphqls
@@ -1,6 +1,8 @@
 type Query {
   random: Int!
-  time: Int!
+  zero: Int!
+  secondsSinceEpoch: Float!
+  valueSharedWithSubscriptions: Int!
 }
 
 type Mutation {
@@ -10,9 +12,9 @@ type Mutation {
 
 type Subscription {
   "Count from 0 until 'to', waiting 'delayMillis' after each response"
-  count(delayMillis: Int!, to: Int!): Int!
+  count(intervalMillis: Int!, to: Int!): Int!
   "Count from 0 until 'to', waiting 'delayMillis' after each response and returns each result as a String"
-  countString(delayMillis: Int!, to: Int!): String!
+  countString(intervalMillis: Int!, to: Int!): String!
   "Trigger an error when accessed"
   operationError: String!
   "Returns a GraphQL error after 'after' items"
@@ -21,5 +23,19 @@ type Subscription {
   "Force close the websocket this subscription is executing on"
   closeWebSocket: String!
 
-  time: Int!
+  "Returns the current seconds every intervalMillis"
+  secondsSinceEpoch(intervalMillis: Int! = 1000): Float!
+
+  "Returns the subscription state"
+  state(intervalMillis: Int! = 1000): State
+
+  valueSharedWithSubscriptions: Int!
+}
+
+type State {
+  "The tag of this server"
+  tag: String!
+
+  "The subscription id"
+  subscriptionId: String!
 }
diff --git a/tests/sqlite/src/commonTest/kotlin/ExpirationTest.kt b/tests/sqlite/src/commonTest/kotlin/ExpirationTest.kt
index 86985b8229..fbf5e502dd 100644
--- a/tests/sqlite/src/commonTest/kotlin/ExpirationTest.kt
+++ b/tests/sqlite/src/commonTest/kotlin/ExpirationTest.kt
@@ -51,7 +51,7 @@ class ExpirationTest {
 
     client.apolloStore.accessCache {
       // store records in the past
-      it.merge(records, cacheHeaders(currentTimeMillis() / 1000 - 15))
+      it.merge(records, cacheHeaders(currentTimeMillis() / 1000 - 15), DefaultRecordMerger)
     }
 
     val e = client.query(GetUserQuery()).fetchPolicy(FetchPolicy.CacheOnly).execute().exception as CacheMissException
@@ -65,7 +65,7 @@ class ExpirationTest {
 
     client.apolloStore.accessCache {
       // update records to be in the present
-      it.merge(records, cacheHeaders(currentTimeMillis() / 1000))
+      it.merge(records, cacheHeaders(currentTimeMillis() / 1000), DefaultRecordMerger)
     }
 
     val response2 = client.query(GetUserQuery()).fetchPolicy(FetchPolicy.CacheOnly).execute()
diff --git a/tests/websockets/src/commonMain/graphql/sample-server/operations.graphql b/tests/websockets/src/commonMain/graphql/sample-server/operations.graphql
index 9f99f236cc..a77423d313 100644
--- a/tests/websockets/src/commonMain/graphql/sample-server/operations.graphql
+++ b/tests/websockets/src/commonMain/graphql/sample-server/operations.graphql
@@ -1,21 +1,29 @@
-subscription Count($to: Int!, $delayMillis: Int!)  {
-  count(to: $to, delayMillis: $delayMillis)
+subscription Count($to: Int!, $intervalMillis: Int!)  {
+  count(to: $to, intervalMillis: $intervalMillis)
 }
 
-subscription CountString($to: Int!, $delayMillis: Int!)  {
-  countString(to: $to, delayMillis: $delayMillis)
+subscription CountString($to: Int!, $intervalMillis: Int!)  {
+  countString(to: $to, intervalMillis: $intervalMillis)
 }
 
-subscription TimeSubscription  {
-  time
+subscription SecondsSinceEpochSubscription($intervalMillis: Int!)  {
+  secondsSinceEpoch(intervalMillis: $intervalMillis)
 }
 
 subscription OperationError {
   operationError
 }
 
-query TimeQuery {
-  time
+query ValueSharedWithSubscriptionsQuery {
+  valueSharedWithSubscriptions
+}
+
+subscription ValueSharedWithSubscriptionsSubscription {
+  valueSharedWithSubscriptions
+}
+
+query ZeroQuery {
+  zero
 }
 
 mutation CloseSocketMutation {
@@ -29,3 +37,10 @@ subscription CloseSocketSubscription {
 subscription GraphqlAccessError($after: Int!) {
   graphqlAccessError(after: $after)
 }
+
+subscription Tag($intervalMillis: Int!)  {
+  state(intervalMillis: $intervalMillis) {
+    tag
+    subscriptionId
+  }
+}
\ No newline at end of file
diff --git a/tests/websockets/src/jvmTest/kotlin/CachedSubscriptionTest.kt b/tests/websockets/src/jvmTest/kotlin/CachedSubscriptionTest.kt
index eb2150997a..e18ecb8e10 100644
--- a/tests/websockets/src/jvmTest/kotlin/CachedSubscriptionTest.kt
+++ b/tests/websockets/src/jvmTest/kotlin/CachedSubscriptionTest.kt
@@ -17,8 +17,8 @@ import kotlinx.coroutines.withTimeout
 import org.junit.AfterClass
 import org.junit.BeforeClass
 import org.junit.Test
-import sample.server.TimeQuery
-import sample.server.TimeSubscription
+import sample.server.ValueSharedWithSubscriptionsQuery
+import sample.server.ValueSharedWithSubscriptionsSubscription
 import kotlin.test.assertEquals
 
 class CachedSubscriptionTest {
@@ -53,11 +53,11 @@ class CachedSubscriptionTest {
     runBlocking {
       val channel = Channel<Int>()
       val job = launch {
-        apolloClient.query(TimeQuery())
+        apolloClient.query(ValueSharedWithSubscriptionsQuery())
             .watch()
             // Ignore cache miss
             .filter { it.data != null }
-            .map { it.data!!.time }
+            .map { it.data!!.valueSharedWithSubscriptions }
             .collect {
               channel.send(it)
               println("watcher received: $it")
@@ -67,10 +67,10 @@ class CachedSubscriptionTest {
       assertEquals(0, channel.receive())
 
       println("starting subscription")
-      apolloClient.subscription(TimeSubscription())
+      apolloClient.subscription(ValueSharedWithSubscriptionsSubscription())
           .toFlow()
           .take(3)
-          .map { it.data!!.time }
+          .map { it.data!!.valueSharedWithSubscriptions }
           .collect {
             println("subscription received: $it")
           }
diff --git a/tests/websockets/src/jvmTest/kotlin/WebSocketErrorsTest.kt b/tests/websockets/src/jvmTest/kotlin/WebSocketErrorsTest.kt
index c4c81a2577..eff7aa92af 100644
--- a/tests/websockets/src/jvmTest/kotlin/WebSocketErrorsTest.kt
+++ b/tests/websockets/src/jvmTest/kotlin/WebSocketErrorsTest.kt
@@ -18,7 +18,6 @@ import org.junit.BeforeClass
 import org.junit.Test
 import sample.server.CloseSocketMutation
 import sample.server.CountSubscription
-import sample.server.TimeSubscription
 import kotlin.test.assertEquals
 import kotlin.test.assertIs
 import kotlin.test.assertTrue
@@ -51,7 +50,7 @@ class WebSocketErrorsTest {
         )
         .build()
 
-    apolloClient.subscription(TimeSubscription())
+    apolloClient.subscription(CountSubscription(to = 100, intervalMillis = 100))
         .toFlow()
         .test {
           val error = awaitItem().exception
@@ -76,7 +75,7 @@ class WebSocketErrorsTest {
         )
         .build()
 
-    apolloClient.subscription(TimeSubscription())
+    apolloClient.subscription(CountSubscription(to = 100, intervalMillis = 100))
         .toFlow()
         .test {
           val error = awaitItem().exception
diff --git a/tests/websockets/src/jvmTest/kotlin/incubating/RetryTest.kt b/tests/websockets/src/jvmTest/kotlin/incubating/RetryTest.kt
new file mode 100644
index 0000000000..1980dd7562
--- /dev/null
+++ b/tests/websockets/src/jvmTest/kotlin/incubating/RetryTest.kt
@@ -0,0 +1,136 @@
+package incubating
+
+import app.cash.turbine.test
+import com.apollographql.apollo.sample.server.SampleServer
+import com.apollographql.apollo3.ApolloClient
+import com.apollographql.apollo3.api.Subscription
+import com.apollographql.apollo3.exception.ApolloHttpException
+import com.apollographql.apollo3.exception.ApolloNetworkException
+import com.apollographql.apollo3.interceptor.RetryOnErrorInterceptor
+import com.apollographql.apollo3.mockserver.MockResponse
+import com.apollographql.apollo3.network.ws.incubating.SubscriptionWsProtocol
+import com.apollographql.apollo3.network.ws.incubating.WebSocketNetworkTransport
+import com.apollographql.apollo3.testing.internal.runTest
+import com.apollographql.apollo3.testing.mockServerTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotEquals
+import org.junit.Test
+import sample.server.TagSubscription
+import sample.server.ZeroQuery
+import kotlin.test.assertIs
+
+class RetryTest {
+  @Test
+  fun retryIsWorking() = runTest(skipDelays = false) {
+    var sampleServer = SampleServer(tag = "tag1")
+    ApolloClient.Builder()
+        .serverUrl(sampleServer.graphqlUrl())
+        .retryOnErrorInterceptor(RetryOnErrorInterceptor { it.operation is Subscription })
+        .subscriptionNetworkTransport(
+            WebSocketNetworkTransport.Builder()
+                .serverUrl(sampleServer.subscriptionsUrl())
+                .wsProtocol(SubscriptionWsProtocol { null })
+                .build()
+        )
+        .build().use { apolloClient ->
+          apolloClient.subscription(TagSubscription(intervalMillis = Int.MAX_VALUE))
+              .toFlow()
+              .test {
+                val item1 = awaitItem()
+                assertEquals("tag1", item1.data?.state?.tag)
+
+                // Reuse the port to keep the url unchanged
+                val port = sampleServer.subscriptionsUrl().extractPort()
+                sampleServer.close()
+                sampleServer = SampleServer(port, "tag2")
+
+                val item2 = awaitItem()
+                assertEquals("tag2", item2.data?.state?.tag)
+
+                // The new sub MUST use a different id
+                assertNotEquals(item1.data?.state?.subscriptionId, item2.data?.state?.subscriptionId)
+
+                cancelAndIgnoreRemainingEvents()
+              }
+        }
+    sampleServer.close()
+  }
+
+  @Test
+  fun retryCanBeDisabled() = runTest(skipDelays = false) {
+    val sampleServer = SampleServer(tag = "tag1")
+    ApolloClient.Builder()
+        .serverUrl(sampleServer.graphqlUrl())
+        .retryOnErrorInterceptor(RetryOnErrorInterceptor { it.operation is Subscription })
+        .subscriptionNetworkTransport(
+            WebSocketNetworkTransport.Builder()
+                .serverUrl(sampleServer.subscriptionsUrl())
+                .wsProtocol(SubscriptionWsProtocol { null })
+                .build()
+        )
+        .build().use { apolloClient ->
+          apolloClient.subscription(TagSubscription(intervalMillis = Int.MAX_VALUE))
+              .retryOnError(false)
+              .toFlow()
+              .test {
+                val item1 = awaitItem()
+                assertEquals("tag1", item1.data?.state?.tag)
+
+                sampleServer.close()
+
+                val item2 = awaitItem()
+                assertIs<ApolloNetworkException>(item2.exception)
+                awaitComplete()
+              }
+        }
+  }
+
+  @Test
+  fun subscriptionsAreNotRetriedByDefault() = runTest(skipDelays = false) {
+    val sampleServer = SampleServer(tag = "tag1")
+    ApolloClient.Builder()
+        .serverUrl(sampleServer.graphqlUrl())
+        .subscriptionNetworkTransport(
+            WebSocketNetworkTransport.Builder()
+                .serverUrl(sampleServer.subscriptionsUrl())
+                .wsProtocol(SubscriptionWsProtocol { null })
+                .build()
+        )
+        .build().use { apolloClient ->
+          apolloClient.subscription(TagSubscription(intervalMillis = Int.MAX_VALUE))
+              .retryOnError(false)
+              .toFlow()
+              .test {
+                val item1 = awaitItem()
+                assertEquals("tag1", item1.data?.state?.tag)
+
+                sampleServer.close()
+
+                val item2 = awaitItem()
+                assertIs<ApolloNetworkException>(item2.exception)
+                awaitComplete()
+              }
+        }
+  }
+
+  @Test
+  fun queriesAreNotRetriedWhenRetrySubscriptionsIsTrue() = mockServerTest(
+      clientBuilder = {
+        retryOnErrorInterceptor(RetryOnErrorInterceptor { it.operation is Subscription })
+      },
+      skipDelays = false
+  ) {
+    mockServer.enqueue(MockResponse.Builder().statusCode(500).build())
+
+    apolloClient.query(ZeroQuery())
+        .toFlow()
+        .test {
+          assertIs<ApolloHttpException>(awaitItem().exception)
+          awaitComplete()
+        }
+  }
+}
+
+private fun String.extractPort(): Int {
+  return Regex("[a-z]*://[a-zA-Z0-9.]*:([0-9]*)").matchAt(this, 0)?.groupValues?.get(1)?.toInt() ?: error("No port found in $this")
+}
\ No newline at end of file
diff --git a/tests/websockets/src/jvmTest/kotlin/incubating/SampleServerTest.kt b/tests/websockets/src/jvmTest/kotlin/incubating/SampleServerTest.kt
index cccdeab6d2..5d3c0a41e1 100644
--- a/tests/websockets/src/jvmTest/kotlin/incubating/SampleServerTest.kt
+++ b/tests/websockets/src/jvmTest/kotlin/incubating/SampleServerTest.kt
@@ -5,6 +5,7 @@ import com.apollographql.apollo3.ApolloClient
 import com.apollographql.apollo3.exception.SubscriptionOperationException
 import com.apollographql.apollo3.network.ws.incubating.SubscriptionWsProtocol
 import com.apollographql.apollo3.network.ws.incubating.WebSocketNetworkTransport
+import com.apollographql.apollo3.testing.internal.runTest
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.map
@@ -12,7 +13,6 @@ import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.single
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 import org.junit.Test
 import sample.server.CountSubscription
 import sample.server.OperationErrorSubscription
@@ -23,7 +23,7 @@ class SampleServerTest {
   private class Scope(val apolloClient: ApolloClient, val sampleServer: SampleServer, val coroutineScope: CoroutineScope)
 
   private fun test(customizeTransport: WebSocketNetworkTransport.Builder.() -> Unit = {}, block: suspend Scope.() -> Unit) {
-    runBlocking {
+    runTest {
       SampleServer().use { sampleServer ->
         ApolloClient.Builder()
             .serverUrl(sampleServer.graphqlUrl())
@@ -35,7 +35,7 @@ class SampleServerTest {
                     .build()
             )
             .build().use { apolloClient ->
-              Scope(apolloClient, sampleServer, this@runBlocking).block()
+              Scope(apolloClient, sampleServer, this@runTest).block()
             }
       }
     }