Architecture

Understanding the III Engine architecture, WebSocket communication, and client-server model

Overview

The III SDK provides a distributed function execution platform that connects multiple workers (services) through a central III Engine. Workers communicate with the Engine via WebSocket connections, enabling real-time function invocation, bidirectional streaming, and event-driven architectures.

System Components

1. III Engine (Central Hub)

   The Engine acts as a message broker and coordinator that:

   • Routes function invocations between workers
   • Manages trigger registrations and event dispatching
   • Provides streaming channels for data transfer
   • Maintains worker registry and health status
   • Handles distributed tracing with OpenTelemetry
2. Workers (SDK Clients)

   Workers are applications that connect to the Engine and:

   • Register functions they can execute
   • Subscribe to triggers (HTTP endpoints, events, schedules)
   • Invoke functions registered by other workers
   • Stream data through channels
   • Report metrics and telemetry
3. WebSocket Protocol

   All communication uses a persistent WebSocket connection with:

   • Binary message framing for efficient data transfer
   • JSON-based message protocol with type discriminators
   • Automatic reconnection with exponential backoff
   • Distributed tracing context propagation (W3C Trace Context)

Communication Model

WebSocket Connection Lifecycle

The SDK manages WebSocket connections with automatic reconnection:

// Connection states
type IIIConnectionState =
  | 'disconnected'  // Initial state or after shutdown
  | 'connecting'    // Attempting initial connection
  | 'connected'     // Active WebSocket connection
  | 'reconnecting'  // Retrying after disconnect
  | 'failed'        // Max retries exceeded

The SDK automatically reconnects with exponential backoff (default: 1s initial delay, 30s max delay, infinite retries). Functions and triggers are re-registered on reconnection.

Message Flow

Message Protocol

All messages follow a consistent structure with a type field:

enum MessageType {
  RegisterFunction = 'registerfunction',
  UnregisterFunction = 'unregisterfunction',
  InvokeFunction = 'invokefunction',
  InvocationResult = 'invocationresult',
  RegisterTrigger = 'registertrigger',
  UnregisterTrigger = 'unregistertrigger',
  RegisterTriggerType = 'registertriggertype',
  UnregisterTriggerType = 'unregistertriggertype',
  WorkerRegistered = 'workerregistered',
}

Function Invocation Messages

InvokeFunction (Request):

{
  type: 'invokefunction',
  invocation_id?: string,  // Optional for fire-and-forget
  function_id: string,     // Target function path
  data: unknown,           // Function input
  traceparent?: string,    // W3C trace context
  baggage?: string         // W3C baggage header
}

InvocationResult (Response):

{
  type: 'invocationresult',
  invocation_id: string,
  function_id: string,
  result?: unknown,        // Success result
  error?: {                // Or error details
    code: string,
    message: string
  },
  traceparent?: string,
  baggage?: string
}

Reconnection Strategy

The SDK implements resilient reconnection with configurable backoff:

interface IIIReconnectionConfig {
  initialDelayMs: number        // Starting delay (default: 1000ms)
  maxDelayMs: number            // Maximum delay cap (default: 30000ms)
  backoffMultiplier: number     // Exponential factor (default: 2)
  jitterFactor: number          // Random jitter 0-1 (default: 0.3)
  maxRetries: number            // Max attempts, -1 for infinite (default: -1)
}

Reconnection behavior:

1. Connection Lost

   WebSocket close event detected, state changes to reconnecting

2. Backoff Calculation

   const exponentialDelay = initialDelayMs * (backoffMultiplier ** attempt)
   const cappedDelay = Math.min(exponentialDelay, maxDelayMs)
   const jitter = cappedDelay * jitterFactor * (2 * Math.random() - 1)
   const delay = Math.floor(cappedDelay + jitter)
   
3. Re-registration

   On successful reconnection:

   • All trigger types are re-registered
   • All services are re-registered
   • All functions (local and HTTP) are re-registered
   • All triggers are re-registered
   • Queued messages are sent

Monitor connection state changes with onConnectionStateChange() to implement custom reconnection logic or user notifications.

Worker Registration

Workers automatically register metadata on connection:

