Add latest changes from gitlab-org/gitlab@master

8 files changed, 380 insertions, 0 deletions

diff --git a/app/assets/javascripts/streaming/chunk_writer.js b/app/assets/javascripts/streaming/chunk_writer.js
new file mode 100644
index 00000000000..4bbd0a5f843
--- /dev/null
+++ b/app/assets/javascripts/streaming/chunk_writer.js
@@ -0,0 +1,144 @@
+import { throttle } from 'lodash';
+import { RenderBalancer } from '~/streaming/render_balancer';
+import {
+  BALANCE_RATE,
+  HIGH_FRAME_TIME,
+  LOW_FRAME_TIME,
+  MAX_CHUNK_SIZE,
+  MIN_CHUNK_SIZE,
+  TIMEOUT,
+} from '~/streaming/constants';
+
+const defaultConfig = {
+  balanceRate: BALANCE_RATE,
+  minChunkSize: MIN_CHUNK_SIZE,
+  maxChunkSize: MAX_CHUNK_SIZE,
+  lowFrameTime: LOW_FRAME_TIME,
+  highFrameTime: HIGH_FRAME_TIME,
+  timeout: TIMEOUT,
+};
+
+function concatUint8Arrays(a, b) {
+  const array = new Uint8Array(a.length + b.length);
+  array.set(a, 0);
+  array.set(b, a.length);
+  return array;
+}
+
+// This class is used to write chunks with a balanced size
+// to avoid blocking main thread for too long.
+//
+// A chunk can be:
+//   1. Too small
+//   2. Too large
+//   3. Delayed in time
+//
+// This class resolves all these problems by
+//   1. Splitting or concatenating chunks to met the size criteria
+//   2. Rendering current chunk buffer immediately if enough time has passed
+//
+// The size of the chunk is determined by RenderBalancer,
+// It measures execution time for each chunk write and adjusts next chunk size.
+export class ChunkWriter {
+  buffer = null;
+  decoder = new TextDecoder('utf-8');
+  timeout = null;
+
+  constructor(htmlStream, config) {
+    this.htmlStream = htmlStream;
+
+    const { balanceRate, minChunkSize, maxChunkSize, lowFrameTime, highFrameTime, timeout } = {
+      ...defaultConfig,
+      ...config,
+    };
+
+    // ensure we still render chunks over time if the size criteria is not met
+    this.scheduleAccumulatorFlush = throttle(this.flushAccumulator.bind(this), timeout);
+
+    const averageSize = Math.round((maxChunkSize + minChunkSize) / 2);
+    this.size = Math.max(averageSize, minChunkSize);
+
+    this.balancer = new RenderBalancer({
+      lowFrameTime,
+      highFrameTime,
+      decrease: () => {
+        this.size = Math.round(Math.max(this.size / balanceRate, minChunkSize));
+      },
+      increase: () => {
+        this.size = Math.round(Math.min(this.size * balanceRate, maxChunkSize));
+      },
+    });
+  }
+
+  write(chunk) {
+    this.scheduleAccumulatorFlush.cancel();
+
+    if (this.buffer) {
+      this.buffer = concatUint8Arrays(this.buffer, chunk);
+    } else {
+      this.buffer = chunk;
+    }
+
+    // accumulate chunks until the size is fulfilled
+    if (this.size > this.buffer.length) {
+      this.scheduleAccumulatorFlush();
+      return Promise.resolve();
+    }
+
+    return this.balancedWrite();
+  }
+
+  balancedWrite() {
+    let cursor = 0;
+
+    return this.balancer.render(() => {
+      const chunkPart = this.buffer.subarray(cursor, cursor + this.size);
+      // accumulate chunks until the size is fulfilled
+      // this is a hot path for the last chunkPart of the chunk
+      if (chunkPart.length < this.size) {
+        this.buffer = chunkPart;
+        this.scheduleAccumulatorFlush();
+        return false;
+      }
+
+      this.writeToDom(chunkPart);
+
+      cursor += this.size;
+      if (cursor >= this.buffer.length) {
+        this.buffer = null;
+        return false;
+      }
+      // continue render
+      return true;
+    });
+  }
+
+  writeToDom(chunk, stream = true) {
+    // stream: true allows us to split chunks with multi-part words
+    const decoded = this.decoder.decode(chunk, { stream });
+    this.htmlStream.write(decoded);
+  }
+
+  flushAccumulator() {
+    if (this.buffer) {
+      this.writeToDom(this.buffer);
+      this.buffer = null;
+    }
+  }
+
+  close() {
+    this.scheduleAccumulatorFlush.cancel();
+    if (this.buffer) {
+      // last chunk should have stream: false to indicate the end of the stream
+      this.writeToDom(this.buffer, false);
+      this.buffer = null;
+    }
+    this.htmlStream.close();
+  }
+
+  abort() {
+    this.scheduleAccumulatorFlush.cancel();
+    this.buffer = null;
+    this.htmlStream.abort();
+  }
+}
diff --git a/app/assets/javascripts/streaming/constants.js b/app/assets/javascripts/streaming/constants.js
new file mode 100644
index 00000000000..224d93a7ac1
--- /dev/null
+++ b/app/assets/javascripts/streaming/constants.js
@@ -0,0 +1,9 @@
+// Lower min chunk numbers can make the page loading take incredibly long
+export const MIN_CHUNK_SIZE = 128 * 1024;
+export const MAX_CHUNK_SIZE = 2048 * 1024;
+export const LOW_FRAME_TIME = 32;
+// Tasks that take more than 50ms are considered Long
+// https://web.dev/optimize-long-tasks/
+export const HIGH_FRAME_TIME = 64;
+export const BALANCE_RATE = 1.2;
+export const TIMEOUT = 500;
diff --git a/app/assets/javascripts/streaming/handle_streamed_anchor_link.js b/app/assets/javascripts/streaming/handle_streamed_anchor_link.js
new file mode 100644
index 00000000000..315dc9bb0a0
--- /dev/null
+++ b/app/assets/javascripts/streaming/handle_streamed_anchor_link.js
@@ -0,0 +1,26 @@
+import { throttle } from 'lodash';
+import { scrollToElement } from '~/lib/utils/common_utils';
+import LineHighlighter from '~/blob/line_highlighter';
+
+const noop = () => {};
+
+export function handleStreamedAnchorLink(rootElement) {
+  // "#L100-200" → ['L100', 'L200']
+  const [anchorStart, end] = window.location.hash.substring(1).split('-');
+  const anchorEnd = end ? `L${end}` : anchorStart;
+  if (!anchorStart || document.getElementById(anchorEnd)) return noop;
+
+  const handler = throttle((mutationList, instance) => {
+    if (!document.getElementById(anchorEnd)) return;
+    scrollToElement(document.getElementById(anchorStart));
+    // eslint-disable-next-line no-new
+    new LineHighlighter();
+    instance.disconnect();
+  }, 300);
+
+  const observer = new MutationObserver(handler);
+
+  observer.observe(rootElement, { childList: true, subtree: true });
+
+  return () => observer.disconnect();
+}
diff --git a/app/assets/javascripts/streaming/html_stream.js b/app/assets/javascripts/streaming/html_stream.js
new file mode 100644
index 00000000000..8182f69a607
--- /dev/null
+++ b/app/assets/javascripts/streaming/html_stream.js
@@ -0,0 +1,33 @@
+import { ChunkWriter } from '~/streaming/chunk_writer';
+
+export class HtmlStream {
+  constructor(element) {
+    const streamDocument = document.implementation.createHTMLDocument('stream');
+
+    streamDocument.open();
+    streamDocument.write('<streaming-element>');
+
+    const virtualStreamingElement = streamDocument.querySelector('streaming-element');
+    element.appendChild(document.adoptNode(virtualStreamingElement));
+
+    this.streamDocument = streamDocument;
+  }
+
+  withChunkWriter(config) {
+    return new ChunkWriter(this, config);
+  }
+
+  write(chunk) {
+    // eslint-disable-next-line no-unsanitized/method
+    this.streamDocument.write(chunk);
+  }
+
+  close() {
+    this.streamDocument.write('</streaming-element>');
+    this.streamDocument.close();
+  }
+
+  abort() {
+    this.streamDocument.close();
+  }
+}
diff --git a/app/assets/javascripts/streaming/polyfills.js b/app/assets/javascripts/streaming/polyfills.js
new file mode 100644
index 00000000000..a9a044a3e99
--- /dev/null
+++ b/app/assets/javascripts/streaming/polyfills.js
@@ -0,0 +1,5 @@
+import { createReadableStreamWrapper } from '@mattiasbuelens/web-streams-adapter';
+import { ReadableStream as PolyfillReadableStream } from 'web-streams-polyfill';
+
+// TODO: remove this when our WebStreams API reaches 100% support
+export const toPolyfillReadable = createReadableStreamWrapper(PolyfillReadableStream);
diff --git a/app/assets/javascripts/streaming/rate_limit_stream_requests.js b/app/assets/javascripts/streaming/rate_limit_stream_requests.js
new file mode 100644
index 00000000000..04a592baa16
--- /dev/null
+++ b/app/assets/javascripts/streaming/rate_limit_stream_requests.js
@@ -0,0 +1,87 @@
+const consumeReadableStream = (stream) => {
+  return new Promise((resolve, reject) => {
+    stream.pipeTo(
+      new WritableStream({
+        close: resolve,
+        abort: reject,
+      }),
+    );
+  });
+};
+
+const wait = (timeout) =>
+  new Promise((resolve) => {
+    setTimeout(resolve, timeout);
+  });
+
+// this rate-limiting approach is specific to Web Streams
+// because streams only resolve when they're fully consumed
+// so we need to split each stream into two pieces:
+//   one for the rate-limiter (wait for all the bytes to be sent)
+//   another for the original consumer
+export const rateLimitStreamRequests = ({
+  factory,
+  total,
+  maxConcurrentRequests,
+  immediateCount = maxConcurrentRequests,
+  timeout = 0,
+}) => {
+  if (total === 0) return [];
+
+  const unsettled = [];
+
+  const pushUnsettled = (promise) => {
+    let res;
+    let rej;
+    const consume = new Promise((resolve, reject) => {
+      res = resolve;
+      rej = reject;
+    });
+    unsettled.push(consume);
+    return promise.then((stream) => {
+      const [first, second] = stream.tee();
+      // eslint-disable-next-line promise/no-nesting
+      consumeReadableStream(first)
+        .then(() => {
+          unsettled.splice(unsettled.indexOf(consume), 1);
+          res();
+        })
+        .catch(rej);
+      return second;
+    }, rej);
+  };
+
+  const immediate = Array.from({ length: Math.min(immediateCount, total) }, (_, i) =>
+    pushUnsettled(factory(i)),
+  );
+
+  const queue = [];
+  const flushQueue = () => {
+    const promises =
+      unsettled.length > maxConcurrentRequests ? unsettled : [...unsettled, wait(timeout)];
+    // errors are handled by the caller
+    // eslint-disable-next-line promise/catch-or-return
+    Promise.race(promises).then(() => {
+      const cb = queue.shift();
+      cb?.();
+      if (queue.length !== 0) {
+        // wait for stream consumer promise to be removed from unsettled
+        queueMicrotask(flushQueue);
+      }
+    });
+  };
+
+  const throttled = Array.from({ length: total - immediateCount }, (_, i) => {
+    return new Promise((resolve, reject) => {
+      queue.push(() => {
+        pushUnsettled(factory(i + immediateCount))
+          .then(resolve)
+          .catch(reject);
+      });
+    });
+  });
+
+  flushQueue();
+
+  return [...immediate, ...throttled];
+};
diff --git a/app/assets/javascripts/streaming/render_balancer.js b/app/assets/javascripts/streaming/render_balancer.js
new file mode 100644
index 00000000000..66929ff3a54
--- /dev/null
+++ b/app/assets/javascripts/streaming/render_balancer.js
@@ -0,0 +1,36 @@
+export class RenderBalancer {
+  previousTimestamp = undefined;
+
+  constructor({ increase, decrease, highFrameTime, lowFrameTime }) {
+    this.increase = increase;
+    this.decrease = decrease;
+    this.highFrameTime = highFrameTime;
+    this.lowFrameTime = lowFrameTime;
+  }
+
+  render(fn) {
+    return new Promise((resolve) => {
+      const callback = (timestamp) => {
+        this.throttle(timestamp);
+        if (fn()) requestAnimationFrame(callback);
+        else resolve();
+      };
+      requestAnimationFrame(callback);
+    });
+  }
+
+  throttle(timestamp) {
+    const { previousTimestamp } = this;
+    this.previousTimestamp = timestamp;
+    if (previousTimestamp === undefined) return;
+
+    const duration = Math.round(timestamp - previousTimestamp);
+    if (!duration) return;
+
+    if (duration >= this.highFrameTime) {
+      this.decrease();
+    } else if (duration < this.lowFrameTime) {
+      this.increase();
+    }
+  }
+}
diff --git a/app/assets/javascripts/streaming/render_html_streams.js b/app/assets/javascripts/streaming/render_html_streams.js
new file mode 100644
index 00000000000..7201e541777
--- /dev/null
+++ b/app/assets/javascripts/streaming/render_html_streams.js
@@ -0,0 +1,40 @@
+import { HtmlStream } from '~/streaming/html_stream';
+
+async function pipeStreams(domWriter, streamPromises) {
+  try {
+    for await (const stream of streamPromises.slice(0, -1)) {
+      await stream.pipeTo(domWriter, { preventClose: true });
+    }
+    const stream = await streamPromises[streamPromises.length - 1];
+    await stream.pipeTo(domWriter);
+  } catch (error) {
+    domWriter.abort(error);
+  }
+}
+
+// this function (and the rest of the pipeline) expects polyfilled streams
+// do not pass native streams here unless our browser support allows for it
+// TODO: remove this notice when our WebStreams API support reaches 100%
+export function renderHtmlStreams(streamPromises, element, config) {
+  if (streamPromises.length === 0) return Promise.resolve();
+
+  const chunkedHtmlStream = new HtmlStream(element).withChunkWriter(config);
+
+  return new Promise((resolve, reject) => {
+    const domWriter = new WritableStream({
+      write(chunk) {
+        return chunkedHtmlStream.write(chunk);
+      },
+      close() {
+        chunkedHtmlStream.close();
+        resolve();
+      },
+      abort(error) {
+        chunkedHtmlStream.abort();
+        reject(error);
+      },
+    });
+
+    pipeStreams(domWriter, streamPromises);
+  });
+}