pdns/dnsdistdist/test-dnsdisttcp_cc.cc

   1 /*
   2  * This file is part of PowerDNS or dnsdist.
   3  * Copyright -- PowerDNS.COM B.V. and its contributors
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of version 2 of the GNU General Public License as
   7  * published by the Free Software Foundation.
   8  *
   9  * In addition, for the avoidance of any doubt, permission is granted to
  10  * link this program with OpenSSL and to (re)distribute the binaries
  11  * produced as the result of such linking.
  12  *
  13  * This program is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16  * GNU General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * along with this program; if not, write to the Free Software
  20  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  21  */
  22 #define BOOST_TEST_DYN_LINK
  23 #define BOOST_TEST_NO_MAIN
  24
  25 #include <boost/test/unit_test.hpp>
  26
  27 #include "dnswriter.hh"
  28 #include "dnsdist.hh"
  29 #include "dnsdist-proxy-protocol.hh"
  30 #include "dnsdist-rings.hh"
  31 #include "dnsdist-tcp-downstream.hh"
  32 #include "dnsdist-tcp-upstream.hh"
  33
  34 struct DNSDistStats g_stats;
  35 GlobalStateHolder<NetmaskGroup> g_ACL;
  36 GlobalStateHolder<vector<DNSDistRuleAction> > g_ruleactions;
  37 GlobalStateHolder<vector<DNSDistResponseRuleAction> > g_respruleactions;
  38 GlobalStateHolder<vector<DNSDistResponseRuleAction> > g_cachehitrespruleactions;
  39 GlobalStateHolder<vector<DNSDistResponseRuleAction> > g_selfansweredrespruleactions;
  40 GlobalStateHolder<servers_t> g_dstates;
  41
  42 QueryCount g_qcount;
  43
  44 const bool TCPIOHandler::s_disableConnectForUnitTests = true;
  45
  46 bool checkDNSCryptQuery(const ClientState& cs, PacketBuffer& query, std::unique_ptr<DNSCryptQuery>& dnsCryptQuery, time_t now, bool tcp)
  47 {
  48   return false;
  49 }
  50
  51 bool checkQueryHeaders(const struct dnsheader* dh, ClientState&)
  52 {
  53   return true;
  54 }
  55
  56 uint64_t uptimeOfProcess(const std::string& str)
  57 {
  58   return 0;
  59 }
  60
  61 void handleResponseSent(const IDState& ids, double udiff, const ComboAddress& client, const ComboAddress& backend, unsigned int size, const dnsheader& cleartextDH, dnsdist::Protocol protocol)
  62 {
  63 }
  64
  65 static std::function<ProcessQueryResult(DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend)> s_processQuery;
  66
  67 ProcessQueryResult processQuery(DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend)
  68 {
  69   if (s_processQuery) {
  70     return s_processQuery(dq, cs, holders, selectedBackend);
  71   }
  72
  73   return ProcessQueryResult::Drop;
  74 }
  75
  76 bool responseContentMatches(const PacketBuffer& response, const DNSName& qname, const uint16_t qtype, const uint16_t qclass, const std::shared_ptr<DownstreamState>& remote, unsigned int& qnameWireLength)
  77 {
  78   return true;
  79 }
  80
  81 static std::function<bool(PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted)> s_processResponse;
  82
  83 bool processResponse(PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted, bool receivedOverUDP)
  84 {
  85   if (s_processResponse) {
  86     return s_processResponse(response, localRespRuleActions, dr, muted);
  87   }
  88
  89   return false;
  90 }
  91
  92 BOOST_AUTO_TEST_SUITE(test_dnsdisttcp_cc)
  93
  94 struct ExpectedStep
  95 {
  96 public:
  97   enum class ExpectedRequest { handshakeClient, readFromClient, writeToClient, closeClient, connectToBackend, readFromBackend, writeToBackend, closeBackend };
  98
  99   ExpectedStep(ExpectedRequest r, IOState n, size_t b = 0, std::function<void(int descriptor)> fn = nullptr): cb(fn), request(r), nextState(n), bytes(b)
 100   {
 101   }
 102
 103   std::function<void(int descriptor)> cb{nullptr};
 104   ExpectedRequest request;
 105   IOState nextState;
 106   size_t bytes{0};
 107 };
 108
 109 static std::deque<ExpectedStep> s_steps;
 110
 111 static PacketBuffer s_readBuffer;
 112 static PacketBuffer s_writeBuffer;
 113 static PacketBuffer s_backendReadBuffer;
 114 static PacketBuffer s_backendWriteBuffer;
 115
 116 std::ostream& operator<<(std::ostream &os, const ExpectedStep::ExpectedRequest d);
 117
 118 std::ostream& operator<<(std::ostream &os, const ExpectedStep::ExpectedRequest d)
 119 {
 120   static const std::vector<std::string> requests = { "handshake with client", "read from client", "write to client", "close connection to client", "connect to the backend", "read from the backend", "write to the backend", "close connection to backend" };
 121   os<<requests.at(static_cast<size_t>(d));
 122   return os;
 123 }
 124
 125 class MockupTLSConnection : public TLSConnection
 126 {
 127 public:
 128   MockupTLSConnection(int descriptor, bool client = false): d_descriptor(descriptor), d_client(client)
 129   {
 130   }
 131
 132   ~MockupTLSConnection() { }
 133
 134   IOState tryHandshake() override
 135   {
 136     auto step = getStep();
 137     BOOST_REQUIRE_EQUAL(step.request, ExpectedStep::ExpectedRequest::handshakeClient);
 138
 139     return step.nextState;
 140   }
 141
 142   IOState tryWrite(const PacketBuffer& buffer, size_t& pos, size_t toWrite) override
 143   {
 144     auto step = getStep();
 145     BOOST_REQUIRE_EQUAL(step.request, !d_client ? ExpectedStep::ExpectedRequest::writeToClient : ExpectedStep::ExpectedRequest::writeToBackend);
 146
 147     if (step.bytes == 0) {
 148       if (step.nextState == IOState::NeedWrite) {
 149         return step.nextState;
 150       }
 151       throw std::runtime_error("Remote host closed the connection");
 152     }
 153
 154     toWrite -= pos;
 155     BOOST_REQUIRE_GE(buffer.size(), pos + toWrite);
 156
 157     if (step.bytes < toWrite) {
 158       toWrite = step.bytes;
 159     }
 160
 161     auto& externalBuffer = d_client ? s_backendWriteBuffer : s_writeBuffer;
 162     externalBuffer.insert(externalBuffer.end(), buffer.begin() + pos, buffer.begin() + pos + toWrite);
 163     pos += toWrite;
 164
 165     return step.nextState;
 166   }
 167
 168   IOState tryRead(PacketBuffer& buffer, size_t& pos, size_t toRead, bool allowIncomplete=false) override
 169   {
 170     auto step = getStep();
 171     BOOST_REQUIRE_EQUAL(step.request, !d_client ? ExpectedStep::ExpectedRequest::readFromClient : ExpectedStep::ExpectedRequest::readFromBackend);
 172
 173     if (step.bytes == 0) {
 174       if (step.nextState == IOState::NeedRead) {
 175         return step.nextState;
 176       }
 177       throw std::runtime_error("Remote host closed the connection");
 178     }
 179
 180     auto& externalBuffer = d_client ? s_backendReadBuffer : s_readBuffer;
 181     toRead -= pos;
 182
 183     if (step.bytes < toRead) {
 184       toRead = step.bytes;
 185     }
 186
 187     BOOST_REQUIRE_GE(buffer.size(), toRead);
 188     BOOST_REQUIRE_GE(externalBuffer.size(), toRead);
 189
 190     std::copy(externalBuffer.begin(), externalBuffer.begin() + toRead, buffer.begin() + pos);
 191     pos += toRead;
 192     externalBuffer.erase(externalBuffer.begin(), externalBuffer.begin() + toRead);
 193
 194     return step.nextState;
 195   }
 196
 197   IOState tryConnect(bool fastOpen, const ComboAddress& remote) override
 198   {
 199     auto step = getStep();
 200     BOOST_REQUIRE_EQUAL(step.request, ExpectedStep::ExpectedRequest::connectToBackend);
 201
 202     return step.nextState;
 203   }
 204
 205   void close() override
 206   {
 207     auto step = getStep();
 208     BOOST_REQUIRE_EQUAL(step.request, !d_client ? ExpectedStep::ExpectedRequest::closeClient : ExpectedStep::ExpectedRequest::closeBackend);
 209   }
 210
 211   bool hasBufferedData() const override
 212   {
 213     return false;
 214   }
 215
 216   bool isUsable() const override
 217   {
 218     return true;
 219   }
 220
 221   std::string getServerNameIndication() const override
 222   {
 223     return "";
 224   }
 225
 226   std::vector<uint8_t> getNextProtocol() const override
 227   {
 228     return std::vector<uint8_t>();
 229   }
 230
 231   LibsslTLSVersion getTLSVersion() const override
 232   {
 233     return LibsslTLSVersion::TLS13;
 234   }
 235
 236   bool hasSessionBeenResumed() const override
 237   {
 238     return false;
 239   }
 240
 241   std::vector<std::unique_ptr<TLSSession>> getSessions() override
 242   {
 243     return {};
 244   }
 245
 246   std::vector<int> getAsyncFDs() override
 247   {
 248     return {};
 249   }
 250
 251   void setSession(std::unique_ptr<TLSSession>& session) override
 252   {
 253   }
 254
 255   /* unused in that context, don't bother */
 256   void doHandshake() override
 257   {
 258   }
 259
 260   void connect(bool fastOpen, const ComboAddress& remote, const struct timeval& timeout) override
 261   {
 262   }
 263
 264   size_t read(void* buffer, size_t bufferSize, const struct timeval&readTimeout, const struct timeval& totalTimeout={0,0}, bool allowIncomplete=false) override
 265   {
 266     return 0;
 267   }
 268
 269   size_t write(const void* buffer, size_t bufferSize, const struct timeval& writeTimeout) override
 270   {
 271     return 0;
 272   }
 273 private:
 274   ExpectedStep getStep() const
 275   {
 276     BOOST_REQUIRE(!s_steps.empty());
 277     auto step = s_steps.front();
 278     s_steps.pop_front();
 279
 280     if (step.cb) {
 281       step.cb(d_descriptor);
 282     }
 283
 284     return step;
 285   }
 286
 287   const int d_descriptor;
 288   bool d_client{false};
 289 };
 290
 291 class MockupTLSCtx : public TLSCtx
 292 {
 293 public:
 294   ~MockupTLSCtx()
 295   {
 296   }
 297
 298   std::unique_ptr<TLSConnection> getConnection(int socket, const struct timeval& timeout, time_t now) override
 299   {
 300     return std::make_unique<MockupTLSConnection>(socket);
 301   }
 302
 303   std::unique_ptr<TLSConnection> getClientConnection(const std::string& host, bool hostIsAddr, int socket, const struct timeval& timeout) override
 304   {
 305     return std::make_unique<MockupTLSConnection>(socket, true);
 306   }
 307
 308   void rotateTicketsKey(time_t now) override
 309   {
 310   }
 311
 312   size_t getTicketsKeysCount() override
 313   {
 314     return 0;
 315   }
 316
 317   std::string getName() const override
 318   {
 319     return "Mockup TLS";
 320   }
 321 };
 322
 323 class MockupFDMultiplexer : public FDMultiplexer
 324 {
 325 public:
 326   MockupFDMultiplexer()
 327   {
 328   }
 329
 330   ~MockupFDMultiplexer()
 331   {
 332   }
 333
 334   int run(struct timeval* tv, int timeout=500) override
 335   {
 336     int ret = 0;
 337
 338     gettimeofday(tv, nullptr); // MANDATORY
 339
 340     /* 'ready' might be altered by a callback while we are iterating */
 341     const auto readyFDs = ready;
 342     for (const auto fd : readyFDs) {
 343       {
 344         const auto& it = d_readCallbacks.find(fd);
 345
 346         if (it != d_readCallbacks.end()) {
 347           it->d_callback(it->d_fd, it->d_parameter);
 348           continue; // so we don't refind ourselves as writable!
 349         }
 350       }
 351
 352       {
 353         const auto& it = d_writeCallbacks.find(fd);
 354
 355         if (it != d_writeCallbacks.end()) {
 356           it->d_callback(it->d_fd, it->d_parameter);
 357         }
 358       }
 359     }
 360
 361     return ret;
 362   }
 363
 364   void getAvailableFDs(std::vector<int>& fds, int timeout) override
 365   {
 366   }
 367
 368   void addFD(int fd, FDMultiplexer::EventKind kind) override
 369   {
 370   }
 371
 372   void removeFD(int fd, FDMultiplexer::EventKind) override
 373   {
 374   }
 375
 376   string getName() const override
 377   {
 378     return "mockup";
 379   }
 380
 381   void setReady(int fd)
 382   {
 383     ready.insert(fd);
 384   }
 385
 386   void setNotReady(int fd)
 387   {
 388     ready.erase(fd);
 389   }
 390
 391 private:
 392   std::set<int> ready;
 393 };
 394
 395 static bool isIPv6Supported()
 396 {
 397   try {
 398     ComboAddress addr("[2001:db8:53::1]:53");
 399     auto socket = std::make_unique<Socket>(addr.sin4.sin_family, SOCK_STREAM, 0);
 400     socket->setNonBlocking();
 401     int res = SConnectWithTimeout(socket->getHandle(), addr, timeval{0, 0});
 402     if (res == 0 || res == EINPROGRESS) {
 403       return true;
 404     }
 405     return false;
 406   }
 407   catch (const std::exception& e) {
 408     return false;
 409   }
 410 }
 411
 412 static ComboAddress getBackendAddress(const std::string& lastDigit, uint16_t port)
 413 {
 414   static const bool useV6 = isIPv6Supported();
 415
 416   if (useV6) {
 417     return ComboAddress("2001:db8:53::" + lastDigit, port);
 418   }
 419
 420   return ComboAddress("192.0.2." + lastDigit, port);
 421 }
 422
 423 static void appendPayloadEditingID(PacketBuffer& buffer, const PacketBuffer& payload, uint16_t newID)
 424 {
 425   PacketBuffer newPayload(payload);
 426   dnsheader dh;
 427   memcpy(&dh, &newPayload.at(sizeof(uint16_t)), sizeof(dh));
 428   dh.id = htons(newID);
 429   memcpy(&newPayload.at(sizeof(uint16_t)), &dh, sizeof(dh));
 430   buffer.insert(buffer.end(), newPayload.begin(), newPayload.end());
 431 }
 432
 433 static void prependPayloadEditingID(PacketBuffer& buffer, const PacketBuffer& payload, uint16_t newID)
 434 {
 435   PacketBuffer newPayload(payload);
 436   dnsheader dh;
 437   memcpy(&dh, &newPayload.at(sizeof(uint16_t)), sizeof(dh));
 438   dh.id = htons(newID);
 439   memcpy(&newPayload.at(sizeof(uint16_t)), &dh, sizeof(dh));
 440   buffer.insert(buffer.begin(), newPayload.begin(), newPayload.end());
 441 }
 442
 443 static void testInit(const std::string& name, TCPClientThreadData& threadData)
 444 {
 445 #ifdef DEBUGLOG_ENABLED
 446   cerr<<name<<endl;
 447 #else
 448   (void) name;
 449 #endif
 450
 451   s_steps.clear();
 452   s_readBuffer.clear();
 453   s_writeBuffer.clear();
 454   s_backendReadBuffer.clear();
 455   s_backendWriteBuffer.clear();
 456
 457   g_proxyProtocolACL.clear();
 458   g_verbose = false;
 459   IncomingTCPConnectionState::clearAllDownstreamConnections();
 460
 461   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
 462 }
 463
 464 #define TEST_INIT(str) testInit(str, threadData)
 465
 466 BOOST_AUTO_TEST_CASE(test_IncomingConnection_SelfAnswered)
 467 {
 468   auto local = getBackendAddress("1", 80);
 469   ClientState localCS(local, true, false, false, "", {});
 470   auto tlsCtx = std::make_shared<MockupTLSCtx>();
 471   localCS.tlsFrontend = std::make_shared<TLSFrontend>(tlsCtx);
 472
 473   TCPClientThreadData threadData;
 474   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
 475
 476   struct timeval now;
 477   gettimeofday(&now, nullptr);
 478
 479   PacketBuffer query;
 480   GenericDNSPacketWriter<PacketBuffer> pwQ(query, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
 481   pwQ.getHeader()->rd = 1;
 482
 483   uint16_t querySize = static_cast<uint16_t>(query.size());
 484   const uint8_t sizeBytes[] = { static_cast<uint8_t>(querySize / 256), static_cast<uint8_t>(querySize % 256) };
 485   query.insert(query.begin(), sizeBytes, sizeBytes + 2);
 486
 487   {
 488     /* drop right away */
 489     TEST_INIT("=> drop right away");
 490     s_readBuffer = query;
 491
 492     s_steps = {
 493       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 494       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 495       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 496       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 497     };
 498     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 499       return ProcessQueryResult::Drop;
 500     };
 501
 502     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 503     IncomingTCPConnectionState::handleIO(state, now);
 504     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 505   }
 506
 507   {
 508     /* self-generated REFUSED, client closes connection right away */
 509     TEST_INIT("=> self-gen");
 510     s_readBuffer = query;
 511
 512     s_steps = {
 513       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 514       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 515       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 516       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 65537 },
 517       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
 518       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 519     };
 520     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 521       // Would be nicer to actually turn it into a response
 522       return ProcessQueryResult::SendAnswer;
 523     };
 524
 525     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 526     IncomingTCPConnectionState::handleIO(state, now);
 527     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size());
 528     BOOST_CHECK(s_writeBuffer == query);
 529   }
 530
 531   {
 532     TEST_INIT("=> shorts");
 533     /* need write then read during handshake,
 534        short read on the size, then on the query itself,
 535        self-generated REFUSED, short write on the response,
 536        client closes connection right away */
 537     s_readBuffer = query;
 538
 539     s_steps = {
 540       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::NeedWrite },
 541       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::NeedRead },
 542       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 543       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 1 },
 544       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 1 },
 545       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, query.size() - 3 },
 546       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 1 },
 547       { ExpectedStep::ExpectedRequest::writeToClient, IOState::NeedWrite, query.size() - 1},
 548       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 1 },
 549       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
 550       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 551     };
 552     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 553       // Would be nicer to actually turn it into a response
 554       return ProcessQueryResult::SendAnswer;
 555     };
 556
 557     /* mark the incoming FD as always ready */
 558     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
 559
 560     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 561     IncomingTCPConnectionState::handleIO(state, now);
 562     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
 563       threadData.mplexer->run(&now);
 564     }
 565     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size());
 566     BOOST_CHECK(s_writeBuffer == query);
 567   }
 568
 569   {
 570     TEST_INIT("=> exception while handling the query");
 571     /* Exception raised while handling the query */
 572     s_readBuffer = query;
 573
 574     s_steps = {
 575       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 576       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 577       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 578       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 579     };
 580     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 581       throw std::runtime_error("Something unexpected happened");
 582     };
 583
 584     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 585     IncomingTCPConnectionState::handleIO(state, now);
 586     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 587   }
 588
 589   {
 590 #if 1
 591     TEST_INIT("=> 10k self-generated pipelined on the same connection");
 592
 593     /* 10k self-generated REFUSED pipelined on the same connection */
 594     size_t count = 10000;
 595
 596     s_steps = { { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done } };
 597
 598     for (size_t idx = 0; idx < count; idx++) {
 599       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
 600       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 });
 601       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 });
 602       s_steps.push_back({ ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() + 2 });
 603     };
 604     s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 });
 605     s_steps.push_back({ ExpectedStep::ExpectedRequest::closeClient, IOState::Done });
 606
 607     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 608       // Would be nicer to actually turn it into a response
 609       return ProcessQueryResult::SendAnswer;
 610     };
 611
 612     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 613     IncomingTCPConnectionState::handleIO(state, now);
 614     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size() * count);
 615 #endif
 616   }
 617
 618   {
 619     TEST_INIT("=> timeout while reading the query");
 620     /* timeout while reading the query */
 621     s_readBuffer = query;
 622
 623     s_steps = {
 624       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 625       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 626       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, query.size() - 2 - 2 },
 627       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 628     };
 629     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 630       /* should not be reached */
 631       BOOST_CHECK(false);
 632       return ProcessQueryResult::SendAnswer;
 633     };
 634
 635     /* mark the incoming FD as NOT ready */
 636     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
 637
 638     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 639     IncomingTCPConnectionState::handleIO(state, now);
 640     BOOST_CHECK_EQUAL(threadData.mplexer->run(&now), 0);
 641     struct timeval later = now;
 642     later.tv_sec += g_tcpRecvTimeout + 1;
 643     auto expiredReadConns = threadData.mplexer->getTimeouts(later, false);
 644     for (const auto& cbData : expiredReadConns) {
 645       BOOST_CHECK_EQUAL(cbData.first, state->d_handler.getDescriptor());
 646       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
 647         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
 648         BOOST_CHECK_EQUAL(cbData.first, cbState->d_handler.getDescriptor());
 649         cbState->handleTimeout(cbState, false);
 650       }
 651     }
 652     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 653   }
 654
 655   {
 656     TEST_INIT("=> timeout while writing the response");
 657     /* timeout while writing the response */
 658     s_readBuffer = query;
 659
 660     s_steps = {
 661       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 662       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 663       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 664       { ExpectedStep::ExpectedRequest::writeToClient, IOState::NeedWrite, 1 },
 665       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 666     };
 667     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 668       return ProcessQueryResult::SendAnswer;
 669     };
 670
 671     /* mark the incoming FD as NOT ready */
 672     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
 673
 674     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 675     IncomingTCPConnectionState::handleIO(state, now);
 676     BOOST_CHECK_EQUAL(threadData.mplexer->run(&now), 0);
 677     struct timeval later = now;
 678     later.tv_sec += g_tcpRecvTimeout + 1;
 679     auto expiredWriteConns = threadData.mplexer->getTimeouts(later, true);
 680     for (const auto& cbData : expiredWriteConns) {
 681       BOOST_CHECK_EQUAL(cbData.first, state->d_handler.getDescriptor());
 682       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
 683         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
 684         BOOST_CHECK_EQUAL(cbData.first, cbState->d_handler.getDescriptor());
 685         cbState->handleTimeout(cbState, false);
 686       }
 687     }
 688     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 1U);
 689   }
 690
 691   {
 692     TEST_INIT("=> Client closes the connection while writing the response (self-answered)");
 693
 694     s_readBuffer = query;
 695
 696     s_steps = {
 697       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 698       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 699       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 700       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 0 },
 701       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 702     };
 703     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 704       return ProcessQueryResult::SendAnswer;
 705     };
 706
 707     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 708     IncomingTCPConnectionState::handleIO(state, now);
 709     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 710   }
 711 }
 712
 713 BOOST_AUTO_TEST_CASE(test_IncomingConnectionWithProxyProtocol_SelfAnswered)
 714 {
 715   auto local = getBackendAddress("1", 80);
 716   ClientState localCS(local, true, false, false, "", {});
 717   auto tlsCtx = std::make_shared<MockupTLSCtx>();
 718   localCS.tlsFrontend = std::make_shared<TLSFrontend>(tlsCtx);
 719
 720   TCPClientThreadData threadData;
 721   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
 722
 723   struct timeval now;
 724   gettimeofday(&now, nullptr);
 725
 726   PacketBuffer query;
 727   GenericDNSPacketWriter<PacketBuffer> pwQ(query, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
 728   pwQ.getHeader()->rd = 1;
 729
 730   uint16_t querySize = static_cast<uint16_t>(query.size());
 731   const uint8_t sizeBytes[] = { static_cast<uint8_t>(querySize / 256), static_cast<uint8_t>(querySize % 256) };
 732   query.insert(query.begin(), sizeBytes, sizeBytes + 2);
 733
 734   {
 735     TEST_INIT("=> reading PP");
 736
 737     g_proxyProtocolACL.addMask("0.0.0.0/0");
 738     g_proxyProtocolACL.addMask("::0/0");
 739
 740     auto proxyPayload = makeProxyHeader(true, ComboAddress("192.0.2.1"), ComboAddress("192.0.2.2"), {});
 741     BOOST_REQUIRE_GT(proxyPayload.size(), s_proxyProtocolMinimumHeaderSize);
 742     s_readBuffer = query;
 743     // preprend the proxy protocol payload
 744     s_readBuffer.insert(s_readBuffer.begin(), proxyPayload.begin(), proxyPayload.end());
 745     // append a second query
 746     s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
 747
 748     s_steps = {
 749       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 750       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, s_proxyProtocolMinimumHeaderSize },
 751       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, proxyPayload.size() - s_proxyProtocolMinimumHeaderSize },
 752       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 753       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 754       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 65537 },
 755       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 756       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 757       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 65537 },
 758       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
 759       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 760     };
 761     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 762       return ProcessQueryResult::SendAnswer;
 763     };
 764
 765     /* mark the incoming FD as NOT ready */
 766     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
 767
 768     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 769     IncomingTCPConnectionState::handleIO(state, now);
 770     BOOST_CHECK_EQUAL(threadData.mplexer->run(&now), 0);
 771     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size() * 2U);
 772   }
 773
 774   {
 775     TEST_INIT("=> Invalid PP");
 776
 777     g_proxyProtocolACL.addMask("0.0.0.0/0");
 778     g_proxyProtocolACL.addMask("::0/0");
 779     auto proxyPayload = std::vector<uint8_t>(s_proxyProtocolMinimumHeaderSize);
 780     std::fill(proxyPayload.begin(), proxyPayload.end(), 0);
 781
 782     s_readBuffer = query;
 783     // preprend the proxy protocol payload
 784     s_readBuffer.insert(s_readBuffer.begin(), proxyPayload.begin(), proxyPayload.end());
 785
 786     s_steps = {
 787       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 788       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, s_proxyProtocolMinimumHeaderSize },
 789       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 790     };
 791     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 792       return ProcessQueryResult::SendAnswer;
 793     };
 794
 795     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 796     IncomingTCPConnectionState::handleIO(state, now);
 797
 798     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 799   }
 800
 801   {
 802     TEST_INIT("=> timeout while reading PP");
 803
 804     g_proxyProtocolACL.addMask("0.0.0.0/0");
 805     g_proxyProtocolACL.addMask("::0/0");
 806     auto proxyPayload = makeProxyHeader(true, ComboAddress("192.0.2.1"), ComboAddress("192.0.2.2"), {});
 807     BOOST_REQUIRE_GT(proxyPayload.size(), s_proxyProtocolMinimumHeaderSize);
 808     s_readBuffer = query;
 809     // preprend the proxy protocol payload
 810     s_readBuffer.insert(s_readBuffer.begin(), proxyPayload.begin(), proxyPayload.end());
 811
 812     s_steps = {
 813       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 814       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, s_proxyProtocolMinimumHeaderSize },
 815       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, proxyPayload.size() - s_proxyProtocolMinimumHeaderSize - 1},
 816       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 817     };
 818     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 819       return ProcessQueryResult::SendAnswer;
 820     };
 821
 822     /* mark the incoming FD as NOT ready */
 823     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
 824
 825     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 826     IncomingTCPConnectionState::handleIO(state, now);
 827     BOOST_CHECK_EQUAL(threadData.mplexer->run(&now), 0);
 828     struct timeval later = now;
 829     later.tv_sec += g_tcpRecvTimeout + 1;
 830     auto expiredReadConns = threadData.mplexer->getTimeouts(later, false);
 831     for (const auto& cbData : expiredReadConns) {
 832       BOOST_CHECK_EQUAL(cbData.first, state->d_handler.getDescriptor());
 833       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
 834         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
 835         BOOST_CHECK_EQUAL(cbData.first, cbState->d_handler.getDescriptor());
 836         cbState->handleTimeout(cbState, false);
 837       }
 838     }
 839     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 840   }
 841 }
 842
 843 BOOST_AUTO_TEST_CASE(test_IncomingConnection_BackendNoOOOR)
 844 {
 845   auto local = getBackendAddress("1", 80);
 846   ClientState localCS(local, true, false, false, "", {});
 847   auto tlsCtx = std::make_shared<MockupTLSCtx>();
 848   localCS.tlsFrontend = std::make_shared<TLSFrontend>(tlsCtx);
 849
 850   TCPClientThreadData threadData;
 851   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
 852
 853   struct timeval now;
 854   gettimeofday(&now, nullptr);
 855
 856   PacketBuffer query;
 857   GenericDNSPacketWriter<PacketBuffer> pwQ(query, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
 858   pwQ.getHeader()->rd = 1;
 859   pwQ.getHeader()->id = 0;
 860
 861   auto shortQuery = query;
 862   shortQuery.resize(sizeof(dnsheader) - 1);
 863   uint16_t shortQuerySize = static_cast<uint16_t>(shortQuery.size());
 864   const uint8_t shortSizeBytes[] = { static_cast<uint8_t>(shortQuerySize / 256), static_cast<uint8_t>(shortQuerySize % 256) };
 865   shortQuery.insert(shortQuery.begin(), shortSizeBytes, shortSizeBytes + 2);
 866
 867   uint16_t querySize = static_cast<uint16_t>(query.size());
 868   const uint8_t sizeBytes[] = { static_cast<uint8_t>(querySize / 256), static_cast<uint8_t>(querySize % 256) };
 869   query.insert(query.begin(), sizeBytes, sizeBytes + 2);
 870
 871   auto backend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
 872   backend->d_tlsCtx = tlsCtx;
 873
 874   {
 875     /* pass to backend, backend answers right away, client closes the connection */
 876     TEST_INIT("=> Query to backend, backend answers right away");
 877     s_readBuffer = query;
 878
 879     s_backendReadBuffer = query;
 880
 881     s_steps = {
 882       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 883       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 884       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 885       /* opening a connection to the backend */
 886       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
 887       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
 888       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
 889       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
 890       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() },
 891       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
 892       /* closing client connection */
 893       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 894       /* closing a connection to the backend */
 895       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
 896     };
 897     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 898       selectedBackend = backend;
 899       return ProcessQueryResult::PassToBackend;
 900     };
 901     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
 902       return true;
 903     };
 904
 905     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 906     IncomingTCPConnectionState::handleIO(state, now);
 907     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size());
 908     BOOST_CHECK(s_writeBuffer == query);
 909     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
 910     BOOST_CHECK(s_backendWriteBuffer == query);
 911     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
 912     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
 913     IncomingTCPConnectionState::clearAllDownstreamConnections();
 914   }
 915
 916   {
 917     /* pass to backend, backend answers right away, exception while handling the response */
 918     TEST_INIT("=> Exception while handling the response sent by the backend");
 919     s_readBuffer = query;
 920
 921     s_backendReadBuffer = query;
 922
 923     s_steps = {
 924       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 925       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 926       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 927       /* opening a connection to the backend */
 928       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
 929       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
 930       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
 931       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
 932       /* closing client connection */
 933       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 934       /* closing a connection to the backend */
 935       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
 936     };
 937     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 938       selectedBackend = backend;
 939       return ProcessQueryResult::PassToBackend;
 940     };
 941     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
 942       throw std::runtime_error("Unexpected error while processing the response");
 943     };
 944
 945     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 946     IncomingTCPConnectionState::handleIO(state, now);
 947     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 948     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
 949     BOOST_CHECK(s_backendWriteBuffer == query);
 950     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
 951     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
 952     IncomingTCPConnectionState::clearAllDownstreamConnections();
 953   }
 954
 955   {
 956     /* pass to backend, backend answers right away, processResponse() fails */
 957     TEST_INIT("=> Response processing fails ");
 958     s_readBuffer = query;
 959
 960     s_backendReadBuffer = query;
 961
 962     s_steps = {
 963       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
 964       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
 965       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
 966       /* opening a connection to the backend */
 967       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
 968       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
 969       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
 970       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
 971       /* closing client connection */
 972       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
 973       /* closing a connection to the backend */
 974       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
 975     };
 976     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
 977       selectedBackend = backend;
 978       return ProcessQueryResult::PassToBackend;
 979     };
 980     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
 981       return false;
 982     };
 983
 984     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
 985     IncomingTCPConnectionState::handleIO(state, now);
 986     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
 987     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
 988     BOOST_CHECK(s_backendWriteBuffer == query);
 989     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
 990     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
 991     IncomingTCPConnectionState::clearAllDownstreamConnections();
 992   }
 993
 994   {
 995     /* pass to backend, backend answers right away, ID matching fails */
 996     TEST_INIT("=> ID matching fails ");
 997     s_readBuffer = query;
 998
 999     auto responsePacket = query;
1000     /* mess with the transaction ID */
1001     responsePacket.at(3) ^= 42;
1002
1003     s_backendReadBuffer = responsePacket;
1004
1005     s_steps = {
1006       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1007       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1008       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1009       /* opening a connection to the backend */
1010       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1011       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1012       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1013       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1014       /* closing client connection */
1015       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1016       /* closing a connection to the backend */
1017       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1018     };
1019     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1020       selectedBackend = backend;
1021       return ProcessQueryResult::PassToBackend;
1022     };
1023     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1024       return true;
1025     };
1026
1027     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1028     IncomingTCPConnectionState::handleIO(state, now);
1029     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1030     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
1031     BOOST_CHECK(s_backendWriteBuffer == query);
1032     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1033     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1034     IncomingTCPConnectionState::clearAllDownstreamConnections();
1035   }
1036
1037   {
1038     TEST_INIT("=> Short (too short) query");
1039     s_readBuffer = shortQuery;
1040
1041     s_steps = {
1042       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1043       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1044       /* closing client connection */
1045       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1046     };
1047     s_processQuery = [](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1048       return ProcessQueryResult::SendAnswer;
1049     };
1050     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1051       return true;
1052     };
1053
1054     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1055     IncomingTCPConnectionState::handleIO(state, now);
1056     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1057     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
1058     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1059
1060     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1061     IncomingTCPConnectionState::clearAllDownstreamConnections();
1062   }
1063
1064   {
1065     TEST_INIT("=> Short (too short) response from backend");
1066     s_readBuffer = query;
1067
1068     s_backendReadBuffer = shortQuery;
1069
1070     s_steps = {
1071       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1072       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1073       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() },
1074       /* opening a connection to the backend */
1075       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1076       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1077       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1078       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1079       /* closing client connection */
1080       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1081       /* closing backend connection */
1082       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1083     };
1084     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1085       selectedBackend = backend;
1086       return ProcessQueryResult::PassToBackend;
1087     };
1088     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1089       return true;
1090     };
1091
1092     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1093     IncomingTCPConnectionState::handleIO(state, now);
1094     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1095     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
1096     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1097
1098     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1099     IncomingTCPConnectionState::clearAllDownstreamConnections();
1100   }
1101
1102   {
1103     /* connect in progress, short write to the backend, short read from the backend, client */
1104     TEST_INIT("=> Short read and write to backend");
1105     s_readBuffer = query;
1106     // append a second query
1107     appendPayloadEditingID(s_readBuffer, query, 1);
1108
1109     s_backendReadBuffer = query;
1110     // append a second query
1111     appendPayloadEditingID(s_backendReadBuffer, query, 1);
1112
1113     s_steps = {
1114       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1115       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1116       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1117       /* connect to backend */
1118       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::NeedWrite, 0, [&threadData](int desc) {
1119           /* set the outgoing descriptor (backend connection) as ready */
1120           dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
1121         }
1122       },
1123       /* send query */
1124       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::NeedWrite, 1 },
1125       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() - 1 },
1126       /* read response */
1127       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 1 },
1128       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 1 },
1129       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, query.size() - 3 },
1130       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 1 },
1131       /* write response to client */
1132       { ExpectedStep::ExpectedRequest::writeToClient, IOState::NeedWrite, query.size() - 1 },
1133       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 1 },
1134       /* read second query */
1135       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1136       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1137       /* write second query to backend */
1138       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1139       /* read second response */
1140       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1141       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1142       /* write second response */
1143       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() },
1144       /* read from client */
1145       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
1146       /* close connection to client */
1147       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1148       /* close connection to the backend, eventually */
1149       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1150     };
1151
1152     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1153       selectedBackend = backend;
1154       return ProcessQueryResult::PassToBackend;
1155     };
1156     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1157       return true;
1158     };
1159
1160     /* set the incoming descriptor as ready! */
1161     dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
1162     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1163     IncomingTCPConnectionState::handleIO(state, now);
1164     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
1165       threadData.mplexer->run(&now);
1166     }
1167     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size() * 2U);
1168     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size() * 2U);
1169     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1170
1171     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1172     IncomingTCPConnectionState::clearAllDownstreamConnections();
1173   }
1174
1175   {
1176     /* connection refused by the backend */
1177     TEST_INIT("=> Connection refused by the backend ");
1178     s_readBuffer = query;
1179
1180     s_steps = {
1181       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1182       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1183       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1184       /* opening a connection to the backend (5 tries by default) */
1185       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1186           throw NetworkError("Connection refused by the backend");
1187         }
1188       },
1189       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1190       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1191           throw NetworkError("Connection refused by the backend");
1192         }
1193       },
1194       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1195       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1196           throw NetworkError("Connection refused by the backend");
1197         }
1198       },
1199       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1200       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1201           throw NetworkError("Connection refused by the backend");
1202         }
1203       },
1204       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1205       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1206           throw NetworkError("Connection refused by the backend");
1207         }
1208       },
1209       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1210       /* closing client connection */
1211       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1212     };
1213
1214     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1215
1216       selectedBackend = backend;
1217       return ProcessQueryResult::PassToBackend;
1218     };
1219     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1220       return true;
1221     };
1222
1223     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1224     IncomingTCPConnectionState::handleIO(state, now);
1225     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1226     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
1227     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1228
1229     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1230     IncomingTCPConnectionState::clearAllDownstreamConnections();
1231   }
1232
1233   {
1234     /* timeout from the backend (write) */
1235     TEST_INIT("=> Timeout from the backend (write) ");
1236     s_readBuffer = query;
1237
1238     s_steps = {
1239       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1240       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1241       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1242       /* opening a connection to the backend (retrying 5 times) */
1243       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1244       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::NeedWrite },
1245       /* closing client connection */
1246       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1247       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1248     };
1249
1250     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1251
1252       selectedBackend = backend;
1253       return ProcessQueryResult::PassToBackend;
1254     };
1255     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1256       return true;
1257     };
1258
1259     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1260     IncomingTCPConnectionState::handleIO(state, now);
1261     struct timeval later = now;
1262     later.tv_sec += backend->d_config.tcpSendTimeout + 1;
1263     auto expiredWriteConns = threadData.mplexer->getTimeouts(later, true);
1264     BOOST_CHECK_EQUAL(expiredWriteConns.size(), 1U);
1265     for (const auto& cbData : expiredWriteConns) {
1266       if (cbData.second.type() == typeid(std::shared_ptr<TCPConnectionToBackend>)) {
1267         auto cbState = boost::any_cast<std::shared_ptr<TCPConnectionToBackend>>(cbData.second);
1268         cbState->handleTimeout(later, true);
1269       }
1270     }
1271     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1272     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
1273     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1274
1275     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1276     IncomingTCPConnectionState::clearAllDownstreamConnections();
1277   }
1278
1279   {
1280     /* timeout from the backend (read) */
1281     TEST_INIT("=> Timeout from the backend (read) ");
1282     s_readBuffer = query;
1283
1284     s_steps = {
1285       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1286       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1287       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1288       /* opening a connection to the backend */
1289       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1290       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1291       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1292       /* closing client connection */
1293       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1294       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1295     };
1296
1297     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1298       selectedBackend = backend;
1299       return ProcessQueryResult::PassToBackend;
1300     };
1301     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1302       return true;
1303     };
1304
1305     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1306     IncomingTCPConnectionState::handleIO(state, now);
1307     struct timeval later = now;
1308     later.tv_sec += backend->d_config.tcpRecvTimeout + 1;
1309     auto expiredConns = threadData.mplexer->getTimeouts(later, false);
1310     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
1311     for (const auto& cbData : expiredConns) {
1312       if (cbData.second.type() == typeid(std::shared_ptr<TCPConnectionToBackend>)) {
1313         auto cbState = boost::any_cast<std::shared_ptr<TCPConnectionToBackend>>(cbData.second);
1314         cbState->handleTimeout(later, false);
1315       }
1316     }
1317     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1318     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
1319     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1320
1321     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1322     IncomingTCPConnectionState::clearAllDownstreamConnections();
1323   }
1324
1325   {
1326     /* connection closed from the backend (write) */
1327     TEST_INIT("=> Connection closed from the backend (write) ");
1328     s_readBuffer = query;
1329
1330     s_steps = {
1331       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1332       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1333       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1334       /* opening a connection to the backend, connection closed on first write (5 attempts) */
1335       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1336       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1337       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1338       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1339       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1340       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1341       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1342       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1343       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1344       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1345       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1346       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1347       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1348       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1349       /* closing client connection */
1350       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1351       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1352     };
1353
1354     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1355       selectedBackend = backend;
1356       return ProcessQueryResult::PassToBackend;
1357     };
1358     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1359       return true;
1360     };
1361
1362     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1363     IncomingTCPConnectionState::handleIO(state, now);
1364     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1365     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
1366     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1367
1368     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1369     IncomingTCPConnectionState::clearAllDownstreamConnections();
1370   }
1371
1372   {
1373     /* connection closed from the backend (write) 4 times then succeeds */
1374     TEST_INIT("=> Connection closed from the backend (write) 4 times then succeeds");
1375     s_readBuffer = query;
1376     s_backendReadBuffer = query;
1377
1378     s_steps = {
1379       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1380       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1381       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1382       /* opening a connection to the backend, connection closed on first write (5 attempts) */
1383       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1384       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1385       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1386       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1387       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1388       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1389       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1390       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1391       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1392       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1393       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1394       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1395       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1396       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1397       /* reading the response */
1398       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1399       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1400       /* send the response to the client */
1401       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() },
1402       /* client closes the connection */
1403       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
1404       /* closing client connection */
1405       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1406       /* then eventually the backend one */
1407       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1408     };
1409
1410     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1411       selectedBackend = backend;
1412       return ProcessQueryResult::PassToBackend;
1413     };
1414     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1415       return true;
1416     };
1417
1418     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1419     IncomingTCPConnectionState::handleIO(state, now);
1420     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size());
1421     BOOST_CHECK(s_writeBuffer == query);
1422     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
1423     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1424
1425     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1426     IncomingTCPConnectionState::clearAllDownstreamConnections();
1427   }
1428
1429   {
1430     TEST_INIT("=> connection closed by the backend on write, then refused");
1431     s_readBuffer = query;
1432
1433     s_steps = {
1434       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1435       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1436       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1437       /* opening a connection to the backend, connection closed on first write */
1438       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1439       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
1440       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1441       /* and now reconnection fails (1) */
1442       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1443           throw NetworkError("Connection refused by the backend");
1444         }
1445       },
1446       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1447       /* 2 */
1448       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1449           throw NetworkError("Connection refused by the backend");
1450         }
1451       },
1452       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1453       /* 3 */
1454       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1455           throw NetworkError("Connection refused by the backend");
1456         }
1457       },
1458       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1459       /* 4 */
1460       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int descriptor) {
1461           throw NetworkError("Connection refused by the backend");
1462         }
1463       },
1464       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1465       /* closing client connection */
1466       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1467     };
1468
1469     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1470       selectedBackend = backend;
1471       return ProcessQueryResult::PassToBackend;
1472     };
1473     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1474       return true;
1475     };
1476
1477     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1478     IncomingTCPConnectionState::handleIO(state, now);
1479     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1480     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
1481     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1482
1483     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1484     IncomingTCPConnectionState::clearAllDownstreamConnections();
1485   }
1486
1487   {
1488     /* connection closed from the backend (read) */
1489     TEST_INIT("=> Connection closed from the backend (read) ");
1490     s_readBuffer = query;
1491
1492     s_steps = {
1493       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1494       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1495       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1496       /* opening a connection to the backend, connection closed on read, 5 attempts, last one succeeds */
1497       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1498       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1499       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1500       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1501       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1502       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1503       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1504       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1505       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1506       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1507       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1508       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1509       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1510       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1511       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1512       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1513       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1514       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1515       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1516       /* closing client connection */
1517       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1518       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1519     };
1520
1521     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1522       selectedBackend = backend;
1523       return ProcessQueryResult::PassToBackend;
1524     };
1525     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1526       return true;
1527     };
1528
1529     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1530     IncomingTCPConnectionState::handleIO(state, now);
1531     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1532     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size() * backend->d_config.d_retries);
1533     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1534
1535     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1536     IncomingTCPConnectionState::clearAllDownstreamConnections();
1537   }
1538
1539   {
1540     /* connection closed from the backend (read) 4 times then succeeds */
1541     TEST_INIT("=> Connection closed from the backend (read) 4 times then succeeds ");
1542     s_readBuffer = query;
1543     s_backendReadBuffer = query;
1544
1545     s_steps = {
1546       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1547       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1548       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1549       /* opening a connection to the backend, connection closed on read, 5 attempts, last one succeeds */
1550       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1551       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1552       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1553       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1554       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1555       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1556       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1557       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1558       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1559       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1560       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1561       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1562       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1563       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1564       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
1565       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1566       /* this time it works */
1567       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1568       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1569       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1570       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1571       /* sending the response to the client */
1572       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() },
1573       /* client closes the connection */
1574       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
1575       /* closing client connection */
1576       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1577       /* the eventually the backend one */
1578       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1579     };
1580
1581     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1582       selectedBackend = backend;
1583       return ProcessQueryResult::PassToBackend;
1584     };
1585     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1586       return true;
1587     };
1588
1589     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1590     IncomingTCPConnectionState::handleIO(state, now);
1591     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size());
1592     BOOST_CHECK(s_writeBuffer == query);
1593     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size() * backend->d_config.d_retries);
1594     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1595
1596     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1597     IncomingTCPConnectionState::clearAllDownstreamConnections();
1598   }
1599
1600   {
1601     TEST_INIT("=> Connection closed by the client when trying to send the response received from the backend");
1602     s_readBuffer = query;
1603     s_backendReadBuffer = query;
1604
1605     s_steps = {
1606       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1607       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1608       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1609       /* opening a connection to the backend */
1610       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1611       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() },
1612       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1613       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1614       /* sending the response to the client, the connection has been closed */
1615       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 0 },
1616       /* closing client connection */
1617       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1618       /* and eventually the backend one */
1619       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1620     };
1621
1622     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1623       selectedBackend = backend;
1624       return ProcessQueryResult::PassToBackend;
1625     };
1626     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1627       return true;
1628     };
1629
1630     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1631     IncomingTCPConnectionState::handleIO(state, now);
1632     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
1633     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), query.size());
1634     BOOST_CHECK(s_backendWriteBuffer == query);
1635     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1636
1637     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1638     IncomingTCPConnectionState::clearAllDownstreamConnections();
1639   }
1640
1641   {
1642 #if 1
1643     /* 101 queries on the same connection, check that the maximum number of queries kicks in */
1644     TEST_INIT("=> 101 queries on the same connection");
1645
1646     g_maxTCPQueriesPerConn = 100;
1647
1648     size_t count = 101;
1649
1650     s_readBuffer = query;
1651
1652     for (size_t idx = 0; idx < count; idx++) {
1653       appendPayloadEditingID(s_readBuffer, query, idx);
1654       appendPayloadEditingID(s_backendReadBuffer, query, idx);
1655     }
1656
1657     s_steps = { { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1658                 { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1659                 { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 },
1660                 /* opening a connection to the backend */
1661                 { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1662                 { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() + 2 },
1663                 { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
1664                 { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 },
1665                 { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() + 2 }
1666     };
1667
1668     for (size_t idx = 0; idx < count - 1; idx++) {
1669       /* read a new query */
1670       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 });
1671       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, query.size() - 2 });
1672       /* pass it to the backend */
1673       s_steps.push_back({ ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, query.size() + 2 });
1674       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 });
1675       s_steps.push_back({ ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, query.size() - 2 });
1676       /* send the response */
1677       s_steps.push_back({ ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, query.size() + 2 });
1678     };
1679     /* close the connection with the client */
1680     s_steps.push_back({ ExpectedStep::ExpectedRequest::closeClient, IOState::Done });
1681     /* eventually with the backend as well */
1682     s_steps.push_back({ ExpectedStep::ExpectedRequest::closeBackend, IOState::Done });
1683
1684     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1685       selectedBackend = backend;
1686       return ProcessQueryResult::PassToBackend;
1687     };
1688     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1689       return true;
1690     };
1691
1692     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1693     IncomingTCPConnectionState::handleIO(state, now);
1694     BOOST_CHECK_EQUAL(s_writeBuffer.size(), query.size() * count);
1695     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1696
1697     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1698     IncomingTCPConnectionState::clearAllDownstreamConnections();
1699
1700     g_maxTCPQueriesPerConn = 0;
1701 #endif
1702   }
1703
1704 }
1705
1706 BOOST_AUTO_TEST_CASE(test_IncomingConnectionOOOR_BackendOOOR)
1707 {
1708   auto local = getBackendAddress("1", 80);
1709   ClientState localCS(local, true, false, false, "", {});
1710   /* enable out-of-order on the front side */
1711   localCS.d_maxInFlightQueriesPerConn = 65536;
1712
1713   auto tlsCtx = std::make_shared<MockupTLSCtx>();
1714   localCS.tlsFrontend = std::make_shared<TLSFrontend>(tlsCtx);
1715
1716   ConnectionInfo connInfo(&localCS);
1717   connInfo.remote = getBackendAddress("84", 4242);
1718
1719   auto backend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
1720   backend->d_tlsCtx = tlsCtx;
1721   /* enable out-of-order on the backend side as well */
1722   backend->d_config.d_maxInFlightQueriesPerConn = 65536;
1723   /* shorter than the client one */
1724   backend->d_config.tcpRecvTimeout = 1;
1725
1726   TCPClientThreadData threadData;
1727   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
1728
1729   struct timeval now;
1730   gettimeofday(&now, nullptr);
1731
1732   std::vector<PacketBuffer> queries(5);
1733   std::vector<PacketBuffer> responses(5);
1734
1735   size_t counter = 0;
1736   size_t totalQueriesSize = 0;
1737   for (auto& query : queries) {
1738     GenericDNSPacketWriter<PacketBuffer> pwQ(query, DNSName("powerdns" + std::to_string(counter) + ".com."), QType::A, QClass::IN, 0);
1739     pwQ.getHeader()->rd = 1;
1740     pwQ.getHeader()->id = htons(counter);
1741     uint16_t querySize = static_cast<uint16_t>(query.size());
1742     const uint8_t sizeBytes[] = { static_cast<uint8_t>(querySize / 256), static_cast<uint8_t>(querySize % 256) };
1743     query.insert(query.begin(), sizeBytes, sizeBytes + 2);
1744     totalQueriesSize += query.size();
1745     ++counter;
1746   }
1747
1748   counter = 0;
1749   size_t totalResponsesSize = 0;
1750   for (auto& response : responses) {
1751     DNSName name("powerdns" + std::to_string(counter) + ".com.");
1752     GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
1753     pwR.getHeader()->qr = 1;
1754     pwR.getHeader()->rd = 1;
1755     pwR.getHeader()->ra = 1;
1756     pwR.getHeader()->id = htons(counter);
1757     pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
1758     pwR.xfr32BitInt(0x01020304);
1759     pwR.commit();
1760
1761     uint16_t responseSize = static_cast<uint16_t>(response.size());
1762     const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
1763     response.insert(response.begin(), sizeBytes, sizeBytes + 2);
1764     totalResponsesSize += response.size();
1765     ++counter;
1766   }
1767
1768   {
1769     TEST_INIT("=> 5 OOOR queries to the backend, backend responds in reverse order");
1770     PacketBuffer expectedWriteBuffer;
1771     PacketBuffer expectedBackendWriteBuffer;
1772
1773     uint16_t backendCounter = 0;
1774     for (const auto& query : queries) {
1775       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
1776       appendPayloadEditingID(expectedBackendWriteBuffer, query, backendCounter++);
1777     }
1778
1779     backendCounter = 0;
1780     for (const auto& response : responses) {
1781       /* reverse order */
1782       prependPayloadEditingID(s_backendReadBuffer, response, backendCounter++);
1783       expectedWriteBuffer.insert(expectedWriteBuffer.begin(), response.begin(), response.end());
1784     }
1785
1786     s_steps = {
1787       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1788       /* reading a query from the client (1) */
1789       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1790       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
1791       /* opening a connection to the backend */
1792       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1793       /* sending query to the backend */
1794       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
1795       /* no response ready yet */
1796       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1797       /* reading a query from the client (2) */
1798       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1799       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
1800       /* sending query to the backend */
1801       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
1802       /* no response ready yet */
1803       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1804       /* reading a query from the client (3) */
1805       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1806       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
1807       /* sending query to the backend */
1808       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
1809       /* no response ready yet */
1810       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1811       /* reading a query from the client (4) */
1812       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1813       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(3).size() - 2 },
1814       /* sending query to the backend */
1815       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(3).size() },
1816       /* no response ready yet */
1817       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1818       /* reading a query from the client (5) */
1819       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1820       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2 },
1821       /* sending query to the backend */
1822       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
1823       /* no response ready yet, but the backend becomes ready */
1824       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
1825         /* set the outgoing descriptor (backend connection) as ready */
1826         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
1827       } },
1828
1829       /* no more queries from the client */
1830       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
1831
1832       /* reading a response from the backend */
1833       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2 },
1834       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() },
1835       /* sending it to the client */
1836       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size() },
1837
1838       /* reading a response from the backend */
1839       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(3).size() - 2 },
1840       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(3).size() },
1841       /* sending it to the client */
1842       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(3).size() },
1843
1844       /* reading a response from the backend */
1845       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() - 2 },
1846       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() },
1847       /* sending it to the client */
1848       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(2).size() },
1849
1850       /* reading a response from the backend */
1851       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
1852       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() },
1853       /* sending it to the client */
1854       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size() },
1855
1856       /* reading a response from the backend */
1857       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
1858       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() },
1859       /* sending it to the client, the client descriptor becomes ready */
1860       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
1861         /* set the incoming descriptor (client connection) as ready */
1862         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
1863       } },
1864
1865       /* client is closing the connection */
1866       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
1867       /* closing client connection */
1868       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1869       /* closing a connection to the backend */
1870       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1871     };
1872
1873     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1874       selectedBackend = backend;
1875       return ProcessQueryResult::PassToBackend;
1876     };
1877     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
1878       return true;
1879     };
1880
1881     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
1882     IncomingTCPConnectionState::handleIO(state, now);
1883     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
1884       threadData.mplexer->run(&now);
1885     }
1886
1887     BOOST_CHECK_EQUAL(s_writeBuffer.size(), totalResponsesSize);
1888     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
1889     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), totalQueriesSize);
1890     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
1891     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
1892
1893     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
1894     IncomingTCPConnectionState::clearAllDownstreamConnections();
1895   }
1896
1897   {
1898     TEST_INIT("=> 3 queries sent to the backend, 1 self-answered, 1 new query sent to the backend which responds to the first query right away, then to the last one, then the connection to the backend times out");
1899
1900     // increase the client timeout for that test, we want the backend to timeout first
1901     g_tcpRecvTimeout = 5;
1902
1903     PacketBuffer expectedWriteBuffer;
1904     PacketBuffer expectedBackendWriteBuffer;
1905
1906     for (const auto& query : queries) {
1907       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
1908     }
1909
1910     uint16_t backendCounter = 0;
1911     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), backendCounter++);
1912     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), backendCounter++);
1913     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), backendCounter++);
1914     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(4), backendCounter++);
1915
1916     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 0);
1917     appendPayloadEditingID(s_backendReadBuffer, responses.at(4), 3);
1918
1919     /* self-answered */
1920     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(3).begin(), responses.at(3).end());
1921     /* from backend */
1922     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(0).begin(), responses.at(0).end());
1923     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(4).begin(), responses.at(4).end());
1924
1925     bool timeout = false;
1926     s_steps = {
1927       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
1928       /* reading a query from the client (1) */
1929       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1930       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
1931       /* opening a connection to the backend */
1932       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
1933       /* sending query to the backend */
1934       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
1935       /* no response ready yet */
1936       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1937       /* reading a query from the client (2) */
1938       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1939       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
1940       /* sending query to the backend */
1941       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
1942       /* no response ready yet */
1943       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1944       /* reading a query from the client (3) */
1945       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1946       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
1947       /* sending query to the backend */
1948       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
1949       /* no response ready yet */
1950       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
1951       /* reading a query from the client (4) */
1952       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1953       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(3).size() - 2 },
1954       /* sending the response right away (self-answered)  */
1955       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(3).size() },
1956       /* reading a query from the client (5) */
1957       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
1958       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2 },
1959       /* sending query to the backend (5) */
1960       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
1961       /* reading a response from the backend (1) */
1962       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
1963       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
1964         /* set the backend descriptor as ready */
1965         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
1966       } },
1967       /* sending it to the client (1) */
1968       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
1969         /* set the client descriptor as NOT ready */
1970         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
1971       } },
1972       /* reading from the client (not ready) */
1973       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
1974       /* reading a response from the backend (5) */
1975       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2 },
1976       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() },
1977       /* sending it to the client (5) */
1978       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size() },
1979
1980       /* try to read from the backend but there is no answer ready yet */
1981       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData, &timeout](int desc) {
1982         /* set the backend descriptor as NOT ready */
1983         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
1984         timeout = true;
1985       } },
1986
1987       /* A timeout occurs */
1988
1989       /* closing client connection */
1990       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
1991
1992       /* closing a connection to the backend */
1993       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
1994     };
1995
1996     s_processQuery = [backend,&responses](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
1997       static size_t count = 0;
1998       if (count++ == 3) {
1999         /* self answered */
2000         dq.getMutableData() = responses.at(3);
2001         /* remove the length */
2002         dq.getMutableData().erase(dq.getMutableData().begin(), dq.getMutableData().begin() + 2);
2003
2004         return ProcessQueryResult::SendAnswer;
2005       }
2006       selectedBackend = backend;
2007       return ProcessQueryResult::PassToBackend;
2008     };
2009     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2010       return true;
2011     };
2012
2013     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2014     IncomingTCPConnectionState::handleIO(state, now);
2015
2016     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2017       threadData.mplexer->run(&now);
2018     }
2019
2020     struct timeval later = now;
2021     later.tv_sec += backend->d_config.tcpRecvTimeout + 1;
2022     auto expiredConns = threadData.mplexer->getTimeouts(later, false);
2023     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
2024     for (const auto& cbData : expiredConns) {
2025       if (cbData.second.type() == typeid(std::shared_ptr<TCPConnectionToBackend>)) {
2026         auto cbState = boost::any_cast<std::shared_ptr<TCPConnectionToBackend>>(cbData.second);
2027         cbState->handleTimeout(later, false);
2028       }
2029     }
2030
2031     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2032     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2033     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2034     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2035     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2036
2037     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2038     IncomingTCPConnectionState::clearAllDownstreamConnections();
2039
2040     // restore the client timeout
2041     g_tcpRecvTimeout = 2;
2042   }
2043
2044   {
2045     TEST_INIT("=> 1 query sent to the backend, short read from the backend, 1 new query arrives in the meantime, the first answer is sent to the client, then the second query is handled, a new one arrives but the connection to the backend dies on us, short write on a new connection, the last query arrives and both are answered");
2046
2047     PacketBuffer expectedWriteBuffer;
2048     PacketBuffer expectedBackendWriteBuffer;
2049
2050     for (const auto& query : queries) {
2051       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
2052     }
2053     for (const auto& response : responses) {
2054       expectedWriteBuffer.insert(expectedWriteBuffer.end(), response.begin(), response.end());
2055     }
2056
2057     uint16_t backendCounter = 0;
2058     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), backendCounter);
2059     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), backendCounter++);
2060     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), backendCounter);
2061     appendPayloadEditingID(s_backendReadBuffer, responses.at(1), backendCounter++);
2062
2063     // new connection
2064     backendCounter = 0;
2065     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), backendCounter);
2066     appendPayloadEditingID(s_backendReadBuffer, responses.at(2), backendCounter++);
2067     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(3), backendCounter);
2068     appendPayloadEditingID(s_backendReadBuffer, responses.at(3), backendCounter++);
2069     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(4), backendCounter);
2070     appendPayloadEditingID(s_backendReadBuffer, responses.at(4), backendCounter++);
2071
2072     bool timeout = false;
2073     int backendDesc;
2074     s_steps = {
2075       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2076       /* reading a query from the client (1) */
2077       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2078       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
2079       /* opening a connection to the backend */
2080       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2081       /* sending query to the backend */
2082       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
2083       /* read response size and the beginning of the response (1) from the backend */
2084       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2085       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 1, [&threadData](int desc) {
2086         /* set the backend descriptor as ready */
2087         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2088       } },
2089       /* reading a query from the client (2) */
2090       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2091       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
2092       /* trying to read an additional query, if any */
2093       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData](int desc) {
2094         /* set the client descriptor as NOT ready */
2095         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2096       } },
2097       /* reading the remaining bytes of response (1) from the backend */
2098       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 3 },
2099       /* sending response (1) to the client */
2100       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size() },
2101       /* sending query (2) to the backend */
2102       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
2103       /* the response (2) is already there */
2104       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2105       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
2106       /* sending response (2) to the client */
2107       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size(), [&threadData](int desc) {
2108         /* set the client descriptor as ready */
2109         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2110       } },
2111       /* reading a query from the client (3) */
2112       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2113       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
2114       /* sending query to the backend */
2115       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, 0 },
2116       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2117       /* opening a connection to the backend */
2118       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2119       /* sending query (3) to the backend, short write */
2120       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::NeedWrite, 1, [&threadData,&backendDesc](int desc) {
2121         /* set the backend descriptor as NOT ready */
2122         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2123         backendDesc = desc;
2124         /* but client is ready */
2125         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
2126       } },
2127       /* reading a query from the client (4) */
2128       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2129       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(3).size() - 2 },
2130       /* reading a query from the client (5) */
2131       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2132       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2, [&threadData,&backendDesc](int desc) {
2133         /* set the backend descriptor as ready now */
2134         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDesc);
2135       } },
2136       /* nothing else to read from the client for now */
2137       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData](int desc) {
2138         /* set the client descriptor as NOT ready */
2139         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2140       } },
2141       /* finishing sending the query (3) to the backend */
2142       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() - 1 },
2143       /* sending the query (4) to the backend */
2144       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(3).size() },
2145       /* sending the query (5) to the backend */
2146       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
2147       /* reading a response from the backend (3) */
2148       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2149       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() - 2 },
2150       /* sending it to the client (3) */
2151       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(2).size() },
2152       /* reading a response from the backend (4) */
2153       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2154       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(3).size() - 2 },
2155       /* sending it to the client (4) but short write */
2156       { ExpectedStep::ExpectedRequest::writeToClient, IOState::NeedWrite, responses.at(3).size() - 1, [&threadData](int desc) {
2157         /* set the client descriptor as NOT ready */
2158         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2159       } },
2160       /* reading a response from the backend (5) */
2161       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2162       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2, [&threadData](int desc) {
2163         /* set the client descriptor as ready to resume sending */
2164         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
2165       } },
2166       /* resume sending it to the client (4) */
2167       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, 1 },
2168       /* sending it to the client (5) */
2169       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size() },
2170
2171       /* nothing to read from the client, then timeout later */
2172       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData,&timeout](int desc) {
2173         /* set the client descriptor as NOT ready */
2174         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2175         timeout = true;
2176       } },
2177
2178       /* closing client connection */
2179       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
2180
2181       /* closing a connection to the backend */
2182       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2183     };
2184
2185     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2186       selectedBackend = backend;
2187       return ProcessQueryResult::PassToBackend;
2188     };
2189     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2190       return true;
2191     };
2192
2193     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2194     IncomingTCPConnectionState::handleIO(state, now);
2195
2196     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2197       threadData.mplexer->run(&now);
2198     }
2199
2200     struct timeval later = now;
2201     later.tv_sec += g_tcpRecvTimeout + 1;
2202     auto expiredConns = threadData.mplexer->getTimeouts(later, false);
2203     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
2204     for (const auto& cbData : expiredConns) {
2205       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
2206         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
2207         cbState->handleTimeout(cbState, false);
2208       }
2209     }
2210
2211     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2212     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2213     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2214     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2215     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2216
2217     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2218     IncomingTCPConnectionState::clearAllDownstreamConnections();
2219   }
2220
2221   {
2222     TEST_INIT("=> 1 query to the backend, second query from the client is dropped, backend times out");
2223     // useful to tests that we check that the client connection is alive in notifyAllQueriesFailed()
2224     PacketBuffer expectedWriteBuffer;
2225     PacketBuffer expectedBackendWriteBuffer;
2226
2227     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2228     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
2229
2230     // only the first query is passed to the backend
2231     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2232
2233     bool timeout = false;
2234     s_steps = {
2235       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2236       /* reading a query from the client (1) */
2237       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2238       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
2239       /* opening a connection to the backend */
2240       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2241       /* sending query to the backend */
2242       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
2243       /* no response ready yet */
2244       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
2245       /* reading a second query from the client */
2246       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2247       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
2248       /* query is dropped, closing the connection to the client */
2249       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0, [&timeout](int desc) {
2250         timeout = true;
2251       } },
2252       /* closing a connection to the backend after a timeout */
2253       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2254     };
2255
2256     counter = 0;
2257     s_processQuery = [backend,&counter](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2258       if (counter == 0) {
2259         ++counter;
2260         selectedBackend = backend;
2261         return ProcessQueryResult::PassToBackend;
2262       }
2263       return ProcessQueryResult::Drop;
2264     };
2265     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2266       return true;
2267     };
2268
2269     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2270     IncomingTCPConnectionState::handleIO(state, now);
2271     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2272       threadData.mplexer->run(&now);
2273     }
2274
2275     struct timeval later = now;
2276     later.tv_sec += backend->d_config.tcpRecvTimeout + 1;
2277     auto expiredConns = threadData.mplexer->getTimeouts(later, false);
2278     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
2279     for (const auto& cbData : expiredConns) {
2280       if (cbData.second.type() == typeid(std::shared_ptr<TCPConnectionToBackend>)) {
2281         auto cbState = boost::any_cast<std::shared_ptr<TCPConnectionToBackend>>(cbData.second);
2282         cbState->handleTimeout(later, false);
2283       }
2284     }
2285
2286     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2287     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2288     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2289     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2290     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2291
2292     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2293     IncomingTCPConnectionState::clearAllDownstreamConnections();
2294   }
2295
2296   {
2297     TEST_INIT("=> 1 query to the backend, second query from the client is dropped, backend sends the answer");
2298     // useful to tests that we check that the client connection is alive in handleResponse()
2299     PacketBuffer expectedWriteBuffer;
2300     PacketBuffer expectedBackendWriteBuffer;
2301
2302     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2303     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
2304
2305     // only the first query is passed to the backend
2306     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2307
2308     s_backendReadBuffer.insert(s_backendReadBuffer.end(), responses.at(0).begin(), responses.at(0).end());
2309
2310     int backendDescriptor = -1;
2311     s_steps = {
2312       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2313       /* reading a query from the client (1) */
2314       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2315       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
2316       /* opening a connection to the backend */
2317       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptor](int desc) {
2318         backendDescriptor = desc;
2319       } },
2320       /* sending query to the backend */
2321       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
2322       /* no response ready yet */
2323       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
2324       /* reading a second query from the client */
2325       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2326       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
2327       /* query is dropped, closing the connection to the client */
2328       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0, [&threadData,&backendDescriptor](int desc) {
2329         /* the backend descriptor becomes ready */
2330         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptor);
2331       } },
2332       /* reading the response to the first query from the backend */
2333       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2334       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
2335       /* closing a connection to the backend */
2336       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2337     };
2338
2339     counter = 0;
2340     s_processQuery = [backend,&counter](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2341       if (counter == 0) {
2342         ++counter;
2343         selectedBackend = backend;
2344         return ProcessQueryResult::PassToBackend;
2345       }
2346       return ProcessQueryResult::Drop;
2347     };
2348     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2349       return true;
2350     };
2351
2352     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2353     IncomingTCPConnectionState::handleIO(state, now);
2354     while ((threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2355       threadData.mplexer->run(&now);
2356     }
2357
2358     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2359     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2360     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2361     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2362     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2363
2364     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2365     IncomingTCPConnectionState::clearAllDownstreamConnections();
2366   }
2367
2368   {
2369     TEST_INIT("=> 2 queries to the backend, client times out, responses arrive and are delivered, we start reading from the client again");
2370
2371     PacketBuffer expectedWriteBuffer;
2372     PacketBuffer expectedBackendWriteBuffer;
2373
2374     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2375     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
2376     s_readBuffer.insert(s_readBuffer.end(), queries.at(4).begin(), queries.at(4).end());
2377
2378     uint16_t backendCounter = 0;
2379     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), backendCounter++);
2380     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), backendCounter++);
2381     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(4), backendCounter++);
2382
2383     appendPayloadEditingID(s_backendReadBuffer, responses.at(1), 1);
2384     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 0);
2385     appendPayloadEditingID(s_backendReadBuffer, responses.at(4), 2);
2386
2387     appendPayloadEditingID(expectedWriteBuffer, responses.at(1), 1);
2388     appendPayloadEditingID(expectedWriteBuffer, responses.at(0), 0);
2389     appendPayloadEditingID(expectedWriteBuffer, responses.at(4), 4);
2390
2391     /* make sure that the backend's timeout is longer than the client's */
2392     backend->d_config.tcpRecvTimeout = 30;
2393
2394     bool timeout = false;
2395     int backendDescriptor = -1;
2396     s_steps = {
2397       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2398       /* reading a query from the client (1) */
2399       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2400       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
2401       /* opening a connection to the backend */
2402       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2403       /* sending query (1) to the backend */
2404       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
2405       /* no response ready yet */
2406       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
2407       /* reading a second query from the client */
2408       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2409       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
2410       /* sending query (2) to the backend */
2411       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
2412       /* no response ready yet */
2413       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&timeout,&backendDescriptor](int desc) {
2414         backendDescriptor = desc;
2415         timeout = true;
2416       } },
2417       /* nothing from the client either */
2418       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
2419       /* the client times out, and we will set the backend descriptor to ready at that point */
2420       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2421       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
2422       /* sending response (2) to the client */
2423       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size() },
2424       /* reading the response (1) from the backend */
2425       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2426       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
2427       /* sending response (1) to the client */
2428       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
2429         /* setting the client descriptor ready */
2430         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2431       } },
2432       /* try to read from the client again, get query (3) */
2433       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2434       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2 },
2435       /* sending query (3) to the backend */
2436       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
2437       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
2438         /* setting the backend descriptor NOT ready */
2439         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2440       } },
2441       /* try to read from the client again, nothing yet */
2442       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData,&backendDescriptor](int desc) {
2443         /* the client descriptor becomes NOT ready */
2444         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2445         /* the backend one is ready, though */
2446         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(backendDescriptor);
2447       } },
2448       /* reading the response (3) from the backend */
2449       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2450       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2 },
2451       /* sending response (3) to the client */
2452       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size(), [&timeout](int desc) {
2453         timeout = true;
2454       } },
2455       /* client times out again, this time we close the connection */
2456       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0 },
2457       /* closing a connection to the backend */
2458       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2459     };
2460
2461     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2462       selectedBackend = backend;
2463       return ProcessQueryResult::PassToBackend;
2464     };
2465     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2466       return true;
2467     };
2468
2469     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2470     IncomingTCPConnectionState::handleIO(state, now);
2471     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2472       threadData.mplexer->run(&now);
2473     }
2474
2475     struct timeval later = now;
2476     later.tv_sec += g_tcpRecvTimeout + 1;
2477     auto expiredConns = threadData.mplexer->getTimeouts(later, false);
2478     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
2479     for (const auto& cbData : expiredConns) {
2480       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
2481         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
2482         cbState->handleTimeout(cbState, false);
2483         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptor);
2484       }
2485     }
2486     timeout = false;
2487     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2488       threadData.mplexer->run(&now);
2489     }
2490
2491     later = now;
2492     later.tv_sec += g_tcpRecvTimeout + 1;
2493     expiredConns = threadData.mplexer->getTimeouts(later, false);
2494     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
2495     for (const auto& cbData : expiredConns) {
2496       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
2497         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
2498         cbState->handleTimeout(cbState, false);
2499       }
2500     }
2501
2502     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2503     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2504     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2505     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2506     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2507
2508     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2509     IncomingTCPConnectionState::clearAllDownstreamConnections();
2510   }
2511
2512   {
2513     TEST_INIT("=> 3 queries to the backend, the first 2 responses arrive and are queued (client write blocks), and the backend closes the connection before sending the last one");
2514
2515     PacketBuffer expectedWriteBuffer;
2516     PacketBuffer expectedBackendWriteBuffer;
2517
2518     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2519     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
2520     s_readBuffer.insert(s_readBuffer.end(), queries.at(2).begin(), queries.at(2).end());
2521
2522     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), queries.at(0).begin(), queries.at(0).end());
2523     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), queries.at(1).begin(), queries.at(1).end());
2524     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), queries.at(2).begin(), queries.at(2).end());
2525
2526     s_backendReadBuffer.insert(s_backendReadBuffer.end(), responses.at(1).begin(), responses.at(1).end());
2527     s_backendReadBuffer.insert(s_backendReadBuffer.end(), responses.at(0).begin(), responses.at(0).end());
2528
2529     expectedWriteBuffer = s_backendReadBuffer;
2530
2531     s_steps = {
2532       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2533       /* reading a query from the client (1) */
2534       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2535       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
2536       /* opening a connection to the backend */
2537       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2538       /* sending query (1) to the backend */
2539       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
2540       /* no response ready yet */
2541       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
2542       /* reading a second query from the client */
2543       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2544       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
2545       /* sending query (2) to the backend */
2546       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
2547       /* no response ready yet */
2548       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
2549       /* reading a third query from the client */
2550       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2551       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
2552       /* sending query (3) to the backend */
2553       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
2554       /* no response ready yet but the backend descriptor becomes ready */
2555       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
2556         /* the backend descriptor becomes ready */
2557         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2558       } },
2559       /* nothing from the client either */
2560       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData](int desc) {
2561         /* the client descriptor is NOT ready */
2562         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
2563       } },
2564       /* read the response (2) from the backend  */
2565       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2566       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
2567       /* trying to send response (2) to the client but blocking */
2568       { ExpectedStep::ExpectedRequest::writeToClient, IOState::NeedWrite, 0 },
2569       /* reading the response (1) from the backend */
2570       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2571       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
2572       /* trying to read from the backend again, connection closes on us */
2573       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
2574       /* so we close the connection */
2575       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2576       /* try opening a new connection to the backend, it fails (5) times */
2577       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [](int desc) {
2578         throw NetworkError("Connection refused by the backend");
2579       } },
2580       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2581       /* try opening a new connection to the backend, it fails (5) times */
2582       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done,0, [](int desc) {
2583         throw NetworkError("Connection refused by the backend");
2584       } },
2585       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2586       /* try opening a new connection to the backend, it fails (5) times */
2587       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done,0, [](int desc) {
2588         throw NetworkError("Connection refused by the backend");
2589       } },
2590       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2591       /* try opening a new connection to the backend, it fails (5) times */
2592       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done,0, [](int desc) {
2593         throw NetworkError("Connection refused by the backend");
2594       } },
2595       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2596       /* try opening a new connection to the backend, it fails (5) times */
2597       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done,0, [](int desc) {
2598         throw NetworkError("Connection refused by the backend");
2599       } },
2600       /* closing a connection to the backend, client becomes ready */
2601       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done, 0, [&threadData](int desc) {
2602         /* the client descriptor is ready */
2603         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
2604       } },
2605       /* sending response (2) to the client */
2606       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size() },
2607       /* sending response (1) to the client */
2608       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size() },
2609       /* closing the client connection */
2610       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0 },
2611     };
2612
2613     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2614       selectedBackend = backend;
2615       return ProcessQueryResult::PassToBackend;
2616     };
2617     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2618       return true;
2619     };
2620
2621     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2622     IncomingTCPConnectionState::handleIO(state, now);
2623     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
2624       threadData.mplexer->run(&now);
2625     }
2626
2627     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2628     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2629     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2630     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2631     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
2632
2633     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2634     IncomingTCPConnectionState::clearAllDownstreamConnections();
2635   }
2636
2637   {
2638     TEST_INIT("=> AXFR");
2639
2640     PacketBuffer axfrQuery;
2641     PacketBuffer secondQuery;
2642     std::vector<PacketBuffer> axfrResponses(3);
2643     PacketBuffer secondResponse;
2644
2645     GenericDNSPacketWriter<PacketBuffer> pwAXFRQuery(axfrQuery, DNSName("powerdns.com."), QType::AXFR, QClass::IN, 0);
2646     pwAXFRQuery.getHeader()->rd = 0;
2647     pwAXFRQuery.getHeader()->id = 42;
2648     uint16_t axfrQuerySize = static_cast<uint16_t>(axfrQuery.size());
2649     const uint8_t axfrQuerySizeBytes[] = { static_cast<uint8_t>(axfrQuerySize / 256), static_cast<uint8_t>(axfrQuerySize % 256) };
2650     axfrQuery.insert(axfrQuery.begin(), axfrQuerySizeBytes, axfrQuerySizeBytes + 2);
2651
2652     const DNSName name("powerdns.com.");
2653     {
2654       /* first message */
2655       auto& response = axfrResponses.at(0);
2656       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2657       pwR.getHeader()->qr = 1;
2658       pwR.getHeader()->id = 42;
2659
2660       /* insert SOA */
2661       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
2662       pwR.xfrName(g_rootdnsname, true);
2663       pwR.xfrName(g_rootdnsname, true);
2664       pwR.xfr32BitInt(1 /* serial */);
2665       pwR.xfr32BitInt(0);
2666       pwR.xfr32BitInt(0);
2667       pwR.xfr32BitInt(0);
2668       pwR.xfr32BitInt(0);
2669       pwR.commit();
2670
2671       /* A record */
2672       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2673       pwR.xfr32BitInt(0x01020304);
2674       pwR.commit();
2675
2676       uint16_t responseSize = static_cast<uint16_t>(response.size());
2677       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2678       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2679     }
2680     {
2681       /* second message */
2682       auto& response = axfrResponses.at(1);
2683       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2684       pwR.getHeader()->qr = 1;
2685       pwR.getHeader()->id = 42;
2686
2687       /* A record */
2688       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2689       pwR.xfr32BitInt(0x01020304);
2690       pwR.commit();
2691
2692       uint16_t responseSize = static_cast<uint16_t>(response.size());
2693       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2694       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2695     }
2696     {
2697       /* third message */
2698       auto& response = axfrResponses.at(2);
2699       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2700       pwR.getHeader()->qr = 1;
2701       pwR.getHeader()->id = 42;
2702
2703       /* A record */
2704       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2705       pwR.xfr32BitInt(0x01020304);
2706       pwR.commit();
2707
2708       /* final SOA */
2709       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
2710       pwR.xfrName(g_rootdnsname, true);
2711       pwR.xfrName(g_rootdnsname, true);
2712       pwR.xfr32BitInt(1 /* serial */);
2713       pwR.xfr32BitInt(0);
2714       pwR.xfr32BitInt(0);
2715       pwR.xfr32BitInt(0);
2716       pwR.xfr32BitInt(0);
2717       pwR.commit();
2718
2719       uint16_t responseSize = static_cast<uint16_t>(response.size());
2720       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2721       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2722     }
2723
2724     {
2725       GenericDNSPacketWriter<PacketBuffer> pwSecondQuery(secondQuery, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
2726       pwSecondQuery.getHeader()->rd = 1;
2727       pwSecondQuery.getHeader()->id = 84;
2728       uint16_t secondQuerySize = static_cast<uint16_t>(secondQuery.size());
2729       const uint8_t secondQuerySizeBytes[] = { static_cast<uint8_t>(secondQuerySize / 256), static_cast<uint8_t>(secondQuerySize % 256) };
2730       secondQuery.insert(secondQuery.begin(), secondQuerySizeBytes, secondQuerySizeBytes + 2);
2731     }
2732
2733     {
2734       GenericDNSPacketWriter<PacketBuffer> pwSecondResponse(secondResponse, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
2735       pwSecondResponse.getHeader()->qr = 1;
2736       pwSecondResponse.getHeader()->rd = 1;
2737       pwSecondResponse.getHeader()->ra = 1;
2738       pwSecondResponse.getHeader()->id = 84;
2739       pwSecondResponse.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2740       pwSecondResponse.xfr32BitInt(0x01020304);
2741       pwSecondResponse.commit();
2742       uint16_t responseSize = static_cast<uint16_t>(secondResponse.size());
2743       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2744       secondResponse.insert(secondResponse.begin(), sizeBytes, sizeBytes + 2);
2745     }
2746
2747     PacketBuffer expectedWriteBuffer;
2748     PacketBuffer expectedBackendWriteBuffer;
2749
2750     s_readBuffer = axfrQuery;
2751     s_readBuffer.insert(s_readBuffer.end(), secondQuery.begin(), secondQuery.end());
2752
2753     uint16_t backendCounter = 0;
2754     appendPayloadEditingID(expectedBackendWriteBuffer, axfrQuery, backendCounter++);
2755     appendPayloadEditingID(expectedBackendWriteBuffer, secondQuery, backendCounter++);
2756
2757     for (const auto& response : axfrResponses) {
2758       appendPayloadEditingID(s_backendReadBuffer, response, 0);
2759       expectedWriteBuffer.insert(expectedWriteBuffer.end(), response.begin(), response.end());
2760     }
2761     appendPayloadEditingID(s_backendReadBuffer, secondResponse, 1);
2762     expectedWriteBuffer.insert(expectedWriteBuffer.end(), secondResponse.begin(), secondResponse.end());
2763
2764     bool timeout = false;
2765     s_steps = {
2766       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2767       /* reading a query from the client (1) */
2768       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2769       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, axfrQuery.size() - 2 },
2770       /* opening a connection to the backend */
2771       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2772       /* sending query (1) to the backend */
2773       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, axfrQuery.size() },
2774       /* no response ready yet, but setting the backend descriptor readable */
2775       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
2776         /* the backend descriptor becomes ready */
2777         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2778       } },
2779       /* no more query from the client for now */
2780       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 , [&threadData](int desc) {
2781         /* the client descriptor becomes NOT ready */
2782         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
2783       } },
2784       /* read the response (1) from the backend  */
2785       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2786       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, axfrResponses.at(0).size() - 2 },
2787       /* sending response (1) to the client */
2788       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, axfrResponses.at(0).size() },
2789       /* reading the response (2) from the backend */
2790       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2791       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, axfrResponses.at(1).size() - 2 },
2792       /* sending response (2) to the client */
2793       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, axfrResponses.at(1).size() },
2794       /* reading the response (3) from the backend */
2795       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2796       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, axfrResponses.at(2).size() - 2 },
2797       /* sending response (3) to the client */
2798       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, axfrResponses.at(2).size(), [&threadData](int desc) {
2799         /* the client descriptor becomes ready */
2800         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
2801       } },
2802       /* trying to read from the client, getting a second query */
2803       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2804       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, secondQuery.size() - 2 },
2805       /* sending query (2) to the backend */
2806       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, secondQuery.size() },
2807       /* reading the response (4) from the backend */
2808       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2809       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, secondResponse.size() - 2 },
2810       /* sending response (4) to the client */
2811       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, secondResponse.size() },
2812       /* trying to read from the client, getting EOF */
2813       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
2814       /* closing the client connection */
2815       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
2816       /* closing the backend connection */
2817       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2818     };
2819
2820     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2821       selectedBackend = backend;
2822       return ProcessQueryResult::PassToBackend;
2823     };
2824     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2825       return true;
2826     };
2827
2828     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
2829     IncomingTCPConnectionState::handleIO(state, now);
2830     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
2831       threadData.mplexer->run(&now);
2832     }
2833
2834     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
2835     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
2836     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
2837     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
2838
2839     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
2840     IncomingTCPConnectionState::clearAllDownstreamConnections();
2841   }
2842
2843   {
2844     TEST_INIT("=> Interrupted AXFR");
2845
2846     PacketBuffer axfrQuery;
2847     PacketBuffer secondQuery;
2848     std::vector<PacketBuffer> axfrResponses(3);
2849     PacketBuffer secondResponse;
2850
2851     auto proxyPayload = makeProxyHeader(true, getBackendAddress("84", 4242), local, {});
2852     BOOST_REQUIRE_GT(proxyPayload.size(), s_proxyProtocolMinimumHeaderSize);
2853
2854     auto proxyEnabledBackend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
2855     proxyEnabledBackend->d_tlsCtx = tlsCtx;
2856     proxyEnabledBackend->d_config.useProxyProtocol = true;
2857
2858     GenericDNSPacketWriter<PacketBuffer> pwAXFRQuery(axfrQuery, DNSName("powerdns.com."), QType::AXFR, QClass::IN, 0);
2859     pwAXFRQuery.getHeader()->rd = 0;
2860     pwAXFRQuery.getHeader()->id = 42;
2861     uint16_t axfrQuerySize = static_cast<uint16_t>(axfrQuery.size());
2862     const uint8_t axfrQuerySizeBytes[] = { static_cast<uint8_t>(axfrQuerySize / 256), static_cast<uint8_t>(axfrQuerySize % 256) };
2863     axfrQuery.insert(axfrQuery.begin(), axfrQuerySizeBytes, axfrQuerySizeBytes + 2);
2864
2865     const DNSName name("powerdns.com.");
2866     {
2867       /* first message */
2868       auto& response = axfrResponses.at(0);
2869       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2870       pwR.getHeader()->qr = 1;
2871       pwR.getHeader()->id = 42;
2872
2873       /* insert SOA */
2874       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
2875       pwR.xfrName(g_rootdnsname, true);
2876       pwR.xfrName(g_rootdnsname, true);
2877       pwR.xfr32BitInt(1 /* serial */);
2878       pwR.xfr32BitInt(0);
2879       pwR.xfr32BitInt(0);
2880       pwR.xfr32BitInt(0);
2881       pwR.xfr32BitInt(0);
2882       pwR.commit();
2883
2884       /* A record */
2885       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2886       pwR.xfr32BitInt(0x01020304);
2887       pwR.commit();
2888
2889       uint16_t responseSize = static_cast<uint16_t>(response.size());
2890       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2891       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2892     }
2893     {
2894       /* second message */
2895       auto& response = axfrResponses.at(1);
2896       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2897       pwR.getHeader()->qr = 1;
2898       pwR.getHeader()->id = 42;
2899
2900       /* A record */
2901       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2902       pwR.xfr32BitInt(0x01020304);
2903       pwR.commit();
2904
2905       uint16_t responseSize = static_cast<uint16_t>(response.size());
2906       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2907       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2908     }
2909     {
2910       /* third message */
2911       auto& response = axfrResponses.at(2);
2912       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
2913       pwR.getHeader()->qr = 1;
2914       pwR.getHeader()->id = 42;
2915
2916       /* A record */
2917       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
2918       pwR.xfr32BitInt(0x01020304);
2919       pwR.commit();
2920
2921       /* final SOA */
2922       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
2923       pwR.xfrName(g_rootdnsname, true);
2924       pwR.xfrName(g_rootdnsname, true);
2925       pwR.xfr32BitInt(1 /* serial */);
2926       pwR.xfr32BitInt(0);
2927       pwR.xfr32BitInt(0);
2928       pwR.xfr32BitInt(0);
2929       pwR.xfr32BitInt(0);
2930       pwR.commit();
2931
2932       uint16_t responseSize = static_cast<uint16_t>(response.size());
2933       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
2934       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
2935     }
2936
2937     PacketBuffer expectedWriteBuffer;
2938     PacketBuffer expectedBackendWriteBuffer;
2939
2940     s_readBuffer = axfrQuery;
2941
2942     uint16_t backendCounter = 0;
2943     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), proxyPayload.begin(), proxyPayload.end());
2944     appendPayloadEditingID(expectedBackendWriteBuffer, axfrQuery, backendCounter++);
2945
2946     for (const auto& response : axfrResponses) {
2947       appendPayloadEditingID(s_backendReadBuffer, response, 0);
2948     }
2949     expectedWriteBuffer.insert(expectedWriteBuffer.end(), axfrResponses.at(0).begin(), axfrResponses.at(0).end());
2950     expectedWriteBuffer.insert(expectedWriteBuffer.end(), axfrResponses.at(1).begin(), axfrResponses.at(1).end());
2951
2952     bool timeout = false;
2953     s_steps = {
2954       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
2955       /* reading a query from the client (1) */
2956       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
2957       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, axfrQuery.size() - 2 },
2958       /* opening a connection to the backend */
2959       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2960       /* sending query (1) to the backend */
2961       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, proxyPayload.size() + axfrQuery.size() },
2962       /* no response ready yet, but setting the backend descriptor readable */
2963       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
2964         /* the backend descriptor becomes ready */
2965         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
2966       } },
2967       /* no more query from the client for now */
2968       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 , [&threadData](int desc) {
2969         /* the client descriptor becomes NOT ready */
2970         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(-1);
2971       } },
2972       /* read the response (1) from the backend  */
2973       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2974       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, axfrResponses.at(0).size() - 2 },
2975       /* sending response (1) to the client */
2976       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, axfrResponses.at(0).size() },
2977       /* reading the response (2) from the backend */
2978       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
2979       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, axfrResponses.at(1).size() - 2 },
2980       /* sending response (2) to the client */
2981       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, axfrResponses.at(1).size() },
2982       /* reading the response (3) from the backend, get EOF!! */
2983       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
2984       /* closing the backend connection */
2985       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2986       /* opening a connection to the backend */
2987       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
2988       /* closing the client connection */
2989       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
2990       /* closing the backend connection */
2991       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
2992     };
2993
2994     s_processQuery = [proxyEnabledBackend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
2995       selectedBackend = proxyEnabledBackend;
2996       return ProcessQueryResult::PassToBackend;
2997     };
2998     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
2999       return true;
3000     };
3001
3002     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3003     IncomingTCPConnectionState::handleIO(state, now);
3004     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
3005       threadData.mplexer->run(&now);
3006     }
3007
3008     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
3009     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3010     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
3011     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3012
3013     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3014     IncomingTCPConnectionState::clearAllDownstreamConnections();
3015   }
3016
3017   {
3018     TEST_INIT("=> IXFR");
3019
3020     PacketBuffer firstQuery;
3021     PacketBuffer ixfrQuery;
3022     PacketBuffer secondQuery;
3023     PacketBuffer firstResponse;
3024     std::vector<PacketBuffer> ixfrResponses(1);
3025     PacketBuffer secondResponse;
3026
3027     {
3028       GenericDNSPacketWriter<PacketBuffer> pwFirstQuery(firstQuery, DNSName("powerdns.com."), QType::SOA, QClass::IN, 0);
3029       pwFirstQuery.getHeader()->rd = 1;
3030       pwFirstQuery.getHeader()->id = 84;
3031       uint16_t firstQuerySize = static_cast<uint16_t>(firstQuery.size());
3032       const uint8_t firstQuerySizeBytes[] = { static_cast<uint8_t>(firstQuerySize / 256), static_cast<uint8_t>(firstQuerySize % 256) };
3033       firstQuery.insert(firstQuery.begin(), firstQuerySizeBytes, firstQuerySizeBytes + 2);
3034     }
3035
3036     {
3037       GenericDNSPacketWriter<PacketBuffer> pwFirstResponse(firstResponse, DNSName("powerdns.com."), QType::SOA, QClass::IN, 0);
3038       pwFirstResponse.getHeader()->qr = 1;
3039       pwFirstResponse.getHeader()->rd = 1;
3040       pwFirstResponse.getHeader()->ra = 1;
3041       pwFirstResponse.getHeader()->id = 84;
3042       pwFirstResponse.startRecord(DNSName("powerdns.com."), QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3043       pwFirstResponse.xfrName(g_rootdnsname, true);
3044       pwFirstResponse.xfrName(g_rootdnsname, true);
3045       pwFirstResponse.xfr32BitInt(3 /* serial */);
3046       pwFirstResponse.xfr32BitInt(0);
3047       pwFirstResponse.xfr32BitInt(0);
3048       pwFirstResponse.xfr32BitInt(0);
3049       pwFirstResponse.xfr32BitInt(0);
3050       pwFirstResponse.commit();
3051
3052       uint16_t responseSize = static_cast<uint16_t>(firstResponse.size());
3053       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
3054       firstResponse.insert(firstResponse.begin(), sizeBytes, sizeBytes + 2);
3055     }
3056
3057     {
3058       GenericDNSPacketWriter<PacketBuffer> pwIXFRQuery(ixfrQuery, DNSName("powerdns.com."), QType::IXFR, QClass::IN, 0);
3059       pwIXFRQuery.getHeader()->rd = 0;
3060       pwIXFRQuery.getHeader()->id = 42;
3061       uint16_t ixfrQuerySize = static_cast<uint16_t>(ixfrQuery.size());
3062       const uint8_t ixfrQuerySizeBytes[] = { static_cast<uint8_t>(ixfrQuerySize / 256), static_cast<uint8_t>(ixfrQuerySize % 256) };
3063       ixfrQuery.insert(ixfrQuery.begin(), ixfrQuerySizeBytes, ixfrQuerySizeBytes + 2);
3064     }
3065
3066     const DNSName name("powerdns.com.");
3067     {
3068       /* first message */
3069       auto& response = ixfrResponses.at(0);
3070       GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
3071       pwR.getHeader()->qr = 1;
3072       pwR.getHeader()->id = 42;
3073
3074       /* insert final SOA */
3075       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3076       pwR.xfrName(g_rootdnsname, true);
3077       pwR.xfrName(g_rootdnsname, true);
3078       pwR.xfr32BitInt(3 /* serial */);
3079       pwR.xfr32BitInt(0);
3080       pwR.xfr32BitInt(0);
3081       pwR.xfr32BitInt(0);
3082       pwR.xfr32BitInt(0);
3083       pwR.commit();
3084
3085       /* insert first SOA */
3086       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3087       pwR.xfrName(g_rootdnsname, true);
3088       pwR.xfrName(g_rootdnsname, true);
3089       pwR.xfr32BitInt(1 /* serial */);
3090       pwR.xfr32BitInt(0);
3091       pwR.xfr32BitInt(0);
3092       pwR.xfr32BitInt(0);
3093       pwR.xfr32BitInt(0);
3094       pwR.commit();
3095
3096       /* removals */
3097       /* A record */
3098       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
3099       pwR.xfr32BitInt(0x01020304);
3100       pwR.commit();
3101
3102       /* additions */
3103       /* insert second SOA */
3104       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3105       pwR.xfrName(g_rootdnsname, true);
3106       pwR.xfrName(g_rootdnsname, true);
3107       pwR.xfr32BitInt(2 /* serial */);
3108       pwR.xfr32BitInt(0);
3109       pwR.xfr32BitInt(0);
3110       pwR.xfr32BitInt(0);
3111       pwR.xfr32BitInt(0);
3112       pwR.commit();
3113       /* A record */
3114       pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
3115       pwR.xfr32BitInt(0x01020305);
3116       pwR.commit();
3117       /* done with 1 -> 2 */
3118       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3119       pwR.xfrName(g_rootdnsname, true);
3120       pwR.xfrName(g_rootdnsname, true);
3121       pwR.xfr32BitInt(2 /* serial */);
3122       pwR.xfr32BitInt(0);
3123       pwR.xfr32BitInt(0);
3124       pwR.xfr32BitInt(0);
3125       pwR.xfr32BitInt(0);
3126       pwR.commit();
3127
3128       /* no removal */
3129
3130       /* additions */
3131       /* insert second SOA */
3132       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3133       pwR.xfrName(g_rootdnsname, true);
3134       pwR.xfrName(g_rootdnsname, true);
3135       pwR.xfr32BitInt(3 /* serial */);
3136       pwR.xfr32BitInt(0);
3137       pwR.xfr32BitInt(0);
3138       pwR.xfr32BitInt(0);
3139       pwR.xfr32BitInt(0);
3140       pwR.commit();
3141
3142       /* actually no addition either */
3143       /* done */
3144       pwR.startRecord(name, QType::SOA, 3600, QClass::IN, DNSResourceRecord::ANSWER);
3145       pwR.xfrName(g_rootdnsname, true);
3146       pwR.xfrName(g_rootdnsname, true);
3147       pwR.xfr32BitInt(3 /* serial */);
3148       pwR.xfr32BitInt(0);
3149       pwR.xfr32BitInt(0);
3150       pwR.xfr32BitInt(0);
3151       pwR.xfr32BitInt(0);
3152       pwR.commit();
3153
3154       uint16_t responseSize = static_cast<uint16_t>(response.size());
3155       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
3156       response.insert(response.begin(), sizeBytes, sizeBytes + 2);
3157     }
3158
3159     {
3160       GenericDNSPacketWriter<PacketBuffer> pwSecondQuery(secondQuery, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
3161       pwSecondQuery.getHeader()->rd = 1;
3162       pwSecondQuery.getHeader()->id = 84;
3163       uint16_t secondQuerySize = static_cast<uint16_t>(secondQuery.size());
3164       const uint8_t secondQuerySizeBytes[] = { static_cast<uint8_t>(secondQuerySize / 256), static_cast<uint8_t>(secondQuerySize % 256) };
3165       secondQuery.insert(secondQuery.begin(), secondQuerySizeBytes, secondQuerySizeBytes + 2);
3166     }
3167
3168     {
3169       GenericDNSPacketWriter<PacketBuffer> pwSecondResponse(secondResponse, DNSName("powerdns.com."), QType::A, QClass::IN, 0);
3170       pwSecondResponse.getHeader()->qr = 1;
3171       pwSecondResponse.getHeader()->rd = 1;
3172       pwSecondResponse.getHeader()->ra = 1;
3173       pwSecondResponse.getHeader()->id = 84;
3174       pwSecondResponse.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
3175       pwSecondResponse.xfr32BitInt(0x01020304);
3176       pwSecondResponse.commit();
3177       uint16_t responseSize = static_cast<uint16_t>(secondResponse.size());
3178       const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
3179       secondResponse.insert(secondResponse.begin(), sizeBytes, sizeBytes + 2);
3180     }
3181
3182     PacketBuffer expectedWriteBuffer;
3183     PacketBuffer expectedBackendWriteBuffer;
3184
3185     s_readBuffer = firstQuery;
3186     s_readBuffer.insert(s_readBuffer.end(), ixfrQuery.begin(), ixfrQuery.end());
3187     s_readBuffer.insert(s_readBuffer.end(), secondQuery.begin(), secondQuery.end());
3188
3189     appendPayloadEditingID(expectedBackendWriteBuffer, firstQuery, 0);
3190     appendPayloadEditingID(expectedBackendWriteBuffer, ixfrQuery, 1);
3191     appendPayloadEditingID(expectedBackendWriteBuffer, secondQuery, 2);
3192
3193     appendPayloadEditingID(s_backendReadBuffer, firstResponse, 0);
3194     expectedWriteBuffer.insert(expectedWriteBuffer.begin(), firstResponse.begin(), firstResponse.end());
3195     for (const auto& response : ixfrResponses) {
3196       appendPayloadEditingID(s_backendReadBuffer, response, 1);
3197       expectedWriteBuffer.insert(expectedWriteBuffer.end(), response.begin(), response.end());
3198     }
3199     appendPayloadEditingID(s_backendReadBuffer, secondResponse, 2);
3200     expectedWriteBuffer.insert(expectedWriteBuffer.end(), secondResponse.begin(), secondResponse.end());
3201
3202     bool timeout = false;
3203     s_steps = {
3204       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3205       /* reading a query from the client (1) */
3206       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3207       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, firstQuery.size() - 2 },
3208       /* opening a connection to the backend */
3209       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3210       /* sending query (1) to the backend */
3211       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, firstQuery.size() },
3212       /* no response ready yet, but setting the backend descriptor readable */
3213       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
3214         /* the backend descriptor becomes ready */
3215         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3216       } },
3217       /* try to read a second query from the client, none yet */
3218       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
3219       /* read the response (1) from the backend  */
3220       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3221       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, firstResponse.size() - 2 },
3222       /* sending response (1) to the client */
3223       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, firstResponse.size(), [&threadData](int desc) {
3224         /* client descriptor becomes ready */
3225         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
3226       } },
3227       /* reading a query from the client (2) */
3228       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3229       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, ixfrQuery.size() - 2 },
3230       /* sending query (2) to the backend */
3231       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, ixfrQuery.size() },
3232       /* read the response (ixfr 1) from the backend  */
3233       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3234       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, ixfrResponses.at(0).size() - 2 },
3235       /* sending response (ixfr 1) to the client */
3236       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, ixfrResponses.at(0).size(), [&threadData](int desc) {
3237         /* the client descriptor becomes ready */
3238         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
3239       } },
3240       /* trying to read from the client, getting a second query */
3241       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3242       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, secondQuery.size() - 2 },
3243       /* sending query (2) to the backend */
3244       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, secondQuery.size() },
3245       /* reading the response (4) from the backend */
3246       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3247       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, secondResponse.size() - 2 },
3248       /* sending response (4) to the client */
3249       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, secondResponse.size() },
3250       /* trying to read from the client, getting EOF */
3251       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
3252       /* closing the client connection */
3253       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
3254       /* closing the backend connection */
3255       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
3256     };
3257
3258     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3259       selectedBackend = backend;
3260       return ProcessQueryResult::PassToBackend;
3261     };
3262     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3263       return true;
3264     };
3265
3266     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3267     IncomingTCPConnectionState::handleIO(state, now);
3268     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
3269       threadData.mplexer->run(&now);
3270     }
3271
3272     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
3273     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3274     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
3275     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3276     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
3277
3278     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3279     IncomingTCPConnectionState::clearAllDownstreamConnections();
3280   }
3281
3282   {
3283     TEST_INIT("=> Outgoing proxy protocol, 3 queries to the backend, first response is sent, connection closed while reading the second one");
3284
3285     PacketBuffer expectedWriteBuffer;
3286     PacketBuffer expectedBackendWriteBuffer;
3287
3288     auto proxyPayload = makeProxyHeader(true, getBackendAddress("84", 4242), local, {});
3289     BOOST_REQUIRE_GT(proxyPayload.size(), s_proxyProtocolMinimumHeaderSize);
3290
3291     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
3292     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
3293     s_readBuffer.insert(s_readBuffer.end(), queries.at(2).begin(), queries.at(2).end());
3294
3295     auto proxyEnabledBackend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
3296     proxyEnabledBackend->d_tlsCtx = tlsCtx;
3297     /* enable out-of-order on the backend side as well */
3298     proxyEnabledBackend->d_config.d_maxInFlightQueriesPerConn = 65536;
3299     proxyEnabledBackend->d_config.useProxyProtocol = true;
3300
3301     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), proxyPayload.begin(), proxyPayload.end());
3302     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), 0);
3303     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), 1);
3304     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), 2);
3305     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), proxyPayload.begin(), proxyPayload.end());
3306     /* we are using an unordered_map, so all bets are off here :-/ */
3307     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), 0);
3308     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), 1);
3309
3310     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 0);
3311     /* after the reconnection */
3312     appendPayloadEditingID(s_backendReadBuffer, responses.at(1), 1);
3313     appendPayloadEditingID(s_backendReadBuffer, responses.at(2), 0);
3314
3315     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(0).begin(), responses.at(0).end());
3316     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(1).begin(), responses.at(1).end());
3317     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(2).begin(), responses.at(2).end());
3318
3319     s_steps = {
3320       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3321       /* reading a query from the client (1) */
3322       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3323       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3324       /* opening a connection to the backend */
3325       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3326       /* sending query (1) to the backend */
3327       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, proxyPayload.size() + queries.at(0).size() },
3328       /* got the response */
3329       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3330       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() },
3331       /* sending the response (1) to the client */
3332       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size() },
3333       /* reading a second query from the client */
3334       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3335       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3336       /* sending query (2) to the backend */
3337       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
3338       /* backend is not ready yet */
3339       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3340       /* reading a third query from the client */
3341       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3342       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
3343       /* sending query (3) to the backend */
3344       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
3345       /* backend is not ready yet, but the descriptor becomes ready */
3346       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
3347         /* the backend descriptor becomes ready */
3348         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3349       }},
3350       /* nothing from the client */
3351       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData](int desc) {
3352         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
3353       } },
3354       /* backend closes the connection on us */
3355       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 0 },
3356       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
3357       /* opening a new connection to the backend */
3358       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3359       /* sending query (2) to the backend */
3360       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, proxyPayload.size() + queries.at(1).size() },
3361       /* sending query (3) to the backend */
3362       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
3363       /* got the response for 2 */
3364       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3365       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() },
3366       /* sending the response (2) to the client */
3367       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size() },
3368       /* got the response for 3 */
3369       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3370       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() },
3371       /* sending the response (3) to the client */
3372       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(2).size(), [&threadData](int desc) {
3373         /* the client descriptor becomes ready */
3374         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3375       } },
3376       /* client closes the connection */
3377       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
3378       /* closing the client connection */
3379       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0 },
3380       /* closing the backend connection */
3381       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done, 0 },
3382     };
3383
3384     s_processQuery = [proxyEnabledBackend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3385       selectedBackend = proxyEnabledBackend;
3386       return ProcessQueryResult::PassToBackend;
3387     };
3388     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3389       return true;
3390     };
3391
3392     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3393     IncomingTCPConnectionState::handleIO(state, now);
3394     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
3395       threadData.mplexer->run(&now);
3396     }
3397
3398     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
3399     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3400     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
3401     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3402     BOOST_CHECK_EQUAL(proxyEnabledBackend->outstanding.load(), 0U);
3403
3404     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3405     /* we should have nothing to clear since the connection cannot be reused due to the Proxy Protocol payload */
3406     BOOST_CHECK_EQUAL(IncomingTCPConnectionState::clearAllDownstreamConnections(), 0U);
3407   }
3408
3409   {
3410     TEST_INIT("=> Outgoing proxy protocol, 3 queries to the backend, the client closes while sending the first response");
3411
3412     PacketBuffer expectedWriteBuffer;
3413     PacketBuffer expectedBackendWriteBuffer;
3414
3415     auto proxyPayload = makeProxyHeader(true, getBackendAddress("84", 4242), local, {});
3416     BOOST_REQUIRE_GT(proxyPayload.size(), s_proxyProtocolMinimumHeaderSize);
3417
3418     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
3419     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
3420     s_readBuffer.insert(s_readBuffer.end(), queries.at(2).begin(), queries.at(2).end());
3421
3422     auto proxyEnabledBackend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
3423     proxyEnabledBackend->d_tlsCtx = tlsCtx;
3424     /* enable out-of-order on the backend side as well */
3425     proxyEnabledBackend->d_config.d_maxInFlightQueriesPerConn = 65536;
3426     proxyEnabledBackend->d_config.useProxyProtocol = true;
3427
3428     expectedBackendWriteBuffer.insert(expectedBackendWriteBuffer.end(), proxyPayload.begin(), proxyPayload.end());
3429     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), 0);
3430     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), 1);
3431     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), 2);
3432
3433     s_steps = {
3434       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3435       /* reading a query from the client (1) */
3436       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3437       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3438       /* opening a connection to the backend */
3439       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3440       /* sending query (1) to the backend */
3441       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, proxyPayload.size() + queries.at(0).size() },
3442       /* we try to read the response, not ready yet */
3443       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3444       /* reading a second query from the client */
3445       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3446       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3447       /* sending query (2) to the backend */
3448       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
3449       /* backend is not ready yet */
3450       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3451       /* reading a third query from the client */
3452       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3453       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
3454       /* sending query (3) to the backend */
3455       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
3456       /* backend is not ready yet, but the descriptor becomes ready */
3457       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
3458         /* the backend descriptor becomes ready */
3459         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3460       }},
3461       /* client closes the connection */
3462       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
3463       /* closing the backend connection */
3464       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done, 0 },
3465       /* closing the client connection */
3466       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0 },
3467     };
3468
3469     s_processQuery = [proxyEnabledBackend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3470       selectedBackend = proxyEnabledBackend;
3471       return ProcessQueryResult::PassToBackend;
3472     };
3473     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3474       return true;
3475     };
3476
3477     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3478     IncomingTCPConnectionState::handleIO(state, now);
3479     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
3480       threadData.mplexer->run(&now);
3481     }
3482
3483     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
3484     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3485     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
3486     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3487     BOOST_CHECK_EQUAL(proxyEnabledBackend->outstanding.load(), 0U);
3488
3489     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3490     /* we should have nothing to clear since the connection cannot be reused due to the Proxy Protocol payload */
3491     BOOST_CHECK_EQUAL(IncomingTCPConnectionState::clearAllDownstreamConnections(), 0U);
3492   }
3493
3494   {
3495     TEST_INIT("=> I/O error with the backend with queries not sent to the backend yet");
3496
3497     PacketBuffer expectedWriteBuffer;
3498     PacketBuffer expectedBackendWriteBuffer;
3499
3500     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
3501     s_readBuffer.insert(s_readBuffer.end(), queries.at(1).begin(), queries.at(1).end());
3502     s_readBuffer.insert(s_readBuffer.end(), queries.at(2).begin(), queries.at(2).end());
3503
3504     /* make sure that the backend's timeout is shorter than the client's */
3505     backend->d_config.tcpConnectTimeout = 1;
3506     g_tcpRecvTimeout = 5;
3507
3508     bool timeout = false;
3509     s_steps = {
3510       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3511       /* reading a query from the client (1) */
3512       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3513       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3514       /* opening a connection to the backend */
3515       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3516       /* backend is not ready yet */
3517       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::NeedWrite, 0 },
3518       /* reading a second query from the client */
3519       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3520       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3521       /* reading a third query from the client */
3522       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3523       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2, [&timeout](int desc) {
3524         timeout = true;
3525       } },
3526       /* trying to read more from the client but nothing to read */
3527       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
3528       /* closing the client connection */
3529       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done, 0 },
3530       /* closing the backend connection */
3531       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done, 0 },
3532     };
3533
3534     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3535       selectedBackend = backend;
3536       return ProcessQueryResult::PassToBackend;
3537     };
3538     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3539       return true;
3540     };
3541
3542     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3543     IncomingTCPConnectionState::handleIO(state, now);
3544     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
3545       threadData.mplexer->run(&now);
3546     }
3547
3548     struct timeval later = now;
3549     later.tv_sec += backend->d_config.tcpConnectTimeout + 1;
3550     auto expiredConns = threadData.mplexer->getTimeouts(later, true);
3551     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
3552     for (const auto& cbData : expiredConns) {
3553       if (cbData.second.type() == typeid(std::shared_ptr<TCPConnectionToBackend>)) {
3554         auto cbState = boost::any_cast<std::shared_ptr<TCPConnectionToBackend>>(cbData.second);
3555         cbState->handleTimeout(later, true);
3556       }
3557     }
3558
3559     BOOST_CHECK_EQUAL(s_writeBuffer.size(), 0U);
3560     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), 0U);
3561     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
3562
3563     /* restore */
3564     backend->d_config.tcpSendTimeout = 30;
3565     g_tcpRecvTimeout = 2;
3566
3567     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3568     /* we have one connection to clear, no proxy protocol */
3569     BOOST_CHECK_EQUAL(IncomingTCPConnectionState::clearAllDownstreamConnections(), 1U);
3570   }
3571
3572   {
3573     TEST_INIT("=> 5 OOOR queries, backend only accepts two at a time");
3574     PacketBuffer expectedWriteBuffer;
3575     PacketBuffer expectedBackendWriteBuffer;
3576
3577     for (const auto& query : queries) {
3578       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
3579     }
3580
3581     /* queries 0, 1 and 4 are sent to the first backend, 2 and 3 to the second */
3582     uint16_t firstBackendCounter = 0;
3583     uint16_t secondBackendCounter = 0;
3584     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), firstBackendCounter++);
3585     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), firstBackendCounter++);
3586     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), secondBackendCounter++);
3587     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(3), secondBackendCounter++);
3588     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(4), firstBackendCounter++);
3589
3590     firstBackendCounter = 0;
3591     secondBackendCounter = 0;
3592     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), firstBackendCounter++);
3593     appendPayloadEditingID(s_backendReadBuffer, responses.at(1), firstBackendCounter++);
3594     appendPayloadEditingID(s_backendReadBuffer, responses.at(2), secondBackendCounter++);
3595     appendPayloadEditingID(s_backendReadBuffer, responses.at(4), firstBackendCounter++);
3596     appendPayloadEditingID(s_backendReadBuffer, responses.at(3), secondBackendCounter++);
3597
3598     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(0).begin(), responses.at(0).end());
3599     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(1).begin(), responses.at(1).end());
3600     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(2).begin(), responses.at(2).end());
3601     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(4).begin(), responses.at(4).end());
3602     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(3).begin(), responses.at(3).end());
3603
3604     auto backend1 = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
3605     backend1->d_tlsCtx = tlsCtx;
3606     /* only two queries in flight! */
3607     backend1->d_config.d_maxInFlightQueriesPerConn = 2;
3608
3609     int backend1Desc = -1;
3610     int backend2Desc = -1;
3611
3612     s_steps = {
3613       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3614       /* reading a query from the client (1) */
3615       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3616       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3617       /* opening a connection to the backend (1) */
3618       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backend1Desc](int desc) {
3619         backend1Desc = desc;
3620       } },
3621       /* sending query (1) to the backend */
3622       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
3623       /* no response ready yet */
3624       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3625       /* reading a query from the client (2) */
3626       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3627       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3628       /* sending query (2) to the backend */
3629       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
3630       /* no response ready yet */
3631       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3632       /* reading a query from the client (3) */
3633       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3634       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
3635       /* opening a connection to the SECOND backend (2) */
3636       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backend2Desc](int desc) {
3637         backend2Desc = desc;
3638       } },
3639       /* sending query (3) to backend 2 */
3640       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
3641       /* no response ready yet */
3642       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3643       /* reading a query from the client (4) */
3644       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3645       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(3).size() - 2 },
3646       /* sending query to the second backend */
3647       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(3).size() },
3648       /* no response ready yet */
3649       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3650       /* nothing more to read from the client at that moment */
3651       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0, [&threadData, &backend1Desc](int desc) {
3652         /* but the first backend becomes readable */
3653         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backend1Desc);
3654       } },
3655       /* reading response (1) from the first backend (1) */
3656       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3657       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
3658       /* sending it to the client */
3659       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
3660         /* client becomes readable */
3661         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3662       } },
3663       /* reading a query from the client (5) */
3664       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3665       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2, [&threadData](int desc) {
3666         /* client is not ready anymore */
3667         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
3668       }  },
3669       /* sending query (5) to the first backend (1) */
3670       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
3671       /* no response ready yet, but the first backend becomes ready */
3672       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
3673         /* set the outgoing descriptor (backend connection) as ready */
3674         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3675       } },
3676       /* trying to read from client, nothing yet */
3677       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
3678       /* reading response (2) from the first backend (1) */
3679       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3680       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
3681       /* sending it to the client */
3682       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size(), [&threadData,&backend1Desc,&backend2Desc](int desc) {
3683         /* client is NOT readable, backend1 is not readable, backend 2 becomes readable */
3684         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
3685         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(backend1Desc);
3686         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backend2Desc);
3687       } },
3688       /* reading response (3) from the second backend (2) */
3689       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3690       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() - 2 },
3691       /* sending it to the client */
3692       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(2).size(), [&threadData,&backend1Desc,&backend2Desc](int desc) {
3693         /* backend 2 is no longer readable, backend 1 becomes readable */
3694         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(backend2Desc);
3695         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backend1Desc);
3696       } },
3697       /* reading response (5) from the first backend (1) */
3698       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3699       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2 },
3700       /* sending it to the client */
3701       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size(), [&threadData,&backend1Desc,&backend2Desc](int desc) {
3702         /* backend 1 is no longer readable, backend 2 becomes readable */
3703         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(backend1Desc);
3704         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backend2Desc);
3705       } },
3706       /* reading response (4) from the second backend (2) */
3707       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3708       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(3).size() - 2 },
3709       /* sending it to the client */
3710       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(3).size(), [&threadData,&backend2Desc](int desc) {
3711         /* backend 2 is no longer readable */
3712         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(backend2Desc);
3713         /* client becomes readable */
3714         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
3715       } },
3716       /* client closes the connection */
3717       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
3718       /* closing client connection */
3719       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
3720       /* closing a connection to the backends */
3721       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
3722       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
3723     };
3724
3725     s_processQuery = [backend1](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3726       selectedBackend = backend1;
3727       return ProcessQueryResult::PassToBackend;
3728     };
3729     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3730       return true;
3731     };
3732
3733     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3734     IncomingTCPConnectionState::handleIO(state, now);
3735     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
3736       threadData.mplexer->run(&now);
3737     }
3738
3739     BOOST_CHECK_EQUAL(s_writeBuffer.size(), totalResponsesSize);
3740     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3741     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), totalQueriesSize);
3742     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3743     BOOST_CHECK_EQUAL(backend1->outstanding.load(), 0U);
3744
3745     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3746     BOOST_CHECK_EQUAL(IncomingTCPConnectionState::clearAllDownstreamConnections(), 2U);
3747   }
3748
3749   {
3750     TEST_INIT("=> 2 OOOR queries to the backend with duplicated IDs");
3751     PacketBuffer expectedWriteBuffer;
3752     PacketBuffer expectedBackendWriteBuffer;
3753
3754     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
3755     s_readBuffer.insert(s_readBuffer.end(), queries.at(0).begin(), queries.at(0).end());
3756
3757     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), 0);
3758     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), 1);
3759
3760     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 0);
3761     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 1);
3762
3763     appendPayloadEditingID(expectedWriteBuffer, responses.at(0), 0);
3764     appendPayloadEditingID(expectedWriteBuffer, responses.at(0), 0);
3765
3766     bool timeout = false;
3767     s_steps = {
3768       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3769       /* reading a query from the client (1) */
3770       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3771       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3772       /* opening a connection to the backend */
3773       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done },
3774       /* sending query to the backend */
3775       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
3776       /* no response ready yet */
3777       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3778       /* reading a query from the client (2) */
3779       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3780       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3781       /* sending query to the backend */
3782       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
3783       /* no response ready yet, but mark the descriptor as ready */
3784       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData](int desc) {
3785         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(desc);
3786       } },
3787       /* nothing more from the client either */
3788       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 },
3789
3790       /* reading response (1) from the backend */
3791       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
3792       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size()},
3793       /* sending it to the client */
3794       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size()},
3795       /* reading response (2) from the backend */
3796       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
3797       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size(), [&threadData](int desc) {
3798         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
3799       } },
3800       /* sending it to the client. we don't have anything else to send to the client, no new query from it either, until we time out */
3801       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&timeout](int desc) {
3802         timeout = true;
3803       } },
3804       /* closing client connection */
3805       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
3806       /* closing a connection to the backend */
3807       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
3808     };
3809
3810     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
3811       selectedBackend = backend;
3812       return ProcessQueryResult::PassToBackend;
3813     };
3814     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
3815       return true;
3816     };
3817
3818     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
3819     IncomingTCPConnectionState::handleIO(state, now);
3820     while (!timeout && (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0)) {
3821       threadData.mplexer->run(&now);
3822     }
3823
3824     struct timeval later = now;
3825     later.tv_sec += g_tcpRecvTimeout + 1;
3826     auto expiredConns = threadData.mplexer->getTimeouts(later);
3827     BOOST_CHECK_EQUAL(expiredConns.size(), 1U);
3828     for (const auto& cbData : expiredConns) {
3829       if (cbData.second.type() == typeid(std::shared_ptr<IncomingTCPConnectionState>)) {
3830         auto cbState = boost::any_cast<std::shared_ptr<IncomingTCPConnectionState>>(cbData.second);
3831         cbState->handleTimeout(cbState, false);
3832       }
3833     }
3834
3835     BOOST_CHECK_EQUAL(s_writeBuffer.size(), expectedWriteBuffer.size());
3836     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
3837     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), expectedBackendWriteBuffer.size());
3838     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
3839     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
3840
3841     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
3842     IncomingTCPConnectionState::clearAllDownstreamConnections();
3843   }
3844 }
3845
3846 BOOST_AUTO_TEST_CASE(test_IncomingConnectionOOOR_BackendNotOOOR)
3847 {
3848   auto local = getBackendAddress("1", 80);
3849   ClientState localCS(local, true, false, false, "", {});
3850   /* enable out-of-order on the front side */
3851   localCS.d_maxInFlightQueriesPerConn = 65536;
3852
3853   auto tlsCtx = std::make_shared<MockupTLSCtx>();
3854   localCS.tlsFrontend = std::make_shared<TLSFrontend>(tlsCtx);
3855
3856   auto backend = std::make_shared<DownstreamState>(getBackendAddress("42", 53));
3857   backend->d_tlsCtx = tlsCtx;
3858   /* shorter than the client one */
3859   backend->d_config.tcpRecvTimeout = 1;
3860
3861   TCPClientThreadData threadData;
3862   threadData.mplexer = std::make_unique<MockupFDMultiplexer>();
3863
3864   struct timeval now;
3865   gettimeofday(&now, nullptr);
3866
3867   std::vector<PacketBuffer> queries(5);
3868   std::vector<PacketBuffer> responses(5);
3869
3870   size_t counter = 0;
3871   size_t totalQueriesSize = 0;
3872   for (auto& query : queries) {
3873     GenericDNSPacketWriter<PacketBuffer> pwQ(query, DNSName("powerdns" + std::to_string(counter) + ".com."), QType::A, QClass::IN, 0);
3874     pwQ.getHeader()->rd = 1;
3875     pwQ.getHeader()->id = counter;
3876     uint16_t querySize = static_cast<uint16_t>(query.size());
3877     const uint8_t sizeBytes[] = { static_cast<uint8_t>(querySize / 256), static_cast<uint8_t>(querySize % 256) };
3878     query.insert(query.begin(), sizeBytes, sizeBytes + 2);
3879     totalQueriesSize += query.size();
3880     ++counter;
3881   }
3882
3883   counter = 0;
3884   size_t totalResponsesSize = 0;
3885   for (auto& response : responses) {
3886     DNSName name("powerdns" + std::to_string(counter) + ".com.");
3887     GenericDNSPacketWriter<PacketBuffer> pwR(response, name, QType::A, QClass::IN, 0);
3888     pwR.getHeader()->qr = 1;
3889     pwR.getHeader()->rd = 1;
3890     pwR.getHeader()->ra = 1;
3891     pwR.getHeader()->id = counter;
3892     pwR.startRecord(name, QType::A, 7200, QClass::IN, DNSResourceRecord::ANSWER);
3893     pwR.xfr32BitInt(0x01020304);
3894     pwR.commit();
3895
3896     uint16_t responseSize = static_cast<uint16_t>(response.size());
3897     const uint8_t sizeBytes[] = { static_cast<uint8_t>(responseSize / 256), static_cast<uint8_t>(responseSize % 256) };
3898     response.insert(response.begin(), sizeBytes, sizeBytes + 2);
3899     totalResponsesSize += response.size();
3900     ++counter;
3901   }
3902
3903   {
3904     TEST_INIT("=> 5 OOOR queries, we will need to open 5 backend connections");
3905     PacketBuffer expectedWriteBuffer;
3906     PacketBuffer expectedBackendWriteBuffer;
3907
3908     for (const auto& query : queries) {
3909       s_readBuffer.insert(s_readBuffer.end(), query.begin(), query.end());
3910     }
3911
3912     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(0), 0);
3913     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(1), 0);
3914     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(2), 0);
3915     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(3), 0);
3916     appendPayloadEditingID(expectedBackendWriteBuffer, queries.at(4), 0);
3917
3918     appendPayloadEditingID(s_backendReadBuffer, responses.at(0), 0);
3919     appendPayloadEditingID(s_backendReadBuffer, responses.at(2), 0);
3920     appendPayloadEditingID(s_backendReadBuffer, responses.at(1), 0);
3921     appendPayloadEditingID(s_backendReadBuffer, responses.at(4), 0);
3922     appendPayloadEditingID(s_backendReadBuffer, responses.at(3), 0);
3923
3924     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(0).begin(), responses.at(0).end());
3925     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(2).begin(), responses.at(2).end());
3926     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(1).begin(), responses.at(1).end());
3927     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(4).begin(), responses.at(4).end());
3928     expectedWriteBuffer.insert(expectedWriteBuffer.end(), responses.at(3).begin(), responses.at(3).end());
3929
3930     std::vector<int> backendDescriptors = { -1, -1, -1, -1, -1 };
3931
3932     s_steps = {
3933       { ExpectedStep::ExpectedRequest::handshakeClient, IOState::Done },
3934       /* reading a query from the client (1) */
3935       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3936       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(0).size() - 2 },
3937       /* opening a connection to the backend (1) */
3938       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptors](int desc) {
3939         backendDescriptors.at(0) = desc;
3940       } },
3941       /* sending query (1) to the backend */
3942       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(0).size() },
3943       /* no response ready yet */
3944       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3945       /* reading a query from the client (2) */
3946       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3947       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(1).size() - 2 },
3948       /* opening a connection to the backend (2) */
3949       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptors](int desc) {
3950         backendDescriptors.at(1) = desc;
3951       } },
3952       /* sending query (2) to the backend */
3953       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(1).size() },
3954       /* no response ready yet */
3955       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3956       /* reading a query from the client (3) */
3957       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3958       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(2).size() - 2 },
3959       /* opening a connection to the backend (3) */
3960       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptors](int desc) {
3961         backendDescriptors.at(2) = desc;
3962       } },
3963       /* sending query (3) to the backend */
3964       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(2).size() },
3965       /* no response ready yet */
3966       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3967       /* reading a query from the client (4) */
3968       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3969       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(3).size() - 2 },
3970       /* opening a connection to the backend (4) */
3971       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptors](int desc) {
3972         backendDescriptors.at(3) = desc;
3973       } },
3974       /* sending query (3) to the backend */
3975       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(3).size() },
3976       /* no response ready yet */
3977       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0 },
3978       /* reading a query from the client (5) */
3979       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 2 },
3980       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, queries.at(4).size() - 2 },
3981       /* opening a connection to the backend (5) */
3982       { ExpectedStep::ExpectedRequest::connectToBackend, IOState::Done, 0, [&backendDescriptors](int desc) {
3983         backendDescriptors.at(4) = desc;
3984       } },
3985       /* sending query (5) to the backend */
3986       { ExpectedStep::ExpectedRequest::writeToBackend, IOState::Done, queries.at(4).size() },
3987       /* no response ready yet, client stops being readable, first backend has a response */
3988       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::NeedRead, 0, [&threadData,&backendDescriptors](int desc) {
3989         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptors.at(0));
3990       } },
3991       /* trying to read from the client but nothing yet */
3992       { ExpectedStep::ExpectedRequest::readFromClient, IOState::NeedRead, 0 , [&threadData](int desc) {
3993         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setNotReady(desc);
3994       } },
3995       /* reading response (1) from the first backend (1) */
3996       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
3997       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(0).size() - 2 },
3998       /* sending it to the client */
3999       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(0).size(), [&threadData,&backendDescriptors](int desc) {
4000         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptors.at(2));
4001       } },
4002       /* reading response (3) from the third backend (3) */
4003       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
4004       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(2).size() - 2 },
4005       /* sending it to the client */
4006       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(2).size(), [&threadData,&backendDescriptors](int desc) {
4007         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptors.at(1));
4008       } },
4009       /* reading response (2) from the second backend (2) */
4010       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
4011       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(1).size() - 2 },
4012       /* sending it to the client */
4013       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(1).size(), [&threadData,&backendDescriptors](int desc) {
4014         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptors.at(4));
4015       } },
4016       /* reading response (5) from the fifth backend (5) */
4017       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
4018       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(4).size() - 2 },
4019       /* sending it to the client */
4020       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(4).size(), [&threadData,&backendDescriptors](int desc) {
4021         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(backendDescriptors.at(3));
4022       } },
4023       /* reading response (4) from the fourth backend (4) */
4024       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, 2 },
4025       { ExpectedStep::ExpectedRequest::readFromBackend, IOState::Done, responses.at(3).size() - 2 },
4026       /* sending it to the client */
4027       { ExpectedStep::ExpectedRequest::writeToClient, IOState::Done, responses.at(3).size(), [&threadData](int desc) {
4028         dynamic_cast<MockupFDMultiplexer*>(threadData.mplexer.get())->setReady(-1);
4029       } },
4030       /* client closes the connection */
4031       { ExpectedStep::ExpectedRequest::readFromClient, IOState::Done, 0 },
4032       /* closing client connection */
4033       { ExpectedStep::ExpectedRequest::closeClient, IOState::Done },
4034       /* closing a connection to the backends */
4035       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
4036       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
4037       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
4038       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
4039       { ExpectedStep::ExpectedRequest::closeBackend, IOState::Done },
4040     };
4041
4042     s_processQuery = [backend](DNSQuestion& dq, ClientState& cs, LocalHolders& holders, std::shared_ptr<DownstreamState>& selectedBackend) -> ProcessQueryResult {
4043       selectedBackend = backend;
4044       return ProcessQueryResult::PassToBackend;
4045     };
4046     s_processResponse = [](PacketBuffer& response, LocalStateHolder<vector<DNSDistResponseRuleAction> >& localRespRuleActions, DNSResponse& dr, bool muted) -> bool {
4047       return true;
4048     };
4049
4050     auto state = std::make_shared<IncomingTCPConnectionState>(ConnectionInfo(&localCS, getBackendAddress("84", 4242)), threadData, now);
4051     IncomingTCPConnectionState::handleIO(state, now);
4052     while (threadData.mplexer->getWatchedFDCount(false) != 0 || threadData.mplexer->getWatchedFDCount(true) != 0) {
4053       threadData.mplexer->run(&now);
4054     }
4055
4056     BOOST_CHECK_EQUAL(s_writeBuffer.size(), totalResponsesSize);
4057     BOOST_CHECK(s_writeBuffer == expectedWriteBuffer);
4058     BOOST_CHECK_EQUAL(s_backendWriteBuffer.size(), totalQueriesSize);
4059     BOOST_CHECK(s_backendWriteBuffer == expectedBackendWriteBuffer);
4060     BOOST_CHECK_EQUAL(backend->outstanding.load(), 0U);
4061
4062     /* we need to clear them now, otherwise we end up with dangling pointers to the steps via the TLS context, etc */
4063     BOOST_CHECK_EQUAL(IncomingTCPConnectionState::clearAllDownstreamConnections(), 5U);
4064   }
4065 }
4066
4067 BOOST_AUTO_TEST_SUITE_END();