19741624072

Committed 27 Nov 2025 03:45PM UTC coverage: 73.086% (+0.02%) from 73.065%

Build # 19741624072

Build Type

Pull #16570

github

Committed by

web-flow

Commit Message

Merge 08a2cdb1d into f94a3f63f

Pull Request Pull Request #16570: rec: rewrite all unwrap calls in web.rs

Run Details

38523 of 63408 branches covered (60.75%)

Branch coverage included in aggregate %.

128044 of 164496 relevant lines covered (77.84%)

6531485.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.89

/pdns/dnsdistdist/dnsdist-tcp-downstream.hh

#pragma once

#include "sstuff.hh"
#include "tcpiohandler-mplexer.hh"
#include "dnsdist.hh"
#include "dnsdist-tcp.hh"

class ConnectionToBackend : public std::enable_shared_from_this<ConnectionToBackend>
{
public:
  ConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now): d_connectionStartTime(now), d_lastDataReceivedTime(now), d_ds(ds), d_mplexer(mplexer), d_enableFastOpen(ds->d_config.tcpFastOpen)
  {
    reconnect();
  }

  virtual ~ConnectionToBackend();

  int getHandle() const
  {
    if (!d_handler) {
      throw std::runtime_error("Attempt to get the socket handle from a non-established TCP connection");
    }

    return d_handler->getDescriptor();
  }

  /* whether the underlying socket has been closed under our feet, basically */
  bool isUsable()
  {
    if (!d_handler) {
      d_connectionDied = true;
      return false;
    }

    if (d_handler->isUsable()) {
      return true;
    }

    d_connectionDied = true;
    return false;
  }

  const std::shared_ptr<DownstreamState>& getDS() const
  {
    return d_ds;
  }

  const ComboAddress& getRemote() const
  {
    return d_ds->d_config.remote;
  }

  const std::string& getBackendName() const
  {
    return d_ds->getName();
  }

  bool isFresh() const
  {
    return d_fresh;
  }

  void setReused()
  {
    d_fresh = false;
  }

  void disableFastOpen()
  {
    d_enableFastOpen = false;
  }

  bool isFastOpenEnabled()
  {
    return d_enableFastOpen;
  }

  /* whether a connection can be used now */
  bool canBeReused(bool sameClient = false) const
  {
    if (d_connectionDied) {
      return false;
    }

    /* we can't reuse a connection where a proxy protocol payload has been sent,
       since:
       - it cannot be reused for a different client
       - we might have different TLV values for each query
    */
    if (d_ds && d_ds->d_config.useProxyProtocol == true && !sameClient) {
      return false;
    }

    if (reachedMaxStreamID()) {
      return false;
    }

    if (reachedMaxConcurrentQueries()) {
      return false;
    }

    return true;
  }

  /* full now but will become usable later */
  bool willBeReusable(bool sameClient) const
  {
    if (d_connectionDied || reachedMaxStreamID()) {
      return false;
    }

    if (d_ds && d_ds->d_config.useProxyProtocol == true) {
      return sameClient;
    }

    return true;
  }

  virtual bool reachedMaxStreamID() const = 0;
  virtual bool reachedMaxConcurrentQueries() const = 0;
  virtual bool isIdle() const = 0;
  virtual void release(bool removeFromCache) = 0;
  virtual void stopIO()
  {
  }

  bool matches(const std::shared_ptr<DownstreamState>& ds) const
  {
    if (!ds || !d_ds) {
      return false;
    }
    return ds == d_ds;
  }

  virtual void queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query) = 0;
  virtual void handleTimeout(const struct timeval& now, bool write) = 0;

  struct timeval getLastDataReceivedTime() const
  {
    return d_lastDataReceivedTime;
  }

  virtual std::string toString() const = 0;

