Repository: libp2p/go-libp2p-kad-dht

Issue:
provider: resume reprovide operations
## Current situtation

A fresh reprovide cycle starts as soon as the `SweepingProvider` is created.

This means that if the node has been providing (and reproviding) content, then is turned off (not just offline) for a few hours/days, and then is restarted, it will take up to `reprovideInterval` for all keys to be reprovided.

## Ideal situation

We want a similar behaviour to when the `provide` goes `offline`, which is to keep track of regions that should have been reprovided but weren't because the node was offline, and reprovide them as fast as possible when the `provider` comes back online.

We need to write reprovide timestamps to a datastore, so that when the `SweepingProvider` is started it can look in the datastore for regions that haven't been provided since at least `reprovideInterval` and schedule them to be reprovided ASAP.

Discussion:

Solution patch:
diff --git a/provider/internal/queue/prefix.go b/provider/internal/queue/prefix.go
index 7cc52e2..d561a0e 100644
--- a/provider/internal/queue/prefix.go
+++ b/provider/internal/queue/prefix.go
@@ -1,3 +1,5 @@
+// Package queue provides queues for batching DHT provide operations by
+// Kademlia prefix, with optional datastore persistence.
 package queue
 
 import (
diff --git a/provider/internal/queue/provide.go b/provider/internal/queue/provide.go
index c6fea1c..4b1f8fa 100644
--- a/provider/internal/queue/provide.go
+++ b/provider/internal/queue/provide.go
@@ -1,8 +1,13 @@
 package queue
 
 import (
+	"context"
+	"fmt"
+	"strings"
 	"sync"
 
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
 	mh "github.com/multiformats/go-multihash"
 
 	"github.com/probe-lab/go-libdht/kad/key/bit256"
@@ -57,7 +62,12 @@ func (q *ProvideQueue) Enqueue(prefix bitstr.Key, keys ...mh.Multihash) {
 	}
 	q.mu.Lock()
 	defer q.mu.Unlock()
+	q.enqueueNoLock(prefix, keys)
+}
 
+// enqueueNoLock adds the supplied keys to the queue under the given prefix
+// without acquiring the mutex. The caller must hold q.mu.
+func (q *ProvideQueue) enqueueNoLock(prefix bitstr.Key, keys []mh.Multihash) {
 	// Enqueue the prefix in the queue if required.
 	q.queue.Push(prefix)
 
@@ -193,3 +203,144 @@ func (q *ProvideQueue) Clear() int {
 
 	return size
 }
+
+// Persist saves the current state of the queue to the provided datastore.
+//
+// The queue state includes the ordered list of prefixes and all associated
+// multihashes. The data is stored under keys formatted as:
+// * {queue-position:012x}/{prefix-bitstring}
+//
+// This operation does not modify the queue's in-memory state.
+func (q *ProvideQueue) Persist(ctx context.Context, d ds.Batching, batchSize int) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	// Create a batch for efficient writes
+	batch, err := d.Batch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create batch: %w", err)
+	}
+
+	i := 0
+	for prefix := range q.queue.queue.Iter() {
+		// Find all keys matching this prefix
+		if subtrie, ok := keyspace.FindSubtrie(q.keys, prefix); ok {
+			keys := keyspace.AllValues(subtrie, bit256.ZeroKey())
+
+			// Concatenate all multihash bytes
+			var buf []byte
+			for _, h := range keys {
+				buf = append(buf, []byte(h)...)
+			}
+
+			// Store with queue position and prefix as key
+			key := ds.NewKey(fmt.Sprintf("%012x/%s", i, string(prefix)))
+			if err := batch.Put(ctx, key, buf); err != nil {
+				return fmt.Errorf("failed to store prefix data: %w", err)
+			}
+		}
+
+		i++
+
+		if i%batchSize == 0 {
+			// Max batch size reached, commit and start a new batch.
+			if err := batch.Commit(ctx); err != nil {
+				return fmt.Errorf("failed to commit batch: %w", err)
+			}
+			batch, err = d.Batch(ctx)
+			if err != nil {
+				return fmt.Errorf("failed to create batch: %w", err)
+			}
+		}
+	}
+
+	if err := batch.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit batch: %w", err)
+	}
+
+	return nil
+}
+
+// decodeMultihashes parses concatenated multihash bytes and returns individual multihashes.
+func decodeMultihashes(data []byte) ([]mh.Multihash, error) {
+	var multihashes []mh.Multihash
+	offset := 0
+
+	for offset < len(data) {
+		// Parse the multihash at current offset
+		// MHFromBytes returns (consumed, multihash, error)
+		consumed, mhash, err := mh.MHFromBytes(data[offset:])
+		if err != nil {
+			return nil, fmt.Errorf("failed to decode multihash at offset %d: %w", offset, err)
+		}
+
+		multihashes = append(multihashes, mhash)
+		offset += consumed
+	}
+
+	return multihashes, nil
+}
+
+// DrainDatastore loads persisted queue entries from the datastore, adds them to
+// the current queue, and removes them from the datastore.
+//
+// This operation does NOT clear the existing queue state - it is additive.
+func (q *ProvideQueue) DrainDatastore(ctx context.Context, d ds.Batching) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	// Query all persisted entries, ordered by key (which preserves queue order)
+	dsQuery := query.Query{
+		Orders: []query.Order{query.OrderByKey{}},
+	}
+
+	results, err := d.Query(ctx, dsQuery)
+	if err != nil {
+		return fmt.Errorf("failed to query datastore: %w", err)
+	}
+
+	// Create a batch for deletes
+	batch, err := d.Batch(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to create batch: %w", err)
+	}
+
+	for result := range results.Next() {
+		if result.Error != nil {
+			return fmt.Errorf("error reading query result: %w", result.Error)
+		}
+
+		// Key format: "/position/prefix"
+		parts := strings.Split(strings.TrimPrefix(result.Key, "/"), "/")
+		if len(parts) != 2 {
+			continue // Skip invalid keys
+		}
+		prefix := bitstr.Key(parts[1])
+
+		// Decode concatenated multihashes
+		keys, err := decodeMultihashes(result.Value)
+		if err != nil {
+			return fmt.Errorf("failed to decode multihashes for prefix %s: %w", prefix, err)
+		}
+		if len(keys) == 0 {
+			continue
+		}
+
+		// Add keys to provide queue
+		q.enqueueNoLock(prefix, keys)
+
+		// Delete key from datastore
+		batch.Delete(ctx, ds.NewKey(result.Key))
+	}
+
+	if err := results.Close(); err != nil {
+		return fmt.Errorf("failed to close query results: %w", err)
+	}
+
+	// Commit deletions
+	if err := batch.Commit(ctx); err != nil {
+		return fmt.Errorf("failed to commit batch: %w", err)
+	}
+
+	return nil
+}
diff --git a/provider/internal/queue/reprovide.go b/provider/internal/queue/reprovide.go
index 4c868f0..666b03a 100644
--- a/provider/internal/queue/reprovide.go
+++ b/provider/internal/queue/reprovide.go
@@ -14,7 +14,7 @@ type ReprovideQueue struct {
 	queue prefixQueue
 }
 
