mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-07 14:09:07 +02:00
Merge remote-tracking branch 'origin/main' into revert/component-types
Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>
This commit is contained in:
@@ -599,6 +599,34 @@ func (s *SqlStore) ApproveAccountPeers(ctx context.Context, accountID string) (i
|
||||
return int(result.RowsAffected), nil
|
||||
}
|
||||
|
||||
// RefreshPeerLastSeen updates only peer_status_last_seen. Every other status
|
||||
// column is left untouched: peer_status_connected and
|
||||
// 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.
|
||||
//
|
||||
// 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.
|
||||
//
|
||||
// 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. The column is nullable — Status is an
|
||||
// embedded pointer, so a peer stored without one leaves it NULL — and NULL
|
||||
// loses every comparison, hence the explicit branch for a peer never seen.
|
||||
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 IS NULL OR peer_status_last_seen < ?)", staleBefore).
|
||||
Update("peer_status_last_seen", gorm.Expr("CURRENT_TIMESTAMP"))
|
||||
if result.Error != nil {
|
||||
return false, status.Errorf(status.Internal, "refresh peer last seen: %v", result.Error)
|
||||
}
|
||||
|
||||
return result.RowsAffected > 0, nil
|
||||
}
|
||||
|
||||
// SaveUsers saves the given list of users to the database.
|
||||
func (s *SqlStore) SaveUsers(ctx context.Context, users []*types.User) error {
|
||||
if len(users) == 0 {
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
nbpeer "github.com/netbirdio/netbird/management/server/peer"
|
||||
"github.com/netbirdio/netbird/management/server/types"
|
||||
)
|
||||
|
||||
const activityAccountID = "activityAccountId"
|
||||
|
||||
func newActivityTestStore(t *testing.T) Store {
|
||||
t.Helper()
|
||||
|
||||
store, cleanUp, err := NewTestStoreFromSQL(context.Background(), "", t.TempDir())
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(cleanUp)
|
||||
|
||||
require.NoError(t, store.SaveAccount(context.Background(), &types.Account{
|
||||
Id: activityAccountID,
|
||||
Domain: "activity.example.com",
|
||||
CreatedAt: time.Now().UTC(),
|
||||
}))
|
||||
|
||||
return store
|
||||
}
|
||||
|
||||
func TestRefreshPeerLastSeen(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newActivityTestStore(t)
|
||||
stored := time.Now().UTC().Add(-3 * time.Hour)
|
||||
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.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)
|
||||
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")
|
||||
}
|
||||
|
||||
// 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")
|
||||
}
|
||||
|
||||
// TestRefreshPeerLastSeenRecordsNeverSeenPeer covers the nullable column. Status
|
||||
// is an embedded pointer, so a peer stored without one leaves last seen NULL,
|
||||
// and NULL loses the cutoff comparison — such a peer would never record its
|
||||
// first activity.
|
||||
func TestRefreshPeerLastSeenRecordsNeverSeenPeer(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newActivityTestStore(t)
|
||||
stored := activityPeer(time.Time{})
|
||||
stored.Status = nil
|
||||
require.NoError(t, store.AddPeerToAccount(ctx, stored))
|
||||
|
||||
refreshed, err := store.RefreshPeerLastSeen(ctx, activityAccountID, "activityPeer", time.Now().UTC().Add(-time.Hour))
|
||||
require.NoError(t, err)
|
||||
assert.True(t, refreshed, "a peer that was never seen must record its first activity")
|
||||
|
||||
peer, err := store.GetPeerByID(ctx, LockingStrengthNone, activityAccountID, "activityPeer")
|
||||
require.NoError(t, err)
|
||||
assert.WithinDuration(t, time.Now().UTC(), peer.Status.LastSeen.UTC(), time.Minute, "last seen should be stamped at write time")
|
||||
}
|
||||
|
||||
// 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
|
||||
// SavePeerStatus is not reused for an activity bump.
|
||||
func TestRefreshPeerLastSeenLeavesSessionStateAlone(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
store := newActivityTestStore(t)
|
||||
|
||||
stored := activityPeer(time.Date(2026, 3, 1, 9, 0, 0, 0, time.UTC))
|
||||
stored.Status.Connected = true
|
||||
stored.Status.SessionStartedAt = 1234567890
|
||||
require.NoError(t, store.AddPeerToAccount(ctx, stored))
|
||||
|
||||
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)
|
||||
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")
|
||||
}
|
||||
|
||||
func activityPeer(lastSeen time.Time) *nbpeer.Peer {
|
||||
return &nbpeer.Peer{
|
||||
ID: "activityPeer",
|
||||
AccountID: activityAccountID,
|
||||
Key: "activityPeerKey",
|
||||
IP: netip.MustParseAddr("100.64.0.9"),
|
||||
Name: "activity-peer",
|
||||
DNSLabel: "activity-peer",
|
||||
Status: &nbpeer.PeerStatus{LastSeen: lastSeen},
|
||||
}
|
||||
}
|
||||
@@ -180,6 +180,14 @@ 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 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.
|
||||
// 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
|
||||
|
||||
@@ -3261,6 +3261,21 @@ func (mr *MockStoreMockRecorder) MarkProxyAccessTokenUsed(ctx, tokenID interface
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarkProxyAccessTokenUsed", reflect.TypeOf((*MockStore)(nil).MarkProxyAccessTokenUsed), ctx, tokenID)
|
||||
}
|
||||
|
||||
// RefreshPeerLastSeen mocks base method.
|
||||
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, 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, 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, staleBefore)
|
||||
}
|
||||
|
||||
// RemovePeerFromAllGroups mocks base method.
|
||||
func (m *MockStore) RemovePeerFromAllGroups(ctx context.Context, peerID string) error {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
Reference in New Issue
Block a user