crashlabs/huskies
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 26 Wiki Activity

huskies: merge 530_story_eliminate_filesystem_markdown_shadows_entirely_crdt_db_is_the_only_story_store

Browse Source
This commit is contained in:
dave
2026-04-10 14:56:13 +00:00
parent 1dd675796b
commit 11d19d8902
26 changed files with 966 additions and 1668 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/io/watcher.rs
+131 -214
View File
@@ -25,7 +25,7 @@ use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher, recommended_
use serde::Serialize;
use std::path::{Path, PathBuf};
use std::sync::mpsc;
use std::time::{Duration, Instant, SystemTime};
use std::time::{Duration, Instant};
use tokio::sync::broadcast;
/// A lifecycle event emitted by the filesystem watcher.
@@ -322,88 +322,38 @@ fn flush_pending(
let _ = event_tx.send(evt);
}
/// Scan `work/5_done/` and move any `.md` files whose mtime is older than
/// `done_retention` to `work/6_archived/`. After each successful promotion,
/// removes the associated git worktree (if any) via [`crate::worktree::prune_worktree_sync`].
/// Sweep items in `5_done` whose `merged_at` timestamp exceeds the retention
/// duration to `6_archived` via CRDT state transitions. Also prunes worktrees
/// for items already in `6_archived`.
///
/// Also scans `work/6_archived/` for stories that still have a live worktree
/// and removes them (catches items that were archived before this sweep was
/// added).
///
/// Worktree removal failures are logged but never block the file move or other
/// cleanup work.
///
/// Called periodically from the watcher thread. File moves will trigger normal
/// watcher events, which `flush_pending` will commit and broadcast.
fn sweep_done_to_archived(work_dir: &Path, git_root: &Path, done_retention: Duration) {
// ── Part 1: promote old items from 5_done/ → 6_archived/ ───────────────
let done_dir = work_dir.join("5_done");
if done_dir.exists() {
let archived_dir = work_dir.join("6_archived");
/// All state is read from CRDT — no filesystem access.
fn sweep_done_to_archived(_work_dir: &Path, git_root: &Path, done_retention: Duration) {
use crate::pipeline_state::{Stage, read_all_typed};
match std::fs::read_dir(&done_dir) {
Err(e) => slog!("[watcher] sweep: failed to read 5_done/: {e}"),
Ok(entries) => {
for entry in entries.flatten() {
let path = entry.path();
if path.extension().is_none_or(|e| e != "md") {
continue;
}
let mtime = match entry.metadata().and_then(|m| m.modified()) {
Ok(t) => t,
Err(_) => continue,
};
let age = SystemTime::now().duration_since(mtime).unwrap_or_default();
if age >= done_retention {
if let Err(e) = std::fs::create_dir_all(&archived_dir) {
slog!("[watcher] sweep: failed to create 6_archived/: {e}");
continue;
}
let dest = archived_dir.join(entry.file_name());
match std::fs::rename(&path, &dest) {
Ok(()) => {
let item_id = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("unknown");
slog!("[watcher] sweep: promoted {item_id} → 6_archived/");
// Prune the worktree for this story (best effort).
if let Err(e) =
crate::worktree::prune_worktree_sync(git_root, item_id)
{
slog!(
"[watcher] sweep: worktree prune failed for {item_id}: {e}"
);
}
}
Err(e) => {
slog!("[watcher] sweep: failed to move {}: {e}", path.display());
}
}
for item in read_all_typed() {
match &item.stage {
Stage::Done { merged_at, .. } => {
let age = chrono::Utc::now()
.signed_duration_since(*merged_at)
.to_std()
.unwrap_or_default();
if age >= done_retention {
let story_id = &item.story_id.0;
crate::db::move_item_stage(story_id, "6_archived", None);
slog!("[watcher] sweep: promoted {story_id} → 6_archived/");
if let Err(e) = crate::worktree::prune_worktree_sync(git_root, story_id) {
slog!("[watcher] sweep: worktree prune failed for {story_id}: {e}");
}
}
}
}
}
// ── Part 2: prune stale worktrees for items already in 6_archived/ ──────
let archived_dir = work_dir.join("6_archived");
if archived_dir.exists()
&& let Ok(entries) = std::fs::read_dir(&archived_dir)
{
for entry in entries.flatten() {
let path = entry.path();
if path.extension().is_none_or(|e| e != "md") {
continue;
}
if let Some(item_id) = path.file_stem().and_then(|s| s.to_str())
&& let Err(e) = crate::worktree::prune_worktree_sync(git_root, item_id)
{
slog!("[watcher] sweep: worktree prune failed for {item_id}: {e}");
Stage::Archived { .. } => {
// Prune stale worktrees for archived items.
let story_id = &item.story_id.0;
if let Err(e) = crate::worktree::prune_worktree_sync(git_root, story_id) {
slog!("[watcher] sweep: worktree prune failed for {story_id}: {e}");
}
}
_ => {}
}
}
}
@@ -1149,105 +1099,90 @@ mod tests {
assert!(!is_config_file(&other_root_config, &git_root));
}
// ── sweep_done_to_archived ────────────────────────────────────────────────
// ── sweep_done_to_archived (CRDT-based) ─────────────────────────────────
//
// The sweep function now reads from `read_all_typed()` and checks
// `Stage::Done { merged_at, .. }`. Items created via
// `write_item_with_content("5_done")` project `merged_at = Utc::now()`,
// so we test with Duration::ZERO to sweep immediately and with a long
// retention to verify items are kept.
#[test]
fn sweep_moves_old_items_to_archived() {
let tmp = TempDir::new().unwrap();
let work_dir = tmp.path().join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
let archived_dir = work_dir.join("6_archived");
fs::create_dir_all(&done_dir).unwrap();
// Write a file and backdate its mtime to 5 hours ago.
let story_path = done_dir.join("10_story_old.md");
fs::write(&story_path, "---\nname: old\n---\n").unwrap();
let past = SystemTime::now()
.checked_sub(Duration::from_secs(5 * 60 * 60))
.unwrap();
filetime::set_file_mtime(&story_path, filetime::FileTime::from_system_time(past)).unwrap();
let retention = Duration::from_secs(4 * 60 * 60);
// tmp.path() has no worktrees dir — prune_worktree_sync is a no-op.
sweep_done_to_archived(&work_dir, tmp.path(), retention);
assert!(
!story_path.exists(),
"old item should be moved out of 5_done/"
crate::db::ensure_content_store();
crate::db::write_item_with_content(
"9880_story_sweep_old",
"5_done",
"---\nname: old\n---\n",
);
// With ZERO retention, any Done item should be swept.
sweep_done_to_archived(
&tmp.path().join(".huskies/work"),
tmp.path(),
Duration::ZERO,
);
// Verify the item was moved to 6_archived in the CRDT.
let items = crate::pipeline_state::read_all_typed();
let item = items.iter().find(|i| i.story_id.0 == "9880_story_sweep_old");
assert!(
archived_dir.join("10_story_old.md").exists(),
"old item should appear in 6_archived/"
item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Archived { .. })),
"item should be archived after sweep"
);
}
#[test]
fn sweep_keeps_recent_items_in_done() {
let tmp = TempDir::new().unwrap();
let work_dir = tmp.path().join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
fs::create_dir_all(&done_dir).unwrap();
// Write a file with a recent mtime (now).
let story_path = done_dir.join("11_story_new.md");
fs::write(&story_path, "---\nname: new\n---\n").unwrap();
crate::db::ensure_content_store();
crate::db::write_item_with_content(
"9881_story_sweep_new",
"5_done",
"---\nname: new\n---\n",
);
let retention = Duration::from_secs(4 * 60 * 60);
sweep_done_to_archived(&work_dir, tmp.path(), retention);
// With a very long retention, the item (merged_at ≈ now) should stay.
sweep_done_to_archived(
&tmp.path().join(".huskies/work"),
tmp.path(),
Duration::from_secs(999_999),
);
assert!(story_path.exists(), "recent item should remain in 5_done/");
let items = crate::pipeline_state::read_all_typed();
let item = items.iter().find(|i| i.story_id.0 == "9881_story_sweep_new");
assert!(
item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Done { .. })),
"item should remain in Done with long retention"
);
}
#[test]
fn sweep_respects_custom_retention() {
let tmp = TempDir::new().unwrap();
let work_dir = tmp.path().join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
let archived_dir = work_dir.join("6_archived");
fs::create_dir_all(&done_dir).unwrap();
// Write a file and backdate its mtime to 2 minutes ago.
let story_path = done_dir.join("12_story_custom.md");
fs::write(&story_path, "---\nname: custom\n---\n").unwrap();
let past = SystemTime::now()
.checked_sub(Duration::from_secs(120))
.unwrap();
filetime::set_file_mtime(&story_path, filetime::FileTime::from_system_time(past)).unwrap();
// With a 1-minute retention, the 2-minute-old file should be swept.
sweep_done_to_archived(&work_dir, tmp.path(), Duration::from_secs(60));
assert!(
!story_path.exists(),
"item older than custom retention should be moved"
crate::db::ensure_content_store();
crate::db::write_item_with_content(
"9882_story_sweep_custom",
"5_done",
"---\nname: custom\n---\n",
);
assert!(
archived_dir.join("12_story_custom.md").exists(),
"item should appear in 6_archived/"
// With ZERO retention, sweep should promote.
sweep_done_to_archived(
&tmp.path().join(".huskies/work"),
tmp.path(),
Duration::ZERO,
);
}
#[test]
fn sweep_custom_retention_keeps_younger_items() {
let tmp = TempDir::new().unwrap();
let work_dir = tmp.path().join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
fs::create_dir_all(&done_dir).unwrap();
// Write a file and backdate its mtime to 30 seconds ago.
let story_path = done_dir.join("13_story_young.md");
fs::write(&story_path, "---\nname: young\n---\n").unwrap();
let past = SystemTime::now()
.checked_sub(Duration::from_secs(30))
.unwrap();
filetime::set_file_mtime(&story_path, filetime::FileTime::from_system_time(past)).unwrap();
// With a 1-minute retention, the 30-second-old file should stay.
sweep_done_to_archived(&work_dir, tmp.path(), Duration::from_secs(60));
let items = crate::pipeline_state::read_all_typed();
let item = items.iter().find(|i| i.story_id.0 == "9882_story_sweep_custom");
assert!(
story_path.exists(),
"item younger than custom retention should remain"
item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Archived { .. })),
"item should be archived with zero retention"
);
}
@@ -1256,7 +1191,6 @@ mod tests {
/// Helper: create a real git worktree at `wt_path` on a new branch.
fn create_git_worktree(git_root: &std::path::Path, wt_path: &std::path::Path, branch: &str) {
use std::process::Command;
// Create the branch first (ignore errors if it already exists).
let _ = Command::new("git")
.args(["branch", branch])
.current_dir(git_root)
@@ -1274,17 +1208,13 @@ mod tests {
let git_root = tmp.path().to_path_buf();
init_git_repo(&git_root);
let work_dir = git_root.join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
fs::create_dir_all(&done_dir).unwrap();
let story_id = "60_story_prune_on_promote";
let story_path = done_dir.join(format!("{story_id}.md"));
fs::write(&story_path, "---\nname: test\n---\n").unwrap();
let past = SystemTime::now()
.checked_sub(Duration::from_secs(5 * 60 * 60))
.unwrap();
filetime::set_file_mtime(&story_path, filetime::FileTime::from_system_time(past)).unwrap();
crate::db::ensure_content_store();
let story_id = "9883_story_prune_on_promote";
crate::db::write_item_with_content(
story_id,
"5_done",
"---\nname: test\n---\n",
);
// Create a real git worktree for this story.
let wt_path = crate::worktree::worktree_path(&git_root, story_id);
@@ -1292,17 +1222,18 @@ mod tests {
create_git_worktree(&git_root, &wt_path, &format!("feature/story-{story_id}"));
assert!(wt_path.exists(), "worktree must exist before sweep");
let retention = Duration::from_secs(4 * 60 * 60);
sweep_done_to_archived(&work_dir, &git_root, retention);
sweep_done_to_archived(
&git_root.join(".huskies/work"),
&git_root,
Duration::ZERO,
);
// Story must be archived.
assert!(!story_path.exists(), "story should be moved out of 5_done/");
// Story must be archived in CRDT.
let items = crate::pipeline_state::read_all_typed();
let item = items.iter().find(|i| i.story_id.0 == story_id);
assert!(
work_dir
.join("6_archived")
.join(format!("{story_id}.md"))
.exists(),
"story should appear in 6_archived/"
item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Archived { .. })),
"story should be archived"
);
// Worktree must be removed.
assert!(
@@ -1317,14 +1248,13 @@ mod tests {
let git_root = tmp.path().to_path_buf();
init_git_repo(&git_root);
let work_dir = git_root.join(".huskies").join("work");
let archived_dir = work_dir.join("6_archived");
fs::create_dir_all(&archived_dir).unwrap();
// Story is already in 6_archived.
let story_id = "61_story_stale_worktree";
let story_path = archived_dir.join(format!("{story_id}.md"));
fs::write(&story_path, "---\nname: stale\n---\n").unwrap();
crate::db::ensure_content_store();
let story_id = "9884_story_stale_worktree";
crate::db::write_item_with_content(
story_id,
"6_archived",
"---\nname: stale\n---\n",
);
// Create a real git worktree that was never cleaned up.
let wt_path = crate::worktree::worktree_path(&git_root, story_id);
@@ -1332,63 +1262,50 @@ mod tests {
create_git_worktree(&git_root, &wt_path, &format!("feature/story-{story_id}"));
assert!(wt_path.exists(), "stale worktree must exist before sweep");
// 5_done/ is empty — only Part 2 runs.
fs::create_dir_all(work_dir.join("5_done")).unwrap();
let retention = Duration::from_secs(4 * 60 * 60);
sweep_done_to_archived(&work_dir, &git_root, retention);
sweep_done_to_archived(
&git_root.join(".huskies/work"),
&git_root,
Duration::from_secs(999_999),
);
// Stale worktree should be pruned.
assert!(
!wt_path.exists(),
"stale worktree should be pruned by sweep"
);
// Story file must remain untouched.
assert!(
story_path.exists(),
"archived story file must not be removed"
);
}
#[test]
fn sweep_archives_story_even_when_worktree_removal_fails() {
// Use a git repo so prune_worktree_sync can attempt removal,
// but the fake directory is not a registered git worktree so
// `git worktree remove` will fail — the story must still be archived.
let tmp = TempDir::new().unwrap();
let git_root = tmp.path().to_path_buf();
init_git_repo(&git_root);
let work_dir = git_root.join(".huskies").join("work");
let done_dir = work_dir.join("5_done");
fs::create_dir_all(&done_dir).unwrap();
let story_id = "62_story_fake_worktree";
let story_path = done_dir.join(format!("{story_id}.md"));
fs::write(&story_path, "---\nname: test\n---\n").unwrap();
let past = SystemTime::now()
.checked_sub(Duration::from_secs(5 * 60 * 60))
.unwrap();
filetime::set_file_mtime(&story_path, filetime::FileTime::from_system_time(past)).unwrap();
crate::db::ensure_content_store();
let story_id = "9885_story_fake_worktree";
crate::db::write_item_with_content(
story_id,
"5_done",
"---\nname: test\n---\n",
);
// Create a plain directory at the expected worktree path — not a real
// git worktree, so `git worktree remove` will fail.
let wt_path = crate::worktree::worktree_path(&git_root, story_id);
fs::create_dir_all(&wt_path).unwrap();
let retention = Duration::from_secs(4 * 60 * 60);
sweep_done_to_archived(&work_dir, &git_root, retention);
// Story must still be archived despite the worktree removal failure.
assert!(
!story_path.exists(),
"story should be archived even when worktree removal fails"
sweep_done_to_archived(
&git_root.join(".huskies/work"),
&git_root,
Duration::ZERO,
);
// Story must be archived in CRDT despite worktree removal failure.
let items = crate::pipeline_state::read_all_typed();
let item = items.iter().find(|i| i.story_id.0 == story_id);
assert!(
work_dir
.join("6_archived")
.join(format!("{story_id}.md"))
.exists(),
"story should appear in 6_archived/ despite worktree removal failure"
item.is_some_and(|i| matches!(i.stage, crate::pipeline_state::Stage::Archived { .. })),
"story should be archived even when worktree removal fails"
);
}
}