From 409e2971be294044807cbb67f76c3e5468b3da77 Mon Sep 17 00:00:00 2001 From: Przemyslaw Olszewski Date: Sun, 30 Aug 2026 11:25:43 +0200 Subject: [PATCH] fix: project only changed run state --- diff --git a/src/projection.rs b/src/projection.rs --- a/src/projection.rs +++ b/src/projection.rs @@ -1,39 +1,53 @@ -use std::collections::HashMap; -use std::time::Duration; - -use syncode_control_node::ProjectionClient; -use syncode_control_runs::{ProjectedRun, RunId, RunLog, Runs, RunsError}; -use thiserror::Error; - -#[derive(Debug, Error)] -pub enum ProjectionError { - #[error("cannot read the run projection: {0}")] - Runs(#[from] RunsError), -} - -pub async fn serve( - runs: Runs, - client: ProjectionClient, - min_run_number: u64, -) -> Result<(), ProjectionError> { - let mut delivered = HashMap::::new(); - loop { - for projection in runs.projections().await? { - if projection.number.get() < min_run_number { - continue; - } - if delivered.get(&projection.run) == Some(&projection) { - continue; - } - match client.send(&projection).await { - Ok(()) => { - delivered.insert(projection.run, projection); - } - Err(error) => { - eprintln!("cannot project run state: {error}"); - } - } - } - tokio::time::sleep(Duration::from_millis(250)).await; - } -} +use std::collections::HashMap; +use std::time::Duration; + +use syncode_control_node::ProjectionClient; +use syncode_control_runs::{ProjectedRun, RunId, RunLog, Runs, RunsError}; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum ProjectionError { + #[error("cannot read the run projection: {0}")] + Runs(#[from] RunsError), + #[error("pending projection for run {0} is missing")] + MissingPending(RunId), +} + +pub async fn serve( + runs: Runs, + client: ProjectionClient, + min_run_number: u64, +) -> Result<(), ProjectionError> { + let mut delivered = HashMap::::new(); + let mut pending = HashMap::::new(); + loop { + for (run, number, sequence, output_version) in runs.projection_versions().await { + if number.get() < min_run_number { + continue; + } + let version = (sequence, output_version); + if delivered.get(&run) == Some(&version) { + continue; + } + if pending.get(&run).map(|(version, _)| *version) != Some(version) { + let Some(projection) = runs.projection(run).await? else { + continue; + }; + pending.insert(run, (version, projection)); + } + let (_, projection) = pending + .get(&run) + .ok_or(ProjectionError::MissingPending(run))?; + match client.send(projection).await { + Ok(()) => { + delivered.insert(run, version); + pending.remove(&run); + } + Err(error) => { + eprintln!("cannot project run state: {error}"); + } + } + } + tokio::time::sleep(Duration::from_millis(250)).await; + } +} diff --git a/tests/projection.rs b/tests/projection.rs --- a/tests/projection.rs +++ b/tests/projection.rs @@ -1,67 +1,110 @@ -#![allow(clippy::expect_used)] - -use std::error::Error; - -use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, ProjectedJobState, RunId, Runs}; - -type TestResult = Result<(), Box>; - -#[tokio::test] -async fn projection_carries_the_current_runtime_authority() -> TestResult { - let runs = Runs::restored(Forgotten::default()).await?; - let job = JobId::fresh(); - let origin = Origin::new( - "syncode/meta".to_owned(), - "commit".to_owned(), - "refs/heads/main".to_owned(), - "push".to_owned(), - ".gitea/workflows/ci.yml".to_owned(), - ) - .with_delivery(Some("delivery".to_owned())); - runs.queue(job, origin, b"plan".to_vec()).await?; - let node = NodeId::fresh(); - let assignment = runs.take_next(node).await?.expect("assignment"); - - let projection = runs.projections().await?.pop().expect("projection"); - let projected = projection.jobs.first().expect("projected job"); - assert_eq!(projected.state, ProjectedJobState::Assigned); - assert_eq!(projected.node, Some(node)); - assert_eq!(projected.fence, Some(assignment.fence())); - - Ok(()) -} - -#[tokio::test] -async fn one_run_can_be_projected_without_sweeping_every_run() -> TestResult { - let runs = Runs::restored(Forgotten::default()).await?; - let origin = Origin::new( - "syncode/meta".to_owned(), - "commit".to_owned(), - "refs/heads/main".to_owned(), - "push".to_owned(), - ".gitea/workflows/ci.yml".to_owned(), - ) - .with_delivery(Some("delivery".to_owned())); - let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?; - - let alone = runs.projection(run).await?.expect("projection"); - let swept = runs.projections().await?; - assert_eq!( - swept.len(), - 1, - "the run added here is the only one to sweep" - ); - assert_eq!( - alone, swept[0], - "one run's projection must match the sweep's" - ); - - Ok(()) -} - -#[tokio::test] -async fn a_run_nobody_queued_has_nothing_to_project() -> TestResult { - let runs = Runs::restored(Forgotten::default()).await?; - assert_eq!(runs.projection(RunId::fresh()).await?, None); - Ok(()) -} +#![allow(clippy::expect_used)] + +use std::error::Error; + +use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, ProjectedJobState, RunId, Runs}; + +type TestResult = Result<(), Box>; + +#[tokio::test] +async fn projection_carries_the_current_runtime_authority() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let job = JobId::fresh(); + let origin = Origin::new( + "syncode/meta".to_owned(), + "commit".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ) + .with_delivery(Some("delivery".to_owned())); + runs.queue(job, origin, b"plan".to_vec()).await?; + let node = NodeId::fresh(); + let assignment = runs.take_next(node).await?.expect("assignment"); + + let projection = runs.projections().await?.pop().expect("projection"); + let projected = projection.jobs.first().expect("projected job"); + assert_eq!(projected.state, ProjectedJobState::Assigned); + assert_eq!(projected.node, Some(node)); + assert_eq!(projected.fence, Some(assignment.fence())); + + Ok(()) +} + +#[tokio::test] +async fn one_run_can_be_projected_without_sweeping_every_run() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let origin = Origin::new( + "syncode/meta".to_owned(), + "commit".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ) + .with_delivery(Some("delivery".to_owned())); + let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?; + + let alone = runs.projection(run).await?.expect("projection"); + let swept = runs.projections().await?; + assert_eq!( + swept.len(), + 1, + "the run added here is the only one to sweep" + ); + assert_eq!( + alone, swept[0], + "one run's projection must match the sweep's" + ); + + Ok(()) +} + +#[tokio::test] +async fn a_run_nobody_queued_has_nothing_to_project() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + assert_eq!(runs.projection(RunId::fresh()).await?, None); + Ok(()) +} + +#[tokio::test] +async fn projection_versions_change_without_building_every_projection() -> TestResult { + let runs = Runs::restored(Forgotten::default()).await?; + let origin = Origin::new( + "syncode/meta".to_owned(), + "commit".to_owned(), + "refs/heads/main".to_owned(), + "push".to_owned(), + ".gitea/workflows/ci.yml".to_owned(), + ) + .with_delivery(Some("delivery".to_owned())); + let run = runs.queue(JobId::fresh(), origin, b"plan".to_vec()).await?; + + let initial = runs.projection_versions().await; + let initial_sequence = initial[0].2; + let initial_output = initial[0].3; + assert_eq!(initial[0].0, run); + + let node = NodeId::fresh(); + let assignment = runs.take_next(node).await?.expect("assignment"); + + let assigned = runs.projection_versions().await; + assert_eq!(assigned[0].0, run); + assert!(assigned[0].2 > initial_sequence); + assert_eq!(assigned[0].3, initial_output); + + runs.logged( + run, + assignment.job(), + node, + assignment.fence(), + 0, + &["line".to_owned()], + ) + .await?; + + let logged = runs.projection_versions().await; + assert_eq!(logged[0].2, assigned[0].2); + assert!(logged[0].3 > assigned[0].3); + + Ok(()) +} diff --git a/crates/control-runs/src/output.rs b/crates/control-runs/src/output.rs --- a/crates/control-runs/src/output.rs +++ b/crates/control-runs/src/output.rs @@ -1,35 +1,37 @@ -//! What a job printed. -//! -//! Output is not a state change: a job that prints nothing is not less valid, -//! and a line arriving late does not move the run anywhere. So logs travel -//! beside the aggregate rather than through it. - -use crate::{Fence, JobId, NodeId, RunId, RunLog, Runs, RunsError}; - -impl Runs { - /// Keep what a node said a job printed. The offset makes a repeated batch - /// the same fact twice rather than new output. - pub async fn logged( - &self, - run: RunId, - job: JobId, - node: NodeId, - fence: Fence, - offset: u64, - lines: &[String], - ) -> Result<(), RunsError> { - self.state - .lock() - .await - .run(run)? - .validate_holder(job, node, fence)?; - self.store - .logged(run, job, offset, lines) - .await - .map_err(RunsError::log) - } - - pub async fn lines(&self, run: RunId, job: JobId) -> Result, RunsError> { - self.store.lines(run, job).await.map_err(RunsError::log) - } -} +//! What a job printed. +//! +//! Output is not a state change: a job that prints nothing is not less valid, +//! and a line arriving late does not move the run anywhere. So logs travel +//! beside the aggregate rather than through it. + +use crate::{Fence, JobId, NodeId, RunId, RunLog, Runs, RunsError}; + +impl Runs { + /// Keep what a node said a job printed. The offset makes a repeated batch + /// the same fact twice rather than new output. + pub async fn logged( + &self, + run: RunId, + job: JobId, + node: NodeId, + fence: Fence, + offset: u64, + lines: &[String], + ) -> Result<(), RunsError> { + self.state + .lock() + .await + .run(run)? + .validate_holder(job, node, fence)?; + self.store + .logged(run, job, offset, lines) + .await + .map_err(RunsError::log)?; + self.state.lock().await.record_output(run); + Ok(()) + } + + pub async fn lines(&self, run: RunId, job: JobId) -> Result, RunsError> { + self.store.lines(run, job).await.map_err(RunsError::log) + } +} diff --git a/crates/control-runs/src/registry/projection.rs b/crates/control-runs/src/registry/projection.rs --- a/crates/control-runs/src/registry/projection.rs +++ b/crates/control-runs/src/registry/projection.rs @@ -1,53 +1,57 @@ -use crate::{ProjectedJob, ProjectedRun, RunId, RunLog}; - -use super::state::ProjectionSeed; -use super::{Runs, RunsError}; - -impl Runs { - pub async fn projections(&self) -> Result, RunsError> { - let seeds = self.state.lock().await.projection_seeds(); - let mut projections = Vec::with_capacity(seeds.len()); - for seed in seeds { - projections.push(self.project(seed).await?); - } - Ok(projections) - } - - /// The current projection of one run, if it has anything to project. Lets - /// a caller push a single run's state out-of-band from the periodic sweep - /// — namely dispatch, which needs the projection to have landed before it - /// hands a node a token that is only valid once it has. - pub async fn projection(&self, run: RunId) -> Result, RunsError> { - let Some(seed) = self.state.lock().await.projection_seed(run) else { - return Ok(None); - }; - Ok(Some(self.project(seed).await?)) - } - - async fn project(&self, seed: ProjectionSeed) -> Result { - let mut jobs = Vec::with_capacity(seed.jobs.len()); - for source in seed.jobs { - let logs = self - .store - .lines(seed.run, source.job) - .await - .map_err(RunsError::log)?; - jobs.push(ProjectedJob { - job: source.job, - key: source.key, - needs: source.needs, - state: source.state, - node: source.node, - fence: source.fence, - logs, - }); - } - Ok(ProjectedRun { - run: seed.run, - number: seed.number, - origin: seed.origin, - sequence: seed.sequence, - jobs, - }) - } -} +use crate::{ProjectedJob, ProjectedRun, RunId, RunLog}; + +use super::state::ProjectionSeed; +use super::{Runs, RunsError}; + +impl Runs { + pub async fn projection_versions(&self) -> Vec<(RunId, crate::RunNumber, u64, u64)> { + self.state.lock().await.projection_versions() + } + + pub async fn projections(&self) -> Result, RunsError> { + let seeds = self.state.lock().await.projection_seeds(); + let mut projections = Vec::with_capacity(seeds.len()); + for seed in seeds { + projections.push(self.project(seed).await?); + } + Ok(projections) + } + + /// The current projection of one run, if it has anything to project. Lets + /// a caller push a single run's state out-of-band from the periodic sweep + /// — namely dispatch, which needs the projection to have landed before it + /// hands a node a token that is only valid once it has. + pub async fn projection(&self, run: RunId) -> Result, RunsError> { + let Some(seed) = self.state.lock().await.projection_seed(run) else { + return Ok(None); + }; + Ok(Some(self.project(seed).await?)) + } + + async fn project(&self, seed: ProjectionSeed) -> Result { + let mut jobs = Vec::with_capacity(seed.jobs.len()); + for source in seed.jobs { + let logs = self + .store + .lines(seed.run, source.job) + .await + .map_err(RunsError::log)?; + jobs.push(ProjectedJob { + job: source.job, + key: source.key, + needs: source.needs, + state: source.state, + node: source.node, + fence: source.fence, + logs, + }); + } + Ok(ProjectedRun { + run: seed.run, + number: seed.number, + origin: seed.origin, + sequence: seed.sequence, + jobs, + }) + } +} diff --git a/crates/control-runs/src/registry/state.rs b/crates/control-runs/src/registry/state.rs --- a/crates/control-runs/src/registry/state.rs +++ b/crates/control-runs/src/registry/state.rs @@ -1,239 +1,190 @@ -mod diagnosis; -mod scheduling; - -use std::collections::BTreeMap; - -use crate::Fence; -use crate::{ - Assignment, JobId, JobState, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError, - SchedulerPolicy, -}; -use crate::{Lease, NodeId}; - -pub struct Entry { - run: Run, - number: RunNumber, - origin: Origin, -} - -pub struct State { - entries: Vec, - organization_turns: BTreeMap, - next_turn: u64, -} - -pub(super) struct ProjectionSeed { - pub run: RunId, - pub number: RunNumber, - pub origin: Origin, - pub sequence: u64, - pub jobs: Vec, -} - -pub(super) struct ProjectionJobSeed { - pub job: JobId, - pub key: String, - pub needs: Vec, - pub state: ProjectedJobState, - pub node: Option, - pub fence: Option, -} - -impl Default for State { - fn default() -> Self { - Self { - entries: Vec::new(), - organization_turns: BTreeMap::new(), - next_turn: 1, - } - } -} - -impl State { - pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) { - self.entries.push(Entry { - run, - number, - origin, - }); - } - - pub fn next_number(&self) -> RunNumber { - self.entries - .iter() - .map(|entry| entry.number) - .max() - .map_or_else(RunNumber::first, RunNumber::after) - } - - pub fn run_by_origin(&self, origin: &Origin) -> Option { - self.entries - .iter() - .find(|entry| &entry.origin == origin) - .map(|entry| entry.run.id()) - } - - pub fn next_queued(&self) -> Option<(RunId, JobId)> { - self.entries.iter().find_map(|entry| { - entry - .run - .queued_jobs() - .next() - .map(|job| (entry.run.id(), job.id())) - }) - } - - pub fn admits_queue( - &self, - origin: &Origin, - jobs: usize, - policy: SchedulerPolicy, - ) -> Result<(), RunsError> { - let organization_open: usize = self - .entries - .iter() - .filter(|entry| entry.origin.organization() == origin.organization()) - .map(|entry| entry.run.open_jobs()) - .sum(); - if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize { - return Err(RunsError::OrganizationQueueQuota { - organization: origin.organization().to_owned(), - limit: policy.organization_queue_quota(), - }); - } - if let Some(principal) = origin.principal() { - let principal_open: usize = self - .entries - .iter() - .filter(|entry| entry.origin.principal() == Some(principal)) - .map(|entry| entry.run.open_jobs()) - .sum(); - if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize { - return Err(RunsError::PrincipalQueueQuota { - principal: principal.to_owned(), - limit: policy.principal_queue_quota(), - }); - } - } - Ok(()) - } - - pub fn assignment(&self, run: RunId, job: JobId) -> Result { - let entry = self.entry(run)?; - let definition = entry.run.job_definition(job)?; - let needs = entry.run.dependencies(job)?; - Ok(Assignment::new( - run, - job, - entry.number, - entry.origin.clone(), - definition, - needs, - )) - } - - pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> { - let organization = self.entry(run)?.origin.organization().to_owned(); - self.organization_turns.insert(organization, self.next_turn); - self.next_turn = self.next_turn.saturating_add(1); - Ok(()) - } - - pub fn runs(&self) -> impl Iterator { - self.entries.iter().map(|entry| &entry.run) - } - - pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> { - self.entries - .iter() - .flat_map(|entry| { - entry - .run - .cancellations(node) - .map(move |(job, lease)| (entry.run.id(), job, lease)) - }) - .collect() - } - - pub fn leases_held_by(&self, node: NodeId) -> usize { - self.entries - .iter() - .flat_map(|entry| entry.run.leased_jobs()) - .filter(|(_, lease)| lease.node() == node) - .count() - } - - pub fn run(&self, id: RunId) -> Result<&Run, RunsError> { - self.entry(id).map(|entry| &entry.run) - } - - pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> { - self.entries - .iter_mut() - .find(|entry| entry.run.id() == id) - .map(|entry| &mut entry.run) - .ok_or(RunsError::UnknownRun(id)) - } - - pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> { - self.entry(id).map(|entry| &entry.origin) - } - - pub(super) fn projection_seeds(&self) -> Vec { - self.entries - .iter() - .filter(|entry| entry.origin.delivery().is_some()) - .map(Self::projection_seed_of) - .collect() - } - - /// The seed for one run, if it exists and is eligible for projection. Used - /// to push a single run's state synchronously (at dispatch, so the token a - /// node just received is already valid where it will be spent) alongside - /// `projection_seeds`, which the periodic sweep uses for everything else. - pub(super) fn projection_seed(&self, run: RunId) -> Option { - self.entry(run) - .ok() - .filter(|entry| entry.origin.delivery().is_some()) - .map(Self::projection_seed_of) - } - - fn projection_seed_of(entry: &Entry) -> ProjectionSeed { - ProjectionSeed { - run: entry.run.id(), - number: entry.number, - origin: entry.origin.clone(), - sequence: entry.run.sequence().get(), - jobs: entry - .run - .projected_jobs() - .map(|(job, state)| { - let (state, authority) = match state { - JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None), - JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)), - JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)), - JobState::Finished(conclusion) => { - (ProjectedJobState::Finished(conclusion), None) - } - JobState::Skipped => (ProjectedJobState::Skipped, None), - }; - ProjectionJobSeed { - job: job.id(), - key: job.key().to_owned(), - needs: job.needs().to_vec(), - state, - node: authority.map(|lease| lease.node()), - fence: authority.map(|lease| lease.fence()), - } - }) - .collect(), - } - } - - fn entry(&self, id: RunId) -> Result<&Entry, RunsError> { - self.entries - .iter() - .find(|entry| entry.run.id() == id) - .ok_or(RunsError::UnknownRun(id)) - } -} +mod diagnosis; +mod projection; +mod scheduling; + +use std::collections::BTreeMap; + +use crate::Fence; +use crate::{ + Assignment, JobId, Origin, ProjectedJobState, Run, RunId, RunNumber, RunsError, SchedulerPolicy, +}; +use crate::{Lease, NodeId}; + +pub struct Entry { + run: Run, + number: RunNumber, + origin: Origin, +} + +pub struct State { + entries: Vec, + output_versions: BTreeMap, + organization_turns: BTreeMap, + next_turn: u64, +} + +pub(super) struct ProjectionSeed { + pub run: RunId, + pub number: RunNumber, + pub origin: Origin, + pub sequence: u64, + pub jobs: Vec, +} + +pub(super) struct ProjectionJobSeed { + pub job: JobId, + pub key: String, + pub needs: Vec, + pub state: ProjectedJobState, + pub node: Option, + pub fence: Option, +} + +impl Default for State { + fn default() -> Self { + Self { + entries: Vec::new(), + output_versions: BTreeMap::new(), + organization_turns: BTreeMap::new(), + next_turn: 1, + } + } +} + +impl State { + pub fn admit(&mut self, run: Run, number: RunNumber, origin: Origin) { + self.entries.push(Entry { + run, + number, + origin, + }); + } + + pub fn next_number(&self) -> RunNumber { + self.entries + .iter() + .map(|entry| entry.number) + .max() + .map_or_else(RunNumber::first, RunNumber::after) + } + + pub fn run_by_origin(&self, origin: &Origin) -> Option { + self.entries + .iter() + .find(|entry| &entry.origin == origin) + .map(|entry| entry.run.id()) + } + + pub fn next_queued(&self) -> Option<(RunId, JobId)> { + self.entries.iter().find_map(|entry| { + entry + .run + .queued_jobs() + .next() + .map(|job| (entry.run.id(), job.id())) + }) + } + + pub fn admits_queue( + &self, + origin: &Origin, + jobs: usize, + policy: SchedulerPolicy, + ) -> Result<(), RunsError> { + let organization_open: usize = self + .entries + .iter() + .filter(|entry| entry.origin.organization() == origin.organization()) + .map(|entry| entry.run.open_jobs()) + .sum(); + if organization_open.saturating_add(jobs) > policy.organization_queue_quota() as usize { + return Err(RunsError::OrganizationQueueQuota { + organization: origin.organization().to_owned(), + limit: policy.organization_queue_quota(), + }); + } + if let Some(principal) = origin.principal() { + let principal_open: usize = self + .entries + .iter() + .filter(|entry| entry.origin.principal() == Some(principal)) + .map(|entry| entry.run.open_jobs()) + .sum(); + if principal_open.saturating_add(jobs) > policy.principal_queue_quota() as usize { + return Err(RunsError::PrincipalQueueQuota { + principal: principal.to_owned(), + limit: policy.principal_queue_quota(), + }); + } + } + Ok(()) + } + + pub fn assignment(&self, run: RunId, job: JobId) -> Result { + let entry = self.entry(run)?; + let definition = entry.run.job_definition(job)?; + let needs = entry.run.dependencies(job)?; + Ok(Assignment::new( + run, + job, + entry.number, + entry.origin.clone(), + definition, + needs, + )) + } + + pub fn record_dispatch(&mut self, run: RunId) -> Result<(), RunsError> { + let organization = self.entry(run)?.origin.organization().to_owned(); + self.organization_turns.insert(organization, self.next_turn); + self.next_turn = self.next_turn.saturating_add(1); + Ok(()) + } + + pub fn runs(&self) -> impl Iterator { + self.entries.iter().map(|entry| &entry.run) + } + + pub fn cancellations(&self, node: NodeId) -> Vec<(RunId, JobId, Lease)> { + self.entries + .iter() + .flat_map(|entry| { + entry + .run + .cancellations(node) + .map(move |(job, lease)| (entry.run.id(), job, lease)) + }) + .collect() + } + + pub fn leases_held_by(&self, node: NodeId) -> usize { + self.entries + .iter() + .flat_map(|entry| entry.run.leased_jobs()) + .filter(|(_, lease)| lease.node() == node) + .count() + } + + pub fn run(&self, id: RunId) -> Result<&Run, RunsError> { + self.entry(id).map(|entry| &entry.run) + } + + pub fn run_mut(&mut self, id: RunId) -> Result<&mut Run, RunsError> { + self.entries + .iter_mut() + .find(|entry| entry.run.id() == id) + .map(|entry| &mut entry.run) + .ok_or(RunsError::UnknownRun(id)) + } + + pub fn origin(&self, id: RunId) -> Result<&Origin, RunsError> { + self.entry(id).map(|entry| &entry.origin) + } + + fn entry(&self, id: RunId) -> Result<&Entry, RunsError> { + self.entries + .iter() + .find(|entry| entry.run.id() == id) + .ok_or(RunsError::UnknownRun(id)) + } +} diff --git a/crates/control-runs/src/registry/state/projection.rs b/crates/control-runs/src/registry/state/projection.rs --- /dev/null +++ b/crates/control-runs/src/registry/state/projection.rs @@ -1,0 +1,78 @@ +use super::{Entry, ProjectionJobSeed, ProjectionSeed, State}; +use crate::{JobState, ProjectedJobState, RunId, RunNumber}; + +impl State { + pub(in crate::registry) fn projection_seeds(&self) -> Vec { + self.entries + .iter() + .filter(|entry| entry.origin.delivery().is_some()) + .map(Self::projection_seed_of) + .collect() + } + + pub(in crate::registry) fn projection_versions(&self) -> Vec<(RunId, RunNumber, u64, u64)> { + self.entries + .iter() + .filter(|entry| entry.origin.delivery().is_some()) + .map(|entry| { + ( + entry.run.id(), + entry.number, + entry.run.sequence().get(), + self.output_versions + .get(&entry.run.id()) + .copied() + .unwrap_or(0), + ) + }) + .collect() + } + + pub(crate) fn record_output(&mut self, run: RunId) { + let version = self.output_versions.entry(run).or_default(); + *version = version.saturating_add(1); + } + + /// The seed for one run, if it exists and is eligible for projection. Used + /// to push a single run's state synchronously (at dispatch, so the token a + /// node just received is already valid where it will be spent) alongside + /// `projection_seeds`, which the periodic sweep uses for everything else. + pub(in crate::registry) fn projection_seed(&self, run: RunId) -> Option { + self.entry(run) + .ok() + .filter(|entry| entry.origin.delivery().is_some()) + .map(Self::projection_seed_of) + } + + fn projection_seed_of(entry: &Entry) -> ProjectionSeed { + ProjectionSeed { + run: entry.run.id(), + number: entry.number, + origin: entry.origin.clone(), + sequence: entry.run.sequence().get(), + jobs: entry + .run + .projected_jobs() + .map(|(job, state)| { + let (state, authority) = match state { + JobState::Waiting | JobState::Queued => (ProjectedJobState::Waiting, None), + JobState::Assigned(lease) => (ProjectedJobState::Assigned, Some(lease)), + JobState::Running(lease) => (ProjectedJobState::Running, Some(lease)), + JobState::Finished(conclusion) => { + (ProjectedJobState::Finished(conclusion), None) + } + JobState::Skipped => (ProjectedJobState::Skipped, None), + }; + ProjectionJobSeed { + job: job.id(), + key: job.key().to_owned(), + needs: job.needs().to_vec(), + state, + node: authority.map(|lease| lease.node()), + fence: authority.map(|lease| lease.fence()), + } + }) + .collect(), + } + } +} -- SynCode