mocklogger/

logger.rs

// Author: D.S. Ljungmark <spider@skuggor.se>, Modio AB
// SPDX-License-Identifier: AGPL-3.0-or-later
#![allow(clippy::module_name_repetitions)]
mod builder;
mod errors;
pub(crate) mod keys;
mod ping;
use super::timefail;
pub use builder::Builder;
pub use errors::LogErr;
use fsipc::legacy::{PreparedPoint, Transaction};
use modio_logger_db::{fxtime, Datastore};
pub use ping::LoggerPing;
use tracing::{debug, error, info, instrument, warn};
use zbus::{interface, SignalContext};

#[derive(Clone)]
pub struct Logger {
    ds: Datastore,
    timefail: timefail::Timefail,
}

impl Logger {
    pub async fn new(timefail: timefail::Timefail, ds: Datastore) -> Result<Self, LogErr> {
        if timefail.is_timefail() {
            info!("Failing all pending change requests due to TIMEFAIL");
            ds.transaction_fail_pending()
                .await
                .inspect_err(|e| error!("Failed to remove pending transactions: {:?}", e))?;
        }
        ds.fail_queued_transactions()
            .await
            .inspect_err(|e| error!("Failed to remove queued transactions: {:?}", e))?;
        Ok(Self { ds, timefail })
    }

    #[instrument(skip(self))]
    pub async fn periodic(&self) -> Result<(), LogErr> {
        debug!("Doing periodic tasks, persistance, timefail");
        if self.ds.should_persist().await? {
            self.ds
                .persist_data()
                .await
                .inspect_err(|e| error!("Failed to persist datastore: {:?}", e))?;
        };
        if self.timefail.is_adjust() {
            if let Some(adjust) = self.timefail.get_adjust().await {
                info!("Time jump has happened, adjusting data with {adjust}");
                let count = self.ds.fix_timefail(adjust).await?;
                if count > 0 {
                    info!("Adjusted timestamps of {count} metrics");
                };
                self.timefail
                    .remove_adjust()
                    .await
                    .inspect_err(|e| error!("Failed to remove timefail file. err={:?}", e))?;
            }
        }
        let stats = self
            .ds
            .get_statistics()
            .await
            .inspect_err(|e| error!("Failed to gather DB statistics: {:?}", e))?;
        info!("Current stats: {stats}");
        Ok(())
    }
    #[cfg(test)]
    pub const fn ds(&self) -> &Datastore {
        &self.ds
    }

    #[must_use]
    pub const fn builder() -> Builder {
        Builder::new()
    }
}

async fn get_mac() -> String {
    use async_std::fs;
    let wan = std::path::Path::new("/sys/class/net/wan/address");
    let mut res = fs::read_to_string(&wan)
        .await
        .unwrap_or_else(|_| String::from("00:00:00:00:00:00"));
    // Lowercase all ascii-chars ( leaving bytes with high bit untouched)
    res.make_ascii_lowercase();
    // Keep only 0-9 a-f
    res.retain(|c| matches!(c, '0'..='9' | 'a'..='f'));
    res
}

#[allow(clippy::use_self)]
#[interface(name = "se.modio.logger.fsipc")]
impl Logger {
    #[allow(clippy::unused_self)]
    const fn ping(&self) -> &str {
        "Ping? Pong"
    }
    #[allow(clippy::unused_self)]
    fn valid_key(&mut self, key: &str) -> bool {
        keys::valid_key(key).is_ok()
    }

    #[allow(clippy::unused_self)]
    async fn get_boxid(&self) -> String {
        get_mac().await
    }

    #[instrument(skip(self))]
    async fn retrieve(&mut self, key: &str) -> Result<(fsipc::legacy::Measure,), LogErr> {
        keys::valid_key(key)?;
        let dat = self
            .ds
            .get_last_datapoint(key)
            .await
            // Having no data for a point is a common case ( point did not exist ) and thus not
            // worth treating as an error.
            .inspect_err(|e| debug!("Failed to retrieve last datapoint. err={:?}", e))?;
        let val = fsipc::legacy::Measure::from(dat);
        Ok((val,))
    }

    #[instrument(skip(self))]
    async fn retrieve_all(&mut self) -> Result<Vec<fsipc::legacy::Measure>, LogErr> {
        let result = self
            .ds
            .get_latest_logdata()
            .await
            .inspect_err(|e| error!("Failed to get latest logdata. err={:?}", e))?;
        let res: Vec<fsipc::legacy::Measure> = result
            .into_iter()
            .map(fsipc::legacy::Measure::from)
            .collect();
        Ok(res)
    }

