1use crate::{
2 AdvanceModuleOperation, JsonFileModuleOperationStore, LinkedWorkspaceError,
3 LinkedWorkspaceTransaction, MODULE_APPROVAL_PROTOCOL, ManagementActor, ModuleApproval,
4 ModuleChangePlan, ModuleEffectOutcome, ModuleEffectReceipt, ModuleEnvironmentPolicy,
5 ModuleManagementEngine, ModuleManagementError, ModuleOperation, ModuleOperationError,
6 ModuleOperationJournal, ModuleOperationKind, ModuleOperationState, ModuleOperationStore,
7 ModuleOperationStoreError, ModulePlanEffect, ModuleRootChange, StartModuleOperation,
8 WorkspaceModuleManagement, WorkspaceModuleManagementError,
9};
10use chrono::{DateTime, Duration, Utc};
11use lenso_contracts::{ArtifactReference, digest_json};
12use std::collections::BTreeSet;
13use std::fs;
14use std::path::{Path, PathBuf};
15use thiserror::Error;
16
17const POLICY_PATH: &str = ".lenso/module-environment-policy.json";
18const MANAGEMENT_ROOT: &str = ".lenso/module-management";
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct ModuleEffectExecution {
22 pub outcome: ModuleEffectOutcome,
23 pub evidence_references: Vec<ArtifactReference>,
24}
25
26pub trait ModuleEffectAdapter: std::fmt::Debug + Send + Sync {
27 fn execute(
28 &self,
29 workspace_root: &Path,
30 operation: &ModuleOperation,
31 effect: &ModulePlanEffect,
32 ) -> Result<ModuleEffectExecution, ModuleEffectAdapterError>;
33}
34
35#[derive(Debug, Error)]
36pub enum ModuleEffectAdapterError {
37 #[error("effect adapter does not support `{effect_id}`: {reason}")]
38 Unsupported { effect_id: String, reason: String },
39 #[error("effect `{effect_id}` failed: {reason}")]
40 Failed { effect_id: String, reason: String },
41}
42
43#[derive(Debug, Clone, PartialEq, Eq)]
44pub struct StartReviewedModulePlan {
45 pub idempotency_key: String,
46 pub plan: ModuleChangePlan,
47}
48
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct ApproveModuleOperation {
51 pub expected_revision: u64,
52 pub boundary_id: String,
53 pub reason: String,
54 pub nonce: String,
55}
56
57#[derive(Debug, Error)]
58pub enum WorkspaceModuleOperatorError {
59 #[error(transparent)]
60 Workspace(#[from] WorkspaceModuleManagementError),
61 #[error(transparent)]
62 Management(#[from] ModuleManagementError),
63 #[error(transparent)]
64 Store(#[from] ModuleOperationStoreError),
65 #[error(transparent)]
66 Linked(#[from] LinkedWorkspaceError),
67 #[error("Module operation artifact I/O failed: {0}")]
68 Io(#[from] std::io::Error),
69 #[error("Module operation artifact JSON failed: {0}")]
70 Json(#[from] serde_json::Error),
71 #[error("reviewed plan is stale or was not produced by this target")]
72 StalePlan,
73 #[error("Module environment policy is unavailable")]
74 PolicyUnavailable,
75 #[error("operation cannot be cancelled after target mutation began")]
76 CancellationUnsafe,
77}
78
79#[derive(Debug)]
80pub struct WorkspaceModuleOperator<A> {
81 root: PathBuf,
82 adapter: A,
83}
84
85impl<A: ModuleEffectAdapter> WorkspaceModuleOperator<A> {
86 pub fn new(root: impl Into<PathBuf>, adapter: A) -> Self {
87 Self {
88 root: root.into(),
89 adapter,
90 }
91 }
92
93 pub fn start(
94 &self,
95 request: &StartReviewedModulePlan,
96 actor: &ManagementActor,
97 holder_id: &str,
98 now: DateTime<Utc>,
99 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
100 if request.idempotency_key.trim().is_empty() {
101 return Err(WorkspaceModuleOperatorError::StalePlan);
102 }
103 let verified = WorkspaceModuleManagement::new(&self.root)
104 .preview(request.plan.request.clone(), request.plan.created_at)?;
105 if verified != request.plan {
106 return Err(WorkspaceModuleOperatorError::StalePlan);
107 }
108 let policy = self.policy()?;
109 self.persist_plan(&verified)?;
110 let operation_id = content_id(
111 "module-op",
112 &(
113 verified.application_id.as_str(),
114 request.idempotency_key.as_str(),
115 ),
116 )?;
117 Ok(self.engine().start(StartModuleOperation {
118 operation_id: &operation_id,
119 idempotency_key: &request.idempotency_key,
120 operation_kind: operation_kind(&verified.request),
121 plan: &verified,
122 policy: &policy,
123 actor,
124 approvals: Vec::new(),
125 holder_id,
126 now,
127 })?)
128 }
129
130 pub fn operation(
131 &self,
132 operation_id: &str,
133 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
134 Ok(self.engine().store().load(operation_id)?)
135 }
136
137 pub fn journal(
138 &self,
139 operation_id: &str,
140 ) -> Result<ModuleOperationJournal, WorkspaceModuleOperatorError> {
141 Ok(self.engine().store().journal(operation_id)?)
142 }
143
144 pub fn approve(
145 &self,
146 operation_id: &str,
147 request: ApproveModuleOperation,
148 actor: &ManagementActor,
149 holder_id: &str,
150 now: DateTime<Utc>,
151 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
152 let operation = self.operation(operation_id)?;
153 let plan = self.load_plan(&operation.plan_digest)?;
154 let policy = self.policy()?;
155 let boundary = plan
156 .approval_boundaries
157 .iter()
158 .find(|boundary| boundary.boundary_id == request.boundary_id)
159 .ok_or(WorkspaceModuleOperatorError::StalePlan)?;
160 let approval = ModuleApproval {
161 protocol: MODULE_APPROVAL_PROTOCOL.to_owned(),
162 approval_id: content_id(
163 "module-approval",
164 &(
165 operation_id,
166 request.boundary_id.as_str(),
167 request.nonce.as_str(),
168 ),
169 )?,
170 plan_digest: plan.plan_digest.clone(),
171 application_id: plan.application_id.clone(),
172 environment_id: plan.environment_id.clone(),
173 expected_target_revision: plan.expected_target_revision,
174 boundary_id: boundary.boundary_id.clone(),
175 risk_class: boundary.risk_class,
176 actor_id: actor.actor_id.clone(),
177 verified_authorities: actor.verified_authorities.iter().cloned().collect(),
178 reason: request.reason,
179 issued_at: now,
180 expires_at: now
181 + Duration::seconds(
182 i64::try_from(policy.maximum_approval_age_seconds).unwrap_or(i64::MAX),
183 ),
184 nonce: request.nonce,
185 };
186 Ok(self.engine().submit_approval(
187 operation_id,
188 request.expected_revision,
189 &plan,
190 &policy,
191 actor,
192 approval,
193 holder_id,
194 now,
195 )?)
196 }
197
198 pub fn apply(
199 &self,
200 operation_id: &str,
201 actor: &ManagementActor,
202 now: DateTime<Utc>,
203 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
204 let plan = self.load_bound_plan(operation_id)?;
205 let engine = self.engine();
206 let mut operation = engine.store().load(operation_id)?;
207 if matches!(
208 operation.state,
209 ModuleOperationState::Ready | ModuleOperationState::ApplyingFiles
210 ) {
211 operation = LinkedWorkspaceTransaction::new(&self.root).apply(
212 &engine,
213 operation_id,
214 &plan,
215 operation.fencing_token,
216 &actor.actor_id,
217 now,
218 )?;
219 }
220 operation = self.run_from_state(&engine, operation, &plan, actor, now)?;
221 if matches!(
222 operation.state,
223 ModuleOperationState::Succeeded | ModuleOperationState::Blocked
224 ) {
225 let _ = engine.release_lease(MANAGEMENT_HOLDER, operation.fencing_token);
226 }
227 Ok(operation)
228 }
229
230 pub fn retry(
231 &self,
232 operation_id: &str,
233 expected_revision: u64,
234 actor: &ManagementActor,
235 now: DateTime<Utc>,
236 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
237 let plan = self.load_bound_plan(operation_id)?;
238 let current = self.operation(operation_id)?;
239 let next_state = next_incomplete_state(¤t, &plan);
240 self.engine().retry_blocked(
241 operation_id,
242 expected_revision,
243 next_state,
244 MANAGEMENT_HOLDER,
245 self.policy()?.maximum_lease_seconds,
246 &actor.actor_id,
247 now,
248 )?;
249 self.apply(operation_id, actor, now)
250 }
251
252 pub fn resume(
253 &self,
254 operation_id: &str,
255 expected_revision: u64,
256 actor: &ManagementActor,
257 now: DateTime<Utc>,
258 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
259 let plan = self.load_bound_plan(operation_id)?;
260 let current = self.operation(operation_id)?;
261 if current.revision != expected_revision {
262 return Err(ModuleOperationStoreError::RevisionConflict {
263 operation_id: operation_id.to_owned(),
264 expected: expected_revision,
265 observed: current.revision,
266 }
267 .into());
268 }
269 let evidence =
270 LinkedWorkspaceTransaction::new(&self.root).resume_evidence(¤t, &plan, now)?;
271 let policy = self.policy()?;
272 self.engine().resume_after_crash(
273 operation_id,
274 expected_revision,
275 &evidence,
276 MANAGEMENT_HOLDER,
277 policy.maximum_lease_seconds,
278 &actor.actor_id,
279 now,
280 )?;
281 self.apply(operation_id, actor, now)
282 }
283
284 pub fn cancel(
285 &self,
286 operation_id: &str,
287 expected_revision: u64,
288 actor: &ManagementActor,
289 now: DateTime<Utc>,
290 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
291 let current = self.operation(operation_id)?;
292 if !matches!(
293 current.state,
294 ModuleOperationState::AwaitingApproval | ModuleOperationState::Ready
295 ) {
296 return Err(WorkspaceModuleOperatorError::CancellationUnsafe);
297 }
298 let operation = self.engine().advance(AdvanceModuleOperation {
299 operation_id: operation_id.to_owned(),
300 expected_revision,
301 fencing_token: current.fencing_token,
302 next_state: ModuleOperationState::Cancelled,
303 actor_id: actor.actor_id.clone(),
304 outcome_code: "operation_cancelled_before_mutation".to_owned(),
305 evidence_references: Vec::new(),
306 error: None,
307 next_actions: Vec::new(),
308 now,
309 })?;
310 if current.fencing_token != 0 {
311 let _ = self
312 .engine()
313 .release_lease(MANAGEMENT_HOLDER, current.fencing_token);
314 }
315 Ok(operation)
316 }
317
318 fn run_from_state(
319 &self,
320 engine: &ModuleManagementEngine<JsonFileModuleOperationStore>,
321 mut operation: ModuleOperation,
322 plan: &ModuleChangePlan,
323 actor: &ManagementActor,
324 now: DateTime<Utc>,
325 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
326 if operation.state == ModuleOperationState::FilesApplied {
327 operation = transition(
328 engine,
329 operation,
330 ModuleOperationState::StagingConfiguration,
331 &actor.actor_id,
332 "configuration_stage_started",
333 now,
334 )?;
335 }
336 if operation.state == ModuleOperationState::StagingConfiguration {
337 operation = self.execute_matching(engine, operation, plan, actor, now, |effect| {
338 matches!(
339 effect,
340 ModulePlanEffect::Configuration { .. }
341 | ModulePlanEffect::Protected { .. }
342 | ModulePlanEffect::ConsoleComposition { .. }
343 | ModulePlanEffect::ServiceInstallation { .. }
344 | ModulePlanEffect::ServiceRemoval { .. }
345 )
346 })?;
347 if operation.state == ModuleOperationState::Blocked {
348 return Ok(operation);
349 }
350 operation = transition(
351 engine,
352 operation,
353 ModuleOperationState::Migrating,
354 &actor.actor_id,
355 "migration_stage_started",
356 now,
357 )?;
358 }
359 if operation.state == ModuleOperationState::Migrating {
360 operation = self.execute_matching(engine, operation, plan, actor, now, |effect| {
361 matches!(effect, ModulePlanEffect::Migration { .. })
362 })?;
363 if operation.state == ModuleOperationState::Blocked {
364 return Ok(operation);
365 }
366 operation = transition(
367 engine,
368 operation,
369 ModuleOperationState::Verifying,
370 &actor.actor_id,
371 "verification_stage_started",
372 now,
373 )?;
374 }
375 if operation.state == ModuleOperationState::Verifying {
376 operation = self.execute_matching(engine, operation, plan, actor, now, |effect| {
377 matches!(effect, ModulePlanEffect::Validate { .. })
378 })?;
379 if operation.state == ModuleOperationState::Blocked {
380 return Ok(operation);
381 }
382 operation = transition(
383 engine,
384 operation,
385 ModuleOperationState::Activating,
386 &actor.actor_id,
387 "activation_stage_started",
388 now,
389 )?;
390 }
391 if operation.state == ModuleOperationState::Activating {
392 operation = self.execute_matching(engine, operation, plan, actor, now, |effect| {
393 matches!(
394 effect,
395 ModulePlanEffect::Restart { .. }
396 | ModulePlanEffect::ServiceRestart { .. }
397 | ModulePlanEffect::Activate { .. }
398 )
399 })?;
400 if operation.state == ModuleOperationState::Blocked {
401 return Ok(operation);
402 }
403 operation = transition(
404 engine,
405 operation,
406 ModuleOperationState::Succeeded,
407 &actor.actor_id,
408 "reviewed_plan_succeeded",
409 now,
410 )?;
411 }
412 Ok(operation)
413 }
414
415 fn execute_matching(
416 &self,
417 engine: &ModuleManagementEngine<JsonFileModuleOperationStore>,
418 mut operation: ModuleOperation,
419 plan: &ModuleChangePlan,
420 actor: &ManagementActor,
421 now: DateTime<Utc>,
422 matches: impl Fn(&ModulePlanEffect) -> bool,
423 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
424 for effect in plan.effects.iter().filter(|effect| matches(effect)) {
425 if operation
426 .effect_receipts
427 .iter()
428 .any(|receipt| receipt.effect_id == effect.effect_id())
429 {
430 continue;
431 }
432 let execution = if matches!(effect, ModulePlanEffect::Protected { .. }) {
433 Ok(ModuleEffectExecution {
434 outcome: ModuleEffectOutcome::Verified,
435 evidence_references: Vec::new(),
436 })
437 } else {
438 self.adapter.execute(&self.root, &operation, effect)
439 };
440 let execution = match execution {
441 Ok(execution) => execution,
442 Err(error) => return Self::block(engine, operation, actor, effect, &error, now),
443 };
444 operation = engine.record_effect_receipt(
445 &operation.operation_id,
446 operation.revision,
447 operation.fencing_token,
448 &actor.actor_id,
449 ModuleEffectReceipt {
450 receipt_id: format!("{}:{}", operation.operation_id, effect.effect_id()),
451 effect_id: effect.effect_id().to_owned(),
452 effect_digest: digest_json(effect)?,
453 operation_id: operation.operation_id.clone(),
454 attempt: operation.attempt,
455 fencing_token: operation.fencing_token,
456 outcome: execution.outcome,
457 evidence_references: execution.evidence_references,
458 committed_at: now,
459 },
460 now,
461 )?;
462 }
463 Ok(operation)
464 }
465
466 fn block(
467 engine: &ModuleManagementEngine<JsonFileModuleOperationStore>,
468 operation: ModuleOperation,
469 actor: &ManagementActor,
470 effect: &ModulePlanEffect,
471 error: &ModuleEffectAdapterError,
472 now: DateTime<Utc>,
473 ) -> Result<ModuleOperation, WorkspaceModuleOperatorError> {
474 let code = match error {
475 ModuleEffectAdapterError::Unsupported { .. } => "effect_adapter_unavailable",
476 ModuleEffectAdapterError::Failed { .. } => "effect_execution_failed",
477 };
478 Ok(engine.advance(AdvanceModuleOperation {
479 operation_id: operation.operation_id,
480 expected_revision: operation.revision,
481 fencing_token: operation.fencing_token,
482 next_state: ModuleOperationState::Blocked,
483 actor_id: actor.actor_id.clone(),
484 outcome_code: code.to_owned(),
485 evidence_references: Vec::new(),
486 error: Some(ModuleOperationError {
487 code: code.to_owned(),
488 message: error.to_string(),
489 evidence_references: Vec::new(),
490 recorded_at: now,
491 }),
492 next_actions: vec![
493 format!("configure_effect_adapter:{}", effect.effect_id()),
494 "retry_operation".to_owned(),
495 ],
496 now,
497 })?)
498 }
499
500 fn engine(&self) -> ModuleManagementEngine<JsonFileModuleOperationStore> {
501 ModuleManagementEngine::new(JsonFileModuleOperationStore::new(
502 self.root.join(MANAGEMENT_ROOT),
503 ))
504 }
505
506 fn policy(&self) -> Result<ModuleEnvironmentPolicy, WorkspaceModuleOperatorError> {
507 let bytes = fs::read(self.root.join(POLICY_PATH)).map_err(|error| {
508 if error.kind() == std::io::ErrorKind::NotFound {
509 WorkspaceModuleOperatorError::PolicyUnavailable
510 } else {
511 error.into()
512 }
513 })?;
514 Ok(serde_json::from_slice(&bytes)?)
515 }
516
517 fn persist_plan(&self, plan: &ModuleChangePlan) -> Result<(), WorkspaceModuleOperatorError> {
518 let root = self.root.join(MANAGEMENT_ROOT).join("plans");
519 fs::create_dir_all(&root)?;
520 let path = root.join(format!("{}.json", plan.plan_digest));
521 let bytes = serde_json::to_vec_pretty(plan)?;
522 if path.exists() {
523 if fs::read(&path)? != bytes {
524 return Err(WorkspaceModuleOperatorError::StalePlan);
525 }
526 return Ok(());
527 }
528 let temporary = root.join(format!("{}.next.json", plan.plan_digest));
529 fs::write(&temporary, &bytes)?;
530 fs::rename(temporary, path)?;
531 Ok(())
532 }
533
534 fn load_plan(&self, digest: &str) -> Result<ModuleChangePlan, WorkspaceModuleOperatorError> {
535 Ok(serde_json::from_slice(&fs::read(
536 self.root
537 .join(MANAGEMENT_ROOT)
538 .join("plans")
539 .join(format!("{digest}.json")),
540 )?)?)
541 }
542
543 fn load_bound_plan(
544 &self,
545 operation_id: &str,
546 ) -> Result<ModuleChangePlan, WorkspaceModuleOperatorError> {
547 let operation = self.operation(operation_id)?;
548 let plan = self.load_plan(&operation.plan_digest)?;
549 if plan.plan_digest != operation.plan_digest {
550 return Err(WorkspaceModuleOperatorError::StalePlan);
551 }
552 Ok(plan)
553 }
554}
555
556pub const MANAGEMENT_HOLDER: &str = "module-management-api";
557
558fn transition(
559 engine: &ModuleManagementEngine<JsonFileModuleOperationStore>,
560 operation: ModuleOperation,
561 next_state: ModuleOperationState,
562 actor_id: &str,
563 outcome_code: &str,
564 now: DateTime<Utc>,
565) -> Result<ModuleOperation, ModuleManagementError> {
566 engine.advance(AdvanceModuleOperation {
567 operation_id: operation.operation_id,
568 expected_revision: operation.revision,
569 fencing_token: operation.fencing_token,
570 next_state,
571 actor_id: actor_id.to_owned(),
572 outcome_code: outcome_code.to_owned(),
573 evidence_references: Vec::new(),
574 error: None,
575 next_actions: Vec::new(),
576 now,
577 })
578}
579
580fn next_incomplete_state(
581 operation: &ModuleOperation,
582 plan: &ModuleChangePlan,
583) -> ModuleOperationState {
584 let completed = operation
585 .effect_receipts
586 .iter()
587 .map(|receipt| receipt.effect_id.as_str())
588 .collect::<BTreeSet<_>>();
589 for effect in &plan.effects {
590 if completed.contains(effect.effect_id()) {
591 continue;
592 }
593 return match effect {
594 ModulePlanEffect::Configuration { .. }
595 | ModulePlanEffect::Protected { .. }
596 | ModulePlanEffect::ConsoleComposition { .. }
597 | ModulePlanEffect::ServiceInstallation { .. }
598 | ModulePlanEffect::ServiceRemoval { .. }
599 | ModulePlanEffect::WorkspaceFile { .. } => ModuleOperationState::StagingConfiguration,
600 ModulePlanEffect::Migration { .. } => ModuleOperationState::Migrating,
601 ModulePlanEffect::Validate { .. } => ModuleOperationState::Verifying,
602 ModulePlanEffect::Restart { .. }
603 | ModulePlanEffect::ServiceRestart { .. }
604 | ModulePlanEffect::Activate { .. } => ModuleOperationState::Activating,
605 };
606 }
607 ModuleOperationState::Activating
608}
609
610fn operation_kind(change: &ModuleRootChange) -> ModuleOperationKind {
611 match change {
612 ModuleRootChange::Install { .. } => ModuleOperationKind::Install,
613 ModuleRootChange::Update { .. } | ModuleRootChange::SelectOptional { .. } => {
614 ModuleOperationKind::Update
615 }
616 ModuleRootChange::Uninstall { .. } => ModuleOperationKind::Uninstall,
617 ModuleRootChange::SwitchDelivery { .. } => ModuleOperationKind::DeliveryTransition,
618 ModuleRootChange::Restore { .. } => ModuleOperationKind::Restore,
619 ModuleRootChange::Repair { .. } => ModuleOperationKind::Repair,
620 }
621}
622
623fn content_id<T: serde::Serialize>(prefix: &str, value: &T) -> Result<String, serde_json::Error> {
624 let digest = digest_json(value)?;
625 Ok(format!("{prefix}-{}", &digest[7..31]))
626}
627
628#[cfg(test)]
629mod tests {
630 use super::*;
631 use crate::{
632 APPLICATION_MODULE_LOCK_PROTOCOL, ApplicationModuleLock,
633 DESIRED_MODULE_COMPOSITION_PROTOCOL, DesiredModuleComposition, EnvironmentManagementMode,
634 MODULE_CHANGE_PLAN_PROTOCOL, MODULE_ENVIRONMENT_POLICY_PROTOCOL, ModulePathPrecondition,
635 application_module_lock_digest, desired_composition_digest, module_change_plan_digest,
636 };
637 use chrono::TimeZone as _;
638 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
639
640 static NEXT_ROOT: AtomicU64 = AtomicU64::new(1);
641
642 #[derive(Debug)]
643 struct FakeAdapter(AtomicBool);
644
645 impl ModuleEffectAdapter for FakeAdapter {
646 fn execute(
647 &self,
648 _root: &Path,
649 _operation: &ModuleOperation,
650 effect: &ModulePlanEffect,
651 ) -> Result<ModuleEffectExecution, ModuleEffectAdapterError> {
652 if matches!(effect, ModulePlanEffect::Validate { .. })
653 && self.0.swap(false, Ordering::SeqCst)
654 {
655 return Err(ModuleEffectAdapterError::Failed {
656 effect_id: effect.effect_id().to_owned(),
657 reason: "injected failure".to_owned(),
658 });
659 }
660 Ok(ModuleEffectExecution {
661 outcome: match effect {
662 ModulePlanEffect::Validate { .. } => ModuleEffectOutcome::Verified,
663 ModulePlanEffect::Activate { .. } => ModuleEffectOutcome::Activated,
664 _ => ModuleEffectOutcome::Applied,
665 },
666 evidence_references: Vec::new(),
667 })
668 }
669 }
670
671 #[test]
672 fn operator_runs_all_remaining_phases_through_one_interface() {
673 let (operator, engine, operation, plan, actor, root) = fixture(false);
674 let completed = operator
675 .run_from_state(&engine, operation, &plan, &actor, now())
676 .unwrap();
677 assert_eq!(completed.state, ModuleOperationState::Succeeded);
678 assert_eq!(completed.effect_receipts.len(), 3);
679 fs::remove_dir_all(root).unwrap();
680 }
681
682 #[test]
683 fn retry_resumes_the_first_incomplete_phase_without_repeating_receipts() {
684 let (operator, engine, operation, plan, actor, root) = fixture(true);
685 let blocked = operator
686 .run_from_state(&engine, operation, &plan, &actor, now())
687 .unwrap();
688 assert_eq!(blocked.state, ModuleOperationState::Blocked);
689 assert!(blocked.effect_receipts.is_empty());
690 let completed = operator
691 .retry(&blocked.operation_id, blocked.revision, &actor, now())
692 .unwrap();
693 assert_eq!(completed.state, ModuleOperationState::Succeeded);
694 assert_eq!(completed.attempt, 2);
695 assert_eq!(completed.effect_receipts.len(), 3);
696 fs::remove_dir_all(root).unwrap();
697 }
698
699 #[allow(clippy::type_complexity)]
700 fn fixture(
701 fail_once: bool,
702 ) -> (
703 WorkspaceModuleOperator<FakeAdapter>,
704 ModuleManagementEngine<JsonFileModuleOperationStore>,
705 ModuleOperation,
706 ModuleChangePlan,
707 ManagementActor,
708 PathBuf,
709 ) {
710 let root = std::env::temp_dir().join(format!(
711 "lenso-module-operator-{}-{}",
712 std::process::id(),
713 NEXT_ROOT.fetch_add(1, Ordering::Relaxed)
714 ));
715 fs::create_dir_all(root.join(".lenso")).unwrap();
716 fs::write(
717 root.join(POLICY_PATH),
718 serde_json::to_vec_pretty(&policy()).unwrap(),
719 )
720 .unwrap();
721 let operator = WorkspaceModuleOperator::new(&root, FakeAdapter(AtomicBool::new(fail_once)));
722 let plan = plan();
723 operator.persist_plan(&plan).unwrap();
724 let actor = actor();
725 let engine = operator.engine();
726 let started = engine
727 .start(StartModuleOperation {
728 operation_id: "operation-1",
729 idempotency_key: "request-1",
730 operation_kind: ModuleOperationKind::Update,
731 plan: &plan,
732 policy: &policy(),
733 actor: &actor,
734 approvals: Vec::new(),
735 holder_id: MANAGEMENT_HOLDER,
736 now: now(),
737 })
738 .unwrap();
739 let applying = transition(
740 &engine,
741 started,
742 ModuleOperationState::ApplyingFiles,
743 &actor.actor_id,
744 "test_apply",
745 now(),
746 )
747 .unwrap();
748 let applied = transition(
749 &engine,
750 applying,
751 ModuleOperationState::FilesApplied,
752 &actor.actor_id,
753 "test_files",
754 now(),
755 )
756 .unwrap();
757 (operator, engine, applied, plan, actor, root)
758 }
759
760 fn plan() -> ModuleChangePlan {
761 let desired = DesiredModuleComposition {
762 protocol: DESIRED_MODULE_COMPOSITION_PROTOCOL.to_owned(),
763 application_id: "app-1".to_owned(),
764 revision: 2,
765 selected: Vec::new(),
766 local_overrides: Vec::new(),
767 };
768 let desired_digest = desired_composition_digest(&desired).unwrap();
769 let target_lock = ApplicationModuleLock {
770 protocol: APPLICATION_MODULE_LOCK_PROTOCOL.to_owned(),
771 application_id: "app-1".to_owned(),
772 desired_composition_digest: desired_digest.clone(),
773 catalog_snapshot_digest: digest('a'),
774 trust_policy_digest: digest('b'),
775 resolver_version: "resolver-1".to_owned(),
776 modules: Vec::new(),
777 capability_bindings: Vec::new(),
778 };
779 let target_lock_digest = application_module_lock_digest(&target_lock).unwrap();
780 let mut plan = ModuleChangePlan {
781 protocol: MODULE_CHANGE_PLAN_PROTOCOL.to_owned(),
782 plan_id: "plan-1".to_owned(),
783 plan_digest: String::new(),
784 application_id: "app-1".to_owned(),
785 environment_id: "local".to_owned(),
786 expected_target_revision: 1,
787 request: ModuleRootChange::Update {
788 module_id: "acme/example".to_owned(),
789 version_requirement: "^1".to_owned(),
790 },
791 current_desired_digest: digest('1'),
792 target_desired: desired,
793 target_desired_digest: desired_digest,
794 current_lock_digest: Some(digest('2')),
795 target_lock,
796 target_lock_digest: target_lock_digest.clone(),
797 catalog_snapshot_digest: digest('a'),
798 resolver_version: "resolver-1".to_owned(),
799 trust_policy_digest: digest('b'),
800 compatibility_evidence_digest: digest('c'),
801 cargo_lock_candidate: None,
802 read_set: Vec::<ModulePathPrecondition>::new(),
803 effects: vec![
804 ModulePlanEffect::Validate {
805 effect_id: "80-validate:test".to_owned(),
806 command: "cargo check --locked".to_owned(),
807 expected_evidence: digest('d'),
808 },
809 ModulePlanEffect::Activate {
810 effect_id: "90-activate:lock".to_owned(),
811 target_lock_digest,
812 },
813 ModulePlanEffect::Restart {
814 effect_id: "99-restart:host".to_owned(),
815 target: "host".to_owned(),
816 },
817 ],
818 approval_boundaries: Vec::new(),
819 validation_commands: vec!["cargo check --locked".to_owned()],
820 next_actions: vec!["review_plan".to_owned()],
821 created_at: now(),
822 };
823 plan.plan_digest = module_change_plan_digest(&plan).unwrap();
824 plan
825 }
826
827 fn policy() -> ModuleEnvironmentPolicy {
828 ModuleEnvironmentPolicy {
829 protocol: MODULE_ENVIRONMENT_POLICY_PROTOCOL.to_owned(),
830 policy_id: "local".to_owned(),
831 revision: "policy-1".to_owned(),
832 mode: EnvironmentManagementMode::Full,
833 require_distinct_approver: false,
834 maximum_approval_age_seconds: 3_600,
835 maximum_lease_seconds: 60,
836 require_backup_for_non_local_destructive_effects: true,
837 }
838 }
839
840 fn actor() -> ManagementActor {
841 ManagementActor {
842 actor_id: "user:operator".to_owned(),
843 verified_authorities: BTreeSet::from(["module.manage".to_owned()]),
844 }
845 }
846 fn now() -> DateTime<Utc> {
847 Utc.with_ymd_and_hms(2026, 7, 30, 12, 0, 0).unwrap()
848 }
849 fn digest(character: char) -> String {
850 format!("sha256:{}", character.to_string().repeat(64))
851 }
852}