Extend BlockStore2 and write a LowToHighLevelBlockStore as wrapper from BlockStore to BlockStore2

src/blockstore/CMakeLists.txt

@@ -30,6 +30,8 @@ set(SOURCES
   implementations/caching/cache/QueueMap.cpp
   implementations/caching/CachedBlock.cpp
   implementations/caching/NewBlock.cpp
+  implementations/low2highlevel/LowToHighLevelBlock.cpp
+  implementations/low2highlevel/LowToHighLevelBlockStore.cpp
   implementations/versioncounting/VersionCountingBlock.cpp
   implementations/versioncounting/VersionCountingBlockStore.cpp
   implementations/versioncounting/VersionCountingBlockStore2.cpp

src/blockstore/implementations/encrypted/EncryptedBlockStore2.h

@@ -45,11 +45,33 @@ public:
     return _baseBlockStore->store(key, encrypted);
   }
 
+  uint64_t numBlocks() const override {
+    return _baseBlockStore->numBlocks();
+  }
+
+  uint64_t estimateNumFreeBytes() const override {
+    return _baseBlockStore->estimateNumFreeBytes();
+  }
+
+  uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const override {
+    uint64_t baseBlockSize = _baseBlockStore->blockSizeFromPhysicalBlockSize(blockSize);
+    if (baseBlockSize <= Cipher::ciphertextSize(HEADER_LENGTH) + sizeof(FORMAT_VERSION_HEADER)) {
+      return 0;
+    }
+    return Cipher::plaintextSize(baseBlockSize - sizeof(FORMAT_VERSION_HEADER)) - HEADER_LENGTH;
+  }
+
+  void forEachBlock(std::function<void (const Key &)> callback) const override {
+    return _baseBlockStore->forEachBlock(std::move(callback));
+  }
+
 private:
 
   // This header is prepended to blocks to allow future versions to have compatibility.
   static constexpr uint16_t FORMAT_VERSION_HEADER = 0;
 
+  static constexpr unsigned int HEADER_LENGTH = Key::BINARY_LENGTH;
+
   cpputils::Data _encrypt(const Key &key, const cpputils::Data &data) const {
     cpputils::Data plaintextWithHeader = _prependKeyHeaderToData(key, data);
     cpputils::Data encrypted = Cipher::encrypt((byte*)plaintextWithHeader.data(), plaintextWithHeader.size(), _encKey);

src/blockstore/implementations/inmemory/InMemoryBlockStore2.cpp

@@ -62,5 +62,23 @@ future<void> InMemoryBlockStore2::store(const Key &key, const Data &data) {
   return make_ready_future();
 }
 
+uint64_t InMemoryBlockStore2::numBlocks() const {
+  return _blocks.size();
+}
+
+uint64_t InMemoryBlockStore2::estimateNumFreeBytes() const {
+  return cpputils::system::get_total_memory();
+}
+
+uint64_t InMemoryBlockStore2::blockSizeFromPhysicalBlockSize(uint64_t blockSize) const {
+  return blockSize;
+}
+
+void InMemoryBlockStore2::forEachBlock(std::function<void (const Key &)> callback) const {
+  for (const auto &entry : _blocks) {
+    callback(entry.first);
+  }
+}
+
 }
 }

src/blockstore/implementations/inmemory/InMemoryBlockStore2.h

@@ -17,6 +17,10 @@ public:
   boost::future<bool> remove(const Key &key) override;
   boost::future<boost::optional<cpputils::Data>> load(const Key &key) const override;
   boost::future<void> store(const Key &key, const cpputils::Data &data) override;
+  uint64_t numBlocks() const override;
+  uint64_t estimateNumFreeBytes() const override;
+  uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const override;
+  void forEachBlock(std::function<void (const Key &)> callback) const override;
 
 private:
   std::unordered_map<Key, cpputils::Data> _blocks;

src/blockstore/implementations/low2highlevel/LowToHighLevelBlock.cpp

@@ -0,0 +1,7 @@
+#include "LowToHighLevelBlock.h"
+
+namespace blockstore {
+namespace lowtohighlevel {
+
+}
+}

