#include "guard.h"
#ifndef OPENSSL_NO_QUIC
#include <aliased_struct-inl.h>
#include <async_wrap-inl.h>
#include <crypto/crypto_util.h>
#include <debug_utils-inl.h>
#include <env-inl.h>
#include <memory_tracker-inl.h>
#include <ngtcp2/ngtcp2.h>
#include <node_bob-inl.h>
#include <node_errors.h>
#include <node_http_common-inl.h>
#include <node_sockaddr-inl.h>
#include <req_wrap-inl.h>
#include <timer_wrap-inl.h>
#include <util-inl.h>
#include <uv.h>
#include <v8.h>
#include <span>
#include "application.h"
#include "bindingdata.h"
#include "cid.h"
#include "data.h"
#include "defs.h"
#include "endpoint.h"
#include "http3.h"
#include "ncrypto.h"
#include "packet.h"
#include "preferredaddress.h"
#include "session.h"
#include "session_manager.h"
#include "sessionticket.h"
#include "streams.h"
#include "tlscontext.h"
#include "transportparams.h"
namespace node {
using v8::Array;
using v8::ArrayBufferView;
using v8::BigInt;
using v8::Boolean;
using v8::Function;
using v8::FunctionCallbackInfo;
using v8::Global;
using v8::HandleScope;
using v8::Int32;
using v8::Integer;
using v8::Just;
using v8::Local;
using v8::LocalVector;
using v8::Maybe;
using v8::MaybeLocal;
using v8::Nothing;
using v8::Number;
using v8::Object;
using v8::ObjectTemplate;
using v8::String;
using v8::Uint32;
using v8::Undefined;
using v8::Value;
namespace quic {
// Listener flags are packed into a single uint32_t bitfield to reduce
// the size of the shared state buffer. Each bit indicates whether a
// corresponding JS callback is registered.
enum class SessionListenerFlags : uint32_t {
PATH_VALIDATION = 1 << 0,
DATAGRAM = 1 << 1,
DATAGRAM_STATUS = 1 << 2,
SESSION_TICKET = 1 << 3,
NEW_TOKEN = 1 << 4,
ORIGIN = 1 << 5,
APPLICATION = 1 << 6
};
inline SessionListenerFlags operator|(SessionListenerFlags a,
SessionListenerFlags b) {
return static_cast<SessionListenerFlags>(static_cast<uint32_t>(a) |
static_cast<uint32_t>(b));
}
inline SessionListenerFlags operator&(SessionListenerFlags a,
SessionListenerFlags b) {
return static_cast<SessionListenerFlags>(static_cast<uint32_t>(a) &
static_cast<uint32_t>(b));
}
inline SessionListenerFlags operator&(uint32_t a, SessionListenerFlags b) {
return static_cast<SessionListenerFlags>(a & static_cast<uint32_t>(b));
}
inline bool operator!(SessionListenerFlags a) {
return static_cast<uint32_t>(a) == 0;
}
inline bool HasListenerFlag(uint32_t flags, SessionListenerFlags flag) {
return !!(flags & flag);
}
// Compute the maximum datagram payload that fits within the peer's
// max_datagram_frame_size transport parameter. The DATAGRAM frame has
// overhead of 1 byte (frame type) + variable-length integer encoding
// of the payload length. This mirrors the check in ngtcp2's
// ngtcp2_pkt_datagram_framelen (1 + varint_len(payload) + payload).
uint64_t MaxDatagramPayload(uint64_t max_frame_size) {
// A DATAGRAM frame needs at least 1 (type) + 1 (varint) + 0 (data).
if (max_frame_size < 2) return 0;
// QUIC variable-length integer encoding sizes (RFC 9000 Section 16).
auto varint_len = [](uint64_t n) -> uint64_t {
if (n < 64) return 1;
if (n < 16384) return 2;
if (n < 1073741824) return 4;
return 8;
};
// Start with the optimistic payload assuming minimum varint (1 byte).
uint64_t payload = max_frame_size - 2;
// If the payload requires a larger varint, the overhead increases.
// Recompute with the actual varint length of the candidate payload.
uint64_t overhead = 1 + varint_len(payload);
if (overhead + payload > max_frame_size) {
payload = max_frame_size - 1 - varint_len(max_frame_size - 3);
}
return payload;
}
#define SESSION_STATE(V) \
V(LISTENER_FLAGS, listener_flags, uint32_t) \
V(CLOSING, closing, uint8_t) \
V(GRACEFUL_CLOSE, graceful_close, uint8_t) \
V(SILENT_CLOSE, silent_close, uint8_t) \
V(STATELESS_RESET, stateless_reset, uint8_t) \
V(HANDSHAKE_COMPLETED, handshake_completed, uint8_t) \
V(HANDSHAKE_CONFIRMED, handshake_confirmed, uint8_t) \
V(STREAM_OPEN_ALLOWED, stream_open_allowed, uint8_t) \
V(PRIORITY_SUPPORTED, priority_supported, uint8_t) \
V(HEADERS_SUPPORTED, headers_supported, uint8_t) \
V(STREAM_CALLBACKS_SUPPORTED, stream_callbacks_supported, uint8_t) \
V(WRAPPED, wrapped, uint8_t) \
V(APPLICATION_TYPE, application_type, uint8_t) \
V(NO_ERROR_CODE, no_error_code, error_code) \
V(INTERNAL_ERROR_CODE, internal_error_code, error_code) \
V(REQUEST_REJECTED_CODE, request_rejected_code, error_code) \
V(MAX_DATAGRAM_SIZE, max_datagram_size, uint16_t) \
V(LAST_DATAGRAM_ID, last_datagram_id, datagram_id) \
V(MAX_PENDING_DATAGRAMS, max_pending_datagrams, uint16_t)
#define SESSION_STATS(V) \
V(CREATED_AT, created_at) \
V(DESTROYED_AT, destroyed_at) \
V(CLOSING_AT, closing_at) \
V(HANDSHAKE_COMPLETED_AT, handshake_completed_at) \
V(HANDSHAKE_CONFIRMED_AT, handshake_confirmed_at) \
V(BYTES_RECEIVED, bytes_received) \
V(BIDI_IN_STREAM_COUNT, bidi_in_stream_count) \
V(BIDI_OUT_STREAM_COUNT, bidi_out_stream_count) \
V(UNI_IN_STREAM_COUNT, uni_in_stream_count) \
V(UNI_OUT_STREAM_COUNT, uni_out_stream_count) \
V(MAX_BYTES_IN_FLIGHT, max_bytes_in_flight) \
V(BYTES_IN_FLIGHT, bytes_in_flight) \
V(BLOCK_COUNT, block_count) \
V(CWND, cwnd) \
V(LATEST_RTT, latest_rtt) \
V(MIN_RTT, min_rtt) \
V(RTTVAR, rttvar) \
V(SMOOTHED_RTT, smoothed_rtt) \
V(SSTHRESH, ssthresh) \
V(PKT_SENT, pkt_sent) \
V(BYTES_SENT, bytes_sent) \
V(PKT_RECV, pkt_recv) \
V(BYTES_RECV, bytes_recv) \
V(PKT_LOST, pkt_lost) \
V(BYTES_LOST, bytes_lost) \
V(PING_RECV, ping_recv) \
V(PKT_DISCARDED, pkt_discarded) \
V(DATAGRAMS_RECEIVED, datagrams_received) \
V(DATAGRAMS_SENT, datagrams_sent) \
V(DATAGRAMS_ACKNOWLEDGED, datagrams_acknowledged) \
V(DATAGRAMS_LOST, datagrams_lost) \
V(STREAMS_IDLE_TIMED_OUT, streams_idle_timed_out)
#define NO_SIDE_EFFECT true
#define SIDE_EFFECT false
#define SESSION_JS_METHODS(V) \
V(Destroy, destroy, SIDE_EFFECT) \
V(GetRemoteAddress, getRemoteAddress, NO_SIDE_EFFECT) \
V(GetServername, getServername, NO_SIDE_EFFECT) \
V(GetAlpnProtocol, getAlpnProtocol, NO_SIDE_EFFECT) \
V(GetLocalAddress, getLocalAddress, NO_SIDE_EFFECT) \
V(GetCertificate, getCertificate, NO_SIDE_EFFECT) \
V(GetEphemeralKeyInfo, getEphemeralKey, NO_SIDE_EFFECT) \
V(GetPeerCertificate, getPeerCertificate, NO_SIDE_EFFECT) \
V(GracefulClose, gracefulClose, SIDE_EFFECT) \
V(SilentClose, silentClose, SIDE_EFFECT) \
V(UpdateKey, updateKey, SIDE_EFFECT) \
V(OpenStream, openStream, SIDE_EFFECT) \
V(SendDatagram, sendDatagram, SIDE_EFFECT) \
V(LocalTransportParams, localTransportParams, NO_SIDE_EFFECT) \
V(RemoteTransportParams, remoteTransportParams, NO_SIDE_EFFECT) \
V(ApplicationOptions, applicationOptions, NO_SIDE_EFFECT)
struct Session::State final {
#define V(_, name, type) type name;
SESSION_STATE(V)
#undef V
};
STAT_STRUCT(Session, SESSION)
using SessionStateArena = AliasedStructArena<Session::State>;
using SessionStatsArena = AliasedStructArena<Session::Stats>;
// Session uses arena-allocated stats, not AliasedStruct, so override the
// STAT_* macros to use impl_->stats() instead of stats_.Data().
#undef STAT_INCREMENT
#undef STAT_INCREMENT_N
#undef STAT_RECORD_TIMESTAMP
#undef STAT_SET
#undef STAT_GET
#define STAT_INCREMENT(Type, name) \
IncrementStat<Type, &Type::name>(impl_->stats());
#define STAT_INCREMENT_N(Type, name, amt) \
IncrementStat<Type, &Type::name>(impl_->stats(), amt);
#define STAT_RECORD_TIMESTAMP(Type, name) \
RecordTimestampStat<Type, &Type::name>(impl_->stats());
#define STAT_SET(Type, name, val) \
SetStat<Type, &Type::name>(impl_->stats(), val)
#define STAT_GET(Type, name) GetStat<Type, &Type::name>(impl_->stats())
namespace {
SessionStateArena& GetSessionStateArena(BindingData& binding) {
if (!binding.session_state_arena_) {
auto* arena = new SessionStateArena();
binding.session_state_arena_ = BindingData::ArenaPtr(
arena, +[](void* p) { delete static_cast<SessionStateArena*>(p); });
}
return *static_cast<SessionStateArena*>(binding.session_state_arena_.get());
}
SessionStatsArena& GetSessionStatsArena(BindingData& binding) {
if (!binding.session_stats_arena_) {
auto* arena = new SessionStatsArena();
binding.session_stats_arena_ = BindingData::ArenaPtr(
arena, +[](void* p) { delete static_cast<SessionStatsArena*>(p); });
}
return *static_cast<SessionStatsArena*>(binding.session_stats_arena_.get());
}
} // namespace
// ============================================================================
class Http3Application;
namespace {
constexpr std::string to_string(PreferredAddress::Policy policy) {
switch (policy) {
case PreferredAddress::Policy::USE_PREFERRED:
return "use";
case PreferredAddress::Policy::IGNORE_PREFERRED:
return "ignore";
}
return "<unknown>";
}
constexpr std::string to_string(Side side) {
switch (side) {
case Side::CLIENT:
return "client";
case Side::SERVER:
return "server";
}
return "<unknown>";
}
constexpr std::string to_string(ngtcp2_encryption_level level) {
switch (level) {
case NGTCP2_ENCRYPTION_LEVEL_1RTT:
return "1rtt";
case NGTCP2_ENCRYPTION_LEVEL_0RTT:
return "0rtt";
case NGTCP2_ENCRYPTION_LEVEL_HANDSHAKE:
return "handshake";
case NGTCP2_ENCRYPTION_LEVEL_INITIAL:
return "initial";
}
return "<unknown>";
}
constexpr std::string to_string(ngtcp2_cc_algo cc_algorithm) {
#define V(name, label) \
case NGTCP2_CC_ALGO_##name: \
return #label;
switch (cc_algorithm) { /* NOLINT(whitespace/newline) */
CC_ALGOS(V)
}
return "<unknown>";
#undef V
}
Maybe<ngtcp2_cc_algo> getAlgoFromString(Environment* env, Local<String> input) {
auto& state = BindingData::Get(env);
#define V(name, str) \
if (input->StringEquals(state.str##_string())) { \
return Just(NGTCP2_CC_ALGO_##name); \
}
CC_ALGOS(V)
#undef V
return Nothing<ngtcp2_cc_algo>();
}
// Qlog is a JSON-based logging format that is being standardized for low-level
// debug logging of QUIC connections and dataflows. The qlog output is generated
// optionally by ngtcp2 for us. The on_qlog_write callback is registered with
// ngtcp2 to emit the qlog information. Every Session will have it's own qlog
// stream.
void on_qlog_write(void* user_data,
uint32_t flags,
const void* data,
size_t len) {
static_cast<Session*>(user_data)->EmitQlog(
flags, std::string_view(static_cast<const char*>(data), len));
}
// Forwards detailed(verbose) debugging information from ngtcp2. Enabled using
// the NODE_DEBUG_NATIVE=NGTCP2 category.
void ngtcp2_debug_log(void* user_data, const char* fmt, ...) {
va_list ap;
va_start(ap, fmt);
std::string format(fmt, strlen(fmt) + 1);
format[strlen(fmt)] = '\n';
// Debug() does not work with the va_list here. So we use vfprintf
// directly instead. Ngtcp2DebugLog is only enabled when the debug
// category is enabled. The thread ID prefix helps distinguish output
// from concurrent sessions across worker threads.
fprintf(stderr, "ngtcp2 ");
vfprintf(stderr, format.c_str(), ap);
va_end(ap);
}
template <typename Opt, PreferredAddress::Policy Opt::*member>
bool SetOption(Environment* env,
Opt* options,
const Local<Object>& object,
const Local<String>& name) {
Local<Value> value;
PreferredAddress::Policy policy = PreferredAddress::Policy::USE_PREFERRED;
if (!object->Get(env->context(), name).ToLocal(&value) ||
!PreferredAddress::tryGetPolicy(env, value).To(&policy)) {
return false;
}
options->*member = policy;
return true;
}
template <typename Opt, TLSContext::Options Opt::*member>
bool SetOption(Environment* env,
Opt* options,
const Local<Object>& object,
const Local<String>& name) {
Local<Value> value;
TLSContext::Options opts;
if (!object->Get(env->context(), name).ToLocal(&value) ||
!TLSContext::Options::From(env, value).To(&opts)) {
return false;
}
options->*member = opts;
return true;
}
template <typename Opt, TransportParams::Options Opt::*member>
bool SetOption(Environment* env,
Opt* options,
const Local<Object>& object,
const Local<String>& name) {
Local<Value> value;
TransportParams::Options opts;
if (!object->Get(env->context(), name).ToLocal(&value) ||
!TransportParams::Options::From(env, value).To(&opts)) {
return false;
}
options->*member = opts;
return true;
}
template <typename Opt, ngtcp2_cc_algo Opt::*member>
bool SetOption(Environment* env,
Opt* options,
const Local<Object>& object,
const Local<String>& name) {
Local<Value> value;
if (!object->Get(env->context(), name).ToLocal(&value)) return false;
if (!value->IsUndefined()) {
ngtcp2_cc_algo algo;
if (value->IsString()) {
if (!getAlgoFromString(env, value.As<String>()).To(&algo)) {
THROW_ERR_INVALID_ARG_VALUE(env, "The cc_algorithm option is invalid");
return false;
}
} else {
if (!value->IsInt32()) {
THROW_ERR_INVALID_ARG_VALUE(
env, "The cc_algorithm option must be a string or an integer");
return false;
}
Local<Int32> num;
if (!value->ToInt32(env->context()).ToLocal(&num)) {
THROW_ERR_INVALID_ARG_VALUE(env, "The cc_algorithm option is invalid");
return false;
}
switch (num->Value()) {
#define V(name, _) \
case NGTCP2_CC_ALGO_##name: \
break;
CC_ALGOS(V)
#undef V
default:
THROW_ERR_INVALID_ARG_VALUE(env,
"The cc_algorithm option is invalid");
return false;
}
algo = FromV8Value<ngtcp2_cc_algo>(num);
}
options->*member = algo;
}
return true;
}
template <typename Opt, uint8_t Opt::*member>
bool SetOption(Environment* env,
Opt* options,
const Local<Object>& object,
const Local<String>& name) {
Local<Value> value;
if (!object->Get(env->context(), name).ToLocal(&value)) return false;
if (!value->IsUndefined()) {
if (!value->IsUint32()) {
Utf8Value nameStr(env->isolate(), name);
THROW_ERR_INVALID_ARG_VALUE(
env, "The %s option must be an uint8", *nameStr);
return false;
}
uint32_t val = value.As<Uint32>()->Value();
if (val > 255) {
Utf8Value nameStr(env->isolate(), name);
THROW_ERR_INVALID_ARG_VALUE(
env, "The %s option must be <= 255", *nameStr);
return false;
}
options->*member = static_cast<uint8_t>(val);
}
return true;
}
} // namespace
// ============================================================================
Session::Config::Config(Environment* env,
Side side,
const Options& options,
uint32_t version,
const SocketAddress& local_address,
const SocketAddress& remote_address,
const CID& dcid,
const CID& scid,
const CID& ocid)
: side(side),
options(options),
version(version),
local_address(local_address),
remote_address(remote_address),
dcid(dcid),
scid(scid),
ocid(ocid) {
ngtcp2_settings_default(&settings);
settings.initial_ts = uv_hrtime();
// Advertise all versions ngtcp2 supports for compatible version
// negotiation (RFC 9368). The preferred list orders the newest
// version first so that negotiation upgrades when possible. The
// initial packet version (options.version) defaults to V1 for
// maximum compatibility with peers that don't support version
// negotiation.
static const uint32_t kSupportedVersions[] = {NGTCP2_PROTO_VER_V2,
NGTCP2_PROTO_VER_V1};
settings.preferred_versions = kSupportedVersions;
settings.preferred_versionslen = std::size(kSupportedVersions);
settings.available_versions = kSupportedVersions;
settings.available_versionslen = std::size(kSupportedVersions);
// TODO(@jasnell): Path MTU Discovery is disabled because libuv does not
// currently expose the IP_DONTFRAG / IP_MTU_DISCOVER socket options
// needed for PMTUD probes to work correctly. Revisit when libuv adds
// support or if we bypass libuv for the UDP socket options.
settings.no_pmtud = 1;
// Per the ngtcp2 documentation, when no_tx_udp_payload_size_shaping is set
// to a non-zero value, it tells ngtcp2 not to limit the UDP payload size to
// NGTCP2_MAX_UDP_PAYLOAD_SIZE` and will instead "use the minimum size among
// the given buffer size, :member:`max_tx_udp_payload_size`, and the
// received max_udp_payload_size QUIC transport parameter." For now, this
// works for us, especially since we do not implement Path MTU discovery.
settings.no_tx_udp_payload_size_shaping = 1;
settings.max_tx_udp_payload_size = options.max_payload_size;
settings.tokenlen = 0;
settings.token = nullptr;
if (options.qlog) {
settings.qlog_write = on_qlog_write;
}
if (env->enabled_debug_list()->enabled(DebugCategory::NGTCP2)) {
settings.log_printf = ngtcp2_debug_log;
}
// The handshake_timeout option is in milliseconds; ngtcp2 expects
// nanoseconds (ngtcp2_duration). UINT64_MAX means no timeout.
settings.handshake_timeout =
options.handshake_timeout == UINT64_MAX
? UINT64_MAX
: options.handshake_timeout * NGTCP2_MILLISECONDS;
// The initial_rtt option is in milliseconds; ngtcp2 expects nanoseconds.
// A value of 0 leaves the ngtcp2 default (333ms) unchanged.
if (options.initial_rtt > 0)
settings.initial_rtt = options.initial_rtt * NGTCP2_MILLISECONDS;
settings.max_stream_window = options.max_stream_window;
settings.max_window = options.max_window;
settings.ack_thresh = options.unacknowledged_packet_threshold;
settings.cc_algo = options.cc_algorithm;
if (side == Side::CLIENT && options.token.has_value()) {
ngtcp2_vec vec = options.token.value();
set_token(vec.base, vec.len, NGTCP2_TOKEN_TYPE_NEW_TOKEN);
}
}
Session::Config::Config(Environment* env,
const Options& options,
const SocketAddress& local_address,
const SocketAddress& remote_address,
const CID& ocid)
: Config(env,
Side::CLIENT,
options,
options.version,
local_address,
remote_address,
CID::Factory::random().Generate(NGTCP2_MIN_INITIAL_DCIDLEN),
options.cid_factory->Generate(),
ocid) {}
void Session::Config::MemoryInfo(MemoryTracker* tracker) const {
tracker->TrackField("options", options);
tracker->TrackField("local_address", local_address);
tracker->TrackField("remote_address", remote_address);
tracker->TrackField("dcid", dcid);
tracker->TrackField("scid", scid);
tracker->TrackField("ocid", ocid);
tracker->TrackField("retry_scid", retry_scid);
}
void Session::Config::set_token(const uint8_t* token,
size_t len,
ngtcp2_token_type type) {
settings.token = token;
settings.tokenlen = len;
settings.token_type = type;
}
void Session::Config::set_token(const RetryToken& token) {
ngtcp2_vec vec = token;
set_token(vec.base, vec.len, NGTCP2_TOKEN_TYPE_RETRY);
}
void Session::Config::set_token(const RegularToken& token) {
ngtcp2_vec vec = token;
set_token(vec.base, vec.len, NGTCP2_TOKEN_TYPE_NEW_TOKEN);
}
std::string Session::Config::ToString() const {
DebugIndentScope indent;
auto prefix = indent.Prefix();
std::string res("{");
res += prefix + "side: " + to_string(side);
res += prefix + "options: " + options.ToString();
res += prefix + "version: " + std::to_string(version);
res += prefix + "local address: " + local_address.ToString();
res += prefix + "remote address: " + remote_address.ToString();
res += prefix + "dcid: " + dcid.ToString();
res += prefix + "scid: " + scid.ToString();
res += prefix + "ocid: " + ocid.ToString();
res += prefix + "retry scid: " + retry_scid.ToString();
res += prefix + "preferred address cid: " + preferred_address_cid.ToString();
res += indent.Close();
return res;
}
// ============================================================================
Maybe<Session::Options> Session::Options::From(Environment* env,
Local<Value> value) {
if (value.IsEmpty() || !value->IsObject()) {
THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
return Nothing<Options>();
}
auto& state = BindingData::Get(env);
auto params = value.As<Object>();
Options options;
#define SET(name) \
SetOption<Session::Options, &Session::Options::name>( \
env, &options, params, state.name##_string())
if (!SET(version) || !SET(min_version) || !SET(preferred_address_strategy) ||
!SET(transport_params) || !SET(tls_options) || !SET(qlog) ||
!SET(handshake_timeout) || !SET(initial_rtt) ||
!SET(keep_alive_timeout) || !SET(max_stream_window) || !SET(max_window) ||
!SET(max_payload_size) || !SET(unacknowledged_packet_threshold) ||
!SET(cc_algorithm) || !SET(draining_period_multiplier) ||
!SET(max_datagram_send_attempts) || !SET(stream_idle_timeout)) {
return Nothing<Options>();
}
#undef SET
// RFC 9000 Section 10.2 requires the draining period to be at least 3x PTO.
static const uint8_t kMinDrainingPeriodMultiplier = 3;
options.draining_period_multiplier = std::max(
options.draining_period_multiplier, kMinDrainingPeriodMultiplier);
// At least 1 send attempt is required.
options.max_datagram_send_attempts =
std::max(options.max_datagram_send_attempts, static_cast<uint8_t>(1));
// Parse the datagram drop policy from a string option.
{
Local<Value> policy_val;
if (params->Get(env->context(), state.datagram_drop_policy_string())
.ToLocal(&policy_val) &&
!policy_val->IsUndefined()) {
Utf8Value policy_str(env->isolate(), policy_val);
if (strcmp(*policy_str, "drop-newest") == 0) {
options.datagram_drop_policy = DatagramDropPolicy::DROP_NEWEST;
}
// Default is DROP_OLDEST, no need to check for "drop-oldest".
}
}
// Parse the application-specific options (HTTP/3 qpack settings, etc.).
// These are used if the negotiated ALPN selects Http3ApplicationImpl.
{
Local<Value> app_val;
if (params->Get(env->context(), state.application_string())
.ToLocal(&app_val) &&
!app_val->IsUndefined()) {
if (!Application_Options::From(env, app_val)
.To(&options.application_options)) {
return Nothing<Options>();
}
}
}
// Parse the SNI map from the tls options.
{
Local<Value> tls_val;
if (params->Get(env->context(), state.tls_options_string())
.ToLocal(&tls_val) &&
tls_val->IsObject()) {
Local<Value> sni_val;
if (tls_val.As<Object>()
->Get(env->context(), state.sni_string())
.ToLocal(&sni_val) &&
sni_val->IsObject()) {
auto sni_obj = sni_val.As<Object>();
Local<Array> hostnames;
if (sni_obj->GetOwnPropertyNames(env->context()).ToLocal(&hostnames)) {
for (uint32_t i = 0; i < hostnames->Length(); i++) {
Local<Value> key;
Local<Value> entry_val;
if (!hostnames->Get(env->context(), i).ToLocal(&key) ||
!key->IsString() ||
!sni_obj->Get(env->context(), key).ToLocal(&entry_val)) {
continue;
}
Utf8Value hostname(env->isolate(), key);
auto entry_options = TLSContext::Options::From(env, entry_val);
if (entry_options.IsNothing()) {
return Nothing<Options>();
}
options.sni[std::string(*hostname, hostname.length())] =
entry_options.FromJust();
}
}
}
}
}
// TODO(@jasnell): Later we will also support setting the CID::Factory.
// For now, we're just using the default random factory.
// Parse the optional NEW_TOKEN for address validation on reconnection.
Local<Value> token_val;
if (params->Get(env->context(), state.token_string()).ToLocal(&token_val) &&
token_val->IsArrayBufferView()) {
Store token_store;
if (Store::From(token_val.As<ArrayBufferView>()).To(&token_store)) {
options.token = std::move(token_store);
}
}
return Just<Options>(options);
}
void Session::Options::MemoryInfo(MemoryTracker* tracker) const {
tracker->TrackField("transport_params", transport_params);
tracker->TrackField("crypto_options", tls_options);
tracker->TrackField("cid_factory_ref", cid_factory_ref);
}
std::string Session::Options::ToString() const {
DebugIndentScope indent;
auto prefix = indent.Prefix();
std::string res("{");
res += prefix + "version: " + std::to_string(version);
res += prefix + "min version: " + std::to_string(min_version);
res += prefix +
"preferred address policy: " + to_string(preferred_address_strategy);
res += prefix + "transport params: " + transport_params.ToString();
res += prefix + "crypto options: " + tls_options.ToString();
if (qlog) {
res += prefix + "qlog: yes";
}
if (handshake_timeout == UINT64_MAX) {
res += prefix + "handshake timeout: <none>";
} else {
res += prefix + "handshake timeout: " + std::to_string(handshake_timeout) +
" nanoseconds";
}
if (initial_rtt > 0) {
res += prefix + "initial rtt: " + std::to_string(initial_rtt) +
" milliseconds";
} else {
res += prefix + "initial rtt: <default>";
}
res += prefix + "max stream window: " + std::to_string(max_stream_window);
res += prefix + "max window: " + std::to_string(max_window);
res += prefix + "max payload size: " + std::to_string(max_payload_size);
if (unacknowledged_packet_threshold != 0) {
res += prefix + "unacknowledged packet threshold: " +
std::to_string(unacknowledged_packet_threshold);
} else {
res += prefix + "unacknowledged packet threshold: <default>";
}
res += prefix + "cc algorithm: " + to_string(cc_algorithm);
res += indent.Close();
return res;
}
// ============================================================================
// ngtcp2 static callback functions
// Utility used only within Session::Impl to reduce boilerplate
// Resolves the Session* from ngtcp2 callback arguments. The
// NgTcp2CallbackScope is NOT created here — it is placed at the
// ngtcp2 entry points (Receive, OnTimeout) so that the deferred
// destroy only fires after all callbacks for that call have completed.
#define NGTCP2_CALLBACK_SCOPE(name) \
auto name = Impl::From(conn, user_data); \
if (name == nullptr) return NGTCP2_ERR_CALLBACK_FAILURE;
// Session::Impl maintains most of the internal state of an active Session.
struct Session::Impl final : public MemoryRetainer {
Session* session_;
ArenaSlotBase stats_slot_;
ArenaSlotBase state_slot_;
BaseObjectWeakPtr<Endpoint> endpoint_;
Config config_;
SocketAddress local_address_;
SocketAddress remote_address_;
std::unique_ptr<Application> application_;
StreamsMap streams_;
// qlog chunks produced before the server session was handed to JS.
std::vector<std::pair<uint32_t, std::string>> pending_qlog_;
TimerWrapHandle timer_;
size_t send_scope_depth_ = 0;
QuicError last_error_;
// Datagrams queued for sending. Serialized into packets by
// SendPendingData alongside stream data.
std::deque<Session::PendingDatagram> pending_datagrams_;
PendingStream::PendingStreamQueue pending_bidi_stream_queue_;
PendingStream::PendingStreamQueue pending_uni_stream_queue_;
// When true, the handshake is deferred until the first stream or
// datagram is sent. This is set for client sessions with a session
// ticket, enabling 0-RTT: the first send triggers the handshake
// and the stream/datagram data is included in the 0-RTT flight.
bool handshake_deferred_ = false;
Stats* stats() { return static_cast<Stats*>(stats_slot_.ptr); }
const Stats* stats() const {
return static_cast<const Stats*>(stats_slot_.ptr);
}
State* state() { return static_cast<State*>(state_slot_.ptr); }
const State* state() const {
return static_cast<const State*>(state_slot_.ptr);
}
Impl(Session* session, Endpoint* endpoint, const Config& config)
: session_(session),
endpoint_(endpoint),
config_(config),
local_address_(config.local_address),
remote_address_(config.remote_address),
timer_(session_->env(), [this] { session_->OnTimeout(); }) {
auto& binding = BindingData::Get(env());
stats_slot_ = GetSessionStatsArena(binding).Allocate(env()->isolate());
state_slot_ = GetSessionStateArena(binding).Allocate(env()->isolate());
timer_.Unref();
}
DISALLOW_COPY_AND_MOVE(Impl)
inline bool is_closing() const { return state()->closing; }
~Impl() {
// Ensure that Close() was called before dropping
DCHECK(is_closing());
DCHECK(endpoint_);
// Removing the session from the endpoint may cause the endpoint to be
// destroyed if it is waiting on the last session to be destroyed. Let's
// grab a reference just to be safe for the rest of the function.
BaseObjectPtr<Endpoint> endpoint = endpoint_;
endpoint_.reset();
MaybeStackBuffer<ngtcp2_cid, 10> cids(
ngtcp2_conn_get_scid(*session_, nullptr));
ngtcp2_conn_get_scid(*session_, cids.out());
MaybeStackBuffer<ngtcp2_cid_token2, 10> tokens(
ngtcp2_conn_get_active_dcid2(*session_, nullptr));
ngtcp2_conn_get_active_dcid2(*session_, tokens.out());
endpoint->DisassociateCID(config_.dcid);
endpoint->DisassociateCID(config_.preferred_address_cid);
for (size_t n = 0; n < cids.length(); n++) {
endpoint->DisassociateCID(CID(cids[n]));
}
for (size_t n = 0; n < tokens.length(); n++) {
if (tokens[n].token_present) {
endpoint->DisassociateStatelessResetToken(
StatelessResetToken(&tokens[n].token));
}
}
auto& binding = BindingData::Get(env());
if (stats_slot_) GetSessionStatsArena(binding).ReleaseSlot(stats_slot_);
if (state_slot_) GetSessionStateArena(binding).ReleaseSlot(state_slot_);
endpoint->RemoveSession(config_.scid, remote_address_);
}
void MemoryInfo(MemoryTracker* tracker) const override {
tracker->TrackField("config", config_);
tracker->TrackField("endpoint", endpoint_);
tracker->TrackField("streams", streams_);
tracker->TrackField("local_address", local_address_);
tracker->TrackField("remote_address", remote_address_);
tracker->TrackField("application", application_);
tracker->TrackField("timer", timer_);
size_t qlog_size = 0;
for (const auto& [flags, data] : pending_qlog_) qlog_size += data.size();
tracker->TrackFieldWithSize("pending_qlog", qlog_size);
}
SET_SELF_SIZE(Impl)
SET_MEMORY_INFO_NAME(Session::Impl)
Environment* env() const { return session_->env(); }
// Gets the Session pointer from the user_data void pointer
// provided by ngtcp2.
static Session* From(ngtcp2_conn* conn, void* user_data) {
if (user_data == nullptr) [[unlikely]] {
return nullptr;
}
auto session = static_cast<Session*>(user_data);
if (session->is_destroyed()) [[unlikely]] {
return nullptr;
}
return session;
}
// JavaScript APIs
// TODO(@jasnell): Fast API alternatives for each of these
// Parse optional close error code options: { code, type, reason }
// Returns true on success (including when no options were provided).
// Returns false on validation error (exception thrown).
// Sets *did_set to true if an error code was actually applied.
static bool MaybeSetCloseError(const FunctionCallbackInfo<Value>& args,
int options_index,
Session* session,
bool* did_set = nullptr) {
if (did_set) *did_set = false;
auto env = Environment::GetCurrent(args);
if (args.Length() <= options_index || args[options_index]->IsUndefined()) {
return true;
}
if (!args[options_index]->IsObject()) {
THROW_ERR_INVALID_ARG_TYPE(env, "options must be an object");
return false;
}
auto options = args[options_index].As<Object>();
auto& state = BindingData::Get(env);
auto context = env->context();
// code: bigint (optional)
Local<Value> code_val;
if (!options->Get(context, state.code_string()).ToLocal(&code_val)) {
return false;
}
if (code_val->IsUndefined()) return true;
uint64_t code;
if (code_val->IsBigInt()) {
bool lossless;
code = code_val.As<BigInt>()->Uint64Value(&lossless);
if (!lossless) {
THROW_ERR_INVALID_ARG_VALUE(env, "options.code is too large");
return false;
}
} else if (code_val->IsNumber()) {
code = static_cast<uint64_t>(code_val.As<Number>()->Value());
} else {
THROW_ERR_INVALID_ARG_TYPE(env,
"options.code must be a bigint or number");
return false;
}
// type: string (optional, default 'transport')
Local<Value> type_val;
if (!options->Get(context, state.type_string()).ToLocal(&type_val)) {
return false;
}
bool is_application = false;
if (!type_val->IsUndefined()) {
if (type_val->StrictEquals(state.application_string())) {
is_application = true;
} else if (!type_val->StrictEquals(state.transport_string())) {
THROW_ERR_INVALID_ARG_VALUE(
env, "options.type must be 'transport' or 'application'");
return false;
}
}
// reason: string (optional)
std::string reason;
Local<Value> reason_val;
if (!options->Get(context, state.reason_string()).ToLocal(&reason_val)) {
return false;
}
if (!reason_val->IsUndefined()) {
Utf8Value reason_str(env->isolate(), reason_val);
reason = std::string(*reason_str, reason_str.length());
}
if (is_application) {
session->SetLastError(QuicError::ForApplication(code, std::move(reason)));
} else {
session->SetLastError(QuicError::ForTransport(code, std::move(reason)));
}
if (did_set) *did_set = true;
return true;
}
JS_METHOD(Destroy) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
// At this layer, Destroy should only be called once. At the
// JavaScript layer calling destroy() multiple times should be
// an idempotent operation. Be sure to check for that there
// as we strictly enforce it here.
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
// args[0] is the optional close error options object.
bool has_close_options = false;
if (!MaybeSetCloseError(args, 0, session, &has_close_options)) return;
// If an error code was provided by the caller, send CONNECTION_CLOSE
// with that code before destroying. SendConnectionClose writes the
// packet and hands it to the endpoint — it doesn't wait for ack.
if (has_close_options) {
session->SendConnectionClose();
}
session->Destroy();
}
// The SNI servername: null until the TLS parameters are final, then the
// host name string, or false if the handshake produced no SNI.
JS_METHOD(GetServername) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed() || !session->tls_info_ready()) {
return args.GetReturnValue().SetNull();
}
auto sn = session->tls_session().servername();
if (sn.empty()) return args.GetReturnValue().Set(false);
Local<Value> ret;
if (ToV8Value(env->context(), sn).ToLocal(&ret)) {
args.GetReturnValue().Set(ret);
}
}
// The negotiated ALPN protocol: null until the TLS parameters are final,
// then the protocol string.
JS_METHOD(GetAlpnProtocol) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed() || !session->tls_info_ready()) {
return args.GetReturnValue().SetNull();
}
auto proto = session->tls_session().protocol();
// QUIC requires ALPN
DCHECK(!proto.empty());
Local<Value> ret;
if (ToV8Value(env->context(), proto).ToLocal(&ret)) {
args.GetReturnValue().Set(ret);
}
}
JS_METHOD(GetRemoteAddress) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
auto address = session->remote_address();
args.GetReturnValue().Set(
SocketAddressBase::Create(env, std::make_shared<SocketAddress>(address))
->object());
}
JS_METHOD(GetLocalAddress) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
auto address = session->local_address();
args.GetReturnValue().Set(
SocketAddressBase::Create(env, std::make_shared<SocketAddress>(address))
->object());
}
JS_METHOD(GetCertificate) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
Local<Value> ret;
if (session->tls_session().cert(env).ToLocal(&ret))
args.GetReturnValue().Set(ret);
}
JS_METHOD(GetEphemeralKeyInfo) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
Local<Object> ret;
if (!session->is_server() &&
session->tls_session().ephemeral_key(env).ToLocal(&ret))
args.GetReturnValue().Set(ret);
}
JS_METHOD(GetPeerCertificate) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
Local<Value> ret;
if (session->tls_session().peer_cert(env).ToLocal(&ret))
args.GetReturnValue().Set(ret);
}
JS_METHOD(GracefulClose) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
// args[0] is the optional close error options object.
if (!MaybeSetCloseError(args, 0, session)) return;
session->Close(CloseMethod::GRACEFUL);
}
JS_METHOD(SilentClose) {
// This is exposed for testing purposes only!
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
session->Close(CloseMethod::SILENT);
}
JS_METHOD(UpdateKey) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
// Initiating a key update may fail if it is done too early (either
// before the TLS handshake has been confirmed or while a previous
// key update is being processed). When it fails, InitiateKeyUpdate()
// will return false.
SendPendingDataScope send_scope(session);
args.GetReturnValue().Set(session->tls_session().InitiateKeyUpdate());
}
JS_METHOD(OpenStream) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) [[unlikely]] {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
DCHECK(args[0]->IsUint32());
// GetDataQueueFromSource handles type validation.
std::shared_ptr<DataQueue> data_source;
if (!Stream::GetDataQueueFromSource(env, args[1]).To(&data_source))
[[unlikely]] {
return THROW_ERR_INVALID_ARG_VALUE(env, "Invalid data source");
}
session->impl_->handshake_deferred_ = false;
SendPendingDataScope send_scope(session);
auto direction = FromV8Value<Direction>(args[0]);
Local<Object> stream;
if (session->OpenStream(direction, std::move(data_source)).ToLocal(&stream))
[[likely]] {
args.GetReturnValue().Set(stream);
}
}
JS_METHOD(SendDatagram) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
if (session->is_destroyed()) {
return THROW_ERR_INVALID_STATE(env, "Session is destroyed");
}
DCHECK(args[0]->IsArrayBufferView());
session->impl_->handshake_deferred_ = false;
SendPendingDataScope send_scope(session);
Store store;
if (!Store::From(args[0].As<ArrayBufferView>()).To(&store)) {
return;
}
args.GetReturnValue().Set(
BigInt::New(env->isolate(), session->SendDatagram(std::move(store))));
}
JS_METHOD(LocalTransportParams) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
ngtcp2_conn* conn = *session;
TransportParams params(ngtcp2_conn_get_local_transport_params(conn));
Local<Object> obj;
if (params.ToObject(env).ToLocal(&obj)) {
args.GetReturnValue().Set(obj);
}
}
JS_METHOD(RemoteTransportParams) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
ngtcp2_conn* conn = *session;
auto params = ngtcp2_conn_get_remote_transport_params(conn);
if (params == nullptr) {
// Remote transport parameters are not yet available.
return args.GetReturnValue().SetUndefined();
}
TransportParams tp(params);
Local<Object> obj;
if (tp.ToObject(env).ToLocal(&obj)) {
args.GetReturnValue().Set(obj);
}
}
JS_METHOD(ApplicationOptions) {
auto env = Environment::GetCurrent(args);
Session* session;
ASSIGN_OR_RETURN_UNWRAP(&session, args.This());
Local<Object> obj;
if (!session->has_application()) {
// The application has not yet been selected (ALPN negotiation is not
// yet complete on the server) or the session has been destroyed. In
// either case, the application options are not available.
return args.GetReturnValue().SetUndefined();
}
auto& options = session->application().options();
if (options.ToObject(env).ToLocal(&obj)) {
args.GetReturnValue().Set(obj);
}
}
// Internal ngtcp2 callbacks
static int on_acknowledge_stream_data_offset(ngtcp2_conn* conn,
stream_id stream_id,
uint64_t offset,
uint64_t datalen,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
// The callback will be invoked with datalen 0 if a zero-length
// stream frame with fin flag set is received. In that case, let's
// just ignore it.
// Per ngtcp2, the range of bytes that are being acknowledged here
// are `[offset, offset + datalen]` but we only really care about
// the datalen as our accounting does not track the offset and
// acknowledges should never come out of order here.
if (datalen == 0) return NGTCP2_SUCCESS;
return session->application().AcknowledgeStreamData(stream_id, datalen)
? NGTCP2_SUCCESS
: NGTCP2_ERR_CALLBACK_FAILURE;
}
static int on_acknowledge_datagram(ngtcp2_conn* conn,
datagram_id dgram_id,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->DatagramStatus(dgram_id, DatagramStatus::ACKNOWLEDGED);
return NGTCP2_SUCCESS;
}
static int on_cid_status(ngtcp2_conn* conn,
ngtcp2_connection_id_status_type type,
uint64_t seq,
const ngtcp2_cid* cid,
const ngtcp2_stateless_reset_token* token,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
std::optional<StatelessResetToken> maybe_reset_token;
if (token != nullptr) maybe_reset_token.emplace(token);
auto& endpoint = session->endpoint();
switch (type) {
case NGTCP2_CONNECTION_ID_STATUS_TYPE_ACTIVATE: {
if (token != nullptr) {
endpoint.AssociateStatelessResetToken(StatelessResetToken(token),
session);
}
break;
}
case NGTCP2_CONNECTION_ID_STATUS_TYPE_DEACTIVATE: {
if (token != nullptr) {
endpoint.DisassociateStatelessResetToken(StatelessResetToken(token));
}
break;
}
}
return NGTCP2_SUCCESS;
}
static int on_extend_max_remote_streams_bidi(ngtcp2_conn* conn,
uint64_t max_streams,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Debug(
session, "Max remote bidi streams increased to %" PRIu64, max_streams);
return NGTCP2_SUCCESS;
}
static int on_extend_max_remote_streams_uni(ngtcp2_conn* conn,
uint64_t max_streams,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Debug(session, "Max remote uni streams increased to %" PRIu64, max_streams);
return NGTCP2_SUCCESS;
}
static int on_extend_max_streams_bidi(ngtcp2_conn* conn,
uint64_t max_streams,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->ProcessPendingBidiStreams();
return NGTCP2_SUCCESS;
}
static int on_extend_max_streams_uni(ngtcp2_conn* conn,
uint64_t max_streams,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->ProcessPendingUniStreams();
return NGTCP2_SUCCESS;
}
static int on_extend_max_stream_data(ngtcp2_conn* conn,
stream_id stream_id,
uint64_t max_data,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
if (auto* stream = Stream::From(stream_user_data)) {
session->application().ExtendMaxStreamData(stream, max_data);
}
return NGTCP2_SUCCESS;
}
static int on_get_new_cid(ngtcp2_conn* conn,
ngtcp2_cid* cid,
ngtcp2_stateless_reset_token* token,
size_t cidlen,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->GenerateNewConnectionId(cid, cidlen, token);
return NGTCP2_SUCCESS;
}
static int on_handshake_completed(ngtcp2_conn* conn, void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
return session->HandshakeCompleted() ? NGTCP2_SUCCESS
: NGTCP2_ERR_CALLBACK_FAILURE;
}
static int on_handshake_confirmed(ngtcp2_conn* conn, void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->HandshakeConfirmed();
return NGTCP2_SUCCESS;
}
static int on_lost_datagram(ngtcp2_conn* conn,
datagram_id dgram_id,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->DatagramStatus(dgram_id, DatagramStatus::LOST);
return NGTCP2_SUCCESS;
}
static int on_path_validation(ngtcp2_conn* conn,
uint32_t flags,
const ngtcp2_path* path,
const ngtcp2_path* old_path,
ngtcp2_path_validation_result res,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
bool flag_preferred_address =
flags & NGTCP2_PATH_VALIDATION_FLAG_PREFERRED_ADDR;
ValidatedPath newValidatedPath{
std::make_shared<SocketAddress>(path->local.addr),
std::make_shared<SocketAddress>(path->remote.addr)};
std::optional<ValidatedPath> oldValidatedPath = std::nullopt;
if (old_path != nullptr) {
oldValidatedPath =
ValidatedPath{std::make_shared<SocketAddress>(old_path->local.addr),
std::make_shared<SocketAddress>(old_path->remote.addr)};
}
session->EmitPathValidation(static_cast<PathValidationResult>(res),
PathValidationFlags{flag_preferred_address},
newValidatedPath,
oldValidatedPath);
return NGTCP2_SUCCESS;
}
static int on_receive_datagram(ngtcp2_conn* conn,
uint32_t flags,
const uint8_t* data,
size_t datalen,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->DatagramReceived(
data,
datalen,
DatagramReceivedFlags{
.early = (flags & NGTCP2_DATAGRAM_FLAG_0RTT) ==
NGTCP2_DATAGRAM_FLAG_0RTT,
});
return NGTCP2_SUCCESS;
}
static int on_receive_new_token(ngtcp2_conn* conn,
const uint8_t* token,
size_t tokenlen,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Debug(session, "Received NEW_TOKEN (%zu bytes)", tokenlen);
session->EmitNewToken(token, tokenlen);
return NGTCP2_SUCCESS;
}
static int on_receive_rx_key(ngtcp2_conn* conn,
ngtcp2_encryption_level level,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
CHECK(!session->is_server());
if (level != NGTCP2_ENCRYPTION_LEVEL_1RTT) return NGTCP2_SUCCESS;
// If the application was already started via on_receive_tx_key
// (0-RTT path), this is a no-op.
if (session->application().is_started()) return NGTCP2_SUCCESS;
Debug(session,
"Receiving RX key for level %s for dcid %s",
to_string(level),
session->config().dcid);
return session->application().Start() ? NGTCP2_SUCCESS
: NGTCP2_ERR_CALLBACK_FAILURE;
}
static int on_receive_stateless_reset(ngtcp2_conn* conn,
const ngtcp2_pkt_stateless_reset2* sr,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Debug(session, "Received stateless reset from peer");
// This callback is informational. ngtcp2 has already set the
// connection state to NGTCP2_CS_DRAINING before invoking this
// callback, and ngtcp2_conn_read_pkt will return
// NGTCP2_ERR_DRAINING. The actual close handling happens in
// Session::Receive when it processes that return value and
// checks this flag.
session->impl_->state()->stateless_reset = 1;
return NGTCP2_SUCCESS;
}
static int on_receive_stream_data(ngtcp2_conn* conn,
uint32_t flags,
stream_id stream_id,
uint64_t offset,
const uint8_t* data,
size_t datalen,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Stream::ReceiveDataFlags data_flags{
// The fin flag indicates that this is the last chunk of data we will
// receive on this stream.
.fin = (flags & NGTCP2_STREAM_DATA_FLAG_FIN) ==
NGTCP2_STREAM_DATA_FLAG_FIN,
// Stream data is early if it is received before the TLS handshake is
// complete.
.early = (flags & NGTCP2_STREAM_DATA_FLAG_0RTT) ==
NGTCP2_STREAM_DATA_FLAG_0RTT,
};
// We received data for a stream! What we don't know yet at this point
// is whether the application wants us to treat this as a control stream
// data (something the application will handle on its own) or a user stream
// data (something that we should create a Stream handle for that is passed
// out to JavaScript). HTTP3, for instance, will generally create three
// control stream in either direction and we want to make sure those are
// never exposed to users and that we don't waste time creating Stream
// handles for them. So, what we do here is pass the stream data on to the
// application for processing. If it ends up being a user stream, the
// application will handle creating the Stream handle and passing that off
// to the JavaScript side.
if (!session->application().ReceiveStreamData(
stream_id, data, datalen, data_flags, stream_user_data)) {
return NGTCP2_ERR_CALLBACK_FAILURE;
}
return NGTCP2_SUCCESS;
}
static int on_receive_tx_key(ngtcp2_conn* conn,
ngtcp2_encryption_level level,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session);
// For SERVER: fires at 1RTT — start the application after handshake.
// For CLIENT: fires at 0RTT — start the application early so that
// HTTP/3 control/QPACK streams are bound before 0-RTT requests.
// Without this, nghttp3_conn_submit_request asserts because the
// QPACK encoder stream isn't bound yet.
if (session->is_server()) {
if (level != NGTCP2_ENCRYPTION_LEVEL_1RTT) return NGTCP2_SUCCESS;
} else {
if (level != NGTCP2_ENCRYPTION_LEVEL_0RTT) return NGTCP2_SUCCESS;
}
// application_ may be null if ALPN selection hasn't happened yet
// (e.g., ALPN mismatch causes the handshake to fail during key
// installation). Without an application, we can't start.
if (!session->impl_->application_) return NGTCP2_ERR_CALLBACK_FAILURE;
Debug(session,
"Receiving TX key for level %s for dcid %s",
to_string(level),
session->config().dcid);
return session->application().Start() ? NGTCP2_SUCCESS
: NGTCP2_ERR_CALLBACK_FAILURE;
}
static int on_receive_version_negotiation(ngtcp2_conn* conn,
const ngtcp2_pkt_hd* hd,
const uint32_t* sv,
size_t nsv,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->EmitVersionNegotiation(*hd, sv, nsv);
return NGTCP2_SUCCESS;
}
static int on_remove_connection_id(ngtcp2_conn* conn,
const ngtcp2_cid* cid,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
session->endpoint().DisassociateCID(CID(cid));
return NGTCP2_SUCCESS;
}
static int on_select_preferred_address(ngtcp2_conn* conn,
ngtcp2_path* dest,
const ngtcp2_preferred_addr* paddr,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
PreferredAddress preferred_address(dest, paddr);
session->SelectPreferredAddress(&preferred_address);
return NGTCP2_SUCCESS;
}
static int on_stream_close(ngtcp2_conn* conn,
uint32_t flags,
stream_id stream_id,
error_code app_error_code,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
// If the peer closes a stream, we return the credit to allow a new one:
session->ExtendMaxStreams(stream_id);
if (!session->has_application()) return NGTCP2_SUCCESS;
// Stream here may be null (e.g. if already destroyed locally) so we also
// pass stream_id to the calls below as well.
auto* stream = Stream::From(stream_user_data);
if (flags & NGTCP2_STREAM_CLOSE_FLAG_APP_ERROR_CODE_SET) {
session->application().ReceiveStreamClose(
stream_id, stream, QuicError::ForApplication(app_error_code));
} else {
session->application().ReceiveStreamClose(stream_id, stream);
}
return NGTCP2_SUCCESS;
}
static int on_stream_open(ngtcp2_conn* conn, stream_id id, void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
if (!session->application().ReceiveStreamOpen(id)) {
return NGTCP2_ERR_CALLBACK_FAILURE;
}
return NGTCP2_SUCCESS;
}
static int on_stream_reset(ngtcp2_conn* conn,
stream_id stream_id,
uint64_t final_size,
error_code app_error_code,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
auto* stream = Stream::From(stream_user_data);
if (stream == nullptr) return NGTCP2_SUCCESS;
session->application().ReceiveStreamReset(
stream, final_size, QuicError::ForApplication(app_error_code));
return NGTCP2_SUCCESS;
}
static int on_receive_stream_stop_sending(ngtcp2_conn* conn,
stream_id stream_id,
error_code app_error_code,
void* user_data,
void* stream_user_data) {
NGTCP2_CALLBACK_SCOPE(session)
auto* stream = Stream::From(stream_user_data);
if (stream == nullptr) return NGTCP2_SUCCESS;
session->application().ReceiveStreamStopSending(
stream, QuicError::ForApplication(app_error_code));
return NGTCP2_SUCCESS;
}
static void on_rand(uint8_t* dest,
size_t destlen,
const ngtcp2_rand_ctx* rand_ctx) {
CHECK(ncrypto::CSPRNG(dest, destlen));
}
static int on_early_data_rejected(ngtcp2_conn* conn, void* user_data) {
auto session = Impl::From(conn, user_data);
if (session == nullptr) return NGTCP2_ERR_CALLBACK_FAILURE;
Debug(session, "Early data was rejected");
if (session->impl_->application_) {
session->application().EarlyDataRejected();
}
return NGTCP2_SUCCESS;
}
static int on_begin_path_validation(ngtcp2_conn* conn,
uint32_t flags,
const ngtcp2_path* path,
const ngtcp2_path* fallback_path,
void* user_data) {
NGTCP2_CALLBACK_SCOPE(session)
Debug(session, "Path validation started");
return NGTCP2_SUCCESS;
}
static constexpr ngtcp2_callbacks CLIENT = {
ngtcp2_crypto_client_initial_cb,
nullptr, // recv_client_initial
ngtcp2_crypto_recv_crypto_data_cb,
on_handshake_completed,
on_receive_version_negotiation,
ngtcp2_crypto_encrypt_cb,
ngtcp2_crypto_decrypt_cb,
ngtcp2_crypto_hp_mask_cb,
on_receive_stream_data,
on_acknowledge_stream_data_offset,
on_stream_open,
on_stream_close,
nullptr, // recv_stateless_reset (deprecated, use v2 below)
ngtcp2_crypto_recv_retry_cb,
on_extend_max_streams_bidi,
on_extend_max_streams_uni,
on_rand,
nullptr, // get_new_connection_id (deprecated, use v2 below)
on_remove_connection_id,
ngtcp2_crypto_update_key_cb,
on_path_validation,
on_select_preferred_address,
on_stream_reset,
on_extend_max_remote_streams_bidi,
on_extend_max_remote_streams_uni,
on_extend_max_stream_data,
nullptr, // dcid_status (deprecated, use v2 below)
on_handshake_confirmed,
on_receive_new_token,
ngtcp2_crypto_delete_crypto_aead_ctx_cb,
ngtcp2_crypto_delete_crypto_cipher_ctx_cb,
on_receive_datagram,
on_acknowledge_datagram,
on_lost_datagram,
nullptr, // get_path_challenge_data (deprecated, use v2 below)
nullptr, // stream_stop_sending (deprecated, use v2 below)
ngtcp2_crypto_version_negotiation_cb,
on_receive_rx_key,
on_receive_tx_key,
on_early_data_rejected,
on_begin_path_validation,
on_receive_stateless_reset,
on_get_new_cid,
on_cid_status,
ngtcp2_crypto_get_path_challenge_data2_cb,
on_receive_stream_stop_sending,
#ifdef NGTCP2_CALLBACKS_V5
nullptr, // stream_close2
#endif
};
static constexpr ngtcp2_callbacks SERVER = {
nullptr, // client_initial
ngtcp2_crypto_recv_client_initial_cb,
ngtcp2_crypto_recv_crypto_data_cb,
on_handshake_completed,
nullptr, // recv_version_negotiation
ngtcp2_crypto_encrypt_cb,
ngtcp2_crypto_decrypt_cb,
ngtcp2_crypto_hp_mask_cb,
on_receive_stream_data,
on_acknowledge_stream_data_offset,
on_stream_open,
on_stream_close,
nullptr, // recv_stateless_reset (deprecated, use v2 below)
nullptr, // recv_retry
on_extend_max_streams_bidi,
on_extend_max_streams_uni,
on_rand,
nullptr, // get_new_connection_id (deprecated, use v2 below)
on_remove_connection_id,
ngtcp2_crypto_update_key_cb,
on_path_validation,
nullptr, // select_preferred_addr
on_stream_reset,
on_extend_max_remote_streams_bidi,
on_extend_max_remote_streams_uni,
on_extend_max_stream_data,
nullptr, // dcid_status (deprecated, use v2 below)
nullptr, // handshake_confirmed
nullptr, // recv_new_token
ngtcp2_crypto_delete_crypto_aead_ctx_cb,
ngtcp2_crypto_delete_crypto_cipher_ctx_cb,
on_receive_datagram,
on_acknowledge_datagram,
on_lost_datagram,
nullptr, // get_path_challenge_data (deprecated, use v2 below)
nullptr, // stream_stop_sending (deprecated, use v2 below)
ngtcp2_crypto_version_negotiation_cb,
nullptr, // recv_rx_key
on_receive_tx_key,
on_early_data_rejected,
on_begin_path_validation,
on_receive_stateless_reset,
on_get_new_cid,
on_cid_status,
ngtcp2_crypto_get_path_challenge_data2_cb,
on_receive_stream_stop_sending,
#ifdef NGTCP2_CALLBACKS_V5
nullptr, // stream_close2
#endif
};
};
#undef NGTCP2_CALLBACK_SCOPE
// ============================================================================
Session::SendPendingDataScope::SendPendingDataScope(Session* session)
: session(session) {
CHECK_NOT_NULL(session);
CHECK(!session->is_destroyed());
++session->impl_->send_scope_depth_;
}
Session::SendPendingDataScope::SendPendingDataScope(
const BaseObjectPtr<Session>& session)
: session(session.get()) {
CHECK_NOT_NULL(session);
CHECK(!session->is_destroyed());
++session->impl_->send_scope_depth_;
}
Session::SendPendingDataScope::~SendPendingDataScope() {
if (session->is_destroyed()) return;
DCHECK_GE(session->impl_->send_scope_depth_, 1);
Debug(session, "Send Scope Depth %zu", session->impl_->send_scope_depth_);
if (--session->impl_->send_scope_depth_ == 0 &&
session->impl_->application_ && !session->impl_->handshake_deferred_) {
session->SendPendingData();
}
}
// ============================================================================
std::string Session::StreamData::ToString() const {
DebugIndentScope indent;
size_t total_bytes = 0;
for (size_t n = 0; n < count; n++) {
total_bytes += data[n].len;
}
auto prefix = indent.Prefix();
std::string res("{");
res += prefix + "count: " + std::to_string(count);
res += prefix + "id: " + std::to_string(id);
res += prefix + "fin: " + std::to_string(fin);
res += prefix + "total: " + std::to_string(total_bytes);
res += indent.Close();
return res;
}
Packet::Ptr Session::CreateStreamDataPacket() {
return endpoint().CreatePacket(
remote_address(), max_packet_size(), "stream data");
}
// Attempts to pack a pending datagram into the current packet.
// Returns the nwrite value from ngtcp2_conn_writev_datagram.
// On fatal error, closes the session and returns the error code.
// The caller should check:
// > 0: packet is complete, send it (pos was NOT advanced — caller
// must add nwrite to pos and send)
// NGTCP2_ERR_WRITE_MORE: datagram packed, room for more
// 0: congestion controlled or doesn't fit, datagram stays in queue
// < 0 (other): fatal error, session already closed
ssize_t Session::TryWritePendingDatagram(PathStorage* path,
uint8_t* dest,
size_t destlen,
uint64_t ts) {
CHECK(HasPendingDatagrams());
auto max_attempts = config().options.max_datagram_send_attempts;
// Skip datagrams that have already exceeded the send attempt limit
// from a previous SendPendingData cycle.
while (HasPendingDatagrams()) {
auto& front = PeekPendingDatagram();
if (front.send_attempts < max_attempts) break;
Debug(this,
"Datagram %" PRIu64 " abandoned after %u attempts",
front.id,
front.send_attempts);
DatagramStatus(front.id, DatagramStatus::ABANDONED);
PopPendingDatagram();
}
if (!HasPendingDatagrams()) return 0;
auto& dg = PeekPendingDatagram();
ngtcp2_vec dgvec = dg.data;
int accepted = 0;
int dg_flags = NGTCP2_WRITE_DATAGRAM_FLAG_MORE;
// PacketInfo for the datagram path. When libuv gains per-socket ECN
// marking, the value from ngtcp2 should be forwarded to the send path.
PacketInfo dg_pi;
ssize_t dg_nwrite = ngtcp2_conn_writev_datagram(*this,
&path->path,
dg_pi,
dest,
destlen,
&accepted,
dg_flags,
dg.id,
&dgvec,
1,
ts);
if (accepted) {
// Nice, the datagram was accepted!
Debug(this, "Datagram %" PRIu64 " accepted into packet", dg.id);
DatagramSent(dg.id);
PopPendingDatagram();
} else {
Debug(this, "Datagram %" PRIu64 " not accepted into packet", dg.id);
}
// ngtcp2 can return a positive number if the packet was nearly full, and so
// it finalized without waiting for more. This isn't an error, and can happen
// with or without `accepted` being true (if false, it stays queued).
if (dg_nwrite > 0) {
if (!accepted) dg.send_attempts++;
return dg_nwrite;
}
switch (dg_nwrite) {
case 0: {
// If dg_nwrite is 0, we are either congestion controlled or
// there wasn't enough room in the packet for the datagram or
// we aren't in a state where we can send.
// We'll skip this attempt and return 0.
CHECK(!accepted);
dg.send_attempts++;
return 0;
}
case NGTCP2_ERR_WRITE_MORE: {
// There's still room left in the packet!
return NGTCP2_ERR_WRITE_MORE;
}
case NGTCP2_ERR_INVALID_STATE:
case NGTCP2_ERR_INVALID_ARGUMENT: {
// Non-fatal error cases. Peer either does not support datagrams
// or the datagram is too large for the peer's max.
// Abandon the datagram and signal skip by returning std::nullopt.
DatagramStatus(dg.id, DatagramStatus::ABANDONED);
PopPendingDatagram();
return 0;
}
default: {
// Fatal errors: PKT_NUM_EXHAUSTED, CALLBACK_FAILURE, NOMEM, etc.
Debug(this, "Fatal datagram error: %zd", dg_nwrite);
SetLastError(QuicError::ForNgtcp2Error(dg_nwrite));
Close(CloseMethod::SILENT);
return dg_nwrite;
}
}
UNREACHABLE();
}
// the SendPendingData method is the primary driver for sending data from the
// application layer. It loops through available stream data and pending
// datagrams and generates packets to send until there is either no more
// data to send or we hit the maximum number of packets to send in one go.
// This method is extremely delicate. A bug in this method can break the
// entire QUIC implementation; so be very careful when making changes here
// and make sure to test thoroughly. When in doubt... don't change it.
void Session::SendPendingData() {
DCHECK(!is_destroyed());
if (!can_send_packets()) [[unlikely]] {
return;
}
// Upper bound on packets per SendPendingData call. ngtcp2's send quantum
// is typically 64 KB, which at 1200-byte minimum packet size is ~53
// packets. 64 covers the worst case with headroom. The actual count per
// call is dynamically capped by ngtcp2_conn_get_send_quantum().
static constexpr size_t kMaxPackets = 64;
Debug(this, "Session sending pending data");
// Cache the timestamp once for the entire send loop. ngtcp2 does not
// require nanosecond-accurate monotonicity within a single burst —
// a single timestamp per SendPendingData call is what other QUIC
// implementations use (e.g., quiche, msquic). When kernel-level
// packet pacing becomes available via libuv, this timestamp becomes
// the base for computing per-packet transmit timestamps.
const uint64_t ts = uv_hrtime();
PathStorage path;
StreamData stream_data;
bool closed = false;
// Batch accumulation: packets are collected here and flushed via
// Session::SendBatch when the loop exits, the batch is full, or
// on early return. This enables synchronous batched delivery via
// uv_udp_try_send2 (sendmmsg) from the deferred flush path.
Packet::Ptr batch[kMaxPackets];
PathStorage batch_paths[kMaxPackets];
size_t batch_count = 0;
auto flush_batch = [&] {
if (batch_count == 0) return;
SendBatch(batch, batch_paths, batch_count);
batch_count = 0;
};
auto update_stats = OnScopeLeave([&] {
if (closed) return;
// Flush any remaining accumulated packets before updating stats.
flush_batch();
if (is_destroyed()) [[unlikely]]
return;
// Get a strong pointer to protect against potential destruction during
// updating the time and data stats.
BaseObjectPtr<Session> s(this);
s->UpdatePacketTxTime();
s->UpdateTimer();
s->UpdateDataStats();
});
// The maximum number of packets to send in this call to SendPendingData.
const size_t max_packet_count = std::min(
kMaxPackets, ngtcp2_conn_get_send_quantum(*this) / max_packet_size());
if (max_packet_count == 0) return;
// The number of packets that have been prepared in this call.
size_t packet_send_count = 0;
Packet::Ptr packet;
auto ensure_packet = [&] {
if (!packet) {
packet = CreateStreamDataPacket();
if (!packet) [[unlikely]]
return false;
}
DCHECK(packet);
return true;
};
// Accumulate a completed packet into the batch.
auto enqueue_packet =
[&](Packet::Ptr& pkt, size_t len, const PacketInfo& pi) {
Debug(this, "Enqueuing packet with %zu bytes into batch", len);
pkt->Truncate(len);
pkt->set_pkt_info(pi);
path.CopyTo(&batch_paths[batch_count]);
batch[batch_count++] = std::move(pkt);
};
// We're going to enter a loop here to prepare and send no more than
// max_packet_count packets.
for (;;) {
// ndatalen is the amount of stream data that was accepted into the packet.
ssize_t ndatalen = 0;
// Make sure we have a packet to write data into.
if (!ensure_packet()) [[unlikely]] {
Debug(this, "Failed to create packet for stream data");
// Doh! Could not create a packet. Time to bail.
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return Close(CloseMethod::SILENT);
}
// The stream_data is the next block of data from the application stream.
// It is reused across iterations, so we reset before it's populated:
stream_data.count = 0;
stream_data.id = -1;
stream_data.fin = false;
stream_data.stream.reset();
if (application().GetStreamData(&stream_data) < 0) {
Debug(this, "Application failed to get stream data");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return Close(CloseMethod::SILENT);
}
// If we got here, we were at least successful in checking for stream data.
// There might not be any stream data to send. If there is no stream data,
// that's perfectly fine, we still need to serialize any frames we do have
// (pings, acks, datagrams, etc) so we'll just keep going.
if (stream_data.id >= 0) {
Debug(this, "Session using stream data: %s", stream_data);
} else {
Debug(this, "No stream data to send");
}
if (HasPendingDatagrams()) {
Debug(this, "There are pending datagrams to send");
}
// Awesome, let's write our packet!
PacketInfo pi;
ssize_t nwrite = WriteVStream(&path,
&pi,
packet->data(),
&ndatalen,
packet->length(),
stream_data,
ts);
// When ndatalen is > 0, that's our indication that stream data was
// accepted in to the packet. Yay!
if (ndatalen > 0) {
Debug(this,
"Session accepted %zu bytes from stream %" PRIi64 " into packet",
ndatalen,
stream_data.id);
if (!application().StreamCommit(&stream_data, ndatalen)) {
// Data was accepted into the packet, but for some reason adjusting
// the stream's committed data failed. Treat as fatal.
Debug(this, "Failed to commit accepted bytes in stream");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return Close(CloseMethod::SILENT);
}
// StreamCommit can call into JS, so a user callback could
// have destroyed the session:
if (is_destroyed()) [[unlikely]] {
closed = true;
return;
}
} else if (stream_data.id >= 0) {
Debug(this,
"Session did not accept any bytes from stream %" PRIi64
" into packet",
stream_data.id);
}
// When nwrite is zero, it means we are congestion limited or it is
// just not our turn to send something. Re-schedule the stream if it
// had unsent data (payload or FIN) so the next timer-triggered
// SendPendingData retries it. Without this, a FIN-only send that
// hits nwrite=0 is lost forever — the stream already returned EOS
// from Pull and won't be re-scheduled by anyone else.
// We call Application::ResumeStream directly (not Session::ResumeStream)
// to avoid creating a SendPendingDataScope — we're already inside
// SendPendingData and re-entering would just hit nwrite=0 again.
if (nwrite == 0 && (stream_data.id >= 0 || !HasPendingDatagrams())) {
Debug(this, "Congestion or not our turn to send");
if (stream_data.id >= 0 && (stream_data.count > 0 || stream_data.fin)) {
application().ResumeStream(stream_data.id);
}
return;
}
// A negative nwrite value indicates either an error or that there is more
// data to write into the packet.
if (nwrite < 0) {
switch (nwrite) {
case NGTCP2_ERR_STREAM_DATA_BLOCKED: {
// We could not write any data for this stream into the packet
// because the flow control for the stream itself indicates that
// the stream is blocked. We'll skip and move on to the next stream.
// ndatalen = -1 means that no stream data was accepted into the
// packet, which is what we want here.
DCHECK_EQ(ndatalen, -1);
// We should only have received this error if there was an actual
// stream identified in the stream data, but let's double check.
DCHECK_GE(stream_data.id, 0);
StreamDataBlocked(stream_data.id);
continue;
}
case NGTCP2_ERR_STREAM_SHUT_WR: {
// Indicates that the writable side of the stream should be closed
// locally or the stream is being reset. In either case, we can't
// send data for this stream!
Debug(this, "Closing stream %" PRIi64 " for writing", stream_data.id);
// ndatalen = -1 means that no stream data was accepted into the
// packet, which is what we want here.
DCHECK_EQ(ndatalen, -1);
// We should only have received this error if there was an actual
// stream identified in the stream data, but let's double check.
DCHECK_GE(stream_data.id, 0);
if (stream_data.stream) [[likely]] {
stream_data.stream->EndWritable();
}
// Notify the application that the stream's write side is shut
// so it stops queuing data. Without this, GetStreamData would
// keep returning the same stream and we'd loop forever.
application().StreamWriteShut(stream_data.id);
continue;
}
case NGTCP2_ERR_WRITE_MORE: {
Debug(this, "Packet buffer not full, coalesce more data into it");
// Room for more in this packet. Try to pack a pending datagram
// if there is one. Otherwise just loop around and keep going.
if (HasPendingDatagrams()) {
auto result = TryWritePendingDatagram(
&path, packet->data(), packet->length(), ts);
// When result is 0, either the datagram was congestion controlled,
// didn't fit in the packet, or was abandoned. Skip and continue.
// When result is > 0, the packet is done and the result is the
// completed size of the packet we're sending.
if (result > 0) {
size_t len = result;
Debug(this, "Sending packet with %zu bytes", len);
enqueue_packet(packet, len, pi);
if (++packet_send_count == max_packet_count) return;
} else if (result < 0) {
// Any negative result other than NGTCP2_ERR_WRITE_MORE
// at this point is fatal. The session will have been
// closed.
if (result != NGTCP2_ERR_WRITE_MORE) return;
}
}
continue;
}
case NGTCP2_ERR_CALLBACK_FAILURE: {
// This case really should not happen. It indicates that the
// ngtcp2 callback failed for some reason. This would be a
// bug in our code.
Debug(this, "Internal failure with ngtcp2 callback");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return Close(CloseMethod::SILENT);
}
}
// Some other type of error happened.
DCHECK_EQ(ndatalen, -1);
Debug(this,
"Session encountered error while writing packet: %s",
ngtcp2_strerror(nwrite));
SetLastError(QuicError::ForNgtcp2Error(nwrite));
closed = true;
return Close(CloseMethod::SILENT);
}
// At this point we have a packet prepared to send. The nwrite
// is the size of the packet we are sending.
size_t len = nwrite;
Debug(this, "Sending packet with %zu bytes", len);
enqueue_packet(packet, len, pi);
if (++packet_send_count == max_packet_count) return;
// If there are pending datagrams, try sending them in a fresh packet.
// This is necessary because ngtcp2_conn_writev_stream only returns
// NGTCP2_ERR_WRITE_MORE when there is actual stream data — when no
// streams are active, the coalescing path above is never reached and
// datagrams would never be sent.
if (HasPendingDatagrams()) {
if (!ensure_packet()) [[unlikely]] {
Debug(this, "Failed to create packet for datagram");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
closed = true;
return Close(CloseMethod::SILENT);
}
auto result =
TryWritePendingDatagram(&path, packet->data(), packet->length(), ts);
if (result > 0) {
Debug(this, "Sending datagram packet with %zd bytes", result);
enqueue_packet(packet, static_cast<size_t>(result), PacketInfo());
if (++packet_send_count == max_packet_count) return;
} else if (result < 0 && result != NGTCP2_ERR_WRITE_MORE) {
// Fatal error — session already closed by TryWritePendingDatagram.
return;
}
// If result == 0 (congestion) or NGTCP2_ERR_WRITE_MORE (datagram
// packed but room for more), the loop continues normally.
}
}
}
ssize_t Session::WriteVStream(PathStorage* path,
PacketInfo* pi,
uint8_t* dest,
ssize_t* ndatalen,
size_t max_packet_size,
const StreamData& stream_data,
uint64_t ts) {
DCHECK_LE(stream_data.count, kMaxVectorCount);
uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_MORE;
if (stream_data.fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN;
// The PacketInfo out-param is populated by ngtcp2 with the ECN codepoint
// to apply when sending this packet. When libuv gains per-socket ECN
// marking, the value should be forwarded to the send path.
return ngtcp2_conn_writev_stream(*this,
&path->path,
*pi,
dest,
max_packet_size,
ndatalen,
flags,
stream_data.id,
stream_data,
stream_data.count,
ts);
}
// ============================================================================
BaseObjectPtr<Session> Session::Create(
Endpoint* endpoint,
const Config& config,
TLSContext* tls_context,
const std::optional<SessionTicket>& ticket) {
JS_NEW_INSTANCE_OR_RETURN(endpoint->env(), obj, {});
return MakeDetachedBaseObject<Session>(
endpoint, obj, config, tls_context, ticket);
}
Session::Session(Endpoint* endpoint,
Local<Object> object,
const Config& config,
TLSContext* tls_context,
const std::optional<SessionTicket>& session_ticket)
: AsyncWrap(endpoint->env(), object, PROVIDER_QUIC_SESSION),
side_(config.side),
allocator_(BindingData::Get(env()).ngtcp2_allocator()),
impl_(std::make_unique<Impl>(this, endpoint, config)),
connection_(InitConnection()),
tls_session_(tls_context->NewSession(this, session_ticket)) {
DCHECK(impl_);
STAT_RECORD_TIMESTAMP(Stats, created_at);
// For clients, select the Application immediately - the ALPN is
// known upfront from the options. For servers, application_ stays
// null until the ClientHello names a protocol.
if (config.side == Side::CLIENT) {
InstallApplicationForAlpn(DecodeAlpn(config.options.tls_options.alpn));
}
// For client sessions with a session ticket and early data enabled,
// defer the handshake until the first stream or datagram is sent.
// This enables 0-RTT: the stream/datagram data is included in the
// first flight alongside the ClientHello. When early data is
// disabled, the handshake starts immediately (no 0-RTT attempt).
if (config.side == Side::CLIENT && session_ticket.has_value() &&
config.options.tls_options.enable_early_data) {
impl_->handshake_deferred_ = true;
}
if (config.options.keep_alive_timeout > 0) {
ngtcp2_conn_set_keep_alive_timeout(
*this, config.options.keep_alive_timeout * NGTCP2_MILLISECONDS);
}
MakeWeak();
Debug(this, "Session created.");
{
const HandleScope handle_scope(env()->isolate());
JS_DEFINE_READONLY_PROPERTY(
env(),
object,
env()->state_string(),
impl_->state_slot_.GetPageDataView(env()->isolate()));
JS_DEFINE_READONLY_PROPERTY(
env(),
object,
FIXED_ONE_BYTE_STRING(env()->isolate(), "stateByteOffset"),
Integer::NewFromUnsigned(
env()->isolate(),
static_cast<uint32_t>(impl_->state_slot_.GetByteOffset())));
JS_DEFINE_READONLY_PROPERTY(
env(),
object,
env()->stats_string(),
impl_->stats_slot_.GetPageBigUint64Array(env()->isolate()));
JS_DEFINE_READONLY_PROPERTY(
env(),
object,
FIXED_ONE_BYTE_STRING(env()->isolate(), "statsByteOffset"),
Integer::NewFromUnsigned(
env()->isolate(),
static_cast<uint32_t>(impl_->stats_slot_.GetByteOffset())));
}
UpdateDataStats();
}
Session::~Session() {
// Double check that Destroy() was called first.
CHECK(is_destroyed());
}
Session::QuicConnectionPointer Session::InitConnection() {
ngtcp2_conn* conn;
Path path(config().local_address, config().remote_address);
TransportParams::Config tp_config(side_, config().ocid, config().retry_scid);
TransportParams transport_params(tp_config,
config().options.transport_params);
transport_params.GenerateSessionTokens(this);
switch (side_) {
case Side::SERVER: {
// On the server side there are certain transport parameters that are
// required to be sent. Let's make sure they are set.
const ngtcp2_transport_params& params = transport_params;
CHECK_EQ(params.original_dcid_present, 1);
CHECK_EQ(ngtcp2_conn_server_new(&conn,
config().dcid,
config().scid,
&path,
config().version,
&Impl::SERVER,
&config().settings,
transport_params,
allocator_,
this),
0);
break;
}
case Side::CLIENT: {
CHECK_EQ(ngtcp2_conn_client_new(&conn,
config().dcid,
config().scid,
&path,
config().version,
&Impl::CLIENT,
&config().settings,
transport_params,
allocator_,
this),
0);
break;
}
}
return QuicConnectionPointer(conn);
}
Session::operator ngtcp2_conn*() const {
return connection_.get();
}
bool Session::is_server() const {
return side_ == Side::SERVER;
}
bool Session::is_destroyed() const {
return !impl_ || flags_.destroy_deferred;
}
bool Session::is_destroyed_or_closing() const {
return !impl_ || impl_->state()->closing;
}
void Session::Close(CloseMethod method) {
if (is_destroyed()) return;
// If the handshake was deferred (0-RTT client that never sent),
// no packets were ever transmitted. Close silently since there is
// nothing to communicate to the peer.
if (impl_->handshake_deferred_) {
impl_->handshake_deferred_ = false;
method = CloseMethod::SILENT;
}
if (impl_->last_error_) {
Debug(this, "Closing with error: %s", impl_->last_error_);
}
STAT_RECORD_TIMESTAMP(Stats, closing_at);
impl_->state()->closing = 1;
// With both the DEFAULT and SILENT options, we will proceed to closing
// the session immediately. All open streams will be immediately destroyed
// with whatever is set as the last error. The session will then be destroyed
// with a possible roundtrip to JavaScript to emit a close event and clean up
// any JavaScript side state. Importantly, these operations are all done
// synchronously, so the session will be destroyed once FinishClose returns.
//
// With the graceful option, we will wait for all streams to close on their
// own before proceeding with the FinishClose operation. New streams will
// be rejected, however.
switch (method) {
case CloseMethod::DEFAULT: {
Debug(this, "Immediately closing session");
impl_->state()->silent_close = 0;
return FinishClose();
}
case CloseMethod::SILENT: {
Debug(this, "Immediately closing session silently");
impl_->state()->silent_close = 1;
return FinishClose();
}
case CloseMethod::GRACEFUL: {
// If we are already closing gracefully, do nothing.
if (impl_->state()->graceful_close) [[unlikely]] {
return;
}
impl_->state()->graceful_close = 1;
// application_ may be null for server sessions if close() is called
// before the TLS handshake selects the ALPN. Without an application
// we cannot do a graceful shutdown (GOAWAY, CONNECTION_CLOSE etc.),
// so fall through to a silent close.
if (!impl_->application_) {
impl_->state()->silent_close = 1;
return FinishClose();
}
// The SendPendingDataScope ensures that the GOAWAY packet queued
// by BeginShutdown is actually sent. Without it, the GOAWAY sits
// in nghttp3's outq until the next Receive() triggers a send.
SendPendingDataScope send_scope(this);
// Signal application-level graceful shutdown (e.g., HTTP/3 GOAWAY).
// BeginShutdown can trigger callbacks that re-enter JS and destroy
// this session, so check is_destroyed() after it returns.
application().BeginShutdown();
if (is_destroyed()) return;
// If there are no open streams, then we can close immediately and
// not worry about waiting around.
if (impl_->streams_.empty()) {
impl_->state()->silent_close = 0;
return FinishClose();
}
// Shut down the writable side of streams whose readable side is
// already ended (e.g., peer called resetStream or sent FIN). Without
// this, such half-closed streams will never fire on_stream_close and
// the graceful close hangs. Streams still actively receiving data
// are left alone to complete naturally.
//
// When the application manages stream FIN (HTTP/3), skip this — a
// writable stream with a closed read side is the normal request/
// response pattern (server received full request, still sending
// response). The application protocol handles stream completion.
if (!application().stream_fin_managed_by_application()) {
Session::SendPendingDataScope send_scope(this);
for (auto& [id, stream] : impl_->streams_) {
if (stream->is_writable() && !stream->is_readable()) {
stream->EndWritable();
ngtcp2_conn_shutdown_stream_write(*this, 0, id, 0);
}
}
}
// The SendPendingDataScope destructor can trigger callbacks that
// re-enter JS and destroy this session.
if (is_destroyed()) return;
Debug(this,
"Gracefully closing session (waiting on %zu streams)",
impl_->streams_.size());
return;
}
}
UNREACHABLE();
}
void Session::FinishClose() {
// FinishClose() should be called only after, and as a result of, Close()
// being called first. However, re-entrancy through MakeCallback or timer
// callbacks can cause impl_ to be destroyed at any point during this
// method. We must check is_destroyed() after every operation that could
// trigger MakeCallback (stream destruction, pending queue rejection,
// SendConnectionClose, EmitClose).
if (is_destroyed()) return;
DCHECK(impl_->state()->closing);
// Clear the graceful_close flag to prevent RemoveStream() from
// re-entering FinishClose() when we destroy streams below.
impl_->state()->graceful_close = 0;
// Destroy all open streams immediately. We copy the map because
// streams remove themselves during destruction. Each Destroy() call
// triggers MakeCallback which can destroy impl_ via JS re-entrancy.
StreamsMap streams = impl_->streams_;
for (auto& stream : streams) {
if (is_destroyed()) return;
stream.second->Destroy(impl_->last_error_);
}
if (is_destroyed()) return;
// Clear pending stream queues.
while (!impl_->pending_bidi_stream_queue_.IsEmpty()) {
impl_->pending_bidi_stream_queue_.PopFront()->reject(impl_->last_error_);
}
while (!impl_->pending_uni_stream_queue_.IsEmpty()) {
impl_->pending_uni_stream_queue_.PopFront()->reject(impl_->last_error_);
}
// Send final application-level shutdown and CONNECTION_CLOSE
// unless this is a silent close.
if (!impl_->state()->silent_close) {
if (impl_->application_) {
application().CompleteShutdown();
}
SendConnectionClose();
}
if (is_destroyed()) return;
impl_->timer_.Close();
// If the session was passed to JavaScript, we need to round-trip
// through JS so it can clean up before we destroy. The JS side
// will synchronously call destroy(), which calls Session::Destroy().
if (impl_->state()->wrapped) {
EmitClose(impl_->last_error_);
} else {
Destroy();
}
}
void Session::Destroy() {
DCHECK(impl_);
// Ensure the closing flag is set for the ~Impl() DCHECK. Normally
// this is set by Session::Close(), but JS destroy() can be called
// directly without going through Close() first.
impl_->state()->closing = 1;
// If we're inside a ngtcp2 or nghttp3 callback scope, we cannot
// destroy impl_ now because the callback is executing methods on
// objects owned by impl_ (e.g., the Application). Defer the
// destruction until the scope exits.
if (flags_.in_ngtcp2_callback_scope || flags_.in_nghttp3_callback_scope) {
Debug(this, "Session destroy deferred (in callback scope)");
flags_.destroy_deferred = true;
return;
}
Debug(this, "Session destroyed");
STAT_RECORD_TIMESTAMP(Stats, destroyed_at);
impl_.reset();
}
PendingStream::PendingStreamQueue& Session::pending_bidi_stream_queue() const {
DCHECK(!is_destroyed());
return impl_->pending_bidi_stream_queue_;
}
PendingStream::PendingStreamQueue& Session::pending_uni_stream_queue() const {
DCHECK(!is_destroyed());
return impl_->pending_uni_stream_queue_;
}
size_t Session::max_packet_size() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_max_tx_udp_payload_size(*this);
}
uint32_t Session::version() const {
DCHECK(!is_destroyed());
return impl_->config_.version;
}
Endpoint& Session::endpoint() const {
DCHECK(!is_destroyed());
return *impl_->endpoint_;
}
TLSSession& Session::tls_session() const {
DCHECK(!is_destroyed());
return *tls_session_;
}
bool Session::has_application() const {
return !is_destroyed() && impl_->application_ != nullptr;
}
Session::Application& Session::application() const {
DCHECK(!is_destroyed());
DCHECK(impl_->application_);
return *impl_->application_;
}
std::string_view Session::DecodeAlpn(std::string_view wire) {
// ALPN wire format is length-prefixed: [len][name]. Extract the first entry.
if (wire.size() >= 2) {
uint8_t len = static_cast<uint8_t>(wire[0]);
if (len > 0 && static_cast<size_t>(len + 1) <= wire.size()) {
return wire.substr(1, len);
}
}
return {};
}
std::unique_ptr<Session::Application> Session::SelectApplicationFromAlpn(
std::string_view alpn) {
// h3 and h3-XX variants use Http3ApplicationImpl.
// Everything else uses DefaultApplication.
if (alpn == "h3" || (alpn.size() > 3 && alpn.substr(0, 3) == "h3-")) {
return CreateHttp3Application(this, config().options.application_options);
}
return CreateDefaultApplication(this, config().options.application_options);
}
void Session::InstallApplicationForAlpn(std::string_view alpn) {
// Acting on the ClientHello twice would install a second Application over
// a live one; TLSSession::EarlySelection is what prevents that.
CHECK(!has_application());
SetApplication(SelectApplicationFromAlpn(alpn));
}
void Session::SetEarlyRemoteTransportParams(std::span<const uint8_t> params) {
DCHECK(!is_destroyed());
if (params.empty()) return;
USE(ngtcp2_conn_decode_and_set_remote_transport_params(
*this, params.data(), params.size()));
}
void Session::SetApplication(std::unique_ptr<Application> app) {
DCHECK(!impl_->application_);
impl_->state()->application_type = static_cast<uint8_t>(app->type());
impl_->state()->headers_supported = static_cast<uint8_t>(
app->SupportsHeaders() ? HeadersSupportState::SUPPORTED
: HeadersSupportState::UNSUPPORTED);
impl_->state()->stream_callbacks_supported =
static_cast<uint8_t>(app->SupportsStreamCallbacks()
? StreamCallbacksSupportState::SUPPORTED
: StreamCallbacksSupportState::UNSUPPORTED);
// Surface the application's "no error" and "internal error" codes via
// session state so that JS-side code (e.g. the stream writer's fail()
// path) can resolve the right wire code for the negotiated ALPN
// without duplicating the per-application table.
impl_->state()->no_error_code = app->GetNoErrorCode();
impl_->state()->internal_error_code = app->GetInternalErrorCode();
impl_->state()->request_rejected_code = app->GetRequestRejectedCode();
impl_->application_ = std::move(app);
}
const SocketAddress& Session::remote_address() const {
DCHECK(!is_destroyed());
return impl_->remote_address_;
}
const SocketAddress& Session::local_address() const {
DCHECK(!is_destroyed());
return impl_->local_address_;
}
std::string Session::diagnostic_name() const {
const auto get_type = [&] { return is_server() ? "server" : "client"; };
return std::string("Session (") + get_type() + "," +
std::to_string(env()->thread_id()) + ":" +
std::to_string(static_cast<int64_t>(get_async_id())) + ")";
}
const Session::Config& Session::config() const {
DCHECK(!is_destroyed());
return impl_->config_;
}
Session::Config& Session::config() {
DCHECK(!is_destroyed());
return impl_->config_;
}
const Session::Options& Session::options() const {
DCHECK(!is_destroyed());
return impl_->config_.options;
}
void Session::EmitQlog(uint32_t flags, std::string_view data) {
if (!env()->can_call_into_js()) return;
if (!is_destroyed() && is_server() && !impl_->state()->wrapped) {
impl_->pending_qlog_.emplace_back(flags, std::string(data));
return;
}
bool fin = (flags & NGTCP2_QLOG_WRITE_FLAG_FIN) != 0;
// Fun fact... ngtcp2 does not emit the final qlog statement until the
// ngtcp2_conn object is destroyed. That means this method is called
// synchronously during impl_.reset() in Session::Destroy(), at which
// point is_destroyed() is true. We cannot use MakeCallback here because
// it can trigger microtask processing and re-entrancy while the
// ngtcp2_conn is mid-destruction. Defer the final chunk via SetImmediate.
if (is_destroyed()) {
auto isolate = env()->isolate();
Global<Object> recv(isolate, object());
Global<Function> cb(isolate,
BindingData::Get(env()).session_qlog_callback());
std::string buf(data);
env()->SetImmediate([recv = std::move(recv),
cb = std::move(cb),
buf = std::move(buf),
fin](Environment* env) {
HandleScope handle_scope(env->isolate());
auto context = env->context();
Local<Value> argv[] = {
Undefined(env->isolate()),
Boolean::New(env->isolate(), fin),
};
if (!ToV8Value(context, buf).ToLocal(&argv[0])) return;
USE(cb.Get(env->isolate())
->Call(context, recv.Get(env->isolate()), arraysize(argv), argv));
});
return;
}
auto isolate = env()->isolate();
Local<Value> argv[] = {Undefined(isolate), Boolean::New(isolate, fin)};
if (!ToV8Value(env()->context(), data).ToLocal(&argv[0])) {
Debug(this, "Failed to convert qlog data to V8 string");
return;
}
Debug(this, "Emitting qlog data");
MakeCallback(
BindingData::Get(env()).session_qlog_callback(), arraysize(argv), argv);
}
const TransportParams Session::local_transport_params() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_local_transport_params(*this);
}
const TransportParams Session::remote_transport_params() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_remote_transport_params(*this);
}
void Session::SetLastError(QuicError&& error) {
DCHECK(!is_destroyed());
Debug(this, "Setting last error to %s", error);
impl_->last_error_ = std::move(error);
}
bool Session::Receive(const uint8_t* data,
size_t len,
const SocketAddress& local_address,
const SocketAddress& remote_address,
const PacketInfo& pkt_info,
uint64_t ts) {
// Convenience wrapper: reads the packet and immediately triggers
// SendPendingData. Used by paths that need an immediate response
// (e.g., Endpoint::Connect for client Initial packets).
// The hot receive path uses ReadPacket() directly with deferred
// flush via BindingData's uv_check callback.
SendPendingDataScope send_scope(this);
return ReadPacket(data, len, local_address, remote_address, pkt_info, ts);
}
bool Session::ReadPacket(const uint8_t* data,
size_t len,
const SocketAddress& local_address,
const SocketAddress& remote_address,
const PacketInfo& pkt_info,
uint64_t ts) {
DCHECK(!is_destroyed());
impl_->remote_address_ = remote_address;
Path path(local_address, remote_address);
Debug(this,
"Session is receiving %zu-byte packet received along path %s",
len,
path);
// It is important to understand that reading the packet will cause
// callback functions to be invoked, any one of which could lead to
// the Session being closed/destroyed synchronously. The callback scope
// ensures that any deferred destroy waits until all callbacks for this
// packet have completed. After calling ngtcp2_conn_read_pkt here, we
// will need to double check that the session is not destroyed before
// we try doing anything with it (like updating stats, etc).
int err;
{
NgTcp2CallbackScope callback_scope(this);
// The PacketInfo carries per-packet metadata (currently ECN codepoint).
// When libuv gains per-packet ECN reporting, the caller should
// populate pkt_info from the receive metadata before calling
// ReadPacket().
// When ts is 0 (the default), call uv_hrtime() here. The batched
// receive path caches a timestamp and passes it to all ReadPacket()
// calls in the same I/O burst.
if (ts == 0) ts = uv_hrtime();
err = ngtcp2_conn_read_pkt(*this, &path, pkt_info, data, len, ts);
}
if (is_destroyed()) return false;
Debug(this, "Session receiving %zu-byte packet with result %d", len, err);
if (err == 0 && !is_destroyed()) [[likely]] {
STAT_INCREMENT_N(Stats, bytes_received, len);
}
return AfterNgtcp2Read(err);
}
void Session::ResumeHandshake() {
DCHECK(!is_destroyed());
DCHECK(is_server());
// From here on the ClientHello callback is a no-op, so the handshake
// runs on into ticket decryption, early data and the rest.
tls_session().set_early_selection(TLSSession::EarlySelection::kComplete);
Debug(this, "Resuming the TLS handshake");
int err;
{
NgTcp2CallbackScope callback_scope(this);
err = ngtcp2_conn_continue_handshake(*this, uv_hrtime());
}
if (is_destroyed()) return;
AfterNgtcp2Read(err);
}
bool Session::AfterNgtcp2Read(int err) {
switch (err) {
case 0: {
if (!is_destroyed()) [[likely]] {
// Process deferred application operations after ALPN selection - not
// necessarily resolved yet as ClientHello can span multiple packets.
if (has_application()) application().PostReceive();
if (is_destroyed()) return true;
// The ClientHello has been processed: SNI and ALPN are selected and
// the Application is installed, but the handshake is stopped short
// of ticket decryption, so no early data exists yet. Surface the
// session, then let the handshake run on. The guard makes this fire
// exactly once, on whichever packet completed the ClientHello, so a
// ClientHello split across datagrams is handled correctly.
if (is_server() && tls_session().early_selection() ==
TLSSession::EarlySelection::kSelected) {
endpoint().EmitNewSession(BaseObjectPtr<Session>(this));
if (!is_destroyed()) ResumeHandshake();
}
}
return true;
}
case NGTCP2_ERR_REQUIRED_TRANSPORT_PARAM: {
Debug(this,
"Receiving packet failed: "
"Remote peer failed to send a required transport parameter");
if (!is_destroyed()) [[likely]] {
SetLastError(QuicError::ForTransport(err));
Close();
}
return false;
}
case NGTCP2_ERR_DRAINING: {
// Connection has entered the draining state, no further data
// should be sent. This can happen for two reasons:
//
// 1. The remote peer sent a CONNECTION_CLOSE. In this case we
// start the draining timer and let OnTimeout handle the
// close, extracting the peer's error via FromConnectionClose.
//
// 2. The remote peer sent a stateless reset. ngtcp2 set the
// draining state internally and invoked our informational
// on_receive_stateless_reset callback (which set the flag).
// There is no point in waiting for a draining period — the
// peer has no state. Close immediately with an error.
if (!is_destroyed()) [[likely]] {
if (impl_->state()->stateless_reset) {
Debug(this, "Session received stateless reset, closing");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_DRAINING));
Close(CloseMethod::SILENT);
} else {
Debug(this, "Session is draining, starting draining timer");
UpdateTimer();
}
}
return false;
}
case NGTCP2_ERR_CLOSING: {
// Connection has entered the closing state, no further data should be
// sent. This happens when the local peer has called
// ngtcp2_conn_write_connection_close.
Debug(this, "Receiving packet failed: Session is closing");
return false;
}
case NGTCP2_ERR_CRYPTO: {
Debug(this, "Receiving packet failed: Crypto error");
// Crypto error happened! Set the last error to the tls alert
if (!is_destroyed()) [[likely]] {
SetLastError(QuicError::ForTlsAlert(ngtcp2_conn_get_tls_alert(*this)));
Close();
}
return false;
}
case NGTCP2_ERR_RETRY: {
// This should only ever happen on the server. We have to send a path
// validation challenge in the form of a RETRY packet to the peer and
// drop the connection.
DCHECK(is_server());
Debug(this, "Receiving packet failed: Server must send a retry packet");
if (!is_destroyed()) {
endpoint().SendRetry(
PathDescriptor{
version(),
config().dcid,
config().scid,
impl_->local_address_,
impl_->remote_address_,
},
uv_hrtime());
Close(CloseMethod::SILENT);
}
return false;
}
case NGTCP2_ERR_DROP_CONN: {
// There's nothing else to do but drop the connection state.
Debug(this, "Receiving packet failed: Session must drop the connection");
if (!is_destroyed()) {
Close(CloseMethod::SILENT);
}
return false;
}
}
// Shouldn't happen but just in case... handle other unknown errors
Debug(this,
"Receiving packet failed: "
"Unexpected error %d while receiving packet",
err);
if (!is_destroyed()) {
SetLastError(QuicError::ForNgtcp2Error(err));
Close();
}
return false;
}
void Session::SendBatch(Packet::Ptr* packets,
PathStorage* paths,
size_t count) {
DCHECK(!is_destroyed());
if (count == 0) return;
// Separate packets into those going to the primary endpoint and those
// redirected to other endpoints (rare: path validation, preferred address).
// Redirected packets are sent individually via the target endpoint.
static constexpr size_t kMaxBatch = 64;
DCHECK_LE(count, kMaxBatch);
Packet::Ptr primary_packets[kMaxBatch];
size_t primary_count = 0;
for (size_t i = 0; i < count; i++) {
if (!packets[i] || !can_send_packets()) {
packets[i].reset();
continue;
}
UpdatePath(paths[i]);
// Check for cross-endpoint redirect.
bool redirected = false;
if (paths[i].path.local.addrlen > 0) {
SocketAddress local_addr(paths[i].path.local.addr);
auto& mgr = BindingData::Get(env()).session_manager();
Endpoint* target = mgr.FindEndpointForAddress(local_addr);
if (target != nullptr && target != &endpoint()) {
SocketAddress remote_addr(paths[i].path.remote.addr);
packets[i]->Redirect(static_cast<Packet::Listener*>(target),
remote_addr);
target->Send(std::move(packets[i]));
redirected = true;
}
}
if (!redirected) {
primary_packets[primary_count++] = std::move(packets[i]);
}
}
if (primary_count == 0) return;
// Use batched send for the primary endpoint.
if (flags_.prefer_try_send) {
endpoint().SendBatch(primary_packets, primary_count);
} else {
// Non-flush path: send individually via async uv_udp_send.
for (size_t i = 0; i < primary_count; i++) {
Send(std::move(primary_packets[i]));
}
}
}
void Session::FlushPendingData() {
DCHECK(!is_destroyed());
if (impl_->application_) {
// Prefer synchronous sends during the deferred flush to avoid the
// one-tick latency of async uv_udp_send from the uv_check callback.
flags_.prefer_try_send = true;
SendPendingData();
flags_.prefer_try_send = false;
}
}
void Session::Send(Packet::Ptr packet) {
// Sending a Packet is generally best effort. If we're not in a state
// where we can send a packet, it's ok to drop it on the floor. The
// packet loss mechanisms will cause the packet data to be resent later
// if appropriate (and possible).
// That said, we should never be trying to send a packet when we're in
// a draining period.
DCHECK(!is_destroyed());
DCHECK(!is_in_draining_period());
if (!can_send_packets()) [[unlikely]] {
// Ptr destructor releases the packet back to the pool.
return;
}
// When called from the deferred flush path (uv_check callback),
// prefer synchronous send to avoid the one-tick latency of async
// uv_udp_send. SendOrTrySend uses uv_udp_try_send first, falling
// back to uv_udp_send on EAGAIN.
if (flags_.prefer_try_send) {
Debug(this, "Session is sending (try_send) %s", packet->ToString());
endpoint().SendOrTrySend(std::move(packet));
return;
}
Debug(this, "Session is sending %s", packet->ToString());
endpoint().Send(std::move(packet));
}
void Session::Send(Packet::Ptr packet, const PathStorage& path) {
DCHECK(!is_destroyed());
DCHECK(!is_in_draining_period());
UpdatePath(path);
// Check if ngtcp2 wants this packet sent on a different path than the
// primary endpoint. This happens during path validation for preferred
// address or connection migration — e.g., a PATH_RESPONSE needs to be
// sent from the preferred address endpoint, not the primary.
if (path.path.local.addrlen > 0) {
SocketAddress local_addr(path.path.local.addr);
auto& mgr = BindingData::Get(env()).session_manager();
Endpoint* target = mgr.FindEndpointForAddress(local_addr);
if (target != nullptr && target != &endpoint()) {
// Redirect the packet to the target endpoint. This updates the
// listener (for pending_callbacks accounting in the ArenaPool
// completion callback) and the destination address.
SocketAddress remote_addr(path.path.remote.addr);
packet->Redirect(static_cast<Packet::Listener*>(target), remote_addr);
if (can_send_packets()) [[likely]] {
Debug(this,
"Sending via non-primary endpoint for path %s",
local_addr.ToString());
target->Send(std::move(packet));
}
return;
}
}
Send(std::move(packet));
}
datagram_id Session::SendDatagram(Store&& data) {
DCHECK(!is_destroyed());
// Sending a datagram is best effort. If we cannot send it for any reason,
// we just return 0 to indicate that the datagram was not sent an the
// data is dropped on the floor.
// If the session is destroyed, draining, or closing, we cannot send.
if (is_destroyed() || is_in_draining_period() || is_in_closing_period()) {
return 0;
}
const ngtcp2_transport_params* tp = remote_transport_params();
uint64_t max_datagram_size = MaxDatagramPayload(tp->max_datagram_frame_size);
// These size and length checks should have been caught by the JavaScript
// side, but handle it gracefully here just in case. We might have some future
// case where datagram frames are sent from C++ code directly, so it's good to
// have these checks as a backstop regardless.
if (max_datagram_size == 0) {
Debug(this, "Datagrams are disabled");
return 0;
}
if (data.length() > max_datagram_size) [[unlikely]] {
Debug(this, "Ignoring oversized datagram");
return 0;
}
if (data.length() == 0) [[unlikely]] {
Debug(this, "Ignoring empty datagram");
return 0;
}
// Assign the datagram ID.
datagram_id did = ++impl_->state()->last_datagram_id;
// Check queue capacity. Apply the drop policy when full.
auto max_pending = impl_->state()->max_pending_datagrams;
if (max_pending > 0 && impl_->pending_datagrams_.size() >= max_pending) {
auto drop_policy = impl_->config_.options.datagram_drop_policy;
if (drop_policy == DatagramDropPolicy::DROP_OLDEST) {
auto& oldest = impl_->pending_datagrams_.front();
Debug(this,
"Datagram queue full, dropping oldest datagram %" PRIu64,
oldest.id);
DatagramStatus(oldest.id, DatagramStatus::ABANDONED);
impl_->pending_datagrams_.pop_front();
} else {
// DROP_NEWEST: reject the incoming datagram.
Debug(
this, "Datagram queue full, dropping newest datagram %" PRIu64, did);
DatagramStatus(did, DatagramStatus::ABANDONED);
return did;
}
}
Session::SendPendingDataScope send_scope(this);
// Queue the datagram. It will be serialized into packets by
// SendPendingData alongside stream data.
Debug(this, "Queuing %zu-byte datagram %" PRIu64, data.length(), did);
impl_->pending_datagrams_.push_back({did, std::move(data)});
return did;
}
void Session::UpdatePacketTxTime() {
DCHECK(!is_destroyed());
ngtcp2_conn_update_pkt_tx_time(*this, uv_hrtime());
}
void Session::UpdatePath(const PathStorage& storage) {
DCHECK(!is_destroyed());
impl_->remote_address_.Update(storage.path.remote.addr,
storage.path.remote.addrlen);
impl_->local_address_.Update(storage.path.local.addr,
storage.path.local.addrlen);
Debug(this,
"path updated. local %s, remote %s",
impl_->local_address_,
impl_->remote_address_);
}
BaseObjectPtr<Stream> Session::FindStream(stream_id id) const {
if (is_destroyed()) return {};
auto it = impl_->streams_.find(id);
if (it == std::end(impl_->streams_)) return {};
return it->second;
}
Session::StreamsMap Session::streams() const {
if (is_destroyed()) return {};
return impl_->streams_;
}
BaseObjectPtr<Stream> Session::CreateStream(
stream_id id,
CreateStreamOption option,
std::shared_ptr<DataQueue> data_source) {
if (!can_create_streams()) [[unlikely]]
return {};
if (auto stream = Stream::Create(this, id, std::move(data_source)))
[[likely]] {
AddStream(stream, option);
return stream;
}
return {};
}
MaybeLocal<Object> Session::OpenStream(Direction direction,
std::shared_ptr<DataQueue> data_source) {
// If can_create_streams() returns false, we are not able to open a stream
// at all now, even in a pending state. The implication is that that session
// is destroyed or closing.
if (!can_create_streams()) [[unlikely]]
return {};
// If can_open_streams() returns false, we are able to create streams but
// they will remain in a pending state. The implication is that the session
// TLS handshake is still progressing. Note that when a pending stream is
// created, it will not be listed in the streams list.
if (!can_open_streams()) {
if (auto stream = Stream::Create(this, direction, std::move(data_source)))
[[likely]] {
return stream->object();
}
return {};
}
stream_id id = -1;
auto open = [&] {
switch (direction) {
case Direction::BIDIRECTIONAL: {
Debug(this, "Opening bidirectional stream");
return ngtcp2_conn_open_bidi_stream(*this, &id, nullptr);
}
case Direction::UNIDIRECTIONAL: {
Debug(this, "Opening uni-directional stream");
return ngtcp2_conn_open_uni_stream(*this, &id, nullptr);
}
}
UNREACHABLE();
};
switch (open()) {
case 0: {
// Woo! Our stream was created.
CHECK_GE(id, 0);
if (auto stream = CreateStream(
id, CreateStreamOption::DO_NOT_NOTIFY, std::move(data_source)))
[[likely]] {
return stream->object();
}
break;
}
case NGTCP2_ERR_STREAM_ID_BLOCKED: {
// The stream cannot yet be opened.
// This is typically caused by the application exceeding the allowed max
// number of concurrent streams. We will allow the stream to be created
// in a pending state.
if (auto stream = Stream::Create(this, direction, std::move(data_source)))
[[likely]] {
return stream->object();
}
break;
}
}
return {};
}
void Session::AddStream(BaseObjectPtr<Stream> stream,
CreateStreamOption option) {
DCHECK(!is_destroyed());
DCHECK(stream);
auto id = stream->id();
auto direction = stream->direction();
// Let's double check that a stream with the given id does not alreadys
// exist. If it does, that means we've got a bug somewhere.
DCHECK_EQ(impl_->streams_.find(id), impl_->streams_.end());
Debug(this, "Adding stream %" PRIi64 " to session", id);
// The streams_ map becomes the sole owner of the Stream instance.
// We mark the stream detached so that when it is removed from
// the session, or is the session is destroyed, the stream will
// also be destroyed.
impl_->streams_[id] = stream;
stream->Detach();
ngtcp2_conn_set_stream_user_data(*this, id, stream.get());
// If the stream already has outbound data (body was provided at creation
// time), resume it now that it is registered in the streams map and can
// be found by FindStream.
if (stream->has_outbound()) {
ResumeStream(id);
}
if (option == CreateStreamOption::NOTIFY) {
EmitStream(stream);
// EmitStream triggers the JS onstream callback via MakeCallback.
// If the callback throws, safeCallbackInvoke calls session.destroy()
// which resets impl_. We must bail out if that happened.
if (is_destroyed()) return;
}
// Update tracking statistics for the number of streams associated with this
// session.
if (ngtcp2_conn_is_local_stream(*this, id)) {
switch (direction) {
case Direction::BIDIRECTIONAL: {
STAT_INCREMENT(Stats, bidi_out_stream_count);
break;
}
case Direction::UNIDIRECTIONAL: {
STAT_INCREMENT(Stats, uni_out_stream_count);
break;
}
}
} else {
switch (direction) {
case Direction::BIDIRECTIONAL: {
STAT_INCREMENT(Stats, bidi_in_stream_count);
break;
}
case Direction::UNIDIRECTIONAL: {
STAT_INCREMENT(Stats, uni_in_stream_count);
break;
}
}
}
}
void Session::RemoveStream(stream_id id) {
DCHECK(!is_destroyed());
Debug(this, "Removing stream %" PRIi64 " from session", id);
ngtcp2_conn_set_stream_user_data(*this, id, nullptr);
if (has_application()) application().StreamRemoved(id);
// Note that removing the stream from the streams map likely releases
// the last BaseObjectPtr holding onto the Stream instance, at which
// point it will be freed. If there are other BaseObjectPtr instances
// or other references to the Stream, however, freeing will be deferred.
// In either case, we cannot assume that the stream still exists after
// this call.
impl_->streams_.erase(id);
// If we are gracefully closing and there are no more streams,
// then we can proceed to finishing the close now. Note that the
// expectation is that the session will be destroyed once FinishClose
// returns.
if (impl_->state()->closing && impl_->state()->graceful_close &&
impl_->streams_.size() == 0) {
FinishClose();
CHECK(is_destroyed());
}
}
void Session::ResumeStream(stream_id id) {
DCHECK(!is_destroyed());
SendPendingDataScope send_scope(this);
application().ResumeStream(id);
}
void Session::ShutdownStream(stream_id id, QuicError error) {
DCHECK(!is_destroyed());
Debug(this, "Shutting down stream %" PRIi64 " with error %s", id, error);
SendPendingDataScope send_scope(this);
// STOP_SENDING and RESET_STREAM frames carry application-level error
// codes (RFC 9000 §19.4, §19.5). Map the QuicError to an appropriate
// application code: APPLICATION errors pass through directly; transport
// no-error maps to the application's no-error code; any other error
// maps to the application's internal error code.
error_code code;
if (error.type() == QuicError::Type::APPLICATION) {
code = error.code();
} else if (error.code() == NGTCP2_NO_ERROR) {
code = application().GetNoErrorCode();
} else {
code = application().GetInternalErrorCode();
}
ngtcp2_conn_shutdown_stream(*this, 0, id, code);
}
void Session::ShutdownStreamWrite(stream_id id, QuicError error) {
DCHECK(!is_destroyed());
Debug(
this, "Shutting down stream %" PRIi64 " write with error %s", id, error);
SendPendingDataScope send_scope(this);
error_code code;
if (error.type() == QuicError::Type::APPLICATION) {
code = error.code();
} else if (error.code() == NGTCP2_NO_ERROR) {
code = application().GetNoErrorCode();
} else {
code = application().GetInternalErrorCode();
}
ngtcp2_conn_shutdown_stream_write(*this, 0, id, code);
}
void Session::StreamDataBlocked(stream_id id) {
DCHECK(!is_destroyed());
STAT_INCREMENT(Stats, block_count);
application().BlockStream(id);
}
void Session::CollectSessionTicketAppData(
SessionTicket::AppData* app_data) const {
DCHECK(!is_destroyed());
application().CollectSessionTicketAppData(app_data);
}
SessionTicket::AppData::Status Session::ExtractSessionTicketAppData(
const SessionTicket::AppData& app_data, Flag flag) {
DCHECK(!is_destroyed());
// Renew, so the client stops offering a ticket that is never accepted.
if (!has_application()) [[unlikely]] {
return SessionTicket::AppData::Status::TICKET_IGNORE_RENEW;
}
return application().ExtractSessionTicketAppData(app_data, flag);
}
void Session::MemoryInfo(MemoryTracker* tracker) const {
if (impl_) {
tracker->TrackField("impl", impl_);
}
tracker->TrackField("tls_session", tls_session_);
}
bool Session::is_in_closing_period() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_in_closing_period(*this) != 0;
}
bool Session::is_in_draining_period() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_in_draining_period(*this) != 0;
}
bool Session::wants_session_ticket() const {
return !is_destroyed() &&
HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::SESSION_TICKET);
}
void Session::SetStreamOpenAllowed() {
DCHECK(!is_destroyed());
impl_->state()->stream_open_allowed = 1;
}
void Session::PopulateEarlyTransportParamsState() {
DCHECK(!is_destroyed());
const ngtcp2_transport_params* tp = remote_transport_params();
if (tp != nullptr) {
impl_->state()->max_datagram_size =
MaxDatagramPayload(tp->max_datagram_frame_size);
}
}
bool Session::can_send_packets() const {
// We can send packets if we're not in the middle of a ngtcp2 callback
// on THIS session, we're not destroyed, and we're not in a draining
// or closing period. The callback scope check is per-session so that
// one session's ngtcp2 callback does not block unrelated sessions
// from sending.
return !is_destroyed() && !flags_.in_ngtcp2_callback_scope &&
!is_in_draining_period() && !is_in_closing_period();
}
bool Session::can_create_streams() const {
return !is_destroyed_or_closing() && !is_in_closing_period() &&
!is_in_draining_period();
}
bool Session::can_open_streams() const {
return !is_destroyed() && impl_->state()->stream_open_allowed;
}
uint64_t Session::max_data_left() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_max_data_left(*this);
}
bool Session::OpenUnidirectionalStream(stream_id* id) {
return ngtcp2_conn_open_uni_stream(*this, id, nullptr) == 0;
}
void Session::ExtendMaxStreams(stream_id id) {
// MAX_STREAMS only limits what the peer opens, and there is nothing to
// grant once the connection is going away.
if (is_in_draining_period() || is_in_closing_period()) return;
if (ngtcp2_conn_is_local_stream(*this, id)) return;
if (ngtcp2_is_bidi_stream(id)) {
ngtcp2_conn_extend_max_streams_bidi(*this, 1);
} else {
ngtcp2_conn_extend_max_streams_uni(*this, 1);
}
}
void Session::Consume(stream_id id, size_t len) {
ExtendStreamOffset(id, len);
ExtendOffset(len);
}
void Session::SetApplicationError(error_code app_error_code) {
SetLastError(QuicError::ForApplication(app_error_code));
}
uint64_t Session::max_local_streams_uni() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_streams_uni_left(*this);
}
uint64_t Session::max_local_streams_bidi() const {
DCHECK(!is_destroyed());
return ngtcp2_conn_get_local_transport_params(*this)
->initial_max_streams_bidi;
}
void Session::set_wrapped() {
DCHECK(!is_destroyed());
impl_->state()->wrapped = 1;
}
bool Session::tls_info_ready() const {
// hello_processed_ is set server-side, handshake_completed covers
// the client. Together they mark the point when SNI/ALPN are final.
return hello_processed_ || impl_->state()->handshake_completed;
}
void Session::FlushPendingQlog() {
if (is_destroyed()) return;
DCHECK(impl_->state()->wrapped);
// Runs synchronously immediately after the new-session callback returns.
auto pending = std::move(impl_->pending_qlog_);
for (auto& [flags, data] : pending) {
if (is_destroyed()) return;
EmitQlog(flags, data);
}
}
void Session::set_priority_supported(bool on) {
DCHECK(!is_destroyed());
impl_->state()->priority_supported = on ? 1 : 0;
}
void Session::ExtendStreamOffset(stream_id id, size_t amount) {
DCHECK(!is_destroyed());
Debug(this, "Extending stream %" PRIi64 " offset by %zu bytes", id, amount);
ngtcp2_conn_extend_max_stream_offset(*this, id, amount);
}
void Session::ExtendOffset(size_t amount) {
DCHECK(!is_destroyed());
Debug(this, "Extending offset by %zu bytes", amount);
ngtcp2_conn_extend_max_offset(*this, amount);
}
bool Session::HasPendingDatagrams() const {
return impl_ && !impl_->pending_datagrams_.empty();
}
Session::PendingDatagram& Session::PeekPendingDatagram() {
return impl_->pending_datagrams_.front();
}
void Session::PopPendingDatagram() {
impl_->pending_datagrams_.pop_front();
}
size_t Session::PendingDatagramCount() const {
return impl_ ? impl_->pending_datagrams_.size() : 0;
}
void Session::DatagramSent(datagram_id id) {
Debug(this, "Datagram %" PRIu64 " sent", id);
STAT_INCREMENT(Stats, datagrams_sent);
}
void Session::UpdateDataStats() {
if (is_destroyed()) return;
Debug(this, "Updating data stats");
ngtcp2_conn_info info;
ngtcp2_conn_get_conn_info(*this, &info);
STAT_SET(Stats, bytes_in_flight, info.bytes_in_flight);
STAT_SET(Stats, cwnd, info.cwnd);
STAT_SET(Stats, latest_rtt, info.latest_rtt);
STAT_SET(Stats, min_rtt, info.min_rtt);
STAT_SET(Stats, rttvar, info.rttvar);
STAT_SET(Stats, smoothed_rtt, info.smoothed_rtt);
STAT_SET(Stats, ssthresh, info.ssthresh);
STAT_SET(Stats, pkt_sent, info.pkt_sent);
STAT_SET(Stats, bytes_sent, info.bytes_sent);
STAT_SET(Stats, pkt_recv, info.pkt_recv);
STAT_SET(Stats, bytes_recv, info.bytes_recv);
STAT_SET(Stats, pkt_lost, info.pkt_lost);
STAT_SET(Stats, bytes_lost, info.bytes_lost);
STAT_SET(Stats, ping_recv, info.ping_recv);
STAT_SET(Stats, pkt_discarded, info.pkt_discarded);
STAT_SET(
Stats,
max_bytes_in_flight,
std::max(STAT_GET(Stats, max_bytes_in_flight), info.bytes_in_flight));
}
void Session::CheckStreamIdleTimeout(uint64_t now) {
if (is_destroyed()) return;
uint64_t timeout = options().stream_idle_timeout;
if (timeout == 0) return;
uint64_t timeout_ns = timeout * NGTCP2_MILLISECONDS;
auto all_streams = streams();
for (const auto& [id, stream] : all_streams) {
if (!stream) continue;
// Only check peer-initiated streams. Locally-initiated streams
// that haven't been written to are the application's concern.
if (ngtcp2_conn_is_local_stream(*this, id)) continue;
uint64_t last_activity = stream->last_activity_timestamp();
if (last_activity > 0 && (now - last_activity) > timeout_ns) {
Debug(this, "Stream %" PRId64 " idle timeout exceeded, destroying", id);
// Notify the peer before destroying. ShutdownStream sends both
// STOP_SENDING and RESET_STREAM as appropriate, using the
// application's no-error code for non-APPLICATION errors (since
// these frames carry application-level error codes per RFC 9000).
// Without this, the peer's stream sits orphaned until the
// session closes.
auto error =
QuicError::ForNgtcp2Error(NGTCP2_ERR_PROTO, "stream idle timeout");
ShutdownStream(id, error);
stream->Destroy(error);
STAT_INCREMENT(Stats, streams_idle_timed_out);
}
}
}
void Session::SendConnectionClose() {
// Method is a non-op if the session is already destroyed or the
// endpoint cannot send. Note: we intentionally do NOT check
// can_send_packets() here because ngtcp2_conn_write_connection_close
// puts the connection into the closing period, and the resulting packet
// must still be sent to the endpoint.
if (is_destroyed()) return;
Debug(this, "Sending connection close packet to peer");
auto ErrorAndSilentClose = [&] {
Debug(this, "Failed to create connection close packet");
SetLastError(QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL));
Close(CloseMethod::SILENT);
};
if (is_server()) {
if (auto packet = Packet::CreateConnectionClosePacket(
endpoint(), impl_->remote_address_, *this, impl_->last_error_))
[[likely]] {
// Send directly to endpoint, bypassing Session::Send which
// would drop the packet because we're now in the closing period.
return endpoint().Send(std::move(packet));
}
// If we are unable to create a connection close packet then
// we are in a bad state. An internal error will be set and
// the session will be silently closed. This is not ideal
// because the remote peer will not know immediately that
// the connection has terminated but there's not much else
// we can do.
return ErrorAndSilentClose();
}
auto packet = endpoint().CreatePacket(impl_->remote_address_,
kDefaultMaxPacketLength,
"immediate connection close (client)");
if (!packet) [[unlikely]] {
return ErrorAndSilentClose();
}
ngtcp2_vec vec = *packet;
Path path(impl_->local_address_, impl_->remote_address_);
ngtcp2_ccerr wire_error = impl_->last_error_.ToNgtcp2ConnectionCloseError();
ssize_t nwrite = ngtcp2_conn_write_connection_close(
*this, &path, nullptr, vec.base, vec.len, &wire_error, uv_hrtime());
if (nwrite < 0) [[unlikely]] {
return ErrorAndSilentClose();
}
packet->Truncate(nwrite);
// Send directly to endpoint — ngtcp2 has entered the closing period
// at this point, so Session::Send() would drop the packet.
return endpoint().Send(std::move(packet));
}
void Session::OnTimeout() {
if (is_destroyed()) return;
if (!impl_->application_) return;
// Hold a strong reference to prevent the Session from being freed during
// re-entrant calls. SendPendingData's scope guard calls UpdateTimer(),
// which can synchronously re-enter OnTimeout() when the timer has already
// expired. That re-entrant path can reach FinishClose → EmitClose →
// Destroy → impl_.reset() → ~Impl → RemoveSession(), dropping the last
// BaseObjectPtr from the endpoint map and freeing the Session. Without
// this guard, the outer OnTimeout / SendPendingData frames would operate
// on a freed object.
BaseObjectPtr<Session> ref(this);
HandleScope scope(env()->isolate());
int ret;
{
NgTcp2CallbackScope callback_scope(this);
ret = ngtcp2_conn_handle_expiry(*this, uv_hrtime());
}
// handle_expiry can trigger ngtcp2 callbacks that invoke MakeCallback,
// which can synchronously destroy the session. Guard before proceeding.
if (is_destroyed()) return;
if (NGTCP2_OK(ret) && !is_in_closing_period() && !is_in_draining_period()) {
SendPendingData();
if (is_destroyed()) return;
CheckStreamIdleTimeout(uv_hrtime());
return;
}
if (is_destroyed()) return;
Debug(this, "Session timed out");
// When the draining period expires, the peer has already sent
// CONNECTION_CLOSE. Use their close error so a clean close (code 0)
// propagates as no-error, allowing stream.closed promises to resolve.
if (is_in_draining_period()) {
SetLastError(QuicError::FromConnectionClose(*this));
} else {
SetLastError(QuicError::ForNgtcp2Error(ret));
}
Close(CloseMethod::SILENT);
}
void Session::UpdateTimer() {
DCHECK(!is_destroyed());
// Both uv_hrtime and ngtcp2_conn_get_expiry return nanosecond units.
uint64_t now = uv_hrtime();
uint64_t expiry;
if (is_in_draining_period()) {
// RFC 9000 Section 10.2: The draining state SHOULD persist for at
// least three times the current Probe Timeout (PTO). ngtcp2 does
// not set a draining timer internally — the application must
// compute it.
ngtcp2_duration pto = ngtcp2_conn_get_pto(*this);
uint8_t multiplier = impl_->config_.options.draining_period_multiplier;
expiry = now + multiplier * pto;
} else {
expiry = ngtcp2_conn_get_expiry(*this);
}
Debug(
this, "Updating timer. Expiry: %" PRIu64 ", now: %" PRIu64, expiry, now);
if (expiry <= now) {
// The timer has already expired.
return OnTimeout();
}
auto timeout = (expiry - now) / NGTCP2_MILLISECONDS;
Debug(this, "Updating timeout to %zu milliseconds", timeout);
// If a stream idle timeout is configured, ensure the timer fires at
// least that often so CheckStreamIdleTimeout runs. Without this, an
// idle session with idle streams might not fire the timer until the
// connection idle timeout, which could be much longer.
uint64_t stream_idle = options().stream_idle_timeout;
if (stream_idle > 0 && timeout > stream_idle) {
timeout = stream_idle;
}
// If timeout is zero here, it means our timer is less than a millisecond
// off from expiry. Let's bump the timer to 1.
impl_->timer_.Update(timeout == 0 ? 1 : timeout);
}
void Session::DatagramStatus(datagram_id datagramId,
quic::DatagramStatus status) {
DCHECK(!is_destroyed());
switch (status) {
case DatagramStatus::ACKNOWLEDGED: {
Debug(this, "Datagram %" PRIu64 " was acknowledged", datagramId);
STAT_INCREMENT(Stats, datagrams_acknowledged);
break;
}
case DatagramStatus::LOST: {
Debug(this, "Datagram %" PRIu64 " was lost", datagramId);
STAT_INCREMENT(Stats, datagrams_lost);
break;
}
case DatagramStatus::ABANDONED: {
Debug(this, "Datagram %" PRIu64 " was abandoned", datagramId);
STAT_INCREMENT(Stats, datagrams_lost);
break;
}
}
if (HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::DATAGRAM_STATUS)) {
EmitDatagramStatus(datagramId, status);
}
}
void Session::DatagramReceived(const uint8_t* data,
size_t datalen,
DatagramReceivedFlags flag) {
DCHECK(!is_destroyed());
// If there is nothing watching for the datagram on the JavaScript side,
// or if the datagram is zero-length, we just drop it on the floor.
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::DATAGRAM) ||
datalen == 0)
return;
Debug(this, "Session is receiving datagram of size %zu", datalen);
STAT_INCREMENT(Stats, datagrams_received);
JS_TRY_ALLOCATE_BACKING(env(), backing, datalen)
memcpy(backing->Data(), data, datalen);
EmitDatagram(Store(std::move(backing), datalen), flag);
}
void Session::GenerateNewConnectionId(ngtcp2_cid* cid,
size_t len,
ngtcp2_stateless_reset_token* token) {
DCHECK(!is_destroyed());
CID cid_ = impl_->config_.options.cid_factory->GenerateInto(cid, len);
Debug(this, "Generated new connection id %s", cid_);
StatelessResetToken new_token(
token, endpoint().options().reset_token_secret, cid_);
endpoint().AssociateCID(cid_, impl_->config_.scid);
endpoint().AssociateStatelessResetToken(new_token, this);
}
bool Session::HandshakeCompleted() {
DCHECK(!is_destroyed());
DCHECK(!impl_->state()->handshake_completed);
Debug(this, "Session handshake completed");
impl_->state()->handshake_completed = 1;
// This implies fully completing a handshake without setting hello_processed
// (set during ALPN negotiation). Should be impossible unless ALPN flow is
// changed drastically, but good to check as it'd lose sessions.
DCHECK(!is_server() || hello_processed_);
STAT_RECORD_TIMESTAMP(Stats, handshake_completed_at);
SetStreamOpenAllowed();
// Capture the peer's max datagram frame size from the remote transport
// parameters so JavaScript can check it without a C++ round-trip.
const ngtcp2_transport_params* tp = remote_transport_params();
impl_->state()->max_datagram_size =
MaxDatagramPayload(tp->max_datagram_frame_size);
// If early data was attempted but rejected by the server,
// tell ngtcp2 so it can retransmit the data as 1-RTT.
// The status of early data will only be rejected if an
// attempt was actually made to send early data.
if (!is_server() && tls_session().early_data_was_rejected())
ngtcp2_conn_tls_early_data_rejected(*this);
// When in a server session, handshake completed == handshake confirmed.
if (is_server()) {
HandshakeConfirmed();
auto& ep = endpoint();
if (!ep.is_closed() && !ep.is_closing()) {
auto token = ep.GenerateNewToken(version(), impl_->remote_address_);
ngtcp2_vec vec = token;
if (NGTCP2_ERR(ngtcp2_conn_submit_new_token(*this, vec.base, vec.len))) {
// Submitting the new token failed... In this case we're going to
// fail because submitting the new token should only fail if we
// ran out of memory or some other unrecoverable state.
return false;
}
}
}
EmitHandshakeComplete();
return true;
}
void Session::HandshakeConfirmed() {
DCHECK(!is_destroyed());
DCHECK(!impl_->state()->handshake_confirmed);
Debug(this, "Session handshake confirmed");
impl_->state()->handshake_confirmed = 1;
STAT_RECORD_TIMESTAMP(Stats, handshake_confirmed_at);
}
void Session::SelectPreferredAddress(PreferredAddress* preferredAddress) {
if (options().preferred_address_strategy ==
PreferredAddress::Policy::IGNORE_PREFERRED) {
Debug(this, "Ignoring preferred address");
return;
}
switch (endpoint().local_address().family()) {
case AF_INET: {
Debug(this, "Selecting preferred address for AF_INET");
auto ipv4 = preferredAddress->ipv4();
if (ipv4.has_value()) {
if (ipv4->host[0] == '\0' || ipv4->port == 0) return;
CHECK(SocketAddress::New(
AF_INET, ipv4->host, ipv4->port, &impl_->remote_address_));
preferredAddress->Use(ipv4.value());
}
break;
}
case AF_INET6: {
Debug(this, "Selecting preferred address for AF_INET6");
auto ipv6 = preferredAddress->ipv6();
if (ipv6.has_value()) {
if (ipv6->host[0] == '\0' || ipv6->port == 0) return;
CHECK(SocketAddress::New(
AF_INET6, ipv6->host, ipv6->port, &impl_->remote_address_));
preferredAddress->Use(ipv6.value());
}
break;
}
}
}
CID Session::new_cid(size_t len) const {
return options().cid_factory->Generate(len);
}
void Session::ProcessPendingBidiStreams() {
// It shouldn't be possible to get here if can_create_streams() is false.
DCHECK(can_create_streams());
stream_id id;
while (!impl_->pending_bidi_stream_queue_.IsEmpty()) {
if (ngtcp2_conn_get_streams_bidi_left(*this) == 0) {
return;
}
switch (ngtcp2_conn_open_bidi_stream(*this, &id, nullptr)) {
case 0: {
impl_->pending_bidi_stream_queue_.PopFront()->fulfill(id);
continue;
}
case NGTCP2_ERR_STREAM_ID_BLOCKED: {
// This case really should not happen since we've checked the number
// of bidi streams left above. However, if it does happen we'll treat
// it the same as if the get_streams_bidi_left call returned zero.
return;
}
default: {
// We failed to open the stream for some reason other than being
// blocked. Report the failure.
impl_->pending_bidi_stream_queue_.PopFront()->reject(
QuicError::ForTransport(NGTCP2_STREAM_LIMIT_ERROR));
continue;
}
}
}
}
void Session::ProcessPendingUniStreams() {
// It shouldn't be possible to get here if can_create_streams() is false.
DCHECK(can_create_streams());
stream_id id;
while (!impl_->pending_uni_stream_queue_.IsEmpty()) {
if (ngtcp2_conn_get_streams_uni_left(*this) == 0) {
return;
}
switch (ngtcp2_conn_open_uni_stream(*this, &id, nullptr)) {
case 0: {
impl_->pending_uni_stream_queue_.PopFront()->fulfill(id);
continue;
}
case NGTCP2_ERR_STREAM_ID_BLOCKED: {
// This case really should not happen since we've checked the number
// of uni streams left above. However, if it does happen we'll treat
// it the same as if the get_streams_uni_left call returned zero.
return;
}
default: {
// We failed to open the stream for some reason other than being
// blocked. Report the failure.
impl_->pending_uni_stream_queue_.PopFront()->reject(
QuicError::ForTransport(NGTCP2_STREAM_LIMIT_ERROR));
continue;
}
}
}
}
// JavaScript callouts
void Session::EmitClose(const QuicError& error) {
DCHECK(!is_destroyed());
// When EmitClose is called, the expectation is that the JavaScript
// side will close the loop and call destroy on the underlying session.
// If we cannot call out into JavaScript at this point, go ahead and
// skip to calling destroy directly.
if (!env()->can_call_into_js()) return Destroy();
CallbackScope<Session> cb_scope(this);
Local<Value> argv[] = {
Integer::New(env()->isolate(), static_cast<int>(error.type())),
BigInt::NewFromUnsigned(env()->isolate(), error.code()),
Undefined(env()->isolate()),
Undefined(env()->isolate()),
};
if (error.reason().length() > 0 &&
!ToV8Value(env()->context(), error.reason()).ToLocal(&argv[2])) {
return;
}
// Attach a human-readable name for known wire codes (RFC 9000 sec. 20.1
// names and OpenSSL TLS alert descriptions for CRYPTO_ERROR). Unknown
// codes leave the slot as undefined. See QuicError::name() for the
// matching path on stream-level errors.
if (const char* n = error.name()) {
argv[3] = BindingData::Get(env()).error_name_string(n);
}
MakeCallback(
BindingData::Get(env()).session_close_callback(), arraysize(argv), argv);
// Importantly, the session instance itself should have been destroyed!
CHECK(is_destroyed());
}
void Session::set_max_datagram_size(uint16_t size) {
if (!is_destroyed()) {
impl_->state()->max_datagram_size = size;
}
}
void Session::EmitGoaway(stream_id last_stream_id) {
if (is_destroyed()) return;
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
Local<Value> argv[] = {
BigInt::New(env()->isolate(), last_stream_id),
};
MakeCallback(
BindingData::Get(env()).session_goaway_callback(), arraysize(argv), argv);
}
void Session::EmitDatagram(Store&& datagram, DatagramReceivedFlags flag) {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cbv_scope(this);
Local<Value> argv[] = {datagram.ToUint8Array(env()),
Boolean::New(env()->isolate(), flag.early)};
MakeCallback(BindingData::Get(env()).session_datagram_callback(),
arraysize(argv),
argv);
}
void Session::EmitDatagramStatus(datagram_id id, quic::DatagramStatus status) {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
auto& state = BindingData::Get(env());
const auto status_to_string = ([&] {
switch (status) {
case DatagramStatus::ACKNOWLEDGED:
return state.acknowledged_string();
case DatagramStatus::LOST:
return state.lost_string();
case DatagramStatus::ABANDONED:
return state.abandoned_string();
}
UNREACHABLE();
})();
Local<Value> argv[] = {BigInt::NewFromUnsigned(env()->isolate(), id),
status_to_string};
MakeCallback(state.session_datagram_status_callback(), arraysize(argv), argv);
}
void Session::EmitHandshakeComplete() {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
auto isolate = env()->isolate();
static constexpr auto kServerName = 0;
static constexpr auto kSelectedAlpn = 1;
static constexpr auto kCipherName = 2;
static constexpr auto kCipherVersion = 3;
static constexpr auto kValidationErrorReason = 4;
static constexpr auto kValidationErrorCode = 5;
Local<Value> argv[] = {
Undefined(isolate), // The negotiated server name
Undefined(isolate), // The selected protocol
Undefined(isolate), // Cipher name
Undefined(isolate), // Cipher version
Undefined(isolate), // Validation error reason
Undefined(isolate), // Validation error code
Boolean::New(isolate, tls_session().early_data_was_attempted()),
Boolean::New(isolate, tls_session().early_data_was_accepted())};
auto& tls = tls_session();
auto peerVerifyError = tls.VerifyPeerIdentity(env());
if (peerVerifyError.has_value() &&
(!peerVerifyError->reason.ToLocal(&argv[kValidationErrorReason]) ||
!peerVerifyError->code.ToLocal(&argv[kValidationErrorCode]))) {
return;
}
if (!ToV8Value(env()->context(), tls.servername())
.ToLocal(&argv[kServerName]) ||
!ToV8Value(env()->context(), tls.protocol())
.ToLocal(&argv[kSelectedAlpn]) ||
!tls.cipher_name(env()).ToLocal(&argv[kCipherName]) ||
!tls.cipher_version(env()).ToLocal(&argv[kCipherVersion])) {
return;
}
MakeCallback(BindingData::Get(env()).session_handshake_callback(),
arraysize(argv),
argv);
}
void Session::EmitPathValidation(PathValidationResult result,
PathValidationFlags flags,
const ValidatedPath& newPath,
const std::optional<ValidatedPath>& oldPath) {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::PATH_VALIDATION)) [[likely]] {
return;
}
auto isolate = env()->isolate();
CallbackScope<Session> cb_scope(this);
auto& state = BindingData::Get(env());
const auto resultToString = ([&] {
switch (result) {
case PathValidationResult::ABORTED:
return state.aborted_string();
case PathValidationResult::FAILURE:
return state.failure_string();
case PathValidationResult::SUCCESS:
return state.success_string();
}
UNREACHABLE();
})();
Local<Value> argv[] = {
resultToString,
SocketAddressBase::Create(env(), newPath.local)->object(),
SocketAddressBase::Create(env(), newPath.remote)->object(),
Undefined(isolate),
Undefined(isolate),
is_server() ? Undefined(isolate)
: Boolean::New(isolate, flags.preferredAddress)};
if (oldPath.has_value()) {
argv[3] = SocketAddressBase::Create(env(), oldPath->local)->object();
argv[4] = SocketAddressBase::Create(env(), oldPath->remote)->object();
}
Debug(this, "Notifying JavaScript of path validation");
MakeCallback(state.session_path_validation_callback(), arraysize(argv), argv);
}
void Session::EmitSessionTicket(Store&& ticket) {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
// If there is nothing listening for the session ticket, don't bother
// emitting.
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::SESSION_TICKET)) [[likely]] {
Debug(this, "Session ticket was discarded");
return;
}
CallbackScope<Session> cb_scope(this);
// Encode the 0-RTT transport params using ngtcp2's matched pair format.
// This must use ngtcp2_conn_encode_0rtt_transport_params (not the
// generic ngtcp2_transport_params_encode_versioned) so that the
// receiver can decode with ngtcp2_conn_decode_and_set_0rtt_transport_params.
ssize_t tp_size = ngtcp2_conn_encode_0rtt_transport_params(*this, nullptr, 0);
if (tp_size > 0) {
JS_TRY_ALLOCATE_BACKING(env(), tp_backing, static_cast<size_t>(tp_size))
ssize_t tp_written = ngtcp2_conn_encode_0rtt_transport_params(
*this,
static_cast<uint8_t*>(tp_backing->Data()),
static_cast<size_t>(tp_size));
if (tp_written > 0) {
Store transport_params(std::move(tp_backing),
static_cast<size_t>(tp_written));
SessionTicket session_ticket(std::move(ticket),
std::move(transport_params));
Local<Value> argv;
if (session_ticket.encode(env()).ToLocal(&argv)) [[likely]] {
MakeCallback(
BindingData::Get(env()).session_ticket_callback(), 1, &argv);
}
}
}
}
void Session::EmitApplication() {
if (is_destroyed()) return;
if (!env()->can_call_into_js()) return;
if (!has_application()) {
// The application has not yet been selected (ALPN negotiation is not
// yet complete on the server) or the session has been destroyed. In
// either case, the application options are not available.
// Should not happen, but we bail out
return;
}
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::APPLICATION)) [[likely]] {
return;
}
CallbackScope<Session> cb_scope(this);
Local<Value> argv;
auto& options = application().options();
if (options.ToObject(env()).ToLocal(&argv)) {
MakeCallback(
BindingData::Get(env()).session_application_callback(), 1, &argv);
}
}
void Session::DestroyAllStreams(const QuicError& error) {
DCHECK(!is_destroyed());
// Copy the streams map since streams remove themselves during
// destruction. Each Destroy() call triggers MakeCallback which
// can destroy impl_ via JS re-entrancy.
StreamsMap streams = impl_->streams_;
for (auto& stream : streams) {
if (is_destroyed()) return;
stream.second->Destroy(error);
}
}
void Session::EmitEarlyDataRejected() {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
MakeCallback(BindingData::Get(env()).session_early_data_rejected_callback(),
0,
nullptr);
}
void Session::EmitNewToken(const uint8_t* token, size_t len) {
DCHECK(!is_destroyed());
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::NEW_TOKEN))
return;
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
Local<Value> argv[2];
auto buf = Buffer::Copy(env(), reinterpret_cast<const char*>(token), len);
if (!buf.ToLocal(&argv[0])) return;
argv[1] = SocketAddressBase::Create(
env(), std::make_shared<SocketAddress>(remote_address()))
->object();
MakeCallback(BindingData::Get(env()).session_new_token_callback(),
arraysize(argv),
argv);
}
void Session::EmitStream(const BaseObjectWeakPtr<Stream>& stream) {
DCHECK(!is_destroyed());
if (!stream) return;
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
auto isolate = env()->isolate();
Local<Value> argv[] = {
stream->object(),
Integer::NewFromUnsigned(isolate,
static_cast<uint32_t>(stream->direction())),
};
MakeCallback(
BindingData::Get(env()).stream_created_callback(), arraysize(argv), argv);
}
void Session::EmitVersionNegotiation(const ngtcp2_pkt_hd& hd,
const uint32_t* sv,
size_t nsv) {
DCHECK(!is_destroyed());
DCHECK(!is_server());
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
auto& opts = options();
// version() is the version that was actually configured for this session.
// versions are the versions requested by the peer.
// supported are the versions supported by Node.js.
LocalVector<Value> versions(env()->isolate(), nsv);
for (size_t n = 0; n < nsv; n++) {
versions[n] = Integer::NewFromUnsigned(env()->isolate(), sv[n]);
}
// supported are the versions we actually support expressed as a range.
// The first value is the minimum version, the second is the maximum.
Local<Value> supported[] = {
Integer::NewFromUnsigned(env()->isolate(), opts.min_version),
Integer::NewFromUnsigned(env()->isolate(), opts.version)};
Local<Value> argv[] = {
// The version configured for this session.
Integer::NewFromUnsigned(env()->isolate(), version()),
// The versions requested.
Array::New(env()->isolate(), versions.data(), nsv),
// The versions we actually support.
Array::New(env()->isolate(), supported, arraysize(supported))};
MakeCallback(BindingData::Get(env()).session_version_negotiation_callback(),
arraysize(argv),
argv);
}
void Session::EmitOrigins(std::vector<std::string>&& origins) {
DCHECK(!is_destroyed());
if (!HasListenerFlag(impl_->state()->listener_flags,
SessionListenerFlags::ORIGIN))
return;
if (!env()->can_call_into_js()) return;
CallbackScope<Session> cb_scope(this);
auto isolate = env()->isolate();
LocalVector<Value> elements(env()->isolate(), origins.size());
for (size_t i = 0; i < origins.size(); i++) {
Local<Value> str;
if (!ToV8Value(env()->context(), origins[i]).ToLocal(&str)) [[unlikely]] {
return;
}
elements[i] = str;
}
Local<Value> argv[] = {Array::New(isolate, elements.data(), elements.size())};
MakeCallback(
BindingData::Get(env()).session_origin_callback(), arraysize(argv), argv);
}
void Session::EmitKeylog(const char* line) {
DCHECK(!is_destroyed());
if (!env()->can_call_into_js()) return;
Local<Value> argv[] = {Undefined(env()->isolate())};
if (!ToV8Value(env()->context(), std::string(line)).ToLocal(&argv[0])) {
Debug(this, "Failed to convert keylog line to V8 string");
return;
}
MakeCallback(
BindingData::Get(env()).session_keylog_callback(), arraysize(argv), argv);
}
// ============================================================================
#define V(name, key, no_side_effect) \
if (no_side_effect) { \
SetProtoMethodNoSideEffect(env->isolate(), tmpl, #key, Impl::name); \
} else { \
SetProtoMethod(env->isolate(), tmpl, #key, Impl::name); \
}
JS_CONSTRUCTOR_IMPL(Session, session_constructor_template, {
JS_ILLEGAL_CONSTRUCTOR();
JS_INHERIT(AsyncWrap);
JS_CLASS(session);
SESSION_JS_METHODS(V)
})
#undef V
void Session::RegisterExternalReferences(ExternalReferenceRegistry* registry) {
#define V(name, _, __) registry->Register(Impl::name);
SESSION_JS_METHODS(V)
#undef V
}
void Session::InitPerIsolate(IsolateData* data, Local<ObjectTemplate> target) {
// TODO(@jasnell): Implement the per-isolate state
}
void Session::InitPerContext(Realm* realm, Local<Object> target) {
#define V(name, str) \
NODE_DEFINE_CONSTANT(target, CC_ALGO_##name); \
NODE_DEFINE_STRING_CONSTANT(target, "CC_ALGO_" #name "_STR", #str);
CC_ALGOS(V)
#undef V
// Make sure the Session constructor template is initialized.
USE(GetConstructorTemplate(realm->env()));
TransportParams::Initialize(realm->env(), target);
PreferredAddress::Initialize(realm->env(), target);
static constexpr auto STREAM_DIRECTION_BIDIRECTIONAL =
static_cast<uint8_t>(Direction::BIDIRECTIONAL);
static constexpr auto STREAM_DIRECTION_UNIDIRECTIONAL =
static_cast<uint8_t>(Direction::UNIDIRECTIONAL);
static constexpr auto QUIC_PROTO_MAX = NGTCP2_PROTO_VER_MAX;
static constexpr auto QUIC_PROTO_MIN = NGTCP2_PROTO_VER_MIN;
NODE_DEFINE_CONSTANT(target, STREAM_DIRECTION_BIDIRECTIONAL);
NODE_DEFINE_CONSTANT(target, STREAM_DIRECTION_UNIDIRECTIONAL);
NODE_DEFINE_CONSTANT(target, DEFAULT_MAX_HEADER_LIST_PAIRS);
NODE_DEFINE_CONSTANT(target, DEFAULT_MAX_HEADER_LENGTH);
NODE_DEFINE_CONSTANT(target, QUIC_PROTO_MAX);
NODE_DEFINE_CONSTANT(target, QUIC_PROTO_MIN);
static constexpr auto DEFAULT_HANDSHAKE_TIMEOUT =
Session::Options::DEFAULT_HANDSHAKE_TIMEOUT;
NODE_DEFINE_CONSTANT(target, DEFAULT_HANDSHAKE_TIMEOUT);
NODE_DEFINE_STRING_CONSTANT(
target, "DEFAULT_CIPHERS", TLSContext::DEFAULT_CIPHERS);
NODE_DEFINE_STRING_CONSTANT(
target, "DEFAULT_GROUPS", TLSContext::DEFAULT_GROUPS);
#define V(name, _) IDX_STATS_SESSION_##name,
enum SessionStatsIdx { SESSION_STATS(V) IDX_STATS_SESSION_COUNT };
#undef V
#define V(name, key, __) \
auto IDX_STATE_SESSION_##name = offsetof(Session::State, key);
SESSION_STATE(V)
#undef V
NODE_DEFINE_CONSTANT(target, IDX_STATS_SESSION_COUNT);
#define V(name, _) NODE_DEFINE_CONSTANT(target, IDX_STATS_SESSION_##name);
SESSION_STATS(V)
NODE_DEFINE_CONSTANT(target, IDX_STATS_SESSION_COUNT);
#undef V
#define V(name, _, __) NODE_DEFINE_CONSTANT(target, IDX_STATE_SESSION_##name);
SESSION_STATE(V)
#undef V
}
} // namespace quic
} // namespace node
#endif // OPENSSL_NO_QUIC