tapdb: ensure the stats cache is always populated

In this commit, we ensure the stats cache is always populated by using a
background goroutine on a timer to hot swap the fresh values in the
cache based on the interval.

Author: Roasbeef

tapdb/universe_stats.go

@@ -449,29 +449,10 @@ func (u *UniverseStats) LogNewProofEvents(ctx context.Context,
 	})
 }
 
-// AggregateSyncStats returns stats aggregated over all assets within the
-// Universe.
-func (u *UniverseStats) AggregateSyncStats(
-	ctx context.Context) (universe.AggregateStats, error) {
-
-	stats := u.statsSnapshot.Load()
-	if stats != nil {
-		u.statsCacheLogger.Hit()
-		return *stats, nil
-	}
-
-	u.statsMtx.Lock()
-	defer u.statsMtx.Unlock()
-
-	// Check to see if the stats were loaded in while we were waiting for
-	// the mutex.
-	stats = u.statsSnapshot.Load()
-	if stats != nil {
-		u.statsCacheLogger.Hit()
-		return *stats, nil
-	}
-
-	u.statsCacheLogger.Miss()
+// querySyncStats is a helper function that's used to query the sync stats for
+// the Universe db.
+func (u *UniverseStats) querySyncStats(ctx context.Context,
+) (universe.AggregateStats, error) {
 
 	var dbStats universe.AggregateStats
 
@@ -516,16 +497,83 @@ func (u *UniverseStats) AggregateSyncStats(
 		return universe.AggregateStats{}, err
 	}
 
+	return dbStats, nil
+}
+
+// populateSyncStatsCache is used to populate the sync stats cache
+// periodically.
+//
+// NOTE: This MUST be run as the call back of a time.AfterFunc.
+func (u *UniverseStats) populateSyncStatsCache() {
+	log.Infof("Refreshing stats cache, duration=%v", u.opts.cacheDuration)
+
+	// If this is a test, then we'll just purge the items.
+	if u.opts.cacheDuration == 0 {
+		u.statsSnapshot.Store(nil)
+		return
+	}
+
+	now := time.Now()
+
+	// To ensure the stats endpoint is always available, we'll repopulate
+	// it async ourselves here. This ensures after the first miss, the
+	// stats are always populated.
+	ctx := context.Background()
+	dbStats, err := u.querySyncStats(ctx)
+	if err != nil {
+		log.Warnf("Unable to refresh stats cache: %v", err)
+		return
+	}
+
+	log.Debugf("Refreshed stats cache, interval=%v, took=%v",
+		u.opts.cacheDuration, time.Since(now))
+
+	u.statsSnapshot.Store(&dbStats)
+
+	// Reset the timer so we'll refresh again after the cache duration.
+	if !u.statsRefresh.Stop() {
+		<-u.statsRefresh.C
+	}
+
+	u.statsRefresh.Reset(u.opts.cacheDuration)
+}
+
+// AggregateSyncStats returns stats aggregated over all assets within the
+// Universe.
+func (u *UniverseStats) AggregateSyncStats(
+	ctx context.Context) (universe.AggregateStats, error) {
+
+	stats := u.statsSnapshot.Load()
+	if stats != nil {
+		u.statsCacheLogger.Hit()
+		return *stats, nil
+	}
+
+	u.statsMtx.Lock()
+	defer u.statsMtx.Unlock()
+
+	// Check to see if the stats were loaded in while we were waiting for
+	// the mutex.
+	stats = u.statsSnapshot.Load()
+	if stats != nil {
+		u.statsCacheLogger.Hit()
+		return *stats, nil
+	}
+
+	u.statsCacheLogger.Miss()
+
+	dbStats, err := u.querySyncStats(ctx)
+	if err != nil {
+		return dbStats, err
+	}
+
 	// We'll store the DB stats then start our time after function to wipe
 	// the stats pointer so we'll refresh it after a period of time.
 	u.statsSnapshot.Store(&dbStats)
 
-	u.statsRefresh = time.AfterFunc(u.opts.cacheDuration, func() {
-		log.Infof("Purging stats cache, duration=%v",
-			u.opts.cacheDuration)
-
-		u.statsSnapshot.Store(nil)
-	})
+	u.statsRefresh = time.AfterFunc(
+		u.opts.cacheDuration, u.populateSyncStatsCache,
+	)
 
 	return dbStats, nil
 }

tapdb/universe_stats_test.go

@@ -4,16 +4,19 @@ import (
 	"bytes"
 	"context"
 	"database/sql"
+	"fmt"
 	"math/rand"
 	"sort"
 	"testing"
 	"time"
 
+	"github.com/davecgh/go-spew/spew"
 	"github.com/lightninglabs/taproot-assets/asset"
 	"github.com/lightninglabs/taproot-assets/fn"
 	"github.com/lightninglabs/taproot-assets/tapdb/sqlc"
 	"github.com/lightninglabs/taproot-assets/universe"
 	"github.com/lightningnetwork/lnd/clock"
+	"github.com/lightningnetwork/lnd/lntest/wait"
 	"github.com/stretchr/testify/require"
 )
 
@@ -27,7 +30,7 @@ func newUniverseStatsWithDB(db *BaseDB, clock clock.Clock) (*UniverseStats,
 	)
 
 	stats := NewUniverseStats(
-		dbTxer, clock, WithStatsCacheDuration(time.Microsecond),
+		dbTxer, clock, WithStatsCacheDuration(0),
 	)
 
 	return stats, db
@@ -100,10 +103,26 @@ func (u *uniStatsHarness) logSyncEventByIndex(i int) {
 func (u *uniStatsHarness) assertUniverseStatsEqual(t *testing.T,
 	stats universe.AggregateStats) {
 
-	uniStats, err := u.db.AggregateSyncStats(context.Background())
-	require.NoError(t, err)
+	var (
+		uniStats universe.AggregateStats
+		err      error
+	)
+
+	err = wait.NoError(func() error {
+		uniStats, err = u.db.AggregateSyncStats(context.Background())
+		if err != nil {
+			return err
+		}
 
-	require.Equal(t, uniStats, stats)
+		if uniStats != stats {
+			return fmt.Errorf("expected %v, got %v",
+				spew.Sdump(stats),
+				spew.Sdump(uniStats))
+		}
+
+		return nil
+	}, time.Second*2)
+	require.NoError(t, err)
 }
 
 // TestUniverseStatsEvents tests that we're able to properly insert, and also