//! Net document storage backed by sqlite3.

//!

//! We store most objects in sqlite tables, except for very large ones,

//! which we store as "blob" files in a separate directory.

use super::ExpirationConfig;

use crate::docmeta::{AuthCertMeta, ConsensusMeta};

use crate::err::ReadOnlyStorageError;

use crate::storage::{InputString, Store};

use crate::{Error, Result};

use fs_mistrust::CheckedDir;

use tor_basic_utils::PathExt as _;

use tor_error::warn_report;

use tor_netdoc::doc::authcert::AuthCertKeyIds;

use tor_netdoc::doc::microdesc::MdDigest;

use tor_netdoc::doc::netstatus::{ConsensusFlavor, Lifetime};

#[cfg(feature = "routerdesc")]

use tor_netdoc::doc::routerdesc::RdDigest;

#[cfg(feature = "bridge-client")]

pub(crate) use {crate::storage::CachedBridgeDescriptor, tor_guardmgr::bridge::BridgeConfig};

use std::collections::HashMap;

use std::fs::OpenOptions;

use std::path::{Path, PathBuf};

use std::sync::Arc;

use std::time::SystemTime;

use rusqlite::{params, OpenFlags, OptionalExtension, Transaction};

use time::OffsetDateTime;

use tracing::{trace, warn};

/// Local directory cache using a Sqlite3 connection.

pub(crate) struct SqliteStore {

    /// Connection to the sqlite3 database.

    conn: rusqlite::Connection,

    /// Location for the sqlite3 database; used to reopen it.

    sql_path: Option<PathBuf>,

    /// Location to store blob files.

    blob_dir: CheckedDir,

    /// Lockfile to prevent concurrent write attempts from different

    /// processes.

    ///

    /// If this is None we aren't using a lockfile.  Watch out!

    ///

    /// (sqlite supports that with connection locking, but we want to

    /// be a little more coarse-grained here)

    lockfile: Option<fslock::LockFile>,

}

impl SqliteStore {

    /// Construct or open a new SqliteStore at some location on disk.

    /// The provided location must be a directory, or a possible

    /// location for a directory: the directory will be created if

    /// necessary.

    ///

    /// If readonly is true, the result will be a read-only store.

    /// Otherwise, when readonly is false, the result may be

    /// read-only or read-write, depending on whether we can acquire

    /// the lock.

    ///

    /// # Limitations:

    ///

    /// The file locking that we use to ensure that only one dirmgr is

    /// writing to a given storage directory at a time is currently

    /// _per process_. Therefore, you might get unexpected results if

    /// two SqliteStores are created in the same process with the

    /// path.

        } else {

        };

        // Check permissions on the sqlite and lock files; don't require them to

        // exist.

            {

            }

        }

        } else {

        };

    /// Construct a new SqliteStore from a database connection and a location

    /// for blob files.

    ///

    /// Used for testing with a memory-backed database.

    ///

    /// Note: `blob_dir` must not be used for anything other than storing the blobs associated with

    /// this database, since we will freely remove unreferenced files from this directory.

    #[cfg(test)]

    /// Construct a new SqliteStore from a database connection and a location

    /// for blob files.

    ///

    /// The `readonly` argument specifies whether the database connection should be read-only.

    /// Check whether this database has a schema format we can read, and

    /// install or upgrade the schema if necessary.

            }

            } else {

                // The other process should have created the database!

            }

            } else {

            }

    /// Read a blob from disk, mapping it if possible.

    ///

    /// Return `Ok(None)` if the file for the blob was not found on disk;

    /// returns an error in other cases.

            Err(fs_mistrust::Error::NotFound(_)) => {

                    path

                );

            }

        };

    /// Write a file to disk as a blob, and record it in the ExtDocs table.

    ///

    /// Return a SavedBlobHandle that describes where the blob is, and which

    /// can be used either to commit the blob or delete it.

    /// Save a blob to disk and commit it.

    #[cfg(test)]

        let SavedBlobHandle {

    /// Return the valid-after time for the latest non non-pending consensus,

    #[cfg(test)]

    // We should revise the tests to use latest_consensus_meta instead.

    /// Remove the blob with name `fname`, but do not give an error on failure.

    /// Delete any blob files that are old enough, and not mentioned in the ExtDocs table.

    ///

    /// There shouldn't actually be any, but we don't want to let our cache grow infinitely

    /// if we have a bug.

        // Now, look for any unreferenced blobs that are a bit old.

            {

                // this file is sufficiently recent that we should not remove it, just to be cautious.

                    );

                }

            };

        }

}

