[stream] seal /stream into chunks a subscriber can start from
1439 unmodified linesExpand all14401441144214431444144514461447144814491450145114521453672 unmodified linesExpand all2126212721282129213021312132213321342135213621372138213921402141214221431493 unmodified linesExpand all36373638363936403641364236433644364536463647364836493650105 unmodified linesExpand all37563757375837593760376137623763376437653766376737683769633 unmodified linesExpand all1439 unmodified linesExpand all "pin-project-lite",]
[[package]]name = "http-range-header"version = "0.4.2"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c"
[[package]]name = "httparse"version = "1.10.1"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"672 unmodified linesExpand allsource = "registry+https://github.com/rust-lang/crates.io-index"checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
[[package]]name = "mime_guess"version = "2.0.5"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e"dependencies = [ "mime", "unicase",]
[[package]]name = "miniz_oxide"version = "0.8.9"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"1493 unmodified linesExpand all "futures-util", "http", "http-body", "http-body-util", "http-range-header", "httpdate", "iri-string", "mime", "mime_guess", "percent-encoding", "pin-project-lite", "tokio", "tokio-util", "tower",105 unmodified linesExpand allsource = "registry+https://github.com/rust-lang/crates.io-index"checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de"
[[package]]name = "unicase"version = "2.9.0"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142"
[[package]]name = "unicode-ident"version = "1.0.24"source = "registry+https://github.com/rust-lang/crates.io-index"checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"633 unmodified linesExpand all13 unmodified lines1415161718181920212226 unmodified lines4950515253535455565744 unmodified lines1021031041051061071081091101111121139 unmodified lines13 unmodified lines__persist_sync_all = []backlinks = []relay = []indexer = []indexer_stream = ["indexer"]indexer_stream = ["indexer", "dep:zstd"]jetstream = ["dep:zstd"]user-keyspace = []firehose-diagnostics = []
26 unmodified lines "json", "query", "rustls", "stream", "http2", "charset", "system-proxy", "gzip", "brotli", "deflate", "zstd",] }axum = { version = "0.8.8", features = ["macros"] }tower-http = { version = "0.6.6", features = ["cors", "trace"] }tower-http = { version = "0.6.6", features = ["cors", "fs", "trace"] }
jacquard-common = { git = "https://tangled.org/nonbinary.computer/jacquard", rev = "9046f2822f4685217300c4a03551b4484fcbc31d", default-features = false, features = ["std", "crypto"] }hickory-resolver = { version = "0.26.2", features = ["system-config", "tokio"], default-features = false }chrono = { version = "0.4.43", features = ["serde"] }44 unmodified lines[[example]]name = "statusphere"required-features = ["indexer_stream"]
[[example]]name = "chunk_load"required-features = ["indexer_stream"]
[profile.dev]opt-level = 1
[profile.bench]9 unmodified lines131 unmodified linesExpand all13213313413513613713813914031 unmodified lines131 unmodified linesExpand all| `THROTTLED_STREAMS` | `jetstream,relay` | the streams whose websocket subscribers have their replay from a cursor capped, comma separated: `indexer` (`/stream`), `jetstream` (`/subscribe`), `relay` (`subscribeRepos`). empty caps none. the live tail and catching up after falling behind it are never capped, and neither are in-process subscribers (`Hydrant::subscribe` and friends) || `STREAM_REPLAY_SPEED` | `10` | how fast a capped replay may go, as a multiple of what keeping up with the live stream costs that subscriber: new events a second (averaged over about 10 seconds) times the bytes its replay sends per event read. events a Jetstream filter drops count 32 bytes each, about what reading them costs in cpu next to sending one. at least `1` || `STREAM_REPLAY_FLOOR_BYTES` | `10000000` | bytes a second a capped replay may always use, however quiet the live stream is. the default is about 80 Mbit/s per replaying subscriber, some 15k Jetstream events a second. each subscriber gets its own budget, so many replaying at once are bounded by how many connect, not by this || `STREAM_PENDING_EVENT_LIMIT` | `4096` | live events a subscriber can hold in memory while its output queue is full. past it, events wait in the shared live buffer (512), and once that laps the subscriber catches up from the database instead. on `/stream` that catch-up skips the `identity` and `account` events it fell behind on, since those are never stored. higher means more memory per slow subscriber and fewer catch-ups. `0` acts like `1` || `STREAM_CHUNK_SPAN` | `0` (off) | ids per sealed `/stream` chunk, see [chunks](api/stream.md#chunks). hydrant seals a span once every id in it is committed, so the stream after the last chunk is at most this many ids long. chunks are merged into bigger ones later. ephemeral databases and ones with `HISTORY_TTL` build none, and starting without chunks deletes the ones built before || `STREAM_SEND_TIMEOUT` | `30sec` | how long a subscriber may block delivery (a full queue, or a websocket write of up to 64 frames or 64 KiB at once that doesn't finish) before it gets a consumer-too-slow error and is disconnected. a stuck subscriber keeps its thread and its buffers until then. `0s` disconnects anyone who falls even a little behind |
## database
31 unmodified lines333 unmodified linesExpand all33433533633733833934034134234395 unmodified lines439440441442443444445446447361 unmodified linesExpand all80981081181281381481581681781881982082182 unmodified lines333 unmodified linesExpand all /// set to a non-zero value to fix the batch size manually. /// /// set via `HYDRANT_STREAM_REPLAY_CHUNK_SIZE`. pub stream_replay_chunk_size: usize, /// set via `HYDRANT_STREAM_CHUNK_SPAN`. pub stream_chunk_span: u64, /// the streams whose websocket readers have their cursor replay capped. in-process /// subscribers are never capped, and neither is anyone's live tail. /// set via `HYDRANT_THROTTLED_STREAMS`, comma separated, empty for none. pub throttled_streams: Vec<StreamKind>,95 unmodified lines // smaller and a sync spends a lot more cpu compacting, see docs/configuration.md db_records_memtable_size_mb: BASE_MEMTABLE_MB, db_records_bloom_filters: false, stream_replay_chunk_size: 0, stream_chunk_span: 0, throttled_streams: DEFAULT_THROTTLED_STREAMS.to_vec(), stream_replay_speed: 10, // about 80 Mbit/s, or some 15k jetstream events a second stream_replay_floor_bytes: 10_000_000,361 unmodified linesExpand all } else { self.stream_replay_chunk_size.to_string() }; config_line!(f, "stream replay chunk", replay_chunk)?; if self.stream_chunk_span == 0 { config_line!(f, "stream chunks", "off")?; } else { config_line!(f, "stream chunk span", self.stream_chunk_span)?; } if self.throttled_streams.is_empty() { config_line!(f, "throttled streams", "none")?; } else { let streams: Vec<_> = self82 unmodified lines292 unmodified linesExpand all293294295296297298299300301302303304305306307308309310337 unmodified linesExpand all292 unmodified linesExpand all .get() .clone() }
#[cfg(feature = "indexer_stream")] pub(crate) fn firehose_backlog(&self) -> usize { let mut backlog = 0; self.firehose_trackers.iter_sync(|_, tracker| { backlog += tracker.pending(); true }); backlog }
pub(crate) async fn identity_reconcile_is_cooled_down( &self, did: &Did, identity: &DocIdentity<'static>,337 unmodified linesExpand all