quic/core/http/quic_spdy_stream.h

// Copyright 2013 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

// The base class for streams which deliver data to/from an application.
// In each direction, the data on such a stream first contains compressed
// headers then body data.

#ifndef QUICHE_QUIC_CORE_HTTP_QUIC_SPDY_STREAM_H_
#define QUICHE_QUIC_CORE_HTTP_QUIC_SPDY_STREAM_H_

#include <sys/types.h>

#include <cstddef>
#include <list>
#include <memory>
#include <string>

#include "absl/base/attributes.h"
#include "absl/strings/string_view.h"
#include "absl/types/span.h"
#include "quic/core/http/capsule.h"
#include "quic/core/http/http_decoder.h"
#include "quic/core/http/http_encoder.h"
#include "quic/core/http/quic_header_list.h"
#include "quic/core/http/quic_spdy_stream_body_manager.h"
#include "quic/core/http/web_transport_stream_adapter.h"
#include "quic/core/qpack/qpack_decoded_headers_accumulator.h"
#include "quic/core/quic_error_codes.h"
#include "quic/core/quic_packets.h"
#include "quic/core/quic_stream.h"
#include "quic/core/quic_stream_sequencer.h"
#include "quic/core/quic_types.h"
#include "quic/core/web_transport_interface.h"
#include "quic/platform/api/quic_export.h"
#include "quic/platform/api/quic_flags.h"
#include "quic/platform/api/quic_socket_address.h"
#include "spdy/core/spdy_framer.h"
#include "spdy/core/spdy_header_block.h"

