diff --git a/src/http2_client.jl b/src/http2_client.jl
index 32df95277..155e34e46 100644
--- a/src/http2_client.jl
+++ b/src/http2_client.jl
@@ -148,6 +148,13 @@ mutable struct H2Connection
     conn_send_window::Int64
     initial_stream_send_window::Int64
     stream_send_window::Dict{UInt32,Int64}
+    # Streams whose send side is closed: the peer completed its response
+    # (END_STREAM) or reset the stream, or the request was canceled. Guarded by
+    # `state_lock` and checked by `_reserve_send_window!` so a body writer
+    # parked on flow control aborts instead of waiting for WINDOW_UPDATE
+    # frames the peer will never send (RFC 9113 §8.1 permits a server to
+    # respond before the request body completes).
+    send_closed_streams::Set{UInt32}
     peer_max_concurrent_streams::Int
     peer_max_header_list_size::Int
     peer_goaway_last_stream_id::UInt32
@@ -258,10 +265,43 @@ end
 @inline function _h2_send_window_ready_locked(conn::H2Connection, stream_id::UInt32)::Bool
     (@atomic :acquire conn.closed) && return true
     conn.conn_error !== nothing && return true
-    stream_window = get(() -> Int64(0), conn.stream_send_window, stream_id)
+    # A send-closed or unregistered stream will never receive more window
+    # credit; report ready so the waiter re-checks and aborts.
+    stream_id in conn.send_closed_streams && return true
+    haskey(conn.stream_send_window, stream_id) || return true
+    stream_window = conn.stream_send_window[stream_id]
     return stream_window > 0 && conn.conn_send_window > 0
 end
 
+"""
+    _mark_h2_send_closed!(conn, stream_id)
+
+Record that no more request DATA may be sent on `stream_id` (the peer ended or
+reset the stream, or the request was canceled) and wake any body writer parked
+in `_reserve_send_window!` waiting for flow-control credit.
+"""
+function _mark_h2_send_closed!(conn::H2Connection, stream_id::UInt32)
+    lock(conn.state_lock)
+    try
+        if haskey(conn.streams, stream_id)
+            push!(conn.send_closed_streams, stream_id)
+        end
+        notify(conn.window_condition; all=true)
+    finally
+        unlock(conn.state_lock)
+    end
+    return nothing
+end
+
+@inline function _h2_send_closed(conn::H2Connection, stream_id::UInt32)::Bool
+    lock(conn.state_lock)
+    try
+        return stream_id in conn.send_closed_streams || !haskey(conn.stream_send_window, stream_id)
+    finally
+        unlock(conn.state_lock)
+    end
+end
+
 function _wait_h2_send_window_locked!(conn::H2Connection, stream_id::UInt32, deadline_ns::Int64)::Nothing
     if deadline_ns == 0
         wait(conn.window_condition)
