Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions apps/server/src/ws.ts
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,7 @@ const RPC_REQUIRED_SCOPE = new Map<string, AuthEnvironmentScope>([
[ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope],
[ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope],
[ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope],
[WS_METHODS.serverProbe, AuthOrchestrationReadScope],
[WS_METHODS.serverGetConfig, AuthOrchestrationReadScope],
[WS_METHODS.serverRefreshProviders, AuthOrchestrationOperateScope],
[WS_METHODS.serverUpdateProvider, AuthOrchestrationOperateScope],
Expand Down Expand Up @@ -1238,6 +1239,10 @@ const makeWsRpcLayer = (
}),
{ "rpc.aggregate": "orchestration" },
),
[WS_METHODS.serverProbe]: (_input) =>
observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), {
"rpc.aggregate": "server",
}),
[WS_METHODS.serverGetConfig]: (_input) =>
observeRpcEffect(WS_METHODS.serverGetConfig, loadServerConfig, {
"rpc.aggregate": "server",
Expand Down
2 changes: 1 addition & 1 deletion packages/client-runtime/src/connection/supervisor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -652,7 +652,7 @@ describe("EnvironmentSupervisor", () => {
}).pipe(Effect.provide(TestClock.layer())),
);

it.effect("keeps a healthy session when the application becomes active", () =>
it.effect("probes the active session without reconnecting on application activation", () =>
Effect.gen(function* () {
const probeCount = yield* Ref.make(0);
const probeCalled = yield* Deferred.make<void>();
Expand Down
27 changes: 26 additions & 1 deletion packages/client-runtime/src/rpc/session.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -169,9 +169,10 @@ const awaitSocket = Effect.fn("TestRpcSessionFactory.awaitSocket")(function* (

const awaitRequest = Effect.fn("TestRpcSessionFactory.awaitRequest")(function* (
socket: TestWebSocket,
index = 0,
) {
for (let attempt = 0; attempt < 100; attempt += 1) {
const request = socket.sent[0];
const request = socket.sent[index];
if (request) {
return decodeRpcRequest(decodeJson(request));
}
Expand Down Expand Up @@ -218,6 +219,30 @@ describe("RpcSessionFactory", () => {
expect(config).toEqual(SERVER_CONFIG);
expect(socket.sent).toHaveLength(1);

const probeFiber = yield* Effect.forkChild(session.probe);
const probeRequest = yield* awaitRequest(socket, 1);
expect(probeRequest).toMatchObject({
_tag: "Request",
tag: WS_METHODS.serverProbe,
payload: {},
});
socket.serverMessage(
encodeJson({
_tag: "Exit",
requestId: probeRequest.id,
exit: {
_tag: "Success",
value: {},
},
}),
);
yield* Fiber.join(probeFiber);

expect(socket.sent.map((request) => decodeRpcRequest(decodeJson(request)).tag)).toEqual([
WS_METHODS.serverGetConfig,
WS_METHODS.serverProbe,
]);

socket.close(1012, "service restart");
const error = yield* Effect.flip(session.closed);

Expand Down
9 changes: 5 additions & 4 deletions packages/client-runtime/src/rpc/session.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,8 +42,9 @@ export class RpcSessionFactory extends Context.Service<
type InitialConfigError = Effect.Error<
ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverGetConfig]>
>;
type ProbeError = Effect.Error<ReturnType<WsRpcProtocolClient[typeof WS_METHODS.serverProbe]>>;

function mapInitialConfigError(error: InitialConfigError): ConnectionAttemptError {
function mapSessionRpcError(error: InitialConfigError | ProbeError): ConnectionAttemptError {
switch (error._tag) {
case "EnvironmentAuthorizationError":
return new ConnectionBlockedError({
Expand Down Expand Up @@ -115,12 +116,12 @@ export const make = Effect.gen(function* () {
const client = yield* makeWsRpcProtocolClient.pipe(Effect.provide(protocolContext));
const initialConfig = yield* Effect.cached(
client[WS_METHODS.serverGetConfig]({}).pipe(
Effect.mapError(mapInitialConfigError),
Effect.mapError(mapSessionRpcError),
Effect.withSpan("environment.initialSync"),
),
);
const probe = client[WS_METHODS.serverGetConfig]({}).pipe(
Effect.mapError(mapInitialConfigError),
const probe = client[WS_METHODS.serverProbe]({}).pipe(
Effect.mapError(mapSessionRpcError),
Comment thread
cursor[bot] marked this conversation as resolved.
Resolved
Effect.asVoid,
Effect.withSpan("clientRuntime.connection.rpcSession.probe"),
);
Expand Down
8 changes: 8 additions & 0 deletions packages/contracts/src/rpc.ts
Original file line number Diff line number Diff line change
Expand Up @@ -201,6 +201,7 @@ export const WS_METHODS = {
previewAutomationFocusHost: "previewAutomation.focusHost",

// Server meta
serverProbe: "server.probe",
serverGetConfig: "server.getConfig",
serverRefreshProviders: "server.refreshProviders",
serverUpdateProvider: "server.updateProvider",
Expand Down Expand Up @@ -246,6 +247,12 @@ export const WsServerRemoveKeybindingRpc = Rpc.make(WS_METHODS.serverRemoveKeybi
error: Schema.Union([KeybindingsConfigError, EnvironmentAuthorizationError]),
});

export const WsServerProbeRpc = Rpc.make(WS_METHODS.serverProbe, {
payload: Schema.Struct({}),
success: Schema.Struct({}),
error: EnvironmentAuthorizationError,
});

export const WsServerGetConfigRpc = Rpc.make(WS_METHODS.serverGetConfig, {
payload: Schema.Struct({}),
success: ServerConfig,
Expand Down Expand Up @@ -682,6 +689,7 @@ export const WsSubscribeAuthAccessRpc = Rpc.make(WS_METHODS.subscribeAuthAccess,
});

export const WsRpcGroup = RpcGroup.make(
WsServerProbeRpc,
WsServerGetConfigRpc,
WsServerRefreshProvidersRpc,
WsServerUpdateProviderRpc,
Expand Down
Loading