gecko-dev/netwerk/protocol/http/Http3WebTransportStream.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/. */

#include "Http3WebTransportStream.h"

#include "HttpLog.h"
#include "Http3Session.h"
#include "Http3WebTransportSession.h"
#include "mozilla/TimeStamp.h"
#include "nsHttpHandler.h"
#include "nsIOService.h"
#include "nsIPipe.h"
#include "nsSocketTransportService2.h"
#include "nsIWebTransportStream.h"

namespace mozilla::net {

namespace {

// This is an nsAHttpTransaction that does nothing.
class DummyWebTransportStreamTransaction : public nsAHttpTransaction {
 public:
  NS_DECL_THREADSAFE_ISUPPORTS

  DummyWebTransportStreamTransaction() = default;

  void SetConnection(nsAHttpConnection*) override {}
  nsAHttpConnection* Connection() override { return nullptr; }
  void GetSecurityCallbacks(nsIInterfaceRequestor**) override {}
  void OnTransportStatus(nsITransport* transport, nsresult status,
                         int64_t progress) override {}
  bool IsDone() override { return false; }
  nsresult Status() override { return NS_OK; }
  uint32_t Caps() override { return 0; }
  [[nodiscard]] nsresult ReadSegments(nsAHttpSegmentReader*, uint32_t,
                                      uint32_t*) override {
    return NS_OK;
  }
  [[nodiscard]] nsresult WriteSegments(nsAHttpSegmentWriter*, uint32_t,
                                       uint32_t*) override {
    return NS_OK;
  }
  void Close(nsresult reason) override {}
  nsHttpConnectionInfo* ConnectionInfo() override { return nullptr; }
  void SetProxyConnectFailed() override {}
  nsHttpRequestHead* RequestHead() override { return nullptr; }
  uint32_t Http1xTransactionCount() override { return 0; }
  [[nodiscard]] nsresult TakeSubTransactions(
      nsTArray<RefPtr<nsAHttpTransaction>>& outTransactions) override {
    return NS_OK;
  }

 private:
  virtual ~DummyWebTransportStreamTransaction() = default;
};

NS_IMPL_ISUPPORTS(DummyWebTransportStreamTransaction, nsISupportsWeakReference)

class WebTransportSendStreamStats : public nsIWebTransportSendStreamStats {
 public:
  NS_DECL_THREADSAFE_ISUPPORTS

  explicit WebTransportSendStreamStats(uint64_t aSent, uint64_t aAcked)
      : mTimeStamp(TimeStamp::Now()),
        mTotalSent(aSent),
        mTotalAcknowledged(aAcked) {}

  NS_IMETHOD GetTimestamp(mozilla::TimeStamp* aTimestamp) override {
    *aTimestamp = mTimeStamp;
    return NS_OK;
  }
  NS_IMETHOD GetBytesSent(uint64_t* aBytesSent) override {
    *aBytesSent = mTotalSent;
    return NS_OK;
  }
  NS_IMETHOD GetBytesAcknowledged(uint64_t* aBytesAcknowledged) override {
    *aBytesAcknowledged = mTotalAcknowledged;
    return NS_OK;
  }

 private:
  virtual ~WebTransportSendStreamStats() = default;

