Query cache is not being invalidated by replication

[gvuyk]

Case

Bug

Issue

If a query is made, then new docs are added via replication, then the same query is made again, the results won't contain the new docs added by replication

Info

• Environment: Node.js

[pubkey]

It looks like the completed-event is fired when the replication is not completed yet. If i add await AsyncTestUtil.wait(10); before the second query, it works.

[gvuyk]

yeah that could be it, measured the times and in absence of the first query, the second takes around 6 ms more to complete, thus allowing the replication to finish

[gvuyk]

Strange thing is, I tried adding the extra wait time on a script outside the test environment and it didn't work, but destroying the queryCache did

const RxDB = require("rxdb");
RxDB.plugin(require("pouchdb-adapter-memory"));

async function test()
{
    // Create a schema
    const schema = {
        version: 0,
        disableKeyCompression: true,
        type: "object",
        properties: {
            name: {type: "string"},
            number: {type: "number"}
        }
    };

    // Create dbA
    const dbA = await RxDB.create({ name: "dba", adapter: "memory" });

    // Create collection A
    const collectionA = await dbA.collection({ name: 'result', schema: schema });

    // Insert documents
    await collectionA.insert({ name: 'aaaa', "number": 1 });
    await collectionA.insert({ name: 'bbbb', "number": 2 });

    console.log("initial collection A: "+(await collectionA.find({}).exec()).length);

    //----------------------------------------------------------------------------

    // Create dbB
    const dbB = await RxDB.create({ name: "dbb", adapter: "memory" });

    // create collection B
    const collectionB = await dbB.collection({ name: 'result', schema: schema });

    // Pull from collection A
    const pullstate = collectionB.sync({
        remote: collectionA.pouch,
        direction: {pull: true, push: false},
        options: {live: false}
    });

    // Wait for replication to complete
    await new Promise((resolve, reject) => {
        pullstate.complete$.subscribe(completed => {
            if(completed){
                if(completed.ok == true){
                    resolve();
                }else{
                    reject(completed.errors);
                }
            }
        });
        pullstate.error$.subscribe(error => {reject(error);});
    });
    await new Promise(r => { setTimeout(r,100); });
    console.log("collection B after pull: "+(await collectionB.find({}).exec()).length);

    // Delete 1 doc from collection B
    var doc = await collectionB.findOne({name: "aaaa"}).exec();
    await doc.remove();
    await new Promise(r => {setTimeout(r,100);});
    console.log("collection B after removal: "+(await collectionB.find({}).exec()).length);

    // Push to collection A
    const pushstate = collectionB.sync({
        remote: collectionA.pouch,
        direction: {pull: false, push: true},
        options: {live: false}
    });

    // Wait for replication to complete
    await new Promise((resolve, reject) => {
        pushstate.complete$.subscribe(completed => {
            if(completed){
                if(completed.ok == true){
                    resolve();
                }else{
                    reject(completed.errors);
                }
            }
        });
        pushstate.error$.subscribe(error => {reject(error);});
    });

    // Check. Length should eq 1
    console.log("collection A after receiving push: "+(await collectionA.find({}).exec()).length);

    // Wait 1 sec
    await new Promise(r => { setTimeout(r,1000); });

    // Check. Length should eq 1
    console.log("collection A after 1 sec of receiving push: "+(await collectionA.find({}).exec()).length);

    // Destroy queryCache
    collectionA._queryCache.destroy();

    // Check. Length now eq 1
    console.log("collection A after destroying queryCache: "+(await collectionA.find({}).exec()).length);
    var docs = await collectionA.find({}).exec();

    await dbA.remove();
    await dbA.destroy();
    await dbB.remove();
    await dbB.destroy();
}

test().then(() => {});

Console output on my end:

initial collection A: 2
collection B after pull: 2
collection B after removal: 1
collection A after receiving push: 2
collection A after 1 sec of receiving push: 2
collection A after destroying queryCache: 1

Just to be clear, this is with the latest release (7.5.1)

[pubkey]

I found the problem, please try out the latest build.

[gvuyk]

Thanks It solves the problem when there's 1 replication, but the problem is still there when replicating twice though (script in my last comment)

[pubkey]

Can you PR a test with this?