diff --git a/modules/cache-material/src/caching_cryptographic_materials_decorators.ts b/modules/cache-material/src/caching_cryptographic_materials_decorators.ts
index 5be6e144b..7e9e531f7 100644
--- a/modules/cache-material/src/caching_cryptographic_materials_decorators.ts
+++ b/modules/cache-material/src/caching_cryptographic_materials_decorators.ts
@@ -21,6 +21,7 @@ import {
   Entry,
 } from './cryptographic_materials_cache'
 import { CryptographicMaterialsCacheKeyHelpersInterface } from './build_cryptographic_materials_cache_key_helpers'
+import { SHARED_REQUESTS, batchesByCache } from './shared_requests'
 
 export function decorateProperties<S extends SupportedAlgorithmSuites>(
   obj: CachingMaterialsManager<S>,
@@ -102,50 +103,139 @@ export function getEncryptionMaterials<S extends SupportedAlgorithmSuites>({
       suite,
       encryptionContext,
     })
-    const entry = this._cache.getEncryptionMaterial(cacheKey, plaintextLength)
-    /* Check for early return (Postcondition): If I have a valid EncryptionMaterial, return it. */
-    if (entry && !this._cacheEntryHasExceededLimits(entry)) {
-      return cloneResponse(entry.response)
-    } else {
-      this._cache.del(cacheKey)
-    }
+    const fetch = async () =>
+      this._backingMaterialsManager
+        /* The cache reuses this data key for other messages,
+         * so don't pass this message's plaintextLength to the backing materials manager.
+         */
+        .getEncryptionMaterials({ suite, encryptionContext, commitmentPolicy })
 
-    const material = await this._backingMaterialsManager
-      /* Strip any information about the plaintext from the backing request,
-       * because the resulting response may be used to encrypt multiple plaintexts.
-       */
-      .getEncryptionMaterials({ suite, encryptionContext, commitmentPolicy })
+    // If the request this caller waited on fails, it tries once more, and a second failure throws.
+    // `retry` is false on that second try.
+    const getShared = async (
+      retry: boolean
+    ): Promise<EncryptionMaterial<S>> => {
+      /* Check for early return (Postcondition): If I have a valid EncryptionMaterial, return it. */
+      const entry = this._cache.getEncryptionMaterial(cacheKey, plaintextLength)
+      if (entry && !this._cacheEntryHasExceededLimits(entry)) {
+        return cloneResponse(entry.response)
+      } else {
+        this._cache.del(cacheKey)
+      }
 
-    /* Check for early return (Postcondition): If I can not cache the EncryptionMaterial, just return it. */
-    if (!material.suite.cacheSafe) return material
+      // Another caller may already be requesting a data key for this cache entry.
+      // Wait for that one if it can also encrypt this message without going over the limits.
+      const joinable = findBatch<EncryptionMaterial<S>>(
+        this._cache,
+        cacheKey,
+        (b) => canJoin(this, b, plaintextLength)
+      )
+      if (joinable) {
+        const outcome = await joinBatch(joinable, plaintextLength)
+        if ('material' in outcome) return outcome.material
+        if ('error' in outcome) {
+          if (!retry) throw outcome.error
+          // The caller that made the failed request already removed it,
+          // so the first waiting caller to get here starts a new request and the others wait on that one.
+          return getShared(false)
+        }
+        // The other caller's data key isn't cache safe, so this message can't use it.
+        // Make a separate request.
+        return cacheEncryptionMaterial(
+          this,
+          cacheKey,
+          await fetch(),
+          plaintextLength
+        )
+      }
 
-    /* It is possible for an entry to exceed limits immediately.
-     * The simplest case is to need to encrypt more than then maxBytesEncrypted.
-     * In this case, I return the response to encrypt the data,
-     * but do not put a know invalid item into the cache.
-     */
-    const testEntry = {
-      response: material,
-      now: Date.now(),
-      messagesEncrypted: 1,
-      bytesEncrypted: plaintextLength,
-    }
-    if (!this._cacheEntryHasExceededLimits(testEntry)) {
-      this._cache.putEncryptionMaterial(
+      const batch = startBatch<EncryptionMaterial<S>>(
+        this._cache,
         cacheKey,
-        material,
-        plaintextLength,
-        this._maxAge
+        this,
+        plaintextLength
       )
-      return cloneResponse(material)
-    } else {
-      /* Postcondition: If the material has exceeded limits it MUST NOT be cloned.
-       * If it is cloned, and the clone is returned,
-       * then there exist a copy of the unencrypted data key.
-       * It is true that this data would be caught by GC, it is better to just not rely on that.
-       */
-      return material
+      let release: Outcome<EncryptionMaterial<S>> = { unshareable: true }
+      try {
+        const material = await fetch()
+
+        // Not cache safe: return it without caching.
+        // The waiting callers can't use it, so each makes its own request.
+        if (!material.suite.cacheSafe) return material
+
+        const used = {
+          response: material,
+          now: Date.now(),
+          messagesEncrypted: batch.messages,
+          bytesEncrypted: batch.bytes,
+        }
+        // This is true only if this message alone is larger than maxBytesEncrypted,
+        // because canJoin turns away any caller that would go over the limits.
+        // So no one is waiting; return the data key without caching it.
+        if (this._cacheEntryHasExceededLimits(used)) return material
+
+        this._cache.putEncryptionMaterial(
+          cacheKey,
+          material,
+          plaintextLength,
+          this._maxAge
+        )
+        // getEncryptionMaterial also adds to the entry's message and byte counts.
+        // Call it once per waiting caller so the cached entry counts their messages too.
+        for (const member of batch.members) {
+          this._cache.getEncryptionMaterial(cacheKey, member.plaintextLength)
+        }
+        for (const member of batch.members) {
+          member.settle({ material: cloneResponse(material) })
+        }
+        return cloneResponse(material)
+      } catch (error) {
+        release = { error }
+        throw error
+      } finally {
+        // Tell the callers still waiting how this request ended:
+        // - on an error, they retry once
+        // - if they can't use this data key, they each make their own request
+        // After a success they already have the data key, so this does nothing.
+        closeBatch(this._cache, cacheKey, batch)
+        settleMembers(batch, release)
+      }
     }
+    return getShared(true)
+  }
+}
+
+function cacheEncryptionMaterial<S extends SupportedAlgorithmSuites>(
+  cmm: CachingMaterialsManager<S>,
+  cacheKey: string,
+  material: EncryptionMaterial<S>,
+  plaintextLength: number
+): EncryptionMaterial<S> {
+  /* Check for early return (Postcondition): If I can not cache the EncryptionMaterial, just return it. */
+  if (!material.suite.cacheSafe) return material
+
+  const testEntry = {
+    response: material,
+    now: Date.now(),
+    messagesEncrypted: 1,
+    bytesEncrypted: plaintextLength,
+  }
+  /* One message can be larger than maxBytesEncrypted by itself.
+   * Return the data key for it, but don't cache it.
+   */
+  if (!cmm._cacheEntryHasExceededLimits(testEntry)) {
+    cmm._cache.putEncryptionMaterial(
+      cacheKey,
+      material,
+      plaintextLength,
+      cmm._maxAge
+    )
+    return cloneResponse(material)
+  } else {
+    /* Postcondition: If the material has exceeded limits it MUST NOT be cloned.
+     * A clone would leave a second copy of the unencrypted data key for GC to clean up.
+     */
+    return material
   }
 }
 
@@ -169,21 +259,175 @@ export function decryptMaterials<S extends SupportedAlgorithmSuites>({
       this._partition,
       request
     )
-    const entry = this._cache.getDecryptionMaterial(cacheKey)
-    /* Check for early return (Postcondition): If I have a valid DecryptionMaterial, return it. */
-    if (entry && !this._cacheEntryHasExceededLimits(entry)) {
-      return cloneResponse(entry.response)
-    } else {
-      this._cache.del(cacheKey)
+    const fetchAndCache = async () => {
+      const material = await this._backingMaterialsManager.decryptMaterials(
+        request
+      )
+      this._cache.putDecryptionMaterial(cacheKey, material, this._maxAge)
+      return material
     }
 
-    const material = await this._backingMaterialsManager.decryptMaterials(
-      request
-    )
+    // If the request this caller waited on fails, it tries once more, and a second failure throws.
+    // `retry` is false on that second try.
+    const getShared = async (
+      retry: boolean
+    ): Promise<DecryptionMaterial<S>> => {
+      /* Check for early return (Postcondition): If I have a valid DecryptionMaterial, return it. */
+      const entry = this._cache.getDecryptionMaterial(cacheKey)
+      if (entry && !this._cacheEntryHasExceededLimits(entry)) {
+        return cloneResponse(entry.response)
+      } else {
+        this._cache.del(cacheKey)
+      }
 
-    this._cache.putDecryptionMaterial(cacheKey, material, this._maxAge)
-    return cloneResponse(material)
+      const joinable = findBatch<DecryptionMaterial<S>>(
+        this._cache,
+        cacheKey,
+        isRecent
+      )
+      if (joinable) {
+        const outcome = await joinBatch(joinable, 0)
+        if ('material' in outcome) return outcome.material
+        if ('error' in outcome && !retry) throw outcome.error
+        // The caller that made the failed request already removed it,
+        // so the first waiting caller to get here starts a new request and the others wait on that one.
+        return getShared(false)
+      }
+
+      const batch = startBatch<DecryptionMaterial<S>>(
+        this._cache,
+        cacheKey,
+        this,
+        0
+      )
+      let release: Outcome<DecryptionMaterial<S>> = { unshareable: true }
+      try {
+        const material = await fetchAndCache()
+        for (const member of batch.members) {
+          member.settle({ material: cloneResponse(material) })
+        }
+        return cloneResponse(material)
+      } catch (error) {
+        release = { error }
+        throw error
+      } finally {
+        // If the request failed, tell the callers still waiting so they retry once.
+        // After a success they already have the material, so this does nothing.
+        closeBatch(this._cache, cacheKey, batch)
+        settleMembers(batch, release)
+      }
+    }
+    return getShared(true)
+  }
+}
+
+// One request to the backing materials manager, and the callers waiting on it.
+interface Batch<M> {
+  startedAt: number
+  // Messages and bytes the data key will encrypt,
+  // counting the caller that made the request and every caller waiting on it.
+  // canJoin checks these against the limits before the data key exists.
+  messages: number
+  bytes: number
+  requester: CachingMaterialsManager<any>
+  members: { plaintextLength: number; settle: (outcome: Outcome<M>) => void }[]
+}
+
+// What a waiting caller gets back:
+// - the material
+// - the request's error
+// - `unshareable`, when the data key can't be used for other messages
+type Outcome<M> = { material: M } | { error: unknown } | { unshareable: true }
+
+function findBatch<M>(
+  cache: CryptographicMaterialsCache<any>,
+  cacheKey: string,
+  predicate: (batch: Batch<M>) => boolean
+): Batch<M> | undefined {
+  const batches = batchesByCache.get(cache)?.get(cacheKey)
+  return batches?.find(predicate)
+}
+
+function isRecent(batch: Batch<any>) {
+  return Date.now() < batch.startedAt + SHARED_REQUESTS.graceInterval
+}
+
+/* The request may come from another caching materials manager that shares this cache.
+ * Join only if this caller's message stays within both managers' limits.
+ */
+function canJoin<S extends SupportedAlgorithmSuites>(
+  cmm: CachingMaterialsManager<S>,
+  batch: Batch<any>,
+  plaintextLength: number
+) {
+  if (!isRecent(batch)) return false
+  const next = {
+    response: undefined as any,
+    now: Date.now(),
+    messagesEncrypted: batch.messages + 1,
+    bytesEncrypted: batch.bytes + plaintextLength,
   }
+  return (
+    !cmm._cacheEntryHasExceededLimits(next) &&
+    !batch.requester._cacheEntryHasExceededLimits(next)
+  )
+}
+
+/* Call this right after findBatch finds nothing, with no await in between.
+ * Otherwise callers arriving at the same time each start their own request.
+ */
+function startBatch<M>(
+  cache: CryptographicMaterialsCache<any>,
+  cacheKey: string,
+  requester: CachingMaterialsManager<any>,
+  plaintextLength: number
+): Batch<M> {
+  const byKey = batchesByCache.get(cache) || new Map<string, Batch<any>[]>()
+  batchesByCache.set(cache, byKey)
+  // A request that hangs never removes itself.
+  // This drops it when the next request for this key starts.
+  const batches = (byKey.get(cacheKey) || []).filter(isRecent)
+  byKey.set(cacheKey, batches)
+  const batch: Batch<M> = {
+    startedAt: Date.now(),
+    messages: 1,
+    bytes: plaintextLength,
+    requester,
+    members: [],
+  }
+  batches.push(batch)
+  return batch
+}
+
+function closeBatch<M>(
+  cache: CryptographicMaterialsCache<any>,
+  cacheKey: string,
+  batch: Batch<M>
+) {
+  const byKey = batchesByCache.get(cache)
+  const batches = byKey?.get(cacheKey)
+  if (!byKey || !batches) return
+  const index = batches.indexOf(batch)
+  if (index !== -1) batches.splice(index, 1)
+  if (!batches.length) byKey.delete(cacheKey)
+}
+
+/* Counts this caller's message against the request now, before the data key exists,
+ * so canJoin sees it when the next caller arrives.
+ */
+async function joinBatch<M>(
+  batch: Batch<M>,
+  plaintextLength: number
+): Promise<Outcome<M>> {
+  batch.messages += 1
+  batch.bytes += plaintextLength
+  return new Promise<Outcome<M>>((settle) =>
+    batch.members.push({ plaintextLength, settle })
+  )
+}
+
+function settleMembers<M>(batch: Batch<M>, outcome: Outcome<M>) {
+  for (const member of batch.members) member.settle(outcome)
 }
 
 export function cacheEntryHasExceededLimits<
diff --git a/modules/cache-material/src/shared_requests.ts b/modules/cache-material/src/shared_requests.ts
new file mode 100644
index 000000000..553e93576
--- /dev/null
+++ b/modules/cache-material/src/shared_requests.ts
@@ -0,0 +1,21 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { CryptographicMaterialsCache } from './cryptographic_materials_cache'
+
+/* Concurrent cache misses for the same entry share one request to the backing materials manager.
+ * If that request fails, its waiting callers share one retry,
+ * so a failing backing materials manager gets one more call, not one per waiting caller.
+ */
+
+export const SHARED_REQUESTS = {
+  // A request older than this may be stuck, so new callers don't join it.
+  graceInterval: 1000, // milliseconds
+}
+
+// Requests in progress, by cache and then by cache key.
+// Keyed by cache, so caching materials managers that share a cache also share requests.
+export const batchesByCache = new WeakMap<
+  CryptographicMaterialsCache<any>,
+  Map<string, any[]>
+>()
diff --git a/modules/cache-material/test/caching_cryptographic_materials_decorators.concurrency.test.ts b/modules/cache-material/test/caching_cryptographic_materials_decorators.concurrency.test.ts
new file mode 100644
index 000000000..fbfdcb6b8
--- /dev/null
+++ b/modules/cache-material/test/caching_cryptographic_materials_decorators.concurrency.test.ts
@@ -0,0 +1,411 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+/* eslint-env mocha */
+
+import { expect } from 'chai'
+import {
+  cacheEntryHasExceededLimits,
+  decorateProperties,
+  getEncryptionMaterials,
+  decryptMaterials,
+} from '../src/caching_cryptographic_materials_decorators'
+import { SHARED_REQUESTS, batchesByCache } from '../src/shared_requests'
+import { getLocalCryptographicMaterialsCache } from '../src/get_local_cryptographic_materials_cache'
+import { buildCryptographicMaterialsCacheKeyHelpers } from '../src/build_cryptographic_materials_cache_key_helpers'
+import { createHash, randomBytes } from 'crypto'
+import {
+  NodeAlgorithmSuite,
+  AlgorithmSuiteIdentifier,
+  KeyringTraceFlag,
+  EncryptedDataKey,
+  NodeEncryptionMaterial,
+  NodeDecryptionMaterial,
+  unwrapDataKey,
+} from '@aws-crypto/material-management'
+
+const suite = new NodeAlgorithmSuite(
+  AlgorithmSuiteIdentifier.ALG_AES128_GCM_IV12_TAG16_HKDF_SHA256
+)
+const uncacheableSuite = new NodeAlgorithmSuite(
+  AlgorithmSuiteIdentifier.ALG_AES128_GCM_IV12_TAG16
+)
+const edk = new EncryptedDataKey({
+  providerId: 'keyNamespace',
+  providerInfo: 'keyName',
+  encryptedDataKey: new Uint8Array([1]),
+})
+const trace = { keyNamespace: 'keyNamespace', keyName: 'keyName' }
+
+const cacheKeyHelpers = buildCryptographicMaterialsCacheKeyHelpers(
+  (input: string) => Buffer.from(input, 'utf8'),
+  (input: Uint8Array) => Buffer.from(input).toString('utf8'),
+  async (...data: (Uint8Array | string)[]) =>
+    data
+      .map((item) =>
+        typeof item === 'string' ? Buffer.from(item, 'hex') : item
+      )
+      .reduce((hash, item) => hash.update(item), createHash('sha512'))
+      .digest()
+)
+
+// A short grace interval, so the hang tests run in milliseconds.
+const FAST = { graceInterval: 20 }
+const DEFAULT = { ...SHARED_REQUESTS }
+
+// A backing materials manager that takes 5 ms per request and returns a new data key each time.
+// `outcomes` decides each call in order: 'ok', 'fail', or 'hang'; later calls use the last one.
+function slowBackingMaterialsManager({
+  outcomes = ['ok'] as ('ok' | 'fail' | 'hang')[],
+  materialSuite = suite,
+} = {}) {
+  const calls = { encrypt: 0, decrypt: 0, active: 0, peak: 0 }
+  let n = 0
+  const respond = async () => {
+    const outcome = outcomes[Math.min(n++, outcomes.length - 1)]
+    if (outcome === 'hang') await new Promise<never>(() => undefined)
+    calls.active += 1
+    calls.peak = Math.max(calls.peak, calls.active)
+    await new Promise((resolve) => setTimeout(resolve, 5))
+    calls.active -= 1
+    if (outcome === 'fail')
+      throw new Error('backing materials manager unavailable')
+  }
+  return {
+    calls,
+    async getEncryptionMaterials() {
+      calls.encrypt += 1
+      await respond()
+      return new NodeEncryptionMaterial(materialSuite, {})
+        .setUnencryptedDataKey(randomBytes(16), {
+          ...trace,
+          flags: KeyringTraceFlag.WRAPPING_KEY_GENERATED_DATA_KEY,
+        })
+        .addEncryptedDataKey(
+          edk,
+          KeyringTraceFlag.WRAPPING_KEY_ENCRYPTED_DATA_KEY
+        )
+    },
+    async decryptMaterials() {
+      calls.decrypt += 1
+      await respond()
+      return new NodeDecryptionMaterial(suite, {}).setUnencryptedDataKey(
+        randomBytes(16),
+        { ...trace, flags: KeyringTraceFlag.WRAPPING_KEY_DECRYPTED_DATA_KEY }
+      )
+    },
+  }
+}
+
+function cachingCMM(
+  backing: any,
+  {
+    maxMessagesEncrypted = 1000,
+    maxBytesEncrypted = 1000,
+    cache = getLocalCryptographicMaterialsCache(100) as any,
+  } = {}
+) {
+  const cmm = {
+    _cacheEntryHasExceededLimits: cacheEntryHasExceededLimits(),
+    getEncryptionMaterials: getEncryptionMaterials(cacheKeyHelpers),
+    decryptMaterials: decryptMaterials(cacheKeyHelpers),
+  } as any
+  decorateProperties(cmm, {
+    cache,
+    backingMaterials: backing,
+    backingMaterialsManager: backing,
+    partition: 'partition',
+    maxAge: 1000 * 60,
+    maxBytesEncrypted,
+    maxMessagesEncrypted,
+  })
+  return cmm
+}
+
+const encryptRequest = { suite, encryptionContext: {}, plaintextLength: 1 }
+const decryptRequest = {
+  suite,
+  encryptionContext: {},
+  encryptedDataKeys: [edk],
+}
+
+function startEncrypts(cmm: any, lengths: number[], request: any = {}) {
+  return lengths.map(async (plaintextLength) =>
+    cmm.getEncryptionMaterials({
+      ...encryptRequest,
+      ...request,
+      plaintextLength,
+    })
+  )
+}
+
+async function encryptConcurrently(cmm: any, lengths: number[], request?: any) {
+  return Promise.all(startEncrypts(cmm, lengths, request))
+}
+
+const times = (count: number, length = 1) =>
+  Array.from({ length: count }, () => length)
+
+// Messages and bytes encrypted under each distinct data key.
+function usageByDataKey(materials: any[], lengths: number[]) {
+  const usage = new Map<string, { messages: number; bytes: number }>()
+  materials.forEach((material, i) => {
+    const dataKey = Buffer.from(
+      unwrapDataKey(material.getUnencryptedDataKey())
+    ).toString('hex')
+    const entry = usage.get(dataKey) || { messages: 0, bytes: 0 }
+    entry.messages += 1
+    entry.bytes += lengths[i]
+    usage.set(dataKey, entry)
+  })
+  return [...usage.values()]
+}
+
+describe('caching materials manager: concurrent cache misses', () => {
+  beforeEach(() => Object.assign(SHARED_REQUESTS, FAST))
+  afterEach(() => Object.assign(SHARED_REQUESTS, DEFAULT))
+
+  it('concurrent encrypts share data keys without exceeding maxMessagesEncrypted', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cmm = cachingCMM(backing, { maxMessagesEncrypted: 5 })
+
+    const materials = await encryptConcurrently(cmm, times(10))
+
+    expect(backing.calls.encrypt).to.equal(2)
+    expect(backing.calls.peak).to.equal(2)
+    expect(
+      usageByDataKey(materials, times(10)).map((u) => u.messages)
+    ).to.deep.equal([5, 5])
+
+    // The cached data key already encrypted 5 messages, so the next one needs a new data key.
+    await cmm.getEncryptionMaterials(encryptRequest)
+    expect(backing.calls.encrypt).to.equal(3)
+  })
+
+  it('callers stop joining a request once it would exceed the limits of the manager that made it', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cache = getLocalCryptographicMaterialsCache(100)
+    const strict = cachingCMM(backing, { cache, maxMessagesEncrypted: 2 })
+    const loose = cachingCMM(backing, { cache, maxMessagesEncrypted: 10 })
+
+    // strict's data key can encrypt only 2 messages, so loose's 5 callers share a second request.
+    await Promise.all([
+      ...startEncrypts(strict, times(2)),
+      ...startEncrypts(loose, times(5)),
+    ])
+
+    expect(backing.calls.encrypt).to.equal(2)
+  })
+
+  it('concurrent encrypts share data keys without exceeding maxBytesEncrypted', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cmm = cachingCMM(backing, { maxBytesEncrypted: 10 })
+
+    const materials = await encryptConcurrently(cmm, times(10, 3))
+
+    expect(backing.calls.encrypt).to.equal(4)
+    expect(
+      usageByDataKey(materials, times(10, 3)).map((u) => u.bytes)
+    ).to.deep.equal([9, 9, 9, 3])
+  })
+
+  it('a warm entry serves callers until its limit, then one request serves the rest', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cmm = cachingCMM(backing, { maxMessagesEncrypted: 5 })
+    const warm = await cmm.getEncryptionMaterials(encryptRequest)
+
+    const materials = await encryptConcurrently(cmm, times(10))
+
+    expect(backing.calls.encrypt).to.equal(3)
+    for (const { messages } of usageByDataKey(
+      [warm, ...materials],
+      times(11)
+    )) {
+      expect(messages).to.be.at.most(5)
+    }
+  })
+
+  it('concurrent decrypts share one backing request', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cmm = cachingCMM(backing)
+
+    await Promise.all(
+      Array.from({ length: 10 }, async () =>
+        cmm.decryptMaterials(decryptRequest)
+      )
+    )
+
+    expect(backing.calls.decrypt).to.equal(1)
+  })
+
+  it('serves every waiting caller even when the cache evicts the entry at once', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cache = getLocalCryptographicMaterialsCache(1)
+    // A cache that deletes each entry one microtask after it is put.
+    const evictingCache = {
+      ...cache,
+      putEncryptionMaterial(...args: [string, any, number, number?]) {
+        cache.putEncryptionMaterial(...args)
+        queueMicrotask(() => cache.del(args[0]))
+      },
+    }
+    const cmm = cachingCMM(backing, { cache: evictingCache })
+
+    const materials = await encryptConcurrently(cmm, times(10))
+
+    expect(materials).to.have.lengthOf(10)
+    expect(backing.calls.encrypt).to.equal(1)
+  })
+
+  it('waiting callers share one retry when the shared request fails', async () => {
+    const backing = slowBackingMaterialsManager({ outcomes: ['fail', 'ok'] })
+    const cmm = cachingCMM(backing)
+
+    const results = await Promise.allSettled(startEncrypts(cmm, times(10)))
+
+    // Only the caller whose request failed sees the error.
+    expect(results.filter((r) => r.status === 'rejected')).to.have.lengthOf(1)
+    expect(backing.calls.encrypt).to.equal(2)
+  })
+
+  it('waiting callers get the error when the shared retry also fails', async () => {
+    const backing = slowBackingMaterialsManager({ outcomes: ['fail'] })
+    const cmm = cachingCMM(backing)
+
+    const results = await Promise.allSettled(startEncrypts(cmm, times(10)))
+
+    for (const result of results) expect(result.status).to.equal('rejected')
+    expect(backing.calls.encrypt).to.equal(2)
+  })
+
+  it('waiting decrypts share one retry when the shared request fails', async () => {
+    const backing = slowBackingMaterialsManager({ outcomes: ['fail', 'ok'] })
+    const cmm = cachingCMM(backing)
+
+    const results = await Promise.allSettled(
+      Array.from({ length: 5 }, async () =>
+        cmm.decryptMaterials(decryptRequest)
+      )
+    )
+
+    expect(results.filter((r) => r.status === 'rejected')).to.have.lengthOf(1)
+    expect(backing.calls.decrypt).to.equal(2)
+  })
+
+  it('starts another request for new callers once one hangs past graceInterval', async () => {
+    const backing = slowBackingMaterialsManager({ outcomes: ['hang', 'ok'] })
+    const cmm = cachingCMM(backing)
+
+    startEncrypts(cmm, times(1))
+    await new Promise((resolve) => setTimeout(resolve, FAST.graceInterval + 5))
+    const materials = await encryptConcurrently(cmm, times(5))
+
+    expect(materials).to.have.lengthOf(5)
+    expect(backing.calls.encrypt).to.equal(2)
+  })
+
+  it('releases every waiting caller when the response is malformed', async () => {
+    // Without a suite, reading the response throws before the waiting callers get it.
+    const malformed = {
+      async getEncryptionMaterials() {
+        await new Promise((resolve) => setTimeout(resolve, 5))
+        return {}
+      },
+      async decryptMaterials() {
+        await new Promise((resolve) => setTimeout(resolve, 5))
+        return {}
+      },
+    }
+    const cmm = cachingCMM(malformed)
+
+    const encrypts = await Promise.allSettled(startEncrypts(cmm, times(3)))
+    const decrypts = await Promise.allSettled(
+      Array.from({ length: 3 }, async () =>
+        cmm.decryptMaterials(decryptRequest)
+      )
+    )
+
+    for (const result of [...encrypts, ...decrypts]) {
+      expect(result.status).to.equal('rejected')
+    }
+  })
+
+  it('forgets a cache key once its requests finish', async () => {
+    const backing = slowBackingMaterialsManager()
+    const cache = getLocalCryptographicMaterialsCache(100)
+    const cmm = cachingCMM(backing, { cache, maxMessagesEncrypted: 2 })
+
+    await encryptConcurrently(cmm, times(5))
+    await Promise.all(
+      Array.from({ length: 5 }, async () =>
+        cmm.decryptMaterials(decryptRequest)
+      )
+    )
+    await Promise.allSettled(
+      startEncrypts(
+        cachingCMM(slowBackingMaterialsManager({ outcomes: ['fail'] }), {
+          cache,
+        }),
+        times(3),
+        { encryptionContext: { a: 'b' } }
+      )
+    )
+
+    expect(batchesByCache.get(cache)?.size).to.equal(0)
+  })
+
+  it('forgets a request that never settles once a newer one starts', async () => {
+    const backing = slowBackingMaterialsManager({ outcomes: ['hang', 'ok'] })
+    const cache = getLocalCryptographicMaterialsCache(100)
+    const cmm = cachingCMM(backing, { cache })
+
+    startEncrypts(cmm, times(1))
+    await new Promise((resolve) => setTimeout(resolve, FAST.graceInterval + 5))
+    await encryptConcurrently(cmm, times(3))
+
+    expect(batchesByCache.get(cache)?.size).to.equal(0)
+  })
+
+  describe('requests run in parallel when the response cannot serve another caller', () => {
+    const uncached = {
+      'plaintextLength exceeds maxBytesEncrypted': {
+        backing: {},
+        cmm: { maxBytesEncrypted: 10 },
+        length: 11,
+        request: {},
+        peak: 10,
+      },
+      'the backing suite is not cache safe': {
+        backing: { materialSuite: uncacheableSuite },
+        cmm: {},
+        length: 1,
+        request: { suite: undefined },
+        // The first response reveals the suite is not cache safe,
+        // so the other 9 requests start after it, in parallel.
+        peak: 9,
+      },
+    }
+
+    for (const [
+      name,
+      { backing: options, cmm: limits, length, request, peak },
+    ] of Object.entries(uncached)) {
+      it(name, async () => {
+        const backing = slowBackingMaterialsManager(options)
+        const cmm = cachingCMM(backing, limits)
+
+        const materials = await encryptConcurrently(
+          cmm,
+          times(10, length),
+          request
+        )
+
+        expect(backing.calls.encrypt).to.equal(10)
+        expect(backing.calls.peak).to.equal(peak)
+        expect(usageByDataKey(materials, times(10, length))).to.have.lengthOf(
+          10
+        )
+      })
+    }
+  })
+})
diff --git a/modules/kms-keyring-node/src/kms_hkeyring_node_helpers.ts b/modules/kms-keyring-node/src/kms_hkeyring_node_helpers.ts
index 0c182f4a7..def5f9086 100644
--- a/modules/kms-keyring-node/src/kms_hkeyring_node_helpers.ts
+++ b/modules/kms-keyring-node/src/kms_hkeyring_node_helpers.ts
@@ -34,6 +34,10 @@ import {
 } from './constants'
 import { BranchKeyIdSupplier } from '@aws-crypto/kms-keyring'
 import { serializeFactory, uuidv4Factory } from '@aws-crypto/serialize'
