1use crate::async_storage::StorageFuture;
9use crate::{
10 ApplyOptions, AsyncAuthorityStore, AsyncObjectStore, CancellationToken, ConflictKey,
11 ConflictSide, DefaultTextMergeDriver, DriverError, ForkOptions, GenerationId, IdempotencyKey,
12 JoinOutcome, MemoryMergeResolutionCache, MergeDriver, MergeDriverRegistry,
13 MergePlanResolutionError, MergeResolution, OperationId, PublicationPermit,
14 PublicationReservation, ReservationOutcome, UnpublishedMergeCandidate, WorkBudget, Workspace,
15 WorkspaceContextError, WorkspaceContextId, WorkspaceContextRegistry, WorkspaceContextStore,
16 WorkspaceDelete, WorkspaceError, WorkspaceGraph, WorkspaceId, WorkspaceLineageError,
17 WorkspaceLineageStore, WorkspaceName, WorkspaceRootId,
18};
19use serde::{Deserialize, Serialize};
20use std::collections::{BTreeMap, BTreeSet};
21use std::future::Future;
22use std::sync::Mutex;
23use thiserror::Error;
24
25const MULTI_ROOT_VERSION: u32 = 2;
26
27#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
29pub struct MultiRootMergeRoot {
30 pub source_workspace_id: WorkspaceId,
32 #[serde(default)]
35 pub merge_workspace_id: Option<WorkspaceId>,
36 pub source_generation: GenerationId,
38 pub target_workspace_id: WorkspaceId,
40 pub target_generation: GenerationId,
42 pub base_generation: GenerationId,
44 #[serde(default)]
47 pub child_wins_bindings: BTreeSet<String>,
48}
49
50#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
52pub struct MultiRootMergePlan {
53 pub operation_id: OperationId,
55 pub parent_context_id: WorkspaceContextId,
57 pub child_context_id: WorkspaceContextId,
59 pub roots: BTreeMap<WorkspaceRootId, MultiRootMergeRoot>,
61}
62
63#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
65pub struct MultiRootMergeCandidate {
66 pub plan: MultiRootMergePlan,
68 pub resolutions: BTreeMap<WorkspaceRootId, BTreeMap<ConflictKey, MergeResolution>>,
70}
71
72#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
75pub struct MultiRootFence {
76 pub token: Vec<u8>,
78}
79
80#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
81struct WorkspacePublicationFence {
82 target_workspace_id: WorkspaceId,
83 reservation: PublicationReservation,
84}
85
86#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
88pub enum MultiRootPublicationPhase {
89 Validating,
91 Conflicted,
93 Aborting,
95 Committed,
97 Publishing,
99 Applied,
101 Paused,
103}
104
105#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
107pub struct MultiRootPublication {
108 pub version: u32,
110 pub revision: u64,
112 pub candidate: MultiRootMergeCandidate,
114 pub phase: MultiRootPublicationPhase,
116 pub published_roots: BTreeSet<WorkspaceRootId>,
118 #[serde(default)]
120 pub published_generations: BTreeMap<WorkspaceRootId, GenerationId>,
121 pub fences: BTreeMap<WorkspaceRootId, MultiRootFence>,
123 #[serde(default)]
125 pub conflicts: BTreeMap<WorkspaceRootId, crate::MergePlan>,
126 #[serde(default)]
128 pub projected_roots: BTreeMap<WorkspaceRootId, GenerationId>,
129 #[serde(default)]
131 pub declared_conflicts: BTreeMap<WorkspaceRootId, BTreeSet<ConflictKey>>,
132 pub paused_root: Option<WorkspaceRootId>,
134}
135
136#[derive(Clone, Debug, Eq, PartialEq)]
138pub enum MultiRootPrepare {
139 Ready(MultiRootFence),
141 Stale,
143 Conflicted(Box<crate::MergePlan>),
145}
146
147#[derive(Clone, Debug, Eq, PartialEq)]
149pub struct MultiRootConflictFinish {
150 pub generation: GenerationId,
152 pub fence: MultiRootFence,
154}
155
156pub trait MultiRootPublicationStore: Send + Sync {
158 type Error: std::error::Error + Send + Sync + 'static;
160
161 fn load(
163 &self,
164 operation_id: OperationId,
165 ) -> impl Future<Output = Result<Option<MultiRootPublication>, Self::Error>> + StorageFuture;
166
167 fn compare_and_swap(
169 &self,
170 operation_id: OperationId,
171 expected_revision: u64,
172 replacement: MultiRootPublication,
173 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
174
175 fn list_operations(
177 &self,
178 ) -> impl Future<Output = Result<Vec<OperationId>, Self::Error>> + StorageFuture;
179
180 fn compare_and_delete(
183 &self,
184 operation_id: OperationId,
185 expected_revision: u64,
186 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
187
188 fn claim_parent(
190 &self,
191 parent_context_id: WorkspaceContextId,
192 operation_id: OperationId,
193 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
194
195 fn release_parent(
197 &self,
198 parent_context_id: WorkspaceContextId,
199 operation_id: OperationId,
200 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
201}
202
203pub trait MultiRootPublisher: Send + Sync {
205 type Error: std::error::Error + Send + Sync + 'static;
207
208 fn prepare(
211 &self,
212 operation_id: OperationId,
213 root_id: WorkspaceRootId,
214 root: &MultiRootMergeRoot,
215 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
216 ) -> impl Future<Output = Result<MultiRootPrepare, Self::Error>> + StorageFuture;
217
218 fn release(
220 &self,
221 operation_id: OperationId,
222 root_id: WorkspaceRootId,
223 fence: &MultiRootFence,
224 ) -> impl Future<Output = Result<(), Self::Error>> + StorageFuture;
225
226 fn project_conflict(
228 &self,
229 operation_id: OperationId,
230 root_id: WorkspaceRootId,
231 root: &MultiRootMergeRoot,
232 conflict: Option<&crate::MergePlan>,
233 ) -> impl Future<Output = Result<GenerationId, Self::Error>> + StorageFuture;
234
235 fn abort_conflict(
237 &self,
238 operation_id: OperationId,
239 root_id: WorkspaceRootId,
240 root: &MultiRootMergeRoot,
241 projected_generation: GenerationId,
242 ) -> impl Future<Output = Result<GenerationId, Self::Error>> + StorageFuture;
243
244 fn declare_conflicts(
246 &self,
247 operation_id: OperationId,
248 root_id: WorkspaceRootId,
249 root: &MultiRootMergeRoot,
250 conflict: &crate::MergePlan,
251 paths: &[String],
252 ) -> impl Future<Output = Result<BTreeSet<ConflictKey>, Self::Error>> + StorageFuture;
253
254 fn finish_conflict(
256 &self,
257 operation_id: OperationId,
258 root_id: WorkspaceRootId,
259 root: &MultiRootMergeRoot,
260 projected_generation: GenerationId,
261 ) -> impl Future<Output = Result<MultiRootConflictFinish, Self::Error>> + StorageFuture;
262
263 fn publish(
265 &self,
266 operation_id: OperationId,
267 root_id: WorkspaceRootId,
268 root: &MultiRootMergeRoot,
269 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
270 fence: &MultiRootFence,
271 ) -> impl Future<Output = Result<MultiRootPublishRoot, Self::Error>> + StorageFuture;
272
273 fn finalize(
277 &self,
278 _operation_id: OperationId,
279 _root_id: WorkspaceRootId,
280 _root: &MultiRootMergeRoot,
281 ) -> impl Future<Output = Result<(), Self::Error>> + StorageFuture {
282 async { Ok(()) }
283 }
284}
285
286pub trait WorkspaceResolver<A, O>: Send + Sync {
288 type Error: std::error::Error + Send + Sync + 'static;
290
291 fn resolve(
293 &self,
294 workspace_id: WorkspaceId,
295 ) -> impl Future<Output = Result<Workspace<A, O>, Self::Error>> + StorageFuture;
296}
297
298pub trait MultiRootMaterializer<A, O>: Send + Sync {
304 type Error: std::error::Error + Send + Sync + 'static;
306
307 fn materialize(
309 &self,
310 operation_id: OperationId,
311 root_id: WorkspaceRootId,
312 workspace: &Workspace<A, O>,
313 from: GenerationId,
314 to: GenerationId,
315 ) -> impl Future<Output = Result<(), Self::Error>> + StorageFuture;
316
317 fn finalize(
320 &self,
321 _operation_id: OperationId,
322 _root_id: WorkspaceRootId,
323 ) -> impl Future<Output = Result<(), Self::Error>> + StorageFuture {
324 async { Ok(()) }
325 }
326}
327
328pub struct WorkspaceMultiRootPublisher<A, O, R> {
334 resolver: R,
335 merge_drivers: Option<std::sync::Arc<MergeDriverRegistry>>,
336 root_merge_drivers: BTreeMap<WorkspaceRootId, std::sync::Arc<MergeDriverRegistry>>,
337 resolution_cache: Mutex<MemoryMergeResolutionCache>,
338 marker: std::marker::PhantomData<fn() -> (A, O)>,
339}
340
341pub struct MaterializingWorkspaceMultiRootPublisher<A, O, R, M> {
347 publisher: WorkspaceMultiRootPublisher<A, O, R>,
348 materializer: M,
349}
350
351impl<A, O, R, M> MaterializingWorkspaceMultiRootPublisher<A, O, R, M> {
352 #[must_use]
354 pub fn new(resolver: R, materializer: M) -> Self {
355 Self {
356 publisher: WorkspaceMultiRootPublisher::new(resolver),
357 materializer,
358 }
359 }
360
361 #[must_use]
363 pub fn with_merge_drivers(mut self, registry: std::sync::Arc<MergeDriverRegistry>) -> Self {
364 self.publisher = self.publisher.with_merge_drivers(registry);
365 self
366 }
367
368 #[must_use]
370 pub fn with_root_merge_drivers(
371 mut self,
372 root_id: WorkspaceRootId,
373 registry: std::sync::Arc<MergeDriverRegistry>,
374 ) -> Self {
375 self.publisher = self.publisher.with_root_merge_drivers(root_id, registry);
376 self
377 }
378}
379
380#[derive(Debug, Error)]
382pub enum MaterializingWorkspaceMultiRootPublisherError<
383 R: std::error::Error + 'static,
384 M: std::error::Error + 'static,
385> {
386 #[error(transparent)]
388 Workspace(#[from] WorkspaceMultiRootPublisherError<R>),
389 #[error("published workspace generation is unavailable for materialization")]
391 MissingPublishedGeneration,
392 #[error("root materialization failed: {0}")]
394 Materializer(M),
395}
396
397impl<A, O, R> WorkspaceMultiRootPublisher<A, O, R> {
398 #[must_use]
400 pub fn new(resolver: R) -> Self {
401 Self {
402 resolver,
403 merge_drivers: None,
404 root_merge_drivers: BTreeMap::new(),
405 resolution_cache: Mutex::new(MemoryMergeResolutionCache::default()),
406 marker: std::marker::PhantomData,
407 }
408 }
409
410 #[must_use]
412 pub fn with_merge_drivers(mut self, registry: std::sync::Arc<MergeDriverRegistry>) -> Self {
413 self.merge_drivers = Some(registry);
414 self
415 }
416
417 #[must_use]
419 pub fn with_root_merge_drivers(
420 mut self,
421 root_id: WorkspaceRootId,
422 registry: std::sync::Arc<MergeDriverRegistry>,
423 ) -> Self {
424 self.root_merge_drivers.insert(root_id, registry);
425 self
426 }
427
428 #[must_use]
430 pub const fn resolver(&self) -> &R {
431 &self.resolver
432 }
433}
434
435#[derive(Debug, Error)]
437pub enum WorkspaceMultiRootPublisherError<E: std::error::Error + 'static> {
438 #[error("workspace resolution failed: {0}")]
440 Resolver(E),
441 #[error(transparent)]
443 Workspace(#[from] WorkspaceError),
444 #[error("merge candidate validation failed: {0}")]
446 Candidate(String),
447 #[error("merge requires declarative conflict resolutions")]
449 Conflicted {
450 plan: Box<crate::MergePlan>,
452 },
453 #[error("multi-root publication fence is invalid")]
455 InvalidFence,
456 #[error("multi-root validation workspace could not be reclaimed")]
458 PreviewCleanup,
459}
460
461impl<A, O, R> WorkspaceMultiRootPublisher<A, O, R>
462where
463 A: AsyncAuthorityStore,
464 O: AsyncObjectStore,
465 R: WorkspaceResolver<A, O>,
466{
467 async fn reserve_operation(
468 &self,
469 target: &Workspace<A, O>,
470 operation_id: OperationId,
471 conflict: &'static str,
472 ) -> Result<PublicationReservation, WorkspaceMultiRootPublisherError<R::Error>> {
473 let authority_head = target
474 .authority()
475 .head(
476 target.authority_id(),
477 WorkBudget::UNBOUNDED,
478 &CancellationToken::new(),
479 )
480 .await
481 .map_err(|failure| WorkspaceError::engine(failure.error))?
482 .value;
483 match target
484 .authority()
485 .reserve_publication(
486 target.authority_id(),
487 authority_head,
488 operation_id,
489 WorkBudget::UNBOUNDED,
490 &CancellationToken::new(),
491 )
492 .await
493 .map_err(|failure| WorkspaceError::engine(failure.error))?
494 .value
495 {
496 ReservationOutcome::Reserved(reservation)
497 | ReservationOutcome::AlreadyReserved(reservation) => Ok(reservation),
498 ReservationOutcome::Conflict { .. } => Err(
499 WorkspaceMultiRootPublisherError::Candidate(conflict.to_owned()),
500 ),
501 }
502 }
503
504 async fn release_operation_reservation(
505 &self,
506 target: &Workspace<A, O>,
507 operation_id: OperationId,
508 ) -> Result<(), WorkspaceMultiRootPublisherError<R::Error>> {
509 let reservation = self
510 .reserve_operation(
511 target,
512 operation_id,
513 "publication reservation is owned by another operation",
514 )
515 .await?;
516 target
517 .authority()
518 .release_publication(
519 reservation,
520 WorkBudget::UNBOUNDED,
521 &CancellationToken::new(),
522 )
523 .await
524 .map_err(|failure| WorkspaceError::engine(failure.error))?;
525 Ok(())
526 }
527
528 async fn pinned_plan(
529 &self,
530 root: &MultiRootMergeRoot,
531 target_override: Option<Workspace<A, O>>,
532 ) -> Result<crate::JoinPlan<A, O>, WorkspaceMultiRootPublisherError<R::Error>> {
533 let source = self
534 .resolver
535 .resolve(root.merge_workspace_id.unwrap_or(root.source_workspace_id))
536 .await
537 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
538 let target = match target_override {
539 Some(target) => target,
540 None => self
541 .resolver
542 .resolve(root.target_workspace_id)
543 .await
544 .map_err(WorkspaceMultiRootPublisherError::Resolver)?,
545 };
546 let source_generation = source.generation(root.source_generation).await?;
547 let target_generation = target.generation(root.target_generation).await?;
548 let plan = source
549 .join_into(&target)
550 .plan_pinned(source_generation, target_generation)
551 .await?;
552 if plan.common_ancestor() != root.base_generation {
553 return Err(WorkspaceMultiRootPublisherError::Candidate(
554 "common ancestor changed".to_owned(),
555 ));
556 }
557 Ok(plan)
558 }
559
560 async fn apply_plan(
561 &self,
562 root_id: WorkspaceRootId,
563 root: &MultiRootMergeRoot,
564 plan: &crate::JoinPlan<A, O>,
565 operation: IdempotencyKey,
566 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
567 permit: PublicationPermit,
568 ) -> Result<JoinOutcome<A, O>, WorkspaceMultiRootPublisherError<R::Error>> {
569 let options = ApplyOptions {
570 if_target: plan.target_head(),
571 idempotency_key: operation,
572 };
573 let initial = plan.apply_with_permit(options, permit).await?;
574 let JoinOutcome::Conflicted {
575 conflicts,
576 truncated,
577 } = initial
578 else {
579 return Ok(initial);
580 };
581 let typed = plan.describe_conflicts(&conflicts, truncated).await?;
582 let automatic = child_wins_resolutions(&typed, &root.child_wins_bindings);
583 let candidate = if resolutions.is_empty() && !typed.conflicts.is_empty() {
584 let remaining = crate::MergePlan {
585 conflicts: typed
586 .conflicts
587 .iter()
588 .filter(|conflict| !automatic.contains_key(&conflict.key))
589 .cloned()
590 .collect(),
591 ..typed.clone()
592 };
593 if remaining.conflicts.is_empty() {
594 UnpublishedMergeCandidate {
595 plan: typed.clone(),
596 resolutions: automatic,
597 }
598 } else {
599 let Some(registry) = self
600 .root_merge_drivers
601 .get(&root_id)
602 .or(self.merge_drivers.as_ref())
603 else {
604 return Err(WorkspaceMultiRootPublisherError::Conflicted {
605 plan: Box::new(typed),
606 });
607 };
608 let mut cache = self.resolution_cache.lock().map_err(|_| {
609 WorkspaceMultiRootPublisherError::Candidate(
610 "merge resolution cache is unavailable".to_owned(),
611 )
612 })?;
613 match crate::resolve_merge_plan(remaining, registry, &mut *cache, false) {
614 Ok(mut candidate) => {
615 candidate.plan = typed.clone();
616 candidate.resolutions.extend(automatic);
617 candidate
618 }
619 Err(
620 MergePlanResolutionError::MissingDriver(_)
621 | MergePlanResolutionError::Driver(DriverError::Unsupported),
622 ) => {
623 return Err(WorkspaceMultiRootPublisherError::Conflicted {
624 plan: Box::new(typed),
625 });
626 }
627 Err(error) => {
628 return Err(WorkspaceMultiRootPublisherError::Candidate(
629 error.to_string(),
630 ));
631 }
632 }
633 }
634 } else {
635 UnpublishedMergeCandidate {
636 plan: typed,
637 resolutions: resolutions.clone().into_iter().chain(automatic).collect(),
638 }
639 };
640 if resolutions.is_empty()
641 && candidate.resolutions.values().any(|resolution| {
642 matches!(resolution, MergeResolution::Text(text)
643 if crate::text_merge::has_conflict_markers(text))
644 })
645 {
646 return Err(WorkspaceMultiRootPublisherError::Conflicted {
647 plan: Box::new(candidate.plan.clone()),
648 });
649 }
650 plan.apply_candidate_with_permit(options, &candidate, permit)
651 .await
652 .map_err(|error| WorkspaceMultiRootPublisherError::Candidate(error.to_string()))
653 }
654
655 async fn preview(
656 &self,
657 operation_id: OperationId,
658 root_id: WorkspaceRootId,
659 root: &MultiRootMergeRoot,
660 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
661 ) -> Result<bool, WorkspaceMultiRootPublisherError<R::Error>> {
662 let target = self
663 .resolver
664 .resolve(root.target_workspace_id)
665 .await
666 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
667 if target.head().await?.id() != root.target_generation {
668 return Ok(false);
669 }
670 let preview_key = IdempotencyKey::new();
673 let preview_name = preview_workspace_name(operation_id, root_id, preview_key)?;
674 let target_generation = target.generation(root.target_generation).await?;
675 let preview = target
676 .fork(
677 preview_name.as_str(),
678 ForkOptions::from_generation(target_generation, preview_key),
679 )
680 .await?;
681 let result = async {
682 let preview_generation = preview.head().await?;
683 let preview_root = MultiRootMergeRoot {
684 target_workspace_id: preview.id(),
685 target_generation: preview_generation.id(),
686 ..root.clone()
687 };
688 let plan = self
689 .pinned_plan(&preview_root, Some(preview.clone()))
690 .await?;
691 Ok(matches!(
692 self.apply_plan(
693 root_id,
694 root,
695 &plan,
696 publisher_key(operation_id, root_id, b"preview-apply"),
697 resolutions,
698 PublicationPermit::Unrestricted,
699 )
700 .await?,
701 JoinOutcome::Applied(_)
702 | JoinOutcome::AlreadyApplied(_)
703 | JoinOutcome::NoChanges(_)
704 ))
705 }
706 .await;
707 let cleanup = preview
708 .delete(publisher_key(operation_id, root_id, b"preview-delete"))
709 .await?;
710 if !matches!(
711 cleanup,
712 WorkspaceDelete::Deleted | WorkspaceDelete::AlreadyDeleted
713 ) {
714 return Err(WorkspaceMultiRootPublisherError::PreviewCleanup);
715 }
716 result
717 }
718}
719
720impl<A, O, R> MultiRootPublisher for WorkspaceMultiRootPublisher<A, O, R>
721where
722 A: AsyncAuthorityStore,
723 O: AsyncObjectStore,
724 R: WorkspaceResolver<A, O>,
725{
726 type Error = WorkspaceMultiRootPublisherError<R::Error>;
727
728 async fn prepare(
729 &self,
730 operation_id: OperationId,
731 root_id: WorkspaceRootId,
732 root: &MultiRootMergeRoot,
733 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
734 ) -> Result<MultiRootPrepare, Self::Error> {
735 match self.preview(operation_id, root_id, root, resolutions).await {
736 Ok(true) => {}
737 Ok(false) => return Ok(MultiRootPrepare::Stale),
738 Err(WorkspaceMultiRootPublisherError::Conflicted { plan }) => {
739 return Ok(MultiRootPrepare::Conflicted(plan));
740 }
741 Err(error) => return Err(error),
742 }
743 let target = self
744 .resolver
745 .resolve(root.target_workspace_id)
746 .await
747 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
748 let authority_head = target
749 .authority()
750 .head(
751 target.authority_id(),
752 WorkBudget::UNBOUNDED,
753 &CancellationToken::new(),
754 )
755 .await
756 .map_err(|failure| WorkspaceError::engine(failure.error))?
757 .value;
758 let reservation = match target
759 .authority()
760 .reserve_publication(
761 target.authority_id(),
762 authority_head,
763 operation_id,
764 WorkBudget::UNBOUNDED,
765 &CancellationToken::new(),
766 )
767 .await
768 .map_err(|failure| WorkspaceError::engine(failure.error))?
769 .value
770 {
771 ReservationOutcome::Reserved(reservation)
772 | ReservationOutcome::AlreadyReserved(reservation) => reservation,
773 ReservationOutcome::Conflict { .. } => return Ok(MultiRootPrepare::Stale),
774 };
775 Ok(MultiRootPrepare::Ready(MultiRootFence {
776 token: serde_json::to_vec(&WorkspacePublicationFence {
777 target_workspace_id: target.id(),
778 reservation,
779 })
780 .map_err(|error| WorkspaceMultiRootPublisherError::Candidate(error.to_string()))?,
781 }))
782 }
783
784 async fn release(
785 &self,
786 operation_id: OperationId,
787 _root_id: WorkspaceRootId,
788 fence: &MultiRootFence,
789 ) -> Result<(), Self::Error> {
790 let decoded = decode_reservation(operation_id, fence)?;
791 let target = self
792 .resolver
793 .resolve(decoded.target_workspace_id)
794 .await
795 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
796 target
797 .authority()
798 .release_publication(
799 decoded.reservation,
800 WorkBudget::UNBOUNDED,
801 &CancellationToken::new(),
802 )
803 .await
804 .map_err(|failure| WorkspaceError::engine(failure.error))?;
805 Ok(())
806 }
807
808 async fn project_conflict(
809 &self,
810 operation_id: OperationId,
811 root_id: WorkspaceRootId,
812 root: &MultiRootMergeRoot,
813 conflict: Option<&crate::MergePlan>,
814 ) -> Result<GenerationId, Self::Error> {
815 let target = self
816 .resolver
817 .resolve(root.target_workspace_id)
818 .await
819 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
820 let projection_key = publisher_key(operation_id, root_id, b"conflict-project");
821 let reservation_operation = conflict_projection_operation_id(operation_id, root_id);
822 if let Some(generation) = target.operation_generation(projection_key).await? {
823 self.release_operation_reservation(&target, reservation_operation)
824 .await?;
825 return Ok(generation.id());
826 }
827 if target.head().await?.id() != root.target_generation {
828 return Err(WorkspaceMultiRootPublisherError::Candidate(
829 "conflict projection target changed".to_owned(),
830 ));
831 }
832 let plan = self.pinned_plan(root, Some(target.clone())).await?;
833 let mut resolutions = BTreeMap::new();
834 if let Some(conflict) = conflict {
835 if conflict.truncated {
836 return Err(WorkspaceMultiRootPublisherError::Candidate(
837 "conflict projection plan changed".to_owned(),
838 ));
839 }
840 for view in &conflict.conflicts {
841 let resolution = if view.kind == crate::ConflictKind::Text {
842 DefaultTextMergeDriver
843 .resolve(view)
844 .unwrap_or(MergeResolution::Select(ConflictSide::Ours))
845 } else {
846 MergeResolution::Select(ConflictSide::Ours)
847 };
848 resolutions.insert(view.key.clone(), resolution);
849 }
850 }
851 let reservation = self
852 .reserve_operation(
853 &target,
854 reservation_operation,
855 "conflict projection reservation changed",
856 )
857 .await?;
858 let outcome = self
859 .apply_plan(
860 root_id,
861 root,
862 &plan,
863 projection_key,
864 &resolutions,
865 PublicationPermit::Reservation {
866 operation_id: reservation_operation.into_bytes(),
867 gate_tail: reservation.gate_tail,
868 expected: reservation.expected,
869 },
870 )
871 .await;
872 let release = target
873 .authority()
874 .release_publication(
875 reservation,
876 WorkBudget::UNBOUNDED,
877 &CancellationToken::new(),
878 )
879 .await
880 .map_err(|failure| WorkspaceError::engine(failure.error));
881 if let Err(error) = release {
882 return Err(error.into());
883 }
884 let outcome = outcome?;
885 match outcome {
886 JoinOutcome::Applied(generation)
887 | JoinOutcome::AlreadyApplied(generation)
888 | JoinOutcome::NoChanges(generation) => Ok(generation.id()),
889 _ => Err(WorkspaceMultiRootPublisherError::Candidate(
890 "conflict projection did not produce a generation".to_owned(),
891 )),
892 }
893 }
894
895 async fn abort_conflict(
896 &self,
897 operation_id: OperationId,
898 root_id: WorkspaceRootId,
899 root: &MultiRootMergeRoot,
900 projected_generation: GenerationId,
901 ) -> Result<GenerationId, Self::Error> {
902 let target = self
903 .resolver
904 .resolve(root.target_workspace_id)
905 .await
906 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
907 let key = publisher_key(operation_id, root_id, b"conflict-abort");
908 let reservation_operation =
909 conflict_abort_materialization_operation_id(operation_id, root_id);
910 if let Some(generation) = target.operation_generation(key).await? {
911 self.release_operation_reservation(&target, reservation_operation)
912 .await?;
913 return Ok(generation.id());
914 }
915 let current = target.head().await?;
916 let original = target.generation(root.target_generation).await?;
917 let projected = target.generation(projected_generation).await?;
918 let changed = original
919 .diff_to(&projected, u32::MAX)
920 .await?
921 .changed_paths(u32::MAX)
922 .await?;
923 let paths = changed
924 .iter()
925 .map(|change| {
926 portable_namespace_path(&change.path).ok_or_else(|| {
927 WorkspaceMultiRootPublisherError::Candidate(
928 "conflict abort cannot represent a non-portable path".to_owned(),
929 )
930 })
931 })
932 .collect::<Result<Vec<_>, _>>()?;
933 let reservation = self
934 .reserve_operation(
935 &target,
936 reservation_operation,
937 "conflict abort target changed while being fenced",
938 )
939 .await?;
940 let outcome = target
941 .apply_paths_from_with_permit(
942 Some(&projected),
943 Some(&original),
944 &paths,
945 current.id(),
946 key,
947 PublicationPermit::Reservation {
948 operation_id: reservation_operation.into_bytes(),
949 gate_tail: reservation.gate_tail,
950 expected: reservation.expected,
951 },
952 )
953 .await;
954 let release = target
955 .authority()
956 .release_publication(
957 reservation,
958 WorkBudget::UNBOUNDED,
959 &CancellationToken::new(),
960 )
961 .await
962 .map_err(|failure| WorkspaceError::engine(failure.error));
963 if let Err(error) = release {
964 return Err(error.into());
965 }
966 let outcome = outcome?;
967 match outcome {
968 crate::WorkspacePathApply::Applied(generation)
969 | crate::WorkspacePathApply::AlreadyApplied(generation)
970 | crate::WorkspacePathApply::NoChanges(generation) => Ok(generation.id()),
971 crate::WorkspacePathApply::Conflicted(_)
972 | crate::WorkspacePathApply::Stale(_)
973 | crate::WorkspacePathApply::Fenced
974 | crate::WorkspacePathApply::IdempotencyConflict => {
975 Err(WorkspaceMultiRootPublisherError::Candidate(
976 "conflict abort overlaps later workspace changes".to_owned(),
977 ))
978 }
979 }
980 }
981
982 async fn declare_conflicts(
983 &self,
984 _operation_id: OperationId,
985 _root_id: WorkspaceRootId,
986 root: &MultiRootMergeRoot,
987 conflict: &crate::MergePlan,
988 paths: &[String],
989 ) -> Result<BTreeSet<ConflictKey>, Self::Error> {
990 let target = self
991 .resolver
992 .resolve(root.target_workspace_id)
993 .await
994 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
995 let current = target.head().await?;
996 let normalized = paths
997 .iter()
998 .map(|path| path.trim_start_matches('/').trim_end_matches('/'))
999 .collect::<Vec<_>>();
1000 let declares_all = normalized
1001 .iter()
1002 .any(|candidate| *candidate == "." || candidate.is_empty());
1003 let mut declared = BTreeSet::new();
1004 for view in &conflict.conflicts {
1005 let Some(raw_path) = view.path.as_deref() else {
1006 if declares_all {
1007 declared.insert(view.key.clone());
1008 }
1009 continue;
1010 };
1011 let path = raw_path.trim_start_matches('/');
1012 if !normalized.iter().any(|candidate| {
1013 *candidate == "."
1014 || *candidate == path
1015 || (!candidate.is_empty()
1016 && path
1017 .strip_prefix(*candidate)
1018 .is_some_and(|suffix| suffix.starts_with('/')))
1019 }) {
1020 continue;
1021 }
1022 if view.kind == crate::ConflictKind::Text {
1023 match current.read(&format!("/{path}"), 64 * 1024 * 1024).await {
1024 Ok(bytes) => {
1025 let text = std::str::from_utf8(&bytes).map_err(|_| {
1026 WorkspaceMultiRootPublisherError::Candidate(format!(
1027 "text conflict {path} is no longer UTF-8"
1028 ))
1029 })?;
1030 if crate::text_merge::has_conflict_markers(text) {
1031 return Err(WorkspaceMultiRootPublisherError::Candidate(format!(
1032 "text conflict {path} still contains conflict markers"
1033 )));
1034 }
1035 }
1036 Err(WorkspaceError::NotFound) => {}
1037 Err(error) => return Err(error.into()),
1038 }
1039 }
1040 declared.insert(view.key.clone());
1041 }
1042 Ok(declared)
1043 }
1044
1045 async fn finish_conflict(
1046 &self,
1047 operation_id: OperationId,
1048 root_id: WorkspaceRootId,
1049 root: &MultiRootMergeRoot,
1050 projected_generation: GenerationId,
1051 ) -> Result<MultiRootConflictFinish, Self::Error> {
1052 let target = self
1053 .resolver
1054 .resolve(root.target_workspace_id)
1055 .await
1056 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
1057 let current = target.head().await?;
1058 if !generation_descends_from(&target, current.id(), projected_generation).await? {
1059 return Err(WorkspaceMultiRootPublisherError::Candidate(
1060 "resolved conflict generation does not descend from its projection".to_owned(),
1061 ));
1062 }
1063 let reservation_operation = conflict_finish_operation_id(operation_id, root_id);
1064 let reservation = self
1065 .reserve_operation(
1066 &target,
1067 reservation_operation,
1068 "resolved conflict generation changed while being fenced",
1069 )
1070 .await?;
1071 Ok(MultiRootConflictFinish {
1072 generation: current.id(),
1073 fence: MultiRootFence {
1074 token: serde_json::to_vec(&WorkspacePublicationFence {
1075 target_workspace_id: target.id(),
1076 reservation,
1077 })
1078 .map_err(|error| WorkspaceMultiRootPublisherError::Candidate(error.to_string()))?,
1079 },
1080 })
1081 }
1082
1083 async fn publish(
1084 &self,
1085 operation_id: OperationId,
1086 root_id: WorkspaceRootId,
1087 root: &MultiRootMergeRoot,
1088 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
1089 fence: &MultiRootFence,
1090 ) -> Result<MultiRootPublishRoot, Self::Error> {
1091 let decoded = decode_reservation(operation_id, fence)?;
1092 if decoded.target_workspace_id != root.target_workspace_id {
1093 return Err(WorkspaceMultiRootPublisherError::InvalidFence);
1094 }
1095 let target = self
1096 .resolver
1097 .resolve(root.target_workspace_id)
1098 .await
1099 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
1100 if target.head().await?.id() != root.target_generation {
1101 if let Some(generation) = target
1102 .operation_generation(publisher_key(operation_id, root_id, b"publish"))
1103 .await?
1104 {
1105 target
1106 .authority()
1107 .release_publication(
1108 decoded.reservation,
1109 WorkBudget::UNBOUNDED,
1110 &CancellationToken::new(),
1111 )
1112 .await
1113 .map_err(|failure| WorkspaceError::engine(failure.error))?;
1114 return Ok(MultiRootPublishRoot::AlreadyPublished(generation.id()));
1115 }
1116 return Ok(MultiRootPublishRoot::UnexpectedTarget);
1117 }
1118 if decoded.reservation.authority_id != target.authority_id() {
1119 return Err(WorkspaceMultiRootPublisherError::InvalidFence);
1120 }
1121 let plan = self.pinned_plan(root, Some(target.clone())).await?;
1122 let outcome = match self
1123 .apply_plan(
1124 root_id,
1125 root,
1126 &plan,
1127 publisher_key(operation_id, root_id, b"publish"),
1128 resolutions,
1129 PublicationPermit::Reservation {
1130 operation_id: operation_id.into_bytes(),
1131 gate_tail: decoded.reservation.gate_tail,
1132 expected: decoded.reservation.expected,
1133 },
1134 )
1135 .await?
1136 {
1137 JoinOutcome::Applied(generation) | JoinOutcome::NoChanges(generation) => {
1138 MultiRootPublishRoot::Published(generation.id())
1139 }
1140 JoinOutcome::AlreadyApplied(generation) => {
1141 MultiRootPublishRoot::AlreadyPublished(generation.id())
1142 }
1143 JoinOutcome::StaleTarget(_)
1144 | JoinOutcome::Fenced
1145 | JoinOutcome::IdempotencyConflict
1146 | JoinOutcome::Conflicted { .. } => MultiRootPublishRoot::UnexpectedTarget,
1147 };
1148 if matches!(
1149 outcome,
1150 MultiRootPublishRoot::Published(_) | MultiRootPublishRoot::AlreadyPublished(_)
1151 ) {
1152 target
1153 .authority()
1154 .release_publication(
1155 decoded.reservation,
1156 WorkBudget::UNBOUNDED,
1157 &CancellationToken::new(),
1158 )
1159 .await
1160 .map_err(|failure| WorkspaceError::engine(failure.error))?;
1161 }
1162 Ok(outcome)
1163 }
1164}
1165
1166impl<A, O, R, M> MultiRootPublisher for MaterializingWorkspaceMultiRootPublisher<A, O, R, M>
1167where
1168 A: AsyncAuthorityStore,
1169 O: AsyncObjectStore,
1170 R: WorkspaceResolver<A, O>,
1171 M: MultiRootMaterializer<A, O>,
1172{
1173 type Error = MaterializingWorkspaceMultiRootPublisherError<R::Error, M::Error>;
1174
1175 async fn prepare(
1176 &self,
1177 operation_id: OperationId,
1178 root_id: WorkspaceRootId,
1179 root: &MultiRootMergeRoot,
1180 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
1181 ) -> Result<MultiRootPrepare, Self::Error> {
1182 self.publisher
1183 .prepare(operation_id, root_id, root, resolutions)
1184 .await
1185 .map_err(MaterializingWorkspaceMultiRootPublisherError::Workspace)
1186 }
1187
1188 async fn release(
1189 &self,
1190 operation_id: OperationId,
1191 root_id: WorkspaceRootId,
1192 fence: &MultiRootFence,
1193 ) -> Result<(), Self::Error> {
1194 self.publisher
1195 .release(operation_id, root_id, fence)
1196 .await
1197 .map_err(Into::into)
1198 }
1199
1200 async fn project_conflict(
1201 &self,
1202 operation_id: OperationId,
1203 root_id: WorkspaceRootId,
1204 root: &MultiRootMergeRoot,
1205 conflict: Option<&crate::MergePlan>,
1206 ) -> Result<GenerationId, Self::Error> {
1207 let to = self
1208 .publisher
1209 .project_conflict(operation_id, root_id, root, conflict)
1210 .await?;
1211 let workspace = self
1212 .publisher
1213 .resolver()
1214 .resolve(root.target_workspace_id)
1215 .await
1216 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
1217 self.materializer
1218 .materialize(
1219 materialization_operation_id(operation_id, root_id),
1220 root_id,
1221 &workspace,
1222 root.target_generation,
1223 to,
1224 )
1225 .await
1226 .map_err(MaterializingWorkspaceMultiRootPublisherError::Materializer)?;
1227 Ok(to)
1228 }
1229
1230 async fn abort_conflict(
1231 &self,
1232 operation_id: OperationId,
1233 root_id: WorkspaceRootId,
1234 root: &MultiRootMergeRoot,
1235 projected_generation: GenerationId,
1236 ) -> Result<GenerationId, Self::Error> {
1237 let workspace = self
1238 .publisher
1239 .resolver()
1240 .resolve(root.target_workspace_id)
1241 .await
1242 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
1243 let from = workspace
1244 .head()
1245 .await
1246 .map_err(WorkspaceMultiRootPublisherError::Workspace)?
1247 .id();
1248 let to = self
1249 .publisher
1250 .abort_conflict(operation_id, root_id, root, projected_generation)
1251 .await?;
1252 self.materializer
1253 .materialize(
1254 conflict_abort_materialization_operation_id(operation_id, root_id),
1255 root_id,
1256 &workspace,
1257 from,
1258 to,
1259 )
1260 .await
1261 .map_err(MaterializingWorkspaceMultiRootPublisherError::Materializer)?;
1262 self.materializer
1263 .finalize(
1264 conflict_abort_materialization_operation_id(operation_id, root_id),
1265 root_id,
1266 )
1267 .await
1268 .map_err(MaterializingWorkspaceMultiRootPublisherError::Materializer)?;
1269 Ok(to)
1270 }
1271
1272 async fn declare_conflicts(
1273 &self,
1274 operation_id: OperationId,
1275 root_id: WorkspaceRootId,
1276 root: &MultiRootMergeRoot,
1277 conflict: &crate::MergePlan,
1278 paths: &[String],
1279 ) -> Result<BTreeSet<ConflictKey>, Self::Error> {
1280 self.publisher
1281 .declare_conflicts(operation_id, root_id, root, conflict, paths)
1282 .await
1283 .map_err(Into::into)
1284 }
1285
1286 async fn finish_conflict(
1287 &self,
1288 operation_id: OperationId,
1289 root_id: WorkspaceRootId,
1290 root: &MultiRootMergeRoot,
1291 projected_generation: GenerationId,
1292 ) -> Result<MultiRootConflictFinish, Self::Error> {
1293 self.publisher
1294 .finish_conflict(operation_id, root_id, root, projected_generation)
1295 .await
1296 .map_err(Into::into)
1297 }
1298
1299 async fn publish(
1300 &self,
1301 operation_id: OperationId,
1302 root_id: WorkspaceRootId,
1303 root: &MultiRootMergeRoot,
1304 resolutions: &BTreeMap<ConflictKey, MergeResolution>,
1305 fence: &MultiRootFence,
1306 ) -> Result<MultiRootPublishRoot, Self::Error> {
1307 let published = self
1308 .publisher
1309 .publish(operation_id, root_id, root, resolutions, fence)
1310 .await?;
1311 if !matches!(
1312 published,
1313 MultiRootPublishRoot::Published(_) | MultiRootPublishRoot::AlreadyPublished(_)
1314 ) {
1315 return Ok(published);
1316 }
1317 let workspace = self
1318 .publisher
1319 .resolver()
1320 .resolve(root.target_workspace_id)
1321 .await
1322 .map_err(WorkspaceMultiRootPublisherError::Resolver)?;
1323 let key = publisher_key(operation_id, root_id, b"publish");
1324 let to = if let Some(generation) = workspace
1325 .operation_generation(key)
1326 .await
1327 .map_err(WorkspaceMultiRootPublisherError::Workspace)?
1328 {
1329 generation.id()
1330 } else {
1331 let head = workspace
1332 .head()
1333 .await
1334 .map_err(WorkspaceMultiRootPublisherError::Workspace)?;
1335 if head.id() != root.target_generation {
1336 return Err(
1337 MaterializingWorkspaceMultiRootPublisherError::MissingPublishedGeneration,
1338 );
1339 }
1340 head.id()
1341 };
1342 self.materializer
1343 .materialize(
1344 materialization_operation_id(operation_id, root_id),
1345 root_id,
1346 &workspace,
1347 root.target_generation,
1348 to,
1349 )
1350 .await
1351 .map_err(MaterializingWorkspaceMultiRootPublisherError::Materializer)?;
1352 Ok(published)
1353 }
1354
1355 async fn finalize(
1356 &self,
1357 operation_id: OperationId,
1358 root_id: WorkspaceRootId,
1359 _root: &MultiRootMergeRoot,
1360 ) -> Result<(), Self::Error> {
1361 self.materializer
1362 .finalize(materialization_operation_id(operation_id, root_id), root_id)
1363 .await
1364 .map_err(MaterializingWorkspaceMultiRootPublisherError::Materializer)
1365 }
1366}
1367
1368fn child_wins_resolutions(
1369 plan: &crate::MergePlan,
1370 paths: &BTreeSet<String>,
1371) -> BTreeMap<ConflictKey, MergeResolution> {
1372 plan.conflicts
1373 .iter()
1374 .filter(|conflict| {
1375 conflict.kind == crate::ConflictKind::Binding
1376 && conflict
1377 .path
1378 .as_ref()
1379 .is_some_and(|path| paths.contains(path))
1380 })
1381 .map(|conflict| {
1382 (
1383 conflict.key.clone(),
1384 MergeResolution::Select(ConflictSide::Theirs),
1385 )
1386 })
1387 .collect()
1388}
1389
1390fn publisher_key(
1391 operation_id: OperationId,
1392 root_id: WorkspaceRootId,
1393 purpose: &[u8],
1394) -> IdempotencyKey {
1395 let mut hasher = blake3::Hasher::new();
1396 hasher.update(b"acyclic.workspace-multi-root-publisher.v1\0");
1397 hasher.update(&operation_id.into_bytes());
1398 hasher.update(&root_id.into_bytes());
1399 hasher.update(purpose);
1400 let mut bytes = [0; 16];
1401 bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
1402 IdempotencyKey::from_bytes(bytes)
1403}
1404
1405fn materialization_operation_id(
1406 operation_id: OperationId,
1407 root_id: WorkspaceRootId,
1408) -> OperationId {
1409 let mut hasher = blake3::Hasher::new();
1410 hasher.update(b"acyclic.multi-root-materialization.v1\0");
1411 hasher.update(&operation_id.into_bytes());
1412 hasher.update(&root_id.into_bytes());
1413 let mut bytes = [0; 16];
1414 bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
1415 OperationId::from_bytes(bytes)
1416}
1417
1418fn conflict_projection_operation_id(
1419 operation_id: OperationId,
1420 root_id: WorkspaceRootId,
1421) -> OperationId {
1422 let mut hasher = blake3::Hasher::new();
1423 hasher.update(b"acyclic.multi-root-conflict-projection.v1\0");
1424 hasher.update(&operation_id.into_bytes());
1425 hasher.update(&root_id.into_bytes());
1426 let mut bytes = [0; 16];
1427 bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
1428 OperationId::from_bytes(bytes)
1429}
1430
1431fn conflict_abort_materialization_operation_id(
1432 operation_id: OperationId,
1433 root_id: WorkspaceRootId,
1434) -> OperationId {
1435 let mut hasher = blake3::Hasher::new();
1436 hasher.update(b"acyclic.multi-root-conflict-abort-materialization.v1\0");
1437 hasher.update(&operation_id.into_bytes());
1438 hasher.update(&root_id.into_bytes());
1439 let mut bytes = [0; 16];
1440 bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
1441 OperationId::from_bytes(bytes)
1442}
1443
1444fn portable_namespace_path(path: &crate::kernel::NamespacePath) -> Option<String> {
1445 let mut value = String::new();
1446 for component in path.components() {
1447 if component.encoding() != crate::kernel::NameEncoding::Utf8 {
1448 return None;
1449 }
1450 value.push('/');
1451 value.push_str(std::str::from_utf8(component.as_bytes()).ok()?);
1452 }
1453 if value.is_empty() {
1454 value.push('/');
1455 }
1456 Some(value)
1457}
1458
1459fn conflict_finish_operation_id(
1460 operation_id: OperationId,
1461 root_id: WorkspaceRootId,
1462) -> OperationId {
1463 let mut hasher = blake3::Hasher::new();
1464 hasher.update(b"acyclic.multi-root-conflict-finish.v1\0");
1465 hasher.update(&operation_id.into_bytes());
1466 hasher.update(&root_id.into_bytes());
1467 let mut bytes = [0; 16];
1468 bytes.copy_from_slice(&hasher.finalize().as_bytes()[..16]);
1469 OperationId::from_bytes(bytes)
1470}
1471
1472async fn generation_descends_from<A: AsyncAuthorityStore, O: AsyncObjectStore>(
1473 workspace: &Workspace<A, O>,
1474 descendant: GenerationId,
1475 ancestor: GenerationId,
1476) -> Result<bool, WorkspaceError> {
1477 if descendant == ancestor {
1478 return Ok(true);
1479 }
1480 let mut pending = vec![descendant];
1481 let mut visited = BTreeSet::new();
1482 while let Some(candidate) = pending.pop() {
1483 if !visited.insert(candidate) || visited.len() > 65_536 {
1484 continue;
1485 }
1486 for parent in workspace.generation(candidate).await?.parents().await? {
1487 if parent == ancestor {
1488 return Ok(true);
1489 }
1490 pending.push(parent);
1491 }
1492 }
1493 Ok(false)
1494}
1495
1496fn decode_reservation<E: std::error::Error + 'static>(
1497 operation_id: OperationId,
1498 fence: &MultiRootFence,
1499) -> Result<WorkspacePublicationFence, WorkspaceMultiRootPublisherError<E>> {
1500 let decoded: WorkspacePublicationFence = serde_json::from_slice(&fence.token)
1501 .map_err(|_| WorkspaceMultiRootPublisherError::InvalidFence)?;
1502 if decoded.reservation.operation_id != operation_id {
1503 return Err(WorkspaceMultiRootPublisherError::InvalidFence);
1504 }
1505 Ok(decoded)
1506}
1507
1508fn preview_workspace_name(
1509 operation_id: OperationId,
1510 root_id: WorkspaceRootId,
1511 preview_key: IdempotencyKey,
1512) -> Result<WorkspaceName, WorkspaceError> {
1513 WorkspaceName::new(format!(
1514 "multi-root-preview-{}-{}-{}",
1515 hex::encode(&operation_id.into_bytes()[..6]),
1516 hex::encode(&root_id.into_bytes()[..6]),
1517 hex::encode(preview_key.operation_id().into_bytes())
1518 ))
1519 .map_err(Into::into)
1520}
1521
1522pub trait MultiRootPublicationAuthorizer: Send + Sync {
1524 type Error: std::error::Error + Send + Sync + 'static;
1526
1527 fn authorize(
1530 &self,
1531 plan: &MultiRootMergePlan,
1532 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
1533
1534 fn authorize_transition(
1537 &self,
1538 plan: &MultiRootMergePlan,
1539 caller_context_id: WorkspaceContextId,
1540 ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture {
1541 async move {
1542 if caller_context_id != plan.parent_context_id {
1543 return Ok(false);
1544 }
1545 self.authorize(plan).await
1546 }
1547 }
1548}
1549
1550impl<S: WorkspaceContextStore> MultiRootPublicationAuthorizer for WorkspaceContextRegistry<S> {
1551 type Error = WorkspaceContextError<S::Error>;
1552
1553 async fn authorize(&self, plan: &MultiRootMergePlan) -> Result<bool, Self::Error> {
1554 let child = self
1555 .authorize_parent(plan.child_context_id, plan.parent_context_id)
1556 .await?;
1557 let parent = self.resolve(plan.parent_context_id).await?;
1558 Ok(plan.roots.iter().all(|(root_id, root)| {
1559 root.merge_workspace_id.is_none()
1560 && child.roots.get(root_id).is_some_and(|binding| {
1561 binding.workspace_id == root.source_workspace_id
1562 && binding.parent_workspace_id == Some(root.target_workspace_id)
1563 })
1564 && parent
1565 .roots
1566 .get(root_id)
1567 .is_some_and(|binding| binding.workspace_id == root.target_workspace_id)
1568 }))
1569 }
1570}
1571
1572pub struct LineageMultiRootPublicationAuthorizer<S> {
1574 store: S,
1575}
1576
1577impl<S> LineageMultiRootPublicationAuthorizer<S> {
1578 #[must_use]
1580 pub const fn new(store: S) -> Self {
1581 Self { store }
1582 }
1583}
1584
1585#[derive(Debug, Error)]
1587pub enum LineageMultiRootPublicationAuthorizationError<E: std::error::Error + 'static> {
1588 #[error(transparent)]
1590 Context(#[from] WorkspaceContextError<E>),
1591 #[error(transparent)]
1593 Lineage(#[from] WorkspaceLineageError<E>),
1594}
1595
1596impl<S> MultiRootPublicationAuthorizer for LineageMultiRootPublicationAuthorizer<S>
1597where
1598 S: WorkspaceContextStore
1599 + WorkspaceLineageStore<Error = <S as WorkspaceContextStore>::Error>
1600 + Clone,
1601{
1602 type Error = LineageMultiRootPublicationAuthorizationError<<S as WorkspaceContextStore>::Error>;
1603
1604 async fn authorize(&self, plan: &MultiRootMergePlan) -> Result<bool, Self::Error> {
1605 let contexts = WorkspaceContextRegistry::new(self.store.clone());
1606 let child = contexts
1607 .authorize_parent(plan.child_context_id, plan.parent_context_id)
1608 .await?;
1609 let parent = contexts.resolve(plan.parent_context_id).await?;
1610 let graph = WorkspaceGraph::new(self.store.clone());
1611 for (root_id, root) in &plan.roots {
1612 if !child.roots.get(root_id).is_some_and(|binding| {
1613 binding.workspace_id == root.source_workspace_id
1614 && binding.parent_workspace_id == Some(root.target_workspace_id)
1615 }) || parent
1616 .roots
1617 .get(root_id)
1618 .is_none_or(|binding| binding.workspace_id != root.target_workspace_id)
1619 {
1620 return Ok(false);
1621 }
1622 if graph
1623 .authorize_join(root.source_workspace_id, root.target_workspace_id)
1624 .await
1625 .is_err()
1626 {
1627 return Ok(false);
1628 }
1629 if let Some(filtered) = root.merge_workspace_id
1630 && graph
1631 .authorize_join(filtered, root.source_workspace_id)
1632 .await
1633 .is_err()
1634 {
1635 return Ok(false);
1636 }
1637 }
1638 Ok(true)
1639 }
1640}
1641
1642#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1644pub enum MultiRootPublishRoot {
1645 Published(GenerationId),
1647 AlreadyPublished(GenerationId),
1649 UnexpectedTarget,
1651}
1652
1653#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1655pub enum Publication {
1656 Applied(MultiRootPublication),
1658 StaleBeforeCommit(WorkspaceRootId),
1660 Conflicted(MultiRootPublication),
1662 Paused(MultiRootPublication),
1664}
1665
1666enum MultiRootValidation {
1667 Committed(MultiRootPublication),
1668 Stale(WorkspaceRootId, MultiRootPublication),
1669 Conflicted(MultiRootPublication),
1670}
1671
1672#[derive(Debug, Error)]
1674pub enum MultiRootPublicationError<
1675 S: std::error::Error + 'static,
1676 P: std::error::Error + 'static,
1677 A: std::error::Error + 'static,
1678> {
1679 #[error("multi-root publication store failed: {0}")]
1681 Store(S),
1682 #[error("multi-root publisher failed: {0}")]
1684 Publisher(P),
1685 #[error("multi-root publication authorization failed: {0}")]
1687 Authorizer(A),
1688 #[error("multi-root publication candidate is invalid or conflicts with its retry identity")]
1690 InvalidCandidate,
1691 #[error("multi-root publication state remained contended")]
1693 Contended,
1694 #[error("parent context already has an unresolved multi-root publication")]
1696 ParentBusy,
1697}
1698
1699type MultiRootResult<T, S, P, A> = Result<T, MultiRootPublicationError<S, P, A>>;
1700
1701pub struct MultiRootPublicationCoordinator<S, P, A> {
1703 store: S,
1704 publisher: P,
1705 authorizer: A,
1706}
1707
1708impl<S, P, A> MultiRootPublicationCoordinator<S, P, A> {
1709 #[must_use]
1711 pub const fn new(store: S, publisher: P, authorizer: A) -> Self {
1712 Self {
1713 store,
1714 publisher,
1715 authorizer,
1716 }
1717 }
1718}
1719
1720impl<S: MultiRootPublicationStore, P: MultiRootPublisher, A: MultiRootPublicationAuthorizer>
1721 MultiRootPublicationCoordinator<S, P, A>
1722{
1723 pub async fn resume(
1726 &self,
1727 operation_id: OperationId,
1728 ) -> Result<Option<Publication>, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
1729 self.resume_with_finalization(operation_id, true).await
1730 }
1731
1732 pub async fn resume_retained(
1735 &self,
1736 operation_id: OperationId,
1737 ) -> Result<Option<Publication>, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
1738 self.resume_with_finalization(operation_id, false).await
1739 }
1740
1741 async fn resume_with_finalization(
1742 &self,
1743 operation_id: OperationId,
1744 finalize: bool,
1745 ) -> Result<Option<Publication>, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
1746 let Some(journal) = self
1747 .store
1748 .load(operation_id)
1749 .await
1750 .map_err(MultiRootPublicationError::Store)?
1751 else {
1752 return Ok(None);
1753 };
1754 validate_journal(&journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
1755 if journal.phase == MultiRootPublicationPhase::Applied {
1756 if finalize {
1757 self.finalize(&journal).await?;
1758 }
1759 return Ok(Some(Publication::Applied(journal)));
1760 }
1761 if journal.phase == MultiRootPublicationPhase::Conflicted {
1762 return self
1763 .project_conflicted(journal)
1764 .await
1765 .map(|journal| Some(Publication::Conflicted(journal)));
1766 }
1767 if journal.phase == MultiRootPublicationPhase::Aborting {
1768 self.abort_journal(journal).await?;
1769 return Ok(None);
1770 }
1771 self.publish_with_finalization(journal.candidate, finalize)
1772 .await
1773 .map(Some)
1774 }
1775
1776 pub async fn pending_operations(
1778 &self,
1779 ) -> Result<Vec<OperationId>, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
1780 self.store
1781 .list_operations()
1782 .await
1783 .map_err(MultiRootPublicationError::Store)
1784 }
1785
1786 pub fn publish(
1788 &self,
1789 candidate: MultiRootMergeCandidate,
1790 ) -> impl Future<Output = MultiRootResult<Publication, S::Error, P::Error, A::Error>>
1791 + StorageFuture
1792 + '_ {
1793 Box::pin(async move { self.publish_with_finalization(candidate, true).await })
1794 }
1795
1796 pub fn publish_retained(
1799 &self,
1800 candidate: MultiRootMergeCandidate,
1801 ) -> impl Future<Output = MultiRootResult<Publication, S::Error, P::Error, A::Error>>
1802 + StorageFuture
1803 + '_ {
1804 Box::pin(async move { self.publish_with_finalization(candidate, false).await })
1805 }
1806
1807 async fn publish_with_finalization(
1808 &self,
1809 candidate: MultiRootMergeCandidate,
1810 finalize: bool,
1811 ) -> MultiRootResult<Publication, S::Error, P::Error, A::Error> {
1812 validate_candidate(&candidate).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
1813 if !self
1814 .authorizer
1815 .authorize(&candidate.plan)
1816 .await
1817 .map_err(MultiRootPublicationError::Authorizer)?
1818 {
1819 return Err(MultiRootPublicationError::InvalidCandidate);
1820 }
1821 let mut journal = self.load_or_create(candidate).await?;
1822 validate_journal(&journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
1823 if journal.phase == MultiRootPublicationPhase::Applied {
1824 if finalize {
1825 self.finalize(&journal).await?;
1826 }
1827 return Ok(Publication::Applied(journal));
1828 }
1829 if journal.phase == MultiRootPublicationPhase::Conflicted {
1830 return self
1831 .project_conflicted(journal)
1832 .await
1833 .map(Publication::Conflicted);
1834 }
1835 if journal.phase == MultiRootPublicationPhase::Validating {
1836 journal = match self.validate_and_commit(journal).await? {
1837 MultiRootValidation::Committed(journal) => journal,
1838 MultiRootValidation::Stale(root_id, journal) => {
1839 self.discard_uncommitted(&journal).await?;
1840 return Ok(Publication::StaleBeforeCommit(root_id));
1841 }
1842 MultiRootValidation::Conflicted(journal) => {
1843 return self
1844 .project_conflicted(journal)
1845 .await
1846 .map(Publication::Conflicted);
1847 }
1848 };
1849 }
1850 self.roll_forward(journal, finalize).await
1851 }
1852
1853 pub fn continue_conflicted(
1855 &self,
1856 operation_id: OperationId,
1857 caller_context_id: WorkspaceContextId,
1858 ) -> impl Future<Output = MultiRootResult<Publication, S::Error, P::Error, A::Error>>
1859 + StorageFuture
1860 + '_ {
1861 Box::pin(async move {
1862 self.continue_conflicted_with_finalization(operation_id, caller_context_id, true)
1863 .await
1864 })
1865 }
1866
1867 pub fn continue_conflicted_retained(
1870 &self,
1871 operation_id: OperationId,
1872 caller_context_id: WorkspaceContextId,
1873 ) -> impl Future<Output = MultiRootResult<Publication, S::Error, P::Error, A::Error>>
1874 + StorageFuture
1875 + '_ {
1876 Box::pin(async move {
1877 self.continue_conflicted_with_finalization(operation_id, caller_context_id, false)
1878 .await
1879 })
1880 }
1881
1882 pub async fn declare_conflicted_paths(
1886 &self,
1887 operation_id: OperationId,
1888 caller_context_id: WorkspaceContextId,
1889 root_id: WorkspaceRootId,
1890 paths: &[String],
1891 ) -> MultiRootResult<(), S::Error, P::Error, A::Error> {
1892 let mut journal = self
1893 .store
1894 .load(operation_id)
1895 .await
1896 .map_err(MultiRootPublicationError::Store)?
1897 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
1898 validate_journal(&journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
1899 if journal.phase != MultiRootPublicationPhase::Conflicted
1900 || !self
1901 .authorizer
1902 .authorize_transition(&journal.candidate.plan, caller_context_id)
1903 .await
1904 .map_err(MultiRootPublicationError::Authorizer)?
1905 {
1906 return Err(MultiRootPublicationError::InvalidCandidate);
1907 }
1908 let root = journal
1909 .candidate
1910 .plan
1911 .roots
1912 .get(&root_id)
1913 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
1914 let conflict = journal
1915 .conflicts
1916 .get(&root_id)
1917 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
1918 let declared = self
1919 .publisher
1920 .declare_conflicts(operation_id, root_id, root, conflict, paths)
1921 .await
1922 .map_err(MultiRootPublicationError::Publisher)?;
1923 if declared.is_empty() {
1924 return Err(MultiRootPublicationError::InvalidCandidate);
1925 }
1926 journal
1927 .declared_conflicts
1928 .entry(root_id)
1929 .or_default()
1930 .extend(declared);
1931 self.persist(journal).await?;
1932 Ok(())
1933 }
1934
1935 async fn continue_conflicted_with_finalization(
1936 &self,
1937 operation_id: OperationId,
1938 caller_context_id: WorkspaceContextId,
1939 finalize: bool,
1940 ) -> MultiRootResult<Publication, S::Error, P::Error, A::Error> {
1941 let journal = self
1942 .store
1943 .load(operation_id)
1944 .await
1945 .map_err(MultiRootPublicationError::Store)?
1946 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
1947 validate_journal(&journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
1948 if !self
1949 .authorizer
1950 .authorize_transition(&journal.candidate.plan, caller_context_id)
1951 .await
1952 .map_err(MultiRootPublicationError::Authorizer)?
1953 {
1954 return Err(MultiRootPublicationError::InvalidCandidate);
1955 }
1956 if journal.phase != MultiRootPublicationPhase::Conflicted
1957 || journal
1958 .projected_roots
1959 .keys()
1960 .ne(journal.candidate.plan.roots.keys())
1961 {
1962 return Err(MultiRootPublicationError::InvalidCandidate);
1963 }
1964 for (root_id, conflict) in &journal.conflicts {
1965 let expected = conflict
1966 .conflicts
1967 .iter()
1968 .map(|view| view.key.clone())
1969 .collect::<BTreeSet<_>>();
1970 if !journal
1971 .declared_conflicts
1972 .get(root_id)
1973 .is_some_and(|declared| expected.is_subset(declared))
1974 {
1975 return Err(MultiRootPublicationError::InvalidCandidate);
1976 }
1977 }
1978 let roots = journal.candidate.plan.roots.clone();
1979 let mut applied = journal;
1980 for (root_id, root) in &roots {
1981 if applied.fences.contains_key(root_id) {
1982 continue;
1983 }
1984 let projected = *applied
1985 .projected_roots
1986 .get(root_id)
1987 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
1988 let finished = self
1989 .publisher
1990 .finish_conflict(operation_id, *root_id, root, projected)
1991 .await
1992 .map_err(MultiRootPublicationError::Publisher)?;
1993 applied
1994 .projected_roots
1995 .insert(*root_id, finished.generation);
1996 applied
1997 .published_generations
1998 .insert(*root_id, finished.generation);
1999 applied.fences.insert(*root_id, finished.fence);
2000 applied = self.persist(applied).await?;
2001 }
2002 applied.phase = MultiRootPublicationPhase::Applied;
2003 applied.published_roots = applied.candidate.plan.roots.keys().copied().collect();
2004 applied.conflicts.clear();
2005 applied.projected_roots.clear();
2006 applied = self.persist(applied).await?;
2007 if finalize {
2008 self.finalize(&applied).await?;
2009 }
2010 Ok(Publication::Applied(applied))
2011 }
2012
2013 pub async fn acknowledge_applied(
2016 &self,
2017 journal: &MultiRootPublication,
2018 ) -> MultiRootResult<(), S::Error, P::Error, A::Error> {
2019 validate_journal(journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
2020 if journal.phase != MultiRootPublicationPhase::Applied {
2021 return Err(MultiRootPublicationError::InvalidCandidate);
2022 }
2023 self.finalize(journal).await
2024 }
2025
2026 pub fn abort_conflicted(
2028 &self,
2029 operation_id: OperationId,
2030 caller_context_id: WorkspaceContextId,
2031 ) -> impl Future<Output = MultiRootResult<(), S::Error, P::Error, A::Error>> + StorageFuture + '_
2032 {
2033 Box::pin(async move {
2034 let mut journal = self
2035 .store
2036 .load(operation_id)
2037 .await
2038 .map_err(MultiRootPublicationError::Store)?
2039 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
2040 validate_journal(&journal).map_err(|()| MultiRootPublicationError::InvalidCandidate)?;
2041 if !self
2042 .authorizer
2043 .authorize_transition(&journal.candidate.plan, caller_context_id)
2044 .await
2045 .map_err(MultiRootPublicationError::Authorizer)?
2046 {
2047 return Err(MultiRootPublicationError::InvalidCandidate);
2048 }
2049 if journal.phase == MultiRootPublicationPhase::Conflicted {
2050 for (root_id, fence) in &journal.fences {
2051 self.publisher
2052 .release(
2053 conflict_finish_operation_id(operation_id, *root_id),
2054 *root_id,
2055 fence,
2056 )
2057 .await
2058 .map_err(MultiRootPublicationError::Publisher)?;
2059 }
2060 journal.fences.clear();
2061 journal.declared_conflicts.clear();
2062 journal.phase = MultiRootPublicationPhase::Aborting;
2063 journal.published_roots.clear();
2064 journal = self.persist(journal).await?;
2065 }
2066 if journal.phase != MultiRootPublicationPhase::Aborting {
2067 return Err(MultiRootPublicationError::InvalidCandidate);
2068 }
2069 self.abort_journal(journal).await
2070 })
2071 }
2072
2073 async fn project_conflicted(
2074 &self,
2075 mut journal: MultiRootPublication,
2076 ) -> Result<MultiRootPublication, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
2077 let operation_id = journal.candidate.plan.operation_id;
2078 let roots = journal.candidate.plan.roots.clone();
2079 for (root_id, root) in roots {
2080 if journal.projected_roots.contains_key(&root_id) {
2081 continue;
2082 }
2083 let generation = self
2084 .publisher
2085 .project_conflict(
2086 operation_id,
2087 root_id,
2088 &root,
2089 journal.conflicts.get(&root_id),
2090 )
2091 .await
2092 .map_err(MultiRootPublicationError::Publisher)?;
2093 journal.projected_roots.insert(root_id, generation);
2094 journal = self.persist(journal).await?;
2095 }
2096 Ok(journal)
2097 }
2098
2099 async fn abort_journal(
2100 &self,
2101 mut journal: MultiRootPublication,
2102 ) -> MultiRootResult<(), S::Error, P::Error, A::Error> {
2103 let operation_id = journal.candidate.plan.operation_id;
2104 let roots = journal.candidate.plan.roots.clone();
2105 for (root_id, root) in roots {
2106 if journal.published_roots.contains(&root_id) {
2107 continue;
2108 }
2109 if let Some(projected_generation) = journal.projected_roots.get(&root_id).copied() {
2110 self.publisher
2111 .abort_conflict(operation_id, root_id, &root, projected_generation)
2112 .await
2113 .map_err(MultiRootPublicationError::Publisher)?;
2114 }
2115 journal.published_roots.insert(root_id);
2116 journal = self.persist(journal).await?;
2117 }
2118 if !self
2119 .store
2120 .release_parent(journal.candidate.plan.parent_context_id, operation_id)
2121 .await
2122 .map_err(MultiRootPublicationError::Store)?
2123 {
2124 return Err(MultiRootPublicationError::Contended);
2125 }
2126 if !self
2127 .store
2128 .compare_and_delete(operation_id, journal.revision)
2129 .await
2130 .map_err(MultiRootPublicationError::Store)?
2131 {
2132 return Err(MultiRootPublicationError::Contended);
2133 }
2134 Ok(())
2135 }
2136
2137 async fn discard_uncommitted(
2138 &self,
2139 journal: &MultiRootPublication,
2140 ) -> MultiRootResult<(), S::Error, P::Error, A::Error> {
2141 let operation_id = journal.candidate.plan.operation_id;
2142 if !self
2143 .store
2144 .release_parent(journal.candidate.plan.parent_context_id, operation_id)
2145 .await
2146 .map_err(MultiRootPublicationError::Store)?
2147 || !self
2148 .store
2149 .compare_and_delete(operation_id, journal.revision)
2150 .await
2151 .map_err(MultiRootPublicationError::Store)?
2152 {
2153 return Err(MultiRootPublicationError::Contended);
2154 }
2155 Ok(())
2156 }
2157
2158 async fn load_or_create(
2159 &self,
2160 candidate: MultiRootMergeCandidate,
2161 ) -> Result<MultiRootPublication, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
2162 let operation_id = candidate.plan.operation_id;
2163 if let Some(existing) = self
2164 .store
2165 .load(operation_id)
2166 .await
2167 .map_err(MultiRootPublicationError::Store)?
2168 {
2169 if existing.candidate != candidate {
2170 return Err(MultiRootPublicationError::InvalidCandidate);
2171 }
2172 if !self
2173 .store
2174 .claim_parent(candidate.plan.parent_context_id, operation_id)
2175 .await
2176 .map_err(MultiRootPublicationError::Store)?
2177 {
2178 if is_pristine_unclaimed(&existing)
2179 && !self
2180 .store
2181 .compare_and_delete(operation_id, existing.revision)
2182 .await
2183 .map_err(MultiRootPublicationError::Store)?
2184 {
2185 return Err(MultiRootPublicationError::Contended);
2186 }
2187 return Err(MultiRootPublicationError::ParentBusy);
2188 }
2189 return Ok(existing);
2190 }
2191 let journal = MultiRootPublication {
2192 version: MULTI_ROOT_VERSION,
2193 revision: 1,
2194 candidate,
2195 phase: MultiRootPublicationPhase::Validating,
2196 published_roots: BTreeSet::new(),
2197 published_generations: BTreeMap::new(),
2198 fences: BTreeMap::new(),
2199 conflicts: BTreeMap::new(),
2200 projected_roots: BTreeMap::new(),
2201 declared_conflicts: BTreeMap::new(),
2202 paused_root: None,
2203 };
2204 let created = match self
2205 .store
2206 .compare_and_swap(operation_id, 0, journal.clone())
2207 .await
2208 {
2209 Ok(created) => created,
2210 Err(error) => return Err(MultiRootPublicationError::Store(error)),
2211 };
2212 if created {
2213 let claimed = self
2214 .store
2215 .claim_parent(journal.candidate.plan.parent_context_id, operation_id)
2216 .await
2217 .map_err(MultiRootPublicationError::Store)?;
2218 if !claimed {
2219 if !self
2220 .store
2221 .compare_and_delete(operation_id, journal.revision)
2222 .await
2223 .map_err(MultiRootPublicationError::Store)?
2224 {
2225 return Err(MultiRootPublicationError::Contended);
2226 }
2227 return Err(MultiRootPublicationError::ParentBusy);
2228 }
2229 return Ok(journal);
2230 }
2231 let existing = self
2232 .store
2233 .load(operation_id)
2234 .await
2235 .map_err(MultiRootPublicationError::Store)?
2236 .ok_or(MultiRootPublicationError::Contended)?;
2237 if existing.candidate != journal.candidate {
2238 return Err(MultiRootPublicationError::InvalidCandidate);
2239 }
2240 if !self
2241 .store
2242 .claim_parent(existing.candidate.plan.parent_context_id, operation_id)
2243 .await
2244 .map_err(MultiRootPublicationError::Store)?
2245 {
2246 if is_pristine_unclaimed(&existing)
2247 && !self
2248 .store
2249 .compare_and_delete(operation_id, existing.revision)
2250 .await
2251 .map_err(MultiRootPublicationError::Store)?
2252 {
2253 return Err(MultiRootPublicationError::Contended);
2254 }
2255 return Err(MultiRootPublicationError::ParentBusy);
2256 }
2257 Ok(existing)
2258 }
2259
2260 async fn validate_and_commit(
2261 &self,
2262 mut journal: MultiRootPublication,
2263 ) -> Result<MultiRootValidation, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
2264 let operation_id = journal.candidate.plan.operation_id;
2265 let roots = journal.candidate.plan.roots.clone();
2266 for (root_id, root) in roots {
2267 if journal.fences.contains_key(&root_id) {
2268 continue;
2269 }
2270 let resolutions = journal
2271 .candidate
2272 .resolutions
2273 .get(&root_id)
2274 .cloned()
2275 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
2276 let prepared = self
2277 .publisher
2278 .prepare(operation_id, root_id, &root, &resolutions)
2279 .await;
2280 let fence = match prepared {
2281 Ok(MultiRootPrepare::Ready(fence)) => fence,
2282 Ok(MultiRootPrepare::Stale) => {
2283 for (prepared_root, prepared_fence) in &journal.fences {
2284 self.publisher
2285 .release(operation_id, *prepared_root, prepared_fence)
2286 .await
2287 .map_err(MultiRootPublicationError::Publisher)?;
2288 }
2289 journal.fences.clear();
2290 journal.conflicts.clear();
2291 journal = self.persist(journal).await?;
2292 return Ok(MultiRootValidation::Stale(root_id, journal));
2293 }
2294 Ok(MultiRootPrepare::Conflicted(plan)) => {
2295 journal.conflicts.insert(root_id, *plan);
2296 continue;
2297 }
2298 Err(error) => {
2299 for (prepared_root, prepared_fence) in &journal.fences {
2300 self.publisher
2301 .release(operation_id, *prepared_root, prepared_fence)
2302 .await
2303 .map_err(MultiRootPublicationError::Publisher)?;
2304 }
2305 journal.fences.clear();
2306 journal.conflicts.clear();
2307 let _ = self.persist(journal).await?;
2308 return Err(MultiRootPublicationError::Publisher(error));
2309 }
2310 };
2311 journal.fences.insert(root_id, fence);
2312 journal = self.persist(journal).await?;
2313 }
2314 if !journal.conflicts.is_empty() {
2315 for (prepared_root, prepared_fence) in &journal.fences {
2316 self.publisher
2317 .release(operation_id, *prepared_root, prepared_fence)
2318 .await
2319 .map_err(MultiRootPublicationError::Publisher)?;
2320 }
2321 journal.fences.clear();
2322 journal.phase = MultiRootPublicationPhase::Conflicted;
2323 return self
2324 .persist(journal)
2325 .await
2326 .map(MultiRootValidation::Conflicted);
2327 }
2328 journal.phase = MultiRootPublicationPhase::Committed;
2329 journal.paused_root = None;
2330 self.persist(journal)
2331 .await
2332 .map(MultiRootValidation::Committed)
2333 }
2334
2335 async fn roll_forward(
2336 &self,
2337 mut journal: MultiRootPublication,
2338 finalize: bool,
2339 ) -> MultiRootResult<Publication, S::Error, P::Error, A::Error> {
2340 let operation_id = journal.candidate.plan.operation_id;
2341 journal.phase = MultiRootPublicationPhase::Publishing;
2342 journal.paused_root = None;
2343 journal = self.persist(journal).await?;
2344 let roots = journal.candidate.plan.roots.clone();
2345 for (root_id, root) in roots {
2346 if journal.published_roots.contains(&root_id) {
2347 continue;
2348 }
2349 let resolutions = journal
2350 .candidate
2351 .resolutions
2352 .get(&root_id)
2353 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
2354 let fence = journal
2355 .fences
2356 .get(&root_id)
2357 .ok_or(MultiRootPublicationError::InvalidCandidate)?;
2358 match self
2359 .publisher
2360 .publish(operation_id, root_id, &root, resolutions, fence)
2361 .await
2362 .map_err(MultiRootPublicationError::Publisher)?
2363 {
2364 MultiRootPublishRoot::Published(generation)
2365 | MultiRootPublishRoot::AlreadyPublished(generation) => {
2366 journal.published_roots.insert(root_id);
2367 journal.published_generations.insert(root_id, generation);
2368 journal = self.persist(journal).await?;
2369 }
2370 MultiRootPublishRoot::UnexpectedTarget => {
2371 journal.phase = MultiRootPublicationPhase::Paused;
2372 journal.paused_root = Some(root_id);
2373 journal = self.persist(journal).await?;
2374 return Ok(Publication::Paused(journal));
2375 }
2376 }
2377 }
2378 journal.phase = MultiRootPublicationPhase::Applied;
2379 journal.paused_root = None;
2380 journal = self.persist(journal).await?;
2381 if finalize {
2382 self.finalize(&journal).await?;
2383 }
2384 Ok(Publication::Applied(journal))
2385 }
2386
2387 async fn finalize(
2388 &self,
2389 journal: &MultiRootPublication,
2390 ) -> MultiRootResult<(), S::Error, P::Error, A::Error> {
2391 let operation_id = journal.candidate.plan.operation_id;
2392 if !journal.declared_conflicts.is_empty() {
2393 for (root_id, fence) in &journal.fences {
2394 self.publisher
2395 .release(
2396 conflict_finish_operation_id(operation_id, *root_id),
2397 *root_id,
2398 fence,
2399 )
2400 .await
2401 .map_err(MultiRootPublicationError::Publisher)?;
2402 }
2403 }
2404 for (root_id, root) in &journal.candidate.plan.roots {
2405 self.publisher
2406 .finalize(operation_id, *root_id, root)
2407 .await
2408 .map_err(MultiRootPublicationError::Publisher)?;
2409 }
2410 if !self
2411 .store
2412 .release_parent(journal.candidate.plan.parent_context_id, operation_id)
2413 .await
2414 .map_err(MultiRootPublicationError::Store)?
2415 {
2416 return Err(MultiRootPublicationError::Contended);
2417 }
2418 if !self
2419 .store
2420 .compare_and_delete(operation_id, journal.revision)
2421 .await
2422 .map_err(MultiRootPublicationError::Store)?
2423 {
2424 return Err(MultiRootPublicationError::Contended);
2425 }
2426 Ok(())
2427 }
2428
2429 async fn persist(
2430 &self,
2431 mut journal: MultiRootPublication,
2432 ) -> Result<MultiRootPublication, MultiRootPublicationError<S::Error, P::Error, A::Error>> {
2433 let expected = journal.revision;
2434 journal.revision = expected.saturating_add(1);
2435 if self
2436 .store
2437 .compare_and_swap(
2438 journal.candidate.plan.operation_id,
2439 expected,
2440 journal.clone(),
2441 )
2442 .await
2443 .map_err(MultiRootPublicationError::Store)?
2444 {
2445 Ok(journal)
2446 } else {
2447 Err(MultiRootPublicationError::Contended)
2448 }
2449 }
2450}
2451
2452fn validate_candidate(candidate: &MultiRootMergeCandidate) -> Result<(), ()> {
2453 if candidate.plan.roots.is_empty()
2454 || candidate.plan.parent_context_id == candidate.plan.child_context_id
2455 || candidate.plan.roots.keys().ne(candidate.resolutions.keys())
2456 || candidate
2457 .plan
2458 .roots
2459 .values()
2460 .any(|root| root.source_workspace_id == root.target_workspace_id)
2461 {
2462 return Err(());
2463 }
2464 Ok(())
2465}
2466
2467fn is_pristine_unclaimed(journal: &MultiRootPublication) -> bool {
2468 journal.phase == MultiRootPublicationPhase::Validating
2469 && journal.fences.is_empty()
2470 && journal.conflicts.is_empty()
2471 && journal.projected_roots.is_empty()
2472 && journal.declared_conflicts.is_empty()
2473 && journal.published_roots.is_empty()
2474 && journal.published_generations.is_empty()
2475 && journal.paused_root.is_none()
2476}
2477
2478fn validate_journal(journal: &MultiRootPublication) -> Result<(), ()> {
2479 let roots: BTreeSet<_> = journal.candidate.plan.roots.keys().copied().collect();
2480 let fence_roots: BTreeSet<_> = journal.fences.keys().copied().collect();
2481 let generation_roots: BTreeSet<_> = journal.published_generations.keys().copied().collect();
2482 let no_publication_progress =
2483 journal.published_roots.is_empty() && journal.published_generations.is_empty();
2484 let no_conflict_state = journal.conflicts.is_empty()
2485 && journal.projected_roots.is_empty()
2486 && journal.declared_conflicts.is_empty();
2487 let phase_valid = match journal.phase {
2488 MultiRootPublicationPhase::Validating => {
2489 no_publication_progress
2490 && journal.projected_roots.is_empty()
2491 && journal.declared_conflicts.is_empty()
2492 && fence_roots.is_subset(&roots)
2493 && journal.paused_root.is_none()
2494 }
2495 MultiRootPublicationPhase::Conflicted => {
2496 no_publication_progress
2497 && journal.fences.is_empty()
2498 && !journal.conflicts.is_empty()
2499 && journal.paused_root.is_none()
2500 }
2501 MultiRootPublicationPhase::Aborting => {
2502 journal.published_generations.is_empty()
2503 && journal.fences.is_empty()
2504 && !journal.conflicts.is_empty()
2505 && journal.declared_conflicts.is_empty()
2506 && journal.paused_root.is_none()
2507 }
2508 MultiRootPublicationPhase::Committed => {
2509 no_publication_progress
2510 && no_conflict_state
2511 && fence_roots == roots
2512 && journal.paused_root.is_none()
2513 }
2514 MultiRootPublicationPhase::Publishing => {
2515 no_conflict_state
2516 && fence_roots == roots
2517 && generation_roots == journal.published_roots
2518 && journal.paused_root.is_none()
2519 }
2520 MultiRootPublicationPhase::Paused => {
2521 no_conflict_state
2522 && fence_roots == roots
2523 && generation_roots == journal.published_roots
2524 && journal.paused_root.is_some_and(|root| {
2525 roots.contains(&root) && !journal.published_roots.contains(&root)
2526 })
2527 }
2528 MultiRootPublicationPhase::Applied => {
2529 journal.conflicts.is_empty()
2530 && journal.projected_roots.is_empty()
2531 && fence_roots == roots
2532 && journal.published_roots == roots
2533 && generation_roots == roots
2534 && journal.paused_root.is_none()
2535 }
2536 };
2537 if journal.version != MULTI_ROOT_VERSION
2538 || journal.revision == 0
2539 || !journal.published_roots.is_subset(&roots)
2540 || !journal.conflicts.keys().all(|root| roots.contains(root))
2541 || !journal
2542 .projected_roots
2543 .keys()
2544 .all(|root| roots.contains(root))
2545 || !journal
2546 .declared_conflicts
2547 .keys()
2548 .all(|root| roots.contains(root))
2549 || !phase_valid
2550 {
2551 return Err(());
2552 }
2553 validate_candidate(&journal.candidate)
2554}
2555
2556#[derive(Default)]
2558pub struct MemoryMultiRootPublicationStore {
2559 records: Mutex<BTreeMap<OperationId, MultiRootPublication>>,
2560 parents: Mutex<BTreeMap<WorkspaceContextId, OperationId>>,
2561 #[cfg(test)]
2562 fail_claim_once: std::sync::atomic::AtomicBool,
2563}
2564
2565#[derive(Debug, Error)]
2567#[error("multi-root publication memory store is unavailable")]
2568pub struct MemoryMultiRootPublicationStoreError;
2569
2570impl MultiRootPublicationStore for MemoryMultiRootPublicationStore {
2571 type Error = MemoryMultiRootPublicationStoreError;
2572
2573 async fn load(
2574 &self,
2575 operation_id: OperationId,
2576 ) -> Result<Option<MultiRootPublication>, Self::Error> {
2577 self.records
2578 .lock()
2579 .map_err(|_| MemoryMultiRootPublicationStoreError)
2580 .map(|records| records.get(&operation_id).cloned())
2581 }
2582
2583 async fn compare_and_swap(
2584 &self,
2585 operation_id: OperationId,
2586 expected_revision: u64,
2587 replacement: MultiRootPublication,
2588 ) -> Result<bool, Self::Error> {
2589 let mut records = self
2590 .records
2591 .lock()
2592 .map_err(|_| MemoryMultiRootPublicationStoreError)?;
2593 if records.get(&operation_id).map_or(0, |value| value.revision) != expected_revision {
2594 return Ok(false);
2595 }
2596 records.insert(operation_id, replacement);
2597 Ok(true)
2598 }
2599
2600 async fn list_operations(&self) -> Result<Vec<OperationId>, Self::Error> {
2601 let records = self
2602 .records
2603 .lock()
2604 .map_err(|_| MemoryMultiRootPublicationStoreError)?;
2605 Ok(records.keys().copied().collect())
2606 }
2607
2608 async fn compare_and_delete(
2609 &self,
2610 operation_id: OperationId,
2611 expected_revision: u64,
2612 ) -> Result<bool, Self::Error> {
2613 let mut records = self
2614 .records
2615 .lock()
2616 .map_err(|_| MemoryMultiRootPublicationStoreError)?;
2617 if records.get(&operation_id).map_or(0, |value| value.revision) != expected_revision {
2618 return Ok(false);
2619 }
2620 records.remove(&operation_id);
2621 Ok(true)
2622 }
2623
2624 async fn claim_parent(
2625 &self,
2626 parent_context_id: WorkspaceContextId,
2627 operation_id: OperationId,
2628 ) -> Result<bool, Self::Error> {
2629 #[cfg(test)]
2630 if self
2631 .fail_claim_once
2632 .swap(false, std::sync::atomic::Ordering::AcqRel)
2633 {
2634 return Err(MemoryMultiRootPublicationStoreError);
2635 }
2636 let mut parents = self
2637 .parents
2638 .lock()
2639 .map_err(|_| MemoryMultiRootPublicationStoreError)?;
2640 if let Some(existing) = parents.get(&parent_context_id) {
2641 Ok(*existing == operation_id)
2642 } else {
2643 parents.insert(parent_context_id, operation_id);
2644 Ok(true)
2645 }
2646 }
2647
2648 async fn release_parent(
2649 &self,
2650 parent_context_id: WorkspaceContextId,
2651 operation_id: OperationId,
2652 ) -> Result<bool, Self::Error> {
2653 let mut parents = self
2654 .parents
2655 .lock()
2656 .map_err(|_| MemoryMultiRootPublicationStoreError)?;
2657 if parents.get(&parent_context_id) != Some(&operation_id) {
2658 return Ok(true);
2659 }
2660 parents.remove(&parent_context_id);
2661 Ok(true)
2662 }
2663}
2664
2665#[cfg(test)]
2666#[allow(clippy::expect_used, clippy::panic)]
2667mod tests {
2668 use super::*;
2669 use crate::{
2670 Digest, Fs, LocalCoreStateStore, MemoryWorkspaceContextStore, TransactionCommit,
2671 WorkspaceContextRoot,
2672 };
2673 use std::path::PathBuf;
2674
2675 #[derive(Debug, Error)]
2676 #[error("test publisher lock failed")]
2677 struct PublisherError;
2678
2679 #[derive(Default)]
2680 struct Publisher {
2681 published: Mutex<BTreeSet<WorkspaceRootId>>,
2682 pause_once: Mutex<Option<WorkspaceRootId>>,
2683 finalize_fail_once: Mutex<Option<WorkspaceRootId>>,
2684 finalized: Mutex<Vec<WorkspaceRootId>>,
2685 }
2686
2687 #[derive(Debug, Error)]
2688 #[error("test authorizer failed")]
2689 struct AuthorizerError;
2690
2691 struct Allow;
2692
2693 #[derive(Debug, Error)]
2694 #[error("test resolver failed")]
2695 struct ResolverError;
2696
2697 struct Resolver<A, O> {
2698 workspaces: Mutex<BTreeMap<WorkspaceId, Workspace<A, O>>>,
2699 }
2700
2701 impl<A: AsyncAuthorityStore, O: AsyncObjectStore> WorkspaceResolver<A, O> for Resolver<A, O> {
2702 type Error = ResolverError;
2703
2704 async fn resolve(&self, workspace_id: WorkspaceId) -> Result<Workspace<A, O>, Self::Error> {
2705 self.workspaces
2706 .lock()
2707 .map_err(|_| ResolverError)?
2708 .get(&workspace_id)
2709 .cloned()
2710 .ok_or(ResolverError)
2711 }
2712 }
2713
2714 impl MultiRootPublicationAuthorizer for Allow {
2715 type Error = AuthorizerError;
2716
2717 async fn authorize(&self, _plan: &MultiRootMergePlan) -> Result<bool, Self::Error> {
2718 Ok(true)
2719 }
2720 }
2721
2722 impl MultiRootPublisher for Publisher {
2723 type Error = PublisherError;
2724
2725 async fn prepare(
2726 &self,
2727 operation_id: OperationId,
2728 _root_id: WorkspaceRootId,
2729 _root: &MultiRootMergeRoot,
2730 _resolutions: &BTreeMap<ConflictKey, MergeResolution>,
2731 ) -> Result<MultiRootPrepare, Self::Error> {
2732 Ok(MultiRootPrepare::Ready(MultiRootFence {
2733 token: operation_id.into_bytes().to_vec(),
2734 }))
2735 }
2736
2737 async fn release(
2738 &self,
2739 _operation_id: OperationId,
2740 _root_id: WorkspaceRootId,
2741 _fence: &MultiRootFence,
2742 ) -> Result<(), Self::Error> {
2743 Ok(())
2744 }
2745
2746 async fn project_conflict(
2747 &self,
2748 _operation_id: OperationId,
2749 _root_id: WorkspaceRootId,
2750 root: &MultiRootMergeRoot,
2751 _conflict: Option<&crate::MergePlan>,
2752 ) -> Result<GenerationId, Self::Error> {
2753 Ok(root.target_generation)
2754 }
2755
2756 async fn abort_conflict(
2757 &self,
2758 _operation_id: OperationId,
2759 _root_id: WorkspaceRootId,
2760 root: &MultiRootMergeRoot,
2761 _projected_generation: GenerationId,
2762 ) -> Result<GenerationId, Self::Error> {
2763 Ok(root.target_generation)
2764 }
2765
2766 async fn declare_conflicts(
2767 &self,
2768 _operation_id: OperationId,
2769 _root_id: WorkspaceRootId,
2770 _root: &MultiRootMergeRoot,
2771 conflict: &crate::MergePlan,
2772 _paths: &[String],
2773 ) -> Result<BTreeSet<ConflictKey>, Self::Error> {
2774 Ok(conflict
2775 .conflicts
2776 .iter()
2777 .map(|view| view.key.clone())
2778 .collect())
2779 }
2780
2781 async fn finish_conflict(
2782 &self,
2783 _operation_id: OperationId,
2784 _root_id: WorkspaceRootId,
2785 root: &MultiRootMergeRoot,
2786 _projected_generation: GenerationId,
2787 ) -> Result<MultiRootConflictFinish, Self::Error> {
2788 Ok(MultiRootConflictFinish {
2789 generation: root.target_generation,
2790 fence: MultiRootFence { token: Vec::new() },
2791 })
2792 }
2793
2794 async fn publish(
2795 &self,
2796 _operation_id: OperationId,
2797 root_id: WorkspaceRootId,
2798 root: &MultiRootMergeRoot,
2799 _resolutions: &BTreeMap<ConflictKey, MergeResolution>,
2800 _fence: &MultiRootFence,
2801 ) -> Result<MultiRootPublishRoot, Self::Error> {
2802 let mut pause = self.pause_once.lock().map_err(|_| PublisherError)?;
2803 if *pause == Some(root_id) {
2804 *pause = None;
2805 return Ok(MultiRootPublishRoot::UnexpectedTarget);
2806 }
2807 let inserted = self
2808 .published
2809 .lock()
2810 .map_err(|_| PublisherError)?
2811 .insert(root_id);
2812 Ok(if inserted {
2813 MultiRootPublishRoot::Published(root.source_generation)
2814 } else {
2815 MultiRootPublishRoot::AlreadyPublished(root.source_generation)
2816 })
2817 }
2818
2819 async fn finalize(
2820 &self,
2821 _operation_id: OperationId,
2822 root_id: WorkspaceRootId,
2823 _root: &MultiRootMergeRoot,
2824 ) -> Result<(), Self::Error> {
2825 self.finalized
2826 .lock()
2827 .map_err(|_| PublisherError)?
2828 .push(root_id);
2829 let mut fail = self.finalize_fail_once.lock().map_err(|_| PublisherError)?;
2830 if *fail == Some(root_id) {
2831 *fail = None;
2832 return Err(PublisherError);
2833 }
2834 Ok(())
2835 }
2836 }
2837
2838 fn generation(byte: u8) -> GenerationId {
2839 GenerationId::new(Digest::from_bytes([byte; 32]))
2840 }
2841
2842 fn candidate() -> MultiRootMergeCandidate {
2843 let roots = [1_u8, 2]
2844 .into_iter()
2845 .map(|byte| {
2846 (
2847 WorkspaceRootId::from_bytes([byte; 16]),
2848 MultiRootMergeRoot {
2849 source_workspace_id: WorkspaceId::from_bytes([byte; 16]),
2850 merge_workspace_id: None,
2851 source_generation: generation(byte),
2852 target_workspace_id: WorkspaceId::from_bytes([byte + 10; 16]),
2853 target_generation: generation(byte + 10),
2854 base_generation: generation(0),
2855 child_wins_bindings: BTreeSet::new(),
2856 },
2857 )
2858 })
2859 .collect::<BTreeMap<_, _>>();
2860 MultiRootMergeCandidate {
2861 plan: MultiRootMergePlan {
2862 operation_id: OperationId::from_bytes([7; 16]),
2863 parent_context_id: WorkspaceContextId::from_bytes([8; 16]),
2864 child_context_id: WorkspaceContextId::from_bytes([9; 16]),
2865 roots,
2866 },
2867 resolutions: [1_u8, 2]
2868 .into_iter()
2869 .map(|byte| (WorkspaceRootId::from_bytes([byte; 16]), BTreeMap::new()))
2870 .collect(),
2871 }
2872 }
2873
2874 #[tokio::test]
2875 async fn parent_claim_allows_only_one_unacknowledged_publication() {
2876 let coordinator = MultiRootPublicationCoordinator::new(
2877 MemoryMultiRootPublicationStore::default(),
2878 Publisher::default(),
2879 Allow,
2880 );
2881 let first = candidate();
2882 let Publication::Applied(retained) = coordinator
2883 .publish_retained(first.clone())
2884 .await
2885 .expect("first retained publication")
2886 else {
2887 panic!("expected applied publication");
2888 };
2889 let mut second = first;
2890 second.plan.operation_id = OperationId::from_bytes([0x55; 16]);
2891 assert!(matches!(
2892 coordinator.publish_retained(second.clone()).await,
2893 Err(MultiRootPublicationError::ParentBusy)
2894 ));
2895 assert_eq!(
2896 coordinator
2897 .pending_operations()
2898 .await
2899 .expect("list retained publication"),
2900 [retained.candidate.plan.operation_id]
2901 );
2902 coordinator
2903 .acknowledge_applied(&retained)
2904 .await
2905 .expect("acknowledge first publication");
2906 assert!(matches!(
2907 coordinator
2908 .publish_retained(second)
2909 .await
2910 .expect("second publication after acknowledgement"),
2911 Publication::Applied(_)
2912 ));
2913 }
2914
2915 #[tokio::test]
2916 async fn add_all_declares_pathless_typed_conflicts() -> Result<(), Box<dyn std::error::Error>> {
2917 let fs = Fs::memory();
2918 let target = fs.create_workspace("pathless-conflict-target").await?;
2919 let head = target.head().await?;
2920 let publisher = WorkspaceMultiRootPublisher::new(Resolver {
2921 workspaces: Mutex::new(BTreeMap::from([(target.id(), target.clone())])),
2922 });
2923 let directory_id = crate::FileId::new();
2924 let key = ConflictKey::Binding {
2925 directory_id,
2926 name: b"entry".to_vec(),
2927 };
2928 let plan = crate::MergePlan {
2929 base: head.id(),
2930 ours: head.id(),
2931 theirs: head.id(),
2932 conflicts: vec![crate::ConflictView {
2933 key: key.clone(),
2934 path: None,
2935 kind: crate::ConflictKind::Binding,
2936 base: crate::ConflictValue::Binding(None),
2937 ours: crate::ConflictValue::Binding(None),
2938 theirs: crate::ConflictValue::Binding(None),
2939 }],
2940 truncated: false,
2941 };
2942 let root = MultiRootMergeRoot {
2943 source_workspace_id: target.id(),
2944 merge_workspace_id: None,
2945 source_generation: head.id(),
2946 target_workspace_id: target.id(),
2947 target_generation: head.id(),
2948 base_generation: head.id(),
2949 child_wins_bindings: BTreeSet::new(),
2950 };
2951
2952 assert!(
2953 publisher
2954 .declare_conflicts(
2955 OperationId::from_bytes([0x71; 16]),
2956 WorkspaceRootId::from_bytes([0x72; 16]),
2957 &root,
2958 &plan,
2959 &["specific/path".to_owned()],
2960 )
2961 .await?
2962 .is_empty()
2963 );
2964 assert_eq!(
2965 publisher
2966 .declare_conflicts(
2967 OperationId::from_bytes([0x71; 16]),
2968 WorkspaceRootId::from_bytes([0x72; 16]),
2969 &root,
2970 &plan,
2971 &[".".to_owned()],
2972 )
2973 .await?,
2974 BTreeSet::from([key])
2975 );
2976 Ok(())
2977 }
2978
2979 #[test]
2980 fn unchanged_child_generation_is_a_valid_noop_publication() {
2981 let mut candidate = candidate();
2982 for root in candidate.plan.roots.values_mut() {
2983 root.source_generation = root.target_generation;
2984 }
2985 assert!(validate_candidate(&candidate).is_ok());
2986 }
2987
2988 #[tokio::test]
2989 async fn committed_publication_pauses_and_rolls_forward_on_retry()
2990 -> Result<(), Box<dyn std::error::Error>> {
2991 let publisher = Publisher::default();
2992 *publisher.pause_once.lock().expect("pause lock") =
2993 Some(WorkspaceRootId::from_bytes([2; 16]));
2994 let coordinator = MultiRootPublicationCoordinator::new(
2995 MemoryMultiRootPublicationStore::default(),
2996 publisher,
2997 Allow,
2998 );
2999 let candidate = candidate();
3000 let first = coordinator
3001 .publish(candidate.clone())
3002 .await
3003 .expect("first publication");
3004 let first = match first {
3005 Publication::Paused(publication) => publication,
3006 Publication::Applied(_)
3007 | Publication::Conflicted(_)
3008 | Publication::StaleBeforeCommit(_) => {
3009 return Err("publication should pause".into());
3010 }
3011 };
3012 assert_eq!(first.phase, MultiRootPublicationPhase::Paused);
3013 assert_eq!(first.published_roots.len(), 1);
3014
3015 let second = coordinator
3016 .publish(candidate)
3017 .await
3018 .expect("recovered publication");
3019 let second = match second {
3020 Publication::Applied(publication) => publication,
3021 Publication::Paused(_)
3022 | Publication::Conflicted(_)
3023 | Publication::StaleBeforeCommit(_) => {
3024 return Err("publication should complete".into());
3025 }
3026 };
3027 assert_eq!(second.phase, MultiRootPublicationPhase::Applied);
3028 assert_eq!(second.published_roots.len(), 2);
3029 assert!(
3030 coordinator
3031 .pending_operations()
3032 .await
3033 .expect("terminal records are reclaimed")
3034 .is_empty()
3035 );
3036 Ok::<(), Box<dyn std::error::Error>>(())
3037 }
3038
3039 #[tokio::test]
3040 async fn local_store_restart_resumes_committed_publication_without_republishing_roots()
3041 -> Result<(), Box<dyn std::error::Error>> {
3042 let directory = tempfile::tempdir()?;
3043 let operation_id = candidate().plan.operation_id;
3044 let first_publisher = Publisher::default();
3045 *first_publisher.pause_once.lock().expect("pause lock") =
3046 Some(WorkspaceRootId::from_bytes([2; 16]));
3047 let first = MultiRootPublicationCoordinator::new(
3048 LocalCoreStateStore::new(directory.path()),
3049 first_publisher,
3050 Allow,
3051 );
3052 let Publication::Paused(paused) = first.publish(candidate()).await? else {
3053 return Err("publication should pause".into());
3054 };
3055 assert_eq!(
3056 paused.published_roots,
3057 [WorkspaceRootId::from_bytes([1; 16])].into()
3058 );
3059 drop(first);
3060
3061 let reopened = MultiRootPublicationCoordinator::new(
3062 LocalCoreStateStore::new(directory.path()),
3063 Publisher::default(),
3064 Allow,
3065 );
3066 let resumed = reopened.resume(operation_id).await?;
3067 let Some(Publication::Applied(applied)) = resumed else {
3068 return Err("reopened publication should complete".into());
3069 };
3070 assert_eq!(applied.phase, MultiRootPublicationPhase::Applied);
3071 assert_eq!(applied.published_roots.len(), 2);
3072 assert_eq!(
3073 *reopened.publisher.published.lock().expect("published lock"),
3074 [WorkspaceRootId::from_bytes([2; 16])].into_iter().collect()
3075 );
3076 assert!(reopened.pending_operations().await?.is_empty());
3077 Ok(())
3078 }
3079
3080 #[tokio::test]
3081 async fn journal_created_before_parent_claim_survives_claim_failure()
3082 -> Result<(), Box<dyn std::error::Error>> {
3083 let store = MemoryMultiRootPublicationStore::default();
3084 store
3085 .fail_claim_once
3086 .store(true, std::sync::atomic::Ordering::Release);
3087 let coordinator = MultiRootPublicationCoordinator::new(store, Publisher::default(), Allow);
3088 let operation_id = candidate().plan.operation_id;
3089
3090 assert!(matches!(
3091 coordinator.publish(candidate()).await,
3092 Err(MultiRootPublicationError::Store(_))
3093 ));
3094 assert_eq!(coordinator.pending_operations().await?, [operation_id]);
3095
3096 assert!(matches!(
3097 coordinator.publish(candidate()).await?,
3098 Publication::Applied(_)
3099 ));
3100 assert!(coordinator.pending_operations().await?.is_empty());
3101 Ok(())
3102 }
3103
3104 #[tokio::test]
3105 async fn finalize_failure_retries_after_the_applied_decision()
3106 -> Result<(), Box<dyn std::error::Error>> {
3107 let publisher = Publisher::default();
3108 *publisher.finalize_fail_once.lock().expect("finalize lock") =
3109 Some(WorkspaceRootId::from_bytes([2; 16]));
3110 let coordinator = MultiRootPublicationCoordinator::new(
3111 MemoryMultiRootPublicationStore::default(),
3112 publisher,
3113 Allow,
3114 );
3115 let operation_id = candidate().plan.operation_id;
3116 assert!(matches!(
3117 coordinator.publish(candidate()).await,
3118 Err(MultiRootPublicationError::Publisher(_))
3119 ));
3120 let pending = coordinator.pending_operations().await?;
3121 assert_eq!(pending, vec![operation_id]);
3122
3123 let resumed = coordinator.resume(operation_id).await?;
3124 assert!(matches!(resumed, Some(Publication::Applied(_))));
3125 assert!(coordinator.pending_operations().await?.is_empty());
3126 let finalized = coordinator
3127 .publisher
3128 .finalized
3129 .lock()
3130 .expect("finalized lock");
3131 assert_eq!(
3132 finalized.as_slice(),
3133 &[
3134 WorkspaceRootId::from_bytes([1; 16]),
3135 WorkspaceRootId::from_bytes([2; 16]),
3136 WorkspaceRootId::from_bytes([1; 16]),
3137 WorkspaceRootId::from_bytes([2; 16]),
3138 ]
3139 );
3140 Ok(())
3141 }
3142
3143 #[tokio::test]
3144 async fn context_authorizer_rejects_forged_root_bindings()
3145 -> Result<(), Box<dyn std::error::Error>> {
3146 let registry = WorkspaceContextRegistry::new(MemoryWorkspaceContextStore::new());
3147 let plan = candidate().plan;
3148 let parent_roots = plan
3149 .roots
3150 .iter()
3151 .map(|(root_id, root)| WorkspaceContextRoot {
3152 root_id: *root_id,
3153 source_path: PathBuf::from(if cfg!(windows) {
3154 format!("C:\\roots\\{}", root_id.into_bytes()[0])
3155 } else {
3156 format!("/roots/{}", root_id.into_bytes()[0])
3157 }),
3158 workspace_id: root.target_workspace_id,
3159 workspace_name: format!("parent-{}", root_id.into_bytes()[0]),
3160 parent_workspace_id: None,
3161 mount_path: None,
3162 })
3163 .collect::<Vec<_>>();
3164 registry
3165 .register_root(plan.parent_context_id, parent_roots.clone())
3166 .await?;
3167 let child_roots = plan
3168 .roots
3169 .iter()
3170 .zip(parent_roots)
3171 .map(|((root_id, root), parent)| WorkspaceContextRoot {
3172 root_id: *root_id,
3173 source_path: parent.source_path,
3174 workspace_id: root.source_workspace_id,
3175 workspace_name: format!("child-{}", root_id.into_bytes()[0]),
3176 parent_workspace_id: Some(root.target_workspace_id),
3177 mount_path: None,
3178 })
3179 .collect::<Vec<_>>();
3180 registry
3181 .register_child(plan.child_context_id, plan.parent_context_id, child_roots)
3182 .await?;
3183 assert!(registry.authorize(&plan).await?);
3184
3185 let mut touched_only = plan.clone();
3186 touched_only
3187 .roots
3188 .remove(&WorkspaceRootId::from_bytes([2; 16]));
3189 assert!(registry.authorize(&touched_only).await?);
3190
3191 let mut forged = plan;
3192 forged
3193 .roots
3194 .get_mut(&WorkspaceRootId::from_bytes([1; 16]))
3195 .expect("root")
3196 .target_workspace_id = WorkspaceId::from_bytes([99; 16]);
3197 assert!(!registry.authorize(&forged).await?);
3198 Ok(())
3199 }
3200
3201 #[tokio::test]
3202 async fn workspace_publisher_previews_then_publishes_the_pinned_child_generation()
3203 -> Result<(), Box<dyn std::error::Error>> {
3204 let fs = Fs::memory();
3205 let parent = fs.create_workspace("publisher-parent").await?;
3206 let mut base_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3207 base_tx.write_text("/base.txt", "base").await?;
3208 let TransactionCommit::Committed(base) = base_tx.commit().await? else {
3209 return Err("base commit did not publish".into());
3210 };
3211 let child = parent
3212 .fork(
3213 "publisher-child",
3214 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3215 )
3216 .await?;
3217 let mut child_tx = child.begin_transaction(IdempotencyKey::new()).await?;
3218 child_tx.write_text("/child.txt", "child").await?;
3219 let TransactionCommit::Committed(child_head) = child_tx.commit().await? else {
3220 return Err("child commit did not publish".into());
3221 };
3222 let mut parent_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3223 parent_tx.write_text("/parent.txt", "parent").await?;
3224 let TransactionCommit::Committed(parent_head) = parent_tx.commit().await? else {
3225 return Err("parent commit did not publish".into());
3226 };
3227 let resolver = Resolver {
3228 workspaces: Mutex::new(BTreeMap::new()),
3229 };
3230 resolver
3231 .workspaces
3232 .lock()
3233 .expect("resolver lock")
3234 .extend([(parent.id(), parent.clone()), (child.id(), child.clone())]);
3235 let publisher = WorkspaceMultiRootPublisher::new(resolver);
3236 let operation = OperationId::from_bytes([31; 16]);
3237 let root_id = WorkspaceRootId::from_bytes([32; 16]);
3238 let root = MultiRootMergeRoot {
3239 source_workspace_id: child.id(),
3240 merge_workspace_id: None,
3241 source_generation: child_head.id(),
3242 target_workspace_id: parent.id(),
3243 target_generation: parent_head.id(),
3244 base_generation: base.id(),
3245 child_wins_bindings: BTreeSet::new(),
3246 };
3247 let resolutions = BTreeMap::new();
3248 let fence = match publisher
3249 .prepare(operation, root_id, &root, &resolutions)
3250 .await?
3251 {
3252 MultiRootPrepare::Ready(fence) => fence,
3253 MultiRootPrepare::Stale | MultiRootPrepare::Conflicted(_) => {
3254 return Err("preview rejected a clean join".into());
3255 }
3256 };
3257 assert_eq!(parent.head().await?.id(), parent_head.id());
3258 let published = publisher
3259 .publish(operation, root_id, &root, &resolutions, &fence)
3260 .await?;
3261 let published_generation = parent.head().await?.id();
3262 assert!(matches!(
3263 published,
3264 MultiRootPublishRoot::Published(generation)
3265 if generation == published_generation
3266 ));
3267 assert_eq!(parent.read("/child.txt", 16).await?.as_ref(), b"child");
3268 assert_eq!(parent.read("/parent.txt", 16).await?.as_ref(), b"parent");
3269 assert_eq!(
3270 publisher
3271 .publish(operation, root_id, &root, &resolutions, &fence)
3272 .await?,
3273 MultiRootPublishRoot::AlreadyPublished(published_generation)
3274 );
3275 Ok(())
3276 }
3277
3278 #[tokio::test]
3279 async fn automatic_text_markers_are_reported_not_published()
3280 -> Result<(), Box<dyn std::error::Error>> {
3281 let fs = Fs::memory();
3282 let parent = fs.create_workspace("publisher-conflict-parent").await?;
3283 let mut base_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3284 base_tx.write_text("/conflicted.txt", "base\n").await?;
3285 let TransactionCommit::Committed(base) = base_tx.commit().await? else {
3286 return Err("base commit did not publish".into());
3287 };
3288 let child = parent
3289 .fork(
3290 "publisher-conflict-child",
3291 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3292 )
3293 .await?;
3294 let mut child_tx = child.begin_transaction(IdempotencyKey::new()).await?;
3295 child_tx.write_text("/conflicted.txt", "child\n").await?;
3296 let TransactionCommit::Committed(child_head) = child_tx.commit().await? else {
3297 return Err("child commit did not publish".into());
3298 };
3299 let mut parent_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3300 parent_tx.write_text("/conflicted.txt", "parent\n").await?;
3301 let TransactionCommit::Committed(parent_head) = parent_tx.commit().await? else {
3302 return Err("parent commit did not publish".into());
3303 };
3304 let resolver = Resolver {
3305 workspaces: Mutex::new(BTreeMap::from([
3306 (parent.id(), parent.clone()),
3307 (child.id(), child.clone()),
3308 ])),
3309 };
3310 let mut registry = MergeDriverRegistry::new();
3311 registry.register("text", std::sync::Arc::new(crate::DefaultTextMergeDriver))?;
3312 registry.set_default("text")?;
3313 let publisher = WorkspaceMultiRootPublisher::new(resolver)
3314 .with_merge_drivers(std::sync::Arc::new(registry));
3315 let root = MultiRootMergeRoot {
3316 source_workspace_id: child.id(),
3317 merge_workspace_id: None,
3318 source_generation: child_head.id(),
3319 target_workspace_id: parent.id(),
3320 target_generation: parent_head.id(),
3321 base_generation: base.id(),
3322 child_wins_bindings: BTreeSet::new(),
3323 };
3324 assert!(matches!(
3325 publisher
3326 .prepare(
3327 OperationId::from_bytes([41; 16]),
3328 WorkspaceRootId::from_bytes([42; 16]),
3329 &root,
3330 &BTreeMap::new(),
3331 )
3332 .await,
3333 Ok(MultiRootPrepare::Conflicted(_))
3334 ));
3335 assert_eq!(parent.head().await?.id(), parent_head.id());
3336 assert_eq!(
3337 parent.read("/conflicted.txt", 32).await?.as_ref(),
3338 b"parent\n"
3339 );
3340 Ok(())
3341 }
3342
3343 #[tokio::test]
3344 async fn child_wins_binding_policy_overrides_explicit_parent_resolution()
3345 -> Result<(), Box<dyn std::error::Error>> {
3346 let fs = Fs::memory();
3347 let parent = fs.create_workspace("child-wins-parent").await?;
3348 parent.write_text("/base.txt", "base").await?;
3349 let base = parent.head().await?;
3350 let child = parent
3351 .fork(
3352 "child-wins-child",
3353 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3354 )
3355 .await?;
3356 child.write_text("/cache.tmp", "child").await?;
3357 parent.write_text("/cache.tmp", "parent").await?;
3358 let child_head = child.head().await?;
3359 let parent_head = parent.head().await?;
3360 let resolver = Resolver {
3361 workspaces: Mutex::new(BTreeMap::from([
3362 (parent.id(), parent.clone()),
3363 (child.id(), child.clone()),
3364 ])),
3365 };
3366 let publisher = WorkspaceMultiRootPublisher::new(resolver);
3367 let operation = OperationId::from_bytes([81; 16]);
3368 let root_id = WorkspaceRootId::from_bytes([82; 16]);
3369 let mut root = MultiRootMergeRoot {
3370 source_workspace_id: child.id(),
3371 merge_workspace_id: None,
3372 source_generation: child_head.id(),
3373 target_workspace_id: parent.id(),
3374 target_generation: parent_head.id(),
3375 base_generation: base.id(),
3376 child_wins_bindings: BTreeSet::new(),
3377 };
3378 let MultiRootPrepare::Conflicted(plan) = publisher
3379 .prepare(operation, root_id, &root, &BTreeMap::new())
3380 .await
3381 .map_err(|error| format!("initial preview: {error}"))?
3382 else {
3383 return Err("same-name creations did not conflict".into());
3384 };
3385 let binding = plan
3386 .conflicts
3387 .iter()
3388 .find(|conflict| {
3389 conflict.kind == crate::ConflictKind::Binding
3390 && conflict.path.as_deref() == Some("/cache.tmp")
3391 })
3392 .ok_or("binding conflict missing")?;
3393 let resolutions = BTreeMap::from([(
3394 binding.key.clone(),
3395 MergeResolution::Select(ConflictSide::Ours),
3396 )]);
3397 root.child_wins_bindings.insert("/cache.tmp".to_owned());
3398 let MultiRootPrepare::Ready(fence) = publisher
3399 .prepare(operation, root_id, &root, &resolutions)
3400 .await
3401 .map_err(|error| format!("resolved preview: {error}"))?
3402 else {
3403 return Err("child-wins binding was not ready".into());
3404 };
3405 assert!(matches!(
3406 publisher
3407 .publish(operation, root_id, &root, &resolutions, &fence)
3408 .await
3409 .map_err(|error| format!("resolved publication: {error}"))?,
3410 MultiRootPublishRoot::Published(_)
3411 ));
3412 assert_eq!(parent.read("/cache.tmp", 16).await?.as_ref(), b"child");
3413 Ok(())
3414 }
3415
3416 #[tokio::test]
3417 async fn independent_same_name_directory_creations_return_a_binding_conflict()
3418 -> Result<(), Box<dyn std::error::Error>> {
3419 let fs = Fs::memory();
3420 let parent = fs.create_workspace("directory-race-parent").await?;
3421 let base = parent.head().await?;
3422 let child = parent
3423 .fork(
3424 "directory-race-child",
3425 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3426 )
3427 .await?;
3428 let mut child_tx = child.begin_transaction(IdempotencyKey::new()).await?;
3429 child_tx.create_dir_all("/sub").await?;
3430 child_tx.write_text("/.gitignore", "*.tmp\n").await?;
3431 child_tx.write_text("/scratch.tmp", "child").await?;
3432 child_tx.write_text("/sub/.gitignore", "*.tmp\n").await?;
3433 child_tx.write_text("/sub/cache.tmp", "child").await?;
3434 child_tx.create_dir_all("/sub/nested").await?;
3435 child_tx
3436 .write_text("/sub/nested/deep.tmp", "nested child")
3437 .await?;
3438 child_tx.commit().await?;
3439 let mut parent_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3440 parent_tx.create_dir_all("/sub").await?;
3441 parent_tx.write_text("/scratch.tmp", "parent").await?;
3442 parent_tx.write_text("/sub/cache.tmp", "parent").await?;
3443 parent_tx.commit().await?;
3444 let resolver = Resolver {
3445 workspaces: Mutex::new(BTreeMap::from([
3446 (parent.id(), parent.clone()),
3447 (child.id(), child.clone()),
3448 ])),
3449 };
3450 let publisher = WorkspaceMultiRootPublisher::new(resolver);
3451 let root = MultiRootMergeRoot {
3452 source_workspace_id: child.id(),
3453 merge_workspace_id: None,
3454 source_generation: child.head().await?.id(),
3455 target_workspace_id: parent.id(),
3456 target_generation: parent.head().await?.id(),
3457 base_generation: base.id(),
3458 child_wins_bindings: BTreeSet::from([
3459 "/scratch.tmp".to_owned(),
3460 "/sub/cache.tmp".to_owned(),
3461 ]),
3462 };
3463 let MultiRootPrepare::Conflicted(plan) = publisher
3464 .prepare(
3465 OperationId::from_bytes([83; 16]),
3466 WorkspaceRootId::from_bytes([84; 16]),
3467 &root,
3468 &BTreeMap::new(),
3469 )
3470 .await?
3471 else {
3472 return Err("independently created directories did not conflict".into());
3473 };
3474 let directory_conflict = plan
3475 .conflicts
3476 .iter()
3477 .find(|conflict| {
3478 conflict.kind == crate::ConflictKind::Binding
3479 && conflict.path.as_deref() == Some("/sub")
3480 })
3481 .ok_or("directory binding conflict missing")?;
3482 let MultiRootPrepare::Ready(fence) = publisher
3483 .prepare(
3484 OperationId::from_bytes([83; 16]),
3485 WorkspaceRootId::from_bytes([84; 16]),
3486 &root,
3487 &BTreeMap::from([(
3488 directory_conflict.key.clone(),
3489 MergeResolution::Select(ConflictSide::Theirs),
3490 )]),
3491 )
3492 .await?
3493 else {
3494 return Err("resolved directory binding was not ready".into());
3495 };
3496 publisher
3497 .release(
3498 OperationId::from_bytes([83; 16]),
3499 WorkspaceRootId::from_bytes([84; 16]),
3500 &fence,
3501 )
3502 .await?;
3503 publisher
3504 .project_conflict(
3505 OperationId::from_bytes([85; 16]),
3506 WorkspaceRootId::from_bytes([84; 16]),
3507 &root,
3508 Some(&plan),
3509 )
3510 .await?;
3511 assert_eq!(parent.read("/sub/cache.tmp", 16).await?.as_ref(), b"parent");
3512 assert_eq!(child.read("/sub/cache.tmp", 16).await?.as_ref(), b"child");
3513 Ok(())
3514 }
3515
3516 #[tokio::test]
3517 async fn conflicted_publication_projects_edits_and_continues_durably()
3518 -> Result<(), Box<dyn std::error::Error>> {
3519 Box::pin(conflicted_publication_projects_edits_and_continues_durably_case()).await
3520 }
3521
3522 async fn conflicted_publication_projects_edits_and_continues_durably_case()
3523 -> Result<(), Box<dyn std::error::Error>> {
3524 let fs = Fs::memory();
3525 let parent = fs.create_workspace("conflict-parent").await?;
3526 let mut base_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3527 base_tx.write_text("/file.txt", "base\n").await?;
3528 let TransactionCommit::Committed(base) = base_tx.commit().await? else {
3529 return Err("base did not commit".into());
3530 };
3531 let child = parent
3532 .fork(
3533 "conflict-child",
3534 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3535 )
3536 .await?;
3537 let mut child_tx = child.begin_transaction(IdempotencyKey::new()).await?;
3538 child_tx.write_text("/file.txt", "child\n").await?;
3539 let TransactionCommit::Committed(child_head) = child_tx.commit().await? else {
3540 return Err("child did not commit".into());
3541 };
3542 let mut parent_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3543 parent_tx.write_text("/file.txt", "parent\n").await?;
3544 let TransactionCommit::Committed(parent_head) = parent_tx.commit().await? else {
3545 return Err("parent did not commit".into());
3546 };
3547 let resolver = Resolver {
3548 workspaces: Mutex::new(BTreeMap::from([
3549 (parent.id(), parent.clone()),
3550 (child.id(), child.clone()),
3551 ])),
3552 };
3553 let mut drivers = MergeDriverRegistry::new();
3554 drivers.register("text", std::sync::Arc::new(DefaultTextMergeDriver))?;
3555 drivers.set_default("text")?;
3556 let operation_id = OperationId::from_bytes([71; 16]);
3557 let root_id = WorkspaceRootId::from_bytes([72; 16]);
3558 let candidate = MultiRootMergeCandidate {
3559 plan: MultiRootMergePlan {
3560 operation_id,
3561 parent_context_id: WorkspaceContextId::from_bytes([73; 16]),
3562 child_context_id: WorkspaceContextId::from_bytes([74; 16]),
3563 roots: BTreeMap::from([(
3564 root_id,
3565 MultiRootMergeRoot {
3566 source_workspace_id: child.id(),
3567 merge_workspace_id: None,
3568 source_generation: child_head.id(),
3569 target_workspace_id: parent.id(),
3570 target_generation: parent_head.id(),
3571 base_generation: base.id(),
3572 child_wins_bindings: BTreeSet::new(),
3573 },
3574 )]),
3575 },
3576 resolutions: BTreeMap::from([(root_id, BTreeMap::new())]),
3577 };
3578 let coordinator = MultiRootPublicationCoordinator::new(
3579 MemoryMultiRootPublicationStore::default(),
3580 WorkspaceMultiRootPublisher::new(resolver)
3581 .with_merge_drivers(std::sync::Arc::new(drivers)),
3582 Allow,
3583 );
3584 let Publication::Conflicted(conflicted) = coordinator.publish(candidate).await? else {
3585 return Err("publication did not enter conflict state".into());
3586 };
3587 assert_eq!(conflicted.projected_roots.len(), 1);
3588 assert!(
3589 String::from_utf8(parent.read("/file.txt", 1_024).await?.to_vec())?
3590 .contains("<<<<<<< ours")
3591 );
3592 assert!(matches!(
3593 coordinator
3594 .continue_conflicted(operation_id, WorkspaceContextId::from_bytes([73; 16]))
3595 .await,
3596 Err(MultiRootPublicationError::InvalidCandidate)
3597 ));
3598 assert!(matches!(
3599 coordinator
3600 .declare_conflicted_paths(
3601 operation_id,
3602 WorkspaceContextId::from_bytes([73; 16]),
3603 root_id,
3604 &["file.txt".to_owned()],
3605 )
3606 .await,
3607 Err(MultiRootPublicationError::Publisher(
3608 WorkspaceMultiRootPublisherError::Candidate(_)
3609 ))
3610 ));
3611 let mut resolution = parent.begin_transaction(IdempotencyKey::new()).await?;
3612 resolution.write_text("/file.txt", "resolved\n").await?;
3613 resolution.commit().await?;
3614 coordinator
3615 .declare_conflicted_paths(
3616 operation_id,
3617 WorkspaceContextId::from_bytes([73; 16]),
3618 root_id,
3619 &["file.txt".to_owned()],
3620 )
3621 .await
3622 .expect("resolved conflict path declaration");
3623 let declared = coordinator
3624 .store
3625 .load(operation_id)
3626 .await?
3627 .expect("conflict journal");
3628 assert_eq!(
3629 declared.conflicts[&root_id]
3630 .conflicts
3631 .iter()
3632 .map(|view| view.key.clone())
3633 .collect::<BTreeSet<_>>(),
3634 declared.declared_conflicts[&root_id]
3635 );
3636 assert!(matches!(
3637 coordinator
3638 .continue_conflicted(operation_id, WorkspaceContextId::from_bytes([73; 16]))
3639 .await?,
3640 Publication::Applied(_)
3641 ));
3642 assert_eq!(parent.read("/file.txt", 32).await?.as_ref(), b"resolved\n");
3643 assert!(coordinator.pending_operations().await?.is_empty());
3644 Ok(())
3645 }
3646
3647 #[tokio::test]
3648 async fn conflicted_publication_abort_restores_premerge_generation()
3649 -> Result<(), Box<dyn std::error::Error>> {
3650 let fs = Fs::memory();
3651 let parent = fs.create_workspace("abort-parent").await?;
3652 let mut base_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3653 base_tx.write_text("/file.txt", "base\n").await?;
3654 let TransactionCommit::Committed(base) = base_tx.commit().await? else {
3655 return Err("base did not commit".into());
3656 };
3657 let child = parent
3658 .fork(
3659 "abort-child",
3660 ForkOptions::from_generation(base.clone(), IdempotencyKey::new()),
3661 )
3662 .await?;
3663 let mut child_tx = child.begin_transaction(IdempotencyKey::new()).await?;
3664 child_tx
3665 .write_text("/file.txt", "discarded child\n")
3666 .await?;
3667 let TransactionCommit::Committed(child_head) = child_tx.commit().await? else {
3668 return Err("child did not commit".into());
3669 };
3670 let mut parent_tx = parent.begin_transaction(IdempotencyKey::new()).await?;
3671 parent_tx.write_text("/file.txt", "kept parent\n").await?;
3672 let TransactionCommit::Committed(parent_head) = parent_tx.commit().await? else {
3673 return Err("parent did not commit".into());
3674 };
3675 let resolver = Resolver {
3676 workspaces: Mutex::new(BTreeMap::from([
3677 (parent.id(), parent.clone()),
3678 (child.id(), child.clone()),
3679 ])),
3680 };
3681 let mut drivers = MergeDriverRegistry::new();
3682 drivers.register("text", std::sync::Arc::new(DefaultTextMergeDriver))?;
3683 drivers.set_default("text")?;
3684 let root_id = WorkspaceRootId::from_bytes([72; 16]);
3685 let abort_operation = OperationId::from_bytes([75; 16]);
3686 let abort_candidate = MultiRootMergeCandidate {
3687 plan: MultiRootMergePlan {
3688 operation_id: abort_operation,
3689 parent_context_id: WorkspaceContextId::from_bytes([76; 16]),
3690 child_context_id: WorkspaceContextId::from_bytes([77; 16]),
3691 roots: BTreeMap::from([(
3692 root_id,
3693 MultiRootMergeRoot {
3694 source_workspace_id: child.id(),
3695 merge_workspace_id: None,
3696 source_generation: child_head.id(),
3697 target_workspace_id: parent.id(),
3698 target_generation: parent_head.id(),
3699 base_generation: base.id(),
3700 child_wins_bindings: BTreeSet::new(),
3701 },
3702 )]),
3703 },
3704 resolutions: BTreeMap::from([(root_id, BTreeMap::new())]),
3705 };
3706 let aborting = MultiRootPublicationCoordinator::new(
3707 MemoryMultiRootPublicationStore::default(),
3708 WorkspaceMultiRootPublisher::new(resolver)
3709 .with_merge_drivers(std::sync::Arc::new(drivers)),
3710 Allow,
3711 );
3712 assert!(matches!(
3713 aborting.publish(abort_candidate).await?,
3714 Publication::Conflicted(_)
3715 ));
3716 let mut unrelated = parent.begin_transaction(IdempotencyKey::new()).await?;
3717 unrelated
3718 .write_text("/unrelated.txt", "preserved\n")
3719 .await?;
3720 let TransactionCommit::Committed(_) = unrelated.commit().await? else {
3721 return Err("unrelated edit did not commit".into());
3722 };
3723 aborting
3724 .abort_conflicted(abort_operation, WorkspaceContextId::from_bytes([76; 16]))
3725 .await?;
3726 assert_eq!(
3727 parent.read("/file.txt", 32).await?.as_ref(),
3728 b"kept parent\n"
3729 );
3730 assert_eq!(
3731 parent.read("/unrelated.txt", 32).await?.as_ref(),
3732 b"preserved\n"
3733 );
3734 assert!(aborting.pending_operations().await?.is_empty());
3735 Ok(())
3736 }
3737}