DOCSEARCHbyAlgolia
Submit Docs

Walrus Documentation

repository·master·Indexed 23 days ago

https://github.com/nubskr/walrus

A high-performance distributed message streaming engine and Write-Ahead Log (WAL) implementation in Rust. Walrus uses Raft consensus for metadata coordination and segment-based partitioning for load balancing and scalable reads. It includes a TCP-based client protocol with commands for topic registration, data ingestion (PUT), and retrieval (GET), as well as a CLI for cluster management.

Tokens
139K
Snippets
209
Records
570
Agent score
82%

What's inside walrus

  1. Overview of Shipping Lane P2P File Transfer

    master

    The Shipping Lane is Octopii's peer-to-peer (P2P) file transfer system. It is specifically designed to handle data transfers that are too large for the standard RPC layer, avoiding the overhead of routing through the Raft leader by enabling direct peer-to-peer communication.

    Key Features

    • Direct P2P Transfer: Data moves directly between peers.
    • Checksum Verification: Uses SHA-256 to ensure data integrity.
    • QUIC-based: Utilizes QUIC for fast, reliable, and multiplexed transport.
    • Streaming Support: Provides mechanisms for efficient memory usage during large transfers.
    • Composable: Allows developers to build custom protocols by combining RPC (for negotiation/control) and Shipping Lane (for bulk data).

  2. Overview of Distributed Walrus Architecture

    master

    Distributed Walrus is a distributed streaming log system built on top of the Walrus storage engine. It provides a fault-tolerant, distributed write-ahead log featuring:

    • Automatic leadership rotation: Uses Raft consensus for managing leadership.
    • Segment-based partitioning: Organizes data into segments.
    • Metadata Coordination: Uses Octopii (Raft consensus) to replicate metadata and handle leader election.
    • Durable Data Storage: Uses the Walrus engine for the actual data storage and I/O operations.

  3. Architecture of the Raft gRPC Key-Value Store

    master

    The project is organized into several functional areas:

    • Server Entry Point: src/bin/main.rs
    • Network Routing: src/network/ (implements RaftNetwork using Tonic)
    • gRPC Services (src/grpc/):
      • api_service.rs: Handles application-level APIs (read/write) and cluster management.
      • raft_service.rs: Handles internal Raft protocol RPCs.
    • Protocol Definitions: protos/ (contains the Protocol Buffer specifications).
    • Storage: src/store/mod.rs (implements in-memory log storage and the StateMachineData state machine).

  4. Octopii System Architecture Overview

    master

    Octopii is a distributed consensus system built on top of OpenRaft. It provides a modular stack for building replicated state machines in Rust, separating the application logic from the consensus, storage, and network layers.

    Layered Architecture

    1. Application Layer: Where you implement your specific business logic by providing a StateMachineTrait implementation.
    2. Public API (OctopiiNode): The primary interface for interacting with the cluster. It exposes methods for proposing changes, querying state, managing learners, and accessing metrics.
    3. Consensus Layer (OpenRaft): Handles core distributed systems tasks including leader election, log replication, and snapshots.
    4. Storage Layer: Manages persistence. It includes WalLogStore for log storage, MemStateMachine for the state machine, and a crash-resistant WriteAheadLog (WAL) that uses checksums and fsync for durability.
    5. Network Layer: Handles communication using a QUIC-based transport (QuinnNetwork). It utilizes RpcHandler and QuicTransport (via Quinn) to provide low-latency, multiplexed, and connection-oriented QUIC/TLS streams.
    ┌─────────────────────────────────────────────────────┐
│                 Application Layer                    │
│           (Your State Machine Implementation)        │
└──────────────────────┬──────────────────────────────┘
                       │ StateMachineTrait
┌──────────────────────┴──────────────────────────────┐
│              OctopiiNode (Public API)                │
│  propose() │ query() │ add_learner() │ metrics()    │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────┴──────────────────────────────┐
│              OpenRaft (Consensus Layer)              │
│   Leader Election │ Log Replication │ Snapshots     │
└─────┬───────────────────────────────────────┬───────┘
      │                                       │
