[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.
This commit is contained in:
mlsmaycon
2026-08-09 10:39:13 +00:00
parent 25b1081933
commit 796b48e49c
6 changed files with 61 additions and 22 deletions
+8 -3
View File
@@ -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.
@@ -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)
+4 -2
View File
@@ -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
+7 -6
View File
@@ -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.