firefox/netwerk/protocol/http/Http3WebTransportSession.cpp

/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

// HttpLog.h should generally be included first
#include "HttpLog.h"
#include "Http3WebTransportSession.h"
#include "Http3WebTransportStream.h"
#include "Http3Session.h"
#include "Http3Stream.h"
#include "nsHttpRequestHead.h"
#include "nsHttpTransaction.h"
#include "nsIClassOfService.h"
#include "nsISocketTransport.h"
#include "nsSocketTransportService2.h"
#include "nsIOService.h"
#include "nsHttpHandler.h"

namespace mozilla::net {

Http3WebTransportSession::Http3WebTransportSession(nsAHttpTransaction* trans,
                                                   Http3Session* aHttp3Session)
    : Http3StreamBase(trans, aHttp3Session) {}

Http3WebTransportSession::~Http3WebTransportSession() = default;

uint64_t Http3WebTransportSession::GetStreamId() const {
  return Http3StreamBase::StreamId();
}

nsresult Http3WebTransportSession::ReadSegments() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::ReadSegments %p mSendState=%d mRecvState=%d",
       this, mSendState, mRecvState));
  if (mSendState == PROCESSING_DATAGRAM) {
    return NS_OK;
  }

  if ((mRecvState == RECV_DONE) || (mRecvState == ACTIVE) ||
      (mRecvState == CLOSE_PENDING)) {
    // Don't transmit any request frames if the peer cannot respond or respone
    // is already done.
    LOG3((
        "Http3WebTransportSession %p ReadSegments request stream aborted due to"
        " response side closure\n",
        this));
    return NS_ERROR_ABORT;
  }

  nsresult rv = NS_OK;
  uint32_t transactionBytes;
  bool again = true;
  do {
    transactionBytes = 0;
    rv = mSocketOutCondition = NS_OK;
    LOG(("Http3WebTransportSession::ReadSegments state=%d [this=%p]",
         mSendState, this));
    switch (mSendState) {
      case PREPARING_HEADERS: {
        rv = mTransaction->ReadSegmentsAgain(
            this, nsIOService::gDefaultSegmentSize, &transactionBytes, &again);
      } break;
      case WAITING_TO_ACTIVATE: {
        // A transaction that had already generated its headers before it was
        // queued at the session level (due to concurrency concerns) may not
        // call onReadSegment off the ReadSegments() stack above.
        LOG3(
            ("Http3WebTransportSession %p ReadSegments forcing OnReadSegment "
             "call\n",
             this));
        uint32_t wasted = 0;
        nsresult rv2 = OnReadSegment("", 0, &wasted);
        LOG3(("  OnReadSegment returned 0x%08" PRIx32,
              static_cast<uint32_t>(rv2)));
      } break;
      default:
        transactionBytes = 0;
        rv = NS_OK;
        break;
    }

    LOG(("Http3WebTransportSession::ReadSegments rv=0x%" PRIx32
         " read=%u sock-cond=%" PRIx32 " again=%d [this=%p]",
         static_cast<uint32_t>(rv), transactionBytes,
         static_cast<uint32_t>(mSocketOutCondition), again, this));

    // XXX some streams return NS_BASE_STREAM_CLOSED to indicate EOF.
    if (rv == NS_BASE_STREAM_CLOSED && !mTransaction->IsDone()) {
      rv = NS_OK;
      transactionBytes = 0;
    }

    if (NS_FAILED(rv)) {
      // if the transaction didn't want to write any more data, then
      // wait for the transaction to call ResumeSend.
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        rv = NS_OK;
      }
      again = false;
    } else if (NS_FAILED(mSocketOutCondition)) {
      if (mSocketOutCondition != NS_BASE_STREAM_WOULD_BLOCK) {
        rv = mSocketOutCondition;
      }
      again = false;
    } else if (!transactionBytes) {
      mTransaction->OnTransportStatus(nullptr, NS_NET_STATUS_WAITING_FOR, 0);

      mSendState = PROCESSING_DATAGRAM;
      rv = NS_OK;
      again = false;
    }
    // write more to the socket until error or end-of-request...
  } while (again && gHttpHandler->Active());
  return rv;
}

