Handle sync ordering vs back-fill ordering (see synapse #7164) resulting in dupe events
Also extra robustness while filling a gap ignore duplicate events from synapse instead considering them an overlapping event with the adjacent fragment
This commit is contained in:
parent
9b2f282534
commit
73ea09f668
3 changed files with 85 additions and 18 deletions
|
@ -17,6 +17,10 @@ export default class Direction {
|
||||||
return this.isForward ? "f" : "b";
|
return this.isForward ? "f" : "b";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
reverse() {
|
||||||
|
return this.isForward ? Direction.Backward : Direction.Forward
|
||||||
|
}
|
||||||
|
|
||||||
static get Forward() {
|
static get Forward() {
|
||||||
return _forward;
|
return _forward;
|
||||||
}
|
}
|
||||||
|
|
|
@ -37,7 +37,11 @@ export default class EventKey {
|
||||||
}
|
}
|
||||||
|
|
||||||
static get defaultLiveKey() {
|
static get defaultLiveKey() {
|
||||||
return new EventKey(Platform.minStorageKey, Platform.middleStorageKey);
|
return EventKey.defaultFragmentKey(Platform.minStorageKey);
|
||||||
|
}
|
||||||
|
|
||||||
|
static defaultFragmentKey(fragmentId) {
|
||||||
|
return new EventKey(fragmentId, Platform.middleStorageKey);
|
||||||
}
|
}
|
||||||
|
|
||||||
toString() {
|
toString() {
|
||||||
|
|
|
@ -10,30 +10,75 @@ export default class GapWriter {
|
||||||
}
|
}
|
||||||
// events is in reverse-chronological order (last event comes at index 0) if backwards
|
// events is in reverse-chronological order (last event comes at index 0) if backwards
|
||||||
async _findOverlappingEvents(fragmentEntry, events, txn) {
|
async _findOverlappingEvents(fragmentEntry, events, txn) {
|
||||||
const eventIds = events.map(e => e.event_id);
|
let expectedOverlappingEventId;
|
||||||
let nonOverlappingEvents = events;
|
if (fragmentEntry.hasLinkedFragment) {
|
||||||
|
expectedOverlappingEventId = await this._findExpectedOverlappingEventId(fragmentEntry, txn);
|
||||||
|
}
|
||||||
|
let remainingEvents = events;
|
||||||
|
let nonOverlappingEvents = [];
|
||||||
let neighbourFragmentEntry;
|
let neighbourFragmentEntry;
|
||||||
const neighbourEventId = await txn.timelineEvents.findFirstOccurringEventId(this._roomId, eventIds);
|
while (remainingEvents && remainingEvents.length) {
|
||||||
if (neighbourEventId) {
|
const eventIds = remainingEvents.map(e => e.event_id);
|
||||||
// trim overlapping events
|
const duplicateEventId = await txn.timelineEvents.findFirstOccurringEventId(this._roomId, eventIds);
|
||||||
const neighbourEventIndex = events.findIndex(e => e.event_id === neighbourEventId);
|
if (duplicateEventId) {
|
||||||
nonOverlappingEvents = events.slice(0, neighbourEventIndex);
|
const duplicateEventIndex = remainingEvents.findIndex(e => e.event_id === duplicateEventId);
|
||||||
|
// should never happen, just being defensive as this *can't* go wrong
|
||||||
|
if (duplicateEventIndex === -1) {
|
||||||
|
throw new Error(`findFirstOccurringEventId returned ${duplicateEventIndex} which wasn't ` +
|
||||||
|
`in [${eventIds.join(",")}] in ${this._roomId}`);
|
||||||
|
}
|
||||||
|
nonOverlappingEvents.push(...remainingEvents.slice(0, duplicateEventIndex));
|
||||||
|
if (!expectedOverlappingEventId || duplicateEventId === expectedOverlappingEventId) {
|
||||||
|
// TODO: check here that the neighbourEvent is at the correct edge of it's fragment
|
||||||
// get neighbour fragment to link it up later on
|
// get neighbour fragment to link it up later on
|
||||||
const neighbourEvent = await txn.timelineEvents.getByEventId(this._roomId, neighbourEventId);
|
const neighbourEvent = await txn.timelineEvents.getByEventId(this._roomId, duplicateEventId);
|
||||||
const neighbourFragment = await txn.timelineFragments.get(this._roomId, neighbourEvent.fragmentId);
|
const neighbourFragment = await txn.timelineFragments.get(this._roomId, neighbourEvent.fragmentId);
|
||||||
neighbourFragmentEntry = fragmentEntry.createNeighbourEntry(neighbourFragment);
|
neighbourFragmentEntry = fragmentEntry.createNeighbourEntry(neighbourFragment);
|
||||||
|
// trim overlapping events
|
||||||
|
remainingEvents = null;
|
||||||
|
} else {
|
||||||
|
// we've hit https://github.com/matrix-org/synapse/issues/7164,
|
||||||
|
// e.g. the event id we found is already in our store but it is not
|
||||||
|
// the adjacent fragment id. Ignore the event, but keep processing the ones after.
|
||||||
|
remainingEvents = remainingEvents.slice(duplicateEventIndex + 1);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
nonOverlappingEvents.push(...remainingEvents);
|
||||||
|
remainingEvents = null;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return {nonOverlappingEvents, neighbourFragmentEntry};
|
return {nonOverlappingEvents, neighbourFragmentEntry};
|
||||||
}
|
}
|
||||||
|
|
||||||
async _findLastFragmentEventKey(fragmentEntry, txn) {
|
async _findExpectedOverlappingEventId(fragmentEntry, txn) {
|
||||||
|
const eventEntry = await this._findFragmentEdgeEvent(
|
||||||
|
fragmentEntry.linkedFragmentId,
|
||||||
|
// reverse because it's the oppose edge of the linked fragment
|
||||||
|
fragmentEntry.direction.reverse(),
|
||||||
|
txn);
|
||||||
|
if (eventEntry) {
|
||||||
|
return eventEntry.event.event_id;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async _findFragmentEdgeEventKey(fragmentEntry, txn) {
|
||||||
const {fragmentId, direction} = fragmentEntry;
|
const {fragmentId, direction} = fragmentEntry;
|
||||||
|
const event = await this._findFragmentEdgeEvent(fragmentId, direction, txn);
|
||||||
|
if (event) {
|
||||||
|
return new EventKey(event.fragmentId, event.eventIndex);
|
||||||
|
} else {
|
||||||
|
// no events yet in the fragment ... odd, but let's not fail and take the default key
|
||||||
|
return EventKey.defaultFragmentKey(fragmentEntry.fragmentId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async _findFragmentEdgeEvent(fragmentId, direction, txn) {
|
||||||
if (direction.isBackward) {
|
if (direction.isBackward) {
|
||||||
const [firstEvent] = await txn.timelineEvents.firstEvents(this._roomId, fragmentId, 1);
|
const [firstEvent] = await txn.timelineEvents.firstEvents(this._roomId, fragmentId, 1);
|
||||||
return new EventKey(firstEvent.fragmentId, firstEvent.eventIndex);
|
return firstEvent;
|
||||||
} else {
|
} else {
|
||||||
const [lastEvent] = await txn.timelineEvents.lastEvents(this._roomId, fragmentId, 1);
|
const [lastEvent] = await txn.timelineEvents.lastEvents(this._roomId, fragmentId, 1);
|
||||||
return new EventKey(lastEvent.fragmentId, lastEvent.eventIndex);
|
return lastEvent;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -57,8 +102,22 @@ export default class GapWriter {
|
||||||
directionalAppend(entries, fragmentEntry, direction);
|
directionalAppend(entries, fragmentEntry, direction);
|
||||||
// set `end` as token, and if we found an event in the step before, link up the fragments in the fragment entry
|
// set `end` as token, and if we found an event in the step before, link up the fragments in the fragment entry
|
||||||
if (neighbourFragmentEntry) {
|
if (neighbourFragmentEntry) {
|
||||||
|
// the throws here should never happen and are only here to detect client or unhandled server bugs
|
||||||
|
// and a last measure to prevent corrupting fragment links
|
||||||
|
if (!fragmentEntry.hasLinkedFragment) {
|
||||||
fragmentEntry.linkedFragmentId = neighbourFragmentEntry.fragmentId;
|
fragmentEntry.linkedFragmentId = neighbourFragmentEntry.fragmentId;
|
||||||
|
} else if (fragmentEntry.linkedFragmentId !== neighbourFragmentEntry.fragmentId) {
|
||||||
|
throw new Error(`Prevented changing fragment ${fragmentEntry.fragmentId} ` +
|
||||||
|
`${fragmentEntry.direction.asApiString()} link from ${fragmentEntry.linkedFragmentId} ` +
|
||||||
|
`to ${neighbourFragmentEntry.fragmentId} in ${this._roomId}`);
|
||||||
|
}
|
||||||
|
if (!neighbourFragmentEntry.hasLinkedFragment) {
|
||||||
neighbourFragmentEntry.linkedFragmentId = fragmentEntry.fragmentId;
|
neighbourFragmentEntry.linkedFragmentId = fragmentEntry.fragmentId;
|
||||||
|
} else if (neighbourFragmentEntry.linkedFragmentId !== fragmentEntry.fragmentId) {
|
||||||
|
throw new Error(`Prevented changing fragment ${neighbourFragmentEntry.fragmentId} ` +
|
||||||
|
`${neighbourFragmentEntry.direction.asApiString()} link from ${neighbourFragmentEntry.linkedFragmentId} ` +
|
||||||
|
`to ${fragmentEntry.fragmentId} in ${this._roomId}`);
|
||||||
|
}
|
||||||
// if neighbourFragmentEntry was found, it means the events were overlapping,
|
// if neighbourFragmentEntry was found, it means the events were overlapping,
|
||||||
// so no pagination should happen anymore.
|
// so no pagination should happen anymore.
|
||||||
neighbourFragmentEntry.token = null;
|
neighbourFragmentEntry.token = null;
|
||||||
|
@ -100,7 +159,7 @@ export default class GapWriter {
|
||||||
throw new Error("start is not equal to prev_batch or next_batch");
|
throw new Error("start is not equal to prev_batch or next_batch");
|
||||||
}
|
}
|
||||||
// find last event in fragment so we get the eventIndex to begin creating keys at
|
// find last event in fragment so we get the eventIndex to begin creating keys at
|
||||||
let lastKey = await this._findLastFragmentEventKey(fragmentEntry, txn);
|
let lastKey = await this._findFragmentEdgeEventKey(fragmentEntry, txn);
|
||||||
// find out if any event in chunk is already present using findFirstOrLastOccurringEventId
|
// find out if any event in chunk is already present using findFirstOrLastOccurringEventId
|
||||||
const {
|
const {
|
||||||
nonOverlappingEvents,
|
nonOverlappingEvents,
|
||||||
|
|
Reference in a new issue