diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index ab60141dd8..f48cf7ab4f 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -47,6 +47,7 @@ * [Host-Mode Manual Subscribe](use-dkg/host-mode-manual-subscribe.md) * [Updates & Rollback](use-dkg/updates-and-rollback.md) * [Migrate to npm](use-dkg/migrate-to-npm.md) +* [Backpressure Observability](use-dkg/backpressure-observability.md) * [Troubleshooting](use-dkg/troubleshooting.md) ## Agent Context diff --git a/docs/assets/pr-2003-scheduler-work-flamegraph.png b/docs/assets/pr-2003-scheduler-work-flamegraph.png new file mode 100644 index 0000000000..f04ef43813 Binary files /dev/null and b/docs/assets/pr-2003-scheduler-work-flamegraph.png differ diff --git a/docs/assets/pr-2003-scheduler-work-flamegraph.svg b/docs/assets/pr-2003-scheduler-work-flamegraph.svg new file mode 100644 index 0000000000..d6e7f5671e --- /dev/null +++ b/docs/assets/pr-2003-scheduler-work-flamegraph.svg @@ -0,0 +1,188 @@ + + PR 2003 realistic 3 MiB scheduler work distribution + Flamegraph of 1660.876 active scheduler slot-seconds sampled every 100 milliseconds on the Blazegraph node of a five-core Hardhat devnet. Four exact 3 MiB RDF payloads exercised online SWM and VM work, offline finalization, restart, durable synchronization, and VM reconciliation with 1200 milliseconds of injected Blazegraph latency. Sync-global durable work used 42.76 percent of observed slot-time, while generic blazegraph.query used 25.73 percent. + + + + + Realistic 3 MiB scheduler work distribution + 5-core Hardhat devnet · 4 × 3 MiB RDF payloads · 577.080 s profile · +1,200 ms Blazegraph/request + + + + 1,660.876 s + observed active slot-time + + + + 4,590 samples + 100 ms diagnostics snapshots + + + + 216 / 216 VM + all 5 nodes · watermark 0 → 4 + + + + 427.384 s · 25.73% + still generic blazegraph.query + + + ALL OBSERVED ACTIVE SCHEDULER SLOTS + + + node 3 scheduler work + 1,660.876 slot-seconds · 100.00% + + + SCHEDULERS + + + store + 947.910 s · 57.07% + + sync-global + 712.966 s · 42.93% + + + LANES + + + store · normal + 733.707 s · 44.18% + + + background + 198.681 s · 11.96% + + + + + sync-global · durable + 710.237 s · 42.76% + + + + + CALLER / PROVIDER SOURCES + + + blazegraph.query + store · normal · generic source + 427.384 s · 25.73% + + + recover + 3.08% + + + run + 2.97% + + + cand. + 2.93% + + + other normal + 157.130 s · 9.46% + + + mat. + 3.08% + + + list + 2.01% + + + other bg + 114.172 s · 6.87% + + + + + durable + sync-global · durable lane + 710.237 s · 42.76% + + + + + Largest observed sources + + + + Scheduler / lane + Source + Slot time + Global share + + + + sync-global / durable + durable + 710.237 s + 42.76% + + + + store / normal + blazegraph.query + 427.384 s + 25.73% + + + + store / background + agent.durableSync.graphScopedMaterialization + 51.187 s + 3.08% + + + + store / normal + publisher.asyncPromote.recoverExpired + 51.116 s + 3.08% + + + + store / normal + publisher.asyncPromote.claimNext.running + 49.404 s + 2.97% + + + + store / normal + publisher.asyncPromote.claimNext.candidates + 48.673 s + 2.93% + + + + store / background + sync.responder.listGraphs + 33.322 s + 2.01% + + + + store / background + sync.responder.readRegisteredSubGraphNames + 24.335 s + 1.47% + + + + + Queue-pressure maxima + store normal: 10 queued · 9.945 s oldest wait · 0 rejected + sync durable: 3 queued · 71.830 s oldest wait · 114.409 s oldest active · 0 rejected + + Slot-time = Σ(sample interval × active-operation count). It is concurrency-weighted, not CPU or wall time, so 1,660.876 s can exceed the 577.080 s profile. + 1,028 snapshots failed while node 3 was intentionally offline or its API was booting; shares describe successfully observed scheduler state only. + + diff --git a/docs/use-dkg/backpressure-observability.md b/docs/use-dkg/backpressure-observability.md new file mode 100644 index 0000000000..6ea8aec4a6 --- /dev/null +++ b/docs/use-dkg/backpressure-observability.md @@ -0,0 +1,173 @@ +--- +status: current +version: v10 +audience: operator+developer +doc_type: how-to +--- + +# Backpressure observability + +DKG exposes one common pressure model for its in-memory schedulers. It answers +four questions without requiring scheduler-specific log archaeology: + +1. Which scheduler and lane are under pressure? +2. Is work waiting, rejected, or admitted but possibly stuck? +3. Which bounded operation classes account for that work? +4. Did the scheduler recover? + +This is an observability layer only. It does not change queue ordering, +concurrency, reservation, displacement, timeout, or retry behavior. + +## Architecture + +```mermaid +flowchart LR + subgraph Producers["Work producers"] + SWM["SWM and store callers"] + Sync["Sync requester stages"] + end + + subgraph Schedulers["Existing scheduling policies"] + Store["StorePriorityScheduler
ack / health / normal / background"] + Global["PriorityAdmissionQueue
global sync admission"] + end + + subgraph Core["Shared core observability"] + Lifecycle["ObservableScheduler lifecycle
enqueue / start / reject / finish"] + Tracker["SchedulerPressureTracker
ages, counts, bounded operation summaries"] + Registry["BackpressureRegistry"] + Monitor["BackpressureMonitor
5 s samples, transition and recovery logs"] + OTel["OpenTelemetry metrics"] + end + + Status["Public /api/status
state only"] + Admin["Node-admin
/api/diagnostics/backpressure"] + Logs["Daemon log"] + + SWM --> Store + Sync --> Global + Store --> Lifecycle + Global --> Lifecycle + Lifecycle --> Tracker + Tracker --> OTel + Tracker --> Registry + Registry --> Monitor + Monitor --> OTel + Monitor --> Logs + Registry --> Status + Registry --> Admin +``` + +The tracker records references to bounded labels and timestamps, not work +closures or payloads. A snapshot can contain a maximum of eight queued and +eight active operation summaries per lane. Labels are sanitized and truncated +before they reach metrics, logs, or the diagnostics response. + +The first registered sources are: + +- `store`: the external-store priority scheduler and its `ack`, `health`, + `normal`, and `background` lanes; +- `sync-global`: the process-wide sync admission queue and its sync lanes. + +Other schedulers can extend `ObservableScheduler` and call its protected +lifecycle methods at their existing admission boundaries. They keep complete +ownership of policy. + +## Pressure states + +| State | Meaning | +| --- | --- | +| `healthy` | No age, utilization, rejection, or active-duration threshold is crossed. | +| `degraded` | A queue is old or at least 75% utilized, but is not full. | +| `saturated` | A queue is full or an admission rejection occurred in the recent visibility window. | +| `stalled` | The oldest admitted operation crossed the scheduler's active-duration threshold. | + +State precedence is `stalled` > `saturated` > `degraded` > `healthy`. A recent +rejection remains visible for 60 seconds so a short full-queue event is not +missed between monitor samples. + +These states describe evidence, not root cause. For example, a stalled store +operation can be caused by Blazegraph, Oxigraph, disk, or a caller that never +settles. Use the operation summary and surrounding store logs to continue the +investigation. + +## Read current pressure + +`GET /api/status` is public and includes only the aggregate state: + +```json +{ + "backpressure": { + "state": "degraded", + "schedulers": [ + { "scheduler": "store", "state": "degraded" }, + { "scheduler": "sync-global", "state": "healthy" } + ], + "diagnosticsAvailable": "/api/diagnostics/backpressure" + } +} +``` + +The detailed route requires the node-level admin token. Agent-scoped tokens are +rejected because the response describes node-wide work. + +```bash +TOKEN=$(dkg auth show) +curl -sS \ + -H "Authorization: Bearer $TOKEN" \ + http://127.0.0.1:9200/api/diagnostics/backpressure | jq +``` + +The response reports current queue/inflight counts and limits, oldest ages, +cumulative lifecycle/rejection counts, and bounded operation summaries. It +does not expose request bodies, SPARQL text, graph or peer identifiers, work +closures, or durable queue payloads. + +## Log behavior + +The daemon samples registered sources every five seconds. It emits: + +- a warning immediately when a lane enters or changes a non-healthy state; +- one warning summary per minute while that state persists; +- an info message when the lane recovers. + +All messages start with `[backpressure]` and carry a JSON object: + +```text +[warn] [backpressure] {"event":"transition","scheduler":"store","lane":"normal","state":"degraded","previousState":"healthy","queued":3,"queueLimit":4,"inflight":4,"inflightLimit":4,"oldestQueuedAgeMs":15234,"oldestActiveAgeMs":19310,"rejectedTotal":0,"queuedOperations":[{"operation":"blazegraph.query","count":3,"oldestAgeMs":15234}],"activeOperations":[{"operation":"publisher.swm.graphScopedReplace","count":4,"oldestAgeMs":19310}]} +``` + +Per-item enqueue/start logs are deliberately avoided. Transition and periodic +summary logging make sustained pressure visible without creating a log storm +that competes with the overloaded scheduler. + +## Metrics + +The common OpenTelemetry instruments use bounded `scheduler` and `lane` +attributes: + +| Metric | Type | Purpose | +| --- | --- | --- | +| `dkg.backpressure.queue_depth` | gauge | Current waiting work | +| `dkg.backpressure.queue_limit` | gauge | Configured queue capacity | +| `dkg.backpressure.inflight` | gauge | Current admitted work | +| `dkg.backpressure.inflight_limit` | gauge | Configured concurrency | +| `dkg.backpressure.oldest_queued_age_ms` | gauge | Head-of-line age | +| `dkg.backpressure.oldest_active_age_ms` | gauge | Oldest admitted duration | +| `dkg.backpressure.events_total` | counter | Lifecycle and rejection events | +| `dkg.backpressure.queue_wait_ms` | histogram | Completed queue waits | +| `dkg.backpressure.active_duration_ms` | histogram | Completed admitted durations | + +Operation names are intentionally excluded from the common current-value +gauges. They remain available in bounded diagnostic/log summaries, while +metrics retain predictable cardinality. + +## Failure containment + +Instrumentation is fail-open: + +- metric recording failures do not alter admission or completion; +- one broken source is reported in the registry's `failures` array without + hiding healthy sources; +- log callback failures do not stop the monitor; +- the monitor timer is unreferenced and is stopped during daemon shutdown. diff --git a/packages/agent/src/dkg-agent-cg-registry.ts b/packages/agent/src/dkg-agent-cg-registry.ts index 95eb1b951f..115602f5b7 100644 --- a/packages/agent/src/dkg-agent-cg-registry.ts +++ b/packages/agent/src/dkg-agent-cg-registry.ts @@ -385,6 +385,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { const contextGraphUri = `did:dkg:context-graph:${contextGraphId}`; const result = await this.store.query( `SELECT ?status WHERE { GRAPH <${cgMetaGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_REGISTRATION_STATUS}> ?status } } LIMIT 1`, + { source: 'agent.contextGraph.registrationStatus' }, ); return result.type === 'bindings' && result.bindings[0]?.['status']?.replace(/^"|"$/g, '') === 'registered'; } @@ -415,7 +416,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { async getContextGraphOnChainId( this: DKGAgent, contextGraphId: string, - options: { signal?: AbortSignal } = {}, + options: { signal?: AbortSignal; source?: string } = {}, ): Promise { const subscribed = this.subscribedContextGraphs.get(contextGraphId)?.onChainId; if (subscribed) return subscribed; @@ -434,7 +435,10 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { const contextGraphUri = `did:dkg:context-graph:${contextGraphId}`; const result = await this.store.query( `SELECT ?id WHERE { GRAPH <${ontologyGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainId> ?id } } LIMIT 1`, - { signal: options.signal }, + { + signal: options.signal, + source: options.source ?? 'agent.contextGraph.onChainId', + }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return null; const value = result.bindings[0]?.['id']; @@ -828,6 +832,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_PUBLISH_POLICY}> ?pp } OPTIONAL { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_PUBLISH_AUTHORITY_ACCOUNT_ID}> ?paa } } } LIMIT 1`, + { source: 'agent.contextGraph.registrationOptions' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return {}; const row = result.bindings[0] ?? {}; @@ -963,7 +968,9 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { }, ); if (!updated) { - await this.store.query(sparql); + await this.store.query(sparql, { + source: 'agent.cg.removeSubGraph.registrationFallback', + }); } } catch { // SPARQL DELETE WHERE may not be supported — delete quads manually @@ -1132,6 +1139,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { const exists = await this.store.query( `SELECT ?hit WHERE { ${existsPatterns.map((p) => `{ ${p} }`).join(' UNION ')} } LIMIT 1`, + { source: 'agent.endorsement.resolveTarget.exists' }, ); if (exists.type !== 'bindings' || exists.bindings.length === 0) { throw new Error( @@ -1149,6 +1157,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { ]); const roots = await this.store.query( `SELECT DISTINCT ?root WHERE { ${rootPatterns.map((p) => `{ ${p} }`).join(' UNION ')} }`, + { source: 'agent.endorsement.resolveTarget.roots' }, ); const rootEntities = roots.type === 'bindings' ? (roots.bindings as Record[]).map((row) => row.root).filter(Boolean) @@ -1183,6 +1192,7 @@ export class ContextGraphRegistryMethods extends DKGAgentBase { .join(' || '); const result = await this.store.query( `SELECT DISTINCT ?s WHERE { GRAPH <${safeGraph}> { ?s ?p ?o . FILTER(${filterClauses}) } }`, + { source: 'agent.endorsement.resolveRootSubjects' }, ); const subjects = new Set(rootEntities); if (result.type === 'bindings') { diff --git a/packages/agent/src/dkg-agent-context-graph.ts b/packages/agent/src/dkg-agent-context-graph.ts index d7628e0671..4317e4ba1b 100644 --- a/packages/agent/src/dkg-agent-context-graph.ts +++ b/packages/agent/src/dkg-agent-context-graph.ts @@ -928,6 +928,7 @@ export class ContextGraphMethods extends DKGAgentBase { UNION { GRAPH <${cgMetaGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> ?ap } } } LIMIT 1`, + { source: 'agent.contextGraph.register.accessPolicy' }, ); const apValue = accessPolicyResult.type === 'bindings' ? accessPolicyResult.bindings[0]?.['ap']?.replace(/^"|"$/g, '') @@ -991,6 +992,7 @@ export class ContextGraphMethods extends DKGAgentBase { const contextGraphUri = `did:dkg:context-graph:${id}`; const statusResult = await this.store.query( `SELECT ?status WHERE { GRAPH <${cgMetaGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_REGISTRATION_STATUS}> ?status } } LIMIT 1`, + { source: 'agent.contextGraph.register.status' }, ); if (statusResult.type === 'bindings' && statusResult.bindings[0]?.['status']?.replace(/^"|"$/g, '') === 'registered') { const existingOnChainId = this.subscribedContextGraphs.get(id)?.onChainId; @@ -1006,6 +1008,7 @@ export class ContextGraphMethods extends DKGAgentBase { UNION { GRAPH <${cgMetaGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.SCHEMA_DESCRIPTION}> ?desc } } } LIMIT 1`, + { source: 'agent.contextGraph.register.description' }, ); const description = descResult.type === 'bindings' ? descResult.bindings[0]?.['desc']?.replace(/^"|"$/g, '') : undefined; diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 34b190fa1b..aecad543da 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -2153,6 +2153,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { for (const literal of [`"${batchId}"^^`, `"${batchId}"`]) { const result = await this.store.query( `SELECT ?root WHERE { GRAPH <${metaGraph}> { ?kc <${ns}merkleRoot> ?root . ?kc <${ns}batchId> ${literal} } } LIMIT 1`, + { source: 'agent.verifyProposal.batchMerkleRoot' }, ); if (result.type === 'bindings' && result.bindings.length > 0) { const hex = (result.bindings[0] as Record)['root']; @@ -6288,17 +6289,20 @@ export class LifecycleSyncMethods extends DKGAgentBase { const metaGraph = contextGraphMetaGraphUri(contextGraphId); const onChainIdPredicate = `${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainId`; const onChainHashPredicate = `${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainHash`; - const registrationResult = await this.store.query(` - SELECT ?predicate ?value WHERE { - GRAPH <${metaGraph}> { - <${contextGraphUri}> ?predicate ?value . - VALUES ?predicate { - <${onChainIdPredicate}> - <${onChainHashPredicate}> + const registrationResult = await this.store.query( + ` + SELECT ?predicate ?value WHERE { + GRAPH <${metaGraph}> { + <${contextGraphUri}> ?predicate ?value . + VALUES ?predicate { + <${onChainIdPredicate}> + <${onChainHashPredicate}> + } } } - } - `); + `, + { source: 'agent.durableSync.registrationBinding' }, + ); let confirmedOnChainId: string | undefined; let confirmedOnChainHash: string | undefined; if (registrationResult.type === 'bindings') { @@ -6868,6 +6872,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { OPTIONAL { <${delegationUri}> <${DKG_ONTOLOGY.DKG_ALLOWED_DELEGATEE_KEY}> ?opKey } } } LIMIT 1`, + { source: 'agent.delegationRefresh.currentState' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return; @@ -7356,6 +7361,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_REGISTRATION_STATUS}> "unregistered" . } }`, + { source: 'agent.contextGraph.confirmedMeta.unregisteredPlaceholder' }, ); const hasUnregisteredPlaceholder = unregisteredPlaceholderResult.type === 'boolean' && unregisteredPlaceholderResult.value === true; @@ -7396,6 +7402,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { } : undefined, ), + { source: 'agent.contextGraph.confirmedMeta.privateDefinition' }, ); if ( authoritativeDefinitionResult.type === 'boolean' && @@ -7412,6 +7419,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { // member delegation. const authoritativePublicDefinitionResult = await this.store.query( buildAuthoritativePublicMetaAskQuery(contextGraphId), + { source: 'agent.contextGraph.confirmedMeta.publicDefinition' }, ); if ( authoritativePublicDefinitionResult.type === 'boolean' && @@ -7480,6 +7488,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . } }`, + { source: 'agent.contextGraph.confirmedMeta.ontologyDeclaration' }, ); return ontologyResult.type === 'boolean' && ontologyResult.value === true; } @@ -7605,6 +7614,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { FILTER(?ts < "${cutoff}"^^) } }`, + { source: 'agent.swmCleanup.expiredOperations' }, ); if (expiredOps.type !== 'bindings' || expiredOps.bindings.length === 0) continue; @@ -7620,6 +7630,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { <${opUri}> ?re . } }`, + { source: 'agent.swmCleanup.operationRoots' }, ); const rootEntities: string[] = []; @@ -7660,6 +7671,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { OPTIONAL { <${opUri}> ?snapshotGraph } } } LIMIT 1`, + { source: 'agent.swmCleanup.graphScopedMetadata' }, ); const v2Row = v2Meta.type === 'bindings' ? v2Meta.bindings[0] : undefined; const scopeVersion = v2Row?.['scopeVersion'] === undefined ? NaN : Number(stripLiteral(v2Row['scopeVersion'])); @@ -7679,6 +7691,7 @@ export class LifecycleSyncMethods extends DKGAgentBase { OPTIONAL { <${headSubject}> ?assertionGraph } } } LIMIT 1`, + { source: 'agent.swmCleanup.currentHeadOwner' }, ); if (headOwned.type === 'bindings' && headOwned.bindings.length > 0) { // Whole KA expired: drop the per-KA SWM assertion graph and @@ -7959,17 +7972,20 @@ async function getSharedMemorySubGraphAdmission( async function isKnownContextGraphUri(store: TripleStore, contextGraphUri: string): Promise { const metaGraph = `${contextGraphUri}/_meta`; - const result = await store.query(` - ASK { - GRAPH <${assertSafeIri(metaGraph)}> { - { - <${assertSafeIri(contextGraphUri)}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - } UNION { - <${assertSafeIri(contextGraphUri)}> <${DKG_ONTOLOGY.DKG_REGISTRATION_STATUS}> ?status . + const result = await store.query( + ` + ASK { + GRAPH <${assertSafeIri(metaGraph)}> { + { + <${assertSafeIri(contextGraphUri)}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + } UNION { + <${assertSafeIri(contextGraphUri)}> <${DKG_ONTOLOGY.DKG_REGISTRATION_STATUS}> ?status . + } } } - } - `); + `, + { source: 'agent.subGraphClassification.knownContextGraph' }, + ); return result.type === 'boolean' && result.value; } /** diff --git a/packages/agent/src/dkg-agent-publish.ts b/packages/agent/src/dkg-agent-publish.ts index 1bbe1d962f..f0a5feac5a 100644 --- a/packages/agent/src/dkg-agent-publish.ts +++ b/packages/agent/src/dkg-agent-publish.ts @@ -551,6 +551,7 @@ async function resolveDirectRootlessUpdateScope( const result = await agent.store.query( `CONSTRUCT { <${safeUal}> ?p ?o } WHERE { ` + `GRAPH <${metaGraph}> { <${safeUal}> ?p ?o } }`, + { source: 'agent.publish.rootlessUpdate.currentMetadata' }, ); const rows = result.type === 'quads' ? result.quads : []; @@ -560,6 +561,7 @@ async function resolveDirectRootlessUpdateScope( + `?ual <${ROOTLESS_UPDATE_DKG_NS}batchId> "${kaId.toString()}"^^<${ROOTLESS_UPDATE_XSD_INTEGER}> . ` + `OPTIONAL { ?ual <${ROOTLESS_UPDATE_DKG_NS}contentScopeVersion> ?scope } ` + `} } LIMIT 2`, + { source: 'agent.publish.rootlessUpdate.legacyBatchLookup' }, ); if (byBatch.type === 'bindings' && byBatch.bindings.length > 0) { throw new LegacyKnowledgeAssetReadOnlyError(); @@ -2420,21 +2422,24 @@ export class PublishMethods extends DKGAgentBase { const contextGraphUri = contextGraphDataGraphUri(contextGraphId); const ontologyGraph = contextGraphDataGraphUri(SYSTEM_CONTEXT_GRAPHS.ONTOLOGY); const cgMetaGraph = contextGraphMetaGraphUri(contextGraphId); - const result = await this.store.query(` - ASK WHERE { - { - GRAPH <${ontologyGraph}> { - <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + const result = await this.store.query( + ` + ASK WHERE { + { + GRAPH <${ontologyGraph}> { + <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + } } - } - UNION - { - GRAPH <${cgMetaGraph}> { - <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + UNION + { + GRAPH <${cgMetaGraph}> { + <${contextGraphUri}> <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + } } } - } - `); + `, + { source: 'agent.publish.contextGraphDefinition' }, + ); return result.type === 'boolean' && result.value === true; } @@ -2669,6 +2674,7 @@ export class PublishMethods extends DKGAgentBase { // recover that partition from the immutable `(UAL, assertionVersion)` graph. const existingMetaResult = await this.store.query( `CONSTRUCT { <${assertionUri}> ?p ?o } WHERE { GRAPH <${metaGraph}> { <${assertionUri}> ?p ?o } }`, + { source: 'agent.assertionFinalize.existingSeal' }, ); const existingMetaQuads = existingMetaResult.type === 'quads' ? existingMetaResult.quads : []; @@ -2844,6 +2850,7 @@ export class PublishMethods extends DKGAgentBase { if (reReservedKaId === undefined) { const reKaIdRes = await this.store.query( `SELECT ?n WHERE { GRAPH <${metaGraph}> { <${assertionLifecycleUri(contextGraphId, agentAddress, name, opts?.subGraphName)}> <${KA_ID_PRED}> ?n } } LIMIT 1`, + { source: 'agent.assertionFinalize.reservedKaId' }, ); const reNum = reKaIdRes.type === 'bindings' && reKaIdRes.bindings[0]?.['n'] !== undefined @@ -3017,6 +3024,7 @@ export class PublishMethods extends DKGAgentBase { OPTIONAL { <${lifecycleUri}> <${ASSERTION_SEAL_PREDICATES.ASSERTION_VERSION}> ?version } OPTIONAL { <${lifecycleUri}> <${VM_CURRENT_ASSERTION_PRED}> ?vm } } } LIMIT 1`, + { source: 'agent.assertionFinalize.lifecycleScope' }, ); const lifecycleScopeRow = lifecycleScopeResult.type === 'bindings' ? lifecycleScopeResult.bindings[0] @@ -3044,6 +3052,7 @@ export class PublishMethods extends DKGAgentBase { } const existingKaIdRes = await this.store.query( `SELECT ?n WHERE { GRAPH <${metaGraph}> { <${lifecycleUri}> <${KA_ID_PRED}> ?n } } LIMIT 1`, + { source: 'agent.assertionFinalize.existingKaId' }, ); const hasExistingKaId = existingKaIdRes.type === 'bindings' && existingKaIdRes.bindings.length > 0; @@ -3106,6 +3115,7 @@ export class PublishMethods extends DKGAgentBase { const stampedNumber = parseStampedNumber(existingKaIdRes.bindings[0]['n']); const reservedUalRes = await this.store.query( `SELECT ?u WHERE { GRAPH <${metaGraph}> { <${lifecycleUri}> <${RESERVED_UAL_PRED}> ?u } } LIMIT 1`, + { source: 'agent.assertionFinalize.reservedUal' }, ); const stampedReservedUal = reservedUalRes.type === 'bindings' && reservedUalRes.bindings[0]?.['u'] !== undefined @@ -4359,6 +4369,7 @@ export class PublishMethods extends DKGAgentBase { const assertionUri = contextGraphAssertionUri(contextGraphId, agentAddress, name, opts?.subGraphName); const metaResult = await this.store.query( `CONSTRUCT { <${assertionUri}> ?p ?o } WHERE { GRAPH <${metaGraph}> { <${assertionUri}> ?p ?o } }`, + { source: 'agent.asyncVmPublish.seal' }, ); const metaQuads = metaResult.type === 'quads' ? metaResult.quads : []; const seal = parseAssertionSealQuads(metaQuads, assertionUri); @@ -5191,6 +5202,7 @@ export class PublishMethods extends DKGAgentBase { OPTIONAL { <${lifecycleUri}> <${VM_CURRENT_ASSERTION_PRED}> ?vm } OPTIONAL { <${lifecycleUri}> <${KA_ID_PRED}> ?kaNum } } } LIMIT 1`, + { source: 'agent.asyncVmPublish.lifecyclePointer' }, ); const stripLit = (v?: string) => v?.replace(/^"/, '').replace(/"(\^\^<[^>]+>)?$/, ''); const pointerRow = pointerRes.type === 'bindings' ? pointerRes.bindings[0] : undefined; @@ -5590,6 +5602,7 @@ export class PublishMethods extends DKGAgentBase { // 1. Read the seal from _meta. const metaResult = await this.store.query( `CONSTRUCT { <${assertionUri}> ?p ?o } WHERE { GRAPH <${metaGraph}> { <${assertionUri}> ?p ?o } }`, + { source: 'agent.vmPublish.seal' }, ); const metaQuads = metaResult.type === 'quads' ? metaResult.quads : []; const seal = parseAssertionSealQuads(metaQuads, assertionUri); @@ -5693,6 +5706,7 @@ export class PublishMethods extends DKGAgentBase { OPTIONAL { <${lifecycleUri}> <${VM_CURRENT_ASSERTION_PRED}> ?vm } OPTIONAL { <${lifecycleUri}> <${KA_ID_PRED}> ?kaNum } } } LIMIT 1`, + { source: 'agent.vmPublish.lifecyclePointer' }, ); const stripLit = (v?: string) => v?.replace(/^"/, '').replace(/"(\^\^<[^>]+>)?$/, ''); const pointerRow = pointerRes.type === 'bindings' ? pointerRes.bindings[0] : undefined; @@ -6291,6 +6305,7 @@ export class PublishMethods extends DKGAgentBase { try { const res = await this.store.query( `SELECT ?vm WHERE { GRAPH <${metaGraph}> { <${lifecycleUri}> <${VM_CURRENT_ASSERTION_PRED}> ?vm } } LIMIT 1`, + { source: 'agent.publish.pointerVmGuard' }, ); const raw = res.type === 'bindings' ? res.bindings[0]?.['vm'] : undefined; vmBare = raw?.replace(/^"/, '').replace(/"(\^\^<[^>]+>)?$/, ''); @@ -6328,6 +6343,7 @@ export class PublishMethods extends DKGAgentBase { const lifecycleUri = assertionLifecycleUri(contextGraphId, agentAddress, name, subGraphName); const metaResult = await this.store.query( `CONSTRUCT { <${assertionUri}> ?p ?o } WHERE { GRAPH <${metaGraph}> { <${assertionUri}> ?p ?o } }`, + { source: 'agent.publish.swmPointerSeal' }, ); const metaQuads = metaResult.type === 'quads' ? metaResult.quads : []; const seal = parseAssertionSealQuads(metaQuads, assertionUri); diff --git a/packages/agent/src/dkg-agent-query.ts b/packages/agent/src/dkg-agent-query.ts index 927a907230..ba336a9577 100644 --- a/packages/agent/src/dkg-agent-query.ts +++ b/packages/agent/src/dkg-agent-query.ts @@ -773,6 +773,7 @@ export class QueryMethods extends DKGAgentBase { ?cg <${DKG_ONTOLOGY.DKG_ACCESS_POLICY}> "private" } }`, + { source: 'agent.query.privateGraphAccessPolicy' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return []; diff --git a/packages/agent/src/dkg-agent-swm-host.ts b/packages/agent/src/dkg-agent-swm-host.ts index 7f509f5dec..bfbcac3441 100644 --- a/packages/agent/src/dkg-agent-swm-host.ts +++ b/packages/agent/src/dkg-agent-swm-host.ts @@ -1015,7 +1015,11 @@ export class SwmHostModeMethods extends DKGAgentBase { const inflight = (async () => { try { const graphManager = new GraphManager(this.store); - const knownCgs = (await graphManager.listContextGraphs()).sort(); + const knownCgs = ( + await graphManager.listContextGraphs({ + source: 'agent.swmHostMode.listContextGraphs', + }) + ).sort(); if (knownCgs.length === 0) { this.hostModeReconcileCursor = 0; return; @@ -2627,7 +2631,9 @@ export class SwmHostModeMethods extends DKGAgentBase { if (!sub.subscribed || sub.onChainId) return null; let resolved: string | null = null; try { - resolved = await this.getContextGraphOnChainId(localCgId); + resolved = await this.getContextGraphOnChainId(localCgId, { + source: 'agent.vmReconcile.resolveOnChainId', + }); } catch { return null; } @@ -2982,6 +2988,7 @@ export class SwmHostModeMethods extends DKGAgentBase { GRAPH <${legacyMeta}> { ?ual <${DKG}batchId> ?b } FILTER NOT EXISTS { GRAPH <${scopedMeta}> { ?ual <${DKG}batchId> ?b } } }`, + { source: 'agent.swm.rsHeal.findLegacyOnly' }, ); if (askGuard.type !== 'boolean' || !askGuard.value) return; @@ -2991,6 +2998,7 @@ export class SwmHostModeMethods extends DKGAgentBase { GRAPH <${legacyMeta}> { ?ual <${DKG}batchId> ?b } FILTER NOT EXISTS { GRAPH <${scopedMeta}> { ?ual <${DKG}batchId> ?b } } }`, + { source: 'agent.swm.rsHeal.listLegacyOnly' }, ); if (stranded.type !== 'bindings') return; @@ -3043,6 +3051,7 @@ export class SwmHostModeMethods extends DKGAgentBase { { <${ual}> <${DKG}rootEntity> ?root . } } }`, + { source: 'agent.swm.rsHeal.readRoots' }, ); if (rootsRes.type !== 'bindings') return; const roots: string[] = []; @@ -3076,6 +3085,7 @@ export class SwmHostModeMethods extends DKGAgentBase { } } }`, + { source: 'agent.swm.rsHeal.checkRootData' }, ); if (present.type !== 'boolean' || !present.value) return; } @@ -3246,12 +3256,15 @@ export class SwmHostModeMethods extends DKGAgentBase { for (const namespace of candidateNamespaces) { const metaGraph = assertSafeIri(namespace.metaGraph); const dataGraph = assertSafeIri(namespace.dataGraph); - const operationRows = await this.store.query(`SELECT ?op ?root ?ts WHERE { - GRAPH <${metaGraph}> { - ?op ?root . - OPTIONAL { ?op ?ts . } - } - } ORDER BY ?op ?root ?ts LIMIT ${maxRows + 1}`); + const operationRows = await this.store.query( + `SELECT ?op ?root ?ts WHERE { + GRAPH <${metaGraph}> { + ?op ?root . + OPTIONAL { ?op ?ts . } + } + } ORDER BY ?op ?root ?ts LIMIT ${maxRows + 1}`, + { source: 'agent.vmReconcile.swmFingerprint.operations' }, + ); if (operationRows.type !== 'bindings') return null; if (isTooLarge(operationRows.bindings)) return null; const operations = operationRows.bindings @@ -3262,9 +3275,12 @@ export class SwmHostModeMethods extends DKGAgentBase { ].join('\0')) .sort(); - const dataRows = await this.store.query(`SELECT ?s ?p ?o WHERE { - GRAPH <${dataGraph}> { ?s ?p ?o . } - } ORDER BY ?s ?p ?o LIMIT ${maxRows + 1}`); + const dataRows = await this.store.query( + `SELECT ?s ?p ?o WHERE { + GRAPH <${dataGraph}> { ?s ?p ?o . } + } ORDER BY ?s ?p ?o LIMIT ${maxRows + 1}`, + { source: 'agent.vmReconcile.swmFingerprint.data' }, + ); if (dataRows.type !== 'bindings') return null; if (isTooLarge(dataRows.bindings)) return null; const dataTriples = dataRows.bindings @@ -3275,9 +3291,14 @@ export class SwmHostModeMethods extends DKGAgentBase { ].join('\0')) .sort(); - const privateRootRows = await this.store.query(`SELECT ?privateEntity ?privateRoot WHERE { - GRAPH <${metaGraph}> { ?privateEntity ?privateRoot . } - } ORDER BY ?privateEntity ?privateRoot LIMIT ${maxRows + 1}`); + const privateRootRows = await this.store.query( + `SELECT ?privateEntity ?privateRoot WHERE { + GRAPH <${metaGraph}> { + ?privateEntity ?privateRoot . + } + } ORDER BY ?privateEntity ?privateRoot LIMIT ${maxRows + 1}`, + { source: 'agent.vmReconcile.swmFingerprint.privateRoots' }, + ); if (privateRootRows.type !== 'bindings') return null; if (isTooLarge(privateRootRows.bindings)) return null; const privateRoots = privateRootRows.bindings diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 97e763d975..3a19fde1ed 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -1294,14 +1294,17 @@ export class DKGAgent extends DKGAgentBase { } }; - const ontologyResult = await this.store.query(` - SELECT ?ctxGraph ?name WHERE { - GRAPH <${ontologyGraph}> { - ?ctxGraph <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } + const ontologyResult = await this.store.query( + ` + SELECT ?ctxGraph ?name WHERE { + GRAPH <${ontologyGraph}> { + ?ctxGraph <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } + } } - } - `); + `, + { source: 'agent.contextGraph.discovery.ontologyDefinitions' }, + ); if (ontologyResult.type === 'bindings') { collectEntries(ontologyResult.bindings as Record[], 'ontology'); } @@ -1310,27 +1313,33 @@ export class DKGAgent extends DKGAgentBase { // persist a complete rdf:type/name definition. Read those binding-only // rows as catalogue entries so an edge restart remains useful while the // chain RPC is unavailable. - const onChainBindingResult = await this.store.query(` - SELECT ?ctxGraph ?name ?onChainId WHERE { - GRAPH <${ontologyGraph}> { - ?ctxGraph <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainId> ?onChainId . - OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } + const onChainBindingResult = await this.store.query( + ` + SELECT ?ctxGraph ?name ?onChainId WHERE { + GRAPH <${ontologyGraph}> { + ?ctxGraph <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainId> ?onChainId . + OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } + } } - } - `); + `, + { source: 'agent.contextGraph.discovery.onChainBindings' }, + ); if (onChainBindingResult.type === 'bindings') { collectEntries(onChainBindingResult.bindings as Record[], 'ontology'); } - const metaResult = await this.store.query(` - SELECT ?ctxGraph ?name WHERE { - GRAPH ?metaGraph { - ?ctxGraph <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . - OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } - FILTER(STRENDS(STR(?metaGraph), "/_meta")) + const metaResult = await this.store.query( + ` + SELECT ?ctxGraph ?name WHERE { + GRAPH ?metaGraph { + ?ctxGraph <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}> . + OPTIONAL { ?ctxGraph <${DKG_ONTOLOGY.SCHEMA_NAME}> ?name } + FILTER(STRENDS(STR(?metaGraph), "/_meta")) + } } - } - `); + `, + { source: 'agent.contextGraph.discovery.metaDefinitions' }, + ); if (metaResult.type === 'bindings') { collectEntries(metaResult.bindings as Record[], 'meta'); } @@ -1471,6 +1480,7 @@ export class DKGAgent extends DKGAgentBase { const contextGraphUri = contextGraphDataGraphUri(contextGraphId); const result = await this.store.query( `SELECT ?id WHERE { GRAPH <${ontologyGraph}> { <${contextGraphUri}> <${DKG_ONTOLOGY.DKG_CONTEXT_GRAPH}OnChainId> ?id } } LIMIT 1`, + { source: 'agent.contextGraph.chainDiscovery.durableOnChainId' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return null; const value = result.bindings[0]?.['id']; @@ -1865,6 +1875,7 @@ export class DKGAgent extends DKGAgentBase { ?network <${DKG_ONTOLOGY.RDF_TYPE}> <${DKG_ONTOLOGY.DKG_NETWORK}> . ?network <${DKG_ONTOLOGY.DKG_GENESIS_VERSION}> ?v . }`, + { source: 'agent.genesis.existingNetwork' }, ); if (existingGenesis.type === 'bindings') { const existingSubjects = existingGenesis.bindings @@ -2802,6 +2813,7 @@ export class DKGAgent extends DKGAgentBase { OPTIONAL { <${candidateLifecycleUri}> <${DKG_NS}contentScopeVersion> ?contentScopeVersion } } } LIMIT 1`, + { source: 'agent.history.lifecycleState' }, ); if (entityResult.type === 'bindings' && entityResult.bindings.length > 0) { lifecycleUri = candidateLifecycleUri; @@ -2863,6 +2875,7 @@ export class DKGAgent extends DKGAgentBase { OPTIONAL { ?event <${DKG_NS}rootEntity> ?rootEntity } } } ORDER BY ?timestamp`, + { source: 'agent.history.lifecycleEvents' }, ); // Member entities on the STABLE lifecycle subject (SUBSTRATE-1 stamp, @@ -2873,6 +2886,7 @@ export class DKGAgent extends DKGAgentBase { `SELECT DISTINCT ?root WHERE { GRAPH <${metaGraph}> { <${lifecycleUri}> ${ENTITY_PRED_ALT} ?root } }`, + { source: 'agent.history.lifecycleRoots' }, ); if (subjectRootsResult.type === 'bindings') { for (const b of subjectRootsResult.bindings) { @@ -2985,6 +2999,7 @@ export class DKGAgent extends DKGAgentBase { OPTIONAL { ?lifecycle <${PROV_NS}wasAttributedTo> ?author } } }`, + { source: 'agent.history.resolveByKaId' }, ); if (res.type !== 'bindings' || res.bindings.length === 0) return null; diff --git a/packages/agent/src/finalization-handler.ts b/packages/agent/src/finalization-handler.ts index a3f1cc5c7a..dba281c7cb 100644 --- a/packages/agent/src/finalization-handler.ts +++ b/packages/agent/src/finalization-handler.ts @@ -1307,6 +1307,7 @@ export class FinalizationHandler { ); const result = await this.store.query( `CONSTRUCT { ?s ?p ?o } WHERE { GRAPH <${assertSafeIri(graphUri)}> { ?s ?p ?o } }`, + { source: 'agent.finalization.verifyExactLayer' }, ); const quads = result.type === 'quads' ? result.quads.map((quad) => ({ ...quad, graph: '' })) @@ -1946,6 +1947,7 @@ export class FinalizationHandler { <${DKG_NS}contentScopeVersion> ?scope . } }`, + { source: 'agent.finalization.confirmedAssertionVersion' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return undefined; const expectedRoot = normalizedHex(ethers.hexlify(input.merkleRoot)); @@ -2014,6 +2016,7 @@ export class FinalizationHandler { `SELECT ?predicate ?object WHERE { GRAPH <${metaGraph}> { <${safeUal}> ?predicate ?object } }`, + { source: 'agent.finalization.graphScopedMetadataState' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return 'absent'; @@ -2147,6 +2150,7 @@ export class FinalizationHandler { UNION { <${safeUal}> <${DKG_NS}kaUal> ?value } UNION { <${safeUal}> <${DKG_NS}assertionGraph> ?value } } }`, + { source: 'agent.finalization.hasGraphScopedMetadata' }, ); return result.type === 'boolean' && result.value; } catch { @@ -2173,7 +2177,9 @@ export class FinalizationHandler { const ask = labelMetaGraph && labelMetaGraph !== metaGraph ? `ASK { { ${partitionPattern} } UNION { GRAPH <${assertSafeIri(labelMetaGraph)}> { <${safeUal}> "confirmed" } } }` : `ASK { ${partitionPattern} }`; - const result = await this.store.query(ask); + const result = await this.store.query(ask, { + source: 'agent.finalization.alreadyConfirmed', + }); return result.type === 'boolean' && result.value === true; } catch { return false; @@ -2311,6 +2317,7 @@ export class FinalizationHandler { const contextGraphUri = contextGraphDataUri(contextGraphId); const result = await this.store.query( `SELECT ?id WHERE { GRAPH <${ontologyGraph}> { <${contextGraphUri}> ?id } } LIMIT 1`, + { source: 'agent.finalization.contextGraphOnChainId' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return undefined; return stripOptionalLiteral(result.bindings[0]?.['id'])?.trim(); @@ -2975,12 +2982,15 @@ export class FinalizationHandler { ? `OPTIONAL { ?op <${SWM_SNAPSHOT_MERKLE_ROOT_PREDICATE}> ?memoRoot . } OPTIONAL { ?op <${SWM_SNAPSHOT_CONTENT_DIGEST_PREDICATE}> ?memoDigest . }` : ''; - const result = await this.store.query(`SELECT ?op ?root ?memoRoot ?memoDigest WHERE { - GRAPH <${assertSafeIri(wsMetaGraph)}> { - ?op <${DKG_NS}rootEntity> ?root . - ${memoPatterns} - } - }`); + const result = await this.store.query( + `SELECT ?op ?root ?memoRoot ?memoDigest WHERE { + GRAPH <${assertSafeIri(wsMetaGraph)}> { + ?op <${DKG_NS}rootEntity> ?root . + ${memoPatterns} + } + }`, + { source: 'agent.finalization.swmSnapshotCandidates' }, + ); if (result.type === 'bindings') { for (const row of result.bindings) { const op = typeof row['op'] === 'string' ? row['op'].replace(/^<(.*)>$/, '$1') : ''; @@ -3114,12 +3124,15 @@ export class FinalizationHandler { const targetHex = ethers.hexlify(merkleRoot); const rootsByOp = new Map(); try { - const result = await this.store.query(`SELECT ?op ?root WHERE { - GRAPH <${assertSafeIri(wsMetaGraph)}> { - ?op <${SWM_SNAPSHOT_MERKLE_ROOT_PREDICATE}> "${targetHex}" . - ?op <${DKG_NS}rootEntity> ?root . - } - }`); + const result = await this.store.query( + `SELECT ?op ?root WHERE { + GRAPH <${assertSafeIri(wsMetaGraph)}> { + ?op <${SWM_SNAPSHOT_MERKLE_ROOT_PREDICATE}> "${targetHex}" . + ?op <${DKG_NS}rootEntity> ?root . + } + }`, + { source: 'agent.finalization.stampedSwmSnapshot' }, + ); if (result.type === 'bindings') { for (const row of result.bindings) { const op = typeof row['op'] === 'string' ? row['op'].replace(/^<(.*)>$/, '$1') : ''; @@ -3299,6 +3312,7 @@ export class FinalizationHandler { ${JSON.stringify(subGraphName)} ; ?createdBy . } }`, + { source: 'agent.finalization.subGraphRegistration' }, ); if (alreadyRegistered.type !== 'boolean' || !alreadyRegistered.value) { const regQuads = generateSubGraphRegistration({ @@ -3653,6 +3667,7 @@ export class FinalizationHandler { private async deleteMetaForRoot(metaGraph: string, rootEntity: string): Promise { const result = await this.store.query( `SELECT DISTINCT ?op WHERE { GRAPH <${assertSafeIri(metaGraph)}> { ?op ${ENTITY_PRED_ALT} <${assertSafeIri(rootEntity)}> } }`, + { source: 'agent.finalization.rootMetadataOperations' }, ); if (result.type !== 'bindings') return; for (const row of result.bindings) { @@ -3666,6 +3681,7 @@ export class FinalizationHandler { ]); const remaining = await this.store.query( `SELECT (COUNT(DISTINCT ?r) AS ?c) WHERE { GRAPH <${assertSafeIri(metaGraph)}> { <${assertSafeIri(op)}> ${ENTITY_PRED_ALT} ?r } }`, + { source: 'agent.finalization.remainingOperationRoots' }, ); const rawCount = remaining.type === 'bindings' && remaining.bindings[0]?.['c']; const countVal = typeof rawCount === 'string' diff --git a/packages/agent/src/gossip-publish-handler.ts b/packages/agent/src/gossip-publish-handler.ts index 165a803aa2..69db636e4e 100644 --- a/packages/agent/src/gossip-publish-handler.ts +++ b/packages/agent/src/gossip-publish-handler.ts @@ -476,6 +476,7 @@ export class GossipPublishHandler { ${JSON.stringify(subGraphName)} ; ?createdBy . } }`, + { source: 'agent.gossipPublish.subGraphRegistration' }, ); if (alreadyRegistered.type !== 'boolean' || !alreadyRegistered.value) { const regQuads = generateSubGraphRegistration({ @@ -618,6 +619,7 @@ export class GossipPublishHandler { `ASK { GRAPH <${assertSafeIri(metaGraph)}> { ` + `<${assertSafeIri(graphPublish.scope.ual)}> ` + ` "confirmed" } }`, + { source: 'agent.gossipPublish.confirmedGuard' }, ); if (confirmed.type === 'boolean' && confirmed.value) { return 'stale' as const; @@ -986,6 +988,7 @@ export class GossipPublishHandler { const cgData = contextGraphDataGraphUri(contextGraphId); const result = await this.store.query( `SELECT ?peer WHERE { GRAPH <${cgMeta}> { <${cgData}> <${DKG_ONTOLOGY.DKG_ALLOWED_PEER}> ?peer } }`, + { source: 'agent.gossipPublish.allowedPeers' }, ); if (result.type !== 'bindings' || result.bindings.length === 0) return null; return result.bindings diff --git a/packages/agent/src/sync/backpressure.ts b/packages/agent/src/sync/backpressure.ts index 7013228a68..99ccc4b7c0 100644 --- a/packages/agent/src/sync/backpressure.ts +++ b/packages/agent/src/sync/backpressure.ts @@ -33,6 +33,23 @@ interface GlobalQueuePayload { contextGraphId?: string; } +export const DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT = 2; +export const DEFAULT_SYNC_GLOBAL_QUEUE_LIMIT_MULTIPLIER = 2; +export const DEFAULT_SYNC_PRIORITY_AGING_MS = 30_000; + +function syncOperationClass(label: string): string { + const operation = label.split(':', 1)[0]; + switch (operation) { + case 'durable': + case 'changelog': + case 'shared-memory': + case 'swm-recovery': + return operation; + default: + return 'sync'; + } +} + let inflight = 0; let lastLimit: number | null = null; let lastQueueLimit: number | null = null; @@ -48,12 +65,20 @@ const queue = new PriorityAdmissionQueue({ }; }, onDepthChange: (depth) => getMetrics().syncBackgroundQueueDepth.record(depth), + observability: { + scheduler: 'sync-global', + // Admission labels also carry CG/peer correlation identifiers. Collapse + // them to a fixed operation class before node-wide diagnostics/logging. + operation: (entry) => syncOperationClass(entry.payload.label), + inflightLimit: (entry) => entry.payload.limit, + thresholds: { + degradedQueueAgeMs: DEFAULT_SYNC_PRIORITY_AGING_MS / 2, + stalledActiveAgeMs: 120_000, + }, + register: true, + }, }); -export const DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT = 2; -export const DEFAULT_SYNC_GLOBAL_QUEUE_LIMIT_MULTIPLIER = 2; -export const DEFAULT_SYNC_PRIORITY_AGING_MS = 30_000; - export type SyncBackpressureBusyReason = 'queue_full' | 'displaced'; export class SyncBackpressureBusyError extends Error { diff --git a/packages/agent/src/sync/priority-admission-queue.ts b/packages/agent/src/sync/priority-admission-queue.ts index aba81eee4e..74d802f35a 100644 --- a/packages/agent/src/sync/priority-admission-queue.ts +++ b/packages/agent/src/sync/priority-admission-queue.ts @@ -1,4 +1,10 @@ -import { getMetrics } from '@origintrail-official/dkg-core'; +import { + backpressureRegistry, + getMetrics, + ObservableScheduler, + type SchedulerPressureThresholds, + type SchedulerPressureTicket, +} from '@origintrail-official/dkg-core'; import type { SyncPriorityClass, SyncSchedulerLane } from './policy.js'; export type PriorityAdmissionRelease = () => void; @@ -44,6 +50,14 @@ export interface PriorityAdmissionQueueHooks { canRun: (entry: PriorityAdmissionEntry) => boolean; onStart: (entry: PriorityAdmissionEntry) => PriorityAdmissionRelease; onDepthChange?: (depth: number) => void; + observability?: { + scheduler: string; + operation: (entry: PriorityAdmissionEntry) => string; + inflightLimit?: (entry: PriorityAdmissionEntry) => number | null; + thresholds?: SchedulerPressureThresholds; + now?: () => number; + register?: boolean; + }; } export interface PriorityAdmissionAcquireOptions extends PriorityAdmissionScheduling { @@ -85,12 +99,22 @@ function abortError(reason: unknown): Error { * Shared priority/FIFO admission queue. Scheduler decision metrics are events: * an aged start emits both `started` and `aged` intentionally. */ -export class PriorityAdmissionQueue { +export class PriorityAdmissionQueue extends ObservableScheduler { private readonly queue: InternalEntry[] = []; private readonly handoffReservations = new Map(); + private readonly pressureTickets = new WeakMap, SchedulerPressureTicket>(); + private readonly hooks: PriorityAdmissionQueueHooks; private nextSequence = 0; - constructor(private readonly hooks: PriorityAdmissionQueueHooks) {} + constructor(hooks: PriorityAdmissionQueueHooks) { + super({ + scheduler: hooks.observability?.scheduler ?? 'priority-admission', + thresholds: hooks.observability?.thresholds, + now: hooks.observability?.now, + }); + this.hooks = hooks; + if (hooks.observability?.register) backpressureRegistry.register(this); + } get length(): number { return this.queue.length; @@ -141,6 +165,12 @@ export class PriorityAdmissionQueue { now, agingThresholdMs: options.agingThresholdMs, }; + if (this.hooks.observability) { + this.updatePressureCapacity({ + queueLimit: options.queueLimit, + inflightLimit: this.hooks.observability.inflightLimit?.(base) ?? null, + }); + } const reservedGlobal = this.handoffReservations.size; const reservedOwner = this.countReservedOwner(ownerKey); @@ -162,6 +192,7 @@ export class PriorityAdmissionQueue { && !reservationOwnerFull ) { this.recordDecision(base, 'started'); + this.observePressureEnqueue(base); const admission: PriorityAdmission = { status: 'running', queuedBefore, @@ -187,9 +218,14 @@ export class PriorityAdmissionQueue { .sort((a, b) => a.priority - b.priority || b.sequence - a.sequence)[0]; if (!victim) { this.recordDecision(base, 'rejected'); + this.observePressureReject( + base, + globalFull ? 'global_queue_full' : 'owner_queue_full', + ); throw options.createBusyError(globalFull ? 'global_queue_full' : 'owner_queue_full'); } if (victim) { + this.observePressureReject(victim, 'displaced'); this.remove(victim); this.recordDecision(victim, 'displaced'); this.rejectOnce(victim, options.createDisplacedError(victim)); @@ -215,6 +251,7 @@ export class PriorityAdmissionQueue { if (options.timeoutMs !== undefined) { internal.timer = setTimeout(() => { if (!this.remove(internal)) return; + this.observePressureReject(internal, 'queue_wait_timeout'); this.rejectOnce( internal, options.createTimeoutError?.() ?? options.createBusyError('global_queue_full'), @@ -223,9 +260,11 @@ export class PriorityAdmissionQueue { } internal.onAbort = () => { if (!this.remove(internal)) return; + this.observePressureCancel(internal, 'aborted'); this.recordDecision(internal, 'aborted'); this.rejectOnce(internal, abortError(options.signal?.reason)); }; + this.observePressureEnqueue(internal); this.queue.push(internal); this.depthChanged(); if (options.signal) { @@ -292,6 +331,7 @@ export class PriorityAdmissionQueue { ownerQueueLimit: options.ownerQueueLimit, }); } + this.observePressureStart(entry); const release = this.hooks.onStart(entry); let released = false; return () => { @@ -299,6 +339,7 @@ export class PriorityAdmissionQueue { released = true; this.handoffReservations.delete(entry.sequence); release(); + this.observePressureFinish(entry); this.pump(); }; } @@ -357,4 +398,46 @@ export class PriorityAdmissionQueue { outcome, }); } + + private pressureWork(entry: PriorityAdmissionEntry) { + return { + lane: entry.lane, + operation: this.hooks.observability?.operation(entry) ?? entry.lane, + }; + } + + private observePressureEnqueue(entry: PriorityAdmissionEntry): void { + if (!this.hooks.observability) return; + this.pressureTickets.set(entry, this.pressureEnqueue(this.pressureWork(entry))); + } + + private observePressureStart(entry: PriorityAdmissionEntry): void { + const ticket = this.pressureTickets.get(entry); + if (ticket) this.pressureStart(ticket); + } + + private observePressureReject(entry: PriorityAdmissionEntry, reason: string): void { + if (!this.hooks.observability) return; + const ticket = this.pressureTickets.get(entry); + if (ticket) { + this.pressureRejectQueued(ticket, reason); + this.pressureTickets.delete(entry); + return; + } + this.pressureReject(this.pressureWork(entry), reason); + } + + private observePressureCancel(entry: PriorityAdmissionEntry, reason: string): void { + const ticket = this.pressureTickets.get(entry); + if (!ticket) return; + this.pressureCancelQueued(ticket, reason); + this.pressureTickets.delete(entry); + } + + private observePressureFinish(entry: PriorityAdmissionEntry): void { + const ticket = this.pressureTickets.get(entry); + if (!ticket) return; + this.pressureFinish(ticket, 'released'); + this.pressureTickets.delete(entry); + } } diff --git a/packages/agent/test/confirmed-meta-source-labels.test.ts b/packages/agent/test/confirmed-meta-source-labels.test.ts new file mode 100644 index 0000000000..c9957f40af --- /dev/null +++ b/packages/agent/test/confirmed-meta-source-labels.test.ts @@ -0,0 +1,34 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { TripleStore } from '@origintrail-official/dkg-storage'; +import { LifecycleSyncMethods } from '../src/dkg-agent-lifecycle.js'; + +describe('confirmed context-graph metadata source labels', () => { + it('attributes every direct store proof by its bounded provider step', async () => { + const query = vi.fn(async () => ({ + type: 'boolean', + value: false, + })); + const store = { query } as unknown as TripleStore; + const agent = { + store, + localApprovedAgentByCG: new Map(), + subscribedContextGraphs: new Map(), + isContextGraphPublicOnChain: vi.fn(async () => false), + isPrivateContextGraph: vi.fn(async () => false), + }; + + await expect( + LifecycleSyncMethods.prototype.hasConfirmedMetaState.call( + agent as never, + 'research', + ), + ).resolves.toBe(false); + + expect(query.mock.calls.map(([, options]) => options?.source)).toEqual([ + 'agent.contextGraph.confirmedMeta.unregisteredPlaceholder', + 'agent.contextGraph.confirmedMeta.privateDefinition', + 'agent.contextGraph.confirmedMeta.publicDefinition', + 'agent.contextGraph.confirmedMeta.ontologyDeclaration', + ]); + }); +}); diff --git a/packages/agent/test/context-graph-on-chain-id-source-labels.test.ts b/packages/agent/test/context-graph-on-chain-id-source-labels.test.ts new file mode 100644 index 0000000000..29fcead12c --- /dev/null +++ b/packages/agent/test/context-graph-on-chain-id-source-labels.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { TripleStore } from '@origintrail-official/dkg-storage'; +import { ContextGraphRegistryMethods } from '../src/dkg-agent-cg-registry.js'; + +function fixture() { + const query = vi.fn(async () => ({ + type: 'bindings', + bindings: [], + })); + return { + query, + agent: { + store: { query } as unknown as TripleStore, + subscribedContextGraphs: new Map(), + contextGraphWireId: (id: string) => id, + localCgIdForWireId: (id: string) => id, + }, + }; +} + +describe('context-graph on-chain id source labels', () => { + it('uses the caller-provided source label', async () => { + const { query, agent } = fixture(); + + await expect( + ContextGraphRegistryMethods.prototype.getContextGraphOnChainId.call( + agent as never, + 'research', + { source: 'agent.vmReconcile.resolveOnChainId' }, + ), + ).resolves.toBeNull(); + + expect(query.mock.calls[0]?.[1]?.source).toBe( + 'agent.vmReconcile.resolveOnChainId', + ); + }); + + it('uses a bounded provider label when the caller omits one', async () => { + const { query, agent } = fixture(); + + await expect( + ContextGraphRegistryMethods.prototype.getContextGraphOnChainId.call( + agent as never, + 'research', + ), + ).resolves.toBeNull(); + + expect(query.mock.calls[0]?.[1]?.source).toBe( + 'agent.contextGraph.onChainId', + ); + }); +}); diff --git a/packages/agent/test/query-source-coverage.test.ts b/packages/agent/test/query-source-coverage.test.ts new file mode 100644 index 0000000000..000e557afc --- /dev/null +++ b/packages/agent/test/query-source-coverage.test.ts @@ -0,0 +1,95 @@ +import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import * as ts from 'typescript'; +import { describe, expect, it } from 'vitest'; + +const TARGETS = [ + '../src/dkg-agent.ts', + '../src/dkg-agent-lifecycle.ts', + '../src/dkg-agent-cg-registry.ts', + '../src/dkg-agent-context-graph.ts', + '../src/dkg-agent-publish.ts', + '../src/finalization-handler.ts', +] as const; + +function propertyCarriesSource(property: ts.ObjectLiteralElementLike): boolean { + // A spread may carry a prepared QueryOptions.source that cannot be resolved + // locally. Keep those and non-literal option variables permissive; the guard + // is intended to reject only source-less option literals we can prove wrong. + if (ts.isSpreadAssignment(property)) return true; + const { name } = property; + if (!name) return false; + if (ts.isIdentifier(name) || ts.isStringLiteral(name)) return name.text === 'source'; + return ts.isComputedPropertyName(name) + && ts.isStringLiteral(name.expression) + && name.expression.text === 'source'; +} + +function missingSourceAttribution(sourceText: string, fileName: string): string[] { + const sourceFile = ts.createSourceFile( + fileName, + sourceText, + ts.ScriptTarget.Latest, + true, + ts.ScriptKind.TS, + ); + const missing: string[] = []; + + const visit = (node: ts.Node): void => { + if ( + ts.isCallExpression(node) + && ts.isPropertyAccessExpression(node.expression) + && node.expression.name.text === 'query' + ) { + const receiver = node.expression.expression.getText(sourceFile); + if (/(?:^|\.)store\??$/.test(receiver)) { + const options = node.arguments[1]; + const literalMissingSource = options !== undefined + && ts.isObjectLiteralExpression(options) + && !options.properties.some(propertyCarriesSource); + if (options === undefined || literalMissingSource) { + const position = sourceFile.getLineAndCharacterOfPosition(node.getStart(sourceFile)); + missing.push(`${fileName}:${position.line + 1}`); + } + } + } + ts.forEachChild(node, visit); + }; + + visit(sourceFile); + return missing; +} + +describe('lifecycle query source coverage', () => { + it('keeps direct store queries attributable in the profiled lifecycle paths', () => { + const missingAttribution: string[] = []; + + // Deliberately scoped to the lifecycle hot paths exercised by the scheduler + // profile. Add a file here when that profiled surface gains another provider. + for (const relativePath of TARGETS) { + const path = fileURLToPath(new URL(relativePath, import.meta.url)); + const sourceText = readFileSync(path, 'utf8'); + missingAttribution.push(...missingSourceAttribution(sourceText, relativePath)); + } + + expect(missingAttribution).toEqual([]); + }); + + it('rejects source-less literals without blocking prepared or spread options', () => { + const sourceText = [ + "store.query('missing-options');", + "store.query('empty-options', {});", + "store.query('other-options', { signal });", + "store.query('literal-source', { source: 'agent.test' });", + "store.query('shorthand-source', { source });", + "store.query('prepared-options', queryOptions);", + "store.query('spread-options', { ...queryOptions });", + ].join('\n'); + + expect(missingSourceAttribution(sourceText, 'fixture.ts')).toEqual([ + 'fixture.ts:1', + 'fixture.ts:2', + 'fixture.ts:3', + ]); + }); +}); diff --git a/packages/agent/test/query-source-labels.test.ts b/packages/agent/test/query-source-labels.test.ts new file mode 100644 index 0000000000..10a5ba5523 --- /dev/null +++ b/packages/agent/test/query-source-labels.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + GraphManager, + type TripleStore, +} from '@origintrail-official/dkg-storage'; +import { QueryMethods } from '../src/dkg-agent-query.js'; + +describe('query caller-provided store labels', () => { + it('attributes the unscoped private-graph access-policy lookup', async () => { + const query = vi.fn(async () => ({ + type: 'bindings', + bindings: [], + })); + const store = { query } as unknown as TripleStore; + + await expect( + QueryMethods.prototype.getDisallowedGraphPrefixes.call( + { store } as never, + ), + ).resolves.toEqual([]); + + expect(query).toHaveBeenCalledOnce(); + expect(query.mock.calls[0]?.[1]?.source).toBe( + 'agent.query.privateGraphAccessPolicy', + ); + }); + + it('forwards caller attribution through context-graph enumeration', async () => { + const listGraphsByPrefix = vi.fn(async () => []); + const store = { + listGraphsByPrefix, + } as unknown as TripleStore; + + await expect( + new GraphManager(store).listContextGraphs({ + source: 'agent.swmHostMode.listContextGraphs', + }), + ).resolves.toEqual([]); + + expect(listGraphsByPrefix).toHaveBeenCalledWith( + 'did:dkg:context-graph:', + { source: 'agent.swmHostMode.listContextGraphs' }, + ); + }); +}); diff --git a/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts b/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts index be1dbc61b3..4838dd39bc 100644 --- a/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts +++ b/packages/agent/test/rs-heal-stranded-kc-decorated.test.ts @@ -329,6 +329,42 @@ describe('healStrandedScopedKCs — through the production store decorator stack expect(metaInsert?.options).toMatchObject({ source: 'agent.swm.rsHeal.materialize', touchedGraphs: [scopedMeta] }); }); + it('labels RS-heal reads by caller operation through the decorator stack', async () => { + const querySources: Array = []; + const capturing = new Proxy(store, { + get(target, prop, receiver) { + if (prop === 'query') { + const orig = Reflect.get(target, prop, receiver) as TripleStore['query']; + return ( + sparql: Parameters[0], + options?: Parameters[1], + ) => { + querySources.push(options?.source); + return orig.call(target, sparql, options); + }; + } + return Reflect.get(target, prop, receiver); + }, + }) as TripleStore; + + await SwmHostModeMethods.prototype.healStrandedScopedKCs.call( + { + store: capturing, + log: { info: () => undefined, warn: () => undefined, error: () => undefined }, + } as never, + TEST_CG, + { subscribed: true, synced: true, onChainId: TEST_ONCHAIN } as never, + ); + + expect(new Set(querySources.filter((source) => source?.startsWith('agent.swm.rsHeal.')))) + .toEqual(new Set([ + 'agent.swm.rsHeal.findLegacyOnly', + 'agent.swm.rsHeal.listLegacyOnly', + 'agent.swm.rsHeal.readRoots', + 'agent.swm.rsHeal.checkRootData', + ])); + }); + it('relocates a VM-graph-only one-shot strand through the full stack (read-both)', async () => { // The publisher's own one-shot publish() lands public data in the per-KA VM // graph, not legacy root data. The read-both heal must recover it through the diff --git a/packages/agent/test/sync-backpressure.test.ts b/packages/agent/test/sync-backpressure.test.ts index 82feb62232..9c43aeaaa1 100644 --- a/packages/agent/test/sync-backpressure.test.ts +++ b/packages/agent/test/sync-backpressure.test.ts @@ -1,5 +1,8 @@ import { describe, expect, it } from 'vitest'; -import { createOperationContext } from '@origintrail-official/dkg-core'; +import { + backpressureRegistry, + createOperationContext, +} from '@origintrail-official/dkg-core'; import { getSyncBackpressureSnapshot, resolveBooleanSwitch, @@ -213,6 +216,51 @@ describe('sync global backpressure', () => { ]); }); + it('removes CG and peer correlation identifiers from node-wide pressure diagnostics', async () => { + const ctx = createOperationContext('sync'); + const policy = resolveSyncGlobalBackpressure({ + syncGlobalMaxInflight: 1, + syncGlobalQueueLimit: 1, + }); + let releaseRunning!: () => void; + const running = withGlobalSyncBackpressure( + { + policy, + ctx, + label: 'durable:urn:cg:private:peer-a', + }, + async () => new Promise((resolve) => { + releaseRunning = resolve; + }), + ); + await tick(); + const queued = withGlobalSyncBackpressure( + { + policy, + ctx, + label: 'swm-recovery:urn:cg:private:peer-b', + }, + async () => undefined, + ); + await tick(); + + const snapshot = backpressureRegistry.capture().schedulers.find( + (scheduler) => scheduler.scheduler === 'sync-global', + ); + expect(snapshot).toMatchObject({ + lanes: [expect.objectContaining({ + activeOperations: [expect.objectContaining({ operation: 'durable' })], + queuedOperations: [expect.objectContaining({ operation: 'swm-recovery' })], + })], + }); + expect(JSON.stringify(snapshot)).not.toContain('urn:cg:private'); + expect(JSON.stringify(snapshot)).not.toContain('peer-a'); + expect(JSON.stringify(snapshot)).not.toContain('peer-b'); + + releaseRunning(); + await Promise.all([running, queued]); + }); + it('starts a later elevated CG before an earlier deprioritized queued CG', async () => { const ctx = createOperationContext('sync'); const policy = resolveSyncGlobalBackpressure({ syncGlobalMaxInflight: 1, syncGlobalQueueLimit: 3 }); @@ -492,4 +540,66 @@ describe('sync global backpressure', () => { else process.env.DKG_STORAGE_ACK_HANDLER_DEADLINE_MS = oldDeadline; } }); + + it('projects priority admission through the generic pressure snapshot', async () => { + let running = 0; + let now = 1_000; + const queue = new PriorityAdmissionQueue({ + canRun: () => running < 1, + onStart: () => { + running += 1; + return () => { running -= 1; }; + }, + observability: { + scheduler: 'test-sync', + operation: (entry) => entry.payload, + inflightLimit: () => 1, + thresholds: { degradedQueueAgeMs: 5_000 }, + now: () => now, + }, + }); + const options = (payload: string) => ({ + payload, + lane: 'durable' as const, + priority: 0, + priorityClass: 'default' as const, + queueLimit: 2, + agingThresholdMs: 30_000, + now: () => now, + createBusyError: () => new Error('full'), + createDisplacedError: () => new Error('displaced'), + }); + + const first = queue.acquire(options('first')); + const releaseFirst = await first.release; + const second = queue.acquire(options('second')); + now += 6_000; + + expect(queue.getBackpressureSnapshot()).toMatchObject({ + scheduler: 'test-sync', + state: 'degraded', + totals: { + queued: 1, + queueLimit: 2, + inflight: 1, + inflightLimit: 1, + }, + lanes: [{ + lane: 'durable', + queuedOperations: [{ + operation: 'second', + count: 1, + oldestAgeMs: 6_000, + }], + }], + }); + + releaseFirst(); + const releaseSecond = await second.release; + releaseSecond(); + expect(queue.getBackpressureSnapshot()).toMatchObject({ + state: 'healthy', + totals: { queued: 0, inflight: 0 }, + }); + }); }); diff --git a/packages/agent/test/vm-reconcile-source-labels.test.ts b/packages/agent/test/vm-reconcile-source-labels.test.ts new file mode 100644 index 0000000000..090aa63037 --- /dev/null +++ b/packages/agent/test/vm-reconcile-source-labels.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { TripleStore } from '@origintrail-official/dkg-storage'; +import { SwmHostModeMethods } from '../src/dkg-agent-swm-host.js'; + +describe('VM reconcile caller-provided store labels', () => { + it('labels each SWM fingerprint read by its bounded operation class', async () => { + const query = vi.fn(async () => ({ + type: 'bindings', + bindings: [], + })); + const store = { query } as unknown as TripleStore; + + await expect( + SwmHostModeMethods.prototype.readVmReconcileSwmGen.call( + { store } as never, + [{ + metaGraph: 'did:dkg:context-graph:test/_shared_memory_meta', + dataGraph: 'did:dkg:context-graph:test/_shared_memory', + }], + ), + ).resolves.toContain('ops:0;'); + + expect(query.mock.calls.map(([, options]) => options?.source)).toEqual([ + 'agent.vmReconcile.swmFingerprint.operations', + 'agent.vmReconcile.swmFingerprint.data', + 'agent.vmReconcile.swmFingerprint.privateRoots', + ]); + }); +}); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index f0a78ba659..9f67418d6e 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -28,6 +28,10 @@ export default defineConfig({ "test/promote-async-default-agent.test.ts", "test/clear-promote-async-facade.test.ts", "test/query-min-trust-alias.test.ts", + "test/query-source-labels.test.ts", + "test/query-source-coverage.test.ts", + "test/context-graph-on-chain-id-source-labels.test.ts", + "test/confirmed-meta-source-labels.test.ts", "test/sync-envelope-cursor.test.ts", "test/exact-assets.test.ts", "test/exact-asset-responder.test.ts", @@ -36,6 +40,8 @@ export default defineConfig({ "test/swm/host-catchup-wire.test.ts", "test/swm/host-mode-store.test.ts", "test/swm/host-mode-key-canonicalization.test.ts", + "test/rs-heal-stranded-kc-decorated.test.ts", + "test/vm-reconcile-source-labels.test.ts", "test/profile-fix-verify.test.ts", "test/pca-v10-facade.test.ts", "test/ensure-registered-for-publish.test.ts", diff --git a/packages/cli/README.md b/packages/cli/README.md index 208e3578af..06339db833 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -401,6 +401,7 @@ When the daemon is running, it exposes a local HTTP API (default: `http://localh - `POST /api/endorse`, `POST /api/verify`, `POST /api/update` — Verifiable Memory trust operations - `GET /api/peers`, `GET /api/connections`, `GET /api/agents` — network introspection - `GET /api/wallets/balances`, `GET /api/chain/rpc-health` — wallet and chain health +- `GET /api/diagnostics/backpressure` — node-admin scheduler pressure snapshot - `GET /api/events` — Server-Sent Events stream for real-time notifications > The V9 `GET /api/apps` endpoint (and the `/apps/*` iframe host) was retired in @@ -409,6 +410,11 @@ When the daemon is running, it exposes a local HTTP API (default: `http://localh All endpoints (except public paths like `/api/status`, `/api/chain/rpc-health`, and `/.well-known/skill.md`) require an API token via `Authorization: Bearer ` header. +`GET /api/diagnostics/backpressure` specifically requires the node-level admin +token; agent-scoped tokens cannot inspect node-wide scheduler work. See the +[backpressure observability operator guide](../../docs/use-dkg/backpressure-observability.md) +for state definitions, log behavior, metrics, and response safety boundaries. + The full API surface — including request bodies, response shapes, and error codes — is documented in [`skills/dkg-node/SKILL.md`](./skills/dkg-node/SKILL.md). ## Local Benchmarks diff --git a/packages/cli/src/daemon/handle-request.ts b/packages/cli/src/daemon/handle-request.ts index ca31ae367d..54e6365506 100644 --- a/packages/cli/src/daemon/handle-request.ts +++ b/packages/cli/src/daemon/handle-request.ts @@ -312,6 +312,7 @@ import { } from './local-agents.js'; import type { MemoryGraphChangedEvent, NotificationSseEvent, RequestContext } from './routes/context.js'; import { handleStatusRoutes } from './routes/status.js'; +import { handleBackpressureRoutes } from './routes/backpressure.js'; import { handleAgentChatRoutes } from './routes/agent-chat.js'; import { handleOpenclawRoutes } from './routes/openclaw.js'; import { handleHermesRoutes } from './routes/hermes.js'; @@ -361,6 +362,9 @@ export async function handleRequest(input: HandleRequestInput): Promise { await handleStatusRoutes(ctx); if (res.writableEnded) return; + await handleBackpressureRoutes(ctx); + if (res.writableEnded) return; + await handleAgentChatRoutes(ctx); if (res.writableEnded) return; diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index 2d3be09e51..b0fae05696 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -70,7 +70,7 @@ import { } from '@origintrail-official/dkg-chain'; import { DKGAgent, loadOpWallets, KaNumberAllocator, resolveSyncAgentsMeta } from '@origintrail-official/dkg-agent'; import { isExternalBackend } from '@origintrail-official/dkg-storage'; -import { computeNetworkId, createOperationContext, createLogRedactor, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri, DEFAULT_PROTOCOL_OUTBOX_BACKOFFS_MS, DEFAULT_PROTOCOL_OUTBOX_MAX_AGE_MS, pickNetworkTunables, isKaPublishLifecycleDebugLoggingEnabled, setKaPublishLifecycleDebugLoggingEnabled, SYSTEM_CONTEXT_GRAPHS } from '@origintrail-official/dkg-core'; +import { BackpressureMonitor, computeNetworkId, createOperationContext, createLogRedactor, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri, DEFAULT_PROTOCOL_OUTBOX_BACKOFFS_MS, DEFAULT_PROTOCOL_OUTBOX_MAX_AGE_MS, pickNetworkTunables, isKaPublishLifecycleDebugLoggingEnabled, setKaPublishLifecycleDebugLoggingEnabled, SYSTEM_CONTEXT_GRAPHS } from '@origintrail-official/dkg-core'; import { DEFAULT_REQUIRED_ACKS, findReservedSubjectPrefix, @@ -1171,6 +1171,9 @@ export async function runDaemonInner( if (foreground) origStdoutWrite(line + "\n"); appendFile(logFile, line + "\n").catch(() => {}); } + const backpressureMonitor = new BackpressureMonitor({ + emit: (level, message) => log(`[${level}] ${message}`), + }); configureApiQueryPriority(process.env.DKG_API_QUERY_PRIORITY, { info: log, @@ -2524,7 +2527,9 @@ export async function runDaemonInner( } }, getContextGraphCount: async () => { - const graphUris = await agent.store.listGraphs(); + const graphUris = await agent.store.listGraphs({ + source: "daemon.metrics.graphInventory", + }); const knownContextGraphIds = new Set(); const subscribedContextGraphs = agent.getSubscribedContextGraphs(); const shadowContextGraphIds = new Set( @@ -2543,7 +2548,9 @@ export async function runDaemonInner( } const declarationQuery = buildContextGraphDeclarationsSparql(graphUris, knownContextGraphIds); const declarationResult = declarationQuery - ? await agent.store.query(declarationQuery) + ? await agent.store.query(declarationQuery, { + source: "daemon.metrics.contextGraphDeclarations", + }) : null; if (declarationResult?.type === "bindings") { for (const contextGraphId of contextGraphIdsFromDeclarationBindings( @@ -2566,7 +2573,9 @@ export async function runDaemonInner( // that are backed by local subscription/declaration state. // These COUNTs are cheap (~0.015 CPU-s/tick on a 75k-triple store). getTotalTriples: async () => { - const r = await agent.query(GET_TOTAL_TRIPLES_SPARQL); + const r = await agent.query(GET_TOTAL_TRIPLES_SPARQL, { + source: "daemon.metrics.totalTriples", + }); return parseRdfInt(r?.bindings?.[0]?.c); }, // RFC ka-metadata-trim (Phase 2 ⊕ / Phase 3 P3.1): the KC/KA counters are @@ -2582,6 +2591,7 @@ export async function runDaemonInner( getTotalKCs: async () => { const r = await agent.query( "SELECT (COUNT(DISTINCT ?kc) AS ?c) WHERE { GRAPH ?g { ?kc ?s } }", + { source: "daemon.metrics.totalKCs" }, ); return parseRdfInt(r?.bindings?.[0]?.c); }, @@ -2594,18 +2604,21 @@ export async function runDaemonInner( ?ka ?re . FILTER NOT EXISTS { ?tok ?ka } } } }`, + { source: "daemon.metrics.totalKAs" }, ); return parseRdfInt(r?.bindings?.[0]?.c); }, getConfirmedKCs: async () => { const r = await agent.query( 'SELECT (COUNT(DISTINCT ?kc) AS ?c) WHERE { GRAPH ?g { ?kc "confirmed" } }', + { source: "daemon.metrics.confirmedKCs" }, ); return parseRdfInt(r?.bindings?.[0]?.c); }, getTentativeKCs: async () => { const r = await agent.query( 'SELECT (COUNT(DISTINCT ?kc) AS ?c) WHERE { GRAPH ?g { ?kc "tentative" } }', + { source: "daemon.metrics.tentativeKCs" }, ); return parseRdfInt(r?.bindings?.[0]?.c); }, @@ -2864,6 +2877,7 @@ export async function runDaemonInner( log(`Telemetry: log exporter not started — ${r.error} (traces/metrics unaffected)`); } } + backpressureMonitor.start(); const PRUNE_INTERVAL_MS = 6 * 60 * 60_000; // 6 hours const pruneRuntimeState = async (): Promise => { @@ -3749,6 +3763,7 @@ export async function runDaemonInner( clearInterval(pingTimer); clearInterval(pruneTimer); logVolumePruner.stop(); + backpressureMonitor.stop(); // Clears the timer AND performs the final best-effort drain (BEFORE // telemetry stops), so a partial window still reaches Loki — keeps // log-derived request totals exact across process lifecycles. diff --git a/packages/cli/src/daemon/routes/backpressure.ts b/packages/cli/src/daemon/routes/backpressure.ts new file mode 100644 index 0000000000..ada7eecaef --- /dev/null +++ b/packages/cli/src/daemon/routes/backpressure.ts @@ -0,0 +1,42 @@ +import { backpressureRegistry } from '@origintrail-official/dkg-core'; +import { jsonResponse } from '../http-utils.js'; +import type { RequestContext } from './context.js'; + +/** + * Node-admin diagnostics for every registered scheduler/backlog source. + * + * The response contains bounded, sanitized operation labels and aggregate + * timings only. It never exposes work closures, request bodies, SPARQL text, + * graph identifiers, peer identifiers, or durable queue payloads. + */ +export async function handleBackpressureRoutes(ctx: RequestContext): Promise { + const { + req, + res, + agent, + config, + validTokens, + path, + requestToken, + } = ctx; + + if (req.method !== 'GET' || path !== '/api/diagnostics/backpressure') return; + + const authEnabled = config.auth?.enabled !== false; + const isNodeAdminCaller = + !authEnabled + || ( + !!requestToken + && validTokens.has(requestToken) + && !agent.resolveAgentByToken(requestToken) + ); + if (!isNodeAdminCaller) { + return jsonResponse(res, 403, { + error: + 'GET /api/diagnostics/backpressure requires a node-level admin token ' + + '(~/.dkg/auth.token); agent-scoped tokens cannot inspect node-wide scheduler work.', + }); + } + + return jsonResponse(res, 200, backpressureRegistry.capture()); +} diff --git a/packages/cli/src/daemon/routes/status.ts b/packages/cli/src/daemon/routes/status.ts index 32595b399b..538eb3b36f 100644 --- a/packages/cli/src/daemon/routes/status.ts +++ b/packages/cli/src/daemon/routes/status.ts @@ -59,7 +59,7 @@ import { enrichEvmError, MockChainAdapter, resolveRpcUrls, getRpcFailoverStats } import { DKGAgent, loadOpWallets } from '@origintrail-official/dkg-agent'; import { isExternalBackend } from '@origintrail-official/dkg-storage'; import { resolveManagedOxigraphPort } from '../oxigraph-managed.js'; -import { computeNetworkId, createOperationContext, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri } from '@origintrail-official/dkg-core'; +import { backpressureRegistry, computeNetworkId, createOperationContext, DKGEvent, Logger, PayloadTooLargeError, GET_VIEWS, TrustLevel, validateSubGraphName, validateAssertionName, validateContextGraphId, isSafeIri, assertSafeIri, sparqlIri, contextGraphSharedMemoryUri, contextGraphAssertionUri, contextGraphMetaUri } from '@origintrail-official/dkg-core'; import { findReservedSubjectPrefix, isSkolemizedUri } from '@origintrail-official/dkg-publisher'; import { DashboardDB, @@ -682,6 +682,7 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { ? getCachedExternalStoreQuads(agent, Date.now()) : peekCachedExternalStoreQuads() : null; + const backpressure = backpressureRegistry.capture(); // RFC-41 §4.9 + §4.3: expose build-info + installMode for // doctor / agent disambiguation. loadBuildInfo() falls back to // the {commit: "uncommitted", distTag: "monorepo", ...} @@ -763,6 +764,16 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { max: admission.max, rejectedTotal: admission.rejectedTotal, }, + // Public status carries state only. Detailed lane timings and operation + // summaries stay behind the node-admin diagnostics route. + backpressure: { + state: backpressure.state, + schedulers: backpressure.schedulers.map((scheduler) => ({ + scheduler: scheduler.scheduler, + state: scheduler.state, + })), + diagnosticsAvailable: '/api/diagnostics/backpressure', + }, connectedPeers: uniquePeers.size, connections: { total: allConns.length, diff --git a/packages/cli/test/backpressure-route.test.ts b/packages/cli/test/backpressure-route.test.ts new file mode 100644 index 0000000000..1fe42ee24f --- /dev/null +++ b/packages/cli/test/backpressure-route.test.ts @@ -0,0 +1,103 @@ +import { createServer, type Server } from 'node:http'; +import { afterEach, describe, expect, it } from 'vitest'; +import { + backpressureRegistry, + type BackpressureSource, +} from '@origintrail-official/dkg-core'; +import { handleBackpressureRoutes } from '../src/daemon/routes/backpressure.js'; + +describe('backpressure diagnostics route', () => { + let server: Server | undefined; + let unregister: (() => void) | undefined; + + afterEach(async () => { + unregister?.(); + unregister = undefined; + if (!server) return; + await new Promise((resolve, reject) => { + server!.close((error) => (error ? reject(error) : resolve())); + }); + server = undefined; + }); + + async function request(options: { + authEnabled?: boolean; + requestToken?: string; + tokenAgentAddress?: string; + } = {}): Promise<{ status: number; body: any }> { + const source: BackpressureSource = { + backpressureId: `route-test-${Date.now()}-${Math.random()}`, + getBackpressureSnapshot: () => ({ + scheduler: 'route-test', + state: 'degraded', + totals: { + queued: 3, + queueLimit: 4, + inflight: 1, + inflightLimit: 1, + oldestQueuedAgeMs: 6_000, + oldestActiveAgeMs: 1_000, + rejectedTotal: 0, + }, + lanes: [], + }), + }; + unregister = backpressureRegistry.register(source); + const agent = { + resolveAgentByToken: (token: string) => + token === options.requestToken ? options.tokenAgentAddress : undefined, + }; + server = createServer(async (req, res) => { + const url = new URL(req.url ?? '/', 'http://127.0.0.1'); + await handleBackpressureRoutes({ + req, + res, + agent, + config: { auth: { enabled: options.authEnabled ?? true } }, + validTokens: new Set(options.requestToken ? [options.requestToken] : []), + url, + path: url.pathname, + requestToken: options.requestToken, + } as any); + if (!res.writableEnded) { + res.statusCode = 404; + res.end(); + } + }); + await new Promise((resolve) => server!.listen(0, '127.0.0.1', resolve)); + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('route test server did not bind'); + const response = await fetch( + `http://127.0.0.1:${address.port}/api/diagnostics/backpressure`, + ); + return { status: response.status, body: await response.json() }; + } + + it('returns registered scheduler snapshots to a node admin', async () => { + const result = await request({ requestToken: 'node-admin-token' }); + + expect(result.status).toBe(200); + expect(result.body.capturedAt).toEqual(expect.any(String)); + expect(result.body.state).toBe('degraded'); + expect(result.body.schedulers).toContainEqual(expect.objectContaining({ + scheduler: 'route-test', + state: 'degraded', + totals: expect.objectContaining({ queued: 3, queueLimit: 4 }), + })); + }); + + it('rejects agent-scoped tokens', async () => { + const result = await request({ + requestToken: 'agent-token', + tokenAgentAddress: '0xagent', + }); + + expect(result.status).toBe(403); + expect(result.body.error).toMatch(/node-level admin token/i); + }); + + it('allows tokenless diagnostics when daemon auth is disabled', async () => { + const result = await request({ authEnabled: false }); + expect(result.status).toBe(200); + }); +}); diff --git a/packages/cli/vitest.unit.config.ts b/packages/cli/vitest.unit.config.ts index edb151458b..c1dee3b3fe 100644 --- a/packages/cli/vitest.unit.config.ts +++ b/packages/cli/vitest.unit.config.ts @@ -30,6 +30,7 @@ export default defineConfig({ 'test/reconcile-503-mapping.test.ts', 'test/config.test.ts', 'test/status-route-rpc.test.ts', + 'test/backpressure-route.test.ts', 'test/status-route-store-quads.test.ts', 'test/query-route-lifecycle.test.ts', 'test/status-command-store.test.ts', diff --git a/packages/core/src/backpressure-observability.ts b/packages/core/src/backpressure-observability.ts new file mode 100644 index 0000000000..f5e968afbf --- /dev/null +++ b/packages/core/src/backpressure-observability.ts @@ -0,0 +1,735 @@ +import { getMetrics } from './telemetry-api.js'; + +export type BackpressureState = 'healthy' | 'degraded' | 'saturated' | 'stalled'; +export type SchedulerPressureOutcome = 'completed' | 'failed' | 'cancelled' | 'released'; + +export interface SchedulerPressureCapacity { + queueLimit?: number | null; + inflightLimit?: number | null; + lanes?: Record; +} + +export interface SchedulerPressureThresholds { + /** Queue age that turns otherwise-low utilization into degraded pressure. */ + degradedQueueAgeMs?: number; + /** Active-work age that indicates an admitted operation may be stuck. */ + stalledActiveAgeMs?: number; + /** Fraction of a bounded queue that marks a lane as degraded. */ + degradedQueueUtilization?: number; + /** Keep a recent admission rejection visible as saturation for this long. */ + rejectionStateWindowMs?: number; +} + +export interface SchedulerPressureWork { + lane: string; + operation: string; +} + +export interface SchedulerPressureTicket { + readonly id: number; +} + +export interface BackpressureOperationSummary { + operation: string; + count: number; + oldestAgeMs: number; +} + +export interface BackpressureLaneSnapshot { + lane: string; + state: BackpressureState; + queued: number; + queueLimit: number | null; + inflight: number; + inflightLimit: number | null; + oldestQueuedAgeMs: number; + oldestActiveAgeMs: number; + queuedOperations: BackpressureOperationSummary[]; + activeOperations: BackpressureOperationSummary[]; + events: Record; + rejectedTotal: number; + rejectedByReason: Record; + /** Age of the latest rejection; safe for monotonic or epoch-based clocks. */ + lastRejectedAgeMs: number | null; +} + +export interface BackpressureSnapshot { + scheduler: string; + state: BackpressureState; + totals: { + queued: number; + queueLimit: number | null; + inflight: number; + inflightLimit: number | null; + oldestQueuedAgeMs: number; + oldestActiveAgeMs: number; + rejectedTotal: number; + }; + lanes: BackpressureLaneSnapshot[]; +} + +export interface BackpressureSource { + readonly backpressureId: string; + getBackpressureSnapshot(): BackpressureSnapshot; +} + +export interface BackpressureRegistrySnapshot { + capturedAt: string; + state: BackpressureState; + schedulers: BackpressureSnapshot[]; + failures: Array<{ scheduler: string; error: string }>; +} + +interface PressureWorkRecord { + id: number; + lane: string; + operation: string; + queuedAt: number; + startedAt?: number; +} + +interface LaneRuntime { + events: Map; + rejectedByReason: Map; + lastRejectedAt: number | null; +} + +const DEFAULT_DEGRADED_QUEUE_AGE_MS = 5_000; +const DEFAULT_STALLED_ACTIVE_AGE_MS = 30_000; +const DEFAULT_DEGRADED_QUEUE_UTILIZATION = 0.75; +const DEFAULT_REJECTION_STATE_WINDOW_MS = 60_000; +const MAX_OPERATION_SUMMARIES = 8; + +const STATE_RANK: Record = { + healthy: 0, + degraded: 1, + saturated: 2, + stalled: 3, +}; + +function maxState(a: BackpressureState, b: BackpressureState): BackpressureState { + return STATE_RANK[a] >= STATE_RANK[b] ? a : b; +} + +function normalizeLimit(value: number | null | undefined): number | null { + return Number.isFinite(value) && (value as number) >= 0 ? value as number : null; +} + +/** + * Keep metric/log dimensions bounded and strip payload-like punctuation. + * Callers should still pass static operation names rather than graph, peer, or + * job identifiers. + */ +export function normalizeBackpressureLabel(value: string, fallback = 'unknown'): string { + const trimmed = value.trim(); + if (!trimmed) return fallback; + return trimmed.replace(/[^\w:./-]/g, '_').slice(0, 80) || fallback; +} + +function mapToRecord(map: ReadonlyMap): Record { + return Object.fromEntries([...map.entries()].sort(([a], [b]) => a.localeCompare(b))); +} + +function operationSummaries( + records: Iterable, + timestamp: (record: PressureWorkRecord) => number, + now: number, +): BackpressureOperationSummary[] { + const byOperation = new Map(); + for (const record of records) { + const existing = byOperation.get(record.operation); + const at = timestamp(record); + if (existing) { + existing.count += 1; + existing.oldestAt = Math.min(existing.oldestAt, at); + } else { + byOperation.set(record.operation, { count: 1, oldestAt: at }); + } + } + return [...byOperation.entries()] + .map(([operation, value]) => ({ + operation, + count: value.count, + oldestAgeMs: Math.max(0, Math.floor(now - value.oldestAt)), + })) + .sort((a, b) => b.oldestAgeMs - a.oldestAgeMs || b.count - a.count || a.operation.localeCompare(b.operation)) + .slice(0, MAX_OPERATION_SUMMARIES); +} + +export interface SchedulerPressureTrackerOptions { + scheduler: string; + capacity?: SchedulerPressureCapacity; + thresholds?: SchedulerPressureThresholds; + now?: () => number; +} + +/** + * Scheduling-policy-neutral lifecycle tracker. + * + * Queue implementations call the five transition methods at their existing + * admission boundaries. The tracker owns timings, state classification, + * bounded metrics, and diagnostic snapshots, but never decides which work may + * run. All observability calls are fail-open so instrumentation cannot change + * scheduler behaviour. + */ +export class SchedulerPressureTracker { + readonly scheduler: string; + + private readonly queued = new Map(); + private readonly active = new Map(); + private readonly lanes = new Map(); + private readonly now: () => number; + private readonly thresholds: Required; + private capacity: SchedulerPressureCapacity; + private nextTicketId = 1; + + constructor(options: SchedulerPressureTrackerOptions) { + this.scheduler = normalizeBackpressureLabel(options.scheduler, 'scheduler'); + this.capacity = options.capacity ?? {}; + this.now = options.now ?? Date.now; + this.thresholds = { + degradedQueueAgeMs: + options.thresholds?.degradedQueueAgeMs ?? DEFAULT_DEGRADED_QUEUE_AGE_MS, + stalledActiveAgeMs: + options.thresholds?.stalledActiveAgeMs ?? DEFAULT_STALLED_ACTIVE_AGE_MS, + degradedQueueUtilization: + options.thresholds?.degradedQueueUtilization ?? DEFAULT_DEGRADED_QUEUE_UTILIZATION, + rejectionStateWindowMs: + options.thresholds?.rejectionStateWindowMs ?? DEFAULT_REJECTION_STATE_WINDOW_MS, + }; + } + + updateCapacity(capacity: SchedulerPressureCapacity): void { + this.capacity = capacity; + } + + enqueue(work: SchedulerPressureWork): SchedulerPressureTicket { + const record: PressureWorkRecord = { + id: this.nextTicketId++, + lane: normalizeBackpressureLabel(work.lane, 'default'), + operation: normalizeBackpressureLabel(work.operation), + queuedAt: this.now(), + }; + this.queued.set(record.id, record); + this.recordEvent(record.lane, 'enqueued'); + return { id: record.id }; + } + + start(ticket: SchedulerPressureTicket): void { + const record = this.queued.get(ticket.id); + if (!record) return; + this.queued.delete(ticket.id); + record.startedAt = this.now(); + this.active.set(ticket.id, record); + this.recordEvent(record.lane, 'started'); + this.safeMetric(() => getMetrics().backpressureQueueWaitMs.record( + Math.max(0, record.startedAt! - record.queuedAt), + { scheduler: this.scheduler, lane: record.lane }, + )); + } + + reject(work: SchedulerPressureWork, reason: string): void { + const lane = normalizeBackpressureLabel(work.lane, 'default'); + this.recordRejection(lane, reason); + } + + rejectQueued(ticket: SchedulerPressureTicket, reason: string): void { + const record = this.queued.get(ticket.id); + if (!record) return; + this.queued.delete(ticket.id); + this.recordRejection(record.lane, reason); + } + + cancelQueued(ticket: SchedulerPressureTicket, reason = 'cancelled'): void { + const record = this.queued.get(ticket.id); + if (!record) return; + this.queued.delete(ticket.id); + this.recordEvent(record.lane, 'cancelled', reason); + } + + finish(ticket: SchedulerPressureTicket, outcome: SchedulerPressureOutcome): void { + const record = this.active.get(ticket.id); + if (!record) return; + this.active.delete(ticket.id); + const finishedAt = this.now(); + this.recordEvent(record.lane, outcome); + this.safeMetric(() => getMetrics().backpressureActiveDurationMs.record( + Math.max(0, finishedAt - (record.startedAt ?? finishedAt)), + { + scheduler: this.scheduler, + lane: record.lane, + outcome, + }, + )); + } + + snapshot(): BackpressureSnapshot { + const now = this.now(); + const laneNames = new Set([ + ...this.lanes.keys(), + ...Object.keys(this.capacity.lanes ?? {}), + ...[...this.queued.values()].map((entry) => entry.lane), + ...[...this.active.values()].map((entry) => entry.lane), + ]); + const snapshots = [...laneNames] + .sort() + .map((lane) => this.laneSnapshot(lane, now)); + const totals = { + queued: snapshots.reduce((sum, lane) => sum + lane.queued, 0), + queueLimit: normalizeLimit(this.capacity.queueLimit) + ?? this.sumLaneLimits('queueLimit', snapshots), + inflight: snapshots.reduce((sum, lane) => sum + lane.inflight, 0), + inflightLimit: normalizeLimit(this.capacity.inflightLimit) + ?? this.sumLaneLimits('inflightLimit', snapshots), + oldestQueuedAgeMs: Math.max(0, ...snapshots.map((lane) => lane.oldestQueuedAgeMs)), + oldestActiveAgeMs: Math.max(0, ...snapshots.map((lane) => lane.oldestActiveAgeMs)), + rejectedTotal: snapshots.reduce((sum, lane) => sum + lane.rejectedTotal, 0), + }; + let state = snapshots.reduce( + (current, lane) => maxState(current, lane.state), + 'healthy', + ); + if ( + totals.queueLimit !== null + && totals.queueLimit > 0 + && totals.queued >= totals.queueLimit + ) { + state = maxState(state, 'saturated'); + } else if ( + totals.queueLimit !== null + && totals.queueLimit > 0 + && totals.queued / totals.queueLimit >= this.thresholds.degradedQueueUtilization + ) { + state = maxState(state, 'degraded'); + } + if (totals.oldestQueuedAgeMs >= this.thresholds.degradedQueueAgeMs) { + state = maxState(state, 'degraded'); + } + if (totals.oldestActiveAgeMs >= this.thresholds.stalledActiveAgeMs) { + state = maxState(state, 'stalled'); + } + return { + scheduler: this.scheduler, + state, + totals, + lanes: snapshots, + }; + } + + private runtimeFor(lane: string): LaneRuntime { + let runtime = this.lanes.get(lane); + if (!runtime) { + runtime = { + events: new Map(), + rejectedByReason: new Map(), + lastRejectedAt: null, + }; + this.lanes.set(lane, runtime); + } + return runtime; + } + + private recordEvent(lane: string, event: string, reason?: string): void { + const runtime = this.runtimeFor(lane); + runtime.events.set(event, (runtime.events.get(event) ?? 0) + 1); + this.safeMetric(() => getMetrics().backpressureEventsTotal.add(1, { + scheduler: this.scheduler, + lane, + event, + ...(reason ? { reason: normalizeBackpressureLabel(reason) } : {}), + })); + } + + private recordRejection(lane: string, reason: string): void { + const normalizedReason = normalizeBackpressureLabel(reason, 'rejected'); + const runtime = this.runtimeFor(lane); + runtime.rejectedByReason.set( + normalizedReason, + (runtime.rejectedByReason.get(normalizedReason) ?? 0) + 1, + ); + runtime.lastRejectedAt = this.now(); + this.recordEvent(lane, 'rejected', normalizedReason); + } + + private laneSnapshot(lane: string, now: number): BackpressureLaneSnapshot { + const runtime = this.runtimeFor(lane); + const queued = [...this.queued.values()].filter((entry) => entry.lane === lane); + const active = [...this.active.values()].filter((entry) => entry.lane === lane); + const queueLimit = normalizeLimit(this.capacity.lanes?.[lane]?.queueLimit); + const inflightLimit = normalizeLimit(this.capacity.lanes?.[lane]?.inflightLimit); + const oldestQueuedAgeMs = queued.length === 0 + ? 0 + : Math.max(...queued.map((entry) => Math.max(0, Math.floor(now - entry.queuedAt)))); + const oldestActiveAgeMs = active.length === 0 + ? 0 + : Math.max(...active.map((entry) => Math.max( + 0, + Math.floor(now - (entry.startedAt ?? now)), + ))); + let state: BackpressureState = 'healthy'; + if (oldestActiveAgeMs >= this.thresholds.stalledActiveAgeMs) { + state = 'stalled'; + } else if ( + (queueLimit !== null && queueLimit > 0 && queued.length >= queueLimit) + || ( + runtime.lastRejectedAt !== null + && now - runtime.lastRejectedAt <= this.thresholds.rejectionStateWindowMs + ) + ) { + state = 'saturated'; + } else if ( + oldestQueuedAgeMs >= this.thresholds.degradedQueueAgeMs + || ( + queueLimit !== null + && queueLimit > 0 + && queued.length / queueLimit >= this.thresholds.degradedQueueUtilization + ) + ) { + state = 'degraded'; + } + return { + lane, + state, + queued: queued.length, + queueLimit, + inflight: active.length, + inflightLimit, + oldestQueuedAgeMs, + oldestActiveAgeMs, + queuedOperations: operationSummaries(queued, (entry) => entry.queuedAt, now), + activeOperations: operationSummaries( + active, + (entry) => entry.startedAt ?? now, + now, + ), + events: mapToRecord(runtime.events), + rejectedTotal: [...runtime.rejectedByReason.values()].reduce((sum, value) => sum + value, 0), + rejectedByReason: mapToRecord(runtime.rejectedByReason), + lastRejectedAgeMs: runtime.lastRejectedAt === null + ? null + : Math.max(0, Math.floor(now - runtime.lastRejectedAt)), + }; + } + + private sumLaneLimits( + field: 'queueLimit' | 'inflightLimit', + lanes: BackpressureLaneSnapshot[], + ): number | null { + if (lanes.length === 0 || lanes.some((lane) => lane[field] === null)) return null; + return lanes.reduce((sum, lane) => sum + (lane[field] ?? 0), 0); + } + + private safeMetric(record: () => void): void { + try { + record(); + } catch { + // Metrics are strictly fail-open. + } + } +} + +/** + * Base class for in-memory schedulers. Subclasses retain complete ownership of + * queue ordering, admission, coalescing, and release semantics and call these + * protected lifecycle methods at their existing boundaries. + */ +export abstract class ObservableScheduler implements BackpressureSource { + protected readonly pressure: SchedulerPressureTracker; + + protected constructor(options: SchedulerPressureTrackerOptions) { + this.pressure = new SchedulerPressureTracker(options); + } + + get backpressureId(): string { + return this.pressure.scheduler; + } + + getBackpressureSnapshot(): BackpressureSnapshot { + return this.pressure.snapshot(); + } + + protected updatePressureCapacity(capacity: SchedulerPressureCapacity): void { + this.pressure.updateCapacity(capacity); + } + + protected pressureEnqueue(work: SchedulerPressureWork): SchedulerPressureTicket { + return this.pressure.enqueue(work); + } + + protected pressureStart(ticket: SchedulerPressureTicket): void { + this.pressure.start(ticket); + } + + protected pressureReject(work: SchedulerPressureWork, reason: string): void { + this.pressure.reject(work, reason); + } + + protected pressureRejectQueued(ticket: SchedulerPressureTicket, reason: string): void { + this.pressure.rejectQueued(ticket, reason); + } + + protected pressureCancelQueued(ticket: SchedulerPressureTicket, reason?: string): void { + this.pressure.cancelQueued(ticket, reason); + } + + protected pressureFinish( + ticket: SchedulerPressureTicket, + outcome: SchedulerPressureOutcome, + ): void { + this.pressure.finish(ticket, outcome); + } +} + +export class BackpressureRegistry { + private readonly sources = new Map(); + + register(source: BackpressureSource): () => void { + const id = normalizeBackpressureLabel(source.backpressureId, 'scheduler'); + const existing = this.sources.get(id); + if (existing && existing !== source) { + throw new Error(`Backpressure source "${id}" is already registered`); + } + this.sources.set(id, source); + return () => { + if (this.sources.get(id) === source) this.sources.delete(id); + }; + } + + capture(): BackpressureRegistrySnapshot { + const schedulers: BackpressureSnapshot[] = []; + const failures: Array<{ scheduler: string; error: string }> = []; + for (const [id, source] of [...this.sources.entries()].sort(([a], [b]) => a.localeCompare(b))) { + try { + schedulers.push(source.getBackpressureSnapshot()); + } catch (error) { + failures.push({ + scheduler: id, + error: error instanceof Error ? error.message : String(error), + }); + } + } + return { + capturedAt: new Date().toISOString(), + state: schedulers.reduce( + (current, scheduler) => maxState(current, scheduler.state), + 'healthy', + ), + schedulers, + failures, + }; + } +} + +export const backpressureRegistry = new BackpressureRegistry(); + +export function recordBackpressureSnapshotMetrics(snapshot: BackpressureSnapshot): void { + const metrics = getMetrics(); + const record = ( + lane: string, + values: { + queued: number; + queueLimit: number | null; + inflight: number; + inflightLimit: number | null; + oldestQueuedAgeMs: number; + oldestActiveAgeMs: number; + }, + ) => { + const attributes = { scheduler: snapshot.scheduler, lane }; + metrics.backpressureQueueDepth.record(values.queued, attributes); + metrics.backpressureInflight.record(values.inflight, attributes); + metrics.backpressureOldestQueuedAgeMs.record(values.oldestQueuedAgeMs, attributes); + metrics.backpressureOldestActiveAgeMs.record(values.oldestActiveAgeMs, attributes); + if (values.queueLimit !== null) { + metrics.backpressureQueueLimit.record(values.queueLimit, attributes); + } + if (values.inflightLimit !== null) { + metrics.backpressureInflightLimit.record(values.inflightLimit, attributes); + } + }; + record('all', { + queued: snapshot.totals.queued, + queueLimit: snapshot.totals.queueLimit, + inflight: snapshot.totals.inflight, + inflightLimit: snapshot.totals.inflightLimit, + oldestQueuedAgeMs: snapshot.totals.oldestQueuedAgeMs, + oldestActiveAgeMs: snapshot.totals.oldestActiveAgeMs, + }); + for (const lane of snapshot.lanes) record(lane.lane, lane); +} + +export interface BackpressureMonitorOptions { + registry?: BackpressureRegistry; + intervalMs?: number; + summaryIntervalMs?: number; + now?: () => number; + emit: ( + level: 'info' | 'warn', + message: string, + snapshot: BackpressureSnapshot, + lane: BackpressureLaneSnapshot | null, + ) => void; +} + +interface LoggedPressureState { + state: BackpressureState; + lastLoggedAt: number; +} + +/** + * One process-wide sampler provides transition/recovery logging and periodic + * summaries. It logs state changes rather than individual queue operations. + */ +export class BackpressureMonitor { + private readonly registry: BackpressureRegistry; + private readonly intervalMs: number; + private readonly summaryIntervalMs: number; + private readonly now: () => number; + private readonly emit: BackpressureMonitorOptions['emit']; + private readonly logged = new Map(); + private timer: ReturnType | null = null; + + constructor(options: BackpressureMonitorOptions) { + this.registry = options.registry ?? backpressureRegistry; + this.intervalMs = Math.max(1_000, options.intervalMs ?? 5_000); + this.summaryIntervalMs = Math.max(this.intervalMs, options.summaryIntervalMs ?? 60_000); + this.now = options.now ?? Date.now; + this.emit = options.emit; + } + + start(): void { + if (this.timer) return; + this.sample(); + this.timer = setInterval(() => this.sample(), this.intervalMs); + if (typeof this.timer.unref === 'function') this.timer.unref(); + } + + stop(): void { + if (!this.timer) return; + clearInterval(this.timer); + this.timer = null; + } + + sample(): void { + const captured = this.registry.capture(); + const now = this.now(); + for (const scheduler of captured.schedulers) { + try { + recordBackpressureSnapshotMetrics(scheduler); + } catch { + // The monitor must keep logging even if a metrics provider misbehaves. + } + const worstLaneState = scheduler.lanes.reduce( + (state, lane) => maxState(state, lane.state), + 'healthy', + ); + const samples: Array<{ key: string; state: BackpressureState; lane: BackpressureLaneSnapshot | null }> = + scheduler.lanes.map((lane) => ({ + key: `${scheduler.scheduler}/${lane.lane}`, + state: lane.state, + lane, + })); + if (STATE_RANK[scheduler.state] > STATE_RANK[worstLaneState]) { + samples.push({ + key: `${scheduler.scheduler}/all`, + state: scheduler.state, + lane: null, + }); + } + for (const sample of samples) { + this.observeSample(sample.key, sample.state, scheduler, sample.lane, now); + } + } + } + + private observeSample( + key: string, + state: BackpressureState, + scheduler: BackpressureSnapshot, + lane: BackpressureLaneSnapshot | null, + now: number, + ): void { + const previous = this.logged.get(key); + if (state === 'healthy') { + if (previous && previous.state !== 'healthy') { + this.safeEmit( + 'info', + this.message('recovered', scheduler, lane, state, previous.state), + scheduler, + lane, + ); + } + this.logged.set(key, { state, lastLoggedAt: previous?.lastLoggedAt ?? now }); + return; + } + + const transition = !previous || previous.state !== state; + const summaryDue = !previous || now - previous.lastLoggedAt >= this.summaryIntervalMs; + if (transition || summaryDue) { + this.safeEmit( + 'warn', + this.message(transition ? 'transition' : 'summary', scheduler, lane, state, previous?.state), + scheduler, + lane, + ); + this.logged.set(key, { state, lastLoggedAt: now }); + } + } + + private message( + event: 'transition' | 'summary' | 'recovered', + scheduler: BackpressureSnapshot, + lane: BackpressureLaneSnapshot | null, + state: BackpressureState, + previousState?: BackpressureState, + ): string { + const values = lane ?? { + lane: 'all', + queued: scheduler.totals.queued, + queueLimit: scheduler.totals.queueLimit, + inflight: scheduler.totals.inflight, + inflightLimit: scheduler.totals.inflightLimit, + oldestQueuedAgeMs: scheduler.totals.oldestQueuedAgeMs, + oldestActiveAgeMs: scheduler.totals.oldestActiveAgeMs, + rejectedTotal: scheduler.totals.rejectedTotal, + queuedOperations: scheduler.lanes.flatMap((item) => item.queuedOperations) + .sort((a, b) => b.oldestAgeMs - a.oldestAgeMs) + .slice(0, MAX_OPERATION_SUMMARIES), + activeOperations: scheduler.lanes.flatMap((item) => item.activeOperations) + .sort((a, b) => b.oldestAgeMs - a.oldestAgeMs) + .slice(0, MAX_OPERATION_SUMMARIES), + }; + return `[backpressure] ${JSON.stringify({ + event, + scheduler: scheduler.scheduler, + lane: values.lane, + state, + ...(previousState ? { previousState } : {}), + queued: values.queued, + queueLimit: values.queueLimit, + inflight: values.inflight, + inflightLimit: values.inflightLimit, + oldestQueuedAgeMs: values.oldestQueuedAgeMs, + oldestActiveAgeMs: values.oldestActiveAgeMs, + rejectedTotal: values.rejectedTotal, + queuedOperations: values.queuedOperations, + activeOperations: values.activeOperations, + })}`; + } + + private safeEmit( + level: 'info' | 'warn', + message: string, + scheduler: BackpressureSnapshot, + lane: BackpressureLaneSnapshot | null, + ): void { + try { + this.emit(level, message, scheduler, lane); + } catch { + // Logging must never break scheduling or the monitor loop. + } + } +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 03a8344e8d..9cf33c79ff 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -68,6 +68,7 @@ export * from './author-catalog-codec.js'; export * from './author-catalog-objects.js'; export * from './author-catalog-directory.js'; export * from './event-bus.js'; +export * from './backpressure-observability.js'; export { Logger, createOperationContext, diff --git a/packages/core/src/telemetry-api.ts b/packages/core/src/telemetry-api.ts index 7c3983c573..ebaa916ed3 100644 --- a/packages/core/src/telemetry-api.ts +++ b/packages/core/src/telemetry-api.ts @@ -183,6 +183,24 @@ export interface DkgMetrics { storeSchedulerQueueWaitMs: Histogram; /** active admitted operations by priority and operation */ storeSchedulerActive: UpDownCounter; + /** current queued work; scheduler and lane are bounded static labels */ + backpressureQueueDepth: Gauge; + /** configured queue capacity; scheduler and lane are bounded static labels */ + backpressureQueueLimit: Gauge; + /** current admitted work; scheduler and lane are bounded static labels */ + backpressureInflight: Gauge; + /** configured concurrent-work capacity; scheduler and lane are bounded static labels */ + backpressureInflightLimit: Gauge; + /** age of the oldest queued item; scheduler and lane are bounded static labels */ + backpressureOldestQueuedAgeMs: Gauge; + /** age of the oldest admitted item; scheduler and lane are bounded static labels */ + backpressureOldestActiveAgeMs: Gauge; + /** scheduler, lane, event, and optional reason are bounded labels */ + backpressureEventsTotal: Counter; + /** queue wait by bounded scheduler and lane */ + backpressureQueueWaitMs: Histogram; + /** admitted-work duration by bounded scheduler, lane, and outcome */ + backpressureActiveDurationMs: Histogram; /** scope={finalization|reconcile} */ storeScanSingleFlightJoinsTotal: Counter; /** active unique scans by scope */ @@ -307,6 +325,39 @@ function buildMetrics(): DkgMetrics { storeSchedulerActive: meter.createUpDownCounter('dkg.store.scheduler_active', { description: 'Currently admitted external-store operations by lane and static caller', }), + backpressureQueueDepth: meter.createGauge('dkg.backpressure.queue_depth', { + description: 'Current scheduler queue depth by bounded scheduler and lane', + }), + backpressureQueueLimit: meter.createGauge('dkg.backpressure.queue_limit', { + description: 'Configured scheduler queue capacity by bounded scheduler and lane', + }), + backpressureInflight: meter.createGauge('dkg.backpressure.inflight', { + description: 'Current admitted scheduler work by bounded scheduler and lane', + }), + backpressureInflightLimit: meter.createGauge('dkg.backpressure.inflight_limit', { + description: 'Configured scheduler concurrent-work capacity by bounded scheduler and lane', + }), + backpressureOldestQueuedAgeMs: meter.createGauge('dkg.backpressure.oldest_queued_age_ms', { + unit: 'ms', + description: 'Current age of the oldest queued scheduler item', + }), + backpressureOldestActiveAgeMs: meter.createGauge('dkg.backpressure.oldest_active_age_ms', { + unit: 'ms', + description: 'Current age of the oldest admitted scheduler item', + }), + backpressureEventsTotal: meter.createCounter('dkg.backpressure.events_total', { + description: 'Scheduler lifecycle events and admission rejections', + }), + backpressureQueueWaitMs: meter.createHistogram('dkg.backpressure.queue_wait_ms', { + unit: 'ms', + description: 'Scheduler queue wait by bounded scheduler and lane', + advice: { explicitBucketBoundaries: OP_DURATION_BUCKETS }, + }), + backpressureActiveDurationMs: meter.createHistogram('dkg.backpressure.active_duration_ms', { + unit: 'ms', + description: 'Scheduler admitted-work duration by bounded scheduler, lane, and outcome', + advice: { explicitBucketBoundaries: OP_DURATION_BUCKETS }, + }), storeScanSingleFlightJoinsTotal: meter.createCounter('dkg.store.scan_singleflight_joins_total', { description: 'Equivalent expensive scans joined to an already running promise', }), diff --git a/packages/core/test/backpressure-observability.test.ts b/packages/core/test/backpressure-observability.test.ts new file mode 100644 index 0000000000..0e59cba33a --- /dev/null +++ b/packages/core/test/backpressure-observability.test.ts @@ -0,0 +1,212 @@ +import { describe, expect, it } from 'vitest'; +import { + BackpressureMonitor, + BackpressureRegistry, + SchedulerPressureTracker, + type BackpressureSnapshot, +} from '../src/backpressure-observability.js'; + +describe('SchedulerPressureTracker', () => { + it('tracks queue and active age without owning scheduling policy', () => { + let now = 1_000; + const tracker = new SchedulerPressureTracker({ + scheduler: 'test scheduler', + now: () => now, + capacity: { + queueLimit: 4, + inflightLimit: 1, + lanes: { + normal: { queueLimit: 4, inflightLimit: 1 }, + }, + }, + thresholds: { + degradedQueueAgeMs: 5_000, + stalledActiveAgeMs: 30_000, + }, + }); + + const ticket = tracker.enqueue({ + lane: 'normal', + operation: 'sparql query with payload-ish punctuation?', + }); + now += 6_000; + expect(tracker.snapshot()).toMatchObject({ + scheduler: 'test_scheduler', + state: 'degraded', + totals: { + queued: 1, + inflight: 0, + oldestQueuedAgeMs: 6_000, + }, + lanes: [{ + lane: 'normal', + state: 'degraded', + queuedOperations: [{ + operation: 'sparql_query_with_payload-ish_punctuation_', + count: 1, + oldestAgeMs: 6_000, + }], + }], + }); + + tracker.start(ticket); + now += 31_000; + expect(tracker.snapshot()).toMatchObject({ + state: 'stalled', + totals: { + queued: 0, + inflight: 1, + oldestActiveAgeMs: 31_000, + }, + }); + + tracker.finish(ticket, 'completed'); + expect(tracker.snapshot()).toMatchObject({ + state: 'healthy', + totals: { queued: 0, inflight: 0 }, + }); + }); + + it('makes recent rejection pressure visible without retaining rejected payloads', () => { + let now = 10_000; + const tracker = new SchedulerPressureTracker({ + scheduler: 'store', + now: () => now, + thresholds: { rejectionStateWindowMs: 60_000 }, + }); + + tracker.reject({ lane: 'normal', operation: 'blazegraph.query' }, 'queue_full'); + expect(tracker.snapshot()).toMatchObject({ + state: 'saturated', + lanes: [{ + lane: 'normal', + rejectedTotal: 1, + rejectedByReason: { queue_full: 1 }, + queuedOperations: [], + activeOperations: [], + }], + }); + + now += 60_001; + expect(tracker.snapshot().state).toBe('healthy'); + }); +}); + +describe('BackpressureRegistry', () => { + it('isolates a broken diagnostic source', () => { + const registry = new BackpressureRegistry(); + registry.register({ + backpressureId: 'broken', + getBackpressureSnapshot: () => { + throw new Error('snapshot failed'); + }, + }); + + expect(registry.capture()).toMatchObject({ + state: 'healthy', + schedulers: [], + failures: [{ scheduler: 'broken', error: 'snapshot failed' }], + }); + }); +}); + +describe('BackpressureMonitor', () => { + it('logs transitions, rate-limited summaries, and recovery', () => { + let now = 1_000; + let state: BackpressureSnapshot['state'] = 'healthy'; + const registry = new BackpressureRegistry(); + registry.register({ + backpressureId: 'test', + getBackpressureSnapshot: () => ({ + scheduler: 'test', + state, + totals: { + queued: state === 'healthy' ? 0 : 3, + queueLimit: 4, + inflight: 1, + inflightLimit: 1, + oldestQueuedAgeMs: state === 'healthy' ? 0 : 6_000, + oldestActiveAgeMs: 1_000, + rejectedTotal: 0, + }, + lanes: [{ + lane: 'normal', + state, + queued: state === 'healthy' ? 0 : 3, + queueLimit: 4, + inflight: 1, + inflightLimit: 1, + oldestQueuedAgeMs: state === 'healthy' ? 0 : 6_000, + oldestActiveAgeMs: 1_000, + queuedOperations: [], + activeOperations: [], + events: {}, + rejectedTotal: 0, + rejectedByReason: {}, + lastRejectedAgeMs: null, + }], + }), + }); + const messages: Array<{ level: string; message: string }> = []; + const monitor = new BackpressureMonitor({ + registry, + now: () => now, + intervalMs: 5_000, + summaryIntervalMs: 60_000, + emit: (level, message) => messages.push({ level, message }), + }); + + monitor.sample(); + expect(messages).toEqual([]); + + state = 'degraded'; + monitor.sample(); + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ level: 'warn' }); + expect(messages[0].message).toContain('"event":"transition"'); + + now += 59_999; + monitor.sample(); + expect(messages).toHaveLength(1); + + now += 1; + monitor.sample(); + expect(messages).toHaveLength(2); + expect(messages[1].message).toContain('"event":"summary"'); + + state = 'healthy'; + monitor.sample(); + expect(messages).toHaveLength(3); + expect(messages[2]).toMatchObject({ level: 'info' }); + expect(messages[2].message).toContain('"event":"recovered"'); + }); + + it('contains logger failures', () => { + const registry = new BackpressureRegistry(); + registry.register({ + backpressureId: 'test', + getBackpressureSnapshot: () => ({ + scheduler: 'test', + state: 'saturated', + totals: { + queued: 1, + queueLimit: 1, + inflight: 0, + inflightLimit: 1, + oldestQueuedAgeMs: 1, + oldestActiveAgeMs: 0, + rejectedTotal: 1, + }, + lanes: [], + }), + }); + const monitor = new BackpressureMonitor({ + registry, + emit: () => { + throw new Error('logger unavailable'); + }, + }); + + expect(() => monitor.sample()).not.toThrow(); + }); +}); diff --git a/packages/publisher/src/async-lift-publisher-impl.ts b/packages/publisher/src/async-lift-publisher-impl.ts index 65e3edeabb..847e7decff 100644 --- a/packages/publisher/src/async-lift-publisher-impl.ts +++ b/packages/publisher/src/async-lift-publisher-impl.ts @@ -382,6 +382,7 @@ export class TripleStoreAsyncLiftPublisher await this.ensureGraph(); const result = await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { <${jobSubject(jobId)}> <${PAYLOAD_PREDICATE}> ?payload } }`, + { source: 'publisher.asyncLift.getStatus' }, ); const rows = expectBindings(result); if (rows.length === 0) return null; @@ -405,6 +406,7 @@ export class TripleStoreAsyncLiftPublisher // so a transient false `none` cannot by itself create a duplicate active job. const result = await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { ?job <${CONTROL_LIFECYCLE_KEY}> ${literal(key)} ; <${PAYLOAD_PREDICATE}> ?payload } }`, + { source: 'publisher.asyncLift.lookupVmPublishIntent' }, ); const jobs = expectBindings(result) .map((row) => this.parseJobPayload(row['payload'])) @@ -451,6 +453,7 @@ export class TripleStoreAsyncLiftPublisher // would over-select raw-lift jobs). const indexed = await this.store.query( `SELECT ?job WHERE { GRAPH <${this.graphUri}> { ?job <${CONTROL_LIFECYCLE_KEY}> ?lifecycleKey } }`, + { source: 'publisher.asyncLift.ensureVmPublishIntentIndex' }, ); const alreadyIndexed = new Set( expectBindings(indexed) @@ -468,6 +471,7 @@ export class TripleStoreAsyncLiftPublisher const statusFilter = filter.status ? `FILTER (?status = ${literal(filter.status)})` : ''; const result = await this.store.query( `SELECT ?payload ?status WHERE { GRAPH <${this.graphUri}> { ?job <${STATUS_PREDICATE}> ?status ; <${PAYLOAD_PREDICATE}> ?payload . ${statusFilter} } }`, + { source: 'publisher.asyncLift.list' }, ); return expectBindings(result) .map((row) => this.parseJobPayload(row['payload'])) @@ -1087,6 +1091,7 @@ export class TripleStoreAsyncLiftPublisher const rows = expectBindings( await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { <${jobSubject(jobId)}> <${PAYLOAD_PREDICATE}> ?payload } }`, + { source: 'publisher.asyncLift.clearTerminalJob' }, ), ); if (rows.length === 0) return { outcome: 'already_absent' }; @@ -1216,6 +1221,7 @@ export class TripleStoreAsyncLiftPublisher private async allocateJournalSeq(lineageKey: string): Promise { const result = await this.store.query( `SELECT (MAX(?seq) AS ?m) WHERE { GRAPH <${this.journalGraphUri}> { ?e <${JOURNAL_LIFECYCLE_KEY}> ${literal(lineageKey)} ; <${JOURNAL_SEQ}> ?seq } }`, + { source: 'publisher.asyncLift.allocateJournalSeq' }, ); const rows = expectBindings(result); const raw = rows.length === 0 ? undefined : rows[0]?.['m']; @@ -1288,6 +1294,7 @@ export class TripleStoreAsyncLiftPublisher private async readJournalEntriesBy(predicate: string, value: string): Promise { const result = await this.store.query( `SELECT ?e ?p ?o WHERE { GRAPH <${this.journalGraphUri}> { ?e <${predicate}> ${literal(value)} . ?e ?p ?o } }`, + { source: 'publisher.asyncLift.readJournalEntries' }, ); const bySubject = new Map>(); for (const row of expectBindings(result)) { @@ -1375,6 +1382,7 @@ export class TripleStoreAsyncLiftPublisher } | null> { const result = await this.store.query( `SELECT ?job ?status ?expiresAt ?claimToken WHERE { GRAPH <${this.walletLockGraphUri}> { <${walletLockSubject(walletId)}> <${CONTROL_LOCKED_JOB}> ?job ; <${CONTROL_LOCK_STATUS}> ?status . OPTIONAL { <${walletLockSubject(walletId)}> <${CONTROL_LOCK_EXPIRES_AT}> ?expiresAt } OPTIONAL { <${walletLockSubject(walletId)}> <${CONTROL_CLAIM_TOKEN}> ?claimToken } } }`, + { source: 'publisher.asyncLift.walletLock.read' }, ); const rows = expectBindings(result); if (rows.length === 0) return null; @@ -1396,6 +1404,7 @@ export class TripleStoreAsyncLiftPublisher const now = this.now(); const result = await this.store.query( `SELECT ?expiresAt WHERE { GRAPH <${this.walletLockGraphUri}> { <${walletLockSubject(walletId)}> <${CONTROL_LOCK_STATUS}> ${literal('active')} ; <${CONTROL_LOCK_EXPIRES_AT}> ?expiresAt . } }`, + { source: 'publisher.asyncLift.walletLock.active' }, ); const rows = expectBindings(result); if (rows.length === 0) return false; @@ -1406,6 +1415,7 @@ export class TripleStoreAsyncLiftPublisher const now = this.now(); const result = await this.store.query( `SELECT ?wallet ?job ?expiresAt ?claimToken WHERE { GRAPH <${this.walletLockGraphUri}> { ?lock <${CONTROL_WALLET_ID}> ?wallet ; <${CONTROL_LOCKED_JOB}> ?job ; <${CONTROL_LOCK_STATUS}> ${literal('active')} ; <${CONTROL_LOCK_EXPIRES_AT}> ?expiresAt . OPTIONAL { ?lock <${CONTROL_CLAIM_TOKEN}> ?claimToken } } }`, + { source: 'publisher.asyncLift.walletLock.sweep' }, ); const expiredWallets: string[] = []; for (const row of expectBindings(result)) { diff --git a/packages/publisher/src/async-promote-queue-impl.ts b/packages/publisher/src/async-promote-queue-impl.ts index d178d2decf..c78819c804 100644 --- a/packages/publisher/src/async-promote-queue-impl.ts +++ b/packages/publisher/src/async-promote-queue-impl.ts @@ -146,11 +146,14 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT async list(filter: PromoteListFilter = {}): Promise { return this.withMutationLock(async () => { await this.ensureGraph(); - return this.listUnlocked(filter); + return this.listUnlocked(filter, 'publisher.asyncPromote.list'); }); } - private async listUnlocked(filter: PromoteListFilter = {}): Promise { + private async listUnlocked( + filter: PromoteListFilter = {}, + source = 'publisher.asyncPromote.list', + ): Promise { const filters: string[] = []; if (filter.state && filter.state.length > 0) { const literals = filter.state.map((s) => literal(s)).join(', '); @@ -161,6 +164,7 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT } const result = await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { ?job <${PROMOTE_STATE}> ?state ; <${PROMOTE_PAYLOAD}> ?payload . ${filters.join(' ')} } }`, + { source }, ); const sorted = expectBindings(result) .map((row) => parseJobPayload(row['payload'])) @@ -241,7 +245,10 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT const now = this.now(); await this.reconcileExpiredRunning(now); - const candidates = (await this.listUnlocked()).filter((j) => { + const candidates = (await this.listUnlocked( + {}, + 'publisher.asyncPromote.claimNext.candidates', + )).filter((j) => { if (j.state === 'queued') return true; if (j.state === 'failed_retrying' && (j.attempt.nextRetryAt ?? 0) <= now) return true; return false; @@ -252,7 +259,10 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT claimableCandidates.push(candidate); } - const running = await this.listUnlocked({ state: ['running'] }); + const running = await this.listUnlocked( + { state: ['running'] }, + 'publisher.asyncPromote.claimNext.running', + ); const eligible = claimableCandidates.filter((candidate) => !running.some((active) => active.jobId !== candidate.jobId && this.jobsShareClaimLane(active, candidate)), ); @@ -404,7 +414,10 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT } private async reconcileExpiredRunning(now: number): Promise { - const running = await this.listUnlocked({ state: ['running'] }); + const running = await this.listUnlocked( + { state: ['running'] }, + 'publisher.asyncPromote.recoverExpired', + ); let reclaimed = 0; let abandoned = 0; @@ -503,7 +516,12 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT return this.withMutationLock(async () => { await this.ensureGraph(); const stats = Object.fromEntries(PROMOTE_JOB_STATES.map((s) => [s, 0])) as PromoteStats; - for (const job of await this.listUnlocked()) stats[job.state] += 1; + for (const job of await this.listUnlocked( + {}, + 'publisher.asyncPromote.stats', + )) { + stats[job.state] += 1; + } return stats; }); } @@ -587,6 +605,7 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT private async readJob(jobId: string): Promise { const result = await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { <${jobSubject(jobId)}> <${PROMOTE_PAYLOAD}> ?payload } }`, + { source: 'publisher.asyncPromote.readJob' }, ); const rows = expectBindings(result); if (rows.length === 0) return null; @@ -667,6 +686,7 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT const rows = expectBindings( await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { <${jobSubject(jobId)}> <${PROMOTE_PAYLOAD}> ?payload } }`, + { source: 'publisher.asyncPromote.clearTerminalJob' }, ), ); const parsed = classifyJobPayload(rows[0]?.['payload']); @@ -722,9 +742,11 @@ export class TripleStoreAsyncPromoteQueue implements AsyncPromoteQueue, PromoteT const result = lookup.lookupQuery.kind === 'assertionWildcard' ? await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { ?job <${PROMOTE_CONTEXT_GRAPH_ID}> ${literal(lookup.request.contextGraphId)} ; <${PROMOTE_ASSERTION_NAME}> ${literal(lookup.request.assertionName)} ; <${PROMOTE_PAYLOAD}> ?payload . } }`, + { source: 'publisher.asyncPromote.findActiveConflict' }, ) : await this.store.query( `SELECT ?payload WHERE { GRAPH <${this.graphUri}> { ?job <${PROMOTE_UNIQUENESS_KEY}> ?key ; <${PROMOTE_PAYLOAD}> ?payload . FILTER (?key IN (${lookup.lookupQuery.keys.map((key) => literal(key)).join(', ')})) } }`, + { source: 'publisher.asyncPromote.findActiveConflict' }, ); const rows = expectBindings(result); for (const row of rows) { diff --git a/packages/publisher/test/async-promote-queue.test.ts b/packages/publisher/test/async-promote-queue.test.ts index 5a55ad8489..0eb4d1fe2b 100644 --- a/packages/publisher/test/async-promote-queue.test.ts +++ b/packages/publisher/test/async-promote-queue.test.ts @@ -616,6 +616,36 @@ describe('TripleStoreAsyncPromoteQueue', () => { expect(await queue.claimNext('worker-1')).toBeNull(); // still backing off }); + it('labels each claimNext store read with its caller-provided operation class', async () => { + const sources: Array = []; + const recordingStore = new Proxy(store, { + get(target, prop, receiver) { + if (prop === 'query') { + return ( + sparql: Parameters[0], + options?: Parameters[1], + ) => { + sources.push(options?.source); + return target.query(sparql, options); + }; + } + const value = Reflect.get(target, prop, receiver); + return typeof value === 'function' ? value.bind(target) : value; + }, + }) as TripleStore; + const queue = new TripleStoreAsyncPromoteQueue(recordingStore, { + now: () => now, + idGenerator: () => `job-${++idCounter}`, + }); + + await expect(queue.claimNext('worker-1')).resolves.toBeNull(); + expect(sources).toEqual([ + 'publisher.asyncPromote.recoverExpired', + 'publisher.asyncPromote.claimNext.candidates', + 'publisher.asyncPromote.claimNext.running', + ]); + }); + it('13. claimNext() does NOT pick a second job for the same (cgId, subGraphName, assertionName) while one is running', async () => { const queue = createQueue(); const reqA = makeRequest(); diff --git a/packages/publisher/test/query-source-coverage.test.ts b/packages/publisher/test/query-source-coverage.test.ts new file mode 100644 index 0000000000..9288d4b98c --- /dev/null +++ b/packages/publisher/test/query-source-coverage.test.ts @@ -0,0 +1,91 @@ +import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import * as ts from 'typescript'; +import { describe, expect, it } from 'vitest'; + +const TARGETS = [ + '../src/async-lift-publisher-impl.ts', + '../src/async-promote-queue-impl.ts', +] as const; + +function propertyCarriesSource(property: ts.ObjectLiteralElementLike): boolean { + // A spread may carry a prepared QueryOptions.source that cannot be resolved + // locally. Keep those and non-literal option variables permissive; the guard + // is intended to reject only source-less option literals we can prove wrong. + if (ts.isSpreadAssignment(property)) return true; + const { name } = property; + if (!name) return false; + if (ts.isIdentifier(name) || ts.isStringLiteral(name)) return name.text === 'source'; + return ts.isComputedPropertyName(name) + && ts.isStringLiteral(name.expression) + && name.expression.text === 'source'; +} + +function missingSourceAttribution(sourceText: string, fileName: string): string[] { + const sourceFile = ts.createSourceFile( + fileName, + sourceText, + ts.ScriptTarget.Latest, + true, + ts.ScriptKind.TS, + ); + const missing: string[] = []; + + const visit = (node: ts.Node): void => { + if ( + ts.isCallExpression(node) + && ts.isPropertyAccessExpression(node.expression) + && node.expression.name.text === 'query' + ) { + const receiver = node.expression.expression.getText(sourceFile); + if (/(?:^|\.)store\??$/.test(receiver)) { + const options = node.arguments[1]; + const literalMissingSource = options !== undefined + && ts.isObjectLiteralExpression(options) + && !options.properties.some(propertyCarriesSource); + if (options === undefined || literalMissingSource) { + const position = sourceFile.getLineAndCharacterOfPosition(node.getStart(sourceFile)); + missing.push(`${fileName}:${position.line + 1}`); + } + } + } + ts.forEachChild(node, visit); + }; + + visit(sourceFile); + return missing; +} + +describe('publisher control-plane query source coverage', () => { + it('keeps direct store queries attributable in async queue and wallet-lock paths', () => { + const missingAttribution: string[] = []; + + // Deliberately scoped to the async publisher hot paths exercised by the + // scheduler profile. Add a file here when that profiled surface expands. + for (const relativePath of TARGETS) { + const path = fileURLToPath(new URL(relativePath, import.meta.url)); + const sourceText = readFileSync(path, 'utf8'); + missingAttribution.push(...missingSourceAttribution(sourceText, relativePath)); + } + + expect(missingAttribution).toEqual([]); + }); + + it('rejects source-less literals without blocking prepared or spread options', () => { + const sourceText = [ + "store.query('missing-options');", + "store.query('empty-options', {});", + "store.query('other-options', { signal });", + "store.query('literal-source', { source: 'publisher.test' });", + "store.query('shorthand-source', { source });", + "store.query('prepared-options', queryOptions);", + "store.query('spread-options', { ...queryOptions });", + ].join('\n'); + + expect(missingSourceAttribution(sourceText, 'fixture.ts')).toEqual([ + 'fixture.ts:1', + 'fixture.ts:2', + 'fixture.ts:3', + ]); + }); +}); diff --git a/packages/publisher/vitest.unit.config.ts b/packages/publisher/vitest.unit.config.ts index 35568a7a58..0b51d7f563 100644 --- a/packages/publisher/vitest.unit.config.ts +++ b/packages/publisher/vitest.unit.config.ts @@ -24,6 +24,7 @@ export default defineConfig({ 'test/async-lift-publish-options.test.ts', 'test/async-promote-queue.test.ts', 'test/async-promote-writejob-atomicity.test.ts', + 'test/query-source-coverage.test.ts', 'test/subject-atomic-write.test.ts', 'test/async-lift-terminal-clear.test.ts', 'test/async-promote-terminal-clear.test.ts', diff --git a/packages/storage/src/graph-manager.ts b/packages/storage/src/graph-manager.ts index 3a6c372860..aed9b1abc9 100644 --- a/packages/storage/src/graph-manager.ts +++ b/packages/storage/src/graph-manager.ts @@ -840,8 +840,8 @@ export class ContextGraphManager { this.ensuredContextGraphs.add(contextGraphId); } - async listContextGraphs(): Promise { - const graphs = await listGraphsByPrefix(this.store, CG_PREFIX); + async listContextGraphs(options?: QueryOptions): Promise { + const graphs = await listGraphsByPrefix(this.store, CG_PREFIX, options); const contextGraphs = new Set(); for (const g of graphs) { if (g.startsWith(CG_PREFIX)) { diff --git a/packages/storage/src/store-priority-scheduler.ts b/packages/storage/src/store-priority-scheduler.ts index 7d2e7e81a2..472fd09423 100644 --- a/packages/storage/src/store-priority-scheduler.ts +++ b/packages/storage/src/store-priority-scheduler.ts @@ -1,5 +1,11 @@ import { performance } from 'node:perf_hooks'; -import { getMetrics } from '@origintrail-official/dkg-core'; +import { + backpressureRegistry, + getMetrics, + ObservableScheduler, + type SchedulerPressureOutcome, + type SchedulerPressureTicket, +} from '@origintrail-official/dkg-core'; import type { StorePressureSnapshot, StoreWorkPriority } from './triple-store.js'; export interface StorePrioritySchedulerSnapshot extends StorePressureSnapshot { @@ -60,6 +66,7 @@ interface QueueEntry { priority: StoreWorkPriority; operation: string; queuedAt: number; + pressureTicket: SchedulerPressureTicket; work: () => Promise; resolve: (value: T) => void; reject: (reason?: unknown) => void; @@ -207,7 +214,7 @@ export function storeWorkPriorityRank(priority: StoreWorkPriority): number { return 3; } -export class StorePriorityScheduler { +export class StorePriorityScheduler extends ObservableScheduler { private readonly queues: Record>> = { ack: [], health: [], @@ -256,6 +263,20 @@ export class StorePriorityScheduler { queueWaitTimeoutMs: legacyQueueWaitTimeoutMs, healthReservedSlots: legacyHealthReservedSlots, }); + const resolvedNow = options.now ?? (() => performance.now()); + const resolvedQueueWaitTimeoutMs = options.queueWaitTimeoutMs + ?? parsePositiveIntegerEnv( + 'DKG_STORE_QUEUE_WAIT_TIMEOUT_MS', + DEFAULT_STORE_QUEUE_WAIT_TIMEOUT_MS, + ); + super({ + scheduler: 'store', + now: resolvedNow, + thresholds: { + degradedQueueAgeMs: Math.max(1, Math.floor(resolvedQueueWaitTimeoutMs / 2)), + stalledActiveAgeMs: 30_000, + }, + }); this.maxConcurrent = options.maxConcurrent ?? parsePositiveIntegerEnv('DKG_STORE_MAX_CONCURRENT', DEFAULT_MAX_CONCURRENT); const requestedAckReservedSlots = options.ackReservedSlots @@ -285,13 +306,32 @@ export class StorePriorityScheduler { requestedNormalFloor, requestedBackgroundFloor, ); - this.queueWaitTimeoutMs = options.queueWaitTimeoutMs - ?? parsePositiveIntegerEnv( - 'DKG_STORE_QUEUE_WAIT_TIMEOUT_MS', - DEFAULT_STORE_QUEUE_WAIT_TIMEOUT_MS, - ); - this.now = options.now ?? (() => performance.now()); + this.queueWaitTimeoutMs = resolvedQueueWaitTimeoutMs; + this.now = resolvedNow; this.queueLimits = normalizeQueueLimits(options.queueLimits ?? resolveQueueLimitsFromEnv()); + const nonAckLimit = Math.max(1, this.maxConcurrent - this.ackReservedSlots); + this.updatePressureCapacity({ + queueLimit: Object.values(this.queueLimits).reduce((sum, value) => sum + value, 0), + inflightLimit: this.maxConcurrent, + lanes: { + ack: { + queueLimit: this.queueLimits.ack, + inflightLimit: this.maxConcurrent, + }, + health: { + queueLimit: this.queueLimits.health, + inflightLimit: nonAckLimit, + }, + normal: { + queueLimit: this.queueLimits.normal, + inflightLimit: this.nonAckLanePolicy.totalLimit, + }, + background: { + queueLimit: this.queueLimits.background, + inflightLimit: this.nonAckLanePolicy.backgroundLimit, + }, + }, + }); } get snapshot(): StorePrioritySchedulerSnapshot { @@ -330,14 +370,23 @@ export class StorePriorityScheduler { } if (this.queues[normalizedPriority].length >= this.queueLimits[normalizedPriority]) { const error = new StoreSchedulerBusyError('queue_full', normalizedPriority, operation); + this.pressureReject( + { lane: normalizedPriority, operation }, + error.reason, + ); this.observeRejection(error); throw error; } return new Promise((resolve, reject) => { + const pressureTicket = this.pressureEnqueue({ + lane: normalizedPriority, + operation, + }); const entry: QueueEntry = { priority: normalizedPriority, operation, queuedAt: this.now(), + pressureTicket, work, resolve, reject, @@ -346,6 +395,7 @@ export class StorePriorityScheduler { const onAbort = () => { if (this.removeQueued(entry as QueueEntry)) { this.cleanupQueuedEntry(entry as QueueEntry); + this.pressureCancelQueued(entry.pressureTicket, 'aborted'); const reason = signal?.reason; reject(reason instanceof Error ? reason : new Error(String(reason ?? 'aborted'))); this.observeDepths(); @@ -363,6 +413,7 @@ export class StorePriorityScheduler { normalizedPriority, operation, ); + this.pressureRejectQueued(entry.pressureTicket, error.reason); this.observeRejection(error); reject(error); this.observeDepths(); @@ -420,6 +471,7 @@ export class StorePriorityScheduler { this.cleanupQueuedEntry(entry); this.increment(entry.priority); const startedAt = this.now(); + this.pressureStart(entry.pressureTicket); const waitMs = Math.max(0, startedAt - entry.queuedAt); const attributes = { priority: entry.priority, @@ -436,17 +488,20 @@ export class StorePriorityScheduler { result = Promise.reject(err); } + let pressureOutcome: SchedulerPressureOutcome = 'completed'; result .then((value) => { this.observeDuration(entry, startedAt, 'ok'); entry.resolve(value); }) .catch((err) => { + pressureOutcome = 'failed'; this.observeDuration(entry, startedAt, 'error'); entry.reject(err); }) .finally(() => { getMetrics().storeSchedulerActive.add(-1, attributes); + this.pressureFinish(entry.pressureTicket, pressureOutcome); this.decrement(entry.priority); this.observeDepths(); this.drain(); @@ -521,6 +576,7 @@ export class StorePriorityScheduler { } export const externalStorePriorityScheduler = new StorePriorityScheduler(); +backpressureRegistry.register(externalStorePriorityScheduler); export function getExternalStorePrioritySchedulerSnapshot(): StorePrioritySchedulerSnapshot { return externalStorePriorityScheduler.snapshot; diff --git a/packages/storage/test/store-priority-scheduler.test.ts b/packages/storage/test/store-priority-scheduler.test.ts index 7a8d29e299..5130bfa26f 100644 --- a/packages/storage/test/store-priority-scheduler.test.ts +++ b/packages/storage/test/store-priority-scheduler.test.ts @@ -649,4 +649,62 @@ describe('StorePriorityScheduler', () => { 'normal', ]); }); + + it('exposes generic pressure diagnostics without changing lane admission', async () => { + let now = 1_000; + const scheduler = new StorePriorityScheduler({ + maxConcurrent: 1, + ackReservedSlots: 0, + healthReservedSlots: 0, + backgroundReservedSlots: 0, + queueLimits: { ack: 1, health: 1, normal: 1, background: 1 }, + queueWaitTimeoutMs: 10_000, + now: () => now, + }); + let release!: () => void; + const blocker = scheduler.run('normal', 'blazegraph.query', async () => { + await new Promise((resolve) => { + release = resolve; + }); + }); + const queued = scheduler.run('normal', 'swm.atomicReplace', async () => undefined); + + now += 6_000; + expect(scheduler.getBackpressureSnapshot()).toMatchObject({ + scheduler: 'store', + state: 'saturated', + totals: { + queued: 1, + queueLimit: 4, + inflight: 1, + inflightLimit: 1, + oldestQueuedAgeMs: 6_000, + }, + lanes: expect.arrayContaining([ + expect.objectContaining({ + lane: 'normal', + queued: 1, + inflight: 1, + queueLimit: 1, + queuedOperations: [{ + operation: 'swm.atomicReplace', + count: 1, + oldestAgeMs: 6_000, + }], + activeOperations: [{ + operation: 'blazegraph.query', + count: 1, + oldestAgeMs: 6_000, + }], + }), + ]), + }); + + release(); + await Promise.all([blocker, queued]); + expect(scheduler.getBackpressureSnapshot()).toMatchObject({ + state: 'healthy', + totals: { queued: 0, inflight: 0 }, + }); + }); });