1use std::collections::{BTreeMap, BTreeSet};
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use std::time::Duration;
7
8use futures::StreamExt;
9use olai_stack_state::{Desired, Instance, InstanceId, Revision, Store};
10use olai_stack_topology::{
11 Catalog, LOCAL_CREDENTIALS, LaunchEnvError, LocalCredentials, LocalSecrets, MaterializedOutput,
12 Plan, PlanError, RenderError, RuntimePlacement, Secrets, Vantage, baseline_catalog_with,
13};
14use serde::Serialize;
15
16use crate::effective::{EffectiveEnv, EnvLayers, Layer};
17use crate::engine::{ComposeEngine, DownOptions, EngineStream, Progress, UpOptions};
18use crate::events::{EngineEvent, EventAction, EventFilter};
19use crate::ports::{BusyPort, probe_host_ports};
20use crate::secrets::{SecretError, SecretProviders};
21use crate::source::{Source, SourceCtx, SourceError};
22use crate::status::Health;
23
24pub const INSTANCES_DIR: &str = "instances";
26
27pub const PLACEMENT_KEYS: &[&str] = &[
29 "COMPOSE_PROJECT_NAME",
30 "TRESTLE_INSTANCE",
31 "TRESTLE_SLOT",
32 "TRESTLE_HOST_LABEL",
33];
34
35#[derive(Debug, thiserror::Error)]
37pub enum InstanceError {
38 #[error(transparent)]
40 Store(#[from] olai_stack_state::Error),
41 #[error(transparent)]
43 Engine(#[from] crate::Error),
44 #[error("module manifest `{path}` of the revision: {reason}")]
46 Module {
47 path: String,
49 reason: String,
51 },
52 #[error(transparent)]
54 Plan(#[from] PlanError),
55 #[error(transparent)]
57 Render(#[from] RenderError),
58 #[error(transparent)]
60 LaunchEnv(#[from] LaunchEnvError),
61 #[error(transparent)]
63 Secret(#[from] SecretError),
64 #[error(transparent)]
66 Source(#[from] SourceError),
67 #[error("local credential: {0}")]
69 LocalCredential(String),
70 #[error("host ports are already in use: {}", ports(.0))]
72 PortsBusy(Vec<BusyPort>),
73 #[error("instance {running} already runs revision {revision} of this environment")]
75 RevisionRunning {
76 revision: u32,
78 running: InstanceId,
80 },
81 #[error("{path}: {source}")]
83 Io {
84 path: PathBuf,
86 #[source]
88 source: std::io::Error,
89 },
90}
91
92fn ports(busy: &[BusyPort]) -> String {
93 busy.iter()
94 .map(|b| format!("{} ({})", b.port, b.owner))
95 .collect::<Vec<_>>()
96 .join(", ")
97}
98
99type Result<T, E = InstanceError> = std::result::Result<T, E>;
100
101#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
103pub struct InstanceEvent {
104 pub instance: InstanceId,
106 pub service: String,
108 pub action: EventAction,
110}
111
112#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
114pub struct Reconciled {
115 pub adopted: Vec<InstanceId>,
117 pub stale: Vec<InstanceId>,
119 pub orphans: Vec<String>,
121 pub removed: Vec<String>,
123}
124
125#[derive(Clone, Copy, Debug, Default)]
127pub struct ReconcileOptions {
128 pub remove_orphans: bool,
130}
131
132#[derive(Debug)]
134pub struct Materialized {
135 pub dir: PathBuf,
137 pub plan: Plan,
139 pub env: EffectiveEnv,
141 launch: BTreeMap<String, String>,
144}
145
146impl Materialized {
147 pub fn launch_var_names(&self) -> impl Iterator<Item = &str> {
149 self.launch.keys().map(String::as_str)
150 }
151}
152
153pub struct Instances<E> {
167 store: Arc<Store>,
168 engine: E,
169 root: PathBuf,
170 providers: SecretProviders,
171 sources: BTreeMap<String, Arc<dyn Source>>,
172 probe_ports: bool,
173 wait_timeout: Option<Duration>,
174}
175
176pub fn instance_project(workspace: &str, env: &str, revision: u32) -> String {
178 format!("{workspace}_{env}_r{revision}")
179}
180
181pub fn parse_instance_project(project: &str) -> Option<(&str, &str, u32)> {
183 let mut parts = project.split('_');
184 let (workspace, env, revision) = (parts.next()?, parts.next()?, parts.next()?);
185 if parts.next().is_some() {
186 return None;
187 }
188 let revision = revision.strip_prefix('r')?.parse().ok()?;
189 let valid = olai_stack_topology::workspace::is_valid_name;
190 (valid(workspace) && valid(env)).then_some((workspace, env, revision))
191}
192
193impl<E: ComposeEngine> Instances<E> {
194 pub fn new(store: Arc<Store>, engine: E, state_dir: impl Into<PathBuf>) -> Self {
198 Instances {
199 store,
200 engine,
201 root: state_dir.into().join(INSTANCES_DIR),
202 providers: SecretProviders::new().with(crate::secrets::EnvProvider),
203 sources: BTreeMap::new(),
204 probe_ports: true,
205 wait_timeout: None,
206 }
207 }
208
209 pub fn with_providers(mut self, providers: SecretProviders) -> Self {
211 self.providers = providers;
212 self
213 }
214
215 pub fn with_source(mut self, source: impl Source + 'static) -> Self {
217 self.sources
218 .insert(source.name().to_owned(), Arc::new(source));
219 self
220 }
221
222 pub fn without_port_probe(mut self) -> Self {
224 self.probe_ports = false;
225 self
226 }
227
228 pub fn with_wait_timeout(mut self, timeout: Duration) -> Self {
230 self.wait_timeout = Some(timeout);
231 self
232 }
233
234 pub fn engine(&self) -> &E {
236 &self.engine
237 }
238
239 pub fn store(&self) -> &Store {
241 &self.store
242 }
243
244 pub fn dir(&self, id: &InstanceId) -> PathBuf {
246 self.root.join(&id.0)
247 }
248
249 pub fn placement(&self, id: &InstanceId) -> Result<RuntimePlacement> {
252 let (instance, workspace, env) = self.describe(id)?;
253 let slot = match instance.slot {
254 Some(slot) => slot,
255 None => self.store.lease_slot(id)?.slot,
256 };
257 Ok(RuntimePlacement {
258 slot,
259 project_name: instance_project(&workspace.name, &env.name, instance.revision),
260 mesh_network: workspace
261 .mesh
262 .then(|| format!("trestle-{}-mesh", workspace.name)),
263 host_label: format!("{}-{}-r{}", workspace.name, env.name, instance.revision),
264 })
265 }
266
267 fn describe(
268 &self,
269 id: &InstanceId,
270 ) -> Result<(
271 Instance,
272 olai_stack_state::Workspace,
273 olai_stack_state::Environment,
274 )> {
275 let instance = self.store.instance(id)?;
276 let env = self.store.environment(&instance.environment)?;
277 let workspace = self.store.workspace_by_id(&env.workspace)?;
278 Ok((instance, workspace, env))
279 }
280
281 async fn base_layers(
283 &self,
284 id: &InstanceId,
285 ) -> Result<(EnvLayers, Revision, RuntimePlacement)> {
286 let (instance, workspace, env) = self.describe(id)?;
287 let revision = self
288 .store
289 .revision(&instance.environment, instance.revision)?;
290 let placement = self.placement(id)?;
291 let defaults: BTreeMap<String, String> = LOCAL_CREDENTIALS
292 .iter()
293 .map(|c| (c.env_var(), c.default.to_owned()))
294 .collect();
295 let reserved: BTreeMap<String, String> = [
296 ("COMPOSE_PROJECT_NAME", placement.project_name.clone()),
297 ("TRESTLE_INSTANCE", id.0.clone()),
298 ("TRESTLE_SLOT", placement.slot.to_string()),
299 ("TRESTLE_HOST_LABEL", placement.host_label.clone()),
300 ]
301 .into_iter()
302 .map(|(k, v)| (k.to_owned(), v))
303 .collect();
304 let mut layers = EnvLayers::new()
305 .with_secrets(Layer::CatalogDefaults, &defaults)
306 .with_plain(Layer::Placement, &reserved)
307 .with_vars(Layer::WorkspaceVars, &workspace.vars)
308 .with_vars(Layer::EnvironmentVars, &env.vars)
309 .with_vars(Layer::RevisionVars, revision.vars())
310 .with_vars(Layer::InstanceOverrides, &instance.overrides);
311 for binding in &instance.sources {
312 let name = &binding.name;
313 let source = self.sources.get(name).ok_or_else(|| SourceError {
314 source_name: name.clone(),
315 reason: "no such source is configured".into(),
316 })?;
317 let ctx = SourceCtx {
318 instance: id.clone(),
319 workspace: workspace.name.clone(),
320 environment: env.name.clone(),
321 revision: instance.revision,
322 config: merge_config(workspace.sources.get(name), &binding.config),
323 };
324 let mut overlay = source.overlay(&ctx).await?;
325 if overlay.is_expired(std::time::SystemTime::now()) {
326 overlay = source.refresh(&ctx, &overlay).await?;
327 }
328 layers = layers.with_overlay(name, &overlay);
329 }
330 Ok((layers, revision, placement))
331 }
332
333 fn plan(
336 &self,
337 revision: &Revision,
338 env_name: &str,
339 placement: &RuntimePlacement,
340 env: &EffectiveEnv,
341 ) -> Result<Plan> {
342 let secrets: Secrets = env.launch_vars().into_iter().collect();
343 let credentials = LocalCredentials::from_secrets(&secrets)
344 .map_err(|e| InstanceError::LocalCredential(e.to_string()))?;
345 let mut catalog = baseline_catalog_with(&credentials);
346 for (path, text) in revision.modules() {
347 let custom =
348 Catalog::from_manifests([text.as_str()]).map_err(|e| InstanceError::Module {
349 path: path.clone(),
350 reason: e.to_string(),
351 })?;
352 catalog = catalog.merge(custom);
353 }
354 let mut ctx = revision.manifest().context.clone();
355 ctx.env_name = env_name.to_owned();
356 placement.apply(&mut ctx);
357 ctx.local_secrets = LocalSecrets::Separate;
359 ctx.secrets = Some(secrets);
360 Ok(catalog.plan(&revision.manifest().selection, &ctx)?)
361 }
362
363 pub async fn effective_env(&self, id: &InstanceId) -> Result<EffectiveEnv> {
365 Ok(self.resolve(id).await?.0)
366 }
367
368 async fn resolve(&self, id: &InstanceId) -> Result<(EffectiveEnv, Plan)> {
369 let (layers, revision, placement) = self.base_layers(id).await?;
370 let env_name = self.store.environment(revision.environment())?.name;
371 let first = layers.resolve(&self.providers)?;
375 let plan = self.plan(&revision, &env_name, &placement, &first)?;
376 let secrets: Vec<&str> = first
379 .vars()
380 .values()
381 .filter(|r| r.value.is_secret() && !r.value.expose().is_empty())
382 .map(|r| r.value.expose())
383 .collect();
384 let split = |env: &olai_stack_topology::InjectedEnv| {
385 let (secret, plain): (BTreeMap<_, _>, BTreeMap<_, _>) = env
386 .iter()
387 .map(|(k, v)| (k.to_owned(), v.to_owned()))
388 .partition(|(_, v)| secrets.iter().any(|s| v.contains(s)));
389 (secret, plain)
390 };
391 let (injected_secret, injected) = split(&plan.env);
392 let (client_secret, client) = split(plan.client_env(Vantage::Host));
393 let env = layers
394 .with_plain(Layer::RevisionManifest, &injected)
395 .with_secrets(Layer::RevisionManifest, &injected_secret)
396 .with_plain(Layer::ClientEnv, &client)
397 .with_secrets(Layer::ClientEnv, &client_secret)
398 .resolve(&self.providers)?;
399 Ok((env, plan))
400 }
401
402 pub async fn materialize(&self, id: &InstanceId) -> Result<Materialized> {
405 let (env, plan) = self.resolve(id).await?;
406 let (output, launch_env) = plan.materialize()?.into_launch_env()?;
407 let dir = self.dir(id);
408 write_tree(&dir, &output)?;
409 let mut launch = env.launch_vars();
410 launch.extend(launch_env.into_vars());
411 Ok(Materialized {
412 dir,
413 plan,
414 env,
415 launch,
416 })
417 }
418
419 pub async fn start(&self, id: &InstanceId) -> Result<Materialized> {
423 let instance = self.store.instance(id)?;
424 let project = self.placement(id)?.project_name;
425 for other in self.store.instances()? {
426 if other.id != *id
427 && other.environment == instance.environment
428 && other.revision == instance.revision
429 && other.desired == Desired::Running
430 {
431 return Err(InstanceError::RevisionRunning {
432 revision: instance.revision,
433 running: other.id,
434 });
435 }
436 }
437 let materialized = self.materialize(id).await?;
438 if self.probe_ports {
439 let running = self
440 .engine
441 .ls()
442 .await?
443 .into_iter()
444 .any(|p| p.name == project && p.is_running());
445 if !running {
447 let busy = probe_host_ports(&materialized.plan);
448 if !busy.is_empty() {
449 return Err(InstanceError::PortsBusy(busy));
450 }
451 }
452 }
453 self.store.set_desired(id, Desired::Running)?;
454 let opts = UpOptions {
455 wait: true,
456 wait_timeout: self.wait_timeout,
457 env: materialized.launch.clone(),
458 progress: Progress::Quiet,
459 };
460 self.engine.up(&materialized.dir, &opts).await?;
461 Ok(materialized)
462 }
463
464 pub async fn stop(&self, id: &InstanceId, volumes: bool) -> Result<()> {
470 let (dir, env) = match self.materialize(id).await {
472 Ok(m) => (m.dir, m.launch),
473 Err(InstanceError::Source(_)) if self.dir(id).join("compose.yaml").is_file() => {
474 (self.dir(id), BTreeMap::new())
475 }
476 Err(e) => return Err(e),
477 };
478 let opts = DownOptions {
479 volumes,
480 env,
481 progress: Progress::Quiet,
482 };
483 self.engine.down(&dir, &opts).await?;
484 self.store.set_desired(id, Desired::Stopped)?;
485 Ok(())
486 }
487
488 pub async fn remove(&self, id: &InstanceId, volumes: bool) -> Result<()> {
491 let project = self.placement(id)?.project_name;
492 if self.engine.ls().await?.iter().any(|p| p.name == project) {
493 self.stop(id, volumes).await?;
494 }
495 let dir = self.dir(id);
496 if dir.exists() {
497 std::fs::remove_dir_all(&dir)
498 .map_err(|source| InstanceError::Io { path: dir, source })?;
499 }
500 self.store.delete_instance(id)?;
501 Ok(())
502 }
503
504 pub async fn health(&self, id: &InstanceId) -> Result<BTreeMap<String, Health>> {
506 let project = self.placement(id)?.project_name;
507 Ok(self
508 .engine
509 .ps(&project)
510 .await?
511 .into_iter()
512 .map(|s| (s.service, s.health))
513 .collect())
514 }
515
516 pub async fn events(&self, id: &InstanceId) -> Result<EngineStream<InstanceEvent>> {
518 let project = self.placement(id)?.project_name;
519 let id = id.clone();
520 Ok(self
521 .engine
522 .events(&EventFilter::Project(project))
523 .await?
524 .map(move |e: crate::Result<EngineEvent>| {
525 e.map(|e| InstanceEvent {
526 instance: id.clone(),
527 service: e.service,
528 action: e.action,
529 })
530 })
531 .boxed())
532 }
533
534 pub async fn reconcile(&self, opts: ReconcileOptions) -> Result<Reconciled> {
539 let projects = self.engine.ls().await?;
540 let mut by_project: BTreeMap<String, InstanceId> = BTreeMap::new();
541 let instances = self.store.instances()?;
542 for instance in &instances {
543 let env = self.store.environment(&instance.environment)?;
544 let ws = self.store.workspace_by_id(&env.workspace)?;
545 by_project.insert(
546 instance_project(&ws.name, &env.name, instance.revision),
547 instance.id.clone(),
548 );
549 }
550 let mut out = Reconciled::default();
551 let mut present = BTreeSet::new();
552 for project in &projects {
553 if parse_instance_project(&project.name).is_none() {
554 continue;
555 }
556 present.insert(project.name.clone());
557 match by_project.get(&project.name) {
558 Some(id) if project.is_running() => {
559 self.store.set_desired(id, Desired::Running)?;
562 out.adopted.push(id.clone());
563 }
564 Some(_) => {}
565 None => {
566 out.orphans.push(project.name.clone());
567 if opts.remove_orphans && self.remove_orphan(&project.config_files).await? {
568 out.removed.push(project.name.clone());
569 }
570 }
571 }
572 }
573 for instance in instances {
574 if instance.desired != Desired::Running {
575 continue;
576 }
577 let project = by_project
578 .iter()
579 .find(|(_, id)| **id == instance.id)
580 .map(|(p, _)| p.clone())
581 .unwrap_or_default();
582 if !present.contains(&project) {
583 self.store.set_desired(&instance.id, Desired::Stopped)?;
584 out.stale.push(instance.id);
585 }
586 }
587 Ok(out)
588 }
589
590 async fn remove_orphan(&self, config_files: &str) -> Result<bool> {
593 let Some(dir) = config_files
595 .split(',')
596 .next()
597 .filter(|f| !f.is_empty())
598 .map(Path::new)
599 .and_then(Path::parent)
600 .filter(|d| d.is_dir())
601 else {
602 return Ok(false);
603 };
604 let opts = DownOptions {
605 progress: Progress::Quiet,
606 ..DownOptions::default()
607 };
608 self.engine.down(dir, &opts).await?;
609 Ok(true)
610 }
611}
612
613fn merge_config(
616 workspace: Option<&serde_json::Value>,
617 binding: &serde_json::Value,
618) -> serde_json::Value {
619 use serde_json::Value;
620 match (workspace, binding) {
621 (None, b) => b.clone(),
622 (Some(w), Value::Null) => w.clone(),
623 (Some(Value::Object(w)), Value::Object(b)) => {
624 let mut merged = w.clone();
625 merged.extend(b.iter().map(|(k, v)| (k.clone(), v.clone())));
626 Value::Object(merged)
627 }
628 (Some(_), b) => b.clone(),
629 }
630}
631
632const RENDERED_FILE: &str = ".trestle-rendered";
635
636fn write_tree(dir: &Path, output: &MaterializedOutput) -> Result<()> {
644 let io = |path: &Path| {
645 let path = path.to_owned();
646 move |source| InstanceError::Io { path, source }
647 };
648 std::fs::create_dir_all(dir).map_err(io(dir))?;
649 #[cfg(unix)]
650 {
651 use std::os::unix::fs::PermissionsExt as _;
652 std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)).map_err(io(dir))?;
653 }
654 let listing = dir.join(RENDERED_FILE);
655 let previous = match std::fs::read_to_string(&listing) {
656 Ok(text) => text,
657 Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
658 Err(e) => return Err(io(&listing)(e)),
659 };
660 output.write_to(dir).map_err(io(dir))?;
663 let current: BTreeSet<&str> = output.files.iter().map(|f| f.path.as_str()).collect();
664 for stale in previous.lines().filter(|l| !l.is_empty()) {
665 let relative = Path::new(stale);
667 let inside = relative
668 .components()
669 .all(|c| matches!(c, std::path::Component::Normal(_)));
670 if current.contains(stale) || !inside {
671 continue;
672 }
673 let path = dir.join(relative);
674 match std::fs::remove_file(&path) {
675 Ok(()) => remove_empty_parents(dir, &path),
676 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
677 Err(e) => return Err(io(&path)(e)),
678 }
679 }
680 let mut text = current.into_iter().collect::<Vec<_>>().join("\n");
681 text.push('\n');
682 std::fs::write(&listing, text).map_err(io(&listing))
683}
684
685fn remove_empty_parents(root: &Path, file: &Path) {
687 let mut dir = file.parent();
688 while let Some(d) = dir {
689 if d == root || !d.starts_with(root) || std::fs::remove_dir(d).is_err() {
690 break;
691 }
692 dir = d.parent();
693 }
694}
695
696#[cfg(test)]
697mod tests {
698 use olai_stack_topology::OutputFile;
699
700 use super::*;
701
702 #[test]
703 fn instance_projects_parse_back() {
704 let name = instance_project("demo-x", "prod", 3);
705 assert_eq!(name, "demo-x_prod_r3");
706 assert_eq!(parse_instance_project(&name), Some(("demo-x", "prod", 3)));
707 for other in ["demo-dev", "a_b", "a_b_c", "a_b_r", "A_b_r1", "a_b_r1_x"] {
708 assert_eq!(parse_instance_project(other), None, "{other}");
709 }
710 }
711
712 fn output(files: &[(&str, &str)]) -> MaterializedOutput {
713 MaterializedOutput {
714 files: files
715 .iter()
716 .map(|(path, contents)| OutputFile {
717 path: (*path).into(),
718 contents: (*contents).into(),
719 sensitive: false,
720 preserve: false,
721 moved_from: None,
722 })
723 .collect(),
724 deploy: Vec::new(),
725 }
726 }
727
728 #[test]
729 fn a_render_replaces_its_own_files_and_leaves_the_containers_alone() {
730 let tmp = tempfile::tempdir().unwrap();
731 let dir = tmp.path().join("inst");
732 write_tree(
733 &dir,
734 &output(&[("compose.yaml", "v1"), ("modules/old/a.env", "A=1")]),
735 )
736 .unwrap();
737
738 let data = dir.join("data/postgres");
741 std::fs::create_dir_all(&data).unwrap();
742 std::fs::write(data.join("PG_VERSION"), "16").unwrap();
743 #[cfg(unix)]
744 {
745 use std::os::unix::fs::PermissionsExt as _;
746 std::fs::set_permissions(&data, std::fs::Permissions::from_mode(0o500)).unwrap();
747 }
748
749 write_tree(
750 &dir,
751 &output(&[("compose.yaml", "v2"), ("modules/new/b.env", "B=2")]),
752 )
753 .unwrap();
754 assert_eq!(
755 std::fs::read_to_string(dir.join("compose.yaml")).unwrap(),
756 "v2"
757 );
758 assert!(dir.join("modules/new/b.env").is_file());
759 assert!(
760 !dir.join("modules/old").exists(),
761 "the previous render's stale file and its empty directory are gone"
762 );
763 assert_eq!(
764 std::fs::read_to_string(data.join("PG_VERSION")).unwrap(),
765 "16",
766 "what the containers wrote stays"
767 );
768 #[cfg(unix)]
769 {
770 use std::os::unix::fs::PermissionsExt as _;
771 std::fs::set_permissions(&data, std::fs::Permissions::from_mode(0o700)).unwrap();
772 }
773 }
774
775 #[test]
776 fn the_listing_never_reaches_outside_the_tree() {
777 let tmp = tempfile::tempdir().unwrap();
778 let outside = tmp.path().join("keep.txt");
779 std::fs::write(&outside, "keep").unwrap();
780 let dir = tmp.path().join("inst");
781 std::fs::create_dir_all(&dir).unwrap();
782 std::fs::write(dir.join(RENDERED_FILE), "../keep.txt\n/etc/hosts\n").unwrap();
783 write_tree(&dir, &output(&[("compose.yaml", "v1")])).unwrap();
784 assert!(outside.is_file());
785 }
786}