Websocket Streaming

WebSocket Streaming

Real-time bidirectional communication between DAG execution engines and dashboards via typed event protocols and connection state management.

Decision Points

Connection State Machine

CONNECTING → OPEN → CLOSING → CLOSED
    ↓         ↓         ↓         ↓
    wait    process   queue     reconnect
  
If connection === 'CONNECTING':
  - Buffer outgoing messages in memory queue
  - Show "connecting..." indicator
  - Timeout after 10s → retry with backoff

If connection === 'OPEN':
  - Send queued messages immediately
  - Process incoming events via typed dispatch
  - Send heartbeat every 30s

If connection === 'CLOSING':
  - Stop sending new messages
  - Finish processing in-flight events
  - Prepare for reconnection

If connection === 'CLOSED':
  - Calculate backoff: min(1000 * 2^attempt, 30000)ms
  - Increment reconnect attempt counter
  - Trigger reconnection after delay

Event Type Routing

If event.type === 'node_state':
  - Update store.nodes[event.node_id].status
  - Trigger UI re-render for affected node
  - Log metrics if present

If event.type === 'cost_update':
  - Update budget display in header
  - Show warning if remaining < 20%
  - Log cost trajectory for analytics

If event.type === 'human_gate_waiting':
  - Show modal with gate presentation
  - Enable approval/rejection buttons
  - Start timeout countdown (default 5min)

If event.type === 'error':
  - Show error toast notification
  - Highlight affected node (if node_id present)
  - Log to error tracking service

Message Buffering Strategy

If connectionState === 'OPEN' && bufferQueue.length === 0:
  - Send message immediately via ws.send()

If connectionState !== 'OPEN' && message.priority === 'high':
  - Add to front of bufferQueue
  - Limit high-priority queue to 50 messages

If connectionState !== 'OPEN' && message.priority === 'normal':
  - Add to back of bufferQueue
  - Drop oldest if queue > 200 messages

If connectionState becomes 'OPEN' && bufferQueue.length > 0:
  - Send all queued messages in order
  - Clear buffer queue
  - Resume normal operation

Failure Modes

Connection Thrashing

Symptom: Rapid connect/disconnect cycles, exponentially increasing CPU usage Detection: If reconnect attempts > 5 in 60 seconds Fix: Implement exponential backoff with max delay (30s), circuit breaker after 10 failures

Message Flooding

Symptom: Dashboard becomes unresponsive, memory usage spikes, UI freezes Detection: If incoming message rate > 100/second or buffer queue > 1000 messages Fix: Implement message throttling, batch state updates, drop non-critical events

State Desync After Reconnect

Symptom: Dashboard shows wrong node statuses, missing execution progress Detection: If timestamp gap > 30s between disconnect and reconnect Fix: Send 'resync_request' on reconnect, server responds with full current state

Memory Leak on Connection Failure

Symptom: Memory usage grows steadily, never decreases, eventual crash Detection: If WebSocket reference count > 5 for single DAG, or buffer never clears Fix: Properly close previous WebSocket before creating new one, clear event listeners

Silent Message Loss

Symptom: Critical events (human gates, errors) never reach dashboard Detection: If expected event doesn't arrive within timeout window Fix: Implement message acknowledgment, server retries unacknowledged critical events

Worked Examples

Network Drop During Node Execution

Scenario: User is monitoring DAG execution. Network drops for 45 seconds during critical node processing. Connection recovers.

Step 1 - Connection Loss Detection:

// WebSocket onclose event fires
ws.onclose = () => {
  setConnectionState('CLOSED');
  // Decision: Network issue or server restart?
  // → Try reconnect (could be temporary network)
  scheduleReconnect();
};

Step 2 - Buffering Decision:

// User tries to send human decision during outage
const sendDecision = (decision) => {
  if (connectionState !== 'OPEN') {
    // Decision: Buffer or reject?
    // → Buffer high-priority messages (human decisions)
    bufferQueue.unshift({ 
      type: 'human_decision', 
      priority: 'high',
      timestamp: Date.now()
    });
    showToast("Decision queued - connection lost");
  }
};

Step 3 - Reconnection with State Gap:

ws.onopen = () => {
  const disconnectDuration = Date.now() - lastDisconnectTime;
  
  if (disconnectDuration > 30000) {
    // Decision: Full resync or continue?
    // → 45s gap requires full resync
    ws.send(JSON.stringify({ type: 'resync_request', last_seen: lastEventTimestamp }));
    
    // Server responds with missed events + current state
    // Trade-off: Higher bandwidth but guaranteed consistency
  }
  
  // Send buffered messages
  flushBufferQueue();
};

What novice misses: Assumes reconnection means everything is fine, doesn't handle state gap What expert catches: Calculates disconnect duration, requests resync for gaps > 30s, preserves critical user actions in buffer

Quality Gates

• Connection establishes within 5 seconds of page load
• Connection remains stable for >5 minutes during normal operation
• No message loss during planned reconnection scenarios
• Memory usage stays <50MB even with 1000+ node events
• Exponential backoff delays: 1s, 2s, 4s, 8s, 16s, 30s (max)
• Critical events (human gates, errors) always reach dashboard within 2s
• UI shows connection status clearly (connecting/connected/disconnected)
• Buffered messages sent in correct order after reconnection
• WebSocket properly closed on component unmount (no leaks)
• Event handlers use exhaustive switch on message.type (TypeScript strict)

Not-For Boundaries

Don't use WebSocket streaming for:

• Simple status polling → Use polling-pattern-optimizer with REST endpoints instead
• File uploads/downloads → Use file-transfer-handler with HTTP multipart instead
• Server-sent events only → Use sse-event-stream for simpler one-way communication
• Request-response APIs → Use api-architect for traditional REST/GraphQL patterns
• Database real-time queries → Use database-change-streams for direct DB subscriptions
• Authentication flows → Use auth-flow-manager for login/logout/token refresh

Delegate to other skills:

• For message queuing/reliability → use message-queue-architect
• For load balancing WebSockets → use websocket-load-balancer
• For WebRTC peer connections → use webrtc-connection-manager