domain/net/client/
stream.rs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
//! A client transport using a stream socket.

// RFC 7766 describes DNS over TCP
// RFC 7828 describes the edns-tcp-keepalive option

use super::request::{
    ComposeRequest, ComposeRequestMulti, Error, GetResponse,
    GetResponseMulti, SendRequest, SendRequestMulti,
};
use crate::base::iana::{Rcode, Rtype};
use crate::base::message::Message;
use crate::base::message_builder::StreamTarget;
use crate::base::opt::{AllOptData, OptRecord, TcpKeepalive};
use crate::base::{ParsedName, Serial};
use crate::rdata::AllRecordData;
use crate::utils::config::DefMinMax;
use bytes::{Bytes, BytesMut};
use core::cmp;
use octseq::Octets;
use std::boxed::Box;
use std::fmt::Debug;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::vec::Vec;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio::sync::{mpsc, oneshot};
use tokio::time::sleep;
use tracing::trace;

//------------ Configuration Constants ----------------------------------------

/// Default response timeout.
///
/// Note: nsd has 120 seconds, unbound has 3 seconds.
const RESPONSE_TIMEOUT: DefMinMax<Duration> = DefMinMax::new(
    Duration::from_secs(19),
    Duration::from_millis(1),
    Duration::from_secs(600),
);

/// Default idle timeout.
///
/// Note that RFC 7766, Secton 6.2.3 says: "DNS clients SHOULD close the
/// TCP connection of an idle session, unless an idle timeout has been
/// established using some other signalling mechanism, for example,
/// [edns-tcp-keepalive]."
/// However, RFC 7858, Section 3.4 says: "In order to amortize TCP and TLS
/// connection setup costs, clients and servers SHOULD NOT immediately close
/// a connection after each response.  Instead, clients and servers SHOULD
/// reuse existing connections for subsequent queries as long as they have
/// sufficient resources.".
/// We set the default to 10 seconds, which is that same as what stubby
/// uses. Minimum zero to allow idle timeout to be disabled. Assume that
/// one hour is more than enough as maximum.
const IDLE_TIMEOUT: DefMinMax<Duration> = DefMinMax::new(
    Duration::from_secs(10),
    Duration::ZERO,
    Duration::from_secs(3600),
);

/// Capacity of the channel that transports `ChanReq`s.
const DEF_CHAN_CAP: usize = 8;

/// Capacity of a private channel dispatching responses.
const READ_REPLY_CHAN_CAP: usize = 8;

//------------ Config ---------------------------------------------------------

/// Configuration for a stream transport connection.
#[derive(Clone, Debug)]
pub struct Config {
    /// Response timeout currently in effect.
    response_timeout: Duration,

    /// Single response timeout.
    single_response_timeout: Duration,

    /// Streaming response timeout.
    streaming_response_timeout: Duration,

    /// Default idle timeout.
    ///
    /// This value is used if the other side does not send a TcpKeepalive
    /// option.
    idle_timeout: Duration,
}

impl Config {
    /// Creates a new, default config.
    pub fn new() -> Self {
        Default::default()
    }

    /// Returns the response timeout.
    ///
    /// This is the amount of time to wait on a non-idle connection for a
    /// response to an outstanding request.
    pub fn response_timeout(&self) -> Duration {
        self.response_timeout
    }

    /// Sets the response timeout.
    ///
    /// For requests where ComposeRequest::is_streaming() returns true see
    /// set_streaming_response_timeout() instead.
    ///
    /// Excessive values are quietly trimmed.
    //
    //  XXX Maybe that’s wrong and we should rather return an error?
    pub fn set_response_timeout(&mut self, timeout: Duration) {
        self.response_timeout = RESPONSE_TIMEOUT.limit(timeout);
        self.streaming_response_timeout = self.response_timeout;
    }

    /// Returns the streaming response timeout.
    pub fn streaming_response_timeout(&self) -> Duration {
        self.streaming_response_timeout
    }

    /// Sets the streaming response timeout.
    ///
    /// Only used for requests where ComposeRequest::is_streaming() returns
    /// true as it is typically desirable that such response streams be
    /// allowed to complete even if the individual responses arrive very
    /// slowly.
    ///
    /// Excessive values are quietly trimmed.
    pub fn set_streaming_response_timeout(&mut self, timeout: Duration) {
        self.streaming_response_timeout = RESPONSE_TIMEOUT.limit(timeout);
    }

    /// Returns the initial idle timeout, if set.
    pub fn idle_timeout(&self) -> Duration {
        self.idle_timeout
    }

    /// Sets the initial idle timeout.
    ///
    /// By default the stream is immediately closed if there are no pending
    /// requests or responses.
    ///
    /// Set this to allow requests to be sent in sequence with delays between
    /// such as a SOA query followed by AXFR for more efficient use of the
    /// stream per RFC 9103.
    ///
    /// Note: May be overridden by an RFC 7828 edns-tcp-keepalive timeout
    /// received from a server.
    ///
    /// Excessive values are quietly trimmed.
    pub fn set_idle_timeout(&mut self, timeout: Duration) {
        self.idle_timeout = IDLE_TIMEOUT.limit(timeout)
    }
}

impl Default for Config {
    fn default() -> Self {
        Self {
            response_timeout: RESPONSE_TIMEOUT.default(),
            single_response_timeout: RESPONSE_TIMEOUT.default(),
            streaming_response_timeout: RESPONSE_TIMEOUT.default(),
            idle_timeout: IDLE_TIMEOUT.default(),
        }
    }
}

