Skip to content

development.md

Latest commit

 

History

History
562 lines (437 loc) · 17.8 KB

development.md

File metadata and controls

562 lines (437 loc) · 17.8 KB
Raw

🔧 Message Broker Core API

RustMQ now includes a fully implemented high-level Message Broker Core that provides intuitive producer and consumer APIs with comprehensive error handling, automatic partition management, and flexible acknowledgment levels.

Architecture Overview

The Message Broker Core is built on a modular architecture that integrates seamlessly with RustMQ's distributed storage and replication systems:

use rustmq::broker::core::*;

// Create a broker core instance with your storage backends
let core = MessageBrokerCore::new(
    wal,               // Write-Ahead Log implementation
    object_storage,    // Object storage backend (S3/GCS/Azure)
    cache,             // Distributed cache layer
    replication_manager, // Replication coordinator
    network_handler,   // Network communication handler
    broker_id,         // Unique broker identifier
);

Producer API

The Producer trait provides a simple, high-performance interface for message production:

#[async_trait]
pub trait Producer {
    /// Send a single record to a topic-partition
    async fn send(&self, record: ProduceRecord) -> Result<ProduceResult>;
    
    /// Send a batch of records for optimized throughput
    async fn send_batch(&self, records: Vec<ProduceRecord>) -> Result<Vec<ProduceResult>>;
    
    /// Flush any pending records to ensure durability
    async fn flush(&self) -> Result<()>;
}

Single Message Production

let producer = core.create_producer();

let record = ProduceRecord {
    topic: "user-events".to_string(),
    partition: Some(0),                    // Optional: let RustMQ choose partition
    key: Some(b"user123".to_vec()),
    value: b"login_event".to_vec(),
    headers: vec![Header {
        key: "content-type".to_string(),
        value: b"application/json".to_vec(),
    }],
    acks: AcknowledgmentLevel::All,        // Wait for all replicas
    timeout_ms: 5000,
};

let result = producer.send(record).await?;
println!("Message produced at offset: {}", result.offset);

Batch Production for High Throughput

let mut batch = Vec::new();
for i in 0..1000 {
    batch.push(ProduceRecord {
        topic: "metrics".to_string(),
        partition: None,  // Auto-partition based on key hash
        key: Some(format!("sensor_{}", i % 10).into_bytes()),
        value: format!("{{\"value\": {}, \"timestamp\": {}}}", i, timestamp).into_bytes(),
        headers: vec![],
        acks: AcknowledgmentLevel::Leader,  // Faster acknowledgment
        timeout_ms: 1000,
    });
}

let results = producer.send_batch(batch).await?;
println!("Produced {} messages", results.len());

Consumer API

The Consumer trait provides flexible message consumption with automatic offset management:

#[async_trait]
pub trait Consumer {
    /// Subscribe to one or more topics
    async fn subscribe(&mut self, topics: Vec<TopicName>) -> Result<()>;
    
    /// Poll for new records with configurable timeout
    async fn poll(&mut self, timeout_ms: u32) -> Result<Vec<ConsumeRecord>>;
    
    /// Commit specific offsets for durability
    async fn commit_offsets(&mut self, offsets: HashMap<TopicPartition, Offset>) -> Result<()>;
    
    /// Seek to a specific offset for replay scenarios
    async fn seek(&mut self, topic_partition: TopicPartition, offset: Offset) -> Result<()>;
}

Basic Consumer Usage

let mut consumer = core.create_consumer("analytics-group".to_string());

// Subscribe to topics
consumer.subscribe(vec!["user-events".to_string(), "orders".to_string()]).await?;

// Consume messages
loop {
    let records = consumer.poll(1000).await?;
    
    for record in records {
        println!("Received: topic={}, partition={}, offset={}", 
                 record.topic_partition.topic,
                 record.topic_partition.partition,
                 record.offset);
        
        // Process your message
        process_message(&record.value).await?;
        
        // Optional: Manual offset commit for exactly-once processing
        let mut offsets = HashMap::new();
        offsets.insert(record.topic_partition.clone(), record.offset + 1);
        consumer.commit_offsets(offsets).await?;
    }
}

Consumer Seek for Message Replay

