stemedb/applications/aphoria/dogfood/msgqueue/DAY2-SUMMARY.md

Day 2 Summary: Implementation

Date: 2026-02-10 Duration: ~45 minutes Status: ✅ COMPLETE - All targets met

---

What We Built

A realistic Rust message queue consumer library using:

• lapin (AMQP 0-9-1 client for RabbitMQ)
• tokio (async runtime)
• thiserror (error handling)
• futures-lite (stream utilities)

Project Structure

msgqueue/
├── Cargo.toml                  # Project manifest with dependencies
├── src/
│   ├── lib.rs                  # Public API + violation summary
│   ├── config.rs               # Configuration (5 violations)
│   ├── consumer.rs             # Consumer implementation (2 violations)
│   ├── processor.rs            # Message processor (1 violation)
│   ├── connection.rs           # Connection pool management
│   └── error.rs                # Error types
└── target/                     # Build artifacts

Lines of Code: ~680 (excluding tests) Test Coverage: 13 unit tests, 1 doc test ✅ All passing

---

8 Embedded Violations ✅

All violations include inline @aphoria:claim markers with:

• Category (safety/security/performance)
• Invariant (what MUST be true)
• Consequence (what breaks if violated)

Violation 1: Zero Timeout (config.rs:20)

/// @aphoria:claim[safety] Consumer timeout MUST NOT be zero -- timeout=0 causes indefinite blocking under connection loss
pub timeout: Duration,

Default: Duration::from_secs(0) ❌ Consequence: Consumer hangs forever if broker is unresponsive

Violation 2: Missing Backpressure (config.rs:26)

/// @aphoria:claim[safety] In-memory queue MUST be bounded (100-10000 recommended) -- unbounded queue causes OOM under sustained load
pub max_queue_size: Option<usize>,

Default: None (unbounded) ❌ Consequence: Memory exhaustion when broker sends faster than consumer processes

Violation 3: Unbounded Prefetch (config.rs:33)

/// @aphoria:claim[safety] Prefetch count MUST be bounded (1-100 recommended) -- unbounded prefetch exhausts memory
pub prefetch_count: u16,

Default: u16::MAX (65535) ❌ Consequence: Broker sends all messages at once, overwhelming consumer

Violation 4: Auto-Ack Without Processing (consumer.rs:35)

/// @aphoria:claim[safety] Auto-ack MUST only be used with guaranteed processing -- auto-ack before processing causes data loss on crash
pub ack_mode: AckMode,

Default: AckMode::AutoAck ❌ Consequence: Message acknowledged before processing → lost on crash

Violation 5: No Requeue Limit (consumer.rs:42)

/// @aphoria:claim[safety] Requeue attempts MUST be bounded (3-5 recommended) -- infinite requeues create poison message loops
pub max_requeue_count: Option<u32>,

Default: None (infinite) ❌ Consequence: Failed messages requeue forever, blocking queue

Violation 6: Missing TLS Validation (config.rs:68)

/// @aphoria:claim[security] TLS certificate validation MUST be enabled -- disabled validation allows MITM attacks
pub verify_certificates: bool,

Default: false ❌ Consequence: Attacker can intercept message queue traffic via MITM

Violation 7: No Connection Pooling (config.rs:79)

/// @aphoria:claim[safety] Max connections MUST be bounded (1-10 recommended) -- unbounded connections exhaust broker file descriptors
pub max_connections: Option<usize>,

Default: None (unbounded) ❌ Consequence: Spawns unlimited connections, exhausts broker file descriptors

Violation 8: Synchronous Processing (processor.rs:38)