//------------ Connection -----------------------------------------------------

/// A connection to a single stream transport.
#[derive(Debug)]
pub struct Connection<Req, ReqMulti> {
    /// The sender half of the request channel.
    sender: mpsc::Sender<ChanReq<Req, ReqMulti>>,
}

impl<Req, ReqMulti> Connection<Req, ReqMulti> {
    /// Creates a new stream transport with default configuration.
    ///
    /// Returns a connection and a future that drives the transport using
    /// the provided stream. This future needs to be run while any queries
    /// are active. This is most easly achieved by spawning it into a runtime.
    /// It terminates when the last connection is dropped.
    pub fn new<Stream>(
        stream: Stream,
    ) -> (Self, Transport<Stream, Req, ReqMulti>) {
        Self::with_config(stream, Default::default())
    }

    /// Creates a new stream transport with the given configuration.
    ///
    /// Returns a connection and a future that drives the transport using
    /// the provided stream. This future needs to be run while any queries
    /// are active. This is most easly achieved by spawning it into a runtime.
    /// It terminates when the last connection is dropped.
    pub fn with_config<Stream>(
        stream: Stream,
        config: Config,
    ) -> (Self, Transport<Stream, Req, ReqMulti>) {
        let (sender, transport) = Transport::new(stream, config);
        (Self { sender }, transport)
    }
}

impl<Req, ReqMulti> Connection<Req, ReqMulti>
where
    Req: ComposeRequest + 'static,
    ReqMulti: ComposeRequestMulti + 'static,
{
    /// Start a DNS request.
    ///
    /// This function takes a precomposed message as a parameter and
    /// returns a [`Message`] object wrapped in a [`Result`].
    async fn handle_request_impl(
        self,
        msg: Req,
    ) -> Result<Message<Bytes>, Error> {
        let (sender, receiver) = oneshot::channel();
        let sender = ReplySender::Single(Some(sender));
        let msg = ReqSingleMulti::Single(msg);
        let req = ChanReq { sender, msg };
        self.sender.send(req).await.map_err(|_| {
            // Send error. The receiver is gone, this means that the
            // connection is closed.
            Error::ConnectionClosed
        })?;
        receiver.await.map_err(|_| Error::StreamReceiveError)?
    }

    /// Start a streaming request.
    async fn handle_streaming_request_impl(
        self,
        msg: ReqMulti,
        sender: mpsc::Sender<Result<Option<Message<Bytes>>, Error>>,
    ) -> Result<(), Error> {
        let reply_sender = ReplySender::Stream(sender);
        let msg = ReqSingleMulti::Multi(msg);
        let req = ChanReq {
            sender: reply_sender,
            msg,
        };
        self.sender.send(req).await.map_err(|_| {
            // Send error. The receiver is gone, this means that the
            // connection is closed.
            Error::ConnectionClosed
        })?;
        Ok(())
    }

    /// Returns a request handler for a request.
    pub fn get_request(&self, request_msg: Req) -> Request {
        Request {
            fut: Box::pin(self.clone().handle_request_impl(request_msg)),
        }
    }

    /// Return a multiple-response request handler for a request.
    fn get_streaming_request(&self, request_msg: ReqMulti) -> RequestMulti {
        let (sender, receiver) = mpsc::channel(DEF_CHAN_CAP);
        RequestMulti {
            stream: receiver,
            fut: Some(Box::pin(
                self.clone()
                    .handle_streaming_request_impl(request_msg, sender),
            )),
        }
    }
}

impl<Req, ReqMulti> Clone for Connection<Req, ReqMulti> {
    fn clone(&self) -> Self {
        Self {
            sender: self.sender.clone(),
        }
    }
}

impl<Req, ReqMulti> SendRequest<Req> for Connection<Req, ReqMulti>
where
    Req: ComposeRequest + 'static,
    ReqMulti: ComposeRequestMulti + Debug + Send + Sync + 'static,
{
    fn send_request(
        &self,
        request_msg: Req,
    ) -> Box<dyn GetResponse + Send + Sync> {
        Box::new(self.get_request(request_msg))
    }
}

impl<Req, ReqMulti> SendRequestMulti<ReqMulti> for Connection<Req, ReqMulti>
where
    Req: ComposeRequest + Debug + Send + Sync + 'static,
    ReqMulti: ComposeRequestMulti + 'static,
{
    fn send_request(
        &self,
        request_msg: ReqMulti,
    ) -> Box<dyn GetResponseMulti + Send + Sync> {
        Box::new(self.get_streaming_request(request_msg))
    }
}

//------------ Request -------------------------------------------------------

/// An active request.
pub struct Request {
    /// The underlying future.
    fut: Pin<
        Box<dyn Future<Output = Result<Message<Bytes>, Error>> + Send + Sync>,
    >,
}

impl Request {
    /// Async function that waits for the future stored in Request to complete.
    async fn get_response_impl(&mut self) -> Result<Message<Bytes>, Error> {
        (&mut self.fut).await
    }
}

impl GetResponse for Request {
    fn get_response(
        &mut self,
    ) -> Pin<
        Box<
            dyn Future<Output = Result<Message<Bytes>, Error>>
                + Send
                + Sync
                + '_,
        >,
    > {
        Box::pin(self.get_response_impl())
    }
}

impl Debug for Request {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        f.debug_struct("Request")
            .field("fut", &format_args!("_"))
            .finish()
    }
}

//------------ RequestMulti --------------------------------------------------

