# Issue a new access token.
Source: https://s2.dev/docs/api/access-tokens/issue

post /access-tokens
Issue a scoped S2 access token for specific resources and operations.

> [card] [Access token concepts](/docs/concepts/access-tokens)
>
> Review token scopes, permissions, expiry, and revocation behavior.

> [!NOTE]
>
> Issued tokens use the scope in this request. Scope fields and `auto_prefix_streams`
> are not inherited from the caller, and requested access must be no broader than
> the caller's effective permissions. See the [token issuance scope
> rules](/docs/concepts/access-tokens#narrowing).


# List access tokens.
Source: https://s2.dev/docs/api/access-tokens/list

get /access-tokens
List S2 access tokens with prefix filtering and cursor-based pagination.

> [card] [Access token concepts](/docs/concepts/access-tokens)
>
> Review token scopes, permissions, expiry, and revocation behavior.

> [!TIP]
>
> Model paging by specifying `start_after` as the last access token ID that was returned.


# Revoke an access token.
Source: https://s2.dev/docs/api/access-tokens/revoke

delete /access-tokens/{id}
Revoke an S2 access token by ID.

> [card] [Access token concepts](/docs/concepts/access-tokens)
>
> Review token scopes, permissions, expiry, and revocation behavior.

> [!WARNING]
>
> Revocation can take up to 10 minutes to propagate, due to caching.


# Create a basin.
Source: https://s2.dev/docs/api/basins/create

post /basins
Create a new S2 basin with a name, location, and optional configuration.

# Delete a basin.
Source: https://s2.dev/docs/api/basins/delete

delete /basins/{basin}
Delete an S2 basin and all streams and records in it.

> [!NOTE]
>
> Deletion is asynchronous, and it can take a few minutes to complete.
>
> During that time, a new basin with the same name will not be allowed.


# Ensure a basin.
Source: https://s2.dev/docs/api/basins/ensure

put /basins/{basin}
Ensure an S2 basin exists with the specified configuration. If it already exists, only its configuration is updated.

# Get basin configuration.
Source: https://s2.dev/docs/api/basins/get-config

get /basins/{basin}
Get an S2 basin's current configuration.

# List basins.
Source: https://s2.dev/docs/api/basins/list

get /basins
List S2 basins with prefix filtering and cursor-based pagination.

> [!TIP]
>
> Model paging by specifying `start_after` as the last basin name that was returned.


# Reconfigure a basin.
Source: https://s2.dev/docs/api/basins/reconfigure

patch /basins/{basin}
Update only provided fields in an S2 basin's configuration; omitted fields are left unchanged.

# Endpoints
Source: https://s2.dev/docs/api/endpoints

Public HTTP endpoints for S2 account-level and per-basin APIs.

S2 public endpoints support HTTP/2 and HTTP/1.1 over TLS 1.3.

> [!TIP]
>
> Use the basin-level endpoint for operating on streams (`/streams`, `/records`), and the account-level endpoint for everything else.
>
> [SDKs](/docs/sdk) will take care of this automatically.

> [card] Account
>
> ```
> https://a.s2.dev
> ```

> [card] Streams and Records
>
> ```
> https://{basin}.b.s2.dev
> ```

> [!NOTE]
>
> [PrivateLink](/docs/platform/private-networking) information is available in the dashboard.

> [!NOTE]
>
> Basin names are globally unique.
>
> Basins are placed in a specific *"location"* when they are created. The currently available locations are `aws:us-east-1`, `aws:us-west-2`, and `aws:eu-north-1`. In the future, you can also expect support for other clouds like GCP.


# Error Codes
Source: https://s2.dev/docs/api/error-codes

Error codes returned by the S2 API, including retryability and structured error bodies.

Conventional HTTP status codes are returned to indicate the success or failure of an API request.

Certain errors are retryable, meaning that re-issuing the same request after a delay could lead to a successful result. Exponential backoff is recommended.

> [!NOTE]
>
> In the context of an [S2S](/docs/api/protocol#sessions) `append` or `read` session, an error that occurs after a `200 OK` status will be a "terminal message" on the response stream. However, pre-flight errors are sent normally.

## Standard Errors

A standard error body has the following JSON form:

```json theme={null}
{
  "code": "brief_reason",
  "message": "more information about the error"
}
```

| Code | HTTP Status | Retryable | Side Effect |
| - | - | - | - |
| `bad_header` | 400 | No | No |
| `bad_path` | 400 | No | No |
| `bad_query` | 400 | No | No |
| `bad_json` | 400 | No | No |
| `bad_proto` | 400 | No | No |
| `bad_frame` | 400 | No | No |
| `decryption_failed` | 400 | No | No |
| `authn` | 401 | No | No |
| `invalid` | 422 | No | No |
| `permission_denied` | 403 | No | No |
| `quota_exhausted` | 403 | No | No |
| `basin_not_found` | 404 | No | No |
| `stream_not_found` | 404 | No | No |
| `access_token_not_found` | 404 | No | No |
| `request_timeout` | 408 | Yes | Maybe |
| `resource_already_exists` | 409 | No | No |
| `basin_deletion_pending` | 409 | No | No |
| `stream_deletion_pending` | 409 | No | No |
| `transaction_conflict` | 409 | Yes | No |
| `rate_limited` | 429 | Yes | No |
| `client_hangup` | 499 | No | Maybe |
| `other` | 500 | Yes | Maybe |
| `storage` | 500 | Yes | Maybe |
| `not_implemented` | 501 | No | No |
| `hot_server` | 502 | Yes | No |
| `server_draining` | 503 | Yes | No |
| `unavailable` | 503 | Yes | Maybe |
| `upstream_timeout` | 504 | Yes | Maybe |

> [!TIP]
>
> Requests that fail with `404` or `409` will succeed on a future attempt once the underlying condition changes (e.g., resource is created, or deletion completes).

## Structured Errors

### Append Condition Failed (412)

See [concurrency control](/docs/concepts/concurrency-control) for more context.

> [tab] Fencing Token Mismatch
>
> ```json theme={null}
> {
>   "fencing_token_mismatch": "the-actual-fencing-token"
> }
> ```
>
> The current fencing token did not match the expected fencing token, which is returned.

> [tab] Sequence Number Mismatch
>
> ```json theme={null}
> {
>   "seq_num_mismatch": 42
> }
> ```
>
> The expected sequence number did not match the assignable sequence number, which is returned.

### Read Range Not Satisfiable (416)

`416` without a `Range` header? Unconventional, but the shoe fits.

See [when reads return `416`](/docs/api/records/read#when-reads-return-416).

```json theme={null}
{
  "tail": {
    "seq_num": 42,
    "timestamp": 1767996832000
  }
}
```

The `tail` field indicates the current end position of the stream.


# Get the default location.
Source: https://s2.dev/docs/api/locations/get-default

get /locations/default
Get the default S2 location for new basins.

> [card] [Basin location concepts](/docs/concepts/basins#location)
>
> Review how locations affect basin placement, latency, and account defaults.


# List locations.
Source: https://s2.dev/docs/api/locations/list

get /locations
List S2 locations available to the account.

> [card] [Basin location concepts](/docs/concepts/basins#location)
>
> Review how locations affect basin placement, latency, and account defaults.


# Set the default location.
Source: https://s2.dev/docs/api/locations/set-default

put /locations/default
Set the default S2 location for new basins.

> [card] [Basin location concepts](/docs/concepts/basins#location)
>
> Review how locations affect basin placement, latency, and account defaults.


# Account-level metrics.
Source: https://s2.dev/docs/api/metrics/account

get /metrics
Get account-level S2 metrics for active basins, account operations, and estimated billable usage.

> [card] [Metrics concepts](/docs/concepts/metrics)
>
> Review metric sets, parameters, scopes, and response types.

### Metric sets

#### `active-basins`

Returns a `label` metric representing all basins which had at least one stream within the specified period.

This can be used to determine which basins to request basin-specific metrics for, when attempting to summarize account-level metrics for a period.

Requires a `start` and `end` timestamp.

#### `account-ops`

Returns a set of `accumulation` metrics, one per account operation. Each account-level request is a single operation, and this endpoint will return as many timeseries as there are distinct account operations with non-zero values within the specified period.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.

> [!NOTE]
>
> Requests for `account-metrics`, `basin-metrics`, and `stream-metrics` are also account operations that are metered the same way!

#### `estimated-billable-usage`

Returns account-wide `accumulation` metrics of estimated charges by [pricing category](https://s2.dev/pricing). The unit is `usd`: values are fractional dollars, not cents.

Series names use the `ops.*`, `storage.*`, and `transfer.*` prefixes for operations, storage, and transfers. `other` captures uncategorized charges, and `total` sums all categories. Do not add `total` to the category series when calculating overall usage.

Requires `start` and `end` timestamps, with a maximum range of 90 days. Accepts `interval=hour` (the default) or `interval=day`; minute intervals are not supported.

```bash theme={null}
curl \
  --request GET \
  --url "https://a.s2.dev/v1/metrics?set=estimated-billable-usage&start=${START}&end=${END}&interval=day" \
  --header "Authorization: Bearer ${S2_ACCESS_TOKEN}"
```

The same data is available in the [dashboard's Billing tab](https://s2.dev/dashboard/billing), with daily totals and hourly drill-down.

> [!NOTE]
>
> Only finalized hours are included, so a daily point may be incomplete. Estimates reflect gross usage; the actual invoice may differ.


# Basin-level metrics.
Source: https://s2.dev/docs/api/metrics/basin

get /metrics/{basin}
Get basin-level S2 metrics for storage, operations, and throughput.

> [card] [Metrics concepts](/docs/concepts/metrics)
>
> Review metric sets, parameters, scopes, and response types.

### Metric sets

#### `storage`

`gauge` of cumulative stored bytes, per hour, aggregated across all streams within the basin.

Requires a `start` and `end` timestamp.

#### `append-ops`

Returns `accumulation` of append operations aggregated across all streams within the basin, grouped by storage class.

An append operation is defined as one minute of appends to a single stream, using a single TCP connection.

For instance, 10 unary appends to stream `X`, each initializing a new connection, would be 10 append operations. 10 unary or streaming session appends to stream `X`, on a single TCP connection, within a single minute, would be 1 append operation.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.

#### `read-ops`

Returns two `accumulation` metrics of read operations, aggregated across all streams within the basin: one for `unary` reads, and another for `streaming` reads.

A streaming read operation is defined as up to one minute of a streaming read response, from a single stream, using a single TCP connection.

A unary read operation is a single read request that does not return a streaming body.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.

#### `read-throughput`

`accumulation` of read throughput in bytes, aggregated across all streams within the basin.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.

> [!NOTE]
>
> This metric set does not currently separate egress traffic by client origin, which will be a [dimension for billing](https://s2.dev/pricing).

#### `append-throughput`

Returns `accumulation` metrics of appended throughput in bytes, aggregated across all streams within the basin and grouped by storage class.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.

#### `basin-ops`

Set of `accumulation` metrics, one per basin operation. Each basin-level request is a single operation, and this endpoint will return as many timeseries as there are distinct basin operations with non-zero values within the specified period.

Requires a `start` and `end` timestamp. Accepts an optional `interval` parameter for `minute` / `hour` / `day` aggregations.


# Stream-level metrics.
Source: https://s2.dev/docs/api/metrics/stream

get /metrics/{basin}/{stream}
Get storage metrics for an individual S2 stream.

> [card] [Metrics concepts](/docs/concepts/metrics)
>
> Review metric sets, parameters, scopes, and response types.

### Metric sets

#### `storage`

Returns a `gauge` representing the cumulative stored bytes, per minute, of the specified stream.

Requires a `start` and `end` timestamp.


# Protocol
Source: https://s2.dev/docs/api/protocol

S2 v1 HTTP API for clients, including authentication, compression, and data plane protocol details.

Clients interact with S2 at its public [`https` endpoints](/docs/api/endpoints), primarily with JSON bodies conforming to its [OpenAPI spec](https://github.com/s2-streamstore/s2-specs/blob/main/s2/v1/openapi.json).

> [!TIP]
>
> Prefer to use an available [SDK](/docs/sdk).

## Authentication

The API requires a `Bearer` [access token](/docs/concepts/access-tokens) to authenticate requests.

S2 SDKs take care of supplying it automatically. If you are using `curl`, you can provide it with `-H "Authorization: Bearer ${S2_ACCESS_TOKEN}"`.

> [!NOTE]
>
> The first usage of an access token on a connection may experience a latency overhead for verification, typically tens of milliseconds. Subsequent requests will not incur such an overhead.

## Compression

To optimize network usage, S2 supports compressed request and response bodies with the following algorithms:

* `zstd` *preferred*
* `gzip`

Standard HTTP semantics around `Content-Encoding` and `Accept-Encoding` apply.

## Data Plane

There are 3 core data operations on a [stream](/docs/concepts/streams):

> [card] [Append records](/docs/api/records/append)
>
> `POST /streams/{stream}/records`

> [card] [Read records](/docs/api/records/read)
>
> `GET /streams/{stream}/records?seq_num=42&count=100`

> [card] [Check the tail](/docs/api/records/check-tail)
>
> `GET /streams/{stream}/records/tail`

### Data in JSON

The following pieces of record data are stored as bytes:

* Header name
* Header value
* Body

The header `s2-format` is used to indicate the desired encoding of these bytes when records are represented in JSON.

> [card] raw
>
> `s2-format: raw` or omit header.
>
> *Use when your record data is valid Unicode.*
>
>  Zero overhead, human-readable.
>
>  Cannot handle binary data safely.

> [card] base64
>
> `s2-format: base64`
>
> *Use when you are working with arbitrary bytes.*
>
>  Always safe.
>
>  33% overhead over the wire.

> [!WARNING]
>
> You can write using one format and read with another. When reading `raw`, S2 is interpreting the stored bytes as UTF-8. This will be a potentially-lossy conversion if it was not also written as `raw`, or as `base64`-encoded valid UTF-8.

### Protobuf messages

Data plane endpoints to [`append`](/docs/api/records/append) and [`read`](/docs/api/records/read) records also support [protobuf](https://protobuf.dev) bodies. This helps avoid the base64 encoding tax compared to binary data in JSON messages.

To send and receive protobuf:

* Set the `Content-Type` header to `application/protobuf` and send a protobuf-encoded payload.
* Set the `Accept` header to `application/protobuf` to receive a protobuf-encoded response. The response will include the `Content-Type: application/protobuf` header if the server returns a protobuf.

Type definitions are available in [ git](https://github.com/s2-streamstore/s2-specs/blob/main/s2/v1/s2.proto) and [Buf](https://buf.build/streamstore/s2/docs/main:s2.v1).

> [card] Append
>
>  [`POST ../{stream}/records`](/docs/api/records/append)
>
>  [`AppendInput`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.AppendInput)
>
>  [`AppendAck`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.AppendAck)

> [card] Read
>
>  [`GET ../{stream}/records?..`](/docs/api/records/read)
>
>  [`ReadBatch`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.ReadBatch)

> [!NOTE]
>
> Sending the `Accept: application/protobuf` request header only guarantees a protobuf response in case of a success (HTTP 200). Other status codes are always accompanied by JSON bodies.

### Sessions

S2S (`S2-Session`) is a minimal binary protocol used by S2 SDKs to encapsulate streaming `append` and `read` session semantics over HTTP/2 (HTTP/3 in future, too!), at the same endpoints as "unary" calls.

It is meant for S2 SDKs, but documented for the adventurous.

> [accordion] S2S spec
>
> #### Setup
>
> * `Content-Type: s2s/proto` signals that a session is being requested.
> * `Accept-Encoding` signals which compression algorithms are supported (service supports `zstd` and `gzip`). `Content-Encoding` is not sent as message-level compression is used.
> * `200 OK` response establishes a session.
>
> #### Message framing
>
> *All integers use big-endian byte order. Messages smaller than 1KiB should not be compressed.*
>
> **Length prefix** (3 bytes): Total message length (flag + body)
>
> **Flag byte** (1 byte): `[T][CC][A][RRRR]`
>
> * `T` (bit 7): Terminal flag (`1` = stream ends after this message)
> * `CC` (bits 6-5): Compression (`00`=`none`, `01`=`zstd`, `10`=`gzip`)
> * `A` (bit 4): Reconnect advised (`1` = server suggests reconnecting, only on messages from *Server → Client*)
> * `RRRR` (bits 3-0): Reserved
>
> **Body** (variable):
>
> * Regular message is a Protobuf
> * Terminal message contains a 2-byte status code, followed by JSON error information (corresponding to unary response behavior)
>
> #### Data flow
>
> **Append** sessions are a bi-directional stream of [`AppendInput`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.AppendInput) messages from *Client → Server*, and [`AppendAck`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.AppendAck) messages from *Server → Client*.
>
> **Read** sessions are a uni-directional stream of [`ReadBatch`](https://buf.build/streamstore/s2/docs/main:s2.v1#s2.v1.ReadBatch) messages from *Server → Client*. When waiting for new records, an empty batch is sent as a heartbeat at least every 15 seconds.
>
> #### Reconnect advice
>
> A server that is about to terminate sets the reconnect-advised flag on regular messages.
>
> * **Read** sessions should be re-established with a fresh request.
> * **Append** sessions should stop sending inputs and half-close the request stream. The server acknowledges all accepted inputs and then ends the session cleanly. A new session can be established concurrently, and should start being used once all acknowledgements for in-flight appends have been received.
> * Append sessions still attached when the server drains are ended by the server: it stops reading inputs, acknowledges accepted inputs, and sends a terminal `503` with error code `server_draining`. Acknowledgements always precede the terminal message, so an input is either acknowledged or was never processed, and unresolved inputs can be safely resubmitted on a new session.

> [!TIP]
>
> Read sessions are also supported over [SSE](/docs/api/records/read#sse) which has the benefit of being usable in the browser context.


# Append records.
Source: https://s2.dev/docs/api/records/append

post /streams/{stream}/records
Append a batch of records atomically to an S2 stream, with optional append conditions.

> [card] [Append concepts](/docs/concepts/appends)
>
> Understand append acknowledgements, batching, and durability.

> [card] [Concurrency control](/docs/concepts/concurrency-control)
>
> Coordinate writers with match sequence numbers and fencing tokens.

> [card] [Appending SDK](/docs/sdk/appending)
>
> Use append sessions, the Producer API, batching, and backpressure.

## Stream config on auto-creation

If the basin has `create_stream_on_append` [configured](/docs/concepts/configs#basin-config), a stream that does not exist is created on first append with the basin's default stream config.

The `s2-stream-config` header can override that default for the new stream. Its value is a JSON-encoded [`StreamConfig`](/docs/concepts/configs#stream-config), the same object accepted by [create](/docs/api/streams/create). Unset fields inherit the basin's default stream config.

The header is only applied when the append creates the stream. It is ignored if the stream already exists, so it is safe to send on every request. To change the config of an existing stream, use [reconfigure](/docs/api/streams/reconfigure).

```bash Request theme={null}
curl --request POST \
  --url 'https://{basin}.b.s2.dev/v1/streams/{stream}/records' \
  --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  --header 'Content-Type: application/json' \
  --header 's2-stream-config: {"retention_policy":{"age":3600},"delete_on_empty":{"min_age_secs":300}}' \
  --data '{
    "records": [
      { "body": "started" }
    ]
  }'
```

A header value that is not a valid `StreamConfig` is rejected with a `400` [`bad_header`](/docs/api/error-codes) error. The same header is accepted on [read](/docs/api/records/read#stream-config-on-auto-creation) for basins with `create_stream_on_read`.


# Check the tail.
Source: https://s2.dev/docs/api/records/check-tail

get /streams/{stream}/records/tail
Check the current tail of an S2 stream.

> [card] [Read interfaces](/docs/concepts/reads#interfaces)
>
> See when to use reads, read sessions, or check-tail calls.

> [card] [Stream positions](/docs/concepts/streams#positions)
>
> Head, tail, sequence numbers, and timestamps.

> [card] [Reading SDK](/docs/sdk/reading#check-tail)
>
> Check the tail from SDK clients.


# Read records.
Source: https://s2.dev/docs/api/records/read

get /streams/{stream}/records
Read records from an S2 stream by sequence number, timestamp, or tail offset.

> [card] [Read concepts](/docs/concepts/reads)
>
> Understand starting points, bounds, wait behavior, and sessions.

> [card] [Reading SDK](/docs/sdk/reading)
>
> Use single-batch reads, read sessions, and check-tail calls.

Use this endpoint for a single-batch read or a streaming read session.

Streaming sessions may be [S2S](/docs/api/protocol#sessions)-based (used by SDK [`readSession`](/docs/sdk/reading#read-session) APIs) or [SSE responses](#sse).

## Parameters

Query parameters choose [where the read starts](#starting-point), [what bounds apply](#bounds), and [whether the read waits at the tail](#waiting-at-the-tail).

### Starting point

A request can specify at most one starting selector. If no selector is provided, the start is equivalent to `tail_offset=0`.

| Query parameter | Unit | Starts at |
| - | - | - |
| `seq_num` | Sequence number | The first record whose sequence number is greater than or equal to `seq_num`. |
| `timestamp` | Milliseconds since Unix epoch | The first record whose timestamp is greater than or equal to `timestamp`. |
| `tail_offset` | Record count | The position `tail.seq_num - tail_offset`, saturated at the start of the stream. |

`tail_offset=0` starts at the tail. The tail is the next sequence number that will be assigned, so there is no existing record at that position.

If a requested `seq_num` or `timestamp` is beyond the current tail, `clamp=true` moves the start to the current tail. This is most useful with `wait > 0`, where the read can then wait for future records. Without `clamp`, starts beyond the tail return `416 Range Not Satisfiable`.

### Bounds

Bounds limit how far a read may continue after its starting point.

| Query parameter | Unit | Behavior |
| - | - | - |
| `count` | Records | Stop after this many records. For single-batch reads, the effective maximum is 1000 records. |
| `bytes` | Bytes | Limit the total metered bytes returned. For single-batch reads, the effective maximum is 1 MiB. |
| `until` | Timestamp | Stop before records whose timestamp is greater than or equal to `until`. |

When multiple bounds are present, the read stops at the first bound reached.

### Waiting at the tail

`wait` is how many seconds the read may wait at the tail for new records after all immediately available records have been returned.

| Request mode | Default `wait` | Tail behavior |
| - | - | - |
| Single-batch read | `0` | Return immediately unless `wait > 0`. |
| Session with `count`, `bytes`, or `until` | `0` | Stop when a bound is reached or when the session catches up to the current tail, unless `wait > 0`. |
| Session without `count`, `bytes`, or `until` | Infinite | Follow the stream indefinitely, waiting at the tail for new records. |

A single-batch read with `wait > 0` may long poll for up to 60 seconds. If the wait elapses before records arrive, the response is an empty batch.

For sessions, `wait` applies throughout the session while still respecting any `count`, `bytes`, or `until` bound.

## Stream config on auto-creation

If the basin has `create_stream_on_read` [configured](/docs/concepts/configs#basin-config), a stream that does not exist is created on first read with the basin's default stream config.

The `s2-stream-config` header can override that default for the new stream. Its value is a JSON-encoded [`StreamConfig`](/docs/concepts/configs#stream-config), the same object accepted by [create](/docs/api/streams/create). Unset fields inherit the basin's default stream config.

The header is only applied when the read creates the stream. It is ignored if the stream already exists, so it is safe to send on every request. To change the config of an existing stream, use [reconfigure](/docs/api/streams/reconfigure).

```bash Request theme={null}
curl --request GET \
  --url 'https://{basin}.b.s2.dev/v1/streams/{stream}/records?tail_offset=0&wait=60' \
  --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  --header 's2-stream-config: {"retention_policy":{"age":3600},"delete_on_empty":{"min_age_secs":300}}'
```

A header value that is not a valid `StreamConfig` is rejected with a `400` [`bad_header`](/docs/api/error-codes) error. The same header is accepted on [append](/docs/api/records/append#stream-config-on-auto-creation) for basins with `create_stream_on_append`.

## When reads return 416

A read returns `416 Range Not Satisfiable` when it cannot return records under the request's effective `wait` and `clamp` behavior. The [`416` response body](/docs/api/error-codes#read-range-not-satisfiable-416) is a `TailResponse` containing the current tail.

| Request behavior | `416` when |
| - | - |
| Effective `wait > 0` | The start is beyond the tail and `clamp` is omitted or `false`: `seq_num > tail.seq_num` or `timestamp > tail.timestamp`. |
| Effective `wait = 0` | No record is available immediately: the stream is empty, `tail_offset = 0`, `seq_num >= tail.seq_num`, or `timestamp > tail.timestamp`. |

**Starting exactly at the tail is valid**; with `wait > 0`, the read waits for future records, and with `wait = 0`, it returns `416` because no record exists at the tail yet.

> [!NOTE]
>
> `clamp=true` only changes starts that are beyond the tail. It moves the start to the current tail, not to the last existing record.

## SSE

Server push with [Server-Sent Events](https://html.spec.whatwg.org/multipage/server-sent-events.html) (SSE) is supported. Clients can request an SSE response by setting the `Accept: text/event-stream` header.

```bash Request theme={null}
curl --request GET \
  --url 'https://hello-world.b.s2.dev/v1/streams/test/records?seq_num=0' \
  --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  --header 'Accept: text/event-stream'
```

```bash Batch theme={null}
event: batch
id: 4,5,74
data: {"records":[{"seq_num":4,"timestamp":1750101582144,"body":"hello"}],"tail":{"seq_num":5,"timestamp":1750101582144}}
```

```bash Ping theme={null}
event: ping
data: {"timestamp":1750101582144}
```

```bash Error theme={null}
event: error
data: Something went wrong
```

```bash Done theme={null}
data: [DONE]
```

Any `error` message will terminate the stream.


# Create a stream.
Source: https://s2.dev/docs/api/streams/create

post /streams
Create a new S2 stream in a basin.

> [!NOTE]
>
> Stream encryption is not configured explicitly when creating a stream. A stream inherits the basin's configured `stream_cipher` when it is created, and that cipher is immutable for the lifetime of the stream.


# Delete a stream.
Source: https://s2.dev/docs/api/streams/delete

delete /streams/{stream}
Delete an S2 stream and its records.

> [!NOTE]
>
> Deletion is an asynchronous operation, and it can take a few minutes to complete.
>
> During that time, a new stream by the same name will not be allowed.


# Ensure a stream.
Source: https://s2.dev/docs/api/streams/ensure

put /streams/{stream}
Ensure an S2 stream exists with the specified configuration. If it already exists, only its configuration is updated.

# Get stream configuration.
Source: https://s2.dev/docs/api/streams/get-config

get /streams/{stream}
Get an S2 stream's current configuration.

> [!NOTE]
>
> This endpoint returns reconfigurable stream settings such as storage class and retention. It does not expose the stream's immutable `cipher` which can be inspected by [listing](/docs/api/streams/list).


# List streams.
Source: https://s2.dev/docs/api/streams/list

get /streams
List streams in an S2 basin with prefix filtering and cursor-based pagination.

> [!TIP]
>
> Model paging by specifying `start_after` as the last stream name that was returned.


# Reconfigure a stream.
Source: https://s2.dev/docs/api/streams/reconfigure

patch /streams/{stream}
Update only provided fields in an S2 stream's configuration; omitted fields are left unchanged.

# Access Tokens
Source: https://s2.dev/docs/cli/access-tokens

Issue, list, and revoke scoped access tokens with the S2 CLI.

### List access tokens

```bash theme={null}
s2 list-access-tokens
```

### Issue an access token

```bash theme={null}
s2 issue-access-token my-token \
  --expires-in 30d \
  --basins-prefix "" \
  --streams-prefix "" \
  --op-group-perms "stream=rw"
```

Scoped to specific basin and stream prefixes:

```bash theme={null}
s2 issue-access-token scoped-token \
  --basins-prefix "my-basin" \
  --streams-prefix "events/" \
  --op-group-perms "stream=rw" \
  --expires-in 7d
```

### Revoke an access token

```bash theme={null}
s2 revoke-access-token my-token
```


# Append
Source: https://s2.dev/docs/cli/append

Append newline-delimited records from stdin, files, or pipelines with the S2 CLI, including options for concurrency control and encrypted streams.

Append records to a stream. Each newline-delimited input becomes one record.

### From stdin

```bash theme={null}
echo -e "hello world\nfoo bar\nthird record" \
  | s2 append s2://my-basin/my-stream
```

```
✓ [APPENDED] 0..3 // tail: 3 @ 1771390606217
```

### From a file

```bash theme={null}
s2 append s2://my-basin/my-stream -i records.txt
```

### Interactive

Start an interactive append session where each line you type becomes a record:

```bash theme={null}
s2 append s2://my-basin/my-stream
```

### Piping from another process

```bash theme={null}
nc starwars.s2.dev 23 | s2 append s2://my-basin/starwars
```

### Encrypted streams

```bash theme={null}
export S2_ENCRYPTION_KEY="$(openssl rand -base64 32)"

echo "top secret" | s2 append s2://my-basin/my-stream
```

Or provide the key explicitly:

```bash theme={null}
echo "top secret" | s2 append s2://my-basin/my-stream \
  --encryption-key-file ./stream-key.b64
```

### Options

| Flag | Description |
| - | - |
| `-f, --fencing-token` | Enforce fencing token. |
| `-m, --match-seq-num` | Enforce that the sequence number issued to the first record matches. |
| `--format` | Input format. |
| `--linger` | How long to wait for more records before flushing a batch. |
| `-i, --input` | Input newline delimited records to append from a file or stdin. Use `-` to read from stdin. |
| `-k, --encryption-key` | Base64-encoded encryption key material. Alternatively, set `S2_ENCRYPTION_KEY`. |
| `--encryption-key-file` | Read base64-encoded encryption key material from file. |


# Apply
Source: https://s2.dev/docs/cli/apply

Apply a JSON spec with the S2 CLI to create or update basins and streams, with a dry run before making changes.

Reads a JSON spec file and ensures the declared basins and streams exist with the specified configuration. Resources that already exist are reconfigured to match the spec, and only the fields present in the spec are updated.

```bash theme={null}
s2 apply -f spec.json
```

## Flags

| Flag | Description |
| - | - |
| `-f, --file` | Path to the JSON spec file (required unless `--schema` is used) |
| `--dry-run` | Preview changes without making any mutations |
| `--schema` | Print the JSON Schema for the spec format to stdout |

For the spec file format, configuration fields, IDE setup, and dry run output, see [Declarative Spec](/docs/infra/declarative-spec).


# Authentication
Source: https://s2.dev/docs/cli/authentication

Log in to the S2 CLI, manage access tokens, and switch between authentication methods.

The CLI can authenticate through browser login or with an S2 access token. Both can remain configured at the same time, and you can choose which one the CLI uses.

## Browser login

To sign in through your browser:

```bash theme={null}
s2 login # or s2 auth login
```

The CLI opens the S2 sign-in page in your default browser. After you sign in and approve access, it saves the browser-login credentials to your system keyring.

If the CLI cannot open a browser, use `s2 login --no-open` to print the URL. The browser must still be able to reach the loopback callback on the machine running the CLI.

The CLI does not silently fall back to plaintext if the keyring is unavailable. To explicitly store the browser-login credentials in a private plaintext file, run `s2 login --insecure-storage`.

### Check your authentication

```bash theme={null}
s2 auth status
```

This verifies the active credential with S2 and reports the selected authentication method, other configured methods, and any storage warnings. It exits successfully when S2 authenticates the credential and exits with a non-zero status when authentication cannot be verified.

For loopback endpoints, such as a local S2 Lite server, the command verifies connectivity instead. Credentials are not verified for those endpoints.

### Log out

```bash theme={null}
s2 logout # or s2 auth logout
```

This removes the local browser login and attempts to revoke it with the OAuth provider. It does not remove a separately stored S2 access token.

## S2 access tokens

Generate an access token from the [dashboard](https://s2.dev/dashboard), then store it with:

```bash theme={null}
s2 auth access-token set
```

Paste the token when prompted. The token is saved to your system credential manager.

### Environment override

Set `S2_ACCESS_TOKEN` to use a token without storing it:

```bash theme={null}
S2_ACCESS_TOKEN="$S2_ACCESS_TOKEN" s2 ls
```

The environment variable takes precedence over browser login and a stored access token.

### Standard input

To persist a token supplied by a secret manager or script, pass it through standard input explicitly:

```bash theme={null}
op read 'op://S2/access-token' | s2 auth access-token set --stdin
```

### Private file storage

The CLI does not silently fall back to plaintext when an OS credential store is unavailable. To explicitly use a private local file, add `--insecure-storage`:

```bash theme={null}
printf '%s\n' "$S2_ACCESS_TOKEN" | \
  s2 auth access-token set --stdin --insecure-storage
```

### Migrate an existing token

Access tokens previously saved as `access_token` in `config.toml` remain readable but are deprecated. Move one into the OS credential store with:

```bash theme={null}
s2 auth access-token migrate
```

If an OS credential store is unavailable, migrate it into the private local file instead:

```bash theme={null}
s2 auth access-token migrate --insecure-storage
```

### Remove a stored token

```bash theme={null}
s2 auth access-token remove
```

This removes the local credential but does not revoke the access token on S2. Use [`s2 revoke-access-token`](/docs/cli/access-tokens#revoke-an-access-token) when the token must stop working everywhere.

## Switch authentication methods

Browser login and a stored access token can coexist. Choose which persistent credential the CLI uses:

```bash theme={null}
s2 auth use browser-login
s2 auth use access-token
```

Switching does not delete either credential.

| Action | Credential used afterward |
| - | - |
| `s2 login` | Browser login |
| `s2 auth access-token set` | Stored access token |
| `s2 auth use <method>` | The selected persistent method |
| Set `S2_ACCESS_TOKEN` | Environment token for that process; overrides the selected method |


# Basins
Source: https://s2.dev/docs/cli/basins

Create, list, inspect, reconfigure, and delete basins with the S2 CLI.

### List basins

```bash theme={null}
s2 ls
```

```
my-basin        active
another-basin   active
```

Use `--prefix` to filter and `--limit` to paginate:

```bash theme={null}
s2 ls --prefix "my-" --limit 10
```

### List locations and storage classes

```bash theme={null}
s2 list-locations
```

The output includes each location's available storage classes and default. Use a class available in your chosen location when setting `--storage-class`. Omit it to inherit the location's default for a new basin.

### Create a basin

```bash theme={null}
s2 create-basin my-basin
```

Create in a specific [location](/docs/concepts/basins#location):

```bash theme={null}
s2 create-basin my-basin --location aws:us-east-1
```

> [!TIP]
>
> Basin names must be globally unique, 8–48 characters, lowercase letters/numbers/hyphens.

Options for default stream configuration:

```bash theme={null}
s2 create-basin my-basin \
  --stream-cipher aegis-256 \
  --storage-class express \
  --retention-policy 7d \
  --create-stream-on-append \
  --create-stream-on-read
```

### Get basin config

```bash theme={null}
s2 get-basin-config my-basin
```

### Reconfigure a basin

```bash theme={null}
s2 reconfigure-basin my-basin \
  --stream-cipher aes-256-gcm \
  --create-stream-on-append true \
  --retention-policy 30d
```

> [!NOTE]
>
> `--stream-cipher`: Encryption algorithm to apply to newly created streams in this basin.
>
> Existing streams keep the `cipher` they were created with.

### Delete a basin

```bash theme={null}
s2 delete-basin my-basin
```


# Bench
Source: https://s2.dev/docs/cli/bench

Benchmark stream latency and throughput with the S2 CLI.

```bash theme={null}
s2 bench my-basin --target-mibps 1 --duration 10s
```

| Flag | Description |
| - | - |
| `-c, --storage-class` | Storage-class name for the test stream. Check available classes with [`s2 list-locations`](/docs/cli/basins#list-locations-and-storage-classes). |
| `-b, --record-size` | Record size in bytes (default: 8192) |
| `-t, --target-mibps` | Target write throughput in MiB/s (default: 1) |
| `-d, --duration` | Test duration (default: 60s) |
| `-w, --catchup-delay` | Delay before starting catchup read (default: 20s) |


# Check Tail
Source: https://s2.dev/docs/cli/check-tail

Check a stream's tail with the S2 CLI to find the next sequence number and latest record timestamp.

Returns the next sequence number and the timestamp of the last record:

```bash theme={null}
s2 check-tail s2://my-basin/my-stream
```

```
3 @ 1771390606217
```


# Configuration
Source: https://s2.dev/docs/cli/configuration

Configure S2 CLI endpoint overrides, default location, and encryption keys.

## Managing Config

```bash theme={null}
# List all config values
s2 config list

# Get a specific value
s2 config get account_endpoint

# Unset a value
s2 config unset account_endpoint
```

`s2 config list` reports a configured access token as `<redacted>`, and `s2 config get access_token` does not reveal the credential. Use the [`s2 auth access-token`](/docs/cli/authentication#s2-access-tokens) commands to manage access tokens.

## Default Location

When `create-basin` is called without `--location`, the basin uses the account's default [location](/docs/concepts/basins#location).

```bash theme={null}
s2 list-locations
s2 get-default-location
s2 set-default-location aws:us-east-1
```

## Environment Variables

The CLI also reads endpoint overrides and command-specific settings from environment variables, which is useful when pointing at [s2-lite](/docs/s2-lite) or working with encrypted streams:

| Variable | Description |
| - | - |
| `S2_ACCESS_TOKEN` | Non-persistent access token override; takes precedence over browser login and a stored access token |
| `S2_ACCOUNT_ENDPOINT` | Account-level API endpoint |
| `S2_BASIN_ENDPOINT` | Basin-level API endpoint |
| `S2_ENCRYPTION_KEY` | [Encryption key](/docs/concepts/encryption) used by `append`, `read`, and `tail` commands, if configured |

```bash theme={null}
# Example: pointing at a local s2-lite instance
export S2_ACCOUNT_ENDPOINT="http://localhost:8080"
export S2_BASIN_ENDPOINT="http://localhost:8080"
export S2_ACCESS_TOKEN="ignored"
```

The SDKs can parse these variables too, but only if you use the associated helper. See [SDK: Endpoints](/docs/sdk/endpoints#environment-variables).


# Fence
Source: https://s2.dev/docs/cli/fence

Set a fencing token with the S2 CLI so cooperating writers can avoid conflicting appends.

Set a fencing token for cooperative write exclusion:

```bash theme={null}
s2 fence s2://my-basin/my-stream "writer-1"
```

Subsequent appends that specify a token will be rejected if it doesn't match:

```bash theme={null}
s2 append s2://my-basin/my-stream -f "writer-1"
```


# Installation
Source: https://s2.dev/docs/cli/installation

Install the S2 CLI with Homebrew, Cargo, release binaries, or Docker.

> [tab] Homebrew
>
> Works on macOS and Linux distributions with [Homebrew](https://brew.sh) installed.
>
> Homebrew 6.0 and later require explicit [tap trust](https://docs.brew.sh/Tap-Trust) for third-party taps. Installing with the fully qualified name trusts only the S2 formula:
>
> ```bash theme={null}
> brew install s2-streamstore/s2/s2
> ```
>
> > [!NOTE]
> >
> > The fully qualified name (`s2-streamstore/s2/s2`) scopes trust to that formula only. To trust every current and future package in the tap instead, run `brew trust s2-streamstore/s2` before installing.

> [tab] Cargo
>
> Works on any system with [Rust](https://www.rust-lang.org/) and [Cargo](https://doc.rust-lang.org/cargo/) installed.
>
> ```bash theme={null}
> cargo install --locked s2-cli
> ```

> [tab] Release Binaries
>
> Linux and macOS users can download the latest release binary:
>
> ```bash theme={null}
> curl -fsSL s2.dev/install.sh | bash
> ```
>
> > [!NOTE]
> >
> > The script installs the binary to `~/.s2/bin` and modifies your shell profile to add it to the path. Reload your shell or open a new terminal after installation.
>
> See all [releases on GitHub](https://github.com/s2-streamstore/s2/releases).

> [tab] Docker
>
> ```bash theme={null}
> docker pull ghcr.io/s2-streamstore/s2
> ```

Verify the installation:

```bash theme={null}
s2 --version
```


# Metrics
Source: https://s2.dev/docs/cli/metrics

Query account, basin, and stream metrics with the S2 CLI.

```bash theme={null}
# Account-level metrics
s2 get-account-metrics

# Basin-level metrics
s2 get-basin-metrics my-basin

# Stream-level metrics
s2 get-stream-metrics s2://my-basin/my-stream
```


# CLI
Source: https://s2.dev/docs/cli/overview

Manage resources, read and write records, benchmark streams, and run local development tools with the S2 CLI.

The `s2` CLI provides access to the full S2 API — managing basins, streams, access tokens, and reading/writing records — plus utilities like [benchmarking](/docs/cli/bench) and [s2-lite](/docs/s2-lite).

```bash theme={null}
s2 --help
```

```
Commands:
  login                 Log in to S2 through your browser
  logout                Log out of the browser-authenticated S2 session
  auth                  Manage CLI authentication
  config                Manage CLI configuration
  ls                    List basins or streams in a basin
  list-basins           List basins
  create-basin          Create a basin
  delete-basin          Delete a basin
  get-basin-config      Get basin config
  reconfigure-basin     Reconfigure a basin
  list-access-tokens    List access tokens
  issue-access-token    Issue an access token
  revoke-access-token   Revoke an access token
  diff                  Compare two basins, streams, or access tokens
  list-locations        List locations
  get-default-location  Get the default location
  set-default-location  Set the default location
  get-account-metrics   Get account metrics
  get-basin-metrics     Get basin metrics
  get-stream-metrics    Get stream metrics
  list-streams          List streams
  create-stream         Create a stream
  delete-stream         Delete a stream
  get-stream-config     Get stream config
  reconfigure-stream    Reconfigure a stream
  check-tail            Check the tail position of a stream
  trim                  Set a trim point for a stream
  fence                 Set a fencing token for a stream
  append                Append records to a stream
  read                  Read records from a stream
  tail                  Tail a stream, showing the last N records
  bench                 Benchmark a stream to measure throughput and latency
  apply                 Apply a declarative spec file, ensuring basins and streams
  lite                  Run S2 Lite server backed by object storage
  update                Update the S2 CLI to the latest release
```

## S2 URIs

Most commands that operate on a stream take an **S2 URI** of the form:

```
s2://{basin}/{stream}
```

For example: `s2://my-basin/my-stream`.

The `ls` command also accepts a URI with just a basin and optional prefix for listing streams: `s2://my-basin/prefix`.


# Read
Source: https://s2.dev/docs/cli/read

Read stream records from a sequence number, timestamp, or tail offset with the S2 CLI.

Read records from a stream.

### Read from the beginning

```bash theme={null}
s2 read s2://my-basin/my-stream -s 0 -n 10
```

```
⦿ 54 bytes (3 records in range 0..=2)
hello world
foo bar
third record
```

### JSON output

```bash theme={null}
s2 read s2://my-basin/my-stream -s 0 -n 3 --format json
```

```json theme={null}
{"seq_num":0,"timestamp":1771390606217,"body":"hello world"}
{"seq_num":1,"timestamp":1771390606217,"body":"foo bar"}
{"seq_num":2,"timestamp":1771390606217,"body":"third record"}
```

### Read from a timestamp

```bash theme={null}
# Absolute (milliseconds since epoch)
s2 read s2://my-basin/my-stream --timestamp 1771390606217

# Relative
s2 read s2://my-basin/my-stream --ago 1h
```

### Read from a tail offset

```bash theme={null}
s2 read s2://my-basin/my-stream --tail-offset 100
```

### Follow for new records

Without a limit (`-n` or `-b`), `read` keeps tailing and waits for new records indefinitely:

```bash theme={null}
s2 read s2://my-basin/my-stream -s 0
```

### Encrypted streams

```bash theme={null}
export S2_ENCRYPTION_KEY="$(cat ./stream-key.b64)"

s2 read s2://my-basin/my-stream -s 0 -n 10
```

Or pass the key on the command line:

```bash theme={null}
s2 read s2://my-basin/my-stream \
  --encryption-key-file ./stream-key.b64 \
  -s 0 -n 10
```

> [!NOTE]
>
> The same `--encryption-key` and `--encryption-key-file` flags also apply to `s2 tail`.

### Options

| Flag | Description |
| - | - |
| `-s, --seq-num` | Starting sequence number (inclusive). |
| `--timestamp` | Starting timestamp in milliseconds since Unix epoch (inclusive). |
| `--ago` | Starting timestamp as a human-friendly delta from current time e.g. "1h", which will be converted to milliseconds since Unix epoch. |
| `--tail-offset` | Start from N records before the tail of the stream. |
| `-n, --count` | Limit the number of records returned. |
| `-b, --bytes` | Limit the number of bytes returned. |
| `--until` | Exclusive end-timestamp in milliseconds since Unix epoch. If provided, results will be limited such that all records returned will have a timestamp \< the one provided via `until`. |
| `--clamp` | Clamp the start position at the tail position. |
| `--format` | Output format. |
| `-o, --output` | Output records to a file or stdout. Use `-` to write to stdout. |
| `-k, --encryption-key` | Base64-encoded encryption key material. Alternatively, set `S2_ENCRYPTION_KEY`. |
| `--encryption-key-file` | Read base64-encoded encryption key material from file. |

***

## Tail

Like `tail` on Unix, it shows the last N records from a stream.

```bash theme={null}
s2 tail s2://my-basin/my-stream -n 5
```

Follow for new records with `-f`:

```bash theme={null}
s2 tail s2://my-basin/my-stream -f
```


# Streams
Source: https://s2.dev/docs/cli/streams

Create, list, inspect, reconfigure, and delete streams inside a basin with the S2 CLI.

### List streams

```bash theme={null}
s2 ls s2://my-basin/
```

```
s2://my-basin/events       2025-09-03T02:28:00Z
s2://my-basin/user-actions 2025-09-03T02:28:48Z
```

Filter by prefix:

```bash theme={null}
s2 ls s2://my-basin/user-
```

### Create a stream

```bash theme={null}
s2 create-stream s2://my-basin/my-stream
```

Omitting `--storage-class` uses the basin's default stream configuration. Choose an explicit class from the offerings for the basin's location shown by `s2 list-locations`:

```bash theme={null}
s2 create-stream s2://my-basin/my-stream \
  --storage-class express \
  --retention-policy 1d
```

### Get stream config

```bash theme={null}
s2 get-stream-config s2://my-basin/my-stream
```

### Reconfigure a stream

```bash theme={null}
s2 reconfigure-stream s2://my-basin/my-stream \
  --retention-policy 30d
```

### Delete a stream

```bash theme={null}
s2 delete-stream s2://my-basin/my-stream
```


# Trim
Source: https://s2.dev/docs/cli/trim

Set an explicit trim point with the S2 CLI to remove records before a sequence number.

Set an explicit trim point. Records before this sequence number will be removed.

Read more about explicit trimming in the [Trimming](/docs/concepts/trimming) concept guide.

```bash theme={null}
s2 trim s2://my-basin/my-stream 1000
```

> [!NOTE]
>
> Trimming is eventually consistent. Trimmed records may be visible for a brief period.


# Access Tokens
Source: https://s2.dev/docs/concepts/access-tokens

Access tokens let callers use S2 with only the permissions and lifetime they need.

Access tokens are a first-class resource in S2. Like basins and streams, there is no limit on how many you may issue. Each token can be **scoped** by resource name or prefix and by the operations it may perform.

Access tokens are **revocable** and support an **optional expiry**. Permanent tokens work well for services; time-limited ones suit ephemeral usage like end-user sessions.

You can manage access tokens from the dashboard or through the API:

> [card] [Dashboard](https://s2.dev/dashboard)
>
> ![](https://mintcdn.com/streamstore/I63bbJM2U3m9xZR6/images/issue-access-token.png?fit=max\&auto=format\&n=I63bbJM2U3m9xZR6\&q=85\&s=fa74c7ab773e3c776aa342f3c714f773)

> [card] API
>
> > [!TIP]
> >
> > API requests require an access token, so use the dashboard to issue the first token for an account.
>
> &#x20;[`GET /access-tokens`](/docs/api/access-tokens/list)
>
> &#x20;[`POST /access-tokens`](/docs/api/access-tokens/issue)
>
> &#x20;[`DELETE /access-tokens/{id}`](/docs/api/access-tokens/revoke)

## Properties

### Identifier

Access token IDs must be unique to the account and between 1 and 96 bytes in length.

> [!TIP]
>
> List requests allow prefix filtering, which is handy for namespacing with a delimiter like `/`. For example, you can name them as `user/{user}` and `service/{service}` to easily filter for access tokens for users vs services.

### Expiration

An optional RFC 3339 timestamp at which the access token becomes invalid.

> [!NOTE]
>
> When one access token issues another, the issuing token's expiry is both the default and the latest possible expiry for the new token.
>
> Dashboard access does not have an expiry, so it can issue permanent access tokens. A token with a limited expiry can issue only tokens bounded by its own expiry.

### Scope

#### Resources

Access to the following resources can be scoped:

* Basins
* Streams
* Access tokens

To specify the set of resources for each of these, you can choose either of:

> [tab] Prefix Match
>
> **Grant access to all resources with a common prefix:**
>
> ```json theme={null}
> { "prefix": "logs/" }
> ```
>
> > [!NOTE]
> >
> > Empty prefix matches all resources (allow all).

> [tab] Exact Match
>
> **Grant access to one specific resource:**
>
> ```json theme={null}
> { "exact": "my-stream" }
> ```
>
> > [!NOTE]
> >
> > Empty exact match will not match any resources (deny all).

> [!TIP]
>
> When configuring access to streams with a prefix, you can also enable *auto-prefixing*.
>
> This acts as a transparent logical namespace within a basin:
>
> * Stream name arguments provided in requests will be automatically prefixed.
> * The prefix will be stripped from a response containing a stream name, such as when listing streams.
>
> For example, with `{ "prefix": "user1/" }` and auto-prefixing enabled, a request for `logs` operates on `user1/logs` in S2. Responses expose that stream as `logs`.
>
> Auto-prefixing is only valid with a prefix stream matcher.

Omitted fields use the following least-permissive defaults:

| Field | Default |
| - | - |
| `basins` | No access |
| `streams` | No access |
| `access_tokens` | No access |
| `op_groups` | All permissions `false` |
| `ops` | Empty |
| `auto_prefix_streams` | `false` |

#### Operations

There are two complementary ways to authorize operations:

> [tab] Operation Groups
>
> Operation groups provide a high-level way to grant read/write access.
>
> ```json theme={null}
> {
>   "op_groups": {
>     "account": { "read": false, "write": false },
>     "basin": { "read": true, "write": false },
>     "stream": { "read": true, "write": false }
>   }
> }
> ```
>
> Group-level permissions will also apply to any new operations added to the group.

> [tab] Individual Operations
>
> For more precise control, explicitly specify the allowed operations:
>
> ```json theme={null}
> {
>   "ops": [
>     "read",
>     "check-tail"
>   ]
> }
> ```

> [!NOTE]
>
> Both groups and individual operations may be specified together — the effective permissions are a union.
>
> An issued token must have at least one effective operation permission.

## Narrowing

When you use an access token to issue another token, the requested scope must be
equal to or narrower than the issuing token's scope. S2 rejects a request with
`permission_denied` if it would expand resource access or operation permissions;
it does not silently reduce the requested scope.

The request defines the new token's complete scope. Scope fields are not
inherited, and omitted fields use the least-permissive defaults above.

> [!NOTE]
>
> [Expiration](#expiration) behaves differently: the new token inherits the
> issuing token's expiration by default and can never outlive it.

For resource matchers, this means:

| Issuing token matcher | New token may use | New token may not use |
| - | - | - |
| `{ "prefix": "foo/" }` | `{ "prefix": "foo/" }`, `{ "prefix": "foo/bar/" }`, `{ "exact": "foo/bar" }`, or no access | `{ "prefix": "" }`, `{ "prefix": "bar/" }`, `{ "exact": "bar" }` |
| `{ "exact": "foo/bar" }` | `{ "exact": "foo/bar" }` or no access | `{ "prefix": "foo/" }`, `{ "exact": "foo/baz" }` |
| No access | No access | Any resource access |

These examples use stream names, but the same matcher rules apply to basin names
and access token IDs. The new token's ID must also match the issuing token's
`access_tokens` matcher, independently of the new token's own `access_tokens`
matcher.

Permissions are narrowed across both `op_groups` and `ops`. An operation group
on the issuing token can authorize individual operations in that group, but
individual operations cannot authorize a group because groups automatically
include future operations.

`auto_prefix_streams` also defaults to `false`; it is not inherited. Matchers in
the issuance request always use names as stored in S2, even when the issuing
token uses auto-prefixing. For example, if the issuing token has
`{ "prefix": "foo/" }`, request `{ "prefix": "foo/bar/" }` for the new token,
not `{ "prefix": "bar/" }`. Enable auto-prefixing on the new token separately if
its callers should use names relative to `foo/bar/`.


# Appends
Source: https://s2.dev/docs/concepts/appends

Appends make records durable at a stream's tail and acknowledge where each batch landed.

Streams are append-only (a *log*) so records are always added to the end, or *tail*, of the stream.

When you append records, S2 responds with an acknowledgement only once your data is fully durable with the range of sequence numbers assigned, the timestamps of the first and last record, and the current tail.

```json theme={null}
{
  "start": { "seq_num": 42, "timestamp": 1713812735000 },
  "end":   { "seq_num": 44, "timestamp": 1713812735012 },
  "tail":  { "seq_num": 44, "timestamp": 1713812735012 }
}
```

* `start` — position of the first record appended.
* `end` — one past the last record appended (so `end.seq_num - start.seq_num` is the number of records).
* `tail` — current tail of the stream, which can exceed `end` if there have been concurrent appends.

## Batching

The append API accepts *batches* of records. A single batch can contain up to 1000 records or 1 MiB of data, and is appended **atomically**: either all records in the batch become durable, or none.

For payloads larger than the 1 MiB [record size limit](/docs/platform/limits#records), the typical approach is to store the data externally (e.g. in object storage) and append a pointer to it as a record. You can also serialize large messages across multiple records, using record headers as metadata — see [this blog post](https://s2.dev/blog/patterns) for patterns and examples with the TypeScript SDK.

## Latency

Acknowledgement latency depends on the stream's [configured](/docs/concepts/configs#stream-config) [storage class](/docs/platform/architecture#storage-classes):

* *Standard* streams ack writes within 400 milliseconds.
* *Express* streams ack writes within 40 milliseconds.

Both storage classes make data durable in multiple availability zones before acknowledging the append.

> [!NOTE]
>
> Latency expectations assume an [authenticated client](/docs/api/protocol#authentication) in the same location as the basin.

## Throughput

To optimize throughput,

1. **Send fewer, larger batches.** This is especially important if you have a very high rate of tiny appends, as each client is [limited](/docs/platform/limits#appends) to 200 batches per second.

2. **Keep multiple writes in-flight.** S2's append session support lets a client pipeline multiple batches on a persistent connection while preserving submission order — if any batch fails, subsequent batches won't become durable. Concurrent single-batch appends can also increase throughput, but they are independent requests, so the order in which batches become durable is not guaranteed.

> [!TIP]
>
> If records arrive one at a time, use the SDK [Producer API](/docs/sdk/appending#producer). It keeps the append session ordering model, while handling batching and backpressure for you.

## See also

> [card] [Appending SDK](/docs/sdk/appending)
>
> Use append sessions, the Producer API, batching, and backpressure.

> [card] [Append API](/docs/api/records/append)
>
> Review request parameters, batch limits, and append conditions.

> [card] [Concurrency control](/docs/concepts/concurrency-control)
>
> Coordinate writers with match sequence numbers and fencing tokens.


# Basins
Source: https://s2.dev/docs/concepts/basins

Basins group streams in a location and define the defaults for new streams.

A container for streams, similar to a *bucket* in object storage.

Each basin has a globally unique name and lives in a specific [cell](/docs/platform/cells) within a location. The basin name is reflected in the data plane [endpoint](/docs/api/endpoints):

```
https://{basin}.b.s2.dev
```

## Configuration

Basins carry [configuration](/docs/concepts/configs#basin-config) that governs stream auto-creation and defaults for streams in the basin.

Enabling `create_stream_on_(append|read)` lets you skip explicit stream creation — just start operating and the stream will be created automatically with the basin's default stream config.

A basin can also set the [encryption](/docs/concepts/encryption) algorithm for newly created streams. Each stream captures that cipher when it is created.

## Location

Every basin is characterized by a logical **location**, established on basin creation. This is fixed for the lifetime of the basin. The main impact of location is on latency with clients; generally, the closer you are to the location, the better latency you will experience.

A basin location corresponds to a replication environment (e.g., a region within a cloud provider) in which S2 is deployed.

The currently available locations are `aws:us-east-1`, `aws:us-west-2`, and `aws:eu-north-1`.

A default location can be set at the account level. This location will be used for newly created basins that do not specify a location.

[List locations](/docs/api/locations/list) also returns each location's available storage classes and default storage class. A new basin inherits that storage-class default unless its configuration specifies one. See [configuration defaults](/docs/concepts/configs#storage-class-selection).

A location can also represent a private, [dedicated cell](/docs/platform/cells#tenancy), if so desired. [Contact us](https://s2.dev/contact) if you need this!

## Naming

Streams within a basin are addressed as `s2://{basin}/{stream}`. Stream names can contain `/` (or any other delimiter) to create logical hierarchies:

```
s2://my-basin/sessions/user-123/run-456
s2://my-basin/logs/prod/api-server
```

This is a naming convention only — there is no directory structure. [Listing streams](/docs/api/streams/list) supports prefix filtering, and [access tokens](/docs/concepts/access-tokens#resources) can be scoped to prefixes as well.

## Limits

The number of basins per account is unlimited (for all customers with a payment method; otherwise limited to 100).

Streams within a basin are unlimited.

See [limits](/docs/platform/limits) for details.

## See also

> [card] [Basin SDK](/docs/sdk/basin-resources)
>
> Create, configure, list, and delete basins from SDK clients.

> [card] [Basin API](/docs/api/basins/create)
>
> Review the basin management endpoints.


# Command Records
Source: https://s2.dev/docs/concepts/command-records

Command records are stream records S2 interprets as service-side directives.

Command records signal operations that are interpreted by the service.

S2 [SDKs](/docs/sdk) and the [CLI](/docs/cli) make it easy to create supported command records, so most applications do not need to construct them directly.

Concretely, a command record is a record with:

1. A single header with an empty name. Empty header names are not allowed in any other context.
2. The operation encoded in that header value.
3. The command payload in the record body.

> [!NOTE]
>
> Command records take up a sequence number on the stream, and will be returned to reads. It is easy to test and filter out commands if needed, with the logic `len(headers) == 1 && headers[0].name == b""`.

## Operations

Operations that are currently supported:

* [`fence`](/docs/concepts/concurrency-control#fencing-tokens) — set a fencing token for cooperative write exclusion. Payload is up to 36 UTF-8 bytes; an empty payload clears the token.
* [`trim`](/docs/concepts/trimming#trim-point) — set a trim point to remove older records. Payload is exactly 8 big-endian bytes representing the desired earliest sequence number.

> [accordion] Constructing command records via REST
>
> > [tab] fence
> >
> > ```bash theme={null}
> > curl -X POST "https://{basin}.b.s2.dev/v1/streams/my-stream/records" \
> >     -H "Authorization: Bearer <token>" \
> >     -H "s2-format: raw" \
> >     -H "Content-Type: application/json" \
> >     -d '{
> >       "records": [
> >         {
> >           "headers": [
> >             ["", "fence"]
> >           ],
> >           "body": "producer-123"
> >         }
> >       ]
> >     }'
> > ```
>
> > [tab] trim
> >
> > ```bash theme={null}
> > curl -X POST "https://{basin}.b.s2.dev/v1/streams/my-stream/records" \
> >     -H "Authorization: Bearer <token>" \
> >     -H "s2-format: base64" \
> >     -H "Content-Type: application/json" \
> >     -d '{
> >       "records": [
> >         {
> >           "headers": [
> >             ["", "dHJpbQ=="]
> >           ],
> >           "body": "AAAAAAAAAGQ="
> >         }
> >       ]
> >     }'
> > ```

## See also

> [card] [Concurrency control](/docs/concepts/concurrency-control)
>
> Coordinate writers with match sequence numbers and fencing tokens.

> [card] [Trimming](/docs/concepts/trimming)
>
> Discard older records by setting a stream trim point.

> [card] [Appends](/docs/concepts/appends)
>
> Understand append acknowledgements, batching, and durability.


# Concurrency Control
Source: https://s2.dev/docs/concepts/concurrency-control

Coordinate writes with conditional appends using a match sequence number or fencing token.

Conditional appends coordinate writers against a durable stream. They let a writer append only when the stream is at an expected position or when the writer holds the expected fencing token.

This is useful for single-writer systems that need safe failover. A replacement writer can inspect the stream, determine the next expected position, and use a conditional append to avoid writing against stale state.

## Mechanisms

The [`append` API](/docs/api/records/append) supports two concurrency-control mechanisms:

### Match sequence number

Specifying expected current state allows *optimistic* concurrency control.

You can provide the sequence number that you expect S2 to assign to the first record in a batch as the `match_seq_num`.

If it does not match, this will result in a [`412 Precondition Failed`](/docs/api/error-codes#append-condition-failed-412) status or corresponding SDK error type.

> [!TIP]
>
> Conditional appends with an appropriate [retry policy](/docs/sdk/retries-timeouts#append-retry-policy) can be used to achieve "exactly-once" semantics.

### Fencing tokens

Fencing is a form of *pessimistic* concurrency control.

It is a cooperative mechanism, so an append that does not specify a fencing token will still be allowed.

When an append does include a `fencing_token` and it does not match, this results in a [`412 Precondition Failed`](/docs/api/error-codes#append-condition-failed-412) status or corresponding SDK error type.

> [!TIP]
>
> Setting the fencing token involves appending a `fence` [command record](/docs/concepts/command-records#operations).

#### Fencing example

Set a fencing token via the CLI:

```bash theme={null}
s2 fence s2://my-basin/my-stream "writer-1"
```

Subsequent appends that specify a token will be rejected if it doesn't match:

```bash theme={null}
s2 append s2://my-basin/my-stream -f "writer-1"
```

Fencing tokens can be up to 36 UTF-8 bytes. An empty token clears the fence. Fencing is strongly consistent — once set, subsequent appends specifying a different token are immediately rejected.

## Failure handling

Append condition failures return structured [`412` error bodies](/docs/api/error-codes#append-condition-failed-412) that identify the current fencing token or assignable sequence number.

> [!NOTE]
>
> In an append session, such a failure poisons the session. Establish a new session before continuing to write.

## Ownership patterns

Conditional appends enforce write-side preconditions at S2.

On the read side, any reader can ensure its knowledge of the stream is current by making a [`check-tail`](/docs/api/records/check-tail) call.

This round-trip can be avoided by a single logical owner of a stream using a [lease](https://en.wikipedia.org/wiki/Lease_\(computer_science\)). The lease itself may be modeled on top of the S2 stream by periodically appending application-level heartbeats identifying the current physical owner.

> [!TIP]
>
> AWS MemoryDB is built on an internal replicated log primitive that offers conditional appends. See §4 of [their paper](https://cdn.amazon.science/e0/1b/ba6c28034babbc1b18f54aa8102e/amazon-memorydb-a-fast-and-durable-memory-first-cloud-database.pdf).
>
> S2 exposes this kind of primitive as a serverless cloud resource, so applications can model many durable streams with independent logical owners, including [agent sessions](https://s2.dev/blog/agent-sessions) and [multiplayer rooms](https://s2.dev/blog/durable-yjs-rooms).

## See also

> [card] [Appends](/docs/concepts/appends)
>
> Understand append acknowledgements, batching, and durability.

> [card] [Command records](/docs/concepts/command-records)
>
> See how S2 applies fence and trim directives from records.

> [card] [Append condition failures](/docs/api/error-codes#append-condition-failed-412)
>
> Inspect structured `412` responses for failed append preconditions.


# Configs
Source: https://s2.dev/docs/concepts/configs

Basin and stream configs define storage class, retention, timestamping, encryption, and stream lifecycle behavior.

Basins and streams each have configuration that can be set at creation time and modified later via reconfigure ([streams](/docs/api/streams/reconfigure), [basins](/docs/api/basins/reconfigure)).

The most important choices are:

* **Storage class**: choose a class available in the basin's location. This controls append latency, not durability. All classes are regionally durable. See [Architecture: Storage classes](/docs/platform/architecture#storage-classes).
* **Retention**: how long records stick around. Defaults to 7 days. See [Retention](#retention).
* **Auto-creation**: whether streams are created implicitly on first append or read, useful when you have many short-lived or dynamic streams.
* **Stream encryption**: optionally require request-time encryption keys for newly created streams. See [Encryption](/docs/concepts/encryption).

You can view and edit configs from the CLI:

```bash theme={null}
# view current config
s2 get-stream-config s2://my-basin/my-stream

# reconfigure stream
s2 reconfigure-stream s2://my-basin/my-stream --storage-class express
```

... or from the [dashboard](https://s2.dev/dashboard):

![Basin config in the S2 dashboard](https://mintcdn.com/streamstore/5UkUxn1E8t4PsPVU/images/config.png?fit=max\&auto=format\&n=5UkUxn1E8t4PsPVU\&q=85\&s=e7f222032fa9bb54af895db965318e79)

## Stream config

For the full set of fields, see the [create](/docs/api/streams/create) and [reconfigure](/docs/api/streams/reconfigure) API docs, or the docs for the SDK you are using.

> [!NOTE]
>
> Encryption is configured on the basin, not in `StreamConfig`. A stream captures the basin's `stream_cipher` when it is created and keeps that `cipher` for life. See [Encryption](/docs/concepts/encryption).

A summary:

| Field | Default | Description |
| - | - | - |
| `storage_class` | Inherited from basin | Storage-class name available in the basin's location. See [selection and defaults](#storage-class-selection). |
| `retention_policy` | 7 days | Age-based (in seconds) or `infinite`. |
| `delete_on_empty` | Disabled | Auto-delete empty streams after a minimum age. |
| `timestamping.mode` | `client-prefer` | How timestamps should be assigned: by S2 on `arrival`; from the client, if provided, with `client-prefer`; or always from the client, with `client-require`. |
| `timestamping.uncapped` | `false` | Allow client timestamps to exceed arrival time. |

### Storage class selection

Use [List locations](/docs/api/locations/list) to check each location's available storage classes and default for your account, or use the CLI:

```bash theme={null}
s2 list-locations
```

Choose a class available in the basin's location. Available classes and defaults can differ by location.

* A new basin without an explicit `default_stream_config.storage_class` uses its location's default.
* A new stream without an explicit `storage_class` uses its basin's default stream configuration.
* Changing a basin's default affects subsequently created streams. It does not change existing streams.

### Retention

S2 is designed to accommodate both infinite accumulation of data and time-bounded retention. You can use a stream as the long-term source of truth for your data, or as a momentary durable buffer on the way to somewhere else.

A retention policy can be configured on a stream, and S2 will automatically delete records that are older than the configured threshold.

This can be set when [creating a stream](/docs/api/streams/create) or by [reconfiguring](/docs/api/streams/reconfigure) an existing one.

> [!NOTE]
>
> Records may stay visible for a brief period after their retention window expires, typically on the order of minutes.

### Delete on empty

When enabled, a stream is automatically deleted once all its records have been trimmed and the stream is empty. The `min_age_secs` field requires the stream to have existed for at least that long before it can be auto-deleted.

Useful for ephemeral streams, e.g. one stream per job run where the data is consumed and no longer needed.

### Timestamping

Every record has a timestamp alongside its sequence number, and reads can start from a timestamp.

`timestamping.mode` controls which clock is used:

* `client-prefer` (**default**): use the client timestamp when present, otherwise use S2 arrival time.
* `client-require`: require the client to provide a timestamp.
* `arrival`: always use S2 arrival time, in milliseconds since the Unix epoch.

S2 keeps timestamps monotonic within each stream. If a timestamp would move backwards, S2 adjusts it to the highest timestamp already seen on that stream. Adjacent records may share the same timestamp.

By default, arrival time caps client timestamps so a bad clock cannot push the stream far into the future. Set `timestamping.uncapped` to `true` when your timestamp scale is not wall-clock time.

> [!TIP]
>
> See [Keeping time on a stream](https://s2.dev/blog/timestamping) for background.

## Basin config

| Field | Default | Description |
| - | - | - |
| `create_stream_on_append` | `false` | Auto-create streams on first append. |
| `create_stream_on_read` | `false` | Auto-create streams on first read. |
| `default_stream_config` | See [above](#stream-config) | Defaults used when not explicitly specified. |
| `stream_cipher` | Unset | Cipher for newly created streams: `aegis-256`, `aes-256-gcm`, or unset for plaintext streams. |


# Encryption
Source: https://s2.dev/docs/concepts/encryption

Encrypted streams keep record contents private at rest with a per-request key that S2 does not store.

S2 supports stream encryption with a key that you supply at request time. Configure a basin with a cipher for new streams, then provide the same key on `append` and `read` on any new streams.

> [card] Determined at stream creation
>
> A stream captures its `cipher` when it is created, through its basin's configuration.

> [card] Key supplied on data plane requests
>
> Encrypted streams require the `s2-encryption-key` header on REST requests, or pass it using the corresponding CLI environment variable or flags.

> [card] Key custody stays with you
>
> S2 does not persist or log the encryption key.

## How it works

```mermaid theme={null}
graph LR
  A["Set basin stream_cipher"] --> B["Create stream"]
  B --> C["Stream requires that cipher for its lifetime"]
  C --> D["Append with s2-encryption-key"]
  C --> E["Read with the same s2-encryption-key"]
```

New streams inherit the `stream_cipher` configured on the basin at the moment they are created. Reconfiguring the basin later only affects streams created after that change.

## What is encrypted

For regular records, S2 encrypts headers and body before storage, and decrypts them only when serving a read with the correct key.

[Command records](/docs/concepts/command-records) stay plaintext because S2 must interpret them to apply service-side operations such as fencing and trimming.

> [!NOTE]
>
> Encrypted records are cryptographically bound to the stream they were written to. S2 includes the stream's unique identifier as associated data, so ciphertext from one stream cannot be accepted as a record in another stream.

## Supported ciphers

| Cipher | API / CLI value | Key length |
| - | - | - |
| **AEGIS-256** (*recommended*) | `aegis-256` | 32 bytes |
| **AES-256-GCM** (*NIST-approved*) | `aes-256-gcm` | 32 bytes |

Keys are base64-encoded when provided in the `s2-encryption-key` header or to the CLI. SDKs allow providing either a string or bytes.

> [!NOTE]
>
> A stream configured with `aes-256-gcm` can have up to 2^32 records (\~4.3 billion), and subsequent appends will be rejected. Otherwise, a stream can have up to 2^64 records (effectively unlimited) — so `aegis-256` is generally preferable.

## Where keys are required

| Operation | Key required? | Notes |
| - | - | - |
| Create or reconfigure basin | No | Set `stream_cipher` here for future streams. |
| Create stream | No | The stream inherits the basin's current `stream_cipher`. |
| Append | Yes, if the stream has a `cipher` | Provide the key on every request or session. |
| Read | Yes, if the stream has a `cipher` | Use the same key that encrypted the records you want to read. |
| Check tail | No | Tail only returns positions, not payloads. |
| List streams | No | Each stream's `cipher` is returned in metadata. |

## Generate a key

```bash theme={null}
openssl rand -base64 32
```

> [!NOTE]
>
> Store the resulting value as a secret to be used for one or more streams. Maintain a mapping of which key was used for which stream, so that it may be specified consistently for appends and reads.

## Key management

For coarse-grained access, manage a shared key in your existing Key Management System (KMS) or secrets infrastructure.

For finer-grained access, such as a key per stream, use envelope encryption:

* Keep a Key Encryption Key (KEK) in your KMS.
* Generate a Data Encryption Key (DEK) per stream, and store the wrapped DEK alongside your stream metadata.
* At request time, your application unwraps the DEK and passes it to S2 as the stream encryption key.

If you need to rotate the key, create a new stream and cut writers and readers over deliberately. Similarly, to change the cipher used, update the basin configuration and then create a new stream.

> [!WARNING]
>
> Do not rotate a stream key in place by simply changing the request header. S2 will accept the append, but later reads with only one key will fail once they reach records encrypted under a different key.

## Configure encryption for new streams

> [tab] CLI
>
> ```bash theme={null}
> s2 create-basin secure-events \
>   --stream-cipher aegis-256
>
> # or change the default for streams created later
> s2 reconfigure-basin secure-events \
>   --stream-cipher aes-256-gcm
> ```

> [tab] REST
>
> ```bash theme={null}
> curl --request POST \
>   --url 'https://a.s2.dev/v1/basins' \
>   --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
>   --header 'Content-Type: application/json' \
>   --data '{
>     "basin": "secure-events",
>     "config": {
>       "stream_cipher": "aegis-256"
>     }
>   }'
> ```
>
> ```bash theme={null}
> curl --request PATCH \
>   --url 'https://a.s2.dev/v1/basins/secure-events' \
>   --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
>   --header 'Content-Type: application/json' \
>   --data '{
>     "stream_cipher": "aes-256-gcm"
>   }'
> ```

> [tab] TypeScript
>
> ```typescript theme={null}
> await client.basins
> 	.create({
> 		basin: basinName,
> 		config: { streamCipher: "aegis-256" },
> 	})
> 	.catch((e) => {
> 		if (!(e instanceof S2Error && e.status === 409)) throw e;
> 	});
>
> await client.basins.reconfigure({
> 	basin: basinName,
> 	streamCipher: "aes-256-gcm",
> });
> ```

> [tab] Python
>
> ```python theme={null}
> await client.create_basin(
>     basin_name,
>     config=BasinConfig(stream_cipher=Encryption.AEGIS_256),
> )
>
> await client.reconfigure_basin(
>     basin_name,
>     config=BasinConfig(stream_cipher=Encryption.AES_256_GCM),
> )
> ```

> [tab] Go
>
> ```go theme={null}
> client.Basins.Create(ctx, s2.CreateBasinArgs{
> 	Basin: basinName,
> 	Config: &s2.BasinConfig{
> 		StreamCipher: s2.Ptr(s2.EncryptionAlgorithmAegis256),
> 	},
> })
>
> client.Basins.Reconfigure(ctx, s2.ReconfigureBasinArgs{
> 	Basin: basinName,
> 	Config: s2.BasinReconfiguration{
> 		StreamCipher: s2.Ptr(s2.EncryptionAlgorithmAes256Gcm),
> 	},
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> client
>     .create_basin(
>         CreateBasinInput::new(basin_name.clone())
>             .with_config(BasinConfig::new().with_stream_cipher(EncryptionAlgorithm::Aegis256)),
>     )
>     .await?;
>
> client
>     .reconfigure_basin(ReconfigureBasinInput::new(
>         basin_name.clone(),
>         BasinReconfiguration::new().with_stream_cipher(EncryptionAlgorithm::Aes256Gcm),
>     ))
>     .await?;
> ```

> [!NOTE]
>
> Stream create endpoints do not take a per-stream encryption field. A stream inherits the basin default when it is created, and that `cipher` is immutable for the lifetime of the stream.

## Append and read with the key

> [tab] CLI
>
> ```bash theme={null}
> export S2_ENCRYPTION_KEY="$(openssl rand -base64 32)"
>
> printf 'top secret\n' | s2 append s2://secure-events/audit-log
> s2 read s2://secure-events/audit-log -s 0 -n 1
> ```
>
> ```bash theme={null}
> # Alternatively, read the key from a file
> s2 read s2://secure-events/audit-log \
>   --encryption-key-file ./stream-key.b64 \
>   -s 0 -n 10
> ```

> [tab] REST
>
> ```bash theme={null}
> curl --request POST \
>   --url 'https://secure-events.b.s2.dev/v1/streams/audit-log/records' \
>   --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
>   --header 'Content-Type: application/json' \
>   --header 's2-format: raw' \
>   --header "s2-encryption-key: ${S2_ENCRYPTION_KEY}" \
>   --data '{
>     "records": [
>       { "body": "top secret" }
>     ]
>   }'
> ```
>
> ```bash theme={null}
> curl --request GET \
>   --url 'https://secure-events.b.s2.dev/v1/streams/audit-log/records?seq_num=0&count=10' \
>   --header "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
>   --header "s2-encryption-key: ${S2_ENCRYPTION_KEY}"
> ```

> [tab] TypeScript
>
> ```typescript theme={null}
> const stream = basin.stream(streamName, { encryptionKey: encKey });
>
> await stream.append(
> 	AppendInput.create([AppendRecord.string({ body: "top secret" })]),
> );
>
> const batch = await stream.read({
> 	start: { from: { seqNum: 0 } },
> 	stop: { limits: { count: 10 } },
> });
> ```

> [tab] Python
>
> ```python theme={null}
> stream = basin.stream(
>     stream_name,
>     encryption_key=os.environ["S2_ENCRYPTION_KEY"],
> )
>
> await stream.append(
>     AppendInput(
>         records=[
>             Record(body=b"top secret"),
>         ]
>     )
> )
>
> batch = await stream.read(
>     start=SeqNum(0),
>     limit=ReadLimit(count=10),
> )
> ```

> [tab] Go
>
> ```go theme={null}
> key, err := s2.NewEncryptionKey(os.Getenv("S2_ENCRYPTION_KEY"))
> if err != nil {
> 	log.Fatalf("S2_ENCRYPTION_KEY: %v", err)
> }
>
> stream := basin.StreamWithOptions(s2.StreamName(streamName), &s2.StreamOptions{
> 	EncryptionKey: &key,
> })
>
> _, _ = stream.Append(ctx, &s2.AppendInput{
> 	Records: []s2.AppendRecord{
> 		{Body: []byte("top secret")},
> 	},
> })
>
> batch, _ := stream.Read(ctx, &s2.ReadOptions{
> 	SeqNum: s2.Uint64(0),
> 	Count:  s2.Uint64(10),
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> let stream = basin
>     .stream(stream_name.clone())
>     .with_encryption_key(std::env::var("S2_ENCRYPTION_KEY")?.parse()?);
>
> stream
>     .append(AppendInput::new(AppendRecordBatch::try_from_iter([
>         AppendRecord::new("top secret")?,
>     ])?))
>     .await?;
>
> let batch = stream
>     .read(
>         ReadInput::new()
>             .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
>             .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(10))),
>     )
>     .await?;
> ```

> [!NOTE]
>
> The same keying pattern applies to single-request operations, append/read sessions, SSE reads, and S2S sessions. The encryption key is just an HTTP header on the wire.

## Inspect cipher config

* [Get basin config](/docs/api/basins/get-config) to see the basin's current `stream_cipher` for new streams.
* [List streams](/docs/api/streams/list) to inspect current streams' `cipher`.

## Errors

| Situation | Result |
| - | - |
| Missing key on an encrypted stream | `422 invalid` |
| Key is not valid base64 or not 32 bytes | `422 invalid` |
| Wrong key | `400 decryption_failed` |


# Metrics
Source: https://s2.dev/docs/concepts/metrics

Metrics expose account, basin, and stream usage through scoped metric sets, query parameters, and response types.

S2 exposes metrics about account, basin, and stream resources.

The metrics API powers visualizations within the [dashboard](https://s2.dev/dashboard/metrics), but is also accessible from the CLI, SDKs, and REST API for custom uses. The account-level [`estimated-billable-usage`](/docs/api/metrics/account#estimated-billable-usage) metric set exposes estimated charges in USD by pricing category, and powers the [Billing tab](https://s2.dev/dashboard/billing).

> [!NOTE]
>
> Operational metrics *may* show duplicate counts or other slight discrepancies due to our use of materialized views for efficiency. Estimated billable usage uses finalized billing data instead, but is not a final invoice. See its [reference](/docs/api/metrics/account#estimated-billable-usage) for details and caveats.

## Metric sets

Metric sets are groupings of metrics evaluated either at the account, basin, or stream level. For example, the account-level [`active-basins`](/docs/api/metrics/account#active-basins) metric set can be obtained with:

```bash theme={null}
curl \
  --request GET \
  --url "https://a.s2.dev/v1/metrics?set=active-basins&start=${START}&end=${END}" \
  --header "Authorization: Bearer ${S2_ACCESS_TOKEN}"
```

#### Common parameters

* `set` specifies the requested metric set
* `start` and `end` timestamps, in [Unix epoch](https://en.wikipedia.org/wiki/Unix_time) seconds
  * The time period specified should not exceed 90 days
  * The `end` timestamp should not be in the future (though some leeway is given to accommodate clock differences)
* `interval` represents a duration for use by metric sets that return timeseries
  * for [`accumulation`](#accumulation), this specifies the requested bucket length, or the window of time which the relevant value is an aggregation over
  * for [`gauge`](#gauge), this specifies the unit to divide the `start`/`end` period into, and the timeseries will reflect a momentary reading at each of these divisions of time

## Metric types

### `scalar`

Represents a single numeric value. Can represent a summary, or aggregation, if the metric set requires a `start` and `end` period, or an instantaneous value if not.

### `accumulation`

Timeseries where each `(epoch, value)` tuple represents an aggregation `value` for a bucket of time beginning at the `epoch`, lasting for the specified `bucket_length`.

Buckets within the time period which have no data will be omitted (unlike with `GaugeTimeseries`).

### `gauge`

Timeseries where each `(epoch, value)` tuple represents a momentary reading of a value, at the specified `epoch` instant.

Gauge timeseries will contain entries for every interval within the requested time period.

### `label`

A set of string labels representing a set. It can represent a summary or aggregation if the metric set requires a `start` and `end` period, or an instantaneous set if not.

## Available metrics

> [card] [Account metrics](/docs/api/metrics/account)
>
> Track active basins, account operations, and estimated billable usage.

> [card] [Basin metrics](/docs/api/metrics/basin)
>
> Track active streams, storage, operations, and throughput.

> [card] [Stream metrics](/docs/api/metrics/stream)
>
> Track stored bytes for one stream.


# Reads
Source: https://s2.dev/docs/concepts/reads

Reads are ordered scans from a stream position, bounded or kept open to follow new records.

Readers can replay history, catch up from a cursor, poll for recent records, or follow a stream live.

S2's CDN-like internal [architecture](/docs/platform/architecture#read-path) allows for a large number of concurrent readers.

## Behavior

A read has three behavioral controls: where it starts, which bounds stop it, and whether it waits at the tail.

Reads can [start from](/docs/api/records/read#starting-point) one of the following:

* record **sequence number**
* record **timestamp**
* a **tail offset**, i.e. N records before the current tail, or the head of the stream if there are fewer records

Sequence numbers are the most precise cursor. Timestamps are useful for time-windowed processing. Tail offsets are useful for "show me the latest activity" views.

> [!NOTE]
>
> S2 returns records in stream order, starting with the earliest record at or after the requested position.

[Bounds](/docs/api/records/read#bounds) may be a record count, a byte budget, an end timestamp, or a combination of these. They apply to the full span of the read, including records that arrive while the read is waiting at the tail.

[Wait](/docs/api/records/read#waiting-at-the-tail) determines what happens after a read catches up to the tail. With default options:

* single-batch reads have implicit record count and byte caps, and return immediately
* read sessions have no bounds, and follow indefinitely

## Interfaces

Use the interface that matches the consumer:

| Interface | Use when |
| - | - |
| Single-batch read | You need one bounded response, optionally long polling at the tail. |
| Read session | You need to scan multiple batches, catch up from a cursor, or follow the stream continuously. |
| Check tail | You only need the current tail position, not the records themselves. |

SDK [read sessions](/docs/sdk/reading#read-session) are the preferred interface for continuous consumers because they provide an iterator-style API and reconnect after transient failures. [SSE](/docs/api/records/read#sse) provides a browser-friendly streaming response for the same session model.

## Consistency

Stream operations are **linearizable**. If an `append` has been acknowledged, any subsequent `read` or `check-tail` operation reflects that write.

## Latency

Time-to-first-byte latency for records written in the last 20 seconds to an *Express* stream is expected to be single-digit milliseconds. Otherwise, it can be up to 200 milliseconds.

Tail checks return only the current [tail position](/docs/concepts/streams#positions) and are expected to complete in single-digit milliseconds.

> [!NOTE]
>
> Latency expectations assume an [authenticated client](/docs/api/protocol#authentication) in the same location as the basin.

## See also

> [card] [Reading SDK](/docs/sdk/reading)
>
> Use single-batch reads, read sessions, and check-tail calls.

> [card] [Read API](/docs/api/records/read)
>
> Review request parameters, SSE sessions, wait behavior, and `416` cases.

> [card] [Stream positions](/docs/concepts/streams#positions)
>
> Head, tail, sequence numbers, and timestamps.


# Records
Source: https://s2.dev/docs/concepts/records

Records are the fundamental unit of an S2 stream: ordered entries with a sequence number, timestamp, and optional headers and body made of arbitrary bytes.

S2 streams allow you to [`append`](/docs/concepts/appends) records at the tail, and [`read`](/docs/concepts/reads) starting from any position in the stream.

seq\_num: 42\

timestamp: 1713812735000\


***

headers:\

  role: agent\

  response-type: tool-call\

  format: json\


***

body:\

  \{"event": "click", "x": 42}

## Anatomy of a record

**Sequence number**

Strictly increasing position within the stream.

Always assigned by S2.

**Timestamp**

Arrival timestamp in milliseconds, when assigned by S2 (default).

It can also be [client-specified](/docs/concepts/configs).

**Headers**

Name-value pairs, similar to HTTP headers.

Both the name and value can be arbitrary bytes.

**Body**

The core payload of a record.

Arbitrary bytes.

## Size

A single record can be up to 1 MiB in [metered bytes](/docs/platform/limits#records).

## See also

> [card] [Appends](/docs/concepts/appends)
>
> Understand append acknowledgements, batching, and durability.

> [card] [Reads](/docs/concepts/reads)
>
> Scan from sequence numbers, timestamps, and tail offsets.

> [card] [Data plane API](/docs/api/protocol#data-plane)
>
> Compare the append, read, and check-tail endpoints.


# Snapshots
Source: https://s2.dev/docs/concepts/snapshots

Use snapshots to bound replay cost while continuing to follow newer stream records.

## Snapshot and follow

Many applications store state as an ordered stream of deltas: each record says what changed. A reader can recover current state by folding those records from the head, then keep the read open to [follow new records](/docs/concepts/reads).

However, that replay cost grows with the stream. The snapshot-and-follow pattern bounds it by materializing state at a known stream position, or a *cursor*. The cursor is the first sequence number not covered by the snapshot — it can come from a writer's current state, or a separate snapshotting process can `checkTail` to ensure its knowledge is current.

## Where snapshots live

### External snapshots

For snapshots stored outside the stream, such as in object storage, attach the cursor to the snapshot itself. For example, include it in the object key or metadata.

To restore, load the snapshot, then read from the cursor and follow the stream. [Trimming](/docs/concepts/trimming) the stream to the cursor is optional; do it when you want to discard history already covered by the snapshot.

For example, a snapshot at cursor `1000` covers records before `1000`, so recovery resumes at `1000`:

```mermaid theme={null}
flowchart LR
  subgraph Stream["stream sequence"]
    direction LR
    R0["0"] --> RMore["..."] --> R999["999"] --> R1000["1000"] --> R1001["1001"] --> RNew["new records"]
  end

  R999 -. "fold into" .-> Snapshot["snapshot<br/>covers 0..999"]
  R1000 --- Cursor["cursor = 1000<br/>first record not covered"]
  Snapshot --> Restore["restore<br/>load snapshot"]
  Restore --> R1000
  R1000 --> Follow["read and follow"]

  classDef covered fill:#e0f2fe,stroke:#0284c7,color:#0f172a;
  classDef cursor fill:#fef3c7,stroke:#d97706,color:#0f172a;
  classDef follow fill:#dcfce7,stroke:#16a34a,color:#0f172a;
  class R0,RMore,R999,Snapshot covered;
  class R1000,Cursor,Restore cursor;
  class R1001,RNew,Follow follow;
```

1. A snapshotting process folds sequence numbers `0` through `999` into state.
2. It writes `snapshots/orders/1000.json`, where `1000` is the cursor.
3. It optionally trims the stream to `1000` to discard the covered records.
4. A reader loads that object, reads from sequence number `1000`, and follows from there.

### In-stream snapshots

Storing the snapshot in the stream lets readers recover from the stream alone. When paired with [trimming](/docs/concepts/trimming), the snapshot record or first fragment can become the new head, so readers simply start from the first record of the stream, apply the snapshot, then follow later deltas.

Because the snapshot is written to the stream, make the first append [conditional](/docs/concepts/concurrency-control#match-sequence-number) on the cursor. If the stream has moved, rebuild or extend the snapshot and retry.

#### Single-record snapshot

If the snapshot and trim command will fit in a [1 MiB batch](/docs/platform/limits#records), use a single atomic batch to append the [`trim` command](/docs/concepts/command-records) and snapshot record. The command that advances the head and the snapshot that remains there become durable together.

1. Choose cursor `1000`.
2. Build or serialize the snapshot for that position.
3. Append one batch with `match_seq_num = 1000`:
   * a `trim` command record to `1001`
   * a snapshot record, e.g. `{ "type": "snapshot", "covers_before": 1000, "state": ... }`
4. On success, the trim command lands at sequence number `1000` and the snapshot lands at `1001`. After trimming takes effect, the command record is removed and the snapshot is the stream head.
5. Readers start at the head, apply the snapshot, and follow later deltas.

#### Framed snapshot

For snapshots too large for one record, split the snapshot into ordered fragments. The fragment records should be distinguishable from normal deltas, usually with headers or body fields that identify a snapshot ID, chunk order, and final marker. Every fragment is written before trimming.

1. Choose cursor `1000`.
2. Build or serialize the snapshot for that position.
3. Append the first snapshot fragment with `match_seq_num = 1000`.
4. Append the remaining fragments and a final marker in order, for example by using an append session or by waiting for each append before sending the next one.
5. Append a `trim` command record to `1000`.

After trimming takes effect, the first snapshot fragment is the stream head. Unlike the single-record case, the trim command remains in the retained stream. Readers can skip it when applying application deltas.

While a reader reconstructs the snapshot from its fragments, it may encounter delta records appended concurrently with the snapshot fragments. The reader should buffer those deltas, finish applying the snapshot, then apply the buffered deltas in order and continue following the stream.

## See also

> [card] [Trimming](/docs/concepts/trimming)
>
> Discard records already covered by a snapshot.

> [card] [Reads](/docs/concepts/reads)
>
> Read from a cursor and follow new records.

> [card] [Concurrency control](/docs/concepts/concurrency-control)
>
> Use conditional appends with match sequence numbers.


# Streams
Source: https://s2.dev/docs/concepts/streams

Streams are durable, ordered sequences of records you can append to, seek through, and follow live.

Streams are always durable and totally ordered.

Every stream has a URI: `s2://{basin}/{stream}`. Basins can hold unlimited streams.

Appends always land at the tail, atomically — either all the records in a batch or none.

Reads can start from any position and optionally follow for live updates.

## Positions

We talk about the **head** and **tail** of a stream.

The head is the beginning, and the tail is where the next record to be appended will go.

Each record on a stream has an associated sequence number and timestamp.

* sequence number increases strictly monotonically
* timestamp increases monotonically, and may be identical between records

> [!TIP]
>
> The *tail* position is `(next_seq_num, last_record_timestamp)`.
>
> Both will be `0` for an empty stream!

## Lifecycle automation

You can automate the creation and deletion of streams. This is very useful for patterns with many dynamic streams, like one per session or job.

### Creation

Streams can be created explicitly, or automatically on first use.

If a basin has `create_stream_on_append` or `create_stream_on_read` [configured](/docs/concepts/configs#basin-config), streams will be created with the basin's default config the first time they're accessed. The [`s2-stream-config`](/docs/api/records/append#stream-config-on-auto-creation) header can override the default config for the new stream on that first request.

### Deletion

Streams can be [configured](/docs/concepts/configs#delete-on-empty) to automatically delete once they become empty. The `min_age_secs` threshold requires the stream to have been empty for a minimum duration before auto-deletion kicks in.

## See also

> [card] [Stream SDK](/docs/sdk/stream-resources)
>
> Create, configure, list, and delete streams from SDK clients.

> [card] [Stream API](/docs/api/streams/create)
>
> Review the stream management endpoints.


# Trimming
Source: https://s2.dev/docs/concepts/trimming

Trimming lets you explicitly discard older records without changing a stream's retention policy.

## Trim point

Trimming removes records before a chosen sequence number, or the **trim point**. Unlike [retention](/docs/concepts/configs#retention), which is configured as a policy on a stream, trimming is an explicit operation.

```bash theme={null}
s2 trim s2://my-basin/my-stream 1000
```

> Trimming involves appending a `trim` [command record](/docs/concepts/command-records#operations).

The effective trim point will be `max(existing_trim_point, min(provided_trim_point, trim_command_seq_num + 1))`, so a trim command can trim through itself but not past itself, and it cannot move the trim point backwards.

> [!NOTE]
>
> Trimming is eventually consistent — trimmed records may be visible for a brief period.

## Snapshots and compaction

[Snapshots](/docs/concepts/snapshots) bound replay cost by materializing state at a cursor. Trimming can then discard records already covered by the snapshot, but it is optional for external snapshots when you want to keep the full stream history.

## See also

> [card] [Trim with the CLI](/docs/cli/trim)
>
> Set a stream trim point from the command line.

> [card] [Command records](/docs/concepts/command-records)
>
> See how trim directives are encoded as records.

> [card] [Snapshots](/docs/concepts/snapshots)
>
> Bound replay cost with snapshot-and-follow.

> [card] [Retention](/docs/concepts/configs#retention)
>
> Configure age-based deletion for older records.


# Demos
Source: https://s2.dev/docs/demos

Try S2 in action: experiment in the Playground, resume an AI chat mid-stream, watch live sandbox logs, and test collaborative editing.

## Interactive

> [card] [Playground](https://s2.dev/playground)
>
> Read and write to a new S2 stream directly from your browser.
>
> Share your link with others for a durable, real-time chat room.

> [card] [AI Chat](https://s2.dev/demos/ai-chat)
>
> Multiplayer, resumable AI chat powered by S2.
>
> Try opening it across multiple devices, or reloading mid-response — the stream picks up where it left off.

> [card] [Sandbox Infra](https://s2.dev/demos/browser-images)
>
> Real-time observability for serverless environments.
>
> Click into tasks to watch their progress live.

> [card] [Real-time Sync with Yjs](https://s2.dev/demos/y-s2)
>
> Collaborative editing with CRDTs over S2.
>
> See the [blog post](https://s2.dev/blog/durable-yjs-rooms) for how it works.

## Technical deep dives

> [card] [s2-term](https://s2.dev/blog/s2-term)
>
> A serverless terminal that replaces SSH with S2 streams, letting you share a term in the browser.

> [card] [Building a KV Store](https://s2.dev/blog/kv-store)
>
> Building a strongly-consistent distributed key-value database using S2 as the write-ahead log.
>
> Also presented at [HYTRADBOI 2025](https://www.hytradboi.com/2025/f2cc03cb-14fc-42f4-ad38-b4b15a15815f-serverless-primitives-for-the-shared-log-architecture).

> [card] [IoT](https://s2.dev/blog/iot)
>
> A Raspberry Pi streams thermal sensor data to S2 at 10 Hz, consumed live via SSE in a Next.js frontend.

> [card] [Parallax](https://s2.dev/blog/distributed-ai-agents)
>
> Distributed AI agent coordination.
>
> Multiple independent agent cohorts reason on shared problems, each writing to their own S2 stream.


# Declarative Spec
Source: https://s2.dev/docs/infra/declarative-spec

Define S2 basins and streams in a JSON spec and apply it with the CLI for declarative, version-controlled infrastructure.

Define your S2 infrastructure in a JSON spec file and apply it declaratively using the CLI. Resources that already exist are reconfigured to match the spec, and only the fields present in the spec are updated.

```bash theme={null}
s2 apply -f spec.json
```

## Spec file format

The spec is a JSON file with a top-level `basins` array. Each basin entry has a `name`, an optional `config`, and an optional `streams` array.

```json theme={null}
{
  "$schema": "https://raw.githubusercontent.com/s2-streamstore/s2/main/cli/schema.json",
  "basins": [
    {
      "name": "my-basin",
      "config": {
        "stream_cipher": "aegis-256",
        "create_stream_on_append": true,
        "default_stream_config": {
          "storage_class": "express",
          "retention_policy": "7days"
        }
      },
      "streams": [
        {
          "name": "events",
          "config": {
            "storage_class": "standard",
            "retention_policy": "infinite"
          }
        }
      ]
    }
  ]
}
```

### IDE setup

Add the `$schema` key to your spec file to point to the published schema hosted on GitHub to enable inline validation and autocomplete in your editor:

```json theme={null}
{
  "$schema": "https://raw.githubusercontent.com/s2-streamstore/s2/main/cli/schema.json",
  "basins": []
}
```

Or generate a local copy and reference it by path:

```bash theme={null}
s2 apply --schema > schema.json
```

```json theme={null}
{
  "$schema": "./schema.json",
  "basins": []
}
```

**VS Code**: configure schema associations in `.vscode/settings.json` so `$schema` is not needed in every file:

```json theme={null}
{
  "json.schemas": [
    {
      "fileMatch": ["*.s2spec.json", "s2-spec.json"],
      "url": "https://raw.githubusercontent.com/s2-streamstore/s2/main/cli/schema.json"
    }
  ]
}
```

You can use any glob pattern in `fileMatch` to match your spec files by naming convention.

**IntelliJ / JetBrains IDEs**: add a schema mapping in the IDE settings:

1. Open **Settings** (`Ctrl+Alt+S` / `⌘,`) → **Languages & Frameworks** → **Schemas and DTDs** → **JSON Schema Mappings**
2. Click **+** to add a new mapping
3. Give the mapping a name (e.g. `S2 Spec`)
4. Set the schema to `https://raw.githubusercontent.com/s2-streamstore/s2/main/cli/schema.json` (or a local path)
5. Set the **Schema version** to **JSON Schema version 7**
6. Add file patterns to associate (e.g. `*.s2spec.json`)

### Basin config fields

| Field | Type | Description |
| - | - | - |
| `default_stream_config` | object | Default stream configuration applied to newly created streams |
| `stream_cipher` | `"aegis-256"` \| `"aes-256-gcm"` | Cipher to apply to streams created after this basin config is in place |
| `create_stream_on_append` | boolean | Auto-create a stream on first append using the default config |
| `create_stream_on_read` | boolean | Auto-create a stream on first read using the default config |

### Stream config fields

| Field | Type | Description |
| - | - | - |
| `storage_class` | string | Storage-class name available in the basin's location. Check available classes with [List locations](/docs/api/locations/list). |
| `retention_policy` | `"infinite"` \| humantime string | How long records are retained (e.g. `"7days"`, `"1week"`) |
| `timestamping` | object | Timestamping behavior (see below) |
| `delete_on_empty` | object | Delete the stream automatically when empty |

#### Timestamping

```json theme={null}
{
  "mode": "client-prefer",
  "uncapped": false
}
```

| Field | Values | Description |
| - | - | - |
| `mode` | `"client-prefer"` \| `"client-require"` \| `"arrival"` | How record timestamps are resolved |
| `uncapped` | boolean | Allow client timestamps to exceed the server arrival time |

#### Delete-on-empty

```json theme={null}
{
  "min_age": "1day"
}
```

Set `min_age` to a humantime duration (e.g. `"1day"`, `"2h 30m"`) to enable automatic deletion of empty streams once they reach that age. The default of `"0s"` disables delete-on-empty.

## Dry run

Preview what would change without making any mutations:

```bash theme={null}
s2 apply -f spec.json --dry-run
```

Output uses the following legend:

| Symbol | Meaning |
| - | - |
| `+` | Will be created |
| `~` | Will be reconfigured (shows field-level diffs) |
| `=` | Already matches spec, no change |

Example output:

```
+ basin new-basin
    storage_class: express
  + stream new-basin/events
    retention_policy: 7days
= basin existing-basin
  ~ stream existing-basin/logs
      retention_policy: 7days → 30days
```

For the full list of CLI flags, see [`s2 apply`](/docs/cli/apply).


# Anthropic
Source: https://s2.dev/docs/integrations/anthropic

Persist Anthropic Messages API streams on S2 and replay them after refreshes or reconnects.

S2 integrates with the Anthropic Messages API through the
`@s2-dev/resumable-stream/anthropic` entrypoint.

The integration includes two helpers:

* The `createResumableChat` server helper stores raw Anthropic stream events in S2 streams and replays them as
  Anthropic-shaped SSE.
* The `subscribe` browser helper is a small async generator that reads the replay endpoint.

```bash theme={null}
export S2_ACCESS_TOKEN="..."
export S2_BASIN="my-basin"
export ANTHROPIC_API_KEY="..."
```

## Server Setup

```ts lib/s2.ts theme={null}
import { createResumableChat } from '@s2-dev/resumable-stream/anthropic';

export const chat = createResumableChat({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  basin: process.env.S2_BASIN!,
  mode: 'session',
});
```

> [!TIP]
>
> Using [S2 Lite](/docs/s2-lite)? Pass `endpoints: S2Environment.parse().endpoints` (from `@s2-dev/streamstore`) to read `S2_ACCOUNT_ENDPOINT` and `S2_BASIN_ENDPOINT`; they aren't picked up automatically. See [reading endpoints from the environment](/docs/sdk/endpoints#environment-variables).

`mode: 'session'` can be useful when you want your chat app to have one S2
stream to hold a long-lived log for a chat. Use `single-use` if each response
needs its own stream.

## Start A Turn

```ts app/api/chat/route.ts theme={null}
import Anthropic from '@anthropic-ai/sdk';
import type { MessageParam } from '@anthropic-ai/sdk/resources/messages';
import { chat } from '@/lib/s2';

const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY!,
});

export async function POST(req: Request) {
  const { id, messages } = (await req.json()) as {
    id: string;
    messages: MessageParam[];
  };

  const source = anthropic.messages.stream({
    model: process.env.ANTHROPIC_MODEL ?? 'claude-haiku-4-5-20251001',
    max_tokens: 1024,
    messages,
  });

  return chat.makeResumable(`chat-${id}`, source, {
    delivery: 'replay',
    waitUntil: (p) => p.catch(console.error),
  });
}
```

`delivery: 'replay'` makes the POST return `202`. The client reads the actual
events from the replay route.

## Replay Route

```ts app/api/chat/stream/route.ts theme={null}
import { chat } from '@/lib/s2';

function parseFromSeqNum(value: string | null): number | undefined {
  if (value === null) return undefined;
  const parsed = Number.parseInt(value, 10);
  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined;
}

export async function GET(req: Request) {
  const url = new URL(req.url);
  const id = url.searchParams.get('id');
  if (!id) return new Response('Missing id query parameter', { status: 400 });

  return chat.replay(`chat-${id}`, {
    fromSeqNum: parseFromSeqNum(url.searchParams.get('from')),
    live: url.searchParams.get('live') === '1',
  });
}
```

Use `live: true` in session mode when the browser should stay connected at the
tail and receive future turns.

## Browser Subscription

```ts theme={null}
import { subscribe } from '@s2-dev/resumable-stream/anthropic/client';

await fetch('/api/chat', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ id: chatId, messages }),
});

for await (const event of subscribe({
  url: (cursor) => {
    const params = new URLSearchParams({ id: chatId, live: '1' });
    if (cursor !== undefined) params.set('from', String(cursor));
    return `/api/chat/stream?${params}`;
  },
})) {
  switch (event.type) {
    case 'message_start':
      break;
    case 'content_block_delta':
      break;
    case 'message_stop':
      break;
  }
}
```

`subscribe` yields Anthropic `RawMessageStreamEvent` values, tracks the replay
cursor from SSE `id:` values, and reconnects from that cursor if the response
drops. It can also yield the adapter's error envelope:

```ts theme={null}
{ type: 'error', error: { type: string, message: string } }
```

## Completed History

The Anthropic helper does not infer user messages and does not build a
transcript for you. You can pair the replay stream with a separate history
store:

1. Append the user message to history before starting the model.
2. Fold Anthropic stream events into a completed assistant message.
3. Append the assistant message when `message_stop` arrives.
4. On page load, render history immediately, then subscribe to replay for any
   active turn.

## Options

`createResumableChat` accepts:

| option | default | description |
| - | - | - |
| `mode` | `"single-use"` | `"single-use"` uses one stream per generation. `"shared"` reuses one active-generation stream. `"session"` appends generations to one durable stream. |
| `endpoints` | S2 defaults | Optional endpoint overrides, commonly used with S2 Lite. |
| `batchSize` | `10` | Maximum number of events per append batch. |
| `lingerDuration` | `50` | Maximum batching delay in milliseconds. |
| `leaseDurationMs` | `5000` | `shared` mode takeover window for stale active generations. |
| `onError` | generic message | Maps upstream errors to an Anthropic-shaped `error` event. |

`replay` accepts:

| option | description |
| - | - |
| `fromSeqNum` | S2 cursor to resume from. The client tracks this from SSE `id:` values. |
| `live` | Session mode only. Keeps the SSE open at the tail for future records. |

`subscribe` accepts:

| option | description |
| - | - |
| `url` | Replay URL. String URLs get `?from=<cursor>` appended on reconnect; function URLs receive the cursor. |
| `signal` | Optional abort signal. |
| `fetch` | Optional fetch implementation override. |
| `headers` | Static or lazy headers sent on every request. |
| `credentials` | Fetch credentials mode. Defaults to `same-origin`. |
| `reconnectBackoffMs` | Millisecond backoff schedule. Pass `[]` to disable reconnect. |

## Example

A complete Bun server and browser client is available here:
[`examples/anthropic-resumable-chat`](https://github.com/s2-streamstore/s2-sdk-typescript/tree/main/examples/anthropic-resumable-chat).


# Bento
Source: https://s2.dev/docs/integrations/bento

Connect Bento pipelines to S2 with S2 input and output components for lightweight stream processing.

## Reference

Documentation for the S2 components for Bento is available at:

* [S2 Input](https://warpstreamlabs.github.io/bento/docs/components/inputs/s2)
* [S2 Output](https://warpstreamlabs.github.io/bento/docs/components/outputs/s2)

## Getting started

We are going to take inputs from a number of S2 streams, process the records using Bento, and output the records merged into another S2 stream.

### Prerequisites

**1. Generate an S2 access token**

Generate an access token by logging into the [dashboard](https://s2.dev/dashboard) and set the `S2_ACCESS_TOKEN` environment variable:

```bash theme={null}
export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN"
```

**2. Install the S2 CLI**

Install the [S2 CLI](/docs/quickstart#get-started-with-the-cli). The CLI automatically reads the `S2_ACCESS_TOKEN` environment variable configured above.

**3. Install the Bento CLI**

Install the latest version of the [Bento CLI](https://warpstreamlabs.github.io/bento/docs/guides/getting_started#install):

```bash theme={null}
curl -Lsf https://warpstreamlabs.github.io/bento/sh/install | bash
```

Ensure that the CLI supports the S2 plugin:

```bash theme={null}
bento list | grep 's2$'
```

### Setup

**1. Create a new basin**

> [!NOTE]
>
> Basin names are globally unique. They must be between 8 and 48 characters long and comprise lowercase letters, numbers and hyphens. They cannot begin or end with a hyphen.

```bash theme={null}
export MY_BASIN_NAME="YOUR_BASIN_NAME"
s2 create-basin ${MY_BASIN_NAME}
```

**2. Create source streams**

Create source streams with the prefix `pup/` and append some "woofs":

```bash theme={null}
PUP_NAMES=("buddy" "yoyo" "scooby")

for PUP in "${PUP_NAMES[@]}"; do
    s2 create-stream "s2://${MY_BASIN_NAME}/pup/${PUP}"

    # Generate and append 10000 random woofs.
    for _ in {1..10000}; do
        echo "${RANDOM}"
    done | s2 append "s2://${MY_BASIN_NAME}/pup/${PUP}"
done
```

We can verify that the streams have been appended with random numbers using:

```bash theme={null}
s2 read s2://${MY_BASIN_NAME}/pup/yoyo -n 10
```

The above command should output a list of 10 random numbers.

![](https://mintcdn.com/streamstore/I63bbJM2U3m9xZR6/images/random-woofs.png?fit=max\&auto=format\&n=I63bbJM2U3m9xZR6\&q=85\&s=d0800e9e52f1833374946c46e8dcb3df)

**Step 3**

Create a new stream called `woofs` to store the processed records:

```bash theme={null}
s2 create-stream "s2://${MY_BASIN_NAME}/woofs"
```

### The Pipeline

**1. Configuration**

Create a file called `woof-pipeline.yml`:

```yaml theme={null}
cache_resources:
  - label: seq_num_cache
    noop: {}

input:
  label: woof_input
  s2:
    basin: ${MY_BASIN_NAME}
    # All streams with the prefix `pup/`
    streams: pup/
    auth_token: ${S2_ACCESS_TOKEN}
    cache: seq_num_cache

pipeline:
  processors:
    - mapping: |
        root = "%v woofs from %s".format(this, meta("s2_stream").re_replace("^pup/", ""))

output:
  label: woof_output
  s2:
    basin: ${MY_BASIN_NAME}
    stream: woofs
    auth_token: ${S2_ACCESS_TOKEN}
```

**2. Running the pipeline**

Start the pipeline using:

```bash theme={null}
bento -c woof-pipeline.yml
```

Open up another terminal and see the records being appended to the `woofs` stream:

```bash theme={null}
s2 read s2://${MY_BASIN_NAME}/woofs
```

![](https://mintcdn.com/streamstore/I63bbJM2U3m9xZR6/images/woofs.gif?s=8dd64d11024e9c83c990b04e0dd661e7)


# Feldera
Source: https://s2.dev/docs/integrations/feldera

Read S2 streams into Feldera SQL pipelines and append view updates back to S2 with Feldera's native S2 input and output connectors.

[Feldera](https://www.feldera.com/) is an incremental compute engine: you define tables and views in SQL, and Feldera keeps the views up to date as records arrive instead of recomputing them. S2 ships as a native transport in Feldera, so a table can read directly from an S2 stream and a view can append its updates to one.

* **`s2_input`** feeds a table from an S2 stream, passing each record body to the configured Feldera input format (e.g. JSON). It checkpoints S2 sequence numbers, so a resumed pipeline picks up where it left off, and it reconnects with bounded backoff on transient errors. `start_from` selects where a fresh pipeline begins: `"Beginning"`, `"Tail"`, a `SeqNum`, a `Timestamp`, or a `TailOffset`.
* **`s2_output`** appends a view's changes to an S2 stream as one record per encoded row, batched into native S2 appends. It is not a fault-tolerant output: a failure between S2 accepting a batch and Feldera observing the acknowledgement can produce duplicates on retry.

Both connectors take a `basin`, `stream`, and `auth_token`, plus an optional `endpoint`. Full configuration tables are in the Feldera docs:

* [S2 input connector](https://docs.feldera.com/connectors/sources/s2)
* [S2 output connector](https://docs.feldera.com/connectors/sinks/s2)

> [!NOTE]
>
> Feldera marks S2 support as experimental.

## Example

A table reads JSON records from the `orders` stream, and a view appends every order of \$1,000 or more to the `large-orders` stream as Feldera [insert/delete](https://docs.feldera.com/formats/json#the-insertdelete-format) envelopes:

Replace `commerce` with your own basin name, and create both streams first (`s2 create-stream s2://{basin}/orders` and `s2 create-stream s2://{basin}/large-orders`) unless the basin has [stream auto-creation](/docs/concepts/configs#basin-config) enabled.

```sql theme={null}
CREATE TABLE orders (
    id BIGINT,
    amount DECIMAL(12, 2)
) WITH (
    'connectors' = '[{
        "transport": {
            "name": "s2_input",
            "config": {
                "basin": "commerce",
                "stream": "orders",
                "auth_token": "${secret:kubernetes:s2/auth-token}",
                "start_from": "Beginning"
            }
        },
        "format": {
            "name": "json",
            "config": { "update_format": "raw" }
        }
    }]'
);

CREATE VIEW large_orders WITH (
    'connectors' = '[{
        "transport": {
            "name": "s2_output",
            "config": {
                "basin": "commerce",
                "stream": "large-orders",
                "auth_token": "${secret:kubernetes:s2/auth-token}"
            }
        },
        "format": {
            "name": "json",
            "config": { "update_format": "insert_delete" }
        }
    }]'
) AS
SELECT * FROM orders WHERE amount >= 1000;
```

`auth_token` is an S2 [access token](/docs/concepts/access-tokens); the example passes it as a Feldera [secret reference](https://docs.feldera.com/connectors/secret-references) rather than inline.

With the pipeline running, appending an order to `orders`:

```bash theme={null}
echo '{"id":5,"amount":1500}' | s2 append s2://commerce/orders
```

shows up moments later on `large-orders` as an insert envelope:

```bash theme={null}
s2 read s2://commerce/large-orders --format json
```

```json theme={null}
{"seq_num":2,"timestamp":1792778129264,"body":"{\"insert\":{\"id\":5,\"amount\":1500}}"}
```

> [!TIP]
>
> Set `endpoint` to run against [s2-lite](/docs/s2-lite) locally, for example `"endpoint": "http://localhost:7070"`. s2-lite ignores the access token, so any non-empty `auth_token` works.


# Goldsky
Source: https://s2.dev/docs/integrations/goldsky

Stream real-time and historical blockchain data from Goldsky Turbo Pipelines into S2 with the Goldsky S2 sink.

[Goldsky](https://goldsky.com/) streams real-time and historical blockchain data via [Turbo Pipelines](https://goldsky.com/products/turbo-pipelines).

Goldsky supports S2 as a sink for Turbo Pipelines, allowing you to stream decoded blockchain data directly into S2 streams. Data is automatically batched according to S2's batch size limits.

See the [Goldsky S2 sink documentation](https://docs.goldsky.com/turbo-pipelines/sinks/s2) for setup and configuration details.


# LiveStore
Source: https://s2.dev/docs/integrations/livestore

Use S2 as a durable backend for LiveStore's event-sourced sync engine.

[LiveStore](https://livestore.dev) is a sync engine powered by event sourcing. The [`@livestore/sync-s2`](https://docs.livestore.dev/sync-providers/s2/) provider uses S2 as the durable backend for the event log — every client materializes state by replaying ordered events from an S2 stream.

## How it works

Each LiveStore `storeId` maps to a single S2 stream. Events are JSON-serialized and appended as records, preserving their total order. Clients sync through a small API proxy that you run on your own server:

* **Client** — the LiveStore client pushes and pulls events through your proxy endpoint.
* **API proxy** — your server authenticates to S2, creates basins and streams as needed, and translates LiveStore sync operations into S2 API calls. The `@livestore/sync-s2` package ships helpers for this, and is also where app-specific concerns like auth and rate limiting live.
* **S2** — stores the event log durably and streams live updates back to clients over SSE.

## Prerequisites

* An [S2 cloud account](https://s2.dev/sign-up)
* A scoped [access token](/docs/concepts/access-tokens) for your basin

## Install

```bash theme={null}
pnpm add @livestore/sync-s2
```

Point the client's sync backend at your proxy endpoint:

```typescript theme={null}
import { makeSyncBackend } from '@livestore/sync-s2'

const backend = makeSyncBackend({
  endpoint: '/api/s2',
})
```

The proxy reads your S2 credentials from the environment:

```bash theme={null}
S2_BASIN=your_basin_name
S2_ACCESS_TOKEN=your_access_token
```

## Resources

* [LiveStore S2 sync provider](https://docs.livestore.dev/sync-providers/s2/) — full setup guide, including the API proxy implementation
* [LiveStore documentation](https://docs.livestore.dev)


# Mastra
Source: https://s2.dev/docs/integrations/mastra

Persist and replay Mastra durable-agent streams with S2.

S2 integrates with [Mastra](https://mastra.ai) through
[`@s2-dev/mastra-pubsub`](https://www.npmjs.com/package/@s2-dev/mastra-pubsub).
The package provides an `S2PubSub` implementation for
[Mastra durable agents](https://mastra.ai/blog/introducing-durable-agents).

Each durable topic maps to an S2 stream. When `S2PubSub` appends an
event, the S2 sequence number becomes the event's Mastra `index`. Live delivery
and replay therefore use the same index, which lets Mastra replay and
deduplicate events without a separate counter.

| Mastra operation | S2 operation |
| - | - |
| Publish an event | Append a record |
| Assign an event index | Use the appended record's sequence number |
| Replay from an index | Read from a sequence number |
| Store a run's events | Use one stream for the run topic |

![A durable agent reconnecting after a browser refresh](https://mintcdn.com/streamstore/B7B_2TOsxpXpJPeO/images/mastra-durable-agent.gif?s=ffb3eb93fbdd33c700c8c3b0baebe61e)

## Install

```bash theme={null}
npm install @s2-dev/mastra-pubsub @mastra/core
```

The package requires Node.js 22.13 or later and `@mastra/core` 1.46 or later
in the 1.x release line. `@mastra/core` is a peer dependency.

Create a basin with **Create Stream on Append** and **Create Stream on Read**
enabled. These settings let `S2PubSub` create a run stream when it first writes
or reads the topic.

You can create the basin in the [dashboard](https://s2.dev/dashboard) or with
the [CLI](/docs/cli/basins#create-a-basin):

```bash theme={null}
s2 create-basin my-basin --create-stream-on-append --create-stream-on-read
```

Set the basin, access token, and model provider key:

```bash theme={null}
export S2_ACCESS_TOKEN="..."
export S2_BASIN="my-basin"
export OPENAI_API_KEY="..."
```

## Configure Mastra

Create an `S2PubSub` and pass it to the Mastra instance that registers your
durable agent:

```ts src/mastra/index.ts theme={null}
import { Mastra } from '@mastra/core';
import { S2PubSub } from '@s2-dev/mastra-pubsub';
import { durableResearchAgent } from './agents/research-agent';

export const mastra = new Mastra({
  agents: { durableResearchAgent },
  pubsub: new S2PubSub({
    accessToken: process.env.S2_ACCESS_TOKEN!,
    basin: process.env.S2_BASIN!,
  }),
});
```

```ts src/mastra/agents/research-agent.ts theme={null}
import { Agent } from '@mastra/core/agent';
import { createDurableAgent } from '@mastra/core/agent/durable';

const agent = new Agent({
  id: 'research',
  name: 'research',
  instructions: 'Answer research questions concisely.',
  model: 'openai/gpt-4o',
});

export const durableResearchAgent = createDurableAgent({ agent });
```

Registered durable agents use the Mastra instance's pubsub. By default,
`S2PubSub` uses S2 for topics beginning with `agent.`. This covers per-run
topics such as `agent.stream.<runId>` and the per-thread topics Mastra uses for
coordination. Topics outside that prefix, and events published with
`localOnly`, use the inner local transport.

## Reconnect to a run

Starting a durable-agent stream returns a `runId`. Keep that ID where a
reconnecting client can retrieve it.

```ts theme={null}
const started = await durableResearchAgent.stream(
  'Summarize the latest on durable streams.',
);

console.log(started.runId);

for await (const chunk of started.output.fullStream) {
  // Send chunks to the client as they arrive.
}
```

After a browser refresh or dropped connection, call `observe` with the same
`runId`:

```ts theme={null}
const observed = await durableResearchAgent.observe(runId);

for await (const chunk of observed.output.fullStream) {
  // Replayed events arrive before new live events.
}
```

Pass an `offset` when the caller already has part of the stream. The offset is
the first event index to replay, so use `42` if index `41` was the last event
processed:

```ts theme={null}
const observed = await durableResearchAgent.observe(runId, { offset: 42 });
```

## How persistence works

`S2PubSub` extends Mastra's `CachingPubSub`, but S2 is the authoritative state
for durable topics:

1. `publish` serializes the event and appends it once to the topic's S2 stream.
   Read sessions deliver the stored record to subscribers in the publishing
   process and in other processes, restoring its `seqNum` as the event `index`.
2. `subscribe` opens a read session at the live tail. If that session
   reconnects, it resumes from the next exact sequence number.
3. `subscribeFromOffset` opens a read session at the requested offset. The same
   session replays retained records and then follows new records, avoiding a
   replay-to-live handoff gap.
4. `getHistory` reads retained events from an offset and restores each record's
   `seqNum` as its event `index`.
5. `clearTopic` stops this instance's subscriptions, clears local state, and
   requests deletion of the corresponding S2 stream. Deletion failures are
   logged rather than thrown.

The adapter does not keep event history, a replay cache, or a durable cursor in
process memory. Call `await pubsub.close()` during graceful shutdown to cancel
active read sessions.

## Delivery semantics

Publishing rejects if S2 does not acknowledge the append; durable topics do not
fall back to a process-local event. The S2 SDK's default append retry policy is
at-least-once, so an ambiguous timeout can produce a duplicate record. If that
is not acceptable, create the S2 client with `appendRetryPolicy:
'noSideEffects'` and pass the client to `S2PubSub`:

```ts theme={null}
import { S2 } from '@s2-dev/streamstore';
import { S2PubSub } from '@s2-dev/mastra-pubsub';

const client = new S2({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  retry: { appendRetryPolicy: 'noSideEffects' },
});

const pubsub = new S2PubSub({
  client,
  basin: process.env.S2_BASIN!,
});
```

S2-delivered callbacks run one at a time per subscription. `S2PubSub` awaits
each callback before reading the next record, which preserves stream order and
applies backpressure when a callback is slow. A callback that never settles
stalls that subscription. If a callback rejects, the adapter logs the error and
considers the record delivered; broadcast observers do not provide
acknowledgement or callback-level redelivery.

Events use JavaScript's `JSON.stringify` semantics. Before appending, the
adapter validates the serialized record, so unsupported values such as
`data: undefined`, functions, symbols, bigints, or circular data reject
`publish` instead of leaving an unreadable record in the stream. Optional
nested properties whose value is `undefined` are omitted.

## Distributed leases

`getLeaseProvider()` returns the S2-backed lease provider used by
[Mastra's signals runtime](https://mastra.ai/docs/long-running-agents/signals#distributed-and-serverless-deployments)
to elect one owner per thread key across processes. The lease owner, expiry,
and nonce are encoded in the thread stream's S2 fencing token, so lease
operations remain atomic without an in-process cursor or a separate lease
stream.

S2 fence commands are filtered out of event history and live delivery, but
they still consume sequence numbers. Resume from `last.index + 1`, not from the
number of events received.

Do not trim a thread stream while its lease is held. `clearTopic` also deletes
the stream, so do not call it for an active thread topic.

## Cleanup and retention

When Mastra cleans up a run, it calls `clearTopic`, which deletes that run's S2
stream. Mastra's default cleanup window is 30 seconds; set `cleanupTimeoutMs`
to `0` if the stream should remain until explicit cleanup or S2 retention
removes it.

If a process exits before cleanup, its stream can remain. You can apply an
age-based [retention policy](/docs/concepts/configs#stream-config) and
`delete_on_empty` in the basin's default stream configuration to remove these
orphaned streams. Keep the retention period longer than the maximum run time
and reconnection window you support. A request whose offset has already been
trimmed fails instead of silently returning partial history.

## Options

`new S2PubSub(config, options?)`

| config | description |
| - | - |
| `accessToken` | S2 access token. Provide this or `client`. |
| `client` | Existing `S2` client. Takes precedence over `accessToken`. |
| `basin` | Basin that stores the durable-agent streams. |
| `endpoints` | Optional endpoint overrides, such as S2 Lite endpoints. |

| option | default | description |
| - | - | - |
| `inner` | `EventEmitterPubSub` | Local transport for non-S2 topics and explicit `localOnly` events. |
| `streamPrefix` | `mastra/durable/` | Prefix added to S2 stream names. |
| `topicPrefix` | `agent.` | Only topics with this prefix use S2, including per-run and per-thread topics. |
| `logger` | `console.error` | Logger used for swallowed or background pubsub failures. |

## S2 Lite

Install the S2 SDK for its environment helper:

```bash theme={null}
npm install @s2-dev/streamstore
```

To use [S2 Lite](/docs/s2-lite), set `S2_ACCOUNT_ENDPOINT` and `S2_BASIN_ENDPOINT`,
then pass the parsed endpoints to `S2PubSub`:

```ts theme={null}
import { S2PubSub } from '@s2-dev/mastra-pubsub';
import { S2Environment } from '@s2-dev/streamstore';

const { endpoints } = S2Environment.parse();

const pubsub = new S2PubSub({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  basin: process.env.S2_BASIN!,
  endpoints,
});
```

## Example

The package repository includes a runnable example adapted from Mastra's
durable-agents example:
[`examples/durable-agents`](https://github.com/s2-streamstore/mastra-pubsub/tree/main/examples/durable-agents).


# OpenTelemetry
Source: https://s2.dev/docs/integrations/otlp

Export OpenTelemetry traces, metrics, and logs to S2 over OTLP/HTTP using direct exporters or the OpenTelemetry Collector.

S2 implements the [OTLP/HTTP](https://opentelemetry.io/docs/specs/otlp/#otlphttp) specification, so it
can be used as a telemetry backend for trace, metric, and log signals.

> [!NOTE]
>
> Available on the cloud service only. [s2-lite](/docs/s2-lite) does not serve OTLP/HTTP.

Signals are routed to the appropriate streams by setting the `s2.stream` [resource](https://opentelemetry.io/docs/concepts/resources/)
attribute. You'd typically set it when initializing your [TracerProvider](https://opentelemetry.io/docs/concepts/signals/traces/#tracer-provider),
[MeterProvider](https://opentelemetry.io/docs/concepts/signals/metrics/#meter-provider),
or [LoggerProvider](https://opentelemetry.io/docs/concepts/signals/logs/#logger-provider).
For example, in Python, if you want all logs from `process-a` to go to `stream-a`, you need to configure it as follows:

```python theme={null}
import logging
from opentelemetry._logs import set_logger_provider
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk.resources import Resource

provider = LoggerProvider(resource=Resource.create({"s2.stream": "stream-a"}))
set_logger_provider(provider)
handler = LoggingHandler(level=logging.INFO, logger_provider=provider)
logging.basicConfig(handlers=[handler], level=logging.INFO)
```

You can export your signals directly to a telemetry backend from your processes, but it is generally
recommended to use a [Collector](https://opentelemetry.io/docs/collector/#when-to-use-a-collector).
If you are already using a Collector, you can easily start exporting signals to S2 by using
an [OTLP/HTTP exporter](https://github.com/open-telemetry/opentelemetry-collector/tree/main/exporter/otlphttpexporter).

S2's `OTLP/HTTP` endpoint supports both `json` and `proto` encodings. You need to set the `Content-Type` header to
`application/json` or `application/x-protobuf` accordingly.

Below is our recommended config for the exporter:

```yaml theme={null}
exporters:
  otlphttp:
    endpoint: https://<your-basin>.b.s2.dev/otlp
    encoding: <your-preferred-encoding>
    headers:
      Content-Type: <value-based-on-encoding>
      Authorization: Bearer <your-access-token>
    sending_queue:
      num_consumers: 1
      batch:
        flush_timeout: 50ms
        min_size: 262144 # 256KiB
        max_size: 524288 # 512KiB
        sizer: bytes
```

**Note:**

* If you don't have an access token already, you can generate one from the [dashboard](https://s2.dev/dashboard).
* You need to create a basin with `create-stream-on-append` flag on, if you want streams to be created automatically
  when signals are exported.
* `sending_queue.num_consumers: 1` ensures batches are sent in order to S2. You can increase it if ordering is not a
  concern.
* If you want to change `sending_queue.batch` config, please be aware of the limits \[[1](/docs/platform/limits#records), [2](/docs/platform/limits#appends)]
  and how that will impact your changes.


# Integrations
Source: https://s2.dev/docs/integrations/overview

Explore S2 integrations for Redis clients, AI agents, OpenTelemetry observability, stream processors, blockchain pipelines, local-first sync, and collaborative editing.

We are [always open](mailto:hi@s2.dev) to exploring more integrations.

> [card] Protocols
>
> * [Redis / Valkey clients](/docs/integrations/redis-streams) — append, replay, and tail S2 streams with Redis or Valkey clients
> * [OpenTelemetry](/docs/integrations/otlp) — traces, metrics, and logs exported to S2 over OTLP/HTTP

> [card] AI SDKs
>
> * [Vercel AI SDK](/docs/integrations/vercel-ai-sdk) — resumable chat generations, message history, and agent sessions
> * [TanStack AI](/docs/integrations/tanstack-ai) — durable chat sessions with replay, live tailing, and reconnect recovery
> * [Anthropic](/docs/integrations/anthropic) — durable Claude chat sessions with replay, live tailing, and reconnect recovery
> * [Mastra](/docs/integrations/mastra) — durable agent streams with replay and reconnect recovery

> [card] Data pipelines
>
> * [Bento](/docs/integrations/bento) — S2 input and output components for Bento pipelines
> * [Streamling](/docs/integrations/streamling) — S2 source and sink plugins for Streamling pipelines
> * [Feldera](/docs/integrations/feldera) — S2 input and output connectors for incremental SQL pipelines
> * [Goldsky](/docs/integrations/goldsky) — real-time and historical blockchain data from Turbo Pipelines

> [card] Sync
>
> * [LiveStore](/docs/integrations/livestore) — event-sourced sync backed by S2
> * [Yjs](/docs/integrations/y-s2) — collaborative editing with y-s2, S2 streams, and R2 snapshots


# Redis / Valkey clients
Source: https://s2.dev/docs/integrations/redis-streams

Use Redis or Valkey clients with S2 streams. Compare command support, understand the guarantees, and configure your connection.

Use Redis or Valkey clients to append, replay, and tail S2 streams over `RESP2` or `RESP3`. A subset of Redis Streams commands is supported, with S2's durability, consistency, and retention semantics.

> [!NOTE]
>
> In beta, and available on the cloud service only. [s2-lite](/docs/s2-lite) does not serve RESP.

## Why S2 via RESP?

If you use Redis Streams to deliver live output, S2 keeps that output available for readers who disconnect, [fall behind](#pagination-and-the-tail), or come back later. You get:

* [durable writes](#xadd)
* [streams that grow without bound](#basin-and-stream-configuration), not limited by memory
* [serverless pricing](#billing)

For workloads that keep history around, that can be [an order of magnitude or more cheaper](#retaining-history) than an equivalent provisioned Redis.

Good fits include:

> [card] AI responses and agent runs
>
> Stream token chunks, tool calls, and progress updates, and resume missed output when a user returns.

> [card] Build logs and background jobs
>
> Stream build, import, or sandbox logs, and debug a completed or failed job later.

> [card] Activity feeds
>
> Stream customer, device, or session events, and let each reader catch up at its own pace.

### When to use something else

* **Other Redis features:** only part of Redis Streams is supported, and no other Redis data types, transactions, scripting, or pub/sub. In particular, there are no consumer groups, explicit `XADD` IDs, exact trimming, `XDEL`, `XREVRANGE`, or `XINFO`. See [command support](#command-support).
* **S2 features beyond Redis:** conditional appends, fencing tokens, and [encrypted streams](/docs/concepts/encryption) need the native [S2 SDKs](/docs/sdk/languages) or [API](/docs/api/protocol).
* **In-memory latency:** S2 acknowledges a write once it is durable in multiple availability zones, within [40 ms on Express and 400 ms on Standard](/docs/concepts/appends#latency) from the same region. If you need sub-millisecond latency, use an in-memory Redis.

> [!TIP]
>
> Copy this guide as Markdown and share it with your coding agent.
>
> Ask it to review your app's Redis usage, flag compatibility gaps, and assess whether S2 fits your workload.

## Command support

Keys are UTF-8 stream names in the connection's basin. Only the forms below are supported; command names link to the Redis reference.

| Command | Supported forms | Notes |
| - | - | - |
| [`XADD`](https://redis.io/docs/latest/commands/xadd/) | `key * field value [field value ...]` | IDs are always generated. Records are limited to [1 MiB](#xadd). Missing streams need [creation on append](#basin-and-stream-configuration). |
| [`XADD NOMKSTREAM`](https://redis.io/docs/latest/commands/xadd/) | `key NOMKSTREAM * field value ...` | Returns `null` if the stream is missing. The check isn't atomic with the append; see [creation on read](#creation-on-read-and-nomkstream). |
| Pipelined [`XADD`](https://redis.io/docs/latest/commands/xadd/) | Many `XADD`s in flight. | Appended in order [per stream on a connection](#ordering-and-retries). |
| [`XTRIM`](https://redis.io/docs/latest/commands/xtrim/), [`XADD` with trimming](https://redis.io/docs/latest/commands/xadd/) | `MAXLEN ~ threshold`, `MINID ~ id` | Approximate only, and [takes effect eventually](#xtrim-and-xlen). Each trim adds an entry with no fields. No `LIMIT`. |
| [`XRANGE`](https://redis.io/docs/latest/commands/xrange/) | Inclusive and exclusive IDs, timestamp-only IDs, `-`, `+`, `COUNT` | Up to 10,000 entries per call. A short page doesn't mean you've reached the end; see [pagination](#pagination-and-the-tail). |
| [`XREAD`](https://redis.io/docs/latest/commands/xread/) | `COUNT`, `STREAMS` with explicit IDs or `$` | Up to 100 streams, sharing 10,000 entries. No `+` ID or cross-stream snapshot. |
| [`XREAD BLOCK`](https://redis.io/docs/latest/commands/xread/) | `BLOCK milliseconds`; `BLOCK 0` waits indefinitely. | See [blocking reads](#blocking-reads). |
| [`XLEN`](https://redis.io/docs/latest/commands/xlen/) | `key` | Counts [all retained entries](#xtrim-and-xlen), including trim records. |
| [`EXISTS`](https://redis.io/docs/latest/commands/exists/) | One or more keys. | Empty streams count as existing. |
| [`TYPE`](https://redis.io/docs/latest/commands/type/) | One key. | Returns `stream` or `none`. |
| [`AUTH`](https://redis.io/docs/latest/commands/auth/) | `token` | See [authentication](#authentication-and-permissions). |
| [`HELLO`](https://redis.io/docs/latest/commands/hello/) | No arguments, or `2` / `3` with optional `AUTH` and `SETNAME`. | Reports `server` `s2-resp`, Redis-compatible `version` `6.2.0`, and `s2_resp_version`. |
| [`CLIENT SETNAME`](https://redis.io/docs/latest/commands/client-setname/), [`CLIENT GETNAME`](https://redis.io/docs/latest/commands/client-getname/) | Set or get the connection name. | Up to 1,024 bytes of printable ASCII, no spaces. |
| [`CLIENT SETINFO`](https://redis.io/docs/latest/commands/client-setinfo/) | `LIB-NAME value`, `LIB-VER value` | Accepted but not stored. |
| [`SELECT`](https://redis.io/docs/latest/commands/select/) | `0` | Only database 0. |
| [`COMMAND`](https://redis.io/docs/latest/commands/command/) | Any subcommand. | Returns an empty array. |
| [`INFO`](https://redis.io/docs/latest/commands/info/) | Optional section names. | Only `server` (`redis_version:6.2.0`, `redis_mode:standalone`, `s2_resp_version`) and `persistence` (`loading:0`). |
| [`PING`](https://redis.io/docs/latest/commands/ping/), [`ECHO`](https://redis.io/docs/latest/commands/echo/), [`QUIT`](https://redis.io/docs/latest/commands/quit/) | `PING [message]`, `ECHO message`, `QUIT` | Standard replies. |

## Connecting

Connect to `{basin}.r.s2.dev:6380` over TLS 1.3, sending the same hostname as TLS SNI; connections without SNI are rejected. Clients that take a URL should use `rediss://`.

You'll need a [basin](/docs/cli/basins#create-a-basin) and an [access token](/docs/concepts/access-tokens). The examples write to a stream called `events`, so [create it](/docs/cli/streams#create-a-stream) or enable [`create_stream_on_append`](#basin-and-stream-configuration) on the basin.

With [redis-cli](https://redis.io/docs/latest/develop/tools/cli/):

```bash theme={null}
export S2_BASIN="your-basin"
export S2_ACCESS_TOKEN="your-access-token"

REDISCLI_AUTH="$S2_ACCESS_TOKEN" \
redis-cli \
  --tls --sni "${S2_BASIN}.r.s2.dev" \
  -h "${S2_BASIN}.r.s2.dev" -p 6380
```

Then, in the CLI:

```text theme={null}
XADD events * kind example
XRANGE events - + COUNT 100
```

### Client libraries

Each example appends three entries to `events`, reads the stream from the start one [page](#pagination-and-the-tail) at a time, then [tails](#tailing-a-stream) it with `XREAD BLOCK`. They read `S2_BASIN` and `S2_ACCESS_TOKEN` from the environment.

> [accordion] Python, Node.js, and Go examples
>
> ```python Python (redis-py) theme={null}
> import os
>
> import redis
>
> basin = os.environ["S2_BASIN"]
> host = f"{basin}.r.s2.dev"
>
> r = redis.Redis(
>     host=host,
>     port=6380,
>     ssl=True,
>     password=os.environ["S2_ACCESS_TOKEN"],
>     decode_responses=True,
>     socket_timeout=30,
> )
>
> stream = "events"
>
> for i in range(3):
>     print("appended", r.xadd(stream, {"kind": "example", "n": str(i)}))
>
> # Read the full stream, one page at a time.
> last_id = None
> start = "-"
> while True:
>     page = r.xrange(stream, min=start, max="+", count=100)
>     if not page:
>         break
>     for entry_id, fields in page:
>         print(entry_id, fields)
>     last_id = page[-1][0]
>     start = f"({last_id}"
>
> # Tail from the last entry read. The block time must be shorter than socket_timeout.
> cursor = last_id or "0-0"
> while True:
>     reply = r.xread({stream: cursor}, count=100, block=10_000)
>     for _, entries in reply:
>         for entry_id, fields in entries:
>             print(entry_id, fields)
>             cursor = entry_id
> ```
>
> ```javascript Node.js (node-redis) theme={null}
> import { createClient } from "redis";
>
> const basin = process.env.S2_BASIN;
> const host = `${basin}.r.s2.dev`;
>
> const client = createClient({
>   socket: { host, port: 6380, tls: true, servername: host },
>   password: process.env.S2_ACCESS_TOKEN,
> });
> client.on("error", (err) => console.error(err));
> await client.connect();
>
> const stream = "events";
>
> for (let i = 0; i < 3; i++) {
>   console.log("appended", await client.xAdd(stream, "*", { kind: "example", n: String(i) }));
> }
>
> // Read the full stream, one page at a time.
> let lastId = null;
> let start = "-";
> while (true) {
>   const page = await client.xRange(stream, start, "+", { COUNT: 100 });
>   if (page.length === 0) break;
>   for (const { id, message } of page) console.log(id, message);
>   lastId = page.at(-1).id;
>   start = `(${lastId}`;
> }
>
> // Blocking reads hold the connection, so tail on a dedicated one.
> const tail = client.duplicate();
> tail.on("error", (err) => console.error(err));
> await tail.connect();
>
> let cursor = lastId ?? "0-0";
> while (true) {
>   const reply = await tail.xRead({ key: stream, id: cursor }, { COUNT: 100, BLOCK: 10_000 });
>   for (const { messages } of reply ?? []) {
>     for (const { id, message } of messages) {
>       console.log(id, message);
>       cursor = id;
>     }
>   }
> }
> ```
>
> ```javascript Node.js (ioredis) theme={null}
> import Redis from "ioredis";
>
> const basin = process.env.S2_BASIN;
> const host = `${basin}.r.s2.dev`;
>
> const redis = new Redis({
>   host,
>   port: 6380,
>   tls: { servername: host },
>   password: process.env.S2_ACCESS_TOKEN,
> });
>
> const stream = "events";
>
> for (let i = 0; i < 3; i++) {
>   console.log("appended", await redis.xadd(stream, "*", "kind", "example", "n", String(i)));
> }
>
> // Read the full stream, one page at a time.
> let lastId = null;
> let start = "-";
> while (true) {
>   const page = await redis.xrange(stream, start, "+", "COUNT", 100);
>   if (page.length === 0) break;
>   for (const [id, fields] of page) console.log(id, fields);
>   lastId = page.at(-1)[0];
>   start = `(${lastId}`;
> }
>
> // Blocking reads hold the connection, so tail on a dedicated one.
> const tail = redis.duplicate();
>
> let cursor = lastId ?? "0-0";
> while (true) {
>   const reply = await tail.xread("COUNT", 100, "BLOCK", 10_000, "STREAMS", stream, cursor);
>   for (const [, entries] of reply ?? []) {
>     for (const [id, fields] of entries) {
>       console.log(id, fields);
>       cursor = id;
>     }
>   }
> }
> ```
>
> ```go Go (go-redis) theme={null}
> package main
>
> import (
> 	"context"
> 	"crypto/tls"
> 	"errors"
> 	"fmt"
> 	"os"
> 	"time"
>
> 	"github.com/redis/go-redis/v9"
> )
>
> func main() {
> 	ctx := context.Background()
> 	basin := os.Getenv("S2_BASIN")
> 	host := basin + ".r.s2.dev"
>
> 	rdb := redis.NewClient(&redis.Options{
> 		Addr:      host + ":6380",
> 		Password:  os.Getenv("S2_ACCESS_TOKEN"),
> 		TLSConfig: &tls.Config{ServerName: host, MinVersion: tls.VersionTLS13},
> 	})
> 	defer rdb.Close()
>
> 	stream := "events"
>
> 	for i := range 3 {
> 		id, err := rdb.XAdd(ctx, &redis.XAddArgs{
> 			Stream: stream,
> 			Values: map[string]any{"kind": "example", "n": i},
> 		}).Result()
> 		if err != nil {
> 			panic(err)
> 		}
> 		fmt.Println("appended", id)
> 	}
>
> 	// Read the full stream, one page at a time.
> 	lastID := ""
> 	start := "-"
> 	for {
> 		page, err := rdb.XRangeN(ctx, stream, start, "+", 100).Result()
> 		if err != nil {
> 			panic(err)
> 		}
> 		if len(page) == 0 {
> 			break
> 		}
> 		for _, msg := range page {
> 			fmt.Println(msg.ID, msg.Values)
> 		}
> 		lastID = page[len(page)-1].ID
> 		start = "(" + lastID
> 	}
>
> 	// Tail from the last entry read. go-redis extends the read timeout by Block.
> 	cursor := lastID
> 	if cursor == "" {
> 		cursor = "0-0"
> 	}
> 	for {
> 		streams, err := rdb.XRead(ctx, &redis.XReadArgs{
> 			Streams: []string{stream, cursor},
> 			Count:   100,
> 			Block:   10 * time.Second,
> 		}).Result()
> 		if errors.Is(err, redis.Nil) {
> 			continue
> 		}
> 		if err != nil {
> 			panic(err)
> 		}
> 		for _, s := range streams {
> 			for _, msg := range s.Messages {
> 				fmt.Println(msg.ID, msg.Values)
> 				cursor = msg.ID
> 			}
> 		}
> 	}
> }
> ```

Client notes:

* Keep the client's read timeout longer than `BLOCK`. redis-py needs an explicit `socket_timeout`; go-redis extends its read deadline automatically.
* A blocking read holds its connection, so clients that share one connection across commands, such as node-redis and ioredis, need a separate connection for it.

### Authentication and permissions

> [!NOTE]
>
> Each connection is tied to the basin in its hostname, `{basin}.r.s2.dev`, which must also be sent as TLS SNI. Only streams in that basin are reachable; use a separate connection for each basin.

Authenticate with `AUTH S2_ACCESS_TOKEN`; in a client library, set the token as the password. No username is needed, though `default` or the basin name work if your client sends one. Authenticate within 30 seconds of connecting.

`AUTH` only checks that the token is valid. Each command is then authorized against the token's [scope](/docs/concepts/access-tokens#scope): the basin, the stream names, and the [operations the command uses](#command-permissions). Anything the token doesn't allow replies `NOPERM`. With [auto-prefixing](/docs/concepts/access-tokens#resources), keys are prefixed too: with the prefix `user1/`, key `logs` is stream `user1/logs`.

### Basin and stream configuration

Basins and streams are configured through the dashboard, [CLI](/docs/cli/basins#reconfigure-a-basin), or API, not through Redis. The [settings](/docs/concepts/configs) that matter here:

* **Creation:** create streams up front, or enable the basin's `create_stream_on_append` so `XADD` creates them. New streams use the basin's default stream config.
* **Retention:** defaults to 7 days; `infinite` keeps everything. Retention beyond the [free-tier limit](/docs/platform/limits#retention) requires a payment method.
* **Storage class:** Express has lower latency than Standard; both are equally durable.
* **Timestamping:** Redis appends carry no client timestamp, so use `arrival` or the default `client-prefer`; `client-require` rejects them.
* **Encryption:** [encrypted streams](/docs/concepts/encryption) need a key on each request, which Redis commands can't pass, so they can't be used here.

#### Creation on read and `NOMKSTREAM`

For a stream that doesn't exist yet:

| Basin setting enabled | `XADD` | `XADD NOMKSTREAM` |
| - | - | - |
| Neither | Error | Returns `null` |
| `create_stream_on_append` | Creates the stream and appends | Returns `null` |
| `create_stream_on_read` | Error | Creates the stream and appends |
| Both | Creates the stream and appends | Creates the stream and appends |

`NOMKSTREAM` checks for the stream before appending, and with `create_stream_on_read` the check itself creates it. The reads made by `XLEN`, `XTRIM`, and inline trimming can also create streams. Because the check and the append are separate, a stream deleted in between can be recreated by `create_stream_on_append`.

### Command permissions

A token needs the [operations](/docs/concepts/access-tokens#operations) a command uses in its `scope.ops`. The table lists what each command can call; with batching and caching, not every call happens every time.

| Command | Possible S2 operations | Permissions to grant |
| - | - | - |
| `AUTH`, `HELLO ... AUTH` | `list-streams` (limit 1) to validate the token | Any valid token; see below |
| `XADD` | `append` | `append` |
| `XADD NOMKSTREAM` | `check-tail`, then `append` | `check-tail`, `append` |
| `XADD` with `MAXLEN ~` or `MINID ~` | `read`, `check-tail`, `append` | `read`, `check-tail`, `append`, `trim` |
| `XTRIM` | `read`, `check-tail`, and `append` if there is anything to trim | `read`, `check-tail`, `append`, `trim` |
| `XRANGE`, nonblocking `XREAD`, `XREAD BLOCK` on several streams | `read`, plus `check-tail` to confirm a cached read is caught up | `read`, `check-tail` |
| `XREAD BLOCK` on one stream | `read` | `read` |
| `XLEN` | `read`, `check-tail` | `read`, `check-tail` |
| `EXISTS`, `TYPE` | `get-stream-config` per key | `get-stream-config` |
| `HELLO`, `CLIENT`, `SELECT`, `COMMAND`, `INFO`, `PING`, `ECHO`, `QUIT` | None | None |

A few details:

* `AUTH` accepts a permission error from its `list-streams` call, so tokens limited to stream operations can still authenticate.
* Grant `check-tail` alongside `read`. A token with only `read` works until a read needs to check the tail.
* `trim` authorizes appending a trim record, so trimming needs both `trim` and `append`. A trim with nothing to remove may use neither.
* Streams created by the basin's creation settings don't need the `create-stream` permission.

Alternatively, grant [operation groups](/docs/concepts/access-tokens#operations): `stream.read` covers `read` and `check-tail`; `stream.write` covers `append`, `trim`, and `fence`; and `basin.read` covers `get-stream-config` and `list-streams`.

## Core semantics

### `XADD`

`XADD key * field value ...` returns an ID once the record is durable in multiple availability zones, on [either storage class](/docs/concepts/appends).

IDs are always generated, as `<timestamp>-<seq_num>`: Unix milliseconds, then the record's S2 sequence number. Sequence numbers only increase; they don't reset when the timestamp changes or after trimming.

Each record can be up to **1 MiB in [metered bytes](/docs/platform/limits#records)**. For `n` field/value pairs:

```text theme={null}
metered bytes = 20 + 2*n + total bytes in all field names and values
```

With inline trimming, the record and its trim record must fit in 1 MiB together.

#### Ordering and retries

Pipelined `XADD`s on one connection are appended in order **per stream**, and replies come back in request order. Appends to different streams, or from different connections, can become visible in any order. Other commands wait for earlier `XADD`s on their connection to finish. A pipeline is not a transaction.

> [!NOTE]
>
> If the connection drops before an `XADD` is acknowledged, the write may or may not have happened. Retrying it can create a duplicate.

The service retries an append only when it knows the earlier attempt had no effect; otherwise it closes the connection. There is no idempotency key over Redis. For stronger guarantees, use native [conditional appends](/docs/concepts/concurrency-control).

### `XREAD` and `XRANGE`

Reads return entries in stream order and are [linearizable](/docs/concepts/reads#consistency): once an `XADD` is acknowledged, later reads see it. An `XREAD` across several streams is not a snapshot of all of them at one point in time.

#### Pagination and the tail

**A page with fewer than `COUNT` entries doesn't mean you've caught up.** Pages can end early because of the 10,000-entry cap or size limits. `XREAD` also splits the cap evenly across its streams, so with 100 streams each returns at most 100 entries.

Keep reading until you get an empty result:

* **`XRANGE`:** start the next page at `(last-id`. An empty array means the range is exhausted.
* **Nonblocking `XREAD`:** pass each stream's last returned ID, keeping the previous cursor for streams that returned nothing. `null` means nothing more is available yet.

For example, after a page ending at `1713812735000-42`:

```text theme={null}
XRANGE events (1713812735000-42 + COUNT 100
XREAD COUNT 100 STREAMS events 1713812735000-42
```

Entries with no fields, such as [trim records](#xtrim-and-xlen), are still entries: continue past their IDs.

Store cursors in your application. `$` means "from the current tail", so using it again between reads can skip entries; continue from the returned IDs instead.

#### Blocking reads

`XREAD BLOCK milliseconds` waits for new entries; `BLOCK 0` waits indefinitely. A finite wait includes setting up the read, so a short one can return `null` even when entries exist. Use a nonblocking read to check whether you've caught up, and after a timeout, retry with the same cursors.

Set the client's socket timeout longer than the wait, or disable it for `BLOCK 0`; redis-py 8, for example, defaults to 5 seconds. Use a separate connection if other commands need to run while a read blocks.

#### Tailing a stream

Once a consumer catches up, use `XREAD BLOCK` instead of polling. An empty `XRANGE` or nonblocking `XREAD` near the tail can cost a billed `check-tail`; polling a quiet stream every 100 ms adds about 864,000 operations a day. `XREAD BLOCK` on one stream waits on its open read without tail checks. On several streams, it checks each quiet stream once per command, so longer waits cost less.

### `XTRIM` and `XLEN`

Only approximate trimming is supported: `MAXLEN ~ threshold` and `MINID ~ id`. A trim appends a [command record](/docs/concepts/command-records), which appears in reads as an entry with an empty field array and counts toward `COUNT`, `MAXLEN`, and `XLEN`. Other command records, such as fences, and records written natively without the [Redis encoding](#message-encoding) appear the same way.

[Trimming is eventual](/docs/concepts/trimming): older entries can stay readable for a while after `XTRIM` returns. Its return value estimates how many entries the trim will remove.

For example, with three entries `A`, `B`, and `C`, `XTRIM events MAXLEN ~ 2` returns `2`:

```text theme={null}
Before:          A -> B -> C          XLEN = 3
Trim pending:    A -> B -> C -> []    XLEN can be 4
Trim applied:              C -> []    XLEN = 2
```

`[]` is the trim record. It has its own ID, so continue from it like any other entry. This means `MAXLEN ~ 1` can leave only the trim record, and `MAXLEN ~ 0` can eventually leave the stream empty.

Inline `XADD` trimming appends the entry and the trim record atomically, but concurrent writers can still leave more than `MAXLEN` entries.

`XLEN` counts retained entries, including trim records, and can be briefly off while trimming or retention takes effect. A stream can be empty and still exist: `XLEN` returns 0 while `EXISTS` returns 1. [`delete_on_empty`](/docs/concepts/configs#delete-on-empty) can remove such streams.

[Retention](/docs/concepts/configs#retention) removes old entries independently of `XTRIM`. Once removed, entries can't be read through any interface.

## Interoperability with S2 APIs

Redis and native S2 clients can use the same streams. Trimming, retention, and deletion affect both.

### IDs and positions

In `1713812735000-42`, `1713812735000` is the record's timestamp and `42` is its S2 sequence number. To resume natively after this entry, read from sequence number 43. Timestamps pass through unchanged, so if native writers use a unit other than milliseconds, Redis IDs do too.

### Message encoding

`XADD` writes each field/value pair as a [record header](/docs/concepts/records#anatomy-of-a-record), in order, after a first header `s2-resp.v: 1`. The body is empty:

```text theme={null}
headers: s2-resp.v: 1, field1: value1, field2: value2, ...
body:    empty
```

For example, this entry:

```text theme={null}
> XADD events * sensor t1 temp 21.5
"1713812735000-42"
> XRANGE events 1713812735000-42 1713812735000-42
1) 1) "1713812735000-42"
   2) 1) "sensor"
      2) "t1"
      3) "temp"
      4) "21.5"
```

reads natively as a record with those headers and no body:

```bash theme={null}
s2 read s2://my-basin/events -s 42 -n 1 --format json
```

```json theme={null}
{
  "seq_num": 42,
  "timestamp": 1713812735000,
  "headers": [
    ["s2-resp.v", "1"],
    ["sensor", "t1"],
    ["temp", "21.5"]
  ]
}
```

Use `--format json-base64` for fields or values that aren't valid UTF-8.

Fields and values are binary-safe, and pair order and duplicates are preserved. Headers can't have empty names, so a field name that is empty or starts with a NUL byte is stored with one extra leading NUL byte, which is removed on read.

Native writers can produce Redis entries the same way. Only the first header marks a Redis entry; a later `s2-resp.v` header is an ordinary field. Records without the marker appear in Redis as entries with no fields. A record with the marker and a non-empty body causes a read error.

### Concurrency controls

Append conditions (`match_seq_num` and fencing tokens) are only available through the [native API](/docs/concepts/concurrency-control). Fencing doesn't stop Redis writers, because their appends never carry a token.

## Performance and quality of service

Latency depends on the storage class and your distance from the basin; see [append](/docs/concepts/appends#latency) and [read](/docs/concepts/reads#latency) latency. For throughput, pipeline `XADD`s: the service batches them per stream and connection. `NOMKSTREAM` and inline trimming take extra round trips, so benchmark them separately.

Reuse connections and keep pipeline depth bounded. Fewer, busier connections batch better and can [reuse read sessions](#read-session-reuse). Give blocking readers their own connections, since a slow reply holds up later replies on the same connection.

Under load, the service can slow down accepting commands, reply `TRYAGAIN`, or close connections. Reconnect with backoff.

### Errors and reconnection

| Reply | Meaning | Connection | What to do |
| - | - | - | - |
| `NOAUTH Authentication required.` | The command was sent before `AUTH`. | Stays open | Authenticate, then retry. |
| `NOPERM ...` | The token doesn't allow this operation, stream, or basin. | Stays open | Fix the token's [permissions](#command-permissions). |
| `TRYAGAIN hot server: ... command dropped, retry` | The server was short on memory and dropped the command before running it. | Stays open | Retry the same command. |
| `TRYAGAIN hot server: memory admission timed out; reconnect and retry` | The server was short on memory. | Closed | Reconnect with backoff and retry the command. |
| `ERR server memory limit reached; try again later` | The server refused a new connection. | Closed | Reconnect with backoff. |
| `ERR S2 append outcome is ambiguous; closing connection: ...` | An append may or may not have committed. | Closed | Reconnect; don't blindly replay [unacknowledged writes](#ordering-and-retries). |
| `ERR session closing; blocking read interrupted` | The server is restarting, or the client closed its side of the connection. | Closing | Reconnect and resume from your cursor. |

Other limits:

* A command must arrive in full within 30 seconds and finish within 30 seconds. `XREAD BLOCK` adds its wait to that; `BLOCK 0` has no limit. A command that runs too long closes the connection without a reply.
* A command can be at most 2 MiB with 16,384 arguments. Many small fields can hit these limits before the 1 MiB record limit.
* On restart or scale-in, connections drain for up to 20 seconds and then close. Treat writes without replies as [ambiguous](#ordering-and-retries).

## Billing

Usage is billed at [S2 pricing](https://s2.dev/pricing) for storage, writes, reads, and operations, including the Redis encoding overhead and trim records. Redis commands don't map one-to-one to operations: many `XADD`s can share one append, and some commands make several calls. The [permissions table](#command-permissions) lists the operations each command can make; this includes one `list-streams` call per connection when it authenticates.

### Retaining history

Consider **100 events per second across 32 streams**, each with a 1 KiB value and read once by one consumer per stream. Each column is the estimated monthly cost of retaining that much history:

| Service | 1 day | 7 days | 30 days |
| - | -: | -: | -: |
| Payload retained | 8 GiB | 58 GiB | 247 GiB |
| **S2 Express** | **\$49** | **\$51** | **\$61** |
| Amazon MemoryDB for Valkey, primary only | \$248 | \$1,259 | \$5,034 |
| Amazon ElastiCache for Valkey 9, synchronous durability | \$566 | \$2,263 | \$9,678 |

These are **illustrative cost estimates** for this workload, not measured hosted bills or a benchmark of equivalent latency. Rates were checked on September 28, 2026, using a 730-hour month and AWS on-demand pricing in `us-east-1`. Sources: [S2](https://s2.dev/pricing), [MemoryDB](https://aws.amazon.com/memorydb/pricing/), and [ElastiCache](https://aws.amazon.com/elasticache/pricing/).

> [accordion] Workload and sizing assumptions
>
> The workload uses 32 equal-rate streams, one persistent writer and one dedicated blocking reader per stream. Each event has one field, `d`, and a 1,024-byte value. The consumer reads every event once without additional historical replay. Retention is at steady state after the full retention period has accumulated.
>
> S2's estimate includes 1,057 [metered bytes](#xadd) per record, retained storage, writes, internet-priced reads, session operations, and resource and authentication charges. It assumes connections and read sessions remain reusable. At 30 days on Express, that is about \$13 for storage, \$19 for writes, \$26 for reads, and \$3 for operations and resources.
>
> AWS sizing uses approximately 1,385 bytes per record measured in a local Valkey 8.1.1 stream. This is a sizing proxy; hosted memory usage and throughput have not been measured. The model assumes balanced key placement, with 20% memory headroom for MemoryDB and ElastiCache's [default 25% memory reserve](https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/redis-memory-management.html).
>
> Each column uses the cheapest fitting on-demand configuration. MemoryDB uses five `db.t4g.medium` shards at 1 day and one `db.r6g.4xlarge` at 7 days. At 30 days, it uses one `db.r6g.16xlarge`, which costs \$6.8957 per hour. MemoryDB persists writes to a Multi-AZ transaction log, so [replicas are optional](https://docs.aws.amazon.com/memorydb/latest/devguide/components.html); adding one replica per shard for availability doubles its estimate. ElastiCache uses one `cache.r6g.xlarge` shard at 1 day, one `cache.r6g.4xlarge` shard at 7 days, and two `cache.r8g.8xlarge` shards at 30 days, priced including the synchronous-durability premium. Synchronous durability requires a replica per shard, so every ElastiCache estimate includes one. ElastiCache's [synchronous durability](https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Durability.Configuring.html) acknowledges writes after persisting them to a Multi-AZ transaction log; this mode [does not support Serverless or data tiering](https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/Durability.Limitations.html). The MemoryDB stream keys exceed its [128 MiB tiering threshold](https://docs.aws.amazon.com/memorydb/latest/devguide/data-tiering-prerequisites.html), so the model keeps their history in memory.
>
> Estimates exclude taxes, promotional credits, commitment discounts, additional snapshots, and application-side compute and network charges. Different memory footprints, reconnects, polling, and unused read-ahead can change the result.

In practice, teams running Redis Streams rarely keep a month of history in memory. Instead, they trim each stream to a short window with `XTRIM` and run a consumer that archives entries to object storage. With a 1-day window, that costs roughly the 1-day column above plus about \$6 a month to keep 247 GiB in [S3 Standard](https://aws.amazon.com/s3/pricing/). The cost is operational: an archiver to run and monitor, two read paths, and replays that must hand off from the archive to the stream at the right ID without gaps or duplicates. With S2, the history stays in the stream and every reader uses the same commands.

Shorter retention or frequent rereading narrows the cost gap. Compare [latency and throughput requirements](#performance-and-quality-of-service) as well as cost.

### Read-session reuse

Reading a stream sequentially on one connection reuses the same S2 read session, which is [billed by the minute](https://s2.dev/pricing). Reconnecting, jumping to a different cursor, or switching between blocking and nonblocking reads can open a new session.

Data the service reads ahead is billed even if the client never requests it. Read transfer is priced by the client's network path; connecting through the public endpoint doesn't qualify for private-network pricing.


# Streamling
Source: https://s2.dev/docs/integrations/streamling

Use Streamling's S2 source and sink plugins to connect S2 with Streamling pipelines.

[Streamling](https://github.com/goldsky-io/streamling) is an open source columnar streaming runtime from [Goldsky](https://goldsky.com/), built for high-throughput real-time pipelines with checkpointing and at-least-once delivery.

S2 support ships as two plugins in [streamling-community-plugins](https://github.com/goldsky-io/streamling-community-plugins) v0.2.0 and later:

* **`s2_source`** reads S2 streams into Arrow batches for any downstream Streamling transform or sink — exact stream names or a prefix, with newly created streams discovered automatically. Records can flow through as the raw envelope (`stream`, `seq_num`, `timestamp`, `headers`, `body`) or as typed columns decoded from JSON bodies.
* **`s2_sink`** appends rows from any Streamling source or transform as JSON records to an S2 stream, with optional per-row routing across streams via a template like `events/{tenant}`.

Use `s2_source` when S2 is the durable input to a Streamling pipeline, and `s2_sink` when S2 is the durable output. The examples below use ClickHouse and Postgres CDC to show useful Streamling sources and sinks in both directions, but the S2 plugins compose with Streamling's other sources, sinks, and transforms.

The two plugins also compose into CDC round trips: the sink carries each row's operation as a [Debezium](https://debezium.io/)-style `dbz.op` record header, and the source restores it, so inserts, updates, and deletes survive a trip through S2 and apply correctly in a downstream database.

## Reference

Configuration tables for both plugins are in the [streamling-community-plugins README](https://github.com/goldsky-io/streamling-community-plugins#available-plugins).

> [!TIP]
>
> Both plugins accept an `endpoint` option, so pipelines can run against [s2-lite](/docs/s2-lite) — handy for CI and local development. See the [CLI endpoint overrides](/docs/cli/configuration#environment-variables) for the accompanying CLI setup; use any non-empty access token value locally, since s2-lite ignores it.

## Prerequisites

**1. Set up the S2 CLI**

Generate an access token from the [dashboard](https://s2.dev/dashboard), then:

```bash theme={null}
export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN"
```

Install the [S2 CLI](/docs/quickstart#get-started-with-the-cli). The CLI automatically reads the same `S2_ACCESS_TOKEN` environment variable.

**2. Create a basin**

```bash theme={null}
export S2_BASIN="YOUR_BASIN_NAME"
s2 create-basin ${S2_BASIN}
```

**3. Install and start Docker**

The ClickHouse and Postgres examples use Docker. [Install Docker](https://docs.docker.com/get-started/get-docker/), start it, and verify that the daemon is available:

```bash theme={null}
docker info
```

**4. Install Streamling**

```bash theme={null}
curl -fsSL https://www.streamling.dev/install.sh | bash
```

**5. Create a walkthrough directory**

Keep the pipeline files and checkpoint state for this walkthrough together:

```bash theme={null}
mkdir -p "$HOME/streamling-s2-demo"
cd "$HOME/streamling-s2-demo"
```

**6. Install the community plugins**

Download the shared library for your platform from the [releases page](https://github.com/goldsky-io/streamling-community-plugins/releases) and tell Streamling where to find it:

> [tab] macOS
>
> ```bash theme={null}
> mkdir -p "$HOME/.streamling/plugins"
> curl -fsSLO https://github.com/goldsky-io/streamling-community-plugins/releases/latest/download/community-plugins-darwin-aarch64.tar.gz
> tar -xzf community-plugins-darwin-aarch64.tar.gz -C "$HOME/.streamling/plugins"
> ```
>
> Use the `darwin-x86_64` asset on Intel Macs.

> [tab] Linux
>
> ```bash theme={null}
> mkdir -p "$HOME/.streamling/plugins"
> curl -fsSLO https://github.com/goldsky-io/streamling-community-plugins/releases/latest/download/community-plugins-linux-x86_64.tar.gz
> tar -xzf community-plugins-linux-x86_64.tar.gz -C "$HOME/.streamling/plugins"
> ```
>
> Use the `linux-aarch64` asset on ARM.

The run commands below set `STREAMLING__PLUGIN__PATH` inline, so they also work in a fresh terminal.

Run every example from `$HOME/streamling-s2-demo`. Streamling stores checkpoints there in `state.db`; see [Clean up and rerun](#clean-up-and-rerun) before replaying the walkthrough from the beginning.

> [!NOTE]
>
> Shell variables and the working directory do not carry into a new terminal. Whenever the walkthrough opens another terminal, return to the walkthrough directory and set `S2_BASIN` again; also set `S2_ACCESS_TOKEN` before starting a Streamling pipeline:
>
> ```bash theme={null}
> cd "$HOME/streamling-s2-demo"
> export S2_BASIN="YOUR_BASIN_NAME"
> export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN"
> ```

## Example: Events into ClickHouse

This example sends `POST → S2 → Streamling → ClickHouse`: services append JSON events to S2 streams over the [REST API](/docs/api/protocol), and a Streamling pipeline tails every stream under a prefix, decodes the events into typed columns, and upserts them into ClickHouse.

**1. Create event streams and append some events**

```bash theme={null}
s2 create-stream s2://${S2_BASIN}/events-web
s2 create-stream s2://${S2_BASIN}/events-mobile
```

Append an event to each stream:

```bash theme={null}
curl -fsS -X POST "https://${S2_BASIN}.b.s2.dev/v1/streams/events-web/records" \
  -H "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"records": [{"body": "{\"id\":1,\"event\":\"page_view\",\"amount\":0.25}"}]}'

curl -fsS -X POST "https://${S2_BASIN}.b.s2.dev/v1/streams/events-mobile/records" \
  -H "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"records": [{"body": "{\"id\":2,\"event\":\"app_open\",\"amount\":0.75}"}]}'
```

Each append is acknowledged with the durable sequence number range:

```json theme={null}
{"start":{"seq_num":0,"timestamp":1783548986349},"end":{"seq_num":1,"timestamp":1783548986349},"tail":{"seq_num":1,"timestamp":1783548986349}}
```

**2. Start ClickHouse**

```bash theme={null}
docker run -d --name s2-streamling-demo-clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server &&
curl -fs --connect-timeout 2 --retry 30 --retry-delay 1 \
  --retry-connrefused --retry-max-time 60 http://localhost:8123/ping > /dev/null ||
  { echo "ClickHouse did not become ready" >&2; false; }
```

**3. Write the pipeline**

Create `events-to-clickhouse.yaml`:

```yaml theme={null}
sources:
  events:
    type: s2_source
    stream_prefix: events-
    schema: "id:int64,event:string,amount:float64"
    include_metadata: true
    primary_key: id

transforms: {}

sinks:
  ch_events:
    type: clickhouse
    from: events
    table: events
    primary_key: id
```

The `schema` option decodes JSON bodies into typed columns, and `include_metadata` adds `_s2_stream`, `_s2_seq_num`, and `_s2_timestamp` columns. The ClickHouse sink creates the table on first write.

**4. Run it**

Connection details are passed as environment variables, keeping secrets out of the pipeline definition:

```bash theme={null}
STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SOURCE__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SOURCE__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
STREAMLING__CLICKHOUSE_SINK__PASSWORD=password \
streamling events-to-clickhouse.yaml
```

The pipeline reads both streams from the beginning and keeps tailing them for new records. Leave it running while you query ClickHouse in the next step.

**5. Query ClickHouse**

In another terminal:

```bash theme={null}
docker exec s2-streamling-demo-clickhouse clickhouse-client --password password \
  --query "SELECT id, event, amount, _s2_stream FROM events FINAL ORDER BY id"
```

```
1	page_view	0.25	events-web
2	app_open	0.75	events-mobile
```

Append more events with `curl` and they show up in ClickHouse within moments. You can even create a brand new `events-*` stream: the source re-lists the prefix every 60 seconds (tunable via `update_streams_interval_secs`) and starts reading it automatically.

> [!TIP]
>
> Events posted to S2 are flowing into ClickHouse as typed rows.

When you are ready to continue, return to the terminal running this pipeline and press <kbd>Ctrl</kbd>+<kbd>C</kbd>.

## Example: Postgres CDC into S2

This example sends `Postgres CDC → Streamling → S2`: Streamling's `postgres_cdc_source` performs an initial table copy followed by continuous logical replication, and `s2_sink` appends every change to an S2 stream with the operation preserved as a `dbz.op` record header.

**1. Start Postgres with logical replication**

```bash theme={null}
docker run -d --name s2-streamling-demo-postgres -p 5433:5432 \
  -e POSTGRES_PASSWORD=postgres postgres:16 -c wal_level=logical &&
for attempt in {1..30}; do
  if docker exec s2-streamling-demo-postgres pg_isready -h 127.0.0.1 -U postgres > /dev/null 2>&1; then
    break
  elif [ "${attempt}" -eq 30 ]; then
    echo "Postgres did not become ready" >&2
    false
  else
    sleep 1
  fi
done
```

Create a table with some rows:

```bash theme={null}
docker exec s2-streamling-demo-postgres psql -v ON_ERROR_STOP=1 -U postgres -c "
  CREATE TABLE products (id BIGINT PRIMARY KEY, name TEXT, price DOUBLE PRECISION);
  INSERT INTO products VALUES (1, 'Avocados', 5.99), (2, 'Salmon', 14.99), (3, 'Sourdough', 6.99);"
```

**2. Write the pipeline**

Create `postgres-cdc-to-s2.yaml`:

```yaml theme={null}
sources:
  products:
    type: postgres_cdc_source
    host: localhost
    port: 5433
    database: postgres
    username: postgres
    password: postgres
    table: products
    primary_key: id
    publication_name: streamling_products
    slot_name: streamling_products
    memory_backpressure_enabled: false

transforms: {}

sinks:
  s2_products:
    type: s2_sink
    from: products
    stream: cdc/products
```

The target stream is created automatically on first append (`ensure_stream` defaults to `true`).

> [!NOTE]
>
> `memory_backpressure_enabled: false` suits local machines, where system memory use often sits above the plugin's default backpressure threshold. See the [plugin README](https://github.com/goldsky-io/streamling-community-plugins#postgres-cdc-source-postgres_cdc_source) for production guidance, including least-privilege Postgres permissions.

**3. Run it**

```bash theme={null}
STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SINK__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SINK__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
streamling postgres-cdc-to-s2.yaml
```

Leave this pipeline running while you read from S2 and apply Postgres changes.

**4. Read the changes from S2**

In another terminal, the initial table copy is already there, each record carrying its operation as a header:

```bash theme={null}
s2 read s2://${S2_BASIN}/cdc/products -s 0 --format json
```

```json theme={null}
{"seq_num":0,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":1,\"name\":\"Avocados\",\"price\":5.99}"}
{"seq_num":1,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":2,\"name\":\"Salmon\",\"price\":14.99}"}
{"seq_num":2,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":3,\"name\":\"Sourdough\",\"price\":6.99}"}
```

Leave the read session tailing. In a third terminal, return to the walkthrough directory, set `S2_BASIN`, and make some changes:

```bash theme={null}
docker exec s2-streamling-demo-postgres psql -v ON_ERROR_STOP=1 -U postgres -c "
  UPDATE products SET price = 4.99 WHERE id = 1;
  DELETE FROM products WHERE id = 2;"
```

The update and delete stream in with `u` and `d` ops:

```json theme={null}
{"seq_num":3,"timestamp":1783549052191,"headers":[["dbz.op","u"]],"body":"{\"id\":1,\"name\":\"Avocados\",\"price\":4.99}"}
{"seq_num":4,"timestamp":1783549052191,"headers":[["dbz.op","d"]],"body":"{\"id\":2}"}
```

> [!TIP]
>
> Postgres changes are flowing into your S2 stream in order, with CDC semantics intact.

## Example: Postgres to ClickHouse through S2

Because `s2_source` maps `dbz.op` headers back to row operations, the CDC stream from the previous section can drive a second pipeline into ClickHouse — S2 acting as the durable, replayable changelog between the two.

If the ClickHouse container from the first example is not already running, start it:

```bash theme={null}
docker run -d --name s2-streamling-demo-clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server &&
curl -fs --connect-timeout 2 --retry 30 --retry-delay 1 \
  --retry-connrefused --retry-max-time 60 http://localhost:8123/ping > /dev/null ||
  { echo "ClickHouse did not become ready" >&2; false; }
```

In a new terminal, create `cdc-to-clickhouse.yaml`:

```yaml theme={null}
sources:
  products_cdc:
    type: s2_source
    streams: cdc/products
    schema: "id:int64,name:string?,price:float64?"
    primary_key: id

transforms: {}

sinks:
  ch_products:
    type: clickhouse
    from: products_cdc
    table: products
    primary_key: id
```

Run the second pipeline:

```bash theme={null}
STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SOURCE__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SOURCE__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
STREAMLING__CLICKHOUSE_SINK__PASSWORD=password \
streamling cdc-to-clickhouse.yaml
```

Non-key columns are marked nullable (`?`) since delete records carry only the primary key. In another terminal, query ClickHouse to see the update applied and the deleted row gone:

```bash theme={null}
docker exec s2-streamling-demo-clickhouse clickhouse-client --password password \
  --query "SELECT id, name, price FROM products FINAL ORDER BY id"
```

```
1	Avocados	4.99
3	Sourdough	6.99
```

> [!NOTE]
>
> Streamling checkpoints source positions in its state backend — by default a SQLite file at `./state.db`, so a restarted pipeline resumes where it left off rather than re-reading streams. Postgres is available as a [state backend](https://github.com/goldsky-io/streamling#state-backends) for production. Delivery is at-least-once end to end. The ClickHouse sink converges on the latest row per primary key, so replayed rows do not change the final table state. The S2 changelog itself can contain duplicates after ambiguous retries or checkpoint replay; consumers should deduplicate if each change must be processed exactly once.

## Clean up and rerun

Stop every Streamling process and the tailing `s2 read` command with <kbd>Ctrl</kbd>+<kbd>C</kbd>. Then remove the walkthrough's containers and their anonymous data volumes:

```bash theme={null}
docker rm -fv s2-streamling-demo-clickhouse s2-streamling-demo-postgres
```

To replay the walkthrough from sequence number 0, also remove its checkpoint file and streams. The `S2_BASIN` guard prevents accidental deletion from an unset variable; only continue if the named streams were created for this walkthrough:

```bash theme={null}
if [ -z "${S2_BASIN:-}" ]; then
  echo "Set S2_BASIN to the walkthrough basin before cleaning up" >&2
else
  rm -f "$HOME/streamling-s2-demo/state.db"
  s2 delete-stream s2://${S2_BASIN}/events-web
  s2 delete-stream s2://${S2_BASIN}/events-mobile
  s2 delete-stream s2://${S2_BASIN}/cdc/products
fi
```

Wait for the stream deletions to finish before recreating them. Keep the existing basin and resume at [Events into ClickHouse](#example-events-into-clickhouse) instead of creating the basin again.


# TanStack AI
Source: https://s2.dev/docs/integrations/tanstack-ai

Persist TanStack AI stream chunks on S2 and connect them to TanStack useChat.

S2 integrates with [TanStack AI](https://tanstack.com/ai) through the
`@s2-dev/resumable-stream/tanstack-ai` entrypoint.

The server helper stores TanStack `StreamChunk` events in S2. The client helper
builds a TanStack `SubscribeConnectionAdapter` so `useChat` can send messages
normally while reading assistant chunks from an S2 replay stream.

## Install

```bash theme={null}
npm install @s2-dev/resumable-stream @tanstack/ai @tanstack/ai-react @tanstack/ai-client
```

If you use TanStack's OpenAI adapter:

```bash theme={null}
npm install @tanstack/ai-openai
```

Create a basin with **Create Stream on Append** and **Create Stream on Read**
enabled.

```bash theme={null}
export S2_ACCESS_TOKEN="..."
export S2_BASIN="my-basin"
export OPENAI_API_KEY="..."
```

## Server Setup

```ts lib/s2.ts theme={null}
import { createResumableChat } from '@s2-dev/resumable-stream/tanstack-ai';

export const chat = createResumableChat({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  basin: process.env.S2_BASIN!,
  mode: 'session',
});
```

> [!TIP]
>
> Using [S2 Lite](/docs/s2-lite)? Pass `endpoints: S2Environment.parse().endpoints` (from `@s2-dev/streamstore`) to read `S2_ACCOUNT_ENDPOINT` and `S2_BASIN_ENDPOINT`; they aren't picked up automatically. See [reading endpoints from the environment](/docs/sdk/endpoints#environment-variables).

## Start A Turn

```ts app/api/chat/route.ts theme={null}
import {
  chat as tanstackChat,
  convertMessagesToModelMessages,
  type UIMessage,
} from '@tanstack/ai';
import { openaiText } from '@tanstack/ai-openai';
import { chat } from '@/lib/s2';

export async function POST(req: Request) {
  const { id, messages } = (await req.json()) as {
    id: string;
    messages: UIMessage[];
  };

  const source = tanstackChat({
    adapter: openaiText(process.env.OPENAI_MODEL ?? 'gpt-4o-mini'),
    messages: convertMessagesToModelMessages(messages),
  });

  return chat.makeResumable(`chat-${id}`, source, {
    delivery: 'replay',
    waitUntil: (p) => p.catch(console.error),
  });
}
```

## Replay Route

```ts app/api/chat/replay/route.ts theme={null}
import { chat } from '@/lib/s2';

function parseFromSeqNum(value: string | null): number | undefined {
  if (value === null) return undefined;
  const parsed = Number.parseInt(value, 10);
  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined;
}

export async function GET(req: Request) {
  const url = new URL(req.url);
  const id = url.searchParams.get('id');
  if (!id) return new Response('Missing id query parameter', { status: 400 });

  return chat.replay(`chat-${id}`, {
    fromSeqNum: parseFromSeqNum(url.searchParams.get('from')),
    live: url.searchParams.get('live') === '1',
  });
}
```

Use `live: true` for session-mode chat UIs so the replay connection stays open
at the tail and receives future turns.

## Client

```tsx app/page.tsx theme={null}
'use client';

import { createConnection } from '@s2-dev/resumable-stream/tanstack-ai/client';
import { useChat } from '@tanstack/ai-react';
import { useMemo } from 'react';

function Chat({ chatId }: { chatId: string }) {
  const connection = useMemo(
    () =>
      createConnection({
        sendUrl: '/api/chat',
        subscribeUrl: (cursor) => {
          const params = new URLSearchParams({ id: chatId, live: '1' });
          if (cursor !== undefined) params.set('from', String(cursor));
          return `/api/chat/replay?${params}`;
        },
        body: { id: chatId },
      }),
    [chatId],
  );

  const chat = useChat({
    id: chatId,
    connection,
    live: true,
  });

  // Render chat.messages and call chat.sendMessage(...)
  return null;
}
```

Every replayed chunk includes an SSE `id:` cursor. `createConnection` remembers
the most recent cursor inside that connection instance and sends it back as
`from` on the next subscribe request, so a dropped subscription can continue
after the chunks already delivered instead of rendering them twice. For a full
page refresh, pass the saved history cursor as shown below.

## Completed History

TanStack `useChat` expects completed messages as app state, so keep transcript
loading in your app:

1. Fetch history before mounting `useChat`.
2. Pass it as `initialMessages`.
3. Start replay from the history cursor so old chunks are not rendered again.

```tsx theme={null}
const history = await fetch(`/api/chat/history?id=${chatId}`).then((r) =>
  r.json(),
);

const connection = createConnection({
  sendUrl: '/api/chat',
  subscribeUrl: (cursor) => {
    const from = cursor ?? history.cursor;
    const params = new URLSearchParams({ id: chatId, live: '1' });
    if (from !== undefined) params.set('from', String(from));
    return `/api/chat/replay?${params}`;
  },
  body: { id: chatId },
});

const chat = useChat({
  id: chatId,
  connection,
  initialMessages: history.messages,
  live: true,
});
```

The runnable example stores periodic message snapshots in the same session
stream and exposes a `/history` route. That history behavior is intentionally
example code, not package API.

## Options

`createResumableChat` accepts:

| option | default | description |
| - | - | - |
| `mode` | `"single-use"` | `"single-use"` uses one stream per generation. `"shared"` reuses one active-generation stream. `"session"` appends generations to one durable stream. |
| `endpoints` | S2 defaults | Optional endpoint overrides, commonly used with S2 Lite. |
| `batchSize` | `10` | Maximum number of chunks per append batch. |
| `lingerDuration` | `50` | Maximum batching delay in milliseconds. |
| `leaseDurationMs` | `5000` | `shared` mode takeover window for stale active generations. |
| `onError` | generic message | Maps upstream errors to a TanStack `RUN_ERROR` chunk. |

`createConnection` accepts:

| option | description |
| - | - |
| `sendUrl` | POST endpoint that starts generation. |
| `subscribeUrl` | GET endpoint that returns replay SSE. String URLs get `?from=<cursor>` appended after chunks have been seen; function URLs receive the cursor. |
| `body` | Extra fields merged into the POST body, usually `{ id: chatId }`. |
| `headers` | Static or lazy headers sent on every request. |
| `credentials` | Fetch credentials mode. Defaults to `same-origin`. |
| `fetch` | Custom fetch implementation for tests or framework integrations. |
| `reconnectBackoffMs` | Millisecond backoff schedule. Defaults to `[]`, so TanStack owns subscription lifecycle. |

## Example

A complete TanStack Start chat app is available here:
[`examples/tanstack-ai-chat`](https://github.com/s2-streamstore/s2-sdk-typescript/tree/main/examples/tanstack-ai-chat).


# Terraform Provider
Source: https://s2.dev/docs/integrations/terraform-provider

Manage S2 basins, streams, and access tokens declaratively with Terraform, including managed S2 and self-hosted s2-lite.

Use the Terraform provider to manage control plane resources.

* Provider source: `s2-streamstore/s2`
* GitHub: [s2-streamstore/terraform-provider-s2](https://github.com/s2-streamstore/terraform-provider-s2)
* Registry: [registry.terraform.io/providers/s2-streamstore/s2](https://registry.terraform.io/providers/s2-streamstore/s2/latest)

The provider supports:

* **Resources**
  * `s2_basin`
  * `s2_stream`
  * `s2_access_token`
* **Data sources**
  * `s2_basin`
  * `s2_stream`
  * `s2_locations`
  * `s2_default_location`

> [!NOTE]
>
> `resource` blocks are managed by Terraform (create/update/delete).\
> `data` blocks are read-only lookups for existing resources.

## Prerequisites

* Terraform `>= 1.5.0`
* An S2 access token (`S2_ACCESS_TOKEN`)

## Quickstart

**1. Set your credentials**

```bash theme={null}
export S2_ACCESS_TOKEN="<your access token>"
# Optional (for self-hosted or s2-lite):
# export S2_ACCOUNT_ENDPOINT="a.s2.dev"
```

**2. Create a Terraform config**

```hcl theme={null}
terraform {
  required_providers {
    s2 = {
      source  = "s2-streamstore/s2"
      version = "~> 0.5"
    }
  }
}

provider "s2" {}

resource "s2_basin" "main" {
  name = "my-example-basin"
}

resource "s2_stream" "events" {
  basin = s2_basin.main.name
  name  = "events"
}

data "s2_basin" "readback" {
  name = s2_basin.main.name
}
```

**3. Initialize and apply**

```bash theme={null}
terraform init
terraform apply
```

## Authentication and Endpoint Configuration

You can configure credentials in provider config or through environment variables:

```hcl theme={null}
provider "s2" {
  # optional if S2_ACCESS_TOKEN is set
  access_token = var.s2_access_token

  # optional
  account_endpoint = var.s2_account_endpoint

  # optional
  basin_endpoint   = var.s2_basin_endpoint
}
```

Environment variables:

* `S2_ACCESS_TOKEN`
* `S2_ACCOUNT_ENDPOINT` (optional)
* `S2_BASIN_ENDPOINT` (optional, supports `{basin}` placeholder e.g. `{basin}.b.s2.dev`)

## Locations and storage classes

Use `s2_locations` to check available locations and their storage classes. Use `s2_default_location` for the account's default location.

```hcl theme={null}
data "s2_locations" "available" {}
data "s2_default_location" "current" {}

output "available_locations" {
  value = data.s2_locations.available.locations
}

output "available_storage_classes" {
  value = data.s2_default_location.current.storage_classes
}

output "default_storage_class" {
  value = data.s2_default_location.current.default_storage_class
}
```

Choose a class available in the basin's location. See [storage-class selection and defaults](/docs/concepts/configs#storage-class-selection).

## Resource Examples

### Basin with defaults

```hcl theme={null}
resource "s2_basin" "analytics" {
  name                    = "analytics-events"
  create_stream_on_append = true
  create_stream_on_read   = false

  default_stream_config {
    storage_class = "standard"

    retention_policy {
      age = 604800
    }

    timestamping {
      mode     = "client-prefer"
      uncapped = false
    }
  }
}
```

### Stream config

```hcl theme={null}
resource "s2_stream" "orders" {
  basin = s2_basin.analytics.name
  name  = "orders"

  storage_class = "express"

  retention_policy {
    age = 86400
  }

  timestamping {
    mode = "arrival"
  }
}
```

### Scoped access token

```hcl theme={null}
resource "s2_access_token" "consumer" {
  token_id = "consumer-token"

  scope = {
    basins = {
      exact = s2_basin.analytics.name
    }
    streams = {
      prefix = "orders"
    }
    ops = ["read"]
  }
}
```

> [!WARNING]
>
> The issued token value is marked as sensitive, but still lives in Terraform state.

## Import Existing Infrastructure

Import formats:

* Basin: `<basin-name>`
* Stream: `<basin-name>/<stream-name>`
* Access token: `<token-id>`

Examples:

```bash theme={null}
terraform import s2_basin.main my-basin
terraform import s2_stream.events my-basin/events
terraform import s2_access_token.consumer my-token-id
```

## Using with s2-lite

When targeting s2-lite, set the account endpoint:

```bash theme={null}
export S2_ACCOUNT_ENDPOINT="localhost:18080"
export S2_ACCESS_TOKEN="test"
```

> [!NOTE]
>
> Access-token API is not supported with s2-lite.

## Recommended Workflow

1. Pin a provider version in `required_providers`.
2. Commit `.terraform.lock.hcl`.
3. Use `terraform plan` to preview changes before applying.
4. Use `terraform import` for existing resources rather than recreating.

## Related Docs

* [Concepts: Basins](/docs/concepts/basins)
* [Concepts: Access Tokens](/docs/concepts/access-tokens)
* [API Endpoints](/docs/api/endpoints)


# Vercel AI SDK
Source: https://s2.dev/docs/integrations/vercel-ai-sdk

Persist AI SDK streams on S2 so chat generations can survive refreshes and reconnects.

S2 integrates with the [AI SDK](https://ai-sdk.dev/) through the
`@s2-dev/resumable-stream/aisdk` entrypoint.

The package stores AI SDK `UIMessageChunk` events in S2 and replays them with
the same SSE format that the AI SDK transport expects.

## Install

```bash theme={null}
npm install @s2-dev/resumable-stream ai @ai-sdk/react @ai-sdk/openai
```

This integration requires `ai >= 5.0`; older AI SDK releases do not export
`UIMessageChunk`.

Create a basin with **Create Stream on Append** enabled. If your app reads
history streams before writing to them, also enable **Create Stream on Read**.

```bash theme={null}
export S2_ACCESS_TOKEN="..."
export S2_BASIN="my-basin"
export OPENAI_API_KEY="..."
```

## Server Setup

Create the S2 chat helper once and reuse it in your routes.

```ts lib/s2.ts theme={null}
import { createResumableChat } from '@s2-dev/resumable-stream/aisdk';

export const chat = createResumableChat({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  basin: process.env.S2_BASIN!,
});
```

> [!TIP]
>
> Using [S2 Lite](/docs/s2-lite)? Pass `endpoints: S2Environment.parse().endpoints` (from `@s2-dev/streamstore`) to read `S2_ACCOUNT_ENDPOINT` and `S2_BASIN_ENDPOINT`; they aren't picked up automatically. See [reading endpoints from the environment](/docs/sdk/endpoints#environment-variables).

## Start A Turn

The POST route starts the model call, writes chunks to S2, and streams the same
chunks back to the AI SDK client.

```ts app/api/chat/route.ts theme={null}
import { openai } from '@ai-sdk/openai';
import { convertToModelMessages, streamText, type UIMessage } from 'ai';
import { after } from 'next/server';
import { chat } from '@/lib/s2';

export async function POST(req: Request) {
  const { id, messages } = (await req.json()) as {
    id: string;
    messages: UIMessage[];
  };

  const result = streamText({
    model: openai('gpt-4o-mini'),
    messages: await convertToModelMessages(messages),
  });

  return chat.makeResumable(`chat-${id}`, result.toUIMessageStream(), {
    waitUntil: (p) => after(async () => { await p; }),
  });
}
```

## Reconnect Route

`resume: true` in `useChat` calls the reconnect route on mount. With the default
AI SDK transport, that route is `${api}/${chatId}/stream`.

```ts app/api/chat/[id]/stream/route.ts theme={null}
import { chat } from '@/lib/s2';

export async function GET(
  _req: Request,
  { params }: { params: Promise<{ id: string }> },
) {
  const { id } = await params;
  return chat.replay(`chat-${id}`);
}
```

If the turn is still active, S2 replays the remaining chunks. If there is no
active turn, the route returns `204`.

## Client

Use the AI SDK `useChat` React hook to create a resumable chat.

```tsx app/page.tsx theme={null}
'use client';

import { useChat } from '@ai-sdk/react';
import { DefaultChatTransport } from 'ai';

const transport = new DefaultChatTransport({ api: '/api/chat' });

export default function Chat() {
  const { messages, sendMessage, status } = useChat({
    id: 'my-chat-id',
    transport,
    resume: true,
  });

  // Render messages and call sendMessage(...)
  return null;
}
```

## Completed History

`@s2-dev/resumable-stream/aisdk` stores the streaming chunks for resumability.
It does not decide where your completed transcript lives.

For a chat app, keep a transcript stream or database table in app code:

1. Load completed messages before rendering the page.
2. Pass them to `useChat` as initial state if your UI needs refresh-after-done
   history.
3. Use `chat.makeResumable(...)` for the active assistant turn.

The runnable example stores completed messages in a separate S2 history stream
and uses a live stream for the active generation.

## Options

`createResumableChat` accepts the shared resumable-stream options:

| option | default | description |
| - | - | - |
| `mode` | `"single-use"` | `"single-use"` uses one stream per generation. `"shared"` reuses one active-generation stream. `"session"` appends generations to one durable stream. |
| `endpoints` | S2 defaults | Optional endpoint overrides, commonly used with S2 Lite. |
| `batchSize` | `10` | Maximum number of chunks per append batch. |
| `lingerDuration` | `50` | Maximum batching delay in milliseconds. |
| `leaseDurationMs` | `5000` | `shared` mode takeover window for stale active generations. |
| `onError` | generic message | Maps upstream errors to the stored AI SDK error chunk. |

`makeResumable` accepts:

| option | default | description |
| - | - | - |
| `delivery` | `"response"` | `"response"` streams chunks on the POST response. `"replay"` returns `202` and expects the client to read from `replay`. |
| `waitUntil` | none | Keeps persistence alive after the HTTP response returns. Use `after`, Cloudflare `waitUntil`, or your server's background-task hook. |

## Example

A complete Bun server and browser client is available here:
[`examples/ai-sdk-resumable-chat`](https://github.com/s2-streamstore/s2-sdk-typescript/tree/main/examples/ai-sdk-resumable-chat).

The TypeScript SDK repo also includes direct S2 examples for:

* [Agent sessions](https://github.com/s2-streamstore/s2-sdk-typescript/blob/main/examples/ai-sdk-agent-session.ts)
* [Chat persistence](https://github.com/s2-streamstore/s2-sdk-typescript/blob/main/examples/ai-sdk-chat-persistence.ts)
* [Multi-agent dinner party](https://github.com/s2-streamstore/s2-sdk-typescript/blob/main/examples/ai-sdk-dinner-party.ts)


# Y-S2
Source: https://s2.dev/docs/integrations/y-s2

Build durable Yjs collaborative editing with y-s2, using S2 streams for updates and Cloudflare R2 for snapshots.

[Y-S2](https://github.com/s2-streamstore/y-s2) is a Cloudflare Worker that serves [Yjs](https://yjs.dev) documents over WebSockets, using S2 as the durable, ordered log of each room's updates and an R2 bucket for snapshots. Any number of Worker instances can serve the same room: they coordinate only through the room's S2 stream, so there is no central server to run or scale.

## How it works

* **Rooms** — each room is a stream named `rooms/<room>/index` in your basin, created on first use.
* **Writes** — document updates and presence (cursor) updates from a client are batched and appended to the room's stream. Messages larger than S2's 1 MiB [record limit](/docs/platform/limits#records), such as a large paste, are stored in R2 and the stream carries a pointer to them.
* **Fan-out** — every connection reads the room's stream and relays new records to its client.
* **Joining** — a new connection loads the room's latest snapshot from R2, reads the stream from where that snapshot ends, and sends the client the full document once it has caught up.
* **Snapshots** — while a room is active, connections periodically write a new snapshot to R2 and then [trim](/docs/concepts/trimming) the stream up to it. A lease built on [fencing tokens](/docs/concepts/concurrency-control) ensures only one connection snapshots a room at a time.

See [Behind y-s2: serverless multiplayer rooms](https://s2.dev/blog/durable-yjs-rooms) for a deeper walkthrough.

## Prerequisites

* A Cloudflare account on the **Workers Paid** plan. Each WebSocket connection is a single Worker invocation, and every S2 and R2 call it makes counts toward the per-invocation subrequest limit, which is 50 on the free plan.
* An R2 bucket for snapshots and large messages.
* An S2 basin with **Create stream on append** and **Create stream on read** enabled, and a scoped [access token](/docs/concepts/access-tokens) for it.
* Node.js 22 or newer.

> [!WARNING]
>
> Set the basin's default stream retention to **infinite**, or at least longer than a room may go without edits. Snapshots are only taken while a room is active, so updates that expire from the stream before a snapshot covers them are lost. Infinite retention requires a payment method on your S2 account (see [limits](/docs/platform/limits#retention)).

## Deploy

1. Clone the repository and install dependencies:

   ```bash theme={null}
   git clone https://github.com/s2-streamstore/y-s2
   cd y-s2
   npm install
   ```

2. Point the `R2_BUCKET` binding in `wrangler.jsonc` at your bucket by setting its `bucket_name`.

3. Store your S2 credentials as Worker secrets:

   ```bash theme={null}
   npx wrangler secret put S2_ACCESS_TOKEN
   npx wrangler secret put S2_BASIN
   ```

4. Deploy:

   ```bash theme={null}
   npm run deploy
   ```

For local development, put `S2_ACCESS_TOKEN` and `S2_BASIN` in a `.dev.vars` file and run `npm run dev`. The repository also includes a GitHub Actions workflow that checks and deploys every push to `main`; see its [README](https://github.com/s2-streamstore/y-s2#continuous-deployment) for the secrets it expects.

## Configuration

These optional variables can be set under `vars` in `wrangler.jsonc` (or in `.dev.vars` locally):

| Variable | Default | Description |
| - | - | - |
| `LOG_MODE` | `CONSOLE` | `CONSOLE` logs to the Worker console. `S2_SINGLE` logs each connection to its own stream (`logs/worker-<cf-ray>`), and `S2_SHARED` logs all connections to `logs/workers-shared`, in the same basin. Case-insensitive. |
| `SNAPSHOT_BACKLOG_SIZE` | `400` | Number of records since the last snapshot that triggers a new one. |
| `SNAPSHOT_BACKLOG_BYTES` | `8388608` | Size in bytes of the records since the last snapshot that triggers a new one. |
| `BACKLOG_BUFFER_AGE` | `60000` | Milliseconds after the oldest record not yet in a snapshot before a new one is taken. |
| `LEASE_DURATION` | `30` | Seconds a connection holds the snapshot lease before others may take it over. |
| `S2_BATCH_SIZE` | `8` | Maximum number of records per append batch. |
| `S2_LINGER_TIME` | `50` | Milliseconds to wait for more records before sending an append batch. |

## Client integration

Connect with the [y-websocket](https://github.com/yjs/y-websocket) provider. The server URL ends in `/ws`, and the room is passed both as the room name and as the `room` parameter:

```javascript theme={null}
import * as Y from 'yjs'
import { WebsocketProvider } from 'y-websocket'

const room = 'room-name'
const doc = new Y.Doc()
const provider = new WebsocketProvider(
  'wss://your-worker.your-subdomain.workers.dev/ws',
  room,
  doc,
  {
    params: { room, yauth: 'your-auth-token' }
  }
)
```

Presence works through the provider's `awareness`, as with any y-websocket server.

> [!WARNING]
>
> Y-S2 does not authenticate clients yet: any non-empty `yauth` value is accepted, and `/auth/token` returns a fixed demo token. Add your own authorization before using it for anything but demos. Anyone who can reach the Worker can read and write any room, and create new rooms in your basin.

## Resources

* [Demo](https://s2.dev/demos/y-s2)
* [GitHub repository](https://github.com/s2-streamstore/y-s2)
* [Behind y-s2: serverless multiplayer rooms](https://s2.dev/blog/durable-yjs-rooms)
* [Yjs documentation](https://docs.yjs.dev)


# Introduction
Source: https://s2.dev/docs/intro

S2 is the API for unlimited, durable, real-time streams.

Stream Store gives you any number of durable, ordered [streams](/docs/concepts/streams) of [records](/docs/concepts/records) that you can [append](/docs/concepts/appends) to and [read](/docs/concepts/reads) from any position.

## The idea

Object storage made data at rest simple: scale, usage-based pricing, granular access control, and no clusters to operate.

S2 brings the same experience to data in motion. Create a durable stream for every [agent session](https://s2.dev/blog/agent-sessions), user journey, or parallel exploration.

Each S2 stream has its own URL. It scales from occasional writes to high-throughput workloads, supports many concurrent live readers, and keeps data replayable in bottomless object storage.

Access streams directly [over HTTP](/docs/api/protocol#data-plane), or use the [SDKs](/docs/sdk), [CLI](/docs/cli), and web-based [Studio](/docs/studio) for higher-level tools.

## Get started

Head to the [quickstart](/docs/quickstart) to get up and running.

You can also try the [playground](https://s2.dev/playground) to explore without signing up.

> [!TIP]
>
> Agent-assisted development works best when your agent can consult S2's [`llms.txt`](https://s2.dev/docs/llms.txt).


# Cachey
Source: https://s2.dev/docs/oss-changelogs/cachey



# CLI
Source: https://s2.dev/docs/oss-changelogs/cli



# Go
Source: https://s2.dev/docs/oss-changelogs/go-sdk



# Python
Source: https://s2.dev/docs/oss-changelogs/python-sdk



# Resumable Stream
Source: https://s2.dev/docs/oss-changelogs/resumable-stream



# Rust
Source: https://s2.dev/docs/oss-changelogs/rust-sdk



# s2-lite
Source: https://s2.dev/docs/oss-changelogs/s2-lite



# Terraform Provider
Source: https://s2.dev/docs/oss-changelogs/terraform-provider



# TypeScript
Source: https://s2.dev/docs/oss-changelogs/typescript-sdk



# Architecture
Source: https://s2.dev/docs/platform/architecture

S2's cellular cloud architecture delivers elastic scale, durable writes, and low-latency reads without operational overhead.

s2.dev is organized into [cells](/docs/platform/cells): isolated deployments located in a specific cloud region.

You don't need to know any of this to use the service.

![Architecture of a cell](https://mintcdn.com/streamstore/h-_jIRPPBcasd4L1/images/cell.png?fit=max\&auto=format\&n=h-_jIRPPBcasd4L1\&q=85\&s=8d49840c7789ce481b34d242d6be5ded)

> [card] Control plane
>
> * `cop` ("channel operator") ensures resilience and elasticity of stateful pieces:
>   * supervises and creates/destroys channels
>   * rebalances load by migrating streams
> * `bgtask` ("background task") workers partitioned on role and keyspace handle async jobs, such as garbage collection.

> [card] Data plane
>
> * `frontend` replicas are an elastic deployment that fronts all API requests, acting as a gateway layer.
> * `cachey` replicas form an elastic read-through cache per zone for all reads from object storage.
> * `channel` pods sequence writes with a WAL maintained as sequential *chunks*, and flush batched stream *segments* accumulated over a longer timeframe.

Our read-through cache, [cachey](https://github.com/s2-streamstore/cachey), is an open source project.

## Write path

The `frontend` talks to the assigned `channel` for the stream to append records.

If necessary it will first activate the stream by assigning to a `channel` using [power of 2 choices](https://www.eecs.harvard.edu/~michaelm/postscripts/handbook2001.pdf) for load balancing.

Channels multiplex stream records in chunks on object storage (effectively a [WAL](https://en.wikipedia.org/wiki/Write-ahead_logging)), flushing based on a combination of elapsed time and accumulated size. Appends are only acknowledged when associated records are durable in a chunk.

Over a longer timeframe, channels batch records from the WAL into per-stream segments and update stream metadata to reference them. This keeps the write acknowledgement path short while giving the read path larger, stream-local object byte ranges to scan.

> [!NOTE]
>
> Once a stream is active, metadata storage is not in the hot-path of writes — only object storage.

## Read path

The `frontend` uses stream metadata to generate a concatenated iteration over relevant stream segments. This is further chained with pointers into chunks (the channel WAL), if the client is reading recently-written data.

It fetches relevant object byte ranges via a `cachey` pod in its own zone, using rendezvous hashing with bounded load to optimize hit rate without hotspotting.

Concurrent reads are combined wherever possible:

* Tailing readers are coalesced on the `frontend` as well as `channel`
* `cachey` will coalesce reads for the same "page" of an object

> [!TIP]
>
> s2.dev effectively implements a purpose-built CDN for streaming data.

## Stream lifecycle

Streams are made up of a chain of *streamlets*. There is usually only one, but multiple may exist when the stream is being migrated between channels, or migration(s) recently completed.

Streamlets take up a tiny amount of memory on channels — and only stick around if there have been recent writes.

Dormant streams live only as metadata and segment data.

## Storage classes

The choice of storage class affects latency of appending and reading recently-written records.

In AWS,

* `Standard` is powered by channels operating against an S3 Standard chunk storage bucket, which guarantees writes are durable in multiple zones.
* `Express` is powered by channels operating against a quorum of *three* S3 Express zonal chunk storage buckets — so writes are still guaranteed to be durable in multiple zones, but with much lower latency.

## Quality of service

Admission control and load balancing logic is embedded into components in the online request path. Auto-scaling and stream migrations handle sustained shifts.

To optimize throughput, the data plane employs pipelining, i.e. multiple in-flight operations with an ordering dependency. This is not just a server-side concern; S2 client SDKs will also pipeline when using the append and read [session](/docs/api/protocol#sessions) APIs.

Stream migrations use a speculation-based protocol that keeps the write pipeline flowing to minimize disruption.

## Correctness

Beyond unit and integration tests, we practice [deterministic simulation testing](https://s2.dev/blog/dst). S2 involves many moving parts, and investing in DST allows us to test the invariants the system must always uphold.

Deterministic testing lets us run the processes that comprise S2 over a simulated network. We can inject faults, crash processes randomly, trigger recovery procedures, and check that we do not violate our commitments to durability and strong consistency.

Our setup includes a custom Rust framework built with [Turmoil](https://github.com/tokio-rs/turmoil), which we run in CI/CD and in larger nightly searches for subtle regressions. We also use [Antithesis](https://antithesis.com/) as a second form of DST, and verify linearizability with the [Porcupine model checker](https://s2.dev/blog/linearizability).

## Local

> [card] [s2-lite](/docs/s2-lite#cloud-service-comparison)
>
> The open source implementation of S2 has a distinct architecture optimized for ease of deployment as a single binary. See how it compares with the cloud service.


# Cells
Source: https://s2.dev/docs/platform/cells

S2 cells isolate compute, storage, and networking by provider and region to define fault domains and support tenancy and scaling.

A cell is an isolated deployment of S2 within a specific [**location**](/docs/concepts/basins#location), which is the user-visible combination of cloud provider and region — for example, `aws:us-east-1`.

Each cell has its own compute, storage, and networking resources; see [Architecture](/docs/platform/architecture) for internals.

An S2 **account**, i.e. your [organization](https://s2.dev/dashboard), has exactly one home cell per location.

When you create a basin, you choose from the available locations. The basin [endpoint](/docs/api/endpoints) routes to the right cell via DNS.

> [!TIP]
>
> Pick a location that most closely matches your workloads when creating a basin, for the lowest latencies.

## Tenancy

Cells are multi-tenant by default, which is how S2 offers purely usage-based [pricing](https://s2.dev/pricing). Idle accounts can scale to zero, while active workloads can burst into pooled capacity.

Dedicated cells are available with a minimum commitment. They are useful when you want stronger isolation than the default multi-tenant fleet — [contact us](mailto:hi@s2.dev).

> [!TIP]
>
> [Private networking](/docs/platform/private-networking) works with both multi-tenant and dedicated cells.


# Limits
Source: https://s2.dev/docs/platform/limits

Check S2 limits for resources, records, throughput, retention, and reconfiguration.

> [!NOTE]
>
> We can increase certain limits if [feedback](mailto:support@s2.dev) suggests they are too restrictive.

## Basins

* Basins are unlimited in number for all accounts with a payment method. Free tier accounts are limited to 100.

* Basin names must be globally unique. The name must be between 8 and 48 characters, comprising lowercase letters, numbers and hyphens. It cannot begin or end with a hyphen.

## Streams

* Streams are unlimited in number.

* Stream names can be an arbitrary string up to 512 characters. They may be hierarchically named (e.g. using a delimiter like `/`) for namespacing — you are able to list streams with prefix filtering.

  > [!TIP]
  >
  > Many use cases around lightweight streams find it useful to configure `create stream on append` / `create stream on read` on the basin, and set a `delete on empty` time threshold for the default stream configuration.

## Records

* Each record may be up to 1 MiB in metered bytes.
* Batches may contain up to 1000 records, and cumulatively total up to 1 MiB in metered bytes.

S2 uses the following formula for metering record size:

```text theme={null}
8 
+ 2 bytes per header
+ size of all header names and values
+ size of the body
```

## Appends

* 100 MiBps per stream.
* 200 batches per second per stream *per client connection*. When exceeded,
  * `Append` requests will receive an HTTP 429.
  * `AppendSession` will experience throttling.

## Reads

Reads are highly scalable on both throughput and concurrency, due to S2's decoupled architecture and read-through caching.

> [!NOTE]
>
> The system is designed to be elastic, but there are bounds to how fast it can scale. The precise limit that will be enforced is TBD.

## Retention

Records may stay visible for a brief period after their retention window expires, typically on the order of minutes.

> [!NOTE]
>
> On the free tier, streams can retain data for 28 days or less. If you want a longer (or infinite) retention, add a payment method in the [Dashboard](http://s2.dev/dashboard).

## Reconfiguration

**Storage class** changes on an existing stream (i.e. from `Standard` to `Express`, or vice versa) typically take effect after the stream has been inactive for approximately 5 to 25 minutes. This will become a live migration in the future.


# Pricing
Source: https://s2.dev/docs/platform/pricing

S2 usage-based pricing model

# Private Networking
Source: https://s2.dev/docs/platform/private-networking

Set up AWS PrivateLink to connect an AWS VPC to S2 privately and keep traffic off the public internet.

AWS PrivateLink allows you to connect to S2 privately from your VPC, without exposing traffic to the public internet. You also benefit from significantly [reduced egress costs](https://s2.dev/pricing).

> [!NOTE]
>
> S2 is currently available in AWS regions, so private networking uses AWS PrivateLink. Equivalent options, such as GCP Private Service Connect, will be added as S2 expands to other clouds.

## Setup

**1. Get Service Name from S2 Dashboard**

Go to the S2 dashboard, in the Basins tab, expand the **Private Connectivity** section to find your PrivateLink service name.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/s2-dashboard.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=ccd40214b2d678da4aa7f1f77d0bad9a)

**2. Navigate to VPC Endpoints**

Search for VPC endpoint on the AWS console and select **"Endpoints"**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-1.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=5f28e375ff848cb67953c111c6fedb9a)

**3. Create Endpoint**

Click on **"Create Endpoint"**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-2.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=2ac969f1ae0f77483e5095f31e384a16)

**4. Select Service Type**

Select **"PrivateLink Ready Partner Services"**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-3.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=b8cf6d3956eb72143313fd6d0d03f7ec)

**5. Verify Service**

Enter the PrivateLink service name from the S2 dashboard and click on **"Verify Service"**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-4.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=9fad871f56db418dd148829f11ab50a7)

**6. Select VPC**

Select the appropriate **VPC**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-5.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=215f0b7924565d88cb59ffd5d7b95e8a)

**7. Enable Private DNS**

Enable private DNS name.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-6.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=751abaac651a014c83086c77fed23110)

**8. Include Subnets**

Include subnets.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-7.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=9c8173f52c651f2685952b52c7bd57bb)

**9. Configure Security Groups**

Select appropriate **Security Groups**.

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-8.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=a9b2b97326d876fbe9e29db9bcc4d787)

![](https://mintcdn.com/streamstore/U551CY5wtdgysPG-/images/privatelink/step-9.png?fit=max\&auto=format\&n=U551CY5wtdgysPG-\&q=85\&s=c291edff7000df36d4969e2e925155d4)

> [!NOTE]
>
> HTTPS (port 443) must be allowed in both inbound and outbound security group rules.
> Access can be left open to `0.0.0.0/0` or, preferably, restricted to the PrivateLink
> endpoint's private IP address or CIDR range.

## Validate Your Connection

1. From a VM on your VPC, try resolving an S2 basin DNS record:

   ```bash theme={null}
   nslookup ${basin}.b.s2.dev
   ```

   You may need to clear caches or wait a few moments for this to take effect.

2. If everything has worked, this DNS record should resolve to a private IPv4 address (e.g., starting with `10.*`).


# Security
Source: https://s2.dev/docs/platform/security

How we secure your data

## Encryption

> [card] Data in Transit
>
> S2 [endpoints](/docs/api/endpoints) are secured by TLS, and S2 uses TLS for service-to-service traffic internally as well.

> [card] Data at Rest
>
> Data at rest is encrypted by the cloud systems S2 relies on.

> [card] [Client-supplied Encryption Keys](/docs/concepts/encryption)
>
> Configure a basin with a cipher and provide the encryption key for reads and writes at request time.

> [card] Client-side Encryption
>
> Get end-to-end protection by encrypting and decrypting on the client-side.

## Compliance

Email [trust@s2.dev](mailto:trust@s2.dev) to request access to our **SOC 2** report, for **HIPAA** BAA, or **GDPR** DPA.

## Responsible Disclosure

Ethical hackers and security researchers can report vulnerabilities to us at [security@s2.dev](mailto:security@s2.dev).

## See also

> [card] [Cells](/docs/platform/cells)
>
> Tenancy options

> [card] [Private networking](/docs/platform/private-networking)
>
> Avoiding the internet


# Status
Source: https://s2.dev/docs/platform/status

S2 service status and availability

# Quickstart
Source: https://s2.dev/docs/quickstart

Create an S2 account, install the CLI, and start appending and reading streams right away.

## Onboarding

**First**, log in to the [dashboard](https://s2.dev/dashboard) and create a new organization.

## Get started with the CLI

**1. Install the S2 CLI**

> [tab] Using Homebrew
>
> This method works on macOS and Linux distributions with
> [Homebrew](https://brew.sh) installed.
>
> Homebrew 6.0 and later require explicit [tap trust](https://docs.brew.sh/Tap-Trust) for third-party taps. Installing with the fully qualified name trusts only the S2 formula:
>
> ```bash theme={null}
> brew install s2-streamstore/s2/s2
> ```
>
> > [!NOTE]
> >
> > The fully qualified name (`s2-streamstore/s2/s2`) scopes trust to that formula only. To trust every current and future package in the tap instead, run `brew trust s2-streamstore/s2` before installing.

> [tab] Using Cargo
>
> This method works on any system with [Rust](https://www.rust-lang.org/)
> and [Cargo](https://doc.rust-lang.org/cargo/) installed.
>
> ```bash theme={null}
> cargo install --locked s2-cli
> ```

> [tab] From Release Binaries
>
> Check out the [S2 CLI Releases](https://github.com/s2-streamstore/s2/releases?q=s2-cli\&expanded=true)
> for prebuilt binaries for many different architectures and operating systems.
>
> Linux and macOS users can download the release binary using:
>
> ```bash theme={null}
> curl -fsSL s2.dev/install.sh | bash
> ```
>
> > [!NOTE]
> >
> > The script will install the binary in `~/.s2/bin` and modify
> > the shell profile file to add it to the path. Make sure to
> > reload the shell or open up a new terminal after completing
> > the installation.

The binary should be installed as `s2`.

**2. Authenticate the CLI**

> [tab] Browser login
>
> ```bash theme={null}
> s2 login
> ```

> [tab] Access token
>
> Issue an access token from the [dashboard](https://s2.dev/dashboard), then store it with:
>
> ```bash theme={null}
> s2 auth access-token set
> ```

**3. Create a new basin**

> [!TIP]
>
> Basin names must be globally unique. Try creating a basin with your
> GitHub handle, something like `username-first-basin`.

> [!NOTE]
>
> A basin name must be between 8 and 48 characters long and comprise
> lowercase letters, numbers and hyphens. It cannot begin or end with a
> hyphen.

```bash theme={null}
export BASIN="YOUR_BASIN_NAME"
s2 create-basin ${BASIN}
```

Let's double-check that the new basin is active. You should see its name in the list with an `active` status:

```bash theme={null}
s2 list-basins
```

**4. Grab your popcorn!**

Nothing better than Star Wars to give us a good sense of how we can exercise `append` and `read` sessions, and see what **data in motion** looks like.

We will start by creating a new stream called `starwars`.

```bash theme={null}
s2 create-stream s2://${BASIN}/starwars
```

Some of you may have watched the movie in ASCII over telnet from `towel.blinkenlights.nl`!

We will do the same, except we will "append" the movie to our `starwars` stream while we also tail it.

Start by creating a read session like so:

```bash theme={null}
s2 read s2://${BASIN}/starwars 2> /dev/null
```

Open another terminal and start an append session:

```bash theme={null}
export BASIN="YOUR_BASIN_NAME" # Set the basin name for new terminal
nc starwars.s2.dev 23 | s2 append s2://${BASIN}/starwars
```

Go back to the previous terminal with the read session to enjoy the movie:

## What You Just Used

You just created a [basin](/docs/concepts/basins), wrote to a [stream](/docs/concepts/streams), and moved [records](/docs/concepts/records) with [appends](/docs/concepts/appends) and [reads](/docs/concepts/reads).

> [card] [Explore S2 concepts](/docs/concepts/records)
>
> Start with records, then follow the model through streams, basins, appends, and reads.

> [card] [Try more demos](/docs/demos)
>
> Explore the playground, resumable AI chat, live sandbox logs, and collaborative editing.


# s2-lite
Source: https://s2.dev/docs/s2-lite

s2-lite is an open source, self-hosted server implementing S2 APIs as a single binary powered by SlateDB for durable storage.

[ s2-lite](https://github.com/s2-streamstore/s2) is designed to run as a single instance, making self-hosting straightforward.

See the [README](https://github.com/s2-streamstore/s2?tab=readme-ov-file#design) for more on its internals and quickstart for running it via the `s2` CLI.

[ Docker images](https://github.com/s2-streamstore/s2/pkgs/container/s2) are available. You can also use the [Helm chart](https://github.com/s2-streamstore/s2/blob/main/charts/s2-lite-helm/README.md) for deploying to Kubernetes.

## Run it locally

With the [CLI](/docs/cli/installation) installed:

```bash theme={null}
s2 lite --port 8080
```

This starts an in-memory server and prints the environment variables to point clients at it:

```bash theme={null}
export S2_ACCOUNT_ENDPOINT="http://localhost:8080"
export S2_BASIN_ENDPOINT="http://localhost:8080"
export S2_ACCESS_TOKEN="ignored"
```

The CLI picks these up automatically. SDKs need you to use the associated helper for reading from the environment; see [SDK: Endpoints](/docs/sdk/endpoints#environment-variables). See [CLI: Configuration](/docs/cli/configuration) for more on the CLI.

It also serves a web dashboard at `http://localhost:8080/dashboard` for browsing basins and streams, and using [Studio](/docs/studio).

> [!TIP]
>
> Even if you plan to exclusively use the cloud service, s2-lite is useful as an S2 emulator for integration tests and CI/CD pipelines. This way your tests run against a real S2 instance without needing network access or credentials.

## Cloud service comparison

Start with what's the same. Both s2.dev and s2-lite implement the core [S2 API](/docs/api) — the [OpenAPI spec](https://github.com/s2-streamstore/s2-specs/blob/main/s2/v1/openapi.json) is in fact generated from s2-lite — and the same [SDKs](/docs/sdk/languages), [CLI](/docs/cli), and [Studio](/docs/studio) work with either.

*For s2-lite, this comparison assumes object storage mode, although local disk and in-memory are also supported.*

| | s2.dev | s2-lite |
| - | - | - |
| **Architecture** | [Elastic](/docs/platform/architecture) — components scale out, and streams migrate to balance load | Single node that scales vertically, which can take you surprisingly far |
| **Storage engine** | Purpose-built: low-latency [WAL with batched segments](/docs/platform/architecture#write-path) on object storage | [SlateDB](https://slatedb.io) on object storage |
| **Append latency** | Within 40 ms on `express` and 400 ms on `standard` [storage class](/docs/platform/architecture#storage-classes) | Your bucket's PUT latency plus tunable [flush interval](https://github.com/s2-streamstore/s2?tab=readme-ov-file#slatedb-settings); typically a few hundred ms |
| **Durability** | Multi-AZ before every ack, even on `express` | Inherits your bucket's durability, typically multi-AZ in the cloud |
| **Read fan-out** | Elastic read-through [caching](/docs/platform/architecture#read-path) — effectively a CDN for streams | Bounded by one node — layer your own caching on top for broadcast scale |
| **Security** | SOC 2, HIPAA, and GDPR [compliant](/docs/platform/security#compliance); [private networking](/docs/platform/private-networking) and [dedicated cells](/docs/platform/cells#tenancy) available | Your perimeter end-to-end — data touches only infrastructure you control |
| **Operations** | None — serverless and scale-to-zero, with usage-based [pricing](https://s2.dev/pricing) | Single binary and a bucket: yours to provision, monitor, and scale |
| **Auth** | Granular [access control](/docs/concepts/access-tokens) | Gate with your own auth layer — access token API support is planned |
| **Open source** | Managed service; MIT-licensed API layer and ecosystem projects | MIT-licensed server implementation |
| **Maturity** | [GA](https://s2.dev/blog/ga) with a 99.99% availability SLO | Newer, though shares code with the cloud service, and builds on top of SlateDB |

**Latency** is mostly about proximity. With the cloud service it depends on how close your workload runs to your basin's [location](/docs/concepts/basins#location); with s2-lite you control placement, so colocating it and its bucket with your workload can win where S2 does not have a nearby location yet. At a comparable distance, the `express` storage class delivers multi-AZ durability with tens-of-milliseconds acks — a combination that is hard to match when flushing to a single bucket. Like the cloud service, s2-lite also pipelines appends, so throughput holds up well even against high-latency object storage, and [`s2 bench`](/docs/cli/bench) makes it easy to measure your own numbers.

**Cost** follows from operations. The cloud service is purely usage-based and scales to zero — there is no instance to keep warm, which for spiky or modest workloads usually makes it cheaper too — and there are no knobs to think about: storage class is the only latency lever, with durability never in question. With s2-lite you pay for the node and the bucket, and the flush interval ties latency to cost: flushing more aggressively acknowledges appends sooner but increases billable object storage requests.

Reach for the cloud service when you want a production stream store without operating one: elasticity, high fan-out, the lowest acknowledgment latencies, granular access control. Reach for s2-lite when self-hosting is the point — keeping data in your own bucket and network, running where the cloud service does not (an edge site, an air-gapped environment, a location S2 does not offer yet), or as a zero-dependency emulator in dev and CI.

Whichever you start with, the shared API means your application can switch with an [endpoint](/docs/sdk/endpoints) override rather than a rewrite.

## Init file

s2-lite can pre-create basins and streams at startup from a JSON spec file using the same format used by [`s2 apply`](/docs/cli/apply).

```bash theme={null}
s2 lite --init-file init.json
```

Or set the environment variable:

```bash theme={null}
S2LITE_INIT_FILE=init.json s2 lite
```

The spec is applied with create-or-reconfigure semantics. Resources that already exist are reconfigured to match the spec, and only the fields present in the spec are updated.

Example `init.json`:

```json theme={null}
{
  "$schema": "https://raw.githubusercontent.com/s2-streamstore/s2/main/cli/schema.json",
  "basins": [
    {
      "name": "my-basin",
      "config": {
        "create_stream_on_append": true
      },
      "streams": [
        { "name": "events" }
      ]
    }
  ]
}
```

See [CLI: Apply](/docs/cli/apply) for the full spec file format reference.


# Access Tokens
Source: https://s2.dev/docs/sdk/access-tokens

Manage access tokens with S2 SDKs: list existing tokens, issue scoped tokens, and revoke tokens.

Access tokens control what a client can do. Each token has a scope that restricts which basins and streams it can access, and what operations it can perform.

See the [access tokens guide](/docs/concepts/access-tokens) for the full specification, including narrowing rules for issued tokens, operation groups, granular permissions, and auto-prefixing for multi-tenant setups.

> [tab] TypeScript
>
> ```typescript theme={null}
> // List tokens (returns metadata, not the secret)
> const tokens = await client.accessTokens.list();
>
> // Issue a token scoped to streams under "users/1234/"
> const { accessToken: issuedToken } = await client.accessTokens.issue({
> 	id: "user-1234-rw-token",
> 	scope: {
> 		basins: { prefix: "" }, // all basins
> 		streams: { prefix: "users/1234/" },
> 		opGroups: { stream: { read: true, write: true } },
> 	},
> 	expiresAt: new Date("2027-01-01"),
> });
>
> // Revoke a token
> await client.accessTokens.revoke({ id: "user-1234-rw-token" });
> ```

> [tab] Python
>
> ```python theme={null}
> # List tokens (returns metadata, not the secret)
> tokens = await client.list_access_tokens()
>
> # Issue a token scoped to streams under "users/1234/"
> issued_token = await client.issue_access_token(
>     "user-1234-rw-token",
>     scope=AccessTokenScope(
>         basins=PrefixMatch(""),  # all basins
>         streams=PrefixMatch("users/1234/"),
>         op_groups=OperationGroupPermissions(
>             stream=Permission.READ_WRITE,
>         ),
>     ),
>     expires_at=datetime(2027, 1, 1, tzinfo=timezone.utc),
> )
>
> # Revoke a token
> await client.revoke_access_token("user-1234-rw-token")
> ```

> [tab] Go
>
> ```go theme={null}
> // List tokens (returns metadata, not the secret)
> tokens, _ := client.AccessTokens.List(ctx, nil)
>
> // Issue a token scoped to streams under "users/1234/"
> expires := time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC)
> result, _ := client.AccessTokens.Issue(ctx, s2.IssueAccessTokenArgs{
> 	ID: "user-1234-rw-token",
> 	Scope: s2.AccessTokenScope{
> 		Basins:  &s2.ResourceSet{Prefix: s2.String("")}, // all basins
> 		Streams: &s2.ResourceSet{Prefix: s2.String("users/1234/")},
> 		OpGroups: &s2.PermittedOperationGroups{
> 			Stream: &s2.ReadWritePermissions{Read: true, Write: true},
> 		},
> 	},
> 	ExpiresAt: &expires,
> })
>
> // Revoke a token
> client.AccessTokens.Revoke(ctx, s2.RevokeAccessTokenArgs{ID: "user-1234-rw-token"})
> ```

> [tab] Rust
>
> ```rust theme={null}
> // List tokens (returns metadata, not the secret)
> let tokens = client.list_access_tokens(Default::default()).await?;
>
> // Issue a token scoped to streams under "users/1234/"
> let result = client
>     .issue_access_token(
>         IssueAccessTokenInput::new(
>             "user-1234-rw-token".parse()?,
>             AccessTokenScopeInput::from_op_group_perms(
>                 OperationGroupPermissions::new()
>                     .with_stream(ReadWritePermissions::read_write()),
>             )
>             .with_basins(BasinMatcher::Prefix("".parse()?)) // all basins
>             .with_streams(StreamMatcher::Prefix("users/1234/".parse()?)),
>         )
>         .with_expires_at("2027-01-01T00:00:00Z".parse()?),
>     )
>     .await?;
>
> // Revoke a token
> client
>     .revoke_access_token("user-1234-rw-token".parse()?)
>     .await?;
> ```


# Appending
Source: https://s2.dev/docs/sdk/appending

Append records with S2 SDKs using single-batch appends, pipelined append sessions, or the Producer API.

SDKs expose three ways to append records:

* `append` for a single atomic batch.
* `appendSession` for ordered, pipelined batches on a persistent connection.
* `Producer` for submitting individual records with per-record tickets.

Batches are the atomic unit of appending. A batch can contain up to 1000 records or 1 MiB of data. Each append — whether a single call or part of a session — writes exactly one batch.

For anything beyond simple one-off writes, use an **append session**. Sessions enable pipelining (multiple batches in flight), maintain strict ordering, and gracefully handle throttling. The [Producer API](#producer) builds on sessions to provide per-record semantics.

> [!TIP]
>
> For high throughput, send batches approaching the 1000-record or 1 MiB limit. Each client is limited to 200 append batches per second, so small batches hit the batch-rate limit sooner. Single-batch callers over the limit may receive a 429 with `Retry-After` info; append sessions and producers use backpressure to match S2's capacity.

> [!NOTE]
>
> Appends also support match sequence numbers and fencing tokens. See [concurrency control](/docs/concepts/concurrency-control) for the write coordination model.

## Single-batch append

A single batch of records can be appended by calling `append` on a stream client:

> [tab] TypeScript
>
> ```typescript theme={null}
> const stream = basin.stream(streamName);
> const ack = await stream.append(
> 	AppendInput.create([
> 		AppendRecord.string({ body: "first event" }),
> 		AppendRecord.string({ body: "second event" }),
> 	]),
> );
>
> // ack tells us where the records landed
> console.log(`Wrote records ${ack.start.seqNum} through ${ack.end.seqNum - 1}`);
> ```

> [tab] Python
>
> ```python theme={null}
> stream = basin.stream(stream_name)
> ack = await stream.append(
>     AppendInput(
>         records=[
>             Record(body=b"first event"),
>             Record(body=b"second event"),
>         ]
>     )
> )
>
> # ack tells us where the records landed
> print(f"Wrote records {ack.start.seq_num} through {ack.end.seq_num - 1}")
> ```

> [tab] Go
>
> ```go theme={null}
> ack, _ := stream.Append(ctx, &s2.AppendInput{
> 	Records: []s2.AppendRecord{
> 		{Body: []byte("first event")},
> 		{Body: []byte("second event")},
> 	},
> })
>
> // ack tells us where the records landed
> fmt.Printf("Wrote records %d through %d\n", ack.Start.SeqNum, ack.End.SeqNum-1)
> ```

> [tab] Rust
>
> ```rust theme={null}
> let stream = basin.stream(stream_name.clone());
>
> let ack = stream
>     .append(AppendInput::new(AppendRecordBatch::try_from_iter([
>         AppendRecord::new("first event")?,
>         AppendRecord::new("second event")?,
>     ])?))
>     .await?;
>
> // ack tells us where the records landed
> println!(
>     "Wrote records {} through {}",
>     ack.start.seq_num,
>     ack.end.seq_num - 1
> );
> ```

This works well for simple cases, but each append is a separate HTTP request.

For higher throughput, and guaranteeing ordering across batches, use an **append session** or the **Producer API**.

## Append session

An append session maintains a bidirectional stream with S2: you send batches of records, and S2 sends back acknowledgements.

> [!NOTE]
>
> Under the hood, sessions use [S2S](/docs/api/protocol#sessions), a minimal binary framing layer over HTTP/2. In environments without HTTP/2 (notably browsers), the TypeScript SDK falls back to HTTP/1.1. You get the same session APIs, but append batches cannot be pipelined, so there will only be one in-flight batch at a time.

This enables [pipelining](https://en.wikipedia.org/wiki/Protocol_pipelining) — you can have multiple batches in flight simultaneously, dramatically improving throughput compared to waiting for each append to complete before sending the next, while still maintaining the ordering of records across batches.

Contrast this with multiple concurrent single-batch appends; while this would also allow high throughput, each concurrent append would be independent, and the ordering in which concurrent batches become durable would not be guaranteed.

A session provides a stateful handle on a stream that is being appended to. Batches can be appended to that session by calling `submit()`.

Submitting a batch is an async function. It resolves to a `BatchSubmitTicket` when the batch is accepted by the session; this async function can exhibit backpressure, to prevent overwhelming the session.

A ticket tracks an append while it is pending. The batch is only durable once it has been acknowledged by S2. This can be awaited via the ticket's `ack()` method, which resolves only when the written batch is fully durable on object storage. The resulting `AppendAck` contains information about the batch's resulting position in the stream.

> [tab] TypeScript
>
> ```typescript theme={null}
> const session = await stream.appendSession();
>
> // Submit a batch - this enqueues it and returns a ticket
> const ticket = await session.submit(
> 	AppendInput.create([
> 		AppendRecord.string({ body: "event-1" }),
> 		AppendRecord.string({ body: "event-2" }),
> 	]),
> );
>
> // The ticket resolves when the batch is durable
> const ack = await ticket.ack();
> console.log(`Durable at seqNum ${ack.start.seqNum}`);
>
> await session.close();
> ```

> [tab] Python
>
> ```python theme={null}
> async with stream.append_session() as session:
>     # Submit a batch — this enqueues it and returns a ticket
>     ticket = await session.submit(
>         AppendInput(
>             records=[
>                 Record(body=b"event-1"),
>                 Record(body=b"event-2"),
>             ]
>         )
>     )
>
>     # The ticket resolves when the batch is durable
>     ack = await ticket
>     print(f"Durable at seq_num {ack.start.seq_num}")
> ```

> [tab] Go
>
> ```go theme={null}
> session, _ := stream.AppendSession(ctx, nil)
> defer session.Close()
>
> // Submit a batch - this enqueues it and returns a ticket
> fut, _ := session.Submit(&s2.AppendInput{
> 	Records: []s2.AppendRecord{
> 		{Body: []byte("event-1")},
> 		{Body: []byte("event-2")},
> 	},
> })
>
> // Wait for enqueue (this is where backpressure happens)
> ticket, _ := fut.Wait(ctx)
>
> // Wait for durability
> ack, _ := ticket.Ack(ctx)
> fmt.Printf("Durable at seqNum %d\n", ack.Start.SeqNum)
> ```

> [tab] Rust
>
> ```rust theme={null}
> let session = stream.append_session(AppendSessionConfig::new());
>
> // Submit a batch - this enqueues it and returns a ticket
> let records = AppendRecordBatch::try_from_iter([
>     AppendRecord::new("event-1")?,
>     AppendRecord::new("event-2")?,
> ])?;
> let ticket = session.submit(AppendInput::new(records)).await?;
>
> // Wait for durability
> let ack = ticket.await?;
> println!("Durable at seqNum {}", ack.start.seq_num);
>
> session.close().await?;
> ```

You do not need to await each ticket immediately. For some use cases, "fire-and-forget" may be fine; for others, you need to confirm that a specific append has finished, or that all writes up to a certain point have been persisted.

If an acknowledgement is not received for a batch within the [configured `requestTimeout`](/docs/sdk/retries-timeouts), the SDK will mark it failed and either retry, if so configured, or surface an error.

### Backpressure

The session tracks how much data is "in flight" (submitted but not yet acknowledged). When you hit the limits, `submit()` blocks until capacity frees up.

This is intentional: it prevents unbounded memory growth and naturally throttles your application to match what S2 can handle.

| Option | Default | Description |
| - | - | - |
| `maxInflightBytes` | 5 MiB | Maximum unacknowledged bytes before `submit()` blocks |
| `maxInflightBatches` | None | Maximum unacknowledged batches (optional additional limit) |

## Producer

The Producer API provides a record-oriented interface over append sessions. You submit individual records and get back a ticket for each one. The producer groups submitted records into batches according to configurable thresholds.

This is particularly useful when:

* You're receiving records one at a time (from a message queue, HTTP requests, etc.)
* You want confirmation that specific records are durable
* You want append-session ordering, pipelining, and backpressure without managing batch boundaries yourself

> [tab] TypeScript
>
> ```typescript theme={null}
> const producer = new Producer(
> 	new BatchTransform({ lingerDurationMillis: 5 }),
> 	await stream.appendSession(),
> );
>
> // Submit individual records
> const ticket = await producer.submit(
> 	AppendRecord.string({ body: "my event" }),
> );
>
> // Get the exact sequence number for this record
> const ack = await ticket.ack();
> console.log(`Record durable at seqNum ${ack.seqNum()}`);
>
> await producer.close();
> ```

> [tab] Python
>
> ```python theme={null}
> async with stream.producer(
>     batching=Batching(linger=timedelta(milliseconds=5)),
> ) as producer:
>     # Submit individual records
>     ticket = await producer.submit(Record(body=b"my event"))
>
>     # Get the exact sequence number for this record
>     ack = await ticket
>     print(f"Record durable at seq_num {ack.seq_num}")
> ```

> [tab] Go
>
> ```go theme={null}
> session, _ := stream.AppendSession(ctx, nil)
> batcher := s2.NewBatcher(ctx, &s2.BatchingOptions{
> 	Linger: 5 * time.Millisecond,
> })
> producer := s2.NewProducer(ctx, batcher, session)
>
> // Submit individual records
> fut, _ := producer.Submit(s2.AppendRecord{Body: []byte("my event")})
> ticket, _ := fut.Wait(ctx)
> ack, _ := ticket.Ack(ctx)
>
> fmt.Printf("Record durable at seqNum %d\n", ack.SeqNum())
>
> producer.Close()
> ```

> [tab] Rust
>
> ```rust theme={null}
> let producer = stream.producer(
>     ProducerConfig::new()
>         .with_batching(BatchingConfig::new().with_linger(Duration::from_millis(5))),
> );
>
> // Submit individual records
> let ticket = producer.submit(AppendRecord::new("my event")?).await?;
>
> // Force the partial batch and wait for durability without closing the producer
> producer.flush().await?;
>
> // Get the exact sequence number
> let ack = ticket.await?;
> println!("Record durable at seqNum {}", ack.seq_num);
>
> producer.close().await?;
> ```

The producer maintains the same ordering guarantee as the underlying append session: records are durable in exactly the order you submitted them, and each per-record ticket resolves with the correct sequence number.

### Batching Configuration

| Option | Default | Description |
| - | - | - |
| `linger` | 5ms | How long to wait for more records before flushing a partial batch |
| `maxBatchRecords` | 1000 | Flush when the batch reaches this many records |
| `maxBatchBytes` | 1 MiB | Flush when the batch reaches this size |

The producer flushes whenever any threshold is hit. For latency-sensitive applications, a shorter linger time means records are written sooner. For throughput-sensitive applications, a longer linger time means more efficient batching.


# Basin Resources
Source: https://s2.dev/docs/sdk/basin-resources

Manage basins with S2 SDKs: list basins, create them, inspect configuration, and delete them.

Basins are containers for streams, similar to buckets in object storage.

Each basin belongs to an account and has its own configuration for default stream settings, which will be used when creating new streams unless explicitly overridden.

> [tab] TypeScript
>
> ```typescript theme={null}
> // List basins
> const basins = await client.basins.list();
>
> // Create a basin
> await client.basins.create({ basin: "my-events" });
>
> // Get configuration
> const config = await client.basins.getConfig({ basin: "my-events" });
>
> // Delete
> await client.basins.delete({ basin: "my-events" });
> ```

> [tab] Python
>
> ```python theme={null}
> # List basins
> basins = await client.list_basins()
>
> # Create a basin
> await client.create_basin("my-events")
>
> # Get configuration
> config = await client.get_basin_config("my-events")
>
> # Delete
> await client.delete_basin("my-events")
> ```

> [tab] Go
>
> ```go theme={null}
> // List basins
> basins, _ := client.Basins.List(ctx, nil)
>
> // Create a basin
> client.Basins.Create(ctx, s2.CreateBasinArgs{Basin: "my-events"})
>
> // Get configuration
> config, _ := client.Basins.GetConfig(ctx, "my-events")
>
> // Delete
> client.Basins.Delete(ctx, "my-events")
> ```

> [tab] Rust
>
> ```rust theme={null}
> // List basins
> let basins = client.list_basins(ListBasinsInput::new()).await?;
>
> // Create a basin
> client
>     .create_basin(CreateBasinInput::new("my-events".parse()?))
>     .await?;
>
> // Get configuration
> let config = client.get_basin_config("my-events".parse()?).await?;
>
> // Delete
> client
>     .delete_basin(DeleteBasinInput::new("my-events".parse()?))
>     .await?;
> ```

See [basin configuration](/docs/api/basins/get-config) for details on available options.


# Endpoint Overrides
Source: https://s2.dev/docs/sdk/endpoints

Configure S2 SDK endpoint overrides for custom account and basin URLs, including local s2-lite development.

By default, SDKs connect to the cloud instance at s2.dev, and only require an access token to get started, which can be generated from the [dashboard](https://s2.dev/dashboard).

SDKs can also connect to any S2-compatible endpoint, such as [s2-lite](https://github.com/s2-streamstore/s2) for local development:

> [tab] TypeScript
>
> ```typescript theme={null}
> const client = new S2({
> 	accessToken: "local-token",
> 	endpoints: {
> 		account: "http://localhost:8080",
> 		basin: "http://localhost:8080",
> 	},
> });
> ```

> [tab] Python
>
> ```python theme={null}
> client = S2(
>     "local-token",
>     endpoints=Endpoints(
>         account="http://localhost:8080",
>         basin="http://localhost:8080",
>     ),
> )
> ```

> [tab] Go
>
> ```go theme={null}
> client := s2.New("local-token", &s2.ClientOptions{
> 	BaseURL: "http://localhost:8080",
> 	MakeBasinBaseURL: func(basin string) string {
> 		return "http://localhost:8080"
> 	},
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> let client = S2::new(
>     S2Config::new("local-token").with_endpoints(S2Endpoints::new(
>         AccountEndpoint::new("http://localhost:8080")?,
>         BasinEndpoint::new("http://localhost:8080")?,
>     )?),
> )?;
> ```

## Environment Variables

SDKs can read endpoint configuration from the environment variables below, but none of them do so automatically. Use the associated helper for reading from the environment:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Reads S2_ACCESS_TOKEN, S2_ACCOUNT_ENDPOINT, and S2_BASIN_ENDPOINT.
> const client = new S2({
> 	...S2Environment.parse(),
> 	accessToken: process.env.S2_ACCESS_TOKEN ?? "local-token",
> });
> ```

> [tab] Python
>
> ```python theme={null}
> # Reads S2_ACCOUNT_ENDPOINT and S2_BASIN_ENDPOINT.
> client = S2(os.environ["S2_ACCESS_TOKEN"], endpoints=Endpoints.from_env())
> ```

> [tab] Go
>
> ```go theme={null}
> // Reads S2_ACCESS_TOKEN, S2_ACCOUNT_ENDPOINT, and S2_BASIN_ENDPOINT.
> client := s2.NewFromEnvironment(nil)
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Reads S2_ACCOUNT_ENDPOINT and S2_BASIN_ENDPOINT.
> let client = S2::new(
>     S2Config::new(std::env::var("S2_ACCESS_TOKEN")?)
>         .with_endpoints(S2Endpoints::from_env()?),
> )?;
> ```

| Variable | Description |
| - | - |
| `S2_ACCOUNT_ENDPOINT` | Account-level API endpoint |
| `S2_BASIN_ENDPOINT` | Basin-level API endpoint (may include `{basin}` placeholder for DNS-based routing) |

The [CLI](/docs/cli/configuration) reads these variables automatically.


# SDKs
Source: https://s2.dev/docs/sdk/languages

Use S2 SDKs for TypeScript, Python, Go, and Rust to create clients, work with basins and streams, and access higher-level APIs.

S2 provides SDKs across several languages to simplify working with our core APIs.

While the [REST API](/docs/api) can be used directly from any HTTP client, the SDKs provide higher-level abstractions (like the [Producer API](/docs/sdk/appending#producer)), retry and error handling logic, rich types, and more performant streaming sessions.

## SDKs

| Language | Repo | Docs | API |
| - | - | - | - |
| TypeScript | [s2-sdk-typescript](https://github.com/s2-streamstore/s2-sdk-typescript) | [typedoc](https://s2-streamstore.github.io/s2-sdk-typescript/index.html) | `v1` |
| Python | [s2-sdk-python](https://github.com/s2-streamstore/s2-sdk-python) | [readthedocs](https://s2-sdk.readthedocs.io/en/stable/api-reference.html) | `v1` |
| Go | [s2-sdk-go](https://github.com/s2-streamstore/s2-sdk-go) | [pkg.go.dev](https://pkg.go.dev/github.com/s2-streamstore/s2-sdk-go/s2) | `v1` |
| Rust | [s2/sdk](https://github.com/s2-streamstore/s2/tree/main/sdk) | [docs.rs](https://docs.rs/s2-sdk/latest/s2_sdk/) | `v1` |

Each repository also contains a variety of examples to demonstrate some common use cases.

## Getting Started

Every SDK follows the same pattern: create a client with an [access token](/docs/concepts/access-tokens), then navigate to basins and streams hierarchically.

Every SDK operation has a simple single-request form. [Appends](/docs/sdk/appending) and [reads](/docs/sdk/reading) also support sessions, which keep a connection open for pipelined writes or streaming records.

> [tab] TypeScript
>
> ```typescript theme={null}
> import { S2 } from "@s2-dev/streamstore";
>
> const client = new S2({
> 	accessToken: process.env.S2_ACCESS_TOKEN!,
> });
>
> const basin = client.basin("my-basin");
> const stream = basin.stream("my-stream");
> ```

> [tab] Python
>
> ```python theme={null}
> import os
>
> from s2_sdk import S2
>
> client = S2(os.environ["S2_ACCESS_TOKEN"])
>
> basin = client.basin("my-basin")
> stream = basin.stream("my-stream")
> ```

> [tab] Go
>
> ```go theme={null}
> package main
>
> import (
> 	"os"
>
> 	"github.com/s2-streamstore/s2-sdk-go/s2"
> )
>
> func main() {
> 	client := s2.New(os.Getenv("S2_ACCESS_TOKEN"), nil)
>
> 	basin := client.Basin("my-basin")
> 	_ = basin.Stream("my-stream")
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> use s2_sdk::{S2, types::S2Config};
>
> let client = S2::new(S2Config::new(std::env::var("S2_ACCESS_TOKEN")?))?;
>
> let basin = client.basin("my-basin".parse()?);
> let stream = basin.stream("my-stream".parse()?);
> ```

> [!TIP]
>
> Using [s2-lite](/docs/s2-lite)? SDKs don't pick up `S2_ACCOUNT_ENDPOINT` / `S2_BASIN_ENDPOINT` automatically. See [reading endpoints from the environment](/docs/sdk/endpoints#environment-variables).

From here, operations fall into two categories: [basin](/docs/sdk/basin-resources) and [stream](/docs/sdk/stream-resources) operations for managing resources, and [appending](/docs/sdk/appending) and [reading](/docs/sdk/reading) for working with records. See [endpoints](/docs/sdk/endpoints) and [retries + timeouts](/docs/sdk/retries-timeouts) for configuration details.


# Metrics
Source: https://s2.dev/docs/sdk/metrics

Query account, basin, and stream metrics with S2 SDKs.

Metrics are available at three levels: account, basin, and stream. Each returns time-series or scalar data for the specified metric set.

> [tab] TypeScript
>
> ```typescript theme={null}
> const now = new Date();
> const thirtyDaysAgo = new Date(Date.now() - 30 * 24 * 3600 * 1000);
> const sixHoursAgo = new Date(Date.now() - 6 * 3600 * 1000);
> const hourAgo = new Date(Date.now() - 3600 * 1000);
>
> // Account-level: active basins over the last 30 days
> const accountMetrics = await client.metrics.account({
> 	set: "active-basins",
> 	start: thirtyDaysAgo,
> 	end: now,
> });
>
> // Basin-level: storage usage with hourly resolution
> const basinMetrics = await client.metrics.basin({
> 	basin: "events",
> 	set: "storage",
> 	start: sixHoursAgo,
> 	end: now,
> 	interval: "hour",
> });
>
> // Stream-level: storage for a specific stream
> const streamMetrics = await client.metrics.stream({
> 	basin: "events",
> 	stream: "user-actions",
> 	set: "storage",
> 	start: hourAgo,
> 	end: now,
> 	interval: "minute",
> });
> ```

> [tab] Python
>
> ```python theme={null}
> now = int(time.time())
> thirty_days_ago = now - 30 * 24 * 3600
> six_hours_ago = now - 6 * 3600
> hour_ago = now - 3600
>
> # Account-level: active basins over the last 30 days
> account_metrics = await client.account_metrics(
>     set=AccountMetricSet.ACTIVE_BASINS,
>     start=thirty_days_ago,
>     end=now,
> )
>
> # Basin-level: storage usage with hourly resolution
> basin_metrics = await client.basin_metrics(
>     "events",
>     set=BasinMetricSet.STORAGE,
>     start=six_hours_ago,
>     end=now,
>     interval=TimeseriesInterval.HOUR,
> )
>
> # Stream-level: storage for a specific stream
> stream_metrics = await client.stream_metrics(
>     "events",
>     "user-actions",
>     set=StreamMetricSet.STORAGE,
>     start=hour_ago,
>     end=now,
>     interval=TimeseriesInterval.MINUTE,
> )
> ```

> [tab] Go
>
> ```go theme={null}
> now := time.Now().Unix()
> thirtyDaysAgo := now - 30*24*3600
> sixHoursAgo := now - 6*3600
> hourAgo := now - 3600
>
> // Account-level: active basins over the last 30 days
> accountMetrics, _ := client.Metrics.Account(ctx, &s2.AccountMetricsArgs{
> 	Set:   s2.AccountMetricSetActiveBasins,
> 	Start: &thirtyDaysAgo,
> 	End:   &now,
> })
>
> // Basin-level: storage usage with hourly resolution
> hourInterval := s2.TimeseriesIntervalHour
> basinMetrics, _ := client.Metrics.Basin(ctx, &s2.BasinMetricsArgs{
> 	Basin:    "events",
> 	Set:      s2.BasinMetricSetStorage,
> 	Start:    &sixHoursAgo,
> 	End:      &now,
> 	Interval: &hourInterval,
> })
>
> // Stream-level: storage for a specific stream
> minuteInterval := s2.TimeseriesIntervalMinute
> streamMetrics, _ := client.Metrics.Stream(ctx, &s2.StreamMetricsArgs{
> 	Basin:    "events",
> 	Stream:   "user-actions",
> 	Set:      s2.StreamMetricSetStorage,
> 	Start:    &hourAgo,
> 	End:      &now,
> 	Interval: &minuteInterval,
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> let now = std::time::SystemTime::now()
>     .duration_since(std::time::UNIX_EPOCH)?
>     .as_secs() as u32;
> let thirty_days_ago = now - 30 * 24 * 3600;
> let six_hours_ago = now - 6 * 3600;
> let hour_ago = now - 3600;
>
> // Account-level: active basins over the last 30 days
> let account_metrics = client
>     .get_account_metrics(GetAccountMetricsInput::new(AccountMetricSet::ActiveBasins(
>         TimeRange::new(thirty_days_ago, now),
>     )))
>     .await?;
>
> // Basin-level: storage usage with hourly resolution
> let basin_metrics = client
>     .get_basin_metrics(GetBasinMetricsInput::new(
>         "events".parse()?,
>         BasinMetricSet::Storage(TimeRange::new(six_hours_ago, now)),
>     ))
>     .await?;
>
> // Stream-level: storage for a specific stream
> let stream_metrics = client
>     .get_stream_metrics(GetStreamMetricsInput::new(
>         "events".parse()?,
>         "user-actions".parse()?,
>         StreamMetricSet::Storage(TimeRange::new(hour_ago, now)),
>     ))
>     .await?;
> ```

See the [metrics guide](/docs/concepts/metrics) for available metric sets and response formats.


# Paginators
Source: https://s2.dev/docs/sdk/paginators

Use S2 SDK paginators to iterate through list results without managing cursors.

List operations like [`listBasins`](/docs/api/basins/list), [`listStreams`](/docs/api/streams/list), and [`listAccessTokens`](/docs/api/access-tokens/list) return paginated results.

SDKs provide automatic pagination helpers that handle async fetching of subsequent pages transparently.

## Usage

Instead of manually handling `hasMore` and `startAfter` cursor parameters, use the paginator variants which return an iterable that automatically fetches pages as needed:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Iterate through all streams with automatic pagination
> for await (const stream of basin.streams.listAll()) {
> 	console.log(stream.name);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Iterate through all streams with automatic pagination
> async for stream in basin.list_all_streams():
>     print(stream.name)
> ```

> [tab] Go
>
> ```go theme={null}
> // Iterate through all streams with automatic pagination
> iter := basin.Streams.Iter(ctx, nil)
> for iter.Next() {
> 	stream := iter.Value()
> 	fmt.Println(stream.Name)
> }
> if err := iter.Err(); err != nil {
> 	log.Fatal(err)
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Iterate through all streams with automatic pagination
> let mut stream = basin.list_all_streams(ListAllStreamsInput::new());
> while let Some(info) = stream.next().await {
>     let info = info?;
>     println!("{}", info.name);
> }
> ```

## Filtering

Paginators accept the same filter options as their single-page list analogs:

> [tab] TypeScript
>
> ```typescript theme={null}
> // List streams with a prefix filter
> for await (const stream of basin.streams.listAll({ prefix: "events/" })) {
> 	console.log(stream.name);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # List streams with a prefix filter
> async for stream in basin.list_all_streams(prefix="events/"):
>     print(stream.name)
> ```

> [tab] Go
>
> ```go theme={null}
> // List streams with a prefix filter
> iter = basin.Streams.Iter(ctx, &s2.ListStreamsArgs{Prefix: "events/"})
> for iter.Next() {
> 	fmt.Println(iter.Value().Name)
> }
> if err := iter.Err(); err != nil {
> 	log.Fatal(err)
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // List streams with a prefix filter
> let input = ListAllStreamsInput::new().with_prefix("events/".parse()?);
> let mut stream = basin.list_all_streams(input);
> while let Some(info) = stream.next().await {
>     println!("{}", info?.name);
> }
> ```

## Deleted Resources

Paginators automatically filter out resources that are pending deletion. This is useful because delete operations mark resources for deletion, but actual garbage collection happens asynchronously.

To include deleted resources in the results:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Include streams that are being deleted
> for await (const stream of basin.streams.listAll({ includeDeleted: true })) {
> 	console.log(stream.name, stream.deletedAt);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Include streams that are being deleted
> async for stream in basin.list_all_streams(include_deleted=True):
>     print(stream.name, stream.deleted_at)
> ```

> [tab] Go
>
> ```go theme={null}
> // Include streams that are being deleted
> iter = basin.Streams.Iter(ctx, &s2.ListStreamsArgs{IncludeDeleted: true})
> for iter.Next() {
> 	stream := iter.Value()
> 	fmt.Println(stream.Name, stream.DeletedAt)
> }
> if err := iter.Err(); err != nil {
> 	log.Fatal(err)
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Include streams that are being deleted
> let input = ListAllStreamsInput::new().with_include_deleted(true);
> let mut stream = basin.list_all_streams(input);
> while let Some(info) = stream.next().await {
>     let info = info?;
>     println!("{} {:?}", info.name, info.deleted_at);
> }
> ```


# Reading
Source: https://s2.dev/docs/sdk/reading

Use S2 SDKs to read records with single-batch reads and read sessions, or inspect the current stream tail.

SDKs expose three ways to read from streams:

* `read` for a single batch of records.
* `readSession` for streaming from a position and optionally following new records.
* `checkTail` for inspecting the current end of a stream.

## Single-batch read

At the simplest level, you can read a batch of records with a single call:

> [tab] TypeScript
>
> ```typescript theme={null}
> const batch = await stream.read({
> 	start: { from: { seqNum: 0 } },
> 	stop: { limits: { count: 100 } },
> });
>
> for (const record of batch.records) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> batch = await stream.read(
>     start=SeqNum(0),
>     limit=ReadLimit(count=100),
> )
>
> for record in batch.records:
>     print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
> batch, _ := stream.Read(ctx, &s2.ReadOptions{
> 	SeqNum: s2.Uint64(0),
> 	Count:  s2.Uint64(100),
> })
>
> for _, record := range batch.Records {
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> let batch = stream
>     .read(
>         ReadInput::new()
>             .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
>             .with_stop(ReadStop::new().with_limits(ReadLimits::new().with_count(100))),
>     )
>     .await?;
>
> for record in batch.records {
>     println!("[{}] {:?}", record.seq_num, record.body);
> }
> ```

Single-batch reads return at most one batch: up to 1000 records or 1 MiB of data. For larger reads or to follow the stream in real-time, use a **read session**.

## Read session

A read session streams records from a starting position. It handles reconnection on transient failures and provides a simple iterator interface. Sessions that do not specify a stop condition will continue to [follow updates](#following-live-updates) in real-time.

### Starting from a sequence number

> [tab] TypeScript
>
> ```typescript theme={null}
> const session = await stream.readSession({
> 	start: { from: { seqNum: 0 } },
> });
>
> for await (const record of session) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> async with stream.read_session(start=SeqNum(0)) as session:
>     async for batch in session:
>         for record in batch.records:
>             print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
> readSession, _ := stream.ReadSession(ctx, &s2.ReadOptions{
> 	SeqNum: s2.Uint64(0),
> })
> defer readSession.Close()
>
> for readSession.Next() {
> 	record := readSession.Record()
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> if err := readSession.Err(); err != nil {
> 	log.Fatal(err)
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> let mut session = stream
>     .read_session(
>         ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0))),
>         ReadSessionConfig::default(),
>     )
>     .await?;
>
> while let Some(batch) = session.next().await {
>     let batch = batch?;
>     for record in batch.records {
>         println!("[{}] {:?}", record.seq_num, record.body);
>     }
> }
> ```

### Starting from a tail offset

Read the last N records in the stream, then follow for new ones:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Start reading from 10 records before the current tail
> const session = await stream.readSession({
> 	start: { from: { tailOffset: 10 } },
> });
>
> for await (const record of session) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Start reading from 10 records before the current tail
> async with stream.read_session(start=TailOffset(10)) as session:
>     async for batch in session:
>         for record in batch.records:
>             print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
> // Start reading from 10 records before the current tail
> tailOffsetSession, _ := stream.ReadSession(ctx, &s2.ReadOptions{
> 	TailOffset: s2.Int64(10),
> })
> defer tailOffsetSession.Close()
>
> for tailOffsetSession.Next() {
> 	record := tailOffsetSession.Record()
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Start reading from 10 records before the current tail
> let mut session = stream
>     .read_session(
>         ReadInput::new().with_start(ReadStart::new().with_from(ReadFrom::TailOffset(10))),
>         ReadSessionConfig::default(),
>     )
>     .await?;
>
> while let Some(batch) = session.next().await {
>     let batch = batch?;
>     for record in batch.records {
>         println!("[{}] {:?}", record.seq_num, record.body);
>     }
> }
> ```

### Starting from a timestamp

Read records starting from a point in time:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Start reading from a specific timestamp
> const oneHourAgo = new Date(Date.now() - 3600 * 1000);
> const session = await stream.readSession({
> 	start: { from: { timestamp: oneHourAgo } },
> });
>
> for await (const record of session) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Start reading from a specific timestamp
> one_hour_ago_ms = int((time.time() - 3600) * 1000)
> async with stream.read_session(start=Timestamp(one_hour_ago_ms)) as session:
>     async for batch in session:
>         for record in batch.records:
>             print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
> // Start reading from a specific timestamp
> oneHourAgo := uint64(time.Now().Add(-time.Hour).UnixMilli())
> timestampSession, _ := stream.ReadSession(ctx, &s2.ReadOptions{
> 	Timestamp: &oneHourAgo,
> })
> defer timestampSession.Close()
>
> for timestampSession.Next() {
> 	record := timestampSession.Record()
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Start reading from a specific timestamp
> let one_hour_ago = std::time::SystemTime::now()
>     .duration_since(std::time::UNIX_EPOCH)?
>     .as_millis() as u64
>     - 3600 * 1000;
> let mut session = stream
>     .read_session(
>         ReadInput::new()
>             .with_start(ReadStart::new().with_from(ReadFrom::Timestamp(one_hour_ago))),
>         ReadSessionConfig::default(),
>     )
>     .await?;
>
> while let Some(batch) = session.next().await {
>     let batch = batch?;
>     for record in batch.records {
>         println!("[{}] {:?}", record.seq_num, record.body);
>     }
> }
> ```

### Reading until a timestamp

Read records up to a point in time, then stop:

> [tab] TypeScript
>
> ```typescript theme={null}
> // Read records until a specific timestamp
> const oneHourAgo = new Date(Date.now() - 3600 * 1000);
> const session = await stream.readSession({
> 	start: { from: { seqNum: 0 } },
> 	stop: { untilTimestamp: oneHourAgo },
> });
>
> for await (const record of session) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Read records until a specific timestamp
> one_hour_ago_ms = int((time.time() - 3600) * 1000)
> async with stream.read_session(
>     start=SeqNum(0),
>     until_timestamp=one_hour_ago_ms,
> ) as session:
>     async for batch in session:
>         for record in batch.records:
>             print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
> // Read records until a specific timestamp
> oneHourAgo = uint64(time.Now().Add(-time.Hour).UnixMilli())
> untilSession, _ := stream.ReadSession(ctx, &s2.ReadOptions{
> 	SeqNum: s2.Uint64(0),
> 	Until:  &oneHourAgo,
> })
> defer untilSession.Close()
>
> for untilSession.Next() {
> 	record := untilSession.Record()
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Read records until a specific timestamp
> let one_hour_ago = std::time::SystemTime::now()
>     .duration_since(std::time::UNIX_EPOCH)?
>     .as_millis() as u64
>     - 3600 * 1000;
> let mut session = stream
>     .read_session(
>         ReadInput::new()
>             .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
>             .with_stop(ReadStop::new().with_until(..one_hour_ago)),
>         ReadSessionConfig::default(),
>     )
>     .await?;
>
> while let Some(batch) = session.next().await {
>     let batch = batch?;
>     for record in batch.records {
>         println!("[{}] {:?}", record.seq_num, record.body);
>     }
> }
> ```

### Following live updates

By default, a read session without stop conditions will follow the stream indefinitely, waiting for new records as they arrive.

When you provide a stop condition (`count`, `bytes`, or `until`), the read stops when either the condition is met **or** it reaches the current tail — whichever comes first.

See the API docs on [waiting at the tail](/docs/api/records/read#waiting-at-the-tail) for the full semantics.

### Long polling

The `wait` parameter controls how long to wait for new records when caught up to the tail. This works for both single-batch reads (long polling) as well as sessions.

> [tab] TypeScript
>
> ```typescript theme={null}
> // Read all available records, and once reaching the current tail, wait an additional 30 seconds for new ones
> const session = await stream.readSession({
> 	start: { from: { seqNum: 0 } },
> 	stop: { waitSecs: 30 },
> });
>
> for await (const record of session) {
> 	console.log(`[${record.seqNum}] ${record.body}`);
> }
> ```

> [tab] Python
>
> ```python theme={null}
> # Read all available records, then wait up to 30 seconds for new ones
> async with stream.read_session(
>     start=SeqNum(0),
>     wait=30,
> ) as session:
>     async for batch in session:
>         for record in batch.records:
>             print(f"[{record.seq_num}] {record.body}")
> ```

> [tab] Go
>
> ```go theme={null}
>    // Read all available records, and once reaching the current tail, wait an additional 30 seconds for new ones
> waitSession, _ := stream.ReadSession(ctx, &s2.ReadOptions{
> 	SeqNum: s2.Uint64(0),
> 	Wait:   s2.Int32(30),
> })
> defer waitSession.Close()
>
> for waitSession.Next() {
> 	record := waitSession.Record()
> 	fmt.Printf("[%d] %s\n", record.SeqNum, string(record.Body))
> }
> ```

> [tab] Rust
>
> ```rust theme={null}
> // Read all available records, and once reaching the current tail, wait an additional 30 seconds
> // for new ones
> let mut session = stream
>     .read_session(
>         ReadInput::new()
>             .with_start(ReadStart::new().with_from(ReadFrom::SeqNum(0)))
>             .with_stop(ReadStop::new().with_wait(30)),
>         ReadSessionConfig::default(),
>     )
>     .await?;
>
> while let Some(batch) = session.next().await {
>     let batch = batch?;
>     for record in batch.records {
>         println!("[{}] {:?}", record.seq_num, record.body);
>     }
> }
> ```

## Check tail

To find the current end of the stream without reading any records:

> [tab] TypeScript
>
> ```typescript theme={null}
> const { tail } = await stream.checkTail();
> console.log(`Stream has ${tail.seqNum} records`);
> ```

> [tab] Python
>
> ```python theme={null}
> tail = await stream.check_tail()
> print(f"Stream has {tail.seq_num} records")
> ```

> [tab] Go
>
> ```go theme={null}
> tail, _ := stream.CheckTail(ctx)
> fmt.Printf("Stream has %d records\n", tail.Tail.SeqNum)
> ```

> [tab] Rust
>
> ```rust theme={null}
> let tail = stream.check_tail().await?;
> println!("Stream has {} records", tail.seq_num);
> ```


# Retries + Timeouts
Source: https://s2.dev/docs/sdk/retries-timeouts

Configure S2 SDK timeouts, retry backoff, and append retry policy.

## Timeouts

SDKs support configuring connection and request timeouts:

> [tab] TypeScript
>
> ```typescript theme={null}
> const client = new S2({
> 	accessToken: accessToken,
> 	connectionTimeoutMillis: 5000,
> 	requestTimeoutMillis: 10000,
> });
> ```

> [tab] Python
>
> ```python theme={null}
> client = S2(
>     access_token,
>     timeout=Timeout(
>         connection=timedelta(seconds=5),
>         request=timedelta(seconds=10),
>     ),
> )
> ```

> [tab] Go
>
> ```go theme={null}
> client := s2.New(accessToken, &s2.ClientOptions{
> 	ConnectionTimeout: 5 * time.Second,
> 	RequestTimeout:    10 * time.Second,
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> let client = S2::new(
>     S2Config::new(access_token)
>         .with_connection_timeout(Duration::from_secs(5))
>         .with_request_timeout(Duration::from_secs(10)),
> )?;
> ```

* **Connection timeout**: Maximum time to wait for a TCP connection to be established. This is a "fail fast" timeout that aborts slow connections early.
* **Request timeout**: Maximum time to wait for a non-streaming request to complete. For [append sessions](/docs/sdk/appending#append-session), this value is also used to determine the SDK-enforced timeout for an individual batch append to be acknowledged.

## Retry Behavior

SDKs automatically retry transient failures using exponential backoff with jitter.

> [tab] TypeScript
>
> ```typescript theme={null}
> const client = new S2({
> 	accessToken: accessToken,
> 	retry: {
> 		maxAttempts: 5,
> 		minBaseDelayMillis: 100,
> 		maxBaseDelayMillis: 2000,
> 	},
> });
> ```

> [tab] Python
>
> ```python theme={null}
> client = S2(
>     access_token,
>     retry=Retry(
>         max_attempts=5,
>         min_base_delay=timedelta(milliseconds=100),
>         max_base_delay=timedelta(seconds=2),
>     ),
> )
> ```

> [tab] Go
>
> ```go theme={null}
> client := s2.New(accessToken, &s2.ClientOptions{
> 	RetryConfig: &s2.RetryConfig{
> 		MaxAttempts:  5,
> 		MinBaseDelay: 100 * time.Millisecond,
> 		MaxBaseDelay: 2 * time.Second,
> 	},
> })
> ```

> [tab] Rust
>
> ```rust theme={null}
> let client = S2::new(
>     S2Config::new(access_token).with_retry(
>         RetryConfig::new()
>             .with_max_attempts(NonZeroU32::new(5).unwrap())
>             .with_min_base_delay(Duration::from_millis(100))
>             .with_max_base_delay(Duration::from_secs(2)),
>     ),
> )?;
> ```

### Append Retry Policy

Retrying appends requires care: if an append succeeds but the acknowledgement is lost (e.g., due to a network timeout), retrying would create duplicate records in a stream.

The `appendRetryPolicy` setting controls how to react in this situation, by treating append retries as a special case.

| Policy | Behavior |
| - | - |
| `all` (default) | Retry all failed appends. Duplicates are possible. |
| `noSideEffects` | Only retry append failures which could not have had a side effect. |

Using `noSideEffects` means that, in the event of a failure, the caller will manually need to verify whether or not the append succeeded (e.g., by reading or otherwise inspecting the state of the stream).

> [!TIP]
>
> To write exactly-once in single-writer setups, append with a `matchSeqNum` precondition. See [concurrency control](/docs/concepts/concurrency-control) for details.
>
> For multi-writer setups, include dedupe metadata such as a writer ID and per-writer idempotency key, then deduplicate later. This is touched upon in [this blog post](https://s2.dev/blog/patterns).

## Configuration Reference

### Timeout Settings

| Setting | Default | Description |
| - | - | - |
| Connection timeout | 3 seconds | Max time to establish TCP connection |
| Request timeout | 5 seconds | Max time for request completion |

### Retry Settings

| Setting | Default | Description |
| - | - | - |
| Max attempts | 3 | Maximum retry attempts |
| Min base delay | 100 ms | Minimum base delay for exponential backoff |
| Max base delay | 1 second | Maximum base delay (actual delay with jitter can be up to 2x) |
| Append retry policy | `all` | How to handle append retries when duplication is possible |


# Stream Resources
Source: https://s2.dev/docs/sdk/stream-resources

Manage streams with S2 SDKs: list streams, create them, inspect configuration, and delete them.

Streams live within basins and are the core data structure in S2.

You can have an unlimited number of streams, and create them instantly (even on first use).

Data retention policies, storage classes, and timestamping behavior are all configurable.

> [tab] TypeScript
>
> ```typescript theme={null}
> // List streams
> const streams = await basin.streams.list({ prefix: "user-" });
>
> // Create a stream
> await basin.streams.create({
> 	stream: "user-actions",
> 	config: {
> 		/* optional configuration */
> 	},
> });
>
> // Get configuration
> const streamConfig = await basin.streams.getConfig({ stream: "user-actions" });
>
> // Delete
> await basin.streams.delete({ stream: "user-actions" });
> ```

> [tab] Python
>
> ```python theme={null}
> # List streams
> streams = await basin.list_streams(prefix="user-")
>
> # Create a stream
> await basin.create_stream(
>     "user-actions",
>     # config=StreamConfig(...)  # optional
> )
>
> # Get configuration
> config = await basin.get_stream_config("user-actions")
>
> # Delete
> await basin.delete_stream("user-actions")
> ```

> [tab] Go
>
> ```go theme={null}
> // List streams
> streams, _ := basin.Streams.List(ctx, &s2.ListStreamsArgs{Prefix: "user-"})
>
> // Create a stream
> basin.Streams.Create(ctx, s2.CreateStreamArgs{
> 	Stream: "user-actions",
> 	Config: &s2.StreamConfig{ /* optional */ },
> })
>
> // Get configuration
> config, _ := basin.Streams.GetConfig(ctx, "user-actions")
>
> // Delete
> basin.Streams.Delete(ctx, "user-actions")
> ```

> [tab] Rust
>
> ```rust theme={null}
> // List streams
> let streams = basin
>     .list_streams(ListStreamsInput::new().with_prefix("user-".parse()?))
>     .await?;
>
> // Create a stream
> // Optionally, pass `.with_config(StreamConfig { .. })` to CreateStreamInput.
> basin
>     .create_stream(CreateStreamInput::new("user-actions".parse()?))
>     .await?;
>
> // Get configuration
> let config = basin.get_stream_config("user-actions".parse()?).await?;
>
> // Delete
> basin
>     .delete_stream(DeleteStreamInput::new("user-actions".parse()?))
>     .await?;
> ```

See [stream configuration](/docs/api/streams/get-config) for more details on all options.

## Storage classes

Check available classes and the default for the basin's location before choosing one:

> [tab] TypeScript
>
> ```typescript theme={null}
> const locations = await s2.locations.list();
> const config = { storageClass: "express" };
> ```

> [tab] Go
>
> ```go theme={null}
> locations, err := client.Locations.List(ctx)
> if err != nil {
>     return err
> }
> config := s2.StreamConfig{StorageClass: s2.Ptr("express")}
> ```

> [tab] Rust
>
> ```rust theme={null}
> let locations = client.list_locations().await?;
> let config = StreamConfig::default().with_storage_class("express");
> ```

Omitting the class on creation uses the location's default for a basin, or the basin's default for a stream.


# Studio
Source: https://s2.dev/docs/studio

S2 Studio is a browser-based data plane explorer built into the dashboard for reading and writing stream records interactively.

## Overview

Studio lets you:

* **Read** records from any stream, starting from a tail offset, sequence number, or timestamp
* **Stream** records live as they arrive
* **Append** records individually, line-by-line, or as a JSON array
* **Trim** records up to a sequence number to reclaim storage
* **Fence** a stream to enforce exclusive write access

***

## Opening Studio

Navigate to any basin in the dashboard, then click **Studio** in the stream list header. You can also switch basins and streams directly from within Studio using the selectors at the top.

![](https://mintcdn.com/streamstore/0pAUJuHAWw3coDj0/images/studio-1.png?fit=max\&auto=format\&n=0pAUJuHAWw3coDj0\&q=85\&s=78faed598ce2a2ce6741c0160f942bd6)

***

## s2-lite

[s2-lite](/docs/s2-lite) serves its own dashboard, including Studio, at `/dashboard` (e.g. `http://localhost:8080/dashboard`).

***

## Read

The **Read** tab lets you fetch records from the selected stream.

![](https://mintcdn.com/streamstore/0pAUJuHAWw3coDj0/images/studio-4.png?fit=max\&auto=format\&n=0pAUJuHAWw3coDj0\&q=85\&s=51e991f06961f61119298b636d86fdb4)

| Field | Description |
| - | - |
| **Start from** | How to anchor the read position: tail offset, sequence number, or timestamp |
| **Tail offset** | Number of records back from the stream tail (e.g. `10` = last 10 records) |
| **Sequence number** | Absolute sequence number of the first record to read |
| **Timestamp** | Start from the first record at or after this time |
| **Limit** | Maximum records to return per batch (up to 1000) |
| **Wait (s)** | Seconds to hold a streaming connection open at the tail before stopping. `0` stops immediately at tail. |
| **Clamp to tail** | If the start position is beyond the tail, clamp to tail instead of erroring |
| **Ignore command records** | Filter out fence and trim command records i.e. only data records are returned |

### Read Once vs Stream

* **Read Once** — fetches a single batch with the current settings
* **Stream** — opens a continuous session that delivers records as they arrive

***

## Append

The **Append** tab lets you write records to the selected stream.

![](https://mintcdn.com/streamstore/0pAUJuHAWw3coDj0/images/studio-5.png?fit=max\&auto=format\&n=0pAUJuHAWw3coDj0\&q=85\&s=9dc05548c24954f153c9767b0ee8876d)

### Input modes

| Mode | Behavior |
| - | - |
| **Single body** | The entire payload is sent as one record |
| **One per line** | Each non-empty line becomes a separate record |
| **JSON array** | Parse payload as a JSON array of `{body, headers?}` objects |

You can also **load a file** directly into the editor — the input mode is auto-detected from the file extension (`.json` → JSON array, `.jsonl`/`.ndjson` → one per line, `.txt` → single body).

### Optional fields

| Field | Description |
| - | - |
| **Match Sequence Number** | Only append if the stream tail matches this sequence number. |
| **Fencing token** | Required if the stream has a fencing token set. Must match exactly (max 36 bytes UTF-8). |

### Commands

| Command | Description |
| - | - |
| **Trim through seq #** | Discard all records up to and including this sequence number. |
| **Fence** | Set a new fencing token (max 36 bytes). Once set, all appends must supply this token to enforce single-writer access. |


# Updates
Source: https://s2.dev/docs/updates

Product updates for S2, the Stream Store, including APIs, integrations, infrastructure changes, and new releases.

**Oct 30, 2026**

[s2-lite](/docs/s2-lite) now ships with a local web dashboard at `/dashboard` (e.g. `http://localhost:8080/dashboard`) for browsing basins and streams and using [Studio](/docs/studio).

**Oct 14, 2026**

View estimated billable usage in USD by pricing category in the [dashboard](https://s2.dev/dashboard/billing) and through the account-level [metrics API](/docs/api/metrics/account#estimated-billable-usage) with `estimated-billable-usage`. Estimates include only finalized hours and are not a final invoice.

**Oct 6, 2026**

Beta support for the [Redis Streams](/docs/integrations/redis-streams) API, so Redis clients can append to, replay, and tail S2 streams.

**Sep 29, 2026**

[Feldera](/docs/integrations/feldera) integration is available, with native `s2_input` and `s2_output` connectors for reading S2 streams into Feldera SQL pipelines and appending view updates back to S2.

**Sep 14, 2026**

The [`s2-stream-config`](/docs/api/records/append#stream-config-on-auto-creation) header sets the config of a stream that is auto-created on append or read, overriding the basin's default stream config for that stream.

**Aug 29, 2026**

[tail.surf](https://tail.surf) is available, powered by s2 — like a gist, but live.

**Jul 22, 2026**

Interactive TUI mode has been removed from the `s2` [CLI](https://github.com/s2-streamstore/s2).

**Jul 15, 2026**

[`@s2-dev/mastra-pubsub`](https://www.npmjs.com/package/@s2-dev/mastra-pubsub) is available. It stores [Mastra durable-agent](/docs/integrations/mastra) stream events in S2 so clients can reconnect to an active run by `runId`.

**Jul 9, 2026**

[Streamling](/docs/integrations/streamling) integration is available, with S2 source and sink plugins for reading from and writing to S2 in Streamling pipelines.

**Jul 7, 2026**

The account-level [endpoint](/docs/api/endpoints) has been updated to `a.s2.dev`.

**Jul 6, 2026**

[opbox](https://opbox.dev) is an open-source daemon that syncs plain-text directories between machines in real time, backed by S2.

**Jun 15, 2026**

Rust clients can use the [`s2-testcontainers`](https://crates.io/crates/s2-testcontainers) crate for easy integration testing against [s2-lite](/docs/s2-lite).

**Jun 1, 2026**

[RePlaya](https://github.com/s2-streamstore/replaya) is a self-hosted browser session replay tool built on S2, with support for live tailing active sessions.

**May 29, 2026**

[Location](/docs/concepts/basins#location) is now a first-class construct, with two new ones available: US West (`aws:us-west-2`) and EU North (`aws:eu-north-1`).

**May 12, 2026**

[Anthropic](/docs/integrations/anthropic) integration in `@s2-dev/resumable-stream` is available. Build durable Claude chat sessions with it.

**May 1, 2026**

[s2.dev](https://s2.dev) is more agent-friendly now as all public pages handle `Accept: text/markdown`. You can also request Markdown at `.md` e.g. [pricing.md](https://s2.dev/pricing.md), [blog.md](https://s2.dev/blog.md).

**Apr 30, 2026**

[TanStack AI](/docs/integrations/tanstack-ai) integration in `@s2-dev/resumable-stream` is available. Build durable chat sessions with it.

**Apr 29, 2026**

S2's `v1alpha` gRPC API has retired. It has been fully superseded by the REST-first `v1` [API](/docs/api).

**Apr 23, 2026**

[Client-supplied encryption keys](/docs/concepts/encryption).

**Apr 22, 2026**

[Vercel AI SDK](/docs/integrations/vercel-ai-sdk) integration in `@s2-dev/resumable-stream` is available. Build durable chat sessions with it.

**Apr 6, 2026**

New [Python SDK](https://github.com/s2-streamstore/s2-sdk-python), powered by the `v1` [API](/docs/api).

**Mar 31, 2026**

Number of basins per account is [now unlimited](/docs/platform/limits) for all users with a payment method. Previously this was capped at 100.

**Mar 23, 2026**

The basin-level [endpoint](/docs/api/endpoints) has been updated to `{basin}.b.s2.dev`. To preserve compatibility, `{basin}.b.aws.s2.dev` will continue to work.

**Mar 18, 2026**

[Video Conferencing with Durable Streams](https://s2.dev/blog/video-conferencing) shows live media, chat, replay, and MP4 export backed by S2 streams.

**Mar 3, 2026**

[Terraform Provider](/docs/integrations/terraform-provider) for S2 is available on the [Terraform Registry](https://registry.terraform.io/providers/s2-streamstore/s2/latest).

**Feb 25, 2026**

S2 cloud service is [GA](https://s2.dev/blog/ga).

**Feb 25, 2026**

`s2 apply` can declaratively create or reconfigure basins and streams from a JSON spec file. Run with `--dry-run` to preview changes before applying. Use `--schema` to get the JSON Schema for IDE validation and autocomplete. See [CLI: Apply](/docs/cli/apply) for the full reference.

s2-lite now accepts `--init-file` (or `S2LITE_INIT_FILE`) to apply a spec at startup. Useful for pre-seeding basins and streams in local dev and CI environments.

**Feb 24, 2026**

[Studio](/docs/studio) is now available in the dashboard. A browser-based data plane explorer for reading, appending, and managing streams interactively.

**Feb 22, 2026**

[Parallax](https://s2.dev/blog/distributed-ai-agents) explores distributed AI agent coordination with isolated S2 streams for each research cohort.

**Jan 29, 2026**

`s2` [CLI](https://github.com/s2-streamstore/s2) has an interactive TUI mode, launch with `s2 -i`.

**Jan 21, 2026**

[s2-lite](https://github.com/s2-streamstore/s2?tab=readme-ov-file#s2-lite) is available. It is an open source, self-hostable server implementation that runs against object storage. You can also run it without pointing at a bucket, as an emulator.

**Jan 20, 2026**

`s2` [CLI](https://github.com/s2-streamstore/s2) has a new `bench` command that measures throughput and latency.

**Jan 14, 2026**

New [Rust SDK](https://github.com/s2-streamstore/s2/tree/main/sdk), powered by the `v1` [API](/docs/api).

**Dec 29, 2025**

S2 endpoints are accessible using IPv6. This could be particularly helpful for avoiding NAT gateway costs.

**Dec 17, 2025**

S2 is SOC 2 Type II compliant.

**Nov 19, 2025**

[`@s2-dev/streamstore-patterns`](https://github.com/s2-streamstore/s2-sdk-typescript/tree/main/packages/patterns) launched as an add-on package for higher-level TypeScript patterns, including typed append/read sessions, large-message chunking, framing, and deduplication helpers.

**Nov 5, 2025**

`PrivateLink` connection info is now available in the basins tab on the dashboard.

**Oct 24, 2025**

[`s2s`](/docs/api/protocol#sessions) is available, and supported for append and read sessions from the TypeScript SDK. Other SDKs will also be migrated away from gRPC to use the REST API and `s2s`.

**Sep 26, 2025**

Long polling reads are supported using the [`?wait=<seconds>` parameter](/docs/api/records/read#waiting-at-the-tail).

**Sep 10, 2025**

Enabled massive scalability for reads and open sourced a key piece of this, [Cachey](https://github.com/s2-streamstore/cachey).

**Aug 29, 2025**

Production-ready support for the [OpenTelemetry Protocol](/docs/integrations/otlp) (OTLP/HTTP).

**Aug 12, 2025**

See [usage metrics](/docs/concepts/metrics) visualized in the dashboard UI.

**Aug 1, 2025**

Read from streams directly in the dashboard UI.

**Jul 24, 2025**

Stream configuration knob [`delete-on-empty`](/docs/concepts/configs#delete-on-empty) to automatically delete a stream if it does not have any records for some period of time.

**Jul 22, 2025**

Reads can [`clamp`](/docs/api/records/read#starting-point) the start position at the tail position.

**Jul 9, 2025**

Interactive [playground](https://s2.dev/playground).

**Jun 24, 2025**

Experimental support for the [OpenTelemetry Protocol](https://opentelemetry.io/docs/specs/otlp) (OTLP/HTTP). Let us know if you would like access.

**Jun 6, 2025**

Reads can be limited [`until`](/docs/api/records/read#bounds) a specific timestamp.

**Jun 3, 2025**

Support for listing, issuing, revoking granular [access tokens](https://s2.dev/blog/access-control).

**May 28, 2025**

Added [metrics endpoints](/docs/concepts/metrics) for usage monitoring to the REST API.

**May 20, 2025**

REST API v1 is ready including [SSE](/docs/api/records/read#sse) support. Besides JSON it also [supports Protobuf](/docs/api/protocol#protobuf-messages) for appending and reading binary data efficiently without the Base64 tax.

**May 8, 2025**

Supercharged metadata scalability.

**May 2, 2025**

CLI got a new `tail` sub-command.

**Apr 28, 2025**

Support for reading from `timestamp` or a `tail_offset`, not just sequence numbers. The implicit cap on record timestamp not exceeding arrival time can now be disabled in stream config.

**Apr 18, 2025**

Prefix filtering in dashboard on basin or stream names.

**Apr 17, 2025**

`first_seq_num` as a `Read` / `ReadSession` response has been removed: records will be returned from the first record `>=` the requested starting point.

**Apr 14, 2025**

Basins can be configured to `create_stream_on_read` automatically when the specified stream does not exist, using the default stream configuration for the basin.

**Apr 1, 2025**

Dashboard UI for stream management.

**Mar 14, 2025**

All records now have a monotonic timestamp in milliseconds since Unix epoch. Streams can also be configured to only utilize client-specified timestamps.

**Mar 11, 2025**

Basins can be configured to `create_stream_on_append` automatically when the specified stream does not exist, using the default stream configuration for the basin.

**Mar 10, 2025**

Dashboard UI for basin management.

**Mar 6, 2025**

`ReadSession` gRPC and REST API support periodic heartbeats.

**Feb 27, 2025**

[Bytewax](https://web.archive.org/web/20260618142759/https://bytewax.io/blog/s2-connector/) integration is available.

**Feb 21, 2025**

[REST API](/docs/api) supports [compression](https://developer.mozilla.org/en-US/docs/Web/HTTP/Compression) using `zstd` or `gzip`, except for streaming reads with SSE.

**Feb 5, 2025**

[Bento](/docs/integrations/bento) integration is available.

**Feb 4, 2025**

gRPC API supports [compression](https://grpc.io/docs/guides/compression) using `zstd` or `gzip`.

**Jan 30, 2025**

[REST API](/docs/api) is ready, including streaming reads with [SSE](https://en.wikipedia.org/wiki/Server-sent_events).

**Jan 14, 2025**

[Python SDK](https://github.com/s2-streamstore/s2-sdk-python-old) available.

**Jan 13, 2025**

[Go SDK](https://github.com/s2-streamstore/s2-sdk-go) available.

**Dec 20, 2024**

[Preview launch](https://s2.dev/blog/intro)  with gRPC API, [Rust SDK](https://github.com/s2-streamstore/s2-sdk-rust), and [CLI](https://github.com/s2-streamstore/s2-cli).

**Jan-Feb 2024**

Prototyped the [guts of S2](https://blog.schmizz.net/designing-serverless-stream-storage#heading-architecture).

**Nov 2023**

S2 [hypothesized](https://blog.schmizz.net/disaggregated-wal).


# Agents
Source: https://s2.dev/docs/use-cases/agents

Use S2 as the event log for agent sessions, so chats, tool calls, and state can be streamed live, resumed, and replayed later.

The [typical pattern](https://s2.dev/blog/agent-sessions) is a stream per agent run.

Each invocation gets its own stream (`s2://my-basin/agents/{run-id}`) containing an ordered log of everything that happened — tool calls, LLM responses, state changes.

Streams can be [created automatically](/docs/concepts/streams#creation) on first append or read, to simplify setup.

## Architecture

```mermaid theme={null}
graph LR
    A[agent] -->|append| S[s2://my-basin/agents/run-123]
    S -->|read SSE| C[client UI]

    subgraph infra["s2.dev"]
        S
    end
```

## Chat / LLM token streaming

The agent (or your backend) appends tokens to a stream as they're generated. The client opens a [read session](/docs/sdk/reading) and renders them as they arrive. If the client disconnects and reconnects, it can easily resume from the last sequence number it processed by starting a new read session from that position.

> [!TIP]
>
> Use [`@s2-dev/resumable-stream`](/docs/integrations/vercel-ai-sdk) with the Vercel AI SDK, or [`@s2-dev/mastra-pubsub`](/docs/integrations/mastra) with Mastra durable agents.

## Agent logs and debugging

Actions the agent takes — tool invocations, API calls, intermediate reasoning — can be appended as structured records using [headers](/docs/concepts/records). This gives you an audit trail for each run. Read the stream later to debug failures, or follow it live to watch an agent work. See the [agent session example](/docs/integrations/vercel-ai-sdk#example) for a working implementation with the Vercel AI SDK.

<video />

## State and context storage

A stream can serve as an append-only log of state changes for an agent session. Instead of overwriting state in a database, append each change as a record. Replay the stream to reconstruct state at any point — essentially event sourcing at the session level.

For long-running sessions, [external snapshots](/docs/concepts/snapshots#external-snapshots) can speed up reconstruction while the stream remains the durable history.

<video />

## Multi-agent coordination

When multiple agents need to communicate, S2 streams provide natural coordination primitives. Use a shared stream as a message bus between agents, with per-agent streams for private memory. The ordered log guarantees all agents see events in the same order. See the [dinner party example](/docs/integrations/vercel-ai-sdk#example) for a multi-agent implementation with the Vercel AI SDK.

<video />

## Getting started

For a ready-made integration, see the [Vercel AI SDK](/docs/integrations/vercel-ai-sdk)
or [Mastra](/docs/integrations/mastra) guide.

For custom setups, this flow can be modeled directly from the CLI:

```bash theme={null}
# Create a basin with auto-stream creation
s2 create-basin agent-runs --create-stream-on-append

# Agent appends (from your backend)
echo "tool_call: search('weather in SF')" \
    | s2 append s2://agent-runs/session-abc

# Client reads (from your frontend, using the token you issued)
s2 read -s0 s2://agent-runs/session-abc
```

In production, you would likely want to build around these same methods via our SDKs — in particular, via the [Producer API](/docs/sdk/appending#producer), which handles batching and pipelining automatically.


# Data Feeds
Source: https://s2.dev/docs/use-cases/data-feeds

Publish data once with S2, then let many consumers follow live updates or replay history at their own pace.

S2 streams are well suited for broadcasting data from a single writer to a large number of readers. One process [appends](/docs/api/records/append) to the stream, and many [read](/docs/api/records/read) from it in such setups.

Readers can follow updates in real-time, or consume from historical prefixes of the stream — either by sequence number, offset from latest message, or timestamp.

## Architecture

```mermaid theme={null}
graph LR
    A[producer] -->|append| S[s2://my-basin/market]
    S -->|read| R1[consumer 1]
    S -->|read| R2[consumer 2]
    S -->|read| R3[consumer 3]

    subgraph infra["s2.dev"]
        S
    end
```

Common use cases include system monitoring, live event streams, and market data distribution.

## Why S2 for broadcast?

S2's [granular access controls](/docs/concepts/access-tokens) let you configure read-only tokens that can be created per-user for fine-grained control, or you can have all consumers share a single token. Streams are accessible directly from the web via REST, so you don't need to run any servers to maintain connections with consumers.

Moreover, S2 supports high write throughput as well as massive read fanouts, so you can offload a lot of scaling challenges.

Messages are only transmitted after they've been sequenced and made durable on object storage, giving you deterministic ordering. You never need to worry about messages getting shuffled, or consumers seeing something slightly different.

The same [read API](/docs/api/records/read) seamlessly integrates historical data with real-time updates. You can read from any point to consume historical data, and when caught up to the current moment, `read` transparently switches to following live updates.

## Simple example

Here's a simple demonstration of the core concept using the CLI. You'd typically use one of our [SDKs](/docs/sdk/overview) to build real applications, but this shows the basic idea.

**Write a log stream:**

```bash theme={null}
tail -f /var/log/app.log | s2 append s2://my-basin/prod/logs
```

Now this feed is available for any number of readers to consume.

**Create a read-only token scoped to this stream:**

```bash theme={null}
s2 issue-access-token public-ro-logs \
  --op-group-perms "stream=r" \
  --basins-exact "my-basin" \
  --streams-exact "prod/logs"
```

This should produce a token, like `I7oAAAAAAABo+9HGXPSurWahOECn5Q21Nf698JSaUh1nmYWG`, which can then be provided to your readers.

**Read historically with timestamp (as milliseconds since Unix epoch) bounds:**

```bash theme={null}
# use the read-only token
export S2_ACCESS_TOKEN="I7oAAAAAAABo+9HGXPSurWahOECn5Q21Nf698JSaUh1nmYWG"
s2 read s2://my-basin/prod/logs \
  --timestamp 1761333455729 \
  --until 1761333456740
```

**Tail the stream in real-time:**

```bash theme={null}
# use the read-only token
export S2_ACCESS_TOKEN="I7oAAAAAAABo+9HGXPSurWahOECn5Q21Nf698JSaUh1nmYWG"
s2 read s2://my-basin/prod/logs
```

Multiple readers can consume the same stream simultaneously, each at their own pace and from their own starting position.


# Live View
Source: https://s2.dev/docs/use-cases/live-view

Stream job output, build logs, and other live events into S2 so dashboards can show what is happening as it happens.

The pattern here is one stream per task, job, or session. A producer appends events as they happen, and consumers [read](/docs/api/records/read) them in real-time. The same read API serves both historical and live data — when a reader catches up to the tail, the connection stays open and new records arrive as they're appended.

## Architecture

```mermaid theme={null}
graph LR
    E[executor] -->|append| S[s2://my-basin/jobs/user-123/job-456]
    S -->|read| D[dashboard]

    subgraph infra["s2.dev"]
        S
    end
```

## Serverless function logs

If you run sandboxed execution environments — CI/CD runners, browsers-as-a-service, coding sandboxes — your customers likely want to see what's happening in real-time.

Store each job's output in its own stream. When a user kicks off a job, create a read-only [access token](/docs/concepts/access-tokens) scoped to that stream and return it alongside the job ID. The executor [appends](/docs/api/records/append) logs directly. The customer [reads](/docs/api/records/read) and tails them live.

See the [Browser Infra demo](/docs/demos) for a live example of this pattern — click into tasks to watch their progress in real-time.

## Build log streaming

Same pattern applied to build pipelines. Stream build output to users as it happens. [Delete-on-empty](/docs/concepts/configs#delete-on-empty) can clean up streams automatically once they've been fully consumed and are no longer needed.

## Getting started

```bash theme={null}
# Create a basin with auto-stream creation
s2 create-basin my-jobs --create-stream-on-append

# Executor appends logs as the job runs
echo "Step 1: Installing dependencies..." \
    | s2 append s2://my-jobs/user-123/job-456
    
echo "Step 2: Running tests..." \
    | s2 append s2://my-jobs/user-123/job-456

# User tails the stream live
s2 read s2://my-jobs/user-123/job-456
```

In production, you'd issue a [scoped access token](/docs/concepts/access-tokens) for the user and have them read via [SSE](/docs/api/records/read#sse) or an SDK.


# Sync
Source: https://s2.dev/docs/use-cases/sync

Use S2 as the shared log for multiplayer apps and data sync, with ordered history that clients and services can replay whenever they need to catch up.

S2 streams can serve as the shared log that keeps data in sync — between clients, between services, or between databases and downstream consumers.

## Event sourcing

Store events instead of current state. Append each change to a stream, and materialize state by replaying from the beginning — or from a [checkpoint](/docs/concepts/snapshots). New consumers can be added at any time by reading the full history.

S2's [strong consistency](/docs/concepts/reads#consistency) and [durable total ordering of records](/docs/concepts/streams) make it a natural fit as a shared write-ahead log.

See the [kv-store blog post](https://s2.dev/blog/kv-store) for a detailed example of how a replicated database with strong consistency can be built directly over the S2 primitive.

## WebSockets alternative

S2 streams can replace stateful WebSocket connections for bidirectional communication. Use two streams (one per direction) or a single shared stream. Records are durable and ordered, clients can reconnect and resume from their last sequence number, and there's no connection state to manage server-side.

## Related integrations

> [card] [Y-S2](/docs/integrations/y-s2)
>
> Build Yjs collaborative editing on S2 streams.

> [card] [LiveStore](/docs/integrations/livestore)
>
> Use S2 as a durable sync backend for LiveStore.