bool Http3WebTransportSession::ConsumeHeaders(const char* buf, uint32_t avail,
                                              uint32_t* countUsed) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG3(("Http3WebTransportSession::ConsumeHeaders %p avail=%u.", this, avail));

  mFlatHttpRequestHeaders.Append(buf, avail);
  // We can use the simple double crlf because firefox is the
  // only client we are parsing
  int32_t endHeader = mFlatHttpRequestHeaders.Find("\r\n\r\n");

  if (endHeader == kNotFound) {
    // We don't have all the headers yet
    LOG3(
        ("Http3WebTransportSession::ConsumeHeaders %p "
         "Need more header bytes. Len = %zu",
         this, mFlatHttpRequestHeaders.Length()));
    *countUsed = avail;
    return false;
  }

  uint32_t oldLen = mFlatHttpRequestHeaders.Length();
  mFlatHttpRequestHeaders.SetLength(endHeader + 2);
  *countUsed = avail - (oldLen - endHeader) + 4;

  return true;
}

nsresult Http3WebTransportSession::TryActivating() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::TryActivating [this=%p]", this));
  nsHttpRequestHead* head = mTransaction->RequestHead();

  nsAutoCString host;
  nsresult rv = head->GetHeader(nsHttp::Host, host);
  if (NS_FAILED(rv)) {
    MOZ_ASSERT(false);
    return rv;
  }
  nsAutoCString path;
  head->Path(path);

  return mSession->TryActivating(""_ns, ""_ns, host, path,
                                 mFlatHttpRequestHeaders, &mStreamId, this);
}

nsresult Http3WebTransportSession::OnReadSegment(const char* buf,
                                                 uint32_t count,
                                                 uint32_t* countRead) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  LOG(("Http3WebTransportSession::OnReadSegment count=%u state=%d [this=%p]",
       count, mSendState, this));

  nsresult rv = NS_OK;

  switch (mSendState) {
    case PREPARING_HEADERS: {
      if (!ConsumeHeaders(buf, count, countRead)) {
        break;
      }
      mSendState = WAITING_TO_ACTIVATE;
    }
      [[fallthrough]];
    case WAITING_TO_ACTIVATE:
      rv = TryActivating();
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        LOG3(
            ("Http3WebTransportSession::OnReadSegment %p cannot activate now. "
             "queued.\n",
             this));
        break;
      }
      if (NS_FAILED(rv)) {
        LOG3(
            ("Http3WebTransportSession::OnReadSegment %p cannot activate "
             "error=0x%" PRIx32 ".",
             this, static_cast<uint32_t>(rv)));
        break;
      }

      // Successfully activated.
      mTransaction->OnTransportStatus(nullptr, NS_NET_STATUS_SENDING_TO, 0);

      mSendState = PROCESSING_DATAGRAM;
      break;
    default:
      MOZ_ASSERT(false, "We are done sending this request!");
      rv = NS_ERROR_UNEXPECTED;
      break;
  }

  mSocketOutCondition = rv;

  return mSocketOutCondition;
}

nsresult Http3WebTransportSession::WriteSegments() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::WriteSegments [this=%p]", this));
  nsresult rv = NS_OK;
  uint32_t countWrittenSingle = 0;
  bool again = true;

  if (mRecvState == CLOSE_PENDING) {
    mSession->CloseWebTransport(mStreamId, mStatus, mReason);
    mRecvState = RECV_DONE;
    // This will closed the steam because the stream is Done().
    return NS_OK;
  }

  do {
    mSocketInCondition = NS_OK;
    countWrittenSingle = 0;
    rv = mTransaction->WriteSegmentsAgain(
        this, nsIOService::gDefaultSegmentSize, &countWrittenSingle, &again);
    LOG(("Http3WebTransportSession::WriteSegments rv=0x%" PRIx32
         " countWrittenSingle=%" PRIu32 " socketin=%" PRIx32 " [this=%p]",
         static_cast<uint32_t>(rv), countWrittenSingle,
         static_cast<uint32_t>(mSocketInCondition), this));
    if (mTransaction->IsDone()) {
      // An HTTP transaction used for setting up a  WebTransport session will
      // receive only response headers and afterward, it will be marked as
      // done. At this point, the session negotiation has finished and the
      // WebTransport session transfers into the ACTIVE state.
      mRecvState = ACTIVE;
    }

    if (NS_FAILED(rv)) {
      // if the transaction didn't want to take any more data, then
      // wait for the transaction to call ResumeRecv.
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        rv = NS_OK;
      }
      again = false;
    } else if (NS_FAILED(mSocketInCondition)) {
      if (mSocketInCondition != NS_BASE_STREAM_WOULD_BLOCK) {
        rv = mSocketInCondition;
      }
      again = false;
    }
    // read more from the socket until error...
  } while (again && gHttpHandler->Active());

  return rv;
}