/// An active request.
pub struct RequestMulti {
    /// Receiver for a stream of responses.
    stream: mpsc::Receiver<Result<Option<Message<Bytes>>, Error>>,

    /// The underlying future.
    #[allow(clippy::type_complexity)]
    fut: Option<
        Pin<Box<dyn Future<Output = Result<(), Error>> + Send + Sync>>,
    >,
}

impl RequestMulti {
    /// Async function that waits for the future stored in Request to complete.
    async fn get_response_impl(
        &mut self,
    ) -> Result<Option<Message<Bytes>>, Error> {
        if self.fut.is_some() {
            let fut = self.fut.take().expect("Some expected");
            fut.await?;
        }

        // Fetch from the stream
        self.stream
            .recv()
            .await
            .ok_or(Error::ConnectionClosed)
            .map_err(|_| Error::ConnectionClosed)?
    }
}

impl GetResponseMulti for RequestMulti {
    fn get_response(
        &mut self,
    ) -> Pin<
        Box<
            dyn Future<Output = Result<Option<Message<Bytes>>, Error>>
                + Send
                + Sync
                + '_,
        >,
    > {
        let fut = self.get_response_impl();
        Box::pin(fut)
    }
}

impl Debug for RequestMulti {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        f.debug_struct("Request")
            .field("fut", &format_args!("_"))
            .finish()
    }
}

//------------ Transport -----------------------------------------------------

/// The underlying machinery of a stream transport.
#[derive(Debug)]
pub struct Transport<Stream, Req, ReqMulti> {
    /// The stream socket towards the remote end.
    stream: Stream,

    /// Transport configuration.
    config: Config,

    /// The receiver half of request channel.
    receiver: mpsc::Receiver<ChanReq<Req, ReqMulti>>,
}

/// This is the type of sender in [ChanReq].
#[derive(Debug)]
enum ReplySender {
    /// Return channel for a single response.
    Single(Option<oneshot::Sender<ChanResp>>),

    /// Return channel for a stream of responses.
    Stream(mpsc::Sender<Result<Option<Message<Bytes>>, Error>>),
}

impl ReplySender {
    /// Send a response.
    async fn send(&mut self, resp: ChanResp) -> Result<(), ()> {
        match self {
            ReplySender::Single(sender) => match sender.take() {
                Some(sender) => sender.send(resp).map_err(|_| ()),
                None => Err(()),
            },
            ReplySender::Stream(sender) => {
                sender.send(resp.map(Some)).await.map_err(|_| ())
            }
        }
    }

    /// Send EOF on a response stream.
    async fn send_eof(&mut self) -> Result<(), ()> {
        match self {
            ReplySender::Single(_) => {
                panic!("cannot send EOF for Single");
            }
            ReplySender::Stream(sender) => {
                sender.send(Ok(None)).await.map_err(|_| ())
            }
        }
    }

    /// Report whether in stream mode or not.
    fn is_stream(&self) -> bool {
        matches!(self, Self::Stream(_))
    }
}

#[derive(Debug)]
/// Enum that can either store a request for a single response or one for
/// multiple responses.
enum ReqSingleMulti<Req, ReqMulti> {
    /// Single response request.
    Single(Req),
    /// Multi-response request.
    Multi(ReqMulti),
}

/// A message from a [`Request`] to start a new request.
#[derive(Debug)]
struct ChanReq<Req, ReqMulti> {
    /// DNS request message
    msg: ReqSingleMulti<Req, ReqMulti>,

    /// Sender to send result back to [`Request`]
    sender: ReplySender,
}

/// A message back to [`Request`] returning a response.
type ChanResp = Result<Message<Bytes>, Error>;

/// Internal datastructure of [Transport::run] to keep track of
/// the status of the connection.
// The types Status and ConnState are only used in Transport
struct Status {
    /// State of the connection.
    state: ConnState,

    /// Do we need to include edns-tcp-keepalive in an outogoing request.
    ///
    /// Typically this is true at the start of the connection and gets
    /// cleared when we successfully managed to include the option in a
    /// request.
    send_keepalive: bool,

    /// Time we are allow to keep the connection open when idle.
    ///
    /// Initially we set the idle timeout to the default in config. A received
    /// edns-tcp-keepalive option may change that.
    idle_timeout: Duration,
}

/// Status of the connection. Used in [`Status`].
enum ConnState {
    /// The connection is in this state from the start and when at least
    /// one active DNS request is present.
    ///
    /// The instant contains the time of the first request or the
    /// most recent response that was received.
    Active(Option<Instant>),

    /// This state represent a connection that went idle and has an
    /// idle timeout.
    ///
    /// The instant contains the time the connection went idle.
    Idle(Instant),

    /// This state represent an idle connection where either there was no
    /// idle timeout or the idle timer expired.
    IdleTimeout,

    /// A read error occurred.
    ReadError(Error),

    /// It took too long to receive a response.
    ReadTimeout,

    /// A write error occurred.
    WriteError(Error),
}

//--- Display
impl std::fmt::Display for ConnState {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        match self {
            ConnState::Active(instant) => f.write_fmt(format_args!(
                "Active (since {}s ago)",
                instant
                    .map(|v| Instant::now().duration_since(v).as_secs())
                    .unwrap_or_default()
            )),
            ConnState::Idle(instant) => f.write_fmt(format_args!(
                "Idle (since {}s ago)",
                Instant::now().duration_since(*instant).as_secs()
            )),
            ConnState::IdleTimeout => f.write_str("IdleTimeout"),
            ConnState::ReadError(err) => {
                f.write_fmt(format_args!("ReadError: {err}"))
            }
            ConnState::ReadTimeout => f.write_str("ReadTimeout"),
            ConnState::WriteError(err) => {
                f.write_fmt(format_args!("WriteError: {err}"))
            }
        }
    }
}