src/blockstore/implementations/low2highlevel/LowToHighLevelBlock.h

@@ -0,0 +1,122 @@
+#pragma once
+#ifndef MESSMER_BLOCKSTORE_IMPLEMENTATIONS_LOWTOHIGHLEVEL_LOWTOHIGHLEVELBLOCK_H_
+#define MESSMER_BLOCKSTORE_IMPLEMENTATIONS_LOWTOHIGHLEVEL_LOWTOHIGHLEVELBLOCK_H_
+
+#include "../../interface/Block.h"
+#include <cpp-utils/data/Data.h>
+#include "../../interface/BlockStore.h"
+#include "../../interface/BlockStore2.h"
+
+#include <cpp-utils/macros.h>
+#include <memory>
+#include <iostream>
+#include <boost/optional.hpp>
+#include <cpp-utils/crypto/symmetric/Cipher.h>
+#include <cpp-utils/assert/assert.h>
+#include <cpp-utils/data/DataUtils.h>
+#include <mutex>
+#include <cpp-utils/logging/logging.h>
+#include "LowToHighLevelBlockStore.h"
+
+namespace blockstore {
+namespace lowtohighlevel {
+
+class LowToHighLevelBlock final: public Block {
+public:
+  static boost::optional<cpputils::unique_ref<LowToHighLevelBlock>> TryCreateNew(BlockStore2 *baseBlockStore, const Key &key, cpputils::Data data);
+  static cpputils::unique_ref<LowToHighLevelBlock> Overwrite(BlockStore2 *baseBlockStore, const Key &key, cpputils::Data data);
+  static boost::optional<cpputils::unique_ref<LowToHighLevelBlock>> Load(BlockStore2 *baseBlockStore, const Key &key);
+
+  LowToHighLevelBlock(const Key& key, cpputils::Data data, BlockStore2 *baseBlockStore);
+  ~LowToHighLevelBlock();
+
+  const void *data() const override;
+  void write(const void *source, uint64_t offset, uint64_t count) override;
+  void flush() override;
+
+  size_t size() const override;
+  void resize(size_t newSize) override;
+
+private:
+  BlockStore2 *_baseBlockStore;
+  cpputils::Data _data;
+  bool _dataChanged;
+  std::mutex _mutex;
+
+  void _storeToBaseBlock();
+
+  DISALLOW_COPY_AND_ASSIGN(LowToHighLevelBlock);
+};
+
+
+inline boost::optional<cpputils::unique_ref<LowToHighLevelBlock>> LowToHighLevelBlock::TryCreateNew(BlockStore2 *baseBlockStore, const Key &key, cpputils::Data data) {
+  // TODO .get() is blocking
+  bool success = baseBlockStore->tryCreate(key, data.copy()).get(); // TODO Copy necessary?
+  if (!success) {
+    return boost::none;
+  }
+
+  return cpputils::make_unique_ref<LowToHighLevelBlock>(key, std::move(data), baseBlockStore);
+}
+
+inline cpputils::unique_ref<LowToHighLevelBlock> LowToHighLevelBlock::Overwrite(BlockStore2 *baseBlockStore, const Key &key, cpputils::Data data) {
+  auto baseBlock = baseBlockStore->store(key, data); // TODO Does it make sense to not store here, but only write back in the destructor of LowToHighLevelBlock? Also: What about tryCreate?
+  return cpputils::make_unique_ref<LowToHighLevelBlock>(key, std::move(data), baseBlockStore);
+}
+
+inline boost::optional<cpputils::unique_ref<LowToHighLevelBlock>> LowToHighLevelBlock::Load(BlockStore2 *baseBlockStore, const Key &key) {
+  boost::optional<cpputils::Data> loadedData = baseBlockStore->load(key).get(); // TODO .get() is blocking
+  if (loadedData == boost::none) {
+    return boost::none;
+  }
+  return cpputils::make_unique_ref<LowToHighLevelBlock>(key, std::move(*loadedData), baseBlockStore);
+}
+
+inline LowToHighLevelBlock::LowToHighLevelBlock(const Key& key, cpputils::Data data, BlockStore2 *baseBlockStore)
+    :Block(key),
+   _baseBlockStore(baseBlockStore),
+   _data(std::move(data)),
+   _dataChanged(false),
+   _mutex() {
+}
+
+inline LowToHighLevelBlock::~LowToHighLevelBlock() {
+  std::unique_lock<std::mutex> lock(_mutex);
+  _storeToBaseBlock();
+}
+
+inline const void *LowToHighLevelBlock::data() const {
+  return (uint8_t*)_data.data();
+}
+
+inline void LowToHighLevelBlock::write(const void *source, uint64_t offset, uint64_t count) {
+  ASSERT(offset <= size() && offset + count <= size(), "Write outside of valid area"); //Also check offset < size() because of possible overflow in the addition
+  std::memcpy((uint8_t*)_data.data()+offset, source, count);
+  _dataChanged = true;
+}
+
+inline void LowToHighLevelBlock::flush() {
+  std::unique_lock<std::mutex> lock(_mutex);
+  _storeToBaseBlock();
+}
+
+inline size_t LowToHighLevelBlock::size() const {
+  return _data.size();
+}
+
+inline void LowToHighLevelBlock::resize(size_t newSize) {
+  _data = cpputils::DataUtils::resize(std::move(_data), newSize);
+  _dataChanged = true;
+}
+
+inline void LowToHighLevelBlock::_storeToBaseBlock() {
+  if (_dataChanged) {
+    _baseBlockStore->store(key(), _data);
+    _dataChanged = false;
+  }
+}
+
+}
+}
+
+#endif

