huskies: merge 1243 refactor Merge jobs stop bloating replicated CRDT state
This commit is contained in:
@@ -299,10 +299,17 @@ impl AgentPool {
|
|||||||
crate::db::write_content(crate::db::ContentKey::MergeSuccess(&sid), "1");
|
crate::db::write_content(crate::db::ContentKey::MergeSuccess(&sid), "1");
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update CRDT with terminal status.
|
// Update CRDT with terminal status. The full untruncated output is
|
||||||
|
// already on disk (write_merge_report, called from
|
||||||
|
// run_merge_pipeline for the Ok(r) case above, or below for the
|
||||||
|
// Err(e) case) — only a bounded summary plus that pointer goes
|
||||||
|
// into the replicated `merge_jobs.error` field, so a large gate
|
||||||
|
// failure doesn't bloat every node's CRDT state.
|
||||||
match &report {
|
match &report {
|
||||||
Ok(r) => {
|
Ok(r) => {
|
||||||
let report_json = serde_json::to_string(r).unwrap_or_else(|_| String::new());
|
let bounded = crate::service::merge::bound_report_for_storage(r);
|
||||||
|
let report_json =
|
||||||
|
serde_json::to_string(&bounded).unwrap_or_else(|_| String::new());
|
||||||
crate::crdt_state::write_merge_job(
|
crate::crdt_state::write_merge_job(
|
||||||
&sid,
|
&sid,
|
||||||
"completed",
|
"completed",
|
||||||
@@ -312,12 +319,15 @@ impl AgentPool {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
|
let report_path = crate::service::merge::io::write_merge_report(&root, &sid, e);
|
||||||
|
let bounded =
|
||||||
|
crate::service::merge::bound_plain_error(e, report_path.as_deref());
|
||||||
crate::crdt_state::write_merge_job(
|
crate::crdt_state::write_merge_job(
|
||||||
&sid,
|
&sid,
|
||||||
"failed",
|
"failed",
|
||||||
started_at,
|
started_at,
|
||||||
Some(finished_at),
|
Some(finished_at),
|
||||||
Some(e),
|
Some(&bounded),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -16,7 +16,10 @@ pub mod summary;
|
|||||||
#[allow(unused_imports)]
|
#[allow(unused_imports)]
|
||||||
pub use status::format_merge_status_message;
|
pub use status::format_merge_status_message;
|
||||||
#[allow(unused_imports)]
|
#[allow(unused_imports)]
|
||||||
pub use summary::{summarize_merge_failure_kind, summarize_merge_result};
|
pub use summary::{
|
||||||
|
bound_plain_error, bound_report_for_storage, summarize_merge_failure_kind,
|
||||||
|
summarize_merge_result,
|
||||||
|
};
|
||||||
|
|
||||||
// ── Error type ────────────────────────────────────────────────────────────────
|
// ── Error type ────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
|||||||
@@ -113,6 +113,78 @@ pub fn summarize_merge_failure_kind(kind: &MergeFailureKind) -> String {
|
|||||||
summarize(failing_gate_label_for_kind(kind), &kind.to_gate_output())
|
summarize(failing_gate_label_for_kind(kind), &kind.to_gate_output())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Return a copy of `result` with its embedded output string (`gate_output`
|
||||||
|
/// on [`MergeResult::Success`], `output` on every other variant) bounded to
|
||||||
|
/// [`SUMMARY_TAIL_LINES`].
|
||||||
|
///
|
||||||
|
/// Used before storing a [`MergeResult`] in the replicated CRDT `merge_jobs`
|
||||||
|
/// collection: the full untruncated text already lives on disk (see
|
||||||
|
/// `service::merge::io::write_merge_report` and `MergeReport::report_path`),
|
||||||
|
/// so only a bounded tail needs to travel through CRDT sync.
|
||||||
|
pub fn bound_result_for_storage(result: &MergeResult) -> MergeResult {
|
||||||
|
match result {
|
||||||
|
MergeResult::Success {
|
||||||
|
conflicts_resolved,
|
||||||
|
conflict_details,
|
||||||
|
gate_output,
|
||||||
|
} => MergeResult::Success {
|
||||||
|
conflicts_resolved: *conflicts_resolved,
|
||||||
|
conflict_details: conflict_details.clone(),
|
||||||
|
gate_output: truncate_gate_output(gate_output, SUMMARY_TAIL_LINES),
|
||||||
|
},
|
||||||
|
MergeResult::Conflict { details, output } => MergeResult::Conflict {
|
||||||
|
details: details.clone(),
|
||||||
|
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
|
||||||
|
},
|
||||||
|
MergeResult::GateFailure {
|
||||||
|
output,
|
||||||
|
failure_kind,
|
||||||
|
} => MergeResult::GateFailure {
|
||||||
|
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
|
||||||
|
failure_kind: failure_kind.clone(),
|
||||||
|
},
|
||||||
|
MergeResult::NoCommits { output } => MergeResult::NoCommits {
|
||||||
|
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
|
||||||
|
},
|
||||||
|
MergeResult::Other {
|
||||||
|
output,
|
||||||
|
conflict_details,
|
||||||
|
} => MergeResult::Other {
|
||||||
|
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
|
||||||
|
conflict_details: conflict_details.clone(),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Return a copy of `report` whose embedded [`MergeResult`] output is bounded
|
||||||
|
/// via [`bound_result_for_storage`]; `report_path` (the pointer to the full
|
||||||
|
/// untruncated text on disk) and every other field are carried over as-is.
|
||||||
|
pub fn bound_report_for_storage(
|
||||||
|
report: &crate::agents::merge::MergeReport,
|
||||||
|
) -> crate::agents::merge::MergeReport {
|
||||||
|
crate::agents::merge::MergeReport {
|
||||||
|
story_id: report.story_id.clone(),
|
||||||
|
result: bound_result_for_storage(&report.result),
|
||||||
|
worktree_cleaned_up: report.worktree_cleaned_up,
|
||||||
|
story_archived: report.story_archived,
|
||||||
|
report_path: report.report_path.clone(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Bound a plain (non-[`MergeResult`]) error string for CRDT storage,
|
||||||
|
/// appending a pointer to the full text on disk when `report_path` is given.
|
||||||
|
///
|
||||||
|
/// Used for the hard-error path (git/process failures that short-circuit
|
||||||
|
/// before a [`MergeResult`] is ever constructed), so the CRDT `merge_jobs`
|
||||||
|
/// error field stays bounded the same way completed-job entries do.
|
||||||
|
pub fn bound_plain_error(error: &str, report_path: Option<&str>) -> String {
|
||||||
|
let truncated = truncate_gate_output(error, SUMMARY_TAIL_LINES);
|
||||||
|
match report_path {
|
||||||
|
Some(path) => format!("{truncated}\n\nFull output: {path}"),
|
||||||
|
None => truncated,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@@ -236,4 +308,118 @@ mod tests {
|
|||||||
let summary = summarize_merge_failure_kind(&kind);
|
let summary = summarize_merge_failure_kind(&kind);
|
||||||
assert!(!summary.contains("Quality gates failed:"));
|
assert!(!summary.contains("Quality gates failed:"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ── bound_result_for_storage ────────────────────────────────────────────────
|
||||||
|
|
||||||
|
fn long_output(n: usize) -> String {
|
||||||
|
(1..=n)
|
||||||
|
.map(|i| format!("line{i}"))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join("\n")
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_result_for_storage_truncates_gate_failure_output() {
|
||||||
|
let result = MergeResult::GateFailure {
|
||||||
|
output: long_output(100),
|
||||||
|
failure_kind: Some(GateFailureKind::Test),
|
||||||
|
};
|
||||||
|
let bounded = bound_result_for_storage(&result);
|
||||||
|
assert!(bounded.output().len() < result.output().len());
|
||||||
|
assert!(bounded.output().contains("line100"));
|
||||||
|
assert!(matches!(
|
||||||
|
bounded,
|
||||||
|
MergeResult::GateFailure {
|
||||||
|
failure_kind: Some(GateFailureKind::Test),
|
||||||
|
..
|
||||||
|
}
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_result_for_storage_truncates_success_gate_output() {
|
||||||
|
let result = MergeResult::Success {
|
||||||
|
conflicts_resolved: true,
|
||||||
|
conflict_details: Some("resolved automatically".to_string()),
|
||||||
|
gate_output: long_output(100),
|
||||||
|
};
|
||||||
|
let bounded = bound_result_for_storage(&result);
|
||||||
|
assert!(bounded.output().len() < result.output().len());
|
||||||
|
assert!(bounded.output().contains("line100"));
|
||||||
|
match bounded {
|
||||||
|
MergeResult::Success {
|
||||||
|
conflicts_resolved,
|
||||||
|
conflict_details,
|
||||||
|
..
|
||||||
|
} => {
|
||||||
|
assert!(conflicts_resolved);
|
||||||
|
assert_eq!(conflict_details.as_deref(), Some("resolved automatically"));
|
||||||
|
}
|
||||||
|
other => panic!("expected Success, got {other:?}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_result_for_storage_leaves_short_output_unchanged() {
|
||||||
|
let result = MergeResult::NoCommits {
|
||||||
|
output: "no commits to merge".to_string(),
|
||||||
|
};
|
||||||
|
let bounded = bound_result_for_storage(&result);
|
||||||
|
assert_eq!(bounded.output(), "no commits to merge");
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── bound_report_for_storage ────────────────────────────────────────────────
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_report_for_storage_truncates_output_and_keeps_report_path() {
|
||||||
|
let report = crate::agents::merge::MergeReport {
|
||||||
|
story_id: "42_story".to_string(),
|
||||||
|
result: MergeResult::GateFailure {
|
||||||
|
output: long_output(100),
|
||||||
|
failure_kind: Some(GateFailureKind::Test),
|
||||||
|
},
|
||||||
|
worktree_cleaned_up: false,
|
||||||
|
story_archived: false,
|
||||||
|
report_path: Some(".huskies/merge_reports/42_story-123.log".to_string()),
|
||||||
|
};
|
||||||
|
let bounded = bound_report_for_storage(&report);
|
||||||
|
assert!(bounded.result.output().len() < report.result.output().len());
|
||||||
|
assert_eq!(
|
||||||
|
bounded.report_path.as_deref(),
|
||||||
|
report.report_path.as_deref()
|
||||||
|
);
|
||||||
|
assert_eq!(bounded.story_id, report.story_id);
|
||||||
|
|
||||||
|
// Round-trips through the same JSON shape `get_merge_status` expects.
|
||||||
|
let json = serde_json::to_string(&bounded).unwrap();
|
||||||
|
let decoded: crate::agents::merge::MergeReport = serde_json::from_str(&json).unwrap();
|
||||||
|
assert!(decoded.result.output().contains("line100"));
|
||||||
|
assert_eq!(
|
||||||
|
decoded.report_path.as_deref(),
|
||||||
|
Some(".huskies/merge_reports/42_story-123.log")
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── bound_plain_error ────────────────────────────────────────────────────────
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_plain_error_appends_pointer_when_path_given() {
|
||||||
|
let bounded = bound_plain_error("boom", Some(".huskies/merge_reports/1_story-1.log"));
|
||||||
|
assert!(bounded.contains("boom"));
|
||||||
|
assert!(bounded.contains("Full output: .huskies/merge_reports/1_story-1.log"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_plain_error_no_pointer_when_no_path() {
|
||||||
|
let bounded = bound_plain_error("boom", None);
|
||||||
|
assert_eq!(bounded, "boom");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bound_plain_error_truncates_long_error() {
|
||||||
|
let error = long_output(100);
|
||||||
|
let bounded = bound_plain_error(&error, None);
|
||||||
|
assert!(bounded.len() < error.len());
|
||||||
|
assert!(bounded.contains("line100"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user