syncode/control
2
0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: expose actions state and publish check events #42

Manually merged
day01 merged 1 commits from feat/0.6-actions-read into develop 2026-08-30 09:27:12 +00:00
Conversation 0 Commits 1 Files Changed 21
+1282 -11
21 changed files with 1282 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -1,76 +1,76 @@
[package]
name = "syncode-control"
description = "SynCode control plane"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
license.workspace = true
repository.workspace = true
publish = false

[workspace]
members = [
".",
"crates/control-node",
"crates/control-nodes",
"crates/control-runs",
"crates/control-store",
]
resolver = "3"

[workspace.package]
version = "0.5.0"
edition = "2024"
rust-version = "1.95"
license = "MIT"
repository = "https://syncode.sh/syncode/control"

[workspace.lints.rust]
unsafe_code = "forbid"

[workspace.lints.clippy]
expect_used = "deny"
panic = "deny"
unwrap_used = "deny"

[dependencies]
base64 = "0.22.1"
clap = { version = "4.6.4", features = ["derive", "env"] }
flate2 = "1.1.5"
futures = "0.3.31"
hex = "0.4.3"
reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
serde = { version = "1.0.229", features = ["derive"] }
url = "2.5.8"
serde_json = "1.0.149"
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-control-node = { path = "crates/control-node" }
syncode-control-nodes = { path = "crates/control-nodes" }
syncode-control-runs = { path = "crates/control-runs" }
syncode-control-store = { version = "0.5.0", path = "crates/control-store" }
syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" }
thiserror = "2.0.19"
tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal"] }
tonic = "0.14.6"
axum = "0.8.9"
hmac = "0.13.0"
sha2 = "0.11.0"
tar = "0.4.44"
uuid = { version = "1.24.0", features = ["v4"] }

[dev-dependencies]
base64 = "0.22.1"
hmac = "0.13.0"
reqwest = { version = "0.13.4", features = ["json", "rustls"] }
serde_json = "1.0.149"
sha2 = "0.11.0"
sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] }
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
tokio-stream = "0.1.18"
tempfile = "3.27.0"
tower = { version = "0.5.3", features = ["util"] }

[lints]
workspace = true
[package]
name = "syncode-control"
description = "SynCode control plane"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
license.workspace = true
repository.workspace = true
publish = false

[workspace]
members = [
".",
"crates/control-node",
"crates/control-nodes",
"crates/control-runs",
"crates/control-store",
]
resolver = "3"

[workspace.package]
version = "0.5.0"
edition = "2024"
rust-version = "1.95"
license = "MIT"
repository = "https://syncode.sh/syncode/control"

[workspace.lints.rust]
unsafe_code = "forbid"

[workspace.lints.clippy]
expect_used = "deny"
panic = "deny"
unwrap_used = "deny"

[dependencies]
base64 = "0.22.1"
clap = { version = "4.6.4", features = ["derive", "env"] }
flate2 = "1.1.5"
futures = "0.3.31"
hex = "0.4.3"
reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
serde = { version = "1.0.229", features = ["derive"] }
url = "2.5.8"
serde_json = "1.0.149"
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-control-node = { path = "crates/control-node" }
syncode-control-nodes = { path = "crates/control-nodes" }
syncode-control-runs = { path = "crates/control-runs" }
syncode-control-store = { version = "0.5.0", path = "crates/control-store" }
syncode-repository-api-grpc = { git = "https://syncode.sh/syncode/repo.git", rev = "d293b2a4b948be059f61283ffc3ab2ec220a64e4" }
thiserror = "2.0.19"
tokio = { version = "1.49.0", features = ["fs", "macros", "net", "rt-multi-thread", "signal", "time"] }
tonic = "0.14.6"
axum = "0.8.9"
hmac = "0.13.0"
sha2 = "0.11.0"
tar = "0.4.44"
uuid = { version = "1.24.0", features = ["v4"] }

[dev-dependencies]
base64 = "0.22.1"
hmac = "0.13.0"
reqwest = { version = "0.13.4", features = ["json", "rustls"] }
serde_json = "1.0.149"
sha2 = "0.11.0"
sqlx = { version = "0.9.0", default-features = false, features = ["macros", "postgres", "runtime-tokio", "tls-rustls-ring-webpki", "uuid"] }
syncode-workflow = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
syncode-workflow-github-actions = { git = "https://syncode.sh/syncode/workflow.git", rev = "14de090ff1961286a947d34f9d81c091081233f1" }
tokio-stream = "0.1.18"
tempfile = "3.27.0"
tower = { version = "0.5.3", features = ["util"] }

[lints]
workspace = true
src/lib.rs
+3
View File
@@ -1,24 +1,27 @@
pub mod action_delivery;
pub mod action_oci;
pub mod action_repository;
pub mod action_store;
pub mod actions;
pub mod admin;
pub mod checks;
pub mod events;
pub mod intake;
pub mod maintenance;
mod native_events;
pub mod projection;
#[path = "forge.rs"]
pub mod repository;
pub mod repository_grpc;
pub mod repository_sources;
pub mod reusable;
mod secret_reference_syntax;
pub mod secret_references;
pub mod secrets;
pub mod sources;
pub mod token;
pub mod trigger;
pub mod webhook;
pub mod action_delivery;
pub mod action_oci;
pub mod action_repository;
pub mod action_store;
pub mod actions;
pub mod actions_read;
pub mod actions_read_identity;
pub mod admin;
pub mod check_events;
pub mod checks;
pub mod events;
pub mod intake;
pub mod maintenance;
mod native_events;
pub mod projection;
#[path = "forge.rs"]
pub mod repository;
pub mod repository_grpc;
pub mod repository_sources;
pub mod reusable;
mod secret_reference_syntax;
pub mod secret_references;
pub mod secrets;
pub mod sources;
pub mod token;
pub mod trigger;
pub mod webhook;
src/main.rs
+51 -3
View File
@@ -1,326 +1,374 @@
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use clap::{Parser, Subcommand, ValueEnum};
use syncode_control::action_delivery::router as action_delivery_router;
use syncode_control::action_oci::PinnedOciResolver;
use syncode_control::action_repository::ForgeActionRepository;
use syncode_control::action_store::FileActionStore;
use syncode_control::actions::ActionResolver;
use syncode_control::admin::{Admin, router as admin_router};
use syncode_control::checks::Checks;
use syncode_control::maintenance;
use syncode_control::projection;
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
use syncode_control::repository_grpc::NativeRepositoryContents;
use syncode_control::repository_sources::RepositorySources;
use syncode_control::secrets::RuntimeSecrets;
use syncode_control::token::EnrolmentScope;
use syncode_control::webhook::{Intake, router};
use syncode_control_node::{
ArtifactTokenAuthority, CapabilityAuthority, GeneratedChecksServer, GeneratedSecretsServer,
GeneratedServer, NodeSessionServer, ProjectionClient,
};
use syncode_control_nodes::{Nodes, Scope};
use syncode_control_runs::{
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
};
use syncode_control_store::Postgres;
use tokio::net::TcpListener;
use tonic::transport::Server;
use url::Url;

#[derive(Debug, Parser)]
#[command(name = "syncode-control", version, about)]
struct Arguments {
#[command(subcommand)]
command: Option<Command>,

/// Where nodes open their session.
#[arg(long, default_value = "127.0.0.1:8090")]
listen: SocketAddr,

/// Where the native repository event feed delivers events.
#[arg(long, default_value = "127.0.0.1:8091")]
listen_events: SocketAddr,

/// The repository service this control plane reads workflows from.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
repository_source: Url,

/// A token allowed to read repository contents.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
repository_source_token: String,

/// Where native repositories are read over gRPC.
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
repository_grpc: String,

#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
action_mirror_url: Url,

#[arg(
long,
env = "SYNCODE_ACTION_ALLOWLIST",
value_delimiter = ',',
num_args = 1..
)]
action_allowlist: Vec<Url>,

#[arg(
long,
env = "SYNCODE_ACTION_OCI_REGISTRIES",
value_delimiter = ',',
num_args = 1..
)]
action_oci_registries: Vec<String>,

#[arg(long, env = "SYNCODE_ACTION_STORE")]
action_store: PathBuf,

#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
action_artifact_public_url: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
secret_source: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
secret_source_token: String,

#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
capability_signing_key: String,

#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
artifact_signing_key: String,

#[arg(long, env = "SYNCODE_PROJECTION_URL")]
projection_url: Url,

#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
projection_token: String,

#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
projection_min_run_number: u64,

/// The secret used to sign the native event feed.
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
event_feed_secret: String,

/// Bearer token protecting the operational API.
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
admin_token: Option<String>,

/// Where the run log is kept.
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
database: String,

#[arg(long, default_value_t = 8)]
database_connections: u32,

#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
organization_concurrency: u32,

#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
repository_concurrency: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
principal_concurrency: u32,

#[arg(
long,
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
default_value_t = 10_000
)]
organization_queue_quota: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
principal_queue_quota: u32,

#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
audit_retention_days: u32,

/// Whether compiled shadow runs may be assigned to nodes.
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
mode: Mode,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
enum Mode {
Shadow,
Active,
}

#[derive(Debug, Subcommand)]
enum Command {
/// Issue an enrolment token a node can spend for an identity.
///
/// The token is stored before it is printed, so one that reaches an
/// operator is one the control plane will honour.
IssueToken {
/// How far the token reaches: `instance`, `organisation:<name>` or
/// `repository:<owner>/<name>`.
#[arg(long, default_value = "instance")]
scope: String,
},
/// Revoke a node identity immediately.
RevokeNode {
/// The node UUID printed at enrolment.
node: String,
},
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let arguments = Arguments::parse();
// The log is read before anything is served: a node that reconnects has to
// meet the run it was holding, not a control plane that has forgotten it.
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
let policy = SchedulerPolicy::new(
arguments.organization_concurrency,
arguments.repository_concurrency,
arguments.principal_concurrency,
arguments.organization_queue_quota,
arguments.principal_queue_quota,
);
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
let nodes = Nodes::restored(store.clone()).await?;

if let Some(command) = arguments.command {
match command {
Command::IssueToken { scope } => {
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
let token = nodes.issue_token(scope).await?;
println!("{}", token.secret().expose());
}
Command::RevokeNode { node } => {
nodes.revoke(node.parse::<NodeId>()?).await?;
}
}
return Ok(());
}

store
.record_audit(AuditEvent::new("control.configuration", "applied").after(
AuditPayload::Configuration(AuditConfiguration {
mode: match arguments.mode {
Mode::Shadow => AuditControlMode::Shadow,
Mode::Active => AuditControlMode::Active,
},
organization_concurrency: arguments.organization_concurrency,
repository_concurrency: arguments.repository_concurrency,
principal_concurrency: arguments.principal_concurrency,
organization_queue_quota: arguments.organization_queue_quota,
principal_queue_quota: arguments.principal_queue_quota,
audit_retention_days: arguments.audit_retention_days,
projection_min_run_number: arguments.projection_min_run_number,
}),
))
.await?;
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
tokio::spawn(maintenance::retain_audit(
store.clone(),
arguments.audit_retention_days,
));

let action_mirror = arguments
.action_mirror_url
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()?;
let action_allowlist = arguments
.action_allowlist
.into_iter()
.map(|repository| {
repository
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()
})
.collect::<Result<Vec<_>, _>>()?;
let action_store = FileActionStore::new(arguments.action_store);
let action_resolver = ActionResolver::new(
ForgeActionRepository::new(arguments.repository_source_token.clone()),
action_store.clone(),
PinnedOciResolver::new(arguments.action_oci_registries)?,
action_mirror,
action_allowlist,
);
let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).await?;
let repository_sources = RepositorySources::new(
native_repository,
RepositoryContents::new(
arguments.repository_source,
arguments.repository_source_token,
),
);
let intake = Arc::new(Intake::new(
repository_sources,
runs.clone(),
action_resolver,
arguments.event_feed_secret,
));
let admin_token = arguments
.admin_token
.filter(|token| !token.is_empty())
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
let http = router(intake)
.merge(admin_router(Arc::new(
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store),
)))
.merge(action_delivery_router(
action_store,
artifact_authority.clone(),
runs.clone(),
));
let events = TcpListener::bind(arguments.listen_events).await?;

let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?;
let secret_service = RuntimeSecrets::new(
runs.clone(),
nodes.clone(),
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
authority.clone(),
);

// Both ends of the control plane run for as long as the other does: without
// events there is nothing to assign, and without sessions there is nobody to
// assign it to. Whichever stops first takes the process down with it.
let node_service = match arguments.mode {
Mode::Shadow => NodeSessionServer::shadow(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
projection.clone(),
),
Mode::Active => NodeSessionServer::new(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
projection.clone(),
),
};
tokio::select! {
served = axum::serve(events, http).into_future() => served?,
served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?,
served = Server::builder()
.add_service(GeneratedServer::new(node_service))
.add_service(GeneratedSecretsServer::new(secret_service))
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
}

Ok(())
}

async fn shutdown() {
if let Err(error) = tokio::signal::ctrl_c().await {
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
std::future::pending::<()>().await;
}
}
use std::error::Error;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;

