fix(cooperation): give tunnel streams collision-free ids

Both peers allocate ids for the streams they initiate, but the ids share
one namespace. Starting both counters at 1 made the barrier stream and a
reverse file transfer collide: the peer rejected the second OPEN as a
duplicate stream, so the transfer never established and stayed at 1%.

Pick a random non-zero id and skip ids already in use locally.

Log: give tunnel streams collision-free ids
PMS: TASK-396845
Influence: concurrent barrier and file-transfer streams no longer collide.

Author: re2zero

src/lib/common/manager/tunnelmanager.cpp

@@ -12,6 +12,7 @@
 #include <QJsonDocument>
 #include <QJsonObject>
 #include <QJsonParseError>
+#include <QRandomGenerator>
 #include <QTcpServer>
 #include <QTcpSocket>
 #include <QTimer>
@@ -195,6 +196,19 @@ quint16 TunnelManager::localPort(const QString &serviceId) const
     return it->port;
 }
 
+quint32 TunnelManager::nextStreamId()
+{
+    // Stream ids are chosen by the initiating side but live in a namespace
+    // shared by both peers. A counter starting at 1 on each side collides as
+    // soon as both initiate a stream (barrier plus a reverse file transfer),
+    // so pick a random id and skip ids already known locally.
+    for (;;) {
+        const quint32 id = QRandomGenerator::global()->generate();
+        if (id != 0 && !_streams.contains(id))
+            return id;
+    }
+}
+
 void TunnelManager::acceptLocalConnection(const QString &serviceId)
 {
     const auto forward = _forwards.constFind(serviceId);
@@ -207,9 +221,7 @@ void TunnelManager::acceptLocalConnection(const QString &serviceId)
         if (!socket)
             break;
 
-        quint32 id = _nextStreamId++;
-        if (_nextStreamId == 0)
-            _nextStreamId = 1;
+        const quint32 id = nextStreamId();
 
         const QString token = makeToken();
         Stream *stream = createStream(id, serviceId, token, false);

src/lib/common/manager/tunnelmanager.h

@@ -101,6 +101,7 @@ public slots:
     Stream *findStream(quint32 id) const;
     Stream *createStream(quint32 id, const QString &serviceId, const QString &token, bool dialer);
     void destroyStream(quint32 id, const QString &reason, bool notifyPeer);
+    quint32 nextStreamId();
 
     void acceptLocalConnection(const QString &serviceId);
     void setupSocket(Stream *stream);
@@ -135,7 +136,6 @@ public slots:
     QHash<quint32, Stream *> _streams;
     QString _peerIp;
     QString _peerRemote;
-    quint32 _nextStreamId { 1 };
 };
 
 #endif   // TUNNELMANAGER_H

tests/manager/tunnelmanager_test.cpp

@@ -17,6 +17,7 @@
 
 #include <QAbstractSocket>
 #include <QFile>
+#include <QElapsedTimer>
 #include <QHostAddress>
 #include <QRandomGenerator>
 #include <QSignalSpy>
@@ -26,9 +27,23 @@
 #include <QTest>
 
 #include <atomic>
+#include <functional>
 #include <memory>
 #include <mutex>
 
+namespace {
+// QTest's QTRY_* macros abort the test function without registering a failure
+// with GoogleTest, so a timed out condition would silently pass here.
+bool waitFor(const std::function<bool()> &condition, int timeoutMs = 3000)
+{
+    QElapsedTimer timer;
+    timer.start();
+    while (!condition() && timer.elapsed() < timeoutMs)
+        QTest::qWait(10);
+    return condition();
+}
+}
+
 TEST(TunnelManagerTest, EchoThroughReverseTunnel)
 {
     QTcpServer echo;
@@ -258,6 +273,75 @@ TEST(TunnelManagerTest, MultipleForwardsRunIndependently)
     a.stopAll();
 }
 
+TEST(TunnelManagerTest, BothSidesCanInitiateStreams)
+{
+    auto makeEcho = [](QTcpServer &server) {
+        EXPECT_TRUE(server.listen(QHostAddress::LocalHost, 0));
+        QObject::connect(&server, &QTcpServer::newConnection, &server, [&server] {
+            while (QTcpSocket *socket = server.nextPendingConnection()) {
+                QObject::connect(socket, &QTcpSocket::readyRead, socket, [socket] {
+                    socket->write(socket->readAll());
+                });
+                QObject::connect(socket, &QTcpSocket::disconnected, socket, &QObject::deleteLater);
+            }
+        });
+    };
+
+    QTcpServer echoA;
+    QTcpServer echoB;
+    makeEcho(echoA);
+    makeEcho(echoB);
+
+    TunnelManager a;
+    TunnelManager b;
+    a.setPeer(QStringLiteral("B"));
+    b.setPeer(QStringLiteral("A"));
+    // A hosts the barrier web service, B hosts a file transfer service.
+    a.registerService(QStringLiteral("barrier"), QStringLiteral("127.0.0.1"), echoA.serverPort());
+    b.registerService(QStringLiteral("transfer"), QStringLiteral("127.0.0.1"), echoB.serverPort());
+
+    a.setSendHandler([&b](const QString &, int mask, const QByteArray &payload) {
+        b.onTunnelMessage(mask, payload, QStringLiteral("A"));
+    });
+    b.setSendHandler([&a](const QString &, int mask, const QByteArray &payload) {
+        a.onTunnelMessage(mask, payload, QStringLiteral("B"));
+    });
+
+    // B opens the barrier stream first, as in an A -> B collaboration.
+    const quint16 barrierPort = b.startLocalForward(QStringLiteral("barrier"));
+    ASSERT_NE(barrierPort, 0);
+    QTcpSocket barrierClient;
+    barrierClient.connectToHost(QHostAddress::LocalHost, barrierPort);
+    ASSERT_TRUE(waitFor([&] {
+        return barrierClient.state() == QAbstractSocket::ConnectedState;
+    }));
+    barrierClient.write(QByteArrayLiteral("barrier"));
+    ASSERT_TRUE(waitFor([&] { return barrierClient.bytesAvailable() >= 7; }));
+    EXPECT_EQ(barrierClient.readAll(), QByteArrayLiteral("barrier"));
+
+    // While that stream is still open, A opens the transfer stream, as in a
+    // B -> A file transfer. Both sides used to allocate stream id 1 here and
+    // B rejected the second OPEN as a duplicate.
+    const quint16 transferPort = a.startLocalForward(QStringLiteral("transfer"));
+    ASSERT_NE(transferPort, 0);
+    QTcpSocket transferClient;
+    transferClient.connectToHost(QHostAddress::LocalHost, transferPort);
+    ASSERT_TRUE(waitFor([&] {
+        return transferClient.state() == QAbstractSocket::ConnectedState;
+    }));
+    transferClient.write(QByteArrayLiteral("transfer"));
+    ASSERT_TRUE(waitFor([&] { return transferClient.bytesAvailable() >= 8; }));
+    EXPECT_EQ(transferClient.readAll(), QByteArrayLiteral("transfer"));
+
+    // The first stream must survive the second one being opened.
+    barrierClient.write(QByteArrayLiteral("still-alive"));
+    ASSERT_TRUE(waitFor([&] { return barrierClient.bytesAvailable() >= 11; }));
+    EXPECT_EQ(barrierClient.readAll(), QByteArrayLiteral("still-alive"));
+
+    b.stopAll();
+    a.stopAll();
+}
+
 TEST(TunnelManagerTest, IncomingRequestIsNotSwallowedByPendingAsyncRequest)
 {
     auto service = std::make_shared<AsioService>();