<commit_before>use std::cmp;
use std::mem;
use std::fmt;
use std::collections::BTreeMap;
use mio;

use lmdb_rs::{self, EnvBuilder, Environment, DbFlags, MdbError};
use tempdir::TempDir;
use byteorder::{ByteOrder, BigEndian};
use spki_sexp;

use super::{Operation,OpResp};
use event_handler::EventHandler;

const REPLICATION_CREDIT : u64 = 1024;

#[derive(Debug)]
struct Forwarder {
    last_sent_downstream: Option<u64>,
    last_acked_downstream: Option<u64>,
    pending_operations: BTreeMap<u64, mio::Token>,
}

#[derive(Debug)]
struct Terminus {
    state: String,
}

#[derive(Debug)]
enum Role {
    Forwarder(Forwarder),
    Terminus(Terminus),
}

struct Log {
    dir: TempDir,
    env: Environment,
}

impl Log {
    fn new() -> Log {
        let d = TempDir::new("lmdb-log").expect("new log");
        info!("DB path: {:?}", d.path());
        let env = EnvBuilder::new().open(d.path(), 0o777).expect("open db");
        
        Log { dir: d, env: env }
    }

    fn tokey(seq: u64) -> [u8; 8] {
        let mut buf : [u8; 8] = [0, 0, 0, 0, 0, 0, 0, 0];
        BigEndian::write_u64(&mut buf, seq);
        buf
    }

    fn fromkey(key: &[u8]) -> u64 {
        BigEndian::read_u64(&key)
    }



    pub fn seqno(&self) -> u64 {
        let mut dbh = self.env.get_default_db(DbFlags::empty()).expect("open-db");
        let mut rdr = self.env.get_reader().expect("get_reader");
        let mut db = rdr.bind(&dbh);
        let mut cur = db.new_cursor().expect("cursor");
        match cur.to_last() { 
            Ok(()) => {
                let last_key = cur.get_key::<&[u8]>().expect("get_key");
                let current = Self::fromkey(last_key);
                let next = current + 1;
                debug!("Last exisiting current: {:?}; key: {:?}; next: {:?}", last_key, current, next);
                next
            },
            Err(MdbError::NotFound) => 0,
            Err(e) => panic!("Unexpected error getting last item: {:?}", e),
        }
    }

    fn read(&self, seqno: u64) -> Option<Operation> {
        let mut dbh = self.env.get_default_db(DbFlags::empty()).expect("open-db");
        let mut rdr = self.env.get_reader().expect("get_reader");
        let mut db = rdr.bind(&dbh);
        let key = Self::tokey(seqno);
        let ret = match db.get(&key.as_ref()) {
            Ok(val) => Some(spki_sexp::from_bytes(val).expect("decode operation")),
            Err(MdbError::NotFound) => None,
            Err(e) => panic!("Unexpected error reading index: {:?}: {:?}", seqno, e),
        };
        debug!("Read: {:?} => {:?}", seqno, ret);
        ret
    }


    fn verify_sequential(&self, seqno: u64) -> bool {
        let current = self.seqno();
        if !(seqno == 0 || current <= seqno) {
            warn!("Hole in history: saw {}; have: {:?}", seqno, current);
            false
        } else {
            true
        }
    }

    fn insert_at(&mut self, seqno: u64, op: &Operation) {
        let bytes = spki_sexp::as_bytes(op).expect("encode operation");
        let mut dbh = self.env.get_default_db(DbFlags::empty()).expect("open-db");
        let txn = self.env.new_transaction().expect("new_transaction");
        debug!("Checking {:?}", seqno);
        let key = Self::tokey(seqno);
        match txn.bind(&dbh).get::<&[u8]>(&key.as_ref()) {
            Err(MdbError::NotFound) => (),
            Err(e) => panic!("Unexpected error reading index: {:?}: {:?}", seqno, e),
            Ok(_) => panic!("Unexpected entry at seqno: {:?}", seqno),
        };

        txn.bind(&dbh).set(&key.as_ref(), &bytes).expect("Persist operation");
        txn.commit().expect("txn commit");
        debug!("Wrote as {:?}/{:?}: {:?}", seqno, key, bytes.len());
    }<commit_msg>Use seperate data space for metadata.<commit_after>use std::cmp;
use std::mem;
use std::fmt;
use std::collections::BTreeMap;
use mio;

use lmdb_rs::{self, EnvBuilder, Environment, DbFlags, MdbError};
use tempdir::TempDir;
use byteorder::{ByteOrder, BigEndian};
use spki_sexp;

use super::{Operation,OpResp};
use event_handler::EventHandler;

const REPLICATION_CREDIT : u64 = 1024;

#[derive(Debug)]
struct Forwarder {
    last_sent_downstream: Option<u64>,
    last_acked_downstream: Option<u64>,
    pending_operations: BTreeMap<u64, mio::Token>,
}

#[derive(Debug)]
struct Terminus {
    state: String,
}

#[derive(Debug)]
enum Role {
    Forwarder(Forwarder),
    Terminus(Terminus),
}

struct Log {
    dir: TempDir,
    env: Environment,
}

const META : &'static str = "meta";
const DATA : &'static str = "data";
const META_COMMITTED : &'static str = "committed";

impl Log {
    fn new() -> Log {
        let d = TempDir::new("lmdb-log").expect("new log");
        info!("DB path: {:?}", d.path());
        let env = EnvBuilder::new().max_dbs(2).open(d.path(), 0o777).expect("open db");

        Log { dir: d, env: env }
    }

