[management] Account-scoped ephemeral peer cleanup

Replace the per-peer linked list with a per-account map keyed by
accountID. Each entry holds only the latest disconnect timestamp we
have observed for that account and a single timer that fires the next
sweep. Sweeps query the database for the authoritative stale set,
batch the deletes through peers.Manager.DeletePeers, then drop the
account from the tracker when lastDisc + lifeTime <= now (else
re-arm at horizon + cleanupWindow).

The drop rule is the entire termination story: an account stays
tracked only while OnPeerDisconnected keeps refreshing the
timestamp. There is no internal feedback loop that can advance
lastDisc on its own, so once disconnects stop the account drops in
at most one sweep.

A timestamp beats the ref-counter alternative because the counter
drifts positive in three real situations the cleanup loop has no
signal for: peers deleted via the API while offline, peers that
reconnect within the lifetime window, and management restarts. The
timestamp design never claims to know the size of the stale set —
it only knows the latest disconnect we observed and uses that to
bound when it is safe to drop the account.

OnPeerConnected becomes a no-op. The sweep query already filters
reconnected peers at the database level (peer_status_connected =
false in the WHERE clause), so there is nothing the in-memory
tracker needs to do on reconnect. The interface method is preserved
for call-site compatibility.

LoadInitialPeers no longer runs the catch-up query synchronously.
It schedules a deferred load via time.AfterFunc at a random delay
between 8 and 10 minutes. Without the jitter, every management
replica in a fleet-wide deploy would issue the catch-up query
simultaneously. The catch-up itself is one GROUP BY against the
peers table:
```sql
  SELECT account_id, MAX(peer_status_last_seen)
  FROM peers
  WHERE ephemeral = true AND peer_status_connected = false
  GROUP BY account_id
```
For each row the tracker seeds an entry and arms a sweep at
max(now, last_seen + lifeTime) + cleanupWindow — so accounts whose
backlog is already stale get cleaned soon after the delay elapses,
and accounts that disconnected recently wait the remaining window.
OnPeerDisconnected calls that arrive during the delay window seed
the tracker live, and the catch-up query skips accounts that are
already tracked.

Stop() cancels both the deferred initial-load timer and every
per-account sweep timer, and flips a stopped flag so subsequent
OnPeerDisconnected calls are ignored. This makes restarts and test
teardown clean.

Two new store methods:
  GetStaleEphemeralPeerIDsForAccount(ctx, accountID, olderThan)
  GetEphemeralAccountsLastDisconnect(ctx)
Both are scoped, indexable queries that the existing peers table
supports without schema changes.

The pending metric is renamed from
management.ephemeral.peers.pending to
management.ephemeral.accounts.tracked to reflect the new semantics
(it now counts accounts on the cleanup list, not peers). Method
names on the metrics type are unchanged so no production call site
has to move. No new metric labels, no per-account cardinality.

The algorithm was validated against an in-memory SQLite peers
table through an 11-scenario prototype kept under proto/, including
pathological-churn and 4-hour randomized simulations. All scenarios
terminate; max observed per-account sweep rate stays bounded near
the lifeTime + cleanupWindow cadence even under sustained
disconnect churn.

Verification: go build, go vet, race-clean tests across the
ephemeral, store, and telemetry packages, plus a clean
golangci-lint pass on the touched packages.

management/server/store/sql_store.go

@@ -3463,6 +3463,49 @@ func (s *SqlStore) GetAllEphemeralPeers(ctx context.Context, lockStrength Lockin
 	return allEphemeralPeers, nil
 }
 
