bug upg: fix pg to userver sockaddr construction

Failed test before fix <https://nda.ya.ru/t/tLXropkU7YGdiX>

Tests: unittests, testing
commit_hash:e8c58b69ba609a2ace0d52af7975d01dd33adcd4

Author: kopturovdim

.mapping.json

@@ -3683,6 +3683,7 @@
   "postgresql/src/storages/postgres/detail/pg_impl_types.hpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pg_impl_types.hpp",
   "postgresql/src/storages/postgres/detail/pg_message_severity.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pg_message_severity.cpp",
   "postgresql/src/storages/postgres/detail/pg_message_severity.hpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pg_message_severity.hpp",
+  "postgresql/src/storages/postgres/detail/pg_version.hpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pg_version.hpp",
   "postgresql/src/storages/postgres/detail/pool.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pool.cpp",
   "postgresql/src/storages/postgres/detail/pool.hpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/pool.hpp",
   "postgresql/src/storages/postgres/detail/query_parameters.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/detail/query_parameters.cpp",
@@ -3743,6 +3744,7 @@
   "postgresql/src/storages/postgres/tests/arrays_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/arrays_pgtest.cpp",
   "postgresql/src/storages/postgres/tests/bitstring_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/bitstring_pgtest.cpp",
   "postgresql/src/storages/postgres/tests/bytea_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/bytea_pgtest.cpp",
+  "postgresql/src/storages/postgres/tests/cancel_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/cancel_pgtest.cpp",
   "postgresql/src/storages/postgres/tests/chrono_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/chrono_pgtest.cpp",
   "postgresql/src/storages/postgres/tests/cluster_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/cluster_pgtest.cpp",
   "postgresql/src/storages/postgres/tests/composite_types_pgtest.cpp":"taxi/uservices/userver/postgresql/src/storages/postgres/tests/composite_types_pgtest.cpp",

postgresql/src/storages/postgres/detail/cancel.cpp

@@ -1,102 +1,125 @@
 #include <storages/postgres/detail/cancel.hpp>
 
+// Internal structures of libpq to send cancel packet using async non blocking socket
+
 #include <array>
 #include <cstdint>
 
+#include <sys/socket.h>
+
 #include <userver/engine/io/sockaddr.hpp>
 #include <userver/engine/io/socket.hpp>
 #include <userver/storages/postgres/exceptions.hpp>
 #include <userver/tracing/span.hpp>
 
-#ifdef __clang__
-// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
-#define USERVER_IMPL_DISABLE_MSAN __attribute__((no_sanitize_memory))
+#include <storages/postgres/detail/pg_version.hpp>
+
+// NOLINTNEXTLINE(modernize-use-using)
+typedef struct {
+    struct sockaddr_storage addr;
+    socklen_t salen;
+} SockAddr;
+
+#if USERVER_LIBPQ_VERSION >= 180000
+
+struct pg_cancel {
+    SockAddr raddr;          /* Remote address */
+    int be_pid;              /* PID of to-be-canceled backend */
+    int pgtcp_user_timeout;  /* tcp user timeout */
+    int keepalives;          /* use TCP keepalives? */
+    int keepalives_idle;     /* time between TCP keepalives */
+    int keepalives_interval; /* time between TCP keepalive
+                              * retransmits */
+    int keepalives_count;    /* maximum number of TCP keepalive
+                              * retransmits */
+
+    /* Pre-constructed cancel request packet starts here */
+    int32_t cancel_pkt_len;                     /* in network byte order */
+    char cancel_req[/*FLEXIBLE_ARRAY_MEMBER*/]; /* CancelRequestPacket */
+};
+
 #else
-// NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
-#define USERVER_IMPL_DISABLE_MSAN
-#endif
 
 // NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
 #define PG_PROTOCOL_MAJOR(v) ((v) >> 16)
 // NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
 #define PG_PROTOCOL_MINOR(v) ((v) & 0x0000ffff)
 // NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
 #define PG_PROTOCOL(m, n) (((m) << 16) | (n))
-
 // NOLINTNEXTLINE(cppcoreguidelines-macro-usage)
 #define CANCEL_REQUEST_CODE PG_PROTOCOL(1234, 5678)
 
-using ACCEPT_TYPE_ARG3 = socklen_t;
-
-struct PgSockaddrStorage {
-    union {
-        struct sockaddr sa; /* get the system-dependent fields */
-        int64_t ss_align;   /* ensures struct is properly aligned */
-        char ss_pad[128];   /* ensures struct has desired size */
-    } ss_stuff;
-};
-
 // NOLINTNEXTLINE(modernize-use-using)
