From 33f284682b9ffeeb467e767b6747f87cae24d708 Mon Sep 17 00:00:00 2001 From: Krasi Georgiev Date: Tue, 15 Jan 2019 23:10:08 +0200 Subject: [PATCH] ignore corrupted blocks that have been replaced by parents. Signed-off-by: Krasi Georgiev --- db.go | 28 ++++++++++++---------------- 1 file changed, 12 insertions(+), 16 deletions(-) diff --git a/db.go b/db.go index 45d5e573..59b51b54 100644 --- a/db.go +++ b/db.go @@ -482,14 +482,22 @@ func (db *DB) reload() (err error) { if err != nil { return err } + deletable := db.deletableBlocks(loadable) - // A corrupted block that is not set for deletion by deletableBlocks() should return an error. - for ulid, err := range corrupted { - if _, ok := deletable[ulid]; !ok { - return errors.Wrap(err, "unexpected corrupted block") + // Corrupted blocks that have been replaced by parents can be safely ignored and deleted. + // This makes it resilient against the process crashing towards the end of a compaction. + // Creation of a new block and deletion of its parents cannot happen atomically. + // By creating blocks with their parents, we can pick up the deletion where it left off during a crash. + for _, block := range loadable { + for _, b := range block.Meta().Compaction.Parents { + delete(corrupted, b.ULID) + deletable[b.ULID] = nil } } + if len(corrupted) > 0 { + return errors.Wrap(err, "unexpected corrupted block") + } // All deletable blocks should not be loaded. var ( @@ -577,18 +585,6 @@ func (db *DB) openBlocks() (blocks []*Block, corrupted map[ulid.ULID]error, err func (db *DB) deletableBlocks(blocks []*Block) map[ulid.ULID]*Block { deletable := make(map[ulid.ULID]*Block) - // Delete old blocks that have been superseded by new ones by gathering their parents. - // Those parents all have newer replacements and - // can be safely deleted after loading the other blocks. - // This makes it resilient against the process crashing towards the end of a compaction. - // Creation of a new block and deletion of its parents cannot happen atomically. - // By creating blocks with their parents, we can pick up the deletion where it left off during a crash. - for _, block := range blocks { - for _, b := range block.Meta().Compaction.Parents { - deletable[b.ULID] = nil - } - } - // Sort the blocks by time - newest to oldest (largest to smallest timestamp). // This ensures that the retentions will remove the oldest blocks. sort.Slice(blocks, func(i, j int) bool {