JetStream Configuration

JetStream for Digital Twins

JetStream provides the persistence layer for fleet digital twins. Each stream type serves a specific purpose with tailored retention and replication settings.

Stream Architecture

Five JetStream resources support the digital twin pattern:

TWIN_STATE: Telemetry Stream

Stores continuous vehicle state with subject-based rollup:

name: TWIN_STATE
subjects:
  - "fleet.prod.veh.*.state.>"
retention: limits
max_age: 1h
max_bytes: 10GB
storage: file
replicas: 3
discard: old
rollup_hdrs: true

Configuration Explained

Subject Rollup

With rollup enabled, consumers can request only the latest message per subject:

// Get latest position for all vehicles
sub, _ := js.Subscribe("fleet.prod.veh.*.state.position",
    nats.DeliverLastPerSubject())

This transforms the stream from a firehose into a queryable state store.

Consumer Patterns

TWIN_EVENTS: Audit Trail

Stores discrete events with long retention for compliance and debugging:

name: TWIN_EVENTS
subjects:
  - "fleet.prod.veh.*.evt.>"
retention: limits
max_age: 90d
max_bytes: 100GB
storage: file
replicas: 3
discard: old

Configuration Explained

Event Properties

Events differ from state:

Example Events

// Armed event
{
  "vid": "VID-001",
  "timestamp": "2024-01-15T10:30:00Z",
  "data": {
    "previous_state": "disarmed",
    "armed_by": "operator-123",
    "reason": "mission_start"
  }
}

// Failsafe event
{
  "vid": "VID-001",
  "timestamp": "2024-01-15T11:45:00Z",
  "data": {
    "failsafe_type": "low_battery",
    "action_taken": "rtl",
    "battery_remaining": 0.15
  }
}

TWIN_CMD: Command Queue

Stores commands with workqueue retention:

name: TWIN_CMD
subjects:
  - "fleet.prod.veh.*.cmd.>"
retention: workqueue
max_age: 5m
storage: file
replicas: 3

Configuration Explained

Workqueue Semantics

1. Command published to stream
2. Vehicle Gateway receives (exactly-once delivery)
3. Gateway processes command
4. Gateway acks message
5. Message removed from stream

Unacked commands re-deliver to ensure commands aren’t lost during disconnections.

Command Flow

Fleet Operator               TWIN_CMD Stream              Vehicle Gateway
     │                             │                             │
     │ Publish: cmd.takeoff        │                             │
     │ ─────────────────────────▶  │                             │
     │                             │ Deliver to consumer         │
     │                             │ ─────────────────────────▶  │
     │                             │                             │ Process
     │                             │                             │ command
     │                             │             Ack             │
     │                             │ ◀─────────────────────────  │
     │                             │ (message removed)           │
     │                             │                             │

TWIN_SHADOW: KV Store

JetStream KV stores maintain digital twin shadow state:

bucket: TWIN_SHADOW
max_value_size: 1MB
history: 5
ttl: 0
replicas: 3
storage: file

Configuration Explained

Key Structure

fleet/prod/veh/VID-001/desired    → Desired state JSON
fleet/prod/veh/VID-001/reported   → Reported state JSON
fleet/prod/veh/VID-001/config     → Vehicle configuration
fleet/prod/veh/VID-001/meta       → Metadata (serial, type, etc.)

Shadow Operations

// Write desired state
kv.Put("fleet/prod/veh/VID-001/desired", desiredJSON)

// Read reported state
entry, _ := kv.Get("fleet/prod/veh/VID-001/reported")

// Watch for changes
watcher, _ := kv.Watch("fleet/prod/veh/VID-001/desired")
for entry := range watcher.Updates() {
    // Desired state changed, reconcile
}

History for Debugging

With history enabled, you can inspect previous states:

// Get all historical values
history, _ := kv.History("fleet/prod/veh/VID-001/desired")
for _, entry := range history {
    fmt.Printf("Rev %d at %s: %s\n",
        entry.Revision(), entry.Created(), entry.Value())
}

TWIN_BLOBS: Object Store

Large artifacts that don’t fit in messages:

bucket: TWIN_BLOBS
max_chunk_size: 128KB
storage: file
replicas: 3

Use Cases

Object Operations

// Upload mission file
obj, _ := obs.PutFile("missions/mission-456.json", file)

// Download to vehicle
reader, _ := obs.Get("missions/mission-456.json")

// List available firmware
objects := obs.List(nats.ObjectSearchPrefix("firmware/"))

Chunked Transfer

Object Store handles large files by:

1. Splitting into chunks (128KB default)
2. Storing chunks as stream messages
3. Tracking metadata (size, hash, chunks)
4. Reassembling on retrieval

This enables resumable transfers over unreliable connections.

Stream Sizing

Per-Vehicle Estimates

Fleet Estimates (1,000 vehicles)

These are baseline estimates. Actual usage depends on message rates and sizes.

Operational Considerations

Stream Health

Monitor these metrics:

Backup Strategy

Migration

Streams can be migrated between clusters:

1. Create stream on target with same configuration
2. Create mirror sourcing from original
3. Wait for sync
4. Switch producers/consumers to target
5. Remove original stream

Summary

Next

Vehicle Gateway →