hugo/RuDS
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: more work on background tasks

Browse Source
This commit is contained in:
Chili Palmer
2026-07-24 11:43:11 +02:00
parent c187108f89
commit da13bac755
45 changed files with 2905 additions and 732 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
crates/bds-core/src/engine/calendar.rs
View File

@@ -6,7 +6,7 @@ use std::path::Path;
use crate::db::DbConnection as Connection;
use crate::db::queries::post as post_q;
use crate::engine::EngineResult;
use crate::engine::{EngineError, EngineResult};
use crate::model::PostStatus;
use crate::util::{atomic_write, timestamp};
@@ -18,6 +18,15 @@ pub fn regenerate_calendar(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<()> {
regenerate_calendar_with_progress(conn, data_dir, project_id, |_, _, _| true)
}
pub fn regenerate_calendar_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
mut on_item: impl FnMut(usize, usize, &str) -> bool,
) -> EngineResult<()> {
let posts = post_q::list_posts_by_project(conn, project_id)?;
@@ -25,7 +34,11 @@ pub fn regenerate_calendar(
let mut months: BTreeMap<String, u32> = BTreeMap::new();
let mut days: BTreeMap<String, u32> = BTreeMap::new();
for post in &posts {
let total = posts.len() + 1;
for (index, post) in posts.iter().enumerate() {
if !on_item(index + 1, total, &post.title) {
return Err(EngineError::Cancelled);
}
if post.status != PostStatus::Published {
continue;
}
@@ -41,6 +54,10 @@ pub fn regenerate_calendar(
*days.entry(day_key).or_insert(0) += 1;
}
if !on_item(total, total, "calendar.json") {
return Err(EngineError::Cancelled);
}
let calendar = serde_json::json!({
"years": years,
"months": months,
@@ -82,4 +99,20 @@ mod tests {
serde_json::from_str(&std::fs::read_to_string(&cal_path).unwrap()).unwrap();
assert!(data["years"].as_object().unwrap().is_empty());
}
#[test]
fn cancellation_before_write_leaves_no_calendar() {
let db = Database::open_in_memory().unwrap();
db.migrate().unwrap();
let dir = tempfile::tempdir().unwrap();
let project =
engine::project::create_project(db.conn(), "Test", Some(dir.path().to_str().unwrap()))
.unwrap();
let result =
regenerate_calendar_with_progress(db.conn(), dir.path(), &project.id, |_, _, _| false);
assert!(matches!(result, Err(EngineError::Cancelled)));
assert!(!dir.path().join("html/calendar.json").exists());
}
}

2
crates/bds-core/src/engine/error.rs
View File

@@ -15,6 +15,8 @@ pub enum EngineError {
Conflict(String),
#[error("validation error: {0}")]
Validation(String),
#[error("operation cancelled")]
Cancelled,
}
impl From<crate::db::DatabaseError> for EngineError {

37
crates/bds-core/src/engine/generation.rs
View File

@@ -63,6 +63,15 @@ pub struct PreparedSiteGeneration {
generated_hashes: Arc<HashMap<String, String>>,
}
impl std::fmt::Debug for PreparedSiteGeneration {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("PreparedSiteGeneration")
.field("source_count", &self.sources.len())
.finish_non_exhaustive()
}
}
pub fn prepare_site_generation(
conn: &Connection,
data_dir: &Path,
@@ -91,6 +100,34 @@ pub fn prepare_site_generation(
})
}
pub fn prepared_section_page_count(
prepared: &PreparedSiteGeneration,
validation: Option<&SiteValidationReport>,
section: GenerationSection,
) -> usize {
let requested = validation.map(|validation| {
validation
.missing_pages
.iter()
.chain(validation.stale_pages.iter())
.cloned()
.collect::<HashSet<_>>()
});
let fallback = validation.is_some_and(|validation| {
validation
.missing_pages
.iter()
.chain(validation.extra_pages.iter())
.chain(validation.stale_pages.iter())
.any(|path| classify_generated_path(path, &prepared.metadata).is_none())
});
crate::render::count_site_render_pages_from_context(
&prepared.render,
section,
(!fallback).then_some(requested.as_ref()).flatten(),
)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum GenerationSection {
Core,

45
crates/bds-core/src/engine/media.rs
View File

@@ -53,9 +53,18 @@ pub fn rebuild_media_links(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<MediaLinkRebuildReport> {
rebuild_media_links_with_progress(conn, data_dir, project_id, None)
}
pub fn rebuild_media_links_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
on_item: Option<ItemProgressFn>,
) -> EngineResult<MediaLinkRebuildReport> {
conn.begin_savepoint()?;
match rebuild_media_links_inner(conn, data_dir, project_id) {
match rebuild_media_links_inner(conn, data_dir, project_id, on_item.as_ref()) {
Ok(report) => {
conn.release_savepoint()?;
Ok(report)
@@ -71,9 +80,16 @@ fn rebuild_media_links_inner(
conn: &Connection,
data_dir: &Path,
project_id: &str,
on_item: Option<&ItemProgressFn>,
) -> EngineResult<MediaLinkRebuildReport> {
let mut report = MediaLinkRebuildReport::default();
for item in qm::list_media_by_project(conn, project_id)? {
let items = qm::list_media_by_project(conn, project_id)?;
for (index, item) in items.iter().enumerate() {
if let Some(callback) = on_item
&& !callback(index + 1, items.len(), &item.original_name)
{
return Err(EngineError::Cancelled);
}
let sidecar = read_sidecar(&fs::read_to_string(data_dir.join(&item.sidecar_path))?)
.map_err(EngineError::Parse)?;
for link in qpm::list_post_media_by_media(conn, &item.id)? {
@@ -110,9 +126,22 @@ pub fn regenerate_missing_thumbnails(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<ThumbnailRepairReport> {
regenerate_missing_thumbnails_with_progress(conn, data_dir, project_id, |_, _, _| true)
}
pub fn regenerate_missing_thumbnails_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
mut on_item: impl FnMut(usize, usize, &str) -> bool,
) -> EngineResult<ThumbnailRepairReport> {
let mut report = ThumbnailRepairReport::default();
for item in qm::list_media_by_project(conn, project_id)? {
let items = qm::list_media_by_project(conn, project_id)?;
for (index, item) in items.iter().enumerate() {
if !on_item(index + 1, items.len(), &item.original_name) {
return Err(EngineError::Cancelled);
}
if !item.mime_type.starts_with("image/") || item.mime_type.contains("svg") {
continue;
}
@@ -653,7 +682,7 @@ pub fn rebuild_media_from_filesystem(
}
/// Per-item progress callback: (current_item, total_items, item_description).
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) + Send>;
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Like `rebuild_media_from_filesystem` but with optional per-item progress.
pub fn rebuild_media_from_filesystem_with_progress(
@@ -699,7 +728,9 @@ pub fn rebuild_media_from_filesystem_with_progress(
for (i, path) in canonical_sidecars.iter().enumerate() {
if let Some(ref cb) = on_item {
let name = path.file_stem().and_then(|s| s.to_str()).unwrap_or("?");
cb(i + 1, total, name);
if !cb(i + 1, total, name) {
return Err(EngineError::Cancelled);
}
}
match rebuild_canonical_media(conn, data_dir, project_id, path) {
Ok(created) => {
@@ -720,7 +751,9 @@ pub fn rebuild_media_from_filesystem_with_progress(
for (i, path) in translation_sidecars.iter().enumerate() {
if let Some(ref cb) = on_item {
let name = path.file_stem().and_then(|s| s.to_str()).unwrap_or("?");
cb(offset + i + 1, total, name);
if !cb(offset + i + 1, total, name) {
return Err(EngineError::Cancelled);
}
}
match rebuild_translation_sidecar(conn, data_dir, project_id, path) {
Ok(created) => {

30
crates/bds-core/src/engine/metadata_diff.rs
View File

@@ -70,8 +70,27 @@ pub fn compute_metadata_diff(
data_dir: &Path,
project_id: &str,
) -> EngineResult<DiffReport> {
compute_metadata_diff_with_progress(conn, data_dir, project_id, |_, _| true)
}
pub fn compute_metadata_diff_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
mut on_progress: impl FnMut(usize, usize) -> bool,
) -> EngineResult<DiffReport> {
const PHASES: usize = 8;
let mut phase = 0;
let mut next_phase = || -> EngineResult<()> {
if !on_progress(phase, PHASES) {
return Err(crate::engine::EngineError::Cancelled);
}
phase += 1;
Ok(())
};
let mut report = DiffReport::default();
next_phase()?;
if let Ok(project) = qproject::get_project_by_id(conn, project_id) {
match diff_project(conn, data_dir, &project) {
Ok(Some(diff)) => report.diffs.push(diff),
@@ -94,6 +113,7 @@ pub fn compute_metadata_diff(
}
// 1. Diff posts
next_phase()?;
let posts = qp::list_posts_by_project(conn, project_id)?;
for post in &posts {
if post.file_path.is_empty() {
@@ -107,6 +127,7 @@ pub fn compute_metadata_diff(
}
// 2. Diff translations
next_phase()?;
for post in &posts {
let translations = qt::list_post_translations_by_post(conn, &post.id)?;
for t in &translations {
@@ -122,6 +143,7 @@ pub fn compute_metadata_diff(
}
// 3. Diff media
next_phase()?;
let media_items = qm::list_media_by_project(conn, project_id)?;
for m in &media_items {
if m.sidecar_path.is_empty() {
@@ -146,6 +168,7 @@ pub fn compute_metadata_diff(
}
// 4. Diff templates
next_phase()?;
let templates = qtpl::list_templates_by_project(conn, project_id)?;
for t in &templates {
if t.file_path.is_empty() {
@@ -159,6 +182,7 @@ pub fn compute_metadata_diff(
}
// 5. Diff scripts
next_phase()?;
let scripts = qs::list_scripts_by_project(conn, project_id)?;
for s in &scripts {
if s.file_path.is_empty() {
@@ -172,6 +196,7 @@ pub fn compute_metadata_diff(
}
// 6. Detect orphans
next_phase()?;
if crate::engine::meta::read_project_json(data_dir)
.is_ok_and(|metadata| metadata.semantic_similarity_enabled)
{
@@ -215,9 +240,14 @@ pub fn compute_metadata_diff(
}
// 7. Detect orphans
next_phase()?;
let orphans = detect_orphan_files(conn, data_dir, project_id)?;
report.orphans = orphans;
if !on_progress(PHASES, PHASES) {
return Err(crate::engine::EngineError::Cancelled);
}
Ok(report)
}

26
crates/bds-core/src/engine/post.rs
View File

@@ -716,7 +716,7 @@ pub fn rebuild_posts_from_filesystem(
}
/// Per-item progress callback: (current_item, total_items, item_description).
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) + Send>;
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Like `rebuild_posts_from_filesystem` but with optional per-item progress.
pub fn rebuild_posts_from_filesystem_with_progress(
@@ -760,7 +760,9 @@ pub fn rebuild_posts_from_filesystem_with_progress(
for (i, path) in canonical_files.iter().enumerate() {
if let Some(ref cb) = on_item {
let name = path.file_stem().and_then(|s| s.to_str()).unwrap_or("?");
cb(i + 1, total, name);
if !cb(i + 1, total, name) {
return Err(EngineError::Cancelled);
}
}
match rebuild_canonical_post(conn, data_dir, project_id, path) {
Ok(created) => {
@@ -781,7 +783,9 @@ pub fn rebuild_posts_from_filesystem_with_progress(
for (i, path) in translation_files.iter().enumerate() {
if let Some(ref cb) = on_item {
let name = path.file_stem().and_then(|s| s.to_str()).unwrap_or("?");
cb(offset + i + 1, total, name);
if !cb(offset + i + 1, total, name) {
return Err(EngineError::Cancelled);
}
}
match rebuild_translation(conn, data_dir, project_id, path) {
Ok(created) => {
@@ -816,11 +820,25 @@ pub fn rebuild_all_links(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<usize> {
rebuild_all_links_with_progress(conn, data_dir, project_id, None)
}
pub fn rebuild_all_links_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
on_item: Option<ItemProgressFn>,
) -> EngineResult<usize> {
let posts = qp::list_posts_by_project(conn, project_id)?;
let mut link_count = 0;
for post in &posts {
for (index, post) in posts.iter().enumerate() {
if let Some(ref callback) = on_item
&& !callback(index + 1, posts.len(), &post.title)
{
return Err(EngineError::Cancelled);
}
// Get post content: from DB or filesystem
let content = if let Some(ref content) = post.content {
content.clone()

115
crates/bds-core/src/engine/publishing.rs
View File

@@ -1,7 +1,10 @@
use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use serde::{Deserialize, Serialize};
use walkdir::WalkDir;
@@ -124,6 +127,26 @@ pub fn upload_site(
private_cache_dir: &Path,
preferences: &PublishingPreferences,
mut on_progress: impl FnMut(usize, usize, UploadTargetKind),
) -> EngineResult<PublishJob> {
let cancelled = AtomicBool::new(false);
upload_site_cancellable(
data_dir,
private_cache_dir,
preferences,
&cancelled,
move |current, total, kind| {
on_progress(current, total, kind);
true
},
)
}
pub fn upload_site_cancellable(
data_dir: &Path,
private_cache_dir: &Path,
preferences: &PublishingPreferences,
cancel_flag: &AtomicBool,
mut on_progress: impl FnMut(usize, usize, UploadTargetKind) -> bool,
) -> EngineResult<PublishJob> {
if std::env::var_os("SSH_AUTH_SOCK").is_none() {
return Err(EngineError::Validation(
@@ -134,7 +157,7 @@ pub fn upload_site(
data_dir,
private_cache_dir,
preferences,
&mut |program, args| run_command(program, args),
&mut |program, args| run_command_cancellable(program, args, cancel_flag),
&mut on_progress,
)
}
@@ -144,7 +167,7 @@ fn upload_site_with_runner(
private_cache_dir: &Path,
preferences: &PublishingPreferences,
runner: &mut CommandRunner<'_>,
on_progress: &mut dyn FnMut(usize, usize, UploadTargetKind),
on_progress: &mut dyn FnMut(usize, usize, UploadTargetKind) -> bool,
) -> EngineResult<PublishJob> {
let credentials = Credentials::from_preferences(preferences)?;
let targets = build_upload_targets(data_dir, &credentials);
@@ -162,7 +185,9 @@ fn upload_site_with_runner(
let cache_path = private_cache_dir.join("publishing-scp-mtimes.json");
let mut cache = read_cache(&cache_path);
for (index, target) in targets.iter().enumerate() {
on_progress(index + 1, targets.len(), target.kind);
if !on_progress(index + 1, targets.len(), target.kind) {
return Err(EngineError::Cancelled);
}
let result = match credentials.mode {
SshMode::Rsync => upload_rsync(target, &credentials, runner),
SshMode::Scp => upload_scp(target, &credentials, &mut cache, runner),
@@ -328,21 +353,49 @@ fn write_cache(path: &Path, cache: &ScpMtimeCache) -> EngineResult<()> {
Ok(())
}
fn run_command(program: &str, args: &[String]) -> Result<(), String> {
let output = Command::new(program)
fn run_command_cancellable(
program: &str,
args: &[String],
cancel_flag: &AtomicBool,
) -> Result<(), String> {
let mut child = Command::new(program)
.args(args)
.output()
.stdout(Stdio::null())
.stderr(Stdio::piped())
.spawn()
.map_err(|error| format!("failed to start {program}: {error}"))?;
if output.status.success() {
Ok(())
} else {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
Err(if stderr.is_empty() {
format!("{program} exited with {}", output.status)
} else {
stderr
})
let stderr = child.stderr.take();
let stderr_reader = std::thread::spawn(move || {
let mut output = String::new();
if let Some(mut stderr) = stderr {
let _ = stderr.read_to_string(&mut output);
}
output
});
let status = loop {
if cancel_flag.load(Ordering::Acquire) {
let _ = child.kill();
let _ = child.wait();
let _ = stderr_reader.join();
return Err("operation cancelled".into());
}
if let Some(status) = child
.try_wait()
.map_err(|error| format!("failed to wait for {program}: {error}"))?
{
break status;
}
std::thread::sleep(Duration::from_millis(50));
};
let stderr = stderr_reader.join().unwrap_or_default().trim().to_string();
if status.success() {
return Ok(());
}
Err(if stderr.is_empty() {
format!("{program} exited with {status}")
} else {
stderr
})
}
#[cfg(test)]
@@ -376,7 +429,7 @@ mod tests {
commands.push((program.to_owned(), args.to_vec()));
Ok(())
},
&mut |_, _, _| {},
&mut |_, _, _| true,
)
.unwrap();
@@ -397,6 +450,28 @@ mod tests {
);
}
#[test]
fn cancellation_stops_before_starting_the_next_target() {
let dir = TempDir::new().unwrap();
let cache = TempDir::new().unwrap();
fs::create_dir_all(dir.path().join("html")).unwrap();
let mut commands = 0;
let result = upload_site_with_runner(
dir.path(),
cache.path(),
&preferences(SshMode::Rsync),
&mut |_, _| {
commands += 1;
Ok(())
},
&mut |_, _, _| false,
);
assert!(matches!(result, Err(EngineError::Cancelled)));
assert_eq!(commands, 0);
}
#[test]
fn scp_excludes_sidecars_and_skips_unchanged_files() {
let dir = TempDir::new().unwrap();
@@ -415,7 +490,7 @@ mod tests {
first.push((program.to_owned(), args.to_vec()));
Ok(())
},
&mut |_, _, _| {},
&mut |_, _, _| true,
)
.unwrap();
assert_eq!(
@@ -437,7 +512,7 @@ mod tests {
second.push(program.to_owned());
Ok(())
},
&mut |_, _, _| {},
&mut |_, _, _| true,
)
.unwrap();
assert!(second.is_empty());
@@ -454,7 +529,7 @@ mod tests {
cache.path(),
&preferences(SshMode::Rsync),
&mut |_, _| Err("network down".into()),
&mut |_, _, _| {},
&mut |_, _, _| true,
)
.unwrap_err();
assert!(error.to_string().contains("network down"));

188
crates/bds-core/src/engine/rebuild.rs
View File

@@ -27,6 +27,8 @@ pub struct FullRebuildReport {
pub templates_updated: usize,
pub scripts_created: usize,
pub scripts_updated: usize,
pub thumbnails_generated: usize,
pub thumbnail_media_failed: usize,
pub errors: Vec<String>,
}
@@ -91,8 +93,19 @@ pub enum RebuildProgress {
name: String,
},
RebuildingTemplates,
TemplateItem {
current: usize,
total: usize,
name: String,
},
RebuildingScripts,
ScriptItem {
current: usize,
total: usize,
name: String,
},
ImportingTags,
RebuildingThumbnails,
RefreshingSemanticIndex,
Complete,
}
@@ -116,8 +129,19 @@ impl RebuildProgress {
name,
} => localize_item(locale, "engine.progress.mediaItem", *current, *total, name),
Self::RebuildingTemplates => translate(locale, "engine.progress.rebuildingTemplates"),
Self::TemplateItem {
current,
total,
name,
} => localize_item(locale, "engine.checkingItem", *current, *total, name),
Self::RebuildingScripts => translate(locale, "engine.progress.rebuildingScripts"),
Self::ScriptItem {
current,
total,
name,
} => localize_item(locale, "engine.checkingItem", *current, *total, name),
Self::ImportingTags => translate(locale, "engine.progress.importingTags"),
Self::RebuildingThumbnails => translate(locale, "engine.progress.rebuildingThumbnails"),
Self::RefreshingSemanticIndex => {
translate(locale, "engine.progress.refreshingSemanticIndex")
}
@@ -137,7 +161,7 @@ fn localize_item(locale: UiLocale, key: &str, current: usize, total: usize, name
}
/// Progress callback: (percent 0.0..1.0, semantic progress event).
pub type ProgressFn = Arc<dyn Fn(f32, &RebuildProgress) + Send + Sync>;
pub type ProgressFn = Arc<dyn Fn(f32, &RebuildProgress) -> bool + Send + Sync>;
/// Orchestrate a full rebuild from filesystem into the database.
///
@@ -178,23 +202,26 @@ fn rebuild_from_filesystem_inner(
on_progress: Option<ProgressFn>,
) -> EngineResult<FullRebuildReport> {
let mut report = FullRebuildReport::default();
let progress = |pct: f32, event: RebuildProgress| {
if let Some(ref f) = on_progress {
f(pct, &event);
let progress = |pct: f32, event: RebuildProgress| -> EngineResult<()> {
if let Some(ref f) = on_progress
&& !f(pct, &event)
{
return Err(crate::engine::EngineError::Cancelled);
}
Ok(())
};
// Phase weights: posts 0.0..0.35, media 0.35..0.70, templates 0.70..0.85, scripts 0.85..1.0
// 1. Load portable project metadata and clear all reconstructible rows.
progress(0.0, RebuildProgress::LoadingProjectMetadata);
progress(0.0, RebuildProgress::LoadingProjectMetadata)?;
fts::ensure_fts_tables(conn)?;
crate::engine::meta::startup_sync(data_dir)?;
crate::engine::meta::sync_metadata_from_filesystem(conn, data_dir, project_id)?;
clear_project_rows(conn, project_id)?;
// 2. Rebuild posts (0.00 .. 0.35)
progress(0.01, RebuildProgress::ScanningPosts);
progress(0.01, RebuildProgress::ScanningPosts)?;
let post_item_cb: Option<post::ItemProgressFn> = on_progress.as_ref().map(|cb| {
let cb = Arc::clone(cb);
let f: post::ItemProgressFn = Box::new(move |current, total, name| {
@@ -211,7 +238,7 @@ fn rebuild_from_filesystem_inner(
total,
name: name.to_string(),
},
);
)
});
f
});
@@ -228,7 +255,7 @@ fn rebuild_from_filesystem_inner(
report.errors.extend(post_report.errors);
// 3. Rebuild media (0.35 .. 0.70)
progress(0.35, RebuildProgress::ScanningMedia);
progress(0.35, RebuildProgress::ScanningMedia)?;
let media_item_cb: Option<media::ItemProgressFn> = on_progress.as_ref().map(|cb| {
let cb = Arc::clone(cb);
let f: media::ItemProgressFn = Box::new(move |current, total, name| {
@@ -245,7 +272,7 @@ fn rebuild_from_filesystem_inner(
total,
name: name.to_string(),
},
);
)
});
f
});
@@ -262,26 +289,87 @@ fn rebuild_from_filesystem_inner(
report.errors.extend(media_report.errors);
// 4. Rebuild templates (0.70 .. 0.85)
progress(0.70, RebuildProgress::RebuildingTemplates);
let tpl_report =
template_rebuild::rebuild_templates_from_filesystem(conn, data_dir, project_id)?;
progress(0.70, RebuildProgress::RebuildingTemplates)?;
let template_progress = on_progress.as_ref().map(|callback| {
let callback = Arc::clone(callback);
Box::new(move |current: usize, total: usize, name: &str| {
callback(
0.70 + current as f32 / total.max(1) as f32 * 0.15,
&RebuildProgress::TemplateItem {
current,
total,
name: name.to_string(),
},
)
}) as template_rebuild::ItemProgressFn
});
let tpl_report = template_rebuild::rebuild_templates_from_filesystem_with_progress(
conn,
data_dir,
project_id,
template_progress,
)?;
report.templates_created = tpl_report.created;
report.templates_updated = tpl_report.updated;
report.errors.extend(tpl_report.errors);
// 5. Rebuild scripts (0.85 .. 0.95)
progress(0.85, RebuildProgress::RebuildingScripts);
let script_report =
script_rebuild::rebuild_scripts_from_filesystem(conn, data_dir, project_id)?;
progress(0.85, RebuildProgress::RebuildingScripts)?;
let script_progress = on_progress.as_ref().map(|callback| {
let callback = Arc::clone(callback);
Box::new(move |current: usize, total: usize, name: &str| {
callback(
0.85 + current as f32 / total.max(1) as f32 * 0.10,
&RebuildProgress::ScriptItem {
current,
total,
name: name.to_string(),
},
)
}) as script_rebuild::ItemProgressFn
});
let script_report = script_rebuild::rebuild_scripts_from_filesystem_with_progress(
conn,
data_dir,
project_id,
script_progress,
)?;
report.scripts_created = script_report.created;
report.scripts_updated = script_report.updated;
report.errors.extend(script_report.errors);
// 6. Restore relationships and tags (0.95 .. 1.0)
progress(0.95, RebuildProgress::ImportingTags);
progress(0.95, RebuildProgress::ImportingTags)?;
super::tag::import_tags_from_file(conn, data_dir, project_id)?;
super::tag::sync_tags_from_posts(conn, project_id)?;
post::rebuild_all_links(conn, data_dir, project_id)?;
let link_progress = on_progress.as_ref().map(|callback| {
let callback = Arc::clone(callback);
Box::new(move |current: usize, total: usize, name: &str| {
callback(
0.95 + current as f32 / total.max(1) as f32 * 0.01,
&RebuildProgress::PostItem {
current,
total,
name: name.to_string(),
},
)
}) as post::ItemProgressFn
});
post::rebuild_all_links_with_progress(conn, data_dir, project_id, link_progress)?;
let media_link_progress = on_progress.as_ref().map(|callback| {
let callback = Arc::clone(callback);
Box::new(move |current: usize, total: usize, name: &str| {
callback(
0.96 + current as f32 / total.max(1) as f32 * 0.01,
&RebuildProgress::MediaItem {
current,
total,
name: name.to_string(),
},
)
}) as media::ItemProgressFn
});
media::rebuild_media_links_with_progress(conn, data_dir, project_id, media_link_progress)?;
if !report.errors.is_empty() {
return Err(crate::engine::EngineError::Validation(format!(
@@ -290,11 +378,39 @@ fn rebuild_from_filesystem_inner(
)));
}
progress(0.98, RebuildProgress::RefreshingSemanticIndex);
crate::engine::embedding::EmbeddingService::production(conn, data_dir)
.index_unindexed(project_id)?;
progress(0.97, RebuildProgress::RebuildingThumbnails)?;
let thumbnail_report = media::regenerate_missing_thumbnails_with_progress(
conn,
data_dir,
project_id,
|current, total, name| {
on_progress.as_ref().is_none_or(|callback| {
callback(
0.97 + current as f32 / total.max(1) as f32 * 0.02,
&RebuildProgress::MediaItem {
current,
total,
name: name.to_string(),
},
)
})
},
)?;
report.thumbnails_generated = thumbnail_report.thumbnails_generated;
report.thumbnail_media_failed = thumbnail_report.media_failed;
progress(1.0, RebuildProgress::Complete);
progress(0.99, RebuildProgress::RefreshingSemanticIndex)?;
crate::engine::embedding::EmbeddingService::production(conn, data_dir)
.index_unindexed_with_progress(project_id, |current, total| {
on_progress.as_ref().is_none_or(|callback| {
callback(
0.99 + current as f32 / total.max(1) as f32 * 0.01,
&RebuildProgress::RefreshingSemanticIndex,
)
})
})?;
progress(1.0, RebuildProgress::Complete)?;
Ok(report)
}
@@ -445,6 +561,36 @@ mod tests {
assert!(report.errors.is_empty());
}
#[test]
fn cancellation_rolls_back_the_full_rebuild_savepoint() {
let (db, dir) = setup();
let post = crate::engine::post::create_post(
db.conn(),
dir.path(),
"p1",
"Preserved",
Some("body"),
vec![],
vec![],
None,
Some("en"),
None,
)
.unwrap();
let result = rebuild_from_filesystem_with_progress(
db.conn(),
dir.path(),
"p1",
Some(Arc::new(|_, event| {
!matches!(event, RebuildProgress::ScanningPosts)
})),
);
assert!(matches!(result, Err(crate::engine::EngineError::Cancelled)));
assert!(qp::get_post_by_id(db.conn(), &post.id).is_ok());
}
#[test]
fn rebuild_creates_posts_and_media() {
let (db, dir) = setup();

33
crates/bds-core/src/engine/script_rebuild.rs
View File

@@ -17,6 +17,8 @@ pub struct ScriptRebuildReport {
pub errors: Vec<String>,
}
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Rebuild scripts from the filesystem into the database.
///
/// Walks the `scripts/` directory for `*.lua` files, parses each
@@ -26,6 +28,15 @@ pub fn rebuild_scripts_from_filesystem(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<ScriptRebuildReport> {
rebuild_scripts_from_filesystem_with_progress(conn, data_dir, project_id, None)
}
pub fn rebuild_scripts_from_filesystem_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
on_item: Option<ItemProgressFn>,
) -> EngineResult<ScriptRebuildReport> {
let mut report = ScriptRebuildReport::default();
let scripts_dir = data_dir.join("scripts");
@@ -34,17 +45,23 @@ pub fn rebuild_scripts_from_filesystem(
return Ok(report);
}
for entry in WalkDir::new(&scripts_dir)
let files = WalkDir::new(&scripts_dir)
.into_iter()
.filter_map(|e| e.ok())
{
.filter(|entry| entry.path().is_file())
.filter(|entry| entry.path().extension().and_then(|ext| ext.to_str()) == Some("lua"))
.collect::<Vec<_>>();
for (index, entry) in files.iter().enumerate() {
let path = entry.path();
if !path.is_file() {
continue;
}
let ext = path.extension().and_then(|e| e.to_str());
if ext != Some("lua") {
continue;
let name = path
.file_stem()
.and_then(|stem| stem.to_str())
.unwrap_or("?");
if let Some(ref callback) = on_item
&& !callback(index + 1, files.len(), name)
{
return Err(EngineError::Cancelled);
}
match rebuild_single_script(conn, data_dir, project_id, path) {

37
crates/bds-core/src/engine/search.rs
View File

@@ -67,7 +67,7 @@ pub struct ReindexReport {
}
/// Per-item progress callback: (current_item, total_items, item_description).
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) + Send>;
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Repair a missing or previously deployed FTS schema and report whether its
/// derived content still needs to be rebuilt.
@@ -215,8 +215,10 @@ fn index_project(
for post in post_q::list_posts_by_project(conn, project_id)? {
*current += 1;
if let Some(callback) = on_item {
callback(*current, total, &post.title);
if let Some(callback) = on_item
&& !callback(*current, total, &post.title)
{
return Err(EngineError::Cancelled);
}
let translations = post_translation::list_post_translations_by_post(conn, &post.id)?;
let translation_data = translations
@@ -255,8 +257,10 @@ fn index_project(
for media in media_q::list_media_by_project(conn, project_id)? {
*current += 1;
if let Some(callback) = on_item {
callback(*current, total, &media.original_name);
if let Some(callback) = on_item
&& !callback(*current, total, &media.original_name)
{
return Err(EngineError::Cancelled);
}
let translations = media_translation::list_media_translations_by_media(conn, &media.id)?;
let translation_data = translations
@@ -374,6 +378,29 @@ mod tests {
assert_eq!(results.len(), 1);
}
#[test]
fn cancellation_rolls_back_shared_index_rebuild() {
let (db, project_id) = setup();
let dir = tempfile::tempdir().unwrap();
engine::post::create_post(
db.conn(),
dir.path(),
&project_id,
"Cancel Me",
Some("body"),
vec![],
vec![],
None,
Some("en"),
None,
)
.unwrap();
let result = rebuild_search_index(db.conn(), Some(Box::new(|_, _, _| false)));
assert!(matches!(result, Err(EngineError::Cancelled)));
}
#[test]
fn rebuild_indexes_published_post_and_translation_bodies_from_files() {
let db = Database::open_in_memory().unwrap();

676
crates/bds-core/src/engine/task.rs
View File

@@ -1,7 +1,10 @@
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::Notify;
/// Unique task identifier.
pub type TaskId = u64;
@@ -25,6 +28,7 @@ pub struct TaskSnapshot {
pub status: TaskStatus,
pub progress: Option<f32>,
pub message: Option<String>,
pub cancellation_requested: bool,
pub created_at: Instant,
}
@@ -42,228 +46,340 @@ struct TaskEntry {
created_at: Instant,
finished_at: Option<Instant>,
last_progress_report: Option<Instant>,
worker_active: bool,
worker_started: bool,
}
/// Manages concurrent tasks with a max concurrency limit and FIFO queue.
pub struct TaskManager {
max_concurrent: usize,
next_id: Mutex<TaskId>,
tasks: Mutex<Vec<TaskEntry>>,
state: Mutex<TaskState>,
state_changed: Condvar,
async_changed: Notify,
}
struct TaskState {
next_id: TaskId,
tasks: HashMap<TaskId, TaskEntry>,
order: VecDeque<TaskId>,
pending: VecDeque<TaskId>,
worker_count: usize,
}
/// Capacity reservation held from asynchronous admission until the worker exits.
pub struct TaskWorker {
manager: Arc<TaskManager>,
task_id: TaskId,
}
impl Drop for TaskWorker {
fn drop(&mut self) {
self.manager.worker_exited(self.task_id);
}
}
impl TaskManager {
/// Create a new task manager with the given concurrency limit.
pub fn new(max_concurrent: usize) -> Self {
Self {
max_concurrent,
next_id: Mutex::new(1),
tasks: Mutex::new(Vec::new()),
max_concurrent: max_concurrent.max(1),
state: Mutex::new(TaskState {
next_id: 1,
tasks: HashMap::new(),
order: VecDeque::new(),
pending: VecDeque::new(),
worker_count: 0,
}),
state_changed: Condvar::new(),
async_changed: Notify::new(),
}
}
/// Submit a new task. Returns its unique identifier.
pub fn submit(&self, label: &str) -> TaskId {
let mut next = self.next_id.lock().unwrap();
let id = *next;
*next += 1;
self.submit_with_group(label, None, None)
}
let entry = TaskEntry {
fn submit_with_group(
&self,
label: &str,
group_id: Option<&str>,
group_name: Option<&str>,
) -> TaskId {
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
let id = state.next_id;
state.next_id += 1;
state.tasks.insert(
id,
label: label.to_owned(),
group_id: None,
group_name: None,
status: TaskStatus::Pending,
cancel_flag: Arc::new(AtomicBool::new(false)),
progress: None,
message: None,
created_at: Instant::now(),
finished_at: None,
last_progress_report: None,
};
let mut tasks = self.tasks.lock().unwrap();
tasks.push(entry);
// Auto-start if under capacity
let running = tasks
.iter()
.filter(|t| t.status == TaskStatus::Running)
.count();
if running < self.max_concurrent
&& let Some(t) = tasks
.iter_mut()
.find(|t| t.id == id && t.status == TaskStatus::Pending)
{
t.status = TaskStatus::Running;
}
TaskEntry {
id,
label: label.to_owned(),
group_id: group_id.map(str::to_owned),
group_name: group_name.map(str::to_owned),
status: TaskStatus::Pending,
cancel_flag: Arc::new(AtomicBool::new(false)),
progress: None,
message: None,
created_at: Instant::now(),
finished_at: None,
last_progress_report: None,
worker_active: false,
worker_started: false,
},
);
state.order.push_back(id);
state.pending.push_back(id);
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
id
}
/// Submit a new task within a group. Returns its unique identifier.
pub fn submit_grouped(&self, label: &str, group_id: &str, group_name: &str) -> TaskId {
let id = self.submit(label);
let mut tasks = self.tasks.lock().unwrap();
if let Some(entry) = tasks.iter_mut().find(|t| t.id == id) {
entry.group_id = Some(group_id.to_owned());
entry.group_name = Some(group_name.to_owned());
}
id
self.submit_with_group(label, Some(group_id), Some(group_name))
}
/// Block a worker until its task may run. Returns false if cancelled.
/// Wait synchronously for admission. Prefer [`Self::admit`] before spawning workers.
pub fn wait_until_runnable(&self, task_id: TaskId) -> bool {
let mut tasks = self.tasks.lock().unwrap();
let mut state = self.state.lock().unwrap();
loop {
match tasks
.iter()
.find(|task| task.id == task_id)
.map(|task| &task.status)
{
Some(TaskStatus::Running) => return true,
Some(TaskStatus::Pending) => tasks = self.state_changed.wait(tasks).unwrap(),
match state.tasks.get(&task_id).map(|task| &task.status) {
Some(TaskStatus::Running) => {
if let Some(task) = state.tasks.get_mut(&task_id) {
task.worker_started = true;
}
return true;
}
Some(TaskStatus::Pending) => state = self.state_changed.wait(state).unwrap(),
_ => return false,
}
}
}
/// Admit without occupying a blocking-pool thread while queued.
pub async fn admit(self: &Arc<Self>, task_id: TaskId) -> Option<TaskWorker> {
loop {
let notified = self.async_changed.notified();
{
let mut state = self.state.lock().unwrap();
match state.tasks.get_mut(&task_id) {
Some(task) if task.status == TaskStatus::Running => {
task.worker_started = true;
return Some(TaskWorker {
manager: Arc::clone(self),
task_id,
});
}
Some(task) if task.status == TaskStatus::Pending => {}
_ => return None,
}
}
notified.await;
}
}
/// Admit a non-Tokio worker while retaining capacity until its guard drops.
pub fn admit_blocking(self: &Arc<Self>, task_id: TaskId) -> Option<TaskWorker> {
if self.wait_until_runnable(task_id) {
Some(TaskWorker {
manager: Arc::clone(self),
task_id,
})
} else {
None
}
}
/// Mark a task as completed.
pub fn complete(&self, task_id: TaskId) {
let mut tasks = self.tasks.lock().unwrap();
if let Some(entry) = tasks.iter_mut().find(|t| t.id == task_id)
&& matches!(entry.status, TaskStatus::Running)
{
entry.status = TaskStatus::Completed;
entry.progress = Some(1.0);
entry.finished_at = Some(Instant::now());
}
Self::promote_next(&mut tasks, self.max_concurrent);
self.state_changed.notify_all();
let mut state = self.state.lock().unwrap();
let released = if let Some(entry) = state.tasks.get_mut(&task_id) {
if matches!(entry.status, TaskStatus::Running) {
entry.status = TaskStatus::Completed;
entry.progress = Some(1.0);
entry.finished_at = Some(Instant::now());
}
let released = entry.worker_active;
entry.worker_active = false;
entry.worker_started = false;
released
} else {
false
};
state.worker_count = state.worker_count.saturating_sub(usize::from(released));
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
}
/// Mark a task as failed with an error message.
pub fn fail(&self, task_id: TaskId, error: String) {
let mut tasks = self.tasks.lock().unwrap();
if let Some(entry) = tasks.iter_mut().find(|t| t.id == task_id)
&& matches!(entry.status, TaskStatus::Running)
{
entry.message = Some(error.clone());
entry.status = TaskStatus::Failed(error);
entry.finished_at = Some(Instant::now());
}
Self::promote_next(&mut tasks, self.max_concurrent);
self.state_changed.notify_all();
let mut state = self.state.lock().unwrap();
let released = if let Some(entry) = state.tasks.get_mut(&task_id) {
if matches!(entry.status, TaskStatus::Running) {
if entry.cancel_flag.load(Ordering::Acquire) {
entry.status = TaskStatus::Cancelled;
} else {
entry.message = Some(error.clone());
entry.status = TaskStatus::Failed(error);
}
entry.finished_at = Some(Instant::now());
}
let released = entry.worker_active;
entry.worker_active = false;
entry.worker_started = false;
released
} else {
false
};
state.worker_count = state.worker_count.saturating_sub(usize::from(released));
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
}
/// Cancel a task by setting its cancel flag and status.
pub fn cancel(&self, task_id: TaskId) {
let mut tasks = self.tasks.lock().unwrap();
if let Some(entry) = tasks.iter_mut().find(|t| t.id == task_id)
/// Cancel queued work immediately, or request a cooperative stop from a worker.
pub fn cancel(&self, task_id: TaskId) -> bool {
let mut state = self.state.lock().unwrap();
let mut cancelled = false;
let mut released = false;
if let Some(entry) = state.tasks.get_mut(&task_id)
&& matches!(entry.status, TaskStatus::Running | TaskStatus::Pending)
{
entry.cancel_flag.store(true, Ordering::Release);
entry.status = TaskStatus::Cancelled;
entry.finished_at = Some(Instant::now());
if !entry.worker_started {
if entry.worker_active {
entry.worker_active = false;
released = true;
}
entry.status = TaskStatus::Cancelled;
entry.finished_at = Some(Instant::now());
}
cancelled = true;
}
Self::promote_next(&mut tasks, self.max_concurrent);
self.state_changed.notify_all();
state.worker_count = state.worker_count.saturating_sub(usize::from(released));
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
cancelled
}
/// Cancel every active task in a group and release their workers.
pub fn cancel_group(&self, group_id: &str) {
let mut tasks = self.tasks.lock().unwrap();
let mut state = self.state.lock().unwrap();
let now = Instant::now();
for entry in tasks.iter_mut().filter(|task| {
task.group_id.as_deref() == Some(group_id)
&& matches!(task.status, TaskStatus::Running | TaskStatus::Pending)
}) {
let group_ids = state
.tasks
.values()
.filter(|task| {
task.group_id.as_deref() == Some(group_id)
&& matches!(task.status, TaskStatus::Running | TaskStatus::Pending)
})
.map(|task| task.id)
.collect::<Vec<_>>();
let mut released = 0;
for id in group_ids {
let entry = state.tasks.get_mut(&id).unwrap();
entry.cancel_flag.store(true, Ordering::Release);
entry.status = TaskStatus::Cancelled;
entry.finished_at = Some(now);
if !entry.worker_started {
if entry.worker_active {
entry.worker_active = false;
released += 1;
}
entry.status = TaskStatus::Cancelled;
entry.finished_at = Some(now);
}
}
Self::promote_next(&mut tasks, self.max_concurrent);
self.state_changed.notify_all();
state.worker_count = state.worker_count.saturating_sub(released);
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
}
/// Return the group containing a task, if any.
pub fn group_id(&self, task_id: TaskId) -> Option<String> {
self.tasks
.lock()
.unwrap()
.iter()
.find(|task| task.id == task_id)
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.tasks
.get(&task_id)
.and_then(|task| task.group_id.clone())
}
/// Check whether a task has been cancelled.
pub fn is_cancelled(&self, task_id: TaskId) -> bool {
let tasks = self.tasks.lock().unwrap();
tasks
.iter()
.find(|t| t.id == task_id)
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.tasks
.get(&task_id)
.map(|t| t.cancel_flag.load(Ordering::Acquire))
.unwrap_or(false)
}
/// Shared cancellation flag for a worker owned by this task.
pub fn cancellation_flag(&self, task_id: TaskId) -> Option<Arc<AtomicBool>> {
self.tasks
.lock()
.unwrap()
.iter()
.find(|task| task.id == task_id)
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.tasks
.get(&task_id)
.map(|task| Arc::clone(&task.cancel_flag))
}
/// Return the current status of a task.
pub fn status(&self, task_id: TaskId) -> Option<TaskStatus> {
let tasks = self.tasks.lock().unwrap();
tasks
.iter()
.find(|t| t.id == task_id)
.map(|t| t.status.clone())
self.get(task_id).map(|task| task.status)
}
/// Count tasks that are still queued.
pub fn pending_count(&self) -> usize {
let tasks = self.tasks.lock().unwrap();
tasks
.iter()
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.tasks
.values()
.filter(|t| t.status == TaskStatus::Pending)
.count()
}
/// Count tasks that are currently running.
pub fn running_count(&self) -> usize {
let tasks = self.tasks.lock().unwrap();
tasks
.iter()
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.tasks
.values()
.filter(|t| t.status == TaskStatus::Running)
.count()
}
/// Remove finished tasks older than the configured retention period.
pub fn evict_expired(&self) {
let cutoff = Instant::now() - FINISHED_TASK_TTL;
let mut tasks = self.tasks.lock().unwrap();
tasks.retain(|task| {
task.finished_at
.is_none_or(|finished_at| finished_at > cutoff)
});
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
}
/// Remove every finished task while preserving running and queued work.
pub fn clear_completed(&self) {
self.tasks
.lock()
.unwrap()
.retain(|task| matches!(task.status, TaskStatus::Pending | TaskStatus::Running));
let mut state = self.state.lock().unwrap();
Self::remove_where(&mut state, |task| task.status == TaskStatus::Completed);
}
/// Remove all terminal task results while preserving active work.
pub fn clear_finished(&self) {
let mut state = self.state.lock().unwrap();
Self::remove_where(&mut state, |task| {
task.finished_at.is_some() && !task.worker_active
});
}
/// Update progress for a running task. Throttled to at most once per 250ms.
pub fn report_progress(&self, task_id: TaskId, progress: Option<f32>, message: Option<String>) {
let mut tasks = self.tasks.lock().unwrap();
if let Some(entry) = tasks.iter_mut().find(|t| t.id == task_id)
let mut state = self.state.lock().unwrap();
if let Some(entry) = state.tasks.get_mut(&task_id)
&& entry.status == TaskStatus::Running
{
let now = Instant::now();
@@ -271,7 +387,7 @@ impl TaskManager {
Some(prev) => now.duration_since(prev).as_millis() >= PROGRESS_THROTTLE_MS as u128,
None => true,
};
if should_report {
if should_report || progress.is_some_and(|value| value >= 1.0) {
entry.progress = progress;
entry.message = message;
entry.last_progress_report = Some(now);
@@ -281,58 +397,157 @@ impl TaskManager {
/// Return the current progress of a task.
pub fn progress(&self, task_id: TaskId) -> Option<f32> {
let tasks = self.tasks.lock().unwrap();
tasks
.iter()
.find(|t| t.id == task_id)
.and_then(|t| t.progress)
self.get(task_id).and_then(|task| task.progress)
}
/// Return a snapshot of all tasks for UI display.
/// Return active tasks plus ten recent finished tasks for status surfaces.
pub fn snapshots(&self) -> Vec<TaskSnapshot> {
let tasks = self.tasks.lock().unwrap();
let mut snapshots = tasks
.iter()
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
let mut active = state
.tasks
.values()
.filter(|task| task.finished_at.is_none())
.chain(
tasks
.iter()
.rev()
.filter(|task| task.finished_at.is_some())
.take(RECENT_FINISHED_LIMIT),
)
.map(|task| TaskSnapshot {
id: task.id,
label: task.label.clone(),
group_id: task.group_id.clone(),
group_name: task.group_name.clone(),
status: task.status.clone(),
progress: task.progress,
message: task.message.clone(),
created_at: task.created_at,
})
.collect::<Vec<_>>();
snapshots.sort_by_key(|snapshot| snapshot.created_at);
snapshots
active.sort_by_key(|task| (task.status != TaskStatus::Running, task.created_at));
let active_groups = active
.iter()
.filter_map(|task| task.group_id.as_deref())
.collect::<HashSet<_>>();
let mut finished = state
.tasks
.values()
.filter(|task| task.finished_at.is_some())
.collect::<Vec<_>>();
finished.sort_by_key(|task| std::cmp::Reverse(task.finished_at));
let recent_ids = finished
.iter()
.take(RECENT_FINISHED_LIMIT)
.map(|task| task.id)
.collect::<HashSet<_>>();
active
.into_iter()
.chain(finished.into_iter().filter(|task| {
recent_ids.contains(&task.id)
|| task
.group_id
.as_deref()
.is_some_and(|group| active_groups.contains(group))
}))
.map(Self::snapshot)
.collect()
}
/// Promote the next queued task to running if capacity allows.
fn promote_next(tasks: &mut [TaskEntry], max_concurrent: usize) {
while tasks
/// Return one retained task by id.
pub fn get(&self, task_id: TaskId) -> Option<TaskSnapshot> {
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state.tasks.get(&task_id).map(Self::snapshot)
}
/// Return every retained task, newest first.
pub fn all(&self) -> Vec<TaskSnapshot> {
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.order
.iter()
.rev()
.filter_map(|id| state.tasks.get(id).map(Self::snapshot))
.collect()
}
/// Return running tasks in start order.
pub fn running(&self) -> Vec<TaskSnapshot> {
let mut state = self.state.lock().unwrap();
Self::prune_expired(&mut state);
state
.order
.iter()
.filter_map(|id| state.tasks.get(id))
.filter(|task| task.status == TaskStatus::Running)
.count()
< max_concurrent
{
let Some(task) = tasks
.iter_mut()
.find(|task| task.status == TaskStatus::Pending)
else {
.map(Self::snapshot)
.collect()
}
fn worker_exited(&self, task_id: TaskId) {
let mut state = self.state.lock().unwrap();
let released = if let Some(task) = state.tasks.get_mut(&task_id) {
if task.cancel_flag.load(Ordering::Acquire) {
task.worker_started = false;
drop(state);
self.notify_changed();
return;
}
let released = task.worker_active;
task.worker_active = false;
task.worker_started = false;
released
} else {
false
};
state.worker_count = state.worker_count.saturating_sub(usize::from(released));
Self::promote_next(&mut state, self.max_concurrent);
drop(state);
self.notify_changed();
}
fn promote_next(state: &mut TaskState, max_concurrent: usize) {
while state.worker_count < max_concurrent {
let Some(id) = state.pending.pop_front() else {
break;
};
let Some(task) = state.tasks.get_mut(&id) else {
continue;
};
if task.status != TaskStatus::Pending {
continue;
}
task.status = TaskStatus::Running;
task.worker_active = true;
state.worker_count += 1;
}
}
fn snapshot(task: &TaskEntry) -> TaskSnapshot {
TaskSnapshot {
id: task.id,
label: task.label.clone(),
group_id: task.group_id.clone(),
group_name: task.group_name.clone(),
status: task.status.clone(),
progress: task.progress,
message: task.message.clone(),
cancellation_requested: task.cancel_flag.load(Ordering::Acquire),
created_at: task.created_at,
}
}
fn prune_expired(state: &mut TaskState) {
let cutoff = Instant::now() - FINISHED_TASK_TTL;
Self::remove_where(state, |task| {
task.finished_at
.is_some_and(|finished_at| finished_at <= cutoff)
&& !task.worker_active
});
}
fn remove_where(state: &mut TaskState, predicate: impl Fn(&TaskEntry) -> bool) {
let removed = state
.tasks
.values()
.filter(|task| predicate(task))
.map(|task| task.id)
.collect::<HashSet<_>>();
state.tasks.retain(|id, _| !removed.contains(id));
state.order.retain(|id| !removed.contains(id));
state.pending.retain(|id| !removed.contains(id));
}
fn notify_changed(&self) {
self.state_changed.notify_all();
self.async_changed.notify_waiters();
}
}
impl Default for TaskManager {
@@ -399,7 +614,7 @@ mod tests {
}
#[test]
fn cancel_sets_flag() {
fn cancelling_unstarted_task_settles_immediately() {
let mgr = TaskManager::default();
let id = mgr.submit("upload");
// Task is auto-started (Running)
@@ -409,6 +624,35 @@ mod tests {
assert_eq!(mgr.status(id), Some(TaskStatus::Cancelled));
}
#[tokio::test]
async fn cancelling_started_task_stays_running_until_worker_stops() {
let mgr = Arc::new(TaskManager::new(1));
let id = mgr.submit("upload");
let worker = mgr.admit(id).await.unwrap();
mgr.cancel(id);
let snapshot = mgr.get(id).unwrap();
assert_eq!(snapshot.status, TaskStatus::Running);
assert!(snapshot.cancellation_requested);
drop(worker);
mgr.fail(id, "operation cancelled".into());
assert_eq!(mgr.status(id), Some(TaskStatus::Cancelled));
}
#[tokio::test]
async fn completed_work_wins_a_late_cancellation_request() {
let mgr = Arc::new(TaskManager::new(1));
let id = mgr.submit("atomic update");
let worker = mgr.admit(id).await.unwrap();
mgr.cancel(id);
drop(worker);
mgr.complete(id);
assert_eq!(mgr.status(id), Some(TaskStatus::Completed));
}
#[test]
fn complete_and_fail() {
let mgr = TaskManager::default();
@@ -444,8 +688,12 @@ mod tests {
// After b fails: d promoted to running
{
let mut tasks = mgr.tasks.lock().unwrap();
for task in tasks.iter_mut().filter(|task| task.finished_at.is_some()) {
let mut state = mgr.state.lock().unwrap();
for task in state
.tasks
.values_mut()
.filter(|task| task.finished_at.is_some())
{
task.finished_at =
Some(Instant::now() - FINISHED_TASK_TTL - Duration::from_secs(1));
}
@@ -469,6 +717,34 @@ mod tests {
assert_eq!(mgr.snapshots().len(), 10);
}
#[test]
fn snapshots_retain_every_active_task_and_complete_active_groups() {
let mgr = TaskManager::new(20);
let finished_group_member = mgr.submit_grouped("finished", "group", "Group");
mgr.complete(finished_group_member);
for index in 0..12 {
mgr.submit_grouped(&format!("active {index}"), "group", "Group");
}
for index in 0..12 {
let id = mgr.submit(&format!("history {index}"));
mgr.complete(id);
}
let snapshots = mgr.snapshots();
assert_eq!(
snapshots
.iter()
.filter(|task| matches!(task.status, TaskStatus::Pending | TaskStatus::Running))
.count(),
12
);
assert!(
snapshots
.iter()
.any(|task| task.id == finished_group_member)
);
}
#[test]
fn clear_completed_preserves_active_tasks() {
let mgr = TaskManager::new(2);
@@ -555,4 +831,70 @@ mod tests {
assert_eq!(mgr.progress(id), Some(0.5));
assert_eq!(mgr.snapshots()[0].message.as_deref(), Some("halfway"));
}
#[tokio::test]
async fn cancellation_holds_capacity_until_worker_exits() {
let mgr = Arc::new(TaskManager::new(1));
let running = mgr.submit("running");
let queued = mgr.submit("queued");
let worker = mgr.admit(running).await.unwrap();
mgr.cancel(running);
assert_eq!(mgr.status(running), Some(TaskStatus::Running));
assert!(mgr.get(running).unwrap().cancellation_requested);
assert_eq!(mgr.status(queued), Some(TaskStatus::Pending));
drop(worker);
assert_eq!(mgr.status(queued), Some(TaskStatus::Pending));
mgr.fail(running, "operation cancelled".into());
assert_eq!(mgr.status(queued), Some(TaskStatus::Running));
}
#[tokio::test]
async fn panicking_worker_guard_releases_capacity() {
let mgr = Arc::new(TaskManager::new(1));
let panicking = mgr.submit("panicking");
let queued = mgr.submit("queued");
let worker = mgr.admit(panicking).await.unwrap();
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
let _worker = worker;
panic!("boom");
}));
assert_eq!(mgr.status(queued), Some(TaskStatus::Running));
}
#[test]
fn terminal_progress_bypasses_throttle() {
let mgr = TaskManager::new(1);
let id = mgr.submit("work");
mgr.report_progress(id, Some(0.5), Some("working".into()));
mgr.report_progress(id, Some(1.0), Some("done".into()));
let task = mgr
.snapshots()
.into_iter()
.find(|task| task.id == id)
.unwrap();
assert_eq!(task.progress, Some(1.0));
assert_eq!(task.message.as_deref(), Some("done"));
}
#[test]
fn clear_completed_keeps_other_terminal_results() {
let mgr = TaskManager::new(3);
let completed = mgr.submit("completed");
let failed = mgr.submit("failed");
let cancelled = mgr.submit("cancelled");
mgr.complete(completed);
mgr.fail(failed, "failed".into());
mgr.cancel(cancelled);
mgr.clear_completed();
assert_eq!(mgr.status(completed), None);
assert!(matches!(mgr.status(failed), Some(TaskStatus::Failed(_))));
assert_eq!(mgr.status(cancelled), Some(TaskStatus::Cancelled));
}
}

33
crates/bds-core/src/engine/template_rebuild.rs
View File

@@ -17,6 +17,8 @@ pub struct TemplateRebuildReport {
pub errors: Vec<String>,
}
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Rebuild templates from the filesystem into the database.
///
/// Walks the `templates/` directory for `*.liquid` files, parses each via
@@ -26,6 +28,15 @@ pub fn rebuild_templates_from_filesystem(
conn: &Connection,
data_dir: &Path,
project_id: &str,
) -> EngineResult<TemplateRebuildReport> {
rebuild_templates_from_filesystem_with_progress(conn, data_dir, project_id, None)
}
pub fn rebuild_templates_from_filesystem_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
on_item: Option<ItemProgressFn>,
) -> EngineResult<TemplateRebuildReport> {
let mut report = TemplateRebuildReport::default();
let templates_dir = data_dir.join("templates");
@@ -34,17 +45,23 @@ pub fn rebuild_templates_from_filesystem(
return Ok(report);
}
for entry in WalkDir::new(&templates_dir)
let files = WalkDir::new(&templates_dir)
.into_iter()
.filter_map(|e| e.ok())
{
.filter(|entry| entry.path().is_file())
.filter(|entry| entry.path().extension().and_then(|ext| ext.to_str()) == Some("liquid"))
.collect::<Vec<_>>();
for (index, entry) in files.iter().enumerate() {
let path = entry.path();
if !path.is_file() {
continue;
}
let ext = path.extension().and_then(|e| e.to_str());
if ext != Some("liquid") {
continue;
let name = path
.file_stem()
.and_then(|stem| stem.to_str())
.unwrap_or("?");
if let Some(ref callback) = on_item
&& !callback(index + 1, files.len(), name)
{
return Err(EngineError::Cancelled);
}
match rebuild_single_template(conn, data_dir, project_id, path) {

28
crates/bds-core/src/engine/validate_site.rs
View File

@@ -6,8 +6,8 @@ use crate::db::DbConnection as Connection;
use walkdir::WalkDir;
use crate::db::queries;
use crate::engine::EngineResult;
use crate::engine::generation::has_published_snapshot;
use crate::engine::{EngineError, EngineResult};
use crate::model::Post;
use crate::render::{build_canonical_post_path, build_site_route_manifest};
@@ -25,11 +25,27 @@ pub fn validate_site(
data_dir: &Path,
project_id: &str,
) -> EngineResult<SiteValidationReport> {
validate_site_with_progress(conn, data_dir, project_id, |_, _| true)
}
pub fn validate_site_with_progress(
conn: &Connection,
data_dir: &Path,
project_id: &str,
mut on_progress: impl FnMut(usize, usize) -> bool,
) -> EngineResult<SiteValidationReport> {
const PHASES: usize = 4;
if !on_progress(0, PHASES) {
return Err(EngineError::Cancelled);
}
let metadata = crate::engine::meta::read_project_json(data_dir)?;
let output_dir = generated_output_dir(data_dir);
let published_posts = load_published_posts(conn, project_id)?;
let route_manifest = build_site_route_manifest(data_dir, &metadata, &published_posts)
.map_err(|error| crate::engine::EngineError::Parse(error.to_string()))?;
if !on_progress(1, PHASES) {
return Err(EngineError::Cancelled);
}
crate::engine::generation::refresh_validation_sitemap(
conn,
&output_dir,
@@ -43,6 +59,9 @@ pub fn validate_site(
.into_iter()
.map(|page| page.relative_path)
.collect::<HashSet<_>>();
if !on_progress(2, PHASES) {
return Err(EngineError::Cancelled);
}
let mut actual = HashSet::new();
let mut zero_byte = HashSet::new();
@@ -67,6 +86,9 @@ pub fn validate_site(
.into_iter()
.map(|file| (file.relative_path, file.updated_at))
.collect::<HashMap<_, _>>();
if !on_progress(3, PHASES) {
return Err(EngineError::Cancelled);
}
let mut stale_pages = stale_post_paths(
data_dir,
&output_dir,
@@ -83,6 +105,10 @@ pub fn validate_site(
stale_pages.sort();
stale_pages.dedup();
if !on_progress(PHASES, PHASES) {
return Err(EngineError::Cancelled);
}
Ok(SiteValidationReport {
missing_pages,
extra_pages,

16
crates/bds-core/src/engine/validate_translations.rs
View File

@@ -5,7 +5,7 @@ use std::path::Path;
use crate::db::DbConnection as Connection;
use crate::db::queries::{post as post_q, post_translation};
use crate::engine::EngineResult;
use crate::engine::{EngineError, EngineResult};
use crate::model::PostStatus;
/// Normalize a language code for comparison (lowercase, strip region).
@@ -53,7 +53,7 @@ pub struct TranslationValidationReport {
}
/// Per-item progress callback: (current_item, total_items, item_description).
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) + Send>;
pub type ItemProgressFn = Box<dyn Fn(usize, usize, &str) -> bool + Send>;
/// Validate all translations with optional per-item progress.
pub fn validate_translations_with_progress(
@@ -75,8 +75,10 @@ pub fn validate_translations_with_progress(
let post_count = posts.len();
for (i, post) in posts.iter().enumerate() {
if let Some(ref cb) = on_item {
cb(i + 1, post_count, &post.title);
if let Some(ref cb) = on_item
&& !cb(i + 1, post_count, &post.title)
{
return Err(EngineError::Cancelled);
}
let translations = post_translation::list_post_translations_by_post(conn, &post.id)?;
@@ -170,8 +172,10 @@ pub fn validate_translations_with_progress(
let path = entry.path();
let stem = path.file_stem().and_then(|s| s.to_str()).unwrap_or("");
if let Some(ref cb) = on_item {
cb(i + 1, fs_total, stem);
if let Some(ref cb) = on_item
&& !cb(i + 1, fs_total, stem)
{
return Err(EngineError::Cancelled);
}
checked_fs_files += 1;

1
crates/bds-core/src/render/mod.rs
View File

@@ -21,6 +21,7 @@ pub use routes::{
render_starter_list_page_with_media_map, render_starter_single_post_page,
render_starter_single_post_page_with_media_map,
};
pub(crate) use site::count_site_render_pages_from_context;
pub use site::{
PagefindDocument, PreviewRenderResult, SitePage, SiteRenderArtifacts, SiteRenderContext,
build_preview_response, build_site_render_artifacts, build_site_render_artifacts_from_context,

40
crates/bds-core/src/render/site.rs
View File

@@ -391,12 +391,7 @@ pub fn prepare_site_render_context(
let linked_media_by_post_id =
build_linked_media_by_post_id(&posts, &linked_media_by_source_post);
let post_data_json_by_id = build_post_data_json_by_id(&posts, &linked_media_by_post_id);
let menu_items = build_menu_items(
data_dir,
&language,
&main_language,
&category_settings,
)?;
let menu_items = build_menu_items(data_dir, &language, &main_language, &category_settings)?;
let canonical_post_path_by_slug =
canonical_post_path_by_slug(&posts, &language, &main_language);
let taxonomy = build_taxonomy_context(&posts, &tags);
@@ -590,6 +585,39 @@ pub fn build_site_render_artifacts_from_context(
Ok(artifacts)
}
pub(crate) fn count_site_render_pages_from_context(
context: &SiteRenderContext,
section: GenerationSection,
requested_paths: Option<&HashSet<String>>,
) -> usize {
context
.languages
.iter()
.map(|language_context| {
let expanded = requested_paths.map(|requested| {
expand_requested_aggregate_paths(
requested,
&language_context.posts,
&language_context.routes,
&language_context.language,
&context.metadata,
)
});
language_context
.routes
.iter()
.filter(|route| {
classify_generated_path(&route.relative_path, &context.metadata)
== Some(section)
&& expanded
.as_ref()
.is_none_or(|requested| requested.contains(&route.relative_path))
})
.count()
})
.sum()
}
pub fn build_preview_response(
conn: &Connection,
data_dir: &Path,

54
crates/bds-core/src/scripting/core_host.rs
View File

@@ -1,5 +1,6 @@
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use base64::Engine as _;
use chrono::{SecondsFormat, TimeZone, Utc};
@@ -1215,33 +1216,39 @@ impl CoreHost {
"get" => {
let id = task_id_arg(args, 0)?;
manager
.snapshots()
.into_iter()
.find(|task| task.id == id)
.get(id)
.map(public_task)
.transpose()
.map(|value| value.unwrap_or(Value::Null))
}
"get_all" => public_tasks(manager.snapshots()),
"get_running" => public_tasks(
manager
.snapshots()
.into_iter()
.filter(|task| task.status == TaskStatus::Running)
.collect(),
),
"get_all" => public_tasks(manager.all()),
"get_running" => public_tasks(manager.running()),
"status_snapshot" => {
let tasks = manager.snapshots();
let active = tasks
.iter()
.filter(|task| matches!(task.status, TaskStatus::Pending | TaskStatus::Running))
.collect::<Vec<_>>();
let running_task_message = active.first().map(|task| {
if task.status == TaskStatus::Pending {
format!("Queued: {}", task.label)
} else if let Some(message) =
task.message.as_deref().filter(|value| !value.is_empty())
{
format!("{}: {message}", task.label)
} else {
task.label.clone()
}
});
Ok(json!({
"active_count": tasks.iter().filter(|task| matches!(task.status, TaskStatus::Pending | TaskStatus::Running)).count(),
"active_count": active.len(),
"running_count": manager.running_count(), "pending_count": manager.pending_count(),
"running_task_message": running_task_message,
"running_task_overflow": active.len().saturating_sub(1),
"tasks": public_tasks(tasks)?,
}))
}
"cancel" => {
manager.cancel(task_id_arg(args, 0)?);
Ok(Value::Bool(true))
}
"cancel" => Ok(Value::Bool(manager.cancel(task_id_arg(args, 0)?))),
"clear_completed" => {
manager.clear_completed();
Ok(Value::Bool(true))
@@ -1355,10 +1362,17 @@ impl CoreHost {
}
let manager = self.task_manager.clone();
let task_id = self.task_id;
engine::publishing::upload_site(
let cancel_flag = manager
.as_ref()
.zip(task_id)
.and_then(|(manager, task_id)| manager.cancellation_flag(task_id))
.unwrap_or_else(|| Arc::new(AtomicBool::new(false)));
let progress_cancel_flag = Arc::clone(&cancel_flag);
engine::publishing::upload_site_cancellable(
&self.data_dir,
&self.private_cache_dir,
&preferences,
cancel_flag.as_ref(),
move |current, total, _| {
if let (Some(manager), Some(task_id)) = (&manager, task_id) {
manager.report_progress(
@@ -1367,6 +1381,7 @@ impl CoreHost {
Some("uploading site".into()),
);
}
!progress_cancel_flag.load(Ordering::Acquire)
},
)?;
match (&self.task_manager, self.task_id) {
@@ -1659,6 +1674,8 @@ fn public_tag(value: Tag) -> HostResult<Value> {
}
fn public_task(value: TaskSnapshot) -> HostResult<Value> {
let cancellable = matches!(value.status, TaskStatus::Pending | TaskStatus::Running)
&& !value.cancellation_requested;
let (status, error) = match value.status {
TaskStatus::Pending => ("pending", None),
TaskStatus::Running => ("running", None),
@@ -1668,6 +1685,9 @@ fn public_task(value: TaskSnapshot) -> HostResult<Value> {
};
Ok(
json!({"id":value.id.to_string(),"name":value.label,"status":status,
"group_id":value.group_id,"group_name":value.group_name,
"cancellable":cancellable,
"cancellation_requested":value.cancellation_requested,
"progress":value.progress,"message":value.message.or(error)}),
)
}

4
crates/bds-core/src/scripting/manifest.rs
View File

@@ -395,6 +395,10 @@ fn field_description(field: &str) -> &'static str {
"updated" => "Whether the Git network command completed successfully.",
"progress" => "Completion value reported by the task.",
"message" => "Latest user-facing task message.",
"cancellable" => "Whether a cancellation request can still be submitted.",
"cancellation_requested" => "Whether cooperative cancellation is in progress.",
"group_id" => "Identifier shared by tasks in one workflow.",
"group_name" => "Human-readable workflow name.",
"created_at" => "Creation timestamp.",
"updated_at" => "Last-update timestamp.",
"language" | "main_language" => "BCP 47 language code.",