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