use std::collections::{HashMap, HashSet};
use std::path::Path;
use serde::Deserialize;
use super::ranked_report::{
MAX_RECORDS_PER_COLLECTION, RankedFunction, RankedProfileDocument, RankedProfileMetadata,
RankedSemantic, fold_inclusive_counts,
};
use super::report_cli::{RankedReportFailure, RankedReportFailurePhase};
use super::report_health::{CAPTURE_HEALTH_SQL, capture_health_input_sql};
use super::report_trace_processor::run_trace_processor_query;
use super::report_trace_sanitizer::sanitize_trace;
const RECORD_HEADER: &[u8] = b"\"record_hex\"";
const MAX_RECORD_HEX_CHARS: usize = 64 * 1024;
const SAMPLE_CORRELATION_SQL: &str = include_str!("sql/sample_correlation.sql");
const RANKED_PROFILE_BASE_SQL: &str = include_str!("sql/ranked_profile_base.sql");
const RANKED_REPORT_SQL: &str = include_str!("sql/ranked_report.sql");
#[derive(Deserialize)]
#[serde(
tag = "record_kind",
content = "record",
rename_all = "snake_case",
deny_unknown_fields
)]
enum CompactRecord {
Metadata(CompactMetadata),
Semantic(Box<RankedSemantic>),
Frame(CompactFrame),
FunctionSelf(CompactFunctionSelf),
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CompactMetadata {
capture_complete: bool,
semantic_complete: bool,
finalization_observed: bool,
incomplete_operation_root_count: i64,
truncation_marker_count: i64,
missing_identity_field_count: i64,
missing_terminal_result_count: i64,
crossing_worker_slice_count: i64,
crossing_planning_activity_slice_count: i64,
crossing_execution_activity_slice_count: i64,
invalid_planning_activity_hierarchy_count: i64,
invalid_execution_activity_hierarchy_count: i64,
perf_sample_without_callsite_count: i64,
perf_samples_skipped: i64,
buffer_loss_count: i64,
data_source_loss_count: i64,
flush_failure_count: i64,
schema_version: u32,
sample_frequency_hz: u32,
sampled_cpu_count: u32,
exact_time_unit: String,
sample_unit: String,
eligible_sample_count: i64,
direct_sample_count: i64,
ambiguous_sample_count: i64,
unattributed_sample_count: i64,
resolved_function_sample_count: i64,
unresolved_function_sample_count: i64,
unwind_error_sample_count: i64,
missing_callstack_sample_count: i64,
trace_profiler_dropped_sample_count: i64,
audit_error_count: i64,
}
#[derive(Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct CompactFrame {
function_id: i64,
parent_function_id: Option<i64>,
name: String,
module_name: Option<String>,
source_file: Option<String>,
line_number: Option<i64>,
official_self_sample_count: i64,
official_inclusive_sample_count: i64,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct CompactFunctionSelf {
semantic_id: i64,
function_id: i64,
self_sample_count: i64,
}
pub(super) fn load_ranked_profile(
input: &Path,
capture_scope_id: Option<u64>,
) -> Result<RankedProfileDocument, RankedReportFailure> {
let health_input_sql = capture_health_input_sql(input)?;
let selection_sql = report_selection_sql(capture_scope_id);
let sql = [
health_input_sql.as_str(),
selection_sql.as_str(),
"CREATE PERFETTO TABLE delta_funnel_capture_health AS",
CAPTURE_HEALTH_SQL,
SAMPLE_CORRELATION_SQL,
RANKED_PROFILE_BASE_SQL,
RANKED_REPORT_SQL,
]
.join("\n");
let sanitized = sanitize_trace(input)?;
let output = run_trace_processor_query(sanitized.path(), sql.as_bytes())?;
parse_ranked_report_output(&output)
}
fn report_selection_sql(capture_scope_id: Option<u64>) -> String {
capture_scope_id.map_or_else(
|| {
"CREATE PERFETTO TABLE delta_funnel_report_selection AS\nSELECT NULL AS capture_scope_id;"
.to_owned()
},
|capture_scope_id| {
format!(
"CREATE PERFETTO TABLE delta_funnel_report_selection AS\nSELECT {capture_scope_id} AS capture_scope_id;"
)
},
)
}
fn parse_ranked_report_output(output: &[u8]) -> Result<RankedProfileDocument, RankedReportFailure> {
let mut lines = output
.split(|byte| *byte == b'\n')
.map(|line| line.strip_suffix(b"\r").unwrap_or(line))
.filter(|line| !line.is_empty());
if lines.next() != Some(RECORD_HEADER) {
return Err(malformed_result());
}
let mut metadata = None;
let mut semantics = Vec::new();
let mut frames = Vec::new();
let mut function_self = Vec::new();
for line in lines {
let hex = line
.strip_prefix(b"\"")
.and_then(|line| line.strip_suffix(b"\""))
.ok_or_else(malformed_result)?;
if hex.len() > MAX_RECORD_HEX_CHARS {
return Err(result_too_large());
}
let record = serde_json::from_slice::<CompactRecord>(&decode_hex(hex)?)
.map_err(|_| malformed_result())?;
match record {
CompactRecord::Metadata(record) => {
if metadata.replace(record).is_some() {
return Err(malformed_result());
}
}
CompactRecord::Semantic(record) => push_bounded(&mut semantics, *record)?,
CompactRecord::Frame(record) => push_bounded(&mut frames, record)?,
CompactRecord::FunctionSelf(record) => push_bounded(&mut function_self, record)?,
}
}
build_document(
metadata.ok_or_else(malformed_result)?,
semantics,
frames,
function_self,
)
}
fn build_document(
metadata: CompactMetadata,
mut semantics: Vec<RankedSemantic>,
frames: Vec<CompactFrame>,
function_self: Vec<CompactFunctionSelf>,
) -> Result<RankedProfileDocument, RankedReportFailure> {
if metadata.missing_identity_field_count > 0 {
return Err(aggregate_failure(
"unsafe_identity",
"ranked profile ownership identity is incomplete",
));
}
if semantics.is_empty() {
return Err(RankedReportFailure::new(
RankedReportFailurePhase::Health,
"incomplete_capture",
"input trace contains no retained Delta Funnel operation",
));
}
if metadata.audit_error_count != 0 {
return Err(aggregate_failure(
"audit_failed",
"Trace Processor aggregate audit failed",
));
}
if [
metadata.crossing_worker_slice_count,
metadata.crossing_planning_activity_slice_count,
metadata.crossing_execution_activity_slice_count,
metadata.invalid_planning_activity_hierarchy_count,
metadata.invalid_execution_activity_hierarchy_count,
]
.into_iter()
.any(|count| count > 0)
{
return Err(aggregate_failure(
"invalid_hierarchy",
"ranked profile semantic hierarchy is invalid",
));
}
semantics.sort_by_key(|semantic| (semantic.operation_id, semantic.semantic_id));
let mut document = RankedProfileDocument {
metadata: RankedProfileMetadata {
capture_complete: metadata.capture_complete,
semantic_complete: metadata.semantic_complete,
finalization_observed: metadata.finalization_observed,
incomplete_operation_root_count: metadata.incomplete_operation_root_count,
truncation_marker_count: metadata.truncation_marker_count,
missing_identity_field_count: metadata.missing_identity_field_count,
missing_terminal_result_count: metadata.missing_terminal_result_count,
crossing_worker_slice_count: metadata.crossing_worker_slice_count,
crossing_planning_activity_slice_count: metadata.crossing_planning_activity_slice_count,
crossing_execution_activity_slice_count: metadata
.crossing_execution_activity_slice_count,
invalid_planning_activity_hierarchy_count: metadata
.invalid_planning_activity_hierarchy_count,
invalid_execution_activity_hierarchy_count: metadata
.invalid_execution_activity_hierarchy_count,
perf_sample_without_callsite_count: metadata.perf_sample_without_callsite_count,
perf_samples_skipped: metadata.perf_samples_skipped,
buffer_loss_count: metadata.buffer_loss_count,
data_source_loss_count: metadata.data_source_loss_count,
flush_failure_count: metadata.flush_failure_count,
schema_version: metadata.schema_version,
sample_frequency_hz: metadata.sample_frequency_hz,
sampled_cpu_count: metadata.sampled_cpu_count,
exact_time_unit: metadata.exact_time_unit,
sample_unit: metadata.sample_unit,
eligible_sample_count: metadata.eligible_sample_count,
direct_sample_count: metadata.direct_sample_count,
ambiguous_sample_count: metadata.ambiguous_sample_count,
unattributed_sample_count: metadata.unattributed_sample_count,
resolved_function_sample_count: metadata.resolved_function_sample_count,
unresolved_function_sample_count: metadata.unresolved_function_sample_count,
unwind_error_sample_count: metadata.unwind_error_sample_count,
missing_callstack_sample_count: metadata.missing_callstack_sample_count,
trace_profiler_dropped_sample_count: metadata.trace_profiler_dropped_sample_count,
},
semantics,
functions: Vec::new(),
};
document
.validate_structure()
.map_err(|_| aggregate_failure("invalid_structure", "semantic hierarchy is invalid"))?;
fold_semantics(&mut document.semantics)?;
let frames = validate_frames(frames)?;
document.functions = expand_functions(&frames, function_self)?;
document
.functions
.sort_by_key(|function| (function.semantic_id, function.function_id));
document
.validate_structure()
.map_err(|_| aggregate_failure("invalid_structure", "function hierarchy is invalid"))?;
fold_functions(&mut document.functions)?;
reconcile_official_self_counts(&frames, &document.functions)?;
document.normalize_source_metadata();
document.validate().map_err(|_| {
aggregate_failure(
"invalid_aggregate",
"ranked profile aggregate validation failed",
)
})?;
Ok(document)
}
fn fold_semantics(semantics: &mut [RankedSemantic]) -> Result<(), RankedReportFailure> {
let parents = semantics
.iter()
.map(|semantic| (semantic.semantic_id, semantic.parent_semantic_id))
.collect::<HashMap<_, _>>();
let direct = semantics
.iter()
.map(|semantic| (semantic.semantic_id, semantic.direct_sample_count))
.collect::<HashMap<_, _>>();
let inclusive = fold_inclusive_counts(&parents, &direct)
.ok_or_else(|| aggregate_failure("count_overflow", "semantic sample fold failed"))?;
for semantic in semantics {
semantic.inclusive_sample_count = *inclusive
.get(&semantic.semantic_id)
.ok_or_else(|| aggregate_failure("invalid_structure", "semantic fold is incomplete"))?;
}
Ok(())
}
fn validate_frames(
frames: Vec<CompactFrame>,
) -> Result<HashMap<i64, CompactFrame>, RankedReportFailure> {
let mut by_id = HashMap::with_capacity(frames.len());
for frame in frames {
if frame.function_id < 0
|| frame.official_self_sample_count < 0
|| frame.official_inclusive_sample_count < 0
|| by_id.insert(frame.function_id, frame).is_some()
{
return Err(aggregate_failure(
"invalid_function_graph",
"native function graph is invalid",
));
}
}
let parents = by_id
.values()
.map(|frame| (frame.function_id, frame.parent_function_id))
.collect::<HashMap<_, _>>();
let self_counts = by_id
.values()
.map(|frame| (frame.function_id, frame.official_self_sample_count))
.collect::<HashMap<_, _>>();
let inclusive = fold_inclusive_counts(&parents, &self_counts).ok_or_else(|| {
aggregate_failure(
"invalid_function_graph",
"native function graph could not be folded",
)
})?;
for frame in by_id.values() {
if inclusive.get(&frame.function_id) != Some(&frame.official_inclusive_sample_count) {
return Err(aggregate_failure(
"official_count_mismatch",
"native function summary did not reconcile",
));
}
}
Ok(by_id)
}
fn expand_functions(
frames: &HashMap<i64, CompactFrame>,
function_self: Vec<CompactFunctionSelf>,
) -> Result<Vec<RankedFunction>, RankedReportFailure> {
let mut functions = HashMap::<(i64, i64), RankedFunction>::new();
let mut self_identities = HashSet::with_capacity(function_self.len());
for record in function_self {
if record.self_sample_count < 0
|| !self_identities.insert((record.semantic_id, record.function_id))
{
return Err(aggregate_failure(
"invalid_function_self_count",
"function self counts are invalid",
));
}
let mut function_id = record.function_id;
let mut is_self = true;
let mut path = HashSet::new();
loop {
if !path.insert(function_id) {
return Err(aggregate_failure(
"invalid_function_graph",
"native function graph contains a cycle",
));
}
let frame = frames.get(&function_id).ok_or_else(|| {
aggregate_failure("invalid_function_graph", "native function frame is missing")
})?;
let identity = (record.semantic_id, function_id);
if let Some(function) = functions.get_mut(&identity) {
if is_self {
function.self_sample_count = function
.self_sample_count
.checked_add(record.self_sample_count)
.ok_or_else(|| {
aggregate_failure("count_overflow", "function self count overflowed")
})?;
}
break;
}
functions.insert(
identity,
RankedFunction {
semantic_id: record.semantic_id,
function_id,
parent_function_id: frame.parent_function_id,
name: frame.name.clone(),
module_name: frame.module_name.clone(),
source_file: frame.source_file.clone(),
line_number: frame.line_number,
self_sample_count: if is_self { record.self_sample_count } else { 0 },
inclusive_sample_count: 0,
},
);
if functions.len() > MAX_RECORDS_PER_COLLECTION {
return Err(result_too_large());
}
let Some(parent_id) = frame.parent_function_id else {
break;
};
function_id = parent_id;
is_self = false;
}
}
Ok(functions.into_values().collect())
}
fn fold_functions(functions: &mut [RankedFunction]) -> Result<(), RankedReportFailure> {
let parents = functions
.iter()
.map(|function| {
(
(function.semantic_id, function.function_id),
function
.parent_function_id
.map(|parent| (function.semantic_id, parent)),
)
})
.collect::<HashMap<_, _>>();
let self_counts = functions
.iter()
.map(|function| {
(
(function.semantic_id, function.function_id),
function.self_sample_count,
)
})
.collect::<HashMap<_, _>>();
let inclusive = fold_inclusive_counts(&parents, &self_counts)
.ok_or_else(|| aggregate_failure("count_overflow", "function sample fold failed"))?;
for function in functions {
function.inclusive_sample_count = *inclusive
.get(&(function.semantic_id, function.function_id))
.ok_or_else(|| aggregate_failure("invalid_structure", "function fold is incomplete"))?;
}
Ok(())
}
fn reconcile_official_self_counts(
frames: &HashMap<i64, CompactFrame>,
functions: &[RankedFunction],
) -> Result<(), RankedReportFailure> {
let mut actual = HashMap::<i64, i64>::new();
for function in functions {
let count = actual.entry(function.function_id).or_default();
*count = count
.checked_add(function.self_sample_count)
.ok_or_else(|| {
aggregate_failure(
"count_overflow",
"official function reconciliation overflowed",
)
})?;
}
if frames.iter().any(|(function_id, frame)| {
actual.get(function_id).copied().unwrap_or_default() != frame.official_self_sample_count
}) {
return Err(aggregate_failure(
"official_count_mismatch",
"native function self counts did not reconcile",
));
}
Ok(())
}
fn push_bounded<T>(records: &mut Vec<T>, record: T) -> Result<(), RankedReportFailure> {
if records.len() == MAX_RECORDS_PER_COLLECTION {
return Err(result_too_large());
}
records.push(record);
Ok(())
}
fn decode_hex(value: &[u8]) -> Result<Vec<u8>, RankedReportFailure> {
if !value.len().is_multiple_of(2) {
return Err(malformed_result());
}
value
.chunks_exact(2)
.map(|pair| {
let high = hex_digit(pair[0]).ok_or_else(malformed_result)?;
let low = hex_digit(pair[1]).ok_or_else(malformed_result)?;
Ok((high << 4) | low)
})
.collect()
}
fn hex_digit(value: u8) -> Option<u8> {
match value {
b'0'..=b'9' => Some(value - b'0'),
b'a'..=b'f' => Some(value - b'a' + 10),
b'A'..=b'F' => Some(value - b'A' + 10),
_ => None,
}
}
fn malformed_result() -> RankedReportFailure {
RankedReportFailure::new(
RankedReportFailurePhase::Query,
"malformed_result",
"ranked profile query returned an unexpected result",
)
}
fn result_too_large() -> RankedReportFailure {
RankedReportFailure::new(
RankedReportFailurePhase::Query,
"result_too_large",
"ranked profile query returned too many records",
)
}
fn aggregate_failure(kind: &'static str, message: &'static str) -> RankedReportFailure {
RankedReportFailure::new(RankedReportFailurePhase::AggregateValidation, kind, message)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ranked_sql_scopes_reports_and_preserves_function_symbols() {
assert_eq!(
report_selection_sql(None),
"CREATE PERFETTO TABLE delta_funnel_report_selection AS\nSELECT NULL AS capture_scope_id;"
);
assert_eq!(
report_selection_sql(Some(42)),
"CREATE PERFETTO TABLE delta_funnel_report_selection AS\nSELECT 42 AS capture_scope_id;"
);
assert!(CAPTURE_HEALTH_SQL.contains("selected_operations AS"));
assert!(
CAPTURE_HEALTH_SQL
.contains("selected_operation_tracks(track_id, expected_operation_id) AS")
);
assert!(CAPTURE_HEALTH_SQL.contains("FROM delta_funnel_report_selection"));
assert!(CAPTURE_HEALTH_SQL.contains("JOIN selected_operation_tracks AS parent"));
assert!(CAPTURE_HEALTH_SQL.contains("selected.expected_operation_id"));
assert!(
CAPTURE_HEALTH_SQL
.contains("operation_id IS expected_operation_id AS operation_identity_valid")
);
assert!(CAPTURE_HEALTH_SQL.contains("AS ranked_semantic_candidate"));
assert!(CAPTURE_HEALTH_SQL.contains("WHEN ranked_semantic_candidate THEN"));
assert!(CAPTURE_HEALTH_SQL.contains("time_semantics NOT IN ('wall_clock', 'lifecycle')"));
let terminal_result_clause = CAPTURE_HEALTH_SQL
.split_once("AS missing_terminal_result_count")
.expect("capture health should emit missing terminal results")
.0
.rsplit_once("coalesce(sum(")
.expect("missing terminal results should be an aggregate")
.1;
assert!(!terminal_result_clause.contains("'operator'"));
assert!(
!CAPTURE_HEALTH_SQL.contains("OR extract_arg(s.arg_set_id, 'debug.operation_id') IN")
);
assert!(RANKED_PROFILE_BASE_SQL.contains("nullif(frame.name, '')"));
assert!(!RANKED_PROFILE_BASE_SQL.contains("substr(frame.name"));
}
#[test]
fn parses_and_folds_compact_ranked_records() -> Result<(), Box<dyn std::error::Error>> {
use super::super::report_html::render_ranked_profile_html;
use super::super::report_terminal::{InspectSelection, InspectSort};
let output = fixture_output(1, 0, true, 0);
let document = parse_ranked_report_output(output.as_bytes())?;
assert_eq!(document.metadata.direct_sample_count, 1);
assert_eq!(document.semantics.len(), 2);
assert_eq!(document.semantics[0].inclusive_sample_count, 1);
assert_eq!(document.functions.len(), 2);
assert_eq!(document.functions[0].function_id, 10);
assert_eq!(document.functions[0].self_sample_count, 0);
assert_eq!(document.functions[0].inclusive_sample_count, 1);
assert_eq!(document.functions[1].function_id, 11);
assert_eq!(document.functions[1].self_sample_count, 1);
let mismatch = parse_ranked_report_output(fixture_output(2, 0, true, 0).as_bytes())
.expect_err("mismatched official count should fail");
assert_eq!(
mismatch.phase(),
RankedReportFailurePhase::AggregateValidation
);
assert_eq!(mismatch.kind(), "official_count_mismatch");
let audit = parse_ranked_report_output(fixture_output(1, 1, true, 0).as_bytes())
.expect_err("failed query audit should fail");
assert_eq!(audit.kind(), "audit_failed");
let metadata_only = |missing_identity_field_count| {
fixture_output(1, 0, true, missing_identity_field_count)
.lines()
.filter(|line| !line.is_empty())
.take(2)
.collect::<Vec<_>>()
.join("\n")
};
let unavailable = parse_ranked_report_output(metadata_only(0).as_bytes())
.expect_err("capture without an operation should remain unavailable");
assert_eq!(unavailable.phase(), RankedReportFailurePhase::Health);
assert_eq!(unavailable.kind(), "incomplete_capture");
let unsafe_metadata_only = parse_ranked_report_output(metadata_only(1).as_bytes())
.expect_err("filtered malformed identity should fail closed");
assert_eq!(
unsafe_metadata_only.phase(),
RankedReportFailurePhase::AggregateValidation
);
assert_eq!(unsafe_metadata_only.kind(), "unsafe_identity");
let incomplete = parse_ranked_report_output(fixture_output(1, 0, false, 0).as_bytes())?;
assert!(!incomplete.metadata.capture_complete);
assert!(!incomplete.metadata.semantic_complete);
assert!(incomplete.metadata.finalization_observed);
assert_eq!(incomplete.metadata.incomplete_operation_root_count, 1);
assert!(!incomplete.semantics[0].is_complete);
assert_eq!(incomplete.semantics[0].end_ns, None);
assert_eq!(incomplete.semantics[0].duration_ns, None);
assert_eq!(incomplete.semantics[0].result, None);
let completed_child = &incomplete.semantics[1];
assert_eq!(completed_child.parent_semantic_id, Some(1));
assert!(completed_child.is_complete);
assert_eq!(completed_child.end_ns, Some(18));
assert_eq!(completed_child.duration_ns, Some(6));
assert_eq!(completed_child.result.as_deref(), Some("ok"));
let terminal = super::super::render_terminal_view(
&incomplete,
InspectSelection::Root,
InspectSort::Duration,
None,
10,
1,
)?;
assert!(terminal.contains(
"id=semantic:2 name=\"Completed child\" kind=\"phase\" duration_ns=6 time_basis=exact:wall_clock operation_wall_percent=n/a complete=true result=\"ok\""
));
let html = render_ranked_profile_html(&incomplete)?;
assert!(html.contains(r#""name":"Completed child""#));
assert!(html.contains(r#""end_ns":18,"duration_ns":6"#));
assert!(html.contains(r#""result":"ok","is_complete":true"#));
let unsafe_identity = parse_ranked_report_output(fixture_output(1, 0, false, 1).as_bytes())
.expect_err("missing ownership identity should fail");
assert_eq!(unsafe_identity.kind(), "unsafe_identity");
let malformed = parse_ranked_report_output(b"\"record_hex\"\n\"xyz\"\n")
.expect_err("invalid hex should fail");
assert_eq!(malformed.phase(), RankedReportFailurePhase::Query);
assert_eq!(malformed.kind(), "malformed_result");
Ok(())
}
#[test]
#[ignore = "requires trace_processor_shell and a real raw trace"]
fn loads_and_renders_a_real_raw_trace_when_requested() -> Result<(), Box<dyn std::error::Error>>
{
use super::super::report_terminal::{InspectSelection, InspectSort};
let trace = std::env::var_os("DELTA_FUNNEL_TEST_PERFETTO_TRACE")
.ok_or("DELTA_FUNNEL_TEST_PERFETTO_TRACE is not set")?;
let document = load_ranked_profile(Path::new(&trace), None)?;
assert!(!document.semantics.is_empty());
assert_eq!(
document.metadata.direct_sample_count,
document
.semantics
.iter()
.map(|semantic| semantic.direct_sample_count)
.sum::<i64>()
);
let output = super::super::render_terminal_view(
&document,
InspectSelection::Root,
InspectSort::Duration,
None,
20,
0,
)?;
assert!(output.starts_with("view: ranked-profile\ncontext: operation-roots\n"));
assert!(output.contains("time_basis=exact:"));
assert!(output.contains("sample_unit: samples"));
Ok(())
}
#[test]
#[ignore = "requires trace_processor_shell and a real interrupted raw trace"]
fn preserves_a_real_interrupted_capture_when_requested()
-> Result<(), Box<dyn std::error::Error>> {
use super::super::report_html::render_ranked_profile_html;
use super::super::report_terminal::{InspectSelection, InspectSort};
let trace = std::env::var_os("DELTA_FUNNEL_TEST_INCOMPLETE_PERFETTO_TRACE")
.ok_or("DELTA_FUNNEL_TEST_INCOMPLETE_PERFETTO_TRACE is not set")?;
let trace = Path::new(&trace);
let original = std::fs::read(trace)?;
let document = load_ranked_profile(trace, None)?;
assert!(!document.metadata.capture_complete);
assert!(!document.metadata.semantic_complete);
assert!(document.metadata.incomplete_operation_root_count > 0);
assert!(document.metadata.missing_terminal_result_count > 0);
assert!(
document
.semantics
.iter()
.filter(|semantic| !semantic.is_complete)
.all(|semantic| {
semantic.end_ns.is_none()
&& semantic.duration_ns.is_none()
&& semantic.result.is_none()
})
);
assert!(document.semantics.iter().any(|semantic| {
semantic.parent_semantic_id.is_some()
&& semantic.is_complete
&& semantic.duration_ns.is_some()
&& semantic.result.is_some()
}));
let terminal = super::super::render_terminal_view(
&document,
InspectSelection::Root,
InspectSort::Duration,
None,
20,
0,
)?;
assert!(terminal.contains("capture_complete: false"));
assert!(terminal.contains("semantic_complete: false"));
assert!(terminal.contains("duration_ns=n/a"));
assert!(terminal.contains("operation_wall_percent=n/a"));
assert!(terminal.contains("result=null"));
let html = render_ranked_profile_html(&document)?;
assert!(html.contains(r#""capture_complete":false"#));
assert!(html.contains(r#""semantic_complete":false"#));
assert!(html.contains("Capture health"));
assert_eq!(std::fs::read(trace)?, original);
Ok(())
}
fn fixture_output(
official_root_count: i64,
audit_error_count: i64,
capture_complete: bool,
missing_identity_field_count: i64,
) -> String {
let records = [
serde_json::json!({
"record_kind": "metadata",
"record": {
"capture_complete": capture_complete,
"semantic_complete": capture_complete,
"finalization_observed": true,
"incomplete_operation_root_count": if capture_complete { 0 } else { 1 },
"truncation_marker_count": 0,
"missing_identity_field_count": missing_identity_field_count,
"missing_terminal_result_count": if capture_complete { 0 } else { 1 },
"crossing_worker_slice_count": 0,
"crossing_planning_activity_slice_count": 0,
"crossing_execution_activity_slice_count": 0,
"invalid_planning_activity_hierarchy_count": 0,
"invalid_execution_activity_hierarchy_count": 0,
"perf_sample_without_callsite_count": 0,
"perf_samples_skipped": 0,
"buffer_loss_count": 0,
"data_source_loss_count": 0,
"flush_failure_count": 0,
"schema_version": 3,
"sample_frequency_hz": 1000,
"sampled_cpu_count": 1,
"exact_time_unit": "nanoseconds",
"sample_unit": "samples",
"eligible_sample_count": 1,
"direct_sample_count": 1,
"ambiguous_sample_count": 0,
"unattributed_sample_count": 0,
"resolved_function_sample_count": 1,
"unresolved_function_sample_count": 0,
"unwind_error_sample_count": 0,
"missing_callstack_sample_count": 0,
"trace_profiler_dropped_sample_count": 0,
"audit_error_count": audit_error_count,
}
}),
serde_json::json!({
"record_kind": "semantic",
"record": {
"semantic_id": 1,
"parent_semantic_id": null,
"operation_id": 1,
"name": "Delta Funnel preview",
"semantic_kind": "operation",
"operation_kind": "preview",
"stage_category": null,
"stage_name": null,
"activity": null,
"start_ns": 10,
"end_ns": capture_complete.then_some(20),
"duration_ns": capture_complete.then_some(10),
"time_semantics": "wall_clock",
"result": capture_complete.then_some("ok"),
"is_complete": capture_complete,
"query_execution_id": null,
"query_scope": null,
"query_owner": null,
"worker_lane_id": null,
"worker_kind": null,
"node_id": null,
"parent_node_id": null,
"operator_partition": null,
"execution_stream_id": null,
"stage_owner_id": null,
"direct_sample_count": 1,
"inclusive_sample_count": 0,
"resolved_function_sample_count": 1,
"unresolved_function_sample_count": 0,
"unwind_error_sample_count": 0,
"missing_callstack_sample_count": 0,
}
}),
serde_json::json!({
"record_kind": "semantic",
"record": {
"semantic_id": 2,
"parent_semantic_id": 1,
"operation_id": 1,
"name": "Completed child",
"semantic_kind": "phase",
"operation_kind": "preview",
"stage_category": null,
"stage_name": null,
"activity": null,
"start_ns": 12,
"end_ns": 18,
"duration_ns": 6,
"time_semantics": "wall_clock",
"result": "ok",
"is_complete": true,
"query_execution_id": null,
"query_scope": null,
"query_owner": null,
"worker_lane_id": null,
"worker_kind": null,
"node_id": null,
"parent_node_id": null,
"operator_partition": null,
"execution_stream_id": null,
"stage_owner_id": null,
"direct_sample_count": 0,
"inclusive_sample_count": 0,
"resolved_function_sample_count": 0,
"unresolved_function_sample_count": 0,
"unwind_error_sample_count": 0,
"missing_callstack_sample_count": 0,
}
}),
frame_record(10, None, 0, official_root_count),
frame_record(11, Some(10), 1, 1),
serde_json::json!({
"record_kind": "function_self",
"record": {
"semantic_id": 1,
"function_id": 11,
"self_sample_count": 1,
}
}),
];
let mut output = String::from("\n\"record_hex\"\n");
for record in records {
output.push('"');
for byte in record.to_string().as_bytes() {
use std::fmt::Write as _;
write!(output, "{byte:02X}").expect("writing to a string should succeed");
}
output.push_str("\"\n");
}
output
}
fn frame_record(
function_id: i64,
parent_function_id: Option<i64>,
self_count: i64,
inclusive_count: i64,
) -> serde_json::Value {
serde_json::json!({
"record_kind": "frame",
"record": {
"function_id": function_id,
"parent_function_id": parent_function_id,
"name": format!("function_{function_id}"),
"module_name": "delta-funnel",
"source_file": "src/lib.rs",
"line_number": function_id,
"official_self_sample_count": self_count,
"official_inclusive_sample_count": inclusive_count,
}
})
}
}