  TimeStamp mTimeStamp;
  uint64_t mTotalSent;
  uint64_t mTotalAcknowledged;
};

NS_IMPL_ISUPPORTS(WebTransportSendStreamStats, nsIWebTransportSendStreamStats)

class WebTransportReceiveStreamStats
    : public nsIWebTransportReceiveStreamStats {
 public:
  NS_DECL_THREADSAFE_ISUPPORTS

  explicit WebTransportReceiveStreamStats(uint64_t aReceived)
      : mTimeStamp(TimeStamp::Now()), mTotalReceived(aReceived) {}

  NS_IMETHOD GetTimestamp(mozilla::TimeStamp* aTimestamp) override {
    *aTimestamp = mTimeStamp;
    return NS_OK;
  }
  NS_IMETHOD GetBytesReceived(uint64_t* aByteReceived) override {
    *aByteReceived = mTotalReceived;
    return NS_OK;
  }

 private:
  virtual ~WebTransportReceiveStreamStats() = default;

  TimeStamp mTimeStamp;
  uint64_t mTotalReceived;
};

NS_IMPL_ISUPPORTS(WebTransportReceiveStreamStats,
                  nsIWebTransportReceiveStreamStats)

}  // namespace

NS_IMPL_ISUPPORTS(Http3WebTransportStream, nsIInputStreamCallback)

Http3WebTransportStream::Http3WebTransportStream(
    Http3Session* aSession, uint64_t aSessionId, WebTransportStreamType aType,
    std::function<void(Result<RefPtr<Http3WebTransportStream>, nsresult>&&)>&&
        aCallback)
    : Http3StreamBase(new DummyWebTransportStreamTransaction(), aSession),
      mSessionId(aSessionId),
      mStreamType(aType),
      mStreamRole(OUTGOING),
      mStreamReadyCallback(std::move(aCallback)) {
  LOG(("Http3WebTransportStream outgoing ctor %p", this));
}

Http3WebTransportStream::Http3WebTransportStream(Http3Session* aSession,
                                                 uint64_t aSessionId,
                                                 WebTransportStreamType aType,
                                                 uint64_t aStreamId)
    : Http3StreamBase(new DummyWebTransportStreamTransaction(), aSession),
      mSessionId(aSessionId),
      mStreamType(aType),
      mStreamRole(INCOMING),
      // WAITING_DATA indicates we are waiting
      // Http3WebTransportStream::OnInputStreamReady to be called.
      mSendState(WAITING_DATA),
      mStreamReadyCallback(nullptr) {
  LOG(("Http3WebTransportStream incoming ctor %p", this));
  mStreamId = aStreamId;
}

Http3WebTransportStream::~Http3WebTransportStream() {
  LOG(("Http3WebTransportStream dtor %p", this));
}

nsresult Http3WebTransportStream::TryActivating() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  return mSession->TryActivatingWebTransportStream(&mStreamId, this);
}

NS_IMETHODIMP Http3WebTransportStream::OnInputStreamReady(
    nsIAsyncInputStream* aStream) {
  LOG1(
      ("Http3WebTransportStream::OnInputStreamReady [this=%p stream=%p "
       "state=%d]",
       this, aStream, mSendState));
  if (mSendState == SEND_DONE) {
    // already closed
    return NS_OK;
  }

  mSendState = SENDING;
  mSession->StreamHasDataToWrite(this);
  return NS_OK;
}

nsresult Http3WebTransportStream::InitOutputPipe() {
  nsCOMPtr<nsIAsyncOutputStream> out;
  nsCOMPtr<nsIAsyncInputStream> in;
  NS_NewPipe2(getter_AddRefs(in), getter_AddRefs(out), true, true,
              nsIOService::gDefaultSegmentSize,
              nsIOService::gDefaultSegmentCount);

  {
    MutexAutoLock lock(mMutex);
    mSendStreamPipeIn = std::move(in);
    mSendStreamPipeOut = std::move(out);
  }

  nsresult rv =
      mSendStreamPipeIn->AsyncWait(this, 0, 0, gSocketTransportService);
  if (NS_FAILED(rv)) {
    return rv;
  }

  mSendState = WAITING_DATA;
  return NS_OK;
}

nsresult Http3WebTransportStream::InitInputPipe() {
  nsCOMPtr<nsIAsyncOutputStream> out;
  nsCOMPtr<nsIAsyncInputStream> in;
  NS_NewPipe2(getter_AddRefs(in), getter_AddRefs(out), true, true,
              nsIOService::gDefaultSegmentSize,
              nsIOService::gDefaultSegmentCount);

  {
    MutexAutoLock lock(mMutex);
    mReceiveStreamPipeIn = std::move(in);
    mReceiveStreamPipeOut = std::move(out);
  }

  mRecvState = READING;
  return NS_OK;
}

void Http3WebTransportStream::GetWriterAndReader(
    nsIAsyncOutputStream** aOutOutputStream,
    nsIAsyncInputStream** aOutInputStream) {
  nsCOMPtr<nsIAsyncOutputStream> output;
  nsCOMPtr<nsIAsyncInputStream> input;
  {
    MutexAutoLock lock(mMutex);
    output = mSendStreamPipeOut;
    input = mReceiveStreamPipeIn;
  }

  output.forget(aOutOutputStream);
  input.forget(aOutInputStream);
}

already_AddRefed<nsIWebTransportSendStreamStats>
Http3WebTransportStream::GetSendStreamStats() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  nsCOMPtr<nsIWebTransportSendStreamStats> stats =
      new WebTransportSendStreamStats(mTotalSent, mTotalAcknowledged);
  return stats.forget();
}