void Http3WebTransportSession::SetResponseHeaders(
    nsTArray<uint8_t>& aResponseHeaders, bool fin, bool interim) {
  MOZ_ASSERT(mRecvState == BEFORE_HEADERS ||
             mRecvState == READING_INTERIM_HEADERS);
  mFlatResponseHeaders.AppendElements(aResponseHeaders);
  mRecvState = (interim) ? READING_INTERIM_HEADERS : READING_HEADERS;
}

nsresult Http3WebTransportSession::OnWriteSegment(char* buf, uint32_t count,
                                                  uint32_t* countWritten) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  LOG(("Http3WebTransportSession::OnWriteSegment [this=%p, state=%d", this,
       mRecvState));
  nsresult rv = NS_OK;
  switch (mRecvState) {
    case BEFORE_HEADERS: {
      *countWritten = 0;
      rv = NS_BASE_STREAM_WOULD_BLOCK;
    } break;
    case READING_HEADERS:
    case READING_INTERIM_HEADERS: {
      // SetResponseHeaders should have been previously called.
      MOZ_ASSERT(!mFlatResponseHeaders.IsEmpty(), "Headers empty!");
      *countWritten = (mFlatResponseHeaders.Length() > count)
                          ? count
                          : mFlatResponseHeaders.Length();
      memcpy(buf, mFlatResponseHeaders.Elements(), *countWritten);

      mFlatResponseHeaders.RemoveElementsAt(0, *countWritten);
      if (mFlatResponseHeaders.Length() == 0) {
        if (mRecvState == READING_INTERIM_HEADERS) {
          // neqo makes sure that fin cannot be received before the final
          // headers are received.
          mRecvState = BEFORE_HEADERS;
        } else {
          mRecvState = ACTIVE;
        }
      }

      if (*countWritten == 0) {
        rv = NS_BASE_STREAM_WOULD_BLOCK;
      } else {
        mTransaction->OnTransportStatus(nullptr, NS_NET_STATUS_RECEIVING_FROM,
                                        0);
      }
    } break;
    case ACTIVE:
    case CLOSE_PENDING:
    case RECV_DONE:
      rv = NS_ERROR_UNEXPECTED;
  }

  // Remember the error received from lower layers. A stream pipe may overwrite
  // it.
  // If rv == NS_OK this will reset mSocketInCondition.
  mSocketInCondition = rv;

  return rv;
}

void Http3WebTransportSession::Close(nsresult aResult) {
  LOG(("Http3WebTransportSession::Close %p", this));
  if (mListener) {
    mListener->OnSessionClosed(NS_SUCCEEDED(aResult), 0, ""_ns);
    mListener = nullptr;
  }
  if (mTransaction) {
    mTransaction->Close(aResult);
    mTransaction = nullptr;
  }
  mRecvState = RECV_DONE;
  mSendState = SEND_DONE;

  if (mSession) {
    mSession->CloseWebTransportConn();
    mSession = nullptr;
  }
}

void Http3WebTransportSession::OnSessionClosed(bool aCleanly, uint32_t aStatus,
                                               const nsACString& aReason) {
  if (mTransaction) {
    mTransaction->Close(NS_BASE_STREAM_CLOSED);
    mTransaction = nullptr;
  }
  if (mListener) {
    mListener->OnSessionClosed(aCleanly, aStatus, aReason);
    mListener = nullptr;
  }
  mRecvState = RECV_DONE;
  mSendState = SEND_DONE;

  mSession->CloseWebTransportConn();
}

void Http3WebTransportSession::CloseSession(uint32_t aStatus,
                                            const nsACString& aReason) {
  if ((mRecvState != CLOSE_PENDING) && (mRecvState != RECV_DONE)) {
    mStatus = aStatus;
    mReason = aReason;
    mSession->ConnectSlowConsumer(this);
    mRecvState = CLOSE_PENDING;
    mSendState = SEND_DONE;
  }
  mListener = nullptr;
}

void Http3WebTransportSession::TransactionIsDone(nsresult aResult) {
  mTransaction->Close(aResult);
  mTransaction = nullptr;
}

void Http3WebTransportSession::CreateOutgoingBidirectionalStream(
    std::function<void(Result<RefPtr<WebTransportStreamBase>, nsresult>&&)>&&
        aCallback) {
  return CreateStreamInternal(true, std::move(aCallback));
}

void Http3WebTransportSession::CreateOutgoingUnidirectionalStream(
    std::function<void(Result<RefPtr<WebTransportStreamBase>, nsresult>&&)>&&
        aCallback) {
  return CreateStreamInternal(false, std::move(aCallback));
}

