provide: handle reprovide (#1125)

* provider: adding provide and reprovide queue

* provider: network operations

* add some tests

* schedule prefix len computations

* provider schedule

* provider: handleProvide

* addressed review

* use go-test/random

* satisfy linter

* log errors during initial prefix len measurement

* address review

* satisfy linter

* address review

* provider: explore swarm

* provider: batch provide

* provider: batch reprovide

* provider: catchup pending work

* provider: options

* provide: handle reprovide

* fix panic when adding key to trie if superstring already exists

* address review

* decrease minimal region size from replicationFactor+1 to replicationFactor

* simplify unscheduleSubsumedPrefixesNoClock

* address review

* fix test to match region size (now: replicationFactor, before: replicationFactor+1)

* dequeue outside of go routine

* refactor and test groupAndScheduleKeysByPrefix

* moved maxPrefixSize const to top

* address review

* address review

* address review

* don't allocate capacity to avgPrefixLenReady

* address review

Author: guillaumemichel

provider/provider.go

@@ -235,6 +235,9 @@ func New(opts ...Option) (*SweepingProvider, error) {
 func (s *SweepingProvider) run() {
 	logger.Debug("Starting SweepingProvider")
 	go s.measureInitialPrefixLen()
+
+	// TODO: complete me
+	s.handleReprovide() // FIXME: to satisfy linter
 }
 
 // Close stops the provider and releases all resources.
@@ -724,6 +727,110 @@ func (s *SweepingProvider) provideKeysToPeer(p peer.ID, keys []mh.Multihash, pme
 	return nil
 }
 
+// handleReprovide advances the reprovider schedule and (asynchronously)
+// reprovides the region at the current schedule cursor.
+//
+// Behavior:
+//   - Determines the next region to reprovide based on the current cursor and
+//     the schedule, reprovides the region under the cursor, and moves the cursor
+//     to the next region.
+//   - Programs the schedule timer (alarm) for the next region’s reprovide
+//     time. When the timer fires, this method must be invoked again.
+//   - If the node has been blocked past the reprovide interval or if one or
+//     more regions’ times are already in the past, those regions are added to
+//     the reprovide queue for catch-up and a connectivity check is triggered.
+//   - If the node is currently offline, it skips the immediate reprovide of
+//     the current region and enqueues it to the reprovide queue for later.
+//   - If the node is online it removes the current region from the reprovide
+//     queue (if present) and starts an asynchronous batch reprovide using a
+//     periodic worker.
+func (s *SweepingProvider) handleReprovide() {
+	s.scheduleLk.Lock()
+	currentPrefix := s.scheduleCursor
+	// Get next prefix to reprovide, and set timer for it.
+	next := keyspace.NextNonEmptyLeaf(s.schedule, currentPrefix, s.order)
+
+	if next == nil {
+		// Schedule is empty, don't reprovide anything.
+		s.scheduleLk.Unlock()
+		return
+	}
+
+	var nextPrefix bitstr.Key
+	var timeUntilNextReprovide time.Duration
+	if next.Key == currentPrefix {
+		// There is a single prefix in the schedule.
+		nextPrefix = currentPrefix
+		timeUntilNextReprovide = s.timeUntil(s.reprovideTimeForPrefix(currentPrefix))
+	} else {
+		currentTimeOffset := s.currentTimeOffset()
+		timeSinceTimerRunning := s.timeBetween(s.timeOffset(s.scheduleTimerStartedAt), currentTimeOffset)
+		timeSinceTimerUntilNext := s.timeBetween(s.timeOffset(s.scheduleTimerStartedAt), next.Data)
+
+		if s.scheduleTimerStartedAt.Add(s.reprovideInterval).Before(s.clock.Now()) {
+			// Alarm was programmed more than reprovideInterval ago, which means that
+			// no regions has been reprovided since. Add all regions to the reprovide
+			// queue. This only happens if the main thread gets blocked for more than
+			// reprovideInterval.
+			nextKeyFound := false
+			scheduleEntries := keyspace.AllEntries(s.schedule, s.order)
+			next = scheduleEntries[0]
+			for _, entry := range scheduleEntries {
+				// Add all regions from the schedule to the reprovide queue. The next
+				// region to be scheduled for reprovide is the one immediately
+				// following the current time offset in the schedule.
+				if !nextKeyFound && entry.Data > currentTimeOffset {
+					next = entry
+					nextKeyFound = true
+				}
+				s.reprovideQueue.Enqueue(entry.Key)
+			}
+			// Don't reprovide any region now, but schedule the next one. All regions
+			// are expected to be reprovided when the provider is catching up with
+			// failed regions.
+			s.scheduleNextReprovideNoLock(next.Key, s.timeUntil(next.Data))
+			s.scheduleLk.Unlock()
+			return
+		}
+		if timeSinceTimerUntilNext < timeSinceTimerRunning {
+			// next is scheduled in the past. While next is in the past, add next to
+			// failedRegions and take nextLeaf as next.
+			count := 0
+			scheduleSize := s.schedule.Size()
+			for timeSinceTimerUntilNext < timeSinceTimerRunning && count < scheduleSize {
+				prefix := next.Key
+				s.reprovideQueue.Enqueue(prefix)
+				next = keyspace.NextNonEmptyLeaf(s.schedule, next.Key, s.order)
+				timeSinceTimerUntilNext = s.timeBetween(s.timeOffset(s.scheduleTimerStartedAt), next.Data)
+				count++
+			}
+		}
+
+		// next is in the future
+		nextPrefix = next.Key
+		timeUntilNextReprovide = s.timeUntil(next.Data)
+	}
+
+	s.scheduleNextReprovideNoLock(nextPrefix, timeUntilNextReprovide)
+	s.scheduleLk.Unlock()
+
+	// If we are offline, don't try to reprovide region.
+	if !s.connectivity.IsOnline() {
+		s.reprovideQueue.Enqueue(currentPrefix)
+		return
+	}
+
+	// Remove prefix that is about to be reprovided from the reprovide queue if
+	// present.
+	s.reprovideQueue.Remove(currentPrefix)
+
+	go func() {
+		s.workerPool.Acquire(periodicWorker)
+		defer s.workerPool.Release(periodicWorker)
+		s.batchReprovide(currentPrefix, true)
+	}()
+}
+
 // handleProvide provides supplied keys to the network if needed and schedules
 // the keys to be reprovided if needed.
 func (s *SweepingProvider) handleProvide(force, reprovide bool, keys ...mh.Multihash) {

provider/provider_test.go

@@ -8,10 +8,12 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/filecoin-project/go-clock"
+	"github.com/guillaumemichel/reservedpool"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/go-test/random"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -546,3 +548,118 @@ func TestIndividualProvideMultiple(t *testing.T) {
 	require.True(t, r.reprovideQueue.IsEmpty())
 	require.True(t, r.provideQueue.IsEmpty())
 }
+
+func waitUntil(t *testing.T, condition func() bool, maxDelay time.Duration, args ...any) {
+	step := time.Millisecond
+	for range maxDelay / step {
+		if condition() {
+			return
+		}
+		time.Sleep(step)
+	}
+	t.Fatal(args...)
+}
+
+func TestHandleReprovide(t *testing.T) {
+	mockClock := clock.NewMock()
+
+	online := atomic.Bool{}
+	online.Store(true)
+	connectivityCheckInterval := time.Second
+	connChecker, err := connectivity.New(
+		func() bool { return online.Load() },
+		func() {},
+		connectivity.WithClock(mockClock),
+		connectivity.WithOfflineCheckInterval(connectivityCheckInterval),
+		connectivity.WithOnlineCheckInterval(connectivityCheckInterval),
+	)
+	require.NoError(t, err)
+	defer connChecker.Close()
+
+	prov := SweepingProvider{
+		order: bit256.ZeroKey(),
+
+		connectivity: connChecker,
+
+		clock:         mockClock,
+		cycleStart:    mockClock.Now(),
+		scheduleTimer: mockClock.Timer(time.Hour),
+		schedule:      trie.New[bitstr.Key, time.Duration](),
+
+		reprovideQueue: queue.NewReprovideQueue(),
+		workerPool:     reservedpool.New[workerType](1, nil), // single worker
+
+		reprovideInterval: time.Minute,
+		maxReprovideDelay: 5 * time.Second,
+
+		getSelfAddrs: func() []ma.Multiaddr { return nil },
+	}
+	prov.scheduleTimer.Stop()
+
+	prefixes := []bitstr.Key{
+		"00",
+		"10",
+		"11",
+	}
+
+	// Empty schedule -> early return
+	prov.handleReprovide()
+	require.Zero(t, prov.scheduleCursor)
+
+	// Single prefix in schedule
+	prov.schedule.Add(prefixes[0], prov.reprovideTimeForPrefix(prefixes[0]))
+	prov.scheduleCursor = prefixes[0]
+	prov.handleReprovide()
+	require.Equal(t, prefixes[0], prov.scheduleCursor)
+
+	// Two prefixes in schedule
+	mockClock.Add(1)
+	prov.schedule.Add(prefixes[1], prov.reprovideTimeForPrefix(prefixes[1]))
+	prov.handleReprovide() // reprovides prefixes[0], set scheduleCursor to prefixes[1]
+	require.Equal(t, prefixes[1], prov.scheduleCursor)
+
+	// Wait more than reprovideInterval to call handleReprovide again.
+	// All prefixes should be added to the reprovide queue.
+	mockClock.Add(prov.reprovideInterval + 1)
+	require.True(t, prov.reprovideQueue.IsEmpty())
+	prov.handleReprovide()
+	require.Equal(t, prefixes[1], prov.scheduleCursor)
+
+	require.Equal(t, 2, prov.reprovideQueue.Size())
+	dequeued, ok := prov.reprovideQueue.Dequeue()
+	require.True(t, ok)
+	require.Equal(t, prefixes[0], dequeued)
+	dequeued, ok = prov.reprovideQueue.Dequeue()
+	require.True(t, ok)
+	require.Equal(t, prefixes[1], dequeued)
+	require.True(t, prov.reprovideQueue.IsEmpty())
+
+	// Go in time past prefixes[1] and prefixes[2]
+	prov.schedule.Add(prefixes[2], prov.reprovideTimeForPrefix(prefixes[2]))
+	mockClock.Add(3 * prov.reprovideInterval / 4)
+	// reprovides prefixes[1], add prefixes[2] to reprovide queue, set
+	// scheduleCursor to prefixes[0]
+	prov.handleReprovide()
+	require.Equal(t, prefixes[0], prov.scheduleCursor)
+
+	require.Equal(t, 1, prov.reprovideQueue.Size())
+	dequeued, ok = prov.reprovideQueue.Dequeue()
+	require.True(t, ok)
+	require.Equal(t, prefixes[2], dequeued)
+	require.True(t, prov.reprovideQueue.IsEmpty())
+
+	mockClock.Add(prov.reprovideInterval / 4)
+
+	// Node goes offline -> prefixes are queued
+	online.Store(false)
+	prov.connectivity.TriggerCheck()
+	waitUntil(t, func() bool { return !prov.connectivity.IsOnline() }, 100*time.Millisecond, "connectivity should be offline")
+	// require.True(t, prov.reprovideQueue.IsEmpty())
+	prov.handleReprovide()
+	// require.Equal(t, 1, prov.reprovideQueue.Size())
+
+	// Node comes back online
+	online.Store(true)
+	mockClock.Add(connectivityCheckInterval)
+	waitUntil(t, func() bool { return prov.connectivity.IsOnline() }, 100*time.Millisecond, "connectivity should be online")
+}