node: Don't re-use timestamps — heartwood

There are rare cases where ref announcements may be sent in close succession, and the current clock time is re-used. This will cause the second (newer) announcement to be ignored by peers.

We add a timestamp method that checks if the timestamp is re-used, and increments it if so, ensuring that we always use a fresh one.

             node::Features::SEED,
             Alias::new("carol"),
 ,
             localtime::LocalTime::now().as_secs(),
             localtime::LocalTime::now().into(),
             [node::KnownAddress::new(
                 // Eve will fail to connect to this address.
                 node::Address::from(net::SocketAddr::from(([0, 0, 0, 0], 19873))),

         .unwrap();
     eve.db
         .routing_mut()
         .insert([&acme], carol, localtime::LocalTime::now().as_secs())
         .insert([&acme], carol, localtime::LocalTime::now().into())
         .unwrap();
     eve.config.peers = node::config::PeerConfig::Static;

     let bob = environment.node(Config::test(Alias::new("bob")));
     let mut eve = environment.node(Config::test(Alias::new("eve")));
     let acme = RepoId::from_str("z42hL2jL4XNk6K8oHQaSWfMgCL7ji").unwrap();
     let now = localtime::LocalTime::now().as_secs();
     let now = localtime::LocalTime::now().into();
     fixtures::repository(working.join("acme"));

                 // If our announcement was made some time ago, the timestamp on it will be old,
                 // and it might not get gossiped to new nodes since it will be purged from caches.
                 // Therefore, we make sure it's never too old.
                 if clock.as_millis() - ann.timestamp
                     <= config.limits.gossip_max_age.as_millis() as u64
                 {
                 if clock - ann.timestamp.to_local_time() <= config.limits.gossip_max_age {
                     Some(ann)
                 } else {
                     None

             );
             ann
         } else {
             service::gossip::node(&config, clock.as_millis())
             service::gossip::node(&config, clock.into())
                 .solve(Default::default())
                 .expect("Runtime::init: unable to solve proof-of-work puzzle")
         };

                     radicle::node::Features::SEED,
                     alias,
 ,
                     clock.as_millis(),
                     clock.into(),
                     [node::KnownAddress::new(addr, address::Source::Bootstrap)],
                 )?;
             }

     last_sync: LocalTime,
     /// Last time the service routing table was pruned.
     last_prune: LocalTime,
     /// Last time the service announced its inventory.
     /// Last time the inventory was announced.
     last_announce: LocalTime,
     /// Last timestamp used for announcements.
     last_timestamp: Timestamp,
     /// Time when the service was initialized, or `None` if it wasn't initialized.
     started_at: Option<LocalTime>,
     /// Publishes events to subscribers.

             last_idle: LocalTime::default(),
             last_sync: LocalTime::default(),
             last_prune: LocalTime::default(),
             last_timestamp: Timestamp::MIN,
             last_announce: LocalTime::default(),
             started_at: None,
             emitter,

         // all of it. It can happen that inventory is not properly seeded if for eg. the
         // user creates a new repository while the node is stopped.
         let rids = self.storage.inventory()?;
         self.db.routing_mut().insert(&rids, nid, time.as_millis())?;
         self.db.routing_mut().insert(&rids, nid, time.into())?;
         let announced = self
             .db

                 &rid,
                 &nid,
                 updated_at.oid,
                 updated_at.timestamp.as_millis(),
                 updated_at.timestamp.into(),
             )? {
                 debug!(target: "service", "Saved local sync status for {rid}..");
             }

             if let Err(err) = self
                 .db
                 .gossip_mut()
                 .prune((now - self.config.limits.gossip_max_age).as_millis())
                 .prune((now - self.config.limits.gossip_max_age).into())
             {
                 error!(target: "service", "Error pruning gossip entries: {err}");
             }

                 // Let all our peers know that we're interested in this repo from now on.
                 self.outbox.broadcast(
                     Message::subscribe(self.filter(), self.time(), Timestamp::MAX),
                     Message::subscribe(self.filter(), self.clock.into(), Timestamp::MAX),
                     self.sessions.connected().map(|(_, s)| s),
                 );
             }

                 info!(target: "service", "Fetched {rid} from {remote} successfully");
                 // Update our routing table in case this fetch was user-initiated and doesn't
                 // come from an announcement.
                 self.seed_discovered(rid, remote, self.time());
                 self.seed_discovered(rid, remote, self.clock.into());
                 for update in &updated {
                     if update.is_skipped() {

         self.emitter.emit(Event::PeerConnected { nid: remote });
         let msgs = self.initial(link);
         let now = self.time();
         if link.is_outbound() {
             if let Some(peer) = self.sessions.get_mut(&remote) {
                 peer.to_connected(self.clock);
                 self.outbox.write_all(peer, msgs);
                 if let Err(e) = self.db.addresses_mut().connected(&remote, &peer.addr, now) {
                 if let Err(e) =
                     self.db
                         .addresses_mut()
                         .connected(&remote, &peer.addr, self.clock.into())
                 {
                     error!(target: "service", "Error updating address book with connection: {e}");
                 }
             }

         match self.db.gossip_mut().announced(announcer, announcement) {
             Ok(fresh) => {
                 if !fresh {
                     debug!(target: "service", "Ignoring stale announcement from {announcer} (t={})", self.time());
                     debug!(target: "service", "Ignoring stale announcement from {announcer} (t={timestamp})");
                     return Ok(false);
                 }
             }

     }
     /// Set of initial messages to send to a peer.
     fn initial(&self, _link: Link) -> Vec<Message> {
         let filter = self.filter();
     fn initial(&mut self, _link: Link) -> Vec<Message> {
         let timestamp = self.timestamp();
         let now = self.clock();
         let filter = self.filter();
         let inventory = match self.storage.inventory() {
             Ok(i) => i,
             Err(e) => {

         // If this is our first connection to the network, we just ask for a fixed backlog
         // of messages to get us started.
         let since = match self.db.gossip().last() {
             Ok(Some(last)) => last - MAX_TIME_DELTA.as_millis() as Timestamp,
             Ok(None) => (*now - INITIAL_SUBSCRIBE_BACKLOG_DELTA).as_millis() as Timestamp,
             Ok(Some(last)) => Timestamp::from(last.to_local_time() - MAX_TIME_DELTA),
             Ok(None) => (*now - INITIAL_SUBSCRIBE_BACKLOG_DELTA).into(),
             Err(e) => {
                 error!(target: "service", "Error getting the lastest gossip message from storage: {e}");
                 return vec![];

         vec![
             Message::node(self.node.clone(), &self.signer),
             Message::inventory(gossip::inventory(now.as_millis(), inventory), &self.signer),
             Message::inventory(gossip::inventory(timestamp, inventory), &self.signer),
             Message::subscribe(filter, since, Timestamp::MAX),
         ]
     }

     /// Update our routing table with our local node's inventory.
     fn sync_inventory(&mut self) -> Result<SyncedRouting, Error> {
         let inventory = self.storage.inventory()?;
         let result = self.sync_routing(inventory, self.node_id(), self.time())?;
         let result = self.sync_routing(inventory, self.node_id(), self.clock.into())?;
         Ok(result)
     }

     /// Return a refs announcement including the given remotes.
     fn refs_announcement_for(
         &self,
         &mut self,
         rid: RepoId,
         remotes: impl IntoIterator<Item = NodeId>,
     ) -> Result<(Announcement, Vec<RefsAt>), Error> {
         let repo = self.storage.repository(rid)?;
         let timestamp = self.time();
         let timestamp = self.timestamp();
         let mut refs = BoundedVec::<_, REF_REMOTE_LIMIT>::new();
         for remote_id in remotes.into_iter() {

     /// Announce our own refs for the given repo.
     fn announce_own_refs(&mut self, rid: RepoId, doc: Doc<Verified>) -> Result<Vec<RefsAt>, Error> {
         let refs = self.announce_refs(rid, doc, [self.node_id()])?;
         let now = self.local_time();
         let (refs, timestamp) = self.announce_refs(rid, doc, [self.node_id()])?;
         // Update refs database with our signed refs branches.
         // This isn't strictly necessary for now, as we only use the database for fetches, and

             self.emitter.emit(Event::LocalRefsAnnounced {
                 rid,
                 refs: r,
                 timestamp: now.as_millis(),
                 timestamp,
             });
             if let Err(e) =
                 self.database_mut()
                     .refs_mut()
                     .set(&rid, &r.remote, &SIGREFS_BRANCH, r.at, now)
             {
             if let Err(e) = self.database_mut().refs_mut().set(
                 &rid,
                 &r.remote,
                 &SIGREFS_BRANCH,
                 r.at,
                 timestamp.to_local_time(),
             ) {
                 error!(
                     target: "service",
                     "Error updating refs database for `rad/sigrefs` of {} in {rid}: {e}",

         rid: RepoId,
         doc: Doc<Verified>,
         remotes: impl IntoIterator<Item = NodeId>,
     ) -> Result<Vec<RefsAt>, Error> {
         let peers = self.sessions.connected().map(|(_, p)| p);
     ) -> Result<(Vec<RefsAt>, Timestamp), Error> {
         let (ann, refs) = self.refs_announcement_for(rid, remotes)?;
         let timestamp = ann.timestamp();
         let peers = self.sessions.connected().map(|(_, p)| p);
         // Update our sync status for our own refs. This is useful for determining if refs were
         // updated while the node was stopped.

         if let Some(refs) = refs.iter().find(|r| r.remote == ann.node) {
             info!(
                 target: "service",
                 "Announcing own refs for {rid} to peers ({}) (t={})..",
                 refs.at, ann.timestamp()
                 "Announcing own refs for {rid} to peers ({}) (t={timestamp})..",
                 refs.at
             );
             if let Err(e) = self
                 .db
                 .seeds_mut()
                 .synced(&rid, &ann.node, refs.at, ann.timestamp())
                 .synced(&rid, &ann.node, refs.at, timestamp)
             {
                 error!(target: "service", "Error updating sync status for local node: {e}");
             }

             }),
             self.db.gossip_mut(),
         );
         Ok(refs)
         Ok((refs, timestamp))
     }
     fn sync_and_announce_inventory(&mut self) {

             return false;
         }
         let persistent = self.config.is_persistent(&nid);
         let time = self.time();
         let timestamp: Timestamp = self.clock.into();
         if let Err(e) = self.db.addresses_mut().attempted(&nid, &addr, time) {
         if let Err(e) = self.db.addresses_mut().attempted(&nid, &addr, timestamp) {
             error!(target: "service", "Error updating address book with connection attempt: {e}");
         }
         self.sessions.insert(

         }
     }
     /// Get the current time.
     fn time(&self) -> Timestamp {
         self.clock.as_millis()
     /// Get a timestamp for using in announcements.
     /// Never returns the same timestamp twice.
     fn timestamp(&mut self) -> Timestamp {
         let now = Timestamp::from(self.clock);
         if *now > *self.last_timestamp {
             self.last_timestamp = now;
         } else {
             self.last_timestamp = self.last_timestamp + 1;
         }
         self.last_timestamp
     }
     ////////////////////////////////////////////////////////////////////////////

     /// Announce our inventory to all connected peers.
     fn announce_inventory(&mut self, inventory: Inventory) -> Result<(), storage::Error> {
         let time = if self.clock > self.last_announce {
             self.clock.as_millis()
         } else if self.last_announce - self.clock < LocalDuration::from_secs(1) {
             // In rare cases where the inventory is updated very quickly, we want to make sure all
             // announcements carry an increasing timestamp. We allow our timestamps to be up to
             // one second in the future.
             self.last_announce.as_millis() + 1
         } else {
             // Announcement is considered redundant, ignore. Nb. This should not happen unless
             // you are trying to spam inventories.
             log::warn!(
                 target: "service",
                 "Ignored outgoing inventory announcement with {} items",
                 inventory.len()
             );
             return Ok(());
         };
         let time = self.timestamp();
         let msg = AnnouncementMessage::from(gossip::inventory(time, inventory));
         self.outbox.announce(

             self.sessions.connected().map(|(_, p)| p),
             self.db.gossip_mut(),
         );
         self.last_announce = LocalTime::from_millis(time as u128);
         self.last_announce = time.to_local_time();
         Ok(())
     }

         let delta = count - self.config.limits.routing_max_size;
         self.db.routing_mut().prune(
             (*now - self.config.limits.routing_max_age).as_millis(),
             (*now - self.config.limits.routing_max_age).into(),
             Some(delta),
         )?;
         Ok(())

 use std::num::TryFromIntError;
 use std::{fmt, io};
 use radicle::crypto::Signature;

     /// An Internal error.
     #[error("internal error: {0}")]
     Internal(#[from] sql::Error),
     /// Unit overflow.
     #[error("unit overflow:: {0}")]
     UnitOverflow(#[from] TryFromIntError),
 }
 /// A database that has access to historical gossip messages.

             .db
             .prepare("DELETE FROM `announcements` WHERE timestamp < ?1")?;
         stmt.bind((1, cutoff.try_into().unwrap_or(i64::MAX)))?;
         stmt.bind((1, &cutoff))?;
         stmt.next()?;
         Ok(self.db.change_count())

             .prepare("SELECT MAX(timestamp) AS latest FROM `announcements`")?;
         if let Some(Ok(row)) = stmt.into_iter().next() {
             let latest = row.try_read::<Option<i64>, _>(0)?;
             return Ok(latest.map(|l| l as Timestamp));
             return match row.try_read::<Option<i64>, _>(0)? {
                 Some(i) => Ok(Some(Timestamp::from(u64::try_from(i)?))),
                 None => Ok(None),
             };
         }
         Ok(None)
     }

             }
         }
         stmt.bind((5, &ann.signature))?;
         stmt.bind((6, ann.message.timestamp().try_into().unwrap_or(i64::MAX)))?;
         stmt.bind((6, &ann.message.timestamp()))?;
         stmt.next()?;
         Ok(self.db.change_count() > 0)