already_AddRefed<nsIWebTransportReceiveStreamStats>
Http3WebTransportStream::GetReceiveStreamStats() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");

  nsCOMPtr<nsIWebTransportReceiveStreamStats> stats =
      new WebTransportReceiveStreamStats(mTotalReceived);
  return stats.forget();
}

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

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

  nsresult rv = NS_OK;

  switch (mSendState) {
    case WAITING_TO_ACTIVATE:
      rv = TryActivating();
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        LOG3(
            ("Http3WebTransportStream::OnReadSegment %p cannot activate now. "
             "queued.\n",
             this));
        break;
      }
      if (NS_FAILED(rv)) {
        LOG3(
            ("Http3WebTransportStream::OnReadSegment %p cannot activate "
             "error=0x%" PRIx32 ".",
             this, static_cast<uint32_t>(rv)));
        mStreamReadyCallback(Err(rv));
        mStreamReadyCallback = nullptr;
        break;
      }

      rv = InitOutputPipe();
      if (NS_SUCCEEDED(rv) && mStreamType == WebTransportStreamType::BiDi) {
        rv = InitInputPipe();
      }
      if (NS_FAILED(rv)) {
        LOG3(
            ("Http3WebTransportStream::OnReadSegment %p failed to create pipe "
             "error=0x%" PRIx32 ".",
             this, static_cast<uint32_t>(rv)));
        mSendState = SEND_DONE;
        mStreamReadyCallback(Err(rv));
        mStreamReadyCallback = nullptr;
        break;
      }

      // Successfully activated.
      mStreamReadyCallback(RefPtr{this});
      mStreamReadyCallback = nullptr;
      break;
    case SENDING: {
      rv = mSession->SendRequestBody(mStreamId, buf, count, countRead);
      LOG3(
          ("Http3WebTransportStream::OnReadSegment %p sending body returns "
           "error=0x%" PRIx32 ".",
           this, static_cast<uint32_t>(rv)));
      mTotalSent += *countRead;
    } break;
    case WAITING_DATA:
      // Still waiting
      LOG3((
          "Http3WebTransportStream::OnReadSegment %p Still waiting for data...",
          this));
      break;
    case SEND_DONE:
      LOG3(("Http3WebTransportStream::OnReadSegment %p called after SEND_DONE ",
            this));
      MOZ_ASSERT(false, "We are done sending this request!");
      MOZ_ASSERT(mStreamReadyCallback);
      rv = NS_ERROR_UNEXPECTED;
      mStreamReadyCallback(Err(rv));
      mStreamReadyCallback = nullptr;
      break;
  }

  mSocketOutCondition = rv;

  return mSocketOutCondition;
}

// static
nsresult Http3WebTransportStream::ReadRequestSegment(
    nsIInputStream* stream, void* closure, const char* buf, uint32_t offset,
    uint32_t count, uint32_t* countRead) {
  Http3WebTransportStream* wtStream = (Http3WebTransportStream*)closure;
  nsresult rv = wtStream->OnReadSegment(buf, count, countRead);
  LOG(("Http3WebTransportStream::ReadRequestSegment %p read=%u", wtStream,
       *countRead));
  return rv;
}

