fix: deduplicate move operations

lib/content/write.js

@@ -15,6 +15,9 @@ const fsm = require('fs-minipass')
 
 module.exports = write
 
+// Cache of move operations in process so we don't duplicate
+const moveOperations = new Map()
+
 async function write (cache, data, opts = {}) {
   const { algorithms, size, integrity } = opts
 
@@ -159,15 +162,23 @@ async function makeTmp (cache, opts) {
 async function moveToDestination (tmp, cache, sri, opts) {
   const destination = contentPath(cache, sri)
   const destDir = path.dirname(destination)
-
-  await fs.mkdir(destDir, { recursive: true })
+  if (moveOperations.has(destination)) {
+    return moveOperations.get(destination)
+  }
+  moveOperations.set(
+    destination,
+    fs.mkdir(destDir, { recursive: true })
+      .then(() => moveFile(tmp.target, destination, { overwrite: false }))
+  )
   try {
-    await moveFile(tmp.target, destination, { overwrite: false })
+    await moveOperations.get(destination)
     tmp.moved = true
   } catch (err) {
     if (!err.message.startsWith('The destination file exists')) {
       throw Object.assign(err, { code: 'EEXIST' })
     }
+  } finally {
+    moveOperations.delete(destination)
   }
 }
 

test/put.js

@@ -1,6 +1,7 @@
 'use strict'
 
 const fs = require('fs/promises')
+const path = require('path')
 const index = require('../lib/entry-index')
 const memo = require('../lib/memoization')
 const t = require('tap')
@@ -128,3 +129,18 @@ t.test('signals error if error writing to cache', async t => {
   t.equal(bulkErr.code, 'EBADSIZE', 'got error from bulk write')
   t.equal(streamErr.code, 'EBADSIZE', 'got error from stream write')
 })
+
+t.test('concurrent puts', async t => {
+  const CACHE = t.testdir()
+  await Promise.all([
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+    put(CACHE, KEY, CONTENT),
+  ])
+  const tmpFiles = await fs.readdir(path.join(CACHE, 'tmp'))
+  t.strictSame(tmpFiles, [], 'Nothing left in tmp')
+})