--- title: "Real-Time Pub/Sub (LISTEN/NOTIFY): Business Use Case for HeliosDB Nano" description: "Planned use case: PostgreSQL-compatible LISTEN/NOTIFY is not in the current HeliosDB Nano release; this page describes the planned pub/sub design." --- # Real-Time Pub/Sub (LISTEN/NOTIFY): Business Use Case for HeliosDB Nano :::caution[Planned: LISTEN/NOTIFY is not in the current release] `LISTEN`, `UNLISTEN`, `NOTIFY` and `pg_notify` are not implemented in the current HeliosDB Nano release: the statements are rejected, and trigger bodies that would call `NOTIFY` do not run. PostgreSQL-compatible LISTEN/NOTIFY is planned. This page describes the planned design; until it is available, clients must poll for changes. See [Known Limitations](https://www.heliosdb.com/docs/nano/reference/compatibility/known-limitations/). ::: **Document ID**: 12_REALTIME_PUBSUB.md **Created**: 2025-11-30 **Category**: Event-Driven Architecture --- ## Executive Summary PostgreSQL-compatible LISTEN/NOTIFY pub/sub is planned for HeliosDB Nano; it is not in the current release. The planned design targets in-process event delivery across many concurrent channels without external dependencies. With thread-safe in-memory channel management, automatic subscription cleanup, and 8KB payload support, the planned design would let event-driven applications implement real-time dashboards, cache invalidation, activity feeds, workflow automation, and microservices coordination entirely within embedded, edge, and microservice deployments without Redis, Kafka, or message queue infrastructure. The planned zero-external-dependency design would avoid network hops for local notifications and the cost of running a separate broker, and would enable offline-first event-driven applications for IoT devices, collaborative tools, and real-time analytics platforms. --- ## Problem Being Solved ### Core Problem Statement Modern applications require real-time event notification for responsive user interfaces, cache coherence, workflow orchestration, and microservices coordination, but existing solutions force teams to choose between heavyweight message brokers (Redis Pub/Sub, Kafka, RabbitMQ) requiring separate infrastructure and network overhead, or inefficient polling-based architectures that waste CPU cycles and increase latency. Teams deploying to edge devices, embedded systems, or resource-constrained environments cannot afford external message queue dependencies, yet lack lightweight pub/sub for event-driven patterns. ### Root Cause Analysis | Factor | Impact | Current Workaround | Limitation | |--------|--------|-------------------|------------| | **External Message Broker Dependency** | 5-50ms network latency per message, $100-1000/month infrastructure cost, complex deployment | Deploy Redis Pub/Sub, Kafka, or RabbitMQ as separate service | Requires network connectivity, adds operational complexity, unsuitable for embedded/edge deployments, single point of failure | | **Database Polling Anti-Pattern** | High CPU usage (10-30% wasted on polling), 100-1000ms notification delay, increased database load | Poll database every 100ms-1s with `SELECT * FROM events WHERE timestamp > last_check` | Wastes resources, poor scalability, delayed notifications, database contention, 10-100x slower than push-based | | **PostgreSQL Server Requirement** | 200MB+ memory overhead, complex deployment, requires client-server architecture | Deploy full PostgreSQL server for LISTEN/NOTIFY | 500MB+ footprint, unsuitable for embedded systems, client-server latency penalty, complex connection management | | **WebSocket/Long-Polling Overhead** | Requires separate HTTP server, complex state management, high memory per connection | Implement custom WebSocket server or long-polling endpoints | 1-5MB memory per connection, stateful server complexity, doesn't integrate with database transactions | | **In-Memory Event Bus Limitations** | No persistence, no integration with database transactions, manual delivery guarantees | Use in-app event emitters (EventEmitter, Tokio broadcast channels) | No transaction integration, lost messages on crash, no database trigger support, manual coordination | ### Business Impact (design targets) The "With HeliosDB Nano" column states design targets for the planned feature; none of these figures has been measured. | Metric | Without HeliosDB Nano | With HeliosDB Nano (target) | Improvement (target) | |--------|----------------------|-------------------|------------| | **Notification Latency (in-process)** | 5-50ms (external broker) + network | <1ms (in-memory) | 5-50x faster | | **Infrastructure Cost** | $100-1000/month (Redis/Kafka) | $0 (embedded) | 100% reduction | | **Memory Footprint (1000 channels)** | 200MB+ (Redis) or 500MB+ (Postgres) | <10MB (in-memory) | 20-50x reduction | | **Deployment Complexity** | 3-5 services (message broker, DB, app) | Single binary | 70% simpler | | **CPU Overhead (polling replaced)** | 10-30% wasted on polling | <1% (event-driven) | 10-30x more efficient | | **Edge Device Viability** | Impossible (requires external broker) | Full support (Raspberry Pi+) | Enables new markets | ### Who Suffers Most 1. **Real-Time Dashboard Teams**: Building analytics dashboards that update live on data changes, forced to poll databases every second, wasting 20-40% CPU and experiencing 500ms-2s update delays when sub-second responsiveness is critical for user experience. 2. **Cache Invalidation Engineers**: Managing cache coherence across distributed application instances without Redis Pub/Sub infrastructure, resulting in stale cache issues, manual cache TTL management, or expensive cache-aside patterns with 10-100x more cache misses than necessary. 3. **IoT/Edge Application Developers**: Building collaborative tools or real-time monitoring on edge devices where external message brokers are unavailable, forcing them to implement inefficient polling or abandon real-time features entirely, degrading user experience significantly. --- ## Why Competitors Cannot Solve This ### Technical Barriers | Competitor Category | Limitation | Root Cause | Time to Match | |--------------------|------------|------------|---------------| | **SQLite, DuckDB** | No pub/sub or notification mechanism | Designed for single-process analytical workloads; no multi-client notification infrastructure | 12-18 months | | **Redis Pub/Sub** | Requires separate server, network overhead, no database transaction integration | Cloud/server-first architecture designed for distributed systems, not embedded use cases | Never (contradicts embedded model) | | **PostgreSQL LISTEN/NOTIFY** | Requires full Postgres server (500MB+ footprint), client-server architecture | Full RDBMS designed for client-server deployment, not in-process embedding | 6-12 months for embedded variant | | **Kafka, RabbitMQ** | Heavy infrastructure (1GB+ memory), complex deployment, designed for durability not speed | Enterprise message broker architecture prioritizing durability over embedded simplicity | Never (enterprise-focused) | | **WebSocket Servers** | Separate HTTP infrastructure, no database integration, stateful connection management | Application-layer protocol disconnected from database layer | 18-24 months to add DB integration | ### Architecture Requirements To match HeliosDB Nano's real-time pub/sub capabilities, competitors would need: 1. **In-Process Thread-Safe Channel Management**: Build concurrent hash map for channel subscriptions with read-write locks, UUID-based subscription tracking, and automatic cleanup on connection drop, requiring deep understanding of Rust concurrency primitives (Arc, RwLock) and lock-free data structures to avoid deadlocks while supporting thousands of channels. 2. **Integration with Database Transaction Lifecycle**: Implement NOTIFY delivery that respects transaction boundaries (only deliver after COMMIT, not on ROLLBACK), coordinate with SQL executor to trigger notifications on database mutations, and handle trigger-based notifications, requiring integration with transaction state machine and SQL parser. 3. **PostgreSQL Wire Protocol Compatibility**: Implement asynchronous notification messages in extended query protocol, support LISTEN/UNLISTEN SQL commands, handle notification delivery during query idle periods, and maintain compatibility with PostgreSQL client libraries (psycopg2, pg, node-postgres), requiring protocol-level implementation expertise. ## HeliosDB Nano Solution ### Architecture Overview ``` ┌─────────────────────────────────────────────────────────────┐ │ Application Code (Multi-threaded) │ ├──────────────────┬──────────────────┬───────────────────────┤ │ LISTEN Thread │ NOTIFY Thread │ Business Logic Thread │ │ (Subscribe) │ (Publish) │ (SQL Queries) │ ├──────────────────┴──────────────────┴───────────────────────┤ │ PubSubManager (Thread-Safe) │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ Channel → Subscribers (HashMap) │ │ │ │ Subscription → Notification Queue (HashMap) │ │ │ │ Arc for concurrent read/write │ │ │ └─────────────────────────────────────────────────────┘ │ ├──────────────────────────────────────────────────────────────┤ │ SQL Executor (Transaction Integration) │ │ - LISTEN/UNLISTEN command parsing │ │ - NOTIFY command execution │ │ - Database trigger → NOTIFY integration │ ├──────────────────────────────────────────────────────────────┤ │ HeliosDB Nano Storage Engine │ └──────────────────────────────────────────────────────────────┘ In-Memory Notification Flow: 1. Thread A: LISTEN 'orders' → Subscribe to 'orders' channel 2. Thread B: NOTIFY 'orders', 'payload' → Enqueue notification 3. Thread A: Poll subscription → Receive notification (<1ms) ``` ### Key Capabilities | Capability | Description | Design target (not measured) | |------------|-------------|-------------| | **PostgreSQL-Compatible LISTEN/NOTIFY** | SQL commands for subscribing (`LISTEN channel`) and publishing (`NOTIFY channel, 'payload'`) with identical syntax to PostgreSQL | 100% PostgreSQL syntax compatibility | | **In-Memory Channel Management** | Thread-safe concurrent HashMap for channels and subscriptions using Arc for lock-free reads | Thousands of concurrent channels, <10MB overhead | | **Sub-Millisecond Notification Delivery** | In-process notification queue delivery without network overhead or serialization | <1ms P99 latency for in-process delivery | | **Automatic Subscription Cleanup** | RAII-based subscription handles with Drop trait implementation for automatic UNLISTEN | Zero memory leaks, automatic resource management | | **8KB Payload Support** | Notification payloads up to 8000 bytes (PostgreSQL limit) for rich event data (JSON, binary) | Full PostgreSQL payload compatibility | | **Multiple Subscribers per Channel** | Broadcast notifications to all subscribers on a channel (fan-out pattern) | O(N) delivery to N subscribers | | **Transaction Integration** | NOTIFY respects transaction boundaries (only delivered after COMMIT, not on ROLLBACK) | ACID-compliant event delivery | | **Database Trigger Support** | Trigger functions can call NOTIFY for automatic event generation on INSERT/UPDATE/DELETE | Zero-code change detection | --- ## Concrete Examples with Code, Config & Architecture ### Example 1: Real-Time Dashboard Updates - Embedded Configuration **Scenario**: Analytics dashboard displaying live sales metrics for 1000 concurrent users, requiring sub-second updates when new orders arrive without polling database every 100ms (which wastes 20% CPU). **Architecture**: ``` Web Browser (WebSocket) ↓ Web Server (Axum/Actix) ↓ HeliosDB Nano (LISTEN 'sales_updates') ↓ In-Process Notification Queue ↓ Database Trigger on INSERT → NOTIFY 'sales_updates' ``` **Configuration** (`heliosdb.toml`): ```toml # HeliosDB Nano configuration for real-time dashboard [database] path = "/var/lib/heliosdb/sales.db" memory_limit_mb = 512 enable_wal = true [pubsub] # Enable pub/sub with optimized settings enabled = true max_channels = 1000 max_payload_bytes = 8000 notification_queue_size = 100 [monitoring] metrics_enabled = true verbose_logging = false ``` **Implementation Code** (Rust with Axum WebSocket): ```rust use heliosdb_nano::{Connection, protocols::PubSubManager}; use axum::{ extract::{ws::WebSocket, State, WebSocketUpgrade}, response::Response, routing::get, Router, }; use std::sync::Arc; use tokio::sync::broadcast; #[derive(Clone)] struct AppState { db: Arc, pubsub: Arc, broadcast_tx: broadcast::Sender, } // Initialize database with trigger for automatic notifications fn init_dashboard_db(conn: &Connection) -> Result<(), Box> { // Create sales table conn.execute( "CREATE TABLE IF NOT EXISTS sales_orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id INTEGER NOT NULL, amount DECIMAL(10,2) NOT NULL, created_at INTEGER DEFAULT (strftime('%s', 'now')) )", [], )?; // Create trigger to automatically NOTIFY on new sales conn.execute( "CREATE TRIGGER IF NOT EXISTS sales_order_notify AFTER INSERT ON sales_orders BEGIN SELECT NOTIFY('sales_updates', json_object( 'order_id', NEW.id, 'customer_id', NEW.customer_id, 'amount', NEW.amount, 'timestamp', NEW.created_at ) ); END", [], )?; Ok(()) } // Background task that listens for database notifications and broadcasts to WebSocket clients async fn notification_broadcaster( pubsub: Arc, tx: broadcast::Sender, ) -> Result<(), Box> { // Subscribe to sales_updates channel let subscription = pubsub.subscribe("sales_updates")?; loop { // Poll for new notifications (non-blocking) let notifications = subscription.poll()?; for notification in notifications { // Broadcast to all WebSocket clients let _ = tx.send(notification.payload); } // Small delay to batch notifications tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; } } // WebSocket handler for individual client connections async fn websocket_handler( ws: WebSocketUpgrade, State(state): State, ) -> Response { ws.on_upgrade(|socket| handle_websocket(socket, state)) } async fn handle_websocket(mut socket: WebSocket, state: AppState) { // Subscribe to broadcast channel let mut rx = state.broadcast_tx.subscribe(); // Forward notifications to WebSocket client while let Ok(message) = rx.recv().await { if socket.send(axum::extract::ws::Message::Text(message)).await.is_err() { break; } } } // API endpoint to insert new sales orders (triggers notification) async fn create_order( State(state): State, axum::Json(order): axum::Json, ) -> Result { state.db.execute( "INSERT INTO sales_orders (customer_id, amount) VALUES (?1, ?2)", [ order["customer_id"].as_i64().unwrap().to_string(), order["amount"].as_f64().unwrap().to_string(), ], ).map_err(|e| e.to_string())?; Ok("Order created".to_string()) } #[tokio::main] async fn main() -> Result<(), Box> { // Initialize database let conn = Arc::new(Connection::open("sales.db")?); init_dashboard_db(&conn)?; // Initialize pub/sub manager let pubsub = Arc::new(PubSubManager::new()); // Create broadcast channel for WebSocket fanout let (broadcast_tx, _) = broadcast::channel(1000); let state = AppState { db: conn, pubsub: Arc::clone(&pubsub), broadcast_tx: broadcast_tx.clone(), }; // Start notification broadcaster task let broadcaster_pubsub = Arc::clone(&pubsub); tokio::spawn(async move { notification_broadcaster(broadcaster_pubsub, broadcast_tx).await }); // Build web server let app = Router::new() .route("/ws", get(websocket_handler)) .route("/orders", axum::routing::post(create_order)) .with_state(state); // Start server let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?; println!("Dashboard server running on http://0.0.0.0:8080"); axum::serve(listener, app).await?; Ok(()) } ``` **Design targets (not measured; feature planned)**: | Metric | Before (Polling) | After (LISTEN/NOTIFY) | Improvement | |--------|------------------|-----------------------|-------------| | **Update Latency** | 500-1000ms (polling interval) | <50ms (event-driven) | 10-20x faster | | **CPU Usage (idle)** | 20% (constant polling) | <1% (event-driven) | 20x more efficient | | **Database Load** | 10 queries/sec per client | 0 queries/sec (notifications only) | 100% reduction | | **Scalability** | 100 concurrent clients max | 1000+ concurrent clients | 10x improvement | | **Infrastructure Cost** | $200/month (Redis + DB) | $50/month (DB only) | 75% reduction | --- ### Example 2: Cache Invalidation - Python Integration **Scenario**: Multi-instance web application with in-memory caches that must invalidate when source data changes, without polling or manual TTL management. **Python Client Code**: ```python import heliosdb_nano from heliosdb_nano import Connection, PubSubManager import asyncio import json from typing import Dict, Any import threading class CacheInvalidationManager: """Manages cache with automatic invalidation via LISTEN/NOTIFY.""" def __init__(self, db_path: str): self.conn = Connection.open(db_path) self.pubsub = PubSubManager() self.cache: Dict[str, Any] = {} self.lock = threading.Lock() def setup_schema(self): """Initialize database schema with invalidation triggers.""" # Create products table self.conn.execute(""" CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, price DECIMAL(10,2) NOT NULL, updated_at INTEGER DEFAULT (strftime('%s', 'now')) ) """) # Create trigger to notify on UPDATE self.conn.execute(""" CREATE TRIGGER IF NOT EXISTS product_update_notify AFTER UPDATE ON products BEGIN SELECT NOTIFY('cache_invalidate', 'product:' || NEW.id ); END """) # Create trigger to notify on DELETE self.conn.execute(""" CREATE TRIGGER IF NOT EXISTS product_delete_notify AFTER DELETE ON products BEGIN SELECT NOTIFY('cache_invalidate', 'product:' || OLD.id ); END """) async def start_invalidation_listener(self): """Background task that listens for cache invalidation events.""" # Subscribe to cache invalidation channel subscription = self.pubsub.subscribe('cache_invalidate') while True: # Poll for notifications notifications = subscription.poll() for notification in notifications: cache_key = notification.payload # Invalidate cache entry with self.lock: if cache_key in self.cache: del self.cache[cache_key] print(f"Cache invalidated: {cache_key}") # Poll every 10ms await asyncio.sleep(0.01) def get_product(self, product_id: int) -> Dict[str, Any]: """Get product with caching.""" cache_key = f"product:{product_id}" # Check cache first with self.lock: if cache_key in self.cache: print(f"Cache HIT: {cache_key}") return self.cache[cache_key] # Cache miss - query database print(f"Cache MISS: {cache_key}") cursor = self.conn.cursor() cursor.execute( "SELECT id, name, price FROM products WHERE id = ?", (product_id,) ) row = cursor.fetchone() if row: product = { 'id': row[0], 'name': row[1], 'price': row[2] } # Store in cache with self.lock: self.cache[cache_key] = product return product return None def update_product(self, product_id: int, name: str, price: float): """Update product (triggers cache invalidation via NOTIFY).""" self.conn.execute( "UPDATE products SET name = ?, price = ?, updated_at = strftime('%s', 'now') WHERE id = ?", (name, price, product_id) ) # Trigger fires automatically, sending NOTIFY # Usage example async def main(): manager = CacheInvalidationManager("products.db") manager.setup_schema() # Insert test data manager.conn.execute( "INSERT OR REPLACE INTO products (id, name, price) VALUES (1, 'Widget', 9.99)" ) # Start background invalidation listener asyncio.create_task(manager.start_invalidation_listener()) # First access - cache miss product = manager.get_product(1) # Cache MISS print(f"Product: {product}") # Second access - cache hit product = manager.get_product(1) # Cache HIT print(f"Product: {product}") # Update product (triggers invalidation) manager.update_product(1, 'Super Widget', 19.99) # Wait for invalidation to propagate await asyncio.sleep(0.1) # Third access - cache miss after invalidation product = manager.get_product(1) # Cache MISS (invalidated) print(f"Product: {product}") # Keep running await asyncio.sleep(3600) if __name__ == "__main__": asyncio.run(main()) ``` **Architecture Pattern**: ``` ┌─────────────────────────────────────────┐ │ Application Instance 1 │ │ ┌────────────────┐ ┌────────────────┐ │ │ │ In-Memory Cache│ │ LISTEN Thread │ │ │ └────────────────┘ └────────────────┘ │ ├─────────────────────────────────────────┤ │ Application Instance 2 │ │ ┌────────────────┐ ┌────────────────┐ │ │ │ In-Memory Cache│ │ LISTEN Thread │ │ │ └────────────────┘ └────────────────┘ │ ├─────────────────────────────────────────┤ │ HeliosDB Nano (Shared) │ │ Database Trigger → NOTIFY all listeners │ └─────────────────────────────────────────┘ Flow: 1. Instance 1: UPDATE products SET price = 19.99 WHERE id = 1 2. Database Trigger: NOTIFY 'cache_invalidate', 'product:1' 3. All Instances: Receive notification, invalidate local cache 4. Next read: Cache miss, fetch fresh data ``` **Design targets (not measured; feature planned)**: - Cache hit rate: 95% (vs 60% with TTL-based invalidation) - Stale cache duration: <50ms (vs 30-60s with TTL) - Cache memory efficiency: 80% reduction (no defensive TTL padding) - Infrastructure cost: $0 (no Redis required) --- ### Example 3: Activity Feed - Microservices Integration (Go/Rust) **Scenario**: Social platform with 10,000 users, displaying real-time activity feed updates (likes, comments, follows) without polling every user's feed. **Rust Service Code** (`src/activity_service.rs`): ```rust use axum::{ extract::{Path, State}, http::StatusCode, routing::{get, post}, Json, Router, }; use heliosdb_nano::{Connection, protocols::PubSubManager}; use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::broadcast; #[derive(Clone)] pub struct ActivityService { db: Arc, pubsub: Arc, } #[derive(Debug, Serialize, Deserialize)] pub struct Activity { id: i64, user_id: i64, action_type: String, // "like", "comment", "follow" target_id: i64, timestamp: i64, } impl ActivityService { pub fn new(db_path: &str) -> Result> { let db = Arc::new(Connection::open(db_path)?); let pubsub = Arc::new(PubSubManager::new()); // Initialize schema with triggers Self::init_schema(&db)?; Ok(Self { db, pubsub }) } fn init_schema(conn: &Connection) -> Result<(), Box> { // Create activities table conn.execute( "CREATE TABLE IF NOT EXISTS activities ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, action_type TEXT NOT NULL, target_id INTEGER NOT NULL, timestamp INTEGER DEFAULT (strftime('%s', 'now')) )", [], )?; // Create trigger to notify followers on new activity conn.execute( "CREATE TRIGGER IF NOT EXISTS activity_notify AFTER INSERT ON activities BEGIN SELECT NOTIFY( 'activity_feed_' || NEW.user_id, json_object( 'id', NEW.id, 'user_id', NEW.user_id, 'action_type', NEW.action_type, 'target_id', NEW.target_id, 'timestamp', NEW.timestamp ) ); END", [], )?; Ok(()) } // Record a new activity (automatically triggers notification) pub fn record_activity( &self, user_id: i64, action_type: &str, target_id: i64, ) -> Result> { let mut stmt = self.db.prepare( "INSERT INTO activities (user_id, action_type, target_id) VALUES (?1, ?2, ?3) RETURNING id, user_id, action_type, target_id, timestamp" )?; let activity = stmt.query_row( [user_id.to_string(), action_type.to_string(), target_id.to_string()], |row| { Ok(Activity { id: row.get(0)?, user_id: row.get(1)?, action_type: row.get(2)?, target_id: row.get(3)?, timestamp: row.get(4)?, }) }, )?; Ok(activity) } // Subscribe to a user's activity feed pub fn subscribe_to_feed(&self, user_id: i64) -> Result, Box> { let channel = format!("activity_feed_{}", user_id); let subscription = self.pubsub.subscribe(&channel)?; Ok(subscription) } } // HTTP handlers async fn post_activity( State(service): State>, Json(req): Json, ) -> (StatusCode, Json) { let user_id = req["user_id"].as_i64().unwrap(); let action_type = req["action_type"].as_str().unwrap(); let target_id = req["target_id"].as_i64().unwrap(); let activity = service.record_activity(user_id, action_type, target_id).unwrap(); (StatusCode::CREATED, Json(activity)) } async fn get_activity_stream( Path(user_id): Path, State(service): State>, ) -> Result>>, StatusCode> { use futures::stream::StreamExt; // Subscribe to user's activity feed let subscription = service.subscribe_to_feed(user_id) .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; // Create SSE stream let stream = futures::stream::unfold(subscription, |sub| async move { loop { tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; if let Ok(notifications) = sub.poll() { for notification in notifications { let event = axum::response::sse::Event::default() .data(notification.payload); return Some((Ok(event), sub)); } } } }); Ok(axum::response::sse::Sse::new(stream)) } pub fn create_router(service: Arc) -> Router { Router::new() .route("/activities", post(post_activity)) .route("/activities/stream/:user_id", get(get_activity_stream)) .with_state(service) } #[tokio::main] async fn main() -> Result<(), Box> { let service = Arc::new(ActivityService::new("activities.db")?); let app = create_router(service); let listener = tokio::net::TcpListener::bind("0.0.0.0:8080").await?; println!("Activity service running on http://0.0.0.0:8080"); axum::serve(listener, app).await?; Ok(()) } ``` **Service Architecture**: ``` ┌─────────────────────────────────────────┐ │ Client (Browser / Mobile App) │ │ Server-Sent Events (SSE) Stream │ ├─────────────────────────────────────────┤ │ Activity Microservice (Rust) │ │ ┌────────────────────────────────────┐ │ │ │ POST /activities │ │ │ │ ↓ │ │ │ │ INSERT INTO activities │ │ │ │ ↓ │ │ │ │ Trigger fires NOTIFY │ │ │ └────────────────────────────────────┘ │ │ ┌────────────────────────────────────┐ │ │ │ GET /activities/stream/:user_id │ │ │ │ ↓ │ │ │ │ LISTEN 'activity_feed_123' │ │ │ │ ↓ │ │ │ │ Poll subscription → SSE stream │ │ │ └────────────────────────────────────┘ │ ├─────────────────────────────────────────┤ │ HeliosDB Nano (In-Process) │ │ PubSubManager → Notification Delivery │ └─────────────────────────────────────────┘ ``` **Design targets (not measured; feature planned)**: - Real-time update latency: <100ms (vs 1-5s with polling) - Memory per connection: 10KB (vs 1MB with WebSocket state) - CPU usage: <5% for 1000 concurrent streams - Infrastructure: Single service (no separate message broker) --- ### Example 4: Workflow Automation - Docker/Container Deployment **Scenario**: Order processing workflow where each step triggers the next automatically via database events, eliminating manual coordination. **Docker Deployment** (`Dockerfile`): ```dockerfile FROM rust:latest as builder WORKDIR /app COPY . . RUN cargo build --release FROM debian:bookworm-slim RUN apt-get update && apt-get install -y \ ca-certificates \ && rm -rf /var/lib/apt/lists/* COPY --from=builder /app/target/release/workflow-engine /usr/local/bin/ RUN mkdir -p /data EXPOSE 8080 VOLUME ["/data"] ENTRYPOINT ["workflow-engine"] CMD ["--config", "/etc/workflow/config.toml"] ``` **Docker Compose** (`docker-compose.yml`): ```yaml version: '3.8' services: workflow-engine: build: context: . dockerfile: Dockerfile image: workflow-engine:latest ports: - "8080:8080" volumes: - ./data:/data - ./config/workflow.toml:/etc/workflow/config.toml:ro environment: RUST_LOG: "workflow_engine=info" HELIOSDB_DATA_DIR: "/data" restart: unless-stopped healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/health"] interval: 30s timeout: 3s retries: 3 networks: default: driver: bridge ``` **Workflow Configuration** (`config/workflow.toml`): ```toml [database] path = "/data/workflow.db" memory_limit_mb = 256 enable_wal = true [pubsub] enabled = true max_channels = 100 [workflow] # Workflow step definitions [workflow.steps.payment_processing] trigger_channel = "order_created" action = "process_payment" next_channel = "payment_completed" [workflow.steps.inventory_allocation] trigger_channel = "payment_completed" action = "allocate_inventory" next_channel = "inventory_allocated" [workflow.steps.shipping] trigger_channel = "inventory_allocated" action = "create_shipment" next_channel = "shipment_created" ``` **Workflow Engine Code** (`src/workflow.rs`): ```rust use heliosdb_nano::{Connection, protocols::PubSubManager}; use std::sync::Arc; use tokio::task::JoinSet; pub struct WorkflowEngine { db: Arc, pubsub: Arc, } impl WorkflowEngine { pub fn new(db_path: &str) -> Result> { let db = Arc::new(Connection::open(db_path)?); let pubsub = Arc::new(PubSubManager::new()); Self::init_schema(&db)?; Ok(Self { db, pubsub }) } fn init_schema(conn: &Connection) -> Result<(), Box> { // Orders table conn.execute( "CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id INTEGER NOT NULL, amount DECIMAL(10,2) NOT NULL, status TEXT DEFAULT 'pending', created_at INTEGER DEFAULT (strftime('%s', 'now')) )", [], )?; // Workflow events table conn.execute( "CREATE TABLE IF NOT EXISTS workflow_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, event_type TEXT NOT NULL, payload TEXT, created_at INTEGER DEFAULT (strftime('%s', 'now')) )", [], )?; // Trigger on order creation conn.execute( "CREATE TRIGGER IF NOT EXISTS order_created_trigger AFTER INSERT ON orders BEGIN INSERT INTO workflow_events (order_id, event_type, payload) VALUES (NEW.id, 'order_created', json_object('order_id', NEW.id, 'amount', NEW.amount) ); SELECT NOTIFY('order_created', json_object('order_id', NEW.id, 'amount', NEW.amount) ); END", [], )?; Ok(()) } // Start workflow step listeners pub async fn start(self: Arc) -> Result<(), Box> { let mut tasks = JoinSet::new(); // Step 1: Listen for new orders, process payment let engine = Arc::clone(&self); tasks.spawn(async move { engine.payment_processor().await }); // Step 2: Listen for payment completion, allocate inventory let engine = Arc::clone(&self); tasks.spawn(async move { engine.inventory_allocator().await }); // Step 3: Listen for inventory allocation, create shipment let engine = Arc::clone(&self); tasks.spawn(async move { engine.shipment_creator().await }); // Wait for all tasks while let Some(result) = tasks.join_next().await { result??; } Ok(()) } async fn payment_processor(&self) -> Result<(), Box> { let subscription = self.pubsub.subscribe("order_created")?; loop { let notifications = subscription.poll()?; for notification in notifications { let payload: serde_json::Value = serde_json::from_str(¬ification.payload)?; let order_id = payload["order_id"].as_i64().unwrap(); println!("Processing payment for order {}", order_id); // Simulate payment processing tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; // Update order status self.db.execute( "UPDATE orders SET status = 'payment_completed' WHERE id = ?1", [order_id.to_string()], )?; // Notify next step self.pubsub.notify( "payment_completed", &serde_json::json!({"order_id": order_id}).to_string(), )?; } tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; } } async fn inventory_allocator(&self) -> Result<(), Box> { let subscription = self.pubsub.subscribe("payment_completed")?; loop { let notifications = subscription.poll()?; for notification in notifications { let payload: serde_json::Value = serde_json::from_str(¬ification.payload)?; let order_id = payload["order_id"].as_i64().unwrap(); println!("Allocating inventory for order {}", order_id); // Simulate inventory allocation tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; // Update order status self.db.execute( "UPDATE orders SET status = 'inventory_allocated' WHERE id = ?1", [order_id.to_string()], )?; // Notify next step self.pubsub.notify( "inventory_allocated", &serde_json::json!({"order_id": order_id}).to_string(), )?; } tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; } } async fn shipment_creator(&self) -> Result<(), Box> { let subscription = self.pubsub.subscribe("inventory_allocated")?; loop { let notifications = subscription.poll()?; for notification in notifications { let payload: serde_json::Value = serde_json::from_str(¬ification.payload)?; let order_id = payload["order_id"].as_i64().unwrap(); println!("Creating shipment for order {}", order_id); // Simulate shipment creation tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; // Update order status self.db.execute( "UPDATE orders SET status = 'shipped' WHERE id = ?1", [order_id.to_string()], )?; println!("Order {} workflow complete!", order_id); } tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; } } } #[tokio::main] async fn main() -> Result<(), Box> { let engine = Arc::new(WorkflowEngine::new("/data/workflow.db")?); // Start workflow engine engine.start().await?; Ok(()) } ``` **Workflow Architecture**: ``` Order Created (INSERT) ↓ Trigger fires NOTIFY 'order_created' ↓ Payment Processor (LISTEN 'order_created') ↓ (process payment) ↓ NOTIFY 'payment_completed' ↓ Inventory Allocator (LISTEN 'payment_completed') ↓ (allocate inventory) ↓ NOTIFY 'inventory_allocated' ↓ Shipment Creator (LISTEN 'inventory_allocated') ↓ (create shipment) ↓ Order Complete ``` **Design targets (not measured; feature planned)**: - Workflow latency: 200-300ms end-to-end - No external message broker required - Automatic retry on failure (reprocess on startup) - Container deployment: Single service --- ### Example 5: Microservices Event Bus - Edge/IoT Deployment **Scenario**: IoT sensor network with 100 edge devices coordinating via lightweight event bus for distributed alerting without cloud connectivity. **Edge Device Configuration** (`edge-config.toml`): ```toml [database] path = "/var/lib/iot/sensors.db" memory_limit_mb = 64 # Low memory for edge devices page_size = 512 enable_wal = true [pubsub] enabled = true max_channels = 50 max_payload_bytes = 1024 # Smaller payloads for IoT [edge] device_id = "sensor_001" location = "warehouse_a" ``` **Edge Device Application** (Rust): ```rust use heliosdb_nano::{Connection, protocols::PubSubManager}; use std::sync::Arc; use serde::{Deserialize, Serialize}; #[derive(Debug, Serialize, Deserialize)] struct SensorReading { device_id: String, sensor_type: String, value: f64, timestamp: i64, alert: bool, } pub struct EdgeSensorNode { db: Arc, pubsub: Arc, device_id: String, } impl EdgeSensorNode { pub fn new(device_id: String) -> Result> { let db = Arc::new(Connection::open("/var/lib/iot/sensors.db")?); let pubsub = Arc::new(PubSubManager::new()); Self::init_schema(&db)?; Ok(Self { db, pubsub, device_id }) } fn init_schema(conn: &Connection) -> Result<(), Box> { conn.execute( "CREATE TABLE IF NOT EXISTS sensor_readings ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_id TEXT NOT NULL, sensor_type TEXT NOT NULL, value REAL NOT NULL, alert BOOLEAN DEFAULT 0, timestamp INTEGER DEFAULT (strftime('%s', 'now')) )", [], )?; // Trigger for critical alerts conn.execute( "CREATE TRIGGER IF NOT EXISTS critical_alert_trigger AFTER INSERT ON sensor_readings WHEN NEW.alert = 1 BEGIN SELECT NOTIFY('critical_alerts', json_object( 'device_id', NEW.device_id, 'sensor_type', NEW.sensor_type, 'value', NEW.value, 'timestamp', NEW.timestamp ) ); END", [], )?; Ok(()) } // Record sensor reading with automatic alert detection pub fn record_reading( &self, sensor_type: &str, value: f64, threshold: f64, ) -> Result<(), Box> { let alert = value > threshold; self.db.execute( "INSERT INTO sensor_readings (device_id, sensor_type, value, alert) VALUES (?1, ?2, ?3, ?4)", [ &self.device_id, sensor_type, &value.to_string(), if alert { "1" } else { "0" }, ], )?; // Trigger fires automatically if alert = 1 Ok(()) } // Listen for alerts from other devices in the network pub async fn alert_listener(&self) -> Result<(), Box> { let subscription = self.pubsub.subscribe("critical_alerts")?; loop { let notifications = subscription.poll()?; for notification in notifications { let alert: serde_json::Value = serde_json::from_str(¬ification.payload)?; println!( "ALERT from {}: {} = {} (threshold exceeded)", alert["device_id"], alert["sensor_type"], alert["value"] ); // Take local action (e.g., activate safety system) self.handle_alert(&alert)?; } tokio::time::sleep(tokio::time::Duration::from_millis(10)).await; } } fn handle_alert(&self, alert: &serde_json::Value) -> Result<(), Box> { // Local alert handling logic // Example: Log alert, activate local safety measures println!("Local device {} taking action on alert", self.device_id); Ok(()) } } #[tokio::main] async fn main() -> Result<(), Box> { let node = Arc::new(EdgeSensorNode::new("sensor_001".to_string())?); // Start alert listener let alert_node = Arc::clone(&node); tokio::spawn(async move { alert_node.alert_listener().await }); // Simulate sensor readings loop { let temperature = 20.0 + (rand::random::() * 10.0); // Record reading (triggers alert if temperature > 25.0) node.record_reading("temperature", temperature, 25.0)?; tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } } ``` **Edge Network Architecture**: ``` ┌────────────────────┐ ┌────────────────────┐ ┌────────────────────┐ │ Edge Device 1 │ │ Edge Device 2 │ │ Edge Device 3 │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ │ Sensor Data │ │ │ │ Sensor Data │ │ │ │ Sensor Data │ │ │ └──────────────┘ │ │ └──────────────┘ │ │ └──────────────┘ │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ │ HeliosDB Nano│ │ │ │ HeliosDB Nano│ │ │ │ HeliosDB Nano│ │ │ │ + PubSub │ │ │ │ + PubSub │ │ │ │ + PubSub │ │ │ └──────────────┘ │ │ └──────────────┘ │ │ └──────────────┘ │ │ LISTEN 'alerts' │ │ LISTEN 'alerts' │ │ LISTEN 'alerts' │ │ NOTIFY on alert │ │ NOTIFY on alert │ │ NOTIFY on alert │ └────────────────────┘ └────────────────────┘ └────────────────────┘ │ │ │ └──────────────────────┴──────────────────────┘ Local Event Propagation (in-memory, sub-millisecond) ``` **Design targets (not measured; feature planned)**: - Alert propagation: <1ms within process - Memory footprint: 32MB per device - No cloud dependency (fully offline) - Scales to hundreds of devices in local network --- ## Technical Advantages ### How the Planned Design Compares The HeliosDB Nano column describes the planned design. Its latency and memory cells are design targets, not measurements. | Aspect | HeliosDB Nano (planned) | Redis Pub/Sub | PostgreSQL LISTEN/NOTIFY | Kafka/RabbitMQ | |--------|---------------|---------------|-------------------------|----------------| | **Notification Latency** | In-process delivery, no network hop (target) | 5-20ms (network) | 10-50ms (network) | 50-200ms (network) | | **Memory Footprint** | In-memory channel registry (target: small, per subscription) | 200MB+ | 500MB+ | 1GB+ | | **Deployment Complexity** | Single binary | Separate service | Separate service | Multiple services | | **Infrastructure Cost** | $0 (embedded) | $100-500/month | $50-200/month | $500-2000/month | | **Offline Capability** | Full support | No | No | No | | **Database Integration** | Native (triggers) | Manual | Native (triggers) | Manual | | **Transaction Support** | ACID-compliant | No | ACID-compliant | No | | **Edge Device Viability** | Yes (planned) | No | No | No | --- ## Key Success Metrics ### Technical KPIs | Metric | Target | Measurement Method | |--------|--------|-------------------| | **Notification Latency (P99)** | <5ms | Prometheus metrics: `histogram(notification_delivery_duration_ms)` | | **Notification Throughput** | 10K+ msgs/sec | Prometheus counter: `rate(notifications_delivered_total[1m])` | | **Memory Overhead** | <10MB for 1000 channels | Process RSS measurement: `process.memory.rss` | | **Subscription Lifecycle** | Zero leaks | Monitor subscription count: `active_subscriptions` should match `LISTEN` commands | | **Delivery Reliability** | 100% | Compare `notifications_sent` to `notifications_received` | ### Business KPIs | Metric | Target | Measurement Method | |--------|--------|-------------------| | **Infrastructure Cost Reduction** | 80-95% | Monthly bill comparison: (Redis cost before - $0 after) / Redis cost before | | **CPU Efficiency Gain** | 20-40% reduction | Application CPU metrics: (polling CPU - event-driven CPU) / polling CPU | | **Real-Time Feature Latency** | <100ms end-to-end | User-facing metrics: time from data change to UI update | | **Developer Productivity** | 2-5x faster feature development | Time to implement real-time feature: polling vs LISTEN/NOTIFY | | **Operational Complexity** | 70% reduction | Count of deployed services: before (DB + Redis + app) vs after (app only) | --- ## Conclusion Real-time event-driven applications, from live dashboards and collaborative tools to cache invalidation and workflow automation, usually choose between an external message broker (Redis, Kafka) that adds a network hop and another service to operate, and polling that spends CPU and delays updates. PostgreSQL-compatible LISTEN/NOTIFY is planned for HeliosDB Nano to offer a third option: in-process notifications with no external dependency, delivered only after `COMMIT`. It is not in the current release, and the figures on this page are design targets, not measurements. **Next Steps**: until LISTEN/NOTIFY is available, poll for changes, for example with an indexed `updated_at` or sequence column that clients read past their last seen value, and keep an external broker where you need push delivery today. See [Known Limitations](https://www.heliosdb.com/docs/nano/reference/compatibility/known-limitations/) for the current status, and the Nano changelog for the release that adds it. --- ## References 1. PostgreSQL LISTEN/NOTIFY Documentation: https://www.postgresql.org/docs/current/sql-notify.html 2. Redis Pub/Sub Performance Benchmarks: https://redis.io/docs/manual/pubsub/ 3. "The Database Poll Pattern" Anti-Pattern Analysis: Martin Fowler, Enterprise Integration Patterns 4. Edge Computing Market Size: IDC IoT and Edge Computing Forecast 2024-2028 5. Real-Time Web Application Architecture: "Designing Data-Intensive Applications" by Martin Kleppmann 6. Cache Invalidation Strategies: "Web Application Performance" (O'Reilly, 2023) --- **Review Cycle**: Quarterly **Owner**: Product Marketing **Adapted for**: HeliosDB Nano Embedded Database