feat(aws): start, attach and stop a project on its AWS environment
ProjectHost::Aws and the tunnel registry: start looks for the project's live task before preparing (a running task is attached to with the token read back from its secret, never rotated), waits for a new access point, runs, waits, tunnels and checks the agent; a task the start ran is stopped on any later failure. Stop, remove (resources; data only on request), reconcile (background re-attach) and the AWS death watch, with exits confirmed through the registry. Docker-only Reset, migration, staleness and container info refuse or skip AWS projects; an AWS project with resources can't be moved off its host. Shutdown closes every tunnel. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
shadowdaoandClaude Opus 5.5 committed
1 parent
7fd6fa6aaa
commit
e05f7e3cd8
20 files changed
+2023
-29
No files matched your search
@@ -292,6 +292,12 @@ docker exec stdout → tokio task → emit("terminal-output-{sessionId}") → li
|
||||
when it drops, so an `AgentEndpoint` stays valid. The `StartSession` token goes to the plugin
|
||||
through its environment, never argv, and the plugin's environment is stripped of every
|
||||
inherited `AWS_*` variable before the session's own credentials are added.
|
||||
`aws_host.rs` is `ProjectHost::Aws` (an ECS Fargate task per project; a running task is
|
||||
attached to with the token read back from its secret, never rotated; a task a failed start ran
|
||||
is stopped again) and `tunnels.rs` the registry of open tunnels by project — an AWS project's
|
||||
`ContainerTarget::for_project` is its tunnel's endpoint, its stored `container_id` is that
|
||||
endpoint's URL, and the AWS death watch confirms exits through the registry. Shutdown closes
|
||||
every tunnel. AWS SDK calls stay in `aws/` (`runtime.rs`, `tasks.rs`, `workspace.rs`).
|
||||
`agent.rs` reuses `docker/exec.rs`'s `OneshotOutput`, host-file reader and uid probe so the two
|
||||
variants cannot drift; an `Agent` target's token never appears in `Debug` or error text.
|
||||
**Container-interior code must go through `ContainerTarget`, never `docker::exec`, bollard or
|
||||
|
||||
@@ -13,13 +13,10 @@ pub mod permissions;
|
||||
pub mod policies;
|
||||
#[cfg(test)]
|
||||
mod policy_eval;
|
||||
// Used by the AWS host (phase 5, Task 6); until then only by its tests.
|
||||
#[allow(dead_code)]
|
||||
pub mod runtime;
|
||||
pub mod tasks;
|
||||
pub mod tofu;
|
||||
pub mod tofu_modules;
|
||||
// Used by the AWS host (phase 5, Task 6); until then only by its tests.
|
||||
#[allow(dead_code)]
|
||||
pub mod workspace;
|
||||
|
||||
pub use connection::{sdk_config, sdk_config_for_keys};
|
||||
|
||||
@@ -337,6 +337,7 @@ pub fn quota_or(e: AwsError) -> AwsError {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // for the UI's start errors (Task 8)
|
||||
pub fn is_quota_error(e: &AwsError) -> bool {
|
||||
matches!(e, AwsError::QuotaExceeded { .. })
|
||||
}
|
||||
@@ -480,6 +481,7 @@ async fn stopped_error<A: TaskApi>(api: &A, logs: &TaskLogs, t: &TaskInfo) -> Aw
|
||||
// ----------------------------------------------------------- stop / watch
|
||||
|
||||
/// Why a task stopped, as the app reports it (spec, "Lifecycle and status").
|
||||
#[allow(dead_code)] // the stop reasons the UI shows (Task 8)
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum StopKind {
|
||||
/// Exited on its own (the essential container ended, idle stop, or AWS
|
||||
@@ -493,6 +495,7 @@ pub enum StopKind {
|
||||
UserInitiated,
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // Task 8
|
||||
pub fn classify_stop(info: &TaskInfo) -> StopKind {
|
||||
match info.stop_code.as_deref() {
|
||||
Some("SpotInterruption") => StopKind::Spot,
|
||||
|
||||
@@ -204,6 +204,7 @@ fn serialize_payload(m: &BTreeMap<String, String>) -> Result<String, String> {
|
||||
/// `TRIPLEC_REPOS`, `TRIPLEC_WIP_PUSH` = `1`/`0`, `TRIPLEC_IDLE_STOP_MINUTES`
|
||||
/// when on, `TRIPLEC_STORAGE` = `ephemeral`/`hybrid`), keys sorted. Fails
|
||||
/// when it would exceed 65,536 bytes.
|
||||
#[cfg_attr(not(test), allow(dead_code))]
|
||||
pub fn project_secret_payload(
|
||||
env_pairs: &[(String, String)],
|
||||
agent_token: &str,
|
||||
|
||||
@@ -34,6 +34,11 @@ pub async fn get_container_info(
|
||||
.projects_store
|
||||
.get(&project_id)
|
||||
.ok_or_else(|| format!("Project {} not found", project_id))?;
|
||||
// An AWS project has no Docker container (its own name could match a
|
||||
// stale one from before it moved to AWS).
|
||||
if project.is_aws() {
|
||||
return Ok(None);
|
||||
}
|
||||
docker::get_container_info(&project).await
|
||||
}
|
||||
|
||||
@@ -432,6 +432,11 @@ pub async fn get_container_staleness(
|
||||
.projects_store
|
||||
.get(&project_id)
|
||||
.ok_or_else(|| format!("Project {} not found", project_id))?;
|
||||
// An AWS task has no local container or snapshot to compare: no claim
|
||||
// either way ("unknown"), and nothing is probed.
|
||||
if project.is_aws() {
|
||||
return Ok(ContainerStaleness::default());
|
||||
}
|
||||
let settings = state.settings_store.get();
|
||||
let base_image = crate::models::container_config::resolve_image_name(
|
||||
&settings.image_source,
|
||||
@@ -708,6 +713,9 @@ pub async fn migrate_project_to_base(
|
||||
app_handle: tauri::AppHandle,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<MigrationReport, String> {
|
||||
if let Some(p) = state.projects_store.get(&project_id) {
|
||||
crate::commands::project_commands::refuse_aws(&p, "Updating to the current base image")?;
|
||||
}
|
||||
let _guard = match ActiveGuard::acquire(&project_id) {
|
||||
Ok(guard) => guard,
|
||||
Err(busy) => return Ok(MigrationReport::failed_preflight(&busy)),
|
||||
|
||||
@@ -655,6 +655,7 @@ pub async fn add_project(
|
||||
#[tauri::command]
|
||||
pub async fn remove_project(
|
||||
project_id: String,
|
||||
discard_data: Option<bool>,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<ProjectRemovalReport, String> {
|
||||
// **H-2: the only writer of these three categories that held nothing.**
|
||||
@@ -673,6 +674,32 @@ pub async fn remove_project(
|
||||
let _guard =
|
||||
crate::project_lock::try_acquire(&project_id, crate::project_lock::ProjectOp::Destroy)?;
|
||||
|
||||
// An AWS project's task and resources go first, and a failure keeps the
|
||||
// project: its record is the only handle on a billed task and its
|
||||
// secret. Its data (the EFS access point) goes only when asked.
|
||||
if let Some(project) = state.projects_store.get(&project_id).filter(|p| p.is_aws()) {
|
||||
let settings = state.settings_store.get();
|
||||
if let Some(target) = ContainerTarget::for_project(&project) {
|
||||
state.exec_manager.close_sessions_for_container(target.id()).await;
|
||||
}
|
||||
let connection_exists = project
|
||||
.host_id
|
||||
.as_deref()
|
||||
.is_some_and(|id| settings.hosts.iter().any(|h| h.id == id));
|
||||
if connection_exists {
|
||||
ProjectHost::for_project(&project)
|
||||
.remove_aws_resources(&project, &settings, discard_data.unwrap_or(false))
|
||||
.await?;
|
||||
} else {
|
||||
log::warn!(
|
||||
"Project {}: its AWS connection no longer exists, so its AWS resources can't be \
|
||||
removed from here",
|
||||
project_id
|
||||
);
|
||||
crate::host::tunnels::close(&project_id).await;
|
||||
}
|
||||
}
|
||||
|
||||
// Release any host loopback ports the auth bridge holds for this project
|
||||
// before the container (and the project record) go away.
|
||||
state.auth_bridge.stop(&project_id).await;
|
||||
@@ -1150,11 +1177,23 @@ fn validate_aws_update(
|
||||
)
|
||||
}
|
||||
}
|
||||
if (stored.host_id != project.host_id || stored.aws_placement != project.aws_placement)
|
||||
&& stored.status != ProjectStatus::Stopped
|
||||
{
|
||||
let moved = stored.host_id != project.host_id || stored.aws_placement != project.aws_placement;
|
||||
if moved && stored.status != ProjectStatus::Stopped {
|
||||
return Err("Stop the project before changing its host or placement".into());
|
||||
}
|
||||
// The task, access point and task definition live in the stored
|
||||
// connection's account and region; moving the project would orphan them.
|
||||
if moved
|
||||
&& stored.aws.as_ref().is_some_and(|a| {
|
||||
a.task_arn.is_some() || a.access_point_id.is_some() || a.task_definition_arn.is_some()
|
||||
})
|
||||
{
|
||||
return Err(
|
||||
"This project still has AWS resources (a task, its data's access point or task \
|
||||
definitions) where it runs now; remove the workspace's AWS resources first"
|
||||
.into(),
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -1359,6 +1398,21 @@ async fn start_project_container_locked(
|
||||
|
||||
// Update project with container info using granular methods (Issue 14: TOCTOU)
|
||||
state.projects_store.set_container_id(&project_id, Some(container_id.clone()))?;
|
||||
// AWS: the task's handles, store-owned like the container id. The target
|
||||
// id stored above is the tunnel's endpoint URL, which is what the death
|
||||
// watch reports a stopped task as.
|
||||
if project.is_aws() {
|
||||
if let Some((task_arn, access_point_id, task_definition_arn)) =
|
||||
crate::host::tunnels::runtime(&project_id)
|
||||
{
|
||||
state.projects_store.set_aws_runtime(
|
||||
&project_id,
|
||||
Some(task_arn),
|
||||
access_point_id,
|
||||
Some(task_definition_arn),
|
||||
)?;
|
||||
}
|
||||
}
|
||||
state.projects_store.update_status(&project_id, ProjectStatus::Running)?;
|
||||
|
||||
// Arm the auth bridge if this project opted in. Purely host-side, so it
|
||||
@@ -1404,7 +1458,30 @@ pub async fn stop_project_container(
|
||||
// Drop host listeners first: they only make sense while the container runs.
|
||||
state.auth_bridge.stop(&project_id).await;
|
||||
|
||||
if let Some(container) = ContainerTarget::for_project(&project) {
|
||||
if project.is_aws() {
|
||||
// A failed StopTask leaves a billed task running, so unlike Docker it
|
||||
// is not papered over: the project keeps its status and the user
|
||||
// sees why (expired SSO, typically), and can retry.
|
||||
emit_progress(&app_handle, &project_id, "Stopping task...");
|
||||
if let Some(container) = ContainerTarget::for_project(&project) {
|
||||
state.exec_manager.close_sessions_for_container(container.id()).await;
|
||||
}
|
||||
let settings = state.settings_store.get();
|
||||
if let Err(e) = ProjectHost::for_project(&project).stop_project(&project, &settings).await {
|
||||
log::warn!("Stop failed for project {}: {}", project_id, e);
|
||||
let _ = state.projects_store.update_status(&project_id, project.status.clone());
|
||||
return Err(format!("Could not stop the task: {e}"));
|
||||
}
|
||||
if let Some(a) = project.aws.as_ref() {
|
||||
state.projects_store.set_aws_runtime(
|
||||
&project_id,
|
||||
None,
|
||||
a.access_point_id.clone(),
|
||||
a.task_definition_arn.clone(),
|
||||
)?;
|
||||
}
|
||||
state.projects_store.set_container_id(&project_id, None)?;
|
||||
} else if let Some(container) = ContainerTarget::for_project(&project) {
|
||||
// Close exec sessions for this project
|
||||
emit_progress(&app_handle, &project_id, "Stopping container...");
|
||||
state.exec_manager.close_sessions_for_container(container.id()).await;
|
||||
@@ -1438,6 +1515,7 @@ pub async fn rebuild_project_container(
|
||||
.projects_store
|
||||
.get(&project_id)
|
||||
.ok_or_else(|| format!("Project {} not found", project_id))?;
|
||||
refuse_aws(&project, "Reset")?;
|
||||
|
||||
// Reset supersedes any migration decision that was still pending: the
|
||||
// snapshot image and both volumes are about to go, so a surviving record
|
||||
@@ -1513,6 +1591,20 @@ pub async fn rebuild_project_container(
|
||||
/// [`crate::commands::migration_commands::reconcile_migration`]. The migration
|
||||
/// pass runs over *every* project, not just the Running ones, because a project
|
||||
/// whose container was removed mid-migration reports Stopped.
|
||||
/// How long reconcile waits on AWS for one project.
|
||||
const AWS_RECONCILE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
|
||||
|
||||
/// Refuse a Docker-only operation on an AWS project.
|
||||
pub(crate) fn refuse_aws(project: &Project, what: &str) -> Result<(), String> {
|
||||
if project.is_aws() {
|
||||
return Err(format!(
|
||||
"{what} is not available for a project that runs on AWS; it applies to local \
|
||||
Docker containers only"
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn reconcile_project_statuses(
|
||||
app_handle: tauri::AppHandle,
|
||||
@@ -1520,7 +1612,7 @@ pub async fn reconcile_project_statuses(
|
||||
) -> Result<Vec<Project>, String> {
|
||||
let projects = state.projects_store.list();
|
||||
|
||||
for project in &projects {
|
||||
for project in projects.iter().filter(|p| !p.is_aws()) {
|
||||
crate::commands::migration_commands::reconcile_migration(project, &app_handle).await;
|
||||
}
|
||||
|
||||
@@ -1554,6 +1646,28 @@ pub async fn reconcile_project_statuses(
|
||||
continue;
|
||||
}
|
||||
|
||||
// AWS: ECS is the authority. One DescribeTasks per project, bounded
|
||||
// so a slow or unreachable AWS can't hold the app's start; a live
|
||||
// task's tunnel is re-opened in the background.
|
||||
if project.is_aws() {
|
||||
let settings = state.settings_store.get();
|
||||
let store = state.projects_store.clone();
|
||||
if tokio::time::timeout(
|
||||
AWS_RECONCILE_TIMEOUT,
|
||||
crate::host::aws_host::reconcile(project, &settings, store),
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
log::warn!(
|
||||
"Project {}: AWS did not answer within {}s; status left as is",
|
||||
project.id,
|
||||
AWS_RECONCILE_TIMEOUT.as_secs()
|
||||
);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
let is_running = if let Some(container) = ContainerTarget::for_project(project) {
|
||||
container.is_running().await.unwrap_or(false)
|
||||
} else {
|
||||
@@ -2279,4 +2393,45 @@ mod tests {
|
||||
let mut edited = running.clone(); edited.aws.as_mut().unwrap().spot = true;
|
||||
assert!(validate_aws_update(&running, &edited, &hosts).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_project_with_aws_resources_cannot_be_moved_off_its_host() {
|
||||
let hosts = vec![host("h1"), host("h2")];
|
||||
let mut stored = aws_project();
|
||||
stored.aws.as_mut().unwrap().access_point_id = Some("fsap-1".into());
|
||||
let mut local = stored.clone();
|
||||
local.host_id = None;
|
||||
local.aws_placement = None;
|
||||
local.aws = None;
|
||||
let e = validate_aws_update(&stored, &local, &hosts).unwrap_err();
|
||||
assert!(e.contains("remove the workspace's AWS resources first"), "{e}");
|
||||
let mut other = stored.clone();
|
||||
other.host_id = Some("h2".into());
|
||||
assert!(validate_aws_update(&stored, &other, &hosts).is_err());
|
||||
let mut moved = stored.clone();
|
||||
moved.aws_placement.as_mut().unwrap().environment = "prod".into();
|
||||
assert!(validate_aws_update(&stored, &moved, &hosts).is_err());
|
||||
for set in [
|
||||
|a: &mut crate::models::AwsWorkspaceConfig| a.task_arn = Some("t".into()),
|
||||
|a: &mut crate::models::AwsWorkspaceConfig| a.task_definition_arn = Some("td".into()),
|
||||
] {
|
||||
let mut st = aws_project();
|
||||
set(st.aws.as_mut().unwrap());
|
||||
assert!(validate_aws_update(&st, &local, &hosts).is_err());
|
||||
}
|
||||
// No handles: free to move. Other edits are fine with handles.
|
||||
assert!(validate_aws_update(&aws_project(), &local, &hosts).is_ok());
|
||||
let mut edited = stored.clone();
|
||||
edited.aws.as_mut().unwrap().cpu = 2048;
|
||||
edited.aws.as_mut().unwrap().memory_mib = 8192;
|
||||
assert!(validate_aws_update(&stored, &edited, &hosts).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn docker_only_operations_refuse_aws_projects() {
|
||||
assert!(refuse_aws(&aws_project(), "Reset").unwrap_err().contains("Reset is not available"));
|
||||
let mut local = aws_project();
|
||||
local.host_id = None;
|
||||
assert!(refuse_aws(&local, "Reset").is_ok());
|
||||
}
|
||||
}
|
||||
@@ -160,7 +160,9 @@ pub async fn open_terminal_session(
|
||||
.get(&project_id)
|
||||
.ok_or_else(|| format!("Project {} not found", project_id))?;
|
||||
|
||||
let target = ContainerTarget::for_project(&project)
|
||||
// Waits while an AWS project's tunnel is being re-opened after a restart.
|
||||
let target = ContainerTarget::for_project_waiting(&project)
|
||||
.await
|
||||
.ok_or_else(|| "Container not running".to_string())?;
|
||||
|
||||
if let Some(id) = resume_session_id.as_deref() {
|
||||
|
||||
@@ -219,13 +219,28 @@ pub fn spawn(
|
||||
cancel: watch::Receiver<bool>,
|
||||
) -> tauri::async_runtime::JoinHandle<()> {
|
||||
tauri::async_runtime::spawn(async move {
|
||||
run(store, source, cancel, |p, reason| {
|
||||
let store_for_aws = store.clone();
|
||||
run(store, source, cancel, move |p, reason| {
|
||||
log::warn!(
|
||||
"Project '{}' ({}): its container stopped without the app asking ({:?}) — marking Stopped",
|
||||
p.name,
|
||||
p.id,
|
||||
reason
|
||||
);
|
||||
// An AWS task that stopped: its tunnel leads nowhere now, and its
|
||||
// task ARN is history (the data's access point stays).
|
||||
if p.is_aws() {
|
||||
if let Some(a) = p.aws.as_ref() {
|
||||
let _ = store_for_aws.set_aws_runtime(
|
||||
&p.id,
|
||||
None,
|
||||
a.access_point_id.clone(),
|
||||
a.task_definition_arn.clone(),
|
||||
);
|
||||
}
|
||||
let id = p.id.clone();
|
||||
tauri::async_runtime::spawn(async move { crate::host::tunnels::close(&id).await });
|
||||
}
|
||||
let _ = app.emit(CONTAINER_EXITED_EVENT, ContainerExited { project: p, reason });
|
||||
})
|
||||
.await;
|
||||
|
||||
@@ -402,7 +402,7 @@ pub(super) fn compute_env_fingerprint(custom_env_vars: &[EnvVar]) -> String {
|
||||
///
|
||||
/// The returned token is never logged and never leaves this module except as
|
||||
/// the env var value handed to Docker.
|
||||
fn shared_claude_auth(project: &Project) -> Option<(String, String)> {
|
||||
pub(super) fn shared_claude_auth(project: &Project) -> Option<(String, String)> {
|
||||
if project.backend != Backend::Anthropic || !project.use_shared_auth_token {
|
||||
return None;
|
||||
}
|
||||
|
||||
@@ -419,6 +419,39 @@ pub(crate) fn workspace_env(
|
||||
env
|
||||
}
|
||||
|
||||
|
||||
/// [`workspace_env`] with its inputs resolved from `settings` exactly as
|
||||
/// `create_container` resolves them (the CA from disk, the shared Claude
|
||||
/// token from the keychain), for a host that does not create a Docker
|
||||
/// container. Also returns the resolved CA, whose files the caller ships.
|
||||
pub(crate) fn workspace_env_from_settings(
|
||||
project: &Project,
|
||||
settings: &crate::models::AppSettings,
|
||||
) -> Result<crate::host::aws_host::AwsEnv, String> {
|
||||
let shared_claude = super::container::shared_claude_auth(project);
|
||||
let effective_ca_path =
|
||||
resolve_with_global(project.ca_cert_path.as_deref(), settings.ca_cert_path.as_deref());
|
||||
let resolved_ca = ca_certs::resolve(effective_ca_path)?;
|
||||
let pairs = workspace_env(
|
||||
project,
|
||||
&WorkspaceEnvInputs {
|
||||
global_aws: &settings.global_aws,
|
||||
global_ollama: &settings.global_ollama,
|
||||
global_llamacpp: &settings.global_llamacpp,
|
||||
global_openai_compatible: &settings.global_openai_compatible,
|
||||
global_claude_instructions: settings.global_claude_instructions.as_deref(),
|
||||
global_custom_env_vars: &settings.global_custom_env_vars,
|
||||
timezone: settings.timezone.as_deref(),
|
||||
global_claude_code_settings: settings.global_claude_code_settings.as_ref(),
|
||||
default_git_user_name: settings.default_git_user_name.as_deref(),
|
||||
default_git_user_email: settings.default_git_user_email.as_deref(),
|
||||
resolved_ca: resolved_ca.as_ref(),
|
||||
shared_claude_token: shared_claude.as_ref().map(|(token, _)| token.as_str()),
|
||||
},
|
||||
);
|
||||
Ok((pairs, resolved_ca))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::ca_certs::ResolvedCa;
|
||||
|
||||
@@ -0,0 +1,833 @@
|
||||
//! `ProjectHost::Aws`: a project's container is an ECS Fargate task in one
|
||||
//! of the account's environments, reached through an SSM port-forward to
|
||||
//! the in-task agent (`ssm.rs`, `agent.rs`).
|
||||
//!
|
||||
//! Start: connection → `SdkConfig` → caller identity → environment record →
|
||||
//! onboarding → the project's live task, if there is one (its token is read
|
||||
//! back from the project secret, P2 — **a running task's token is never
|
||||
//! rotated**); otherwise `prepare_workspace` (which rotates the token) →
|
||||
//! access point available → RunTask → ready → tunnel → agent health →
|
||||
//! registry ([`super::tunnels`]). A task this start ran is stopped again on
|
||||
//! any later failure, so nothing billed is left behind.
|
||||
//!
|
||||
//! The decisions are made by [`start_core`] over the `aws/` traits and a
|
||||
//! [`Connector`], so fakes exercise them; AWS SDK calls stay in `aws/`.
|
||||
|
||||
use std::future::Future;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use aws_config::SdkConfig;
|
||||
use futures_util::future::BoxFuture;
|
||||
|
||||
use super::agent::{self, AgentEndpoint};
|
||||
use super::ssm::{EcsTask, PortForward, AGENT_PORT};
|
||||
use super::tunnels::{self, LiveTunnel, TunnelHandle};
|
||||
use super::watch::{AwsTaskLookup, TaskGroup, WatchedTask};
|
||||
use super::ContainerTarget;
|
||||
use crate::aws::environments::EnvironmentInfo;
|
||||
use crate::aws::runtime::{
|
||||
check_onboarded, cluster_arn_of_task, cluster_of_task, read_agent_token, read_environment,
|
||||
remove_workspace_resources, wait_access_point, PollTiming, RemoveRequest, RuntimeApi,
|
||||
SdkRuntimeApi,
|
||||
};
|
||||
use crate::aws::tasks::{
|
||||
find_or_run, find_project_task, stopped_among, task_log_stream, wait_ready, SdkTaskApi,
|
||||
TaskApi, TaskInfo, TaskLogs, WaitTiming, LOG_TAIL_LINES, STOP_REASON,
|
||||
};
|
||||
use crate::aws::tofu_modules::{log_stream_prefix, owner_key, task_family};
|
||||
use crate::aws::workspace::{prepare_workspace, SdkWorkspaceApi, WorkspaceApi};
|
||||
use crate::aws::{caller_identity, sdk_config, CallerIdentity};
|
||||
use crate::docker::ca_certs::{container_cert_name, ResolvedCa};
|
||||
use crate::models::app_settings::ImageSource;
|
||||
use crate::models::container_config::REGISTRY_IMAGE;
|
||||
use crate::models::{AppSettings, Project, ProjectStatus};
|
||||
use crate::storage::projects_store::ProjectsStore;
|
||||
use crate::storage::settings_store::SettingsStore;
|
||||
|
||||
// Progress strings the start view shows, in order (Task 8 matches on none
|
||||
// of them; they are for the user).
|
||||
pub const PROGRESS_READING: &str = "Reading environment…";
|
||||
pub const PROGRESS_PREPARING: &str = "Preparing workspace…";
|
||||
pub const PROGRESS_STARTING: &str = "Starting task…";
|
||||
pub const PROGRESS_WAITING: &str = "Waiting for the task (about 2 minutes)…";
|
||||
pub const PROGRESS_CONNECTING: &str = "Connecting…";
|
||||
|
||||
/// A concurrent start won the RunTask race: the secret now holds a token
|
||||
/// that task never saw.
|
||||
pub const START_RACE: &str =
|
||||
"Another start of this workspace is in progress; try again in a minute";
|
||||
|
||||
/// Per-file cap on what is shipped in the project secret (R1).
|
||||
pub const MAX_FILE_BYTES: u64 = 64 * 1024;
|
||||
/// How long the agent gets to answer through a fresh tunnel.
|
||||
const HEALTH_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
const HEALTH_INTERVAL: Duration = Duration::from_secs(1);
|
||||
/// How long a terminal waits for a background re-attach.
|
||||
pub const REATTACH_WAIT: Duration = Duration::from_secs(90);
|
||||
|
||||
fn s<E: std::fmt::Display>(e: E) -> String {
|
||||
e.to_string()
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────── inputs
|
||||
|
||||
/// The image an AWS task runs (P5): registry images only.
|
||||
pub fn aws_image(settings: &AppSettings) -> Result<String, String> {
|
||||
match settings.image_source {
|
||||
ImageSource::Registry => Ok(REGISTRY_IMAGE.to_string()),
|
||||
ImageSource::LocalBuild => {
|
||||
log::info!("AWS projects run the registry image; the local build exists only on this machine");
|
||||
Ok(REGISTRY_IMAGE.to_string())
|
||||
}
|
||||
ImageSource::Custom => {
|
||||
let name = settings.custom_image_name.as_deref().unwrap_or("").trim();
|
||||
if is_registry_reference(name) {
|
||||
Ok(name.to_string())
|
||||
} else {
|
||||
Err(format!(
|
||||
"The custom image \"{}\" is not a registry reference, so AWS can't pull it. \
|
||||
Use a name like registry.example.com/team/image:tag (Settings → Docker).",
|
||||
name.escape_debug()
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `host/path[:tag]` or `user/repo[:tag]`: something Fargate can pull. A
|
||||
/// bare name (a local tag) or a loopback registry is not.
|
||||
fn is_registry_reference(name: &str) -> bool {
|
||||
if name.is_empty() || name.len() > 512 || name.chars().any(|c| c.is_whitespace() || c.is_control()) {
|
||||
return false;
|
||||
}
|
||||
let Some((first, rest)) = name.split_once('/') else {
|
||||
return false;
|
||||
};
|
||||
if first.is_empty() || rest.is_empty() || rest.starts_with('/') {
|
||||
return false;
|
||||
}
|
||||
let host = first.split(':').next().unwrap_or("");
|
||||
!(host == "localhost" || host.starts_with("127.") || host == "[::1]")
|
||||
}
|
||||
|
||||
/// [`aws_env`]'s answer: the pairs, and the resolved CA whose files ship too.
|
||||
pub type AwsEnv = (Vec<(String, String)>, Option<ResolvedCa>);
|
||||
|
||||
/// The environment pairs an AWS task gets: [`workspace_env`] with the same
|
||||
/// inputs Docker resolves, and Claude Code's sandbox forced off (R4: Fargate
|
||||
/// support for bubblewrap is unverified). Returns the resolved CA too.
|
||||
///
|
||||
/// [`workspace_env`]: crate::docker::workspace_env::workspace_env
|
||||
pub fn aws_env(project: &Project, settings: &AppSettings) -> Result<AwsEnv, String> {
|
||||
let mut p = project.clone();
|
||||
if p.sandbox_mode_enabled {
|
||||
log::info!(
|
||||
"Project {}: Claude Code's sandbox is turned off on AWS (not yet supported on Fargate)",
|
||||
project.id
|
||||
);
|
||||
p.sandbox_mode_enabled = false;
|
||||
}
|
||||
crate::docker::workspace_env::workspace_env_from_settings(&p, settings)
|
||||
}
|
||||
|
||||
/// One file read for the secret, refusing anything over [`MAX_FILE_BYTES`].
|
||||
fn read_capped(path: &Path, shown: &str) -> Result<Vec<u8>, String> {
|
||||
let len = std::fs::metadata(path)
|
||||
.map_err(|e| format!("Could not read {shown}: {e}"))?
|
||||
.len();
|
||||
if len > MAX_FILE_BYTES {
|
||||
return Err(format!(
|
||||
"{shown} is {len} bytes; files sent to an AWS workspace are limited to {} KiB each",
|
||||
MAX_FILE_BYTES / 1024
|
||||
));
|
||||
}
|
||||
std::fs::read(path).map_err(|e| format!("Could not read {shown}: {e}"))
|
||||
}
|
||||
|
||||
/// The regular files directly in `dir`: subdirectories skipped, and a
|
||||
/// symlink only when it resolves to a file inside `dir`.
|
||||
fn files_in(dir: &Path) -> Result<Vec<(String, PathBuf)>, String> {
|
||||
let root = std::fs::canonicalize(dir)
|
||||
.map_err(|e| format!("Could not read {}: {e}", dir.display()))?;
|
||||
let mut out = Vec::new();
|
||||
let entries =
|
||||
std::fs::read_dir(&root).map_err(|e| format!("Could not read {}: {e}", dir.display()))?;
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
let Some(name) = entry.file_name().to_str().map(str::to_string) else {
|
||||
log::warn!("Skipping {} (not a UTF-8 name)", path.display());
|
||||
continue;
|
||||
};
|
||||
let Ok(ft) = entry.file_type() else { continue };
|
||||
let real = if ft.is_file() {
|
||||
path.clone()
|
||||
} else if ft.is_symlink() {
|
||||
match std::fs::canonicalize(&path) {
|
||||
Ok(t) if t.starts_with(&root) && t.is_file() => t,
|
||||
_ => {
|
||||
log::warn!("Skipping {} (a link out of the folder, or not to a file)", path.display());
|
||||
continue;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
continue;
|
||||
};
|
||||
out.push((name, real));
|
||||
}
|
||||
out.sort();
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// The files the entrypoint writes to `/tmp/.host-ssh` and `/tmp/.host-ca`
|
||||
/// (R1), as `ssh/<file>` / `ca/<file>` pairs: the project's effective SSH
|
||||
/// key folder (or file) and the resolved CA, the same sources Docker mounts.
|
||||
pub fn workspace_files(
|
||||
ssh_path: Option<&str>,
|
||||
ca: Option<&ResolvedCa>,
|
||||
) -> Result<Vec<(String, Vec<u8>)>, String> {
|
||||
let mut out = Vec::new();
|
||||
if let Some(ssh) = ssh_path.map(str::trim).filter(|p| !p.is_empty()) {
|
||||
crate::commands::project_commands::validate_mounted_host_path(
|
||||
"the SSH key folder",
|
||||
None,
|
||||
Some(ssh),
|
||||
)?;
|
||||
let path = Path::new(ssh);
|
||||
if path.is_dir() {
|
||||
for (name, real) in files_in(path)? {
|
||||
let bytes = read_capped(&real, &format!("SSH file {name}"))?;
|
||||
out.push((format!("ssh/{name}"), bytes));
|
||||
}
|
||||
} else if path.is_file() {
|
||||
let name = path
|
||||
.file_name()
|
||||
.and_then(|n| n.to_str())
|
||||
.ok_or_else(|| format!("The SSH key path {ssh} has no usable file name"))?;
|
||||
out.push((format!("ssh/{name}"), read_capped(path, &format!("SSH file {name}"))?));
|
||||
} else {
|
||||
return Err(format!("The SSH key path {ssh} does not exist"));
|
||||
}
|
||||
}
|
||||
if let Some(ca) = ca {
|
||||
crate::commands::project_commands::validate_mounted_host_path(
|
||||
"the CA certificate path",
|
||||
None,
|
||||
Some(&ca.host_path),
|
||||
)?;
|
||||
if ca.is_dir {
|
||||
let allowed: Vec<PathBuf> = ca.cert_files.clone();
|
||||
for (name, real) in files_in(Path::new(&ca.host_path))? {
|
||||
// Only what `collect_cert_files` picked (by extension).
|
||||
let picked = allowed.iter().any(|f| {
|
||||
f.file_name().and_then(|n| n.to_str()) == Some(name.as_str())
|
||||
});
|
||||
if picked {
|
||||
out.push((format!("ca/{name}"), read_capped(&real, &format!("certificate {name}"))?));
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let path = Path::new(&ca.host_path);
|
||||
let name = container_cert_name(
|
||||
&path.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(),
|
||||
);
|
||||
out.push((format!("ca/{name}"), read_capped(path, &format!("certificate {name}"))?));
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────── connector
|
||||
|
||||
/// Opens the tunnel to a ready task and checks its agent answers.
|
||||
pub trait Connector: Send + Sync {
|
||||
fn connect(
|
||||
&self,
|
||||
task: &TaskInfo,
|
||||
token: &str,
|
||||
) -> impl Future<Output = Result<(Box<dyn TunnelHandle>, AgentEndpoint), String>> + Send;
|
||||
}
|
||||
|
||||
/// The real one: an SSM port-forward, then `/v1/health` through it.
|
||||
pub struct SsmConnector {
|
||||
pub sdk: SdkConfig,
|
||||
}
|
||||
|
||||
/// The SSM target of a ready task's `workspace` container.
|
||||
pub fn ecs_task(task: &TaskInfo) -> Result<EcsTask, String> {
|
||||
let cluster = cluster_of_task(&task.arn)
|
||||
.ok_or_else(|| format!("{} is not a task ARN with a cluster", task.arn))?;
|
||||
let task_id = task.arn.rsplit('/').next().unwrap_or_default().to_string();
|
||||
let runtime_id = task
|
||||
.runtime_id
|
||||
.clone()
|
||||
.ok_or_else(|| "The task's container has no runtime id yet".to_string())?;
|
||||
Ok(EcsTask { cluster: cluster.to_string(), task_id, runtime_id })
|
||||
}
|
||||
|
||||
/// Wait for the agent behind `ep` to answer. A refusal (wrong token) is
|
||||
/// final; silence is retried until [`HEALTH_TIMEOUT`].
|
||||
async fn wait_healthy(ep: &AgentEndpoint) -> Result<(), String> {
|
||||
let deadline = tokio::time::Instant::now() + HEALTH_TIMEOUT;
|
||||
loop {
|
||||
match agent::is_running(ep).await {
|
||||
Ok(true) => return Ok(()),
|
||||
Ok(false) => {}
|
||||
Err(e) => return Err(format!("The workspace's agent refused the connection: {e}")),
|
||||
}
|
||||
if tokio::time::Instant::now() + HEALTH_INTERVAL > deadline {
|
||||
return Err(
|
||||
"The workspace's agent did not answer through the tunnel. Check that the image \
|
||||
includes triple-c-agent, then start the project again."
|
||||
.into(),
|
||||
);
|
||||
}
|
||||
tokio::time::sleep(HEALTH_INTERVAL).await;
|
||||
}
|
||||
}
|
||||
|
||||
impl Connector for SsmConnector {
|
||||
async fn connect(
|
||||
&self,
|
||||
task: &TaskInfo,
|
||||
token: &str,
|
||||
) -> Result<(Box<dyn TunnelHandle>, AgentEndpoint), String> {
|
||||
let target = ecs_task(task)?;
|
||||
let forward = PortForward::open(&self.sdk, &target, AGENT_PORT).await?;
|
||||
let ep = forward.endpoint(token);
|
||||
if let Err(e) = wait_healthy(&ep).await {
|
||||
forward.close().await;
|
||||
return Err(e);
|
||||
}
|
||||
Ok((Box::new(forward), ep))
|
||||
}
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────── start
|
||||
|
||||
/// What [`start_core`] needs besides the APIs.
|
||||
pub struct StartInputs<'a> {
|
||||
pub project: &'a Project,
|
||||
pub who: &'a CallerIdentity,
|
||||
pub env_pairs: &'a [(String, String)],
|
||||
pub files: &'a [(String, Vec<u8>)],
|
||||
pub image: &'a str,
|
||||
pub wait: WaitTiming,
|
||||
pub access_point_wait: PollTiming,
|
||||
}
|
||||
|
||||
/// A connected task.
|
||||
pub struct Started {
|
||||
pub env: EnvironmentInfo,
|
||||
pub task: TaskInfo,
|
||||
pub forward: Box<dyn TunnelHandle>,
|
||||
pub endpoint: AgentEndpoint,
|
||||
/// The access point the task mounts (Hybrid). `None` when an existing
|
||||
/// task was attached to: the store's value stands.
|
||||
pub access_point_id: Option<String>,
|
||||
/// This start ran the task (rather than attaching to a running one).
|
||||
pub created: bool,
|
||||
}
|
||||
|
||||
/// Append the task's last log lines to a wait failure that lacks them.
|
||||
async fn with_log_tail<T: TaskApi>(api: &T, logs: &TaskLogs, arn: &str, err: String) -> String {
|
||||
if err.contains("Last log lines") || err.contains("wrote no log lines") || err.contains("log could not be read") {
|
||||
return err;
|
||||
}
|
||||
match api.log_tail(&logs.group, &task_log_stream(&logs.stream_prefix, arn), LOG_TAIL_LINES).await {
|
||||
Ok(lines) if !lines.is_empty() => format!("{err}\nLast log lines:\n{}", lines.join("\n")),
|
||||
_ => err,
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop a task this start ran, best-effort, so nothing billed is orphaned.
|
||||
async fn stop_created<T: TaskApi>(api: &T, cluster: &str, arn: &str) {
|
||||
if let Err(e) = api.stop_task(cluster, arn, STOP_REASON).await {
|
||||
log::warn!("Could not stop task {arn} after a failed start: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
/// The start sequence; see the module docs.
|
||||
pub async fn start_core<T: TaskApi, W: WorkspaceApi, R: RuntimeApi, C: Connector>(
|
||||
tasks: &T,
|
||||
work: &W,
|
||||
rt: &R,
|
||||
connector: &C,
|
||||
inp: &StartInputs<'_>,
|
||||
progress: &(dyn Fn(&str) + Send + Sync),
|
||||
) -> Result<Started, String> {
|
||||
let project = inp.project;
|
||||
let (placement, cfg) = match (&project.aws_placement, &project.aws) {
|
||||
(Some(p), Some(c)) if project.is_aws() => (p, c),
|
||||
_ => return Err("The project is not set up to run on AWS".into()),
|
||||
};
|
||||
let region = placement.region.as_str();
|
||||
let ok = owner_key(region, &inp.who.user_id);
|
||||
|
||||
progress(PROGRESS_READING);
|
||||
let env = read_environment(rt, &placement.environment, &inp.who.account, region)
|
||||
.await
|
||||
.map_err(s)?;
|
||||
check_onboarded(rt, &env.name, &ok).await.map_err(s)?;
|
||||
let family = task_family(&env.name, &ok, &project.id);
|
||||
let logs = TaskLogs {
|
||||
group: env.log_group.clone(),
|
||||
stream_prefix: log_stream_prefix(&ok, &project.id),
|
||||
};
|
||||
|
||||
let existing = find_project_task(tasks, &env.cluster_arn, &family, &inp.who.user_id, &project.id)
|
||||
.await
|
||||
.map_err(s)?;
|
||||
|
||||
let (task, token, access_point_id, created) = if let Some(task) = existing {
|
||||
// Attach: never rotate a running task's token (P2).
|
||||
log::info!("Project {}: attaching to its running task {}", project.id, task.arn);
|
||||
let token = read_agent_token(rt, &env.name, &ok, &project.id).await.map_err(s)?;
|
||||
progress(PROGRESS_WAITING);
|
||||
let task = match wait_ready(tasks, &env.cluster_arn, &task.arn, &logs, inp.wait, |_| {}).await {
|
||||
Ok(t) => t,
|
||||
Err(e) => return Err(with_log_tail(tasks, &logs, &task.arn, e.to_string()).await),
|
||||
};
|
||||
(task, token, None, false)
|
||||
} else {
|
||||
progress(PROGRESS_PREPARING);
|
||||
let prepared = prepare_workspace(
|
||||
work,
|
||||
inp.who,
|
||||
&env,
|
||||
project,
|
||||
inp.env_pairs,
|
||||
inp.files,
|
||||
inp.image,
|
||||
)
|
||||
.await
|
||||
.map_err(s)?;
|
||||
if let Some(ap) = &prepared.access_point_id {
|
||||
wait_access_point(rt, ap, inp.access_point_wait).await.map_err(s)?;
|
||||
}
|
||||
progress(PROGRESS_STARTING);
|
||||
let start = find_or_run(
|
||||
tasks,
|
||||
&env,
|
||||
&inp.who.user_id,
|
||||
&project.id,
|
||||
&prepared.task_definition_arn,
|
||||
cfg.spot,
|
||||
)
|
||||
.await
|
||||
.map_err(s)?;
|
||||
if start.reused {
|
||||
return Err(START_RACE.into());
|
||||
}
|
||||
progress(PROGRESS_WAITING);
|
||||
let arn = start.task.arn.clone();
|
||||
let task = match wait_ready(tasks, &env.cluster_arn, &arn, &logs, inp.wait, |_| {}).await {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
let msg = with_log_tail(tasks, &logs, &arn, e.to_string()).await;
|
||||
stop_created(tasks, &env.cluster_arn, &arn).await;
|
||||
return Err(msg);
|
||||
}
|
||||
};
|
||||
(task, prepared.agent_token, prepared.access_point_id, true)
|
||||
};
|
||||
|
||||
progress(PROGRESS_CONNECTING);
|
||||
match connector.connect(&task, &token).await {
|
||||
Ok((forward, endpoint)) => Ok(Started {
|
||||
env,
|
||||
task,
|
||||
forward,
|
||||
endpoint,
|
||||
access_point_id,
|
||||
created,
|
||||
}),
|
||||
Err(e) => {
|
||||
if created {
|
||||
stop_created(tasks, &env.cluster_arn, &task.arn).await;
|
||||
}
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The connection, region and SDK config of an AWS project.
|
||||
async fn project_sdk(project: &Project, settings: &AppSettings) -> Result<(String, String, SdkConfig), String> {
|
||||
let host_id = project.host_id.as_deref().ok_or("The project has no host")?;
|
||||
let region = project
|
||||
.aws_placement
|
||||
.as_ref()
|
||||
.map(|p| p.region.clone())
|
||||
.ok_or("The project has no AWS placement")?;
|
||||
let conn = crate::commands::host_commands::aws_conn(host_id, settings)?;
|
||||
let sdk = sdk_config(&conn, ®ion).await.map_err(s)?;
|
||||
Ok((host_id.to_string(), region, sdk))
|
||||
}
|
||||
|
||||
/// `ProjectHost::Aws::ensure_started`.
|
||||
pub async fn ensure_started(
|
||||
project: &Project,
|
||||
settings: &AppSettings,
|
||||
progress: &(dyn Fn(&str) + Send + Sync),
|
||||
) -> Result<ContainerTarget, String> {
|
||||
// Already connected and answering: nothing to do.
|
||||
if let Some(ep) = tunnels::get_endpoint(&project.id) {
|
||||
if matches!(agent::is_running(&ep).await, Ok(true)) {
|
||||
return Ok(ContainerTarget::Agent(ep));
|
||||
}
|
||||
tunnels::close(&project.id).await;
|
||||
}
|
||||
|
||||
// Everything local first: a bad image or file fails before any AWS call.
|
||||
let image = aws_image(settings)?;
|
||||
let (env_pairs, ca) = aws_env(project, settings)?;
|
||||
let ssh = project
|
||||
.ssh_key_path
|
||||
.as_deref()
|
||||
.or(settings.default_ssh_key_path.as_deref());
|
||||
let files = workspace_files(ssh, ca.as_ref())?;
|
||||
|
||||
let (connection_id, region, sdk) = project_sdk(project, settings).await?;
|
||||
let who = caller_identity(&sdk).await.map_err(s)?;
|
||||
let started = start_core(
|
||||
&SdkTaskApi::new(&sdk),
|
||||
&SdkWorkspaceApi::new(&sdk),
|
||||
&SdkRuntimeApi::new(&sdk),
|
||||
&SsmConnector { sdk: sdk.clone() },
|
||||
&StartInputs {
|
||||
project,
|
||||
who: &who,
|
||||
env_pairs: &env_pairs,
|
||||
files: &files,
|
||||
image: &image,
|
||||
wait: WaitTiming::default(),
|
||||
access_point_wait: PollTiming::default(),
|
||||
},
|
||||
progress,
|
||||
)
|
||||
.await?;
|
||||
|
||||
log::info!(
|
||||
"Project {}: {} task {}",
|
||||
project.id,
|
||||
if started.created { "started" } else { "attached to" },
|
||||
started.task.arn
|
||||
);
|
||||
let endpoint = started.endpoint.clone();
|
||||
let access_point_id = started
|
||||
.access_point_id
|
||||
.or_else(|| project.aws.as_ref().and_then(|a| a.access_point_id.clone()));
|
||||
let ecs = ecs_task(&started.task)?;
|
||||
if let Some(old) = tunnels::register(
|
||||
&project.id,
|
||||
LiveTunnel {
|
||||
task: ecs,
|
||||
task_arn: started.task.arn.clone(),
|
||||
cluster_arn: started.env.cluster_arn.clone(),
|
||||
connection_id,
|
||||
region,
|
||||
sdk,
|
||||
forward: started.forward,
|
||||
endpoint: endpoint.clone(),
|
||||
access_point_id,
|
||||
task_definition_arn: started.task.task_definition_arn.clone(),
|
||||
},
|
||||
) {
|
||||
old.forward.close().await;
|
||||
}
|
||||
Ok(ContainerTarget::Agent(endpoint))
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────── stop
|
||||
|
||||
/// StopTask the task in `task_arn`'s cluster.
|
||||
pub async fn stop_task_arn<T: TaskApi>(api: &T, task_arn: &str) -> Result<(), String> {
|
||||
let cluster = cluster_of_task(task_arn)
|
||||
.ok_or_else(|| format!("{task_arn} is not a task ARN with a cluster"))?;
|
||||
api.stop_task(cluster, task_arn, STOP_REASON).await.map_err(s)
|
||||
}
|
||||
|
||||
/// `ProjectHost::stop` for an agent target: the task behind a registered
|
||||
/// tunnel. `None` when `ep` is not one of ours.
|
||||
pub async fn stop_endpoint(ep: &AgentEndpoint) -> Option<Result<(), String>> {
|
||||
let (_connection, _region, cluster, arn) = tunnels::task_of(ep)?;
|
||||
let (project_id, sdk) = tunnels::owner_of(ep)?;
|
||||
let res = SdkTaskApi::new(&sdk)
|
||||
.stop_task(&cluster, &arn, STOP_REASON)
|
||||
.await
|
||||
.map_err(s);
|
||||
if res.is_ok() {
|
||||
tunnels::close(&project_id).await;
|
||||
}
|
||||
Some(res)
|
||||
}
|
||||
|
||||
/// Stop an AWS project's task: through its tunnel when there is one, else
|
||||
/// the stored task ARN, else whatever live task the project has (a crash
|
||||
/// between RunTask and the store write). Closes the tunnel.
|
||||
pub async fn stop_project(project: &Project, settings: &AppSettings) -> Result<(), String> {
|
||||
if let Some(ep) = tunnels::get_endpoint(&project.id) {
|
||||
if let Some(res) = stop_endpoint(&ep).await {
|
||||
return res;
|
||||
}
|
||||
}
|
||||
let (_, region, sdk) = project_sdk(project, settings).await?;
|
||||
let api = SdkTaskApi::new(&sdk);
|
||||
let stored = project.aws.as_ref().and_then(|a| a.task_arn.clone());
|
||||
let arn = match stored {
|
||||
Some(a) => Some(a),
|
||||
None => {
|
||||
let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default();
|
||||
let who = caller_identity(&sdk).await.map_err(s)?;
|
||||
let ok = owner_key(®ion, &who.user_id);
|
||||
find_project_task(&api, &format!("triple-c-{env}"), &task_family(&env, &ok, &project.id), &who.user_id, &project.id)
|
||||
.await
|
||||
.map_err(s)?
|
||||
.map(|t| t.arn)
|
||||
}
|
||||
};
|
||||
if let Some(arn) = arn {
|
||||
stop_task_arn(&api, &arn).await?;
|
||||
}
|
||||
tunnels::close(&project.id).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ────────────────────────────────────────────────────────────── remove
|
||||
|
||||
/// Stop the task and remove the project's AWS resources (task definition
|
||||
/// revisions, the secret with a 7-day window, and with `discard_data` the
|
||||
/// access point).
|
||||
pub async fn remove_resources(
|
||||
project: &Project,
|
||||
settings: &AppSettings,
|
||||
discard_data: bool,
|
||||
) -> Result<(), String> {
|
||||
tunnels::close(&project.id).await;
|
||||
let (_, region, sdk) = project_sdk(project, settings).await?;
|
||||
let who = caller_identity(&sdk).await.map_err(s)?;
|
||||
let ok = owner_key(®ion, &who.user_id);
|
||||
let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default();
|
||||
let aws = project.aws.as_ref();
|
||||
remove_workspace_resources(
|
||||
&SdkTaskApi::new(&sdk),
|
||||
&SdkWorkspaceApi::new(&sdk),
|
||||
&SdkRuntimeApi::new(&sdk),
|
||||
&RemoveRequest {
|
||||
env: &env,
|
||||
region: ®ion,
|
||||
owner_user_id: &who.user_id,
|
||||
owner_key: &ok,
|
||||
project_id: &project.id,
|
||||
stored_task_arn: aws.and_then(|a| a.task_arn.as_deref()),
|
||||
stored_access_point_id: aws.and_then(|a| a.access_point_id.as_deref()),
|
||||
discard_data,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.map_err(s)
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────── reconcile
|
||||
|
||||
/// What reconcile found for a Running AWS project.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Reconciled {
|
||||
/// The task is up (or on its way): re-open the tunnel.
|
||||
Live(Box<TaskInfo>),
|
||||
/// The task stopped, or ECS no longer knows it: the project is Stopped.
|
||||
Gone,
|
||||
}
|
||||
|
||||
/// Ask ECS about a project's stored task. No stored task is `Gone`.
|
||||
pub async fn reconcile_task<T: TaskApi>(api: &T, task_arn: Option<&str>) -> Result<Reconciled, String> {
|
||||
let Some(arn) = task_arn else {
|
||||
return Ok(Reconciled::Gone);
|
||||
};
|
||||
let Some(cluster) = cluster_of_task(arn) else {
|
||||
return Ok(Reconciled::Gone);
|
||||
};
|
||||
let found = api.describe(cluster, &[arn.to_string()]).await.map_err(s)?;
|
||||
Ok(match found.into_iter().find(|t| t.arn == arn) {
|
||||
Some(t)
|
||||
if t.desired_status == "RUNNING"
|
||||
&& matches!(t.last_status.as_str(), "PROVISIONING" | "PENDING" | "ACTIVATING" | "RUNNING") =>
|
||||
{
|
||||
Reconciled::Live(Box::new(t))
|
||||
}
|
||||
_ => Reconciled::Gone,
|
||||
})
|
||||
}
|
||||
|
||||
/// Reconcile one AWS project the store has as Running (or Starting/
|
||||
/// Stopping/Error left by a crash): a gone task makes it Stopped (its task
|
||||
/// ARN cleared); a live one keeps it Running. An ECS that can't be asked
|
||||
/// changes nothing — marking Stopped would hide a billed task.
|
||||
pub async fn reconcile_project<T: TaskApi>(
|
||||
api: &T,
|
||||
store: &ProjectsStore,
|
||||
project: &Project,
|
||||
) -> Result<Reconciled, String> {
|
||||
let arn = project.aws.as_ref().and_then(|a| a.task_arn.clone());
|
||||
let r = reconcile_task(api, arn.as_deref()).await?;
|
||||
match &r {
|
||||
Reconciled::Gone => {
|
||||
store.update_status(&project.id, ProjectStatus::Stopped)?;
|
||||
if let Some(a) = project.aws.as_ref() {
|
||||
store.set_aws_runtime(&project.id, None, a.access_point_id.clone(), a.task_definition_arn.clone())?;
|
||||
}
|
||||
}
|
||||
Reconciled::Live(_) => {
|
||||
if project.status != ProjectStatus::Running {
|
||||
store.update_status(&project.id, ProjectStatus::Running)?;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(r)
|
||||
}
|
||||
|
||||
/// `reconcile_project_statuses` for an AWS project: decide, and when the
|
||||
/// task is live re-open its tunnel in the background (terminals wait for
|
||||
/// it through [`tunnels::wait_endpoint`]). Never blocks on the tunnel.
|
||||
pub async fn reconcile(project: &Project, settings: &AppSettings, store: Arc<ProjectsStore>) {
|
||||
let (connection_id, region, sdk) = match project_sdk(project, settings).await {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
log::warn!("Project {}: could not reach AWS to reconcile ({e}); status left as is", project.id);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let api = SdkTaskApi::new(&sdk);
|
||||
match reconcile_project(&api, &store, project).await {
|
||||
Ok(Reconciled::Live(task)) => {
|
||||
tunnels::set_pending(&project.id, true);
|
||||
let project = project.clone();
|
||||
tauri::async_runtime::spawn(async move {
|
||||
if let Err(e) = reattach(&project, *task, connection_id, region, sdk, &store).await {
|
||||
log::warn!("Project {}: could not re-open the tunnel to its task: {e}", project.id);
|
||||
tunnels::set_pending(&project.id, false);
|
||||
}
|
||||
});
|
||||
}
|
||||
Ok(Reconciled::Gone) => {
|
||||
log::info!("Project '{}' ({}): its task is gone — set to Stopped", project.name, project.id)
|
||||
}
|
||||
Err(e) => log::warn!("Project {}: could not check its task ({e}); status left as is", project.id),
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-open the tunnel to a live task found at startup.
|
||||
async fn reattach(
|
||||
project: &Project,
|
||||
task: TaskInfo,
|
||||
connection_id: String,
|
||||
region: String,
|
||||
sdk: SdkConfig,
|
||||
store: &ProjectsStore,
|
||||
) -> Result<(), String> {
|
||||
let env = project.aws_placement.as_ref().map(|p| p.environment.clone()).unwrap_or_default();
|
||||
let who = caller_identity(&sdk).await.map_err(s)?;
|
||||
let ok = owner_key(®ion, &who.user_id);
|
||||
let rt = SdkRuntimeApi::new(&sdk);
|
||||
let api = SdkTaskApi::new(&sdk);
|
||||
let token = read_agent_token(&rt, &env, &ok, &project.id).await.map_err(s)?;
|
||||
let cluster_arn = cluster_arn_of_task(&task.arn).ok_or("bad task ARN")?;
|
||||
let logs = TaskLogs { group: format!("/triple-c/{env}"), stream_prefix: log_stream_prefix(&ok, &project.id) };
|
||||
let task = wait_ready(&api, &cluster_arn, &task.arn, &logs, WaitTiming::default(), |_| {})
|
||||
.await
|
||||
.map_err(s)?;
|
||||
let (forward, endpoint) = SsmConnector { sdk: sdk.clone() }.connect(&task, &token).await?;
|
||||
if let Some(old) = tunnels::register(
|
||||
&project.id,
|
||||
LiveTunnel {
|
||||
task: ecs_task(&task)?,
|
||||
task_arn: task.arn.clone(),
|
||||
cluster_arn,
|
||||
connection_id,
|
||||
region,
|
||||
sdk,
|
||||
forward,
|
||||
endpoint: endpoint.clone(),
|
||||
access_point_id: project.aws.as_ref().and_then(|a| a.access_point_id.clone()),
|
||||
task_definition_arn: task.task_definition_arn.clone(),
|
||||
},
|
||||
) {
|
||||
old.forward.close().await;
|
||||
}
|
||||
// The local port is new, so the stored target id is too.
|
||||
store.set_container_id(&project.id, Some(endpoint.base_url.to_string()))?;
|
||||
log::info!("Project {}: tunnel to task {} re-opened", project.id, task.arn);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────── watch
|
||||
|
||||
/// The AWS death watch's lookup: Running AWS projects from the store, their
|
||||
/// targets from the registry (or, while unattached, an endpoint with the
|
||||
/// stored id so `container_watch::settle` still matches), and
|
||||
/// `stopped_among` per connection/region/cluster.
|
||||
pub struct StoreTaskLookup {
|
||||
pub projects: Arc<ProjectsStore>,
|
||||
pub settings: Arc<SettingsStore>,
|
||||
}
|
||||
|
||||
/// A Running AWS project's watched task, if it has one.
|
||||
pub fn watched_task(p: &Project) -> Option<WatchedTask> {
|
||||
if p.status != ProjectStatus::Running || !p.is_aws() {
|
||||
return None;
|
||||
}
|
||||
let arn = p.aws.as_ref()?.task_arn.clone()?;
|
||||
let target = match tunnels::get_endpoint(&p.id) {
|
||||
Some(ep) => ep,
|
||||
None => AgentEndpoint::new(url::Url::parse(p.container_id.as_deref()?).ok()?, "unattached").ok()?,
|
||||
};
|
||||
if Some(target.base_url.as_str()) != p.container_id.as_deref() {
|
||||
return None;
|
||||
}
|
||||
Some(WatchedTask {
|
||||
target: ContainerTarget::Agent(target),
|
||||
group: TaskGroup {
|
||||
connection_id: p.host_id.clone()?,
|
||||
region: p.aws_placement.as_ref()?.region.clone(),
|
||||
cluster_arn: cluster_arn_of_task(&arn)?,
|
||||
},
|
||||
task_arn: arn,
|
||||
})
|
||||
}
|
||||
|
||||
impl AwsTaskLookup for StoreTaskLookup {
|
||||
fn running_tasks(&self) -> Vec<WatchedTask> {
|
||||
self.projects.list().iter().filter_map(watched_task).collect()
|
||||
}
|
||||
|
||||
fn stopped<'a>(
|
||||
&'a self,
|
||||
group: &'a TaskGroup,
|
||||
arns: &'a [String],
|
||||
) -> BoxFuture<'a, Result<Vec<String>, String>> {
|
||||
Box::pin(async move {
|
||||
let settings = self.settings.get();
|
||||
let conn = crate::commands::host_commands::aws_conn(&group.connection_id, &settings)?;
|
||||
let sdk = sdk_config(&conn, &group.region).await.map_err(s)?;
|
||||
let stopped = stopped_among(&SdkTaskApi::new(&sdk), &group.cluster_arn, arns)
|
||||
.await
|
||||
.map_err(s)?;
|
||||
record_stopped(&self.projects.list(), &stopped);
|
||||
Ok(stopped)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Tell the registry which targets' tasks ECS reports stopped, so
|
||||
/// `has_exited` confirms them (R3).
|
||||
pub fn record_stopped(projects: &[Project], stopped: &[String]) {
|
||||
for w in projects.iter().filter_map(watched_task) {
|
||||
if stopped.contains(&w.task_arn) {
|
||||
tunnels::mark_stopped(w.target.id());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,543 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use aws_sdk_ecs::operation::register_task_definition::RegisterTaskDefinitionInput;
|
||||
use aws_sdk_ecs::operation::run_task::RunTaskInput;
|
||||
|
||||
use super::*;
|
||||
use crate::aws::environments::Knowable;
|
||||
use crate::aws::tasks::TaskInfo;
|
||||
use crate::aws::workspace::{AccessPointRef, FileSystemRef, TaskDefinitionRef};
|
||||
use crate::aws::AwsError;
|
||||
use crate::host::lifecycle::ProjectHost;
|
||||
use crate::host::tunnels::tests::{fake_tunnel, FakeTunnel};
|
||||
use crate::models::{AwsPlacement, AwsWorkspaceConfig, StorageMode};
|
||||
|
||||
const REGION: &str = "eu-west-1";
|
||||
const ACCOUNT: &str = "111122223333";
|
||||
const USER: &str = "AROAEXAMPLEID:alice@example.com";
|
||||
|
||||
fn who() -> CallerIdentity {
|
||||
CallerIdentity {
|
||||
account: ACCOUNT.into(),
|
||||
arn: format!("arn:aws:sts::{ACCOUNT}:assumed-role/dev/alice"),
|
||||
user_id: USER.into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn aws_project(storage: StorageMode) -> Project {
|
||||
let mut p = Project::new("aws".into(), Vec::new());
|
||||
p.host_id = Some("c1".into());
|
||||
p.aws_placement = Some(AwsPlacement { region: REGION.into(), environment: "dev".into() });
|
||||
p.aws = Some(AwsWorkspaceConfig { storage, ..Default::default() });
|
||||
p
|
||||
}
|
||||
|
||||
fn family(p: &Project) -> String {
|
||||
task_family("dev", &owner_key(REGION, USER), &p.id)
|
||||
}
|
||||
|
||||
fn task_arn(id: &str) -> String {
|
||||
format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task/triple-c-dev/{id}")
|
||||
}
|
||||
|
||||
fn cluster_arn() -> String {
|
||||
format!("arn:aws:ecs:{REGION}:{ACCOUNT}:cluster/triple-c-dev")
|
||||
}
|
||||
|
||||
fn task(p: &Project, id: &str, last: &str, ready: bool) -> TaskInfo {
|
||||
TaskInfo {
|
||||
arn: task_arn(id),
|
||||
last_status: last.into(),
|
||||
desired_status: if last == "STOPPED" { "STOPPED" } else { "RUNNING" }.into(),
|
||||
health: None,
|
||||
stop_code: (last == "STOPPED").then(|| "EssentialContainerExited".into()),
|
||||
stopped_reason: (last == "STOPPED").then(|| "Essential container exited".into()),
|
||||
task_definition_arn: format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:1", family(p)),
|
||||
runtime_id: ready.then(|| "rt-1".into()),
|
||||
exec_agent_running: ready,
|
||||
owner: Some(USER.into()),
|
||||
project: Some(p.id.clone()),
|
||||
}
|
||||
}
|
||||
|
||||
fn record() -> String {
|
||||
serde_json::json!({
|
||||
"schema": 1, "name": "dev", "template_version": 1,
|
||||
"cluster_arn": cluster_arn(),
|
||||
"security_group_id": "sg-0123456789abcdef0",
|
||||
"subnet_ids": ["subnet-0a", "subnet-0b"],
|
||||
"assign_public_ip": true,
|
||||
"kms_key_arn": format!("arn:aws:kms:{REGION}:{ACCOUNT}:key/1234-abcd"),
|
||||
"log_group": "/triple-c/dev", "network_mode": "public", "tenancy": "strict"
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// One fake for all three AWS traits and the connector, logging calls.
|
||||
#[derive(Default)]
|
||||
struct Fake {
|
||||
calls: Mutex<Vec<String>>,
|
||||
/// ListTasks answers in turn; the last repeats.
|
||||
listed: Mutex<Vec<Vec<String>>>,
|
||||
tasks: Mutex<HashMap<String, TaskInfo>>,
|
||||
/// What RunTask starts.
|
||||
run: Option<TaskInfo>,
|
||||
logs: Vec<String>,
|
||||
role: Option<Knowable>,
|
||||
connect_fails: bool,
|
||||
connect_tokens: Mutex<Vec<String>>,
|
||||
secret_json: Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
fn log(&self, s: impl Into<String>) {
|
||||
self.calls.lock().unwrap().push(s.into());
|
||||
}
|
||||
fn calls(&self) -> Vec<String> {
|
||||
self.calls.lock().unwrap().clone()
|
||||
}
|
||||
fn called(&self, prefix: &str) -> bool {
|
||||
self.calls().iter().any(|c| c.starts_with(prefix))
|
||||
}
|
||||
fn with_task(self, t: TaskInfo) -> Self {
|
||||
self.tasks.lock().unwrap().insert(t.arn.clone(), t);
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
impl TaskApi for Fake {
|
||||
async fn run_task(&self, _req: RunTaskInput) -> Result<TaskInfo, AwsError> {
|
||||
self.log("run");
|
||||
let t = self.run.clone().expect("a run was expected");
|
||||
self.tasks.lock().unwrap().insert(t.arn.clone(), t.clone());
|
||||
Ok(t)
|
||||
}
|
||||
async fn describe(&self, _c: &str, arns: &[String]) -> Result<Vec<TaskInfo>, AwsError> {
|
||||
let t = self.tasks.lock().unwrap();
|
||||
Ok(arns.iter().filter_map(|a| t.get(a).cloned()).collect())
|
||||
}
|
||||
async fn list_project_tasks(&self, _c: &str, _f: &str) -> Result<Vec<String>, AwsError> {
|
||||
self.log("list");
|
||||
let mut l = self.listed.lock().unwrap();
|
||||
Ok(if l.len() > 1 { l.remove(0) } else { l.first().cloned().unwrap_or_default() })
|
||||
}
|
||||
async fn stop_task(&self, _c: &str, arn: &str, reason: &str) -> Result<(), AwsError> {
|
||||
self.log(format!("stop {arn} {reason}"));
|
||||
Ok(())
|
||||
}
|
||||
async fn log_tail(&self, _g: &str, _s: &str, _l: i32) -> Result<Vec<String>, AwsError> {
|
||||
Ok(self.logs.clone())
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkspaceApi for Fake {
|
||||
async fn find_file_system(&self, _t: &str) -> Result<Option<FileSystemRef>, AwsError> {
|
||||
Ok(Some(FileSystemRef { id: "fs-1".into(), owner: Some(USER.into()) }))
|
||||
}
|
||||
async fn access_points(&self, _fs: &str) -> Result<Vec<AccessPointRef>, AwsError> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
async fn create_access_point(
|
||||
&self,
|
||||
_fs: &str,
|
||||
_root: &str,
|
||||
_tags: &[(String, String)],
|
||||
) -> Result<String, AwsError> {
|
||||
self.log("create-ap");
|
||||
Ok("fsap-new".into())
|
||||
}
|
||||
async fn put_project_secret(
|
||||
&self,
|
||||
_name: &str,
|
||||
_kms: &str,
|
||||
json: &str,
|
||||
_tags: &[(String, String)],
|
||||
) -> Result<String, AwsError> {
|
||||
self.log("put-secret");
|
||||
*self.secret_json.lock().unwrap() = Some(json.into());
|
||||
Ok(format!("arn:aws:secretsmanager:{REGION}:{ACCOUNT}:secret:triple-c/dev/x-AbCdEf"))
|
||||
}
|
||||
async fn task_definition_revisions(&self, _f: &str) -> Result<Vec<String>, AwsError> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
async fn describe_task_definition(
|
||||
&self,
|
||||
_arn: &str,
|
||||
) -> Result<Option<TaskDefinitionRef>, AwsError> {
|
||||
Ok(None)
|
||||
}
|
||||
async fn register_task_definition(
|
||||
&self,
|
||||
req: RegisterTaskDefinitionInput,
|
||||
) -> Result<String, AwsError> {
|
||||
self.log("register");
|
||||
Ok(format!(
|
||||
"arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:1",
|
||||
req.family().unwrap()
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeApi for Fake {
|
||||
async fn environment_record(&self, env: &str) -> Result<Option<String>, AwsError> {
|
||||
self.log(format!("record {env}"));
|
||||
Ok(Some(record()))
|
||||
}
|
||||
async fn role_exists(&self, _name: &str) -> Result<Knowable, AwsError> {
|
||||
self.log("role");
|
||||
Ok(self.role.clone().unwrap_or(Knowable::Yes))
|
||||
}
|
||||
async fn secret_string(&self, _name: &str) -> Result<String, AwsError> {
|
||||
self.log("get-secret");
|
||||
Ok(r#"{"TRIPLEC_AGENT_TOKEN":"token-of-the-running-task"}"#.into())
|
||||
}
|
||||
async fn access_point_state(&self, id: &str) -> Result<Option<String>, AwsError> {
|
||||
self.log(format!("ap-state {id}"));
|
||||
Ok(Some("available".into()))
|
||||
}
|
||||
async fn deregister_task_definition(&self, _arn: &str) -> Result<(), AwsError> {
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_task_definitions(&self, _arns: &[String]) -> Result<(), AwsError> {
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_secret(&self, _name: &str) -> Result<(), AwsError> {
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_access_point(&self, _id: &str) -> Result<(), AwsError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Connector for Fake {
|
||||
async fn connect(
|
||||
&self,
|
||||
_task: &TaskInfo,
|
||||
token: &str,
|
||||
) -> Result<(Box<dyn TunnelHandle>, AgentEndpoint), String> {
|
||||
self.log("connect");
|
||||
self.connect_tokens.lock().unwrap().push(token.into());
|
||||
if self.connect_fails {
|
||||
return Err("session-manager-plugin was not found".into());
|
||||
}
|
||||
Ok((
|
||||
Box::new(FakeTunnel(std::sync::Arc::new(AtomicUsize::new(0)))),
|
||||
AgentEndpoint::loopback(61100, token),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
fn inputs<'a>(p: &'a Project, who: &'a CallerIdentity) -> StartInputs<'a> {
|
||||
StartInputs {
|
||||
project: p,
|
||||
who,
|
||||
env_pairs: &[],
|
||||
files: &[],
|
||||
image: REGISTRY_IMAGE,
|
||||
wait: WaitTiming {
|
||||
interval: Duration::from_millis(1),
|
||||
timeout: Duration::from_millis(30),
|
||||
},
|
||||
access_point_wait: PollTiming {
|
||||
interval: Duration::from_millis(1),
|
||||
timeout: Duration::from_millis(30),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
async fn start(fake: &Fake, p: &Project) -> (Result<Started, String>, Vec<String>) {
|
||||
let w = who();
|
||||
let steps = Mutex::new(Vec::new());
|
||||
let progress = |m: &str| steps.lock().unwrap().push(m.to_string());
|
||||
let r = start_core(fake, fake, fake, fake, &inputs(p, &w), &progress).await;
|
||||
(r, steps.into_inner().unwrap())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_first_start_prepares_runs_waits_and_connects_in_that_order() {
|
||||
let p = aws_project(StorageMode::Hybrid);
|
||||
let fake = Fake { run: Some(task(&p, "new", "RUNNING", true)), ..Default::default() };
|
||||
let (r, steps) = start(&fake, &p).await;
|
||||
let started = r.unwrap();
|
||||
assert!(started.created);
|
||||
assert_eq!(started.access_point_id.as_deref(), Some("fsap-new"));
|
||||
assert_eq!(started.task.arn, task_arn("new"));
|
||||
assert_eq!(
|
||||
fake.calls(),
|
||||
vec![
|
||||
"record dev", "role", "list", "create-ap", "put-secret", "register",
|
||||
"ap-state fsap-new", "list", "run", "connect"
|
||||
],
|
||||
"look for a live task before rotating the token; the access point is available before RunTask"
|
||||
);
|
||||
assert_eq!(steps, vec![PROGRESS_READING, PROGRESS_PREPARING, PROGRESS_STARTING, PROGRESS_WAITING, PROGRESS_CONNECTING]);
|
||||
// The connector got the token the secret was just written with.
|
||||
let secret: serde_json::Value =
|
||||
serde_json::from_str(fake.secret_json.lock().unwrap().as_deref().unwrap()).unwrap();
|
||||
assert_eq!(
|
||||
fake.connect_tokens.lock().unwrap()[0],
|
||||
secret["TRIPLEC_AGENT_TOKEN"].as_str().unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_running_task_is_attached_with_its_own_token_and_nothing_is_rewritten() {
|
||||
let p = aws_project(StorageMode::Hybrid);
|
||||
let live = task(&p, "live", "RUNNING", true);
|
||||
let fake = Fake {
|
||||
listed: Mutex::new(vec![vec![live.arn.clone()]]),
|
||||
..Default::default()
|
||||
}
|
||||
.with_task(live);
|
||||
let (r, steps) = start(&fake, &p).await;
|
||||
let started = r.unwrap();
|
||||
assert!(!started.created);
|
||||
assert_eq!(started.access_point_id, None);
|
||||
assert_eq!(fake.calls(), vec!["record dev", "role", "list", "get-secret", "connect"]);
|
||||
assert_eq!(fake.connect_tokens.lock().unwrap()[0], "token-of-the-running-task");
|
||||
assert_eq!(steps, vec![PROGRESS_READING, PROGRESS_WAITING, PROGRESS_CONNECTING]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn losing_the_run_race_is_a_clear_error_and_nothing_is_stopped() {
|
||||
let p = aws_project(StorageMode::Ephemeral);
|
||||
let other = task(&p, "other", "RUNNING", true);
|
||||
let fake = Fake {
|
||||
// Nothing at the first look; a concurrent start's task at the second.
|
||||
listed: Mutex::new(vec![vec![], vec![other.arn.clone()]]),
|
||||
..Default::default()
|
||||
}
|
||||
.with_task(other);
|
||||
let (r, _) = start(&fake, &p).await;
|
||||
assert_eq!(r.err().as_deref(), Some(START_RACE));
|
||||
assert!(!fake.called("run") && !fake.called("stop") && !fake.called("connect"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_task_that_stops_while_starting_is_reported_with_its_log() {
|
||||
let p = aws_project(StorageMode::Ephemeral);
|
||||
let fake = Fake {
|
||||
run: Some(task(&p, "dies", "STOPPED", false)),
|
||||
logs: vec!["entrypoint: cannot decode TRIPLEC_FILES".into()],
|
||||
..Default::default()
|
||||
};
|
||||
let (r, _) = start(&fake, &p).await;
|
||||
let e = r.err().unwrap();
|
||||
assert!(e.contains("Essential container exited") && e.contains("cannot decode"), "{e}");
|
||||
assert!(!fake.called("connect"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_task_that_never_gets_ready_is_stopped_and_its_log_shown() {
|
||||
let p = aws_project(StorageMode::Ephemeral);
|
||||
let fake = Fake {
|
||||
run: Some(task(&p, "slow", "PENDING", false)),
|
||||
logs: vec!["pulling image".into()],
|
||||
..Default::default()
|
||||
};
|
||||
let (r, _) = start(&fake, &p).await;
|
||||
let e = r.err().unwrap();
|
||||
assert!(e.contains("did not become ready") && e.contains("pulling image"), "{e}");
|
||||
assert!(fake.calls().contains(&format!("stop {} {STOP_REASON}", task_arn("slow"))));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_tunnel_failure_stops_a_task_this_start_ran_but_never_an_attached_one() {
|
||||
let p = aws_project(StorageMode::Ephemeral);
|
||||
let fake = Fake {
|
||||
run: Some(task(&p, "new", "RUNNING", true)),
|
||||
connect_fails: true,
|
||||
..Default::default()
|
||||
};
|
||||
let (r, _) = start(&fake, &p).await;
|
||||
assert!(r.err().unwrap().contains("session-manager-plugin"));
|
||||
assert!(fake.calls().contains(&format!("stop {} {STOP_REASON}", task_arn("new"))));
|
||||
|
||||
let live = task(&p, "live", "RUNNING", true);
|
||||
let attached = Fake {
|
||||
listed: Mutex::new(vec![vec![live.arn.clone()]]),
|
||||
connect_fails: true,
|
||||
..Default::default()
|
||||
}
|
||||
.with_task(live);
|
||||
let (r, _) = start(&attached, &p).await;
|
||||
assert!(r.is_err());
|
||||
assert!(!attached.called("stop"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_user_who_has_not_joined_is_told_before_anything_is_written() {
|
||||
let p = aws_project(StorageMode::Hybrid);
|
||||
let fake = Fake { role: Some(Knowable::No), ..Default::default() };
|
||||
let (r, _) = start(&fake, &p).await;
|
||||
assert!(r.err().unwrap().contains("not joined"));
|
||||
assert_eq!(fake.calls(), vec!["record dev", "role"]);
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────── inputs
|
||||
|
||||
#[test]
|
||||
fn aws_runs_registry_images_only() {
|
||||
let mut s = AppSettings::default();
|
||||
s.image_source = ImageSource::LocalBuild;
|
||||
assert_eq!(aws_image(&s).unwrap(), REGISTRY_IMAGE);
|
||||
s.image_source = ImageSource::Registry;
|
||||
assert_eq!(aws_image(&s).unwrap(), REGISTRY_IMAGE);
|
||||
s.image_source = ImageSource::Custom;
|
||||
for good in ["registry.example.com/team/img:1", "ghcr.io/x/y", "user/repo:tag", "reg:5000/a"] {
|
||||
s.custom_image_name = Some(good.into());
|
||||
assert_eq!(aws_image(&s).unwrap(), good);
|
||||
}
|
||||
for bad in ["", "triple-c:latest", "ubuntu", "localhost/x", "127.0.0.1:5000/x", "a /b", "/x"] {
|
||||
s.custom_image_name = Some(bad.into());
|
||||
assert!(aws_image(&s).is_err(), "{bad}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_sandbox_is_forced_off_on_aws() {
|
||||
let mut p = aws_project(StorageMode::Ephemeral);
|
||||
p.sandbox_mode_enabled = true;
|
||||
let (pairs, _) = aws_env(&p, &AppSettings::default()).unwrap();
|
||||
let settings = &pairs.iter().find(|(k, _)| k == "CLAUDE_CODE_SETTINGS_JSON").unwrap().1;
|
||||
let v: serde_json::Value = serde_json::from_str(settings).unwrap();
|
||||
assert_eq!(v["sandbox"]["enabled"], false, "{settings}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn files_for_the_secret_come_from_the_ssh_folder_without_leaving_it() {
|
||||
let base = tempfile::tempdir().unwrap();
|
||||
let ssh = base.path().join("ssh");
|
||||
std::fs::create_dir(&ssh).unwrap();
|
||||
std::fs::write(ssh.join("id_ed25519"), b"KEY").unwrap();
|
||||
std::fs::write(ssh.join("config"), b"Host *").unwrap();
|
||||
std::fs::create_dir(ssh.join("sub")).unwrap();
|
||||
std::fs::write(ssh.join("sub/nested"), b"x").unwrap();
|
||||
std::fs::write(base.path().join("outside"), b"SECRET-OUTSIDE").unwrap();
|
||||
#[cfg(unix)]
|
||||
{
|
||||
std::os::unix::fs::symlink(base.path().join("outside"), ssh.join("escape")).unwrap();
|
||||
std::os::unix::fs::symlink(ssh.join("config"), ssh.join("alias")).unwrap();
|
||||
}
|
||||
let got = workspace_files(Some(ssh.to_str().unwrap()), None).unwrap();
|
||||
let names: Vec<&str> = got.iter().map(|(k, _)| k.as_str()).collect();
|
||||
#[cfg(unix)]
|
||||
assert_eq!(names, vec!["ssh/alias", "ssh/config", "ssh/id_ed25519"]);
|
||||
#[cfg(not(unix))]
|
||||
assert_eq!(names, vec!["ssh/config", "ssh/id_ed25519"]);
|
||||
assert!(got.iter().all(|(_, v)| v != b"SECRET-OUTSIDE"));
|
||||
|
||||
std::fs::write(ssh.join("huge"), vec![b'x'; MAX_FILE_BYTES as usize + 1]).unwrap();
|
||||
let e = workspace_files(Some(ssh.to_str().unwrap()), None).unwrap_err();
|
||||
assert!(e.contains("huge") && e.contains("64 KiB"), "{e}");
|
||||
|
||||
assert!(workspace_files(Some("/"), None).is_err(), "a filesystem root is refused");
|
||||
assert!(workspace_files(None, None).unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ca_files_are_what_docker_would_mount() {
|
||||
let base = tempfile::tempdir().unwrap();
|
||||
let dir = base.path().join("ca");
|
||||
std::fs::create_dir(&dir).unwrap();
|
||||
std::fs::write(dir.join("corp.pem"), b"PEM").unwrap();
|
||||
std::fs::write(dir.join("README"), b"not a cert").unwrap();
|
||||
let ca = crate::docker::ca_certs::resolve(Some(dir.to_str().unwrap())).unwrap().unwrap();
|
||||
let got = workspace_files(None, Some(&ca)).unwrap();
|
||||
assert_eq!(got, vec![("ca/corp.pem".to_string(), b"PEM".to_vec())]);
|
||||
|
||||
let one = base.path().join("Root CA.pem");
|
||||
std::fs::write(&one, b"ONE").unwrap();
|
||||
let ca = crate::docker::ca_certs::resolve(Some(one.to_str().unwrap())).unwrap().unwrap();
|
||||
let got = workspace_files(None, Some(&ca)).unwrap();
|
||||
assert_eq!(got, vec![("ca/Root_CA.crt".to_string(), b"ONE".to_vec())]);
|
||||
}
|
||||
|
||||
// ───────────────────────────────────────────────── targets and reconcile
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_aws_projects_target_is_its_registered_tunnel() {
|
||||
let mut p = aws_project(StorageMode::Ephemeral);
|
||||
p.container_id = Some("http://127.0.0.1:61200/".into());
|
||||
assert_eq!(ProjectHost::for_project(&p), ProjectHost::Aws);
|
||||
assert_eq!(ContainerTarget::for_project(&p), None, "no tunnel, no target, whatever is stored");
|
||||
tunnels::register(&p.id, fake_tunnel(61200, std::sync::Arc::new(AtomicUsize::new(0))));
|
||||
assert_eq!(
|
||||
ContainerTarget::for_project(&p),
|
||||
Some(ContainerTarget::Agent(AgentEndpoint::loopback(61200, "tok")))
|
||||
);
|
||||
assert_eq!(
|
||||
ContainerTarget::for_project_waiting(&p).await,
|
||||
ContainerTarget::for_project(&p)
|
||||
);
|
||||
tunnels::close(&p.id).await;
|
||||
|
||||
let local = Project::new("local".into(), Vec::new());
|
||||
assert_eq!(ProjectHost::for_project(&local), ProjectHost::Docker);
|
||||
}
|
||||
|
||||
fn store_with(p: &Project) -> (tempfile::TempDir, ProjectsStore) {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let store = ProjectsStore::over_file_for_test(dir.path().join("projects.json"), vec![p.clone()]);
|
||||
(dir, store)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reconcile_marks_a_project_whose_task_stopped_as_stopped() {
|
||||
let mut p = aws_project(StorageMode::Hybrid);
|
||||
p.status = ProjectStatus::Running;
|
||||
{
|
||||
let a = p.aws.as_mut().unwrap();
|
||||
a.task_arn = Some(task_arn("gone"));
|
||||
a.access_point_id = Some("fsap-1".into());
|
||||
a.task_definition_arn = Some("arn:td:1".into());
|
||||
}
|
||||
let fake = Fake::default().with_task(task(&p, "gone", "STOPPED", false));
|
||||
let (_d, store) = store_with(&p);
|
||||
assert_eq!(reconcile_project(&fake, &store, &p).await.unwrap(), Reconciled::Gone);
|
||||
let after = store.get(&p.id).unwrap();
|
||||
assert_eq!(after.status, ProjectStatus::Stopped);
|
||||
let a = after.aws.unwrap();
|
||||
assert_eq!(a.task_arn, None, "a stopped task's ARN is cleared");
|
||||
assert_eq!(a.access_point_id.as_deref(), Some("fsap-1"), "the data's access point stays");
|
||||
|
||||
// ECS has forgotten it: also gone.
|
||||
let forgotten = Fake::default();
|
||||
let (_d, store) = store_with(&p);
|
||||
assert_eq!(reconcile_project(&forgotten, &store, &p).await.unwrap(), Reconciled::Gone);
|
||||
assert_eq!(store.get(&p.id).unwrap().status, ProjectStatus::Stopped);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reconcile_keeps_a_project_with_a_live_task_running() {
|
||||
let mut p = aws_project(StorageMode::Hybrid);
|
||||
p.status = ProjectStatus::Running;
|
||||
p.aws.as_mut().unwrap().task_arn = Some(task_arn("up"));
|
||||
let fake = Fake::default().with_task(task(&p, "up", "RUNNING", true));
|
||||
let (_d, store) = store_with(&p);
|
||||
assert!(matches!(reconcile_project(&fake, &store, &p).await.unwrap(), Reconciled::Live(_)));
|
||||
assert_eq!(store.get(&p.id).unwrap().status, ProjectStatus::Running);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_watch_reports_a_running_aws_project_as_its_stored_target_and_confirms_a_recorded_stop() {
|
||||
let mut p = aws_project(StorageMode::Ephemeral);
|
||||
p.status = ProjectStatus::Running;
|
||||
p.container_id = Some("http://127.0.0.1:61300/".into());
|
||||
p.aws.as_mut().unwrap().task_arn = Some(task_arn("w1"));
|
||||
let w = watched_task(&p).expect("watched");
|
||||
assert_eq!(w.target.id(), "http://127.0.0.1:61300/", "settle's running_on matches the stored id");
|
||||
assert_eq!(w.group.cluster_arn, cluster_arn());
|
||||
assert_eq!(w.group.connection_id, "c1");
|
||||
assert!(!w.target.has_exited().await, "nothing recorded yet");
|
||||
record_stopped(std::slice::from_ref(&p), &[task_arn("w1")]);
|
||||
assert!(w.target.has_exited().await);
|
||||
|
||||
let mut stopped = p.clone();
|
||||
stopped.status = ProjectStatus::Stopped;
|
||||
assert!(watched_task(&stopped).is_none(), "a Stopped project is not polled again");
|
||||
let local = Project::new("l".into(), Vec::new());
|
||||
assert!(watched_task(&local).is_none());
|
||||
}
|
||||
@@ -18,11 +18,13 @@ use crate::docker;
|
||||
use crate::models::{AppSettings, Project};
|
||||
use crate::models::container_config;
|
||||
|
||||
/// The host a project runs on. One variant today; an AWS variant will carry
|
||||
/// the connection, region and environment (see the spec's data model).
|
||||
/// The host a project runs on: the local Docker daemon, or an ECS Fargate
|
||||
/// task in an AWS environment (`aws_host.rs`; the connection, region and
|
||||
/// environment come from the project).
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum ProjectHost {
|
||||
Docker,
|
||||
Aws,
|
||||
}
|
||||
|
||||
/// An agent target is a way *into* a container, not a handle on its
|
||||
@@ -41,8 +43,12 @@ pub struct DiscardLeftovers {
|
||||
}
|
||||
|
||||
impl ProjectHost {
|
||||
pub fn for_project(_project: &Project) -> Self {
|
||||
ProjectHost::Docker
|
||||
pub fn for_project(project: &Project) -> Self {
|
||||
if project.is_aws() {
|
||||
ProjectHost::Aws
|
||||
} else {
|
||||
ProjectHost::Docker
|
||||
}
|
||||
}
|
||||
|
||||
/// Make sure the project's container exists, matches the project's
|
||||
@@ -60,6 +66,7 @@ impl ProjectHost {
|
||||
ProjectHost::Docker => docker_ensure_started(project, settings, progress)
|
||||
.await
|
||||
.map(ContainerTarget::docker),
|
||||
ProjectHost::Aws => super::aws_host::ensure_started(project, settings, progress).await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,24 +79,66 @@ impl ProjectHost {
|
||||
ProjectHost::Docker => Ok(docker::find_existing_container(project)
|
||||
.await?
|
||||
.map(ContainerTarget::docker)),
|
||||
// The tunnel registry: an AWS project's target exists while its
|
||||
// tunnel is open.
|
||||
ProjectHost::Aws => Ok(ContainerTarget::for_project(project)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop the container behind `target`. For an agent target that is the
|
||||
/// ECS task behind a registered tunnel (`StopTask`, reason "Stopped from
|
||||
/// Triple-C"), and the tunnel is closed.
|
||||
pub async fn stop(&self, target: &ContainerTarget) -> Result<(), String> {
|
||||
match target {
|
||||
ContainerTarget::Docker { container_id } => docker::stop_container(container_id).await,
|
||||
ContainerTarget::Agent(_) => Err(AGENT_LIFECYCLE.replace("{op}", "Stopping")),
|
||||
ContainerTarget::Agent(ep) => super::aws_host::stop_endpoint(ep)
|
||||
.await
|
||||
.unwrap_or_else(|| Err(AGENT_LIFECYCLE.replace("{op}", "Stopping"))),
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop a project's container whether or not the app holds a target for
|
||||
/// it. Docker: the stored container. AWS: the tunnel's task, else the
|
||||
/// stored task ARN, else the project's live task — a project whose
|
||||
/// tunnel could not be re-opened must still be stoppable.
|
||||
pub async fn stop_project(&self, project: &Project, settings: &AppSettings) -> Result<(), String> {
|
||||
match self {
|
||||
ProjectHost::Docker => match ContainerTarget::for_project(project) {
|
||||
Some(t) => self.stop(&t).await,
|
||||
None => Ok(()),
|
||||
},
|
||||
ProjectHost::Aws => super::aws_host::stop_project(project, settings).await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop (errors ignored — a stopped container is the goal) and remove.
|
||||
/// An agent target is stopped; its resources go with
|
||||
/// [`Self::remove_aws_resources`].
|
||||
pub async fn remove(&self, target: &ContainerTarget) -> Result<(), String> {
|
||||
match target {
|
||||
ContainerTarget::Docker { container_id } => {
|
||||
let _ = docker::stop_container(container_id).await;
|
||||
docker::remove_container(container_id).await
|
||||
}
|
||||
ContainerTarget::Agent(_) => Err(AGENT_LIFECYCLE.replace("{op}", "Removing")),
|
||||
ContainerTarget::Agent(ep) => super::aws_host::stop_endpoint(ep)
|
||||
.await
|
||||
.unwrap_or_else(|| Err(AGENT_LIFECYCLE.replace("{op}", "Removing"))),
|
||||
}
|
||||
}
|
||||
|
||||
/// AWS: stop the task and remove the project's resources — task
|
||||
/// definition revisions, the secret (7-day recovery window) and, only
|
||||
/// with `discard_data`, the EFS access point. Docker: nothing (its state
|
||||
/// goes with [`Self::discard_project_state`]).
|
||||
pub async fn remove_aws_resources(
|
||||
&self,
|
||||
project: &Project,
|
||||
settings: &AppSettings,
|
||||
discard_data: bool,
|
||||
) -> Result<(), String> {
|
||||
match self {
|
||||
ProjectHost::Docker => Ok(()),
|
||||
ProjectHost::Aws => super::aws_host::remove_resources(project, settings, discard_data).await,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -107,6 +156,8 @@ impl ProjectHost {
|
||||
left.volumes = docker::remove_project_volumes(project).await;
|
||||
left
|
||||
}
|
||||
// Removed by `remove_aws_resources`, which needs the settings.
|
||||
ProjectHost::Aws => DiscardLeftovers::default(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -118,6 +169,7 @@ impl ProjectHost {
|
||||
docker::remove_legacy_mcp_containers(&project.id).await;
|
||||
docker::remove_legacy_project_network(&project.id).await;
|
||||
}
|
||||
ProjectHost::Aws => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,12 +28,11 @@
|
||||
//! may still do so.
|
||||
|
||||
pub mod agent;
|
||||
pub mod aws_host;
|
||||
pub mod lifecycle;
|
||||
pub mod sessions;
|
||||
// Nothing in the app opens an SSM tunnel until the AWS host lands (phase 5 of
|
||||
// the AWS spec); its tests exercise it meanwhile. Remove this allow then.
|
||||
#[allow(dead_code)]
|
||||
pub mod ssm;
|
||||
pub mod tunnels;
|
||||
pub mod watch;
|
||||
|
||||
pub use crate::docker::exec::{
|
||||
@@ -154,11 +153,28 @@ impl ContainerTarget {
|
||||
}
|
||||
}
|
||||
|
||||
/// The target for a project's container, if it has one yet.
|
||||
/// The target for a project's container, if it has one yet. An AWS
|
||||
/// project's is its registered tunnel's agent endpoint (`tunnels.rs`):
|
||||
/// none while no tunnel is open.
|
||||
pub fn for_project(project: &Project) -> Option<Self> {
|
||||
if project.is_aws() {
|
||||
return tunnels::get_endpoint(&project.id).map(ContainerTarget::Agent);
|
||||
}
|
||||
project.container_id.as_deref().map(Self::docker)
|
||||
}
|
||||
|
||||
/// [`Self::for_project`], waiting (up to `aws_host::REATTACH_WAIT`) for an
|
||||
/// AWS project whose tunnel is being re-opened in the background after a
|
||||
/// restart. For a Docker project it is `for_project`.
|
||||
pub async fn for_project_waiting(project: &Project) -> Option<Self> {
|
||||
if project.is_aws() {
|
||||
return tunnels::wait_endpoint(&project.id, aws_host::REATTACH_WAIT)
|
||||
.await
|
||||
.map(ContainerTarget::Agent);
|
||||
}
|
||||
Self::for_project(project)
|
||||
}
|
||||
|
||||
/// The host-specific identifier (a Docker container id today). For logs
|
||||
/// and for keying per-container state; never parse it.
|
||||
pub fn id(&self) -> &str {
|
||||
|
||||
@@ -115,6 +115,10 @@ impl EcsTask {
|
||||
/// ARN), and check it can take a session now: the task is running and
|
||||
/// its ECS Exec agent is up. Fargate starts that agent a few seconds after
|
||||
/// the container, so "not ready yet" is a normal answer just after start.
|
||||
///
|
||||
/// The AWS host builds its target from `aws::tasks::wait_ready`'s answer
|
||||
/// instead (`aws_host::ecs_task`); this stays for the live tests.
|
||||
#[allow(dead_code)]
|
||||
pub async fn resolve(
|
||||
ecs: &aws_sdk_ecs::Client,
|
||||
cluster: &str,
|
||||
@@ -175,6 +179,7 @@ impl EcsTask {
|
||||
|
||||
/// The part of an ARN after the last `marker` (`.../cluster/name` → `name`),
|
||||
/// or the whole string when it is not an ARN.
|
||||
#[allow(dead_code)] // `EcsTask::resolve`
|
||||
fn last_arn_segment<'a>(s: &'a str, marker: &str) -> &'a str {
|
||||
if !s.starts_with("arn:") {
|
||||
return s;
|
||||
@@ -606,6 +611,7 @@ pub enum TunnelState {
|
||||
pub struct PortForward {
|
||||
local_port: u16,
|
||||
state: watch::Receiver<TunnelState>,
|
||||
#[allow(dead_code)] // `plugin_pid()`, for diagnostics and the tests
|
||||
plugin_pid: watch::Receiver<Option<u32>>,
|
||||
supervisor: Option<tokio::task::JoinHandle<()>>,
|
||||
_cancel: DropGuard,
|
||||
@@ -678,6 +684,7 @@ impl PortForward {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state)
|
||||
pub fn local_port(&self) -> u16 {
|
||||
self.local_port
|
||||
}
|
||||
@@ -688,11 +695,13 @@ impl PortForward {
|
||||
AgentEndpoint::loopback(self.local_port, token)
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state)
|
||||
pub fn state(&self) -> watch::Receiver<TunnelState> {
|
||||
self.state.clone()
|
||||
}
|
||||
|
||||
/// The running plugin's process id, if one is running.
|
||||
#[allow(dead_code)] // diagnostics and the tests (Task 8 may show the state)
|
||||
pub fn plugin_pid(&self) -> Option<u32> {
|
||||
*self.plugin_pid.borrow()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,301 @@
|
||||
//! The live SSM tunnels to AWS projects' agents, by project id.
|
||||
//!
|
||||
//! An AWS project's [`ContainerTarget`](super::ContainerTarget) is an agent
|
||||
//! endpoint at the loopback end of a [`PortForward`](super::ssm::PortForward);
|
||||
//! this registry is where `ContainerTarget::for_project` finds it, where
|
||||
//! `ProjectHost::stop` finds the task behind an endpoint, and what the
|
||||
//! shutdown path closes (every forward runs a `session-manager-plugin` child
|
||||
//! that would otherwise outlive the app — see `ssm.rs`).
|
||||
//!
|
||||
//! It also remembers which targets the AWS death watch found **stopped**, so
|
||||
//! `ContainerTarget::has_exited` can answer for an agent target (R3): only
|
||||
//! the host that runs the task can confirm a death, never the agent.
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Mutex;
|
||||
use std::time::Duration;
|
||||
|
||||
use aws_config::SdkConfig;
|
||||
use futures_util::future::BoxFuture;
|
||||
use tokio::sync::watch;
|
||||
|
||||
use super::agent::AgentEndpoint;
|
||||
use super::ssm::{EcsTask, PortForward};
|
||||
|
||||
/// Something that owns a tunnel's resources: the real [`PortForward`], or a
|
||||
/// test stand-in.
|
||||
pub trait TunnelHandle: Send + Sync {
|
||||
fn close(self: Box<Self>) -> BoxFuture<'static, ()>;
|
||||
}
|
||||
|
||||
impl TunnelHandle for PortForward {
|
||||
fn close(self: Box<Self>) -> BoxFuture<'static, ()> {
|
||||
Box::pin(async move { (*self).close().await })
|
||||
}
|
||||
}
|
||||
|
||||
/// One project's tunnel and the task it reaches.
|
||||
pub struct LiveTunnel {
|
||||
/// The SSM target (kept for diagnostics; the forward already uses it).
|
||||
#[allow(dead_code)]
|
||||
pub task: EcsTask,
|
||||
pub task_arn: String,
|
||||
pub cluster_arn: String,
|
||||
pub connection_id: String,
|
||||
pub region: String,
|
||||
/// The connection's SDK config, so a stop needs no settings lookup.
|
||||
pub sdk: SdkConfig,
|
||||
pub forward: Box<dyn TunnelHandle>,
|
||||
pub endpoint: AgentEndpoint,
|
||||
/// Runtime handles for the store (`set_aws_runtime`).
|
||||
pub access_point_id: Option<String>,
|
||||
pub task_definition_arn: String,
|
||||
}
|
||||
|
||||
/// Where an endpoint's task is: `(connection, region, cluster, task ARN)`.
|
||||
pub type TaskOf = (String, String, String, String);
|
||||
|
||||
#[derive(Default)]
|
||||
struct Registry {
|
||||
live: HashMap<String, LiveTunnel>,
|
||||
/// Target ids (endpoint base URLs) whose task the AWS watch saw STOPPED.
|
||||
stopped: HashSet<String>,
|
||||
/// Projects whose tunnel is being re-opened in the background.
|
||||
pending: HashSet<String>,
|
||||
}
|
||||
|
||||
fn registry() -> &'static Mutex<Registry> {
|
||||
static R: std::sync::OnceLock<Mutex<Registry>> = std::sync::OnceLock::new();
|
||||
R.get_or_init(Default::default)
|
||||
}
|
||||
|
||||
fn lock() -> std::sync::MutexGuard<'static, Registry> {
|
||||
registry().lock().unwrap_or_else(|e| e.into_inner())
|
||||
}
|
||||
|
||||
/// Wakes [`wait_endpoint`] callers when the registry changes.
|
||||
fn changed() -> &'static watch::Sender<u64> {
|
||||
static C: std::sync::OnceLock<watch::Sender<u64>> = std::sync::OnceLock::new();
|
||||
C.get_or_init(|| watch::channel(0).0)
|
||||
}
|
||||
|
||||
fn notify() {
|
||||
changed().send_modify(|n| *n = n.wrapping_add(1));
|
||||
}
|
||||
|
||||
/// Register `tunnel` for `project_id`, returning the one it replaces (the
|
||||
/// caller closes it).
|
||||
pub fn register(project_id: &str, tunnel: LiveTunnel) -> Option<LiveTunnel> {
|
||||
let old = {
|
||||
let mut r = lock();
|
||||
r.stopped.remove(tunnel.endpoint.base_url.as_str());
|
||||
r.pending.remove(project_id);
|
||||
r.live.insert(project_id.to_string(), tunnel)
|
||||
};
|
||||
notify();
|
||||
old
|
||||
}
|
||||
|
||||
pub fn get_endpoint(project_id: &str) -> Option<AgentEndpoint> {
|
||||
lock().live.get(project_id).map(|t| t.endpoint.clone())
|
||||
}
|
||||
|
||||
/// `(task ARN, access point id, task definition ARN)` of the project's tunnel.
|
||||
pub fn runtime(project_id: &str) -> Option<(String, Option<String>, String)> {
|
||||
lock().live.get(project_id).map(|t| {
|
||||
(
|
||||
t.task_arn.clone(),
|
||||
t.access_point_id.clone(),
|
||||
t.task_definition_arn.clone(),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// The project whose tunnel is `ep` and its task.
|
||||
pub fn task_of(ep: &AgentEndpoint) -> Option<TaskOf> {
|
||||
lock().live.values().find(|t| t.endpoint.base_url == ep.base_url).map(|t| {
|
||||
(
|
||||
t.connection_id.clone(),
|
||||
t.region.clone(),
|
||||
t.cluster_arn.clone(),
|
||||
t.task_arn.clone(),
|
||||
)
|
||||
})
|
||||
}
|
||||
|
||||
/// The project id and SDK config behind `ep`.
|
||||
pub(crate) fn owner_of(ep: &AgentEndpoint) -> Option<(String, SdkConfig)> {
|
||||
lock()
|
||||
.live
|
||||
.iter()
|
||||
.find(|(_, t)| t.endpoint.base_url == ep.base_url)
|
||||
.map(|(id, t)| (id.clone(), t.sdk.clone()))
|
||||
}
|
||||
|
||||
/// Take the project's tunnel out and close it.
|
||||
pub async fn close(project_id: &str) {
|
||||
let taken = {
|
||||
let mut r = lock();
|
||||
r.pending.remove(project_id);
|
||||
let t = r.live.remove(project_id);
|
||||
if let Some(ref t) = t {
|
||||
r.stopped.remove(t.endpoint.base_url.as_str());
|
||||
}
|
||||
t
|
||||
};
|
||||
notify();
|
||||
if let Some(t) = taken {
|
||||
t.forward.close().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Close every tunnel, concurrently. The shutdown path's step.
|
||||
pub async fn close_all() {
|
||||
let all: Vec<LiveTunnel> = {
|
||||
let mut r = lock();
|
||||
r.pending.clear();
|
||||
r.stopped.clear();
|
||||
r.live.drain().map(|(_, t)| t).collect()
|
||||
};
|
||||
notify();
|
||||
futures_util::future::join_all(all.into_iter().map(|t| t.forward.close())).await;
|
||||
}
|
||||
|
||||
/// The AWS watch saw the task behind target `target_id` STOPPED.
|
||||
pub fn mark_stopped(target_id: &str) {
|
||||
lock().stopped.insert(target_id.to_string());
|
||||
}
|
||||
|
||||
/// Whether the AWS watch has confirmed the task behind `target_id` stopped.
|
||||
pub fn is_stopped(target_id: &str) -> bool {
|
||||
lock().stopped.contains(target_id)
|
||||
}
|
||||
|
||||
/// A background re-attach for this project has begun / ended without a tunnel.
|
||||
pub fn set_pending(project_id: &str, pending: bool) {
|
||||
{
|
||||
let mut r = lock();
|
||||
if pending {
|
||||
r.pending.insert(project_id.to_string());
|
||||
} else {
|
||||
r.pending.remove(project_id);
|
||||
}
|
||||
}
|
||||
notify();
|
||||
}
|
||||
|
||||
pub fn is_pending(project_id: &str) -> bool {
|
||||
lock().pending.contains(project_id)
|
||||
}
|
||||
|
||||
/// The project's endpoint, waiting up to `timeout` while a background
|
||||
/// re-attach is still opening it. `None` when there is none.
|
||||
pub async fn wait_endpoint(project_id: &str, timeout: Duration) -> Option<AgentEndpoint> {
|
||||
let mut rx = changed().subscribe();
|
||||
let deadline = tokio::time::Instant::now() + timeout;
|
||||
loop {
|
||||
if let Some(ep) = get_endpoint(project_id) {
|
||||
return Some(ep);
|
||||
}
|
||||
if !is_pending(project_id) {
|
||||
return None;
|
||||
}
|
||||
if tokio::time::timeout_at(deadline, rx.changed()).await.is_err() {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) mod tests {
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
/// A stand-in tunnel that counts its closes.
|
||||
pub(crate) struct FakeTunnel(pub Arc<AtomicUsize>);
|
||||
impl TunnelHandle for FakeTunnel {
|
||||
fn close(self: Box<Self>) -> BoxFuture<'static, ()> {
|
||||
let n = self.0.clone();
|
||||
Box::pin(async move {
|
||||
n.fetch_add(1, Ordering::SeqCst);
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn fake_tunnel(port: u16, closes: Arc<AtomicUsize>) -> LiveTunnel {
|
||||
LiveTunnel {
|
||||
task: EcsTask {
|
||||
cluster: "triple-c-dev".into(),
|
||||
task_id: format!("t{port}"),
|
||||
runtime_id: "r".into(),
|
||||
},
|
||||
task_arn: format!("arn:aws:ecs:eu-west-1:111122223333:task/triple-c-dev/t{port}"),
|
||||
cluster_arn: "arn:aws:ecs:eu-west-1:111122223333:cluster/triple-c-dev".into(),
|
||||
connection_id: "c1".into(),
|
||||
region: "eu-west-1".into(),
|
||||
sdk: SdkConfig::builder().build(),
|
||||
forward: Box::new(FakeTunnel(closes)),
|
||||
endpoint: AgentEndpoint::loopback(port, "tok"),
|
||||
access_point_id: Some("fsap-1".into()),
|
||||
task_definition_arn: "arn:td:1".into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn pid(tag: &str) -> String {
|
||||
format!("tunnels-test-{tag}-{}", uuid::Uuid::new_v4())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_registered_tunnel_is_found_by_project_and_by_endpoint_and_closed_once() {
|
||||
let closes = Arc::new(AtomicUsize::new(0));
|
||||
let p = pid("reg");
|
||||
assert!(get_endpoint(&p).is_none());
|
||||
assert!(register(&p, fake_tunnel(61001, closes.clone())).is_none());
|
||||
let ep = get_endpoint(&p).unwrap();
|
||||
assert_eq!(ep, AgentEndpoint::loopback(61001, "tok"));
|
||||
let (conn, region, _, arn) = task_of(&ep).unwrap();
|
||||
assert_eq!((conn.as_str(), region.as_str()), ("c1", "eu-west-1"));
|
||||
assert!(arn.ends_with("/t61001"));
|
||||
assert_eq!(runtime(&p).unwrap().1.as_deref(), Some("fsap-1"));
|
||||
close(&p).await;
|
||||
close(&p).await;
|
||||
assert_eq!(closes.load(Ordering::SeqCst), 1);
|
||||
assert!(get_endpoint(&p).is_none() && task_of(&ep).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn re_registering_hands_back_the_old_tunnel_and_clears_a_stale_stop() {
|
||||
let closes = Arc::new(AtomicUsize::new(0));
|
||||
let p = pid("rereg");
|
||||
register(&p, fake_tunnel(61002, closes.clone()));
|
||||
let id = AgentEndpoint::loopback(61003, "tok").base_url.to_string();
|
||||
mark_stopped(&id);
|
||||
let old = register(&p, fake_tunnel(61003, closes.clone())).unwrap();
|
||||
assert_eq!(old.endpoint.base_url.port(), Some(61002));
|
||||
assert!(!is_stopped(&id), "a new task on a reused port is not stopped");
|
||||
old.forward.close().await;
|
||||
close(&p).await;
|
||||
assert_eq!(closes.load(Ordering::SeqCst), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn waiting_for_an_endpoint_ends_when_it_is_registered_or_the_reattach_gives_up() {
|
||||
let p = pid("wait");
|
||||
assert!(wait_endpoint(&p, Duration::from_secs(5)).await.is_none(), "nothing pending");
|
||||
set_pending(&p, true);
|
||||
let p2 = p.clone();
|
||||
let waiter = tokio::spawn(async move { wait_endpoint(&p2, Duration::from_secs(5)).await });
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
register(&p, fake_tunnel(61004, Arc::new(AtomicUsize::new(0))));
|
||||
assert_eq!(waiter.await.unwrap().unwrap().base_url.port(), Some(61004));
|
||||
close(&p).await;
|
||||
|
||||
set_pending(&p, true);
|
||||
let p2 = p.clone();
|
||||
let waiter = tokio::spawn(async move { wait_endpoint(&p2, Duration::from_secs(5)).await });
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
set_pending(&p, false);
|
||||
assert!(waiter.await.unwrap().is_none());
|
||||
}
|
||||
}
|
||||
@@ -27,8 +27,6 @@ pub const AWS_POLL_INTERVAL: Duration = Duration::from_secs(15);
|
||||
#[derive(Clone)]
|
||||
pub enum ExitSource {
|
||||
Docker,
|
||||
// Constructed by the AWS host (phase 5, Task 6); until then only by tests.
|
||||
#[allow(dead_code)]
|
||||
Aws(AwsWatch),
|
||||
}
|
||||
|
||||
@@ -80,7 +78,6 @@ pub struct AwsWatch {
|
||||
interval: Duration,
|
||||
}
|
||||
|
||||
#[allow(dead_code)] // Task 6
|
||||
impl AwsWatch {
|
||||
pub fn new(lookup: Arc<dyn AwsTaskLookup>) -> Self {
|
||||
Self::every(lookup, AWS_POLL_INTERVAL)
|
||||
@@ -180,9 +177,10 @@ impl ContainerTarget {
|
||||
ContainerTarget::Docker { container_id } => docker::events::has_exited(container_id).await,
|
||||
// The agent can only say "I answer" or "I don't"; silence is not a
|
||||
// positive exit (an SSM blip looks the same). Only the host that
|
||||
// runs the container can confirm a death — an AWS host answers
|
||||
// from ECS — so through the agent alone this never confirms one.
|
||||
ContainerTarget::Agent(_) => false,
|
||||
// runs the container can confirm a death: the AWS watch records
|
||||
// the targets whose task ECS reports STOPPED in the tunnel
|
||||
// registry (R3), and nothing else ever confirms one.
|
||||
ContainerTarget::Agent(ep) => super::tunnels::is_stopped(ep.base_url.as_str()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -332,6 +332,19 @@ pub fn run() {
|
||||
crate::host::watch::ExitSource::Docker,
|
||||
lifecycle_setup.cancellation(),
|
||||
));
|
||||
// The same for AWS projects: a DescribeTasks poll every 15 s over
|
||||
// the tasks of Running AWS projects (none: no AWS calls at all).
|
||||
lifecycle_setup.track(container_watch::spawn(
|
||||
app.handle().clone(),
|
||||
projects_store_setup.clone(),
|
||||
crate::host::watch::ExitSource::Aws(crate::host::watch::AwsWatch::new(
|
||||
std::sync::Arc::new(crate::host::aws_host::StoreTaskLookup {
|
||||
projects: projects_store_setup.clone(),
|
||||
settings: settings_store_setup.clone(),
|
||||
}),
|
||||
)),
|
||||
lifecycle_setup.cancellation(),
|
||||
));
|
||||
|
||||
// Marketplaces: refresh each once at startup, in the background.
|
||||
// Failures are logged, not toasted — the Marketplace tab shows them.
|
||||
@@ -509,6 +522,9 @@ pub fn run() {
|
||||
exec_manager.close_all_sessions(),
|
||||
auth_bridge.stop_all(),
|
||||
browser_view::manager().stop_all(),
|
||||
// Each AWS tunnel is a session-manager-plugin
|
||||
// child that `exit` would not end.
|
||||
crate::host::tunnels::close_all(),
|
||||
);
|
||||
};
|
||||
|
||||
|
||||
@@ -295,7 +295,8 @@ async fn handle_open(
|
||||
return Err(format!("Project '{}' is not running", project.name));
|
||||
}
|
||||
|
||||
let container = crate::host::ContainerTarget::for_project(&project)
|
||||
let container = crate::host::ContainerTarget::for_project_waiting(&project)
|
||||
.await
|
||||
.ok_or_else(|| "Container not running".to_string())?;
|
||||
|
||||
let cmd = match session_type {
|
||||
|
||||
Reference in new issue
Block a user