fix(share/eds): skip concurrent writes on edsstore Put

share/eds/store.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -64,6 +65,9 @@ type Store struct {
 	// lastGCResult is only stored on the store for testing purposes.
 	lastGCResult atomic.Pointer[dagstore.GCResult]
 
+	// stripedLocks is used to synchronize parallel operations
+	stripedLocks [256]sync.Mutex
+
 	metrics *metrics
 }
 
@@ -197,6 +201,10 @@ func (s *Store) Put(ctx context.Context, root share.DataHash, square *rsmt2d.Ext
 }
 
 func (s *Store) put(ctx context.Context, root share.DataHash, square *rsmt2d.ExtendedDataSquare) (err error) {
+	lk := &s.stripedLocks[root[len(root)-1]]
+	lk.Lock()
+	defer lk.Unlock()
+
 	// if root already exists, short-circuit
 	if has, _ := s.Has(ctx, root); has {
 		return dagstore.ErrShardExists

share/eds/store_test.go

@@ -3,9 +3,11 @@ package eds
 import (
 	"context"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/filecoin-project/dagstore"
 	"github.com/filecoin-project/dagstore/shard"
 	"github.com/ipfs/go-datastore"
 	ds_sync "github.com/ipfs/go-datastore/sync"
@@ -171,6 +173,30 @@ func TestEDSStore(t *testing.T) {
 			assert.Contains(t, hashesOut, hash)
 		}
 	})
+
+	t.Run("Parallel put", func(t *testing.T) {
+		const amount = 20
+		eds, dah := randomEDS(t)
+
+		wg := sync.WaitGroup{}
+		for i := 1; i < amount; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				err := edsStore.Put(ctx, dah.Hash(), eds)
+				if err != nil {
+					require.ErrorIs(t, err, dagstore.ErrShardExists)
+				}
+			}()
+		}
+		wg.Wait()
+
+		eds, err := edsStore.Get(ctx, dah.Hash())
+		require.NoError(t, err)
+		newDah, err := da.NewDataAvailabilityHeader(eds)
+		require.NoError(t, err)
+		require.Equal(t, dah.Hash(), newDah.Hash())
+	})
 }
 
 // TestEDSStore_GC verifies that unused transient shards are collected by the GC periodically.