WEBDIENSTE/NLnetLabs-krill
1
0
Fork 0
mirror of https://github.com/NLnetLabs/krill.git synced 2026-08-17 21:25:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

Remove listeners from event sourcing.

Browse Source
This commit is contained in:
Martin Hoffmann
2025-12-08 19:08:20 +01:00
parent b41d54c8f1
commit 7e44ba41e2
15 changed files with 159 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/commons/eventsourcing/agg.rs
+30 -24
View File
@@ -131,6 +131,36 @@ pub trait Aggregate: Storable + 'static {
}
}
}
/// Process events before they are saved.
///
/// This method is called on the updated aggregate, i.e., the events
/// given by `events` have already been applied to it.
///
/// The method is allowed to return an error, in which case all the
/// changes made by `events` are rolled back to the previous version of
/// the aggregate.
///
/// The default implementation of this method does nothing and returns
/// `Ok(())`.
fn pre_save_events(
&self, events: &[Self::Event], context: &Self::Context
) -> Result<(), Self::Error> {
let _ = (events, context);
Ok(())
}
/// Process events after they have been saved.
///
/// This method is called on the updated aggregate, i.e., the events
/// given by `events` have already been applied to it.
///
/// The default implementation does nothing.
fn post_save_events(
&self, events: &[Self::Event], context: &Self::Context
) {
let _ = (events, context);
}
}
@@ -686,30 +716,6 @@ impl<E: Event, I: InitEvent> StoredEffect<E, I> {
}
//------------ PreSaveEventListener ------------------------------------------
/// A listener that receives events before the aggregate is saved.
///
/// The listener is allowed to return an error in case of issues, which will
/// will result in rolling back the intended change to an aggregate.
pub trait PreSaveEventListener<A: Aggregate>: Send + Sync + 'static {
fn listen(
&self, agg: &A, events: &[A::Event], context: &A::Context,
) -> Result<(), A::Error>;
}
//------------ PostSaveEventListener -----------------------------------------
/// A listener that receives events after the aggregate is saved.
///
/// The listener is not allowed to fail.
pub trait PostSaveEventListener<A: Aggregate>: Send + Sync + 'static {
fn listen(
&self, agg: &A, events: &[A::Event], context: &A::Context,
);
}
//------------ Helper Functions ----------------------------------------------
/// Unfailably creates a JSON value from a serializable object.
src/commons/eventsourcing/mod.rs
+2 -3
View File
@@ -301,9 +301,8 @@ mod wal;
pub use self::agg::{
Aggregate, Command, CommandDetails, Event, InitCommand,
InitCommandDetails, InitEvent, PostSaveEventListener,
PreSaveEventListener, SentCommand, SentInitCommand, StoredCommand,
StoredCommandBuilder, StoredEffect, WithStorableDetails
InitCommandDetails, InitEvent, SentCommand, SentInitCommand,
StoredCommand, StoredCommandBuilder, StoredEffect, WithStorableDetails
};
pub use self::store::{AggregateStore, AggregateStoreError, Storable};
pub use self::wal::{
src/commons/eventsourcing/store.rs
+9 -45
View File
@@ -18,10 +18,7 @@ use crate::api::history::{
};
use crate::commons::error::KrillIoError;
use crate::commons::storage::{Ident, KeyValueError, KeyValueStore};
use super::agg::{
Aggregate, Command, InitCommand, PostSaveEventListener,
PreSaveEventListener, StoredCommand
};
use super::agg::{Aggregate, Command, InitCommand, StoredCommand};
//------------ Storable ------------------------------------------------------
@@ -61,12 +58,6 @@ pub struct AggregateStore<A: Aggregate> {
/// A cache for the command history of an instance.
history_cache: Option<Mutex<HashMap<MyHandle, Vec<CommandHistoryRecord>>>>,
/// The pre-save listeners.
pre_save_listeners: Vec<Arc<dyn PreSaveEventListener<A>>>,
/// The post-save listeners.
post_save_listeners: Vec<Arc<dyn PostSaveEventListener<A>>>,
}
/// # Starting up
@@ -114,8 +105,6 @@ impl<A: Aggregate> AggregateStore<A> {
else {
None
},
pre_save_listeners: Vec::new(),
post_save_listeners: Vec::new(),
}
}
@@ -135,23 +124,6 @@ impl<A: Aggregate> AggregateStore<A> {
}
Ok(())
}
/// Adds a listener that will receive all events before they are stored.
pub fn add_pre_save_listener<L: PreSaveEventListener<A>>(
&mut self,
sync_listener: Arc<L>,
) {
self.pre_save_listeners.push(sync_listener);
}
/// Adds a listener that will receive a reference to all events after they
/// are stored.
pub fn add_post_save_listener<L: PostSaveEventListener<A>>(
&mut self,
listener: Arc<L>,
) {
self.post_save_listeners.push(listener);
}
}
/// # Manage Aggregates
@@ -475,18 +447,12 @@ impl<A: Aggregate> AggregateStore<A> {
// should inform the pre-save listeners. They may
// still generate errors, and if they do, then we
// return with an error, without saving.
let mut opt_err: Option<A::Error> = None;
let mut opt_err = None;
if let Some(events) = processed.events() {
for pre_save_listener
in &self.pre_save_listeners {
if let Err(e)
= pre_save_listener.as_ref().listen(
aggregate, events, context
)
{
opt_err = Some(e);
break;
}
if let Err(err) = aggregate.pre_save_events(
events, context
) {
opt_err = Some(err);
}
}
@@ -505,11 +471,9 @@ impl<A: Aggregate> AggregateStore<A> {
// Now send the events to the 'post-save'
// listeners.
if let Some(events) = processed.events() {
for listener in &self.post_save_listeners {
listener.as_ref().listen(
aggregate, events, context
);
}
aggregate.post_save_events(
events, context
);
}
Ok(())
src/commons/eventsourcing/test.rs
+1 -10
View File
@@ -384,14 +384,6 @@ impl EventCounter {
}
}
impl<A: Aggregate> PostSaveEventListener<A> for EventCounter {
fn listen(
&self, _agg: &A, events: &[A::Event], _context: &A::Context,
) {
self.counter.write().unwrap().total += events.len();
}
}
//------------ Test Function -------------------------------------------------
@@ -401,13 +393,12 @@ fn event_sourcing_framework() {
let counter = Arc::new(EventCounter::default());
let mut manager = AggregateStore::<Person>::create(
let manager = AggregateStore::<Person>::create(
&storage_uri,
const { Ident::make("person") },
false,
)
.unwrap();
manager.add_post_save_listener(counter.clone());
let alice_name = "alice smith".to_string();
let alice_handle = MyHandle::from_str("alice").unwrap();
src/daemon/http/auth/authorizer.rs
+2 -1
View File
@@ -37,6 +37,7 @@ use super::providers::{config_file, openid_connect};
///
/// This type is a wrapper around the available backend specific auth
/// providers that can be found in the [super::providers] module.
#[allow(clippy::large_enum_variant)]
enum AuthProvider {
Token(admin_token::AuthProvider),
@@ -205,7 +206,7 @@ impl Authorizer {
#[cfg(feature = "multi-user")]
AuthType::ConfigFile => {
(
config_file::AuthProvider::new(&config)?.into(),
config_file::AuthProvider::new(config)?.into(),
Some(admin_token::AuthProvider::new(config))
)
}
src/server/ca/certauth.rs
+16
View File
@@ -645,6 +645,22 @@ impl Aggregate for CertAuth {
}
}
}
fn pre_save_events(
&self, events: &[Self::Event], context: &Self::Context
) -> Result<(), Self::Error> {
context.ca_manager().cert_auth_pre_save_events(
self, events, context
)?;
context.tasks().cert_auth_pre_save_events(self, events, context)?;
Ok(())
}
fn post_save_events(
&self, events: &[Self::Event], context: &Self::Context
) {
context.tasks().cert_auth_post_save_events(self, events, context);
}
}
/// # Data presentation
src/server/ca/manager/child.rs
+5 -3
View File
@@ -118,7 +118,8 @@ impl CaManager {
TrustAnchorProxyCommand::make_signer_request(
&ta_handle,
krill.system_actor(),
)
),
krill,
)?;
// Get sign request for signer.
@@ -138,7 +139,7 @@ impl CaManager {
None, // do not override next manifest number
krill.signer(),
krill.system_actor(),
)
),
)?;
// Get the response from the signer and give it to the proxy.
@@ -148,7 +149,8 @@ impl CaManager {
&ta_handle,
exchange.clone().response,
krill.system_actor(),
)
),
krill,
)?;
Ok(())
}
src/server/ca/manager/mod.rs
+39 -59
View File
@@ -61,7 +61,7 @@ use crate::constants::{
};
use crate::daemon::http::auth::{AuthInfo, Permission}; // XXX remove
use crate::server::manager::KrillContext;
use crate::server::mq::{now, Task, TaskQueue};
use crate::server::mq::{now, Task};
use crate::server::runtime;
use crate::server::taproxy::{
TrustAnchorProxy, TrustAnchorProxyCommand, TrustAnchorProxyInitCommand,
@@ -74,6 +74,7 @@ use super::certauth::CertAuth;
use super::commands::{
CertAuthCommandDetails, CertAuthInitCommand, CertAuthInitCommandDetails,
};
use super::events::CertAuthEvent;
use super::publishing::{CaObjectsStore, DeprecatedRepository};
use super::status::{CaStatus, CaStatusStore};
@@ -122,12 +123,11 @@ impl CaManager {
/// Return an error if any of the various stores cannot be initialized.
pub fn build(
config: &Config,
tasks: &Arc<TaskQueue>,
runtime: runtime::Handle,
) -> KrillResult<Self> {
// Create the AggregateStore for the event-sourced `CertAuth`
// structures that handle most CA functions.
let mut ca_store = AggregateStore::<CertAuth>::create(
let ca_store = AggregateStore::<CertAuth>::create(
&config.storage_uri,
CASERVER_NS,
config.use_history_cache,
@@ -162,56 +162,13 @@ impl CaManager {
&config.storage_uri
)?);
// Register the `CaObjectsStore` as a pre-save listener to the
// 'ca_store' so that it can update its ROAs and issued
// certificates and/or generate manifests and CRLs when relevant
// changes occur in a `CertAuth`.
ca_store.add_pre_save_listener(ca_objects_store.clone());
// Register the `MessageQueue` as a pre-save listener to 'ca_store' so
// that relevant changes in a `CertAuth` can trigger follow-up
// actions. This is done as pre-save listener, because commands
// that would result in a follow-up should fail, if the task cannot be
// planned.
//
// Tasks will typically be picked up after the CA changes are
// committed, but they may also be picked up sooner by another
// thread. Because of that the tasks will remember which minimal
// version of the CA they are intended for, so that they can
// be rescheduled should they have been picked up too soon.
//
// An example of a triggered task: schedule a synchronisation with the
// repository (publication server) in case ROAs have been
// updated.
ca_store.add_pre_save_listener(tasks.clone());
// Now also register the `MessageQueue` as a post-save listener. We
// use this to send best-effort post-save signals to children
// in case a certificate was updated or a child key was revoked.
// This is a no-op for remote children (we cannot send a signal over
// RFC 6492).
ca_store.add_post_save_listener(tasks.clone());
// Create TA proxy store if we need it.
let ta_proxy_store = if config.ta_proxy_enabled() {
let mut store = AggregateStore::<TrustAnchorProxy>::create(
Some(AggregateStore::<TrustAnchorProxy>::create(
&config.storage_uri,
TA_PROXY_SERVER_NS,
config.use_history_cache,
)?;
// We need a pre-save listener so that we can schedule:
// - publication on updates
// - signing by the Trust Anchor Signer when there are requests
// [in testbed mode]
store.add_pre_save_listener(tasks.clone());
// We need a post-save listener so that we can schedule:
// - re-sync for local children when the proxy has new responses
// AND is saved
store.add_post_save_listener(tasks.clone());
Some(store)
)?)
}
else {
None
@@ -298,6 +255,16 @@ impl CaManager {
}
Ok(res)
}
pub(super) fn cert_auth_pre_save_events(
&self,
ca: &CertAuth,
events: &[CertAuthEvent],
krill: &KrillContext,
) -> KrillResult<()> {
self.ca_objects_store.cert_auth_pre_save_events(ca, events, krill)
}
}
/// # Trust Anchor Support
@@ -309,10 +276,11 @@ impl CaManager {
fn send_ta_proxy_command(
&self,
cmd: TrustAnchorProxyCommand,
krill: &KrillContext,
) -> KrillResult<Arc<TrustAnchorProxy>> {
self.ta_proxy_store.as_ref().ok_or_else(|| {
Error::custom("ta_support_enabled is false")
})?.command(cmd)
})?.command_with_context(cmd, krill.tasks())
}
/// Sends a command to the TA signer.
@@ -371,12 +339,13 @@ impl CaManager {
ta_proxy_store.add(
ta_proxy_store.add_with_context(
TrustAnchorProxyInitCommand::make(
ta_handle,
krill.signer(),
krill.system_actor(),
)
),
krill.tasks(),
)?;
Ok(())
}
@@ -449,13 +418,15 @@ impl CaManager {
&self,
contact: RepositoryContact,
actor: &Actor,
krill: &KrillContext,
) -> KrillResult<()> {
self.send_ta_proxy_command(
TrustAnchorProxyCommand::add_repo(
&ta_handle(),
contact,
actor,
)
),
krill,
)?;
Ok(())
}
@@ -479,9 +450,11 @@ impl CaManager {
&self,
info: TrustAnchorSignerInfo,
actor: &Actor,
krill: &KrillContext,
) -> KrillResult<()> {
self.send_ta_proxy_command(
TrustAnchorProxyCommand::add_signer(&ta_handle(), info, actor)
TrustAnchorProxyCommand::add_signer(&ta_handle(), info, actor),
krill,
)?;
Ok(())
}
@@ -493,9 +466,11 @@ impl CaManager {
&self,
info: TrustAnchorSignerInfo,
actor: &Actor,
krill: &KrillContext,
) -> KrillResult<()> {
self.send_ta_proxy_command(
TrustAnchorProxyCommand::update_signer(&ta_handle(), info, actor)
TrustAnchorProxyCommand::update_signer(&ta_handle(), info, actor),
krill,
)?;
Ok(())
}
@@ -507,7 +482,8 @@ impl CaManager {
&self, actor: &Actor, krill: &KrillContext,
) -> KrillResult<ApiTrustAnchorSignedRequest> {
self.send_ta_proxy_command(
TrustAnchorProxyCommand::make_signer_request(&ta_handle(), actor)
TrustAnchorProxyCommand::make_signer_request(&ta_handle(), actor),
krill,
)?.get_signer_request(krill.config().ta_timing, krill.signer())
}
@@ -525,13 +501,15 @@ impl CaManager {
&self,
response: TrustAnchorSignedResponse,
actor: &Actor,
krill: &KrillContext,
) -> KrillResult<()> {
self.send_ta_proxy_command(
TrustAnchorProxyCommand::process_signer_response(
&ta_handle(),
response,
actor,
)
),
krill,
)?;
Ok(())
}
@@ -563,14 +541,16 @@ impl CaManager {
let contact = RepositoryContact::try_from_response(
repository_response
).map_err(Error::rfc8183)?;
self.ta_proxy_repository_update(contact, krill.system_actor())?;
self.ta_proxy_repository_update(
contact, krill.system_actor(), krill
)?;
// Initialise signer
self.ta_signer_init(ta_uris, ta_aia, ta_key_pem, krill)?;
// Add signer to proxy
let signer_info = self.get_trust_anchor_signer()?.get_signer_info();
self.ta_proxy_signer_add(signer_info, krill.system_actor())?;
self.ta_proxy_signer_add(signer_info, krill.system_actor(), krill)?;
self.sync_ta_proxy_signer_if_possible(krill)?;
self.cas_repo_sync_single(&ta_handle, 0, krill)?;
@@ -853,7 +833,7 @@ impl CaManager {
let child_handle = req.handle.clone();
let add_child_cmd =
TrustAnchorProxyCommand::add_child(ca, req, actor);
self.send_ta_proxy_command(add_child_cmd)?;
self.send_ta_proxy_command(add_child_cmd, krill)?;
self.ca_parent_response(ca, child_handle, service_uri)
}
}
src/server/ca/manager/parent.rs
+9 -3
View File
@@ -219,6 +219,7 @@ impl CaManager {
child_handle,
request,
actor,
krill,
)
}
else {
@@ -264,7 +265,9 @@ impl CaManager {
) -> KrillResult<provisioning::Message> {
if ca_handle.as_str() == TA_NAME {
let request = ProvisioningRequest::Revocation(revoke_request);
self.ta_slow_rfc6492_request(ca_handle, child, request, actor)
self.ta_slow_rfc6492_request(
ca_handle, child, request, actor, krill
)
}
else {
let res = RevocationResponse::from(&revoke_request);
@@ -291,6 +294,7 @@ impl CaManager {
child: ChildHandle,
request: ProvisioningRequest,
actor: &Actor,
krill: &KrillContext,
) -> KrillResult<provisioning::Message> {
let proxy = self.get_trust_anchor_proxy()?;
if let Some(response) = proxy.response_for_child(&child, &request)? {
@@ -307,7 +311,8 @@ impl CaManager {
child,
request.key_identifier(),
actor,
)
),
krill,
)?;
Ok(response)
@@ -335,7 +340,8 @@ impl CaManager {
child.clone(),
request,
actor,
)
),
krill,
)?;
provisioning::Message::not_performed_response(
src/server/ca/publishing.rs
+1 -5
View File
@@ -26,7 +26,6 @@ use crate::api::roa::RoaInfo;
use crate::commons::KrillResult;
use crate::commons::crypto::KrillSigner;
use crate::commons::error::Error;
use crate::commons::eventsourcing::PreSaveEventListener;
use crate::commons::storage::{Ident, KeyValueStore};
use crate::constants::CA_OBJECTS_NS;
use crate::config::IssuanceTimingConfig;
@@ -179,11 +178,8 @@ impl CaObjectsStore {
objects.re_issue(force, issuance_timing, signer)
})
}
}
/// React to any events on a CA that cause the set of object to change.
impl PreSaveEventListener<CertAuth> for CaObjectsStore {
fn listen(
pub(super) fn cert_auth_pre_save_events(
&self,
ca: &CertAuth,
events: &[CertAuthEvent],
src/server/manager/mod.rs
+8 -7
View File
@@ -213,7 +213,6 @@ impl KrillManager {
let ca_manager = CaManager::build(
&config,
&tasks,
runtime,
)?;
@@ -492,8 +491,9 @@ impl KrillManager {
contact: RepositoryContact,
actor: &Actor,
) -> KrillResult<()> {
self.ca_manager()
.ta_proxy_repository_update(contact, actor)
self.ca_manager().ta_proxy_repository_update(
contact, actor, self.context()
)
}
pub fn ta_proxy_repository_contact(
@@ -507,7 +507,7 @@ impl KrillManager {
info: TrustAnchorSignerInfo,
actor: &Actor,
) -> KrillResult<()> {
self.ca_manager().ta_proxy_signer_add(info, actor)
self.ca_manager().ta_proxy_signer_add(info, actor, self.context())
}
pub fn ta_proxy_signer_update(
@@ -515,7 +515,7 @@ impl KrillManager {
info: TrustAnchorSignerInfo,
actor: &Actor,
) -> KrillResult<()> {
self.ca_manager().ta_proxy_signer_update(info, actor)
self.ca_manager().ta_proxy_signer_update(info, actor, self.context())
}
pub fn ta_proxy_signer_make_request(
@@ -536,8 +536,9 @@ impl KrillManager {
response: TrustAnchorSignedResponse,
actor: &Actor,
) -> KrillResult<()> {
self.ca_manager()
.ta_proxy_signer_process_response(response, actor)
self.ca_manager().ta_proxy_signer_process_response(
response, actor, self.context()
)
}
pub fn ta_proxy_children_add(
src/server/mq.rs
+4 -22
View File
@@ -12,7 +12,6 @@ use rpki::repository::x509::Time;
use serde::{Deserialize, Serialize};
use url::Url;
use crate::api::ca::Timestamp;
use crate::commons::eventsourcing;
use crate::commons::{Error, KrillResult};
use crate::commons::eventsourcing::Aggregate;
use crate::commons::queue::{Queue, ScheduleMode};
@@ -583,11 +582,8 @@ impl TaskQueue {
_ => Ok(()),
}
}
}
/// Implement pre-save listening for CertAuth events.
impl eventsourcing::PreSaveEventListener<CertAuth> for TaskQueue {
fn listen(
pub fn cert_auth_pre_save_events(
&self,
ca: &CertAuth,
events: &[CertAuthEvent],
@@ -598,14 +594,8 @@ impl eventsourcing::PreSaveEventListener<CertAuth> for TaskQueue {
}
Ok(())
}
}
/// Implement post-save listening for CertAuth events.
///
/// Used for best effort signaling to local child CAs that a sync with
/// their parent is needed.
impl eventsourcing::PostSaveEventListener<CertAuth> for TaskQueue {
fn listen(
pub fn cert_auth_post_save_events(
&self,
ca: &CertAuth,
events: &[CertAuthEvent],
@@ -640,15 +630,11 @@ impl eventsourcing::PostSaveEventListener<CertAuth> for TaskQueue {
}
}
}
}
/// Implement pre-save listening for TrustAnchorProxy events.
impl eventsourcing::PreSaveEventListener<TrustAnchorProxy> for TaskQueue {
fn listen(
pub fn ta_proxy_pre_save_events(
&self,
proxy: &TrustAnchorProxy,
events: &[TrustAnchorProxyEvent],
_context: &(),
) -> KrillResult<()> {
for event in events {
trace!("Seen TrustAnchorProxy event '{event}'");
@@ -685,15 +671,11 @@ impl eventsourcing::PreSaveEventListener<TrustAnchorProxy> for TaskQueue {
}
Ok(())
}
}
/// Implement post-save listening for TrustAnchorProxy events.
impl eventsourcing::PostSaveEventListener<TrustAnchorProxy> for TaskQueue {
fn listen(
pub fn ta_proxy_post_save_events(
&self,
_proxy: &TrustAnchorProxy,
events: &[TrustAnchorProxyEvent],
_context: &(),
) {
for event in events {
match event {
src/server/pubd/manager.rs
+2 -2
View File
@@ -59,8 +59,8 @@ impl RepositoryManager {
pub fn build(
config: &Config,
) -> Result<Self, Error> {
let access_proxy = RepositoryAccessProxy::create(&config)?;
let content_proxy = RepositoryContentProxy::create(&config)?;
let access_proxy = RepositoryAccessProxy::create(config)?;
let content_proxy = RepositoryContentProxy::create(config)?;
Ok(RepositoryManager {
access: access_proxy,
src/server/taproxy.rs
+30 -8
View File
@@ -40,6 +40,7 @@ use crate::api::ta::{
};
use crate::constants::ta_resource_class_name;
use crate::server::ca::UsedKeyState;
use crate::server::mq::TaskQueue;
use crate::tasigner::TaTimingConfig;
@@ -113,7 +114,7 @@ impl eventsourcing::Aggregate for TrustAnchorProxy {
type InitCommand<'a> = TrustAnchorProxyInitCommand<'a>;
type InitEvent = TrustAnchorProxyInitEvent;
type Error = Error;
type Context = ();
type Context = TaskQueue;
fn init(
handle: &CaHandle, event: TrustAnchorProxyInitEvent,
@@ -299,6 +300,19 @@ impl eventsourcing::Aggregate for TrustAnchorProxy {
) => self.process_give_child_response(child_handle, key),
}
}
fn pre_save_events(
&self, events: &[Self::Event], context: &Self::Context
) -> Result<(), Self::Error> {
context.ta_proxy_pre_save_events(self, events)?;
Ok(())
}
fn post_save_events(
&self, events: &[Self::Event], context: &Self::Context
) {
context.ta_proxy_post_save_events(self, events)
}
}
// # Process command details
@@ -1278,6 +1292,8 @@ mod tests {
.unwrap(),
);
let tasks = TaskQueue::new(storage_uri).unwrap();
let timing = TaTimingConfig::default();
let actor = crate::constants::ACTOR_DEF_KRILL;
@@ -1289,7 +1305,7 @@ mod tests {
&actor,
);
ta_proxy_store.add(proxy_init).unwrap();
ta_proxy_store.add_with_context(proxy_init, &tasks).unwrap();
let repository = {
let repo_info = RepoInfo::new(
@@ -1316,7 +1332,9 @@ mod tests {
repository,
&actor,
);
let mut proxy = ta_proxy_store.command(add_repo_cmd).unwrap();
let mut proxy = ta_proxy_store.command_with_context(
add_repo_cmd, &tasks
).unwrap();
let signer_handle = CaHandle::new("signer".into());
let tal_https =
@@ -1354,7 +1372,9 @@ mod tests {
&actor,
);
proxy = ta_proxy_store.command(add_signer_cmd).unwrap();
proxy = ta_proxy_store.command_with_context(
add_signer_cmd, &tasks,
).unwrap();
// The initial signer starts off with a TA certificate
// and a CRL and manifest with revision number 42, as specified in
@@ -1373,7 +1393,9 @@ mod tests {
&proxy_handle,
&actor,
);
proxy = ta_proxy_store.command(make_publish_request_cmd).unwrap();
proxy = ta_proxy_store.command_with_context(
make_publish_request_cmd, &tasks,
).unwrap();
let signed_request =
proxy.get_signer_request(timing, &signer).unwrap();
@@ -1400,9 +1422,9 @@ mod tests {
&actor,
);
proxy = ta_proxy_store
.command(ta_proxy_process_signer_response_command)
.unwrap();
proxy = ta_proxy_store.command_with_context(
ta_proxy_process_signer_response_command, &tasks,
).unwrap();
// The TA should have published again, the revision used for
// manifest and crl will have been updated to the
tests/common.rs
+1 -1
View File
@@ -398,7 +398,7 @@ impl KrillServer {
let mut res = Self {
join: tokio::spawn(async {
if let Err(err) = start_krill_daemon(
config.into(), Some(tx)
config, Some(tx)
).await {
error!("Krill failed to start: {err}");
}