mirror of
https://github.com/vector-im/hydrogen-web.git
synced 2024-12-23 11:35:04 +01:00
Merge pull request #377 from vector-im/bwindels/fix-sync-opentimeline-race
Fix race between sync and subscribing to timeline after opening it
This commit is contained in:
commit
fef0832074
@ -404,10 +404,16 @@ export class BaseRoom extends EventEmitter {
|
|||||||
clock: this._platform.clock,
|
clock: this._platform.clock,
|
||||||
logger: this._platform.logger,
|
logger: this._platform.logger,
|
||||||
});
|
});
|
||||||
|
try {
|
||||||
if (this._roomEncryption) {
|
if (this._roomEncryption) {
|
||||||
this._timeline.enableEncryption(this._decryptEntries.bind(this, DecryptionSource.Timeline));
|
this._timeline.enableEncryption(this._decryptEntries.bind(this, DecryptionSource.Timeline));
|
||||||
}
|
}
|
||||||
await this._timeline.load(this._user, this.membership, log);
|
await this._timeline.load(this._user, this.membership, log);
|
||||||
|
} catch (err) {
|
||||||
|
// this also clears this._timeline in the closeCallback
|
||||||
|
this._timeline.dispose();
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
return this._timeline;
|
return this._timeline;
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
@ -315,7 +315,7 @@ export class SendQueue {
|
|||||||
import {HomeServer as MockHomeServer} from "../../../mocks/HomeServer.js";
|
import {HomeServer as MockHomeServer} from "../../../mocks/HomeServer.js";
|
||||||
import {createMockStorage} from "../../../mocks/Storage.js";
|
import {createMockStorage} from "../../../mocks/Storage.js";
|
||||||
import {NullLogger} from "../../../logging/NullLogger.js";
|
import {NullLogger} from "../../../logging/NullLogger.js";
|
||||||
import {event, withTextBody, withTxnId} from "../../../mocks/event.js";
|
import {createEvent, withTextBody, withTxnId} from "../../../mocks/event.js";
|
||||||
import {poll} from "../../../mocks/poll.js";
|
import {poll} from "../../../mocks/poll.js";
|
||||||
|
|
||||||
export function tests() {
|
export function tests() {
|
||||||
@ -326,12 +326,12 @@ export function tests() {
|
|||||||
const hs = new MockHomeServer();
|
const hs = new MockHomeServer();
|
||||||
// 1. enqueue and start send event 1
|
// 1. enqueue and start send event 1
|
||||||
const queue = new SendQueue({roomId: "!abc", storage, hsApi: hs.api});
|
const queue = new SendQueue({roomId: "!abc", storage, hsApi: hs.api});
|
||||||
const event1 = withTextBody(event("m.room.message", "$123"), "message 1");
|
const event1 = withTextBody("message 1", createEvent("m.room.message", "$123"));
|
||||||
await logger.run("event1", log => queue.enqueueEvent(event1.type, event1.content, null, log));
|
await logger.run("event1", log => queue.enqueueEvent(event1.type, event1.content, null, log));
|
||||||
assert.equal(queue.pendingEvents.length, 1);
|
assert.equal(queue.pendingEvents.length, 1);
|
||||||
const sendRequest1 = hs.requests.send[0];
|
const sendRequest1 = hs.requests.send[0];
|
||||||
// 2. receive remote echo, before /send has returned
|
// 2. receive remote echo, before /send has returned
|
||||||
const remoteEcho = withTxnId(event1, sendRequest1.arguments[2]);
|
const remoteEcho = withTxnId(sendRequest1.arguments[2], event1);
|
||||||
const txn = await storage.readWriteTxn([storage.storeNames.pendingEvents]);
|
const txn = await storage.readWriteTxn([storage.storeNames.pendingEvents]);
|
||||||
const removal = await logger.run("remote echo", log => queue.removeRemoteEchos([remoteEcho], txn, log));
|
const removal = await logger.run("remote echo", log => queue.removeRemoteEchos([remoteEcho], txn, log));
|
||||||
await txn.complete();
|
await txn.complete();
|
||||||
@ -339,7 +339,7 @@ export function tests() {
|
|||||||
queue.emitRemovals(removal);
|
queue.emitRemovals(removal);
|
||||||
assert.equal(queue.pendingEvents.length, 0);
|
assert.equal(queue.pendingEvents.length, 0);
|
||||||
// 3. now enqueue event 2
|
// 3. now enqueue event 2
|
||||||
const event2 = withTextBody(event("m.room.message", "$456"), "message 2");
|
const event2 = withTextBody("message 2", createEvent("m.room.message", "$456"));
|
||||||
await logger.run("event2", log => queue.enqueueEvent(event2.type, event2.content, null, log));
|
await logger.run("event2", log => queue.enqueueEvent(event2.type, event2.content, null, log));
|
||||||
// even though the first pending event has been removed by the remote echo,
|
// even though the first pending event has been removed by the remote echo,
|
||||||
// the second should get the next index, as the send loop is still blocking on the first one
|
// the second should get the next index, as the send loop is still blocking on the first one
|
||||||
|
@ -76,6 +76,7 @@ export class Timeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async _loadPowerLevels(txn) {
|
async _loadPowerLevels(txn) {
|
||||||
|
// TODO: update power levels as state is updated
|
||||||
const powerLevelsState = await txn.roomState.get(this._roomId, "m.room.power_levels", "");
|
const powerLevelsState = await txn.roomState.get(this._roomId, "m.room.power_levels", "");
|
||||||
if (powerLevelsState) {
|
if (powerLevelsState) {
|
||||||
return new PowerLevels({
|
return new PowerLevels({
|
||||||
@ -84,10 +85,14 @@ export class Timeline {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
const createState = await txn.roomState.get(this._roomId, "m.room.create", "");
|
const createState = await txn.roomState.get(this._roomId, "m.room.create", "");
|
||||||
|
if (createState) {
|
||||||
return new PowerLevels({
|
return new PowerLevels({
|
||||||
createEvent: createState.event,
|
createEvent: createState.event,
|
||||||
ownUserId: this._ownMember.userId
|
ownUserId: this._ownMember.userId
|
||||||
});
|
});
|
||||||
|
} else {
|
||||||
|
return new PowerLevels({ownUserId: this._ownMember.userId});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
_setupEntries(timelineEntries) {
|
_setupEntries(timelineEntries) {
|
||||||
@ -116,12 +121,17 @@ export class Timeline {
|
|||||||
return params ? params : false;
|
return params ? params : false;
|
||||||
};
|
};
|
||||||
// first, look in local entries based on txn id
|
// first, look in local entries based on txn id
|
||||||
const foundInLocalEntries = this._localEntries.findAndUpdate(
|
if (pe.relatedTxnId) {
|
||||||
|
const found = this._localEntries.findAndUpdate(
|
||||||
e => e.id === pe.relatedTxnId,
|
e => e.id === pe.relatedTxnId,
|
||||||
updateOrFalse,
|
updateOrFalse,
|
||||||
);
|
);
|
||||||
|
if (found) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
// now look in remote entries based on event id
|
// now look in remote entries based on event id
|
||||||
if (!foundInLocalEntries && pe.relatedEventId) {
|
if (pe.relatedEventId) {
|
||||||
this._remoteEntries.findAndUpdate(
|
this._remoteEntries.findAndUpdate(
|
||||||
e => e.id === pe.relatedEventId,
|
e => e.id === pe.relatedEventId,
|
||||||
updateOrFalse
|
updateOrFalse
|
||||||
@ -141,6 +151,17 @@ export class Timeline {
|
|||||||
}
|
}
|
||||||
|
|
||||||
_addLocalRelationsToNewRemoteEntries(entries) {
|
_addLocalRelationsToNewRemoteEntries(entries) {
|
||||||
|
// because it is not safe to iterate a derived observable collection
|
||||||
|
// before it has any subscriptions, we bail out if this isn't
|
||||||
|
// the case yet. This can happen when sync adds or replaces entries
|
||||||
|
// before load has finished and the view has subscribed to the timeline.
|
||||||
|
//
|
||||||
|
// Once the subscription is setup, MappedList will set up the local
|
||||||
|
// relations as needed with _applyAndEmitLocalRelationChange,
|
||||||
|
// so we're not missing anything by bailing out.
|
||||||
|
if (!this._localEntries.hasSubscriptions) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
// find any local relations to this new remote event
|
// find any local relations to this new remote event
|
||||||
for (const pee of this._localEntries) {
|
for (const pee of this._localEntries) {
|
||||||
// this will work because we set relatedEventId when removing remote echos
|
// this will work because we set relatedEventId when removing remote echos
|
||||||
@ -232,3 +253,52 @@ export class Timeline {
|
|||||||
return this._ownMember;
|
return this._ownMember;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
import {FragmentIdComparer} from "./FragmentIdComparer.js";
|
||||||
|
import {Clock as MockClock} from "../../../mocks/Clock.js";
|
||||||
|
import {createMockStorage} from "../../../mocks/Storage.js";
|
||||||
|
import {createEvent, withTextBody, withSender} from "../../../mocks/event.js";
|
||||||
|
import {NullLogItem} from "../../../logging/NullLogger.js";
|
||||||
|
import {EventEntry} from "./entries/EventEntry.js";
|
||||||
|
import {User} from "../../User.js";
|
||||||
|
import {PendingEvent} from "../sending/PendingEvent.js";
|
||||||
|
|
||||||
|
export function tests() {
|
||||||
|
const fragmentIdComparer = new FragmentIdComparer([]);
|
||||||
|
const roomId = "$abc";
|
||||||
|
return {
|
||||||
|
"adding or replacing entries before subscribing to entries does not loose local relations": async assert => {
|
||||||
|
const pendingEvents = new ObservableArray();
|
||||||
|
const timeline = new Timeline({
|
||||||
|
roomId,
|
||||||
|
storage: await createMockStorage(),
|
||||||
|
closeCallback: () => {},
|
||||||
|
fragmentIdComparer,
|
||||||
|
pendingEvents,
|
||||||
|
clock: new MockClock(),
|
||||||
|
});
|
||||||
|
// 1. load timeline
|
||||||
|
await timeline.load(new User("@alice:hs.tld"), "join", new NullLogItem());
|
||||||
|
// 2. test replaceEntries and addOrReplaceEntries don't fail
|
||||||
|
const event1 = withTextBody("hi!", withSender("@bob:hs.tld", createEvent("m.room.message", "!abc")));
|
||||||
|
const entry1 = new EventEntry({event: event1, fragmentId: 1, eventIndex: 1}, fragmentIdComparer);
|
||||||
|
timeline.replaceEntries([entry1]);
|
||||||
|
const event2 = withTextBody("hi bob!", withSender("@alice:hs.tld", createEvent("m.room.message", "!def")));
|
||||||
|
const entry2 = new EventEntry({event: event2, fragmentId: 1, eventIndex: 2}, fragmentIdComparer);
|
||||||
|
timeline.addOrReplaceEntries([entry2]);
|
||||||
|
// 3. add local relation (redaction)
|
||||||
|
pendingEvents.append(new PendingEvent({data: {
|
||||||
|
roomId,
|
||||||
|
queueIndex: 1,
|
||||||
|
eventType: "m.room.redaction",
|
||||||
|
txnId: "t123",
|
||||||
|
content: {},
|
||||||
|
relatedEventId: event2.event_id
|
||||||
|
}}));
|
||||||
|
// 4. subscribe (it's now safe to iterate timeline.entries)
|
||||||
|
timeline.entries.subscribe({});
|
||||||
|
// 5. check the local relation got correctly aggregated
|
||||||
|
assert.equal(Array.from(timeline.entries)[0].isRedacting, true);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
@ -14,18 +14,22 @@ See the License for the specific language governing permissions and
|
|||||||
limitations under the License.
|
limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
export function event(type, id = null) {
|
export function createEvent(type, id = null) {
|
||||||
return {type, event_id: id};
|
return {type, event_id: id};
|
||||||
}
|
}
|
||||||
|
|
||||||
export function withContent(event, content) {
|
export function withContent(content, event) {
|
||||||
return Object.assign({}, event, {content});
|
return Object.assign({}, event, {content});
|
||||||
}
|
}
|
||||||
|
|
||||||
export function withTextBody(event, body) {
|
export function withSender(sender, event) {
|
||||||
return withContent(event, {body, msgtype: "m.text"});
|
return Object.assign({}, event, {sender});
|
||||||
}
|
}
|
||||||
|
|
||||||
export function withTxnId(event, txnId) {
|
export function withTextBody(body, event) {
|
||||||
|
return withContent({body, msgtype: "m.text"}, event);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function withTxnId(txnId, event) {
|
||||||
return Object.assign({}, event, {unsigned: {transaction_id: txnId}});
|
return Object.assign({}, event, {unsigned: {transaction_id: txnId}});
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user