protected:
  bool reconnect();

  boost::optional<struct timeval> getBackendHealthCheckTTD(const struct timeval& now) const
  {
    if (d_ds == nullptr) {
      throw std::runtime_error("getBackendReadTTD() without any backend selected");
    }
    if (d_ds->d_config.checkTimeout == 0) {
      return boost::none;
    }

    struct timeval res = now;
    res.tv_sec += d_ds->d_config.checkTimeout / 1000; /* ms to s */
    res.tv_usec += (d_ds->d_config.checkTimeout % 1000) * 1000; /* remaining ms to µs */
    normalizeTV(res);

    return res;
  }

  boost::optional<struct timeval> getBackendReadTTD(const struct timeval& now) const
  {
    if (d_ds == nullptr) {
      throw std::runtime_error("getBackendReadTTD() without any backend selected");
    }
    if (d_ds->d_config.tcpRecvTimeout == 0) {
      return boost::none;
    }

    struct timeval res = now;
    res.tv_sec += d_ds->d_config.tcpRecvTimeout;

    return res;
  }

  boost::optional<struct timeval> getBackendWriteTTD(const struct timeval& now) const
  {
    if (d_ds == nullptr) {
      throw std::runtime_error("getBackendWriteTTD() called without any backend selected");
    }
    if (d_ds->d_config.tcpSendTimeout == 0) {
      return boost::none;
    }

    struct timeval res = now;
    res.tv_sec += d_ds->d_config.tcpSendTimeout;

    return res;
  }

  boost::optional<struct timeval> getBackendConnectTTD(const struct timeval& now) const
  {
    if (d_ds == nullptr) {
      throw std::runtime_error("getBackendConnectTTD() called without any backend selected");
    }
    if (d_ds->d_config.tcpConnectTimeout == 0) {
      return boost::none;
    }

    struct timeval res = now;
    res.tv_sec += d_ds->d_config.tcpConnectTimeout;

    return res;
  }

  struct timeval d_connectionStartTime;
  struct timeval d_lastDataReceivedTime;
  const std::shared_ptr<DownstreamState> d_ds{nullptr};
  std::shared_ptr<TCPQuerySender> d_sender{nullptr};
  std::unique_ptr<FDMultiplexer>& d_mplexer;
  std::unique_ptr<TCPIOHandler> d_handler{nullptr};
  std::unique_ptr<IOStateHandler> d_ioState{nullptr};
  uint64_t d_queries{0};
  uint32_t d_highestStreamID{0};
  uint16_t d_downstreamFailures{0};
  bool d_proxyProtocolPayloadSent{false};
  bool d_enableFastOpen{false};
  bool d_connectionDied{false};
  bool d_fresh{true};
};

class TCPConnectionToBackend : public ConnectionToBackend
{
public:
  TCPConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now, std::string&& /* proxyProtocolPayload*, unused but there to match the HTTP2 connections, so we can use the same templated connections manager class */): ConnectionToBackend(ds, mplexer, now), d_responseBuffer(512)
  {
  }

  virtual ~TCPConnectionToBackend();

  bool isIdle() const override
  {
    return d_state == State::idle && d_pendingQueries.size() == 0 && d_pendingResponses.size() == 0;
  }

  bool reachedMaxStreamID() const override
  {
    /* TCP/DoT has only 2^16 usable identifiers, DoH has 2^32 */
    const uint32_t maximumStreamID = std::numeric_limits<uint16_t>::max() - 1;
    return d_highestStreamID == maximumStreamID;
  }

  bool reachedMaxConcurrentQueries() const override
  {
    const size_t concurrent = d_pendingQueries.size() + d_pendingResponses.size() + (d_state == State::sendingQueryToBackend ? 1 : 0);
    if (concurrent > 0 && concurrent >= d_ds->d_config.d_maxInFlightQueriesPerConn) {
      return true;
    }
    return false;
  }
  bool matchesTLVs(const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs) const;

  void queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query) override;
  void handleTimeout(const struct timeval& now, bool write) override;
  void release(bool removeFromCache) override;

  std::string toString() const override
  {
    ostringstream o;
    o << "TCP connection to backend "<<(d_ds ? d_ds->getName() : "empty")<<" over FD "<<(d_handler ? std::to_string(d_handler->getDescriptor()) : "no socket")<<", state is "<<(int)d_state<<", io state is "<<(d_ioState ? d_ioState->getState() : "empty")<<", queries count is "<<d_queries<<", pending queries count is "<<d_pendingQueries.size()<<", "<<d_pendingResponses.size()<<" pending responses";
    return o.str();
  }

  void setProxyProtocolValuesSent(std::unique_ptr<std::vector<ProxyProtocolValue>>&& proxyProtocolValuesSent);

