Fix issues found in review of host pairing

- Saves of a datastore run one at a time, and changes made while saving
  are saved by the next save. Overlapping saves from approving and
  revoking hosts could write an older snapshot after a newer one.
- Map subscribers fetch everything again when a new session starts, as
  the controller only sends values that exist. A host revoked while a
  browser was disconnected stayed in its list.
- The host token check accepts the same pre-rename tokens as the
  WebSocket, so upgrading a host with such a token does not make it ask
  to join again.

packages/controller/src/Controller.ts

@@ -1025,6 +1025,30 @@ export default class Controller {
 		return jwt.sign({ aud: "host", host: hostId }, this.authSecret);
 	}
 
+	/**
+	 * Check a host token
+	 *
+	 * @param token - Host token to check.
+	 * @returns ID of the host the token was issued to.
+	 * @throws if the token is invalid, its host is not registered or it was issued before tokenValidAfter.
+	 */
+	verifyHostToken(token: string): number {
+		// migrate: allow pre-rename tokens issued to hosts before alpha-14
+		const payload = jwt.verify(token, this.authSecret, { audience: ["host", "slave"] });
+		if (typeof payload === "string") {
+			throw new Error("unexpected JsonWebToken type");
+		}
+		const hostId = payload.host !== undefined ? payload.host : payload.slave;
+		const host = this.hosts.get(hostId);
+		if (!host) {
+			throw new Error("host is not registered, start the host again to request to join the cluster");
+		}
+		if (!payload.iat || payload.iat < host.tokenValidAfter) {
+			throw new Error("invalid token");
+		}
+		return hostId;
+	}
+
 	hostsUpdated(hosts: HostRecord[]) {
 		let updates = hosts.map(host => host.toHostDetails());
 		this.subscriptions.broadcast(new lib.HostUpdatesEvent(updates));

packages/controller/src/WsServer.ts

@@ -252,24 +252,9 @@ ${err.stack}`
 		}
 
 		try {
-			// migrate: allow pre-rename tokens issued to hosts before alpha-14
-			const tokenPayload = jwt.verify(data.token, this.controller.authSecret, { audience: ["host", "slave"] });
-			if (typeof tokenPayload === "string") {
-				throw new Error("unexpected JsonWebToken type");
-			}
-
-			// migrate: allow pre-rename tokens issued to hosts before alpha-14
-			if ((tokenPayload.host !== undefined ? tokenPayload.host : tokenPayload.slave) !== data.id) {
+			if (this.controller.verifyHostToken(data.token) !== data.id) {
 				throw new Error("missmatched host id");
 			}
-
-			let host = this.controller.hosts.get(data.id);
-			if (!host) {
-				throw new Error("host is not registered, start the host again to request to join the cluster");
-			}
-			if (!tokenPayload.iat || tokenPayload.iat < host.tokenValidAfter) {
-				throw new Error("invalid token");
-			}
 		} catch (err: any) {
 			logger.verbose(`WsServer | authentication failed for ${this.remoteAddr(req)}`);
 			wsRejectedConnectionsCounter.inc();

packages/controller/src/routes.ts

@@ -189,17 +189,7 @@ function validateHostToken(req: Request, res: Response, next: any) {
 	}
 
 	try {
-		const tokenPayload = jwt.verify(token, req.app.locals.controller.authSecret, { audience: "host" });
-		if (typeof tokenPayload === "string") {
-			throw new Error("unexpected JsonWebToken type");
-		}
-		let host = req.app.locals.controller.hosts.get(tokenPayload.host);
-		if (!host) {
-			throw new Error("invalid host");
-		}
-		if (!tokenPayload.iat || tokenPayload.iat < (host.tokenValidAfter ?? 0)) {
-			throw new Error("invalid token");
-		}
+		req.app.locals.controller.verifyHostToken(token);
 	} catch (err) {
 		res.sendStatus(401);
 		return;

packages/lib/src/datastore.ts

@@ -166,6 +166,7 @@ export abstract class Datastore<
 		"update": [ updates: U[] ],
 	}> {
 	private dirty = false;
+	private saving: Promise<void> = Promise.resolve();
 
 	constructor(
 		private provider: DatastoreProvider<K, V> = new MemoryDatastoreProvider(),
@@ -181,10 +182,22 @@ export abstract class Datastore<
 
 	// Save the datastore to the provider
 	async save() {
-		if (this.dirty) {
-			await this.provider.save(this.data);
+		// Saves run one at a time so an older snapshot can't be written after a newer one.
+		const save = this.saving.then(async () => {
+			if (!this.dirty) {
+				return;
+			}
+			// Cleared before writing so changes made while saving are saved by the next save.
 			this.dirty = false;
-		}
+			try {
+				await this.provider.save(this.data);
+			} catch (err) {
+				this.dirty = true;
+				throw err;
+			}
+		});
+		this.saving = save.catch(() => {});
+		await save;
 	}
 
 	// Load the datastore from the provider

packages/lib/src/subscriptions.ts

@@ -725,6 +725,16 @@ export class MapSubscriber<
 	/** Current value of the subscribed resource */
 	values = new Map<K, V>();
 
+	override handleConnectionEvent(event: "connect" | "drop" | "resume" | "close") {
+		if (event === "connect") {
+			// A new session is only sent values that exist, not ones removed while disconnected,
+			// so everything is fetched again.
+			this.values.clear();
+			this.lastUpdatedMs = -1;
+		}
+		super.handleConnectionEvent(event);
+	}
+
 	protected override getLastUpdatedTimeMs(event: E) {
 		let maxTime = this.lastUpdatedMs;
 		for (const update of event.updates) {

test/controller/routes.js

@@ -221,6 +221,7 @@ describe("controller/src/routes", function() {
 			controller.hosts = new SubscribableDatastore();
 			controller.hostPairing = new HostPairingManager(controller);
 			controller.generateHostToken = Controller.prototype.generateHostToken;
+			controller.verifyHostToken = Controller.prototype.verifyHostToken;
 			controller.wsServer.remoteAddr = req => req.socket.remoteAddress;
 			adminToken = controller.users.signUserToken(controller.users.getByName("test"));
 			playerToken = controller.users.signUserToken(controller.users.getByName("player"));
@@ -478,6 +479,11 @@ describe("controller/src/routes", function() {
 				}
 				assert(controller.hosts.has(hostId));
 			});
+			it("should respond with 200 for a host token from before the slave rename", async function() {
+				const { hostId } = await controller.hostPairing.createHost("Host A");
+				const token = jwt.sign({ aud: "slave", slave: hostId }, controller.authSecret);
+				await assertResponse(await check({ "Authorization": `Bearer ${token}` }), 200, { valid: true });
+			});
 			it("should respond with 401 without a valid host token", async function() {
 				await controller.hostPairing.createHost("Host A");
 				assert.equal((await check({})).status, 401);

test/lib/datastore.js

@@ -346,6 +346,41 @@ describe("lib/datastore", function() {
 				await datastore.save();
 				assert.equal(datastoreProvider.methodCalls.length, 2);
 			});
+			it("should write overlapping saves in order with the latest data", async function() {
+				const written = [];
+				let release;
+				datastoreProvider.save = async data => {
+					const keys = [...data.keys()];
+					if (!written.length) {
+						await new Promise(resolve => { release = resolve; });
+					}
+					written.push(keys);
+				};
+				datastore = new MockDatastore(datastoreProvider, new Map([["a", 1]]));
+				datastore.callTouch();
+				const first = datastore.save();
+				await new Promise(resolve => setImmediate(resolve)); // First write in progress
+				datastore.getData().set("b", 2);
+				datastore.callTouch();
+				const second = datastore.save();
+				release();
+				await Promise.all([first, second]);
+				assert.deepEqual(written, [["a"], ["a", "b"]]);
+			});
+			it("should save again after a failed save", async function() {
+				let fail = true;
+				datastoreProvider.save = async () => {
+					if (fail) {
+						fail = false;
+						throw new Error("disk full");
+					}
+					datastoreProvider.methodCalls.push(["save"]);
+				};
+				datastore.callTouch();
+				await assert.rejects(datastore.save(), /disk full/);
+				await datastore.save();
+				assert.deepEqual(datastoreProvider.methodCalls.at(-1), ["save"]);
+			});
 		});
 
 		describe("load", function() {

test/lib/subscriptions.test.js

@@ -1584,6 +1584,34 @@ describe("lib/subscriptions", function() {
 			assert.equal(snap2.has(2), true);
 		});
 
+		it("should fetch everything again when a new session starts", async function() {
+			subscriber.subscribe(() => {});
+			await events.once(mockControl.connector, "send");
+			await subscriber._handleEvent(new MapEvent([
+				{ id: 1, updatedAtMs: 10, isDeleted: false },
+			]));
+			const [before] = subscriber.getSnapshot();
+			mockControl.connector.sentMessages.length = 0;
+
+			mockControl.connector.emit("connect");
+			assert.equal(subscriber.values.size, 0);
+			assert.notStrictEqual(subscriber.getSnapshot()[0], before);
+			await new Promise(resolve => setImmediate(resolve));
+			const requests = mockControl.connector.sentMessages.filter(m => m.name === "SubscriptionRequest");
+			assert.equal(requests.length, 1);
+			assert.equal(requests[0].data.lastRequestTimeMs, -1);
+		});
+
+		it("should keep values when a session is resumed", async function() {
+			await subscriber._handleEvent(new MapEvent([
+				{ id: 1, updatedAtMs: 10, isDeleted: false },
+			]));
+			subscriber.handleConnectionEvent("drop");
+			subscriber.handleConnectionEvent("resume");
+			assert.equal(subscriber.values.size, 1);
+			assert.equal(subscriber.lastUpdatedMs, 10);
+		});
+
 		it("should handle multiple updates correctly", async function() {
 			await subscriber._handleEvent(new MapEvent([
 				{ id: 1, updatedAtMs: 10, isDeleted: false },