node: Use millisecond precision for timestamps

Previously we used seconds, which could result in false positives
when checking for stale announcements, if things happened too quickly.

Cargo.lock

@@ -1355,9 +1355,9 @@ checksum = "f051f77a7c8e6957c0696eac88f26b0117e54f52d3fc682ab19397a8812846a4"
 
 [[package]]
 name = "localtime"
-version = "1.0.0"
+version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b58b1a2626b0920eec1d591e94c4feb3d1353695faad25c38f2a3c3efaf3db19"
+checksum = "f752d9826e5de9509879546e12b57259942fcbf2356bc2006f7da5f3fb2be587"
 
 [[package]]
 name = "log"

radicle-node/Cargo.toml

@@ -23,7 +23,7 @@ io-reactor = { version = "0.1.0", features = ["popol"] }
 lexopt = { version = "0.2.1" }
 libc = { version = "0.2.137" }
 log = { version = "0.4.17", features = ["std"] }
-localtime = { version = "1" }
+localtime = { version = "1.1.0" }
 netservices = { version = "0.1.0", features = ["io-reactor", "socket2"] }
 nonempty = { version = "0.8.1", features = ["serialize"] }
 qcheck = { version = "1", default-features = false, optional = true }

radicle-node/src/address/store.rs

@@ -330,7 +330,7 @@ mod test {
         let alice = arbitrary::gen::<NodeId>(1);
         let mut cache = Book::memory().unwrap();
         let features = node::Features::SEED;
-        let timestamp = LocalTime::now().as_secs();
+        let timestamp = LocalTime::now().as_millis();
 
         let ka = KnownAddress {
             addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(),
@@ -356,7 +356,7 @@ mod test {
         let alice = arbitrary::gen::<NodeId>(1);
         let mut cache = Book::memory().unwrap();
         let features = node::Features::SEED;
-        let timestamp = LocalTime::now().as_secs();
+        let timestamp = LocalTime::now().as_millis();
 
         let ka = KnownAddress {
             addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(),
@@ -381,7 +381,7 @@ mod test {
     fn test_insert_and_update() {
         let alice = arbitrary::gen::<NodeId>(1);
         let mut cache = Book::memory().unwrap();
-        let timestamp = LocalTime::now().as_secs();
+        let timestamp = LocalTime::now().as_millis();
         let features = node::Features::SEED;
         let ka = KnownAddress {
             addr: net::SocketAddr::from(([4, 4, 4, 4], 8776)).into(),
@@ -431,7 +431,7 @@ mod test {
         let alice = arbitrary::gen::<NodeId>(1);
         let bob = arbitrary::gen::<NodeId>(1);
         let mut cache = Book::memory().unwrap();
-        let timestamp = LocalTime::now().as_secs();
+        let timestamp = LocalTime::now().as_millis();
         let features = node::Features::SEED;
 
         for addr in [
@@ -469,7 +469,7 @@ mod test {
         let rng = fastrand::Rng::new();
         let mut cache = Book::memory().unwrap();
         let mut expected = Vec::new();
-        let timestamp = LocalTime::now().as_secs();
+        let timestamp = LocalTime::now().as_millis();
         let features = node::Features::SEED;
 
         for id in ids {

radicle-node/src/clock.rs

@@ -1,2 +1,2 @@
-/// Seconds since epoch.
+/// Milliseconds since epoch.
 pub type Timestamp = u64;

radicle-node/src/service.rs

@@ -347,7 +347,7 @@ where
     }
 
     pub fn initialize(&mut self, time: LocalTime) -> Result<(), Error> {
-        debug!(target: "service", "Init @{}", time.as_secs());
+        debug!(target: "service", "Init @{}", time.as_millis());
 
         self.start_time = time;
 
@@ -358,7 +358,7 @@ where
         }
         // Ensure that our inventory is recorded in our routing table.
         for id in self.storage.inventory()? {
-            self.routing.insert(id, self.node_id(), time.as_secs())?;
+            self.routing.insert(id, self.node_id(), time.as_millis())?;
         }
         // Setup subscription filter for tracked repos.
         self.filter = Filter::new(
@@ -632,7 +632,7 @@ where
                 self.reactor.write_all(
                     remote,
                     gossip::handshake(
-                        self.clock.as_secs(),
+                        self.clock.as_millis(),
                         &self.storage,
                         &self.signer,
                         filter,
@@ -745,7 +745,7 @@ where
         let peer = self.nodes.entry(*announcer).or_insert_with(Node::default);
 
         // Don't allow messages from too far in the future.
-        if timestamp.saturating_sub(now.as_secs()) > MAX_TIME_DELTA.as_secs() {
+        if timestamp.saturating_sub(now.as_millis()) > MAX_TIME_DELTA.as_millis() as u64 {
             return Err(session::Error::InvalidTimestamp(timestamp));
         }
 
@@ -754,7 +754,7 @@ where
                 // Discard inventory messages we've already seen, otherwise update
                 // out last seen time.
                 if !peer.inventory_announced(timestamp) {
-                    debug!(target: "service", "Ignoring stale inventory announcement from {announcer} (t={})", self.clock.as_secs());
+                    debug!(target: "service", "Ignoring stale inventory announcement from {announcer} (t={})", self.clock.as_millis());
                     return Ok(false);
                 }
 
@@ -940,7 +940,7 @@ where
                     self.reactor.write_all(
                         peer.id,
                         gossip::handshake(
-                            self.clock.as_secs(),
+                            self.clock.as_millis(),
                             &self.storage,
                             &self.signer,
                             filter,
@@ -1050,7 +1050,7 @@ where
     /// Sync, and if needed, announce our local inventory.
     fn sync_and_announce_inventory(&mut self) -> Result<bool, Error> {
         let inventory = self.storage.inventory()?;
-        let updated = self.sync_routing(&inventory, self.node_id(), self.clock.as_secs())?;
+        let updated = self.sync_routing(&inventory, self.node_id(), self.clock.as_millis())?;
 
         if updated {
             self.announce_inventory(inventory)?;
@@ -1104,7 +1104,7 @@ where
         let repo = self.storage.repository(id)?;
         let remote = repo.remote(&node)?;
         let peers = self.sessions.negotiated().map(|(_, p)| p);
-        let timestamp = self.clock.as_secs();
+        let timestamp = self.clock.as_millis();
 
         if remote.refs.len() > Refs::max() {
             error!(
@@ -1152,7 +1152,7 @@ where
 
     /// Announce our inventory to all connected peers.
     fn announce_inventory(&mut self, inventory: Vec<Id>) -> Result<(), storage::Error> {
-        let time = self.clock.as_secs();
+        let time = self.clock.as_millis();
         let inv = Message::inventory(gossip::inventory(time, inventory), &self.signer);
         for id in self.sessions.negotiated().map(|(id, _)| id) {
             self.reactor.write(*id, inv.clone());
@@ -1168,7 +1168,7 @@ where
 
         let delta = count - self.config.limits.routing_max_size;
         self.routing.prune(
-            (*now - self.config.limits.routing_max_age).as_secs(),
+            (*now - self.config.limits.routing_max_age).as_millis(),
             Some(delta),
         )?;
         Ok(())
@@ -1505,7 +1505,7 @@ mod gossip {
             Message::inventory(gossip::inventory(now, inventory), signer),
             Message::subscribe(
                 filter,
-                now - SUBSCRIBE_BACKLOG_DELTA.as_secs(),
+                now - SUBSCRIBE_BACKLOG_DELTA.as_millis() as u64,
                 Timestamp::MAX,
             ),
         ];

radicle-node/src/service/message.rs

@@ -440,7 +440,7 @@ mod tests {
         let msg: Message = AnnouncementMessage::from(RefsAnnouncement {
             id: arbitrary::gen(1),
             refs: bounded_refs,
-            timestamp: LocalTime::now().as_secs(),
+            timestamp: LocalTime::now().as_millis(),
         })
         .signed(&MockSigner::default())
         .into();
@@ -470,7 +470,7 @@ mod tests {
                 inventory: arbitrary::vec(INVENTORY_LIMIT)
                     .try_into()
                     .expect("size within bounds limit"),
-                timestamp: LocalTime::now().as_secs(),
+                timestamp: LocalTime::now().as_millis(),
             },
             &MockSigner::default(),
         );

radicle-node/src/service/routing.rs

@@ -312,7 +312,7 @@ mod test {
         let node = arbitrary::gen(1);
 
         for id in ids {
-            db.insert(id, node, LocalTime::now().as_secs()).unwrap();
+            db.insert(id, node, LocalTime::now().as_millis()).unwrap();
         }
 
         assert_eq!(10, db.len().unwrap(), "correct number of rows in table");
@@ -328,7 +328,7 @@ mod test {
 
         for id in &ids {
             for node in &nodes {
-                let time = rng.u64(..now.as_secs());
+                let time = rng.u64(..now.as_millis());
                 db.insert(*id, *node, time).unwrap();
             }
         }
@@ -338,18 +338,18 @@ mod test {
 
         for id in &ids {
             for node in &nodes {
-                let time = rng.u64(now.as_secs()..i64::MAX as u64);
+                let time = rng.u64(now.as_millis()..i64::MAX as u64);
                 db.insert(*id, *node, time).unwrap();
             }
         }
 
-        let pruned = db.prune(now.as_secs(), None).unwrap();
+        let pruned = db.prune(now.as_millis(), None).unwrap();
         assert_eq!(pruned, ids.len() * nodes.len());
 
         for id in &ids {
             for node in &nodes {
                 let t = db.entry(id, node).unwrap().unwrap();
-                assert!(t >= now.as_secs());
+                assert!(t >= now.as_millis());
             }
         }
     }

radicle-node/src/test/gossip.rs

@@ -27,7 +27,7 @@ pub fn messages(count: usize, now: LocalTime, delta: LocalDuration) -> Vec<Messa
         msgs.push(Message::inventory(
             InventoryAnnouncement {
                 inventory: arbitrary::vec(3).try_into().unwrap(),
-                timestamp: time.as_secs(),
+                timestamp: time.as_millis(),
             },
             &signer,
         ))

radicle-node/src/test/peer.rs

@@ -180,7 +180,7 @@ where
     }
 
     pub fn timestamp(&self) -> Timestamp {
-        self.clock().as_secs()
+        self.clock().as_millis()
     }
 
     pub fn git_url(&self, repo: Id, namespace: Option<RemoteId>) -> remote::Url {

radicle-node/src/tests.rs

@@ -47,7 +47,7 @@ use crate::{git, identity, rad, runtime, service, test};
 #[test]
 fn test_inventory_decode() {
     let inventory: Vec<Id> = arbitrary::gen(300);
-    let timestamp = LocalTime::now().as_secs();
+    let timestamp = LocalTime::now().as_millis();
 
     let mut buf = Vec::new();
     inventory.as_slice().encode(&mut buf).unwrap();
@@ -214,7 +214,7 @@ fn test_inventory_sync() {
     let bob_signer = MockSigner::default();
     let bob_storage = fixtures::storage(tmp.path().join("bob"), &bob_signer).unwrap();
     let bob = Peer::config("bob", [8, 8, 8, 8], bob_storage, peer::Config::default());
-    let now = LocalTime::now().as_secs();
+    let now = LocalTime::now().as_millis();
     let projs = bob.storage().inventory().unwrap();
 
     alice.connect_to(&bob);
@@ -319,7 +319,7 @@ fn test_inventory_pruning() {
                 Message::inventory(
                     InventoryAnnouncement {
                         inventory: test::arbitrary::vec::<Id>(num_projs).try_into().unwrap(),
-                        timestamp: bob.local_time().as_secs(),
+                        timestamp: bob.local_time().as_millis(),
                     },
                     &MockSigner::default(),
                 ),
@@ -359,7 +359,7 @@ fn test_tracking() {
 fn test_inventory_relay_bad_timestamp() {
     let mut alice = Peer::new("alice", [7, 7, 7, 7]);
     let bob = Peer::new("bob", [8, 8, 8, 8]);
-    let two_hours = 3600 * 2;
+    let two_hours = 3600 * 1000 * 2;
     let timestamp = alice.timestamp() + two_hours;
 
     alice.connect_to(&bob);
@@ -436,8 +436,8 @@ fn test_announcement_rebroadcast_timestamp_filtered() {
         eve.id(),
         Message::Subscribe(Subscribe {
             filter: Filter::default(),
-            since: alice.local_time().as_secs(),
+            since: alice.local_time().as_millis(),
-            until: (alice.local_time() + delta).as_secs(),
+            until: (alice.local_time() + delta).as_millis(),
         }),
     );
 
@@ -599,7 +599,7 @@ fn test_inventory_relay() {
     let bob = Peer::new("bob", [8, 8, 8, 8]);
     let eve = Peer::new("eve", [9, 9, 9, 9]);
     let inv = BoundedVec::try_from(arbitrary::vec(1)).unwrap();
-    let now = LocalTime::now().as_secs();
+    let now = LocalTime::now().as_millis();
 
     // Inventory from Bob relayed to Eve.
     alice.connect_to(&bob);