RFC-64 M1 5/7: activate role-aware adaptive Core capacity

packages/agent/src/dkg-agent-base.ts

@@ -164,6 +164,10 @@ import {
   resolveCorePublicSyncBatchSize,
   type CorePublicSyncCoverageStatus,
 } from './sync/core-public-coverage-scheduler.js';
+import {
+  SyncCapacityRuntime,
+  type SyncCapacityStatus,
+} from './sync/capacity-runtime.js';
 import { bindRandomSampling, type RandomSamplingDisabledReason, type RandomSamplingHandle, type RandomSamplingStatus } from './random-sampling-bind.js';
 import { connectToMultiaddr, ensurePeerConnected as ensurePeerConnectedAtom, primeCatchupConnections as primeCatchupConnectionsAtom } from './p2p/peer-connect.js';
 import { Messenger, type SloProtocolStats } from './p2p/messenger.js';
@@ -1060,6 +1064,8 @@ export class DKGAgentBase {
    * capped by this scheduler; Edge nodes never register automatic coverage.
    */
   protected readonly corePublicSyncCoverageScheduler: CorePublicSyncCoverageScheduler;
+  /** Role-aware requester admission and Core automatic-coverage capacity. */
+  protected readonly syncCapacityRuntime: SyncCapacityRuntime;
   protected started = false;
   /**
    * One OT-RFC-64 persistence owner for the inventory lease and every resource
@@ -1624,12 +1630,18 @@ export class DKGAgentBase {
     publicSnapshotStore?: WorkspacePublicSnapshotStore,
   ) {
     this.config = config;
+    const resolvedCoverageBatch = resolveCorePublicSyncBatchSize(
+      config.syncCorePublicBatchSize,
+    );
     this.corePublicSyncCoverageScheduler = new CorePublicSyncCoverageScheduler(
-      resolveCorePublicSyncBatchSize(config.syncCorePublicBatchSize),
+      resolvedCoverageBatch,
     );
     this.wallet = wallet;
     this.node = node;
     this.store = store;
+    this.syncCapacityRuntime = SyncCapacityRuntime.create(config, store, {
+      resolvedCoverageBatch,
+    });
     this.contextGraphMetaProjection = new ContextGraphMetaProjection(store);
     this.publisher = publisher;
     this.queryEngine = queryEngine;
@@ -1716,13 +1728,15 @@ export class DKGAgentBase {
   /** Build one canonical peer-round scope with named snapshot/live phases. */
   protected planCorePublicSyncPeerRound(remotePeer: string): PeerSyncScope {
     const selected = [...(this.config.syncContextGraphs ?? [])];
-    const automaticContextGraphIds = (this.config.nodeRole ?? 'edge') === 'core'
-      ? this.corePublicSyncCoverageScheduler.planAutomaticCoverage(
-        selected,
-        this.config.syncContextGraphPriorities,
-        remotePeer,
-      )
-      : [];
+    let automaticContextGraphIds: string[] = [];
+    if ((this.config.nodeRole ?? 'edge') === 'core') {
+      automaticContextGraphIds = this.corePublicSyncCoverageScheduler
+        .planAutomaticCoverageWithOptions(selected, {
+          priorities: this.config.syncContextGraphPriorities,
+          planningLane: remotePeer,
+          effectiveBatchSize: this.syncCapacityRuntime.getEffectiveCoverageBatch(),
+        });
+    }
     const initialDurableContextGraphIds = [...new Set([
       ...selected,
       ...automaticContextGraphIds,
@@ -1743,6 +1757,10 @@ export class DKGAgentBase {
     );
   }
 
+  getSyncCapacityStatus(): SyncCapacityStatus {
+    return this.syncCapacityRuntime.getStatus();
+  }
+
   /**
    * Acquire the RFC-64 inventory, finish bounded stale-candidate cleanup, and
    * open the inherited-owner control-object tree before network consumers.

packages/agent/src/dkg-agent-lifecycle.ts

@@ -303,11 +303,9 @@ import {
   type DurableSyncAccumulator,
 } from './sync/durable-progress.js';
 import {
-  getSyncBackpressureSnapshot,
   getSyncBackpressureBusyError,
   resolveBooleanSwitch,
   resolveNonNegativeIntegerSwitch,
-  resolveSyncGlobalBackpressure,
   withGlobalSyncBackpressure,
 } from './sync/backpressure.js';
 import {
@@ -1233,9 +1231,10 @@ export class LifecycleSyncMethods extends DKGAgentBase {
       operationSignal,
     );
     try {
+      const capacityAdmission = this.syncCapacityRuntime.getAdmissionOptions();
       return await withGlobalSyncBackpressure(
         {
-          policy: resolveSyncGlobalBackpressure(this.config),
+          ...capacityAdmission,
           ctx,
           label,
           contextGraphId,
@@ -2021,7 +2020,7 @@ export class LifecycleSyncMethods extends DKGAgentBase {
                 );
               },
               onDecline: (details) => {
-                const syncPressure = getSyncBackpressureSnapshot(resolveSyncGlobalBackpressure(this.config));
+                const syncPressure = this.syncCapacityRuntime.getBackpressureSnapshot();
                 const syncPressureLabel =
                   `syncGlobalInflight=${syncPressure.inflight} ` +
                   `syncGlobalQueued=${syncPressure.queued} ` +
@@ -2544,15 +2543,19 @@ export class LifecycleSyncMethods extends DKGAgentBase {
       process.env,
       (message) => this.log.warn(ctx, message),
     );
-    const syncGlobalPolicy = resolveSyncGlobalBackpressure(this.config);
+    const syncGlobalPolicy = this.syncCapacityRuntime.getResolvedPolicyStatus();
+    const syncCapacity = this.syncCapacityRuntime.getStatus();
     const configuredPriorityCounts = countSyncPriorityClasses(this.config.syncContextGraphPriorities);
     this.log.info(ctx, `Resolved sync policy ${JSON.stringify({
       snapshotGlobalRows: snapshotPolicy.budget.maxRows,
       snapshotGlobalBytesEstimate: snapshotPolicy.budget.maxBytesEstimate,
       snapshotLocalRows: snapshotPolicy.budget.maxSnapshotRows,
       snapshotLocalBytesEstimate: snapshotPolicy.budget.maxSnapshotBytesEstimate,
-      syncGlobalInflightLimit: syncGlobalPolicy.limit ?? 0,
+      syncGlobalInflightLimit: syncGlobalPolicy.inflightLimit ?? 0,
       syncGlobalQueueLimit: syncGlobalPolicy.queueLimit ?? 0,
+      syncCapacityMode: syncCapacity.mode,
+      syncCapacityCurrentInflight: syncCapacity.currentInflight ?? 0,
+      syncCapacityCoverageBatch: syncCapacity.currentCoverageBatch,
       configuredPriorities: configuredPriorityCounts,
       snapshotLocalClamped: snapshotPolicy.localRowsClamped || snapshotPolicy.localBytesEstimateClamped,
     })}`);
@@ -3249,6 +3252,19 @@ export class LifecycleSyncMethods extends DKGAgentBase {
       this.log.warn(ctx, `Skipping periodic sync reconciler startup (DKG_SYNC_RECONCILER_ENABLED=0)`);
     }
 
+    this.syncCapacityRuntime.startSampling({
+      hasSupplementalDemand: () => (
+        this.corePublicSyncCoverageScheduler.hasAutomaticCoverageBacklog(
+          this.config.syncContextGraphs ?? [],
+          this.syncCapacityRuntime.getEffectiveCoverageBatch(),
+        )
+      ),
+      onError: (error) => {
+        const message = error instanceof Error ? error.message : String(error);
+        this.log.warn(ctx, `Adaptive sync capacity sample failed: ${message}`);
+      },
+    });
+
     // A.4-lite+: keep a small set of Core nodes warm (connection pinned +
     // auto-redialed by libp2p) so catch-up / chain reconciliation never pays
     // a cold circuit-relay dial to reach a Core. Opt-in via

packages/agent/src/dkg-agent-types.ts

@@ -1160,6 +1160,23 @@ export interface Rfc64PublicCatalogBootstrapConfigV1 {
   readonly retryIntervalMs?: number;
 }
 
+/**
+ * Optional bounds for role-aware adaptive sync concurrency.
+ *
+ * These values are operator policy, not instantaneous capacity:
+ * `minInflight` is the floor the controller may reduce to and `maxInflight` is
+ * the ceiling it may grow to. Omitting the whole block preserves role-aware
+ * defaults; `enabled` can explicitly opt in or out without restating bounds.
+ */
+export interface SyncAdaptiveCapacityConfig {
+  /** Explicitly enable or disable adaptive sync capacity. Omit for the role-aware default. */
+  readonly enabled?: boolean;
+  /** Minimum effective global sync concurrency the controller may select. */
+  readonly minInflight?: number;
+  /** Maximum effective global sync concurrency the controller may select. */
+  readonly maxInflight?: number;
+}
+
 export interface DKGAgentConfig {
   name: string;
   /** Selected genesis document. Defaults to the compatibility Base testnet genesis. */
@@ -1272,6 +1289,11 @@ export interface DKGAgentConfig {
   syncGlobalLimit?: number;
   /** Max sync jobs waiting behind the global cap. Defaults to 2x the inflight cap. */
   syncGlobalQueueLimit?: number;
+  /**
+   * Optional role-aware adaptive sync-concurrency policy. Omission preserves
+   * role-aware defaults and is distinct from an explicit disable.
+   */
+  syncAdaptiveCapacity?: SyncAdaptiveCapacityConfig;
   /**
    * Maximum automatically discovered public CGs a Core adds to one peer-sync
    * round. Explicitly selected CGs are not capped. Defaults to 8; 0 disables

packages/agent/src/dkg-agent.ts

@@ -361,6 +361,7 @@ import {
   type DurableSyncResult,
   type SharedMemorySyncResult,
   type DKGAgentConfig,
+  type SyncAdaptiveCapacityConfig,
   type Rfc64CatalogAccessPolicyAuthorityConfigV1,
   type DKGAgentACKTransportOptions,
   type ImportedArtifactByteStore,
@@ -473,6 +474,7 @@ export type {
   SharedMemorySyncDiagnostics,
   CatchupSyncDiagnostics,
   DKGAgentConfig,
+  SyncAdaptiveCapacityConfig,
   Rfc64CatalogAccessPolicyAuthorityConfigV1,
   DKGAgentACKTransportOptions,
   ImportedArtifactByteStore,
@@ -1720,6 +1722,7 @@ export class DKGAgent extends DKGAgentBase {
 
   async stop(): Promise<void> {
     if (!this.started) return;
+    this.syncCapacityRuntime.stopSampling();
     if (this.chainPoller) {
       // Await so any in-flight poll (and its HTTP keep-alive socket) settles
       // BEFORE we tear down the chain adapter — otherwise the RPC connection

packages/agent/src/index.ts

@@ -194,6 +194,7 @@ export {
   InvalidContentError,
   StaleSenderKeyTargetError,
   type DKGAgentConfig,
+  type SyncAdaptiveCapacityConfig,
   type Rfc64CatalogAccessPolicyAuthorityConfigV1,
   type DKGAgentACKTransportOptions,
   type ContextGraphSub,

packages/agent/src/sync/backpressure.ts

@@ -313,22 +313,27 @@ export function resolveExplicitSyncGlobalLimit(
     ?? nonNegativeInteger(config.syncGlobalLimit);
 }
 
-/** Resolve one node-wide policy; the optional live limit is shared by every admission. */
-export function resolveSyncGlobalBackpressure(
-  config: SyncGlobalBackpressureConfig,
+/** Build a policy from a final hard limit without re-reading global-limit env inputs. */
+export function createSyncGlobalBackpressurePolicy(
+  resolvedLimit: number,
+  configuredQueueLimit?: number,
   currentLimit?: SyncBackpressureCurrentLimit,
 ): SyncGlobalBackpressurePolicy {
-  const limit = resolveExplicitSyncGlobalLimit(config)
-    ?? DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT;
+  const limit = nonNegativeInteger(resolvedLimit);
+  if (limit === undefined) {
+    throw new TypeError('resolved sync global limit must be a non-negative integer');
+  }
   if (limit === 0) {
     return Object.freeze({
       limit: undefined,
       queueLimit: undefined,
     }) as SyncGlobalBackpressurePolicy;
   }
 
+  // Queue capacity remains an independent operator control. Only the already-
+  // resolved hard limit is protected from a second global-limit env lookup.
   const queueLimit = nonNegativeInteger(parseIntegerEnv('DKG_SYNC_GLOBAL_QUEUE_LIMIT'))
-    ?? nonNegativeInteger(config.syncGlobalQueueLimit)
+    ?? nonNegativeInteger(configuredQueueLimit)
     ?? limit * DEFAULT_SYNC_GLOBAL_QUEUE_LIMIT_MULTIPLIER;
   return Object.freeze({
     limit,
@@ -337,6 +342,20 @@ export function resolveSyncGlobalBackpressure(
   }) as SyncGlobalBackpressurePolicy;
 }
 
+/** Resolve one node-wide policy; the optional live limit is shared by every admission. */
+export function resolveSyncGlobalBackpressure(
+  config: SyncGlobalBackpressureConfig,
+  currentLimit?: SyncBackpressureCurrentLimit,
+): SyncGlobalBackpressurePolicy {
+  const limit = resolveExplicitSyncGlobalLimit(config)
+    ?? DEFAULT_SYNC_GLOBAL_MAX_INFLIGHT;
+  return createSyncGlobalBackpressurePolicy(
+    limit,
+    config.syncGlobalQueueLimit,
+    currentLimit,
+  );
+}
+
 export function getSyncBackpressureSnapshot(
   policy?: SyncGlobalBackpressurePolicy,
   now = Date.now(),