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
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

//! Child of native_global_journal_io. Explicit closed-storage maintenance;
//! ordinary opening never scans journals or performs this cleanup network work.
use super::*;

pub(crate) async fn retry_published_global_conversion_cleanup<
    S: Storage + Clone + Send + Sync + 'static,
>(
    storage: &S,
    authenticated: &crate::sync::AuthenticatedPartialConversion,
) -> Result<usize, LixError> {
    let admitted = admit_partial_epoch(storage).await?;
    let expected = &admitted.state;
    if authenticated.state().repository_id() != expected.repository_id()
        || authenticated.state().active_account_id() != expected.active_account_id()
        || authenticated.state().remote_id() != expected.remote_id()
    {
        return Err(epoch_error(
            "cleanup retry authority/account differs from published partial replica",
        ));
    }
    let (pointer, active) = load_pointer(storage)
        .await?
        .ok_or_else(|| epoch_error("cleanup retry lacks active epoch"))?;
    if !matches!(pointer, PointerState::Active { .. }) {
        return Err(epoch_error("cleanup retry cannot run during migration"));
    }
    // Inventory keys only. One bounded native journal payload is loaded at a
    // time below; the storage owner prevents concurrent migration/key insertion.
    let keys = {
        let read = storage
            .begin_read(ReadOptions {
                durability: crate::storage_adapter::StorageReadDurability::Durable,
                ..Default::default()
            })
            .await
            .map_err(storage_error)?;
        let mut cursor = read
            .begin_scan(
                REPOSITORY_EPOCH_SPACE,
                KeyRange {
                    lower: Bound::Included(Key(Bytes::from_static(b"partial-global-conversion/"))),
                    upper: Bound::Excluded(Key(Bytes::from_static(b"partial-global-conversion0"))),
                },
                BeginScanOptions {
                    projection: CoreProjection::KeyOnly,
                    ..Default::default()
                },
            )
            .await
            .map_err(storage_error)?;
        let mut keys = Vec::new();
        while let Some(chunk) = cursor.next_chunk().await.map_err(storage_error)? {
            if keys.len() + chunk.len() > 65536 {
                return Err(epoch_error(
                    "explicit conversion cleanup inventory exceeds its operation budget",
                ));
            }
            keys.extend(chunk.into_iter().map(|entry| entry.key));
        }
        keys
    };
    let mut completed = 0;
    for key in keys {
        let raw = {
            let read = storage
                .begin_read(ReadOptions {
                    durability: crate::storage_adapter::StorageReadDurability::Durable,
                    ..Default::default()
                })
                .await
                .map_err(storage_error)?;
            let keys = [key.clone()];
            let values = read
                .get_many(&[GetManyRequest {
                    space: REPOSITORY_EPOCH_SPACE,
                    keys: &keys,
                    opts: GetOptions::default(),
                }])
                .await
                .map_err(storage_error)?
                .values;
            if values.len() != 1 {
                return Err(epoch_error("cleanup journal read cardinality changed"));
            }
            match values.into_iter().next().flatten() {
                Some(ProjectedValue::FullValue(raw)) => raw,
                _ => return Err(epoch_error("cleanup journal payload disappeared")),
            }
        };
        if raw.len() > 1024 * 1024 {
            return Err(epoch_error("cleanup journal exceeds fixed payload bound"));
        }
        let mut journal: GlobalConversionJournal = serde_json::from_slice(&raw)
            .map_err(|_| epoch_error("cleanup journal is malformed"))?;
        journal.validate()?;
        // Do not submit another account/repository's private source coordinates.
        if global_journal_key(
            &journal.source_bank,
            expected.repository_id(),
            expected.active_account_id(),
        )? != key
        {
            continue;
        }
        if journal.cleanup_complete || journal.receipt.is_none() {
            continue;
        }
        // No read transaction survives this network await. The authority validates
        // the exact immutable M receipt and guards the surviving root atomically.
        crate::sync::cleanup_global_conversion_authenticated(authenticated, &journal.request)
            .await?;
        journal.cleanup_complete = true;
        persist_global_conversion_journal(
            storage,
            &active,
            expected.repository_id(),
            expected.active_account_id(),
            &journal,
            Some(raw),
        )
        .await?;
        completed += 1;
    }
    Ok(completed)
}