┌─────┴──────────┐                   ┌────────┴──────┐
│  Storage Layer │                   │ Network Layer │
│                │                   │               │
│ ┌────────────┐ │                   │ ┌───────────┐ │
│ │ WalLogStore│ │                   │ │QuinnNetwork││
│ └─────┬──────┘ │                   │ └─────┬─────┘ │
│       │        │                   │       │       │
│ ┌─────┴──────┐ │                   │ ┌─────┴─────┐ │
│ │WriteAheadLog│ │                   │ │RpcHandler │ │
│ │  (Walrus)  │ │                   │ └─────┬─────┘ │
│ └────────────┘ │                   │       │       │
│                │                   │ ┌─────┴─────┐ │
│ ┌────────────┐ │                   │ │QuicTransport│ │
│ │MemStateMach│ │                   │ │  (Quinn)  │ │
│ │   ine      │ │                   │ └───────────┘ │
│ └────────────┘ │                   │               │
└────────────────┘                   └───────────────┘
         │                                    │
         │ Persistence                        │ Network
         ▼                                    ▼
    [Disk: WAL]                      [QUIC/TLS Streams]

  5. Explore OpenRaft implementation examples

    master
    The octopii/openraft/examples directory provides several complete application examples that demonstrate how to implement different components of the OpenRaft ecosystem. These examples vary based on their storage engines, network protocols, and runtime characteristics. Use these examples to understand how to compose LogStore, StateMachine, RaftNetwork, and client/server components.

  7. What is a ReplicationSessionId and how does it ensure consistency?

    master

    In OpenRaft, a ReplicationSessionId is a unique identifier used to distinguish individual replication sessions. It is used to guarantee that replication states (such as progress updates) are correctly handled during leader changes or cluster membership modifications.

    When a Leader initiates replication to a set of target nodes, it starts a session. As replication progresses (e.g., a node receives a specific log entry), the replication module sends update messages containing the ReplicationSessionId. This allows the Raft core to track progress accurately and ensures that updates from an old session (e.g., from a previous Leader or a previous membership configuration) are not mistakenly applied to a new session.

    A new ReplicationSessionId is automatically created whenever:

    • The Leader's identity changes.
    • The cluster's membership changes.

  8. Overview of Walrus: The Write-Ahead Log Engine

    master

    Walrus is Octopii's high-performance Write-Ahead Log (WAL) implementation. It provides durable persistence for state machines using a topic-based organization similar to Kafka. Data is stored in 64KB mmap'd, checksummed blocks.

    Walrus supports configurable consistency modes and fsync scheduling to balance the trade-offs between latency, throughput, and durability.

  9. Understand the Chunk Transfer Protocol

    master

    For large file or snapshot transfers, Octopii uses a custom framing protocol with checksums to ensure reliability.

    Protocol Flow:

    Sender:

    1. Send size (as u64).
    2. Send data[0..N] (split into 64KB chunks).
    3. Send SHA256(data) checksum.

    Receiver:

    1. Receive size.
    2. Receive exact amount of data.
    3. Receive checksum.
    4. Verify SHA256(data) == checksum.
    5. Send ACK (0 for OK, 1 for checksum mismatch, 2 for error).

  10. Atomicity and Failure Recovery in Batch Writes

    master

    Walrus guarantees atomicity for batch writes through a four-phase execution model involving pre-allocation, io_uring submission, and rollback capabilities.

    Atomicity Guarantees

    • All-or-nothing: Readers will see either all entries in a batch or none. No partial batches are ever visible.
    • Kernel-level atomicity: Uses io_uring batched submission to ensure write operations are treated as an atomic unit by the kernel.

    Recovery Behavior

    If a failure occurs during the batch process, the system attempts to restore the topic to its original state:

    Failure PointRecovery ActionResulting State
    Block allocation failsRelease locks, return errorOriginal state preserved
    io_uring write failsRollback offsets, mark blocks unlockedOriginal state restored
    fsync failsRollback offsets, mark blocks unlockedOriginal state restored
    Panic during batchRAII guard releases atomic flagFlag released; process restart triggers normal recovery

    Note: A panic during a batch is considered a catastrophic failure. While the RAII guard ensures the is_batch_writing flag is released, a process restart is required to ensure full recovery.

  11. High-Level Architecture of Distributed Walrus

    master

    The system consists of multiple nodes communicating via TCP. Clients (such as walrus-cli or Python scripts) connect to nodes on ports :8080-9093 using length-prefixed protocols.

    Node Components

    Each node contains several key layers:

    1. Client Listener: Handles incoming REGISTER, PUT, GET, STATE, and METRICS requests.
    2. NodeController:
      • Routing Logic: The leader uses ensure_topic() and append/read operations. Followers forward requests to the leader.
      • Lease Sync: A 100ms loop that runs update_leases() to maintain synchronization.
    3. Storage Layers:
      • Metadata State: Stores Topics, Nodes, and Segments via Octopii.
      • Bucket (Storage): Stores Leases (e.g., logs:1) and logs (e.g., logs:2) using the Walrus engine.
    4. Walrus Engine: Performs batch_append, read_next, and Disk I/O.
    5. Octopii (Raft Engine): Handles metadata replication, leader election, and log commits via Raft RPC (e.g., AppendEntries, RequestVote).

  12. How gRPC payload chunking works in the Raft network

    master

    The implementation includes automatic payload chunking to handle cases where append_entries RPCs exceed the gRPC message size limit.

    When a message is too large, the system detects the error and retries using a chunked transmission strategy. This ensures that large log entries can be replicated across the cluster without violating gRPC constraints.

    To observe the chunking logic and logs in action, run the specific chunking test with warning-level logging enabled:

    RUST_LOG=warn cargo test --test test_chunk -- --nocapture