netget/server/igmp/mod.rs
1//! IGMP server implementation using raw IP sockets
2pub mod actions;
3
4use crate::server::connection::ConnectionId;
5use actions::IgmpProtocol;
6use anyhow::{Context, Result};
7use socket2::Socket;
8use std::collections::HashSet;
9use std::net::{Ipv4Addr, SocketAddr};
10use std::os::fd::FromRawFd;
11use std::sync::Arc;
12use tokio::sync::{mpsc, Mutex};
13use tracing::{debug, info};
14
15use crate::llm::action_helper::call_llm;
16use crate::llm::ollama_client::OllamaClient;
17use crate::logging::emit::Log;
18use crate::protocol::Event;
19use crate::state::app_state::AppState;
20use actions::{IGMP_LEAVE_RECEIVED_EVENT, IGMP_QUERY_RECEIVED_EVENT, IGMP_REPORT_RECEIVED_EVENT};
21
22/// IGMP message types (RFC 2236)
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24#[repr(u8)]
25pub enum IgmpMessageType {
26 /// Membership Query (0x11)
27 MembershipQuery = 0x11,
28 /// IGMPv1 Membership Report (0x12)
29 V1MembershipReport = 0x12,
30 /// IGMPv2 Membership Report (0x16)
31 V2MembershipReport = 0x16,
32 /// Leave Group (0x17)
33 LeaveGroup = 0x17,
34 /// IGMPv3 Membership Report (0x22)
35 V3MembershipReport = 0x22,
36}
37
38impl IgmpMessageType {
39 fn from_u8(value: u8) -> Option<Self> {
40 match value {
41 0x11 => Some(Self::MembershipQuery),
42 0x12 => Some(Self::V1MembershipReport),
43 0x16 => Some(Self::V2MembershipReport),
44 0x17 => Some(Self::LeaveGroup),
45 0x22 => Some(Self::V3MembershipReport),
46 _ => None,
47 }
48 }
49
50 fn as_str(&self) -> &'static str {
51 match self {
52 Self::MembershipQuery => "Membership Query",
53 Self::V1MembershipReport => "IGMPv1 Membership Report",
54 Self::V2MembershipReport => "IGMPv2 Membership Report",
55 Self::LeaveGroup => "Leave Group",
56 Self::V3MembershipReport => "IGMPv3 Membership Report",
57 }
58 }
59}
60
61/// Smallest legal IPv4 header (IHL 5 words), in bytes.
62const MIN_IP_HEADER: usize = 20;
63/// Largest legal IPv4 header (IHL 15 words), in bytes.
64const MAX_IP_HEADER: usize = 60;
65/// IANA protocol number for IGMP.
66const IP_PROTO_IGMP: u8 = 2;
67/// `224.0.0.2`, the ALL-ROUTERS group a Leave Group is addressed to (RFC 2236 §2.9).
68const ALL_ROUTERS: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 2);
69/// `224.0.0.22`, the group every IGMPv3 Membership Report is addressed to (RFC 3376 §4.2.14).
70const ALL_IGMPV3_ROUTERS: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 22);
71
72/// Why a captured IP packet carries no IGMP message we can act on.
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74pub enum PayloadReject {
75 /// Fewer bytes than the smallest possible IPv4 header.
76 TooShort(usize),
77 /// The IHL field names a header length that is illegal, or longer than the packet.
78 HeaderLength { ihl: usize, len: usize },
79 /// The IPv4 protocol field is not 2.
80 NotIgmp(u8),
81}
82
83impl std::fmt::Display for PayloadReject {
84 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85 match self {
86 Self::TooShort(len) => write!(f, "packet too short for an IPv4 header ({len} bytes)"),
87 Self::HeaderLength { ihl, len } => {
88 write!(f, "malformed IPv4 header (IHL={ihl} bytes, packet={len})")
89 }
90 Self::NotIgmp(proto) => write!(f, "IP protocol {proto} is not IGMP"),
91 }
92 }
93}
94
95/// Strip the IPv4 header a raw socket delivers and return the IGMP message inside.
96///
97/// Every index below is reachable from the wire, so each one is bounds-checked before it is
98/// taken. In particular the IHL field is four bits, so a hostile packet can name a header of
99/// 0..=60 bytes: without the lower bound `&raw[ihl..]` would silently hand the IP header itself
100/// to the IGMP parser, and without the upper one it would be an out-of-range slice and a panic
101/// inside the capture task — where `tokio::spawn` swallows it and the server goes on reporting
102/// `Running`.
103pub fn igmp_payload(raw: &[u8]) -> std::result::Result<&[u8], PayloadReject> {
104 if raw.len() < MIN_IP_HEADER {
105 return Err(PayloadReject::TooShort(raw.len()));
106 }
107 let ihl = (raw[0] & 0x0F) as usize * 4;
108 if !(MIN_IP_HEADER..=MAX_IP_HEADER).contains(&ihl) || raw.len() < ihl {
109 return Err(PayloadReject::HeaderLength {
110 ihl,
111 len: raw.len(),
112 });
113 }
114 if raw[9] != IP_PROTO_IGMP {
115 return Err(PayloadReject::NotIgmp(raw[9]));
116 }
117 Ok(&raw[ihl..])
118}
119
120/// Where an IGMP message we are about to emit has to be addressed.
121///
122/// This is spec-fixed, not a choice: a Membership Report goes to the group it reports
123/// (RFC 2236 §9), a Leave Group to ALL-ROUTERS `224.0.0.2` (§2.9), and an IGMPv3 report to
124/// `224.0.0.22` (RFC 3376 §4.2.14). Sending any of them back to whoever queried us — which is
125/// what the fallback does, and what v1 and v3 reports used to get — puts a multicast control
126/// message on a unicast address, where no router is listening for it.
127pub fn response_destination(packet: &[u8], peer: SocketAddr) -> SocketAddr {
128 if packet.len() < 8 {
129 return peer;
130 }
131 let group = Ipv4Addr::new(packet[4], packet[5], packet[6], packet[7]);
132 match packet[0] {
133 0x12 | 0x16 if group.is_multicast() => SocketAddr::new(std::net::IpAddr::V4(group), 0),
134 0x17 => SocketAddr::new(std::net::IpAddr::V4(ALL_ROUTERS), 0),
135 0x22 => SocketAddr::new(std::net::IpAddr::V4(ALL_IGMPV3_ROUTERS), 0),
136 _ => peer,
137 }
138}
139
140/// Parsed IGMP message
141#[derive(Debug, Clone)]
142pub struct IgmpMessage {
143 pub msg_type: IgmpMessageType,
144 pub max_response_time: u8,
145 /// The checksum field as it arrived, bytes 2-3. See [`Self::checksum_valid`].
146 pub checksum: u16,
147 pub group_address: Ipv4Addr,
148 pub raw_data: Vec<u8>,
149}
150
151impl IgmpMessage {
152 /// Parse an IGMP message from raw bytes
153 pub fn parse(data: &[u8]) -> Result<Self> {
154 if data.len() < 8 {
155 return Err(anyhow::anyhow!(
156 "IGMP message too short: {} bytes",
157 data.len()
158 ));
159 }
160
161 let msg_type = IgmpMessageType::from_u8(data[0]).context("Unknown IGMP message type")?;
162
163 let max_response_time = data[1];
164
165 let checksum = u16::from_be_bytes([data[2], data[3]]);
166
167 let group_address = Ipv4Addr::new(data[4], data[5], data[6], data[7]);
168
169 Ok(Self {
170 msg_type,
171 max_response_time,
172 checksum,
173 group_address,
174 raw_data: data.to_vec(),
175 })
176 }
177
178 /// Whether the RFC 1071 checksum over the whole message is correct.
179 ///
180 /// Checked separately from [`Self::parse`] so the two failures stay distinguishable in the
181 /// log: a message we cannot parse is not IGMP, one that fails here *is* IGMP and arrived
182 /// corrupt. Acting on a corrupt message would spend an LLM round-trip — and possibly a
183 /// Membership Report, which is a positive claim — on bytes no conforming receiver would
184 /// have accepted.
185 pub fn checksum_valid(&self) -> bool {
186 crate::server::igmp::actions::igmp_checksum(&self.raw_data) == 0
187 }
188
189 /// Check if this is a general query (group address is 0.0.0.0)
190 pub fn is_general_query(&self) -> bool {
191 self.msg_type == IgmpMessageType::MembershipQuery && self.group_address.is_unspecified()
192 }
193
194 /// Get human-readable description
195 pub fn description(&self) -> String {
196 format!(
197 "{} for group {} (max_resp={})",
198 self.msg_type.as_str(),
199 self.group_address,
200 self.max_response_time
201 )
202 }
203}
204
205/// IGMP server state
206pub struct IgmpServerState {
207 /// Set of multicast groups we've joined
208 pub joined_groups: HashSet<Ipv4Addr>,
209}
210
211impl IgmpServerState {
212 fn new() -> Self {
213 Self {
214 joined_groups: HashSet::new(),
215 }
216 }
217}
218
219/// IGMP server that manages multicast group membership
220pub struct IgmpServer;
221
222impl IgmpServer {
223 /// Spawn IGMP server with action-based LLM handling using raw IP sockets
224 pub async fn spawn_with_llm_actions(
225 listen_addr: SocketAddr,
226 llm_client: OllamaClient,
227 app_state: Arc<AppState>,
228 status_tx: mpsc::UnboundedSender<String>,
229 server_id: crate::state::ServerId,
230 ) -> Result<SocketAddr> {
231 Log::new(Some(&status_tx))
232 .info("IGMP server starting with raw IP sockets (requires root privileges)");
233
234 // Create raw socket for IGMP (protocol 2).
235 // AF_INET/SOCK_RAW/IPPROTO_IGMP are POSIX constants available on Linux, macOS and the
236 // BSDs alike; raw sockets require root (or CAP_NET_RAW on Linux) on all of them.
237 //
238 // The fd MUST be validated *before* handing it to `Socket::from_raw_fd`: that function's
239 // safety contract requires an open, exclusively-owned descriptor, and wrapping -1 would
240 // also make the eventual `Drop` call `close(-1)`.
241 let raw_fd = unsafe { libc::socket(libc::AF_INET, libc::SOCK_RAW, libc::IPPROTO_IGMP) };
242 if raw_fd < 0 {
243 let os_err = std::io::Error::last_os_error();
244 let msg = format!(
245 "Failed to create raw IGMP socket on {}: {} \
246 (IGMP needs root/CAP_NET_RAW - re-run netget with sudo)",
247 listen_addr.ip(),
248 os_err
249 );
250 Log::new(Some(&status_tx)).error(&msg);
251 return Err(anyhow::anyhow!(msg));
252 }
253 // SAFETY: `raw_fd` was just returned by `socket(2)`, is >= 0, is open, and nothing else
254 // owns it, so transferring ownership to `Socket` is sound.
255 let socket = unsafe { Socket::from_raw_fd(raw_fd) };
256
257 // Bind to the interface address (0.0.0.0 to listen on all interfaces)
258 let bind_addr = std::net::SocketAddrV4::new(
259 match listen_addr.ip() {
260 std::net::IpAddr::V4(addr) => addr,
261 std::net::IpAddr::V6(_) => Ipv4Addr::UNSPECIFIED,
262 },
263 0, // Port is ignored for raw sockets
264 );
265
266 // Any of these can still fail (e.g. binding an address the host does not own). Report the
267 // failure on both channels and propagate so the server is marked Error, never Running.
268 let setup = (|| -> Result<()> {
269 socket
270 .set_reuse_address(true)
271 .context("set_reuse_address failed")?;
272 socket
273 .bind(&bind_addr.into())
274 .with_context(|| format!("bind to {} failed", bind_addr))?;
275 socket
276 .set_nonblocking(true)
277 .context("set_nonblocking failed")?;
278 Ok(())
279 })();
280 if let Err(e) = setup {
281 let msg = format!("IGMP raw socket setup failed: {:#}", e);
282 Log::new(Some(&status_tx)).error(&msg);
283 return Err(e);
284 }
285
286 // Get the local address (for display purposes)
287 let local_addr = SocketAddr::new(
288 std::net::IpAddr::V4(*bind_addr.ip()),
289 2, // IGMP protocol number
290 );
291
292 info!(
293 "IGMP server listening on {} (raw socket, protocol 2)",
294 local_addr
295 );
296
297 let protocol = Arc::new(IgmpProtocol::new());
298 let server_state = Arc::new(Mutex::new(IgmpServerState::new()));
299
300 // Convert socket2::Socket to std::net::UdpSocket for tokio
301 // Even though this is a raw socket, we can wrap it as UdpSocket
302 let std_socket: std::net::UdpSocket = socket.into();
303 let socket = Arc::new(
304 tokio::net::UdpSocket::from_std(std_socket)
305 .inspect_err(|e| {
306 Log::new(Some(&status_tx)).error(format!(
307 "IGMP failed to register raw socket with tokio: {}",
308 e
309 ));
310 })
311 .context("failed to register IGMP raw socket with the tokio reactor")?,
312 );
313
314 let task_registrar = app_state.clone();
315 let accept_handle = tokio::spawn(async move {
316 let mut buffer = vec![0u8; 65535];
317 let log = Log::new(Some(&status_tx));
318
319 loop {
320 match socket.recv_from(&mut buffer).await {
321 Ok((n, peer_addr)) => {
322 let raw_data = &buffer[..n];
323
324 // Raw sockets receive the whole IP packet; every bounds check that
325 // makes stripping the header safe lives in `igmp_payload`.
326 let data = match igmp_payload(raw_data) {
327 Ok(payload) => payload.to_vec(),
328 Err(reject) => {
329 debug!("IGMP discarded a capture from {}: {}", peer_addr, reject);
330 continue;
331 }
332 };
333 let connection_id =
334 ConnectionId::new(app_state.get_next_unified_id().await);
335
336 // Add connection to ServerInstance
337 use crate::state::server::{
338 ConnectionState as ServerConnectionState, ConnectionStatus,
339 ProtocolConnectionInfo,
340 };
341 let now = crate::utils::clock::Instant::now();
342 let conn_state = ServerConnectionState {
343 id: connection_id,
344 remote_addr: peer_addr,
345 local_addr,
346 bytes_sent: 0,
347 bytes_received: data.len() as u64,
348 packets_sent: 0,
349 packets_received: 1,
350 last_activity: now,
351 status: ConnectionStatus::Active,
352 status_changed_at: now,
353 protocol_info: ProtocolConnectionInfo::empty(),
354 };
355 app_state
356 .add_connection_to_server(server_id, conn_state)
357 .await;
358 let _ = status_tx.send("__UPDATE_UI__".to_string());
359
360 // Parse IGMP message
361 let igmp_msg = match IgmpMessage::parse(&data) {
362 Ok(msg) => msg,
363 Err(e) => {
364 log.debug(format!(
365 "IGMP received non-IGMP packet ({} bytes): {}",
366 data.len(),
367 e
368 ));
369 continue;
370 }
371 };
372
373 // A message that fails its own checksum is corrupt, not merely
374 // unfamiliar. Answering one would spend an LLM round-trip - and
375 // possibly a Membership Report, which is a positive membership claim -
376 // on bytes a conforming receiver would have dropped.
377 if !igmp_msg.checksum_valid() {
378 log.warn(format!(
379 "IGMP {} from {} discarded: bad checksum 0x{:04x} decision=bad_checksum",
380 igmp_msg.msg_type.as_str(),
381 peer_addr,
382 igmp_msg.checksum
383 ));
384 continue;
385 }
386
387 // Summary + full payload FileOnly: the igmp_* event templates
388 // render the equivalent line to the TUI.
389 log.debug(format!(
390 "IGMP received from {}: {}",
391 peer_addr,
392 igmp_msg.description()
393 ));
394 log.trace(format!("IGMP data (hex): {}", hex::encode(&data)));
395
396 let llm_clone = llm_client.clone();
397 let state_clone = app_state.clone();
398 let status_clone = status_tx.clone();
399 let socket_clone = socket.clone();
400 let protocol_clone = protocol.clone();
401 let server_state_clone = server_state.clone();
402
403 // Tracked, not detached: stop_server must abort this task too.
404 let task_owner = app_state.clone();
405 task_owner.spawn_server_task(server_id, async move {
406 let log = Log::new(Some(&status_clone));
407 // Determine event type and build event data
408 let (event, _event_type_ref) = match igmp_msg.msg_type {
409 IgmpMessageType::MembershipQuery => {
410 let query_type = if igmp_msg.is_general_query() {
411 "General"
412 } else {
413 "Group-Specific"
414 };
415 (
416 Event::new(
417 &IGMP_QUERY_RECEIVED_EVENT,
418 serde_json::json!({
419 "query_type": query_type,
420 "group_address": igmp_msg.group_address.to_string(),
421 "max_response_time": igmp_msg.max_response_time
422 }),
423 ),
424 &IGMP_QUERY_RECEIVED_EVENT,
425 )
426 }
427 IgmpMessageType::V1MembershipReport
428 | IgmpMessageType::V2MembershipReport
429 | IgmpMessageType::V3MembershipReport => (
430 Event::new(
431 &IGMP_REPORT_RECEIVED_EVENT,
432 serde_json::json!({
433 "group_address": igmp_msg.group_address.to_string()
434 }),
435 ),
436 &IGMP_REPORT_RECEIVED_EVENT,
437 ),
438 IgmpMessageType::LeaveGroup => (
439 Event::new(
440 &IGMP_LEAVE_RECEIVED_EVENT,
441 serde_json::json!({
442 "group_address": igmp_msg.group_address.to_string()
443 }),
444 ),
445 &IGMP_LEAVE_RECEIVED_EVENT,
446 ),
447 };
448
449 // What to do with an IGMP message is not wire-determined. Which
450 // groups to report membership in (answering a query) is a membership
451 // policy choice, not derivable from the query (a general query names
452 // group 0.0.0.0); and an observed report or leave has no spec-mandated
453 // packet response at all (report suppression / passive observation).
454 // So with no operator policy — no server instruction and no per-event
455 // handler — the spec-safe default is to advertise no memberships and
456 // stay silent, WITHOUT burning an LLM round-trip per captured packet.
457 // The model is consulted only when the operator supplies the
458 // membership policy it should apply.
459 if !operator_wants_dynamic(
460 &state_clone,
461 server_id,
462 &event.event_type.id,
463 )
464 .await
465 {
466 log.info(format!(
467 "IGMP {} from {} ignored: no membership policy configured (static default, no LLM) decision=static_no_policy",
468 igmp_msg.msg_type.as_str(),
469 peer_addr
470 ));
471 return;
472 }
473
474 log.debug(format!(
475 "IGMP calling LLM for {} from {}",
476 igmp_msg.msg_type.as_str(),
477 peer_addr
478 ));
479
480 match call_llm(
481 &llm_clone,
482 &state_clone,
483 server_id,
484 None,
485 &event,
486 protocol_clone.as_ref(),
487 )
488 .await
489 {
490 Ok(execution_result) => {
491 for message in &execution_result.messages {
492 log.info(message);
493 }
494
495 // Three outcomes have to stay apart in the log even though
496 // all three are silent on the wire (see the Err arm below
497 // for why silence is the only correct IGMP answer):
498 // the model explicitly declined (`ignore_message` ->
499 // NoAction), the model answered nothing at all, and the
500 // backend errored.
501 let model_declined = execution_result
502 .protocol_results
503 .iter()
504 .any(|r| {
505 matches!(
506 r,
507 crate::llm::actions::protocol_trait::ActionResult::NoAction
508 )
509 });
510 let has_output = execution_result
511 .protocol_results
512 .iter()
513 .any(|r| !r.get_all_output().is_empty());
514 if !has_output {
515 if model_declined {
516 log.info(format!(
517 "IGMP {} from {}: model chose to send nothing decision=model_ignore",
518 igmp_msg.msg_type.as_str(),
519 peer_addr
520 ));
521 } else {
522 log.warn(format!(
523 "IGMP {} from {}: model produced no usable action, staying silent decision=model_no_answer",
524 igmp_msg.msg_type.as_str(),
525 peer_addr
526 ));
527 }
528 }
529
530 log.debug(format!(
531 "IGMP got {} protocol results",
532 execution_result.protocol_results.len()
533 ));
534
535 // Process protocol results
536 for protocol_result in &execution_result.protocol_results {
537 if let Some(output_data) =
538 protocol_result.get_all_output().first()
539 {
540 // Spec-fixed, and a pure function so it is testable
541 // without a raw socket.
542 let dest_addr =
543 response_destination(output_data, peer_addr);
544
545 // Send the IGMP packet via raw socket
546 if let Err(e) =
547 socket_clone.send_to(output_data, dest_addr).await
548 {
549 log.error(format!(
550 "Failed to send IGMP response: {}",
551 e
552 ));
553 } else {
554 // Summary + full payload FileOnly: the
555 // send action template already reports the
556 // send to the TUI.
557 log.debug(format!(
558 "IGMP sent {} bytes to {}",
559 output_data.len(),
560 dest_addr
561 ));
562 log.trace(format!(
563 "IGMP sent (hex): {}",
564 hex::encode(output_data)
565 ));
566 }
567 }
568 }
569
570 // Process async custom actions (join_group/leave_group)
571 use crate::llm::actions::protocol_trait::ActionResult;
572 for protocol_result in &execution_result.protocol_results {
573 if let ActionResult::Custom { name, data } = protocol_result
574 {
575 match name.as_str() {
576 "igmp_join_group" => {
577 if let Some(group_str) = data
578 .get("group_address")
579 .and_then(|v| v.as_str())
580 {
581 if let Ok(group_addr) =
582 group_str.parse::<Ipv4Addr>()
583 {
584 // `joined_groups` is what makes a
585 // repeated join idempotent: without
586 // this check the second one fails
587 // with EADDRINUSE and is logged as
588 // an error, when in fact the
589 // membership the model asked for
590 // already exists.
591 if server_state_clone
592 .lock()
593 .await
594 .joined_groups
595 .contains(&group_addr)
596 {
597 log.debug(format!("IGMP already a member of {}; join ignored", group_addr));
598 continue;
599 }
600 // Join the multicast group on all interfaces (0.0.0.0)
601 match socket_clone.join_multicast_v4(
602 group_addr,
603 Ipv4Addr::UNSPECIFIED,
604 ) {
605 Ok(_) => {
606 let mut state =
607 server_state_clone
608 .lock()
609 .await;
610 state
611 .joined_groups
612 .insert(group_addr);
613 log.info(format!("IGMP joined multicast group {}", group_addr));
614 }
615 Err(e) => {
616 log.error(format!("Failed to join multicast group {}: {}", group_addr, e));
617 }
618 }
619 }
620 }
621 }
622 "igmp_leave_group" => {
623 if let Some(group_str) = data
624 .get("group_address")
625 .and_then(|v| v.as_str())
626 {
627 if let Ok(group_addr) =
628 group_str.parse::<Ipv4Addr>()
629 {
630 // Leave the multicast group on all interfaces (0.0.0.0)
631 match socket_clone.leave_multicast_v4(
632 group_addr,
633 Ipv4Addr::UNSPECIFIED,
634 ) {
635 Ok(_) => {
636 let mut state =
637 server_state_clone
638 .lock()
639 .await;
640 state
641 .joined_groups
642 .remove(&group_addr);
643 log.info(format!("IGMP left multicast group {}", group_addr));
644 }
645 Err(e) => {
646 log.error(format!("Failed to leave multicast group {}: {}", group_addr, e));
647 }
648 }
649 }
650 }
651 }
652 _ => {}
653 }
654 }
655 }
656 }
657 Err(e) => {
658 // IGMP has no error message and no way to answer without
659 // asserting a membership. A Membership Report is a positive
660 // claim ("this host is in group G") that makes the querier
661 // forward that group's traffic to this segment; emitting one
662 // we cannot substantiate is worse than not answering, and
663 // reports/leaves have no spec response at all. So the wire
664 // stays silent and the failure goes to the operator only —
665 // classified, so an overload reads differently from a hard
666 // backend fault, and never rendered onto the socket.
667 let category =
668 crate::utils::wire_failure::WireFailure::classify(&e);
669 let category_tag = if category.is_overloaded() {
670 "overloaded"
671 } else {
672 "unavailable"
673 };
674 log.error(format!(
675 "IGMP {} from {} unanswered: LLM call failed: {} decision=fail_closed_silent category={}",
676 igmp_msg.msg_type.as_str(),
677 peer_addr,
678 e,
679 category_tag
680 ));
681 }
682 }
683 }).await;
684 }
685 Err(e) => {
686 // A raw socket that keeps erroring would otherwise spin this loop at
687 // 100% CPU forever, so stop the capture loop and say so loudly.
688 log.error(format!("IGMP receive error, stopping capture loop: {}", e));
689 break;
690 }
691 }
692 }
693 });
694
695 task_registrar
696 .register_server_task(server_id, accept_handle)
697 .await;
698
699 Ok(local_addr)
700 }
701}
702
703/// Returns true if the operator opted into dynamic (LLM- or handler-driven) responses for this
704/// server: either a non-empty server instruction was given, or an event handler is configured
705/// for `event_id`. When false the protocol applies its static default and never consults the
706/// model — for IGMP that default is to advertise no memberships and stay silent, because with no
707/// configured membership policy there is nothing to report and an observed report/leave needs no
708/// reply.
709async fn operator_wants_dynamic(
710 state: &AppState,
711 server_id: crate::state::ServerId,
712 event_id: &str,
713) -> bool {
714 state
715 .with_server_mut(server_id, |server| {
716 let has_instruction = !server.instruction.trim().is_empty();
717 let has_handler = server
718 .event_handler_config
719 .as_ref()
720 .map(|c| c.find_handler(event_id).is_some())
721 .unwrap_or(false);
722 has_instruction || has_handler
723 })
724 .await
725 .unwrap_or(false)
726}