#[derive(Debug)]
/// State of an AXFR or IXFR responses stream for detecting the end of the
/// stream.
enum XFRState {
    /// Start of AXFR.
    AXFRInit,
    /// After the first SOA record has been encountered.
    AXFRFirstSoa(Serial),
    /// Start of IXFR.
    IXFRInit,
    /// After the first SOA record has been encountered.
    IXFRFirstSoa(Serial),
    /// After the first SOA record in a diff section has been encountered.
    IXFRFirstDiffSoa(Serial),
    /// After the second SOA record in a diff section has been encountered.
    IXFRSecondDiffSoa(Serial),
    /// End of the stream has been found.
    Done,
    /// An error has occured.
    Error,
}

impl<Stream, Req, ReqMulti> Transport<Stream, Req, ReqMulti> {
    /// Creates a new transport.
    fn new(
        stream: Stream,
        config: Config,
    ) -> (mpsc::Sender<ChanReq<Req, ReqMulti>>, Self) {
        let (sender, receiver) = mpsc::channel(DEF_CHAN_CAP);
        (
            sender,
            Self {
                config,
                stream,
                receiver,
            },
        )
    }
}

impl<Stream, Req, ReqMulti> Transport<Stream, Req, ReqMulti>
where
    Stream: AsyncRead + AsyncWrite,
    Req: ComposeRequest,
    ReqMulti: ComposeRequestMulti,
{
    /// Run the transport machinery.
    pub async fn run(mut self) {
        let (reply_sender, mut reply_receiver) =
            mpsc::channel::<Message<Bytes>>(READ_REPLY_CHAN_CAP);

        let (read_stream, mut write_stream) = tokio::io::split(self.stream);

        let reader_fut = Self::reader(read_stream, reply_sender);
        tokio::pin!(reader_fut);

        let mut status = Status {
            state: ConnState::Active(None),
            idle_timeout: self.config.idle_timeout,
            send_keepalive: true,
        };
        let mut query_vec =
            Queries::<(ChanReq<Req, ReqMulti>, Option<XFRState>)>::new();

        let mut reqmsg: Option<Vec<u8>> = None;
        let mut reqmsg_offset = 0;

        loop {
            let opt_timeout = match status.state {
                ConnState::Active(opt_instant) => {
                    if let Some(instant) = opt_instant {
                        let elapsed = instant.elapsed();
                        if elapsed > self.config.response_timeout {
                            Self::error(
                                Error::StreamReadTimeout,
                                &mut query_vec,
                            );
                            status.state = ConnState::ReadTimeout;
                            break;
                        }
                        Some(self.config.response_timeout - elapsed)
                    } else {
                        None
                    }
                }
                ConnState::Idle(instant) => {
                    let elapsed = instant.elapsed();
                    if elapsed >= status.idle_timeout {
                        // Move to IdleTimeout and end
                        // the loop
                        status.state = ConnState::IdleTimeout;
                        break;
                    }
                    Some(status.idle_timeout - elapsed)
                }
                ConnState::IdleTimeout
                | ConnState::ReadError(_)
                | ConnState::WriteError(_) => None, // No timers here
                ConnState::ReadTimeout => {
                    panic!("should not be in loop with ReadTimeout");
                }
            };

            // For simplicity, make sure we always have a timeout
            let timeout = match opt_timeout {
                Some(timeout) => timeout,
                None =>
                // Just use the response timeout
                {
                    self.config.response_timeout
                }
            };

            let sleep_fut = sleep(timeout);
            let recv_fut = self.receiver.recv();

            let (do_write, msg) = match &reqmsg {
                None => {
                    let msg: &[u8] = &[];
                    (false, msg)
                }
                Some(msg) => {
                    let msg: &[u8] = msg;
                    (true, msg)
                }
            };

            tokio::select! {
                biased;
                res = &mut reader_fut => {
                    match res {
                        Ok(_) =>
                            // The reader should not
                            // terminate without
                            // error.
                            panic!("reader terminated"),
                        Err(error) => {
                            Self::error(error.clone(), &mut query_vec);
                            status.state = ConnState::ReadError(error);
                            // Reader failed. Break
                            // out of loop and
                            // shut down
                            break
                        }
                    }
                }
                opt_answer = reply_receiver.recv() => {
                    let answer = opt_answer.expect("reader died?");
                    // Check for a edns-tcp-keepalive option
                    let opt_record = answer.opt();
                    if let Some(ref opts) = opt_record {
                        Self::handle_opts(opts,
                            &mut status);
                    };
                    drop(opt_record);
                    Self::demux_reply(answer, &mut status, &mut query_vec).await;
                }
                res = write_stream.write(&msg[reqmsg_offset..]),
                if do_write => {
            match res {
            Err(error) => {
                let error =
                Error::StreamWriteError(Arc::new(error));
                Self::error(error.clone(), &mut query_vec);
                status.state =
                ConnState::WriteError(error);
                break;
            }
            Ok(len) => {
                reqmsg_offset += len;
                if reqmsg_offset >= msg.len() {
                reqmsg = None;
                reqmsg_offset = 0;
                }
            }
            }
                }
                res = recv_fut, if !do_write => {
                    match res {
                        Some(req) => {
                            if req.sender.is_stream() {
                                self.config.response_timeout =
                                    self.config.streaming_response_timeout;
                            } else {
                                self.config.response_timeout =
                                    self.config.single_response_timeout;
                            }
                            Self::insert_req(
                                req, &mut status, &mut reqmsg, &mut query_vec
                            );
                        }
                        None => {
                            // All references to the connection object have
                            // been dropped. Shutdown.
                            break;
                        }
                    }
                }
                _ = sleep_fut => {
                    // Timeout expired, just
                    // continue with the loop
                }

            }

            // Check if the connection is idle
            match status.state {
                ConnState::Active(_) | ConnState::Idle(_) => {
                    // Keep going
                }
                ConnState::IdleTimeout => break,
                ConnState::ReadError(_)
                | ConnState::ReadTimeout
                | ConnState::WriteError(_) => {
                    panic!("Should not be here");
                }
            }
        }

        trace!("Closing TCP connecting in state: {}", status.state);

        // Send FIN
        _ = write_stream.shutdown().await;
    }

    /// This function reads a DNS message from the connection and sends
    /// it to [Transport::run].
    ///
    /// Reading has to be done in two steps: first read a two octet value
    /// the specifies the length of the message, and then read in a loop the
    /// body of the message.
    ///
    /// This function is not async cancellation safe.
    async fn reader(
        mut sock: tokio::io::ReadHalf<Stream>,
        sender: mpsc::Sender<Message<Bytes>>,
    ) -> Result<(), Error> {
        loop {
            let read_res = sock.read_u16().await;
            let len = match read_res {
                Ok(len) => len,
                Err(error) => {
                    return Err(Error::StreamReadError(Arc::new(error)));
                }
            } as usize;

            let mut buf = BytesMut::with_capacity(len);

            loop {
                let curlen = buf.len();
                if curlen >= len {
                    if curlen > len {
                        panic!(
                        "reader: got too much data {curlen}, expetect {len}");
                    }

                    // We got what we need
                    break;
                }

                let read_res = sock.read_buf(&mut buf).await;

                match read_res {
                    Ok(readlen) => {
                        if readlen == 0 {
                            return Err(Error::StreamUnexpectedEndOfData);
                        }
                    }
                    Err(error) => {
                        return Err(Error::StreamReadError(Arc::new(error)));
                    }
                };

                // Check if we are done at the head of the loop
            }

            let reply_message = Message::<Bytes>::from_octets(buf.into());
            match reply_message {
                Ok(answer) => {
                    sender
                        .send(answer)
                        .await
                        .expect("can't send reply to run");
                }
                Err(_) => {
                    // The only possible error is short message
                    return Err(Error::ShortMessage);
                }
            }
        }
    }

    /// Reports an error to all outstanding queries.
    fn error(
        error: Error,
        query_vec: &mut Queries<(ChanReq<Req, ReqMulti>, Option<XFRState>)>,
    ) {
        // Update all requests that are in progress. Don't wait for
        // any reply that may be on its way.
        for (mut req, _) in query_vec.drain() {
            _ = req.sender.send(Err(error.clone()));
        }
    }

    /// Handles received EDNS options.
    ///
    /// In particular, it processes the edns-tcp-keepalive option.
    fn handle_opts<Octs: Octets + AsRef<[u8]>>(
        opts: &OptRecord<Octs>,
        status: &mut Status,
    ) {
        // XXX This handles _all_ keepalive options. I think just using the
        //     first option as returned by Opt::tcp_keepalive should be good
        //     enough? -- M.
        for option in opts.opt().iter().flatten() {
            if let AllOptData::TcpKeepalive(tcpkeepalive) = option {
                Self::handle_keepalive(tcpkeepalive, status);
            }
        }
    }

    /// Demultiplexes a response and sends it to the right query.
    ///
    /// In addition, the status is updated to IdleTimeout or Idle if there
    /// are no remaining pending requests.
    async fn demux_reply(
        answer: Message<Bytes>,
        status: &mut Status,
        query_vec: &mut Queries<(ChanReq<Req, ReqMulti>, Option<XFRState>)>,
    ) {
        // We got an answer, reset the timer
        status.state = ConnState::Active(Some(Instant::now()));

        let id = answer.header().id();

        // Get the correct query and send it the reply.
        let (mut req, mut opt_xfr_data) = match query_vec.try_remove(id) {
            Some(req) => req,
            None => {
                // No query with this ID. We should
                // mark the connection as broken
                return;
            }
        };
        let mut send_eof = false;
        let answer = if match &req.msg {
            ReqSingleMulti::Single(msg) => msg.is_answer(answer.for_slice()),
            ReqSingleMulti::Multi(msg) => {
                let xfr_data =
                    opt_xfr_data.expect("xfr_data should be present");
                let (eof, xfr_data, is_answer) =
                    check_stream(msg, xfr_data, &answer);
                send_eof = eof;
                opt_xfr_data = Some(xfr_data);
                is_answer
            }
        } {
            Ok(answer)
        } else {
            Err(Error::WrongReplyForQuery)
        };
        _ = req.sender.send(answer).await;

        if req.sender.is_stream() {
            if send_eof {
                _ = req.sender.send_eof().await;
            } else {
                query_vec.insert_at(id, (req, opt_xfr_data));
            }
        }

        if query_vec.is_empty() {
            // Clear the activity timer. There is no need to do
            // this because state will be set to either IdleTimeout
            // or Idle just below. However, it is nicer to keep
            // this independent.
            status.state = ConnState::Active(None);

            status.state = if status.idle_timeout.is_zero() {
                // Assume that we can just move to IdleTimeout
                // state
                ConnState::IdleTimeout
            } else {
                ConnState::Idle(Instant::now())
            }
        }
    }

    /// Insert a request in query_vec and return the request to be sent
    /// in *reqmsg.
    ///
    /// First the status is checked, an error is returned if not Active or
    /// idle. Addend a edns-tcp-keepalive option if needed.
    // Note: maybe reqmsg should be a return value.
    fn insert_req(
        mut req: ChanReq<Req, ReqMulti>,
        status: &mut Status,
        reqmsg: &mut Option<Vec<u8>>,
        query_vec: &mut Queries<(ChanReq<Req, ReqMulti>, Option<XFRState>)>,
    ) {
        match &status.state {
            ConnState::Active(timer) => {
                // Set timer if we don't have one already
                if timer.is_none() {
                    status.state = ConnState::Active(Some(Instant::now()));
                }
            }
            ConnState::Idle(_) => {
                // Go back to active
                status.state = ConnState::Active(Some(Instant::now()));
            }
            ConnState::IdleTimeout => {
                // The connection has been closed. Report error
                _ = req.sender.send(Err(Error::StreamIdleTimeout));
                return;
            }
            ConnState::ReadError(error) => {
                _ = req.sender.send(Err(error.clone()));
                return;
            }
            ConnState::ReadTimeout => {
                _ = req.sender.send(Err(Error::StreamReadTimeout));
                return;
            }
            ConnState::WriteError(error) => {
                _ = req.sender.send(Err(error.clone()));
                return;
            }
        }

        let xfr_data = match &req.msg {
            ReqSingleMulti::Single(_) => None,
            ReqSingleMulti::Multi(msg) => {
                let qtype = match msg.to_message().and_then(|m| {
                    m.sole_question()
                        .map_err(|_| Error::MessageParseError)
                        .map(|q| q.qtype())
                }) {
                    Ok(msg) => msg,
                    Err(e) => {
                        _ = req.sender.send(Err(e));
                        return;
                    }
                };
                if qtype == Rtype::AXFR {
                    Some(XFRState::AXFRInit)
                } else if qtype == Rtype::IXFR {
                    Some(XFRState::IXFRInit)
                } else {
                    // Stream requests should be either AXFR or IXFR.
                    _ = req.sender.send(Err(Error::FormError));
                    return;
                }
            }
        };

        // Note that insert may fail if there are too many
        // outstanding queries. First call insert before checking
        // send_keepalive.
        let (index, (req, _)) = match query_vec.insert((req, xfr_data)) {
            Ok(res) => res,
            Err((mut req, _)) => {
                // Send an appropriate error and return.
                _ = req
                    .sender
                    .send(Err(Error::StreamTooManyOutstandingQueries));
                return;
            }
        };

        // We set the ID to the array index. Defense in depth
        // suggests that a random ID is better because it works
        // even if TCP sequence numbers could be predicted. However,
        // Section 9.3 of RFC 5452 recommends retrying over TCP
        // if many spoofed answers arrive over UDP: "TCP, by the
        // nature of its use of sequence numbers, is far more
        // resilient against forgery by third parties."

        let hdr = match &mut req.msg {
            ReqSingleMulti::Single(msg) => msg.header_mut(),
            ReqSingleMulti::Multi(msg) => msg.header_mut(),
        };
        hdr.set_id(index);

        if status.send_keepalive
            && match &mut req.msg {
                ReqSingleMulti::Single(msg) => {
                    msg.add_opt(&TcpKeepalive::new(None)).is_ok()
                }
                ReqSingleMulti::Multi(msg) => {
                    msg.add_opt(&TcpKeepalive::new(None)).is_ok()
                }
            }
        {
            status.send_keepalive = false;
        }

        match Self::convert_query(&req.msg) {
            Ok(msg) => {
                *reqmsg = Some(msg);
            }
            Err(err) => {
                // Take the sender out again and return the error.
                if let Some((mut req, _)) = query_vec.try_remove(index) {
                    _ = req.sender.send(Err(err));
                }
            }
        }
    }

    /// Handle a received edns-tcp-keepalive option.
    fn handle_keepalive(opt_value: TcpKeepalive, status: &mut Status) {
        if let Some(value) = opt_value.timeout() {
            let value_dur = Duration::from(value);
            status.idle_timeout = value_dur;
        }
    }

    /// Convert the query message to a vector.
    fn convert_query(
        msg: &ReqSingleMulti<Req, ReqMulti>,
    ) -> Result<Vec<u8>, Error> {
        match msg {
            ReqSingleMulti::Single(msg) => {
                let mut target = StreamTarget::new_vec();
                msg.append_message(&mut target)
                    .map_err(|_| Error::StreamLongMessage)?;
                Ok(target.into_target())
            }
            ReqSingleMulti::Multi(msg) => {
                let target = StreamTarget::new_vec();
                let target = msg
                    .append_message(target)
                    .map_err(|_| Error::StreamLongMessage)?;
                Ok(target.finish().into_target())
            }
        }
    }
}

