sequoia-cert-store 0.7.3

use std::borrow::Cow;
use std::collections::BTreeSet;
use std::collections::btree_map::{BTreeMap, Entry};
use std::ffi::OsString;
use std::path::Path;
use std::path::PathBuf;
use std::str;
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use std::str::FromStr;
use std::thread::sleep;

use anyhow::Context;

use rusqlite::{
    Connection,
    Error as RError,
    ErrorCode,
    OpenFlags,
    Result as RResult,
    Transaction,
    TransactionBehavior,
    params,
};
use smallvec::{SmallVec, smallvec};

use sequoia_openpgp as openpgp;
use openpgp::Fingerprint;
use openpgp::KeyHandle;
use openpgp::KeyID;
use openpgp::Result;
use openpgp::cert::prelude::*;
use openpgp::cert::raw::{RawCert, RawCertParser};
use openpgp::packet::UserID;
use openpgp::parse::Parse;
use openpgp::parse::buffered_reader as br;

use openpgp_cert_d as cert_d;

use crate::LazyCert;
use crate::store::MergeCerts;
use crate::store::{Store, StoreError};
use crate::store::StoreUpdate;
use crate::store::UserIDIndex;
use crate::store::UserIDQueryParams;

const TRACE: bool = cfg!(test) || crate::TRACE;

mod cache;
use cache::CertdSignatureVerificationCache;

mod shadow_ca;
use shadow_ca::CA;

/// Do not scan more often than every SCAN_LIMIT.
#[cfg(not(test))]
const SCAN_LIMIT: Duration = Duration::from_millis(50);

/// No SCAN_LIMIT for tests.  We want to make race conditions and lock
/// contention more likely.
#[cfg(test)]
const SCAN_LIMIT: Duration = Duration::from_millis(0);

/// The tag is the tag of the in-memory index.
///
/// We need to keep both the persistent index (the DB) in sync
/// with the disk, and the in-memory index in sync with the disk.
/// To detect if the persistent index is out of sync, we compare
/// the tag stored in the persistent index with the cert's Tag.
///
/// This is not enough to keep the in-memory index in sync.
/// Consider: an external process changes the cert-d, and updates
/// the persistent index.  The cert-d and persistent index are
/// consistent, but our in-memory index is out of sync!
///
/// Thus, we track the tag that the in-memory index corresponds
/// to.  This tag is associated with the connection to avoid
/// taking additional locks.
struct DB {
    in_memory_tag: Option<cert_d::Tag>,
    conn: Connection,
}

/// Statistics about the number of times the in-memory index, and the
/// persistent index were loaded.
pub(crate) struct LoadStats {
    /// The number of times the in-memory index was loaded from the
    /// database.
    #[cfg(test)]
    in_memory_loads: std::sync::atomic::AtomicUsize,
    /// The number of times the persistent index was loaded from the
    /// cert-d.
    #[cfg(test)]
    persistent_index_scans: std::sync::atomic::AtomicUsize,
}

impl LoadStats {
    /// Gets the in-memory-load counter.
    ///
    /// The number of times the in-memory index was loaded from the
    /// database.
    ///
    /// This counter is incremented when we load the persistent index
    /// into memory.  Note: when the memory index is initialized as a
    /// side effect of creating the persistent index, this is not
    /// incremented.
    #[allow(dead_code)]
    pub(crate) fn in_memory_loads(&self) -> usize {
        #[allow(unused_mut, unused_assignments)]
        let mut counter = 0;

        #[cfg(test)]
        {
            counter = self.in_memory_loads.load(
                std::sync::atomic::Ordering::Relaxed);
        }

        counter
    }

    /// Increments the in-memory-load counter.
    fn in_memory_loads_inc(&self) {
        #[cfg(test)]
        self.in_memory_loads.fetch_add(
            1, std::sync::atomic::Ordering::Relaxed);
    }

    /// Gets the persistent-index-scans counter.
    ///
    /// The number of times the persistent index was loaded from the
    /// cert-d.
    ///
    /// This counter is incremented when we scan the cert-d in order
    /// to build or update the persistent index.
    #[allow(dead_code)]
    pub(crate) fn persistent_index_scans(&self) -> usize {
        #[allow(unused_mut, unused_assignments)]
        let mut counter = 0;

        #[cfg(test)]
        {
            counter = self.persistent_index_scans.load(
                std::sync::atomic::Ordering::Relaxed);
        }

        counter
    }

    /// Increments the persistent-index-scans counter.
    fn persistent_index_scans_inc(&self) {
        #[cfg(test)]
        self.persistent_index_scans.fetch_add(
            1, std::sync::atomic::Ordering::Relaxed);
    }
}

impl Default for LoadStats {
    fn default() -> Self {
        Self {
            #[cfg(test)]
            in_memory_loads: 0.into(),
            #[cfg(test)]
            persistent_index_scans: 0.into(),
        }
    }
}

/// A cert-d backed store with persistent indices.
///
/// We use the cert-d as the source of truth, and maintain in-core
/// indices mapping from key handles, subkey handles, and user IDs to
/// cert fingerprints.  We serve all requests from the in-core caches.
///
/// We persist the indices in SQLite to cache them, but this is a
/// best-effort mechanism: concurrent database mutators may have
/// exclusive write rights to the database and our update may fail.
/// If that happens, we stop trying to update the database, but still
/// update the in-core indices.  We abort the transaction.  And, since
/// we didn't update the cert-d tag in the database, we (or another
/// process) will recognize the db as stale next time it is accessed,
/// and update it.
///
/// # Lock order
///
/// This struct uses a series of locks, which must be taken in the
/// following order to avoid deadlocks:
///
///   - CertD::cas
///   - CertD::conn
///   - CertD::last_scan
///   - CertD::certs_cache
///   - CertD::index_cache
pub struct CertD<'a> {
    certd: cert_d::CertD,

    /// A cache of the trust root, and shadow CAs keyed by the cert-d
    /// special name.
    cas: Mutex<BTreeMap<String, CA<'a>>>,

    path: PathBuf,

    conn: Mutex<DB>,

    /// Cache indexed by primary key fingerprint.
    certs_cache: RwLock<BTreeMap<Fingerprint, (Arc<LazyCert<'a>>, cert_d::Tag)>>,

    /// We rate limit scans for changes.  This is the last time a scan
    /// was done (if ever).
    last_scan: Mutex<Option<Instant>>,

    /// Indices mapping from (sub)key fingerprint, (sub)key key ID,
    /// and user ID to certificate fingerprint.
    index_cache: RwLock<Index>,

    pub(crate) load_stats: LoadStats,

    // We keep this for its drop handler.
    #[allow(dead_code)]
    signature_cache: Option<CertdSignatureVerificationCache>,
}
assert_send_and_sync!(CertD<'_>);

/// Indices mapping from (sub)key fingerprint, (sub)key key ID, and
/// user ID to certificate fingerprint.
#[derive(Default)]
struct Index {
    /// Maps cert and subkey fingerprints to cert fingerprints.
    ///
    /// The values are a sorted, de-duplicated list of certificate
    /// fingerprints.  Use `insert_unique` to maintain order and
    /// uniqueness.
    by_key_fingerprint: BTreeMap<Fingerprint, SmallVec<[Fingerprint; 1]>>,

    /// Maps cert and subkey key IDs to cert fingerprints.
    ///
    /// The values are a sorted, de-duplicated list of certificate
    /// fingerprints.  Use `insert_unique` to maintain order and
    /// uniqueness.
    by_key_id: BTreeMap<KeyID, SmallVec<[Fingerprint; 1]>>,

    /// Maps user IDs to cert fingerprints.
    ///
    /// XXX: UserIDIndex has an internal RwLock.  We could avoid that
    /// by using UserIDIndexInner, but the query interface is on
    /// UserIDIndex.
    by_userid: UserIDIndex,

}

/// Inserts `fp` into `bucket` if it is not present.
fn insert_unique(bucket: &mut SmallVec<[Fingerprint; 1]>,
                 fp: Cow<Fingerprint>)
{
    if let Err(i) = bucket.binary_search(&fp) {
        bucket.insert(i, fp.into_owned());
    }
}

impl<'a> CertD<'a> {
    /// Returns the canonicalized path.
    ///
    /// If path is `None`, then returns the default location.
    fn path(path: Option<&Path>) -> Result<PathBuf>
    {
        if let Some(path) = path {
            Ok(path.to_owned())
        } else {
            Ok(cert_d::CertD::user_configured_store_path()?)
        }
    }

    /// Opens the default cert-d for reading and writing.
    pub fn open_default() -> Result<Self>
    {
        let path = Self::path(None)?;
        Self::open(path)
    }

    /// Opens a cert-d for reading and writing.
    pub fn open<P>(path: P) -> Result<Self>
        where P: AsRef<Path>,
    {
        tracer!(TRACE, "CertD::open");

        let path = path.as_ref();
        t!("loading cert-d {:?}", path);

        let certd = cert_d::CertD::with_base_dir(&path)
            .map_err(|err| {
                t!("While opening the certd {:?}: {}", path, err);
                let err = anyhow::Error::from(err)
                    .context(format!("While opening the certd {:?}", path));
                err
            })?;

        // We include the hostname in the file name to prevent having
        // to share databases and their locks across network file
        // systems.
        let mut database_name =
            OsString::from("_sequoia_cert_store_index_v1_on_");
        database_name.push(&gethostname::gethostname());
        database_name.push(".sqlite");

        let index_path = path.join(database_name);
        let mut conn = Connection::open_with_flags(
            &index_path,
            OpenFlags::SQLITE_OPEN_READ_WRITE
                | OpenFlags::SQLITE_OPEN_CREATE
                | OpenFlags::SQLITE_OPEN_PRIVATE_CACHE)?;

        if false { // TRACE {
            fn trace(m: &str) {
                eprintln!("SQLite: {}", m);
            }
            conn.trace(Some(trace));
        }

        // for tests, deactivate the 'double-quoted string literal' feature
        #[cfg(test)]
        {
            conn.set_db_config(rusqlite::config::DbConfig::SQLITE_DBCONFIG_DQS_DDL, false)?;
            conn.set_db_config(rusqlite::config::DbConfig::SQLITE_DBCONFIG_DQS_DML, false)?;
        }

        conn.pragma_update_and_check(None, "secure_delete", true, |_| Ok(()))?;
        conn.pragma_update(None, "foreign_keys", true)?;

        // Changing to WAL requires a lock on the database. The
        // default is not WAL, so creating a new file makes this
        // actually a change.  This might return
        // `ErrorCode::DatabaseBusy` if two instances perform this at
        // the same time.  In that case, try again.
        //
        // See:
        //
        // - https://sqlite.org/forum/forumpost/e236e37925b628ba
        //
        //   the issue with the 'Database locked' error is that since
        //   WAL mode is a database property and not just a connection
        //   property, to set it requires getting a lock to write to
        //   the database, and I think it will need an 'Exclusive'
        //   lock, so if the database is busy with other connections,
        //   it might get locked out from being able to make the
        //   change.
        //
        // - https://docs.rs/sqlx/latest/sqlx/sqlite/struct.SqliteConnectOptions.html#method.journal_mode
        //
        //   Opening a connection to a database created in WAL mode
        //   with a different journal_mode will erase the setting on
        //   the database, requiring an exclusive lock to do so. You
        //   may get a database is locked (corresponding to
        //   SQLITE_BUSY) error if another connection is accessing the
        //   database file at the same time.
        repeat_while_busy(|| {
            conn.pragma_update_and_check(
                None, "journal_mode", "WAL", |_| Ok(()))
        })?;

        let certd = Self {
            certd,
            cas: Default::default(),
            path: path.into(),
            conn: Mutex::new(DB {
                in_memory_tag: None,
                conn,
            }),
            certs_cache: Default::default(),
            last_scan: Mutex::new(None),
            index_cache: RwLock::new(Index {
                by_key_fingerprint: Default::default(),
                by_key_id: Default::default(),
                by_userid: Default::default(),
            }),
            load_stats: Default::default(),
            signature_cache: CertdSignatureVerificationCache::cache_file(path)
                .map(|filename| {
                    CertdSignatureVerificationCache::load(filename)
                })
                .ok(),
        };

        certd.initialize()?;
        Ok(certd)
    }

    /// Returns a reference to the low-level `CertD`.
    pub fn certd(&self) -> &cert_d::CertD {
        &self.certd
    }

    /// Returns a mutable reference to the low-level `CertD`.
    pub fn certd_mut(&mut self) -> &mut cert_d::CertD {
        &mut self.certd
    }

    /// Initializes a certd by creating the persistent index, reading
    /// the entries and populating the index.
    fn initialize(&self) -> Result<()>
    {
        tracer!(TRACE, "CertD::initialize");

        repeat_while_busy(|| -> std::result::Result<_, RepeatError> {
            let mut db = self.conn.lock().unwrap();

            let tx = Transaction::new(&mut db.conn, TransactionBehavior::Deferred)?;

            // First, see which database version we're dealing with.
            match self.db_version(&tx) {
                Ok(1) => {
                    drop(tx);
                    self.scan(&mut db)?;
                    Ok(())
                },
                Ok(n) => {
                    t!("Expected version 1, got version {}.  Re-initializing.", n);
                    Self::initialize_v1(&tx)?;
                    tx.commit()?;
                    self.initial_scan(&mut db)?;
                    Ok(())
                },
                Err(RError::SqliteFailure(e, _)) if e.code == ErrorCode::Unknown =>
                {
                    t!("Initial indexing.");
                    Self::initialize_v1(&tx)?;
                    tx.commit()?;
                    self.initial_scan(&mut db)?;
                    Ok(())
                },
                Err(e) => Err(e.into()),
            }
        })?;

        Ok(())
    }

    /// Creates the disk index, doing the initial scan in parallel.
    ///
    /// It's okay if we fail to create the disk index, because the
    /// database is busy: someone else will do it later.  What's
    /// essential is that the in-memory index reflects the certificate
    /// store.
    fn initial_scan(&self, db: &mut DB) -> Result<()>
    {
        tracer!(false, "CertD::initial_scan");

        // Even though the first SQL statement will be a write, we ask
        // for a Deferred transaction instead of an Immediate
        // transaction.  A Deferred transaction will not return
        // `DatabaseBusy`, because it doesn't actually interact with
        // the database:
        //
        //   Internally, the BEGIN DEFERRED statement merely sets a
        //   flag on the database connection that turns off the
        //   automatic commit that would normally occur when the last
        //   statement finishes.
        //
        //   ...
        //
        //   IMMEDIATE causes the database connection to start a new
        //   write immediately, without waiting for a write
        //   statement. The BEGIN IMMEDIATE might fail with
        //   SQLITE_BUSY if another write transaction is already
        //   active on another database connection.
        //
        //   https://www.sqlite.org/lang_transaction.html
        //
        // This means even if we would get `DatabaseBusy`, we'll at
        // least have a `Transaction`, which we can pass to
        // `db_insert`, which we have to call to build the in-memory
        // index.
        let tx = Transaction::new_unchecked(
            &db.conn, TransactionBehavior::Deferred)?;

        let mut commit = true;
        let mut last_scan = self.last_scan.lock().unwrap();
        let certd_tag = self.certd.tag();
        t!("Initial certd tag is {:?}", certd_tag);

        // Load all certs without canonicalizing them.
        self.prefetch_internal(true, &[], false);

        for (fp, (cert, tag)) in self.certs_cache.read().unwrap().iter() {
            t!("Inserting {}", cert.fingerprint());
            self.db_insert(
                &tx,
                &mut commit,
                tag.clone(),
                fp,
                Box::new(cert.keys().map(|sk| sk.fingerprint())),
                Box::new(cert.userids()))?;
        }

        // Save some cycles next time.
        self.db_set_certd_tag(&tx, certd_tag)?;
        t!("Setting DB tag to {:?}", certd_tag);
        if commit {
            ignore_sqlite_busy(tx.commit())?;
        } else {
            ignore_sqlite_busy(tx.rollback())?;
        }
        *last_scan = Some(Instant::now());

        self.load_stats.persistent_index_scans_inc();

        // As a side effect of updating the persistent index, we build
        // the in-memory index.
        db.in_memory_tag = Some(certd_tag);

        Ok(())
    }

    /// Initializes the in-memory index from the database.
    ///
    /// This does a hard reinitialization.  It's up to the caller to
    /// decide whether this is necessary.
    fn initialize_in_memory_index(&self, tx: &Transaction)
        -> Result<()>
    {
        tracer!(TRACE, "CertD::initialize_in_memory_index");

        let mut index_cache = self.index_cache.write().unwrap();

        let mut by_key_fingerprint = || -> Result<_> {
            let mut stmt = tx.prepare("SELECT fingerprint, cert_fingerprint \
                                       FROM keys")?;
            let r = stmt.query_map(
                [],
                |row| {
                    Ok((Fingerprint::from_str(&row.get::<_, String>(0)?),
                        Fingerprint::from_str(&row.get::<_, String>(1)?)))
                })?
                .filter_map(|row| {
                    if let Ok((subkey, cert)) = row {
                        if let (Ok(subkey), Ok(cert)) = (subkey, cert) {
                            Some((subkey, smallvec![cert]))
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                })
                .collect::<Vec<(Fingerprint, SmallVec<[Fingerprint; 1]>)>>();
            Ok(r)
        }().unwrap_or_else(|_| Default::default());

        // Sort by subkey fingerprint, then collate adjacent lists
        // into the first.
        by_key_fingerprint.sort_unstable_by(|a, b| a.0.cmp(&b.0));

        // We maintain a pointer to the first item within a run of the
        // same fingerprint.  We'll collate all other fingerprints
        // into the bucket of the first item.
        let mut current: Option<&mut (Fingerprint, SmallVec<[Fingerprint; 1]>)>
            = None;
        for item in by_key_fingerprint.iter_mut() {
            if let Some(c) = current.as_mut() {
                if c.0 == item.0 {
                    insert_unique(&mut c.1, Cow::Owned(item.1.pop().unwrap()));
                } else {
                    // The current run of fingerprints has ended and
                    // it is time to update the pointer.
                    current = Some(item);
                }
            } else {
                // Current was None.  We found our first current.
                current = Some(item);
            }
        }

        // Only retain the entries with non-empty buckets.
        by_key_fingerprint.retain(|(_, cert_bucket)| ! cert_bucket.is_empty());

        let by_key_id = by_key_fingerprint.iter()
            .map(|(subkey, cert_bucket)| {
                (KeyID::from(subkey), cert_bucket.clone())
            })
            .collect();

        let by_userid = || -> Result<_> {
            let mut stmt = tx.prepare(
                "SELECT cert_fingerprint, userid, email \
                 FROM userids")?;
            let r = stmt.query_map(
                [],
                |row| {
                    Ok((Fingerprint::from_str(&row.get::<_, String>(0)?),
                        UserID::from(row.get::<_, String>(1)?),
                        row.get::<_, Option<String>>(2)?))
                })?
                .filter_map(|row| {
                    if let Ok((Ok(fp), userid, email)) = row {
                        Some((fp, userid, email.filter(|e| ! e.is_empty())))
                    } else {
                        None
                    }
                })
                .collect::<UserIDIndex>();
            Ok(r)
        }().unwrap_or_else(|_| Default::default());

        *index_cache = Index {
                by_key_fingerprint: by_key_fingerprint.into_iter().collect(),
                by_key_id,
                by_userid,
        };

        self.load_stats.in_memory_loads_inc();

        Ok(())
    }

    /// Makes sure that the on-disk index and the in-memory index are
    /// up-to-date.
    fn scan(&self, db: &mut DB)
        -> Result<()>
    {
        tracer!(false, "CertD::scan");

        // Rate-limit scanning.  Don't rescan if we did a scan (very)
        // recently.
        let mut last_scan = self.last_scan.lock().unwrap();
        let now = Instant::now();
        if let Some(last_scan) = last_scan.as_ref() {
            if now < *last_scan {
                // Time travel?
                t!("Maybe rescanning (time travel)");
            } else if now.duration_since(*last_scan) > SCAN_LIMIT {
                // It's been too long since the last scan.
                t!("Maybe rescanning");
            } else {
                return Ok(())
            };
        } else {
            t!("Never scanned");
        }
        // We hold last_scan.  This blocks other threads from doing a
        // scan, which is exactly what we want as when they wake up,
        // they'll see that a scan was just done.
        *last_scan = Some(now);

        // See if the database is up-to-date.
        let disk_certd_tag = self.certd.tag();

        // Start a transaction in deferred mode.  We may still
        // discover that then database is current, so we don't need an
        // immediate write lock on the database.
        let tx = Transaction::new(
            &mut db.conn, TransactionBehavior::Deferred)?;
        let mut commit = true;

        let db_tag = self.db_certd_tag(&tx)?;
        if db_tag == Some(disk_certd_tag) {
            // It is.  We're done.
            t!("Persistent index is up to date, exiting.");

            if db.in_memory_tag != Some(disk_certd_tag) {
                t!("In-memory index's tag does not match disk tag, reloading");
                self.initialize_in_memory_index(&tx)?;
                db.in_memory_tag = db_tag;
            } else {
                t!("In-memory index's tag matches disk tag, exiting");
            }

            return Ok(());
        }
        t!("Persistent index's tag does not match disk tag, rescanning");

        // Read all tags indexed by fingerprint into the core.  This
        // has two benefits.  First, we have a quick way to compare
        // tags later.  Second, we can detect corrupted fingerprints
        // and key IDs and remove them from the index.
        let mut stmt = tx.prepare_cached(
            "SELECT rowid, fingerprint, keyid, tag FROM certs")?;
        let mut bad = Vec::new();
        let tags: BTreeMap<_, _> = stmt.query([])?
            .mapped(|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
            .filter_map(|row| {
                let (rowid, fp_str, keyid_str, tag):
                (i64, String, String, i64) = row.ok()?;
                match (fp_str.parse::<Fingerprint>(),
                       keyid_str.parse::<KeyID>()) {
                    (Ok(fp), Ok(_)) =>
                        Some((fp, db2tag(tag))),
                    _ => {
                        bad.push(rowid);
                        None
                    },
                }
            })
            .collect();
        drop(stmt);

        for bad_rowid in bad {
            commit = commit && {
                let mut stmt = tx.prepare_cached(
                    "DELETE FROM certs WHERE rowid = ?0")?;
                ignore_sqlite_busy(stmt.execute(params![bad_rowid]))?
            };
        }

        // Scan the subkey index for corrupted fingerprints and key
        // IDs and remove them from the index.
        let mut stmt = tx.prepare_cached(
            "SELECT cert_fingerprint, fingerprint, keyid FROM keys")?;
        let mut bad = BTreeSet::new();
        for row in stmt.query([])?
            .mapped(|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
        {
            let (cert_fp, fp_str, keyid_str): (String, String, String) =
                row?;
            match (fp_str.parse::<Fingerprint>(),
                   keyid_str.parse::<KeyID>()) {
                (Ok(_), Ok(_)) => (),
                _ => {
                    bad.insert(cert_fp);
                },
            }
        }
        drop(stmt);
        for bad_cert_fp in bad {
            commit = commit && {
                let mut stmt = tx.prepare_cached(
                    "DELETE FROM keys WHERE cert_fingerprint = ?0")?;
                ignore_sqlite_busy(stmt.execute(params![bad_cert_fp]))?
            };
        }

        // Scan the subkey index for corrupted fingerprints and key
        // IDs and remove them from the index.
        let mut stmt = tx.prepare_cached(
            "SELECT cert_fingerprint FROM userids")?;
        let mut bad = BTreeSet::new();
        for row in stmt.query([])?
            .mapped(|r| Ok(r.get::<_, String>(0)?))
        {
            if let Ok(fp) = row {
                if fp.parse::<Fingerprint>().is_err() {
                    bad.insert(fp);
                }
             }
        }
        drop(stmt);
        for bad_cert_fp in bad {
            commit = commit && {
                let mut stmt = tx.prepare_cached(
                    "DELETE FROM userids WHERE cert_fingerprint = ?0")?;
                ignore_sqlite_busy(stmt.execute(params![bad_cert_fp]))?
            };
        }

        let transaction = |commit: &mut bool, fp_str: &str, file|
            -> Result<()>
        {
            t!("Considering {}", fp_str);
            let fp: Fingerprint = fp_str.parse()?;
            let tag = cert_d::Tag::try_from(&file)?;

            if tags.get(&fp) == Some(&tag) {
                t!("{} is up to date!", fp_str);
                return Ok(());
            }

            let mut parser = br::File::new_with_cookie(
                file, self.certd.get_path(fp_str)?, Default::default())
                .map_err(Into::into)
                .and_then(RawCertParser::from_buffered_reader)
                .map_err(|err| {
                t!("While reading {:?} from the certd {:?}: {}",
                   fp, self.path, err);
                err
            })?;

            match parser.next() {
                Some(Ok(cert)) => {
                    t!("inserting");
                    self.insert_rawcert(&tx, commit, cert, tag)?;
                    Ok(())
                },

                Some(Err(err)) => {
                    t!("While parsing {:?} from the certd {:?}: {}",
                       fp, self.path, err);
                    Err(err)
                },

                None => {
                    t!("While parsing {:?} from the certd {:?}: empty file",
                       fp, self.path);
                    Err(anyhow::anyhow!("empty file"))
                },
            }
        };

        for (fp, file) in self.certd.iter_files().filter_map(cert_d::Result::ok)
        {
            if let Err(e) = transaction(&mut commit, &fp, file) {
                t!("{}: {}", fp, e);
                let mut stmt = tx.prepare_cached(
                    "DELETE FROM certs WHERE fingerprint = ?1")?;
                commit = commit &&
                    ignore_sqlite_busy(stmt.execute(params![fp]))?;
            }
        }

        // Save some cycles next time.
        self.db_set_certd_tag(&tx, disk_certd_tag)?;

        // Make sure the in-memory index is up to date.
        if db_tag == db.in_memory_tag {
            // The in-memory index was consistent with the persistent
            // index.  As we updated the persistent index, we also
            // updated the in-memory index so the in-memory index is
            // not also consistent with the cert-d.
            db.in_memory_tag = Some(disk_certd_tag);
        } else if db.in_memory_tag != Some(disk_certd_tag) {
            // The in-memory index was not consistent with the cert-d.
            // Rebuild it.
            t!("In-memory index out of date relative to db, reloading");
            self.initialize_in_memory_index(&tx)?;
            db.in_memory_tag = db_tag;
        }

        *last_scan = Some(Instant::now());
        if commit {
            tx.commit()?;
        } else {
            tx.rollback()?;
        }

        self.load_stats.persistent_index_scans_inc();

        Ok(())
    }

    /// Returns the database version.
    fn db_version(&self, tx: &Transaction) -> RResult<u32> {
        let mut stmt = tx.prepare_cached(
            "SELECT version FROM version WHERE id == 0")?;
        let r = stmt.query([])?.mapped(|r| r.get(0)).next().unwrap_or(Ok(0))?;
        Ok(r)
    }

    /// Returns the certd tag recorded in the database.
    fn db_certd_tag(&self, tx: &Transaction) -> Result<Option<cert_d::Tag>> {
        let mut stmt = tx.prepare_cached(
            "SELECT tag FROM certd_tag WHERE id == 0")?;
        let r =
            stmt.query([])?.mapped(|r| Ok(db2tag(r.get(0)?)))
           .next().transpose()?;
        Ok(r)
    }

    /// Sets the certd tag recorded in the database.
    fn db_set_certd_tag(&self, tx: &Transaction, t: cert_d::Tag)
                        -> std::result::Result<(), RError>
    {
        if let Ok(mut stmt) = tx.prepare_cached(
            "INSERT OR REPLACE INTO certd_tag (id, tag) \
             VALUES (0, ?1)")
        {
            ignore_sqlite_busy(stmt.execute(params![tag2db(t)]))?;
        }
        Ok(())
    }

    /// Inserts the raw cert into the database and in-core cache.
    ///
    /// Only updates the database as long as `*commit` is true, and
    /// will clear the flag if writing to the database fails because
    /// the database is locked for writing.
    fn insert_rawcert(&self,
                      tx: &Transaction,
                      commit: &mut bool,
                      cert: RawCert<'a>,
                      tag: cert_d::Tag)
                      -> Result<()>
    {
        let fp = cert.fingerprint();
        self.db_insert(
            tx,
            commit,
            tag,
            &fp,
            Box::new(cert.keys().map(|sk| sk.fingerprint())),
            Box::new(cert.userids()))?;

        // Now insert it into the cache.
        self.certs_cache.write().unwrap()
            .insert(fp, (Arc::new(LazyCert::from(cert)), tag));
        Ok(())
    }

    /// Inserts the cert into the database and in-core cache.
    ///
    /// `cert_tag` is the certificate's tag.
    ///
    /// `certd_tag` is a tuple of the certd's tag prior to the
    /// insertion, and after the insertion, if known.
    fn insert_cert(&self, cert: Arc<LazyCert<'a>>, cert_tag: cert_d::Tag,
                   certd_tags: Option<(cert_d::Tag, cert_d::Tag)>)
                   -> Result<()>
    {
        tracer!(TRACE, "CertD::insert_cert");

        let fp = cert.fingerprint();

        let mut db = self.conn.lock().unwrap();
        let mut set_in_memory_tag = None;

        if let Ok(tx) = Transaction::new_unchecked(
            &db.conn, TransactionBehavior::Immediate)
        {
            let mut commit = true;
            self.db_insert(
                &tx,
                &mut commit,
                cert_tag,
                &fp,
                Box::new(cert.keys().map(|skb| skb.fingerprint())),
                Box::new(cert.userids().map(|uidb| uidb.clone())))?;

            if let Some((pre, post)) = certd_tags {
                if let Ok(Some(db_certd_tag)) = self.db_certd_tag(&tx) {
                    if pre == db_certd_tag {
                        // If we fail to update the certd tag, nothing
                        // bad will happen: the next time we check the
                        // tag, we'll notice that it's out of date.
                        let _ = self.db_set_certd_tag(&tx, post);
                        t!("DB tag matched cert-d tag, rolling forward");

                        if Some(pre) == db.in_memory_tag {
                            set_in_memory_tag = Some(post);
                            t!("In-memory tag matched db's cert-d tag, \
                                rolling forward");
                        } else {
                            t!("In-memory tag ({:?}) did NOT match db's \
                                cert-d tag ({:?}), NOT rolling forward \
                                (to {:?})",
                               db.in_memory_tag, pre, post);
                        }
                    } else {
                        t!("DB tag ({:?}) did NOT match cert-d tag ({:?}), \
                            NOT rolling forward (to {:?})",
                           db_certd_tag, pre, post);
                    }
                }
            }

            // Commit so that we release the exclusive lock on the
            // database before acquiring the exclusive lock on the cache.
            if commit {
                tx.commit()?;
            } else {
                tx.rollback()?;
            }
        }
        if let Some(set_in_memory_tag) = set_in_memory_tag {
            db.in_memory_tag = Some(set_in_memory_tag);
        }
        drop(db);

        // Now insert it into the cache.
        self.certs_cache.write().unwrap().insert(fp, (cert, cert_tag));
        Ok(())
    }

    /// Inserts information about a cert (subkey fingerprints and user
    /// IDs) into the database.
    ///
    /// Only updates the database as long as `*commit` is true, and
    /// will clear the flag if writing to the database fails because
    /// the database is locked for writing.
    fn db_insert<'i>(&self,
                     tx: &Transaction,
                     commit: &mut bool,
                     tag: cert_d::Tag,
                     cert_fp: &Fingerprint,
                     keys: Box<dyn Iterator<Item = Fingerprint> + 'i>,
                     userids: Box<dyn Iterator<Item = UserID> + 'i>)
                     -> Result<()>
    {
        let keyid = openpgp::KeyID::from(cert_fp);
        let cert_fp_db = fingerprint2db(cert_fp);

        // First, insert the cert and get its row id.
        *commit = *commit && {
            let mut stmt = tx.prepare_cached(
                "INSERT OR REPLACE INTO certs (tag, keyid, fingerprint) \
                 VALUES (?1, ?2, ?3)")?;
            ignore_sqlite_busy(stmt.execute(params![
                tag2db(tag), keyid2db(&keyid), &cert_fp_db]))?
        };

        // Now insert the keys.  Note that this doesn't remove
        // entries for subkeys that have been stripped from the
        // certificate.
        let mut stmt = tx.prepare_cached(
            "INSERT OR IGNORE INTO keys \
             (cert_fingerprint, keyid, fingerprint) \
             VALUES (?1, ?2, ?3)")?;

        // And, also update the in-memory index.
        let mut index = self.index_cache.write().unwrap();

        for subkey_fp in keys {
            let keyid = openpgp::KeyID::from(&subkey_fp);
            *commit = *commit &&
                ignore_sqlite_busy(stmt.execute(params![
                    &cert_fp_db, keyid2db(&keyid),
                    fingerprint2db(&subkey_fp)]))?;

            insert_unique(index.by_key_fingerprint.entry(subkey_fp).or_default(),
                          Cow::Borrowed(cert_fp));
            insert_unique(index.by_key_id.entry(keyid).or_default(),
                          Cow::Borrowed(cert_fp));
        }

        // Now insert the user IDs.  Note that this doesn't remove
        // entries for user IDs that have been stripped from the
        // certificate.
        let mut by_userid = index.by_userid.inner_mut();
        for uid in userids {
            let uidstr = String::from_utf8_lossy(uid.value()).to_string();
            let email = uid.email_normalized().ok().flatten();
            by_userid.insert(cert_fp.clone(), uid, email.clone());

            *commit = *commit && {
                let domain =
                    email.as_ref().and_then(|e| e.rfind("@").map(|i| &e[i + 1..]));

                let mut stmt = tx.prepare_cached(
                    "INSERT OR IGNORE INTO userids \
                     (cert_fingerprint, userid, email, domain) \
                     VALUES (?1, ?2, ?3, ?4)")?;
                ignore_sqlite_busy(stmt.execute(params![
                    &cert_fp_db, uidstr, email, domain]))?
            };
        }
        drop(by_userid);
        drop(index);

        Ok(())
    }

    /// Initializes the version 1 persistent indices.
    ///
    /// Any existing content is lost.
    fn initialize_v1(tx: &Transaction) -> std::result::Result<(), RError> {
        tx.execute_batch("
-- A table identifying the version and a human-readable magic.
DROP TABLE IF EXISTS version;
CREATE TABLE version (
    id INTEGER PRIMARY KEY,
    version INTEGER NOT NULL,
    comment TEXT NOT NULL
);

-- Record the schema version.
INSERT OR IGNORE INTO version VALUES (0, 1, 'sequoia cert cache v1');

-- Tag of the certd directory to quickly detect whether the db is stale.
DROP TABLE IF EXISTS certd_tag;
CREATE TABLE certd_tag (
    id INTEGER PRIMARY KEY,
    tag INTEGER NOT NULL
);

-- A table for all the certificates.
DROP TABLE IF EXISTS certs;
CREATE TABLE certs (
    tag INTEGER NOT NULL,
    keyid TEXT NOT NULL,
    fingerprint TEXT PRIMARY KEY NOT NULL
);

CREATE INDEX IF NOT EXISTS certs_keyid
    ON certs (keyid);

-- A mapping from subkey key IDs and fingerprints to certs.
DROP TABLE IF EXISTS keys;
CREATE TABLE keys (
    cert_fingerprint TEXT NOT NULL,
    keyid TEXT NOT NULL,
    fingerprint TEXT NOT NULL,
    FOREIGN KEY(cert_fingerprint) REFERENCES certs(fingerprint) ON DELETE CASCADE,
    UNIQUE(cert_fingerprint, fingerprint)
);

CREATE INDEX IF NOT EXISTS keys_fingerprint
    ON keys (fingerprint);
CREATE INDEX IF NOT EXISTS keys_keyid
    ON keys (keyid);

-- A mapping from user IDs to certs.
DROP TABLE IF EXISTS userids;
CREATE TABLE userids (
    cert_fingerprint TEXT NOT NULL,
    userid TEXT NOT NULL,
    email TEXT,
    domain TEXT,
    FOREIGN KEY(cert_fingerprint) REFERENCES certs(fingerprint) ON DELETE CASCADE,
    UNIQUE(cert_fingerprint, userid)
);

CREATE INDEX IF NOT EXISTS userids_userid
    ON userids (userid);
CREATE INDEX IF NOT EXISTS userids_email
    ON userids (email);
CREATE INDEX IF NOT EXISTS userids_domain
    ON userids (domain);
")?;

        Ok(())
    }

    /// Loads the cert identified by `fp` either from the cache or the
    /// disk.
    ///
    /// This takes the `CertD::certs_cache` and `CertD::index_cache`
    /// locks.  The caller must be careful to not hold them.
    fn load(&self, fp: &Fingerprint) -> Result<Arc<LazyCert<'a>>> {
        tracer!(TRACE, "CertD::load");
        let fp_name = format!("{:x}", fp);

        // We want to retry once.  See below for the explanation.
        let mut tries = 1;
        'retry: loop {
            let transaction = || {
                let fh = self.certd.get_file(&fp_name)?
                    .ok_or_else(|| StoreError::NotFound(fp.into()))?;
                let disk_tag = cert_d::Tag::try_from(&fh)?;
                Ok((fh, disk_tag))
            };

            let (fh, disk_tag) = transaction().map_err(|e: anyhow::Error| {
                // Evict the entry from the cache.
                self.certs_cache.write().unwrap().remove(fp);
                e
            })?;

            // Happy path: look into the cache without acquiring a write
            // lock and see if the tag is current.
            let mut initial_cached_tag = None;
            if let Some((cached_cert, cached_tag))
                = self.certs_cache.read().unwrap().get(fp)
            {
                if cached_tag == &disk_tag {
                    t!("{} loaded from cache on happy path", fp);
                    return Ok(cached_cert.clone());
                }
                initial_cached_tag = Some(cached_tag.clone());
            }

            // Slow path: either the cert was not present or outdated.
            // Load it from disk, and update the cache after the I/O is
            // done.
            t!("{} outdated in cache, loaded from disk", fp);
            let raw = br::File::new_with_cookie(
                fh, self.certd.get_path(&fp_name)?, Default::default())
                .map_err(Into::into)
                .and_then(RawCert::from_buffered_reader)
                .map_err(|e| {
                    t!("Failed to load cert: {}", e);
                    e
                })?;

            let update_index = |raw: &RawCert| {
                let mut index = self.index_cache.write().unwrap();

                for subkey in raw.keys() {
                    let subkey_fp = subkey.fingerprint();
                    let keyid = openpgp::KeyID::from(&subkey_fp);
                    insert_unique(
                        index.by_key_fingerprint.entry(subkey_fp).or_default(),
                        Cow::Borrowed(fp));
                    insert_unique(
                        index.by_key_id.entry(keyid).or_default(),
                        Cow::Borrowed(fp));
                }
                index.by_userid.insert(fp, raw.userids());
                drop(index);
            };

            // After we acquire the `CertD::certs_cache` lock, we need
            // to resolve a potential race.
            //
            // Let's say we have two threads, A and B.  Both threads
            // want to load the same certificate.  At some point, an
            // external task, E, may modify the on-disk certificate
            // changing it from C to C'.  At the end, C' should be
            // stored in the cache; C must not overwrite C'.
            //
            // - Scenario #0: No mutation, no conflict.  This is fine.
            //
            //   1. A: Read C
            //   2. B: Read C
            //   3. A: Insert C
            //   4. B: Insert C
            //
            //   (Steps 1 & 2, and/or steps 3 & 4 can be reversed
            //   without changing the analysis.)
            //
            // - Scenario #1: Mutation after A and B read C.  This is
            //   fine: the next time some thread reads C, they will
            //   update the cache.
            //
            //   1. A: Read C
            //   2. B: Read C
            //   3. E: Modifies C -> C'
            //   4. A: Insert C    <- Safe.
            //   5. B: Insert C    <- Safe.
            //
            //   (Steps 1 & 2, and/or steps 4 & 5 can be reversed
            //   without changing the analysis.)
            //
            // - Scenario #2.A: A reads C.  C is mutated.  B reads C'.
            //   A inserts C into the cache.  Then B inserts C' into
            //   the cache.  This is fine: the latest version is
            //   stored in the cache.
            //
            //   1. A: Read C
            //   2. E: Modifies C -> C'
            //   3. B: Read C'
            //   4. A: Insert C    <- Safe.
            //   5. B: Insert C'   <- Safe.
            //
            // - Scenario #2.B: B reads C.  C is mutated.  A reads C'.
            //   B inserts C into the cache.  Then A inserts C' into the
            //   cache.  This is fine: the latest version is stored in
            //   the cache.
            //
            //   1. B: Read C
            //   2. E: Modifies C -> C'
            //   3. A: Read C'
            //   4. B: Insert C    <- Safe.
            //   5. A: Insert C'   <- Safe.
            //
            // - Scenario #3.A: A reads C.  C is mutated.  B reads C',
            //   and inserts it into the cache.  A inserts C into the
            //   cache overwriting C'.  STEP 5 IS A DISASTER.
            //
            //   1. A: Read C
            //   2. E: Modifies C -> C'
            //   3. B: Read C'
            //   4. B: Insert C'   <- Safe.
            //   5. A: Insert C    <- DOOM.
            //
            // - Scenario #3.B: B reads C.  C is mutated.  A reads C',
            //   and inserts it into the cache.  B inserts C into the
            //   cache overwriting C'.  STEP 5 IS A DISASTER.
            //
            //   1. B: Read C
            //   2. E: Modifies C -> C'
            //   3. A: Read C'
            //   4. A: Insert C'   <- Safe.
            //   5. B: Insert C    <- DOOM.
            //
            // To prevent scenario #3, we can do the following: after
            // reacquiring the lock, but before inserting the
            // certificate into the cache, the thread checks that:
            //
            //   - The tag in the cache (`new_cached_tag`) matches
            //     what it originally saw in the cache
            //     (`initial_cached_tag`)
            //
            //     If this is the case, then no one else updated the
            //     in-memory cache, so it is safe to update it.
            //
            //   - OR, the tag in the cache (`new_cached_tag`) matches
            //     what it originally saw on disk (`disk_tag`)
            //
            //     If this is the case, then someone else updated the
            //     in-memory cache with the same update the thread
            //     wants to make, so it is safe to return what's in
            //     the in-memory cache.
            //
            // If neither hold, then the in-memory cache was updated
            // to a different value, and the current thread doesn't
            // know if it has C or C'.
            //
            // Unfortunately, this also means that these conditions
            // hold for step 2.A.5 and step 2.B.5.
            //
            // As such we can use the following mitigation to check
            // that the current thread is NOT at step 3.A.5 or step
            // 3.B.5:
            //
            // ```text
            // let safe = new_cached_tag == initial_cached_tag
            //     || new_cached_tag == disk_tag;
            // ```
            //
            // If this condition holds, we simply retry.  If we race
            // again, then we we prefer to avoid live lock.
            match self.certs_cache.write().unwrap().entry(fp.clone()) {
                Entry::Occupied(mut e) => {
                    let (cached_cert, new_cached_tag) = e.get();
                    if &disk_tag == new_cached_tag {
                        // Another thread updated the in-memory cache
                        // with what we expect.  Use the in-memory
                        // cache.
                        t!("{} lost race: loading from cache", fp);
                        return Ok(cached_cert.clone());
                    } else if initial_cached_tag.as_ref() == Some(new_cached_tag) {
                        // The in-memory cache was not updated.
                        // Insert what we read.
                        t!("{}: won race: updating index", fp);
                        update_index(&raw);
                        let cert = Arc::new(LazyCert::from(raw));
                        e.insert((cert.clone(), disk_tag));
                        return Ok(cert);
                    } else if tries > 0 {
                        // The tag changed.  We don't know which tag
                        // is newer, so we redo the lookup (there
                        // could have been two quick updates, we got
                        // the old tag and cert, the other thread that
                        // won the race got the new tag and cert).  In
                        // the likely case that the cached_tag is the
                        // current version, the next iteration will
                        // take the happy path.
                        t!("{} tag mismatch, retrying lookup once", fp);
                        tries -= 1;
                        continue 'retry;
                    } else {
                        // There's high contention.
                        //
                        // To break possible live lock, take one.
                        t!("{} tag mismatch, avoiding live lock", fp);
                        update_index(&raw);
                        let cert = Arc::new(LazyCert::from(raw));
                        e.insert((cert.clone(), disk_tag));
                        return Ok(cert);
                    }
                },
                Entry::Vacant(e) => {
                    t!("{} not in cache, loaded from disk", fp);
                    update_index(&raw);
                    let cert = Arc::new(LazyCert::from(raw));
                    e.insert((cert.clone(), disk_tag));
                    return Ok(cert);
                },
            }
        }
    }
}

/// Serializes a Fingerprint for storage in the database.
fn fingerprint2db(fp: &Fingerprint) -> String {
    format!("{:x}", fp)
}

/// Serializes a Key ID for storage in the database.
fn keyid2db(id: &KeyID) -> String {
    format!("{:x}", id)
}

/// Converts a `Tag` into something representable by SQLite.
fn tag2db(t: cert_d::Tag) -> i64 {
    u64::from(t) as i64
}

/// Converts a `Tag` from something representable by SQLite.
fn db2tag(t: i64) -> cert_d::Tag {
    (t as u64).into()
}

/// Ignores SQLITE_BUSY.
///
/// Returns `Ok(true)` if the update was successful, `Ok(false)` if
/// the database was locked for writing, and any other error as-is.
///
/// This can be used to implement stricter transaction semantics than
/// SQLite offers.  If you have a SQLite transaction in deferred
/// (i.e. read) mode, and try to update a table, the update fails with
/// SQLITE_BUSY, but that doesn't invalidate the transaction.  Hence,
/// a second update in the same transaction may succeed.  But that is
/// not what we want, we want either all updates to succeed, or none.
///
/// # Examples
///
/// ```rust
/// # use rusqlite::*;
/// # fn ignore_sqlite_busy<T>(r: Result<T>) -> Result<bool> { todo!() }
/// # fn f() -> anyhow::Result<()> {
/// let conn = Connection::open_in_memory()?;
/// let tx = Transaction::new_unchecked(
///     &conn, TransactionBehavior::Deferred)?;
/// let mut commit = true;
///
/// commit = commit && {
///     let mut stmt = tx.prepare_cached(
///         "CREATE TABLE foo")?;
///     ignore_sqlite_busy(stmt.execute(params![]))?
/// };
///
/// if commit {
///     tx.commit()?;
/// } else {
///     tx.rollback()?;
/// }
/// # Ok(()) }
/// ```
fn ignore_sqlite_busy<T>(r: rusqlite::Result<T>)
    -> rusqlite::Result<bool, RError>
{
    match r {
        Ok(_) => Ok(true),
        Err(e) => if is_sqlite_busy(&e) {
            Ok(false)
        } else {
            Err(e.into())
        },
    }
}

/// Tests if the error is SQLITE_BUSY.
fn is_sqlite_busy(e: &rusqlite::Error) -> bool {
    Some(rusqlite::ErrorCode::DatabaseBusy) == e.sqlite_error_code()
}

#[derive(Debug, thiserror::Error)]
enum RepeatError {
    #[error(transparent)]
    Db(#[from] RError),
    #[error(transparent)]
    Anyhow(#[from] anyhow::Error),
}

/// Repeat an operation as long as the database returns "busy."
fn repeat_while_busy<T, E>(f: impl Fn() -> std::result::Result<T, E>)
    -> Result<T>
where
    E: Into<RepeatError>,
{
    tracer!(TRACE, "repeat_while_busy");

    const REPEAT_DEADLINE: Duration = Duration::from_secs(5);
    const REPEAT_DELAY: Duration = Duration::from_millis(1);

    let start = Instant::now();

    loop {
        match f().map_err(Into::into) {
            err @ Err(RepeatError::Db(RError::SqliteFailure(e, _)))
                if e.code == ErrorCode::DatabaseBusy =>
            {
                if start.elapsed() > REPEAT_DEADLINE {
                    break err.with_context(|| {
                        format!("reached deadline {} ms waiting for DB \
                                 to become available",
                                REPEAT_DEADLINE.as_millis()
                        )
                    });
                }

                t!("database busy, re-trying");
                sleep(REPEAT_DELAY);
                continue;
            }
            Err(RepeatError::Db(err)) => break Err(err.into()),
            Err(RepeatError::Anyhow(err)) => break Err(err),
            Ok(r) => break Ok(r),
        }
    }
}

/// Asserts that the given vector contains at least one cert, or
/// returns the appropriate error.
fn ok_or_not_found<'a, K>(handle: K,
                          c: Vec<Arc<LazyCert<'a>>>)
                          -> Result<Vec<Arc<LazyCert<'a>>>>
where
    K: Into<KeyHandle>,
{
    if c.is_empty() {
        Err(crate::store::StoreError::NotFound(handle.into()).into())
    } else {
        Ok(c)
    }
}

impl<'a> Store<'a> for CertD<'a> {
    fn lookup_by_cert(&self, kh: &KeyHandle) -> Result<Vec<Arc<LazyCert<'a>>>> {
        match kh {
            KeyHandle::Fingerprint(fp) => Ok(vec![self.load(fp)?]),
            KeyHandle::KeyID(id) => {
                let mut db = self.conn.lock().unwrap();
                self.scan(&mut db)?;
                drop(db);

                let primaries =
                    self.index_cache.read().unwrap().by_key_id.get(id).cloned();
                // Note: it is of the utmost importance to compute
                // `primaries`, then drop the `CertD::index_cache`
                // lock, then map over self.load, because we must not
                // hold the lock over calls to self.load.  If this is
                // done on the right hand side of the `if let`
                // statement, the lock will be held across the call to
                // self.load.
                let mut matches = Vec::new();
                if let Some(primaries) = primaries {
                    matches = primaries.iter().filter_map(|fp| {
                        if &KeyID::from(fp) == id {
                            Some(self.load(fp))
                        } else {
                            None
                        }
                    }).collect::<Result<Vec<_>>>()?;
                }

                // We didn't find a certificate.
                ok_or_not_found(id, matches)
            },
        }
    }

    fn lookup_by_cert_or_subkey(&self, kh: &KeyHandle) -> Result<Vec<Arc<LazyCert<'a>>>> {
        tracer!(false, "lookup_by_cert_or_subkey");
        t!("{}", kh);

        let mut db = self.conn.lock().unwrap();
        self.scan(&mut db)?;
        drop(db);

        match kh {
            KeyHandle::Fingerprint(fp) => {
                let primaries = self.index_cache.read().unwrap()
                    .by_key_fingerprint.get(fp).cloned();
                // Note: it is of the utmost importance to compute
                // `primaries`, then drop the `CertD::index_cache`
                // lock, then map over self.load, because we must not
                // hold the lock over calls to self.load.  If this is
                // done on the right hand side of the `if let`
                // statement, the lock will be held across the call to
                // self.load.
                if let Some(primaries) = primaries {
                    t!("In-memory cache hit!");
                    return primaries.iter().map(|fp| self.load(fp)).collect();
                }
            }
            KeyHandle::KeyID(keyid) => {
                let primaries = self.index_cache.read().unwrap()
                    .by_key_id.get(keyid).cloned();
                // Note: it is of the utmost importance to compute
                // `primaries`, then drop the `CertD::index_cache`
                // lock, then map over self.load, because we must not
                // hold the lock over calls to self.load.  If this is
                // done on the right hand side of the `if let`
                // statement, the lock will be held across the call to
                // self.load.
                if let Some(primaries) = primaries {
                    t!("In-memory cache hit!");
                    return primaries.iter().map(|fp| self.load(fp)).collect();
                }
            }
        }
        t!("In-memory cache miss!");

        // We didn't find a certificate.
        ok_or_not_found(kh.clone(), Vec::new())
    }

    fn select_userid(&self, params: &UserIDQueryParams, pattern: &str)
        -> Result<Vec<Arc<LazyCert<'a>>>>
    {
        let mut db = self.conn.lock().unwrap();
        self.scan(&mut db)?;
        drop(db);

        tracer!(false, "CertD::select_userid");
        t!("params: {:?}, pattern: {:?}", params, pattern);

        let matches = self.index_cache.read().unwrap()
            .by_userid.select_userid(params, pattern)?;
        // Note: it is of the utmost importance to compute `matches`,
        // then drop the `CertD::index_cache` lock, then map over
        // self.load, because we must not hold the lock over calls to
        // self.load.
        let matches = matches
            .into_iter()
            .map(|fp| self.load(&fp))
            .collect::<Result<Vec<_>>>()?;

        Ok(matches)
    }

    fn fingerprints<'b>(&'b self) -> Box<dyn Iterator<Item=Fingerprint> + 'b> {
        let transaction = || -> Result<Box<dyn Iterator<Item=Fingerprint> + 'b>>
        {
            let mut db = self.conn.lock().unwrap();
            self.scan(&mut db)?;
            drop(db);

            let mut fps = self.index_cache.read().unwrap()
                .by_key_fingerprint.iter()
                .filter(|(sk_fp, cert_fps)|
                        cert_fps.binary_search(sk_fp).is_ok())
                .map(|(fp, _)| fp.clone())
                .collect::<Vec<_>>();
            fps.sort_unstable();
            fps.dedup();

            Ok(Box::new(fps.into_iter()))
        };

        match transaction() {
            Ok(v) => v,
            Err(_) => {
                // We cannot return errors here, fall back to
                // accessing the certd.
                Box::new(self.certd.fingerprints()
                         .filter_map(|fp| fp.ok().and_then(|s| s.parse().ok())))
            },
        }
    }

    fn certs<'b>(&'b self)
        -> Box<dyn Iterator<Item=Arc<LazyCert<'a>>> + 'b>
        where 'a: 'b
    {
        self.prefetch_internal(true, &[], false);
        Box::new(
            self.certs_cache.read().unwrap().values()
                .map(|(cert, _)| cert.clone())
                .collect::<Vec<_>>()
                .into_iter())
    }

    fn prefetch_all(&self) {
        self.prefetch_internal(true, &[], true);
    }

    fn prefetch_some(&self, certs: &[KeyHandle]) {
        self.prefetch_internal(false, certs, true);
    }
}

impl<'a> CertD<'a> {
    /// Canonicalize the certs identified by `khs`, or `all`.
    fn prefetch_internal(&self,
                         all: bool,
                         khs: &[KeyHandle],
                         canonicalize: bool)
    {
        // LazyCert is Sync and Send, but keeping a reference to the
        // RawCerts in certs prevents us from later updating
        // certs.  This requires a bit of acrobatics to get
        // right.

        tracer!(TRACE, "CertD::prefetch_internal");
        if all {
            t!("Prefetch all certificates");
        } else {
            t!("Prefetch {} certificates", khs.len());
        }

        /// Work items.
        enum Work<'a> {
            /// Load the certificate from disk, canonicalize if
            /// `canonicalize` is given.
            Load(Fingerprint),

            /// Canonicalize a raw cert that is already in our cache.
            Canonicalize(Box<RawCert<'a>>, cert_d::Tag),
        }

        let mut work: Vec<Work> = Default::default();
        let certs_cache = self.certs_cache.read().unwrap();

        // First, schedule all raw certs in our cache that we
        // should canonicalize.
        if canonicalize {
            for w in certs_cache.iter()
                .filter_map(|(fpr, (cert, tag))| {
                    if ! cert.is_parsed() {
                        if all
                            || khs.iter()
                            .any(|kh| {
                                kh.aliases(&KeyHandle::from(fpr.clone()))
                            })
                        {
                            // Unfortunately we have to clone the bytes,
                            // because we cannot keep certs borrowed.
                            t!("Queuing {} to be prefetched", fpr);
                            use std::ops::Deref;
                            cert.deref().clone().into_raw_cert().ok()
                                .map(|raw_cert| Work::Canonicalize(Box::new(raw_cert), tag.clone()))
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                })
            {
                work.push(w);
            }
        }

        // Second, we consider all certs not in the cache.
        if all {
            // Either all of them.
            for fp in self.certd.fingerprints()
                .into_iter().flatten()
                .filter_map(|f| f.parse::<Fingerprint>().ok())
                .filter(|f| ! certs_cache.contains_key(f))
            {
                work.push(Work::Load(fp));
            }
        } else {
            // Or just those in `khs`.  Here, for simplicity, we
            // only consider the fingerprints.
            for fp in khs.iter()
                .filter_map(|kh| Fingerprint::try_from(kh).ok())
                .filter(|f| ! certs_cache.contains_key(f))
            {
                work.push(Work::Load(fp));
            }
        }

        // Release our read lock.
        drop(certs_cache);

        let cert_count = work.len();

        // Use not more threads than we have cores.  No point in
        // spawning a single thread, just do it inline.  Also, some
        // platforms don't support spawning threads.
        let mut threads = num_cpus::get();
        if threads <= 1 {
            threads = 0;
        } else if cert_count < 16 {
            // The keyring is small, limit the number of threads.
            threads = 2;
        } else {
            // Sort the certificates so they are ordered from most
            // packets to least.  More packets implies more work, and
            // this will hopefully result in a more equal distribution
            // of load.
            work.sort_unstable_by_key(|w| match w {
                Work::Load(_) => 0,
                Work::Canonicalize(raw_cert, _tag) =>
                    usize::MAX - raw_cert.count(),
            });
        }
        t!("Using {} threads", threads);

        let mut count = 0;

        let process_work = |w: Work| -> Option<(LazyCert, cert_d::Tag)> {
            match w {
                Work::Load(fp) => {
                    // Silently ignore errors.  This will be caught
                    // later when the caller looks this one up.
                    let file = self.certd.get_file(&fp.to_string())
                        .ok().flatten()?;
                    let tag = cert_d::Tag::try_from(&file).ok()?;
                    match br::File::new_with_cookie(
                        file,
                        self.certd.get_path(&fp.to_string()).ok()?,
                        Default::default())
                        .map_err(Into::<anyhow::Error>::into)
                        .and_then(|c| {
                            if canonicalize {
                                Ok(LazyCert::from(
                                    Cert::from_buffered_reader(c)?))
                            } else {
                                Ok(LazyCert::from(
                                    RawCert::from_buffered_reader(c)?))
                            }
                        })
                    {
                        Ok(cert) => Some((cert, tag)),
                        Err(err) => {
                            t!("Error loading {}: {}", fp, err);
                            None
                        },
                    }
                },
                Work::Canonicalize(raw, tag) => {
                    // Silently ignore errors.  This will be
                    // caught later when the caller looks this
                    // one up.
                    match Cert::try_from(&*raw) {
                        Ok(cert) =>
                            Some((LazyCert::from(cert), tag)),
                        Err(err) => {
                            t!("Parsing raw cert {}: {}",
                               raw.keyid(), err);
                            None
                        }
                    }
                },
            }
        };

        if threads == 0 {
            let mut certs = self.certs_cache.write().unwrap();
            for w in work {
                match &w {
                    Work::Load(fp) =>
                        t!("Inline loading{} {}",
                           if canonicalize {
                               " + canonicalizing"
                           } else {
                               ""
                           },
                           fp),
                    Work::Canonicalize(raw, _) =>
                        t!("Inline canonicalizing {}!",
                           raw.keyid()),
                }
                if let Some((cert, tag)) = process_work(w) {
                    let fpr = cert.fingerprint();
                    t!("Caching {}", fpr);
                    certs.insert(fpr, (Arc::new(cert), tag));
                    count += 1;
                }
            }
        } else {
            use crossbeam::thread;
            use crossbeam::channel::unbounded as channel;

            // Avoid an extra level of indentation.
            let result = thread::scope(|thread_scope| {
            // A communication channel for sending work to the workers.
            let (work_tx, work_rx) = channel();
            // A communication channel for returning results to the
            // main thread.
            let (results_tx, results_rx) = channel();

            let mut threads_extant = Vec::new();

            for cert in work.into_iter() {
                if threads_extant.len() < threads {
                    let tid = threads_extant.len();
                    t!("Starting thread {} of {}",
                       tid, threads);

                    let mut work = Some(Ok(cert));

                    // The thread's state.
                    let work_rx = work_rx.clone();
                    let results_tx = results_tx.clone();

                    threads_extant.push(thread_scope.spawn(move |_| {
                        loop {
                            match work.take().unwrap_or_else(|| work_rx.recv()) {
                                Err(_) => break,
                                Ok(w) => {
                                    match &w {
                                        Work::Load(fp) =>
                                            t!("Thread {} loading{} {}",
                                               tid,
                                               if canonicalize {
                                                   " + canonicalizing"
                                               } else {
                                                   ""
                                               },
                                               fp),
                                        Work::Canonicalize(raw, _) =>
                                            t!("Thread {} canonicalizing {}!",
                                               tid, raw.keyid()),
                                    }
                                    if let Some(r) = process_work(w) {
                                        let _ = results_tx.send(r);
                                    }
                                },
                            }
                        }

                        t!("Thread {} exiting", tid);
                    }));
                } else {
                    work_tx.send(cert).unwrap();
                }
            }

            // When the threads see this drop, they will exit.
            drop(work_tx);

            // Drop our reference to results_tx.  When the last thread
            // exits, the last reference will be dropped and the loop
            // below will exit.
            drop(results_tx);

            let mut certs = self.certs_cache.write().unwrap();
            while let Ok((cert, tag)) = results_rx.recv() {
                let fpr = cert.fingerprint();
                t!("Caching {}", fpr);
                certs.insert(fpr, (Arc::new(cert), tag));
                count += 1;
            }
            }); // thread scope.

            // We're just caching results so we can ignore errors.
            if let Err(err) = result {
                t!("{:?}", err);
            }
        }

        t!("Prefetched {} certificates, ({} had errors)",
           count, cert_count - count);
    }
}

impl<'a> StoreUpdate<'a> for CertD<'a> {
    fn update_by(&self, cert: Arc<LazyCert<'a>>,
                 merge_strategy: &dyn MergeCerts<'a>)
        -> Result<Arc<LazyCert<'a>>>
    {
        tracer!(TRACE, "CertD::update_by");
        t!("Inserting {}", cert.fingerprint());

        // This is slightly annoying: cert-d expects bytes.  But
        // serializing cert is a complete waste if we have to merge
        // the certificate with another one.  cert-d actually only
        // needs the primary key, which it uses to derive the
        // fingerprint, so, we only serialize that.
        let fpr = cert.fingerprint();
        let fpr_str = format!("{:x}", fpr);

        let mut merged = None;

        let mut pre_certd_tag = None;
        let mut post_certd_tag = None;
        let (cert_tag, _) = self.certd.insert_extended(
            &fpr_str, (), false,
            |_| {
                // Compute the tag with the cert-d lock held.
                pre_certd_tag = Some(self.certd.tag());
                Ok(())
            },
            |(), disk_bytes| {
                let disk: Option<Arc<LazyCert>>
                    = if let Some(disk_bytes) = disk_bytes
                {
                    let mut parser = RawCertParser::from_bytes(disk_bytes)
                        .with_context(|| {
                            format!("Parsing {} as returned from the cert directory",
                                    fpr)
                        })
                        .map_err(|err| {
                            t!("Reading disk version: {}", err);
                            err
                        })?;
                    let disk = parser.next().transpose()
                        .with_context(|| {
                            format!("Parsing {} as returned from the cert directory",
                                    fpr)
                        })
                        .map_err(|err| {
                            t!("Parsing disk version: {}", err);
                            err
                        })?;
                    disk.map(|disk| Arc::new(LazyCert::from(disk)))
                } else {
                    t!("No disk version");
                    None
                };

                let merged_ = merge_strategy.merge_public(cert, disk)
                    .with_context(|| {
                        format!("Merging versions of {}", fpr)
                    })
                    .map_err(|err| {
                        t!("Merging: {}", err);
                        err
                    })?;
                let bytes = merged_.to_vec()?;
                merged = Some(merged_);
                Ok(cert_d::MergeResult::Data(bytes))
            },
            |r| {
                // Compute the tag with the cert-d lock held.
                post_certd_tag = Some(self.certd.tag());
                Ok(r)
            },
        )?;

        let certd_tags = if let (Some(pre), Some(post))
            = (pre_certd_tag, post_certd_tag)
        {
            Some((pre, post))
        } else {
            None
        };

        let merged = merged.expect("set");
        self.insert_cert(merged.clone(), cert_tag, certd_tags)?;
        Ok(merged)
    }
}

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

    use std::sync::OnceLock;
    use anyhow::Context;

    use openpgp::packet::UserID;
    use openpgp::serialize::Serialize;
    use openpgp::serialize::SerializeInto;

    use crate::store::StoreError;
    use crate::tests::print_error_chain;
    const N: usize = 1050;

    fn roughly_a_thousand_certs() -> &'static [Cert] {
        static CERTS: OnceLock<Vec<Cert>> = OnceLock::new();
        CERTS.get_or_init(
            || (0..N).into_iter().map(|i| {
                let userid = format!("<{}@example.org>", i);

                let (cert, _rev) = CertBuilder::new()
                    .set_cipher_suite(CipherSuite::Cv25519)
                    .add_userid(UserID::from(&userid[..]))
                    .add_storage_encryption_subkey()
                    .generate()
                    .expect("ok");
                cert
            }).collect()).as_slice()
    }

    /// Checks that the CertD at `path`
    fn check_certd(certd: &CertD, certs: &[&Cert],
                   certs_fpr: Option<&[Fingerprint]>,
                   subkeys_fpr: &[Fingerprint],
                   userids: &[&UserID]) -> Result<()> {
        // Test Store::iter.  In particular, make sure we get
        // everything back.
        if let Some(certs_fpr) = certs_fpr {
            let mut certs_read = certd.certs().collect::<Vec<_>>();
            assert_eq!(
                certs_read.len(), certs.len(),
                "Looks like you're exhausting the available file descriptors");

            certs_read.sort_by_key(|c| c.fingerprint());
            let certs_read_fpr
                = certs_read.iter().map(|c| c.fingerprint()).collect::<Vec<_>>();
            assert_eq!(certs_fpr, certs_read_fpr);
        }

        // Test Store::by_cert.
        for cert in certs.iter() {
            let certs_read = certd.lookup_by_cert(&cert.key_handle()).expect("present");
            // We expect exactly one cert.
            assert_eq!(certs_read.len(), 1);
            let cert_read = certs_read.iter().next().expect("have one")
                .to_cert().expect("valid");
            assert_eq!(&cert_read, cert);
        }

        for subkey in subkeys_fpr.iter() {
            let kh = KeyHandle::from(subkey.clone());
            match certd.lookup_by_cert(&kh) {
                Ok(certs) => panic!("Expected nothing, got {} certs", certs.len()),
                Err(err) => {
                    if let Some(&StoreError::NotFound(ref got))
                        = err.downcast_ref::<StoreError>()
                    {
                        assert_eq!(&kh, got);
                    } else {
                        panic!("Expected NotFound, got: {}", err);
                    }
                }
            }
        }

        // Test Store::lookup_by_cert_or_subkey.
        for fpr in certs.iter().map(|cert| cert.fingerprint())
            .chain(subkeys_fpr.iter().cloned())
        {
            let certs_read
                = certd.lookup_by_cert_or_subkey(&KeyHandle::from(fpr.clone())).expect("present");
            // We expect exactly one cert.
            assert_eq!(certs_read.len(), 1);
            let cert_read = certs_read.iter().next().expect("have one")
                .to_cert().expect("valid");

            assert!(cert_read.keys().any(|k| k.key().fingerprint() == fpr));
        }

        // Test Store::lookup_by_userid.
        for userid in userids.iter() {
            let certs_read
                = certd.lookup_by_userid(&userid).expect("present");
            // We expect exactly one cert.
            assert_eq!(certs_read.len(), 1);
            let cert_read = certs_read.iter().next().expect("have one")
                .to_cert().expect("valid");

            assert!(cert_read.userids().any(|u| &u.userid() == userid));
        }
        Ok(())
    }

