Compare commits

..

No commits in common. "570633de10cdb421a4d2492fe0cd2f180b9bc40a" and "5e264093ec84a8e5117696731ded965a338d52f3" have entirely different histories.

6 changed files with 18 additions and 92 deletions

View File

@ -1,8 +1,8 @@
use std::{sync, time}; use std::{future, pin, sync, time};
use crate::domain::acme::{self, Certificate, PrivateKey}; use crate::domain::acme::{Certificate, PrivateKey};
use crate::domain::{self, acme};
use crate::error::unexpected::{self, Intoable, Mappable}; use crate::error::unexpected::{self, Intoable, Mappable};
use crate::{domain, util};
const LETS_ENCRYPT_URL: &str = "https://acme-v02.api.letsencrypt.org/directory"; const LETS_ENCRYPT_URL: &str = "https://acme-v02.api.letsencrypt.org/directory";
@ -14,8 +14,7 @@ pub trait Manager {
fn sync_domain<'mgr>( fn sync_domain<'mgr>(
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>>; ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>;
fn get_http01_challenge_key(&self, token: &str) -> Result<String, GetHttp01ChallengeKeyError>; fn get_http01_challenge_key(&self, token: &str) -> Result<String, GetHttp01ChallengeKeyError>;
/// Returned vec is guaranteed to have len > 0 /// Returned vec is guaranteed to have len > 0
@ -87,7 +86,8 @@ impl Manager for ManagerImpl {
fn sync_domain<'mgr>( fn sync_domain<'mgr>(
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>> { ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>
{
Box::pin(async move { Box::pin(async move {
// if there's an existing cert, and its expiry (determined by the soonest value of // if there's an existing cert, and its expiry (determined by the soonest value of
// not_after amongst its parts) is later than 30 days from now, then we consider it to be // not_after amongst its parts) is later than 30 days from now, then we consider it to be

View File

@ -3,7 +3,7 @@ use crate::error::unexpected::{self, Mappable};
use crate::origin; use crate::origin;
use crate::util; use crate::util;
use std::sync; use std::{future, pin, sync};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
#[derive(thiserror::Error, Debug)] #[derive(thiserror::Error, Debug)]
@ -146,13 +146,13 @@ pub trait Manager: Sync + Send + rustls::server::ResolvesServerCert {
fn sync_cert<'mgr>( fn sync_cert<'mgr>(
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>>; ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>;
fn sync_with_config<'mgr>( fn sync_with_config<'mgr>(
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
config: config::Config, config: config::Config,
) -> util::BoxedFuture<'mgr, Result<(), SyncWithConfigError>>; ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>;
fn get_acme_http01_challenge_key( fn get_acme_http01_challenge_key(
&self, &self,
@ -244,7 +244,8 @@ impl Manager for ManagerImpl {
fn sync_cert<'mgr>( fn sync_cert<'mgr>(
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
) -> util::BoxedFuture<'mgr, Result<(), unexpected::Error>> { ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), unexpected::Error>> + Send + 'mgr>>
{
Box::pin(async move { Box::pin(async move {
if let Some(ref acme_manager) = self.acme_manager { if let Some(ref acme_manager) = self.acme_manager {
acme_manager.sync_domain(domain.clone()).await?; acme_manager.sync_domain(domain.clone()).await?;
@ -258,7 +259,8 @@ impl Manager for ManagerImpl {
&'mgr self, &'mgr self,
domain: domain::Name, domain: domain::Name,
config: config::Config, config: config::Config,
) -> util::BoxedFuture<'mgr, Result<(), SyncWithConfigError>> { ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>
{
Box::pin(async move { Box::pin(async move {
let config_hash = config let config_hash = config
.hash() .hash()

View File

@ -1,6 +1,5 @@
use crate::error::unexpected; use crate::error::unexpected;
use futures::stream; use std::sync;
use std::{io, sync};
pub mod git; pub mod git;
pub mod mux; pub mod mux;
@ -30,7 +29,7 @@ pub enum AllDescrsError {
} }
#[derive(thiserror::Error, Debug)] #[derive(thiserror::Error, Debug)]
pub enum GetFileError { pub enum ReadFileIntoError {
#[error("descr not synced")] #[error("descr not synced")]
DescrNotSynced, DescrNotSynced,
@ -41,8 +40,6 @@ pub enum GetFileError {
Unexpected(#[from] unexpected::Error), Unexpected(#[from] unexpected::Error),
} }
pub type ReadFileIntoError = GetFileError;
#[mockall::automock] #[mockall::automock]
/// Describes a storage mechanism for Origins. Each Origin is uniquely identified by its Descr. /// Describes a storage mechanism for Origins. Each Origin is uniquely identified by its Descr.
pub trait Store { pub trait Store {
@ -50,8 +47,6 @@ pub trait Store {
/// the origin into the storage. /// the origin into the storage.
fn sync(&self, descr: &Descr) -> Result<(), SyncError>; fn sync(&self, descr: &Descr) -> Result<(), SyncError>;
fn all_descrs(&self) -> Result<Vec<Descr>, AllDescrsError>;
fn read_file_into( fn read_file_into(
&self, &self,
descr: &Descr, descr: &Descr,
@ -59,11 +54,7 @@ pub trait Store {
into: &mut dyn std::io::Write, into: &mut dyn std::io::Write,
) -> Result<(), ReadFileIntoError>; ) -> Result<(), ReadFileIntoError>;
fn get_file( fn all_descrs(&self) -> Result<Vec<Descr>, AllDescrsError>;
&self,
descr: &Descr,
path: &str,
) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError>;
} }
pub fn new_mock() -> sync::Arc<sync::Mutex<MockStore>> { pub fn new_mock() -> sync::Arc<sync::Mutex<MockStore>> {
@ -79,7 +70,6 @@ impl Store for sync::Arc<sync::Mutex<MockStore>> {
self.lock().unwrap().all_descrs() self.lock().unwrap().all_descrs()
} }
/// Deprecated, use get_file
fn read_file_into( fn read_file_into(
&self, &self,
descr: &Descr, descr: &Descr,
@ -88,12 +78,4 @@ impl Store for sync::Arc<sync::Mutex<MockStore>> {
) -> Result<(), ReadFileIntoError> { ) -> Result<(), ReadFileIntoError> {
self.lock().unwrap().read_file_into(descr, path, into) self.lock().unwrap().read_file_into(descr, path, into)
} }
fn get_file<'store>(
&'store self,
descr: &Descr,
path: &str,
) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, GetFileError> {
self.lock().unwrap().get_file(descr, path)
}
} }

View File

@ -4,8 +4,6 @@ use crate::origin;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::{collections, fs, io, sync}; use std::{collections, fs, io, sync};
use futures::stream;
#[derive(Clone)] #[derive(Clone)]
struct RepoSnapshot { struct RepoSnapshot {
repo: sync::Arc<gix::ThreadSafeRepository>, repo: sync::Arc<gix::ThreadSafeRepository>,
@ -333,50 +331,6 @@ impl super::Store for FSStore {
Ok(()) Ok(())
} }
// TODO test this
fn get_file<'store>(
&'store self,
descr: &origin::Descr,
path: &str,
) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, origin::GetFileError> {
let repo_snapshot = match self.get_repo_snapshot(descr) {
Ok(Some(repo_snapshot)) => repo_snapshot,
Ok(None) => return Err(origin::GetFileError::DescrNotSynced),
Err(e) => return Err(e.into()),
};
let mut clean_path = Path::new(path);
clean_path = clean_path.strip_prefix("/").unwrap_or(clean_path);
let repo = repo_snapshot.repo.to_thread_local();
let file_object = repo
.find_object(repo_snapshot.tree_object_id)
.map_unexpected_while(|| {
format!("finding tree object {}", repo_snapshot.tree_object_id)
})?
.peel_to_tree()
.map_unexpected_while(|| {
format!("peeling tree object {}", repo_snapshot.tree_object_id)
})?
.lookup_entry_by_path(clean_path)
.map_unexpected_while(|| {
format!(
"looking up {} in tree object {}",
clean_path.display(),
repo_snapshot.tree_object_id
)
})?
.ok_or(origin::GetFileError::FileNotFound)?
.object()
.or_unexpected()?;
// TODO this is very not ideal, the whole file is first read totally into memory, and then
// that is cloned.
let data = file_object.data.clone();
Ok(Box::pin(stream::once(async move { Ok(data) })))
}
} }
#[cfg(test)] #[cfg(test)]

View File

@ -1,7 +1,5 @@
use crate::error::unexpected::Mappable; use crate::error::unexpected::Mappable;
use crate::origin; use crate::origin;
use futures::stream;
use std::io;
pub struct Store<F, S> pub struct Store<F, S>
where where
@ -53,16 +51,6 @@ where
.or_unexpected_while(format!("mapping {:?} to store", &descr))? .or_unexpected_while(format!("mapping {:?} to store", &descr))?
.read_file_into(descr, path, into) .read_file_into(descr, path, into)
} }
fn get_file<'store>(
&'store self,
descr: &origin::Descr,
path: &str,
) -> Result<stream::BoxStream<'static, io::Result<Vec<u8>>>, origin::GetFileError> {
(self.mapping_fn)(descr)
.or_unexpected_while(format!("mapping {:?} to store", &descr))?
.get_file(descr, path)
}
} }
#[cfg(test)] #[cfg(test)]

View File

@ -12,13 +12,13 @@ pub fn open_file(path: &path::Path) -> io::Result<Option<fs::File>> {
} }
} }
pub type BoxedFuture<'a, O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'a>>; type StaticFuture<O> = pin::Pin<Box<dyn futures::Future<Output = O> + Send + 'static>>;
pub struct TaskStack<E> pub struct TaskStack<E>
where where
E: error::Error + Send + 'static, E: error::Error + Send + 'static,
{ {
wait_group: Vec<BoxedFuture<'static, Result<(), E>>>, wait_group: Vec<StaticFuture<Result<(), E>>>,
} }
impl<E> TaskStack<E> impl<E> TaskStack<E>