feat(iroh-dns-server)!: eviction of stale zonestore entries (#2997)

## Description

Configurable eviction of stale zonestore entries.

This works by taking a snapshot of the database in regular intervals and
checking if a record is possibly expired. For all possibly expired
records a fire and forget message will be sent to the io write actor to
check again. The actor will check the expiry again (entry could have
been renewed since the last snapshot) and then deletes it if the final
check confirms expiry.

Between expiry checks there is a configurable delay, so the thread is
not constantly spinning checking for expiry.

We use a second thread since we don't want to block writing new entries
by sifting through old entries.

## Breaking Changes

- struct config::Config has a new field zone_store
- struct metrics::Metrics has a new field store_packets_expired

## Notes & open questions

Note: there are two ways to do eviction. One is to carefully keep track
of the time for each entry by having a second table that has (timestamp,
key) as the key and () as the value. Then you could just evict without
doing a full scan by sorting by time ascending.

The downside of course is that you need an entire new table, and you
need to update this table every time you update an entry (delete (old
time, id), insert (new time, id).

~~So I decided to just do a full scan instead for simplicity. We can
change it if it becomes a problem.~~

~~Hm, maybe we should avoid the full scan after all. I can imagine the
thing being a bit less responsive than usual while the scan is ongoing.
Another idea would be to have an "event log" where you just store (time,
id) -> () and then use that to look for eviction *candidates*. Don't
bother cleaning up this event log on every update.~~

I have now implemented a second table. It is a multimap table from
timestamp to id. This gets updated on every write (slight perf downside
here), and can be used to scan for evictions without having to do a full
scan.

There is quite a lot of code just to expose these config options. We
could also omit this and just use reasonable defaults.

## Change checklist

- [ ] Self-review.
- [ ] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [ ] Tests if relevant.
- [ ] All breaking changes documented.

---------

Co-authored-by: dignifiedquire <[email protected]>
Co-authored-by: Asmir Avdicevic <[email protected]>

Author: 3 people

Cargo.lock

@@ -29,7 +29,8 @@ governor = "0.6.3" #needs new release of tower_governor for 0.7.0
 hickory-proto = "=0.25.0-alpha.2"
 hickory-server = { version = "=0.25.0-alpha.2", features = ["dns-over-rustls"] }
 http = "1.0.0"
-iroh-metrics = "0.29"
+humantime-serde = "1.1.1"
+iroh-metrics = { version = "0.29.0" }
 lru = "0.12.3"
 parking_lot = "0.12.1"
 pkarr = { version = "2.2.0", features = [ "async", "relay", "dht"], default-features = false }
@@ -64,6 +65,7 @@ hickory-resolver = "=0.25.0-alpha.2"
 iroh = { version = "0.29.0", path = "../iroh" }
 iroh-test = { version = "0.29.0", path = "../iroh-test" }
 pkarr = { version = "2.2.0", features = ["rand"] }
+testresult = "0.4.1"
 
 [[bench]]
 name = "write"

iroh-dns-server/Cargo.toml

@@ -7,7 +7,7 @@ use tokio::runtime::Runtime;
 const LOCALHOST_PKARR: &str = "http://localhost:8080/pkarr";
 
 async fn start_dns_server(config: Config) -> Result<Server> {
-    let store = ZoneStore::persistent(Config::signed_packet_store_path()?)?;
+    let store = ZoneStore::persistent(Config::signed_packet_store_path()?, Default::default())?;
     Server::spawn(config, store).await
 }
 

iroh-dns-server/benches/write.rs

@@ -4,6 +4,7 @@ use std::{
     env,
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::{Path, PathBuf},
+    time::Duration,
 };
 
 use anyhow::{anyhow, Context, Result};
@@ -13,6 +14,7 @@ use tracing::info;
 use crate::{
     dns::DnsConfig,
     http::{CertMode, HttpConfig, HttpsConfig, RateLimitConfig},
+    store::ZoneStoreOptions,
 };
 
 const DEFAULT_METRICS_ADDR: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9117);
@@ -44,11 +46,61 @@ pub struct Config {
     /// Config for the mainline lookup.
     pub mainline: Option<MainlineConfig>,
 
+    /// Config for the zone store.
+    pub zone_store: Option<StoreConfig>,
+
     /// Config for pkarr rate limit
     #[serde(default)]
     pub pkarr_put_rate_limit: RateLimitConfig,
 }
 
+/// The config for the store.
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct StoreConfig {
+    /// Maximum number of packets to process in a single write transaction.
+    max_batch_size: usize,
+
+    /// Maximum time to keep a write transaction open.
+    #[serde(with = "humantime_serde")]
+    max_batch_time: Duration,
+
+    /// Time to keep packets in the store before eviction.
+    #[serde(with = "humantime_serde")]
+    eviction: Duration,
+
+    /// Pause between eviction checks.
+    #[serde(with = "humantime_serde")]
+    eviction_interval: Duration,
+}
+
+impl Default for StoreConfig {
+    fn default() -> Self {
+        ZoneStoreOptions::default().into()
+    }
+}
+
+impl From<ZoneStoreOptions> for StoreConfig {
+    fn from(value: ZoneStoreOptions) -> Self {
+        Self {
+            max_batch_size: value.max_batch_size,
+            max_batch_time: value.max_batch_time,
+            eviction: value.eviction,
+            eviction_interval: value.eviction_interval,
+        }
+    }
+}
+
+impl From<StoreConfig> for ZoneStoreOptions {
+    fn from(value: StoreConfig) -> Self {
+        Self {
+            max_batch_size: value.max_batch_size,
+            max_batch_time: value.max_batch_time,
+            eviction: value.eviction,
+            eviction_interval: value.eviction_interval,
+        }
+    }
+}
+
 /// The config for the metrics server.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct MetricsConfig {
@@ -187,6 +239,7 @@ impl Default for Config {
                 rr_aaaa: None,
                 rr_ns: Some("ns1.irohdns.example.".to_string()),
             },
+            zone_store: None,
             metrics: None,
             mainline: None,
             pkarr_put_rate_limit: RateLimitConfig::default(),

iroh-dns-server/src/config.rs

@@ -16,7 +16,10 @@ pub use store::ZoneStore;
 
 #[cfg(test)]
 mod tests {
-    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+    use std::{
+        net::{Ipv4Addr, Ipv6Addr, SocketAddr},
+        time::Duration,
+    };
 
     use anyhow::Result;
     use hickory_resolver::{
@@ -29,9 +32,16 @@ mod tests {
         key::SecretKey,
     };
     use pkarr::{PkarrClient, SignedPacket};
+    use testresult::TestResult;
     use url::Url;
 
-    use crate::{config::BootstrapOption, server::Server};
+    use crate::{
+        config::BootstrapOption,
+        server::Server,
+        store::{PacketSource, ZoneStoreOptions},
+        util::PublicKeyBytes,
+        ZoneStore,
+    };
 
     #[tokio::test]
     async fn pkarr_publish_dns_resolve() -> Result<()> {
@@ -178,6 +188,36 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn store_eviction() -> TestResult<()> {
+        iroh_test::logging::setup_multithreaded();
+        let options = ZoneStoreOptions {
+            eviction: Duration::from_millis(100),
+            eviction_interval: Duration::from_millis(100),
+            max_batch_time: Duration::from_millis(100),
+            ..Default::default()
+        };
+        let store = ZoneStore::in_memory(options)?;
+
+        // create a signed packet
+        let signed_packet = random_signed_packet()?;
+        let key = PublicKeyBytes::from_signed_packet(&signed_packet);
+
+        store
+            .insert(signed_packet, PacketSource::PkarrPublish)
+            .await?;
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        for _ in 0..10 {
+            let entry = store.get_signed_packet(&key).await?;
+            if entry.is_none() {
+                return Ok(());
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+        panic!("store did not evict packet");
+    }
+
     #[tokio::test]
     async fn integration_mainline() -> Result<()> {
         iroh_test::logging::setup_multithreaded();
@@ -188,7 +228,8 @@ mod tests {
 
         // spawn our server with mainline support
         let (server, nameserver, _http_url) =
-            Server::spawn_for_tests_with_mainline(Some(BootstrapOption::Custom(bootstrap))).await?;
+            Server::spawn_for_tests_with_options(Some(BootstrapOption::Custom(bootstrap)), None)
+                .await?;
 
         let origin = "irohdns.example.";
 
@@ -228,4 +269,12 @@ mod tests {
         config.add_name_server(nameserver_config);
         AsyncResolver::tokio(config, Default::default())
     }
+
+    fn random_signed_packet() -> Result<SignedPacket> {
+        let secret_key = SecretKey::generate();
+        let node_id = secret_key.public();
+        let relay_url: Url = "https://relay.example.".parse()?;
+        let node_info = NodeInfo::new(node_id, Some(relay_url.clone()), Default::default());
+        node_info.to_pkarr_signed_packet(&secret_key, 30)
+    }
 }

iroh-dns-server/src/lib.rs

@@ -22,6 +22,7 @@ pub struct Metrics {
     pub store_packets_inserted: Counter,
     pub store_packets_removed: Counter,
     pub store_packets_updated: Counter,
+    pub store_packets_expired: Counter,
 }
 
 impl Default for Metrics {
@@ -44,6 +45,7 @@ impl Default for Metrics {
             store_packets_inserted: Counter::new("Signed packets inserted into the store"),
             store_packets_removed: Counter::new("Signed packets removed from the store"),
             store_packets_updated: Counter::new("Number of updates to existing packets"),
+            store_packets_expired: Counter::new("Number of expired packets"),
         }
     }
 }

iroh-dns-server/src/metrics.rs

@@ -14,7 +14,11 @@ use crate::{
 
 /// Spawn the server and run until the `Ctrl-C` signal is received, then shutdown.
 pub async fn run_with_config_until_ctrl_c(config: Config) -> Result<()> {
-    let mut store = ZoneStore::persistent(Config::signed_packet_store_path()?)?;
+    let zone_store_options = config.zone_store.clone().unwrap_or_default();
+    let mut store = ZoneStore::persistent(
+        Config::signed_packet_store_path()?,
+        zone_store_options.into(),
+    )?;
     if let Some(bootstrap) = config.mainline_enabled() {
         info!("mainline fallback enabled");
         store = store.with_mainline_fallback(bootstrap);
@@ -96,14 +100,15 @@ impl Server {
     /// HTTP server.
     #[cfg(test)]
     pub async fn spawn_for_tests() -> Result<(Self, std::net::SocketAddr, url::Url)> {
-        Self::spawn_for_tests_with_mainline(None).await
+        Self::spawn_for_tests_with_options(None, None).await
     }
 
     /// Spawn a server suitable for testing, while optionally enabling mainline with custom
     /// bootstrap addresses.
     #[cfg(test)]
-    pub async fn spawn_for_tests_with_mainline(
+    pub async fn spawn_for_tests_with_options(
         mainline: Option<crate::config::BootstrapOption>,
+        options: Option<crate::store::ZoneStoreOptions>,
     ) -> Result<(Self, std::net::SocketAddr, url::Url)> {
         use std::net::{IpAddr, Ipv4Addr};
 
@@ -117,7 +122,7 @@ impl Server {
         config.https = None;
         config.metrics = Some(MetricsConfig::disabled());
 
-        let mut store = ZoneStore::in_memory()?;
+        let mut store = ZoneStore::in_memory(options.unwrap_or_default())?;
         if let Some(bootstrap) = mainline {
             info!("mainline fallback enabled");
             store = store.with_mainline_fallback(bootstrap);

iroh-dns-server/src/server.rs

@@ -19,6 +19,7 @@ use crate::{
 };
 
 mod signed_packets;
+pub use signed_packets::Options as ZoneStoreOptions;
 
 /// Cache up to 1 million pkarr zones by default
 pub const DEFAULT_CACHE_CAPACITY: usize = 1024 * 1024;
@@ -44,14 +45,14 @@ pub struct ZoneStore {
 
 impl ZoneStore {
     /// Create a persistent store
-    pub fn persistent(path: impl AsRef<Path>) -> Result<Self> {
-        let packet_store = SignedPacketStore::persistent(path)?;
+    pub fn persistent(path: impl AsRef<Path>, options: ZoneStoreOptions) -> Result<Self> {
+        let packet_store = SignedPacketStore::persistent(path, options)?;
         Ok(Self::new(packet_store))
     }
 
     /// Create an in-memory store.
-    pub fn in_memory() -> Result<Self> {
-        let packet_store = SignedPacketStore::in_memory()?;
+    pub fn in_memory(options: ZoneStoreOptions) -> Result<Self> {
+        let packet_store = SignedPacketStore::in_memory(options)?;
         Ok(Self::new(packet_store))
     }