chore: Merge resolver service into client service

This commit is contained in:
theS1LV3R authored and Ellis Git committed 2026-07-09 18:22:25 +00:00
1 parent 2a342bcbc9
commit 0d5789f727
14 files changed
+91 -376

No files matched your search

Generated
+1 -2
View File
@@ -4540,8 +4540,7 @@ checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7"
[[package]]
name = "resolvematrix"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "19ba4985332bf3cc4f998f7da7efb88bf1c68b753eee5bbd11c33afe12a192f6"
source = "git+https://forgejo.ellis.link/continuwuation/resolvematrix?rev=b2280bc4eaacea3de18bc47534cddeeeb1ceab9c#b2280bc4eaacea3de18bc47534cddeeeb1ceab9c"
dependencies = [
"futures",
"hickory-resolver",
+4 -1
View File
@@ -570,7 +570,10 @@ features = ["std"]
version = "0.3.0"
[workspace.dependencies.resolvematrix]
version = "0.1.2"
#version = "0.1.2"
rev = "b2280bc4eaacea3de18bc47534cddeeeb1ceab9c"
git = "https://forgejo.ellis.link/continuwuation/resolvematrix"
#branch = "zoe/cache-get-all"
[workspace.dependencies.serde_urlencoded]
version = "0.7.1"
+5 -2
View File
@@ -1017,7 +1017,7 @@ pub(super) async fn resolve_true_destination(
.http_client(self.services.client.default.clone())
.build()?
} else {
&self.services.resolver.resolver
&self.services.client.resolver
};
let actual = resolver.resolve_server(server_name.as_str()).await?;
@@ -1027,7 +1027,10 @@ pub(super) async fn resolve_true_destination(
| ResolvedDestination::Named(host, port) => format!("{host}:{port}"),
};
let msg = format!("Destination: {}\nHostname URI (SNI): {}", destination, actual.host);
let msg = format!(
"Destination: {}\nHostname URI (SNI): {}\nIs override?: {}\nResolution step: {}",
destination, actual.host, actual.is_override, actual.resolution_step
);
self.write_str(&msg).await
}
+36 -49
View File
@@ -1,6 +1,6 @@
use clap::Subcommand;
use conduwuit::{Err, Result, utils::time};
use futures::StreamExt;
use resolvematrix::resolution::Resolution;
use ruma::OwnedServerName;
use crate::admin_command_dispatch;
@@ -10,14 +10,13 @@
#[allow(clippy::enum_variant_names)]
/// Resolver service and caches
pub enum ResolverCommand {
/// Query the destinations cache
DestinationsCache {
/// Query the destinations or overrides cache, depending on the value of the
/// `overrides` flag (default false)
Cache {
server_name: Option<OwnedServerName>,
},
/// Query the overrides cache
OverridesCache {
name: Option<String>,
#[arg(short, long)]
overrides: Option<bool>,
},
/// Flush a given server from the resolver caches or flush them completely
@@ -40,49 +39,39 @@ pub enum ResolverCommand {
}
impl crate::Context<'_> {
async fn destinations_cache(&self, server_name: Option<OwnedServerName>) -> Result {
use service::resolver::cache::CachedDest;
async fn cache(
&self,
server_name: Option<OwnedServerName>,
overrides: Option<bool>,
) -> Result {
writeln!(self, "| Server Name | Destination | SNI | Override | Step | Expires |").await?;
writeln!(self, "| ----------- | ----------- | --- | -------- | ---- | ------- |").await?;
writeln!(self, "| Server Name | Destination | Hostname | Expires |").await?;
writeln!(self, "| ----------- | ----------- | -------- | ------- |").await?;
let entries = self.services.client.resolver.get_all_cache_entries();
let mut destinations = self.services.resolver.dns.cache.destinations().boxed();
while let Some((name, CachedDest { dest, host, expire })) = destinations.next().await {
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
}
for (host, entry) in entries {
if let Some(ors) = overrides
&& entry.is_override != ors
{
continue;
}
let expire = time::format(expire, "%+");
self.write_str(&format!("| {name} | {dest} | {host} | {expire} |\n"))
.await?;
}
Ok(())
}
async fn overrides_cache(&self, server_name: Option<String>) -> Result {
use service::resolver::cache::CachedOverride;
writeln!(self, "| Server Name | IP | Port | Expires | Overriding |").await?;
writeln!(self, "| ----------- | --- | ----:| ------- | ---------- |").await?;
let mut overrides = self.services.resolver.dns.cache.overrides().boxed();
while let Some((name, CachedOverride { ips, port, expire, overriding })) =
overrides.next().await
{
if let Some(server_name) = server_name.as_ref() {
if name != *server_name {
continue;
}
if let Some(server_name) = server_name.as_ref()
&& host != *server_name
{
continue;
}
let expire = time::format(expire, "%+");
let Resolution {
destination,
is_override,
host: sni_host,
resolution_step,
} = entry.resolution;
let expires = time::format(entry.expires_at, "%Y-%m-%dT%H:%M:%S%.3f%z");
self.write_str(&format!(
"| {name} | {ips:?} | {port} | {expire} | {overriding:?} |\n"
"| {host} | {destination:?} | {sni_host} | {is_override:?} | {resolution_step} \
| {expires} |\n"
))
.await?;
}
@@ -92,16 +81,14 @@ async fn overrides_cache(&self, server_name: Option<String>) -> Result {
async fn flush_cache(&self, name: Option<OwnedServerName>, all: bool) -> Result {
if all {
self.services.resolver.resolver.clear_cache();
self.services.resolver.dns.cache.clear().await;
writeln!(self, "Resolver caches cleared!").await
self.services.client.resolver.clear_cache();
self.services.client.dns.clear_cache();
writeln!(self, "Resolver and DNS caches cleared!").await
} else if let Some(name) = name {
self.services
.resolver
.client
.resolver
.remove_cache_entry(name.as_str());
self.services.resolver.dns.cache.del_destination(&name);
self.services.resolver.dns.cache.del_override(&name);
self.write_str(&format!("Cleared {name} from resolver caches!"))
.await
} else {
@@ -8,17 +8,8 @@
};
use reqwest::dns::{Addrs, Name, Resolve, Resolving};
use super::cache::{Cache, CachedOverride};
pub struct Resolver {
pub(crate) resolver: Arc<TokioResolver>,
pub(crate) hooked: Arc<Hooked>,
server: Arc<Server>,
}
pub(crate) struct Hooked {
resolver: Arc<TokioResolver>,
cache: Arc<Cache>,
server: Arc<Server>,
}
@@ -26,7 +17,7 @@ pub(crate) struct Hooked {
impl Resolver {
#[allow(clippy::as_conversions, clippy::cast_sign_loss, clippy::cast_possible_truncation)]
pub(super) fn build(server: &Arc<Server>, cache: Arc<Cache>) -> Result<Arc<Self>> {
pub(super) fn build(server: &Arc<Server>) -> Result<Arc<Self>> {
let config = &server.config;
let (sys_conf, mut opts) = hickory_resolver::system_conf::read_system_conf()
.map_err(|e| err!(error!("Failed to configure DNS resolver from system: {e}")))?;
@@ -80,7 +71,6 @@ pub(super) fn build(server: &Arc<Server>, cache: Arc<Cache>) -> Result<Arc<Self>
Ok(Arc::new(Self {
resolver: resolver.clone(),
hooked: Arc::new(Hooked { resolver, cache, server: server.clone() }),
server: server.clone(),
}))
}
@@ -96,43 +86,6 @@ fn resolve(&self, name: Name) -> Resolving {
}
}
impl Resolve for Hooked {
fn resolve(&self, name: Name) -> Resolving {
hooked_resolve(self.cache.clone(), self.server.clone(), self.resolver.clone(), name)
.boxed()
}
}
#[tracing::instrument(
level = "debug",
skip_all,
fields(name = ?name.as_str())
)]
async fn hooked_resolve(
cache: Arc<Cache>,
server: Arc<Server>,
resolver: Arc<TokioResolver>,
name: Name,
) -> Result<Addrs, Box<dyn std::error::Error + Send + Sync>> {
match cache.get_override(name.as_str()).await {
| Ok(cached) if cached.valid() => cached_to_reqwest(cached).await,
| Ok(CachedOverride { overriding, .. }) if overriding.is_some() =>
resolve_to_reqwest(
server,
resolver,
overriding
.as_deref()
.map(str::parse)
.expect("overriding is set for this record")
.expect("overriding is a valid internet name"),
)
.boxed()
.await,
| _ => resolve_to_reqwest(server, resolver, name).boxed().await,
}
}
async fn resolve_to_reqwest(
server: Arc<Server>,
resolver: Arc<TokioResolver>,
@@ -156,12 +109,3 @@ async fn resolve_to_reqwest(
() = server.until_shutdown() => Err(handle_shutdown()),
}
}
async fn cached_to_reqwest(cached: CachedOverride) -> ResolvingResult {
let addrs = cached
.ips
.into_iter()
.map(move |ip| SocketAddr::new(ip, cached.port));
Ok(Box::new(addrs))
}
+39 -12
View File
@@ -1,13 +1,20 @@
mod dns;
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use conduwuit::{Config, Result, err, trace};
use either::Either;
use ipaddress::IPAddress;
use reqwest::redirect;
use resolvematrix::server::{MatrixResolver, MatrixResolverBuilder};
use crate::{resolver, service};
use crate::{client::dns::Resolver, service};
pub struct Service {
pub resolver: MatrixResolver,
pub dns: Arc<Resolver>,
pub default: reqwest::Client,
pub url_preview: reqwest::Client,
pub extern_media: reqwest::Client,
@@ -20,10 +27,24 @@ pub struct Service {
pub cidr_range_denylist: Vec<IPAddress>,
}
#[async_trait]
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
let config = &args.server.config;
let resolver = args.require::<resolver::Service>("resolver");
let dns = Resolver::build(args.server)?;
let resolver = MatrixResolverBuilder::new()
.dangerous_tls_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
.http_client(
base(&args.server.config)?
.connect_timeout(Duration::from_secs(args.server.config.well_known_conn_timeout))
.read_timeout(Duration::from_secs(args.server.config.well_known_timeout))
.timeout(Duration::from_secs(args.server.config.well_known_timeout))
.pool_max_idle_per_host(0)
.redirect(redirect::Policy::limited(4))
.build()?
)
.dns_resolver(dns.resolver.clone())
.build()?;
let url_preview_bind_addr = config
.url_preview_bound_interface
@@ -41,28 +62,29 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.unwrap_or_else(|| conduwuit::user_agent_media().to_owned());
Ok(Arc::new(Self {
default: base(config)?
.dns_resolver(resolver.dns.resolver.clone())
.build()?,
resolver,
dns: dns.clone(),
default: base(config)?.dns_resolver(dns.clone()).build()?,
url_preview: base(config)
.and_then(|builder| {
builder_interface(builder, url_preview_bind_iface.as_deref())
})?
.local_address(url_preview_bind_addr)
.dns_resolver(resolver.dns.resolver.clone())
.dns_resolver(dns.clone())
.timeout(Duration::from_secs(config.url_preview_timeout))
.redirect(redirect::Policy::limited(3))
.user_agent(url_preview_user_agent)
.build()?,
extern_media: base(config)?
.dns_resolver(resolver.dns.resolver.clone())
.dns_resolver(dns.clone())
.redirect(redirect::Policy::limited(3))
.build()?,
federation: base(config)?
.dns_resolver(resolver.dns.resolver.hooked.clone())
.dns_resolver(dns.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.federation_timeout))
.timeout(Duration::from_secs(
@@ -76,7 +98,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
federation_slow: base(config)?
.dns_resolver(resolver.dns.resolver.hooked.clone())
.dns_resolver(dns.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.federation_timeout.saturating_mul(6)))
.timeout(Duration::from_secs(
@@ -90,7 +112,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
sender: base(config)?
.dns_resolver(resolver.dns.resolver.hooked.clone())
.dns_resolver(dns.clone())
.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
.read_timeout(Duration::from_secs(config.sender_timeout))
.timeout(Duration::from_secs(config.sender_timeout))
@@ -100,7 +122,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
appservice: base(config)?
.dns_resolver(resolver.dns.resolver.clone())
.dns_resolver(dns.clone())
.connect_timeout(Duration::from_secs(5))
.read_timeout(Duration::from_secs(config.appservice_timeout))
.timeout(Duration::from_secs(config.appservice_timeout))
@@ -110,7 +132,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
.build()?,
pusher: base(config)?
.dns_resolver(resolver.dns.resolver.clone())
.dns_resolver(dns.clone())
.connect_timeout(Duration::from_secs(config.pusher_conn_timeout))
.timeout(Duration::from_secs(config.pusher_timeout))
.pool_max_idle_per_host(1)
@@ -128,6 +150,11 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
}))
}
async fn clear_cache(&self) {
self.resolver.clear_cache();
self.dns.resolver.clear_cache();
}
fn name(&self) -> &str { service::make_name(std::module_path!()) }
}
+1 -1
View File
@@ -133,7 +133,7 @@ pub async fn execute_on<'i, T, PathBuilderInput>(
let actual = self
.services
.resolver
.client
.resolver
.resolve_server(dest.as_str())
.await?;
+1 -3
View File
@@ -4,7 +4,7 @@
use conduwuit::{Result, Server};
pub(crate) use execute::FederationPathBuilderInput;
use crate::{Dep, client, moderation, resolver, server_keys};
use crate::{Dep, client, moderation, server_keys};
pub struct Service {
services: Services,
@@ -13,7 +13,6 @@ pub struct Service {
struct Services {
server: Arc<Server>,
client: Dep<client::Service>,
resolver: Dep<resolver::Service>,
server_keys: Dep<server_keys::Service>,
moderation: Dep<moderation::Service>,
}
@@ -24,7 +23,6 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
services: Services {
server: args.server.clone(),
client: args.depend::<client::Service>("client"),
resolver: args.depend::<resolver::Service>("resolver"),
server_keys: args.depend::<server_keys::Service>("server_keys"),
moderation: args.depend::<moderation::Service>("moderation"),
},
-1
View File
@@ -32,7 +32,6 @@
pub mod presence;
pub mod pusher;
pub mod registration_tokens;
pub mod resolver;
pub mod rooms;
pub mod sending;
pub mod server_keys;
-135
View File
@@ -1,135 +0,0 @@
use std::{net::IpAddr, sync::Arc, time::SystemTime};
use conduwuit::{
Result,
arrayvec::ArrayVec,
at, err,
utils::{math::Expected, stream::TryIgnore},
};
use database::{Cbor, Deserialized, Map};
use futures::{Stream, StreamExt, future::join};
use ruma::{OwnedServerName, ServerName};
use serde::{Deserialize, Serialize};
use super::fed::FedDest;
pub struct Cache {
destinations: Arc<Map>,
overrides: Arc<Map>,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct CachedDest {
pub dest: FedDest,
pub host: String,
pub expire: SystemTime,
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct CachedOverride {
pub ips: IpAddrs,
pub port: u16,
pub expire: SystemTime,
pub overriding: Option<String>,
}
pub type IpAddrs = ArrayVec<IpAddr, MAX_IPS>;
pub(crate) const MAX_IPS: usize = 3;
impl Cache {
pub(super) fn new(args: &crate::Args<'_>) -> Arc<Self> {
Arc::new(Self {
destinations: args.db["servername_destination"].clone(),
overrides: args.db["servername_override"].clone(),
})
}
pub async fn clear(&self) { join(self.clear_destinations(), self.clear_overrides()).await; }
pub async fn clear_destinations(&self) { self.destinations.clear().await; }
pub async fn clear_overrides(&self) { self.overrides.clear().await; }
pub fn del_destination(&self, name: &ServerName) { self.destinations.remove(name); }
pub fn del_override(&self, name: &ServerName) { self.overrides.remove(name); }
pub fn set_destination(&self, name: &ServerName, dest: &CachedDest) {
self.destinations.raw_put(name, Cbor(dest));
}
pub fn set_override(&self, name: &str, over: &CachedOverride) {
self.overrides.raw_put(name, Cbor(over));
}
#[must_use]
pub async fn has_destination(&self, destination: &ServerName) -> bool {
self.get_destination(destination).await.is_ok()
}
#[must_use]
pub async fn has_override(&self, destination: &str) -> bool {
self.get_override(destination)
.await
.iter()
.any(CachedOverride::valid)
}
pub async fn get_destination(&self, name: &ServerName) -> Result<CachedDest> {
self.destinations
.get(name)
.await
.deserialized::<Cbor<_>>()
.map(at!(0))
.into_iter()
.find(CachedDest::valid)
.ok_or(err!(Request(NotFound("Expired from cache"))))
}
pub async fn get_override(&self, name: &str) -> Result<CachedOverride> {
self.overrides
.get(name)
.await
.deserialized::<Cbor<_>>()
.map(at!(0))
}
pub fn destinations(&self) -> impl Stream<Item = (OwnedServerName, CachedDest)> + Send + '_ {
self.destinations
.stream()
.ignore_err()
.map(|item: (OwnedServerName, Cbor<_>)| (item.0, item.1.0))
}
pub fn overrides(&self) -> impl Stream<Item = (OwnedServerName, CachedOverride)> + Send + '_ {
self.overrides
.stream()
.ignore_err()
.map(|item: (OwnedServerName, Cbor<_>)| (item.0, item.1.0))
}
}
impl CachedDest {
#[inline]
#[must_use]
pub fn valid(&self) -> bool { self.expire > SystemTime::now() }
#[inline]
#[must_use]
pub fn size(&self) -> usize {
self.dest
.size()
.expected_add(self.host.len())
.expected_add(size_of_val(&self.expire))
}
}
impl CachedOverride {
#[inline]
#[must_use]
pub fn valid(&self) -> bool { self.expire > SystemTime::now() }
#[inline]
#[must_use]
pub fn size(&self) -> usize { size_of_val(self) }
}
-45
View File
@@ -1,45 +0,0 @@
use std::{fmt, net::SocketAddr};
use conduwuit::{arrayvec::ArrayString, utils::math::Expected};
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
pub enum FedDest {
Literal(SocketAddr), // "ip:port"
Named(String, PortString), // ("hostname", ":port")
}
/// numeric or service-name
pub type PortString = ArrayString<16>;
const DEFAULT_PORT: &str = ":8448";
impl FedDest {
pub(crate) fn uri_string(&self) -> String {
match self {
| Self::Literal(addr) => addr.to_string(),
| Self::Named(host, port) => format!("{host}{port}"),
}
}
#[inline]
#[must_use]
pub fn default_port() -> PortString {
PortString::from(DEFAULT_PORT).expect("default port string")
}
#[inline]
#[must_use]
pub fn size(&self) -> usize {
match self {
| Self::Literal(saddr) => size_of_val(saddr),
| Self::Named(host, port) => host.len().expected_add(port.capacity()),
}
}
}
impl fmt::Display for FedDest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(self.uri_string().as_str())
}
}
-64
View File
@@ -1,64 +0,0 @@
pub mod cache;
mod dns;
pub mod fed;
use std::{sync::Arc, time::Duration};
use async_trait::async_trait;
use conduwuit::Result;
use reqwest::redirect;
use resolvematrix::server::{MatrixResolver, MatrixResolverBuilder};
use self::{cache::Cache, dns::Resolver};
use crate::client::base;
pub struct Service {
pub resolver: MatrixResolver,
pub dns: Dns,
#[allow(dead_code)] // This service doesn't access services after construction
services: Services,
}
struct Services;
pub struct Dns {
pub cache: Arc<Cache>,
pub resolver: Arc<Resolver>,
}
#[async_trait]
impl crate::Service for Service {
#[allow(clippy::as_conversions, clippy::cast_sign_loss, clippy::cast_possible_truncation)]
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
let cache = Cache::new(&args);
let resolver = Resolver::build(args.server, cache.clone())?;
Ok(Arc::new(Self {
resolver: MatrixResolverBuilder::new()
.dangerous_tls_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
.http_client(
base(&args.server.config)?
.connect_timeout(Duration::from_secs(args.server.config.well_known_conn_timeout))
.read_timeout(Duration::from_secs(args.server.config.well_known_timeout))
.timeout(Duration::from_secs(args.server.config.well_known_timeout))
.pool_max_idle_per_host(0)
.redirect(redirect::Policy::limited(4))
.build()?
)
.dns_resolver(resolver.resolver.clone())
.build()?,
dns: Dns {
cache,
resolver,
},
services: Services {},
}))
}
async fn clear_cache(&self) {
self.resolver.clear_cache();
self.dns.resolver.clear_cache();
self.dns.cache.clear().await;
}
fn name(&self) -> &str { crate::service::make_name(module_path!()) }
}
+1
View File
@@ -121,6 +121,7 @@ pub(crate) fn depend<T: Service>(&'a self, name: &'static str) -> Dep<T> {
/// Create a reference immediately to a service when constructing another
/// Service. The other service must be constructed.
#[inline]
#[allow(dead_code)]
pub(crate) fn require<T: Service>(&'a self, name: &str) -> Arc<T> {
require::<T>(self.service, name)
}
+2 -4
View File
@@ -11,8 +11,8 @@
account_data, admin, announcements, antispam, appservice, client, config, emergency,
federation, firstrun, globals, key_backups, mailer,
manager::Manager,
media, moderation, oauth, oidc, presence, pusher, registration_tokens, resolver, rooms,
sending, server_keys,
media, moderation, oauth, oidc, presence, pusher, registration_tokens, rooms, sending,
server_keys,
service::{self, Args, Map, Service},
sync, threepid, transactions, uiaa, users,
};
@@ -33,7 +33,6 @@ pub struct Services {
pub presence: Arc<presence::Service>,
pub pusher: Arc<pusher::Service>,
pub registration_tokens: Arc<registration_tokens::Service>,
pub resolver: Arc<resolver::Service>,
pub rooms: rooms::Service,
pub federation: Arc<federation::Service>,
pub firstrun: Arc<firstrun::Service>,
@@ -78,7 +77,6 @@ macro_rules! build {
account_data: build!(account_data::Service),
admin: build!(admin::Service),
appservice: build!(appservice::Service),
resolver: build!(resolver::Service),
client: build!(client::Service),
config: build!(config::Service),
emergency: build!(emergency::Service),