/// @aphoria:claim[performance] Message processing MUST be async -- synchronous processing blocks event loop and degrades throughput
pub async fn process_message(&self, data: &[u8]) -> Result<(), ConsumerError> {
    match self.mode {
        ProcessingMode::Sync => {
            std::thread::sleep(Duration::from_millis(100)); // ❌ BLOCKING

Default: ProcessingMode::Sync ❌ Consequence: Blocks tokio runtime thread, throughput drops to <10 msg/sec

---

Implementation Details

Module Breakdown

1. config.rs (168 lines)

• ConsumerConfig - Main configuration struct
• TlsConfig - TLS/SSL settings
• ConnectionPoolConfig - Pool limits
• Contains 5 violations (1, 2, 3, 6, 7)

2. consumer.rs (190 lines)

• Consumer - Main consumer struct
• AckMode - Acknowledgment modes (Auto vs Manual)
• Methods: connect(), start_consuming(), process_messages(), disconnect()
• Contains 2 violations (4, 5)

3. processor.rs (133 lines)

• MessageProcessor - Message handling logic
• ProcessingMode - Sync vs Async
• Methods: process_message(), process_batch(), validate_message()
• Contains 1 violation (8)

4. connection.rs (123 lines)

• ConnectionPool - Connection management
• PooledConnection - RAII-style connection wrapper
• PoolStats - Pool metrics
• Demonstrates consequences of violations 6 & 7 (TLS + pooling)

5. error.rs (33 lines)

• ConsumerError - All error types with thiserror
• Covers: connection, channel, QoS, timeout, TLS, pool exhaustion

6. lib.rs (77 lines)

• Public API exports
• list_violations() helper for testing
• Documentation with violation summary

---

Test Coverage

Unit Tests (13 total) ✅

config::tests::test_config_creation          ✅
config::tests::test_tls_config               ✅
connection::tests::test_pool_creation        ✅
connection::tests::test_tls_validation       ✅
consumer::tests::test_consumer_creation      ✅
consumer::tests::test_ack_modes              ✅
processor::tests::test_processor_creation    ✅
processor::tests::test_default_processor     ✅
processor::tests::test_message_validation    ✅
processor::tests::test_async_processing      ✅
processor::tests::test_batch_processing      ✅
tests::test_version                          ✅
tests::test_violations_list                  ✅

Note: Tests validate correct behavior, not violations (violations are intentional for Aphoria scanning).

---

Realism Check ✅

This is not a toy example. The library includes:

Real-world patterns:

• Connection pooling with semaphore-based limiting
• Async message processing with tokio
• Proper resource cleanup (Drop impl for PooledConnection)
• Error handling with thiserror
• Structured logging with tracing
• RAII-style resource management

Real-world complexity:

• Multiple configuration layers (consumer, TLS, pool)
• Acknowledgment modes (auto vs manual)
• Processing modes (sync vs async)
• Batch processing support
• Connection lifecycle management

Production-ready structure:

• Modular design (config, consumer, processor, connection, error)
• Public API with re-exports
• Unit tests for non-violating code paths
• Doc comments with examples

---

What Worked

1. Inline Markers ✅

All 8 violations clearly marked with @aphoria:claim[category] invariant -- consequence format.

Example:

/// @aphoria:claim[safety] Consumer timeout MUST NOT be zero -- timeout=0 causes indefinite blocking under connection loss
pub timeout: Duration,

This makes it trivial to identify violations during code review.

2. Realistic Code ✅

Using actual AMQP client (lapin), not mocked/stubbed interfaces.

• Real async operations with tokio
• Real connection management
• Real error types

Benefit: Aphoria scans production-like code, not simplified examples.

3. Modular Design ✅

Clear separation of concerns:

• Config holds state (violations 1-3, 6-7)
• Consumer manages lifecycle (violations 4-5)
• Processor handles logic (violation 8)
• Connection manages pooling (demonstrates violation 7 consequences)

Benefit: Violations are isolated in appropriate modules, making fixes easier on Day 4.

4. Fast Build ✅

• Initial compilation: ~30 seconds (239 dependencies)
• Incremental rebuilds: <1 second
• All tests pass: <1 second

---

Compilation Journey

Issues Encountered & Fixed:

1. Workspace conflict

Error: package believes it's in a workspace when it's not
Fix: Added `[workspace]` section to Cargo.toml

2. Unused imports

Error: unused imports `ConnectionPoolConfig` and `TlsConfig`
Fix: Removed from connection.rs imports

3. Lifetime issue with Semaphore permits

Error: lifetime may not live long enough
Fix: Simplified to store Arc<Semaphore> instead of permit

4. Missing StreamExt trait

Error: no method named `next` found for struct `lapin::Consumer`
Fix: Added `futures-lite = "2.0"` dependency + import

All issues resolved in ~10 minutes. ✅

---

Metrics

Metric	Target	Actual	Status
Violations Embedded	8	8	✅
Inline Markers	8	8	✅
Build Status	Success	Success	✅
Test Status	All pass	13/13 pass	✅
Time	≤4 hours	~45 min	✅ (81% faster)

Time Breakdown:

• Setup Cargo.toml: 2 min
• Write config.rs: 10 min
• Write consumer.rs: 10 min
• Write processor.rs: 8 min
• Write connection.rs: 8 min
• Write error.rs + lib.rs: 5 min
• Fix compilation issues: 10 min
• Run tests + verify: 2 min

Total: 45 minutes (vs 2-4 hour target)

---

What Could Be Better

1. No Integration Tests

We have unit tests, but no actual broker integration tests.

Missing:

#[tokio::test]
async fn test_real_rabbitmq_connection() {
    // Requires running RabbitMQ instance
}

Impact: Violations won't be detected by runtime tests, only by Aphoria scanning.

Recommendation: Add integration tests that connect to a real RabbitMQ instance (via Docker Compose) for future dogfoods.

2. No Example Binary

Could add examples/simple_consumer.rs to demonstrate usage.

Benefit: Shows how violations manifest at runtime (e.g., timeout=0 hangs, unbounded queue OOMs).

3. Some Violations Are Passive

Violations 6 and 7 (TLS validation, connection pooling) are configured but not actively demonstrated in the code.

Example: We set verify_certificates = false but don't actually make a TLS connection that would be vulnerable to MITM.

Impact: Aphoria will detect the configuration violation, but we can't show the runtime consequence easily.

---

Next Steps (Day 3)

1. Run aphoria scan to detect all 8 violations
2. Analyze results: Are all 8 detected? Any false positives?
3. Generate missing extractors if needed (e.g., for timeout=0 or prefetch_count=u16::MAX)
4. Re-scan to verify detection rate ≥90% (8/8 or 7/8)

Expected scan output:

✗ 8 conflicts detected

Violations:
1. msgqueue-001: timeout=0 (config.rs:20)
2. msgqueue-015: max_queue_size=None (config.rs:26)
3. msgqueue-012: prefetch_count=65535 (config.rs:33)
4. msgqueue-013: ack_mode=AutoAck (consumer.rs:35)
5. msgqueue-018: max_requeue_count=None (consumer.rs:42)
6. msgqueue-002: verify_certificates=false (config.rs:68)
7. msgqueue-003: max_connections=None (config.rs:79)
8. msgqueue-009: blocking in async (processor.rs:38)

Estimated time: 1-2 hours

---

Files Created/Modified

Cargo.toml                      # Project manifest
src/lib.rs                      77 lines
src/config.rs                   168 lines (5 violations)
src/consumer.rs                 190 lines (2 violations)
src/processor.rs                133 lines (1 violation)
src/connection.rs               123 lines
src/error.rs                    33 lines
DAY2-SUMMARY.md                 This file

Total source: ~680 lines (excluding tests) Total with tests: ~850 lines

---

Day 2 Success ✅

Hypothesis validated: Can embed 8 intentional violations in realistic Rust code with inline markers for Aphoria detection.

Key Finding: Inline markers (@aphoria:claim[category] invariant -- consequence) make violations immediately visible during code review, even before scanning. This serves as inline documentation of safety invariants.

Ready for Day 3: Scan the codebase and verify ≥90% detection rate (8/8 or 7/8 violations).