use clap::{Parser, Subcommand, ValueEnum};
use syncode_control::action_delivery::router as action_delivery_router;
use syncode_control::action_oci::PinnedOciResolver;
use syncode_control::action_repository::ForgeActionRepository;
use syncode_control::action_store::FileActionStore;
use syncode_control::actions::ActionResolver;
use syncode_control::actions_read::ActionsRead;
use syncode_control::actions_read_identity::IdentityActionsAuthorization;
use syncode_control::admin::{Admin, router as admin_router};
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
use syncode_control::checks::Checks;
use syncode_control::maintenance;
use syncode_control::projection;
use syncode_control::repository::{RepositoryContents, RepositorySecrets};
use syncode_control::repository_grpc::NativeRepositoryContents;
use syncode_control::repository_sources::RepositorySources;
use syncode_control::secrets::RuntimeSecrets;
use syncode_control::token::EnrolmentScope;
use syncode_control::webhook::{Intake, router};
use syncode_control_node::{
ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer,
GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient,
};
use syncode_control_nodes::{Nodes, Scope};
use syncode_control_runs::{
AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy,
};
use syncode_control_store::Postgres;
use tokio::net::TcpListener;
use tonic::transport::Server;
use url::Url;

#[derive(Debug, Parser)]
#[command(name = "syncode-control", version, about)]
struct Arguments {
#[command(subcommand)]
command: Option<Command>,

/// Where nodes open their session.
#[arg(long, default_value = "127.0.0.1:8090")]
listen: SocketAddr,

/// Where the native repository event feed delivers events.
#[arg(long, default_value = "127.0.0.1:8091")]
listen_events: SocketAddr,

/// The repository service this control plane reads workflows from.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_URL")]
repository_source: Url,

/// A token allowed to read repository contents.
#[arg(long, env = "SYNCODE_REPOSITORY_SOURCE_TOKEN", hide_env_values = true)]
repository_source_token: String,

/// Where native repositories are read over gRPC.
#[arg(long, env = "SYNCODE_REPOSITORY_GRPC_URL")]
repository_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_GRPC_URL")]
identity_grpc: String,

#[arg(long, env = "SYNCODE_IDENTITY_SHARED_SECRET", hide_env_values = true)]
identity_shared_secret: String,

#[arg(long, env = "SYNCODE_ACTION_MIRROR_URL")]
action_mirror_url: Url,

#[arg(
long,
env = "SYNCODE_ACTION_ALLOWLIST",
value_delimiter = ',',
num_args = 1..
)]
action_allowlist: Vec<Url>,

#[arg(
long,
env = "SYNCODE_ACTION_OCI_REGISTRIES",
value_delimiter = ',',
num_args = 1..
)]
action_oci_registries: Vec<String>,

#[arg(long, env = "SYNCODE_ACTION_STORE")]
action_store: PathBuf,

#[arg(long, env = "SYNCODE_ACTION_ARTIFACT_PUBLIC_URL")]
action_artifact_public_url: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_URL")]
secret_source: Url,

#[arg(long, env = "SYNCODE_SECRET_SOURCE_TOKEN", hide_env_values = true)]
secret_source_token: String,

#[arg(long, env = "SYNCODE_CAPABILITY_SIGNING_KEY", hide_env_values = true)]
capability_signing_key: String,

#[arg(long, env = "SYNCODE_ARTIFACT_SIGNING_KEY", hide_env_values = true)]
artifact_signing_key: String,

#[arg(long, env = "SYNCODE_PROJECTION_URL")]
projection_url: Url,

#[arg(long, env = "SYNCODE_PROJECTION_TOKEN", hide_env_values = true)]
projection_token: String,

#[arg(long, env = "SYNCODE_PROJECTION_MIN_RUN_NUMBER")]
projection_min_run_number: u64,

/// The secret used to sign the native event feed.
#[arg(long, env = "SYNCODE_EVENT_FEED_SECRET", hide_env_values = true)]
event_feed_secret: String,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_URL")]
collaboration_event_url: Url,

#[arg(long, env = "SYNCODE_COLLAB_EVENT_SECRET", hide_env_values = true)]
collaboration_event_secret: String,

#[arg(long, env = "SYNCODE_CONTROL_SOURCE_NODE_ID")]
source_node_id: uuid::Uuid,

#[arg(
long,
env = "SYNCODE_CHECK_EVENT_INTERVAL_SECONDS",
default_value_t = 1
)]
check_event_interval_seconds: u64,

#[arg(long, env = "SYNCODE_CHECK_EVENT_BATCH", default_value_t = 100)]
check_event_batch: i64,

/// Bearer token protecting the operational API.
#[arg(long, env = "SYNCODE_CONTROL_ADMIN_TOKEN", hide_env_values = true)]
admin_token: Option<String>,

/// Where the run log is kept.
#[arg(long, env = "SYNCODE_DATABASE_URL", hide_env_values = true)]
database: String,

#[arg(long, default_value_t = 8)]
database_connections: u32,

#[arg(long, env = "SYNCODE_ORGANIZATION_CONCURRENCY", default_value_t = 100)]
organization_concurrency: u32,

#[arg(long, env = "SYNCODE_REPOSITORY_CONCURRENCY", default_value_t = 20)]
repository_concurrency: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_CONCURRENCY", default_value_t = 20)]
principal_concurrency: u32,

#[arg(
long,
env = "SYNCODE_ORGANIZATION_QUEUE_QUOTA",
default_value_t = 10_000
)]
organization_queue_quota: u32,

#[arg(long, env = "SYNCODE_PRINCIPAL_QUEUE_QUOTA", default_value_t = 1_000)]
principal_queue_quota: u32,

#[arg(long, env = "SYNCODE_AUDIT_RETENTION_DAYS", default_value_t = 90)]
audit_retention_days: u32,

/// Whether compiled shadow runs may be assigned to nodes.
#[arg(long, value_enum, default_value_t = Mode::Shadow)]
mode: Mode,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)]
enum Mode {
Shadow,
Active,
}

#[derive(Debug, Subcommand)]
enum Command {
/// Issue an enrolment token a node can spend for an identity.
///
/// The token is stored before it is printed, so one that reaches an
/// operator is one the control plane will honour.
IssueToken {
/// How far the token reaches: `instance`, `organisation:<name>` or
/// `repository:<owner>/<name>`.
#[arg(long, default_value = "instance")]
scope: String,
},
/// Revoke a node identity immediately.
RevokeNode {
/// The node UUID printed at enrolment.
node: String,
},
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let arguments = Arguments::parse();
// The log is read before anything is served: a node that reconnects has to
// meet the run it was holding, not a control plane that has forgotten it.
let store = Postgres::connect(&arguments.database, arguments.database_connections).await?;
let policy = SchedulerPolicy::new(
arguments.organization_concurrency,
arguments.repository_concurrency,
arguments.principal_concurrency,
arguments.organization_queue_quota,
arguments.principal_queue_quota,
);
let runs = Runs::restored_with_policy(store.clone(), policy).await?;
let nodes = Nodes::restored(store.clone()).await?;

if let Some(command) = arguments.command {
match command {
Command::IssueToken { scope } => {
let scope: Scope = scope.parse::<EnrolmentScope>()?.into();
let token = nodes.issue_token(scope).await?;
println!("{}", token.secret().expose());
}
Command::RevokeNode { node } => {
nodes.revoke(node.parse::<NodeId>()?).await?;
}
}
return Ok(());
}

store
.record_audit(AuditEvent::new("control.configuration", "applied").after(
AuditPayload::Configuration(AuditConfiguration {
mode: match arguments.mode {
Mode::Shadow => AuditControlMode::Shadow,
Mode::Active => AuditControlMode::Active,
},
organization_concurrency: arguments.organization_concurrency,
repository_concurrency: arguments.repository_concurrency,
principal_concurrency: arguments.principal_concurrency,
organization_queue_quota: arguments.organization_queue_quota,
principal_queue_quota: arguments.principal_queue_quota,
audit_retention_days: arguments.audit_retention_days,
projection_min_run_number: arguments.projection_min_run_number,
}),
))
.await?;
tokio::spawn(maintenance::sweep(runs.clone(), nodes.clone()));
tokio::spawn(maintenance::retain_audit(
store.clone(),
arguments.audit_retention_days,
));

let action_mirror = arguments
.action_mirror_url
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()?;
let action_allowlist = arguments
.action_allowlist
.into_iter()
.map(|repository| {
repository
.to_string()
.parse::<syncode_workflow::RepositoryUrl>()
})
.collect::<Result<Vec<_>, _>>()?;
let action_store = FileActionStore::new(arguments.action_store);
let action_resolver = ActionResolver::new(
ForgeActionRepository::new(arguments.repository_source_token.clone()),
action_store.clone(),
PinnedOciResolver::new(arguments.action_oci_registries)?,
action_mirror,
action_allowlist,
);
let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).await?;
let repository_sources = RepositorySources::new(
native_repository,
RepositoryContents::new(
arguments.repository_source,
arguments.repository_source_token,
),
);
let intake = Arc::new(Intake::new(
repository_sources,
runs.clone(),
action_resolver,
arguments.event_feed_secret,
));
let admin_token = arguments
.admin_token
.filter(|token| !token.is_empty())
.ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?;
let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?;
let http = router(intake)
.merge(admin_router(Arc::new(
Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()),
)))
.merge(action_delivery_router(
action_store,
artifact_authority.clone(),
runs.clone(),
));
let events = TcpListener::bind(arguments.listen_events).await?;
let check_events = CheckEventPublisher::new(
store.clone(),
CheckEventPublisherConfig {
endpoint: arguments.collaboration_event_url.as_str(),
secret: arguments.collaboration_event_secret,
source_node_id: arguments.source_node_id,
interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds),
batch: arguments.check_event_batch,
},
)?;

let authority = CapabilityAuthority::new(arguments.capability_signing_key)?;
let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?;
let secret_service = RuntimeSecrets::new(
runs.clone(),
nodes.clone(),
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
authority.clone(),
);
let actions_read = ActionsRead::new(
runs.clone(),
IdentityActionsAuthorization::connect(
arguments.identity_grpc,
arguments.identity_shared_secret,
)
.await?,
);

// Both ends of the control plane run for as long as the other does: without
// events there is nothing to assign, and without sessions there is nobody to
// assign it to. Whichever stops first takes the process down with it.
let node_service = match arguments.mode {
Mode::Shadow => NodeSessionServer::shadow(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
projection.clone(),
),
Mode::Active => NodeSessionServer::new(
runs.clone(),
nodes,
authority,
artifact_authority,
arguments.action_artifact_public_url,
projection.clone(),
),
};
tokio::select! {
served = axum::serve(events, http).into_future() => served?,
served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?,
published = check_events.run() => published?,
served = Server::builder()
.add_service(GeneratedServer::new(node_service))
.add_service(GeneratedSecretsServer::new(secret_service))
.add_service(GeneratedChecksServer::new(Checks::new(runs.clone())))
.add_service(GeneratedActionsReadServer::new(actions_read))
.serve_with_shutdown(arguments.listen, shutdown()) => served?,
}

Ok(())
}

async fn shutdown() {
if let Err(error) = tokio::signal::ctrl_c().await {
eprintln!("cannot listen for shutdown, keeping the service running: {error}");
std::future::pending::<()>().await;
}
}
crates/control-node/build.rs
+5 -1
View File
@@ -1,15 +1,19 @@
use std::error::Error;

fn main() -> Result<(), Box<dyn Error>> {
let mut config = prost_build::Config::new();
config.protoc_executable(protoc_bin_vendored::protoc_bin_path()?);

tonic_prost_build::configure().compile_with_config(
config,
&["proto/node.proto"],
&["proto"],
)?;

println!("cargo:rerun-if-changed=proto");
Ok(())
}
use std::error::Error;

fn main() -> Result<(), Box<dyn Error>> {
let mut config = prost_build::Config::new();
config.protoc_executable(protoc_bin_vendored::protoc_bin_path()?);

tonic_prost_build::configure().compile_with_config(
config,
&[
"proto/node.proto",
"proto/actions.proto",
"proto/identity.proto",
],
&["proto"],
)?;

println!("cargo:rerun-if-changed=proto");
Ok(())
}
crates/control-node/src/lib.rs
+14
View File
@@ -1,27 +1,41 @@
mod artifact;
mod capability;
mod declaration;
mod error;
mod identity;
mod outbound;
mod projection;
mod reports;
mod session;

pub mod wire {
//! Generated from `proto/node.proto`.
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]

include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs"));
}

pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError};
pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError};
pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError};
pub use session::NodeSessionServer;
pub use wire::checks_server::Checks as ChecksService;
pub use wire::checks_server::ChecksServer as GeneratedChecksServer;
pub use wire::node_session_client::NodeSessionClient;
pub use wire::node_session_server::NodeSessionServer as GeneratedServer;
pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService;
pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer;
mod artifact;
mod capability;
mod declaration;
mod error;
mod identity;
mod outbound;
mod projection;
mod reports;
mod session;

pub mod wire {
//! Generated from `proto/node.proto`.
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]

include!(concat!(env!("OUT_DIR"), "/syncode.node.v1.rs"));
}

pub mod identity_wire {
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]

include!(concat!(env!("OUT_DIR"), "/syncode.identity.v1.rs"));
}

pub mod actions_wire {
#![allow(clippy::doc_markdown, clippy::large_enum_variant, clippy::use_self)]

include!(concat!(env!("OUT_DIR"), "/syncode.control.v1.rs"));
}

pub use actions_wire::actions_read_server::ActionsRead as ActionsReadService;
pub use actions_wire::actions_read_server::ActionsReadServer as GeneratedActionsReadServer;
pub use artifact::{ArtifactCapability, ArtifactTokenAuthority, ArtifactTokenError};
pub use capability::{CapabilityAuthority, CapabilityClaims, CapabilityError};
pub use projection::{ProjectionClient, ProjectionClientError, ProjectionRequestError};
pub use session::NodeSessionServer;
pub use wire::checks_server::Checks as ChecksService;
pub use wire::checks_server::ChecksServer as GeneratedChecksServer;
pub use wire::node_session_client::NodeSessionClient;
pub use wire::node_session_server::NodeSessionServer as GeneratedServer;
pub use wire::runtime_secrets_server::RuntimeSecrets as RuntimeSecretsService;
pub use wire::runtime_secrets_server::RuntimeSecretsServer as GeneratedSecretsServer;
crates/control-runs/src/lib.rs
+2
View File
@@ -1,35 +1,37 @@
mod assignment;
mod audit;
mod diagnosis;
mod identity;
mod job;
mod lease;
mod log;
mod origin;
mod output;
mod projection;
mod record;
mod registry;
mod run;
mod scheduler;