    // Make sure that we can read a huge cert-d.  Specifically, the
    // typical file descriptor limit is 1024.  Make sure we can
    // initialize and iterate over a cert-d with a few more entries
    // than that.
    #[test]
    fn huge_cert_d() -> Result<()> {
        let path = tempfile::tempdir()?;
        let certd = cert_d::CertD::with_base_dir(&path)
            .map_err(|err| {
                let err = anyhow::Error::from(err)
                    .context(format!("While opening the certd {:?}", path));
                print_error_chain(&err);
                err
            })?;

        // Generate some certificates and write them to a cert-d using
        // the low-level interface.

        let mut certs = Vec::new();
        let mut certs_fpr = Vec::new();
        let mut subkeys_fpr = Vec::new();
        let mut userids = Vec::new();

        for cert in roughly_a_thousand_certs() {
            certs_fpr.push(cert.fingerprint());
            subkeys_fpr.extend(cert.keys().subkeys().map(|ka| ka.key().fingerprint()));
            userids.push(cert.userids().next().unwrap().userid());

            let mut bytes = Vec::new();
            cert.serialize(&mut bytes).expect("can serialize to a vec");
            certd
                .insert_data(&bytes, false, |new, disk| {
                    assert!(disk.is_none());

                    Ok(cert_d::MergeResult::DataRef(new))
                })
                .with_context(|| {
                    format!("{:?} ({})", path, cert.fingerprint())
                })
                .expect("can insert");

            certs.push(cert);
        }

        // One subkey per certificate.
        assert_eq!(certs_fpr.len(), subkeys_fpr.len());

        certs_fpr.sort();

        // Open the cert-d and make sure we can read what we wrote via
        // the low-level interface.
        let certd = CertD::open(&path).expect("exists");

        check_certd(
            &certd, &certs[..], Some(&certs_fpr), &subkeys_fpr, &userids)
    }

