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)
375 .with_cwd(Some(REMOTE_WORKDIR.into()))
376 .with_timeout_sec(Some(delete_timeout.as_secs_f64()));
377 let result = self.run_sync(request, delete_wait(delete_timeout)).await?;
378 match result.outcome {
379 Some(outcome) if outcome.exit_code == Some(0) => deleted = count,
380 outcome => {
381 let detail = outcome
382 .map(|outcome| {
383 let stderr = outcome.stderr.trim();
384 if stderr.is_empty() {
385 format!("exit {:?}", outcome.exit_code)
386 } else {
387 stderr.to_string()
388 }
389 })
390 .unwrap_or_else(|| "still running at the deadline".into());
391 return Err(Error::new(
392 ErrorKind::ExecFailed,
393 format!(
394 "the in-guest removal of {count} deleted path(s) failed: {detail}. \
395 The uploaded members landed; the guest manifest was left as it \
396 was, so the next sync will order these deletions again."
397 ),
398 ));
399 }
400 }
401 }
402
403 let body = serde_json::to_vec(&local).map_err(|error| {
404 Error::new(
405 ErrorKind::Unexpected,
406 format!("the manifest will not serialize: {error}"),
407 )
408 })?;
409 self.upload_file(MANIFEST_PATH, &body, None).await?;
410
411 Ok(SyncPass {
412 uploaded_bytes,
413 uploaded_members,
414 deleted,
415 refused_deletions,
416 full,
417 unchanged: false,
418 manifest: local,
419 })
420 }
421
422 pub async fn download_dir(
427 &self,
428 tree: &dyn LocalTree,
429 remote: &str,
430 globs: &[String],
431 dir: &Path,
432 ) -> Result<Vec<Artifact>, Error> {
433 let archive = self.download_tar(remote).await?;
434 Ok(tree.extract(&archive, globs, dir)?)
435 }
436}
437
438#[cfg(test)]
439mod tests {
440 use std::sync::{Arc, Mutex};
441
442 use super::*;
443 use crate::session::testing::{Recorder, Reply, session_with};
444
445 #[derive(Default)]
448 struct FakeTree {
449 manifest: Manifest,
450 extracted: Mutex<Vec<(Vec<u8>, Vec<String>)>>,
451 }
452
453 impl LocalTree for FakeTree {
454 fn manifest(&self, _dir: &Path) -> Result<Manifest, WorkspaceError> {
455 Ok(self.manifest.clone())
456 }
457
458 fn pack(&self, _dir: &Path) -> Result<Packed, WorkspaceError> {
459 Err(WorkspaceError::new(
460 "a sync pass packs only the changed paths",
461 ))
462 }
463
464 fn pack_paths(&self, _dir: &Path, paths: &[String]) -> Result<Packed, WorkspaceError> {
465 Ok(Packed {
466 archive: paths.join("\n").into_bytes(),
467 members: paths.len(),
468 })
469 }
470
471 fn extract(
472 &self,
473 archive: &[u8],
474 globs: &[String],
475 _dir: &Path,
476 ) -> Result<Vec<Artifact>, WorkspaceError> {
477 self.extracted
478 .lock()
479 .expect("not poisoned")
480 .push((archive.to_vec(), globs.to_vec()));
481 Ok(vec![Artifact {
482 path: "dist/app".into(),
483 bytes: 3,
484 }])
485 }
486 }
487
488 fn files(entries: &[(&str, &str)]) -> Manifest {
489 Manifest {
490 version: MANIFEST_VERSION,
491 files: entries
492 .iter()
493 .map(|(path, hash)| ((*path).to_string(), (*hash).to_string()))
494 .collect(),
495 ..Manifest::default()
496 }
497 }
498
499 fn exited(exit_code: i32) -> Reply {
500 Reply::ok(serde_json::json!({
501 "exec_id": "x", "phase": "exited", "exit_code": exit_code, "signal": null,
502 "stdout": "", "stderr": "", "truncated": false, "writers_may_be_alive": false
503 }))
504 }
505
506 #[tokio::test]
514 async fn a_pass_after_one_edit_uploads_that_member_and_deletes_nothing() {
515 let recorder = Recorder::with([Reply::Body(200, Vec::new()), Reply::Body(200, Vec::new())]);
516 let (session, _, _) = session_with(Arc::clone(&recorder));
517 let tree = FakeTree {
518 manifest: files(&[("a.txt", "edited"), ("b.txt", "same")]),
519 ..FakeTree::default()
520 };
521 let baseline = files(&[("a.txt", "old"), ("b.txt", "same")]);
522
523 let pass = session
524 .sync_pass(
525 &tree,
526 Path::new("/tmp/project"),
527 Some(baseline),
528 Duration::from_secs(60),
529 )
530 .await
531 .expect("syncs");
532
533 let requests = recorder.requests();
534 assert_eq!(requests.len(), 2, "{requests:?}");
535 assert_eq!(requests[0].path, "/v1/fs/tar?path=%2Fworkspace");
536 assert_eq!(requests[0].body, b"a.txt", "only the edited member travels");
537 assert_eq!(
538 requests[1].path,
539 "/v1/fs/file?path=%2Fworkspace%2F.microvm-sync-manifest.json"
540 );
541 let written: Manifest = serde_json::from_slice(&requests[1].body).expect("a manifest");
542 assert_eq!(written, tree.manifest);
543 assert_eq!((pass.uploaded_members, pass.deleted), (1, 0));
544 assert!(!pass.full && !pass.unchanged);
545 }
546
547 #[tokio::test]
549 async fn an_unchanged_tree_sends_nothing() {
550 let recorder = Recorder::with([]);
551 let (session, _, _) = session_with(Arc::clone(&recorder));
552 let tree = FakeTree {
553 manifest: files(&[("a.txt", "same")]),
554 ..FakeTree::default()
555 };
556 let pass = session
557 .sync_pass(
558 &tree,
559 Path::new("/tmp/p"),
560 Some(tree.manifest.clone()),
561 Duration::from_secs(60),
562 )
563 .await
564 .expect("syncs");
565 assert!(pass.unchanged);
566 assert!(recorder.requests().is_empty());
567 }
568
569 #[tokio::test(start_paused = true)]
579 async fn a_hostile_deletion_path_is_refused_not_executed() {
580 let recorder = Recorder::with([
581 Reply::ok(serde_json::json!({"exec_id": "x", "phase": "running"})),
582 exited(0),
583 exited(0),
584 Reply::Body(200, Vec::new()),
585 ]);
586 let (session, _, _) = session_with(Arc::clone(&recorder));
587 let tree = FakeTree {
588 manifest: files(&[("kept.txt", "same")]),
589 ..FakeTree::default()
590 };
591 let baseline = files(&[
592 ("kept.txt", "same"),
593 ("sub/gone.txt", "x"),
594 ("../../etc/passwd", "x"),
595 ]);
596
597 let pass = session
598 .sync_pass(
599 &tree,
600 Path::new("/tmp/p"),
601 Some(baseline),
602 Duration::from_secs(60),
603 )
604 .await
605 .expect("syncs");
606
607 let start = recorder
608 .requests()
609 .into_iter()
610 .find(|request| request.path == "/v1/exec/start")
611 .expect("an rm went out");
612 let body: serde_json::Value = serde_json::from_slice(&start.body).expect("JSON");
613 assert_eq!(
614 body["command"],
615 serde_json::json!(["rm", "-rf", "--", "sub/gone.txt"]),
616 "the rm carries only the deletable path"
617 );
618 assert_eq!(body["cwd"], "/workspace");
619 assert_eq!(body["timeout_sec"], 60.0);
620 assert!(
621 body["user"].is_null(),
622 "manifest deletion must not elevate above the image default user"
623 );
624 assert_eq!((pass.deleted, pass.refused_deletions), (1, 1));
625 }
626
627 #[tokio::test(start_paused = true)]
630 async fn a_failed_removal_fails_the_pass_and_keeps_the_old_manifest() {
631 let recorder = Recorder::with([
632 Reply::ok(serde_json::json!({"exec_id": "x", "phase": "running"})),
633 exited(1),
634 exited(1),
635 ]);
636 let (session, _, _) = session_with(Arc::clone(&recorder));
637 let tree = FakeTree::default();
638 let err = session
639 .sync_pass(
640 &tree,
641 Path::new("/tmp/p"),
642 Some(files(&[("gone.txt", "x")])),
643 Duration::from_secs(60),
644 )
645 .await
646 .expect_err("the rm failed");
647 assert_eq!(err.code(), "ERR_EXEC_FAILED", "{err}");
648 assert!(err.to_string().contains("next sync will order"), "{err}");
649 assert!(
650 !recorder
651 .requests()
652 .iter()
653 .any(|request| request.path.contains("manifest")),
654 "the manifest was rewritten after a failed removal"
655 );
656 }
657
658 #[tokio::test]
660 async fn the_guest_manifest_is_none_when_absent_or_unreadable() {
661 let written = files(&[("a.txt", "h")]);
662 let recorder = Recorder::with([
663 Reply::Body(404, b"no such file".to_vec()),
664 Reply::Body(200, b"not json".to_vec()),
665 Reply::Body(200, serde_json::to_vec(&written).expect("JSON")),
666 Reply::Body(401, b"wrong token".to_vec()),
667 ]);
668 let (session, _, _) = session_with(recorder);
669 assert_eq!(session.sync_manifest().await.expect("absent"), None);
670 assert_eq!(session.sync_manifest().await.expect("unreadable"), None);
671 assert_eq!(session.sync_manifest().await.expect("read"), Some(written));
672 let err = session
673 .sync_manifest()
674 .await
675 .expect_err("a 401 is not an absence");
676 assert_eq!(err.code(), "ERR_CREDENTIALS");
677 }
678
679 #[tokio::test]
685 async fn sync_dir_diffs_against_the_guests_manifest_unless_full() {
686 let tree = FakeTree {
687 manifest: files(&[("a.txt", "same"), ("b.txt", "new")]),
688 ..FakeTree::default()
689 };
690 let guest = files(&[("a.txt", "same")]);
691 let recorder = Recorder::with([
692 Reply::Body(200, serde_json::to_vec(&guest).expect("JSON")),
693 Reply::Body(200, Vec::new()),
694 Reply::Body(200, Vec::new()),
695 Reply::Body(200, Vec::new()),
696 Reply::Body(200, Vec::new()),
697 ]);
698 let (session, _, _) = session_with(Arc::clone(&recorder));
699
700 let incremental = session
701 .sync_dir(
702 &tree,
703 Path::new("/tmp/p"),
704 false,
705 DEFAULT_SYNC_DELETE_TIMEOUT,
706 )
707 .await
708 .expect("syncs");
709 assert_eq!(incremental.uploaded_members, 1, "only b.txt is new");
710 assert!(!incremental.full);
711
712 let full = session
713 .sync_dir(
714 &tree,
715 Path::new("/tmp/p"),
716 true,
717 DEFAULT_SYNC_DELETE_TIMEOUT,
718 )
719 .await
720 .expect("syncs");
721 assert!(full.full);
722 assert_eq!(full.uploaded_members, 2, "a full sync sends everything");
723 let paths: Vec<String> = recorder.requests().into_iter().map(|r| r.path).collect();
724 assert_eq!(
725 paths[3], "/v1/fs/tar?path=%2Fworkspace",
726 "a full sync starts with the upload, not a manifest read: {paths:?}"
727 );
728 }
729
730 #[tokio::test]
733 async fn download_dir_extracts_the_remote_archive_with_the_callers_globs() {
734 let recorder = Recorder::with([Reply::Body(200, b"tar bytes".to_vec())]);
735 let (session, _, _) = session_with(Arc::clone(&recorder));
736 let tree = FakeTree::default();
737 let got = session
738 .download_dir(
739 &tree,
740 "/workspace/out",
741 &["dist/**".into()],
742 Path::new("/tmp/d"),
743 )
744 .await
745 .expect("downloads");
746 assert_eq!(got.len(), 1);
747 assert_eq!(recorder.last().path, "/v1/fs/tar?path=%2Fworkspace%2Fout");
748 assert_eq!(
749 *tree.extracted.lock().expect("not poisoned"),
750 [(b"tar bytes".to_vec(), vec!["dist/**".to_string()])]
751 );
752 }
753
754 #[test]
757 fn the_pack_budgets_are_the_daemons_caps_inclusive() {
758 assert_eq!(MAX_PACK_BYTES, 536_870_912);
759 assert_eq!(MAX_PACK_MEMBERS, 100_000);
760 assert!(within_byte_budget(536_870_912));
761 assert!(!within_byte_budget(536_870_913));
762 assert!(within_member_budget(100_000));
763 assert!(!within_member_budget(100_001));
764 }
765
766 #[test]
772 fn a_sync_waits_its_budget_plus_the_delete_grace() {
773 assert_eq!(
774 delete_wait(Duration::from_secs(60)),
775 Duration::from_secs(90)
776 );
777 assert_eq!(delete_wait(Duration::ZERO), Duration::from_secs(30));
778 }
779
780 #[test]
783 fn a_workspace_error_is_invalid_arg_with_itself_as_the_source() {
784 let err = Error::from(WorkspaceError::new("the tree exceeds the budget"));
785 assert_eq!(err.to_string(), "the tree exceeds the budget");
786 assert_eq!(err.code(), "ERR_INVALID_ARG");
787 assert_eq!(err.wire_kind(), None);
788 let source = std::error::Error::source(&err).expect("a source");
789 assert_eq!(
790 source.downcast_ref::<WorkspaceError>(),
791 Some(&WorkspaceError::new("the tree exceeds the budget"))
792 );
793 }
794
795 #[test]
798 fn a_diff_names_changed_new_and_deleted_members_and_nothing_else() {
799 let mut remote = Manifest {
800 version: MANIFEST_VERSION,
801 ..Manifest::default()
802 };
803 remote.files.insert("same.txt".into(), "hash-same".into());
804 remote.files.insert("edited.txt".into(), "hash-old".into());
805 remote.files.insert("removed.txt".into(), "hash-x".into());
806 remote.dirs.insert("gone-dir".into());
807 remote.dirs.insert("gone-dir/nested".into());
808 remote.symlinks.insert("link".into(), "old-target".into());
809
810 let mut local = Manifest {
811 version: MANIFEST_VERSION,
812 ..Manifest::default()
813 };
814 local.files.insert("same.txt".into(), "hash-same".into());
815 local.files.insert("edited.txt".into(), "hash-new".into());
816 local.files.insert("added.txt".into(), "hash-add".into());
817 local.symlinks.insert("link".into(), "new-target".into());
818 remote.dirs.insert("kept-dir".into());
819 local.dirs.insert("kept-dir".into());
820 local.dirs.insert("new-dir".into());
821
822 let delta = diff(&local, &remote);
823 assert_eq!(delta.upload, ["added.txt", "edited.txt", "link", "new-dir"]);
824 assert_eq!(delta.delete, ["removed.txt", "gone-dir/nested", "gone-dir"]);
826 }
827
828 #[test]
834 fn a_diff_never_deletes_the_manifest_itself() {
835 let mut remote = Manifest::default();
836 remote.files.insert(MANIFEST_NAME.into(), "hash".into());
837 let delta = diff(&Manifest::default(), &remote);
838 assert!(delta.is_empty(), "{delta:?}");
839 }
840
841 #[test]
844 fn only_a_plain_relative_path_is_deletable() {
845 for kept in ["a.txt", "src/lib.rs", "a/b/c", ".hidden", "x..y"] {
846 assert!(deletable(kept), "{kept}");
847 }
848 for refused in [
849 "",
850 "/etc/passwd",
851 "../x",
852 "a/../b",
853 "a//b",
854 "a/",
855 "a\\b",
856 "..",
857 ] {
858 assert!(!deletable(refused), "{refused:?}");
859 }
860 }
861}