moshpit: stream bulk publishes instead of buffering them (#419)

Publishing a lot at once meant sending a lot at once, and the batch
endpoint reads the whole request before the handler runs. Its ceiling is
therefore however much we are willing to hold in memory for whoever
asks, so "let us publish more" and "do not let anyone exhaust the
daemon" pulled in opposite directions.

Streaming removes the trade rather than picking a side. NDJSON is one
item per line: the parser holds one line, the handler writes it, and a
500-item upload costs the same memory as a 1-item one. The cap can then
be about total work instead of about a single allocation.

POST /api/moshpit/sites/:name/content/stream takes application/x-ndjson
and answers in NDJSON as it goes:

  {"type":"accepted","limits":{"items":500,...}}
  {"type":"progress","index":1,"ok":true,"created":true,"slug":"..."}
  {"type":"done","items":500,"created":500,"failed":0,"ms":1605}

The first progress line arrives long before the last item is uploaded,
which is what makes a progress bar possible at all. /pit/publish is a
page that does this with a file picker; it hands the File straight to
fetch, so the browser streams it from disk and the tab does not hold the
payload either.

Every cap bounds a distinct way of abusing a streaming endpoint, which
is a better DoS target than a buffered one because it holds a socket for
as long as a client keeps trickling:

- 500 items, equal to a name's whole capacity, so one upload can fill a
  site and cannot do more than fill it
- a per-line cap, checked after complete lines are drained, so a body
  with no newline in it cannot grow without bound under the total cap
- a total-byte cap
- a 20s idle timeout, rearmed on data, so a slow-but-live upload
  survives and a slowloris does not
- four concurrent streams process-wide, then 503 with Retry-After
  rather than a queue, because a queue is a slower way to run out of
  memory

A bad line costs that line, not the stream. A cap hit mid-stream cannot
be a status code -- the 200 went out before the first item was written
-- so it arrives as a final error line and everything already written
stays written.

The vhost needed its own block for this. nginx buffers a request body to
disk before contacting the app by default, which would have turned
"items land while the rest uploads" into one long silence followed by
everything at once, and its 2m body cap would have refused the upload
outright.

Verified in a real browser over CDP, not just in tests: 300 items from a
file picker drove the bar 0% -> 33% -> 62% -> 100%, 300 created in
1738ms, no console errors. A 500-item curl upload showed the first
progress line at 70ms against a 1667ms upload. 575/575 apps/pwa and
2377/2382 root tests pass.

Author: ralyodio

apps/pwa/deploy/README.md

@@ -80,6 +80,13 @@ are in the right order already.
   documented-length posts (`MAX_PUBLISH_BYTES`, ~1.9M) and the vhost allows 2m
   so nginx is never the one saying no. Past that you get a 413 in JSON telling
   you to split the batch — safe to do, because publishing upserts on the slug.
+  For a genuinely large import use the streaming endpoint instead, which does
+  not buffer at all and has its own vhost block.
+- **Bulk upload appears to hang, then finishes all at once.** That is nginx
+  buffering, and it means the `content/stream` location did not take. The
+  endpoint streams both ways; `proxy_request_buffering off` and
+  `proxy_buffering off` are what let items land while the rest is still
+  uploading. `nginx -T | grep -A5 content/stream` shows whether it is live.
 
 ## Verifying, one layer at a time
 

apps/pwa/deploy/nginx-vhost.conf

@@ -70,6 +70,43 @@ server {
         proxy_send_timeout    30s;
     }
 
+    # Bulk publish. Everything about this location exists because the endpoint
+    # streams in both directions, and nginx's defaults quietly undo that.
+    location ~ ^/api/moshpit/sites/[^/]+/content/stream$ {
+        proxy_pass http://127.0.0.1:@APP_PORT@;
+
+        proxy_set_header Host              $host;
+        proxy_set_header X-Real-IP         $remote_addr;
+        proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+
+        # The upload is NDJSON read a line at a time, so the app never holds
+        # more than one item. This is the cap on the whole upload rather than
+        # on what fits in memory — the app enforces its own item and byte
+        # limits underneath it.
+        client_max_body_size 128m;
+
+        # THE important one. nginx buffers a request body to disk by default and
+        # only then contacts the app, which would turn "items land while the
+        # rest uploads" into one long silence followed by everything at once —
+        # the exact behaviour the streaming endpoint exists to avoid.
+        proxy_request_buffering off;
+        # The same, for the progress lines coming back. The app also sends
+        # X-Accel-Buffering: no, so this is belt and braces for an nginx that
+        # was built to ignore the header.
+        proxy_buffering off;
+        # Chunked upload needs HTTP/1.1 upstream; the default is 1.0, which
+        # cannot express a body of unknown length.
+        proxy_http_version 1.1;
+
+        # A big import is legitimately slow. The progress lines keep the
+        # response side busy, but the request side can be quiet while a client
+        # on a poor uplink sends one item, so both directions get room. The
+        # app's own idle timeout is the tighter, smarter limit.
+        proxy_read_timeout    10m;
+        proxy_send_timeout    10m;
+    }
+
     # Cheap liveness probe that does not touch the database.
     location = /healthz {
         proxy_pass http://127.0.0.1:@APP_PORT@;

apps/pwa/src/lib/moshpit-publish-stream.mjs

@@ -0,0 +1,210 @@
+// Publishing a lot at once, without holding it all in memory.
+//
+// The JSON batch endpoint buffers the whole request before the handler runs, so
+// the most it can accept is the most we are willing to hold per request — and
+// that ceiling is set by what a hostile caller could do, not by what a real one
+// needs. Raising it to fit a real bulk import means agreeing to buffer that
+// much for anyone who asks.
+//
+// Streaming removes the trade. NDJSON is one item per line, so the parser holds
+// one line, the handler writes it, and the memory cost of a 500-item upload is
+// the same as a 1-item one. The cap can then be about the daemon's total work
+// rather than about a single allocation.
+//
+// The caps below exist because a streaming endpoint is a better DoS target than
+// a buffered one, not a worse one: it accepts an open socket for as long as the
+// client keeps trickling bytes. Every one of them bounds a different attack —
+// total bytes, a single unbounded line, item count, idle time, and how many of
+// these can run at once.
+
+import { MAX_BODY, MAX_ITEMS_PER_NAME, MAX_TITLE } from "./moshpit-content.mjs";
+
+/** The wire format: one JSON object per line. */
+export const STREAM_CONTENT_TYPES = ["application/x-ndjson", "application/ndjson", "application/jsonl"];
+
+/**
+ * The most items one stream may carry.
+ *
+ * Equal to a name's whole capacity on purpose: one upload can fill a site and
+ * cannot do more than fill it, so there is no batch size that accomplishes
+ * something a single stream could not.
+ */
+export const MAX_STREAM_ITEMS = MAX_ITEMS_PER_NAME;
+
+/**
+ * The most one line may weigh.
+ *
+ * An item's fields are already bounded (MAX_BODY + MAX_TITLE and a gallery of
+ * URLs), and JSON escaping can widen a character to six bytes. This is that,
+ * rounded up — enough that no legitimate item is refused, small enough that a
+ * single line can never be an unbounded allocation.
+ */
+export const MAX_LINE_BYTES = (MAX_BODY + MAX_TITLE) * 6 + 64 * 1024;
+
+/** The most a whole stream may weigh, however it is split into lines. */
+export const MAX_STREAM_BYTES = MAX_STREAM_ITEMS * MAX_LINE_BYTES;
+
+/**
+ * How long the stream may go without producing bytes.
+ *
+ * This is the slowloris guard. Without it a client can hold a connection and a
+ * database handle open indefinitely by sending one byte a minute, which costs
+ * them nothing and costs us a slot. Generous enough for a slow uplink mid-item.
+ */
+export const STREAM_IDLE_MS = 20_000;
+
+/**
+ * How many of these may run at once, process-wide.
+ *
+ * The real limit on concurrent bulk imports is the database behind them, not
+ * the sockets. Past this the answer is 503 with a Retry-After rather than
+ * queueing, because a queue is just a slower way to run out of memory.
+ */
+export const MAX_CONCURRENT_STREAMS = 4;
+
+/** A cap was hit. Carries the status the route should report. */
+export class StreamLimitError extends Error {
+  constructor(code, message, status = 413) {
+    super(message);
+    this.name = "StreamLimitError";
+    this.code = code;
+    this.status = status;
+  }
+}
+
+let active = 0;
+
+/**
+ * Take one of the concurrent-stream slots, or null when they are all taken.
+ *
+ * Returns the release function rather than a boolean so a caller cannot
+ * acquire and forget which slot to give back — the only thing to hold onto is
+ * the thing that frees it.
+ */
+export function acquireStreamSlot() {
+  if (active >= MAX_CONCURRENT_STREAMS) return null;
+  active += 1;
+  let released = false;
+  return () => {
+    // Idempotent: the route releases in a finally block that can run after an
+    // error path already released, and double-decrementing would hand out slots
+    // that do not exist.
+    if (released) return;
+    released = true;
+    active -= 1;
+  };
+}
+
+/** Live count, for tests and for the 503's message. */
+export function activeStreams() {
+  return active;
+}
+
+/** Test seam — drops any leaked slots between cases. */
+export function resetStreamSlots() {
+  active = 0;
+}
+
+export function isStreamContentType(header) {
+  const type = String(header || "").split(";")[0].trim().toLowerCase();
+  return STREAM_CONTENT_TYPES.includes(type);
+}
+
+/**
+ * Parse an NDJSON request body into items, one at a time.
+ *
+ * Yields `{ index, item }` as each line completes, so the caller can write it
+ * and report progress before the next line has even arrived. Blank lines are
+ * skipped — a trailing newline is the normal way to end a file, and treating it
+ * as an empty item would fail every upload written by a well-behaved tool.
+ *
+ * A malformed line is yielded as `{ index, error }` rather than thrown. One bad
+ * line in a bulk import should cost that line, not the several hundred good
+ * ones already written before it — the same reasoning the batch endpoint's 207
+ * follows.
+ */
+export async function* ndjsonItems(readable, {
+  maxItems = MAX_STREAM_ITEMS,
+  maxLineBytes = MAX_LINE_BYTES,
+  maxBytes = MAX_STREAM_BYTES,
+  idleMs = STREAM_IDLE_MS,
+} = {}) {
+  const decoder = new TextDecoder("utf-8");
+  let buffered = "";
+  let bytes = 0;
+  let index = 0;
+
+  let timer = null;
+  let idle = null;
+  const arm = () => {
+    clearTimeout(timer);
+    timer = setTimeout(() => {
+      idle = new StreamLimitError("idle", `no data for ${idleMs}ms`, 408);
+      // Ends the for-await below. Destroying is the point: an idle client is
+      // holding a slot, so the socket goes with it.
+      readable.destroy?.(idle);
+    }, idleMs);
+  };
+
+  // One line, validated and parsed. Returns null for a blank line.
+  const take = (line) => {
+    const text = line.trim();
+    if (!text) return null;
+    index += 1;
+    if (index > maxItems) {
+      throw new StreamLimitError("too_many_items", `publish up to ${maxItems} items in one stream`);
+    }
+    try {
+      const parsed = JSON.parse(text);
+      if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) {
+        return { index, error: "each line must be a JSON object" };
+      }
+      return { index, item: parsed };
+    } catch {
+      return { index, error: "that line is not valid JSON" };
+    }
+  };
+
+  arm();
+  try {
+    for await (const chunk of readable) {
+      arm();
+
+      bytes += chunk.length;
+      if (bytes > maxBytes) {
+        throw new StreamLimitError("too_large", `a stream may carry up to ${maxBytes} bytes`);
+      }
+
+      buffered += decoder.decode(chunk, { stream: true });
+
+      let nl = buffered.indexOf("\n");
+      while (nl !== -1) {
+        const line = buffered.slice(0, nl);
+        buffered = buffered.slice(nl + 1);
+        const out = take(line);
+        if (out) yield out;
+        nl = buffered.indexOf("\n");
+      }
+
+      // Checked after draining complete lines, so this only ever measures an
+      // unterminated one. Without it a client can send bytes forever with no
+      // newline and grow `buffered` without bound — under the total cap the
+      // whole time.
+      if (Buffer.byteLength(buffered, "utf8") > maxLineBytes) {
+        throw new StreamLimitError("line_too_large", `one item may be up to ${maxLineBytes} bytes`);
+      }
+    }
+
+    buffered += decoder.decode();
+    const last = take(buffered);
+    if (last) yield last;
+  } catch (err) {
+    // `readable.destroy(idle)` surfaces here as the error we constructed, but
+    // some streams report an aborted read as a generic premature-close instead.
+    // Reporting a timeout as a parse failure would send someone looking at
+    // their JSON, so the idle flag wins over whatever the stream said.
+    throw idle ?? err;
+  } finally {
+    clearTimeout(timer);
+  }
+}