diff --git a/config/scripts/pr-code-change-scope-cross-version-send-path.test.mjs b/config/scripts/pr-code-change-scope-cross-version-send-path.test.mjs index 2e0cd544467..14da3c177c8 100644 --- a/config/scripts/pr-code-change-scope-cross-version-send-path.test.mjs +++ b/config/scripts/pr-code-change-scope-cross-version-send-path.test.mjs @@ -10,7 +10,6 @@ describe('cross-version wire routing for the send path', () => { 'src/shared/structured-agent-session-send-mutation.ts', 'src/main/runtime/rpc/core.ts', 'src/main/runtime/rpc/errors.ts', - 'src/main/runtime/rpc/rpc-streaming-dispatcher.ts', 'src/main/runtime/rpc/orchestration-contract-fence.ts', 'src/main/runtime/rpc/orchestration-session-caller.ts', 'src/main/runtime/rpc/orchestration-legacy-compatibility.ts', diff --git a/config/scripts/pr-code-change-scope.mjs b/config/scripts/pr-code-change-scope.mjs index 5ed417ba2a9..c9d52d9f36d 100644 --- a/config/scripts/pr-code-change-scope.mjs +++ b/config/scripts/pr-code-change-scope.mjs @@ -206,7 +206,6 @@ const CROSS_VERSION_WIRE_PREFIXES = [ // Run on every request the suites dispatch, whatever its method. 'src/main/runtime/rpc/core.ts', 'src/main/runtime/rpc/errors.ts', - 'src/main/runtime/rpc/rpc-streaming-dispatcher.ts', 'src/main/runtime/rpc/orchestration-contract-fence.ts', 'src/main/runtime/rpc/orchestration-session-caller.ts', 'src/main/runtime/rpc/orchestration-legacy-compatibility.ts', diff --git a/src/main/runtime/rpc/dispatcher-caller-fingerprint.ts b/src/main/runtime/rpc/dispatcher-caller-fingerprint.ts deleted file mode 100644 index f94551d5d6c..00000000000 --- a/src/main/runtime/rpc/dispatcher-caller-fingerprint.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { isDurableMutation } from '../../../shared/orchestration-rpc-contract' -import type { RpcRequest } from './core' - -export function needsLocalCallerFingerprint(request: RpcRequest, params: unknown): boolean { - return ( - request.method.startsWith('orchestration.federation') || - (!!request.orchestrationRequestId && isDurableMutation(request.method, params)) - ) -} diff --git a/src/main/runtime/rpc/dispatcher-client-browser-routing.ts b/src/main/runtime/rpc/dispatcher-client-browser-routing.ts deleted file mode 100644 index a70d0687353..00000000000 --- a/src/main/runtime/rpc/dispatcher-client-browser-routing.ts +++ /dev/null @@ -1,12 +0,0 @@ -import type { OrcaRuntimeService } from '../orca-runtime' - -export function routeDispatcherClientHostedBrowserRpc( - runtime: OrcaRuntimeService, - method: string, - params: unknown -) { - const candidate = runtime as OrcaRuntimeService & { - routeClientHostedBrowserRpc?: OrcaRuntimeService['routeClientHostedBrowserRpc'] - } - return candidate.routeClientHostedBrowserRpc?.(method, params) ?? { handled: false as const } -} diff --git a/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts b/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts deleted file mode 100644 index 7b26823026c..00000000000 --- a/src/main/runtime/rpc/dispatcher-unary-method-invocation.ts +++ /dev/null @@ -1,91 +0,0 @@ -import type { OrcaRuntimeService } from '../orca-runtime' -import type { RpcContext, RpcMethod, RpcRequest } from './core' -import { routeDispatcherClientHostedBrowserRpc } from './dispatcher-client-browser-routing' -import { needsLocalCallerFingerprint } from './dispatcher-caller-fingerprint' -import type { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' -import type { - DurableMutationInvocation, - OrchestrationMutationExecutor -} from './orchestration-mutation-executor' -import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction' - -type DispatcherUnaryMethodInvocation = { - runtime: OrcaRuntimeService - request: RpcRequest - method: RpcMethod - params: unknown - context: RpcContext - orchestrationMutations: OrchestrationMutationExecutor - legacyOrchestration: OrchestrationLegacyCompatibility -} - -export async function invokeDispatcherUnaryMethod({ - runtime, - request, - method, - params, - context, - orchestrationMutations, - legacyOrchestration -}: DispatcherUnaryMethodInvocation): Promise { - const clientHostedBrowser = await routeDispatcherClientHostedBrowserRpc( - runtime, - request.method, - params - ) - if (clientHostedBrowser.handled) { - recordRuntimeFeatureInteraction( - runtime, - request.method, - clientHostedBrowser.result, - undefined, - request.params - ) - return clientHostedBrowser.result - } - - const compatibility = await legacyOrchestration.tryHandle(request, params, context.signal) - if (compatibility.handled) { - return compatibility.result - } - const effectiveParams = compatibility.params ?? params - const legacyCoordinator = legacyOrchestration.createCoordinatorInvocation( - request, - compatibility.legacyCoordinatorAuthority - ) - // Why: a legacy coordinator's fingerprint outranks the transport's, so skip minting a local one. - const fallbackCallerFingerprint = - context.authenticatedCallerFingerprint ?? - (!legacyCoordinator && needsLocalCallerFingerprint(request, effectiveParams) - ? orchestrationMutations.getLocalAuthenticatedCallerFingerprint() - : undefined) - const callerFingerprint = - legacyCoordinator?.mutationCallerFingerprint ?? fallbackCallerFingerprint - const invoke = (mutation?: DurableMutationInvocation) => { - const legacyCoordinatorRunId = legacyCoordinator?.revalidate() - return method.handler(effectiveParams, { - ...context, - authenticatedCallerFingerprint: mutation?.identity.callerFingerprint ?? callerFingerprint, - recordMutationReceipt: mutation?.recordReceipt, - markWorkerDoneMutationEffectFree: mutation?.markWorkerDoneEffectFree, - markMutationEffectPossible: mutation?.markEffectPossible, - orchestrationMutation: mutation?.identity, - replayedMutationReceipt: mutation?.replayedReceipt, - legacyCoordinatorRunId, - legacyCoordinatorAuthority: legacyCoordinator?.authority, - revalidateLegacyCoordinator: legacyCoordinator?.revalidate, - orchestrationCompatibilityCallerAuthority: - compatibility.orchestrationCompatibilityCallerAuthority, - orchestrationCompatibilityEvidence: request.orchestrationCompatibilityEvidence - }) - } - const result = await orchestrationMutations.run( - request, - effectiveParams, - invoke, - callerFingerprint, - context.orchestrationCaller?.orcaSessionId - ) - recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params) - return result -} diff --git a/src/main/runtime/rpc/dispatcher.ts b/src/main/runtime/rpc/dispatcher.ts index 2fa14f238e1..6911684def3 100644 --- a/src/main/runtime/rpc/dispatcher.ts +++ b/src/main/runtime/rpc/dispatcher.ts @@ -14,8 +14,11 @@ import { successResponse } from './errors' import { ALL_RPC_METHODS } from './methods' import { emulatorProbe, emulatorProbeError } from '../../emulator/emulator-probe' import type { OrcaRuntimeService } from '../orca-runtime' +import { isDurableMutation } from '../../../shared/orchestration-rpc-contract' +import type { OrchestrationSessionCaller } from '../orchestration/orchestration-caller-identity' import { getOrchestrationMutationExecutor, + type DurableMutationInvocation, type OrchestrationMutationExecutor } from './orchestration-mutation-executor' import { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' @@ -24,8 +27,8 @@ import { type RpcDispatchStreamingOptions } from './dispatcher-stream-options' import { mapDispatcherError } from './dispatcher-error-response' -import { RpcStreamingDispatcher } from './rpc-streaming-dispatcher' -import { invokeDispatcherUnaryMethod } from './dispatcher-unary-method-invocation' +import { createDispatcherStreamingFeatureEmitter } from './dispatcher-streaming-feature-emitter' +import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction' import type { RpcCallerScope } from './rpc-caller-scope' import { admitRpcRequest } from './rpc-request-admission-prologue' @@ -36,12 +39,18 @@ export type DispatcherOptions = { callerScope?: RpcCallerScope } +type AdmittedUnaryCall = { + request: RpcRequest + method: RpcMethod + params: unknown + caller: OrchestrationSessionCaller | undefined +} + export class RpcDispatcher { private readonly runtime: OrcaRuntimeService private readonly registry: RpcRegistry private readonly orchestrationMutations: OrchestrationMutationExecutor private readonly legacyOrchestration: OrchestrationLegacyCompatibility - private readonly streamingDispatcher: RpcStreamingDispatcher private readonly pinnedCallerScope: RpcCallerScope | undefined constructor({ runtime, methods = ALL_RPC_METHODS, callerScope }: DispatcherOptions) { @@ -50,27 +59,14 @@ export class RpcDispatcher { this.registry = buildRegistry(methods) this.orchestrationMutations = getOrchestrationMutationExecutor(runtime) this.legacyOrchestration = new OrchestrationLegacyCompatibility(runtime) - this.streamingDispatcher = new RpcStreamingDispatcher({ - runtime, - registry: this.registry, - orchestrationMutations: this.orchestrationMutations, - legacyOrchestration: this.legacyOrchestration, - meta: () => this.meta(), - pinnedCallerScope: callerScope - }) } async dispatch(request: RpcRequest, options?: RpcDispatchStreamingOptions): Promise { const meta = this.meta() - const pendingAdmission = admitRpcRequest( - { - runtime: this.runtime, - registry: this.registry, - meta, - pinnedCallerScope: this.pinnedCallerScope - }, + const pendingAdmission = this.admit( request, options, + meta, (method: RpcMethod | RpcStreamingMethod): method is RpcMethod => !isStreamingMethod(method) ) const admission = @@ -79,27 +75,15 @@ export class RpcDispatcher { return admission.rejected } - if (request.method.startsWith('emulator.')) { + // Why: only the request/response transport probes emulator calls; streaming-unary never did. + const probeEmulator = request.method.startsWith('emulator.') + if (probeEmulator) { emulatorProbe(`rpc ${request.method}`, request.params) } try { - const result = await invokeDispatcherUnaryMethod({ - runtime: this.runtime, - request: admission.request, - method: admission.method, - params: admission.params, - context: rpcContextFromTransport( - this.runtime, - admission.request, - options, - admission.caller - ), - orchestrationMutations: this.orchestrationMutations, - legacyOrchestration: this.legacyOrchestration - }) - return successResponse(request.id, meta, result) + return successResponse(request.id, meta, await this.invokeUnary(admission, options)) } catch (error) { - if (request.method.startsWith('emulator.')) { + if (probeEmulator) { emulatorProbeError(`rpc ${request.method}`, error, { params: request.params }) @@ -108,12 +92,148 @@ export class RpcDispatcher { } } + // Why: streaming dispatch sends multiple responses through the reply callback instead of a Promise. async dispatchStreaming( - request: RpcRequest, + rawRequest: RpcRequest, reply: (response: string) => void, options?: RpcDispatchStreamingOptions ): Promise { - return this.streamingDispatcher.dispatch(request, reply, options) + const meta = this.meta() + const pendingAdmission = this.admit( + rawRequest, + options, + meta, + (_method: RpcMethod | RpcStreamingMethod): _method is RpcMethod | RpcStreamingMethod => true + ) + const admission = + pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission + if ('rejected' in admission) { + reply(JSON.stringify(admission.rejected)) + return + } + const { method, request, params, caller } = admission + + if (!isStreamingMethod(method)) { + try { + const result = await this.invokeUnary({ method, request, params, caller }, options) + reply(JSON.stringify(successResponse(request.id, meta, result))) + } catch (error) { + reply(JSON.stringify(mapDispatcherError(request, meta, error))) + } + return + } + + const { emit, recordedFeatureInteractions } = createDispatcherStreamingFeatureEmitter( + this.runtime, + request, + meta, + reply + ) + try { + const result = await method.handler( + params, + rpcContextFromTransport(this.runtime, request, options, caller), + emit + ) + recordRuntimeFeatureInteraction( + this.runtime, + request.method, + result, + recordedFeatureInteractions, + request.params + ) + } catch (error) { + reply(JSON.stringify(mapDispatcherError(request, meta, error))) + } + } + + private admit( + request: RpcRequest, + options: RpcDispatchStreamingOptions | undefined, + meta: RpcEnvelopeMeta, + isTransportSupported: (method: RpcMethod | RpcStreamingMethod) => method is M + ) { + return admitRpcRequest( + { + runtime: this.runtime, + registry: this.registry, + meta, + pinnedCallerScope: this.pinnedCallerScope + }, + request, + options, + isTransportSupported + ) + } + + private async invokeUnary( + { request, method, params, caller }: AdmittedUnaryCall, + options: RpcDispatchStreamingOptions | undefined + ): Promise { + const { runtime, orchestrationMutations, legacyOrchestration } = this + const context = rpcContextFromTransport(runtime, request, options, caller) + // Why: test fakes of the runtime omit the client-hosted browser route. + const clientHostedBrowser = await (runtime.routeClientHostedBrowserRpc?.( + request.method, + params + ) ?? { handled: false as const }) + if (clientHostedBrowser.handled) { + recordRuntimeFeatureInteraction( + runtime, + request.method, + clientHostedBrowser.result, + undefined, + request.params + ) + return clientHostedBrowser.result + } + + const compatibility = await legacyOrchestration.tryHandle(request, params, context.signal) + if (compatibility.handled) { + return compatibility.result + } + const effectiveParams = compatibility.params ?? params + const legacyCoordinator = legacyOrchestration.createCoordinatorInvocation( + request, + compatibility.legacyCoordinatorAuthority + ) + // Why: a legacy coordinator's fingerprint outranks the transport's, so skip minting a local one. + const fallbackCallerFingerprint = + context.authenticatedCallerFingerprint ?? + (!legacyCoordinator && + (request.method.startsWith('orchestration.federation') || + (!!request.orchestrationRequestId && isDurableMutation(request.method, effectiveParams))) + ? orchestrationMutations.getLocalAuthenticatedCallerFingerprint() + : undefined) + const callerFingerprint = + legacyCoordinator?.mutationCallerFingerprint ?? fallbackCallerFingerprint + const invoke = (mutation?: DurableMutationInvocation) => { + const legacyCoordinatorRunId = legacyCoordinator?.revalidate() + return method.handler(effectiveParams, { + ...context, + authenticatedCallerFingerprint: mutation?.identity.callerFingerprint ?? callerFingerprint, + recordMutationReceipt: mutation?.recordReceipt, + markWorkerDoneMutationEffectFree: mutation?.markWorkerDoneEffectFree, + markMutationEffectPossible: mutation?.markEffectPossible, + orchestrationMutation: mutation?.identity, + replayedMutationReceipt: mutation?.replayedReceipt, + legacyCoordinatorRunId, + legacyCoordinatorAuthority: legacyCoordinator?.authority, + revalidateLegacyCoordinator: legacyCoordinator?.revalidate, + orchestrationCompatibilityCallerAuthority: + compatibility.orchestrationCompatibilityCallerAuthority, + orchestrationCompatibilityEvidence: request.orchestrationCompatibilityEvidence + }) + } + const result = await orchestrationMutations.run( + request, + effectiveParams, + invoke, + callerFingerprint, + context.orchestrationCaller?.orcaSessionId + ) + recordRuntimeFeatureInteraction(runtime, request.method, result, undefined, request.params) + return result } private meta(): RpcEnvelopeMeta { diff --git a/src/main/runtime/rpc/rpc-streaming-dispatcher.ts b/src/main/runtime/rpc/rpc-streaming-dispatcher.ts deleted file mode 100644 index 10c572d7ac2..00000000000 --- a/src/main/runtime/rpc/rpc-streaming-dispatcher.ts +++ /dev/null @@ -1,107 +0,0 @@ -import { - isStreamingMethod, - type RpcEnvelopeMeta, - type RpcMethod, - type RpcRegistry, - type RpcRequest, - type RpcStreamingMethod -} from './core' - -import { successResponse } from './errors' -import type { OrcaRuntimeService } from '../orca-runtime' -import type { OrchestrationMutationExecutor } from './orchestration-mutation-executor' -import { recordRuntimeFeatureInteraction } from './runtime-feature-interaction' -import type { OrchestrationLegacyCompatibility } from './orchestration-legacy-compatibility' -import { - rpcContextFromTransport, - type RpcDispatchStreamingOptions -} from './dispatcher-stream-options' -import { mapDispatcherError } from './dispatcher-error-response' -import { createDispatcherStreamingFeatureEmitter } from './dispatcher-streaming-feature-emitter' -import { invokeDispatcherUnaryMethod } from './dispatcher-unary-method-invocation' -import type { RpcCallerScope } from './rpc-caller-scope' -import { admitRpcRequest } from './rpc-request-admission-prologue' - -export type RpcStreamingDispatcherDependencies = { - runtime: OrcaRuntimeService - registry: RpcRegistry - orchestrationMutations: OrchestrationMutationExecutor - legacyOrchestration: OrchestrationLegacyCompatibility - meta: () => RpcEnvelopeMeta - pinnedCallerScope?: RpcCallerScope -} - -export class RpcStreamingDispatcher { - constructor(private readonly dependencies: RpcStreamingDispatcherDependencies) {} - - // Why: streaming dispatch sends multiple responses through the reply callback instead of a Promise. - async dispatch( - rawRequest: RpcRequest, - reply: (response: string) => void, - options?: RpcDispatchStreamingOptions - ): Promise { - const { runtime, registry, orchestrationMutations, legacyOrchestration, meta } = - this.dependencies - const envelopeMeta = meta() - const pendingAdmission = admitRpcRequest( - { - runtime, - registry, - meta: envelopeMeta, - pinnedCallerScope: this.dependencies.pinnedCallerScope - }, - rawRequest, - options, - (_method: RpcMethod | RpcStreamingMethod): _method is RpcMethod | RpcStreamingMethod => true - ) - const admission = - pendingAdmission instanceof Promise ? await pendingAdmission : pendingAdmission - if ('rejected' in admission) { - reply(JSON.stringify(admission.rejected)) - return - } - const { method, request, params, caller } = admission - - if (!isStreamingMethod(method)) { - try { - const result = await invokeDispatcherUnaryMethod({ - runtime, - request, - method, - params, - context: rpcContextFromTransport(runtime, request, options, caller), - orchestrationMutations, - legacyOrchestration - }) - reply(JSON.stringify(successResponse(request.id, envelopeMeta, result))) - } catch (error) { - reply(JSON.stringify(mapDispatcherError(request, envelopeMeta, error))) - } - return - } - - const { emit, recordedFeatureInteractions } = createDispatcherStreamingFeatureEmitter( - runtime, - request, - envelopeMeta, - reply - ) - - try { - const result = await method.handler( - params, - rpcContextFromTransport(runtime, request, options, caller), - emit - ) - recordRuntimeFeatureInteraction( - runtime, - request.method, - result, - recordedFeatureInteractions, - request.params - ) - } catch (error) { - reply(JSON.stringify(mapDispatcherError(request, envelopeMeta, error))) - } - } -}