Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/database/models/document-meta.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ interface IDocumentMeta extends MongooseDocument {
ownerDid: string | null;
ownerIdentityDid: string | null;
portalAddress: string | null;
appType?: "ddoc" | "dsheet";
editLock: string | null; // roomKey-wrapped
title: string | null; // roomKey-encrypted
updatedAt: number;
Expand All @@ -21,6 +22,8 @@ const DocumentMetaSchema = new Schema<IDocumentMeta>({
ownerDid: { type: String, default: null },
ownerIdentityDid: { type: String, default: null },
portalAddress: { type: String, default: null },
// Which Fileverse app owns this document. Absent ⇒ "ddoc" (legacy).
appType: { type: String, enum: ["ddoc", "dsheet"], default: "ddoc" },
editLock: { type: String, default: null },
title: { type: String, default: null },
updatedAt: { type: Number, required: true },
Expand Down
6 changes: 3 additions & 3 deletions src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ class CollaborationServer {
"/documents/:documentId/rotate-session",
createRotateSessionHandler({
authService, sessionManager, mongodbStore, rotationCoordinator,
terminateOldSession: async (documentId, sessionDid, appType) => {
terminateOldSession: async (documentId, sessionDid) => {
const room = getRoomName(documentId, sessionDid);
const sockets = await this.io!.in(room).fetchSockets();
// Laggard sockets stay AUTHED: their next write must reach createUpdate and get the
Expand All @@ -128,7 +128,7 @@ class CollaborationServer {
// See docs/architecture/gp-semaphore.md.
for (const s of sockets) { s.leave(room); }
await sessionManager.deactivateSession(documentId, sessionDid);
await sessionManager.terminateSession(documentId, sessionDid, appType);
await sessionManager.terminateSession(documentId, sessionDid);
},
}, this.io)
);
Expand All @@ -150,7 +150,7 @@ class CollaborationServer {
const room = getRoomName(documentId, s.sessionDid);
this.io!.to(room).emit("/session/terminated", { roomId: documentId });
for (const sock of await this.io!.in(room).fetchSockets()) sock.leave(room);
await sessionManager.terminateSession(documentId, s.sessionDid, s.appType ?? "ddoc");
await sessionManager.terminateSession(documentId, s.sessionDid);
}
},
})
Expand Down
5 changes: 4 additions & 1 deletion src/scripts/backfill-apptype.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import {
DocumentUpdateModel,
DocumentCommitModel,
SessionModel,
DocumentMetaModel,
} from "../database/models";

async function backfill(): Promise<void> {
Expand All @@ -24,16 +25,18 @@ async function backfill(): Promise<void> {
const filter = { appType: { $exists: false } };
const set = { $set: { appType: "ddoc" as const } };

const [updates, commits, sessions] = await Promise.all([
const [updates, commits, sessions, metas] = await Promise.all([
DocumentUpdateModel.updateMany(filter, set),
DocumentCommitModel.updateMany(filter, set),
SessionModel.updateMany(filter, set),
DocumentMetaModel.updateMany(filter, set),
]);

console.log("appType backfill complete:");
console.log(` DocumentUpdate: ${updates.modifiedCount} tagged "ddoc"`);
console.log(` DocumentCommit: ${commits.modifiedCount} tagged "ddoc"`);
console.log(` Session: ${sessions.modifiedCount} tagged "ddoc"`);
console.log(` DocumentMeta: ${metas.modifiedCount} tagged "ddoc"`);

await databaseService.disconnect();
}
Expand Down
1 change: 0 additions & 1 deletion src/services/flush-route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,6 @@ export function createFlushHandler(deps: FlushDeps) {
commitCid: null,
createdAt: Date.now(),
sessionDid,
appType: "ddoc",
});
} catch (err) {
if (err instanceof SessionTerminatedError) {
Expand Down
29 changes: 22 additions & 7 deletions src/services/mongodb-store.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { DocumentUpdate, DocumentCommit } from "../types/index";
import { DocumentUpdate, DocumentCommit, AppType } from "../types/index";
import { DocumentUpdateModel, DocumentCommitModel, CounterModel, SessionModel, DocumentMetaModel, DocumentMirrorModel, DocumentEditEpochModel } from "../database/models";

export class SessionTerminatedError extends Error {
Expand Down Expand Up @@ -37,7 +37,9 @@ export class MongoDBStore {
commitCid: update.commitCid,
createdAt: update.createdAt,
sessionDid: update.sessionDid,
appType: update.appType,
// Server-authoritative: the session row decides which app owns this stream —
// an HTTP caller (e.g. /flush) has no trustworthy claim of its own.
appType: session.appType ?? "ddoc",
});

await mongoUpdate.save();
Expand Down Expand Up @@ -257,6 +259,7 @@ export class MongoDBStore {
ownerDid: string | null;
ownerIdentityDid: string | null;
portalAddress: string | null;
appType: AppType;
editLock: string | null;
title: string | null;
}): Promise<void> {
Expand All @@ -267,6 +270,7 @@ export class MongoDBStore {
ownerDid: meta.ownerDid,
ownerIdentityDid: meta.ownerIdentityDid,
portalAddress: meta.portalAddress,
appType: meta.appType,
},
$set: {
sessionDid: meta.sessionDid,
Expand All @@ -292,6 +296,7 @@ export class MongoDBStore {
ownerDid: string | null;
ownerIdentityDid: string | null;
sessionDid: string;
appType: AppType;
}): Promise<{ portalAddress: string | null }> {
// 1. Ensure the row exists; pin all binding fields on INSERT only.
await DocumentMetaModel.findOneAndUpdate(
Expand All @@ -304,6 +309,7 @@ export class MongoDBStore {
sessionDid: p.sessionDid,
updatedAt: Date.now(),
isPublished: false,
appType: p.appType,
},
},
{ upsert: true, writeConcern: { w: "majority", j: true } }
Expand Down Expand Up @@ -363,7 +369,7 @@ export class MongoDBStore {
// Discovery: docs bound to the proven owner (identity DID or portal owner DID) — recovery for a wiped device.
async listDocumentsForOwner(
by: { ownerIdentityDid?: string; ownerDid?: string }
): Promise<Array<{ documentId: string; editLock: string | null; title: string | null }>> {
): Promise<Array<{ documentId: string; editLock: string | null; title: string | null; appType: AppType }>> {
const filter: Record<string, any> = {};
if (by.ownerIdentityDid) filter.ownerIdentityDid = by.ownerIdentityDid;
else if (by.ownerDid) filter.ownerDid = by.ownerDid;
Expand All @@ -372,11 +378,12 @@ export class MongoDBStore {
// unpublished durable docs (the publish reconciler flips this flag).
filter.isPublished = { $ne: true };

const metas: any[] = await DocumentMetaModel.find(filter).select("editLock title").lean();
const metas: any[] = await DocumentMetaModel.find(filter).select("editLock title appType").lean();
return metas.map((m) => ({
documentId: m._id,
editLock: m.editLock ?? null,
title: m.title ?? null,
appType: (m.appType as AppType) ?? "ddoc",
}));
}

Expand Down Expand Up @@ -599,26 +606,34 @@ export class MongoDBStore {
return ids;
}

// Conservative orphan GC: terminated ddoc streams with no editLock and no snapshot row.
// A real draft always has an editLock or a snapshot, so it is never swept.
// Conservative orphan GC: terminated update streams (both apps) with no live sibling
// session, no editLock, and no snapshot row. The purge is documentId-scoped, so a live
// sibling must be ruled out first — a real draft always has an editLock or a snapshot,
// so it is never swept; a terminated legacy dsheet stream's durability is its IPFS commit.
async collectOrphans(graceMs: number): Promise<number> {
const cutoff = new Date(Date.now() - graceMs);
const terminated: any[] = await SessionModel.find({
appType: "ddoc", state: "terminated", createdAt: { $lt: cutoff },
state: "terminated", createdAt: { $lt: cutoff },
}).lean();

let purged = 0;
for (const s of terminated) {
// Pre-durable sessions (insert-only portalAddress key absent) predate the
// editLock/snapshot invariant this sweep relies on — never sweep them.
if (!Object.prototype.hasOwnProperty.call(s, "portalAddress")) continue;
// A live sibling session owns this document's stream — never purge under it.
const live = await SessionModel.findOne({ documentId: s.documentId, state: { $ne: "terminated" } })
.select("_id")
.lean();
if (live) continue;
const meta: any = await DocumentMetaModel.findById(s.documentId).lean();
if (meta?.editLock) continue; // real draft — never sweep
const snap: any = await DocumentUpdateModel.findOne({ documentId: s.documentId, updateType: "snapshot" }).sort({ seq: -1 }).lean();
if (snap) continue; // has a hydration base — never sweep
const legacy: any = await DocumentUpdateModel.findOne({ documentId: s.documentId, seq: { $exists: false } }).select("_id").lean();
if (legacy) continue; // pre-durable rows awaiting seq backfill — never sweep
await DocumentUpdateModel.deleteMany({ documentId: s.documentId });
await DocumentCommitModel.deleteMany({ documentId: s.documentId });
await SessionModel.deleteMany({ documentId: s.documentId });
await CounterModel.deleteOne({ _id: s.documentId });
purged++;
Expand Down
4 changes: 2 additions & 2 deletions src/services/rotate-route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ export interface RotateDeps {
sessionManager: Pick<SessionManager, "getSession" | "createSession">;
mongodbStore: Pick<MongoDBStore, "setMinEditEpoch">;
rotationCoordinator: Pick<RotationCoordinator, "begin" | "isActive">;
terminateOldSession: (documentId: string, sessionDid: string, appType: "ddoc" | "dsheet") => Promise<void>;
terminateOldSession: (documentId: string, sessionDid: string) => Promise<void>;
}

export function createRotateSessionHandler(deps: RotateDeps, io: AppServer) {
Expand Down Expand Up @@ -95,7 +95,7 @@ export function createRotateSessionHandler(deps: RotateDeps, io: AppServer) {
onCutover: () => {
io.to(oldRoom).emit("/session/cutover", { roomId: documentId, epoch: gateEpoch });
setTimeout(() => {
void deps.terminateOldSession(documentId, oldSessionDid, oldSession.appType ?? "ddoc")
void deps.terminateOldSession(documentId, oldSessionDid)
.catch((e) => console.error("rotate terminate error:", e));
}, T_DRAIN_MS);
},
Expand Down
13 changes: 2 additions & 11 deletions src/services/session-manager.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { SessionModel, DocumentUpdateModel, DocumentCommitModel } from "../database/models";
import { SessionModel } from "../database/models";

interface RuntimeSession {
documentId: string;
Expand Down Expand Up @@ -302,11 +302,7 @@ export class SessionManager {
}
}

async terminateSession(
documentId: string,
sessionDid: string,
appType: "ddoc" | "dsheet" = "ddoc"
): Promise<void> {
async terminateSession(documentId: string, sessionDid: string): Promise<void> {
const sessionKey = this.getSessionKey(documentId, sessionDid);
this.inMemorySessions.delete(sessionKey);

Expand All @@ -315,11 +311,6 @@ export class SessionManager {
{ documentId, sessionDid },
{ state: "terminated", roomInfo: null }
);

if (appType === "dsheet") {
await DocumentUpdateModel.deleteMany({ documentId, sessionDid });
await DocumentCommitModel.deleteMany({ documentId, sessionDid });
}
} catch (error) {
console.error("Error terminating session in database:", error);
}
Expand Down
Loading
Loading