nginx: buffer responses in nginx, not application layer

files/nginx/odk.conf.template

@@ -209,9 +209,8 @@ server {
     proxy_pass http://service:8383;
     proxy_redirect off;
 
-    # buffer requests, but not responses, so streaming out works.
     proxy_request_buffering on;
-    proxy_buffering off;
+    proxy_buffering on;
     proxy_read_timeout 2m;
   }
 

test/nginx/mock-http-server/index.js

@@ -1,9 +1,13 @@
+const { Readable } = require('node:stream');
+
 const express = require('express');
 
 const port = process.env.PORT || 80;
 const log = (...args) => console.log('[mock-http-server]', ...args);
 
 const requests = [];
+let openProcessorCount = 0;
+let completedProcessorCount = 0;
 
 const app = express();
 app.set('case sensitive routing', true);
@@ -29,9 +33,57 @@ app.get('/__mock_http_server/health',      (req, res) => res.send('OK'));
 app.get('/__mock_http_server/request-log', (req, res) => res.json(requests));
 app.get('/__mock_http_server/reset',       (req, res) => {
   requests.length = 0;
+  openProcessorCount = 0;
+  completedProcessorCount = 0;
   res.json('OK');
 });
 
+app.get(new RegExp('^/v1/.*/100MB\\.csv$'), (req, res) => {
+  const csvSizeBytes = 100_000_000;
+
+  res.set('Content-Disposition', `attachment; filename="100MB.csv"; filename*=UTF-8''100MB.csv`);
+  res.set('Content-Type', 'text/csv; charset=utf-8');
+
+  ++openProcessorCount;
+
+  async function* generateCsv(targetByteLength) {
+    let rowCount = 0;
+    let totalWritten = 0;
+
+    const batchSize = Math.pow(2, 18);
+
+    const header = Buffer.from('row_number,timestamp,random-number\n', 'utf8');
+    totalWritten += header.byteLength;
+    yield header;
+
+    while(totalWritten < targetByteLength) {
+      await new Promise(resolve => setTimeout(resolve, 1));
+
+      const batch = Buffer.allocUnsafe(Math.min(batchSize, targetByteLength - totalWritten));
+      let bufpos = 0;
+      while(bufpos < batch.length) {
+        const line = `${++rowCount},${new Date().toISOString()},${Math.random()}\n`;
+        const bytesWritten = batch.write(line, bufpos, batch.length-bufpos, 'utf8');
+        bufpos += bytesWritten;
+        totalWritten += bytesWritten;
+      }
+      yield batch;
+    }
+
+    ++completedProcessorCount;
+  }
+
+  const randomStream = Readable.from(generateCsv(csvSizeBytes));
+  randomStream.pipe(res);
+  req.on('close', () => {
+    randomStream.destroy();
+    --openProcessorCount;
+  });
+});
+app.get('/__mock_http_server/open-processor-count', (req, res) => {
+  res.send({ openProcessorCount, completedProcessorCount });
+});
+
 app.get('/v1/reflect-headers', (req, res) => res.json(req.headers));
 
 // Central-Backend can set Cache headers and those should have highest precedence

test/nginx/src/lib.js

@@ -10,6 +10,7 @@ module.exports = {
   assertSentryReceived,
   requestSentryMock,
   resetSentryMock,
+  sleep,
 };
 
 async function assertSentryReceived(...expectedRequests) {
@@ -56,3 +57,7 @@ function requestSentryMock(opts) {
     req.end();
   });
 }
+
+function sleep(ms) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}

test/nginx/src/mocha/nginx.spec.js

@@ -5,6 +5,7 @@ const {
   assertSentryReceived,
   requestSentryMock,
   resetSentryMock,
+  sleep,
 } = require('../lib');
 const request = require('./request');
 
@@ -436,6 +437,82 @@ function standardTestSuite({ fetchHttp, fetchHttp6, apiFetch, apiFetch6, forward
     });
   });
 
+  describe('response buffering', () => {
+    it('should buffer responses in nginx, not backend services', async function() {
+      // NOTE the final check should pass before the test times out.  If timeout
+      // is reached, there may be back-pressure on the NodeJS server.  This
+      // would imply that nginx buffering is not working correctly.
+      // !!! ONLY CHANGE THIS TIMEOUT IF THE ABOVE COMMENT IS WELL UNDERSTOOD !!!
+      const testTimeout = 5_000;
+      this.timeout(testTimeout);
+
+      let controller;
+
+      try {
+        // given
+        controller = new AbortController();
+        const { signal } = controller;
+
+        // when
+        const res = await apiFetch('/v1/projects/123/forms/some_form_id/attachments/100MB.csv', { signal });
+        // then
+        assert.equal(res.status, 200);
+
+        // when
+        const reader = res.body.getReader();
+        const initialRead = await reader.read();
+        let bytesRead = initialRead.value.length;
+        // then
+        assert.isFalse(initialRead.done);
+        assert.isAtMost(bytesRead, 16_384);
+        assert.equal(new TextDecoder('utf8').decode(initialRead.value).split('\n', 1)[0], 'row_number,timestamp,random-number');
+        // and
+        assert.deepEqual(await getOpenProcessorCount(), { openProcessorCount:1, completedProcessorCount:0 });
+
+        // when
+        await untilOpenProcessorCountIs({ timeout:testTimeout, openProcessorCount:0, completedProcessorCount:1 });
+        // and
+        while(true) {
+          const { done, value } = await reader.read();
+          if(done) break;
+          bytesRead += value.length;
+        }
+        // then
+        assert.equal(bytesRead, 100_000_000);
+      } finally {
+        controller.abort();
+      }
+    });
+
+    async function getOpenProcessorCount() {
+      const res = await request(`http://localhost:8383/__mock_http_server/open-processor-count`);
+      assert.isTrue(res.ok);
+      return await res.json();
+    }
+
+    async function untilOpenProcessorCountIs({ timeout, ...expected }) {
+      let timeoutId;
+      try {
+        let timedOut;
+        timeoutId = setTimeout(() => { timedOut = true; }, timeout);
+
+        while(true) {
+          const { openProcessorCount, completedProcessorCount } = await getOpenProcessorCount();
+          if(openProcessorCount === expected.openProcessorCount &&
+              completedProcessorCount === expected.completedProcessorCount) {
+            break;
+          }
+
+          if(timedOut) throw new Error(`Timeout of ${timeout} ms exceeded.`);
+
+          await sleep(100);
+        }
+      } finally {
+        clearTimeout(timeoutId);
+      }
+    }
+  });
+
   it('should serve generated client-config.json', async () => {
     // when
     const res = await apiFetch('/client-config.json');