    /// Signal sent when a new value is stored
    #[zbus(signal)]
    async fn store_signal(
        ctxt: &SignalContext<'_>,
        key: &str,
        value: &str,
        when: u64,
    ) -> zbus::Result<()>;

    #[instrument(skip_all)]
    async fn store(
        &self,
        #[zbus(signal_context)] ctxt: SignalContext<'_>,
        key: &str,
        value: &str,
    ) -> Result<(), LogErr> {
        keys::valid_key(key)?;
        let when = fxtime();
        let timefail = self.timefail.is_timefail();
        self.ds
            .insert(key, value, when, timefail)
            .await
            .inspect_err(|e| error!("Failed to store single metric. err={:?}", e))?;

        #[allow(
            clippy::cast_precision_loss,
            clippy::cast_sign_loss,
            clippy::cast_possible_truncation
        )]
        let trunc_when = when as u64;

        if !key.starts_with("modio.") {
            Self::store_signal(&ctxt, key, value, trunc_when).await?;
        };
        Ok(())
    }

    #[instrument(skip_all)]
    async fn store_with_time(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalContext<'_>,
        key: &str,
        value: &str,
        when: u64,
    ) -> Result<(), LogErr> {
        keys::valid_key(key)?;
        // timefail is always false when storing with timestamp, as the timestamp is expected to be
        // correct from another source.
        let timefail = false;

        // if time wraps we have another problem.
        #[allow(clippy::cast_precision_loss)]
        let db_when = when as f64;

        self.ds
            .insert(key, value, db_when, timefail)
            .await
            .inspect_err(|e| error!("Failed to store single metric with timestamp. err={:?}", e))?;
        if !key.starts_with("modio.") {
            Self::store_signal(&ctxt, key, value, when).await?;
        };
        Ok(())
    }
    /// Signal sent when a new transaction is added
    #[zbus(signal)]
    async fn transaction_added(ctxt: &SignalContext<'_>, key: &str) -> zbus::Result<()>;

    #[instrument(skip_all)]
    async fn transaction_add(
        &mut self,
        #[zbus(signal_context)] ctxt: SignalContext<'_>,
        key: &str,
        expected: &str,
        target: &str,
        token: &str,
    ) -> Result<(), LogErr> {
        keys::valid_key(key)?;
        keys::valid_token(token)?;
        if self
            .ds
            .has_transaction(token)
            .await
            .inspect_err(|e| error!("Failed to check for transaction for key. err={:?}", e))?
        {
            warn!(
                "Duplicate transaction (key: {}, token: {}) Ignoring for backwards compatibility.",
                key, token
            );
            return Ok(());
        }
        self.ds
            .transaction_add(key, expected, target, token)
            .await
            .inspect_err(|e| error!("Failed to add transaction for key. err={:?}", e))?;
        Self::transaction_added(&ctxt, key).await?;
        Ok(())
    }

    #[instrument(skip(self))]
    async fn transaction_get(&mut self, prefix: &str) -> Result<Vec<Transaction>, LogErr> {
        debug!("Retrieving transactions beginning with {}", prefix);
        let res = self
            .ds
            .transaction_get(prefix)
            .await
            .inspect_err(|e| error!("Failed to get transaction for key. err={:?}", e))?;

        let res: Vec<Transaction> = res.into_iter().map(Transaction::from).collect();
        Ok(res)
    }

    #[instrument(skip(self))]
    async fn transaction_fail(&mut self, t_id: u64) -> Result<(), LogErr> {
        debug!("Marking transaction: t_id={}, failed", t_id);
        let timefail = self.timefail.is_timefail();

        // if it wraps, that is fine.
        #[allow(clippy::cast_possible_wrap)]
        let t_id = t_id as i64;
        let count = self
            .ds
            .transaction_fail(t_id, timefail)
            .await
            .inspect_err(|e| error!("Failed to mark transaction as failed. err={:?}", e))?;
        if count > 0 {
            Ok(())
        } else {
            Err(LogErr::TransactionNotFound)
        }
    }

    #[instrument(skip(self))]
    async fn transaction_pass(&mut self, t_id: u64) -> Result<(), LogErr> {
        debug!("Marking transaction: t_id={}, passed ", t_id);
        let timefail = self.timefail.is_timefail();
        // if it wraps, that is fine.
        #[allow(clippy::cast_possible_wrap)]
        let t_id = t_id as i64;
        let count = self
            .ds
            .transaction_pass(t_id, timefail)
            .await
            .inspect_err(|e| error!("Failed to mark transaction as passed. err={:?}", e))?;
        if count > 0 {
            Ok(())
        } else {
            Err(LogErr::TransactionNotFound)
        }
    }

    #[instrument(skip(self))]
    async fn prepare_datapoints(&mut self, maximum: u32) -> Result<Vec<PreparedPoint>, LogErr> {
        prepare_range_check(maximum)?;
        let data = self
            .ds
            .get_batch(maximum)
            .await
            .inspect_err(|e| error!("Failed to get batch of datapoints. err={:?}", e))?;
        let result: Vec<PreparedPoint> = data.into_iter().map(PreparedPoint::from).collect();
        Ok(result)
    }

    #[instrument(skip(self))]
    async fn prepare_modio_datapoints(
        &mut self,
        maximum: u32,
    ) -> Result<Vec<PreparedPoint>, LogErr> {
        prepare_range_check(maximum)?;
        let data = self
            .ds
            .get_internal_batch(maximum)
            .await
            .inspect_err(|e| error!("Failed to get_internal_batch. err={:?}", e))?;
        let result: Vec<PreparedPoint> = data.into_iter().map(PreparedPoint::from).collect();
        Ok(result)
    }

    #[instrument(skip_all)]
    async fn remove_prepared(&mut self, items: Vec<i64>) -> Result<(), LogErr> {
        prepare_remove_check(&items)?;
        self.ds
            .drop_batch(&items)
            .await
            .inspect_err(|e| error!("Failed to delete batch of submitted items. err={:?}", e))?;
        Ok(())
    }
}

