provider: daemon (#1126)

* 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

* provider: daemon

* cancel context of external functions + tests

* 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

* close connectivity

* address review

* optimise expensive calls to trie.Size()

Author: guillaumemichel

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/gammazero/deque v1.0.0
 	github.com/google/gopacket v1.1.19
 	github.com/google/uuid v1.6.0
-	github.com/guillaumemichel/reservedpool v0.1.0
+	github.com/guillaumemichel/reservedpool v0.2.0
 	github.com/hashicorp/golang-lru v1.0.2
 	github.com/ipfs/boxo v0.33.1
 	github.com/ipfs/go-cid v0.5.0

go.sum

@@ -113,8 +113,8 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/grpc-ecosystem/grpc-gateway v1.5.0/go.mod h1:RSKVYQBd5MCa4OVpNdGskqpgL2+G+NZTnrVHpWWfpdw=
-github.com/guillaumemichel/reservedpool v0.1.0 h1:QOG3bsExi+Erk2TBzGan87uqA48Ns27uhhnwgCJup+Y=
-github.com/guillaumemichel/reservedpool v0.1.0/go.mod h1:sXSDIaef81TFdAJglsCFCMfgF5E5Z5xK1tFhjDhvbUc=
+github.com/guillaumemichel/reservedpool v0.2.0 h1:q73gtdMFJHtW+dDJ/fwtk34p7JprQv8fJSK7dEjf8Sw=
+github.com/guillaumemichel/reservedpool v0.2.0/go.mod h1:sXSDIaef81TFdAJglsCFCMfgF5E5Z5xK1tFhjDhvbUc=
 github.com/gxed/hashland/keccakpg v0.0.1/go.mod h1:kRzw3HkwxFU1mpmPP8v1WyQzwdGfmKFJ6tItnhQ67kU=
 github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmvhST0bie/0lS48=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=

provider/datastore/keystore.go

@@ -151,6 +151,7 @@ func NewKeyStore(d ds.Batching, opts ...KeyStoreOption) (*KeyStore, error) {
 		}
 	}
 	keyStore := KeyStore{
+		done:      make(chan struct{}),
 		ds:        d,
 		base:      ds.NewKey(cfg.base),
 		prefixLen: cfg.prefixBits,

provider/internal/connectivity/connectivity.go

@@ -23,6 +23,7 @@ import (
 //     is invoked exactly once.
 type ConnectivityChecker struct {
 	done      chan struct{}
+	closed    bool
 	closeOnce sync.Once
 
 	online atomic.Bool
@@ -58,17 +59,14 @@ func New(checkFunc func() bool, backOnlineNotify func(), opts ...Option) (*Conne
 }
 
 // Close stops any running connectivity checks and prevents future ones.
-func (c *ConnectivityChecker) Close() {
-	c.closeOnce.Do(func() { close(c.done) })
-}
-
-func (c *ConnectivityChecker) closed() bool {
-	select {
-	case <-c.done:
-		return true
-	default:
-		return false
-	}
+func (c *ConnectivityChecker) Close() error {
+	c.closeOnce.Do(func() {
+		close(c.done)
+		c.mutex.Lock()
+		c.closed = true
+		c.mutex.Unlock()
+	})
+	return nil
 }
 
 // IsOnline returns true if the node is currently online, false otherwise.
@@ -86,13 +84,13 @@ func (c *ConnectivityChecker) IsOnline() bool {
 //   - Exit if context is cancelled, or ConnectivityChecker is closed.
 //   - When node is found back online, run the `backOnlineNotify` callback.
 func (c *ConnectivityChecker) TriggerCheck() {
-	if c.closed() {
-		// Noop
-		return
-	}
 	if !c.mutex.TryLock() {
 		return // already checking
 	}
+	if c.closed {
+		c.mutex.Unlock()
+		return
+	}
 	if c.online.Load() && c.clock.Now().Sub(c.lastCheck) < c.onlineCheckInterval {
 		c.mutex.Unlock()
 		return // last check was too recent
@@ -117,7 +115,9 @@ func (c *ConnectivityChecker) TriggerCheck() {
 				return
 			case <-ticker.C:
 				if c.checkFunc() {
-					if !c.closed() {
+					select {
+					case <-c.done:
+					default:
 						// Node is back online.
 						c.online.Store(true)
 						c.lastCheck = c.clock.Now()