Make loading configs and modules deterministic

lib/loading/ComponentRegistry.ts

@@ -20,6 +20,7 @@ export class ComponentRegistry {
   private readonly componentResources: Record<string, Resource>;
   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<void> {
+    // 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<void> {
+    // 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,

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<void> {
+    // 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,

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,10 @@ 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.
+   */
+  blankNodePrefix?: string;
   /**
    * An optional logger.
    */

lib/rdf/RdfStreamIncluder.ts

@@ -11,9 +11,8 @@ import type { RdfParserOptions } from './RdfParser';
  * A RdfStreamIncluder takes a triple stream and detects owl:includes to automatically include other files.
  */
 export class RdfStreamIncluder extends Transform {
-  private runningImporters = 1;
   private readonly parserOptions: RdfParserOptions;
-  private flushCallback: TransformCallback | undefined;
+  private readonly imports: Promise<Readable>[] = [];
 
   public constructor(parserOptions: RdfParserOptions) {
     super({ objectMode: true });
@@ -31,10 +30,19 @@ 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(), () => {
+      // The error has already been emitted
+    });
+  }
+
+  /**
+   * Emit the quads of all imports in a deterministic order
+   */
+  protected async flushImports(): Promise<void> {
+    for (const importStream of this.imports) {
+      for await (const quad of await importStream) {
+        this.push(quad);
+      }
     }
   }
 
@@ -44,7 +52,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 +67,36 @@ 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.
+      // Errors are emitted immediately, so they don't have to wait for preceding imports.
+      const importStream = this.parseImport(relativeFilePath, this.imports.length);
+      importStream.catch((error: Error) => this.emit('error', error));
+      this.imports.push(importStream);
+    }
+  }
+
+  /**
+   * 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<Readable> {
+    let rawStream: Readable;
+    try {
+      rawStream = await RdfParser.fetchFileOrUrl(relativeFilePath);
+    } catch (error: unknown) {
+      throw RdfParser.addPathToError(<Error> error, this.parserOptions.path);
     }
+    return 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}_`,
+    }).on('error', (error: Error) => this.emit('error', error));
   }
 
   /**

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)

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', () => {