Skip to content

Commit 5deaaa6

Browse files
lastsundaylastsunday
committed
fix(extension): pglite transaction error
1 parent 227cda6 commit 5deaaa6

11 files changed

Lines changed: 136 additions & 80 deletions

File tree

‎apps/extension/entrypoints/offscreen/worker/connectionManager.js‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88
* 其他属性（close、closed、on 等）直接透传原始实例。
99
*/
1010

11+
import { debugLog, errorLog } from '@/common/log';
12+
1113
const STATE = {
1214
UNINITIALIZED: 0,
1315
INITIALIZING: 1,
@@ -109,6 +111,12 @@ class ConnectionManager {
109111
resolve(await target[method](...args));
110112
} catch (e) {
111113
reject(e);
114+
errorLog('[ConnectionManager] operation failed, attempting ROLLBACK: ' + e.message);
115+
try {
116+
await target.exec('ROLLBACK');
117+
} catch (re) {
118+
debugLog('[ConnectionManager] ROLLBACK after error: ' + re.message);
119+
}
112120
}
113121
}
114122
this.#busy = false;

‎apps/extension/entrypoints/offscreen/worker/database.js‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -168,7 +168,7 @@ export async function batchInsertOrReplace(
168168
for (let i = 0; i < count; i++) {
169169
const start = i * maxRecordCountForOneExec;
170170
let end = (i + 1) * maxRecordCountForOneExec;
171-
if (i == count.length - 1) {
171+
if (i == count - 1) {
172172
//last index
173173
end = recordTotal;
174174
}
@@ -673,6 +673,8 @@ const initDb = async function ({ dataDir = `opfs-ahp://${JOB_DB_PATH}` } = {}) {
673673
});
674674
} catch (e) {
675675
errorLog('[DB] schema upgrade fail,' + e.message);
676+
await db.close();
677+
throw e;
676678
}
677679
return db;
678680
};

