Skip to content

feat(perf): decide captured observed-performance policies offline #26

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
plotarmordev merged 1 commit into from
Sep 10, 2026
Merged

feat(perf): decide captured observed-performance policies offline #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 104 additions & 15 deletions crates/grill-perf/src/evidence.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use std::io::{Read, Write};
use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
use std::path::{Path, PathBuf};

fn hex(bytes: &[u8]) -> String {
pub(crate) fn hex(bytes: &[u8]) -> String {
const DIGITS: &[u8] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for byte in bytes {
Expand Down Expand Up @@ -127,7 +127,7 @@ pub fn throughput(attempts: &[Attempt], elapsed_us: u64) -> (bool, Option<u64>,
.map(|n| n as f64 * 1_000_000.0 / elapsed_us as f64);
(eligible && rate.is_some(), tokens, rate)
}
fn decode_rate(attempt: &Attempt) -> Option<f64> {
pub(crate) fn decode_sample(attempt: &Attempt) -> Option<(u64, u64)> {
if attempt.status != Status::Complete {
return None;
}
Expand All @@ -137,9 +137,9 @@ fn decode_rate(attempt: &Attempt) -> Option<f64> {
if tokens < 2 || settle <= first {
return None;
}
Some((tokens - 1) as f64 * 1_000_000.0 / (settle - first) as f64)
Some((tokens - 1, settle - first))
}
fn prefill_rate(attempt: &Attempt) -> Option<f64> {
pub(crate) fn prefill_sample(attempt: &Attempt) -> Option<(u64, u64)> {
if attempt.status != Status::Complete
|| attempt.usage.cached_prompt_tokens.is_some_and(|n| n > 0)
{
Expand All @@ -150,16 +150,46 @@ fn prefill_rate(attempt: &Attempt) -> Option<f64> {
if tokens == 0 || first == 0 {
return None;
}
Some(tokens as f64 * 1_000_000.0 / first as f64)
Some((tokens, first))
}
fn decode_rate(attempt: &Attempt) -> Option<f64> {
decode_sample(attempt).map(|(n, d)| n as f64 * 1_000_000.0 / d as f64)
}
fn prefill_rate(attempt: &Attempt) -> Option<f64> {
prefill_sample(attempt).map(|(n, d)| n as f64 * 1_000_000.0 / d as f64)
}
pub struct Loaded {
pub plan: Plan,
pub waves: Vec<Option<Wave>>,
pub states: Vec<&'static str>,
pub history: crate::lifecycle::History,
pub metrics: Option<crate::metrics::Summary>,
pub policy: Option<crate::policy::Policy>,
pub plan_sha256: String,
pub evidence_sha256: String,
pub lineage_sha256: String,
}
pub fn load(root: &Path) -> Result<Loaded> {
load_verified(root).map_err(|error| error.detail)
}
pub(crate) struct LoadError {
pub reason: crate::policy::Reason,
pub detail: String,
}
impl From<String> for LoadError {
fn from(detail: String) -> Self {
Self {
reason: crate::policy::Reason::InvalidEvidence,
detail,
}
}
}
impl From<&str> for LoadError {
fn from(detail: &str) -> Self {
detail.to_owned().into()
}
}
pub(crate) fn load_verified(root: &Path) -> Result<Loaded, LoadError> {
directory(root)?;
let plan_bytes = read(&root.join("plan.json"), 8 * 1024 * 1024)?;
let plan: Plan = decode(&plan_bytes)?;
Expand Down Expand Up @@ -212,22 +242,53 @@ pub fn load(root: &Path) -> Result<Loaded> {
{
return Err("workload or schedule identity mismatch".into());
}
let policy = plan
.policy_sha256
.as_ref()
.map(|hash| {
let bytes = read(&root.join("policy.json"), crate::policy::CAP).map_err(|detail| {
LoadError {
reason: crate::policy::Reason::InvalidPolicy,
detail,
}
})?;
if digest(&bytes) != *hash {
return Err(LoadError {
reason: crate::policy::Reason::PolicyHashMismatch,
detail: "policy sidecar hash mismatch".into(),
});
}
crate::policy::parse(&bytes, &plan).map_err(|reason| LoadError {
detail: reason.as_str().into(),
reason,
})
})
.transpose()?;
crate::wire::endpoint(&plan.endpoint, plan.local_http)?;
let plan_hash = digest(&plan_bytes);
let mut fingerprint = Sha256::new();
fingerprint.update(b"grill-perf-evidence-v1\0");
fingerprint.update(plan_hash.as_bytes());
fingerprint.update(plan.source_sha256.as_bytes());
let mut lineage = Sha256::new();
lineage.update(b"grill-perf-acquisition-lineage-v1\0");
let mut waves = Vec::with_capacity(plan.waves.len());
let mut states = Vec::with_capacity(plan.waves.len());
let mut metrics_budget = crate::metrics::Budget::default();
let mut metrics_waves = Vec::new();
for spec in &plan.waves {
let dir = wave_dir(root, spec.index);
if !exists(&dir)? {
fingerprint.update(b"not_started\0");
waves.push(None);
states.push("not_started");
continue;
}
directory(&dir)?;
let reservation_bytes = read(&dir.join("reservation.json"), 40 * 1024 * 1024)?;
let reservation: Reservation = decode(&reservation_bytes)?;
let reservation_hash = digest(&reservation_bytes);
fingerprint.update(reservation_hash.as_bytes());
if reservation.version != 1
|| reservation.plan_sha256 != plan_hash
|| reservation.wave != *spec
Expand All @@ -251,14 +312,33 @@ pub fn load(root: &Path) -> Result<Loaded> {
}
let path = dir.join("wave.json");
if !exists(&path)? {
fingerprint.update(b"reserved_unsettled\0");
waves.push(None);
states.push("reserved_unsettled");
continue;
}
let wave: Wave = decode(&read(&path, FILE_CAP)?)?;
let wave_bytes = read(&path, FILE_CAP)?;
fingerprint.update(b"published\0");
fingerprint.update(digest(&wave_bytes).as_bytes());
let wave: Wave = decode(&wave_bytes)?;
lineage.update(
digest(
&serde_json::to_vec(&(
&wave.spec,
&wave.attempts,
wave.elapsed_us,
wave.dispatch_spread_us,
wave.preparation_us,
wave.reservation_publication_us,
wave.body_publication_us,
))
.map_err(|e| e.to_string())?,
)
.as_bytes(),
);
if wave.version != 1
|| wave.plan_sha256 != plan_hash
|| wave.reservation_sha256 != digest(&reservation_bytes)
|| wave.reservation_sha256 != reservation_hash
|| wave.spec != *spec
|| wave.attempts.len() != spec.concurrency as usize
{
Expand Down Expand Up @@ -363,6 +443,8 @@ pub fn load(root: &Path) -> Result<Loaded> {
states.push("published");
}
let history = crate::lifecycle::history(root, &plan, &plan_hash, &states, &waves)?;
fingerprint.update(history.evidence_sha256.as_bytes());
lineage.update(history.lineage_sha256.as_bytes());
Ok(Loaded {
metrics: plan.metrics.as_ref().map(|config| crate::metrics::Summary {
config: config.clone(),
Expand All @@ -373,6 +455,10 @@ pub fn load(root: &Path) -> Result<Loaded> {
waves,
states,
history,
policy,
plan_sha256: plan_hash,
evidence_sha256: hex(&fingerprint.finalize()),
lineage_sha256: hex(&lineage.finalize()),
})
}
#[derive(Serialize)]
Expand Down Expand Up @@ -646,7 +732,7 @@ pub struct ReferenceIdentity {
pub reasons: Vec<String>,
pub scope: &'static str,
}
fn reference_identity(a: &Plan, a2: &Plan) -> ReferenceIdentity {
pub(crate) fn reference_identity(a: &Plan, a2: &Plan) -> ReferenceIdentity {
let mut reasons = Vec::new();
let mut mismatch = false;
let mut declaration = |name: &str, a: Option<&str>, a2: Option<&str>| match (
Expand Down Expand Up @@ -718,6 +804,15 @@ fn complete_lane_observations(values: &[Vec<Option<f64>>]) -> bool {
values.iter().flatten().all(Option::is_some)
}

pub(crate) fn compatible(a: &Plan, b: &Plan) -> bool {
a.workload_sha256 == b.workload_sha256
&& a.tool_version == b.tool_version
&& a.collector_sha256 == b.collector_sha256
&& a.local_http == b.local_http
&& a.pool_max_idle_per_host == b.pool_max_idle_per_host
&& a.metrics == b.metrics
}

pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result<Comparison> {
let left = load(a)?;
let right = load(b)?;
Expand All @@ -727,13 +822,7 @@ pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result<Compariso
for (side, other) in
std::iter::once(("candidate", &right)).chain(reference.iter().map(|run| ("reference", run)))
{
if left.plan.workload_sha256 != other.plan.workload_sha256
|| left.plan.tool_version != other.plan.tool_version
|| left.plan.collector_sha256 != other.plan.collector_sha256
|| left.plan.local_http != other.plan.local_http
|| left.plan.pool_max_idle_per_host != other.plan.pool_max_idle_per_host
|| left.plan.metrics != other.plan.metrics
{
if !compatible(&left.plan, &other.plan) {
return Err(format!(
"{side}: incompatible workload, tool, transport or metrics controls; no matched comparison produced"
));
Expand Down
22 changes: 20 additions & 2 deletions crates/grill-perf/src/lifecycle.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
use crate::{evidence, model::*, run::Summary};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::fs::{self, File, OpenOptions};
use std::os::fd::AsRawFd;
use std::os::unix::fs::OpenOptionsExt;
Expand Down Expand Up @@ -55,6 +56,8 @@ pub struct History {
pub next_wave: usize,
pub last_status: Option<String>,
pub open: bool,
pub evidence_sha256: String,
pub lineage_sha256: String,
}

pub fn dir(root: &Path, index: usize) -> PathBuf {
Expand Down Expand Up @@ -115,6 +118,8 @@ pub fn history(
next_wave: 0,
last_status: None,
open: false,
evidence_sha256: evidence::digest(b"legacy-session-absent"),
lineage_sha256: evidence::digest(b"legacy-session-absent"),
});
}
if indices.is_empty() {
Expand All @@ -124,15 +129,20 @@ pub fn history(
let mut last_status = None;
let mut open = false;
let mut publication_uncertain = false;
let mut fingerprint = Sha256::new();
let mut lineage = Sha256::new();
for (expected, index) in indices.iter().copied().enumerate() {
if index != expected || open {
return Err("noncontiguous or overlapping execution sessions".into());
}
let path = dir(root, index);
evidence::directory(&path)?;
let session_bytes = evidence::read(&path.join("session.json"), FILE_CAP)?;
fingerprint.update(evidence::digest(&session_bytes).as_bytes());
let session: Session =
serde_json::from_slice(&evidence::read(&path.join("session.json"), FILE_CAP)?)
.map_err(|e| format!("invalid session: {e}"))?;
serde_json::from_slice(&session_bytes).map_err(|e| format!("invalid session: {e}"))?;
lineage.update(session.started_unix_ms.to_le_bytes());
lineage.update((session.first_wave as u64).to_le_bytes());
if session.version != 1
|| session.index != index
|| session.first_wave != next_wave
Expand All @@ -146,18 +156,23 @@ pub fn history(
}
if exists(&path.join("run.json"))? {
let end_bytes = evidence::read(&path.join("run.json"), FILE_CAP)?;
fingerprint.update(b"settled\0");
fingerprint.update(evidence::digest(&end_bytes).as_bytes());
lineage.update(evidence::digest(&end_bytes).as_bytes());
let end: Summary = serde_json::from_slice(&end_bytes)
.map_err(|e| format!("invalid session outcome: {e}"))?;
if index == 0 {
let receipt = root.join("run.json");
if exists(&receipt)? {
fingerprint.update(b"root-present\0");
if evidence::read(&receipt, FILE_CAP)? != end_bytes {
return Err("first root receipt differs from session zero outcome".into());
}
} else {
// Session publication precedes root publication. Do not repair or
// mistake this crash window (or later loss) for settled evidence.
publication_uncertain = true;
fingerprint.update(b"root-missing\0");
}
}
if end.version != 1
Expand Down Expand Up @@ -217,6 +232,7 @@ pub fn history(
next_wave = end_wave;
last_status = Some(end.status);
} else {
fingerprint.update(b"unsettled\0");
open = true;
last_status = None;
}
Expand All @@ -238,6 +254,8 @@ pub fn history(
next_wave,
last_status,
open,
evidence_sha256: evidence::hex(&fingerprint.finalize()),
lineage_sha256: evidence::hex(&lineage.finalize()),
})
}

Expand Down
Loading