Gruf::Queue

A high-performance, queue-based gRPC server extension for Gruf that provides enhanced thread pool management, intelligent resource handling, and seamless database connection management.

Features

🚀 Enhanced Performance

Queue-based request processing with intelligent job scheduling
Resource exhaustion protection with automatic RESOURCE_EXHAUSTED responses
Structured logging with comprehensive error handling and debugging support

🔌 Plugin Architecture

Zero-configuration auto-installation
Non-invasive integration with existing Gruf applications
Modular design with configurable components

🛡️ Reliability & Safety

Thread-safe operations with proper synchronization
Graceful error handling and recovery
Smart ActiveRecord connection management

📊 Observability

Structured logging with metadata for monitoring
Thread naming for better debugging
Comprehensive error reporting with context

Installation

Add this line to your application's Gemfile:

gem 'gruf-queue'

And then execute:

$ bundle install

Quick Start

Zero Configuration Setup

The simplest way to use gruf-queue is to just require it:

require 'gruf-queue'

# That's it! Everything is automatically configured.
# Start your Gruf server as usual:
Gruf::Server.new(
  hostname: '0.0.0.0:9001',
  services: [YourService]
).start

When you require gruf-queue, it automatically:

✅ Configures Gruf to use QueuedRpcServer
✅ Registers the ConnectionReset interceptor (if ActiveRecord is available)
✅ Sets up enhanced thread pool management
✅ Enables structured logging

Manual Installation Control

To disable auto-installation and configure manually:

# Disable auto-installation
ENV['GRUF_QUEUE_NO_AUTO_INSTALL'] = 'true'

require 'gruf-queue'

# Manual installation
Gruf::Queue::Plugin.install!

# Or configure components individually
Gruf.configure do |config|
  config.rpc_server = Gruf::Queue::QueuedRpcServer
  
  config.interceptors.use(
    Gruf::Queue::Interceptors::ConnectionReset,
    enabled: true,
    target_classes: [ActiveRecord::Base]
  )
end

Configuration

Server Configuration

# Configure the QueuedRpcServer with custom settings
Gruf.configure do |config|
  config.rpc_server = Gruf::Queue::QueuedRpcServer
  config.rpc_server_options = {
    pool_size: 10,                    # Thread pool size
    max_waiting_requests: 20,         # Queue capacity
    pool_keep_alive: 300,             # Thread keep-alive time
    poll_period: 1                    # Polling interval
  }
end

Using Server Factory

For advanced server creation:

# Create custom server instances
server = Gruf::Queue::ServerFactory.create_server(
  pool_size: 15,
  max_waiting_requests: 30,
  server: Gruf::Queue::QueuedRpcServer,
  interceptors: [
    Gruf::Queue::Interceptors::ConnectionReset
  ]
)

Gruf::Server.new(
  hostname: '0.0.0.0:9001',
  services: [YourService],
  rpc_server: server
).start

ActiveRecord Connection Management

The ConnectionReset interceptor automatically manages database connections:

# Automatically registered when ActiveRecord is detected
# Customizable with additional target classes:
Gruf.configure do |config|
  config.interceptors.use(
    Gruf::Queue::Interceptors::ConnectionReset,
    enabled: true,
    target_classes: [ActiveRecord::Base, CustomConnectionClass]
  )
end

Architecture

Core Components

`Gruf::Queue::QueuedRpcServer`

Enhanced gRPC server with intelligent resource management:

Monitors thread pool capacity
Automatically sends RESOURCE_EXHAUSTED when overloaded
Provides structured error handling

`Gruf::Queue::Pool`

Advanced thread pool implementation:

Extends GRPC::Pool with enhanced job scheduling
Structured logging with thread identification
Comprehensive error isolation and recovery

`Gruf::Queue::Interceptors::ConnectionReset`

Smart database connection management:

Automatically resets ActiveRecord connections after each request
Validates connection handlers before attempting reset
Graceful error handling to prevent request failures

`Gruf::Queue::Plugin`

