Skip to content

Commit b25ea43

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
refactor(server): instrument WS RPCs in the group middleware
Every WebSocket RPC handler wrapped its body in observeRpcEffect, observeRpcStream or observeRpcStreamEffect to get its span and request metrics, while the group middleware already did the scope check. Calls the scope check rejected never reached a wrapper, so they were neither traced nor counted. The middleware, renamed WsRpcGuard, now does both: it checks the scope and wraps the call, rejected or not, in the ws.rpc.<method> span and the t3_rpc_requests_total / t3_rpc_request_duration metrics. Handlers are plain service calls again. The rpc.aggregate labels move into an exhaustive per-method map typed over WsRpcGroup, with every existing value kept as is. Per-call attributes (thread, command, task, provider instance ids) move into the handlers as Effect.annotateCurrentSpan. For stream RPCs the middleware wraps the server's stream runner, so the span and the duration cover the subscription until it ends, fails or is interrupted. The tracing opt-outs for the diagnostics RPCs are unchanged. deviceList keeps its input-dependent check in the handler. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
1 parent f870c41 commit b25ea43

10 files changed

Lines changed: 1418 additions & 2003 deletions

File tree

‎apps/server/src/auth/RpcAuthorization.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ import {
1515
RPC_REQUIRED_SCOPES,
1616
requiredScopeForRpcMethod,
1717
requiredScopeForDeviceList,
18-
rpcScopeAuthorizationLayer,
18+
wsRpcGuardLayer,
1919
} from "./RpcAuthorization.ts";
2020

2121
describe("RPC authorization scopes", () => {
@@ -140,7 +140,7 @@ describe("RPC scope middleware", () => {
140140
group.toLayerHandler(WS_METHODS.serverRetryResourceTelemetry, () =>
141141
Effect.sync(() => handled.push("retry")).pipe(Effect.andThen(Effect.never)),
142142
),
143-
rpcScopeAuthorizationLayer([AuthOrchestrationReadScope]),
143+
wsRpcGuardLayer([AuthOrchestrationReadScope]),
144144
),
145145
),
146146
);

‎apps/server/src/auth/RpcAuthorization.ts‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,14 +10,16 @@ import {
1010
ORCHESTRATION_V2_WS_METHODS,
1111
type AuthEnvironmentScope,
1212
EnvironmentAuthorizationError,
13-
RpcScopeAuthorization,
1413
WS_METHODS,
1514
WsRpcGroup,
15+
WsRpcGuard,
1616
} from "@t3tools/contracts";
1717
import * as Effect from "effect/Effect";
1818
import * as Layer from "effect/Layer";
1919
import type * as RpcGroup from "effect/unstable/rpc/RpcGroup";
2020

21+
import { instrumentRpc } from "../observability/RpcInstrumentation.ts";
22+
2123
type WsRpcMethod = RpcGroup.Rpcs<typeof WsRpcGroup>["_tag"];
2224

2325
/**
@@ -222,13 +224,17 @@ export const rpcAuthorizationError = (requiredScope: AuthEnvironmentScope) =>
222224
requiredScope,
223225
});
224226

225-
/** Authorizes every RPC on one connection against that connection's session scopes. */
226-
export const rpcScopeAuthorizationLayer = (scopes: ReadonlyArray<AuthEnvironmentScope>) =>
227-
Layer.succeed(RpcScopeAuthorization)((effect, { rpc }) => {
227+
/**
228+
* Guards every RPC on one connection: authorizes it against that connection's session scopes and
229+
* instruments it. A rejected call still gets its span and a failed request metric.
230+
*/
231+
export const wsRpcGuardLayer = (scopes: ReadonlyArray<AuthEnvironmentScope>) =>
232+
Layer.succeed(WsRpcGuard)((effect, { rpc }) => {
228233
const requiredScope = requiredScopeForRpcMethod(rpc._tag);
229-
return scopes.includes(requiredScope)
230-
? effect
231-
: Effect.fail(rpcAuthorizationError(requiredScope));
234+
return instrumentRpc(
235+
rpc,
236+
scopes.includes(requiredScope) ? effect : Effect.fail(rpcAuthorizationError(requiredScope)),
237+
);
232238
});
233239

234240
/** Retrying can install or restart tools even though ordinary listing is readable. */

‎apps/server/src/mcp/PreviewAutomationBroker.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ import * as TestClock from "effect/testing/TestClock";
2929
import * as RpcGroup from "effect/unstable/rpc/RpcGroup";
3030
import * as RpcTest from "effect/unstable/rpc/RpcTest";
3131

32-
import { rpcScopeAuthorizationLayer } from "../auth/RpcAuthorization.ts";
32+
import { wsRpcGuardLayer } from "../auth/RpcAuthorization.ts";
3333
import * as PreviewAutomationBroker from "./PreviewAutomationBroker.ts";
3434

3535
const makeBroker = PreviewAutomationBroker.make.pipe(Effect.provide(NodeServices.layer));
@@ -1262,7 +1262,7 @@ it.effect("evicts an unanswered host and lets later calls use a healthy runtime"
12621262
group.toLayer({
12631263
[WS_METHODS.previewAutomationConnect]: (host) => Stream.unwrap(broker.connect(host)),
12641264
}),
1265-
rpcScopeAuthorizationLayer([AuthOrchestrationOperateScope]),
1265+
wsRpcGuardLayer([AuthOrchestrationOperateScope]),
12661266
),
12671267
),
12681268
);

0 commit comments

Comments
 (0)