weida_runtime/
resolve.rs

1//! Name resolution, and the seam that lets an application replace it.
2//!
3//! A name becomes a set of addresses, and **how** is not this library's
4//! decision to fix. The default is the system resolver, which answers A and
5//! AAAA and therefore one port for the whole set; that is exactly right for a
6//! Kubernetes headless Service, where every Pod listens on the same container
7//! port, and exactly wrong for a cloud load balancer, where one IP forwards
8//! several ports to several backends. The second case is what DNS SRV
9//! expresses — one host, one record per port — and what no A/AAAA answer can.
10//!
11//! So the shape is the one this workspace already uses for the decisions that
12//! belong to the application: `Trust` decides what a dial believes,
13//! `weida-amqp` takes the caller's `rustls::ClientConfig`, and a
14//! [`Resolver`] decides what a name means. Shipping only the system resolver
15//! and calling SRV unsupported would put a DNS stack in everybody's dependency
16//! graph to serve one deployment shape; refusing SRV outright would make that
17//! deployment shape unreachable. A seam does neither.
18
19use std::fmt;
20use std::future::Future;
21use std::net::{IpAddr, SocketAddr};
22use std::pin::Pin;
23use std::sync::Arc;
24
25use weida_core::{DEFAULT_PORT, Error};
26
27use crate::exec::Exec;
28
29/// The future a [`Resolver`] returns.
30///
31/// Boxed, because a resolver is held as a trait object in configuration and
32/// `async fn` in a trait is not dyn-compatible. One allocation per dial, on a
33/// path that is already doing a DNS round trip.
34pub type Resolved<'a> = Pin<Box<dyn Future<Output = Result<Vec<SocketAddr>, Error>> + Send + 'a>>;
35
36/// What a name means.
37///
38/// Implement this to answer a `weida://` authority with whatever the
39/// deployment actually uses — DNS SRV, a service registry, the cloud
40/// provider's API, or a table in a configuration file. The default is
41/// [`SystemResolver`].
42///
43/// # The contract
44///
45/// - **Order matters.** The addresses are dialled in the order returned, so a
46///   resolver that knows a preference expresses it by sorting. The caller
47///   tries the next one when a dial fails.
48/// - **`port` is the port the URL wrote, or `None` when it wrote none.**
49///   `None` means the authority names a *set*
50///   ([decisions/0020](../../../docs/decisions/0020-cluster-and-discovery.md)
51///   §4.2), and it is the resolver that decides which ports that set listens
52///   on: the system resolver uses [`DEFAULT_PORT`], an SRV resolver uses the
53///   ports the records carry.
54/// - **`max_addresses` is a cap, not a hint.** A resolver answer is remote
55///   input, so returning more than asked for is a bug in the resolver
56///   (`docs/INVARIANTS.md`); the caller does not re-truncate.
57/// - **An empty answer is an error**, not an empty set: a caller with no
58///   address has nothing to dial and deserves the reason.
59///
60/// # The `Exec`
61///
62/// Resolution runs on the runtime the library was given, not on an ambient
63/// one: `tokio::net::lookup_host` needs a Tokio context, and a library that
64/// demanded its caller be inside one would be the mistake `Exec` exists to
65/// prevent. A resolver that spawns work does so through the handle it is
66/// passed.
67pub trait Resolver: fmt::Debug + Send + Sync + 'static {
68    /// Resolves `name` into the addresses to dial, in order.
69    fn resolve<'a>(
70        &'a self,
71        exec: &'a Exec,
72        name: &'a str,
73        port: Option<u16>,
74        max_addresses: usize,
75    ) -> Resolved<'a>;
76}
77
78/// The system resolver: an IP literal in place, everything else through
79/// `getaddrinfo`.
80///
81/// What it answers is A and AAAA records, so the whole set shares one port —
82/// the one the URL wrote, or [`DEFAULT_PORT`] when it wrote none. That covers
83/// a literal, a plain name and a Kubernetes headless Service, whose A/AAAA
84/// answer *is* the set of Pod addresses. It cannot express a port-forwarding
85/// load balancer; [`Resolver`] is how that deployment brings its own answer.
86#[derive(Clone, Copy, Debug, Default)]
87pub struct SystemResolver;
88
89impl Resolver for SystemResolver {
90    fn resolve<'a>(
91        &'a self,
92        exec: &'a Exec,
93        name: &'a str,
94        port: Option<u16>,
95        max_addresses: usize,
96    ) -> Resolved<'a> {
97        Box::pin(async move {
98            let port = port.unwrap_or(DEFAULT_PORT);
99            if let Ok(ip) = name.parse::<IpAddr>() {
100                return Ok(vec![SocketAddr::new(ip, port)]);
101            }
102            let query = (name.to_owned(), port);
103            let looked_up = exec
104                .spawn(async move {
105                    tokio::net::lookup_host(query)
106                        .await
107                        .map(|addrs| addrs.collect::<Vec<SocketAddr>>())
108                })
109                .await
110                .map_err(|e| Error::Runtime(format!("name resolution task failed: {e}")))?;
111            let addrs: Vec<SocketAddr> = looked_up
112                .map_err(|e| Error::InvalidAddress(format!("cannot resolve {name}:{port}: {e}")))?
113                .into_iter()
114                .take(max_addresses)
115                .collect();
116            if addrs.is_empty() {
117                return Err(Error::InvalidAddress(format!(
118                    "{name}:{port} resolved to no addresses"
119                )));
120            }
121            Ok(addrs)
122        })
123    }
124}
125
126/// The resolver a configuration holds.
127///
128/// An `Arc` because a runtime's configuration is cloned per endpoint and a
129/// resolver may hold a cache, a connection to a registry or nothing at all.
130#[derive(Clone, Debug)]
131pub struct SharedResolver(Arc<dyn Resolver>);
132
133impl SharedResolver {
134    /// Wraps a resolver.
135    pub fn new(resolver: impl Resolver) -> SharedResolver {
136        SharedResolver(Arc::new(resolver))
137    }
138
139    /// Resolves through the wrapped resolver.
140    pub fn resolve<'a>(
141        &'a self,
142        exec: &'a Exec,
143        name: &'a str,
144        port: Option<u16>,
145        max_addresses: usize,
146    ) -> Resolved<'a> {
147        self.0.resolve(exec, name, port, max_addresses)
148    }
149}
150
151impl Default for SharedResolver {
152    fn default() -> SharedResolver {
153        SharedResolver::new(SystemResolver)
154    }
155}
156
157impl PartialEq for SharedResolver {
158    /// Two shared resolvers are equal when they are the same resolver.
159    ///
160    /// Pointer identity, because a resolver is an implementation and not a
161    /// value: the connection pool keys on configuration, and two different
162    /// resolvers must not be treated as one even if they happen to answer the
163    /// same way today.
164    fn eq(&self, other: &SharedResolver) -> bool {
165        Arc::ptr_eq(&self.0, &other.0)
166    }
167}
168
169impl Eq for SharedResolver {}
170
171#[cfg(test)]
172mod tests {
173    use super::*;
174
175    /// A resolver that answers from a table, which is what a test needs and
176    /// also what a static deployment would write.
177    #[derive(Debug)]
178    struct Table(Vec<SocketAddr>);
179
180    impl Resolver for Table {
181        fn resolve<'a>(
182            &'a self,
183            _exec: &'a Exec,
184            _name: &'a str,
185            _port: Option<u16>,
186            max_addresses: usize,
187        ) -> Resolved<'a> {
188            Box::pin(async move { Ok(self.0.iter().copied().take(max_addresses).collect()) })
189        }
190    }
191
192    #[tokio::test]
193    async fn a_literal_needs_no_resolver_and_takes_the_written_port() {
194        let exec = Exec::current().expect("ambient runtime");
195        let addrs = SystemResolver
196            .resolve(&exec, "127.0.0.1", Some(9000), 8)
197            .await
198            .expect("literal");
199        assert_eq!(addrs, vec!["127.0.0.1:9000".parse().expect("addr")]);
200    }
201
202    #[tokio::test]
203    async fn a_literal_without_a_written_port_takes_the_default() {
204        let exec = Exec::current().expect("ambient runtime");
205        let addrs = SystemResolver
206            .resolve(&exec, "127.0.0.1", None, 8)
207            .await
208            .expect("literal");
209        assert_eq!(addrs[0].port(), DEFAULT_PORT);
210    }
211
212    #[tokio::test]
213    async fn an_unresolvable_name_is_an_error_rather_than_an_empty_set() {
214        let exec = Exec::current().expect("ambient runtime");
215        let err = SystemResolver
216            .resolve(&exec, "no-such-host.invalid", Some(1), 8)
217            .await
218            .expect_err("`.invalid` never resolves");
219        assert!(
220            err.to_string().contains("no-such-host.invalid"),
221            "the error names what could not be resolved: {err}"
222        );
223    }
224
225    /// The load-balancer shape: one address, several ports, which no A/AAAA
226    /// answer can express and a replaced resolver can.
227    #[tokio::test]
228    async fn a_replaced_resolver_may_answer_several_ports_on_one_address() {
229        let exec = Exec::current().expect("ambient runtime");
230        let table = SharedResolver::new(Table(vec![
231            "203.0.113.7:7443".parse().expect("addr"),
232            "203.0.113.7:7444".parse().expect("addr"),
233            "203.0.113.7:7445".parse().expect("addr"),
234        ]));
235        let addrs = table
236            .resolve(&exec, "lb.example", None, 8)
237            .await
238            .expect("table");
239        assert_eq!(addrs.len(), 3);
240        assert!(addrs.iter().all(|a| a.ip().to_string() == "203.0.113.7"));
241        assert_eq!(
242            addrs.iter().map(|a| a.port()).collect::<Vec<_>>(),
243            vec![7443, 7444, 7445],
244            "the order is the resolver's and is preserved"
245        );
246    }
247
248    #[tokio::test]
249    async fn the_cap_is_the_callers_and_the_resolver_honours_it() {
250        let exec = Exec::current().expect("ambient runtime");
251        let table = SharedResolver::new(Table(vec![
252            "203.0.113.7:7443".parse().expect("addr"),
253            "203.0.113.7:7444".parse().expect("addr"),
254            "203.0.113.7:7445".parse().expect("addr"),
255        ]));
256        let addrs = table
257            .resolve(&exec, "lb.example", None, 2)
258            .await
259            .expect("table");
260        assert_eq!(addrs.len(), 2, "a resolver answer is remote input");
261    }
262
263    #[test]
264    fn two_resolvers_are_equal_only_when_they_are_the_same_one() {
265        let one = SharedResolver::default();
266        let same = one.clone();
267        let other = SharedResolver::default();
268        assert_eq!(one, same);
269        assert_ne!(
270            one, other,
271            "identical behaviour is not identity: the pool keys on this"
272        );
273    }
274
275    #[tokio::test]
276    async fn resolves_ip_literals_without_dns() {
277        let exec = Exec::current().expect("ambient runtime");
278        assert_eq!(
279            SystemResolver
280                .resolve(&exec, "127.0.0.1", Some(7443), 8)
281                .await
282                .expect("v4"),
283            vec![SocketAddr::from(([127, 0, 0, 1], 7443))]
284        );
285        let v6 = SystemResolver
286            .resolve(&exec, "::1", Some(7443), 8)
287            .await
288            .expect("v6");
289        assert_eq!(v6.len(), 1);
290        assert_eq!(v6[0].port(), 7443);
291        assert!(v6[0].is_ipv6());
292    }
293
294    /// Claim: a hostname yields every address the resolver offers, in its
295    /// order and no more than the cap. `localhost` is the case that matters —
296    /// it commonly resolves to both `::1` and `127.0.0.1`, and dialling only
297    /// the first reaches a server bound to the other never.
298    #[tokio::test]
299    async fn a_hostname_resolves_to_every_address_up_to_the_cap() {
300        let exec = Exec::current().expect("ambient runtime");
301        let all = SystemResolver
302            .resolve(&exec, "localhost", Some(7443), 8)
303            .await
304            .expect("localhost");
305        assert!(!all.is_empty());
306        assert!(all.iter().all(|a| a.port() == 7443));
307
308        let capped = SystemResolver
309            .resolve(&exec, "localhost", Some(7443), 1)
310            .await
311            .expect("localhost");
312        assert_eq!(capped.len(), 1, "the cap must bound the answer");
313        assert_eq!(capped[0], all[0], "and it must keep the resolver's order");
314    }
315}
weida_runtime/resolve.rs

weida_runtime/
resolve.rs
