fix(iroh-dns-server): Fixes for packet expiry (#3297)

## Description

Based on #3295 

I noticed in the debug logs of our staging server that some entries in
the expiry table were never deleted, and thus they were processed again
and again on each expiry tick.

This PR has the following changes:

* Bugfix: When upserting a packet, we were not properly clearing the
*old* expiry entry! [This
line](https://github.com/n0-computer/iroh/blob/f3c9af35542a7f2e00303090d1ca5273f86bc375/iroh-dns-server/src/store/signed_packets.rs#L148)
is wrong: It should be `existing.timestamp()`, because that's the entry
we want to remove, not the (not-yet-existing) entry for the timestamp of
the *new* packet.
* Improve type handling and debug formatting around expiry timestamps,
to make logs better readable and be less error prone to bad type
conversions
* When an expiry entry is being processed, and the corresponding packet
does not exist anymore, the expiry entry is deleted

## Breaking Changes

<!-- Optional, if there are any breaking changes document them,
including how to migrate older code. -->

## Notes & open questions

We need tests around expiry. This should've been caught earlier. There's
currently only a [single
test](https://github.com/n0-computer/iroh/blob/f3c9af35542a7f2e00303090d1ca5273f86bc375/iroh-dns-server/src/lib.rs#L190)
around eviction, added in the PR that added eviction (#2997). But that
test only tests that expiry works at all. There's no test currently that
verifies if entries in the expiry table are removed when a packet is
upserted, which is the test that would fail without this PR.

## Change checklist
<!-- Remove any that are not relevant. -->
- [x] Self-review.
- [x] 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: Would be relevant, but needs to come later.

Author: Frando

Cargo.lock

@@ -27,6 +27,7 @@ dirs-next = "2.0.0"
 governor = "0.8"
 hickory-server = { version = "0.25.1", features = ["https-ring"] }
 http = "1.0.0"
+humantime = "2.2.0"
 humantime-serde = "1.1.1"
 iroh-metrics = { version = "0.34", features = ["service"] }
 lru = "0.12.3"

iroh-dns-server/Cargo.toml

@@ -105,6 +105,7 @@ async fn main() -> Result<()> {
         .with_direct_addresses(args.addr.into_iter().collect())
         .with_user_data(args.user_data);
     let signed_packet = node_info.to_pkarr_signed_packet(&secret_key, 30)?;
+    tracing::debug!("signed packet: {signed_packet:?}");
     pkarr.publish(&signed_packet).await?;
 
     println!("signed packet published.");

iroh-dns-server/examples/publish.rs

@@ -1,4 +1,10 @@
-use std::{future::Future, path::Path, result, sync::Arc, time::Duration};
+use std::{
+    future::Future,
+    path::Path,
+    result,
+    sync::Arc,
+    time::{Duration, SystemTime},
+};
 
 use anyhow::{Context, Result};
 use pkarr::{SignedPacket, Timestamp};
@@ -52,7 +58,7 @@ enum Message {
         res: oneshot::Sender<Snapshot>,
     },
     CheckExpired {
-        time: [u8; 8],
+        time: Timestamp,
         key: PublicKeyBytes,
     },
 }
@@ -119,7 +125,6 @@ impl Actor {
             let transaction = self.db.begin_write()?;
             let mut tables = Tables::new(&transaction)?;
             let timeout = tokio::time::sleep(self.options.max_batch_time);
-            let expired = Timestamp::now() - expiry_us;
             tokio::pin!(timeout);
             for _ in 0..self.options.max_batch_size {
                 tokio::select! {
@@ -151,8 +156,8 @@ impl Actor {
                                         res.send(false).ok();
                                         continue;
                                     } else {
-                                        // remove the packet from the update time index
-                                        tables.update_time.remove(&packet.timestamp().to_bytes(), key.as_bytes())?;
+                                        // remove the old packet from the update time index
+                                        tables.update_time.remove(&existing.timestamp().to_bytes(), key.as_bytes())?;
                                         true
                                     }
                                 } else {
@@ -185,13 +190,21 @@ impl Actor {
                                 res.send(Snapshot::new(&self.db)?).ok();
                             }
                             Message::CheckExpired { key, time } => {
-                                trace!("check expired {} at {}", key, u64::from_be_bytes(time));
+                                trace!("check expired {} at {}", key, fmt_time(time));
                                 if let Some(packet) = get_packet(&tables.signed_packets, &key)? {
+                                    let expired = Timestamp::now() - expiry_us;
                                     if packet.timestamp() < expired {
-                                        tables.update_time.remove(&time, key.as_bytes())?;
+                                        tables.update_time.remove(&time.to_bytes(), key.as_bytes())?;
                                         let _ = tables.signed_packets.remove(key.as_bytes())?;
                                         self.metrics.store_packets_expired.inc();
+                                        debug!("removed expired packet {key}");
+                                    } else {
+                                        debug!("packet {key} is no longer expired, removing obsolete expiry entry");
+                                        tables.update_time.remove(&time.to_bytes(), key.as_bytes())?;
                                     }
+                                } else {
+                                    debug!("expired packet {key} not found, remove from expiry table");
+                                    tables.update_time.remove(&time.to_bytes(), key.as_bytes())?;
                                 }
                             }
                         }
@@ -205,6 +218,10 @@ impl Actor {
     }
 }
 
+fn fmt_time(t: Timestamp) -> String {
+    humantime::format_rfc3339_micros(SystemTime::from(t)).to_string()
+}
+
 /// A struct similar to [`redb::Table`] but for all tables that make up the
 /// signed packet store.
 pub(super) struct Tables<'a> {
@@ -368,7 +385,7 @@ async fn evict_task_inner(send: mpsc::Sender<Message>, options: Options) -> anyh
             anyhow::bail!("failed to get snapshot");
         };
         let expired = Timestamp::now() - expiry_us;
-        trace!("evicting packets older than {}", expired);
+        trace!("evicting packets older than {}", fmt_time(expired));
         // if getting the range fails we exit the loop and shut down
         // if individual reads fail we log the error and limp on
         for item in snapshot.update_time.range(..expired.to_bytes())? {
@@ -379,26 +396,22 @@ async fn evict_task_inner(send: mpsc::Sender<Message>, options: Options) -> anyh
                     continue;
                 }
             };
-            let time = time.value();
-            trace!("evicting expired packets at {}", u64::from_be_bytes(time));
+            let time = Timestamp::from(time.value());
+            trace!("evicting expired packets at {}", fmt_time(time));
             for item in keys {
                 let key = match item {
                     Ok(v) => v,
                     Err(e) => {
                         error!(
                             "failed to read update_time item at {}: {:?}",
-                            u64::from_be_bytes(time),
+                            fmt_time(time),
                             e
                         );
                         continue;
                     }
                 };
                 let key = PublicKeyBytes::new(key.value());
-                debug!(
-                    "evicting expired packet {} {}",
-                    u64::from_be_bytes(time),
-                    key
-                );
+                debug!("evicting expired packet {} {}", fmt_time(time), key);
                 send.send(Message::CheckExpired { time, key }).await?;
             }
         }