srt-runtime 0.4.0

SRT (Secure Reliable Transport, draft-sharabayko-srt-01) packet codecs + sans-IO HSv5 Caller-Listener and Rendezvous handshake state machines, an ARQ reliability engine, a TSBPD delivery scheduler, LiveCC packet pacing, FileCC window-based congestion control, optional (crypto feature) payload encryption, and an optional (tokio feature) async UDP socket adapter; no_std core.
Documentation
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
//! Rendezvous handshake engine — `draft-sharabayko-srt-01` §4.3.2 (Rendezvous
//! Handshake), curated at `specs/rules/srt-rendezvous.md`. Line cites below
//! (`LNNNN`) are the source draft's line numbers, exactly as recorded in that
//! curation.
//!
//! [`RendezvousHandshake`] is symmetric: both peers run the *same* engine
//! (unlike [`crate::caller::CallerHandshake`] / [`crate::listener::ListenerHandshake`],
//! which run different code for different roles). Which of the two logical
//! roles — **Initiator** or **Responder** — a given instance ends up playing
//! is decided at runtime by the **cookie contest** (§4.3.2, L2107-2135):
//! each side supplies its own 32-bit cookie to [`RendezvousHandshake::new`]
//! (this crate never reads a clock or a socket address — see
//! [`crate::handshake_sm::derive_cookie`] for a ready-made, non-standardized
//! derivation helper, exactly as for [`crate::listener::ListenerHandshake`]),
//! and the greater cookie value wins ("becomes Initiator", L2133-2135).
//!
//! # State machine
//!
//! States are named exactly as the draft's Parallel Handshake Flow diagram
//! (§4.3.2.2, L2280, quoted verbatim): **Waving → Attention → Initiated →
//! Connected** (plus `Idle` before [`RendezvousHandshake::start`] is called —
//! not spec-named, mirrors [`crate::caller::CallerHandshakeState::Idle`] — and
//! `Rejected`/`TimedOut` terminal states, also not spec-named). Every
//! transition below is the Initiator table (L2301-2334) or Responder table
//! (L2336-2382), or a missing-packet recovery rule (L2383-2432).
//!
//! ## Serial vs Parallel flow: one engine, not two
//!
//! The draft narrates two "flows" (§4.3.2.1 Serial, L2140-2269; §4.3.2.2
//! Parallel, L2270-2432) that differ only in *message interleaving*, not in
//! any new transition rule: the Parallel flow's Initiator/Responder tables are
//! complete state × received-message tables, driven purely by message
//! content — so they already cover the Serial flow's crossing case. That case
//! is: a peer still in `Waving` (having sent its own WAVEAHAND but never
//! having received the other side's) receives a **CONCLUSION** directly
//! instead of a WAVEAHAND (§4.3.2.1 step 3, L2203-2219 — the draft calls the
//! resulting state "fine"). Tracing the draft's own worked example confirms
//! the action taken is identical to applying the Parallel Attention row to
//! that CONCLUSION: an Initiator receiving an extension-less CONCLUSION here
//! behaves exactly like the Initiator-Attention-row "no extensions" case
//! (L2312-2318); a Responder receiving a CONCLUSION+HSREQ here behaves
//! exactly like the Responder-Attention-row HSREQ case (L2360-2364) — because
//! by this point in the exchange the peer's role-appropriate first CONCLUSION
//! already carries whatever extension its role dictates (Initiator: HSREQ
//! immediately, L2286-2287; Responder: none until it has seen HSREQ,
//! L2357-2360/L2287-2288). This implementation therefore has **no separate
//! "Fine" state** — a received CONCLUSION while still in `Waving` dispatches
//! straight into the same Attention-row logic used when genuinely in
//! `Attention`, rather than inventing a fourth, behaviourally-divergent state
//! the tables do not define.
//!
//! ## Resolved ambiguities (not explicit in the curated rules)
//!
//! - **Cookie collision → rejection.** The draft says only "the connection
//!   will not be made until new, unique cookies are generated" (L2119-2124),
//!   describing an out-of-band retry, not a wire action. This engine surfaces
//!   it as [`RejectionReason::RdvCookie`] (Table 7 code `1009`, "rendezvous
//!   cookie collision" — an exact fit) rather than blocking or retrying
//!   internally; a driver that wants the "regenerate and retry" behaviour
//!   builds a fresh [`RendezvousHandshake`] with a new cookie.
//! - **CONCLUSION `SYN Cookie` field.** The draft specifies WAVEAHAND's SYN
//!   Cookie (L2156-2165) but not what later CONCLUSION/AGREEMENT messages
//!   carry in that field. Every message this engine sends after WAVEAHAND
//!   continues to carry *this side's own* cookie (never an echo of the peer's,
//!   unlike the Caller-Listener flood-protection cookie) — consistent with
//!   the cookie's role here being mutual identification/contest, not a
//!   flood-protection echo-token.
//! - **No Stream ID / Group Membership exchange.** Neither extension is
//!   mentioned anywhere in §4.3.2; unlike
//!   [`crate::handshake_sm::NegotiatedParams`] from the Caller-Listener flow,
//!   this engine never sends them and always reports `stream_id: None`,
//!   `group: None` — flagged rather than fabricated.
//! - **Latency/flags reconciliation.** §4.3.2 does not restate the
//!   greater-latency / AND-flags rule from §4.3.1.2, but it is a property of
//!   the Handshake Extension Message itself (§3.2.1.1), not of the flow that
//!   carried it, so the same shared `handshake_sm` reconciliation helper used
//!   by the Caller-Listener flow is reused unchanged.
//! - **Duplicate WAVEAHAND while already `Attention`.** Not covered by either
//!   table (which only fire on the *first* WAVEAHAND). Treated as a benign
//!   duplicate: resend the last message, no state change.
//! - **Recovery rule 3 (data packet promotes a stuck Responder, L2413-2422)**
//!   is exposed as [`RendezvousHandshake::on_recovery_trigger`], since this
//!   engine's `feed` takes a [`ControlPacket`] (matching
//!   [`crate::caller::CallerHandshake`] / [`crate::listener::ListenerHandshake`]),
//!   not the data-plane [`crate::packet::SrtPacket`]; a driver that receives a
//!   `SrtPacket::Data` (or any Control packet normally only sent between
//!   connected parties — [`RendezvousHandshake::feed`] already treats any
//!   *inbound* non-Handshake `ControlPacket` this way for exactly this
//!   reason) calls it directly.
//!
//! Explicit non-goals, unchanged from the crate root: ARQ/loss, TSBPD
//! delivery, congestion control, AES key-wrap/unwrap crypto, a `tokio` socket
//! adapter, and the Version-4 legacy Rendezvous path (L2101-2105, out of
//! scope of the draft excerpt this crate implements against).

