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

huskies: merge 1222 bug show fails with "content unavailable" on done/archived stories — content should never be evicted

Browse Source
This commit is contained in:
Huskies Agent
2026-07-18 19:56:30 +00:00
parent 2b0e8e6f10
commit a6cce683f5
4 changed files with 373 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/db/gc.rs
+309 -34
View File
@@ -1,8 +1,10 @@
//! Content-store garbage collection: TransitionFired subscriber and startup sweep.
//! Content-store garbage collection: TransitionFired subscriber, startup
//! sweep, and story-content backfill.
//!
//! When a pipeline item reaches a terminal stage (Done, Archived, Abandoned,
//! Superseded, Rejected) every `ContentKey::*` entry for that story is purged
//! from the in-memory content store. There are two purge paths:
//! Two purge paths clear the *ephemeral* `ContentKey::*` working-state keys
//! (gate output, respawn counters, merge-fixup flags, etc.) when a pipeline
//! item reaches a terminal stage (Done, Archived, Abandoned, Superseded,
//! Rejected):
//!
//! 1. **Subscriber** ([`spawn_content_gc_subscriber`]) — reacts to
//! [`crate::pipeline_state::TransitionFired`] events and runs for new
@@ -10,19 +12,35 @@
//!
//! 2. **Startup sweep** ([`sweep_zombie_content_on_startup`]) — cleans up
//! zombie entries left over from sessions that predate the subscriber.
//!
//! `ContentKey::Story` — the story's markdown body — is deliberately
//! excluded from both paths (story 1222): reaching a terminal stage is not
//! deletion, and `show()` must keep returning content for Done/Archived
//! stories indefinitely. The story body is only fully purged
//! ([`purge_content_keys_for_story`]) when a story is genuinely tombstoned
//! (absent from the live CRDT index — `evict_item`/`purge_story`), which
//! the startup sweep still detects and cleans up.
//!
//! [`backfill_evicted_story_content`] is an idempotent, run-on-every-startup
//! repair pass that restores `ContentKey::Story` for any terminal-stage
//! story whose in-memory content was wiped by the pre-1222 purge bug, using
//! the durable SQLite `pipeline_items.content` shadow column (which the
//! purge never touched). Stories with no SQLite copy either are reported as
//! unrecoverable.
use crate::db::{ContentKey, all_content_ids, delete_content};
use crate::db::{ContentKey, all_content_ids, delete_content, get_shared_pool, write_content};
use crate::pipeline_state::{Pipeline, Stage, Status};
use crate::slog;
use crate::slog_warn;
/// Purge every [`ContentKey`] variant for `story_id` from the in-memory content store.
/// Purge the eight ephemeral (non-body) `ContentKey` variants for
/// `story_id` — everything except `ContentKey::Story`.
///
/// All eight key namespaces are deleted unconditionally — deletes for absent
/// keys are no-ops. Call this when a work item reaches a terminal stage to
/// prevent long-lived zombie entries from accumulating in the process heap.
pub(crate) fn purge_content_keys_for_story(story_id: &str) {
delete_content(ContentKey::Story(story_id));
/// Call this when a work item reaches a terminal stage: agent working-state
/// (gate output, respawn counters, merge-fixup flags) is no longer needed,
/// but the story's markdown body must be retained indefinitely so `show()`
/// keeps working (story 1222). Deletes for absent keys are no-ops.
pub(crate) fn purge_ephemeral_content_keys_for_story(story_id: &str) {
delete_content(ContentKey::GateOutput(story_id));
delete_content(ContentKey::AbortRespawnCount(story_id));
delete_content(ContentKey::MergeMasterSpawnCount(story_id));
@@ -33,13 +51,28 @@ pub(crate) fn purge_content_keys_for_story(story_id: &str) {
delete_content(ContentKey::MergeReport(story_id));
}
/// Spawn a background task that purges content-store entries when a story reaches a terminal stage.
/// Purge every `ContentKey` variant for `story_id`, including the markdown
/// body (`ContentKey::Story`), from the in-memory content store.
///
/// Subscribes to [`crate::pipeline_state::subscribe_transitions`]. On each
/// Use this ONLY for stories that are genuinely gone — tombstoned / absent
/// from the live CRDT index via `evict_item`/`purge_story`. A terminal
/// pipeline stage (Done, Archived, …) is NOT deletion — use
/// [`purge_ephemeral_content_keys_for_story`] for that case so `show()`
/// keeps returning content (story 1222). Deletes for absent keys are no-ops.
pub(crate) fn purge_content_keys_for_story(story_id: &str) {
delete_content(ContentKey::Story(story_id));
purge_ephemeral_content_keys_for_story(story_id);
}
/// Spawn a background task that purges ephemeral content-store entries when a story reaches a terminal stage.
///
/// Subscribes to [`crate::pipeline_state::subscribe_transitions`]. On each
/// [`crate::pipeline_state::TransitionFired`] where `after` is `Done`,
/// `Archived`, `Abandoned`, `Superseded`, or `Rejected`, all `ContentKey::*`
/// entries for that story are purged. Lag events are logged as warnings —
/// a missed event leaves zombie entries that the next startup sweep will remove.
/// `Archived`, `Abandoned`, `Superseded`, or `Rejected`, the eight ephemeral
/// `ContentKey::*` entries for that story are purged — the markdown body
/// (`ContentKey::Story`) is retained (story 1222). Lag events are logged as
/// warnings — a missed event leaves zombie entries that the next startup
/// sweep will remove.
pub(crate) fn spawn_content_gc_subscriber() {
let mut rx = crate::pipeline_state::subscribe_transitions();
tokio::spawn(async move {
@@ -50,9 +83,9 @@ pub(crate) fn spawn_content_gc_subscriber() {
let story_id = &fired.story_id.0;
slog!(
"[content-gc] Story '{story_id}' reached terminal stage; \
purging all content-store entries."
purging ephemeral content-store entries (body retained)."
);
purge_content_keys_for_story(story_id);
purge_ephemeral_content_keys_for_story(story_id);
}
}
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
@@ -70,7 +103,12 @@ pub(crate) fn spawn_content_gc_subscriber() {
/// One-shot startup sweep: purge content-store entries for stories that have
/// already reached terminal stages or are absent from the CRDT.
///
/// Idempotent — safe to call more than once. Intended to clean up zombie
/// Genuinely tombstoned/absent stories get a full purge (including the
/// markdown body). Stories that are still live in the CRDT but sitting in a
/// terminal stage only have their ephemeral working-state keys purged — the
/// markdown body is retained so `show()` keeps working (story 1222).
///
/// Idempotent — safe to call more than once. Intended to clean up zombie
/// entries left over from server sessions that predate the GC subscriber.
pub(crate) fn sweep_zombie_content_on_startup() {
let raw_keys = all_content_ids();
@@ -92,25 +130,94 @@ pub(crate) fn sweep_zombie_content_on_startup() {
let mut swept = 0usize;
for story_id in &story_ids {
let should_purge = match crate::crdt_state::read_item(story_id) {
// Tombstoned or absent from the live CRDT index — purge.
None => true,
Some(item) => is_terminal_stage(item.stage()),
};
if should_purge {
purge_content_keys_for_story(story_id);
swept += 1;
match crate::crdt_state::read_item(story_id) {
// Tombstoned or absent from the live CRDT index — genuinely
// gone, safe to purge everything including the story body.
None => {
purge_content_keys_for_story(story_id);
swept += 1;
}
// Still a live item, just sitting in a terminal stage — retain
// the story body (story 1222), only purge ephemeral state.
Some(item) if is_terminal_stage(item.stage()) => {
purge_ephemeral_content_keys_for_story(story_id);
swept += 1;
}
Some(_) => {}
}
}
if swept > 0 {
slog!(
"[content-gc] Startup sweep purged content-store entries for \
"[content-gc] Startup sweep purged ephemeral content-store entries for \
{swept} zombie story(s)."
);
}
}
/// Restore in-memory `ContentKey::Story` content for terminal-stage stories
/// whose markdown body was wiped by the pre-1222 purge bug (which deleted
/// `ContentKey::Story` on every terminal transition), using the durable
/// SQLite `pipeline_items.content` shadow column — `delete_content` only
/// ever touched the in-memory map, so the SQLite copy survived.
///
/// Idempotent and cheap to run on every startup: stories that already have
/// in-memory content are skipped. Returns the story IDs that could not be
/// recovered (also absent from the SQLite shadow column) — the caller logs
/// this as a report (story 1222, AC4).
pub(crate) async fn backfill_evicted_story_content() -> Vec<String> {
let Some(pool) = get_shared_pool() else {
return Vec::new();
};
let Some(items) = crate::crdt_state::read_all_items() else {
return Vec::new();
};
let mut restored = 0usize;
let mut unrecoverable = Vec::new();
for item in items {
if !is_terminal_stage(item.stage()) {
continue;
}
let story_id = item.story_id();
if crate::db::read_content(ContentKey::Story(story_id)).is_some() {
continue;
}
let row: Result<Option<(Option<String>,)>, sqlx::Error> =
sqlx::query_as("SELECT content FROM pipeline_items WHERE id = ?1")
.bind(story_id)
.fetch_optional(pool)
.await;
match row {
Ok(Some((Some(content),))) => {
write_content(ContentKey::Story(story_id), &content);
restored += 1;
}
_ => unrecoverable.push(story_id.to_string()),
}
}
if restored > 0 {
slog!(
"[content-gc] Backfill restored content for {restored} terminal-stage \
story(s) from the SQLite shadow table."
);
}
if !unrecoverable.is_empty() {
slog_warn!(
"[content-gc] Backfill could not recover content for {} story(s) — \
compacted away with no SQLite copy: {}",
unrecoverable.len(),
unrecoverable.join(", ")
);
}
unrecoverable
}
/// Return `true` when `stage` is one of the terminal pipeline classifications.
///
/// Story 1086: matches via the [`Status`] projection (Done / Abandoned /
@@ -181,6 +288,44 @@ mod tests {
);
}
/// Assert the seven ephemeral ContentKey variants seeded by
/// `seed_all_keys` are absent, but `ContentKey::Story` (the markdown
/// body) is still present — story 1222.
fn assert_ephemeral_keys_absent_but_story_present(story_id: &str) {
assert!(
read_content(ContentKey::Story(story_id)).is_some(),
"Story key must be RETAINED across a terminal-stage transition (story 1222)"
);
assert!(
read_content(ContentKey::GateOutput(story_id)).is_none(),
"GateOutput key must be absent"
);
assert!(
read_content(ContentKey::AbortRespawnCount(story_id)).is_none(),
"AbortRespawnCount key must be absent"
);
assert!(
read_content(ContentKey::MergeMasterSpawnCount(story_id)).is_none(),
"MergeMasterSpawnCount key must be absent"
);
assert!(
read_content(ContentKey::RunTestsOk(story_id)).is_none(),
"RunTestsOk key must be absent"
);
assert!(
read_content(ContentKey::CommitRecoveryPending(story_id)).is_none(),
"CommitRecoveryPending key must be absent"
);
assert!(
read_content(ContentKey::MergeFixupPending(story_id)).is_none(),
"MergeFixupPending key must be absent"
);
assert!(
read_content(ContentKey::MergeFailureKind(story_id)).is_none(),
"MergeFailureKind key must be absent"
);
}
/// AC1: purge_content_keys_for_story removes all eight ContentKey namespaces.
#[test]
fn purge_clears_all_eight_content_key_namespaces() {
@@ -216,10 +361,11 @@ mod tests {
assert_all_keys_absent(id);
}
/// AC1 + AC4: the GC subscriber reacts to an Abandoned terminal transition and
/// purges all content-store entries for the story.
/// Story 1222, AC1: the GC subscriber reacts to an Abandoned terminal
/// transition and purges ephemeral content-store entries for the story,
/// but RETAINS the story body so `show()` keeps working.
#[tokio::test]
async fn subscriber_purges_content_on_terminal_transition() {
async fn subscriber_purges_ephemeral_keys_but_retains_story_content_on_terminal_transition() {
crate::crdt_state::init_for_test();
ensure_content_store();
@@ -242,7 +388,7 @@ mod tests {
// Give the subscriber task time to run.
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
assert_all_keys_absent(story_id);
assert_ephemeral_keys_absent_but_story_present(story_id);
}
/// AC4: the subscriber does NOT purge content for stories that remain in
@@ -272,8 +418,9 @@ mod tests {
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
// Terminal story's content must be gone.
assert_all_keys_absent(terminal_id);
// Terminal story's ephemeral keys must be gone, but body retained
// (story 1222).
assert_ephemeral_keys_absent_but_story_present(terminal_id);
// Active story's main content key must still be present.
assert!(
@@ -309,6 +456,49 @@ mod tests {
assert_all_keys_absent(story_id);
}
/// Story 1222, AC1 + AC3: a story that reaches Done but is still LIVE in
/// the CRDT (not tombstoned) must keep its body across the startup sweep
/// — only ephemeral working-state keys are purged.
#[test]
fn startup_sweep_retains_story_content_for_live_done_item() {
crate::crdt_state::init_for_test();
ensure_content_store();
let story_id = "1222_test_sweep_live_done";
write_item_with_content(
story_id,
"5_done",
"---\nname: Live Done Test\n---\n",
ItemMeta::named("Live Done Test"),
);
seed_all_keys(story_id);
sweep_zombie_content_on_startup();
assert_ephemeral_keys_absent_but_story_present(story_id);
}
/// Story 1222, AC1 + AC3: same as above but for Archived, the other
/// stage explicitly named in the AC.
#[test]
fn startup_sweep_retains_story_content_for_live_archived_item() {
crate::crdt_state::init_for_test();
ensure_content_store();
let story_id = "1222_test_sweep_live_archived";
write_item_with_content(
story_id,
"6_archived",
"---\nname: Live Archived Test\n---\n",
ItemMeta::named("Live Archived Test"),
);
seed_all_keys(story_id);
sweep_zombie_content_on_startup();
assert_ephemeral_keys_absent_but_story_present(story_id);
}
/// AC2: sweep_zombie_content_on_startup leaves active stories' content intact.
#[test]
fn startup_sweep_preserves_active_story_content() {
@@ -342,4 +532,89 @@ mod tests {
sweep_zombie_content_on_startup();
sweep_zombie_content_on_startup();
}
/// Story 1222, AC4: with no shared SQLite pool initialised, the backfill
/// is a safe no-op (returns no unrecoverable IDs, does not panic).
#[tokio::test]
async fn backfill_is_noop_without_shared_pool() {
crate::crdt_state::init_for_test();
ensure_content_store();
let unrecoverable = backfill_evicted_story_content().await;
assert!(unrecoverable.is_empty());
}
/// Story 1222, AC4: the backfill restores in-memory Story content for a
/// terminal-stage story whose body was wiped from the in-memory store,
/// using the durable SQLite shadow column that the (buggy, now-fixed)
/// terminal-transition purge never touched.
#[tokio::test]
async fn backfill_restores_story_content_from_sqlite_shadow_column() {
crate::crdt_state::init_for_test();
ensure_content_store();
crate::db::tests::ensure_shadow_db().await;
let story_id = "1222_test_backfill_restore";
let body = "---\nname: Backfill Restore Test\n---\n# Body\n";
write_item_with_content(
story_id,
"5_done",
body,
ItemMeta::named("Backfill Restore Test"),
);
// Let the shadow-write background task flush the insert to SQLite.
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
// Simulate the pre-1222 bug: wipe the in-memory Story key only.
// SQLite still has it — delete_content never touches the shadow
// table.
delete_content(ContentKey::Story(story_id));
assert!(read_content(ContentKey::Story(story_id)).is_none());
let unrecoverable = backfill_evicted_story_content().await;
assert!(
!unrecoverable.contains(&story_id.to_string()),
"story with a surviving SQLite copy must not be reported unrecoverable"
);
assert_eq!(
read_content(ContentKey::Story(story_id)).as_deref(),
Some(body),
"backfill must restore the body from the SQLite shadow column"
);
}
/// Story 1222, AC4: a terminal-stage story with no SQLite copy either
/// (never shadow-written) is reported as unrecoverable, not silently
/// dropped.
#[tokio::test]
async fn backfill_reports_unrecoverable_story_with_no_sqlite_copy() {
crate::crdt_state::init_for_test();
ensure_content_store();
crate::db::tests::ensure_shadow_db().await;
let story_id = "1222_test_backfill_unrecoverable";
// Write directly to the CRDT only — bypass write_item_with_content
// so no shadow-write message is ever sent, and never write
// in-memory content either, simulating a story whose SQLite copy
// was never captured (or was compacted away).
crate::crdt_state::write_item_str(
story_id,
"5_done",
Some("Unrecoverable Test"),
None,
None,
None,
);
let unrecoverable = backfill_evicted_story_content().await;
assert!(
unrecoverable.contains(&story_id.to_string()),
"story with no in-memory AND no SQLite content must be reported unrecoverable"
);
}
}
server/src/db/mod.rs
+11 -3
View File
@@ -37,8 +37,10 @@ pub use shadow_write::{check_schema_drift, get_shared_pool, init};
#[cfg(test)]
pub use content_store::ensure_content_store;
/// Shared test helpers for the `db` module, including [`tests::ensure_shadow_db`]
/// which `db::gc::tests` reuses for backfill tests (story 1222).
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
use std::fs;
@@ -597,10 +599,16 @@ mod tests {
/// Park the init on a leaked multi-thread runtime so the bg task lives for
/// the whole test process; mirrors `db::ops::tests::ensure_shadow_db`.
#[cfg(test)]
static SHADOW_RT: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
pub(crate) static SHADOW_RT: std::sync::OnceLock<tokio::runtime::Runtime> =
std::sync::OnceLock::new();
/// Shared test helper: initialise the shadow SQLite DB exactly once per
/// test binary, parked on a leaked multi-thread runtime so the
/// background write task survives past any single `#[tokio::test]`'s
/// per-test runtime teardown. Reused by `db::gc::tests` for backfill
/// tests (story 1222) — do not duplicate this dance elsewhere.
#[cfg(test)]
async fn ensure_shadow_db() {
pub(crate) async fn ensure_shadow_db() {
static INIT: std::sync::OnceLock<()> = std::sync::OnceLock::new();
if INIT.get().is_some() {
return;
server/src/http/mcp/status_tools.rs
+42
View File
@@ -382,6 +382,48 @@ mod tests {
assert_eq!(depends_on[1], 200);
}
/// Story 1222 regression: `show` must return full content for a story
/// after it reaches Done and again after it reaches Archived — the
/// terminal-stage content purge must never evict the story body, only
/// its ephemeral bookkeeping keys.
#[tokio::test]
async fn tool_show_returns_content_after_done_and_after_archived() {
let tmp = tempdir().unwrap();
crate::crdt_state::init_for_test();
crate::db::ensure_content_store();
let story_id = "1222_story_done_archived_test";
let story_content = "# Story\n\n## Acceptance Criteria\n\n- [ ] Ship it\n";
crate::db::write_item_with_content(
story_id,
"5_done",
story_content,
crate::db::ItemMeta::named("Done Archived Test"),
);
// Simulate the terminal-transition purge that fires when a story
// reaches Done (story 996 GC subscriber / sweep).
crate::db::gc::purge_ephemeral_content_keys_for_story(story_id);
let ctx = crate::http::context::AppContext::new_test(tmp.path().to_path_buf());
let result = tool_show(&json!({"story_id": story_id}), &ctx)
.await
.expect("show must succeed for a Done story, not return content unavailable");
let parsed: serde_json::Value = serde_json::from_str(&result).unwrap();
assert_eq!(parsed["story_id"], story_id);
assert_eq!(parsed["front_matter"]["name"], "Done Archived Test");
// Move to Archived and purge again — content must still be readable.
crate::db::move_item_stage(story_id, "6_archived", None);
crate::db::gc::purge_ephemeral_content_keys_for_story(story_id);
let result = tool_show(&json!({"story_id": story_id}), &ctx)
.await
.expect("show must succeed for an Archived story, not return content unavailable");
let parsed: serde_json::Value = serde_json::from_str(&result).unwrap();
assert_eq!(parsed["story_id"], story_id);
}
#[tokio::test]
async fn tool_show_returns_story_data() {
let tmp = tempdir().unwrap();
server/src/startup/tick_loop.rs
+11 -3
View File
@@ -588,6 +588,10 @@ pub(crate) async fn run_reconcile_pass(
// Content-GC: purge content-store entries for terminal/tombstoned stories.
crate::db::gc::sweep_zombie_content_on_startup();
// Content backfill: restore story bodies wiped by the pre-1222 buggy
// terminal-transition purge from the durable SQLite shadow column.
crate::db::gc::backfill_evicted_story_content().await;
// Worktree create: ensure every Coding story has a worktree.
crate::agents::pool::worktree_lifecycle::reconcile_worktree_create(root, agents.port()).await;
@@ -749,12 +753,16 @@ mod tests {
"run_reconcile_pass must not broadcast through the transition channel (no Lagged)"
);
// ── Assert: zombie content purged for all 200 Abandoned stories ────
// ── Assert: story content is RETAINED for all 200 Abandoned stories ──
// Story 1222: content must never be evicted on a pipeline transition
// (Done/Archived/Abandoned/etc.) — only genuinely deleted/tombstoned
// stories get their body purged. These stories are still live in the
// CRDT (merely Abandoned), so their content-store entry must survive.
for i in 0..200u32 {
let id = format!("1066_abandoned_{i:04}");
assert!(
crate::db::read_content(ContentKey::Story(&id)).is_none(),
"zombie content must be purged for abandoned story {id}"
crate::db::read_content(ContentKey::Story(&id)).is_some(),
"story content must be retained for abandoned story {id} (story 1222)"
);
}
}