mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-28 07:19:08 +01:00
[management] Stamp proxy peer activity with the database clock
The activity write took a Go-side timestamp, which is exactly what MarkPeerConnectedIfNewerSession documents as the cause of previous ordering bugs: a value read before the write can land after a connect that stamped CURRENT_TIMESTAMP, dragging LastSeen backwards. The write now uses the database clock like the other status writers, so the column only ever moves forward. The throttle is unaffected; it reads the peer already in hand and never needed the write's timestamp.
This commit is contained in:
@@ -604,15 +604,15 @@ func (s *SqlStore) ApproveAccountPeers(ctx context.Context, accountID string) (i
|
||||
// peer_status_session_started_at belong to the sync stream that owns the
|
||||
// session, and a blind write here would corrupt the fencing
|
||||
// MarkPeerConnectedIfNewerSession relies on.
|
||||
func (s *SqlStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, seenAt time.Time) error {
|
||||
if seenAt.IsZero() {
|
||||
return nil
|
||||
}
|
||||
|
||||
//
|
||||
// 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 {
|
||||
result := s.db.WithContext(ctx).
|
||||
Model(&nbpeer.Peer{}).
|
||||
Where(accountAndIDQueryCondition, accountID, peerID).
|
||||
Update("peer_status_last_seen", seenAt)
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -104,27 +104,15 @@ func TestRefreshUserLastLoginUnknownUserIsNotAnError(t *testing.T) {
|
||||
func TestRefreshPeerLastSeen(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newActivityTestStore(t)
|
||||
require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC))))
|
||||
|
||||
seenAt := time.Date(2026, 3, 1, 12, 0, 0, 0, time.UTC)
|
||||
require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", seenAt))
|
||||
|
||||
peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
|
||||
require.NoError(t, err)
|
||||
assert.WithinDuration(t, seenAt, peer.Status.LastSeen.UTC(), time.Second, "unexpected stored last seen")
|
||||
}
|
||||
|
||||
func TestRefreshPeerLastSeenZeroIsIgnored(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newActivityTestStore(t)
|
||||
stored := time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC)
|
||||
stored := time.Now().UTC().Add(-3 * time.Hour)
|
||||
require.NoError(t, store.AddPeerToAccount(ctx, activityPeer(stored)))
|
||||
|
||||
require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Time{}))
|
||||
require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer"))
|
||||
|
||||
peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
|
||||
require.NoError(t, err)
|
||||
assert.WithinDuration(t, stored, peer.Status.LastSeen.UTC(), time.Second, "a zero timestamp must not clear last seen")
|
||||
assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time")
|
||||
assert.True(t, peer.Status.LastSeen.After(stored), "last seen must move forward")
|
||||
}
|
||||
|
||||
// TestRefreshPeerLastSeenLeavesSessionStateAlone pins the column boundary: the
|
||||
@@ -139,12 +127,11 @@ func TestRefreshPeerLastSeenLeavesSessionStateAlone(t *testing.T) {
|
||||
stored.Status.SessionStartedAt = 1234567890
|
||||
require.NoError(t, store.AddPeerToAccount(ctx, stored))
|
||||
|
||||
seenAt := time.Date(2026, 3, 1, 12, 0, 0, 0, time.UTC)
|
||||
require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", seenAt))
|
||||
require.NoError(t, store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer"))
|
||||
|
||||
peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
|
||||
require.NoError(t, err)
|
||||
assert.WithinDuration(t, seenAt, peer.Status.LastSeen.UTC(), time.Second, "last seen should move forward")
|
||||
assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should move forward")
|
||||
assert.True(t, peer.Status.Connected, "connected flag must survive an activity write")
|
||||
assert.Equal(t, int64(1234567890), peer.Status.SessionStartedAt, "session token must survive an activity write")
|
||||
}
|
||||
|
||||
@@ -185,11 +185,12 @@ type Store interface {
|
||||
// Returns true when the update happened, false when this stream lost
|
||||
// the race against a newer session.
|
||||
MarkPeerConnectedIfNewerSession(ctx context.Context, accountID, peerID string, newSessionStartedAt int64) (bool, error)
|
||||
// RefreshPeerLastSeen records that a peer was seen at seenAt. Connected and
|
||||
// RefreshPeerLastSeen records that a peer was just seen, stamping the
|
||||
// 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, seenAt time.Time) error
|
||||
RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error
|
||||
// MarkPeerDisconnectedIfSameSession sets the peer to disconnected and
|
||||
// resets SessionStartedAt to zero, but only when the stored
|
||||
// SessionStartedAt equals the given sessionStartedAt. LastSeen is
|
||||
|
||||
@@ -3204,17 +3204,17 @@ func (mr *MockStoreMockRecorder) MarkProxyAccessTokenUsed(ctx, tokenID interface
|
||||
}
|
||||
|
||||
// RefreshPeerLastSeen mocks base method.
|
||||
func (m *MockStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string, seenAt time.Time) error {
|
||||
func (m *MockStore) RefreshPeerLastSeen(ctx context.Context, accountID, peerID string) error {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "RefreshPeerLastSeen", ctx, accountID, peerID, seenAt)
|
||||
ret := m.ctrl.Call(m, "RefreshPeerLastSeen", ctx, accountID, peerID)
|
||||
ret0, _ := ret[0].(error)
|
||||
return ret0
|
||||
}
|
||||
|
||||
// RefreshPeerLastSeen indicates an expected call of RefreshPeerLastSeen.
|
||||
func (mr *MockStoreMockRecorder) RefreshPeerLastSeen(ctx, accountID, peerID, seenAt interface{}) *gomock.Call {
|
||||
func (mr *MockStoreMockRecorder) RefreshPeerLastSeen(ctx, accountID, peerID interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPeerLastSeen", reflect.TypeOf((*MockStore)(nil).RefreshPeerLastSeen), ctx, accountID, peerID, seenAt)
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshPeerLastSeen", reflect.TypeOf((*MockStore)(nil).RefreshPeerLastSeen), ctx, accountID, peerID)
|
||||
}
|
||||
|
||||
// RefreshUserLastLogin mocks base method.
|
||||
|
||||
Reference in New Issue
Block a user