Fix protocol conformance issues found in interop testing

Implement the Poll Sequence (initiate on reaching Up when the configured interval is lower than the slow-start one; respond to Poll with Final; terminate on Final) and apply DesiredMinTxInterval changes per RFC 5880 section 6.8.3, keeping the slow rate until the peer has acknowledged. Never set the Demand bit unless demand mode is active, clear the C bit (this implementation shares fate with the control plane), discard packets whose Length exceeds the received payload, and log effective timer changes.

src/bin/rust-bfdd.rs

@@ -242,6 +242,10 @@ struct BfdSession {
     tx: mpsc::Sender<SessionControlCommand>,
     rx_watchdog: Option<mpsc::Sender<WatchdogReset>>,
     cur_interval: BfdInterval,
+    // Configured DesiredMinTxInterval; advertised only once Up (RFC 5880 6.8.3 requires >= 1s otherwise)
+    configured_desired_min_tx: BfdInterval,
+    // DesiredMinTxInterval actually used for the transmit rate; lags the advertised value during a Poll Sequence
+    effective_desired_min_tx: BfdInterval,
     event_dispatcher: EventDispatcherHandle,
 }
 
@@ -424,6 +428,9 @@ impl BfdSession {
         event_dispatcher: EventDispatcherHandle,
     ) -> Self {
         let mut rng = rand::thread_rng();
+        let configured_desired_min_tx = BfdInterval::from_millis(300);
+        let slow_desired_min_tx =
+            std::cmp::max(configured_desired_min_tx, BfdInterval::from_secs(1));
 
         // control_sock
         //     .connect(SocketAddr::new(remote_addr, CONTROL_PORT))
@@ -436,6 +443,8 @@ impl BfdSession {
             periodic,
             rx_watchdog: None,
             cur_interval: BfdInterval::from_secs(1),
+            configured_desired_min_tx,
+            effective_desired_min_tx: slow_desired_min_tx,
             event_dispatcher,
             state: BfdSessionState {
                 control_sock: Arc::new(control_sock),
@@ -446,7 +455,7 @@ impl BfdSession {
                 local_discr,
                 remote_discr: Atomic::new(BfdDiscriminator(0)),
                 local_diag: Atomic::new(BfdDiagnostic::None),
-                desired_min_tx_interval: Atomic::new(BfdInterval::from_millis(300)),
+                desired_min_tx_interval: Atomic::new(slow_desired_min_tx),
                 required_min_rx_interval: Atomic::new(BfdInterval::from_millis(300)),
                 remote_min_rx_interval: Atomic::new(BfdInterval::from_millis(300)),
                 demand_mode: Atomic::new(false),
@@ -502,15 +511,27 @@ impl BfdSession {
                 //    less than one second (1,000,000 microseconds).  This is intended to ensure that the bandwidth
                 //    consumed by BFD sessions that are not Up is negligible, particularly in the case where a neighbor
                 //    may not be running BFD.
-                self.state.set_desired_min_tx_interval(std::cmp::min(
+                self.state.set_desired_min_tx_interval(std::cmp::max(
-                    self.state.desired_min_tx_interval(),
+                    self.configured_desired_min_tx,
                     BfdInterval::from_secs(1),
                 ));
+                self.state.set_poll_mode(false);
+                self.set_effective_desired_min_tx(self.state.desired_min_tx_interval());
                 self.update_transmit_interval().await;
             }
             (BfdState::AdminDown | BfdState::Down | BfdState::Init, BfdState::Up) => {
                 self.start_watchdog().await;
                 self.state.set_local_diag(BfdDiagnostic::None); // This doesn't seem to be required by the spec but seems sensible and at least FRR does it.
+                                                                // Advertise the configured interval now, but per section 6.8.3 keep transmitting at the old (slower)
+                                                                // rate until a Poll Sequence confirms the peer has seen the change.
+                self.state
+                    .set_desired_min_tx_interval(self.configured_desired_min_tx);
+                if self.configured_desired_min_tx < self.effective_desired_min_tx {
+                    self.state.set_poll_mode(true);
+                } else {
+                    self.set_effective_desired_min_tx(self.configured_desired_min_tx);
+                    self.update_transmit_interval().await;
+                }
             }
             _ => {}
         }
@@ -553,11 +574,8 @@ impl BfdSession {
                     self.state.poll_mode()
                 })
                 .with_final_(poll_response)
-                .with_cpi(true)
+                .with_cpi(false)
-                .with_demand(
+                .with_demand(self.state.demand_mode())
-                    self.state.session_state() == BfdState::Up
-                        && self.state.remote_session_state() == BfdState::Up,
-                )
                 .with_detect_mult(self.state.detect_mult())
                 .with_length(24),
             my_disc: self.state.local_discr(),
@@ -671,8 +689,6 @@ impl BfdSession {
             warn!("Invalid packet length {}, discarding", p.flags.length());
             return;
         }
-        // TODO: If the Length field is greater than the payload of the encapsulating protocol, the packet MUST be
-        //   discarded.
 
         // If the Detect Mult field is zero, the packet MUST be discarded.
         if p.flags.detect_mult() == 0 {
@@ -749,8 +765,10 @@ impl BfdSession {
         }
         // If a Poll Sequence is being transmitted by the local system and the Final (F) bit in the received packet is
         //   set, the Poll Sequence MUST be terminated.
-        //
+        if self.state.poll_mode() && p.flags.final_() {
-        // TODO: poll stuff
+            self.state.set_poll_mode(false);
+            self.set_effective_desired_min_tx(self.state.desired_min_tx_interval());
+        }
 
         // Update the transmit interval as described in section 6.8.2.
         self.update_transmit_interval().await;
@@ -825,19 +843,39 @@ impl BfdSession {
 
             // If the Poll (P) bit is set, send a BFD Control packet to the remote system with the Poll (P) bit clear,
             // and the Final (F) bit set (see section 6.8.7).
-            if p.flags.poll() {}
+            if p.flags.poll() {
+                self.transmit_control_packet(true).await;
+            }
             // If the packet was not discarded, it has been received for purposes of the Detection Time expiration rules
             // in section 6.8.4.
         }
     }
 
+    fn set_effective_desired_min_tx(&mut self, value: BfdInterval) {
+        if value != self.effective_desired_min_tx {
+            info!(
+                "Peer {} effective desired-min-tx {} -> {}",
+                self.state.peer_addr(),
+                self.effective_desired_min_tx,
+                value
+            );
+        }
+        self.effective_desired_min_tx = value;
+    }
+
     // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.2
     async fn update_transmit_interval(&mut self) {
         let new_base_interval = std::cmp::max(
-            self.state.desired_min_tx_interval(),
+            self.effective_desired_min_tx,
             self.state.remote_min_rx_interval(),
         );
         if new_base_interval != self.cur_interval {
+            info!(
+                "Peer {} transmit interval {} -> {}",
+                self.state.peer_addr(),
+                self.cur_interval,
+                new_base_interval
+            );
             self.periodic
                 .tx
                 .send(PeriodicControlCommand::SetMinInterval(new_base_interval))
@@ -863,6 +901,12 @@ impl BfdSession {
             .into()
         });
         if old != self.state.detection_time() {
+            info!(
+                "Peer {} detection time {} -> {}",
+                self.state.peer_addr(),
+                old,
+                self.state.detection_time()
+            );
             self.start_watchdog().await
         }
     }
@@ -1097,6 +1141,12 @@ impl Bfdd {
                     Ok((len, addr)) => {
                         if let Ok((_leftover, packet)) = BfdPacket::parse(&buf[..len]) {
                             debug!("rx packet ({:?}): {:?}", addr.ip(), packet);
+                            // If the Length field is greater than the payload of the encapsulating protocol, the packet MUST be
+                            // discarded.
+                            if packet.flags.length() as usize > len {
+                                warn!("discarding packet with length > payload");
+                                continue;
+                            }
                             self.rx_packet(addr, packet).await
                         } else {
                             warn!("Failed to parse packet");