use alloc::vec;
use alloc::vec::Vec;

use crate::error::{Error, Result};
use crate::handshake_sm::{
    self, HANDSHAKE_VERSION_5, HandshakeConfig, HandshakeOutput, NegotiatedParams, RejectionReason,
};
use crate::packet::{
    ControlPacket, EncryptionField, ExtensionType, HandshakeExtensionFlags,
    HandshakeExtensionMessageFlags, HandshakeExtensions, HandshakePacket, HandshakeType,
    HsExtMessage,
};

/// Rendezvous handshake lifecycle state (`draft-sharabayko-srt-01` §4.3.2).
/// State names are exactly the Parallel Handshake Flow diagram (§4.3.2.2,
/// L2280): `Waving -> Attention -> Initiated -> Connected`. See the module
/// doc "Serial vs Parallel flow" note for why there is no separate state for
/// the Serial flow's "fine".
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
#[non_exhaustive]
pub enum RendezvousHandshakeState {
    /// No handshake message sent yet. Not spec-named — mirrors
    /// [`crate::caller::CallerHandshakeState::Idle`].
    Idle,
    /// L2100/L2153/L2280 "waving"/"Waving": both parties' initial state.
    Waving,
    /// L2172/L2280 "attention"/"Attention": a WAVEAHAND was received, the
    /// cookie contest is resolved, awaiting the peer's CONCLUSION.
    Attention,
    /// L2234/L2280 "initiated"/"Initiated": role-appropriate extension
    /// content has been seen at least once; awaiting the peer's confirmation.
    Initiated,
    /// L2224/L2251/L2280 "connected"/"Connected": the handshake completed;
    /// [`NegotiatedParams`] are available.
    Connected,
    /// The handshake was rejected (locally, or by an explicit peer Table 7
    /// code). Not spec-named.
    Rejected,
    /// No response arrived after the configured retry budget. Not spec-named.
    TimedOut,
}

impl RendezvousHandshakeState {
    /// A short label for this state.
    pub fn name(&self) -> &'static str {
        match self {
            RendezvousHandshakeState::Idle => "Idle",
            RendezvousHandshakeState::Waving => "Waving",
            RendezvousHandshakeState::Attention => "Attention",
            RendezvousHandshakeState::Initiated => "Initiated",
            RendezvousHandshakeState::Connected => "Connected",
            RendezvousHandshakeState::Rejected => "Rejected",
            RendezvousHandshakeState::TimedOut => "TimedOut",
        }
    }
}

impl core::fmt::Display for RendezvousHandshakeState {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        f.write_str(self.name())
    }
}

/// The role a [`RendezvousHandshake`] plays, resolved by the cookie contest
/// (`draft-sharabayko-srt-01` §4.3.2, L2107-2135): "When one party's cookie
/// value is greater than its peer's, it wins the cookie contest and becomes
/// Initiator (the other party becomes the Responder)."
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
#[non_exhaustive]
pub enum RendezvousRole {
    /// Wins the cookie contest (the greater cookie value, L2133-2135). MUST
    /// attach the HSREQ extension (L2286-2287).
    Initiator,
    /// Loses the cookie contest. MUST attach the HSRSP extension
    /// (L2287-2288).
    Responder,
}

impl RendezvousRole {
    /// A short label for this role.
    pub fn name(&self) -> &'static str {
        match self {
            RendezvousRole::Initiator => "Initiator",
            RendezvousRole::Responder => "Responder",
        }
    }
}

