#include "ngtcp2/ngtcp2.h"

#if HAVE_OPENSSL && HAVE_QUIC

#include "guard.h"

#ifndef OPENSSL_NO_QUIC

#include <aliased_struct-inl.h>

#include <async_wrap-inl.h>

#include <base_object-inl.h>

#include <env-inl.h>

#include <memory_tracker-inl.h>

#include <node_blob.h>

#include <node_bob-inl.h>

#include <node_file.h>

#include <node_sockaddr-inl.h>

#include "application.h"

#include "bindingdata.h"

#include "defs.h"

#include "session.h"

#include "streams.h"

namespace node {

using v8::Array;

using v8::ArrayBuffer;

using v8::ArrayBufferView;

using v8::BackingStore;

using v8::BackingStoreInitializationMode;

using v8::BigInt;

using v8::FunctionCallbackInfo;

using v8::Global;

using v8::HandleScope;

using v8::Integer;

using v8::Just;

using v8::Local;

using v8::LocalVector;

using v8::Maybe;

using v8::Nothing;

using v8::Object;

using v8::ObjectTemplate;

using v8::SharedArrayBuffer;

using v8::String;

using v8::Uint32;

using v8::Uint8Array;

using v8::Value;

namespace quic {

#define STREAM_STATE(V) \

V(ID, id, stream_id) \

V(PENDING, pending, uint8_t) \

V(FIN_SENT, fin_sent, uint8_t) \

V(FIN_RECEIVED, fin_received, uint8_t) \

V(READ_ENDED, read_ended, uint8_t) \

V(WRITE_ENDED, write_ended, uint8_t) \

V(RESET, reset, uint8_t) \

V(RESET_CODE, reset_code, uint64_t) \

V(HAS_OUTBOUND, has_outbound, uint8_t) \

V(HAS_READER, has_reader, uint8_t) \

/* Set when the stream has a block event handler */ \

V(WANTS_BLOCK, wants_block, uint8_t) \

/* Set when the stream has a headers event handler */ \

V(WANTS_HEADERS, wants_headers, uint8_t) \

/* Set when the stream has a reset event handler */ \

V(WANTS_RESET, wants_reset, uint8_t) \

/* Set when the stream has a trailers event handler */ \

V(WANTS_TRAILERS, wants_trailers, uint8_t) \

/* True when 0-RTT early data was received */ \

V(RECEIVED_EARLY_DATA, received_early_data, uint8_t) \

V(WRITE_DESIRED_SIZE, write_desired_size, uint32_t) \

V(HIGH_WATER_MARK, high_water_mark, uint32_t)

#define STREAM_STATS(V) \

/* Marks the timestamp when the stream object was created. */ \

V(CREATED_AT, created_at) \

/* Marks the timestamp when the stream was opened. This can be different */ \

/* from the created_at timestamp if the stream was created in as pending */ \

V(OPENED_AT, opened_at) \

/* Marks the timestamp when the stream last received data */ \

V(RECEIVED_AT, received_at) \

/* Marks the timestamp when the stream last received an acknowledgement */ \

V(ACKED_AT, acked_at) \

/* Marks the timestamp when the stream was destroyed */ \

V(DESTROYED_AT, destroyed_at) \

/* Records the total number of bytes received by the stream */ \

V(BYTES_RECEIVED, bytes_received) \

/* Records the total number of bytes sent by the stream */ \

V(BYTES_SENT, bytes_sent) \

V(MAX_OFFSET, max_offset) \

V(MAX_OFFSET_ACK, max_offset_ack) \

V(MAX_OFFSET_RECV, max_offset_received) \

V(FINAL_SIZE, final_size) \

/* Bytes in the receive accumulation buffer */ \

V(BYTES_ACCUMULATED, bytes_accumulated) \

/* Peak bytes accumulated over stream lifetime */ \

V(MAX_BYTES_ACCUMULATED, max_bytes_accumulated)

#define STREAM_JS_METHODS(V) \

V(AttachSource, attachSource, false) \

V(Destroy, destroy, false) \

V(SendHeaders, sendHeaders, false) \

V(StopSending, stopSending, false) \

V(ResetStream, resetStream, false) \

V(SetPriority, setPriority, false) \

V(GetPriority, getPriority, true) \

V(GetReader, getReader, false) \

V(InitStreamingSource, initStreamingSource, false) \

V(Write, write, false) \

V(EndWrite, endWrite, false)

// ============================================================================

// RecvAccumulator implementation

RecvAccumulator::RecvAccumulator(size_t max_capacity)

: buf_(kMinCapacity), max_capacity_(std::max(max_capacity, kMinCapacity)) {}

size_t RecvAccumulator::Write(const uint8_t* data, size_t len) {

if (len == 0) return 0;

size_t capacity = buf_.size();

// If the buffer is full, caller must flush or grow first.

size_t space = remaining();

if (space == 0) return 0;

size_t to_write = std::min(len, space);

// Write into the buffer, handling wrap-around.

size_t physical_write = write_pos_ % capacity;

size_t first_chunk = std::min(to_write, capacity - physical_write);

memcpy(buf_.data() + physical_write, data, first_chunk);

if (first_chunk < to_write) {

memcpy(buf_.data(), data + first_chunk, to_write - first_chunk);

}

write_pos_ += to_write;

len_ += to_write;

return to_write;

}

std::unique_ptr<DataQueue::Entry> RecvAccumulator::Flush(Environment* env) {

if (len_ == 0) return nullptr;

size_t capacity = buf_.size();

auto store = ArrayBuffer::NewBackingStore(

env->isolate(), len_, BackingStoreInitializationMode::kUninitialized);

auto* dest = static_cast<uint8_t*>(store->Data());

// Copy from the ring buffer. Handle the wrap-around case.

size_t physical_read = read_pos_ % capacity;

size_t first_chunk = std::min(len_, capacity - physical_read);

memcpy(dest, buf_.data() + physical_read, first_chunk);

if (first_chunk < len_) {

memcpy(dest + first_chunk, buf_.data(), len_ - first_chunk);

}

size_t flushed = len_;

// Reset cursors.

read_pos_ = 0;

write_pos_ = 0;

len_ = 0;

// Shrink back toward kMinCapacity if we had expanded.

if (capacity > kMinCapacity) {

buf_.resize(kMinCapacity);

buf_.shrink_to_fit();

}

return DataQueue::CreateInMemoryEntryFromBackingStore(

std::move(store), 0, flushed);

}

void RecvAccumulator::Grow() {

size_t capacity = buf_.size();

size_t new_capacity = std::min(capacity * 2, max_capacity_);

if (new_capacity <= capacity) return; // already at max

// Linearize the data and grow in one step.

std::vector<uint8_t> new_buf(new_capacity);

if (len_ > 0) {

size_t physical_read = read_pos_ % capacity;

size_t first_chunk = std::min(len_, capacity - physical_read);

memcpy(new_buf.data(), buf_.data() + physical_read, first_chunk);

if (first_chunk < len_) {

memcpy(new_buf.data() + first_chunk, buf_.data(), len_ - first_chunk);

}

}

buf_ = std::move(new_buf);

read_pos_ = 0;

write_pos_ = len_;

}

// ============================================================================

PendingStream::PendingStream(Direction direction,

Stream* stream,

BaseObjectWeakPtr<Session> session)

: direction_(direction), stream_(stream), session_(session) {

if (session_) {

if (direction == Direction::BIDIRECTIONAL) {

session_->pending_bidi_stream_queue().PushBack(this);

} else {

session_->pending_uni_stream_queue().PushBack(this);

}

}

}

PendingStream::~PendingStream() {

pending_stream_queue_.Remove();

if (waiting_) {

Debug(stream_, "A pending stream was canceled");

}

}

void PendingStream::fulfill(stream_id id) {

CHECK(waiting_);

waiting_ = false;

stream_->NotifyStreamOpened(id);

}

void PendingStream::reject(QuicError error) {

CHECK(waiting_);

waiting_ = false;

stream_->Destroy(error);

}

struct Stream::PendingHeaders {

HeadersKind kind;

Global<Array> headers;

HeadersFlags flags;

PendingHeaders(HeadersKind kind_, Global<Array> headers_, HeadersFlags flags_)

: kind(kind_), headers(std::move(headers_)), flags(flags_) {}

DISALLOW_COPY_AND_MOVE(PendingHeaders)

};

// ============================================================================

struct Stream::State {

#define V(_, name, type) type name;

STREAM_STATE(V)

#undef V

};

STAT_STRUCT(Stream, STREAM)

// Stream uses arena-allocated stats, not AliasedStruct, so override the

// STAT_* macros to use the stats() accessor 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>(stats());

#define STAT_INCREMENT_N(Type, name, amt) \

IncrementStat<Type, &Type::name>(stats(), amt);

#define STAT_RECORD_TIMESTAMP(Type, name) \

RecordTimestampStat<Type, &Type::name>(stats());

#define STAT_SET(Type, name, val) SetStat<Type, &Type::name>(stats(), val)

#define STAT_GET(Type, name) GetStat<Type, &Type::name>(stats())

using StreamStateArena = AliasedStructArena<Stream::State>;

using StreamStatsArena = AliasedStructArena<Stream::Stats>;

namespace {

StreamStateArena& GetStreamStateArena(BindingData& binding) {

if (!binding.stream_state_arena_) {

auto* arena = new StreamStateArena();

binding.stream_state_arena_ = BindingData::ArenaPtr(

arena, +[](void* p) { delete static_cast<StreamStateArena*>(p); });

}

return *static_cast<StreamStateArena*>(binding.stream_state_arena_.get());

}

StreamStatsArena& GetStreamStatsArena(BindingData& binding) {

if (!binding.stream_stats_arena_) {

auto* arena = new StreamStatsArena();

binding.stream_stats_arena_ = BindingData::ArenaPtr(

arena, +[](void* p) { delete static_cast<StreamStatsArena*>(p); });

}

return *static_cast<StreamStatsArena*>(binding.stream_stats_arena_.get());

}

} // namespace

// ============================================================================

namespace {

// Creates an in-memory DataQueue entry by copying the requested range of

// the given ArrayBuffer into a fresh BackingStore. The caller's buffer is

// not detached or otherwise modified, so callers can safely reuse or

// mutate it after the call returns. Callers that want to ensure their

// buffer cannot be mutated after handing it off can call

// `ArrayBuffer.prototype.transfer()` themselves before calling into the

// QUIC API.

// Returns nullptr on zero length or allocation failure.

std::unique_ptr<DataQueue::Entry> CreateEntryFromBuffer(

Environment* env, Local<ArrayBuffer> buffer, size_t offset, size_t length) {

if (length == 0) return nullptr;

JS_TRY_ALLOCATE_BACKING_OR_RETURN(env, copy, length, nullptr);

memcpy(copy->Data(),

static_cast<const uint8_t*>(buffer->Data()) + offset,

length);

return DataQueue::CreateInMemoryEntryFromBackingStore(

std::move(copy), 0, length);

}

} // namespace

Maybe<std::shared_ptr<DataQueue>> Stream::GetDataQueueFromSource(

Environment* env, Local<Value> value) {

DCHECK_IMPLIES(!value->IsUndefined(), value->IsObject());

std::vector<std::unique_ptr<DataQueue::Entry>> entries;

if (value->IsUndefined()) {

// Return an empty DataQueue.

return Just(std::shared_ptr<DataQueue>());

} else if (value->IsArrayBuffer()) {

auto buffer = value.As<ArrayBuffer>();

auto length = buffer->ByteLength();

if (length > 0) {

auto entry = CreateEntryFromBuffer(env, buffer, 0, length);

if (!entry) {

return Nothing<std::shared_ptr<DataQueue>>();

}

entries.push_back(std::move(entry));

}

return Just(DataQueue::CreateIdempotent(std::move(entries)));

} else if (value->IsSharedArrayBuffer()) {

auto sab = value.As<SharedArrayBuffer>();

auto length = sab->ByteLength();

if (length > 0) {

// SharedArrayBuffer cannot be detached, so we always copy. Note that

// because of the nature of SAB, another thread can end up modifying

// the SAB while we're copying, which is racy but unavoidable.

JS_TRY_ALLOCATE_BACKING_OR_RETURN(

env, backing, length, Nothing<std::shared_ptr<DataQueue>>());

memcpy(backing->Data(), sab->Data(), length);

entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(

std::move(backing), 0, length));

}

return Just(DataQueue::CreateIdempotent(std::move(entries)));

} else if (value->IsArrayBufferView()) {

auto view = value.As<ArrayBufferView>();

auto offset = view->ByteOffset();

auto length = view->ByteLength();

if (length > 0) {

auto entry = CreateEntryFromBuffer(env, view->Buffer(), offset, length);

if (!entry) {

return Nothing<std::shared_ptr<DataQueue>>();

}

entries.push_back(std::move(entry));

}

return Just(DataQueue::CreateIdempotent(std::move(entries)));

} else if (Blob::HasInstance(env, value)) {

Blob* blob;

ASSIGN_OR_RETURN_UNWRAP(

&blob, value, Nothing<std::shared_ptr<DataQueue>>());

return Just(blob->getDataQueue().slice(0));

} else if (value->IsString()) {

Utf8Value str(env->isolate(), value);

JS_TRY_ALLOCATE_BACKING_OR_RETURN(

env, backing, str.length(), Nothing<std::shared_ptr<DataQueue>>());

memcpy(backing->Data(), *str, str.length());

auto len = backing->ByteLength();

entries.push_back(DataQueue::CreateInMemoryEntryFromBackingStore(

std::move(backing), 0, len));

return Just(DataQueue::CreateIdempotent(std::move(entries)));

}

// FileHandle — create an fd-backed DataQueue from the file path.

// The JS side validates and locks the FileHandle before passing

// the C++ handle here. We detect FileHandle by checking if the

// object's constructor name is "FileHandle".

if (value->IsObject()) {

auto obj = value.As<Object>();

Local<String> ctor_name;

auto maybe_name = obj->GetConstructorName();

if (!maybe_name.IsEmpty()) {

ctor_name = maybe_name;

Utf8Value name(env->isolate(), ctor_name);

if (strcmp(*name, "FileHandle") == 0) {

fs::FileHandle* file_handle;

ASSIGN_OR_RETURN_UNWRAP(

&file_handle, value, Nothing<std::shared_ptr<DataQueue>>());

Local<Value> path;

if (!ToV8Value(env->context(), file_handle->original_name())

.ToLocal(&path)) {

return Nothing<std::shared_ptr<DataQueue>>();

}

auto entry = DataQueue::CreateFdEntry(env, path);

if (!entry) return Nothing<std::shared_ptr<DataQueue>>();

size_t size = entry->size().value_or(0);

auto queue = DataQueue::Create();

if (!queue) return Nothing<std::shared_ptr<DataQueue>>();

queue->append(std::move(entry));

queue->cap(size);

return Just(std::move(queue));

}

}

}

// TODO(jasnell): Add streaming sources...

THROW_ERR_INVALID_ARG_TYPE(env, "Invalid data source type");

return Nothing<std::shared_ptr<DataQueue>>();

}

// Provides the implementation of the various JavaScript APIs for the

// Stream object.

struct Stream::Impl {

// Attaches an outbound data source to the stream.

JS_METHOD(AttachSource) {

Environment* env = Environment::GetCurrent(args);

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

std::shared_ptr<DataQueue> dataqueue;

if (GetDataQueueFromSource(env, args[0]).To(&dataqueue)) {

stream->set_outbound(std::move(dataqueue));

// set_outbound does not call ResumeStream because during

// construction the stream is not yet registered with the session.

// When attaching a source after creation (via setBody), the

// stream is already registered and must be resumed to enter the

// send queue.

if (!stream->is_pending()) {

stream->session().ResumeStream(stream->id());

}

}

}

// Immediately and forcefully destroys the stream.

JS_METHOD(Destroy) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

if (args.Length() >= 1) {

CHECK(args[0]->IsBigInt());

bool lossless = false;

uint64_t code = args[0].As<BigInt>()->Uint64Value(&lossless);

// If the code cannot be represented in 64 bits, it is too large to be

// a valid QUIC error code, error!

if (!lossless) {

THROW_ERR_INVALID_ARG_TYPE(stream->env(), "Error code is too large");

return;

}

stream->Destroy(QuicError::ForApplication(code));

} else {

stream->Destroy();

}

}

// Sends a block of headers to the peer. If the stream is not yet open,

// the headers will be queued and sent immediately when the stream is

// opened. Returns false if the application does not support headers.

JS_METHOD(SendHeaders) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

CHECK(args[0]->IsUint32()); // Kind

CHECK(args[1]->IsArray()); // Headers

CHECK(args[2]->IsUint32()); // Flags

HeadersKind kind = FromV8Value<HeadersKind>(args[0]);

Local<Array> headers = args[1].As<Array>();

HeadersFlags flags = FromV8Value<HeadersFlags>(args[2]);

// If the stream is pending, the headers will be queued until the

// stream is opened, at which time the queued header block will be

// immediately sent when the stream is opened. If we already know

// that the application does not support headers, return false

// immediately so the JS side can throw an appropriate error.

if (stream->is_pending()) {

if (!stream->session().application().SupportsHeaders()) {

return args.GetReturnValue().Set(false);

}

stream->EnqueuePendingHeaders(kind, headers, flags);

return args.GetReturnValue().Set(true);

}

args.GetReturnValue().Set(stream->session().application().SendHeaders(

*stream, kind, headers, flags));

}

