minikv 0.7.0

A production-grade distributed KV store with Raft consensus in ~1,500 lines
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

//! Volume gRPC service implementation
//!
//! This module exposes the internal gRPC API for volume operations.
//! Security features (TLS, authentication) and cross-datacenter replication are planned for future releases.

use crate::proto::volume_internal_server::{VolumeInternal, VolumeInternalServer};
use crate::proto::*;
use crate::volume::blob::BlobStore;
use std::sync::{Arc, Mutex};
use tonic::{Request, Response, Status};

pub struct VolumeGrpcService {
    store: Arc<Mutex<BlobStore>>,
}

impl VolumeGrpcService {
    pub fn new(store: BlobStore) -> Self {
        VolumeGrpcService {
            store: Arc::new(Mutex::new(store)),
        }
    }

    pub fn into_server(self) -> VolumeInternalServer<Self> {
        VolumeInternalServer::new(self)
    }
}

#[tonic::async_trait]
impl VolumeInternal for VolumeGrpcService {
    async fn prepare(
        &self,
        req: Request<PrepareRequest>,
    ) -> Result<Response<PrepareResponse>, Status> {
        let inner = req.into_inner();

        // Validate request
        if inner.key.is_empty() {
            return Ok(Response::new(PrepareResponse {
                ok: false,
                error: "key cannot be empty".to_string(),
            }));
        }

        // Check if we have space (simplified check)
        // In production: check disk space, quotas, etc.

        Ok(Response::new(PrepareResponse {
            ok: true,
            error: String::new(),
        }))
    }

    async fn commit(
        &self,
        req: Request<CommitRequest>,
    ) -> Result<Response<CommitResponse>, Status> {
        let _inner = req.into_inner();

        // CI trigger: commit for relaunch

        // For now, we just acknowledge
        // In production: finalize the transaction, make data durable

        Ok(Response::new(CommitResponse {
            ok: true,
            error: String::new(),
        }))
    }

    async fn abort(&self, req: Request<AbortRequest>) -> Result<Response<AbortResponse>, Status> {
        let _inner = req.into_inner();

        // Clean up any prepared state
        // In production: delete temp files, release locks

        Ok(Response::new(AbortResponse { ok: true }))
    }

    async fn pull(&self, _req: Request<PullRequest>) -> Result<Response<Self::PullStream>, Status> {
        Err(Status::unimplemented("Pull not implemented"))
    }

    async fn delete(
        &self,
        req: Request<DeleteRequest>,
    ) -> Result<Response<DeleteResponse>, Status> {
        let inner = req.into_inner();

        match self.store.lock().unwrap().delete(&inner.key) {
            Ok(_) => Ok(Response::new(DeleteResponse {
                ok: true,
                error: String::new(),
            })),
            Err(e) => Ok(Response::new(DeleteResponse {
                ok: false,
                error: e.to_string(),
            })),
        }
    }

    async fn ping(&self, _req: Request<PingRequest>) -> Result<Response<PingResponse>, Status> {
        Ok(Response::new(PingResponse {
            volume_id: "vol-1".to_string(),
            uptime_secs: 0,
            total_keys: 0,
            total_bytes: 0,
        }))
    }

    async fn stats(&self, _req: Request<StatsRequest>) -> Result<Response<StatsResponse>, Status> {
        Ok(Response::new(StatsResponse {
            total_keys: 0,
            total_bytes: 0,
            free_bytes: 0,
            shards: vec![],
        }))
    }

    type PullStream = tokio_stream::wrappers::ReceiverStream<Result<Chunk, Status>>;
}