Merge fc0c698 into 7a338da

Author: dcherednik

ydb/core/protos/config.proto

@@ -450,7 +450,7 @@ message TInterconnectConfig {
     optional bool SuppressConnectivityCheck = 39 [default = false];
     optional uint32 PreallocatedBufferSize = 40;
     optional uint32 NumPreallocatedBuffers = 41;
-    optional bool EnableExternalDataChannel = 42;
+    optional bool EnableExternalDataChannel = 42 [default = true];
     optional bool ValidateIncomingPeerViaDirectLookup = 44;
     optional uint32 SocketBacklogSize = 45; // SOMAXCONN if not set or zero
 

ydb/library/actors/interconnect/interconnect_common.h

@@ -49,7 +49,7 @@ namespace NActors {
         ui32 MaxSerializedEventSize = NActors::EventMaxByteSize;
         ui32 PreallocatedBufferSize = 8 << 10; // 8 KB
         ui32 NumPreallocatedBuffers = 16;
-        bool EnableExternalDataChannel = false;
+        bool EnableExternalDataChannel = true;
         bool ValidateIncomingPeerViaDirectLookup = false;
         ui32 SocketBacklogSize = 0; // SOMAXCONN if zero
         TDuration FirstErrorSleep = TDuration::MilliSeconds(10);

ydb/library/actors/interconnect/ut/lib/test_events.h

@@ -15,6 +15,10 @@ namespace NActors {
     struct TEvTest : TEventPB<TEvTest, NInterconnectTest::TEvTest, EvTest> {
         TEvTest() = default;
 
+        explicit TEvTest(ui64 sequenceNumber) {
+            Record.SetSequenceNumber(sequenceNumber);
+        }
+
         TEvTest(ui64 sequenceNumber, const TString& payload) {
             Record.SetSequenceNumber(sequenceNumber);
             Record.SetPayload(payload);

ydb/library/actors/interconnect/ut/protos/interconnect_test.proto

@@ -2,7 +2,11 @@ package NInterconnectTest;
 
 message TEvTest {
     optional uint64 SequenceNumber = 1;
-    optional bytes Payload = 2;
+    optional uint64 DataCrc = 2;
+    oneof Data {
+        bytes Payload = 3;
+        uint32 PayloadId = 4;
+    }
 }
 
 message TEvTestChan {

ydb/library/actors/interconnect/ut_fat/main.cpp

@@ -9,6 +9,7 @@
 
 #include <library/cpp/testing/unittest/tests_data.h>
 #include <library/cpp/testing/unittest/registar.h>
+#include <library/cpp/digest/crc32c/crc32c.h>
 
 #include <util/network/sock.h>
 #include <util/network/poller.h>
@@ -21,11 +22,13 @@ Y_UNIT_TEST_SUITE(InterconnectUnstableConnection) {
     class TSenderActor: public TSenderBaseActor {
         TDeque<ui64> InFly;
         ui16 SendFlags;
+        bool UseRope;
 
     public:
-        TSenderActor(const TActorId& recipientActorId, ui16 sendFlags)
+        TSenderActor(const TActorId& recipientActorId, ui16 sendFlags, bool useRope)
             : TSenderBaseActor(recipientActorId, 32)
             , SendFlags(sendFlags)
+            , UseRope(useRope)
         {
         }
 
@@ -36,8 +39,19 @@ Y_UNIT_TEST_SUITE(InterconnectUnstableConnection) {
         void SendMessage(const TActorContext& ctx) override {
             const ui32 flags = IEventHandle::MakeFlags(0, SendFlags);
             const ui64 cookie = SequenceNumber;
-            const TString payload('@', RandomNumber<size_t>(65536) + 4096);
-            ctx.Send(RecipientActorId, new TEvTest(SequenceNumber, payload), flags, cookie);
+
+            const TString payload(RandomNumber<size_t>(65536) + 4096, '@');
+
+            auto ev = new TEvTest(SequenceNumber);
+            ev->Record.SetDataCrc(Crc32c(payload.data(), payload.size()));
+
+            if (UseRope) {
+                ev->Record.SetPayloadId(ev->AddPayload(TRope(payload)));
+            } else {
+                ev->Record.SetPayload(payload);
+            }
+
+            ctx.Send(RecipientActorId, ev, flags, cookie);
             InFly.push_back(SequenceNumber);
             ++InFlySize;
             ++SequenceNumber;
@@ -90,6 +104,14 @@ Y_UNIT_TEST_SUITE(InterconnectUnstableConnection) {
             Y_ABORT_UNLESS(m.HasSequenceNumber());
             Y_ABORT_UNLESS(m.GetSequenceNumber() >= ReceivedCount, "got #%" PRIu64 " expected at least #%" PRIu64,
                      m.GetSequenceNumber(), ReceivedCount);
+            if (m.HasPayloadId()) {
+                auto rope = ev->Get()->GetPayload(m.GetPayloadId());
+                auto data = rope.GetContiguousSpan();
+                auto crc = Crc32c(data.data(), data.size());
+                Y_ABORT_UNLESS(m.GetDataCrc() == crc);
+            } else {
+                Y_ABORT_UNLESS(m.HasPayload());
+            }
             ++ReceivedCount;
             SenderNode->Send(ev->Sender, new TEvTestResponse(m.GetSequenceNumber()));
         }
@@ -109,7 +131,23 @@ Y_UNIT_TEST_SUITE(InterconnectUnstableConnection) {
 
         TReceiverActor* receiverActor = new TReceiverActor(testCluster.GetNode(1));
         const TActorId recipient = testCluster.RegisterActor(receiverActor, 2);
-        TSenderActor* senderActor = new TSenderActor(recipient, flags);
+        TSenderActor* senderActor = new TSenderActor(recipient, flags, false);
+        testCluster.RegisterActor(senderActor, 1);
+
+        NanoSleep(30ULL * 1000 * 1000 * 1000);
+    }
+
+    Y_UNIT_TEST(InterconnectTestWithProxyUnsureUndeliveredWithRopeXdc) {
+        ui32 numNodes = 2;
+        double bandWidth = 1000000;
+        ui16 flags = IEventHandle::FlagTrackDelivery | IEventHandle::FlagGenerateUnsureUndelivered;
+        TTestICCluster::TTrafficInterrupterSettings interrupterSettings{TDuration::Seconds(2), bandWidth, true};
+
+        TTestICCluster testCluster(numNodes, TChannelsConfig(), &interrupterSettings);
+
+        TReceiverActor* receiverActor = new TReceiverActor(testCluster.GetNode(1));
+        const TActorId recipient = testCluster.RegisterActor(receiverActor, 2);
+        TSenderActor* senderActor = new TSenderActor(recipient, flags, true);
         testCluster.RegisterActor(senderActor, 1);
 
         NanoSleep(30ULL * 1000 * 1000 * 1000);
@@ -125,7 +163,7 @@ Y_UNIT_TEST_SUITE(InterconnectUnstableConnection) {
 
         TReceiverActor* receiverActor = new TReceiverActor(testCluster.GetNode(1));
         const TActorId recipient = testCluster.RegisterActor(receiverActor, 2);
-        TSenderActor* senderActor = new TSenderActor(recipient, flags);
+        TSenderActor* senderActor = new TSenderActor(recipient, flags, false);
         testCluster.RegisterActor(senderActor, 1);
 
         NanoSleep(30ULL * 1000 * 1000 * 1000);

ydb/library/yql/providers/dq/config/config.proto

@@ -25,7 +25,7 @@ message TDqConfig {
         optional uint64 MessagePendingTimeoutMs = 15 [default = 5000];
         optional uint64 MessagePendingSize = 16 [default = 18446744073709551615];
         optional uint32 MaxSerializedEventSize = 17 [default = 67108000];
-        optional bool EnableExternalDataChannel = 27 [default = false];
+        optional bool EnableExternalDataChannel = 27 [default = true];
 
         // Scheduler
         optional uint64 ResolutionMicroseconds = 18 [default = 1024];