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

fix: retain preparation diagnostics and source-owned cleanup

Browse Source
This commit is contained in:
Hare
2026-10-15 18:51:00 +09:00
parent cafa2fa6ea
commit 62d71ed17e
5 changed files with 58 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/job_notification_routing.rs
+1 -1
View File
@@ -192,7 +192,7 @@ pub(in crate::store) fn route(
let ack:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM notification_acknowledgements WHERE workspace_id=?1 AND delivery_id=?2) OR EXISTS(SELECT 1 FROM notification_job_legacy_receipts WHERE workspace_id=?1 AND delivery_id=?2 AND runtime_accepted=1)",params![workspace,source.delivery_id],|r|r.get(0))?;
tx.commit()?; return Ok(disposition=="deliver" && !ack);
}
if let JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)=facts {
if job.request.notification_target.is_some() && matches!(facts, JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)) {
// A transient checker read is not suppression or an immutable defer.
// With no decision, the existing pending scan retries the same source.
// Check existing decisions first: never replace pinned request evidence.
crates/workspace-server/src/job_notification_tests.rs
+13
View File
@@ -86,6 +86,14 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur
"no_recipient",
None,
),
(
"no-target-lookup-failure",
false,
JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed),
"suppress",
"no_recipient",
Some("source_lookup_failed"),
),
(
"clean",
true,
@@ -201,6 +209,11 @@ async fn job_notification_preparation_failures_remain_pending_and_append_only()
for _ in 0..2 {
assert!(!store.route_job_notification("w", &source, &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)).unwrap());
}
if !target {
assert_eq!(job_notice_evidence(&store,&source),("suppress".into(),"no_recipient".into(),false,Some("source_lookup_failed".into())));
assert!(!store.list_pending_job_notifications("w",100).unwrap().iter().any(|s|s.delivery_id==source.delivery_id));
continue;
}
store.with_conn(|conn| {
let sequence: i64 = conn.query_row("SELECT sequence FROM notification_sources WHERE input_event_id=?1", [&source.delivery_id], |r| r.get(0))?;
assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures WHERE source_sequence=?1 AND code='source_lookup_failed'", [sequence], |r| r.get::<_, i64>(0))?, 2);
crates/workspace-server/src/notification_diagnostics.rs
+25 -5
View File
@@ -33,7 +33,12 @@ fn source_checks(
});
}
if let Some(id) = &query.source_event_id {
let captured: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_sources s LEFT JOIN workspace_resource_keys k ON k.workspace_id=s.workspace_id AND k.resource_kind='ticket' AND k.resource_id=s.ticket_id WHERE s.workspace_id=?1 AND s.input_event_id=?2 AND (?3 IS NULL OR s.ticket_id=?3 OR k.resource_key=?3))",params![workspace,id,query.ticket],|r|r.get(0))?;
// Ticket event ordinals are Ticket-local; MR/result source identities are global within Workspace.
let scoped_ticket = id
.starts_with("ticket-event:")
.then_some(query.ticket.as_deref())
.flatten();
let captured: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_sources s LEFT JOIN workspace_resource_keys k ON k.workspace_id=s.workspace_id AND k.resource_kind='ticket' AND k.resource_id=s.ticket_id WHERE s.workspace_id=?1 AND s.input_event_id=?2 AND (?3 IS NULL OR s.ticket_id=?3 OR k.resource_key=?3))",params![workspace,id,scoped_ticket],|r|r.get(0))?;
let origin = if captured {
Some(true)
} else if let (Some(ticket), Some(index)) = (
@@ -46,6 +51,13 @@ fn source_checks(
id.strip_prefix("mr-event:").and_then(|v| v.split_once(':'))
{
Some(conn.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2 AND event_id=?3)",params![workspace,mr,event],|r|r.get(0))?)
} else if let Some(result) = id.strip_prefix("implementation-result:") {
Some(conn.query_row("SELECT EXISTS(SELECT 1 FROM ticket_implementation_results WHERE workspace_id=?1 AND result_id=?2)",params![workspace,result],|r|r.get(0))?)
} else if let Some((mr, sequence)) = id
.strip_prefix("mr-source:")
.and_then(|v| v.split_once(':'))
{
Some(conn.query_row("SELECT EXISTS(SELECT 1 FROM merge_request_source_observations WHERE workspace_id=?1 AND merge_request_id=?2 AND sequence=?3)",params![workspace,mr,sequence],|r|r.get(0))?)
} else {
None
};
@@ -160,12 +172,14 @@ pub(super) fn page_with_connection(
(a.delivery_id IS NOT NULL OR aa.delivery_id IS NOT NULL OR COALESCE(l.runtime_accepted,0)=1),
COALESCE(json_extract(a.target_json,'$.worker.runtime_id'),json_extract(aa.source_json,'$.runtime_id'),json_extract(r.target_json,'$.worker.runtime_id'),json_extract(r.target_json,'$.runtime_id'),json_extract(ar.reserved_source_json,'$.runtime_id'),json_extract(l.record_json,'$.target.runtime_id')),
COALESCE(json_extract(a.target_json,'$.worker.worker_id'),json_extract(aa.source_json,'$.worker_id'),json_extract(r.target_json,'$.worker.worker_id'),json_extract(r.target_json,'$.worker_id'),json_extract(ar.reserved_source_json,'$.worker_id'),json_extract(l.record_json,'$.target.worker_id')),
f.code,f.recorded_at,
COALESCE(f.code,pf.code),COALESCE(f.recorded_at,pf.recorded_at),
CASE WHEN s.role IN ('orchestrator','worker') AND s.assignment_id!='' THEN EXISTS(SELECT 1 FROM ticket_current_worker_assignments c WHERE c.workspace_id=s.workspace_id AND c.ticket_id=s.ticket_id AND c.role=s.role AND c.assignment_id=s.assignment_id) END,
EXISTS(SELECT 1 FROM notification_failures uf WHERE uf.workspace_id=s.workspace_id AND uf.delivery_id=d.delivery_id AND uf.code IN ('acceptance_unknown','acknowledgement_failed')),
(SELECT COUNT(*) FROM (SELECT DISTINCT at.runtime_id,at.worker_id FROM notification_assignment_targets at WHERE at.workspace_id=s.workspace_id AND at.delivery_id=d.delivery_id)),
l.delivery_id IS NOT NULL
l.delivery_id IS NOT NULL,
pf.sequence IS NOT NULL
{joins}
LEFT JOIN notification_source_preparation_failures pf ON pf.sequence=(SELECT MAX(p.sequence) FROM notification_source_preparation_failures p WHERE p.source_sequence=s.sequence)
LEFT JOIN notification_job_legacy_receipts l ON l.workspace_id=s.workspace_id AND l.delivery_id=d.delivery_id
LEFT JOIN notification_requests r ON r.workspace_id=s.workspace_id AND r.delivery_id=d.delivery_id
LEFT JOIN notification_acknowledgements a ON a.workspace_id=s.workspace_id AND a.delivery_id=d.delivery_id
@@ -186,6 +200,7 @@ pub(super) fn page_with_connection(
let recipient_current: Option<bool> = row.get(19)?;
let unresolved_send: bool = row.get(20)?;
let multiple_targets = row.get::<_, i64>(21)? > 1;
let source_failure: bool = row.get(23)?;
let acceptance_outcome_unknown = !accepted && (prepared || unresolved_send);
use server_api::NotificationTargetResolution as Target;
let target_resolution = if accepted {
@@ -214,7 +229,11 @@ pub(super) fn page_with_connection(
} else if recipient_current == Some(false) {
Stage::Historical
} else if disposition.is_none() {
Stage::AwaitingRouting
if source_failure {
Stage::DeliveryFailed
} else {
Stage::AwaitingRouting
}
} else if disposition.as_deref() == Some("suppress") {
Stage::Suppressed
} else if disposition.as_deref() == Some("defer") {
@@ -235,7 +254,8 @@ pub(super) fn page_with_connection(
let retry_candidate = !accepted
&& recipient_current != Some(false)
&& failure.as_deref() != Some("assignment_changed")
&& disposition.as_deref() == Some("deliver");
&& (disposition.as_deref() == Some("deliver")
|| (disposition.is_none() && source_failure));
items.push(NotificationDiagnostic {
source_id: format!("notification-source:{}", row.get::<_, i64>(0)?),
source_event_id: bounded(row.get(1)?).unwrap_or_default(),
crates/workspace-server/src/server_job_notification_tests.rs
+9
View File
@@ -163,6 +163,11 @@ async fn job_notification_checker_read_failure_reopen_recovers_same_source_to_re
let failed_once = failures();
assert_eq!(failed_once.len(), 1);
assert_eq!(failed_once[0].2, "source_lookup_failed");
let diagnostic=api.store.notification_diagnostics(TEST_WORKSPACE_ID,&server_api::NotificationDiagnosticsQuery{job_id:Some(request.job_id.clone()),..Default::default()}).unwrap().items.remove(0);
assert_eq!(diagnostic.stage,server_api::NotificationDiagnosticStage::DeliveryFailed);
assert!(diagnostic.disposition.is_none() && diagnostic.retry_candidate);
assert!(!diagnostic.acceptance_outcome_unknown);
assert_eq!(diagnostic.last_failure_code.as_deref(),Some("source_lookup_failed"));
assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_decisions WHERE source_sequence=?1", [original_source.0], |r| r.get::<_,i64>(0)).unwrap(), 0);
assert!(api.store.job_notification_request(TEST_WORKSPACE_ID, &source.delivery_id).unwrap().is_none());
assert!(delivery_receipts(&api, &target).is_empty());
@@ -189,6 +194,10 @@ async fn job_notification_checker_read_failure_reopen_recovers_same_source_to_re
assert!(execution.attempts.lock().unwrap().iter().any(|attempt| attempt.0 == source.delivery_id && attempt.1 == pinned.0));
assert_eq!(source_snapshot(), original_source);
assert_eq!(failures(), failed_once, "recovery must retain the original failure observation");
let diagnostic=reopened.store.notification_diagnostics(TEST_WORKSPACE_ID,&server_api::NotificationDiagnosticsQuery{job_id:Some(request.job_id.clone()),..Default::default()}).unwrap().items.remove(0);
assert_eq!(diagnostic.stage,server_api::NotificationDiagnosticStage::RuntimeAccepted);
assert_eq!(diagnostic.last_failure_code.as_deref(),Some("source_lookup_failed"));
assert!(!diagnostic.retry_candidate && !diagnostic.acceptance_outcome_unknown && !diagnostic.model_processing_confirmed);
let job = reopened.store.get_backend_job(TEST_WORKSPACE_ID, &request.job_id).unwrap().unwrap();
assert_eq!(job.state, BackendJobState::Completed);
assert_eq!(job.current_attempt, accepted.job.current_attempt);
crates/workspace-server/src/workspace_deletion.rs
+10
View File
@@ -491,6 +491,7 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
// commit; update guards remain installed throughout.
let mut ticket_delete_guards = Vec::new();
for (table, trigger) in [
("notification_source_preparation_failures", "notification_source_preparation_failures_no_delete"),
("notification_job_targets", "notification_job_targets_no_delete"),
("notification_job_legacy_receipts", "notification_job_legacy_no_delete"),
("notification_assignment_targets", "notification_assignment_targets_no_delete"),
@@ -521,6 +522,11 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
tx.execute_batch(&format!("DROP TRIGGER \"{trigger}\";"))?;
ticket_delete_guards.push(definition);
}
// Source-owned children have no Workspace column; purge them through
// the authenticated parent's scope before removing the source graph.
for table in ["notification_source_preparation_failures", "notification_source_offers"] {
tx.execute(&format!("DELETE FROM {table} WHERE source_sequence IN (SELECT sequence FROM notification_sources WHERE workspace_id=?1)"),[&workspace_id])?;
}
for table in WORKSPACE_DELETION_PURGE_TABLES {
tx.execute(
&format!("DELETE FROM \"{table}\" WHERE workspace_id = ?1"),
@@ -1113,6 +1119,8 @@ mod tests {
serde_json::to_string(&snapshot).unwrap()
],
)?;
conn.execute("INSERT INTO notification_source_preparation_failures(source_sequence,code) SELECT sequence,'source_lookup_failed' FROM notification_sources WHERE workspace_id=?1 LIMIT 1",[workspace])?;
conn.execute("INSERT INTO notification_source_offers(source_sequence,last_offered) SELECT sequence,1 FROM notification_sources WHERE workspace_id=?1 LIMIT 1",[workspace])?;
Ok(())
})
.unwrap();
@@ -1123,6 +1131,8 @@ mod tests {
assert!(conn.execute(&format!("DELETE FROM {table} WHERE workspace_id=?1"), [workspace]).is_err());
assert!(conn.execute(&format!("UPDATE {table} SET ticket_id=ticket_id WHERE workspace_id=?1"), [workspace]).is_err());
}
assert!(conn.execute("DELETE FROM notification_source_preparation_failures WHERE source_sequence IN (SELECT sequence FROM notification_sources WHERE workspace_id=?1)",[workspace]).is_err());
assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures p JOIN notification_sources s ON s.sequence=p.source_sequence WHERE s.workspace_id=?1",[workspace],|r|r.get::<_,i64>(0))?,1);
assert!(conn.execute("DELETE FROM typed_tickets WHERE workspace_id=?1", [workspace]).is_err(),
"ordinary Ticket deletion must not cascade away case evidence");
Ok(())