diff --git a/src/blockstore/blockstore_heap.cpp b/src/blockstore/blockstore_heap.cpp index 08158679..de17c3ca 100644 --- a/src/blockstore/blockstore_heap.cpp +++ b/src/blockstore/blockstore_heap.cpp @@ -307,7 +307,10 @@ int blockstore_heap_t::read_blocks(uint64_t disk_offset, uint64_t disk_size, uin wr->entry_type &= ~BS_HEAP_GARBAGE; if ((wr->entry_type & BS_HEAP_TYPE) < BS_HEAP_BIG_WRITE || (wr->entry_type & BS_HEAP_TYPE) > BS_HEAP_ROLLBACK || - (wr->entry_type & ~(BS_HEAP_TYPE|BS_HEAP_STABLE))) + (wr->entry_type & ~(BS_HEAP_TYPE|BS_HEAP_STABLE)) || + (wr->entry_type == BS_HEAP_DELETE) || + (wr->entry_type == (BS_HEAP_ROLLBACK|BS_HEAP_STABLE)) || + (wr->entry_type == (BS_HEAP_COMMIT|BS_HEAP_STABLE))) { fprintf(stderr, "Error: entry has unknown type %u in metadata block %u at %u. ", wr->entry_type, block_num, block_offset); @@ -324,6 +327,12 @@ corrupted_object: return EDOM; } } + if (wr->entry_type == BS_HEAP_COMMIT && !wr->version) + { + fprintf(stderr, "Error: commit entry has zero version in metadata block %u at %u. ", + block_num, block_offset); + goto corrupted_object; + } if (wr->size != wr->get_size(this)) { fprintf(stderr, "Error: entry %jx:%jx v%ju has invalid size in metadata block %u at %u (%u != expected %u bytes). Metadata is corrupted, aborting\n", @@ -398,9 +407,91 @@ int blockstore_heap_t::load_blocks(uint64_t disk_offset, uint64_t size, uint8_t }); } +// Validate object entry sequence +bool blockstore_heap_t::validate_object(heap_entry_t *obj) +{ + heap_entry_t *small_wr = NULL; + heap_entry_t *commit_wr = NULL, *rollback_wr = NULL; + heap_entry_t *stable_wr = NULL; + heap_entry_t *next_wr = NULL; + for (auto wr = obj; wr && !wr->is_garbage(); wr = prev(wr)) + { + if (next_wr && wr->lsn == next_wr->lsn && (wr->is_overwrite() == next_wr->is_overwrite())) + { + // Check duplicate lsns + fprintf(stderr, "Error: there are two entries for %jx:%jx with lsn %ju\n", wr->inode, wr->stripe, wr->lsn); + return false; + } + next_wr = wr; + if (wr->type() == BS_HEAP_ROLLBACK) + { + if (commit_wr && wr->version > commit_wr->version) + { + // rollback may not come before commit with a smaller version + fprintf(stderr, "Error: rollback entry %jx:%jx v%ju l%ju comes before a commit entry v%ju l%ju\n", + wr->inode, wr->stripe, wr->version, wr->lsn, commit_wr->version, commit_wr->lsn); + return false; + } + rollback_wr = wr; + continue; + } + if (wr->type() == BS_HEAP_COMMIT) + { + commit_wr = wr; + continue; + } + if (wr->entry_type & BS_HEAP_STABLE) + { + stable_wr = wr; + } + else if (rollback_wr && wr->version > rollback_wr->version) + { + // neither stable nor unstable but ignored + } + else if (commit_wr && wr->version <= commit_wr->version) + { + stable_wr = wr; + } + else + { + if (stable_wr) + { + // a stable write may not come over unstable + fprintf(stderr, "Error: uncommitted entry %jx:%jx v%ju l%ju comes before a committed entry v%ju l%ju\n", + wr->inode, wr->stripe, wr->version, wr->lsn, stable_wr->version, stable_wr->lsn); + return false; + } + } + if (wr->type() == BS_HEAP_SMALL_WRITE || wr->type() == BS_HEAP_INTENT_WRITE) + { + small_wr = wr; + } + else if (wr->type() == BS_HEAP_BIG_WRITE || wr->type() == BS_HEAP_BIG_INTENT) + { + small_wr = NULL; + } + else if (wr->type() == BS_HEAP_DELETE) + { + if (small_wr) + { + // small_write may not come over delete + fprintf(stderr, "Error: entry %jx:%jx v%ju l%ju comes over a DELETE but a BIG_WRITE or BIG_INTENT is expected\n", + small_wr->inode, small_wr->stripe, small_wr->version, small_wr->lsn); + return false; + } + } + } + if (small_wr) + { + fprintf(stderr, "Error: entry %jx:%jx v%ju l%ju comes first but a BIG_WRITE or BIG_INTENT is expected before it\n", + small_wr->inode, small_wr->stripe, small_wr->version, small_wr->lsn); + return false; + } + return true; +} + void blockstore_heap_t::fill_recheck_queue() { - // FIXME: Validate objects for (auto & pgp: block_index) { for (auto & ip: pgp.second) @@ -408,6 +499,7 @@ void blockstore_heap_t::fill_recheck_queue() for (auto & op: ip.second) { auto obj = &op.second.ptr->entry; + // Add object to recheck queue if (obj->type() == BS_HEAP_INTENT_WRITE || obj->type() == BS_HEAP_BIG_INTENT) { // Recheck only the latest intent_write @@ -419,7 +511,7 @@ void blockstore_heap_t::fill_recheck_queue() } else { - // Or a series of small_writes + // Or recheck a series of small_writes for (auto wr = obj; wr && wr->type() == BS_HEAP_SMALL_WRITE; wr = prev(wr)) { if (wr->small().len > 0) @@ -433,7 +525,7 @@ void blockstore_heap_t::fill_recheck_queue() } } -void blockstore_heap_t::mark_used_blocks() +int blockstore_heap_t::mark_used_blocks() { for (auto & pgp: block_index) { @@ -444,6 +536,10 @@ void blockstore_heap_t::mark_used_blocks() bool added = false; auto li = op.second.ptr; auto wr = &li->entry; + if (!validate_object(wr)) + { + return EDOM; + } if (wr->entry_type == (BS_HEAP_DELETE|BS_HEAP_STABLE) && !li->prev) { mark_garbage(li->block_num, wr, UINT32_MAX); @@ -477,6 +573,7 @@ void blockstore_heap_t::mark_used_blocks() } } } + return 0; } void blockstore_heap_t::recheck_buffer(heap_entry_t *cwr, uint8_t *buf) @@ -632,12 +729,16 @@ bool blockstore_heap_t::recheck_small_writes(std::functionlsn < bo->lsn; }); + return 0; } bool blockstore_heap_t::calc_checksums(heap_entry_t *wr, uint8_t *data, bool set, uint32_t offset, uint32_t len) diff --git a/src/blockstore/blockstore_heap.h b/src/blockstore/blockstore_heap.h index f171e5eb..3bb56de1 100644 --- a/src/blockstore/blockstore_heap.h +++ b/src/blockstore/blockstore_heap.h @@ -191,8 +191,9 @@ class blockstore_heap_t int recheck_queue_depth = 0; uint64_t get_pg_id(inode_t inode, uint64_t stripe); + bool validate_object(heap_entry_t *obj); void fill_recheck_queue(); - void mark_used_blocks(); + int mark_used_blocks(); void recheck_buffer(heap_entry_t *cwr, uint8_t *buf); void defragment_block(uint32_t block_num); @@ -219,7 +220,7 @@ public: int load_blocks(uint64_t disk_offset, uint64_t size, uint8_t *buf, bool allow_corrupted, uint64_t &entries_loaded); // finish loading - void finish_load(); + int finish_load(bool allow_corrupted = false); // recheck small write data after reading the database from disk bool recheck_small_writes(std::function)> read_buffer, int queue_depth); // reshard database according to the pool's PG count diff --git a/src/blockstore/blockstore_init.cpp b/src/blockstore/blockstore_init.cpp index 9e6169b9..554b0ecb 100644 --- a/src/blockstore/blockstore_init.cpp +++ b/src/blockstore/blockstore_init.cpp @@ -283,7 +283,10 @@ resume_6: }, bs->meta_write_recheck_parallelism); return 1; resume_7: - bs->heap->finish_load(); + if (bs->heap->finish_load() != 0) + { + exit(1); + } free(metadata_buffer); metadata_buffer = NULL; return 0;