use super::*;
use crate::request_task::reserved::*;
use saddle_core::{DiagnosticOutcomeAxes, OperationOutcome, PhysicalDispositionFact};
use saddle_observability::root_diagnostic::{RootOutcomeFacts, RootRequestEvent};
use std::sync::atomic::{AtomicBool, Ordering};
static SIGNAL: AtomicBool = AtomicBool::new(false);
static POLLED: AtomicBool = AtomicBool::new(false);
static AT_HANDOFF: AtomicBool = AtomicBool::new(false);
static CHECK_BODY_STORAGE: AtomicBool = AtomicBool::new(false);
struct Cancel;
impl Future for Cancel {
type Output = ();
fn poll(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<()> {
if SIGNAL.load(Ordering::SeqCst) {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
#[derive(Clone, Copy, Debug, PartialEq)]
enum Case {
Serial,
Cancel,
Deadline,
Panic,
ReadyStop,
ReadyCleanup,
Abort,
HandoffAbort,
Preparation,
MultiSuccess,
MultiFailure,
MultiCancel,
}
struct Body(
Case,
std::cell::Cell<usize>,
Option<ReservedRequestView>,
usize,
bool,
);
impl Future for Body {
type Output = u32;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
let address = self.as_ref().get_ref() as *const Self as usize;
if self.1.get() == 0 {
self.1.set(address);
} else {
assert_eq!(self.1.get(), address);
}
POLLED.store(true, Ordering::SeqCst);
match self.0 {
Case::Panic => panic!("RG_SCOPE_PRIMARY"),
Case::Cancel => {
SIGNAL.store(true, Ordering::SeqCst);
cx.waker().wake_by_ref();
Poll::Pending
}
Case::Deadline | Case::Abort => Poll::Pending,
Case::ReadyStop => {
SIGNAL.store(true, Ordering::SeqCst);
Poll::Ready(41)
}
_ => Poll::Ready(41),
}
}
}
impl Drop for Body {
fn drop(&mut self) {
if let Some(view) = self.2.as_ref() {
assert!(
view.test_available_storage() < self.3,
"body reservation still held inside F::drop"
);
println!("RG_SEND_DROP {:?} pinned reservation_held", self.0);
}
if self.1.get() != 0 {
assert_eq!(
self.1.get(),
self as *const Self as usize,
"pinned until Drop"
);
}
if (self.0 == Case::Preparation && self.4) || matches!(
self.0,
Case::Panic | Case::ReadyCleanup | Case::HandoffAbort
) {
panic!("RG_SCOPE_CLEANUP")
}
}
}
struct Input {
serial: Option<ProfuseGwSerialScope<Cancel>>,
terminal: Option<ProfuseGwPostDatabaseRequestTerminal>,
saved_value: Option<u32>,
physical: saddle_core::DbPhysicalProcessCapability,
observer: Observer,
output: saddle_observability::EmergencyDiagnosticHandle,
case: Case,
preparation_cleanup: bool,
}
type Owner = crate::profusegw::ReservedDispatchOwner<Input>;
fn facts(operation: OperationOutcome) -> RootOutcomeFacts {
RootOutcomeFacts {
axes: DiagnosticOutcomeAxes {
operation,
physical: PhysicalDispositionFact::Unknown,
..Default::default()
},
..Default::default()
}
}
fn body(
owner: &mut Owner,
context: ReservedTaskContext,
) -> impl Future<Output = Result<u32, ReservedRequestFailure>> + Send + '_ {
async move {
owner.1.serial = Some(
owner
.0
.take()
.unwrap()
.into_database_request()
.into_serial_scope(Cancel),
);
if owner.1.case == Case::Deadline {
owner
.1
.serial
.as_mut()
.unwrap()
.lease
.deadline
.timer
.as_mut()
.reset(tokio::time::Instant::now() + Duration::from_millis(2));
}
if matches!(
owner.1.case,
Case::MultiSuccess | Case::MultiFailure | Case::MultiCancel
) {
return multistep(owner, &context).await;
}
let rounds = if matches!(owner.1.case, Case::Serial | Case::Preparation | Case::Panic) {
2
} else {
1
};
for round in 0..rounds {
let running_case =
if round == 0 && matches!(owner.1.case, Case::Panic | Case::Preparation) {
Case::Serial
} else {
owner.1.case
};
let driver = owner.1.serial.as_mut().unwrap().driver();
let (supervisor, observation) = driver.prepare_reserved(&context).ok().unwrap();
let stage = observation
.start_database_stage(&owner.1.observer, Some(&owner.1.output))
.unwrap();
assert!(
observation
.start_database_stage(&owner.1.observer, None)
.is_err()
);
let (other_supervisor, other_observation) =
driver.prepare_reserved(&context).ok().unwrap();
if matches!(owner.1.case, Case::Serial | Case::Preparation | Case::Panic) && round == 0
{
let view = observation.diagnostic_view();
let error =
std::io::Error::new(std::io::ErrorKind::PermissionDenied, "RG_DB_FIRST_SCOPE");
let failure = view.source_error(
&error,
Some(&owner.1.output),
saddle_core::DiagnosticStage::RequestDb,
RootRequestEvent::Database,
facts(OperationOutcome::Failed),
);
let failure = other_observation
.retain_database_failure(failure)
.err()
.expect("same scope, wrong invocation must return source");
assert!(observation.retain_database_failure(failure).is_ok());
}
let other_stage = other_observation
.start_database_stage(&owner.1.observer, None)
.unwrap();
let other = other_supervisor
.supervise(None, std::future::ready(7))
.await;
assert_eq!(other.result.ok(), Some(7));
let (stage, foreign) = stage
.finish(other.completion, facts(OperationOutcome::Succeeded))
.err()
.unwrap();
assert_eq!(
other_stage
.finish(foreign, facts(OperationOutcome::Succeeded))
.ok(),
Some(saddle_observability::DiagnosticSubmission::OutputUnavailable)
);
drop(other_observation);
let pressure = (owner.1.case == Case::Preparation && round == 1)
.then(|| observation.exhaust_test_storage());
let available = context.test_storage_available();
let check_view = CHECK_BODY_STORAGE
.load(Ordering::SeqCst)
.then(|| observation.diagnostic_view());
if running_case == Case::Preparation && round == 1 {
POLLED.store(false, Ordering::SeqCst);
}
let outcome = supervisor
.supervise(
Some(&owner.1.output),
Body(running_case, std::cell::Cell::new(0), check_view, available, owner.1.preparation_cleanup),
)
.await;
if CHECK_BODY_STORAGE.load(Ordering::SeqCst) {
assert_eq!(
context.test_storage_available(),
available,
"body refund after normal/panic teardown"
);
}
drop(pressure);
let supervision = outcome.completion.supervision();
match running_case {
Case::Serial | Case::ReadyStop | Case::ReadyCleanup | Case::HandoffAbort => {
assert_eq!(outcome.result.as_ref().ok(), Some(&41))
}
Case::Preparation if round == 1 => {
assert!(matches!(
&outcome.result,
Err(ProfuseGwReservedScopeFailure::Preparation(
ReservedContextError::Storage(saddle_admission::AdmissionError::FrameworkReserveExceeded { .. })
))
));
assert!(!POLLED.load(Ordering::SeqCst));
assert!(observation.primary().is_some(), "preparation keeps its original receipt");
},
Case::Preparation => assert_eq!(outcome.result.as_ref().ok(), Some(&41)),
Case::Panic => assert!(supervision.panicked),
Case::Cancel => assert_eq!(supervision.stop, Some(ProfuseGwScopeStop::Cancelled)),
Case::Deadline => assert_eq!(supervision.stop, Some(ProfuseGwScopeStop::TimedOut)),
Case::Abort => unreachable!("parent aborts the Pending body"),
Case::MultiSuccess | Case::MultiFailure | Case::MultiCancel => unreachable!(),
}
if running_case == Case::ReadyStop {
assert_eq!(supervision.stop, Some(ProfuseGwScopeStop::Cancelled));
}
if matches!(
running_case,
Case::Panic | Case::ReadyCleanup | Case::HandoffAbort
) {
assert!(supervision.cleanup_failed);
}
drop(driver);
let (physical, execution) = owner.1.serial.take().unwrap().into_physical_finalization();
let receipt = owner
.1
.physical
.connection_returned(execution, outcome.result)
.ok()
.unwrap();
let mut suspended = physical.complete(receipt).ok().unwrap();
if running_case == Case::HandoffAbort {
let (value, terminal) = suspended.into_response_parts().ok().unwrap();
owner.1.saved_value = value.ok();
owner.1.terminal = Some(terminal);
AT_HANDOFF.store(true, Ordering::SeqCst);
std::future::pending::<()>().await;
unreachable!("parent aborts this controlled handoff");
}
let operation = match running_case {
Case::Serial | Case::Preparation if round == 0 => OperationOutcome::Failed,
Case::Panic => OperationOutcome::Panicked,
Case::Cancel => OperationOutcome::Cancelled,
Case::Deadline => OperationOutcome::TimedOut,
Case::Preparation => OperationOutcome::Failed,
_ => OperationOutcome::Succeeded,
};
let mut stage_facts = facts(operation);
stage_facts.axes.physical = PhysicalDispositionFact::Returned;
assert!(stage.finish(outcome.completion, stage_facts).is_ok());
assert!(
observation
.start_database_stage(&owner.1.observer, None)
.is_err()
);
drop(observation);
if round + 1 < rounds {
let (next, value) = suspended.resume().await.unwrap();
assert_eq!(value.ok(), Some(41));
owner.1.serial = Some(next);
} else {
let (_, terminal) = suspended.into_response_parts().ok().unwrap();
owner.1.terminal = Some(terminal);
}
}
Ok(41)
}
}
fn factory<'a>(
owner: &'a mut Owner,
context: ReservedTaskContext,
) -> ReservedBorrowedFuture<'a, u32> {
Box::pin(body(owner, context))
}
async fn multistep(
owner: &mut Owner,
context: &ReservedTaskContext,
) -> Result<u32, ReservedRequestFailure> {
use saddle_core::request_context::RegisteredContextOperation as Op;
let code = saddle_core::DiagnosticCode::new("rg2.original").unwrap();
let rounds = if owner.1.case == Case::MultiSuccess {
24
} else {
1
};
let baseline = context.test_storage_available();
for round in 0..rounds {
let driver = owner.1.serial.as_mut().unwrap().driver();
let (begin, begin_obs) = driver.prepare_reserved(context).ok().unwrap();
let begin_stage = begin_obs
.start_database_stage(&owner.1.observer, None)
.unwrap();
let begun = begin
.supervise(None, async {
driver.checkpoint().await.unwrap();
1u32
})
.await;
assert_eq!(begun.result.ok(), Some(1));
let (supervisor, observation) = driver.prepare_reserved(context).ok().unwrap();
let stage = observation
.start_database_stage(&owner.1.observer, Some(&owner.1.output))
.unwrap();
let foreign = begin_obs
.operation(Op::checked("db.dynamic.write").unwrap())
.unwrap();
let pressure = observation.exhaust_test_storage();
assert!(
observation
.operation(Op::checked("db.no_capacity").unwrap())
.is_err()
);
drop(pressure); let mut first_source = None;
let result = supervisor
.supervise(Some(&owner.1.output), async {
let first = observation
.operation(Op::checked("db.dynamic.read").unwrap())
.unwrap();
driver.checkpoint().await.unwrap();
let from_first = 41u32;
drop(first); tokio::task::yield_now().await; let operation = if from_first == 41 {
"db.dynamic.write"
} else {
"db.dynamic.other"
};
let step = observation
.operation(Op::checked(operation).unwrap())
.unwrap();
driver.checkpoint().await.unwrap();
if owner.1.case == Case::MultiFailure {
let failure = step.test_original(
&std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"RG2_DYNAMIC_WRITE",
),
Some(&owner.1.output),
);
let failure = foreign
.retain_failure(failure)
.err()
.expect("foreign invocation");
first_source = Some(step.retain_failure(failure).ok().unwrap());
let cleanup_step = observation
.operation(Op::checked("db.dynamic.cleanup").unwrap())
.unwrap();
let failure = cleanup_step.test_original(
&std::io::Error::other("RG2_SECOND_ORIGINAL"),
Some(&owner.1.output),
);
let original_id = failure.occurrence();
let pressure = observation.exhaust_test_storage();
let failure = cleanup_step
.retain_failure(failure)
.err()
.expect("new receipt node requires storage");
drop(pressure);
assert_eq!(
serde_json::to_value(failure.occurrence()).unwrap(),
serde_json::to_value(original_id).unwrap()
);
assert!(cleanup_step.retain_failure(failure).is_ok());
}
if owner.1.case == Case::MultiCancel {
SIGNAL.store(true, Ordering::SeqCst);
std::future::poll_fn(|cx| {
cx.waker().wake_by_ref();
Poll::<()>::Pending
})
.await;
}
from_first
})
.await;
drop(foreign);
if owner.1.case == Case::MultiCancel {
assert_eq!(
result.completion.supervision().stop,
Some(ProfuseGwScopeStop::Cancelled)
);
} else {
assert_eq!(result.result.as_ref().ok(), Some(&41));
}
let (end, end_obs) = driver.prepare_reserved(context).ok().unwrap();
let end_stage = end_obs
.start_database_stage(&owner.1.observer, Some(&owner.1.output))
.unwrap();
let ended = end
.supervise(Some(&owner.1.output), async {
let op = end_obs
.operation(
Op::checked(if first_source.is_some() {
"db.rollback"
} else {
"db.commit"
})
.unwrap(),
)
.unwrap();
driver.checkpoint().await.unwrap();
if let Some(primary) = first_source {
let diagnostic = crate::diagnostics::failure(
saddle_core::DiagnosticStage::RequestDb,
saddle_core::DiagnosticCategory::ExpectedRejection,
"rg2.rollback",
)
.during_cleanup_of_occurrence(&primary);
op.source_cleanup_existing_error(
&std::io::Error::other("RG2_ROLLBACK_ORIGINAL"),
diagnostic,
code,
Some(&owner.1.output),
facts(OperationOutcome::Failed),
)
.ok()
.unwrap();
}
})
.await;
drop(driver);
let (physical, execution) = owner.1.serial.take().unwrap().into_physical_finalization();
let receipt = owner
.1
.physical
.connection_returned(execution, 41u32)
.ok()
.unwrap();
let mut suspended = physical.complete(receipt).ok().unwrap();
let mut stage_facts = facts(OperationOutcome::Succeeded);
stage_facts.axes.physical = PhysicalDispositionFact::Returned;
assert!(begin_stage.finish(begun.completion, stage_facts).is_ok());
let mut body_facts = stage_facts;
body_facts.axes.operation = match owner.1.case {
Case::MultiFailure => OperationOutcome::Failed,
Case::MultiCancel => OperationOutcome::Cancelled,
_ => OperationOutcome::Succeeded,
};
assert!(stage.finish(result.completion, body_facts).is_ok());
assert!(end_stage.finish(ended.completion, body_facts).is_ok());
if owner.1.case == Case::MultiSuccess {
assert_eq!(
context.test_scope_nodes(),
0,
"no successful history in task"
);
let held = context.test_storage_available();
drop(begin_obs);
drop(observation);
drop(end_obs);
assert!(
context.test_storage_available() > held,
"last handles refund"
);
} else {
drop(begin_obs);
drop(observation);
drop(end_obs);
}
if round + 1 < rounds {
let (next, value) = suspended.resume().await.unwrap();
assert_eq!(value, 41);
owner.1.serial = Some(next);
assert_eq!(
context.test_storage_available(),
baseline,
"success storage does not accumulate round={round}"
);
} else {
let (_, terminal) = suspended.into_response_parts().ok().unwrap();
owner.1.terminal = Some(terminal);
}
}
println!(
"RG2_MULTISTEP {:?} rounds={rounds} retired_success_and_originals",
owner.1.case
);
Ok(41)
}
fn layout<I, R>(_: impl FnOnce(I) -> R) -> std::alloc::Layout {
std::alloc::Layout::new::<R>()
}
#[test]
fn db_shaped_supervision_handoff() {
run_cases(
&[
Case::Serial,
Case::Cancel,
Case::Deadline,
Case::Panic,
Case::ReadyStop,
Case::ReadyCleanup,
Case::Abort,
Case::HandoffAbort,
Case::Preparation,
],
0,
);
}
#[test]
fn multistep_invocation_and_retirement() {
ReservedTaskContext::test_scope_layouts();
run_cases(
&[Case::MultiSuccess, Case::MultiFailure, Case::MultiCancel],
1,
);
}
#[test]
fn send_body_lifecycle() {
use std::alloc::Layout;
let demand = saddle_admission::StorageDemand::embedded(
Layout::new::<Pin<Box<Body>>>(),
Layout::new::<()>(),
&[(Layout::new::<Body>(), 1)],
)
.unwrap();
assert_eq!(
demand.bytes(),
Layout::new::<ReservedScopeHeap<Body>>().size() + Layout::new::<Body>().size()
);
println!(
"RG_SEND_LAYOUT body={} heap={} owner={} demand={} arc_control=0",
Layout::new::<Body>().size(),
Layout::new::<ReservedScopeHeap<Body>>().size(),
Layout::new::<ReservedScopeOwned<Body>>().size(),
demand.bytes()
);
run_cases(
&[Case::Serial, Case::Abort, Case::Panic, Case::ReadyCleanup],
2,
);
}
#[test]
fn preparation_failure_original_and_refund() {
run_cases(&[Case::Preparation], 3);
run_cases(&[Case::Preparation], 4);
}
fn run_cases(cases: &[Case], multistep: u8) {
CHECK_BODY_STORAGE.store(multistep == 2, Ordering::SeqCst);
let path = std::env::var_os("RG_COMPOSITION_EVIDENCE")
.map(std::path::PathBuf::from)
.unwrap_or_else(|| std::env::temp_dir().join(format!("rg-compose-{}", std::process::id())));
std::fs::create_dir_all(&path).unwrap();
let output = saddle_observability::EmergencyDiagnostics::start(
&saddle_observability::FileLoggingConfig::new(
path.clone(),
saddle_observability::Rotation::Daily,
),
)
.unwrap();
let old_hook = std::panic::take_hook();
std::panic::set_hook(Box::new(|info| {
let _ = crate::diagnostics::capture_current_panic(info);
}));
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
.block_on(async {
for &case in cases {
SIGNAL.store(false, Ordering::SeqCst);
POLLED.store(false, Ordering::SeqCst);
AT_HANDOFF.store(false, Ordering::SeqCst);
let shared = Arc::new(SharedRuntimeProcess {
process: Mutex::new(Some(ProfuseGwRuntimeProcess::new(
crate::request_task::reserved::tests::process(),
))),
});
let lease = ProfuseGwProcessLease {
shared: shared.clone(),
};
let physical = lease
.take_database_startup_half()
.unwrap()
.into_process_capability();
let observer = Observer::with_writer(
saddle_observability::ObserverConfig::default(),
std::io::sink(),
)
.unwrap();
let input = Input {
serial: None,
terminal: None,
saved_value: None,
physical,
observer: observer.clone(),
output: output.handle(),
case,
preparation_cleanup: multistep == 4,
};
let body_layout = layout(
|(owner, context): (&'static mut Owner, ReservedTaskContext)| {
body(owner, context)
},
);
let admission = lease.try_reserved_dispatch(
saddle_core::ContextLabel::checked("rg-composition").unwrap(),
Some(output.handle()),
body_layout,
&[(std::alloc::Layout::new::<Cancel>(), 1)],
input,
factory,
);
let ReservedDispatchOutcome::Ready {
root,
future,
mut ticket,
..
} = admission
else {
panic!("isolated consumer fit case={case:?} body={body_layout:?}")
};
let mut tasks = tokio::task::JoinSet::new();
let handle = tasks.spawn(future);
ticket.bind(handle.id()).unwrap();
if case == Case::Abort {
while !POLLED.load(Ordering::SeqCst) {
tokio::task::yield_now().await;
}
handle.abort();
}
if case == Case::HandoffAbort {
while !AT_HANDOFF.load(Ordering::SeqCst) {
tokio::task::yield_now().await;
}
handle.abort();
}
let mut joined = ticket
.complete(tasks.join_next().await.unwrap())
.ok()
.unwrap();
if let Some(scope) = joined.owner_mut().1.serial.take() {
let (completion, execution) = scope.into_physical_finalization();
let receipt = joined
.owner_mut()
.1
.physical
.connection_discarded(execution, ())
.ok()
.unwrap();
let (_, terminal) = completion
.complete(receipt)
.ok()
.unwrap()
.into_response_parts()
.ok()
.unwrap();
joined.owner_mut().1.terminal = Some(terminal);
}
finish_profusegw_after_database(joined.owner_mut().1.terminal.take().unwrap());
drop(root);
let mut terminal_facts = facts(OperationOutcome::Succeeded);
terminal_facts.axes.physical = if case == Case::Abort {
PhysicalDispositionFact::Discarded
} else {
PhysicalDispositionFact::Returned
};
let recovered = joined.recover(terminal_facts).ok().unwrap();
if !matches!(case, Case::Abort | Case::HandoffAbort) {
assert_eq!(recovered.result.as_ref().unwrap().as_ref().ok(), Some(&41));
}
if case == Case::HandoffAbort {
assert_eq!(recovered.owner.1.saved_value, Some(41));
assert!(recovered.cleanup.is_some());
}
if case == Case::MultiSuccess {
assert!(recovered.primary.is_none() && recovered.cleanup.is_none());
} else {
assert!(recovered.primary.is_some() || recovered.cleanup.is_some());
}
let expected_stages = if case == Case::MultiSuccess {
72
} else if matches!(case, Case::MultiFailure | Case::MultiCancel) {
3
} else if matches!(case, Case::Serial | Case::Preparation | Case::Panic) {
4
} else {
2
};
assert_eq!(
observer
.metrics_snapshot()
.stage_latency(saddle_observability::Stage::Database)
.iter()
.sum::<u64>(),
expected_stages
);
drop(recovered);
match lease.try_admit() {
ProfuseGwCoordinatorAdmissionOutcome::Ready(d, _) => d.cancel(),
_ => panic!("same process healthy/zero {case:?}"),
}
drop(lease);
shared
.process
.lock()
.unwrap()
.take()
.unwrap()
.finish()
.unwrap();
println!("RG_COMPOSITION case={case:?} owner_returned ZERO");
}
});
std::panic::set_hook(old_hook);
let exit = crate::diagnostics::close_output(
output,
Some(std::time::Instant::now() + Duration::from_secs(3)),
);
assert_eq!(exit.snapshot.enqueued, exit.snapshot.written);
assert_eq!(exit.snapshot.dropped, 0);
let records: Vec<serde_json::Value> =
std::fs::read_to_string(path.join("saddle.emergency.log"))
.unwrap()
.lines()
.map(|s| serde_json::from_str(s).unwrap())
.collect();
let sources: Vec<_> = records
.iter()
.filter(|r| r["event"] == "request_error_original" && r["channel"] == "context")
.map(|r| {
(
r["occurrence"].clone(),
serde_json::from_str::<serde_json::Value>(r["payload"].as_str().unwrap()).unwrap(),
)
})
.collect();
assert!(!sources.is_empty());
if multistep >= 3 {
let originals: Vec<_> = records.iter().filter(|r| {
r["channel"] == "debug" && r["payload"].as_str().is_some_and(|s| s.contains("FrameworkReserveExceeded"))
}).collect();
assert!(!originals.is_empty(), "original Storage Debug must be readable");
for original in originals {
assert!(original["payload"].as_str().unwrap().contains("requested:"));
assert!(original["payload"].as_str().unwrap().contains("available:"));
}
if multistep == 4 {
assert!(sources.iter().any(|(occurrence, _)| occurrence["primary_diagnostic_id"].as_u64().is_some()),
"pre-poll destructor panic remains secondary to original storage failure");
}
}
for (occurrence, source) in &sources {
assert!(
source["facts"]["origin_line"]
.as_u64()
.or_else(|| source["facts"]["origin"]["line"].as_u64())
.unwrap()
> 0
);
let finals: Vec<_> = records
.iter()
.filter(|r| {
r["event"] == "request_failure_boundary"
&& r["stage"] == "finalization"
&& r["occurrence"] == *occurrence
})
.collect();
assert_eq!(
finals.len(),
1,
"one final projection for each original {occurrence}"
);
assert_eq!(finals[0]["source_context"], source["context"]);
let stages: Vec<_> = records
.iter()
.filter(|r| {
r["event"] == "request_failure_boundary"
&& r["stage"] == "database"
&& r["occurrence"] == *occurrence
})
.collect();
assert!(stages.len() <= 1);
for stage in stages {
assert_eq!(stage["source_context"], source["context"]);
}
if let Some(primary) = occurrence["primary_diagnostic_id"].as_u64() {
let original = sources
.iter()
.find(|(o, _)| o["diagnostic_id"] == primary)
.unwrap();
assert_eq!(
source["context"]["scope"], original.1["context"]["scope"],
"cleanup must not link previous scope"
);
assert_eq!(finals[0]["axes"]["axes"]["cleanup"], "failed");
}
}
if multistep == 1 {
let operations = sources
.iter()
.map(|(_, s)| s["context"]["db_operation"].clone())
.collect::<Vec<_>>();
println!("RG2_OPERATION_SNAPSHOTS {operations:?}");
for expected in ["db.dynamic.write", "db.dynamic.cleanup", "db.rollback"] {
assert_eq!(
operations.iter().filter(|v| v["value"] == expected).count(),
1,
"original operation snapshot {expected}"
);
}
assert!(
sources.len() >= 3,
"body original, rollback original and cancellation"
);
} else if multistep == 0 {
let successful_second = records
.iter()
.filter(|r| {
r["event"] == "request_stage_finished"
&& r["context"]["scope"]["value"]["transaction_scope"] == 1
&& r["axes"]["axes"]["operation"] == "succeeded"
})
.collect::<Vec<_>>();
assert_eq!(successful_second.len(), 1); assert!(successful_second.iter().all(|r| r["occurrence"].is_null()));
let ready_cleanup = records
.iter()
.find(|r| {
r["event"] == "request_failure_boundary"
&& r["stage"] == "database"
&& r["axes"]["axes"]["operation"] == "succeeded"
&& r["axes"]["axes"]["cleanup"] == "failed"
})
.unwrap();
assert_eq!(ready_cleanup["axes"]["axes"]["physical"], "returned");
let incomplete = records
.iter()
.find(|r| {
r["event"] == "request_stage_finished"
&& r["axes"]["axes"]["operation"] == "cancelled"
})
.unwrap();
assert_eq!(
incomplete["axes"]["axes"]["physical"], "unknown",
"dropped stage is not a physical receipt"
);
}
for (name, layout) in supervised_scope_layouts() {
println!(
"RG_COMPOSITION_LAYOUT {name} size={} align={}",
layout.size(),
layout.align()
);
}
println!("RG_COMPOSITION PASS physical=SIMULATED HTTP_DB=NOT_RUN");
}