Skip to content

Commit

Permalink
Retry to-device messages
Browse files Browse the repository at this point in the history 
This adds a queueToDevice API alongside sendToDevice which is a
much higher-level API that adds the messages to a queue, stored in
persistent storage, and retries them periodically. Also converts
sending of megolm keys to use the new API.

Other uses of sendToDevice are nopt converted in this PR, but could
be later.

Requires matrix-org/matrix-mock-request#17
  • Loading branch information
@dbkr
dbkr committed Jul 28, 2022
1 parent 5367ee1 commit d8d87e5
Show file tree
Hide file tree
Showing 14 changed files with 735 additions and 61 deletions.
339 changes: 339 additions & 0 deletions spec/unit/queueToDevice.spec.ts
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,339 @@
/*
Copyright 2022 The Matrix.org Foundation C.I.C.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

import MockHttpBackend from 'matrix-mock-request';
import { indexedDB as fakeIndexedDB } from 'fake-indexeddb';

import { IHttpOpts, IndexedDBStore, MatrixEvent, MemoryStore, Room } from "../../src";
import { MatrixClient } from "../../src/client";
import { ToDeviceBatch } from '../../src/models/ToDeviceMessage';
import { logger } from '../../src/logger';

const FAKE_USER = "@alice:example.org";
const FAKE_DEVICE_ID = "AAAAAAAA";
const FAKE_PAYLOAD = {
"foo": 42,
};
const EXPECTED_BODY = {
messages: {
[FAKE_USER]: {
[FAKE_DEVICE_ID]: FAKE_PAYLOAD,
},
},
};

const FAKE_MSG = {
userId: FAKE_USER,
deviceId: FAKE_DEVICE_ID,
payload: FAKE_PAYLOAD,
};

enum StoreType {
Memory = 'Memory',
IndexedDB = 'IndexedDB',
}

// Jest now uses @sinonjs/fake-timers which exposes tickAsync() and a number of
// other async methods which break the event loop, letting scheduled promise
// callbacks run. Unfortunately, Jest doesn't expose these, so we have to do
// it manually (this is what sinon does under the hood). We do both in a loop
// until the thing we expect happens: hopefully this is the least flakey way
// and avoids assuming anything about the app's behaviour.
const realSetTimeout = setTimeout;
function flushPromises() {
return new Promise(r => {
realSetTimeout(r, 1);
});
}

async function flushAndRunTimersUntil(cond: () => boolean) {
while (!cond()) {
await flushPromises();
if (cond()) break;
jest.advanceTimersToNextTimer();
}
}

describe.each([
[StoreType.Memory], [StoreType.IndexedDB],
])("queueToDevice (%s store)", function(storeType) {
let httpBackend: MockHttpBackend;
let client: MatrixClient;

beforeEach(function() {
httpBackend = new MockHttpBackend();

const store = storeType === StoreType.IndexedDB ?
new IndexedDBStore({ indexedDB: fakeIndexedDB }) : new MemoryStore();

client = new MatrixClient({
baseUrl: "https://my.home.server",
accessToken: "my.access.token",
request: httpBackend.requestFn as IHttpOpts["request"],
store,
});
});

afterEach(function() {
jest.useRealTimers();
});

it("sends a to-device message", async function() {
httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).check((request) => {
expect(request.data).toEqual(EXPECTED_BODY);
}).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});

await httpBackend.flushAllExpected();
});

it("retries on error", async function() {
jest.useFakeTimers();

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(500);

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).check((request) => {
expect(request.data).toEqual(EXPECTED_BODY);
}).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});
expect(httpBackend.flushSync(null, 1)).toEqual(1);

await flushAndRunTimersUntil(() => httpBackend.requests.length > 0);

expect(httpBackend.flushSync(null, 1)).toEqual(1);
});

it("stops retrying on 4xx errors", async function() {
jest.useFakeTimers();

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(400);

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});
expect(httpBackend.flushSync(null, 1)).toEqual(1);

// Asserting that another request is never made is obviously
// a bit tricky - we just flush the queue what should hopefully
// be plenty of times and assert that nothing comes through.
let tries = 0;
await flushAndRunTimersUntil(() => ++tries === 10);

expect(httpBackend.requests.length).toEqual(0);
});

it("honours ratelimiting", async function() {
jest.useFakeTimers();

// pick something obscure enough it's unlikley to clash with a
// retry delay the algorithm uses anyway
const retryDelay = 279 * 1000;

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(429, {
errcode: "M_LIMIT_EXCEEDED",
retry_after_ms: retryDelay,
});

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});
expect(httpBackend.flushSync(null, 1)).toEqual(1);
await flushPromises();

logger.info("Advancing clock to just before expected retry time...");

jest.advanceTimersByTime(retryDelay - 1000);
await flushPromises();

expect(httpBackend.requests.length).toEqual(0);

logger.info("Advancing clock past expected retry time...");

jest.advanceTimersByTime(2000);
await flushPromises();

expect(httpBackend.flushSync(null, 1)).toEqual(1);
});

it("retries on retryImmediately()", async function() {
httpBackend.when("GET", "/_matrix/client/versions").respond(200, {
versions: ["r0.0.1"],
});

await Promise.all([client.startClient(), httpBackend.flush(null, 1, 20)]);

try {
httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(500);

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});
expect(await httpBackend.flush(null, 1, 1)).toEqual(1);
await flushPromises();

client.retryImmediately();

expect(await httpBackend.flush(null, 1, 20)).toEqual(1);
} finally {
client.stopClient();
}
});

it("retries on when client is started", async function() {
httpBackend.when("GET", "/_matrix/client/versions").respond(200, {
versions: ["r0.0.1"],
});

await Promise.all([client.startClient(), httpBackend.flush("/_matrix/client/versions", 1, 20)]);

try {
httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(500);

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});
expect(await httpBackend.flush(null, 1, 1)).toEqual(1);
await flushPromises();

client.stopClient();
await Promise.all([client.startClient(), httpBackend.flush("/_matrix/client/versions", 1, 20)]);

expect(await httpBackend.flush(null, 1, 20)).toEqual(1);
} finally {
client.stopClient();
}
});

it("retries when a message is retried", async function() {
httpBackend.when("GET", "/_matrix/client/versions").respond(200, {
versions: ["r0.0.1"],
});

await Promise.all([client.startClient(), httpBackend.flush(null, 1, 20)]);

try {
httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(500);

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).respond(200, {});

await client.queueToDevice({
eventType: "org.example.foo",
batch: [
FAKE_MSG,
],
});

expect(await httpBackend.flush(null, 1, 1)).toEqual(1);
await flushPromises();

const dummyEvent = new MatrixEvent({
event_id: "!fake:example.org",
});
const mockRoom = {
updatePendingEvent: jest.fn(),
} as unknown as Room;
client.resendEvent(dummyEvent, mockRoom);

expect(await httpBackend.flush(null, 1, 20)).toEqual(1);
} finally {
client.stopClient();
}
});

it("splits many messages into multiple HTTP requests", async function() {
const batch: ToDeviceBatch = {
eventType: "org.example.foo",
batch: [],
};

for (let i = 0; i <= 20; ++i) {
batch.batch.push({
userId: `@user${i}:example.org`,
deviceId: FAKE_DEVICE_ID,
payload: FAKE_PAYLOAD,
});
}

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).check((request) => {
expect(Object.keys(request.data.messages).length).toEqual(20);
}).respond(200, {});

httpBackend.when(
"PUT", "/sendToDevice/org.example.foo/",
).check((request) => {
expect(Object.keys(request.data.messages).length).toEqual(1);
}).respond(200, {});

await client.queueToDevice(batch);
await httpBackend.flushAllExpected();
});
});
Loading

0 comments on commit d8d87e5

Please sign in to comment.