Change the logic and add moc data

client/internal/lazyconn/inactivity/inactivity.go

@@ -1,70 +0,0 @@
-package inactivity
-
-import (
-	"context"
-	"time"
-
-	peer "github.com/netbirdio/netbird/client/internal/peer/id"
-)
-
-const (
-	DefaultInactivityThreshold = 60 * time.Minute // idle after 1 hour inactivity
-	MinimumInactivityThreshold = 3 * time.Minute
-)
-
-type Monitor struct {
-	id                  peer.ConnID
-	timer               *time.Timer
-	cancel              context.CancelFunc
-	inactivityThreshold time.Duration
-}
-
-func NewInactivityMonitor(peerID peer.ConnID, threshold time.Duration) *Monitor {
-	i := &Monitor{
-		id:                  peerID,
-		timer:               time.NewTimer(0),
-		inactivityThreshold: threshold,
-	}
-	i.timer.Stop()
-	return i
-}
-
-func (i *Monitor) Start(ctx context.Context, timeoutChan chan peer.ConnID) {
-	i.timer.Reset(i.inactivityThreshold)
-	defer i.timer.Stop()
-
-	ctx, i.cancel = context.WithCancel(ctx)
-	defer func() {
-		defer i.cancel()
-		select {
-		case <-i.timer.C:
-		default:
-		}
-	}()
-
-	select {
-	case <-i.timer.C:
-		select {
-		case timeoutChan <- i.id:
-		case <-ctx.Done():
-			return
-		}
-	case <-ctx.Done():
-		return
-	}
-}
-
-func (i *Monitor) Stop() {
-	if i.cancel == nil {
-		return
-	}
-	i.cancel()
-}
-
-func (i *Monitor) PauseTimer() {
-	i.timer.Stop()
-}
-
-func (i *Monitor) ResetTimer() {
-	i.timer.Reset(i.inactivityThreshold)
-}

client/internal/lazyconn/inactivity/inactivity_test.go

@@ -1,156 +0,0 @@
-package inactivity
-
-import (
-	"context"
-	"testing"
-	"time"
-
-	peerid "github.com/netbirdio/netbird/client/internal/peer/id"
-)
-
-type MocPeer struct {
-}
-
-func (m *MocPeer) ConnID() peerid.ConnID {
-	return peerid.ConnID(m)
-}
-
-func TestInactivityMonitor(t *testing.T) {
-	tCtx, testTimeoutCancel := context.WithTimeout(context.Background(), time.Second*5)
-	defer testTimeoutCancel()
-
-	p := &MocPeer{}
-	im := NewInactivityMonitor(p.ConnID(), time.Second*2)
-
-	timeoutChan := make(chan peerid.ConnID)
-
-	exitChan := make(chan struct{})
-
-	go func() {
-		defer close(exitChan)
-		im.Start(tCtx, timeoutChan)
-	}()
-
-	select {
-	case <-timeoutChan:
-	case <-tCtx.Done():
-		t.Fatal("timeout")
-	}
-
-	select {
-	case <-exitChan:
-	case <-tCtx.Done():
-		t.Fatal("timeout")
-	}
-}
-
-func TestReuseInactivityMonitor(t *testing.T) {
-	p := &MocPeer{}
-	im := NewInactivityMonitor(p.ConnID(), time.Second*2)
-
-	timeoutChan := make(chan peerid.ConnID)
-
-	for i := 2; i > 0; i-- {
-		exitChan := make(chan struct{})
-
-		testTimeoutCtx, testTimeoutCancel := context.WithTimeout(context.Background(), time.Second*5)
-
-		go func() {
-			defer close(exitChan)
-			im.Start(testTimeoutCtx, timeoutChan)
-		}()
-
-		select {
-		case <-timeoutChan:
-		case <-testTimeoutCtx.Done():
-			t.Fatal("timeout")
-		}
-
-		select {
-		case <-exitChan:
-		case <-testTimeoutCtx.Done():
-			t.Fatal("timeout")
-		}
-		testTimeoutCancel()
-	}
-}
-
-func TestStopInactivityMonitor(t *testing.T) {
-	tCtx, testTimeoutCancel := context.WithTimeout(context.Background(), time.Second*5)
-	defer testTimeoutCancel()
-
-	p := &MocPeer{}
-	im := NewInactivityMonitor(p.ConnID(), DefaultInactivityThreshold)
-
-	timeoutChan := make(chan peerid.ConnID)
-
-	exitChan := make(chan struct{})
-
-	go func() {
-		defer close(exitChan)
-		im.Start(tCtx, timeoutChan)
-	}()
-
-	go func() {
-		time.Sleep(3 * time.Second)
-		im.Stop()
-	}()
-
-	select {
-	case <-timeoutChan:
-		t.Fatal("unexpected timeout")
-	case <-exitChan:
-	case <-tCtx.Done():
-		t.Fatal("timeout")
-	}
-}
-
-func TestPauseInactivityMonitor(t *testing.T) {
-	tCtx, testTimeoutCancel := context.WithTimeout(context.Background(), time.Second*10)
-	defer testTimeoutCancel()
-
-	p := &MocPeer{}
-	trashHold := time.Second * 3
-	im := NewInactivityMonitor(p.ConnID(), trashHold)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	timeoutChan := make(chan peerid.ConnID)
-
-	exitChan := make(chan struct{})
-
-	go func() {
-		defer close(exitChan)
-		im.Start(ctx, timeoutChan)
-	}()
-
-	time.Sleep(1 * time.Second) // grant time to start the monitor
-	im.PauseTimer()
-
-	// check to do not receive timeout
-	thresholdCtx, thresholdCancel := context.WithTimeout(context.Background(), trashHold+time.Second)
-	defer thresholdCancel()
-	select {
-	case <-exitChan:
-		t.Fatal("unexpected exit")
-	case <-timeoutChan:
-		t.Fatal("unexpected timeout")
-	case <-thresholdCtx.Done():
-		// test ok
-	case <-tCtx.Done():
-		t.Fatal("test timed out")
-	}
-
-	// test reset timer
-	im.ResetTimer()
-
-	select {
-	case <-tCtx.Done():
-		t.Fatal("test timed out")
-	case <-exitChan:
-		t.Fatal("unexpected exit")
-	case <-timeoutChan:
-		// expected timeout
-	}
-}