impl Store for SqliteStore {

        }

                // Somebody else has the lock.

                }

            }

        // This works around a false positive; see

        //   https://github.com/rust-lang/rust-clippy/issues/8114

        #[allow(clippy::let_and_return)]

        // In theory bad system clocks might generate table rows with times far in the future.

        // However, for data which is cached here which comes from the network consensus,

        // we rely on the fact that no consensus from the future exists, so this can't happen.

        // Bridge descriptors come from bridges and bridges might send crazy times,

        // so we need to discard any that look like they are from the future,

        // since otherwise wrong far-future timestamps might live in our DB indefinitely.

        #[cfg(feature = "bridge-client")]

        }

        };

            // TODO: If the cache is corrupt (because this blob is missing), and the cache has not yet

            // been cleaned, this may fail to find the latest consensus that we actually have.

        } else {

        }

        } else {

        }

    #[cfg(test)]

        {

        } else {

        }

        /// How long to keep a consensus around after it has expired

        const CONSENSUS_LIFETIME: time::Duration = time::Duration::days(4);

        // After a few days have passed, a consensus is no good for

        // anything at all, not even diffs.

        // TODO: We should probably remove the blob as well, but for now

        // this is enough.

        // TODO(nickm): Do I need to get a transaction here for performance?

        }

        }

        // TODO(nickm): Should I speed this up with a transaction, or

        // does it not matter for queries?

        }

        }

        }

    #[cfg(feature = "routerdesc")]

        // TODO(nickm): Should I speed this up with a transaction, or

        // does it not matter for queries?

        }

    #[cfg(feature = "routerdesc")]

        }

    #[cfg(feature = "bridge-client")]

    #[cfg(feature = "bridge-client")]

            // Hopefully whoever *does* have the lock will update the cache.

            // Otherwise it will contain a stale entry forever

            // (which we'll ignore, but waste effort on).

    #[cfg(feature = "bridge-client")]

            // This is called when we find corrupted or stale cache entries,

            // to stop us wasting time on them next time.

            // Hopefully whoever *does* have the lock will do this.

}

/// Handle to a blob that we have saved to disk but not yet committed to

/// the database.

struct SavedBlobHandle<'a> {

    /// Transaction we're using to add the blob to the ExtDocs table.

    tx: Transaction<'a>,

    /// Filename for the file, with respect to the blob directory.

    #[allow(unused)]

    fname: String,

    /// Declared digest string for this blob. Of the format

    /// "digesttype-hexstr".

    digeststr: String,

    /// An 'unlinker' for the blob file.

    unlinker: Unlinker,

}

/// Handle to a file which we might have to delete.

///

/// When this handle is dropped, the file gets deleted, unless you have

/// first called [`Unlinker::forget`].

struct Unlinker {

    /// The location of the file to remove, or None if we shouldn't

    /// remove it.

    p: Option<PathBuf>,

}

impl Unlinker {

    /// Make a new Unlinker for a given filename.

    /// Forget about this unlinker, so that the corresponding file won't

    /// get dropped.

}

impl Drop for Unlinker {

}

/// Convert a hexadecimal sha3-256 digest from the database into an array.

/// Convert a hexadecimal sha3-256 "digest string" as used in the

/// digest column from the database into an array.

    } else {

    }

/// Create a ConsensusMeta from a `Row` returned by one of

/// `FIND_LATEST_CONSENSUS_META` or `FIND_CONSENSUS_AND_META_BY_DIGEST`.

    Ok(ConsensusMeta::new(

    ))

/// Set up the tables for the arti cache schema in a sqlite database.

