periodic.go

// Collect rev 10 times frequently than the actual interval
// so that we can retry in error case with a shorter delay
const intervalCoefficient = 10

func (t *Periodic) Run() {
	t.ctx, t.cancel = context.WithCancel(context.Background())
	interval := t.getInterval()
	subInterval := interval / intervalCoefficient
	// expected amount of revs after t.period
	expectedNum := int(math.Ceil(float64(t.period) / float64(subInterval)))

	go func() {
		for {
			t.revs = append(t.revs, t.rg.Rev())

			select {
			case <-t.ctx.Done():
				return
			case <-t.clock.After(subInterval):
			}

			t.mu.Lock()
			p := t.paused
			t.mu.Unlock()
			if p {
				continue
			}

			if len(t.revs) < expectedNum {
				continue
			}
			indexPeriodAgo := len(t.revs) - expectedNum
			rev := t.revs[indexPeriodAgo]

			plog.Noticef("Starting auto-compaction at revision %d (retention: %v)", rev, t.period)
			_, err := t.c.Compact(t.ctx, &pb.CompactionRequest{Revision: rev})
			if err == nil || err == mvcc.ErrCompacted {
				plog.Noticef("Finished auto-compaction at revision %d", rev)
				// move to next sliding window
				t.revs = t.revs[indexPeriodAgo+1:]
			} else {
				plog.Noticef("Failed auto-compaction at revision %d (%v)", rev, err)
				plog.Noticef("Retry after %v", subInterval)
			}
		}
	}()
}