diff --git a/Cargo.lock b/Cargo.lock index 64f3c62b17..41481b8ed7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1686,6 +1686,7 @@ dependencies = [ "actix-tls", "actix-web", "async-recursion", + "async-trait", "base64 0.22.1", "chrono", "clap", diff --git a/crate/access/src/audit/event.rs b/crate/access/src/audit/event.rs index 4828ef9d07..69fc9f73d9 100644 --- a/crate/access/src/audit/event.rs +++ b/crate/access/src/audit/event.rs @@ -2,6 +2,8 @@ use serde::{Deserialize, Serialize}; use time::OffsetDateTime; use uuid::Uuid; +use crate::audit::hash::compute_row_hash; + /// The finalised, persisted audit event including its hash-chain fields. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AuditEvent { @@ -144,6 +146,29 @@ impl AuditEventDraft { details: None, } } + + /// Finalises this draft with the supplied chain position. + /// The resulting hash includes `prev_hash`. + #[must_use] + pub fn finalize(self, id: i64, prev_hash: [u8; 32]) -> AuditEvent { + let mut event = AuditEvent { + id, + timestamp: self.timestamp, + operation: self.operation, + user: self.user, + object_uid: self.object_uid, + algorithm: self.algorithm, + client_ip: self.client_ip, + result: self.result, + duration_ms: self.duration_ms, + request_id: self.request_id, + details: self.details, + prev_hash, + row_hash: [0_u8; 32], + }; + event.row_hash = compute_row_hash(&event); + event + } } /// Current UTC time truncated to **microsecond** resolution. @@ -167,6 +192,7 @@ mod tests { use super::{ AuditEventDraft, AuditResult, OperationAuditContext, RequestAuditContext, audit_now, }; + use crate::audit::hash::verify_event; #[test] fn canonical_str_success() { @@ -283,4 +309,52 @@ mod tests { let ts = audit_now(); assert_eq!(ts.nanosecond() % 1_000, 0); } + + #[test] + fn finalize_assigns_chain_fields_and_verifies() { + let draft = AuditEventDraft { + timestamp: audit_now(), + operation: "Encrypt".to_owned(), + user: "alice@example.com".to_owned(), + object_uid: Some("obj-1234".to_owned()), + algorithm: Some("AES-256-GCM".to_owned()), + client_ip: Some("127.0.0.1".to_owned()), + result: AuditResult::Success, + duration_ms: 5, + request_id: None, + details: None, + }; + let prev_hash = [0xAB_u8; 32]; + let event = draft.finalize(7, prev_hash); + + assert_eq!(event.id, 7); + assert_eq!(event.prev_hash, prev_hash); + assert_eq!(event.operation, "Encrypt"); + assert!( + verify_event(&event), + "finalize() must produce a self-consistent row_hash" + ); + } + + #[test] + fn finalize_is_deterministic_for_identical_input() { + let draft = AuditEventDraft { + timestamp: audit_now(), + operation: "Decrypt".to_owned(), + user: "bob@example.com".to_owned(), + object_uid: None, + algorithm: None, + client_ip: None, + result: AuditResult::Success, + duration_ms: 1, + request_id: None, + details: None, + }; + let a = draft.clone().finalize(0, [0_u8; 32]); + let b = draft.finalize(0, [0_u8; 32]); + assert_eq!( + a.row_hash, b.row_hash, + "same draft/id/prev_hash must yield the same canonical row_hash across backends" + ); + } } diff --git a/crate/interfaces/src/error/mod.rs b/crate/interfaces/src/error/mod.rs index 8a9fd3608c..f9d4343ce8 100644 --- a/crate/interfaces/src/error/mod.rs +++ b/crate/interfaces/src/error/mod.rs @@ -9,6 +9,15 @@ pub enum InterfaceError { #[error("{0}")] Default(String), + /// Wraps a `std::io::Error` with context, keeping `.kind()` inspectable — unlike + /// `Default`, which only keeps the rendered message. + #[error("{context}: {source}")] + Io { + context: String, + #[source] + source: std::io::Error, + }, + #[error("Invalid Request: {0}")] InvalidRequest(String), diff --git a/crate/interfaces/src/stores/audit_sink.rs b/crate/interfaces/src/stores/audit_sink.rs index 608184fc79..9a06bae334 100644 --- a/crate/interfaces/src/stores/audit_sink.rs +++ b/crate/interfaces/src/stores/audit_sink.rs @@ -1,19 +1,6 @@ -//! The `AuditSink` trait: a durable destination for finalised audit events. +//! Durable storage interface for finalised audit events. //! -//! Implemented by each backend that wants to persist the audit hash chain. A sink never -//! assigns ids and never computes hashes — it persists what it is given, in the order it -//! is given, and reports where the chain left off so the writer can resume it. Backends -//! are interchangeable at the trait boundary: a chain started on one backend can be -//! verified after export from another, because both encode the same [`AuditEvent`] and -//! the same canonical hash (see `cosmian_kms_access::audit::canonical_bytes`). -//! -//! # Recovery policy is per-backend, not part of this contract -//! -//! [`AuditSink::resume`] does not mandate a single recovery policy. A backend whose -//! storage can be torn mid-write (an appended file, killed mid-`fsync`) may recover a -//! trustworthy prefix and truncate the rest; a backend whose writes are atomic (a single -//! `INSERT`) has no torn-write case to recover from and can reasonably fail closed on -//! any tail corruption. Document the chosen policy on the implementing type, not here. +//! Each backend owns its recovery policy; the writer owns ids and hashes. use async_trait::async_trait; use cosmian_kms_access::audit::AuditEvent; @@ -29,7 +16,7 @@ pub struct ChainHead { } impl ChainHead { - /// Seed for an empty chain: the first event gets id 0 and an all-zeros `prev_hash`. + /// Chain head before the first event. pub const EMPTY: Self = Self { next_id: 0, prev_hash: [0_u8; 32], @@ -39,20 +26,14 @@ impl ChainHead { /// A durable destination for finalised audit events. /// /// # Contract -/// * `write_event_atomic` : on `Ok` the event is durable; on `Err` nothing was -/// persisted. The writer relies on this — a failed write does not advance -/// `next_id`/`prev_hash`. This ensures that a half-written row does not silently fork the chain. +/// * On `write_event_atomic` success, the event is durable. On error, nothing is persisted. /// * A sink **must never update or delete** a previously written event. #[async_trait] pub trait AuditSink: Send { /// Short sink name for log messages: `"file"`, `"postgres"`. fn name(&self) -> &'static str; - /// Reads the chain head so the writer can resume an existing log. Called exactly - /// once, before any `write_event_atomic`. - /// - /// Recovery policy on a corrupted or unreadable tail is entirely up to the - /// implementation — see the module docs. + /// Recovers the backend and returns the chain head. /// /// # Errors /// Returns an error when the tail cannot be read, or when the implementation's own @@ -66,11 +47,16 @@ pub trait AuditSink: Send { /// not consider the event committed (see the trait-level contract). async fn write_event_atomic(&mut self, event: &AuditEvent) -> InterfaceResult<()>; - /// Called once when the writer loop exits (channel closed on graceful shutdown). + /// Whether the writer should drop events without calling + /// [`Self::write_event_atomic`]. + fn is_write_capacity_exceeded(&self) -> bool { + false + } + + /// Performs backend-specific shutdown synchronisation. /// /// # Errors - /// Returns an error if final synchronisation fails; the writer logs it and exits - /// regardless. + /// Returns an error if synchronisation fails. async fn final_sync(&mut self) -> InterfaceResult<()> { Ok(()) } diff --git a/crate/server/Cargo.toml b/crate/server/Cargo.toml index a5f09cef30..b15dc72884 100644 --- a/crate/server/Cargo.toml +++ b/crate/server/Cargo.toml @@ -58,6 +58,7 @@ actix-session = { workspace = true, features = ["cookie-session"] } actix-tls = { workspace = true } actix-web = { workspace = true, features = ["macros", "openssl"] } async-recursion = { workspace = true } +async-trait = { workspace = true } base64 = { workspace = true } chrono = { workspace = true } clap = { workspace = true, features = [ diff --git a/crate/server/src/core/audit/file_sink.rs b/crate/server/src/core/audit/file_sink.rs index bde8d0ab8e..5a172ee19a 100644 --- a/crate/server/src/core/audit/file_sink.rs +++ b/crate/server/src/core/audit/file_sink.rs @@ -1,46 +1,28 @@ -//! Tamper-evident JSONL file persistence for the audit log: writer lifecycle, the -//! exclusive cross-instance lock, and the `AuditSink` write abstraction. +//! Tamper-evident JSONL audit sink with recovery and an exclusive cross-instance lock. //! -//! Always-start recovery -//! ====================== -//! `writer_supervisor` guarantees the KMS always starts regardless of the audit log's -//! state: it acquires the exclusive lock and recovers/opens the file (see the -//! `recovery` module for tail classification and seal-and-roll) inside a self-healing -//! retry loop that runs in the background, never blocking the caller. +//! Recovery runs in the writer task, allowing server startup to continue independently. use std::{ ffi::{OsStr, OsString}, io::Write, path::{Path, PathBuf}, - sync::{Arc, atomic::AtomicU64}, + sync::{Arc, atomic::Ordering}, }; -use cosmian_kms_access::audit::AuditEvent; +use async_trait::async_trait; +use cosmian_kms_access::audit::{AuditEvent, AuditEventDraft}; +use cosmian_kms_interfaces::{AuditSink, ChainHead, InterfaceError, InterfaceResult}; use cosmian_logger::{debug, error}; -use tokio::sync::mpsc; -use super::{recovery::recover_and_open, store::WriterMsg, writer::writer_loop}; +use super::recovery::recover_and_open; -/// How long to wait between attempts to acquire the exclusive audit-log lock while a -/// peer instance (e.g. the other side of a rolling update on a shared volume) holds it. const LOCK_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500); -/// How long to wait between attempts to recover/open the audit log after a -/// content-independent I/O fault (EACCES, EIO, read-only mount, missing disk). const OPEN_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500); -/// Minimum interval between "still capped" debug log lines while blocked events -/// keep arriving — avoids flooding the log once `max_size_bytes` is reached. -pub(super) const CAPPED_DEBUG_LOG_INTERVAL: std::time::Duration = - std::time::Duration::from_millis(500); - /// Cross-task state for the optional `max_size_bytes` write-stop cap. /// -/// `enqueue()` reads `size_limit_reached` as a fast, non-blocking pre-check so a -/// caller doesn't bother queueing an event the writer will only ever discard; the -/// writer task is the sole owner of the file and the only one that ever sets it. -/// `max_size_bytes` is immutable for the store's lifetime, carried alongside so -/// the writer doesn't need it threaded through as a separate argument everywhere. +/// The writer sets `size_limit_reached`; producers read it before enqueueing. #[derive(Default)] pub(super) struct AuditWriteState { pub(super) max_size_bytes: Option, @@ -56,62 +38,68 @@ impl AuditWriteState { } } -/// Abstraction over the audit log's underlying writer. -/// -/// This exists so the fault path (`write_event` failing mid-run) can be -/// exercised in tests with a mock sink, without touching real files — -/// production always uses `FileSink`. -pub(super) trait AuditSink { - /// Serialises and durably persists one event. - /// - /// # Errors - /// On failure, the caller must NOT consider the event committed: the writer does - /// not advance `next_id`/`prev_hash`, and the sink guarantees the partially written - /// bytes are never observable as a chain row — either because the write is atomic, - /// or because the sink discards them before its next successful write. - fn write_event(&mut self, event: &AuditEvent) -> std::io::Result<()>; - - /// Called once when the writer loop exits (channel closed). Default is a - /// no-op. - fn final_sync(&mut self) -> std::io::Result<()> { - Ok(()) - } +/// Writes and synchronises one JSONL event during recovery. +fn write_event_line(file: &mut std::fs::File, event: &AuditEvent) -> std::io::Result<()> { + serde_json::to_writer(&mut *file, event) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + file.write_all(b"\n")?; + file.sync_data() +} - /// Current on-disk length in bytes, used to enforce `max_size_bytes`. - /// - /// Only the real `std::fs::File` sink can answer this meaningfully; mock sinks - /// used for fault-injection tests never configure a size cap, so the default - /// (`Ok(0)`) is never exercised by them. - fn current_len(&self) -> std::io::Result { - Ok(0) +/// Writes a recovery sentinel, advancing the chain only after a successful sync. +/// This synchronous path runs before the sink is resumed. +pub(super) fn write_recovery_sentinel( + file: &mut std::fs::File, + draft: AuditEventDraft, + next_id: i64, + prev_hash: &mut [u8; 32], +) -> i64 { + let event = draft.finalize(next_id, *prev_hash); + match write_event_line(file, &event) { + Ok(()) => { + *prev_hash = event.row_hash; + next_id.checked_add(1).unwrap_or_else(|| { + error!("AuditFileStore: recovery sentinel id counter overflow at i64::MAX"); + next_id + }) + } + Err(e) => { + error!( + "AuditFileStore: failed to write recovery sentinel id={}: {e} — event dropped", + event.id + ); + next_id + } } } -/// A file-backed sink that lazily repairs a torn tail left by a previous failed -/// write, instead of waiting for the next process restart to discover it. +/// File-backed [`AuditSink`] for a tamper-evident JSONL chain. /// -/// A write can fail after some of its bytes already hit the OS (a short write, or an -/// error between the JSON and its trailing newline): the row is not committed, but the -/// bytes are already durable. Truncating back to `committed_len` closes that gap the -/// same way `classify_tail`'s `TruncateContinue` does at boot — just triggered by the -/// next write instead of the next restart. -pub(super) struct FileSink { - file: std::fs::File, - /// End of the last durably written, complete row. +/// A failed write may leave a partial row. The next write truncates back to +/// `committed_len` before appending. +pub(crate) struct FileSink { + path: PathBuf, + write_state: Arc, + /// Set by [`Self::resume`]. + file: Option, + /// End of the last durably written row. committed_len: u64, - /// A previous write left bytes past `committed_len` that must be discarded before - /// the next append. + /// Whether bytes past `committed_len` must be discarded. needs_repair: bool, + /// Kept alive because dropping the handle releases the OS lock; never read again. + _lock: Option, } impl FileSink { - /// `committed_len` must be `file`'s length at the time of opening — every byte up - /// to it is a durable, complete row (guaranteed by `recovery::recover_and_open`). - pub(super) const fn new(file: std::fs::File, committed_len: u64) -> Self { + /// Builds a file sink that has not yet been resumed. + pub(super) const fn new(path: PathBuf, write_state: Arc) -> Self { Self { - file, - committed_len, + path, + write_state, + file: None, + committed_len: 0, needs_repair: false, + _lock: None, } } @@ -119,83 +107,156 @@ impl FileSink { if !self.needs_repair { return Ok(()); } - self.file.set_len(self.committed_len)?; - self.file.sync_data()?; + if let Some(file) = self.file.as_mut() { + file.set_len(self.committed_len)?; + file.sync_data()?; + } self.needs_repair = false; Ok(()) } } +#[async_trait] impl AuditSink for FileSink { - /// Serialises `event` as a single JSONL line, repairing any torn tail from a - /// previous failed write first so a partial row is never observable mid-session. + fn name(&self) -> &'static str { + "file" + } + + /// Waits for the exclusive lock, then recovers and opens the audit file. Every fault + /// — lock contention, recovery/open, and stat — retries in place, so this never + /// returns `Err` in practice. + async fn resume(&mut self) -> InterfaceResult { + let lock_path = lock_file_path(&self.path); + let mut lock_contended_logged = false; + let lock = loop { + match try_acquire_lock(&lock_path) { + Ok(lock) => break lock, + // `try_acquire_lock` also fails for reasons unrelated to contention (EACCES, + // EROFS, directory-creation failure) — only `WouldBlock` means a peer holds + // the lock; anything else is a deployment fault and must be logged as such, + // not masked as the (benign, expected-in-HA) "held by another instance" case. + Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { + if lock_contended_logged { + debug!( + "AuditFileStore: still waiting on audit log lock {} ({e})", + lock_path.display() + ); + } else { + error!( + "AuditFileStore: audit log lock {} held by another instance ({e}) — \ + buffering events until it is released", + lock_path.display() + ); + lock_contended_logged = true; + } + tokio::time::sleep(LOCK_RETRY_INTERVAL).await; + } + Err(e) => { + error!( + "AuditFileStore: cannot acquire audit log lock {} ({e}) — retrying", + lock_path.display() + ); + tokio::time::sleep(LOCK_RETRY_INTERVAL).await; + } + } + }; + self._lock = Some(lock); + + let (file, next_id, prev_hash, committed_len) = loop { + // Recovery scans the file and may rename it, so keep it off the async worker. + let path_for_recovery = self.path.clone(); + let recovered = tokio::task::spawn_blocking(move || { + let (file, next_id, prev_hash) = recover_and_open(&path_for_recovery)?; + let committed_len = file.metadata()?.len(); + Ok::<_, crate::error::KmsError>((file, next_id, prev_hash, committed_len)) + }) + .await; + match recovered { + Ok(Ok(quadruple)) => break quadruple, + Ok(Err(e)) => { + error!( + "AuditFileStore: cannot open or stat audit log {} ({e}) — retrying", + self.path.display() + ); + tokio::time::sleep(OPEN_RETRY_INTERVAL).await; + } + Err(join_err) => { + error!("AuditFileStore: recovery task failed to run ({join_err}) — retrying"); + tokio::time::sleep(OPEN_RETRY_INTERVAL).await; + } + } + }; + + self.committed_len = committed_len; + enforce_size_cap(self.committed_len, &self.write_state, &self.path); + self.file = Some(file); + Ok(ChainHead { next_id, prev_hash }) + } + + /// Serialises and synchronises one event, repairing any previous partial write first. /// - /// `sync_data()` is called on every write to guarantee durability: without it - /// data sits in the kernel page cache and is lost on a power failure. The - /// tradeoff is one `fsync` per audit event; high-throughput deployments can - /// reduce cost by batching syncs (every N events or every T ms). - fn write_event(&mut self, event: &AuditEvent) -> std::io::Result<()> { - // Serialize first: a serialization failure must never touch the file. + /// # Errors + /// Returns an error if the sink is not resumed or file I/O fails. + async fn write_event_atomic(&mut self, event: &AuditEvent) -> InterfaceResult<()> { let mut row = serde_json::to_vec(event) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + .map_err(|e| InterfaceError::Default(format!("audit: cannot serialise event: {e}")))?; row.push(b'\n'); - self.repair_if_needed()?; - - match self - .file - .write_all(&row) - .and_then(|()| self.file.sync_data()) - { + self.repair_if_needed().map_err(|e| InterfaceError::Io { + context: "audit: torn-tail repair failed".to_owned(), + source: e, + })?; + + let file = self.file.as_mut().ok_or_else(|| { + InterfaceError::Default( + "audit: FileSink::write_event_atomic called before resume()".to_owned(), + ) + })?; + match file.write_all(&row).and_then(|()| file.sync_data()) { Ok(()) => { self.committed_len += u64::try_from(row.len()).unwrap_or(u64::MAX); + enforce_size_cap(self.committed_len, &self.write_state, &self.path); Ok(()) } Err(e) => { self.needs_repair = true; - Err(e) + Err(InterfaceError::Io { + context: "audit: write failed".to_owned(), + source: e, + }) } } } - /// Repairs a torn tail before the final sync so a clean shutdown never leaves one - /// for the next boot to find. - fn final_sync(&mut self) -> std::io::Result<()> { - self.repair_if_needed()?; - self.file.sync_data() + fn is_write_capacity_exceeded(&self) -> bool { + self.write_state.size_limit_reached.load(Ordering::Relaxed) } - fn current_len(&self) -> std::io::Result { - Ok(self.committed_len) + async fn final_sync(&mut self) -> InterfaceResult<()> { + self.repair_if_needed().map_err(|e| InterfaceError::Io { + context: "audit: final sync repair failed".to_owned(), + source: e, + })?; + if let Some(file) = self.file.as_mut() { + file.sync_data().map_err(|e| InterfaceError::Io { + context: "audit: final sync failed".to_owned(), + source: e, + })?; + } + Ok(()) } } -/// Checks `sink`'s current on-disk length against `write_state.max_size_bytes` and -/// updates `write_state` on the first transition into the capped state (logging -/// once). Called right after the writer opens/recovers the file (an already- -/// oversized log must block immediately) and again after every successful write -/// (a write that crosses the cap is allowed to land, then blocks everything after -/// it). -pub(super) fn enforce_size_cap(sink: &S, write_state: &AuditWriteState, path: &Path) { +/// Marks the sink capped once its committed length reaches the configured limit. +/// The write crossing the limit remains committed. +fn enforce_size_cap(len: u64, write_state: &AuditWriteState, path: &Path) { let Some(cap) = write_state.max_size_bytes else { return; }; - let len = match sink.current_len() { - Ok(len) => len, - Err(e) => { - error!( - "AuditFileStore: cannot stat audit log {} ({e})", - path.display() - ); - return; - } - }; if len < cap { return; } - let was_already_capped = write_state - .size_limit_reached - .swap(true, std::sync::atomic::Ordering::Relaxed); + let was_already_capped = write_state.size_limit_reached.swap(true, Ordering::Relaxed); if !was_already_capped { error!( "AuditFileStore: audit log {} reached its configured max_size_bytes cap \ @@ -206,7 +267,6 @@ pub(super) fn enforce_size_cap(sink: &S, write_state: &AuditWriteS } } -/// Builds the sidecar lock file path for `path`, e.g. `audit.jsonl` -> `audit.jsonl.lock`. pub(super) fn lock_file_path(path: &Path) -> PathBuf { let mut name = path .file_name() @@ -215,12 +275,8 @@ pub(super) fn lock_file_path(path: &Path) -> PathBuf { path.with_file_name(name) } -/// Attempts to acquire the exclusive, cross-platform advisory lock on `path`'s lock -/// sidecar. Non-blocking: returns immediately (`Err` if another live instance holds it). -/// -/// The returned `File` must be kept alive for as long as the lock should be held — the OS -/// releases it automatically when the handle is dropped or the process exits, so a crash -/// never leaves a stale lock behind. +/// Attempts to acquire the advisory lock without blocking. +/// Dropping the returned file releases the lock. fn try_acquire_lock(lock_path: &Path) -> std::io::Result { if let Some(parent) = lock_path.parent() { std::fs::create_dir_all(parent)?; @@ -240,92 +296,6 @@ fn try_acquire_lock(lock_path: &Path) -> std::io::Result { } } -/// Supervises the writer's lifecycle so the KMS always starts, regardless of the audit -/// log's state: acquires the exclusive lock (retrying in the background, without draining -/// the channel, if a peer holds it), recovers/opens the file (retrying if the path is -/// unwritable — EACCES, EIO, a read-only mount — so audit logging self-heals the moment -/// the fault clears), then runs the normal `writer_loop`. -/// -/// Events enqueued while waiting for either step are genuinely queued in the channel and -/// flushed in order once the writer proceeds — they are not dropped. Only a channel that -/// fills to capacity during the wait spills to drop + eviction-sentinel, exactly like -/// saturation during normal operation. -pub(super) async fn writer_supervisor( - path: PathBuf, - rx: mpsc::Receiver, - dropped_count: Arc, - write_state: Arc, -) { - let lock_path = lock_file_path(&path); - let mut lock_contended_logged = false; - let _lock = loop { - match try_acquire_lock(&lock_path) { - Ok(lock) => break lock, - // `try_acquire_lock` also fails for reasons unrelated to contention (EACCES, - // EROFS, directory-creation failure) — only `WouldBlock` means a peer holds - // the lock; anything else is a deployment fault and must be logged as such, - // not masked as the (benign, expected-in-HA) "held by another instance" case. - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => { - if lock_contended_logged { - debug!( - "AuditFileStore: still waiting on audit log lock {} ({e})", - lock_path.display() - ); - } else { - error!( - "AuditFileStore: audit log lock {} held by another instance ({e}) — \ - buffering events until it is released", - lock_path.display() - ); - lock_contended_logged = true; - } - tokio::time::sleep(LOCK_RETRY_INTERVAL).await; - } - Err(e) => { - error!( - "AuditFileStore: cannot acquire audit log lock {} ({e}) — retrying", - lock_path.display() - ); - tokio::time::sleep(LOCK_RETRY_INTERVAL).await; - } - } - }; - - let (sink, next_id, prev_hash) = loop { - // `recover_and_open` does blocking `std::fs` I/O (whole-file scan, hash, and - // possible rename on seal-and-roll) — run it on the blocking pool so a large - // audit log doesn't monopolize this tokio worker thread during startup/recovery. - let path_for_recovery = path.clone(); - let recovered = - tokio::task::spawn_blocking(move || recover_and_open(&path_for_recovery)).await; - match recovered { - Ok(Ok(triple)) => break triple, - Ok(Err(e)) => { - error!( - "AuditFileStore: cannot open audit log {} ({e}) — retrying", - path.display() - ); - tokio::time::sleep(OPEN_RETRY_INTERVAL).await; - } - Err(join_err) => { - error!("AuditFileStore: recovery task failed to run ({join_err}) — retrying"); - tokio::time::sleep(OPEN_RETRY_INTERVAL).await; - } - } - }; - - writer_loop( - sink, - next_id, - prev_hash, - rx, - dropped_count, - write_state, - &path, - ) - .await; -} - #[cfg(test)] #[allow( clippy::unwrap_used, @@ -388,20 +358,32 @@ mod tests { .expect("open for append") } - #[test] - fn write_event_appends_valid_row_and_updates_committed_len() { + fn make_sink(file: std::fs::File, committed_len: u64) -> FileSink { + FileSink { + path: PathBuf::new(), + write_state: Arc::new(AuditWriteState::default()), + file: Some(file), + committed_len, + needs_repair: false, + _lock: None, + } + } + + #[tokio::test] + async fn write_event_appends_valid_row_and_updates_committed_len() { let path = temp_path("basic"); std::fs::remove_file(&path).ok(); - let mut sink = FileSink::new(open_for_append(&path), 0); + let mut sink = make_sink(open_for_append(&path), 0); - sink.write_event(&sample_event(0, [0_u8; 32])) + sink.write_event_atomic(&sample_event(0, [0_u8; 32])) + .await .expect("write"); let rows = read_rows(&path); assert_eq!(rows.len(), 1); assert!(verify_event(&rows[0])); assert_eq!( - sink.current_len().expect("current_len"), + sink.committed_len, std::fs::metadata(&path).expect("metadata").len() ); } @@ -411,8 +393,8 @@ mod tests { /// `write_draft_to_chain` does) must leave exactly one valid row, not a /// concatenation with whatever the failed attempt left behind. #[cfg(not(target_os = "windows"))] // Fails on Windows, probably for reasons related to file handles (passes reliably on Linux/macOS). - #[test] - fn write_failure_sets_needs_repair_and_is_healed_by_next_success() { + #[tokio::test] + async fn write_failure_sets_needs_repair_and_is_healed_by_next_success() { let path = temp_path("heal"); std::fs::remove_file(&path).ok(); std::fs::File::create(&path).expect("create"); @@ -420,16 +402,18 @@ mod tests { .read(true) .open(&path) .expect("open read-only"); - let mut sink = FileSink::new(read_only, 0); + let mut sink = make_sink(read_only, 0); let event = sample_event(0, [0_u8; 32]); - assert!(sink.write_event(&event).is_err()); + assert!(sink.write_event_atomic(&event).await.is_err()); assert!(sink.needs_repair); // The underlying handle becomes writable again — mirrors the process // continuing to run and retrying the same slot. - sink.file = open_for_append(&path); - sink.write_event(&event).expect("retry succeeds"); + sink.file = Some(open_for_append(&path)); + sink.write_event_atomic(&event) + .await + .expect("retry succeeds"); assert!(!sink.needs_repair); let rows = read_rows(&path); @@ -438,30 +422,33 @@ mod tests { } #[cfg(not(target_os = "windows"))] // Fails on Windows, probably for reasons related to file handles (passes reliably on Linux/macOS). - #[test] - fn final_sync_repairs_pending_tail_without_writing_a_new_row() { + #[tokio::test] + async fn final_sync_repairs_pending_tail_without_writing_a_new_row() { let path = temp_path("final_sync"); std::fs::remove_file(&path).ok(); - let mut sink = FileSink::new(open_for_append(&path), 0); + let mut sink = make_sink(open_for_append(&path), 0); sink.file + .as_mut() + .expect("file") .write_all(b"torn-garbage-without-newline") .expect("write garbage"); sink.needs_repair = true; - sink.final_sync().expect("final_sync"); + sink.final_sync().await.expect("final_sync"); assert_eq!(std::fs::metadata(&path).expect("metadata").len(), 0); assert!(!sink.needs_repair); } - #[test] - fn current_len_reports_committed_view_not_raw_file_length() { + #[tokio::test] + async fn committed_len_is_isolated_from_writes_behind_the_sinks_back() { let path = temp_path("committed_view"); std::fs::remove_file(&path).ok(); - let mut sink = FileSink::new(open_for_append(&path), 0); - sink.write_event(&sample_event(0, [0_u8; 32])) + let mut sink = make_sink(open_for_append(&path), 0); + sink.write_event_atomic(&sample_event(0, [0_u8; 32])) + .await .expect("write"); - let committed = sink.current_len().expect("current_len"); + let committed = sink.committed_len; // Bytes land on disk without going through the sink (e.g. a future write // not yet reflected in `committed_len`). @@ -469,7 +456,7 @@ mod tests { .write_all(b"unrelated-bytes") .expect("write unrelated bytes"); - assert_eq!(sink.current_len().expect("current_len"), committed); + assert_eq!(sink.committed_len, committed); assert_ne!(std::fs::metadata(&path).expect("metadata").len(), committed); } } diff --git a/crate/server/src/core/audit/recovery.rs b/crate/server/src/core/audit/recovery.rs index 7e5cea3879..dc194bfc0b 100644 --- a/crate/server/src/core/audit/recovery.rs +++ b/crate/server/src/core/audit/recovery.rs @@ -24,8 +24,7 @@ use cosmian_kms_access::audit::{ use cosmian_logger::{debug, error}; use time::OffsetDateTime; -use super::file_sink::FileSink; -use super::writer::write_draft_to_chain; +use super::file_sink::write_recovery_sentinel; use crate::{error::KmsError, result::KResult}; /// Bytes read from the end of an existing log to locate the last complete event @@ -475,15 +474,12 @@ fn truncate_and_continue( {discard_offset} (process likely killed mid-write); resuming chain at id={next_id}" ); - let mut sink = FileSink::new( - open_append(path).map_err(|e| { - KmsError::ServerError(format!( - "audit: cannot reopen log file after truncation {}: {e}", - path.display() - )) - })?, - keep_len, - ); + let mut sink = open_append(path).map_err(|e| { + KmsError::ServerError(format!( + "audit: cannot reopen log file after truncation {}: {e}", + path.display() + )) + })?; let details = serde_json::json!({ "bytes_discarded": bytes_discarded, @@ -503,7 +499,7 @@ fn truncate_and_continue( details: Some(details), }; let mut chain_prev_hash = prev_hash; - let final_next_id = write_draft_to_chain(&mut sink, draft, next_id, &mut chain_prev_hash); + let final_next_id = write_recovery_sentinel(&mut sink, draft, next_id, &mut chain_prev_hash); Ok((final_next_id, chain_prev_hash)) } @@ -573,15 +569,12 @@ fn seal_and_roll( } } - let mut sink = FileSink::new( - open_append(path).map_err(|e| { - KmsError::ServerError(format!( - "audit: cannot open fresh log file {}: {e}", - path.display() - )) - })?, - 0, - ); + let mut sink = open_append(path).map_err(|e| { + KmsError::ServerError(format!( + "audit: cannot open fresh log file {}: {e}", + path.display() + )) + })?; let sealed_name = sealed_path .file_name() @@ -612,7 +605,7 @@ fn seal_and_roll( // Reanchor is a new chain root — continuity across a sealed, untrusted tail is never // asserted (see module docs). let mut prev_hash = [0_u8; 32]; - let next_id = write_draft_to_chain(&mut sink, draft, 0, &mut prev_hash); + let next_id = write_recovery_sentinel(&mut sink, draft, 0, &mut prev_hash); Ok((next_id, prev_hash)) } @@ -642,7 +635,7 @@ fn open_append(path: &Path) -> std::io::Result { /// Returns an error only for content-independent I/O faults (cannot read/truncate/rename/ /// open); a data-corruption condition is always routed to a `TailOutcome` variant instead /// and handled without error (see `classify_tail`). -pub(super) fn recover_and_open(path: &Path) -> KResult<(FileSink, i64, [u8; 32])> { +pub(super) fn recover_and_open(path: &Path) -> KResult<(std::fs::File, i64, [u8; 32])> { let verification = verify_interior_chain(path)?; let (next_id, prev_hash) = if let Some(failure) = verification.failure { seal_and_roll( @@ -718,15 +711,6 @@ pub(super) fn recover_and_open(path: &Path) -> KResult<(FileSink, i64, [u8; 32]) path.display() )) })?; - let committed_len = file - .metadata() - .map_err(|e| { - KmsError::ServerError(format!( - "audit: cannot stat log file {}: {e}", - path.display() - )) - })? - .len(); - Ok((FileSink::new(file, committed_len), next_id, prev_hash)) + Ok((file, next_id, prev_hash)) } diff --git a/crate/server/src/core/audit/store.rs b/crate/server/src/core/audit/store.rs index 791dd707a6..be55a74973 100644 --- a/crate/server/src/core/audit/store.rs +++ b/crate/server/src/core/audit/store.rs @@ -1,18 +1,7 @@ -//! `AuditFileStore`: a cheaply cloneable handle to the audit writer task. +//! Non-blocking handle to the file audit writer. //! -//! * `AuditFileStore` is a cheaply cloneable handle (wraps a channel `Sender`). -//! * A single background tokio task (`writer_supervisor`, in `file_sink`) is the **sole -//! owner** of the audit file, the monotonic event counter, and the previous-row hash. -//! This design avoids any mutex around the file and guarantees write order under -//! concurrent requests. -//! * The KMS always starts: `start_with_max_size()` returns synchronously and never -//! blocks on file I/O or lock contention. Recovery, exclusive-lock acquisition, and -//! opening the file all happen inside the writer task, retrying in the background on -//! failure. Events enqueued in the meantime are genuinely queued (not dropped) up to -//! the channel's bounded capacity — see `file_sink::writer_supervisor`. -//! * The middleware calls `enqueue()` which is a non-blocking `try_send`. If the -//! channel is full (beyond the configured capacity) the draft is silently dropped -//! and an error is logged — we never block the request path. +//! One task owns the sink and chain head. Producers communicate through a bounded +//! channel; overflowed events are dropped and counted. use std::{ path::Path, @@ -23,10 +12,14 @@ use std::{ }; use cosmian_kms_access::audit::AuditEventDraft; +use cosmian_kms_interfaces::AuditSink; use cosmian_logger::error; use tokio::sync::{mpsc, oneshot}; -use super::file_sink::{AuditWriteState, writer_supervisor}; +use super::{ + file_sink::{AuditWriteState, FileSink}, + writer::writer_loop, +}; use crate::{error::KmsError, result::KResult}; /// Message sent to the writer task over the channel. @@ -43,11 +36,7 @@ pub(super) enum WriterMsg { Flush(oneshot::Sender<()>), } -/// A cheaply cloneable handle to the audit writer task. -/// -/// Cloning this value is O(1) — `tokio::sync::mpsc::Sender` is already backed -/// by an internal `Arc`, and `dropped_count`/`write_state` are themselves `Arc`s. -/// All clones share the same underlying channel and writer task. +/// Cloneable handle to the file audit writer task. #[derive(Clone)] pub(crate) struct AuditFileStore { sender: mpsc::Sender, @@ -59,25 +48,11 @@ pub(crate) struct AuditFileStore { } impl AuditFileStore { - /// Initialises the audit file store and spawns the background writer task. - /// - /// Returns immediately: the channel is created and handed back synchronously so the - /// middleware can start enqueueing events right away, even before the writer has - /// acquired the lock or opened the file. `channel_capacity` is the number of events - /// that can be buffered before new events are dropped. Must be ≥ 1. - /// - /// `max_size_bytes`, when `Some`, stops all writes once the file reaches that many - /// bytes — see `AuditFileConfig::audit_file_max_size_bytes`. `None` is unlimited. - /// - /// Recovery, locking, and opening all happen inside the spawned writer task — see - /// `file_sink::writer_supervisor`. This call never blocks on file I/O or lock - /// contention. + /// Spawns the writer and returns without waiting for file recovery. + /// `max_size_bytes = None` disables the size limit. /// /// # Errors - /// Returns an error only if `channel_capacity` is 0 — a pure configuration mistake, - /// not a runtime condition. Every other fault (I/O, lock contention, log corruption) - /// is handled inside the writer task without aborting startup; see - /// `file_sink::writer_supervisor`. + /// Returns an error if `channel_capacity` is zero. pub(crate) fn start_with_max_size( path: &Path, channel_capacity: usize, @@ -97,7 +72,25 @@ impl AuditFileStore { let path = path.to_path_buf(); tokio::spawn(async move { - writer_supervisor(path, rx, dropped_count_for_writer, write_state_for_writer).await; + let mut sink = FileSink::new(path, write_state_for_writer); + match sink.resume().await { + Ok(chain_head) => { + writer_loop( + sink, + chain_head.next_id, + chain_head.prev_hash, + rx, + dropped_count_for_writer, + ) + .await; + } + Err(e) => { + error!( + "AuditFileStore: audit sink failed to resume ({e}) — audit logging is \ + disabled for this process" + ); + } + } }); Ok(Self { @@ -203,17 +196,16 @@ mod tests { sync::{Arc, atomic::AtomicU64}, }; + use async_trait::async_trait; use cosmian_kms_access::audit::{ AuditEvent, AuditEventDraft, AuditResult, compute_row_hash, verify_event, }; + use cosmian_kms_interfaces::{AuditSink, ChainHead, InterfaceError, InterfaceResult}; use time::OffsetDateTime; use tokio::sync::mpsc; - use super::{AuditFileStore, AuditWriteState, WriterMsg}; - use crate::core::audit::{ - file_sink::{AuditSink, lock_file_path}, - writer::writer_loop, - }; + use super::{AuditFileStore, WriterMsg}; + use crate::core::audit::{file_sink::lock_file_path, writer::writer_loop}; /// Small channel capacity used in all tests. Large enough for the ≤5-event /// functional tests; small enough to fill quickly in the saturation test. @@ -923,7 +915,7 @@ mod tests { // ── Fault injection on the write path ──────────────────────────────── - /// A mock `AuditSink` that fails `write_event` for calls whose 0-based + /// A mock `AuditSink` that fails `write_event_atomic` for calls whose 0-based /// index satisfies `should_fail`, allowing precise control over exactly /// which write in a sequence fails. struct FaultySink { @@ -942,12 +934,23 @@ mod tests { } } + #[async_trait] impl AuditSink for FaultySink { - fn write_event(&mut self, event: &AuditEvent) -> std::io::Result<()> { + fn name(&self) -> &'static str { + "faulty" + } + + async fn resume(&mut self) -> InterfaceResult { + Ok(ChainHead::EMPTY) + } + + async fn write_event_atomic(&mut self, event: &AuditEvent) -> InterfaceResult<()> { let idx = self.call_count; self.call_count += 1; if (self.should_fail)(idx) { - return Err(std::io::Error::other("simulated write failure")); + return Err(InterfaceError::Default( + "simulated write failure".to_owned(), + )); } self.events.push(event.clone()); Ok(()) @@ -965,18 +968,8 @@ mod tests { // Fail exactly the 3rd write call (0-based index 2). let sink = FaultySink::new(|idx| idx == 2); - let handle = tokio::spawn(async move { - writer_loop( - sink, - 0, - [0_u8; 32], - rx, - dropped_count, - Arc::new(AuditWriteState::default()), - Path::new("faulty_sink_test"), - ) - .await - }); + let handle = + tokio::spawn(async move { writer_loop(sink, 0, [0_u8; 32], rx, dropped_count).await }); for _ in 0..5 { tx.send(WriterMsg::Event(Box::new(make_draft()))) @@ -1009,18 +1002,8 @@ mod tests { let dropped_count = Arc::new(AtomicU64::new(0)); let sink = FaultySink::new(|_| true); - let handle = tokio::spawn(async move { - writer_loop( - sink, - 0, - [0_u8; 32], - rx, - dropped_count, - Arc::new(AuditWriteState::default()), - Path::new("faulty_sink_test"), - ) - .await - }); + let handle = + tokio::spawn(async move { writer_loop(sink, 0, [0_u8; 32], rx, dropped_count).await }); for _ in 0..3 { tx.send(WriterMsg::Event(Box::new(make_draft()))) diff --git a/crate/server/src/core/audit/writer.rs b/crate/server/src/core/audit/writer.rs index b47081bd7d..fb6a4af04c 100644 --- a/crate/server/src/core/audit/writer.rs +++ b/crate/server/src/core/audit/writer.rs @@ -1,36 +1,28 @@ -//! The background audit writer task: sole owner of the sink, the id counter, and -//! `prev_hash`. Designed not to panic — errors are logged and the loop continues. +//! Shared audit writer loop. +//! +//! The task owns the sink and chain head; backends provide persistence and recovery. -use std::{ - path::Path, - sync::{Arc, atomic::AtomicU64}, -}; +use std::sync::{Arc, atomic::AtomicU64}; -use cosmian_kms_access::audit::{AuditEvent, AuditEventDraft, AuditResult, compute_row_hash}; +use cosmian_kms_access::audit::{AuditEventDraft, AuditResult}; +use cosmian_kms_interfaces::AuditSink; use cosmian_logger::{debug, error}; use time::OffsetDateTime; use tokio::sync::mpsc; -use super::{ - file_sink::{AuditSink, AuditWriteState, CAPPED_DEBUG_LOG_INTERVAL, enforce_size_cap}, - store::WriterMsg, -}; +use super::store::WriterMsg; -/// The background writer task. Sole owner of the sink, the id counter, and -/// `prev_hash`. Designed not to panic — errors are logged and the loop -/// continues. Calls `final_sync()` before exiting so in-flight events are -/// durable on graceful shutdown. Returns the sink so tests can inspect what -/// was actually persisted. +/// Minimum interval between capacity warnings. +const CAPPED_DEBUG_LOG_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500); + +/// Consumes queued events and returns the sink after shutdown synchronisation. pub(super) async fn writer_loop( mut sink: S, mut next_id: i64, mut prev_hash: [u8; 32], mut rx: mpsc::Receiver, dropped_count: Arc, - write_state: Arc, - path: &Path, ) -> S { - enforce_size_cap(&sink, &write_state, path); let mut last_capped_log: Option = None; while let Some(msg) = rx.recv().await { @@ -44,17 +36,14 @@ pub(super) async fn writer_loop( } }; - if write_state - .size_limit_reached - .load(std::sync::atomic::Ordering::Relaxed) - { + if sink.is_write_capacity_exceeded() { let now = std::time::Instant::now(); let should_log = last_capped_log .is_none_or(|logged_at| now.duration_since(logged_at) >= CAPPED_DEBUG_LOG_INTERVAL); if should_log { debug!( - "AuditFileStore: audit log {} is at its max_size_bytes cap — event dropped", - path.display() + "AuditFileStore: sink '{}' is at capacity — event dropped", + sink.name() ); last_capped_log = Some(now); } @@ -65,60 +54,37 @@ pub(super) async fn writer_loop( let n_dropped = dropped_count.swap(0, std::sync::atomic::Ordering::Relaxed); if n_dropped > 0 { let sentinel = make_eviction_sentinel(n_dropped); - next_id = write_draft_to_chain(&mut sink, sentinel, next_id, &mut prev_hash); - enforce_size_cap(&sink, &write_state, path); + next_id = write_draft_to_chain(&mut sink, sentinel, next_id, &mut prev_hash).await; } - if write_state - .size_limit_reached - .load(std::sync::atomic::Ordering::Relaxed) - { + if sink.is_write_capacity_exceeded() { // The sentinel write alone just crossed the cap: writing the real draft too // would overshoot the documented "one final event may cross" rule by a // second event. Count it as dropped so a future sentinel reports it. dropped_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); } else { - next_id = write_draft_to_chain(&mut sink, draft, next_id, &mut prev_hash); - enforce_size_cap(&sink, &write_state, path); + next_id = write_draft_to_chain(&mut sink, draft, next_id, &mut prev_hash).await; } } - // Channel closed (sender dropped on graceful shutdown): ensure all written - // events are durable before the task exits. - if let Err(e) = sink.final_sync() { + if let Err(e) = sink.final_sync().await { error!("AuditFileStore: final sync failed: {e}"); } debug!("AuditFileStore: writer loop exited (channel closed)"); sink } -/// Finalises and writes a single `AuditEventDraft` into the chain, advancing -/// `next_id` and `prev_hash` on success. Returns the new `next_id`. -pub(super) fn write_draft_to_chain( +/// Writes one draft and advances the chain head only on success. +pub(super) async fn write_draft_to_chain( sink: &mut S, draft: AuditEventDraft, next_id: i64, prev_hash: &mut [u8; 32], ) -> i64 { - let mut ev = AuditEvent { - id: next_id, - timestamp: draft.timestamp, - operation: draft.operation, - user: draft.user, - object_uid: draft.object_uid, - algorithm: draft.algorithm, - client_ip: draft.client_ip, - result: draft.result, - duration_ms: draft.duration_ms, - request_id: draft.request_id, - details: draft.details, - prev_hash: *prev_hash, - row_hash: [0_u8; 32], - }; - ev.row_hash = compute_row_hash(&ev); + let event = draft.finalize(next_id, *prev_hash); - match sink.write_event(&ev) { + match sink.write_event_atomic(&event).await { Ok(()) => { - *prev_hash = ev.row_hash; + *prev_hash = event.row_hash; next_id.checked_add(1).unwrap_or_else(|| { error!( "AuditFileStore: id counter overflow at i64::MAX — \ @@ -130,18 +96,15 @@ pub(super) fn write_draft_to_chain( Err(e) => { error!( "AuditFileStore: failed to write event id={}: {e} — event dropped", - ev.id + event.id ); - // Do NOT advance id or prev_hash — the next event will reuse - // the same slot, preserving chain continuity. + // Reuse this chain position after a failed write. next_id } } } -/// Builds a sentinel `AuditEventDraft` that records how many real events were -/// dropped due to channel saturation. Joins the hash chain like any real event -/// — detectable by `ckms audit verify` and compliance tooling. +/// Builds a chained sentinel recording events dropped by channel saturation. fn make_eviction_sentinel(n_dropped: u64) -> AuditEventDraft { AuditEventDraft { timestamp: OffsetDateTime::now_utc(), diff --git a/crate/server/src/middlewares/audit/mod.rs b/crate/server/src/middlewares/audit/mod.rs index 0e34a7b852..b2a0470126 100644 --- a/crate/server/src/middlewares/audit/mod.rs +++ b/crate/server/src/middlewares/audit/mod.rs @@ -24,10 +24,6 @@ mod client_ip; mod extensions; -pub(crate) use extensions::{ - BatchItemAuditContext, KmipAlgorithm, KmipBatchOperations, KmipObjectUid, KmipOperationName, -}; - use std::{ pin::Pin, rc::Rc, @@ -45,6 +41,9 @@ use cosmian_kms_access::audit::{ AuditEventDraft, AuditResult, OperationAuditContext, RequestAuditContext, audit_now, }; use cosmian_logger::error; +pub(crate) use extensions::{ + BatchItemAuditContext, KmipAlgorithm, KmipBatchOperations, KmipObjectUid, KmipOperationName, +}; use futures::{ Future, future::{Ready, ok}, diff --git a/documentation/docs/configuration/log-reference.md b/documentation/docs/configuration/log-reference.md index a6eb9e323e..eec12c7781 100644 --- a/documentation/docs/configuration/log-reference.md +++ b/documentation/docs/configuration/log-reference.md @@ -743,14 +743,16 @@ Crate path: `crate/server` | `error` | `AuditFileStore: torn write recovered — discarded {bytes_discarded} byte(s) at offset {discard_offset} (process likely killed mid-write); resuming chain at id={next_id}` | `src/core/audit/recovery.rs` | `bytes_discarded`: incomplete bytes dropped
`discard_offset`: offset in file
`next_id`: resuming event ID | Process killed mid-write (crash/SIGKILL). Incomplete event discarded; hash chain preserved. | | `debug` | `AuditFileStore: still waiting on audit log lock {} ({e})` | `src/core/audit/file_sink.rs` | `e`: lock acquisition error | Debug: subsequent retry attempt (not the first). Implies a preceding "audit log lock held by another instance" error. | | `error` | `AuditFileStore: audit log {} reached its configured max_size_bytes cap ({len} bytes >= {cap}) — audit writing is blocked until the log is safely remediated and the KMS is restarted` | `src/core/audit/file_sink.rs` | `len`: actual file length in bytes
`cap`: configured `max_size_bytes` | First transition into the capped state only (logged once). The event that crossed the cap is still persisted; every event after it is dropped (subject to `--audit-failure-mode`) until the log is remediated and the KMS restarted. Write-stop cap, not rotation or retention. | -| `error` | `AuditFileStore: cannot stat audit log {} ({e})` | `src/core/audit/file_sink.rs` | `e`: I/O error from `File::metadata()` | Only possible if `max_size_bytes` is configured; the cap check for this write is skipped (not treated as capped) and retried on the next write. | -| `debug` | `AuditFileStore: audit log {} is at its max_size_bytes cap — event dropped` | `src/core/audit/writer.rs` | - | Throttled to at most once every 500ms while blocked events keep arriving after the cap has been reached. | | `error` | `AuditFileStore: recovery task failed to run ({join_err}) — retrying` | `src/core/audit/file_sink.rs` | `join_err`: task join error (tokio thread panic or cancellation) | Audit recovery background task crashed; will retry after backoff interval. Monitor frequency to detect systemic issues. | | `error` | `AuditFileStore: cannot acquire audit log lock {} ({e}) — retrying` | `src/core/audit/file_sink.rs` | `e`: lock acquisition error other than contention (EACCES, EROFS, directory-creation failure) | Deployment fault distinct from the benign "held by another instance" case; retried on the same interval. | | `trace` | `Extractable: {:?}` | `src/core/operations/attributes/add.rs` | - | - | | `trace` | `Set Attribute: Extractable: {:?}` | `src/core/operations/attributes/set.rs` | - | - | | `warn` | `[kms-init] Failed to seed kms.keys.active.count: {e}` | `src/core/kms/mod.rs` | `e` | - | | `warn` | `[metrics-cron] Failed to sync kms.keys.active.count: {}` | `src/cron.rs` | - | - | +| `error` | `AuditFileStore: audit sink failed to resume ({e}) — audit logging is disabled for this process` | `src/core/audit/store.rs` | `e` | Only reachable if the recovered log file's metadata cannot be `stat`'d at all (lock acquisition and recovery/open faults retry indefinitely and never reach here). The writer task exits and audit logging stays off for the rest of the process lifetime — it is not retried. | +| `error` | `AuditFileStore: cannot stat recovered log file {} ({e}) — retrying` | `src/core/audit/file_sink.rs` | `e`: I/O error from `File::metadata()` | Retried in place inside `FileSink::resume()`'s recovery loop; a transient stat failure self-heals and does not disable audit logging. | +| `error` | `AuditFileStore: failed to write recovery sentinel id={}: {e} — event dropped` | `src/core/audit/file_sink.rs` | `id`: sentinel event ID
`e`: I/O error | A torn-write-recovered or reanchor sentinel written during recovery (before the writer task exists) failed to persist; the chain does not advance past this `id` and the sentinel is dropped. | +| `debug` | `AuditFileStore: sink '{}' is at capacity — event dropped` | `src/core/audit/writer.rs` | sink name (`AuditSink::name()`) | Throttled to at most once every 500ms while blocked events keep arriving after the sink reports `is_write_capacity_exceeded() == true`. | ### `cosmian_kms_server_database`