Backport #95466 to 25.10: Avoid possible crash for distributed queries in case of cancellation

robot-clickhouse · robot-clickhouse · commit 73e4e10d0cd5 · 2026-01-30T10:21:01.000Z
diff --git a/src/Client/Connection.cpp b/src/Client/Connection.cpp
@@ -85,6 +85,7 @@ namespace ErrorCodes
     extern const int SUPPORT_IS_DISABLED;
     extern const int BAD_ARGUMENTS;
     extern const int EMPTY_DATA_PASSED;
+    extern const int LOGICAL_ERROR;
 }
 
 Connection::~Connection()
@@ -1302,6 +1303,11 @@ Packet Connection::receivePacket()
 {
     try
     {
+        /// We are trying to send something to already disconnected connection,
+        /// this means that we continue using Connection after exception.
+        if (!in)
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Connection to {} is terminated", getDescription());
+
         Packet res;
 
         /// Have we already read packet type?
diff --git a/src/Processors/Sources/RemoteSource.cpp b/src/Processors/Sources/RemoteSource.cpp
@@ -249,6 +249,8 @@ void RemoteSource::onCancel() noexcept
 
 void RemoteSource::onUpdatePorts()
 {
+    if (isCancelled())
+        return;
     if (getPort().isFinished())
         query_executor->finish();
 }
diff --git a/src/QueryPipeline/RemoteQueryExecutor.cpp b/src/QueryPipeline/RemoteQueryExecutor.cpp
@@ -785,7 +785,7 @@ void RemoteQueryExecutor::finish()
       * - received an unknown packet from one replica;
       * then you do not need to read anything.
       */
-    if (!isQueryPending() || hasThrownException())
+    if (!isQueryPending() || hasThrownException() || was_cancelled)
         return;
 
     /// To make sure finish is only called once

Original file line number	Diff line number	Diff line change
`@@ -85,6 +85,7 @@ namespace ErrorCodes`
`85`	`85`	`extern const int SUPPORT_IS_DISABLED;`
`86`	`86`	`extern const int BAD_ARGUMENTS;`
`87`	`87`	`extern const int EMPTY_DATA_PASSED;`
	`88`	`+ extern const int LOGICAL_ERROR;`
`88`	`89`	`}`
`89`	`90`
`90`	`91`	`Connection::~Connection()`
`@@ -1302,6 +1303,11 @@ Packet Connection::receivePacket()`
`1302`	`1303`	`{`
`1303`	`1304`	`try`
`1304`	`1305`	`{`
	`1306`	`+ /// We are trying to send something to already disconnected connection,`
	`1307`	`+ /// this means that we continue using Connection after exception.`
	`1308`	`+ if (!in)`
	`1309`	`+ throw Exception(ErrorCodes::LOGICAL_ERROR, "Connection to {} is terminated", getDescription());`
	`1310`	`+`
`1305`	`1311`	`Packet res;`
`1306`	`1312`
`1307`	`1313`	`/// Have we already read packet type?`
Original file line number	Diff line number	Diff line change
`@@ -249,6 +249,8 @@ void RemoteSource::onCancel() noexcept`
`249`	`249`
`250`	`250`	`void RemoteSource::onUpdatePorts()`
`251`	`251`	`{`
	`252`	`+ if (isCancelled())`
	`253`	`+ return;`
`252`	`254`	`if (getPort().isFinished())`
`253`	`255`	`query_executor->finish();`
`254`	`256`	`}`