From 796b48e49c2c72b26914cfdbc9ef7340d0e9a9a3 Mon Sep 17 00:00:00 2001 From: mlsmaycon Date: Sun, 9 Aug 2026 10:39:13 +0000 Subject: [PATCH] [management] Enforce the peer activity throttle inside the update The manager checked LastSeen on the peer it already held and then issued an unconditional UPDATE, so concurrent requests for one peer could each pass the check off the same stale read and write. The cutoff now travels to the store and lands in the statement's WHERE, matching how MarkPeerConnectedIfNewerSession fences its own write, and the local check stays as the query-free fast path. --- .../reverseproxy/activity/manager/manager.go | 13 +++++++--- .../activity/manager/manager_test.go | 14 ++++++---- management/server/store/sql_store.go | 11 +++++--- .../server/store/sql_store_activity_test.go | 26 +++++++++++++++++-- management/server/store/store.go | 6 +++-- management/server/store/store_mock.go | 13 +++++----- 6 files changed, 61 insertions(+), 22 deletions(-) diff --git a/management/internals/modules/reverseproxy/activity/manager/manager.go b/management/internals/modules/reverseproxy/activity/manager/manager.go index 65c592e77..978604fa0 100644 --- a/management/internals/modules/reverseproxy/activity/manager/manager.go +++ b/management/internals/modules/reverseproxy/activity/manager/manager.go @@ -38,18 +38,23 @@ func (m *managerImpl) RecordUserLogin(ctx context.Context, accountID string, use } // RecordPeerSeen stamps LastSeen, the column a peer activates its owner -// through. The throttle reads the peer the caller already holds, so a peer seen -// inside the interval costs nothing to skip. +// through. The peer the caller already holds answers the throttle without a +// query, so a peer seen inside the interval costs nothing to skip; the same +// cutoff goes to the store, which enforces it inside the UPDATE so concurrent +// requests for one peer cannot each write off their own stale read. func (m *managerImpl) RecordPeerSeen(ctx context.Context, accountID string, peer *peer.Peer) error { if peer == nil || !countsTowardActivity(peer) { return nil } - if peer.Status != nil && time.Since(peer.Status.LastSeen) < peerSeenInterval { + staleBefore := time.Now().UTC().Add(-peerSeenInterval) + if peer.Status != nil && peer.Status.LastSeen.After(staleBefore) { return nil } - return m.store.RefreshPeerLastSeen(ctx, accountID, peer.ID) + _, err := m.store.RefreshPeerLastSeen(ctx, accountID, peer.ID, staleBefore) + + return err } // countsTowardActivity reports whether the peer represents a device a person diff --git a/management/internals/modules/reverseproxy/activity/manager/manager_test.go b/management/internals/modules/reverseproxy/activity/manager/manager_test.go index 2044616ff..c89bce0a1 100644 --- a/management/internals/modules/reverseproxy/activity/manager/manager_test.go +++ b/management/internals/modules/reverseproxy/activity/manager/manager_test.go @@ -29,8 +29,9 @@ type loginWrite struct { } type seenWrite struct { - accountID string - peerID string + accountID string + peerID string + staleBefore time.Time } func (s *recordingStore) SaveUserLastLogin(_ context.Context, accountID, userID string, lastLogin time.Time) error { @@ -38,9 +39,9 @@ func (s *recordingStore) SaveUserLastLogin(_ context.Context, accountID, userID return nil } -func (s *recordingStore) RefreshPeerLastSeen(_ context.Context, accountID, peerID string) error { - s.seen = append(s.seen, seenWrite{accountID: accountID, peerID: peerID}) - return nil +func (s *recordingStore) RefreshPeerLastSeen(_ context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { + s.seen = append(s.seen, seenWrite{accountID: accountID, peerID: peerID, staleBefore: staleBefore}) + return true, nil } func TestRecordUserLogin(t *testing.T) { @@ -140,6 +141,9 @@ func TestRecordPeerSeen(t *testing.T) { require.Len(t, st.seen, 1, "exactly one activity write should have been recorded") assert.Equal(t, "account1", st.seen[0].accountID, "activity must be recorded against the service account") assert.Equal(t, tt.peer.ID, st.seen[0].peerID, "activity must be recorded against the calling peer") + assert.Equal(t, time.UTC, st.seen[0].staleBefore.Location(), "cutoffs are passed in UTC") + assert.WithinDuration(t, time.Now().UTC().Add(-peerSeenInterval), st.seen[0].staleBefore, time.Minute, + "the store must enforce the same interval the local check applies") }) } } diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go index d3c796e9c..198377312 100644 --- a/management/server/store/sql_store.go +++ b/management/server/store/sql_store.go @@ -608,16 +608,21 @@ func (s *SqlStore) ApproveAccountPeers(ctx context.Context, accountID string) (i // LastSeen comes from the database clock for the same reason it does there: a // Go-side timestamp is taken before the write and can land after a connect that // used CURRENT_TIMESTAMP, dragging the column backwards. -func (s *SqlStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error { +// +// staleBefore carries the caller's throttle into the same statement, so +// concurrent requests for one peer collapse into a single write instead of +// each racing on its own stale read. +func (s *SqlStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { result := s.db.WithContext(ctx). Model(&nbpeer.Peer{}). Where(accountAndIDQueryCondition, accountID, peerID). + Where("peer_status_last_seen < ?", staleBefore). Update("peer_status_last_seen", gorm.Expr("CURRENT_TIMESTAMP")) if result.Error != nil { - return status.Errorf(status.Internal, "refresh peer last seen: %v", result.Error) + return false, status.Errorf(status.Internal, "refresh peer last seen: %v", result.Error) } - return nil + return result.RowsAffected > 0, nil } // SaveUsers saves the given list of users to the database. diff --git a/management/server/store/sql_store_activity_test.go b/management/server/store/sql_store_activity_test.go index d34bbc0dc..77a211bf4 100644 --- a/management/server/store/sql_store_activity_test.go +++ b/management/server/store/sql_store_activity_test.go @@ -37,7 +37,9 @@ func TestRefreshPeerLastSeen(t *testing.T) { stored := time.Now().UTC().Add(-3 * time.Hour) require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored))) - require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer")) + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + assert.True(t, refreshed, "a peer seen three hours ago is stale enough to refresh") peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") require.NoError(t, err) @@ -45,6 +47,24 @@ func TestRefreshPeerLastSeen(t *testing.T) { assert.True(t, peer.Status.LastSeen.After(stored), "last seen must move forward") } +// TestRefreshPeerLastSeenHonoursCutoff covers the throttle the caller relies on: +// two concurrent requests both read the same stale peer, but only the statement +// that still finds LastSeen behind the cutoff writes. +func TestRefreshPeerLastSeenHonoursCutoff(t *testing.T) { + ctx := context.Background() + store := newActivityTestStore(t) + stored := time.Now().UTC().Add(-10 * time.Minute) + require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored))) + + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + assert.False(t, refreshed, "a peer seen inside the interval must not be written") + + peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") + require.NoError(t, err) + assert.WithinDuration(t, stored, peer.Status.LastSeen.UTC(), time.Second, "last seen must be left where it was") +} + // TestRefreshPeerLastSeenLeavesSessionStateAlone pins the column boundary: the // connected flag and the session token belong to the sync stream that owns the // peer's session, and a blind write here would corrupt its fencing. This is why @@ -58,7 +78,9 @@ func TestRefreshPeerLastSeenLeavesSessionStateAlone(t *testing.T) { stored.Status.SessionStartedAt = 1234567890 require.NoError(t, store.AddPeerToAccount(ctx, stored)) - require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer")) + refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour)) + require.NoError(t, err) + require.True(t, refreshed, "the peer is stale enough to refresh") peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer") require.NoError(t, err) diff --git a/management/server/store/store.go b/management/server/store/store.go index 7997d24e2..c0db8d626 100644 --- a/management/server/store/store.go +++ b/management/server/store/store.go @@ -184,8 +184,10 @@ type Store interface { // database clock like the other status writers. Connected and // SessionStartedAt are left alone, so this never interferes with the // session-ownership protocol MarkPeerConnectedIfNewerSession implements. - // Callers decide how often to call it; the store does not throttle. - RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error + // The write only lands when the stored LastSeen is older than + // staleBefore, which keeps a caller's throttle atomic under concurrent + // requests for the same peer. Returns true when the update happened. + RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) // MarkPeerDisconnectedIfSameSession sets the peer to disconnected and // resets SessionStartedAt to zero, but only when the stored // SessionStartedAt equals the given sessionStartedAt. LastSeen is diff --git a/management/server/store/store_mock.go b/management/server/store/store_mock.go index 8c39ab126..9b194ed12 100644 --- a/management/server/store/store_mock.go +++ b/management/server/store/store_mock.go @@ -3204,17 +3204,18 @@ func (mr *MockStoreMockRecorder) MarkProxyAccessTokenUsed(ctx, tokenID interface } // RefreshPeerLastSeen mocks base method. -func (m *MockStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error { +func (m *MockStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, staleBefore time.Time) (bool, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "RefreshPeerLastSeen", ctx, accountID, peerID) - ret0, _ := ret[0].(error) - return ret0 + ret := m.ctrl.Call(m, "RefreshPeerLastSeen", ctx, accountID, peerID, staleBefore) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 } // RefreshPeerLastSeen indicates an expected call of RefreshPeerLastSeen. -func (mr *MockStoreMockRecorder) RefreshPeerLastSeen(ctx, accountID, peerID interface{}) *gomock.Call { +func (mr *MockStoreMockRecorder) RefreshPeerLastSeen(ctx, accountID, peerID, staleBefore interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPeerLastSeen", reflect.TypeOf((*MockStore)(nil).RefreshPeerLastSeen), ctx, accountID, peerID) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPeerLastSeen", reflect.TypeOf((*MockStore)(nil).RefreshPeerLastSeen), ctx, accountID, peerID, staleBefore) } // RemovePeerFromAllGroups mocks base method.