d6abac7c4b
A stream dropped once the relay has asked to close a WebTransport session would send its FIN or STOP_SENDING at once, ahead of the WT_CLOSE_SESSION capsule. In a browser on the build host, a page whose stream was stopped just before the close lost its renderer (reproduced on loopback: a page writing a stream the relay refused crashed in most runs). The binding now keeps such streams (WtSend, WtRecv hold the connection's Closing) until the QUIC connection has closed, at most ten seconds, when they end with nothing more on the wire; read_setup keeps the stream it refuses in `early`, so it too is dropped after the close. tests/webtransport_close.rs: a stream that breaks a rule before SETUP is refused with PROTOCOL_VIOLATION, and the client reads the close and no STOP_SENDING ahead of it. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1921 lines
91 KiB
Rust
1921 lines
91 KiB
Rust
//! One session at the relay, from SETUP to its summary, and every request it makes
|
|
//! (`draft-ietf-moq-transport-22` §6, §7, §9).
|
|
|
|
use crate::auth::{Resolved, Right, TokenCache};
|
|
use crate::cache::CachedObject;
|
|
use crate::door::Verdict;
|
|
use crate::dvr::{room_of, is_events_track, Dvr, Slot, Snapshot};
|
|
use crate::relay::{
|
|
now_unix_ms, signed_of, DownSub, NsEvent, NsSub, ObjectIn, PublishedNs, Relay, SessionCtx, SubParams, TrackKey, Upstream,
|
|
UpstreamKind, DEFAULT_PRIORITY,
|
|
};
|
|
use crate::sched::{self, Item, LaneKey, LaneKind, ObjMeta, Op, Prio, SchedHandle};
|
|
use crate::summary::{RelayEnergy, SessionSummary, SignedSessionSummary, Tally, FORMAT};
|
|
use crate::transport::{BoxRecv, BoxSend, SharedConnection, TransportError, TransportKind};
|
|
use crate::wire::{open_request, responder, MessageReader, ReqRx, ReqTx, StreamError, UniReader};
|
|
use joule_moq::carriage::{read_object_properties, SETUP_OPTION_JWP_RECEIPTS};
|
|
use bytes::Bytes;
|
|
use joule_moq::draft22::data::{Datagram, FetchEntry, FetchObject, ObjectStatus, RangeEnd};
|
|
use joule_moq::draft22::kvp::{opt, prop, KeyValuePair, Properties, SetupOptions};
|
|
use joule_moq::draft22::names::{is_reserved, prefix_matches, prefixes_overlap, SESSION_NAMESPACE};
|
|
use joule_moq::draft22::params::param;
|
|
use joule_moq::draft22::{
|
|
Consequence, ControlMessage, FetchOk, Goaway, GroupOrder, Location, LocationFilter, NamespaceRequest, Parameter, Parameters, PeerRequestIds,
|
|
Perspective, Publish, PublishDoneCode, RequestError, RequestErrorCode, RequestIds, RequestKind, RequestOk, SessionError, Setup, StreamHeader,
|
|
StreamResetCode, SubscribeOk, Token, TrackRequest, UniEvent,
|
|
};
|
|
use joule_moq::messages::ObjectHeader;
|
|
use joule_moq::grant_proof;
|
|
use joule_moq::session::MoqSession;
|
|
use joule_moq::{GroupId, ObjectId, TrackAlias, TrackNamespace, JWP_MOQ_EXTENSION_VERSION};
|
|
use std::collections::HashSet;
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
use tokio::sync::{mpsc, oneshot, watch};
|
|
use tokio::time::{timeout, Instant};
|
|
|
|
/// How long a peer has to open its control stream, and a request stream to state its
|
|
/// request (§6.3).
|
|
const SETUP_TIMEOUT: Duration = Duration::from_secs(10);
|
|
/// Data streams that arrive before the control stream, held until SETUP (§6.3).
|
|
const EARLY_STREAMS: usize = 64;
|
|
|
|
/// The implementation the relay names in SETUP (§9.1.5).
|
|
pub fn implementation() -> String {
|
|
format!("joule-moq-relay/{}", env!("CARGO_PKG_VERSION"))
|
|
}
|
|
|
|
fn params_of(m: &ControlMessage) -> Option<&Parameters> {
|
|
match m {
|
|
ControlMessage::Subscribe(r) | ControlMessage::Fetch(r) | ControlMessage::TrackStatus(r) => Some(&r.params),
|
|
ControlMessage::Publish(p) => Some(&p.params),
|
|
ControlMessage::PublishNamespace(r) | ControlMessage::SubscribeNamespace(r) | ControlMessage::SubscribeTracks(r) => Some(&r.params),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
fn tokens_in(p: &Parameters) -> Vec<Token> {
|
|
p.0.iter()
|
|
.filter_map(|x| match x {
|
|
Parameter::AuthorizationToken(t) => Some(t.clone()),
|
|
_ => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn varint_param(p: &Parameters, ty: u64) -> Option<u64> {
|
|
p.0.iter().find_map(|x| match (x.type_code() == ty, x) {
|
|
(true, Parameter::RendezvousTimeout(v)) | (true, Parameter::SubgroupDeliveryTimeout(v)) | (true, Parameter::FillTimeout(v)) => Some(*v),
|
|
_ => None,
|
|
})
|
|
}
|
|
|
|
fn is_session_namespace(ns: &TrackNamespace) -> bool {
|
|
ns.0.first().is_some_and(|f| f.as_slice() == SESSION_NAMESPACE)
|
|
}
|
|
|
|
/// What a wire error means for the session.
|
|
fn fail(ctx: &SessionCtx, e: &StreamError) {
|
|
if let StreamError::Wire(w) = e {
|
|
match w.consequence() {
|
|
Consequence::CloseSession(code) => ctx.close(code.0, &w.to_string()),
|
|
_ => ctx.close(SessionError::PROTOCOL_VIOLATION.0, &w.to_string()),
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn reject(tx: &ReqTx, code: RequestErrorCode, reason: &str) {
|
|
let e = ControlMessage::RequestError(RequestError { code, retry_interval: 0, reason: reason.as_bytes().to_vec(), redirect: None });
|
|
let _ = tx.send(&e).await;
|
|
tx.fin().await;
|
|
}
|
|
|
|
/// A Subscription's start and end from its Location Filter, given the Largest Object
|
|
/// (§3.3.1, §9.20.9).
|
|
pub fn subscription_range(f: Option<LocationFilter>, largest: Option<Location>) -> (Location, Option<Location>) {
|
|
match f.unwrap_or(LocationFilter::None) {
|
|
LocationFilter::None => (Location::new(0, 0), None),
|
|
LocationFilter::NextObject => (largest.map_or(Location::new(0, 0), |l| Location::new(l.group, l.object.saturating_add(1))), None),
|
|
LocationFilter::RelativeStart { start_group } => (Location::new(LocationFilter::relative_start(start_group, largest.map(|l| l.group)), 0), None),
|
|
LocationFilter::AbsoluteStart { start } => (start, None),
|
|
LocationFilter::AbsoluteStartGroupEnd { start, end_group_delta } => (start, Some(Location::new(start.group.saturating_add(end_group_delta), u64::MAX))),
|
|
LocationFilter::AbsoluteRange { start, end_group_delta, end_object } => (start, Some(Location::new(start.group.saturating_add(end_group_delta), end_object))),
|
|
}
|
|
}
|
|
|
|
/// A fetch's range from its Location Filter (§3.2, §9.20.9): never past the Largest
|
|
/// Object. `None` when it starts after it (INVALID_RANGE).
|
|
pub fn fetch_range(f: Option<LocationFilter>, largest: Location) -> Option<(Location, Location)> {
|
|
let (start, end) = match f.unwrap_or(LocationFilter::None) {
|
|
LocationFilter::None => (Location::new(0, 0), largest),
|
|
LocationFilter::NextObject => return None,
|
|
LocationFilter::RelativeStart { start_group } => (Location::new(LocationFilter::relative_start(start_group, Some(largest.group)), 0), largest),
|
|
LocationFilter::AbsoluteStart { start } => (start, largest),
|
|
LocationFilter::AbsoluteStartGroupEnd { start, end_group_delta } => (start, Location::new(start.group.saturating_add(end_group_delta), u64::MAX).min(largest)),
|
|
LocationFilter::AbsoluteRange { start, end_group_delta, end_object } => (start, Location::new(start.group.saturating_add(end_group_delta), end_object).min(largest)),
|
|
};
|
|
(start <= largest && start <= end).then_some((start, end))
|
|
}
|
|
|
|
// ── The session ────────────────────────────────────────────────────────────
|
|
|
|
/// The client's SETUP, from its control stream; data streams that come first wait in
|
|
/// `early`. A stream that breaks a rule is put there too, so that it ends after the
|
|
/// session's close, not before it (`transport::webtransport` holds what is dropped then).
|
|
async fn read_setup(conn: &SharedConnection, max_payload: u64, early: &mut Vec<UniReader>) -> Result<(Setup, UniReader), String> {
|
|
loop {
|
|
let recv = conn.accept_uni().await.map_err(|e| e.to_string())?;
|
|
let mut r = UniReader::new(recv, max_payload);
|
|
let first = match r.start().await.map_err(|e| e.to_string()) {
|
|
Ok(StreamHeader::Control) => r.next().await.map_err(|e| e.to_string()),
|
|
Ok(_) if early.len() < EARLY_STREAMS => {
|
|
early.push(r);
|
|
continue;
|
|
}
|
|
Ok(_) => Err("too many data streams before SETUP".into()),
|
|
Err(e) => Err(e),
|
|
};
|
|
return match first {
|
|
Ok(Some(UniEvent::Control(ControlMessage::Setup(s)))) => Ok((s, r)),
|
|
Ok(_) => {
|
|
early.push(r);
|
|
Err("the control stream does not begin with SETUP".into())
|
|
}
|
|
Err(e) => {
|
|
early.push(r);
|
|
Err(e)
|
|
}
|
|
};
|
|
}
|
|
}
|
|
|
|
/// Run one session to its end.
|
|
pub async fn serve(relay: Relay, conn: SharedConnection) {
|
|
let cfg = relay.config().clone();
|
|
let id = relay.next_id();
|
|
relay.inner.stats.sessions_opened.fetch_add(1, Ordering::Relaxed);
|
|
let opened = now_unix_ms();
|
|
if relay.inner.state.lock().unwrap().sessions.len() >= cfg.limits.max_sessions {
|
|
conn.close(SessionError::INTERNAL_ERROR.0, b"the relay is at its session limit");
|
|
return;
|
|
}
|
|
// The relay's SETUP goes first, at once (MOQT §6.3: either endpoint may send first): it
|
|
// depends on nothing of the client's, and it carries the session's challenge, which the
|
|
// client's grants' proofs sign (WAI live-rooms §10.5.2). The receipt extension is
|
|
// declared whatever the client says; it is on when both declare it (§6.3.2).
|
|
let mut challenge = [0u8; 32];
|
|
rand_core_06::RngCore::fill_bytes(&mut rand_core_06::OsRng, &mut challenge);
|
|
let mut o = SetupOptions::new();
|
|
o.push(KeyValuePair::bytes(opt::MOQT_IMPLEMENTATION, implementation().into_bytes()));
|
|
if cfg.limits.max_request_updates != 0 {
|
|
o.push(KeyValuePair::varint(opt::MAX_REQUEST_UPDATES, cfg.limits.max_request_updates));
|
|
}
|
|
if cfg.limits.max_auth_token_cache_size != 0 {
|
|
o.push(KeyValuePair::varint(opt::MAX_AUTH_TOKEN_CACHE_SIZE, cfg.limits.max_auth_token_cache_size));
|
|
}
|
|
o.push(KeyValuePair::varint(SETUP_OPTION_JWP_RECEIPTS, JWP_MOQ_EXTENSION_VERSION as u64));
|
|
grant_proof::add_challenge(&mut o, challenge);
|
|
let ours = match ControlMessage::Setup(Setup { options: o }).encode() {
|
|
Ok(b) => b,
|
|
Err(_) => return,
|
|
};
|
|
let control = match conn.open_uni().await {
|
|
Ok(mut s) => {
|
|
s.set_priority(i32::MAX);
|
|
if s.write_all(&ours).await.is_err() {
|
|
return;
|
|
}
|
|
s
|
|
}
|
|
Err(_) => return,
|
|
};
|
|
let mut early = Vec::new();
|
|
let (setup, ctl_rx) = match timeout(SETUP_TIMEOUT, read_setup(&conn, cfg.limits.max_object_bytes, &mut early)).await {
|
|
Ok(Ok(x)) => x,
|
|
Ok(Err(why)) => {
|
|
conn.close(SessionError::PROTOCOL_VIOLATION.0, why.as_bytes());
|
|
return;
|
|
}
|
|
Err(_) => {
|
|
conn.close(SessionError::CONTROL_MESSAGE_TIMEOUT.0, b"no SETUP");
|
|
return;
|
|
}
|
|
};
|
|
let over_wt = conn.kind() == TransportKind::WebTransport;
|
|
if let Err(code) = setup.check_from(Perspective::Client, over_wt) {
|
|
conn.close(code.0, b"the client's SETUP breaks a rule for its end");
|
|
return;
|
|
}
|
|
let mut tokens = TokenCache::new(cfg.limits.max_auth_token_cache_size);
|
|
let setup_tokens = match tokens.setup(&setup.options.authorization_tokens()) {
|
|
Ok(t) => t,
|
|
Err(code) => {
|
|
conn.close(code.0, b"an authorization token in SETUP");
|
|
return;
|
|
}
|
|
};
|
|
let jwp = MoqSession::declares_jwp(&setup);
|
|
// Every room grant the SETUP presents is proved before any request is served (WAI
|
|
// live-rooms §10.5.2, §10.5.4); a request that arrived first waits in `early` or on its
|
|
// stream. A grant without a proof that passes closes the session: UNAUTHORIZED (§12.2).
|
|
let target = conn.target().unwrap_or_else(|| {
|
|
let s = |b: Option<&[u8]>| String::from_utf8_lossy(b.unwrap_or_default()).into_owned();
|
|
let path = s(setup.options.path());
|
|
format!("//{}{}", s(setup.options.authority()), path.split('?').next().unwrap_or(""))
|
|
});
|
|
let context = format!("{} {target}", grant_proof::VERB_SETUP);
|
|
let proofs = grant_proof::proofs_of(&setup.options);
|
|
let mut proven = HashSet::new();
|
|
for (ty, value) in &setup_tokens {
|
|
if *ty != 0 || !relay.inner.auth.requires_proof(value) {
|
|
continue;
|
|
}
|
|
let checked = match grant_proof::grant_digest_and_principal(value) {
|
|
None => Err("the token its rule names is not a room grant's record".to_string()),
|
|
Some((digest, principal)) => {
|
|
let proof = grant_proof::proof_for(&proofs, &digest).or(if proofs.len() == 1 { Some(proofs[0]) } else { None });
|
|
grant_proof::check_proof(&digest, &principal, proof, &challenge, |_| false, &context).map_err(|r| r.code().to_string())
|
|
}
|
|
};
|
|
match checked {
|
|
Ok(_) => {
|
|
proven.insert(crate::auth::sha256(value));
|
|
}
|
|
Err(why) => {
|
|
tracing::info!(session = id, remote = %conn.remote(), %context, reason = %why, "a grant presented without a proof that passes: the session is refused");
|
|
conn.close(SessionError::UNAUTHORIZED.0, format!("live-rooms §10.5: {why}").as_bytes());
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
// When the session must end: at the relay's session limit, or when the earliest room
|
|
// grant an authority signed that it proved expires (live-rooms §10.2: with no
|
|
// re-validation, expiry alone ends a token).
|
|
let mut ends: Option<(Instant, u64, String)> = None;
|
|
let mut end_at = |at: Instant, code: SessionError, why: String| {
|
|
if ends.as_ref().is_none_or(|(t, _, _)| at < *t) {
|
|
ends = Some((at, code.0, why));
|
|
}
|
|
};
|
|
let unlimited = relay.inner.auth.unlimited(&setup_tokens, &proven);
|
|
if cfg.limits.max_session_ms != 0 && !unlimited {
|
|
end_at(Instant::now() + Duration::from_millis(cfg.limits.max_session_ms), SessionError::NO_ERROR, format!("the relay's session limit of {} s", cfg.limits.max_session_ms / 1000));
|
|
}
|
|
for (_, value) in setup_tokens.iter().filter(|(ty, _)| *ty == 0) {
|
|
if let Some(g) = relay.inner.auth.authority_grant(value).filter(|_| proven.contains(&crate::auth::sha256(value))) {
|
|
let left = g.exp.saturating_mul(1000).saturating_sub(opened);
|
|
end_at(Instant::now() + Duration::from_millis(left), SessionError::EXPIRED_AUTH_TOKEN, "the room grant expired".into());
|
|
}
|
|
}
|
|
let opening = Tally { room_grant: relay.metered_grant(&setup_tokens), ..Tally::default() };
|
|
let tally = Arc::new(Mutex::new(opening));
|
|
let sched = sched::spawn(conn.clone(), cfg.limits.write_concurrency, tally.clone());
|
|
let ctx = Arc::new(SessionCtx {
|
|
id,
|
|
conn: conn.clone(),
|
|
kind: conn.kind(),
|
|
remote: conn.remote(),
|
|
sched,
|
|
tally,
|
|
peer_ids: Mutex::new(PeerRequestIds::new(Perspective::Client)),
|
|
my_ids: Mutex::new(RequestIds::new(Perspective::Server)),
|
|
tokens: Mutex::new(tokens),
|
|
setup_tokens,
|
|
proven,
|
|
peer_max_updates: setup.options.max_request_updates(),
|
|
my_max_updates: cfg.limits.max_request_updates,
|
|
control: tokio::sync::Mutex::new(Some(control)),
|
|
goaway_sent: AtomicBool::new(false),
|
|
pending_fetch: Mutex::new(Default::default()),
|
|
opened_unix_ms: opened,
|
|
close_code: Mutex::new(None),
|
|
jwp,
|
|
ingress: (cfg.limits.max_ingress_kbps != 0 && !unlimited).then(|| Mutex::new(crate::relay::Ingress::new(cfg.limits.max_ingress_kbps, cfg.limits.ingress_burst_ms))),
|
|
});
|
|
relay.inner.state.lock().unwrap().sessions.insert(id, ctx.clone());
|
|
relay.inner.change.notify_waiters();
|
|
if relay.going_away() {
|
|
send_goaway(&ctx, cfg.goaway_uri.as_bytes(), cfg.limits.goaway_timeout_ms).await;
|
|
}
|
|
|
|
let mut tasks = vec![
|
|
tokio::spawn(control_reader(ctx.clone(), ctl_rx)),
|
|
tokio::spawn(bi_loop(relay.clone(), ctx.clone())),
|
|
tokio::spawn(uni_loop(relay.clone(), ctx.clone(), early)),
|
|
tokio::spawn(datagram_loop(relay.clone(), ctx.clone())),
|
|
];
|
|
if let Some((at, code, why)) = ends {
|
|
let c = ctx.clone();
|
|
tasks.push(tokio::spawn(async move {
|
|
tokio::time::sleep_until(at.into()).await;
|
|
tracing::info!(session = c.id, remote = %c.remote, reason = %why, "a session at its limit: closed");
|
|
c.close(code, &why);
|
|
}));
|
|
}
|
|
let why = conn.closed().await;
|
|
for t in tasks {
|
|
t.abort();
|
|
}
|
|
cleanup(&relay, &ctx, why).await;
|
|
}
|
|
|
|
/// The rest of the peer's control stream: at most a GOAWAY. Its end is a protocol
|
|
/// violation (§6.3).
|
|
async fn control_reader(ctx: Arc<SessionCtx>, mut r: UniReader) {
|
|
loop {
|
|
match r.next().await {
|
|
Ok(Some(UniEvent::Control(ControlMessage::Goaway(g)))) => {
|
|
if g.check_from(Perspective::Client).is_err() {
|
|
ctx.close(SessionError::PROTOCOL_VIOLATION.0, "a client's GOAWAY names a URI");
|
|
return;
|
|
}
|
|
tracing::debug!(session = ctx.id, "the client sent GOAWAY");
|
|
}
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => {
|
|
ctx.close(SessionError::PROTOCOL_VIOLATION.0, "the control stream closed");
|
|
return;
|
|
}
|
|
Err(StreamError::Wire(e)) => {
|
|
fail(&ctx, &StreamError::Wire(e));
|
|
return;
|
|
}
|
|
Err(StreamError::Transport(_)) => return,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Send GOAWAY on a session's control stream, once (§9.2).
|
|
pub async fn send_goaway(ctx: &SessionCtx, uri: &[u8], timeout_ms: u64) {
|
|
if ctx.goaway_sent.swap(true, Ordering::AcqRel) {
|
|
return;
|
|
}
|
|
let m = ControlMessage::Goaway(Goaway { new_session_uri: uri.to_vec(), timeout_ms });
|
|
if let Ok(b) = m.encode()
|
|
&& let Some(s) = ctx.control.lock().await.as_mut()
|
|
{
|
|
let _ = s.write_all(&b).await;
|
|
}
|
|
}
|
|
|
|
async fn bi_loop(relay: Relay, ctx: Arc<SessionCtx>) {
|
|
while let Ok((send, recv)) = ctx.conn.accept_bi().await {
|
|
tokio::spawn(request(relay.clone(), ctx.clone(), send, recv));
|
|
}
|
|
}
|
|
|
|
async fn uni_loop(relay: Relay, ctx: Arc<SessionCtx>, early: Vec<UniReader>) {
|
|
for r in early {
|
|
tokio::spawn(data_stream(relay.clone(), ctx.clone(), r));
|
|
}
|
|
let max = relay.config().limits.max_object_bytes;
|
|
while let Ok(recv) = ctx.conn.accept_uni().await {
|
|
tokio::spawn(data_stream(relay.clone(), ctx.clone(), UniReader::new(recv, max)));
|
|
}
|
|
}
|
|
|
|
async fn datagram_loop(relay: Relay, ctx: Arc<SessionCtx>) {
|
|
while let Ok(b) = ctx.conn.recv_datagram().await {
|
|
let arrived = Instant::now();
|
|
let d = match Datagram::decode(&b) {
|
|
Ok(Datagram::Object(d)) => d,
|
|
Ok(Datagram::Padding(_)) => continue,
|
|
Err(e) => {
|
|
fail(&ctx, &StreamError::Wire(e));
|
|
return;
|
|
}
|
|
};
|
|
let key = match relay.inner.state.lock().unwrap().aliases.get(&(ctx.id, d.track_alias)) {
|
|
Some(k) => k.clone(),
|
|
// §3.1.3.1: a datagram whose alias is not yet known MAY be dropped.
|
|
None => continue,
|
|
};
|
|
let len = match &d.content {
|
|
joule_moq::draft22::data::DatagramContent::Payload(p) => p.len() as u64,
|
|
_ => 0,
|
|
};
|
|
{
|
|
let mut t = ctx.tally.lock().unwrap();
|
|
t.publisher.objects_received += 1;
|
|
t.publisher.object_bytes_received += len;
|
|
let tr = t.published(&joule_moq::draft22::names::render(&key.0, Some(&key.1)));
|
|
tr.objects_received += 1;
|
|
tr.object_bytes_received += len;
|
|
}
|
|
if !ctx.take_ingress(len, relay.config().limits.max_ingress_kbps) {
|
|
return;
|
|
}
|
|
relay.inner.stats.objects_in.fetch_add(1, Ordering::Relaxed);
|
|
if relay.inner.door.checks(&key.0) && relay.inner.door.media_of(&key.1).is_none() {
|
|
let relay = relay.clone();
|
|
let ctx = ctx.clone();
|
|
tokio::spawn(async move {
|
|
let payload = match &d.content {
|
|
joule_moq::draft22::data::DatagramContent::Payload(p) => p.clone(),
|
|
_ => Vec::new(),
|
|
};
|
|
let v = relay.inner.door.object(&key.0, &key.1, d.group_id, d.object_id, &payload, None).await;
|
|
if let Verdict::Refuse(why) = &v {
|
|
refused(&relay, &ctx, &key, d.group_id, d.object_id, why);
|
|
return;
|
|
}
|
|
relay.fan_out_datagram(&key, ctx.id, &d, arrived, signed_of(&v));
|
|
});
|
|
} else {
|
|
relay.fan_out_datagram(&key, ctx.id, &d, arrived, None);
|
|
}
|
|
}
|
|
}
|
|
|
|
fn refused(relay: &Relay, ctx: &SessionCtx, key: &TrackKey, group: u64, object: u64, why: &str) {
|
|
relay.inner.stats.refused_at_door.fetch_add(1, Ordering::Relaxed);
|
|
{
|
|
let mut t = ctx.tally.lock().unwrap();
|
|
t.publisher.refused_at_door += 1;
|
|
t.published(&joule_moq::draft22::names::render(&key.0, Some(&key.1))).refused_at_door += 1;
|
|
}
|
|
tracing::warn!(
|
|
target: "joule_moq_relay::door",
|
|
session = ctx.id,
|
|
track = %joule_moq::draft22::names::render(&key.0, Some(&key.1)),
|
|
group,
|
|
object,
|
|
reason = why,
|
|
"refused at the door"
|
|
);
|
|
}
|
|
|
|
/// One unidirectional stream from the peer.
|
|
async fn data_stream(relay: Relay, ctx: Arc<SessionCtx>, mut r: UniReader) {
|
|
let header = match timeout(SETUP_TIMEOUT, r.start()).await {
|
|
Ok(Ok(h)) => h,
|
|
Ok(Err(e)) => {
|
|
fail(&ctx, &e);
|
|
return;
|
|
}
|
|
Err(_) => {
|
|
r.stop(StreamResetCode::CANCELLED.0);
|
|
return;
|
|
}
|
|
};
|
|
match header {
|
|
StreamHeader::Subgroup(h) => upstream_subgroup(relay, ctx, h, r).await,
|
|
StreamHeader::Fetch(fh) => {
|
|
let waiting = ctx.pending_fetch.lock().unwrap().remove(&fh.request_id);
|
|
match waiting {
|
|
Some(tx) => {
|
|
let _ = tx.send(r);
|
|
}
|
|
None => r.stop(StreamResetCode::CANCELLED.0),
|
|
}
|
|
}
|
|
StreamHeader::Padding => {
|
|
// Read and discarded (§11.5.1), each read bounded.
|
|
while let Ok(Ok(Some(_))) = timeout(SETUP_TIMEOUT, r.next()).await {}
|
|
}
|
|
StreamHeader::Control => ctx.close(SessionError::PROTOCOL_VIOLATION.0, "a second control stream"),
|
|
}
|
|
}
|
|
|
|
/// Closes the cache's count of an upstream subgroup stream however the stream's reading
|
|
/// ends: cleanly only at its FIN.
|
|
struct StreamCount<'a> {
|
|
relay: &'a Relay,
|
|
key: &'a TrackKey,
|
|
group: u64,
|
|
counted: bool,
|
|
clean: bool,
|
|
}
|
|
|
|
impl Drop for StreamCount<'_> {
|
|
fn drop(&mut self) {
|
|
if self.counted {
|
|
self.relay.cache_stream_ended(self.key, self.group, self.clean);
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn upstream_subgroup(relay: Relay, ctx: Arc<SessionCtx>, h: joule_moq::draft22::SubgroupHeader, mut r: UniReader) {
|
|
let wait = Duration::from_millis(relay.config().limits.unknown_alias_wait_ms);
|
|
let Some(key) = relay.resolve_alias(ctx.id, h.track_alias, wait).await else {
|
|
r.stop(StreamResetCode::CANCELLED.0);
|
|
return;
|
|
};
|
|
let door_on = relay.inner.door.checks(&key.0);
|
|
let media_of = relay.inner.door.media_of(&key.1);
|
|
let rendered = joule_moq::draft22::names::render(&key.0, Some(&key.1));
|
|
let (mut object_timeout, mut subgroup_timeout) = {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
st.tracks.get(&key).map_or((0, 0), |t| (t.pub_object_timeout, t.pub_subgroup_timeout))
|
|
};
|
|
// A room's events track the DVR store keeps: its decisions are read as they pass, and
|
|
// the moderators among this publisher's grants are the ones it honours (dvr.rs).
|
|
let events_dvr = relay.dvr().filter(|d| is_events_track(&key) && d.keeps(&key.0)).cloned();
|
|
let moderators: Vec<[u8; 32]> = match &events_dvr {
|
|
Some(_) => relay.inner.auth.proven_grants(&ctx.setup_tokens, &ctx.proven).into_iter().filter(|g| g.moderates() && Some(g.room) == room_of(&key.0)).map(|g| g.principal).collect(),
|
|
None => Vec::new(),
|
|
};
|
|
let mut started = HashSet::new();
|
|
let mut first: Option<u64> = None;
|
|
let mut subgroup_id: Option<u64> = None;
|
|
let mut count = StreamCount { relay: &relay, key: &key, group: h.group_id, counted: false, clean: false };
|
|
loop {
|
|
match r.next().await {
|
|
Ok(Some(UniEvent::SubgroupObject(o))) => {
|
|
let arrived = Instant::now();
|
|
if first.is_none() {
|
|
first = Some(o.object_id);
|
|
subgroup_id = Some(h.resolve_subgroup_id(o.object_id));
|
|
// Either timeout set on the subgroup's first object overrides the
|
|
// track's for this subgroup (§5.2).
|
|
let v = |ty| o.properties.find(ty).ok().flatten().and_then(|k| k.as_varint());
|
|
if let Some(t) = v(prop::OBJECT_DELIVERY_TIMEOUT) {
|
|
object_timeout = t;
|
|
}
|
|
if let Some(t) = v(prop::SUBGROUP_DELIVERY_TIMEOUT) {
|
|
subgroup_timeout = t;
|
|
}
|
|
}
|
|
let sid = subgroup_id.unwrap_or(0);
|
|
{
|
|
let mut t = ctx.tally.lock().unwrap();
|
|
t.publisher.objects_received += 1;
|
|
t.publisher.object_bytes_received += o.payload.len() as u64;
|
|
let tr = t.published(&rendered);
|
|
tr.objects_received += 1;
|
|
tr.object_bytes_received += o.payload.len() as u64;
|
|
}
|
|
if !ctx.take_ingress(o.payload.len() as u64, relay.config().limits.max_ingress_kbps) {
|
|
if let Some(sid) = subgroup_id {
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(StreamResetCode::SESSION_CLOSED.0));
|
|
}
|
|
return;
|
|
}
|
|
relay.inner.stats.objects_in.fetch_add(1, Ordering::Relaxed);
|
|
let mut signed = None;
|
|
if door_on && o.status == ObjectStatus::Normal {
|
|
if let Some(media) = &media_of {
|
|
if let Err(why) = relay.inner.door.bundle(&key.0, media, h.group_id, &o.payload) {
|
|
relay.inner.stats.groups_dropped_at_door.fetch_add(1, Ordering::Relaxed);
|
|
refused(&relay, &ctx, &key, h.group_id, o.object_id, &why);
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(StreamResetCode::MALFORMED_TRACK.0));
|
|
r.stop(StreamResetCode::CANCELLED.0);
|
|
return;
|
|
}
|
|
} else {
|
|
let mut hh = ObjectHeader {
|
|
track_alias: TrackAlias(h.track_alias),
|
|
group_id: GroupId(h.group_id),
|
|
object_id: ObjectId(o.object_id),
|
|
publisher_priority: 0,
|
|
status: joule_moq::ObjectStatus::Normal,
|
|
payload_len: 0,
|
|
mrl_layer: 0,
|
|
mrl_total_layers: 0,
|
|
wai_envelope: false,
|
|
};
|
|
let property = read_object_properties(&o.properties, &mut hh).ok().and_then(|x| x.receipt);
|
|
let v = relay.inner.door.object(&key.0, &key.1, h.group_id, o.object_id, &o.payload, property.as_ref()).await;
|
|
if let Verdict::Refuse(why) = &v {
|
|
// The rest of a multi-object stream is not forwarded around a
|
|
// hole: its downstream streams are reset (§7.4).
|
|
refused(&relay, &ctx, &key, h.group_id, o.object_id, why);
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(StreamResetCode::MALFORMED_TRACK.0));
|
|
r.stop(StreamResetCode::CANCELLED.0);
|
|
return;
|
|
}
|
|
signed = signed_of(&v);
|
|
}
|
|
}
|
|
let is_first = first == Some(o.object_id);
|
|
let taken = relay.fan_out_object(
|
|
&key,
|
|
ObjectIn { upstream: ctx.id, header: &h, subgroup_id: sid, first_in_stream: first, object: &o, arrived, object_timeout, signed },
|
|
&mut started,
|
|
);
|
|
if is_first {
|
|
count.counted = taken;
|
|
}
|
|
if let Some(d) = &events_dvr {
|
|
d.events_object(&key.0, &o.payload, &moderators);
|
|
}
|
|
}
|
|
Ok(Some(_)) => {
|
|
ctx.close(SessionError::PROTOCOL_VIOLATION.0, "a subgroup stream carried something other than an object");
|
|
return;
|
|
}
|
|
Ok(None) => {
|
|
if let Some(sid) = subgroup_id {
|
|
relay.end_subgroup(&key, h.group_id, sid, subgroup_timeout, &started, None);
|
|
}
|
|
count.clean = true;
|
|
return;
|
|
}
|
|
Err(StreamError::Transport(TransportError::Reset(code))) => {
|
|
if let Some(sid) = subgroup_id {
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(code));
|
|
}
|
|
return;
|
|
}
|
|
Err(StreamError::Wire(e)) => {
|
|
if let Some(sid) = subgroup_id {
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(StreamResetCode::MALFORMED_TRACK.0));
|
|
}
|
|
if let Consequence::CloseSession(code) = e.consequence() {
|
|
ctx.close(code.0, &e.to_string());
|
|
} else {
|
|
r.stop(StreamResetCode::MALFORMED_TRACK.0);
|
|
}
|
|
return;
|
|
}
|
|
Err(StreamError::Transport(_)) => {
|
|
if let Some(sid) = subgroup_id {
|
|
relay.end_subgroup(&key, h.group_id, sid, 0, &started, Some(StreamResetCode::SESSION_CLOSED.0));
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── Requests ───────────────────────────────────────────────────────────────
|
|
|
|
async fn request(relay: Relay, ctx: Arc<SessionCtx>, mut send: BoxSend, recv: BoxRecv) {
|
|
send.set_priority(i32::MAX - 1);
|
|
let mut reader = MessageReader::new(recv);
|
|
let first = match timeout(SETUP_TIMEOUT, reader.next()).await {
|
|
Ok(Ok(Some(m))) => m,
|
|
Ok(Err(e)) => {
|
|
fail(&ctx, &e);
|
|
return;
|
|
}
|
|
_ => return,
|
|
};
|
|
if first.opens_request().is_none() {
|
|
ctx.close(SessionError::PROTOCOL_VIOLATION.0, "a request stream that does not begin with a request");
|
|
return;
|
|
}
|
|
if let Some(rid) = first.request_id()
|
|
&& ctx.peer_ids.lock().unwrap().accept(rid).is_err()
|
|
{
|
|
ctx.close(SessionError::INVALID_REQUEST_ID.0, "a Request ID of the wrong parity or seen before");
|
|
return;
|
|
}
|
|
let (tx, rx) = match responder(send, reader, &first, ctx.my_max_updates, ctx.peer_max_updates) {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
fail(&ctx, &StreamError::Wire(e));
|
|
return;
|
|
}
|
|
};
|
|
let mut tokens = ctx.setup_tokens.clone();
|
|
if let Some(p) = params_of(&first) {
|
|
match ctx.tokens.lock().unwrap().resolve(&tokens_in(p)) {
|
|
Ok(t) => tokens.extend(t),
|
|
Err(code) => {
|
|
ctx.close(code.0, "an authorization token");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
if relay.going_away() {
|
|
reject(&tx, RequestErrorCode::GOING_AWAY, "the relay is going away").await;
|
|
return;
|
|
}
|
|
match first {
|
|
ControlMessage::Subscribe(r) => subscribe(relay, ctx, r, tx, rx, tokens).await,
|
|
ControlMessage::Publish(p) => publish(relay, ctx, p, tx, rx, tokens).await,
|
|
ControlMessage::Fetch(r) => fetch(relay, ctx, r, tx, rx, tokens).await,
|
|
ControlMessage::TrackStatus(r) => track_status(relay, ctx, r, tx, tokens).await,
|
|
ControlMessage::PublishNamespace(r) => publish_namespace(relay, ctx, r, tx, rx, tokens).await,
|
|
ControlMessage::SubscribeNamespace(r) => subscribe_namespace(relay, ctx, r, tx, rx, tokens).await,
|
|
ControlMessage::SubscribeTracks(_) => reject(&tx, RequestErrorCode::NOT_SUPPORTED, "SUBSCRIBE_TRACKS is not implemented by this relay").await,
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
/// Make sure `key` has an Established upstream subscription (§7.4: "The relay MUST have
|
|
/// an Established upstream subscription before sending SUBSCRIBE_OK"), subscribing to
|
|
/// every publisher of a namespace it falls under, or waiting up to `rendezvous_ms` for
|
|
/// one to appear (§9.20.6).
|
|
async fn ensure_upstream(relay: &Relay, key: &TrackKey, requester: u64, rendezvous_ms: u64) -> Result<(), (RequestErrorCode, String)> {
|
|
let deadline = Instant::now() + Duration::from_millis(rendezvous_ms);
|
|
let upstream_timeout = Duration::from_millis(relay.config().limits.upstream_timeout_ms);
|
|
loop {
|
|
let notified = relay.inner.change.notified();
|
|
tokio::pin!(notified);
|
|
notified.as_mut().enable();
|
|
enum Next {
|
|
Ready,
|
|
Wait(crate::relay::UpstreamWait),
|
|
Subscribe(Vec<Arc<SessionCtx>>, watch::Sender<Option<Result<(), (u64, String)>>>),
|
|
Nobody,
|
|
}
|
|
let next = {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
if st.tracks.get(key).is_some_and(|t| !t.upstreams.is_empty()) {
|
|
Next::Ready
|
|
} else if let Some(w) = st.pending_upstream.get(key) {
|
|
Next::Wait(w.clone())
|
|
} else {
|
|
let pubs = relay.publishers_of(&st, &key.0, requester);
|
|
if pubs.is_empty() {
|
|
Next::Nobody
|
|
} else {
|
|
let (wtx, wrx) = watch::channel(None);
|
|
st.pending_upstream.insert(key.clone(), wrx);
|
|
Next::Subscribe(pubs, wtx)
|
|
}
|
|
}
|
|
};
|
|
match next {
|
|
Next::Ready => return Ok(()),
|
|
Next::Wait(mut w) => {
|
|
let r = timeout(upstream_timeout, async {
|
|
loop {
|
|
if let Some(r) = w.borrow_and_update().clone() {
|
|
return r;
|
|
}
|
|
if w.changed().await.is_err() {
|
|
return Err((RequestErrorCode::INTERNAL_ERROR.0, "the upstream request ended".into()));
|
|
}
|
|
}
|
|
})
|
|
.await;
|
|
match r {
|
|
Ok(Ok(())) => continue,
|
|
Ok(Err((c, why))) => return Err((RequestErrorCode(c), why)),
|
|
Err(_) => return Err((RequestErrorCode::TIMEOUT, "the upstream did not answer".into())),
|
|
}
|
|
}
|
|
Next::Subscribe(pubs, wtx) => {
|
|
let r = subscribe_upstream_any(relay, key, pubs).await;
|
|
relay.inner.state.lock().unwrap().pending_upstream.remove(key);
|
|
let _ = wtx.send(Some(r.clone()));
|
|
return r.map_err(|(c, w)| (RequestErrorCode(c), w));
|
|
}
|
|
Next::Nobody => {
|
|
if rendezvous_ms == 0 {
|
|
return Err((RequestErrorCode::DOES_NOT_EXIST, "no publisher of the track".into()));
|
|
}
|
|
if tokio::time::timeout_at(deadline, notified).await.is_err() {
|
|
return Err((RequestErrorCode::TIMEOUT, "no publisher within RENDEZVOUS_TIMEOUT".into()));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// SUBSCRIBE upstream to every publisher; done when the first is Established (the rest
|
|
/// join the track as they answer).
|
|
async fn subscribe_upstream_any(relay: &Relay, key: &TrackKey, pubs: Vec<Arc<SessionCtx>>) -> Result<(), (u64, String)> {
|
|
let (rtx, mut rrx) = mpsc::unbounded_channel();
|
|
let n = pubs.len();
|
|
for p in pubs {
|
|
let (relay, key, rtx) = (relay.clone(), key.clone(), rtx.clone());
|
|
tokio::spawn(async move {
|
|
let _ = rtx.send(upstream_subscribe(&relay, p, &key).await);
|
|
});
|
|
}
|
|
let mut first_err = None;
|
|
for _ in 0..n {
|
|
match rrx.recv().await {
|
|
Some(Ok(())) => return Ok(()),
|
|
Some(Err(e)) => {
|
|
first_err.get_or_insert(e);
|
|
}
|
|
None => break,
|
|
}
|
|
}
|
|
Err(first_err.unwrap_or((RequestErrorCode::INTERNAL_ERROR.0, "no upstream answered".into())))
|
|
}
|
|
|
|
/// The relay's own SUBSCRIBE to a publisher. It asks for every object (no filter), at the
|
|
/// default priority and with no delivery timeout: §5.1.3 has a relay not copy a
|
|
/// subscriber's priority or Group Order upstream, and §9.20.3, §9.20.4 its timeouts.
|
|
pub async fn upstream_subscribe(relay: &Relay, pubc: Arc<SessionCtx>, key: &TrackKey) -> Result<(), (u64, String)> {
|
|
let Some(id) = pubc.next_request_id() else { return Err((RequestErrorCode::INTERNAL_ERROR.0, "Request IDs spent".into())) };
|
|
let msg = ControlMessage::Subscribe(TrackRequest {
|
|
request_id: id,
|
|
namespace: key.0.clone(),
|
|
name: key.1.clone(),
|
|
params: Parameters::new().with(Parameter::SubscriberPriority(DEFAULT_PRIORITY)).with(Parameter::Forward(true)),
|
|
});
|
|
let (utx, mut urx) = open_request(&*pubc.conn, &msg, pubc.my_max_updates, pubc.peer_max_updates).await.map_err(|e| (RequestErrorCode::INTERNAL_ERROR.0, e.to_string()))?;
|
|
let t = Duration::from_millis(relay.config().limits.upstream_timeout_ms);
|
|
match timeout(t, urx.next()).await {
|
|
Ok(Ok(Some(ControlMessage::SubscribeOk(ok)))) => {
|
|
{
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
if let Some(other) = st.aliases.get(&(pubc.id, ok.track_alias))
|
|
&& other != key
|
|
{
|
|
drop(st);
|
|
pubc.close(SessionError::DUPLICATE_TRACK_ALIAS.0, "a Track Alias already in use");
|
|
return Err((RequestErrorCode::INTERNAL_ERROR.0, "a duplicate Track Alias".into()));
|
|
}
|
|
let track = relay.track_entry(&mut st, key);
|
|
if track.upstreams.is_empty() {
|
|
track.set_props(ok.track_properties.clone());
|
|
}
|
|
if let Some(l) = ok.params.largest_object() {
|
|
track.note_largest(l);
|
|
}
|
|
track.ended = None;
|
|
track.upstreams.push(Upstream { session: pubc.id, alias: ok.track_alias, kind: UpstreamKind::Subscribe(utx.clone()) });
|
|
st.aliases.insert((pubc.id, ok.track_alias), key.clone());
|
|
}
|
|
relay.inner.change.notify_waiters();
|
|
let (relay, key, sid) = (relay.clone(), key.clone(), pubc.id);
|
|
tokio::spawn(async move {
|
|
let status = loop {
|
|
match urx.next().await {
|
|
Ok(Some(ControlMessage::PublishDone(d))) => break d.status,
|
|
Ok(Some(_)) => continue,
|
|
Ok(None) | Err(_) => break PublishDoneCode::GOING_AWAY,
|
|
}
|
|
};
|
|
relay.upstream_gone(sid, &key, status);
|
|
});
|
|
Ok(())
|
|
}
|
|
Ok(Ok(Some(ControlMessage::RequestError(e)))) => Err((e.code.0, String::from_utf8_lossy(&e.reason).into_owned())),
|
|
Ok(Err(e)) => {
|
|
fail(&pubc, &e);
|
|
Err((RequestErrorCode::INTERNAL_ERROR.0, e.to_string()))
|
|
}
|
|
_ => {
|
|
utx.reset(StreamResetCode::CANCELLED.0).await;
|
|
Err((RequestErrorCode::TIMEOUT.0, "the publisher did not answer SUBSCRIBE".into()))
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn subscribe(relay: Relay, ctx: Arc<SessionCtx>, r: TrackRequest, tx: ReqTx, mut rx: ReqRx, tokens: Vec<Resolved>) {
|
|
let key: TrackKey = (r.namespace.clone(), r.name.clone());
|
|
if is_session_namespace(&key.0) {
|
|
return reject(&tx, RequestErrorCode::DOES_NOT_EXIST, "no session-level track is defined here").await;
|
|
}
|
|
if !relay.inner.auth.allows(&key.0, Right::Subscribe, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this subscription").await;
|
|
}
|
|
if r.params.range_filters().next().is_some() {
|
|
return reject(&tx, RequestErrorCode::INVALID_FILTER, "Range Filters need MAX_FILTER_RANGES, which this relay declares 0").await;
|
|
}
|
|
// A fill that starts below the live edge seeks back (WAI live-rooms §10.6).
|
|
if let Some(Parameter::FillParameters(fill)) = r.params.get(param::FILL_PARAMETERS)
|
|
&& let Some(largest) = relay.largest(&key)
|
|
&& fetch_range(fill.location_filter().or(r.params.location_filter()), largest).is_some_and(|(s, _)| s.group < largest.group)
|
|
&& !relay.inner.auth.may_seek(&key.0, &tokens, &ctx.proven)
|
|
{
|
|
seek_refused(&relay, &ctx, &key);
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, SEEK_NOT_GRANTED).await;
|
|
}
|
|
let rendezvous = varint_param(&r.params, param::RENDEZVOUS_TIMEOUT).unwrap_or(0).min(relay.config().limits.max_rendezvous_ms);
|
|
if let Err((code, why)) = ensure_upstream(&relay, &key, ctx.id, rendezvous).await {
|
|
return reject(&tx, code, &why).await;
|
|
}
|
|
let is_receipts = relay.is_receipts_track(&key.1);
|
|
let made = {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
match st.tracks.get_mut(&key) {
|
|
Some(track) if !track.upstreams.is_empty() => {
|
|
let (start, end) = subscription_range(r.params.location_filter(), track.largest);
|
|
let p = SubParams {
|
|
subscriber_priority: r.params.subscriber_priority().unwrap_or(DEFAULT_PRIORITY),
|
|
order: r.params.group_order().unwrap_or(track.default_order),
|
|
forward: r.params.forward().unwrap_or(true),
|
|
start,
|
|
end,
|
|
object_timeout: r.params.object_delivery_timeout().unwrap_or(0),
|
|
subgroup_timeout: varint_param(&r.params, param::SUBGROUP_DELIVERY_TIMEOUT).unwrap_or(0),
|
|
};
|
|
let id = relay.next_id();
|
|
let sub = Arc::new(DownSub {
|
|
id,
|
|
session: ctx.id,
|
|
request_id: r.request_id,
|
|
alias: id,
|
|
track: key.clone(),
|
|
render: track.render.clone(),
|
|
sched: ctx.sched.clone(),
|
|
params: Mutex::new(p),
|
|
receipts_track: is_receipts,
|
|
tally: ctx.tally.clone(),
|
|
tx: tx.clone(),
|
|
ended: watch::channel(false).0,
|
|
});
|
|
track.downs.push(sub.clone());
|
|
let mut params = Parameters::new();
|
|
if let Some(l) = track.largest {
|
|
params.push(Parameter::LargestObject(l));
|
|
}
|
|
Some((sub, SubscribeOk { track_alias: id, params, track_properties: track.props.clone() }))
|
|
}
|
|
_ => None,
|
|
}
|
|
};
|
|
let Some((sub, ok)) = made else {
|
|
return reject(&tx, RequestErrorCode::DOES_NOT_EXIST, "the publisher left").await;
|
|
};
|
|
ctx.tally.lock().unwrap().subscribed(&sub.render, crate::relay::now_unix_us());
|
|
let mut ended = sub.ended.subscribe();
|
|
if tx.send(&ControlMessage::SubscribeOk(ok)).await.is_err() {
|
|
relay.remove_down(&sub);
|
|
ctx.sched.cancel(sub.id, StreamResetCode::CANCELLED.0);
|
|
return;
|
|
}
|
|
if let Some(Parameter::FillParameters(fill)) = r.params.get(param::FILL_PARAMETERS) {
|
|
let p = sub.params.lock().unwrap().clone();
|
|
if p.forward {
|
|
let fill = fill.clone();
|
|
let filter = fill.location_filter().or(r.params.location_filter());
|
|
let order = fill.group_order().unwrap_or(p.order);
|
|
let prio = fill.subscriber_priority().unwrap_or(p.subscriber_priority);
|
|
let budget = varint_param(&fill, param::FILL_TIMEOUT);
|
|
let (relay, ctx, key, request_id, lane_sub) = (relay.clone(), ctx.clone(), key.clone(), r.request_id, sub.id);
|
|
let may_seek = relay.inner.auth.may_seek(&key.0, &tokens, &ctx.proven);
|
|
tokio::spawn(async move {
|
|
serve_fill(&relay, &ctx, &key, filter, order, prio, request_id, lane_sub, budget, may_seek).await;
|
|
});
|
|
}
|
|
}
|
|
let mut reading = true;
|
|
loop {
|
|
tokio::select! {
|
|
_ = ended.changed() => return,
|
|
m = rx.next(), if reading => match m {
|
|
Ok(Some(ControlMessage::RequestUpdate(u))) => {
|
|
if u.check_for(RequestKind::Subscribe).is_err() {
|
|
let e = ControlMessage::RequestError(RequestError { code: RequestErrorCode::INVALID_FILTER, retry_interval: 0, reason: b"a parameter a subscription update does not take".to_vec(), redirect: None });
|
|
let _ = tx.send(&e).await;
|
|
continue;
|
|
}
|
|
let largest = relay.inner.state.lock().unwrap().tracks.get(&key).and_then(|t| t.largest);
|
|
{
|
|
let mut p = sub.params.lock().unwrap();
|
|
if let Some(v) = u.params.subscriber_priority() {
|
|
p.subscriber_priority = v;
|
|
ctx.sched.set_priority(sub.id, v);
|
|
}
|
|
if let Some(f) = u.params.forward() {
|
|
p.forward = f;
|
|
}
|
|
if let Some(f) = u.params.location_filter() {
|
|
let (s, e) = subscription_range(Some(f), largest);
|
|
p.start = s;
|
|
p.end = e;
|
|
}
|
|
if let Some(t) = u.params.object_delivery_timeout() {
|
|
p.object_timeout = t;
|
|
}
|
|
if let Some(t) = varint_param(&u.params, param::SUBGROUP_DELIVERY_TIMEOUT) {
|
|
p.subgroup_timeout = t;
|
|
}
|
|
}
|
|
let mut params = Parameters::new();
|
|
if let Some(l) = largest {
|
|
params.push(Parameter::LargestObject(l));
|
|
}
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params, track_properties: Default::default() })).await;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
// A FIN says no more updates; the subscription goes on (§6.4.2.2).
|
|
Ok(None) => reading = false,
|
|
Err(StreamError::Wire(e)) => {
|
|
fail(&ctx, &StreamError::Wire(e));
|
|
return;
|
|
}
|
|
// The subscriber cancelled (§6.4.2.3).
|
|
Err(StreamError::Transport(_)) => {
|
|
relay.remove_down(&sub);
|
|
ctx.sched.cancel(sub.id, StreamResetCode::CANCELLED.0);
|
|
tx.reset(StreamResetCode::CANCELLED.0).await;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn publish(relay: Relay, ctx: Arc<SessionCtx>, p: Publish, tx: ReqTx, mut rx: ReqRx, tokens: Vec<Resolved>) {
|
|
let key: TrackKey = (p.namespace.clone(), p.name.clone());
|
|
if is_reserved(&key.0) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "a reserved namespace").await;
|
|
}
|
|
if !relay.inner.auth.allows(&key.0, Right::Publish, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this track").await;
|
|
}
|
|
{
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
if st.aliases.get(&(ctx.id, p.track_alias)).is_some_and(|k| *k != key) {
|
|
drop(st);
|
|
ctx.close(SessionError::DUPLICATE_TRACK_ALIAS.0, "a Track Alias already in use");
|
|
return;
|
|
}
|
|
let track = relay.track_entry(&mut st, &key);
|
|
if track.upstreams.is_empty() {
|
|
track.set_props(p.track_properties.clone());
|
|
}
|
|
if let Some(l) = p.params.largest_object() {
|
|
track.note_largest(l);
|
|
}
|
|
track.ended = None;
|
|
track.upstreams.push(Upstream { session: ctx.id, alias: p.track_alias, kind: UpstreamKind::Publish });
|
|
st.aliases.insert((ctx.id, p.track_alias), key.clone());
|
|
}
|
|
relay.inner.change.notify_waiters();
|
|
if tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await.is_err() {
|
|
relay.upstream_gone(ctx.id, &key, PublishDoneCode::INTERNAL_ERROR);
|
|
return;
|
|
}
|
|
let status = loop {
|
|
match rx.next().await {
|
|
Ok(Some(ControlMessage::PublishDone(d))) => break d.status,
|
|
Ok(Some(ControlMessage::RequestUpdate(_))) => {
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
// A FIN before PUBLISH_DONE: the request failed (§6.4.2.2).
|
|
Ok(None) => break PublishDoneCode::INTERNAL_ERROR,
|
|
Err(StreamError::Wire(e)) => {
|
|
fail(&ctx, &StreamError::Wire(e));
|
|
break PublishDoneCode::INTERNAL_ERROR;
|
|
}
|
|
Err(StreamError::Transport(_)) => break PublishDoneCode::GOING_AWAY,
|
|
}
|
|
};
|
|
relay.upstream_gone(ctx.id, &key, status);
|
|
tx.fin().await;
|
|
}
|
|
|
|
/// The reason phrase of a request refused for seeking back without the power to (WAI
|
|
/// live-rooms §10.6).
|
|
pub const SEEK_NOT_GRANTED: &str = "seek_not_granted";
|
|
|
|
fn seek_refused(relay: &Relay, ctx: &SessionCtx, key: &TrackKey) {
|
|
relay.inner.stats.seek_refused.fetch_add(1, Ordering::Relaxed);
|
|
tracing::info!(session = ctx.id, remote = %ctx.remote, track = %joule_moq::draft22::names::render(&key.0, Some(&key.1)), "a request seeks back below the live edge without the seek power: refused");
|
|
}
|
|
|
|
/// What withholds an object from a FETCH or a fill (WAI live-moderation §6.3): the standing
|
|
/// decisions the DVR store honoured in the track's room, unless the session moderates there.
|
|
#[derive(Clone)]
|
|
struct Withhold {
|
|
dvr: Arc<Dvr>,
|
|
room: Option<[u8; 32]>,
|
|
events: bool,
|
|
}
|
|
|
|
/// Whether the session proved a grant with the moderate power in `room` (live-rooms §3.3):
|
|
/// a moderator's session is given what is withheld from the others (live-moderation §6.3).
|
|
fn moderates(relay: &Relay, ctx: &SessionCtx, room: Option<[u8; 32]>) -> bool {
|
|
room.is_some() && relay.inner.auth.proven_grants(&ctx.setup_tokens, &ctx.proven).iter().any(|g| g.moderates() && Some(g.room) == room)
|
|
}
|
|
|
|
impl Withhold {
|
|
fn of(relay: &Relay, ctx: &SessionCtx, key: &TrackKey) -> Option<Withhold> {
|
|
let dvr = relay.dvr()?.clone();
|
|
let room = room_of(&key.0);
|
|
if !dvr.withholds_in(room) || moderates(relay, ctx, room) {
|
|
return None;
|
|
}
|
|
Some(Withhold { dvr, room, events: is_events_track(key) })
|
|
}
|
|
|
|
/// The payload to send, or `None` when it is withheld whole. Of an events-track object
|
|
/// only the withheld records are left out (`crate::events::withhold`).
|
|
fn apply(&self, payload: Bytes) -> Option<Bytes> {
|
|
if self.events {
|
|
return match crate::events::withhold(&payload, |d| self.dvr.withheld(self.room, d)) {
|
|
None => None,
|
|
Some(std::borrow::Cow::Borrowed(_)) => Some(payload.clone()),
|
|
Some(std::borrow::Cow::Owned(v)) => Some(Bytes::from(v)),
|
|
};
|
|
}
|
|
(!self.dvr.withheld(self.room, &joule_moq::receipt::ObjectReceipt::hash_payload(&payload))).then_some(payload)
|
|
}
|
|
}
|
|
|
|
/// An End of Unknown Range at `(g, o)` on a fetch lane: what the relay cannot or must not
|
|
/// serve there (MOQT §3.2.2, §11.4.1.2).
|
|
fn unknown_item(request_id: u64, order: GroupOrder, prio: u8, fill: bool, g: u64, o: u64) -> Item {
|
|
Item {
|
|
op: Op::FetchEntry { request_id, order, entry: FetchEntry::EndOfRange { kind: RangeEnd::Unknown, location: Location::new(g, o) } },
|
|
prio: Prio { subscriber: prio, publisher: u8::MAX, group_rank: Prio::group_rank(order, g), fill_rank: if fill { 0 } else { 1 }, mode_rank: 1, within: o },
|
|
deadline: None,
|
|
meta: None,
|
|
}
|
|
}
|
|
|
|
/// Push the objects from the cache in `[start, end]` onto a fetch or fill lane, then its
|
|
/// end. Returns how many objects. An object a standing decision withholds is an End of
|
|
/// Unknown Range in its place.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn queue_from_cache(relay: &Relay, sched: &SchedHandle, key: &TrackKey, start: Location, end: Location, order: GroupOrder, prio: u8, request_id: u64, lane_sub: u64, fill: bool, withhold: Option<&Withhold>) -> usize {
|
|
let (objects, render) = {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
match st.tracks.get(key) {
|
|
Some(t) => (t.cache.range(start, end, order == GroupOrder::Descending), t.render.clone()),
|
|
None => (Vec::new(), Arc::from("")),
|
|
}
|
|
};
|
|
let lane = LaneKey { sub: lane_sub, kind: LaneKind::Fetch { request_id } };
|
|
let mut n = 0;
|
|
let mut group: Option<u64> = None;
|
|
let mut slots: Vec<(u64, Option<Item>)> = Vec::new();
|
|
for ((g, o), c) in objects {
|
|
if group != Some(g) {
|
|
if let Some(prev) = group {
|
|
n += push_group(sched, lane, prev, std::mem::take(&mut slots), request_id, order, prio, fill).0;
|
|
}
|
|
group = Some(g);
|
|
}
|
|
if c.status != ObjectStatus::Normal {
|
|
// A fetch carries no status (§11.4.1); a gap says an object does not exist.
|
|
continue;
|
|
}
|
|
let CachedObject { subgroup, publisher_priority, properties, payload, .. } = c;
|
|
let payload = match withhold {
|
|
None => payload,
|
|
Some(w) => match w.apply(payload) {
|
|
Some(p) => p,
|
|
None => {
|
|
slots.push((o, None));
|
|
continue;
|
|
}
|
|
},
|
|
};
|
|
slots.push((o, Some(fetch_item(&render, request_id, order, prio, fill, g, o, subgroup, publisher_priority, properties, payload, false))));
|
|
}
|
|
if let Some(prev) = group {
|
|
n += push_group(sched, lane, prev, slots, request_id, order, prio, fill).0;
|
|
}
|
|
let p = Prio { subscriber: prio, publisher: u8::MAX, group_rank: u64::MAX, fill_rank: 1, mode_rank: 1, within: u64::MAX };
|
|
sched.push(lane, Item { op: Op::FetchFin { request_id, order }, prio: p, deadline: None, meta: None });
|
|
n
|
|
}
|
|
|
|
/// One object on a fetch or fill lane.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn fetch_item(render: &Arc<str>, request_id: u64, order: GroupOrder, prio: u8, fill: bool, g: u64, o: u64, subgroup: Option<u64>, publisher_priority: u8, properties: Properties, payload: Bytes, from_dvr: bool) -> Item {
|
|
let meta = ObjMeta { track: render.clone(), payload_len: payload.len() as u64, object_id: o, receipts_subscribed: false, fetched: true, dvr: from_dvr, signed: None };
|
|
let p = Prio { subscriber: prio, publisher: publisher_priority, group_rank: Prio::group_rank(order, g), fill_rank: if fill { 0 } else { 1 }, mode_rank: 1, within: o };
|
|
let entry = FetchEntry::Object(FetchObject { group_id: g, object_id: o, subgroup_id: subgroup, publisher_priority, properties: (!properties.is_empty()).then_some(properties), payload: payload.to_vec() });
|
|
Item { op: Op::FetchEntry { request_id, order, entry }, prio: p, deadline: None, meta: Some(meta) }
|
|
}
|
|
|
|
/// Push one group's objects onto a fetch lane, in Object ID order; a slot without an item is
|
|
/// an object the relay cannot or must not serve. An End of Range moves to a later group than
|
|
/// the entry before it (MOQT §11.4.1.2, as `joule-moq` encodes it: its Group ID is a delta
|
|
/// from the prior entry's), so it cannot mark one object in the middle of a group. When
|
|
/// objects of the group cannot be served, one End of Unknown Range from the group's start
|
|
/// through the last of them goes first, and the group's objects after it follow. Returns
|
|
/// the objects pushed and their payload bytes.
|
|
#[allow(clippy::too_many_arguments)]
|
|
fn push_group(sched: &SchedHandle, lane: LaneKey, g: u64, slots: Vec<(u64, Option<Item>)>, request_id: u64, order: GroupOrder, prio: u8, fill: bool) -> (usize, usize) {
|
|
let last_unknown = slots.iter().filter(|(_, i)| i.is_none()).map(|(o, _)| *o).max();
|
|
if let Some(u) = last_unknown {
|
|
sched.push(lane, unknown_item(request_id, order, prio, fill, g, u));
|
|
}
|
|
let (mut n, mut bytes) = (0, 0);
|
|
for (o, item) in slots {
|
|
if let Some(item) = item
|
|
&& last_unknown.is_none_or(|u| o > u)
|
|
{
|
|
bytes += item.meta.as_ref().map_or(0, |m| m.payload_len as usize);
|
|
sched.push(lane, item);
|
|
n += 1;
|
|
}
|
|
}
|
|
(n, bytes)
|
|
}
|
|
|
|
/// The session of an upstream that can answer a FETCH for `key`: one of the track's
|
|
/// upstreams, else a publisher of its namespace (§7.6: the same matching as SUBSCRIBE).
|
|
fn fetch_upstream_of(relay: &Relay, key: &TrackKey, except: u64) -> Option<Arc<SessionCtx>> {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
if let Some(t) = st.tracks.get(key)
|
|
&& let Some(u) = t.upstreams.iter().find(|u| u.session != except)
|
|
&& let Some(c) = st.sessions.get(&u.session)
|
|
{
|
|
return Some(c.clone());
|
|
}
|
|
relay.publishers_of(&st, &key.0, except).into_iter().next()
|
|
}
|
|
|
|
/// A FETCH upstream whose objects go onto a downstream lane as they arrive (§3.2.3).
|
|
/// Returns the publisher's FETCH_OK, or its refusal.
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn upstream_fetch(
|
|
relay: &Relay,
|
|
pubc: Arc<SessionCtx>,
|
|
sched: SchedHandle,
|
|
key: &TrackKey,
|
|
start: Location,
|
|
end: Option<Location>,
|
|
order: GroupOrder,
|
|
prio: u8,
|
|
request_id: u64,
|
|
lane_sub: u64,
|
|
fill: bool,
|
|
budget: Duration,
|
|
) -> Result<FetchOk, (RequestErrorCode, String)> {
|
|
let Some(id) = pubc.next_request_id() else { return Err((RequestErrorCode::INTERNAL_ERROR, "Request IDs spent".into())) };
|
|
let (otx, orx) = oneshot::channel();
|
|
pubc.pending_fetch.lock().unwrap().insert(id, otx);
|
|
let filter = match end {
|
|
Some(e) => LocationFilter::AbsoluteRange { start, end_group_delta: e.group - start.group, end_object: e.object },
|
|
None => LocationFilter::AbsoluteStart { start },
|
|
};
|
|
let msg = ControlMessage::Fetch(TrackRequest {
|
|
request_id: id,
|
|
namespace: key.0.clone(),
|
|
name: key.1.clone(),
|
|
params: Parameters::new().with(Parameter::LocationFilter(filter)).with(Parameter::GroupOrder(order)).with(Parameter::SubscriberPriority(DEFAULT_PRIORITY)),
|
|
});
|
|
let fail_up = |why: String| {
|
|
pubc.pending_fetch.lock().unwrap().remove(&id);
|
|
Err((RequestErrorCode::INTERNAL_ERROR, why))
|
|
};
|
|
let (utx, mut urx) = match open_request(&*pubc.conn, &msg, pubc.my_max_updates, pubc.peer_max_updates).await {
|
|
Ok(p) => p,
|
|
Err(e) => return fail_up(e.to_string()),
|
|
};
|
|
let t = Duration::from_millis(relay.config().limits.upstream_timeout_ms);
|
|
let ok = match timeout(t, urx.next()).await {
|
|
Ok(Ok(Some(ControlMessage::FetchOk(ok)))) => ok,
|
|
Ok(Ok(Some(ControlMessage::RequestError(e)))) => {
|
|
pubc.pending_fetch.lock().unwrap().remove(&id);
|
|
return Err((e.code, String::from_utf8_lossy(&e.reason).into_owned()));
|
|
}
|
|
_ => {
|
|
utx.reset(StreamResetCode::CANCELLED.0).await;
|
|
return fail_up("the publisher did not answer FETCH".into());
|
|
}
|
|
};
|
|
relay.inner.stats.fetch_served_upstream.fetch_add(1, Ordering::Relaxed);
|
|
let render: Arc<str> = joule_moq::draft22::names::render(&key.0, Some(&key.1)).into();
|
|
let end_location = ok.end_location;
|
|
tokio::spawn(async move {
|
|
let lane = LaneKey { sub: lane_sub, kind: LaneKind::Fetch { request_id } };
|
|
let fin = |op| Item { op, prio: Prio { subscriber: prio, publisher: u8::MAX, group_rank: u64::MAX, fill_rank: 1, mode_rank: 1, within: u64::MAX }, deadline: None, meta: None };
|
|
let reader = timeout(budget, orx).await;
|
|
let Ok(Ok(mut reader)) = reader else {
|
|
// §3.2.3: past the Fill Timeout what is not there is a Timed-Out gap.
|
|
sched.push(lane, fin(Op::FetchEntry { request_id, order, entry: FetchEntry::EndOfRange { kind: RangeEnd::TimedOut, location: end_location } }));
|
|
sched.push(lane, fin(Op::FetchFin { request_id, order }));
|
|
return;
|
|
};
|
|
reader.set_group_order(order);
|
|
loop {
|
|
match timeout(budget, reader.next()).await {
|
|
Ok(Ok(Some(UniEvent::FetchEntry(entry)))) => {
|
|
let (pp, g, o, len) = match &entry {
|
|
FetchEntry::Object(f) => (f.publisher_priority, f.group_id, f.object_id, f.payload.len() as u64),
|
|
FetchEntry::EndOfRange { location, .. } => (u8::MAX, location.group, location.object, 0),
|
|
};
|
|
let meta = matches!(entry, FetchEntry::Object(_)).then(|| ObjMeta { track: render.clone(), payload_len: len, object_id: o, receipts_subscribed: false, fetched: true, dvr: false, signed: None });
|
|
let p = Prio { subscriber: prio, publisher: pp, group_rank: Prio::group_rank(order, g), fill_rank: if fill { 0 } else { 1 }, mode_rank: 1, within: o };
|
|
sched.push(lane, Item { op: Op::FetchEntry { request_id, order, entry }, prio: p, deadline: None, meta });
|
|
}
|
|
Ok(Ok(None)) => {
|
|
sched.push(lane, fin(Op::FetchFin { request_id, order }));
|
|
return;
|
|
}
|
|
Ok(Err(StreamError::Transport(TransportError::Reset(code)))) => {
|
|
sched.push(lane, fin(Op::FetchReset { request_id, order, code }));
|
|
return;
|
|
}
|
|
Ok(Ok(Some(_))) | Ok(Err(_)) => {
|
|
sched.push(lane, fin(Op::FetchReset { request_id, order, code: StreamResetCode::INTERNAL_ERROR.0 }));
|
|
return;
|
|
}
|
|
Err(_) => {
|
|
sched.push(lane, fin(Op::FetchEntry { request_id, order, entry: FetchEntry::EndOfRange { kind: RangeEnd::TimedOut, location: end_location } }));
|
|
sched.push(lane, fin(Op::FetchFin { request_id, order }));
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
Ok(ok)
|
|
}
|
|
|
|
/// Where a group of a FETCH's range comes from when the relay answers it alone.
|
|
enum Src {
|
|
/// The cache holds it: its objects in the range.
|
|
Cache(Vec<((u64, u64), CachedObject)>),
|
|
/// The DVR store holds it.
|
|
Dvr(u64),
|
|
/// Groups `lo ..= hi` were evicted from the DVR store: no longer known.
|
|
Unknown(u64, u64),
|
|
}
|
|
|
|
/// The groups of `[start, end]`, ascending, as the relay answers them alone: each held whole
|
|
/// in the cache (the group of the Largest Object, `last`, from its start), held in the DVR
|
|
/// store, known not to exist (left out), or evicted from the store. `None` when a group is
|
|
/// none of these: the relay then asks upstream (the cache's rule).
|
|
fn local_plan(c: Option<&crate::cache::TrackCache>, snap: &Snapshot, start: Location, end: Location, last: u64) -> Option<Vec<Src>> {
|
|
let mut out = Vec::new();
|
|
let mut g = start.group;
|
|
while g <= end.group {
|
|
let cs = c.and_then(|c| c.groups.get(&g).copied().filter(|_| c.evicted_through.is_none_or(|e| g > e)));
|
|
let next = if cs.is_some_and(|s| s.whole() || (g == last && s.start && !s.broken)) {
|
|
let c = c.expect("held");
|
|
let objs = c.objects.range((g, 0)..=(g, u64::MAX)).filter(|((gg, oo), _)| (start..=end).contains(&Location::new(*gg, *oo))).map(|(k, o)| (*k, o.clone())).collect();
|
|
out.push(Src::Cache(objs));
|
|
g.checked_add(1)
|
|
} else if snap.held.contains(&g) {
|
|
out.push(Src::Dvr(g));
|
|
g.checked_add(1)
|
|
} else if let Some(hi) = c.and_then(|c| c.absent_through(g)).or_else(|| snap.absent_through(g)) {
|
|
hi.checked_add(1)
|
|
} else {
|
|
// Evicted from the store, or none of these: the relay asks upstream.
|
|
let e = snap.evicted_through.filter(|e| g <= *e)?;
|
|
// The run stops before the next group the cache still holds whole and has not
|
|
// evicted. The search starts after `g`, which the cache cannot serve (the first
|
|
// branch), so the run is never empty and `c - 1` never passes below `g`.
|
|
let top = e.min(end.group);
|
|
let cached = g.checked_add(1).filter(|from| *from <= top).and_then(|from| {
|
|
c.and_then(|c| {
|
|
c.groups
|
|
.range(from..=top)
|
|
.find(|(gg, s)| c.evicted_through.is_none_or(|ce| **gg > ce) && (s.whole() || (**gg == last && s.start && !s.broken)))
|
|
.map(|(gg, _)| *gg)
|
|
})
|
|
});
|
|
let hi = cached.map_or(top, |c| c - 1);
|
|
out.push(Src::Unknown(g, hi));
|
|
hi.checked_add(1)
|
|
};
|
|
match next {
|
|
Some(n) => g = n,
|
|
None => break,
|
|
}
|
|
}
|
|
Some(out)
|
|
}
|
|
|
|
/// Serve a planned range from the cache and the DVR store onto a fetch or fill lane, groups
|
|
/// in the requested order, then its end. The store's groups are read as the lane drains,
|
|
/// about a mebibyte at a time, so a long range is never held in memory whole. What the
|
|
/// store evicted, dropped or a decision withholds is an End of Unknown Range: those objects
|
|
/// existed, and an unmarked gap would say they did not (MOQT §3.2.2).
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn serve_local(relay: Relay, sched: SchedHandle, key: TrackKey, groups: Vec<Src>, start: Location, end: Location, order: GroupOrder, prio: u8, request_id: u64, lane_sub: u64, fill: bool, withhold: Option<Withhold>, moderator: bool) {
|
|
let render: Arc<str> = joule_moq::draft22::names::render(&key.0, Some(&key.1)).into();
|
|
let lane = LaneKey { sub: lane_sub, kind: LaneKind::Fetch { request_id } };
|
|
let descending = order == GroupOrder::Descending;
|
|
let mut groups = groups;
|
|
if descending {
|
|
groups.reverse();
|
|
}
|
|
let dvr = relay.dvr().cloned();
|
|
let mut pending = 0usize;
|
|
for src in groups {
|
|
match src {
|
|
Src::Unknown(lo, hi) => {
|
|
let at = if descending { Location::new(lo, u64::MAX) } else { Location::new(hi, u64::MAX).min(end) };
|
|
sched.push(lane, unknown_item(request_id, order, prio, fill, at.group, at.object));
|
|
}
|
|
Src::Cache(objs) => {
|
|
let Some(g) = objs.first().map(|((g, _), _)| *g) else { continue };
|
|
let mut slots = Vec::with_capacity(objs.len());
|
|
for ((_, o), c) in objs {
|
|
if c.status != ObjectStatus::Normal {
|
|
continue;
|
|
}
|
|
let payload = match &withhold {
|
|
None => Some(c.payload),
|
|
Some(w) => w.apply(c.payload),
|
|
};
|
|
slots.push((o, payload.map(|p| fetch_item(&render, request_id, order, prio, fill, g, o, c.subgroup, c.publisher_priority, c.properties, p, false))));
|
|
}
|
|
push_group(&sched, lane, g, slots, request_id, order, prio, fill);
|
|
}
|
|
Src::Dvr(g) => {
|
|
let from = if g == start.group { start.object } else { 0 };
|
|
let to = if g == end.group { end.object } else { u64::MAX };
|
|
let read = match dvr.clone() {
|
|
Some(d) => {
|
|
let k = key.clone();
|
|
tokio::task::spawn_blocking(move || d.read_group(&k, g, from, to, !moderator)).await.ok().flatten()
|
|
}
|
|
None => None,
|
|
};
|
|
match read {
|
|
None => sched.push(lane, unknown_item(request_id, order, prio, fill, g, to)),
|
|
Some(read) => {
|
|
let slots = read
|
|
.into_iter()
|
|
.filter_map(|slot| match slot {
|
|
Slot::Object(so) if so.status == ObjectStatus::Normal => {
|
|
Some((so.id, Some(fetch_item(&render, request_id, order, prio, fill, g, so.id, so.subgroup, so.priority, so.properties, so.payload, true))))
|
|
}
|
|
Slot::Object(_) => None,
|
|
Slot::Unknown(o) => Some((o, None)),
|
|
})
|
|
.collect();
|
|
pending += push_group(&sched, lane, g, slots, request_id, order, prio, fill).1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if pending >= 1 << 20 {
|
|
sched.drained(lane_sub).await;
|
|
pending = 0;
|
|
}
|
|
}
|
|
let p = Prio { subscriber: prio, publisher: u8::MAX, group_rank: u64::MAX, fill_rank: 1, mode_rank: 1, within: u64::MAX };
|
|
sched.push(lane, Item { op: Op::FetchFin { request_id, order }, prio: p, deadline: None, meta: None });
|
|
}
|
|
|
|
async fn fetch(relay: Relay, ctx: Arc<SessionCtx>, r: TrackRequest, tx: ReqTx, mut rx: ReqRx, tokens: Vec<Resolved>) {
|
|
let key: TrackKey = (r.namespace.clone(), r.name.clone());
|
|
if is_session_namespace(&key.0) {
|
|
return reject(&tx, RequestErrorCode::DOES_NOT_EXIST, "no session-level track is defined here").await;
|
|
}
|
|
if !relay.inner.auth.allows(&key.0, Right::Subscribe, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this fetch").await;
|
|
}
|
|
if r.params.range_filters().next().is_some() {
|
|
return reject(&tx, RequestErrorCode::INVALID_FILTER, "Range Filters need MAX_FILTER_RANGES, which this relay declares 0").await;
|
|
}
|
|
let order = r.params.group_order().unwrap_or(GroupOrder::Ascending);
|
|
let prio = r.params.subscriber_priority().unwrap_or(DEFAULT_PRIORITY);
|
|
let budget_ms = varint_param(&r.params, param::FILL_TIMEOUT).unwrap_or(relay.config().limits.fill_timeout_ms).min(relay.config().limits.fill_timeout_ms);
|
|
let lane_sub = relay.next_id();
|
|
let may_seek = relay.inner.auth.may_seek(&key.0, &tokens, &ctx.proven);
|
|
let moderator = moderates(&relay, &ctx, room_of(&key.0));
|
|
enum Plan {
|
|
Cache { start: Location, end: Location, eot: bool, props: joule_moq::draft22::kvp::Properties },
|
|
Local { start: Location, end: Location, eot: bool, props: joule_moq::draft22::kvp::Properties, groups: Vec<Src> },
|
|
Upstream { start: Location, end: Option<Location> },
|
|
Refuse(RequestErrorCode, &'static str),
|
|
}
|
|
let plan_now = || {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
let t = st.tracks.get(&key);
|
|
let dvr = relay.dvr().filter(|d| d.keeps(&key.0));
|
|
let kept = dvr.and_then(|d| d.largest(&key));
|
|
let live = t.and_then(|t| t.largest);
|
|
if t.is_none() && kept.is_none() {
|
|
return (Plan::Upstream { start: Location::new(0, 0), end: None }, false);
|
|
}
|
|
let Some(largest) = live.max(kept) else { return (Plan::Refuse(RequestErrorCode::INVALID_RANGE, "no object has been published on the track"), false) };
|
|
let Some((start, end)) = fetch_range(r.params.location_filter(), largest) else {
|
|
return (Plan::Refuse(RequestErrorCode::INVALID_RANGE, "the range starts after the Largest Object"), false);
|
|
};
|
|
// Below the live edge, the group of the Largest Object, a request seeks back
|
|
// (WAI live-rooms §10.6).
|
|
if start.group < largest.group && !may_seek {
|
|
return (Plan::Refuse(RequestErrorCode::UNAUTHORIZED, SEEK_NOT_GRANTED), false);
|
|
}
|
|
if let Some(t) = t
|
|
&& live == Some(largest)
|
|
&& t.cache.covers(start)
|
|
{
|
|
return (Plan::Cache { start, end, eot: t.cache.end_of_track == Some(end), props: t.props.clone() }, false);
|
|
}
|
|
if let Some(d) = dvr
|
|
&& let Some(snap) = d.snapshot(&key, start.group, end.group)
|
|
&& let Some(groups) = local_plan(t.map(|t| &t.cache), &snap, start, end, largest.group)
|
|
{
|
|
let eot = t.is_some_and(|t| t.cache.end_of_track == Some(end)) || snap.end_of_track == Some(end);
|
|
let props = t.map(|t| t.props.clone()).filter(|p| !p.is_empty()).unwrap_or_else(|| snap.props.clone());
|
|
return (Plan::Local { start, end, eot, props, groups }, false);
|
|
}
|
|
match t {
|
|
None => (Plan::Upstream { start, end: Some(end) }, false),
|
|
// Not covered while the track's publisher is still sending and nothing from
|
|
// `start` was evicted: the groups it lacks may be ones still arriving (each
|
|
// group its own stream, finishing out of order).
|
|
Some(t) => (Plan::Upstream { start, end: Some(end) }, !t.upstreams.is_empty() && t.cache.evicted_through.is_none_or(|e| start.group > e)),
|
|
}
|
|
};
|
|
// A fetch the cache does not yet cover, of a track still arriving, waits a little for
|
|
// its groups to finish (within the fetch's fill budget, at most a second) before the
|
|
// relay asks upstream: a publisher by PUBLISH need not answer a FETCH at all.
|
|
let wait_until = Instant::now() + Duration::from_millis(budget_ms.min(1_000));
|
|
let plan = loop {
|
|
let (p, may_complete) = plan_now();
|
|
if !(may_complete && Instant::now() < wait_until) {
|
|
break p;
|
|
}
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
};
|
|
match plan {
|
|
Plan::Refuse(code, why) => {
|
|
if why == SEEK_NOT_GRANTED {
|
|
seek_refused(&relay, &ctx, &key);
|
|
}
|
|
reject(&tx, code, why).await
|
|
}
|
|
Plan::Cache { start, end, eot, props } => {
|
|
relay.inner.stats.fetch_served_from_cache.fetch_add(1, Ordering::Relaxed);
|
|
let ok = FetchOk { end_of_track: eot, end_location: end, params: Parameters::new(), track_properties: props };
|
|
if tx.send(&ControlMessage::FetchOk(ok)).await.is_err() {
|
|
return;
|
|
}
|
|
let withhold = Withhold::of(&relay, &ctx, &key);
|
|
queue_from_cache(&relay, &ctx.sched, &key, start, end, order, prio, r.request_id, lane_sub, false, withhold.as_ref());
|
|
finish_fetch(&ctx, lane_sub, &tx, &mut rx, None).await;
|
|
}
|
|
Plan::Local { start, end, eot, props, groups } => {
|
|
if groups.iter().any(|g| !matches!(g, Src::Cache(_))) {
|
|
relay.inner.stats.fetch_served_from_dvr.fetch_add(1, Ordering::Relaxed);
|
|
} else {
|
|
relay.inner.stats.fetch_served_from_cache.fetch_add(1, Ordering::Relaxed);
|
|
}
|
|
let ok = FetchOk { end_of_track: eot, end_location: end, params: Parameters::new(), track_properties: props };
|
|
if tx.send(&ControlMessage::FetchOk(ok)).await.is_err() {
|
|
return;
|
|
}
|
|
let withhold = Withhold::of(&relay, &ctx, &key);
|
|
let producer = tokio::spawn(serve_local(relay.clone(), ctx.sched.clone(), key.clone(), groups, start, end, order, prio, r.request_id, lane_sub, false, withhold, moderator));
|
|
finish_fetch(&ctx, lane_sub, &tx, &mut rx, Some(producer)).await;
|
|
}
|
|
Plan::Upstream { start, end } => {
|
|
let Some(pubc) = fetch_upstream_of(&relay, &key, ctx.id) else {
|
|
return reject(&tx, RequestErrorCode::DOES_NOT_EXIST, "the relay holds no such objects and has no publisher to ask").await;
|
|
};
|
|
// With no Location Filter the request's own range goes upstream unchanged.
|
|
let (start, end) = match (r.params.location_filter(), end) {
|
|
(Some(_), _) | (None, Some(_)) => (start, end),
|
|
(None, None) => (Location::new(0, 0), None),
|
|
};
|
|
match upstream_fetch(&relay, pubc, ctx.sched.clone(), &key, start, end, order, prio, r.request_id, lane_sub, false, Duration::from_millis(budget_ms)).await {
|
|
Ok(ok) => {
|
|
let ok = FetchOk { end_of_track: ok.end_of_track, end_location: ok.end_location, params: Parameters::new(), track_properties: ok.track_properties };
|
|
if tx.send(&ControlMessage::FetchOk(ok)).await.is_err() {
|
|
ctx.sched.cancel(lane_sub, StreamResetCode::CANCELLED.0);
|
|
return;
|
|
}
|
|
finish_fetch(&ctx, lane_sub, &tx, &mut rx, None).await;
|
|
}
|
|
Err((code, why)) => reject(&tx, code, &why).await,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// After FETCH_OK: answer priority updates until the fetch stream is sent, then FIN; a
|
|
/// cancelled request cancels the stream (§3.2.4).
|
|
async fn finish_fetch(ctx: &SessionCtx, lane_sub: u64, tx: &ReqTx, rx: &mut ReqRx, producer: Option<tokio::task::JoinHandle<()>>) {
|
|
let sched = ctx.sched.clone();
|
|
let abort = producer.as_ref().map(|p| p.abort_handle());
|
|
// With a producer still pushing (a range read from the DVR store as the lane drains),
|
|
// the lane is drained only once it has pushed the end.
|
|
let drained = async {
|
|
if let Some(p) = producer {
|
|
let _ = p.await;
|
|
}
|
|
sched.drained(lane_sub).await
|
|
};
|
|
tokio::pin!(drained);
|
|
let mut reading = true;
|
|
loop {
|
|
tokio::select! {
|
|
_ = &mut drained => break,
|
|
m = rx.next(), if reading => match m {
|
|
Ok(Some(ControlMessage::RequestUpdate(u))) => {
|
|
if let Some(p) = u.params.subscriber_priority() {
|
|
ctx.sched.set_priority(lane_sub, p);
|
|
}
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => reading = false,
|
|
Err(_) => {
|
|
if let Some(a) = &abort {
|
|
a.abort();
|
|
}
|
|
ctx.sched.cancel(lane_sub, StreamResetCode::CANCELLED.0);
|
|
tx.reset(StreamResetCode::CANCELLED.0).await;
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
tx.fin().await;
|
|
}
|
|
|
|
/// A subscription's fill (§3.4): the fill range from the cache, or upstream, on a fill
|
|
/// fetch stream whose header names the SUBSCRIBE's Request ID. A fill the relay cannot
|
|
/// serve is a fetch stream reset right after its header (§3.4.1).
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn serve_fill(relay: &Relay, ctx: &Arc<SessionCtx>, key: &TrackKey, filter: Option<LocationFilter>, order: GroupOrder, prio: u8, request_id: u64, lane_sub: u64, budget: Option<u64>, may_seek: bool) {
|
|
let range = {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
st.tracks.get(key).and_then(|t| t.largest.and_then(|l| fetch_range(filter, l).map(|(s, e)| (s, e, t.cache.covers(s), l))))
|
|
};
|
|
let Some((start, end, covered, largest)) = range else { return };
|
|
let withhold = Withhold::of(relay, ctx, key);
|
|
if covered {
|
|
relay.inner.stats.fetch_served_from_cache.fetch_add(1, Ordering::Relaxed);
|
|
queue_from_cache(relay, &ctx.sched, key, start, end, order, prio, request_id, lane_sub, true, withhold.as_ref());
|
|
return;
|
|
}
|
|
// The fill's earlier groups from the DVR store: a window whose end lies ahead of the
|
|
// live edge is a SUBSCRIBE with that range and a fill (WAI live-recording §8.3).
|
|
if may_seek || start.group >= largest.group {
|
|
let local = {
|
|
let st = relay.inner.state.lock().unwrap();
|
|
relay.dvr().filter(|d| d.keeps(&key.0)).and_then(|d| d.snapshot(key, start.group, end.group)).and_then(|snap| local_plan(st.tracks.get(key).map(|t| &t.cache), &snap, start, end, largest.group))
|
|
};
|
|
if let Some(groups) = local {
|
|
relay.inner.stats.fetch_served_from_dvr.fetch_add(1, Ordering::Relaxed);
|
|
let moderator = moderates(relay, ctx, room_of(&key.0));
|
|
serve_local(relay.clone(), ctx.sched.clone(), key.clone(), groups, start, end, order, prio, request_id, lane_sub, true, withhold, moderator).await;
|
|
return;
|
|
}
|
|
}
|
|
let budget = Duration::from_millis(budget.unwrap_or(relay.config().limits.fill_timeout_ms).min(relay.config().limits.fill_timeout_ms));
|
|
let fail_lane = |sched: &SchedHandle| {
|
|
let lane = LaneKey { sub: lane_sub, kind: LaneKind::Fetch { request_id } };
|
|
let p = Prio { subscriber: prio, publisher: u8::MAX, group_rank: 0, fill_rank: 0, mode_rank: 1, within: 0 };
|
|
sched.push(lane, Item { op: Op::FetchReset { request_id, order, code: StreamResetCode::UNKNOWN_OBJECT_STATUS.0 }, prio: p, deadline: None, meta: None });
|
|
};
|
|
match fetch_upstream_of(relay, key, ctx.id) {
|
|
Some(pubc) => {
|
|
if upstream_fetch(relay, pubc, ctx.sched.clone(), key, start, Some(end), order, prio, request_id, lane_sub, true, budget).await.is_err() {
|
|
fail_lane(&ctx.sched);
|
|
}
|
|
}
|
|
None => fail_lane(&ctx.sched),
|
|
}
|
|
}
|
|
|
|
async fn track_status(relay: Relay, ctx: Arc<SessionCtx>, r: TrackRequest, tx: ReqTx, tokens: Vec<Resolved>) {
|
|
let key: TrackKey = (r.namespace.clone(), r.name.clone());
|
|
if !relay.inner.auth.allows(&key.0, Right::Subscribe, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this track").await;
|
|
}
|
|
let found = relay.inner.state.lock().unwrap().tracks.get(&key).map(|t| (t.largest, t.props.clone()));
|
|
// A track the relay keeps in its DVR store, its publisher gone (or before a restart).
|
|
let found = found.or_else(|| relay.dvr().and_then(|d| d.largest(&key).map(|l| (Some(l), d.props(&key).unwrap_or_default()))));
|
|
match found {
|
|
Some((largest, props)) => {
|
|
let mut params = Parameters::new();
|
|
if let Some(l) = largest {
|
|
params.push(Parameter::LargestObject(l));
|
|
}
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params, track_properties: props })).await;
|
|
tx.fin().await;
|
|
}
|
|
None => reject(&tx, RequestErrorCode::DOES_NOT_EXIST, "the relay knows no such track").await,
|
|
}
|
|
}
|
|
|
|
/// Withdraw a published namespace, telling namespace subscribers when nobody publishes
|
|
/// it any longer.
|
|
fn withdraw(relay: &Relay, session: u64, id: u64) {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
let Some(pos) = st.namespaces.iter().position(|n| n.id == id) else { return };
|
|
let gone = st.namespaces.remove(pos);
|
|
let still = st.namespaces.iter().any(|n| n.namespace == gone.namespace);
|
|
if !still {
|
|
for s in st.ns_subs.iter().filter(|s| s.session != session && prefix_matches(&s.prefix, &gone.namespace)) {
|
|
let _ = s.tx.send(NsEvent::Remove(gone.namespace.clone()));
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn publish_namespace(relay: Relay, ctx: Arc<SessionCtx>, r: NamespaceRequest, tx: ReqTx, mut rx: ReqRx, tokens: Vec<Resolved>) {
|
|
if is_reserved(&r.prefix) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "a reserved namespace").await;
|
|
}
|
|
if !relay.inner.auth.allows(&r.prefix, Right::Publish, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this namespace").await;
|
|
}
|
|
let id = relay.next_id();
|
|
let to_subscribe = {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
let known = st.namespaces.iter().any(|n| n.namespace == r.prefix);
|
|
st.namespaces.push(PublishedNs { namespace: r.prefix.clone(), session: ctx.id, id });
|
|
if !known {
|
|
for s in st.ns_subs.iter().filter(|s| s.session != ctx.id && prefix_matches(&s.prefix, &r.prefix)) {
|
|
let _ = s.tx.send(NsEvent::Add(r.prefix.clone()));
|
|
}
|
|
}
|
|
// §7.6: a new publisher of a namespace gets a SUBSCRIBE for every subscription
|
|
// the relay holds under it.
|
|
st.tracks
|
|
.values()
|
|
.filter(|t| !t.downs.is_empty() && prefix_matches(&r.prefix, &t.key.0) && !t.upstreams.iter().any(|u| u.session == ctx.id))
|
|
.map(|t| t.key.clone())
|
|
.collect::<Vec<_>>()
|
|
};
|
|
if tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await.is_err() {
|
|
withdraw(&relay, ctx.id, id);
|
|
return;
|
|
}
|
|
relay.inner.change.notify_waiters();
|
|
for key in to_subscribe {
|
|
let (relay, ctx) = (relay.clone(), ctx.clone());
|
|
tokio::spawn(async move {
|
|
let _ = upstream_subscribe(&relay, ctx, &key).await;
|
|
});
|
|
}
|
|
loop {
|
|
match rx.next().await {
|
|
Ok(Some(ControlMessage::RequestUpdate(_))) => {
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
// A FIN is not a withdrawal; the namespace stands until the request is
|
|
// cancelled or the session ends (§4.1).
|
|
Ok(None) => {
|
|
let _ = ctx.conn.closed().await;
|
|
return;
|
|
}
|
|
Err(_) => {
|
|
withdraw(&relay, ctx.id, id);
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn suffix(ns: &TrackNamespace, prefix: &TrackNamespace) -> TrackNamespace {
|
|
TrackNamespace(ns.0[prefix.0.len().min(ns.0.len())..].to_vec())
|
|
}
|
|
|
|
async fn subscribe_namespace(relay: Relay, ctx: Arc<SessionCtx>, r: NamespaceRequest, tx: ReqTx, mut rx: ReqRx, tokens: Vec<Resolved>) {
|
|
if !relay.inner.auth.allows(&r.prefix, Right::Subscribe, &tokens, &ctx.proven) {
|
|
return reject(&tx, RequestErrorCode::UNAUTHORIZED, "no grant covers this namespace").await;
|
|
}
|
|
let (etx, mut erx) = mpsc::unbounded_channel();
|
|
let id = relay.next_id();
|
|
let initial = {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
if st.ns_subs.iter().any(|s| s.session == ctx.id && prefixes_overlap(&s.prefix, &r.prefix)) {
|
|
None
|
|
} else {
|
|
st.ns_subs.push(NsSub { prefix: r.prefix.clone(), session: ctx.id, id, tx: etx });
|
|
Some(relay.published_under(&st, &r.prefix, ctx.id))
|
|
}
|
|
};
|
|
let Some(initial) = initial else {
|
|
return reject(&tx, RequestErrorCode::PREFIX_OVERLAP, "it overlaps another namespace subscription of this session").await;
|
|
};
|
|
let remove = |relay: &Relay| relay.inner.state.lock().unwrap().ns_subs.retain(|s| s.id != id);
|
|
if tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await.is_err() {
|
|
remove(&relay);
|
|
return;
|
|
}
|
|
for ns in initial {
|
|
if tx.send(&ControlMessage::Namespace(suffix(&ns, &r.prefix))).await.is_err() {
|
|
remove(&relay);
|
|
return;
|
|
}
|
|
}
|
|
let mut reading = true;
|
|
loop {
|
|
tokio::select! {
|
|
ev = erx.recv() => {
|
|
let m = match ev {
|
|
Some(NsEvent::Add(ns)) => ControlMessage::Namespace(suffix(&ns, &r.prefix)),
|
|
Some(NsEvent::Remove(ns)) => ControlMessage::NamespaceDone(suffix(&ns, &r.prefix)),
|
|
None => break,
|
|
};
|
|
if tx.send(&m).await.is_err() {
|
|
break;
|
|
}
|
|
}
|
|
m = rx.next(), if reading => match m {
|
|
Ok(Some(ControlMessage::RequestUpdate(_))) => {
|
|
let _ = tx.send(&ControlMessage::RequestOk(RequestOk { params: Parameters::new(), track_properties: Default::default() })).await;
|
|
}
|
|
Ok(Some(_)) => {}
|
|
Ok(None) => reading = false,
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
}
|
|
remove(&relay);
|
|
}
|
|
|
|
// ── The end of a session ───────────────────────────────────────────────────
|
|
|
|
async fn cleanup(relay: &Relay, ctx: &Arc<SessionCtx>, why: TransportError) {
|
|
let (downs, ups, nss) = {
|
|
let mut st = relay.inner.state.lock().unwrap();
|
|
st.sessions.remove(&ctx.id);
|
|
let mut downs = Vec::new();
|
|
let mut ups = Vec::new();
|
|
for t in st.tracks.values_mut() {
|
|
t.downs.retain(|d| {
|
|
if d.session == ctx.id {
|
|
downs.push(d.clone());
|
|
false
|
|
} else {
|
|
true
|
|
}
|
|
});
|
|
if t.upstreams.iter().any(|u| u.session == ctx.id) {
|
|
ups.push(t.key.clone());
|
|
}
|
|
}
|
|
let nss: Vec<u64> = st.namespaces.iter().filter(|n| n.session == ctx.id).map(|n| n.id).collect();
|
|
st.ns_subs.retain(|s| s.session != ctx.id);
|
|
(downs, ups, nss)
|
|
};
|
|
ctx.pending_fetch.lock().unwrap().clear();
|
|
for d in &downs {
|
|
let _ = d.ended.send(true);
|
|
relay.remove_down(d);
|
|
}
|
|
// The publisher's session ended: its tracks end with GOING_AWAY (§12.4).
|
|
for key in ups {
|
|
relay.upstream_gone(ctx.id, &key, PublishDoneCode::GOING_AWAY);
|
|
}
|
|
for id in nss {
|
|
withdraw(relay, ctx.id, id);
|
|
}
|
|
ctx.sched.shutdown_and_wait().await;
|
|
let close_code = match &why {
|
|
TransportError::Closed { code, by_peer: true, .. } if *code != u64::MAX => Some(*code),
|
|
_ => *ctx.close_code.lock().unwrap(),
|
|
};
|
|
let closed_us = crate::relay::now_unix_us();
|
|
let tally = {
|
|
let mut t = ctx.tally.lock().unwrap();
|
|
t.close_spans(closed_us);
|
|
t.clone()
|
|
};
|
|
let summary = SessionSummary {
|
|
format: FORMAT.into(),
|
|
relay_key: hex::encode(relay.summary_key().to_bytes()),
|
|
session: ctx.id,
|
|
transport: ctx.kind.label().into(),
|
|
remote: ctx.remote.clone(),
|
|
draft: String::from_utf8_lossy(joule_moq::draft22::ALPN).into_owned(),
|
|
opened_unix_ms: ctx.opened_unix_ms,
|
|
closed_unix_ms: closed_us / 1000,
|
|
close_code,
|
|
relay_energy: RelayEnergy::unmetered(),
|
|
tally,
|
|
};
|
|
let signed = SignedSessionSummary::sign(summary, &relay.inner.key);
|
|
if let Some(path) = &relay.config().summary.out_path
|
|
&& let Ok(line) = serde_json::to_string(&signed)
|
|
{
|
|
use std::io::Write;
|
|
// One write of the line and its newline: sessions that end together append whole
|
|
// lines, never one inside another.
|
|
if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(path) {
|
|
let _ = f.write_all(format!("{line}\n").as_bytes());
|
|
}
|
|
}
|
|
tracing::info!(session = ctx.id, transport = ctx.kind.label(), "session ended");
|
|
relay.inner.summaries.lock().unwrap().push(signed);
|
|
relay.inner.sessions_gone.notify_waiters();
|
|
relay.inner.change.notify_waiters();
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod local_plan_tests {
|
|
use super::{local_plan, Src};
|
|
use crate::cache::{GroupState, TrackCache};
|
|
use crate::dvr::Snapshot;
|
|
use joule_moq::draft22::Location;
|
|
|
|
fn whole() -> GroupState {
|
|
GroupState { start: true, finished: 1, ..GroupState::default() }
|
|
}
|
|
|
|
/// A group the cache still holds whole at or below its own eviction mark is not the
|
|
/// cache's to serve, and a run of groups the store evicted that starts there must end
|
|
/// at or after it. Group 0 used to underflow (`0 - 1`) and panic the relay's task.
|
|
#[test]
|
|
fn an_evicted_run_starting_at_a_whole_group_the_cache_evicted() {
|
|
let mut c = TrackCache::default();
|
|
c.groups.insert(0, whole());
|
|
c.groups.insert(1, whole());
|
|
c.evicted_through = Some(0);
|
|
let snap = Snapshot { evicted_through: Some(0), ..Snapshot::default() };
|
|
let plan = local_plan(Some(&c), &snap, Location::new(0, 0), Location::new(1, u64::MAX), 1).expect("a plan");
|
|
assert!(matches!(plan[0], Src::Unknown(0, 0)), "group 0 is evicted, alone");
|
|
assert!(matches!(plan[1], Src::Cache(_)), "group 1 comes from the cache");
|
|
assert_eq!(plan.len(), 2);
|
|
}
|
|
|
|
/// The same at a later group: the run is never inverted.
|
|
#[test]
|
|
fn an_evicted_run_is_never_inverted() {
|
|
let mut c = TrackCache::default();
|
|
for g in 4..=7 {
|
|
c.groups.insert(g, whole());
|
|
}
|
|
c.evicted_through = Some(5);
|
|
let snap = Snapshot { evicted_through: Some(5), ..Snapshot::default() };
|
|
let plan = local_plan(Some(&c), &snap, Location::new(4, 0), Location::new(7, u64::MAX), 7).expect("a plan");
|
|
assert!(matches!(plan[0], Src::Unknown(4, 5)), "groups 4 and 5 are evicted, as one run");
|
|
assert!(matches!(plan[1], Src::Cache(_)) && matches!(plan[2], Src::Cache(_)));
|
|
assert_eq!(plan.len(), 3);
|
|
}
|
|
}
|