client/internal/lazyconn/inactivity/manager.go

@@ -0,0 +1,174 @@
+package inactivity
+
+import (
+	"context"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/iface/configurer"
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+)
+
+// Responder: vmp2
+// - Receive handshake initiation: 148 bytes + extra 32 bytes, every 02:00 - 03:00 minutes
+// - Receive keep alive: 32 bytes, every 25 sec
+// Initiator: mp1
+// - Receive handshake response:
+// - Receive keep alive: 32 bytes, every 25 sec
+
+const (
+	keepAliveBytes       = 32
+	keepAliveInterval    = 25 * time.Second
+	handshakeInitBytes   = 148
+	handshakeRespBytes   = 92
+	handshakeMaxInterval = 3 * time.Minute
+
+	checkInterval        = keepAliveInterval // todo: 5 * time.Second
+	keepAliveCheckPeriod = keepAliveInterval
+)
+
+const (
+	// todo make it configurable
+	DefaultInactivityThreshold = 60 * time.Minute // idle after 1 hour inactivity
+	MinimumInactivityThreshold = 3 * time.Minute
+)
+
+type WgInterface interface {
+	GetStats() (map[string]configurer.WGStats, error)
+}
+
+type peerInfo struct {
+	lastRxBytesAtLastIdleCheck int64
+	lastIdleCheckAt            time.Time
+	inActivityInRow            int
+	log                        *log.Entry
+}
+
+type Manager struct {
+	InactivePeersChan chan []string
+	iface             WgInterface
+	interestedPeers   map[string]*peerInfo
+}
+
+func NewManager(iface WgInterface) *Manager {
+	return &Manager{
+		InactivePeersChan: make(chan []string, 1),
+		iface:             iface,
+		interestedPeers:   make(map[string]*peerInfo),
+	}
+}
+
+func (m *Manager) AddPeer(peerCfg *lazyconn.PeerConfig) {
+	if _, exists := m.interestedPeers[peerCfg.PublicKey]; !exists {
+		m.interestedPeers[peerCfg.PublicKey] = &peerInfo{
+			log: peerCfg.Log,
+		}
+	}
+}
+
+func (m *Manager) RemovePeer(peer string) {
+	delete(m.interestedPeers, peer)
+}
+
+func (m *Manager) Start(ctx context.Context) {
+	ticker := newTicker(checkInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case tickTime := <-ticker.C():
+			idlePeers, err := m.checkStats(tickTime)
+			if err != nil {
+				log.Errorf("error checking stats: %v", err)
+				return
+			}
+
+			if len(idlePeers) == 0 {
+				continue
+			}
+			select {
+			case m.InactivePeersChan <- idlePeers:
+			case <-ctx.Done():
+				continue
+			default:
+				continue
+			}
+		}
+	}
+}
+
+func (m *Manager) checkStats(now time.Time) ([]string, error) {
+	stats, err := m.iface.GetStats()
+	if err != nil {
+		return nil, err
+	}
+
+	var idlePeers []string
+
+	for peer, info := range m.interestedPeers {
+		stat, found := stats[peer]
+		if !found {
+			info.log.Warnf("peer not found in wg stats")
+			continue
+		}
+
+		// First measurement: initialize
+		if info.lastIdleCheckAt.IsZero() {
+			info.lastIdleCheckAt = now
+			info.lastRxBytesAtLastIdleCheck = stat.RxBytes
+			info.log.Infof("initializing RxBytes: %v, %v", now, stat.RxBytes)
+			continue
+		}
+
+		// check only every idleCheckDuration
+		if shouldSkipIdleCheck(now, info.lastIdleCheckAt) {
+			continue
+		}
+
+		// sometimes we measure false inactivity, so we need to check if we have activity in a row
+		inactive := isInactive(stat, info)
+		if inactive {
+			info.inActivityInRow++
+		} else {
+			info.inActivityInRow = 0
+		}
+
+		if info.inActivityInRow >= 3 {
+			info.log.Infof("peer is inactive for %d checks, marking as inactive", info.inActivityInRow)
+			idlePeers = append(idlePeers, peer)
+		}
+		info.lastIdleCheckAt = now
+		info.lastRxBytesAtLastIdleCheck = stat.RxBytes
+	}
+
+	return idlePeers, nil
+}
+
+func isInactive(stat configurer.WGStats, info *peerInfo) bool {
+	rxSyncPrevPeriod := stat.RxBytes - info.lastRxBytesAtLastIdleCheck
+	switch rxSyncPrevPeriod {
+	case 0:
+		info.log.Tracef("peer inactive, received 0 bytes")
+		return true
+	case keepAliveBytes:
+		info.log.Tracef("peer inactive, only keep alive received, current RxBytes: %d", rxSyncPrevPeriod)
+		return true
+	case handshakeInitBytes + keepAliveBytes:
+		info.log.Tracef("peer inactive, only handshakeInitBytes + keepAliveBytes, current RxBytes: %d", rxSyncPrevPeriod)
+		return true
+	case handshakeRespBytes + keepAliveBytes:
+		info.log.Tracef("peer inactive, only handshakeRespBytes + keepAliveBytes, current RxBytes: %d", rxSyncPrevPeriod)
+		return true
+	default:
+		info.log.Infof("active, RxBytes: %d", rxSyncPrevPeriod)
+		return false
+	}
+}
+
+func shouldSkipIdleCheck(now, lastIdleCheckAt time.Time) bool {
+	minDuration := keepAliveCheckPeriod - (checkInterval / 2)
+	return now.Sub(lastIdleCheckAt) < minDuration
+}

