Handle inventory stream backpressure

destiny2/destiny2.routes.js

@@ -12,6 +12,36 @@ import toTemporalInstant from '../helpers/to-temporal-instant.js';
 
 import configuration from '../helpers/config.js';
 
+async function writeChunk(res, chunk) {
+    if (res.writableEnded || res.destroyed) {
+        return false;
+    }
+
+    if (res.write(chunk)) {
+        return true;
+    }
+
+    await new Promise(resolve => {
+        const handleDrain = () => {
+            cleanup();
+            resolve();
+        };
+        const handleClose = () => {
+            cleanup();
+            resolve();
+        };
+        const cleanup = () => {
+            res.off('drain', handleDrain);
+            res.off('close', handleClose);
+        };
+
+        res.on('drain', handleDrain);
+        res.on('close', handleClose);
+    });
+
+    return !res.writableEnded && !res.destroyed;
+}
+
 /**
  * @openapi
  *  components:
@@ -124,13 +154,28 @@ const routes = ({ authenticationController, destiny2Controller }) => {
                     return res.end();
                 }
 
-                res.write(first ? `[${JSON.stringify(item)}` : `,${JSON.stringify(item)}`);
+                const canContinue = await writeChunk(
+                    res,
+                    first ? `[${JSON.stringify(item)}` : `,${JSON.stringify(item)}`,
+                );
+
+                if (!canContinue) {
+                    log.info(
+                        `${req.method} ${req.url} request closed while streaming item ${index} of ${items.length}`,
+                    );
+                    return res.end();
+                }
                 first = false;
 
                 // Introduce a delay to allow the event loop to process the 'close' event
                 await new Promise(resolve => setImmediate(resolve));
             }
-            res.write(']');
+
+            if (!(await writeChunk(res, ']'))) {
+                log.info(`${req.method} ${req.url} request closed before inventory completed`);
+                return res.end();
+            }
+
             res.end();
         } else {
             if (Number.isNaN(page)) page = 1;

destiny2/destiny2.routes.spec.js

@@ -4,9 +4,14 @@ import { StatusCodes } from 'http-status-codes';
 import Chance from 'chance';
 import { createResponse, createRequest } from 'node-mocks-http';
 
+vi.mock('../authorization/authorization.middleware.js', () => ({
+    default: vi.fn((_req, _res, next) => next()),
+}));
+
 import Destiny2Router from './destiny2.routes.js';
 import Destiny2Controller from './destiny2.controller.js';
 import manifest2Response from '../mocks/manifest2Response.json';
+import authorizeUser from '../authorization/authorization.middleware.js';
 
 const { Response: manifest } = manifest2Response;
 const chance = new Chance();
@@ -64,6 +69,7 @@ describe('Destiny2Router', () => {
         res = createResponse({
             eventEmitter: EventEmitter,
         });
+        authorizeUser.mockImplementation((_req, _res, next) => next());
     });
 
     describe('getCharacters', () => {
@@ -137,4 +143,55 @@ describe('Destiny2Router', () => {
                 }));
         });
     });
+
+    describe('getInventory', () => {
+        it('should wait for drain when the response stream applies backpressure', () =>
+            new Promise((done, reject) => {
+                const req = createRequest({
+                    method: 'GET',
+                    url: '/inventory',
+                    query: {},
+                });
+                const originalWrite = res.write.bind(res);
+                let writeCalls = 0;
+
+                world.items = [
+                    { hash: 1, displayProperties: { name: 'One' } },
+                    { hash: 2, displayProperties: { name: 'Two' } },
+                ];
+
+                res.write = vi.fn(chunk => {
+                    writeCalls += 1;
+                    originalWrite(chunk);
+
+                    if (writeCalls === 1) {
+                        return false;
+                    }
+
+                    return true;
+                });
+
+                res.on('end', () => {
+                    try {
+                        expect(res.statusCode).toEqual(StatusCodes.OK);
+                        expect(res.write).toHaveBeenCalledTimes(3);
+                        expect(JSON.parse(res._getData())).toEqual(world.items);
+                        done();
+                    } catch (err) {
+                        reject(err);
+                    }
+                });
+
+                destiny2Router(req, res, next);
+
+                setImmediate(() => {
+                    try {
+                        expect(res.write).toHaveBeenCalledTimes(1);
+                        res.emit('drain');
+                    } catch (err) {
+                        reject(err);
+                    }
+                });
+            }));
+    });
 });