feat(transcode): lock the destination transcode cache path

transcode/transcoder_caching.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sync"
 )
 
 const perm = 0o644
@@ -14,6 +15,7 @@ const perm = 0o644
 type CachingTranscoder struct {
 	cachePath  string
 	transcoder Transcoder
+	locks      keyedMutex
 }
 
 var _ Transcoder = (*CachingTranscoder)(nil)
@@ -38,8 +40,10 @@ func (t *CachingTranscoder) Transcode(ctx context.Context, profile Profile, in s
 	}
 
 	key := cacheKey(name, args)
-	path := filepath.Join(t.cachePath, key)
+	unlock := t.locks.Lock(key)
+	defer unlock()
 
+	path := filepath.Join(t.cachePath, key)
 	cf, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o644)
 	if err != nil {
 		return fmt.Errorf("open cache file: %w", err)
@@ -51,7 +55,8 @@ func (t *CachingTranscoder) Transcode(ctx context.Context, profile Profile, in s
 		return nil
 	}
 
-	if err := t.transcoder.Transcode(ctx, profile, in, io.MultiWriter(out, cf)); err != nil {
+	dest := io.MultiWriter(out, cf)
+	if err := t.transcoder.Transcode(ctx, profile, in, dest); err != nil {
 		os.Remove(path)
 		return fmt.Errorf("internal transcode: %w", err)
 	}
@@ -69,3 +74,15 @@ func cacheKey(cmd string, args []string) string {
 	}
 	return fmt.Sprintf("%x", sum.Sum(nil))
 }
+
+type keyedMutex struct {
+	sync.Map
+}
+
+func (km *keyedMutex) Lock(key string) func() {
+	value, _ := km.LoadOrStore(key, &sync.Mutex{})
+	mu := value.(*sync.Mutex)
+	mu.Lock()
+	// TODO: remove key entry from map to save some space?
+	return mu.Unlock
+}