src/blockstore/implementations/low2highlevel/LowToHighLevelBlockStore.cpp

@@ -0,0 +1,69 @@
+#include <unordered_set>
+#include "LowToHighLevelBlockStore.h"
+#include "LowToHighLevelBlock.h"
+
+using cpputils::unique_ref;
+using cpputils::make_unique_ref;
+using cpputils::Data;
+using boost::none;
+using boost::optional;
+using std::string;
+namespace bf = boost::filesystem;
+
+namespace blockstore {
+namespace lowtohighlevel {
+
+LowToHighLevelBlockStore::LowToHighLevelBlockStore(unique_ref<BlockStore2> baseBlockStore)
+    : _baseBlockStore(std::move(baseBlockStore)) {
+}
+
+Key LowToHighLevelBlockStore::createKey() {
+    // TODO Is this the right way?
+    return cpputils::Random::PseudoRandom().getFixedSize<Key::BINARY_LENGTH>();
+}
+
+optional<unique_ref<Block>> LowToHighLevelBlockStore::tryCreate(const Key &key, Data data) {
+    //TODO Easier implementation? This is only so complicated because of the cast LowToHighLevelBlock -> Block
+    auto result = LowToHighLevelBlock::TryCreateNew(_baseBlockStore.get(), key, std::move(data));
+    if (result == boost::none) {
+        return boost::none;
+    }
+    return unique_ref<Block>(std::move(*result));
+}
+
+unique_ref<Block> LowToHighLevelBlockStore::overwrite(const Key &key, Data data) {
+    return unique_ref<Block>(
+        LowToHighLevelBlock::Overwrite(_baseBlockStore.get(), key, std::move(data))
+    );
+}
+
+optional<unique_ref<Block>> LowToHighLevelBlockStore::load(const Key &key) {
+    auto result = optional<unique_ref<Block>>(LowToHighLevelBlock::Load(_baseBlockStore.get(), key));
+    if (result == boost::none) {
+      return boost::none;
+    }
+    return unique_ref<Block>(std::move(*result));
+}
+
+void LowToHighLevelBlockStore::remove(const Key &key) {
+    _baseBlockStore->remove(key);
+}
+
+uint64_t LowToHighLevelBlockStore::numBlocks() const {
+    return _baseBlockStore->numBlocks();
+}
+
+uint64_t LowToHighLevelBlockStore::estimateNumFreeBytes() const {
+    return _baseBlockStore->estimateNumFreeBytes();
+}
+
+uint64_t LowToHighLevelBlockStore::blockSizeFromPhysicalBlockSize(uint64_t blockSize) const {
+    return _baseBlockStore->blockSizeFromPhysicalBlockSize(blockSize);
+}
+
+void LowToHighLevelBlockStore::forEachBlock(std::function<void (const Key &)> callback) const {
+    _baseBlockStore->forEachBlock(std::move(callback));
+}
+
+}
+}