void Http3WebTransportSession::CreateStreamInternal(
    bool aBidi,
    std::function<void(Result<RefPtr<WebTransportStreamBase>, nsresult>&&)>&&
        aCallback) {
  LOG(("Http3WebTransportSession::CreateStreamInternal this=%p aBidi=%d", this,
       aBidi));
  if (mRecvState != ACTIVE) {
    aCallback(Err(NS_ERROR_NOT_AVAILABLE));
    return;
  }

  RefPtr<Http3WebTransportStream> stream =
      aBidi ? new Http3WebTransportStream(mSession, mStreamId,
                                          WebTransportStreamType::BiDi,
                                          std::move(aCallback))
            : new Http3WebTransportStream(mSession, mStreamId,
                                          WebTransportStreamType::UniDi,
                                          std::move(aCallback));
  mSession->StreamHasDataToWrite(stream);
  // Put the newly created stream in to |mStreams| to keep it alive.
  mStreams.AppendElement(std::move(stream));
}

// This is called by Http3Session::TryActivatingWebTransportStream. When called,
// this means a WebTransport stream is successfully activated and the stream
// will be managed by Http3Session.
void Http3WebTransportSession::RemoveWebTransportStream(
    Http3WebTransportStream* aStream) {
  LOG(
      ("Http3WebTransportSession::RemoveWebTransportStream "
       "this=%p aStream=%p",
       this, aStream));
  DebugOnly<bool> existed = mStreams.RemoveElement(aStream);
  MOZ_ASSERT(existed);
}

already_AddRefed<Http3WebTransportStream>
Http3WebTransportSession::OnIncomingWebTransportStream(
    WebTransportStreamType aType, uint64_t aId) {
  LOG(
      ("Http3WebTransportSession::OnIncomingWebTransportStream "
       "this=%p",
       this));

  if (mRecvState != ACTIVE) {
    return nullptr;
  }

  MOZ_ASSERT(!mTransaction);
  RefPtr<Http3WebTransportStream> stream =
      new Http3WebTransportStream(mSession, mStreamId, aType, aId);
  if (NS_FAILED(stream->InitInputPipe())) {
    return nullptr;
  }

  if (aType == WebTransportStreamType::BiDi) {
    if (NS_FAILED(stream->InitOutputPipe())) {
      return nullptr;
    }
  }

  if (!mListener) {
    return nullptr;
  }

  if (nsCOMPtr<WebTransportSessionEventListenerInternal> listener =
          do_QueryInterface(mListener)) {
    listener->OnIncomingStreamAvailableInternal(stream);
  }
  return stream.forget();
}

void Http3WebTransportSession::SendDatagram(nsTArray<uint8_t>&& aData,
                                            uint64_t aTrackingId) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::SendDatagram this=%p", this));
  if (mSendState != PROCESSING_DATAGRAM) {
    return;
  }

  mSession->SendDatagram(this, aData, aTrackingId);
  mSession->StreamHasDataToWrite(this);
}

void Http3WebTransportSession::OnDatagramReceived(nsTArray<uint8_t>&& aData) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::OnDatagramReceived this=%p", this));
  if (mRecvState != ACTIVE || !mListener) {
    return;
  }

  if (nsCOMPtr<WebTransportSessionEventListenerInternal> listener =
          do_QueryInterface(mListener)) {
    listener->OnDatagramReceivedInternal(std::move(aData));
  }
}

void Http3WebTransportSession::GetMaxDatagramSize() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  if (mRecvState != ACTIVE || !mListener) {
    return;
  }

  uint64_t size = mSession->MaxDatagramSize(mStreamId);
  mListener->OnMaxDatagramSize(size);
}

void Http3WebTransportSession::OnOutgoingDatagramOutCome(
    uint64_t aId, WebTransportSessionEventListener::DatagramOutcome aOutCome) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportSession::OnOutgoingDatagramOutCome this=%p id=%" PRIx64
       ", outCome=%d mRecvState=%d",
       this, aId, static_cast<uint32_t>(aOutCome), mRecvState));
  if (mRecvState != ACTIVE || !mListener || !aId) {
    return;
  }

  mListener->OnOutgoingDatagramOutCome(aId, aOutCome);
}

void Http3WebTransportSession::OnStreamStopSending(uint64_t aId,
                                                   nsresult aError) {
  LOG(("OnStreamStopSending id:%" PRId64, aId));
  if (!mListener) {
    return;
  }

  mListener->OnStopSending(aId, aError);
}

void Http3WebTransportSession::OnStreamReset(uint64_t aId, nsresult aError) {
  LOG(("OnStreamReset id:%" PRId64, aId));
  if (!mListener) {
    return;
  }

  mListener->OnResetReceived(aId, aError);
}

}  // namespace mozilla::net