frontend-patterns/research/experiments/matrix-recovery/experiment.mjs

import assert from 'node:assert/strict';
import {createHash} from 'node:crypto';
import {readFileSync} from 'node:fs';
import {pathToFileURL} from 'node:url';

// An installed SDK directory can be supplied to keep node_modules outside the repo.
const sdkDir = process.env.FEP_MATRIX_SDK_DIR;
const sdk = await import(sdkDir ? pathToFileURL(`${sdkDir}/lib/index.js`).href : 'matrix-js-sdk');
const {createClient, MatrixScheduler, Room, MatrixEvent} = sdk;
const loggerModule = await import(sdkDir ? pathToFileURL(`${sdkDir}/lib/logger.js`).href : 'matrix-js-sdk/lib/logger.js');
loggerModule.logger.setLevel('silent');
const installedDir = sdkDir ? pathToFileURL(`${sdkDir}/`) : new URL('./node_modules/matrix-js-sdk/', import.meta.url);
const packageUrl = new URL('package.json', installedDir);
assert.equal(JSON.parse(readFileSync(packageUrl)).version, '42.3.0');
const userId = '@research:example.invalid';
const roomId = '!research:example.invalid';
const digest = data => createHash('sha256').update(data).digest('hex');

function fixture({dropFirst = false, holdFirst = false} = {}) {
    const requests = [];
    // Synthetic receiver: idempotency is implemented here, not proved about a homeserver.
    const accepted = new Map();
    let release;
    let arrived;
    const firstRequest = new Promise(resolve => { arrived = resolve; });
    const gate = new Promise(resolve => { release = resolve; });
    const client = createClient({
        baseUrl: 'https://example.invalid', userId, deviceId: 'RESEARCH',
        accessToken: 'synthetic-not-a-credential',
        // Disable scheduler retries to expose the first failed attempt deterministically.
        scheduler: new MatrixScheduler(() => -1),
        fetchFn: async (url, init) => {
            const path = new URL(url).pathname;
            assert.equal(init.method, 'PUT');
            assert.match(path, /\/send\/m.room.message\//);
            const body = JSON.parse(init.body);
            requests.push({path, body});
            let result = accepted.get(path);
            if (result) assert.deepEqual(result.body, body, 'retry payload changed');
            else {
                result = {body, event_id: `$accepted-${accepted.size + 1}`};
                accepted.set(path, result);
            }
            if (requests.length === 1) {
                arrived();
                if (holdFirst) await gate;
                if (dropFirst) throw new TypeError('synthetic response loss after acceptance');
            }
            return new Response(JSON.stringify({event_id: result.event_id}), {
                status: 200, headers: {'Content-Type': 'application/json'},
            });
        },
    });
    const room = new Room(roomId, client, userId, {pendingEventOrdering: 'detached'});
    client.store.storeRoom(room);
    async function echo() {
        const [path, data] = [...accepted][0];
        const txn = decodeURIComponent(path.split('/').at(-1));
        await room.addLiveEvents([new MatrixEvent({
            event_id: data.event_id, room_id: roomId, sender: userId,
            type: 'm.room.message', content: data.body, origin_server_ts: 1,
            unsigned: {transaction_id: txn},
        })], {});
    }
    return {client, room, requests, accepted, release, firstRequest, echo};
}

const results = [];
{
    const f = fixture({dropFirst: true});
    await assert.rejects(f.client.sendTextMessage(roomId, 'preserve this text', 'operation-a'));
    const [pending] = f.room.getPendingEvents();
    assert.equal(pending.status, 'not_sent');
    assert.equal(f.accepted.size, 1, 'receiver accepted despite local not_sent');
    assert.equal(pending.getContent().body, 'preserve this text');
    const txn = pending.getTxnId();
    await f.client.resendEvent(pending, f.room);
    assert.equal(pending.getTxnId(), txn);
    assert.equal(f.requests.length, 2);
    assert.deepEqual(f.requests[0], f.requests[1]);
    assert.equal(f.accepted.size, 1);
    assert.equal(pending.status, 'sent');
    await f.echo();
    assert.equal(f.room.getPendingEvents().length, 0);
    assert.equal(f.room.getLiveTimeline().getEvents().length, 1);
    results.push({case: 'lost-response-same-operation-retry', passed: true,
        requests: 2, synthetic_receiver_commits: 1, status_after_loss: 'not_sent',
        pending_after_echo: 0, timeline_events: 1});
    f.client.stopClient();
}
{
    const f = fixture({dropFirst: true});
    await assert.rejects(f.client.sendTextMessage(roomId, 'same intent', 'operation-b'));
    await assert.rejects(f.client.sendTextMessage(roomId, 'same intent', 'operation-b-new'), /blocked/);
    assert.equal(f.requests.length, 1, 'pending queue should block a fresh send');
    for (const event of [...f.room.getPendingEvents()]) f.client.cancelPendingEvent(event);
    await f.client.sendTextMessage(roomId, 'same intent', 'operation-b-after-cancel');
    assert.equal(f.accepted.size, 2);
    results.push({case: 'negative-control-new-identity', passed: true,
        synthetic_receiver_commits: 2, fresh_send_initially_blocked: true,
        conclusion: 'cancelling local pending records then using a new identity permits a duplicate logical send'});
    f.client.stopClient();
}
for (const dropFirst of [false, true]) {
    const f = fixture({holdFirst: true, dropFirst});
    // Attach rejection handling immediately, before deliberately delaying the response.
    const outcome = f.client.sendTextMessage(roomId, 'echo first', 'operation-c')
        .then(() => 'resolved', () => 'rejected');
    await f.firstRequest;
    const pending = f.room.getPendingEvents()[0];
    await f.echo();
    assert.equal(f.room.getPendingEvents().length, 0);
    assert.equal(pending.status, null);
    f.release();
    const promiseOutcome = await outcome;
    assert.equal(promiseOutcome, dropFirst ? 'rejected' : 'resolved');
    assert.equal(f.room.getPendingEvents().length, 0);
    assert.equal(f.room.getLiveTimeline().getEvents().length, 1);
    assert.equal(pending.status, null, 'late transport result regressed reconciled event');
    results.push({case: dropFirst ? 'echo-before-transport-error' : 'echo-before-success-response',
        passed: true, promise_outcome: promiseOutcome, event_status: pending.status,
        pending_events: 0, timeline_events: 1});
    f.client.stopClient();
}
console.log(JSON.stringify({
    kind: 'real-sdk-with-synthetic-transport', sdk: 'matrix-js-sdk@42.3.0',
    node: process.version, live_server: false, encryption: false,
    scheduler_automatic_retries: false,
    experiment_sha256: digest(readFileSync(new URL(import.meta.url))),
    lock_sha256: digest(readFileSync(new URL('./package-lock.json', import.meta.url))),
    sdk_file_sha256: Object.fromEntries(['package.json', 'lib/client.js', 'lib/models/room.js', 'lib/scheduler.js']
        .map(file => [file, digest(readFileSync(new URL(file, installedDir)))])),
    results,
}, null, 2));