Skip to content

Commit 2990789

Browse files
coodoscoodos
authored
feat: optmize evault CRUD performance (#951)
* chore: JWKs caching * chore: debug logs * chore: remove log noise * fix: rate-limit failfast * fix: indexing on mid * chore: atomic update * feat: atomic updates * fix: regression on registry participants array * chore: debug logs * chore: fix e2e test
1 parent 97c66fa commit 2990789

8 files changed

Lines changed: 309 additions & 144 deletions

File tree

‎infrastructure/evault-core/src/core/db/db.service.ts‎

Lines changed: 114 additions & 123 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
import neo4j, { type Driver } from "neo4j-driver";
22
import { W3IDBuilder } from "w3id";
3+
import { timed } from "../utils/timing";
34
import { deserializeValue, serializeValue } from "./schema";
45
import type {
56
AppendEnvelopeOperationLogParams,
@@ -41,12 +42,15 @@ export class DbService {
4142
* @returns The result of the query execution
4243
*/
4344
private async runQueryInternal(query: string, params: Record<string, any>) {
44-
const session = this.driver.session();
45-
try {
46-
return await session.run(query, params);
47-
} finally {
48-
await session.close();
49-
}
45+
const firstLine = query.trim().split("\n")[0].slice(0, 80);
46+
return timed(`db.query "${firstLine}"`, async () => {
47+
const session = this.driver.session();
48+
try {
49+
return await session.run(query, params);
50+
} finally {
51+
await session.close();
52+
}
53+
});
5054
}
5155

5256
/**
@@ -74,11 +78,14 @@ export class DbService {
7478
acl: string[],
7579
eName: string,
7680
): Promise<StoreMetaEnvelopeResult<T>> {
81+
return timed("db.storeMetaEnvelope", async () => {
7782
if (!eName) {
7883
throw new Error("eName is required for storing meta-envelopes");
7984
}
8085

81-
const w3id = await new W3IDBuilder().build();
86+
const w3id = await timed("db.storeMetaEnvelope.buildMetaId", () =>
87+
new W3IDBuilder().build(),
88+
);
8289

8390
const cypher: string[] = [
8491
`CREATE (m:MetaEnvelope { id: $metaId, ontology: $ontology, acl: $acl, eName: $eName })`,
@@ -128,7 +135,9 @@ export class DbService {
128135
counter++;
129136
}
130137

131-
await this.runQueryInternal(cypher.join("\n"), envelopeParams);
138+
await timed("db.storeMetaEnvelope.runQuery", () =>
139+
this.runQueryInternal(cypher.join("\n"), envelopeParams),
140+
);
132141

133142
return {
134143
metaEnvelope: {
@@ -138,6 +147,7 @@ export class DbService {
138147
},
139148
envelopes: createdEnvelopes,
140149
};
150+
});
141151
}
142152

143153
/**
@@ -580,91 +590,81 @@ export class DbService {
580590
acl: string[],
581591
eName: string,
582592
): Promise<StoreMetaEnvelopeResult<T>> {
593+
return timed("db.updateMetaEnvelopeById", async () => {
583594
if (!eName) {
584595
throw new Error("eName is required for updating meta-envelopes");
585596
}
586597

598+
// The whole read-modify-write cycle runs inside a single Neo4j write
599+
// transaction. The opening MERGE+SET acquires a write lock on the
600+
// MetaEnvelope node, so concurrent updates to the same id serialize
601+
// here — without this, request B's "delete stale envelopes" step
602+
// could clobber fields that request A just wrote.
603+
const session = this.driver.session();
587604
try {
588-
let existing = await this.findMetaEnvelopeById<T>(id, eName);
589-
if (!existing) {
590-
const metaW3id = await new W3IDBuilder().build();
591-
await this.runQueryInternal(
605+
return await session.executeWrite(async (tx) => {
606+
const findResult = await tx.run(
592607
`
593-
CREATE (m:MetaEnvelope {
594-
id: $id,
595-
ontology: $ontology,
596-
acl: $acl,
597-
eName: $eName
598-
})
608+
MERGE (m:MetaEnvelope { id: $id, eName: $eName })
609+
ON CREATE SET m.ontology = $ontology, m.acl = $acl
610+
ON MATCH SET m.ontology = $ontology, m.acl = $acl
611+
WITH m
612+
OPTIONAL MATCH (m)-[:LINKS_TO]->(e:Envelope)
613+
RETURN collect(e) AS envelopes
599614
`,
600-
{ id, ontology: meta.ontology, acl, eName },
615+
{ id, eName, ontology: meta.ontology, acl },
601616
);
602-
existing = {
603-
id,
604-
ontology: meta.ontology,
605-
acl,
606-
parsed: meta.payload,
607-
envelopes: [],
608-
};
609-
}
610617

611-
// Update the meta-envelope properties (ensure eName matches)
612-
await this.runQueryInternal(
613-
`
614-
MATCH (m:MetaEnvelope { id: $id, eName: $eName })
615-
SET m.ontology = $ontology, m.acl = $acl
616-
`,
617-
{ id, ontology: meta.ontology, acl, eName },
618-
);
618+
const envelopeNodes: any[] = (
619+
findResult.records[0]?.get("envelopes") ?? []
620+
).filter((n: any) => n !== null && n !== undefined);
619621

620-
// Deduplicate envelopes — if multiple Envelope nodes share the
621-
// same ontology (field name), keep the first and delete the rest.
622-
// This prevents non-deterministic reads where collect(e) returns
623-
// duplicates in undefined order and reduce picks the wrong one.
624-
const seen = new Map<string, string>(); // ontology → kept envelope id
625-
const dupsToDelete: string[] = [];
626-
for (const env of existing.envelopes) {
627-
if (seen.has(env.ontology)) {
628-
dupsToDelete.push(env.id);
629-
} else {
630-
seen.set(env.ontology, env.id);
622+
let workingEnvelopes: Envelope<T[keyof T]>[] =
623+
envelopeNodes.map((node: any) => ({
624+
id: node.properties.id,
625+
ontology: node.properties.ontology,
626+
value: deserializeValue(
627+
node.properties.value,
628+
node.properties.valueType,
629+
) as T[keyof T],
630+
valueType: node.properties.valueType,
631+
}));
632+
633+
// Deduplicate envelopes — if multiple Envelope nodes share the
634+
// same ontology, keep the first and delete the rest.
635+
const seen = new Map<string, string>();
636+
const dupsToDelete: string[] = [];
637+
for (const env of workingEnvelopes) {
638+
if (seen.has(env.ontology)) {
639+
dupsToDelete.push(env.id);
640+
} else {
641+
seen.set(env.ontology, env.id);
642+
}
631643
}
632-
}
633-
if (dupsToDelete.length > 0) {
634-
console.warn(
635-
`[eVault] Cleaning ${dupsToDelete.length} duplicate envelope(s) for MetaEnvelope ${id}`,
636-
);
637-
for (const dupId of dupsToDelete) {
638-
await this.runQueryInternal(
639-
`MATCH (e:Envelope { id: $envelopeId }) DETACH DELETE e`,
640-
{ envelopeId: dupId },
644+
if (dupsToDelete.length > 0) {
645+
console.warn(
646+
`[eVault] Cleaning ${dupsToDelete.length} duplicate envelope(s) for MetaEnvelope ${id}`,
647+
);
648+
await tx.run(
649+
`MATCH (e:Envelope) WHERE e.id IN $ids DETACH DELETE e`,
650+
{ ids: dupsToDelete },
651+
);
652+
workingEnvelopes = workingEnvelopes.filter(
653+
(e) => !dupsToDelete.includes(e.id),
641654
);
642655
}
643-
// Remove deleted dupes from the existing list so the update
644-
// loop below doesn't try to reference them.
645-
existing.envelopes = existing.envelopes.filter(
646-
(e) => !dupsToDelete.includes(e.id),
647-
);
648-
}
649656

650-
const createdEnvelopes: Envelope<T[keyof T]>[] = [];
651-
let counter = 0;
657+
const createdEnvelopes: Envelope<T[keyof T]>[] = [];
652658

653-
// For each field in the new payload
654-
for (const [key, value] of Object.entries(meta.payload)) {
655-
try {
659+
for (const [key, value] of Object.entries(meta.payload)) {
656660
const { value: storedValue, type: valueType } =
657661
serializeValue(value);
658-
const alias = `e${counter}`;
659-
660-
// Check if an envelope with this ontology already exists
661-
const existingEnvelope = existing.envelopes.find(
662+
const existingEnvelope = workingEnvelopes.find(
662663
(e) => e.ontology === key,
663664
);
664665

665666
if (existingEnvelope) {
666-
// Update existing envelope
667-
await this.runQueryInternal(
667+
await tx.run(
668668
`
669669
MATCH (e:Envelope { id: $envelopeId })
670670
SET e.value = $newValue, e.valueType = $valueType
@@ -675,88 +675,79 @@ export class DbService {
675675
valueType,
676676
},
677677
);
678-
679678
createdEnvelopes.push({
680679
id: existingEnvelope.id,
681680
ontology: key,
682681
value: value as T[keyof T],
683682
valueType,
684683
});
685684
} else {
686-
// Create new envelope — use MERGE on the relationship
687-
// + ontology to prevent duplicate Envelopes if two
688-
// concurrent updates race.
689685
const envW3id = await new W3IDBuilder().build();
690686
const envelopeId = envW3id.id;
691-
692-
await this.runQueryInternal(
687+
await tx.run(
693688
`
694689
MATCH (m:MetaEnvelope { id: $metaId, eName: $eName })
695-
MERGE (m)-[:LINKS_TO]->(${alias}:Envelope { ontology: $${alias}_ontology })
696-
ON CREATE SET ${alias}.id = $${alias}_id, ${alias}.value = $${alias}_value, ${alias}.valueType = $${alias}_type
697-
ON MATCH SET ${alias}.value = $${alias}_value, ${alias}.valueType = $${alias}_type
690+
MERGE (m)-[:LINKS_TO]->(e:Envelope { ontology: $ontology })
691+
ON CREATE SET e.id = $envelopeId, e.value = $newValue, e.valueType = $valueType
692+
ON MATCH SET e.value = $newValue, e.valueType = $valueType
698693
`,
699694
{
700695
metaId: id,
701-
eName: eName,
702-
[`${alias}_id`]: envelopeId,
703-
[`${alias}_ontology`]: key,
704-
[`${alias}_value`]: storedValue,
705-
[`${alias}_type`]: valueType,
696+
eName,
697+
envelopeId,
698+
ontology: key,
699+
newValue: storedValue,
700+
valueType,
706701
},
707702
);
708-
709703
createdEnvelopes.push({
710704
id: envelopeId,
711705
ontology: key,
712706
value: value as T[keyof T],
713707
valueType,
714708
});
715709
}
716-
717-
counter++;
718-
} catch (error) {
719-
console.error(`Error processing field ${key}:`, error);
720-
throw error;
721710
}
722-
}
723711

724-
// Delete envelopes that are no longer in the payload
725-
const existingOntologies = new Set(Object.keys(meta.payload));
726-
const envelopesToDelete = existing.envelopes.filter(
727-
(e) => !existingOntologies.has(e.ontology),
728-
);
729-
730-
for (const envelope of envelopesToDelete) {
731-
try {
732-
await this.runQueryInternal(
733-
`
734-
MATCH (e:Envelope { id: $envelopeId })
735-
DETACH DELETE e
736-
`,
737-
{ envelopeId: envelope.id },
738-
);
739-
} catch (error) {
740-
console.error(
741-
`Error deleting envelope ${envelope.id}:`,
742-
error,
743-
);
744-
throw error;
712+
// PATCH semantics: fields absent from the new payload are
713+
// left alone. Callers (notably web3-adapter) project partial
714+
// platform updates through toGlobal — if the platform only
715+
// touched one column, only one ontology reaches us, and
716+
// deleting "stale" envelopes here would clobber every other
717+
// field on the meta-envelope (e.g. wiping participantIds when
718+
// a read-receipt update arrives).
719+
720+
// Build the full post-write state by merging the pre-write
721+
// envelope set with everything we just wrote. Used by
722+
// resolvers to fan out webhooks containing the complete
723+
// merged state — receivers overwrite their local row with
724+
// whatever the webhook carries, so a partial diff would
725+
// make them lose every untouched field.
726+
const mergedPayload: Record<string, any> = {};
727+
for (const env of workingEnvelopes) {
728+
mergedPayload[env.ontology] = env.value;
729+
}
730+
for (const env of createdEnvelopes) {
731+
mergedPayload[env.ontology] = env.value;
745732
}
746-
}
747733

748-
return {
749-
metaEnvelope: {
750-
id,
751-
ontology: meta.ontology,
752-
acl,
753-
},
754-
envelopes: createdEnvelopes,
755-
};
734+
return {
735+
metaEnvelope: {
736+
id,
737+
ontology: meta.ontology,
738+
acl,
739+
},
740+
envelopes: createdEnvelopes,
741+
mergedPayload,
742+
};
743+
});
756744
} catch (error) {
757745
console.error("Error in updateMetaEnvelopeById:", error);
758746
throw error;
747+
} finally {
748+
await session.close();
759749
}
750+
});
760751
}
761752

762753
/**

‎infrastructure/evault-core/src/core/db/migrations/add-id-indexes.ts‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/**
2+
* Neo4j Migration: Add point-lookup indexes on Envelope.id and MetaEnvelope.id
3+
*
4+
* Without these, every `MATCH (e:Envelope { id: $id })` and
5+
* `MATCH (m:MetaEnvelope { id: $id })` does a NodeByLabelScan over the entire
6+
* Envelope / MetaEnvelope population, which scales linearly with total stored
7+
* data. Update operations call these queries once per payload field, so a
8+
* 5-field chat update can take 10+ seconds.
9+
*
10+
* `id` is generated by W3IDBuilder and is unique by construction, so a plain
11+
* range index on the property is sufficient — no need for a composite index
12+
* with eName, since the id alone is selective.
13+
*/
14+
15+
import type { Driver } from "neo4j-driver";
16+
17+
const STATEMENTS: { name: string; cypher: string }[] = [
18+
{
19+
name: "envelope_id_index",
20+
cypher: `CREATE INDEX envelope_id_index IF NOT EXISTS FOR (e:Envelope) ON (e.id)`,
21+
},
22+
{
23+
name: "meta_envelope_id_index",
24+
cypher: `CREATE INDEX meta_envelope_id_index IF NOT EXISTS FOR (m:MetaEnvelope) ON (m.id)`,
25+
},
26+
];
27+
28+
export async function createIdIndexes(driver: Driver): Promise<void> {
29+
const session = driver.session();
30+
try {
31+
for (const { name, cypher } of STATEMENTS) {
32+
try {
33+
await session.run(cypher);
34+
console.log(`Ensured ${name}`);
35+
} catch (error) {
36+
if (
37+
error instanceof Error &&
38+
error.message.includes("already exists")
39+
) {
40+
console.log(`${name} already exists`);
41+
} else {
42+
console.error(`Error creating ${name}:`, error);
43+
throw error;
44+
}
45+
}
46+
}
47+
} finally {
48+
await session.close();
49+
}
50+
}

0 commit comments

Comments
 (0)