Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: assign unique event refs to merge integration comments

Browse Source
This commit is contained in:
Hare
2026-10-27 06:15:48 +09:00
parent af719e387d
commit 623c31615d
3 changed files with 184 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/merge-request/src/lib.rs
+17 -1
View File
@@ -1747,7 +1747,23 @@ fn ticket_event(
) -> Result<(), MergeRequestError> {
let ticket = &mr.ticket_ids[0];
let n:i64=t.query_row("SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2",params![mr.workspace_id,ticket],|r|r.get(0))?;
t.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body)VALUES(?1,?2,?3,'comment',?4,?5,NULL,NULL,'Merge Request integrated',?6)",params![mr.workspace_id,ticket,n,format!("worker:{}:{}",e.merged_by.runtime_id,e.merged_by.worker_id),e.created_at.to_rfc3339(),format!("Merge Request `{}` integrated approved source ref `{}`; Ticket remains in progress until guarded Ticket completion.",mr.merge_request_id,e.approved_source_ref)])?;
t.execute(
"INSERT INTO typed_ticket_events(
workspace_id,ticket_id,event_index,kind,author,at,from_state,to_state,heading,body,event_ref
) VALUES(?1,?2,?3,'comment',?4,?5,NULL,NULL,'Merge Request integrated',?6,?7)",
params![
mr.workspace_id,
ticket,
n,
format!("worker:{}:{}", e.merged_by.runtime_id, e.merged_by.worker_id),
e.created_at.to_rfc3339(),
format!(
"Merge Request `{}` integrated approved source ref `{}`; Ticket remains in progress until guarded Ticket completion.",
mr.merge_request_id, e.approved_source_ref
),
format!("{ticket}:{n}"),
],
)?;
for (k, v) in [
("implementation_assignment_id", a),
("merge_request_id", &mr.merge_request_id),
crates/merge-request/tests/store.rs
+116 -5
View File
@@ -53,8 +53,8 @@ fn fixture() -> (tempfile::TempDir, MergeRequestStore) {
let d = tempfile::tempdir().unwrap();
let p = d.path().join("db");
let c = Connection::open(&p).unwrap();
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,title TEXT NOT NULL,body TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_targets(workspace_id TEXT,ticket_id TEXT,ordinal INTEGER,repository_key TEXT,ref_selector TEXT,access TEXT);CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t','Title','Requirements');CREATE VIEW ticket_active_worker_assignments AS SELECT current.* FROM ticket_current_worker_assignments current JOIN typed_tickets ticket ON ticket.workspace_id=current.workspace_id AND ticket.ticket_id=current.ticket_id WHERE ticket.workflow_state NOT IN ('done','closed');").unwrap();
c.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body) VALUES('W','T',-1,'comment','Requirements')",[]).unwrap();
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,title TEXT NOT NULL,body TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_targets(workspace_id TEXT,ticket_id TEXT,ordinal INTEGER,repository_key TEXT,ref_selector TEXT,access TEXT);CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,event_ref TEXT NOT NULL DEFAULT '',PRIMARY KEY(workspace_id,ticket_id,event_index),UNIQUE(workspace_id,ticket_id,event_ref));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO repositories VALUES('W','R2');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t','Title','Requirements');CREATE VIEW ticket_active_worker_assignments AS SELECT current.* FROM ticket_current_worker_assignments current JOIN typed_tickets ticket ON ticket.workspace_id=current.workspace_id AND ticket.ticket_id=current.ticket_id WHERE ticket.workflow_state NOT IN ('done','closed');").unwrap();
c.execute("INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref) VALUES('W','T',-1,'comment','Requirements','T:-1')",[]).unwrap();
drop(c);
let a = Assignments(Arc::new(Mutex::new(CurrentAssignment {
assignment_id: "A".into(),
@@ -457,6 +457,18 @@ fn integration_preserves_selectors_ticket_state_and_assignment_and_replays_its_m
.unwrap();
assert_eq!(replayed, merged);
let connection = Connection::open(d.path().join("db")).unwrap();
let ticket_refs = connection
.prepare("SELECT event_ref FROM typed_ticket_events WHERE event_index >= 0")
.unwrap()
.query_map([], |row| row.get::<_, String>(0))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
ticket_refs,
vec!["T:0"],
"replay must not append another post"
);
let (state, active): (String, bool) = connection.query_row(
"SELECT workflow_state, EXISTS(SELECT 1 FROM ticket_active_worker_assignments WHERE workspace_id='W' AND ticket_id='T') FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
[],
@@ -477,6 +489,87 @@ fn integration_preserves_selectors_ticket_state_and_assignment_and_replays_its_m
assert!(!json.contains(banned), "{banned} in {json}")
}
}
#[test]
fn integration_post_failure_rolls_back_and_same_operation_recovers_with_legacy_empty_ref() {
let (dir, store) = fixture();
open(&store);
let review = approve(&store, "source", "approved");
request(&store, "source", "pending");
let completion = CompleteMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
operation_id: "recover-integration".into(),
approval_event_id: review.event_id,
current_subject_ref: "source".into(),
target_ref_before: "before".into(),
target_ref_after: "after".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth(),
now: at(5),
};
let connection = Connection::open(dir.path().join("db")).unwrap();
// Retain a comment written before integration populated event_ref.
connection.execute_batch(
"INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,kind,body,event_ref)
VALUES('W','T',0,'comment','Legacy integration','');
CREATE TRIGGER reject_integration_post BEFORE INSERT ON typed_ticket_events
WHEN NEW.heading='Merge Request integrated'
BEGIN SELECT RAISE(ABORT,'injected integration post failure'); END;",
).unwrap();
let before = store.get_by_id("W", "MR").unwrap();
let error = store.complete(completion.clone()).unwrap_err();
assert!(
error
.to_string()
.contains("injected integration post failure")
);
let failed = store.get_by_id("W", "MR").unwrap();
assert_eq!(failed.state, MergeRequestState::Open);
assert_eq!(failed.thread, before.thread);
assert!(store.authorize_review_submission("MR", "pending").is_ok());
assert_eq!(
connection
.query_row(
"SELECT COUNT(*) FROM typed_ticket_event_attributes",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
0
);
connection
.execute_batch("DROP TRIGGER reject_integration_post")
.unwrap();
let merged = store.complete(completion.clone()).unwrap();
assert_eq!(store.complete(completion).unwrap(), merged);
assert_eq!(
store.get_by_id("W", "MR").unwrap().state,
MergeRequestState::Merged
);
assert!(store.authorize_review_submission("MR", "pending").is_err());
let posts = connection
.prepare("SELECT event_ref,body FROM typed_ticket_events WHERE event_index >= 0 ORDER BY event_index")
.unwrap()
.query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)))
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
posts.len(),
2,
"retry and replay must append exactly one post"
);
assert_eq!(posts[0], (String::new(), "Legacy integration".into()));
assert_eq!(posts[1].0, "T:1");
assert!(posts[1].1.contains("`MR`"));
assert_eq!(connection.query_row(
"SELECT value FROM typed_ticket_event_attributes WHERE event_index=1 AND key='operation_id'",
[], |row| row.get::<_, String>(0),
).unwrap(), "recover-integration");
}
#[test]
fn source_move_cancels_submission_and_old_approval_is_reusable_when_source_returns() {
let (_d, s) = fixture();
@@ -955,7 +1048,7 @@ fn ticket_rescope_requires_fresh_review_before_merge_and_remains_recoverable() {
let connection = Connection::open(dir.path().join("db")).unwrap();
connection
.execute_batch(
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL)",
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL,'T:1')",
)
.unwrap();
connection
@@ -1082,7 +1175,7 @@ fn ticket_rescope_after_integration_accepts_fresh_requirement_attestation() {
let connection = Connection::open(dir.path().join("db")).unwrap();
connection
.execute_batch(
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL)",
"UPDATE typed_tickets SET title='Rescoped' WHERE workspace_id='W' AND ticket_id='T'; INSERT INTO typed_ticket_events VALUES('W','T',1,'item_edit','user','t2',NULL,NULL,NULL,NULL,'T:1')",
)
.unwrap();
connection
@@ -1307,6 +1400,24 @@ fn partial_and_full_integration_retain_ticket_state_and_active_assignment() {
assert_eq!(second_request.merged_result().unwrap(), &second_merge);
assert_eq!(first_request.integration_approval().unwrap(), &first);
assert_eq!(second_request.integration_approval().unwrap(), &second);
let posts = connection
.prepare(
"SELECT event.event_ref, attribute.value
FROM typed_ticket_events event
JOIN typed_ticket_event_attributes attribute USING(workspace_id,ticket_id,event_index)
WHERE attribute.key='merge_request_id' ORDER BY event.event_index",
)
.unwrap()
.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})
.unwrap()
.collect::<Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
posts,
vec![("T:0".into(), "MR".into()), ("T:1".into(), "MR-2".into())]
);
}
#[test]
@@ -1324,7 +1435,7 @@ fn historical_ticket_completion_payload_remains_readable_and_unchanged_on_migrat
}"#;
connection.execute_batch(
"UPDATE typed_tickets SET workflow_state='done' WHERE workspace_id='W' AND ticket_id='T';
INSERT INTO typed_ticket_events VALUES('W','T',0,'state_changed','worker:runtime:coder','2026-07-26T12:00:07Z','inprogress','done','Ticket requirements completed','historical body');"
INSERT INTO typed_ticket_events VALUES('W','T',0,'state_changed','worker:runtime:coder','2026-07-26T12:00:07Z','inprogress','done','Ticket requirements completed','historical body','T:0');"
).unwrap();
connection
.execute(
crates/workspace-server/src/server/tests/ticket_evidence_tests.rs
+51
View File
@@ -18,6 +18,57 @@ const EVIDENCE_MR: &str = "ticket-evidence-mr";
const EVIDENCE_SOURCE: &str = "immutable-approved-source";
const EVIDENCE_TARGET: &str = "immutable-target-result";
#[tokio::test]
async fn integration_preserves_legacy_empty_post_ref_and_exposes_new_ref_with_server_schema() {
let fixture = EvidenceApiFixture::new().await;
let approval = fixture.approve("integration-post-ref");
let connection = Connection::open(&fixture.api.config.database_path).unwrap();
let legacy_index: i64 = connection.query_row(
"SELECT COALESCE(MAX(event_index),-1)+1 FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2",
params![TEST_WORKSPACE_ID, fixture.ticket_id], |row| row.get(0),
).unwrap();
connection.execute(
"INSERT INTO typed_ticket_events(workspace_id,ticket_id,event_index,event_ref,kind,body)
VALUES(?1,?2,?3,'','comment','Legacy integration')",
params![TEST_WORKSPACE_ID, fixture.ticket_id, legacy_index],
).unwrap();
let merged = fixture.integrate(&approval);
assert_eq!(fixture.integrate(&approval), merged);
let detail = fixture
.backend
.show(fixture.ticket_id.clone().into())
.unwrap();
let legacy = detail
.events
.iter()
.find(|event| event.body.as_str() == "Legacy integration")
.unwrap();
assert_eq!(legacy.event_ref, "");
let integrations = detail
.events
.iter()
.filter(|event| {
event.attributes.get("merge_request_id").map(String::as_str) == Some(EVIDENCE_MR)
})
.collect::<Vec<_>>();
assert_eq!(
integrations.len(),
1,
"completion replay must not duplicate the Ticket post"
);
assert_eq!(
integrations[0].event_ref,
format!("{}:{}", fixture.ticket_id, legacy_index + 1)
);
assert_eq!(
integrations[0].attributes["operation_id"],
"evidence-integration"
);
let shown = fixture.show().await;
assert_eq!(shown.merge_requests[0].state, "merged");
}
#[tokio::test]
async fn ticket_browse_does_not_observe_sources_but_authoritative_query_and_show_still_do() {
struct CountingObserver(Arc<std::sync::atomic::AtomicUsize>);