broke utils out into separate files

src/message.rs

@@ -1,9 +1,9 @@
 use anyhow::Context;
 use console::{style, Color};
+use serde::{Deserialize, Serialize};
 
 use crate::traits::FeroxSerialize;
 use crate::utils::fmt_err;
-use serde::{Deserialize, Serialize};
 
 #[derive(Serialize, Deserialize, Default)]
 /// Representation of a log entry, can be represented as a human readable string or JSON

src/scanner/ferox_scanner.rs

@@ -5,8 +5,6 @@ use futures::{stream, StreamExt};
 use lazy_static::lazy_static;
 use tokio::sync::Semaphore;
 
-use super::utils::Requester;
-
 use crate::{
     event_handlers::{
         Command::{AddError, AddToF64Field, SubtractFromUsizeField},
@@ -22,6 +20,8 @@ use crate::{
     utils::fmt_err,
 };
 
+use super::requester::Requester;
+
 lazy_static! {
     /// Vector of FeroxResponse objects
     pub static ref RESPONSES: FeroxResponses = FeroxResponses::default();

src/scanner/limit_heap.rs

@@ -0,0 +1,158 @@
+/// bespoke variation on an array-backed max-heap
+///
+/// 255 possible values generated from the initial requests/second
+///
+/// when no additional errors are encountered, the left child is taken (increasing req/sec)
+/// if errors have increased since the last interval, the right child is taken (decreasing req/sec)
+///
+/// formula for each child:
+/// - left: (|parent - current|) / 2 + current
+/// - right: current - ((|parent - current|) / 2)
+#[derive(Debug)]
+pub(super) struct LimitHeap {
+    /// backing array, 255 nodes == height of 7 ( 2^(h+1) -1 nodes )
+    pub(super) inner: [i32; 255],
+
+    /// original # of requests / second
+    pub(super) original: i32,
+
+    /// current position w/in the backing array
+    pub(super) current: usize,
+}
+
+/// default implementation of a LimitHeap
+impl Default for LimitHeap {
+    /// zero-initialize the backing array
+    fn default() -> Self {
+        Self {
+            inner: [0; 255],
+            original: 0,
+            current: 0,
+        }
+    }
+}
+
+/// implementation of a LimitHeap
+impl LimitHeap {
+    /// move to right child, return node's index from which the move was requested
+    pub(super) fn move_right(&mut self) -> usize {
+        if self.has_children() {
+            let tmp = self.current;
+            self.current = self.current * 2 + 2;
+            return tmp;
+        }
+        self.current
+    }
+
+    /// move to left child, return node's index from which the move was requested
+    pub(super) fn move_left(&mut self) -> usize {
+        if self.has_children() {
+            let tmp = self.current;
+            self.current = self.current * 2 + 1;
+            return tmp;
+        }
+        self.current
+    }
+
+    /// move to parent, return node's index from which the move was requested
+    pub(super) fn move_up(&mut self) -> usize {
+        if self.has_parent() {
+            let tmp = self.current;
+            self.current = (self.current - 1) / 2;
+            return tmp;
+        }
+        self.current
+    }
+
+    /// move directly to the given index
+    pub(super) fn move_to(&mut self, index: usize) {
+        self.current = index;
+    }
+
+    /// get the current node's value
+    pub(super) fn value(&self) -> i32 {
+        self.inner[self.current]
+    }
+
+    /// set the current node's value
+    pub(super) fn set_value(&mut self, value: i32) {
+        self.inner[self.current] = value;
+    }
+
+    /// check that this node has a parent (true for all except root)
+    pub(super) fn has_parent(&self) -> bool {
+        self.current > 0
+    }
+
+    /// get node's parent's value or self.original if at the root
+    pub(super) fn parent_value(&mut self) -> i32 {
+        if self.has_parent() {
+            let current = self.move_up();
+            let val = self.value();
+            self.move_to(current);
+            return val;
+        }
+        self.original
+    }
+
+    /// check if the current node has children
+    pub(super) fn has_children(&self) -> bool {
+        // inner structure is a complete tree, just check for the right child
+        self.current * 2 + 2 <= self.inner.len()
+    }
+
+    /// get current node's right child's value
+    fn right_child_value(&mut self) -> i32 {
+        let tmp = self.move_right();
+        let val = self.value();
+        self.move_to(tmp);
+        val
+    }
+
+    /// set current node's left child's value
+    fn set_left_child(&mut self) {
+        let parent = self.parent_value();
+        let current = self.value();
+        let value = ((parent - current).abs() / 2) + current;
+
+        self.move_left();
+        self.set_value(value);
+        self.move_up();
+    }
+
+    /// set current node's right child's value
+    fn set_right_child(&mut self) {
+        let parent = self.parent_value();
+        let current = self.value();
+        let value = current - ((parent - current).abs() / 2);
+
+        self.move_right();
+        self.set_value(value);
+        self.move_up();
+    }
+
+    /// iterate over the backing array, filling in each child's value based on the original value
+    pub(super) fn build(&mut self) {
+        // ex: original is 400
+        // arr[0] == 200
+        // arr[1] (left child) == 300
+        // arr[2] (right child) == 100
+        let root = self.original / 2;
+
+        self.inner[0] = root; // set root node to half of the original value
+        self.inner[1] = ((self.original - root).abs() / 2) + root;
+        self.inner[2] = root - ((self.original - root).abs() / 2);
+
+        // start with index 1 and fill in each child below that node
+        for i in 1..self.inner.len() {
+            self.move_to(i);
+
+            if self.has_children() && self.right_child_value() == 0 {
+                // this node has an unset child since the rchild is 0
+                self.set_left_child();
+                self.set_right_child();
+            }
+        }
+        self.move_to(0); // reset current index to the root of the tree
+    }
+}

src/scanner/mod.rs

@@ -1,9 +1,12 @@
-mod container;
+mod ferox_scanner;
 mod utils;
 mod init;
 #[cfg(test)]
 mod tests;
+mod limit_heap;
+mod policy_data;
+mod requester;
 
-pub use self::container::{FeroxScanner, RESPONSES};
+pub use self::ferox_scanner::{FeroxScanner, RESPONSES};
 pub use self::init::initialize;
 pub use self::utils::PolicyTrigger;

src/scanner/policy_data.rs

@@ -0,0 +1,309 @@
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+
+use crate::{atomic_load, atomic_store, config::RequesterPolicy};
+
+use super::limit_heap::LimitHeap;
+
+/// data regarding policy and metadata about last enforced trigger etc...
+#[derive(Default, Debug)]
+pub struct PolicyData {
+    /// how to handle exceptional cases such as too many errors / 403s / 429s etc
+    pub(super) policy: RequesterPolicy,
+
+    /// whether or not we're in the middle of a cooldown period
+    pub(super) cooling_down: AtomicBool,
+
+    /// length of time to pause tuning after making an adjustment
+    pub(super) wait_time: u64,
+
+    /// rate limit (at last interval)
+    limit: AtomicUsize,
+
+    /// number of errors (at last interval)
+    pub(super) errors: AtomicUsize,
+
+    /// whether or not the owning Requester should remove the rate_limiter, happens when a scan
+    /// has been limited and moves back up to the point of its original scan speed
+    pub(super) remove_limit: AtomicBool,
+
+    /// heap of values used for adjusting # of requests/second
+    pub(super) heap: std::sync::RwLock<LimitHeap>,
+}
+
+/// implementation of PolicyData
+impl PolicyData {
+    /// given a RequesterPolicy, create a new PolicyData
+    pub fn new(policy: RequesterPolicy, timeout: u64) -> Self {
+        // can use this as a tweak for how aggressively adjustments should be made when tuning
+        let wait_time = ((timeout as f64 / 2.0) * 1000.0) as u64;
+
+        Self {
+            policy,
+            wait_time,
+            ..Default::default()
+        }
+    }
+
+    /// setter for requests / second; populates the underlying heap with values from req/sec seed
+    pub(super) fn set_reqs_sec(&self, reqs_sec: usize) {
+        if let Ok(mut guard) = self.heap.write() {
+            guard.original = reqs_sec as i32;
+            guard.build();
+            self.set_limit(guard.inner[0] as usize); // set limit to 1/2 of current request rate
+        }
+    }
+
+    /// setter for errors
+    pub(super) fn set_errors(&self, errors: usize) {
+        atomic_store!(self.errors, errors);
+    }
+
+    /// setter for limit
+    fn set_limit(&self, limit: usize) {
+        atomic_store!(self.limit, limit);
+    }
+
+    /// getter for limit
+    pub(super) fn get_limit(&self) -> usize {
+        atomic_load!(self.limit)
+    }
+
+    /// adjust the rate of requests per second up (increase rate)
+    pub(super) fn adjust_up(&self, streak_counter: &usize) {
+        if let Ok(mut heap) = self.heap.try_write() {
+            if *streak_counter > 2 {
+                // streak of 3 upward moves in a row, traverse the tree upward instead of to a
+                // higher-valued branch lower in the tree
+                let current = heap.value();
+                heap.move_up();
+                heap.move_up();
+                if current > heap.value() {
+                    // the tree's structure makes it so that sometimes 2 moves up results in a
+                    // value greater than the current node's and other times we need to move 3 up
+                    // to arrive at a greater value
+                    if heap.has_parent() && heap.parent_value() > current {
+                        // all nodes except 0th node (root)
+                        heap.move_up();
+                    } else if !heap.has_parent() {
+                        // been here enough that we can try resuming the scan to its original
+                        // speed (no limiting at all)
+                        atomic_store!(self.remove_limit, true);
+                    }
+                }
+                self.set_limit(heap.value() as usize);
+            } else if heap.has_children() {
+                // streak not at 3, just check that we can move down, and do so
+                heap.move_left();
+                self.set_limit(heap.value() as usize);
+            } else {
+                // tree bottomed out, need to move back up the tree a bit
+                let current = heap.value();
+                heap.move_up();
+                heap.move_up();
+
+                if current > heap.value() {
+                    heap.move_up();
+                }
+
+                self.set_limit(heap.value() as usize);
+            }
+        }
+    }
+
+    /// adjust the rate of requests per second down (decrease rate)
+    pub(super) fn adjust_down(&self) {
+        if let Ok(mut heap) = self.heap.try_write() {
+            if heap.has_children() {
+                heap.move_right();
+                self.set_limit(heap.value() as usize);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    /// PolicyData builds and sets correct values for the inner heap when set_reqs_sec is called
+    fn set_reqs_sec_builds_heap_and_sets_initial_value() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        assert_eq!(pd.wait_time, 3500);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+        assert_eq!(pd.heap.read().unwrap().original, 400);
+        assert_eq!(pd.heap.read().unwrap().current, 0);
+        assert_eq!(pd.heap.read().unwrap().inner[0], 200);
+        assert_eq!(pd.heap.read().unwrap().inner[1], 300);
+        assert_eq!(pd.heap.read().unwrap().inner[2], 100);
+    }
+
+    #[test]
+    /// PolicyData setters/getters tests for code coverage / sanity
+    fn policy_data_getters_and_setters() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_errors(20);
+        assert_eq!(pd.errors.load(Ordering::Relaxed), 20);
+        pd.set_limit(200);
+        assert_eq!(pd.get_limit(), 200);
+    }
+
+    #[test]
+    /// PolicyData adjust_down sets the limit to the correct value
+    fn policy_data_adjust_down_simple() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+        pd.adjust_down();
+        assert_eq!(pd.get_limit(), 100);
+    }
+
+    #[test]
+    /// PolicyData adjust_down sets the limit to the correct value when no child nodes are present
+    fn policy_data_adjust_down_no_children() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+        let mut guard = pd.heap.write().unwrap();
+        guard.move_to(250);
+        guard.set_value(27);
+        pd.set_limit(guard.value() as usize);
+        drop(guard);
+
+        pd.adjust_down();
+        assert_eq!(pd.get_limit(), 27);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_simple() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+        pd.adjust_up(&0);
+        assert_eq!(pd.get_limit(), 300);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_streak_and_2_moves() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        // 2 moves
+        pd.heap.write().unwrap().move_to(9);
+        assert_eq!(pd.heap.read().unwrap().value(), 275);
+        pd.adjust_up(&3);
+        assert_eq!(pd.heap.read().unwrap().value(), 300);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 300);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), false);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_streak_and_2_moves_to_arrive_at_root() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        pd.heap.write().unwrap().move_to(4);
+        assert_eq!(pd.heap.read().unwrap().value(), 250);
+        pd.adjust_up(&3);
+        assert_eq!(pd.heap.read().unwrap().value(), 200);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 200);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), true);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_streak_and_2_moves_to_find_less_than_current() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        pd.heap.write().unwrap().move_to(15);
+        assert_eq!(pd.heap.read().unwrap().value(), 387);
+        pd.adjust_up(&3);
+        assert_eq!(pd.heap.read().unwrap().value(), 350);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 350);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), false);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_streak_and_3_moves() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        pd.heap.write().unwrap().move_to(19);
+        assert_eq!(pd.heap.read().unwrap().value(), 287);
+        pd.adjust_up(&3);
+        assert_eq!(pd.heap.read().unwrap().value(), 300);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 300);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), false);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_no_children_2_moves() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        pd.heap.write().unwrap().move_to(241);
+
+        assert_eq!(pd.heap.read().unwrap().value(), 41);
+        pd.adjust_up(&0);
+        assert_eq!(pd.heap.read().unwrap().value(), 43);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 43);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), false);
+    }
+
+    #[test]
+    /// PolicyData adjust_up sets the limit to the correct value
+    fn policy_data_adjust_up_with_no_children_3_moves() {
+        // original: 400
+        // [200, 300, 100, 350, 250, 150, 50, 375, 325, 275, 225, 175, 125, 75, 25, ...]
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+        assert_eq!(pd.get_limit(), 200);
+
+        pd.heap.write().unwrap().move_to(240);
+
+        assert_eq!(pd.heap.read().unwrap().value(), 45);
+        pd.adjust_up(&0);
+        assert_eq!(pd.heap.read().unwrap().value(), 37);
+        assert_eq!(pd.limit.load(Ordering::Relaxed), 37);
+        assert_eq!(pd.remove_limit.load(Ordering::Relaxed), false);
+    }
+
+    #[test]
+    /// hit some of the out of the way corners of limitheap for coverage
+    fn increase_limit_heap_coverage_by_hitting_edge_cases() {
+        let pd = PolicyData::new(RequesterPolicy::AutoBail, 7);
+        pd.set_reqs_sec(400);
+
+        println!("{:?}", pd.heap.read().unwrap()); // debug derivation
+
+        pd.heap.write().unwrap().move_to(240);
+        assert_eq!(pd.heap.write().unwrap().move_right(), 240);
+        assert_eq!(pd.heap.write().unwrap().move_left(), 240);
+
+        pd.heap.write().unwrap().move_to(0);
+        assert_eq!(pd.heap.write().unwrap().move_up(), 0);
+        assert_eq!(pd.heap.write().unwrap().parent_value(), 400);
+    }
+}

src/scanner/tests.rs

@@ -1,11 +1,14 @@
-use super::*;
+use std::sync::Arc;
+
+use tokio::sync::Semaphore;
+
 use crate::{
     config::OutputLevel,
     event_handlers::Handles,
     scan_manager::{FeroxScans, ScanOrder},
 };
-use std::sync::Arc;
-use tokio::sync::Semaphore;
+
+use super::*;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 #[should_panic]