Skip to content

Commit ec343f2

Browse files
pubkeypubkey
authoredJun 7, 2022
WIP CHANGE no _meta in revision (#3838)
* CHANGE no _meta in revision * ADD logs * CHANGE (RxStorage) revision hash must not include the `_meta` field * FIX lint
·
16.8.112.5.0
1 parent 3160e9d commit ec343f2

14 files changed

+197
-60
lines changed
 

‎CHANGELOG.md

+2-1
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,8 @@
44
<!-- CHANGELOG NEWEST -->
55

66
- FIX: RxStorage should never emit an eventBulk with an empty events array.
7-
- Update PouchDB to `7.3.0` Thanks [@cetsupport](https://github.com/cetsupport)
7+
- Update PouchDB to `7.3.0` Thanks [@cetsupport](https://github.com/cetsupport).
8+
- CHANGE (RxStorage) revision hash must not include the `_meta` field.
89

910
<!-- ADD new changes here! -->
1011

‎src/plugins/local-documents/rx-local-document.ts

+7-5
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ import objectPath from 'object-path';
22
import { distinctUntilChanged, map } from 'rxjs/operators';
33
import { overwritable } from '../../overwritable';
44
import { basePrototype, createRxDocumentConstructor } from '../../rx-document';
5-
import { isPouchdbConflictError, newRxError, newRxTypeError } from '../../rx-error';
5+
import { isBulkWriteConflictError, newRxError, newRxTypeError } from '../../rx-error';
66
import { writeSingle } from '../../rx-storage-helper';
77
import type {
88
LocalDocumentAtomicUpdateFunction,
@@ -16,7 +16,7 @@ import type {
1616
RxLocalDocument,
1717
RxLocalDocumentData
1818
} from '../../types';
19-
import { clone, flatClone, getDefaultRevision, getDefaultRxDocumentMeta, getFromObjectOrThrow } from '../../util';
19+
import { clone, createRevision, flatClone, getDefaultRevision, getDefaultRxDocumentMeta, getFromObjectOrThrow } from '../../util';
2020
import { getLocalDocStateByParent } from './local-documents-helper';
2121

2222
const RxDocumentParent = createRxDocumentConstructor() as any;
@@ -125,9 +125,9 @@ const RxLocalDocumentPrototype: any = {
125125
// while still having the option to run a retry on conflicts
126126
while (!done) {
127127
const oldDocData = this._dataSync$.getValue();
128+
const newData = await mutationFunction(clone(oldDocData.data), this);
128129
try {
129130
// always await because mutationFunction might be async
130-
const newData = await mutationFunction(clone(oldDocData.data), this);
131131

132132
const newDocData = flatClone(oldDocData);
133133
newDocData.data = newData;
@@ -142,8 +142,10 @@ const RxLocalDocumentPrototype: any = {
142142
* Because atomicUpdate has a mutation function,
143143
* we can just re-run the mutation until there is no conflict
144144
*/
145-
if (isPouchdbConflictError(err as any)) {
146-
// pouchdb conflict error -> retrying
145+
const isConflict = isBulkWriteConflictError(err as any);
146+
if (isConflict) {
147+
// conflict error -> retrying
148+
newData._rev = createRevision(newData, isConflict.documentInDb);
147149
} else {
148150
rej(err);
149151
return;

‎src/plugins/pouchdb/custom-events-plugin.ts

+19-11
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
* Also we can better define what data we need for our events.
77
* @link http://jsbin.com/pagebi/1/edit?js,output
88
* @link https://github.com/pubkey/rxdb/blob/1f4115b69bdacbb853af9c637d70f5f184d4e474/src/rx-storage-pouchdb.ts#L273
9+
* @link https://hasura.io/blog/couchdb-style-conflict-resolution-rxdb-hasura/
910
*/
1011

1112
import type {
@@ -140,10 +141,14 @@ export function addCustomEventsPluginToPouch() {
140141
options: PouchBulkDocOptions,
141142
callback: Function
142143
) {
144+
143145
const startTime = now();
144146
const runId = i++;
145147

146-
// normalize input
148+
/**
149+
* Normalize inputs
150+
* because there are many ways to call pouchdb.bulkDocs()
151+
*/
147152
if (typeof options === 'function') {
148153
callback = options;
149154
options = {};
@@ -298,16 +303,15 @@ export function addCustomEventsPluginToPouch() {
298303
const useNewRev = useRevisions.start + '-' + newRev.hash;
299304

300305
hasNonErrorWrite = true;
301-
docs.push(
302-
Object.assign(
303-
{},
304-
insertDocsById.get(id),
305-
{
306-
_revisions: useRevisions,
307-
_rev: useNewRev
308-
}
309-
)
306+
const writeToPouchDocData = Object.assign(
307+
{},
308+
insertDocsById.get(id),
309+
{
310+
_revisions: useRevisions,
311+
_rev: useNewRev
312+
}
310313
);
314+
docs.push(writeToPouchDocData);
311315
usePouchResult.push({
312316
ok: true,
313317
id,
@@ -336,7 +340,8 @@ export function addCustomEventsPluginToPouch() {
336340
let callReturn: any;
337341
const callPromise = new Promise((res, rej) => {
338342
callReturn = oldBulkDocs.call(
339-
this, docs,
343+
this,
344+
docs,
340345
deeperOptions,
341346
(err: any, result: (PouchBulkDocResultRow | PouchWriteError)[]) => {
342347
if (err) {
@@ -396,6 +401,9 @@ export function addCustomEventsPluginToPouch() {
396401
if (options.custom) {
397402
return callPromise;
398403
}
404+
405+
406+
399407
return callReturn;
400408
};
401409

‎src/rx-collection.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -494,7 +494,7 @@ export class RxCollectionBase<
494494
}
495495
queue = queue
496496
.then(() => _atomicUpsertEnsureRxDocumentExists(this as any, primary as any, useJson))
497-
.then((wasInserted: any) => {
497+
.then((wasInserted) => {
498498
if (!wasInserted.inserted) {
499499
return _atomicUpsertUpdate(wasInserted.doc, useJson)
500500
.then(() => wasInserted.doc);

‎src/rx-document.ts

+36-12
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,15 @@ import {
1515
flatClone,
1616
PROMISE_RESOLVE_NULL,
1717
PROMISE_RESOLVE_VOID,
18-
ensureNotFalsy
18+
ensureNotFalsy,
19+
parseRevision,
20+
createRevision,
21+
promiseWait
1922
} from './util';
2023
import {
2124
newRxError,
2225
newRxTypeError,
23-
isPouchdbConflictError
26+
isBulkWriteConflictError
2427
} from './rx-error';
2528
import {
2629
runPluginHooks
@@ -321,34 +324,49 @@ export const basePrototype = {
321324
*/
322325
atomicUpdate(this: RxDocument, mutationFunction: Function): Promise<RxDocument> {
323326
return new Promise((res, rej) => {
324-
this._atomicQueue = this._atomicQueue
327+
this._atomicQueue = this
328+
._atomicQueue
325329
.then(async () => {
326330
let done = false;
327331
// we need a hacky while loop to stay incide the chain-link of _atomicQueue
328332
// while still having the option to run a retry on conflicts
329333
while (!done) {
330334
const oldData = this._dataSync$.getValue();
335+
// always await because mutationFunction might be async
336+
let newData;
337+
331338
try {
332-
// always await because mutationFunction might be async
333-
let newData = await mutationFunction(clone(this._dataSync$.getValue()), this);
339+
newData = await mutationFunction(
340+
clone(oldData),
341+
this
342+
);
334343
if (this.collection) {
335344
newData = this.collection.schema.fillObjectWithDefaults(newData);
336345
}
346+
} catch (err) {
347+
rej(err);
348+
return;
349+
}
337350

351+
try {
338352
await this._saveData(newData, oldData);
339353
done = true;
340-
} catch (err) {
354+
} catch (err: any) {
355+
const useError = err.parameters && err.parameters.error ? err.parameters.error : err;
341356
/**
342357
* conflicts cannot happen by just using RxDB in one process
343358
* There are two ways they still can appear which is
344359
* replication and multi-tab usage
345360
* Because atomicUpdate has a mutation function,
346361
* we can just re-run the mutation until there is no conflict
347362
*/
348-
if (isPouchdbConflictError(err as any)) {
349-
// pouchdb conflict error -> retrying
363+
const isConflict = isBulkWriteConflictError(useError as any);
364+
if (isConflict) {
365+
// conflict error -> retrying
366+
newData._rev = createRevision(newData, isConflict.documentInDb);
367+
await promiseWait(300);
350368
} else {
351-
rej(err);
369+
rej(useError);
352370
return;
353371
}
354372
}
@@ -385,7 +403,6 @@ export const basePrototype = {
385403
newData: RxDocumentWriteData<RxDocumentType>,
386404
oldData: RxDocumentData<RxDocumentType>
387405
): Promise<void> {
388-
newData = newData;
389406

390407
// deleted documents cannot be changed
391408
if (this._isDeleted$.getValue()) {
@@ -400,15 +417,22 @@ export const basePrototype = {
400417
await this.collection._runHooks('pre', 'save', newData, this);
401418
this.collection.schema.validate(newData);
402419

420+
421+
// TODO REMOVE THIS CHECK
422+
const p1 = parseRevision(oldData._rev);
423+
const p2 = parseRevision(newData._rev);
424+
newData._rev = createRevision(newData, oldData);
425+
if ((p1.height + 1 !== p2.height)) {
426+
// throw new Error('REVISION NOT INCREMENTED! ' + p1.height + ' ' + p2.height);
427+
}
428+
403429
const writeResult = await this.collection.storageInstance.bulkWrite([{
404430
previous: oldData,
405431
document: newData
406432
}]);
407433

408434
const isError = writeResult.error[this.primary];
409435
throwIfIsStorageWriteError(this.collection, this.primary, newData, isError);
410-
ensureNotFalsy(writeResult.success[this.primary]);
411-
412436

413437
return this.collection._runHooks('post', 'save', newData, this);
414438
},

‎src/rx-error.ts

+12-6
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@
55
import { overwritable } from './overwritable';
66
import type {
77
RxErrorParameters,
8-
PouchWriteError,
9-
RxErrorKey
8+
RxErrorKey,
9+
RxStorageBulkWriteError
1010
} from './types';
1111

1212
/**
@@ -122,12 +122,18 @@ export function newRxTypeError(
122122
);
123123
}
124124

125-
export function isPouchdbConflictError(err: RxError | RxTypeError): boolean {
125+
126+
/**
127+
* Returns the error if it is a 409 conflict,
128+
* return false if it is another error.
129+
*/
130+
export function isBulkWriteConflictError<RxDocType>(
131+
err: RxStorageBulkWriteError<RxDocType> | any
132+
): RxStorageBulkWriteError<RxDocType> | false {
126133
if (
127-
err.parameters && err.parameters.pouchDbError &&
128-
(err.parameters.pouchDbError as PouchWriteError).status === 409
134+
err.status === 409
129135
) {
130-
return true;
136+
return err;
131137
} else {
132138
return false;
133139
}

‎src/rx-storage-helper.ts

+6-1
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,7 @@ export function throwIfIsStorageWriteError<RxDocType>(
130130
throw newRxError('COL19', {
131131
collection: collection.name,
132132
id: documentId,
133-
pouchDbError: error,
133+
error,
134134
data: writeData
135135
});
136136
} else {
@@ -579,6 +579,11 @@ export function getWrappedStorageInstance<RxDocType, Internals, InstanceCreation
579579
runPluginHooks('preWriteToStorageInstance', hookParams);
580580
data = hookParams.doc;
581581

582+
583+
// console.log('----------------------');
584+
// console.dir(writeRow.previous);
585+
// console.dir(data);
586+
582587
/**
583588
* Update the revision after the hooks have run.
584589
* Do not update the revision if no previous is given,

‎src/types/pouch.d.ts

-1
Original file line numberDiff line numberDiff line change
@@ -236,7 +236,6 @@ export type PouchBulkDocOptions = {
236236
new_edits?: boolean;
237237

238238
// custom options for RxDB
239-
set_new_edit_as_latest_revision?: boolean;
240239
isDeeper?: boolean;
241240
custom?: any;
242241
}

‎src/types/rx-error.d.ts

+1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ export declare class RxTypeError extends TypeError {
2626
* this lists all possible parameters
2727
*/
2828
export interface RxErrorParameters {
29+
readonly error?: any;
2930
readonly errors?: RxErrorItem[];
3031
readonly schemaPath?: string;
3132
readonly objPath?: string;

‎src/util.ts

+25-2
Original file line numberDiff line numberDiff line change
@@ -432,10 +432,33 @@ export function createRevision<RxDocType>(
432432
const previousRevisionHeigth = previousRevision ? parseRevision(previousRevision).height : 0;
433433
const newRevisionHeight = previousRevisionHeigth + 1;
434434

435-
const docWithoutRev = Object.assign({}, docData, {
435+
436+
const docWithoutRev: any = Object.assign({}, docData, {
436437
_rev: undefined,
437-
_rev_tree: undefined
438+
_rev_tree: undefined,
439+
/**
440+
* All _meta properties MUST NOT be part of the
441+
* revision hash.
442+
* Plugins might temporarily store data in the _meta
443+
* field and strip it away when the document is replicated
444+
* or written to another storage.
445+
*/
446+
_meta: undefined
438447
});
448+
449+
/**
450+
* The revision height must be part of the hash
451+
* as the last parameter of the document data.
452+
* This is required to ensure we never ever create
453+
* two different document states that have the same revision
454+
* hash. Even writing the exact same document data
455+
* must have to result in a different hash so that
456+
* the replication can known if the state just looks equal
457+
* or if it is really exactly the equal state in data and time.
458+
*/
459+
delete docWithoutRev._rev;
460+
docWithoutRev._rev = previousDocData ? newRevisionHeight : 1;
461+
439462
const diggestString = JSON.stringify(docWithoutRev);
440463
const revisionHash = Md5.hash(diggestString);
441464

‎test/unit/rx-collection.test.ts

+15-5
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@ import {
2626
now,
2727
RxDocument,
2828
getFromMapOrThrow,
29-
RxCollectionCreator
29+
RxCollectionCreator,
30+
parseRevision
3031
} from '../../';
3132

3233
import {
@@ -1597,12 +1598,14 @@ config.parallel('rx-collection.test.js', () => {
15971598
assert.ok(isRxDocument(docs[0]));
15981599
c.database.destroy();
15991600
});
1600-
it('should not crash when upserting the same doc in parallel many times with random waits', async () => {
1601+
it('should not crash when upserting the same doc in parallel many times with random waits', async function () {
16011602
const c = await humansCollection.createPrimary(0);
16021603
const docData = schemaObjects.simpleHuman();
1604+
docData.firstName = 'test-many-atomic-upsert';
16031605

16041606
let t = 0;
1605-
const amount = config.isFastMode() ? 20 : 200;
1607+
const amount = config.isFastMode() ? 20 : 5;
1608+
16061609
const docs = await Promise.all(
16071610
new Array(amount)
16081611
.fill(0)
@@ -1611,7 +1614,7 @@ config.parallel('rx-collection.test.js', () => {
16111614
upsertData.lastName = idx + '';
16121615
const randomWait = randomBoolean() ? wait(randomNumber(0, 30)) : Promise.resolve();
16131616
return randomWait
1614-
.then(() => c.atomicUpsert(docData))
1617+
.then(() => c.atomicUpsert(upsertData))
16151618
.then(doc => {
16161619
t++;
16171620
return doc;
@@ -1624,22 +1627,29 @@ config.parallel('rx-collection.test.js', () => {
16241627

16251628
c.database.destroy();
16261629
});
1627-
it('should update the value', async () => {
1630+
it('should update the value', async function () {
16281631
const c = await humansCollection.createPrimary(0);
16291632
const docData = schemaObjects.simpleHuman();
1633+
const docId = docData.passportId;
16301634

16311635
await Promise.all([
16321636
c.atomicUpsert(docData),
16331637
c.atomicUpsert(docData),
16341638
c.atomicUpsert(docData)
16351639
]);
16361640

1641+
1642+
const viaStorage = await c.storageInstance.findDocumentsById([docId], true);
1643+
const viaStorageDoc = viaStorage[docId];
1644+
assert.strictEqual(parseRevision(viaStorageDoc._rev).height, 3);
1645+
16371646
const docData2 = clone(docData);
16381647
docData2.firstName = 'foobar';
16391648
await c.atomicUpsert(docData2);
16401649
const doc = await c.findOne().exec(true);
16411650
assert.strictEqual(doc.firstName, 'foobar');
16421651

1652+
16431653
c.database.destroy();
16441654
});
16451655
it('should work when upserting to existing document', async () => {

‎test/unit/rx-document.test.ts

+5-2
Original file line numberDiff line numberDiff line change
@@ -611,19 +611,22 @@ config.parallel('rx-document.test.js', () => {
611611
const col = await humansCollection.create(1);
612612
const doc = await col.findOne().exec(true);
613613

614+
// non-async mutation
614615
try {
615616
await doc.atomicUpdate(() => {
616-
throw new Error('ouch');
617+
throw new Error('throws intentional A');
617618
});
618619
} catch (err) { }
620+
619621
// async mutation
620622
try {
621623
await doc.atomicUpdate(async () => {
622624
await wait(10);
623-
throw new Error('ouch');
625+
throw new Error('throws intentional B');
624626
});
625627
} catch (err) { }
626628

629+
// non throwing mutation
627630
await doc.atomicUpdate(d => {
628631
d.age = 150;
629632
return d;

‎test/unit/rx-storage-implementations.test.ts

+68
Original file line numberDiff line numberDiff line change
@@ -642,6 +642,74 @@ config.parallel('rx-storage-implementations.test.js (implementation: ' + config.
642642

643643
storageInstance.close();
644644
});
645+
it('should be able to do a write where only _meta fields are changed', async () => {
646+
const storageInstance = await config.storage.getStorage().createStorageInstance<TestDocType>({
647+
databaseName: randomCouchString(12),
648+
collectionName: randomCouchString(12),
649+
schema: getPseudoSchemaForVersion<TestDocType>(0, 'key'),
650+
options: {},
651+
multiInstance: false
652+
});
653+
654+
const key = 'foobar';
655+
let docData: RxDocumentWriteData<TestDocType> = {
656+
key,
657+
value: 'barfoo1',
658+
_attachments: {},
659+
_deleted: false,
660+
_rev: EXAMPLE_REVISION_1,
661+
_meta: {
662+
lwt: now(),
663+
foobar: 0
664+
}
665+
};
666+
docData._rev = createRevision(docData);
667+
668+
const res1 = await storageInstance.bulkWrite(
669+
[{
670+
document: clone(docData)
671+
}]
672+
);
673+
assert.deepStrictEqual(res1.error, {});
674+
675+
// change once
676+
let newDocData: RxDocumentWriteData<TestDocType> = clone(docData);
677+
newDocData._meta.foobar = 1;
678+
newDocData._meta.lwt = now();
679+
newDocData._rev = createRevision(newDocData, docData);
680+
681+
const res2 = await storageInstance.bulkWrite(
682+
[{
683+
previous: docData,
684+
document: clone(newDocData)
685+
}]
686+
);
687+
assert.deepStrictEqual(res2.error, {});
688+
docData = newDocData;
689+
690+
// change again
691+
newDocData = clone(docData);
692+
newDocData._meta.foobar = 2;
693+
newDocData._meta.lwt = now();
694+
newDocData._rev = createRevision(newDocData, docData);
695+
assert.strictEqual(parseRevision(newDocData._rev).height, 3);
696+
697+
const res3 = await storageInstance.bulkWrite(
698+
[{
699+
previous: docData,
700+
document: clone(newDocData)
701+
}]
702+
);
703+
assert.deepStrictEqual(res3.error, {});
704+
docData = newDocData;
705+
706+
707+
const viaStorage = await storageInstance.findDocumentsById([key], true);
708+
const viaStorageDoc = ensureNotFalsy(viaStorage[key]);
709+
assert.strictEqual(parseRevision(viaStorageDoc._rev).height, 3);
710+
711+
storageInstance.close();
712+
});
645713
it('should be able to create another instance after a write', async () => {
646714
const databaseName = randomCouchString(12);
647715
const storageInstance = await config.storage.getStorage().createStorageInstance<TestDocType>({

‎test/unit/util.test.ts

-13
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,6 @@ import {
2020
validateDatabaseName,
2121
deepFreezeWhenDevMode
2222
} from '../../plugins/dev-mode';
23-
import {
24-
rev as pouchCreateRevisison
25-
} from 'pouchdb-utils';
2623
import { EXAMPLE_REVISION_1 } from '../helper/revisions';
2724

2825
describe('util.test.js', () => {
@@ -77,16 +74,6 @@ describe('util.test.js', () => {
7774
});
7875
assert.strictEqual(hash1, hash2);
7976
});
80-
it('should return the same value as pouchdb', async () => {
81-
const docData = {
82-
foo: 'bar',
83-
bar: 'foo',
84-
_rev_tree: '1-asdfasdf'
85-
};
86-
const ownRev = createRevision(docData as any);
87-
const pouchRev = '1-' + pouchCreateRevisison(docData, true);
88-
assert.strictEqual(ownRev, pouchRev);
89-
});
9077
});
9178
describe('.sortObject()', () => {
9279
it('should sort when regex in object', () => {

0 commit comments

Comments
 (0)
Please sign in to comment.