Leave failed user stats writes as they are, to keep this change small

user/manager.go

@@ -57,7 +57,6 @@ type Manager struct {
 	db          *db.DB
 	queries     queries
 	statsQueue  map[string]*Stats       // Accumulated stats deltas to be written to the database (UserID -> Stats)
-	statsResets int                     // Number of daily resets, so a failed write from before one is not retried after it
 	tokenQueue  map[string]*TokenUpdate // "Queue" to asynchronously write token access stats to the database (Token ID -> TokenUpdate)
 	accessCache *accessCache            // In-memory snapshot of user_access; refreshed by maybeReloadAccessCache after every ACL mutation
 	quit        chan struct{}           // Closed by Close() to signal background goroutines to stop
@@ -441,7 +440,6 @@ func (a *Manager) ResetStats() error {
 		return err
 	}
 	a.statsQueue = make(map[string]*Stats)
-	a.statsResets++
 	return nil
 }
 
@@ -452,11 +450,6 @@ func (a *Manager) ResetStats() error {
 func (a *Manager) EnqueueUserStats(userID string, delta *Stats) {
 	a.mu.Lock()
 	defer a.mu.Unlock()
-	a.enqueueUserStatsNoLock(userID, delta)
-}
-
-// enqueueUserStatsNoLock adds the delta to the user's queued stats; a.mu must be held
-func (a *Manager) enqueueUserStatsNoLock(userID string, delta *Stats) {
 	if queued, ok := a.statsQueue[userID]; ok {
 		queued.Messages += delta.Messages
 		queued.Emails += delta.Emails
@@ -491,11 +484,11 @@ func (a *Manager) writeUserStatsQueue() error {
 		log.Tag(tag).Trace("No user stats updates to commit")
 		return nil
 	}
-	statsQueue, resets := a.statsQueue, a.statsResets
+	statsQueue := a.statsQueue
 	a.statsQueue = make(map[string]*Stats)
 	a.mu.Unlock()
 
-	err := db.ExecTx(a.db, func(tx *sql.Tx) error {
+	return db.ExecTx(a.db, func(tx *sql.Tx) error {
 		log.Tag(tag).Debug("Writing user stats queue for %d user(s)", len(statsQueue))
 		for userID, update := range statsQueue {
 			log.
@@ -513,23 +506,6 @@ func (a *Manager) writeUserStatsQueue() error {
 		}
 		return nil
 	})
-	if err != nil {
-		a.requeueUserStats(statsQueue, resets)
-	}
-	return err
-}
-
-// requeueUserStats puts the deltas of a failed write back, to be retried with the next flush: they
-// are increments, so nothing else will write them again. Deltas from before a daily reset are dropped.
-func (a *Manager) requeueUserStats(statsQueue map[string]*Stats, resets int) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if resets != a.statsResets {
-		return
-	}
-	for userID, delta := range statsQueue {
-		a.enqueueUserStatsNoLock(userID, delta)
-	}
 }
 
 // User returns the user with the given username if it exists, or ErrUserNotFound otherwise

user/manager_test.go

@@ -3597,59 +3597,3 @@ func TestManager_EnqueueStats_DeltasAccumulate(t *testing.T) {
 		require.Equal(t, int64(4), u.Stats.Calls)
 	})
 }
-
-func TestManager_EnqueueStats_FailedFlushIsRetried(t *testing.T) {
-	// A failed write keeps its deltas for the next flush; they are increments, so nothing else
-	// would ever write them again
-	a := newFailingStatsManager(t)
-	u, err := a.User("ben")
-	require.Nil(t, err)
-	failStatsWrites(t, a, true)
-	a.EnqueueUserStats(u.ID, &Stats{Messages: 3})
-	require.Error(t, a.writeUserStatsQueue())
-
-	failStatsWrites(t, a, false)
-	a.EnqueueUserStats(u.ID, &Stats{Messages: 1})
-	require.Nil(t, a.writeUserStatsQueue())
-	u, err = a.User("ben")
-	require.Nil(t, err)
-	require.Equal(t, int64(4), u.Stats.Messages)
-}
-
-func TestManager_EnqueueStats_FailedFlushIsDroppedAfterReset(t *testing.T) {
-	// Deltas of a failed write from before the daily reset belong to the old day
-	a := newFailingStatsManager(t)
-	u, err := a.User("ben")
-	require.Nil(t, err)
-	failStatsWrites(t, a, true)
-	a.EnqueueUserStats(u.ID, &Stats{Messages: 3})
-	require.Error(t, a.writeUserStatsQueue())
-
-	failStatsWrites(t, a, false)
-	require.Nil(t, a.ResetStats())
-	require.Nil(t, a.writeUserStatsQueue())
-	u, err = a.User("ben")
-	require.Nil(t, err)
-	require.Equal(t, int64(0), u.Stats.Messages)
-}
-
-func newFailingStatsManager(t *testing.T) *Manager {
-	a, err := NewSQLiteManager(filepath.Join(t.TempDir(), "user.db"), "", &Config{
-		DefaultAccess:       PermissionReadWrite,
-		BcryptCost:          bcrypt.MinCost,
-		QueueWriterInterval: time.Hour, // Flushed manually
-	})
-	require.Nil(t, err)
-	t.Cleanup(func() { a.Close() })
-	require.Nil(t, a.AddUser("ben", "ben", RoleUser, false))
-	return a
-}
-
-func failStatsWrites(t *testing.T, a *Manager, fail bool) {
-	query := `DROP TRIGGER fail_stats`
-	if fail {
-		query = `CREATE TRIGGER fail_stats BEFORE UPDATE OF stats_messages ON user BEGIN SELECT RAISE(FAIL, 'write failed'); END`
-	}
-	_, err := a.db.Exec(query)
-	require.Nil(t, err)
-}