Merge pull request #200 from libp2p/marco/with-clock-2

feat: Use a clock interface in pstoreds as well

go.mod

@@ -5,6 +5,7 @@ go 1.17
 retract v0.2.9 // Contains backwards-incompatible changes. Use v0.3.0 instead.
 
 require (
+	github.com/benbjohnson/clock v1.3.0
 	github.com/gogo/protobuf v1.3.2
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/ipfs/go-datastore v0.5.0

go.sum

@@ -6,6 +6,8 @@ github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE
 github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
+github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
+github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/btcsuite/btcd v0.20.1-beta h1:Ik4hyJqN8Jfyv3S4AGBOmyouMsYE3EdYODkMbQjwPGw=
 github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
 github.com/btcsuite/btclog v0.0.0-20170628155309-84c8d2346e9f/go.mod h1:TdznJufoqS23FtqVCzL0ZqgP5MqXbb4fg/WgDys70nA=

pstoreds/addr_book.go

@@ -83,11 +83,11 @@ func (r *addrsRecord) flush(write ds.Write) (err error) {
 // * after an entry has been modified (e.g. addresses have been added or removed, TTLs updated, etc.)
 //
 // If the return value is true, the caller should perform a flush immediately to sync the record with the store.
-func (r *addrsRecord) clean() (chgd bool) {
-	now := time.Now().Unix()
+func (r *addrsRecord) clean(now time.Time) (chgd bool) {
+	nowUnix := now.Unix()
 	addrsLen := len(r.Addrs)
 
-	if !r.dirty && !r.hasExpiredAddrs(now) {
+	if !r.dirty && !r.hasExpiredAddrs(nowUnix) {
 		// record is not dirty, and we have no expired entries to purge.
 		return false
 	}
@@ -104,7 +104,7 @@ func (r *addrsRecord) clean() (chgd bool) {
 		})
 	}
 
-	r.Addrs = removeExpired(r.Addrs, now)
+	r.Addrs = removeExpired(r.Addrs, nowUnix)
 
 	return r.dirty || len(r.Addrs) != addrsLen
 }
@@ -144,6 +144,23 @@ type dsAddrBook struct {
 	// controls children goroutine lifetime.
 	childrenDone sync.WaitGroup
 	cancelFn     func()
+
+	clock clock
+}
+
+type clock interface {
+	Now() time.Time
+	After(d time.Duration) <-chan time.Time
+}
+
+type realclock struct{}
+
+func (rc realclock) Now() time.Time {
+	return time.Now()
+}
+
+func (rc realclock) After(d time.Duration) <-chan time.Time {
+	return time.After(d)
 }
 
 var _ pstore.AddrBook = (*dsAddrBook)(nil)
@@ -176,6 +193,11 @@ func NewAddrBook(ctx context.Context, store ds.Batching, opts Options) (ab *dsAd
 		opts:        opts,
 		cancelFn:    cancelFn,
 		subsManager: pstoremem.NewAddrSubManager(),
+		clock:       realclock{},
+	}
+
+	if opts.Clock != nil {
+		ab.clock = opts.Clock
 	}
 
 	if opts.CacheSize > 0 {
@@ -212,7 +234,7 @@ func (ab *dsAddrBook) loadRecord(id peer.ID, cache bool, update bool) (pr *addrs
 		pr.Lock()
 		defer pr.Unlock()
 
-		if pr.clean() && update {
+		if pr.clean(ab.clock.Now()) && update {
 			err = pr.flush(ab.ds)
 		}
 		return pr, err
@@ -231,7 +253,7 @@ func (ab *dsAddrBook) loadRecord(id peer.ID, cache bool, update bool) (pr *addrs
 			return nil, err
 		}
 		// this record is new and local for now (not in cache), so we don't need to lock.
-		if pr.clean() && update {
+		if pr.clean(ab.clock.Now()) && update {
 			err = pr.flush(ab.ds)
 		}
 	default:
@@ -383,7 +405,7 @@ func (ab *dsAddrBook) UpdateAddrs(p peer.ID, oldTTL time.Duration, newTTL time.D
 	pr.Lock()
 	defer pr.Unlock()
 
-	newExp := time.Now().Add(newTTL).Unix()
+	newExp := ab.clock.Now().Add(newTTL).Unix()
 	for _, entry := range pr.Addrs {
 		if entry.Ttl != int64(oldTTL) {
 			continue
@@ -392,7 +414,7 @@ func (ab *dsAddrBook) UpdateAddrs(p peer.ID, oldTTL time.Duration, newTTL time.D
 		pr.dirty = true
 	}
 
-	if pr.clean() {
+	if pr.clean(ab.clock.Now()) {
 		pr.flush(ab.ds)
 	}
 }
@@ -461,7 +483,7 @@ func (ab *dsAddrBook) setAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Duratio
 	// 	return nil
 	// }
 
-	newExp := time.Now().Add(ttl).Unix()
+	newExp := ab.clock.Now().Add(ttl).Unix()
 	addrsMap := make(map[string]*pb.AddrBookRecord_AddrEntry, len(pr.Addrs))
 	for _, addr := range pr.Addrs {
 		addrsMap[string(addr.Addr.Bytes())] = addr
@@ -521,7 +543,7 @@ func (ab *dsAddrBook) setAddrs(p peer.ID, addrs []ma.Multiaddr, ttl time.Duratio
 	// }
 
 	pr.dirty = true
-	pr.clean()
+	pr.clean(ab.clock.Now())
 	return pr.flush(ab.ds)
 }
 
@@ -567,7 +589,7 @@ func (ab *dsAddrBook) deleteAddrs(p peer.ID, addrs []ma.Multiaddr) (err error) {
 	pr.Addrs = deleteInPlace(pr.Addrs, addrs)
 
 	pr.dirty = true
-	pr.clean()
+	pr.clean(ab.clock.Now())
 	return pr.flush(ab.ds)
 }
 

pstoreds/addr_book_gc.go

@@ -25,7 +25,7 @@ var (
 	// queries
 	purgeLookaheadQuery = query.Query{
 		Prefix:   gcLookaheadBase.String(),
-		Orders:   []query.Order{query.OrderByKey{}},
+		Orders:   []query.Order{query.OrderByFunction(orderByTimestampInKey)},
 		KeysOnly: true,
 	}
 
@@ -95,7 +95,7 @@ func (gc *dsAddrBookGc) background() {
 	defer gc.ab.childrenDone.Done()
 
 	select {
-	case <-time.After(gc.ab.opts.GCInitialDelay):
+	case <-gc.ab.clock.After(gc.ab.opts.GCInitialDelay):
 	case <-gc.ab.ctx.Done():
 		// yield if we have been cancelled/closed before the delay elapses.
 		return
@@ -180,7 +180,7 @@ func (gc *dsAddrBookGc) purgeLookahead() {
 	}
 	defer results.Close()
 
-	now := time.Now().Unix()
+	now := gc.ab.clock.Now().Unix()
 
 	// keys: 	/peers/gc/addrs/<unix timestamp of next visit>/<peer ID b32>
 	// values: 	nil
@@ -214,7 +214,7 @@ func (gc *dsAddrBookGc) purgeLookahead() {
 		if e, ok := gc.ab.cache.Peek(id); ok {
 			cached := e.(*addrsRecord)
 			cached.Lock()
-			if cached.clean() {
+			if cached.clean(gc.ab.clock.Now()) {
 				if err = cached.flush(batch); err != nil {
 					log.Warnf("failed to flush entry modified by GC for peer: &v, err: %v", id.Pretty(), err)
 				}
@@ -239,7 +239,7 @@ func (gc *dsAddrBookGc) purgeLookahead() {
 			dropInError(gcKey, err, "unmarshalling entry")
 			continue
 		}
-		if record.clean() {
+		if record.clean(gc.ab.clock.Now()) {
 			err = record.flush(batch)
 			if err != nil {
 				log.Warnf("failed to flush entry modified by GC for peer: &v, err: %v", id.Pretty(), err)
@@ -284,7 +284,7 @@ func (gc *dsAddrBookGc) purgeStore() {
 		}
 
 		id := record.Id.ID
-		if !record.clean() {
+		if !record.clean(gc.ab.clock.Now()) {
 			continue
 		}
 
@@ -317,7 +317,7 @@ func (gc *dsAddrBookGc) populateLookahead() {
 		return
 	}
 
-	until := time.Now().Add(gc.ab.opts.GCLookaheadInterval).Unix()
+	until := gc.ab.clock.Now().Add(gc.ab.opts.GCLookaheadInterval).Unix()
 
 	var id peer.ID
 	record := &addrsRecord{AddrBookRecord: &pb.AddrBookRecord{}}
@@ -386,3 +386,25 @@ func (gc *dsAddrBookGc) populateLookahead() {
 
 	gc.currWindowEnd = until
 }
+
+// orderByTimestampInKey orders the results by comparing the timestamp in the
+// key. A lexiographic sort by itself is wrong since "10" is less than "2", but
+// as an int 2 is obviously less than 10.
+func orderByTimestampInKey(a, b query.Entry) int {
+	aKey := ds.RawKey(a.Key)
+	aInt, err := strconv.ParseInt(aKey.Parent().Name(), 10, 64)
+	if err != nil {
+		return -1
+	}
+	bKey := ds.RawKey(b.Key)
+	bInt, err := strconv.ParseInt(bKey.Parent().Name(), 10, 64)
+	if err != nil {
+		return -1
+	}
+	if aInt < bInt {
+		return -1
+	} else if aInt == bInt {
+		return 0
+	}
+	return 1
+}

pstoreds/addr_book_gc_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	mockClock "github.com/benbjohnson/clock"
 	query "github.com/ipfs/go-datastore/query"
 	pstore "github.com/libp2p/go-libp2p-core/peerstore"
 	test "github.com/libp2p/go-libp2p-peerstore/test"
@@ -90,6 +91,8 @@ func TestGCPurging(t *testing.T) {
 	opts.GCInitialDelay = 90 * time.Hour
 	opts.GCLookaheadInterval = 20 * time.Second
 	opts.GCPurgeInterval = 1 * time.Second
+	clk := mockClock.NewMock()
+	opts.Clock = clk
 
 	factory := addressBookFactory(t, leveldbStore, opts)
 	ab, closeFn := factory()
@@ -120,7 +123,7 @@ func TestGCPurging(t *testing.T) {
 		t.Errorf("expected 4 GC lookahead entries, got: %v", i)
 	}
 
-	<-time.After(2 * time.Second)
+	clk.Add(2 * time.Second)
 	gc.purgeLookahead()
 	if i := tp.countLookaheadEntries(); i != 3 {
 		t.Errorf("expected 3 GC lookahead entries, got: %v", i)
@@ -129,13 +132,13 @@ func TestGCPurging(t *testing.T) {
 	// Purge the cache, to exercise a different path in the purge cycle.
 	tp.clearCache()
 
-	<-time.After(5 * time.Second)
+	clk.Add(5 * time.Second)
 	gc.purgeLookahead()
 	if i := tp.countLookaheadEntries(); i != 3 {
 		t.Errorf("expected 3 GC lookahead entries, got: %v", i)
 	}
 
-	<-time.After(5 * time.Second)
+	clk.Add(5 * time.Second)
 	gc.purgeLookahead()
 	if i := tp.countLookaheadEntries(); i != 1 {
 		t.Errorf("expected 1 GC lookahead entries, got: %v", i)
@@ -157,6 +160,8 @@ func TestGCDelay(t *testing.T) {
 	opts.GCInitialDelay = 2 * time.Second
 	opts.GCLookaheadInterval = 1 * time.Minute
 	opts.GCPurgeInterval = 30 * time.Second
+	clk := mockClock.NewMock()
+	opts.Clock = clk
 
 	factory := addressBookFactory(t, leveldbStore, opts)
 	ab, closeFn := factory()
@@ -172,7 +177,7 @@ func TestGCDelay(t *testing.T) {
 	}
 
 	// after the initial delay has passed.
-	<-time.After(3 * time.Second)
+	clk.Add(3 * time.Second)
 	if i := tp.countLookaheadEntries(); i != 1 {
 		t.Errorf("expected 1 lookahead entry, got: %d", i)
 	}
@@ -188,6 +193,8 @@ func TestGCLookaheadDisabled(t *testing.T) {
 	opts.GCInitialDelay = 90 * time.Hour
 	opts.GCLookaheadInterval = 0 // disable lookahead
 	opts.GCPurgeInterval = 9 * time.Hour
+	clk := mockClock.NewMock()
+	opts.Clock = clk
 
 	factory := addressBookFactory(t, leveldbStore, opts)
 	ab, closeFn := factory()
@@ -206,13 +213,13 @@ func TestGCLookaheadDisabled(t *testing.T) {
 	ab.AddAddrs(ids[2], addrs[30:40], 10*time.Hour)
 	ab.AddAddrs(ids[3], addrs[40:], 10*time.Hour)
 
-	time.Sleep(100 * time.Millisecond)
+	clk.Add(100 * time.Millisecond)
 
 	if i := tp.countLookaheadEntries(); i != 0 {
 		t.Errorf("expected no GC lookahead entries, got: %v", i)
 	}
 
-	time.Sleep(500 * time.Millisecond)
+	clk.Add(500 * time.Millisecond)
 	gc := ab.(*dsAddrBook).gc
 	gc.purgeFunc()
 

pstoreds/ds_test.go

@@ -15,6 +15,8 @@ import (
 
 	pstore "github.com/libp2p/go-libp2p-core/peerstore"
 	pt "github.com/libp2p/go-libp2p-peerstore/test"
+
+	mockClock "github.com/benbjohnson/clock"
 )
 
 type datastoreFactory func(tb testing.TB) (ds.Batching, func())
@@ -50,16 +52,20 @@ func TestDsAddrBook(t *testing.T) {
 			opts := DefaultOpts()
 			opts.GCPurgeInterval = 1 * time.Second
 			opts.CacheSize = 1024
+			clk := mockClock.NewMock()
+			opts.Clock = clk
 
-			pt.TestAddrBook(t, addressBookFactory(t, dsFactory, opts))
+			pt.TestAddrBook(t, addressBookFactory(t, dsFactory, opts), clk)
 		})
 
 		t.Run(name+" Cacheless", func(t *testing.T) {
 			opts := DefaultOpts()
 			opts.GCPurgeInterval = 1 * time.Second
 			opts.CacheSize = 0
+			clk := mockClock.NewMock()
+			opts.Clock = clk
 
-			pt.TestAddrBook(t, addressBookFactory(t, dsFactory, opts))
+			pt.TestAddrBook(t, addressBookFactory(t, dsFactory, opts), clk)
 		})
 	}
 }

pstoreds/peerstore.go

@@ -34,6 +34,8 @@ type Options struct {
 	// Initial delay before GC processes start. Intended to give the system breathing room to fully boot
 	// before starting GC.
 	GCInitialDelay time.Duration
+
+	Clock clock
 }
 
 // DefaultOpts returns the default options for a persistent peerstore, with the full-purge GC algorithm:
@@ -50,6 +52,7 @@ func DefaultOpts() Options {
 		GCPurgeInterval:     2 * time.Hour,
 		GCLookaheadInterval: 0,
 		GCInitialDelay:      60 * time.Second,
+		Clock:               realclock{},
 	}
 }
 

pstoremem/inmem_test.go

@@ -7,6 +7,7 @@ import (
 
 	pstore "github.com/libp2p/go-libp2p-core/peerstore"
 
+	mockClock "github.com/benbjohnson/clock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 )
@@ -43,11 +44,12 @@ func TestPeerstoreProtoStoreLimits(t *testing.T) {
 }
 
 func TestInMemoryAddrBook(t *testing.T) {
+	clk := mockClock.NewMock()
 	pt.TestAddrBook(t, func() (pstore.AddrBook, func()) {
-		ps, err := NewPeerstore()
+		ps, err := NewPeerstore(WithClock(clk))
 		require.NoError(t, err)
 		return ps, func() { ps.Close() }
-	})
+	}, clk)
 }
 
 func TestInMemoryKeyBook(t *testing.T) {