-typedef struct {
-    struct PgSockaddrStorage addr;
-    ACCEPT_TYPE_ARG3 salen;
-} SockAddr;
+typedef struct CancelRequestPacket {
+    /* Note that each field is stored in network byte order! */
+    uint32_t cancelRequestCode; /* code to identify a cancel request */
+    uint32_t backendPID;        /* PID of client's backend */
+    uint32_t cancelAuthCode;    /* secret key to authorize cancel */
+} CancelRequestPacket;
 
 struct pg_cancel {
     SockAddr raddr;
     int be_pid;
     int be_key;
 };
 
-// NOLINTNEXTLINE(modernize-use-using)
-typedef struct CancelRequestPacket {
-    /* Note that each field is stored in network byte order! */
-    uint32_t cancel_request_code; /* code to identify a cancel request */
-    uint32_t backend_pid;         /* PID of client's backend */
-    uint32_t cancel_auth_code;    /* secret key to authorize cancel */
-} CancelRequestPacket;
-
-struct CancelPacket {
-    uint32_t packetlen;
-    CancelRequestPacket cp;
-};
+#endif
 
 USERVER_NAMESPACE_BEGIN
 
 namespace storages::postgres::detail {
 
 namespace {
 
-// TODO(TAXICOMMON-11213) investigate why memory sanitizer complains.
-USERVER_IMPL_DISABLE_MSAN CancelPacket MakeCancelPacket(const PGcancel& cn) noexcept {
-    CancelPacket cp{};
-    cp.packetlen = sizeof(cp);
-    cp.cp.cancel_request_code = static_cast<uint32_t>(htonl(CANCEL_REQUEST_CODE));
-    cp.cp.backend_pid = htonl(cn.be_pid);
-    cp.cp.cancel_auth_code = htonl(cn.be_key);
-    return cp;
+void SendCancelPacket(engine::io::Socket& tmp_sock, engine::Deadline deadline, const PGcancel& cn) {
+#if USERVER_LIBPQ_VERSION >= 180000
+    // reference code: libpq/fe-cancel.c
+    // In libpq cancel packet is stored directly in pg_cancel struct
+    // starting from cancel_pkt_len field
+    // After this field pg_cancel contains flexible array with binary prepared packet
+    // NOTE: cancel_pkt_len in pg_cancel stored in network byte order
+    size_t cancel_pkt_len = ntohl(cn.cancel_pkt_len);
+    auto ret = tmp_sock.SendAll(reinterpret_cast<const char*>(&cn.cancel_pkt_len), cancel_pkt_len, deadline);
+    if (ret != cancel_pkt_len) {
+        throw CommandError(fmt::format(
+            "SendCancelPacket: Failed to call SendAll(), expected bytes to send {}, actual bytes sent {}",
+            cancel_pkt_len,
+            ret
+        ));
+    }
+#else
+    struct {
+        uint32_t packetlen;
+        CancelRequestPacket cp;
+    } crp;
+
+    crp.packetlen = htonl(sizeof(crp));
+    crp.cp.cancelRequestCode = htonl(CANCEL_REQUEST_CODE);
+    crp.cp.backendPID = htonl(cn.be_pid);
+    crp.cp.cancelAuthCode = htonl(cn.be_key);
+
+    auto ret = tmp_sock.SendAll(&crp, sizeof(crp), deadline);
+    if (ret != sizeof(crp)) {
+        throw CommandError("SendAll()");
+    }
+#endif
 }
 
 }  // namespace
 
-USERVER_IMPL_DISABLE_MSAN void Cancel(PGcancel* cn, engine::Deadline deadline) {
+void Cancel(PGcancel* cn, engine::Deadline deadline) {
     tracing::Span span{"pg_cancel"};
     if (!cn) {
         return;
     }
 
-    engine::io::Sockaddr addr;
-    memcpy(addr.Data(), &cn->raddr, std::min<size_t>(sizeof(cn->raddr), addr.Size()));
-
+    engine::io::Sockaddr addr(&cn->raddr.addr);
     engine::io::Socket tmp_sock(addr.Domain(), engine::io::SocketType::kStream);
 
     tmp_sock.Connect(addr, deadline);
+    LOG_DEBUG() << "Connected to " << addr.PrimaryAddressString() << " on port " << addr.Port();
 
-    CancelPacket cp = MakeCancelPacket(*cn);
-    auto ret = tmp_sock.SendAll(&cp, sizeof(cp), deadline);
-    if (ret != sizeof(cp)) {
-        throw CommandError("SendAll()");
-    }
+    SendCancelPacket(tmp_sock, deadline, *cn);
 
     /*
      * Comment from libpq's sources, fe-connect.c, inside internal_cancel():
@@ -107,8 +130,14 @@ USERVER_IMPL_DISABLE_MSAN void Cancel(PGcancel* cn, engine::Deadline deadline) {
      * one we thought we were canceling.  Note we don't actually expect this
      * read to obtain any data, we are just waiting for EOF to be signaled.
      */
-    std::array<char, 1024> c{};
-    [[maybe_unused]] auto ret2 = tmp_sock.RecvAll(c.data(), c.size(), deadline);
+    try {
+        std::array<char, 1024> c{};
+        [[maybe_unused]] auto ret = tmp_sock.RecvAll(c.data(), c.size(), deadline);
+    } catch (const engine::io::IoSystemError& ex) {
+        LOG_LIMITED_INFO() << "Got exception during RecvAll after pg_cancel: " << ex;
+        // Do not propagate exception here, because postmaster sends no response
+        // for cancel packets, just silently close socket
+    }
 }
 
 }  // namespace storages::postgres::detail

postgresql/src/storages/postgres/detail/pg_connection_wrapper.cpp

@@ -16,12 +16,6 @@ auto PQXsendQueryPrepared(PGconn* conn, const char* stmtName, int nParams, const
 }
 #endif
 
-#ifdef ARCADIA_ROOT
-#define USERVER_LIBPQ_VERSION PG_VERSION_NUM
-#else
-#include <userver_libpq_version.hpp>  // Y_IGNORE
-#endif
-
 #include <userver/concurrent/background_task_storage.hpp>
 #include <userver/crypto/openssl.hpp>
 #include <userver/engine/task/cancel.hpp>
@@ -32,6 +26,7 @@ auto PQXsendQueryPrepared(PGconn* conn, const char* stmtName, int nParams, const
 
 #include <storages/postgres/detail/cancel.hpp>
 #include <storages/postgres/detail/pg_message_severity.hpp>
+#include <storages/postgres/detail/pg_version.hpp>
 #include <storages/postgres/detail/tracing_tags.hpp>
 #include <userver/storages/postgres/dsn.hpp>
 #include <userver/storages/postgres/exceptions.hpp>

postgresql/src/storages/postgres/detail/pg_version.hpp

@@ -0,0 +1,9 @@
+#pragma once
+
+#include <pg_config.h>
+
+#ifdef ARCADIA_ROOT
+#define USERVER_LIBPQ_VERSION PG_VERSION_NUM
+#else
+#include <userver_libpq_version.hpp>  // Y_IGNORE
+#endif

postgresql/src/storages/postgres/tests/cancel_pgtest.cpp

@@ -0,0 +1,59 @@
+#include <storages/postgres/tests/util_pgtest.hpp>
+#include <userver/engine/async.hpp>
+#include <userver/engine/sleep.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace pg = storages::postgres;
+
+namespace {
+
+UTEST_P(PostgreConnection, Cancel) {
+    /*
+     * IF THIS TEST FAILS DURING UPDATE OF libpq VERSION IN CONTRIB,
+     * WRITE SOMEONE FROM TPS-60916
+     * We verify the logic of request cancellation in PostgreSQL.
+     * To do this, we start an asynchronous task that performs a long sleep on the PostgreSQL side (putting the
+     * connection into the busy state). Then we cancel this task. The connection should remain busy. Next, we issue a
+     * query cancellation request with a timeout shorter than the sleep duration on the PostgreSQL side (this is
+     * important!). If the query is canceled successfully, the connection will transition to the idle state. If the
+     * cancellation does not work, we will catch a timeout exception during DiscardInput. This happens specifically
+     * because the cancel timeout is shorter than the duration passed to pg_sleep.
+     */
+    auto& conn = GetConn();
+    CheckConnection(conn);
+
+    // Setup query timeout for 1 minute
+    const auto cmd_ctrl = pg::CommandControl{
+        /*network_timeout*/ std::chrono::seconds{60},
+        /*statement_timeout*/ std::chrono::seconds{60},
+    };
+
+    // sleep on postgres side for 1 minute
+    auto task = engine::CriticalAsyncNoSpan([&conn, &cmd_ctrl]() {
+        LOG_DEBUG() << "Enter pg_sleep";
+        conn->Execute("select pg_sleep(60)", /*query_params*/ {}, cmd_ctrl);
+        LOG_DEBUG() << "Return from pg_sleep";
+    });
+
+    // short sleep to ensure select in async task was started
+    engine::SleepFor(std::chrono::milliseconds{10});
+    ASSERT_FALSE(conn->IsIdle()) << "connection must be in busy state";
+
+    // cancel task
+    task.SyncCancel();
+    ASSERT_FALSE(conn->IsIdle()) << "connection must remain in busy state";
+
+    // try to cleanup with short timeout
+    // if cancel did not work, here we will get timeout exception
+    // because cancel timeout (5s) is less than pg_sleep(60s)
+    ASSERT_NO_THROW(conn->CancelAndCleanup(std::chrono::seconds{5}));
+
+    ASSERT_TRUE(conn->IsIdle()) << "connection must be in idle state";
+
+    ASSERT_NO_THROW(conn->Execute("select 1", /*query_params*/ {}, cmd_ctrl)) << "connection must be usable";
+}
+
+}  // namespace
+
+USERVER_NAMESPACE_END