Return write error on reaching blob dir size limit

Summary:
I found that we continue accepting writes even when the blob db goes beyond the configured blob directory size limit. Now, we return an error for writes on reaching `blob_dir_size` limit and if `is_fifo` is set to false. (We cannot just drop any file when `is_fifo` is true.)

Deleting the oldest file when `is_fifo` is true will be handled in a later PR.
Closes https://github.com/facebook/rocksdb/pull/3060

Differential Revision: D6136156

Pulled By: sagar0

fbshipit-source-id: 2f11cb3f2eedfa94524fbfa2613dd64bfad7a23c

utilities/blob_db/blob_db_impl.cc

@@ -1050,6 +1050,14 @@ uint64_t BlobDBImpl::ExtractExpiration(const Slice& key, const Slice& value,
 Status BlobDBImpl::AppendBlob(const std::shared_ptr<BlobFile>& bfile,
                               const std::string& headerbuf, const Slice& key,
                               const Slice& value, std::string* index_entry) {
+  auto size_put = BlobLogRecord::kHeaderSize + key.size() + value.size();
+  if (bdb_options_.blob_dir_size > 0 &&
+      (total_blob_space_.load() + size_put) > bdb_options_.blob_dir_size) {
+    if (!bdb_options_.is_fifo) {
+      return Status::NoSpace("Blob DB reached the maximum configured size.");
+    }
+  }
+
   Status s;
 
   uint64_t blob_offset = 0;
@@ -1073,7 +1081,6 @@ Status BlobDBImpl::AppendBlob(const std::shared_ptr<BlobFile>& bfile,
 
   // increment blob count
   bfile->blob_count_++;
-  auto size_put = BlobLogRecord::kHeaderSize + key.size() + value.size();
 
   bfile->file_size_ += size_put;
   last_period_write_ += size_put;

utilities/blob_db/blob_db_test.cc

@@ -653,6 +653,7 @@ TEST_F(BlobDBTest, GCOldestSimpleBlobFileWhenOutOfSpace) {
   Options options;
   options.env = mock_env_.get();
   BlobDBOptions bdb_options;
+  bdb_options.is_fifo = true;
   bdb_options.blob_dir_size = 100;
   bdb_options.blob_file_size = 100;
   bdb_options.disable_background_tasks = true;
@@ -870,6 +871,29 @@ TEST_F(BlobDBTest, MigrateFromPlainRocksDB) {
   delete db;
 }
 
+// Test to verify that a NoSpace IOError Status is returned on reaching
+// blob_dir_size limit. 
+TEST_F(BlobDBTest, OutOfSpace) {
+  // Use mock env to stop wall clock.
+  Options options;
+  options.env = mock_env_.get();
+  BlobDBOptions bdb_options;
+  bdb_options.blob_dir_size = 150;
+  bdb_options.disable_background_tasks = true;
+  Open(bdb_options);
+
+  // Each stored blob has an overhead of about 42 bytes currently.
+  // So a small key + a 100 byte blob should take up ~150 bytes in the db.
+  std::string value(100, 'v');
+  ASSERT_OK(blob_db_->PutWithTTL(WriteOptions(), "key1", value, 60));
+
+  // Putting another blob should fail as ading it would exceed the blob_dir_size
+  // limit.
+  Status s = blob_db_->PutWithTTL(WriteOptions(), "key2", value, 60);
+  ASSERT_TRUE(s.IsIOError());
+  ASSERT_TRUE(s.IsNoSpace());
+}
+
 }  //  namespace blob_db
 }  //  namespace rocksdb