feat(mailserver): stop reconciling history if we were online

pkg/messaging/api.go

@@ -55,10 +55,16 @@ func (a *API) ConnectionStatus() types.ConnectionStatus {
 // be reconciled with the store nodes (#7568): periodically while connectivity
 // is not reliable (relay mesh not Connected on every default shard), and once
 // more when it recovers.
-func (a *API) OnHistoryReconcileNeeded() <-chan struct{} {
+func (a *API) OnHistoryReconcileNeeded() <-chan types.HistoryReconcileWindow {
 	return a.core.stack.Transport.OnHistoryReconcileNeeded()
 }
 
+// HistoryDeliveryReliable reports whether live delivery can safely advance
+// initialized history cursors without issuing a store query.
+func (a *API) HistoryDeliveryReliable() bool {
+	return a.core.stack.Transport.HistoryDeliveryReliable()
+}
+
 // SubscribeFilterMatched returns a channel that is notified whenever an incoming
 // envelope matches at least one installed filter. bufSize should be 1.
 // Callers must call UnsubscribeFilterMatched when done.

pkg/messaging/layers/transport/transport.go

@@ -778,13 +778,17 @@ func (t *Transport) ConnectionState() types.ConnectionState {
 // (#7568): fired periodically while connectivity is not reliable and once when
 // it recovers. Without a waku node (offline transport) it returns a nil
 // channel, which blocks forever — i.e. never signals.
-func (t *Transport) OnHistoryReconcileNeeded() <-chan struct{} {
+func (t *Transport) OnHistoryReconcileNeeded() <-chan types.HistoryReconcileWindow {
 	if t.waku == nil {
 		return nil
 	}
 	return t.waku.OnHistoryReconcileNeeded()
 }
 
+func (t *Transport) HistoryDeliveryReliable() bool {
+	return t.waku != nil && t.waku.HistoryDeliveryReliable()
+}
+
 func (t *Transport) Peers() types.PeerStats {
 	return t.waku.Peers()
 }

pkg/messaging/types/connection_status.go

@@ -8,6 +8,7 @@ import (
 // PartiallyConnected / Connected) surfaced to the Messaging API, re-exported
 // from the waku layer that produces it.
 type ConnectionState = wakutypes.ConnectionState
+type HistoryReconcileWindow = wakutypes.HistoryReconcileWindow
 
 const (
 	ConnectionStateDisconnected       = wakutypes.ConnectionStateDisconnected

pkg/messaging/waku/gowaku.go

@@ -175,12 +175,10 @@ type Waku struct {
 	// so it needs no lock of its own.
 	topicHealth   map[string]peermanager.TopicHealth
 	onlineChecker *onlinechecker.DefaultOnlineChecker
-	// historyReconcileNeeded is signalled by the history-reconcile loop (see
-	// history_reconcile.go) whenever the consumer should fetch history from the
-	// store nodes. Buffered (1) level-trigger: sends never block, a pending
-	// signal coalesces with later ones. Temporary until logos-delivery owns
+	// historyReconcileNeeded carries unreliable delivery windows detected by
+	// the history-reconcile loop. Temporary until logos-delivery owns
 	// reconciliation end to end — see OnHistoryReconcileNeeded.
-	historyReconcileNeeded chan struct{}
+	historyReconcileNeeded chan types.HistoryReconcileWindow
 	// stateMu guards state and stateInitialized. ConnectionChanged is invoked
 	// from the OS/mobile path while checkForConnectionChanges and
 	// handleNetworkChangeFromApp run on the internal poller goroutine, so all
@@ -256,7 +254,7 @@ func New(nodeKey *ecdsa.PrivateKey, cfg *Config, logger *zap.Logger, ts timesour
 		connectionNotifChan:         make(chan node.PeerConnection, 20),
 		connStatusSubscriptions:     make(map[string]*types.ConnStatusSubscription),
 		topicHealth:                 make(map[string]peermanager.TopicHealth),
-		historyReconcileNeeded:      make(chan struct{}, 1),
+		historyReconcileNeeded:      make(chan types.HistoryReconcileWindow, 16),
 		ctx:                         ctx,
 		cancel:                      cancel,
 		wg:                          sync.WaitGroup{},

pkg/messaging/waku/history_reconcile.go

@@ -18,64 +18,102 @@ const (
 	historyReconcileMinInterval   = 30 * time.Second
 )
 
-// OnHistoryReconcileNeeded returns a channel signalled whenever history should
-// be reconciled with the store nodes: periodically while connectivity is not
+type historyReconcileTracker struct {
+	reliable       bool
+	checkedAt      time.Time
+	unreliableFrom time.Time
+	lastReconcile  time.Time
+}
+
+func newHistoryReconcileTracker(reliable bool, now time.Time) historyReconcileTracker {
+	tracker := historyReconcileTracker{reliable: reliable, checkedAt: now}
+	if !reliable {
+		tracker.unreliableFrom = now
+	}
+	return tracker
+}
+
+func (t *historyReconcileTracker) observe(reliable bool, now time.Time, minInterval time.Duration) *types.HistoryReconcileWindow {
+	wasReliable := t.reliable
+	t.reliable = reliable
+	if wasReliable && !reliable {
+		t.unreliableFrom = t.checkedAt
+	}
+	t.checkedAt = now
+
+	if !shouldReconcileHistory(reliable, wasReliable, t.lastReconcile, now, minInterval) {
+		return nil
+	}
+	if t.unreliableFrom.IsZero() {
+		t.unreliableFrom = now
+	}
+	t.lastReconcile = now
+	window := types.HistoryReconcileWindow{From: t.unreliableFrom, To: now}
+	if reliable {
+		t.unreliableFrom = time.Time{}
+	}
+	return &window
+}
+
+// OnHistoryReconcileNeeded returns unreliable delivery windows that should be
+// reconciled with the store nodes: periodically while connectivity is not
 // reliable, and once more when it recovers (closing the unreliable window).
-// It is a buffered level-trigger; consumers that are slow or paused coalesce
-// pending signals instead of queueing them.
 //
 // Temporary: this channel exists because the Waku node cannot yet own the
 // fetch itself. It already has the subscription (filter) list, but lacks the
-// per-topic "when was it last fetched" watermark, which the Messenger persists
-// in the app DB (mailserver_topics). Eventually logos-delivery will own
-// reconciliation and history backfill entirely, including persisting
-// lastFetched per topic: its Messaging API already reconciles
+// per-topic "known complete through" cursor, which the Messenger persists in
+// the app DB (mailserver_topics). Eventually logos-delivery will own
+// reconciliation and history backfill entirely, including persisting that
+// cursor per topic: its Messaging API already reconciles
 // (https://github.com/logos-messaging/logos-delivery/issues/3941) but does not
-// yet fetch history, and neither exposes nor persists lastFetched. That gap
+// yet fetch history, and neither exposes nor persists a completeness cursor.
+// That gap
 // should be closed in logos-delivery before integration — otherwise ownership
-// is split and persistence breaks, since we cannot persist lastFetched on its
+// is split and persistence breaks, since we cannot persist the cursor on its
 // behalf. Until then the fetch stays in the Messenger (Transport would be a
 // nicer interim home, but it is a stopgap either way), and this signal bridges
 // the two.
-func (w *Waku) OnHistoryReconcileNeeded() <-chan struct{} {
+func (w *Waku) OnHistoryReconcileNeeded() <-chan types.HistoryReconcileWindow {
 	return w.historyReconcileNeeded
 }
 
 // startHistoryReconcileLoop runs the timing/decision half of history
 // reconciliation. The fetch itself lives with the consumer (the protocol
-// layer), which owns the topics and per-chat watermarks; this loop only owns
-// connectivity confidence and cadence. Suspended while the node is paused
-// (app backgrounded, no ticker armed): SetPaused(false) triggers its own
-// fetch on foreground.
+// layer), which owns the topics and per-chat history cursors; this loop only owns
+// connectivity confidence and cadence. It continues observing while paused;
+// the consumer queues the resulting windows until it resumes.
 func (w *Waku) startHistoryReconcileLoop() {
 	w.wg.Add(1)
 	go func() {
 		defer gocommon.LogOnPanic()
 		defer w.wg.Done()
 
-		sub := w.PauseBroadcaster.Subscribe()
-		defer sub.Unsubscribe()
-
-		reliable := w.reliablyConnected()
-		var lastReconcile time.Time
+		tracker := newHistoryReconcileTracker(w.reliablyConnected(), time.Now())
+		ticker := time.NewTicker(historyReconcileCheckInterval)
+		defer ticker.Stop()
 
-		pt := gocommon.NewPausableTicker(gocommon.PausableTickerConfig{
-			Interval: historyReconcileCheckInterval,
-			OnTick: func() {
-				wasReliable := reliable
-				reliable = w.reliablyConnected()
-				if !shouldReconcileHistory(reliable, wasReliable, lastReconcile, time.Now(), historyReconcileMinInterval) {
-					return
+		for {
+			select {
+			case <-w.ctx.Done():
+				return
+			case <-ticker.C:
+				now := time.Now()
+				reliable := w.reliablyConnected()
+				window := tracker.observe(reliable, now, historyReconcileMinInterval)
+				if window == nil {
+					continue
 				}
-				lastReconcile = time.Now()
-				w.logger.Debug("history reconciliation needed", zap.Bool("reliable", reliable))
+				w.logger.Debug("history reconciliation needed",
+					zap.Bool("reliable", reliable),
+					zap.Time("from", window.From),
+					zap.Time("to", window.To))
 				select {
-				case w.historyReconcileNeeded <- struct{}{}:
-				default: // a signal is already pending; coalesce
+				case w.historyReconcileNeeded <- *window:
+				case <-w.ctx.Done():
+					return
 				}
-			},
-		}, sub.C())
-		pt.Run(w.ctx.Done())
+			}
+		}
 	}()
 }
 
@@ -89,6 +127,13 @@ func (w *Waku) reliablyConnected() bool {
 	return w.ConnectionState() == types.ConnectionStateConnected
 }
 
+// HistoryDeliveryReliable is stricter than reliablyConnected for light nodes:
+// their filter-subscription health is not observable, so Connected alone must
+// not advance persisted history completeness cursors.
+func (w *Waku) HistoryDeliveryReliable() bool {
+	return !w.cfg.IsLightClient() && w.reliablyConnected()
+}
+
 // shouldReconcileHistory decides whether a reconciliation is due at a tick:
 // when the connection just recovered (an unreliable window closed, fetch what
 // it may have missed), or while it stays unreliable and minInterval has passed

pkg/messaging/waku/history_reconcile_test.go

@@ -5,6 +5,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/status-im/status-go/pkg/messaging/waku/types"
 )
 
 func TestShouldReconcileHistory(t *testing.T) {
@@ -66,3 +68,67 @@ func TestShouldReconcileHistory(t *testing.T) {
 		})
 	}
 }
+
+func TestHistoryReconcileTrackerBoundsUnreliableWindow(t *testing.T) {
+	start := time.Unix(1_000, 0)
+	tracker := newHistoryReconcileTracker(true, start)
+
+	window := tracker.observe(true, start.Add(10*time.Second), historyReconcileMinInterval)
+	require.Nil(t, window)
+
+	// The transition occurred between observations, so use the previous known
+	// reliable time as the conservative lower boundary.
+	window = tracker.observe(false, start.Add(20*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, window)
+	require.Equal(t, start.Add(10*time.Second), window.From)
+	require.Equal(t, start.Add(20*time.Second), window.To)
+
+	window = tracker.observe(false, start.Add(50*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, window)
+	require.Equal(t, start.Add(10*time.Second), window.From)
+	require.Equal(t, start.Add(50*time.Second), window.To)
+
+	window = tracker.observe(true, start.Add(60*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, window)
+	require.Equal(t, start.Add(10*time.Second), window.From)
+	require.Equal(t, start.Add(60*time.Second), window.To)
+
+	window = tracker.observe(true, start.Add(90*time.Second), historyReconcileMinInterval)
+	require.Nil(t, window)
+}
+
+func TestHistoryReconcileTrackerPreservesDisjointWindows(t *testing.T) {
+	start := time.Unix(2_000, 0)
+	tracker := newHistoryReconcileTracker(true, start)
+
+	first := tracker.observe(false, start.Add(10*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, first)
+	window := tracker.observe(true, start.Add(20*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, window)
+
+	window = tracker.observe(true, start.Add(50*time.Second), historyReconcileMinInterval)
+	require.Nil(t, window)
+	second := tracker.observe(false, start.Add(60*time.Second), historyReconcileMinInterval)
+	require.NotNil(t, second)
+
+	require.Equal(t, start, first.From)
+	require.Equal(t, start.Add(50*time.Second), second.From)
+	require.True(t, first.To.Before(second.From))
+}
+
+func TestHistoryDeliveryReliableExcludesLightNodes(t *testing.T) {
+	core := &Waku{
+		cfg:       &Config{Mode: ModeCore},
+		connState: types.ConnectionStateConnected,
+	}
+	require.True(t, core.HistoryDeliveryReliable())
+
+	core.connState = types.ConnectionStatePartiallyConnected
+	require.False(t, core.HistoryDeliveryReliable())
+
+	edge := &Waku{
+		cfg:       &Config{Mode: ModeEdge},
+		connState: types.ConnectionStateConnected,
+	}
+	require.False(t, edge.HistoryDeliveryReliable())
+}

pkg/messaging/waku/types/connection.go

@@ -1,5 +1,7 @@
 package types
 
+import "time"
+
 // ConnectionState mirrors the logos-delivery Messaging API's three-state
 // connection status (see logos-delivery waku/api/types.nim `ConnectionStatus`).
 // It lets status-go speak the same vocabulary as the Messaging API:
@@ -35,3 +37,11 @@ func (s ConnectionState) String() string {
 		return "disconnected"
 	}
 }
+
+// HistoryReconcileWindow is a period during which live delivery could not be
+// trusted. Store reconciliation must not query beyond these bounds (apart from
+// the caller's small out-of-order tolerance).
+type HistoryReconcileWindow struct {
+	From time.Time
+	To   time.Time
+}

pkg/messaging/waku/types/waku.go

@@ -111,12 +111,14 @@ type Waku interface {
 	// from it via ConnectionState.IsOnline().
 	ConnectionState() ConnectionState
 
-	// OnHistoryReconcileNeeded returns a channel signalled whenever history
-	// should be reconciled with the store nodes (#7568): periodically while
-	// connectivity is not reliable (relay mesh not Connected on every default
-	// shard), and once more when connectivity recovers. Buffered
-	// level-trigger; pending signals coalesce.
-	OnHistoryReconcileNeeded() <-chan struct{}
+	// OnHistoryReconcileNeeded returns the unreliable delivery windows that
+	// should be reconciled with store nodes (#7568).
+	OnHistoryReconcileNeeded() <-chan HistoryReconcileWindow
+
+	// HistoryDeliveryReliable reports whether a full-node relay mesh is healthy
+	// enough that active topics can advance their history completeness cursor
+	// without querying a store node. It is deliberately false for light nodes.
+	HistoryDeliveryReliable() bool
 
 	SubscribeEnvelopeEvents(events chan<- EnvelopeEvent) Subscription