Improved Postgres connection stability (#2286)

Author: Demilivor

orm_lib/src/DbClientImpl.cc

@@ -429,6 +429,8 @@ DbConnectionPtr DbClientImpl::newConnection(trantor::EventLoop *loop)
         }
         // Reconnect after 1 second
         auto loop = closeConnPtr->loop();
+        // closeConnPtr may be not valid. Close the connection file descriptor.
+        closeConnPtr->disconnect();
         loop->runAfter(1, [weakPtr, loop, closeConnPtr] {
             auto thisPtr = weakPtr.lock();
             if (!thisPtr)

orm_lib/src/postgresql_impl/PgBatchConnection.cc

@@ -87,11 +87,14 @@ PgConnection::PgConnection(trantor::EventLoop *loop,
                                   [](PGconn *conn) { PQfinish(conn); })),
       channel_(loop, PQsocket(connectionPtr_.get()))
 {
+    if (channel_.fd() < 0)
+    {
+        LOG_ERROR << "Failed to create Postgres connection";
+    }
 }
 
 void PgConnection::init()
 {
-    PQsetnonblocking(connectionPtr_.get(), 1);
     if (channel_.fd() < 0)
     {
         LOG_ERROR << "Connection with Postgres could not be established";
@@ -103,6 +106,8 @@ void PgConnection::init()
         }
         return;
     }
+
+    PQsetnonblocking(connectionPtr_.get(), 1);
     channel_.setReadCallback([this]() {
         if (status_ == ConnectStatus::Bad)
         {
@@ -165,8 +170,11 @@ void PgConnection::disconnect()
     auto thisPtr = shared_from_this();
     loop_->runInLoop([thisPtr, &pro]() {
         thisPtr->status_ = ConnectStatus::Bad;
-        thisPtr->channel_.disableAll();
-        thisPtr->channel_.remove();
+        if (thisPtr->channel_.fd() >= 0)
+        {
+            thisPtr->channel_.disableAll();
+            thisPtr->channel_.remove();
+        }
         thisPtr->connectionPtr_.reset();
         pro.set_value(1);
     });
@@ -522,11 +530,17 @@ void PgConnection::handleFatalError(bool clearAll, bool isAbortPipeline)
     {
         for (auto &cmd : batchCommandsForWaitingResults_)
         {
-            cmd->exceptionCallback_(exceptPtr);
+            if (cmd->exceptionCallback_)
+            {
+                cmd->exceptionCallback_(exceptPtr);
+            }
         }
         for (auto &cmd : batchSqlCommands_)
         {
-            cmd->exceptionCallback_(exceptPtr);
+            if (cmd->exceptionCallback_)
+            {
+                cmd->exceptionCallback_(exceptPtr);
+            }
         }
         batchCommandsForWaitingResults_.clear();
         batchSqlCommands_.clear();
@@ -536,13 +550,19 @@ void PgConnection::handleFatalError(bool clearAll, bool isAbortPipeline)
         if (!batchSqlCommands_.empty() &&
             !batchSqlCommands_.front()->preparingStatement_.empty())
         {
-            batchSqlCommands_.front()->exceptionCallback_(exceptPtr);
+            if (batchSqlCommands_.front()->exceptionCallback_)
+            {
+                batchSqlCommands_.front()->exceptionCallback_(exceptPtr);
+            }
             batchSqlCommands_.pop_front();
         }
         else if (!batchCommandsForWaitingResults_.empty())
         {
             auto &cmd = batchCommandsForWaitingResults_.front();
-            cmd->exceptionCallback_(exceptPtr);
+            if (cmd->exceptionCallback_)
+            {
+                cmd->exceptionCallback_(exceptPtr);
+            }
             batchCommandsForWaitingResults_.pop_front();
         }
         else

orm_lib/src/postgresql_impl/PgConnection.cc

@@ -65,11 +65,14 @@ PgConnection::PgConnection(trantor::EventLoop *loop,
                                   [](PGconn *conn) { PQfinish(conn); })),
       channel_(loop, PQsocket(connectionPtr_.get()))
 {
+    if (channel_.fd() < 0)
+    {
+        LOG_ERROR << "Failed to create Postgres connection";
+    }
 }
 
 void PgConnection::init()
 {
-    PQsetnonblocking(connectionPtr_.get(), 1);
     if (channel_.fd() < 0)
     {
         LOG_ERROR << "Connection with Postgres could not be established";
@@ -80,6 +83,8 @@ void PgConnection::init()
         }
         return;
     }
+
+    PQsetnonblocking(connectionPtr_.get(), 1);
     channel_.setReadCallback([this]() {
         if (status_ == ConnectStatus::Bad)
         {
@@ -142,8 +147,11 @@ void PgConnection::disconnect()
     auto thisPtr = shared_from_this();
     loop_->runInLoop([thisPtr, &pro]() {
         thisPtr->status_ = ConnectStatus::Bad;
-        thisPtr->channel_.disableAll();
-        thisPtr->channel_.remove();
+        if (thisPtr->channel_.fd() >= 0)
+        {
+            thisPtr->channel_.disableAll();
+            thisPtr->channel_.remove();
+        }
         thisPtr->connectionPtr_.reset();
         pro.set_value(1);
     });
@@ -400,7 +408,8 @@ void PgConnection::handleFatalError()
 {
     auto exceptPtr =
         std::make_exception_ptr(Failure(PQerrorMessage(connectionPtr_.get())));
-    exceptionCallback_(exceptPtr);
+    if (exceptionCallback_)
+        exceptionCallback_(exceptPtr);
     exceptionCallback_ = nullptr;
 }