    /// Tests four concurrent mutators on the same CertD.
    #[test]
    fn concurrent_mutators() -> Result<()> {
        use rand::prelude::*;
        let tmpdir = tempfile::tempdir()?;
        let path = tmpdir.path();
        let certd_shared = CertD::open(path)?;
        let roughly_a_hundred_certs = &roughly_a_thousand_certs()[..101];

        std::thread::scope(|s| {
            for tid in 0..3 {
                let certd_shared = &certd_shared;

                s.spawn(move || {
                    // Thread 0 has its own CertD, thread 1 and 2 get
                    // to share one.
                    let certd_owned;
                    let certd = if tid % 3 == 0 {
                        certd_owned = CertD::open(path).unwrap();
                        &certd_owned
                    } else {
                        certd_shared
                    };

                    let mut rng = rand::thread_rng();
                    let mut source_certs = roughly_a_hundred_certs.to_vec();
                    source_certs.shuffle(&mut rng);
                    let mut certs = Vec::new();
                    let mut certs_fpr = Vec::new();
                    let mut subkeys_fpr = Vec::new();
                    let mut userids = Vec::new();

                    for cert in source_certs.iter() {
                        eprint!("{}", tid);
                        certs_fpr.push(cert.fingerprint());
                        subkeys_fpr.extend(cert.keys().subkeys().map(|ka| ka.key().fingerprint()));
                        userids.push(cert.userids().next().unwrap().userid());

                        certd.update(Arc::new(cert.clone().into()))
                            .expect("can insert");

                        certs.push(cert);

                        check_certd(
                            &certd, &certs[..], None, &subkeys_fpr, &userids)
                            .expect("consistency and robustness");
                    }
                });
            }

            // Finally, add a fourth mutator that doesn't update the
            // database.
            s.spawn(move || {
                let certd = cert_d::CertD::with_base_dir(path).unwrap();
                for cert in roughly_a_hundred_certs {
                    std::thread::sleep(Duration::from_millis(10));
                    eprint!(".");
                    certd.insert_data(
                        &cert.to_vec().unwrap(), false,
                        |new, _disk| Ok(cert_d::MergeResult::DataRef(new)))
                        .with_context(|| {
                            format!("{:?} ({})", path, cert.fingerprint())
                        })
                        .expect("can insert");
                }
            });
        });

        Ok(())
    }

