fix(addressbook): refresh last-seen on hive gossip and serialize updates

Author: martinconic

pkg/addressbook/addressbook.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/bzz"
@@ -18,6 +19,12 @@ import (
 
 const keyPrefix = "addressbook_entry_"
 
+// lastSeenUpdateInterval is the resolution at which UpdateLastSeen persists.
+// Hive reports the same peer on every gossip round, while pruning operates on
+// a scale of weeks, so a write is skipped when the stored value is already
+// this fresh.
+const lastSeenUpdateInterval = 24 * time.Hour
+
 var _ Interface = (*store)(nil)
 
 var ErrNotFound = errors.New("addressbook: not found")
@@ -54,6 +61,14 @@ type GetPutter interface {
 	Putter
 }
 
+// GetPutUpdater is the addressbook surface needed by hive: it stores peers it
+// learns about and refreshes the last-seen time of the ones it already knows.
+type GetPutUpdater interface {
+	GetPutter
+	// UpdateLastSeen marks the overlay as seen at the current time.
+	UpdateLastSeen(overlay swarm.Address) error
+}
+
 type Getter interface {
 	// Get returns the saved bzz.Address for the requested overlay together
 	// with its verification flag.
@@ -73,6 +88,10 @@ type Remover interface {
 type store struct {
 	store storage.StateStorer
 	now   func() time.Time
+
+	// mu serializes the read-modify-write in UpdateLastSeen against Put, so a
+	// concurrent Put is not rolled back by a stale copy of the entry.
+	mu sync.Mutex
 }
 
 // New creates new addressbook for state storer.
@@ -97,6 +116,9 @@ func (s *store) Get(overlay swarm.Address) (*bzz.Address, bool, error) {
 }
 
 func (s *store) Put(overlay swarm.Address, addr bzz.Address, verified bool) (err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	key := keyPrefix + overlay.String()
 	return s.store.Put(key, &verifiedAddress{
 		Address:  &addr,
@@ -106,8 +128,12 @@ func (s *store) Put(overlay swarm.Address, addr bzz.Address, verified bool) (err
 }
 
 // UpdateLastSeen marks the overlay as seen at the current time. It is a no-op
-// if the overlay is not present in the addressbook.
+// if the overlay is not present in the addressbook, or if the recorded time is
+// younger than lastSeenUpdateInterval.
 func (s *store) UpdateLastSeen(overlay swarm.Address) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	key := keyPrefix + overlay.String()
 	v := &verifiedAddress{}
 	if err := s.store.Get(key, v); err != nil {
@@ -116,7 +142,13 @@ func (s *store) UpdateLastSeen(overlay swarm.Address) error {
 		}
 		return err
 	}
-	v.LastSeen = s.now().Unix()
+
+	now := s.now().Unix()
+	if now-v.LastSeen < int64(lastSeenUpdateInterval.Seconds()) {
+		return nil
+	}
+
+	v.LastSeen = now
 	return s.store.Put(key, v)
 }
 

pkg/addressbook/addressbook_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/bzz"
 	"github.com/ethersphere/bee/v2/pkg/crypto"
 	"github.com/ethersphere/bee/v2/pkg/statestore/mock"
+	"github.com/ethersphere/bee/v2/pkg/storage"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	ma "github.com/multiformats/go-multiaddr"
 )
@@ -136,21 +137,134 @@ func TestUpdateLastSeen(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// advance the clock and bump last-seen; the entry must survive a prune at
-	// the original time.
-	now = time.Unix(5000, 0)
+	// advance the clock past the update interval and bump last-seen; the entry
+	// must survive a prune at the original time.
+	seenAt := now.Add(2 * 24 * time.Hour)
+	now = seenAt
 	if err := store.UpdateLastSeen(overlay); err != nil {
 		t.Fatal(err)
 	}
 
-	if err := store.Prune(time.Unix(4000, 0)); err != nil {
+	if err := store.Prune(seenAt.Add(-time.Hour)); err != nil {
 		t.Fatal(err)
 	}
 	if _, _, err := store.Get(overlay); err != nil {
 		t.Fatalf("entry pruned despite recent last-seen: %v", err)
 	}
 }
 
+// TestUpdateLastSeenThrottled asserts that a bump within lastSeenUpdateInterval
+// does not write. Hive calls UpdateLastSeen on every gossip sighting, so this
+// is what keeps the write rate bounded to roughly one per peer per day.
+func TestUpdateLastSeenThrottled(t *testing.T) {
+	t.Parallel()
+
+	base := time.Unix(1_000_000, 0)
+	now := base
+	mockStore := mock.NewStateStore()
+	store := addressbook.NewWithClock(mockStore, func() time.Time { return now })
+
+	overlay := swarm.NewAddress([]byte{0, 1, 2, 3})
+	if err := store.Put(overlay, newTestAddr(t, overlay), true); err != nil {
+		t.Fatal(err)
+	}
+
+	// well inside the interval: must not touch the record.
+	now = base.Add(time.Hour)
+	if err := store.UpdateLastSeen(overlay); err != nil {
+		t.Fatal(err)
+	}
+	if got := lastSeenOf(t, mockStore, overlay); got != base.Unix() {
+		t.Fatalf("throttled update wrote: last_seen = %d, want %d", got, base.Unix())
+	}
+
+	// past the interval: must write.
+	now = base.Add(addressbook.LastSeenUpdateInterval + time.Second)
+	if err := store.UpdateLastSeen(overlay); err != nil {
+		t.Fatal(err)
+	}
+	if got := lastSeenOf(t, mockStore, overlay); got != now.Unix() {
+		t.Fatalf("update past interval did not write: last_seen = %d, want %d", got, now.Unix())
+	}
+}
+
+// TestUpdateLastSeenKeepsConcurrentPut pins the read-modify-write in
+// UpdateLastSeen against a Put that lands between its read and its write.
+// Without serialization the Put's Verified flag is rolled back, which would
+// also desync the addressbook from hive's chequebook registry.
+func TestUpdateLastSeenKeepsConcurrentPut(t *testing.T) {
+	t.Parallel()
+
+	base := time.Unix(1_000_000, 0)
+	now := base
+	hooked := &hookStore{StateStorer: mock.NewStateStore()}
+	book := addressbook.NewWithClock(hooked, func() time.Time { return now })
+
+	overlay := swarm.NewAddress([]byte{0, 1, 2, 3})
+	addr := newTestAddr(t, overlay)
+
+	// a known, not-yet-verified peer.
+	if err := book.Put(overlay, addr, false); err != nil {
+		t.Fatal(err)
+	}
+	// move past the throttle so UpdateLastSeen really writes.
+	now = base.Add(addressbook.LastSeenUpdateInterval + time.Second)
+
+	// While UpdateLastSeen holds the entry it has just read, hive verifies the
+	// same peer and stores it with Verified=true.
+	started, finished := make(chan struct{}), make(chan struct{})
+	hooked.onGet = func() {
+		go func() {
+			defer close(finished)
+			close(started)
+			if err := book.Put(overlay, addr, true); err != nil {
+				t.Error(err)
+			}
+		}()
+		<-started
+		// Give the writer time to land. Serialized, it blocks on the
+		// addressbook lock until UpdateLastSeen returns; unsynchronized, its
+		// write completes here and is then overwritten below.
+		time.Sleep(100 * time.Millisecond)
+	}
+
+	if err := book.UpdateLastSeen(overlay); err != nil {
+		t.Fatal(err)
+	}
+	<-finished
+
+	if _, verified, err := book.Get(overlay); err != nil || !verified {
+		t.Fatalf("concurrent Put(verified=true) was rolled back: verified=%v err=%v", verified, err)
+	}
+}
+
+// hookStore fires onGet once, immediately after a Get returns, to interleave a
+// concurrent writer inside UpdateLastSeen's read-modify-write.
+type hookStore struct {
+	storage.StateStorer
+	onGet func()
+}
+
+func (h *hookStore) Get(key string, i any) error {
+	err := h.StateStorer.Get(key, i)
+	if h.onGet != nil {
+		f := h.onGet
+		h.onGet = nil
+		f()
+	}
+	return err
+}
+
+func lastSeenOf(t *testing.T, store storage.StateStorer, overlay swarm.Address) int64 {
+	t.Helper()
+
+	v := &addressbook.VerifiedAddress{}
+	if err := store.Get("addressbook_entry_"+overlay.String(), v); err != nil {
+		t.Fatalf("get entry: %v", err)
+	}
+	return v.LastSeen
+}
+
 func TestPrune(t *testing.T) {
 	t.Parallel()
 

pkg/addressbook/export_test.go

@@ -12,6 +12,9 @@ import (
 
 type VerifiedAddress = verifiedAddress
 
+// LastSeenUpdateInterval exposes the UpdateLastSeen write throttle.
+const LastSeenUpdateInterval = lastSeenUpdateInterval
+
 // NewWithClock creates an addressbook with an overridable clock, for testing.
 func NewWithClock(storer storage.StateStorer, now func() time.Time) Interface {
 	return &store{

pkg/hive/hive.go

@@ -70,7 +70,7 @@ type Options struct {
 
 type Service struct {
 	streamer          p2p.Streamer
-	addressBook       addressbook.GetPutter
+	addressBook       addressbook.GetPutUpdater
 	addPeersHandler   func(...swarm.Address)
 	networkID         uint64
 	logger            log.Logger
@@ -92,7 +92,7 @@ type Service struct {
 	chequebookStorer   ChequebookStorer
 }
 
-func New(streamer p2p.Streamer, addressbook addressbook.GetPutter, networkID uint64, overlay swarm.Address, logger log.Logger, o Options) *Service {
+func New(streamer p2p.Streamer, addressbook addressbook.GetPutUpdater, networkID uint64, overlay swarm.Address, logger log.Logger, o Options) *Service {
 	svc := &Service{
 		streamer:           streamer,
 		logger:             logger.WithName(loggerName).Register(),
@@ -370,6 +370,17 @@ func (s *Service) checkAndAddPeers(ctx context.Context, peers pb.Peers) {
 		}
 
 		if err := bzz.CheckTimestamp(bzzAddress.Timestamp, existing, bzz.TimestampSourceGossip, s.now()); err != nil {
+			// A peer re-presenting a record we already hold has nothing new to
+			// store, but it is still a sighting: peers mint their bzz.Address
+			// once and gossip it unchanged for their whole uptime. Refresh
+			// last-seen so peers we keep hearing about, but never dial, do not
+			// look stale to the pruner. Both errors imply a known peer, since
+			// an unknown one short-circuits inside CheckTimestamp.
+			if errors.Is(err, bzz.ErrTimestampStale) || errors.Is(err, bzz.ErrTimestampTooSoon) {
+				if err := s.addressBook.UpdateLastSeen(overlayAddr); err != nil {
+					s.logger.Debug("hive gossip: update last seen", "overlay", overlayAddr.String(), "error", err)
+				}
+			}
 			s.bumpTimestampMetric(err)
 			s.logger.Debug("hive gossip: timestamp validation failed", "overlay", overlayAddr.String(), "error", err)
 			continue