-// New creates a new ReprovideQueue instance.
+// NewReprovideQueue creates a new ReprovideQueue instance.
 func NewReprovideQueue() *ReprovideQueue {
 	return &ReprovideQueue{queue: prefixQueue{prefixes: trie.New[bitstr.Key, struct{}]()}}
 }
diff --git a/provider/keystore/keystore.go b/provider/keystore/keystore.go
index 962f37d..d72ef7f 100644
--- a/provider/keystore/keystore.go
+++ b/provider/keystore/keystore.go
@@ -29,6 +29,7 @@ type Keystore interface {
 	Delete(context.Context, ...mh.Multihash) error
 	Empty(context.Context) error
 	Size(context.Context) (int, error)
+	BatchSize() int
 	Close() error
 }
 
@@ -383,6 +384,11 @@ func (s *keystore) executeOperation(op opType, ctx context.Context, keys []mh.Mu
 	}
 }
 
+// BatchSize returns the configured batch size for datastore operations.
+func (s *keystore) BatchSize() int {
+	return s.batchSize
+}
+
 // Put stores the provided keys in the underlying datastore, grouping them by
 // the first prefixLen bits. It returns only the keys that were not previously
 // persisted in the datastore (i.e., newly added keys).
diff --git a/provider/options.go b/provider/options.go
index 8e109cf..09749cc 100644
--- a/provider/options.go
+++ b/provider/options.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/ipfs/go-datastore"
 	"github.com/libp2p/go-libp2p-kad-dht/amino"
 	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
 	"github.com/libp2p/go-libp2p-kad-dht/provider/keystore"