+import {
+  deepCopyBranchKeyMaterial,
+  sharedFetch,
+} from './shared_branch_key_fetches'
 
 export const stringToUtf8Bytes = (input: string): Buffer =>
   Buffer.from(input, 'utf-8')
@@ -194,38 +198,43 @@ export async function getBranchKeyMaterials(
 ): Promise<NodeBranchKeyMaterial> {
   const { keyStore, cacheLimitTtl } = hKeyring
 
-  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
-  //# The hierarchical keyring MUST attempt to find [branch key materials](../structures.md#branch-key-materials)
-  //# from the underlying [cryptographic materials cache](../local-cryptographic-materials-cache.md).
-  const cacheEntry = cmc.getBranchKeyMaterial(cacheEntryId)
-  let branchKeyMaterials: NodeBranchKeyMaterial
-  // if the cache entry is false, branch key materials were not found
-  if (!cacheEntry || hKeyring.cacheEntryHasExceededLimits(cacheEntry)) {
+  // The cached entry, or undefined if it's missing or past this keyring's TTL.
+  const lookUp = () => {
     //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
-    //# If this is NOT true, then we MUST treat the cache entry as expired.
-
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#ondecrypt
-    //# If this is NOT true, then we MUST treat the cache entry as expired.
+    //# The hierarchical keyring MUST attempt to find [branch key materials](../structures.md#branch-key-materials)
+    //# from the underlying [cryptographic materials cache](../local-cryptographic-materials-cache.md).
+    const cacheEntry = cmc.getBranchKeyMaterial(cacheEntryId)
+    if (!cacheEntry || hKeyring.cacheEntryHasExceededLimits(cacheEntry)) {
+      //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
+      //# If this is NOT true, then we MUST treat the cache entry as expired.
+
+      //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#ondecrypt
+      //# If this is NOT true, then we MUST treat the cache entry as expired.
+      return undefined
+    }
+    return cacheEntry
+  }
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
-    //# If a cache entry is not found or the cache entry is expired, the hierarchical keyring MUST attempt to obtain the branch key materials
-    //# by querying the backing branch keystore specified in the [retrieve OnEncrypt branch key materials](#query-branch-keystore-onencrypt) section.
-    //# If the keyring is not able to retrieve [branch key materials](../structures.md#branch-key-materials)
-    //# through the underlying cryptographic materials cache or
-    //# it no longer has access to them through the backing keystore, OnEncrypt MUST fail.
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
+  //# If a cache entry is not found or the cache entry is expired, the hierarchical keyring MUST attempt to obtain the branch key materials
+  //# by querying the backing branch keystore specified in the [retrieve OnEncrypt branch key materials](#query-branch-keystore-onencrypt) section.
+  //# If the keyring is not able to retrieve [branch key materials](../structures.md#branch-key-materials)
+  //# through the underlying cryptographic materials cache or
+  //# it no longer has access to them through the backing keystore, OnEncrypt MUST fail.
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
-    //# Otherwise, OnEncrypt MUST fail.
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
+  //# Otherwise, OnEncrypt MUST fail.
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
-    //# Otherwise, OnDecrypt MUST fail.
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
+  //# Otherwise, OnDecrypt MUST fail.
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
-    //# OnEncrypt MUST call the Keystore's [GetActiveBranchKey](../branch-key-store.md#getactivebranchkey) operation with the following inputs:
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
+  //# OnEncrypt MUST call the Keystore's [GetActiveBranchKey](../branch-key-store.md#getactivebranchkey) operation with the following inputs:
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
-    //# OnDecrypt MUST call the Keystore's [GetBranchKeyVersion](../branch-key-store.md#getbranchkeyversion) operation with the following inputs:
-    branchKeyMaterials = branchKeyVersion
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
+  //# OnDecrypt MUST call the Keystore's [GetBranchKeyVersion](../branch-key-store.md#getbranchkeyversion) operation with the following inputs:
+  const fetch = async () =>
+    branchKeyVersion
       ? await keyStore.getBranchKeyVersion(branchKeyId, branchKeyVersion)
       : // The complice needs a line
         //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
@@ -233,41 +242,28 @@ export async function getBranchKeyMaterials(
         //# - the `branchKeyId` used in this operation
         await keyStore.getActiveBranchKey(branchKeyId)
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
-    //# If the Keystore's GetActiveBranchKey operation succeeds
-    //# the keyring MUST put the returned branch key materials in the cache using the
-    //# formula defined in [Appendix A](#appendix-a-cache-entry-identifier-formulas).
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#query-branch-keystore-onencrypt
+  //# If the Keystore's GetActiveBranchKey operation succeeds
+  //# the keyring MUST put the returned branch key materials in the cache using the
+  //# formula defined in [Appendix A](#appendix-a-cache-entry-identifier-formulas).
 
-    //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
-    //# If the Keystore's GetBranchKeyVersion operation succeeds
-    //# the keyring MUST put the returned branch key materials in the cache using the
-    //# formula defined in [Appendix A](#appendix-a-cache-entry-identifier-formulas).
-    cmc.putBranchKeyMaterial(cacheEntryId, branchKeyMaterials, cacheLimitTtl)
-  } else {
+  //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#getitem-branch-keystore-ondecrypt
+  //# If the Keystore's GetBranchKeyVersion operation succeeds
+  //# the keyring MUST put the returned branch key materials in the cache using the
+  //# formula defined in [Appendix A](#appendix-a-cache-entry-identifier-formulas).
+  const put = (material: NodeBranchKeyMaterial) =>
+    cmc.putBranchKeyMaterial(cacheEntryId, material, cacheLimitTtl)
+
+  const cacheEntry = lookUp()
+  if (cacheEntry) {
     //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#ondecrypt
     //# If a cache entry is found and the entry's TTL has not expired, the hierarchical keyring MUST use those branch key materials for key unwrapping.
 
     //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
     //# If a cache entry is found and the entry's TTL has not expired, the hierarchical keyring MUST use those branch key materials for key wrapping.
-    branchKeyMaterials = cacheEntry.response
+    return deepCopyBranchKeyMaterial(cacheEntry.response)
   }
-
-  /* Hand back a copy the cache can never touch. The CMC zeroes a material's
-   * buffer in place on eviction; callers read the branch key AFTER this await,
-   * so a concurrent eviction (overwrite, TTL, or tail) could otherwise zero the
-   * buffer mid-derivation. */
-  return deepCopyBranchKeyMaterial(branchKeyMaterials)
-}
-
-function deepCopyBranchKeyMaterial(
-  material: NodeBranchKeyMaterial
-): NodeBranchKeyMaterial {
-  return new NodeBranchKeyMaterial(
-    Buffer.from(material.branchKey()),
-    material.branchKeyIdentifier,
-    material.branchKeyVersion.toString('utf-8'),
-    { ...material.encryptionContext }
-  )
+  return sharedFetch(cmc, cacheEntryId, fetch, put)
 }
 
 //= aws-encryption-sdk-specification/framework/aws-kms/aws-kms-hierarchical-keyring.md#onencrypt
diff --git a/modules/kms-keyring-node/src/shared_branch_key_fetches.ts b/modules/kms-keyring-node/src/shared_branch_key_fetches.ts
new file mode 100644
index 000000000..c01e53c20
--- /dev/null
+++ b/modules/kms-keyring-node/src/shared_branch_key_fetches.ts
@@ -0,0 +1,120 @@
+// Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import {
+  NodeAlgorithmSuite,
+  NodeBranchKeyMaterial,
+} from '@aws-crypto/material-management'
+import { CryptographicMaterialsCache } from '@aws-crypto/cache-material'
+
+/* Concurrent cache misses for the same branch key share one keystore fetch.
+ * If that fetch fails, its waiting callers share one retry,
+ * so a failing keystore gets one more call, not one per waiting caller.
+ */
+
+export const SHARED_FETCHES = {
+  // A fetch older than this may be stuck, so new callers don't join it.
+  graceInterval: 1000, // milliseconds
+}
+
+interface SharedFetch {
+  startedAt: number
+  members: ((outcome: SharedOutcome) => void)[]
+}
+type SharedOutcome = { material: NodeBranchKeyMaterial } | { error: unknown }
+
+/* Fetches in progress, by cache and then by cache entry id.
+ * Keyed by cache, so keyrings that share a cache also share fetches.
+ * Each entry id holds a list, so a stuck fetch that finishes late removes only itself,
+ * not a newer fetch for the same entry.
+ */
+const sharedFetches = new WeakMap<
+  CryptographicMaterialsCache<NodeAlgorithmSuite>,
+  Map<string, SharedFetch[]>
+>()
+
+export function sharedFetchesFor(
+  cmc: CryptographicMaterialsCache<NodeAlgorithmSuite>
+) {
+  return sharedFetches.get(cmc)
+}
+
+function isRecent(fetch: SharedFetch) {
+  return Date.now() < fetch.startedAt + SHARED_FETCHES.graceInterval
+}
+
+/* Fetch the branch key materials, or wait on another caller's fetch of the same entry.
+ * Returns a copy with its own buffer (see deepCopyBranchKeyMaterial).
+ * If the fetch this caller waited on fails, it tries once more, and a second failure throws.
+ * `retry` is false on that second try.
+ */
+export async function sharedFetch(
+  cmc: CryptographicMaterialsCache<NodeAlgorithmSuite>,
+  cacheEntryId: string,
+  fetch: () => Promise<NodeBranchKeyMaterial>,
+  put: (material: NodeBranchKeyMaterial) => void,
+  retry = true
+): Promise<NodeBranchKeyMaterial> {
+  const byId = sharedFetches.get(cmc) || new Map<string, SharedFetch[]>()
+  sharedFetches.set(cmc, byId)
+  // A fetch that hangs never removes itself.
+  // This drops it when the next fetch for this entry starts.
+  const fetches = (byId.get(cacheEntryId) || []).filter(isRecent)
+  const current = fetches[fetches.length - 1]
+
+  if (current) {
+    const outcome = await new Promise<SharedOutcome>((resolve) =>
+      current.members.push(resolve)
+    )
+    if ('material' in outcome) return outcome.material
+    if (!retry) throw outcome.error
+    // The caller that made the failed fetch already removed it,
+    // so the first waiting caller to get here starts a new fetch and the others wait on that one.
+    return sharedFetch(cmc, cacheEntryId, fetch, put, false)
+  }
+
+  const mine: SharedFetch = { startedAt: Date.now(), members: [] }
+  fetches.push(mine)
+  byId.set(cacheEntryId, fetches)
+  const close = () => {
+    const list = byId.get(cacheEntryId)
+    if (!list) return
+    const index = list.indexOf(mine)
+    if (index !== -1) list.splice(index, 1)
+    if (!list.length) byId.delete(cacheEntryId)
+  }
+
+  let failure: SharedOutcome = { error: undefined }
+  try {
+    const material = await fetch()
+    // Copy before put: once cached, an eviction can zero the material's buffer.
+    for (const settle of mine.members) {
+      settle({ material: deepCopyBranchKeyMaterial(material) })
+    }
+    put(material)
+    return deepCopyBranchKeyMaterial(material)
+  } catch (error) {
+    failure = { error }
+    throw error
+  } finally {
+    // If the fetch failed, tell the callers still waiting so they retry once.
+    // After a success they already have the material, so this does nothing.
+    close()
+    for (const settle of mine.members) settle(failure)
+  }
+}
+
+/* The cache zeroes a material's buffer when it evicts it.
+ * Callers derive keys from the branch key after getBranchKeyMaterials returns,
+ * so they need their own buffer, or an eviction can zero the key mid-use (#1691).
+ */
+export function deepCopyBranchKeyMaterial(
+  material: NodeBranchKeyMaterial
+): NodeBranchKeyMaterial {
+  return new NodeBranchKeyMaterial(
+    Buffer.from(material.branchKey()),
+    material.branchKeyIdentifier,
+    material.branchKeyVersion.toString('utf-8'),
+    { ...material.encryptionContext }
+  )
+}
diff --git a/modules/kms-keyring-node/test/kms_hkeyring_node.concurrency.test.ts b/modules/kms-keyring-node/test/kms_hkeyring_node.concurrency.test.ts
index 2369b8dd7..53242191c 100644
--- a/modules/kms-keyring-node/test/kms_hkeyring_node.concurrency.test.ts
+++ b/modules/kms-keyring-node/test/kms_hkeyring_node.concurrency.test.ts
@@ -29,6 +29,15 @@ import {
   KeyStoreInfoOutput,
 } from '@aws-crypto/branch-keystore-node'
 
+import { getBranchKeyMaterials } from '../src/kms_hkeyring_node_helpers'
+import {
+  SHARED_FETCHES,
+  sharedFetchesFor,
+} from '../src/shared_branch_key_fetches'
+import { getLocalCryptographicMaterialsCache } from '@aws-crypto/cache-material'
+import { NodeAlgorithmSuite } from '@aws-crypto/material-management'
+import { v4 } from 'uuid'
+
 const CONCURRENCY = 25
 
 let activeMaterialA: NodeBranchKeyMaterial
@@ -143,3 +152,271 @@ describe('KmsHierarchicalKeyRingNode: concurrent cold-cache operations (#1691)',
     }
   })
 })
