provider: deduplicate cids in queue

CHANGELOG.md

@@ -36,6 +36,7 @@ The following emojis are used to highlight certain changes:
 - upgrade to `go-libp2p` [v0.41.1](https://github.com/libp2p/go-libp2p/releases/tag/v0.41.1)
 - `bitswap/network`: Add a new `requests_in_flight` metric gauge that measures how many bitswap streams are being written or read at a given time.
 - improve speed of data onboarding by batching/bufering provider queue writes [#888](https://github.com/ipfs/boxo/pull/888)
+- `provider/queue` deduplicates CIDs [#910](https://github.com/ipfs/boxo/pull/910)
 
 ### Removed
 

provider/internal/queue/queue.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/gammazero/deque"
+	lru "github.com/hashicorp/golang-lru/v2"
 	cid "github.com/ipfs/go-cid"
 	datastore "github.com/ipfs/go-datastore"
 	namespace "github.com/ipfs/go-datastore/namespace"
@@ -127,12 +128,13 @@ func (q *Queue) worker(ctx context.Context) {
 	var (
 		c       cid.Cid
 		counter uint64
-		k       datastore.Key = datastore.Key{}
 		inBuf   deque.Deque[cid.Cid]
 	)
 
 	const baseCap = 1024
 	inBuf.SetBaseCap(baseCap)
+	k := datastore.Key{}
+	dedupCache, _ := lru.New[cid.Cid, struct{}](baseCap)
 
 	defer func() {
 		if c != cid.Undef {
@@ -206,6 +208,9 @@ func (q *Queue) worker(ctx context.Context) {
 			if !ok {
 				return
 			}
+			if found, _ := dedupCache.ContainsOrAdd(toQueue, struct{}{}); found {
+				continue
+			}
 			idle = false
 
 			if c == cid.Undef {
@@ -283,7 +288,7 @@ func (q *Queue) commitInput(ctx context.Context, counter uint64, cids *deque.Deq
 
 	cstr := makeCidString(cids.Front())
 	n := cids.Len()
-	for i := 0; i < n; i++ {
+	for i := range n {
 		c := cids.At(i)
 		key := datastore.NewKey(fmt.Sprintf("%020d/%s", counter, cstr))
 		if err = b.Put(ctx, key, c.Bytes()); err != nil {

provider/internal/queue/queue_test.go

@@ -118,3 +118,21 @@ func TestInitializationWithManyCids(t *testing.T) {
 
 	assertOrdered(cids, queue, t)
 }
+
+func TestDeduplicateCids(t *testing.T) {
+	ds := sync.MutexWrap(datastore.NewMapDatastore())
+	queue := New(ds)
+	defer queue.Close()
+
+	cids := random.Cids(5)
+	queue.Enqueue(cids[0])
+	queue.Enqueue(cids[0])
+	queue.Enqueue(cids[1])
+	queue.Enqueue(cids[2])
+	queue.Enqueue(cids[1])
+	queue.Enqueue(cids[3])
+	queue.Enqueue(cids[0])
+	queue.Enqueue(cids[4])
+
+	assertOrdered(cids, queue, t)
+}