refactor: move write_batch_delete into rocksdb_wrapper

src/server/pegasus_write_service_impl.h

@@ -169,21 +169,19 @@ class pegasus_write_service::impl : public dsn::replication::replica_base
             return empty_put(decree);
         }
 
+        auto cleanup = dsn::defer([this]() { _rocksdb_wrapper->clear_up_write_batch(); });
         for (auto &sort_key : update.sort_keys) {
-            resp.error =
-                db_write_batch_delete(decree, composite_raw_key(update.hash_key, sort_key));
+            resp.error = _rocksdb_wrapper->write_batch_delete(
+                decree, composite_raw_key(update.hash_key, sort_key));
             if (resp.error) {
-                clear_up_batch_states(decree, resp.error);
                 return resp.error;
             }
         }
 
-        resp.error = db_write(decree);
+        resp.error = _rocksdb_wrapper->write(decree);
         if (resp.error == 0) {
             resp.count = update.sort_keys.size();
         }
-
-        clear_up_batch_states(decree, resp.error);
         return resp.error;
     }
 

src/server/rocksdb_wrapper.cpp

@@ -160,6 +160,25 @@ int rocksdb_wrapper::write(int64_t decree)
     return status.code();
 }
 
+int rocksdb_wrapper::write_batch_delete(int64_t decree, dsn::string_view raw_key)
+{
+    FAIL_POINT_INJECT_F("db_write_batch_delete",
+                        [](dsn::string_view) -> int { return FAIL_DB_WRITE_BATCH_DELETE; });
+
+    rocksdb::Status s = _write_batch->Delete(utils::to_rocksdb_slice(raw_key));
+    if (dsn_unlikely(!s.ok())) {
+        dsn::blob hash_key, sort_key;
+        pegasus_restore_key(dsn::blob(raw_key.data(), 0, raw_key.size()), hash_key, sort_key);
+        derror_rocksdb("write_batch_delete",
+                       s.ToString(),
+                       "decree: {}, hash_key: {}, sort_key: {}",
+                       decree,
+                       utils::c_escape_string(hash_key),
+                       utils::c_escape_string(sort_key));
+    }
+    return s.code();
+}
+
 void rocksdb_wrapper::clear_up_write_batch() { _write_batch->Clear(); }
 
 void rocksdb_wrapper::set_default_ttl(uint32_t ttl)

src/server/rocksdb_wrapper.h

@@ -62,6 +62,7 @@ class rocksdb_wrapper : public dsn::replication::replica_base
                             dsn::string_view value,
                             uint32_t expire_sec);
     int write(int64_t decree);
+    int write_batch_delete(int64_t decree, dsn::string_view raw_key);
     void clear_up_write_batch();
 
     void set_default_ttl(uint32_t ttl);