// Source: packages/node/iii/src/iii.ts:361-380
private registerWorkerMetadata(): void {
  this.triggerVoid(EngineFunctions.REGISTER_WORKER, {
    runtime: 'node',              // or 'python', 'rust'
    version: SDK_VERSION,         // SDK version
    name: this.workerName,        // Hostname:PID or custom
    os: getOsInfo(),              // Platform and architecture
    telemetry: {
      language: 'en-US',          // User locale
      project_name: '...',        // Optional project identifier
      framework: '...',           // Optional framework name
      amplitude_api_key: '...'    // Optional analytics key
    }
  })
}

The Engine responds with WorkerRegistered message containing a unique worker_id.

Distributed Tracing

The architecture supports W3C Trace Context for end-to-end observability:

Every function invocation automatically propagates traceparent and baggage headers, enabling distributed tracing across workers without manual instrumentation.

Trace propagation flow:

1. Caller injects trace context:

   const traceparent = injectTraceparent()  // "00-{trace_id}-{span_id}-01"
   const baggage = injectBaggage()          // "key1=value1,key2=value2"
   

2. Engine forwards context with invocation message

3. Handler extracts context and creates child span:

   const parentContext = extractContext(traceparent, baggage)
   return context.with(parentContext, () =>
     withSpan(`call ${function_id}`, { kind: SpanKind.SERVER }, async span => {
       // Handler execution within trace context
     })
   )
   

4. Response includes updated trace context from handler

See packages/node/iii/src/iii.ts:202-217 for implementation details.

Multi-Runtime Support

The III SDK is available in three runtimes with consistent APIs:

Node.js

import { init } from 'iii-sdk'

const iii = init('ws://localhost:8080', {
  workerName: 'my-service',
  enableMetricsReporting: true,
  invocationTimeoutMs: 30000
})

Location: packages/node/iii/src/iii.ts

Python

from iii import III, InitOptions

iii = III('ws://localhost:8080', InitOptions(
  worker_name='my-service',
  enable_metrics_reporting=True,
  invocation_timeout_ms=30000
))
await iii.connect()

Location: packages/python/iii/src/iii/iii.py

Rust

use iii::{III, InitOptions};

let iii = III::init("ws://localhost:8080", InitOptions {
    worker_name: Some("my-service".into()),
    enable_metrics_reporting: true,
    invocation_timeout_ms: 30000,
    ..Default::default()
}).await?;

Location: packages/rust/iii/src/lib.rs

Performance Considerations

Message Batching

The SDK queues messages when WebSocket is not ready and sends them in batch on connection:

// Source: packages/node/iii/src/iii.ts:653-666
const pending = this.messagesToSend
this.messagesToSend = []
for (const message of pending) {
  // Skip cancelled invocations
  if (message.type === MessageType.InvokeFunction &&
      !this.invocations.has(message.invocation_id)) {
    continue
  }
  this.sendMessageRaw(JSON.stringify(message))
}

Invocation Timeouts

All function calls have configurable timeouts (default 30s):

const result = await iii.call('service::function', data, 5000)  // 5s timeout

Timeouts prevent resource leaks from hanging invocations.

Connection State Management

Monitor and react to connection state changes:

const unsubscribe = iii.onConnectionStateChange((state) => {
  switch (state) {
    case 'connected':
      console.log('Ready to process requests')
      break
    case 'reconnecting':
      console.warn('Connection lost, retrying...')
      break
    case 'failed':
      console.error('Max retries exceeded')
      process.exit(1)
  }
})

// Later: unsubscribe()

See packages/node/iii/src/iii.ts:473-488 for implementation.

Graceful Shutdown

The SDK provides graceful shutdown that:

• Stops accepting new invocations
• Rejects pending invocations with error
• Closes WebSocket connection
• Flushes OpenTelemetry data
• Clears all callbacks

// Source: packages/node/iii/src/iii.ts:492-524
await iii.shutdown()

Always call shutdown() before process termination to ensure telemetry data is flushed and in-flight requests are properly handled.

Next Steps

Functions

Learn how to register and invoke functions

Triggers

Understand trigger types and event handling

Channels

Implement bidirectional streaming

Streaming

Build real-time data operations