mirror of
https://github.com/toeverything/AFFiNE.git
synced 2026-09-13 04:42:56 +08:00
fix(server): wrap updates applying in a transaction (#4922)
This commit is contained in:
@@ -8,7 +8,13 @@ import {
|
|||||||
import { Snapshot, Update } from '@prisma/client';
|
import { Snapshot, Update } from '@prisma/client';
|
||||||
import { chunk } from 'lodash-es';
|
import { chunk } from 'lodash-es';
|
||||||
import { defer, retry } from 'rxjs';
|
import { defer, retry } from 'rxjs';
|
||||||
import { applyUpdate, Doc, encodeStateAsUpdate, encodeStateVector } from 'yjs';
|
import {
|
||||||
|
applyUpdate,
|
||||||
|
Doc,
|
||||||
|
encodeStateAsUpdate,
|
||||||
|
encodeStateVector,
|
||||||
|
transact,
|
||||||
|
} from 'yjs';
|
||||||
|
|
||||||
import { Config } from '../../config';
|
import { Config } from '../../config';
|
||||||
import { Metrics } from '../../metrics/metrics';
|
import { Metrics } from '../../metrics/metrics';
|
||||||
@@ -84,16 +90,18 @@ export class DocManager implements OnModuleInit, OnModuleDestroy {
|
|||||||
const next = () => {
|
const next = () => {
|
||||||
const updates = chunks.shift();
|
const updates = chunks.shift();
|
||||||
if (updates?.length) {
|
if (updates?.length) {
|
||||||
updates.forEach(u => {
|
transact(doc, () => {
|
||||||
try {
|
updates.forEach(u => {
|
||||||
applyUpdate(doc, u);
|
try {
|
||||||
} catch (e) {
|
applyUpdate(doc, u);
|
||||||
this.logger.error(
|
} catch (e) {
|
||||||
`Failed to apply update: ${updates
|
this.logger.error(
|
||||||
.map(u => u.toString('hex'))
|
`Failed to apply update: ${updates
|
||||||
.join('\n')}`
|
.map(u => u.toString('hex'))
|
||||||
);
|
.join('\n')}`
|
||||||
}
|
);
|
||||||
|
}
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
// avoid applying too many updates in single round which will take the whole cpu time like dead lock
|
// avoid applying too many updates in single round which will take the whole cpu time like dead lock
|
||||||
|
|||||||
Reference in New Issue
Block a user