client/internal/lazyconn/inactivity/manager_test.go

@@ -0,0 +1,46 @@
+package inactivity
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/netbirdio/netbird/util"
+)
+
+func init() {
+	_ = util.InitLog("trace", "console")
+}
+
+func TestNewManager(t *testing.T) {
+	for i, sc := range scenarios {
+		timer := NewFakeTimer()
+		newTicker = func(d time.Duration) Ticker {
+			return newFakeTicker(d, timer)
+		}
+
+		t.Run(fmt.Sprintf("Scenario %d", i), func(t *testing.T) {
+			mock := newMockWgInterface("peer1", sc.Data, timer)
+			manager := NewManager(mock)
+			manager.AddPeer("peer1")
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			manager.Start(ctx)
+
+			var inactiveResult bool
+			select {
+			case <-manager.InactivePeersChan:
+				inactiveResult = true
+			default:
+				inactiveResult = false
+			}
+
+			if inactiveResult != sc.ExpectedInactive {
+				t.Errorf("Expected inactive peers: %v, got: %v", sc.ExpectedInactive, inactiveResult)
+			}
+		})
+	}
+}

client/internal/lazyconn/inactivity/moc_test.go

@@ -0,0 +1,102 @@
+package inactivity
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/netbirdio/netbird/client/iface/configurer"
+)
+
+type rxHistory struct {
+	when    time.Duration
+	RxBytes int64
+}
+
+// mockWgInterface mocks WgInterface to simulate peer stats.
+type mockWgInterface struct {
+	peerID        string
+	statsSequence []rxHistory
+	timer         *FakeTimer
+	initialTime   time.Time
+	reachedLast   bool
+}
+
+func newMockWgInterface(peerID string, history []rxHistory, timer *FakeTimer) *mockWgInterface {
+	return &mockWgInterface{
+		peerID:        peerID,
+		statsSequence: history,
+		timer:         timer,
+		initialTime:   timer.Now(),
+	}
+}
+
+func (m *mockWgInterface) GetStats() (map[string]configurer.WGStats, error) {
+	if m.reachedLast {
+		return nil, fmt.Errorf("no more data")
+	}
+
+	now := m.timer.Now()
+	var rx int64
+	for i, history := range m.statsSequence {
+		if now.Before(m.initialTime.Add(history.when)) {
+			break
+		}
+
+		if len(m.statsSequence)-1 == i {
+			m.reachedLast = true
+		}
+
+		rx += history.RxBytes
+	}
+
+	wgStats := make(map[string]configurer.WGStats)
+	wgStats[m.peerID] = configurer.WGStats{
+		RxBytes: rx,
+	}
+	return wgStats, nil
+}
+
+// fakeTicker is a controllable ticker for use in tests
+type fakeTicker struct {
+	interval time.Duration
+	timer    *FakeTimer
+
+	ch  chan time.Time
+	now time.Time
+}
+
+func newFakeTicker(interval time.Duration, timer *FakeTimer) *fakeTicker {
+	return &fakeTicker{
+		interval: interval,
+		timer:    timer,
+		ch:       make(chan time.Time, 1),
+		now:      timer.Now(),
+	}
+}
+
+func (f *fakeTicker) C() <-chan time.Time {
+	f.now = f.now.Add(f.interval)
+	f.timer.Set(f.now)
+	f.ch <- f.now
+	return f.ch
+}
+
+func (f *fakeTicker) Stop() {}
+
+type FakeTimer struct {
+	now time.Time
+}
+
+func NewFakeTimer() *FakeTimer {
+	return &FakeTimer{
+		now: time.Date(2025, time.June, 1, 0, 0, 0, 0, time.UTC),
+	}
+}
+
+func (f *FakeTimer) Set(t time.Time) {
+	f.now = t
+}
+
+func (f *FakeTimer) Now() time.Time {
+	return f.now
+}