‎apps/extension/entrypoints/offscreen/worker/service/baseBridgeService.js‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,9 +44,10 @@ export const addTransactionServiceMethod = ({ bridgeService = null, methodName =
4444
bridgeService.methodNameMapping.set(methodName, targetMethodName);
4545
bridgeService[targetMethodName] = async (message, param) => {
4646
try {
47-
await (await getDb()).transaction(async (tx) => {
48-
await postSuccessMessage(message, await methodFunction({ param, tx }));
47+
const result = await (await getDb()).transaction(async (tx) => {
48+
return await methodFunction({ param, tx });
4949
});
50+
await postSuccessMessage(message, result);
5051
} catch (e) {
5152
await postErrorMessage(message, `[worker] ${targetMethodName} error : ` + e);
5253
}

‎apps/extension/entrypoints/offscreen/worker/service/companyTagService.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -412,12 +412,12 @@ export async function _batchAddOrUpdateCompanyTag({ companyTagBOs = null, overri
412412
}
413413
sourceTypeSourceAndIdsMap.get(key).push(genIdFromText(item.companyName));
414414
}
415-
sourceTypeSourceAndIdsMap.forEach(async (value, key, map) => {
415+
for (const [key] of sourceTypeSourceAndIdsMap) {
416416
const ids = sourceTypeSourceAndIdsMap.get(key);
417417
const sourceType = sourceTypeSourceAndSourceTypeMap.get(key);
418418
const source = sourceTypeSourceAndSourceMap.get(key);
419419
await SERVICE_INSTANCE._deleteByIds(ids, COMPANY_ID_COLUMN, { connection, otherCondition: `source_type=${sourceType} AND ${source ? "source = '" + source + "'" : "source IS NULL"}` });
420-
});
420+
}
421421
const companyTags = [];
422422
for (let i = 0; i < companyTagBOs.length; i++) {
423423
const item = companyTagBOs[i];

‎apps/extension/entrypoints/offscreen/worker/service/jobPublicService.js‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,12 +49,12 @@ export const _jobPublicBatchAddJobPublicAndUpdateJob = async (param, { connectio
4949
value.jobIdList.push(item.jobId);
5050
jobPublicSourceAndJobIdMap.set(key, value);
5151
});
52-
jobPublicSourceAndJobIdMap.values().forEach(item => {
53-
SERVICE_INSTANCE._deleteByIds(item.jobIdList, "job_id", {
52+
for (const item of jobPublicSourceAndJobIdMap.values()) {
53+
await SERVICE_INSTANCE._deleteByIds(item.jobIdList, "job_id", {
5454
connection,
5555
otherCondition: `source_type = ${item.sourceType} AND ${item.source ? `source = '${item.source}'` : `source is null`} `
5656
});
57-
});
57+
}
5858
await SERVICE_INSTANCE._batchAddOrUpdate(jobPublicList, { overrideCreateDatetime: true, overrideUpdateDatetime: true, connection });
5959
}
6060
if (jobList && jobList.length > 0) {

‎apps/extension/entrypoints/offscreen/worker/service/jobSnapshotService.js‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,8 +100,8 @@ export const JobSnapshotService = {
100100
try {
101101
await (await getDb()).transaction(async (tx) => {
102102
await SERVICE_INSTANCE._batchAddOrUpdate(param.items, { overrideUpdateDatetime: param.overrideUpdateDatetime, connection: tx })
103-
postSuccessMessage(message, {});
104103
});
104+
postSuccessMessage(message, {});
105105
} catch (e) {
106106
postErrorMessage(
107107
message,

‎apps/extension/entrypoints/offscreen/worker/service/jobTagService.js‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -360,12 +360,12 @@ export async function _jobTagBatchAddOrUpdate(jobTagBOs, overrideUpdateDatetime,
360360
}
361361
sourceTypeSourceAndJobIdsMap.get(key).push(item.jobId);
362362
}
363-
sourceTypeSourceAndJobIdsMap.forEach(async (value, key, map) => {
363+
for (const [key] of sourceTypeSourceAndJobIdsMap) {
364364
const ids = sourceTypeSourceAndJobIdsMap.get(key);
365365
const sourceType = sourceTypeSourceAndSourceTypeMap.get(key);
366366
const source = sourceTypeSourceAndSourceMap.get(key);
367367
await SERVICE_INSTANCE._deleteByIds(ids, JOB_ID_COLUMN, { connection, otherCondition: `source_type=${sourceType} AND ${source ? "source = '" + source + "'" : "source IS NULL"}` });
368-
});
368+
}
369369
const jobTags = [];
370370
for (let i = 0; i < jobTagBOs.length; i++) {
371371
const item = jobTagBOs[i];

‎apps/extension/tests/worker/database.test.js‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -300,3 +300,17 @@ test('database init correct', async () => {
300300
});
301301
});
302302
});
303+
304+
test('connection recovers after failed transaction', async () => {
305+
const db = await getDb({ dataDir: 'memory://' });
306+
expect(db).toBeInstanceOf(PGlite);
307+
308+
await expect(
309+
db.transaction(async (tx) => {
310+
await tx.exec('INVALID SQL SYNTAX');
311+
})
312+
).rejects.toThrow();
313+
314+
const { rows } = await db.query('SELECT 1 AS val');
315+
expect(rows[0].val).toBe(1);
316+
});

‎apps/extension/tests/worker/service/companyTagService.test.js‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
import * as modUtil from "@/common/extension/worker/util";
2+
import { getDb } from "@/entrypoints/offscreen/worker/database";
3+
import { _batchAddOrUpdateCompanyTag } from "@/entrypoints/offscreen/worker/service/companyTagService";
4+
import { BaseService } from "@/entrypoints/offscreen/worker/service/baseService";
5+
import { PGlite } from "@electric-sql/pglite";
6+
import { expect, test, vi } from "vitest";
7+
8+
test('_batchAddOrUpdateCompanyTag executes deletes sequentially, not concurrently', async () => {
9+
vi.spyOn(modUtil, 'postErrorMessage').mockImplementation(async (message, error) => {
10+
throw error;
11+
});
12+
const db = await getDb({ dataDir: 'memory://' });
13+
expect(db).toBeInstanceOf(PGlite);
14+
15+
const callOrder = [];
16+
const originalDeleteByIds = BaseService.prototype._deleteByIds;
17+
vi.spyOn(BaseService.prototype, '_deleteByIds').mockImplementation(async function (...args) {
18+
callOrder.push('del_start');
19+
const result = await originalDeleteByIds.apply(this, args);
20+
callOrder.push('del_end');
21+
return result;
22+
});
23+
24+
const items = [
25+
{ companyName: 'CorpA', sourceType: 0, source: null, tags: ['startup', 'fintech'], updateDatetime: new Date().toISOString() },
26+
{ companyName: 'CorpB', sourceType: 0, source: 'linkedin', tags: ['enterprise'], updateDatetime: new Date().toISOString() },
27+
];
28+
29+
await (await getDb()).transaction(async (tx) => {
30+
await _batchAddOrUpdateCompanyTag({ companyTagBOs: items, overrideUpdateDatetime: false, connection: tx });
31+
});
32+
33+
for (let i = 0; i < callOrder.length - 1; i++) {
34+
if (callOrder[i] === 'del_start') {
35+
expect(callOrder[i + 1]).not.toBe('del_start');
36+
}
37+
}
38+
});

‎apps/extension/tests/worker/service/jobPublicService.test.js‎

Lines changed: 24 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -1,80 +1,35 @@
11
import * as modUtil from "@/common/extension/worker/util";
2-
import { parse } from "@/common/utils/date";
32
import { getDb } from "@/entrypoints/offscreen/worker/database";
4-
import {
5-
METHOD_ADD_OR_UPDATE, METHOD_BATCH_ADD_OR_UPDATE, METHOD_DELETE_BY_ID, METHOD_DELETE_BY_IDS,
6-
METHOD_GET_BY_ID, METHOD_GET_BY_IDS, METHOD_SEARCH
7-
} from "@/entrypoints/offscreen/worker/service/baseBridgeService";
8-
import JobPublicService from "@/entrypoints/offscreen/worker/service/jobPublicService";
3+
import { _jobPublicBatchAddJobPublicAndUpdateJob } from "@/entrypoints/offscreen/worker/service/jobPublicService";
94
import { PGlite } from "@electric-sql/pglite";
105
import { expect, test, vi } from "vitest";
11-
test('job public service method name correct', async () => {
12-
const result = JobPublicService.getMethodNameMap();
13-
expect(result.size).greaterThan(0);
14-
});
156

16-
test('job public service crud logic correct', async () => {
17-
const item = { id: null, jobId: "jobId1", createDatetime: parse("2025-05-18"), updateDatetime: parse("2025-05-18") };
18-
const item2 = { id: null, jobId: "jobId2", createDatetime: parse("2025-05-19"), updateDatetime: parse("2025-05-19") };
7+
test('forEach bug: source with single quote no longer masks error as "transaction aborted"', async () => {
198
vi.spyOn(modUtil, 'postErrorMessage').mockImplementation(async (message, error) => {
209
throw error;
2110
});
2211
const db = await getDb({ dataDir: 'memory://' });
2312
expect(db).toBeInstanceOf(PGlite);
24-
let itemId = null;
25-
let itemId2 = null;
26-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
27-
const item = data;
28-
expect(item.jobId).toBe(("jobId1"));
29-
expect(parse(item.createDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
30-
expect(parse(item.updateDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
31-
itemId = item.id;
32-
});
33-
await JobPublicService[JobPublicService.getMethodName(METHOD_ADD_OR_UPDATE)]({}, item);
34-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
35-
const item = data[0];
36-
expect(item.jobId).toBe(("jobId2"));
37-
expect(parse(item.createDatetime).valueOf()).toMatchObject(parse("2025-05-19").valueOf());
38-
expect(parse(item.updateDatetime).valueOf()).toMatchObject(parse("2025-05-19").valueOf());
39-
itemId2 = item.id;
40-
});
41-
await JobPublicService[JobPublicService.getMethodName(METHOD_BATCH_ADD_OR_UPDATE)]({}, { items: [item2] });
42-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
43-
let item = data.items[0];
44-
expect(item.jobId).toBe(("jobId1"));
45-
expect(parse(item.createDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
46-
expect(parse(item.updateDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
47-
expect(data.total).toBe(2);
48-
});
49-
await JobPublicService[JobPublicService.getMethodName(METHOD_SEARCH)]({}, { pageNum: 1, pageSize: 1 });
50-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
51-
let item = data;
52-
expect(item.jobId).toBe(("jobId1"));
53-
expect(parse(item.createDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
54-
expect(parse(item.updateDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
55-
});
56-
await JobPublicService[JobPublicService.getMethodName(METHOD_GET_BY_ID)]({}, itemId);
57-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
58-
let item = data[0];
59-
expect(item.jobId).toBe(("jobId1"));
60-
expect(parse(item.createDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
61-
expect(parse(item.updateDatetime).valueOf()).toMatchObject(parse("2025-05-18").valueOf());
62-
});
63-
await JobPublicService[JobPublicService.getMethodName(METHOD_GET_BY_IDS)]({}, [itemId]);
64-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
65-
expect(data[0].affectedRows).toBe(1);
66-
});
67-
await JobPublicService[JobPublicService.getMethodName(METHOD_DELETE_BY_ID)]({}, itemId);
68-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
69-
expect(data.total).toBe(1);
70-
});
71-
await JobPublicService[JobPublicService.getMethodName(METHOD_SEARCH)]({}, { pageNum: 1, pageSize: 1 });
72-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
73-
expect(data[0].affectedRows).toBe(1);
74-
});
75-
await JobPublicService[JobPublicService.getMethodName(METHOD_DELETE_BY_IDS)]({}, [itemId2]);
76-
vi.spyOn(modUtil, 'postSuccessMessage').mockImplementation(async (message, data) => {
77-
expect(data.total).toBe(0);
13+
14+
await db.exec(`INSERT INTO job_public (id, job_id, source_type, source, create_datetime, update_datetime) VALUES
15+
('old-x', 'job-x', 0, NULL, NOW(), NOW()),
16+
('old-y', 'job-y', 0, 'normal', NOW(), NOW())`);
17+
18+
const error = await (await getDb()).transaction(async (tx) => {
19+
try {
20+
await _jobPublicBatchAddJobPublicAndUpdateJob({
21+
jobPublicList: [
22+
{ id: 'new-x', jobId: 'job-x', sourceType: 0, source: null, createDatetime: new Date().toISOString(), updateDatetime: new Date().toISOString() },
23+
{ id: 'new-y', jobId: 'job-y', sourceType: 0, source: "it's", createDatetime: new Date().toISOString(), updateDatetime: new Date().toISOString() },
24+
]
25+
}, { connection: tx });
26+
return null;
27+
} catch (e) {
28+
return e.message;
29+
}
7830
});
79-
await JobPublicService[JobPublicService.getMethodName(METHOD_SEARCH)]({}, { pageNum: 1, pageSize: 1 });
80-
})
31+
32+
expect(error).not.toBeNull();
33+
expect(error).not.toContain('current transaction is aborted');
34+
expect(error).toContain('syntax error');
35+
});

0 commit comments

Comments
 (0)