const INSTALL_V0_SCHEMA: &str = "

  -- Helps us version the schema.  The schema here corresponds to a

  -- version number called 'version', and it should be readable by

  -- anybody who is compliant with versions of at least 'readable_by'.

  CREATE TABLE TorSchemaMeta (

     name TEXT NOT NULL PRIMARY KEY,

     version INTEGER NOT NULL,

     readable_by INTEGER NOT NULL

  );

  INSERT INTO TorSchemaMeta (name, version, readable_by) VALUES ( 'TorDirStorage', 0, 0 );

  -- Keeps track of external blobs on disk.

  CREATE TABLE ExtDocs (

    -- Records a digest of the file contents, in the form 'dtype-hexstr'

    digest TEXT PRIMARY KEY NOT NULL,

    -- When was this file created?

    created DATE NOT NULL,

    -- After what time will this file definitely be useless?

    expires DATE NOT NULL,

    -- What is the type of this file? Currently supported are 'con:<flavor>'.

    type TEXT NOT NULL,

    -- Filename for this file within our blob directory.

    filename TEXT NOT NULL

  );

  -- All the microdescriptors we know about.

  CREATE TABLE Microdescs (

    sha256_digest TEXT PRIMARY KEY NOT NULL,

    last_listed DATE NOT NULL,

    contents BLOB NOT NULL

  );

  -- All the authority certificates we know.

  CREATE TABLE Authcerts (

    id_digest TEXT NOT NULL,

    sk_digest TEXT NOT NULL,

    published DATE NOT NULL,

    expires DATE NOT NULL,

    contents BLOB NOT NULL,

    PRIMARY KEY (id_digest, sk_digest)

  );

  -- All the consensuses we're storing.

  CREATE TABLE Consensuses (

    valid_after DATE NOT NULL,

    fresh_until DATE NOT NULL,

    valid_until DATE NOT NULL,

    flavor TEXT NOT NULL,

    pending BOOLEAN NOT NULL,

    sha3_of_signed_part TEXT NOT NULL,

    digest TEXT NOT NULL,

    FOREIGN KEY (digest) REFERENCES ExtDocs (digest) ON DELETE CASCADE

  );

  CREATE INDEX Consensuses_vu on CONSENSUSES(valid_until);

";

/// Update the database schema, from each version to the next

