lix 0.18.0

Embeddable version control for apps and AI agents.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

//! Background reconciliation of an authority descriptor into the retained SQL
//! working set. Every network step precedes the final owned publication task.
use super::http::{HttpSyncTransport, RawHttpClient};
use super::partial_state::PartialReplicaState;
use crate::LixError;
use crate::engine::Engine;
use crate::storage_adapter::Storage;
use std::sync::Arc;

pub(super) enum PreparedDescriptor {
    NoChange,
    /// Durable bookkeeping advanced while native serving controls stayed local.
    LocalProgress,
    Ready(super::partial_publication::PreparedPartialPublication),
}

pub(super) async fn prepare_clean_descriptor<S, C>(
    engine: Arc<Engine<S>>,
    previous: Arc<PartialReplicaState>,
    transport: &HttpSyncTransport<C>,
    wrapper: super::http::TimedLeasedPartialDescriptor,
    recovery: super::partial_publication::PartialRecoveryPolicy,
) -> Result<PreparedDescriptor, LixError>
where
    S: Storage + Clone + Send + Sync + 'static,
    C: RawHttpClient + Clone + 'static,
{
    let deadline = wrapper.deadline;
    deadline.check(&wrapper.wire.lease.lease_id)?;
    if engine.sync_mode().partial_admission().as_deref() != Some(previous.as_ref()) {
        return Err(LixError::new(
            LixError::CODE_TRANSACTION_CONFLICT,
            "lease recovery source changed",
        ));
    }
    if matches!(
        recovery,
        super::partial_publication::PartialRecoveryPolicy::ExpiredBaseline
            | super::partial_publication::PartialRecoveryPolicy::AuthorityWins
    ) {
        // A cold read can register interests and then fail for missing inputs,
        // before the session's successful-read flush. Recovery must durably
        // retain those interests (and restore a reopened registry) itself.
        // This journal write uses its own admission/epoch CAS guards; it does
        // not acquire session write admission or weaken publication's checks.
        let registry = engine.sync_mode().read_interests().ok_or_else(|| {
            LixError::new(
                LixError::CODE_TRANSACTION_CONFLICT,
                "lease recovery has no retained interests",
            )
        })?;
        super::partial_interest_journal::flush_partial_read_interests(
            &engine.storage(),
            &previous,
            &registry,
        )
        .await?;
    }
    if recovery == super::partial_publication::PartialRecoveryPolicy::ExpiredBaseline
        && super::partial_publication::same_serving_basis(
            previous.descriptor(),
            &wrapper.wire.descriptor,
        )
    {
        return super::partial_publication::prepare_lease_reacquisition(
            &engine,
            &wrapper.wire,
            deadline,
        )
        .await
        .map(PreparedDescriptor::Ready);
    }
    let next = Arc::new(previous.with_leased_descriptor_and_fresh_generations(wrapper.wire)?);
    let candidate = transport.fork_native_baseline_lease(next.baseline_lease())?;
    let storage = engine.storage();
    let mut demands = std::collections::BTreeSet::new();
    loop {
        deadline.check(&next.baseline_lease().lease_id)?;
        engine.sync_mode().ensure_partial_admission_healthy()?;
        if engine.sync_mode().partial_admission().as_deref() != Some(previous.as_ref()) {
            return Err(LixError::new(
                LixError::CODE_TRANSACTION_CONFLICT,
                "candidate source admission changed",
            ));
        }
        if next.descriptor().cursor < previous.descriptor().cursor {
            return Err(LixError::new(
                LixError::CODE_TRANSACTION_CONFLICT,
                "candidate descriptor cursor regressed",
            ));
        }
        let error = match super::partial_publication::prepare_partial_publication(
            &engine,
            next.clone(),
            deadline.clone(),
            recovery,
        )
        .await
        {
            Ok(None) => return Ok(PreparedDescriptor::NoChange),
            Ok(Some(prepared)) => {
                return Ok(PreparedDescriptor::Ready(prepared));
            }
            Err(error) => error,
        };
        let Some(demand) = super::runtime::native_sync_demand_request_for_error(&error)? else {
            return Err(error);
        };
        // Same typed immutable input cannot be demanded twice by one candidate.
        // Native corruption and unsupported history remain terminal errors.
        let fingerprint = format!("{demand:?}");
        if !demands.insert(fingerprint) {
            return Err(LixError::new(
                "LIX_PARTIAL_SCOPE_PREPARATION_STALLED",
                "candidate repeated a hydrated native dependency",
            ));
        }
        // Keep the original HTTP observation's monotonic budget across every
        // hydration retry. Neither partial progress nor transport cancellation
        // renews authority retention for this candidate.
        use futures_util::FutureExt;
        let hydrate =
            super::partial_runtime::hydrate_demand(&storage, &previous, &candidate, demand).fuse();
        let expires = super::platform::sleep(deadline.remaining()?).fuse();
        futures_util::pin_mut!(hydrate, expires);
        futures_util::select_biased! {
            _ = expires => return Err(LixError::new("LIX_PARTIAL_CANDIDATE_EXPIRED", "candidate hydration exceeded its original baseline deadline")),
            result = hydrate => result?,
        }
        deadline.check(&next.baseline_lease().lease_id)?;
    }
}