From c8379d5d2861480b38a69400d55b3daba66f97f0 Mon Sep 17 00:00:00 2001 From: Przemyslaw Olszewski Date: Sun, 30 Aug 2026 15:22:00 +0200 Subject: [PATCH] fix: read action sources from repository plane --- diff --git a/src/action_repository.rs b/src/action_repository.rs --- a/src/action_repository.rs +++ b/src/action_repository.rs @@ -1,125 +1,169 @@ -use std::io::Read; - -use flate2::read::GzDecoder; -use serde::Deserialize; -use syncode_workflow::{GitCommit, RepositoryUrl}; -use url::Url; - -use crate::actions::{ActionRepositoryPort, ActionResolutionError, RepositorySnapshot}; - -#[derive(Clone)] -pub struct ForgeActionRepository { - client: reqwest::Client, - token: String, -} - -#[derive(Deserialize)] -struct CommitResponse { - sha: String, -} - -impl ForgeActionRepository { - #[must_use] - pub fn new(token: String) -> Self { - Self { - client: reqwest::Client::new(), - token, - } - } - - async fn get(&self, url: Url) -> Result { - let response = self - .client - .get(url.clone()) - .header("Authorization", format!("token {}", self.token)) - .send() - .await - .map_err(|error| ActionResolutionError::Repository(error.to_string()))?; - if !response.status().is_success() { - return Err(ActionResolutionError::Repository(format!( - "{} returned {}", - url, - response.status() - ))); - } - Ok(response) - } -} - -impl ActionRepositoryPort for ForgeActionRepository { - async fn fetch( - &self, - repository: RepositoryUrl, - revision: String, - ) -> Result { - let (mut api, owner, name) = repository_coordinates(&repository)?; - { - let mut segments = api - .path_segments_mut() - .map_err(|_| ActionResolutionError::Repository(repository.to_string()))?; - segments.extend([ - "api", "v1", "repos", &owner, &name, "git", "commits", &revision, - ]); - } - api.query_pairs_mut() - .append_pair("stat", "false") - .append_pair("verification", "false") - .append_pair("files", "false"); - let commit = self - .get(api) - .await? - .json::() - .await - .map_err(|error| ActionResolutionError::Repository(error.to_string()))? - .sha - .parse::() - .map_err(|error| ActionResolutionError::Repository(error.to_string()))?; - - let (mut archive_url, owner, name) = repository_coordinates(&repository)?; - { - let archive = format!("{commit}.tar.gz"); - let mut segments = archive_url - .path_segments_mut() - .map_err(|_| ActionResolutionError::Repository(repository.to_string()))?; - segments.extend(["api", "v1", "repos", &owner, &name, "archive", &archive]); - } - let compressed = self - .get(archive_url) - .await? - .bytes() - .await - .map_err(|error| ActionResolutionError::Repository(error.to_string()))?; - let mut archive = Vec::new(); - GzDecoder::new(compressed.as_ref()) - .read_to_end(&mut archive) - .map_err(|error| ActionResolutionError::Repository(error.to_string()))?; - Ok(RepositorySnapshot { commit, archive }) - } -} - -fn repository_coordinates( - repository: &RepositoryUrl, -) -> Result<(Url, String, String), ActionResolutionError> { - let url = Url::parse(repository.as_ref()) - .map_err(|error| ActionResolutionError::Repository(error.to_string()))?; - let segments = url - .path_segments() - .ok_or_else(|| ActionResolutionError::Repository(repository.to_string()))? - .filter(|segment| !segment.is_empty()) - .map(str::to_owned) - .collect::>(); - let [owner, name] = segments.as_slice() else { - return Err(ActionResolutionError::Repository(format!( - "{repository} must identify exactly one owner and repository" - ))); - }; - let mut base = url; - base.set_path("/"); - base.set_query(None); - base.set_fragment(None); - Ok(( - base, - owner.to_owned(), - name.trim_end_matches(".git").to_owned(), - )) -} +use syncode_control_node::identity_wire::ResolveRepositoryRequest; +use syncode_control_node::identity_wire::identity_client::IdentityClient; +use syncode_repository_api_grpc::proto::repository_read_client::RepositoryReadClient; +use syncode_repository_api_grpc::proto::{ + ArchiveFormat, ArchiveRequest, GetCommitRequest, ObjectId, ResolveRevisionRequest, +}; +use syncode_workflow::{GitCommit, RepositoryUrl}; +use tonic::Request; +use tonic::metadata::{Ascii, MetadataValue}; +use tonic::transport::Channel; +use url::Url; + +use crate::actions::{ActionRepositoryPort, ActionResolutionError, RepositorySnapshot}; + +#[derive(Clone)] +pub struct NativeActionRepository { + repositories: RepositoryReadClient, + identity: IdentityClient, + authorization: MetadataValue, +} + +impl NativeActionRepository { + pub async fn connect( + repository_endpoint: String, + identity_endpoint: String, + identity_shared_secret: String, + ) -> Result { + let repositories = RepositoryReadClient::connect(repository_endpoint) + .await + .map_err(repository_error)?; + let identity = IdentityClient::connect(identity_endpoint) + .await + .map_err(repository_error)?; + let authorization = format!("Bearer {identity_shared_secret}") + .parse() + .map_err(repository_error)?; + Ok(Self { + repositories, + identity, + authorization, + }) + } + + async fn repository_id( + &self, + repository: &RepositoryUrl, + ) -> Result { + match repository_address(repository)? { + RepositoryAddress::Id(id) => Ok(id), + RepositoryAddress::Coordinates { owner, name } => { + let mut request = Request::new(ResolveRepositoryRequest { owner, name }); + request + .metadata_mut() + .insert("authorization", self.authorization.clone()); + Ok(self + .identity + .clone() + .resolve_repository(request) + .await + .map_err(repository_error)? + .into_inner() + .repository_id) + } + } + } + + async fn archive( + &self, + repository_id: String, + tree: ObjectId, + ) -> Result, ActionResolutionError> { + let mut stream = self + .repositories + .clone() + .archive(ArchiveRequest { + repository_id, + tree: Some(tree), + format: ArchiveFormat::Tar.into(), + prefix: "source/".to_owned(), + }) + .await + .map_err(repository_error)? + .into_inner(); + let mut archive = Vec::new(); + while let Some(chunk) = stream.message().await.map_err(repository_error)? { + archive.extend(chunk.data); + } + Ok(archive) + } +} + +impl ActionRepositoryPort for NativeActionRepository { + async fn fetch( + &self, + repository: RepositoryUrl, + revision: String, + ) -> Result { + let repository_id = self.repository_id(&repository).await?; + let resolved = self + .repositories + .clone() + .resolve_revision(ResolveRevisionRequest { + repository_id: repository_id.clone(), + revision, + }) + .await + .map_err(repository_error)? + .into_inner(); + let object = resolved.peeled.ok_or_else(|| { + ActionResolutionError::Repository( + "repository service omitted the peeled revision".to_owned(), + ) + })?; + let commit = self + .repositories + .clone() + .get_commit(GetCommitRequest { + repository_id: repository_id.clone(), + object_id: Some(object), + }) + .await + .map_err(repository_error)? + .into_inner(); + let commit_id = commit.id.ok_or_else(|| { + ActionResolutionError::Repository("repository service omitted the commit id".to_owned()) + })?; + let tree = commit.tree.ok_or_else(|| { + ActionResolutionError::Repository( + "repository service omitted the commit tree".to_owned(), + ) + })?; + let commit = commit_id + .hex + .parse::() + .map_err(repository_error)?; + let archive = self.archive(repository_id, tree).await?; + Ok(RepositorySnapshot { commit, archive }) + } +} + +enum RepositoryAddress { + Id(String), + Coordinates { owner: String, name: String }, +} + +fn repository_address( + repository: &RepositoryUrl, +) -> Result { + let url = Url::parse(repository.as_ref()).map_err(repository_error)?; + let segments = url + .path_segments() + .ok_or_else(|| ActionResolutionError::Repository(repository.to_string()))? + .filter(|segment| !segment.is_empty()) + .collect::>(); + match segments.as_slice() { + [id] if uuid::Uuid::parse_str(id).is_ok() => Ok(RepositoryAddress::Id((*id).to_owned())), + [owner, name] => Ok(RepositoryAddress::Coordinates { + owner: (*owner).to_owned(), + name: name.trim_end_matches(".git").to_owned(), + }), + _ => Err(ActionResolutionError::Repository(format!( + "{repository} must identify one repository id or owner and repository" + ))), + } +} + +fn repository_error(error: impl std::fmt::Display) -> ActionResolutionError { + ActionResolutionError::Repository(error.to_string()) +} diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -1,374 +1,381 @@ -use std::error::Error; -use std::net::SocketAddr; -use std::path::PathBuf; -use std::sync::Arc; - -use clap::{Parser, Subcommand, ValueEnum}; -use syncode_control::action_delivery::router as action_delivery_router; -use syncode_control::action_oci::PinnedOciResolver; -use syncode_control::action_repository::ForgeActionRepository; -use syncode_control::action_store::FileActionStore; -use syncode_control::actions::ActionResolver; -use syncode_control::actions_read::ActionsRead; -use syncode_control::actions_read_identity::IdentityActionsAuthorization; -use syncode_control::admin::{Admin, router as admin_router}; -use syncode_control::check_events::{CheckEventPublisher, CheckEventPublisherConfig}; -use syncode_control::checks::Checks; -use syncode_control::maintenance; -use syncode_control::projection; -use syncode_control::repository::{RepositoryContents, RepositorySecrets}; -use syncode_control::repository_grpc::NativeRepositoryContents; -use syncode_control::repository_sources::RepositorySources; -use syncode_control::secrets::RuntimeSecrets; -use syncode_control::token::EnrolmentScope; -use syncode_control::webhook::{Intake, router}; -use syncode_control_node::{ - ArtifactTokenAuthority, CapabilityAuthority, GeneratedActionsReadServer, GeneratedChecksServer, - GeneratedSecretsServer, GeneratedServer, NodeSessionServer, ProjectionClient, -}; -use syncode_control_nodes::{Nodes, Scope}; -use syncode_control_runs::{ - AuditConfiguration, AuditControlMode, AuditEvent, AuditPayload, NodeId, Runs, SchedulerPolicy, -}; -use syncode_control_store::Postgres; -use tokio::net::TcpListener; -use tonic::transport::Server; -use url::Url; - -#[derive(Debug, Parser)] -#[command(name = "syncode-control", version, about)] -struct Arguments { - #[command(subcommand)] - command: Option, - - /// 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, - - #[arg( - long, - env = "SYNCODE_ACTION_OCI_REGISTRIES", - value_delimiter = ',', - num_args = 1.. - )] - action_oci_registries: Vec, - - #[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, - - /// 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:` or - /// `repository:/`. - #[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> { - 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::()?.into(); - let token = nodes.issue_token(scope).await?; - println!("{}", token.secret().expose()); - } - Command::RevokeNode { node } => { - nodes.revoke(node.parse::()?).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::()?; - let action_allowlist = arguments - .action_allowlist - .into_iter() - .map(|repository| { - repository - .to_string() - .parse::() - }) - .collect::, _>>()?; - let action_store = FileActionStore::new(arguments.action_store); - let action_resolver = ActionResolver::new( - ForgeActionRepository::new(arguments.repository_source_token.clone()), - action_store.clone(), - PinnedOciResolver::new(arguments.action_oci_registries)?, - action_mirror, - action_allowlist, - ); - let native_repository = NativeRepositoryContents::connect(arguments.repository_grpc).await?; - let repository_sources = RepositorySources::new( - native_repository, - RepositoryContents::new( - arguments.repository_source, - arguments.repository_source_token, - ), - ); - let intake = Arc::new(Intake::new( - repository_sources, - runs.clone(), - action_resolver, - arguments.event_feed_secret, - )); - let admin_token = arguments - .admin_token - .filter(|token| !token.is_empty()) - .ok_or("SYNCODE_CONTROL_ADMIN_TOKEN is required while serving")?; - let artifact_authority = ArtifactTokenAuthority::new(arguments.artifact_signing_key)?; - let http = router(intake) - .merge(admin_router(Arc::new( - Admin::new(runs.clone(), nodes.clone(), admin_token).with_operations(store.clone()), - ))) - .merge(action_delivery_router( - action_store, - artifact_authority.clone(), - runs.clone(), - )); - let events = TcpListener::bind(arguments.listen_events).await?; - let check_events = CheckEventPublisher::new( - store.clone(), - CheckEventPublisherConfig { - endpoint: arguments.collaboration_event_url.as_str(), - secret: arguments.collaboration_event_secret, - source_node_id: arguments.source_node_id, - interval: std::time::Duration::from_secs(arguments.check_event_interval_seconds), - batch: arguments.check_event_batch, - }, - )?; - - let authority = CapabilityAuthority::new(arguments.capability_signing_key)?; - let projection = ProjectionClient::new(arguments.projection_url, arguments.projection_token)?; - let secret_service = RuntimeSecrets::new( - runs.clone(), - nodes.clone(), - RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token), - authority.clone(), - ); - let actions_read = ActionsRead::new( - runs.clone(), - IdentityActionsAuthorization::connect( - arguments.identity_grpc, - arguments.identity_shared_secret, - ) - .await?, - ); - - // Both ends of the control plane run for as long as the other does: without - // events there is nothing to assign, and without sessions there is nobody to - // assign it to. Whichever stops first takes the process down with it. - let node_service = match arguments.mode { - Mode::Shadow => NodeSessionServer::shadow( - runs.clone(), - nodes, - authority, - artifact_authority, - arguments.action_artifact_public_url, - projection.clone(), - ), - Mode::Active => NodeSessionServer::new( - runs.clone(), - nodes, - authority, - artifact_authority, - arguments.action_artifact_public_url, - projection.clone(), - ), - }; - tokio::select! { - served = axum::serve(events, http).into_future() => served?, - served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?, - published = check_events.run() => published?, - served = Server::builder() - .add_service(GeneratedServer::new(node_service)) - .add_service(GeneratedSecretsServer::new(secret_service)) - .add_service(GeneratedChecksServer::new(Checks::new(runs.clone()))) - .add_service(GeneratedActionsReadServer::new(actions_read)) - .serve_with_shutdown(arguments.listen, shutdown()) => served?, - } - - Ok(()) -} - -async fn shutdown() { - if let Err(error) = tokio::signal::ctrl_c().await { - eprintln!("cannot listen for shutdown, keeping the service running: {error}"); - std::future::pending::<()>().await; - } -} +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_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, + + /// 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, + + #[arg( + long, + env = "SYNCODE_ACTION_OCI_REGISTRIES", + value_delimiter = ',', + num_args = 1.. + )] + action_oci_registries: Vec, + + #[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, + + /// 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:` or + /// `repository:/`. + #[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> { + 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::()?.into(); + let token = nodes.issue_token(scope).await?; + println!("{}", token.secret().expose()); + } + Command::RevokeNode { node } => { + nodes.revoke(node.parse::()?).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::()?; + let action_allowlist = arguments + .action_allowlist + .into_iter() + .map(|repository| { + repository + .to_string() + .parse::() + }) + .collect::, _>>()?; + 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)?; + let secret_service = RuntimeSecrets::new( + runs.clone(), + nodes.clone(), + RepositorySecrets::new(arguments.secret_source, arguments.secret_source_token), + authority.clone(), + ); + let actions_read = ActionsRead::new( + runs.clone(), + IdentityActionsAuthorization::connect( + arguments.identity_grpc, + arguments.identity_shared_secret, + ) + .await?, + ); + + // Both ends of the control plane run for as long as the other does: without + // events there is nothing to assign, and without sessions there is nobody to + // assign it to. Whichever stops first takes the process down with it. + let node_service = match arguments.mode { + Mode::Shadow => NodeSessionServer::shadow( + runs.clone(), + nodes, + authority, + artifact_authority, + arguments.action_artifact_public_url, + projection.clone(), + ), + Mode::Active => NodeSessionServer::new( + runs.clone(), + nodes, + authority, + artifact_authority, + arguments.action_artifact_public_url, + projection.clone(), + ), + }; + tokio::select! { + served = axum::serve(events, http).into_future() => served?, + served = projection::serve(runs.clone(), projection, arguments.projection_min_run_number) => served?, + published = check_events.run() => published?, + served = Server::builder() + .add_service(GeneratedServer::new(node_service)) + .add_service(GeneratedSecretsServer::new(secret_service)) + .add_service(GeneratedChecksServer::new(Checks::new(runs.clone()))) + .add_service(GeneratedActionsReadServer::new(actions_read)) + .serve_with_shutdown(arguments.listen, shutdown()) => served?, + } + + Ok(()) +} + +async fn shutdown() { + if let Err(error) = tokio::signal::ctrl_c().await { + eprintln!("cannot listen for shutdown, keeping the service running: {error}"); + std::future::pending::<()>().await; + } +} diff --git a/tests/support/repository.rs b/tests/support/repository.rs --- a/tests/support/repository.rs +++ b/tests/support/repository.rs @@ -1,215 +1,240 @@ -use std::pin::Pin; - -use syncode_repository_api_grpc::proto::repository_read_server::{ - RepositoryRead, RepositoryReadServer, -}; -use syncode_repository_api_grpc::proto::*; -use tokio_stream::Stream; -use tonic::{Request, Response, Status}; - -pub const REPOSITORY: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89210"; -pub const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c"; -pub const TAG_OBJECT: &str = "8888888888888888888888888888888888888888"; -const ROOT: &str = "1111111111111111111111111111111111111111"; -const GITEA: &str = "2222222222222222222222222222222222222222"; -const WORKFLOWS: &str = "3333333333333333333333333333333333333333"; -const CI: &str = "4444444444444444444444444444444444444444"; -const OTHER: &str = "5555555555555555555555555555555555555555"; -const SRC: &str = "6666666666666666666666666666666666666666"; -const MAIN: &str = "7777777777777777777777777777777777777777"; - -pub const WORKFLOW: &str = r#" -name: CI -on: - push: - branches: [main] - tags: ["v*"] - pull_request: - branches: [main] - paths: ["src/**"] -jobs: - build: - runs-on: [self-hosted, linux] - steps: - - run: echo native -"#; - -const OTHER_WORKFLOW: &str = r#" -name: Other -on: [workflow_dispatch] -jobs: - other: - runs-on: [self-hosted, linux] - steps: - - run: echo other -"#; - -#[derive(Default)] -struct FixtureRepository; - -pub fn service() -> RepositoryReadServer { - RepositoryReadServer::new(FixtureRepository) -} - -#[tonic::async_trait] -impl RepositoryRead for FixtureRepository { - async fn get_commit( - &self, - _request: Request, - ) -> Result, Status> { - Ok(Response::new(Commit { - id: Some(object(COMMIT)), - tree: Some(object(ROOT)), - parents: Vec::new(), - message: Vec::new(), - })) - } - - async fn list_tree(&self, request: Request) -> Result, Status> { - let tree = request - .into_inner() - .object_id - .ok_or_else(|| Status::invalid_argument("missing tree"))?; - let entries = match tree.hex.as_str() { - ROOT => vec![ - entry(b".gitea", GITEA, TreeEntryKind::Tree), - entry(b"src", SRC, TreeEntryKind::Tree), - ], - GITEA => vec![entry(b"workflows", WORKFLOWS, TreeEntryKind::Tree)], - WORKFLOWS => vec![ - entry(b"ci.yml", CI, TreeEntryKind::Blob), - entry(b"other.yml", OTHER, TreeEntryKind::Blob), - ], - SRC => vec![entry(b"main.rs", MAIN, TreeEntryKind::Blob)], - _ => return Err(Status::not_found("tree")), - }; - Ok(Response::new(Tree { entries })) - } - - type GetBlobStream = Pin> + Send>>; - - async fn get_blob( - &self, - request: Request, - ) -> Result, Status> { - let object = request - .into_inner() - .object_id - .ok_or_else(|| Status::invalid_argument("missing blob"))?; - let content = match object.hex.as_str() { - CI => WORKFLOW, - OTHER => OTHER_WORKFLOW, - _ => return Err(Status::not_found("blob")), - }; - Ok(Response::new(Box::pin(tokio_stream::iter([Ok( - BlobChunk { - data: content.as_bytes().to_vec(), - }, - )])))) - } - - async fn diff(&self, _request: Request) -> Result, Status> { - Ok(Response::new(DiffResult { - files: vec![DiffFile { - old_path: b"src/main.rs".to_vec(), - new_path: b"src/main.rs".to_vec(), - kind: DiffChangeKind::Modified.into(), - ..Default::default() - }], - })) - } - - type ArchiveStream = Pin> + Send>>; - - async fn resolve_revision( - &self, - request: Request, - ) -> Result, Status> { - let revision = request.into_inner().revision; - let target = object(&revision); - Ok(Response::new(ResolvedRevision { - target: Some(target), - peeled: Some(object(COMMIT)), - kind: ObjectKind::Commit.into(), - })) - } - - async fn list_commits( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("list_commits")) - } - - async fn merge_base( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("merge_base")) - } - - async fn blame( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("blame")) - } - - async fn language_statistics( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("language_statistics")) - } - - async fn activity( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("activity")) - } - - async fn verify_commit_signature( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("verify_commit_signature")) - } - - async fn list_lfs_pointers( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("list_lfs_pointers")) - } - - async fn search_code( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("search_code")) - } - - async fn archive( - &self, - _request: Request, - ) -> Result, Status> { - Err(Status::unimplemented("archive")) - } -} - -fn object(hex: &str) -> ObjectId { - ObjectId { - format: ObjectFormat::Sha1.into(), - hex: hex.to_owned(), - } -} - -fn entry(name: &[u8], id: &str, kind: TreeEntryKind) -> TreeEntry { - TreeEntry { - name: name.to_vec(), - id: Some(object(id)), - kind: kind.into(), - mode: 0, - } -} +use std::pin::Pin; + +use syncode_repository_api_grpc::proto::repository_read_server::{ + RepositoryRead, RepositoryReadServer, +}; +use syncode_repository_api_grpc::proto::*; +use tokio_stream::Stream; +use tonic::{Request, Response, Status}; + +pub const REPOSITORY: &str = "018f47e2-b2c4-7f19-8a6d-13ef76c89210"; +pub const COMMIT: &str = "9f2c1e4a7b3d5f6081a2c3d4e5f60718293a4b5c"; +pub const TAG_OBJECT: &str = "8888888888888888888888888888888888888888"; +pub const ROOT: &str = "1111111111111111111111111111111111111111"; +const GITEA: &str = "2222222222222222222222222222222222222222"; +const WORKFLOWS: &str = "3333333333333333333333333333333333333333"; +const CI: &str = "4444444444444444444444444444444444444444"; +const OTHER: &str = "5555555555555555555555555555555555555555"; +const SRC: &str = "6666666666666666666666666666666666666666"; +const MAIN: &str = "7777777777777777777777777777777777777777"; + +pub const WORKFLOW: &str = r#" +name: CI +on: + push: + branches: [main] + tags: ["v*"] + pull_request: + branches: [main] + paths: ["src/**"] +jobs: + build: + runs-on: [self-hosted, linux] + steps: + - run: echo native +"#; + +const OTHER_WORKFLOW: &str = r#" +name: Other +on: [workflow_dispatch] +jobs: + other: + runs-on: [self-hosted, linux] + steps: + - run: echo other +"#; + +#[derive(Default)] +struct FixtureRepository; + +pub fn service() -> RepositoryReadServer { + RepositoryReadServer::new(FixtureRepository) +} + +#[tonic::async_trait] +impl RepositoryRead for FixtureRepository { + async fn get_commit( + &self, + _request: Request, + ) -> Result, Status> { + Ok(Response::new(Commit { + id: Some(object(COMMIT)), + tree: Some(object(ROOT)), + parents: Vec::new(), + message: Vec::new(), + })) + } + + async fn list_tree(&self, request: Request) -> Result, Status> { + let tree = request + .into_inner() + .object_id + .ok_or_else(|| Status::invalid_argument("missing tree"))?; + let entries = match tree.hex.as_str() { + ROOT => vec![ + entry(b".gitea", GITEA, TreeEntryKind::Tree), + entry(b"src", SRC, TreeEntryKind::Tree), + ], + GITEA => vec![entry(b"workflows", WORKFLOWS, TreeEntryKind::Tree)], + WORKFLOWS => vec![ + entry(b"ci.yml", CI, TreeEntryKind::Blob), + entry(b"other.yml", OTHER, TreeEntryKind::Blob), + ], + SRC => vec![entry(b"main.rs", MAIN, TreeEntryKind::Blob)], + _ => return Err(Status::not_found("tree")), + }; + Ok(Response::new(Tree { entries })) + } + + type GetBlobStream = Pin> + Send>>; + + async fn get_blob( + &self, + request: Request, + ) -> Result, Status> { + let object = request + .into_inner() + .object_id + .ok_or_else(|| Status::invalid_argument("missing blob"))?; + let content = match object.hex.as_str() { + CI => WORKFLOW, + OTHER => OTHER_WORKFLOW, + _ => return Err(Status::not_found("blob")), + }; + Ok(Response::new(Box::pin(tokio_stream::iter([Ok( + BlobChunk { + data: content.as_bytes().to_vec(), + }, + )])))) + } + + async fn diff(&self, _request: Request) -> Result, Status> { + Ok(Response::new(DiffResult { + files: vec![DiffFile { + old_path: b"src/main.rs".to_vec(), + new_path: b"src/main.rs".to_vec(), + kind: DiffChangeKind::Modified.into(), + ..Default::default() + }], + })) + } + + type ArchiveStream = Pin> + Send>>; + + async fn resolve_revision( + &self, + request: Request, + ) -> Result, Status> { + let revision = request.into_inner().revision; + let target = object(&revision); + Ok(Response::new(ResolvedRevision { + target: Some(target), + peeled: Some(object(COMMIT)), + kind: ObjectKind::Commit.into(), + })) + } + + async fn list_commits( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("list_commits")) + } + + async fn merge_base( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("merge_base")) + } + + async fn blame( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("blame")) + } + + async fn language_statistics( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("language_statistics")) + } + + async fn activity( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("activity")) + } + + async fn verify_commit_signature( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("verify_commit_signature")) + } + + async fn list_lfs_pointers( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("list_lfs_pointers")) + } + + async fn search_code( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("search_code")) + } + + async fn archive( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + if request.repository_id != REPOSITORY + || request.tree.as_ref().map(|tree| tree.hex.as_str()) != Some(ROOT) + || request.format != i32::from(ArchiveFormat::Tar) + || request.prefix != "source/" + { + return Err(Status::invalid_argument("unexpected archive request")); + } + let content = b"name: fixture\nruns:\n using: node20\n main: index.js\n"; + let mut archive = tar::Builder::new(Vec::new()); + let mut header = tar::Header::new_gnu(); + header + .set_path("source/action.yml") + .map_err(|error| Status::internal(error.to_string()))?; + header.set_size(content.len() as u64); + header.set_mode(0o644); + header.set_cksum(); + archive + .append(&header, content.as_slice()) + .map_err(|error| Status::internal(error.to_string()))?; + let bytes = archive + .into_inner() + .map_err(|error| Status::internal(error.to_string()))?; + Ok(Response::new(Box::pin(tokio_stream::iter([Ok( + BlobChunk { data: bytes }, + )])))) + } +} + +fn object(hex: &str) -> ObjectId { + ObjectId { + format: ObjectFormat::Sha1.into(), + hex: hex.to_owned(), + } +} + +fn entry(name: &[u8], id: &str, kind: TreeEntryKind) -> TreeEntry { + TreeEntry { + name: name.to_vec(), + id: Some(object(id)), + kind: kind.into(), + mode: 0, + } +} diff --git a/tests/action_repository.rs b/tests/action_repository.rs --- /dev/null +++ b/tests/action_repository.rs @@ -1,0 +1,121 @@ +#![allow(clippy::expect_used)] + +#[path = "support/repository.rs"] +#[allow(dead_code)] +mod repository; + +use std::error::Error; + +use syncode_control::action_repository::NativeActionRepository; +use syncode_control::actions::ActionRepositoryPort; +use syncode_control_node::identity_wire::identity_server::{Identity, IdentityServer}; +use syncode_control_node::identity_wire::{ + CheckCapabilityRequest, CheckCapabilityResponse, ResolveRepositoryRequest, + ResolveRepositoryResponse, ValidateSessionRequest, ValidateSessionResponse, +}; +use tokio::net::TcpListener; +use tokio_stream::wrappers::TcpListenerStream; +use tonic::transport::Server; +use tonic::{Request, Response, Status}; + +type TestResult = Result>; + +#[derive(Default)] +struct FixtureIdentity; + +#[tonic::async_trait] +impl Identity for FixtureIdentity { + async fn validate_session( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("validate_session")) + } + + async fn check_capability( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unimplemented("check_capability")) + } + + async fn resolve_repository( + &self, + request: Request, + ) -> Result, Status> { + if request + .metadata() + .get("authorization") + .and_then(|value| value.to_str().ok()) + != Some("Bearer shared-secret") + { + return Err(Status::unauthenticated("missing authorization")); + } + let request = request.into_inner(); + if request.owner != "actions" || request.name != "checkout" { + return Err(Status::not_found("repository")); + } + Ok(Response::new(ResolveRepositoryResponse { + repository_id: repository::REPOSITORY.to_owned(), + })) + } +} + +async fn source() -> TestResult { + let repository_listener = TcpListener::bind("127.0.0.1:0").await?; + let repository_endpoint = format!("http://{}", repository_listener.local_addr()?); + tokio::spawn(async move { + let _ = Server::builder() + .add_service(repository::service()) + .serve_with_incoming(TcpListenerStream::new(repository_listener)) + .await; + }); + + 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; + }); + + Ok(NativeActionRepository::connect( + repository_endpoint, + identity_endpoint, + "shared-secret".to_owned(), + ) + .await?) +} + +#[tokio::test] +async fn resolves_coordinates_and_fetches_an_immutable_native_archive() -> TestResult { + let source = source().await?; + let snapshot = source + .fetch( + "https://dev.syncode.sh/actions/checkout".parse()?, + "v4".to_owned(), + ) + .await?; + + assert_eq!(snapshot.commit.as_ref(), repository::COMMIT); + let paths = tar::Archive::new(snapshot.archive.as_slice()) + .entries()? + .map(|entry| entry.and_then(|entry| entry.path().map(|path| path.into_owned()))) + .collect::, _>>()?; + assert_eq!(paths, [std::path::PathBuf::from("source/action.yml")]); + Ok(()) +} + +#[tokio::test] +async fn accepts_a_native_repository_id_without_coordinate_resolution() -> TestResult { + let source = source().await?; + let snapshot = source + .fetch( + format!("https://dev.syncode.sh/{}", repository::REPOSITORY).parse()?, + repository::COMMIT.to_owned(), + ) + .await?; + assert_eq!(snapshot.commit.as_ref(), repository::COMMIT); + Ok(()) +} -- SynCode