1use std::collections::{BTreeMap, BTreeSet};
29use std::path::Path;
30use std::time::Duration;
31
32use crate::error::{Error, ErrorKind, WireKind};
33use crate::session::Session;
34
35pub const SKIPPED_DIRS: [&str; 4] = [".git", "target", "node_modules", ".venv"];
42
43pub const MAX_PACK_BYTES: u64 = 512 * 1024 * 1024;
46
47pub const MAX_PACK_MEMBERS: usize = 100_000;
49
50pub fn within_byte_budget(total: u64) -> bool {
53 total <= MAX_PACK_BYTES
54}
55
56pub fn within_member_budget(count: usize) -> bool {
58 count <= MAX_PACK_MEMBERS
59}
60
61pub const REMOTE_WORKDIR: &str = "/workspace";
66
67pub const MANIFEST_PATH: &str = "/workspace/.microvm-sync-manifest.json";
74
75pub const MANIFEST_NAME: &str = ".microvm-sync-manifest.json";
78
79pub const MANIFEST_VERSION: u32 = 1;
81
82pub const DEFAULT_SYNC_DELETE_TIMEOUT: Duration = Duration::from_secs(60);
85
86pub const SYNC_DELETE_GRACE: Duration = Duration::from_secs(30);
92
93fn delete_wait(budget: Duration) -> Duration {
97 budget.saturating_add(SYNC_DELETE_GRACE)
98}
99
100#[derive(Debug, Clone, Eq, PartialEq)]
106pub struct WorkspaceError(String);
107
108impl WorkspaceError {
109 pub fn new(message: impl Into<String>) -> Self {
110 Self(message.into())
111 }
112}
113
114impl std::fmt::Display for WorkspaceError {
115 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116 self.0.fmt(f)
117 }
118}
119
120impl std::error::Error for WorkspaceError {}
121
122impl From<WorkspaceError> for Error {
123 fn from(error: WorkspaceError) -> Self {
126 Error::new(ErrorKind::InvalidArg, error.to_string()).with_source(error)
127 }
128}
129
130#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
135pub struct Manifest {
136 pub version: u32,
138 pub files: BTreeMap<String, String>,
140 pub symlinks: BTreeMap<String, String>,
143 pub dirs: BTreeSet<String>,
145}
146
147#[derive(Debug, Default, PartialEq)]
149pub struct Delta {
150 pub upload: Vec<String>,
152 pub delete: Vec<String>,
155}
156
157impl Delta {
158 pub fn is_empty(&self) -> bool {
160 self.upload.is_empty() && self.delete.is_empty()
161 }
162}
163
164pub fn diff(local: &Manifest, remote: &Manifest) -> Delta {
170 let mut delta = Delta::default();
171 for (path, hash) in &local.files {
172 if remote.files.get(path) != Some(hash) {
173 delta.upload.push(path.clone());
174 }
175 }
176 for (path, target) in &local.symlinks {
177 if remote.symlinks.get(path) != Some(target) {
178 delta.upload.push(path.clone());
179 }
180 }
181 for path in &local.dirs {
182 if !remote.dirs.contains(path) {
183 delta.upload.push(path.clone());
184 }
185 }
186 let lives_on = |path: &String| {
187 local.files.contains_key(path)
188 || local.symlinks.contains_key(path)
189 || local.dirs.contains(path)
190 || path == MANIFEST_NAME
191 };
192 delta.delete.extend(
193 remote
194 .files
195 .keys()
196 .chain(remote.symlinks.keys())
197 .chain(remote.dirs.iter())
198 .filter(|path| !lives_on(path))
199 .cloned(),
200 );
201 delta.upload.sort();
202 delta.delete.sort_by(|a, b| b.cmp(a));
204 delta
205}
206
207pub fn deletable(path: &str) -> bool {
214 !path.is_empty()
215 && !path.starts_with('/')
216 && !path.contains('\\')
217 && path
218 .split('/')
219 .all(|component| !component.is_empty() && component != "..")
220}
221
222#[derive(Debug)]
224pub struct Packed {
225 pub archive: Vec<u8>,
226 pub members: usize,
227}
228
229#[derive(Clone, Debug, Eq, PartialEq)]
231pub struct Artifact {
232 pub path: String,
234 pub bytes: u64,
235}
236
237pub trait LocalTree: Send + Sync {
244 fn manifest(&self, dir: &Path) -> Result<Manifest, WorkspaceError>;
246
247 fn pack(&self, dir: &Path) -> Result<Packed, WorkspaceError>;
250
251 fn pack_paths(&self, dir: &Path, paths: &[String]) -> Result<Packed, WorkspaceError>;
253
254 fn extract(
259 &self,
260 archive: &[u8],
261 globs: &[String],
262 dir: &Path,
263 ) -> Result<Vec<Artifact>, WorkspaceError>;
264}
265
266#[derive(Debug)]
268pub struct SyncPass {
269 pub uploaded_bytes: usize,
271 pub uploaded_members: usize,
272 pub deleted: usize,
274 pub refused_deletions: usize,
276 pub full: bool,
278 pub unchanged: bool,
280 pub manifest: Manifest,
282}
283
284impl Session {
285 pub async fn sync_manifest(&self) -> Result<Option<Manifest>, Error> {
291 match self.download_file(MANIFEST_PATH).await {
292 Ok(bytes) => Ok(serde_json::from_slice(&bytes).ok()),
293 Err(error) if error.wire_kind() == Some(WireKind::NotFound) => Ok(None),
294 Err(error) => Err(error),
295 }
296 }
297
298 pub async fn sync_dir(
302 &self,
303 tree: &dyn LocalTree,
304 dir: &Path,
305 full: bool,
306 delete_timeout: Duration,
307 ) -> Result<SyncPass, Error> {
308 let baseline = if full {
309 None
310 } else {
311 self.sync_manifest().await?
312 };
313 self.sync_pass(tree, dir, baseline, delete_timeout).await
314 }
315
316 pub async fn sync_pass(
325 &self,
326 tree: &dyn LocalTree,
327 dir: &Path,
328 baseline: Option<Manifest>,
329 delete_timeout: Duration,
330 ) -> Result<SyncPass, Error> {
331 let local = tree.manifest(dir)?;
332 let full = baseline.is_none();
333 let delta = diff(&local, &baseline.unwrap_or_default());
334 if delta.is_empty() && !full {
335 return Ok(SyncPass {
336 uploaded_bytes: 0,
337 uploaded_members: 0,
338 deleted: 0,
339 refused_deletions: 0,
340 full: false,
341 unchanged: true,
342 manifest: local,
343 });
344 }
345
346 let mut uploaded_bytes = 0;
347 let mut uploaded_members = 0;
348 if !delta.upload.is_empty() {
349 let packed = tree.pack_paths(dir, &delta.upload)?;
350 self.upload_tar(REMOTE_WORKDIR, &packed.archive).await?;
351 uploaded_bytes = packed.archive.len();
352 uploaded_members = packed.members;
353 }
354
355 let doomed: Vec<String> = delta
356 .delete
357 .iter()
358 .filter(|path| deletable(path))
359 .cloned()
360 .collect();
361 let refused_deletions = delta.delete.len() - doomed.len();
362 let mut deleted = 0;
363 if !doomed.is_empty() {
364 let count = doomed.len();
365 let command = ["rm", "-rf", "--"]
366 .into_iter()
367 .map(String::from)
368 .chain(doomed)
369 .collect();
370 let request = protocol::exec::StartRequest::new(self.mint_exec_id(), command)
371 .with_cwd(Some(REMOTE_WORKDIR.into()))
372 .with_timeout_sec(Some(delete_timeout.as_secs_f64()));
373 let result = self.run_sync(request, delete_wait(delete_timeout)).await?;
374 match result.outcome {
375 Some(outcome) if outcome.exit_code == Some(0) => deleted = count,
376 outcome => {
377 let detail = outcome
378 .map(|outcome| {
379 let stderr = outcome.stderr.trim();
380 if stderr.is_empty() {
381 format!("exit {:?}", outcome.exit_code)
382 } else {
383 stderr.to_string()
384 }
385 })
386 .unwrap_or_else(|| "still running at the deadline".into());
387 return Err(Error::new(
388 ErrorKind::ExecFailed,
389 format!(
390 "the in-guest removal of {count} deleted path(s) failed: {detail}. \
391 The uploaded members landed; the guest manifest was left as it \
392 was, so the next sync will order these deletions again."
393 ),
394 ));
395 }
396 }
397 }
398
399 let body = serde_json::to_vec(&local).map_err(|error| {
400 Error::new(
401 ErrorKind::Unexpected,
402 format!("the manifest will not serialize: {error}"),
403 )
404 })?;
405 self.upload_file(MANIFEST_PATH, &body, None).await?;
406
407 Ok(SyncPass {
408 uploaded_bytes,
409 uploaded_members,
410 deleted,
411 refused_deletions,
412 full,
413 unchanged: false,
414 manifest: local,
415 })
416 }
417
418 pub async fn download_dir(
423 &self,
424 tree: &dyn LocalTree,
425 remote: &str,
426 globs: &[String],
427 dir: &Path,
428 ) -> Result<Vec<Artifact>, Error> {
429 let archive = self.download_tar(remote).await?;
430 Ok(tree.extract(&archive, globs, dir)?)
431 }
432}
433
434#[cfg(test)]
435mod tests {
436 use std::sync::{Arc, Mutex};
437
438 use super::*;
439 use crate::session::testing::{Recorder, Reply, session_with};
440
441 #[derive(Default)]
444 struct FakeTree {
445 manifest: Manifest,
446 extracted: Mutex<Vec<(Vec<u8>, Vec<String>)>>,
447 }
448
449 impl LocalTree for FakeTree {
450 fn manifest(&self, _dir: &Path) -> Result<Manifest, WorkspaceError> {
451 Ok(self.manifest.clone())
452 }
453
454 fn pack(&self, _dir: &Path) -> Result<Packed, WorkspaceError> {
455 Err(WorkspaceError::new(
456 "a sync pass packs only the changed paths",
457 ))
458 }
459
460 fn pack_paths(&self, _dir: &Path, paths: &[String]) -> Result<Packed, WorkspaceError> {
461 Ok(Packed {
462 archive: paths.join("\n").into_bytes(),
463 members: paths.len(),
464 })
465 }
466
467 fn extract(
468 &self,
469 archive: &[u8],
470 globs: &[String],
471 _dir: &Path,
472 ) -> Result<Vec<Artifact>, WorkspaceError> {
473 self.extracted
474 .lock()
475 .expect("not poisoned")
476 .push((archive.to_vec(), globs.to_vec()));
477 Ok(vec![Artifact {
478 path: "dist/app".into(),
479 bytes: 3,
480 }])
481 }
482 }
483
484 fn files(entries: &[(&str, &str)]) -> Manifest {
485 Manifest {
486 version: MANIFEST_VERSION,
487 files: entries
488 .iter()
489 .map(|(path, hash)| ((*path).to_string(), (*hash).to_string()))
490 .collect(),
491 ..Manifest::default()
492 }
493 }
494
495 fn exited(exit_code: i32) -> Reply {
496 Reply::ok(serde_json::json!({
497 "exec_id": "x", "phase": "exited", "exit_code": exit_code, "signal": null,
498 "stdout": "", "stderr": "", "truncated": false, "writers_may_be_alive": false
499 }))
500 }
501
502 #[tokio::test]
510 async fn a_pass_after_one_edit_uploads_that_member_and_deletes_nothing() {
511 let recorder = Recorder::with([Reply::Body(200, Vec::new()), Reply::Body(200, Vec::new())]);
512 let (session, _, _) = session_with(Arc::clone(&recorder));
513 let tree = FakeTree {
514 manifest: files(&[("a.txt", "edited"), ("b.txt", "same")]),
515 ..FakeTree::default()
516 };
517 let baseline = files(&[("a.txt", "old"), ("b.txt", "same")]);
518
519 let pass = session
520 .sync_pass(
521 &tree,
522 Path::new("/tmp/project"),
523 Some(baseline),
524 Duration::from_secs(60),
525 )
526 .await
527 .expect("syncs");
528
529 let requests = recorder.requests();
530 assert_eq!(requests.len(), 2, "{requests:?}");
531 assert_eq!(requests[0].path, "/v1/fs/tar?path=%2Fworkspace");
532 assert_eq!(requests[0].body, b"a.txt", "only the edited member travels");
533 assert_eq!(
534 requests[1].path,
535 "/v1/fs/file?path=%2Fworkspace%2F.microvm-sync-manifest.json"
536 );
537 let written: Manifest = serde_json::from_slice(&requests[1].body).expect("a manifest");
538 assert_eq!(written, tree.manifest);
539 assert_eq!((pass.uploaded_members, pass.deleted), (1, 0));
540 assert!(!pass.full && !pass.unchanged);
541 }
542
543 #[tokio::test]
545 async fn an_unchanged_tree_sends_nothing() {
546 let recorder = Recorder::with([]);
547 let (session, _, _) = session_with(Arc::clone(&recorder));
548 let tree = FakeTree {
549 manifest: files(&[("a.txt", "same")]),
550 ..FakeTree::default()
551 };
552 let pass = session
553 .sync_pass(
554 &tree,
555 Path::new("/tmp/p"),
556 Some(tree.manifest.clone()),
557 Duration::from_secs(60),
558 )
559 .await
560 .expect("syncs");
561 assert!(pass.unchanged);
562 assert!(recorder.requests().is_empty());
563 }
564
565 #[tokio::test(start_paused = true)]
573 async fn a_hostile_deletion_path_is_refused_not_executed() {
574 let recorder = Recorder::with([
575 Reply::ok(serde_json::json!({"exec_id": "x", "phase": "running"})),
576 exited(0),
577 exited(0),
578 Reply::Body(200, Vec::new()),
579 ]);
580 let (session, _, _) = session_with(Arc::clone(&recorder));
581 let tree = FakeTree {
582 manifest: files(&[("kept.txt", "same")]),
583 ..FakeTree::default()
584 };
585 let baseline = files(&[
586 ("kept.txt", "same"),
587 ("gone.txt", "x"),
588 ("../../etc/passwd", "x"),
589 ]);
590
591 let pass = session
592 .sync_pass(
593 &tree,
594 Path::new("/tmp/p"),
595 Some(baseline),
596 Duration::from_secs(60),
597 )
598 .await
599 .expect("syncs");
600
601 let start = recorder
602 .requests()
603 .into_iter()
604 .find(|request| request.path == "/v1/exec/start")
605 .expect("an rm went out");
606 let body: serde_json::Value = serde_json::from_slice(&start.body).expect("JSON");
607 assert_eq!(
608 body["command"],
609 serde_json::json!(["rm", "-rf", "--", "gone.txt"]),
610 "the rm carries only the deletable path"
611 );
612 assert_eq!(body["cwd"], "/workspace");
613 assert_eq!(body["timeout_sec"], 60.0);
614 assert_eq!((pass.deleted, pass.refused_deletions), (1, 1));
615 }
616
617 #[tokio::test(start_paused = true)]
620 async fn a_failed_removal_fails_the_pass_and_keeps_the_old_manifest() {
621 let recorder = Recorder::with([
622 Reply::ok(serde_json::json!({"exec_id": "x", "phase": "running"})),
623 exited(1),
624 exited(1),
625 ]);
626 let (session, _, _) = session_with(Arc::clone(&recorder));
627 let tree = FakeTree::default();
628 let err = session
629 .sync_pass(
630 &tree,
631 Path::new("/tmp/p"),
632 Some(files(&[("gone.txt", "x")])),
633 Duration::from_secs(60),
634 )
635 .await
636 .expect_err("the rm failed");
637 assert_eq!(err.code(), "ERR_EXEC_FAILED", "{err}");
638 assert!(err.to_string().contains("next sync will order"), "{err}");
639 assert!(
640 !recorder
641 .requests()
642 .iter()
643 .any(|request| request.path.contains("manifest")),
644 "the manifest was rewritten after a failed removal"
645 );
646 }
647
648 #[tokio::test]
650 async fn the_guest_manifest_is_none_when_absent_or_unreadable() {
651 let written = files(&[("a.txt", "h")]);
652 let recorder = Recorder::with([
653 Reply::Body(404, b"no such file".to_vec()),
654 Reply::Body(200, b"not json".to_vec()),
655 Reply::Body(200, serde_json::to_vec(&written).expect("JSON")),
656 Reply::Body(401, b"wrong token".to_vec()),
657 ]);
658 let (session, _, _) = session_with(recorder);
659 assert_eq!(session.sync_manifest().await.expect("absent"), None);
660 assert_eq!(session.sync_manifest().await.expect("unreadable"), None);
661 assert_eq!(session.sync_manifest().await.expect("read"), Some(written));
662 let err = session
663 .sync_manifest()
664 .await
665 .expect_err("a 401 is not an absence");
666 assert_eq!(err.code(), "ERR_CREDENTIALS");
667 }
668
669 #[tokio::test]
675 async fn sync_dir_diffs_against_the_guests_manifest_unless_full() {
676 let tree = FakeTree {
677 manifest: files(&[("a.txt", "same"), ("b.txt", "new")]),
678 ..FakeTree::default()
679 };
680 let guest = files(&[("a.txt", "same")]);
681 let recorder = Recorder::with([
682 Reply::Body(200, serde_json::to_vec(&guest).expect("JSON")),
683 Reply::Body(200, Vec::new()),
684 Reply::Body(200, Vec::new()),
685 Reply::Body(200, Vec::new()),
686 Reply::Body(200, Vec::new()),
687 ]);
688 let (session, _, _) = session_with(Arc::clone(&recorder));
689
690 let incremental = session
691 .sync_dir(
692 &tree,
693 Path::new("/tmp/p"),
694 false,
695 DEFAULT_SYNC_DELETE_TIMEOUT,
696 )
697 .await
698 .expect("syncs");
699 assert_eq!(incremental.uploaded_members, 1, "only b.txt is new");
700 assert!(!incremental.full);
701
702 let full = session
703 .sync_dir(
704 &tree,
705 Path::new("/tmp/p"),
706 true,
707 DEFAULT_SYNC_DELETE_TIMEOUT,
708 )
709 .await
710 .expect("syncs");
711 assert!(full.full);
712 assert_eq!(full.uploaded_members, 2, "a full sync sends everything");
713 let paths: Vec<String> = recorder.requests().into_iter().map(|r| r.path).collect();
714 assert_eq!(
715 paths[3], "/v1/fs/tar?path=%2Fworkspace",
716 "a full sync starts with the upload, not a manifest read: {paths:?}"
717 );
718 }
719
720 #[tokio::test]
723 async fn download_dir_extracts_the_remote_archive_with_the_callers_globs() {
724 let recorder = Recorder::with([Reply::Body(200, b"tar bytes".to_vec())]);
725 let (session, _, _) = session_with(Arc::clone(&recorder));
726 let tree = FakeTree::default();
727 let got = session
728 .download_dir(
729 &tree,
730 "/workspace/out",
731 &["dist/**".into()],
732 Path::new("/tmp/d"),
733 )
734 .await
735 .expect("downloads");
736 assert_eq!(got.len(), 1);
737 assert_eq!(recorder.last().path, "/v1/fs/tar?path=%2Fworkspace%2Fout");
738 assert_eq!(
739 *tree.extracted.lock().expect("not poisoned"),
740 [(b"tar bytes".to_vec(), vec!["dist/**".to_string()])]
741 );
742 }
743
744 #[test]
747 fn the_pack_budgets_are_the_daemons_caps_inclusive() {
748 assert_eq!(MAX_PACK_BYTES, 536_870_912);
749 assert_eq!(MAX_PACK_MEMBERS, 100_000);
750 assert!(within_byte_budget(536_870_912));
751 assert!(!within_byte_budget(536_870_913));
752 assert!(within_member_budget(100_000));
753 assert!(!within_member_budget(100_001));
754 }
755
756 #[test]
762 fn a_sync_waits_its_budget_plus_the_delete_grace() {
763 assert_eq!(
764 delete_wait(Duration::from_secs(60)),
765 Duration::from_secs(90)
766 );
767 assert_eq!(delete_wait(Duration::ZERO), Duration::from_secs(30));
768 }
769
770 #[test]
773 fn a_workspace_error_is_invalid_arg_with_itself_as_the_source() {
774 let err = Error::from(WorkspaceError::new("the tree exceeds the budget"));
775 assert_eq!(err.to_string(), "the tree exceeds the budget");
776 assert_eq!(err.code(), "ERR_INVALID_ARG");
777 assert_eq!(err.wire_kind(), None);
778 let source = std::error::Error::source(&err).expect("a source");
779 assert_eq!(
780 source.downcast_ref::<WorkspaceError>(),
781 Some(&WorkspaceError::new("the tree exceeds the budget"))
782 );
783 }
784
785 #[test]
788 fn a_diff_names_changed_new_and_deleted_members_and_nothing_else() {
789 let mut remote = Manifest {
790 version: MANIFEST_VERSION,
791 ..Manifest::default()
792 };
793 remote.files.insert("same.txt".into(), "hash-same".into());
794 remote.files.insert("edited.txt".into(), "hash-old".into());
795 remote.files.insert("removed.txt".into(), "hash-x".into());
796 remote.dirs.insert("gone-dir".into());
797 remote.dirs.insert("gone-dir/nested".into());
798 remote.symlinks.insert("link".into(), "old-target".into());
799
800 let mut local = Manifest {
801 version: MANIFEST_VERSION,
802 ..Manifest::default()
803 };
804 local.files.insert("same.txt".into(), "hash-same".into());
805 local.files.insert("edited.txt".into(), "hash-new".into());
806 local.files.insert("added.txt".into(), "hash-add".into());
807 local.symlinks.insert("link".into(), "new-target".into());
808 remote.dirs.insert("kept-dir".into());
809 local.dirs.insert("kept-dir".into());
810 local.dirs.insert("new-dir".into());
811
812 let delta = diff(&local, &remote);
813 assert_eq!(delta.upload, ["added.txt", "edited.txt", "link", "new-dir"]);
814 assert_eq!(delta.delete, ["removed.txt", "gone-dir/nested", "gone-dir"]);
816 }
817
818 #[test]
824 fn a_diff_never_deletes_the_manifest_itself() {
825 let mut remote = Manifest::default();
826 remote.files.insert(MANIFEST_NAME.into(), "hash".into());
827 let delta = diff(&Manifest::default(), &remote);
828 assert!(delta.is_empty(), "{delta:?}");
829 }
830
831 #[test]
834 fn only_a_plain_relative_path_is_deletable() {
835 for kept in ["a.txt", "src/lib.rs", "a/b/c", ".hidden", "x..y"] {
836 assert!(deletable(kept), "{kept}");
837 }
838 for refused in [
839 "",
840 "/etc/passwd",
841 "../x",
842 "a/../b",
843 "a//b",
844 "a/",
845 "a\\b",
846 "..",
847 ] {
848 assert!(!deletable(refused), "{refused:?}");
849 }
850 }
851}