pub use kcode_k1_audio_classification_projection::{
FragmentId, FragmentStatus, OverallState, SpeakerLabelV1,
};
pub use kcode_k1_txn_ordering::TxId;
#[cfg(any(test, feature = "testkit"))]
pub use kcode_k1_audio_classification_driver::{EngineFuture, FragmentEngine};
use kcode_k1_audio_classification_driver::{AudioClassificationDriver, StartOutcome};
use kcode_k1_audio_classification_projection::{Projection, ProjectionEffect};
use kcode_k1_audio_fragment_transactions as transactions;
use kcode_k1_objects::K1Objects;
use kcode_k1_peering::K1Peering;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem, SubsystemId};
use kcode_speaker_v3_analysis::Analyzer;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
#[cfg(any(test, feature = "testkit"))]
use std::{collections::HashMap, path::PathBuf};
const RESTART_ERROR: &str = "analysis interrupted by restart";
const SUBSYSTEM_NAME: &str = "audio-classification";
pub struct AudioClassificationCoordinator {
projection: Arc<Projection>,
_ordering: Arc<K1TxnOrdering>,
driver: Arc<AudioClassificationDriver>,
health: Arc<Health>,
#[cfg(test)]
callback: Arc<Callback>,
}
impl AudioClassificationCoordinator {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
analyzer: Analyzer,
) -> Result<Self, String> {
let driver = Arc::new(AudioClassificationDriver::open(
peering.clone(),
objects,
analyzer,
));
Self::open_with_driver(root, ordering, peering, driver)
}
#[cfg(any(test, feature = "testkit"))]
pub fn with_engine(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
objects: Arc<K1Objects>,
engine: Arc<dyn FragmentEngine>,
) -> Result<Self, String> {
let driver = Arc::new(AudioClassificationDriver::with_engine(
peering.clone(),
objects,
engine,
));
Self::open_with_driver(root, ordering, peering, driver)
}
fn open_with_driver(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
driver: Arc<AudioClassificationDriver>,
) -> Result<Self, String> {
let (projection, cursor) = Projection::open(root, &ordering)?;
let projection = Arc::new(projection);
let health = Arc::new(Health::default());
let replaying = Arc::new(AtomicBool::new(true));
let replayed = Arc::new(AtomicU64::new(0));
let callback = Arc::new(Callback {
projection: projection.clone(),
driver: driver.clone(),
health: health.clone(),
replaying: replaying.clone(),
replayed: replayed.clone(),
});
let startup = (|| {
ordering
.register_subsystem(subsystem_id()?, cursor, callback.clone())
.map_err(|error| format!("register audio classification: {error}"))?;
replaying.store(false, Ordering::Release);
record_replay(root, replayed.load(Ordering::Acquire));
for fragment in projection.running()? {
transactions::submit_failure(
&peering,
fragment.fragment_id,
fragment.stage,
None,
RESTART_ERROR.to_owned(),
)?;
}
for fragment_id in projection.queued()? {
start_driver(&driver, fragment_id)?;
}
driver.ensure_healthy()?;
health.ensure()
})();
if let Err(error) = startup {
driver.shutdown();
return Err(error);
}
Ok(Self {
projection,
_ordering: ordering,
driver,
health,
#[cfg(test)]
callback,
})
}
pub fn status(&self, fragment_id: FragmentId) -> Result<Option<FragmentStatus>, String> {
self.ensure_healthy()?;
self.projection.status(fragment_id)
}
pub fn validate_labels(
&self,
fragment_id: FragmentId,
labels: &[SpeakerLabelV1],
) -> Result<TxId, String> {
self.ensure_healthy()?;
self.projection.validate_labels(fragment_id, labels)
}
pub fn ensure_healthy(&self) -> Result<(), String> {
self.health.ensure()?;
if let Err(error) = self.driver.ensure_healthy() {
self.fault(format!("audio classification driver: {error}"));
}
self.health.ensure()
}
pub fn fault_ambiguous_commitment(&self) {
self.fault("transaction commitment is ambiguous".to_owned());
}
pub fn shutdown(&self) {
self.driver.shutdown();
}
#[cfg(any(test, feature = "testkit"))]
pub fn inject_errors(
&self,
fragment_id: FragmentId,
errors: Vec<String>,
) -> Result<(), String> {
self.projection.inject_errors(fragment_id, errors)
}
#[cfg(any(test, feature = "testkit"))]
pub fn startup_replay_count(root: &Path) -> Result<u64, String> {
lock(replay_counts())
.get(root)
.copied()
.ok_or_else(|| "startup replay count is unavailable".to_owned())
}
fn fault(&self, diagnostic: String) {
self.health.fault(diagnostic);
self.driver.shutdown();
}
}
impl Drop for AudioClassificationCoordinator {
fn drop(&mut self) {
self.driver.shutdown();
}
}
#[derive(Default)]
struct Health(Mutex<Option<String>>);
impl Health {
fn ensure(&self) -> Result<(), String> {
lock(&self.0).as_ref().map_or_else(
|| Ok(()),
|error| Err(format!("audio classification requires reopen: {error}")),
)
}
fn fault(&self, diagnostic: String) {
let mut current = lock(&self.0);
if current.is_none() {
*current = Some(diagnostic);
}
}
}
struct Callback {
projection: Arc<Projection>,
driver: Arc<AudioClassificationDriver>,
health: Arc<Health>,
replaying: Arc<AtomicBool>,
replayed: Arc<AtomicU64>,
}
impl Callback {
fn react(&self, fragment_id: FragmentId, effect: ProjectionEffect) -> Result<(), String> {
match effect {
ProjectionEffect::Start => start_driver(&self.driver, fragment_id),
ProjectionEffect::Abort | ProjectionEffect::LabelsCommitted => {
self.driver.abort(fragment_id);
Ok(())
}
ProjectionEffect::None => {
let state = self
.projection
.status(fragment_id)?
.ok_or_else(|| "applied event has no projected fragment".to_owned())?
.state;
if is_terminal(state) {
self.driver.abort(fragment_id);
}
Ok(())
}
}
}
fn fault(&self, diagnostic: String) -> String {
self.health.fault(diagnostic.clone());
self.driver.shutdown();
diagnostic
}
}
impl Subsystem for Callback {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
let applied = self
.projection
.apply(id, payload)
.map_err(|error| self.fault(format!("apply audio classification event: {error}")))?;
if self.replaying.load(Ordering::Acquire) {
self.replayed.fetch_add(1, Ordering::AcqRel);
return Ok(());
}
self.react(applied.fragment_id, applied.effect)
.map_err(|error| self.fault(format!("apply audio classification effect: {error}")))
}
fn reorg(&self) -> Result<(), String> {
let result = self.projection.clear();
self.health
.fault("canonical reorganization requires reopen".to_owned());
self.driver.shutdown();
result
}
}
fn start_driver(driver: &AudioClassificationDriver, id: FragmentId) -> Result<(), String> {
match driver.start(id)? {
StartOutcome::Started | StartOutcome::Pending => Ok(()),
StartOutcome::AlreadyActive => Err("audio fragment is already active".to_owned()),
}
}
fn subsystem_id() -> Result<SubsystemId, String> {
SubsystemId::from_str(SUBSYSTEM_NAME)
}
fn is_terminal(state: OverallState) -> bool {
matches!(
state,
OverallState::Failed
| OverallState::Completed
| OverallState::Confirmed
| OverallState::Discarded
)
}
fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
mutex.lock().unwrap_or_else(|error| error.into_inner())
}
#[cfg(any(test, feature = "testkit"))]
fn record_replay(root: &Path, count: u64) {
lock(replay_counts()).insert(root.to_path_buf(), count);
}
#[cfg(not(any(test, feature = "testkit")))]
fn record_replay(_root: &Path, _count: u64) {}
#[cfg(any(test, feature = "testkit"))]
fn replay_counts() -> &'static Mutex<HashMap<PathBuf, u64>> {
use std::sync::OnceLock;
static COUNTS: OnceLock<Mutex<HashMap<PathBuf, u64>>> = OnceLock::new();
COUNTS.get_or_init(|| Mutex::new(HashMap::new()))
}
#[cfg(test)]
mod tests;