pub fn recover<P: AsRef<Path>>(
        storage_directory: P,
    ) -> io::Result<(
        // Metadata writer
        MetadataStore,
        // Metadata - node id, value, user data
        Vec<UpdateMetadata>,
    )> {
        use fs2::FileExt;

        // TODO NOCOMMIT
        let sync_status = std::process::Command::new("sync")
            .status()
            .map(|status| status.success());

        if !matches!(sync_status, Ok(true)) {
            log::warn!(
                "sync command before recovery failed: {:?}",
                sync_status
            );
        }

        let path = storage_directory.as_ref();

        // initialize directories if not present
        if let Err(e) = fs::read_dir(path) {
            if e.kind() == io::ErrorKind::NotFound {
                fallible!(fs::create_dir_all(path));
            }
        }

        let _ = fs::File::create(path.join(WARN));

        let directory_lock = fallible!(fs::File::open(path));
        fallible!(directory_lock.sync_all());
        fallible!(directory_lock.try_lock_exclusive());

        let recovery =
            MetadataStore::recover_inner(&storage_directory, &directory_lock)?;

        let new_log = LogAndStats {
            log_sequence_number: recovery.id_for_next_log,
            bytes_written: 0,
            file: fallible!(fs::File::create(log_path(
                path,
                recovery.id_for_next_log
            ))),
        };

        let (tx, rx) = unbounded();

        let inner = Inner {
            snapshot_size: Arc::new(recovery.snapshot_size.into()),
            storage_directory: path.into(),
            directory_lock: Arc::new(directory_lock),
            global_error: Default::default(),
            active_log: Arc::new(Mutex::new(new_log)),
            worker_outbox: tx,
        };

        let worker_inner = inner.clone();

        let spawn_res = std::thread::Builder::new()
            .name("sled_flusher".into())
            .spawn(move || {
                worker(
                    rx,
                    recovery.id_for_next_log.checked_sub(1).unwrap(),
                    worker_inner,
                )
            });

        if let Err(e) = spawn_res {
            return Err(io::Error::other(format!(
                "unable to spawn metadata compactor thread for sled database: {:?}",
                e
            )));
        }

        Ok((MetadataStore { inner, is_shut_down: false }, recovery.recovered))
    }
Returns the writer handle `MetadataStore`, a sorted array of metadata, and a sorted array
of free keys.
src/metadata_store.rs