src/blockstore/implementations/low2highlevel/LowToHighLevelBlockStore.h

@@ -0,0 +1,42 @@
+#pragma once
+#ifndef MESSMER_BLOCKSTORE_IMPLEMENTATIONS_LOWTOHIGHLEVEL_LOWTOHIGHLEVELBLOCKSTORE_H_
+#define MESSMER_BLOCKSTORE_IMPLEMENTATIONS_LOWTOHIGHLEVEL_LOWTOHIGHLEVELBLOCKSTORE_H_
+
+#include "../../interface/BlockStore.h"
+#include "../../interface/BlockStore2.h"
+#include <cpp-utils/macros.h>
+#include <cpp-utils/pointer/cast.h>
+#include <iostream>
+
+// TODO Think each function through and make sure it's as performant
+//      to use LowToHighLevelBlockStore<OnDiskBlockStore2> as to use
+//      OnDiskBlockStore directly (i.e. no additional stores/loads from the disk)
+//      (same for other base block stores)
+
+namespace blockstore {
+namespace lowtohighlevel {
+
+class LowToHighLevelBlockStore final: public BlockStore {
+public:
+  LowToHighLevelBlockStore(cpputils::unique_ref<BlockStore2> baseBlockStore);
+
+  Key createKey() override;
+  boost::optional<cpputils::unique_ref<Block>> tryCreate(const Key &key, cpputils::Data data) override;
+  cpputils::unique_ref<Block> overwrite(const blockstore::Key &key, cpputils::Data data) override;
+  boost::optional<cpputils::unique_ref<Block>> load(const Key &key) override;
+  void remove(const Key &key) override;
+  uint64_t numBlocks() const override;
+  uint64_t estimateNumFreeBytes() const override;
+  uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const override;
+  void forEachBlock(std::function<void (const Key &)> callback) const override;
+
+private:
+  cpputils::unique_ref<BlockStore2> _baseBlockStore;
+
+  DISALLOW_COPY_AND_ASSIGN(LowToHighLevelBlockStore);
+};
+
+}
+}
+
+#endif

src/blockstore/implementations/ondisk/OnDiskBlockStore2.h

@@ -8,6 +8,7 @@
 #include <cpp-utils/pointer/unique_ref.h>
 #include "OnDiskBlockStore.h"
 #include <cpp-utils/logging/logging.h>
+#include <sys/statvfs.h>
 
 namespace blockstore {
 namespace ondisk {
@@ -61,6 +62,41 @@ public:
     return boost::make_ready_future();
   }
 
+  uint64_t numBlocks() const override {
+    uint64_t count = 0;
+    for (auto prefixDir = boost::filesystem::directory_iterator(_rootDir); prefixDir != boost::filesystem::directory_iterator(); ++prefixDir) {
+      if (boost::filesystem::is_directory(prefixDir->path())) {
+        count += std::distance(boost::filesystem::directory_iterator(prefixDir->path()), boost::filesystem::directory_iterator());
+      }
+    }
+    return count;
+  }
+
+  uint64_t estimateNumFreeBytes() const override {
+    struct statvfs stat;
+    ::statvfs(_rootDir.c_str(), &stat);
+    return stat.f_bsize*stat.f_bavail;
+  }
+
+  uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const override {
+    if(blockSize <= formatVersionHeaderSize()) {
+      return 0;
+    }
+    return blockSize - formatVersionHeaderSize();
+  }
+
+  void forEachBlock(std::function<void (const Key &)> callback) const override {
+    for (auto prefixDir = boost::filesystem::directory_iterator(_rootDir); prefixDir != boost::filesystem::directory_iterator(); ++prefixDir) {
+      if (boost::filesystem::is_directory(prefixDir->path())) {
+        std::string blockKeyPrefix = prefixDir->path().filename().native();
+        for (auto block = boost::filesystem::directory_iterator(prefixDir->path()); block != boost::filesystem::directory_iterator(); ++block) {
+          std::string blockKeyPostfix = block->path().filename().native();
+          callback(Key::FromString(blockKeyPrefix + blockKeyPostfix));
+        }
+      }
+    }
+  }
+
 private:
   boost::filesystem::path _rootDir;
 

src/blockstore/implementations/versioncounting/VersionCountingBlockStore2.h

@@ -53,6 +53,41 @@ public:
     return _baseBlockStore->store(key, dataWithHeader);
   }
 
