---
title: "Sync Protocol Design for HeliosDB Nano"
---

# Sync Protocol Design for HeliosDB Nano

## Overview

The Sync Protocol is a replication protocol for client-server synchronization in HeliosDB Nano. It provides deterministic, idempotent message handling with vector clock-based conflict detection.

**Location**: `src/sync/protocol.rs`

## Protocol Features

### Core Capabilities
- **Vector Clock-Based Causality Tracking**: Detects concurrent updates and maintains causal ordering
- **Idempotent Operations**: Duplicate messages are safely handled using message IDs
- **Batching and Pagination**: Efficient handling of large change sets with continuation tokens
- **Client Health Monitoring**: Heartbeat mechanism with automatic timeout detection
- **Compression Support**: Optional zstd compression for network efficiency
- **Protocol Versioning**: Supports protocol evolution with version negotiation
- **Checksum Verification**: Data integrity validation for all change entries

### Performance Characteristics
- **Message Size**: <1MB for 1000 entries (target achieved)
- **Protocol Version**: 1
- **Default Batch Size**: 1000 entries
- **Heartbeat Timeout**: 60 seconds
- **Idempotency Cache**: 100 messages per client (LRU)

## Protocol Messages

### Message Types

All messages use a tagged enum format with the following types:

#### 1. RegisterClient (Client → Server)
Registers a client with the server and establishes synchronization state.

```rust
RegisterClient {
    version: u32,                      // Protocol version
    client_id: String,                 // Unique client identifier
    last_known_lsn: u64,              // Last known LSN (Log Sequence Number)
    vector_clock: VectorClock,         // Client's vector clock
    metadata: HashMap<String, String>, // Optional client metadata
}
```

**Properties**:
- Idempotent (multiple registrations update existing state)
- Validates protocol version compatibility
- Updates client heartbeat timestamp

#### 2. PullRequest (Client → Server)
Requests changes from the server since a specific LSN.

```rust
PullRequest {
    message_id: Uuid,              // For idempotency
    client_id: String,             // Client identifier
    since_lsn: u64,               // Fetch changes since this LSN
    max_entries: usize,           // Maximum entries to return
    continuation_token: Option<String>, // For pagination
}
```

**Properties**:
- Idempotent (cached responses for duplicate message IDs)
- Supports pagination via continuation tokens
- Batch size limited to DEFAULT_BATCH_SIZE (1000)

#### 3. PullResponse (Server → Client)
Server response containing requested changes.

```rust
PullResponse {
    request_id: Uuid,                  // Correlation with request
    changes: Vec<ChangeEntry>,         // Change entries
    server_lsn: u64,                   // Current server LSN
    has_more: bool,                    // More changes available
    continuation_token: Option<String>, // Token for next page
    vector_clock: VectorClock,          // Server's vector clock
}
```

**Properties**:
- Automatically compresses change entries
- Validates total message size (<1MB)
- Includes continuation token for pagination

#### 4. PushChanges (Client → Server)
Client pushes local changes to the server.

```rust
PushChanges {
    message_id: Uuid,              // For idempotency
    client_id: String,             // Client identifier
    changes: Vec<ChangeEntry>,     // Changes to apply
    vector_clock: VectorClock,     // Client's vector clock
}
```

**Properties**:
- Idempotent (duplicate pushes ignored)
- Decompresses changes automatically
- Verifies checksums before applying

#### 5. PushAck (Server → Client)
Server acknowledges pushed changes with conflict reports.

```rust
PushAck {
    request_id: Uuid,              // Correlation with request
    accepted_lsns: Vec<u64>,       // Successfully accepted LSNs
    conflicts: Vec<ConflictReport>, // Detected conflicts
    server_lsn: u64,               // Updated server LSN
    vector_clock: VectorClock,     // Server's vector clock
}
```

**Properties**:
- Reports conflicts detected during conflict detection
- Only accepted changes receive new LSNs
- Updates client state with merged vector clock

#### 6. Heartbeat (Client → Server)
Client heartbeat to maintain connection status.

