feat: pause background work when app is in background

internal/ipfs/ipfs.go

@@ -19,6 +19,7 @@ import (
 	"github.com/status-im/status-go/common"
 	"github.com/status-im/status-go/internal/logutils"
 	"github.com/status-im/status-go/params"
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
 )
 
 const maxRequestsPerSecond = 3
@@ -100,15 +101,39 @@ func (d *Downloader) taskDispatcher() {
 	defer common.LogOnPanic()
 	ticker := time.NewTicker(time.Second / maxRequestsPerSecond)
 	defer ticker.Stop()
+	sub := messaginglifecycle.SubscribePausedBackground()
+	defer sub.Unsubscribe()
+
+	paused := <-sub.C()
+	var tickerC <-chan time.Time
+	if !paused {
+		tickerC = ticker.C
+	}
 
 	for {
-		<-ticker.C
-		request, ok := <-d.inputTaskChan
-		if !ok {
+		select {
+		case <-d.quit:
 			return
+		case pausedState, ok := <-sub.C():
+			if !ok {
+				return
+			}
+			paused = pausedState
+			if paused {
+				tickerC = nil
+			} else {
+				tickerC = ticker.C
+			}
+		case <-tickerC:
+			select {
+			case request, ok := <-d.inputTaskChan:
+				if !ok {
+					return
+				}
+				d.rateLimiterChan <- request
+			default:
+			}
 		}
-		d.rateLimiterChan <- request
-
 	}
 }
 

internal/ipfs/ipfs_test.go

@@ -0,0 +1,42 @@
+package ipfs
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
+)
+
+func TestTaskDispatcherPausesAndResumesByLifecycle(t *testing.T) {
+	messaginglifecycle.SetPausedBackground(true)
+	defer messaginglifecycle.SetPausedBackground(false)
+
+	d := &Downloader{
+		inputTaskChan:   make(chan taskRequest, 1),
+		rateLimiterChan: make(chan taskRequest, 1),
+		quit:            make(chan struct{}, 1),
+	}
+
+	go d.taskDispatcher()
+	defer close(d.quit)
+
+	req := taskRequest{cid: "cid-1", doneChan: make(chan taskResponse, 1)}
+	d.inputTaskChan <- req
+
+	select {
+	case <-d.rateLimiterChan:
+		t.Fatal("task was dispatched while paused")
+	case <-time.After(450 * time.Millisecond):
+	}
+
+	messaginglifecycle.SetPausedBackground(false)
+
+	select {
+	case got := <-d.rateLimiterChan:
+		require.Equal(t, req.cid, got.cid)
+	case <-time.After(2 * time.Second):
+		t.Fatal("task was not dispatched after resume")
+	}
+}

internal/rpc/chain/rpclimiter/rpc_limiter.go

