Skip to content

Commit d21f2eb

Browse files
fix: crud upload queue monitoring
1 parent d5f755d commit d21f2eb

File tree

6 files changed

+170
-22
lines changed

6 files changed

+170
-22
lines changed

.changeset/afraid-apples-learn.md

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
---
2+
'@powersync/common': patch
3+
'@powersync/web': patch
4+
'@powersync/react-native': patch
5+
---
6+
7+
Fixed issue where sequentially mutating the same row multiple times could cause the CRUD upload queue monitoring to think CRUD operations have not been processed correctly by the `BackendConnector` `uploadData` method.

packages/common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ import Logger, { ILogger } from 'js-logger';
33
import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus.js';
44
import { AbortOperation } from '../../../utils/AbortOperation.js';
55
import { BaseListener, BaseObserver, Disposable } from '../../../utils/BaseObserver.js';
6+
import { throttleLeadingTrailing } from '../../../utils/throttle.js';
67
import { BucketChecksum, BucketStorageAdapter, Checkpoint } from '../bucket/BucketStorageAdapter.js';
78
import { CrudEntry } from '../bucket/CrudEntry.js';
89
import { SyncDataBucket } from '../bucket/SyncDataBucket.js';
@@ -16,7 +17,6 @@ import {
1617
isStreamingSyncCheckpointDiff,
1718
isStreamingSyncData
1819
} from './streaming-sync-types.js';
19-
import { throttleLeadingTrailing } from '../../../utils/throttle.js';
2020