broadcast_common::impl_spec_display!(RendezvousRole);

/// The Handshake Extension content found on one received CONCLUSION, keyed by
/// which side sends which (§4.3.2.2, L2282-2288). Internal to this module —
/// [`crate::handshake_sm::parse_peer_extensions`] does not distinguish
/// HSREQ from HSRSP, which the Rendezvous tables need to.
enum PeerHsExt {
    /// No Handshake Extension block present.
    None,
    /// An `HSREQ` block was present (only an Initiator sends this).
    HsReq(HsExtMessage),
    /// An `HSRSP` block was present (only a Responder sends this).
    HsRsp(HsExtMessage),
}

/// Walks `hp`'s extension blocks looking for an HSREQ or HSRSP Handshake
/// Extension Message. Returns [`Error::InvalidField`] on any malformed block
/// or malformed HSREQ/HSRSP contents — never panics on untrusted input.
fn parse_peer_hs_ext(hp: &HandshakePacket<'_>) -> Result<PeerHsExt> {
    let mut found = PeerHsExt::None;
    for block in hp.extensions.iter() {
        let block = block.map_err(|_| Error::InvalidField {
            what: "rendezvous handshake extensions",
            reason: "malformed extension block",
        })?;
        match block.ext_type {
            ExtensionType::HsReq => {
                let msg = block.as_hs_ext_message().map_err(|_| Error::InvalidField {
                    what: "HSREQ extension message",
                    reason: "malformed contents",
                })?;
                found = PeerHsExt::HsReq(msg);
            }
            ExtensionType::HsRsp => {
                let msg = block.as_hs_ext_message().map_err(|_| Error::InvalidField {
                    what: "HSRSP extension message",
                    reason: "malformed contents",
                })?;
                found = PeerHsExt::HsRsp(msg);
            }
            _ => {}
        }
    }
    Ok(found)
}

/// A driveable, symmetric SRT Rendezvous handshake
/// (`draft-sharabayko-srt-01` §4.3.2). See the module docs for the state
/// machine and the design decisions not explicit in the curated rules.
#[derive(Debug)]
pub struct RendezvousHandshake {
    own_socket_id: u32,
    own_cookie: u32,
    config: HandshakeConfig,
    state: RendezvousHandshakeState,
    role: Option<RendezvousRole>,
    peer_socket_id: u32,
    peer_cookie: u32,
    peer_hs_msg: Option<HsExtMessage>,
    last_sent: Option<Vec<u8>>,
    ticks_since_send: u32,
    retries: u32,
    negotiated: Option<NegotiatedParams>,
}

impl RendezvousHandshake {
    /// Creates a fresh Rendezvous handshake in
    /// [`RendezvousHandshakeState::Idle`], with the 32-bit cookie this side
    /// will offer in the cookie contest (§4.3.2, L2107-2135). This crate never
    /// reads a clock or a socket address — see
    /// [`crate::handshake_sm::derive_cookie`] for a ready-made,
    /// non-standardized derivation helper.
    pub fn new(own_socket_id: u32, own_cookie: u32, config: HandshakeConfig) -> Self {
        RendezvousHandshake {
            own_socket_id,
            own_cookie,
            config,
            state: RendezvousHandshakeState::Idle,
            role: None,
            peer_socket_id: 0,
            peer_cookie: 0,
            peer_hs_msg: None,
            last_sent: None,
            ticks_since_send: 0,
            retries: 0,
            negotiated: None,
        }
    }

    /// The current state.
    pub fn state(&self) -> RendezvousHandshakeState {
        self.state
    }

    /// The role resolved by the cookie contest, once past
    /// [`RendezvousHandshakeState::Waving`].
    pub fn role(&self) -> Option<RendezvousRole> {
        self.role
    }

    /// The negotiated parameters, once [`RendezvousHandshakeState::Connected`].
    pub fn negotiated(&self) -> Option<&NegotiatedParams> {
        self.negotiated.as_ref()
    }

    /// Builds the initial WAVEAHAND handshake (§4.3.2, L2100/L2153-2170:
    /// Version 5, this side's own cookie, no extensions) and transitions to
    /// [`RendezvousHandshakeState::Waving`].
    ///
    /// # Errors
    /// [`Error::HandshakeOutOfSequence`] if called more than once.
    pub fn start(&mut self) -> Result<Vec<u8>> {
        if self.state != RendezvousHandshakeState::Idle {
            return Err(Error::HandshakeOutOfSequence {
                state: self.state.name(),
                reason: "start() called after the handshake already began",
            });
        }
        let hp = HandshakePacket {
            timestamp: 0,
            dest_socket_id: 0, // Unknown yet — mirrors CallerHandshake::start's INDUCTION.
            version: HANDSHAKE_VERSION_5,
            encryption_field: self.config.encryption_field,
            extension_field: HandshakeExtensionFlags(0),
            initial_seq_number: self.config.initial_seq_number,
            mtu: self.config.mtu,
            max_flow_window_size: self.config.max_flow_window_size,
            handshake_type: HandshakeType::Wavehand,
            srt_socket_id: self.own_socket_id,
            syn_cookie: self.own_cookie,
            peer_ip: self.config.local_ip,
            extensions: HandshakeExtensions(&[]),
        };
        let bytes = handshake_sm::build_bytes(hp)?;
        self.last_sent = Some(bytes.clone());
        self.ticks_since_send = 0;
        self.state = RendezvousHandshakeState::Waving;
        Ok(bytes)
    }

