Skip to content

refactor(storage): add bounded managed mutation capabilities #2186

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Jurij89 merged 13 commits into from
Aug 9, 2026
Merged

refactor(storage): add bounded managed mutation capabilities #2186

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
fb83d61
refactor(storage): add bounded managed mutations
Aug 9, 2026
4f1c1ff
fix(agent): drain bounded join retention backlog
Aug 9, 2026
4de42ad
test(agent): exercise unified projection mutation
Aug 9, 2026
f6544ca
fix(storage): harden structured mutation boundary
Aug 9, 2026
e150951
fix(agent): preserve update-only join moderation
Aug 9, 2026
2157a79
fix(storage): forward structured capability truthfully
Aug 9, 2026
b06bc60
fix(agent): chunk large RS heal projections
Aug 9, 2026
7cba154
refactor(storage): centralize bounded operation probes
Aug 9, 2026
c7d4c48
fix(storage): preserve bounded update compatibility
Aug 9, 2026
d115d07
refactor(agent): unify join retention policy
Aug 9, 2026
6ef9bb9
fix(storage): preserve blob-backed bounded replacements
Aug 9, 2026
19f1ced
refactor(storage): centralize structured quad rewriting
Aug 9, 2026
2c7f33c
fix(storage): retain update-only linked-record pruning
Aug 9, 2026
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
53 changes: 9 additions & 44 deletions packages/agent/src/curator-meta-refresh.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,15 +3,14 @@
import { randomUUID } from 'node:crypto';
import { multiaddr } from '@multiformats/multiaddr';
import {
assertSafeIri,
contextGraphDataGraphUri,
contextGraphMetaGraphUri,
createOperationContext,
DKG_ONTOLOGY,
type OperationContext,
} from '@origintrail-official/dkg-core';
import {
tryUpdateWithTouchedGraphs,
tryReplaceProjectionFromGraphAtomically,
type Quad,
type TripleStore,
} from '@origintrail-official/dkg-storage';
Expand Down Expand Up @@ -415,45 +414,6 @@ async function fetchAuthoritativeMetaSnapshot(
return { checkpointKey: result.checkpointKey, quads: controlMetaQuads };
}

/**
* Replace only the curator-replicated portion of a CG's root `_meta` graph.
* `dkg:revokedAgent` is deliberately retained as a node-local tombstone.
*/
function replaceCuratorMetaProjectionSparql(
contextGraphId: string,
metaGraph: string,
stagingGraph: string,
): string {
const contextGraphUri = assertSafeIri(contextGraphDataGraphUri(contextGraphId));
const delegationPrefix = `did:dkg:agent-delegation:${contextGraphId}:`;
assertSafeIri(metaGraph);
assertSafeIri(stagingGraph);
return `DELETE {
GRAPH <${metaGraph}> { ?staleSubject ?stalePredicate ?staleObject . }
}
INSERT {
GRAPH <${metaGraph}> { ?freshSubject ?freshPredicate ?freshObject . }
}
WHERE {
{
GRAPH <${metaGraph}> {
?staleSubject ?stalePredicate ?staleObject .
FILTER (
(
?staleSubject = <${contextGraphUri}> &&
?stalePredicate != <${DKG_ONTOLOGY.DKG_REVOKED_AGENT}>
) ||
STRSTARTS(STR(?staleSubject), ${JSON.stringify(delegationPrefix)})
)
}
}
UNION
{
GRAPH <${stagingGraph}> { ?freshSubject ?freshPredicate ?freshObject . }
}
}`;
}