nsresult Http3WebTransportStream::ReadSegments() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportStream::ReadSegments [this=%p]", this));
  nsresult rv = NS_OK;
  uint32_t sendBytes = 0;
  bool again = true;
  do {
    sendBytes = 0;
    rv = mSocketOutCondition = NS_OK;
    LOG(("Http3WebTransportStream::ReadSegments state=%d [this=%p]", mSendState,
         this));
    switch (mSendState) {
      case WAITING_TO_ACTIVATE: {
        LOG3(
            ("Http3WebTransportStream %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)));
        if (mSendState != WAITING_DATA) {
          break;
        }
      }
        [[fallthrough]];
      case WAITING_DATA:
        [[fallthrough]];
      case SENDING: {
        if (mStreamRole == INCOMING &&
            mStreamType == WebTransportStreamType::UniDi) {
          rv = NS_OK;
          break;
        }
        mSendState = SENDING;
        rv = mSendStreamPipeIn->ReadSegments(ReadRequestSegment, this,
                                             nsIOService::gDefaultSegmentSize,
                                             &sendBytes);
      } break;
      case SEND_DONE: {
        return NS_OK;
      }
      default:
        sendBytes = 0;
        rv = NS_OK;
        break;
    }

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

    // XXX some streams return NS_BASE_STREAM_CLOSED to indicate EOF.
    if (rv == NS_BASE_STREAM_CLOSED || !mPendingTasks.IsEmpty()) {
      rv = NS_OK;
      sendBytes = 0;
    }

    if (NS_FAILED(rv)) {
      // if the writer didn't want to write any more data, then
      // wait for the transaction to call ResumeSend.
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        mSendState = WAITING_DATA;
        rv = mSendStreamPipeIn->AsyncWait(this, 0, 0, gSocketTransportService);
      }
      again = false;

      // Got a  WebTransport specific error
      if (rv >= NS_ERROR_WEBTRANSPORT_CODE_BASE &&
          rv <= NS_ERROR_WEBTRANSPORT_CODE_END) {
        uint8_t errorCode = GetWebTransportErrorFromNSResult(rv);
        mSendState = SEND_DONE;
        Reset(WebTransportErrorToHttp3Error(errorCode));
        rv = NS_OK;
      }
    } else if (NS_FAILED(mSocketOutCondition)) {
      if (mSocketOutCondition != NS_BASE_STREAM_WOULD_BLOCK) {
        rv = mSocketOutCondition;
      }
      again = false;
    } else if (!sendBytes) {
      mSendState = SEND_DONE;
      rv = NS_OK;
      again = false;
      if (!mPendingTasks.IsEmpty()) {
        LOG(("Has pending tasks to do"));
        nsTArray<std::function<void()>> tasks = std::move(mPendingTasks);
        for (const auto& task : tasks) {
          task();
        }
      }
      // Tell the underlying stream we're done
      SendFin();
    }

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

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

  LOG(("Http3WebTransportStream::OnWriteSegment [this=%p, state=%d", this,
       static_cast<uint32_t>(mRecvState)));
  nsresult rv = NS_OK;
  switch (mRecvState) {
    case READING: {
      rv = mSession->ReadResponseData(mStreamId, buf, count, countWritten,
                                      &mFin);
      if (*countWritten == 0) {
        if (mFin) {
          mRecvState = RECV_DONE;
          rv = NS_BASE_STREAM_CLOSED;
        } else {
          rv = NS_BASE_STREAM_WOULD_BLOCK;
        }
      } else {
        mTotalReceived += *countWritten;
        if (mFin) {
          mRecvState = RECEIVED_FIN;
        }
      }
    } break;
    case RECEIVED_FIN:
      rv = NS_BASE_STREAM_CLOSED;
      mRecvState = RECV_DONE;
      break;
    case RECV_DONE:
      rv = NS_ERROR_UNEXPECTED;
      break;
    default:
      rv = NS_ERROR_UNEXPECTED;
      break;
  }

  // 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;
}

// static
nsresult Http3WebTransportStream::WritePipeSegment(nsIOutputStream* stream,
                                                   void* closure, char* buf,
                                                   uint32_t offset,
                                                   uint32_t count,
                                                   uint32_t* countWritten) {
  Http3WebTransportStream* self = (Http3WebTransportStream*)closure;

  nsresult rv = self->OnWriteSegment(buf, count, countWritten);
  if (NS_FAILED(rv)) {
    return rv;
  }

  LOG(("Http3WebTransportStream::WritePipeSegment %p written=%u", self,
       *countWritten));

  return rv;
}

nsresult Http3WebTransportStream::WriteSegments() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  if (!mReceiveStreamPipeOut) {
    return NS_OK;
  }

  LOG(("Http3WebTransportStream::WriteSegments [this=%p]", this));

  nsresult rv = NS_OK;
  uint32_t countWrittenSingle = 0;
  bool again = true;

  do {
    mSocketInCondition = NS_OK;
    countWrittenSingle = 0;
    rv = mReceiveStreamPipeOut->WriteSegments(WritePipeSegment, this,
                                              nsIOService::gDefaultSegmentSize,
                                              &countWrittenSingle);
    LOG(("Http3WebTransportStream::WriteSegments rv=0x%" PRIx32
         " countWrittenSingle=%" PRIu32 " socketin=%" PRIx32 " [this=%p]",
         static_cast<uint32_t>(rv), countWrittenSingle,
         static_cast<uint32_t>(mSocketInCondition), this));
    if (NS_FAILED(rv)) {
      if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
        rv = NS_OK;
      }
      if (rv == NS_BASE_STREAM_CLOSED) {
        mReceiveStreamPipeOut->Close();
        rv = NS_OK;
      }
      again = false;
    } else if (NS_FAILED(mSocketInCondition)) {
      if (mSocketInCondition != NS_BASE_STREAM_WOULD_BLOCK) {
        rv = mSocketInCondition;
        if (rv == NS_BASE_STREAM_CLOSED) {
          mReceiveStreamPipeOut->Close();
          rv = NS_OK;
        }
      }
      again = false;
    }
    // read more from the socket until error...
  } while (again && gHttpHandler->Active());

  return rv;
}