    /// Feeds an inbound control packet.
    ///
    /// A non-Handshake `ControlPacket` fed while this side is a Responder
    /// stuck in [`RendezvousHandshakeState::Initiated`] is treated as
    /// missing-packet recovery rule 3 (L2413-2422: "any control packet
    /// normally only sent between connected parties") rather than an error —
    /// see [`Self::on_recovery_trigger`] for the data-packet analogue.
    ///
    /// # Errors
    /// [`Error::UnexpectedControlPacket`] for a non-Handshake packet outside
    /// that one recovery case; [`Error::HandshakeOutOfSequence`] if fed before
    /// [`Self::start`] or after a terminal state (a driver bug, not a peer
    /// protocol failure).
    pub fn feed(&mut self, packet: &ControlPacket<'_>) -> Result<Vec<HandshakeOutput>> {
        let hp = match packet {
            ControlPacket::Handshake(hp) => hp,
            other => {
                if self.state == RendezvousHandshakeState::Initiated
                    && self.role == Some(RendezvousRole::Responder)
                {
                    return self.enter_connected();
                }
                return Err(Error::UnexpectedControlPacket {
                    actual: other.control_type().name(),
                });
            }
        };
        if let Some(reason) = RejectionReason::from_handshake_type(hp.handshake_type) {
            return Ok(self.reject(reason));
        }
        if hp.version != HANDSHAKE_VERSION_5 {
            // §4.3.2, L2101-2105: Version-4 legacy Rendezvous is out of scope.
            return Ok(self.reject(RejectionReason::Version));
        }
        match self.state {
            RendezvousHandshakeState::Idle => Err(Error::HandshakeOutOfSequence {
                state: self.state.name(),
                reason: "feed() called before start()",
            }),
            RendezvousHandshakeState::Waving => self.on_waving(hp),
            RendezvousHandshakeState::Attention => self.on_attention(hp),
            RendezvousHandshakeState::Initiated => self.on_initiated(hp),
            RendezvousHandshakeState::Connected => self.on_connected(hp),
            RendezvousHandshakeState::Rejected | RendezvousHandshakeState::TimedOut => {
                Err(Error::HandshakeOutOfSequence {
                    state: self.state.name(),
                    reason: "handshake already reached a terminal state",
                })
            }
        }
    }

    /// Convenience wrapper: parses `bytes` as a [`ControlPacket`] then
    /// [`Self::feed`]s it.
    pub fn feed_bytes(&mut self, bytes: &[u8]) -> Result<Vec<HandshakeOutput>> {
        let packet = ControlPacket::parse(bytes)?;
        self.feed(&packet)
    }

    /// Missing-packet recovery rule 3 (§4.3.2.2, L2413-2422): call this when
    /// the driver receives a data packet (`SrtPacket::Data`) while this side
    /// has not yet reached [`RendezvousHandshakeState::Connected`]. A no-op
    /// (`Ok(Vec::new())`) unless this side is a Responder stuck in
    /// [`RendezvousHandshakeState::Initiated`] — the one case the draft says
    /// is "exceptionally allowed" to promote to Connected "as if it had
    /// received AGREEMENT".
    pub fn on_recovery_trigger(&mut self) -> Result<Vec<HandshakeOutput>> {
        if self.state == RendezvousHandshakeState::Initiated
            && self.role == Some(RendezvousRole::Responder)
        {
            self.enter_connected()
        } else {
            Ok(Vec::new())
        }
    }

    /// Advances retransmit timing by one caller-defined tick, mirroring
    /// [`crate::caller::CallerHandshake::tick`] /
    /// [`crate::listener::ListenerHandshake::tick`].
    pub fn tick(&mut self) -> Vec<HandshakeOutput> {
        if matches!(
            self.state,
            RendezvousHandshakeState::Idle
                | RendezvousHandshakeState::Connected
                | RendezvousHandshakeState::Rejected
                | RendezvousHandshakeState::TimedOut
        ) {
            return Vec::new();
        }
        self.ticks_since_send += 1;
        if self.ticks_since_send < self.config.retransmit_after_ticks {
            return Vec::new();
        }
        self.ticks_since_send = 0;
        self.retries += 1;
        if self.retries > self.config.max_retries {
            self.state = RendezvousHandshakeState::TimedOut;
            return vec![HandshakeOutput::TimedOut];
        }
        match self.last_sent.clone() {
            Some(bytes) => vec![HandshakeOutput::Send(bytes)],
            None => Vec::new(),
        }
    }

