hydrant

stream

created 6 minutes ago :: modified 6 minutes ago :: 4.14 KB .md

Tools

download raw

available in builds compiled with the indexer_stream feature. /stream is hydrant's ordered websocket event stream. each subscriber receives a full copy of the stream.

/stream is not wire- or protocol-compatible with tap. tap uses a sharded work queue, server-managed acknowledgements, and redelivery; hydrant uses broadcast delivery and client-managed cursors.

GET /stream#

subscribe to the ordered event stream.

query parameters#

paramtypedescription
cursorintegerinclusive u64 event ID from which to replay durable record events.

if cursor is omitted, the connection starts at the current stream head and receives only new events. if it is present, hydrant replays durable record events whose IDs are greater than or equal to the cursor, up to the head observed when the connection starts, then switches to live delivery.

cursor=0 replays all retained record events. IDs are monotonic but may have gaps: live-only identity and account events consume IDs without being persisted, and events that cannot be inflated are skipped.

event envelope#

events are JSON text frames with this envelope:

{
  "id": 42,
  "type": "record",
  "record": {}
}

fieldtypedescription
idintegeru64 stream event ID.
typestringrecord, identity, or account.
recordobjectpresent when type is record.
identityobjectpresent when type is identity.
accountobjectpresent when type is account.

only the payload matching type is present.

record#

record events are durable and available through cursor replay.

{
  "id": 42,
  "type": "record",
  "record": {
    "live": true,
    "did": "did:plc:abc123xyz",
    "rev": "3kpjxabc123",
    "collection": "app.bsky.feed.post",
    "rkey": "3kpjxabc123",
    "action": "create",
    "record": {
      "$type": "app.bsky.feed.post",
      "text": "hello, world!",
      "createdAt": "2026-05-27T12:00:00.000Z"
    },
    "cid": "bafyreihy..."
  }
}

fieldtypedescription
livebooleantrue for live ingestion; false for backfill.
didstringrepository DID.
revstringrepository revision TID.
collectionstringcollection NSID.
rkeystringrecord key.
actionstringcreate, update, or delete.
recordobjectdecoded DAG-CBOR record; omitted when no record content is stored.
cidstringCID of record; omitted when no record content is stored.

record and cid are omitted for deletes and when hydrant is configured not to store record content, including HYDRANT_ONLY_INDEX_LINKS=true.

identity#

identity events are live-only and are not available through cursor replay.

{
  "id": 43,
  "type": "identity",
  "identity": {
    "did": "did:plc:abc123xyz",
    "handle": "user.bsky.social"
  }
}

fieldtypedescription
didstringrepository DID.
handlestringcurrent handle; omitted when unavailable.

account#

account events are live-only and are not available through cursor replay.

{
  "id": 44,
  "type": "account",
  "account": {
    "did": "did:plc:abc123xyz",
    "active": false,
    "status": "deactivated"
  }
}

fieldtypedescription
didstringrepository DID.
activebooleanwhether the repository is active.
statusstringaccount status; omitted when unavailable.

websocket behavior#

  • the server sends an empty ping every 30 seconds and responds to client ping frames with pong frames.
  • client text and binary frames are not accepted; hydrant closes the connection when it receives one.
  • if sending blocks for HYDRANT_STREAM_SEND_TIMEOUT (30 seconds by default), hydrant sends an error text frame and closes the connection:

{
  "type": "error",
  "error": "ConsumerTooSlow",
  "message": "stream socket send blocked for at least 30 seconds"
}