The 'use-heap-after-free` error in the Topic SDK code (#15088)

Alek5andr-Kotov · Gazizonoki · commit b1cb7522e979 · 2025-03-11T19:27:09.000+03:00
diff --git a/src/client/topic/impl/topic_impl.h b/src/client/topic/impl/topic_impl.h
@@ -296,8 +296,8 @@ class TTopicClient::TImpl : public TClientImplCommon<TTopicClient::TImpl> {
     {
         auto request = MakeOperationRequest<Ydb::Topic::UpdateOffsetsInTransactionRequest>(settings);
 
-        request.mutable_tx()->set_id(TStringType{GetTxId(tx)});
-        request.mutable_tx()->set_session(TStringType{GetSessionId(tx)});
+        request.mutable_tx()->set_id(tx.TxId);
+        request.mutable_tx()->set_session(tx.SessionId);
 
         for (auto& t : topics) {
             auto* topic = request.mutable_topics()->Add();
diff --git a/src/client/topic/impl/transaction.h b/src/client/topic/impl/transaction.h
@@ -10,18 +10,21 @@ class TTransaction;
 
 namespace NYdb::inline V3::NTopic {
 
-using TTransactionId = std::pair<std::string, std::string>;
+struct TTransactionId {
+    std::string SessionId;
+    std::string TxId;
+};
 
 inline
-const std::string& GetSessionId(const TTransactionId& x)
+bool operator==(const TTransactionId& lhs, const TTransactionId& rhs)
 {
-    return x.first;
+    return (lhs.SessionId == rhs.SessionId) && (lhs.TxId == rhs.TxId);
 }
 
 inline
-const std::string& GetTxId(const TTransactionId& x)
+bool operator!=(const TTransactionId& lhs, const TTransactionId& rhs)
 {
-    return x.second;
+    return !(lhs == rhs);
 }
 
 TTransactionId MakeTransactionId(const NTable::TTransaction& tx);
@@ -30,3 +33,10 @@ TStatus MakeSessionExpiredError();
 TStatus MakeCommitTransactionSuccess();
 
 }
+
+template <>
+struct THash<NYdb::NTopic::TTransactionId> {
+    size_t operator()(const NYdb::NTopic::TTransactionId& v) const noexcept {
+        return CombineHashes(THash<std::string>()(v.SessionId), THash<std::string>()(v.TxId));
+    }
+};
diff --git a/src/client/topic/impl/write_session_impl.cpp b/src/client/topic/impl/write_session_impl.cpp
@@ -12,6 +12,12 @@
 #include <util/stream/buffer.h>
 #include <util/generic/guid.h>
 
+template <>
+void Out<NYdb::NTopic::TTransactionId>(IOutputStream& s, const NYdb::NTopic::TTransactionId& v)
+{
+    s << "{" << v.SessionId << ", " << v.TxId << "}";
+}
+
 namespace NYdb::inline V3::NTopic {
 
 const TDuration UPDATE_TOKEN_PERIOD = TDuration::Hours(1);
@@ -35,13 +41,22 @@ TTxIdOpt GetTransactionId(const Ydb::Topic::StreamWriteMessage_WriteRequest& req
     return TTxId(tx.session(), tx.id());
 }
 
-TTxIdOpt GetTransactionId(const NTable::TTransaction* tx)
+TTxIdOpt GetTransactionId(const std::optional<TTransactionId>& tx)
+{
+    if (!tx) {
+        return std::nullopt;
+    }
+
+    return TTxId(tx->SessionId, tx->TxId);
+}
+
+std::optional<TTransactionId> MakeTransactionId(const NTable::TTransaction* tx)
 {
     if (!tx) {
         return std::nullopt;
     }
 
-    return TTxId(tx->GetSession().GetId(), tx->GetId());
+    return TTransactionId{tx->GetSession().GetId(), tx->GetId()};
 }
 
 }
@@ -584,7 +599,7 @@ void TWriteSessionImpl::TrySignalAllAcksReceived(ui64 seqNo)
         ++txInfo->AckCount;
 
         LOG_LAZY(DbDriverState->Log, TLOG_DEBUG,
-                 LogPrefixImpl() << "OnAck: seqNo=" << seqNo << ", txId=" << GetTxId(txId) << ", WriteCount=" << txInfo->WriteCount << ", AckCount=" << txInfo->AckCount);
+                 LogPrefixImpl() << "OnAck: seqNo=" << seqNo << ", txId=" << txId << ", WriteCount=" << txInfo->WriteCount << ", AckCount=" << txInfo->AckCount);
 
         if (txInfo->CommitCalled && (txInfo->WriteCount == txInfo->AckCount)) {
             txInfo->AllAcksReceived.SetValue(MakeCommitTransactionSuccess());
@@ -631,15 +646,15 @@ void TWriteSessionImpl::WriteInternal(TContinuationToken&&, TWriteMessage&& mess
                 ++txInfo->WriteCount;
 
                 LOG_LAZY(DbDriverState->Log, TLOG_DEBUG,
-                         LogPrefixImpl() << "OnWrite: seqNo=" << seqNo << ", txId=" << GetTxId(txId) << ", WriteCount=" << txInfo->WriteCount << ", AckCount=" << txInfo->AckCount);
+                         LogPrefixImpl() << "OnWrite: seqNo=" << seqNo << ", txId=" << txId << ", WriteCount=" << txInfo->WriteCount << ", AckCount=" << txInfo->AckCount);
             }
             WrittenInTx[seqNo] = txId;
         }
 
         CurrentBatch.Add(
                 seqNo, createdAtValue, message.Data, message.Codec, message.OriginalSize,
                 message.MessageMeta_,
-                message.GetTxPtr()
+                MakeTransactionId(message.GetTxPtr())
         );
 
         FlushWriteIfRequiredImpl();
@@ -1412,10 +1427,10 @@ size_t TWriteSessionImpl::WriteBatchImpl() {
             if (!currMessage.MessageMeta.empty()) {
                 OriginalMessagesToSend.emplace(id, createTs, datum.size(),
                                                std::move(currMessage.MessageMeta),
-                                               currMessage.Tx);
+                                               std::move(currMessage.Tx));
             } else {
                 OriginalMessagesToSend.emplace(id, createTs, datum.size(),
-                                               currMessage.Tx);
+                                               std::move(currMessage.Tx));
             }
         }
         block.Data = std::move(CurrentBatch.Data);