    /// [`RendezvousHandshakeState::Waving`]: the first message ever received
    /// from the peer — either a genuine WAVEAHAND (§4.3.2.2 Waving row,
    /// L2301-2306 Initiator / L2336-2342 Responder) or, in the Serial flow's
    /// crossing case, a CONCLUSION directly (see module docs). Either way the
    /// cookie contest is resolved here, from `hp.syn_cookie`.
    fn on_waving(&mut self, hp: &HandshakePacket<'_>) -> Result<Vec<HandshakeOutput>> {
        if !matches!(
            hp.handshake_type,
            HandshakeType::Wavehand | HandshakeType::Conclusion
        ) {
            return Ok(self.reject(RejectionReason::Rogue));
        }
        self.peer_socket_id = hp.srt_socket_id;
        let role = match self.resolve_role(hp.syn_cookie) {
            Ok(r) => r,
            Err(_) => return Ok(self.reject(RejectionReason::RdvCookie)),
        };
        self.peer_cookie = hp.syn_cookie;
        self.role = Some(role);

        if hp.handshake_type == HandshakeType::Wavehand {
            self.state = RendezvousHandshakeState::Attention;
            match role {
                // Initiator Waving row (L2301-2306): send CONCLUSION+HSREQ.
                RendezvousRole::Initiator => self.send_conclusion(Some(ExtensionType::HsReq)),
                // Responder Waving row (L2336-2342): send CONCLUSION, no
                // extensions (it has not seen the peer's HSREQ yet).
                RendezvousRole::Responder => self.send_conclusion(None),
            }
        } else {
            // Serial-flow crossing (§4.3.2.1 step 3, L2203-2219): apply the
            // same content-driven Attention-row logic (module docs).
            self.on_attention_conclusion(hp, role)
        }
    }

    /// [`RendezvousHandshakeState::Attention`]: a WAVEAHAND was already seen
    /// and processed; only a CONCLUSION (Attention row) or a duplicate
    /// WAVEAHAND (not covered by either table — treated as a benign resend,
    /// module docs) is expected here.
    fn on_attention(&mut self, hp: &HandshakePacket<'_>) -> Result<Vec<HandshakeOutput>> {
        let role = self
            .role
            .expect("role is always resolved before Attention is reached");
        if hp.handshake_type == HandshakeType::Conclusion {
            return self.on_attention_conclusion(hp, role);
        }
        if hp.handshake_type == HandshakeType::Wavehand {
            return Ok(self.resend());
        }
        Ok(self.reject(RejectionReason::Rogue))
    }

    /// The Attention row's CONCLUSION-handling logic (§4.3.2.2, L2307-2320
    /// Initiator / L2343-2364 Responder), shared between a genuine
    /// [`RendezvousHandshakeState::Attention`] and the Serial-flow crossing
    /// case fed straight from [`RendezvousHandshakeState::Waving`].
    fn on_attention_conclusion(
        &mut self,
        hp: &HandshakePacket<'_>,
        role: RendezvousRole,
    ) -> Result<Vec<HandshakeOutput>> {
        self.peer_socket_id = hp.srt_socket_id;
        let ext = match parse_peer_hs_ext(hp) {
            Ok(e) => e,
            Err(_) => return Ok(self.reject(RejectionReason::Rogue)),
        };
        match (role, ext) {
            (RendezvousRole::Initiator, PeerHsExt::None) => {
                // L2312-2318: no extensions -> Initiated, still send
                // CONCLUSION+HSREQ.
                self.state = RendezvousHandshakeState::Initiated;
                self.send_conclusion(Some(ExtensionType::HsReq))
            }
            (RendezvousRole::Initiator, PeerHsExt::HsRsp(msg)) => {
                // L2318-2320: contains HSRSP -> Connected, send AGREEMENT.
                self.peer_hs_msg = Some(msg);
                self.enter_connected()
            }
            (RendezvousRole::Responder, PeerHsExt::None) => {
                // L2357-2360: no extensions yet -> resend the empty
                // CONCLUSION, remain in Attention.
                self.state = RendezvousHandshakeState::Attention;
                self.send_conclusion(None)
            }
            (RendezvousRole::Responder, PeerHsExt::HsReq(msg)) => {
                // L2360-2364: HSREQ present -> Initiated, send
                // CONCLUSION+HSRSP.
                self.peer_hs_msg = Some(msg);
                self.state = RendezvousHandshakeState::Initiated;
                self.send_conclusion(Some(ExtensionType::HsRsp))
            }
            // A peer sending the extension kind only its own role should ever
            // send (HSREQ into an Initiator, HSRSP into a Responder) is not a
            // transition either table defines — reject rather than guess.
            _ => Ok(self.reject(RejectionReason::Rogue)),
        }
    }

