diff --git a/README.md b/README.md index 4a5995f..c45bed9 100644 --- a/README.md +++ b/README.md @@ -186,6 +186,10 @@ New uploads use each event's UTC day, while day and `event-partitions//track..json` records each immutable object's range. Athena combines projected stream/day partitions with its hidden `$path` column, so a wide legacy capture day can prune unrelated objects. +Exact `trace_show` and `trace_compare` lookups recover event time from Synty's +ULID ids. If an exact id or resolved-session lookup still spans more object +paths than the request guard permits, Athena falls back to the bounded +stream/day partitions while retaining the exact id/session predicate. Once a stream index exists, readers include its legacy days and objects missing from the object index conservatively. A stream with no partition metadata falls back to physical days overlapping the requested window; initialize or backfill diff --git a/src/event.rs b/src/event.rs index fbade9b..43f466c 100644 --- a/src/event.rs +++ b/src/event.rs @@ -106,6 +106,29 @@ pub fn deterministic_ulid(ts_ms: u64, key: &str) -> String { ulid_string(ts_ms, &entropy) } +/// Recover the 48-bit millisecond timestamp from a canonical ULID. Trace +/// lookups use this to prune the raw event lake before applying an exact id +/// predicate; foreign/non-ULID ids deliberately return `None`. +pub fn ulid_timestamp_ms(id: &str) -> Option { + if id.len() != 26 { + return None; + } + let mut timestamp = 0u64; + for byte in id.bytes().take(10) { + let value = match byte.to_ascii_uppercase() { + b'0'..=b'9' => byte - b'0', + b'A'..=b'H' => byte.to_ascii_uppercase() - b'A' + 10, + b'J'..=b'K' => byte.to_ascii_uppercase() - b'J' + 18, + b'M'..=b'N' => byte.to_ascii_uppercase() - b'M' + 20, + b'P'..=b'T' => byte.to_ascii_uppercase() - b'P' + 22, + b'V'..=b'Z' => byte.to_ascii_uppercase() - b'V' + 27, + _ => return None, + }; + timestamp = (timestamp << 5) | u64::from(value); + } + (timestamp <= 0x0000_ffff_ffff_ffff).then_some(timestamp) +} + /// Encode (48-bit time, 80-bit entropy) into the canonical 26-char Crockford /// base32 ULID string (the oklog/ulid byte packing). fn ulid_string(ts_ms: u64, entropy: &[u8; 10]) -> String { @@ -192,6 +215,16 @@ mod tests { assert!(early[..10] < late[..10]); } + #[test] + fn ulid_timestamp_roundtrips_and_rejects_foreign_ids() { + let timestamp = 1_700_000_000_123; + let id = deterministic_ulid(timestamp, "session"); + assert_eq!(ulid_timestamp_ms(&id), Some(timestamp)); + assert_eq!(ulid_timestamp_ms(&id.to_ascii_lowercase()), Some(timestamp)); + assert_eq!(ulid_timestamp_ms("job:call-1"), None); + assert_eq!(ulid_timestamp_ms("Z0000000000000000000000000"), None); + } + // The envelope round-trips and matches the wire field names ingest reads. #[test] fn event_json_uses_canonical_field_names() { diff --git a/src/trace_athena.rs b/src/trace_athena.rs index ee90d02..7e3df21 100644 --- a/src/trace_athena.rs +++ b/src/trace_athena.rs @@ -13,6 +13,7 @@ use std::time::{Duration as StdDuration, Instant}; const DEFAULT_LIST_HOURS: i64 = 1; const DEFAULT_LOOKUP_HOURS: i64 = 24 * 7; +const ID_LOOKUP_MINUTES: i64 = 5; const MAX_LOOKBACK_HOURS: i64 = 24 * 7; const MAX_EVENTS: usize = 50_000; const MAX_RESULT_BYTES: usize = 64 * 1024 * 1024; @@ -378,7 +379,8 @@ impl Backend { return Ok(out); } let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; - let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?; + let window = id_lookup_window(id) + .unwrap_or(Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?); let store = self.load_store(window, None, None, None, None, &[id], true, scope, deadline)?; let out = trace::show_store_text(&store, id, before, after, false, Some(scope))?; @@ -431,7 +433,8 @@ impl Backend { return Ok(out); } let deadline = Instant::now() + REQUEST_QUERY_TIMEOUT; - let window = Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?; + let window = ids_lookup_window(&[left, right]) + .unwrap_or(Window::parse(None, None, DEFAULT_LOOKUP_HOURS)?); let store = self.load_store( window, None, @@ -472,6 +475,16 @@ impl Backend { }; let first = self.select(&streams, window, &predicate, deadline)?; let sessions = event_sessions(&first.lines); + let matched_streams = event_streams(&first.lines); + let context_streams = if matched_streams.is_empty() { + streams.clone() + } else { + streams + .iter() + .filter(|stream| matched_streams.contains(*stream)) + .cloned() + .collect() + }; if sessions.len() > MAX_SESSIONS { return Err(limit_error(format!( "Athena trace selection spans more than {MAX_SESSIONS} sessions; narrow the time, machine, source, or operation filter" @@ -488,7 +501,7 @@ impl Backend { && !sessions.is_empty(); let mut lines = if expands_sessions { self.select( - &streams, + &context_streams, context_window, &Predicate { sessions: sessions.iter().cloned().collect(), @@ -504,7 +517,7 @@ impl Backend { if !sessions.is_empty() && !expands_sessions { let mut contexts = self .select( - &streams, + &context_streams, context_window, &Predicate { sessions: sessions.into_iter().collect(), @@ -529,8 +542,28 @@ impl Backend { predicate: &Predicate, deadline: Instant, ) -> Result { - let selection = self.selected_objects(streams, window)?; - if selection.paths.is_empty() { + let permits_partition_scan = + !predicate.ids.is_empty() || !predicate.sessions.is_empty(); + let (selection, partition_scan) = match self.selected_objects(streams, window) { + Ok(selection) => (selection, false), + Err(error) + if permits_partition_scan + && matches!( + error.downcast_ref::(), + Some(TraceQueryError::Limit(_)) + ) => + { + ( + ObjectSelection { + days: window_days(window).into_iter().collect(), + paths: Vec::new(), + }, + true, + ) + } + Err(error) => return Err(error), + }; + if selection.days.is_empty() || (!partition_scan && selection.paths.is_empty()) { metrics::Run::new("athena_trace") .set("outcome", "empty") .set("rows", 0) @@ -751,7 +784,6 @@ fn select_sql( "Athena trace needs at least one stream" ); anyhow::ensure!(!days.is_empty(), "Athena trace needs at least one day"); - anyhow::ensure!(!paths.is_empty(), "Athena trace needs at least one object path"); let stream_values = streams .iter() .map(|stream| sql_string(stream)) @@ -762,15 +794,19 @@ fn select_sql( .map(|day| sql_string(day)) .collect::>() .join(", "); - let path_values = paths - .iter() - .map(|path| sql_string(path)) - .collect::>() - .join(", "); let mut clauses = vec![ format!("stream IN ({stream_values})"), format!("day IN ({day_values})"), - format!("\"$path\" IN ({path_values})"), + ]; + if !paths.is_empty() { + let path_values = paths + .iter() + .map(|path| sql_string(path)) + .collect::>() + .join(", "); + clauses.push(format!("\"$path\" IN ({path_values})")); + } + clauses.extend([ format!( "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) >= from_iso8601_timestamp({})", sql_string(&window.since.to_rfc3339()) @@ -779,7 +815,7 @@ fn select_sql( "from_iso8601_timestamp(json_extract_scalar(line, '$.ts')) < from_iso8601_timestamp({})", sql_string(&window.until.to_rfc3339()) ), - ]; + ]); if let Some(source) = predicate.source.as_deref() { clauses.push(format!( "strpos(lower(coalesce(json_extract_scalar(line, '$.source'), '')), {}) > 0", @@ -881,6 +917,15 @@ fn event_sessions(lines: &[String]) -> BTreeSet { sessions } +fn event_streams(lines: &[String]) -> BTreeSet { + lines + .iter() + .filter_map(|line| serde_json::from_str::(line).ok()) + .map(|event| event.stream) + .filter(|stream| !stream.is_empty()) + .collect() +} + fn sql_string(value: &str) -> String { format!("'{}'", value.replace('\'', "''")) } @@ -889,6 +934,30 @@ fn query_id(id: &str) -> &str { id.strip_prefix("job:").unwrap_or(id) } +fn id_lookup_window(id: &str) -> Option { + ids_lookup_window(&[id]) +} + +fn ids_lookup_window(ids: &[&str]) -> Option { + let mut timestamps = ids.iter().map(|id| { + let timestamp = crate::event::ulid_timestamp_ms(query_id(id))?; + DateTime::::from_timestamp_millis(timestamp as i64) + }); + let first = timestamps.next()??; + let (mut since, mut until) = (first, first); + for timestamp in timestamps { + let timestamp = timestamp?; + since = since.min(timestamp); + until = until.max(timestamp); + } + let window = Window { + since: since - Duration::minutes(ID_LOOKUP_MINUTES), + until: until + Duration::minutes(ID_LOOKUP_MINUTES), + }; + (window.until - window.since <= Duration::hours(MAX_LOOKBACK_HOURS)) + .then_some(window) +} + fn normalized_stream_source(source: &str) -> String { match source.to_ascii_lowercase().as_str() { "codex" | "codex_cli" | "codex-cli" => "codex".into(), @@ -1006,6 +1075,66 @@ mod tests { } } + #[test] + fn exact_id_falls_back_to_stream_day_partitions_when_paths_exceed_the_guard() { + use crate::bucket::Bucket; + + let root = std::env::temp_dir().join(format!( + "synty-athena-id-partition-fallback-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&root); + let bucket = crate::bucket::LocalFs::new(&root); + for index in 0..=MAX_OBJECT_PATHS { + bucket + .put( + &format!( + "events/edge-m-codex/chunks/track.2026-07-22/{index:04}.jsonl" + ), + b"{}\n", + ) + .unwrap(); + } + let calls = Arc::new(Mutex::new(Vec::new())); + let mut backend = Backend { + config: Config { + bucket: root.to_string_lossy().into_owned(), + workgroup: "wg".into(), + database: "synty".into(), + table: "raw_events".into(), + }, + query: Box::new(FakeQuery { + lines: Vec::new(), + calls: Arc::clone(&calls), + }), + streams: Some(vec!["edge-m-codex".into()]), + days: None, + cached: None, + }; + + backend + .select( + &["edge-m-codex".into()], + Window { + since: parse_time("2026-07-22T10:00:00Z").unwrap(), + until: parse_time("2026-07-22T10:10:00Z").unwrap(), + }, + &Predicate { + ids: vec!["event-id".into()], + ..Default::default() + }, + Instant::now() + StdDuration::from_secs(5), + ) + .unwrap(); + + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 1); + assert!(calls[0].contains("day IN ('2026-07-22')")); + assert!(!calls[0].contains("\"$path\"")); + assert!(calls[0].contains("'event-id'")); + let _ = std::fs::remove_dir_all(&root); + } + #[test] fn raw_athena_rows_reconstruct_the_existing_span_surface() { let lines = vec![ @@ -1182,6 +1311,73 @@ mod tests { assert!(!calls[1].contains("synty_context_rank")); } + #[test] + fn show_uses_ulid_time_and_matching_stream_for_bounded_lookup() { + let timestamp = parse_time("2026-07-22T10:00:01Z").unwrap(); + let id = crate::event::deterministic_ulid(timestamp.timestamp_millis() as u64, "call"); + let lines = vec![ + event( + "start", + "2026-07-22T10:00:00Z", + "session_start", + json!({"cwd":"/work/synty"}), + ), + event( + &id, + "2026-07-22T10:00:01Z", + "tool_call", + json!({"name":"exec_command","call_id":"c1","arguments":"{\"cmd\":\"cargo test\"}"}), + ), + ]; + let calls = Arc::new(Mutex::new(Vec::new())); + let mut backend = Backend { + config: Config::new( + "s3://bucket".into(), + "wg".into(), + "synty".into(), + "raw_events".into(), + ) + .unwrap(), + query: Box::new(FakeQuery { + lines, + calls: Arc::clone(&calls), + }), + streams: Some(vec![ + "edge-m-codex".into(), + "edge-other-claudecode".into(), + ]), + days: Some(vec!["2026-07-22".into()]), + cached: None, + }; + + let out = backend + .show(&id, 3, 5, &ReadScope::default()) + .unwrap(); + + assert!(out.contains("cargo test")); + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 2); + assert!(calls[0].contains("2026-07-22T09:55:01+00:00")); + assert!(calls[0].contains("2026-07-22T10:05:01+00:00")); + assert!(calls[0].contains("'edge-other-claudecode'")); + assert!(calls[1].contains("stream IN ('edge-m-codex')")); + assert!(!calls[1].contains("'edge-other-claudecode'")); + } + + #[test] + fn compare_window_covers_both_ulid_timestamps() { + let early = parse_time("2026-07-22T10:00:00Z").unwrap(); + let late = parse_time("2026-07-22T10:30:00Z").unwrap(); + let left = crate::event::deterministic_ulid(early.timestamp_millis() as u64, "left"); + let right = crate::event::deterministic_ulid(late.timestamp_millis() as u64, "right"); + + let window = ids_lookup_window(&[&left, &format!("job:{right}")]).unwrap(); + + assert_eq!(window.since, parse_time("2026-07-22T09:55:00Z").unwrap()); + assert_eq!(window.until, parse_time("2026-07-22T10:35:00Z").unwrap()); + assert!(ids_lookup_window(&[&left, "foreign-id"]).is_none()); + } + #[test] fn listed_job_ids_query_the_native_span_id() { let lines = vec![