lix 0.17.1

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

use super::*;

pub(super) async fn wait_descriptor<S>(
    lease: SessionLease<S>,
    branch_id: Option<String>,
    after: Option<u64>,
) -> Result<Response, ApiError>
where
    S: Storage + Clone + Send + Sync + 'static,
{
    wait_descriptor_until(
        lease,
        branch_id,
        after,
        tokio::time::Instant::now() + SYNC_LONG_POLL_TIMEOUT,
    )
    .await
}

pub(super) async fn wait_descriptor_until<S>(
    lease: SessionLease<S>,
    branch_id: Option<String>,
    after: Option<u64>,
    deadline: tokio::time::Instant,
) -> Result<Response, ApiError>
where
    S: Storage + Clone + Send + Sync + 'static,
{
    let descriptor = load_descriptor_until(lease, branch_id, after, deadline).await?;
    bounded_sync_json_response(
        descriptor,
        "partial replica descriptor",
        crate::sync::MAX_LEASED_DESCRIPTOR_BYTES,
    )
}

pub(super) async fn load_descriptor_until<S>(
    lease: SessionLease<S>,
    branch_id: Option<String>,
    after: Option<u64>,
    deadline: tokio::time::Instant,
) -> Result<crate::sync::LeasedPartialReplicaDescriptor, ApiError>
where
    S: Storage + Clone + Send + Sync + 'static,
{
    if after.is_some() && branch_id.is_none() {
        return Err(ApiError::bad_request(
            "descriptor continuation requires the selected branchId",
        ));
    }
    if after.is_none() {
        let descriptor = lease
            .run_cancellable_read(move |lix| async move {
                lix.leased_partial_replica_descriptor(branch_id.as_deref())
                    .await
            })
            .await?;
        return Ok(descriptor);
    }
    // Subscribe first: a publication during/between descriptor read and wait
    // leaves an unseen channel version and makes changed() immediately ready.
    let mut changed = lease.record.lix.sync_mode_state().change_watcher();
    let mut finish = false;
    loop {
        let branch_id = branch_id.clone();
        let descriptor = lease
            .run_cancellable_read(move |lix| async move {
                lix.partial_replica_descriptor(branch_id.as_deref()).await
            })
            .await?;
        if after.is_some_and(|after| descriptor.cursor < after) {
            return Err(ApiError::from(LixError::new(
                LixError::CODE_TRANSACTION_CONFLICT,
                "partial descriptor cursor is ahead of the authority; admitted baseline must be reconciled",
            ).with_details(serde_json::json!({ "after":after, "cursor":descriptor.cursor, "partialBaselineResetRequired":true }))));
        }
        if after.is_none_or(|after| descriptor.cursor > after)
            || finish
            || tokio::time::Instant::now() >= deadline
        {
            let branch = descriptor.selected_branch.branch_id;
            let leased = lease
                .run_cancellable_read(move |lix| async move {
                    lix.leased_partial_replica_descriptor(Some(&branch)).await
                })
                .await?;
            return Ok(leased);
        }
        tokio::select! {
            result = changed.changed() => { finish = result.is_err(); },
            _ = tokio::time::sleep_until(deadline) => { finish = true; },
        }
        // Re-read even on timeout/closed notifier. The response always contains
        // one final coherent descriptor, without resetting the fixed deadline.
    }
}