const UPDATE_SCHEMA: &[&str] = &["

  -- Update the database schema from version 0 to version 1.

  CREATE TABLE RouterDescs (

    sha1_digest TEXT PRIMARY KEY NOT NULL,

    published DATE NOT NULL,

    contents BLOB NOT NULL

  );

","

  -- Update the database schema from version 1 to version 2.

  -- We create this table even if the bridge-client feature is disabled, but then don't touch it at all.

  CREATE TABLE BridgeDescs (

    bridge_line TEXT PRIMARY KEY NOT NULL,

    fetched DATE NOT NULL,

    until DATE NOT NULL,

    contents BLOB NOT NULL

  );

"];

/// Update the database schema version tracking, from each version to the next

const UPDATE_SCHEMA_VERSION: &str = "

  UPDATE TorSchemaMeta SET version=? WHERE version<?;

";

/// Version number used for this version of the arti cache schema.

const SCHEMA_VERSION: u32 = UPDATE_SCHEMA.len() as u32;

/// Query: find the latest-expiring microdesc consensus with a given

/// pending status.

const FIND_CONSENSUS_P: &str = "

  SELECT valid_after, valid_until, filename

  FROM Consensuses

  INNER JOIN ExtDocs ON ExtDocs.digest = Consensuses.digest

  WHERE pending = ? AND flavor = ?

  ORDER BY valid_until DESC

  LIMIT 1;

";

/// Query: find the latest-expiring microdesc consensus, regardless of

/// pending status.

const FIND_CONSENSUS: &str = "

  SELECT valid_after, valid_until, filename

  FROM Consensuses

  INNER JOIN ExtDocs ON ExtDocs.digest = Consensuses.digest

  WHERE flavor = ?

  ORDER BY valid_until DESC

  LIMIT 1;

";

/// Query: Find the valid-after time for the latest-expiring

/// non-pending consensus of a given flavor.

const FIND_LATEST_CONSENSUS_META: &str = "

  SELECT valid_after, fresh_until, valid_until, sha3_of_signed_part, digest

  FROM Consensuses

  WHERE pending = 0 AND flavor = ?

  ORDER BY valid_until DESC

  LIMIT 1;

";

/// Look up a consensus by its digest-of-signed-part string.

const FIND_CONSENSUS_AND_META_BY_DIGEST_OF_SIGNED: &str = "

  SELECT valid_after, fresh_until, valid_until, sha3_of_signed_part, Consensuses.digest, filename

  FROM Consensuses

  INNER JOIN ExtDocs on ExtDocs.digest = Consensuses.digest

  WHERE Consensuses.sha3_of_signed_part = ?

  LIMIT 1;

";

/// Query: Update the consensus whose digest field is 'digest' to call it

/// no longer pending.

const MARK_CONSENSUS_NON_PENDING: &str = "

  UPDATE Consensuses

  SET pending = 0

  WHERE digest = ?;

";

/// Query: Remove the consensus with a given digest field.

#[allow(dead_code)]

const REMOVE_CONSENSUS: &str = "

  DELETE FROM Consensuses

  WHERE digest = ?;

";

/// Query: Find the authority certificate with given key digests.

const FIND_AUTHCERT: &str = "

  SELECT contents FROM AuthCerts WHERE id_digest = ? AND sk_digest = ?;

";

/// Query: find the microdescriptor with a given hex-encoded sha256 digest

const FIND_MD: &str = "

  SELECT contents

  FROM Microdescs

  WHERE sha256_digest = ?

";

/// Query: find the router descriptors with a given hex-encoded sha1 digest

#[cfg(feature = "routerdesc")]

const FIND_RD: &str = "

  SELECT contents

  FROM RouterDescs

  WHERE sha1_digest = ?

";

/// Query: find every ExtDocs member that has expired.

const FIND_EXPIRED_EXTDOCS: &str = "

  SELECT filename FROM ExtDocs where expires < datetime('now');

";

/// Query: find whether an ExtDoc is listed.

const COUNT_EXTDOC_BY_PATH: &str = "

  SELECT COUNT(*) FROM ExtDocs WHERE filename = ?;

";

/// Query: Add a new entry to ExtDocs.

const INSERT_EXTDOC: &str = "

  INSERT OR REPLACE INTO ExtDocs ( digest, created, expires, type, filename )

  VALUES ( ?, datetime('now'), ?, ?, ? );

";

/// Query: Add a new consensus.

const INSERT_CONSENSUS: &str = "

  INSERT OR REPLACE INTO Consensuses

    ( valid_after, fresh_until, valid_until, flavor, pending, sha3_of_signed_part, digest )

  VALUES ( ?, ?, ?, ?, ?, ?, ? );

";

/// Query: Add a new AuthCert

const INSERT_AUTHCERT: &str = "

  INSERT OR REPLACE INTO Authcerts

    ( id_digest, sk_digest, published, expires, contents)

  VALUES ( ?, ?, ?, ?, ? );

";

/// Query: Add a new microdescriptor

const INSERT_MD: &str = "

  INSERT OR REPLACE INTO Microdescs ( sha256_digest, last_listed, contents )

  VALUES ( ?, ?, ? );

";

/// Query: Add a new router descriptor

#[allow(unused)]

#[cfg(feature = "routerdesc")]

const INSERT_RD: &str = "

  INSERT OR REPLACE INTO RouterDescs ( sha1_digest, published, contents )

  VALUES ( ?, ?, ? );

";

/// Query: Change the time when a given microdescriptor was last listed.

const UPDATE_MD_LISTED: &str = "

  UPDATE Microdescs

  SET last_listed = max(last_listed, ?)

  WHERE sha256_digest = ?;

";

/// Query: Find a cached bridge descriptor

#[cfg(feature = "bridge-client")]

const FIND_BRIDGEDESC: &str = "SELECT fetched, contents FROM BridgeDescs WHERE bridge_line = ?;";

/// Query: Record a cached bridge descriptor

#[cfg(feature = "bridge-client")]

const INSERT_BRIDGEDESC: &str = "

  INSERT OR REPLACE INTO BridgeDescs ( bridge_line, fetched, until, contents )

  VALUES ( ?, ?, ?, ? );

";

/// Query: Remove a cached bridge descriptor

#[cfg(feature = "bridge-client")]

#[allow(dead_code)]

const DELETE_BRIDGEDESC: &str = "DELETE FROM BridgeDescs WHERE bridge_line = ?;";

/// Query: Discard every expired extdoc.

///

/// External documents aren't exposed through [`Store`].

const DROP_OLD_EXTDOCS: &str = "DELETE FROM ExtDocs WHERE expires < datetime('now');";

/// Query: Discard an extdoc with a given path.

const DELETE_EXTDOC_BY_FILENAME: &str = "DELETE FROM ExtDocs WHERE filename = ?;";

/// Query: Discard every router descriptor that hasn't been listed for 3

/// months.

// TODO: Choose a more realistic time.

const DROP_OLD_ROUTERDESCS: &str = "DELETE FROM RouterDescs WHERE published < ?;";

/// Query: Discard every microdescriptor that hasn't been listed for 3 months.

// TODO: Choose a more realistic time.

const DROP_OLD_MICRODESCS: &str = "DELETE FROM Microdescs WHERE last_listed < ?;";

/// Query: Discard every expired authority certificate.

const DROP_OLD_AUTHCERTS: &str = "DELETE FROM Authcerts WHERE expires < ?;";

/// Query: Discard every consensus that's been expired for at least

/// two days.

const DROP_OLD_CONSENSUSES: &str = "DELETE FROM Consensuses WHERE valid_until < ?;";

/// Query: Discard every bridge descriptor that is too old, or from the future.  (Both ?=now.)

#[cfg(feature = "bridge-client")]

const DROP_OLD_BRIDGEDESCS: &str = "DELETE FROM BridgeDescs WHERE ? > until OR fetched > ?;";

#[cfg(test)]

pub(crate) mod test {

    #![allow(clippy::unwrap_used)]

    use super::*;

    use crate::storage::EXPIRATION_DEFAULTS;

    use hex_literal::hex;

    use tempfile::{tempdir, TempDir};

    use time::ext::NumericalDuration;

    pub(crate) fn new_empty() -> Result<(TempDir, SqliteStore)> {

        let tmp_dir = tempdir().unwrap();

        let sql_path = tmp_dir.path().join("db.sql");

        let conn = rusqlite::Connection::open(sql_path)?;

        let blob_path = tmp_dir.path().join("blobs");

        let blob_dir = fs_mistrust::Mistrust::builder()

            .dangerously_trust_everyone()

            .build()

            .unwrap()

            .verifier()

            .make_secure_dir(blob_path)

            .unwrap();

        let store = SqliteStore::from_conn(conn, blob_dir)?;

        Ok((tmp_dir, store))

    }

    #[test]

    fn init() -> Result<()> {

        let tmp_dir = tempdir().unwrap();

        let blob_dir = fs_mistrust::Mistrust::builder()

            .dangerously_trust_everyone()

            .build()

            .unwrap()

            .verifier()

            .secure_dir(&tmp_dir)

            .unwrap();

        let sql_path = tmp_dir.path().join("db.sql");

        // Initial setup: everything should work.

        {

            let conn = rusqlite::Connection::open(&sql_path)?;

            let _store = SqliteStore::from_conn(conn, blob_dir.clone())?;

        }

        // Second setup: shouldn't need to upgrade.

        {

            let conn = rusqlite::Connection::open(&sql_path)?;

            let _store = SqliteStore::from_conn(conn, blob_dir.clone())?;

        }

        // Third setup: shouldn't need to upgrade.

        {

            let conn = rusqlite::Connection::open(&sql_path)?;

            conn.execute_batch("UPDATE TorSchemaMeta SET version = 9002;")?;

            let _store = SqliteStore::from_conn(conn, blob_dir.clone())?;

        }

        // Fourth: this says we can't read it, so we'll get an error.

        {

            let conn = rusqlite::Connection::open(&sql_path)?;

            conn.execute_batch("UPDATE TorSchemaMeta SET readable_by = 9001;")?;

            let val = SqliteStore::from_conn(conn, blob_dir);

            assert!(val.is_err());

        }

        Ok(())

    }

    #[test]

    fn bad_blob_fname() -> Result<()> {

        let (_tmp_dir, store) = new_empty()?;

        assert!(store.blob_dir.join("abcd").is_ok());

        assert!(store.blob_dir.join("abcd..").is_ok());

        assert!(store.blob_dir.join("..abcd..").is_ok());

        assert!(store.blob_dir.join(".abcd").is_ok());

        assert!(store.blob_dir.join("..").is_err());

        assert!(store.blob_dir.join("../abcd").is_err());

        assert!(store.blob_dir.join("/abcd").is_err());

        Ok(())

    }

    #[test]

    fn blobs() -> Result<()> {

        let (_tmp_dir, mut store) = new_empty()?;

        let now = OffsetDateTime::now_utc();

        let one_week = 1.weeks();

        let fname1 = store.save_blob(

            b"Hello world",

            "greeting",

            "sha1",

            &hex!("7b502c3a1f48c8609ae212cdfb639dee39673f5e"),

            now + one_week,

        )?;

        let fname2 = store.save_blob(

            b"Goodbye, dear friends",

            "greeting",

            "sha1",

            &hex!("2149c2a7dbf5be2bb36fb3c5080d0fb14cb3355c"),

            now - one_week,

        )?;

        assert_eq!(

            fname1,

            "greeting_sha1-7b502c3a1f48c8609ae212cdfb639dee39673f5e"

        );

        assert_eq!(

            &std::fs::read(store.blob_dir.join(&fname1)?).unwrap()[..],

            b"Hello world"

        );

        assert_eq!(

            &std::fs::read(store.blob_dir.join(&fname2)?).unwrap()[..],

            b"Goodbye, dear friends"

        );

        let n: u32 = store

            .conn

            .query_row("SELECT COUNT(filename) FROM ExtDocs", [], |row| row.get(0))?;

        assert_eq!(n, 2);

        let blob = store.read_blob(&fname2)?.unwrap();

        assert_eq!(blob.as_str().unwrap(), "Goodbye, dear friends");

        // Now expire: the second file should go away.

        store.expire_all(&EXPIRATION_DEFAULTS)?;

        assert_eq!(

            &std::fs::read(store.blob_dir.join(&fname1)?).unwrap()[..],

            b"Hello world"

        );

        assert!(std::fs::read(store.blob_dir.join(&fname2)?).is_err());

        let n: u32 = store

            .conn

            .query_row("SELECT COUNT(filename) FROM ExtDocs", [], |row| row.get(0))?;

        assert_eq!(n, 1);

        Ok(())

    }

    #[test]

    fn consensus() -> Result<()> {

        use tor_netdoc::doc::netstatus;

        let (_tmp_dir, mut store) = new_empty()?;

        let now = OffsetDateTime::now_utc();

        let one_hour = 1.hours();

        assert_eq!(

            store.latest_consensus_time(ConsensusFlavor::Microdesc)?,

            None

        );

        let cmeta = ConsensusMeta::new(

            netstatus::Lifetime::new(

                now.into(),

                (now + one_hour).into(),

                SystemTime::from(now + one_hour * 2),

            )

            .unwrap(),

            [0xAB; 32],

            [0xBC; 32],

        );

        store.store_consensus(

            &cmeta,

            ConsensusFlavor::Microdesc,

            true,

            "Pretend this is a consensus",

        )?;

        {

            assert_eq!(

                store.latest_consensus_time(ConsensusFlavor::Microdesc)?,

                None

            );

            let consensus = store

                .latest_consensus(ConsensusFlavor::Microdesc, None)?

                .unwrap();

            assert_eq!(consensus.as_str()?, "Pretend this is a consensus");

            let consensus = store.latest_consensus(ConsensusFlavor::Microdesc, Some(false))?;

            assert!(consensus.is_none());

        }

        store.mark_consensus_usable(&cmeta)?;

        {

            assert_eq!(

                store.latest_consensus_time(ConsensusFlavor::Microdesc)?,

                now.into()

            );

            let consensus = store

                .latest_consensus(ConsensusFlavor::Microdesc, None)?

                .unwrap();

            assert_eq!(consensus.as_str()?, "Pretend this is a consensus");

            let consensus = store

                .latest_consensus(ConsensusFlavor::Microdesc, Some(false))?

                .unwrap();

            assert_eq!(consensus.as_str()?, "Pretend this is a consensus");

        }

        {

            let consensus_text = store.consensus_by_meta(&cmeta)?;

            assert_eq!(consensus_text.as_str()?, "Pretend this is a consensus");

            let (is, _cmeta2) = store

                .consensus_by_sha3_digest_of_signed_part(&[0xAB; 32])?

                .unwrap();

            assert_eq!(is.as_str()?, "Pretend this is a consensus");

            let cmeta3 = ConsensusMeta::new(

                netstatus::Lifetime::new(

                    now.into(),

                    (now + one_hour).into(),

                    SystemTime::from(now + one_hour * 2),

                )

                .unwrap(),

                [0x99; 32],

                [0x99; 32],

            );

            assert!(store.consensus_by_meta(&cmeta3).is_err());

            assert!(store

                .consensus_by_sha3_digest_of_signed_part(&[0x99; 32])?

                .is_none());

        }

        {

            assert!(store

                .consensus_by_sha3_digest_of_signed_part(&[0xAB; 32])?

                .is_some());

            store.delete_consensus(&cmeta)?;

            assert!(store

                .consensus_by_sha3_digest_of_signed_part(&[0xAB; 32])?

                .is_none());

        }

        Ok(())

    }