Keep fast typing from being silently dropped

e2e/collab.e2e.ts

@@ -11,7 +11,7 @@
  * login session cookie. They still meet in one repository-authorized channel.
  */
 
-import { content, expect, ready, test } from "./room";
+import { content, expect, ready, test, written } from "./room";
 
 test("an edit by one appears for the other", async ({ join }) => {
 	let ana = await join("ana");
@@ -30,6 +30,21 @@ test("an edit by one appears for the other", async ({ join }) => {
 	await expect(content(ana)).toContainText("Readable and diffable.");
 });
 
+test("typing faster than the server takes updates still reaches everyone", async ({ join, room }) => {
+	let ana = await join("ana");
+	let bo = await join("bo");
+
+	// Without a delay this is well past the 200 updates a second the server
+	// accepts from one socket. The excess used to be dropped, and everything
+	// typed afterwards was stranded behind it.
+	await content(ana).click();
+	await ana.keyboard.type("Typed faster than anybody types. ".repeat(8).trim());
+	await ana.keyboard.type(" Then the rest.", { delay: 50 });
+
+	await expect(content(bo)).toContainText("anybody types. Then the rest.");
+	await written(ana, room, "anybody types. Then the rest.");
+});
+
 test("the header represents everyone here as faces", async ({ join }) => {
 	let ana = await join("ana");
 	let bo = await join("bo");

packages/editor/src/provider.test.ts

@@ -11,7 +11,7 @@
  * layout, and happy-dom returns zero for every measurement.
  */
 
-import { describe, expect, it } from "bun:test";
+import { afterEach, describe, expect, it, jest } from "bun:test";
 import * as Y from "yjs";
 
 import { PlanProvider } from "./provider";
@@ -302,3 +302,205 @@ describe("opening the plan", () => {
 		});
 	});
 });
