feat(benchmarks): add flexible queue mode testing with emulator support

- Update run_benchmarks.sh to accept queue_mode argument (async|pubsub)
- Results now organized by queue mode: results/{queue_mode}/
- Add Pub/Sub + GCS emulator setup instructions
- Document AsyncQueue vs PubSub trade-offs and characteristics
- Enable comparative testing: single-server vs distributed architectures
- Include emulator commands for local PubSub benchmarking

Author: prosdev

benchmarks/README.md

@@ -6,20 +6,49 @@ This directory contains load testing scenarios for validating EventKit's perform
 
 ### 1. Start EventKit
 
+Choose your queue mode:
+
+**AsyncQueue Mode (Single-Server, In-Process)**:
+```bash
+# Terminal 1: Start EventKit API with AsyncQueue
+GCP_PROJECT_ID=eventkit-benchmark \
+GCP_GCS_BUCKET=eventkit-events \
+STORAGE_EMULATOR_HOST=http://localhost:4443 \
+EVENTKIT_RING_BUFFER_ENABLED=true \
+EVENTKIT_QUEUE_MODE=async \
+uv run uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000
+```
+
+**PubSub Mode (Distributed, External Queue)**:
 ```bash
-# Terminal 1: Start EventKit API
-cd /path/to/eventkit
-uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000
+# Terminal 1: Start Pub/Sub + GCS emulators
+docker compose up -d pubsub-emulator gcs-emulator
+
+# Terminal 2: Start EventKit API with PubSub
+GCP_PROJECT_ID=eventkit-benchmark \
+GCP_GCS_BUCKET=eventkit-events \
+PUBSUB_EMULATOR_HOST=localhost:8085 \
+STORAGE_EMULATOR_HOST=http://localhost:9023 \
+EVENTKIT_RING_BUFFER_ENABLED=true \
+EVENTKIT_QUEUE_MODE=pubsub \
+EVENTKIT_PUBSUB_TOPIC=eventkit-events \
+uv run uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000
+
+# Terminal 3: Start EventSubscriptionCoordinator workers
+# (Implementation specific - see streaming/coordinator.py for examples)
 ```
 
 ### 2. Run All Benchmarks
 
 ```bash
-# Terminal 2: Run benchmark suite
-./benchmarks/run_benchmarks.sh
+# Test AsyncQueue mode
+./benchmarks/run_benchmarks.sh async
+
+# Or test PubSub mode
+./benchmarks/run_benchmarks.sh pubsub
 ```
 
-This will run all scenarios and save results to `benchmarks/results/`.
+Results will be saved to `benchmarks/results/{queue_mode}/`.
 
 ## Manual Testing
 
@@ -171,6 +200,63 @@ uv run memray run -o memory.bin \
 memray flamegraph memory.bin
 ```
 
+## Comparing Queue Modes
+
+EventKit supports two queue modes with different trade-offs:
+
+### AsyncQueue (Single-Server)
+**Best for**: Development, single-instance deployments
+
+**Characteristics**:
+- In-process Python `asyncio.Queue`
+- Low latency (microseconds to enqueue)
+- No external dependencies
+- Simpler architecture
+- Limited to single server's resources
+
+**Benchmark this to measure**:
+- Maximum throughput for single instance
+- Memory pressure under load
+- Ring buffer → queue → loader latency
+
+### PubSub (Distributed)
+**Best for**: Production, horizontal scaling
+
+**Characteristics**:
+- External queue (Google Cloud Pub/Sub)
+- Higher latency (milliseconds to publish)
+- Distributed processing
+- Horizontal scalability
+- More complex architecture
+
+**Benchmark this to measure**:
+- Network overhead (API → Pub/Sub)
+- Multi-worker scalability
+- Fault tolerance (nack/redelivery)
+
+### Running Comparisons
+
+```bash
+# Benchmark AsyncQueue (single emulator needed)
+docker compose up -d gcs-emulator
+./benchmarks/run_benchmarks.sh async
+
+# Benchmark PubSub (both emulators needed)
+docker compose up -d pubsub-emulator gcs-emulator
+./benchmarks/run_benchmarks.sh pubsub
+
+# Compare results
+diff benchmarks/results/async/baseline.log \
+     benchmarks/results/pubsub/baseline.log
+
+# Stop emulators when done
+docker compose down
+```
+
+**What to expect**:
+- **AsyncQueue**: Higher throughput, lower latency (no network hops)
+- **PubSub**: Lower throughput (network + serialization overhead), but horizontally scalable
+
 ## Configuration Tuning
 
 Test different configurations by setting environment variables:
@@ -187,7 +273,7 @@ export EVENTKIT_RING_BUFFER_ENABLED=false
 export EVENTKIT_EVENT_LOADER_FLUSH_INTERVAL=60
 
 # Run benchmarks
-./benchmarks/run_benchmarks.sh
+./benchmarks/run_benchmarks.sh async
 ```
 
 ## Expected Results

