Avoid race conditions during DELETE/INSERT operations

Closes #1301
comunica · Apr 24, 2024 · 6a3ebf5 · 6a3ebf5
1 parent 052ac7d
commit 6a3ebf5
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 10 deletions.
diff --git a/packages/bus-rdf-update-quads/lib/ActorRdfUpdateQuadsDestination.ts b/packages/bus-rdf-update-quads/lib/ActorRdfUpdateQuadsDestination.ts
@@ -51,22 +51,20 @@ export abstract class ActorRdfUpdateQuadsDestination extends ActorRdfUpdateQuads
     destination: IQuadDestination,
     action: IActionRdfUpdateQuads,
   ): Promise<IActorRdfUpdateQuadsOutput> {
-    const execute = (): Promise<void> => Promise.all([
-      action.quadStreamInsert ? destination.insert(action.quadStreamInsert) : Promise.resolve(),
-      action.quadStreamDelete ? destination.delete(action.quadStreamDelete) : Promise.resolve(),
-      action.deleteGraphs ?
+    const execute = async(): Promise<void> => {
+      await (action.quadStreamDelete ? destination.delete(action.quadStreamDelete) : Promise.resolve());
+      await (action.deleteGraphs ?
         destination.deleteGraphs(
           action.deleteGraphs.graphs,
           action.deleteGraphs.requireExistence,
           action.deleteGraphs.dropGraphs,
         ) :
-        Promise.resolve(),
-      action.createGraphs ?
+        Promise.resolve());
+      await (action.createGraphs ?
         destination.createGraphs(action.createGraphs.graphs, action.createGraphs.requireNonExistence) :
-        Promise.resolve(),
-    ]).then(() => {
-      // Return void
-    });
+        Promise.resolve());
+      await (action.quadStreamInsert ? destination.insert(action.quadStreamInsert) : Promise.resolve());
+    };
     return { execute };
   }
 

diff --git a/packages/bus-rdf-update-quads/test/ActorRdfUpdateQuadsDestination-test.ts b/packages/bus-rdf-update-quads/test/ActorRdfUpdateQuadsDestination-test.ts
@@ -131,6 +131,28 @@ describe('ActorRdfUpdateQuadsDestination', () => {
 
       expect(store.getQuads(null, null, null, null)).toEqual([ skolemized ]);
     });
+
+    it('should not delete quads that are being inserted', async() => {
+      const q1 = quad(namedNode('http://example.org/1'), namedNode('http://example.org#type'), namedNode('http://example.org#thing'));
+      const q2 = quad(namedNode('http://example.org/2'), namedNode('http://example.org#type'), namedNode('http://example.org#thing'));
+
+      const store = new Store();
+      store.addQuads([ q1, q2 ]);
+      const context: IActionContext = new ActionContext({
+        [KeysRdfUpdateQuads.destination.name]: store,
+        [KeysQuerySourceIdentify.sourceIds.name]: new Map([[ store, '3' ]]),
+      });
+
+      const output: IActorRdfUpdateQuadsOutput = await actor.run({
+        quadStreamInsert: new ArrayIterator([ q1 ], { autoStart: false }),
+        quadStreamDelete: new ArrayIterator([ q1, q2 ], { autoStart: false }),
+        context,
+      });
+
+      await output.execute();
+
+      expect(store.getQuads(null, null, null, null)).toEqual([ q1 ]);
+    });
   });
 });