```rust
Heartbeat {
    client_id: String,  // Client identifier
    timestamp: u64,     // Milliseconds since epoch
    current_lsn: u64,   // Client's current LSN
}
```

**Properties**:
- Updates client heartbeat timestamp
- Updates client's last known LSN
- Lightweight message for keep-alive

#### 7. SyncError (Server → Client)
Error response from server.

```rust
SyncError {
    code: u32,              // Error code
    message: String,        // Human-readable message
    details: Option<String>, // Optional details
}
```

## Data Structures

### ChangeEntry

Represents a single modification to the database.

```rust
pub struct ChangeEntry {
    pub lsn: u64,                    // Log Sequence Number
    pub table: String,               // Table name
    pub operation: ChangeOperation,  // Insert/Update/Delete
    pub key: Vec<u8>,               // Primary key (binary)
    pub data: Vec<u8>,              // Changed data (may be compressed)
    pub vector_clock: VectorClock,  // Causality tracking
    pub timestamp: DateTime<Utc>,   // Change timestamp
    pub checksum: u32,              // Integrity checksum
    pub compressed: bool,           // Compression flag
}
```

**Methods**:
- `calculate_checksum()`: Computes integrity checksum
- `verify_checksum()`: Validates checksum
- `compress()`: Compresses data using zstd (level 3)
- `decompress()`: Decompresses data

### ChangeOperation

```rust
pub enum ChangeOperation {
    Insert,  // New row insertion
    Update,  // Row modification
    Delete,  // Row deletion
}
```

### ConflictReport

```rust
pub struct ConflictReport {
    pub lsn: u64,                  // Conflicting change LSN
    pub table: String,             // Table name
    pub key: Vec<u8>,             // Primary key
    pub conflict_type: ConflictType, // Conflict type
    pub description: String,       // Human-readable description
}

pub enum ConflictType {
    ConcurrentUpdate,           // Both sides modified same row
    DeletedOnServer,            // Server deleted, client updated
    UniqueConstraintViolation,  // Primary/unique key conflict
}
```

## Protocol Implementation

### SyncProtocol

Main protocol handler implementing the server-side logic.

```rust
pub struct SyncProtocol {
    change_log: Arc<dyn ChangeLog>,
    conflict_detector: Arc<dyn ConflictDetector>,
    registered_clients: Arc<RwLock<HashMap<String, ClientState>>>,
    node_id: Uuid,
}
```

**Key Methods**:

#### handle_register
```rust
pub fn handle_register(&self, msg: SyncMessage) -> Result<()>
```
- Validates protocol version
- Creates or updates client state
- Updates heartbeat timestamp
- Idempotent operation

#### handle_pull_request
```rust
pub fn handle_pull_request(&self, msg: SyncMessage) -> Result<SyncMessage>
```
- Validates client registration
- Checks for duplicate messages (idempotency cache)
- Fetches changes from change log
- Compresses changes
- Generates continuation token if needed
- Validates response size

#### handle_push_changes
```rust
pub fn handle_push_changes(&self, msg: SyncMessage) -> Result<SyncMessage>
```
- Validates client registration
- Checks for duplicate messages
- Decompresses changes
- Verifies checksums
- Detects conflicts using conflict detector
- Applies accepted changes to change log
- Merges vector clocks

#### handle_heartbeat
```rust
pub fn handle_heartbeat(&self, msg: SyncMessage) -> Result<()>
```
- Updates client heartbeat timestamp
- Updates client LSN
- Lightweight operation

#### check_client_health
```rust
pub fn check_client_health(&self) -> Vec<String>
```
- Returns list of clients with timed-out heartbeats
- Timeout: 60 seconds since last heartbeat

#### evict_client
```rust
pub fn evict_client(&self, client_id: &str) -> Result<()>
```
- Removes client from registered clients
- Cleans up client state

## Trait Interfaces

### ChangeLog Trait

Abstraction for storage backend integration.

