graphforge_knowledge/
retention_dependency.rs

1//! Required-local versus outside-reference retention pins (#1349).
2
3use std::collections::HashSet;
4use std::sync::{Arc, LazyLock};
5
6use arrow::array::{FixedSizeBinaryBuilder, StringArray, TimestampMicrosecondArray, UInt32Array};
7use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit};
8use arrow::record_batch::RecordBatch;
9use graphforge_core::canonical::{CANONICAL_CONTRACT_VERSION, CanonicalDomain, fingerprint};
10use uuid::Uuid;
11
12use crate::{
13    KNOWLEDGE_CAPABILITY_VERSION, KnowledgeError, MAX_KNOWLEDGE_ROWS, SchemaRegistryEntry,
14    check_limit, fixed_column, invalid, require_schema, require_uuid, require_v7, required_i64,
15    required_text, required_u32, string_column, timestamp_column, u32_column, uuid_at, uuid_field,
16};
17
18/// Immutable retention-dependency record contract.
19pub const RETENTION_DEPENDENCY_CONTRACT_VERSION: u32 = 1;
20/// Closed required-subject-kind registry version.
21pub const RETENTION_REQUIRED_KIND_REGISTRY_VERSION: u32 = 1;
22/// Closed dependency-class registry version.
23pub const RETENTION_DEPENDENCY_CLASS_REGISTRY_VERSION: u32 = 1;
24
25/// Authoritative `knowledge/retention_dependencies.parquet` schema.
26pub static RETENTION_DEPENDENCY_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(|| {
27    Arc::new(Schema::new(vec![
28        uuid_field("dependency_uuid", false),
29        uuid_field("scope_uuid", false),
30        uuid_field("required_uuid", false),
31        Field::new("required_kind", DataType::Utf8, false),
32        Field::new("dependency_class", DataType::Utf8, false),
33        uuid_field("provenance_uuid", false),
34        Field::new(
35            "recorded_at",
36            DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
37            false,
38        ),
39        Field::new("contract_version", DataType::UInt32, false),
40    ]))
41});
42
43static RETENTION_DEPENDENCY_SCHEMA_FINGERPRINT: LazyLock<[u8; 32]> = LazyLock::new(|| {
44    fingerprint(
45        CanonicalDomain::Schema,
46        CANONICAL_CONTRACT_VERSION,
47        b"retention_dependency/1|dependency_uuid:fixed[16]:required|scope_uuid:fixed[16]:required|required_uuid:fixed[16]:required|required_kind:utf8:required|dependency_class:utf8:required|provenance_uuid:fixed[16]:required|recorded_at:timestamp_us_utc:required|contract_version:u32:required",
48    )
49    .expect("registered retention dependency schema is within canonical bounds")
50});
51
52/// Closed subject kind for one retention dependency endpoint.
53#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)]
54#[serde(rename_all = "snake_case")]
55pub enum RetentionRequiredKind {
56    /// Immutable research Artifact.
57    Artifact,
58    /// Evidence-link UUID.
59    EvidenceLink,
60    /// Immutable research Source.
61    Source,
62    /// Immutable assertion UUID.
63    Assertion,
64}
65
66impl RetentionRequiredKind {
67    /// Canonical persisted spelling.
68    #[must_use]
69    pub const fn as_str(self) -> &'static str {
70        match self {
71            Self::Artifact => "artifact",
72            Self::EvidenceLink => "evidence_link",
73            Self::Source => "source",
74            Self::Assertion => "assertion",
75        }
76    }
77
78    pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
79        match value {
80            "artifact" => Ok(Self::Artifact),
81            "evidence_link" => Ok(Self::EvidenceLink),
82            "source" => Ok(Self::Source),
83            "assertion" => Ok(Self::Assertion),
84            _ => Err(invalid("retention_required_kind", "unknown closed value")),
85        }
86    }
87}
88
89/// Closed retention dependency class.
90#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, serde::Serialize, serde::Deserialize)]
91#[serde(rename_all = "snake_case")]
92pub enum RetentionDependencyClass {
93    /// Locally retained bytes required for the scoped selection.
94    RequiredLocalBytes,
95    /// Historical outside reference without local substitution.
96    OutsideReference,
97    /// Comparison baseline retained for analyst review.
98    ComparisonBaseline,
99}
100
101impl RetentionDependencyClass {
102    /// Canonical persisted spelling.
103    #[must_use]
104    pub const fn as_str(self) -> &'static str {
105        match self {
106            Self::RequiredLocalBytes => "required_local_bytes",
107            Self::OutsideReference => "outside_reference",
108            Self::ComparisonBaseline => "comparison_baseline",
109        }
110    }
111
112    pub(super) fn parse(value: &str) -> Result<Self, KnowledgeError> {
113        match value {
114            "required_local_bytes" => Ok(Self::RequiredLocalBytes),
115            "outside_reference" => Ok(Self::OutsideReference),
116            "comparison_baseline" => Ok(Self::ComparisonBaseline),
117            _ => Err(invalid("dependency_class", "unknown closed value")),
118        }
119    }
120}
121
122/// One explicit retention dependency pin.
123#[derive(Clone, Debug, Eq, PartialEq)]
124pub struct RetentionDependency {
125    /// Caller-supplied UUIDv7 identity and idempotency key.
126    pub dependency_uuid: Uuid,
127    /// Selection or retention root UUID.
128    pub scope_uuid: Uuid,
129    /// Required subject UUID.
130    pub required_uuid: Uuid,
131    /// Closed required subject kind.
132    pub required_kind: RetentionRequiredKind,
133    /// Closed dependency class.
134    pub dependency_class: RetentionDependencyClass,
135    /// Provenance event that published the dependency.
136    pub provenance_uuid: Uuid,
137    /// Durable transaction time.
138    pub recorded_at_micros: i64,
139    /// Frozen record contract.
140    pub contract_version: u32,
141}
142
143impl RetentionDependency {
144    /// Construct one validated retention dependency.
145    pub fn new(
146        dependency_uuid: Uuid,
147        scope_uuid: Uuid,
148        required_uuid: Uuid,
149        required_kind: RetentionRequiredKind,
150        dependency_class: RetentionDependencyClass,
151        provenance_uuid: Uuid,
152        recorded_at_micros: i64,
153    ) -> Result<Self, KnowledgeError> {
154        let row = Self {
155            dependency_uuid,
156            scope_uuid,
157            required_uuid,
158            required_kind,
159            dependency_class,
160            provenance_uuid,
161            recorded_at_micros,
162            contract_version: RETENTION_DEPENDENCY_CONTRACT_VERSION,
163        };
164        validate_dependency(&row)?;
165        Ok(row)
166    }
167}
168
169/// Validated retention-dependency table.
170#[derive(Clone, Debug, Default, Eq, PartialEq)]
171pub struct RetentionDependencyLedger {
172    /// Dependencies ordered by `(recorded_at, dependency_uuid)`.
173    pub dependencies: Vec<RetentionDependency>,
174}
175
176impl RetentionDependencyLedger {
177    /// Validate, sort, and construct a dependency table.
178    pub fn new(mut dependencies: Vec<RetentionDependency>) -> Result<Self, KnowledgeError> {
179        dependencies.sort_by_key(|row| (row.recorded_at_micros, row.dependency_uuid));
180        validate_dependencies(&dependencies)?;
181        Ok(Self { dependencies })
182    }
183
184    /// Merge staged rows into an existing ledger.
185    pub fn merge(&self, staged: &Self) -> Result<Self, KnowledgeError> {
186        let mut merged = self.dependencies.clone();
187        for row in &staged.dependencies {
188            if let Some(existing) = merged
189                .iter()
190                .find(|candidate| candidate.dependency_uuid == row.dependency_uuid)
191            {
192                if existing != row {
193                    return Err(KnowledgeError::Conflict("dependency_uuid"));
194                }
195                continue;
196            }
197            merged.push(row.clone());
198        }
199        Self::new(merged)
200    }
201
202    /// Encode the authoritative Arrow batch.
203    pub fn batch(&self) -> Result<RecordBatch, KnowledgeError> {
204        dependency_batch(&self.dependencies)
205    }
206
207    /// Decode one or more Arrow batches.
208    pub fn from_batches(batches: &[RecordBatch]) -> Result<Self, KnowledgeError> {
209        let mut dependencies = Vec::new();
210        for batch in batches {
211            require_schema(
212                batch,
213                &RETENTION_DEPENDENCY_SCHEMA,
214                "retention_dependencies.schema",
215            )?;
216            let ids = fixed_column(batch, "dependency_uuid")?;
217            let scopes = fixed_column(batch, "scope_uuid")?;
218            let required = fixed_column(batch, "required_uuid")?;
219            let kinds = string_column(batch, "required_kind")?;
220            let classes = string_column(batch, "dependency_class")?;
221            let provenance = fixed_column(batch, "provenance_uuid")?;
222            let recorded = timestamp_column(batch, "recorded_at")?;
223            let versions = u32_column(batch, "contract_version")?;
224            for row in 0..batch.num_rows() {
225                dependencies.push(RetentionDependency {
226                    dependency_uuid: uuid_at(ids, row, "dependency_uuid")?,
227                    scope_uuid: uuid_at(scopes, row, "scope_uuid")?,
228                    required_uuid: uuid_at(required, row, "required_uuid")?,
229                    required_kind: RetentionRequiredKind::parse(required_text(
230                        kinds,
231                        row,
232                        "required_kind",
233                    )?)?,
234                    dependency_class: RetentionDependencyClass::parse(required_text(
235                        classes,
236                        row,
237                        "dependency_class",
238                    )?)?,
239                    provenance_uuid: uuid_at(provenance, row, "provenance_uuid")?,
240                    recorded_at_micros: required_i64(recorded, row, "recorded_at")?,
241                    contract_version: required_u32(versions, row, "contract_version")?,
242                });
243            }
244        }
245        Self::new(dependencies)
246    }
247}
248
249pub(crate) fn schema_registry_entry() -> SchemaRegistryEntry {
250    SchemaRegistryEntry {
251        capability_id: "knowledge",
252        capability_version: KNOWLEDGE_CAPABILITY_VERSION,
253        record_family: "retention_dependencies",
254        record_version: RETENTION_DEPENDENCY_CONTRACT_VERSION,
255        schema: Arc::clone(&RETENTION_DEPENDENCY_SCHEMA),
256        schema_fingerprint: *RETENTION_DEPENDENCY_SCHEMA_FINGERPRINT,
257        enum_registry_versions: &[
258            ("required_kind", RETENTION_REQUIRED_KIND_REGISTRY_VERSION),
259            (
260                "dependency_class",
261                RETENTION_DEPENDENCY_CLASS_REGISTRY_VERSION,
262            ),
263        ],
264        sort_key: &["recorded_at", "dependency_uuid"],
265        diff_identity_fields: &["dependency_uuid"],
266        diff_record_uuid_field: Some("dependency_uuid"),
267        fingerprint_domain: CanonicalDomain::RetentionDependency,
268        owner: "graphforge-knowledge",
269        implementation_issue: 1349,
270        max_rows: MAX_KNOWLEDGE_ROWS,
271    }
272}
273
274fn validate_dependency(row: &RetentionDependency) -> Result<(), KnowledgeError> {
275    if row.contract_version != RETENTION_DEPENDENCY_CONTRACT_VERSION {
276        return Err(invalid(
277            "dependency.contract_version",
278            "unsupported version",
279        ));
280    }
281    require_v7(row.dependency_uuid, "dependency_uuid")?;
282    require_uuid(row.scope_uuid, "scope_uuid")?;
283    require_uuid(row.required_uuid, "required_uuid")?;
284    require_uuid(row.provenance_uuid, "provenance_uuid")?;
285    Ok(())
286}
287
288fn validate_dependencies(dependencies: &[RetentionDependency]) -> Result<(), KnowledgeError> {
289    check_limit("retention_dependencies", dependencies.len())?;
290    let mut ids = HashSet::with_capacity(dependencies.len());
291    for row in dependencies {
292        validate_dependency(row)?;
293        if !ids.insert(row.dependency_uuid) {
294            return Err(KnowledgeError::Duplicate("dependency_uuid"));
295        }
296    }
297    Ok(())
298}
299
300fn dependency_batch(rows: &[RetentionDependency]) -> Result<RecordBatch, KnowledgeError> {
301    let mut ids = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
302    let mut scopes = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
303    let mut required = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
304    let mut provenance = FixedSizeBinaryBuilder::with_capacity(rows.len(), 16);
305    for row in rows {
306        ids.append_value(row.dependency_uuid.as_bytes())?;
307        scopes.append_value(row.scope_uuid.as_bytes())?;
308        required.append_value(row.required_uuid.as_bytes())?;
309        provenance.append_value(row.provenance_uuid.as_bytes())?;
310    }
311    RecordBatch::try_new(
312        Arc::clone(&RETENTION_DEPENDENCY_SCHEMA),
313        vec![
314            Arc::new(ids.finish()),
315            Arc::new(scopes.finish()),
316            Arc::new(required.finish()),
317            Arc::new(StringArray::from_iter_values(
318                rows.iter().map(|row| row.required_kind.as_str()),
319            )),
320            Arc::new(StringArray::from_iter_values(
321                rows.iter().map(|row| row.dependency_class.as_str()),
322            )),
323            Arc::new(provenance.finish()),
324            Arc::new(
325                TimestampMicrosecondArray::from_iter_values(
326                    rows.iter().map(|row| row.recorded_at_micros),
327                )
328                .with_timezone("UTC"),
329            ),
330            Arc::new(UInt32Array::from_iter_values(
331                rows.iter().map(|row| row.contract_version),
332            )),
333        ],
334    )
335    .map_err(Into::into)
336}
337
338#[cfg(test)]
339mod tests {
340    use super::*;
341    use crate::tests::uuid7;
342
343    #[test]
344    fn retention_dependency_round_trip() {
345        let local = RetentionDependency::new(
346            uuid7(1),
347            uuid7(2),
348            uuid7(3),
349            RetentionRequiredKind::Artifact,
350            RetentionDependencyClass::RequiredLocalBytes,
351            uuid7(4),
352            100,
353        )
354        .unwrap();
355        let outside = RetentionDependency::new(
356            uuid7(5),
357            uuid7(2),
358            uuid7(6),
359            RetentionRequiredKind::Artifact,
360            RetentionDependencyClass::OutsideReference,
361            uuid7(7),
362            200,
363        )
364        .unwrap();
365        let ledger = RetentionDependencyLedger::new(vec![local, outside]).unwrap();
366        let reopened = RetentionDependencyLedger::from_batches(&[ledger.batch().unwrap()]).unwrap();
367        assert_eq!(reopened, ledger);
368    }
369
370    #[test]
371    fn retention_registries_round_trip_and_reject_unknown_tokens() {
372        for kind in [
373            RetentionRequiredKind::Artifact,
374            RetentionRequiredKind::EvidenceLink,
375            RetentionRequiredKind::Source,
376            RetentionRequiredKind::Assertion,
377        ] {
378            assert_eq!(RetentionRequiredKind::parse(kind.as_str()).unwrap(), kind);
379        }
380        for class in [
381            RetentionDependencyClass::RequiredLocalBytes,
382            RetentionDependencyClass::OutsideReference,
383            RetentionDependencyClass::ComparisonBaseline,
384        ] {
385            assert_eq!(
386                RetentionDependencyClass::parse(class.as_str()).unwrap(),
387                class
388            );
389        }
390        assert!(RetentionRequiredKind::parse("node").is_err());
391        assert!(RetentionDependencyClass::parse("genealogy").is_err());
392    }
393}
graphforge_knowledge/retention_dependency.rs

graphforge_knowledge/
retention_dependency.rs
