Add AICL example: 19_observer_system.aicl

data/aicl/examples/19_observer_system.aicl

@@ -0,0 +1,413 @@
+# AICL Example: Observer Pattern Event Bus
+# Implements a robust publish/subscribe event bus with topic routing, guaranteed message ordering,
+# backpressure management, dead letter queue handling, and replay capabilities.
+
+# Level 1: Architecture
+Goal: Build a high-throughput event bus implementing the Observer pattern with topic-based routing,
+    strict per-topic message ordering, backpressure to protect slow subscribers,
+    dead letter queues for poison messages, and replay support for missed events.
+
+Constraint: Publishers must be decoupled from subscribers; no direct references allowed.
+Constraint: Each topic maintains FIFO ordering within a partition key.
+Constraint: Subscribers must acknowledge messages within a configurable timeout.
+Constraint: Backpressure must be applied before memory pressure reaches critical threshold.
+Constraint: Dead letter messages must never be silently dropped; always persisted for inspection.
+
+Risk: Slow subscriber causes unbounded buffer growth
+Recovery: Apply per-subscriber backpressure with configurable max buffer; overflow goes to DLQ
+
+Risk: Message ordering violation under concurrent publishing
+Recovery: Partition by routing key; single-writer per partition with sequencer
+
+Risk: Dead letter queue grows unbounded without reprocessing
+Recovery: Scheduled DLQ reprocessor with exponential backoff; alert at queue depth threshold
+
+Risk: Subscriber crash loses in-flight messages
+Recovery: Redeliver unacknowledged messages after visibility timeout with at-least-once guarantee
+
+Risk: Topic routing misconfiguration routes messages to wrong subscribers
+Recovery: Validate topic subscription patterns at registration; audit log all routing decisions
+
+Risk: Replay floods subscribers with historical volume
+Recovery: Throttle replay delivery rate; subscribers opt-in to replay with max throughput setting
+
+Layer: Core
+    SubLayer: Publisher
+    SubLayer: TopicRouter
+    SubLayer: PartitionManager
+Layer: Subscription
+    SubLayer: SubscriberRegistry
+    SubLayer: SubscriptionManager
+    SubLayer: AcknowledgmentTracker
+Layer: Reliability
+    SubLayer: BackpressureController
+    SubLayer: DeadLetterQueue
+    SubLayer: ReplayEngine
+
+Validation: Topic name must match [a-zA-Z][a-zA-Z0-9_.-] pattern
+Validation: Subscriber callback must complete within acknowledgment timeout
+Validation: Partition key must be present for ordered topics
+Validation: Replay request must specify valid time range and subscriber ID
+Validation: Backpressure threshold must be positive and less than system memory limit
+Validation: Dead letter message must preserve original headers and failure reason
+
+# Level 2: Entities
+Entity Topic
+    topicId: string
+    topicName: string
+    partitionCount: integer
+    retentionHours: integer
+    isOrdered: boolean
+    createdAt: datetime
+    messageCount: integer
+    byteRate: float
+    schema: dict
+
+Entity Subscription
+    subscriptionId: string
+    subscriberName: string
+    topicPattern: string
+    callback: string
+    filterExpression: string
+    maxInFlight: integer
+    ackTimeout: integer
+    retryPolicy: dict
+    createdAt: datetime
+
+Entity Message
+    messageId: string
+    topic: string
+    partitionKey: string
+    sequenceNumber: integer
+    payload: dict
+    headers: dict
+    publishedAt: datetime
+    expiresAt: datetime
+    traceId: string
+
+Entity DeadLetterEntry
+    entryId: string
+    originalMessage: Message
+    subscriberName: string
+    failureReason: string
+    failureCount: integer
+    firstFailedAt: datetime
+    lastFailedAt: datetime
+    nextRetryAt: datetime
+    deadLetteredAt: datetime
+
+Entity BackpressureState
+    subscriberId: string
+    currentBufferSize: integer
+    maxBufferSize: integer
+    pressureLevel: string
+    throttleRate: float
+    droppedCount: integer
+    lastAdjustedAt: datetime
+
+Entity SubscriberMetrics
+    subscriberId: string
+    messagesProcessed: integer
+    messagesFailed: integer
+    averageLatency: float
+    p99Latency: float
+    lastMessageAt: datetime
+    activeSubscriptions: integer
+    dlqDepth: integer
+
+# Level 3: Behaviors
+Behavior PublishMessage
+    Input:
+        topic: string
+        partitionKey: string
+        payload: dict
+        headers: dict
+    Output:
+        messageId: string
+        sequenceNumber: integer
+    Action:
+        Validate topic exists and schema matches payload
+        Determine partition from partition key hash
+        Assign monotonic sequence number within partition
+        Persist message to topic log
+        Route message to all matching subscriptions
+        Apply backpressure check per subscriber before delivery
+        Return message ID and sequence number
+
+Behavior Subscribe
+    Input:
+        topicPattern: string
+        callback: string
+        config: dict
+    Output:
+        subscriptionId: string
+    Action:
+        Validate topic pattern syntax and matching topics
+        Register subscription in subscriber registry
+        Initialize delivery buffer and backpressure controller
+        Begin message delivery from current head
+        Return subscription identifier
+
+Behavior AcknowledgeMessage
+    Input:
+        subscriptionId: string
+        messageId: string
+        status: string
+    Output:
+        acknowledged: boolean
+    Action:
+        Verify message belongs to subscription's in-flight set
+        If status is success, remove from in-flight and advance checkpoint
+        If status is failure, apply retry policy or move to DLQ
+        Update subscriber metrics
+        Release backpressure slot
+        Return acknowledgment confirmation
+
+Behavior RouteToSubscriber
+    Input:
+        message: Message
+        subscription: Subscription
+    Output:
+        delivered: boolean
+    Action:
+        Evaluate filter expression against message headers
+        Check subscriber backpressure state
+        If pressure is critical, route to DLQ with backpressure reason
+        If pressure is moderate, apply throttle delay
+        If acceptable, deliver to subscriber callback
+        Track delivery in acknowledgment tracker
+        Return delivery result
+
+Behavior ProcessDeadLetter
+    Input:
+        entryId: string
+        action: string
+    Output:
+        result: string
+    Action:
+        Load dead letter entry from DLQ store
+        If action is retry, re-inject message to original subscription
+        If action is discard, log and permanently remove
+        If action is redirect, publish to alternate topic
+        Update DLQ metrics and entry status
+        Return action result
+
+Behavior ReplayMessages
+    Input:
+        subscriptionId: string
+        fromTimestamp: datetime
+        toTimestamp: datetime
+        maxThroughput: integer
+    Output:
+        replayId: string
+        messageCount: integer
+    Action:
+        Validate subscription and time range
+        Load messages from topic log for time range
+        Apply max throughput throttling
+        Deliver messages to subscriber callback
+        Skip already-acknowledged messages
+        Track replay progress and completion
+        Return replay statistics
+
+# Level 4: Conditions
+Condition: BackpressureCritical
+    When subscriber buffer reaches 90% of max capacity
+    Then redirect new messages to DLQ with backpressure flag and alert subscriber owner
+
+Condition: AcknowledgmentTimeout
+    When message remains unacknowledged beyond ackTimeout
+    Then redeliver message with incremented delivery attempt counter; after max attempts move to DLQ
+
+Condition: InvalidMessageSchema
+    When message payload does not conform to topic schema
+    Then reject publication with validation error; if already published, route to DLQ
+
+# Level 5: Events
+Event: OnSubscriberRegistered
+    On new subscription created for a topic pattern
+    Action: Warm subscriber connection pool and validate callback endpoint reachability
+
+Event: OnBackpressureThreshold
+    On subscriber buffer crosses 75% capacity
+    Action: Apply throttle, notify subscriber via health endpoint, and begin buffering upstream
+
+Event: OnDeadLetterCreated
+    On message moved to dead letter queue
+    Action: Emit metric, send notification to subscriber owner, and log for audit trail
+
+Event: OnReplayCompleted
+    On message replay batch fully delivered
+    Action: Log replay summary, update subscriber checkpoint, and emit completion metric
+
+# Level 6: Concurrency
+Parallel:
+    Message publishing to multiple partitions concurrently
+    Message delivery to independent subscribers in parallel
+    Dead letter queue reprocessing on separate worker pool
+    Backpressure monitoring and adjustment across all subscribers
+    Metrics aggregation from all subscriber delivery threads
+
+# Level 7: Optimization
+Optimize: Message delivery throughput
+    Priority: Batch deliveries per subscriber; minimize lock contention on partition sequencer
+
+Optimize: Topic routing latency
+    Priority: Cache subscription pattern matching results; pre-compile filter expressions
+
+# Level 8: Learning
+Learn: Subscriber processing capacity
+    Goal: Predict optimal max-in-flight setting per subscriber to maximize throughput without overflow
+    Adapt: Per-subscriber maxInFlight and buffer size
+    Based: Subscriber acknowledgment rate and latency distribution over sliding window
+
+Learn: DLQ root cause patterns
+    Goal: Categorize dead-lettered messages by failure type to identify systemic issues
+    Adapt: Alerting rules and auto-retry thresholds per failure category
+    Based: Historical DLQ entry failure reason distribution and retry success rates
+
+# Level 9: Security
+Security:
+    Encrypt: Message payload in transit via TLS 1.3 with pinned certificates between bus and subscribers
+    Encrypt: DLQ entries at rest using AES-256 with per-topic encryption keys
+    Protect: Topic publication with IAM-style authorization and topic-level ACL
+    Protect: Subscription callbacks from message injection via HMAC message signing
+    Protect: Replay endpoints from unauthorized access with admin role requirement and audit logging
+
+# Level 10: Native
+Rust
+{
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+#[derive(Clone, Debug)]
+pub struct Message {
+    pub message_id: String,
+    pub topic: String,
+    pub partition_key: String,
+    pub sequence_number: u64,
+    pub payload: Vec<u8>,
+    pub headers: HashMap<String, String>,
+    pub published_at: Instant,
+}
+
+#[derive(Clone, Debug)]
+pub struct Subscription {
+    pub subscription_id: String,
+    pub subscriber_name: String,
+    pub topic_pattern: String,
+    pub max_in_flight: usize,
+    pub ack_timeout: Duration,
+    pub retry_max: u32,
+}
+
+#[derive(Debug)]
+pub enum DeliverStatus {
+    Acknowledged,
+    Failed(String),
+    Timeout,
+}
+
+pub struct DeadLetterEntry {
+    pub entry_id: String,
+    pub original_message: Message,
+    pub subscriber_name: String,
+    pub failure_reason: String,
+    pub failure_count: u32,
+    pub first_failed_at: Instant,
+    pub dead_lettered_at: Instant,
+}
+
+pub struct EventBus {
+    topics: Arc<Mutex<HashMap<String, Vec<Message>>>>,
+    subscriptions: Arc<Mutex<Vec<Subscription>>>,
+    in_flight: Arc<Mutex<HashMap<String, (Message, Instant, u32)>>>,
+    dead_letter_queue: Arc<Mutex<Vec<DeadLetterEntry>>>,
+    max_buffer: usize,
+}
+
+impl EventBus {
+    pub fn new(max_buffer: usize) -> Self {
+        EventBus {
+            topics: Arc::new(Mutex::new(HashMap::new())),
+            subscriptions: Arc::new(Mutex::new(Vec::new())),
+            in_flight: Arc::new(Mutex::new(HashMap::new())),
+            dead_letter_queue: Arc::new(Mutex::new(Vec::new())),
+            max_buffer,
+        }
+    }
+
+    pub fn publish(&self, topic: &str, msg: Message) -> u64 {
+        let mut topics = self.topics.lock().unwrap();
+        let partition = topics.entry(topic.to_string()).or_insert_with(Vec::new);
+        let seq = partition.len() as u64;
+        let mut msg = msg;
+        msg.sequence_number = seq;
+        partition.push(msg.clone());
+        seq
+    }
+
+    pub fn subscribe(&self, sub: Subscription) {
+        let mut subs = self.subscriptions.lock().unwrap();
+        subs.push(sub);
+    }
+
+    pub fn acknowledge(&self, subscription_id: &str, message_id: &str) -> bool {
+        let mut in_flight = self.in_flight.lock().unwrap();
+        in_flight.remove(&format!("{}:{}", subscription_id, message_id)).is_some()
+    }
+
+    pub fn check_timeouts(&self) {
+        let mut in_flight = self.in_flight.lock().unwrap();
+        let mut dlq = self.dead_letter_queue.lock().unwrap();
+        let now = Instant::now();
+        let expired: Vec<String> = in_flight.iter()
+            .filter(|(_, (msg, sent_at, retries))| {
+                now.duration_since(*sent_at) > Duration::from_secs(30) && *retries >= 3
+            })
+            .map(|(k, _)| k.clone())
+            .collect();
+        for key in expired {
+            if let Some((msg, _, retries)) = in_flight.remove(&key) {
+                dlq.push(DeadLetterEntry {
+                    entry_id: format!("dlq-{}", msg.message_id),
+                    original_message: msg,
+                    subscriber_name: String::new(),
+                    failure_reason: "ack_timeout_exceeded".to_string(),
+                    failure_count: retries,
+                    first_failed_at: now,
+                    dead_lettered_at: now,
+                });
+            }
+        }
+    }
+
+    pub fn dead_letter_count(&self) -> usize {
+        self.dead_letter_queue.lock().unwrap().len()
+    }
+}
+
+fn main() {
+    let bus = EventBus::new(10_000);
+    let sub = Subscription {
+        subscription_id: "sub-1".to_string(),
+        subscriber_name: "order-service".to_string(),
+        topic_pattern: "orders.*".to_string(),
+        max_in_flight: 50,
+        ack_timeout: Duration::from_secs(30),
+        retry_max: 3,
+    };
+    bus.subscribe(sub);
+    let msg = Message {
+        message_id: "msg-1".to_string(),
+        topic: "orders.created".to_string(),
+        partition_key: "order-123".to_string(),
+        sequence_number: 0,
+        payload: vec![],
+        headers: HashMap::new(),
+        published_at: Instant::now(),
+    };
+    let seq = bus.publish("orders.created", msg);
+    println!("Published message with sequence: {}", seq);
+}