// Tells the peer to stop sending data for this stream. This has the effect

// of shutting down the readable side of the stream for this peer. Any data

// that has already been received is still readable.

JS_METHOD(StopSending) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

error_code code = 0;

CHECK_IMPLIES(!args[0]->IsUndefined(), args[0]->IsBigInt());

if (!args[0]->IsUndefined()) {

bool unused = false; // not used but still necessary.

code = args[0].As<BigInt>()->Uint64Value(&unused);

}

stream->EndReadable();

if (!stream->is_pending()) {

// If the stream is a local unidirectional there's nothing to do here.

if (stream->is_local_unidirectional()) return;

stream->NotifyReadableEnded(code);

} else {

stream->pending_close_read_code_ = code;

}

}

// Sends a reset stream to the peer to tell it we will not be sending any

// more data for this stream. This has the effect of shutting down the

// writable side of the stream for this peer. Any data that is held in the

// outbound queue will be dropped. The stream may still be readable.

JS_METHOD(ResetStream) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

error_code code = 0;

CHECK_IMPLIES(!args[0]->IsUndefined(), args[0]->IsBigInt());

if (!args[0]->IsUndefined()) {

bool lossless = false; // not used but still necessary.

code = args[0].As<BigInt>()->Uint64Value(&lossless);

}

if (stream->state()->reset == 1) return;

stream->EndWritable();

// We can release our outbound here now. Since the stream is being reset

// on the ngtcp2 side, we do not need to keep any of the data around

// waiting for acknowledgement that will never come.

stream->outbound_.reset();

stream->state()->reset = 1;

if (!stream->is_pending()) {

if (stream->is_remote_unidirectional()) return;

stream->NotifyWritableEnded(code);

} else {

stream->pending_close_write_code_ = code;

}

}

JS_METHOD(SetPriority) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

CHECK(args[0]->IsUint32()); // Packed: (urgency << 1) | incremental

uint32_t packed = args[0].As<Uint32>()->Value();

StreamPriority priority = static_cast<StreamPriority>(packed >> 1);

StreamPriorityFlags flags = (packed & 1)

? StreamPriorityFlags::INCREMENTAL

: StreamPriorityFlags::NON_INCREMENTAL;

// Always update the stored priority on the stream.

stream->priority_ = StoredPriority{

.priority = priority,

.flags = flags,

.pending = stream->is_pending(),

};

if (!stream->is_pending()) {

stream->session().application().SetStreamPriority(

*stream, priority, flags);

}

}

JS_METHOD(GetPriority) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

// On the client side, priority is always read from the stream's

// stored value since the client is the one setting it. On the

// server side, we delegate to the application which can read

// the peer's requested priority (e.g., from PRIORITY_UPDATE

// frames in HTTP/3).

if (!stream->session().is_server()) {

auto& pri = stream->priority_;

uint32_t packed = (static_cast<uint32_t>(pri.priority) << 1) |

(pri.flags == StreamPriorityFlags::INCREMENTAL ? 1 : 0);

return args.GetReturnValue().Set(packed);

}

auto result = stream->session().application().GetStreamPriority(*stream);

uint32_t packed =

(static_cast<uint32_t>(result.priority) << 1) |

(result.flags == StreamPriorityFlags::INCREMENTAL ? 1 : 0);

args.GetReturnValue().Set(packed);

}

// Returns a Blob::Reader that can be used to read data that has been

// received on the stream.

JS_METHOD(GetReader) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

BaseObjectPtr<Blob::Reader> reader = stream->get_reader();

if (reader) args.GetReturnValue().Set(reader->object());

// Returns undefined when the stream is not readable (e.g. a local

// unidirectional stream). The JS side checks for this.

}

JS_METHOD(InitStreamingSource) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

stream->InitStreaming();

}

JS_METHOD(Write) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

stream->WriteStreamData(args);

}

JS_METHOD(EndWrite) {

Stream* stream;

ASSIGN_OR_RETURN_UNWRAP(&stream, args.This());

stream->EndWriting();

}

};

// ============================================================================

class Stream::Outbound final : public MemoryRetainer {

public:

explicit Outbound(Stream* stream, std::shared_ptr<DataQueue> queue)

: stream_(stream),

queue_(std::move(queue)),

reader_(queue_->get_reader()) {}

// Creates an Outbound in streaming mode with a non-idempotent DataQueue

// that can be appended to via AppendEntry().

explicit Outbound(Stream* stream)

: stream_(stream),

queue_(DataQueue::Create()),

reader_(queue_->get_reader()) {

flags_.streaming = true;

}

void Acknowledge(size_t amount) {

size_t remaining = std::min(amount, total_ - uncommitted_);

while (remaining > 0 && head_ != nullptr) {

DCHECK_LE(head_->ack_offset, head_->offset);

// The amount to acknowledge in this chunk is the lesser of the total

// amount remaining to acknowledge or the total remaining unacknowledged

// bytes in the chunk.

size_t amount_to_ack =

std::min(remaining, head_->offset - head_->ack_offset);

// If the amount to ack is zero here, it means our ack offset has caught

// up to our commit offset, which means there's nothing left to

// acknowledge yet. We could treat this as an error but let's just stop

// here.

if (amount_to_ack == 0) break;

// Adjust our remaining down and our ack_offset up...

remaining -= amount_to_ack;

head_->ack_offset += amount_to_ack;

// If we've fully acknowledged this chunk, free it and decrement total.

if (head_->ack_offset == head_->buf.len) {

DCHECK_GE(total_, head_->buf.len);

total_ -= head_->buf.len;

// if tail_ == head_ here, it means we've fully acknowledged our current

// buffer. Set tail to nullptr since we're freeing it here.

if (head_.get() == tail_) {

// In this case, commit_head_ should have already been set to nullptr.

// Because we should only have hit this case if the entire buffer

// had been committed.

DCHECK(commit_head_ == nullptr);

tail_ = nullptr;

}

head_ = std::move(head_->next);

DCHECK_IMPLIES(head_ == nullptr, tail_ == nullptr);

}

}

}

void Commit(size_t amount) {

// Commit amount number of bytes from the current uncommitted

// byte queue. Importantly, this does not remove the bytes

// from the byte queue.

size_t remaining = std::min(uncommitted_, amount);

// There's nothing to commit.

while (remaining > 0 && commit_head_ != nullptr) {

// The amount to commit is the lesser of the total amount remaining to

// commit and the remaining uncommitted bytes in this chunk.

size_t amount_to_commit = std::min(

remaining,

static_cast<size_t>(commit_head_->buf.len - commit_head_->offset));

// The amount to commit here should never be zero because that means we

// should have already advanced the commit head.

DCHECK_NE(amount_to_commit, 0);

uncommitted_ -= amount_to_commit;

remaining -= amount_to_commit;

commit_head_->offset += amount_to_commit;

if (commit_head_->offset == commit_head_->buf.len) {

count_--;

commit_head_ = commit_head_->next.get();

}

}

}

void Cap() {

// Calling cap without a value halts the ability to add any

// new data to the queue if it is not idempotent. If it is

// idempotent, it's a non-op.

if (queue_) queue_->cap();

}

bool is_streaming() const { return flags_.streaming; }

size_t total() const { return total_; }

size_t uncommitted() const { return uncommitted_; }

// Total bytes in the pipeline: data appended to the DataQueue that

// hasn't been pulled yet, plus data pulled but not yet acknowledged.

// This is the number to compare against highWaterMark for backpressure.

size_t queued_bytes() const { return queued_ + total_; }

// Appends an entry to the underlying DataQueue. Only valid when

// the Outbound was created in streaming mode.

bool AppendEntry(std::unique_ptr<DataQueue::Entry> entry) {

if (!flags_.streaming || !queue_) return false;

auto size = entry->size();

auto result = queue_->append(std::move(entry));

if (result.has_value() && result.value()) {

if (size.has_value()) queued_ += size.value();

return true;

}

return false;

}

int Pull(bob::Next<ngtcp2_vec> next,

int options,

ngtcp2_vec* data,

size_t count,

size_t max_count_hint) {

if (flags_.next_pending) {

// An async read is in flight, but there may be uncommitted bytes

// from a previous read that ngtcp2 didn't accept (nwrite=0 due

// to pacing/congestion). Return those bytes so the send loop can

// retry rather than blocking until the async read completes.

if (uncommitted_ > 0) {

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](int) {});

return bob::Status::STATUS_BLOCK;

}

if (flags_.errored) {

std::move(next)(UV_EBADF, nullptr, 0, [](int) {});

return UV_EBADF;

}

// If eos_ is true and there are no uncommitted bytes we'll return eos,

// otherwise, return whatever is in the uncommitted queue.

if (flags_.eos) {

if (uncommitted_ > 0) {

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](int) {});

return bob::Status::STATUS_EOS;

}

