rkvs 0.4.0

A high-performance, namespace-based key-value storage system with async operations and batch processing
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//! Contains all configuration-related methods for the `Namespace`.
use super::{DEFAULT_SHARD_CAPACITY, Namespace, Result, RkvsError};
use crate::data_table::DataTable;
use crate::types::NamespaceConfig;
use std::sync::Arc;
use tokio::sync::RwLock;

impl Namespace {
    /// Resizes the namespace to use a new number of shards.
    /// Only supports increasing shard count
    pub async fn resize_shards(&self, new_shard_count: usize) -> Result<()> {
        let mut shards_guard = self
            .timeout_write_lock(&self.shards, "resize_shards")
            .await?;
        let current_shard_count = shards_guard.len();

        if new_shard_count <= current_shard_count {
            return Err(crate::RkvsError::Storage(
                "New shard count must be greater than current shard count".to_string(),
            ));
        }

        let num_new_shards = new_shard_count - current_shard_count;

        let new_shards: Vec<Arc<RwLock<DataTable>>> = (0..num_new_shards)
            .map(|_| {
                Arc::new(RwLock::new(DataTable::with_capacity(
                    DEFAULT_SHARD_CAPACITY,
                )))
            })
            .collect();

        // Append new shards to existing ones
        shards_guard.extend(new_shards);

        // Now redistribute keys that need to move to new shards
        // We only check the first current_shard_count shards (the old ones)
        for shard_idx in 0..current_shard_count {
            let shard = Arc::clone(&shards_guard[shard_idx]);
            let mut data = self
                .timeout_write_lock(
                    &shard,
                    &format!("data redistribution from shard {}", shard_idx),
                )
                .await?;
            let keys_to_check: Vec<_> = data.all_keys();

            for key in keys_to_check {
                let new_shard_idx = super::helpers::jump_consistent_hash(&key, new_shard_count);

                // If key should move to a different shard
                if new_shard_idx != shard_idx {
                    if let Some(value) = data.delete_value(&key) {
                        // Move to new shard
                        let target_shard = Arc::clone(&shards_guard[new_shard_idx]);
                        let mut target_data = self
                            .timeout_write_lock(
                                &target_shard,
                                &format!("data redistribution to shard {}", new_shard_idx),
                            )
                            .await?;
                        target_data.set_value(key, (*value).clone());
                    }
                }
                // If new_shard_idx == shard_idx, key stays where it is
            }
        }

        self.config.set_shard_count(new_shard_count);

        Ok(())
    }

    /// Gets a clone of the namespace's configuration.
    pub async fn get_config(&self) -> NamespaceConfig {
        self.config.clone()
    }

    /// Sets the maximum number of keys allowed in the namespace.
    pub async fn set_max_keys(&self, value: usize) -> Result<()> {
        let new_max = if value == 0 { usize::MAX } else { value };
        if self.metadata.key_count() > new_max {
            return Err(crate::RkvsError::Storage(format!(
                "Cannot set max_keys to {} when namespace already has {} keys",
                new_max,
                self.metadata.key_count()
            )));
        }
        self.config.set_max_keys(new_max);
        Ok(())
    }

    /// Sets the maximum value size allowed in the namespace.
    pub async fn set_max_value_size(&self, value: usize) -> Result<()> {
        let new_max = if value == 0 { usize::MAX } else { value };
        self.config.set_max_value_size(new_max);
        Ok(())
    }

    /// Sets the lock acquisition timeout for this namespace.
    pub async fn set_lock_timeout(&self, value_ms: usize) -> Result<()> {
        if value_ms == 0 {
            return Err(RkvsError::Storage(
                "Lock timeout must be greater than 0.".to_string(),
            ));
        }
        self.config.set_lock_timeout(value_ms);
        Ok(())
    }

    /// Updates multiple config fields at once.
    pub async fn update_config(&self, new_config: NamespaceConfig) -> Result<()> {
        self.set_max_keys(new_config.max_keys()).await?;
        self.set_max_value_size(new_config.max_value_size()).await?;
        self.set_lock_timeout(
            new_config
                .lock_timeout
                .load(std::sync::atomic::Ordering::SeqCst),
        )
        .await?;
        Ok(())
    }
}