    /// [`RendezvousHandshakeState::Initiated`] (§4.3.2.2, L2321-2334 Initiator
    /// / L2365-2382 Responder), including the idempotent-resend recovery
    /// rules (L2383-2422).
    fn on_initiated(&mut self, hp: &HandshakePacket<'_>) -> Result<Vec<HandshakeOutput>> {
        let role = self
            .role
            .expect("role is always resolved before Initiated is reached");
        match role {
            RendezvousRole::Initiator => {
                if hp.handshake_type != HandshakeType::Conclusion {
                    return Ok(self.reject(RejectionReason::Rogue));
                }
                let ext = match parse_peer_hs_ext(hp) {
                    Ok(e) => e,
                    Err(_) => return Ok(self.reject(RejectionReason::Rogue)),
                };
                match ext {
                    // L2325: "REMAINS IN THIS STATE" — still resend
                    // CONCLUSION+HSREQ.
                    PeerHsExt::None => self.send_conclusion(Some(ExtensionType::HsReq)),
                    // L2325-2334ish: contains HSRSP -> Connected, AGREEMENT.
                    PeerHsExt::HsRsp(msg) => {
                        self.peer_hs_msg = Some(msg);
                        self.enter_connected()
                    }
                    PeerHsExt::HsReq(_) => Ok(self.reject(RejectionReason::Rogue)),
                }
            }
            RendezvousRole::Responder => {
                if hp.handshake_type == HandshakeType::Agreement {
                    // Responder Initiated row: AGREEMENT -> respond AGREEMENT,
                    // switch to Connected.
                    return self.enter_connected();
                }
                if hp.handshake_type != HandshakeType::Conclusion {
                    return Ok(self.reject(RejectionReason::Rogue));
                }
                let ext = match parse_peer_hs_ext(hp) {
                    Ok(e) => e,
                    Err(_) => return Ok(self.reject(RejectionReason::Rogue)),
                };
                match ext {
                    // Recovery rule 2 (L2391-2395): MUST always resend HSRSP,
                    // even if this HSREQ was already seen and processed once.
                    PeerHsExt::HsReq(msg) => {
                        self.peer_hs_msg = Some(msg);
                        self.send_conclusion(Some(ExtensionType::HsRsp))
                    }
                    _ => Ok(self.reject(RejectionReason::Rogue)),
                }
            }
        }
    }

    /// [`RendezvousHandshakeState::Connected`] (Initiator row item 4,
    /// L2331-2334; Responder row item 4, L2377-2381): normally no more
    /// handshake traffic, but a repeated CONCLUSION is answered with another
    /// AGREEMENT (recovery rule 4, L2424-2432, from the other side's point of
    /// view); anything else is disregarded rather than regressing a completed
    /// handshake.
    fn on_connected(&mut self, hp: &HandshakePacket<'_>) -> Result<Vec<HandshakeOutput>> {
        if hp.handshake_type == HandshakeType::Conclusion {
            return self.send_agreement();
        }
        Ok(Vec::new())
    }

    /// The cookie contest (§4.3.2, L2107-2135). Identical cookies are a
    /// collision the draft says must not connect (L2119-2124) — surfaced by
    /// the caller as [`RejectionReason::RdvCookie`].
    fn resolve_role(&self, peer_cookie: u32) -> Result<RendezvousRole> {
        if peer_cookie == self.own_cookie {
            return Err(Error::InvalidField {
                what: "rendezvous cookie",
                reason: "identical to the peer's cookie (collision, L2119-2124)",
            });
        }
        Ok(if self.own_cookie > peer_cookie {
            RendezvousRole::Initiator
        } else {
            RendezvousRole::Responder
        })
    }

    fn reject(&mut self, reason: RejectionReason) -> Vec<HandshakeOutput> {
        self.state = RendezvousHandshakeState::Rejected;
        vec![HandshakeOutput::Rejected(reason)]
    }

    fn resend(&mut self) -> Vec<HandshakeOutput> {
        match &self.last_sent {
            Some(bytes) => vec![HandshakeOutput::Send(bytes.clone())],
            None => Vec::new(),
        }
    }

    /// Builds and sends a CONCLUSION: `ext_type` picks HSREQ/HSRSP, or `None`
    /// for the extension-less greeting (§4.3.2.2, L2286-2288: no Stream ID /
    /// Group Membership is ever attached — see module docs).
    fn send_conclusion(&mut self, ext_type: Option<ExtensionType>) -> Result<Vec<HandshakeOutput>> {
        let (ext_bytes, ext_flags): (Vec<u8>, u16) = match ext_type {
            None => (Vec::new(), 0),
            Some(t) => {
                let hs_msg = HsExtMessage {
                    srt_version: self.config.srt_version,
                    srt_flags: self.config.flags,
                    receiver_tsbpd_delay_ms: self.config.latency_ms,
                    sender_tsbpd_delay_ms: self.config.latency_ms,
                };
                handshake_sm::build_conclusion_extensions(t, &hs_msg, None, None)?
            }
        };
        let hp = HandshakePacket {
            timestamp: 0,
            dest_socket_id: self.peer_socket_id,
            version: HANDSHAKE_VERSION_5,
            encryption_field: self.config.encryption_field,
            extension_field: HandshakeExtensionFlags(ext_flags),
            initial_seq_number: self.config.initial_seq_number,
            mtu: self.config.mtu,
            max_flow_window_size: self.config.max_flow_window_size,
            handshake_type: HandshakeType::Conclusion,
            srt_socket_id: self.own_socket_id,
            syn_cookie: self.own_cookie,
            peer_ip: self.config.local_ip,
            extensions: HandshakeExtensions(&ext_bytes),
        };
        let bytes = handshake_sm::build_bytes(hp)?;
        self.last_sent = Some(bytes.clone());
        self.ticks_since_send = 0;
        self.retries = 0;
        Ok(vec![HandshakeOutput::Send(bytes)])
    }

