diff --git a/src/dns_cache.rs b/src/dns_cache.rs index febdf357..395ada05 100644 --- a/src/dns_cache.rs +++ b/src/dns_cache.rs @@ -259,8 +259,15 @@ impl DnsCache { .map_or(true, |records| records.is_empty()); // No existing records for this name and type, and not for us. + // + // `is_for_us` is heuristic per message, not per record. We never grow the cache for + // records nobody asked for, but always keep what's already there coherent. if empty_records && !is_for_us { - trace!("add_or_update: not for us: {}", incoming.get_name()); + trace!( + "add_or_update: dropping new record not for us: {} (type {:?})", + incoming.get_name(), + incoming.get_type() + ); return None; } diff --git a/src/service_daemon.rs b/src/service_daemon.rs index 31b553e8..f04ea640 100644 --- a/src/service_daemon.rs +++ b/src/service_daemon.rs @@ -83,7 +83,14 @@ const GROUP_ADDR_V4: Ipv4Addr = Ipv4Addr::new(224, 0, 0, 251); const GROUP_ADDR_V6: Ipv6Addr = Ipv6Addr::new(0xff02, 0, 0, 0, 0, 0, 0, 0xfb); const LOOPBACK_V4: Ipv4Addr = Ipv4Addr::new(127, 0, 0, 1); -const RESOLVE_WAIT_IN_MILLIS: u64 = 500; +/// The first fast-path follow-up resolve query fires this soon after an +/// instance is found via PTR. Subsequent tries back off exponentially +/// (200ms, 400ms, 800ms, 1600ms), see `exec_command_resolve`. +const RESOLVE_RETRY_BASE_MILLIS: u64 = 200; + +/// Max number of fast-path resolve tries before the browse retransmission +/// cycle takes over re-querying (see `query_unresolved_instances`). +const RESOLVE_MAX_TRY: u16 = 4; /// RFC 6762 §8.3: the two unsolicited announcements are sent "one second apart". /// We schedule the second one strictly wider than the §6 multicast rate-limit @@ -2722,6 +2729,17 @@ impl Zeroconf { match DnsIncoming::new(buf, my_intf.into()) { Ok(msg) => { + debug!( + "handle_read: {} bytes from {} on if_index {}: {} ({} questions {} answers {} authorities {} additionals)", + sz, + pktinfo.addr_src, + pkt_if_index, + if msg.is_query() { "query" } else { "response" }, + msg.questions().len(), + msg.answers().len(), + msg.authorities().len(), + msg.additionals().len(), + ); if msg.is_query() { let querier_addr = pktinfo.addr_src; self.handle_query(msg, pkt_if_index, querier_addr); @@ -2748,19 +2766,56 @@ impl Zeroconf { for record in records { if let Some(srv) = record.record.any().downcast_ref::() { if self.cache.get_addr(srv.host()).is_none() { + debug!( + "query_unresolved: SRV record found for instance: {}, host: {}, sending address query", + instance, + srv.host() + ); self.send_query_vec(&[(srv.host(), RRType::A), (srv.host(), RRType::AAAA)]); return true; } } } } else { - self.send_query(instance, RRType::ANY); + debug!( + "query_unresolved: SRV record not found for instance: {}, sending SRV+TXT query", + instance + ); + // Query SRV and TXT explicitly rather than RRType::ANY. RFC 6762 §6.5 makes + // ANY legal, but SRV+TXT is what Bonjour's DNSServiceResolve and Avahi send, + // so it is the path every responder is actually exercised on; it also lets + // known-answer suppression apply (`get_known_answers` has no ANY case). + self.send_query_vec(&[(instance, RRType::SRV), (instance, RRType::TXT)]); return true; } false } + /// Re-issues follow-up queries (SRV / address) for every instance of + /// `ty_domain` that has been found via PTR but is not yet resolved. + /// + /// This is meant to be driven by the browse retransmission cycle so a + /// pending instance keeps being queried for as long as the browse is + /// active. + fn query_unresolved_instances(&mut self, ty_domain: &str) { + let now = current_time_millis(); + let mut instances = Vec::new(); + if let Some(records) = self.cache.get_ptr(ty_domain) { + for record in records.iter().filter(|r| !r.record.expires_soon(now)) { + if let Some(ptr) = record.record.any().downcast_ref::() { + instances.push(ptr.alias().to_string()); + } + } + } + + for instance in instances { + if !self.resolved.contains(&instance) { + self.query_unresolved(&instance); + } + } + } + /// Checks if `ty_domain` has records in the cache. If yes, sends the /// cached records via `sender`. fn query_cache_for_service( @@ -2783,7 +2838,12 @@ impl Zeroconf { new_event = Some(ServiceEvent::ServiceResolved(Box::new(resolved_service))); } else { - debug!("Resolved service is not valid: {}", ptr.alias()); + debug!( + "query_cache_for_service: not valid: {} (host_empty={}, addrs_empty={})", + ptr.alias(), + resolved_service.get_hostname().is_empty(), + resolved_service.get_addresses().is_empty(), + ); } } Err(err) => { @@ -2844,7 +2904,7 @@ impl Zeroconf { fn add_pending_resolve(&mut self, instance: String) { if !self.pending_resolves.contains(&instance) { - let next_time = current_time_millis() + RESOLVE_WAIT_IN_MILLIS; + let next_time = current_time_millis() + RESOLVE_RETRY_BASE_MILLIS; self.add_retransmission(next_time, Command::Resolve(instance.clone(), 1)); self.pending_resolves.insert(instance); } @@ -3329,7 +3389,11 @@ impl Zeroconf { let event = ServiceEvent::ServiceResolved(Box::new(resolved_service)); call_service_listener(&self.service_queriers, ty_domain, event); } else { - debug!("Resolved service is not valid: {instance}"); + debug!( + "resolve_updated_instances: not valid: {instance} (host_empty={}, addrs_empty={})", + resolved_service.get_hostname().is_empty(), + resolved_service.get_addresses().is_empty(), + ); if self.resolved.remove(dns_ptr.alias()) { removed_instances .entry(ty_domain.to_string()) @@ -3895,6 +3959,9 @@ impl Zeroconf { } self.send_query(&ty, RRType::PTR); + + self.query_unresolved_instances(&ty); + self.increase_counter(Counter::Browse, 1); let next_time = now + (next_delay * 1000) as u64; @@ -3962,12 +4029,17 @@ impl Zeroconf { fn exec_command_resolve(&mut self, instance: String, try_count: u16) { let pending_query = self.query_unresolved(&instance); - let max_try = 3; - if pending_query && try_count < max_try { + if pending_query && try_count < RESOLVE_MAX_TRY { // Note that if the current try already succeeds, the next retransmission // will be no-op as the cache has been updated. - let next_time = current_time_millis() + RESOLVE_WAIT_IN_MILLIS; + // + // Back off exponentially + let next_delay = RESOLVE_RETRY_BASE_MILLIS << try_count; + let next_time = current_time_millis() + next_delay; self.add_retransmission(next_time, Command::Resolve(instance, try_count + 1)); + } else { + // This fast-path retry chain is ending. + self.pending_resolves.remove(&instance); } } @@ -5230,13 +5302,14 @@ mod tests { valid_instance_name, valid_ip_on_intf, DaemonEvent, HostnameResolutionEvent, IfKind, MaxPacketSizeSelection, MyIntf, SendConfig, ServiceDaemon, ServiceEvent, ServiceInfo, GROUP_ADDR_V4, INITIAL_QUERY_DELAY_MAX_MILLIS, INITIAL_QUERY_DELAY_MIN_MILLIS, - MAX_PKT_ABSOLUTE_IPV6, MAX_PKT_DEFAULT, MDNS_PORT, MIN_MAX_PACKET_SIZE, + MAX_PKT_ABSOLUTE_IPV6, MAX_PKT_DEFAULT, MDNS_PORT, MIN_MAX_PACKET_SIZE, RESOLVE_MAX_TRY, SHARED_RESPONSE_DELAY_MAX_MILLIS, SHARED_RESPONSE_DELAY_MIN_MILLIS, }; use crate::{ dns_parser::{ - DnsEntryExt, DnsIncoming, DnsOutgoing, DnsPointer, InterfaceId, RRType, ScopedIp, - CLASS_IN, FLAGS_AA, FLAGS_QR_QUERY, FLAGS_QR_RESPONSE, LEGACY_UNICAST_MAX_TTL, + DnsAddress, DnsEntryExt, DnsIncoming, DnsOutgoing, DnsPointer, DnsSrv, InterfaceId, + RRType, ScopedIp, CLASS_IN, FLAGS_AA, FLAGS_QR_QUERY, FLAGS_QR_RESPONSE, + LEGACY_UNICAST_MAX_TTL, }, service_daemon::{add_answer_of_service, check_hostname}, }; @@ -6596,4 +6669,193 @@ mod tests { server_default.shutdown().unwrap(); client_default.shutdown().unwrap(); } + + /// Regression test for such issue: an instance that is found (via PTR) but + /// whose SRV/address answers are lost during the initial resolve attempts + /// must not be stranded. As long as the browse is active, the daemon must + /// keep re-querying for it. + #[test] + fn test_unresolved_instance_not_stranded() { + use socket2::{Domain, Protocol, Socket, Type}; + use std::net::SocketAddrV4; + + // Pick the first IPv4 interface, like the other multicast tests. + let intf_ip = match my_ip_interfaces(false) + .into_iter() + .find_map(|intf| match intf.ip() { + IpAddr::V4(ip) if !ip.is_loopback() => Some(ip), + _ => None, + }) { + Some(ip) => ip, + None => { + println!("No non-loopback IPv4 interface available; skipping test."); + return; + } + }; + + let unique = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap() + .as_micros(); + let ty_domain = format!("_strandtest{unique}._udp.local."); + let instance = format!("inst.{ty_domain}"); + // The SRV target is a distinct hostname, which is the case that needs a + // separate address query. + let host = format!("strandhost{unique}.local."); + let port = 1234u16; + let ttl = 4500u32; + + let if_id = InterfaceId { + name: "test".to_string(), + index: 0, + }; + + // Build our hand-rolled responder socket: bound to the mDNS port, + // joined to the group, with loopback on so it exchanges packets with + // the in-process daemon. + let sock = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)) + .expect("create responder socket"); + sock.set_reuse_address(true).expect("set reuse_address"); + #[cfg(unix)] + let _ = sock.set_reuse_port(true); + sock.bind(&SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, MDNS_PORT).into()) + .expect("bind responder socket"); + sock.join_multicast_v4(&GROUP_ADDR_V4, &intf_ip) + .expect("join multicast group"); + sock.set_multicast_if_v4(&intf_ip) + .expect("set multicast_if"); + sock.set_multicast_loop_v4(true).expect("enable loopback"); + sock.set_read_timeout(Some(Duration::from_millis(100))) + .expect("set read timeout"); + let responder: UdpSocket = sock.into(); + + // A response carrying PTR + SRV, but deliberately no address record. + let announce_packets = || { + let mut out = DnsOutgoing::new(FLAGS_QR_RESPONSE | FLAGS_AA); + out.add_answer_at_time( + DnsPointer::new(&ty_domain, RRType::PTR, CLASS_IN, ttl, instance.clone()), + 0, + ); + out.add_answer_at_time( + DnsSrv::new(&instance, CLASS_IN, ttl, 0, 0, port, host.clone()), + 0, + ); + out.to_data_on_wire(MAX_PKT_DEFAULT, true) + }; + + // A response carrying just the withheld address record. + let addr_packets = || { + let mut out = DnsOutgoing::new(FLAGS_QR_RESPONSE | FLAGS_AA); + out.add_answer_at_time( + DnsAddress::new( + &host, + RRType::A, + CLASS_IN, + ttl, + IpAddr::V4(intf_ip), + if_id.clone(), + ), + 0, + ); + out.to_data_on_wire(MAX_PKT_DEFAULT, true) + }; + + let send_all = |packets: Vec>| { + for packet in packets { + let _ = responder.send_to(&packet, (GROUP_ADDR_V4, MDNS_PORT)); + } + }; + + // Start the browse, then announce (unsolicited) to seed the cache. + let daemon = ServiceDaemon::new().expect("create daemon"); + let browse_rx = daemon.browse(&ty_domain).expect("start browse"); + send_all(announce_packets()); + + // The number of address-query packets we must see before we start + // answering. We withhold past the fast-path budget (`RESOLVE_MAX_TRY` + // tries in `exec_command_resolve`) so that resolution can only come + // from the browse retransmission cycle. The buggy code went silent + // once the fast path was exhausted and never resolved the instance. + let answer_after = RESOLVE_MAX_TRY as i32; + let mut addr_query_count = 0; + let mut resolved = false; + let mut buf = [0u8; 2048]; + let deadline = Instant::now() + Duration::from_secs(20); + + while Instant::now() < deadline { + // Drive the responder: react to whatever queries have arrived. + while let Ok((len, from)) = responder.recv_from(&mut buf) { + // The daemon sends each query on every interface (loopback + // included) and on Linux this socket receives all copies. Count + // only the copy sent on the interface we joined, so one daemon + // query counts exactly once. + if from.ip() != IpAddr::V4(intf_ip) { + continue; + } + let Ok(msg) = DnsIncoming::new(buf[..len].to_vec(), if_id.clone()) else { + continue; + }; + if msg.is_response() { + continue; + } + + let mut saw_addr_query = false; + let mut saw_service_query = false; + for q in msg.questions() { + let qname = q.entry_name(); + if qname.eq_ignore_ascii_case(&host) + && matches!(q.entry_type(), RRType::A | RRType::AAAA) + { + saw_addr_query = true; + } else if qname.eq_ignore_ascii_case(&ty_domain) + || qname.eq_ignore_ascii_case(&instance) + { + saw_service_query = true; + } + } + + // Keep the PTR/SRV fresh if the daemon asks for them. + if saw_service_query { + send_all(announce_packets()); + } + + // Count address queries as one per packet (the daemon batches + // A and AAAA into a single query). Only answer once the buggy + // give-up window is behind us. + if saw_addr_query { + addr_query_count += 1; + if addr_query_count > answer_after { + send_all(addr_packets()); + } + } + } + + // Has the daemon resolved the instance yet? + while let Ok(event) = browse_rx.try_recv() { + if let ServiceEvent::ServiceResolved(info) = event { + if info.get_fullname().eq_ignore_ascii_case(&instance) { + resolved = true; + } + } + } + + if resolved { + break; + } + } + + daemon.shutdown().unwrap(); + + assert!( + addr_query_count > answer_after, + "daemon stopped querying for the address after {} tries; \ + an unresolved instance must keep being queried while the browse is active", + addr_query_count + ); + assert!( + resolved, + "instance was found but never resolved even though its address was \ + eventually answered" + ); + } }