fix: Integrate StreamMonitor to prevent uncaught errors

src/ldp/AuthenticatedLdpHandler.ts

@@ -4,6 +4,7 @@ import type { Authorizer } from '../authorization/Authorizer';
 import { HttpHandler } from '../server/HttpHandler';
 import type { HttpRequest } from '../server/HttpRequest';
 import type { HttpResponse } from '../server/HttpResponse';
+import { StreamMonitor } from '../util/StreamMonitor';
 import type { RequestParser } from './http/RequestParser';
 import type { ResponseDescription } from './http/response/ResponseDescription';
 import type { ResponseWriter } from './http/ResponseWriter';
@@ -113,7 +114,19 @@ export class AuthenticatedLdpHandler extends HttpHandler {
     const op: Operation = await this.requestParser.handleSafe(request);
     const credentials: Credentials = await this.credentialsExtractor.handleSafe(request);
     const permissions: PermissionSet = await this.permissionsExtractor.handleSafe(op);
+
+    // Monitor the data stream while authorizer is checking the permissions
+    let monitor: StreamMonitor | undefined;
+    if (op.body) {
+      monitor = new StreamMonitor(op.body.data, 'AuthenticatedLdpHandler');
+    }
+
     await this.authorizer.handleSafe({ credentials, identifier: op.target, permissions });
+
+    if (monitor) {
+      monitor.release();
+    }
+
     return this.operationHandler.handleSafe(op);
   }
 }

src/storage/DataAccessorBasedStore.ts

@@ -13,6 +13,7 @@ import { NotFoundHttpError } from '../util/errors/NotFoundHttpError';
 import { NotImplementedError } from '../util/errors/NotImplementedError';
 import { UnsupportedHttpError } from '../util/errors/UnsupportedHttpError';
 import type { MetadataController } from '../util/MetadataController';
+import { StreamMonitor } from '../util/StreamMonitor';
 import { CONTENT_TYPE, HTTP, LDP, RDF } from '../util/UriConstants';
 import { ensureTrailingSlash, trimTrailingSlashes } from '../util/Util';
 import type { DataAccessor } from './accessors/DataAccessor';
