return unavailable if pipe is dead

return unavailable if pipe is dead in write session

4 files changed, 9 insertions, 4 deletions

diff --git a/ydb/public/api/protos/persqueue_error_codes_v1.proto b/ydb/public/api/protos/persqueue_error_codes_v1.proto
index c6658305c2..fab4aece53 100644
--- a/ydb/public/api/protos/persqueue_error_codes_v1.proto
+++ b/ydb/public/api/protos/persqueue_error_codes_v1.proto
@@ -39,5 +39,8 @@ enum ErrorCode {
 
     WRONG_PARTITION_NUMBER = 500021;
     PREFERRED_CLUSTER_MISMATCHED = 500022;
+
+    TABLET_PIPE_DISCONNECTED = 500023;
+
     ERROR = 500100;
 }
diff --git a/ydb/services/persqueue_v1/actors/persqueue_utils.cpp b/ydb/services/persqueue_v1/actors/persqueue_utils.cpp
index 1b286f6c3d..986801f5f6 100644
--- a/ydb/services/persqueue_v1/actors/persqueue_utils.cpp
+++ b/ydb/services/persqueue_v1/actors/persqueue_utils.cpp
@@ -94,6 +94,7 @@ Ydb::StatusIds::StatusCode ConvertPersQueueInternalCodeToStatus(const Ydb::PersQ
             return Ydb::StatusIds::SUCCESS;
         case INITIALIZING:
         case CLUSTER_DISABLED:
+        case TABLET_PIPE_DISCONNECTED:
             return Ydb::StatusIds::UNAVAILABLE;
         case PREFERRED_CLUSTER_MISMATCHED:
             return Ydb::StatusIds::ABORTED;
diff --git a/ydb/services/persqueue_v1/actors/persqueue_utils.h b/ydb/services/persqueue_v1/actors/persqueue_utils.h
index 1442536e6b..ea90f8b427 100644
--- a/ydb/services/persqueue_v1/actors/persqueue_utils.h
+++ b/ydb/services/persqueue_v1/actors/persqueue_utils.h
@@ -61,6 +61,7 @@ static inline bool InternalErrorCode(Ydb::PersQueue::ErrorCode::ErrorCode errorC
         case Ydb::PersQueue::ErrorCode::INITIALIZING:
         case Ydb::PersQueue::ErrorCode::OVERLOAD:
         case Ydb::PersQueue::ErrorCode::WRITE_ERROR_DISK_IS_FULL:
+        case Ydb::PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED:
             return true;
         default:
             return false;
diff --git a/ydb/services/persqueue_v1/actors/write_session_actor.ipp b/ydb/services/persqueue_v1/actors/write_session_actor.ipp
index 6887857823..429f329ba3 100644
--- a/ydb/services/persqueue_v1/actors/write_session_actor.ipp
+++ b/ydb/services/persqueue_v1/actors/write_session_actor.ipp
@@ -1247,7 +1247,7 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::T
 
 template<bool UseMigrationProtocol>
 void TWriteSessionActor<UseMigrationProtocol>::Handle(NPQ::TEvPartitionWriter::TEvDisconnected::TPtr&, const TActorContext& ctx) {
-    CloseSession("pipe to partition's tablet is dead", PersQueue::ErrorCode::ERROR, ctx);
+    CloseSession("pipe to partition's tablet is dead", PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED, ctx);
 }
 
 template<bool UseMigrationProtocol>
@@ -1255,7 +1255,7 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(TEvTabletPipe::TEvClientCo
     TEvTabletPipe::TEvClientConnected *msg = ev->Get();
     //TODO: add here retries for connecting to PQRB
     if (msg->Status != NKikimrProto::OK) {
-        CloseSession(TStringBuilder() << "pipe to tablet is dead " << msg->TabletId, PersQueue::ErrorCode::ERROR, ctx);
+        CloseSession(TStringBuilder() << "pipe to tablet is dead " << msg->TabletId, PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED, ctx);
         return;
     }
 }
@@ -1263,7 +1263,7 @@ void TWriteSessionActor<UseMigrationProtocol>::Handle(TEvTabletPipe::TEvClientCo
 template<bool UseMigrationProtocol>
 void TWriteSessionActor<UseMigrationProtocol>::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev, const TActorContext& ctx) {
     //TODO: add here retries for connecting to PQRB
-    CloseSession(TStringBuilder() << "pipe to tablet is dead " << ev->Get()->TabletId, PersQueue::ErrorCode::ERROR, ctx);
+    CloseSession(TStringBuilder() << "pipe to tablet is dead " << ev->Get()->TabletId, PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED, ctx);
 }
 
 template<bool UseMigrationProtocol>
@@ -1271,7 +1271,7 @@ void TWriteSessionActor<UseMigrationProtocol>::PrepareRequest(THolder<TEvWrite>&
     if (!PendingRequest) {
         PendingRequest = new TWriteRequestInfo(++NextRequestCookie);
     }
-   
+
     auto& request = PendingRequest->PartitionWriteRequest->Record;
     ui64 payloadSize = 0;