+  uint64_t numBlocks() const override {
+    return _baseBlockStore->numBlocks();
+  }
+
+  uint64_t estimateNumFreeBytes() const override {
+    return _baseBlockStore->estimateNumFreeBytes();
+  }
+
+  uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const override {
+    uint64_t baseBlockSize = _baseBlockStore->blockSizeFromPhysicalBlockSize(blockSize);
+    if (baseBlockSize <= HEADER_LENGTH) {
+      return 0;
+    }
+    return baseBlockSize - HEADER_LENGTH;
+  }
+
+  void forEachBlock(std::function<void (const Key &)> callback) const override {
+    if (!_missingBlockIsIntegrityViolation) {
+      return _baseBlockStore->forEachBlock(std::move(callback));
+    }
+
+    std::unordered_set<blockstore::Key> existingBlocks = _knownBlockVersions.existingBlocks();
+    _baseBlockStore->forEachBlock([&existingBlocks, callback] (const Key &key) {
+      callback(key);
+
+      auto found = existingBlocks.find(key);
+      if (found != existingBlocks.end()) {
+        existingBlocks.erase(found);
+      }
+    });
+    if (!existingBlocks.empty()) {
+      integrityViolationDetected("A block that should have existed wasn't found.");
+    }
+  }
+
 private:
   // This header is prepended to blocks to allow future versions to have compatibility.
   static constexpr uint16_t FORMAT_VERSION_HEADER = 0;

src/blockstore/interface/BlockStore.h

@@ -28,7 +28,7 @@ public:
 
   // Returns, how much space a block has if we allow it to take the given physical block size (i.e. after removing headers, checksums, whatever else).
   // This can be used to create blocks with a certain physical block size.
-  virtual uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const = 0;
+  virtual uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const = 0; // TODO Test
 
   virtual void forEachBlock(std::function<void (const Key &)> callback) const = 0;
 

src/blockstore/interface/BlockStore2.h

@@ -18,14 +18,19 @@ class BlockStore2 {
 public:
   virtual ~BlockStore2() {}
 
+  __attribute__((warn_unused_result))
   virtual boost::future<bool> tryCreate(const Key &key, const cpputils::Data &data) = 0;
+  __attribute__((warn_unused_result))
   virtual boost::future<bool> remove(const Key &key) = 0;
 
+  __attribute__((warn_unused_result))
   virtual boost::future<boost::optional<cpputils::Data>> load(const Key &key) const = 0;
 
   // Store the block with the given key. If it doesn't exist, it is created.
+  __attribute__((warn_unused_result))
   virtual boost::future<void> store(const Key &key, const cpputils::Data &data) = 0;
 
+  __attribute__((warn_unused_result))
   boost::future<Key> create(cpputils::Data data) {
     Key key = cpputils::Random::PseudoRandom().getFixedSize<Key::BINARY_LENGTH>();
     boost::future<bool> successFuture = tryCreate(key, data);
@@ -37,6 +42,12 @@ public:
       }
     });
   }
+
+  virtual uint64_t numBlocks() const = 0;
+  //TODO Test estimateNumFreeBytes
+  virtual uint64_t estimateNumFreeBytes() const = 0;
+  virtual uint64_t blockSizeFromPhysicalBlockSize(uint64_t blockSize) const = 0; // TODO Test
+  virtual void forEachBlock(std::function<void (const Key &)> callback) const = 0;
 };
 
 }

test/blockstore/CMakeLists.txt

