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.