/// Check of incoming values to be removed, Returns error on invalid items
fn prepare_remove_check(items: &[i64]) -> Result<(), PreparedError> {
    if items.is_empty() {
        return Err(PreparedError::Empty);
    }
    if items.iter().any(|x| *x < 0_i64) {
        return Err(PreparedError::InvalidIndex);
    }
    Ok(())
}

#[derive(thiserror::Error, Debug)]
pub enum PreparedError {
    #[error("Too few items")]
    TooSmall,
    #[error("Too many items")]
    TooMany,
    #[error("Empty item set")]
    Empty,
    #[error("Invalid index")]
    InvalidIndex,
}

// Converting From a Prepared Error to LogError (which is used by DBus layer)
impl From<PreparedError> for LogErr {
    fn from(e: PreparedError) -> Self {
        Self::InvalidPrepared(e.to_string())
    }
}

/// Range check helper for `prepare_datapoints` and `prepare_modio_datapoints`
///
/// Errors:
///    `PreparedError` with a specific error item.
const fn prepare_range_check(num: u32) -> Result<(), PreparedError> {
    match num {
        0 => Err(PreparedError::TooSmall),
        1..=250 => Ok(()),
        _ => Err(PreparedError::TooMany),
    }
}

#[cfg(test)]
pub mod tests {
    use super::*;

    use crate::testing::{Tempbase, TestServer};
    use fsipc::unixtime;
    use std::error::Error;
    use test_log::test;
    use timeout_macro::timeouttest;

    type TestResult = Result<(), Box<dyn Error>>;

    use futures_util::{FutureExt, StreamExt};

