Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Threads work stash #3542

Closed
wants to merge 8 commits into from
Closed

Threads work stash #3542

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e781af8
Deduplicate calls to Thread::fetchRootEvent
t3chguy Jul 5, 2023
bbe2686
De-duplicate GET requests at the http-api level too
t3chguy Jul 5, 2023
387c4e2
Add test for fetch deduplication
t3chguy Jul 5, 2023
1889f43
Fix fetch consolidation to not trigger promise errors
t3chguy Jul 5, 2023
0b94480
Iterate tests
t3chguy Jul 5, 2023
4ae55f6
Iterate tests
t3chguy Jul 5, 2023
ca0abf7
Fix more tests
t3chguy Jul 5, 2023
94fe9c5
Stash
t3chguy Jul 5, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 2 additions & 91 deletions spec/integ/matrix-client-event-timeline.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -604,17 +604,6 @@ describe("MatrixClient event timelines", function () {
return THREAD_ROOT;
});

httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});

httpBackend
.when(
"GET",
Expand Down Expand Up @@ -1258,7 +1247,7 @@ describe("MatrixClient event timelines", function () {
encodeURIComponent(THREAD_RELATION_TYPE.name) +
buildRelationPaginationQuery({
dir: Direction.Backward,
limit: 3,
limit: 1,
recurse: true,
}),
)
Expand Down Expand Up @@ -1549,8 +1538,6 @@ describe("MatrixClient event timelines", function () {
expect(timelineSets).not.toBeNull();
respondToThreads(threadsResponse);
respondToThreads(threadsResponse);
respondToEvent(THREAD_ROOT);
respondToEvent(THREAD2_ROOT);
respondToThread(THREAD_ROOT, [THREAD_REPLY]);
respondToThread(THREAD2_ROOT, [THREAD2_REPLY]);
await flushHttp(room.fetchRoomThreads());
Expand All @@ -1570,9 +1557,6 @@ describe("MatrixClient event timelines", function () {
thread.initialEventsFetched = true;
const prom = emitPromise(room, ThreadEvent.NewReply);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD2_ROOT);
await room.addLiveEvents([THREAD_REPLY2]);
await httpBackend.flushAllExpected();
Expand Down Expand Up @@ -1678,8 +1662,6 @@ describe("MatrixClient event timelines", function () {
expect(timelineSets).not.toBeNull();
respondToThreads(threadsResponse);
respondToThreads(threadsResponse);
respondToEvent(THREAD_ROOT);
respondToEvent(THREAD2_ROOT);
respondToThread(THREAD_ROOT, [THREAD_REPLY]);
respondToThread(THREAD2_ROOT, [THREAD2_REPLY]);
await flushHttp(room.fetchRoomThreads());
Expand All @@ -1699,8 +1681,6 @@ describe("MatrixClient event timelines", function () {
thread.initialEventsFetched = true;
const prom = emitPromise(room, ThreadEvent.Update);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD_ROOT_UPDATED);
respondToEvent(THREAD2_ROOT);
await room.addLiveEvents([THREAD_REPLY_REACTION]);
await httpBackend.flushAllExpected();
Expand Down Expand Up @@ -2016,11 +1996,6 @@ describe("MatrixClient event timelines", function () {
},
},
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
Expand All @@ -2047,71 +2022,7 @@ describe("MatrixClient event timelines", function () {
expect(thread.initialEventsFetched).toBeTruthy();
const timelineSet = thread.timelineSet;

httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/event/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return THREAD_ROOT;
});
httpBackend
.when("GET", "/rooms/!foo%3Abar/context/" + encodeURIComponent(THREAD_ROOT.event_id!))
.respond(200, function () {
return {
start: "start_token",
events_before: [],
event: THREAD_ROOT,
events_after: [],
end: "end_token",
state: [],
};
});
httpBackend
.when(
"GET",
"/_matrix/client/v1/rooms/!foo%3Abar/relations/" +
encodeURIComponent(THREAD_ROOT.event_id!) +
"/" +
encodeURIComponent(THREAD_RELATION_TYPE.name) +
buildRelationPaginationQuery({
dir: Direction.Backward,
from: "start_token",
}),
)
.respond(200, function () {
return {
chunk: [],
};
});
httpBackend
.when(
"GET",
"/_matrix/client/v1/rooms/!foo%3Abar/relations/" +
encodeURIComponent(THREAD_ROOT.event_id!) +
"/" +
encodeURIComponent(THREAD_RELATION_TYPE.name) +
buildRelationPaginationQuery({ dir: Direction.Forward, from: "end_token" }),
)
.respond(200, function () {
return {
chunk: [THREAD_REPLY],
};
});

const timeline = await flushHttp(client.getEventTimeline(timelineSet, THREAD_ROOT.event_id!));
const timeline = await client.getEventTimeline(timelineSet, THREAD_ROOT.event_id!);

httpBackend.when("GET", "/sync").respond(200, {
next_batch: "s_5_5",
Expand Down
21 changes: 20 additions & 1 deletion spec/unit/http-api/fetch.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import { FetchHttpApi } from "../../../src/http-api/fetch";
import { TypedEventEmitter } from "../../../src/models/typed-event-emitter";
import { ClientPrefix, HttpApiEvent, HttpApiEventHandlerMap, IdentityPrefix, IHttpOpts, Method } from "../../../src";
import { emitPromise } from "../../test-utils/test-utils";
import { QueryDict } from "../../../src/utils";
import { defer, QueryDict } from "../../../src/utils";

describe("FetchHttpApi", () => {
const baseUrl = "http://baseUrl";
Expand Down Expand Up @@ -290,4 +290,23 @@ describe("FetchHttpApi", () => {
runTests(baseUrlWithTrailingSlash);
});
});

it("should consolidate simultaneous GET requests to the same URL", () => {
const deferred = defer<Response>();
const fetchFn = jest.fn().mockReturnValue(deferred.promise);
const api = new FetchHttpApi(new TypedEventEmitter<any, any>(), {
baseUrl,
prefix,
fetchFn,
accessToken: "token",
});

api.authedRequest(Method.Get, "/path");
api.authedRequest(Method.Get, "/path");
api.authedRequest(Method.Get, "/path");
api.authedRequest(Method.Get, "/path");
deferred.resolve({ ok: true, text: () => "foobar" } as unknown as Response);

expect(fetchFn).toHaveBeenCalledTimes(1);
});
});
7 changes: 2 additions & 5 deletions spec/unit/room.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2780,17 +2780,16 @@ describe("Room", function () {
opts: IRelationsRequestOpts = { dir: Direction.Backward },
) =>
Promise.resolve({
chunk: [threadResponse1.event] as IEvent[],
chunk: [threadResponse1.event, threadResponse2.event] as IEvent[],
next_batch: "start_token",
});

let prom = emitPromise(room, ThreadEvent.New);
await room.addLiveEvents([threadRoot, threadResponse1]);
await room.addLiveEvents([threadRoot, threadResponse1, threadResponse2]);
const thread: Thread = await prom;
await emitPromise(room, ThreadEvent.Update);

expect(thread.initialEventsFetched).toBeTruthy();
await room.addLiveEvents([threadResponse2]);
expect(thread).toHaveLength(2);
expect(thread.replyToEvent!.getId()).toBe(threadResponse2.getId());

Expand All @@ -2809,10 +2808,8 @@ describe("Room", function () {
},
});

prom = emitPromise(room, ThreadEvent.Update);
const threadResponse2Redaction = mkRedaction(threadResponse2);
await room.addLiveEvents([threadResponse2Redaction]);
await prom;
await emitPromise(room, ThreadEvent.Update);
expect(thread).toHaveLength(1);
expect(thread.replyToEvent!.getId()).toBe(threadResponse1.getId());
Expand Down
21 changes: 18 additions & 3 deletions src/http-api/fetch.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,12 @@ limitations under the License.
* This is an internal module. See {@link MatrixHttpApi} for the public class.
*/

import { checkObjectHasKeys, encodeParams } from "../utils";
import { checkObjectHasKeys, encodeParams, QueryDict } from "../utils";
import { TypedEventEmitter } from "../models/typed-event-emitter";
import { Method } from "./method";
import { ConnectionError, MatrixError } from "./errors";
import { HttpApiEvent, HttpApiEventHandlerMap, IHttpOpts, IRequestOpts } from "./interface";
import { anySignal, parseErrorResponse, timeoutSignal } from "./utils";
import { QueryDict } from "../utils";

type Body = Record<string, any> | BodyInit;

Expand All @@ -40,6 +39,9 @@ export type ResponseType<T, O extends IHttpOpts> = O extends undefined

export class FetchHttpApi<O extends IHttpOpts> {
private abortController = new AbortController();
// We track in flight GET requests to de-duplicate them
// as semantically GET requests done at the same time should yield the same result.
private inFlight = new Map<string, Promise<ResponseType<any, O>>>();

public constructor(
private eventEmitter: TypedEventEmitter<HttpApiEvent, HttpApiEventHandlerMap>,
Expand Down Expand Up @@ -170,6 +172,7 @@ export class FetchHttpApi<O extends IHttpOpts> {

/**
* Perform a request to the homeserver without any credentials.
* De-duplicates simultaneous GET requests to the same URL to improve performance.
* @param method - The HTTP method e.g. "GET".
* @param path - The HTTP path <b>after</b> the supplied prefix e.g.
* "/createRoom".
Expand Down Expand Up @@ -202,7 +205,19 @@ export class FetchHttpApi<O extends IHttpOpts> {
opts?: IRequestOpts,
): Promise<ResponseType<T, O>> {
const fullUri = this.getUrl(path, queryParams, opts?.prefix, opts?.baseUrl);
return this.requestOtherUrl<T>(method, fullUri, body, opts);
const cacheKey = fullUri.toString();
if (method === Method.Get && this.inFlight.has(cacheKey)) {
return this.inFlight.get(cacheKey)!;
}

let prom = this.requestOtherUrl<T>(method, fullUri, body, opts);
if (method === Method.Get) {
this.inFlight.set(cacheKey, prom);
prom = prom.finally(() => {
this.inFlight.delete(cacheKey);
});
}
return prom;
}

/**
Expand Down
3 changes: 2 additions & 1 deletion src/models/read-receipt.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,7 +167,8 @@ export abstract class ReadReceipt<
);
}

const preferSynthetic = ordering === null || ordering < 0;
// We only prefer the synthetic if it is the one we just added or if it points to a later event than the real
const preferSynthetic = (synthetic && ordering === null) || (ordering !== null && ordering < 0);

// we don't bother caching just real receipts by event ID as there's nothing that would read it.
// Take the current cached receipt before we overwrite the pair elements.
Expand Down
48 changes: 31 additions & 17 deletions src/models/thread.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
public readonly room: Room;
public readonly client: MatrixClient;
private readonly pendingEventOrdering: PendingEventOrdering;
private processRootEventPromise?: Promise<void>;

public initialEventsFetched = !Thread.hasServerSideSupport;
/**
Expand Down Expand Up @@ -146,13 +147,16 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM

private async fetchRootEvent(): Promise<void> {
this.rootEvent = this.room.findEventById(this.id);
// If the rootEvent does not exist in the local stores, then fetch it from the server.
try {
const eventData = await this.client.fetchRoomEvent(this.roomId, this.id);
const mapper = this.client.getEventMapper();
this.rootEvent = mapper(eventData); // will merge with existing event object if such is known
} catch (e) {
logger.error("Failed to fetch thread root to construct thread with", e);
// If the rootEvent does not exist in the local stores or it doesn't contain the unsigned m.thread data,
// then fetch it from the server.
if (!this.rootEvent || this.getRootEventBundledRelationship() === undefined) {
try {
const eventData = await this.client.fetchRoomEvent(this.roomId, this.id);
const mapper = this.client.getEventMapper();
this.rootEvent = mapper(eventData); // will merge with existing event object if such is known
} catch (e) {
logger.error("Failed to fetch thread root to construct thread with", e);
}
}
await this.processEvent(this.rootEvent);
}
Expand Down Expand Up @@ -195,9 +199,11 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
}
this.lastEvent = this.rootEvent;
this._currentUserParticipated = false;
// Clear the promise so the thread is re-fetched when we next see it
this.processRootEventPromise = undefined;
this.emit(ThreadEvent.Delete, this);
} else {
await this.updateThreadMetadata();
await this.updateThreadMetadata(true);
}
};

Expand Down Expand Up @@ -475,18 +481,24 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
}
}

private async updateThreadMetadata(): Promise<void> {
private async updateThreadMetadata(forceUpdateRoot = false): Promise<void> {
this.updatePendingReplyCount();

if (Thread.hasServerSideSupport) {
// Ensure we show *something* as soon as possible, we'll update it as soon as we get better data, but we
// don't want the thread preview to be empty if we can avoid it
if (!this.initialEventsFetched) {
await this.processRootEvent();
if (!this.processRootEventPromise || forceUpdateRoot) {
if (Thread.hasServerSideSupport) {
// Ensure we show *something* as soon as possible, we'll update it as soon as we get better data, but we
// don't want the thread preview to be empty if we can avoid it
if (!this.initialEventsFetched) {
this.processRootEventPromise = this.processRootEvent();
await this.processRootEventPromise;
}

this.processRootEventPromise = this.fetchRootEvent();
await this.processRootEventPromise;
}
await this.fetchRootEvent();
this.processRootEventPromise = this.processRootEvent();
}
await this.processRootEvent();
await this.processRootEventPromise;

if (!this.initialEventsFetched) {
this.initialEventsFetched = true;
Expand Down Expand Up @@ -566,7 +578,9 @@ export class Thread extends ReadReceipt<ThreadEmittedEvents, ThreadEventHandlerM
* Finds an event by ID in the current thread
*/
public findEventById(eventId: string): MatrixEvent | undefined {
return this.timelineSet.findEventById(eventId);
if (this.rootEvent?.getId() === eventId) return this.rootEvent;
if (this.lastEvent?.getId() === eventId) return this.lastEvent;
return this.timelineSet.findEventById(eventId) ?? this.replayEvents?.find((event) => event.getId() === eventId);
}

/**
Expand Down