fix: Integrate wrapStreamError to prevent uncaught errors

src/ldp/http/SparqlUpdateBodyParser.ts

@@ -5,7 +5,7 @@ import { getLoggerFor } from '../../logging/LogUtil';
 import { APPLICATION_SPARQL_UPDATE } from '../../util/ContentTypes';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
 import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMediaTypeHttpError';
-import { pipeSafe, readableToString } from '../../util/Util';
+import { pipeSafe, readableToString, wrapStreamError } from '../../util/Util';
 import type { BodyParserArgs } from './BodyParser';
 import { BodyParser } from './BodyParser';
 import type { SparqlUpdatePatch } from './SparqlUpdatePatch';
@@ -30,7 +30,7 @@ export class SparqlUpdateBodyParser extends BodyParser {
     // It is impossible to check if object mode is enabled in Node 10 (without accessing private variables)
     const options = { objectMode: request.readableObjectMode };
     const toAlgebraStream = pipeSafe(request, new PassThrough(options));
-    const dataCopy = pipeSafe(request, new PassThrough(options));
+    const dataCopy = wrapStreamError(pipeSafe(request, new PassThrough(options)));
     let algebra: Algebra.Operation;
     try {
       const sparql = await readableToString(toAlgebraStream);

src/server/ExpressHttpServer.ts

@@ -3,6 +3,7 @@ import cors from 'cors';
 import type { Express } from 'express';
 import express from 'express';
 import { getLoggerFor } from '../logging/LogUtil';
+import { wrapStreamError } from '../util/Util';
 import type { HttpHandler } from './HttpHandler';
 
 export class ExpressHttpServer {
@@ -39,7 +40,7 @@ export class ExpressHttpServer {
     app.use(async(request, response, done): Promise<void> => {
       try {
         this.logger.info(`Received request for ${request.url}`);
-        await this.handler.handleSafe({ request, response });
+        await this.handler.handleSafe({ request: wrapStreamError(request), response });
       } catch (error: unknown) {
         const errMsg = error instanceof Error ? `${error.name}: ${error.message}\n${error.stack}` : 'Unknown error.';
         this.logger.error(errMsg);

src/storage/accessors/FileDataAccessor.ts

@@ -15,7 +15,7 @@ import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMedi
 import type { MetadataController } from '../../util/MetadataController';
 import { CONTENT_TYPE, DCTERMS, POSIX, RDF, XSD } from '../../util/UriConstants';
 import { toNamedNode, toTypedLiteral } from '../../util/UriUtil';
-import { pushQuad } from '../../util/Util';
+import { pushQuad, wrapStreamError } from '../../util/Util';
 import type { FileIdentifierMapper, ResourceLink } from '../FileIdentifierMapper';
 import type { DataAccessor } from './DataAccessor';
 
@@ -51,7 +51,7 @@ export class FileDataAccessor implements DataAccessor {
     const stats = await this.getStats(link.filePath);
 
     if (stats.isFile()) {
-      return createReadStream(link.filePath);
+      return wrapStreamError(createReadStream(link.filePath));
     }
 
     throw new NotFoundHttpError();

src/storage/accessors/SparqlDataAccessor.ts

@@ -26,7 +26,7 @@ import { UnsupportedMediaTypeHttpError } from '../../util/errors/UnsupportedMedi
 import type { MetadataController } from '../../util/MetadataController';
 import { CONTENT_TYPE, LDP } from '../../util/UriConstants';
 import { toNamedNode } from '../../util/UriUtil';
-import { ensureTrailingSlash } from '../../util/Util';
+import { ensureTrailingSlash, wrapStreamError } from '../../util/Util';
 import type { ContainerManager } from '../ContainerManager';
 import type { DataAccessor } from './DataAccessor';
 
@@ -78,7 +78,7 @@ export class SparqlDataAccessor implements DataAccessor {
    */
   public async getData(identifier: ResourceIdentifier): Promise<Readable> {
     const name = namedNode(identifier.path);
-    return this.sendSparqlConstruct(this.sparqlConstruct(name));
+    return wrapStreamError(await this.sendSparqlConstruct(this.sparqlConstruct(name)));
   }
 
   /**

src/storage/conversion/QuadToRdfConverter.ts

@@ -5,6 +5,7 @@ import { RepresentationMetadata } from '../../ldp/representation/RepresentationM
 import type { RepresentationPreferences } from '../../ldp/representation/RepresentationPreferences';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { CONTENT_TYPE } from '../../util/UriConstants';
+import { wrapStreamError } from '../../util/Util';
 import { checkRequest, matchingTypes } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
 import { TypedRepresentationConverter } from './TypedRepresentationConverter';
@@ -34,7 +35,7 @@ export class QuadToRdfConverter extends TypedRepresentationConverter {
     const metadata = new RepresentationMetadata(quads.metadata, { [CONTENT_TYPE]: contentType });
     return {
       binary: true,
-      data: rdfSerializer.serialize(quads.data, { contentType }) as Readable,
+      data: wrapStreamError(rdfSerializer.serialize(quads.data, { contentType })) as Readable,
       metadata,
     };
   }

src/storage/conversion/RdfToQuadConverter.ts

@@ -5,7 +5,7 @@ import { RepresentationMetadata } from '../../ldp/representation/RepresentationM
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
 import { CONTENT_TYPE } from '../../util/UriConstants';
-import { pipeSafe } from '../../util/Util';
+import { pipeSafe, wrapStreamError } from '../../util/Util';
 import { checkRequest } from './ConversionUtil';
 import type { RepresentationConverterArgs } from './RepresentationConverter';
 import { TypedRepresentationConverter } from './TypedRepresentationConverter';
@@ -44,7 +44,7 @@ export class RdfToQuadConverter extends TypedRepresentationConverter {
 
     return {
       binary: false,
-      data,
+      data: wrapStreamError(data),
       metadata,
     };
   }

src/storage/patch/SparqlUpdatePatchHandler.ts

@@ -12,6 +12,7 @@ import { getLoggerFor } from '../../logging/LogUtil';
 import { INTERNAL_QUADS } from '../../util/ContentTypes';
 import { UnsupportedHttpError } from '../../util/errors/UnsupportedHttpError';
 import { CONTENT_TYPE } from '../../util/UriConstants';
+import { wrapStreamError } from '../../util/Util';
 import type { ResourceLocker } from '../ResourceLocker';
 import type { ResourceStore } from '../ResourceStore';
 import { PatchHandler } from './PatchHandler';
@@ -77,7 +78,7 @@ export class SparqlUpdatePatchHandler extends PatchHandler {
     const metadata = new RepresentationMetadata(input.identifier.path, { [CONTENT_TYPE]: INTERNAL_QUADS });
     const representation: Representation = {
       binary: false,
-      data: store.match() as Readable,
+      data: wrapStreamError(store.match()) as Readable,
       metadata,
     };
     await this.source.setRepresentation(input.identifier, representation);