1use std::{
2 collections::BTreeMap,
3 path::{Path, PathBuf},
4};
5
6use dora_core::build::BuildInfo;
7use dora_message::{
8 BuildId, SessionId,
9 common::GitSource,
10 descriptor::{CoreNodeKind, NodeSource, ResolvedNode},
11 id::NodeId,
12};
13use eyre::{Context, ContextCompat};
14
15const FINGERPRINT_SCHEMA_VERSION: u32 = 3;
29
30#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
31pub struct DataflowSession {
32 pub build_id: Option<BuildId>,
33 pub session_id: SessionId,
34 pub git_sources: BTreeMap<NodeId, GitSource>,
35 pub local_build: Option<BuildInfo>,
36 #[serde(default, skip_serializing_if = "Option::is_none")]
40 pub build_fingerprint: Option<String>,
41 #[serde(default, skip_serializing_if = "Option::is_none")]
46 pub resolved_dataflow: Option<dora_message::descriptor::Descriptor>,
47 #[serde(default, skip_serializing_if = "Option::is_none")]
53 pub source_fingerprint: Option<String>,
54}
55
56impl Default for DataflowSession {
57 fn default() -> Self {
58 Self {
59 build_id: None,
60 session_id: SessionId::generate(),
61 git_sources: Default::default(),
62 local_build: Default::default(),
63 build_fingerprint: None,
64 resolved_dataflow: None,
65 source_fingerprint: None,
66 }
67 }
68}
69
70impl DataflowSession {
71 pub fn read_session(dataflow_path: &Path) -> eyre::Result<Self> {
72 let session_file = session_file_path(dataflow_path)?;
73 if session_file.exists() {
74 match deserialize(&session_file) {
75 Ok(parsed) => return Ok(parsed),
76 Err(err) => {
77 tracing::warn!(
78 "failed to read dataflow session file at {}: {err:#}, regenerating (you might need to run `dora build` again)",
79 session_file.display()
80 );
81 }
82 }
83 }
84
85 let default_session = DataflowSession::default();
86 default_session.write_out_for_dataflow(dataflow_path)?;
87 Ok(default_session)
88 }
89
90 pub fn write_out_for_dataflow(&self, dataflow_path: &Path) -> eyre::Result<()> {
91 let session_file = session_file_path(dataflow_path)?;
92 let filename = session_file
93 .file_name()
94 .context("session file has no file name")?
95 .to_str()
96 .context("session file name is no utf8")?;
97 if let Some(parent) = session_file.parent() {
98 std::fs::create_dir_all(parent).context("failed to create out dir")?;
99 }
100 std::fs::write(&session_file, self.serialize()?)
101 .context("failed to write dataflow session file")?;
102 let gitignore = session_file.with_file_name(".gitignore");
103 if gitignore.exists() {
104 let existing =
105 std::fs::read_to_string(&gitignore).context("failed to read gitignore")?;
106 if !existing
107 .lines()
108 .any(|l| l.split_once('/') == Some(("", filename)))
109 {
110 let new = existing + &format!("\n/{filename}\n");
111 std::fs::write(gitignore, new).context("failed to update gitignore")?;
112 }
113 } else {
114 std::fs::write(gitignore, format!("/{filename}\n"))
115 .context("failed to write gitignore")?;
116 }
117 Ok(())
118 }
119
120 fn serialize(&self) -> eyre::Result<String> {
121 serde_yaml::to_string(&self).context("failed to serialize dataflow session file")
122 }
123
124 pub fn fingerprint_source(descriptor: &dora_message::descriptor::Descriptor) -> Option<String> {
129 let yaml = serde_yaml::to_string(descriptor).ok()?;
130 Some(format!("src-v1-{}", fnv1a_64_hex(yaml.as_bytes())))
131 }
132
133 pub fn fingerprint_build_inputs(resolved_nodes: &BTreeMap<NodeId, ResolvedNode>) -> String {
154 let mut canonical = String::new();
155 canonical.push_str(&format!(
159 "dora-session-build-inputs-v{FINGERPRINT_SCHEMA_VERSION}\n"
160 ));
161
162 for (node_id, node) in resolved_nodes {
164 canonical.push_str("node:");
165 canonical.push_str(node_id.as_ref());
166 canonical.push('\n');
167
168 let build_field = node_build_command(&node.kind);
172 match build_field {
173 Some(cmd) => {
174 canonical.push_str("build:");
175 canonical.push_str(cmd);
176 canonical.push('\n');
177 }
178 None => canonical.push_str("build:none\n"),
179 }
180
181 let source = node_source(&node.kind);
183 match source {
184 Some(NodeSource::Local) => canonical.push_str("source:local\n"),
185 Some(NodeSource::GitBranch { repo, rev }) => {
186 canonical.push_str("source:git\nrepo:");
187 canonical.push_str(repo);
188 canonical.push('\n');
189 let (kind, value) = match rev {
190 Some(dora_message::descriptor::GitRepoRev::Branch(v)) => {
191 ("branch", v.as_str())
192 }
193 Some(dora_message::descriptor::GitRepoRev::Tag(v)) => ("tag", v.as_str()),
194 Some(dora_message::descriptor::GitRepoRev::Rev(v)) => ("rev", v.as_str()),
195 None => ("head", ""),
196 };
197 canonical.push_str("rev:");
198 canonical.push_str(kind);
199 canonical.push(':');
200 canonical.push_str(value);
201 canonical.push('\n');
202 }
203 None => canonical.push_str("source:none\n"),
204 }
205
206 if let Some(env) = &node.env {
210 canonical.push_str("env:\n");
211 for (k, v) in env {
212 canonical.push_str(" ");
213 canonical.push_str(k);
214 canonical.push('=');
215 canonical.push_str(&v.to_string());
216 canonical.push('\n');
217 }
218 } else {
219 canonical.push_str("env:none\n");
220 }
221
222 if let Some(deploy) = &node.deploy {
224 if let Some(cwd) = &deploy.working_dir {
225 canonical.push_str("cwd:");
226 canonical.push_str(&cwd.to_string_lossy());
227 canonical.push('\n');
228 } else {
229 canonical.push_str("cwd:none\n");
230 }
231 } else {
232 canonical.push_str("cwd:none\n");
233 }
234
235 if let CoreNodeKind::Runtime(runtime) = &node.kind {
246 let mut by_id: Vec<_> = runtime.operators.iter().collect();
247 by_id.sort_by_key(|op| op.id.as_ref());
248 if by_id.is_empty() {
249 canonical.push_str("operators:none\n");
250 } else {
251 canonical.push_str("operators:\n");
252 for op in by_id {
253 canonical.push_str(" op:");
254 canonical.push_str(op.id.as_ref());
255 canonical.push('\n');
256 match op.config.build.as_deref() {
257 Some(cmd) => {
258 canonical.push_str(" build:");
259 canonical.push_str(cmd);
260 canonical.push('\n');
261 }
262 None => canonical.push_str(" build:none\n"),
263 }
264 let kind_tag = op.config.source.runtime_name();
276 canonical.push_str(" source-kind:");
277 canonical.push_str(kind_tag);
278 canonical.push('\n');
279 }
280 }
281 }
282 }
283
284 fnv1a_64_hex(canonical.as_bytes())
285 }
286
287 pub fn invalidate_if_build_inputs_changed(
297 &mut self,
298 resolved_nodes: &BTreeMap<NodeId, ResolvedNode>,
299 ) -> bool {
300 let current = Self::fingerprint_build_inputs(resolved_nodes);
301 if self.build_fingerprint.as_deref() == Some(¤t) {
302 return false;
303 }
304 let had_build_id = self.build_id.is_some();
305 self.build_id = None;
306 self.local_build = None;
307 self.git_sources.clear();
308 self.resolved_dataflow = None;
310 self.build_fingerprint = Some(current);
311 if had_build_id {
312 tracing::info!(
313 "dataflow build inputs changed since last session — discarding cached \
314 build_id; run `dora build` to rebuild"
315 );
316 }
317 true
318 }
319}
320
321fn node_build_command(kind: &CoreNodeKind) -> Option<&str> {
329 match kind {
330 CoreNodeKind::Custom(custom) => custom.build.as_deref(),
331 CoreNodeKind::Runtime(_) => None,
332 }
333}
334
335fn node_source(kind: &CoreNodeKind) -> Option<&NodeSource> {
337 match kind {
338 CoreNodeKind::Custom(custom) => Some(&custom.source),
339 CoreNodeKind::Runtime(_) => None,
340 }
341}
342
343fn fnv1a_64_hex(bytes: &[u8]) -> String {
347 const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
348 const FNV_PRIME: u64 = 0x100000001b3;
349 let mut hash = FNV_OFFSET_BASIS;
350 for byte in bytes {
351 hash ^= *byte as u64;
352 hash = hash.wrapping_mul(FNV_PRIME);
353 }
354 format!("{hash:016x}")
355}
356
357fn deserialize(session_file: &Path) -> eyre::Result<DataflowSession> {
358 let s = std::fs::read_to_string(session_file).with_context(|| {
359 format!(
360 "failed to read DataflowSession file at {}",
361 session_file.display()
362 )
363 })?;
364 serde_yaml::from_str(&s).with_context(|| {
365 format!(
366 "failed to deserialize DataflowSession file at {}",
367 session_file.display()
368 )
369 })
370}
371
372fn session_file_path(dataflow_path: &Path) -> eyre::Result<PathBuf> {
373 let file_stem = dataflow_path
374 .file_stem()
375 .wrap_err("dataflow path has no file stem")?
376 .to_str()
377 .wrap_err("dataflow file stem is not valid utf-8")?;
378 let session_file = dataflow_path
379 .with_file_name("out")
380 .join(format!("{file_stem}.dora-session.yaml"));
381 Ok(session_file)
382}
383
384#[cfg(test)]
385mod tests {
386 use super::*;
387 use dora_core::descriptor::DescriptorExt;
388 use dora_message::descriptor::Descriptor;
389
390 fn resolved(yaml: &str) -> BTreeMap<NodeId, ResolvedNode> {
394 let desc: Descriptor = serde_yaml::from_str(yaml).expect("yaml parses as Descriptor");
395 desc.resolve_aliases_and_set_defaults()
396 .expect("descriptor resolves cleanly")
397 }
398
399 const BASELINE: &str = "\
400nodes:
401 - id: a
402 path: ./a
403 build: cargo build
404";
405
406 #[test]
407 fn source_fingerprint_changes_with_any_edit() {
408 let base: Descriptor =
411 serde_yaml::from_str("nodes:\n - id: a\n hub: test/x@^0.1\n").unwrap();
412 let fp = DataflowSession::fingerprint_source(&base);
413 assert!(fp.is_some());
414 let edited: Descriptor = serde_yaml::from_str(
415 "nodes:\n - id: a\n hub: test/x@^0.1\n - id: b\n path: ./b\n",
416 )
417 .unwrap();
418 assert_ne!(fp, DataflowSession::fingerprint_source(&edited));
419 let same: Descriptor =
421 serde_yaml::from_str("nodes:\n - id: a\n hub: test/x@^0.1\n").unwrap();
422 assert_eq!(fp, DataflowSession::fingerprint_source(&same));
423 }
424
425 #[test]
426 fn fingerprint_is_stable_for_same_inputs() {
427 let nodes = resolved(BASELINE);
428 let fp1 = DataflowSession::fingerprint_build_inputs(&nodes);
429 let fp2 = DataflowSession::fingerprint_build_inputs(&nodes);
430 assert_eq!(fp1, fp2, "same inputs must produce same fingerprint");
431 }
432
433 #[test]
434 fn fingerprint_changes_when_build_command_changes() {
435 let before = resolved(BASELINE);
436 let after = resolved(
437 "\
438nodes:
439 - id: a
440 path: ./a
441 build: cargo build --release
442",
443 );
444 assert_ne!(
445 DataflowSession::fingerprint_build_inputs(&before),
446 DataflowSession::fingerprint_build_inputs(&after),
447 "build command change must change fingerprint",
448 );
449 }
450
451 #[test]
452 fn fingerprint_changes_when_source_changes_local_to_git() {
453 let local = resolved(BASELINE);
454 let git = resolved(
455 "\
456nodes:
457 - id: a
458 path: ./a
459 git: https://github.com/dora-rs/dora.git
460 build: cargo build
461",
462 );
463 assert_ne!(
464 DataflowSession::fingerprint_build_inputs(&local),
465 DataflowSession::fingerprint_build_inputs(&git),
466 "switching local->git source must change fingerprint",
467 );
468 }
469
470 #[test]
471 fn fingerprint_changes_when_git_rev_changes() {
472 let rev_a = resolved(
473 "\
474nodes:
475 - id: a
476 path: ./a
477 git: https://github.com/dora-rs/dora.git
478 rev: aaaaaaa
479 build: cargo build
480",
481 );
482 let rev_b = resolved(
483 "\
484nodes:
485 - id: a
486 path: ./a
487 git: https://github.com/dora-rs/dora.git
488 rev: bbbbbbb
489 build: cargo build
490",
491 );
492 assert_ne!(
493 DataflowSession::fingerprint_build_inputs(&rev_a),
494 DataflowSession::fingerprint_build_inputs(&rev_b),
495 "different git revs must produce different fingerprints",
496 );
497 }
498
499 #[test]
500 fn fingerprint_changes_when_env_changes() {
501 let without_env = resolved(BASELINE);
502 let with_env = resolved(
503 "\
504nodes:
505 - id: a
506 path: ./a
507 build: cargo build
508 env:
509 RUST_LOG: info
510",
511 );
512 assert_ne!(
513 DataflowSession::fingerprint_build_inputs(&without_env),
514 DataflowSession::fingerprint_build_inputs(&with_env),
515 "adding env must change fingerprint",
516 );
517 }
518
519 #[test]
520 fn fingerprint_unchanged_when_only_path_changes() {
521 let map_a = resolved(
525 "\
526nodes:
527 - id: a
528 path: ./target/debug/a
529 build: cargo build
530",
531 );
532 let map_b = resolved(
533 "\
534nodes:
535 - id: a
536 path: ./target/release/a
537 build: cargo build
538",
539 );
540 assert_eq!(
541 DataflowSession::fingerprint_build_inputs(&map_a),
542 DataflowSession::fingerprint_build_inputs(&map_b),
543 "path-only change MUST NOT change fingerprint (per #1444 policy)",
544 );
545 }
546
547 #[test]
548 fn invalidate_clears_build_metadata_on_mismatch() {
549 let nodes = resolved(BASELINE);
550 let mut session = DataflowSession {
551 build_id: Some(BuildId::generate()),
552 git_sources: BTreeMap::from([(
553 NodeId::from("a".to_string()),
554 GitSource {
555 subdir: None,
556 hub: None,
557 repo: "x".to_string(),
558 commit_hash: "y".to_string(),
559 },
560 )]),
561 ..Default::default()
562 };
563
564 let invalidated = session.invalidate_if_build_inputs_changed(&nodes);
565 assert!(
566 invalidated,
567 "first call against an un-fingerprinted session must invalidate"
568 );
569 assert!(session.build_id.is_none(), "build_id must be cleared");
570 assert!(session.local_build.is_none(), "local_build must be cleared");
571 assert!(
572 session.git_sources.is_empty(),
573 "git_sources must be cleared"
574 );
575 assert!(
576 session.build_fingerprint.is_some(),
577 "new fingerprint must be stored"
578 );
579 }
580
581 #[test]
582 fn invalidate_is_noop_when_fingerprint_matches() {
583 let nodes = resolved(BASELINE);
584 let mut session = DataflowSession::default();
585 session.invalidate_if_build_inputs_changed(&nodes);
587 let post_build_id = Some(BuildId::generate());
589 session.build_id = post_build_id;
590
591 let invalidated = session.invalidate_if_build_inputs_changed(&nodes);
592 assert!(!invalidated, "matching fingerprint must NOT invalidate");
593 assert_eq!(
594 session.build_id, post_build_id,
595 "build_id must survive matching call"
596 );
597 }
598
599 #[test]
600 fn session_roundtrip_with_new_field() {
601 let session = DataflowSession {
602 build_fingerprint: Some("abcdef0123456789".to_string()),
603 ..Default::default()
604 };
605 let yaml = serde_yaml::to_string(&session).unwrap();
606 let parsed: DataflowSession = serde_yaml::from_str(&yaml).unwrap();
607 assert_eq!(
608 parsed.build_fingerprint.as_deref(),
609 Some("abcdef0123456789")
610 );
611 }
612
613 #[test]
614 fn session_roundtrip_backward_compatible_without_field() {
615 let legacy_yaml = "build_id: null\n\
616 session_id: 00000000-0000-0000-0000-000000000000\n\
617 git_sources: {}\n\
618 local_build: null\n";
619 let parsed: DataflowSession =
620 serde_yaml::from_str(legacy_yaml).expect("legacy session must deserialize");
621 assert!(parsed.build_fingerprint.is_none());
622 }
623
624 #[test]
625 fn read_session_corrupt_file_regenerates_with_warning() {
626 let dir = tempfile::tempdir().unwrap();
627 let dataflow_path = dir.path().join("dataflow.yml");
628 std::fs::write(&dataflow_path, "nodes: []").unwrap();
629
630 let out_dir = dir.path().join("out");
631 std::fs::create_dir_all(&out_dir).unwrap();
632 let session_path = out_dir.join("dataflow.dora-session.yaml");
633 std::fs::write(&session_path, "invalid yaml syntax: : :").unwrap();
634
635 let session = DataflowSession::read_session(&dataflow_path).unwrap();
638 assert!(session.build_id.is_none());
639 assert!(session_path.exists());
640
641 let re_parsed = DataflowSession::read_session(&dataflow_path).unwrap();
643 assert_eq!(session.session_id, re_parsed.session_id);
644 }
645
646 #[test]
658 fn fingerprint_unchanged_when_only_deploy_machine_changes() {
659 let on_m1 = resolved(
660 "\
661nodes:
662 - id: a
663 path: ./a
664 build: cargo build
665 deploy:
666 machine: machine-1
667",
668 );
669 let on_m2 = resolved(
670 "\
671nodes:
672 - id: a
673 path: ./a
674 build: cargo build
675 deploy:
676 machine: machine-2
677",
678 );
679 assert_eq!(
680 DataflowSession::fingerprint_build_inputs(&on_m1),
681 DataflowSession::fingerprint_build_inputs(&on_m2),
682 "deploy.machine-only change MUST NOT change fingerprint (per #1444 policy)",
683 );
684 }
685
686 #[test]
689 fn fingerprint_changes_when_deploy_working_dir_changes() {
690 let cwd_a = resolved(
691 "\
692nodes:
693 - id: a
694 path: ./a
695 build: cargo build
696 deploy:
697 working_dir: /tmp/a
698",
699 );
700 let cwd_b = resolved(
701 "\
702nodes:
703 - id: a
704 path: ./a
705 build: cargo build
706 deploy:
707 working_dir: /tmp/b
708",
709 );
710 assert_ne!(
711 DataflowSession::fingerprint_build_inputs(&cwd_a),
712 DataflowSession::fingerprint_build_inputs(&cwd_b),
713 "deploy.working_dir change must change fingerprint",
714 );
715 }
716
717 #[test]
721 fn fingerprint_changes_when_env_value_mutates() {
722 let info = resolved(
723 "\
724nodes:
725 - id: a
726 path: ./a
727 build: cargo build
728 env:
729 RUST_LOG: info
730",
731 );
732 let debug = resolved(
733 "\
734nodes:
735 - id: a
736 path: ./a
737 build: cargo build
738 env:
739 RUST_LOG: debug
740",
741 );
742 assert_ne!(
743 DataflowSession::fingerprint_build_inputs(&info),
744 DataflowSession::fingerprint_build_inputs(&debug),
745 "mutating an env value must change fingerprint",
746 );
747 }
748
749 #[test]
750 fn fingerprint_changes_when_env_key_removed() {
751 let two_keys = resolved(
752 "\
753nodes:
754 - id: a
755 path: ./a
756 build: cargo build
757 env:
758 RUST_LOG: info
759 FEATURE: x
760",
761 );
762 let one_key = resolved(
763 "\
764nodes:
765 - id: a
766 path: ./a
767 build: cargo build
768 env:
769 RUST_LOG: info
770",
771 );
772 assert_ne!(
773 DataflowSession::fingerprint_build_inputs(&two_keys),
774 DataflowSession::fingerprint_build_inputs(&one_key),
775 "removing an env key must change fingerprint",
776 );
777 }
778
779 #[test]
784 fn fingerprint_unchanged_when_only_node_yaml_order_changes() {
785 let ab = resolved(
786 "\
787nodes:
788 - id: a
789 path: ./a
790 build: cargo build
791 - id: b
792 path: ./b
793 build: make
794",
795 );
796 let ba = resolved(
797 "\
798nodes:
799 - id: b
800 path: ./b
801 build: make
802 - id: a
803 path: ./a
804 build: cargo build
805",
806 );
807 assert_eq!(
808 DataflowSession::fingerprint_build_inputs(&ab),
809 DataflowSession::fingerprint_build_inputs(&ba),
810 "node YAML ordering must not affect fingerprint (BTreeMap sorts by id)",
811 );
812 }
813
814 #[test]
819 fn fingerprint_changes_when_operator_build_changes() {
820 let pandas_v2 = resolved(
821 "\
822nodes:
823 - id: foo
824 operators:
825 - id: op
826 python: ./op.py
827 build: pip install pandas==2.0
828 inputs: {}
829 outputs: []
830",
831 );
832 let pandas_v2_1 = resolved(
833 "\
834nodes:
835 - id: foo
836 operators:
837 - id: op
838 python: ./op.py
839 build: pip install pandas==2.1
840 inputs: {}
841 outputs: []
842",
843 );
844 assert_ne!(
845 DataflowSession::fingerprint_build_inputs(&pandas_v2),
846 DataflowSession::fingerprint_build_inputs(&pandas_v2_1),
847 "operator-level build change must change fingerprint",
848 );
849 }
850
851 #[test]
858 fn fingerprint_unchanged_when_only_operator_python_path_changes() {
859 let py_a = resolved(
860 "\
861nodes:
862 - id: foo
863 operators:
864 - id: op
865 python: ./op_a.py
866 inputs: {}
867 outputs: []
868",
869 );
870 let py_b = resolved(
871 "\
872nodes:
873 - id: foo
874 operators:
875 - id: op
876 python: ./op_b.py
877 inputs: {}
878 outputs: []
879",
880 );
881 assert_eq!(
882 DataflowSession::fingerprint_build_inputs(&py_a),
883 DataflowSession::fingerprint_build_inputs(&py_b),
884 "operator script path-only change MUST NOT change fingerprint \
885 (symmetric to Custom `path` exclusion in #1444)",
886 );
887 }
888
889 #[test]
893 fn fingerprint_changes_when_operator_kind_changes() {
894 let python = resolved(
895 "\
896nodes:
897 - id: foo
898 operators:
899 - id: op
900 python: ./op.py
901 inputs: {}
902 outputs: []
903",
904 );
905 let shared_lib = resolved(
906 "\
907nodes:
908 - id: foo
909 operators:
910 - id: op
911 shared-library: ./op
912 inputs: {}
913 outputs: []
914",
915 );
916 assert_ne!(
917 DataflowSession::fingerprint_build_inputs(&python),
918 DataflowSession::fingerprint_build_inputs(&shared_lib),
919 "switching operator kind (Python -> SharedLibrary) must change fingerprint",
920 );
921 }
922
923 #[test]
924 fn fingerprint_unchanged_when_only_operator_yaml_order_changes() {
925 let ab = resolved(
929 "\
930nodes:
931 - id: foo
932 operators:
933 - id: a
934 python: ./a.py
935 inputs: {}
936 outputs: []
937 - id: b
938 python: ./b.py
939 inputs: {}
940 outputs: []
941",
942 );
943 let ba = resolved(
944 "\
945nodes:
946 - id: foo
947 operators:
948 - id: b
949 python: ./b.py
950 inputs: {}
951 outputs: []
952 - id: a
953 python: ./a.py
954 inputs: {}
955 outputs: []
956",
957 );
958 assert_eq!(
959 DataflowSession::fingerprint_build_inputs(&ab),
960 DataflowSession::fingerprint_build_inputs(&ba),
961 "operator YAML ordering within a node must not affect fingerprint",
962 );
963 }
964}