zeppelin_embed/lifecycle/
cascade.rs

1//! Root-scoped cascade declarations, serialized with the namespace coordinator.
2use super::namespace_batch::{
3    body, durable_write, envelope, invalid, io, name_valid, read_optional, sync_dir,
4};
5use super::{
6    CancelToken, DocumentFields, DocumentScanRequest, NamespaceMutation, QueryControl, Store,
7    StoreError,
8};
9use crate::ingest::DocId;
10use crate::meta::{ColumnId, ColumnType, PredicateValue};
11use crate::vfs::{StdVfs, Vfs};
12use std::collections::{BTreeMap, BTreeSet, VecDeque};
13use std::path::Path;
14
15const RECORD: &str = ".ze-cascades";
16const VERSION: &str = "ZECASCADE1\n";
17
18/// A child namespace's ownership reference to a parent namespace.
19#[derive(Clone, Debug, Eq, PartialEq, Ord, PartialOrd)]
20pub struct CascadeRule {
21    /// Parent namespace in the same root.
22    pub parent: String,
23    /// Child namespace in the same root.
24    pub child: String,
25    /// Child's id128 attribute holding the parent document ID.
26    pub attribute: ColumnId,
27}
28
29fn encode(rules: &BTreeSet<CascadeRule>) -> Vec<u8> {
30    let mut text = VERSION.to_owned();
31    for rule in rules {
32        text.push_str(&format!(
33            "{}\t{}\t{}\n",
34            rule.parent,
35            rule.child,
36            rule.attribute.get()
37        ));
38    }
39    envelope(text.as_bytes())
40}
41fn read(root: &Path) -> Result<BTreeSet<CascadeRule>, StoreError> {
42    let path = root.join(RECORD);
43    let Some(bytes) = read_optional(&path)? else {
44        return Ok(BTreeSet::new());
45    };
46    let text =
47        std::str::from_utf8(body(&path, &bytes)?).map_err(|_| invalid(&path, "cascade UTF-8"))?;
48    let text = text
49        .strip_prefix(VERSION)
50        .ok_or_else(|| invalid(&path, "cascade version"))?;
51    let mut rules = BTreeSet::new();
52    for line in text.lines() {
53        let mut fields = line.split('\t');
54        let parent = fields
55            .next()
56            .ok_or_else(|| invalid(&path, "cascade parent"))?;
57        let child = fields
58            .next()
59            .ok_or_else(|| invalid(&path, "cascade child"))?;
60        let attribute = fields
61            .next()
62            .and_then(|s| s.parse::<u32>().ok())
63            .ok_or_else(|| invalid(&path, "cascade attribute"))?;
64        if fields.next().is_some() || !name_valid(parent) || !name_valid(child) {
65            return Err(invalid(&path, "invalid cascade rule"));
66        }
67        let rule = CascadeRule {
68            parent: parent.into(),
69            child: child.into(),
70            attribute: ColumnId::new(attribute),
71        };
72        reject_cycle(&rules, &rule)?;
73        if !rules.insert(rule) {
74            return Err(invalid(&path, "duplicate cascade rule"));
75        }
76    }
77    if encode(&rules) != bytes {
78        return Err(invalid(&path, "noncanonical cascade record"));
79    }
80    Ok(rules)
81}
82fn reject_cycle(rules: &BTreeSet<CascadeRule>, new: &CascadeRule) -> Result<(), StoreError> {
83    let mut queue = VecDeque::from([(
84        new.child.clone(),
85        vec![new.parent.clone(), new.child.clone()],
86    )]);
87    let mut visited = BTreeSet::new();
88    while let Some((name, path)) = queue.pop_front() {
89        if name == new.parent {
90            return Err(StoreError::CascadeCycle { cycle: path });
91        }
92        if !visited.insert(name.clone()) {
93            continue;
94        }
95        for rule in rules.iter().filter(|r| r.parent == name) {
96            let mut path = path.clone();
97            path.push(rule.child.clone());
98            queue.push_back((rule.child.clone(), path));
99        }
100    }
101    Ok(())
102}
103fn validate(
104    root: &Path,
105    sources: &BTreeMap<String, Store>,
106    rule: &CascadeRule,
107) -> Result<(), StoreError> {
108    if !sources.contains_key(&rule.parent) {
109        return Err(invalid(
110            root,
111            &format!("missing cascade participant {}", rule.parent),
112        ));
113    }
114    let child = sources
115        .get(&rule.child)
116        .ok_or_else(|| invalid(root, &format!("missing cascade participant {}", rule.child)))?;
117    if child
118        .schema()
119        .column(rule.attribute)
120        .map(|c| c.column_type())
121        != Some(ColumnType::Id128)
122    {
123        return Err(invalid(
124            root,
125            "cascade attribute must be a declared id128 column",
126        ));
127    }
128    Ok(())
129}
130
131pub(super) fn declare(
132    root: &Path,
133    sources: &BTreeMap<String, Store>,
134    rule: CascadeRule,
135    step: &mut dyn FnMut(&str) -> std::io::Result<()>,
136) -> Result<(), StoreError> {
137    validate(root, sources, &rule)?;
138    let mut rules = read(root)?;
139    reject_cycle(&rules, &rule)?;
140    if !rules.insert(rule) {
141        return Ok(());
142    }
143    let bytes = encode(&rules);
144    if bytes.len() > 1024 * 1024 {
145        return Err(invalid(root, "cascade record too large"));
146    }
147    let temporary = root.join(".ze-cascades.tmp");
148    durable_write(&StdVfs, &temporary, &bytes, step)?;
149    StdVfs
150        .rename(&temporary, &root.join(RECORD))
151        .map_err(|e| io(root, e))?;
152    step("cascade declaration rename").map_err(|e| io(root, e))?;
153    sync_dir(&StdVfs, root, step)
154}
155
156pub(super) fn expand(
157    root: &Path,
158    sources: &BTreeMap<String, Store>,
159    mutations: &mut [NamespaceMutation],
160) -> Result<(), StoreError> {
161    let rules = read(root)?;
162    // Require the entire declared closure even when a parent currently has no
163    // matching rows. Missing namespaces must never silently disable a rule.
164    let mut reachable: BTreeSet<String> = sources.keys().cloned().collect();
165    loop {
166        let before = reachable.len();
167        for rule in &rules {
168            if reachable.contains(&rule.parent) {
169                reachable.insert(rule.child.clone());
170            }
171        }
172        if before == reachable.len() {
173            break;
174        }
175    }
176    for name in &reachable {
177        if !sources.contains_key(name) {
178            return Err(invalid(
179                root,
180                &format!("missing cascade participant {name}"),
181            ));
182        }
183    }
184    let active: Vec<_> = rules
185        .iter()
186        .filter(|r| reachable.contains(&r.parent))
187        .collect();
188    for rule in &active {
189        validate(root, sources, rule)?;
190    }
191    let mut ids: BTreeMap<String, BTreeSet<DocId>> = mutations
192        .iter()
193        .map(|m| (m.name.clone(), m.deletes.iter().copied().collect()))
194        .collect();
195    // Read references once while all logical writer locks are held. Null
196    // attributes have no parent; shared children are deleted if any owner is.
197    let mut references = Vec::new();
198    for (name, source) in sources {
199        let child_rules: Vec<_> = active.iter().filter(|r| r.child == *name).collect();
200        if child_rules.is_empty() {
201            continue;
202        }
203        let mut cursor = None;
204        loop {
205            let mut request = DocumentScanRequest::new(
206                1024,
207                DocumentFields::ATTRIBUTES,
208                QueryControl::Cancel(CancelToken::new()),
209            );
210            if let Some(cursor) = cursor {
211                request = request.with_cursor(cursor);
212            }
213            let page = source
214                .scan_documents(request)
215                .map_err(|e| invalid(root, &e.to_string()))?;
216            for document in page.documents {
217                for (column, value) in document.attributes.into_iter().flatten() {
218                    for rule in child_rules.iter().filter(|r| r.attribute == column) {
219                        let PredicateValue::Id128(parent_id) = value else {
220                            return Err(invalid(root, "cascade reference is not id128"));
221                        };
222                        references.push((
223                            rule.parent.clone(),
224                            parent_id,
225                            name.clone(),
226                            document.doc_id,
227                        ));
228                    }
229                }
230            }
231            cursor = page.continuation;
232            if cursor.is_none() {
233                break;
234            }
235        }
236    }
237    loop {
238        let mut changed = false;
239        for (parent, parent_id, child, child_id) in &references {
240            if ids.get(parent).is_some_and(|set| set.contains(parent_id)) {
241                let set = ids
242                    .get_mut(child)
243                    .ok_or_else(|| invalid(root, "missing child ID set"))?;
244                changed |= set.insert(*child_id);
245            }
246        }
247        if !changed {
248            break;
249        }
250    }
251    for mutation in mutations {
252        mutation.deletes = ids
253            .remove(&mutation.name)
254            .ok_or_else(|| invalid(root, "missing cascade ID set"))?
255            .into_iter()
256            .collect();
257    }
258    Ok(())
259}
zeppelin_embed/lifecycle/cascade.rs

zeppelin_embed/lifecycle/
cascade.rs