// Replay messages from a specific point in time
let topic_partition = TopicPartition {
    topic: "user-events".to_string(),
    partition: 0,
};

// Seek to offset 1000 to replay messages
consumer.seek(topic_partition, 1000).await?;

// Continue normal polling - will start from offset 1000
let records = consumer.poll(5000).await?;

Acknowledgment Levels

RustMQ supports flexible acknowledgment levels for different durability and performance requirements:

use rustmq::types::AcknowledgmentLevel;

// Maximum performance - fire and forget
acks: AcknowledgmentLevel::None,

// Fast acknowledgment - leader only
acks: AcknowledgmentLevel::Leader,

// High availability - majority of replicas
acks: AcknowledgmentLevel::Majority,

// Maximum durability - all replicas
acks: AcknowledgmentLevel::All,

// Custom requirement - specific number of replicas
acks: AcknowledgmentLevel::Custom(3),

Error Handling

The Broker Core provides comprehensive error handling with detailed error types:

use rustmq::error::RustMqError;

match producer.send(record).await {
    Ok(result) => println!("Success: offset {}", result.offset),
    Err(RustMqError::NotLeader(partition)) => {
        println!("Not leader for partition: {}", partition);
        // Retry with updated metadata
    },
    Err(RustMqError::OffsetOutOfRange(msg)) => {
        println!("Offset out of range: {}", msg);
        // Seek to valid offset
    },
    Err(RustMqError::Timeout) => {
        println!("Request timed out");
        // Implement retry logic
    },
    Err(e) => println!("Other error: {}", e),
}

Integration with Storage Layers

The Broker Core seamlessly integrates with RustMQ's tiered storage architecture:

  • Local WAL: Recent messages are served from high-speed local NVMe storage
  • Cache Layer: Frequently accessed messages are cached for optimal performance
  • Object Storage: Historical messages are automatically migrated to cost-effective cloud storage
  • Intelligent Routing: The core automatically routes read requests to the optimal storage tier

Testing and Validation

The Message Broker Core includes comprehensive test coverage:

  • Unit Tests: Core functionality with 88 passing tests
  • Integration Tests: End-to-end workflows with 9 comprehensive test scenarios
  • Mock Implementations: Complete test doubles for all dependencies
  • Error Scenarios: Comprehensive error condition testing

Performance Characteristics

WAL I/O Design

RustMQ uses a single shared, sequential write-ahead log optimized for append throughput and durability:

  • O_DIRECT block-aligned writer: WAL records are packed into 4 KiB-aligned blocks and written with pwrite, bypassing the page cache to avoid double-buffering and cache pollution from cold reads.
  • Dedicated writer thread: the WAL runs on its own OS thread issuing blocking pwrite/fsync (std::os::unix::fs::FileExt), fed over an mpsc channel. (tokio::fs cannot do O_DIRECT — it copies through an unaligned internal buffer — so a dedicated thread is required for aligned I/O.)
  • Group commit: appends are acknowledged after the WAL append; fsync is batched across records to amortize device-flush latency rather than paying it per record.
  • Buffered fallback: O_DIRECT support is probed at runtime; on filesystems that reject it (e.g. tmpfs, some sandboxes) the identical write path runs with buffered I/O — only the open flag differs.

Note: io_uring was evaluated and deliberately not adopted. For a single sequential, already block-batched and group-committed append stream, io_uring offers no throughput gain (the bottleneck is device bandwidth + fsync latency, which it does not change) while adding kernel attack surface and deployment fragility in sandboxed environments (gVisor, restrictive seccomp). See plan.md for the full rationale.

Overall Performance

  • Low Latency: Sub-millisecond produce latency for local WAL writes (O_DIRECT + group commit)
  • High Throughput: Batch production for maximum throughput scenarios
  • Automatic Partitioning: Intelligent partition selection based on message keys
  • Zero-Copy Operations: Efficient memory usage throughout the message path with buffer reuse (detailed optimization guide)
  • Async Throughout: Non-blocking I/O for maximum concurrency

📊 Future Performance Tuning (Not Yet Implemented)

Planned Broker Optimization