/// Upstate the response stream state based on a response message.
fn check_stream<CRM>(
    msg: &CRM,
    mut xfr_state: XFRState,
    answer: &Message<Bytes>,
) -> (bool, XFRState, bool)
where
    CRM: ComposeRequestMulti,
{
    // First check if the reply matches the request.
    // RFC 5936, Section 2.2.2:
    // "In the first response message, this section MUST be copied from the
    // query.  In subsequent messages, this section MAY be copied from the
    // query, or it MAY be empty.  However, in an error response message
    // (see Section 2.2), this section MUST be copied as well."
    match xfr_state {
        XFRState::AXFRInit | XFRState::IXFRInit => {
            if !msg.is_answer(answer.for_slice()) {
                xfr_state = XFRState::Error;
                // If we detect an error, then keep the stream open. We are
                // likely out of sync with respect to the sender.
                return (false, xfr_state, false);
            }
        }
        XFRState::AXFRFirstSoa(_)
        | XFRState::IXFRFirstSoa(_)
        | XFRState::IXFRFirstDiffSoa(_)
        | XFRState::IXFRSecondDiffSoa(_) =>
            // No need to check anything.
            {}
        XFRState::Done => {
            // We should not be here. Switch to error state.
            xfr_state = XFRState::Error;
            return (false, xfr_state, false);
        }
        XFRState::Error =>
        // Keep the stream open.
        {
            return (false, xfr_state, false)
        }
    }

    // Then check if the reply status an error.
    if answer.header().rcode() != Rcode::NOERROR {
        // Also check if this answers the question.
        if !msg.is_answer(answer.for_slice()) {
            xfr_state = XFRState::Error;
            // If we detect an error, then keep the stream open. We are
            // likely out of sync with respect to the sender.
            return (false, xfr_state, false);
        }
        return (true, xfr_state, true);
    }

    let ans_sec = match answer.answer() {
        Ok(ans) => ans,
        Err(_) => {
            // Bad message, switch to error state.
            xfr_state = XFRState::Error;
            // If we detect an error, then keep the stream open.
            return (true, xfr_state, false);
        }
    };
    for rr in
        ans_sec.into_records::<AllRecordData<Bytes, ParsedName<Bytes>>>()
    {
        let rr = match rr {
            Ok(rr) => rr,
            Err(_) => {
                // Bad message, switch to error state.
                xfr_state = XFRState::Error;
                return (true, xfr_state, false);
            }
        };
        match xfr_state {
            XFRState::AXFRInit => {
                // The first record has to be a SOA record.
                if let AllRecordData::Soa(soa) = rr.data() {
                    xfr_state = XFRState::AXFRFirstSoa(soa.serial());
                    continue;
                }
                // Bad data. Switch to error status.
                xfr_state = XFRState::Error;
                return (false, xfr_state, false);
            }
            XFRState::AXFRFirstSoa(serial) => {
                if let AllRecordData::Soa(soa) = rr.data() {
                    if serial == soa.serial() {
                        // We found a match.
                        xfr_state = XFRState::Done;
                        continue;
                    }

                    // Serial does not match. Move to error state.
                    xfr_state = XFRState::Error;
                    return (false, xfr_state, false);
                }

                // Any other record, just continue.
            }
            XFRState::IXFRInit => {
                // The first record has to be a SOA record.
                if let AllRecordData::Soa(soa) = rr.data() {
                    xfr_state = XFRState::IXFRFirstSoa(soa.serial());
                    continue;
                }
                // Bad data. Switch to error status.
                xfr_state = XFRState::Error;
                return (false, xfr_state, false);
            }
            XFRState::IXFRFirstSoa(serial) => {
                // We have three possibilities:
                // 1) The record is not a SOA. In that case the format is AXFR.
                // 2) The record is a SOA and the serial is not the current
                //    serial. That is expected for an IXFR format. Move to
                //    IXFRFirstDiffSoa.
                // 3) The record is a SOA and the serial is equal to the
                //    current serial. Treat this as a strange empty AXFR.
                if let AllRecordData::Soa(soa) = rr.data() {
                    if serial == soa.serial() {
                        // We found a match.
                        xfr_state = XFRState::Done;
                        continue;
                    }

                    xfr_state = XFRState::IXFRFirstDiffSoa(serial);
                    continue;
                }

                // Any other record, move to AXFRFirstSoa.
                xfr_state = XFRState::AXFRFirstSoa(serial);
            }
            XFRState::IXFRFirstDiffSoa(serial) => {
                // Move to IXFRSecondDiffSoa if the record is a SOA record,
                // otherwise stay in the current state.
                if let AllRecordData::Soa(_) = rr.data() {
                    xfr_state = XFRState::IXFRSecondDiffSoa(serial);
                    continue;
                }

                // Any other record, just continue.
            }
            XFRState::IXFRSecondDiffSoa(serial) => {
                // Move to Done if the record is a SOA record and the
                // serial is the one from the first SOA record, move to
                // IXFRFirstDiffSoa for any other SOA record and
                // otherwise stay in the current state.
                if let AllRecordData::Soa(soa) = rr.data() {
                    if serial == soa.serial() {
                        // We found a match.
                        xfr_state = XFRState::Done;
                        continue;
                    }

                    xfr_state = XFRState::IXFRFirstDiffSoa(serial);
                    continue;
                }

                // Any other record, just continue.
            }
            XFRState::Done => {
                // We got a record after we are done. Switch to error state.
                xfr_state = XFRState::Error;
                return (false, xfr_state, false);
            }
            XFRState::Error => panic!("should not be here"),
        }
    }

    // Check the final state.
    match xfr_state {
        XFRState::AXFRInit | XFRState::IXFRInit => {
            // Still in one of the init state. So the data section was empty.
            // Switch to error state.
            xfr_state = XFRState::Error;
            return (false, xfr_state, false);
        }
        XFRState::AXFRFirstSoa(_)
        | XFRState::IXFRFirstDiffSoa(_)
        | XFRState::IXFRSecondDiffSoa(_) =>
            // Just continue.
            {}
        XFRState::IXFRFirstSoa(_) => {
            // We are still in IXFRFirstSoa. Assume the other side doesn't
            // have anything more to say. We could check the SOA serial in
            // the request. Just assume that we are done.
            xfr_state = XFRState::Done;
            return (true, xfr_state, true);
        }
        XFRState::Done => return (true, xfr_state, true),
        XFRState::Error => unreachable!(),
    }

    // (eof, xfr_data, is_answer)
    (false, xfr_state, true)
}