@@ -41,7 +42,8 @@ type config struct {
 	peerid peer.ID
 	router KadClosestPeersRouter
 
-	keystore keystore.Keystore
+	keystore  keystore.Keystore
+	datastore datastore.Batching
 
 	msgSender      pb.MessageSender
 	selfAddrs      func() []ma.Multiaddr
@@ -279,3 +281,15 @@ func WithKeystore(ks keystore.Keystore) Option {
 		return nil
 	}
 }
+
+// WithDatastore defines the datastore used to keep track of the keyspace
+// region reprovides and persist the provide queue on close.
+func WithDatastore(ds datastore.Batching) Option {
+	return func(cfg *config) error {
+		if ds == nil {
+			return errors.New("reprovider config: datastore cannot be nil")
+		}
+		cfg.datastore = ds
+		return nil
+	}
+}
diff --git a/provider/provider.go b/provider/provider.go
index 0f2bfc3..2514ddf 100644
--- a/provider/provider.go
+++ b/provider/provider.go
@@ -31,6 +31,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/go-test/random"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -187,9 +188,10 @@ func New(opts ...Option) (*SweepingProvider, error) {
 	}
 	cleanupFuncs := []func() error{}
 
+	var mapDs *ds.MapDatastore
 	if cfg.keystore == nil {
 		// Setup KeyStore if missing
-		mapDs := ds.NewMapDatastore()
+		mapDs = ds.NewMapDatastore()
 		cleanupFuncs = append(cleanupFuncs, mapDs.Close)
 		cfg.keystore, err = keystore.NewKeystore(mapDs)
 		if err != nil {
@@ -198,6 +200,15 @@ func New(opts ...Option) (*SweepingProvider, error) {
 		}
 		cleanupFuncs = append(cleanupFuncs, cfg.keystore.Close)
 	}
+	if cfg.datastore == nil {
+		// Setup datastore if missing
+		if mapDs == nil {
+			mapDs = ds.NewMapDatastore()
+			cleanupFuncs = append(cleanupFuncs, mapDs.Close)
+		}
+		cfg.datastore = mapDs
+	}
+
 	meter := otel.Meter("github.com/libp2p/go-libp2p-kad-dht/provider")
 	providerCounter, err := meter.Int64Counter(
 		"total_provide_count",
@@ -272,11 +283,20 @@ func New(opts ...Option) (*SweepingProvider, error) {
 	prov.connectivity.SetCallbacks(prov.onOnline, prov.onOffline)
 	prov.connectivity.Start()
 
+	pqueueDs := namespace.Wrap(cfg.datastore, ds.NewKey("pqueue"))
+	// Load keys that were logged to the datastore back to the provide queue.
+	if err := prov.provideQueue.DrainDatastore(ctx, pqueueDs); err != nil {
+		logger.Errorw("failed to drain provide queue from datastore", "error", err)
+	}
+
+	// Provide queue is persisted on close, reuse keystore batch size.
+	batchSize := prov.keystore.BatchSize()
+	persistProvideQueue := func() error { return prov.provideQueue.Persist(ctx, pqueueDs, batchSize) }
+	prov.cleanupFuncs = append(prov.cleanupFuncs, prov.workerPool.Close, persistProvideQueue)
+
 	// Don't need to start schedule timer yet
 	prov.scheduleTimer.Stop()
 
-	prov.cleanupFuncs = append(prov.cleanupFuncs, prov.workerPool.Close)
-
 	prov.wg.Add(1)
 	go prov.run()
 



Test patch:
diff --git a/provider/internal/queue/provide_test.go b/provider/internal/queue/provide_test.go
index 908648b..5599957 100644
--- a/provider/internal/queue/provide_test.go
+++ b/provider/internal/queue/provide_test.go
@@ -1,8 +1,12 @@
 package queue
 
 import (
+	"context"
 	"testing"
 
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/query"
+	dssync "github.com/ipfs/go-datastore/sync"
 	"github.com/ipfs/go-test/random"
 	mh "github.com/multiformats/go-multihash"
 
@@ -282,3 +286,326 @@ func TestProvideClearQueue(t *testing.T) {
 	require.True(t, q.queue.prefixes.IsEmptyLeaf())
 	require.Equal(t, 0, q.queue.queue.Len())
 }
+
+func TestProvidePersistAndDrainDatastore(t *testing.T) {
+	ctx := context.Background()
+	nMultihashesPerPrefix := 1 << 3
+
+	// Create and populate a queue
+	q1 := NewProvideQueue()
+	prefixes := []bitstr.Key{
+		"000",
+		"001",
+		"010",
+		"011",
+		"10",
+	}
+
+	mhMap := make(map[bitstr.Key][]mh.Multihash)
+	for _, prefix := range prefixes {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q1.Enqueue(prefix, mhs...)
+		mhMap[prefix] = mhs
+	}
+
+	require.Equal(t, len(prefixes), q1.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q1.Size())
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist the queue
+	err := q1.Persist(ctx, store, 100)
+	require.NoError(t, err)
+
+	// Verify original queue is unchanged
+	require.Equal(t, len(prefixes), q1.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q1.Size())
+
+	// Create a new queue and load from datastore
+	q2 := NewProvideQueue()
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	// Verify loaded queue matches original
+	require.Equal(t, q1.Size(), q2.Size())
+	require.Equal(t, q1.NumRegions(), q2.NumRegions())
+
+	// Verify prefix order is preserved
+	for i := range prefixes {
+		require.Equal(t, q1.queue.queue.At(i), q2.queue.queue.At(i))
+	}
+
+	// Verify all keys are present by dequeuing and comparing
+	for range prefixes {
+		prefix1, mhs1, ok1 := q1.Dequeue()
+		prefix2, mhs2, ok2 := q2.Dequeue()
+
+		require.True(t, ok1)
+		require.True(t, ok2)
+		require.Equal(t, prefix1, prefix2)
+		require.ElementsMatch(t, mhs1, mhs2)
+	}
+
+	require.True(t, q1.IsEmpty())
+	require.True(t, q2.IsEmpty())
+
+	// Verify datastore is empty after DrainDatastore
+	results, err := store.Query(ctx, query.Query{KeysOnly: true})
+	require.NoError(t, err)
+	count := 0
+	for range results.Next() {
+		count++
+	}
+	require.Equal(t, 0, count)
+}
+
+func TestProvidePersistEmptyQueue(t *testing.T) {
+	ctx := context.Background()
+
+	// Create empty queue
+	q := NewProvideQueue()
+	require.True(t, q.IsEmpty())
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist empty queue
+	err := q.Persist(ctx, store, 100)
+	require.NoError(t, err)
+
+	// Load into new queue
+	q2 := NewProvideQueue()
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	require.True(t, q2.IsEmpty())
+}
+
+func TestProvideDrainDatastoreFromEmptyDatastore(t *testing.T) {
+	ctx := context.Background()
+
+	// Create a datastore with no persisted data
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Load into queue
+	q := NewProvideQueue()
+	err := q.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	require.True(t, q.IsEmpty())
+}
+
+func TestProvideDrainDatastoreIsAdditive(t *testing.T) {
+	ctx := context.Background()
+	nMultihashesPerPrefix := 1 << 3
+
+	// Create and populate first queue
+	q1 := NewProvideQueue()
+	prefixes1 := []bitstr.Key{"000", "001"}
+	for _, prefix := range prefixes1 {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q1.Enqueue(prefix, mhs...)
+	}
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist first queue
+	err := q1.Persist(ctx, store, 100)
+	require.NoError(t, err)
+
+	// Create and populate second queue with different prefixes
+	q2 := NewProvideQueue()
+	prefixes2 := []bitstr.Key{"10", "11"}
+	for _, prefix := range prefixes2 {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q2.Enqueue(prefix, mhs...)
+	}
+
+	require.Equal(t, len(prefixes2), q2.NumRegions())
+	initialSize := q2.Size()
+
+	// Drain from datastore (should ADD to existing queue, not replace)
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	// Verify q2 now contains BOTH sets of prefixes
+	require.Equal(t, initialSize+q1.Size(), q2.Size())
+	require.Equal(t, len(prefixes1)+len(prefixes2), q2.NumRegions())
+
+	// Verify datastore is empty after DrainDatastore
+	results, err := store.Query(ctx, query.Query{KeysOnly: true})
+	require.NoError(t, err)
+	count := 0
+	for range results.Next() {
+		count++
+	}
+	require.Equal(t, 0, count)
+}
+
+func TestProvideDrainDatastoreRemovesPersisted(t *testing.T) {
+	ctx := context.Background()
+	nMultihashesPerPrefix := 1 << 3
+
+	// Create and populate a queue
+	q := NewProvideQueue()
+	prefixes := []bitstr.Key{"000", "001", "010"}
+	for _, prefix := range prefixes {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q.Enqueue(prefix, mhs...)
+	}
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist the queue
+	err := q.Persist(ctx, store, 100)
+	require.NoError(t, err)
+
+	// Verify data is in datastore
+	results, err := store.Query(ctx, query.Query{KeysOnly: true})
+	require.NoError(t, err)
+	count := 0
+	for range results.Next() {
+		count++
+	}
+	require.Greater(t, count, 0)
+
+	// Load and clear
+	q2 := NewProvideQueue()
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	// Verify data is removed from datastore
+	results, err = store.Query(ctx, query.Query{KeysOnly: true})
+	require.NoError(t, err)
+	count = 0
+	for range results.Next() {
+		count++
+	}
+	require.Equal(t, 0, count)
+
+	// Verify loaded queue matches original
+	require.Equal(t, q.Size(), q2.Size())
+	require.Equal(t, q.NumRegions(), q2.NumRegions())
+
+	// Loading from cleared datastore should result in empty queue
+	q3 := NewProvideQueue()
+	err = q3.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+	require.True(t, q3.IsEmpty())
+}
+
+func TestProvidePersistWithBatching(t *testing.T) {
+	ctx := context.Background()
+	nMultihashesPerPrefix := 1 << 3
+
+	// Create and populate a queue with enough prefixes to trigger batch commits
+	q1 := NewProvideQueue()
+	prefixes := []bitstr.Key{
+		"000",
+		"001",
+		"010",
+		"011",
+		"100",
+		"101",
+		"110",
+	}
+
+	mhMap := make(map[bitstr.Key][]mh.Multihash)
+	for _, prefix := range prefixes {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q1.Enqueue(prefix, mhs...)
+		mhMap[prefix] = mhs
+	}
+
+	require.Equal(t, len(prefixes), q1.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q1.Size())
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist with small batch size to trigger mid-loop batch commits
+	// With 7 prefixes and batchSize=3, we'll commit at i=3 and i=6, plus final commit
+	err := q1.Persist(ctx, store, 3)
+	require.NoError(t, err)
+
+	// Verify original queue is unchanged
+	require.Equal(t, len(prefixes), q1.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q1.Size())
+
+	// Create a new queue and load from datastore
+	q2 := NewProvideQueue()
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	// Verify loaded queue matches original
+	require.Equal(t, q1.Size(), q2.Size())
+	require.Equal(t, q1.NumRegions(), q2.NumRegions())
+
+	// Verify prefix order is preserved
+	for i := range prefixes {
+		require.Equal(t, q1.queue.queue.At(i), q2.queue.queue.At(i))
+	}
+
+	// Verify all keys are present by dequeuing and comparing
+	for range prefixes {
+		prefix1, mhs1, ok1 := q1.Dequeue()
+		prefix2, mhs2, ok2 := q2.Dequeue()
+
+		require.True(t, ok1)
+		require.True(t, ok2)
+		require.Equal(t, prefix1, prefix2)
+		require.ElementsMatch(t, mhs1, mhs2)
+	}
+
+	require.True(t, q1.IsEmpty())
+	require.True(t, q2.IsEmpty())
+
+	// Verify datastore is empty after DrainDatastore
+	results, err := store.Query(ctx, query.Query{KeysOnly: true})
+	require.NoError(t, err)
+	count := 0
+	for range results.Next() {
+		count++
+	}
+	require.Equal(t, 0, count)
+}
+
+func TestProvidePersistLoadWithOverlappingPrefixes(t *testing.T) {
+	ctx := context.Background()
+	nMultihashesPerPrefix := 1 << 3
+
+	// Create queue with overlapping prefixes that get consolidated
+	q1 := NewProvideQueue()
+	prefixes := []bitstr.Key{
+		"000",
+		"0000", // Should be consolidated into "000"
+	}
+
+	for _, prefix := range prefixes {
+		mhs := genMultihashesMatchingPrefix(prefix, nMultihashesPerPrefix)
+		q1.Enqueue(prefix, mhs...)
+	}
+
+	// After consolidation, should have 1 region
+	require.Equal(t, 1, q1.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q1.Size())
+
+	// Create a datastore
+	store := dssync.MutexWrap(ds.NewMapDatastore())
+
+	// Persist the queue
+	err := q1.Persist(ctx, store, 100)
+	require.NoError(t, err)
+
+	// Load into new queue
+	q2 := NewProvideQueue()
+	err = q2.DrainDatastore(ctx, store)
+	require.NoError(t, err)
+
+	// Verify loaded queue matches consolidated state
+	require.Equal(t, 1, q2.NumRegions())
+	require.Equal(t, len(prefixes)*nMultihashesPerPrefix, q2.Size())
+}
diff --git a/provider/provider_test.go b/provider/provider_test.go
index 214b5e2..6ca9e36 100644
--- a/provider/provider_test.go
+++ b/provider/provider_test.go
@@ -1497,3 +1497,111 @@ func TestClosestPeersToPrefixErrors(t *testing.T) {
 		})
 	})
 }
