fix: Remove legacy run projection #51
File diff suppressed because it is too large
Load Diff
@@ -1,37 +1,37 @@
|
||||
# SynCode Control Plane
|
||||
|
||||
`syncode-control` is the authority over CI runs in SynCode: it triggers runs,
|
||||
resolves and compiles workflows, owns the run aggregate, keeps the queue,
|
||||
assigns work to nodes, holds leases, collects state and logs, issues secrets on
|
||||
demand, enforces quotas, and records the audit trail.
|
||||
|
||||
It replaces the Actions control plane the forge used to run. After 0.3 a node
|
||||
does not talk to the forge at all: it enrolls here, receives a compiled
|
||||
execution plan here, and reports here. The forge keeps Git hosting, the secret
|
||||
store, artifact hosting, the run UI reading a one-way projection, and the event
|
||||
feed this service triggers runs from.
|
||||
|
||||
## Boundaries
|
||||
|
||||
A workflow is compiled exactly once, in this service, with the shared compiler
|
||||
from [`syncode/workflow`](https://syncode.sh/syncode/workflow). The assignment
|
||||
carries a versioned execution plan, not a workflow file: the runner never sees
|
||||
workflow source and rejects a plan whose schema version it does not support.
|
||||
|
||||
What stays on the node is what cannot be decided before the job runs: fetching
|
||||
actions, expanding composite actions, and evaluating expressions that depend on
|
||||
step results, job status, and the environment.
|
||||
|
||||
## Development
|
||||
|
||||
```sh
|
||||
cargo fmt --check
|
||||
cargo clippy --workspace --all-targets --all-features -- -D warnings
|
||||
cargo test --workspace --all-targets --all-features
|
||||
./scripts/check-architecture.sh
|
||||
./scripts/check-rust-loc.sh
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
MIT. See [LICENSE](LICENSE).
|
||||
# SynCode Control Plane
|
||||
|
||||
`syncode-control` is the authority over CI runs in SynCode: it triggers runs,
|
||||
resolves and compiles workflows, owns the run aggregate, keeps the queue,
|
||||
assigns work to nodes, holds leases, collects state and logs, issues secrets on
|
||||
demand, enforces quotas, and records the audit trail.
|
||||
|
||||
It replaces the Actions control plane the forge used to run. After 0.3 a node
|
||||
does not talk to the forge at all: it enrolls here, receives a compiled
|
||||
execution plan here, and reports here. The SynCode UI reads runs and logs from
|
||||
this service. The forge temporarily keeps the action mirror, secret store, and
|
||||
the event feed this service triggers runs from.
|
||||
|
||||
## Boundaries
|
||||
|
||||
A workflow is compiled exactly once, in this service, with the shared compiler
|
||||
from [`syncode/workflow`](https://syncode.sh/syncode/workflow). The assignment
|
||||
carries a versioned execution plan, not a workflow file: the runner never sees
|
||||
workflow source and rejects a plan whose schema version it does not support.
|
||||
|
||||
What stays on the node is what cannot be decided before the job runs: fetching
|
||||
actions, expanding composite actions, and evaluating expressions that depend on
|
||||
step results, job status, and the environment.
|
||||
|
||||
## Development
|
||||
|
||||
```sh
|
||||
cargo fmt --check
|
||||
cargo clippy --workspace --all-targets --all-features -- -D warnings
|
||||
cargo test --workspace --all-targets --all-features
|
||||
./scripts/check-architecture.sh
|
||||
./scripts/check-rust-loc.sh
|
||||
```
|
||||
|
||||
## License
|
||||
|
||||
MIT. See [LICENSE](LICENSE).
|
||||
@@ -1,28 +1,27 @@
|
||||
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_credentials;
|
||||
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;
|
||||
#[path = "forge.rs"]
|
||||
pub mod repository;
|
||||
pub mod repository_credentials;
|
||||
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;
|
||||
+4
-18
@@ -1,392 +1,378 @@
|
||||
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::NativeActionRepository;
|
||||
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_credentials::IdentityRepositoryCredentials;
|
||||
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_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).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,
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)?;
|
||||
let secret_service = RuntimeSecrets::new(
|
||||
runs.clone(),
|
||||
nodes.clone(),
|
||||
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
|
||||
IdentityRepositoryCredentials::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
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;
|
||||
}
|
||||
}
|
||||
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::NativeActionRepository;
|
||||
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::repository::{RepositoryContents, RepositorySecrets};
|
||||
use syncode_control::repository_credentials::IdentityRepositoryCredentials;
|
||||
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, RepositoryCoordinates,
|
||||
};
|
||||
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,
|
||||
|
||||
/// 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,
|
||||
}),
|
||||
))
|
||||
.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_repository = NativeActionRepository::connect(
|
||||
arguments.repository_grpc.clone(),
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?;
|
||||
let action_resolver = ActionResolver::new(
|
||||
action_repository,
|
||||
action_store.clone(),
|
||||
PinnedOciResolver::new(arguments.action_oci_registries)?,
|
||||
action_mirror,
|
||||
action_allowlist,
|
||||
);
|
||||
let native_repository =
|
||||
NativeRepositoryContents::connect(arguments.repository_grpc.clone()).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 repositories = RepositoryCoordinates::new(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)?;
|
||||
let secret_service = RuntimeSecrets::new(
|
||||
runs.clone(),
|
||||
nodes.clone(),
|
||||
RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token),
|
||||
IdentityRepositoryCredentials::connect(
|
||||
arguments.identity_grpc.clone(),
|
||||
arguments.identity_shared_secret.clone(),
|
||||
)
|
||||
.await?,
|
||||
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,
|
||||
repositories.clone(),
|
||||
),
|
||||
Mode::Active => NodeSessionServer::new(
|
||||
runs.clone(),
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
arguments.action_artifact_public_url,
|
||||
repositories,
|
||||
),
|
||||
};
|
||||
tokio::select! {
|
||||
served = axum::serve(events, http).into_future() => 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;
|
||||
}
|
||||
}
|
||||
+3
-116
File diff suppressed because it is too large
Load Diff
+6
-16
File diff suppressed because it is too large
Load Diff
@@ -1,35 +1,34 @@
|
||||
[package]
|
||||
name = "syncode-control-node"
|
||||
description = "The gRPC stream a node opens to the control plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[dependencies]
|
||||
base64 = "0.22.1"
|
||||
hmac = "0.12.1"
|
||||
prost = "0.14.4"
|
||||
reqwest = { version = "0.13.4", default-features = false, features = ["json", "rustls"] }
|
||||
serde = { version = "1.0.229", features = ["derive"] }
|
||||
serde_json = "1.0.149"
|
||||
sha2 = "0.10.9"
|
||||
syncode-control-nodes = { path = "../control-nodes" }
|
||||
syncode-control-runs = { path = "../control-runs" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["sync"] }
|
||||
tokio-stream = "0.1.18"
|
||||
tonic = "0.14.6"
|
||||
tonic-prost = "0.14.6"
|
||||
url = "2.5.8"
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[build-dependencies]
|
||||
prost-build = "0.14.4"
|
||||
protoc-bin-vendored = "3.2.0"
|
||||
tonic-prost-build = "0.14.6"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
[package]
|
||||
name = "syncode-control-node"
|
||||
description = "The gRPC stream a node opens to the control plane"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
rust-version.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish = false
|
||||
|
||||
[dependencies]
|
||||
base64 = "0.22.1"
|
||||
hmac = "0.12.1"
|
||||
prost = "0.14.4"
|
||||
serde = { version = "1.0.229", features = ["derive"] }
|
||||
serde_json = "1.0.149"
|
||||
sha2 = "0.10.9"
|
||||
syncode-control-nodes = { path = "../control-nodes" }
|
||||
syncode-control-runs = { path = "../control-runs" }
|
||||
thiserror = "2.0.19"
|
||||
tokio = { version = "1.49.0", features = ["sync"] }
|
||||
tokio-stream = "0.1.18"
|
||||
tonic = "0.14.6"
|
||||
tonic-prost = "0.14.6"
|
||||
url = "2.5.8"
|
||||
uuid = { version = "1.24.0", features = ["v4"] }
|
||||
|
||||
[build-dependencies]
|
||||
prost-build = "0.14.4"
|
||||
protoc-bin-vendored = "3.2.0"
|
||||
tonic-prost-build = "0.14.6"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
@@ -1,110 +1,106 @@
|
||||
use prost::DecodeError;
|
||||
use syncode_control_nodes::NodesError;
|
||||
use syncode_control_runs::{IdentityError, RunsError};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::ArtifactTokenError;
|
||||
use crate::ProjectionRequestError;
|
||||
use tonic::Status;
|
||||
|
||||
use crate::CapabilityError;
|
||||
use crate::declaration::DeclarationError;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum SessionError {
|
||||
#[error("node message has no message_id or idempotency_key")]
|
||||
Unidentified,
|
||||
|
||||
#[error("node message_id {0:?} was reused for different content")]
|
||||
MessageIdConflict(String),
|
||||
|
||||
#[error("node idempotency_key {0:?} was reused for different content")]
|
||||
IdempotencyConflict(String),
|
||||
|
||||
#[error("node declared invalid parallelism {0}")]
|
||||
InvalidParallelism(u32),
|
||||
|
||||
#[error("node {0} already has an active session")]
|
||||
ConcurrentSession(syncode_control_runs::NodeId),
|
||||
|
||||
#[error("node message sequence {actual} arrived while {expected} was required")]
|
||||
OutOfSequence { expected: u64, actual: u64 },
|
||||
#[error(transparent)]
|
||||
Identity(#[from] IdentityError),
|
||||
|
||||
#[error(transparent)]
|
||||
Service(#[from] RunsError),
|
||||
|
||||
#[error(transparent)]
|
||||
Nodes(#[from] NodesError),
|
||||
|
||||
#[error(transparent)]
|
||||
Declaration(#[from] DeclarationError),
|
||||
|
||||
#[error(transparent)]
|
||||
Capability(#[from] CapabilityError),
|
||||
|
||||
#[error(transparent)]
|
||||
ArtifactToken(#[from] ArtifactTokenError),
|
||||
|
||||
#[error(transparent)]
|
||||
Projection(#[from] ProjectionRequestError),
|
||||
|
||||
#[error("run {0} has no projection before dispatch")]
|
||||
MissingProjection(syncode_control_runs::RunId),
|
||||
|
||||
#[error("a node must say what it can do and how much room it has")]
|
||||
Undeclared,
|
||||
|
||||
#[error("the node has not said hello on this stream")]
|
||||
NoHello,
|
||||
|
||||
#[error("node message carried no body")]
|
||||
EmptyMessage,
|
||||
|
||||
#[error("progress reported an unknown conclusion {0}")]
|
||||
UnknownConclusion(i32),
|
||||
|
||||
#[error("the node stopped reading its stream")]
|
||||
Unreadable,
|
||||
|
||||
#[error(transparent)]
|
||||
Transport(#[from] Status),
|
||||
|
||||
#[error("a stored control message cannot be decoded: {0}")]
|
||||
Outbox(DecodeError),
|
||||
|
||||
#[error("durable control message {0:?} disappeared from the outbox")]
|
||||
MissingOutbox(String),
|
||||
}
|
||||
|
||||
impl From<SessionError> for Status {
|
||||
fn from(error: SessionError) -> Self {
|
||||
match error {
|
||||
SessionError::Transport(status) => status,
|
||||
SessionError::Unidentified
|
||||
| SessionError::MessageIdConflict(_)
|
||||
| SessionError::IdempotencyConflict(_)
|
||||
| SessionError::Identity(_)
|
||||
| SessionError::EmptyMessage
|
||||
| SessionError::InvalidParallelism(_)
|
||||
| SessionError::OutOfSequence { .. }
|
||||
| SessionError::UnknownConclusion(_) => Self::invalid_argument(error.to_string()),
|
||||
SessionError::NoHello | SessionError::ConcurrentSession(_) => {
|
||||
Self::failed_precondition(error.to_string())
|
||||
}
|
||||
SessionError::Unreadable => Self::unavailable(error.to_string()),
|
||||
SessionError::Undeclared | SessionError::Declaration(_) => {
|
||||
Self::invalid_argument(error.to_string())
|
||||
}
|
||||
SessionError::Capability(_)
|
||||
| SessionError::ArtifactToken(_)
|
||||
| SessionError::Projection(_)
|
||||
| SessionError::MissingProjection(_) => Self::internal(error.to_string()),
|
||||
SessionError::Nodes(_) => Self::permission_denied(error.to_string()),
|
||||
SessionError::Service(_) | SessionError::Outbox(_) | SessionError::MissingOutbox(_) => {
|
||||
Self::internal(error.to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
use prost::DecodeError;
|
||||
use syncode_control_nodes::NodesError;
|
||||
use syncode_control_runs::{IdentityError, RunsError};
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::ArtifactTokenError;
|
||||
use crate::RepositoryCoordinateRequestError;
|
||||
use tonic::Status;
|
||||
|
||||
use crate::CapabilityError;
|
||||
use crate::declaration::DeclarationError;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum SessionError {
|
||||
#[error("node message has no message_id or idempotency_key")]
|
||||
Unidentified,
|
||||
|
||||
#[error("node message_id {0:?} was reused for different content")]
|
||||
MessageIdConflict(String),
|
||||
|
||||
#[error("node idempotency_key {0:?} was reused for different content")]
|
||||
IdempotencyConflict(String),
|
||||
|
||||
#[error("node declared invalid parallelism {0}")]
|
||||
InvalidParallelism(u32),
|
||||
|
||||
#[error("node {0} already has an active session")]
|
||||
ConcurrentSession(syncode_control_runs::NodeId),
|
||||
|
||||
#[error("node message sequence {actual} arrived while {expected} was required")]
|
||||
OutOfSequence { expected: u64, actual: u64 },
|
||||
#[error(transparent)]
|
||||
Identity(#[from] IdentityError),
|
||||
|
||||
#[error(transparent)]
|
||||
Service(#[from] RunsError),
|
||||
|
||||
#[error(transparent)]
|
||||
Nodes(#[from] NodesError),
|
||||
|
||||
#[error(transparent)]
|
||||
Declaration(#[from] DeclarationError),
|
||||
|
||||
#[error(transparent)]
|
||||
Capability(#[from] CapabilityError),
|
||||
|
||||
#[error(transparent)]
|
||||
ArtifactToken(#[from] ArtifactTokenError),
|
||||
|
||||
#[error(transparent)]
|
||||
RepositoryCoordinates(#[from] RepositoryCoordinateRequestError),
|
||||
|
||||
#[error("a node must say what it can do and how much room it has")]
|
||||
Undeclared,
|
||||
|
||||
#[error("the node has not said hello on this stream")]
|
||||
NoHello,
|
||||
|
||||
#[error("node message carried no body")]
|
||||
EmptyMessage,
|
||||
|
||||
#[error("progress reported an unknown conclusion {0}")]
|
||||
UnknownConclusion(i32),
|
||||
|
||||
#[error("the node stopped reading its stream")]
|
||||
Unreadable,
|
||||
|
||||
#[error(transparent)]
|
||||
Transport(#[from] Status),
|
||||
|
||||
#[error("a stored control message cannot be decoded: {0}")]
|
||||
Outbox(DecodeError),
|
||||
|
||||
#[error("durable control message {0:?} disappeared from the outbox")]
|
||||
MissingOutbox(String),
|
||||
}
|
||||
|
||||
impl From<SessionError> for Status {
|
||||
fn from(error: SessionError) -> Self {
|
||||
match error {
|
||||
SessionError::Transport(status) => status,
|
||||
SessionError::Unidentified
|
||||
| SessionError::MessageIdConflict(_)
|
||||
| SessionError::IdempotencyConflict(_)
|
||||
| SessionError::Identity(_)
|
||||
| SessionError::EmptyMessage
|
||||
| SessionError::InvalidParallelism(_)
|
||||
| SessionError::OutOfSequence { .. }
|
||||
| SessionError::UnknownConclusion(_) => Self::invalid_argument(error.to_string()),
|
||||
SessionError::NoHello | SessionError::ConcurrentSession(_) => {
|
||||
Self::failed_precondition(error.to_string())
|
||||
}
|
||||
SessionError::Unreadable => Self::unavailable(error.to_string()),
|
||||
SessionError::Undeclared | SessionError::Declaration(_) => {
|
||||
Self::invalid_argument(error.to_string())
|
||||
}
|
||||
SessionError::Capability(_)
|
||||
| SessionError::ArtifactToken(_)
|
||||
| SessionError::RepositoryCoordinates(_) => Self::internal(error.to_string()),
|
||||
SessionError::Nodes(_) => Self::permission_denied(error.to_string()),
|
||||
SessionError::Service(_) | SessionError::Outbox(_) | SessionError::MissingOutbox(_) => {
|
||||
Self::internal(error.to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,43 +1,45 @@
|
||||
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;
|
||||
|
||||
pub const REPOSITORY_TOKEN_SECRET: &str = "SYNCODE_REPOSITORY_TOKEN";
|
||||
mod artifact;
|
||||
mod capability;
|
||||
mod declaration;
|
||||
mod error;
|
||||
mod identity;
|
||||
mod outbound;
|
||||
mod reports;
|
||||
mod repository_coordinates;
|
||||
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 repository_coordinates::{
|
||||
RepositoryCoordinateRequestError, RepositoryCoordinates, RepositoryCoordinatesError,
|
||||
};
|
||||
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;
|
||||
|
||||
pub const REPOSITORY_TOKEN_SECRET: &str = "SYNCODE_REPOSITORY_TOKEN";
|
||||
@@ -1,243 +1,243 @@
|
||||
mod acknowledgement;
|
||||
mod dispatch;
|
||||
mod obsolete;
|
||||
mod server;
|
||||
mod state;
|
||||
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use prost::Message;
|
||||
use syncode_control_nodes::{InboxDecision, NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::Status;
|
||||
use url::Url;
|
||||
|
||||
use crate::error::SessionError;
|
||||
use crate::identity::{enrol_node, greet, renew_credential};
|
||||
use crate::outbound::{Outbound, send};
|
||||
use crate::reports::{accept_logs, report};
|
||||
use crate::wire::{ControlMessage, NodeMessage, Welcome, control_message, node_message};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority, ProjectionClient};
|
||||
use acknowledgement::{acknowledge_node_message, acknowledgement};
|
||||
use dispatch::{assignment_is_current, beat, holder, offer_work};
|
||||
use obsolete::obsolete_report;
|
||||
use state::Session;
|
||||
|
||||
const CHANNEL_DEPTH: usize = 32;
|
||||
const OUTBOUND_RETRY_INTERVAL: Duration = Duration::from_secs(2);
|
||||
|
||||
pub struct NodeSessionServer<L, S> {
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: Url,
|
||||
projection: ProjectionClient,
|
||||
assignments: bool,
|
||||
}
|
||||
|
||||
struct Registries<'a, L, S> {
|
||||
runs: &'a Runs<L>,
|
||||
nodes: &'a Nodes<S>,
|
||||
authority: &'a CapabilityAuthority,
|
||||
artifact_authority: &'a ArtifactTokenAuthority,
|
||||
action_artifact_url: &'a Url,
|
||||
projection: &'a ProjectionClient,
|
||||
assignments: bool,
|
||||
}
|
||||
|
||||
impl<L, S> Clone for Registries<'_, L, S> {
|
||||
fn clone(&self) -> Self {
|
||||
*self
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, S> Copy for Registries<'_, L, S> {}
|
||||
|
||||
async fn serve<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
message: NodeMessage,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
let control = registries.runs;
|
||||
if !session.accept(&message)? {
|
||||
acknowledge_node_message(outbound, &message, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
let tracked = matches!(
|
||||
message.body.as_ref(),
|
||||
Some(
|
||||
node_message::Body::Progress(_)
|
||||
| node_message::Body::Heartbeat(_)
|
||||
| node_message::Body::RotationAccepted(_)
|
||||
| node_message::Body::Logs(_)
|
||||
)
|
||||
);
|
||||
let node = tracked.then(|| holder(session.node)).transpose()?;
|
||||
if let Some(node) = node {
|
||||
let payload = NodeMessage {
|
||||
sequence: 0,
|
||||
message_id: String::new(),
|
||||
idempotency_key: String::new(),
|
||||
body: message.body.clone(),
|
||||
}
|
||||
.encode_to_vec();
|
||||
match registries
|
||||
.nodes
|
||||
.message_started(
|
||||
node,
|
||||
&message.message_id,
|
||||
&message.idempotency_key,
|
||||
&payload,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
InboxDecision::New => {}
|
||||
InboxDecision::Duplicate => {
|
||||
acknowledge_node_message(outbound, &message, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
InboxDecision::MessageIdConflict => {
|
||||
return Err(SessionError::MessageIdConflict(message.message_id));
|
||||
}
|
||||
InboxDecision::IdempotencyConflict => {
|
||||
return Err(SessionError::IdempotencyConflict(message.idempotency_key));
|
||||
}
|
||||
}
|
||||
}
|
||||
let acknowledgement = acknowledgement(&message);
|
||||
let progress = matches!(message.body.as_ref(), Some(node_message::Body::Progress(_)));
|
||||
let result = match message.body {
|
||||
Some(node_message::Body::Enrol(enrol)) => {
|
||||
enrol_node(registries.nodes, outbound, &enrol, sender).await
|
||||
}
|
||||
Some(node_message::Body::Hello(hello)) => {
|
||||
let identity: syncode_control_runs::NodeId = hello.node.parse()?;
|
||||
registries
|
||||
.nodes
|
||||
.authenticate(identity, &hello.credential, SystemTime::now())
|
||||
.await?;
|
||||
if !registries.nodes.session_opened(identity).await {
|
||||
return Err(SessionError::ConcurrentSession(identity));
|
||||
}
|
||||
if let Err(error) = greet(registries.nodes, identity, &hello).await {
|
||||
registries.nodes.session_abandoned(identity).await;
|
||||
return Err(error);
|
||||
}
|
||||
session.node = Some(identity);
|
||||
session.maximum = usize::try_from(hello.max_parallel)
|
||||
.map_err(|_| SessionError::InvalidParallelism(hello.max_parallel))?;
|
||||
if session.maximum == 0 {
|
||||
return Err(SessionError::InvalidParallelism(hello.max_parallel));
|
||||
}
|
||||
session.available = session
|
||||
.maximum
|
||||
.saturating_sub(control.leases_held_by(identity).await);
|
||||
session.work_version = control.work_version();
|
||||
let welcome = control_message::Body::Welcome(Welcome {
|
||||
node: identity.to_string(),
|
||||
});
|
||||
send(sender, outbound.next(welcome)).await?;
|
||||
let mut replayed_rotation = false;
|
||||
for pending in registries.nodes.outbound_pending(identity).await? {
|
||||
let stored = ControlMessage::decode(pending.payload.as_slice())
|
||||
.map_err(SessionError::Outbox)?;
|
||||
if !assignment_is_current(control, identity, &stored).await? {
|
||||
registries
|
||||
.nodes
|
||||
.outbound_acknowledged(identity, &pending.message_id)
|
||||
.await?;
|
||||
continue;
|
||||
}
|
||||
replayed_rotation |= matches!(stored.body, Some(control_message::Body::Rotated(_)));
|
||||
let replayed = outbound.resequence(stored);
|
||||
send(sender, replayed.clone()).await?;
|
||||
session.track(replayed);
|
||||
}
|
||||
// A node reconnecting late in the life of its credential is handed
|
||||
// the next one before it is handed work, so a long job never
|
||||
// outlives the credential that has to report it.
|
||||
if !replayed_rotation
|
||||
&& let Some(rotation) =
|
||||
renew_credential(registries.nodes, outbound, identity, sender).await?
|
||||
{
|
||||
session.track(rotation);
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await
|
||||
}
|
||||
Some(node_message::Body::Progress(progress)) => {
|
||||
let identity = holder(session.node)?;
|
||||
if report(control, identity, progress).await? {
|
||||
session.release();
|
||||
offer_work(registries, outbound, session, sender).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Some(node_message::Body::Heartbeat(heartbeat)) => {
|
||||
let identity = holder(session.node)?;
|
||||
beat(registries, identity, &heartbeat).await?;
|
||||
if let Some(rotation) =
|
||||
renew_credential(registries.nodes, outbound, identity, sender).await?
|
||||
{
|
||||
session.track(rotation);
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await
|
||||
}
|
||||
Some(node_message::Body::RotationAccepted(accepted)) => {
|
||||
let identity = holder(session.node)?;
|
||||
registries
|
||||
.nodes
|
||||
.accept_rotation(identity, &accepted.credential)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
Some(node_message::Body::Logs(logs)) => {
|
||||
let identity = holder(session.node)?;
|
||||
accept_logs(control, identity, &logs).await
|
||||
}
|
||||
Some(node_message::Body::Acknowledgement(acknowledgement)) => {
|
||||
let identity = holder(session.node)?;
|
||||
registries
|
||||
.nodes
|
||||
.outbound_acknowledged(identity, &acknowledgement.message_id)
|
||||
.await?;
|
||||
session.acknowledged(&acknowledgement.message_id);
|
||||
Ok(())
|
||||
}
|
||||
None => Err(SessionError::EmptyMessage),
|
||||
};
|
||||
let obsolete = result.as_ref().err().is_some_and(obsolete_report);
|
||||
if let Some(node) = node {
|
||||
registries
|
||||
.nodes
|
||||
.message_finished(node, &message.message_id, result.is_ok() || obsolete)
|
||||
.await?;
|
||||
}
|
||||
if obsolete {
|
||||
if let Some(mut acknowledgement) = acknowledgement {
|
||||
if progress {
|
||||
acknowledgement.conclusion = crate::wire::Conclusion::Cancelled as i32;
|
||||
}
|
||||
send(
|
||||
sender,
|
||||
outbound.next(control_message::Body::Acknowledgement(acknowledgement)),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
if result.is_ok()
|
||||
&& let Some(acknowledgement) = acknowledgement
|
||||
{
|
||||
send(
|
||||
sender,
|
||||
outbound.next(control_message::Body::Acknowledgement(acknowledgement)),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
result
|
||||
}
|
||||
mod acknowledgement;
|
||||
mod dispatch;
|
||||
mod obsolete;
|
||||
mod server;
|
||||
mod state;
|
||||
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use prost::Message;
|
||||
use syncode_control_nodes::{InboxDecision, NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::Status;
|
||||
use url::Url;
|
||||
|
||||
use crate::error::SessionError;
|
||||
use crate::identity::{enrol_node, greet, renew_credential};
|
||||
use crate::outbound::{Outbound, send};
|
||||
use crate::reports::{accept_logs, report};
|
||||
use crate::wire::{ControlMessage, NodeMessage, Welcome, control_message, node_message};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority, RepositoryCoordinates};
|
||||
use acknowledgement::{acknowledge_node_message, acknowledgement};
|
||||
use dispatch::{assignment_is_current, beat, holder, offer_work};
|
||||
use obsolete::obsolete_report;
|
||||
use state::Session;
|
||||
|
||||
const CHANNEL_DEPTH: usize = 32;
|
||||
const OUTBOUND_RETRY_INTERVAL: Duration = Duration::from_secs(2);
|
||||
|
||||
pub struct NodeSessionServer<L, S> {
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: Url,
|
||||
repositories: RepositoryCoordinates,
|
||||
assignments: bool,
|
||||
}
|
||||
|
||||
struct Registries<'a, L, S> {
|
||||
runs: &'a Runs<L>,
|
||||
nodes: &'a Nodes<S>,
|
||||
authority: &'a CapabilityAuthority,
|
||||
artifact_authority: &'a ArtifactTokenAuthority,
|
||||
action_artifact_url: &'a Url,
|
||||
repositories: &'a RepositoryCoordinates,
|
||||
assignments: bool,
|
||||
}
|
||||
|
||||
impl<L, S> Clone for Registries<'_, L, S> {
|
||||
fn clone(&self) -> Self {
|
||||
*self
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, S> Copy for Registries<'_, L, S> {}
|
||||
|
||||
async fn serve<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
message: NodeMessage,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
let control = registries.runs;
|
||||
if !session.accept(&message)? {
|
||||
acknowledge_node_message(outbound, &message, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
let tracked = matches!(
|
||||
message.body.as_ref(),
|
||||
Some(
|
||||
node_message::Body::Progress(_)
|
||||
| node_message::Body::Heartbeat(_)
|
||||
| node_message::Body::RotationAccepted(_)
|
||||
| node_message::Body::Logs(_)
|
||||
)
|
||||
);
|
||||
let node = tracked.then(|| holder(session.node)).transpose()?;
|
||||
if let Some(node) = node {
|
||||
let payload = NodeMessage {
|
||||
sequence: 0,
|
||||
message_id: String::new(),
|
||||
idempotency_key: String::new(),
|
||||
body: message.body.clone(),
|
||||
}
|
||||
.encode_to_vec();
|
||||
match registries
|
||||
.nodes
|
||||
.message_started(
|
||||
node,
|
||||
&message.message_id,
|
||||
&message.idempotency_key,
|
||||
&payload,
|
||||
)
|
||||
.await?
|
||||
{
|
||||
InboxDecision::New => {}
|
||||
InboxDecision::Duplicate => {
|
||||
acknowledge_node_message(outbound, &message, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
InboxDecision::MessageIdConflict => {
|
||||
return Err(SessionError::MessageIdConflict(message.message_id));
|
||||
}
|
||||
InboxDecision::IdempotencyConflict => {
|
||||
return Err(SessionError::IdempotencyConflict(message.idempotency_key));
|
||||
}
|
||||
}
|
||||
}
|
||||
let acknowledgement = acknowledgement(&message);
|
||||
let progress = matches!(message.body.as_ref(), Some(node_message::Body::Progress(_)));
|
||||
let result = match message.body {
|
||||
Some(node_message::Body::Enrol(enrol)) => {
|
||||
enrol_node(registries.nodes, outbound, &enrol, sender).await
|
||||
}
|
||||
Some(node_message::Body::Hello(hello)) => {
|
||||
let identity: syncode_control_runs::NodeId = hello.node.parse()?;
|
||||
registries
|
||||
.nodes
|
||||
.authenticate(identity, &hello.credential, SystemTime::now())
|
||||
.await?;
|
||||
if !registries.nodes.session_opened(identity).await {
|
||||
return Err(SessionError::ConcurrentSession(identity));
|
||||
}
|
||||
if let Err(error) = greet(registries.nodes, identity, &hello).await {
|
||||
registries.nodes.session_abandoned(identity).await;
|
||||
return Err(error);
|
||||
}
|
||||
session.node = Some(identity);
|
||||
session.maximum = usize::try_from(hello.max_parallel)
|
||||
.map_err(|_| SessionError::InvalidParallelism(hello.max_parallel))?;
|
||||
if session.maximum == 0 {
|
||||
return Err(SessionError::InvalidParallelism(hello.max_parallel));
|
||||
}
|
||||
session.available = session
|
||||
.maximum
|
||||
.saturating_sub(control.leases_held_by(identity).await);
|
||||
session.work_version = control.work_version();
|
||||
let welcome = control_message::Body::Welcome(Welcome {
|
||||
node: identity.to_string(),
|
||||
});
|
||||
send(sender, outbound.next(welcome)).await?;
|
||||
let mut replayed_rotation = false;
|
||||
for pending in registries.nodes.outbound_pending(identity).await? {
|
||||
let stored = ControlMessage::decode(pending.payload.as_slice())
|
||||
.map_err(SessionError::Outbox)?;
|
||||
if !assignment_is_current(control, identity, &stored).await? {
|
||||
registries
|
||||
.nodes
|
||||
.outbound_acknowledged(identity, &pending.message_id)
|
||||
.await?;
|
||||
continue;
|
||||
}
|
||||
replayed_rotation |= matches!(stored.body, Some(control_message::Body::Rotated(_)));
|
||||
let replayed = outbound.resequence(stored);
|
||||
send(sender, replayed.clone()).await?;
|
||||
session.track(replayed);
|
||||
}
|
||||
// A node reconnecting late in the life of its credential is handed
|
||||
// the next one before it is handed work, so a long job never
|
||||
// outlives the credential that has to report it.
|
||||
if !replayed_rotation
|
||||
&& let Some(rotation) =
|
||||
renew_credential(registries.nodes, outbound, identity, sender).await?
|
||||
{
|
||||
session.track(rotation);
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await
|
||||
}
|
||||
Some(node_message::Body::Progress(progress)) => {
|
||||
let identity = holder(session.node)?;
|
||||
if report(control, identity, progress).await? {
|
||||
session.release();
|
||||
offer_work(registries, outbound, session, sender).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
Some(node_message::Body::Heartbeat(heartbeat)) => {
|
||||
let identity = holder(session.node)?;
|
||||
beat(registries, identity, &heartbeat).await?;
|
||||
if let Some(rotation) =
|
||||
renew_credential(registries.nodes, outbound, identity, sender).await?
|
||||
{
|
||||
session.track(rotation);
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await
|
||||
}
|
||||
Some(node_message::Body::RotationAccepted(accepted)) => {
|
||||
let identity = holder(session.node)?;
|
||||
registries
|
||||
.nodes
|
||||
.accept_rotation(identity, &accepted.credential)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
Some(node_message::Body::Logs(logs)) => {
|
||||
let identity = holder(session.node)?;
|
||||
accept_logs(control, identity, &logs).await
|
||||
}
|
||||
Some(node_message::Body::Acknowledgement(acknowledgement)) => {
|
||||
let identity = holder(session.node)?;
|
||||
registries
|
||||
.nodes
|
||||
.outbound_acknowledged(identity, &acknowledgement.message_id)
|
||||
.await?;
|
||||
session.acknowledged(&acknowledgement.message_id);
|
||||
Ok(())
|
||||
}
|
||||
None => Err(SessionError::EmptyMessage),
|
||||
};
|
||||
let obsolete = result.as_ref().err().is_some_and(obsolete_report);
|
||||
if let Some(node) = node {
|
||||
registries
|
||||
.nodes
|
||||
.message_finished(node, &message.message_id, result.is_ok() || obsolete)
|
||||
.await?;
|
||||
}
|
||||
if obsolete {
|
||||
if let Some(mut acknowledgement) = acknowledgement {
|
||||
if progress {
|
||||
acknowledgement.conclusion = crate::wire::Conclusion::Cancelled as i32;
|
||||
}
|
||||
send(
|
||||
sender,
|
||||
outbound.next(control_message::Body::Acknowledgement(acknowledgement)),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
offer_work(registries, outbound, session, sender).await?;
|
||||
return Ok(());
|
||||
}
|
||||
if result.is_ok()
|
||||
&& let Some(acknowledgement) = acknowledgement
|
||||
{
|
||||
send(
|
||||
sender,
|
||||
outbound.next(control_message::Body::Acknowledgement(acknowledgement)),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
result
|
||||
}
|
||||
@@ -1,211 +1,196 @@
|
||||
mod assignment;
|
||||
|
||||
use std::time::SystemTime;
|
||||
|
||||
use prost::Message;
|
||||
use syncode_control_nodes::{NodeStore, NodesError, Refusal};
|
||||
use syncode_control_runs::{Dispatch, Fence, NodeId, RunId, RunLog, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::Status;
|
||||
|
||||
use super::Registries;
|
||||
use super::state::Session;
|
||||
use crate::declaration;
|
||||
use crate::error::SessionError;
|
||||
use crate::outbound::{DurableDelivery, Outbound, send, send_durable};
|
||||
use crate::projection::ProjectionClient;
|
||||
use crate::wire::{ControlMessage, Heartbeat, Refused, control_message};
|
||||
use assignment::assignment_body;
|
||||
|
||||
pub(super) async fn offer_work<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
if !registries.assignments {
|
||||
return Ok(());
|
||||
}
|
||||
let node = holder(session.node)?;
|
||||
offer_cancellations(registries, outbound, session, sender, node).await?;
|
||||
session.available = session
|
||||
.maximum
|
||||
.saturating_sub(registries.runs.leases_held_by(node).await);
|
||||
while session.available > 0 {
|
||||
registries
|
||||
.nodes
|
||||
.session_available(node, session.available)
|
||||
.await;
|
||||
let candidate = match registries.nodes.scheduling_candidate(node).await {
|
||||
Ok(candidate) => candidate,
|
||||
Err(NodesError::Refused(Refusal::Undeclared(_))) => {
|
||||
session.available = 0;
|
||||
registries.nodes.session_available(node, 0).await;
|
||||
return Ok(());
|
||||
}
|
||||
Err(NodesError::Refused(refusal)) => {
|
||||
let body = control_message::Body::Refused(Refused {
|
||||
reason: refusal.to_string(),
|
||||
});
|
||||
return send(sender, outbound.next(body)).await;
|
||||
}
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
let candidates = registries.nodes.available_candidates().await;
|
||||
let assignment = match registries
|
||||
.runs
|
||||
.take_next_among(&candidate, &candidates)
|
||||
.await?
|
||||
{
|
||||
Dispatch::Assigned(assignment) => assignment,
|
||||
Dispatch::Empty => break,
|
||||
Dispatch::Refused(_) => break,
|
||||
};
|
||||
retire_previous_assignments(registries.nodes, node, session, &assignment).await?;
|
||||
project_before_dispatch(registries.runs, registries.projection, assignment.run()).await?;
|
||||
let repository = registries
|
||||
.projection
|
||||
.repository_coordinates(assignment.origin().repository())
|
||||
.await?;
|
||||
let body = assignment_body(
|
||||
&assignment,
|
||||
repository,
|
||||
node,
|
||||
registries.authority,
|
||||
registries.artifact_authority,
|
||||
registries.action_artifact_url,
|
||||
)?;
|
||||
match send_durable(registries.nodes, node, sender, outbound, body).await? {
|
||||
DurableDelivery::Pending(sent) => session.track(*sent),
|
||||
DurableDelivery::Acknowledged => {}
|
||||
}
|
||||
session.available -= 1;
|
||||
}
|
||||
registries
|
||||
.nodes
|
||||
.session_available(node, session.available)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn retire_previous_assignments<S: NodeStore>(
|
||||
nodes: &syncode_control_nodes::Nodes<S>,
|
||||
node: NodeId,
|
||||
session: &mut Session,
|
||||
assignment: &syncode_control_runs::Assignment,
|
||||
) -> Result<(), SessionError> {
|
||||
for pending in nodes.outbound_pending(node).await? {
|
||||
let stored =
|
||||
ControlMessage::decode(pending.payload.as_slice()).map_err(SessionError::Outbox)?;
|
||||
let same_job = matches!(
|
||||
stored.body,
|
||||
Some(control_message::Body::Assignment(ref previous))
|
||||
if previous.run == assignment.run().to_string()
|
||||
&& previous.job == assignment.job().to_string()
|
||||
&& previous.fence != assignment.fence().get()
|
||||
);
|
||||
if same_job {
|
||||
nodes
|
||||
.outbound_acknowledged(node, &pending.message_id)
|
||||
.await?;
|
||||
session.acknowledged(&pending.message_id);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn assignment_is_current<L: RunLog>(
|
||||
runs: &syncode_control_runs::Runs<L>,
|
||||
node: NodeId,
|
||||
message: &ControlMessage,
|
||||
) -> Result<bool, SessionError> {
|
||||
let Some(control_message::Body::Assignment(assignment)) = message.body.as_ref() else {
|
||||
return Ok(true);
|
||||
};
|
||||
Ok(runs
|
||||
.assignment_is_current(
|
||||
assignment.run.parse()?,
|
||||
assignment.job.parse()?,
|
||||
node,
|
||||
Fence::from(assignment.fence),
|
||||
SystemTime::now(),
|
||||
)
|
||||
.await)
|
||||
}
|
||||
|
||||
async fn offer_cancellations<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
node: NodeId,
|
||||
) -> Result<(), SessionError> {
|
||||
for cancellation in registries.runs.pending_cancellations(node).await {
|
||||
let body = control_message::Body::Cancel(crate::wire::Cancel {
|
||||
run: cancellation.run().to_string(),
|
||||
fence: cancellation.fence().get(),
|
||||
job: cancellation.job().to_string(),
|
||||
});
|
||||
if let DurableDelivery::Pending(sent) =
|
||||
send_durable(registries.nodes, node, sender, outbound, body).await?
|
||||
{
|
||||
session.track(*sent);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn project_before_dispatch<L: RunLog>(
|
||||
runs: &Runs<L>,
|
||||
projection: &ProjectionClient,
|
||||
run: RunId,
|
||||
) -> Result<(), SessionError> {
|
||||
let projected = runs
|
||||
.projection(run)
|
||||
.await?
|
||||
.ok_or(SessionError::MissingProjection(run))?;
|
||||
projection.send(&projected).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn retry_pending(
|
||||
session: &Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
for message in session.pending() {
|
||||
send(sender, message.clone()).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn holder(node: Option<NodeId>) -> Result<NodeId, SessionError> {
|
||||
node.ok_or(SessionError::NoHello)
|
||||
}
|
||||
|
||||
pub(super) async fn beat<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
node: NodeId,
|
||||
heartbeat: &Heartbeat,
|
||||
) -> Result<(), SessionError> {
|
||||
let capacity = heartbeat
|
||||
.capacity
|
||||
.as_ref()
|
||||
.map(declaration::capacity)
|
||||
.transpose()?;
|
||||
registries
|
||||
.nodes
|
||||
.beat(node, capacity, SystemTime::now())
|
||||
.await?;
|
||||
for held in &heartbeat.held {
|
||||
registries
|
||||
.runs
|
||||
.renewed_job(
|
||||
held.run.parse()?,
|
||||
held.job.parse()?,
|
||||
node,
|
||||
Fence::from(held.fence),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
mod assignment;
|
||||
|
||||
use std::time::SystemTime;
|
||||
|
||||
use prost::Message;
|
||||
use syncode_control_nodes::{NodeStore, NodesError, Refusal};
|
||||
use syncode_control_runs::{Dispatch, Fence, NodeId, RunLog};
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::Status;
|
||||
|
||||
use super::Registries;
|
||||
use super::state::Session;
|
||||
use crate::declaration;
|
||||
use crate::error::SessionError;
|
||||
use crate::outbound::{DurableDelivery, Outbound, send, send_durable};
|
||||
use crate::wire::{ControlMessage, Heartbeat, Refused, control_message};
|
||||
use assignment::assignment_body;
|
||||
|
||||
pub(super) async fn offer_work<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
if !registries.assignments {
|
||||
return Ok(());
|
||||
}
|
||||
let node = holder(session.node)?;
|
||||
offer_cancellations(registries, outbound, session, sender, node).await?;
|
||||
session.available = session
|
||||
.maximum
|
||||
.saturating_sub(registries.runs.leases_held_by(node).await);
|
||||
while session.available > 0 {
|
||||
registries
|
||||
.nodes
|
||||
.session_available(node, session.available)
|
||||
.await;
|
||||
let candidate = match registries.nodes.scheduling_candidate(node).await {
|
||||
Ok(candidate) => candidate,
|
||||
Err(NodesError::Refused(Refusal::Undeclared(_))) => {
|
||||
session.available = 0;
|
||||
registries.nodes.session_available(node, 0).await;
|
||||
return Ok(());
|
||||
}
|
||||
Err(NodesError::Refused(refusal)) => {
|
||||
let body = control_message::Body::Refused(Refused {
|
||||
reason: refusal.to_string(),
|
||||
});
|
||||
return send(sender, outbound.next(body)).await;
|
||||
}
|
||||
Err(error) => return Err(error.into()),
|
||||
};
|
||||
let candidates = registries.nodes.available_candidates().await;
|
||||
let assignment = match registries
|
||||
.runs
|
||||
.take_next_among(&candidate, &candidates)
|
||||
.await?
|
||||
{
|
||||
Dispatch::Assigned(assignment) => assignment,
|
||||
Dispatch::Empty => break,
|
||||
Dispatch::Refused(_) => break,
|
||||
};
|
||||
retire_previous_assignments(registries.nodes, node, session, &assignment).await?;
|
||||
let repository = registries
|
||||
.repositories
|
||||
.resolve(assignment.origin().repository())
|
||||
.await?;
|
||||
let body = assignment_body(
|
||||
&assignment,
|
||||
repository,
|
||||
node,
|
||||
registries.authority,
|
||||
registries.artifact_authority,
|
||||
registries.action_artifact_url,
|
||||
)?;
|
||||
match send_durable(registries.nodes, node, sender, outbound, body).await? {
|
||||
DurableDelivery::Pending(sent) => session.track(*sent),
|
||||
DurableDelivery::Acknowledged => {}
|
||||
}
|
||||
session.available -= 1;
|
||||
}
|
||||
registries
|
||||
.nodes
|
||||
.session_available(node, session.available)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn retire_previous_assignments<S: NodeStore>(
|
||||
nodes: &syncode_control_nodes::Nodes<S>,
|
||||
node: NodeId,
|
||||
session: &mut Session,
|
||||
assignment: &syncode_control_runs::Assignment,
|
||||
) -> Result<(), SessionError> {
|
||||
for pending in nodes.outbound_pending(node).await? {
|
||||
let stored =
|
||||
ControlMessage::decode(pending.payload.as_slice()).map_err(SessionError::Outbox)?;
|
||||
let same_job = matches!(
|
||||
stored.body,
|
||||
Some(control_message::Body::Assignment(ref previous))
|
||||
if previous.run == assignment.run().to_string()
|
||||
&& previous.job == assignment.job().to_string()
|
||||
&& previous.fence != assignment.fence().get()
|
||||
);
|
||||
if same_job {
|
||||
nodes
|
||||
.outbound_acknowledged(node, &pending.message_id)
|
||||
.await?;
|
||||
session.acknowledged(&pending.message_id);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn assignment_is_current<L: RunLog>(
|
||||
runs: &syncode_control_runs::Runs<L>,
|
||||
node: NodeId,
|
||||
message: &ControlMessage,
|
||||
) -> Result<bool, SessionError> {
|
||||
let Some(control_message::Body::Assignment(assignment)) = message.body.as_ref() else {
|
||||
return Ok(true);
|
||||
};
|
||||
Ok(runs
|
||||
.assignment_is_current(
|
||||
assignment.run.parse()?,
|
||||
assignment.job.parse()?,
|
||||
node,
|
||||
Fence::from(assignment.fence),
|
||||
SystemTime::now(),
|
||||
)
|
||||
.await)
|
||||
}
|
||||
|
||||
async fn offer_cancellations<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
outbound: &mut Outbound,
|
||||
session: &mut Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
node: NodeId,
|
||||
) -> Result<(), SessionError> {
|
||||
for cancellation in registries.runs.pending_cancellations(node).await {
|
||||
let body = control_message::Body::Cancel(crate::wire::Cancel {
|
||||
run: cancellation.run().to_string(),
|
||||
fence: cancellation.fence().get(),
|
||||
job: cancellation.job().to_string(),
|
||||
});
|
||||
if let DurableDelivery::Pending(sent) =
|
||||
send_durable(registries.nodes, node, sender, outbound, body).await?
|
||||
{
|
||||
session.track(*sent);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) async fn retry_pending(
|
||||
session: &Session,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
) -> Result<(), SessionError> {
|
||||
for message in session.pending() {
|
||||
send(sender, message.clone()).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(super) fn holder(node: Option<NodeId>) -> Result<NodeId, SessionError> {
|
||||
node.ok_or(SessionError::NoHello)
|
||||
}
|
||||
|
||||
pub(super) async fn beat<L: RunLog, S: NodeStore>(
|
||||
registries: Registries<'_, L, S>,
|
||||
node: NodeId,
|
||||
heartbeat: &Heartbeat,
|
||||
) -> Result<(), SessionError> {
|
||||
let capacity = heartbeat
|
||||
.capacity
|
||||
.as_ref()
|
||||
.map(declaration::capacity)
|
||||
.transpose()?;
|
||||
registries
|
||||
.nodes
|
||||
.beat(node, capacity, SystemTime::now())
|
||||
.await?;
|
||||
for held in &heartbeat.held {
|
||||
registries
|
||||
.runs
|
||||
.renewed_job(
|
||||
held.run.parse()?,
|
||||
held.job.parse()?,
|
||||
node,
|
||||
Fence::from(held.fence),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,154 +1,154 @@
|
||||
use std::pin::Pin;
|
||||
|
||||
use syncode_control_nodes::{NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
use tokio_stream::{Stream, StreamExt};
|
||||
use tonic::{Request, Response, Status, Streaming};
|
||||
|
||||
use super::dispatch::{offer_work, retry_pending};
|
||||
use super::state::Session;
|
||||
use super::{CHANNEL_DEPTH, NodeSessionServer, OUTBOUND_RETRY_INTERVAL, Registries, serve};
|
||||
use crate::error::SessionError;
|
||||
use crate::outbound::Outbound;
|
||||
use crate::wire::node_session_server::NodeSession;
|
||||
use crate::wire::{ControlMessage, NodeMessage};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority};
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
// A derive would add `L: Clone, S: Clone` bounds neither `Runs<L>` nor
|
||||
// `Nodes<S>` need — both clone by sharing their own internal handle.
|
||||
impl<L, S> Clone for NodeSessionServer<L, S> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
control: self.control.clone(),
|
||||
nodes: self.nodes.clone(),
|
||||
authority: self.authority.clone(),
|
||||
artifact_authority: self.artifact_authority.clone(),
|
||||
action_artifact_url: self.action_artifact_url.clone(),
|
||||
projection: self.projection.clone(),
|
||||
assignments: self.assignments,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, S> NodeSessionServer<L, S> {
|
||||
#[must_use]
|
||||
pub const fn new(
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: url::Url,
|
||||
projection: crate::ProjectionClient,
|
||||
) -> Self {
|
||||
Self {
|
||||
control,
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
action_artifact_url,
|
||||
projection,
|
||||
assignments: true,
|
||||
}
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn shadow(
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: url::Url,
|
||||
projection: crate::ProjectionClient,
|
||||
) -> Self {
|
||||
Self {
|
||||
control,
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
action_artifact_url,
|
||||
projection,
|
||||
assignments: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl<L: RunLog + 'static, S: NodeStore + 'static> NodeSession for NodeSessionServer<L, S> {
|
||||
type OpenStream = Pin<Box<dyn Stream<Item = Result<ControlMessage, Status>> + Send>>;
|
||||
|
||||
async fn open(
|
||||
&self,
|
||||
request: Request<Streaming<NodeMessage>>,
|
||||
) -> Result<Response<Self::OpenStream>, Status> {
|
||||
let mut inbound = request.into_inner();
|
||||
let (sender, receiver) = mpsc::channel(CHANNEL_DEPTH);
|
||||
let server = self.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(error) = run_session(&mut inbound, &sender, server).await {
|
||||
eprintln!("syncode control: node session failed: {error}");
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Response::new(Box::pin(ReceiverStream::new(receiver))))
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_session<L: RunLog, S: NodeStore, I>(
|
||||
inbound: &mut I,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
server: NodeSessionServer<L, S>,
|
||||
) -> Result<(), SessionError>
|
||||
where
|
||||
I: Stream<Item = Result<NodeMessage, Status>> + Unpin,
|
||||
{
|
||||
let mut outbound = Outbound::default();
|
||||
let mut session = Session::default();
|
||||
let registries = Registries {
|
||||
runs: &server.control,
|
||||
nodes: &server.nodes,
|
||||
authority: &server.authority,
|
||||
artifact_authority: &server.artifact_authority,
|
||||
action_artifact_url: &server.action_artifact_url,
|
||||
projection: &server.projection,
|
||||
assignments: server.assignments,
|
||||
};
|
||||
let mut retry = tokio::time::interval(OUTBOUND_RETRY_INTERVAL);
|
||||
retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
retry.tick().await;
|
||||
loop {
|
||||
let work_version = session.work_version;
|
||||
let served = tokio::select! {
|
||||
message = inbound.next() => match message {
|
||||
Some(Ok(message)) => serve(
|
||||
registries, &mut outbound, &mut session, message, sender,
|
||||
).await,
|
||||
Some(Err(error)) => Err(SessionError::Transport(error)),
|
||||
None => break,
|
||||
},
|
||||
version = server.control.wait_for_work(work_version), if session.is_open() => {
|
||||
session.work_version = version;
|
||||
offer_work(registries, &mut outbound, &mut session, sender).await
|
||||
}
|
||||
_ = retry.tick(), if session.has_pending() => retry_pending(&session, sender).await
|
||||
};
|
||||
if let Err(error) = served {
|
||||
let _ = sender.send(Err(error.into())).await;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if let Some(id) = session.node
|
||||
&& let Err(error) = server.nodes.session_closed(id).await
|
||||
{
|
||||
sender
|
||||
.send(Err(SessionError::from(error).into()))
|
||||
.await
|
||||
.map_err(|_| SessionError::Unreadable)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
use std::pin::Pin;
|
||||
|
||||
use syncode_control_nodes::{NodeStore, Nodes};
|
||||
use syncode_control_runs::{RunLog, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
use tokio_stream::{Stream, StreamExt};
|
||||
use tonic::{Request, Response, Status, Streaming};
|
||||
|
||||
use super::dispatch::{offer_work, retry_pending};
|
||||
use super::state::Session;
|
||||
use super::{CHANNEL_DEPTH, NodeSessionServer, OUTBOUND_RETRY_INTERVAL, Registries, serve};
|
||||
use crate::error::SessionError;
|
||||
use crate::outbound::Outbound;
|
||||
use crate::wire::node_session_server::NodeSession;
|
||||
use crate::wire::{ControlMessage, NodeMessage};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority};
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
// A derive would add `L: Clone, S: Clone` bounds neither `Runs<L>` nor
|
||||
// `Nodes<S>` need — both clone by sharing their own internal handle.
|
||||
impl<L, S> Clone for NodeSessionServer<L, S> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
control: self.control.clone(),
|
||||
nodes: self.nodes.clone(),
|
||||
authority: self.authority.clone(),
|
||||
artifact_authority: self.artifact_authority.clone(),
|
||||
action_artifact_url: self.action_artifact_url.clone(),
|
||||
repositories: self.repositories.clone(),
|
||||
assignments: self.assignments,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<L, S> NodeSessionServer<L, S> {
|
||||
#[must_use]
|
||||
pub const fn new(
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: url::Url,
|
||||
repositories: crate::RepositoryCoordinates,
|
||||
) -> Self {
|
||||
Self {
|
||||
control,
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
action_artifact_url,
|
||||
repositories,
|
||||
assignments: true,
|
||||
}
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub const fn shadow(
|
||||
control: Runs<L>,
|
||||
nodes: Nodes<S>,
|
||||
authority: CapabilityAuthority,
|
||||
artifact_authority: ArtifactTokenAuthority,
|
||||
action_artifact_url: url::Url,
|
||||
repositories: crate::RepositoryCoordinates,
|
||||
) -> Self {
|
||||
Self {
|
||||
control,
|
||||
nodes,
|
||||
authority,
|
||||
artifact_authority,
|
||||
action_artifact_url,
|
||||
repositories,
|
||||
assignments: false,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl<L: RunLog + 'static, S: NodeStore + 'static> NodeSession for NodeSessionServer<L, S> {
|
||||
type OpenStream = Pin<Box<dyn Stream<Item = Result<ControlMessage, Status>> + Send>>;
|
||||
|
||||
async fn open(
|
||||
&self,
|
||||
request: Request<Streaming<NodeMessage>>,
|
||||
) -> Result<Response<Self::OpenStream>, Status> {
|
||||
let mut inbound = request.into_inner();
|
||||
let (sender, receiver) = mpsc::channel(CHANNEL_DEPTH);
|
||||
let server = self.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(error) = run_session(&mut inbound, &sender, server).await {
|
||||
eprintln!("syncode control: node session failed: {error}");
|
||||
}
|
||||
});
|
||||
|
||||
Ok(Response::new(Box::pin(ReceiverStream::new(receiver))))
|
||||
}
|
||||
}
|
||||
|
||||
async fn run_session<L: RunLog, S: NodeStore, I>(
|
||||
inbound: &mut I,
|
||||
sender: &mpsc::Sender<Result<ControlMessage, Status>>,
|
||||
server: NodeSessionServer<L, S>,
|
||||
) -> Result<(), SessionError>
|
||||
where
|
||||
I: Stream<Item = Result<NodeMessage, Status>> + Unpin,
|
||||
{
|
||||
let mut outbound = Outbound::default();
|
||||
let mut session = Session::default();
|
||||
let registries = Registries {
|
||||
runs: &server.control,
|
||||
nodes: &server.nodes,
|
||||
authority: &server.authority,
|
||||
artifact_authority: &server.artifact_authority,
|
||||
action_artifact_url: &server.action_artifact_url,
|
||||
repositories: &server.repositories,
|
||||
assignments: server.assignments,
|
||||
};
|
||||
let mut retry = tokio::time::interval(OUTBOUND_RETRY_INTERVAL);
|
||||
retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
|
||||
retry.tick().await;
|
||||
loop {
|
||||
let work_version = session.work_version;
|
||||
let served = tokio::select! {
|
||||
message = inbound.next() => match message {
|
||||
Some(Ok(message)) => serve(
|
||||
registries, &mut outbound, &mut session, message, sender,
|
||||
).await,
|
||||
Some(Err(error)) => Err(SessionError::Transport(error)),
|
||||
None => break,
|
||||
},
|
||||
version = server.control.wait_for_work(work_version), if session.is_open() => {
|
||||
session.work_version = version;
|
||||
offer_work(registries, &mut outbound, &mut session, sender).await
|
||||
}
|
||||
_ = retry.tick(), if session.has_pending() => retry_pending(&session, sender).await
|
||||
};
|
||||
if let Err(error) = served {
|
||||
let _ = sender.send(Err(error.into())).await;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if let Some(id) = session.node
|
||||
&& let Err(error) = server.nodes.session_closed(id).await
|
||||
{
|
||||
sender
|
||||
.send(Err(SessionError::from(error).into()))
|
||||
.await
|
||||
.map_err(|_| SessionError::Unreadable)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,126 +1,125 @@
|
||||
use std::time::SystemTime;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{
|
||||
Architecture, JobId, MatrixPolicy, NodeId, OperatingSystem, Origin, Priority, Requirements,
|
||||
RunEvent, SchedulingRefusal,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum AuditPayload {
|
||||
Origin(Origin),
|
||||
Jobs(AuditJobs),
|
||||
Enrolment(AuditEnrolment),
|
||||
Node(AuditNodeSnapshot),
|
||||
RunEvent(RunEvent),
|
||||
Configuration(AuditConfiguration),
|
||||
RunAdmission(AuditRunAdmission),
|
||||
SchedulingRefusals(Vec<SchedulingRefusal>),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
pub struct AuditJobs {
|
||||
pub jobs: Vec<AuditJob>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditJob {
|
||||
pub id: JobId,
|
||||
pub key: String,
|
||||
pub needs: Vec<String>,
|
||||
pub matrix: MatrixPolicy,
|
||||
pub priority: Priority,
|
||||
pub requirements: Requirements,
|
||||
pub secrets: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditEnrolment {
|
||||
pub scope: AuditNodeScope,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditNodeScope {
|
||||
Instance,
|
||||
Organisation(String),
|
||||
Repository(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditNodeLifecycle {
|
||||
Enrolled,
|
||||
Active,
|
||||
Draining,
|
||||
Offline,
|
||||
Revoked,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditContainerRuntime {
|
||||
pub name: String,
|
||||
pub version: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeCapabilities {
|
||||
pub architecture: Architecture,
|
||||
pub operating_system: OperatingSystem,
|
||||
pub container_runtime: AuditContainerRuntime,
|
||||
pub cores: u32,
|
||||
pub memory_bytes: u64,
|
||||
pub labels: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditVolume {
|
||||
pub total_bytes: u64,
|
||||
pub used_bytes: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeCapacity {
|
||||
pub build_volume_free_bytes: u64,
|
||||
pub cache_volume: Option<AuditVolume>,
|
||||
pub cache_volume_path: Option<String>,
|
||||
pub layer_store_bytes: u64,
|
||||
pub cached_images: Vec<String>,
|
||||
pub cached_actions: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeSnapshot {
|
||||
pub id: NodeId,
|
||||
pub scope: AuditNodeScope,
|
||||
pub lifecycle: AuditNodeLifecycle,
|
||||
pub capabilities: Option<AuditNodeCapabilities>,
|
||||
pub capacity: Option<AuditNodeCapacity>,
|
||||
pub last_seen: SystemTime,
|
||||
pub credential_expires_at: SystemTime,
|
||||
pub rotation_pending: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditControlMode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditConfiguration {
|
||||
pub mode: AuditControlMode,
|
||||
pub organization_concurrency: u32,
|
||||
pub repository_concurrency: u32,
|
||||
pub principal_concurrency: u32,
|
||||
pub organization_queue_quota: u32,
|
||||
pub principal_queue_quota: u32,
|
||||
pub audit_retention_days: u32,
|
||||
pub projection_min_run_number: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditRunAdmission {
|
||||
pub repository: String,
|
||||
pub jobs: usize,
|
||||
}
|
||||
use std::time::SystemTime;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{
|
||||
Architecture, JobId, MatrixPolicy, NodeId, OperatingSystem, Origin, Priority, Requirements,
|
||||
RunEvent, SchedulingRefusal,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
#[serde(untagged)]
|
||||
pub enum AuditPayload {
|
||||
Origin(Origin),
|
||||
Jobs(AuditJobs),
|
||||
Enrolment(AuditEnrolment),
|
||||
Node(AuditNodeSnapshot),
|
||||
RunEvent(RunEvent),
|
||||
Configuration(AuditConfiguration),
|
||||
RunAdmission(AuditRunAdmission),
|
||||
SchedulingRefusals(Vec<SchedulingRefusal>),
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
|
||||
pub struct AuditJobs {
|
||||
pub jobs: Vec<AuditJob>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditJob {
|
||||
pub id: JobId,
|
||||
pub key: String,
|
||||
pub needs: Vec<String>,
|
||||
pub matrix: MatrixPolicy,
|
||||
pub priority: Priority,
|
||||
pub requirements: Requirements,
|
||||
pub secrets: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditEnrolment {
|
||||
pub scope: AuditNodeScope,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditNodeScope {
|
||||
Instance,
|
||||
Organisation(String),
|
||||
Repository(String),
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditNodeLifecycle {
|
||||
Enrolled,
|
||||
Active,
|
||||
Draining,
|
||||
Offline,
|
||||
Revoked,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditContainerRuntime {
|
||||
pub name: String,
|
||||
pub version: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeCapabilities {
|
||||
pub architecture: Architecture,
|
||||
pub operating_system: OperatingSystem,
|
||||
pub container_runtime: AuditContainerRuntime,
|
||||
pub cores: u32,
|
||||
pub memory_bytes: u64,
|
||||
pub labels: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditVolume {
|
||||
pub total_bytes: u64,
|
||||
pub used_bytes: u64,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeCapacity {
|
||||
pub build_volume_free_bytes: u64,
|
||||
pub cache_volume: Option<AuditVolume>,
|
||||
pub cache_volume_path: Option<String>,
|
||||
pub layer_store_bytes: u64,
|
||||
pub cached_images: Vec<String>,
|
||||
pub cached_actions: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditNodeSnapshot {
|
||||
pub id: NodeId,
|
||||
pub scope: AuditNodeScope,
|
||||
pub lifecycle: AuditNodeLifecycle,
|
||||
pub capabilities: Option<AuditNodeCapabilities>,
|
||||
pub capacity: Option<AuditNodeCapacity>,
|
||||
pub last_seen: SystemTime,
|
||||
pub credential_expires_at: SystemTime,
|
||||
pub rotation_pending: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub enum AuditControlMode {
|
||||
Shadow,
|
||||
Active,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditConfiguration {
|
||||
pub mode: AuditControlMode,
|
||||
pub organization_concurrency: u32,
|
||||
pub repository_concurrency: u32,
|
||||
pub principal_concurrency: u32,
|
||||
pub organization_queue_quota: u32,
|
||||
pub principal_queue_quota: u32,
|
||||
pub audit_retention_days: u32,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||
pub struct AuditRunAdmission {
|
||||
pub repository: String,
|
||||
pub jobs: usize,
|
||||
}
|
||||
@@ -1,141 +1,136 @@
|
||||
#![allow(clippy::expect_used, clippy::panic)]
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use syncode_control_nodes::{Ephemeral, Lifecycle, Nodes, Scope};
|
||||
use syncode_control_runs::{Forgotten, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
|
||||
use super::{NodeSessionServer, run_session};
|
||||
use crate::wire::{
|
||||
Capabilities, Capacity, Enrol, Heartbeat, Hello, NodeMessage, control_message, node_message,
|
||||
};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority, ProjectionClient};
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_unreadable_client_releases_its_session() {
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
let nodes = Nodes::restored(Ephemeral::default()).await.expect("nodes");
|
||||
let token = nodes
|
||||
.issue_token(Scope::Instance)
|
||||
.await
|
||||
.expect("enrolment token");
|
||||
let server = NodeSessionServer::new(
|
||||
runs,
|
||||
nodes.clone(),
|
||||
CapabilityAuthority::new("test-capability-key").expect("capability authority"),
|
||||
ArtifactTokenAuthority::new("test-artifact-key").expect("artifact authority"),
|
||||
"http://127.0.0.1:1/".parse().expect("artifact URL"),
|
||||
ProjectionClient::new(
|
||||
"http://127.0.0.1:1/".parse().expect("projection URL"),
|
||||
String::new(),
|
||||
"http://127.0.0.1:1".to_owned(),
|
||||
String::new(),
|
||||
)
|
||||
.expect("projection client"),
|
||||
);
|
||||
let (input, input_receiver) = mpsc::channel(8);
|
||||
let (output, mut output_receiver) = mpsc::channel(8);
|
||||
let session = tokio::spawn(async move {
|
||||
let mut inbound = ReceiverStream::new(input_receiver);
|
||||
run_session(&mut inbound, &output, server).await
|
||||
});
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
1,
|
||||
node_message::Body::Enrol(Enrol {
|
||||
token: token.secret().expose().to_owned(),
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("enrol");
|
||||
let enrolled = output_receiver
|
||||
.recv()
|
||||
.await
|
||||
.expect("enrolment response")
|
||||
.expect("enrolment status");
|
||||
let control_message::Body::Enrolled(enrolled) = enrolled.body.expect("enrolment body") else {
|
||||
panic!("expected enrolment");
|
||||
};
|
||||
let node = enrolled.node.parse().expect("node id");
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
2,
|
||||
node_message::Body::Hello(Hello {
|
||||
node: enrolled.node,
|
||||
credential: enrolled.credential,
|
||||
capabilities: Some(capabilities()),
|
||||
capacity: Some(capacity()),
|
||||
max_parallel: 1,
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("hello");
|
||||
let welcome = output_receiver
|
||||
.recv()
|
||||
.await
|
||||
.expect("welcome response")
|
||||
.expect("welcome status");
|
||||
assert!(matches!(
|
||||
welcome.body,
|
||||
Some(control_message::Body::Welcome(_))
|
||||
));
|
||||
drop(output_receiver);
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
3,
|
||||
node_message::Body::Heartbeat(Heartbeat {
|
||||
held: Vec::new(),
|
||||
capacity: Some(capacity()),
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("heartbeat");
|
||||
tokio::time::timeout(Duration::from_secs(1), session)
|
||||
.await
|
||||
.expect("session shutdown")
|
||||
.expect("session task")
|
||||
.expect("session cleanup");
|
||||
assert_eq!(
|
||||
nodes.lifecycle(node).await.expect("lifecycle"),
|
||||
Lifecycle::Offline
|
||||
);
|
||||
}
|
||||
|
||||
fn message(sequence: u64, body: node_message::Body) -> NodeMessage {
|
||||
NodeMessage {
|
||||
sequence,
|
||||
message_id: format!("message-{sequence}"),
|
||||
idempotency_key: format!("message-{sequence}"),
|
||||
body: Some(body),
|
||||
}
|
||||
}
|
||||
|
||||
fn capabilities() -> Capabilities {
|
||||
Capabilities {
|
||||
architecture: "amd64".to_owned(),
|
||||
operating_system: "linux".to_owned(),
|
||||
container_runtime: "docker".to_owned(),
|
||||
container_runtime_version: "29.6.2".to_owned(),
|
||||
cores: 2,
|
||||
memory_bytes: 8 * 1024 * 1024 * 1024,
|
||||
labels: vec!["ubuntu-latest".to_owned()],
|
||||
}
|
||||
}
|
||||
|
||||
fn capacity() -> Capacity {
|
||||
Capacity {
|
||||
build_volume_free_bytes: 60 * 1024 * 1024 * 1024,
|
||||
layer_store_bytes: 10 * 1024 * 1024 * 1024,
|
||||
cache_volume_present: true,
|
||||
cache_volume_total_bytes: 100,
|
||||
cache_volume_used_bytes: 40,
|
||||
cache_volume_path: "/var/cache/syncode".to_owned(),
|
||||
cached_images: Vec::new(),
|
||||
cached_actions: Vec::new(),
|
||||
}
|
||||
}
|
||||
#![allow(clippy::expect_used, clippy::panic)]
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use syncode_control_nodes::{Ephemeral, Lifecycle, Nodes, Scope};
|
||||
use syncode_control_runs::{Forgotten, Runs};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
|
||||
use super::{NodeSessionServer, run_session};
|
||||
use crate::wire::{
|
||||
Capabilities, Capacity, Enrol, Heartbeat, Hello, NodeMessage, control_message, node_message,
|
||||
};
|
||||
use crate::{ArtifactTokenAuthority, CapabilityAuthority, RepositoryCoordinates};
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_unreadable_client_releases_its_session() {
|
||||
let runs = Runs::restored(Forgotten::default()).await.expect("runs");
|
||||
let nodes = Nodes::restored(Ephemeral::default()).await.expect("nodes");
|
||||
let token = nodes
|
||||
.issue_token(Scope::Instance)
|
||||
.await
|
||||
.expect("enrolment token");
|
||||
let server = NodeSessionServer::new(
|
||||
runs,
|
||||
nodes.clone(),
|
||||
CapabilityAuthority::new("test-capability-key").expect("capability authority"),
|
||||
ArtifactTokenAuthority::new("test-artifact-key").expect("artifact authority"),
|
||||
"http://127.0.0.1:1/".parse().expect("artifact URL"),
|
||||
RepositoryCoordinates::new("http://127.0.0.1:1".to_owned(), String::new())
|
||||
.expect("repository coordinates"),
|
||||
);
|
||||
let (input, input_receiver) = mpsc::channel(8);
|
||||
let (output, mut output_receiver) = mpsc::channel(8);
|
||||
let session = tokio::spawn(async move {
|
||||
let mut inbound = ReceiverStream::new(input_receiver);
|
||||
run_session(&mut inbound, &output, server).await
|
||||
});
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
1,
|
||||
node_message::Body::Enrol(Enrol {
|
||||
token: token.secret().expose().to_owned(),
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("enrol");
|
||||
let enrolled = output_receiver
|
||||
.recv()
|
||||
.await
|
||||
.expect("enrolment response")
|
||||
.expect("enrolment status");
|
||||
let control_message::Body::Enrolled(enrolled) = enrolled.body.expect("enrolment body") else {
|
||||
panic!("expected enrolment");
|
||||
};
|
||||
let node = enrolled.node.parse().expect("node id");
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
2,
|
||||
node_message::Body::Hello(Hello {
|
||||
node: enrolled.node,
|
||||
credential: enrolled.credential,
|
||||
capabilities: Some(capabilities()),
|
||||
capacity: Some(capacity()),
|
||||
max_parallel: 1,
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("hello");
|
||||
let welcome = output_receiver
|
||||
.recv()
|
||||
.await
|
||||
.expect("welcome response")
|
||||
.expect("welcome status");
|
||||
assert!(matches!(
|
||||
welcome.body,
|
||||
Some(control_message::Body::Welcome(_))
|
||||
));
|
||||
drop(output_receiver);
|
||||
|
||||
input
|
||||
.send(Ok(message(
|
||||
3,
|
||||
node_message::Body::Heartbeat(Heartbeat {
|
||||
held: Vec::new(),
|
||||
capacity: Some(capacity()),
|
||||
}),
|
||||
)))
|
||||
.await
|
||||
.expect("heartbeat");
|
||||
tokio::time::timeout(Duration::from_secs(1), session)
|
||||
.await
|
||||
.expect("session shutdown")
|
||||
.expect("session task")
|
||||
.expect("session cleanup");
|
||||
assert_eq!(
|
||||
nodes.lifecycle(node).await.expect("lifecycle"),
|
||||
Lifecycle::Offline
|
||||
);
|
||||
}
|
||||
|
||||
fn message(sequence: u64, body: node_message::Body) -> NodeMessage {
|
||||
NodeMessage {
|
||||
sequence,
|
||||
message_id: format!("message-{sequence}"),
|
||||
idempotency_key: format!("message-{sequence}"),
|
||||
body: Some(body),
|
||||
}
|
||||
}
|
||||
|
||||
fn capabilities() -> Capabilities {
|
||||
Capabilities {
|
||||
architecture: "amd64".to_owned(),
|
||||
operating_system: "linux".to_owned(),
|
||||
container_runtime: "docker".to_owned(),
|
||||
container_runtime_version: "29.6.2".to_owned(),
|
||||
cores: 2,
|
||||
memory_bytes: 8 * 1024 * 1024 * 1024,
|
||||
labels: vec!["ubuntu-latest".to_owned()],
|
||||
}
|
||||
}
|
||||
|
||||
fn capacity() -> Capacity {
|
||||
Capacity {
|
||||
build_volume_free_bytes: 60 * 1024 * 1024 * 1024,
|
||||
layer_store_bytes: 10 * 1024 * 1024 * 1024,
|
||||
cache_volume_present: true,
|
||||
cache_volume_total_bytes: 100,
|
||||
cache_volume_used_bytes: 40,
|
||||
cache_volume_path: "/var/cache/syncode".to_owned(),
|
||||
cached_images: Vec::new(),
|
||||
cached_actions: Vec::new(),
|
||||
}
|
||||
}
|
||||
@@ -1,138 +1,90 @@
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use axum::Json;
|
||||
use axum::http::StatusCode;
|
||||
use axum::routing::post;
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use syncode_control_runs::{Forgotten, JobId, Origin, Runs};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
_request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
Err(Status::unimplemented("issue_workflow_repository_token"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_runs_are_projected_with_legacy_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let projection_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let projection_base = format!("http://{}/", projection_listener.local_addr()?).parse()?;
|
||||
let (sent, mut received) = mpsc::unbounded_channel();
|
||||
let app = axum::Router::new().route(
|
||||
"/api/internal/actions/syncode/projection",
|
||||
post(move |Json(body): Json<serde_json::Value>| {
|
||||
let sent = sent.clone();
|
||||
async move {
|
||||
let _ = sent.send(body);
|
||||
StatusCode::NO_CONTENT
|
||||
}
|
||||
}),
|
||||
);
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(projection_listener, app).await;
|
||||
});
|
||||
|
||||
let runs = Runs::restored(Forgotten::default()).await?;
|
||||
let run = runs
|
||||
.queue(
|
||||
JobId::fresh(),
|
||||
Origin::new(
|
||||
"76128383-1df5-4979-9b13-c048a5287e9a".to_owned(),
|
||||
"commit".to_owned(),
|
||||
"refs/heads/main".to_owned(),
|
||||
"push".to_owned(),
|
||||
".gitea/workflows/ci.yml".to_owned(),
|
||||
)
|
||||
.with_delivery(Some("delivery".to_owned())),
|
||||
b"plan".to_vec(),
|
||||
)
|
||||
.await?;
|
||||
let projection = runs.projection(run).await?.expect("projection");
|
||||
let client = ProjectionClient::new(
|
||||
projection_base,
|
||||
String::new(),
|
||||
identity_endpoint,
|
||||
"shared-secret".to_owned(),
|
||||
)?;
|
||||
assert_eq!(
|
||||
client
|
||||
.repository_coordinates("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
client.send(&projection).await?;
|
||||
|
||||
let body = received.recv().await.expect("projection body");
|
||||
assert_eq!(body["origin"]["repository"], "syncode/pipelines-demo");
|
||||
assert_eq!(body["origin"]["native"], true);
|
||||
Ok(())
|
||||
}
|
||||
#![allow(clippy::expect_used)]
|
||||
|
||||
use std::error::Error;
|
||||
|
||||
use syncode_control_node::RepositoryCoordinates;
|
||||
use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer};
|
||||
use syncode_control_node::identity_wire::{
|
||||
CheckCapabilityRequest, CheckCapabilityResponse, GetRepositoryCoordinatesRequest,
|
||||
GetRepositoryCoordinatesResponse, IssueWorkflowRepositoryTokenRequest,
|
||||
IssueWorkflowRepositoryTokenResponse, ResolveRepositoryRequest, ResolveRepositoryResponse,
|
||||
ValidateSessionRequest, ValidateSessionResponse,
|
||||
};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
type TestResult<T = ()> = Result<T, Box<dyn Error + Send + Sync>>;
|
||||
|
||||
struct FixtureIdentity;
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Identity for FixtureIdentity {
|
||||
async fn issue_workflow_repository_token(
|
||||
&self,
|
||||
_request: Request<IssueWorkflowRepositoryTokenRequest>,
|
||||
) -> Result<Response<IssueWorkflowRepositoryTokenResponse>, Status> {
|
||||
Err(Status::unimplemented("issue_workflow_repository_token"))
|
||||
}
|
||||
|
||||
async fn validate_session(
|
||||
&self,
|
||||
_request: Request<ValidateSessionRequest>,
|
||||
) -> Result<Response<ValidateSessionResponse>, Status> {
|
||||
Err(Status::unimplemented("validate_session"))
|
||||
}
|
||||
|
||||
async fn check_capability(
|
||||
&self,
|
||||
_request: Request<CheckCapabilityRequest>,
|
||||
) -> Result<Response<CheckCapabilityResponse>, Status> {
|
||||
Err(Status::unimplemented("check_capability"))
|
||||
}
|
||||
|
||||
async fn resolve_repository(
|
||||
&self,
|
||||
_request: Request<ResolveRepositoryRequest>,
|
||||
) -> Result<Response<ResolveRepositoryResponse>, Status> {
|
||||
Err(Status::unimplemented("resolve_repository"))
|
||||
}
|
||||
|
||||
async fn get_repository_coordinates(
|
||||
&self,
|
||||
request: Request<GetRepositoryCoordinatesRequest>,
|
||||
) -> Result<Response<GetRepositoryCoordinatesResponse>, Status> {
|
||||
if request
|
||||
.metadata()
|
||||
.get("authorization")
|
||||
.and_then(|value| value.to_str().ok())
|
||||
!= Some("Bearer shared-secret")
|
||||
{
|
||||
return Err(Status::unauthenticated("missing authorization"));
|
||||
}
|
||||
Ok(Response::new(GetRepositoryCoordinatesResponse {
|
||||
owner: "syncode".to_owned(),
|
||||
name: "pipelines-demo".to_owned(),
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn native_repository_ids_resolve_to_coordinates() -> TestResult {
|
||||
let identity_listener = TcpListener::bind("127.0.0.1:0").await?;
|
||||
let identity_endpoint = format!("http://{}", identity_listener.local_addr()?);
|
||||
tokio::spawn(async move {
|
||||
let _ = Server::builder()
|
||||
.add_service(IdentityServer::new(FixtureIdentity))
|
||||
.serve_with_incoming(TcpListenerStream::new(identity_listener))
|
||||
.await;
|
||||
});
|
||||
|
||||
let client = RepositoryCoordinates::new(identity_endpoint, "shared-secret".to_owned())?;
|
||||
assert_eq!(
|
||||
client
|
||||
.resolve("76128383-1df5-4979-9b13-c048a5287e9a")
|
||||
.await?,
|
||||
"syncode/pipelines-demo"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
@@ -1,184 +1,0 @@
|
||||
use serde::Serialize;
|
||||
use syncode_control_runs::{ProjectedJob, ProjectedRun, RunId, RunNumber};
|
||||
use thiserror::Error;
|
||||
use tonic::Request;
|
||||
use tonic::metadata::{Ascii, MetadataValue};
|
||||
use tonic::transport::{Channel, Endpoint};
|
||||
use url::Url;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::identity_wire::GetRepositoryCoordinatesRequest;
|
||||
use crate::identity_wire::identity_client::IdentityClient;
|
||||
|
||||
/// Pushes a run's projected state to the forge. Dispatch uses it to close the
|
||||
/// gap between handing a node a token and the forge knowing what that token is
|
||||
/// for; the periodic sweep in the binary uses it for everything after.
|
||||
#[derive(Clone)]
|
||||
pub struct ProjectionClient {
|
||||
endpoint: Url,
|
||||
token: String,
|
||||
client: reqwest::Client,
|
||||
identity: IdentityClient<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ProjectionClientError {
|
||||
#[error("cannot address the projection endpoint")]
|
||||
Address,
|
||||
#[error("cannot address the identity endpoint")]
|
||||
IdentityAddress,
|
||||
#[error("cannot authorize identity requests")]
|
||||
IdentityAuthorization,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ProjectionRequestError {
|
||||
#[error(transparent)]
|
||||
Transport(#[from] reqwest::Error),
|
||||
#[error("HTTP {status}: {body}")]
|
||||
Rejected {
|
||||
status: reqwest::StatusCode,
|
||||
body: String,
|
||||
},
|
||||
#[error("identity rejected the repository projection: {0}")]
|
||||
Identity(#[from] tonic::Status),
|
||||
#[error("invalid repository projection origin {0:?}")]
|
||||
InvalidRepository(String),
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct LegacyProjection<'a> {
|
||||
run: &'a RunId,
|
||||
number: &'a RunNumber,
|
||||
origin: LegacyOrigin<'a>,
|
||||
sequence: u64,
|
||||
jobs: &'a [ProjectedJob],
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct LegacyOrigin<'a> {
|
||||
repository: String,
|
||||
commit: &'a str,
|
||||
reference: &'a str,
|
||||
event: &'a str,
|
||||
workflow: &'a str,
|
||||
delivery: Option<&'a str>,
|
||||
principal: Option<&'a str>,
|
||||
secrets_allowed: bool,
|
||||
native: bool,
|
||||
}
|
||||
|
||||
impl ProjectionClient {
|
||||
pub fn new(
|
||||
base: Url,
|
||||
token: String,
|
||||
identity_endpoint: String,
|
||||
identity_shared_secret: String,
|
||||
) -> Result<Self, ProjectionClientError> {
|
||||
let endpoint = base
|
||||
.join("api/internal/actions/syncode/projection")
|
||||
.map_err(|_| ProjectionClientError::Address)?;
|
||||
let identity = IdentityClient::new(
|
||||
Endpoint::from_shared(identity_endpoint)
|
||||
.map_err(|_| ProjectionClientError::IdentityAddress)?
|
||||
.connect_lazy(),
|
||||
);
|
||||
let authorization = format!("Bearer {identity_shared_secret}")
|
||||
.parse()
|
||||
.map_err(|_| ProjectionClientError::IdentityAuthorization)?;
|
||||
Ok(Self {
|
||||
endpoint,
|
||||
token,
|
||||
client: reqwest::Client::new(),
|
||||
identity,
|
||||
authorization,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn send(&self, projection: &ProjectedRun) -> Result<(), ProjectionRequestError> {
|
||||
let projection = self.for_legacy_projection(projection).await?;
|
||||
let response = self
|
||||
.client
|
||||
.post(self.endpoint.clone())
|
||||
.header("X-Gitea-Internal-Auth", format!("Bearer {}", self.token))
|
||||
.json(&projection)
|
||||
.send()
|
||||
.await?;
|
||||
let status = response.status();
|
||||
if status.is_success() {
|
||||
return Ok(());
|
||||
}
|
||||
let body = response.text().await?;
|
||||
Err(ProjectionRequestError::Rejected { status, body })
|
||||
}
|
||||
|
||||
async fn for_legacy_projection<'a>(
|
||||
&self,
|
||||
projection: &'a ProjectedRun,
|
||||
) -> Result<LegacyProjection<'a>, ProjectionRequestError> {
|
||||
let repository = projection.origin.repository();
|
||||
let native = Uuid::parse_str(repository).is_ok();
|
||||
let repository = self.repository_coordinates(repository).await?;
|
||||
Ok(LegacyProjection {
|
||||
run: &projection.run,
|
||||
number: &projection.number,
|
||||
origin: LegacyOrigin {
|
||||
repository,
|
||||
commit: projection.origin.commit(),
|
||||
reference: projection.origin.reference(),
|
||||
event: projection.origin.event(),
|
||||
workflow: projection.origin.workflow(),
|
||||
delivery: projection.origin.delivery(),
|
||||
principal: projection.origin.principal(),
|
||||
secrets_allowed: projection.origin.secrets_allowed(),
|
||||
native,
|
||||
},
|
||||
sequence: projection.sequence,
|
||||
jobs: &projection.jobs,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn repository_coordinates(
|
||||
&self,
|
||||
repository: &str,
|
||||
) -> Result<String, ProjectionRequestError> {
|
||||
if Uuid::parse_str(repository).is_ok() {
|
||||
let mut request = Request::new(GetRepositoryCoordinatesRequest {
|
||||
repository_id: repository.to_owned(),
|
||||
});
|
||||
request
|
||||
.metadata_mut()
|
||||
.insert("authorization", self.authorization.clone());
|
||||
let coordinates = self
|
||||
.identity
|
||||
.clone()
|
||||
.get_repository_coordinates(request)
|
||||
.await?
|
||||
.into_inner();
|
||||
if coordinates.owner.is_empty()
|
||||
|| coordinates.name.is_empty()
|
||||
|| coordinates.owner.contains('/')
|
||||
|| coordinates.name.contains('/')
|
||||
{
|
||||
return Err(ProjectionRequestError::InvalidRepository(format!(
|
||||
"{}/{}",
|
||||
coordinates.owner, coordinates.name
|
||||
)));
|
||||
}
|
||||
Ok(format!("{}/{}", coordinates.owner, coordinates.name))
|
||||
} else {
|
||||
let Some((owner, name)) = repository.split_once('/') else {
|
||||
return Err(ProjectionRequestError::InvalidRepository(
|
||||
repository.to_owned(),
|
||||
));
|
||||
};
|
||||
if owner.is_empty() || name.is_empty() || name.contains('/') {
|
||||
return Err(ProjectionRequestError::InvalidRepository(
|
||||
repository.to_owned(),
|
||||
));
|
||||
}
|
||||
Ok(repository.to_owned())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,0 +1,92 @@
|
||||
use thiserror::Error;
|
||||
use tonic::Request;
|
||||
use tonic::metadata::{Ascii, MetadataValue};
|
||||
use tonic::transport::{Channel, Endpoint};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::identity_wire::GetRepositoryCoordinatesRequest;
|
||||
use crate::identity_wire::identity_client::IdentityClient;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct RepositoryCoordinates {
|
||||
identity: IdentityClient<Channel>,
|
||||
authorization: MetadataValue<Ascii>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RepositoryCoordinatesError {
|
||||
#[error("cannot address the identity endpoint")]
|
||||
IdentityAddress,
|
||||
#[error("cannot authorize identity requests")]
|
||||
IdentityAuthorization,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RepositoryCoordinateRequestError {
|
||||
#[error("identity rejected the repository coordinate request: {0}")]
|
||||
Identity(#[from] tonic::Status),
|
||||
#[error("invalid repository origin {0:?}")]
|
||||
InvalidRepository(String),
|
||||
}
|
||||
|
||||
impl RepositoryCoordinates {
|
||||
pub fn new(
|
||||
identity_endpoint: String,
|
||||
identity_shared_secret: String,
|
||||
) -> Result<Self, RepositoryCoordinatesError> {
|
||||
let identity = IdentityClient::new(
|
||||
Endpoint::from_shared(identity_endpoint)
|
||||
.map_err(|_| RepositoryCoordinatesError::IdentityAddress)?
|
||||
.connect_lazy(),
|
||||
);
|
||||
let authorization = format!("Bearer {identity_shared_secret}")
|
||||
.parse()
|
||||
.map_err(|_| RepositoryCoordinatesError::IdentityAuthorization)?;
|
||||
Ok(Self {
|
||||
identity,
|
||||
authorization,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn resolve(
|
||||
&self,
|
||||
repository: &str,
|
||||
) -> Result<String, RepositoryCoordinateRequestError> {
|
||||
if Uuid::parse_str(repository).is_ok() {
|
||||
let mut request = Request::new(GetRepositoryCoordinatesRequest {
|
||||
repository_id: repository.to_owned(),
|
||||
});
|
||||
request
|
||||
.metadata_mut()
|
||||
.insert("authorization", self.authorization.clone());
|
||||
let coordinates = self
|
||||
.identity
|
||||
.clone()
|
||||
.get_repository_coordinates(request)
|
||||
.await?
|
||||
.into_inner();
|
||||
if coordinates.owner.is_empty()
|
||||
|| coordinates.name.is_empty()
|
||||
|| coordinates.owner.contains('/')
|
||||
|| coordinates.name.contains('/')
|
||||
{
|
||||
return Err(RepositoryCoordinateRequestError::InvalidRepository(
|
||||
format!("{}/{}", coordinates.owner, coordinates.name),
|
||||
));
|
||||
}
|
||||
Ok(format!("{}/{}", coordinates.owner, coordinates.name))
|
||||
} else {
|
||||
let Some((owner, name)) = repository.split_once('/') else {
|
||||
return Err(RepositoryCoordinateRequestError::InvalidRepository(
|
||||
repository.to_owned(),
|
||||
));
|
||||
};
|
||||
if owner.is_empty() || name.is_empty() || name.contains('/') {
|
||||
return Err(RepositoryCoordinateRequestError::InvalidRepository(
|
||||
repository.to_owned(),
|
||||
));
|
||||
}
|
||||
Ok(repository.to_owned())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,53 +1,0 @@
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
|
||||
use syncode_control_node::ProjectionClient;
|
||||
use syncode_control_runs::{ProjectedRun, RunId, RunLog, Runs, RunsError};
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum ProjectionError {
|
||||
#[error("cannot read the run projection: {0}")]
|
||||
Runs(#[from] RunsError),
|
||||
#[error("pending projection for run {0} is missing")]
|
||||
MissingPending(RunId),
|
||||
}
|
||||
|
||||
pub async fn serve<L: RunLog>(
|
||||
runs: Runs<L>,
|
||||
client: ProjectionClient,
|
||||
min_run_number: u64,
|
||||
) -> Result<(), ProjectionError> {
|
||||
let mut delivered = HashMap::<RunId, (u64, u64)>::new();
|
||||
let mut pending = HashMap::<RunId, ((u64, u64), ProjectedRun)>::new();
|
||||
loop {
|
||||
for (run, number, sequence, output_version) in runs.projection_versions().await {
|
||||
if number.get() < min_run_number {
|
||||
continue;
|
||||
}
|
||||
let version = (sequence, output_version);
|
||||
if delivered.get(&run) == Some(&version) {
|
||||
continue;
|
||||
}
|
||||
if pending.get(&run).map(|(version, _)| *version) != Some(version) {
|
||||
let Some(projection) = runs.projection(run).await? else {
|
||||
continue;
|
||||
};
|
||||
pending.insert(run, (version, projection));
|
||||
}
|
||||
let (_, projection) = pending
|
||||
.get(&run)
|
||||
.ok_or(ProjectionError::MissingPending(run))?;
|
||||
match client.send(projection).await {
|
||||
Ok(()) => {
|
||||
delivered.insert(run, version);
|
||||
pending.remove(&run);
|
||||
}
|
||||
Err(error) => {
|
||||
eprintln!("cannot project run state: {error}");
|
||||
}
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(250)).await;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user