fixed query hanging after local document inserted (#7349)

Author: epodgaysky

src/plugins/pipeline/rx-pipeline.ts

@@ -76,7 +76,9 @@ export class RxPipeline<RxDocType> {
         this.source.onClose.push(() => this.close());
         this.destination.awaitBeforeReads.add(this.waitBeforeWriteFn);
         this.subs.push(
-            this.source.eventBulks$.subscribe((bulk) => {
+            this.source.eventBulks$.pipe(
+                filter(bulk => !this.stopped && !bulk.isLocal)
+            ).subscribe((bulk) => {
                 this.lastSourceDocTime.next(bulk.events[0].documentData._meta.lwt);
                 this.somethingChanged.next({});
             })

test/unit/rx-pipeline.test.ts

@@ -17,7 +17,9 @@ import {
 } from '../../plugins/test-utils/index.mjs';
 import { wrappedValidateAjvStorage } from '../../plugins/validate-ajv/index.mjs';
 import { RxDBPipelinePlugin } from '../../plugins/pipeline/index.mjs';
+import { RxDBLocalDocumentsPlugin } from '../../plugins/local-documents/index.mjs';
 addRxPlugin(RxDBPipelinePlugin);
+addRxPlugin(RxDBLocalDocumentsPlugin);
 import { RxDBLeaderElectionPlugin } from '../../plugins/leader-election/index.mjs';
 import { assertThrows } from 'async-test-util';
 addRxPlugin(RxDBLeaderElectionPlugin);
@@ -392,6 +394,31 @@ describe('rx-pipeline.test.js', () => {
             assert.strictEqual(c2After.firstName, 'foobar');
 
 
+            c1.database.close();
+            c2.database.close();
+        });
+        it('should not block reads when localDocument inserted', async () => {
+            const c1 = await humansCollection.create(0);
+            await c1.database.waitForLeadership();
+            const c2 = await humansCollection.create(0);
+
+            await c1.addPipeline({
+                destination: c2,
+                handler: async (docs) => {
+                    for (const doc of docs) {
+                        await c2.insert(schemaObjects.humanData(doc.passportId));
+                    }
+                },
+                identifier: randomToken(10)
+            });
+            await c1.insert(schemaObjects.humanData('foobar'));
+            const doc2 = await c2.findOne().exec(true);
+            assert.strictEqual(doc2.passportId, 'foobar');
+
+            await c1.insertLocal('LOCAL_KEY', { data: true });
+            const doc2AfterLocalInserted = await c2.findOne().exec(true);
+            assert.strictEqual(doc2AfterLocalInserted.passportId, 'foobar');
+
             c1.database.close();
             c2.database.close();
         });