    #[test(timeouttest)]
    async fn ping_pong_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        let first = proxy.ping().await?;
        let second = proxy.ping().await?;
        assert_eq!(first, "Ping? Pong");
        assert_eq!(second, "Ping? Pong");
        Ok(())
    }

    #[test(timeouttest)]
    async fn done_gives_error_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        let res = proxy.done().await;
        assert!(res.is_err());
        Ok(())
    }

    #[test(timeouttest)]
    async fn store_retrieve() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        proxy.store("test.key", "abc123").await?;
        let m = proxy.retrieve("test.key").await?;
        assert_eq!(m.key, "test.key");
        assert_eq!(m.value, "abc123");
        Ok(())
    }

    #[test(timeouttest)]
    async fn store_buffer() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        proxy
            .store_with_time("test.key", "abc123", 1_494_602_107)
            .await?;
        proxy
            .store_with_time("test.key", "abc1234", 1_494_602_108)
            .await?;
        let m = proxy.retrieve("test.key").await?;
        assert_eq!(m.key, "test.key");
        assert_eq!(m.value, "abc1234");
        assert_eq!(m.timestamp, 1_494_602_108);
        Ok(())
    }

    #[test(timeouttest)]
    async fn retrieve_all_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        debug!("I have a party here");
        proxy
            .store_with_time("test.key", "abc123", 1_494_602_107)
            .await?;
        proxy
            .store_with_time("test.key", "abc1234", 1_494_602_108)
            .await?;
        proxy.store("test.key2", "abcdefg").await?;

        let all = proxy.retrieve_all().await?;
        assert_eq!(all.len(), 2);
        let m0 = all.first().expect("Should have value");
        assert_eq!(m0.key, "test.key");
        assert_eq!(m0.value, "abc1234");
        assert_eq!(m0.timestamp, 1_494_602_108);
        let m1 = all.get(1).expect("Should have value");
        assert_eq!(m1.key, "test.key2");
        assert_eq!(m1.value, "abcdefg");
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_adding_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        proxy
            .transaction_add("test.test.one", "first", "second", "012")
            .await?;
        proxy
            .transaction_add("dummy.test.one", "should not", "be present", "013")
            .await?;
        let transactions = proxy.transaction_get("test.test").await?;
        assert_eq!(transactions.len(), 1);
        let res = &transactions[0];
        assert_eq!(res.key, "test.test.one");
        assert_eq!(res.t_id, 1, "Transaction ID mismatch");
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_dupe_adding_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let proxy = server.proxy().await?;
        proxy
            .transaction_add("test.test.one", "first", "second", "1638290048")
            .await?;
        let res = proxy
            .transaction_add("test.test.one", "first", "second", "1638290048")
            .await;
        res.expect("duplicated tokens should not cause error");
        let transactions = proxy.transaction_get("test.test").await?;
        assert_eq!(transactions.len(), 1);
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_signal_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let logger = server.proxy().await?;
        let mut stream = logger.receive_transaction_added().await?;
        logger
            .transaction_add("test.test.transaction_signal", "first", "second", "012")
            .await?;

        let signal = stream.next().await.unwrap();
        let payload = signal.args()?;
        assert_eq!(payload.key, "test.test.transaction_signal");
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_passing_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let logger = server.proxy().await?;
        logger
            .transaction_add("test.test.one", "first", "second", "012")
            .await?;
        logger
            .transaction_add("test.test.two", "uno", "dos", "0113")
            .await?;
        let trans = logger.transaction_get("test.test").await?;
        logger.transaction_fail(trans[0].t_id).await?;
        logger.transaction_pass(trans[1].t_id).await?;
        logger
            .transaction_add("test.test.three", "etta", "tvåa", "0114")
            .await?;
        let transactions = logger.transaction_get("test.test").await?;
        assert_eq!(transactions.len(), 1);
        let res = &transactions[0];
        assert_eq!(res.key, "test.test.three");
        assert_eq!(res.t_id, 3, "Transaction id mismatch");
        Ok(())
    }

    #[test(timeouttest)]
    async fn retrieving_data_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        ipc.store("test.test.one", "first").await?;
        ipc.store("test.test.one", "second").await?;
        ipc.store("test.test.one", "third").await?;

        ipc.store("test.test.two", "1").await?;
        ipc.store("test.test.two", "2").await?;
        ipc.store("test.test.two", "3").await?;

        let res = ipc.retrieve_all().await?;
        for measure in &res {
            let data = ipc.retrieve(&measure.key).await?;
            assert_eq!(data.key, measure.key);
        }
        Ok(())
    }

    #[test(timeouttest)]
    async fn valid_key_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        assert!(ipc.valid_key("modio.software.development").await?);
        assert!(ipc.valid_key("abc").await?);
        assert!(ipc.valid_key("a.b.c").await?);
        assert!(ipc.valid_key("a_b.c").await?);
        Ok(())
    }

    #[test(timeouttest)]
    async fn invalid_key_test() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        assert!(!ipc.valid_key("modio..invalid").await?);
        assert!(!ipc.valid_key(".modio..invalid").await?);
        assert!(!ipc.valid_key("modio.invalid.").await?);
        assert!(!ipc.valid_key("modio. invalid").await?);
        assert!(!ipc.valid_key("modio.in valid").await?);
        assert!(!ipc.valid_key("modio.invalid ").await?);
        assert!(!ipc.valid_key(" modio.invalid").await?);
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_double() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let key = "test.test.one";
        let first = "first";
        let second = "second";
        let guid4 = zbus::Guid::generate();
        // Store the key
        ipc.store(key, first).await?;

        ipc.transaction_add(key, first, second, guid4.as_str())
            .await?;
        let transactions = ipc.transaction_get(key).await?;
        let first_transaction = transactions
            .first()
            .expect("Should have at least one transaction");
        let res = ipc.transaction_pass(first_transaction.t_id).await;
        assert!(res.is_ok());
        let res = ipc.transaction_pass(first_transaction.t_id).await;
        assert!(res.is_err());
        Ok(())
    }

    #[test(timeouttest)]
    async fn transaction_tests() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let key = "test.test.one";
        let first = "first";
        let mut our_value = "first";
        let second = "second";

        ipc.store(key, our_value).await?;
        let when = unixtime();

        ipc.store_with_time(key, our_value, when).await?;

        let guid1 = zbus::Guid::generate(); // Transactions "id" are either a UUID or an timestamp-as-string
        ipc.transaction_add(key, first, second, guid1.as_str())
            .await?;
        let guid2 = zbus::Guid::generate();
        ipc.transaction_add(key, first, second, guid2.as_str())
            .await?;
        let guid3 = zbus::Guid::generate();
        ipc.transaction_add(key, first, second, guid3.as_str())
            .await?;

        let transactions = ipc.transaction_get(key).await?;
        // The "id" here is an internal ID in the current logger instance, not the same as
        // our GUID ID we entered above.
        for trn in &transactions {
            if trn.key == key {
                if our_value == trn.expected {
                    our_value = &trn.target;
                    ipc.transaction_pass(trn.t_id).await?;
                } else {
                    ipc.transaction_fail(trn.t_id).await?;
                }
            }
            ipc.store(key, our_value).await?;
        }
        // Transaction  get should _succeed_ but return empty data.
        let res = ipc.transaction_get(key).await?;
        assert_eq!(res.len(), 0);

        Ok(())
    }

    #[test(timeouttest)]
    async fn no_modio_signals() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let mut stream = ipc.receive_store_signal().await?;
        // Send a transactiona
        ipc.store("test.test.test", "value").await?;
        let first = stream.next().await.unwrap();
        let payload = first.args()?;
        assert_eq!(payload.key, "test.test.test");

        // A store to a "modio." prefix key should not end up in a signal
        // Therefore we should have an empty result here.
        ipc.store("modio.test.test", "value").await?;
        let second = stream.next().now_or_never();
        assert!(second.is_none());
        Ok(())
    }

    #[test(timeouttest)]
    async fn submit_consume() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        for x in 0..5 {
            ipc.store_with_time("test.foo", &x.to_string(), unixtime() - 100)
                .await?;
        }
        let vals = ipc.prepare_datapoints(10).await?;
        assert_eq!(vals.len(), 5);

        let point = &vals[0];
        assert_eq!(point.key, "test.foo");
        assert_eq!(point.value, "0");

        let more = ipc.prepare_datapoints(10).await?;
        // Should be available still
        assert_eq!(more.len(), 5);

        let last = &vals[4];
        assert_eq!(last.key, "test.foo");
        assert_eq!(last.value, "4");

        let to_remove: Vec<_> = vals.iter().map(|m| m.id).collect();

        ipc.remove_prepared(to_remove).await?;
        let after = ipc.prepare_datapoints(30).await?;
        assert!(after.is_empty());
        Ok(())
    }

    #[test(timeouttest)]
    async fn submit_modio_consume() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        for x in 0..5 {
            ipc.store_with_time("test.foo", &x.to_string(), unixtime() - 100)
                .await?;
            ipc.store_with_time("modio.test.foo", &x.to_string(), unixtime() - 100)
                .await?;
        }

        let vals = ipc.prepare_modio_datapoints(20).await?;
        assert_eq!(vals.len(), 5);
        for point in &vals {
            assert_eq!(&point.key, "modio.test.foo");
        }

        let more = ipc.prepare_modio_datapoints(10).await?;
        // Should be available still
        assert_eq!(more.len(), 5);

        // Figure out how to remove the ID's here
        //
        let to_remove: Vec<i64> = vals.iter().map(|x| x.id).collect();
        ipc.remove_prepared(to_remove).await?;

        let modio_after = ipc.prepare_modio_datapoints(30).await?;
        assert!(modio_after.is_empty());

        let after = ipc.prepare_datapoints(30).await?;
        assert!(!after.is_empty());
        Ok(())
    }

    // Prepare datapoints should cause a write to disk only if the buffer is eitheer full or too
    // old, as for how the database considers things.
    #[test(timeouttest)]
    async fn test_get_batch_expiry() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        for x in 0..25 {
            ipc.store("test.foo", &x.to_string()).await?;
        }
        let vals = ipc.prepare_datapoints(10).await?;
        assert_eq!(
            vals.len(),
            0,
            "Data should not automatically be flushed to disk"
        );

        ipc.store_with_time("test.foo", "26", unixtime() - 200)
            .await?;
        let vals = ipc.prepare_datapoints(10).await?;
        assert_eq!(
            vals.len(),
            10,
            "Oldest data in buffer should cause a flush to disk"
        );
        Ok(())
    }

    // Prepare (internal) datapoints should cause a write to disk only if the buffer is either full
    // or too old, as for how the database considers things.
    #[test(timeouttest)]
    async fn test_internal_get_batch_expiry() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        for x in 0..25 {
            ipc.store("modio.test.foo", &x.to_string()).await?;
            ipc.store("test.foo", &x.to_string()).await?;
        }
        let vals = ipc.prepare_modio_datapoints(10).await?;
        assert_eq!(
            vals.len(),
            0,
            "Data should not automatically be flushed to disk"
        );

        ipc.store_with_time("modio.test.foo", "26", unixtime() - 200)
            .await?;
        let vals = ipc.prepare_modio_datapoints(10).await?;
        assert_eq!(
            vals.len(),
            10,
            "Oldest data in buffer should cause a flush to disk"
        );
        Ok(())
    }

    #[test(timeouttest)]
    async fn test_get_batch() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        for x in 0..25 {
            ipc.store_with_time("test.foo", &x.to_string(), unixtime() - 200)
                .await?;
        }

        let vals = ipc.prepare_datapoints(10).await?;
        assert_eq!(vals.len(), 10);
        let point = &vals[0];
        assert_eq!(point.key, "test.foo");
        assert_eq!(point.value, "0");

        let to_remove: Vec<i64> = vals.iter().map(|x| x.id).collect();
        ipc.remove_prepared(to_remove).await?;

        // Requst more than we have left, should return only the 25-10=15 items
        let vals = ipc.prepare_datapoints(30).await?;
        assert_eq!(vals.len(), 15);
        let point = &vals[0];
        assert_eq!(point.key, "test.foo");
        assert_eq!(point.value, "10");

        let to_remove: Vec<i64> = vals.iter().map(|x| x.id).collect();
        ipc.remove_prepared(to_remove).await?;

        let vals = ipc.prepare_datapoints(30).await?;
        assert!(vals.is_empty());
        Ok(())
    }

    #[test(timeouttest)]
    /// Technically, this test might fail if you have an ethernet interface named "wan"
    /// I'm ok with that for now as that is unlikely to happen, if it does, disable the test, or
    /// check it as something better
    async fn test_get_mac() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let logger = server.proxy().await?;
        let res = logger.get_boxid().await?;
        assert_eq!(res, "000000000000");
        Ok(())
    }

    #[test(timeouttest)]
    async fn test_empty_transactions() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let transactions = ipc.transaction_get("test").await?;
        assert_eq!(transactions.len(), 0);
        let values = ipc.retrieve_all().await?;
        assert_eq!(values.len(), 0);
        Ok(())
    }

    #[test(timeouttest)]
    async fn test_transaction_get_prefix() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let transactions = ipc.transaction_get("mbus.").await?;
        assert_eq!(transactions.len(), 0);
        Ok(())
    }

    // This test is speshul.
    // It dupes the logic for the temp server in order to start it several times and compare
    // results between executions, in order to make sure that temp data (ram only) is accessible
    // later.
    #[test(timeouttest)]
    async fn resume_database_test() -> TestResult {
        let base = Tempbase::default();
        let dbfile = Tempbase::dbfile();
        let name = format!("se.modio.logger.TestResumeDb{}", line!());

        let expected = vec![
            ("test.test.one", "1"),
            ("test.test.two", "2"),
            ("test.test.three", "3"),
            ("test.test.four", "4"),
            ("test.test.five", "5"),
            ("test.test.six", "6"),
        ];

        // Spawn server, store some data and then shut it down
        {
            let name = name.clone();
            let base = base.clone();
            let dbfile = dbfile.clone();
            let server = TestServer::new_with_base(name, base, dbfile).await?;
            let logger = server.proxy().await?;
            info!("Filling datastore");
            for (key, val) in &expected {
                logger.store(key, val).await?;
            }
            // Read the data to ensure we have it in storage
            for (key, value) in &expected {
                let data = logger.retrieve(key).await?;
                assert_eq!(&data.value, value);
                assert_eq!(&data.key, key);
            }
            info!("Done, settling things");
        };

        // Spawn another server, retrieving the previous data and ensure it is the same.
        {
            let server = TestServer::new_with_base(name, base, dbfile).await?;
            let logger = server.proxy().await?;
            info!("Checking content after re-start");
            for (key, value) in &expected {
                let data = logger.retrieve(key).await?;
                assert_eq!(&data.value, value);
                assert_eq!(&data.key, key);
            }
        }
        Ok(())
    }

    // This is a big test-case that covers multiple concurrent reader-writers while running
    // maintenance jobs at the same time.
    // The test triggers load problems with the system if there are too few DB connections, and
    // more.
    #[allow(clippy::assertions_on_constants)]
    #[test(timeouttest)]
    async fn load_and_maintain() -> TestResult {
        let server = TestServer::new(line!()).await?;
        let ipc = server.proxy().await?;
        let logger1 = server.logger1().await?;
        let submit1 = server.submit1().await?;
        use crate::testing::proxytest::{Logger1Proxy, Submit1Proxy};
        use crate::LOGGER_PATH;
        use async_std::task::{sleep, spawn};
        use fsipc::legacy::fsipcProxy;
        use futures::try_join;
        use std::collections::HashMap;
        use std::sync::atomic::{AtomicBool, Ordering};
        use std::sync::Arc;
        use std::time::Duration;

        // Atomic flag that the tasks check before writing, prevents them from causing trouble once
        // we reach shutdown
        let stop = Arc::new(AtomicBool::new(false));
        // Amount of keys to have traffic for that are using "one write per key" mode.
        const NUM_SLOW_KEYS: usize = 1000;
        const NUM_TRANSACTIONS: usize = 50;
        assert!(
            NUM_TRANSACTIONS < NUM_SLOW_KEYS,
            "Transactions work on slow keys."
        );
        const NUM_BULK_KEYS: usize = 2000;
        // How many items the submitter will consume each loop
        const NUM_SUBMIT_DATA: usize = 200;

        // How many loops to run for each of maintenance / clean until we are done
        // The test will end after either of these reach their limit, or something fails.
        const NUM_CLEAN_LOOPS: usize = 50;
        const NUM_MAINT_LOOPS: usize = 50;

        // 1 task to generate "trickle" of data, using ipc.store with delays.
        //  ( also sets metadata for each iteration )
        async fn trickle_gen(
            stop: Arc<AtomicBool>,
            ipc: fsipcProxy<'static>,
            logger1: Logger1Proxy<'static>,
        ) -> zbus::Result<()> {
            loop {
                for n in 0..NUM_SLOW_KEYS {
                    let key = format!("test.trickle.key.{n}");
                    let name = format!("Name of tricke test key #{n}");
                    let desc = format!("Description {n} of trickle test key #{n}");
                    let val = format!("{n}");
                    logger1.set_metadata_name(&key, &name).await?;
                    logger1.set_metadata_description(&key, &desc).await?;
                    ipc.store(&key, &val).await?;
                }
                if stop.load(Ordering::Relaxed) {
                    break Ok(());
                }
            }
        }
        // 1 task to generate "bulk" of data, using logger1.store func
        async fn bulk_gen(
            stop: Arc<AtomicBool>,
            logger1: Logger1Proxy<'static>,
        ) -> zbus::Result<()> {
            let delay = Duration::from_millis(1);
            loop {
                let mut map = HashMap::with_capacity(NUM_BULK_KEYS);
                for n in 0..=NUM_BULK_KEYS {
                    let key = format!("test.bulk.key.{n}");
                    let val = format!("{n}");
                    map.insert(key, val.into());
                }
                logger1.store_batch(map).await?;
                sleep(delay).await;
                if stop.load(Ordering::Relaxed) {
                    break Ok(());
                }
            }
        }

        // 1 task to pretend to submit, calling "get all metadata" and then "prepare batch"
        async fn submit_gen(
            stop: Arc<AtomicBool>,
            ipc: fsipcProxy<'static>,
            submit1: Submit1Proxy<'static>,
        ) -> zbus::Result<()> {
            let delay = Duration::from_millis(5);
            loop {
                if stop.load(Ordering::Relaxed) {
                    break Ok(());
                }
                let _meta = submit1.get_all_metadata().await?;
                let dp = ipc.prepare_datapoints(NUM_SUBMIT_DATA as u32).await?;
                let to_remove: Vec<_> = dp.iter().map(|m| m.id).collect();
                if to_remove.is_empty() {
                    sleep(delay).await;
                } else {
                    ipc.remove_prepared(to_remove).await?;
                }
            }
        }

        // 1 task to poll keys and add transactions
        async fn tran_gen(stop: Arc<AtomicBool>, ipc: fsipcProxy<'static>) -> zbus::Result<()> {
            let delay = Duration::from_millis(5);
            let mut t_id = 0;
            loop {
                // Sleep first to give the trickle code a chance to add measures. Doing it that way
                // cuts down on errors.
                sleep(delay).await;
                if stop.load(Ordering::Relaxed) {
                    break Ok(());
                }
                for n in 0..=NUM_TRANSACTIONS {
                    t_id += 1;
                    let transaction_id = format!("{t_id}");
                    let key = format!("test.trickle.key.{n}");
                    // No datapoint exists yet, that is not a problem for this code
                    if let Ok(val) = ipc.retrieve(&key).await {
                        ipc.transaction_add(&val.key, &val.value, "0", &transaction_id)
                            .await?;
                    }
                }
            }
        }

        async fn maint_task(stop: Arc<AtomicBool>, conn: zbus::Connection) -> zbus::Result<()> {
            let delay = Duration::from_millis(5);
            use crate::LOGGER_PATH;
            let iface_ref = conn
                .object_server()
                .interface::<_, Logger>(LOGGER_PATH)
                .await?;
            for _ in 0..=NUM_MAINT_LOOPS {
                sleep(delay).await;
                if stop.load(Ordering::Relaxed) {
                    return Ok(());
                }
                iface_ref
                    .get_mut()
                    .await
                    .periodic()
                    .await
                    // Since the proxies return a zbus::Error<()> and we want to use try_join below, we
                    // need to return the same kind of error, and I don't feel like implementing the
                    // conversions just for some test-cases.
                    .map_err(|e| zbus::Error::Failure(e.to_string()))?;
            }
            stop.store(true, Ordering::Relaxed);
            Ok(())
        }

        async fn clean_task(stop: Arc<AtomicBool>, conn: zbus::Connection) -> zbus::Result<()> {
            let delay = Duration::from_millis(5);
            use crate::LOGGER_PATH;
            let iface_ref = conn
                .object_server()
                .interface::<_, Logger>(LOGGER_PATH)
                .await?;
            let pool = iface_ref.get_mut().await.ds().pool();
            for _ in 0..=NUM_CLEAN_LOOPS {
                sleep(delay).await;
                if stop.load(Ordering::Relaxed) {
                    return Ok(());
                }
                Datastore::clean_maintenance(pool.clone())
                    .await
                    .map_err(|e| zbus::Error::Failure(e.to_string()))?;
            }
            stop.store(true, Ordering::Relaxed);
            Ok(())
        }

        let trickle = spawn(trickle_gen(stop.clone(), ipc.clone(), logger1.clone()));
        let bulk = spawn(bulk_gen(stop.clone(), logger1.clone()));
        let submit = spawn(submit_gen(stop.clone(), ipc.clone(), submit1.clone()));
        let trans = spawn(tran_gen(stop.clone(), ipc.clone()));
        let maint = spawn(maint_task(stop.clone(), server.conn.clone()));
        let cleant = spawn(clean_task(stop.clone(), server.conn.clone()));
        let iface_ref = server
            .conn
            .object_server()
            .interface::<_, Logger>(LOGGER_PATH)
            .await?;
        /* Artificially  drain a few connetions from our pool. default pool is 4, so we steal 2
         * more. 1 for maintenance and one for the pool. */
        let _conn4 = iface_ref.get_mut().await.ds().pool().acquire().await?;
        let _conn5 = iface_ref.get_mut().await.ds().pool().acquire().await?;
        try_join!(trickle, bulk, submit, trans, maint, cleant)?;
        stop.store(true, Ordering::Relaxed);

        Ok(())
    }
}