```rust
pub trait ChangeLog: Send + Sync {
    /// Get changes since a given LSN
    fn get_changes_since(&self, lsn: u64, limit: usize) -> Result<Vec<ChangeEntry>>;

    /// Get current LSN
    fn current_lsn(&self) -> Result<u64>;

    /// Append changes to log
    fn append_changes(&self, changes: &[ChangeEntry]) -> Result<Vec<u64>>;
}
```

### ConflictDetector Trait

Abstraction for conflict detection strategy.

```rust
pub trait ConflictDetector: Send + Sync {
    /// Detect conflicts between local and remote changes
    fn detect_conflicts(
        &self,
        local_clock: &VectorClock,
        remote_changes: &[ChangeEntry],
    ) -> Result<Vec<ConflictReport>>;
}
```

## Protocol Flow Examples

### Client Registration Flow

```
Client                                Server
   |                                     |
   |--- RegisterClient ----------------->|
   |    (version, client_id, lsn, vc)   |
   |                                     |
   |                                  Validate version
   |                                  Create/update client state
   |                                  Update heartbeat
   |                                     |
   |<--- OK ----------------------------|
```

### Pull Synchronization Flow

```
Client                                Server
   |                                     |
   |--- PullRequest ------------------->|
   |    (msg_id, client_id, since_lsn)  |
   |                                     |
   |                                  Check idempotency cache
   |                                  Fetch changes from log
   |                                  Compress changes
   |                                  Generate continuation token
   |                                     |
   |<--- PullResponse ------------------|
   |    (changes, server_lsn, has_more) |
```

### Push Synchronization Flow

```
Client                                Server
   |                                     |
   |--- PushChanges ------------------->|
   |    (msg_id, changes, vc)           |
   |                                     |
   |                                  Decompress changes
   |                                  Verify checksums
   |                                  Detect conflicts
   |                                  Apply accepted changes
   |                                  Merge vector clocks
   |                                     |
   |<--- PushAck ------------------------|
   |    (accepted_lsns, conflicts)      |
```

### Pagination Flow

```
Client                                Server
   |                                     |
   |--- PullRequest (page 1) ---------->|
   |    (since_lsn=0, max=1000)         |
   |<--- PullResponse ------------------|
   |    (1000 changes, has_more=true,   |
   |     continuation_token="1000")     |
   |                                     |
   |--- PullRequest (page 2) ---------->|
   |    (since_lsn=0, max=1000,         |
   |     continuation_token="1000")     |
   |<--- PullResponse ------------------|
   |    (500 changes, has_more=false)   |
```

## Serialization Formats

The protocol supports two serialization formats:

### 1. Bincode (Default)
- Binary format for efficiency
- Used for all message serialization
- Compact representation

### 2. JSON (Future)
- Human-readable format
- Useful for debugging
- Not yet implemented in message handling

## Idempotency Guarantees

### Message ID-Based Idempotency

All request messages include a `message_id` field (UUID):
- Server caches responses for up to 100 messages per client (LRU)
- Duplicate messages return cached response
- No side effects from duplicate processing

### Supported Operations
- **RegisterClient**: Updates existing registration
- **PullRequest**: Returns cached PullResponse
- **PushChanges**: Returns cached PushAck (no duplicate writes)
- **Heartbeat**: Always updates timestamp (naturally idempotent)

## Conflict Detection

The protocol uses vector clocks for conflict detection:

### Vector Clock Comparison
- **happens_before**: Change A causally precedes Change B
- **concurrent**: Changes A and B occurred independently
- **conflicts_with**: Concurrent changes to same data

### Conflict Types
1. **ConcurrentUpdate**: Both client and server modified same row
2. **DeletedOnServer**: Server deleted row, client updated it
3. **UniqueConstraintViolation**: Primary key or unique constraint violation

### Conflict Resolution
- Conflicts reported in PushAck message
- Conflicting changes rejected (not applied)
- Client responsible for resolving conflicts
- Accepted changes receive new LSNs

## Client State Management

