qualia_core_db/
sync.rs

1// no_std is declared at the crate level if needed
2
3use crate::NQuin;
4
5/// Represents a 12-byte structural pointer used for $O(N)$ Merkle DAG diffing.
6#[repr(C, packed)]
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct JumpTableEntry {
9    pub chunk_id: u64,
10    pub byte_offset: u32,
11}
12
13/// The Zero-Allocation CRDT Synchronizer for the Qualia Engine.
14/// Operates strictly within pre-allocated stack limits for offline-first mesh syncs.
15pub struct MerkleCrdtSynchronizer;
16
17impl MerkleCrdtSynchronizer {
18    /// Extracts the 12-bit Lamport clock integer from the 5th Vector (Metadata).
19    /// The Lamport clock is embedded in bits 32-43.
20    #[inline(always)]
21    pub fn extract_lamport_time(quin: &NQuin) -> u16 {
22        ((quin.metadata >> 32) & 0x0FFF) as u16
23    }
24
25    /// Compares two arrays of JumpTableEntry to find the divergent 128KB frames.
26    /// Returns a stack-allocated slice of byte offsets indicating which blocks to pull.
27    /// Operates in exactly $O(N)$ time with zero heap allocation.
28    pub fn diff_jump_tables<'a>(
29        local_table: &[JumpTableEntry],
30        remote_table: &[JumpTableEntry],
31        diff_buffer: &'a mut [u32],
32    ) -> &'a [u32] {
33        let mut local_idx = 0;
34        let mut remote_idx = 0;
35        let mut diff_count = 0;
36
37        // $O(N)$ linear scan comparing Virtual Chunk IDs (Merkle Hashes)
38        while local_idx < local_table.len()
39            && remote_idx < remote_table.len()
40            && diff_count < diff_buffer.len()
41        {
42            let local_chunk = local_table[local_idx].chunk_id;
43            let remote_chunk = remote_table[remote_idx].chunk_id;
44
45            if local_chunk == remote_chunk {
46                // Frames match, advance both pointers
47                local_idx += 1;
48                remote_idx += 1;
49            } else if local_chunk < remote_chunk {
50                local_idx += 1;
51            } else {
52                // Remote has a divergent frame
53                diff_buffer[diff_count] = remote_table[remote_idx].byte_offset;
54                diff_count += 1;
55                remote_idx += 1;
56            }
57        }
58
59        // Add any remaining remote frames
60        while remote_idx < remote_table.len() && diff_count < diff_buffer.len() {
61            diff_buffer[diff_count] = remote_table[remote_idx].byte_offset;
62            diff_count += 1;
63            remote_idx += 1;
64        }
65
66        &diff_buffer[..diff_count]
67    }
68
69    /// Resolves structural conflicts between an incoming 128KB divergent frame and the local frame.
70    /// Follows a strict selectable compaction policy triggered by the context metadata.
71    pub fn resolve_frame_conflict(
72        local_frame: &mut [NQuin],
73        incoming_frame: &[NQuin],
74        compaction_policy_metadata: u16,
75    ) {
76        const MASK_STRICT_HISTORY: u16 = 0x0010;
77        const MASK_EPOCH_COMPACT: u16 = 0x0020;
78
79        if (compaction_policy_metadata & MASK_EPOCH_COMPACT) != 0 {
80            // Epoch Compaction: Zero-out Tombstone Quins to shrink the active data footprint
81            for incoming in incoming_frame.iter() {
82                // In an epoch compact, we look for tombstones (e.g. Quins with metadata flags marking deletion)
83                let is_tombstone = (incoming.metadata & 0x1) != 0; // Simulated tombstone flag
84                if is_tombstone {
85                    for local in local_frame.iter_mut() {
86                        if local.subject == incoming.subject
87                            && local.predicate == incoming.predicate
88                            && local.object == incoming.object
89                        {
90                            // Match found. Zero-out both to compress the dataset.
91                            local.subject = 0;
92                            local.predicate = 0;
93                            local.object = 0;
94                            local.context = 0;
95                            local.metadata = 0;
96                            local.parity = 0;
97                            break;
98                        }
99                    }
100                } else {
101                    // Regular merge logic via Lamport clock comparison
102                    // Find empty slot or matching subject/predicate
103                    for local in local_frame.iter_mut() {
104                        if local.subject == incoming.subject
105                            && local.predicate == incoming.predicate
106                        {
107                            let local_time = Self::extract_lamport_time(local);
108                            let incoming_time = Self::extract_lamport_time(incoming);
109
110                            // Keep the Quin with the higher Lamport clock
111                            if incoming_time > local_time {
112                                *local = *incoming;
113                            }
114                            break;
115                        } else if local.subject == 0 {
116                            // Empty slot found, insert
117                            *local = *incoming;
118                            break;
119                        }
120                    }
121                }
122            }
123        } else if (compaction_policy_metadata & MASK_STRICT_HISTORY) != 0 {
124            // Strict History: Append Tombstone Quins (never delete)
125            for incoming in incoming_frame.iter() {
126                // Find empty slot in local frame to append to
127                for local in local_frame.iter_mut() {
128                    if local.subject == 0 {
129                        // 0 denotes empty slot
130                        *local = *incoming;
131                        break;
132                    }
133                }
134            }
135        }
136    }
137}
qualia_core_db/sync.rs

qualia_core_db/
sync.rs