bool Http3WebTransportStream::Done() const {
  return mSendState == SEND_DONE && mRecvState == RECV_DONE;
}

void Http3WebTransportStream::Close(nsresult aResult) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportStream::Close [this=%p]", this));
  mTransaction = nullptr;
  if (mSendStreamPipeIn) {
    mSendStreamPipeIn->AsyncWait(nullptr, 0, 0, nullptr);
    mSendStreamPipeIn->CloseWithStatus(aResult);
  }
  if (mReceiveStreamPipeOut) {
    mReceiveStreamPipeOut->AsyncWait(nullptr, 0, 0, nullptr);
    mReceiveStreamPipeOut->CloseWithStatus(aResult);
  }
  mSendState = SEND_DONE;
  mRecvState = RECV_DONE;
  mSession = nullptr;
}

void Http3WebTransportStream::SendFin() {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportStream::SendFin [this=%p mSendState=%d]", this,
       mSendState));

  if (mSendFin || !mSession || mResetError) {
    // Already closed.
    return;
  }

  mSendFin = true;

  switch (mSendState) {
    case SENDING: {
      mPendingTasks.AppendElement([self = RefPtr{this}]() {
        self->mSession->CloseSendingSide(self->mStreamId);
      });
    } break;
    case WAITING_DATA:
      mSendState = SEND_DONE;
      [[fallthrough]];
    case SEND_DONE:
      mSession->CloseSendingSide(mStreamId);
      // StreamHasDataToWrite needs to be called to trigger ProcessOutput.
      mSession->StreamHasDataToWrite(this);
      break;
    default:
      MOZ_ASSERT_UNREACHABLE("invalid mSendState!");
      break;
  }
}

void Http3WebTransportStream::Reset(uint64_t aErrorCode) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportStream::Reset [this=%p, mSendState=%d]", this,
       mSendState));

  if (mResetError || !mSession || mSendFin) {
    // The stream is already reset.
    return;
  }

  mResetError = Some(aErrorCode);

  switch (mSendState) {
    case SENDING: {
      LOG(("Http3WebTransportStream::Reset [this=%p] reset after sending data",
           this));
      mPendingTasks.AppendElement([self = RefPtr{this}]() {
        // "Reset" needs a special treatment here. If we are sending data and
        // ResetWebTransportStream is called before Http3Session::ProcessOutput,
        // neqo will drop the last piece of data.
        NS_DispatchToCurrentThread(
            NS_NewRunnableFunction("Http3WebTransportStream::Reset", [self]() {
              self->mSession->ResetWebTransportStream(self, *self->mResetError);
              self->mSession->StreamHasDataToWrite(self);
              self->mSession->ConnectSlowConsumer(self);
            }));
      });
    } break;
    case WAITING_DATA:
      mSendState = SEND_DONE;
      [[fallthrough]];
    case SEND_DONE:
      mSession->ResetWebTransportStream(this, *mResetError);
      // StreamHasDataToWrite needs to be called to trigger ProcessOutput.
      mSession->StreamHasDataToWrite(this);
      mSession->ConnectSlowConsumer(this);
      break;
    default:
      MOZ_ASSERT_UNREACHABLE("invalid mSendState!");
      break;
  }
}

void Http3WebTransportStream::SendStopSending(uint8_t aErrorCode) {
  MOZ_ASSERT(OnSocketThread(), "not on socket thread");
  LOG(("Http3WebTransportStream::SendStopSending [this=%p, mSendState=%d]",
       this, mSendState));

  if (mSendState == WAITING_TO_ACTIVATE) {
    return;
  }

  if (mStopSendingError || !mSession) {
    return;
  }

  mStopSendingError = Some(aErrorCode);

  mSession->StreamStopSending(this, *mStopSendingError);
  // StreamHasDataToWrite needs to be called to trigger ProcessOutput.
  mSession->StreamHasDataToWrite(this);
}

void Http3WebTransportStream::SetSendOrder(Maybe<int64_t> aSendOrder) {
  mSession->SetSendOrder(this, aSendOrder);
}

}  // namespace mozilla::net