Streams

​

What is a stream?

• Ordered: Records maintain strict sequential ordering via sequence numbers
• Append-only: Records can only be added to the end, never modified
• Identified: Each stream has a unique name within its basin
• Durable: All acknowledged appends are persisted to object storage
• Streamable: Supports both historical reads and real-time streaming

​

Stream naming

• Length: Between 1 and 512 bytes
• Characters: Letters, numbers, underscores (_), and hyphens (-)
• Case-sensitive: MyStream and mystream are different streams
• Unique: Must be unique within a basin

​

Examples

user-123-events
device_telemetry
orders/2024/03
log.application.production

​

Stream positions

pub struct StreamPosition {
    pub seq_num: u64,      // Unique sequence number
    pub timestamp: u64,    // Timestamp in milliseconds
}

​

Sequence numbers

• Start at 0: The first record in a stream has seq_num: 0
• Sequential: Each new record increments the sequence number by 1
• Unique: No two records in a stream have the same sequence number
• Immutable: Sequence numbers never change once assigned

​

Timestamps

• Milliseconds since epoch: Unix timestamp in milliseconds
• Monotonic: Always increasing or equal within a stream
• Configurable source: Can be client-specified, server-assigned, or hybrid

​

Stream tail

{
  "tail": {
    "seq_num": 1000,
    "timestamp": 1709481600000
  }
}

• tail.seq_num is the sequence number that will be assigned to the next appended record
• tail.timestamp is the timestamp of the most recent record

​

Configuration

{
  "stream": "my-stream",
  "config": {
    "storage_class": "express",
    "retention_policy": {
      "age": 2592000
    },
    "timestamping": {
      "mode": "client-require",
      "uncapped": true
    },
    "delete_on_empty": {
      "min_age_secs": 7200
    }
  }
}

​

Storage class

• standard: Append tail latency under 400ms, cost-optimized
• express: Append tail latency under 40ms, performance-optimized

​

Retention policy

{
  "retention_policy": {
    "age": 604800  // 7 days in seconds
  }
}

{
  "retention_policy": {
    "infinite": {}
  }
}

​

Timestamping configuration

​

Mode

• Use client-provided timestamp if present
• Fall back to server arrival time if not provided
• Best for most use cases

• Client must provide timestamp
• Append fails if timestamp is missing
• Use when client timestamp is critical

• Always use server arrival time
• Ignore any client-provided timestamp
• Use for strict server-side ordering

​

Uncapped

• false (default): Client timestamps are capped at server arrival time
• true: Client can provide future timestamps

if !uncapped && timestamp > now {
    timestamp = now;
}

​

Delete-on-empty

{
  "delete_on_empty": {
    "min_age_secs": 3600  // Delete if empty for 1 hour
  }
}

• Set min_age_secs: 0 to disable (default)
• Empty streams are deleted after the configured period
• Useful for temporary or ephemeral streams

​

Stream lifecycle

​

Creation

s2 create-stream s2://my-basin/my-stream

​

Listing streams

s2 list-streams s2://my-basin/ --prefix device-

{
  "streams": [
    {
      "name": "device-001",
      "created_at": "2024-03-03T10:00:00Z",
      "deleted_at": null
    },
    {
      "name": "device-002",
      "created_at": "2024-03-03T10:15:00Z",
      "deleted_at": null
    }
  ],
  "has_more": false
}

​

Checking the tail

curl https://api.s2.dev/streams/my-stream/tail \
  -H "S2-Basin: my-basin" \
  -H "Authorization: Bearer ${S2_ACCESS_TOKEN}"

{
  "tail": {
    "seq_num": 42,
    "timestamp": 1709481600000
  }
}

​

Deletion

s2 delete-stream s2://my-basin/my-stream

​

Streamer implementation

struct Streamer {
    db: slatedb::Db,
    stream_id: StreamId,
    config: OptionalStreamConfig,
    fencing_token: CommandState<FencingToken>,
    trim_point: CommandState<RangeTo<SeqNum>>,
    append_futs: FuturesOrdered<...>,
    pending_appends: append::PendingAppends,
    stable_pos: StreamPosition,
    follow_tx: broadcast::Sender<...>,
    bgtask_trigger_tx: broadcast::Sender<...>,
}

• Serializes appends: Ensures strict ordering of writes
• Assigns sequence numbers: Manages the tail position
• Enforces fencing: Implements conditional write semantics
• Broadcasts to followers: Notifies real-time readers of new records
• Manages durability: Coordinates with SlateDB for persistence

​

Stream storage

/// (SM) per-stream, updatable
/// Key: BasinName \0 StreamName
/// Value: StreamMeta
StreamMeta(BasinName, StreamName),

/// (SIM) per-stream, immutable
/// Key: StreamID
/// Value: BasinName \0 StreamName
StreamIdMapping(StreamId),

/// (SP) per-stream, updatable
/// Key: StreamID
/// Value: SeqNum Timestamp WriteTimestampSecs
StreamTailPosition(StreamId),

/// (SFT) per-stream, updatable, optional
/// Key: StreamID
/// Value: FencingToken
StreamFencingToken(StreamId),

/// (STP) per-stream, updatable, optional
/// Key: StreamID
/// Value: NonZeroSeqNum
StreamTrimPoint(StreamId),

​

Performance considerations

​

Append throughput

• Pipelining: Multiple appends can be in-flight simultaneously
• Batch size: Up to 1000 records per batch, max 1 MiB metered bytes
• Backpressure: Semaphore-based flow control prevents overwhelming storage

​

Read patterns

• Sequential reads: Most efficient, leverages SlateDB’s LSM structure
• Timestamp-based: Requires secondary index lookup
• Tail reads: Can follow live stream with minimal latency

​

Best practices

1. Choose meaningful names: Use hierarchical naming for related streams
2. Set appropriate retention: Balance storage costs with data retention needs
3. Configure timestamping: Match client/server timestamp behavior to your use case
4. Monitor tail lag: Track the difference between append and read positions
5. Use fencing tokens: Implement exactly-once semantics for critical workflows

​

Next steps