@@ -38,6 +38,7 @@ set(SOURCES
     implementations/versioncounting/KnownBlockVersionsTest.cpp
     implementations/versioncounting/VersionCountingBlockStoreTest_Generic.cpp
     implementations/versioncounting/VersionCountingBlockStoreTest_Specific.cpp
+    implementations/low2highlevel/LowToHighLevelBlockStoreTest.cpp
 )
 
 add_executable(${PROJECT_NAME} ${SOURCES})

test/blockstore/implementations/low2highlevel/LowToHighLevelBlockStoreTest.cpp

@@ -0,0 +1,31 @@
+#include "blockstore/implementations/low2highlevel/LowToHighLevelBlockStore.h"
+#include "blockstore/implementations/testfake/FakeBlockStore.h"
+#include "blockstore/implementations/inmemory/InMemoryBlockStore2.h"
+#include "../../testutils/BlockStoreTest.h"
+#include <gtest/gtest.h>
+#include <cpp-utils/tempfile/TempFile.h>
+
+using ::testing::Test;
+
+using blockstore::BlockStore;
+using blockstore::BlockStore2;
+using blockstore::lowtohighlevel::LowToHighLevelBlockStore;
+using blockstore::testfake::FakeBlockStore;
+using blockstore::inmemory::InMemoryBlockStore2;
+
+using cpputils::Data;
+using cpputils::DataFixture;
+using cpputils::make_unique_ref;
+using cpputils::unique_ref;
+using cpputils::TempFile;
+
+class LowToHighLevelBlockStoreTestFixture: public BlockStoreTestFixture {
+public:
+  LowToHighLevelBlockStoreTestFixture() {}
+
+  unique_ref<BlockStore> createBlockStore() override {
+    return make_unique_ref<LowToHighLevelBlockStore>(make_unique_ref<InMemoryBlockStore2>());
+  }
+};
+
+INSTANTIATE_TYPED_TEST_CASE_P(LowToHighLevel, BlockStoreTest, LowToHighLevelBlockStoreTestFixture);

test/blockstore/testutils/BlockStore2Test.h

@@ -33,6 +33,24 @@ public:
 
   ConcreteBlockStoreTestFixture fixture;
   cpputils::unique_ref<blockstore::BlockStore2> blockStore;
+
+  template<class Entry>
+  void EXPECT_UNORDERED_EQ(const std::vector<Entry> &expected, std::vector<Entry> actual) {
+    EXPECT_EQ(expected.size(), actual.size());
+    for (const Entry &expectedEntry : expected) {
+      removeOne(&actual, expectedEntry);
+    }
+  }
+
+  template<class Entry>
+  void removeOne(std::vector<Entry> *entries, const Entry &toRemove) {
+    auto found = std::find(entries->begin(), entries->end(), toRemove);
+    if (found != entries->end()) {
+      entries->erase(found);
+      return;
+    }
+    EXPECT_TRUE(false);
+  }
 };
 
 TYPED_TEST_CASE_P(BlockStore2Test);
@@ -265,6 +283,91 @@ TYPED_TEST_P(BlockStore2Test, givenNonEmptyBlockStore_whenLoadingNonExistingBloc
   EXPECT_EQ(boost::none, this->blockStore->load(key).get());
 }
 
