dom/fetch/FetchStreamReader.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=8 sts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this
   5  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 #include "FetchStreamReader.h"
   8 #include "InternalResponse.h"
   9 #include "mozilla/ConsoleReportCollector.h"
  10 #include "mozilla/ErrorResult.h"
  11 #include "mozilla/StaticAnalysisFunctions.h"
  12 #include "mozilla/dom/AutoEntryScript.h"
  13 #include "mozilla/dom/Promise.h"
  14 #include "mozilla/dom/PromiseBinding.h"
  15 #include "mozilla/dom/ReadableStream.h"
  16 #include "mozilla/dom/ReadableStreamDefaultController.h"
  17 #include "mozilla/dom/ReadableStreamDefaultReader.h"
  18 #include "mozilla/dom/WorkerPrivate.h"
  19 #include "mozilla/dom/WorkerRef.h"
  20 #include "mozilla/HoldDropJSObjects.h"
  21 #include "nsContentUtils.h"
  22 #include "nsDebug.h"
  23 #include "nsIAsyncInputStream.h"
  24 #include "nsIPipe.h"
  25 #include "nsIScriptError.h"
  26 #include "nsPIDOMWindow.h"
  27 #include "jsapi.h"
  28
  29 namespace mozilla::dom {
  30
  31 NS_IMPL_CYCLE_COLLECTING_ADDREF(FetchStreamReader)
  32 NS_IMPL_CYCLE_COLLECTING_RELEASE(FetchStreamReader)
  33
  34 NS_IMPL_CYCLE_COLLECTION(FetchStreamReader, mGlobal, mReader)
  35
  36 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(FetchStreamReader)
  37   NS_INTERFACE_MAP_ENTRY(nsIOutputStreamCallback)
  38   NS_INTERFACE_MAP_ENTRY(nsISupports)
  39 NS_INTERFACE_MAP_END
  40
  41 /* static */
  42 nsresult FetchStreamReader::Create(JSContext* aCx, nsIGlobalObject* aGlobal,
  43                                    FetchStreamReader** aStreamReader,
  44                                    nsIInputStream** aInputStream) {
  45   MOZ_ASSERT(aCx);
  46   MOZ_ASSERT(aGlobal);
  47   MOZ_ASSERT(aStreamReader);
  48   MOZ_ASSERT(aInputStream);
  49
  50   RefPtr<FetchStreamReader> streamReader = new FetchStreamReader(aGlobal);
  51
  52   nsCOMPtr<nsIAsyncInputStream> pipeIn;
  53
  54   NS_NewPipe2(getter_AddRefs(pipeIn), getter_AddRefs(streamReader->mPipeOut),
  55               true, true, 0, 0);
  56
  57   pipeIn.forget(aInputStream);
  58   streamReader.forget(aStreamReader);
  59   return NS_OK;
  60 }
  61
  62 nsresult FetchStreamReader::MaybeGrabStrongWorkerRef(JSContext* aCx) {
  63   if (NS_IsMainThread()) {
  64     return NS_OK;
  65   }
  66
  67   WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
  68   MOZ_ASSERT(workerPrivate);
  69
  70   RefPtr<StrongWorkerRef> workerRef = StrongWorkerRef::Create(
  71       workerPrivate, "FetchStreamReader", [streamReader = RefPtr(this)]() {
  72         MOZ_ASSERT(streamReader);
  73
  74         // mAsyncWaitWorkerRef may keep the (same) StrongWorkerRef alive even
  75         // when mWorkerRef has already been nulled out by a previous call to
  76         // CloseAndRelease, we can just safely ignore this callback then
  77         // (as would the CloseAndRelease do on a second call).
  78         if (streamReader->mWorkerRef) {
  79           streamReader->CloseAndRelease(
  80               streamReader->mWorkerRef->Private()->GetJSContext(),
  81               NS_ERROR_DOM_INVALID_STATE_ERR);
  82         } else {
  83           MOZ_DIAGNOSTIC_ASSERT(streamReader->mAsyncWaitWorkerRef);
  84         }
  85       });
  86
  87   if (NS_WARN_IF(!workerRef)) {
  88     return NS_ERROR_DOM_INVALID_STATE_ERR;
  89   }
  90
  91   // These 2 objects create a ref-cycle here that is broken when the stream is
  92   // closed or the worker shutsdown.
  93   mWorkerRef = std::move(workerRef);
  94
  95   return NS_OK;
  96 }
  97
  98 FetchStreamReader::FetchStreamReader(nsIGlobalObject* aGlobal)
  99     : mGlobal(aGlobal), mOwningEventTarget(mGlobal->SerialEventTarget()) {
 100   MOZ_ASSERT(aGlobal);
 101 }
 102
 103 FetchStreamReader::~FetchStreamReader() {
 104   CloseAndRelease(nullptr, NS_BASE_STREAM_CLOSED);
 105 }
 106
 107 // If a context is provided, an attempt will be made to cancel the reader.  The
 108 // only situation where we don't expect to have a context is when closure is
 109 // being triggered from the destructor or the WorkerRef is notifying.  If
 110 // we're at the destructor, it's far too late to cancel anything.  And if the
 111 // WorkerRef is being notified, the global is going away, so there's also
 112 // no need to do further JS work.
 113 void FetchStreamReader::CloseAndRelease(JSContext* aCx, nsresult aStatus) {
 114   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
 115
 116   if (mStreamClosed) {
 117     // Already closed.
 118     return;
 119   }
 120
 121   RefPtr<FetchStreamReader> kungFuDeathGrip = this;
 122   if (aCx && mReader) {
 123     ErrorResult rv;
 124     if (aStatus == NS_ERROR_DOM_WRONG_TYPE_ERR) {
 125       rv.ThrowTypeError<MSG_FETCH_BODY_WRONG_TYPE>();
 126     } else {
 127       rv = aStatus;
 128     }
 129     JS::Rooted<JS::Value> errorValue(aCx);
 130     if (ToJSValue(aCx, std::move(rv), &errorValue)) {
 131       IgnoredErrorResult ignoredError;
 132       // It's currently safe to cancel an already closed reader because, per the
 133       // comments in ReadableStream::cancel() conveying the spec, step 2 of
 134       // 3.4.3 that specified ReadableStreamCancel is: If stream.[[state]] is
 135       // "closed", return a new promise resolved with undefined.
 136       RefPtr<Promise> cancelResultPromise =
 137           MOZ_KnownLive(mReader)->Cancel(aCx, errorValue, ignoredError);
 138       NS_WARNING_ASSERTION(!ignoredError.Failed(),
 139                            "Failed to cancel stream during close and release");
 140       if (cancelResultPromise) {
 141         bool setHandled = cancelResultPromise->SetAnyPromiseIsHandled();
 142         NS_WARNING_ASSERTION(setHandled,
 143                              "Failed to mark cancel promise as handled.");
 144         (void)setHandled;
 145       }
 146     }
 147
 148     // We don't want to propagate exceptions during the cleanup.
 149     JS_ClearPendingException(aCx);
 150   }
 151
 152   mStreamClosed = true;
 153
 154   mGlobal = nullptr;
 155
 156   if (mPipeOut) {
 157     mPipeOut->CloseWithStatus(aStatus);
 158   }
 159   mPipeOut = nullptr;
 160
 161   mWorkerRef = nullptr;
 162
 163   mReader = nullptr;
 164   mBuffer.Clear();
 165 }
 166
 167 // https://fetch.spec.whatwg.org/#body-incrementally-read
 168 void FetchStreamReader::StartConsuming(JSContext* aCx, ReadableStream* aStream,
 169                                        ErrorResult& aRv) {
 170   MOZ_DIAGNOSTIC_ASSERT(!mReader);
 171   MOZ_DIAGNOSTIC_ASSERT(aStream);
 172   MOZ_ASSERT(!aStream->MaybeGetInputStreamIfUnread(),
 173              "FetchStreamReader is for JS streams but we got a stream based on "
 174              "nsIInputStream here. Extract nsIInputStream and read it instead "
 175              "to reduce overhead.");
 176
 177   aRv = MaybeGrabStrongWorkerRef(aCx);
 178   if (aRv.Failed()) {
 179     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 180     return;
 181   }
 182
 183   // Step 2: Let reader be the result of getting a reader for body’s stream.
 184   RefPtr<ReadableStreamDefaultReader> reader = aStream->GetReader(aRv);
 185   if (aRv.Failed()) {
 186     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 187     return;
 188   }
 189
 190   mReader = reader;
 191
 192   mAsyncWaitWorkerRef = mWorkerRef;
 193   aRv = mPipeOut->AsyncWait(this, 0, 0, mOwningEventTarget);
 194   if (NS_WARN_IF(aRv.Failed())) {
 195     mAsyncWaitWorkerRef = nullptr;
 196     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 197   }
 198 }
 199
 200 struct FetchReadRequest : public ReadRequest {
 201  public:
 202   NS_DECL_ISUPPORTS_INHERITED
 203   NS_DECL_CYCLE_COLLECTION_CLASS_INHERITED(FetchReadRequest, ReadRequest)
 204
 205   explicit FetchReadRequest(FetchStreamReader* aReader)
 206       : mFetchStreamReader(aReader) {}
 207
 208   MOZ_CAN_RUN_SCRIPT_BOUNDARY
 209   void ChunkSteps(JSContext* aCx, JS::Handle<JS::Value> aChunk,
 210                   ErrorResult& aRv) override {
 211     mFetchStreamReader->ChunkSteps(aCx, aChunk, aRv);
 212   }
 213
 214   MOZ_CAN_RUN_SCRIPT_BOUNDARY
 215   void CloseSteps(JSContext* aCx, ErrorResult& aRv) override {
 216     mFetchStreamReader->CloseSteps(aCx, aRv);
 217   }
 218
 219   MOZ_CAN_RUN_SCRIPT_BOUNDARY
 220   void ErrorSteps(JSContext* aCx, JS::Handle<JS::Value> aError,
 221                   ErrorResult& aRv) override {
 222     mFetchStreamReader->ErrorSteps(aCx, aError, aRv);
 223   }
 224
 225  protected:
 226   virtual ~FetchReadRequest() = default;
 227
 228   MOZ_KNOWN_LIVE RefPtr<FetchStreamReader> mFetchStreamReader;
 229 };
 230
 231 NS_IMPL_CYCLE_COLLECTION_INHERITED(FetchReadRequest, ReadRequest,
 232                                    mFetchStreamReader)
 233 NS_IMPL_ADDREF_INHERITED(FetchReadRequest, ReadRequest)
 234 NS_IMPL_RELEASE_INHERITED(FetchReadRequest, ReadRequest)
 235 NS_INTERFACE_MAP_BEGIN_CYCLE_COLLECTION(FetchReadRequest)
 236 NS_INTERFACE_MAP_END_INHERITING(ReadRequest)
 237
 238 // nsIOutputStreamCallback interface
 239 MOZ_CAN_RUN_SCRIPT_BOUNDARY
 240 NS_IMETHODIMP
 241 FetchStreamReader::OnOutputStreamReady(nsIAsyncOutputStream* aStream) {
 242   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
 243   if (mStreamClosed) {
 244     mAsyncWaitWorkerRef = nullptr;
 245     return NS_OK;
 246   }
 247
 248   AutoEntryScript aes(mGlobal, "ReadableStreamReader.read", !mWorkerRef);
 249   if (!Process(aes.cx())) {
 250     // We're done processing data, and haven't queued up a new AsyncWait - we
 251     // can clear our mAsyncWaitWorkerRef.
 252     mAsyncWaitWorkerRef = nullptr;
 253   }
 254   return NS_OK;
 255 }
 256
 257 bool FetchStreamReader::Process(JSContext* aCx) {
 258   NS_ASSERT_OWNINGTHREAD(FetchStreamReader);
 259   MOZ_ASSERT(mReader);
 260
 261   if (!mBuffer.IsEmpty()) {
 262     nsresult rv = WriteBuffer();
 263     if (NS_WARN_IF(NS_FAILED(rv))) {
 264       CloseAndRelease(aCx, NS_ERROR_DOM_ABORT_ERR);
 265       return false;
 266     }
 267     return true;
 268   }
 269
 270   // Check if the output stream has already been closed. This lets us propagate
 271   // errors eagerly, and detect output stream closures even when we have no data
 272   // to write.
 273   if (NS_WARN_IF(NS_FAILED(mPipeOut->StreamStatus()))) {
 274     CloseAndRelease(aCx, NS_ERROR_DOM_ABORT_ERR);
 275     return false;
 276   }
 277
 278   // We're waiting on new data - set up a WAIT_CLOSURE_ONLY callback so we
 279   // notice if the reader closes.
 280   nsresult rv = mPipeOut->AsyncWait(
 281       this, nsIAsyncOutputStream::WAIT_CLOSURE_ONLY, 0, mOwningEventTarget);
 282   if (NS_WARN_IF(NS_FAILED(rv))) {
 283     CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 284     return false;
 285   }
 286
 287   // If we already have an outstanding read request, don't start another one
 288   // concurrently.
 289   if (!mHasOutstandingReadRequest) {
 290     // https://fetch.spec.whatwg.org/#incrementally-read-loop
 291     // The below very loosely tries to implement the incrementally-read-loop
 292     // from the fetch spec.
 293     // Step 2: Read a chunk from reader given readRequest.
 294     RefPtr<ReadRequest> readRequest = new FetchReadRequest(this);
 295     RefPtr<ReadableStreamDefaultReader> reader = mReader;
 296     mHasOutstandingReadRequest = true;
 297
 298     IgnoredErrorResult err;
 299     reader->ReadChunk(aCx, *readRequest, err);
 300     if (NS_WARN_IF(err.Failed())) {
 301       // Let's close the stream.
 302       mHasOutstandingReadRequest = false;
 303       CloseAndRelease(aCx, NS_ERROR_DOM_INVALID_STATE_ERR);
 304       // Don't return false, as we've already called `AsyncWait`.
 305     }
 306   }
 307   return true;
 308 }
 309
 310 void FetchStreamReader::ChunkSteps(JSContext* aCx, JS::Handle<JS::Value> aChunk,
 311                                    ErrorResult& aRv) {
 312   // This roughly implements the chunk steps from
 313   // https://fetch.spec.whatwg.org/#incrementally-read-loop.
 314
 315   mHasOutstandingReadRequest = false;
 316
 317   // Step 2. If chunk is not a Uint8Array object, then set continueAlgorithm to
 318   // this step: run processBodyError given a TypeError.
 319   RootedSpiderMonkeyInterface<Uint8Array> chunk(aCx);
 320   if (!aChunk.isObject() || !chunk.Init(&aChunk.toObject())) {
 321     CloseAndRelease(aCx, NS_ERROR_DOM_WRONG_TYPE_ERR);
 322     return;
 323   }
 324
 325   MOZ_DIAGNOSTIC_ASSERT(mBuffer.IsEmpty());
 326
 327   // Let's take a copy of the data.
 328   // FIXME: We could sometimes avoid this copy by trying to write `chunk`
 329   // directly into `mPipeOut` eagerly, and only filling `mBuffer` if there isn't
 330   // enough space in the pipe's buffer.
 331   if (!chunk.AppendDataTo(mBuffer)) {
 332     CloseAndRelease(aCx, NS_ERROR_OUT_OF_MEMORY);
 333     return;
 334   }
 335
 336   mBufferOffset = 0;
 337   mBufferRemaining = mBuffer.Length();
 338
 339   nsresult rv = WriteBuffer();
 340   if (NS_WARN_IF(NS_FAILED(rv))) {
 341     CloseAndRelease(aCx, NS_ERROR_DOM_ABORT_ERR);
 342   }
 343 }
 344
 345 void FetchStreamReader::CloseSteps(JSContext* aCx, ErrorResult& aRv) {
 346   mHasOutstandingReadRequest = false;
 347   CloseAndRelease(aCx, NS_BASE_STREAM_CLOSED);
 348 }
 349
 350 void FetchStreamReader::ErrorSteps(JSContext* aCx, JS::Handle<JS::Value> aError,
 351                                    ErrorResult& aRv) {
 352   mHasOutstandingReadRequest = false;
 353   ReportErrorToConsole(aCx, aError);
 354   CloseAndRelease(aCx, NS_ERROR_FAILURE);
 355 }
 356
 357 nsresult FetchStreamReader::WriteBuffer() {
 358   MOZ_ASSERT(mBuffer.Length() == (mBufferOffset + mBufferRemaining));
 359
 360   char* data = reinterpret_cast<char*>(mBuffer.Elements());
 361
 362   while (mBufferRemaining > 0) {
 363     uint32_t written = 0;
 364     nsresult rv =
 365         mPipeOut->Write(data + mBufferOffset, mBufferRemaining, &written);
 366
 367     if (rv == NS_BASE_STREAM_WOULD_BLOCK) {
 368       break;
 369     }
 370
 371     if (NS_WARN_IF(NS_FAILED(rv))) {
 372       return rv;
 373     }
 374
 375     MOZ_ASSERT(written <= mBufferRemaining);
 376     mBufferRemaining -= written;
 377     mBufferOffset += written;
 378
 379     if (mBufferRemaining == 0) {
 380       mBuffer.Clear();
 381       break;
 382     }
 383   }
 384
 385   nsresult rv = mPipeOut->AsyncWait(this, 0, 0, mOwningEventTarget);
 386   if (NS_WARN_IF(NS_FAILED(rv))) {
 387     return rv;
 388   }
 389
 390   return NS_OK;
 391 }
 392
 393 void FetchStreamReader::ReportErrorToConsole(JSContext* aCx,
 394                                              JS::Handle<JS::Value> aValue) {
 395   nsCString sourceSpec;
 396   uint32_t line = 0;
 397   uint32_t column = 0;
 398   nsString valueString;
 399
 400   nsContentUtils::ExtractErrorValues(aCx, aValue, sourceSpec, &line, &column,
 401                                      valueString);
 402
 403   nsTArray<nsString> params;
 404   params.AppendElement(valueString);
 405
 406   RefPtr<ConsoleReportCollector> reporter = new ConsoleReportCollector();
 407   reporter->AddConsoleReport(nsIScriptError::errorFlag,
 408                              "ReadableStreamReader.read"_ns,
 409                              nsContentUtils::eDOM_PROPERTIES, sourceSpec, line,
 410                              column, "ReadableStreamReadingFailed"_ns, params);
 411
 412   uint64_t innerWindowId = 0;
 413
 414   if (NS_IsMainThread()) {
 415     nsCOMPtr<nsPIDOMWindowInner> window = do_QueryInterface(mGlobal);
 416     if (window) {
 417       innerWindowId = window->WindowID();
 418     }
 419     reporter->FlushReportsToConsole(innerWindowId);
 420     return;
 421   }
 422
 423   WorkerPrivate* workerPrivate = GetWorkerPrivateFromContext(aCx);
 424   if (workerPrivate) {
 425     innerWindowId = workerPrivate->WindowID();
 426   }
 427
 428   RefPtr<Runnable> r = NS_NewRunnableFunction(
 429       "FetchStreamReader::ReportErrorToConsole", [reporter, innerWindowId]() {
 430         reporter->FlushReportsToConsole(innerWindowId);
 431       });
 432
 433   workerPrivate->DispatchToMainThread(r.forget());
 434 }
 435
 436 }  // namespace mozilla::dom