@@ -1523,8 +1538,8 @@ void TWriteSessionImpl::SendImpl() {
                 auto* msgData = writeRequest->add_messages();
 
                 if (message.Tx) {
-                    writeRequest->mutable_tx()->set_id(TStringType{message.Tx->GetId()});
-                    writeRequest->mutable_tx()->set_session(TStringType{message.Tx->GetSession().GetId()});
+                    writeRequest->mutable_tx()->set_id(message.Tx->TxId);
+                    writeRequest->mutable_tx()->set_session(message.Tx->SessionId);
                 }
 
                 msgData->set_seq_no(GetSeqNoImpl(message.Id));
diff --git a/src/client/topic/impl/write_session_impl.h b/src/client/topic/impl/write_session_impl.h
@@ -164,18 +164,18 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
         std::optional<ECodec> Codec;
         ui32 OriginalSize; // only for coded messages
         std::vector<std::pair<std::string, std::string>> MessageMeta;
-        const NTable::TTransaction* Tx;
+        std::optional<TTransactionId> Tx;
 
         TMessage(uint64_t id, const TInstant& createdAt, std::string_view data, std::optional<ECodec> codec = {},
                  ui32 originalSize = 0, const std::vector<std::pair<std::string, std::string>>& messageMeta = {},
-                 const NTable::TTransaction* tx = nullptr)
+                 std::optional<TTransactionId>&& tx = {})
             : Id(id)
             , CreatedAt(createdAt)
             , DataRef(data)
             , Codec(codec)
             , OriginalSize(originalSize)
             , MessageMeta(messageMeta)
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
     };
 
@@ -189,11 +189,11 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
 
         void Add(uint64_t id, const TInstant& createdAt, std::string_view data, std::optional<ECodec> codec, ui32 originalSize,
                  const std::vector<std::pair<std::string, std::string>>& messageMeta,
-                 const NTable::TTransaction* tx) {
+                 std::optional<TTransactionId>&& tx) {
             if (StartedAt == TInstant::Zero())
                 StartedAt = TInstant::Now();
             CurrentSize += codec ? originalSize : data.size();
-            Messages.emplace_back(id, createdAt, data, codec, originalSize, messageMeta, tx);
+            Messages.emplace_back(id, createdAt, data, codec, originalSize, messageMeta, std::move(tx));
             Acquired = false;
         }
 
@@ -263,24 +263,24 @@ class TWriteSessionImpl : public TContinuationTokenIssuer,
         TInstant CreatedAt;
         size_t Size;
         std::vector<std::pair<std::string, std::string>> MessageMeta;
-        const NTable::TTransaction* Tx;
+        std::optional<TTransactionId> Tx;
 
         TOriginalMessage(const uint64_t id, const TInstant createdAt, const size_t size,
-                         const NTable::TTransaction* tx)
+                         std::optional<TTransactionId>&& tx)
             : Id(id)
             , CreatedAt(createdAt)
             , Size(size)
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
 
         TOriginalMessage(const uint64_t id, const TInstant createdAt, const size_t size,
                          std::vector<std::pair<std::string, std::string>>&& messageMeta,
-                         const NTable::TTransaction* tx)
+                         std::optional<TTransactionId>&& tx)
             : Id(id)
             , CreatedAt(createdAt)
             , Size(size)
             , MessageMeta(std::move(messageMeta))
-            , Tx(tx)
+            , Tx(std::move(tx))
         {}
     };
 

Original file line number	Diff line number	Diff line change
`@@ -296,8 +296,8 @@ class TTopicClient::TImpl : public TClientImplCommon<TTopicClient::TImpl> {`
`296`	`296`	`{`
`297`	`297`	`auto request = MakeOperationRequest<Ydb::Topic::UpdateOffsetsInTransactionRequest>(settings);`
`298`	`298`
`299`		`- request.mutable_tx()->set_id(TStringType{GetTxId(tx)});`
`300`		`- request.mutable_tx()->set_session(TStringType{GetSessionId(tx)});`
	`299`	`+ request.mutable_tx()->set_id(tx.TxId);`
	`300`	`+ request.mutable_tx()->set_session(tx.SessionId);`
`301`	`301`
`302`	`302`	`for (auto& t : topics) {`
`303`	`303`	`auto* topic = request.mutable_topics()->Add();`