1use std::borrow::Cow;
4use std::fmt;
5use std::fs;
6use std::marker::PhantomData;
7use std::path::Path;
8use std::path::PathBuf;
9use std::sync::Arc;
10use std::thread::available_parallelism;
11
12use anyhow::Context;
13use anyhow::Result;
14use anyhow::anyhow;
15use anyhow::bail;
16use anyhow::ensure;
17use bytesize::ByteSize;
18use codespan_reporting::files::SimpleFile;
19use codespan_reporting::term;
20use codespan_reporting::term::termcolor::Buffer;
21use indexmap::IndexMap;
22use rowan::GreenNode;
23use secrecy::ExposeSecret;
24use tokio::process::Command;
25use toml_spanner::Arena;
26use toml_spanner::ErrorKind;
27use toml_spanner::Failed;
28use toml_spanner::FromToml;
29use toml_spanner::Item;
30use toml_spanner::Table;
31use toml_spanner::ToToml;
32use toml_spanner::ToTomlError;
33use toml_spanner::Toml;
34use toml_spanner::helper::display;
35use toml_spanner::helper::flatten_any;
36use toml_spanner::helper::parse_string;
37use tracing::error;
38use tracing::warn;
39use url::Url;
40use wdl_analysis::DiagnosticsConfig;
41use wdl_analysis::diagnostics::unknown_name;
42use wdl_analysis::diagnostics::unknown_type;
43use wdl_analysis::document::Task;
44use wdl_analysis::types::PrimitiveType;
45use wdl_analysis::types::Type;
46use wdl_analysis::types::v1::ExprTypeEvaluator;
47use wdl_ast::AstNode;
48use wdl_ast::Diagnostic;
49use wdl_ast::Span;
50use wdl_ast::SupportedVersion;
51use wdl_ast::lexer::Lexer;
52use wdl_ast::v1::Expr;
53use wdl_grammar::construct_tree;
54use wdl_grammar::grammar::v1;
55use wdl_grammar::grammar::v1::Parser;
56
57use crate::CancellationContext;
58use crate::EvaluationContext;
59use crate::EvaluationPath;
60use crate::Events;
61use crate::NoneValue;
62use crate::Object;
63use crate::SYSTEM;
64use crate::Value;
65use crate::backend::ExecuteTaskRequest;
66use crate::backend::TaskExecutionBackend;
67use crate::convert_unit_string;
68use crate::diagnostics::unknown_enum_choice;
69use crate::http::Transferer;
70use crate::path::is_supported_url;
71use crate::tree::SyntaxNode;
72use crate::v1::DEFAULT_TASK_REQUIREMENT_MAX_RETRIES;
73use crate::v1::ExprEvaluator;
74
75pub(crate) const MAX_RETRIES: u64 = 100;
77
78pub(crate) const DEFAULT_TASK_SHELL: &str = "bash";
80
81pub(crate) const DEFAULT_TASK_CONTAINER: &str = "ubuntu:latest";
83
84const DEFAULT_BACKEND_NAME: &str = "default";
86
87const MAX_LSF_JOB_NAME_PREFIX: usize = 100;
89
90const REDACTED: &str = "<REDACTED>";
92
93const CACHE_DIR_SENTINEL: &str = "system";
95
96const DEFAULT_HTTP_RETRIES: u32 = 5;
100
101const DEFAULT_APPTAINER_EXECUTABLE: &str = "apptainer";
103
104const DEFAULT_SCATTER_CONCURRENCY: u64 = 1000;
107
108pub(crate) fn cache_dir() -> Result<PathBuf> {
110 const CACHE_DIR_ROOT: &str = "sprocket";
112
113 Ok(dirs::cache_dir()
114 .context("failed to determine user cache directory")?
115 .join(CACHE_DIR_ROOT))
116}
117
118fn escape_mapping(toml: &str) -> Vec<(usize, usize)> {
149 let mut iter = toml.char_indices();
150 let mut mapping = Vec::new();
151 let mut new = 0;
152 while let Some((old, c)) = iter.next() {
153 if c != '\\' {
154 new += c.len_utf8();
155 continue;
156 }
157
158 match iter.next() {
159 Some((_, 'u')) => {
160 let c = u32::from_str_radix(&toml[old + 2 ..old + 6 ], 16)
161 .map(char::from_u32)
162 .expect("invalid TOML escape sequence")
163 .expect("invalid TOML escape character");
164 new += c.len_utf8();
165
166 iter.nth(3);
168 mapping.push((new, old + 6 ));
169 }
170 Some((_, 'U')) => {
171 let c = u32::from_str_radix(&toml[old + 2 ..old + 10 ], 16)
172 .map(char::from_u32)
173 .expect("invalid TOML escape sequence")
174 .expect("invalid TOML escape character");
175 new += c.len_utf8();
176
177 iter.nth(7);
179 mapping.push((new, old + 10 ));
180 }
181 Some(_) => {
182 new += 1;
184 mapping.push((new, old + 2 ));
185 }
186 None => break,
187 }
188 }
189
190 mapping
191}
192
193#[derive(Default, Debug, Clone)]
197pub struct SecretString {
198 inner: secrecy::SecretString,
202 redacted: bool,
211}
212
213impl SecretString {
214 pub fn redact(mut self) -> Self {
219 self.redacted = true;
220 self
221 }
222
223 pub fn inner(&self) -> &secrecy::SecretString {
225 &self.inner
226 }
227}
228
229impl From<String> for SecretString {
230 fn from(s: String) -> Self {
231 Self {
232 inner: s.into(),
233 redacted: false,
234 }
235 }
236}
237
238impl From<&str> for SecretString {
239 fn from(s: &str) -> Self {
240 Self {
241 inner: s.into(),
242 redacted: false,
243 }
244 }
245}
246
247impl<'de> FromToml<'de> for SecretString {
248 fn from_toml(ctx: &mut toml_spanner::Context<'de>, item: &Item<'de>) -> Result<Self, Failed> {
249 Ok(String::from_toml(ctx, item)?.into())
250 }
251}
252
253impl ToToml for SecretString {
254 fn to_toml<'a>(&'a self, arena: &'a Arena) -> Result<Item<'a>, ToTomlError> {
255 use secrecy::ExposeSecret;
256
257 if self.redacted {
258 REDACTED.to_toml(arena)
259 } else {
260 self.inner.expose_secret().to_toml(arena)
261 }
262 }
263}
264
265impl PartialEq for SecretString {
266 fn eq(&self, other: &Self) -> bool {
267 use secrecy::ExposeSecret;
268
269 self.inner.expose_secret() == other.inner.expose_secret()
271 }
272}
273
274impl Eq for SecretString {}
275
276#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Hash, Toml)]
279#[toml(Toml, rename_all = "snake_case")]
280pub enum FailureMode {
281 #[default]
284 Slow,
285 Fast,
289}
290
291mod index_map {
293 use indexmap::IndexMap;
294 use toml_spanner::Arena;
295 use toml_spanner::Context;
296 use toml_spanner::Failed;
297 use toml_spanner::FromToml;
298 use toml_spanner::Item;
299 use toml_spanner::Key;
300 use toml_spanner::Table;
301 use toml_spanner::TableStyle;
302 use toml_spanner::ToToml;
303 use toml_spanner::ToTomlError;
304
305 pub fn from_toml<'de, V>(
307 ctx: &mut Context<'de>,
308 item: &Item<'de>,
309 ) -> Result<IndexMap<String, V>, Failed>
310 where
311 V: FromToml<'de>,
312 {
313 let table = item.require_table(ctx)?;
314 let mut map = IndexMap::default();
315 let mut had_error = false;
316 for (key, item) in table {
317 match V::from_toml(ctx, item) {
318 Ok(v) => {
319 map.insert(key.name.into(), v);
320 }
321 Err(_) => had_error = true,
322 }
323 }
324
325 if had_error { Err(Failed) } else { Ok(map) }
326 }
327
328 pub fn to_toml<'a, V>(
330 value: &'a IndexMap<String, V>,
331 arena: &'a Arena,
332 ) -> Result<Item<'a>, ToTomlError>
333 where
334 V: ToToml,
335 {
336 let Some(mut table) = Table::try_with_capacity(value.len(), arena) else {
337 return Err(ToTomlError::from(
338 "length of table exceeded maximum capacity",
339 ));
340 };
341
342 table.set_style(TableStyle::Implicit);
343
344 for (k, v) in value {
345 table.insert_unique(Key::new(k), v.to_toml(arena)?, arena);
346 }
347
348 Ok(table.into_item())
349 }
350}
351
352#[derive(Debug, Clone, Toml, PartialEq, Eq)]
363#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
364pub struct Config {
365 #[toml(default, style = Header)]
367 pub http: HttpConfig,
368 #[toml(default, style = Header)]
370 pub workflow: WorkflowConfig,
371 #[toml(default, style = Header)]
373 pub task: TaskConfig,
374 #[toml(default = DEFAULT_BACKEND_NAME.into())]
376 pub backend: String,
377 #[toml(default, with = index_map)]
382 pub backends: IndexMap<String, BackendConfig>,
383 #[toml(default, style = Header)]
385 pub storage: StorageConfig,
386 #[toml(default)]
400 pub suppress_env_specific_output: bool,
401 #[toml(default)]
408 pub experimental_features_enabled: bool,
409 #[toml(default, rename = "fail")]
417 pub failure_mode: FailureMode,
418}
419
420impl Default for Config {
421 fn default() -> Self {
422 Self {
423 http: Default::default(),
424 workflow: Default::default(),
425 task: Default::default(),
426 backend: DEFAULT_BACKEND_NAME.into(),
427 backends: Default::default(),
428 storage: Default::default(),
429 suppress_env_specific_output: Default::default(),
430 experimental_features_enabled: Default::default(),
431 failure_mode: Default::default(),
432 }
433 }
434}
435
436impl Config {
437 pub fn builder() -> ConfigBuilder<Self> {
439 ConfigBuilder::default()
440 }
441
442 pub async fn validate(&self) -> Result<()> {
444 self.http.validate()?;
445 self.workflow.validate()?;
446 self.task.validate()?;
447
448 if self.backends.is_empty() && self.backend == DEFAULT_BACKEND_NAME {
449 } else {
451 let backend = &self.backend;
452 if !self.backends.contains_key(backend) {
453 bail!("a backend named `{backend}` is not present in the configuration");
454 }
455 }
456
457 for backend in self.backends.values() {
458 backend.validate().await?;
459 }
460
461 self.storage.validate()?;
462
463 if self.suppress_env_specific_output && !self.experimental_features_enabled {
464 bail!("`suppress_env_specific_output` requires enabling experimental features");
465 }
466
467 Ok(())
468 }
469
470 pub fn redact(mut self) -> Self {
474 for backend in self.backends.values_mut() {
475 *backend = std::mem::take(backend).redact();
476 }
477
478 if let Some(auth) = self.storage.azure.auth.take() {
479 self.storage.azure.auth = Some(auth.redact());
480 }
481
482 if let Some(auth) = self.storage.s3.auth.take() {
483 self.storage.s3.auth = Some(auth.redact());
484 }
485
486 if let Some(auth) = self.storage.google.auth.take() {
487 self.storage.google.auth = Some(auth.redact());
488 }
489
490 self
491 }
492
493 pub fn backend(&self) -> Result<Cow<'_, BackendConfig>> {
498 if !self.backends.is_empty() {
499 let backend = &self.backend;
500 return Ok(Cow::Borrowed(self.backends.get(backend).ok_or_else(
501 || anyhow!("a backend named `{backend}` is not present in the configuration"),
502 )?));
503 }
504 Ok(Cow::Owned(BackendConfig::default()))
506 }
507
508 pub(crate) async fn create_backend(
510 self: &Arc<Self>,
511 run_root_dir: &Path,
512 events: Events,
513 cancellation: CancellationContext,
514 ) -> Result<Arc<dyn TaskExecutionBackend>> {
515 use crate::backend::*;
516
517 match self.backend()?.as_ref() {
518 BackendConfig::Local { .. } => {
519 warn!(
520 "the engine is configured to use the local backend: tasks will not be run \
521 inside of a container"
522 );
523 Ok(Arc::new(LocalBackend::new(
524 self.clone(),
525 events,
526 cancellation,
527 )?))
528 }
529 BackendConfig::Docker { .. } => Ok(Arc::new(
530 DockerBackend::new(self.clone(), events, cancellation).await?,
531 )),
532 BackendConfig::Tes { .. } => Ok(Arc::new(
533 TesBackend::new(self.clone(), events, cancellation).await?,
534 )),
535 BackendConfig::LsfApptainer { .. } => Ok(Arc::new(LsfApptainerBackend::new(
536 self.clone(),
537 run_root_dir,
538 events,
539 cancellation,
540 )?)),
541 BackendConfig::SlurmApptainer { .. } => Ok(Arc::new(SlurmApptainerBackend::new(
542 self.clone(),
543 run_root_dir,
544 events,
545 cancellation,
546 )?)),
547 }
548 }
549}
550
551#[derive(Copy, Clone, Default, Debug, PartialEq, Eq)]
553pub enum Parallelism {
554 #[default]
556 Available,
557 Use(usize),
559}
560
561impl From<usize> for Parallelism {
562 fn from(value: usize) -> Self {
563 Self::Use(value)
564 }
565}
566
567impl From<Parallelism> for usize {
568 fn from(value: Parallelism) -> Self {
569 match value {
570 Parallelism::Available => available_parallelism().map(Into::into).unwrap_or(1),
571 Parallelism::Use(value) => value,
572 }
573 }
574}
575
576impl<'de> FromToml<'de> for Parallelism {
577 fn from_toml(ctx: &mut toml_spanner::Context<'de>, item: &Item<'de>) -> Result<Self, Failed> {
578 if let Some("available") = item.as_str() {
579 return Ok(Self::Available);
580 }
581
582 if let Some(n) = item.as_u64().and_then(|n| usize::try_from(n).ok())
583 && n > 0
584 {
585 return Ok(Self::Use(n));
586 }
587
588 Err(ctx.report_custom_error(
589 "expected a positive integer or `available` for parallelism",
590 item,
591 ))
592 }
593}
594
595impl ToToml for Parallelism {
596 fn to_toml<'a>(&'a self, _: &'a Arena) -> Result<Item<'a>, ToTomlError> {
597 match self {
598 Self::Available => Ok(Item::string("available")),
599 Self::Use(n) => Ok(i64::try_from(*n)
600 .map_err(|e| ToTomlError {
601 message: format!("invalid parallelism: {e}").into(),
602 })?
603 .into()),
604 }
605 }
606}
607
608#[derive(Debug, Clone, Toml, PartialEq, Eq)]
610#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
611pub struct HttpConfig {
612 #[toml(default = CACHE_DIR_SENTINEL.into())]
616 pub cache_dir: String,
617 #[toml(default = DEFAULT_HTTP_RETRIES)]
619 pub retries: u32,
620 #[toml(default)]
624 pub parallelism: Parallelism,
625 #[toml(default, FromToml with = parse_string, ToToml with = display)]
630 pub hash_algorithm: cloud_copy::HashAlgorithm,
631}
632
633impl Default for HttpConfig {
634 fn default() -> Self {
635 Self {
636 cache_dir: CACHE_DIR_SENTINEL.into(),
637 retries: DEFAULT_HTTP_RETRIES,
638 parallelism: Default::default(),
639 hash_algorithm: Default::default(),
640 }
641 }
642}
643
644impl HttpConfig {
645 pub fn validate(&self) -> Result<()> {
647 if let Parallelism::Use(parallelism) = self.parallelism
648 && parallelism == 0
649 {
650 bail!("configuration value `http.parallelism` cannot be zero");
651 }
652 Ok(())
653 }
654
655 pub fn cache_dir(&self) -> Result<PathBuf> {
657 const DOWNLOADS_CACHE_SUBDIR: &str = "downloads";
658
659 if self.using_system_cache_dir() {
660 cache_dir().map(|d| d.join(DOWNLOADS_CACHE_SUBDIR))
661 } else {
662 Ok(PathBuf::from(&self.cache_dir))
663 }
664 }
665
666 pub fn using_system_cache_dir(&self) -> bool {
668 self.cache_dir == CACHE_DIR_SENTINEL
669 }
670}
671
672#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
674#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
675pub struct StorageConfig {
676 #[toml(default, style = Header)]
678 pub azure: AzureStorageConfig,
679 #[toml(default, style = Header)]
681 pub s3: S3StorageConfig,
682 #[toml(default, style = Header)]
684 pub google: GoogleStorageConfig,
685}
686
687impl StorageConfig {
688 pub fn validate(&self) -> Result<()> {
690 self.azure.validate()?;
691 self.s3.validate()?;
692 self.google.validate()?;
693 Ok(())
694 }
695}
696
697#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
699#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
700pub struct AzureStorageAuthConfig {
701 pub account_name: String,
703 pub access_key: SecretString,
705}
706
707impl AzureStorageAuthConfig {
708 pub fn validate(&self) -> Result<()> {
710 if self.account_name.is_empty() {
711 bail!("configuration value `storage.azure.auth.account_name` is required");
712 }
713
714 if self.access_key.inner.expose_secret().is_empty() {
715 bail!("configuration value `storage.azure.auth.access_key` is required");
716 }
717
718 Ok(())
719 }
720
721 pub fn redact(mut self) -> Self {
724 self.access_key = self.access_key.redact();
725 self
726 }
727}
728
729#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
731#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
732pub struct AzureStorageConfig {
733 #[toml(style = Header)]
735 pub auth: Option<AzureStorageAuthConfig>,
736}
737
738impl AzureStorageConfig {
739 pub fn validate(&self) -> Result<()> {
741 if let Some(auth) = &self.auth {
742 auth.validate()?;
743 }
744
745 Ok(())
746 }
747}
748
749#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
751#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
752pub struct S3StorageAuthConfig {
753 pub access_key_id: String,
755 pub secret_access_key: SecretString,
757}
758
759impl S3StorageAuthConfig {
760 pub fn validate(&self) -> Result<()> {
762 if self.access_key_id.is_empty() {
763 bail!("configuration value `storage.s3.auth.access_key_id` is required");
764 }
765
766 if self.secret_access_key.inner.expose_secret().is_empty() {
767 bail!("configuration value `storage.s3.auth.secret_access_key` is required");
768 }
769
770 Ok(())
771 }
772
773 pub fn redact(mut self) -> Self {
776 self.secret_access_key = self.secret_access_key.redact();
777 self
778 }
779}
780
781#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
783#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
784pub struct S3StorageConfig {
785 pub region: Option<String>,
790
791 #[toml(style = Header)]
793 pub auth: Option<S3StorageAuthConfig>,
794}
795
796impl S3StorageConfig {
797 pub fn validate(&self) -> Result<()> {
799 if let Some(auth) = &self.auth {
800 auth.validate()?;
801 }
802
803 Ok(())
804 }
805}
806
807#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
809#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
810pub struct GoogleStorageAuthConfig {
811 pub access_key: String,
813 pub secret: SecretString,
815}
816
817impl GoogleStorageAuthConfig {
818 pub fn validate(&self) -> Result<()> {
820 if self.access_key.is_empty() {
821 bail!("configuration value `storage.google.auth.access_key` is required");
822 }
823
824 if self.secret.inner.expose_secret().is_empty() {
825 bail!("configuration value `storage.google.auth.secret` is required");
826 }
827
828 Ok(())
829 }
830
831 pub fn redact(mut self) -> Self {
834 self.secret = self.secret.redact();
835 self
836 }
837}
838
839#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
841#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
842pub struct GoogleStorageConfig {
843 #[toml(style = Header)]
845 pub auth: Option<GoogleStorageAuthConfig>,
846}
847
848impl GoogleStorageConfig {
849 pub fn validate(&self) -> Result<()> {
851 if let Some(auth) = &self.auth {
852 auth.validate()?;
853 }
854
855 Ok(())
856 }
857}
858
859#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
861#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
862pub struct WorkflowConfig {
863 #[toml(default, style = Header)]
865 pub scatter: ScatterConfig,
866}
867
868impl WorkflowConfig {
869 pub fn validate(&self) -> Result<()> {
871 self.scatter.validate()?;
872 Ok(())
873 }
874}
875
876#[derive(Debug, Clone, PartialEq, Eq, Toml)]
878#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
879pub struct ScatterConfig {
880 #[toml(default = DEFAULT_SCATTER_CONCURRENCY)]
933 pub concurrency: u64,
934}
935
936impl Default for ScatterConfig {
937 fn default() -> Self {
938 Self {
939 concurrency: DEFAULT_SCATTER_CONCURRENCY,
940 }
941 }
942}
943
944impl ScatterConfig {
945 pub fn validate(&self) -> Result<()> {
947 if self.concurrency == 0 {
948 bail!("configuration value `workflow.scatter.concurrency` cannot be zero");
949 }
950
951 Ok(())
952 }
953}
954
955#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Hash, Toml)]
957#[toml(Toml, rename_all = "snake_case")]
958pub enum CallCachingMode {
959 #[default]
966 Off,
967 On,
973 Explicit,
981}
982
983#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Hash, Toml)]
985#[toml(Toml, rename_all = "snake_case")]
986pub enum ContentDigestMode {
987 Strong,
994 #[default]
1005 Weak,
1006}
1007
1008#[derive(Debug, Default, Copy, Clone, PartialEq, Eq)]
1010pub enum Retries {
1011 #[default]
1013 Default,
1014 Use(u64),
1016}
1017
1018impl From<u64> for Retries {
1019 fn from(value: u64) -> Self {
1020 Self::Use(value)
1021 }
1022}
1023
1024impl From<Retries> for u64 {
1025 fn from(value: Retries) -> Self {
1026 match value {
1027 Retries::Default => DEFAULT_TASK_REQUIREMENT_MAX_RETRIES,
1028 Retries::Use(value) => value,
1029 }
1030 }
1031}
1032
1033impl<'de> FromToml<'de> for Retries {
1034 fn from_toml(ctx: &mut toml_spanner::Context<'de>, item: &Item<'de>) -> Result<Self, Failed> {
1035 if let Some("default") = item.as_str() {
1036 return Ok(Self::Default);
1037 }
1038
1039 if let Some(n) = item.as_u64()
1040 && n < MAX_RETRIES
1041 {
1042 return Ok(Self::Use(n));
1043 }
1044
1045 Err(ctx.report_custom_error(
1046 format!("expected an integer less than {MAX_RETRIES} or `default` for retries"),
1047 item,
1048 ))
1049 }
1050}
1051
1052impl ToToml for Retries {
1053 fn to_toml<'a>(&'a self, _: &'a Arena) -> Result<Item<'a>, ToTomlError> {
1054 match self {
1055 Self::Default => Ok(Item::string("default")),
1056 Self::Use(n) => Ok(i64::try_from(*n)
1057 .map_err(|e| ToTomlError {
1058 message: format!("invalid retries: {e}").into(),
1059 })?
1060 .into()),
1061 }
1062 }
1063}
1064
1065#[derive(Debug, Clone, Toml, PartialEq, Eq)]
1067#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1068pub struct TaskConfig {
1069 #[toml(default)]
1073 pub retries: Retries,
1074 #[toml(default = DEFAULT_TASK_CONTAINER.into())]
1077 pub container: String,
1078 #[toml(default = DEFAULT_TASK_SHELL.into())]
1093 pub shell: String,
1094 #[toml(default)]
1096 pub cpu_limit_behavior: TaskResourceLimitBehavior,
1097 #[toml(default)]
1099 pub memory_limit_behavior: TaskResourceLimitBehavior,
1100 #[toml(default = CACHE_DIR_SENTINEL.into())]
1104 pub cache_dir: String,
1105 #[toml(default)]
1107 pub cache: CallCachingMode,
1108 #[toml(default)]
1112 pub digests: ContentDigestMode,
1113 #[toml(default)]
1122 pub excluded_cache_requirements: Vec<String>,
1123 #[toml(default)]
1131 pub excluded_cache_hints: Vec<String>,
1132 #[toml(default)]
1140 pub excluded_cache_inputs: Vec<String>,
1141}
1142
1143impl Default for TaskConfig {
1144 fn default() -> Self {
1145 Self {
1146 retries: Default::default(),
1147 container: DEFAULT_TASK_CONTAINER.into(),
1148 shell: DEFAULT_TASK_SHELL.into(),
1149 cpu_limit_behavior: Default::default(),
1150 memory_limit_behavior: Default::default(),
1151 cache_dir: CACHE_DIR_SENTINEL.into(),
1152 cache: Default::default(),
1153 digests: Default::default(),
1154 excluded_cache_requirements: Default::default(),
1155 excluded_cache_hints: Default::default(),
1156 excluded_cache_inputs: Default::default(),
1157 }
1158 }
1159}
1160
1161impl TaskConfig {
1162 pub fn validate(&self) -> Result<()> {
1164 if let Retries::Use(value) = self.retries
1165 && value >= MAX_RETRIES
1166 {
1167 bail!("configuration value `task.retries` cannot exceed {MAX_RETRIES}");
1168 }
1169
1170 Ok(())
1171 }
1172
1173 pub fn cache_dir(&self) -> Option<PathBuf> {
1175 if self.cache_dir == CACHE_DIR_SENTINEL {
1176 None
1177 } else {
1178 Some(PathBuf::from(&self.cache_dir))
1179 }
1180 }
1181}
1182
1183#[derive(Debug, Default, Copy, Clone, PartialEq, Eq, Toml)]
1186#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1187pub enum TaskResourceLimitBehavior {
1188 TryWithMax,
1191 #[default]
1195 Deny,
1196}
1197
1198#[derive(Debug, Clone, Toml, PartialEq, Eq)]
1200#[toml(Toml, rename_all = "snake_case", tag = "type")]
1201pub enum BackendConfig {
1202 Local {
1204 #[toml(default, style = Header, flatten, with = flatten_any)]
1206 config: LocalBackendConfig,
1207 },
1208 Docker {
1210 #[toml(default, style = Header, flatten, with = flatten_any)]
1212 config: DockerBackendConfig,
1213 },
1214 Tes {
1216 #[toml(default, style = Header, flatten, with = flatten_any)]
1218 config: TesBackendConfig,
1219 },
1220 LsfApptainer {
1224 #[toml(default, style = Header, flatten, with = flatten_any)]
1226 config: LsfApptainerBackendConfig,
1227 },
1228 SlurmApptainer {
1232 #[toml(default, style = Header, flatten, with = flatten_any)]
1234 config: SlurmApptainerBackendConfig,
1235 },
1236}
1237
1238impl Default for BackendConfig {
1239 fn default() -> Self {
1240 Self::Docker {
1241 config: Default::default(),
1242 }
1243 }
1244}
1245
1246impl BackendConfig {
1247 pub async fn validate(&self) -> Result<()> {
1249 match self {
1250 Self::Local { config } => config.validate(),
1251 Self::Docker { config } => config.validate(),
1252 Self::Tes { config } => config.validate(),
1253 Self::LsfApptainer { config } => config.validate().await,
1254 Self::SlurmApptainer { config } => config.validate().await,
1255 }
1256 }
1257
1258 pub fn as_local(&self) -> Option<&LocalBackendConfig> {
1262 match self {
1263 Self::Local { config } => Some(config),
1264 _ => None,
1265 }
1266 }
1267
1268 pub fn as_docker(&self) -> Option<&DockerBackendConfig> {
1272 match self {
1273 Self::Docker { config } => Some(config),
1274 _ => None,
1275 }
1276 }
1277
1278 pub fn as_tes(&self) -> Option<&TesBackendConfig> {
1282 match self {
1283 Self::Tes { config } => Some(config),
1284 _ => None,
1285 }
1286 }
1287
1288 pub fn as_lsf_apptainer(&self) -> Option<&LsfApptainerBackendConfig> {
1293 match self {
1294 Self::LsfApptainer { config } => Some(config),
1295 _ => None,
1296 }
1297 }
1298
1299 pub fn as_slurm_apptainer(&self) -> Option<&SlurmApptainerBackendConfig> {
1304 match self {
1305 Self::SlurmApptainer { config } => Some(config),
1306 _ => None,
1307 }
1308 }
1309
1310 pub fn redact(self) -> Self {
1312 match self {
1313 Self::Local { .. }
1314 | Self::Docker { .. }
1315 | Self::LsfApptainer { .. }
1316 | Self::SlurmApptainer { .. } => self,
1317 Self::Tes { config } => Self::Tes {
1318 config: config.redact(),
1319 },
1320 }
1321 }
1322}
1323
1324impl From<LocalBackendConfig> for BackendConfig {
1325 fn from(config: LocalBackendConfig) -> Self {
1326 Self::Local { config }
1327 }
1328}
1329
1330impl From<DockerBackendConfig> for BackendConfig {
1331 fn from(config: DockerBackendConfig) -> Self {
1332 Self::Docker { config }
1333 }
1334}
1335
1336impl From<TesBackendConfig> for BackendConfig {
1337 fn from(config: TesBackendConfig) -> Self {
1338 Self::Tes { config }
1339 }
1340}
1341
1342impl From<LsfApptainerBackendConfig> for BackendConfig {
1343 fn from(config: LsfApptainerBackendConfig) -> Self {
1344 Self::LsfApptainer { config }
1345 }
1346}
1347
1348impl From<SlurmApptainerBackendConfig> for BackendConfig {
1349 fn from(config: SlurmApptainerBackendConfig) -> Self {
1350 Self::SlurmApptainer { config }
1351 }
1352}
1353
1354#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
1361#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1362pub struct LocalBackendConfig {
1363 pub cpu: Option<u64>,
1369
1370 pub memory: Option<String>,
1377}
1378
1379impl LocalBackendConfig {
1380 pub fn validate(&self) -> Result<()> {
1382 if let Some(cpu) = self.cpu {
1383 if cpu == 0 {
1384 bail!("local backend configuration value `cpu` cannot be zero");
1385 }
1386
1387 let total = SYSTEM.cpus().len() as u64;
1388 if cpu > total {
1389 bail!(
1390 "local backend configuration value `cpu` cannot exceed the virtual CPUs \
1391 available to the host ({total})"
1392 );
1393 }
1394 }
1395
1396 if let Some(memory) = &self.memory {
1397 let memory = convert_unit_string(memory).with_context(|| {
1398 format!("local backend configuration value `memory` has invalid value `{memory}`")
1399 })?;
1400
1401 if memory == 0 {
1402 bail!("local backend configuration value `memory` cannot be zero");
1403 }
1404
1405 let total = SYSTEM.total_memory();
1406 if memory > total {
1407 bail!(
1408 "local backend configuration value `memory` cannot exceed the total memory of \
1409 the host ({total} bytes)"
1410 );
1411 }
1412 }
1413
1414 Ok(())
1415 }
1416}
1417
1418#[derive(Debug, Clone, PartialEq, Eq, Toml)]
1420#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1421pub struct DockerBackendConfig {
1422 #[toml(default = true)]
1426 pub cleanup: bool,
1427}
1428
1429impl DockerBackendConfig {
1430 pub fn validate(&self) -> Result<()> {
1432 Ok(())
1433 }
1434}
1435
1436impl Default for DockerBackendConfig {
1437 fn default() -> Self {
1438 Self { cleanup: true }
1439 }
1440}
1441
1442#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
1444#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1445pub struct BasicAuthConfig {
1446 pub username: String,
1448 pub password: SecretString,
1450}
1451
1452impl BasicAuthConfig {
1453 pub fn validate(&self) -> Result<()> {
1455 Ok(())
1456 }
1457
1458 pub fn redact(mut self) -> Self {
1460 self.password = self.password.redact();
1461 self
1462 }
1463}
1464
1465#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
1467#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1468pub struct BearerAuthConfig {
1469 pub token: SecretString,
1471}
1472
1473impl BearerAuthConfig {
1474 pub fn validate(&self) -> Result<()> {
1476 Ok(())
1477 }
1478
1479 pub fn redact(mut self) -> Self {
1481 self.token = self.token.redact();
1482 self
1483 }
1484}
1485
1486#[derive(Debug, Clone, PartialEq, Eq, Toml)]
1488#[toml(Toml, rename_all = "snake_case", tag = "type")]
1489pub enum TesBackendAuthConfig {
1490 Basic {
1492 #[toml(default, style = Header, flatten, with = flatten_any)]
1494 config: BasicAuthConfig,
1495 },
1496 Bearer {
1498 #[toml(default, style = Header, flatten, with = flatten_any)]
1500 config: BearerAuthConfig,
1501 },
1502}
1503
1504impl TesBackendAuthConfig {
1505 pub fn validate(&self) -> Result<()> {
1507 match self {
1508 Self::Basic { config } => config.validate(),
1509 Self::Bearer { config } => config.validate(),
1510 }
1511 }
1512
1513 pub fn redact(self) -> Self {
1516 match self {
1517 Self::Basic { config } => Self::Basic {
1518 config: config.redact(),
1519 },
1520 Self::Bearer { config } => Self::Bearer {
1521 config: config.redact(),
1522 },
1523 }
1524 }
1525}
1526
1527impl From<BasicAuthConfig> for TesBackendAuthConfig {
1528 fn from(config: BasicAuthConfig) -> Self {
1529 Self::Basic { config }
1530 }
1531}
1532
1533impl From<BearerAuthConfig> for TesBackendAuthConfig {
1534 fn from(config: BearerAuthConfig) -> Self {
1535 Self::Bearer { config }
1536 }
1537}
1538
1539#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
1541#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1542pub struct TesBackendConfig {
1543 #[toml(FromToml with = parse_string, ToToml with = display)]
1545 pub url: Option<Url>,
1546
1547 #[toml(style = Header)]
1549 pub auth: Option<TesBackendAuthConfig>,
1550
1551 #[toml(FromToml with = parse_string, ToToml with = display)]
1553 pub inputs: Option<Url>,
1554
1555 #[toml(FromToml with = parse_string, ToToml with = display)]
1557 pub outputs: Option<Url>,
1558
1559 pub interval: Option<u64>,
1563
1564 pub retries: Option<u32>,
1569
1570 pub max_concurrency: Option<u32>,
1575
1576 #[toml(default)]
1579 pub insecure: bool,
1580}
1581
1582impl TesBackendConfig {
1583 pub fn validate(&self) -> Result<()> {
1585 match &self.url {
1586 Some(url) => {
1587 if !self.insecure && url.scheme() != "https" {
1588 bail!(
1589 "TES backend configuration value `url` has invalid value `{url}`: URL \
1590 must use a HTTPS scheme"
1591 );
1592 }
1593 }
1594 None => bail!("TES backend configuration value `url` is required"),
1595 }
1596
1597 if let Some(auth) = &self.auth {
1598 auth.validate()?;
1599 }
1600
1601 if let Some(max_concurrency) = self.max_concurrency
1602 && max_concurrency == 0
1603 {
1604 bail!("TES backend configuration value `max_concurrency` cannot be zero");
1605 }
1606
1607 match &self.inputs {
1608 Some(url) => {
1609 if !is_supported_url(url.as_str()) {
1610 bail!(
1611 "TES backend storage configuration value `inputs` has invalid value \
1612 `{url}`: URL scheme is not supported"
1613 );
1614 }
1615
1616 if !url.path().ends_with('/') {
1617 bail!(
1618 "TES backend storage configuration value `inputs` has invalid value \
1619 `{url}`: URL path must end with a slash"
1620 );
1621 }
1622 }
1623 None => bail!("TES backend configuration value `inputs` is required"),
1624 }
1625
1626 match &self.outputs {
1627 Some(url) => {
1628 if !is_supported_url(url.as_str()) {
1629 bail!(
1630 "TES backend storage configuration value `outputs` has invalid value \
1631 `{url}`: URL scheme is not supported"
1632 );
1633 }
1634
1635 if !url.path().ends_with('/') {
1636 bail!(
1637 "TES backend storage configuration value `outputs` has invalid value \
1638 `{url}`: URL path must end with a slash"
1639 );
1640 }
1641 }
1642 None => bail!("TES backend storage configuration value `outputs` is required"),
1643 }
1644
1645 Ok(())
1646 }
1647
1648 pub fn redact(mut self) -> Self {
1650 if let Some(auth) = self.auth.take() {
1651 self.auth = Some(auth.redact());
1652 }
1653
1654 self
1655 }
1656}
1657
1658#[derive(Debug, Clone, PartialEq, Eq, Toml)]
1660#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
1661pub struct ApptainerConfig {
1662 #[toml(default = DEFAULT_APPTAINER_EXECUTABLE.into())]
1668 pub executable: String,
1669
1670 pub image_cache_dir: Option<PathBuf>,
1676
1677 #[toml(default)]
1680 pub extra_args: Vec<String>,
1681}
1682
1683impl Default for ApptainerConfig {
1684 fn default() -> Self {
1685 Self {
1686 executable: DEFAULT_APPTAINER_EXECUTABLE.into(),
1687 image_cache_dir: None,
1688 extra_args: Default::default(),
1689 }
1690 }
1691}
1692
1693impl ApptainerConfig {
1694 pub async fn validate(&self) -> Result<(), anyhow::Error> {
1696 Ok(())
1697 }
1698}
1699
1700#[derive(Debug, Clone, PartialEq, Eq)]
1709pub struct Condition {
1710 pub raw: String,
1712 expr: GreenNode,
1714}
1715
1716impl Condition {
1717 pub fn new(raw: impl Into<String>) -> Result<Self, Vec<Diagnostic>> {
1719 #[derive(Default)]
1722 struct Context(Vec<Diagnostic>);
1723
1724 impl wdl_analysis::types::v1::EvaluationContext for Context {
1725 fn version(&self) -> SupportedVersion {
1726 Default::default()
1727 }
1728
1729 fn resolve_name(&mut self, name: &str, span: Span) -> Option<Type> {
1730 match name {
1731 "cpu" => Some(PrimitiveType::Float.into()),
1732 "memory" => Some(PrimitiveType::Integer.into()),
1733 "gpu" | "fpga" => Some(PrimitiveType::Boolean.into()),
1734 "disks" => Some(PrimitiveType::Integer.into()),
1735 "hint" => Some(Type::Object),
1736 _ => {
1737 self.add_diagnostic(unknown_name(name, span));
1738 None
1739 }
1740 }
1741 }
1742
1743 fn resolve_type_name(&mut self, name: &str, span: Span) -> Result<Type, Diagnostic> {
1744 Err(unknown_type(name, span))
1745 }
1746
1747 fn task(&self) -> Option<&Task> {
1748 None
1749 }
1750
1751 fn diagnostics_config(&self) -> DiagnosticsConfig {
1752 Default::default()
1753 }
1754
1755 fn add_diagnostic(&mut self, diagnostic: Diagnostic) {
1756 self.0.push(diagnostic);
1757 }
1758 }
1759
1760 let raw = raw.into();
1761 let mut parser = Parser::new(Lexer::new(&raw));
1762 let marker = parser.start();
1763 match v1::expr(&mut parser, marker) {
1764 Ok(()) => {
1765 if let Some((_, span)) = parser.next() {
1766 return Err(vec![
1767 Diagnostic::error("expected a single WDL expression")
1768 .with_label("extraneous WDL source starts here", span),
1769 ]);
1770 }
1771
1772 let output = parser.finish();
1773 if !output.diagnostics.is_empty() {
1774 return Err(output.diagnostics);
1775 }
1776
1777 let expr = Expr::cast(construct_tree(raw.as_ref(), output.events))
1778 .expect("node should cast");
1779
1780 let mut context = Context::default();
1782 let ty = ExprTypeEvaluator::new(&mut context)
1783 .evaluate_expr(&expr)
1784 .unwrap_or(Type::Union);
1785
1786 if !context.0.is_empty() {
1787 return Err(context.0);
1788 }
1789
1790 match ty {
1791 Type::Primitive(PrimitiveType::Boolean, false) | Type::Union => {}
1792 _ => {
1793 return Err(vec![
1794 Diagnostic::error(format!(
1795 "conditional expression is expected to be type `Boolean`, but \
1796 found type `{ty}`",
1797 ))
1798 .with_highlight(expr.span()),
1799 ]);
1800 }
1801 }
1802
1803 Ok(Self {
1804 raw,
1805 expr: expr.inner().green().into_owned(),
1806 })
1807 }
1808 Err((marker, diagnostic)) => {
1809 marker.abandon(&mut parser);
1810 Err(vec![diagnostic.into()])
1811 }
1812 }
1813 }
1814
1815 pub(crate) async fn evaluate(
1820 &self,
1821 request: &ExecuteTaskRequest<'_>,
1822 transferer: &dyn Transferer,
1823 ) -> Result<bool> {
1824 struct Context<'a> {
1826 request: &'a ExecuteTaskRequest<'a>,
1828 transferer: &'a dyn Transferer,
1830 }
1831
1832 impl EvaluationContext for Context<'_> {
1833 fn version(&self) -> SupportedVersion {
1834 Default::default()
1835 }
1836
1837 fn resolve_name(&self, name: &str, span: Span) -> Result<Value, Diagnostic> {
1838 match name {
1839 "cpu" => Ok(self.request.constraints.cpu.into()),
1840 "memory" => Ok((self.request.constraints.memory as i64).into()),
1841 "gpu" => Ok((!self.request.constraints.gpu.is_empty()).into()),
1842 "fpga" => Ok((!self.request.constraints.fpga.is_empty()).into()),
1843 "disks" => Ok(self
1844 .request
1845 .constraints
1846 .disks
1847 .iter()
1848 .map(|(_, s)| *s)
1849 .sum::<i64>()
1850 .into()),
1851 "hint" => Ok(self.request.hints.clone().into()),
1852 _ => Err(unknown_name(name, span)),
1853 }
1854 }
1855
1856 fn resolve_type_name(&self, name: &str, span: Span) -> Result<Type, Diagnostic> {
1857 Err(unknown_type(name, span))
1858 }
1859
1860 fn enum_choice_value(
1861 &self,
1862 enum_name: &str,
1863 choice_name: &str,
1864 ) -> Result<Value, Diagnostic> {
1865 Err(unknown_enum_choice(enum_name, choice_name))
1866 }
1867
1868 fn base_dir(&self) -> &EvaluationPath {
1869 self.request.base_dir
1870 }
1871
1872 fn temp_dir(&self) -> &Path {
1873 self.request.temp_dir
1874 }
1875
1876 fn transferer(&self) -> &dyn Transferer {
1877 self.transferer
1878 }
1879
1880 fn object_access(&self, object: &Object, name: &str) -> Option<Value> {
1881 if !Arc::ptr_eq(&object.members, &self.request.hints.members) {
1884 return None;
1885 }
1886
1887 Some(
1891 self.request
1892 .inputs
1893 .hint(name)
1894 .or_else(|| object.get(name))
1895 .cloned()
1896 .unwrap_or_else(|| NoneValue::untyped().into()),
1897 )
1898 }
1899 }
1900
1901 async fn eval(context: Context<'_>, expr: &Expr<SyntaxNode>) -> Result<bool, Diagnostic> {
1906 let mut evaluator = ExprEvaluator::new(context);
1907 let value = evaluator.evaluate_expr(expr).await?;
1908 match value.as_boolean() {
1909 Some(res) => Ok(res),
1910 None => Err(Diagnostic::error(format!(
1911 "conditional expression is expected to be type `Boolean`, but found type \
1912 `{ty}`",
1913 ty = value.ty()
1914 ))
1915 .with_highlight(expr.span())),
1916 }
1917 }
1918
1919 let expr = Expr::cast(self.expr.clone().into()).expect("should be an expression node");
1920 match eval(
1921 Context {
1922 request,
1923 transferer,
1924 },
1925 &expr,
1926 )
1927 .await
1928 {
1929 Ok(res) => Ok(res),
1930 Err(diagnostic) => {
1931 let file: SimpleFile<_, _> = SimpleFile::new("<condition>", &self.raw);
1932 let mut buffer = Buffer::no_color();
1933 term::emit_to_write_style(
1934 &mut buffer,
1935 &Default::default(),
1936 &file,
1937 &diagnostic.to_codespan(()),
1938 )
1939 .context("failed to write diagnostic to buffer")?;
1940 let diagnostic = String::from_utf8(buffer.into_inner())
1941 .context("diagnostic buffer contents are not UTF-8")?;
1942 bail!(
1943 "failed to evaluate backend dynamic arguments condition: {diagnostic}",
1944 diagnostic = diagnostic
1945 .strip_prefix("error: ")
1946 .unwrap_or(&diagnostic)
1947 .trim()
1948 );
1949 }
1950 }
1951 }
1952}
1953
1954impl ToToml for Condition {
1955 fn to_toml<'a>(&'a self, arena: &'a Arena) -> Result<Item<'a>, ToTomlError> {
1956 self.raw.to_toml(arena)
1957 }
1958}
1959
1960impl<'de> FromToml<'de> for Condition {
1961 fn from_toml(ctx: &mut toml_spanner::Context<'de>, item: &Item<'de>) -> Result<Self, Failed> {
1962 fn remap(mapping: &[(usize, usize)], unescaped: usize) -> usize {
1965 match mapping.binary_search_by_key(&unescaped, |x| x.0) {
1966 Ok(i) => {
1967 mapping[i].1
1969 }
1970 Err(i) => {
1971 unescaped
1974 + if i == 0 {
1975 0
1978 } else {
1979 mapping[i - 1].1 - mapping[i - 1].0
1981 }
1982 }
1983 }
1984 }
1985
1986 fn push_errors(
1989 ctx: &mut toml_spanner::Context<'_>,
1990 item: &Item<'_>,
1991 diagnostics: Vec<Diagnostic>,
1992 ) -> Failed {
1993 for diagnostic in diagnostics {
1994 let span = if let Some(label) = diagnostic.labels().next() {
1995 let label_span = label.span();
1996 let span = item.span();
1997
1998 let source = &ctx.source()[span.start as usize..span.end as usize];
1999 let offset = if source.starts_with(r#"""""#) | source.starts_with("'''") {
2000 3
2001 } else {
2002 1
2003 };
2004
2005 let mapping = escape_mapping(
2006 source
2007 .get(offset..(source.len() - offset))
2008 .expect("invalid TOML string"),
2009 );
2010
2011 let label_start = remap(&mapping, label_span.start());
2013 let label_end = remap(&mapping, label_span.end());
2014
2015 toml_spanner::Span::new(
2016 span.start + offset as u32 + label_start as u32,
2017 span.start + offset as u32 + label_end as u32,
2018 )
2019 } else {
2020 item.span()
2021 };
2022
2023 ctx.errors
2024 .push(toml_spanner::Error::custom(diagnostic.message(), span));
2025 }
2026
2027 Failed
2028 }
2029
2030 Self::new(String::from_toml(ctx, item)?).map_err(|diags| push_errors(ctx, item, diags))
2031 }
2032}
2033
2034#[derive(Debug, Clone, PartialEq, Eq, Toml)]
2039#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2040pub struct ConditionalArgs {
2041 pub condition: Condition,
2043 #[toml(default)]
2045 pub args: Vec<String>,
2046}
2047
2048impl ConditionalArgs {
2049 pub fn validate(&self) -> Result<()> {
2054 if self.args.is_empty() {
2055 bail!("backend conditional arguments must have at least one argument specified");
2056 }
2057
2058 Ok(())
2059 }
2060}
2061
2062#[derive(Debug, Clone, Default, PartialEq, Eq, Toml)]
2066#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2067pub struct AdditionalArgs {
2068 #[toml(default)]
2070 pub args: Vec<String>,
2071 #[toml(default)]
2076 pub conditional: Vec<ConditionalArgs>,
2077}
2078
2079impl AdditionalArgs {
2080 pub fn validate(&self) -> Result<()> {
2082 for arg in &self.conditional {
2083 arg.validate()?;
2084 }
2085
2086 Ok(())
2087 }
2088}
2089
2090mod byte_size {
2092 use bytesize::ByteSize;
2093 use toml_spanner::Context;
2094 use toml_spanner::Failed;
2095 use toml_spanner::Item;
2096
2097 pub fn from_toml<'de>(ctx: &mut Context<'de>, item: &Item<'de>) -> Result<ByteSize, Failed> {
2099 if let Some(s) = item.as_u64() {
2100 return Ok(ByteSize(s));
2101 }
2102
2103 if let Some(s) = item.as_str() {
2104 return s
2105 .parse()
2106 .map_err(|e| ctx.report_custom_error(format!("invalid byte size: {e}"), item));
2107 }
2108
2109 Err(ctx.report_expected_but_found(&"integer or string", item))
2110 }
2111}
2112
2113#[derive(Debug, Clone, PartialEq, Eq, Toml)]
2122#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2123pub struct LsfQueueConfig {
2124 pub name: String,
2127 pub max_cpu_per_task: Option<u64>,
2129 #[toml(FromToml with = byte_size, ToToml with = display)]
2131 pub max_memory_per_task: Option<ByteSize>,
2132}
2133
2134impl LsfQueueConfig {
2135 pub async fn validate(&self, name: &str) -> Result<(), anyhow::Error> {
2137 let queue = &self.name;
2138 ensure!(!queue.is_empty(), "{name}_lsf_queue name cannot be empty");
2139 if let Some(max_cpu_per_task) = self.max_cpu_per_task {
2140 ensure!(
2141 max_cpu_per_task > 0,
2142 "{name}_lsf_queue `{queue}` must allow at least 1 CPU to be provisioned"
2143 );
2144 }
2145 if let Some(max_memory_per_task) = self.max_memory_per_task {
2146 ensure!(
2147 max_memory_per_task.as_u64() > 0,
2148 "{name}_lsf_queue `{queue}` must allow at least some memory to be provisioned"
2149 );
2150 }
2151 match tokio::time::timeout(
2152 std::time::Duration::from_secs(10),
2155 Command::new("bqueues").arg(queue).output(),
2156 )
2157 .await
2158 {
2159 Ok(output) => {
2160 let output = output.context("validating LSF queue")?;
2161 if !output.status.success() {
2162 let stdout = String::from_utf8_lossy(&output.stdout);
2163 let stderr = String::from_utf8_lossy(&output.stderr);
2164 error!(%stdout, %stderr, %queue, "failed to validate {name}_lsf_queue");
2165 Err(anyhow!("failed to validate {name}_lsf_queue `{queue}`"))
2166 } else {
2167 Ok(())
2168 }
2169 }
2170 Err(_) => Err(anyhow!(
2171 "timed out trying to validate {name}_lsf_queue `{queue}`"
2172 )),
2173 }
2174 }
2175}
2176
2177#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
2181#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2182pub struct LsfApptainerBackendConfig {
2183 pub interval: Option<u64>,
2187 pub max_concurrency: Option<u32>,
2195 #[toml(style = Header)]
2202 pub default_lsf_queue: Option<LsfQueueConfig>,
2203 #[toml(style = Header)]
2210 pub short_task_lsf_queue: Option<LsfQueueConfig>,
2211 #[toml(style = Header)]
2215 pub gpu_lsf_queue: Option<LsfQueueConfig>,
2216 #[toml(style = Header)]
2220 pub fpga_lsf_queue: Option<LsfQueueConfig>,
2221 pub job_name_prefix: Option<String>,
2224 #[toml(default)]
2226 pub bsub: AdditionalArgs,
2227 #[toml(default)]
2234 pub apptainer: ApptainerConfig,
2235}
2236
2237impl LsfApptainerBackendConfig {
2238 pub async fn validate(&self) -> Result<(), anyhow::Error> {
2240 if cfg!(not(unix)) {
2241 bail!("LSF + Apptainer backend is not supported on non-unix platforms");
2242 }
2243
2244 if let Some(queue) = &self.default_lsf_queue {
2250 queue.validate("default").await?;
2251 }
2252
2253 if let Some(queue) = &self.short_task_lsf_queue {
2254 queue.validate("short_task").await?;
2255 }
2256
2257 if let Some(queue) = &self.gpu_lsf_queue {
2258 queue.validate("gpu").await?;
2259 }
2260
2261 if let Some(queue) = &self.fpga_lsf_queue {
2262 queue.validate("fpga").await?;
2263 }
2264
2265 if let Some(prefix) = &self.job_name_prefix
2266 && prefix.len() > MAX_LSF_JOB_NAME_PREFIX
2267 {
2268 bail!(
2269 "LSF job name prefix `{prefix}` exceeds the maximum {MAX_LSF_JOB_NAME_PREFIX} \
2270 bytes"
2271 );
2272 }
2273
2274 self.bsub.validate()?;
2276
2277 self.apptainer.validate().await?;
2279
2280 Ok(())
2281 }
2282
2283 pub(crate) fn lsf_queue_for_task(
2288 &self,
2289 requirements: &Object,
2290 hints: &Object,
2291 ) -> Option<&LsfQueueConfig> {
2292 if let Some(queue) = self.fpga_lsf_queue.as_ref()
2294 && let Some(true) = requirements
2295 .get(wdl_ast::v1::TASK_REQUIREMENT_FPGA)
2296 .and_then(Value::as_boolean)
2297 {
2298 return Some(queue);
2299 }
2300
2301 if let Some(queue) = self.gpu_lsf_queue.as_ref()
2302 && let Some(true) = requirements
2303 .get(wdl_ast::v1::TASK_REQUIREMENT_GPU)
2304 .and_then(Value::as_boolean)
2305 {
2306 return Some(queue);
2307 }
2308
2309 if let Some(queue) = self.short_task_lsf_queue.as_ref()
2311 && let Some(true) = hints
2312 .get(wdl_ast::v1::TASK_HINT_SHORT_TASK)
2313 .and_then(Value::as_boolean)
2314 {
2315 return Some(queue);
2316 }
2317
2318 self.default_lsf_queue.as_ref()
2321 }
2322}
2323
2324#[derive(Debug, Clone, PartialEq, Eq, Toml)]
2333#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2334pub struct SlurmPartitionConfig {
2335 pub name: String,
2338 pub max_cpu_per_task: Option<u64>,
2341 #[toml(FromToml with = byte_size, ToToml with = display)]
2343 pub max_memory_per_task: Option<ByteSize>,
2344}
2345
2346impl SlurmPartitionConfig {
2347 pub async fn validate(&self, name: &str) -> Result<(), anyhow::Error> {
2350 let partition = &self.name;
2351 ensure!(
2352 !partition.is_empty(),
2353 "{name}_slurm_partition name cannot be empty"
2354 );
2355 if let Some(max_cpu_per_task) = self.max_cpu_per_task {
2356 ensure!(
2357 max_cpu_per_task > 0,
2358 "{name}_slurm_partition `{partition}` must allow at least 1 CPU to be provisioned"
2359 );
2360 }
2361 if let Some(max_memory_per_task) = self.max_memory_per_task {
2362 ensure!(
2363 max_memory_per_task.as_u64() > 0,
2364 "{name}_slurm_partition `{partition}` must allow at least some memory to be \
2365 provisioned"
2366 );
2367 }
2368 match tokio::time::timeout(
2369 std::time::Duration::from_secs(10),
2372 Command::new("scontrol")
2373 .arg("show")
2374 .arg("partition")
2375 .arg(partition)
2376 .output(),
2377 )
2378 .await
2379 {
2380 Ok(output) => {
2381 let output = output.context("validating Slurm partition")?;
2382 if !output.status.success() {
2383 let stdout = String::from_utf8_lossy(&output.stdout);
2384 let stderr = String::from_utf8_lossy(&output.stderr);
2385 error!(%stdout, %stderr, %partition, "failed to validate {name}_slurm_partition");
2386 Err(anyhow!(
2387 "failed to validate {name}_slurm_partition `{partition}`"
2388 ))
2389 } else {
2390 Ok(())
2391 }
2392 }
2393 Err(_) => Err(anyhow!(
2394 "timed out trying to validate {name}_slurm_partition `{partition}`"
2395 )),
2396 }
2397 }
2398}
2399
2400#[derive(Debug, Default, Clone, PartialEq, Eq, Toml)]
2404#[toml(Toml, rename_all = "snake_case", deny_unknown_fields)]
2405pub struct SlurmApptainerBackendConfig {
2406 pub interval: Option<u64>,
2410 pub max_concurrency: Option<u32>,
2418 #[toml(style = Header)]
2427 pub default_slurm_partition: Option<SlurmPartitionConfig>,
2428 #[toml(style = Header)]
2436 pub short_task_slurm_partition: Option<SlurmPartitionConfig>,
2437 #[toml(style = Header)]
2441 pub gpu_slurm_partition: Option<SlurmPartitionConfig>,
2442 #[toml(style = Header)]
2446 pub fpga_slurm_partition: Option<SlurmPartitionConfig>,
2447 #[toml(default)]
2449 pub sbatch: AdditionalArgs,
2450 pub job_name_prefix: Option<String>,
2452 #[toml(default)]
2459 pub apptainer: ApptainerConfig,
2460}
2461
2462impl SlurmApptainerBackendConfig {
2463 pub async fn validate(&self) -> Result<(), anyhow::Error> {
2465 if cfg!(not(unix)) {
2466 bail!("Slurm + Apptainer backend is not supported on non-unix platforms");
2467 }
2468
2469 if let Some(partition) = &self.default_slurm_partition {
2475 partition.validate("default").await?;
2476 }
2477 if let Some(partition) = &self.short_task_slurm_partition {
2478 partition.validate("short_task").await?;
2479 }
2480 if let Some(partition) = &self.gpu_slurm_partition {
2481 partition.validate("gpu").await?;
2482 }
2483 if let Some(partition) = &self.fpga_slurm_partition {
2484 partition.validate("fpga").await?;
2485 }
2486
2487 self.sbatch.validate()?;
2489
2490 self.apptainer.validate().await?;
2492
2493 Ok(())
2494 }
2495
2496 pub(crate) fn slurm_partition_for_task(
2501 &self,
2502 requirements: &Object,
2503 hints: &Object,
2504 ) -> Option<&SlurmPartitionConfig> {
2505 if let Some(partition) = self.fpga_slurm_partition.as_ref()
2511 && let Some(true) = requirements
2512 .get(wdl_ast::v1::TASK_REQUIREMENT_FPGA)
2513 .and_then(Value::as_boolean)
2514 {
2515 return Some(partition);
2516 }
2517
2518 if let Some(partition) = self.gpu_slurm_partition.as_ref()
2519 && let Some(true) = requirements
2520 .get(wdl_ast::v1::TASK_REQUIREMENT_GPU)
2521 .and_then(Value::as_boolean)
2522 {
2523 return Some(partition);
2524 }
2525
2526 if let Some(partition) = self.short_task_slurm_partition.as_ref()
2528 && let Some(true) = hints
2529 .get(wdl_ast::v1::TASK_HINT_SHORT_TASK)
2530 .and_then(Value::as_boolean)
2531 {
2532 return Some(partition);
2533 }
2534
2535 self.default_slurm_partition.as_ref()
2538 }
2539}
2540
2541#[derive(Debug, thiserror::Error)]
2543pub enum BuilderMergeError {
2544 #[error("failed to serialize after merging configuration")]
2546 Serialize(#[from] ToTomlError),
2547 #[error("failed to parse after merging configuration")]
2549 Parse {
2550 source: String,
2552 #[source]
2554 error: toml_spanner::Error,
2555 },
2556 #[error("failed to deserialize after merging configuration")]
2558 Deserialize {
2559 source: String,
2561 #[source]
2563 error: toml_spanner::FromTomlError,
2564 },
2565}
2566
2567struct BuilderErrorDisplay<'a>(&'static str, &'a Option<PathBuf>);
2569
2570impl fmt::Display for BuilderErrorDisplay<'_> {
2571 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2572 match self.1 {
2573 Some(path) => write!(
2574 f,
2575 "failed to {op} configuration file `{path}`",
2576 op = self.0,
2577 path = path.display()
2578 ),
2579 None => write!(
2580 f,
2581 "failed to {op} in-memory configuration string",
2582 op = self.0
2583 ),
2584 }
2585 }
2586}
2587
2588#[derive(Debug, thiserror::Error)]
2590pub enum BuilderError {
2591 #[error("failed to read configuration file `{path}`")]
2593 Io {
2594 path: PathBuf,
2596 #[source]
2598 error: std::io::Error,
2599 },
2600 #[error("{}", BuilderErrorDisplay("parse", .path))]
2602 Parse {
2603 path: Option<PathBuf>,
2607 source: String,
2609 #[source]
2611 error: toml_spanner::Error,
2612 },
2613 #[error("{}", BuilderErrorDisplay("deserialize", .path))]
2615 Deserialize {
2616 path: Option<PathBuf>,
2620 source: String,
2622 #[source]
2624 error: toml_spanner::FromTomlError,
2625 },
2626 #[error(transparent)]
2628 Merge(#[from] BuilderMergeError),
2629}
2630
2631impl BuilderError {
2632 pub fn path(&self) -> impl fmt::Display + Clone {
2640 #[derive(Clone)]
2641 struct Helper<'a>(&'a BuilderError);
2642
2643 impl fmt::Display for Helper<'_> {
2644 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
2645 match self.0 {
2646 BuilderError::Io { path, .. }
2647 | BuilderError::Parse {
2648 path: Some(path), ..
2649 }
2650 | BuilderError::Deserialize {
2651 path: Some(path), ..
2652 } => path.display().fmt(f),
2653 BuilderError::Parse { path: None, .. }
2654 | BuilderError::Deserialize { path: None, .. } => write!(f, "<string>"),
2655 BuilderError::Merge(_) => write!(f, "<merged>"),
2656 }
2657 }
2658 }
2659
2660 Helper(self)
2661 }
2662
2663 pub fn toml_error(&self) -> Option<&toml_spanner::Error> {
2668 match &self {
2669 Self::Parse { error, .. } | Self::Merge(BuilderMergeError::Parse { error, .. }) => {
2670 Some(error)
2671 }
2672 Self::Deserialize { error, .. }
2673 | Self::Merge(BuilderMergeError::Deserialize { error, .. }) => error.errors.first(),
2674 _ => None,
2675 }
2676 }
2677
2678 pub fn source(&self) -> Option<&str> {
2683 match &self {
2684 Self::Parse { source, .. }
2685 | Self::Deserialize { source, .. }
2686 | Self::Merge(BuilderMergeError::Parse { source, .. })
2687 | Self::Merge(BuilderMergeError::Deserialize { source, .. }) => Some(source),
2688 _ => None,
2689 }
2690 }
2691
2692 pub fn to_diagnostic(&self) -> Diagnostic {
2694 let mut diagnostic = Diagnostic::error(self.to_string());
2695
2696 if let Some(e) = self.toml_error() {
2697 for (span, text) in [e.primary_label(), e.secondary_label()]
2698 .into_iter()
2699 .flatten()
2700 {
2701 let span = Span::new(span.start as usize, (span.end - span.start) as usize);
2702 let text: &str = text.trim();
2703
2704 let text = if text.is_empty() {
2706 match e.kind() {
2707 ErrorKind::UnexpectedEof => "unexpected end of file",
2708 ErrorKind::RedefineAsArray { .. } => {
2709 "a previously defined table was redefined as an array"
2710 }
2711 ErrorKind::FileTooLarge => "file is too large",
2712 ErrorKind::Custom(message) => message,
2713 _ => text,
2714 }
2715 } else {
2716 text
2717 };
2718
2719 if text.is_empty() {
2720 diagnostic = diagnostic.with_highlight(span);
2721 } else {
2722 diagnostic = diagnostic.with_label(text, span);
2723 }
2724 }
2725 }
2726
2727 if matches!(self, Self::Merge(_)) {
2728 diagnostic = diagnostic.with_help("reported line numbers reflect merged TOML source");
2729 }
2730
2731 diagnostic
2732 }
2733}
2734
2735#[derive(Debug)]
2737enum Source {
2738 Path(PathBuf),
2740 String(String),
2742}
2743
2744#[derive(Default, Debug)]
2758pub struct ConfigBuilder<T> {
2759 sources: Vec<Source>,
2761 _phantom: PhantomData<T>,
2763}
2764
2765impl<T> ConfigBuilder<T> {
2766 pub fn with_string_source(mut self, toml: impl Into<String>) -> Self {
2768 self.sources.push(Source::String(toml.into()));
2769 self
2770 }
2771
2772 pub fn with_file_source(mut self, path: impl Into<PathBuf>) -> Self {
2774 self.sources.push(Source::Path(path.into()));
2775 self
2776 }
2777
2778 pub fn try_build(self) -> Result<T, BuilderError>
2783 where
2784 T: ToToml + for<'de> FromToml<'de>,
2785 {
2786 let sources: Vec<(Option<PathBuf>, String)> = self
2788 .sources
2789 .into_iter()
2790 .map(|s| match s {
2791 Source::Path(path) => {
2792 let source = fs::read_to_string(&path).map_err(|e| BuilderError::Io {
2793 path: path.clone(),
2794 error: e,
2795 })?;
2796
2797 Ok((Some(path), source))
2798 }
2799 Source::String(source) => Ok((None, source)),
2800 })
2801 .collect::<Result<_, BuilderError>>()?;
2802
2803 let arena = Arena::new();
2805 let documents = sources
2806 .iter()
2807 .map(|(path, source)| {
2808 toml_spanner::parse(source, &arena).map_err(|e| BuilderError::Parse {
2809 path: path.clone(),
2810 source: source.clone(),
2811 error: e,
2812 })
2813 })
2814 .collect::<Result<Vec<_>, _>>()?;
2815
2816 let mut merged_table: Table<'_> = Table::new();
2818 for (index, mut document) in documents.into_iter().enumerate() {
2819 document.to::<T>().map_err(|e| {
2822 let (path, source) = &sources[index];
2823 BuilderError::Deserialize {
2824 path: path.clone(),
2825 source: source.clone(),
2826 error: e,
2827 }
2828 })?;
2829
2830 Self::merge_tables(document.into_table(), &mut merged_table, &arena);
2832 }
2833
2834 let source =
2838 toml_spanner::to_string(&merged_table).map_err(BuilderMergeError::Serialize)?;
2839
2840 Ok(toml_spanner::parse(&source, &arena)
2842 .map_err(|e| BuilderMergeError::Parse {
2843 source: source.clone(),
2844 error: e,
2845 })?
2846 .to()
2847 .map_err(|e| BuilderMergeError::Deserialize {
2848 source: source.clone(),
2849 error: e,
2850 })?)
2851 }
2852
2853 fn merge_tables<'de>(src: Table<'de>, dest: &mut Table<'de>, arena: &'de Arena) {
2864 for (key, src_item) in src {
2865 let Some(dest_item) = dest.get_mut(key.name) else {
2866 dest.insert(key, src_item, arena);
2867 continue;
2868 };
2869
2870 if let Some(src_array) = src_item.as_array()
2872 && let Some(dest_array) = dest_item.as_array_mut()
2873 {
2874 for element in src_array {
2875 dest_array.push(element.clone_in(arena), arena);
2876 }
2877 continue;
2878 }
2879
2880 if let Some(_) = src_item.as_table()
2882 && let Some(dest_table) = dest_item.as_table_mut()
2883 {
2884 Self::merge_tables(src_item.into_table().unwrap(), dest_table, arena);
2885 continue;
2886 }
2887
2888 dest.insert(key, src_item, arena);
2890 }
2891 }
2892}
2893
2894#[cfg(test)]
2895mod test {
2896 use std::collections::HashMap;
2897 use std::io::Write;
2898
2899 use codespan_reporting::files::SimpleFile;
2900 use codespan_reporting::term::DisplayStyle;
2901 use codespan_reporting::term::emit_into_string;
2902 use codespan_reporting::term::{self};
2903 use futures::future::BoxFuture;
2904 use pretty_assertions::assert_eq;
2905 use tempfile::TempPath;
2906 use tempfile::tempdir;
2907
2908 use super::*;
2909 use crate::ONE_GIBIBYTE;
2910 use crate::TaskInputs;
2911 use crate::backend::TaskExecutionConstraints;
2912 use crate::http::Location;
2913 use crate::v1::DEFAULT_TASK_REQUIREMENT_CPU;
2914 use crate::v1::DEFAULT_TASK_REQUIREMENT_DISKS;
2915 use crate::v1::DEFAULT_TASK_REQUIREMENT_MEMORY;
2916
2917 #[test]
2918 fn redacted_secret() {
2919 let mut map: HashMap<_, SecretString> = HashMap::new();
2920 map.insert(
2921 "foo",
2922 SecretString {
2923 inner: "secret".into(),
2924 redacted: false,
2925 },
2926 );
2927
2928 assert_eq!(
2929 toml_spanner::to_string(&map).unwrap().trim(),
2930 format!(r#"foo = "secret""#)
2931 );
2932
2933 map.insert(
2934 "foo",
2935 SecretString {
2936 inner: "secret".into(),
2937 redacted: true,
2938 },
2939 );
2940 assert_eq!(
2941 toml_spanner::to_string(&map).unwrap().trim(),
2942 format!(r#"foo = "{REDACTED}""#)
2943 );
2944 }
2945
2946 #[test]
2947 fn redacted_config() {
2948 let config = Config {
2949 backends: [
2950 (
2951 "first".to_string(),
2952 TesBackendConfig {
2953 auth: Some(TesBackendAuthConfig::Basic {
2954 config: BasicAuthConfig {
2955 username: "foo".into(),
2956 password: "secret".into(),
2957 },
2958 }),
2959 ..Default::default()
2960 }
2961 .into(),
2962 ),
2963 (
2964 "second".to_string(),
2965 TesBackendConfig {
2966 auth: Some(
2967 BearerAuthConfig {
2968 token: "secret".into(),
2969 }
2970 .into(),
2971 ),
2972 ..Default::default()
2973 }
2974 .into(),
2975 ),
2976 ]
2977 .into(),
2978 storage: StorageConfig {
2979 azure: AzureStorageConfig {
2980 auth: Some(AzureStorageAuthConfig {
2981 account_name: "foo".into(),
2982 access_key: "secret".into(),
2983 }),
2984 },
2985 s3: S3StorageConfig {
2986 auth: Some(S3StorageAuthConfig {
2987 access_key_id: "foo".into(),
2988 secret_access_key: "secret".into(),
2989 }),
2990 ..Default::default()
2991 },
2992 google: GoogleStorageConfig {
2993 auth: Some(GoogleStorageAuthConfig {
2994 access_key: "foo".into(),
2995 secret: "secret".into(),
2996 }),
2997 },
2998 },
2999 ..Default::default()
3000 };
3001
3002 let toml = toml_spanner::to_string(&config).unwrap();
3003 assert!(toml.contains("secret"), "`{toml}` contains a secret");
3004 }
3005
3006 #[tokio::test]
3007 async fn test_config_validate() {
3008 let mut config = Config::default();
3010 config.task.retries = 255.into();
3011 assert_eq!(
3012 config.validate().await.unwrap_err().to_string(),
3013 "configuration value `task.retries` cannot exceed 100"
3014 );
3015
3016 let mut config = Config::default();
3018 config.workflow.scatter.concurrency = 0;
3019 assert_eq!(
3020 config.validate().await.unwrap_err().to_string(),
3021 "configuration value `workflow.scatter.concurrency` cannot be zero"
3022 );
3023
3024 let config = Config {
3026 backend: "foo".into(),
3027 ..Default::default()
3028 };
3029 assert_eq!(
3030 config.validate().await.unwrap_err().to_string(),
3031 "a backend named `foo` is not present in the configuration"
3032 );
3033 let config = Config {
3034 backend: "bar".into(),
3035 backends: [("foo".to_string(), BackendConfig::default())].into(),
3036 ..Default::default()
3037 };
3038 assert_eq!(
3039 config.validate().await.unwrap_err().to_string(),
3040 "a backend named `bar` is not present in the configuration"
3041 );
3042
3043 let config = Config {
3045 backend: "foo".to_string(),
3046 backends: [("foo".to_string(), BackendConfig::default())].into(),
3047 ..Default::default()
3048 };
3049 config.validate().await.expect("config should validate");
3050
3051 let config = Config {
3053 backends: [(
3054 "default".to_string(),
3055 LocalBackendConfig {
3056 cpu: Some(0),
3057 ..Default::default()
3058 }
3059 .into(),
3060 )]
3061 .into(),
3062 ..Default::default()
3063 };
3064 assert_eq!(
3065 config.validate().await.unwrap_err().to_string(),
3066 "local backend configuration value `cpu` cannot be zero"
3067 );
3068 let config = Config {
3069 backends: [(
3070 "default".to_string(),
3071 LocalBackendConfig {
3072 cpu: Some(10000000),
3073 ..Default::default()
3074 }
3075 .into(),
3076 )]
3077 .into(),
3078 ..Default::default()
3079 };
3080 assert!(
3081 config
3082 .validate()
3083 .await
3084 .unwrap_err()
3085 .to_string()
3086 .starts_with(
3087 "local backend configuration value `cpu` cannot exceed the virtual CPUs \
3088 available to the host"
3089 )
3090 );
3091
3092 let config = Config {
3094 backends: [(
3095 "default".to_string(),
3096 LocalBackendConfig {
3097 memory: Some("0 GiB".to_string()),
3098 ..Default::default()
3099 }
3100 .into(),
3101 )]
3102 .into(),
3103 ..Default::default()
3104 };
3105 assert_eq!(
3106 config.validate().await.unwrap_err().to_string(),
3107 "local backend configuration value `memory` cannot be zero"
3108 );
3109 let config = Config {
3110 backends: [(
3111 "default".to_string(),
3112 LocalBackendConfig {
3113 memory: Some("100 meows".to_string()),
3114 ..Default::default()
3115 }
3116 .into(),
3117 )]
3118 .into(),
3119 ..Default::default()
3120 };
3121 assert_eq!(
3122 config.validate().await.unwrap_err().to_string(),
3123 "local backend configuration value `memory` has invalid value `100 meows`"
3124 );
3125
3126 let config = Config {
3127 backends: [(
3128 "default".to_string(),
3129 LocalBackendConfig {
3130 memory: Some("1000 TiB".to_string()),
3131 ..Default::default()
3132 }
3133 .into(),
3134 )]
3135 .into(),
3136 ..Default::default()
3137 };
3138 assert!(
3139 config
3140 .validate()
3141 .await
3142 .unwrap_err()
3143 .to_string()
3144 .starts_with(
3145 "local backend configuration value `memory` cannot exceed the total memory of \
3146 the host"
3147 )
3148 );
3149
3150 let config = Config {
3152 backends: [("default".to_string(), TesBackendConfig::default().into())].into(),
3153 ..Default::default()
3154 };
3155 assert_eq!(
3156 config.validate().await.unwrap_err().to_string(),
3157 "TES backend configuration value `url` is required"
3158 );
3159
3160 let config = Config {
3162 backends: [(
3163 "default".to_string(),
3164 TesBackendConfig {
3165 url: Some("https://example.com".parse().unwrap()),
3166 max_concurrency: Some(0),
3167 ..Default::default()
3168 }
3169 .into(),
3170 )]
3171 .into(),
3172 ..Default::default()
3173 };
3174 assert_eq!(
3175 config.validate().await.unwrap_err().to_string(),
3176 "TES backend configuration value `max_concurrency` cannot be zero"
3177 );
3178
3179 let config = Config {
3181 backends: [(
3182 "default".to_string(),
3183 TesBackendConfig {
3184 url: Some("http://example.com".parse().unwrap()),
3185 inputs: Some("http://example.com".parse().unwrap()),
3186 outputs: Some("http://example.com".parse().unwrap()),
3187 ..Default::default()
3188 }
3189 .into(),
3190 )]
3191 .into(),
3192 ..Default::default()
3193 };
3194 assert_eq!(
3195 config.validate().await.unwrap_err().to_string(),
3196 "TES backend configuration value `url` has invalid value `http://example.com/`: URL \
3197 must use a HTTPS scheme"
3198 );
3199
3200 let config = Config {
3202 backends: [(
3203 "default".to_string(),
3204 TesBackendConfig {
3205 url: Some("http://example.com".parse().unwrap()),
3206 inputs: Some("http://example.com".parse().unwrap()),
3207 outputs: Some("http://example.com".parse().unwrap()),
3208 insecure: true,
3209 ..Default::default()
3210 }
3211 .into(),
3212 )]
3213 .into(),
3214 ..Default::default()
3215 };
3216 config
3217 .validate()
3218 .await
3219 .expect("configuration should validate");
3220
3221 let mut config = Config::default();
3223 config.http.parallelism = 0.into();
3224 assert_eq!(
3225 config.validate().await.unwrap_err().to_string(),
3226 "configuration value `http.parallelism` cannot be zero"
3227 );
3228
3229 let mut config = Config::default();
3231 config.http.parallelism = 5.into();
3232 assert!(
3233 config.validate().await.is_ok(),
3234 "should pass for valid configuration"
3235 );
3236 let mut config = Config::default();
3237 config.http.parallelism = Parallelism::default();
3238 assert!(config.validate().await.is_ok(), "should pass for default");
3239
3240 #[cfg(unix)]
3242 {
3243 let job_name_prefix = "A".repeat(MAX_LSF_JOB_NAME_PREFIX * 2);
3244 let mut config = Config {
3245 experimental_features_enabled: true,
3246 ..Default::default()
3247 };
3248 config.backends.insert(
3249 "default".to_string(),
3250 LsfApptainerBackendConfig {
3251 job_name_prefix: Some(job_name_prefix.clone()),
3252 ..Default::default()
3253 }
3254 .into(),
3255 );
3256 assert_eq!(
3257 config.validate().await.unwrap_err().to_string(),
3258 format!("LSF job name prefix `{job_name_prefix}` exceeds the maximum 100 bytes")
3259 );
3260 }
3261 }
3262
3263 fn create_temp_file(contents: &str) -> TempPath {
3264 let mut file: tempfile::NamedTempFile =
3265 tempfile::NamedTempFile::new().expect("failed to create temporary file");
3266 file.write_all(contents.as_bytes())
3267 .expect("failed to write temporary file");
3268 file.into_temp_path()
3269 }
3270
3271 #[test]
3272 fn it_builds_with_no_sources() {
3273 let config = Config::builder().try_build().expect("should build");
3274 assert_eq!(config, Config::default(), "should be equal");
3275 }
3276
3277 #[test]
3278 fn it_builds_with_one_source() {
3279 let path = create_temp_file("backend = 'foo'");
3280
3281 let config = Config::builder()
3282 .with_file_source(&path)
3283 .try_build()
3284 .expect("should build");
3285 assert_eq!(
3286 config,
3287 Config {
3288 backend: "foo".into(),
3289 ..Default::default()
3290 },
3291 "should be equal"
3292 );
3293 }
3294
3295 #[test]
3296 fn it_errors_on_invalid_parse() {
3297 let path = create_temp_file("invalid");
3298
3299 let e = Config::builder()
3300 .with_file_source(&path)
3301 .try_build()
3302 .expect_err("should fail");
3303
3304 let source = e.source().expect("should have source");
3305
3306 let diagnostic = e.to_diagnostic();
3307 let error = emit_into_string(
3308 &term::Config {
3309 display_style: DisplayStyle::Rich,
3310 ..Default::default()
3311 },
3312 &SimpleFile::new(e.path(), source),
3313 &diagnostic.to_codespan(()),
3314 )
3315 .expect("should emit");
3316
3317 assert!(
3318 error.contains("expected an equals"),
3319 "the error `{error}` does not contain the expected message"
3320 );
3321 }
3322
3323 #[test]
3324 fn it_errors_on_invalid_deserialization() {
3325 let path = create_temp_file("backend = 42");
3326
3327 let e = Config::builder()
3328 .with_file_source(&path)
3329 .try_build()
3330 .expect_err("should fail");
3331
3332 let source = e.source().expect("should have source");
3333
3334 let diagnostic = e.to_diagnostic();
3335 let error = emit_into_string(
3336 &term::Config {
3337 display_style: DisplayStyle::Rich,
3338 ..Default::default()
3339 },
3340 &SimpleFile::new(e.path(), source),
3341 &diagnostic.to_codespan(()),
3342 )
3343 .expect("should emit");
3344
3345 assert!(
3346 error.contains("expected a string"),
3347 "the error `{error}` does not contain the expected message"
3348 );
3349 }
3350
3351 #[test]
3352 fn it_merges_sources() {
3353 let first = create_temp_file(
3354 r#"
3355backend = 'foo'
3356
3357[task]
3358excluded_cache_inputs = ['1', '2', '3']
3359
3360[backends.foo]
3361type = 'local'
3362"#,
3363 );
3364
3365 let second = create_temp_file(
3366 r#"
3367[task]
3368excluded_cache_inputs = ['4', '5']
3369
3370[backends.bar]
3371type = 'docker'
3372"#,
3373 );
3374
3375 let third: TempPath = create_temp_file(
3376 r#"
3377backend = 'baz'
3378
3379[task]
3380excluded_cache_inputs = ['6', '7', '8']
3381
3382[backends.baz]
3383type = 'tes'
3384"#,
3385 );
3386
3387 let fourth = r#"
3388backend = 'qux'
3389
3390[task]
3391excluded_cache_inputs = ['9', '10']
3392
3393[backends.qux]
3394type = 'lsf_apptainer'
3395"#;
3396
3397 let config = Config::builder()
3398 .with_file_source(&first)
3399 .with_file_source(&second)
3400 .with_file_source(&third)
3401 .with_string_source(fourth)
3402 .try_build()
3403 .expect("should build");
3404
3405 assert_eq!(
3406 config,
3407 Config {
3408 backend: "qux".into(),
3409 task: TaskConfig {
3410 excluded_cache_inputs: vec![
3411 "1".to_string(),
3412 "2".to_string(),
3413 "3".to_string(),
3414 "4".to_string(),
3415 "5".to_string(),
3416 "6".to_string(),
3417 "7".to_string(),
3418 "8".to_string(),
3419 "9".to_string(),
3420 "10".to_string(),
3421 ],
3422 ..Default::default()
3423 },
3424 backends: IndexMap::from_iter([
3425 ("foo".to_string(), LocalBackendConfig::default().into()),
3426 ("bar".to_string(), DockerBackendConfig::default().into()),
3427 ("baz".to_string(), TesBackendConfig::default().into()),
3428 (
3429 "qux".to_string(),
3430 LsfApptainerBackendConfig::default().into()
3431 ),
3432 ]),
3433 ..Default::default()
3434 },
3435 "should be equal"
3436 );
3437 }
3438
3439 #[test]
3440 fn parallelism_serialization() {
3441 let map: HashMap<&str, Parallelism> =
3442 HashMap::from_iter([("value", Parallelism::Available)]);
3443 assert_eq!(
3444 toml_spanner::to_string(&map).unwrap(),
3445 format!("value = \"available\"\n")
3446 );
3447
3448 let map: HashMap<&str, Parallelism> =
3449 HashMap::from_iter([("value", Parallelism::Use(123))]);
3450 assert_eq!(toml_spanner::to_string(&map).unwrap(), "value = 123\n");
3451 }
3452
3453 #[test]
3454 fn parallelism_deserialization() {
3455 let map: HashMap<String, Parallelism> =
3456 toml_spanner::from_str("value = 'available'").unwrap();
3457 assert_eq!(map["value"], Parallelism::Available);
3458
3459 let map: HashMap<String, Parallelism> = toml_spanner::from_str("value = 123").unwrap();
3460 assert_eq!(map["value"], Parallelism::Use(123));
3461
3462 let expected_error =
3463 "expected a positive integer or `available` for parallelism at `value`";
3464
3465 let error =
3466 toml_spanner::from_str::<HashMap<String, Parallelism>>("value = 'wrong'").unwrap_err();
3467 assert_eq!(error.to_string(), expected_error);
3468
3469 let error =
3470 toml_spanner::from_str::<HashMap<String, Parallelism>>("value = 0").unwrap_err();
3471 assert_eq!(error.to_string(), expected_error);
3472
3473 let error =
3474 toml_spanner::from_str::<HashMap<String, Parallelism>>("value = -10").unwrap_err();
3475 assert_eq!(error.to_string(), expected_error);
3476 }
3477
3478 #[test]
3479 fn retries_serialization() {
3480 let map: HashMap<&str, Retries> = HashMap::from_iter([("value", Retries::Default)]);
3481 assert_eq!(
3482 toml_spanner::to_string(&map).unwrap(),
3483 format!("value = \"default\"\n")
3484 );
3485
3486 let map: HashMap<&str, Retries> = HashMap::from_iter([("value", Retries::Use(123))]);
3487 assert_eq!(toml_spanner::to_string(&map).unwrap(), "value = 123\n");
3488 }
3489
3490 #[test]
3491 fn retries_deserialization() {
3492 let map: HashMap<String, Retries> = toml_spanner::from_str("value = 'default'").unwrap();
3493 assert_eq!(map["value"], Retries::Default);
3494
3495 let map: HashMap<String, Retries> = toml_spanner::from_str("value = 12").unwrap();
3496 assert_eq!(map["value"], Retries::Use(12));
3497
3498 let map: HashMap<String, Retries> = toml_spanner::from_str("value = 0").unwrap();
3499 assert_eq!(map["value"], Retries::Use(0));
3500
3501 let expected_error = format!(
3502 "expected an integer less than {MAX_RETRIES} or `default` for retries at `value`"
3503 );
3504
3505 let error =
3506 toml_spanner::from_str::<HashMap<String, Retries>>("value = 'wrong'").unwrap_err();
3507 assert_eq!(error.to_string(), expected_error);
3508
3509 let error = toml_spanner::from_str::<HashMap<String, Retries>>("value = 101").unwrap_err();
3510 assert_eq!(error.to_string(), expected_error);
3511
3512 let error = toml_spanner::from_str::<HashMap<String, Retries>>("value = -10").unwrap_err();
3513 assert_eq!(error.to_string(), expected_error);
3514 }
3515
3516 #[test]
3517 fn mapping_escape_indexes() {
3518 assert!(escape_mapping("").is_empty());
3520
3521 assert!(escape_mapping("hello world!").is_empty());
3523
3524 assert_eq!(escape_mapping(r#"\"\""#), &[(1, 2), (2, 4)]);
3527 assert_eq!(escape_mapping(r#"\u0022\u0022"#), &[(1, 6), (2, 12)]);
3528 assert_eq!(
3529 escape_mapping(r#"\U00000022\U00000022"#),
3530 &[(1, 10), (2, 20)]
3531 );
3532
3533 assert_eq!(
3535 escape_mapping(r#"\"foo\u0022 == \U00000022bar\" && \"\" == \"\n\""#),
3536 &[
3537 (1, 2), (5, 11), (10, 25), (14, 30), (19, 36), (20, 38), (25, 44), (26, 46), (27, 48) ]
3547 );
3548 }
3549
3550 #[test]
3551 fn conditional_args_serialization() {
3552 let error = toml_spanner::from_str::<ConditionalArgs>("condition = 1").unwrap_err();
3554 assert_eq!(
3555 error.to_string(),
3556 "expected a string, found integer at `condition`"
3557 );
3558
3559 let error =
3561 toml_spanner::from_str::<ConditionalArgs>(r#"condition = "foo bar""#).unwrap_err();
3562 assert_eq!(error.to_string(), "expected a single WDL expression");
3563
3564 let error =
3566 toml_spanner::from_str::<ConditionalArgs>(r#"condition = "{ foo: }""#).unwrap_err();
3567 assert_eq!(error.to_string(), "expected expression, but found `}`");
3568
3569 let error = toml_spanner::from_str::<ConditionalArgs>(r#"condition = "1""#).unwrap_err();
3571 assert_eq!(
3572 error.to_string(),
3573 "conditional expression is expected to be type `Boolean`, but found type `Int`"
3574 );
3575 let error = toml_spanner::from_str::<ConditionalArgs>(r#"condition = "hint""#).unwrap_err();
3576 assert_eq!(
3577 error.to_string(),
3578 "conditional expression is expected to be type `Boolean`, but found type `Object`"
3579 );
3580
3581 let error = toml_spanner::from_str::<ConditionalArgs>(r#"condition = "foo""#).unwrap_err();
3583 assert_eq!(error.to_string(), "unknown name `foo`");
3584
3585 let error =
3587 toml_spanner::from_str::<ConditionalArgs>(r#"condition = "Foo {}""#).unwrap_err();
3588 assert_eq!(error.to_string(), "unknown type name `Foo`");
3589
3590 let args: ConditionalArgs = toml_spanner::from_str(r#"condition = "true""#).unwrap();
3592 assert_eq!(args.condition.raw, "true");
3593 assert_eq!(
3594 toml_spanner::to_string(&args).unwrap(),
3595 "condition = \"true\"\nargs = []\n"
3596 );
3597
3598 let args: ConditionalArgs =
3599 toml_spanner::from_str(r#"condition = "cpu == 1 && hint.bar == \"foo\"""#).unwrap();
3600 assert_eq!(args.condition.raw, r#"cpu == 1 && hint.bar == "foo""#);
3601 assert_eq!(
3602 toml_spanner::to_string(&args).unwrap(),
3603 "condition = 'cpu == 1 && hint.bar == \"foo\"'\nargs = []\n"
3604 );
3605 }
3606
3607 #[test]
3608 fn validate_conditional_args() {
3609 let args: ConditionalArgs = toml_spanner::from_str(r#"condition = "true""#).unwrap();
3611 assert_eq!(
3612 args.validate().unwrap_err().to_string(),
3613 "backend conditional arguments must have at least one argument specified"
3614 );
3615 }
3616
3617 #[tokio::test]
3618 async fn evaluate_conditions() {
3619 struct Context {
3621 cpu: f64,
3622 memory: u64,
3623 gpu: bool,
3624 fpga: bool,
3625 disks: i64,
3626 inputs: TaskInputs,
3627 hints: Object,
3628 }
3629
3630 impl Default for Context {
3631 fn default() -> Self {
3632 Self {
3633 cpu: DEFAULT_TASK_REQUIREMENT_CPU,
3634 memory: DEFAULT_TASK_REQUIREMENT_MEMORY as u64,
3635 gpu: false,
3636 fpga: false,
3637 disks: (DEFAULT_TASK_REQUIREMENT_DISKS * ONE_GIBIBYTE) as i64,
3638 inputs: Default::default(),
3639 hints: Default::default(),
3640 }
3641 }
3642 }
3643
3644 struct Transferer;
3645
3646 impl crate::http::Transferer for Transferer {
3647 fn download<'a>(&'a self, _: &'a Url) -> BoxFuture<'a, Result<Location>> {
3648 unimplemented!()
3649 }
3650
3651 fn upload<'a>(&'a self, _: &'a Path, _: &'a Url) -> BoxFuture<'a, Result<()>> {
3652 unimplemented!()
3653 }
3654
3655 fn size<'a>(&'a self, _: &'a Url) -> BoxFuture<'a, anyhow::Result<Option<u64>>> {
3656 unimplemented!()
3657 }
3658
3659 fn walk<'a>(&'a self, _: &'a Url) -> BoxFuture<'a, Result<Arc<[String]>>> {
3660 unimplemented!()
3661 }
3662
3663 fn exists<'a>(&'a self, _: &'a Url) -> BoxFuture<'a, Result<bool>> {
3664 unimplemented!()
3665 }
3666
3667 fn digest<'a>(
3668 &'a self,
3669 _: &'a Url,
3670 ) -> BoxFuture<'a, Result<Option<Arc<cloud_copy::ContentDigest>>>> {
3671 unimplemented!()
3672 }
3673 }
3674
3675 async fn eval(context: Context, expression: &str) -> Result<bool> {
3679 let dir = tempdir().context("failed to create temporary directory")?;
3680 let condition = Condition::new(expression).expect("invalid expression");
3681 condition
3682 .evaluate(
3683 &ExecuteTaskRequest {
3684 id: "test",
3685 command: "",
3686 inputs: &context.inputs,
3687 backend_inputs: &[],
3688 requirements: &Object::empty(),
3689 hints: &context.hints,
3690 env: &Default::default(),
3691 constraints: &TaskExecutionConstraints {
3692 container: None,
3693 cpu: context.cpu,
3694 memory: context.memory,
3695 gpu: if context.gpu {
3696 vec![String::new()]
3697 } else {
3698 Default::default()
3699 },
3700 fpga: if context.fpga {
3701 vec![String::new()]
3702 } else {
3703 Default::default()
3704 },
3705 disks: IndexMap::from_iter([("".into(), context.disks)]),
3706 },
3707 base_dir: &EvaluationPath::from_local_path(dir.path().into()),
3708 attempt_dir: &dir.path().join("0"),
3709 temp_dir: &dir.path().join("tmp"),
3710 },
3711 &Transferer,
3712 )
3713 .await
3714 }
3715
3716 assert_eq!(eval(Context::default(), "true").await.unwrap(), true);
3718 assert_eq!(eval(Context::default(), "false").await.unwrap(), false);
3719 assert_eq!(eval(Context::default(), "cpu == 1").await.unwrap(), true);
3720 assert_eq!(
3721 eval(Context::default(), "memory == 2147483648")
3722 .await
3723 .unwrap(),
3724 true
3725 );
3726 assert_eq!(eval(Context::default(), "gpu").await.unwrap(), false);
3727 assert_eq!(eval(Context::default(), "fpga").await.unwrap(), false);
3728 assert_eq!(
3729 eval(Context::default(), "disks == 1073741824")
3730 .await
3731 .unwrap(),
3732 true
3733 );
3734 assert_eq!(
3735 eval(Context::default(), "defined(hint.foo)").await.unwrap(),
3736 false
3737 );
3738
3739 assert_eq!(
3741 eval(
3742 Context {
3743 cpu: 10.,
3744 memory: 10 * 1024 * 1024,
3745 gpu: true,
3746 fpga: true,
3747 disks: 1024 * 1024,
3748 inputs: Default::default(),
3749 hints: Object::new(IndexMap::from_iter([(
3750 "foo".into(),
3751 "hi".to_string().into()
3752 )]))
3753 },
3754 r#"cpu == 10 && memory == 10*1024*1024 && gpu && fpga && disks == 1024 * 1024 && hint.foo == "hi""#
3755 )
3756 .await
3757 .unwrap(),
3758 true
3759 );
3760
3761 let mut context = Context {
3763 hints: Object::new(IndexMap::from_iter([(
3764 "foo".into(),
3765 "hi".to_string().into(),
3766 )])),
3767 ..Default::default()
3768 };
3769 context
3770 .inputs
3771 .override_hint("foo", "overridden!".to_string());
3772 assert_eq!(
3773 eval(context, r#"hint.foo == "overridden!""#).await.unwrap(),
3774 true
3775 );
3776 }
3777}