@@ -286,6 +326,9 @@ function _wait_h2_send_window_locked!(conn::H2Connection, stream_id::UInt32, dea
     return nothing
 end
 
+# Returns the number of DATA payload bytes the caller may send (> 0), or 0 when
+# the stream's send side closed (peer response complete, stream reset, request
+# canceled, or stream unregistered) and the caller must stop sending.
 function _reserve_send_window!(conn::H2Connection, stream_id::UInt32, wanted::Int, deadline_ns::Int64)::Int
     wanted > 0 || throw(ArgumentError("wanted send window must be > 0"))
     lock(conn.state_lock)
@@ -293,6 +336,8 @@ function _reserve_send_window!(conn::H2Connection, stream_id::UInt32, wanted::In
         while true
             (@atomic :acquire conn.closed) && throw(ProtocolError("HTTP/2 connection is closed"))
             conn.conn_error === nothing || throw(conn.conn_error::ProtocolError)
+            stream_id in conn.send_closed_streams && return 0
+            haskey(conn.stream_send_window, stream_id) || return 0
             stream_window = get(() -> Int64(0), conn.stream_send_window, stream_id)
             if stream_window <= 0 || conn.conn_send_window <= 0
                 # Both the connection-level and per-stream windows must allow
@@ -310,21 +355,24 @@ function _reserve_send_window!(conn::H2Connection, stream_id::UInt32, wanted::In
     end
 end
 
-function _write_data_frames_h2!(conn::H2Connection, stream_id::UInt32, request::Request, data::Vector{UInt8}, end_stream::Bool)
-    isempty(data) && return nothing
+# Returns `true` when all of `data` was written, `false` when the stream's send
+# side closed mid-body and the remaining payload must be abandoned.
+function _write_data_frames_h2!(conn::H2Connection, stream_id::UInt32, request::Request, data::Vector{UInt8}, end_stream::Bool)::Bool
+    isempty(data) && return true
     offset = 1
     total_len = length(data)
     while offset <= total_len
         remaining = total_len - offset + 1
         write_deadline_ns = _request_write_deadline_ns(request)
         chunk_len = _reserve_send_window!(conn, stream_id, remaining, write_deadline_ns)
+        chunk_len == 0 && return false
         chunk = Vector{UInt8}(undef, chunk_len)
         copyto!(chunk, 1, data, offset, chunk_len)
         final_chunk = (offset + chunk_len - 1) == total_len
         _write_frame_h2_threadsafe!(conn, DataFrame(stream_id, end_stream && final_chunk, chunk), write_deadline_ns)
         offset += chunk_len
     end
-    return nothing
+    return true
 end
 
 function _apply_peer_settings!(conn::H2Connection, settings::Vector{Pair{UInt16,UInt32}})
@@ -646,6 +694,7 @@ function _unregister_stream!(conn::H2Connection, stream_id::UInt32)
     try
         pop!(conn.streams, stream_id, nothing)
         pop!(conn.stream_send_window, stream_id, nothing)
+        delete!(conn.send_closed_streams, stream_id)
         notify(conn.stream_condition; all=true)
     finally
         unlock(conn.state_lock)
@@ -871,7 +920,21 @@ function _process_incoming_frame!(conn::H2Connection, frame::AbstractFrame)
         rst = frame::RSTStreamFrame
         state = _stream_state(conn, rst.stream_id)
         state === nothing && return nothing
-        _set_stream_error!(state::H2StreamState, H2StreamResetError(rst.error_code))
+        # RFC 9113 §8.1: a server MAY send RST_STREAM with NO_ERROR after a
+        # complete response to ask the client to abort the request body;
+        # "clients MUST NOT discard responses as a result". Only surface the
+        # reset as a stream error when it does not follow a complete response.
+        benign = false
+        lock((state::H2StreamState).lock)
+        try
+            benign = rst.error_code == UInt32(0) && (state::H2StreamState).stream_done
+        finally
+            unlock((state::H2StreamState).lock)
+        end
+        benign || _set_stream_error!(state::H2StreamState, H2StreamResetError(rst.error_code))
+        # Either way the peer will grant no more send window: wake any body
+        # writer parked on flow control so it stops uploading.
+        _mark_h2_send_closed!(conn, rst.stream_id)
         return nothing
     elseif frame isa WindowUpdateFrame
         update = frame::WindowUpdateFrame
@@ -900,6 +963,10 @@ function _process_incoming_frame!(conn::H2Connection, frame::AbstractFrame)
             _consume_orphan_header_fragment!(conn, headers.stream_id, headers.header_block_fragment, headers.end_headers)
             return nothing
         end
+        # A response bearing END_STREAM means the peer is done with the stream
+        # and will grant no more send window; wake any body writer parked on
+        # flow control so the roundtrip can return this early response.
+        headers.end_stream && _mark_h2_send_closed!(conn, headers.stream_id)
         _handle_stream_header_fragment!(conn, state::H2StreamState, headers.header_block_fragment, headers.end_headers, headers.end_stream)
         # Track an open header block so a missing END_HEADERS forces CONTINUATION
         # (RFC 7540 6.10) and the trailing frames still reach this stream even if
@@ -938,6 +1005,9 @@ function _process_incoming_frame!(conn::H2Connection, frame::AbstractFrame)
         data = frame::DataFrame
         state = _stream_state(conn, data.stream_id)
         state === nothing && return nothing
+        # Mark before `_handle_stream_data!`, which can block while the
+        # buffered body is drained.
+        data.end_stream && _mark_h2_send_closed!(conn, data.stream_id)
         _handle_stream_data!(state::H2StreamState, data)
         return nothing
     end
@@ -1114,6 +1184,7 @@ function _connect_h2_from_tcp!(
             Int64(65_535),
             Int64(65_535),
             Dict{UInt32,Int64}(),
+            Set{UInt32}(),
             typemax(Int),
             0,
             typemax(UInt32),
@@ -1226,8 +1297,13 @@ function Base.close(conn::H2Connection)
     return nothing
 end
 
-function _write_request_body_h2!(conn::H2Connection, stream_id::UInt32, request::Request)
-    request.body isa EmptyBody && return nothing
+# Returns `true` when the full body (and END_STREAM) was sent, `false` when the
+# upload was abandoned because the stream's send side closed first — e.g. the
+# peer answered with a complete response before reading the whole request body
+# (RFC 9113 §8.1), reset the stream, or the request was canceled. The caller
+# decides how to close out the abandoned stream.
+function _write_request_body_h2!(conn::H2Connection, stream_id::UInt32, request::Request)::Bool
+    request.body isa EmptyBody && return true
     buf = Vector{UInt8}(undef, 16 * 1024)
     pending = UInt8[]
     have_pending = false
@@ -1236,16 +1312,16 @@ function _write_request_body_h2!(conn::H2Connection, stream_id::UInt32, request:
             n = body_read!(request.body, buf)
             if n == 0
                 if have_pending
-                    _write_data_frames_h2!(conn, stream_id, request, pending, true)
-                else
-                    _write_frame_h2_threadsafe!(conn, DataFrame(stream_id, true, UInt8[]), _request_write_deadline_ns(request))
+                    return _write_data_frames_h2!(conn, stream_id, request, pending, true)
                 end
-                return nothing
+                _h2_send_closed(conn, stream_id) && return false
+                _write_frame_h2_threadsafe!(conn, DataFrame(stream_id, true, UInt8[]), _request_write_deadline_ns(request))
+                return true
             end
             current = Vector{UInt8}(undef, n)
             copyto!(current, 1, buf, 1, n)
             if have_pending
-                _write_data_frames_h2!(conn, stream_id, request, pending, false)
+                _write_data_frames_h2!(conn, stream_id, request, pending, false) || return false
             end
             pending = current
             have_pending = true
@@ -1302,6 +1378,14 @@ function _request_headers_for_h2(address::String, request::Request, secure::Bool
             push!(fields, HeaderField(lowered, normalized, false))
         end
     end
+    # HTTP/2 carries no Transfer-Encoding; DATA framing alone delimits the
+    # body. Still advertise `content-length` when the body size is known
+    # (RFC 9113 §8.1.1), matching the HTTP/1.1 client: intermediaries that
+    # translate to an HTTP/1.1 origin otherwise forward the body without a
+    # declared length, and origins commonly treat that as an empty body.
+    if request.content_length >= 0 && !(request.body isa EmptyBody) && !hasheader(request.headers, "content-length")
+        push!(fields, HeaderField("content-length", string(request.content_length), false))
+    end
     return fields
 end
 
@@ -1606,6 +1690,9 @@ function _h2_roundtrip_incoming!(
     cancel_cb = let conn = conn, stream_state = stream_state, request_ctx = request_ctx
         () -> begin
             _set_stream_error!(stream_state, _canceled_error(request_ctx))
+            # Stop a body writer parked on flow control; cancellation means no
+            # more request DATA may be sent on this stream.
+            _mark_h2_send_closed!(conn, stream_state.stream_id)
             @try_ignore _write_frame_h2_threadsafe!(conn, RSTStreamFrame(stream_state.stream_id, UInt32(0x8)))
             lock(conn.state_lock)
             try
@@ -1625,7 +1712,19 @@ function _h2_roundtrip_incoming!(
         end
         try
             end_stream = request.body isa EmptyBody
-            end_stream || _write_request_body_h2!(conn, stream_state.stream_id, request)
+            if !end_stream && !_write_request_body_h2!(conn, stream_state.stream_id, request)
+                # The upload was abandoned because the peer finished with the
+                # stream first. When that was a benign early complete response
+                # (no stream error), actively cancel our half of the stream
+                # with NO_ERROR so the peer can release its stream state; a
+                # peer-initiated reset or cancellation needs no reply.
+                send_rst = lock(stream_state.lock) do
+                    stream_state.stream_error === nothing && !stream_state.conn_errored
+                end
+                if send_rst
+                    @try_ignore _write_frame_h2_threadsafe!(conn, RSTStreamFrame(stream_state.stream_id, UInt32(0x0)))
+                end
+            end
         catch err
             _fail_h2_connection!(conn, ProtocolError("HTTP/2 write failed", err::Exception))
             rethrow()
diff --git a/test/http2_client_tests.jl b/test/http2_client_tests.jl
index e0acc75aa..1ab3604d1 100644
--- a/test/http2_client_tests.jl
+++ b/test/http2_client_tests.jl
@@ -1228,6 +1228,214 @@ end
     end
 end
 
+@testset "HTTP/2 client advertises content-length for known-size request bodies" begin
+    # RFC 9113 §8.1.1 permits `content-length` on HTTP/2 requests, and
+    # intermediaries translating to HTTP/1.1 origins need it: without one they
+    # forward the body with no declared length, and origins commonly treat
+    # that as an empty body (e.g. Django rejects such uploads outright).
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    seen_headers = Ref{Vector{HT.HeaderField}}(HT.HeaderField[])
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_encoder = HT.Encoder()
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[]))
+            _ = HT.read_frame!(reader)
+            headers_frame = _read_next_headers_frame!(reader)
+            hf = headers_frame::HT.HeadersFrame
+            seen_headers[] = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+            done = false
+            while !done
+                frame = HT.read_frame!(reader)
+                frame isa HT.DataFrame || continue
+                done = (frame::HT.DataFrame).end_stream
+            end
+            encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "200", false)])
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(hf.stream_id, true, true, encoded))
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        payload = fill(UInt8('x'), 96)
+        request = HT.Request("POST", "/sized-upload"; host = address, body = HT.BytesBody(payload), content_length = length(payload))
+        response = HT.h2_roundtrip!(h2_conn, request)
+        @test response.status == 200
+        _wait_task_h2!(server_task)
+        lengths = [field.value for field in seen_headers[] if field.name == "content-length"]
+        @test lengths == ["96"]
+    finally
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "HTTP/2 client aborts request body on early complete response" begin
+    # RFC 9113 §8.1: a server can send a complete response before the client
+    # finishes the request body (e.g. rejecting an upload at auth time) and
+    # then stops granting flow-control window. A body writer parked in
+    # `_reserve_send_window!` must wake up, abandon the upload, cancel its half
+    # of the stream with RST_STREAM(NO_ERROR), and surface the response.
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    data_bytes = Threads.Atomic{Int}(0)
+    client_rst_code = Threads.Atomic{Int}(-1)
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_encoder = HT.Encoder()
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[UInt16(0x4) => UInt32(32)]))
+            _ = HT.read_frame!(reader)
+            headers_frame = _read_next_headers_frame!(reader)
+            hf = headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+            # Complete 401 response immediately; never send WINDOW_UPDATE.
+            encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "401", false)])
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(hf.stream_id, true, true, encoded))
+            while true
+                frame = HT.read_frame!(reader)
+                if frame isa HT.DataFrame
+                    Threads.atomic_add!(data_bytes, length((frame::HT.DataFrame).data))
+                elseif frame isa HT.RSTStreamFrame
+                    client_rst_code[] = Int((frame::HT.RSTStreamFrame).error_code)
+                    break
+                end
+            end
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        payload = fill(UInt8('x'), 128)
+        request = HT.Request("POST", "/early-response"; host = address, body = HT.BytesBody(payload), content_length = length(payload))
+        response = HT.h2_roundtrip!(h2_conn, request)
+        @test response.status == 401
+        _wait_task_h2!(server_task)
+        # Only the first 32-byte window's worth of the body was sent.
+        # (Connection reusability after a stream-level event is asserted in the
+        # mid-upload reset testset below, whose server outlives the check; here
+        # the server closes its socket as soon as it sees the client RST, so
+        # the read loop may legitimately have shut the connection down.)
+        @test data_bytes[] < length(payload)
+        @test client_rst_code[] == 0
+    finally
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "HTTP/2 client keeps response when peer resets after complete response" begin
+    # RFC 9113 §8.1: RST_STREAM with NO_ERROR after a complete response asks
+    # the client to abort the request body, and "clients MUST NOT discard
+    # responses as a result".
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_encoder = HT.Encoder()
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[UInt16(0x4) => UInt32(32)]))
+            _ = HT.read_frame!(reader)
+            headers_frame = _read_next_headers_frame!(reader)
+            hf = headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+            encoded = HT.encode_header_block(server_encoder, HT.HeaderField[HT.HeaderField(":status", "401", false)])
+            _write_frame_to_conn!(accepted_conn, HT.HeadersFrame(hf.stream_id, true, true, encoded))
+            _write_frame_to_conn!(accepted_conn, HT.RSTStreamFrame(hf.stream_id, UInt32(0x0)))
+            while true
+                frame = HT.read_frame!(reader)
+                frame isa HT.RSTStreamFrame && break
+            end
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        payload = fill(UInt8('x'), 128)
+        request = HT.Request("POST", "/early-response-rst"; host = address, body = HT.BytesBody(payload), content_length = length(payload))
+        response = HT.h2_roundtrip!(h2_conn, request)
+        @test response.status == 401
+        _wait_task_h2!(server_task)
+    finally
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
+@testset "HTTP/2 client surfaces mid-upload stream reset" begin
+    # A reset arriving while the body writer is parked on flow control must
+    # wake it and raise the stream error instead of stalling until a timeout.
+    listener = ND.listen("tcp", "127.0.0.1:0"; backlog = 8)
+    laddr = NC.addr(listener)::NC.SocketAddrV4
+    address = ND.join_host_port("127.0.0.1", Int(laddr.port))
+    server_task = errormonitor(Threads.@spawn begin
+        accepted_conn = NC.accept(listener)
+        reader = HT._ConnReader(accepted_conn)
+        server_decoder = HT.Decoder()
+        try
+            _ = _read_exact_h2_tcp!(accepted_conn, length(HT._H2_PREFACE))
+            _ = HT.read_frame!(reader)
+            _write_frame_to_conn!(accepted_conn, HT.SettingsFrame(false, Pair{UInt16, UInt32}[UInt16(0x4) => UInt32(32)]))
+            _ = HT.read_frame!(reader)
+            headers_frame = _read_next_headers_frame!(reader)
+            hf = headers_frame::HT.HeadersFrame
+            _ = HT.decode_header_block(server_decoder, hf.header_block_fragment)
+            # Wait for the first DATA frame, then cancel the stream without a
+            # response and without ever granting more window.
+            while true
+                frame = HT.read_frame!(reader)
+                frame isa HT.DataFrame && break
+            end
+            _write_frame_to_conn!(accepted_conn, HT.RSTStreamFrame(hf.stream_id, UInt32(0x8)))
+            sleep(0.20)
+        finally
+            HTTP.@try_ignore NC.close(accepted_conn)
+        end
+        return nothing
+    end)
+    h2_conn = HT.connect_h2!(address; secure = false)
+    try
+        payload = fill(UInt8('x'), 128)
+        request = HT.Request("POST", "/mid-upload-rst"; host = address, body = HT.BytesBody(payload), content_length = length(payload))
+        err = try
+            HT.h2_roundtrip!(h2_conn, request)
+            nothing
+        catch e
+            e
+        end
+        @test err isa HT.H2StreamResetError
+        @test (err::HT.H2StreamResetError).error_code == UInt32(0x8)
+        # A stream-level reset must not poison the shared connection.
+        @test HT._h2_conn_reusable(h2_conn)
+        _wait_task_h2!(server_task)
+    finally
+        close(h2_conn)
+        HTTP.@try_ignore NC.close(listener)
+    end
+end
+
 @testset "HTTP/2 client connection window stays non-negative under a raised stream window" begin
     # A raised SETTINGS_INITIAL_WINDOW_SIZE only enlarges the per-stream send
     # window; the connection-level window is separate and stays at the protocol
@@ -1883,6 +2091,7 @@ function _build_bare_h2_connection()
         Int64(65_535),
         Int64(65_535),
         Dict{UInt32,Int64}(),
+        Set{UInt32}(),
         typemax(Int),
         0,
         typemax(UInt32),