    /// Builds and sends an AGREEMENT: no extensions (§4.3.2.1, L2224-2230).
    fn send_agreement(&mut self) -> Result<Vec<HandshakeOutput>> {
        let hp = HandshakePacket {
            timestamp: 0,
            dest_socket_id: self.peer_socket_id,
            version: HANDSHAKE_VERSION_5,
            encryption_field: EncryptionField::NoEncryption,
            extension_field: HandshakeExtensionFlags(0),
            initial_seq_number: self.config.initial_seq_number,
            mtu: self.config.mtu,
            max_flow_window_size: self.config.max_flow_window_size,
            handshake_type: HandshakeType::Agreement,
            srt_socket_id: self.own_socket_id,
            syn_cookie: self.own_cookie,
            peer_ip: self.config.local_ip,
            extensions: HandshakeExtensions(&[]),
        };
        let bytes = handshake_sm::build_bytes(hp)?;
        self.last_sent = Some(bytes.clone());
        self.ticks_since_send = 0;
        self.retries = 0;
        Ok(vec![HandshakeOutput::Send(bytes)])
    }

    /// Reaches [`RendezvousHandshakeState::Connected`]: builds
    /// [`NegotiatedParams`] from `self.peer_hs_msg` (always captured by every
    /// caller of this method before it is called) and sends the AGREEMENT
    /// every documented transition into Connected requires.
    fn enter_connected(&mut self) -> Result<Vec<HandshakeOutput>> {
        let negotiated = self.build_negotiated();
        self.negotiated = Some(negotiated.clone());
        self.state = RendezvousHandshakeState::Connected;
        let mut out = self.send_agreement()?;
        out.push(HandshakeOutput::Connected(negotiated));
        Ok(out)
    }

    fn build_negotiated(&self) -> NegotiatedParams {
        let peer_msg = self
            .peer_hs_msg
            .expect("peer_hs_msg is always captured before any transition reaches Connected");
        NegotiatedParams {
            version: HANDSHAKE_VERSION_5,
            flags: HandshakeExtensionMessageFlags(self.config.flags.0 & peer_msg.srt_flags.0),
            latency_ms: handshake_sm::negotiate_latency_ms(self.config.latency_ms, &peer_msg),
            own_socket_id: self.own_socket_id,
            peer_socket_id: self.peer_socket_id,
            // §4.3.2 never mentions a Stream ID / Group Membership exchange —
            // module docs "Resolved ambiguities". The same applies to §6.1.5
            // Key Material Exchange: this engine does not implement it (see
            // the crate root docs' explicit-follow-ups list), so encryption
            // is never negotiated over Rendezvous.
            stream_id: None,
            group: None,
            #[cfg(feature = "crypto")]
            sek: None,
            #[cfg(feature = "crypto")]
            salt: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::packet::handshake::HS_EXT_FLAG_HSREQ;

    #[test]
    fn start_is_idempotent_guard() {
        let mut r = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        assert!(r.start().is_ok());
        assert!(r.start().is_err());
    }

    #[test]
    fn wavehand_wire_values_match_draft_4_3_2() {
        let mut r = RendezvousHandshake::new(0xAAAA_BBBB, 0xC0FF_EE00, HandshakeConfig::default());
        let bytes = r.start().unwrap();
        let pkt = ControlPacket::parse(&bytes).unwrap();
        match pkt {
            ControlPacket::Handshake(hp) => {
                assert_eq!(hp.version, HANDSHAKE_VERSION_5);
                assert_eq!(hp.handshake_type, HandshakeType::Wavehand);
                assert_eq!(hp.srt_socket_id, 0xAAAA_BBBB);
                assert_eq!(hp.syn_cookie, 0xC0FF_EE00);
                assert_eq!(hp.extension_field.0, 0);
            }
            _ => panic!("expected handshake"),
        }
        assert_eq!(r.state(), RendezvousHandshakeState::Waving);
    }

    fn wavehand(socket_id: u32, cookie: u32) -> ControlPacket<'static> {
        ControlPacket::Handshake(HandshakePacket {
            timestamp: 0,
            dest_socket_id: 0,
            version: HANDSHAKE_VERSION_5,
            encryption_field: EncryptionField::NoEncryption,
            extension_field: HandshakeExtensionFlags(0),
            initial_seq_number: 0,
            mtu: 1500,
            max_flow_window_size: 8192,
            handshake_type: HandshakeType::Wavehand,
            srt_socket_id: socket_id,
            syn_cookie: cookie,
            peer_ip: [0; 4],
            extensions: HandshakeExtensions(&[]),
        })
    }

    #[test]
    fn greater_cookie_wins_initiator() {
        let mut a = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        let mut b = RendezvousHandshake::new(2, 100, HandshakeConfig::default());
        a.start().unwrap();
        b.start().unwrap();

        a.feed(&wavehand(2, 100)).unwrap();
        b.feed(&wavehand(1, 500)).unwrap();

        assert_eq!(a.role(), Some(RendezvousRole::Initiator));
        assert_eq!(b.role(), Some(RendezvousRole::Responder));
        assert_eq!(a.state(), RendezvousHandshakeState::Attention);
        assert_eq!(b.state(), RendezvousHandshakeState::Attention);
    }

    #[test]
    fn identical_cookies_are_rejected_as_a_collision() {
        let mut a = RendezvousHandshake::new(1, 0x00C0_FFEE, HandshakeConfig::default());
        a.start().unwrap();
        let outputs = a.feed(&wavehand(2, 0x00C0_FFEE)).unwrap();
        assert_eq!(
            outputs,
            vec![HandshakeOutput::Rejected(RejectionReason::RdvCookie)]
        );
        assert_eq!(a.state(), RendezvousHandshakeState::Rejected);
    }

    #[test]
    fn initiator_attention_entry_sends_hsreq() {
        let mut a = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        a.start().unwrap();
        let outputs = a.feed(&wavehand(2, 100)).unwrap();
        assert_eq!(outputs.len(), 1);
        let bytes = match &outputs[0] {
            HandshakeOutput::Send(b) => b.clone(),
            other => panic!("expected Send, got {other:?}"),
        };
        let pkt = ControlPacket::parse(&bytes).unwrap();
        match pkt {
            ControlPacket::Handshake(hp) => {
                assert_eq!(hp.handshake_type, HandshakeType::Conclusion);
                assert_eq!(hp.extension_field.0 & HS_EXT_FLAG_HSREQ, HS_EXT_FLAG_HSREQ);
                let blocks: Vec<_> = hp.extensions.iter().map(|b| b.unwrap()).collect();
                assert_eq!(blocks.len(), 1);
                assert_eq!(blocks[0].ext_type, ExtensionType::HsReq);
            }
            _ => panic!("expected handshake"),
        }
    }

    #[test]
    fn responder_attention_entry_sends_empty_conclusion() {
        let mut b = RendezvousHandshake::new(2, 100, HandshakeConfig::default());
        b.start().unwrap();
        let outputs = b.feed(&wavehand(1, 500)).unwrap();
        assert_eq!(outputs.len(), 1);
        let bytes = match &outputs[0] {
            HandshakeOutput::Send(b) => b.clone(),
            other => panic!("expected Send, got {other:?}"),
        };
        let pkt = ControlPacket::parse(&bytes).unwrap();
        match pkt {
            ControlPacket::Handshake(hp) => {
                assert_eq!(hp.handshake_type, HandshakeType::Conclusion);
                assert_eq!(hp.extension_field.0, 0);
                assert_eq!(hp.extensions.iter().count(), 0);
            }
            _ => panic!("expected handshake"),
        }
    }

    #[test]
    fn malformed_extension_mid_flow_is_rejected_not_panicking() {
        let mut a = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        a.start().unwrap();
        a.feed(&wavehand(2, 100)).unwrap();
        assert_eq!(a.state(), RendezvousHandshakeState::Attention);

        // A CONCLUSION whose extension block declares a length far larger
        // than the bytes actually present.
        let bad_ext: &'static [u8] = &[0x00, 0x01, 0xFF, 0xFF];
        let bad = ControlPacket::Handshake(HandshakePacket {
            timestamp: 0,
            dest_socket_id: 1,
            version: HANDSHAKE_VERSION_5,
            encryption_field: EncryptionField::NoEncryption,
            extension_field: HandshakeExtensionFlags(HS_EXT_FLAG_HSREQ),
            initial_seq_number: 0,
            mtu: 1500,
            max_flow_window_size: 8192,
            handshake_type: HandshakeType::Conclusion,
            srt_socket_id: 2,
            syn_cookie: 100,
            peer_ip: [0; 4],
            extensions: HandshakeExtensions(bad_ext),
        });
        let outputs = a.feed(&bad).unwrap();
        assert_eq!(
            outputs,
            vec![HandshakeOutput::Rejected(RejectionReason::Rogue)]
        );
        assert_eq!(a.state(), RendezvousHandshakeState::Rejected);
    }

    #[test]
    fn feed_before_start_is_out_of_sequence() {
        let mut r = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        assert!(matches!(
            r.feed(&wavehand(2, 100)),
            Err(Error::HandshakeOutOfSequence { .. })
        ));
    }

    #[test]
    fn feed_rejects_non_handshake_packets_outside_recovery_case() {
        use crate::packet::misc::KeepAlivePacket;
        let mut r = RendezvousHandshake::new(1, 500, HandshakeConfig::default());
        r.start().unwrap();
        let ka = ControlPacket::KeepAlive(KeepAlivePacket {
            timestamp: 0,
            dest_socket_id: 0,
        });
        assert!(matches!(
            r.feed(&ka),
            Err(Error::UnexpectedControlPacket { .. })
        ));
    }
}