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

huskies: merge 997

Browse Source
This commit is contained in:
dave
2026-05-14 11:01:06 +00:00
parent 0572af2193
commit c7a7cb4281
40 changed files with 256 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/crdt_state/ops.rs
+1 -9
View File
@@ -542,15 +542,7 @@ mod tests {
);
// Attempt resurrection via write_item — must be rejected by tombstone check.
write_item_str(
story_id,
"1_backlog",
Some("Resurrected"),
None,
None,
None,
None,
);
write_item_str(story_id, "1_backlog", Some("Resurrected"), None, None, None);
assert!(
read_item(story_id).is_none(),
"tombstoned story must not be resurrected by write_item after remote delete"
server/src/crdt_state/read.rs
+13 -7
View File
@@ -348,7 +348,7 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
JsonValue::String(s) if !s.is_empty() => s.parse::<crate::config::AgentName>().ok(),
_ => None,
};
let retry_count = match item.retry_count.view() {
let retry_count_register = match item.retry_count.view() {
JsonValue::Number(n) if n >= 0.0 => n as u32,
_ => 0u32,
};
@@ -411,6 +411,7 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
claim_agent.as_deref(),
claim_ts_secs,
plan_state_str.as_deref(),
retry_count_register,
)?;
Some(PipelineItemView {
@@ -418,7 +419,6 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
stage,
name,
agent,
retry_count,
depends_on,
qa_mode,
item_type,
@@ -439,6 +439,7 @@ pub(super) fn extract_item_view(item: &PipelineItemCrdt) -> Option<PipelineItemV
/// before the typed projection. This keeps remote ops from older nodes (and
/// raw-CRDT test inserts that bypass `migrate_legacy_stage_strings`) from
/// silently disappearing from the typed read path.
#[allow(clippy::too_many_arguments)]
fn project_stage_for_view(
stage_str: &str,
story_id: &str,
@@ -447,6 +448,7 @@ fn project_stage_for_view(
claim_agent: Option<&str>,
claim_ts_secs: Option<u64>,
plan_state_str: Option<&str>,
retries: u32,
) -> Option<crate::pipeline_state::Stage> {
use crate::pipeline_state::{
AgentClaim, AgentName, ArchiveReason, BranchName, GitSha, PlanState, Stage,
@@ -482,6 +484,7 @@ fn project_stage_for_view(
.unwrap_or(Stage::Coding {
claim: None,
plan: PlanState::Missing,
retries: 0,
}),
)
};
@@ -504,6 +507,7 @@ fn project_stage_for_view(
"coding" => Some(Stage::Coding {
claim,
plan: PlanState::from_str(plan_state_str.unwrap_or("")),
retries,
}),
"qa" => Some(Stage::Qa),
"blocked" => Some(Stage::Blocked {
@@ -513,6 +517,7 @@ fn project_stage_for_view(
feature_branch: BranchName(format!("feature/story-{story_id}")),
commits_ahead: NonZeroU32::new(1).expect("1 is non-zero"),
claim,
retries,
}),
"merge_failure" => {
// Story 986: read the typed kind directly from ContentKey::MergeFailureKind
@@ -675,7 +680,7 @@ mod tests {
let item_json: JsonValue = json!({
"story_id": "40_story_view",
"stage": "qa",
"stage": "2_current",
"name": "View Test",
"agent": "coder-1",
"retry_count": 2.0,
@@ -691,10 +696,13 @@ mod tests {
let view = extract_item_view(&crdt.doc.items[0]).unwrap();
assert_eq!(view.story_id, "40_story_view");
assert!(matches!(view.stage, crate::pipeline_state::Stage::Qa));
assert!(matches!(
view.stage,
crate::pipeline_state::Stage::Coding { .. }
));
assert_eq!(view.name, "View Test");
assert_eq!(view.agent.map(|a| a.as_str()), Some("coder-1"));
assert_eq!(view.retry_count, 2u32);
assert_eq!(view.retry_count(), 2u32);
assert_eq!(view.depends_on, vec![10u32, 20u32]);
}
@@ -749,7 +757,6 @@ mod tests {
None,
None,
None,
None,
);
// The story is live on this node.
@@ -817,7 +824,6 @@ mod tests {
None,
None,
None,
None,
);
assert!(
read_item(story_id).is_none(),
server/src/crdt_state/state/tests.rs
+1 -10
View File
@@ -116,7 +116,6 @@ async fn subscribe_receives_stage_transition_events() {
None,
None,
None,
None,
);
// Drain any stale events from concurrent tests until we see ours.
@@ -136,15 +135,7 @@ async fn subscribe_receives_stage_transition_events() {
));
// Update stage — emit_event fires again with the real from_stage.
write_item_str(
"906_story_subscribe",
"2_current",
None,
None,
None,
None,
None,
);
write_item_str("906_story_subscribe", "2_current", None, None, None, None);
let evt: CrdtEvent = loop {
let e = rx
server/src/crdt_state/types.rs
+11 -6
View File
@@ -190,8 +190,6 @@ pub struct WorkItem {
/// treated as malformed, not surfaced with an empty string.
pub(super) name: String,
pub(super) agent: Option<crate::config::AgentName>,
/// Retry counter — `0` when the CRDT register is unset.
pub(super) retry_count: u32,
/// Dependency story numbers — empty `Vec` when the register is unset.
pub(super) depends_on: Vec<u32>,
/// QA mode override. `None` means "use the project default".
@@ -226,9 +224,16 @@ impl WorkItem {
self.agent
}
/// Retry counter. Returns `0` when the register is unset.
/// Retry counter projected from the Stage variant.
///
/// Returns `retries` from `Stage::Coding` or `Stage::Merge`; `0` for all
/// other stages (QA, Backlog, Done, etc. carry no retry state).
pub fn retry_count(&self) -> u32 {
self.retry_count
match &self.stage {
crate::pipeline_state::Stage::Coding { retries, .. } => *retries,
crate::pipeline_state::Stage::Merge { retries, .. } => *retries,
_ => 0,
}
}
/// Dependency story numbers. Returns an empty slice when unset.
@@ -262,7 +267,6 @@ impl WorkItem {
stage: crate::pipeline_state::Stage,
name: impl Into<String>,
agent: Option<crate::config::AgentName>,
retry_count: u32,
depends_on: Vec<u32>,
qa_mode: Option<crate::io::story_metadata::QaMode>,
item_type: Option<crate::io::story_metadata::ItemType>,
@@ -273,7 +277,6 @@ impl WorkItem {
stage,
name: name.into(),
agent,
retry_count,
depends_on,
qa_mode,
item_type,
@@ -525,6 +528,7 @@ mod tests {
to_stage: crate::pipeline_state::Stage::Coding {
claim: None,
plan: crate::pipeline_state::PlanState::Missing,
retries: 0,
},
name: "Foo Feature".to_string(),
};
@@ -688,6 +692,7 @@ mod tests {
to_stage: Stage::Coding {
claim: None,
plan: crate::pipeline_state::PlanState::Missing,
retries: 0,
},
name: "Broadcast Test".to_string(),
};
server/src/crdt_state/write/item.rs
+82 -48
View File
@@ -245,12 +245,13 @@ pub fn set_plan_state(story_id: &str, state: crate::pipeline_state::PlanState) -
///
/// `stage` is the typed pipeline state; it is serialised to the canonical
/// clean wire form (story 934) via [`stage_dir_name`] at the CRDT boundary.
/// The `retries` count embedded in `Stage::Coding` / `Stage::Merge` is
/// automatically written to the `retry_count` CRDT register (story 997).
pub fn write_item(
story_id: &str,
stage: &Stage,
name: Option<&str>,
agent: Option<&str>,
retry_count: Option<i64>,
depends_on: Option<&str>,
merged_at: Option<f64>,
) {
@@ -260,6 +261,12 @@ pub fn write_item(
Stage::Merge { claim, .. } => claim.as_ref(),
_ => None,
};
// Extract retries from the Stage payload; non-Coding/Merge stages store 0.
let stage_retries: f64 = match stage {
Stage::Coding { retries, .. } => *retries as f64,
Stage::Merge { retries, .. } => *retries as f64,
_ => 0.0,
};
let Some(state_mutex) = get_crdt() else {
return;
};
@@ -307,11 +314,9 @@ pub fn write_item(
s.crdt.doc.items[idx].agent.set(a.to_string())
});
}
if let Some(rc) = retry_count {
apply_and_persist(&mut state, |s| {
s.crdt.doc.items[idx].retry_count.set(rc as f64)
});
}
apply_and_persist(&mut state, |s| {
s.crdt.doc.items[idx].retry_count.set(stage_retries)
});
if let Some(d) = depends_on {
apply_and_persist(&mut state, |s| {
s.crdt.doc.items[idx].depends_on.set(d.to_string())
@@ -365,7 +370,7 @@ pub fn write_item(
"stage": stage_str,
"name": name.unwrap_or(""),
"agent": agent.unwrap_or(""),
"retry_count": retry_count.unwrap_or(0) as f64,
"retry_count": stage_retries,
"depends_on": depends_on.unwrap_or(""),
"claim_agent": insert_claim_agent,
"claim_ts": insert_claim_ts,
@@ -429,7 +434,6 @@ pub fn write_item_str(
stage: &str,
name: Option<&str>,
agent: Option<&str>,
retry_count: Option<i64>,
depends_on: Option<&str>,
merged_at: Option<f64>,
) {
@@ -453,58 +457,88 @@ pub fn write_item_str(
crate::slog!("[crdt_state] write_item_str: unknown stage '{stage}' for {story_id}");
return;
};
write_item(
story_id,
&typed,
name,
agent,
retry_count,
depends_on,
merged_at,
);
write_item(story_id, &typed, name, agent, depends_on, merged_at);
}
/// Set `retry_count` to an explicit value for a pipeline item.
/// Set `retries` to an explicit value for a pipeline item via a Stage transition.
///
/// Pure metadata operation — the item's stage is not changed.
/// Call `set_retry_count(story_id, 0)` to reset the counter after a
/// stage transition or an explicit unblock.
/// Reads the current Stage from the CRDT, updates the `retries` field (only
/// meaningful for `Stage::Coding` and `Stage::Merge`), and writes back via
/// `write_item`. No-op for items not in a Coding or Merge stage.
pub fn set_retry_count(story_id: &str, count: i64) {
let Some(state_mutex) = get_crdt() else {
let Some(item) = super::super::read::read_item(story_id) else {
return;
};
let Ok(mut state) = state_mutex.lock() else {
return;
let new_stage = match item.stage().clone() {
Stage::Coding {
claim,
plan,
retries: _,
} => Stage::Coding {
claim,
plan,
retries: count.max(0) as u32,
},
Stage::Merge {
feature_branch,
commits_ahead,
claim,
retries: _,
} => Stage::Merge {
feature_branch,
commits_ahead,
claim,
retries: count.max(0) as u32,
},
_ => return,
};
if let Some(&idx) = state.index.get(story_id) {
apply_and_persist(&mut state, |s| {
s.crdt.doc.items[idx].retry_count.set(count as f64)
});
}
write_item(story_id, &new_stage, None, None, None, None);
}
/// Increment `retry_count` by 1 and return the new value.
/// Increment `retries` by 1 and return the new value.
///
/// Pure metadata operation — the item's stage is not changed.
/// Returns 0 if the item is not found in the CRDT (no-op in that case).
/// Use the returned value to decide whether the story should be blocked.
/// Reads the current Stage, increments the embedded `retries` field, and
/// writes back via `write_item`. Returns `0` if the item is not found or is
/// not in a Coding or Merge stage (no-op in that case).
pub fn bump_retry_count(story_id: &str) -> i64 {
let Some(state_mutex) = get_crdt() else {
let Some(item) = super::super::read::read_item(story_id) else {
return 0;
};
let Ok(mut state) = state_mutex.lock() else {
return 0;
let (new_stage, new_retries) = match item.stage().clone() {
Stage::Coding {
claim,
plan,
retries,
} => {
let n = retries + 1;
(
Stage::Coding {
claim,
plan,
retries: n,
},
n,
)
}
Stage::Merge {
feature_branch,
commits_ahead,
claim,
retries,
} => {
let n = retries + 1;
(
Stage::Merge {
feature_branch,
commits_ahead,
claim,
retries: n,
},
n,
)
}
_ => return 0,
};
let Some(&idx) = state.index.get(story_id) else {
return 0;
};
let current = match state.crdt.doc.items[idx].retry_count.view() {
JsonValue::Number(n) => n as i64,
_ => 0,
};
let new_count = current + 1;
apply_and_persist(&mut state, |s| {
s.crdt.doc.items[idx].retry_count.set(new_count as f64)
});
new_count
write_item(story_id, &new_stage, None, None, None, None);
new_retries as i64
}
server/src/crdt_state/write/migrations.rs
+3 -2
View File
@@ -349,7 +349,6 @@ mod stage_migration_tests {
None,
None,
None,
None,
);
// Then overwrite the stage register with the raw legacy string,
// bypassing `db::normalise_stage_str` / `write_item_str`'s mapping.
@@ -373,6 +372,7 @@ mod stage_migration_tests {
Stage::Coding {
claim: None,
plan: PlanState::Missing,
retries: 0,
},
),
(
@@ -390,6 +390,7 @@ mod stage_migration_tests {
feature_branch: BranchName(String::new()),
commits_ahead: NonZeroU32::new(1).unwrap(),
claim: None,
retries: 0,
},
),
(
@@ -458,12 +459,12 @@ mod stage_migration_tests {
&Stage::Coding {
claim: None,
plan: PlanState::Missing,
retries: 0,
},
Some("Already Clean"),
None,
None,
None,
None,
);
seed_with_raw_stage("9521_needs_migration", "2_current");
server/src/crdt_state/write/tests.rs
+13 -42
View File
@@ -97,7 +97,6 @@ fn migrate_story_ids_to_numeric_rewrites_slug_ids() {
None,
None,
None,
None,
);
let result = migrate_story_ids_to_numeric();
@@ -120,15 +119,7 @@ fn migrate_story_ids_to_numeric_rewrites_slug_ids() {
fn migrate_story_ids_to_numeric_is_idempotent() {
init_for_test();
write_item_str(
"43",
"1_backlog",
Some("Already Numeric"),
None,
None,
None,
None,
);
write_item_str("43", "1_backlog", Some("Already Numeric"), None, None, None);
// First call — nothing to migrate.
let r1 = migrate_story_ids_to_numeric();
@@ -154,17 +145,8 @@ fn migrate_story_ids_to_numeric_skips_conflict() {
None,
None,
None,
None,
);
write_item_str(
"44",
"2_current",
Some("Foo numeric"),
None,
None,
None,
None,
);
write_item_str("44", "2_current", Some("Foo numeric"), None, None, None);
let result = migrate_story_ids_to_numeric();
// The slug entry must NOT be migrated because "44" is already occupied.
@@ -195,7 +177,6 @@ fn migrate_story_ids_to_numeric_preserves_stage_and_name() {
Some("coder-1"),
None,
None,
None,
);
migrate_story_ids_to_numeric();
@@ -214,15 +195,7 @@ fn migrate_names_from_slugs_fills_empty_names() {
init_for_test();
// Write an item without a name.
write_item_str(
"42_story_my_feature",
"1_backlog",
None,
None,
None,
None,
None,
);
write_item_str("42_story_my_feature", "1_backlog", None, None, None, None);
// Before migration: nameless item is filtered by read_item (AC 5).
assert!(
@@ -251,7 +224,6 @@ fn migrate_names_from_slugs_leaves_existing_names_unchanged() {
None,
None,
None,
None,
);
migrate_names_from_slugs();
@@ -285,7 +257,6 @@ fn set_depends_on_round_trip_and_clear() {
None,
None,
None,
None,
);
// Set depends_on to [837] and verify CRDT register holds the list.
@@ -338,7 +309,6 @@ fn set_agent_some_writes_name() {
None,
None,
None,
None,
);
let found = set_agent(
@@ -366,7 +336,6 @@ fn set_agent_none_clears_register() {
Some("coder-2"),
None,
None,
None,
);
// Confirm agent is set.
@@ -412,7 +381,6 @@ fn set_qa_mode_round_trip_server_then_human() {
None,
None,
None,
None,
);
// Set qa=server via typed path and assert CRDT register reflects it.
@@ -465,7 +433,6 @@ fn set_qa_mode_round_trip_all_variants() {
None,
None,
None,
None,
);
for mode in [QaMode::Server, QaMode::Agent, QaMode::Human] {
@@ -501,7 +468,6 @@ fn bump_retry_count_increments_by_one() {
None,
None,
None,
None,
);
let v1 = bump_retry_count("9001_story_bump_test");
@@ -511,7 +477,11 @@ fn bump_retry_count_increments_by_one() {
assert_eq!(v2, 2, "second bump should return 2");
let item = read_item("9001_story_bump_test").expect("item must exist");
assert_eq!(item.retry_count, 2u32, "CRDT must reflect final bump value");
assert_eq!(
item.retry_count(),
2u32,
"CRDT must reflect final bump value"
);
}
#[test]
@@ -522,7 +492,6 @@ fn set_retry_count_resets_to_zero() {
"2_current",
Some("Set Test"),
None,
Some(5),
None,
None,
);
@@ -530,7 +499,11 @@ fn set_retry_count_resets_to_zero() {
set_retry_count("9002_story_set_test", 0);
let item = read_item("9002_story_set_test").expect("item must exist");
assert_eq!(item.retry_count, 0u32, "set_retry_count(0) must reset to 0");
assert_eq!(
item.retry_count(),
0u32,
"set_retry_count(0) must reset to 0"
);
}
#[test]
@@ -701,7 +674,6 @@ async fn tombstone_survives_concurrent_writes() {
None,
None,
None,
None,
);
assert!(
read_item(story_id).is_some(),
@@ -720,7 +692,6 @@ async fn tombstone_survives_concurrent_writes() {
None,
None,
None,
None,
);
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
}