fix: keep re-querying unresolved instances while browsing (#493)

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;
         }
 

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::<DnsSrv>() {
                     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::<DnsPointer>() {
+                    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<Vec<u8>>| {
+            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"
+        );
+    }
 }