namespace quic {

namespace test {
class QuicSpdyStreamPeer;
class QuicStreamPeer;
}  // namespace test

class QuicSpdySession;
class WebTransportHttp3;

// A QUIC stream that can send and receive HTTP2 (SPDY) headers.
class QUIC_EXPORT_PRIVATE QuicSpdyStream
    : public QuicStream,
      public CapsuleParser::Visitor,
      public QpackDecodedHeadersAccumulator::Visitor {
 public:
  // Visitor receives callbacks from the stream.
  class QUIC_EXPORT_PRIVATE Visitor {
   public:
    Visitor() {}
    Visitor(const Visitor&) = delete;
    Visitor& operator=(const Visitor&) = delete;

    // Called when the stream is closed.
    virtual void OnClose(QuicSpdyStream* stream) = 0;

    // Allows subclasses to override and do work.
    virtual void OnPromiseHeadersComplete(QuicStreamId /*promised_id*/,
                                          size_t /*frame_len*/) {}

   protected:
    virtual ~Visitor() {}
  };

  QuicSpdyStream(QuicStreamId id, QuicSpdySession* spdy_session,
                 StreamType type);
  QuicSpdyStream(PendingStream* pending, QuicSpdySession* spdy_session);
  QuicSpdyStream(const QuicSpdyStream&) = delete;
  QuicSpdyStream& operator=(const QuicSpdyStream&) = delete;
  ~QuicSpdyStream() override;

  // QuicStream implementation
  void OnClose() override;

  // Override to maybe close the write side after writing.
  void OnCanWrite() override;

  // Called by the session when headers with a priority have been received
  // for this stream.  This method will only be called for server streams.
  virtual void OnStreamHeadersPriority(
      const spdy::SpdyStreamPrecedence& precedence);

  // Called by the session when decompressed headers have been completely
  // delivered to this stream.  If |fin| is true, then this stream
  // should be closed; no more data will be sent by the peer.
  virtual void OnStreamHeaderList(bool fin, size_t frame_len,
                                  const QuicHeaderList& header_list);

  // Called by the session when decompressed push promise headers have
  // been completely delivered to this stream.
  virtual void OnPromiseHeaderList(QuicStreamId promised_id, size_t frame_len,
                                   const QuicHeaderList& header_list);

  // Called by the session when a PRIORITY frame has been been received for this
  // stream. This method will only be called for server streams.
  void OnPriorityFrame(const spdy::SpdyStreamPrecedence& precedence);

  // Override the base class to not discard response when receiving
  // QUIC_STREAM_NO_ERROR.
  void OnStreamReset(const QuicRstStreamFrame& frame) override;
  void ResetWithError(QuicResetStreamError error) override;
  bool OnStopSending(QuicResetStreamError error) override;

  // Called by the sequencer when new data is available. Decodes the data and
  // calls OnBodyAvailable() to pass to the upper layer.
  void OnDataAvailable() override;

  // Called in OnDataAvailable() after it finishes the decoding job.
  virtual void OnBodyAvailable() = 0;

  // Writes the headers contained in |header_block| on the dedicated headers
  // stream or on this stream, depending on VersionUsesHttp3().  Returns the
  // number of bytes sent, including data sent on the encoder stream when using
  // QPACK.
  virtual size_t WriteHeaders(
      spdy::SpdyHeaderBlock header_block, bool fin,
      QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener);

  // Sends |data| to the peer, or buffers if it can't be sent immediately.
  void WriteOrBufferBody(absl::string_view data, bool fin);

  // Writes the trailers contained in |trailer_block| on the dedicated headers
  // stream or on this stream, depending on VersionUsesHttp3().  Trailers will
  // always have the FIN flag set.  Returns the number of bytes sent, including
  // data sent on the encoder stream when using QPACK.
  virtual size_t WriteTrailers(
      spdy::SpdyHeaderBlock trailer_block,
      QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener);

  // Override to report newly acked bytes via ack_listener_.
  bool OnStreamFrameAcked(QuicStreamOffset offset, QuicByteCount data_length,
                          bool fin_acked, QuicTime::Delta ack_delay_time,
                          QuicTime receive_timestamp,
                          QuicByteCount* newly_acked_length) override;

  // Override to report bytes retransmitted via ack_listener_.
  void OnStreamFrameRetransmitted(QuicStreamOffset offset,
                                  QuicByteCount data_length,
                                  bool fin_retransmitted) override;

  // Does the same thing as WriteOrBufferBody except this method takes iovec
  // as the data input. Right now it only calls WritevData.
  QuicConsumedData WritevBody(const struct iovec* iov, int count, bool fin);

  // Does the same thing as WriteOrBufferBody except this method takes
  // memslicespan as the data input. Right now it only calls WriteMemSlices.
  QuicConsumedData WriteBodySlices(absl::Span<QuicMemSlice> slices, bool fin);

  // Marks the trailers as consumed. This applies to the case where this object
  // receives headers and trailers as QuicHeaderLists via calls to
  // OnStreamHeaderList(). Trailer data will be consumed from the sequencer only
  // once all body data has been consumed.
  void MarkTrailersConsumed();

  // Clears |header_list_|.
  void ConsumeHeaderList();

  // This block of functions wraps the sequencer's functions of the same
  // name.  These methods return uncompressed data until that has
  // been fully processed.  Then they simply delegate to the sequencer.
  virtual size_t Readv(const struct iovec* iov, size_t iov_len);
  virtual int GetReadableRegions(iovec* iov, size_t iov_len) const;
  void MarkConsumed(size_t num_bytes);

  // Returns true if header contains a valid 3-digit status and parse the status
  // code to |status_code|.
  static bool ParseHeaderStatusCode(const spdy::SpdyHeaderBlock& header,
                                    int* status_code);

  // Returns true when all data from the peer has been read and consumed,
  // including the fin.
  bool IsDoneReading() const;
  bool HasBytesToRead() const;

  void set_visitor(Visitor* visitor) { visitor_ = visitor; }

  bool headers_decompressed() const { return headers_decompressed_; }

  // Returns total amount of body bytes that have been read.
  uint64_t total_body_bytes_read() const;

  const QuicHeaderList& header_list() const { return header_list_; }

  bool trailers_decompressed() const { return trailers_decompressed_; }

  // Returns whatever trailers have been received for this stream.
  const spdy::SpdyHeaderBlock& received_trailers() const {
    return received_trailers_;
  }

  // Returns true if headers have been fully read and consumed.
  bool FinishedReadingHeaders() const;

  // Returns true if FIN has been received and either trailers have been fully
  // read and consumed or there are no trailers.
  bool FinishedReadingTrailers() const;

  // Returns true if the sequencer has delivered the FIN, and no more body bytes
  // will be available.
  bool IsSequencerClosed() { return sequencer()->IsClosed(); }

  // QpackDecodedHeadersAccumulator::Visitor implementation.
  void OnHeadersDecoded(QuicHeaderList headers,
                        bool header_list_size_limit_exceeded) override;
  void OnHeaderDecodingError(QuicErrorCode error_code,
                             absl::string_view error_message) override;

  QuicSpdySession* spdy_session() const { return spdy_session_; }

  // Send PRIORITY_UPDATE frame and update |last_sent_urgency_| if
  // |last_sent_urgency_| is different from current priority.
  void MaybeSendPriorityUpdateFrame() override;

  // Returns the WebTransport session owned by this stream, if one exists.
  WebTransportHttp3* web_transport() { return web_transport_.get(); }

  // Returns the WebTransport data stream associated with this QUIC stream, or
  // null if this is not a WebTransport data stream.
  WebTransportStream* web_transport_stream() {
    if (web_transport_data_ == nullptr) {
      return nullptr;
    }
    return &web_transport_data_->adapter;
  }

  // Sends a WEBTRANSPORT_STREAM frame and sets up the appropriate metadata.
  void ConvertToWebTransportDataStream(WebTransportSessionId session_id);

  void OnCanWriteNewData() override;

  // If this stream is a WebTransport data stream, closes the connection with an
  // error, and returns false.
  bool AssertNotWebTransportDataStream(absl::string_view operation);

  // Indicates whether a call to WriteBodySlices will be successful and not
  // rejected due to buffer being full.  |write_size| must be non-zero.
  bool CanWriteNewBodyData(QuicByteCount write_size) const;

  // From CapsuleParser::Visitor.
  bool OnCapsule(const Capsule& capsule) override;
  void OnCapsuleParseFailure(const std::string& error_message) override;

  // Sends an HTTP/3 datagram. The stream and context IDs are not part of
  // |payload|.
  MessageStatus SendHttp3Datagram(
      absl::optional<QuicDatagramContextId> context_id,
      absl::string_view payload);

  class QUIC_EXPORT_PRIVATE Http3DatagramVisitor {
   public:
    virtual ~Http3DatagramVisitor() {}

    // Called when an HTTP/3 datagram is received. |payload| does not contain
    // the stream or context IDs. Note that this contains the stream ID even if
    // flow IDs from draft-ietf-masque-h3-datagram-00 are in use.
    virtual void OnHttp3Datagram(
        QuicStreamId stream_id,
        absl::optional<QuicDatagramContextId> context_id,
        absl::string_view payload) = 0;
  };

  class QUIC_EXPORT_PRIVATE Http3DatagramRegistrationVisitor {
   public:
    virtual ~Http3DatagramRegistrationVisitor() {}

    // Called when a REGISTER_DATAGRAM_CONTEXT or REGISTER_DATAGRAM_NO_CONTEXT
    // capsule is received. Note that this contains the stream ID even if flow
    // IDs from draft-ietf-masque-h3-datagram-00 are in use.
    virtual void OnContextReceived(
        QuicStreamId stream_id,
        absl::optional<QuicDatagramContextId> context_id,
        DatagramFormatType format_type,
        absl::string_view format_additional_data) = 0;

    // Called when a CLOSE_DATAGRAM_CONTEXT capsule is received. Note that this
    // contains the stream ID even if flow IDs from
    // draft-ietf-masque-h3-datagram-00 are in use.
    virtual void OnContextClosed(
        QuicStreamId stream_id,
        absl::optional<QuicDatagramContextId> context_id,
        ContextCloseCode close_code, absl::string_view close_details) = 0;
  };

  // Registers |visitor| to receive HTTP/3 datagram context registrations. This
  // must not be called without first calling
  // UnregisterHttp3DatagramRegistrationVisitor. |visitor| must be valid until a
  // corresponding call to UnregisterHttp3DatagramRegistrationVisitor.
  void RegisterHttp3DatagramRegistrationVisitor(
      Http3DatagramRegistrationVisitor* visitor);

  // Unregisters for HTTP/3 datagram context registrations. Must not be called
  // unless previously registered.
  void UnregisterHttp3DatagramRegistrationVisitor();

  // Moves an HTTP/3 datagram registration to a different visitor. Mainly meant
  // to be used by the visitors' move operators.
  void MoveHttp3DatagramRegistration(Http3DatagramRegistrationVisitor* visitor);

  // Registers |visitor| to receive HTTP/3 datagrams for optional context ID
  // |context_id|. This must not be called on a previously registered context ID
  // without first calling UnregisterHttp3DatagramContextId. |visitor| must be
  // valid until a corresponding call to UnregisterHttp3DatagramContextId. If
  // this method is called multiple times, the context ID MUST either be always
  // present, or always absent.
  void RegisterHttp3DatagramContextId(
      absl::optional<QuicDatagramContextId> context_id,
      DatagramFormatType format_type, absl::string_view format_additional_data,
      Http3DatagramVisitor* visitor);

  // Unregisters an HTTP/3 datagram context ID. Must be called on a previously
  // registered context.
  void UnregisterHttp3DatagramContextId(
      absl::optional<QuicDatagramContextId> context_id);

  // Moves an HTTP/3 datagram context ID to a different visitor. Mainly meant
  // to be used by the visitors' move operators.
  void MoveHttp3DatagramContextIdRegistration(
      absl::optional<QuicDatagramContextId> context_id,
      Http3DatagramVisitor* visitor);

  // Sets max datagram time in queue.
  void SetMaxDatagramTimeInQueue(QuicTime::Delta max_time_in_queue);

  // Generates a new HTTP/3 datagram context ID for this stream. A datagram
  // registration visitor must be currently registered on this stream.
  QuicDatagramContextId GetNextDatagramContextId();

  void OnDatagramReceived(QuicDataReader* reader);

  void RegisterHttp3DatagramFlowId(QuicDatagramStreamId flow_id);

  QuicByteCount GetMaxDatagramSize(
      absl::optional<QuicDatagramContextId> context_id) const;

  // Writes |capsule| onto the DATA stream.
  void WriteCapsule(const Capsule& capsule, bool fin = false);

  void WriteGreaseCapsule();

 protected:
  // Called when the received headers are too large. By default this will
  // reset the stream.
  virtual void OnHeadersTooLarge();

  virtual void OnInitialHeadersComplete(bool fin, size_t frame_len,
                                        const QuicHeaderList& header_list);
  virtual void OnTrailingHeadersComplete(bool fin, size_t frame_len,
                                         const QuicHeaderList& header_list);
  virtual size_t WriteHeadersImpl(
      spdy::SpdyHeaderBlock header_block, bool fin,
      QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener);

  Visitor* visitor() { return visitor_; }

  void set_headers_decompressed(bool val) { headers_decompressed_ = val; }

  void set_ack_listener(
      QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener) {
    ack_listener_ = std::move(ack_listener);
  }

  void OnWriteSideInDataRecvdState() override;

  virtual bool AreHeadersValid(const QuicHeaderList& header_list) const;

  // Reset stream upon invalid request headers.
  virtual void OnInvalidHeaders();

 private:
  friend class test::QuicSpdyStreamPeer;
  friend class test::QuicStreamPeer;
  friend class QuicStreamUtils;
  class HttpDecoderVisitor;

  struct QUIC_EXPORT_PRIVATE WebTransportDataStream {
    WebTransportDataStream(QuicSpdyStream* stream,
                           WebTransportSessionId session_id);

    WebTransportSessionId session_id;
    WebTransportStreamAdapter adapter;
  };

  // Called by HttpDecoderVisitor.
  bool OnDataFrameStart(QuicByteCount header_length,
                        QuicByteCount payload_length);
  bool OnDataFramePayload(absl::string_view payload);
  bool OnDataFrameEnd();
  bool OnHeadersFrameStart(QuicByteCount header_length,
                           QuicByteCount payload_length);
  bool OnHeadersFramePayload(absl::string_view payload);
  bool OnHeadersFrameEnd();
  void OnWebTransportStreamFrameType(QuicByteCount header_length,
                                     WebTransportSessionId session_id);
  bool OnUnknownFrameStart(uint64_t frame_type, QuicByteCount header_length,
                           QuicByteCount payload_length);
  bool OnUnknownFramePayload(absl::string_view payload);
  bool OnUnknownFrameEnd();

  // Given the interval marked by [|offset|, |offset| + |data_length|), return
  // the number of frame header bytes contained in it.
  QuicByteCount GetNumFrameHeadersInInterval(QuicStreamOffset offset,
                                             QuicByteCount data_length) const;

  void MaybeProcessSentWebTransportHeaders(spdy::SpdyHeaderBlock& headers);
  void MaybeProcessReceivedWebTransportHeaders();

  // Writes HTTP/3 DATA frame header. If |force_write| is true, use
  // WriteOrBufferData if send buffer cannot accomodate the header + data.
  ABSL_MUST_USE_RESULT bool WriteDataFrameHeader(QuicByteCount data_length,
                                                 bool force_write);

  // Simply calls OnBodyAvailable() unless capsules are in use, in which case
  // pass the capsule fragments to the capsule manager.
  void HandleBodyAvailable();

  // Called when a datagram frame or capsule is received.
  void HandleReceivedDatagram(absl::optional<QuicDatagramContextId> context_id,
                              absl::string_view payload);

  QuicSpdySession* spdy_session_;

  bool on_body_available_called_because_sequencer_is_closed_;

  Visitor* visitor_;

  // True if read side processing is blocked while waiting for callback from
  // QPACK decoder.
  bool blocked_on_decoding_headers_;
  // True if the headers have been completely decompressed.
  bool headers_decompressed_;
  // True if uncompressed headers or trailers exceed maximum allowed size
  // advertised to peer via SETTINGS_MAX_HEADER_LIST_SIZE.
  bool header_list_size_limit_exceeded_;
  // Contains a copy of the decompressed header (name, value) pairs until they
  // are consumed via Readv.
  QuicHeaderList header_list_;
  // Length of most recently received HEADERS frame payload.
  QuicByteCount headers_payload_length_;

  // True if the trailers have been completely decompressed.
  bool trailers_decompressed_;
  // True if the trailers have been consumed.
  bool trailers_consumed_;

  // The parsed trailers received from the peer.
  spdy::SpdyHeaderBlock received_trailers_;

  // Headers accumulator for decoding HEADERS frame payload.
  std::unique_ptr<QpackDecodedHeadersAccumulator>
      qpack_decoded_headers_accumulator_;
  // Visitor of the HttpDecoder.
  std::unique_ptr<HttpDecoderVisitor> http_decoder_visitor_;
  // HttpDecoder for processing raw incoming stream frames.
  HttpDecoder decoder_;
  // Object that manages references to DATA frame payload fragments buffered by
  // the sequencer and calculates how much data should be marked consumed with
  // the sequencer each time new stream data is processed.
  QuicSpdyStreamBodyManager body_manager_;

  std::unique_ptr<CapsuleParser> capsule_parser_;

  // Sequencer offset keeping track of how much data HttpDecoder has processed.
  // Initial value is zero for fresh streams, or sequencer()->NumBytesConsumed()
  // at time of construction if a PendingStream is converted to account for the
  // length of the unidirectional stream type at the beginning of the stream.
  QuicStreamOffset sequencer_offset_;

  // True when inside an HttpDecoder::ProcessInput() call.
  // Used for detecting reentrancy.
  bool is_decoder_processing_input_;

  // Ack listener of this stream, and it is notified when any of written bytes
  // are acked or retransmitted.
  QuicReferenceCountedPointer<QuicAckListenerInterface> ack_listener_;

  // Offset of unacked frame headers.
  QuicIntervalSet<QuicStreamOffset> unacked_frame_headers_offsets_;

  // Urgency value sent in the last PRIORITY_UPDATE frame, or default urgency
  // defined by the spec if no PRIORITY_UPDATE frame has been sent.
  int last_sent_urgency_;

  // If this stream is a WebTransport extended CONNECT stream, contains the
  // WebTransport session associated with this stream.
  std::unique_ptr<WebTransportHttp3> web_transport_;

  // If this stream is a WebTransport data stream, |web_transport_data_|
  // contains all of the associated metadata.
  std::unique_ptr<WebTransportDataStream> web_transport_data_;

  // HTTP/3 Datagram support.
  Http3DatagramRegistrationVisitor* datagram_registration_visitor_ = nullptr;
  Http3DatagramVisitor* datagram_no_context_visitor_ = nullptr;
  absl::optional<QuicDatagramStreamId> datagram_flow_id_;
  QuicDatagramContextId datagram_next_available_context_id_;
  absl::flat_hash_map<QuicDatagramContextId, Http3DatagramVisitor*>
      datagram_context_visitors_;
};

}  // namespace quic

#endif  // QUICHE_QUIC_CORE_HTTP_QUIC_SPDY_STREAM_H_