FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

quic: fix stop sending behaviour & callback · nodejs/node@7de3d09 · GitHub

/ node Public

Commit 7de3d09

Browse files
pimterry
authored and
aduh95
committed
quic: fix stop sending behaviour & callback
Signed-off-by: Tim Perry <pimterry@gmail.com> PR-URL: #64710 Reviewed-By: James M Snell <jasnell@gmail.com>
1 parent 2632d60 commit 7de3d09

15 files changed

Lines changed: 193 additions & 164 deletions

‎doc/api/quic.md‎

Lines changed: 21 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2031,8 +2031,7 @@ added: v23.8.0
20312031

20322032
The callback to invoke when the peer aborts a direction of the stream by
20332033
sending a `RESET_STREAM` frame (the peer abandons their writable side, so
2034-
no further data will arrive on our readable side) or a `STOP_SENDING`
2035-
frame (the peer asks us to stop writing on our writable side).
2034+
no further data will arrive on our readable side).
20362035

20372036
The callback receives a Node.js error whose `errorCode` (`bigint`)
20382037
property carries the application error code from the wire frame.
@@ -2043,6 +2042,21 @@ continue using the still-active direction on a bidirectional stream),
20432042
abort the other direction with [`writer.fail()`][], or tear down the
20442043
whole stream with [`stream.destroy()`][]. Read/write.
20452044

2045+
### `stream.onstopsending`
2046+
2047+
<!-- YAML
2048+
added: REPLACEME
2049+
-->
2050+
2051+
* Type: {quic.OnStreamErrorCallback}
2052+
2053+
The callback to invoke when the peer aborts a direction of the stream by
2054+
sending a `STOP_SENDING` frame (the peer asks us to stop writing on our
2055+
writable side).
2056+
2057+
The callback receives a Node.js error whose `errorCode` (`bigint`)
2058+
property carries the application error code from the wire frame. Read/write.
2059+
20462060
### `stream.headers`
20472061

20482062
<!-- YAML
@@ -3587,8 +3601,8 @@ functions. If a callback throws synchronously or returns a promise that
35873601
rejects, the error is caught and the owning session or stream is destroyed
35883602
with that error:
35893603