// If there are uncommitted bytes in the queue_, and there are enough to

// fill a full data packet, then pull will just return the current

// uncommitted bytes currently in the queue rather than reading more from

// the queue.

if (uncommitted_ >= kDefaultMaxPacketLength) {

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

DCHECK(queue_);

DCHECK(reader_);

// At this point, we know our reader hasn't finished yet, there might be

// uncommitted bytes but we want to go ahead and pull some more. We request

// that the pull is sync but allow for it to be async.

int ret = reader_->Pull(

[this](auto status, auto vecs, auto count, auto done) {

// The status should never be wait here.

DCHECK_NE(status, bob::Status::STATUS_WAIT);

if (status < 0) {

// If next_pending_ is true then a pull from the reader ended up

// being asynchronous, our stream is blocking waiting for the data,

// but we have an error! oh no! We need to error the stream.

if (flags_.next_pending) {

flags_.next_pending = false;

stream_->Destroy(

QuicError::ForNgtcp2Error(NGTCP2_INTERNAL_ERROR));

// We do not need to worry about calling MarkErrored in this case

// since we are immediately destroying the stream which will

// release the outbound buffer anyway.

}

return;

}

if (status == bob::Status::STATUS_EOS) {

DCHECK_EQ(count, 0);

DCHECK_NULL(vecs);

MarkEnded();

// If next_pending_ is true then a pull from the reader ended up

// being asynchronous, our stream is blocking waiting for the data.

// Here, there is no more data to read, but we will might have data

// in the uncommitted queue. We'll resume the stream so that the

// session will try to read from it again.

// We must clear next_pending_ before calling ResumeStream because

// ResumeStream can synchronously re-enter Outbound::Pull.

if (flags_.next_pending) {

flags_.next_pending = false;

stream_->session().ResumeStream(stream_->id());

}

return;

}

if (status == bob::Status::STATUS_BLOCK) {

DCHECK_EQ(count, 0);

DCHECK_NULL(vecs);

// If next_pending_ is true then a pull from the reader ended up

// being asynchronous, our stream is blocking waiting for the data.

// Here, we're still blocking! so there's nothing left for us to do!

return;

}

DCHECK_EQ(status, bob::Status::STATUS_CONTINUE);

// If the read returns bytes, those will be added to the uncommitted

// bytes in the queue.

Append(vecs, count, std::move(done));

// If next_pending_ is true, then a pull from the reader ended up

// being asynchronous, our stream is blocking waiting for the data.

// Now that we have data, let's resume the stream so the session will

// pull from it again.

// We must clear next_pending_ before calling ResumeStream because

// ResumeStream can synchronously re-enter Outbound::Pull.

if (flags_.next_pending) {

flags_.next_pending = false;

stream_->session().ResumeStream(stream_->id());

}

},

bob::OPTIONS_SYNC,

nullptr,

0,

kMaxVectorCount);

// There was an error. We'll report that immediately. We do not have

// to destroy the stream here since that will be taken care of by

// the caller.

if (ret < 0) {

MarkErrored();

std::move(next)(ret, nullptr, 0, [](int) {});

// Since we are erroring and won't be able to make use of this DataQueue

// any longer, let's free both it and the reader and put ourselves into

// an errored state. Further attempts to read from the outbound will

// result in a UV_EBADF error. The caller, however, should handle this by

// closing down the stream so that doesn't happen.

return ret;

}

if (ret == bob::Status::STATUS_EOS) {

// Here, we know we are done with the DataQueue and the Reader, but we

// might not yet have committed or acknowledged all of the queued data.

// We'll release our references to the queue_ and reader_ but everything

// else is untouched.

MarkEnded();

if (uncommitted_ > 0) {

// If the read returns eos, and there are uncommitted bytes in the

// queue, we'll set eos_ to true and return the current set of

// uncommitted bytes.

PullUncommitted(std::move(next));

return bob::STATUS_CONTINUE;

}

// If the read returns eos, and there are no uncommitted bytes in the

// queue, we'll return eos with no data.

std::move(next)(bob::Status::STATUS_EOS, nullptr, 0, [](int) {});

return bob::Status::STATUS_EOS;

}

if (ret == bob::Status::STATUS_BLOCK) {

// If the read returns blocked, and there are uncommitted bytes in the

// queue, we'll return the current set of uncommitted bytes.

if (uncommitted_ > 0) {

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

// If the read returns blocked, and there are no uncommitted bytes in the

// queue, we'll return blocked.

std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](int) {});

return bob::Status::STATUS_BLOCK;

}