+// GetStaleEphemeralPeerIDsForAccount returns IDs of disconnected
+// ephemeral peers in the given account whose last_seen is strictly
+// older than olderThan.
+func (s *SqlStore) GetStaleEphemeralPeerIDsForAccount(ctx context.Context, accountID string, olderThan time.Time) ([]string, error) {
+	var ids []string
+	err := s.db.WithContext(ctx).
+		Model(&nbpeer.Peer{}).
+		Where("account_id = ? AND ephemeral = ? AND peer_status_connected = ? AND peer_status_last_seen < ?",
+			accountID, true, false, olderThan).
+		Pluck("id", &ids).Error
+	if err != nil {
+		log.WithContext(ctx).Errorf("failed to query stale ephemeral peers for account %s: %v", accountID, err)
+		return nil, status.Errorf(status.Internal, "query stale ephemeral peers")
+	}
+	return ids, nil
+}
+
+// GetEphemeralAccountsLastDisconnect returns the latest peer_status_last_seen
+// per account across disconnected ephemeral peers. Returns one entry per
+// account that has at least one such peer.
+func (s *SqlStore) GetEphemeralAccountsLastDisconnect(ctx context.Context) (map[string]time.Time, error) {
+	type row struct {
+		AccountID string
+		LastSeen  time.Time
+	}
+	var rows []row
+	err := s.db.WithContext(ctx).
+		Model(&nbpeer.Peer{}).
+		Select("account_id, MAX(peer_status_last_seen) AS last_seen").
+		Where("ephemeral = ? AND peer_status_connected = ?", true, false).
+		Group("account_id").
+		Scan(&rows).Error
+	if err != nil {
+		log.WithContext(ctx).Errorf("failed to load ephemeral-account last disconnect map: %v", err)
+		return nil, status.Errorf(status.Internal, "load ephemeral accounts")
+	}
+	out := make(map[string]time.Time, len(rows))
+	for _, r := range rows {
+		out[r.AccountID] = r.LastSeen
+	}
+	return out, nil
+}
+
 // DeletePeer removes a peer from the store.
 func (s *SqlStore) DeletePeer(ctx context.Context, accountID string, peerID string) error {
 	result := s.db.Delete(&nbpeer.Peer{}, accountAndIDQueryCondition, accountID, peerID)

management/server/store/store.go

@@ -165,6 +165,15 @@ type Store interface {
 	GetAccountPeersWithExpiration(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*nbpeer.Peer, error)
 	GetAccountPeersWithInactivity(ctx context.Context, lockStrength LockingStrength, accountID string) ([]*nbpeer.Peer, error)
 	GetAllEphemeralPeers(ctx context.Context, lockStrength LockingStrength) ([]*nbpeer.Peer, error)
+	// GetStaleEphemeralPeerIDsForAccount returns the IDs of disconnected
+	// ephemeral peers whose last_seen is strictly older than olderThan,
+	// scoped to a single account. Used by the per-account cleanup sweep.
+	GetStaleEphemeralPeerIDsForAccount(ctx context.Context, accountID string, olderThan time.Time) ([]string, error)
+	// GetEphemeralAccountsLastDisconnect returns, for every account that
+	// has at least one disconnected ephemeral peer, the most recent
+	// last_seen across that account's disconnected ephemeral peers. Used
+	// to reconstruct the per-account cleanup tracker after a restart.
+	GetEphemeralAccountsLastDisconnect(ctx context.Context) (map[string]time.Time, error)
 	SavePeer(ctx context.Context, accountID string, peer *nbpeer.Peer) error
 	SavePeerStatus(ctx context.Context, accountID, peerID string, status nbpeer.PeerStatus) error
 	// MarkPeerConnectedIfNewerSession sets the peer to connected with the

management/server/store/store_mock.go

@@ -1376,6 +1376,36 @@ func (mr *MockStoreMockRecorder) GetAllEphemeralPeers(ctx, lockStrength interfac
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetAllEphemeralPeers", reflect.TypeOf((*MockStore)(nil).GetAllEphemeralPeers), ctx, lockStrength)
 }
 
+// GetStaleEphemeralPeerIDsForAccount mocks base method.
+func (m *MockStore) GetStaleEphemeralPeerIDsForAccount(ctx context.Context, accountID string, olderThan time.Time) ([]string, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetStaleEphemeralPeerIDsForAccount", ctx, accountID, olderThan)
+	ret0, _ := ret[0].([]string)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetStaleEphemeralPeerIDsForAccount indicates an expected call of GetStaleEphemeralPeerIDsForAccount.
+func (mr *MockStoreMockRecorder) GetStaleEphemeralPeerIDsForAccount(ctx, accountID, olderThan interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetStaleEphemeralPeerIDsForAccount", reflect.TypeOf((*MockStore)(nil).GetStaleEphemeralPeerIDsForAccount), ctx, accountID, olderThan)
+}
+
+// GetEphemeralAccountsLastDisconnect mocks base method.
+func (m *MockStore) GetEphemeralAccountsLastDisconnect(ctx context.Context) (map[string]time.Time, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetEphemeralAccountsLastDisconnect", ctx)
+	ret0, _ := ret[0].(map[string]time.Time)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetEphemeralAccountsLastDisconnect indicates an expected call of GetEphemeralAccountsLastDisconnect.
+func (mr *MockStoreMockRecorder) GetEphemeralAccountsLastDisconnect(ctx interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEphemeralAccountsLastDisconnect", reflect.TypeOf((*MockStore)(nil).GetEphemeralAccountsLastDisconnect), ctx)
+}
+
 // GetAllProxyAccessTokens mocks base method.
 func (m *MockStore) GetAllProxyAccessTokens(ctx context.Context, lockStrength LockingStrength) ([]*types2.ProxyAccessToken, error) {
 	m.ctrl.T.Helper()

management/server/telemetry/ephemeral_metrics.go

@@ -7,9 +7,9 @@ import (
 )
 
 // EphemeralPeersMetrics tracks the ephemeral peer cleanup pipeline: how
-// many peers are currently scheduled for deletion, how many tick runs
-// the cleaner has performed, how many peers it has removed, and how
-// many delete batches failed.
+// many accounts are currently being tracked for cleanup, how many sweep
+// runs deleted at least one peer, how many peers have been removed, and
+// how many delete batches failed.
 type EphemeralPeersMetrics struct {
 	ctx context.Context
 
@@ -21,16 +21,16 @@ type EphemeralPeersMetrics struct {
 
 // NewEphemeralPeersMetrics constructs the ephemeral cleanup counters.
 func NewEphemeralPeersMetrics(ctx context.Context, meter metric.Meter) (*EphemeralPeersMetrics, error) {
-	pending, err := meter.Int64UpDownCounter("management.ephemeral.peers.pending",
+	pending, err := meter.Int64UpDownCounter("management.ephemeral.accounts.tracked",
 		metric.WithUnit("1"),
-		metric.WithDescription("Number of ephemeral peers currently waiting to be cleaned up"))
+		metric.WithDescription("Number of accounts currently tracked for ephemeral peer cleanup"))
 	if err != nil {
 		return nil, err
 	}
 
 	cleanupRuns, err := meter.Int64Counter("management.ephemeral.cleanup.runs.counter",
 		metric.WithUnit("1"),
-		metric.WithDescription("Number of ephemeral cleanup ticks that processed at least one peer"))
+		metric.WithDescription("Number of ephemeral cleanup sweeps that deleted at least one peer"))
 	if err != nil {
 		return nil, err
 	}
@@ -61,7 +61,8 @@ func NewEphemeralPeersMetrics(ctx context.Context, meter metric.Meter) (*Ephemer
 // All methods are nil-receiver safe so callers that haven't wired metrics
 // (tests, self-hosted with metrics off) can invoke them unconditionally.
 
-// IncPending bumps the pending gauge when a peer is added to the cleanup list.
+// IncPending bumps the tracked-accounts gauge when a new account
+// becomes eligible for ephemeral cleanup tracking.
 func (m *EphemeralPeersMetrics) IncPending() {
 	if m == nil {
 		return
@@ -69,8 +70,8 @@ func (m *EphemeralPeersMetrics) IncPending() {
 	m.pending.Add(m.ctx, 1)
 }
 
-// AddPending bumps the pending gauge by n — used at startup when the
-// initial set of ephemeral peers is loaded from the store.
+// AddPending bumps the tracked-accounts gauge by n — used at startup
+// when the catch-up query seeds the tracker.
 func (m *EphemeralPeersMetrics) AddPending(n int64) {
 	if m == nil || n <= 0 {
 		return
@@ -78,9 +79,8 @@ func (m *EphemeralPeersMetrics) AddPending(n int64) {
 	m.pending.Add(m.ctx, n)
 }
 
-// DecPending decreases the pending gauge — used both when a peer reconnects
-// before its deadline (removed from the list) and when a cleanup tick
-// actually deletes it.
+// DecPending decreases the tracked-accounts gauge when an account is
+// dropped from the tracker (no more disconnects to chase).
 func (m *EphemeralPeersMetrics) DecPending(n int64) {
 	if m == nil || n <= 0 {
 		return