feat(wire-format): per-document XChaCha announce and lock (TEC-2936)

.env.example

@@ -11,4 +11,7 @@ SERVER_DID=did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK
 
 # Rate Limiting
 RATE_LIMIT_WINDOW_MS=900000
-RATE_LIMIT_MAX=100
+RATE_LIMIT_MAX=100
+# Collab wire format rollout: unset (ECIES everywhere, never locks) or "xchacha"
+# (announce XChaCha and lock a document once every socket in its room can read it).
+# WIRE_FORMAT_TARGET=xchacha

src/config/index.ts

@@ -54,4 +54,7 @@ export const config: ServerConfig = {
   webhook: {
     apiKey: process.env.COLLAB_WEBHOOK_API_KEY,
   },
+  wireFormat: {
+    target: process.env.WIRE_FORMAT_TARGET === "xchacha" ? "xchacha" : "ecies",
+  },
 };

src/database/models/document-wire-format.ts

@@ -0,0 +1,22 @@
+import mongoose, { Schema, Document as MongooseDocument } from "mongoose";
+
+// Per-document wire-format lock. A row means the document is locked to XChaCha; no row
+// means ECIES. Separate from DocumentMeta on purpose: that row is created by the owner's
+// meta upload with $setOnInsert fields, and an auth-time upsert would strand them.
+interface IDocumentWireFormat extends MongooseDocument {
+  _id: string; // documentId
+  wireFormat: "xchacha";
+  lockedAt: number;
+}
+
+const DocumentWireFormatSchema = new Schema<IDocumentWireFormat>({
+  _id: { type: String, required: true },
+  wireFormat: { type: String, enum: ["xchacha"], required: true },
+  lockedAt: { type: Number, required: true },
+});
+
+export const DocumentWireFormatModel = mongoose.model<IDocumentWireFormat>(
+  "DocumentWireFormat",
+  DocumentWireFormatSchema
+);
+export type { IDocumentWireFormat };

src/database/models/index.ts

@@ -5,3 +5,4 @@ export { CounterModel, type ICounter } from "./counter";
 export { DocumentMetaModel, type IDocumentMeta } from "./document-meta";
 export { DocumentMirrorModel, type IDocumentMirror } from "./document-mirror";
 export { DocumentEditEpochModel, type IDocumentEditEpoch } from "./document-edit-epoch";
+export { DocumentWireFormatModel, type IDocumentWireFormat } from "./document-wire-format";

src/index.ts

@@ -28,6 +28,7 @@ import { createRotateSessionHandler } from "./services/rotate-route";
 import { rotationCoordinator } from "./services/rotation-coordinator";
 import { createFlushHandler } from "./services/flush-route";
 import { createDeletedFileWebhookHandler } from "./services/deleted-file-webhook";
+import { markDraining } from "./services/lifecycle";
 import { createLightNode } from "@waku/sdk";
 import protobuf from "protobufjs";
 import { generateKeyPairFromSeed } from "@libp2p/crypto/keys";
@@ -253,6 +254,7 @@ class CollaborationServer {
 
   private shutdown(signal: string) {
     logger.info(`\n Received ${signal}. Shutting down gracefully...`);
+    markDraining();
 
     if (this.orphanGcInterval) {
       clearInterval(this.orphanGcInterval);

src/services/lifecycle.ts

@@ -0,0 +1,16 @@
+// Set at the top of the graceful shutdown, before io.close() fires "disconnecting" for
+// every socket on the instance. Per-socket work that only matters on a live server (the
+// wire-format ratchet) checks it and returns: the reconnect auth after the restart
+// resolves the same state, and running it here only burns a Mongo read and a
+// cross-instance enumeration per socket against peers that are also going down.
+let draining = false;
+
+export const markDraining = (): void => {
+  draining = true;
+};
+
+export const isDraining = (): boolean => draining;
+
+export const resetDrainingForTests = (): void => {
+  draining = false;
+};

src/services/mongodb-store.ts

@@ -1,6 +1,7 @@
 import { DocumentUpdate, DocumentCommit, AppType } from "../types/index";
-import { DocumentUpdateModel, DocumentCommitModel, CounterModel, SessionModel, DocumentMetaModel, DocumentMirrorModel, DocumentEditEpochModel } from "../database/models";
+import { DocumentUpdateModel, DocumentCommitModel, CounterModel, SessionModel, DocumentMetaModel, DocumentMirrorModel, DocumentEditEpochModel, DocumentWireFormatModel } from "../database/models";
 import { logger, perfLogger } from "./logger";
+import type { WireFormat } from "./wire-format";
 
 export class SessionTerminatedError extends Error {
   constructor() { super("session terminated"); this.name = "SessionTerminatedError"; }
@@ -375,6 +376,31 @@ export class MongoDBStore {
     return { editLock: meta.editLock ?? null, title: meta.title ?? null };
   }
 
+  // Wire-format lock. No row means ECIES.
+  async getWireFormat(documentId: string): Promise<WireFormat> {
+    const row: any = await DocumentWireFormatModel.findById(documentId).select("wireFormat").lean();
+    return row?.wireFormat === "xchacha" ? "xchacha" : "ecies";
+  }
+
+  // Idempotent under a race: two concurrent auths both end locked, exactly one sees true.
+  async lockWireFormat(documentId: string): Promise<boolean> {
+    try {
+      const res = await DocumentWireFormatModel.updateOne(
+        { _id: documentId },
+        { $setOnInsert: { wireFormat: "xchacha", lockedAt: Date.now() } },
+        { upsert: true, writeConcern: { w: "majority", j: true } }
+      );
+      return res.upsertedCount === 1;
+    } catch (err) {
+      // A concurrent upsert on the same _id can surface as a duplicate-key error instead
+      // of a clean loss of the upsertedCount race; the other caller took the lock.
+      if (typeof err === "object" && err !== null && "code" in err && (err as { code: unknown }).code === 11000) {
+        return false;
+      }
+      throw err;
+    }
+  }
+
   async setMinEditEpoch(documentId: string, epoch: number): Promise<void> {
     await DocumentEditEpochModel.findOneAndUpdate(
       { _id: documentId },
@@ -632,6 +658,7 @@ export class MongoDBStore {
       CounterModel.deleteOne({ _id: documentId }),
       DocumentMirrorModel.deleteMany({ documentId }),
       DocumentEditEpochModel.deleteOne({ _id: documentId }),
+      DocumentWireFormatModel.deleteOne({ _id: documentId }),
     ]);
   }
 
@@ -694,6 +721,7 @@ export class MongoDBStore {
       await DocumentCommitModel.deleteMany({ documentId: s.documentId });
       await SessionModel.deleteMany({ documentId: s.documentId });
       await CounterModel.deleteOne({ _id: s.documentId });
+      await DocumentWireFormatModel.deleteOne({ _id: s.documentId });
       purged++;
     }
     return purged;