Method rewrite_session_turns

atomic-core/src/pristine/txn/write/mod.rs:448–521

Rewrite one session's derived index from its complete immutable turn set. This keeps local keys, manifest order, lifecycle pointers, and KG turn numbers consistent even when provenance arrives out of order.

(
        &mut self,
        mut record: crate::change::session::SessionRecord,
        turns: Vec<crate::change::session::SessionTurn>,
    )

    /// set. This keeps local keys, manifest order, lifecycle pointers, and KG
    /// turn numbers consistent even when provenance arrives out of order.
    fn rewrite_session_turns(
        &mut self,
        mut record: crate::change::session::SessionRecord,
        turns: Vec<crate::change::session::SessionTurn>,
    ) -> PristineResult<()> {
        use crate::change::session::{
            canonicalize_session_turns, encode_session_turn_key, session_turn_namespace,
        };

        let old_turns = self.load_session_turns_for_write(&record.session_id)?;
        let turns = canonicalize_session_turns(turns);
        let preserved_edges = self.clear_session_turn_kg(&old_turns, &turns)?;

        let namespace = session_turn_namespace(&record.session_id);
        let scan_start = encode_session_turn_key(namespace, 0);
        let scan_end = encode_session_turn_key(namespace, u32::MAX);
        let existing_keys = {
            let table = self.txn.open_table(SESSION_TURNS)?;
            let mut keys = Vec::new();
            for row in table.range::<&[u8; 40]>(&scan_start..=&scan_end)? {
                let (key, _value) = row?;
                keys.push(*key.value());
            }
            keys
        };
        {
            let mut table = self.txn.open_table(SESSION_TURNS)?;
            for key in existing_keys {
                table.remove(&key)?;
            }
        }
        {
            let mut table = self.txn.open_table(SESSION_TURNS)?;
            let mut reverse = self.txn.open_table(SESSION_PROVENANCE)?;
            for turn in &turns {
                let key = encode_session_turn_key(namespace, turn.turn_number);
                let bytes = turn.to_bytes();
                table.insert(&key, bytes.as_slice())?;
                reverse.insert(turn.provenance_hash.as_bytes(), &key)?;
            }
        }

        record.first_provenance = turns.first().map(|turn| turn.provenance_hash);
        record.latest_provenance = turns.last().map(|turn| turn.provenance_hash);
        record.turn_count = turns.len() as u32;
        if let Some(first) = turns.first() {
            record.started_at = record.started_at.min(first.timestamp);
        }
        {
            let mut sessions = self.txn.open_table(SESSIONS)?;
            let bytes = record.to_bytes();
            sessions.insert(record.session_id.as_str(), bytes.as_slice())?;
        }

        let turn_numbers: std::collections::HashMap<Hash, u32> = turns
            .iter()
            .map(|turn| (turn.provenance_hash, turn.turn_number))
            .collect();