client/internal/lazyconn/inactivity/scenarios_test.go

@@ -0,0 +1,103 @@
+package inactivity
+
+import "time"
+
+type scenario struct {
+	ExpectedInactive bool
+	Data             []rxHistory
+}
+
+var scenarios = []scenario{
+	{
+		ExpectedInactive: true,
+		Data: []rxHistory{
+			{when: 0 * time.Second, RxBytes: 32},
+			{when: 25 * time.Second, RxBytes: 32},
+			{when: 50 * time.Second, RxBytes: 32},
+			{when: 75 * time.Second, RxBytes: 32},
+			{when: 100 * time.Second, RxBytes: 32},
+			{when: 100 * time.Second, RxBytes: 92},
+			{when: 150 * time.Second, RxBytes: 32},
+			{when: 175 * time.Second, RxBytes: 32},
+			{when: 200 * time.Second, RxBytes: 32},
+			{when: 225 * time.Second, RxBytes: 32},
+			{when: 250 * time.Second, RxBytes: 32},
+			{when: 250 * time.Second, RxBytes: 92},
+			{when: 300 * time.Second, RxBytes: 32},
+			{when: 325 * time.Second, RxBytes: 32},
+			{when: 350 * time.Second, RxBytes: 32},
+			{when: 375 * time.Second, RxBytes: 32},
+			{when: 375 * time.Second, RxBytes: 92},
+			{when: 400 * time.Second, RxBytes: 32},
+			{when: 425 * time.Second, RxBytes: 32},
+			{when: 450 * time.Second, RxBytes: 32},
+			{when: 475 * time.Second, RxBytes: 32},
+			{when: 500 * time.Second, RxBytes: 32},
+			{when: 500 * time.Second, RxBytes: 92},
+			{when: 525 * time.Second, RxBytes: 32},
+			{when: 550 * time.Second, RxBytes: 32},
+			{when: 575 * time.Second, RxBytes: 32},
+			{when: 600 * time.Second, RxBytes: 32},
+			{when: 625 * time.Second, RxBytes: 32},
+			{when: 625 * time.Second, RxBytes: 92},
+			{when: 650 * time.Second, RxBytes: 32},
+			{when: 675 * time.Second, RxBytes: 32},
+			{when: 700 * time.Second, RxBytes: 32},
+			{when: 725 * time.Second, RxBytes: 32},
+			{when: 750 * time.Second, RxBytes: 32},
+			{when: 750 * time.Second, RxBytes: 92},
+			{when: 775 * time.Second, RxBytes: 32},
+		},
+	},
+	{
+		ExpectedInactive: true,
+		Data: []rxHistory{
+			{when: 0 * time.Second, RxBytes: 32},
+			{when: 25 * time.Second, RxBytes: 32},
+			{when: 50 * time.Second, RxBytes: 32},
+			{when: 75 * time.Second, RxBytes: 32},
+			{when: 100 * time.Second, RxBytes: 32},
+			{when: 100 * time.Second, RxBytes: 148},
+			{when: 125 * time.Second, RxBytes: 32},
+			{when: 150 * time.Second, RxBytes: 32},
+			{when: 175 * time.Second, RxBytes: 32},
+			{when: 200 * time.Second, RxBytes: 32},
+			{when: 225 * time.Second, RxBytes: 32},
+			{when: 225 * time.Second, RxBytes: 148},
+			{when: 250 * time.Second, RxBytes: 32},
+			{when: 275 * time.Second, RxBytes: 32},
+			{when: 300 * time.Second, RxBytes: 32},
+			{when: 325 * time.Second, RxBytes: 32},
+			{when: 350 * time.Second, RxBytes: 32},
+			{when: 350 * time.Second, RxBytes: 148},
+			{when: 375 * time.Second, RxBytes: 32},
+			{when: 400 * time.Second, RxBytes: 32},
+			{when: 425 * time.Second, RxBytes: 32},
+			{when: 450 * time.Second, RxBytes: 32},
+			{when: 475 * time.Second, RxBytes: 32},
+			{when: 475 * time.Second, RxBytes: 148},
+			{when: 500 * time.Second, RxBytes: 32},
+			{when: 525 * time.Second, RxBytes: 32},
+			{when: 550 * time.Second, RxBytes: 32},
+			{when: 575 * time.Second, RxBytes: 32},
+			{when: 600 * time.Second, RxBytes: 32},
+			{when: 600 * time.Second, RxBytes: 148},
+			{when: 625 * time.Second, RxBytes: 32},
+			{when: 650 * time.Second, RxBytes: 32},
+			{when: 675 * time.Second, RxBytes: 32},
+			{when: 700 * time.Second, RxBytes: 32},
+			{when: 725 * time.Second, RxBytes: 32},
+			{when: 725 * time.Second, RxBytes: 148},
+			{when: 750 * time.Second, RxBytes: 32},
+		},
+	},
+	{
+		ExpectedInactive: false,
+		Data: []rxHistory{
+			{when: 0 * time.Second, RxBytes: 32},
+			{when: 25 * time.Second, RxBytes: 32},
+			{when: 50 * time.Second, RxBytes: 100},
+			{when: 75 * time.Second, RxBytes: 32},
+		},
+	},
+}

client/internal/lazyconn/inactivity/ticker.go

@@ -0,0 +1,24 @@
+package inactivity
+
+import "time"
+
+var newTicker = func(d time.Duration) Ticker {
+	return &realTicker{t: time.NewTicker(d)}
+}
+
+type Ticker interface {
+	C() <-chan time.Time
+	Stop()
+}
+
+type realTicker struct {
+	t *time.Ticker
+}
+
+func (r *realTicker) C() <-chan time.Time {
+	return r.t.C
+}
+
+func (r *realTicker) Stop() {
+	r.t.Stop()
+}