Skip to main content

acyclic_fs/
multi_root.rs

1//! Durable all-roots publication decisions.
2//!
3//! Each root still uses the ordinary immutable join/candidate APIs. This
4//! coordinator validates every pinned root before recording one commit
5//! decision, then rolls all roots forward idempotently. A changed target after
6//! the decision pauses recovery instead of overwriting external work.
7
8use 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/// One root pinned by an immutable cross-root merge plan.
28#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
29pub struct MultiRootMergeRoot {
30    /// Workspace whose pinned generation is being published.
31    pub source_workspace_id: WorkspaceId,
32    /// Optional filtered compatibility fork authenticated as a direct child of
33    /// `source_workspace_id`. Ordinary filesystem joins leave this absent.
34    #[serde(default)]
35    pub merge_workspace_id: Option<WorkspaceId>,
36    /// Child generation captured when planning began.
37    pub source_generation: GenerationId,
38    /// Exact direct-parent workspace.
39    pub target_workspace_id: WorkspaceId,
40    /// Target head that must validate before the commit decision.
41    pub target_generation: GenerationId,
42    /// Three-way common ancestor authenticated by the per-root join plan.
43    pub base_generation: GenerationId,
44    /// Caller-selected child-wins binding paths pinned with the publication
45    /// journal. This does not alter ordinary filesystem merge semantics.
46    #[serde(default)]
47    pub child_wins_bindings: BTreeSet<String>,
48}
49
50/// Immutable pinned plan for one logical multi-root publication.
51#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
52pub struct MultiRootMergePlan {
53    /// Stable retry and recovery identity.
54    pub operation_id: OperationId,
55    /// Direct-parent context receiving the publication.
56    pub parent_context_id: WorkspaceContextId,
57    /// Exact child context being published.
58    pub child_context_id: WorkspaceContextId,
59    /// Non-empty ordered root set.
60    pub roots: BTreeMap<WorkspaceRootId, MultiRootMergeRoot>,
61}
62
63/// Caller-selected declarative resolutions bound to an exact plan.
64#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
65pub struct MultiRootMergeCandidate {
66    /// Exact immutable plan.
67    pub plan: MultiRootMergePlan,
68    /// Per-root conflict declarations. Empty maps mean no conflicts.
69    pub resolutions: BTreeMap<WorkspaceRootId, BTreeMap<ConflictKey, MergeResolution>>,
70}
71
72/// Opaque durable reservation proving that one target was validated and
73/// fenced for an exact publication operation.
74#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
75pub struct MultiRootFence {
76    /// Publisher-defined stable reservation token.
77    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/// Durable logical publication phase.
87#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
88pub enum MultiRootPublicationPhase {
89    /// Every root is being checked before any durable commit decision.
90    Validating,
91    /// Typed conflicts are durable and await declarative resolution.
92    Conflicted,
93    /// A conflicted working state is being restored to its pre-merge state.
94    Aborting,
95    /// The all-roots roll-forward decision is durable.
96    Committed,
97    /// At least one root has been advanced.
98    Publishing,
99    /// Every root is durable.
100    Applied,
101    /// Unexpected concurrent target state requires reconciliation.
102    Paused,
103}
104
105/// Complete recoverable cross-root publication record.
106#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
107pub struct MultiRootPublication {
108    /// Serialization contract version.
109    pub version: u32,
110    /// Monotonic optimistic-concurrency revision.
111    pub revision: u64,
112    /// Validated unpublished candidate.
113    pub candidate: MultiRootMergeCandidate,
114    /// Durable decision/progress phase.
115    pub phase: MultiRootPublicationPhase,
116    /// Roots known durable under the commit decision.
117    pub published_roots: BTreeSet<WorkspaceRootId>,
118    /// Exact durable generation produced for each published root.
119    #[serde(default)]
120    pub published_generations: BTreeMap<WorkspaceRootId, GenerationId>,
121    /// Per-root reservations acquired before the commit decision.
122    pub fences: BTreeMap<WorkspaceRootId, MultiRootFence>,
123    /// Immutable typed conflicts pinned during validation.
124    #[serde(default)]
125    pub conflicts: BTreeMap<WorkspaceRootId, crate::MergePlan>,
126    /// Parent generations containing the durable conflict projection.
127    #[serde(default)]
128    pub projected_roots: BTreeMap<WorkspaceRootId, GenerationId>,
129    /// Conflict keys explicitly accepted from the projected working copy.
130    #[serde(default)]
131    pub declared_conflicts: BTreeMap<WorkspaceRootId, BTreeSet<ConflictKey>>,
132    /// Root that observed unexpected target state, if paused.
133    pub paused_root: Option<WorkspaceRootId>,
134}
135
136/// Result of validating and reserving one root before the commit decision.
137#[derive(Clone, Debug, Eq, PartialEq)]
138pub enum MultiRootPrepare {
139    /// The exact target is reserved for publication.
140    Ready(MultiRootFence),
141    /// The exact target changed before it could be reserved.
142    Stale,
143    /// Immutable typed conflicts require a later declarative resolution.
144    Conflicted(Box<crate::MergePlan>),
145}
146
147/// Exact resolved generation and its durable publication reservation.
148#[derive(Clone, Debug, Eq, PartialEq)]
149pub struct MultiRootConflictFinish {
150    /// Current generation descending from the projected conflict state.
151    pub generation: GenerationId,
152    /// Reservation fencing that generation until the Applied decision.
153    pub fence: MultiRootFence,
154}
155
156/// Durable optimistic store for publication decisions.
157pub trait MultiRootPublicationStore: Send + Sync {
158    /// Adapter error.
159    type Error: std::error::Error + Send + Sync + 'static;
160
161    /// Loads one operation record.
162    fn load(
163        &self,
164        operation_id: OperationId,
165    ) -> impl Future<Output = Result<Option<MultiRootPublication>, Self::Error>> + StorageFuture;
166
167    /// Atomically replaces one expected revision; zero creates the record.
168    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    /// Lists durable operations that may require restart recovery.
176    fn list_operations(
177        &self,
178    ) -> impl Future<Output = Result<Vec<OperationId>, Self::Error>> + StorageFuture;
179
180    /// Removes a terminal record only when its exact revision still matches.
181    /// Failed physical cleanup leaves the record intact for restart recovery.
182    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    /// Atomically claims the one publication slot for a parent context.
189    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    /// Releases the exact parent claim after terminal cleanup.
196    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
203/// Root-specific publisher backed by the ordinary join and materializer APIs.
204pub trait MultiRootPublisher: Send + Sync {
205    /// Publisher error.
206    type Error: std::error::Error + Send + Sync + 'static;
207
208    /// Atomically validates and reserves the exact pinned target. Repeating
209    /// this call for one operation must return the same live reservation.
210    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    /// Releases a reservation when validation fails before a commit decision.
219    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    /// Projects one root into its durable conflicted working state.
227    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    /// Restores one projected root to its exact pre-merge generation.
236    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    /// Validates explicitly staged conflict paths and returns their exact keys.
245    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    /// Captures the current resolved working generation during continue.
255    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    /// Idempotently rolls one root forward after the durable commit decision.
264    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    /// Reclaims terminal per-root recovery state after the all-roots Applied
274    /// decision is durable. It must be idempotent because restart recovery
275    /// retries cleanup independently of logical publication.
276    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
286/// Resolves one authenticated workspace identity for core publication.
287pub trait WorkspaceResolver<A, O>: Send + Sync {
288    /// Resolver failure.
289    type Error: std::error::Error + Send + Sync + 'static;
290
291    /// Opens the exact workspace or fails closed.
292    fn resolve(
293        &self,
294        workspace_id: WorkspaceId,
295    ) -> impl Future<Output = Result<Workspace<A, O>, Self::Error>> + StorageFuture;
296}
297
298/// Applies one logically published root generation to its host checkout.
299///
300/// Implementations are injected capabilities: the coordinator owns retry and
301/// ordering, while a native host may use the core journaled materializer and
302/// a non-native host may deliberately provide no physical checkout.
303pub trait MultiRootMaterializer<A, O>: Send + Sync {
304    /// Materialization failure.
305    type Error: std::error::Error + Send + Sync + 'static;
306
307    /// Idempotently applies the exact published generation for one root.
308    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    /// Reclaims terminal physical recovery state. Logical publication is
318    /// already durable when this runs, so failures are retried on recovery.
319    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
328/// SDK-owned publisher over ordinary immutable workspace joins.
329///
330/// Preparation validates pinned inputs and resolutions against a disposable
331/// SDK workspace. It never changes the target. Publication then repeats the
332/// same deterministic join through the target's fenced authority CAS.
333pub 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
341/// Workspace publisher composed with an injected physical materializer.
342///
343/// Logical publication happens first under the durable all-roots decision.
344/// If materialization is interrupted, retry discovers the exact generation
345/// from the workspace idempotency record and resumes the same materializer.
346pub 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    /// Creates one composed logical and physical publisher.
353    #[must_use]
354    pub fn new(resolver: R, materializer: M) -> Self {
355        Self {
356            publisher: WorkspaceMultiRootPublisher::new(resolver),
357            materializer,
358        }
359    }
360
361    /// Uses one immutable-input driver registry for conflict resolution.
362    #[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    /// Overrides merge-driver selection for one root in a multi-root plan.
369    #[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/// Failure from a composed logical and physical root publication.
381#[derive(Debug, Error)]
382pub enum MaterializingWorkspaceMultiRootPublisherError<
383    R: std::error::Error + 'static,
384    M: std::error::Error + 'static,
385> {
386    /// Logical workspace validation or publication failed.
387    #[error(transparent)]
388    Workspace(#[from] WorkspaceMultiRootPublisherError<R>),
389    /// The exact published generation could not be recovered.
390    #[error("published workspace generation is unavailable for materialization")]
391    MissingPublishedGeneration,
392    /// Physical checkout application failed and may be retried.
393    #[error("root materialization failed: {0}")]
394    Materializer(M),
395}
396
397impl<A, O, R> WorkspaceMultiRootPublisher<A, O, R> {
398    /// Creates a publisher over an injected authenticated resolver.
399    #[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    /// Uses one immutable-input driver registry for conflict resolution.
411    #[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    /// Overrides merge-driver selection for one root in a multi-root plan.
418    #[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    /// Borrows the resolver capability.
429    #[must_use]
430    pub const fn resolver(&self) -> &R {
431        &self.resolver
432    }
433}
434
435/// Production workspace publisher failure.
436#[derive(Debug, Error)]
437pub enum WorkspaceMultiRootPublisherError<E: std::error::Error + 'static> {
438    /// Workspace resolution failed.
439    #[error("workspace resolution failed: {0}")]
440    Resolver(E),
441    /// Workspace planning, preview, or publication failed.
442    #[error(transparent)]
443    Workspace(#[from] WorkspaceError),
444    /// Declarative conflict resolution did not match the pinned plan.
445    #[error("merge candidate validation failed: {0}")]
446    Candidate(String),
447    /// Validation found typed conflicts for which no complete resolution was supplied.
448    #[error("merge requires declarative conflict resolutions")]
449    Conflicted {
450        /// Immutable exact alternatives suitable for presentation or drivers.
451        plan: Box<crate::MergePlan>,
452    },
453    /// The supplied opaque fence does not describe these exact inputs.
454    #[error("multi-root publication fence is invalid")]
455    InvalidFence,
456    /// Disposable validation state could not be reclaimed safely.
457    #[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        // A prior preview may have been deleted after reporting conflicts.
671        // Reusing its fork identity would reopen that tombstone on continuation.
672        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
1522/// Authorization boundary for direct-parent and root-binding validation.
1523pub trait MultiRootPublicationAuthorizer: Send + Sync {
1524    /// Authorizer error.
1525    type Error: std::error::Error + Send + Sync + 'static;
1526
1527    /// Verifies that the plan publishes the exact child's bound roots only to
1528    /// its exact direct parent.
1529    fn authorize(
1530        &self,
1531        plan: &MultiRootMergePlan,
1532    ) -> impl Future<Output = Result<bool, Self::Error>> + StorageFuture;
1533
1534    /// Verifies that an interactive transition is being requested by the
1535    /// exact direct parent that owns the durable merge state.
1536    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
1572/// Context and workspace-lineage authorizer for filtered compatibility forks.
1573pub struct LineageMultiRootPublicationAuthorizer<S> {
1574    store: S,
1575}
1576
1577impl<S> LineageMultiRootPublicationAuthorizer<S> {
1578    /// Creates an authorizer over one shared durable core-state store.
1579    #[must_use]
1580    pub const fn new(store: S) -> Self {
1581        Self { store }
1582    }
1583}
1584
1585/// Failure while authenticating both context and compatibility-fork lineage.
1586#[derive(Debug, Error)]
1587pub enum LineageMultiRootPublicationAuthorizationError<E: std::error::Error + 'static> {
1588    /// Context binding validation failed.
1589    #[error(transparent)]
1590    Context(#[from] WorkspaceContextError<E>),
1591    /// Workspace lineage validation failed.
1592    #[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/// One root's publication result.
1643#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1644pub enum MultiRootPublishRoot {
1645    /// Root became durable during this attempt.
1646    Published(GenerationId),
1647    /// Exact result was already durable.
1648    AlreadyPublished(GenerationId),
1649    /// Target no longer matches and was not overwritten.
1650    UnexpectedTarget,
1651}
1652
1653/// Terminal coordinator outcome.
1654#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
1655pub enum Publication {
1656    /// Every root is durable.
1657    Applied(MultiRootPublication),
1658    /// Validation failed before a commit decision; no root was changed.
1659    StaleBeforeCommit(WorkspaceRootId),
1660    /// Typed conflicts are durable; no root was published.
1661    Conflicted(MultiRootPublication),
1662    /// Roll-forward paused on unexpected state after a commit decision.
1663    Paused(MultiRootPublication),
1664}
1665
1666enum MultiRootValidation {
1667    Committed(MultiRootPublication),
1668    Stale(WorkspaceRootId, MultiRootPublication),
1669    Conflicted(MultiRootPublication),
1670}
1671
1672/// Cross-root publication failure.
1673#[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    /// Durable state failed.
1680    #[error("multi-root publication store failed: {0}")]
1681    Store(S),
1682    /// A root-specific join/materialization failed.
1683    #[error("multi-root publisher failed: {0}")]
1684    Publisher(P),
1685    /// Context lineage authorization failed.
1686    #[error("multi-root publication authorization failed: {0}")]
1687    Authorizer(A),
1688    /// Retry supplied different immutable inputs or malformed roots.
1689    #[error("multi-root publication candidate is invalid or conflicts with its retry identity")]
1690    InvalidCandidate,
1691    /// Durable state remained contended beyond the bounded retry policy.
1692    #[error("multi-root publication state remained contended")]
1693    Contended,
1694    /// Another unresolved publication already belongs to this parent context.
1695    #[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
1701/// Core coordinator for one injected durable store and root publisher.
1702pub 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    /// Creates a coordinator.
1710    #[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    /// Resumes one durable publication, returning `None` when no such
1724    /// operation exists. The stored immutable candidate remains authoritative.
1725    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    /// Resumes publication while retaining an Applied journal until an
1733    /// external, idempotent compatibility projection has been recorded.
1734    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    /// Lists durable publication operations for restart recovery.
1777    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    /// Validates, commits, and rolls forward one exact candidate.
1787    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    /// Publishes while retaining the terminal journal for a caller-owned,
1797    /// idempotent compatibility projection before acknowledgement.
1798    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    /// Accepts the ordinary edits in a fully projected conflicted working copy.
1854    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    /// Continues a conflict while retaining the terminal journal until the
1868    /// caller acknowledges its compatibility projection.
1869    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    /// Declares the current projected contents of conflict paths as the
1883    /// caller-selected resolutions. Text paths still containing generated
1884    /// markers are rejected.
1885    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    /// Acknowledges a retained Applied journal after external compatibility
2014    /// state has been durably and idempotently projected.
2015    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    /// Restores every conflicted root to its exact pre-merge generation.
2027    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/// Process-local durable-state substitute for deterministic tests.
2557#[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/// Memory store synchronization failure.
2566#[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}