private:
  /* waitingForResponseFromBackend is a state where we have not yet started reading the size,
     so we can still switch to sending instead */
  enum class State : uint8_t { idle, sendingQueryToBackend, waitingForResponseFromBackend, readingResponseSizeFromBackend, readingResponseFromBackend };
  enum class FailureReason : uint8_t { /* too many attempts */ gaveUp, timeout, unexpectedQueryID };

  static void handleIO(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
  static void handleIOCallback(int fd, FDMultiplexer::funcparam_t& param);
  static void handleReconnectionAttempt(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now, IOStateGuard& ioGuard, IOState& iostate, bool& reconnected, bool& connectionDied);
  static IOState queueNextQuery(std::shared_ptr<TCPConnectionToBackend>& conn);
  static IOState sendQuery(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
  static bool isXFRFinished(const TCPResponse& response, TCPQuery& query);

  IOState handleResponse(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
  uint16_t getQueryIdFromResponse() const;
  void notifyAllQueriesFailed(const struct timeval& now, FailureReason reason);
  bool needProxyProtocolPayload() const
  {
    return !d_proxyProtocolPayloadSent && (d_ds && d_ds->d_config.useProxyProtocol);
  }

  class PendingRequest
  {
  public:
    std::shared_ptr<TCPQuerySender> d_sender{nullptr};
    TCPQuery d_query;
  };

  PacketBuffer d_responseBuffer;
  std::list<PendingRequest> d_pendingQueries;
  std::unordered_map<uint16_t, PendingRequest> d_pendingResponses;
  std::unique_ptr<std::vector<ProxyProtocolValue>> d_proxyProtocolValuesSent{nullptr};
  PendingRequest d_currentQuery;
  size_t d_currentPos{0};
  uint16_t d_responseSize{0};
  State d_state{State::idle};
  bool d_handlingIO{false};
  bool d_lastIOBlocked{false};
};

void setTCPDownstreamMaxIdleConnectionsPerBackend(uint64_t max);
void setTCPDownstreamCleanupInterval(uint64_t interval);
void setTCPDownstreamMaxIdleTime(uint64_t max);

1	#pragma once
2
3	#include "sstuff.hh"
4	#include "tcpiohandler-mplexer.hh"
5	#include "dnsdist.hh"
6	#include "dnsdist-tcp.hh"
7
8	class ConnectionToBackend : public std::enable_shared_from_this<ConnectionToBackend>
9	{
10	public:
11	ConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now): d_connectionStartTime(now), d_lastDataReceivedTime(now), d_ds(ds), d_mplexer(mplexer), d_enableFastOpen(ds->d_config.tcpFastOpen)	1,095✔
12	{	1,095✔
13	reconnect();	1,095✔
14	}	1,095✔
15
16	virtual ~ConnectionToBackend();
17
18	int getHandle() const
19	{	3,903✔
20	if (!d_handler) {	3,903!
21	throw std::runtime_error("Attempt to get the socket handle from a non-established TCP connection");	×
22	}	×
23
24	return d_handler->getDescriptor();	3,903✔
25	}	3,903✔
26
27	/* whether the underlying socket has been closed under our feet, basically */
28	bool isUsable()
29	{	98✔
30	if (!d_handler) {	98!
31	d_connectionDied = true;	×
32	return false;	×
33	}	×
34
35	if (d_handler->isUsable()) {	98✔
36	return true;	4✔
37	}	4✔
38
39	d_connectionDied = true;	94✔
40	return false;	94✔
41	}	98✔
42
43	const std::shared_ptr<DownstreamState>& getDS() const
44	{	164,061✔
45	return d_ds;	164,061✔
46	}	164,061✔
47
48	const ComboAddress& getRemote() const
49	{	×
50	return d_ds->d_config.remote;	×
51	}	×
52
53	const std::string& getBackendName() const
54	{	21✔
55	return d_ds->getName();	21✔
56	}	21✔
57
58	bool isFresh() const
59	{	2,003✔
60	return d_fresh;	2,003✔
61	}	2,003✔
62
63	void setReused()
64	{	32,138✔
65	d_fresh = false;	32,138✔
66	}	32,138✔
67
68	void disableFastOpen()
69	{	×
70	d_enableFastOpen = false;	×
71	}	×
72
73	bool isFastOpenEnabled()
74	{	5✔
75	return d_enableFastOpen;	5✔
76	}	5✔
77
78	/* whether a connection can be used now */
79	bool canBeReused(bool sameClient = false) const
80	{	32,346✔
81	if (d_connectionDied) {	32,346✔
82	return false;	2✔
83	}	2✔
84
85	/* we can't reuse a connection where a proxy protocol payload has been sent,
86	since:
87	- it cannot be reused for a different client
88	- we might have different TLV values for each query
89	*/
90	if (d_ds && d_ds->d_config.useProxyProtocol == true && !sameClient) {	32,344!
91	return false;	×
92	}	×
93
94	if (reachedMaxStreamID()) {	32,344!
95	return false;	×
96	}	×
97
98	if (reachedMaxConcurrentQueries()) {	32,344✔
99	return false;	107✔
100	}	107✔
101
102	return true;	32,237✔
103	}	32,344✔
104
105	/* full now but will become usable later */
106	bool willBeReusable(bool sameClient) const
107	{	1,002✔
108	if (d_connectionDied \|\| reachedMaxStreamID()) {	1,002!
109	return false;	235✔
110	}	235✔
111
112	if (d_ds && d_ds->d_config.useProxyProtocol == true) {	767!
113	return sameClient;	104✔
114	}	104✔
115
116	return true;	663✔
117	}	767✔
118
119	virtual bool reachedMaxStreamID() const = 0;
120	virtual bool reachedMaxConcurrentQueries() const = 0;
121	virtual bool isIdle() const = 0;
122	virtual void release(bool removeFromCache) = 0;
123	virtual void stopIO()
124	{	69✔
125	}	69✔
126
127	bool matches(const std::shared_ptr<DownstreamState>& ds) const
128	{	×
129	if (!ds \|\| !d_ds) {	×
130	return false;	×
131	}	×
132	return ds == d_ds;	×
133	}	×
134
135	virtual void queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query) = 0;
136	virtual void handleTimeout(const struct timeval& now, bool write) = 0;
137
138	struct timeval getLastDataReceivedTime() const
139	{	32,196✔
140	return d_lastDataReceivedTime;	32,196✔
141	}	32,196✔
142
143	virtual std::string toString() const = 0;
144
145	protected:
146	bool reconnect();
147
148	boost::optional<struct timeval> getBackendHealthCheckTTD(const struct timeval& now) const
149	{	42✔
150	if (d_ds == nullptr) {	42!
151	throw std::runtime_error("getBackendReadTTD() without any backend selected");	×
152	}	×
153	if (d_ds->d_config.checkTimeout == 0) {	42!
154	return boost::none;	×
155	}	×
156
157	struct timeval res = now;	42✔
158	res.tv_sec += d_ds->d_config.checkTimeout / 1000; /* ms to s */	42✔
159	res.tv_usec += (d_ds->d_config.checkTimeout % 1000) * 1000; /* remaining ms to µs */	42✔
160	normalizeTV(res);	42✔
161
162	return res;	42✔
163	}	42✔
164
165	boost::optional<struct timeval> getBackendReadTTD(const struct timeval& now) const
166	{	4,804✔
167	if (d_ds == nullptr) {	4,804!
168	throw std::runtime_error("getBackendReadTTD() without any backend selected");	×
169	}	×
170	if (d_ds->d_config.tcpRecvTimeout == 0) {	4,804!
171	return boost::none;	×
172	}	×
173
174	struct timeval res = now;	4,804✔
175	res.tv_sec += d_ds->d_config.tcpRecvTimeout;	4,804✔
176
177	return res;	4,804✔
178	}	4,804✔
179
180	boost::optional<struct timeval> getBackendWriteTTD(const struct timeval& now) const
181	{	493✔
182	if (d_ds == nullptr) {	493!
183	throw std::runtime_error("getBackendWriteTTD() called without any backend selected");	×
184	}	×
185	if (d_ds->d_config.tcpSendTimeout == 0) {	493!
186	return boost::none;	×
187	}	×
188
189	struct timeval res = now;	493✔
190	res.tv_sec += d_ds->d_config.tcpSendTimeout;	493✔
191
192	return res;	493✔
193	}	493✔
194
195	boost::optional<struct timeval> getBackendConnectTTD(const struct timeval& now) const
196	{	1,504✔
197	if (d_ds == nullptr) {	1,504!
198	throw std::runtime_error("getBackendConnectTTD() called without any backend selected");	×
199	}	×
200	if (d_ds->d_config.tcpConnectTimeout == 0) {	1,504!
201	return boost::none;	×
202	}	×
203
204	struct timeval res = now;	1,504✔
205	res.tv_sec += d_ds->d_config.tcpConnectTimeout;	1,504✔
206
207	return res;	1,504✔
208	}	1,504✔
209
210	struct timeval d_connectionStartTime;
211	struct timeval d_lastDataReceivedTime;
212	const std::shared_ptr<DownstreamState> d_ds{nullptr};
213	std::shared_ptr<TCPQuerySender> d_sender{nullptr};
214	std::unique_ptr<FDMultiplexer>& d_mplexer;
215	std::unique_ptr<TCPIOHandler> d_handler{nullptr};
216	std::unique_ptr<IOStateHandler> d_ioState{nullptr};
217	uint64_t d_queries{0};
218	uint32_t d_highestStreamID{0};
219	uint16_t d_downstreamFailures{0};
220	bool d_proxyProtocolPayloadSent{false};
221	bool d_enableFastOpen{false};
222	bool d_connectionDied{false};
223	bool d_fresh{true};
224	};
225
226	class TCPConnectionToBackend : public ConnectionToBackend
227	{
228	public:
229	TCPConnectionToBackend(const std::shared_ptr<DownstreamState>& ds, std::unique_ptr<FDMultiplexer>& mplexer, const struct timeval& now, std::string&& /* proxyProtocolPayload, unused but there to match the HTTP2 connections, so we can use the same templated connections manager class /): ConnectionToBackend(ds, mplexer, now), d_responseBuffer(512)	1,010✔
230	{	1,010✔
231	}	1,010✔
232
233	virtual ~TCPConnectionToBackend();
234
235	bool isIdle() const override
236	{	×
237	return d_state == State::idle && d_pendingQueries.size() == 0 && d_pendingResponses.size() == 0;	×
238	}	×
239
240	bool reachedMaxStreamID() const override
241	{	32,418✔
242	/* TCP/DoT has only 2^16 usable identifiers, DoH has 2^32 */
243	const uint32_t maximumStreamID = std::numeric_limits<uint16_t>::max() - 1;	32,418✔
244	return d_highestStreamID == maximumStreamID;	32,418✔
245	}	32,418✔
246
247	bool reachedMaxConcurrentQueries() const override
248	{	32,240✔
249	const size_t concurrent = d_pendingQueries.size() + d_pendingResponses.size() + (d_state == State::sendingQueryToBackend ? 1 : 0);	32,240✔
250	if (concurrent > 0 && concurrent >= d_ds->d_config.d_maxInFlightQueriesPerConn) {	32,240✔
251	return true;	107✔
252	}	107✔
253	return false;	32,133✔
254	}	32,240✔
255	bool matchesTLVs(const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs) const;
256
257	void queueQuery(std::shared_ptr<TCPQuerySender>& sender, TCPQuery&& query) override;
258	void handleTimeout(const struct timeval& now, bool write) override;
259	void release(bool removeFromCache) override;
260
261	std::string toString() const override
262	{	×
263	ostringstream o;	×
264	o << "TCP connection to backend "<<(d_ds ? d_ds->getName() : "empty")<<" over FD "<<(d_handler ? std::to_string(d_handler->getDescriptor()) : "no socket")<<", state is "<<(int)d_state<<", io state is "<<(d_ioState ? d_ioState->getState() : "empty")<<", queries count is "<<d_queries<<", pending queries count is "<<d_pendingQueries.size()<<", "<<d_pendingResponses.size()<<" pending responses";	×
265	return o.str();	×
266	}	×
267
268	void setProxyProtocolValuesSent(std::unique_ptr<std::vector<ProxyProtocolValue>>&& proxyProtocolValuesSent);
269
270	private:
271	/* waitingForResponseFromBackend is a state where we have not yet started reading the size,
272	so we can still switch to sending instead */
273	enum class State : uint8_t { idle, sendingQueryToBackend, waitingForResponseFromBackend, readingResponseSizeFromBackend, readingResponseFromBackend };
274	enum class FailureReason : uint8_t { /* too many attempts */ gaveUp, timeout, unexpectedQueryID };
275
276	static void handleIO(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
277	static void handleIOCallback(int fd, FDMultiplexer::funcparam_t& param);
278	static void handleReconnectionAttempt(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now, IOStateGuard& ioGuard, IOState& iostate, bool& reconnected, bool& connectionDied);
279	static IOState queueNextQuery(std::shared_ptr<TCPConnectionToBackend>& conn);
280	static IOState sendQuery(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
281	static bool isXFRFinished(const TCPResponse& response, TCPQuery& query);
282
283	IOState handleResponse(std::shared_ptr<TCPConnectionToBackend>& conn, const struct timeval& now);
284	uint16_t getQueryIdFromResponse() const;
285	void notifyAllQueriesFailed(const struct timeval& now, FailureReason reason);
286	bool needProxyProtocolPayload() const
287	{	34,498✔
288	return !d_proxyProtocolPayloadSent && (d_ds && d_ds->d_config.useProxyProtocol);	34,498!
289	}	34,498✔
290
291	class PendingRequest
292	{
293	public:
294	std::shared_ptr<TCPQuerySender> d_sender{nullptr};
295	TCPQuery d_query;
296	};
297
298	PacketBuffer d_responseBuffer;
299	std::list<PendingRequest> d_pendingQueries;
300	std::unordered_map<uint16_t, PendingRequest> d_pendingResponses;
301	std::unique_ptr<std::vector<ProxyProtocolValue>> d_proxyProtocolValuesSent{nullptr};
302	PendingRequest d_currentQuery;
303	size_t d_currentPos{0};
304	uint16_t d_responseSize{0};
305	State d_state{State::idle};
306	bool d_handlingIO{false};
307	bool d_lastIOBlocked{false};
308	};
309
310	void setTCPDownstreamMaxIdleConnectionsPerBackend(uint64_t max);
311	void setTCPDownstreamCleanupInterval(uint64_t interval);
312	void setTCPDownstreamMaxIdleTime(uint64_t max);

PowerDNS / pdns / 19741624072

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous