accounting: add RemoveDoneTransfers method to fix bisync race #8815

Before this change bisync was adjusting MaxCompletedTransfers in order
to clear the done transfers from the stats.

This wasn't working (because it was only clearing one transfer) and
was part of a race adjusting MaxCompletedTransfers.

This fixes the problem by introducing a new method RemoveDoneTransfers
to clear the done transfers explicitly and calling it in bisync.

cmd/bisync/listing.go

@@ -707,8 +707,7 @@ func (b *bisyncRun) modifyListing(ctx context.Context, src fs.Fs, dst fs.Fs, res
 		prettyprint(dstList.list, "dstList", fs.LogLevelDebug)
 
 		// clear stats so we only do this once
-		accounting.MaxCompletedTransfers = 0
-		accounting.Stats(ctx).PruneTransfers()
+		accounting.Stats(ctx).RemoveDoneTransfers()
 	}
 
 	if b.DebugName != "" {

fs/accounting/stats.go

@@ -912,22 +912,31 @@ func (s *StatsInfo) RemoveTransfer(transfer *Transfer) {
 }
 
 // PruneTransfers makes sure there aren't too many old transfers by removing
-// single finished transfer.
-func (s *StatsInfo) PruneTransfers() {
+// a single finished transfer. Returns true if it removed a transfer.
+func (s *StatsInfo) PruneTransfers() bool {
 	if MaxCompletedTransfers < 0 {
-		return
+		return false
 	}
+	removed := false
 	s.mu.Lock()
 	// remove a transfer from the start if we are over quota
 	if len(s.startedTransfers) > MaxCompletedTransfers+s.ci.Transfers {
 		for i, tr := range s.startedTransfers {
 			if tr.IsDone() {
 				s._removeTransfer(tr, i)
+				removed = true
 				break
 			}
 		}
 	}
 	s.mu.Unlock()
+	return removed
+}
+
+// RemoveDoneTransfers removes all Done transfers.
+func (s *StatsInfo) RemoveDoneTransfers() {
+	for s.PruneTransfers() {
+	}
 }
 
 // AddServerSideMove counts a server side move

fs/accounting/stats_test.go

@@ -465,3 +465,27 @@ func TestPruneTransfers(t *testing.T) {
 		})
 	}
 }
+
+func TestRemoveDoneTransfers(t *testing.T) {
+	ctx := context.Background()
+	s := NewStats(ctx)
+	const transfers = 10
+	for i := int64(1); i <= int64(transfers); i++ {
+		s.AddTransfer(&Transfer{
+			startedAt:   time.Unix(i, 0),
+			completedAt: time.Unix(i+1, 0),
+		})
+	}
+
+	s.mu.Lock()
+	assert.Equal(t, time.Duration(transfers)*time.Second, s._totalDuration())
+	assert.Equal(t, transfers, len(s.startedTransfers))
+	s.mu.Unlock()
+
+	s.RemoveDoneTransfers()
+
+	s.mu.Lock()
+	assert.Equal(t, time.Duration(transfers)*time.Second, s._totalDuration())
+	assert.Equal(t, transfers, len(s.startedTransfers))
+	s.mu.Unlock()
+}