+TYPED_TEST_P(BlockStore2Test, NumBlocksIsCorrectOnEmptyBlockstore) {
+  auto blockStore = this->fixture.createBlockStore();
+  EXPECT_EQ(0u, blockStore->numBlocks());
+}
+
+TYPED_TEST_P(BlockStore2Test, NumBlocksIsCorrectAfterAddingOneBlock) {
+  auto blockStore = this->fixture.createBlockStore();
+  blockStore->create(cpputils::Data(1)).wait();
+  EXPECT_EQ(1u, blockStore->numBlocks());
+}
+
+TYPED_TEST_P(BlockStore2Test, NumBlocksIsCorrectAfterRemovingTheLastBlock) {
+  auto blockStore = this->fixture.createBlockStore();
+  blockstore::Key key = blockStore->create(cpputils::Data(1)).get();
+  blockStore->remove(key).wait();
+  EXPECT_EQ(0u, blockStore->numBlocks());
+}
+
+TYPED_TEST_P(BlockStore2Test, NumBlocksIsCorrectAfterAddingTwoBlocks) {
+  auto blockStore = this->fixture.createBlockStore();
+  blockStore->create(cpputils::Data(1)).wait();
+  blockStore->create(cpputils::Data(0)).wait();
+  EXPECT_EQ(2u, blockStore->numBlocks());
+}
+
+TYPED_TEST_P(BlockStore2Test, NumBlocksIsCorrectAfterRemovingABlock) {
+  auto blockStore = this->fixture.createBlockStore();
+  blockstore::Key key = blockStore->create(cpputils::Data(1)).get();
+  blockStore->create(cpputils::Data(1)).wait();
+  blockStore->remove(key).wait();
+  EXPECT_EQ(1u, blockStore->numBlocks());
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_zeroblocks) {
+  auto blockStore = this->fixture.createBlockStore();
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({}, mockForEachBlockCallback.called_with);
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_oneblock) {
+  auto blockStore = this->fixture.createBlockStore();
+  auto key = blockStore->create(cpputils::Data(1)).get();
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({key}, mockForEachBlockCallback.called_with);
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_twoblocks) {
+  auto blockStore = this->fixture.createBlockStore();
+  auto key1 = blockStore->create(cpputils::Data(1)).get();
+  auto key2 = blockStore->create(cpputils::Data(1)).get();
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({key1, key2}, mockForEachBlockCallback.called_with);
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_threeblocks) {
+  auto blockStore = this->fixture.createBlockStore();
+  auto key1 = blockStore->create(cpputils::Data(1)).get();
+  auto key2 = blockStore->create(cpputils::Data(1)).get();
+  auto key3 = blockStore->create(cpputils::Data(1)).get();
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({key1, key2, key3}, mockForEachBlockCallback.called_with);
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_doesntListRemovedBlocks_oneblock) {
+  auto blockStore = this->fixture.createBlockStore();
+  auto key1 = blockStore->create(cpputils::Data(1)).get();
+  blockStore->remove(key1).wait();
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({}, mockForEachBlockCallback.called_with);
+}
+
+TYPED_TEST_P(BlockStore2Test, ForEachBlock_doesntListRemovedBlocks_twoblocks) {
+  auto blockStore = this->fixture.createBlockStore();
+  auto key1 = blockStore->create(cpputils::Data(1)).get();
+  auto key2 = blockStore->create(cpputils::Data(1)).get();
+  blockStore->remove(key1);
+  MockForEachBlockCallback mockForEachBlockCallback;
+  blockStore->forEachBlock(mockForEachBlockCallback.callback());
+  this->EXPECT_UNORDERED_EQ({key2}, mockForEachBlockCallback.called_with);
+}
 
 REGISTER_TYPED_TEST_CASE_P(BlockStore2Test,
   givenNonEmptyBlockStore_whenCallingTryCreateOnExistingBlock_thenFails,
@@ -300,7 +403,18 @@ REGISTER_TYPED_TEST_CASE_P(BlockStore2Test,
   givenEmptyBlockStore_whenStoringAndLoadingExistingNonEmptyBlock_thenCorrectBlockLoads,
   givenNonEmptyBlockStore_whenStoringAndLoadingExistingNonEmptyBlock_thenCorrectBlockLoads,
   givenEmptyBlockStore_whenLoadingNonExistingBlock_thenFails,
-  givenNonEmptyBlockStore_whenLoadingNonExistingBlock_thenFails
+  givenNonEmptyBlockStore_whenLoadingNonExistingBlock_thenFails,
+  NumBlocksIsCorrectOnEmptyBlockstore,
+  NumBlocksIsCorrectAfterAddingOneBlock,
+  NumBlocksIsCorrectAfterRemovingTheLastBlock,
+  NumBlocksIsCorrectAfterAddingTwoBlocks,
+  NumBlocksIsCorrectAfterRemovingABlock,
+  ForEachBlock_zeroblocks,
+  ForEachBlock_oneblock,
+  ForEachBlock_twoblocks,
+  ForEachBlock_threeblocks,
+  ForEachBlock_doesntListRemovedBlocks_oneblock,
+  ForEachBlock_doesntListRemovedBlocks_twoblocks
 );