@@ -93,9 +94,13 @@ export class DataAccessorBasedStore implements ResourceStore {
     // Ensure the representation is supported by the accessor
     await this.accessor.canHandle(representation);
 
+    const monitor = new StreamMonitor(representation.data, 'DataAccessorBasedStore-addResource');
+
     // Using the parent metadata as we can also use that later to check if the nested containers maybe need to be made
     const parentMetadata = await this.getSafeNormalizedMetadata(container);
 
+    monitor.release();
+
     // When a POST method request targets a non-container resource without an existing representation,
     // the server MUST respond with the 404 status code.
     if (!parentMetadata && !container.path.endsWith('/')) {
@@ -120,9 +125,13 @@ export class DataAccessorBasedStore implements ResourceStore {
     // Ensure the representation is supported by the accessor
     await this.accessor.canHandle(representation);
 
+    const monitor = new StreamMonitor(representation.data, 'DataAccessorBasedStore-setRepresentation');
+
     // Check if the resource already exists
     const oldMetadata = await this.getSafeNormalizedMetadata(identifier);
 
+    monitor.release();
+
     // Might want to redirect in the future
     if (oldMetadata && oldMetadata.identifier.value !== identifier.path) {
       throw new ConflictHttpError(`${identifier.path} conflicts with existing path ${oldMetadata.identifier.value}`);
@@ -218,10 +227,14 @@ export class DataAccessorBasedStore implements ResourceStore {
       await this.handleContainerData(representation);
     }
 
+    const monitor = new StreamMonitor(representation.data, 'DataAccessorBasedStore-writeData');
+
     if (createContainers) {
       await this.createRecursiveContainers(await this.containerManager.getContainer(identifier));
     }
 
+    monitor.release();
+
     // Make sure the metadata has the correct identifier and correct type quads
     const { metadata } = representation;
     metadata.identifier = DataFactory.namedNode(identifier.path);

src/storage/accessors/FileDataAccessor.ts

@@ -13,6 +13,7 @@ import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import { isSystemError } from '../../util/errors/SystemError';
 import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMediaTypeHttpError';
 import type { MetadataController } from '../../util/MetadataController';
+import { StreamMonitor } from '../../util/StreamMonitor';
 import { CONTENT_TYPE, DCTERMS, POSIX, RDF, XSD } from '../../util/UriConstants';
 import { toNamedNode, toTypedLiteral } from '../../util/UriUtil';
 import { pushQuad } from '../../util/Util';
@@ -82,13 +83,18 @@ export class FileDataAccessor implements DataAccessor {
     if (this.isMetadataPath(identifier.path)) {
       throw new ConflictHttpError('Not allowed to create files with the metadata extension.');
     }
+
+    const monitor = new StreamMonitor(data, 'FileDataAccessor-writeDocument');
+
     const link = await this.resourceMapper.mapUrlToFilePath(identifier, metadata.contentType);
 
     // Check if we already have a corresponding file with a different extension
     await this.verifyExistingExtension(link);
 
     const wroteMetadata = await this.writeMetadata(link, metadata);
 
+    monitor.release();
+
     try {
       await this.writeDataFile(link.filePath, data);
     } catch (error: unknown) {

src/storage/accessors/SparqlDataAccessor.ts

@@ -24,6 +24,7 @@ import { NotFoundHttpError } from '../../util/errors/NotFoundHttpError';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
 import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMediaTypeHttpError';
 import type { MetadataController } from '../../util/MetadataController';
+import { StreamMonitor } from '../../util/StreamMonitor';
 import { CONTENT_TYPE, LDP } from '../../util/UriConstants';
 import { toNamedNode } from '../../util/UriUtil';
 import { ensureTrailingSlash } from '../../util/Util';
@@ -125,8 +126,13 @@ export class SparqlDataAccessor implements DataAccessor {
     if (this.isMetadataIdentifier(identifier)) {
       throw new ConflictHttpError('Not allowed to create NamedNodes with the metadata extension.');
     }
+
+    const monitor = new StreamMonitor(data, 'SparqlDataAccessor-writeDocument');
+
     const { name, parent } = await this.getRelatedNames(identifier);
 
+    monitor.release();
+
     const triples = await arrayifyStream(data) as Quad[];
     const def = defaultGraph();
     if (triples.some((triple): boolean => !def.equals(triple.graph))) {

src/storage/conversion/ChainedConverter.ts

@@ -1,5 +1,6 @@
 import type { Representation } from '../../ldp/representation/Representation';
 import { getLoggerFor } from '../../logging/LogUtil';
+import { StreamMonitor } from '../../util/StreamMonitor';
 import { matchingMediaType } from '../../util/Util';
 import { checkRequest } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
@@ -44,10 +45,15 @@ export class ChainedConverter extends TypedRepresentationConverter {
   }
 
   public async canHandle(input: RepresentationConverterArgs): Promise<void> {
+    const monitor = new StreamMonitor(input.representation.data, 'ChainedConverter-canHandle');
+
     // We assume a chain can be constructed, otherwise there would be a configuration issue
     // So we only check if the input can be parsed and the preferred type can be written
     const inTypes = this.filterTypes(await this.first.getInputTypes());
     const outTypes = this.filterTypes(await this.last.getOutputTypes());
+
+    monitor.release();
+
     checkRequest(input, inTypes, outTypes);
   }
 
@@ -58,7 +64,9 @@ export class ChainedConverter extends TypedRepresentationConverter {
   public async handle(input: RepresentationConverterArgs): Promise<Representation> {
     const args = { ...input };
     for (let i = 0; i < this.converters.length - 1; ++i) {
+      const monitor = new StreamMonitor(args.representation.data, `ChainedConverter-chain${i}`);
       const value = await this.getMatchingType(this.converters[i], this.converters[i + 1]);
+      monitor.release();
       args.preferences = { type: [{ value, weight: 1 }]};
       args.representation = await this.converters[i].handle(args);
     }

src/storage/conversion/QuadToRdfConverter.ts

@@ -4,6 +4,7 @@ import type { Representation } from '../../ldp/representation/Representation';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import type { RepresentationPreferences } from '../../ldp/representation/RepresentationPreferences';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
+import { StreamMonitor } from '../../util/StreamMonitor';
 import { CONTENT_TYPE } from '../../util/UriConstants';
 import { checkRequest, matchingTypes } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
@@ -22,7 +23,9 @@ export class QuadToRdfConverter extends TypedRepresentationConverter {
   }
 
   public async canHandle(input: RepresentationConverterArgs): Promise<void> {
+    const monitor = new StreamMonitor(input.representation.data, 'QuadToRdfConverter');
     checkRequest(input, [ INTERNAL_QUADS ], await rdfSerializer.getContentTypes());
+    monitor.release();
   }
 
   public async handle(input: RepresentationConverterArgs): Promise<Representation> {

src/storage/conversion/RdfToQuadConverter.ts

@@ -4,6 +4,7 @@ import type { Representation } from '../../ldp/representation/Representation';
 import { RepresentationMetadata } from '../../ldp/representation/RepresentationMetadata';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
+import { StreamMonitor } from '../../util/StreamMonitor';
 import { CONTENT_TYPE } from '../../util/UriConstants';
 import { pipeSafe } from '../../util/Util';
 import { checkRequest } from './ConversionUtil';
@@ -23,7 +24,9 @@ export class RdfToQuadConverter extends TypedRepresentationConverter {
   }
 
   public async canHandle(input: RepresentationConverterArgs): Promise<void> {
+    const monitor = new StreamMonitor(input.representation.data, 'RdfToQuadConverter');
     checkRequest(input, await rdfParser.getContentTypes(), [ INTERNAL_QUADS ]);
+    monitor.release();
   }
 
   public async handle(input: RepresentationConverterArgs): Promise<Representation> {

test/unit/ldp/AuthenticatedLdpHandler.test.ts

@@ -1,3 +1,4 @@
+import streamifyArray from 'streamify-array';
 import type { CredentialsExtractor } from '../../../src/authentication/CredentialsExtractor';
 import type { Authorizer } from '../../../src/authorization/Authorizer';
 import type { AuthenticatedLdpHandlerArgs } from '../../../src/ldp/AuthenticatedLdpHandler';
@@ -77,4 +78,11 @@ describe('An AuthenticatedLdpHandler', (): void => {
 
     await expect(handler.handle({ request: 'request' as any, response: {} as HttpResponse })).rejects.toEqual('apple');
   });
+
+  it('can handle operations with data.', async(): Promise< void> => {
+    args.requestParser.handle = async(): Promise<any> => ({ body: { data: streamifyArray([ 'data' ]) }});
+    const handler = new AuthenticatedLdpHandler(args);
+
+    await expect(handler.handle({ request: 'request' as any, response: 'response' as any })).resolves.toBeUndefined();
+  });
 });

test/unit/storage/conversion/ChainedConverter.test.ts

@@ -1,3 +1,4 @@
+import streamifyArray from 'streamify-array';
 import type { Representation } from '../../../../src/ldp/representation/Representation';
 import { RepresentationMetadata } from '../../../../src/ldp/representation/RepresentationMetadata';
 import type { RepresentationPreferences } from '../../../../src/ldp/representation/RepresentationPreferences';
@@ -52,7 +53,7 @@ describe('A ChainedConverter', (): void => {
     converter = new ChainedConverter(converters);
 
     const metadata = new RepresentationMetadata({ [CONTENT_TYPE]: 'text/turtle' });
-    representation = { metadata } as Representation;
+    representation = { data: streamifyArray([]), metadata } as Representation;
     preferences = { type: [{ value: 'internal/quads', weight: 1 }]};
     args = { representation, preferences, identifier: { path: 'path' }};
   });

test/unit/storage/conversion/QuadToRdfConverter.test.ts

@@ -24,13 +24,13 @@ describe('A QuadToRdfConverter', (): void => {
   });
 
   it('can handle quad to turtle conversions.', async(): Promise<void> => {
-    const representation = { metadata } as Representation;
+    const representation = { data: streamifyArray([]), metadata } as Representation;
     const preferences: RepresentationPreferences = { type: [{ value: 'text/turtle', weight: 1 }]};
     await expect(converter.canHandle({ identifier, representation, preferences })).resolves.toBeUndefined();
   });
 
   it('can handle quad to JSON-LD conversions.', async(): Promise<void> => {
-    const representation = { metadata } as Representation;
+    const representation = { data: streamifyArray([]), metadata } as Representation;
     const preferences: RepresentationPreferences = { type: [{ value: 'application/ld+json', weight: 1 }]};
     await expect(converter.canHandle({ identifier, representation, preferences })).resolves.toBeUndefined();
   });

test/unit/storage/conversion/RdfToQuadConverter.test.ts

@@ -26,14 +26,14 @@ describe('A RdfToQuadConverter.test.ts', (): void => {
 
   it('can handle turtle to quad conversions.', async(): Promise<void> => {
     const metadata = new RepresentationMetadata({ [CONTENT_TYPE]: 'text/turtle' });
-    const representation = { metadata } as Representation;
+    const representation = { data: streamifyArray([]), metadata } as Representation;
     const preferences: RepresentationPreferences = { type: [{ value: INTERNAL_QUADS, weight: 1 }]};
     await expect(converter.canHandle({ identifier, representation, preferences })).resolves.toBeUndefined();
   });
 
   it('can handle JSON-LD to quad conversions.', async(): Promise<void> => {
     const metadata = new RepresentationMetadata({ [CONTENT_TYPE]: 'application/ld+json' });
-    const representation = { metadata } as Representation;
+    const representation = { data: streamifyArray([]), metadata } as Representation;
     const preferences: RepresentationPreferences = { type: [{ value: INTERNAL_QUADS, weight: 1 }]};
     await expect(converter.canHandle({ identifier, representation, preferences })).resolves.toBeUndefined();
   });