Replace dsqueue

go.mod

@@ -3,6 +3,7 @@ module github.com/libp2p/go-libp2p-kad-dht
 go 1.25.7
 
 require (
+	github.com/gammazero/cascadeq v0.2.0
 	github.com/gammazero/deque v1.2.1
 	github.com/google/gopacket v1.1.19
 	github.com/google/uuid v1.6.0
@@ -13,7 +14,6 @@ require (
 	github.com/ipfs/go-datastore v0.9.2
 	github.com/ipfs/go-detect-race v0.0.1
 	github.com/ipfs/go-ds-pebble v0.5.12
-	github.com/ipfs/go-dsqueue v0.2.0
 	github.com/ipfs/go-libdht v0.5.0
 	github.com/ipfs/go-log/v2 v2.9.2
 	github.com/ipfs/go-test v0.4.1
@@ -63,13 +63,13 @@ require (
 	github.com/dunglas/httpsfv v1.1.0 // indirect
 	github.com/filecoin-project/go-clock v0.1.0 // indirect
 	github.com/flynn/noise v1.1.0 // indirect
+	github.com/gammazero/fsutil v0.2.0 // indirect
 	github.com/getsentry/sentry-go v0.27.0 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/snappy v0.0.5-0.20231225225746-43d5d4cd4e0e // indirect
 	github.com/gorilla/websocket v1.5.3 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/ipfs/go-block-format v0.2.4 // indirect
 	github.com/ipld/go-ipld-prime v0.24.0 // indirect

go.sum

@@ -81,8 +81,12 @@ github.com/flynn/noise v1.1.0/go.mod h1:xbMo+0i6+IGbYdJhF31t2eR1BIU0CYc12+BNAKwU
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/gammazero/cascadeq v0.2.0 h1:LHq6hWLQvQCHVtCl5hDWYSavqXRFCKZ8upCdRtsPpfo=
+github.com/gammazero/cascadeq v0.2.0/go.mod h1:aEkjsO3wVhT09lheCjiKbEmT7GtGa03WXGmclOZp5wY=
 github.com/gammazero/deque v1.2.1 h1:9fnQVFCCZ9/NOc7ccTNqzoKd1tCWOqeI05/lPqFPMGQ=
 github.com/gammazero/deque v1.2.1/go.mod h1:5nSFkzVm+afG9+gy0VIowlqVAW4N8zNcMne+CMQVD2g=
+github.com/gammazero/fsutil v0.2.0 h1:/MqQHCBoT07KGY75avKqG/SI0zS693zr1Ljx8cwKWhs=
+github.com/gammazero/fsutil v0.2.0/go.mod h1:UhNgS1Hr75DBX6zqBEOB4AAZQiNytnr3Mc0ZpiLKPz4=
 github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps=
 github.com/getsentry/sentry-go v0.27.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w1gwTf25trprY=
 github.com/ghemawat/stream v0.0.0-20171120220530-696b145b53b9 h1:r5GgOLGbza2wVHRzK7aAj6lWZjfbAwiu/RDCVOKjRyM=
@@ -127,8 +131,6 @@ github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmv
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c=
 github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
-github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
-github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc=
@@ -149,12 +151,8 @@ github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ
 github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
 github.com/ipfs/go-ds-badger v0.0.7/go.mod h1:qt0/fWzZDoPW6jpQeqUjR5kBfhDNB65jd9YlmAvpQBk=
 github.com/ipfs/go-ds-leveldb v0.1.0/go.mod h1:hqAW8y4bwX5LWcCtku2rFNX3vjDZCy5LZCg+cSZvYb8=
-github.com/ipfs/go-ds-leveldb v0.5.2 h1:6nmxlQ2zbp4LCNdJVsmHfs9GP0eylfBNxpmY1csp0x0=
-github.com/ipfs/go-ds-leveldb v0.5.2/go.mod h1:2fAwmcvD3WoRT72PzEekHBkQmBDhc39DJGoREiuGmYo=
 github.com/ipfs/go-ds-pebble v0.5.12 h1:idO/w4i3IBA6vZtVWsyG5IlPIgwd62iUaQZBl/Kv+yI=
 github.com/ipfs/go-ds-pebble v0.5.12/go.mod h1:H2zy28KMQSiAflUxpKzKHqbpSHRWPZS5/bi4ymAJOjY=
-github.com/ipfs/go-dsqueue v0.2.0 h1:MBi9w3oSiX98Xc+Y7NuJ9G8MI6mAT4IGdO9dHEMCZzU=
-github.com/ipfs/go-dsqueue v0.2.0/go.mod h1:8FfNQC4DMF/KkzBXRNB9Rb3MKDW0Sh98HMtXYl1mLQE=
 github.com/ipfs/go-ipfs-delay v0.0.0-20181109222059-70721b86a9a8/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw=
 github.com/ipfs/go-ipfs-util v0.0.1/go.mod h1:spsl5z8KUnrve+73pOhSVZND1SIxPW5RyBCNzQxlJBc=
 github.com/ipfs/go-libdht v0.5.0 h1:ZN+eCqwahZvUeT0e4DsIxRtm78Mc9UR5tmZUiMsrGjQ=
@@ -398,7 +396,6 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE=
 github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/wangjia184/sortedset v0.0.0-20160527075905-f5d03557ba30/go.mod h1:YkocrP2K2tcw938x9gCOmT5G5eCD6jsTz0SZuyAqwIE=

provider/buffered/provider.go

@@ -4,8 +4,8 @@ import (
 	"errors"
 	"sync"
 
+	"github.com/gammazero/cascadeq"
 	"github.com/ipfs/go-datastore"
-	"github.com/ipfs/go-dsqueue"
 	"github.com/ipfs/go-log/v2"
 	"github.com/libp2p/go-libp2p-kad-dht/provider/internal"
 	mh "github.com/multiformats/go-multihash"
@@ -36,31 +36,34 @@ type SweepingProvider struct {
 
 	newItems  chan struct{}
 	Provider  internal.Provider
-	queue     *dsqueue.DSQueue
+	queue     *cascadeq.Queue
 	batchSize int
 	logger    *log.ZapEventLogger
 }
 
 // New creates a new SweepingProvider that wraps the given provider with
 // buffering capabilities. Operations are queued and processed asynchronously
 // in batches for improved performance.
-func New(prov internal.Provider, ds datastore.Batching, opts ...Option) *SweepingProvider {
+func New(prov internal.Provider, ds datastore.Batching, queuePath string, opts ...Option) (*SweepingProvider, error) {
 	cfg := getOpts(opts)
+
+	queue, err := cascadeq.New(queuePath,
+		cascadeq.WithSnapshotInterval(cfg.idleWriteTime))
+	if err != nil {
+		return nil, err
+	}
 	s := &SweepingProvider{
 		done:   make(chan struct{}),
 		closed: make(chan struct{}),
 
-		newItems: make(chan struct{}, 1),
-		Provider: prov,
-		queue: dsqueue.New(ds, cfg.dsName,
-			dsqueue.WithDedupCacheSize(0), // disable deduplication
-			dsqueue.WithIdleWriteTime(cfg.idleWriteTime),
-		),
+		newItems:  make(chan struct{}, 1),
+		Provider:  prov,
+		queue:     queue,
 		batchSize: cfg.batchSize,
 		logger:    log.Logger(cfg.loggerName),
 	}
 	go s.worker()
-	return s
+	return s, nil
 }
 
 // Close stops the provider and releases all resources.
@@ -136,6 +139,8 @@ func (s *SweepingProvider) executeOperation(f func(...mh.Multihash) error, keys
 // It runs in a separate goroutine and continues until the provider is closed.
 func (s *SweepingProvider) worker() {
 	defer close(s.done)
+	res := make([][]byte, s.batchSize)
+
 	var emptyQueue bool
 	for {
 		if emptyQueue {
@@ -154,16 +159,15 @@ func (s *SweepingProvider) worker() {
 			}
 		}
 
-		res, err := s.queue.GetN(s.batchSize)
-		if err != nil {
-			s.logger.Warnf("BufferedSweepingProvider unable to dequeue: %v", err)
-			continue
-		}
-		if len(res) < s.batchSize {
+		n := s.queue.Drain(res)
+		if n < s.batchSize {
 			// Queue was fully drained.
 			emptyQueue = true
+			if n == 0 {
+				continue
+			}
 		}
-		ops, err := getOperations(res)
+		ops, err := getOperations(res[:n])
 		if err != nil {
 			s.logger.Warnf("BufferedSweepingProvider unable to parse dequeued item: %v", err)
 			continue
@@ -187,11 +191,16 @@ func (s *SweepingProvider) worker() {
 
 // enqueue adds operations to the queue for asynchronous processing.
 func (s *SweepingProvider) enqueue(op byte, keys ...mh.Multihash) error {
+	data := make([][]byte, 0, len(keys))
 	for _, h := range keys {
-		if err := s.queue.Put(toBytes(op, h)); err != nil {
-			return err
-		}
+		data = append(data, toBytes(op, h))
 	}
+
+	err := s.queue.PutBatch(data)
+	if err != nil {
+		return err
+	}
+
 	select {
 	case s.newItems <- struct{}{}:
 	default:
@@ -242,12 +251,11 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error {
 	return s.enqueue(stopProvidingOp, keys...)
 }
 
-// Clear clears the all the keys from the provide queue and returns the number
-// of keys that were cleared.
+// Clear clears the all the keys from the provide queue.
 //
 // The keys are not deleted from the keystore, so they will continue to be
 // reprovided as scheduled.
-func (s *SweepingProvider) Clear() int {
+func (s *SweepingProvider) Clear() error {
 	return s.Provider.Clear()
 }
 

provider/buffered/provider_test.go

@@ -78,9 +78,9 @@ func (f *fakeProvider) StopProviding(keys ...mh.Multihash) error {
 	return nil
 }
 
-func (f *fakeProvider) Clear() int {
+func (f *fakeProvider) Clear() error {
 	// Unused
-	return 0
+	return nil
 }
 
 func (f *fakeProvider) RefreshSchedule() error {
@@ -103,10 +103,13 @@ func TestQueueingMechanism(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		fake := newFakeProvider()
 		ds := datastore.NewMapDatastore()
-		provider := New(fake, ds,
+		provider, err := New(fake, ds, t.TempDir(),
 			WithDsName("test1"),
 			WithIdleWriteTime(time.Millisecond),
 			WithBatchSize(10))
+		if err != nil {
+			t.Fatal(err)
+		}
 		defer provider.Close()
 
 		keys := random.Multihashes(3)
@@ -240,19 +243,23 @@ func TestBatchProcessing(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		fake := newFakeProvider()
 		ds := datastore.NewMapDatastore()
-		provider := New(fake, ds,
+		provider, err := New(fake, ds, t.TempDir(),
 			WithDsName("test4"),
-			WithBatchSize(3), // Process 3 operations at once
+			WithBatchSize(1000), // Process 1000 operations at once
 			WithIdleWriteTime(time.Second))
+		if err != nil {
+			t.Fatal(err)
+		}
 		defer provider.Close()
 
-		// Queue multiple keys - total of 3 operations (2 from ProvideOnce + 1 from StartProviding)
-		keys := random.Multihashes(3)
+		// Queue multiple keys - total of 30000 operations (2 from ProvideOnce + 29998
+		// from StartProviding). Should be enough to overflow to disk.
+		keys := random.Multihashes(30000)
 
 		if err := provider.ProvideOnce(keys[0], keys[1]); err != nil {
 			t.Fatalf("ProvideOnce failed: %v", err)
 		}
-		if err := provider.StartProviding(false, keys[2]); err != nil {
+		if err := provider.StartProviding(false, keys[2:]...); err != nil {
 			t.Fatalf("StartProviding failed: %v", err)
 		}
 		synctest.Wait()
@@ -273,8 +280,8 @@ func TestBatchProcessing(t *testing.T) {
 		for _, call := range fake.startProvidingCalls {
 			totalStartProvidingCalls += len(call.keys)
 		}
-		if totalStartProvidingCalls != 1 {
-			t.Errorf("Expected 1 total key in StartProviding calls, got %d", totalStartProvidingCalls)
+		if totalStartProvidingCalls != len(keys)-2 {
+			t.Errorf("Expected %d total key in StartProviding calls, got %d", len(keys)-2, totalStartProvidingCalls)
 		}
 	})
 }

provider/dual/provider.go

@@ -208,13 +208,12 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error {
 	return nil
 }
 
-// Clear clears the all the keys from the provide queues of both DHTs and
-// returns the number of keys that were cleared (sum of both queues).
+// Clear clears the all the keys from the provide queues of both DHTs.
 //
 // The keys are not deleted from the keystore, so they will continue to be
 // reprovided as scheduled.
-func (s *SweepingProvider) Clear() int {
-	return s.LAN.Clear() + s.WAN.Clear()
+func (s *SweepingProvider) Clear() error {
+	return errors.Join(s.LAN.Clear(), s.WAN.Clear())
 }
 
 // RefreshSchedule scans the Keystore for any keys that are not currently

provider/internal/interface.go

@@ -12,7 +12,7 @@ type Provider interface {
 	StartProviding(force bool, keys ...mh.Multihash) error
 	StopProviding(keys ...mh.Multihash) error
 	ProvideOnce(keys ...mh.Multihash) error
-	Clear() int
+	Clear() error
 	RefreshSchedule() error
 	Close() error
 }

provider/internal/queue/prefix.go

@@ -68,15 +68,12 @@ func (q *prefixQueue) Size() int {
 	return q.queue.Len()
 }
 
-// Clear removes all keys from the queue and returns the number of keys that
-// were removed.
-func (q *prefixQueue) Clear() int {
-	size := q.Size()
-
+// Clear removes all keys from the queue.
+func (q *prefixQueue) Clear() error {
 	q.queue.Clear()
 	*q.prefixes = trie.Trie[bitstr.Key, struct{}]{}
 
-	return size
+	return nil
 }
 
 // removeSuperstrings finds all superstrings of `prefix` in the trie, removes

provider/internal/queue/provide.go

@@ -191,17 +191,15 @@ func (q *ProvideQueue) NumRegions() int {
 	return q.queue.Size()
 }
 
-// Clear removes all keys from the queue and returns the number of keys that
-// were removed.
-func (q *ProvideQueue) Clear() int {
+// Clear removes all keys from the queue.
+func (q *ProvideQueue) Clear() error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
-	size := q.keys.Size()
 
 	q.queue.Clear()
 	*q.keys = trie.Trie[bit256.Key, mh.Multihash]{}
 
-	return size
+	return nil
 }
 
 // Persist saves the current state of the queue to the provided datastore.

provider/internal/queue/provide_test.go

@@ -282,8 +282,7 @@ func TestProvideClearQueue(t *testing.T) {
 	require.Equal(t, q.queue.queue.Len(), len(prefixes))
 	require.Equal(t, q.Size(), len(prefixes)*nMultihashesPerPrefix)
 
-	cleared := q.Clear()
-	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, cleared)
+	require.NoError(t, q.Clear())
 	require.True(t, q.IsEmpty())
 
 	require.True(t, q.keys.IsEmptyLeaf())

provider/internal/queue/reprovide.go

@@ -61,9 +61,8 @@ func (q *ReprovideQueue) Size() int {
 	return q.queue.Size()
 }
 
-// Clear removes all prefixes from the queue and returns the number of removed
-// prefixes.
-func (q *ReprovideQueue) Clear() int {
+// Clear removes all prefixes from the queue.
+func (q *ReprovideQueue) Clear() error {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 	return q.queue.Clear()

provider/internal/queue/reprovide_test.go

@@ -104,8 +104,7 @@ func TestReprovideClearQueue(t *testing.T) {
 		q.Enqueue(k)
 	}
 
-	cleared := q.Clear()
-	require.Equal(t, len(keys), cleared)
+	require.NoError(t, q.Clear())
 	require.True(t, q.IsEmpty())
 	require.Equal(t, 0, q.queue.prefixes.Size())
 	require.Equal(t, 0, q.queue.queue.Len())

provider/provider.go

@@ -2163,14 +2163,13 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error {
 	return err
 }
 
-// Clear clears the all the keys from the provide queue and returns the number
-// of keys that were cleared.
+// Clear clears all the keys from the provide queue.
 //
 // The keys are not deleted from the keystore, so they will continue to be
 // reprovided as scheduled.
-func (s *SweepingProvider) Clear() int {
+func (s *SweepingProvider) Clear() error {
 	if s.closed() {
-		return 0
+		return nil
 	}
 	return s.provideQueue.Clear()
 }

provider/provider_test.go

@@ -1426,7 +1426,7 @@ func TestClose(t *testing.T) {
 		require.ErrorIs(t, err, ErrClosed)
 		err = prov.ProvideOnce(newMh)
 		require.ErrorIs(t, err, ErrClosed)
-		require.Equal(t, 0, prov.Clear())
+		require.NoError(t, prov.Clear())
 
 		err = prov.workerPool.Acquire(burstWorker)
 		require.ErrorIs(t, err, reservedpool.ErrClosed)