    fn tokey(seq: u64) -> [u8; 8] {
        let mut buf : [u8; 8] = [0, 0, 0, 0, 0, 0, 0, 0];
        BigEndian::write_u64(&mut buf, seq);
        buf
    }

    fn fromkey(key: &[u8]) -> u64 {
        BigEndian::read_u64(&key)
    }

    pub fn seqno(&self) -> u64 {
        let mut data = self.env.create_db(META, DbFlags::empty()).expect("open-db");
        let mut rdr = self.env.get_reader().expect("get_reader");

        match rdr.bind(&data).get(&META_COMMITTED) {
            Ok(val) => {
                let current = Self::fromkey(val);
                let next = current + 1;
                debug!("Last exisiting current: {:?}; next: {:?}", current, next);
                next
            },
            Err(MdbError::NotFound) => 0,
            Err(e) => panic!("Unexpected error getting last item: {:?}", e),
        }
    }

    fn read(&self, seqno: u64) -> Option<Operation> {
        let mut data = self.env.create_db(DATA, DbFlags::empty()).expect("open-db");
        let mut rdr = self.env.get_reader().expect("get_reader");
        let mut db = rdr.bind(&data);
        let key = Self::tokey(seqno);
        let ret = match db.get(&key.as_ref()) {
            Ok(val) => Some(spki_sexp::from_bytes(val).expect("decode operation")),
            Err(MdbError::NotFound) => None,
            Err(e) => panic!("Unexpected error reading index: {:?}: {:?}", seqno, e),
        };
        debug!("Read: {:?} => {:?}", seqno, ret);
        ret
    }


    fn verify_sequential(&self, seqno: u64) -> bool {
        let current = self.seqno();
        if !(seqno == 0 || current <= seqno) {
            warn!("Hole in history: saw {}; have: {:?}", seqno, current);
            false
        } else {
            true
        }
    }

    fn insert_at(&mut self, seqno: u64, op: &Operation) {
        let data_bytes = spki_sexp::as_bytes(op).expect("encode operation");
        let mut data = self.env.create_db(DATA, DbFlags::empty()).expect("open-db");
        let mut meta = self.env.create_db(META, DbFlags::empty()).expect("open-db");
        let txn = self.env.new_transaction().expect("new_transaction");
        debug!("Checking {:?}", seqno);
        let key = Self::tokey(seqno);
        match txn.bind(&data).get::<&[u8]>(&key.as_ref()) {
            Err(MdbError::NotFound) => (),
            Err(e) => panic!("Unexpected error reading index: {:?}: {:?}", seqno, e),
            Ok(_) => panic!("Unexpected entry at seqno: {:?}", seqno),
        };

        txn.bind(&data).set(&key.as_ref(), &data_bytes).expect("Persist operation");
        txn.bind(&meta).set(&META_COMMITTED, &key.as_ref()).expect("Persist operation");
        txn.commit().expect("txn commit");
        debug!("Wrote as {:?}/{:?}: {:?}", seqno, key, data_bytes.len());
    }
{'fraction_non_alphanumeric': 0.18700039313327219, 'fraction_numerical': 0.013104442405975626, 'mean_word_length': 2.588152327221439, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '295561', 'n_tokens_mistral': 2687, 'n_tokens_neox': 2578, 'n_words': 556}