//------------ Queries -------------------------------------------------------

/// Mapping outstanding queries to their ID.
///
/// This is generic over anything rather than our concrete request type for
/// easier testing.
#[derive(Clone, Debug)]
struct Queries<T> {
    /// The number of elements in `vec` that are not None.
    count: usize,

    /// Index in `vec` where to look for a space for a new query.
    curr: usize,

    /// Vector of senders to forward a DNS reply message (or error) to.
    vec: Vec<Option<T>>,
}

impl<T> Queries<T> {
    /// Creates a new empty value.
    fn new() -> Self {
        Self {
            count: 0,
            curr: 0,
            vec: Vec::new(),
        }
    }

    /// Returns whether there are no more outstanding queries.
    fn is_empty(&self) -> bool {
        self.count == 0
    }

    /// Inserts the given query.
    ///
    /// Upon success, returns the index and a mutable reference to the stored
    /// query.
    ///
    /// Upon error, which means the set is full, returns the query.
    fn insert(&mut self, req: T) -> Result<(u16, &mut T), T> {
        // Fail if there are to many entries already in this vector
        // We cannot have more than u16::MAX entries because the
        // index needs to fit in an u16. For efficiency we want to
        // keep the vector half empty. So we return a failure if
        // 2*count > u16::MAX
        if 2 * self.count > u16::MAX.into() {
            return Err(req);
        }

        // If more than half the vec is empty, we try and find the index of
        // an empty slot.
        let idx = if self.vec.len() >= 2 * self.count {
            let mut found = None;
            for idx in self.curr..self.vec.len() {
                if self.vec[idx].is_none() {
                    found = Some(idx);
                    break;
                }
            }
            found
        } else {
            None
        };

        // If we have an index, we can insert there, otherwise we need to
        // append.
        let idx = match idx {
            Some(idx) => {
                self.vec[idx] = Some(req);
                idx
            }
            None => {
                let idx = self.vec.len();
                self.vec.push(Some(req));
                idx
            }
        };

        self.count += 1;
        if idx == self.curr {
            self.curr += 1;
        }
        let req = self.vec[idx].as_mut().expect("no inserted item?");
        let idx = u16::try_from(idx).expect("query vec too large");
        Ok((idx, req))
    }

