Add retention deletion

db.go

@@ -207,10 +207,13 @@ func (db *DB) run() {
 
 			var merr MultiError
 
-			changes, err := db.compact()
+			changes1, err := db.retentionCutoff()
 			merr.Add(err)
 
-			if changes {
+			changes2, err := db.compact()
+			merr.Add(err)
+
+			if changes1 || changes2 {
 				merr.Add(db.reloadBlocks())
 			}
 			if err := merr.Err(); err != nil {
@@ -223,11 +226,31 @@ func (db *DB) run() {
 	}
 }
 
+func (db *DB) retentionCutoff() (bool, error) {
+	db.headmtx.RLock()
+
+	if db.opts.RetentionDuration == 0 {
+		return false, nil
+	}
+	// We don't count the span covered by head blocks towards the
+	// retention time as it generally makes up a fraction of it.
+	if len(db.persisted) == 0 {
+		return false, nil
+	}
+
+	last := db.persisted[len(db.persisted)-1]
+	mint := last.Meta().MaxTime - int64(db.opts.RetentionDuration)
+
+	db.headmtx.RUnlock()
+
+	return retentionCutoff(db.dir, mint)
+}
+
 func (db *DB) compact() (changes bool, err error) {
-	// Check whether we have pending head blocks that are ready to be persisted.
-	// They have the highest priority.
 	db.headmtx.RLock()
 
+	// Check whether we have pending head blocks that are ready to be persisted.
+	// They have the highest priority.
 	var singles []*headBlock
 
 	// Collect head blocks that are ready for compaction. Write them after
@@ -297,36 +320,35 @@ Loop:
 	return changes, nil
 }
 
-// func (db *DB) retentionCutoff() error {
-// 	if db.opts.RetentionDuration == 0 {
-// 		return nil
-// 	}
-// 	h := db.heads[len(db.heads)-1]
-// 	t := h.meta.MinTime - int64(db.opts.RetentionDuration)
-
-// 	var (
-// 		blocks = db.blocks()
-// 		i      int
-// 		b      Block
-// 	)
-// 	for i, b = range blocks {
-// 		if b.Meta().MinTime >= t {
-// 			break
-// 		}
-// 	}
-// 	if i <= 1 {
-// 		return nil
-// 	}
-// 	db.logger.Log("msg", "retention cutoff", "idx", i-1)
-// 	db.removeBlocks(0, i)
-
-// 	for _, b := range blocks[:i] {
-// 		if err := os.RemoveAll(b.Dir()); err != nil {
-// 			return errors.Wrap(err, "removing old block")
-// 		}
-// 	}
-// 	return nil
-// }
+// retentionCutoff deletes all directories of blocks in dir that are strictly
+// before mint.
+func retentionCutoff(dir string, mint int64) (bool, error) {
+	dirs, err := blockDirs(dir)
+	if err != nil {
+		return false, errors.Wrapf(err, "list block dirs %s", dir)
+	}
+
+	changes := false
+
+	for _, dir := range dirs {
+		meta, err := readMetaFile(dir)
+		if err != nil {
+			return changes, errors.Wrapf(err, "read block meta %s", dir)
+		}
+		// The first block we encounter marks that we crossed the boundary
+		// of deletable blocks.
+		if meta.MaxTime >= mint {
+			break
+		}
+		changes = true
+
+		if err := os.RemoveAll(dir); err != nil {
+			return changes, err
+		}
+	}
+
+	return changes, nil
+}
 
 func (db *DB) reloadBlocks() error {
 	var cs []io.Closer