pub use assignment::{Assignment, Dependency};
pub use audit::{
AuditConfiguration, AuditContainerRuntime, AuditControlMode, AuditEnrolment, AuditEvent,
AuditJob, AuditJobs, AuditNodeCapabilities, AuditNodeCapacity, AuditNodeLifecycle,
AuditNodeScope, AuditNodeSnapshot, AuditPayload, AuditRunAdmission, AuditVolume,
};
pub use diagnosis::QueueDiagnosis;
pub use identity::{IdentityError, JobId, NodeId, RunId, Sequence};
pub use job::{JobState, MatrixPolicy, QueuedJob};
pub use lease::{Fence, Lease};
pub use log::{Forgotten, ForgottenError, OpenedRun, RunLog, StoredRun};
pub use origin::{Origin, RunNumber};
pub use projection::{ProjectedJob, ProjectedJobState, ProjectedRun};
pub use record::RunRecord;
pub use registry::{Cancellation, CancellationDispatch, Dispatch, LEASE_TERM, Runs, RunsError};
pub use run::{Conclusion, Run, RunCommand, RunError, RunEvent, RunState, SecretReadOutcome};
pub use scheduler::{
Architecture, Candidate, OperatingSystem, Priority, Refusal as SchedulingRefusal, Requirements,
SchedulerPolicy, Scope as SchedulingScope,
};
mod assignment;
mod audit;
mod diagnosis;
mod identity;
mod job;
mod lease;
mod log;
mod origin;
mod output;
mod projection;
mod read;
mod record;
mod registry;
mod run;
mod scheduler;

pub use assignment::{Assignment, Dependency};
pub use audit::{
AuditConfiguration, AuditContainerRuntime, AuditControlMode, AuditEnrolment, AuditEvent,
AuditJob, AuditJobs, AuditNodeCapabilities, AuditNodeCapacity, AuditNodeLifecycle,
AuditNodeScope, AuditNodeSnapshot, AuditPayload, AuditRunAdmission, AuditVolume,
};
pub use diagnosis::QueueDiagnosis;
pub use identity::{IdentityError, JobId, NodeId, RunId, Sequence};
pub use job::{JobState, MatrixPolicy, QueuedJob};
pub use lease::{Fence, Lease};
pub use log::{Forgotten, ForgottenError, OpenedRun, RunLog, StoredRun};
pub use origin::{Origin, RunNumber};
pub use projection::{ProjectedJob, ProjectedJobState, ProjectedRun};
pub use read::{JobView, RunView};
pub use record::RunRecord;
pub use registry::{Cancellation, CancellationDispatch, Dispatch, LEASE_TERM, Runs, RunsError};
pub use run::{Conclusion, Run, RunCommand, RunError, RunEvent, RunState, SecretReadOutcome};
pub use scheduler::{
Architecture, Candidate, OperatingSystem, Priority, Refusal as SchedulingRefusal, Requirements,
SchedulerPolicy, Scope as SchedulingScope,
};
crates/control-store/src/lib.rs
+17
View File
@@ -1,197 +1,214 @@
//! What the control plane writes down, kept in Postgres.
//!
//! Nothing here decides anything. The aggregate decides, this writes the
//! decision down and reads it back, which is why a run has no stored state
//! column: its state is what replaying its events produces. A node is the other
//! shape — a current fact rather than a history — so its state is stored and
//! overwritten.

mod audit;
mod nodes;
mod runs;

pub use audit::RecordedAuditEvent;

use sqlx::Executor;
use sqlx::postgres::{PgPool, PgPoolOptions};
use thiserror::Error;