+
+/**
+ * A server that applies, acknowledges and relays updates, and drops some
+ * without a word, as the real one does past its rate limit.
+ *
+ * The plan service accepts at most 200 updates a second from one socket and
+ * silently ignores the rest. A dropped update also strands every later one
+ * from the same client, because Yjs holds an update whose predecessor never
+ * arrived instead of integrating it.
+ */
+function server(drop: (index: number) => boolean) {
+	let doc = new Y.Doc();
+	let transport = wire();
+	let received: Array<{ id: string; bytes: number }> = [];
+	let delivered = 0;
+	let send = transport.send;
+	let room = {
+		doc,
+		transport,
+		received,
+		drop,
+		get delivered() {
+			return delivered;
+		},
+	};
+	transport.send = (kind: string, payload: Record<string, unknown> = {}) => {
+		send.call(transport, kind);
+		if (kind !== "plan:update") return;
+		let binary = atob(payload.update as string);
+		let update = Uint8Array.from(binary, char => char.charCodeAt(0));
+		received.push({ id: payload.id as string, bytes: update.byteLength });
+		if (room.drop(received.length - 1)) return;
+		delivered++;
+		Y.applyUpdate(doc, update);
+		transport.emit("plan:ack", { kind: "plan:ack", id: payload.id });
+	};
+	return room;
+}
+
+function typing(doc: Y.Doc, count: number) {
+	let text = doc.getText("typed");
+	for (let i = 0; i < count; i++) text.insert(text.length, "x");
+}
+
+describe("delivering local edits", () => {
+	afterEach(() => {
+		jest.useRealTimers();
+	});
+
+	it("does not send faster than the server accepts", async () => {
+		jest.useFakeTimers();
+		// Past 200 a second the real server drops the rest.
+		let room = server(() => room.delivered >= 200);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 300);
+		jest.advanceTimersByTime(1_000);
+
+		expect(room.doc.getText("typed").toString()).toBe("x".repeat(300));
+		expect(room.delivered).toBeLessThan(200);
+	});
+
+	it("resends an edit the server never acknowledged", async () => {
+		jest.useFakeTimers();
+		let room = server(index => index === 0);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		jest.advanceTimersByTime(100);
+		typing(doc, 1);
+		jest.advanceTimersByTime(100);
+		// The second keystroke arrived but cannot be applied without the first.
+		expect(room.doc.getText("typed").toString()).toBe("");
+
+		jest.advanceTimersByTime(10_000);
+		expect(room.doc.getText("typed").toString()).toBe("xx");
+	});
+
+	it("leaves acknowledged edits alone", async () => {
+		jest.useFakeTimers();
+		let room = server(() => false);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		jest.advanceTimersByTime(10_000);
+
+		expect(room.delivered).toBe(1);
+	});
+
+	it("backs off and then stops resending an edit nobody acknowledges", async () => {
+		jest.useFakeTimers();
+		let room = server(() => true);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		// Sent at once, resent at 4s, then at 8s, 16s and 32s.
+		jest.advanceTimersByTime(7_000);
+		expect(room.received).toHaveLength(2);
+		jest.advanceTimersByTime(7_000);
+		expect(room.received).toHaveLength(3);
+
+		jest.advanceTimersByTime(10 * 60_000);
+		expect(room.received).toHaveLength(5);
+	});
+
+	it("does not repeat an edit the server refused", async () => {
+		jest.useFakeTimers();
+		let room = server(() => true);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		room.transport.emit("session:error", {
+			kind: "session:error",
+			message: "implementation is active",
+		});
+		typing(doc, 1);
+		// Too large, in the real server: refused without rotating the epoch.
+		jest.advanceTimersByTime(100);
+		room.transport.emit("plan:reset", {
+			kind: "plan:reset",
+			epoch: "01K0N4TR8K7JGM4R1J7PW4R8YJ",
+			reason: "rebuilt",
+		});
+		jest.advanceTimersByTime(10 * 60_000);
+
+		expect(room.received).toHaveLength(2);
+	});
+
+	it("stops resending once the epoch is replaced", async () => {
+		jest.useFakeTimers();
+		let room = server(() => true);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		room.transport.emit("plan:reset", {
+			kind: "plan:reset",
+			epoch: "01K0N4TR8K7JGM4R1J7PW4R8YK",
+			reason: "rebuilt",
+		});
+		jest.advanceTimersByTime(10 * 60_000);
+
+		expect(room.received).toHaveLength(1);
+	});
+
+	it("keeps a merged resend outstanding when the old ids are acknowledged late", async () => {
+		jest.useFakeTimers();
+		let room = server(() => true);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+
+		typing(doc, 1);
+		jest.advanceTimersByTime(100);
+		typing(doc, 1);
+		jest.advanceTimersByTime(4_000);
+		let [first, second, merged] = room.received.map(entry => entry.id);
+		expect(merged).toStartWith("merged-");
+
+		for (let id of [first, second]) {
+			room.transport.emit("plan:ack", { kind: "plan:ack", id });
+		}
+		room.drop = () => false;
+		jest.advanceTimersByTime(60_000);
+
+		expect(room.received.at(-1)?.id).toBe(merged);
+		expect(room.doc.getText("typed").toString()).toBe("xx");
+	});
+
+	it("replays a large backlog in chunks the server accepts", async () => {
+		jest.useFakeTimers();
+		let room = server(() => true);
+		let doc = new Y.Doc();
+		let provider = new PlanProvider({ wire: room.transport, doc });
+		await provider.connect();
+		let text = doc.getText("typed");
+		for (let i = 0; i < 64; i++) {
+			text.insert(text.length, "y".repeat(4_096));
+			jest.advanceTimersByTime(20);
+		}
+
+		room.drop = () => false;
+		let before = room.received.length;
+		await provider.resume();
+		let replayed = room.received.slice(before);
+
+		expect(replayed.length).toBeGreaterThan(1);
+		for (let entry of replayed) expect(entry.bytes).toBeLessThanOrEqual(64 * 1024);
+		expect(room.doc.getText("typed").length).toBe(64 * 4_096);
+	});
+});