diff --git a/pkg/addressbook/addressbook.go b/pkg/addressbook/addressbook.go index 32b14e0a771..7f31a4f4e2b 100644 --- a/pkg/addressbook/addressbook.go +++ b/pkg/addressbook/addressbook.go @@ -9,6 +9,7 @@ import ( "errors" "fmt" "strings" + "time" "github.com/ethersphere/bee/v2/pkg/bzz" "github.com/ethersphere/bee/v2/pkg/storage" @@ -22,22 +23,30 @@ var _ Interface = (*store)(nil) var ErrNotFound = errors.New("addressbook: not found") // verifiedAddress pairs a bzz.Address with a flag indicating whether the peer -// has been verified. +// has been verified, and the last time the overlay was seen. type verifiedAddress struct { Address *bzz.Address `json:"address"` Verified bool `json:"verified"` + // LastSeen is the Unix timestamp (seconds) of the last time the overlay + // was seen over hive or in kademlia. Used to prune stale entries. + LastSeen int64 `json:"last_seen"` } // Interface is the AddressBook interface. type Interface interface { GetPutter Remover + // UpdateLastSeen marks the overlay as seen at the current time. + UpdateLastSeen(overlay swarm.Address) error // Overlays returns a list of all overlay addresses saved in addressbook. Overlays() ([]swarm.Address, error) // IterateOverlays exposes overlays in a form of an iterator. IterateOverlays(func(swarm.Address) (bool, error)) error // Addresses returns a list of all bzz.Address-es saved in addressbook. Addresses() ([]bzz.Address, error) + // Prune removes all entries whose overlay has not been seen since the + // given time. + Prune(before time.Time) error } type GetPutter interface { @@ -63,12 +72,14 @@ type Remover interface { type store struct { store storage.StateStorer + now func() time.Time } // New creates new addressbook for state storer. func New(storer storage.StateStorer) Interface { return &store{ store: storer, + now: time.Now, } } @@ -90,13 +101,63 @@ func (s *store) Put(overlay swarm.Address, addr bzz.Address, verified bool) (err return s.store.Put(key, &verifiedAddress{ Address: &addr, Verified: verified, + LastSeen: s.now().Unix(), }) } +// UpdateLastSeen marks the overlay as seen at the current time. It is a no-op +// if the overlay is not present in the addressbook. +func (s *store) UpdateLastSeen(overlay swarm.Address) error { + key := keyPrefix + overlay.String() + v := &verifiedAddress{} + if err := s.store.Get(key, v); err != nil { + if errors.Is(err, storage.ErrNotFound) { + return nil + } + return err + } + v.LastSeen = s.now().Unix() + return s.store.Put(key, v) +} + func (s *store) Remove(overlay swarm.Address) error { return s.store.Delete(keyPrefix + overlay.String()) } +// Prune removes all entries whose overlay has not been seen since before. +// Entries without a recorded last-seen time (LastSeen == 0) are kept, leaving +// pruning to a later run once they have been observed. +func (s *store) Prune(before time.Time) error { + cutoff := before.Unix() + + var stale []swarm.Address + err := s.store.Iterate(keyPrefix, func(key, value []byte) (stop bool, err error) { + entry := &verifiedAddress{} + if err := json.Unmarshal(value, entry); err != nil { + return true, err + } + if entry.LastSeen != 0 && entry.LastSeen < cutoff { + addr, err := swarm.ParseHexAddress(strings.TrimPrefix(string(key), keyPrefix)) + if err != nil { + return true, err + } + stale = append(stale, addr) + } + return false, nil + }) + if err != nil { + return err + } + + for _, addr := range stale { + if err := s.Remove(addr); err != nil { + return err + } + } + + return nil +} + func (s *store) IterateOverlays(cb func(swarm.Address) (bool, error)) error { return s.store.Iterate(keyPrefix, func(key, _ []byte) (stop bool, err error) { k := string(key) diff --git a/pkg/addressbook/addressbook_test.go b/pkg/addressbook/addressbook_test.go index f515527d21e..ee355e9cde0 100644 --- a/pkg/addressbook/addressbook_test.go +++ b/pkg/addressbook/addressbook_test.go @@ -7,6 +7,7 @@ package addressbook_test import ( "errors" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethersphere/bee/v2/pkg/addressbook" @@ -17,6 +18,24 @@ import ( ma "github.com/multiformats/go-multiaddr" ) +func newTestAddr(t *testing.T, overlay swarm.Address) bzz.Address { + t.Helper() + + multiaddr, err := ma.NewMultiaddr("/ip4/1.1.1.1") + if err != nil { + t.Fatal(err) + } + pk, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + bzzAddr, err := bzz.NewAddress(crypto.NewDefaultSigner(pk), []ma.Multiaddr{multiaddr}, overlay, 1, common.HexToHash("0x1").Bytes(), 1, common.Address{}) + if err != nil { + t.Fatal(err) + } + return *bzzAddr +} + type bookFunc func() (book addressbook.Interface) func TestInMem(t *testing.T) { @@ -96,3 +115,96 @@ func run(t *testing.T, f bookFunc) { t.Fatalf("expected addresses len %v, got %v", 1, len(addresses)) } } + +func TestUpdateLastSeen(t *testing.T) { + t.Parallel() + + now := time.Unix(1000, 0) + store := addressbook.NewWithClock(mock.NewStateStore(), func() time.Time { return now }) + + overlay := swarm.NewAddress([]byte{0, 1, 2, 3}) + + // UpdateLastSeen on a missing overlay is a no-op and must not create an entry. + if err := store.UpdateLastSeen(overlay); err != nil { + t.Fatal(err) + } + if _, _, err := store.Get(overlay); !errors.Is(err, addressbook.ErrNotFound) { + t.Fatalf("expected ErrNotFound, got %v", err) + } + + if err := store.Put(overlay, newTestAddr(t, overlay), true); err != nil { + 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) + if err := store.UpdateLastSeen(overlay); err != nil { + t.Fatal(err) + } + + if err := store.Prune(time.Unix(4000, 0)); err != nil { + t.Fatal(err) + } + if _, _, err := store.Get(overlay); err != nil { + t.Fatalf("entry pruned despite recent last-seen: %v", err) + } +} + +func TestPrune(t *testing.T) { + t.Parallel() + + now := time.Unix(0, 0) + store := addressbook.NewWithClock(mock.NewStateStore(), func() time.Time { return now }) + + stale := swarm.NewAddress([]byte{0, 1, 2, 3}) + fresh := swarm.NewAddress([]byte{0, 1, 2, 4}) + + now = time.Unix(1000, 0) + if err := store.Put(stale, newTestAddr(t, stale), true); err != nil { + t.Fatal(err) + } + + now = time.Unix(9000, 0) + if err := store.Put(fresh, newTestAddr(t, fresh), true); err != nil { + t.Fatal(err) + } + + if err := store.Prune(time.Unix(5000, 0)); err != nil { + t.Fatal(err) + } + + if _, _, err := store.Get(stale); !errors.Is(err, addressbook.ErrNotFound) { + t.Fatalf("stale entry should have been pruned, got err=%v", err) + } + if _, _, err := store.Get(fresh); err != nil { + t.Fatalf("fresh entry should survive: %v", err) + } +} + +func TestPruneKeepsEntriesWithoutLastSeen(t *testing.T) { + t.Parallel() + + mockStore := mock.NewStateStore() + overlay := swarm.NewAddress([]byte{0, 1, 2, 3}) + + // Seed an entry without a last_seen field, mirroring records that predate + // pruning before the stamping migration runs. + if err := mockStore.Put("addressbook_entry_"+overlay.String(), &addressbook.VerifiedAddress{ + Address: addrPtr(newTestAddr(t, overlay)), + Verified: true, + }); err != nil { + t.Fatal(err) + } + + store := addressbook.New(mockStore) + if err := store.Prune(time.Unix(5000, 0)); err != nil { + t.Fatal(err) + } + + if _, _, err := store.Get(overlay); err != nil { + t.Fatalf("entry without last_seen must not be pruned: %v", err) + } +} + +func addrPtr(a bzz.Address) *bzz.Address { return &a } diff --git a/pkg/addressbook/export_test.go b/pkg/addressbook/export_test.go index 9db017e1f27..325621e1266 100644 --- a/pkg/addressbook/export_test.go +++ b/pkg/addressbook/export_test.go @@ -4,4 +4,18 @@ package addressbook +import ( + "time" + + "github.com/ethersphere/bee/v2/pkg/storage" +) + type VerifiedAddress = verifiedAddress + +// NewWithClock creates an addressbook with an overridable clock, for testing. +func NewWithClock(storer storage.StateStorer, now func() time.Time) Interface { + return &store{ + store: storer, + now: now, + } +} diff --git a/pkg/node/node.go b/pkg/node/node.go index c6a1dc54981..2a8069763ff 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -216,6 +216,7 @@ const ( reserveMinEvictCount = 1_000 cacheMinEvictCount = 10_000 maxAllowedDoubling = 1 + addressbookPruneAfter = 30 * 24 * time.Hour // remove addressbook entries not seen for this long ) func NewBee( @@ -382,6 +383,12 @@ func NewBee( addressbook := addressbook.New(stateStore) + // Prune addressbook entries whose overlays have not been seen recently, so + // the address book does not accumulate stale peers indefinitely. + if err := addressbook.Prune(time.Now().Add(-addressbookPruneAfter)); err != nil { + logger.Warning("addressbook prune failed", "error", err) + } + logger.Info("using overlay address", "address", swarmAddress) // this will set overlay if it was not set before diff --git a/pkg/statestore/storeadapter/export_test.go b/pkg/statestore/storeadapter/export_test.go index d857e0a66d5..bffe33c70ff 100644 --- a/pkg/statestore/storeadapter/export_test.go +++ b/pkg/statestore/storeadapter/export_test.go @@ -4,8 +4,13 @@ package storeadapter -var RewriteAddressbookEnvelope = rewriteAddressbookEnvelope +var ( + RewriteAddressbookEnvelope = rewriteAddressbookEnvelope + StampAddressbookLastSeen = stampAddressbookLastSeen +) -type LegacyEntry = legacyEntry -type MigratedEntry = migratedEntry -type MigratedAddress = migratedAddress +type ( + LegacyEntry = legacyEntry + MigratedEntry = migratedEntry + MigratedAddress = migratedAddress +) diff --git a/pkg/statestore/storeadapter/migration.go b/pkg/statestore/storeadapter/migration.go index 6542a5f3c78..9f3a83ff804 100644 --- a/pkg/statestore/storeadapter/migration.go +++ b/pkg/statestore/storeadapter/migration.go @@ -7,6 +7,7 @@ package storeadapter import ( "encoding/json" "fmt" + "time" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/storage/migration" @@ -23,15 +24,16 @@ func allSteps(st storage.Store) migration.Steps { // and never execute newly added migrations. noop := func() error { return nil } return map[uint64]migration.StepFn{ - 1: noop, - 2: noop, - 3: noop, - 4: noop, - 5: noop, - 6: noop, - 7: noop, - 8: noop, - 9: rewriteAddressbookEnvelope(st), + 1: noop, + 2: noop, + 3: noop, + 4: noop, + 5: noop, + 6: noop, + 7: noop, + 8: noop, + 9: rewriteAddressbookEnvelope(st), + 10: stampAddressbookLastSeen(st), } } @@ -55,6 +57,7 @@ type migratedAddress struct { type migratedEntry struct { Address migratedAddress `json:"address"` Verified bool `json:"verified"` + LastSeen int64 `json:"last_seen"` } // rewriteAddressbookEnvelope wraps each "addressbook_entry_*" legacy @@ -118,3 +121,65 @@ func rewriteAddressbookEnvelope(s storage.Store) migration.StepFn { return nil } } + +// stampAddressbookLastSeen sets "last_seen" to the current time on every +// "addressbook_entry_*" record that lacks it, so that addresses carried over +// from before pruning was introduced are not immediately pruned. Entries that +// already carry a non-zero last_seen are left untouched. The whole record is +// preserved by merging into the decoded JSON object rather than re-encoding a +// typed struct. +func stampAddressbookLastSeen(s storage.Store) migration.StepFn { + return func() error { + store := &StateStorerAdapter{s} + + type item struct { + key string + val []byte + } + + var batch []item + if err := store.Iterate("addressbook_entry_", func(key, val []byte) (stop bool, err error) { + batch = append(batch, item{ + key: string(key), + val: append([]byte(nil), val...), + }) + return false, nil + }); err != nil { + return fmt.Errorf("iterate addressbook entries: %w", err) + } + + now := time.Now().Unix() + + for _, e := range batch { + var fields map[string]json.RawMessage + if err := json.Unmarshal(e.val, &fields); err != nil { + _ = store.Delete(e.key) + continue + } + + if raw, ok := fields["last_seen"]; ok { + var ls int64 + if json.Unmarshal(raw, &ls) == nil && ls != 0 { + continue + } + } + + stamp, err := json.Marshal(now) + if err != nil { + return fmt.Errorf("marshal last_seen: %w", err) + } + fields["last_seen"] = stamp + + out, err := json.Marshal(fields) + if err != nil { + return fmt.Errorf("marshal addressbook entry %q: %w", e.key, err) + } + + if err := store.Put(e.key, json.RawMessage(out)); err != nil { + return fmt.Errorf("stamp addressbook entry %q: %w", e.key, err) + } + } + + return nil + } +} diff --git a/pkg/statestore/storeadapter/migration_test.go b/pkg/statestore/storeadapter/migration_test.go index 3a74eb2a90b..51182dc58b5 100644 --- a/pkg/statestore/storeadapter/migration_test.go +++ b/pkg/statestore/storeadapter/migration_test.go @@ -7,6 +7,7 @@ package storeadapter_test import ( "errors" "testing" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethersphere/bee/v2/pkg/addressbook" @@ -191,6 +192,171 @@ func TestRewriteAddressbookEnvelope_AddressbookConsumes(t *testing.T) { } } +func TestStampAddressbookLastSeen(t *testing.T) { + t.Parallel() + + raw := newTestStore(t) + store, err := storeadapter.NewStateStorerAdapter(raw) + if err != nil { + t.Fatalf("NewStateStorerAdapter: %v", err) + } + + const prefix = "addressbook_entry_" + + // entry carried over from before pruning: no last_seen. + stampKey := prefix + "aabb" + if err := store.Put(stampKey, &storeadapter.MigratedEntry{ + Address: storeadapter.MigratedAddress{ + Overlay: "aabb", + Underlays: []string{"/ip4/1.1.1.1"}, + Signature: "sig==", + Nonce: "deadbeef", + Timestamp: 12345, + }, + Verified: true, + }); err != nil { + t.Fatalf("seed stamp: %v", err) + } + + // entry that already has a last_seen must not be touched. + keepKey := prefix + "ccdd" + const existingLastSeen = int64(42) + if err := store.Put(keepKey, &storeadapter.MigratedEntry{ + Address: storeadapter.MigratedAddress{Overlay: "ccdd"}, + LastSeen: existingLastSeen, + }); err != nil { + t.Fatalf("seed keep: %v", err) + } + + if err := storeadapter.StampAddressbookLastSeen(raw)(); err != nil { + t.Fatalf("migration: %v", err) + } + + var stamped storeadapter.MigratedEntry + if err := store.Get(stampKey, &stamped); err != nil { + t.Fatalf("get stamped: %v", err) + } + if stamped.LastSeen == 0 { + t.Fatal("last_seen was not stamped") + } + // other fields must survive the merge. + if !stamped.Verified || stamped.Address.Overlay != "aabb" || stamped.Address.Timestamp != 12345 { + t.Fatalf("entry mutated unexpectedly: %+v", stamped) + } + if len(stamped.Address.Underlays) != 1 || stamped.Address.Underlays[0] != "/ip4/1.1.1.1" { + t.Fatalf("underlays lost: %v", stamped.Address.Underlays) + } + + var kept storeadapter.MigratedEntry + if err := store.Get(keepKey, &kept); err != nil { + t.Fatalf("get kept: %v", err) + } + if kept.LastSeen != existingLastSeen { + t.Fatalf("existing last_seen overwritten: got %d want %d", kept.LastSeen, existingLastSeen) + } +} + +// TestAddressbookPruneRealStore drives addressbook.Prune over the production +// storage path (leveldb behind StateStorerAdapter), whose iterator key +// semantics differ from the in-memory mock used in the addressbook package's +// own tests. It confirms that overlays are correctly reconstructed from the +// iterated keys so the right entries are pruned and the survivors remain +// readable through the addressbook. +func TestAddressbookPruneRealStore(t *testing.T) { + t.Parallel() + + const ( + prefix = "addressbook_entry_" + validSignature = "c2lnbmF0dXJl" // base64("signature") + validUnderlay = "/ip4/127.0.0.1/tcp/1634" + nonceHex = "deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef" + ) + + store, err := storeadapter.NewStateStorerAdapter(newTestStore(t)) + if err != nil { + t.Fatalf("NewStateStorerAdapter: %v", err) + } + + seed := func(overlayHex string, lastSeen int64) { + t.Helper() + if err := store.Put(prefix+overlayHex, &storeadapter.MigratedEntry{ + Address: storeadapter.MigratedAddress{ + Overlay: overlayHex, + Underlays: []string{validUnderlay}, + Signature: validSignature, + Nonce: nonceHex, + }, + Verified: true, + LastSeen: lastSeen, + }); err != nil { + t.Fatalf("seed %s: %v", overlayHex, err) + } + } + + stale := swarm.MustParseHexAddress("aabb") + fresh := swarm.MustParseHexAddress("ccdd") + seed("aabb", 1000) + seed("ccdd", 9000) + + book := addressbook.New(store) + if err := book.Prune(time.Unix(5000, 0)); err != nil { + t.Fatalf("prune: %v", err) + } + + if _, _, err := book.Get(stale); !errors.Is(err, addressbook.ErrNotFound) { + t.Fatalf("stale entry should have been pruned, got err=%v", err) + } + + got, _, err := book.Get(fresh) + if err != nil { + t.Fatalf("fresh entry should survive prune: %v", err) + } + if !got.Overlay.Equal(fresh) { + t.Fatalf("survivor overlay mismatch: got %s want %s", got.Overlay, fresh) + } +} + +func TestStampAddressbookLastSeen_Idempotent(t *testing.T) { + t.Parallel() + + raw := newTestStore(t) + store, err := storeadapter.NewStateStorerAdapter(raw) + if err != nil { + t.Fatalf("NewStateStorerAdapter: %v", err) + } + + key := "addressbook_entry_aabb" + if err := store.Put(key, &storeadapter.MigratedEntry{ + Address: storeadapter.MigratedAddress{Overlay: "aabb"}, + Verified: true, + }); err != nil { + t.Fatalf("seed: %v", err) + } + + if err := storeadapter.StampAddressbookLastSeen(raw)(); err != nil { + t.Fatalf("first run: %v", err) + } + + var first storeadapter.MigratedEntry + if err := store.Get(key, &first); err != nil { + t.Fatalf("get after first run: %v", err) + } + + for i := 0; i < 2; i++ { + if err := storeadapter.StampAddressbookLastSeen(raw)(); err != nil { + t.Fatalf("rerun %d: %v", i, err) + } + } + + var got storeadapter.MigratedEntry + if err := store.Get(key, &got); err != nil { + t.Fatalf("get after repeated runs: %v", err) + } + if got.LastSeen != first.LastSeen { + t.Fatalf("last_seen changed across reruns: got %d want %d", got.LastSeen, first.LastSeen) + } +} + func TestRewriteAddressbookEnvelope_Idempotent(t *testing.T) { t.Parallel() diff --git a/pkg/topology/kademlia/kademlia.go b/pkg/topology/kademlia/kademlia.go index 57c4c41a950..6f0d3fc6a0b 100644 --- a/pkg/topology/kademlia/kademlia.go +++ b/pkg/topology/kademlia/kademlia.go @@ -455,6 +455,10 @@ func (k *Kad) connectionAttemptsHandler(ctx context.Context, wg *sync.WaitGroup, k.connectedPeers.Add(peer.addr) + if err := k.addressBook.UpdateLastSeen(peer.addr); err != nil { + k.logger.Debug("could not update last seen for peer", "peer_address", peer.addr, "error", err) + } + k.metrics.TotalOutboundConnections.Inc() k.collector.Record(peer.addr, im.PeerLogIn(time.Now(), im.PeerConnectionDirectionOutbound)) @@ -1214,6 +1218,9 @@ func (k *Kad) onConnected(ctx context.Context, addr swarm.Address) error { k.knownPeers.Add(addr) k.connectedPeers.Add(addr) k.waitNext.Remove(addr) + if err := k.addressBook.UpdateLastSeen(addr); err != nil { + k.logger.Debug("could not update last seen for peer", "peer_address", addr, "error", err) + } k.recalcDepth() k.notifyManageLoop() k.notifyPeerSig() diff --git a/pkg/topology/kademlia/lastseen_test.go b/pkg/topology/kademlia/lastseen_test.go new file mode 100644 index 00000000000..3ceb6b2daf2 --- /dev/null +++ b/pkg/topology/kademlia/lastseen_test.go @@ -0,0 +1,94 @@ +// Copyright 2026 The Swarm Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package kademlia_test + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/ethersphere/bee/v2/pkg/addressbook" + beeCrypto "github.com/ethersphere/bee/v2/pkg/crypto" + "github.com/ethersphere/bee/v2/pkg/discovery/mock" + "github.com/ethersphere/bee/v2/pkg/log" + "github.com/ethersphere/bee/v2/pkg/stabilization" + mockstate "github.com/ethersphere/bee/v2/pkg/statestore/mock" + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/ethersphere/bee/v2/pkg/topology/kademlia" + "github.com/ethersphere/bee/v2/pkg/util/testutil" +) + +type spyBook struct { + addressbook.Interface + mu sync.Mutex + seen map[string]int +} + +func (s *spyBook) UpdateLastSeen(o swarm.Address) error { + s.mu.Lock() + s.seen[o.String()]++ + s.mu.Unlock() + return s.Interface.UpdateLastSeen(o) +} + +func (s *spyBook) count(o swarm.Address) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.seen[o.String()] +} + +func TestKademliaBumpsLastSeenOnConnect(t *testing.T) { + t.Parallel() + + detector, err := stabilization.NewDetector(stabilization.Config{ + PeriodDuration: 1 * time.Second, + NumPeriodsForStabilization: 2, + StabilizationFactor: 1, + WarmupTime: 0, + }) + if err != nil { + t.Fatal(err) + } + + var conns, failed int32 + spy := &spyBook{Interface: addressbook.New(mockstate.NewStateStore()), seen: map[string]int{}} + base := swarm.RandAddress(t) + disc := mock.NewDiscovery() + + pk, _ := beeCrypto.GenerateSecp256k1Key() + signer := beeCrypto.NewDefaultSigner(pk) + p2ps := p2pMock(t, spy, signer, &conns, &failed) + + bit := -1 + kad, err := kademlia.New(base, spy, disc, p2ps, detector, log.Noop, kademlia.Options{ + BitSuffixLength: &bit, + ExcludeFunc: defaultExcludeFunc, + }) + if err != nil { + t.Fatal(err) + } + p2ps.SetPickyNotifier(kad) + if err := kad.Start(context.Background()); err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, kad) + kad.SetStorageRadius(0) + + // Inbound path: Connected -> onConnected -> UpdateLastSeen. + inbound := swarm.RandAddress(t) + connectOne(t, signer, kad, spy, inbound, nil) + if got := spy.count(inbound); got == 0 { + t.Fatalf("inbound connect did not bump last-seen for %s", inbound) + } + + // Outbound path: manage loop dials -> connect closure -> UpdateLastSeen. + outbound := swarm.RandAddressAt(t, base, 0) + addOne(t, signer, kad, spy, outbound) + waitConn(t, &conns) + if got := spy.count(outbound); got == 0 { + t.Fatalf("outbound connect did not bump last-seen for %s", outbound) + } +}