+
+// TestQueuePersistence tests that the provide queue is persisted when the
+// provider is closed and loaded again when it restarts.
+func TestQueuePersistence(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		pid, err := peer.Decode("12BoooooPEER")
+		require.NoError(t, err)
+
+		// Create a datastore that will be shared across provider instances
+		mapDs := ds.NewMapDatastore()
+		defer mapDs.Close()
+
+		router := &mockRouter{
+			getClosestPeersFunc: func(ctx context.Context, k string) ([]peer.ID, error) {
+				return []peer.ID{pid}, nil
+			},
+		}
+		msgSender := &mockMsgSender{
+			sendMessageFunc: func(ctx context.Context, p peer.ID, m *pb.Message) error {
+				// Don't actually send messages in this test
+				return nil
+			},
+		}
+
+		checkInterval := time.Second
+
+		// Create the first provider instance with NO workers
+		// This ensures keys stay in the queue and won't be processed
+		opts0 := []Option{
+			WithPeerID(pid),
+			WithRouter(router),
+			WithMessageSender(msgSender),
+			WithDatastore(mapDs),
+			WithSelfAddrs(func() []ma.Multiaddr {
+				addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/4001")
+				require.NoError(t, err)
+				return []ma.Multiaddr{addr}
+			}),
+			WithConnectivityCheckOnlineInterval(checkInterval),
+			WithMaxWorkers(0),               // No workers = keys stay in queue
+			WithDedicatedBurstWorkers(0),    // No dedicated burst workers
+			WithDedicatedPeriodicWorkers(0), // No dedicated periodic workers
+		}
+		prov0, err := New(opts0...)
+		require.NoError(t, err)
+
+		synctest.Wait()
+		require.True(t, prov0.connectivity.IsOnline())
+
+		// SKip the avg prefix length estimation
+		prov0.avgPrefixLenLk.Lock()
+		prov0.cachedAvgPrefixLen = 4
+		prov0.avgPrefixLenLk.Unlock()
+
+		// Generate test keys
+		mhs := random.Multihashes(8)
+
+		// Provide keys - they will be queued but not processed (no workers)
+		err = prov0.ProvideOnce(mhs...)
+		require.NoError(t, err)
+		synctest.Wait()
+
+		// Verify the queue is not empty before closing
+		queueSizeBefore := prov0.provideQueue.Size()
+		require.Equal(t, len(mhs), queueSizeBefore, "queue size should match number of provided keys")
+
+		// Close the first provider - this should persist the queue
+		err = prov0.Close()
+		require.NoError(t, err)
+
+		// Create a second provider instance with the same datastore
+		opts1 := []Option{
+			WithPeerID(pid),
+			WithRouter(router),
+			WithMessageSender(msgSender),
+			WithDatastore(mapDs),
+			WithSelfAddrs(func() []ma.Multiaddr {
+				addr, err := ma.NewMultiaddr("/ip4/127.0.0.1/tcp/4001")
+				require.NoError(t, err)
+				return []ma.Multiaddr{addr}
+			}),
+			WithConnectivityCheckOnlineInterval(checkInterval),
+			WithMaxWorkers(0),               // Still no workers to keep queue stable
+			WithDedicatedBurstWorkers(0),    // No dedicated burst workers
+			WithDedicatedPeriodicWorkers(0), // No dedicated periodic workers
+		}
+		prov1, err := New(opts1...)
+		require.NoError(t, err)
+		defer prov1.Close()
+
+		synctest.Wait()
+		require.True(t, prov1.connectivity.IsOnline())
+
+		// Verify the queue was loaded from the datastore
+		queueSizeAfter := prov1.provideQueue.Size()
+		require.Equal(t, queueSizeBefore, queueSizeAfter, "queue size should match after restart")
+		require.False(t, prov1.provideQueue.IsEmpty(), "queue should not be empty after restart")
+
+		// Verify the keys are the same by dequeuing them
+		var dequeuedKeys []mh.Multihash
+		for !prov1.provideQueue.IsEmpty() {
+			_, keys, ok := prov1.provideQueue.Dequeue()
+			require.True(t, ok)
+			dequeuedKeys = append(dequeuedKeys, keys...)
+		}
+		require.ElementsMatch(t, mhs, dequeuedKeys, "dequeued keys should match original keys")
+	})
+}