fix no response when calling cypher plugin through neo4j driver (TuGr…

…aph-family#434) * add error_code in Bolt Failure msg fix no response when calling cypher plugin through bolt * fix ci failure * fix typo
ljcui · Mar 14, 2024 · d1220bc · d1220bc
1 parent 0ec400d
commit d1220bc
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 5 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -25,6 +25,7 @@ jobs:
           echo "Listing 100 largest packages"
           dpkg-query -Wf '${Installed-Size}\t${Package}\n' | sort -n | tail -n 100
           echo "Removing large packages"
+          sudo apt-get update
           sudo apt-get remove -y '^dotnet-.*'
           sudo apt-get remove -y '^llvm-.*'
           sudo apt-get remove -y 'php.*'
@@ -68,6 +69,7 @@ jobs:
           echo "Listing 100 largest packages"
           dpkg-query -Wf '${Installed-Size}\t${Package}\n' | sort -n | tail -n 100
           echo "Removing large packages"
+          sudo apt-get update
           sudo apt-get remove -y '^dotnet-.*'
           sudo apt-get remove -y '^llvm-.*'
           sudo apt-get remove -y 'php.*'
@@ -111,6 +113,7 @@ jobs:
           echo "Listing 100 largest packages"
           dpkg-query -Wf '${Installed-Size}\t${Package}\n' | sort -n | tail -n 100
           echo "Removing large packages"
+          sudo apt-get update
           sudo apt-get remove -y '^dotnet-.*'
           sudo apt-get remove -y '^llvm-.*'
           sudo apt-get remove -y 'php.*'

diff --git a/include/lgraph/lgraph_exceptions.h b/include/lgraph/lgraph_exceptions.h
@@ -18,6 +18,7 @@
 namespace lgraph_api {
 
 #define ERROR_CODES \
+X(UnknownError, "Unknown error.") \
 X(InvalidGalaxy, "Invalid Galaxy.") \
 X(InvalidGraphDB, "Invalid GraphDB.") \
 X(InvalidTxn, "Invalid transaction.") \

diff --git a/src/cypher/execution_plan/ops/op_produce_results.h b/src/cypher/execution_plan/ops/op_produce_results.h
@@ -274,6 +274,10 @@ class ProduceResults : public OpBase {
             auto child = children[0];
             auto res = child->Consume(ctx);
             if (res != OP_OK) {
+                if (ctx->result_->Size() > 0 &&
+                    session->streaming_msg.value().type == bolt::BoltMsg::PullN) {
+                    session->ps.AppendRecords(ctx->result_->BoltRecords());
+                }
                 session->ps.AppendSuccess();
                 session->state = bolt::SessionState::READY;
                 ctx->bolt_conn_->PostResponse(std::move(session->ps.MutableBuffer()));

diff --git a/src/server/bolt_handler.cpp b/src/server/bolt_handler.cpp
@@ -20,6 +20,7 @@
 #include "server/bolt_server.h"
 #include "server/bolt_session.h"
 #include "db/galaxy.h"
+using namespace lgraph_api;
 namespace bolt {
 extern boost::asio::io_service workers;
 std::unordered_map<std::string, cypher::FieldData> ConvertParameters(
@@ -53,6 +54,14 @@ void BoltFSM(std::shared_ptr<BoltConnection> conn) {
     auto conn_id = conn->conn_id();
     LOG_DEBUG() << FMA_FMT("bolt fsm thread[conn_id:{}] start.", conn_id);
     auto session = (BoltSession*)conn->GetContext();
+    auto RespondFailure = [&conn, &session](ErrorCode code, const std::string& msg){
+        bolt::PackStream ps;
+        ps.AppendFailure(
+            {{"code", ErrorCodeToString(code)},
+             {"message", msg}});
+        conn->PostResponse(std::move(ps.MutableBuffer()));
+        session->state = SessionState::FAILED;
+    };
     while (!conn->has_closed()) {
         auto msg = session->msgs.Pop(std::chrono::milliseconds(100));
         if (!msg) {  // msgs pop timeout
@@ -147,13 +156,12 @@ void BoltFSM(std::shared_ptr<BoltConnection> conn) {
                     sm->GetCypherScheduler()->Eval(&ctx, lgraph_api::GraphQueryType::CYPHER,
                                                    cypher, elapsed);
                     LOG_DEBUG() << "Cypher execution completed";
+                } catch (const lgraph_api::LgraphException& e) {
+                    LOG_ERROR() << e.what();
+                    RespondFailure(e.code(), e.msg());
                 } catch (std::exception& e) {
                     LOG_ERROR() << e.what();
-                    bolt::PackStream ps;
-                    ps.AppendFailure({{"code", "error"},
-                                      {"message", e.what()}});
-                    conn->PostResponse(std::move(ps.MutableBuffer()));
-                    session->state = SessionState::FAILED;
+                    RespondFailure(ErrorCode::UnknownError, e.what());
                 }
             }
         }