# High-throughput configuration
[wal]
capacity_bytes = 53687091200        # 50GB for high-volume topics
fsync_on_write = false              # Disable for maximum throughput
segment_size_bytes = 2147483648     # 2GB segments
buffer_size = 1048576               # 1MB buffer

[cache]
write_cache_size_bytes = 8589934592  # 8GB hot cache
read_cache_size_bytes = 17179869184  # 16GB cold cache

[network]
max_connections = 50000             # Increase connection limit
connection_timeout_ms = 60000       # Longer timeout for slow clients

[object_storage]
max_concurrent_uploads = 50         # More concurrent uploads
multipart_threshold = 52428800      # 50MB threshold

# Low-latency configuration
[wal]
fsync_on_write = true               # Enable for durability
buffer_size = 4096                  # Smaller buffers for low latency

[replication]
min_in_sync_replicas = 1            # Reduce for lower latency
ack_timeout_ms = 1000               # Faster timeouts
heartbeat_timeout_ms = 10000        # Shorter heartbeat timeout for faster failover

Planned Kubernetes Resource Tuning

# High-performance broker configuration
resources:
  requests:
    memory: "16Gi"
    cpu: "8"
    ephemeral-storage: "100Gi"
  limits:
    memory: "32Gi"
    cpu: "16"
    ephemeral-storage: "200Gi"

# Node affinity for performance
nodeSelector:
  cloud.google.com/gke-nodepool: high-performance
  
affinity:
  podAntiAffinity:
    requiredDuringSchedulingIgnoredDuringExecution:
    - labelSelector:
        matchLabels:
          app: rustmq-broker
      topologyKey: kubernetes.io/hostname

# Volume configuration for maximum IOPS
volumeClaimTemplates:
- metadata:
    name: wal-storage
  spec:
    accessModes: ["ReadWriteOnce"]
    storageClassName: fast-ssd
    resources:
      requests:
        storage: 500Gi

📈 Future Monitoring (Not Yet Implemented)

Planned Prometheus Configuration

# prometheus-config.yaml - future monitoring setup
apiVersion: v1
kind: ConfigMap
metadata:
  name: prometheus-config
data:
  prometheus.yml: |
    global:
      scrape_interval: 15s
      
    scrape_configs:
    - job_name: 'rustmq-brokers'
      kubernetes_sd_configs:
      - role: pod
        namespaces:
          names: [rustmq]
      relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        action: keep
        regex: rustmq-broker
      - source_labels: [__meta_kubernetes_pod_ip]
        target_label: __address__
        replacement: ${1}:9642
        
    - job_name: 'rustmq-controllers'
      kubernetes_sd_configs:
      - role: pod
        namespaces:
          names: [rustmq]
      relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        action: keep
        regex: rustmq-controller
      - source_labels: [__meta_kubernetes_pod_ip]
        target_label: __address__
        replacement: ${1}:9642

Future Monitoring (Not Yet Implemented)

Planned metrics to monitor:

  • Throughput: rate(messages_produced_total[5m]), rate(messages_consumed_total[5m])
  • Latency: produce_latency_seconds, consume_latency_seconds
  • Storage: wal_size_bytes, cache_hit_ratio, object_storage_upload_rate
  • Replication: replication_lag, in_sync_replicas_count
  • System: cpu_usage, memory_usage, disk_iops, network_throughput

Future Alerting (Not Yet Implemented)

# alerts.yaml - planned alerting rules
groups:
- name: rustmq.rules
  rules:
  - alert: HighProduceLatency
    expr: histogram_quantile(0.95, produce_latency_seconds) > 0.1
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "High produce latency detected"
      
  - alert: ReplicationLagHigh
    expr: replication_lag > 10000
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Replication lag is too high"
      
  - alert: BrokerDown
    expr: up{job="rustmq-brokers"} == 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "RustMQ broker is down"

🔧 Development & Troubleshooting

🚀 Environment Setup

RustMQ provides a comprehensive environment setup script that can configure both development and production environments:

Development Environment Setup

Set up a complete local development environment with certificates, configurations, and services:

# Set up complete development environment (recommended)
./generate-certs.sh develop

# Force regenerate existing setup
./generate-certs.sh develop --force

# View available options
./generate-certs.sh --help

