Revert "node: Use `std::time` for reactor and wire" — heartwood

This reverts commit e404f1038f461264f9395742ef74f5b710bef54a.

The commit above introduced the Epoch type to reduce the need for depending on the localtime crate. Eventually, node operators noticed that announcements would appear to be in the future and cause connection issues.

After more research, an issue¹ was found that describes how on some platforms, primarily Linux and MacOS, Instant::duration_since would return a shortened duration. This is the cause of this apparent drift in time.

¹

https://github.com/rust-lang/rust/issues/87906

 use std::io::ErrorKind;
 use std::sync::Arc;
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
 use std::time::Duration;
 use std::{io, thread};
 use crossbeam_channel::{unbounded, Receiver, TryRecvError};
 use localtime::LocalTime;
 use mio::event::{Event, Source};
 use mio::{Events, Interest, Poll, Waker};
 use thiserror::Error;

     type Transport: EventHandler + Source + Send + Debug + WriteAtomic;
     /// Method called by the reactor on the start of each event loop once the poll has returned.
     fn tick(&mut self, instant: Instant);
     fn tick(&mut self, time: localtime::LocalTime);
     /// Method called by the reactor when a previously set timeout is fired.
     ///

         &mut self,
         token: Token,
         reaction: <Self::Listener as EventHandler>::Reaction,
         instant: Instant,
         time: localtime::LocalTime,
     );
     /// Method called by the reactor upon a reaction to an I/O event on a transport resource.

         &mut self,
         token: Token,
         reaction: <Self::Transport as EventHandler>::Reaction,
         instant: Instant,
         time: localtime::LocalTime,
     );
     /// Method called by the reactor when a given resource was successfully registered

     fn run(mut self) {
         loop {
             let before_poll = Instant::now();
             let before_poll = LocalTime::now();
             let timeout = self
                 .timeouts
                 .next_expiring_from(before_poll)

             // Blocking
             let res = self.poll.poll(&mut events, Some(timeout));
             let tick = Instant::now();
             self.service.tick(tick);
             let now = LocalTime::now();
             self.service.tick(now);
             // The way this is currently used basically ignores which keys have
             // timed out. So as long as *something* timed out, we wake the service.
             let timers_fired = self.timeouts.remove_expired_by(tick);
             let timers_fired = self.timeouts.remove_expired_by(now);
             if timers_fired > 0 {
                 log::trace!(target: "reactor", "Timer has fired");
                 self.service.timer_reacted();

                 self.service.handle_error(Error::Poll(err));
             }
             let awoken = self.handle_events(tick, events);
             log::trace!(target: "reactor", "Duration between tick and events handled: {:?}", Instant::now().duration_since(tick));
             let awoken = self.handle_events(now, events);
             // Process the commands only if we awoken by the waker.
             if awoken {

                 }
             }
             self.handle_actions(tick);
             self.handle_actions(now);
         }
     }
     /// # Returns
     ///
     /// Whether one of the events was originated from the waker.
     fn handle_events(&mut self, instant: Instant, events: Events) -> bool {
     fn handle_events(&mut self, time: LocalTime, events: Events) -> bool {
         log::trace!(target: "reactor", "Handling events");
         let mut awoken = false;
         let mut deregistered = Vec::new();

                         .handle(event)
                         .into_iter()
                         .for_each(|service_event| {
                             self.service.listener_reacted(token, service_event, instant);
                             self.service.listener_reacted(token, service_event, time);
                         });
                 } else {
                     let listener = self.deregister_listener(token).unwrap_or_else(|| {

         awoken
     }
     fn handle_actions(&mut self, instant: Instant) {
     fn handle_actions(&mut self, time: LocalTime) {
         while let Some(action) = self.service.next() {
             log::trace!(target: "reactor", "Handling action {action} from the service");
             // Deadlock may happen here if the service will generate events over and over
             // in the handle_* calls we may never get out of this loop
             if let Err(err) = self.handle_action(action, instant) {
             if let Err(err) = self.handle_action(action, time) {
                 log::error!(target: "reactor", "Error: {err}");
                 self.service.handle_error(err);
             }

     fn handle_action(
         &mut self,
         action: Action<H::Listener, H::Transport>,
         instant: Instant,
         time: LocalTime,
     ) -> Result<(), Error<H::Listener, H::Transport>> {
         match action {
             Action::RegisterListener(token, mut listener) => {

             Action::SetTimer(duration) => {
                 log::trace!(target: "reactor", "Adding timer {duration:?} from now");
                 self.timeouts.set_timeout(duration, instant);
                 self.timeouts.set_timeout(duration, time);
             }
         }
         Ok(())

 use std::collections::BTreeSet;
 use std::time::Duration;
 use std::{collections::BTreeSet, time::Instant};
 use localtime::{LocalDuration, LocalTime};
 /// Manages timers and triggers timeouts.
 #[derive(Debug, Default)]
 pub struct Timer {
     /// Timeouts are durations since the UNIX epoch.
     timeouts: BTreeSet<Instant>,
     timeouts: BTreeSet<localtime::LocalTime>,
 }
 impl Timer {

     }
     /// Register a new timeout relative to a certain point in time.
     pub fn set_timeout(&mut self, timeout: Duration, after: Instant) {
         let time = after + timeout;
     pub fn set_timeout(&mut self, timeout: Duration, after: LocalTime) {
         let time = after + LocalDuration::from_millis(timeout.as_millis());
         self.timeouts.insert(time);
     }
     /// Get the first timeout expiring right at or after certain moment of time.
     /// Returns [`None`] if there are no timeouts.
     pub fn next_expiring_from(&self, time: impl Into<Instant>) -> Option<Duration> {
     pub fn next_expiring_from(&self, time: impl Into<LocalTime>) -> Option<Duration> {
         let time = time.into();
         let last = *self.timeouts.first()?;
         Some(if last >= time {
             last - time
             Duration::from_millis(last.as_millis() - time.as_millis())
         } else {
             Duration::default()
             Duration::from_secs(0)
         })
     }
     /// Removes timeouts which expire by a certain moment of time (inclusive),
     /// returning total number of timeouts which were removed.
     pub fn remove_expired_by(&mut self, instant: Instant) -> usize {
     pub fn remove_expired_by(&mut self, time: LocalTime) -> usize {
         // Since `split_off` returns everything *after* the given key, including the key,
         // if a timer is set for exactly the given time, it would remain in the "after"
         // set of unexpired keys. This isn't what we want, therefore we add `1` to the
         // given time value so that it is put in the "before" set that gets expired
         // and overwritten.
         let at = instant + Duration::from_millis(1);
         let at = time + LocalDuration::from_millis(1);
         let unexpired = self.timeouts.split_off(&at);
         let fired = self.timeouts.len();
         self.timeouts = unexpired;

     fn test_wake_exact() {
         let mut tm = Timer::new();
         let now = Instant::now();
         let now = LocalTime::now();
         tm.set_timeout(Duration::from_secs(8), now);
         tm.set_timeout(Duration::from_secs(9), now);
         tm.set_timeout(Duration::from_secs(10), now);
         assert_eq!(tm.remove_expired_by(now + Duration::from_secs(9)), 2);
         assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(9)), 2);
         assert_eq!(tm.count(), 1);
     }

         assert_eq!(tm.remove_expired_by(now), 0);
         assert_eq!(tm.count(), 4);
         assert_eq!(tm.remove_expired_by(now + Duration::from_secs(9)), 1);
         assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(9)), 1);
         assert_eq!(tm.count(), 3, "one timeout has expired");
         assert_eq!(tm.remove_expired_by(now + Duration::from_secs(66)), 2);
         assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(66)), 2);
         assert_eq!(tm.count(), 1, "another two timeouts have expired");
         assert_eq!(tm.remove_expired_by(now + Duration::from_secs(96)), 1);
         assert_eq!(tm.remove_expired_by(now + LocalDuration::from_secs(96)), 1);
         assert!(!tm.has_timeouts(), "all timeouts have expired");
     }

     fn test_next() {
         let mut tm = Timer::new();
         let mut now = Instant::now();
         let mut now = LocalTime::now();
         tm.set_timeout(Duration::from_secs(3), now);
         assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(3)));
         now += Duration::from_secs(2);
         now = now + LocalDuration::from_secs(2);
         assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(1)));
         now += Duration::from_secs(1);
         now = now + LocalDuration::from_secs(1);
         assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(0)));
         now += Duration::from_secs(1);
         now = now + LocalDuration::from_secs(1);
         assert_eq!(tm.next_expiring_from(now), Some(Duration::from_secs(0)));
         assert_eq!(tm.remove_expired_by(now), 1);

 use std::collections::VecDeque;
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::{Instant, SystemTime};
 use std::{io, net, time};
 use crossbeam_channel as chan;

 use cyphernet::encrypt::noise::{HandshakePattern, Keyset, NoiseState};
 use cyphernet::proxy::socks5;
 use cyphernet::{Digest, EcSk, Ecdh, Sha256};
 use localtime::LocalTime;
 use mio::net::TcpStream;
 use radicle::node::device::Device;

     }
 }
 /// The epoch time of when the node started.
 struct Epoch {
     /// The system time when the node started.
     started_time: SystemTime,
     /// The instant when the node started.
     started_at: Instant,
 }
 impl Epoch {
     /// Construct a new [`Epoch`].
     fn new(started_time: SystemTime, started_at: Instant) -> Self {
         Self {
             started_time,
             started_at,
         }
     }
     /// Construct an [`Epoch`] where both values are recorded using their
     /// equivalent `now` constructors.
     fn now() -> Self {
         Self::new(SystemTime::now(), Instant::now())
     }
     /// Get the elapsed [`SystemTime`] given a later [`Instant`].
     fn elapsed_time(&self, later: Instant) -> SystemTime {
         self.started_time + (later - self.started_at)
     }
 }
 /// Wire protocol implementation for a set of peers.
 pub(crate) struct Wire<D, S, G: crypto::signature::Signer<crypto::Signature> + Ecdh> {
     /// Backing service instance.

     peers: Peers,
     /// A (practically) infinite source of tokens to identify transports and listeners.
     tokens: Tokens,
     /// Record of system time and instant when the node started.
     epoch: Epoch,
 }
 impl<D, S, G> Wire<D, S, G>

             listening: RandomMap::default(),
             peers: Peers(RandomMap::default()),
             tokens: Tokens::default(),
             epoch: Epoch::now(),
         }
     }
     fn time(&self, instant: Instant) -> SystemTime {
         self.epoch.elapsed_time(instant)
     }
     pub fn listen(&mut self, socket: Listener) {
         let token = self.tokens.advance();
         self.listening.insert(token, socket.local_addr());

     type Listener = Listener;
     type Transport = Transport<WireSession<G>>;
     fn tick(&mut self, time: Instant) {
     fn tick(&mut self, time: LocalTime) {
         self.metrics.open_channels = self
             .peers
             .iter()

             })
             .sum();
         self.metrics.worker_queue_size = self.worker.len();
         self.service.tick(self.time(time).into(), &self.metrics);
         self.service.tick(
             LocalTime::from_millis(time.as_millis() as u128),
             &self.metrics,
         );
     }
     fn timer_reacted(&mut self) {

         &mut self,
         _: Token, // Note that this is the token of the listener socket.
         event: io::Result<(TcpStream, std::net::SocketAddr)>,
         _: Instant,
         _: LocalTime,
     ) {
         match event {
             Ok((connection, peer)) => {

         }
     }
     fn transport_reacted(&mut self, token: Token, event: SessionEvent<WireSession<G>>, _: Instant) {
     fn transport_reacted(
         &mut self,
         token: Token,
         event: SessionEvent<WireSession<G>>,
         _: LocalTime,
     ) {
         match event {
             SessionEvent::Established(ProtocolArtifact { state, session }) => {
                 // SAFETY: With the NoiseXK protocol, there is always a remote static key.