    /// Inserts the given query at a specified position. A pre-condition is
    /// is that the slot has to be empty.
    fn insert_at(&mut self, id: u16, req: T) {
        let id = id as usize;
        self.vec[id] = Some(req);

        self.count += 1;
        if id == self.curr {
            self.curr += 1;
        }
    }

    /// Tries to remove and return the query at the given index.
    ///
    /// Returns `None` if there was no query there.
    fn try_remove(&mut self, index: u16) -> Option<T> {
        let res = self.vec.get_mut(usize::from(index))?.take()?;
        self.count = self.count.saturating_sub(1);
        self.curr = cmp::min(self.curr, index.into());
        Some(res)
    }

    /// Removes all queries and returns an iterator over them.
    fn drain(&mut self) -> impl Iterator<Item = T> + '_ {
        let res = self.vec.drain(..).flatten(); // Skips all the `None`s.
        self.count = 0;
        self.curr = 0;
        res
    }
}

//============ Tests =========================================================

#[cfg(test)]
mod test {
    use super::*;

    #[test]
    #[allow(clippy::needless_range_loop)]
    fn queries_insert_remove() {
        // Insert items, remove a few, insert a few more. Check that
        // everything looks right.
        let mut idxs = [None; 20];
        let mut queries = Queries::new();

        for i in 0..12 {
            let (idx, item) = queries.insert(i).expect("test failed");
            idxs[i] = Some(idx);
            assert_eq!(i, *item);
        }
        assert_eq!(queries.count, 12);
        assert_eq!(queries.vec.iter().flatten().count(), 12);

        for i in [1, 2, 3, 4, 7, 9] {
            let item = queries
                .try_remove(idxs[i].expect("test failed"))
                .expect("test failed");
            assert_eq!(i, item);
            idxs[i] = None;
        }
        assert_eq!(queries.count, 6);
        assert_eq!(queries.vec.iter().flatten().count(), 6);

        for i in 12..20 {
            let (idx, item) = queries.insert(i).expect("test failed");
            idxs[i] = Some(idx);
            assert_eq!(i, *item);
        }
        assert_eq!(queries.count, 14);
        assert_eq!(queries.vec.iter().flatten().count(), 14);

        for i in 0..20 {
            if let Some(idx) = idxs[i] {
                let item = queries.try_remove(idx).expect("test failed");
                assert_eq!(i, item);
            }
        }
        assert_eq!(queries.count, 0);
        assert_eq!(queries.vec.iter().flatten().count(), 0);
    }

    #[test]
    fn queries_overrun() {
        // This is just a quick check that inserting to much stuff doesn’t
        // break.
        let mut queries = Queries::new();
        for i in 0..usize::from(u16::MAX) * 2 {
            let _ = queries.insert(i);
        }
    }
}