What the development setup provides:

  • ✅ Self-signed development certificates with simplified CA chain
  • ✅ Development configuration files for broker, controller, and admin
  • ✅ Local data directories and startup scripts
  • ✅ Example applications and test clients
  • ✅ Ready-to-run local cluster

Quick start after setup:

# Start the complete cluster
./start-cluster-dev.sh

# Or start services individually
./start-controller-dev.sh  # Start controller first
./start-broker-dev.sh      # Start broker

# Test with examples
cargo run --example secure_producer
cargo run --example secure_consumer

# Admin operations
cargo run --bin rustmq-admin -- --config config/admin-dev.toml cluster status

Production Environment Setup

Get comprehensive guidance for production deployment:

# Show production setup guidance
./generate-certs.sh production

What the production guidance provides:

  • 📋 Production setup guidance and checklists
  • 📋 Security best practices and hardening
  • 📋 Deployment options (Kubernetes, Docker, systemd)
  • 📋 Monitoring and observability setup
  • 📋 Certificate management with external CA

Environment Setup Features

Development Environment (./generate-certs.sh develop):

  • Complete Setup: Creates certificates, configs, data directories, and startup scripts
  • Certificate Chain: Proper CA-signed certificates (fixes August 2025 certificate signing issues)
  • Ready to Use: Start developing immediately with ./start-cluster-dev.sh
  • Examples Included: Secure producer/consumer examples with mTLS
  • Validation: Automatic setup validation and certificate verification

Production Environment (./generate-certs.sh production):

  • Security Guidance: Enterprise-grade security setup instructions
  • Certificate Management: External CA integration and certificate lifecycle
  • Deployment Options: Kubernetes, Docker, and systemd deployment guides
  • Monitoring Setup: Comprehensive observability and alerting configuration
  • Best Practices: Production hardening and operational procedures

🔐 Development Certificates

The development environment automatically generates certificates with simplified signing chains:

  • certs/ca.pem - Root CA certificate (self-signed for development)
  • certs/server.pem + certs/server.key - Server certificate and private key (CA-signed)
  • certs/client.pem + certs/client.key - Client certificate and private key (CA-signed)
  • certs/admin.pem + certs/admin.key - Admin certificate and private key (CA-signed)

⚠️ Security Notice: These are development-only certificates. For production, follow the production setup guide!

Current Development Issues

Note: Since RustMQ is in early development, most "issues" are actually missing implementations.

  1. Services Not Responding
# Both broker and controller services are now production-ready with full functionality
# Check if they started successfully
docker-compose logs rustmq-broker-1
docker-compose logs rustmq-controller-1

# Look for configuration loading messages
# Services should log "started successfully" then sleep
  1. Build Issues
# Ensure Rust toolchain is up to date
rustup update

# Clean build if needed
cargo clean
cargo build --release

# Run tests to verify implementation
cargo test
  1. Configuration Issues
# Validate configuration
cargo run --bin rustmq-broker -- --config config/broker.toml

# Check configuration structure in src/config.rs
# All fields must be present in TOML files

Log Analysis

# View service logs (from docker/ directory)
cd docker
docker-compose logs rustmq-broker-1
docker-compose logs rustmq-controller-1

# Check for configuration validation errors
docker-compose logs | grep ERROR

# Monitor BigQuery subscriber demo
docker-compose logs rustmq-bigquery-subscriber

For complete Docker and Kubernetes deployment guides, troubleshooting, and configuration details, see docker/README.md.

Local GCS Testing

To run GCS integration tests locally during development, you can use a real Google Cloud Storage bucket.

  1. Copy the configuration template: 
    cp gcs-test.yaml.template gcs-test.yaml
  2. Update gcs-test.yaml with your GCP project ID and test bucket name.
  3. Authenticate via one of these methods:
    • Provide a path to a service account JSON key in the credentials_path field.
    • Set the GOOGLE_APPLICATION_CREDENTIALS environment variable.
    • Use Application Default Credentials (ADC) by running gcloud auth application-default login.
  4. Run the integration tests: 
    cargo test --test integration_gcs -- --ignored
    Note: If gcs-test.yaml is not found, these tests will be skipped automatically.