2121
export enum LockType {
2222
CRUD = 'crud',
@@ -230,7 +230,7 @@ export abstract class AbstractStreamingSyncImplementation
230230
*/
231231
const nextCrudItem = await this.options.adapter.nextCrudItem();
232232
if (nextCrudItem) {
233-
if (nextCrudItem.id == checkedCrudItem?.id) {
233+
if (nextCrudItem.clientId == checkedCrudItem?.clientId) {
234234
// This will force a higher log level than exceptions which are caught here.
235235
this.logger.warn(`Potentially previously uploaded CRUD entries are still present in the upload queue.
236236
Make sure to handle uploads and complete CRUD transactions or batches by calling and awaiting their [.complete()] method.

packages/web/src/db/PowerSyncDatabase.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,8 @@ import {
1616
import { Mutex } from 'async-mutex';
1717
import { WASQLiteOpenFactory } from './adapters/wa-sqlite/WASQLiteOpenFactory';
1818
import {
19-
ResolvedWebSQLOpenOptions,
2019
DEFAULT_WEB_SQL_FLAGS,
20+
ResolvedWebSQLOpenOptions,
2121
resolveWebSQLFlags,
2222
WebSQLFlags
2323
} from './adapters/web-sql-flags';

packages/web/tests/stream.test.ts

Lines changed: 42 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,39 @@
1-
import { Schema, TableV2, column } from '@powersync/common';
1+
import { Schema, Table, column } from '@powersync/common';
2+
import { WebPowerSyncOpenFactoryOptions } from '@powersync/web';
23
import Logger from 'js-logger';
34
import { v4 as uuid } from 'uuid';
45
import { beforeAll, describe, expect, it, vi } from 'vitest';
56
import { MockRemote, MockStreamOpenFactory, TestConnector } from './utils/MockStreamOpenFactory';
67

8+
type UnwrapPromise<T> = T extends Promise<infer U> ? U : T;
9+
10+
export type ConnectedDatabaseUtils = UnwrapPromise<ReturnType<typeof generateConnectedDatabase>>;
11+
export type GenerateConnectedDatabaseOptions = {
12+
powerSyncOptions: Partial<WebPowerSyncOpenFactoryOptions>;
13+
};
14+
715
const UPLOAD_TIMEOUT_MS = 3000;
816

9-
export async function generateConnectedDatabase({ useWebWorker } = { useWebWorker: true }) {
17+
export const DEFAULT_CONNECTED_POWERSYNC_OPTIONS = {
18+
powerSyncOptions: {
19+
dbFilename: 'test-stream-connection.db',
20+
flags: {
21+
enableMultiTabs: false,
22+
useWebWorker: true
23+
},
24+
// Makes tests faster
25+
crudUploadThrottleMs: 0,
26+
schema: new Schema({
27+
users: new Table({ name: column.text })
28+
})
29+
}
30+
};
31+
32+
export async function generateConnectedDatabase(
33+
options: GenerateConnectedDatabaseOptions = DEFAULT_CONNECTED_POWERSYNC_OPTIONS
34+
) {
35+
const { powerSyncOptions } = options;
36+
const { powerSyncOptions: defaultPowerSyncOptions } = DEFAULT_CONNECTED_POWERSYNC_OPTIONS;
1037
/**
1138
* Very basic implementation of a listener pattern.
1239
* Required since we cannot extend multiple classes.
@@ -16,24 +43,14 @@ export async function generateConnectedDatabase({ useWebWorker } = { useWebWorke
1643
const uploadSpy = vi.spyOn(connector, 'uploadData');
1744
const remote = new MockRemote(connector, () => callbacks.forEach((c) => c()));
1845

19-
const users = new TableV2({
20-
name: column.text
21-
});
22-
23-
const schema = new Schema({
24-
users
25-
});
26-
2746
const factory = new MockStreamOpenFactory(
2847
{
29-
dbFilename: 'test-stream-connection.db',
48+
...defaultPowerSyncOptions,
49+
...powerSyncOptions,
3050
flags: {
31-
enableMultiTabs: false,
32-
useWebWorker
33-
},
34-
// Makes tests faster
35-
crudUploadThrottleMs: 0,
36-
schema
51+
...(defaultPowerSyncOptions.flags ?? {}),
52+
...(powerSyncOptions.flags ?? {})
53+
}
3754
},
3855
remote
3956
);
@@ -83,7 +100,14 @@ describe('Streaming', () => {
83100
test: (createConnectedDatabase: () => ReturnType<typeof generateConnectedDatabase>) => Promise<void>
84101
) => {
85102
const funcWithWebWorker = generateConnectedDatabase;
86-
const funcWithoutWebWorker = () => generateConnectedDatabase({ useWebWorker: false });
103+
const funcWithoutWebWorker = () =>
104+
generateConnectedDatabase({
105+
powerSyncOptions: {
106+
flags: {
107+
useWebWorker: false
108+
}
109+
}
110+
});
87111

88112
it(`${name} - with web worker`, () => test(funcWithWebWorker));
89113
it(`${name} - without web worker`, () => test(funcWithoutWebWorker));

packages/web/tests/uploads.test.ts

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
import Logger from 'js-logger';
2+
import p from 'p-defer';
3+
import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
4+
import { ConnectedDatabaseUtils, generateConnectedDatabase } from './stream.test';
5+
6+
describe('CRUD Uploads', () => {
7+
let connectedUtils: ConnectedDatabaseUtils;
8+
const logger = Logger.get('crud-logger');
9+
10+
beforeAll(() => Logger.useDefaults());
11+
12+
beforeEach(async () => {
13+
connectedUtils = await generateConnectedDatabase({
14+
powerSyncOptions: {
15+
logger,
16+
/**
17+
* The timeout here is set to longer than the default test timeout
18+
* A retry wil cause tests to fail.
19+
*/
20+
crudUploadThrottleMs: 10_000,
21+
flags: {
22+
enableMultiTabs: false
23+
}
24+
}
25+
});
26+
});
27+
28+
afterEach(async () => {
29+
connectedUtils.remote.streamController?.close();
30+
await connectedUtils.powersync.disconnectAndClear();
31+
await connectedUtils.powersync.close();
32+
});
33+
34+
it('should warn for missing upload operations in uploadData', async () => {
35+
const { powersync, uploadSpy } = connectedUtils;
36+
const loggerSpy = vi.spyOn(logger, 'warn');
37+
38+
const deferred = p();
39+
40+
uploadSpy.mockImplementation(async (db) => {
41+
// This upload method does not perform an upload
42+
deferred.resolve();
43+
});
44+
45+
// Create something with CRUD in it.
46+
await powersync.execute('INSERT into users (id, name) VALUES (uuid(), ?)', ['steven']);
47+
48+
// The empty upload handler should have been called
49+
// Timeouts seem to be weird in Vitest Browser mode.
50+
// This makes the check below more stable.
51+
await deferred.promise;
52+
53+
await vi.waitFor(
54+
() => {
55+
expect(
56+
loggerSpy.mock.calls.find((logArgs) =>
57+
logArgs[0].includes('Potentially previously uploaded CRUD entries are still present')
58+
)
59+
).exist;
60+
},
61+
{
62+
timeout: 500
63+
}
64+
);
65+
});
66+
67+
it('should immediately upload sequential transactions', async () => {
68+
const { powersync, uploadSpy } = connectedUtils;
69+
const deferred = p();
70+
71+
uploadSpy.mockImplementation(async (db) => {
72+
// This upload method does not perform an upload
73+
const nextTransaction = await db.getNextCrudTransaction();
74+
console.log('uploading trans', nextTransaction);
75+
if (!nextTransaction) {
76+
return;
77+
}
78+
79+
// Mockingly delete the crud op in order to progress through the CRUD queue
80+
for (const op of nextTransaction.crud) {
81+
await db.execute(`DELETE FROM ps_crud WHERE id = ?`, [op.clientId]);
82+
}
83+
84+
deferred.resolve();
85+
});
86+
87+
// Create the first item
88+
await powersync.execute('INSERT into users (id, name) VALUES (uuid(), ?)', ['steven']);
89+
90+
// Modify the first item in a new transaction
91+
await powersync.execute(`
92+
UPDATE
93+
users
94+
SET
95+
name = 'Mugi'
96+
WHERE
97+
name = 'steven'`);
98+
99+
// Create a second item
100+
await powersync.execute('INSERT into users (id, name) VALUES (uuid(), ?)', ['steven2']);
101+
102+
// The empty upload handler should have been called
103+
// Timeouts seem to be weird in Vitest Browser mode.
104+
// This makes the check below more stable.
105+
await deferred.promise;
106+
107+
await vi.waitFor(
108+
() => {
109+
expect(uploadSpy.mock.calls.length).eq(3);
110+
},
111+
{
112+
timeout: 5_000
113+
}
114+
);
115+
});
116+
});

packages/web/tests/utils/MockStreamOpenFactory.ts

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -133,14 +133,15 @@ export class MockedStreamPowerSync extends PowerSyncDatabase {
133133
connector: PowerSyncBackendConnector
134134
): AbstractStreamingSyncImplementation {
135135
return new WebStreamingSyncImplementation({
136+
logger: this.options.logger,
136137
adapter: this.bucketStorageAdapter,
137138
remote: this.remote,
138139
uploadCrud: async () => {
139140
await this.waitForReady();
140141
await connector.uploadData(this);
141142
},
142143
identifier: this.database.name,
143-
retryDelayMs: 0
144+
retryDelayMs: this.options.crudUploadThrottleMs ?? 0 // The zero here makes tests faster
144145
});
145146
}
146147
}

0 commit comments

Comments
 (0)