Optionally archive old events (#307)

Changelog.md

@@ -5,6 +5,38 @@ for planned releases.
 
 ## 0.8.0 RC
 
+
+### Archiving old events
+
+You can now choose to archive old publication events which will allow you to
+delete or move them in order to save space on your system. To use this feature,
+which is disabled by default, you need to set the following directive in your
+config file:
+
+```
+ archive_threshold_days = <nr> 
+```
+
+If enabled this option will make sure that the republish events, where your CA
+simply generates a new Manifest and CRL are archived after the given number of
+days.
+
+If you run Krill as a Publication Server then this option will enable the archiving
+of publication deltas received by your server from CAs.
+
+Archived commands (containing e.g. details of when the change happened), and following
+events will be moved to an "archived" subdirectory under your data directory as follows:
+```
+   $data_dir/pubd/0/archived      <-- if you run Krill as a Publication Server
+   $data_dir/cas/ca/archived      <-- for a CA named 'ca'
+```
+
+If you want to save space you can delete the files from these archived directories,
+e.g. from cron. However, you could also archive them in a different way: e.g. compress
+and move to long term storage. Krill will no longer need this data, but if you ever wanted
+to see these details in history you would need to move them back into the parent directory
+of the 'archived' directory.
+
 Features:
 * Will check all state on startup and try to recover in case of issues (#314)
 

defaults/krill.conf

@@ -22,6 +22,34 @@
 #
 ### data_dir = "./data"
 
+# Archive publication events after X days. If you do NOT set this
+# value then Krill will not do any archiving.
+#
+# If enabled this option will make sure that the republish events,
+# where your CA simply generates a new Manifest and CRL are archived
+# after the given number of days.
+#
+# If you run Krill as a Publication Server then this option will
+# enable the archiving of publication deltas received by your server
+# from CAs.
+#
+# Archived commands (containing e.g. details of when the change happened),
+# and following events will be moved to an "archived" subdirectory under
+# your data directory as follows:
+#   $data_dir/pubd/0/archived      <-- for Publication Server
+#   $data_dir/cas/ca/archived      <-- for a CA named 'ca'
+#
+# If you want to save space you can delete the files from these archived
+# directories, e.g. from cron. However, you could also archive them in
+# a different way: e.g. compress and move to long term storage. Krill will
+# no longer need this data, but if you ever wanted to see these details in
+# history you would need to move them back into the parent directory of
+# the 'archived' directory.
+#
+# To enable this set the following key value pair.
+#
+### archive_threshold_days = 7
+
 # Specify the path to the PID file for Krill.
 #
 # Defaults to "krill.pid" under the 'data_dir' specified above.

src/commons/api/history.rs

@@ -155,6 +155,18 @@ impl CommandHistoryRecord {
         Time::from(DateTime::from_utc(time, Utc))
     }
 
+    pub fn resulting_version(&self) -> u64 {
+        if let Some(versions) = self.effect.events() {
+            if let Some(last) = versions.last() {
+                *last
+            } else {
+                self.version
+            }
+        } else {
+            self.version
+        }
+    }
+
     pub fn command_key(&self) -> Result<CommandKey, CommandKeyError> {
         CommandKey::from_str(&self.key)
     }
@@ -290,10 +302,14 @@ pub struct CommandHistoryCriteria {
 }
 
 impl CommandHistoryCriteria {
-    pub fn set_exclude(&mut self, labels: &[&str]) {
+    pub fn set_excludes(&mut self, labels: &[&str]) {
         self.label_excludes = Some(labels.iter().map(|s| (*s).to_string()).collect());
     }
 
+    pub fn set_includes(&mut self, labels: &[&str]) {
+        self.label_includes = Some(labels.iter().map(|s| (*s).to_string()).collect());
+    }
+
     pub fn set_after(&mut self, timestamp: i64) {
         self.after = Some(timestamp);
     }

src/commons/eventsourcing/agg_store.rs

@@ -3,6 +3,7 @@ use std::io;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
 
+use chrono::Duration;
 use rpki::x509::Time;
 
 use crate::commons::api::{CommandHistory, CommandHistoryCriteria, Handle};
@@ -90,6 +91,9 @@ pub enum AggregateStoreError {
 
     #[display(fmt = "Could not recover state for '{}', aborting recover. Use backup!!", _0)]
     CouldNotRecover(Handle),
+
+    #[display(fmt = "Could not archive commands and events for '{}'. Error: {}", _0, _1)]
+    CouldNotArchive(Handle, String),
 }
 
 impl From<KeyStoreError> for AggregateStoreError {
@@ -136,8 +140,15 @@ where
     /// In that case the user may want to use the recover option to see what can be salvaged.
     pub fn warm(&self) -> StoreResult<()> {
         for handle in self.list() {
-            self.get_latest(&handle)
-                .map_err(|e| AggregateStoreError::WarmupFailed(handle, e.to_string()))?;
+            if !self.store.has_key(&handle, &DiskKeyStore::key_for_snapshot()) {
+                return Err(AggregateStoreError::WarmupFailed(
+                    handle,
+                    "No snapshot found.".to_string(),
+                ));
+            } else {
+                self.get_latest(&handle)
+                    .map_err(|e| AggregateStoreError::WarmupFailed(handle, e.to_string()))?;
+            }
         }
         Ok(())
     }
@@ -233,6 +244,50 @@ where
 
         Ok(())
     }
+
+    /// Archive old commands if they are:
+    /// - older than the backup snapshot
+    /// - AND older then the threshold days
+    /// - AND they are eligible for archiving
+    pub fn archive_old_commands(&self, handle: &Handle, days: i64) -> StoreResult<()> {
+        let mut crit = CommandHistoryCriteria::default();
+        let before = (Time::now() - Duration::days(days)).timestamp();
+        crit.set_before(before);
+        crit.set_includes(&["cmd-ca-publish", "pubd-publish"]);
+
+        let info = self
+            .store
+            .get_info(handle)
+            .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?;
+
+        let archivable = self.command_history(handle, crit)?;
+
+        let commands = archivable.commands();
+
+        for command in commands {
+            let key = command
+                .command_key()
+                .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?;
+
+            if command.resulting_version() < info.snapshot_version {
+                info!("Archiving command {} for {}", command.key, handle);
+
+                self.store
+                    .archive(handle, &key.into())
+                    .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?;
+
+                if let Some(evt_versions) = command.effect.events() {
+                    for version in evt_versions {
+                        info!("Archiving event {} for {}", version, handle);
+                        self.store
+                            .archive_event(handle, *version)
+                            .map_err(|e| AggregateStoreError::CouldNotArchive(handle.clone(), e.to_string()))?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
 }
 
 impl<A: Aggregate> DiskAggregateStore<A>

src/commons/eventsourcing/mod.rs

@@ -329,7 +329,7 @@ mod tests {
 
         // Get history excluding 'around the sun' commands
         let mut crit = CommandHistoryCriteria::default();
-        crit.set_exclude(&["person-around-sun"]);
+        crit.set_excludes(&["person-around-sun"]);
         let history = manager.command_history(&id_alice, crit).unwrap();
         assert_eq!(history.total(), 1);
 

src/commons/eventsourcing/store.rs

@@ -123,6 +123,12 @@ impl TryFrom<PathBuf> for CommandKey {
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct CommandKeyError;
 
+impl fmt::Display for CommandKeyError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "invalid command key")
+    }
+}
+
 //------------ KeyStore ------------------------------------------------------
 
 /// Generic KeyStore for AggregateManager

src/daemon/ca/server.rs

@@ -24,7 +24,7 @@ use crate::commons::eventsourcing::{Aggregate, AggregateStore, Command, CommandK
 use crate::commons::remote::cmslogger::CmsLogger;
 use crate::commons::remote::{rfc6492, rfc8181, rfc8183};
 use crate::commons::util::httpclient;
-use crate::commons::KrillResult;
+use crate::commons::{KrillEmptyResult, KrillResult};
 use crate::constants::{CASERVER_DIR, STATUS_DIR};
 use crate::daemon::ca::{
     self, ta_handle, CertAuth, Cmd, CmdDet, IniDet, ResourceTaggedAttestation, RouteAuthorizationUpdates,
@@ -376,6 +376,16 @@ impl CaServer {
             .map_err(|_| Error::CaUnknown(handle.clone()))
     }
 
+    /// Archive old (eligible) commands for CA
+    pub async fn archive_ca_commands(&self, days: i64) -> KrillEmptyResult {
+        for ca in self.ca_list().await.cas() {
+            let lock = self.locks.ca(ca.handle()).await;
+            let _ = lock.write().await;
+            self.ca_store.archive_old_commands(ca.handle(), days)?;
+        }
+        Ok(())
+    }
+
     /// Shows the details for a CA command
     pub fn get_ca_command_details(
         &self,

src/daemon/config.rs

@@ -61,6 +61,13 @@ impl ConfigDefaults {
     fn data_dir() -> PathBuf {
         PathBuf::from("./data")
     }
+    fn archive_threshold_days() -> Option<i64> {
+        if Self::test_mode() {
+            Some(0)
+        } else {
+            None
+        }
+    }
     fn rsync_base() -> uri::Rsync {
         uri::Rsync::from_str("rsync://localhost/repo/").unwrap()
     }
@@ -206,6 +213,9 @@ pub struct Config {
     #[serde(default = "ConfigDefaults::data_dir")]
     pub data_dir: PathBuf,
 
+    #[serde(default = "ConfigDefaults::archive_threshold_days")]
+    pub archive_threshold_days: Option<i64>,
+
     pub pid_file: Option<PathBuf>,
 
     #[serde(default = "ConfigDefaults::rsync_base")]
@@ -359,6 +369,7 @@ impl Config {
         let testbed_enabled = true;
         let https_mode = HttpsMode::Generate;
         let data_dir = data_dir.clone();
+        let archive_threshold_days = Some(0);
         let rsync_base = ConfigDefaults::rsync_base();
         let service_uri = ConfigDefaults::service_uri();
         let rrdp_service_uri = Some("https://localhost:3000/test-rrdp/".to_string());
@@ -409,6 +420,7 @@ impl Config {
             testbed_enabled,
             https_mode,
             data_dir,
+            archive_threshold_days,
             rsync_base,
             service_uri,
             rrdp_service_uri,

src/daemon/http/server.rs

@@ -237,7 +237,7 @@ pub async fn metrics(req: Request) -> RoutingResult {
         res.push_str(&format!("krill_version_patch {}\n", KRILL_VERSION_PATCH));
         res.push_str("\n");
 
-        if let Ok(stats) = server.repo_stats().await {
+        if let Ok(stats) = server.repo_stats() {
             let publishers = stats.get_publishers();
 
             res.push_str("# HELP krill_repo_publisher number of publishers in repository\n");
@@ -435,7 +435,7 @@ pub async fn rfc8181(req: Request) -> RoutingResult {
         };
 
         let read = state.read().await;
-        match read.rfc8181(publisher, bytes).await {
+        match read.rfc8181(publisher, bytes) {
             Ok(bytes) => Ok(HttpResponse::rfc8181(bytes.to_vec())),
             Err(e) => render_error(e),
         }
@@ -504,7 +504,7 @@ async fn stats(req: Request) -> RoutingResult {
     match *req.method() {
         Method::GET => match req.path().full() {
             "/stats/info" => render_json(req.state().read().await.server_info()),
-            "/stats/repo" => render_json_res(req.state().read().await.repo_stats().await),
+            "/stats/repo" => render_json_res(req.state().read().await.repo_stats()),
             "/stats/cas" => render_json(req.state().read().await.cas_stats().await),
             _ => Err(req),
         },
@@ -674,7 +674,6 @@ pub async fn stale_publishers(req: Request, seconds: Option<&str>) -> RoutingRes
                 .read()
                 .await
                 .repo_stats()
-                .await
                 .map(|stats| PublisherList::build(&stats.stale_publishers(seconds), "/api/v1/publishers")),
         ),
         Err(_) => render_error(Error::ApiInvalidSeconds),
@@ -688,7 +687,6 @@ pub async fn list_pbl(req: Request) -> RoutingResult {
             .read()
             .await
             .publishers()
-            .await
             .map(|publishers| PublisherList::build(&publishers, "/api/v1/publishers")),
     )
 }
@@ -697,22 +695,20 @@ pub async fn list_pbl(req: Request) -> RoutingResult {
 async fn add_pbl(req: Request) -> RoutingResult {
     let server = req.state().clone();
     match req.json().await {
-        Ok(pbl) => render_json_res(server.write().await.add_publisher(pbl).await),
+        Ok(pbl) => render_json_res(server.write().await.add_publisher(pbl)),
         Err(e) => render_error(e),
     }
 }
 
 /// Removes a publisher. Should be idempotent! If if did not exist then
 /// that's just fine.
-#[allow(clippy::needless_pass_by_value)]
 pub async fn remove_pbl(req: Request, publisher: Handle) -> RoutingResult {
-    render_empty_res(req.state().write().await.remove_publisher(publisher).await)
+    render_empty_res(req.state().write().await.remove_publisher(publisher))
 }
 
 /// Returns a json structure with publisher details
-#[allow(clippy::needless_pass_by_value)]
 pub async fn show_pbl(req: Request, publisher: Handle) -> RoutingResult {
-    render_json_res(req.state().read().await.get_publisher(&publisher).await)
+    render_json_res(req.state().read().await.get_publisher(&publisher))
 }
 
 //------------ repository_response ---------------------------------------------
@@ -732,7 +728,7 @@ pub async fn repository_response_json(req: Request, publisher: Handle) -> Routin
 }
 
 async fn repository_response(req: &Request, publisher: &Handle) -> Result<rfc8183::RepositoryResponse, Error> {
-    req.state().read().await.repository_response(publisher).await
+    req.state().read().await.repository_response(publisher)
 }
 
 async fn ca_add_child(req: Request, parent: ParentHandle) -> RoutingResult {
@@ -867,7 +863,7 @@ fn parse_history_path(path: &mut RequestPath) -> Option<CommandHistoryCriteria>
     let mut crit = CommandHistoryCriteria::default();
 
     match path.next() {
-        Some("short") => crit.set_exclude(&["cmd-ca-publish"]),
+        Some("short") => crit.set_excludes(&["cmd-ca-publish"]),
         Some("full") => {}
         _ => return None,
     };

src/daemon/krillserver.rs

@@ -114,16 +114,13 @@ impl KrillServer {
 
         let pubserver = {
             if CONFIG.repo_enabled {
-                Some(
-                    PubServer::build(
-                        &base_uri,
-                        rrdp_base_uri.clone(),
-                        work_dir,
-                        CONFIG.rfc8181_log_dir.as_ref(),
-                        Arc::new(signer.clone()),
-                    )
-                    .await?,
-                )
+                Some(PubServer::build(
+                    &base_uri,
+                    rrdp_base_uri.clone(),
+                    work_dir,
+                    CONFIG.rfc8181_log_dir.as_ref(),
+                    Arc::new(signer.clone()),
+                )?)
             } else {
                 PubServer::remove_if_empty(
                     &base_uri,
@@ -131,8 +128,7 @@ impl KrillServer {
                     work_dir,
                     CONFIG.rfc8181_log_dir.as_ref(),
                     Arc::new(signer.clone()),
-                )
-                .await?
+                )?
             }
         };
         let pubserver: Option<Arc<PubServer>> = pubserver.map(Arc::new);
@@ -155,7 +151,7 @@ impl KrillServer {
                 info!("Creating embedded Trust Anchor");
 
                 let pubserver = pubserver.as_ref().ok_or_else(|| Error::PublisherNoEmbeddedRepo)?;
-                let repo_info: RepoInfo = pubserver.repo_info_for(&ta_handle).await?;
+                let repo_info: RepoInfo = pubserver.repo_info_for(&ta_handle)?;
 
                 let ta_uri = CONFIG.ta_cert_uri();
 
@@ -170,7 +166,7 @@ impl KrillServer {
                 // Add publisher
                 let req = rfc8183::PublisherRequest::new(None, ta_handle.clone(), ta.id_cert().clone());
 
-                pubserver.create_publisher(req).await?;
+                pubserver.create_publisher(req)?;
 
                 // Force initial publication
                 caserver.republish(&ta_handle).await?;
@@ -192,10 +188,10 @@ impl KrillServer {
                     let pubserver = pubserver.as_ref().ok_or_else(|| Error::PublisherNoEmbeddedRepo)?;
                     let pub_req =
                         rfc8183::PublisherRequest::new(None, testbed_ca_handle.clone(), testbed_ca.id_cert().clone());
-                    pubserver.create_publisher(pub_req).await?;
+                    pubserver.create_publisher(pub_req)?;
                     let rfc8181_uri =
                         uri::Https::from_string(format!("{}rfc8181/{}", service_uri, testbed_ca_handle)).unwrap();
-                    let repo_response = pubserver.repository_response(rfc8181_uri, &testbed_ca_handle).await?;
+                    let repo_response = pubserver.repository_response(rfc8181_uri, &testbed_ca_handle)?;
                     let repo_contact = RepositoryContact::Rfc8181(repo_response);
                     caserver.update_repo(testbed_ca_handle.clone(), repo_contact).await?;
                     caserver.republish(&testbed_ca_handle).await?;
@@ -282,32 +278,32 @@ impl KrillServer {
     }
 
     /// Returns the repository server stats
-    pub async fn repo_stats(&self) -> KrillResult<RepoStats> {
-        self.get_embedded()?.repo_stats().await
+    pub fn repo_stats(&self) -> KrillResult<RepoStats> {
+        self.get_embedded()?.repo_stats()
     }
 
     /// Returns all currently CONFIGured publishers. (excludes deactivated)
-    pub async fn publishers(&self) -> KrillResult<Vec<Handle>> {
-        self.get_embedded()?.publishers().await
+    pub fn publishers(&self) -> KrillResult<Vec<Handle>> {
+        self.get_embedded()?.publishers()
     }
 
     /// Adds the publishers, blows up if it already existed.
-    pub async fn add_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<rfc8183::RepositoryResponse> {
+    pub fn add_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<rfc8183::RepositoryResponse> {
         let publisher_handle = req.publisher_handle().clone();
 
-        self.get_embedded()?.create_publisher(req).await?;
+        self.get_embedded()?.create_publisher(req)?;
 
-        self.repository_response(&publisher_handle).await
+        self.repository_response(&publisher_handle)
     }
 
     /// Removes a publisher, blows up if it didn't exist.
-    pub async fn remove_publisher(&mut self, publisher: PublisherHandle) -> KrillEmptyResult {
-        self.get_embedded()?.remove_publisher(publisher).await
+    pub fn remove_publisher(&mut self, publisher: PublisherHandle) -> KrillEmptyResult {
+        self.get_embedded()?.remove_publisher(publisher)
     }
 
     /// Returns a publisher.
-    pub async fn get_publisher(&self, publisher: &PublisherHandle) -> KrillResult<PublisherDetails> {
-        self.get_embedded()?.get_publisher_details(publisher).await
+    pub fn get_publisher(&self, publisher: &PublisherHandle) -> KrillResult<PublisherDetails> {
+        self.get_embedded()?.get_publisher_details(publisher)
     }
 
     pub fn rrdp_base_path(&self) -> PathBuf {
@@ -320,14 +316,13 @@ impl KrillServer {
 /// # Manage RFC8181 clients
 ///
 impl KrillServer {
-    pub async fn repository_response(&self, publisher: &PublisherHandle) -> KrillResult<rfc8183::RepositoryResponse> {
+    pub fn repository_response(&self, publisher: &PublisherHandle) -> KrillResult<rfc8183::RepositoryResponse> {
         let rfc8181_uri = uri::Https::from_string(format!("{}rfc8181/{}", self.service_uri, publisher)).unwrap();
-
-        self.get_embedded()?.repository_response(rfc8181_uri, publisher).await
+        self.get_embedded()?.repository_response(rfc8181_uri, publisher)
     }
 
-    pub async fn rfc8181(&self, publisher: PublisherHandle, msg_bytes: Bytes) -> KrillResult<Bytes> {
-        self.get_embedded()?.rfc8181(publisher, msg_bytes).await
+    pub fn rfc8181(&self, publisher: PublisherHandle, msg_bytes: Bytes) -> KrillResult<Bytes> {
+        self.get_embedded()?.rfc8181(publisher, msg_bytes)
     }
 }
 
@@ -596,7 +591,7 @@ impl KrillServer {
         let contact = match update {
             RepositoryUpdate::Embedded => {
                 // Add to embedded publication server if not present
-                if self.get_embedded()?.get_publisher_details(&handle).await.is_err() {
+                if self.get_embedded()?.get_publisher_details(&handle).is_err() {
                     let id_cert = {
                         let ca = self.caserver.get_ca(&handle).await?;
                         ca.id_cert().clone()
@@ -604,10 +599,10 @@ impl KrillServer {
 
                     // Add publisher
                     let req = rfc8183::PublisherRequest::new(None, handle.clone(), id_cert);
-                    self.add_publisher(req).await?;
+                    self.add_publisher(req)?;
                 }
 
-                RepositoryContact::embedded(self.get_embedded()?.repo_info_for(&handle).await?)
+                RepositoryContact::embedded(self.get_embedded()?.repo_info_for(&handle)?)
             }
             RepositoryUpdate::Rfc8181(response) => {
                 // first check that the new repo can be contacted
@@ -701,13 +696,13 @@ impl KrillServer {
 impl KrillServer {
     /// Handles a publish delta request sent to the API, or.. through
     /// the CmsProxy.
-    pub async fn handle_delta(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillEmptyResult {
-        self.get_embedded()?.publish(publisher, delta).await
+    pub fn handle_delta(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillEmptyResult {
+        self.get_embedded()?.publish(publisher, delta)
     }
 
     /// Handles a list request sent to the API, or.. through the CmsProxy.
-    pub async fn handle_list(&self, publisher: &PublisherHandle) -> KrillResult<ListReply> {
-        self.get_embedded()?.list(publisher).await
+    pub fn handle_list(&self, publisher: &PublisherHandle) -> KrillResult<ListReply> {
+        self.get_embedded()?.list(publisher)
     }
 }
 

src/daemon/scheduler.rs

@@ -36,6 +36,10 @@ pub struct Scheduler {
     /// Responsible for refreshing announcement information
     #[allow(dead_code)] // just need to keep this in scope
     announcements_refresh_sh: ScheduleHandle,
+
+    /// Responsible for archiving old commands
+    #[allow(dead_code)] // just need to keep this in scope
+    archive_old_commands_sh: ScheduleHandle,
 }
 
 impl Scheduler {
@@ -46,16 +50,17 @@ impl Scheduler {
         bgp_analyser: Arc<BgpAnalyser>,
         ca_refresh_rate: u32,
     ) -> Self {
-        let event_sh = make_event_sh(event_queue, caserver.clone(), pubserver);
+        let event_sh = make_event_sh(event_queue, caserver.clone(), pubserver.clone());
         let republish_sh = make_republish_sh(caserver.clone());
-        let ca_refresh_sh = make_ca_refresh_sh(caserver, ca_refresh_rate);
+        let ca_refresh_sh = make_ca_refresh_sh(caserver.clone(), ca_refresh_rate);
         let announcements_refresh_sh = make_announcements_refresh_sh(bgp_analyser);
-
+        let archive_old_commands_sh = make_archive_old_commands_sh(caserver, pubserver);
         Scheduler {
             event_sh,
             republish_sh,
             ca_refresh_sh,
             announcements_refresh_sh,
+            archive_old_commands_sh,
         }
     }
 }
@@ -230,3 +235,24 @@ fn make_announcements_refresh_sh(bgp_analyser: Arc<BgpAnalyser>) -> ScheduleHand
     });
     scheduler.watch_thread(Duration::from_millis(100))
 }
+
+fn make_archive_old_commands_sh(caserver: Arc<CaServer>, pubserver: Option<Arc<PubServer>>) -> ScheduleHandle {
+    let mut scheduler = clokwerk::Scheduler::new();
+    scheduler.every(60.seconds()).run(move || {
+        let mut rt = Runtime::new().unwrap();
+        rt.block_on(async {
+            if let Some(days) = CONFIG.archive_threshold_days {
+                if let Err(e) = caserver.archive_ca_commands(days).await {
+                    error!("Failed to archive old CA commands: {}", e)
+                }
+
+                if let Some(pubserver) = pubserver.as_ref() {
+                    if let Err(e) = pubserver.history_archive_old(days) {
+                        error!("Failed to archive old Publication Server commands: {}", e)
+                    }
+                }
+            }
+        })
+    });
+    scheduler.watch_thread(Duration::from_millis(100))
+}

src/pubd/pubserver.rs

@@ -42,7 +42,7 @@ pub struct PubServer {
 /// # Constructing
 ///
 impl PubServer {
-    pub async fn remove_if_empty(
+    pub fn remove_if_empty(
         rsync_base: &uri::Rsync,
         rrdp_base_uri: uri::Https,         // for the RRDP files
         work_dir: &PathBuf,                // for the aggregate stores
@@ -52,8 +52,8 @@ impl PubServer {
         let mut pub_server_dir = work_dir.clone();
         pub_server_dir.push(PUBSERVER_DIR);
         if pub_server_dir.exists() {
-            let server = PubServer::build(rsync_base, rrdp_base_uri, work_dir, rfc8181_log_dir, signer).await?;
-            if server.publishers().await?.is_empty() {
+            let server = PubServer::build(rsync_base, rrdp_base_uri, work_dir, rfc8181_log_dir, signer)?;
+            if server.publishers()?.is_empty() {
                 let _result = fs::remove_dir_all(pub_server_dir);
                 Ok(None)
             } else {
@@ -64,7 +64,7 @@ impl PubServer {
         }
     }
 
-    pub async fn build(
+    pub fn build(
         rsync_base: &uri::Rsync,
         rrdp_base_uri: uri::Https,         // for the RRDP files
         work_dir: &PathBuf,                // for the aggregate stores
@@ -105,7 +105,7 @@ impl PubServer {
         Handle::from_str(PUBSERVER_DFLT).unwrap()
     }
 
-    async fn repository(&self) -> KrillResult<Arc<Repository>> {
+    fn repository(&self) -> KrillResult<Arc<Repository>> {
         let handle = Self::repository_handle();
 
         match self.store.get_latest(&handle) {
@@ -118,8 +118,8 @@ impl PubServer {
     }
 
     /// Handle an RFC8181 request and sign the response
-    pub async fn rfc8181(&self, publisher_handle: PublisherHandle, msg_bytes: Bytes) -> KrillResult<Bytes> {
-        let repository = self.repository().await?;
+    pub fn rfc8181(&self, publisher_handle: PublisherHandle, msg_bytes: Bytes) -> KrillResult<Bytes> {
+        let repository = self.repository()?;
         let publisher = repository.get_publisher(&publisher_handle)?;
 
         let msg = ProtocolCms::decode(msg_bytes.clone(), false).map_err(|e| Error::Rfc8181Decode(e.to_string()))?;
@@ -135,7 +135,7 @@ impl PubServer {
                 let list_reply = publisher.list_current();
                 (rfc8181::Message::list_reply(list_reply), false)
             }
-            rfc8181::QueryMessage::PublishDelta(delta) => match self.publish(publisher_handle, delta).await {
+            rfc8181::QueryMessage::PublishDelta(delta) => match self.publish(publisher_handle, delta) {
                 Ok(()) => (rfc8181::Message::success_reply(), true),
                 Err(e) => {
                     let error_code = e.to_rfc8181_error_code();
@@ -161,26 +161,26 @@ impl PubServer {
     }
 
     /// Let a known publisher publish in a repository.
-    pub async fn publish(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillResult<()> {
+    pub fn publish(&self, publisher: PublisherHandle, delta: PublishDelta) -> KrillResult<()> {
         let repository_handle = Self::repository_handle();
         let cmd = CmdDet::publish(&repository_handle, publisher, delta);
         self.store.command(cmd)?;
-        self.write_repository().await
+        self.write_repository()
     }
 
-    pub async fn repo_stats(&self) -> KrillResult<RepoStats> {
-        let repo = self.repository().await?;
+    pub fn repo_stats(&self) -> KrillResult<RepoStats> {
+        let repo = self.repository()?;
         Ok(repo.stats().clone())
     }
 
-    pub async fn publishers(&self) -> KrillResult<Vec<PublisherHandle>> {
-        let repository = self.repository().await?;
+    pub fn publishers(&self) -> KrillResult<Vec<PublisherHandle>> {
+        let repository = self.repository()?;
         Ok(repository.publishers())
     }
 
     /// Returns a list reply for a known publisher in a repository
-    pub async fn list(&self, publisher: &PublisherHandle) -> KrillResult<ListReply> {
-        let repository = self.repository().await?;
+    pub fn list(&self, publisher: &PublisherHandle) -> KrillResult<ListReply> {
+        let repository = self.repository()?;
         let publisher = repository.get_publisher(publisher)?;
         Ok(publisher.list_current())
     }
@@ -189,31 +189,31 @@ impl PubServer {
 /// # Manage publishers
 ///
 impl PubServer {
-    pub async fn repo_info_for(&self, publisher: &PublisherHandle) -> KrillResult<RepoInfo> {
-        let repository = self.repository().await?;
+    pub fn repo_info_for(&self, publisher: &PublisherHandle) -> KrillResult<RepoInfo> {
+        let repository = self.repository()?;
         Ok(repository.repo_info_for(publisher))
     }
 
-    pub async fn get_publisher_details(&self, publisher_handle: &PublisherHandle) -> KrillResult<PublisherDetails> {
-        let repository = self.repository().await?;
+    pub fn get_publisher_details(&self, publisher_handle: &PublisherHandle) -> KrillResult<PublisherDetails> {
+        let repository = self.repository()?;
         repository
             .get_publisher(publisher_handle)
             .map(|p| p.as_api_details(publisher_handle))
     }
 
     /// Returns the RFC8183 Repository Response for the publisher
-    pub async fn repository_response(
+    pub fn repository_response(
         &self,
         rfc8181_uri: uri::Https,
         publisher: &PublisherHandle,
     ) -> KrillResult<rfc8183::RepositoryResponse> {
-        let repository = self.repository().await?;
+        let repository = self.repository()?;
         repository.repository_response(rfc8181_uri, publisher)
     }
 
     /// Adds a publisher. Will complain if a publisher already exists for this
     /// handle. Will also verify that the base_uri is allowed.
-    pub async fn create_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<()> {
+    pub fn create_publisher(&self, req: rfc8183::PublisherRequest) -> KrillResult<()> {
         let repository_handle = Self::repository_handle();
         let cmd = CmdDet::add_publisher(&repository_handle, req);
         self.store.command(cmd)?;
@@ -224,11 +224,11 @@ impl PubServer {
     /// re-activation in future. Reason is that we never forget the history
     /// of the old publisher, and if handles are re-used by different
     /// entities that would get confusing.
-    pub async fn remove_publisher(&self, publisher: PublisherHandle) -> KrillResult<()> {
+    pub fn remove_publisher(&self, publisher: PublisherHandle) -> KrillResult<()> {
         let repository_handle = Self::repository_handle();
         let cmd = CmdDet::remove_publisher(&repository_handle, publisher);
         self.store.command(cmd)?;
-        self.write_repository().await
+        self.write_repository()
     }
 }
 
@@ -236,12 +236,22 @@ impl PubServer {
 ///
 impl PubServer {
     /// Update the RRDP files and rsync content on disk.
-    pub async fn write_repository(&self) -> KrillResult<()> {
-        let repository = self.repository().await?;
+    pub fn write_repository(&self) -> KrillResult<()> {
+        let repository = self.repository()?;
         repository.write()
     }
 }
 
+/// # Manage history
+///
+impl PubServer {
+    /// Archive old commands
+    pub fn history_archive_old(&self, days: i64) -> KrillResult<()> {
+        let handle = Self::repository_handle();
+        self.store.archive_old_commands(&handle, days)?;
+        Ok(())
+    }
+}
 //------------ Tests ---------------------------------------------------------
 
 #[cfg(test)]
@@ -285,28 +295,26 @@ mod tests {
         rfc8183::PublisherRequest::new(None, handle, id_cert.clone())
     }
 
-    async fn make_server(work_dir: &PathBuf) -> PubServer {
+    fn make_server(work_dir: &PathBuf) -> PubServer {
         let signer = KrillSigner::build(work_dir).unwrap();
         let signer = Arc::new(signer);
 
-        PubServer::build(&server_base_uri(), server_base_http_uri(), work_dir, None, signer)
-            .await
-            .unwrap()
+        PubServer::build(&server_base_uri(), server_base_http_uri(), work_dir, None, signer).unwrap()
     }
 
-    #[tokio::test]
-    async fn should_add_publisher() {
+    #[test]
+    fn should_add_publisher() {
         let d = test::tmp_dir();
-        let server = make_server(&d).await;
+        let server = make_server(&d);
 
         let alice = publisher_alice(&d);
 
         let alice_handle = Handle::from_str("alice").unwrap();
         let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert());
 
-        server.create_publisher(publisher_req).await.unwrap();
+        server.create_publisher(publisher_req).unwrap();
 
-        let alice_found = server.get_publisher_details(&alice_handle).await.unwrap();
+        let alice_found = server.get_publisher_details(&alice_handle).unwrap();
 
         assert_eq!(alice_found.base_uri(), alice.base_uri());
         assert_eq!(alice_found.id_cert(), alice.id_cert());
@@ -315,47 +323,47 @@ mod tests {
         let _ = fs::remove_dir_all(d);
     }
 
-    #[tokio::test]
-    async fn should_not_add_publisher_twice() {
+    #[test]
+    fn should_not_add_publisher_twice() {
         let d = test::tmp_dir();
-        let server = make_server(&d).await;
+        let server = make_server(&d);
 
         let alice = publisher_alice(&d);
 
         let alice_handle = Handle::from_str("alice").unwrap();
         let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert());
 
-        server.create_publisher(publisher_req.clone()).await.unwrap();
+        server.create_publisher(publisher_req.clone()).unwrap();
 
-        match server.create_publisher(publisher_req).await {
+        match server.create_publisher(publisher_req) {
             Err(Error::PublisherDuplicate(name)) => assert_eq!(name, alice_handle),
             _ => panic!("Expected error"),
         }
         let _ = fs::remove_dir_all(d);
     }
 
-    #[tokio::test]
-    async fn should_list_files() {
+    #[test]
+    fn should_list_files() {
         let d = test::tmp_dir();
-        let server = make_server(&d).await;
+        let server = make_server(&d);
 
         let alice = publisher_alice(&d);
 
         let alice_handle = Handle::from_str("alice").unwrap();
         let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert());
 
-        server.create_publisher(publisher_req).await.unwrap();
+        server.create_publisher(publisher_req).unwrap();
 
-        let list_reply = server.list(&alice_handle).await.unwrap();
+        let list_reply = server.list(&alice_handle).unwrap();
         assert_eq!(0, list_reply.elements().len());
 
         let _ = fs::remove_dir_all(d);
     }
 
-    #[tokio::test]
-    async fn should_publish_files() {
+    #[test]
+    fn should_publish_files() {
         let d = test::tmp_dir();
-        let server = make_server(&d).await;
+        let server = make_server(&d);
 
         // set up server with default repository, and publisher alice
         let alice = publisher_alice(&d);
@@ -363,7 +371,7 @@ mod tests {
         let alice_handle = Handle::from_str("alice").unwrap();
         let publisher_req = make_publisher_req(alice_handle.as_str(), alice.id_cert());
 
-        server.create_publisher(publisher_req).await.unwrap();
+        server.create_publisher(publisher_req).unwrap();
 
         // get the file out of a list_reply
         fn find_in_reply<'a>(reply: &'a ListReply, uri: &uri::Rsync) -> Option<&'a ListElement> {
@@ -386,10 +394,10 @@ mod tests {
         builder.add_publish(file2.as_publish());
         let delta = builder.finish();
 
-        server.publish(alice_handle.clone(), delta).await.unwrap();
+        server.publish(alice_handle.clone(), delta).unwrap();
 
         // Two files should now appear in the list
-        let list_reply = server.list(&alice_handle).await.unwrap();
+        let list_reply = server.list(&alice_handle).unwrap();
         assert_eq!(2, list_reply.elements().len());
         assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file.txt")).is_some());
         assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file2.txt")).is_some());
@@ -415,10 +423,10 @@ mod tests {
         builder.add_publish(file3.as_publish());
         let delta = builder.finish();
 
-        server.publish(alice_handle.clone(), delta).await.unwrap();
+        server.publish(alice_handle.clone(), delta).unwrap();
 
         // Two files should now appear in the list
-        let list_reply = server.list(&alice_handle).await.unwrap();
+        let list_reply = server.list(&alice_handle).unwrap();
 
         assert_eq!(2, list_reply.elements().len());
         assert!(find_in_reply(&list_reply, &test::rsync("rsync://localhost/repo/alice/file.txt")).is_some());
@@ -439,7 +447,7 @@ mod tests {
         builder.add_publish(file_outside.as_publish());
         let delta = builder.finish();
 
-        match server.publish(alice_handle.clone(), delta).await {
+        match server.publish(alice_handle.clone(), delta) {
             Err(Error::Rfc8181Delta(PublicationDeltaError::UriOutsideJail(_, _))) => {} // ok
             _ => panic!("Expected error publishing outside of base uri jail"),
         }
@@ -453,7 +461,7 @@ mod tests {
         builder.add_update(file2_update.as_update(file2.hash()));
         let delta = builder.finish();
 
-        match server.publish(alice_handle.clone(), delta).await {
+        match server.publish(alice_handle.clone(), delta) {
             Err(Error::Rfc8181Delta(PublicationDeltaError::NoObjectForHashAndOrUri(_))) => {}
             _ => panic!("Expected error when file for update can't be found"),
         }
@@ -463,7 +471,7 @@ mod tests {
         builder.add_withdraw(file2.as_withdraw());
         let delta = builder.finish();
 
-        match server.publish(alice_handle.clone(), delta).await {
+        match server.publish(alice_handle.clone(), delta) {
             Err(Error::Rfc8181Delta(PublicationDeltaError::NoObjectForHashAndOrUri(_))) => {} // ok
             _ => panic!("Expected error withdrawing file that does not exist"),
         }
@@ -473,7 +481,7 @@ mod tests {
         builder.add_publish(file3.as_publish());
         let delta = builder.finish();
 
-        match server.publish(alice_handle, delta).await {
+        match server.publish(alice_handle, delta) {
             Err(Error::Rfc8181Delta(PublicationDeltaError::ObjectAlreadyPresent(uri))) => {
                 assert_eq!(uri, test::rsync("rsync://localhost/repo/alice/file3.txt"))
             }

src/publish/mod.rs

@@ -41,7 +41,7 @@ impl CaPublisher {
         };
 
         let list_reply = match &repo_contact {
-            RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle).await?,
+            RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle)?,
             RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_list(ca_handle, repo, false).await?,
         };
 
@@ -72,7 +72,7 @@ impl CaPublisher {
         };
 
         match &repo_contact {
-            RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta).await?,
+            RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta)?,
             RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_delta(ca_handle, repo, delta, false).await?,
         };
 
@@ -90,14 +90,14 @@ impl CaPublisher {
         info!("Will perform best effort clean up of old repository: {}", repo);
 
         let list_reply = match repo {
-            RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle).await?,
+            RepositoryContact::Embedded(_) => self.get_embedded()?.list(ca_handle)?,
             RepositoryContact::Rfc8181(repo) => self.caserver.send_rfc8181_list(ca_handle, repo, true).await?,
         };
 
         let delta = list_reply.into_withdraw_delta();
 
         match repo {
-            RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta).await?,
+            RepositoryContact::Embedded(_) => self.get_embedded()?.publish(ca_handle.clone(), delta)?,
             RepositoryContact::Rfc8181(res) => self.caserver.send_rfc8181_delta(ca_handle, res, delta, true).await?,
         }
 

test-resources/krill-init.conf

@@ -22,6 +22,34 @@
 #
 data_dir = "/var/lib/krill/data/"
 
+# Archive publication events after X days. If you do NOT set this
+# value then Krill will not do any archiving.
+#
+# If enabled this option will make sure that the republish events,
+# where your CA simply generates a new Manifest and CRL are archived
+# after the given number of days.
+#
+# If you run Krill as a Publication Server then this option will
+# enable the archiving of publication deltas received by your server
+# from CAs.
+#
+# Archived commands (containing e.g. details of when the change happened),
+# and following events will be moved to an "archived" subdirectory under
+# your data directory as follows:
+#   $data_dir/pubd/0/archived      <-- for Publication Server
+#   $data_dir/cas/ca/archived      <-- for a CA named 'ca'
+#
+# If you want to save space you can delete the files from these archived
+# directories, e.g. from cron. However, you could also archive them in
+# a different way: e.g. compress and move to long term storage. Krill will
+# no longer need this data, but if you ever wanted to see these details in
+# history you would need to move them back into the parent directory of
+# the 'archived' directory.
+#
+# To enable this set the following key value pair.
+#
+### archive_threshold_days = 7
+
 # Specify the path to the PID file for Krill.
 #
 # Defaults to "krill.pid" under the 'data_dir' specified above.