benchmarks/run_benchmarks.sh

@@ -4,6 +4,19 @@
 #
 # Runs all benchmark scenarios and collects results.
 #
+# Usage:
+#   ./benchmarks/run_benchmarks.sh [queue_mode]
+#
+# Arguments:
+#   queue_mode: "async" (default) or "pubsub"
+#
+# Examples:
+#   # Test AsyncQueue (single-server)
+#   ./benchmarks/run_benchmarks.sh async
+#
+#   # Test PubSub (distributed)
+#   ./benchmarks/run_benchmarks.sh pubsub
+#
 
 set -euo pipefail
 
@@ -14,14 +27,22 @@ YELLOW='\033[1;33m'
 NC='\033[0m' # No Color
 
 # Configuration
+QUEUE_MODE="${1:-async}"
 HOST="${EVENTKIT_HOST:-http://localhost:8000}"
-RESULTS_DIR="benchmarks/results"
+RESULTS_DIR="benchmarks/results/${QUEUE_MODE}"
 DURATION="${DURATION:-60s}"
 
+# Validate queue mode
+if [[ "$QUEUE_MODE" != "async" && "$QUEUE_MODE" != "pubsub" ]]; then
+    echo "ERROR: Invalid queue mode '$QUEUE_MODE'. Must be 'async' or 'pubsub'"
+    exit 1
+fi
+
 echo -e "${BLUE}========================================${NC}"
 echo -e "${BLUE}EventKit Performance Benchmarks${NC}"
 echo -e "${BLUE}========================================${NC}"
 echo ""
+echo "Queue Mode: $QUEUE_MODE"
 echo "Host: $HOST"
 echo "Duration: $DURATION"
 echo "Results: $RESULTS_DIR"
@@ -59,7 +80,35 @@ run_scenario() {
 echo -e "${YELLOW}Checking if EventKit is running...${NC}"
 if ! curl -s "$HOST/health" > /dev/null 2>&1; then
     echo "ERROR: EventKit is not running at $HOST"
-    echo "Start it with: uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000"
+    echo ""
+    echo "Start EventKit with the following configuration:"
+    echo ""
+    if [[ "$QUEUE_MODE" == "async" ]]; then
+        echo "  # AsyncQueue mode (single-server, in-process)"
+        echo "  GCP_PROJECT_ID=eventkit-benchmark \\"
+        echo "  GCP_GCS_BUCKET=eventkit-events \\"
+        echo "  STORAGE_EMULATOR_HOST=http://localhost:4443 \\"
+        echo "  EVENTKIT_RING_BUFFER_ENABLED=true \\"
+        echo "  EVENTKIT_QUEUE_MODE=async \\"
+        echo "  uv run uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000"
+    else
+        echo "  # PubSub mode (distributed, external queue)"
+        echo ""
+        echo "  # Start emulators"
+        echo "  docker compose up -d pubsub-emulator gcs-emulator"
+        echo ""
+        echo "  # Start EventKit API"
+        echo "  GCP_PROJECT_ID=eventkit-benchmark \\"
+        echo "  GCP_GCS_BUCKET=eventkit-events \\"
+        echo "  PUBSUB_EMULATOR_HOST=localhost:8085 \\"
+        echo "  STORAGE_EMULATOR_HOST=http://localhost:9023 \\"
+        echo "  EVENTKIT_RING_BUFFER_ENABLED=true \\"
+        echo "  EVENTKIT_QUEUE_MODE=pubsub \\"
+        echo "  EVENTKIT_PUBSUB_TOPIC=eventkit-events \\"
+        echo "  uv run uvicorn eventkit.api.app:app --host=0.0.0.0 --port=8000"
+        echo ""
+        echo "  Note: Ensure you have EventSubscriptionCoordinator workers running separately"
+    fi
     exit 1
 fi
 echo -e "${GREEN}✓ EventKit is running${NC}"