3590-
* Stream callbacks (`onblocked`, `onreset`, `onheaders`, `ontrailers`,
3591-
`oninfo`, `onwanttrailers`): the stream is destroyed.
3604+
* Stream callbacks (`onblocked`, `onreset`, `onstopsending`, `onheaders`,
3605+
`ontrailers`, `oninfo`, `onwanttrailers`): the stream is destroyed.
35923606
* Session callbacks (`onapplication`, `onstream`, `ondatagram`,
35933607
`ondatagramstatus`, `onpathvalidation`, `onsessionticket`,
35943608
`onnewtoken`, `onversionnegotiation`, `onorigin`, `ongoaway`,
@@ -4457,10 +4471,9 @@ added: v26.2.0
44574471
* `session` {quic.QuicSession}
44584472
* `error` {any} The QUIC error associated with the reset.
44594473
4460-
Published when a stream receives a STOP\_SENDING or RESET\_STREAM frame
4461-
from the peer, indicating the peer has aborted the stream. This is a
4462-
key signal for diagnosing application-level issues such as cancelled
4463-
requests.
4474+
Published when a stream receives a RESET\_STREAM frame from the peer,
4475+
indicating the peer has aborted its sending direction. This is a key signal
4476+
for diagnosing application-level issues such as cancelled requests.
44644477
44654478
### Channel: `quic.stream.blocked`
44664479

‎lib/internal/quic/quic.js‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ const {
1212
DataViewPrototypeGetByteLength,
1313
ErrorCaptureStackTrace,
1414
FunctionPrototypeBind,
15+
FunctionPrototypeCall,
1516
Number,
1617
ObjectDefineProperties,
1718
ObjectKeys,
@@ -210,6 +211,7 @@ const {
210211
kSendHeaders,
211212
kSessionApplication,
212213
kSessionTicket,
214+
kStopSending,
213215
kTrailers,
214216
kVersionNegotiation,
215217
kInspect,
@@ -991,6 +993,14 @@ setCallbacks({
991993
this[kOwner][kReset](error);
992994
},
993995

996+
onStreamStopSending(error) {
997+
if (error !== undefined) {
998+
error = convertQuicError(error);
999+
}
1000+
debug('stream stop sending callback', this[kOwner], error);
1001+
this[kOwner][kStopSending](error);
1002+
},
1003+
9941004
onStreamHeaders(headers, kind) {
9951005
// Called when the stream C++ handle has received a full block of headers.
9961006
debug(`stream ${this[kOwner].id} headers callback`, headers, kind);
@@ -1576,6 +1586,7 @@ class QuicStream {
15761586
onerror: undefined,
15771587
onblocked: undefined,
15781588
onreset: undefined,
1589+
onstopsending: undefined,
15791590
onheaders: undefined,
15801591
ontrailers: undefined,
15811592
oninfo: undefined,
@@ -1781,6 +1792,25 @@ class QuicStream {
17811792
}
17821793
}
17831794

1795+
/** @type {OnStreamErrorCallback} */
1796+
get onstopsending() {
1797+
assertIsQuicStream(this);
1798+
return this.#inner.onstopsending;
1799+
}
1800+
1801+
set onstopsending(fn) {
1802+
assertIsQuicStream(this);
1803+
const inner = this.#inner;
1804+
if (fn === undefined) {
1805+
inner.onstopsending = undefined;
1806+
inner.state.wantsStopSending = false;
1807+
} else {
1808+
validateFunction(fn, 'onstopsending');
1809+
inner.onstopsending = FunctionPrototypeBind(fn, this);
1810+
inner.state.wantsStopSending = true;
1811+
}
1812+
}
1813+
17841814
/** @type {OnHeadersCallback} */
17851815
get onheaders() {
17861816
assertIsQuicStream(this);
@@ -2143,6 +2173,19 @@ class QuicStream {
21432173
}
21442174
};
21452175

2176+
const onStopSending = stream[kStopSending];
2177+
stream[kStopSending] = (reason) => {
2178+
if (!closed && !errored) {
2179+
errored = true;
2180+
error = reason;
2181+
if (drainWakeup != null) {
2182+
drainWakeup.reject(error);
2183+
drainWakeup = null;
2184+
}
2185+
}
2186+
FunctionPrototypeCall(onStopSending, stream, reason);
2187+
};
2188+
21462189
// A note on backpressure handling: per the stream/iter spec, the default
21472190
// backpressure policy for writers is strict, meaning that if the stream
21482191
// signals backpressure additional writes are rejected until the buffer has
@@ -2543,6 +2586,7 @@ class QuicStream {
25432586
inner.pendingClose.resolve = undefined;
25442587
inner.onblocked = undefined;
25452588
inner.onreset = undefined;
2589+
inner.onstopsending = undefined;
25462590
inner.onheaders = undefined;
25472591
inner.onerror = undefined;
25482592
inner.ontrailers = undefined;
@@ -2596,6 +2640,12 @@ class QuicStream {
25962640
safeCallbackInvoke(inner.onreset, this, error);
25972641
}
25982642

2643+
[kStopSending](error) {
2644+
const inner = this.#inner;
2645+
assert(inner.onstopsending, 'Unexpected stop sending event');
2646+
safeCallbackInvoke(inner.onstopsending, this, error);
2647+
}
2648+
25992649
[kHeaders](headers, kind) {
26002650
const block = parseHeaderPairs(headers);
26012651
const kindName = kHeadersKindName[kind] ?? kind;

‎lib/internal/quic/state.js‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@ const {
101101
IDX_STATE_STREAM_WANTS_BLOCK,
102102
IDX_STATE_STREAM_WANTS_HEADERS,
103103
IDX_STATE_STREAM_WANTS_RESET,
104+
IDX_STATE_STREAM_WANTS_STOP_SENDING,
104105
IDX_STATE_STREAM_WANTS_TRAILERS,
105106
IDX_STATE_STREAM_RECEIVED_EARLY_DATA,
106107
IDX_STATE_STREAM_WRITE_DESIRED_SIZE,
@@ -142,6 +143,7 @@ assert(IDX_STATE_STREAM_HAS_READER !== undefined);
142143
assert(IDX_STATE_STREAM_WANTS_BLOCK !== undefined);
143144
assert(IDX_STATE_STREAM_WANTS_HEADERS !== undefined);
144145
assert(IDX_STATE_STREAM_WANTS_RESET !== undefined);
146+
assert(IDX_STATE_STREAM_WANTS_STOP_SENDING !== undefined);
145147
assert(IDX_STATE_STREAM_WANTS_TRAILERS !== undefined);
146148
assert(IDX_STATE_STREAM_WRITE_DESIRED_SIZE !== undefined);
147149
assert(IDX_STATE_STREAM_RESET_CODE !== undefined);
@@ -826,6 +828,24 @@ class QuicStreamState {
826828
DataViewPrototypeSetUint8(handle, this.#offset + IDX_STATE_STREAM_WANTS_RESET, val ? 1 : 0);
827829
}
828830

831+
/** @type {boolean} */
832+
get wantsStopSending() {
833+
const handle = this.#handle;
834+
if (handle === undefined) return undefined;
835+
return DataViewPrototypeGetUint8(
836+
handle, this.#offset + IDX_STATE_STREAM_WANTS_STOP_SENDING) !== 0;
837+
}
838+
839+
/** @type {boolean} */
840+
set wantsStopSending(val) {
841+
const handle = this.#handle;
842+
if (handle === undefined) return;
843+
DataViewPrototypeSetUint8(
844+
handle,
845+
this.#offset + IDX_STATE_STREAM_WANTS_STOP_SENDING,
846+
val ? 1 : 0);
847+
}
848+
829849
/** @type {boolean} */
830850
get wantsTrailers() {
831851
const handle = this.#handle;
@@ -903,6 +923,7 @@ class QuicStreamState {
903923
hasReader,
904924
wantsBlock,
905925
wantsReset,
926+
wantsStopSending,
906927
wantsHeaders,
907928
wantsTrailers,
908929
early,
@@ -923,6 +944,7 @@ class QuicStreamState {
923944
hasReader,
924945
wantsBlock,
925946
wantsReset,
947+
wantsStopSending,
926948
wantsHeaders,
927949
wantsTrailers,
928950
early,
@@ -960,6 +982,7 @@ class QuicStreamState {
960982
hasReader,
961983
wantsBlock,
962984
wantsReset,
985+
wantsStopSending,
963986
wantsHeaders,
964987
wantsTrailers,
965988
early,
@@ -980,6 +1003,7 @@ class QuicStreamState {
9801003
hasReader,
9811004
wantsBlock,
9821005
wantsReset,
1006+
wantsStopSending,
9831007
wantsHeaders,
9841008
wantsTrailers,
9851009
early,

‎lib/internal/quic/symbols.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ const kReset = Symbol('kReset');
5757
const kSendHeaders = Symbol('kSendHeaders');
5858
const kSessionApplication = Symbol('kSessionApplication');
5959
const kSessionTicket = Symbol('kSessionTicket');
60+
const kStopSending = Symbol('kStopSending');
6061
const kTrailers = Symbol('kTrailers');
6162
const kVersionNegotiation = Symbol('kVersionNegotiation');
6263

@@ -93,6 +94,7 @@ module.exports = {
9394
kSendHeaders,
9495
kSessionApplication,
9596
kSessionTicket,
97+
kStopSending,
9698
kTrailers,
9799
kVersionNegotiation,
98100
};

‎src/quic/bindingdata.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ class SessionManager;
6161
V(stream_drain, StreamDrain) \
6262
V(stream_headers, StreamHeaders) \
6363
V(stream_reset, StreamReset) \
64+
V(stream_stop_sending, StreamStopSending) \
6465
V(stream_trailers, StreamTrailers)
6566

6667
// The various JS strings the implementation uses.

‎src/quic/session.cc‎

Lines changed: 11 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1611,11 +1611,11 @@ struct Session::Impl final : public MemoryRetainer {
16111611
return NGTCP2_SUCCESS;
16121612
}
16131613

1614-
static int on_stream_stop_sending(ngtcp2_conn* conn,
1615-
stream_id stream_id,
1616-
error_code app_error_code,
1617-
void* user_data,
1618-
void* stream_user_data) {
1614+
static int on_receive_stream_stop_sending(ngtcp2_conn* conn,
1615+
stream_id stream_id,
1616+
error_code app_error_code,
1617+
void* user_data,
1618+
void* stream_user_data) {
16191619
NGTCP2_CALLBACK_SCOPE(session)
16201620
auto* stream = Stream::From(stream_user_data);
16211621
if (stream == nullptr) return NGTCP2_SUCCESS;
@@ -1652,7 +1652,7 @@ struct Session::Impl final : public MemoryRetainer {
16521652

16531653
static constexpr ngtcp2_callbacks CLIENT = {
16541654
ngtcp2_crypto_client_initial_cb,
1655-
nullptr,
1655+
nullptr, // stream_stop_sending
16561656
ngtcp2_crypto_recv_crypto_data_cb,
16571657
on_handshake_completed,
16581658
on_receive_version_negotiation,
@@ -1686,7 +1686,7 @@ struct Session::Impl final : public MemoryRetainer {
16861686
on_acknowledge_datagram,
16871687
on_lost_datagram,
16881688
nullptr, // get_path_challenge_data (deprecated, use v2 below)
1689-
on_stream_stop_sending,
1689+
nullptr, // stream_stop_sending
16901690
ngtcp2_crypto_version_negotiation_cb,
16911691
on_receive_rx_key,
16921692
on_receive_tx_key,
@@ -1697,12 +1697,12 @@ struct Session::Impl final : public MemoryRetainer {
16971697
on_cid_status,
16981698
ngtcp2_crypto_get_path_challenge_data2_cb,
16991699
#ifdef NGTCP2_CALLBACKS_V4
1700-
nullptr,
1700+
on_receive_stream_stop_sending,
17011701
#endif
17021702
};
17031703

17041704
static constexpr ngtcp2_callbacks SERVER = {
1705-
nullptr,
1705+
nullptr, // stream_stop_sending
17061706
ngtcp2_crypto_recv_client_initial_cb,
17071707
ngtcp2_crypto_recv_crypto_data_cb,
17081708
on_handshake_completed,
@@ -1737,7 +1737,7 @@ struct Session::Impl final : public MemoryRetainer {
17371737
on_acknowledge_datagram,
17381738
on_lost_datagram,
17391739
nullptr, // get_path_challenge_data (deprecated, use v2 below)
1740-
on_stream_stop_sending,
1740+
nullptr, // stream_stop_sending
17411741
ngtcp2_crypto_version_negotiation_cb,
17421742
nullptr,
17431743
on_receive_tx_key,
@@ -1748,7 +1748,7 @@ struct Session::Impl final : public MemoryRetainer {
17481748
on_cid_status,
17491749
ngtcp2_crypto_get_path_challenge_data2_cb,
17501750
#ifdef NGTCP2_CALLBACKS_V4
1751-
nullptr,
1751+
on_receive_stream_stop_sending,
17521752
#endif
17531753
};
17541754
};

‎src/quic/streams.cc‎

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,8 @@ namespace quic {
6161
V(WANTS_HEADERS, wants_headers, uint8_t) \
6262
/* Set when the stream has a reset event handler */ \
6363
V(WANTS_RESET, wants_reset, uint8_t) \
64+
/* Set when the stream has a stop sending event handler */ \
65+
V(WANTS_STOP_SENDING, wants_stop_sending, uint8_t) \
6466
/* Set when the stream has a trailers event handler */ \
6567
V(WANTS_TRAILERS, wants_trailers, uint8_t) \
6668
/* True when 0-RTT early data was received */ \
@@ -1774,19 +1776,11 @@ void Stream::ReceiveData(const uint8_t* data,
17741776
}
17751777

17761778
void Stream::ReceiveStopSending(QuicError error) {
1777-
// STOP_SENDING from the peer asks us to stop sending. Per RFC 9000
1778-
// §3.5 the receiver SHOULD respond with RESET_STREAM, which is what
1779-
// ngtcp2_conn_shutdown_stream_write below schedules. If our
1780-
// writable side has already been shut down (e.g. we already sent
1781-
// RESET_STREAM ourselves or finished sending with FIN) there is
1782-
// nothing more to do here. The previous guard checked
1783-
// `state()->read_ended` which is unrelated to the writable side and
1784-
// suppressed STOP_SENDING handling whenever a sibling RESET_STREAM
1785-
// frame had been processed first within the same packet.
1786-
if (state()->write_ended) return;
1779+
// STOP_SENDING from the peer asks us to stop sending. The required
1780+
// RESET_STREAM response is scheduled automatically.
17871781
Debug(this, "Received stop sending with error %s", error);
1788-
ngtcp2_conn_shutdown_stream_write(session(), 0, id(), error.code());
17891782
EndWritable();
1783+
EmitStopSending(error);
17901784
}
17911785

17921786
void Stream::ReceiveStreamReset(uint64_t final_size, QuicError error) {
@@ -1958,6 +1952,17 @@ void Stream::EmitReset(const QuicError& error) {
19581952
MakeCallback(BindingData::Get(env()).stream_reset_callback(), 1, &err);
19591953
}
19601954

1955+
void Stream::EmitStopSending(const QuicError& error) {
1956+
if (!env()->can_call_into_js() || !state()->wants_stop_sending) {
1957+
return;
1958+
}
1959+
CallbackScope<Stream> cb_scope(this);
1960+
Local<Value> err;
1961+
if (!error.ToV8Value(env()).ToLocal(&err)) return;
1962+
1963+
MakeCallback(BindingData::Get(env()).stream_stop_sending_callback(), 1, &err);
1964+
}
1965+
19611966
void Stream::EmitWantTrailers() {
19621967
// state()->wants_trailers will be set from the javascript side if the
19631968
// stream object has a handler for the trailers event.

‎src/quic/streams.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -417,6 +417,9 @@ class Stream final : public AsyncWrap,
417417
// Notifies the JavaScript side that the stream has been reset.
418418
void EmitReset(const QuicError& error);
419419

420+
// Notifies the JavaScript side that the peer asked it to stop sending.
421+
void EmitStopSending(const QuicError& error);
422+
420423
// Notifies the JavaScript side that the application is ready to receive
421424
// trailing headers. Any trailing headers must be sent immediately, and
422425
// synchronously when this callback is triggered.

‎test/parallel/test-quic-internal-endpoint-stats-state.mjs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,7 @@ assert.strictEqual(streamState.reset, false);
156156
assert.strictEqual(streamState.hasReader, false);
157157
assert.strictEqual(streamState.wantsBlock, false);
158158
assert.strictEqual(streamState.wantsReset, false);
159+
assert.strictEqual(streamState.wantsStopSending, false);
159160

160161
assert.strictEqual(sessionState.hasPathValidationListener, false);
161162
assert.strictEqual(sessionState.hasDatagramListener, false);

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL