kache_core/
timeline.rs

1//! Build timeline records: what one build session did, in order, with timings.
2//!
3//! The client assembles a record from the logs it already writes and sends it
4//! to kache-service, which stores it as submitted. Every time is Unix epoch
5//! milliseconds. Records carry no filesystem paths and no environment values
6//! outside the allowlisted run context.
7
8use std::collections::BTreeMap;
9
10use serde::{Deserialize, Serialize};
11
12/// Version of [`BuildTimeline`]. A server rejects a record whose schema it
13/// does not know.
14pub const BUILD_TIMELINE_SCHEMA: u32 = 7;
15
16/// One build session: its compiler invocations and the remote transfers that
17/// belong to it.
18#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
19pub struct BuildTimeline {
20    pub schema: u32,
21    /// Stable across re-submissions of the same session from the same run, so
22    /// a later, more complete submission replaces the earlier one.
23    pub client_record_id: String,
24    pub session_id: String,
25    #[serde(default)]
26    pub kache_version: String,
27    /// Earliest unit start.
28    pub started_at_ms: u64,
29    /// Latest unit finish.
30    pub finished_at_ms: u64,
31    #[serde(default)]
32    pub identity: TimelineIdentity,
33    /// Hash of the build root, so sessions from one tree can be grouped without
34    /// sending its path.
35    #[serde(default)]
36    pub root_hash: String,
37    #[serde(default)]
38    pub context: RunContext,
39    #[serde(default)]
40    pub log: LogLimits,
41    /// Prefetch plan summary for the session, when the daemon had already
42    /// closed it when the record was assembled.
43    #[serde(default, skip_serializing_if = "Option::is_none")]
44    pub summary: Option<TimelineSummary>,
45    #[serde(default)]
46    pub units: Vec<TimelineUnit>,
47    #[serde(default)]
48    pub transfers: Vec<TimelineTransfer>,
49}
50
51/// What build this was, as far as the client could tell.
52#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
53pub struct TimelineIdentity {
54    /// Truncated content hash of the root's `Cargo.lock`.
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub lock_digest: Option<String>,
57    /// Prefetch identity key (`id/{lock}/{target}/{profile}`), only when the
58    /// profile was known or the key was set explicitly.
59    #[serde(default, skip_serializing_if = "Option::is_none")]
60    pub identity_key: Option<String>,
61    #[serde(default)]
62    pub source: IdentitySource,
63}
64
65#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
66#[serde(rename_all = "snake_case")]
67pub enum IdentitySource {
68    /// Set explicitly by the build environment.
69    Explicit,
70    /// Derived from the lockfile and a profile named in the environment.
71    LockEnv,
72    #[default]
73    #[serde(other)]
74    Absent,
75}
76
77/// Where the build ran. Every field is optional; only allowlisted CI variables
78/// and explicit labels are sent.
79#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
80pub struct RunContext {
81    #[serde(default, skip_serializing_if = "Option::is_none")]
82    pub repository: Option<String>,
83    #[serde(default, skip_serializing_if = "Option::is_none")]
84    pub workflow: Option<String>,
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub job: Option<String>,
87    #[serde(default, skip_serializing_if = "Option::is_none")]
88    pub run_id: Option<String>,
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub run_attempt: Option<String>,
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub event: Option<String>,
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub git_ref: Option<String>,
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub commit: Option<String>,
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub runner_os: Option<String>,
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub runner_arch: Option<String>,
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub runner_pool: Option<String>,
103    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
104    pub labels: BTreeMap<String, String>,
105}
106
107/// Log rotation limits in force on the client. Once the event log passes
108/// `event_log_max_size` only the last `event_log_keep_lines` lines survive, so
109/// a record from a large build may be missing its first units.
110#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
111pub struct LogLimits {
112    #[serde(default)]
113    pub event_log_max_size: u64,
114    #[serde(default)]
115    pub event_log_keep_lines: u64,
116}
117
118/// The daemon's per-session prefetch plan summary.
119#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
120pub struct TimelineSummary {
121    /// Some prefetch outcomes were unavailable when the session closed.
122    #[serde(default)]
123    pub incomplete: bool,
124    #[serde(default)]
125    pub plan_id: String,
126    #[serde(default)]
127    pub plan_source: String,
128    #[serde(default)]
129    pub closure_reason: String,
130    #[serde(default)]
131    pub started_at_ms: u64,
132    #[serde(default)]
133    pub last_activity_ms: u64,
134    #[serde(default)]
135    pub candidate_keys: u64,
136    #[serde(default)]
137    pub downloaded_keys: u64,
138    #[serde(default)]
139    pub downloaded_bytes: u64,
140    #[serde(default)]
141    pub used_keys: u64,
142    #[serde(default)]
143    pub demanded_keys: u64,
144    #[serde(default)]
145    pub demanded_candidate_keys: u64,
146    #[serde(default)]
147    pub cancelled: bool,
148    /// Speculative GET keys that a wrapper then consumed as a local or
149    /// prefetch hit. Daemon `used_keys` can miss these.
150    #[serde(default)]
151    pub consumed_prefetch_keys: u64,
152    #[serde(default)]
153    pub consumed_prefetch_bytes: u64,
154    /// Consumed prefetch keys whose import finished at or before first demand.
155    #[serde(default)]
156    pub useful_prefetch_keys: u64,
157    #[serde(default)]
158    pub useful_prefetch_bytes: u64,
159    /// Sum of wrapper remote-check waits on this session's units.
160    #[serde(default)]
161    pub remote_wait_ms: u64,
162    #[serde(default)]
163    pub get_not_found: u64,
164    #[serde(default)]
165    pub get_errors: u64,
166    /// Consumed prefetch keys whose GET had started but not finished at first
167    /// demand. The demand waited on the rest of that download, not a new GET.
168    #[serde(default)]
169    pub in_flight_prefetch_keys: u64,
170    #[serde(default)]
171    pub in_flight_prefetch_bytes: u64,
172    /// Speculative GETs or packed entries whose receipt says `cancelled`.
173    #[serde(default)]
174    pub get_cancelled: u64,
175}
176
177/// One key requested by a wrapper, including unsuccessful predictions.
178#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
179pub struct KeyDemand {
180    pub cache_key: String,
181    pub first_demand_at_ms: u64,
182    /// Wall time blocked on remote-check IPC, including daemon admission and
183    /// failed requests. It is not an estimate of time saved by prefetch.
184    #[serde(default)]
185    pub remote_wait_ms: u64,
186}
187
188/// One compiler invocation.
189#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
190pub struct TimelineUnit {
191    pub cache_key: String,
192    pub crate_name: String,
193    /// Wrapper outcome as logged: `local_hit`, `prefetch_hit`, `remote_hit`,
194    /// `dup`, `miss`, `error`, `passthrough`, `skipped`. A prefetched entry
195    /// consumed from the local store logs `local_hit`.
196    pub result: String,
197    /// When the build started waiting for this unit.
198    pub started_at_ms: u64,
199    pub finished_at_ms: u64,
200    #[serde(default)]
201    pub compile_time_ms: u64,
202    #[serde(default)]
203    pub size: u64,
204    #[serde(default)]
205    pub key_ms: u64,
206    #[serde(default)]
207    pub lookup_ms: u64,
208    #[serde(default)]
209    pub restore_ms: u64,
210    #[serde(default)]
211    pub store_ms: u64,
212    #[serde(default)]
213    pub startup_ms: u64,
214    #[serde(default)]
215    pub flight_wait_ms: u64,
216    #[serde(default)]
217    pub permit_wait_ms: u64,
218    #[serde(default)]
219    pub compiler_runs: u32,
220    /// Schema of the wrapper event this unit came from.
221    #[serde(default)]
222    pub event_schema: u32,
223    /// Empty for wrapper events before schema 20, or invocations with no lookup.
224    #[serde(default, skip_serializing_if = "Vec::is_empty")]
225    pub demands: Vec<KeyDemand>,
226    /// The session's earliest speculative delivery of this unit's key, joined
227    /// from transfer receipts when the record is assembled. A `local_hit`
228    /// with `before_demand` timing is a prefetched local hit.
229    #[serde(default, skip_serializing_if = "Option::is_none")]
230    pub prefetch: Option<UnitPrefetch>,
231}
232
233/// A speculative download that delivered a unit's key in the same session.
234#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
235pub struct UnitPrefetch {
236    /// Plan, source and candidate rank that scheduled the download.
237    pub origin: PrefetchOrigin,
238    /// When the GET started. A packed entry reports its pack's GET.
239    pub started_at_ms: u64,
240    /// When the entry finished importing into the local store.
241    pub delivered_at_ms: u64,
242    pub compressed_bytes: u64,
243    pub timing: PrefetchTiming,
244}
245
246/// Where a delivery fell relative to the key's first demand in the session.
247#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
248#[serde(rename_all = "snake_case")]
249pub enum PrefetchTiming {
250    /// Imported at or before first demand.
251    #[default]
252    BeforeDemand,
253    /// Started before first demand and finished after it.
254    InFlight,
255    /// Started at or after first demand.
256    AfterDemand,
257}
258
259impl PrefetchTiming {
260    /// Classify one delivery against the key's first demand.
261    pub fn classify(started_at_ms: u64, delivered_at_ms: u64, first_demand_at_ms: u64) -> Self {
262        if delivered_at_ms <= first_demand_at_ms {
263            Self::BeforeDemand
264        } else if started_at_ms < first_demand_at_ms {
265            Self::InFlight
266        } else {
267            Self::AfterDemand
268        }
269    }
270}
271
272/// One remote transfer attributed to the session.
273#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
274pub struct TimelineTransfer {
275    #[serde(default, skip_serializing_if = "Option::is_none")]
276    pub accounting: Option<PrefetchAccounting>,
277    pub cache_key: String,
278    #[serde(default)]
279    pub crate_name: String,
280    pub direction: TransferDirection,
281    pub ok: bool,
282    #[serde(default)]
283    pub compressed_bytes: u64,
284    #[serde(default)]
285    pub original_bytes: u64,
286    pub started_at_ms: u64,
287    pub finished_at_ms: u64,
288    #[serde(default)]
289    pub network_ms: u64,
290    #[serde(default)]
291    pub semaphore_wait_ms: u64,
292    #[serde(default)]
293    pub request_count: u32,
294    #[serde(default)]
295    pub import_ms: u64,
296    pub attribution: TransferAttribution,
297    /// Immutable origin captured when this candidate was scheduled. Absent
298    /// on demand downloads and logs older than transfer schema 4.
299    #[serde(default, skip_serializing_if = "Option::is_none")]
300    pub prefetch: Option<PrefetchOrigin>,
301    /// Operation/import outcome, including neutral `not_found`, `cancelled`,
302    /// and `skipped`; empty for demand downloads, uploads, and old logs.
303    #[serde(default, skip_serializing_if = "String::is_empty")]
304    pub outcome: String,
305}
306
307#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
308#[serde(rename_all = "snake_case")]
309pub enum TransferDirection {
310    Upload,
311    #[default]
312    Download,
313}
314
315/// Accounting for one physical backend operation. Nested entries describe
316/// payload attribution; their bytes must not be added to the physical total.
317#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
318pub struct PrefetchAccounting {
319    pub operation: PrefetchOperation,
320    /// True when compressed_bytes covers the complete received body. False
321    /// means partial transport bytes are unavailable (including LIST bodies).
322    pub bytes_complete: bool,
323    /// Counts calls to the backend, not SDK retries or LIST pages.
324    pub requests_complete: bool,
325    #[serde(default, skip_serializing_if = "Option::is_none")]
326    pub list_result_count: Option<u64>,
327    #[serde(default, skip_serializing_if = "Vec::is_empty")]
328    pub entries: Vec<PackedEntryTransfer>,
329}
330
331#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
332#[serde(rename_all = "snake_case")]
333pub enum PrefetchOperation {
334    #[default]
335    Get,
336    List,
337}
338
339/// One compressed entry frame within a physical pack body. A successful local
340/// import makes the entry available at finished_at_ms; zero means no import.
341#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
342pub struct PackedEntryTransfer {
343    pub cache_key: String,
344    pub crate_name: String,
345    pub compressed_bytes: u64,
346    pub finished_at_ms: u64,
347    pub outcome: String,
348    pub prefetch: PrefetchOrigin,
349}
350
351/// The plan that scheduled one speculative download. Kept with the task,
352/// so a later build cannot claim a download that was already in flight.
353#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
354pub struct PrefetchOrigin {
355    #[serde(default)]
356    pub session_id: String,
357    #[serde(default)]
358    /// Planner-issued ID when available; empty for fallback and unscoped work.
359    pub plan_id: String,
360    /// `advisory`, `fallback`, or `unscoped` (startup or direct requests).
361    #[serde(default)]
362    pub source: String,
363    /// Zero-based order in the request list, before local-hit, in-flight,
364    /// and budget filtering. Planner requests validate their list first;
365    /// direct IPC requests can include rejected keys. None for keys added
366    /// by whole-remote warming.
367    #[serde(default)]
368    pub candidate_rank: Option<u64>,
369    #[serde(default)]
370    pub candidate_source: crate::CandidateSource,
371}
372
373/// How the client tied a transfer to this session.
374#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
375#[serde(rename_all = "snake_case")]
376pub enum TransferAttribution {
377    /// The download task carries this session's immutable plan origin.
378    Session,
379    /// Same cache key as one of the session's units.
380    Key,
381    /// No key match; its time overlaps this session and no other.
382    #[default]
383    Window,
384}
385
386#[cfg(test)]
387mod tests {
388    use super::*;
389
390    fn sample() -> BuildTimeline {
391        BuildTimeline {
392            schema: BUILD_TIMELINE_SCHEMA,
393            client_record_id: "0123456789abcdef".into(),
394            session_id: "fedcba9876543210".into(),
395            kache_version: "0.23.1".into(),
396            started_at_ms: 1_000,
397            finished_at_ms: 5_000,
398            identity: TimelineIdentity {
399                lock_digest: Some("aaaabbbbccccdddd".into()),
400                identity_key: None,
401                source: IdentitySource::Absent,
402            },
403            root_hash: "1111222233334444".into(),
404            context: RunContext {
405                repository: Some("org/repo".into()),
406                labels: BTreeMap::from([("phase".into(), "cold".into())]),
407                ..RunContext::default()
408            },
409            log: LogLimits {
410                event_log_max_size: 10 << 20,
411                event_log_keep_lines: 1000,
412            },
413            summary: None,
414            units: vec![TimelineUnit {
415                cache_key: "k1".into(),
416                crate_name: "serde".into(),
417                result: "local_hit".into(),
418                demands: vec![KeyDemand {
419                    cache_key: "k1".into(),
420                    first_demand_at_ms: 1_050,
421                    remote_wait_ms: 0,
422                }],
423                started_at_ms: 1_000,
424                finished_at_ms: 1_200,
425                prefetch: Some(UnitPrefetch {
426                    origin: PrefetchOrigin {
427                        session_id: "fedcba9876543210".into(),
428                        plan_id: "p1".into(),
429                        source: "advisory".into(),
430                        candidate_rank: Some(3),
431                        ..PrefetchOrigin::default()
432                    },
433                    started_at_ms: 900,
434                    delivered_at_ms: 990,
435                    compressed_bytes: 64,
436                    timing: PrefetchTiming::BeforeDemand,
437                }),
438                ..TimelineUnit::default()
439            }],
440            transfers: vec![TimelineTransfer {
441                cache_key: "k1".into(),
442                direction: TransferDirection::Download,
443                ok: true,
444                started_at_ms: 900,
445                finished_at_ms: 990,
446                attribution: TransferAttribution::Key,
447                ..TimelineTransfer::default()
448            }],
449        }
450    }
451
452    #[test]
453    fn legacy_unit_has_no_demand_observation() {
454        let unit: TimelineUnit = serde_json::from_str(
455            r#"{"cache_key":"k","crate_name":"crate","result":"local_hit","started_at_ms":100,"finished_at_ms":200}"#,
456        ).unwrap();
457        assert!(unit.demands.is_empty());
458        assert!(unit.prefetch.is_none());
459    }
460
461    #[test]
462    fn prefetch_timing_splits_on_first_demand() {
463        let classify = PrefetchTiming::classify;
464        assert_eq!(classify(10, 99, 100), PrefetchTiming::BeforeDemand);
465        assert_eq!(classify(10, 100, 100), PrefetchTiming::BeforeDemand);
466        assert_eq!(classify(10, 101, 100), PrefetchTiming::InFlight);
467        assert_eq!(classify(99, 150, 100), PrefetchTiming::InFlight);
468        assert_eq!(classify(100, 150, 100), PrefetchTiming::AfterDemand);
469        assert_eq!(classify(120, 150, 100), PrefetchTiming::AfterDemand);
470    }
471
472    #[test]
473    fn unit_prefetch_serializes_timing_in_snake_case() {
474        let unit = TimelineUnit {
475            prefetch: Some(UnitPrefetch {
476                timing: PrefetchTiming::InFlight,
477                ..UnitPrefetch::default()
478            }),
479            ..TimelineUnit::default()
480        };
481        let json = serde_json::to_value(&unit).unwrap();
482        assert_eq!(json["prefetch"]["timing"], "in_flight");
483        let absent = serde_json::to_value(TimelineUnit::default()).unwrap();
484        assert!(absent.get("prefetch").is_none());
485    }
486
487    #[test]
488    fn record_round_trips_through_json() {
489        let record = sample();
490        let json = serde_json::to_string(&record).unwrap();
491        assert_eq!(
492            serde_json::from_str::<BuildTimeline>(&json).unwrap(),
493            record
494        );
495    }
496
497    #[test]
498    fn empty_optional_context_is_not_serialized() {
499        let json = serde_json::to_value(sample()).unwrap();
500        let context = json["context"].as_object().unwrap();
501        assert_eq!(
502            context.keys().collect::<Vec<_>>(),
503            vec!["labels", "repository"]
504        );
505        assert!(json.get("summary").is_none());
506    }
507
508    #[test]
509    fn enums_use_snake_case_names() {
510        let json = serde_json::to_value(sample()).unwrap();
511        assert_eq!(json["identity"]["source"], "absent");
512        assert_eq!(json["transfers"][0]["direction"], "download");
513        assert_eq!(json["transfers"][0]["attribution"], "key");
514        assert_eq!(
515            serde_json::to_value(IdentitySource::LockEnv).unwrap(),
516            "lock_env"
517        );
518    }
519
520    #[test]
521    fn unknown_identity_source_reads_as_absent() {
522        let source: IdentitySource = serde_json::from_str("\"from_the_future\"").unwrap();
523        assert_eq!(source, IdentitySource::Absent);
524    }
525
526    #[test]
527    fn minimal_record_uses_defaults() {
528        let record: BuildTimeline = serde_json::from_str(
529            r#"{"schema":1,"client_record_id":"r","session_id":"s","started_at_ms":1,"finished_at_ms":2}"#,
530        )
531        .unwrap();
532        assert!(record.units.is_empty());
533        assert!(record.transfers.is_empty());
534        assert_eq!(record.identity.source, IdentitySource::Absent);
535    }
536}
kache_core/timeline.rs

kache_core/
timeline.rs
