💾 Kawa Storage - 高性能イベントストレージエンジン

kawa-storageは、Kawaプロジェクトの核となる高性能ストレージエンジンです。Write-Ahead Log（WAL）とセグメント化による永続化、イベントソーシングパターンを実装しています。

🎯 設計目標

• ⚡ 高性能: 274K+ writes/sec, 1.2M+ reads/sec
• 🔒 整合性: CRC32チェックサムによるデータ整合性保証
• 📈 スケーラビリティ: セグメント化による効率的なデータ管理
• 🛡️ 信頼性: 障害耐性とデータ復旧機能

🏗️ アーキテクチャ

┌─────────────────────────────────────────────────────────────┐
│                 Kawa Storage Engine                         │
├─────────────────────────────────────────────────────────────┤
│  📤 Public API                                             │
│   ├── StorageEngine                                        │
│   ├── append_event() / read_events()                       │
│   └── get_latest_offset()                                  │
├─────────────────────────────────────────────────────────────┤
│  📝 Write-Ahead Log (WAL)                                  │
│   ├── パーティション別オフセット管理                          │
│   ├── セグメント自動ローテーション                           │
│   └── 非同期書き込み・読み取り                             │
├─────────────────────────────────────────────────────────────┤
│  📦 Segment Management                                      │
│   ├── 固定サイズセグメント（1GB default）                   │
│   ├── メモリマップドI/O                                     │
│   ├── CRC32整合性チェック                                   │
│   └── 並行アクセス制御                                     │
├─────────────────────────────────────────────────────────────┤
│  🔍 Event Management                                        │
│   ├── EventId（UUID v4）                                   │
│   ├── イベントメタデータ                                     │
│   ├── JSON・バイナリデータサポート                          │
│   └── タイムスタンプ管理                                     │
└─────────────────────────────────────────────────────────────┘

🚀 クイックスタート

基本的な使用方法

use kawa_storage::{
    StorageEngine, StorageConfig, 
    Topic, Partition, EventData, Offset
};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // ストレージエンジンを初期化
    let config = StorageConfig {
        data_dir: "./kawa-data".into(),
        segment_size: 1024 * 1024 * 1024, // 1GB
        sync_interval_ms: 1000,
        enable_compression: false,
    };
    
    let storage = StorageEngine::new(config).await?;
    
    // イベントを追加
    let topic = Topic::new("user-events");
    let partition = Partition::new(0);
    let event_data = EventData::from_json(r#"{"user_id": "123", "action": "login"}"#);
    
    let (event_id, offset) = storage.append_event(&topic, partition, event_data).await?;
    println!("Event {} stored at offset {}", event_id, offset);
    
    // イベントを読み取り
    let events = storage.read_events(&topic, partition, Offset::new(0), 10).await?;
    for event in events {
        println!("Event: {:?}", event);
    }
    
    // 最新オフセットを取得
    if let Some(latest) = storage.get_latest_offset(&topic, partition).await? {
        println!("Latest offset: {}", latest);
    }
    
    Ok(())
}

パフォーマンス最適化

// 大量データの効率的な書き込み
for i in 0..10000 {
    let event_data = EventData::from_bytes(format!("Event {}", i).into_bytes());
    storage.append_event(&topic, partition, event_data).await?;
}

// バッチ読み取り
let events = storage.read_events(&topic, partition, Offset::new(0), 1000).await?;

📊 性能特性

ベンチマーク結果

スケーラビリティ

• 最大セグメントサイズ: 制限なし（推奨1GB）
• 最大パーティション数: 制限なし
• 最大トピック数: 制限なし
• 並行アクセス: 完全対応

🔧 設定オプション

use kawa_storage::StorageConfig;

let config = StorageConfig {
    // データ保存ディレクトリ
    data_dir: PathBuf::from("./kawa-data"),
    
    // セグメントファイルサイズ（バイト）
    segment_size: 1024 * 1024 * 1024, // 1GB
    
    // 同期間隔（ミリ秒）
    sync_interval_ms: 1000,
    
    // 圧縮有効化（将来実装）
    enable_compression: false,
};

🗃️ データ構造

Event

pub struct Event {
    pub id: EventId,           // UUID v4
    pub topic: Topic,          // トピック名
    pub partition: Partition,  // パーティション番号
    pub data: EventData,       // イベントデータ
    pub timestamp: Timestamp,  // 作成時刻
    pub offset: Option<Offset>, // オフセット（読み取り時設定）
}

EventData

pub enum EventData {
    Json(String),      // JSON文字列
    Binary(Vec<u8>),   // バイナリデータ
    Text(String),      // プレーンテキスト
}

// 使用例
let json_data = EventData::from_json(r#"{"key": "value"}"#);
let binary_data = EventData::from_bytes(vec![1, 2, 3, 4]);
let text_data = EventData::from_text("Hello, World!");

🧪 テスト

テスト実行

# 全テスト実行
cargo test

# 統合テスト
cargo test --test integration_tests

# ベンチマーク
cargo test bench_ --release -- --nocapture

# カバレッジ
cargo tarpaulin --out Html

テスト構成

• Unit Tests: 各モジュールの基本機能
• Integration Tests: エンドツーエンドシナリオ
• Benchmark Tests: 性能測定
• Property Tests: ランダムデータでの検証

🚧 将来の機能

v0.2.0

• 圧縮サポート: LZ4/Snappy圧縮
• インデックス構築: 高速検索機能
• メモリプール: アロケーション最適化
• バッチAPI: 複数イベント一括処理

v0.3.0

• レプリケーション: 多ノード対応
• 暗号化: 保存時暗号化
• スキーマ検証: イベント構造検証
• 自動圧縮: 古いセグメント整理

⚠️ 制限事項

• トランザクション: 現在未対応
• 削除操作: イベントの削除は未対応
• 分散ストレージ: 単一ノードのみ
• スキーマ進化: 現在未対応

🔗 関連ドキュメント

• API Documentation
• Storage Design
• Performance Tuning
• Error Handling

📝 エラーハンドリング

use kawa_storage::{StorageError, StorageResult};

match storage.append_event(&topic, partition, event_data).await {
    Ok((event_id, offset)) => {
        println!("Success: {} at {}", event_id, offset);
    }
    Err(StorageError::Io { source, .. }) => {
        eprintln!("I/O Error: {}", source);
    }
    Err(StorageError::Serialization { source, .. }) => {
        eprintln!("Serialization Error: {}", source);
    }
    Err(e) => {
        eprintln!("Other Error: {}", e);
    }
}

kawa-storage - Rustで実装された高性能イベントストレージ 🚀