+
+const CONCURRENT_OPERATIONS = 3000
+
+// A short grace interval, so the hang tests run in milliseconds.
+const FAST = { graceInterval: 40 }
+const DEFAULT = { ...SHARED_FETCHES }
+
+const wait = async (ms: number) =>
+  new Promise((resolve) => setTimeout(resolve, ms))
+
+// A keystore stub where each request takes 5 ms.
+// `outcomes` decides each call in order: 'ok', 'fail', or 'hang'; later calls use the last one.
+function slowKeyStore(
+  outcomes: ('ok' | 'fail' | 'hang')[] = ['ok']
+): Sinon.SinonStubbedInstance<BranchKeyStoreNode> {
+  const material = new NodeBranchKeyMaterial(
+    Buffer.alloc(32, 1),
+    BRANCH_KEY_ID_A,
+    v4(),
+    {}
+  )
+  const keyStore = stubKeyStore() as any
+  let calls = 0
+  const fetch = async () => {
+    const outcome = outcomes[Math.min(calls++, outcomes.length - 1)]
+    if (outcome === 'hang') return new Promise<never>(() => undefined)
+    await wait(5)
+    if (outcome === 'fail') throw new Error('keystore unavailable')
+    return deepCopyBranchKeyMaterial(material)
+  }
+  keyStore.getActiveBranchKey.callsFake(fetch)
+  keyStore.getBranchKeyVersion.callsFake(fetch)
+  return keyStore
+}
+
+function keyringFor(
+  keyStore: BranchKeyStoreNode,
+  {
+    cache,
+    partitionId,
+  }: {
+    cache?: KmsHierarchicalKeyRingNode['_cmc']
+    partitionId?: string
+  } = {}
+): KmsHierarchicalKeyRingNode {
+  return new KmsHierarchicalKeyRingNode({
+    branchKeyIdSupplier: BRANCH_KEY_ID_SUPPLIER,
+    keyStore,
+    cacheLimitTtl: TTL,
+    cache,
+    partitionId,
+  })
+}
+
+// A minimal keyring for calling getBranchKeyMaterials directly.
+function hKeyringFor(keyStore: any) {
+  return {
+    keyStore,
+    cacheLimitTtl: TTL * 1000,
+    cacheEntryHasExceededLimits: () => false,
+  } as any
+}
+
+function startEncrypts(hkr: KmsHierarchicalKeyRingNode, count: number) {
+  return Array.from({ length: count }, async () =>
+    hkr.onEncrypt(new NodeEncryptionMaterial(TEST_ESDK_ALG_SUITE, EC_A))
+  )
+}
+
+async function encryptConcurrently(
+  hkr: KmsHierarchicalKeyRingNode,
+  count: number
+) {
+  return Promise.all(startEncrypts(hkr, count))
+}
+
+async function encryptSettled(hkr: KmsHierarchicalKeyRingNode, count: number) {
+  return Promise.allSettled(startEncrypts(hkr, count))
+}
+
+describe('KmsHierarchicalKeyRingNode: shared keystore fetches (#1663)', () => {
+  beforeEach(() => Object.assign(SHARED_FETCHES, FAST))
+  afterEach(() => Object.assign(SHARED_FETCHES, DEFAULT))
+
+  it(`${CONCURRENT_OPERATIONS} concurrent onEncrypt misses share one keystore call`, async () => {
+    // On a slow runner, starting 3000 callers can take longer than a short grace interval,
+    // which starts a second fetch.
+    Object.assign(SHARED_FETCHES, { graceInterval: 60 * 1000 })
+    const keyStore = slowKeyStore()
+    await encryptConcurrently(keyringFor(keyStore), CONCURRENT_OPERATIONS)
+    expect(keyStore.getActiveBranchKey.callCount).to.equal(1)
+  })
+
+  it(`${CONCURRENT_OPERATIONS} concurrent onDecrypt misses share one keystore call`, async () => {
+    // On a slow runner, starting 3000 callers can take longer than a short grace interval,
+    // which starts a second fetch.
+    Object.assign(SHARED_FETCHES, { graceInterval: 60 * 1000 })
+    const keyStore = slowKeyStore()
+    const [encrypted] = await encryptConcurrently(keyringFor(keyStore), 1)
+    const expectedPdk = unwrapDataKey(encrypted.getUnencryptedDataKey())
+
+    const hkr = keyringFor(keyStore)
+    const recovered = await Promise.all(
+      Array.from({ length: CONCURRENT_OPERATIONS }, async () => {
+        const material = new NodeDecryptionMaterial(TEST_ESDK_ALG_SUITE, EC_A)
+        await hkr.onDecrypt(material, encrypted.encryptedDataKeys)
+        return unwrapDataKey(material.getUnencryptedDataKey())
+      })
+    )
+
+    expect(keyStore.getBranchKeyVersion.callCount).to.equal(1)
+    for (const pdk of recovered) expect(pdk).to.deep.equal(expectedPdk)
+  })
+
+  it('keyrings that share a cache and partition share keystore calls', async () => {
+    const keyStore = slowKeyStore()
+    const cache = getLocalCryptographicMaterialsCache<NodeAlgorithmSuite>(100)
+    const partitionId = v4()
+    await Promise.all([
+      encryptConcurrently(keyringFor(keyStore, { cache, partitionId }), 10),
+      encryptConcurrently(keyringFor(keyStore, { cache, partitionId }), 10),
+    ])
+    expect(keyStore.getActiveBranchKey.callCount).to.equal(1)
+  })
+
+  it('waiting callers share one retry when the shared fetch fails', async () => {
+    const keyStore = slowKeyStore(['fail', 'ok'])
+    const results = await encryptSettled(keyringFor(keyStore), 10)
+
+    // Only the caller whose fetch failed sees the error.
+    expect(results.filter((r) => r.status === 'rejected')).to.have.lengthOf(1)
+    expect(keyStore.getActiveBranchKey.callCount).to.equal(2)
+  })
+
+  it('waiting callers get the error when the shared retry also fails', async () => {
+    const keyStore = slowKeyStore(['fail'])
+    const results = await encryptSettled(keyringFor(keyStore), 10)
+
+    for (const result of results) expect(result.status).to.equal('rejected')
+    expect(keyStore.getActiveBranchKey.callCount).to.equal(2)
+  })
+
+  it('releases every waiting caller when the keystore response is malformed', async () => {
+    // Copying this malformed response throws before the waiting callers get it.
+    const keyStore = {
+      getBranchKeyVersion: async () => {
+        await wait(5)
+        return {} as NodeBranchKeyMaterial
+      },
+    }
+    const cache = getLocalCryptographicMaterialsCache<NodeAlgorithmSuite>(100)
+
+    const results = await Promise.allSettled(
+      Array.from({ length: 3 }, async () =>
+        getBranchKeyMaterials(
+          hKeyringFor(keyStore),
+          cache,
+          BRANCH_KEY_ID_A,
+          'entry',
+          'version'
+        )
+      )
+    )
+
+    for (const result of results) expect(result.status).to.equal('rejected')
+    expect(sharedFetchesFor(cache)?.size).to.equal(0)
+  })
+
+  it('starts another fetch for new callers once one hangs past graceInterval', async () => {
+    const keyStore = slowKeyStore(['hang', 'ok'])
+    const hkr = keyringFor(keyStore)
+
+    startEncrypts(hkr, 1)
+    await wait(FAST.graceInterval + 5)
+    await encryptConcurrently(hkr, 5)
+
+    expect(keyStore.getActiveBranchKey.callCount).to.equal(2)
+  })
+
+  it('a stale fetch that fails late does not affect callers waiting on a newer one', async () => {
+    const deferreds: {
+      resolve: (m: NodeBranchKeyMaterial) => void
+      reject: (e: Error) => void
+    }[] = []
+    const keyStore = {
+      getBranchKeyVersion: async () =>
+        new Promise<NodeBranchKeyMaterial>((resolve, reject) =>
+          deferreds.push({ resolve, reject })
+        ),
+    }
+    const cache = getLocalCryptographicMaterialsCache<NodeAlgorithmSuite>(100)
+    const get = async () =>
+      getBranchKeyMaterials(
+        hKeyringFor(keyStore),
+        cache,
+        BRANCH_KEY_ID_A,
+        'entry',
+        'version'
+      )
+
+    const stale = get()
+    await wait(FAST.graceInterval + 5)
+    const newer = [get(), get(), get()]
+    await wait(1)
+    expect(deferreds).to.have.lengthOf(2)
+
+    deferreds[0].reject(new Error('stale'))
+    const [staleResult] = await Promise.allSettled([stale])
+    expect((staleResult as PromiseRejectedResult).reason.message).to.equal(
+      'stale'
+    )
+    // A caller arriving now still joins the newer fetch.
+    newer.push(get())
+    deferreds[1].resolve(
+      new NodeBranchKeyMaterial(Buffer.alloc(32, 7), BRANCH_KEY_ID_A, v4(), {})
+    )
+    for (const result of await Promise.all(newer)) {
+      expect(result.branchKey()).to.deep.equal(Buffer.alloc(32, 7))
+    }
+    expect(deferreds).to.have.lengthOf(2)
+  })
+
+  it('forgets a cache entry once its fetches settle', async () => {
+    const cache = getLocalCryptographicMaterialsCache<NodeAlgorithmSuite>(100)
+    await encryptConcurrently(keyringFor(slowKeyStore(), { cache }), 5)
+    await encryptSettled(keyringFor(slowKeyStore(['fail']), { cache }), 5)
+    expect(sharedFetchesFor(cache)?.size).to.equal(0)
+  })
+})
+
+describe('getBranchKeyMaterials: eviction while a request settles', () => {
+  const material = (fill: number) =>
+    new NodeBranchKeyMaterial(Buffer.alloc(32, fill), BRANCH_KEY_ID_A, v4(), {})
+
+  it('returns an intact branch key when another put evicts the entry before callers resume', async () => {
+    const cache = getLocalCryptographicMaterialsCache<NodeAlgorithmSuite>(1)
+    // Another branch key is cached one microtask after this one,
+    // which evicts and zeroes this entry in the one-entry cache.
+    const racingCache = {
+      ...cache,
+      putBranchKeyMaterial(
+        key: string,
+        m: NodeBranchKeyMaterial,
+        ttl?: number
+      ) {
+        cache.putBranchKeyMaterial(key, m, ttl)
+        queueMicrotask(() => cache.putBranchKeyMaterial('other', material(2)))
+      },
+    }
+    const keyStore = { getBranchKeyVersion: async () => material(1) }
+
+    const results = await Promise.all(
+      Array.from({ length: 3 }, async () =>
+        getBranchKeyMaterials(
+          hKeyringFor(keyStore),
+          racingCache,
+          BRANCH_KEY_ID_A,
+          'entry',
+          'version'
+        )
+      )
+    )
+
+    for (const result of results) {
+      expect(result.branchKey()).to.deep.equal(Buffer.alloc(32, 1))
+    }
+  })
+})