From 4b47d12137940cf051dbd2062bd783e9c940d156 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 17:35:09 +0000 Subject: [PATCH 1/3] Make loading configs and modules deterministic Imports were parsed in parallel, and their quads were emitted as soon as they were parsed, so the order of quads depended on I/O timing. This made for example the order of list values that are spread over multiple files, and thereby the order of instantiations, differ between runs. Imports are still fetched and parsed in parallel, but their quads are now emitted after the quads of the importing file, in the order in which the imports occur. Furthermore, labels of blank nodes depended on the order in which files started being parsed. Each file now receives a blank node prefix that depends on its position in the registration order or the import tree. As a result, compiling the same config multiple times now produces the same output. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UeSuG8npLjD7EpaYC6LCWE --- lib/loading/ComponentRegistry.ts | 6 + lib/loading/ConfigRegistry.ts | 4 + lib/rdf/RdfParser.ts | 16 ++- lib/rdf/RdfStreamIncluder.ts | 80 ++++++++----- test/unit/loading/ComponentRegistry-test.ts | 25 ++++ test/unit/loading/ConfigRegistry-test.ts | 14 +++ test/unit/rdf/RdfParser-test.ts | 121 ++++++++++++++++++++ 7 files changed, 239 insertions(+), 27 deletions(-) diff --git a/lib/loading/ComponentRegistry.ts b/lib/loading/ComponentRegistry.ts index 4975c73..34c3338 100644 --- a/lib/loading/ComponentRegistry.ts +++ b/lib/loading/ComponentRegistry.ts @@ -20,6 +20,7 @@ export class ComponentRegistry { private readonly componentResources: Record; private readonly skipContextValidation: boolean; private readonly remoteContextLookups: boolean; + private registeredModules = 0; public constructor(options: IComponentLoaderRegistryOptions) { this.moduleState = options.moduleState; @@ -36,8 +37,10 @@ export class ComponentRegistry { * Effectively, this will load in all components that are available in the main package and its dependencies. */ public async registerAvailableModules(): Promise { + // Sorted, as blank node labels depend on the registration order, and modules are discovered in any order. await Promise.all(Object.values(this.moduleState.componentModules) .flatMap(x => Object.values(x)) + .sort((left, right) => left < right ? -1 : Number(left > right)) .map((moduleResourceUrl: string) => this.registerModule(moduleResourceUrl))); } @@ -46,9 +49,12 @@ export class ComponentRegistry { * @param urlOrPath An RDF document URL or local file path. */ public async registerModule(urlOrPath: string): Promise { + // Determined before loading, so that blank node labels only depend on the order of the calls. + const blankNodePrefix = `module_${this.registeredModules++}_`; const data = await RdfParser.fetchFileOrUrl(urlOrPath); await this.registerModuleStream(new RdfParser().parse(data, { path: urlOrPath, + blankNodePrefix, contexts: this.moduleState.contexts, importPaths: this.moduleState.importPaths, logger: this.logger, diff --git a/lib/loading/ConfigRegistry.ts b/lib/loading/ConfigRegistry.ts index 60a8690..6b7000c 100644 --- a/lib/loading/ConfigRegistry.ts +++ b/lib/loading/ConfigRegistry.ts @@ -15,6 +15,7 @@ export class ConfigRegistry { private readonly logger: Logger; private readonly skipContextValidation: boolean; private readonly remoteContextLookups: boolean; + private registeredConfigs = 0; public constructor(options: IConfigLoaderRegistryOptions) { this.moduleState = options.moduleState; @@ -29,9 +30,12 @@ export class ConfigRegistry { * @param urlOrPath An RDF document URL or local file path. */ public async register(urlOrPath: string): Promise { + // Determined before loading, so that blank node labels only depend on the order of the calls. + const blankNodePrefix = `config_${this.registeredConfigs++}_`; const data = await RdfParser.fetchFileOrUrl(urlOrPath); return await this.registerStream(new RdfParser().parse(data, { path: urlOrPath, + blankNodePrefix, contexts: this.moduleState.contexts, importPaths: this.moduleState.importPaths, ignoreImports: false, diff --git a/lib/rdf/RdfParser.ts b/lib/rdf/RdfParser.ts index 0ff9f92..3b342ca 100644 --- a/lib/rdf/RdfParser.ts +++ b/lib/rdf/RdfParser.ts @@ -1,6 +1,7 @@ import { createReadStream, promises as fs } from 'node:fs'; import type { Readable } from 'node:stream'; import type * as RDF from '@rdfjs/types'; +import { DataFactory } from 'rdf-data-factory'; import type { ParseOptions } from 'rdf-parse'; import { rdfParser } from 'rdf-parse'; import type { Logger } from 'winston'; @@ -57,7 +58,12 @@ export class RdfParser { }; // Execute parsing - const quadStream = rdfParser.parse(textStream, options); + const quadStream = rdfParser.parse(textStream, { + ...options, + dataFactory: options.dataFactory ?? (options.blankNodePrefix === undefined ? + undefined : + new DataFactory({ blankNodePrefix: options.blankNodePrefix })), + }); const includedQuadStream = quadStream.pipe(new RdfStreamIncluder(options)); quadStream.on('error', (error: Error) => includedQuadStream .emit('error', RdfParser.addPathToError(error, options.path))); @@ -115,6 +121,14 @@ export type RdfParserOptions = ParseOptions & { * Undefined if this file is the root file. */ importedFromPath?: string; + /** + * The prefix of the labels of blank nodes that are created while parsing. + * Imported files derive their prefix from it and the position of their import, + * so that labels don't depend on the order in which files are loaded. + * If undefined, labels depend on the order in which files are parsed. + * This is ignored if a data factory is passed. + */ + blankNodePrefix?: string; /** * An optional logger. */ diff --git a/lib/rdf/RdfStreamIncluder.ts b/lib/rdf/RdfStreamIncluder.ts index 85894d1..0a3436a 100644 --- a/lib/rdf/RdfStreamIncluder.ts +++ b/lib/rdf/RdfStreamIncluder.ts @@ -9,11 +9,15 @@ import type { RdfParserOptions } from './RdfParser'; /** * A RdfStreamIncluder takes a triple stream and detects owl:includes to automatically include other files. + * + * Imports are fetched and parsed in parallel, + * but their quads are only emitted after the quads of the current stream, + * one import after the other in the order in which they were detected. + * This makes the resulting quad order independent of I/O timing. */ export class RdfStreamIncluder extends Transform { - private runningImporters = 1; private readonly parserOptions: RdfParserOptions; - private flushCallback: TransformCallback | undefined; + private readonly imports: Promise[] = []; public constructor(parserOptions: RdfParserOptions) { super({ objectMode: true }); @@ -31,10 +35,17 @@ export class RdfStreamIncluder extends Transform { // eslint-disable-next-line ts/naming-convention public _flush(callback: TransformCallback): void { - if (--this.runningImporters === 0) { - callback(); - } else { - this.flushCallback = callback; + this.flushImports().then(() => callback(), callback); + } + + /** + * Emit the quads of all imports, in the order in which they were detected. + */ + protected async flushImports(): Promise { + for (const importedQuads of this.imports) { + for (const quad of await importedQuads) { + this.push(quad); + } } } @@ -44,7 +55,6 @@ export class RdfStreamIncluder extends Transform { */ public handleImports(quad: RDF.Quad): void { if (!this.parserOptions.ignoreImports && quad.predicate.value === IRIS_RDFS.seeAlso) { - this.runningImporters++; let relativeFilePath = decodeURI(quad.object.value); // Try overriding path using defined import paths @@ -60,26 +70,44 @@ export class RdfStreamIncluder extends Transform { } } - // Recursively call the parser - RdfParser.fetchFileOrUrl(relativeFilePath) - .then((rawStream: Readable) => { - const data: Readable = new RdfParser().parse(rawStream, { - ...this.parserOptions, - baseIRI: undefined, - path: relativeFilePath, - importedFromPath: this.parserOptions.path, - }); - data - .on('data', (subData: RDF.Quad) => this.push(subData)) - .on('error', (error: Error): boolean => this.emit('error', error)) - .on('end', () => { - if (this.flushCallback && --this.runningImporters === 0) { - this.flushCallback(); - } - }); - }) - .catch((error: Error) => this.emit('error', RdfParser.addPathToError(error, this.parserOptions.path))); + // Recursively call the parser, and buffer the quads until they can be emitted in order. + // Errors are emitted immediately, so they don't have to wait for preceding imports. + this.imports.push(this.parseImport(relativeFilePath, this.imports.length).catch((error: Error) => { + this.emit('error', error); + return []; + })); + } + } + + /** + * Parse the given imported file into an array of quads. + * @param relativeFilePath The path or URL of the imported file. + * @param index The position of the import among the imports of the current stream. + */ + protected async parseImport(relativeFilePath: string, index: number): Promise { + let rawStream: Readable; + try { + rawStream = await RdfParser.fetchFileOrUrl(relativeFilePath); + } catch (error: unknown) { + throw RdfParser.addPathToError( error, this.parserOptions.path); } + const data: Readable = new RdfParser().parse(rawStream, { + ...this.parserOptions, + baseIRI: undefined, + path: relativeFilePath, + importedFromPath: this.parserOptions.path, + // Labels of the current stream are its prefix followed by digits, so these never clash. + blankNodePrefix: this.parserOptions.blankNodePrefix === undefined ? + undefined : + `${this.parserOptions.blankNodePrefix}${index}_`, + }); + return new Promise((resolve, reject) => { + const quads: RDF.Quad[] = []; + data + .on('data', (quad: RDF.Quad) => quads.push(quad)) + .on('error', reject) + .on('end', () => resolve(quads)); + }); } /** diff --git a/test/unit/loading/ComponentRegistry-test.ts b/test/unit/loading/ComponentRegistry-test.ts index 1a45ae8..cfd90d7 100644 --- a/test/unit/loading/ComponentRegistry-test.ts +++ b/test/unit/loading/ComponentRegistry-test.ts @@ -5,6 +5,7 @@ import { RdfObjectLoader } from 'rdf-object'; import type { Logger } from 'winston'; import { ComponentRegistry } from '../../../lib/loading/ComponentRegistry'; import type { IModuleState } from '../../../lib/loading/ModuleStateBuilder'; +import { RdfParser } from '../../../lib/rdf/RdfParser'; describe('ComponentRegistry', () => { let moduleState: IModuleState; @@ -54,9 +55,33 @@ describe('ComponentRegistry', () => { expect(Object.keys(objectLoader.resources) .includes('http://example.org/HelloWorldModule#SayHelloComponentNested')).toBeTruthy(); }); + + it('should register discovered modules in sorted order', async() => { + const registerModule = jest.spyOn(componentRegistry, 'registerModule').mockResolvedValue(); + moduleState.componentModules = { + B: { 1: 'b1', 2: 'b2' }, + A: { 1: 'a1' }, + C: { 1: 'c1' }, + }; + await componentRegistry.registerAvailableModules(); + expect(registerModule.mock.calls.map(args => args[0])).toEqual([ 'a1', 'b1', 'b2', 'c1' ]); + }); }); describe('registerModule', () => { + it('should assign blank node prefixes in the order of the calls', async() => { + const parse = jest.spyOn(RdfParser.prototype, 'parse'); + await Promise.all([ + componentRegistry.registerModule(Path.join(__dirname, '../../assets/module.jsonld')), + componentRegistry.registerModule(Path.join(__dirname, '../../assets/module-dynamicentries.jsonld')), + ]); + expect(parse.mock.calls.map(args => [ args[1].path, args[1].blankNodePrefix ])).toEqual(expect.arrayContaining([ + [ Path.join(__dirname, '../../assets/module.jsonld').replaceAll('\\', '/'), 'module_0_' ], + [ Path.join(__dirname, '../../assets/module-dynamicentries.jsonld').replaceAll('\\', '/'), 'module_1_' ], + ])); + parse.mockRestore(); + }); + it('should handle a valid module file', async() => { await componentRegistry.registerModule(Path.join(__dirname, '../../assets/module.jsonld')); expect(Object.keys(objectLoader.resources) diff --git a/test/unit/loading/ConfigRegistry-test.ts b/test/unit/loading/ConfigRegistry-test.ts index b5f79ab..5b0696f 100644 --- a/test/unit/loading/ConfigRegistry-test.ts +++ b/test/unit/loading/ConfigRegistry-test.ts @@ -5,6 +5,7 @@ import { stringToTerm } from 'rdf-string'; import type { Logger } from 'winston'; import { ConfigRegistry } from '../../../lib/loading/ConfigRegistry'; import type { IModuleState } from '../../../lib/loading/ModuleStateBuilder'; +import { RdfParser } from '../../../lib/rdf/RdfParser'; describe('ConfigRegistry', () => { let moduleState: IModuleState; @@ -42,6 +43,19 @@ describe('ConfigRegistry', () => { it('should throw on an invalid module file', async() => { await expect(configRegistry.register(`not-exists.jsonld`)).rejects.toThrow(/ENOENT/u); }); + + it('should assign blank node prefixes in the order of the calls', async() => { + const parse = jest.spyOn(RdfParser.prototype, 'parse'); + await Promise.all([ + configRegistry.register(Path.join(__dirname, '../../assets/config.jsonld')), + configRegistry.register(Path.join(__dirname, '../../assets/config-lazy.jsonld')), + ]); + expect(parse.mock.calls.map(args => [ args[1].path, args[1].blankNodePrefix ])).toEqual(expect.arrayContaining([ + [ Path.join(__dirname, '../../assets/config.jsonld').replaceAll('\\', '/'), 'config_0_' ], + [ Path.join(__dirname, '../../assets/config-lazy.jsonld').replaceAll('\\', '/'), 'config_1_' ], + ])); + parse.mockRestore(); + }); }); describe('registerCustom', () => { diff --git a/test/unit/rdf/RdfParser-test.ts b/test/unit/rdf/RdfParser-test.ts index d7e19af..165e061 100644 --- a/test/unit/rdf/RdfParser-test.ts +++ b/test/unit/rdf/RdfParser-test.ts @@ -1,4 +1,5 @@ import * as Path from 'node:path'; +import { DataFactory } from 'rdf-data-factory'; import type { RdfParserOptions } from '../../../lib/rdf/RdfParser'; import { RdfParser } from '../../../lib/rdf/RdfParser'; import 'jest-rdf'; @@ -34,6 +35,41 @@ globalThis.fetch = jest.fn(async(url: string) => { statusText: 'OK', }; } + if (url === 'http://example.org/myfileslow.ttl') { + await new Promise(resolve => setTimeout(resolve, 50)); + return { + body: streamifyString(` .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileslownest.ttl') { + await new Promise(resolve => setTimeout(resolve, 50)); + return { + body: streamifyString(` .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileblank.ttl') { + return { + body: streamifyString(`[] . + .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } + if (url === 'http://example.org/myfileblanknest.ttl') { + return { + body: streamifyString(`[] .`), + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } if (url === 'http://example.org/myfileerror.ttl') { return { body: streamifyString(`<...`), @@ -296,6 +332,91 @@ describe('RdfParser', () => { ]); }); + it('for a Turtle stream with imports should emit imports in their order, regardless of loading time', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + . + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('ex:s', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileslow.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile2.ttl'), + quad('ex:s3', 'ex:p3', 'ex:o3'), + quad('ex:s2', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with nested imports should emit each import before the next one', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileslownest.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile2.ttl'), + quad('ex:s3', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile1.ttl'), + quad('ex:s1', 'ex:p1', 'ex:o1'), + quad('ex:s2', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with a slow import followed by an import to unknown file', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).rejects + .toThrow(new Error(`Error while parsing file "path/to/file.ttl": URL not found: http://example.org/myfileunknown.ttl`)); + }); + + it('for a Turtle stream with imports and a blank node prefix should derive prefixes for imports', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + blankNodePrefix: 'p_', + }; + const doc = ` +[] . + , . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('_:p_0', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfile1.ttl'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblank.ttl'), + quad('ex:s1', 'ex:p1', 'ex:o1'), + quad('_:p_1_0', 'ex:p1', 'ex:o1'), + quad('ex:s1', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblanknest.ttl'), + quad('_:p_1_0_0', 'ex:p2', 'ex:o2'), + ]); + }); + + it('for a Turtle stream with a blank node prefix and a data factory should use the data factory', async() => { + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + blankNodePrefix: 'p_', + dataFactory: new DataFactory({ blankNodePrefix: 'f_' }), + }; + const doc = ` +[] . + . +`; + await expect(arrayifyStream(parser.parse(streamifyString(doc), options))).resolves + .toEqualRdfQuadArray([ + quad('_:f_0', 'ex:p', 'ex:o'), + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfileblanknest.ttl'), + quad('_:f_1', 'ex:p2', 'ex:o2'), + ]); + }); + it('for a Turtle stream with imports, when ignoring imports', async() => { const options: RdfParserOptions = { path: 'path/to/file.ttl', From 150d073542f0b4a0b62f76e8a1ac58196d0e4cd3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Nov 2026 18:28:49 +0000 Subject: [PATCH 2/3] Stream imports in order instead of buffering them Imports are still started as soon as they are detected, but instead of buffering all their quads, they are read one after the other once the importing stream has ended. Imports that are not being read yet pause once their stream buffer is full, so quads keep flowing as they are parsed, and memory does not grow with the size of imports. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UeSuG8npLjD7EpaYC6LCWE --- lib/rdf/RdfParser.ts | 4 --- lib/rdf/RdfStreamIncluder.ts | 43 ++++++++++---------------- test/unit/rdf/RdfParser-test.ts | 55 +++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 30 deletions(-) diff --git a/lib/rdf/RdfParser.ts b/lib/rdf/RdfParser.ts index 3b342ca..7c196c8 100644 --- a/lib/rdf/RdfParser.ts +++ b/lib/rdf/RdfParser.ts @@ -123,10 +123,6 @@ export type RdfParserOptions = ParseOptions & { importedFromPath?: string; /** * The prefix of the labels of blank nodes that are created while parsing. - * Imported files derive their prefix from it and the position of their import, - * so that labels don't depend on the order in which files are loaded. - * If undefined, labels depend on the order in which files are parsed. - * This is ignored if a data factory is passed. */ blankNodePrefix?: string; /** diff --git a/lib/rdf/RdfStreamIncluder.ts b/lib/rdf/RdfStreamIncluder.ts index 0a3436a..d35b462 100644 --- a/lib/rdf/RdfStreamIncluder.ts +++ b/lib/rdf/RdfStreamIncluder.ts @@ -9,15 +9,10 @@ import type { RdfParserOptions } from './RdfParser'; /** * A RdfStreamIncluder takes a triple stream and detects owl:includes to automatically include other files. - * - * Imports are fetched and parsed in parallel, - * but their quads are only emitted after the quads of the current stream, - * one import after the other in the order in which they were detected. - * This makes the resulting quad order independent of I/O timing. */ export class RdfStreamIncluder extends Transform { private readonly parserOptions: RdfParserOptions; - private readonly imports: Promise[] = []; + private readonly imports: Promise[] = []; public constructor(parserOptions: RdfParserOptions) { super({ objectMode: true }); @@ -35,15 +30,19 @@ export class RdfStreamIncluder extends Transform { // eslint-disable-next-line ts/naming-convention public _flush(callback: TransformCallback): void { - this.flushImports().then(() => callback(), callback); + this.flushImports().then(() => callback(), () => { + // The error has already been emitted + }); } /** - * Emit the quads of all imports, in the order in which they were detected. + * Emit the quads of all imports after the quads of this stream, one import after the other, + * in the order in which they were detected, so that the order does not depend on I/O timing. + * Imports are already being parsed in the meantime, until their stream buffer is full. */ protected async flushImports(): Promise { - for (const importedQuads of this.imports) { - for (const quad of await importedQuads) { + for (const importStream of this.imports) { + for await (const quad of await importStream) { this.push(quad); } } @@ -70,28 +69,27 @@ export class RdfStreamIncluder extends Transform { } } - // Recursively call the parser, and buffer the quads until they can be emitted in order. + // Recursively call the parser. // Errors are emitted immediately, so they don't have to wait for preceding imports. - this.imports.push(this.parseImport(relativeFilePath, this.imports.length).catch((error: Error) => { - this.emit('error', error); - return []; - })); + const importStream = this.parseImport(relativeFilePath, this.imports.length); + importStream.catch((error: Error) => this.emit('error', error)); + this.imports.push(importStream); } } /** - * Parse the given imported file into an array of quads. + * Start parsing the given imported file. * @param relativeFilePath The path or URL of the imported file. * @param index The position of the import among the imports of the current stream. */ - protected async parseImport(relativeFilePath: string, index: number): Promise { + protected async parseImport(relativeFilePath: string, index: number): Promise { let rawStream: Readable; try { rawStream = await RdfParser.fetchFileOrUrl(relativeFilePath); } catch (error: unknown) { throw RdfParser.addPathToError( error, this.parserOptions.path); } - const data: Readable = new RdfParser().parse(rawStream, { + return new RdfParser().parse(rawStream, { ...this.parserOptions, baseIRI: undefined, path: relativeFilePath, @@ -100,14 +98,7 @@ export class RdfStreamIncluder extends Transform { blankNodePrefix: this.parserOptions.blankNodePrefix === undefined ? undefined : `${this.parserOptions.blankNodePrefix}${index}_`, - }); - return new Promise((resolve, reject) => { - const quads: RDF.Quad[] = []; - data - .on('data', (quad: RDF.Quad) => quads.push(quad)) - .on('error', reject) - .on('end', () => resolve(quads)); - }); + }).on('error', (error: Error) => this.emit('error', error)); } /** diff --git a/test/unit/rdf/RdfParser-test.ts b/test/unit/rdf/RdfParser-test.ts index 165e061..4cba72f 100644 --- a/test/unit/rdf/RdfParser-test.ts +++ b/test/unit/rdf/RdfParser-test.ts @@ -1,4 +1,6 @@ import * as Path from 'node:path'; +import { PassThrough } from 'node:stream'; +import type * as RDF from '@rdfjs/types'; import { DataFactory } from 'rdf-data-factory'; import type { RdfParserOptions } from '../../../lib/rdf/RdfParser'; import { RdfParser } from '../../../lib/rdf/RdfParser'; @@ -10,7 +12,21 @@ const quad = require('rdf-quad'); const stringifyStream = require('stream-to-string'); const streamifyString = require('streamify-string'); +// The body of http://example.org/myfilestreaming.ttl, which tests write to themselves +let resolveStreamingBody: (body: PassThrough) => void; +let streamingBody: Promise; + globalThis.fetch = jest.fn(async(url: string) => { + if (url === 'http://example.org/myfilestreaming.ttl') { + const body = new PassThrough(); + resolveStreamingBody(body); + return { + body, + ok: true, + headers: new Headers({ 'Content-Type': 'text/turtle' }), + statusText: 'OK', + }; + } if (url === 'http://example.org/myfile1.ttl') { return { body: streamifyString(` .`), @@ -367,6 +383,45 @@ describe('RdfParser', () => { ]); }); + it('for a Turtle stream with an import should emit quads of the import while it is being parsed', async() => { + streamingBody = new Promise(resolve => resolveStreamingBody = resolve); + const options: RdfParserOptions = { + path: 'path/to/file.ttl', + }; + const doc = ` + . +`; + const quads: RDF.Quad[] = []; + let resolveImportedQuad: () => void; + const importedQuad = new Promise(resolve => resolveImportedQuad = resolve); + const stream = parser.parse(streamifyString(doc), options) + .on('data', (data: RDF.Quad) => { + quads.push(data); + if (data.subject.value === 'ex:s4') { + resolveImportedQuad(); + } + }); + const ended = new Promise(resolve => stream.on('end', resolve)); + + // The first quad of the import is emitted before the import has ended + const body = await streamingBody; + body.write(` .\n`); + await importedQuad; + await new Promise(resolve => setImmediate(resolve)); + expect(quads).toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfilestreaming.ttl'), + quad('ex:s4', 'ex:p4', 'ex:o4'), + ]); + + body.end(` .\n`); + await ended; + expect(quads).toEqualRdfQuadArray([ + quad('ex:s', 'http://www.w3.org/2000/01/rdf-schema#seeAlso', 'http://example.org/myfilestreaming.ttl'), + quad('ex:s4', 'ex:p4', 'ex:o4'), + quad('ex:s5', 'ex:p5', 'ex:o5'), + ]); + }); + it('for a Turtle stream with a slow import followed by an import to unknown file', async() => { const options: RdfParserOptions = { path: 'path/to/file.ttl', From 8716f47e6c34fe0b89a118ed1dbcb835ccfa811d Mon Sep 17 00:00:00 2001 From: Ruben Taelman Date: Sat, 7 Nov 2026 16:30:28 +0100 Subject: [PATCH 3/3] Update lib/rdf/RdfStreamIncluder.ts --- lib/rdf/RdfStreamIncluder.ts | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/lib/rdf/RdfStreamIncluder.ts b/lib/rdf/RdfStreamIncluder.ts index d35b462..1279635 100644 --- a/lib/rdf/RdfStreamIncluder.ts +++ b/lib/rdf/RdfStreamIncluder.ts @@ -36,9 +36,7 @@ export class RdfStreamIncluder extends Transform { } /** - * Emit the quads of all imports after the quads of this stream, one import after the other, - * in the order in which they were detected, so that the order does not depend on I/O timing. - * Imports are already being parsed in the meantime, until their stream buffer is full. + * Emit the quads of all imports in a deterministic order */ protected async flushImports(): Promise { for (const importStream of this.imports) {