refactor(interactive): Let transction returns commit result (#4507)

In Interactive's execution engine, transactions such as
`ReadTransaction`, `UpdateTransaction`, and `InsertTransaction` are
employed to maintain data consistency and integrity throughout
operations. Each transaction can either succeed or fail, and it is
important to understand the transactional guarantees provided by
Interactive:

1. For every transaction, we first write the Write-Ahead Log (WAL) to
persistent storage before applying it to the graph data. This ensures a
reliable record of the transaction steps.

2. If a transaction returns `false` during the `commit()` process, the
error occurred prior to applying the WAL to the graph data. This type of
failure could arise during the construction of the WAL or during its
writing phase.

3. It is important to note that errors can still occur when replaying
the WAL to the graph database. Replaying might fail due to limitations
in resources or due to unforeseen bugs. **However,** any errors
encountered during this stage will be handled via exceptions or may
result in process failure. Currently, there is no established mechanism
to handle such failures. Future improvements should focus on
implementing failover strategies, potentially allowing the GraphDB to
continue replaying the WAL until it succeeds.

Author: zhanglei1949

docs/flex/interactive/development/dev_and_test.md

@@ -250,4 +250,15 @@ The mapping between status codes and HTTP codes is shown in the table below.
 | INVALID_IMPORT_FILE(104)   | 400         |
 | IO_ERROR(105)             | 500         |
 | QUERY_FAILED(106)         | 500         |
-| default                             | 500         |
+| default                             | 500         |
+
+
+### Transactions
+
+In Interactive's execution engine, transactions such as `ReadTransaction`, `UpdateTransaction`, and `InsertTransaction` are employed to maintain data consistency and integrity throughout operations. Each transaction can either succeed or fail, and it is important to understand the transactional guarantees provided by Interactive:
+
+1. For every transaction, we first write the Write-Ahead Log (WAL) to persistent storage before applying it to the graph data. This ensures a reliable record of the transaction steps.
+
+2. If a transaction returns `false` during the `commit()` process, the error occurred prior to applying the WAL to the graph data. This type of failure could arise during the construction of the WAL or during its writing phase.
+
+3. It is important to note that errors can still occur when replaying the WAL to the graph database. Replaying might fail due to limitations in resources or due to unforeseen bugs. **However,** any errors encountered during this stage will be handled via exceptions or may result in process failure. Currently, there is no established mechanism to handle such failures. Future improvements should focus on implementing failover strategies, potentially allowing the GraphDB to continue replaying the WAL until it succeeds.

flex/engines/graph_db/database/compact_transaction.cc

@@ -31,14 +31,18 @@ CompactTransaction::~CompactTransaction() { Abort(); }
 
 timestamp_t CompactTransaction::timestamp() const { return timestamp_; }
 
-void CompactTransaction::Commit() {
+bool CompactTransaction::Commit() {
   if (timestamp_ != std::numeric_limits<timestamp_t>::max()) {
     auto* header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
     header->length = 0;
     header->timestamp = timestamp_;
     header->type = 1;
 
-    logger_.append(arc_.GetBuffer(), arc_.GetSize());
+    if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+      LOG(ERROR) << "Failed to append wal log";
+      Abort();
+      return false;
+    }
     arc_.Clear();
 
     LOG(INFO) << "before compact - " << timestamp_;
@@ -48,6 +52,7 @@ void CompactTransaction::Commit() {
     vm_.release_update_timestamp(timestamp_);
     timestamp_ = std::numeric_limits<timestamp_t>::max();
   }
+  return true;
 }
 
 void CompactTransaction::Abort() {

flex/engines/graph_db/database/compact_transaction.h

@@ -33,7 +33,7 @@ class CompactTransaction {
 
   timestamp_t timestamp() const;
 
-  void Commit();
+  bool Commit();
 
   void Abort();
 

flex/engines/graph_db/database/graph_db_session.cc

@@ -58,8 +58,7 @@ UpdateTransaction GraphDBSession::GetUpdateTransaction() {
 }
 
 bool GraphDBSession::BatchUpdate(UpdateBatch& batch) {
-  GetUpdateTransaction().batch_commit(batch);
-  return true;
+  return GetUpdateTransaction().batch_commit(batch);
 }
 
 const MutablePropertyFragment& GraphDBSession::graph() const {

flex/engines/graph_db/database/insert_transaction.cc

@@ -144,26 +144,31 @@ bool InsertTransaction::AddEdge(label_t src_label, const Any& src,
   return true;
 }
 
-void InsertTransaction::Commit() {
+bool InsertTransaction::Commit() {
   if (timestamp_ == std::numeric_limits<timestamp_t>::max()) {
-    return;
+    return true;
   }
   if (arc_.GetSize() == sizeof(WalHeader)) {
     vm_.release_insert_timestamp(timestamp_);
     clear();
-    return;
+    return true;
   }
   auto* header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
   header->length = arc_.GetSize() - sizeof(WalHeader);
   header->type = 0;
   header->timestamp = timestamp_;
 
-  logger_.append(arc_.GetBuffer(), arc_.GetSize());
+  if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+    LOG(ERROR) << "Failed to append wal log";
+    Abort();
+    return false;
+  }
   IngestWal(graph_, timestamp_, arc_.GetBuffer() + sizeof(WalHeader),
             header->length, alloc_);
 
   vm_.release_insert_timestamp(timestamp_);
   clear();
+  return true;
 }
 
 void InsertTransaction::Abort() {

flex/engines/graph_db/database/insert_transaction.h

@@ -48,7 +48,7 @@ class InsertTransaction {
   bool AddEdge(label_t src_label, const Any& src, label_t dst_label,
                const Any& dst, label_t edge_label, const Any& prop);
 
-  void Commit();
+  bool Commit();
 
   void Abort();
 

flex/engines/graph_db/database/read_transaction.cc

@@ -34,7 +34,10 @@ std::string ReadTransaction::run(
 
 timestamp_t ReadTransaction::timestamp() const { return timestamp_; }
 
-void ReadTransaction::Commit() { release(); }
+bool ReadTransaction::Commit() {
+  release();
+  return true;
+}
 
 void ReadTransaction::Abort() { release(); }
 

flex/engines/graph_db/database/read_transaction.h

@@ -384,7 +384,7 @@ class ReadTransaction {
 
   timestamp_t timestamp() const;
 
-  void Commit();
+  bool Commit();
 
   void Abort();
 

flex/engines/graph_db/database/single_edge_insert_transaction.cc

@@ -109,15 +109,19 @@ timestamp_t SingleEdgeInsertTransaction::timestamp() const {
   return timestamp_;
 }
 
-void SingleEdgeInsertTransaction::Commit() {
+bool SingleEdgeInsertTransaction::Commit() {
   if (timestamp_ == std::numeric_limits<timestamp_t>::max()) {
-    return;
+    return false;
   }
   auto* header = reinterpret_cast<WalHeader*>(arc_.GetBuffer());
   header->length = arc_.GetSize() - sizeof(WalHeader);
   header->type = 0;
   header->timestamp = timestamp_;
-  logger_.append(arc_.GetBuffer(), arc_.GetSize());
+  if (!logger_.append(arc_.GetBuffer(), arc_.GetSize())) {
+    LOG(ERROR) << "Failed to append wal log";
+    Abort();
+    return false;
+  }
 
   grape::OutArchive arc;
   {
@@ -134,6 +138,7 @@ void SingleEdgeInsertTransaction::Commit() {
                     timestamp_, arc, alloc_);
   vm_.release_insert_timestamp(timestamp_);
   clear();
+  return true;
 }
 
 void SingleEdgeInsertTransaction::clear() {

flex/engines/graph_db/database/single_edge_insert_transaction.h

@@ -41,7 +41,7 @@ class SingleEdgeInsertTransaction {
 
   timestamp_t timestamp() const;
 
-  void Commit();
+  bool Commit();
 
  private:
   void clear();