// Reads here are generally expected to be synchronous. If we have a reader

// that insists on providing data asynchronously, then we'll have to block

// until the data is actually available. However, if there are uncommitted

// bytes already buffered (from a previous async read), return those now

// rather than blocking — the async callback will resume the stream when

// more data arrives.

if (ret == bob::Status::STATUS_WAIT) {

flags_.next_pending = true;

if (uncommitted_ > 0) {

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

std::move(next)(bob::Status::STATUS_BLOCK, nullptr, 0, [](int) {});

return bob::Status::STATUS_BLOCK;

}

DCHECK_EQ(ret, bob::Status::STATUS_CONTINUE);

PullUncommitted(std::move(next));

return bob::Status::STATUS_CONTINUE;

}

void MemoryInfo(MemoryTracker* tracker) const override {

tracker->TrackField("queue", queue_);

tracker->TrackField("reader", reader_);

tracker->TrackFieldWithSize("buffer", total_);

}

SET_MEMORY_INFO_NAME(Stream::Outbound)

SET_SELF_SIZE(Outbound)

private:

struct OnComplete {

bob::Done done;

explicit OnComplete(bob::Done done) : done(std::move(done)) {}

~OnComplete() { std::move(done)(0); }

};

void PullUncommitted(bob::Next<ngtcp2_vec> next) {

MaybeStackBuffer<ngtcp2_vec, 16> chunks;

chunks.AllocateSufficientStorage(count_);

auto head = commit_head_;

size_t n = 0;

while (head != nullptr && n < count_) {

// There might only be one byte here but there should never be zero.

DCHECK_LT(head->offset, head->buf.len);

chunks[n].base = head->buf.base + head->offset;

chunks[n].len = head->buf.len - head->offset;

head = head->next.get();

n++;

}

std::move(next)(bob::Status::STATUS_CONTINUE, chunks.out(), n, [](int) {});

}