async function atomicallyReplaceCuratorMetaSnapshot(
agent: CuratorMetaRefreshAgent,
contextGraphId: string,
Expand Down Expand Up @@ -488,10 +448,15 @@ async function atomicallyReplaceCuratorMetaSnapshot(
invalidateTargetProjections();
let replaced = false;
try {
replaced = await tryUpdateWithTouchedGraphs(
replaced = await tryReplaceProjectionFromGraphAtomically(
agent.store,
replaceCuratorMetaProjectionSparql(contextGraphId, metaGraph, stagingGraph),
[metaGraph],
{
targetGraphUri: metaGraph,
stagingGraphUri: stagingGraph,
targetSubject: contextGraphDataGraphUri(contextGraphId),
preservedTargetPredicates: [DKG_ONTOLOGY.DKG_REVOKED_AGENT],
targetSubjectPrefixes: [`did:dkg:agent-delegation:${contextGraphId}:`],
},
{ source: 'agent.metaRefresh.replace' },
);
} finally {
Expand Down
15 changes: 14 additions & 1 deletion packages/agent/src/dkg-agent-base.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,7 @@ import {
pickNetworkTunables,
isSparqlUpdateOperation,
} from '@origintrail-official/dkg-core';
import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, createTripleStore, isExternalBackend, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage';
import { GraphManager, PrivateContentStore, SystemRecordLaneForwarderV1, createTripleStore, isExternalBackend, structuredMutationMightMutate, structuredMutationTouchedGraphs, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig, type QueryOptions } from '@origintrail-official/dkg-storage';
import { emptyRpcUsageWindow, EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo, type RpcUsageWindow } from '@origintrail-official/dkg-chain';
import {
DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient,
Expand Down Expand Up @@ -576,6 +576,19 @@ export function createListContextGraphsCacheInvalidatingStore(
() => markProjectionDirty?.(undefined, graphUri),
)
: undefined,
structuredMutation: innerStore.structuredMutation
Comment thread
Jurij89 marked this conversation as resolved.
? (mutation, options) => {
// Capture scope before the first await so caller-side mutation cannot
// redirect cache invalidation after the backend has committed.
const targetGraphs = [...structuredMutationTouchedGraphs(mutation)];
const mightMutate = structuredMutationMightMutate(mutation);
return invalidateAfterMutation(
() => innerStore.structuredMutation!(mutation, options),
() => mightMutate,
() => targetGraphs.forEach((graph) => markProjectionDirty?.(undefined, graph)),
);
}
: undefined,
listGraphs(options) {
return innerStore.listGraphs(options);
},
Expand Down
166 changes: 66 additions & 100 deletions packages/agent/src/dkg-agent-join.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ import {
OPEN_ENROLLMENT_MAX_APPROVALS_PER_HOUR,
isBoundedOpenEnrollmentPolicy,
} from '@origintrail-official/dkg-core';
import { GraphManager, PrivateContentStore, createTripleStore, tryReplaceSubjectAtomically, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage';
import { GraphManager, PrivateContentStore, createTripleStore, tryReplaceSubjectAtomically, tryReplaceSubjectPredicatesAtomically, type TripleStore, type TripleStoreConfig, type Quad, type LargeLiteralStorageConfig } from '@origintrail-official/dkg-storage';
import { EVMChainAdapter, NoChainAdapter, enrichEvmError, buildKnowledgeAssetUal, type EVMAdapterConfig, type ChainAdapter, type CreateContextGraphParams, type CreateOnChainContextGraphParams, type CreateOnChainContextGraphResult, type TxResult, type V10PublishingConvictionAccountInfo } from '@origintrail-official/dkg-chain';
import {
DKGPublisher, PublishHandler, SharedMemoryHandler, UpdateHandler, ChainEventPoller, AccessHandler, AccessClient,
Expand Down Expand Up @@ -396,6 +396,37 @@ const JOIN_REQUEST_INGRESS_PER_PEER = 20;
const JOIN_REQUEST_INGRESS_PER_AGENT = 6;
const JOIN_REQUEST_INGRESS_PER_CONTEXT_GRAPH = 100;
const JOIN_REQUEST_INGRESS_MAX_QUEUE_DEPTH = 64;
const JOIN_REQUEST_STATUS_PREDICATE = 'https://dkg.network/ontology#requestStatus';
const JOIN_REQUEST_DECISION_TIMESTAMP_PREDICATE = 'https://dkg.network/ontology#decisionTimestamp';

function joinRequestDecisionReplacement(
contextGraphId: string,
agentAddress: string,
status: 'approved' | 'rejected',
decidedAt = Date.now(),
) {
const graphUri = contextGraphMetaGraphUri(contextGraphId);
const subject = `did:dkg:join-request:${contextGraphId}:${agentAddress.toLowerCase()}`;
return {
graphUri,
subject,
predicates: [JOIN_REQUEST_STATUS_PREDICATE, JOIN_REQUEST_DECISION_TIMESTAMP_PREDICATE],
replacementQuads: [
{
subject,
predicate: JOIN_REQUEST_STATUS_PREDICATE,
object: `"${status}"`,
graph: graphUri,
},
{
subject,
predicate: JOIN_REQUEST_DECISION_TIMESTAMP_PREDICATE,
object: `"${decidedAt}"`,
graph: graphUri,
},
],
};
}

export interface ContextGraphJoinPolicyStatus {
contextGraphId: string;
Expand Down Expand Up @@ -1957,12 +1988,11 @@ export class JoinRequestMethods extends DKGAgentBase {
);
}

// Every production adapter implements SPARQL UPDATE. Refuse to cross the
// membership boundary on a custom adapter that cannot atomically replace
// the moderation status.
if (typeof this.store.update !== 'function') {
// Refuse to cross the membership boundary on a custom adapter that cannot
// atomically replace the moderation predicates after the invite commits.
if (typeof this.store.structuredMutation !== 'function') {
Comment thread
Jurij89 marked this conversation as resolved.
Outdated
throw new Error(
'Join approval requires atomic SPARQL UPDATE support from the configured triple store.',
'Join approval requires atomic subject-predicate replacement support from the configured triple store.',
);
}

Expand Down Expand Up @@ -2009,44 +2039,23 @@ export class JoinRequestMethods extends DKGAgentBase {

/** Complete the moderation side of an approval; safe to repeat after retry. */
async markJoinRequestApproved(this: DKGAgent, contextGraphId: string, agentAddress: string): Promise<void> {
const cgMetaGraph = contextGraphMetaGraphUri(contextGraphId);
const requestUri = `did:dkg:join-request:${contextGraphId}:${agentAddress.toLowerCase()}`;
const requestStatus = 'https://dkg.network/ontology#requestStatus';
const decisionTimestamp = 'https://dkg.network/ontology#decisionTimestamp';
const decidedAt = Date.now();
if (typeof this.store.update !== 'function') {
const replaced = await tryReplaceSubjectPredicatesAtomically(
Comment thread
Jurij89 marked this conversation as resolved.
this.store,
joinRequestDecisionReplacement(contextGraphId, agentAddress, 'approved'),
{ source: 'join-approval-status' },
);
if (!replaced) {
throw new Error(
'Join approval requires atomic SPARQL UPDATE support from the configured triple store.',
'Join approval requires atomic subject-predicate replacement support from the configured triple store.',
);
}
// Status and retention ordering must cross the terminal boundary together.
await this.store.update(`
DELETE { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> ?oldStatus .
<${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp .
} }
INSERT { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> "approved" .
<${requestUri}> <${decisionTimestamp}> "${decidedAt}" .
} }
WHERE {
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${requestStatus}> ?oldStatus . } }
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp . } }
}
`, { touchedGraphs: [cgMetaGraph], source: 'join-approval-status' });
await this.pruneTerminalJoinRequestHistory(contextGraphId);
}

/** Keep curator/requester moderation state bounded without risking the decision path. */
async pruneTerminalJoinRequestHistory(this: DKGAgent, contextGraphId: string): Promise<void> {
try {
const pruned = await pruneTerminalJoinRequestRecords(this.store, contextGraphId);
if (pruned > 0) {
this.log.info(
createOperationContext('system'),
`Pruned ${pruned} terminal join-request record(s) for "${contextGraphId}"`,
);
}
await pruneTerminalJoinRequestRecords(this.store, contextGraphId);
} catch (error) {
// Retention is resource hygiene, not part of the authorization commit.
// Keep the terminal decision durable and retry pruning on the next one.
Expand Down Expand Up @@ -2256,55 +2265,27 @@ export class JoinRequestMethods extends DKGAgentBase {
agentAddress: string,
status: 'approved' | 'rejected',
): Promise<void> {
const cgMetaGraph = contextGraphMetaGraphUri(contextGraphId);
const requestUri = `did:dkg:join-request:${contextGraphId}:${agentAddress.toLowerCase()}`;
const requestStatus = 'https://dkg.network/ontology#requestStatus';
const decisionTimestamp = 'https://dkg.network/ontology#decisionTimestamp';
const decidedAt = Date.now();

if (typeof this.store.update === 'function') {
await this.store.update(`
DELETE { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> ?oldStatus .
<${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp .
} }
INSERT { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> "${status}" .
<${requestUri}> <${decisionTimestamp}> "${decidedAt}" .
} }
WHERE {
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${requestStatus}> ?oldStatus . } }
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp . } }
}
`, { touchedGraphs: [cgMetaGraph], source: `local-join-${status}-status` });
} else {
// Compatibility fallback for custom stores without SPARQL UPDATE.
const replacement = joinRequestDecisionReplacement(contextGraphId, agentAddress, status);
const replaced = await tryReplaceSubjectPredicatesAtomically(
this.store,
replacement,
{ source: `local-join-${status}-status` },
);
if (!replaced) {
// Compatibility fallback for custom stores without atomic replacement.
// The curator remains authoritative if a crash lands between these two
// mutations, and an approval can be redelivered to repair local state.
await this.store.deleteByPattern({
graph: cgMetaGraph,
subject: requestUri,
predicate: requestStatus,
graph: replacement.graphUri,
subject: replacement.subject,
predicate: JOIN_REQUEST_STATUS_PREDICATE,
});
await this.store.deleteByPattern({
graph: cgMetaGraph,
subject: requestUri,
predicate: decisionTimestamp,
graph: replacement.graphUri,
subject: replacement.subject,
predicate: JOIN_REQUEST_DECISION_TIMESTAMP_PREDICATE,
});
await this.store.insert([
{
subject: requestUri,
predicate: requestStatus,
object: `"${status}"`,
graph: cgMetaGraph,
},
{
subject: requestUri,
predicate: decisionTimestamp,
object: `"${decidedAt}"`,
graph: cgMetaGraph,
},
]);
await this.store.insert(replacement.replacementQuads);
}

await this.pruneTerminalJoinRequestHistory(contextGraphId);
Expand Down Expand Up @@ -2488,9 +2469,6 @@ export class JoinRequestMethods extends DKGAgentBase {
// graph curator can …" (403 at the route) for a non-curator.
await this.assertContextGraphOwner(contextGraphId, callerAgentAddress, 'manage join requests');

const cgMetaGraph = contextGraphMetaGraphUri(contextGraphId);
const requestUri = `did:dkg:join-request:${contextGraphId}:${agentAddress.toLowerCase()}`;
const DKG = 'https://dkg.network/ontology#';
const delegation = await this.loadPendingJoinDelegation(contextGraphId, agentAddress);
if (!delegation) {
throw new Error(`No pending join request found from ${agentAddress}`);
Expand All @@ -2506,28 +2484,16 @@ export class JoinRequestMethods extends DKGAgentBase {
) {
throw new Error('Pending join request generation does not match its signed delegation');
}
const requestStatus = `${DKG}requestStatus`;
const decisionTimestamp = `${DKG}decisionTimestamp`;
const decidedAt = Date.now();
if (typeof this.store.update !== 'function') {
const replaced = await tryReplaceSubjectPredicatesAtomically(
this.store,
joinRequestDecisionReplacement(contextGraphId, agentAddress, 'rejected'),
{ source: 'join-rejected' },
);
if (!replaced) {
throw new Error(
'Join rejection requires atomic SPARQL UPDATE support from the configured triple store.',
'Join rejection requires atomic subject-predicate replacement support from the configured triple store.',
);
}
await this.store.update(`
DELETE { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> ?oldStatus .
<${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp .
} }
INSERT { GRAPH <${cgMetaGraph}> {
<${requestUri}> <${requestStatus}> "rejected" .
<${requestUri}> <${decisionTimestamp}> "${decidedAt}" .
} }
WHERE {
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${requestStatus}> ?oldStatus . } }
OPTIONAL { GRAPH <${cgMetaGraph}> { <${requestUri}> <${decisionTimestamp}> ?oldDecisionTimestamp . } }
}
`, { touchedGraphs: [cgMetaGraph], source: 'join-rejected' });
await this.pruneTerminalJoinRequestHistory(contextGraphId);
this.upsertContextGraphMember({
contextGraphId,
Expand Down
Loading
Loading