@@ -12,6 +12,7 @@ import (
 
 	gocommon "github.com/status-im/status-go/common"
 	"github.com/status-im/status-go/internal/logutils"
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
 )
 
 const (
@@ -260,9 +261,29 @@ func (rl *RPCRpsLimiter) start() {
 	ticker := time.NewTicker(tickerInterval)
 	go func() {
 		defer gocommon.LogOnPanic()
+		sub := messaginglifecycle.SubscribePausedBackground()
+		defer sub.Unsubscribe()
+
+		paused := <-sub.C()
+		var tickerC <-chan time.Time
+		if !paused {
+			tickerC = ticker.C
+		}
+
 		for {
 			select {
-			case <-ticker.C:
+			case pausedState, ok := <-sub.C():
+				if !ok {
+					ticker.Stop()
+					return
+				}
+				paused = pausedState
+				if paused {
+					tickerC = nil
+				} else {
+					tickerC = ticker.C
+				}
+			case <-tickerC:
 				{
 					rl.requestsMadeWithinSecondMutex.Lock()
 					oldrequestsMadeWithinSecond := rl.requestsMadeWithinSecond

internal/rpc/chain/rpclimiter/rpc_limiter_test.go

@@ -4,7 +4,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
+
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
 )
 
 func setupTest() (*InMemRequestsMapStorage, RequestLimiter) {
@@ -193,3 +196,36 @@ func TestAllowWhenLimitNotReachedForInfinitePeriod(t *testing.T) {
 	// Verify the result
 	require.True(t, allow)
 }
+
+func TestRPCRpsLimiterStartPausesAndResumesReleases(t *testing.T) {
+	messaginglifecycle.SetPausedBackground(true)
+	defer messaginglifecycle.SetPausedBackground(false)
+
+	waitCh := make(chan bool, 1)
+	rl := &RPCRpsLimiter{
+		uuid:                     uuid.New(),
+		maxRequestsPerSecond:     1,
+		requestsMadeWithinSecond: 1,
+		callersOnWaitForRequests: []callerOnWait{
+			{requests: 1, ch: waitCh},
+		},
+		quit: make(chan bool),
+	}
+
+	rl.start()
+	defer rl.Stop()
+
+	select {
+	case <-waitCh:
+		t.Fatal("limiter released callers while paused")
+	case <-time.After(tickerInterval + 200*time.Millisecond):
+	}
+
+	messaginglifecycle.SetPausedBackground(false)
+
+	select {
+	case <-waitCh:
+	case <-time.After(2 * tickerInterval):
+		t.Fatal("limiter did not release callers after resume")
+	}
+}

internal/rpc/client.go

@@ -1,3 +1,3 @@
 package rpc
 
 //go:generate go tool mockgen -package=mock_rpcclient -source=client.go -destination=mock/client/client.go
@@ -145,6 +145,15 @@
 	c.signalsTransmitter.Stop()
 	c.networkManager.Stop()
 
+	c.rpsLimiterMutex.Lock()
+	for key, limiter := range c.limiterPerProvider {
+		if limiter != nil {
+			limiter.Stop()
+		}
+		delete(c.limiterPerProvider, key)
+	}
+	c.rpsLimiterMutex.Unlock()
+
 	c.rpcClientsMutex.Lock()
 	for _, client := range c.rpcClients {
 		client.Close()

internal/timesource/ntp.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/status-im/status-go/common"
 	"github.com/status-im/status-go/internal/logutils"
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
 )
 
 const (
@@ -221,14 +222,52 @@ func (s *ntpTimeSource) runPeriodically(ctx context.Context, fn func() error, st
 	}
 	go func() {
 		defer common.LogOnPanic()
+		sub := messaginglifecycle.SubscribePausedBackground()
+		defer sub.Unsubscribe()
+
+		paused := <-sub.C()
+		if paused && period != s.slowNTPSyncPeriod {
+			period = s.slowNTPSyncPeriod
+		}
+
+		timer := time.NewTimer(period)
+		defer timer.Stop()
+
+		resetTimer := func(d time.Duration) {
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			timer.Reset(d)
+		}
+
 		for {
 			select {
-			case <-time.After(period):
+			case pausedState, ok := <-sub.C():
+				if !ok {
+					return
+				}
+				paused = pausedState
+				if paused && period != s.slowNTPSyncPeriod {
+					period = s.slowNTPSyncPeriod
+				}
+				resetTimer(period)
+			case <-timer.C:
+				if paused {
+					resetTimer(period)
+					continue
+				}
 				if err := fn(); err == nil {
 					period = s.slowNTPSyncPeriod
 				} else if period != s.slowNTPSyncPeriod {
 					period = s.fastNTPSyncPeriod
 				}
+				if paused && period != s.slowNTPSyncPeriod {
+					period = s.slowNTPSyncPeriod
+				}
+				resetTimer(period)
 
 			case <-ctx.Done():
 				return

internal/timesource/ntp_test.go

@@ -10,6 +10,8 @@ import (
 	"github.com/beevik/ntp"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	messaginglifecycle "github.com/status-im/status-go/pkg/messaging/lifecycle"
 )
 
 const (
@@ -287,6 +289,39 @@ func TestGetCurrentTimeInMillis(t *testing.T) {
 	ts.Stop()
 }
 
+func TestRunPeriodicallyPausesAndResumesByLifecycle(t *testing.T) {
+	messaginglifecycle.SetPausedBackground(true)
+	defer messaginglifecycle.SetPausedBackground(false)
+
+	source := &ntpTimeSource{
+		fastNTPSyncPeriod: 20 * time.Millisecond,
+		slowNTPSyncPeriod: 120 * time.Millisecond,
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	hitsCh := make(chan struct{}, 4)
+	source.runPeriodically(ctx, func() error {
+		hitsCh <- struct{}{}
+		return nil
+	}, false)
+
+	select {
+	case <-hitsCh:
+		t.Fatal("periodic function ran while paused")
+	case <-time.After(90 * time.Millisecond):
+	}
+
+	messaginglifecycle.SetPausedBackground(false)
+
+	select {
+	case <-hitsCh:
+	case <-time.After(600 * time.Millisecond):
+		t.Fatal("periodic function did not run after resume")
+	}
+}
+
 func TestGetCurrentTimeOffline(t *testing.T) {
 	// covers https://github.com/status-im/status-desktop/issues/12691
 	ts := &ntpTimeSource{