use super::TAP_SUBJECT_PREFIX;
use crate::services::lys_http::LysHttp;
use std::collections::{BTreeMap, BTreeSet};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, mpsc};
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum PlacementResult {
Placed,
Existing,
Skipped,
}
#[derive(serde::Serialize, serde::Deserialize)]
struct Journal {
v: u8,
network: String,
subjects: BTreeMap<String, bool>,
}
pub struct PlacementBook {
path: PathBuf,
journal: Journal,
attempted: BTreeSet<String>,
}
impl PlacementBook {
pub(crate) fn relocate_legacy(
legacy: &Path,
destination: &Path,
network: &str,
) -> Result<(), String> {
if !legacy.try_exists().map_err(|error| {
format!(
"tap placement legacy journal cannot be inspected: {}",
error.kind()
)
})? {
return Ok(());
}
let previous = Self::open(legacy, network)?;
let mut current = Self::open(destination, network)?;
for (subject, placed) in previous.journal.subjects {
current
.journal
.subjects
.entry(subject)
.and_modify(|existing| *existing |= placed)
.or_insert(placed);
}
current.save()?;
std::fs::remove_file(legacy).map_err(|error| {
format!(
"tap placement legacy journal cannot be removed: {}",
error.kind()
)
})?;
tracing::info!(from = %legacy.display(), to = %destination.display(), "TAP_SUBJECT_JOURNAL_MIGRATED");
Ok(())
}
pub(crate) fn open(path: &Path, network: &str) -> Result<Self, String> {
if network.trim().is_empty() {
return Err("tap placement names no network".to_owned());
}
let journal = match std::fs::read(path) {
Ok(bytes) => {
let journal: Journal = serde_json::from_slice(&bytes)
.map_err(|error| format!("tap placement journal cannot be read: {error}"))?;
if journal.v != 1 || journal.network != network {
return Err(
"tap placement journal has an unknown version or another network"
.to_owned(),
);
}
for subject in journal.subjects.keys() {
validate(subject)?;
}
journal
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Journal {
v: 1,
network: network.to_owned(),
subjects: BTreeMap::new(),
},
Err(error) => {
return Err(format!(
"tap placement journal cannot be read: {}",
error.kind()
));
}
};
Ok(Self {
path: path.to_owned(),
journal,
attempted: BTreeSet::new(),
})
}
pub(crate) fn pending(&self) -> Vec<String> {
self.journal
.subjects
.iter()
.filter(|(_, placed)| !**placed)
.map(|(subject, _)| subject.clone())
.collect()
}
pub(crate) fn attempt(
&mut self,
subject: &str,
http: &dyn LysHttp,
credential: &str,
) -> Result<PlacementResult, String> {
validate(subject)?;
if self.journal.subjects.get(subject) == Some(&true)
|| !self.attempted.insert(subject.to_owned())
{
return Ok(PlacementResult::Skipped);
}
self.journal.subjects.insert(subject.to_owned(), false);
self.save()?;
let child = format!("{TAP_SUBJECT_PREFIX}{subject}");
let mut hash = blake3::Hasher::new();
hash.update(b"liminal-console-tap-placement\0");
hash.update(self.journal.network.as_bytes());
hash.update(b"\0");
hash.update(child.as_bytes());
let operation = format!("op-{}", hex::encode(&hash.finalize().as_bytes()[..16]));
let body = serde_json::json!({
"operation": operation,
"child": {"kind": "liminal.subject", "id": child},
"parent": {"kind": "liminal.network", "id": self.journal.network},
});
let answer = http
.post(credential, "apps/liminal/placements", &body)
.map_err(|error| error.to_string())?;
let result = if answer.succeeded() {
PlacementResult::Placed
} else if answer.status == 400
&& answer.body["refusal"] == "placement_invalid"
&& answer.body["reason"]
== "placement_invalid: the resource is already placed in a parent"
{
PlacementResult::Existing
} else {
return Err(format!(
"{}: {}",
answer.refusal(),
answer.body["reason"]
.as_str()
.unwrap_or("no reason supplied")
));
};
self.journal.subjects.insert(subject.to_owned(), true);
self.save()?;
if result == PlacementResult::Placed {
tracing::info!(
subject = child,
network = self.journal.network,
"TAP_SUBJECT_PLACED"
);
}
Ok(result)
}
fn save(&self) -> Result<(), String> {
let parent = self
.path
.parent()
.ok_or("tap placement journal names no directory")?;
std::fs::create_dir_all(parent)
.map_err(|error| format!("tap placement journal directory failed: {}", error.kind()))?;
let next = self.path.with_extension("writing");
let bytes = serde_json::to_vec(&self.journal)
.map_err(|error| format!("tap placement journal encoding failed: {error}"))?;
let mut file = std::fs::File::create(&next)
.map_err(|error| format!("tap placement journal create failed: {}", error.kind()))?;
file.write_all(&bytes)
.and_then(|()| file.sync_all())
.map_err(|error| format!("tap placement journal flush failed: {}", error.kind()))?;
std::fs::rename(&next, &self.path)
.map_err(|error| format!("tap placement journal replace failed: {}", error.kind()))
}
}
fn validate(subject: &str) -> Result<(), String> {
if subject.is_empty()
|| subject.starts_with("console.")
|| subject.chars().any(char::is_control)
{
Err("tap placement target is empty, reserved or contains a control character".to_owned())
} else {
Ok(())
}
}
pub struct TapPlacements {
send: Option<mpsc::Sender<String>>,
queued: Mutex<BTreeSet<String>>,
worker: Option<std::thread::JoinHandle<()>>,
}
impl std::fmt::Debug for TapPlacements {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("TapPlacements")
.finish_non_exhaustive()
}
}
impl TapPlacements {
pub(crate) fn start(
path: &Path,
network: &str,
http: Arc<dyn LysHttp>,
credential: String,
) -> Result<Arc<Self>, String> {
let mut book = PlacementBook::open(path, network)?;
let pending = book.pending();
let (send, receive) = mpsc::channel::<String>();
let worker = std::thread::Builder::new().name("liminal-tap-placements".to_owned()).spawn(move || {
for subject in receive {
if let Err(error) = book.attempt(&subject, http.as_ref(), &credential) {
tracing::warn!(%subject, %error, "TAP_SUBJECT_PLACEMENT_FAILED: retry at next boot");
}
}
}).map_err(|error| format!("tap placement worker could not start: {}", error.kind()))?;
let placements = Arc::new(Self {
send: Some(send),
queued: Mutex::new(BTreeSet::new()),
worker: Some(worker),
});
for subject in pending {
placements.offer(&subject)?;
}
Ok(placements)
}
pub(crate) fn offer(&self, subject: &str) -> Result<(), String> {
validate(subject)?;
if !self
.queued
.lock()
.map_err(|_| "tap placement queue poisoned")?
.insert(subject.to_owned())
{
return Ok(());
}
self.send
.as_ref()
.ok_or("tap placement worker stopped")?
.send(subject.to_owned())
.map_err(|_| "tap placement worker exited".to_owned())
}
}
impl Drop for TapPlacements {
fn drop(&mut self) {
drop(self.send.take());
if let Some(worker) = self.worker.take()
&& worker.join().is_err()
{
tracing::error!("TAP_SUBJECT_PLACEMENT_WORKER_FAILED");
}
}
}