    /// Tests that the database index preserves all the information.
    #[test]
    fn database_index() -> Result<()> {
        let tmpdir = tempfile::tempdir()?;
        let p = tmpdir.path();

        let c = CertBuilder::new()
            .add_userid("Alice Lovelace")
            .add_userid("<alice@lovelace.name>")
            .add_userid("Alice Lovelace <alice@lovelace.name>")
            .add_signing_subkey()
            .generate()?.0;

        // Insert the cert using the low-level cert-d.
        {
            let certd = cert_d::CertD::with_base_dir(p)?;
            certd.insert_data(
                &c.to_vec()?, false,
                |new, _disk| Ok(cert_d::MergeResult::DataRef(new)))?;
        }

        // Construct the database index.
        let _ = CertD::open(p)?;

        // Do the queries.  Construct a new CertD for each query so
        // that we don't get the benefit of updating the in-core index
        // when loading a cert from disk.
        for (i, k) in c.keys().enumerate() {
            let primary = i == 0;

            if primary {
                let d = CertD::open(p)?.lookup_by_cert(&k.key().fingerprint().into())?;
                assert_eq!(&c, d[0].to_cert()?);
                let d = CertD::open(p)?.lookup_by_cert(&k.key().keyid().into())?;
                assert_eq!(&c, d[0].to_cert()?);
            }

            let d = CertD::open(p)?.lookup_by_cert_or_subkey(&k.key().fingerprint().into())?;
            assert_eq!(&c, d[0].to_cert()?);
            let d = CertD::open(p)?.lookup_by_cert_or_subkey(&k.key().keyid().into())?;
            assert_eq!(&c, d[0].to_cert()?);
        }

        for u in c.userids() {
            let d = CertD::open(p)?.select_userid(
                UserIDQueryParams::new()
                    .set_anchor_start(true)
                    .set_anchor_end(true)
                    .set_email(false)
                    .set_ignore_case(false),
                &String::from_utf8(u.userid().value().to_vec())?)?;
            assert_eq!(&c, d[0].to_cert()?);

            if let Ok(Some(e)) = u.userid().email() {
                let d = CertD::open(p)?.select_userid(
                    UserIDQueryParams::new()
                        .set_anchor_start(true)
                        .set_anchor_end(true)
                        .set_email(true)
                        .set_ignore_case(false),
                    e)?;
                assert_eq!(&c, d[0].to_cert()?);
            }

            if let Ok(Some(n)) = u.userid().name() {
                let d = CertD::open(p)?.select_userid(
                    UserIDQueryParams::new()
                        .set_anchor_start(true)
                        .set_anchor_end(false)
                        .set_email(false)
                        .set_ignore_case(false),
                    n)?;
                assert_eq!(&c, d[0].to_cert()?);
            }
        }

        Ok(())
    }
}