#[derive(Debug, Error)]
pub enum StoreError {
#[error("the store cannot be reached: {0}")]
Unreachable(#[from] sqlx::Error),

#[error("something stored cannot be read back: {0}")]
Unreadable(#[from] serde_json::Error),

#[error("sequence {0} in the log does not fit the range a run can hold")]
Sequence(i64),

#[error("a stored run contains no jobs")]
EmptyRun,

#[error("audit retention of {0} days is outside the database range")]
Retention(u32),
}

#[derive(Clone)]
pub struct Postgres {
pool: PgPool,
}

/// The tables the log needs, created if this is the first time it runs.
///
/// A run's events are keyed by the run and its sequence, so the same event
/// cannot be written twice and a gap cannot be filled in later by accident.
///
/// The columns a table created by an earlier version does not have are added
/// nullable first, because there is no honest value to give a run that was
/// opened before the control plane wrote down where a run came from.
///
/// Such a run is then dropped rather than kept, and the columns are made
/// mandatory so no later one can look like it. Inventing an origin is still off
/// the table; what changed is the cost of refusing. A run with no origin can
/// never be assigned — a node cannot be told what it is building — so keeping it
/// bought nothing, while reading it back took the whole control plane down on
/// every start, which is how one unusable row became an outage.
///
/// An origin that cannot name its workflow file goes the same way, and for the
/// same reason: it is a shape this control plane no longer writes and can no
/// longer read, so leaving it in the table is leaving a start to fail on.
const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS runs (
id UUID PRIMARY KEY,
job UUID NOT NULL,
number BIGINT NOT NULL,
origin JSONB NOT NULL,
plan BYTEA NOT NULL,
opened_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE runs ADD COLUMN IF NOT EXISTS number BIGINT;
ALTER TABLE runs ADD COLUMN IF NOT EXISTS origin JSONB;
DELETE FROM runs WHERE number IS NULL OR origin IS NULL OR origin->>'workflow' IS NULL;
ALTER TABLE runs ALTER COLUMN number SET NOT NULL;
ALTER TABLE runs ALTER COLUMN origin SET NOT NULL;
CREATE TABLE IF NOT EXISTS run_jobs (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
id UUID NOT NULL,
position BIGINT NOT NULL,
key TEXT NOT NULL,
needs JSONB NOT NULL,
matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb,
priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb,
requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb,
secrets JSONB NOT NULL DEFAULT '[]'::jsonb,
plan BYTEA NOT NULL,
PRIMARY KEY (run, id),
UNIQUE (run, position)
);
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS secrets JSONB NOT NULL DEFAULT '[]'::jsonb;
INSERT INTO run_jobs (run, id, position, key, needs, plan)
SELECT id, job, 0, job::text, '[]'::jsonb, plan FROM runs
ON CONFLICT (run, id) DO NOTHING;
CREATE TABLE IF NOT EXISTS run_events (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
sequence BIGINT NOT NULL,
event JSONB NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (run, sequence)
);
UPDATE run_events e SET event = to_jsonb('Opened'::text)
FROM runs r WHERE e.run = r.id AND e.event = to_jsonb('Queued'::text);
UPDATE run_events e SET event = jsonb_build_object(
'Assigned', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Assigned'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Assigned'
AND NOT ((e.event->'Assigned') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Renewed', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Renewed'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Renewed'
AND NOT ((e.event->'Renewed') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Started', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Started'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Started'
AND NOT ((e.event->'Started') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Finished', jsonb_build_object('job', to_jsonb(r.job), 'conclusion', e.event->'Finished'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Finished'
AND NOT ((e.event->'Finished') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'LeaseExpired', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'LeaseExpired'))
FROM runs r WHERE e.run = r.id AND e.event ? 'LeaseExpired'
AND NOT ((e.event->'LeaseExpired') ? 'job');
CREATE TABLE IF NOT EXISTS run_logs (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
job UUID NOT NULL,
at_offset BIGINT NOT NULL,
lines JSONB NOT NULL,
received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (run, job, at_offset)
);
CREATE TABLE IF NOT EXISTS nodes (
id UUID PRIMARY KEY,
state JSONB NOT NULL,
saved_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS enrolment_tokens (
secret TEXT PRIMARY KEY,
token JSONB NOT NULL,
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS node_message_inbox (
node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE,
message_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
payload BYTEA NOT NULL,
applied BOOLEAN NOT NULL DEFAULT false,
received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (node, message_id),
UNIQUE (node, idempotency_key)
);
CREATE TABLE IF NOT EXISTS node_control_outbox (
node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE,
message_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
payload BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
acknowledged_at TIMESTAMPTZ,
PRIMARY KEY (node, message_id),
UNIQUE (node, idempotency_key)
);
CREATE TABLE IF NOT EXISTS audit_entries (
id BIGSERIAL PRIMARY KEY,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
kind TEXT NOT NULL,
outcome TEXT NOT NULL,
principal TEXT,
node UUID,
run UUID,
job UUID,
correlation TEXT,
before_state JSONB,
after_state JSONB,
reason TEXT
);
CREATE INDEX IF NOT EXISTS audit_entries_occurred_at ON audit_entries (occurred_at);
CREATE INDEX IF NOT EXISTS audit_entries_run_job ON audit_entries (run, job, id);
CREATE INDEX IF NOT EXISTS audit_entries_node ON audit_entries (node, id);
"#;

impl Postgres {
/// Connect and make sure the schema is there.
pub async fn connect(url: &str, connections: u32) -> Result<Self, StoreError> {
let pool = PgPoolOptions::new()
.max_connections(connections)
.connect(url)
.await?;
pool.execute(SCHEMA).await?;
Ok(Self { pool })
}

#[must_use]
pub const fn pool(&self) -> &PgPool {
&self.pool
}
}
//! What the control plane writes down, kept in Postgres.
//!
//! Nothing here decides anything. The aggregate decides, this writes the
//! decision down and reads it back, which is why a run has no stored state
//! column: its state is what replaying its events produces. A node is the other
//! shape — a current fact rather than a history — so its state is stored and
//! overwritten.

mod audit;
mod check_events;
mod nodes;
mod runs;

pub use audit::RecordedAuditEvent;
pub use check_events::PendingCheckEvent;

use sqlx::Executor;
use sqlx::postgres::{PgPool, PgPoolOptions};
use thiserror::Error;

#[derive(Debug, Error)]
pub enum StoreError {
#[error("the store cannot be reached: {0}")]
Unreachable(#[from] sqlx::Error),

#[error("something stored cannot be read back: {0}")]
Unreadable(#[from] serde_json::Error),

#[error("sequence {0} in the log does not fit the range a run can hold")]
Sequence(i64),

#[error("a stored run contains no jobs")]
EmptyRun,

#[error("audit retention of {0} days is outside the database range")]
Retention(u32),

#[error("a stored run is invalid: {0}")]
InvalidRun(String),
}

#[derive(Clone)]
pub struct Postgres {
pool: PgPool,
}

/// The tables the log needs, created if this is the first time it runs.
///
/// A run's events are keyed by the run and its sequence, so the same event
/// cannot be written twice and a gap cannot be filled in later by accident.
///
/// The columns a table created by an earlier version does not have are added
/// nullable first, because there is no honest value to give a run that was
/// opened before the control plane wrote down where a run came from.
///
/// Such a run is then dropped rather than kept, and the columns are made
/// mandatory so no later one can look like it. Inventing an origin is still off
/// the table; what changed is the cost of refusing. A run with no origin can
/// never be assigned — a node cannot be told what it is building — so keeping it
/// bought nothing, while reading it back took the whole control plane down on
/// every start, which is how one unusable row became an outage.
///
/// An origin that cannot name its workflow file goes the same way, and for the
/// same reason: it is a shape this control plane no longer writes and can no
/// longer read, so leaving it in the table is leaving a start to fail on.
const SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS runs (
id UUID PRIMARY KEY,
job UUID NOT NULL,
number BIGINT NOT NULL,
origin JSONB NOT NULL,
plan BYTEA NOT NULL,
opened_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE runs ADD COLUMN IF NOT EXISTS number BIGINT;
ALTER TABLE runs ADD COLUMN IF NOT EXISTS origin JSONB;
DELETE FROM runs WHERE number IS NULL OR origin IS NULL OR origin->>'workflow' IS NULL;
ALTER TABLE runs ALTER COLUMN number SET NOT NULL;
ALTER TABLE runs ALTER COLUMN origin SET NOT NULL;
CREATE TABLE IF NOT EXISTS run_jobs (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
id UUID NOT NULL,
position BIGINT NOT NULL,
key TEXT NOT NULL,
needs JSONB NOT NULL,
matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb,
priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb,
requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb,
secrets JSONB NOT NULL DEFAULT '[]'::jsonb,
plan BYTEA NOT NULL,
PRIMARY KEY (run, id),
UNIQUE (run, position)
);
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS matrix JSONB NOT NULL DEFAULT '{"fail_fast":true,"max_parallel":null}'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS priority JSONB NOT NULL DEFAULT '"Normal"'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS requirements JSONB NOT NULL DEFAULT '{"architecture":null,"operating_system":null,"container_runtime":null,"labels":[],"minimum_cores":0,"minimum_memory_bytes":0,"build_volume_bytes":0,"preferred_images":[],"preferred_actions":[]}'::jsonb;
ALTER TABLE run_jobs ADD COLUMN IF NOT EXISTS secrets JSONB NOT NULL DEFAULT '[]'::jsonb;
INSERT INTO run_jobs (run, id, position, key, needs, plan)
SELECT id, job, 0, job::text, '[]'::jsonb, plan FROM runs
ON CONFLICT (run, id) DO NOTHING;
CREATE TABLE IF NOT EXISTS run_events (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
sequence BIGINT NOT NULL,
event JSONB NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (run, sequence)
);
UPDATE run_events e SET event = to_jsonb('Opened'::text)
FROM runs r WHERE e.run = r.id AND e.event = to_jsonb('Queued'::text);
UPDATE run_events e SET event = jsonb_build_object(
'Assigned', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Assigned'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Assigned'
AND NOT ((e.event->'Assigned') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Renewed', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Renewed'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Renewed'
AND NOT ((e.event->'Renewed') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Started', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'Started'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Started'
AND NOT ((e.event->'Started') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'Finished', jsonb_build_object('job', to_jsonb(r.job), 'conclusion', e.event->'Finished'))
FROM runs r WHERE e.run = r.id AND e.event ? 'Finished'
AND NOT ((e.event->'Finished') ? 'job');
UPDATE run_events e SET event = jsonb_build_object(
'LeaseExpired', jsonb_build_object('job', to_jsonb(r.job), 'lease', e.event->'LeaseExpired'))
FROM runs r WHERE e.run = r.id AND e.event ? 'LeaseExpired'
AND NOT ((e.event->'LeaseExpired') ? 'job');
CREATE TABLE IF NOT EXISTS run_logs (
run UUID NOT NULL REFERENCES runs (id) ON DELETE CASCADE,
job UUID NOT NULL,
at_offset BIGINT NOT NULL,
lines JSONB NOT NULL,
received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (run, job, at_offset)
);
CREATE TABLE IF NOT EXISTS check_event_outbox (
id BIGSERIAL PRIMARY KEY,
message_id UUID NOT NULL UNIQUE,
repository_id UUID NOT NULL,
payload JSONB NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
available_at TIMESTAMPTZ NOT NULL DEFAULT now(),
delivered_at TIMESTAMPTZ,
last_error TEXT
);
CREATE INDEX IF NOT EXISTS check_event_outbox_pending
ON check_event_outbox (available_at, id) WHERE delivered_at IS NULL;
CREATE TABLE IF NOT EXISTS nodes (
id UUID PRIMARY KEY,
state JSONB NOT NULL,
saved_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS enrolment_tokens (
secret TEXT PRIMARY KEY,
token JSONB NOT NULL,
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS node_message_inbox (
node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE,
message_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
payload BYTEA NOT NULL,
applied BOOLEAN NOT NULL DEFAULT false,
received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (node, message_id),
UNIQUE (node, idempotency_key)
);
CREATE TABLE IF NOT EXISTS node_control_outbox (
node UUID NOT NULL REFERENCES nodes (id) ON DELETE CASCADE,
message_id TEXT NOT NULL,
idempotency_key TEXT NOT NULL,
payload BYTEA NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
acknowledged_at TIMESTAMPTZ,
PRIMARY KEY (node, message_id),
UNIQUE (node, idempotency_key)
);
CREATE TABLE IF NOT EXISTS audit_entries (
id BIGSERIAL PRIMARY KEY,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
kind TEXT NOT NULL,
outcome TEXT NOT NULL,
principal TEXT,
node UUID,
run UUID,
job UUID,
correlation TEXT,
before_state JSONB,
after_state JSONB,
reason TEXT
);
CREATE INDEX IF NOT EXISTS audit_entries_occurred_at ON audit_entries (occurred_at);
CREATE INDEX IF NOT EXISTS audit_entries_run_job ON audit_entries (run, job, id);
CREATE INDEX IF NOT EXISTS audit_entries_node ON audit_entries (node, id);
"#;

impl Postgres {
/// Connect and make sure the schema is there.
pub async fn connect(url: &str, connections: u32) -> Result<Self, StoreError> {
let pool = PgPoolOptions::new()
.max_connections(connections)
.connect(url)
.await?;
pool.execute(SCHEMA).await?;
Ok(Self { pool })
}

#[must_use]
pub const fn pool(&self) -> &PgPool {
&self.pool
}
}
crates/control-store/src/runs.rs
+16 -1
View File
@@ -1,234 +1,249 @@
//! How a run's history is written down and read back.

mod audit;
mod restore;

use std::collections::BTreeMap;

use sqlx::types::Json;
use syncode_control_runs::{
AuditEvent, AuditJob, AuditJobs, AuditPayload, JobId, OpenedRun, Origin, Priority, QueuedJob,
Requirements, RunId, RunLog, RunRecord, StoredRun,
};
use uuid::Uuid;

use self::audit::run_audit;
use self::restore::Opened;
use crate::audit::write_audit;
use crate::{Postgres, StoreError};

impl Postgres {
async fn write(&self, run: RunId, record: RunRecord) -> Result<(), StoreError> {
let sequence =
i64::try_from(record.sequence().get()).map_err(|_| StoreError::Sequence(i64::MAX))?;
let mut transaction = self.pool.begin().await?;
sqlx::query!(
"INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)",
Uuid::from(run),
sequence,
serde_json::to_value(record.event())?,
)
.execute(&mut *transaction)
.await?;
let origin = sqlx::query_scalar!(
r#"SELECT origin AS "origin!: Json<Origin>" FROM runs WHERE id = $1"#,
Uuid::from(run),
)
.fetch_one(&mut *transaction)
.await?;
write_audit(&mut *transaction, &run_audit(run, &record, &origin.0)).await?;
transaction.commit().await?;
Ok(())
}
}

impl RunLog for Postgres {
type Error = StoreError;

async fn opened(&self, run: OpenedRun<'_>) -> Result<(), Self::Error> {
let number = i64::try_from(run.number.get()).map_err(|_| StoreError::Sequence(i64::MAX))?;
let first = run.jobs.first().ok_or(StoreError::EmptyRun)?;
// The run and the event that opened it are one fact. Writing them apart
// would allow a run with no history, which nothing could rebuild.
let mut transaction = self.pool.begin().await?;
sqlx::query!(
"INSERT INTO runs (id, job, number, origin, plan) VALUES ($1, $2, $3, $4, $5)",
Uuid::from(run.run),
Uuid::from(first.id()),
number,
serde_json::to_value(run.origin)?,
first.plan(),
)
.execute(&mut *transaction)
.await?;
for (position, job) in run.jobs.iter().enumerate() {
let position = i64::try_from(position).map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_jobs (
run, id, position, key, needs, matrix, priority, requirements, secrets, plan
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
Uuid::from(run.run),
Uuid::from(job.id()),
position,
job.key(),
serde_json::to_value(job.needs())?,
serde_json::to_value(job.matrix())?,
serde_json::to_value(job.priority())?,
serde_json::to_value(job.requirements())?,
serde_json::to_value(job.secrets())?,
job.plan(),
)
.execute(&mut *transaction)
.await?;
}
let sequence = i64::try_from(run.record.sequence().get())
.map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)",
Uuid::from(run.run),
sequence,
serde_json::to_value(run.record.event())?,
)
.execute(&mut *transaction)
.await?;
write_audit(
&mut *transaction,
&AuditEvent::new("run.triggered", "accepted")
.run(run.run)
.principal(run.origin.principal())
.correlation(run.origin.delivery())
.after(AuditPayload::Origin(run.origin.clone())),
)
.await?;
let jobs = run
.jobs
.iter()
.map(|job| AuditJob {
id: job.id(),
key: job.key().to_owned(),
needs: job.needs().to_vec(),
matrix: job.matrix(),
priority: job.priority(),
requirements: job.requirements().clone(),
secrets: job.secrets().to_vec(),
})
.collect::<Vec<_>>();
write_audit(
&mut *transaction,
&AuditEvent::new("workflow.compiled", "accepted")
.run(run.run)
.principal(run.origin.principal())
.correlation(run.origin.delivery())
.after(AuditPayload::Jobs(AuditJobs { jobs })),
)
.await?;
transaction.commit().await?;
Ok(())
}

async fn appended(&self, run: RunId, record: RunRecord) -> Result<(), Self::Error> {
self.write(run, record).await
}

/// The offset is the primary key, so a batch that arrives twice is written
/// once. A node retrying after a broken connection must not double the log.
async fn logged(
&self,
run: RunId,
job: JobId,
offset: u64,
lines: &[String],
) -> Result<(), Self::Error> {
let offset = i64::try_from(offset).map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_logs (run, job, at_offset, lines) VALUES ($1, $2, $3, $4)
ON CONFLICT (run, job, at_offset) DO NOTHING",
Uuid::from(run),
Uuid::from(job),
offset,
serde_json::to_value(lines)?,
)
.execute(&self.pool)
.await?;
Ok(())
}

async fn lines(&self, run: RunId, job: JobId) -> Result<Vec<String>, Self::Error> {
let rows = sqlx::query!(
"SELECT lines FROM run_logs WHERE run = $1 AND job = $2 ORDER BY at_offset",
Uuid::from(run),
Uuid::from(job),
)
.fetch_all(&self.pool)
.await?;

let mut printed = Vec::new();
for row in rows {
let batch: Vec<String> = serde_json::from_value(row.lines)?;
printed.extend(batch);
}
Ok(printed)
}

async fn audited(&self, event: AuditEvent) -> Result<(), Self::Error> {
self.record_audit(event).await
}

async fn restore(&self) -> Result<Vec<StoredRun>, Self::Error> {
let mut jobs: BTreeMap<RunId, Vec<QueuedJob>> = BTreeMap::new();
for row in sqlx::query!(
"SELECT run, id, key, needs, matrix, priority, requirements, secrets, plan
FROM run_jobs ORDER BY run, position",
)
.fetch_all(&self.pool)
.await?
{
let run = RunId::from(row.run);
let priority: Priority = serde_json::from_value(row.priority)?;
let requirements: Requirements = serde_json::from_value(row.requirements)?;
let job = QueuedJob::new(
JobId::from(row.id),
row.key,
serde_json::from_value(row.needs)?,
serde_json::from_value(row.matrix)?,
row.plan,
)
.scheduled(priority, requirements)
.referencing(serde_json::from_value(row.secrets)?);
jobs.entry(run).or_default().push(job);
}
let rows = sqlx::query!(
r#"SELECT r.id, r.number, r.origin AS "origin!: Json<Origin>", e.sequence, e.event
FROM runs r
JOIN run_events e ON e.run = r.id
ORDER BY r.opened_at, r.id, e.sequence"#,
)
.fetch_all(&self.pool)
.await?;

let mut restored: Vec<StoredRun> = Vec::new();
let mut log: Vec<RunRecord> = Vec::new();
let mut current: Option<Opened> = None;

for row in rows {
let opened = Opened::read(row.id, row.number, row.origin.0)?;
let sequence =
u64::try_from(row.sequence).map_err(|_| StoreError::Sequence(row.sequence))?;
let event = serde_json::from_value(row.event)?;

if current.as_ref().is_some_and(|open| open.id != opened.id)
&& let Some(open) = current.take()
{
let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?;
restored.push(open.stored(definitions, std::mem::take(&mut log)));
}
current.get_or_insert(opened);
log.push(RunRecord::new(sequence.into(), event));
}
if let Some(open) = current {
let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?;
restored.push(open.stored(definitions, log));
}
Ok(restored)
}
}
//! How a run's history is written down and read back.

mod audit;
mod restore;

use std::collections::BTreeMap;

use sqlx::types::Json;
use syncode_control_runs::{
AuditEvent, AuditJob, AuditJobs, AuditPayload, JobId, OpenedRun, Origin, Priority, QueuedJob,
Requirements, RunEvent, RunId, RunLog, RunRecord, StoredRun,
};
use uuid::Uuid;

use self::audit::run_audit;
use self::restore::Opened;
use crate::audit::write_audit;
use crate::check_events::enqueue_check_snapshot;
use crate::{Postgres, StoreError};

impl Postgres {
async fn write(&self, run: RunId, record: RunRecord) -> Result<(), StoreError> {
let sequence =
i64::try_from(record.sequence().get()).map_err(|_| StoreError::Sequence(i64::MAX))?;
let mut transaction = self.pool.begin().await?;
sqlx::query!(
"INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)",
Uuid::from(run),
sequence,
serde_json::to_value(record.event())?,
)
.execute(&mut *transaction)
.await?;
if affects_check_status(&record.event()) {
enqueue_check_snapshot(&mut transaction, run).await?;
}
let origin = sqlx::query_scalar!(
r#"SELECT origin AS "origin!: Json<Origin>" FROM runs WHERE id = $1"#,
Uuid::from(run),
)
.fetch_one(&mut *transaction)
.await?;
write_audit(&mut *transaction, &run_audit(run, &record, &origin.0)).await?;
transaction.commit().await?;
Ok(())
}
}

impl RunLog for Postgres {
type Error = StoreError;

async fn opened(&self, run: OpenedRun<'_>) -> Result<(), Self::Error> {
let number = i64::try_from(run.number.get()).map_err(|_| StoreError::Sequence(i64::MAX))?;
let first = run.jobs.first().ok_or(StoreError::EmptyRun)?;
// The run and the event that opened it are one fact. Writing them apart
// would allow a run with no history, which nothing could rebuild.
let mut transaction = self.pool.begin().await?;
sqlx::query!(
"INSERT INTO runs (id, job, number, origin, plan) VALUES ($1, $2, $3, $4, $5)",
Uuid::from(run.run),
Uuid::from(first.id()),
number,
serde_json::to_value(run.origin)?,
first.plan(),
)
.execute(&mut *transaction)
.await?;
for (position, job) in run.jobs.iter().enumerate() {
let position = i64::try_from(position).map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_jobs (
run, id, position, key, needs, matrix, priority, requirements, secrets, plan
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)",
Uuid::from(run.run),
Uuid::from(job.id()),
position,
job.key(),
serde_json::to_value(job.needs())?,
serde_json::to_value(job.matrix())?,
serde_json::to_value(job.priority())?,
serde_json::to_value(job.requirements())?,
serde_json::to_value(job.secrets())?,
job.plan(),
)
.execute(&mut *transaction)
.await?;
}
let sequence = i64::try_from(run.record.sequence().get())
.map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_events (run, sequence, event) VALUES ($1, $2, $3)",
Uuid::from(run.run),
sequence,
serde_json::to_value(run.record.event())?,
)
.execute(&mut *transaction)
.await?;
enqueue_check_snapshot(&mut transaction, run.run).await?;
write_audit(
&mut *transaction,
&AuditEvent::new("run.triggered", "accepted")
.run(run.run)
.principal(run.origin.principal())
.correlation(run.origin.delivery())
.after(AuditPayload::Origin(run.origin.clone())),
)
.await?;
let jobs = run
.jobs
.iter()
.map(|job| AuditJob {
id: job.id(),
key: job.key().to_owned(),
needs: job.needs().to_vec(),
matrix: job.matrix(),
priority: job.priority(),
requirements: job.requirements().clone(),
secrets: job.secrets().to_vec(),
})
.collect::<Vec<_>>();
write_audit(
&mut *transaction,
&AuditEvent::new("workflow.compiled", "accepted")
.run(run.run)
.principal(run.origin.principal())
.correlation(run.origin.delivery())
.after(AuditPayload::Jobs(AuditJobs { jobs })),
)
.await?;
transaction.commit().await?;
Ok(())
}

async fn appended(&self, run: RunId, record: RunRecord) -> Result<(), Self::Error> {
self.write(run, record).await
}

/// The offset is the primary key, so a batch that arrives twice is written
/// once. A node retrying after a broken connection must not double the log.
async fn logged(
&self,
run: RunId,
job: JobId,
offset: u64,
lines: &[String],
) -> Result<(), Self::Error> {
let offset = i64::try_from(offset).map_err(|_| StoreError::Sequence(i64::MAX))?;
sqlx::query!(
"INSERT INTO run_logs (run, job, at_offset, lines) VALUES ($1, $2, $3, $4)
ON CONFLICT (run, job, at_offset) DO NOTHING",
Uuid::from(run),
Uuid::from(job),
offset,
serde_json::to_value(lines)?,
)
.execute(&self.pool)
.await?;
Ok(())
}

async fn lines(&self, run: RunId, job: JobId) -> Result<Vec<String>, Self::Error> {
let rows = sqlx::query!(
"SELECT lines FROM run_logs WHERE run = $1 AND job = $2 ORDER BY at_offset",
Uuid::from(run),
Uuid::from(job),
)
.fetch_all(&self.pool)
.await?;

let mut printed = Vec::new();
for row in rows {
let batch: Vec<String> = serde_json::from_value(row.lines)?;
printed.extend(batch);
}
Ok(printed)
}

async fn audited(&self, event: AuditEvent) -> Result<(), Self::Error> {
self.record_audit(event).await
}

async fn restore(&self) -> Result<Vec<StoredRun>, Self::Error> {
let mut jobs: BTreeMap<RunId, Vec<QueuedJob>> = BTreeMap::new();
for row in sqlx::query!(
"SELECT run, id, key, needs, matrix, priority, requirements, secrets, plan
FROM run_jobs ORDER BY run, position",
)
.fetch_all(&self.pool)
.await?
{
let run = RunId::from(row.run);
let priority: Priority = serde_json::from_value(row.priority)?;
let requirements: Requirements = serde_json::from_value(row.requirements)?;
let job = QueuedJob::new(
JobId::from(row.id),
row.key,
serde_json::from_value(row.needs)?,
serde_json::from_value(row.matrix)?,
row.plan,
)
.scheduled(priority, requirements)
.referencing(serde_json::from_value(row.secrets)?);
jobs.entry(run).or_default().push(job);
}
let rows = sqlx::query!(
r#"SELECT r.id, r.number, r.origin AS "origin!: Json<Origin>", e.sequence, e.event
FROM runs r
JOIN run_events e ON e.run = r.id
ORDER BY r.opened_at, r.id, e.sequence"#,
)
.fetch_all(&self.pool)
.await?;

let mut restored: Vec<StoredRun> = Vec::new();
let mut log: Vec<RunRecord> = Vec::new();
let mut current: Option<Opened> = None;

for row in rows {
let opened = Opened::read(row.id, row.number, row.origin.0)?;
let sequence =
u64::try_from(row.sequence).map_err(|_| StoreError::Sequence(row.sequence))?;
let event = serde_json::from_value(row.event)?;

if current.as_ref().is_some_and(|open| open.id != opened.id)
&& let Some(open) = current.take()
{
let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?;
restored.push(open.stored(definitions, std::mem::take(&mut log)));
}
current.get_or_insert(opened);
log.push(RunRecord::new(sequence.into(), event));
}
if let Some(open) = current {
let definitions = jobs.remove(&open.id).ok_or(StoreError::EmptyRun)?;
restored.push(open.stored(definitions, log));
}
Ok(restored)
}
}

const fn affects_check_status(event: &RunEvent) -> bool {
matches!(
event,
RunEvent::Assigned { .. }
| RunEvent::Started { .. }
| RunEvent::Finished { .. }
| RunEvent::LeaseExpired { .. }
)
}
crates/control-runs/src/registry/mod.rs
+1
View File
@@ -1,49 +1,50 @@
mod cancellation;
mod diagnosis;
mod error;
mod projection;
mod reporting;
mod scheduling;
mod secrets;
mod state;
mod work;

use std::sync::{Arc, atomic::AtomicU64};
use std::time::Duration;

use async_lock::Mutex;
use event_listener::Event;

use self::state::State;
use crate::{Assignment, SchedulerPolicy, SchedulingRefusal};
pub use cancellation::{Cancellation, CancellationDispatch};
pub use error::RunsError;

pub const LEASE_TERM: Duration = Duration::from_secs(60);

pub struct Runs<L> {
pub(crate) store: Arc<L>,
pub(crate) state: Arc<Mutex<State>>,
work: Arc<Event>,
work_version: Arc<AtomicU64>,
policy: SchedulerPolicy,
}

#[derive(Clone, Debug)]
pub enum Dispatch {
Assigned(Box<Assignment>),
Empty,
Refused(Vec<SchedulingRefusal>),
}

impl<L> Clone for Runs<L> {
fn clone(&self) -> Self {
Self {
store: Arc::clone(&self.store),
state: Arc::clone(&self.state),
work: Arc::clone(&self.work),
work_version: Arc::clone(&self.work_version),
policy: self.policy,
}
}
}
mod cancellation;
mod diagnosis;
mod error;
mod projection;
mod read;
mod reporting;
mod scheduling;
mod secrets;
mod state;
mod work;

use std::sync::{Arc, atomic::AtomicU64};
use std::time::Duration;

use async_lock::Mutex;
use event_listener::Event;

use self::state::State;
use crate::{Assignment, SchedulerPolicy, SchedulingRefusal};
pub use cancellation::{Cancellation, CancellationDispatch};
pub use error::RunsError;

pub const LEASE_TERM: Duration = Duration::from_secs(60);

pub struct Runs<L> {
pub(crate) store: Arc<L>,
pub(crate) state: Arc<Mutex<State>>,
work: Arc<Event>,
work_version: Arc<AtomicU64>,
policy: SchedulerPolicy,
}

#[derive(Clone, Debug)]
pub enum Dispatch {
Assigned(Box<Assignment>),
Empty,
Refused(Vec<SchedulingRefusal>),
}

impl<L> Clone for Runs<L> {
fn clone(&self) -> Self {
Self {
store: Arc::clone(&self.store),
state: Arc::clone(&self.state),
work: Arc::clone(&self.work),
work_version: Arc::clone(&self.work_version),
policy: self.policy,
}
}
}
crates/control-runs/src/registry/state.rs
+4 -4
View File
@@ -1,250 +1,250 @@
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<Entry>,
organization_turns: BTreeMap<String, u64>,
next_turn: u64,
}

pub(super) struct ProjectionSeed {
pub run: RunId,
pub number: RunNumber,
pub origin: Origin,
pub sequence: u64,
pub jobs: Vec<ProjectionJobSeed>,
}

pub(super) struct ProjectionJobSeed {
pub job: JobId,
pub key: String,
pub needs: Vec<String>,
pub state: ProjectedJobState,
pub node: Option<NodeId>,
pub fence: Option<Fence>,
}

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<RunId> {
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<Assignment, RunsError> {
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<Item = &Run> {
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)
}

/// Every run of one commit, unfiltered by delivery.
pub(super) fn seeds_for_commit(&self, repository: &str, commit: &str) -> Vec<ProjectionSeed> {
self.entries
.iter()
.filter(|entry| {
entry.origin.repository() == repository && entry.origin.commit() == commit
})
.map(Self::projection_seed_of)
.collect()
}

pub(super) fn projection_seeds(&self) -> Vec<ProjectionSeed> {
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<ProjectionSeed> {
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 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 {
pub(super) run: Run,
number: RunNumber,
pub(super) origin: Origin,
}

pub struct State {
pub(super) entries: Vec<Entry>,
organization_turns: BTreeMap<String, u64>,
next_turn: u64,
}

pub(super) struct ProjectionSeed {
pub run: RunId,
pub number: RunNumber,
pub origin: Origin,
pub sequence: u64,
pub jobs: Vec<ProjectionJobSeed>,
}

pub(super) struct ProjectionJobSeed {
pub job: JobId,
pub key: String,
pub needs: Vec<String>,
pub state: ProjectedJobState,
pub node: Option<NodeId>,
pub fence: Option<Fence>,
}

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<RunId> {
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<Assignment, RunsError> {
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<Item = &Run> {
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)
}

/// Every run of one commit, unfiltered by delivery.
pub(super) fn seeds_for_commit(&self, repository: &str, commit: &str) -> Vec<ProjectionSeed> {
self.entries
.iter()
.filter(|entry| {
entry.origin.repository() == repository && entry.origin.commit() == commit
})
.map(Self::projection_seed_of)
.collect()
}

pub(super) fn projection_seeds(&self) -> Vec<ProjectionSeed> {
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<ProjectionSeed> {
self.entry(run)
.ok()
.filter(|entry| entry.origin.delivery().is_some())
.map(Self::projection_seed_of)
}

pub(super) 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))
}
}
crates/control-runs/src/run/access.rs
+1 -1
View File
@@ -1,212 +1,212 @@
use crate::job::Job;
use crate::{Dependency, Fence, JobId, JobState, Lease, QueuedJob, RunId, RunRecord, Sequence};
use std::collections::BTreeMap;

use super::{Conclusion, Run, RunError};

impl Run {
pub(super) fn advance(&mut self) {
loop {
let mut changes = Vec::new();
for job in self
.jobs
.iter()
.filter(|job| job.state() == JobState::Waiting)
{
let dependencies: Vec<&Job> = job
.needs()
.iter()
.flat_map(|key| {
self.jobs
.iter()
.filter(move |candidate| candidate.key() == key)
})
.collect();
if dependencies.iter().all(|dependency| {
matches!(
dependency.state(),
JobState::Finished(_) | JobState::Skipped
)
}) {
changes.push((job.id(), JobState::Queued));
}
}
if changes.is_empty() {
return;
}
for (job, state) in changes {
if let Ok(job) = self.job_mut(job) {
job.set_state(state);
}
}
}
}

pub(super) fn job(&self, id: JobId) -> Result<&Job, RunError> {
self.jobs
.iter()
.find(|job| job.id() == id)
.ok_or(RunError::UnknownJob {
run: self.id,
job: id,
})
}

pub(super) fn job_mut(&mut self, id: JobId) -> Result<&mut Job, RunError> {
self.jobs
.iter_mut()
.find(|job| job.id() == id)
.ok_or(RunError::UnknownJob {
run: self.id,
job: id,
})
}

pub(crate) fn queued_jobs(&self) -> impl Iterator<Item = &QueuedJob> {
self.jobs.iter().filter_map(|job| {
if job.state() != JobState::Queued {
return None;
}
let Some(limit) = job.matrix().max_parallel() else {
return Some(job.definition());
};
let active = self
.jobs
.iter()
.filter(|candidate| {
candidate.key() == job.key()
&& matches!(
candidate.state(),
JobState::Assigned(_) | JobState::Running(_)
)
})
.count() as u64;
(active < limit).then(|| job.definition())
})
}

pub(crate) fn active_jobs(&self) -> usize {
self.jobs
.iter()
.filter(|job| matches!(job.state(), JobState::Assigned(_) | JobState::Running(_)))
.count()
}

pub(crate) fn open_jobs(&self) -> usize {
self.jobs
.iter()
.filter(|job| {
matches!(
job.state(),
JobState::Waiting
| JobState::Queued
| JobState::Assigned(_)
| JobState::Running(_)
)
})
.count()
}

pub(crate) fn dependencies(&self, job: JobId) -> Result<Vec<Dependency>, RunError> {
let job = self.job(job)?;
job.needs()
.iter()
.map(|key| {
let states: Vec<JobState> = self
.jobs
.iter()
.filter(|candidate| candidate.key() == key)
.map(Job::state)
.collect();
let conclusion = if states
.iter()
.any(|state| matches!(state, JobState::Finished(Conclusion::Failure)))
{
Conclusion::Failure
} else if states
.iter()
.any(|state| matches!(state, JobState::Finished(Conclusion::Cancelled)))
{
Conclusion::Cancelled
} else if states.iter().any(|state| {
matches!(
state,
JobState::Finished(Conclusion::Skipped) | JobState::Skipped
)
}) {
Conclusion::Skipped
} else {
Conclusion::Success
};
let outputs = self
.jobs
.iter()
.filter(|candidate| candidate.key() == key)
.flat_map(|candidate| candidate.outputs().clone())
.collect();
Ok(Dependency::new(key.clone(), conclusion, outputs))
})
.collect()
}

pub(crate) fn jobs(&self) -> impl Iterator<Item = &QueuedJob> {
self.jobs.iter().map(Job::definition)
}

pub(crate) fn projected_jobs(&self) -> impl Iterator<Item = (&QueuedJob, JobState)> {
self.jobs.iter().map(|job| (job.definition(), job.state()))
}

pub(crate) fn job_definition(&self, job: JobId) -> Result<&QueuedJob, RunError> {
self.job(job).map(Job::definition)
}

pub fn job_state(&self, job: JobId) -> Result<JobState, RunError> {
self.job(job).map(Job::state)
}

pub(crate) fn repeats_finish(
&self,
job: JobId,
fence: Fence,
conclusion: Conclusion,
outputs: &BTreeMap<String, String>,
) -> Result<bool, RunError> {
let job = self.job(job)?;
Ok(job.fence() == fence
&& job.state() == JobState::Finished(conclusion)
&& job.outputs() == outputs)
}

pub fn lease(&self, job: JobId) -> Result<Option<Lease>, RunError> {
Ok(match self.job(job)?.state() {
JobState::Assigned(lease) | JobState::Running(lease) => Some(lease),
_ => None,
})
}

pub(crate) fn leased_jobs(&self) -> impl Iterator<Item = (JobId, Lease)> + '_ {
self.jobs.iter().filter_map(|job| match job.state() {
JobState::Assigned(lease) | JobState::Running(lease) => Some((job.id(), lease)),
_ => None,
})
}

pub(crate) fn single_job(&self) -> Option<JobId> {
(self.jobs.len() == 1).then(|| self.jobs[0].id())
}

pub const fn id(&self) -> RunId {
self.id
}

pub fn sequence(&self) -> Sequence {
self.log
.last()
.map_or_else(Sequence::default, RunRecord::sequence)
}

pub fn events(&self) -> impl Iterator<Item = RunRecord> {
self.log.iter().cloned()
}
}
use crate::job::Job;
use crate::{Dependency, Fence, JobId, JobState, Lease, QueuedJob, RunId, RunRecord, Sequence};
use std::collections::BTreeMap;

use super::{Conclusion, Run, RunError};

impl Run {
pub(super) fn advance(&mut self) {
loop {
let mut changes = Vec::new();
for job in self
.jobs
.iter()
.filter(|job| job.state() == JobState::Waiting)
{
let dependencies: Vec<&Job> = job
.needs()
.iter()
.flat_map(|key| {
self.jobs
.iter()
.filter(move |candidate| candidate.key() == key)
})
.collect();
if dependencies.iter().all(|dependency| {
matches!(
dependency.state(),
JobState::Finished(_) | JobState::Skipped
)
}) {
changes.push((job.id(), JobState::Queued));
}
}
if changes.is_empty() {
return;
}
for (job, state) in changes {
if let Ok(job) = self.job_mut(job) {
job.set_state(state);
}
}
}
}

pub(super) fn job(&self, id: JobId) -> Result<&Job, RunError> {
self.jobs
.iter()
.find(|job| job.id() == id)
.ok_or(RunError::UnknownJob {
run: self.id,
job: id,
})
}

pub(super) fn job_mut(&mut self, id: JobId) -> Result<&mut Job, RunError> {
self.jobs
.iter_mut()
.find(|job| job.id() == id)
.ok_or(RunError::UnknownJob {
run: self.id,
job: id,
})
}

pub(crate) fn queued_jobs(&self) -> impl Iterator<Item = &QueuedJob> {
self.jobs.iter().filter_map(|job| {
if job.state() != JobState::Queued {
return None;
}
let Some(limit) = job.matrix().max_parallel() else {
return Some(job.definition());
};
let active = self
.jobs
.iter()
.filter(|candidate| {
candidate.key() == job.key()
&& matches!(
candidate.state(),
JobState::Assigned(_) | JobState::Running(_)
)
})
.count() as u64;
(active < limit).then(|| job.definition())
})
}

pub(crate) fn active_jobs(&self) -> usize {
self.jobs
.iter()
.filter(|job| matches!(job.state(), JobState::Assigned(_) | JobState::Running(_)))
.count()
}

pub(crate) fn open_jobs(&self) -> usize {
self.jobs
.iter()
.filter(|job| {
matches!(
job.state(),
JobState::Waiting
| JobState::Queued
| JobState::Assigned(_)
| JobState::Running(_)
)
})
.count()
}

pub(crate) fn dependencies(&self, job: JobId) -> Result<Vec<Dependency>, RunError> {
let job = self.job(job)?;
job.needs()
.iter()
.map(|key| {
let states: Vec<JobState> = self
.jobs
.iter()
.filter(|candidate| candidate.key() == key)
.map(Job::state)
.collect();
let conclusion = if states
.iter()
.any(|state| matches!(state, JobState::Finished(Conclusion::Failure)))
{
Conclusion::Failure
} else if states
.iter()
.any(|state| matches!(state, JobState::Finished(Conclusion::Cancelled)))
{
Conclusion::Cancelled
} else if states.iter().any(|state| {
matches!(
state,
JobState::Finished(Conclusion::Skipped) | JobState::Skipped
)
}) {
Conclusion::Skipped
} else {
Conclusion::Success
};
let outputs = self
.jobs
.iter()
.filter(|candidate| candidate.key() == key)
.flat_map(|candidate| candidate.outputs().clone())
.collect();
Ok(Dependency::new(key.clone(), conclusion, outputs))
})
.collect()
}

pub(crate) fn jobs(&self) -> impl Iterator<Item = &QueuedJob> {
self.jobs.iter().map(Job::definition)
}

pub fn projected_jobs(&self) -> impl Iterator<Item = (&QueuedJob, JobState)> {
self.jobs.iter().map(|job| (job.definition(), job.state()))
}

pub(crate) fn job_definition(&self, job: JobId) -> Result<&QueuedJob, RunError> {
self.job(job).map(Job::definition)
}

pub fn job_state(&self, job: JobId) -> Result<JobState, RunError> {
self.job(job).map(Job::state)
}

pub(crate) fn repeats_finish(
&self,
job: JobId,
fence: Fence,
conclusion: Conclusion,
outputs: &BTreeMap<String, String>,
) -> Result<bool, RunError> {
let job = self.job(job)?;
Ok(job.fence() == fence
&& job.state() == JobState::Finished(conclusion)
&& job.outputs() == outputs)
}

pub fn lease(&self, job: JobId) -> Result<Option<Lease>, RunError> {
Ok(match self.job(job)?.state() {
JobState::Assigned(lease) | JobState::Running(lease) => Some(lease),
_ => None,
})
}

pub(crate) fn leased_jobs(&self) -> impl Iterator<Item = (JobId, Lease)> + '_ {
self.jobs.iter().filter_map(|job| match job.state() {
JobState::Assigned(lease) | JobState::Running(lease) => Some((job.id(), lease)),
_ => None,
})
}

pub(crate) fn single_job(&self) -> Option<JobId> {
(self.jobs.len() == 1).then(|| self.jobs[0].id())
}

pub const fn id(&self) -> RunId {
self.id
}

pub fn sequence(&self) -> Sequence {
self.log
.last()
.map_or_else(Sequence::default, RunRecord::sequence)
}

pub fn events(&self) -> impl Iterator<Item = RunRecord> {
self.log.iter().cloned()
}
}
crates/control-node/proto/actions.proto
+55
View File
@@ -1,0 +1,55 @@
syntax = "proto3";

package syncode.control.v1;

service ActionsRead {
rpc ListRuns(ListRunsRequest) returns (ListRunsResponse);
rpc GetRun(GetRunRequest) returns (ActionRun);
rpc GetJobLogs(GetJobLogsRequest) returns (GetJobLogsResponse);
}

message ListRunsRequest {
string session_token = 1;
string owner = 2;
string repository = 3;
}

message GetRunRequest {
string session_token = 1;
string run_id = 2;
}

message GetJobLogsRequest {
string session_token = 1;
string run_id = 2;
string job_id = 3;
}

message ListRunsResponse {
repeated ActionRun runs = 1;
}

message ActionRun {
string id = 1;
uint64 number = 2;
string repository_id = 3;
string commit = 4;
string reference = 5;
string event = 6;
string workflow = 7;
string state = 8;
string conclusion = 9;
repeated ActionJob jobs = 10;
}

message ActionJob {
string id = 1;
string key = 2;
string state = 3;
string conclusion = 4;
string node_id = 5;
}

message GetJobLogsResponse {
repeated string lines = 1;
}
crates/control-node/proto/identity.proto
+60
View File
@@ -1,0 +1,60 @@
syntax = "proto3";

package syncode.identity.v1;

service Identity {
rpc ValidateSession(ValidateSessionRequest) returns (ValidateSessionResponse);
rpc CheckCapability(CheckCapabilityRequest) returns (CheckCapabilityResponse);
rpc ResolveRepository(ResolveRepositoryRequest) returns (ResolveRepositoryResponse);
}

message ResolveRepositoryRequest {
string owner = 1;
string name = 2;
}

message ResolveRepositoryResponse {
string repository_id = 1;
}

enum PrincipalKind {
PRINCIPAL_KIND_UNSPECIFIED = 0;
PRINCIPAL_KIND_USER = 1;
PRINCIPAL_KIND_PLATFORM_AGENT = 2;
PRINCIPAL_KIND_LOCAL_AGENT = 3;
PRINCIPAL_KIND_ACCESS_TOKEN = 4;
}

enum ResourceKind {
RESOURCE_KIND_UNSPECIFIED = 0;
RESOURCE_KIND_ORGANIZATION = 1;
RESOURCE_KIND_REPOSITORY = 2;
RESOURCE_KIND_INSTANCE = 3;
}

message ValidateSessionRequest {
string session_token = 1;
}

message ValidateSessionResponse {
string principal_id = 1;
PrincipalKind principal_kind = 2;
int64 expires_at_unix = 3;
string owner_user_id = 4;
repeated string capabilities = 5;
string audience = 6;
ResourceKind resource_kind = 7;
string resource_id = 8;
}

message CheckCapabilityRequest {
string principal_id = 1;
PrincipalKind principal_kind = 2;
ResourceKind resource_kind = 3;
string resource_id = 4;
string capability = 5;
}

message CheckCapabilityResponse {
bool allowed = 1;
}
crates/control-runs/src/read.rs
+17
View File
@@ -1,0 +1,17 @@
use crate::{JobId, NodeId, Origin, ProjectedJobState, RunId, RunNumber};

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RunView {
pub run: RunId,
pub number: RunNumber,
pub origin: Origin,
pub jobs: Vec<JobView>,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct JobView {
pub job: JobId,
pub key: String,
pub state: ProjectedJobState,
pub node: Option<NodeId>,
}
crates/control-runs/src/registry/read.rs
+63
View File
@@ -1,0 +1,63 @@
use crate::{JobView, RunId, RunView};

use super::state::{ProjectionSeed, State};
use super::{Runs, RunsError};

impl State {
fn seeds_for_repository(&self, repository: &str) -> Vec<ProjectionSeed> {
self.entries
.iter()
.filter(|entry| entry.origin.repository() == repository)
.map(Self::projection_seed_of)
.collect()
}

fn seed_for_run(&self, run: RunId) -> Option<ProjectionSeed> {
self.entries
.iter()
.find(|entry| entry.run.id() == run)
.map(Self::projection_seed_of)
}
}

impl<L> Runs<L> {
pub async fn runs_for_repository(&self, repository: &str) -> Vec<RunView> {
let mut runs = self
.state
.lock()
.await
.seeds_for_repository(repository)
.into_iter()
.map(view)
.collect::<Vec<_>>();
runs.sort_by_key(|run| std::cmp::Reverse(run.number));
runs
}

pub async fn run_view(&self, run: RunId) -> Result<RunView, RunsError> {
self.state
.lock()
.await
.seed_for_run(run)
.map(view)
.ok_or(RunsError::UnknownRun(run))
}
}

fn view(seed: ProjectionSeed) -> RunView {
RunView {
run: seed.run,
number: seed.number,
origin: seed.origin,
jobs: seed
.jobs
.into_iter()
.map(|job| JobView {
job: job.job,
key: job.key,
state: job.state,
node: job.node,
})
.collect(),
}
}
crates/control-store/src/check_events.rs
+203
View File
@@ -1,0 +1,203 @@
use sqlx::{Postgres as Database, Row, Transaction};
use syncode_control_runs::{JobState, Origin, QueuedJob, Run, RunId, RunRecord};
use uuid::Uuid;

use crate::{Postgres, StoreError};

#[derive(Clone, Debug)]
pub struct PendingCheckEvent {
pub id: i64,
pub message_id: Uuid,
pub repository_id: Uuid,
pub payload: serde_json::Value,
pub attempts: i32,
}

pub(crate) async fn enqueue_check_snapshot(
transaction: &mut Transaction<'_, Database>,
run: RunId,
) -> Result<(), StoreError> {
let Some((repository_id, payload)) = snapshot(transaction, run).await? else {
return Ok(());
};
sqlx::query(
"INSERT INTO check_event_outbox (message_id, repository_id, payload) VALUES ($1, $2, $3)",
)
.bind(Uuid::new_v4())
.bind(repository_id)
.bind(payload)
.execute(&mut **transaction)
.await?;
Ok(())
}

async fn snapshot(
transaction: &mut Transaction<'_, Database>,
run_id: RunId,
) -> Result<Option<(Uuid, serde_json::Value)>, StoreError> {
let row = sqlx::query("SELECT number, origin FROM runs WHERE id = $1")
.bind(Uuid::from(run_id))
.fetch_one(&mut **transaction)
.await?;
let number: i64 = row.try_get("number")?;
let origin: Origin = serde_json::from_value(row.try_get("origin")?)?;
let Ok(repository_id) = Uuid::parse_str(origin.repository()) else {
return Ok(None);
};
let jobs = load_jobs(transaction, run_id).await?;
let records = load_records(transaction, run_id).await?;
let run = Run::replay(run_id, jobs, records)
.map_err(|error| StoreError::InvalidRun(error.to_string()))?;
let checks = run
.projected_jobs()
.map(|(job, state)| {
let (state, conclusion) = check_state(state);
serde_json::json!({
"job_id": job.id().to_string(),
"job": job.key(),
"state": state,
"conclusion": conclusion,
})
})
.collect::<Vec<_>>();
Ok(Some((
repository_id,
serde_json::json!({
"repository_id": repository_id,
"commit": origin.commit(),
"workflow": origin.workflow(),
"run_id": run_id.to_string(),
"run_number": number,
"run_sequence": run.sequence().get(),
"checks": checks,
}),
)))
}

async fn load_jobs(
transaction: &mut Transaction<'_, Database>,
run: RunId,
) -> Result<Vec<QueuedJob>, StoreError> {
let rows = sqlx::query(
"SELECT id, key, needs, matrix, priority, requirements, secrets, plan \
FROM run_jobs WHERE run = $1 ORDER BY position",
)
.bind(Uuid::from(run))
.fetch_all(&mut **transaction)
.await?;
rows.into_iter()
.map(|row| {
Ok(QueuedJob::new(
row.try_get::<Uuid, _>("id")?.into(),
row.try_get("key")?,
serde_json::from_value(row.try_get("needs")?)?,
serde_json::from_value(row.try_get("matrix")?)?,
row.try_get("plan")?,
)
.scheduled(
serde_json::from_value(row.try_get("priority")?)?,
serde_json::from_value(row.try_get("requirements")?)?,
)
.referencing(serde_json::from_value(row.try_get("secrets")?)?))
})
.collect()
}

async fn load_records(
transaction: &mut Transaction<'_, Database>,
run: RunId,
) -> Result<Vec<RunRecord>, StoreError> {
let rows =
sqlx::query("SELECT sequence, event FROM run_events WHERE run = $1 ORDER BY sequence")
.bind(Uuid::from(run))
.fetch_all(&mut **transaction)
.await?;
rows.into_iter()
.map(|row| {
let sequence: i64 = row.try_get("sequence")?;
let sequence = u64::try_from(sequence).map_err(|_| StoreError::Sequence(sequence))?;
Ok(RunRecord::new(
sequence.into(),
serde_json::from_value(row.try_get("event")?)?,
))
})
.collect()
}

fn check_state(state: JobState) -> (&'static str, Option<&'static str>) {
match state {
JobState::Waiting | JobState::Queued => ("waiting", None),
JobState::Assigned(_) => ("assigned", None),
JobState::Running(_) => ("running", None),
JobState::Finished(conclusion) => ("finished", Some(conclusion_name(conclusion))),
JobState::Skipped => ("skipped", Some("skipped")),
}
}

const fn conclusion_name(conclusion: syncode_control_runs::Conclusion) -> &'static str {
match conclusion {
syncode_control_runs::Conclusion::Success => "success",
syncode_control_runs::Conclusion::Failure => "failure",
syncode_control_runs::Conclusion::Cancelled => "cancelled",
syncode_control_runs::Conclusion::Skipped => "skipped",
}
}

impl Postgres {
pub async fn claim_check_events(
&self,
limit: i64,
lease: i64,
) -> Result<Vec<PendingCheckEvent>, StoreError> {
let rows = sqlx::query(
"UPDATE check_event_outbox SET attempts = attempts + 1, \
available_at = now() + make_interval(secs => $2::int) WHERE id IN ( \
SELECT id FROM check_event_outbox WHERE delivered_at IS NULL \
AND available_at <= now() ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED \
) RETURNING id, message_id, repository_id, payload, attempts",
)
.bind(limit)
.bind(lease)
.fetch_all(&self.pool)
.await?;
rows.into_iter()
.map(|row| {
Ok(PendingCheckEvent {
id: row.try_get("id")?,
message_id: row.try_get("message_id")?,
repository_id: row.try_get("repository_id")?,
payload: row.try_get("payload")?,
attempts: row.try_get("attempts")?,
})
})
.collect()
}

pub async fn mark_check_event_delivered(&self, id: i64) -> Result<(), StoreError> {
sqlx::query(
"UPDATE check_event_outbox SET delivered_at = now(), last_error = NULL WHERE id = $1",
)
.bind(id)
.execute(&self.pool)
.await?;
Ok(())
}

pub async fn mark_check_event_failed(
&self,
id: i64,
error: &str,
retry_in: i64,
) -> Result<(), StoreError> {
sqlx::query(
"UPDATE check_event_outbox SET last_error = $2, \
available_at = now() + make_interval(secs => $3::int) WHERE id = $1",
)
.bind(id)
.bind(error)
.bind(retry_in)
.execute(&self.pool)
.await?;
Ok(())
}
}
src/actions_read.rs
+237
View File
@@ -1,0 +1,237 @@
use syncode_control_node::ActionsReadService;
use syncode_control_node::actions_wire::{
ActionJob, ActionRun, GetJobLogsRequest, GetJobLogsResponse, GetRunRequest, ListRunsRequest,
ListRunsResponse,
};
use syncode_control_runs::{Conclusion, JobView, ProjectedJobState, RunId, RunLog, RunView, Runs};
use thiserror::Error;
use tonic::{Request, Response, Status};

#[derive(Debug, Error)]
pub enum AuthorizationError {
#[error("the session is not valid")]
Unauthenticated,
#[error("repository access is denied")]
Denied,
#[error("repository was not found")]
NotFound,
#[error("identity is unavailable: {0}")]
Unavailable(String),
}

#[tonic::async_trait]
pub trait ActionsAuthorization: Send + Sync + 'static {
async fn repository(
&self,
session_token: &str,
owner: &str,
name: &str,
) -> Result<String, AuthorizationError>;

async fn authorize(
&self,
session_token: &str,
repository_id: &str,
) -> Result<(), AuthorizationError>;
}

pub struct ActionsRead<L, A> {
runs: Runs<L>,
authorization: A,
}

impl<L, A> ActionsRead<L, A> {
pub const fn new(runs: Runs<L>, authorization: A) -> Self {
Self {
runs,
authorization,
}
}
}

#[tonic::async_trait]
impl<L, A> ActionsReadService for ActionsRead<L, A>
where
L: RunLog + Send + Sync + 'static,
A: ActionsAuthorization,
{
async fn list_runs(
&self,
request: Request<ListRunsRequest>,
) -> Result<Response<ListRunsResponse>, Status> {
let request = request.into_inner();
required(&request.session_token, "session token")?;
required(&request.owner, "repository owner")?;
required(&request.repository, "repository name")?;
let repository_id = self
.authorization
.repository(&request.session_token, &request.owner, &request.repository)
.await
.map_err(authorize_status)?;
let runs = self
.runs
.runs_for_repository(&repository_id)
.await
.into_iter()
.map(action_run)
.collect();
Ok(Response::new(ListRunsResponse { runs }))
}

async fn get_run(
&self,
request: Request<GetRunRequest>,
) -> Result<Response<ActionRun>, Status> {
let request = request.into_inner();
required(&request.session_token, "session token")?;
let run = parse_run(&request.run_id)?;
let view = self
.runs
.run_view(run)
.await
.map_err(|_| Status::not_found("run was not found"))?;
self.authorization
.authorize(&request.session_token, view.origin.repository())
.await
.map_err(authorize_status)?;
Ok(Response::new(action_run(view)))
}

async fn get_job_logs(
&self,
request: Request<GetJobLogsRequest>,
) -> Result<Response<GetJobLogsResponse>, Status> {
let request = request.into_inner();
required(&request.session_token, "session token")?;
let run = parse_run(&request.run_id)?;
let job = request
.job_id
.parse()
.map_err(|_| Status::invalid_argument("job id is invalid"))?;
let view = self
.runs
.run_view(run)
.await
.map_err(|_| Status::not_found("run was not found"))?;
if !view.jobs.iter().any(|candidate| candidate.job == job) {
return Err(Status::not_found("job was not found in the run"));
}
self.authorization
.authorize(&request.session_token, view.origin.repository())
.await
.map_err(authorize_status)?;
let lines = self
.runs
.lines(run, job)
.await
.map_err(|error| Status::unavailable(error.to_string()))?;
Ok(Response::new(GetJobLogsResponse { lines }))
}
}

fn required(value: &str, field: &'static str) -> Result<(), Status> {
if value.is_empty() {
Err(Status::invalid_argument(format!("{field} is required")))
} else {
Ok(())
}
}

fn parse_run(value: &str) -> Result<RunId, Status> {
value
.parse()
.map_err(|_| Status::invalid_argument("run id is invalid"))
}

fn authorize_status(error: AuthorizationError) -> Status {
match error {
AuthorizationError::Unauthenticated => Status::unauthenticated(error.to_string()),
AuthorizationError::Denied => Status::permission_denied(error.to_string()),
AuthorizationError::NotFound => Status::not_found(error.to_string()),
AuthorizationError::Unavailable(_) => Status::unavailable(error.to_string()),
}
}

fn action_run(run: RunView) -> ActionRun {
let (state, conclusion) = run_status(&run.jobs);
ActionRun {
id: run.run.to_string(),
number: run.number.get(),
repository_id: run.origin.repository().to_owned(),
commit: run.origin.commit().to_owned(),
reference: run.origin.reference().to_owned(),
event: run.origin.event().to_owned(),
workflow: run.origin.workflow().to_owned(),
state: state.to_owned(),
conclusion: conclusion.to_owned(),
jobs: run.jobs.into_iter().map(action_job).collect(),
}
}

fn action_job(job: JobView) -> ActionJob {
let (state, conclusion) = job_status(job.state);
ActionJob {
id: job.job.to_string(),
key: job.key,
state: state.to_owned(),
conclusion: conclusion.to_owned(),
node_id: job.node.map_or_else(String::new, |node| node.to_string()),
}
}

fn run_status(jobs: &[JobView]) -> (&'static str, &'static str) {
if jobs
.iter()
.any(|job| matches!(job.state, ProjectedJobState::Running))
{
return ("running", "");
}
if jobs
.iter()
.any(|job| matches!(job.state, ProjectedJobState::Assigned))
{
return ("assigned", "");
}
if jobs
.iter()
.any(|job| matches!(job.state, ProjectedJobState::Waiting))
{
return ("waiting", "");
}
let conclusion = if jobs
.iter()
.any(|job| matches!(job.state, ProjectedJobState::Finished(Conclusion::Failure)))
{
"failure"
} else if jobs.iter().any(|job| {
matches!(
job.state,
ProjectedJobState::Finished(Conclusion::Cancelled)
)
}) {
"cancelled"
} else if jobs.iter().all(|job| {
matches!(
job.state,
ProjectedJobState::Skipped | ProjectedJobState::Finished(Conclusion::Skipped)
)
}) {
"skipped"
} else {
"success"
};
("finished", conclusion)
}

const fn job_status(state: ProjectedJobState) -> (&'static str, &'static str) {
match state {
ProjectedJobState::Waiting => ("waiting", ""),
ProjectedJobState::Assigned => ("assigned", ""),
ProjectedJobState::Running => ("running", ""),
ProjectedJobState::Skipped => ("finished", "skipped"),
ProjectedJobState::Finished(Conclusion::Success) => ("finished", "success"),
ProjectedJobState::Finished(Conclusion::Failure) => ("finished", "failure"),
ProjectedJobState::Finished(Conclusion::Cancelled) => ("finished", "cancelled"),
ProjectedJobState::Finished(Conclusion::Skipped) => ("finished", "skipped"),
}
}
src/actions_read_identity.rs
+135
View File
@@ -1,0 +1,135 @@
use syncode_control_node::identity_wire::identity_client::IdentityClient;
use syncode_control_node::identity_wire::{
CheckCapabilityRequest, PrincipalKind, ResolveRepositoryRequest, ResourceKind,
ValidateSessionRequest,
};
use tonic::metadata::{Ascii, MetadataValue};
use tonic::transport::Channel;
use tonic::{Code, Request};

use crate::actions_read::{ActionsAuthorization, AuthorizationError};

#[derive(Clone)]
pub struct IdentityActionsAuthorization {
client: IdentityClient<Channel>,
authorization: MetadataValue<Ascii>,
}

impl IdentityActionsAuthorization {
pub async fn connect(
endpoint: String,
shared_secret: String,
) -> Result<Self, AuthorizationError> {
let authorization = format!("Bearer {shared_secret}").parse().map_err(|_| {
AuthorizationError::Unavailable("identity secret is invalid".to_owned())
})?;
let client = IdentityClient::connect(endpoint)
.await
.map_err(|error| AuthorizationError::Unavailable(error.to_string()))?;
Ok(Self {
client,
authorization,
})
}

async fn principal(
&self,
session_token: &str,
) -> Result<(String, PrincipalKind), AuthorizationError> {
let response = self
.client
.clone()
.validate_session(self.request(ValidateSessionRequest {
session_token: session_token.to_owned(),
}))
.await
.map_err(map_identity_status)?
.into_inner();
let kind = PrincipalKind::try_from(response.principal_kind).map_err(|_| {
AuthorizationError::Unavailable(
"identity returned an invalid principal kind".to_owned(),
)
})?;
Ok((response.principal_id, kind))
}

async fn require_read(
&self,
principal_id: String,
principal_kind: PrincipalKind,
repository_id: &str,
) -> Result<(), AuthorizationError> {
let allowed = self
.client
.clone()
.check_capability(self.request(CheckCapabilityRequest {
principal_id,
principal_kind: principal_kind.into(),
resource_kind: ResourceKind::Repository.into(),
resource_id: repository_id.to_owned(),
capability: "repo:read".to_owned(),
}))
.await
.map_err(map_identity_status)?
.into_inner()
.allowed;
if allowed {
Ok(())
} else {
Err(AuthorizationError::Denied)
}
}

fn request<T>(&self, body: T) -> Request<T> {
let mut request = Request::new(body);
request
.metadata_mut()
.insert("authorization", self.authorization.clone());
request
}
}

#[tonic::async_trait]
impl ActionsAuthorization for IdentityActionsAuthorization {
async fn repository(
&self,
session_token: &str,
owner: &str,
name: &str,
) -> Result<String, AuthorizationError> {
let (principal_id, principal_kind) = self.principal(session_token).await?;
let repository_id = self
.client
.clone()
.resolve_repository(self.request(ResolveRepositoryRequest {
owner: owner.to_owned(),
name: name.to_owned(),
}))
.await
.map_err(map_identity_status)?
.into_inner()
.repository_id;
self.require_read(principal_id, principal_kind, &repository_id)
.await?;
Ok(repository_id)
}

async fn authorize(
&self,
session_token: &str,
repository_id: &str,
) -> Result<(), AuthorizationError> {
let (principal_id, principal_kind) = self.principal(session_token).await?;
self.require_read(principal_id, principal_kind, repository_id)
.await
}
}

fn map_identity_status(status: tonic::Status) -> AuthorizationError {
match status.code() {
Code::Unauthenticated => AuthorizationError::Unauthenticated,
Code::PermissionDenied => AuthorizationError::Denied,
Code::NotFound => AuthorizationError::NotFound,
_ => AuthorizationError::Unavailable(status.to_string()),
}
}
src/check_events.rs
+145
View File
@@ -1,0 +1,145 @@
use std::time::Duration;

use hmac::{Hmac, KeyInit, Mac};
use serde::Serialize;
use sha2::Sha256;
use syncode_control_store::{PendingCheckEvent, Postgres};
use thiserror::Error;
use uuid::Uuid;

const MESSAGE_TYPE: &str = "check.status.changed";

pub struct CheckEventPublisher {
store: Postgres,
endpoint: reqwest::Url,
secret: String,
source_node_id: Uuid,
client: reqwest::Client,
interval: Duration,
batch: i64,
}

pub struct CheckEventPublisherConfig<'a> {
pub endpoint: &'a str,
pub secret: String,
pub source_node_id: Uuid,
pub interval: Duration,
pub batch: i64,
}

#[derive(Debug, Error)]
pub enum CheckEventPublisherError {
#[error("check event interval, batch, and signing secret must be non-empty")]
InvalidConfiguration,
#[error("check event endpoint is invalid: {0}")]
InvalidEndpoint(#[from] url::ParseError),
#[error("check event HTTP client cannot be created: {0}")]
HttpClient(#[from] reqwest::Error),
#[error("check event outbox failed: {0}")]
Store(#[from] syncode_control_store::StoreError),
}

#[derive(Serialize)]
struct Envelope<'a> {
message_id: Uuid,
protocol_version: &'static str,
schema_version: u32,
source_node_id: Uuid,
repository_id: Uuid,
message_type: &'static str,
sequence: i64,
term: i64,
payload: &'a serde_json::Value,
}

impl CheckEventPublisher {
pub fn new(
store: Postgres,
config: CheckEventPublisherConfig<'_>,
) -> Result<Self, CheckEventPublisherError> {
if config.interval.is_zero() || config.batch <= 0 || config.secret.is_empty() {
return Err(CheckEventPublisherError::InvalidConfiguration);
}
Ok(Self {
store,
endpoint: config.endpoint.parse()?,
secret: config.secret,
source_node_id: config.source_node_id,
client: reqwest::Client::builder()
.timeout(Duration::from_secs(10))
.build()?,
interval: config.interval,
batch: config.batch,
})
}

pub async fn run(self) -> Result<(), CheckEventPublisherError> {
let mut ticker = tokio::time::interval(self.interval);
loop {
ticker.tick().await;
self.publish_once().await?;
}
}

pub async fn publish_once(&self) -> Result<(), CheckEventPublisherError> {
let lease = i64::try_from(self.interval.as_secs().max(30)).unwrap_or(i64::MAX);
let pending = self.store.claim_check_events(self.batch, lease).await?;
for message in pending {
match self.publish(&message).await {
Ok(()) => self.store.mark_check_event_delivered(message.id).await?,
Err(error) => {
let retry = retry_delay(message.attempts);
self.store
.mark_check_event_failed(message.id, &error, retry)
.await?;
}
}
}
Ok(())
}

async fn publish(&self, message: &PendingCheckEvent) -> Result<(), String> {
let body = serde_json::to_vec(&Envelope {
message_id: message.message_id,
protocol_version: "1.0",
schema_version: 1,
source_node_id: self.source_node_id,
repository_id: message.repository_id,
message_type: MESSAGE_TYPE,
sequence: message.id,
term: 1,
payload: &message.payload,
})
.map_err(|error| error.to_string())?;
let mut mac = Hmac::<Sha256>::new_from_slice(self.secret.as_bytes())
.map_err(|error| error.to_string())?;
mac.update(&body);
let signature = hex::encode(mac.finalize().into_bytes());
let response = self
.client
.post(self.endpoint.clone())
.header("x-syncode-event", MESSAGE_TYPE)
.header("x-syncode-delivery", message.message_id.to_string())
.header("x-syncode-signature", signature)
.header("content-type", "application/json")
.body(body)
.send()
.await
.map_err(|error| error.to_string())?;
if response.status().is_success() {
Ok(())
} else {
Err(format!(
"check event endpoint returned HTTP {}",
response.status().as_u16()
))
}
}
}

fn retry_delay(attempts: i32) -> i64 {
let exponent = u32::try_from(attempts.saturating_sub(1))
.unwrap_or(u32::MAX)
.min(8);
1_i64.checked_shl(exponent).unwrap_or(300).min(300)
}
tests/actions_read.rs
+149
View File
@@ -1,0 +1,149 @@
#![allow(clippy::expect_used)]

use std::error::Error;

use syncode_control::actions_read::{ActionsAuthorization, ActionsRead, AuthorizationError};
use syncode_control_node::GeneratedActionsReadServer;
use syncode_control_node::actions_wire::actions_read_client::ActionsReadClient;
use syncode_control_node::actions_wire::{GetJobLogsRequest, GetRunRequest, ListRunsRequest};
use syncode_control_runs::{Forgotten, JobId, NodeId, Origin, Runs};
use tokio::net::TcpListener;
use tokio_stream::wrappers::TcpListenerStream;
use tonic::Code;
use tonic::transport::Server;

type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

const REPOSITORY_ID: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89214";

#[derive(Clone)]
struct Authorization;

#[tonic::async_trait]
impl ActionsAuthorization for Authorization {
async fn repository(
&self,
session_token: &str,
owner: &str,
name: &str,
) -> Result<String, AuthorizationError> {
if session_token != "valid-session" {
return Err(AuthorizationError::Unauthenticated);
}
if owner == "syncode" && name == "control" {
Ok(REPOSITORY_ID.to_owned())
} else {
Err(AuthorizationError::NotFound)
}
}

async fn authorize(
&self,
session_token: &str,
repository_id: &str,
) -> Result<(), AuthorizationError> {
if session_token != "valid-session" {
return Err(AuthorizationError::Unauthenticated);
}
if repository_id == REPOSITORY_ID {
Ok(())
} else {
Err(AuthorizationError::Denied)
}
}
}

async fn server(runs: Runs<Forgotten>) -> TestResult<ActionsReadClient<tonic::transport::Channel>> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let endpoint = format!("http://{}", listener.local_addr()?);
tokio::spawn(async move {
let _ = Server::builder()
.add_service(GeneratedActionsReadServer::new(ActionsRead::new(
runs,
Authorization,
)))
.serve_with_incoming(TcpListenerStream::new(listener))
.await;
});
Ok(ActionsReadClient::connect(endpoint).await?)
}

#[tokio::test]
async fn lists_views_and_reads_logs_through_the_public_contract() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let job = JobId::fresh();
let run = runs
.queue(
job,
Origin::new(
REPOSITORY_ID.to_owned(),
"abc123".to_owned(),
"refs/heads/main".to_owned(),
"push".to_owned(),
".syncode/workflows/ci.yml".to_owned(),
),
b"plan".to_vec(),
)
.await?;
let node = NodeId::fresh();
let assignment = runs.take_next(node).await?.expect("assignment");
runs.logged(
run,
job,
node,
assignment.fence(),
0,
&["first".to_owned(), "second".to_owned()],
)
.await?;

let mut client = server(runs).await?;
let listed = client
.list_runs(ListRunsRequest {
session_token: "valid-session".to_owned(),
owner: "syncode".to_owned(),
repository: "control".to_owned(),
})
.await?
.into_inner();
assert_eq!(1, listed.runs.len());
assert_eq!(run.to_string(), listed.runs[0].id);
assert_eq!("assigned", listed.runs[0].state);

let viewed = client
.get_run(GetRunRequest {
session_token: "valid-session".to_owned(),
run_id: run.to_string(),
})
.await?
.into_inner();
assert_eq!("abc123", viewed.commit);
assert_eq!(job.to_string(), viewed.jobs[0].id);

let logs = client
.get_job_logs(GetJobLogsRequest {
session_token: "valid-session".to_owned(),
run_id: run.to_string(),
job_id: job.to_string(),
})
.await?
.into_inner();
assert_eq!(vec!["first", "second"], logs.lines);
Ok(())
}

#[tokio::test]
async fn rejects_an_invalid_session() -> TestResult {
let runs = Runs::restored(Forgotten::default()).await?;
let mut client = server(runs).await?;
let error = client
.list_runs(ListRunsRequest {
session_token: "invalid-session".to_owned(),
owner: "syncode".to_owned(),
repository: "control".to_owned(),
})
.await
.expect_err("invalid session must be rejected");
assert_eq!(Code::Unauthenticated, error.code());
Ok(())
}
tests/check_events.rs
+103
View File
@@ -1,0 +1,103 @@
#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]

use std::error::Error;
use std::sync::{Arc, Mutex};

use axum::body::Bytes;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::routing::post;
use hmac::{Hmac, KeyInit, Mac};
use sha2::Sha256;
use sqlx::{AssertSqlSafe, Connection, Executor, PgConnection};
use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig};
use syncode_control_runs::{JobId, Origin, Runs};
use syncode_control_store::Postgres;
use uuid::Uuid;

type Recorded = Arc<Mutex<Option<(HeaderMap, Vec<u8>)>>>;
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;

async fn record(State(recorded): State<Recorded>, headers: HeaderMap, body: Bytes) -> StatusCode {
recorded
.lock()
.map(|mut value| *value = Some((headers, body.to_vec())))
.map_or(StatusCode::INTERNAL_SERVER_ERROR, |()| {
StatusCode::NO_CONTENT
})
}

async fn fresh() -> TestResult<String> {
let base = std::env::var("SYNCODE_TEST_DATABASE_URL")?;
let schema = format!("check_events_{}", Uuid::new_v4().simple());
let mut connection = PgConnection::connect(&base).await?;
connection
.execute(AssertSqlSafe(format!("CREATE SCHEMA \"{schema}\"")))
.await?;
connection.close().await?;
let separator = if base.contains('?') { '&' } else { '?' };
Ok(format!(
"{base}{separator}options=-c%20search_path%3D{schema}"
))
}

#[tokio::test]
async fn check_status_change_is_delivered_and_acknowledged() -> TestResult {
let store = Postgres::connect(&fresh().await?, 2).await?;
let repository_id = Uuid::new_v4();
let origin = Origin::new(
repository_id.to_string(),
"1111111111111111111111111111111111111111".to_owned(),
"refs/heads/main".to_owned(),
"push".to_owned(),
".gitea/workflows/ci.yml".to_owned(),
);
Runs::restored(store.clone())
.await?
.queue(JobId::fresh(), origin, br#"{"schema":1}"#.to_vec())
.await?;

let recorded = Recorded::default();
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
let endpoint = format!("http://{}/events", listener.local_addr()?);
let app = axum::Router::new()
.route("/events", post(record))
.with_state(Arc::clone(&recorded));
let server = tokio::spawn(async move { axum::serve(listener, app).await });

CheckEventPublisher::new(
store.clone(),
CheckEventPublisherConfig {
endpoint: &endpoint,
secret: "check-secret".to_owned(),
source_node_id: Uuid::new_v4(),
interval: std::time::Duration::from_secs(1),
batch: 10,
},
)?
.publish_once()
.await?;

let (headers, body) = recorded
.lock()
.map_err(|_| "recorded event lock was poisoned")?
.clone()
.ok_or("check event was not delivered")?;
let signature = hex::decode(headers["x-syncode-signature"].to_str()?)?;
let mut mac = Hmac::<Sha256>::new_from_slice(b"check-secret")?;
mac.update(&body);
mac.verify_slice(&signature)?;
let envelope: serde_json::Value = serde_json::from_slice(&body)?;
assert_eq!(envelope["protocol_version"], "1.0");
assert_eq!(envelope["repository_id"], repository_id.to_string());
assert_eq!(envelope["message_type"], "check.status.changed");
assert_eq!(envelope["payload"]["checks"][0]["state"], "waiting");
let delivered = sqlx::query_scalar::<_, bool>(
"SELECT delivered_at IS NOT NULL FROM check_event_outbox LIMIT 1",
)
.fetch_one(store.pool())
.await?;
assert!(delivered);
server.abort();
Ok(())
}