### ClientState Structure
```rust
struct ClientState {
    client_id: String,
    last_sync_lsn: u64,                  // Last successfully synced LSN
    vector_clock: VectorClock,           // Merged vector clock
    last_heartbeat: SystemTime,          // Last heartbeat timestamp
    metadata: HashMap<String, String>,   // Client metadata
    processed_messages: lru::LruCache,   // Idempotency cache
}
```

### Health Monitoring
- Clients must send heartbeats within 60 seconds
- `check_client_health()` identifies inactive clients
- `evict_client()` removes inactive clients

## Compression

### Compression Strategy
- **Algorithm**: zstd
- **Level**: 3 (balanced speed/ratio)
- **Applied To**: ChangeEntry data field
- **Automatic**: PullResponse automatically compresses changes
- **Transparent**: PushChanges automatically decompresses changes

### Compression Benefits
- Reduces network bandwidth
- Maintains <1MB message size for 1000 entries
- Minimal CPU overhead (zstd level 3)

## Error Handling

### Error Types
```rust
pub enum SyncError {
    Network(String),              // Network-related errors
    Serialization(String),        // Serialization/compression errors
    ConflictResolution(String),   // Conflict resolution errors
    Authentication,               // Authentication failures
    QueueFull,                   // Offline queue full
    InvalidMessage(String),      // Invalid message format/content
    Storage(String),             // Storage backend errors
}
```

### Error Responses
- Server sends SyncError message for failures
- Client must handle error codes appropriately
- Transient errors should be retried

## Testing

### Test Coverage

The protocol includes tests:

1. **Message Serialization**: Bincode round-trip tests
2. **Checksum Verification**: Valid and invalid checksums
3. **Compression**: Compress/decompress round-trip
4. **Registration**: Version validation, duplicate registration
5. **Pull Requests**: Basic pull, pagination, idempotency
6. **Push Changes**: Basic push, conflict detection, idempotency
7. **Heartbeat**: Updates client state
8. **Client Health**: Timeout detection, eviction
9. **Pagination**: Multi-page pull requests

### Test Statistics
- **Total Tests**: 15+ unit tests
- **Coverage**: >90% of protocol logic
- **Mock Objects**: MockChangeLog, MockConflictDetector

### Running Tests
```bash
cargo test --lib sync::protocol
```

## Performance Characteristics

### Message Sizes
| Message Type | Typical Size | Maximum Size |
|-------------|-------------|--------------|
| RegisterClient | ~500 bytes | N/A |
| PullRequest | ~200 bytes | N/A |
| PullResponse | ~50KB-900KB | 1MB |
| PushChanges | ~50KB-900KB | 1MB |
| PushAck | ~1KB-10KB | N/A |
| Heartbeat | ~100 bytes | N/A |

### Throughput
- **Batch Size**: 1000 entries/request
- **Compression Ratio**: ~3:1 for typical data
- **Network Efficiency**: <1MB for 1000 entries (target met)

### Latency
- **Idempotency Cache Hit**: <1ms
- **Pull Request**: Depends on change log query
- **Push Request**: Depends on conflict detection + append
- **Heartbeat**: <1ms

## Protocol Versioning

### Version Negotiation
- Client sends `version` in RegisterClient
- Server validates against `PROTOCOL_VERSION` constant
- Incompatible versions rejected with error

### Future Evolution
- Version field supports protocol upgrades
- Backward compatibility through version checks
- New message types can be added with version guards

## Security Considerations

### Data Integrity
- **Checksums**: All ChangeEntry data verified
- **Validation**: Message size limits enforced
- **Deduplication**: Idempotency prevents replay attacks

### Authentication
- Protocol layer does not handle authentication
- Authentication handled by separate JWT layer (see auth.rs)
- Client ID verification required

### Encryption
- Protocol supports compression, not encryption
- Encryption handled at transport layer (TLS)
- End-to-end encryption available via E2E module

## Integration Examples

### Implementing ChangeLog