void MarkErrored() {

flags_.errored = true;

head_.reset();

tail_ = nullptr;

commit_head_ = nullptr;

total_ = 0;

count_ = 0;

uncommitted_ = 0;

MarkEnded();

}

void MarkEnded() {

flags_.eos = true;

queue_.reset();

reader_.reset();

}

void Append(const DataQueue::Vec* vectors, size_t count, bob::Done done) {

if (count == 0) return;

// The done callback should only be invoked after we're done with

// all of the vectors passed in this call. To ensure of that, we

// wrap it with a shared pointer that calls done when the final

// instance is dropped.

auto on_complete = std::make_shared<OnComplete>(std::move(done));

for (size_t n = 0; n < count; n++) {

if (vectors[n].len == 0 || vectors[n].base == nullptr) continue;

auto entry = std::make_unique<Entry>(vectors[n], on_complete);

if (tail_ == nullptr) {

head_ = std::move(entry);

tail_ = head_.get();

commit_head_ = head_.get();

} else {

DCHECK_NULL(tail_->next);

tail_->next = std::move(entry);

tail_ = tail_->next.get();

if (commit_head_ == nullptr) commit_head_ = tail_;

}

count_++;

total_ += vectors[n].len;

uncommitted_ += vectors[n].len;

if (queued_ >= vectors[n].len) {

queued_ -= vectors[n].len;

} else {

queued_ = 0;

}

}

}

Stream* stream_;

std::shared_ptr<DataQueue> queue_;

std::shared_ptr<DataQueue::Reader> reader_;

struct Flags {

uint8_t errored : 1 = 0;

// True when in streaming mode (non-idempotent queue, appendable).

uint8_t streaming : 1 = 0;

// Will be set to true if the reader_ ends up providing a pull result

// asynchronously.

uint8_t next_pending : 1 = 0;

// Will be set to true once reader_ has returned eos.

uint8_t eos : 1 = 0;

};

Flags flags_;