node: init policies DB early

Instead of creating the DB policies handle when needed, construct the
handle when the `Worker` pool is being constructed.

This replaces the usage of the DB handle with the `Worker`'s field
instead.

Signed-off-by: Fintan Halpenny <fintan.halpenny@gmail.com>
X-Clacks-Overhead: GNU Terry Pratchett
This commit is contained in:
Fintan Halpenny 2023-12-13 13:03:52 +00:00 committed by cloudhead
parent e894b6512d
commit 63469fd3bc
No known key found for this signature in database
2 changed files with 27 additions and 25 deletions

View File

@ -235,9 +235,6 @@ impl Runtime {
let nid = *signer.public_key();
let fetch = worker::FetchConfig {
policy,
scope,
policies_db: home.node().join(node::POLICIES_DB_FILE),
limit: FetchLimit::default(),
local: nid,
expiry: worker::garbage::Expiry::default(),
@ -251,8 +248,11 @@ impl Runtime {
timeout: time::Duration::from_secs(9),
storage: storage.clone(),
fetch,
policy,
scope,
policies_db: home.node().join(node::POLICIES_DB_FILE),
},
);
)?;
let control = match UnixListener::bind(home.socket()) {
Ok(sock) => sock,
Err(err) if err.kind() == io::ErrorKind::AddrInUse => {

View File

@ -34,6 +34,12 @@ pub struct Config {
pub storage: Storage,
/// Configuration for performing fetched.
pub fetch: FetchConfig,
/// Default policy, if a policy for a specific node or repository was not found.
pub policy: Policy,
/// Default scope, if a scope for a specific repository was not found.
pub scope: policy::Scope,
/// Path to the policies database.
pub policies_db: PathBuf,
}
/// Error returned by fetch.
@ -154,12 +160,6 @@ pub struct TaskResult {
#[derive(Debug, Clone)]
pub struct FetchConfig {
/// Default policy, if a policy for a specific node or repository was not found.
pub policy: Policy,
/// Default scope, if a scope for a specific repository was not found.
pub scope: policy::Scope,
/// Path to the policies database.
pub policies_db: PathBuf,
/// Data limits when fetching from a remote.
pub limit: FetchLimit,
/// Public key of the local peer.
@ -176,6 +176,7 @@ struct Worker {
fetch_config: FetchConfig,
tasks: chan::Receiver<Task>,
handle: Handle,
policies: policy::Config<policy::store::Read>,
}
impl Worker {
@ -261,13 +262,7 @@ impl Worker {
}
fn is_authorized(&self, remote: NodeId, rid: Id) -> Result<(), UploadError> {
let policy = {
let policy = self.fetch_config.policy;
let scope = self.fetch_config.scope;
let db = &self.fetch_config.policies_db;
let policies = policy::Config::new(policy, scope, policy::Store::reader(db)?);
policies.repo_policy(&rid)?.policy
};
let policy = self.policies.repo_policy(&rid)?.policy;
let repo = self.storage.repository(rid)?;
let doc = repo.canonical_identity_doc()?;
if !doc.is_visible_to(&remote) || policy == Policy::Block {
@ -285,18 +280,14 @@ impl Worker {
channels: channels::ChannelsFlush,
) -> Result<fetch::FetchResult, FetchError> {
let FetchConfig {
policy,
scope,
policies_db,
limit,
local,
expiry,
} = &self.fetch_config;
let policies = policy::Config::new(*policy, *scope, policy::Store::reader(policies_db)?);
// N.b. if the `rid` is blocked this will return an error, so
// we won't continue with any further set up of the fetch.
let allowed = radicle_fetch::Allowed::from_config(rid, &policies)?;
let blocked = radicle_fetch::BlockList::from_config(&policies)?;
let allowed = radicle_fetch::Allowed::from_config(rid, &self.policies)?;
let blocked = radicle_fetch::BlockList::from_config(&self.policies)?;
let handle = fetch::Handle::new(rid, *local, &self.storage, allowed, blocked, channels)?;
let result = handle.fetch(rid, &self.storage, *limit, remote, refs_at)?;
@ -318,21 +309,32 @@ pub struct Pool {
impl Pool {
/// Create a new worker pool with the given parameters.
pub fn with(tasks: chan::Receiver<Task>, nid: NodeId, handle: Handle, config: Config) -> Self {
pub fn with(
tasks: chan::Receiver<Task>,
nid: NodeId,
handle: Handle,
config: Config,
) -> Result<Self, policy::Error> {
let mut pool = Vec::with_capacity(config.capacity);
for i in 0..config.capacity {
let policies = policy::Config::new(
config.policy,
config.scope,
policy::Store::reader(&config.policies_db)?,
);
let worker = Worker {
nid,
tasks: tasks.clone(),
handle: handle.clone(),
storage: config.storage.clone(),
fetch_config: config.fetch.clone(),
policies,
};
let thread = thread::spawn(&nid, format!("worker#{i}"), || worker.run());
pool.push(thread);
}
Self { pool }
Ok(Self { pool })
}
/// Run the worker pool.