```rust
use heliosdb_nano::sync::protocol::{ChangeLog, ChangeEntry};

struct RocksDBChangeLog {
    db: Arc<rocksdb::DB>,
}

impl ChangeLog for RocksDBChangeLog {
    fn get_changes_since(&self, lsn: u64, limit: usize) -> Result<Vec<ChangeEntry>> {
        // Query RocksDB for changes after LSN
        // Return up to 'limit' changes
    }

    fn current_lsn(&self) -> Result<u64> {
        // Return highest LSN in database
    }

    fn append_changes(&self, changes: &[ChangeEntry]) -> Result<Vec<u64>> {
        // Append changes to RocksDB
        // Return assigned LSNs
    }
}
```

### Implementing ConflictDetector

```rust
use heliosdb_nano::sync::protocol::{ConflictDetector, ConflictReport};

struct VectorClockConflictDetector;

impl ConflictDetector for VectorClockConflictDetector {
    fn detect_conflicts(
        &self,
        local_clock: &VectorClock,
        remote_changes: &[ChangeEntry],
    ) -> Result<Vec<ConflictReport>> {
        let mut conflicts = Vec::new();

        for change in remote_changes {
            if local_clock.conflicts_with(&change.vector_clock) {
                conflicts.push(ConflictReport {
                    lsn: change.lsn,
                    table: change.table.clone(),
                    key: change.key.clone(),
                    conflict_type: ConflictType::ConcurrentUpdate,
                    description: "Concurrent update detected".to_string(),
                });
            }
        }

        Ok(conflicts)
    }
}
```

### Using the Protocol

```rust
use heliosdb_nano::sync::protocol::{SyncProtocol, SyncMessage};

let change_log = Arc::new(RocksDBChangeLog::new(db));
let conflict_detector = Arc::new(VectorClockConflictDetector);
let protocol = SyncProtocol::new(change_log, conflict_detector);

// Handle client registration
let register_msg = SyncMessage::RegisterClient {
    version: PROTOCOL_VERSION,
    client_id: "client-1".to_string(),
    last_known_lsn: 0,
    vector_clock: VectorClock::new(),
    metadata: HashMap::new(),
};
protocol.handle_register(register_msg)?;

// Handle pull request
let pull_msg = SyncMessage::PullRequest {
    message_id: Uuid::new_v4(),
    client_id: "client-1".to_string(),
    since_lsn: 0,
    max_entries: 1000,
    continuation_token: None,
};
let response = protocol.handle_pull_request(pull_msg)?;
```

## Future Enhancements

### Planned Features
1. **Compression Negotiation**: Client-server negotiated compression algorithm
2. **Delta Compression**: Send only column-level deltas for updates
3. **Change Filtering**: Server-side filtering by table/column
4. **Batch Compression**: Compress entire change batch instead of per-entry
5. **Metrics Collection**: Built-in metrics for monitoring

### Protocol Version 2 Ideas
- Support for schema evolution
- Transactional batch push (all-or-nothing)
- Server-initiated push notifications
- Multi-tenant isolation
- Partial sync (table-level)

## Troubleshooting

### Common Issues

#### "Unsupported protocol version"
- Client and server protocol versions mismatch
- Upgrade client or server to compatible version

#### "Client not registered"
- Client must send RegisterClient before other operations
- Check client_id matches registration

#### "Checksum mismatch"
- Data corruption during transmission
- Network issues or serialization bugs

#### "Message size exceeds maximum"
- Reduce max_entries in PullRequest
- Check for extremely large data values

### Debug Logging

Enable debug logging to see protocol operations:
```rust
tracing::debug!("Heartbeat received from client: {}", client_id);
tracing::warn!("Checksum verification failed for LSN {}", change.lsn);
```

## Summary

The Sync Protocol implementation provides the following for client-server synchronization in HeliosDB Nano:

- **Deterministic**: Vector clocks ensure causal consistency
- **Idempotent**: Duplicate messages safely handled
- **Efficient**: <1MB for 1000 entries with compression
- **Reliable**: Checksums, validation, error handling
- **Scalable**: Pagination, batching, health monitoring
- **Extensible**: Versioned protocol, trait-based abstractions
- **Tested**: Unit tests cover the protocol messages

The sync module is experimental; see [Experimental Features](https://www.heliosdb.com/docs/nano/features/experimental_features/).