Plugin management system:

Idempotent installation process
Comprehensive error handling during setup
Validation of dependencies and requirements

Plugin Lifecycle

# Plugin installation process
Gruf::Queue::Plugin.install!  # => true (success) or false (failure)

# Check installation status
Gruf::Queue::Plugin.installed?  # => true/false

# Reset for testing (test environments only)
Gruf::Queue::Plugin.reset!

Advanced Usage

Custom Server Implementation

# Create a custom server class
class MyCustomServer < Gruf::Queue::QueuedRpcServer
  def initialize(*)
    super
    # Custom initialization logic
  end
  
  private
  
  def handle_custom_logic
    # Your custom server logic
  end
end

# Use with gruf-queue
Gruf.configure do |config|
  config.rpc_server = MyCustomServer
end

Monitoring and Observability

# Structured logging is automatically enabled
# Logs include contextual information:
# - Thread IDs for debugging
# - Error details with full context
# - Performance metrics
# - Resource utilization info

# Example log output:
# [INFO] Starting thread pool target_size=10
# [DEBUG] Worker thread started thread_id=47185656920560
# [WARN] Job execution failed error=SomeError error_class=StandardError

Environment-specific Configuration

# Production settings
if Rails.env.production?
  Gruf.configure do |config|
    config.rpc_server_options = {
      pool_size: 20,
      max_waiting_requests: 50,
      pool_keep_alive: 600
    }
  end
end

# Development settings
if Rails.env.development?
  Gruf.configure do |config|
    config.rpc_server_options = {
      pool_size: 5,
      max_waiting_requests: 10,
      pool_keep_alive: 60
    }
  end
end

Performance Considerations

Thread Pool Sizing: Set pool_size based on your server's CPU cores and expected load
Queue Capacity: Configure max_waiting_requests to handle traffic spikes
Connection Management: The ConnectionReset interceptor prevents connection leaks in threaded environments
Resource Monitoring: Use structured logs to monitor resource utilization

Troubleshooting

Common Issues

Server not starting with custom configuration:

# Ensure plugin is installed before configuration
Gruf::Queue::Plugin.install!
# Then configure...

ActiveRecord connection issues:

# Verify ActiveRecord is loaded before gruf-queue
require 'active_record'
require 'gruf-queue'

High memory usage:

# Adjust pool size and keep-alive settings
Gruf.configure do |config|
  config.rpc_server_options = {
    pool_size: 8,           # Reduce if memory constrained
    pool_keep_alive: 60     # Shorter keep-alive
  }
end

Requirements

Ruby >= 3.2.0
Gruf >= 2.21.0
gRPC >= 1.0

Development

After checking out the repo, run bin/setup to install dependencies. Then, run rake spec to run the tests.

# Install dependencies
$ bundle install

# Run tests
$ bundle exec rspec

# Run linting
$ bundle exec rubocop

# Install locally
$ bundle exec rake install

Contributing

Bug reports and pull requests are welcome on GitHub at https://github.com/ether-moon/gruf-queue. This project is intended to be a safe, welcoming space for collaboration.

License

The gem is available as open source under the terms of the MIT License.

gruf-queue

Development

Runtime

Gruf::Queue

Features

Installation

Quick Start

Zero Configuration Setup

Manual Installation Control

Configuration

Server Configuration

Using Server Factory

ActiveRecord Connection Management

Architecture

Core Components

Gruf::Queue::QueuedRpcServer

Gruf::Queue::Pool

Gruf::Queue::Interceptors::ConnectionReset

Gruf::Queue::Plugin

Plugin Lifecycle

Advanced Usage

Custom Server Implementation

Monitoring and Observability

Environment-specific Configuration

Performance Considerations

Troubleshooting

Common Issues

Requirements

Development

Contributing

License

`Gruf::Queue::QueuedRpcServer`

`Gruf::Queue::Pool`

`Gruf::Queue::Interceptors::ConnectionReset`

`Gruf::Queue::Plugin`